diff --git a/docs/profile-recovery-acceptance-map.md b/docs/profile-recovery-acceptance-map.md new file mode 100644 index 000000000..fc31bf047 --- /dev/null +++ b/docs/profile-recovery-acceptance-map.md @@ -0,0 +1,101 @@ +# Profile recovery and rollback acceptance map (#1025 / M17.02.2) + +This map states what repository tests prove about profile recovery, drain, +rollback, backup/restore, reconnect revalidation, and fencing. It does not +certify any live deployment. A "proven" result means a deterministic test in +this repository drives the named primitive; "unproven" means no test does. +Unsupported hosted candidates fail closed (`PROFILE_RUNTIME_NOT_QUALIFIED` for +the declared `CLOUD_PROFILE_UNQUALIFIED` denial, `PROFILE_RUNTIME_TRANSPORT_MISMATCH` +for the Node Pi Durable adapter's direct-local identity) rather than falling back +to a local or self-hosted runtime. The source-level profile map is in +`packages/profile-adapters/README.md`. + +#1025 remains open. The acceptance clauses below are not all satisfied. + +## Test ownership + +| File | Lane | Proves | +| -------------------------------------------------------------- | ------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `tests/profile-recovery-crash-restore.test.mjs` | smoke | Local composition: crash/restart recovery, fencing of the crashed owner, graceful drain of a scheduled reconciliation pass, portable backup refusal and total-loss restore | +| `tests/profile-recovery-fenced-rollback.test.mjs` | smoke | Rollback continuation input binding, reconnect eligibility reasons, channel-generation fencing (fake transaction) | +| `tests/profile-recovery-rebind-fencing.test.mjs` | smoke | Adapter-level rollback fence (synthetic `MockRuntimeAdapter` and managed-Pi driver fixtures): after the fence closes, the eight post-start operations on the previous binding fail `PROFILE_AUTHORITY_REJECTED` before the upstream runtime; a new owner at the next generation resumes the exact retained handle; a rebound managed-Pi attempt does not resend the physical start | +| `tests/profile-recovery-postgres-restore.integration.test.mjs` | integration (`bun run test:integration`, PR lane) | Self-hosted-server PostgreSQL total-loss backup/restore into a freshly provisioned database; disposable-database cleanup contract | + +Primitive semantics are owned by package tests and are not repeated in the +files above: + +- `packages/workflow-runtime/src/embedded-job-store.test.mjs`: single-use lease + tokens, reclaim after expiry, terminal jobs never reclaimed, token-holder-only + renew/complete/fail, queue reopen, and `WORKFLOW_RECOVERY_PARENT_ALREADY_CONTINUED`. +- `packages/domain/src/execution-cancellation-command.test.mjs`: lost-ACK replay + keeps the first identity, `EXECUTION_CANCELLATION_PAYLOAD_CONFLICT`, + `EXECUTION_CANCELLATION_CALLER_MISMATCH`, `EXECUTION_CANCELLATION_EXECUTION_INACTIVE`. +- `packages/profile-portability/src/index.test.mjs`: `PORTABLE_ACTIVE_WORK`, + interrupted import rollback to a pristine destination (`SIMULATED_IMPORT_INTERRUPTION`), + idempotent replay, Local and Hosted Simple persistence round trips. +- `packages/sqlite-persistence/src/restore-validation.test.mjs` and `index.test.mjs`: + `SQLITE_BACKUP_INVALID` digest rejection and restore validation. +- `packages/profile-adapters/src/index.test.mjs`: profile-to-deployment binding, + `PROFILE_DEPLOYMENT_MISMATCH` / `PROFILE_PERSISTENCE_MISMATCH` / `PROFILE_WAKE_MISMATCH`, + exact adapter/transport topology guards, the hosted cloud denial, and start-handle + attempt binding. +- `packages/runtime-sdk/src/eligibility.test.mjs`: offline, stale, revoked, + incompatible, and unverified candidates. +- `packages/database/src/runtime-channel-ownership-repository.test.mjs`: invalid + credential fences fail closed before any write. +- `packages/deployment/src/reconciliation-scheduler.test.mjs`: the scheduler's own + drain semantics. + +## Clause map + +| #1025 clause | Proof in this PR | Result | +| ------------------------------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Crash/recovery from a supported composition root | `LocalControlPlaneComposition` test "an abandoned job is recovered once under the next attempt and the crashed owner is fenced": abandoned lease retained across reopen, not reclaimed before expiry, recovered at attempt 2 with the identical `workflowKey` and `input`, crashed token rejected, recovered owner completes once, terminal job never reclaimed | Proven for Local. Self-hosted and Hosted: unproven | +| Drain | `LocalControlPlaneComposition` test "close waits for an in-flight scheduled reconciliation pass before storage closes": `close()` does not resolve while the pass is held, and resolves after it completes | Proven for Local scheduled reconciliation only. Workflow-queue stop, runtime-transport drain, Restate drain, and Pi-owned `drain()` are unproven | +| Upgrade | No upgrade executor is reachable from a supported composition root, and no test drives one. Portable import rollback is a data-import primitive, proven in `packages/profile-portability` | Unproven for every profile | +| Rollback | (a) Recovery continuation admission rejects a mismatched parent input or checkpoint (`WORKFLOW_RECOVERY_PARENT_MISMATCH`) and an ambiguous parent (`WORKFLOW_RECOVERY_PARENT_AMBIGUOUS`), in `profile-recovery-fenced-rollback`. (b) Interrupted portable import rolls back to a pristine destination, in `packages/profile-portability`. No release or version rollback primitive exists | Recovery-continuation admission proven at store level for Local. Import rollback proven as a primitive. No composition-level rollback of a deployed version is proven | +| Backup, Local | Composition test "backup refuses active work, then restores the evaluation into a fresh composition after total loss": `PORTABLE_ACTIVE_WORK` refusal, export from the composition's persistence, restore into a fresh composition directory, reopen read-back | Proven for Local | +| Backup/restore, Self-hosted server | `profile-recovery-postgres-restore.integration.test.mjs`: export from a seeded database, restore into a freshly provisioned database, replay returns `replayed`. Runs in the PR integration lane (`postgres-pull-request.yml` → `bun run test:integration`) | Proven against disposable local PostgreSQL. CI result pending the PR's integration run | +| Backup/restore, Self-hosted simple | Storage-level only: `restore-validation` and the Local/Hosted Simple port round trip in `packages/profile-portability`. No composition-level test | Unproven at composition level | +| Backup/restore, Hosted | Portability export is covered by `packages/profile-portability/src/postgres.integration.test.mjs` (Postgres source and destination). No live hosted restore | Unproven live | +| Revalidate reconnect grants | `evaluateRuntimeEligibility` reasons: `RUNTIME_OFFLINE`, `RUNTIME_EXPIRED`, `LOCAL_PROJECT_GRANT_REVOKED`, `LOCAL_PROJECT_GRANT_MISSING`, `CAPABILITY_SNAPSHOT_STALE`. Plus package `RUNTIME_STALE`, `RUNTIME_REVOKED` | Contract proven. No live reconnect handshake exercised | +| Revalidate target generations | `PostgresRuntimeChannelOwnershipRepository` rejects a claim at or below the current channel generation, rejects a foreign workspace (`RUNTIME_CHANNEL_WORKSPACE_MISMATCH`), and refuses a heartbeat from a superseded generation. Driven through the repository's fake-transaction seam | Proven at repository contract level, not against a live Postgres. Not exercised for Local | +| Revalidate input | Recovery continuation binds the parent's exact input and checkpoint (`WORKFLOW_RECOVERY_PARENT_MISMATCH`). Composition recovery returns the identical retained input. Cancellation payload conflict never re-dispatches (package test) | Proven for Local queue and recovery continuation | +| Fence one attempt owner throughout | Lease tokens are single-use (package test). Composition test: the crashed owner's token is rejected after recovery, and exactly one owner completes. Synthetic adapter-level fence in `tests/profile-recovery-rebind-fencing.test.mjs` | Proven for Local. Self-hosted and Hosted: unproven | + +## Per-profile matrix + +| Clause | Local | Self-hosted simple (SQLite + Restate) | Self-hosted server (PostgreSQL + Restate) | Hosted (cloud) | +| ---------------------------- | ----------------------------------------------------------------------- | ------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------ | +| Crash/recovery | Proven (composition) | Unproven | Unproven | Fail-closed binding only | +| Drain | Partly proven (scheduled reconciliation) | Unproven | Unproven | Unproven | +| Upgrade | Unproven | Unproven | Unproven | Unproven | +| Rollback | Recovery admission and import rollback proven as primitives | Unproven | Partly: PostgreSQL rollback on invalid derivation (`packages/profile-portability/src/postgres.integration.test.mjs`). Interrupted rollback not tested on PostgreSQL | Unproven | +| Backup/restore | Proven (composition) | Storage-level only | Proven against disposable local PostgreSQL | Unproven live | +| Reconnect-grant revalidation | Contract proven (one profile-independent function; no per-profile test) | Same | Same | Same | +| Target-generation fencing | Not exercised | Unproven | Repository contract (fake transaction) | Fail-closed binding only | +| Exact input binding | Proven | Unproven at composition | Unproven at composition | Unproven | +| One attempt owner | Proven (composition, package, and synthetic adapter-level fence) | Unproven | Unproven | Fail-closed binding only | + +"Fail-closed binding only" means `tests/...` and `packages/profile-adapters` +prove that Hosted refuses local storage, the embedded queue, direct-local +transports, and non-cloud placement. No hosted recovery behaviour is claimed. + +## Remaining gaps (required before #1025 can close) + +1. **Upgrade is unproven for every profile.** No upgrade executor is reachable + from a supported composition root, so no test can drive one yet. Adding one + needs an upstream-owned upgrade entry point; this PR does not build a + recovery engine. +2. **Drain is partly proven.** Only scheduled-reconciliation drain at Local + composition close is tested. Workflow-queue stop, runtime-transport drain, + and Restate drain are unproven. The Pi-owned `drain()` is excluded from this + scope. +3. **Restate wake and replay** for both self-hosted variants. No in-test harness + runs the pinned `@restatedev/restate-server`. +4. **Self-hosted simple has no composition-level recovery or backup test.** +5. **Hosted live composition** over the managed-cloud adapter is unproven. The + repository fails closed for unqualified candidates. +6. **Live reconnect handshake** is unproven. Grant revalidation is a contract. +7. **Channel-generation fencing** runs only against the fake transaction. + It has no PostgreSQL integration test. diff --git a/scripts/integration-shards.mjs b/scripts/integration-shards.mjs index 73b10dc2b..05b8fa2f9 100644 --- a/scripts/integration-shards.mjs +++ b/scripts/integration-shards.mjs @@ -114,6 +114,10 @@ export const INTEGRATION_SHARDS = [ package: '.', files: ['./tests/memory-process-loss.integration.test.mjs'], }, + { + package: '.', + files: ['./tests/profile-recovery-postgres-restore.integration.test.mjs'], + }, { package: 'apps/hosted-control-plane', files: [ diff --git a/scripts/run-bun-test-group.mjs b/scripts/run-bun-test-group.mjs index f5a927c21..31a6a28ce 100644 --- a/scripts/run-bun-test-group.mjs +++ b/scripts/run-bun-test-group.mjs @@ -68,6 +68,9 @@ const repositoryGroups = { 'tests/m11-security-probes.test.mjs', 'tests/m11-sqlite-benchmark.test.mjs', 'tests/neon-workflow.test.mjs', + 'tests/profile-recovery-crash-restore.test.mjs', + 'tests/profile-recovery-fenced-rollback.test.mjs', + 'tests/profile-recovery-rebind-fencing.test.mjs', 'tests/skill-library.test.mjs', 'tests/m11-recovery-rpo-rto.test.mjs', 'tests/m11-retention-apply-cli.test.mjs', diff --git a/scripts/run-integration-tests.mjs b/scripts/run-integration-tests.mjs index 5495798ca..a61282c8e 100644 --- a/scripts/run-integration-tests.mjs +++ b/scripts/run-integration-tests.mjs @@ -169,6 +169,22 @@ try { run('bun', ['x', 'turbo', 'run', 'test:integration', '--concurrency=1', '--log-order=stream'], { environment: integrationEnvironment, }) + // The test:integration task builds only each package's dependencies (^build), + // so a package that is consumed by root-level files but has no dependent + // integration task has no dist/ output. Package exports resolve to dist/, + // so build the consumed package explicitly before those root-level files run. + run( + 'bun', + [ + 'x', + 'turbo', + 'run', + 'build', + '--filter=@control-plane/profile-portability', + '--concurrency=1', + ], + { environment: integrationEnvironment } + ) // Repository-owned scenarios span composition roots without importing // repository fixtures from inside an application package. run( @@ -179,6 +195,7 @@ try { remoteDatabase ? '120000' : '30000', './tests/memory-process-loss.integration.test.mjs', './tests/langgraph-retirement-inventory-pg.integration.test.mjs', + './tests/profile-recovery-postgres-restore.integration.test.mjs', ], { environment: integrationEnvironment, diff --git a/tests/integration-runner-lifecycle.test.mjs b/tests/integration-runner-lifecycle.test.mjs index d0e23bdd1..39debeef4 100644 --- a/tests/integration-runner-lifecycle.test.mjs +++ b/tests/integration-runner-lifecycle.test.mjs @@ -141,7 +141,8 @@ describe('integration runner resource ownership', () => { const result = executeRunner(remoteTarget) expect(result.status).toBe(0) expect(result.calls.filter(({ program }) => program === 'docker')).toEqual([]) - expect(result.calls.filter(({ program }) => program === 'bun')).toHaveLength(3) + // turbo integration, the consumed-package build, root-level files, and the cloud drill. + expect(result.calls.filter(({ program }) => program === 'bun')).toHaveLength(4) }) test('remote configuration rejection does not contact Docker', () => { @@ -187,6 +188,21 @@ describe('integration runner resource ownership', () => { expect(startedProject(first)).not.toBe(startedProject(second)) }) + test('root-level integration files run only after their consumed workspace package is built', () => { + const result = executeRunner() + expect(result.status).toBe(0) + const bunCalls = result.calls.filter(({ program }) => program === 'bun') + const build = bunCalls.findIndex( + ({ args }) => + args.includes('build') && args.includes('--filter=@control-plane/profile-portability') + ) + const rootFile = bunCalls.findIndex(({ args }) => + args.includes('./tests/profile-recovery-postgres-restore.integration.test.mjs') + ) + expect(build).toBeGreaterThanOrEqual(0) + expect(rootFile).toBeGreaterThan(build) + }) + for (const [label, overrides] of [ ['partial startup failure', { FAKE_DOCKER_FAIL_ON: 'up' }], ['integration failure', { FAKE_BUN_FAIL: 'true' }], diff --git a/tests/j1-unfinished-child-process.fixture.mjs b/tests/j1-unfinished-child-process.fixture.mjs index 2c07f9285..f84610f5d 100644 --- a/tests/j1-unfinished-child-process.fixture.mjs +++ b/tests/j1-unfinished-child-process.fixture.mjs @@ -131,7 +131,11 @@ export async function createUnfinishedChildProcessHarness(options = {}) { } async function evidence() { try { - return (await readFile(join(directory, 'process-evidence.jsonl'), 'utf8')) + // Children append rows while this poll reads. Only newline-terminated rows are + // complete; a trailing fragment is left for the next poll, as stdout rows are. + const text = await readFile(join(directory, 'process-evidence.jsonl'), 'utf8') + return text + .slice(0, text.lastIndexOf('\n') + 1) .split('\n') .filter(Boolean) .map(JSON.parse) diff --git a/tests/profile-recovery-crash-restore.test.mjs b/tests/profile-recovery-crash-restore.test.mjs new file mode 100644 index 000000000..c227c0e0e --- /dev/null +++ b/tests/profile-recovery-crash-restore.test.mjs @@ -0,0 +1,330 @@ +// M17.02.2 (#1025): crash/recovery and portable backup/restore through the +// LocalControlPlaneComposition root (real SQLite persistence and embedded queue). +// +// - Crash/restart: a job whose worker dies keeps its abandoned lease across +// reopen, is recovered once under the next attempt with the identical workflow +// key and input, and fences the crashed owner's single-use token. +// - Backup/restore: the composition's own persistence exports through the +// portable source, refuses while work is active, and restores into a freshly +// provisioned composition after the source composition is closed. +// +// Primitive semantics (lease token rules, record planning, digest validation, +// binding guards) are owned by the package tests listed in +// docs/profile-recovery-acceptance-map.md and are not repeated here. The +// self-hosted-server PostgreSQL restore is in +// tests/profile-recovery-postgres-restore.integration.test.mjs. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, test } from 'bun:test' +import { LocalControlPlaneComposition } from '../apps/local-control-plane/src/composition.ts' +import { + SqliteContextPackageRepository, + SqliteEvaluationRepository, + SqliteExecutionPlanRepository, + SqliteExecutionRepository, +} from '@control-plane/sqlite-persistence' +import { ExecutionLifecycleService } from '@control-plane/domain' +import { contextPackageSerializationFixtures } from '@control-plane/context' +import { createExecutionPlanTestFixture } from '@control-plane/execution-plan/testing' +import { + PersistencePortableStateDestination, + PersistencePortableStateSource, + applyPortableImport, + exportPortableState, + planPortableImport, +} from '@control-plane/profile-portability' +import { observedEvaluationFixture } from '../packages/profile-portability/src/evaluation-fixture.mjs' + +const createdAt = '2026-10-09T00:00:00.000Z' +const laterAt = '2026-10-09T00:01:00.000Z' +const executionKey = 'exe_01JABCDEF0123456789ABCDEFG' +const leaseMs = 1_000 + +const temporaryDirectories = [] +const temporaryCompositions = [] + +/** Releases what a composition holds before its directory is removed. A started + * composition drains and closes itself through `close()`. An unstarted one is + * not closed by `close()`, so its storage handle is released explicitly. Every + * release is attempted, and failures surface together. */ +async function disposeComposition(composition) { + const failures = [] + const release = async (step) => { + try { + await step() + } catch (error) { + failures.push(error) + } + } + await release(() => composition.close()) + await release(() => composition.persistence.close()) + await release(() => composition.coordination.close()) + await release(() => composition.observability.close()) + if (failures.length > 0) { + throw new AggregateError(failures, 'LocalControlPlaneComposition disposal failed') + } +} + +afterEach(async () => { + const failures = [] + for (const composition of temporaryCompositions.splice(0)) { + try { + await disposeComposition(composition) + } catch (error) { + failures.push(error) + } + } + // Directories are removed only after every composition has released its storage. + for (const directory of temporaryDirectories.splice(0)) { + try { + await rm(directory, { recursive: true, force: true }) + } catch (error) { + failures.push(error) + } + } + if (failures.length > 0) throw new AggregateError(failures, 'profile recovery cleanup failed') +}) + +async function openTempDirectory(prefix) { + const directory = await mkdtemp(join(tmpdir(), prefix)) + temporaryDirectories.push(directory) + return directory +} + +function openComposition(directory, options = {}) { + const composition = new LocalControlPlaneComposition({ dataDirectory: directory, ...options }) + temporaryCompositions.push(composition) + return composition +} + +/** Retains the full guarded parent tuple the local composition's queue + * requires (context package, execution plan, lifecycle execution) and returns + * the accepted workflow input shape. */ +async function seedRetainedExecution(composition) { + const package_ = contextPackageSerializationFixtures.futurePi + await new SqliteContextPackageRepository(composition.persistence).put(package_) + const plan = createExecutionPlanTestFixture({ contextPackage: package_ }) + const reference = { + ...(await new SqliteExecutionPlanRepository(composition.persistence).put(plan)), + schemaVersion: plan.schemaVersion, + } + const input = { + executionId: executionKey, + workflowId: `wfl_${executionKey.slice(4)}`, + executionPlan: reference, + marketplacePluginReferences: [ + { + pluginId: 'plugin:control-plane:guard-fixture', + releaseId: `release:${'a'.repeat(64)}`, + canonicalContentDigest: `sha256:${'b'.repeat(64)}`, + }, + ], + deadlineAt: '2026-10-09T01:00:00.000Z', + } + await new ExecutionLifecycleService( + new SqliteExecutionRepository(composition.persistence) + ).createExecution({ + executionId: input.executionId, + correlation: plan.correlation, + executionPlan: reference, + marketplacePluginReferences: input.marketplacePluginReferences, + acceptedAt: createdAt, + }) + return input +} + +describe('M17.02.2 local composition crash recovery (embedded SQLite queue)', () => { + test('an abandoned job is recovered once under the next attempt and the crashed owner is fenced', async () => { + const directory = await openTempDirectory('profile-recovery-local-crash-') + const crashed = openComposition(directory) + expect(crashed.profile).toBe('local') + expect(crashed.durableExecution).toBe('embedded-sqlite') + await crashed.persistence.migrate() + + const input = await seedRetainedExecution(crashed) + const enqueued = await crashed.workflowJobs.enqueue({ + workflowKey: executionKey, + input, + at: createdAt, + }) + expect(enqueued.outcome).toBe('created') + const [inflight] = await crashed.workflowJobs.claimDue({ + owner: 'worker-crashed', + leaseMs, + now: createdAt, + limit: 5, + }) + expect(inflight).toMatchObject({ status: 'running', attempt: 1 }) + const crashedToken = inflight.lease.token + + // Simulated crash: storage closes mid-attempt with the lease held and no outcome. + crashed.persistence.close() + + // Restart the same composition root over the same retained data directory. + const restarted = openComposition(directory) + await restarted.persistence.migrate() + const retained = await restarted.workflowJobs.get(executionKey) + expect(retained).toMatchObject({ + status: 'running', + attempt: 1, + lease: { owner: 'worker-crashed' }, + }) + expect(retained.input).toEqual(input) + + // The abandoned lease is still live at the crash instant, so nothing is reclaimed. + await expect( + restarted.workflowJobs.claimDue({ owner: 'worker-early', leaseMs, now: createdAt, limit: 5 }) + ).resolves.toEqual([]) + + // After the lease expires a new owner recovers the same job identity and input. + const [recovered] = await restarted.workflowJobs.claimDue({ + owner: 'worker-recovered', + leaseMs, + now: laterAt, + limit: 5, + }) + expect(recovered).toMatchObject({ + workflowKey: executionKey, + attempt: 2, + input, + lease: { owner: 'worker-recovered' }, + }) + expect(recovered.lease.token).not.toBe(crashedToken) + + // The crashed owner's single-use token lands nothing after recovery. + await expect( + restarted.workflowJobs.complete({ + workflowKey: executionKey, + owner: 'worker-crashed', + token: crashedToken, + outcome: { executionId: executionKey, status: 'failed' }, + at: laterAt, + }) + ).resolves.toBe(false) + expect((await restarted.workflowJobs.get(executionKey)).outcome).toBeUndefined() + + // The recovered owner completes exactly once; a terminal job is never reclaimed. + await expect( + restarted.workflowJobs.complete({ + workflowKey: executionKey, + owner: 'worker-recovered', + token: recovered.lease.token, + outcome: { executionId: executionKey, status: 'completed' }, + at: laterAt, + }) + ).resolves.toBe(true) + await expect( + restarted.workflowJobs.claimDue({ owner: 'worker-late', leaseMs, now: laterAt, limit: 5 }) + ).resolves.toHaveLength(0) + }) +}) + +describe('M17.02.2 local composition graceful drain', () => { + test('close waits for an in-flight scheduled reconciliation pass before storage closes', async () => { + const directory = await openTempDirectory('profile-recovery-local-drain-') + let releasePass + const gate = new Promise((resolve) => { + releasePass = resolve + }) + let markPassStarted + const passStarted = new Promise((resolve) => { + markPassStarted = resolve + }) + let passFinished = false + const source = { + async load() { + throw new Error('UNEXPECTED_OBSERVATION_LOAD') + }, + async listCandidates() { + markPassStarted() + await gate + passFinished = true + return [] + }, + } + const composition = openComposition(directory, { + reconciliation: { source, effects: {}, intervalMs: 1, batchLimit: 1 }, + }) + await composition.start() + await passStarted + + let closed = false + const closing = composition.close().then(() => { + closed = true + }) + try { + // The pass is held open, so the drain cannot complete however long we wait. + await new Promise((resolve) => setTimeout(resolve, 20)) + expect(closed).toBe(false) + expect(passFinished).toBe(false) + } finally { + // Release the gate even when an assertion fails, so the drain settles and + // storage is not left closing under a pending pass. + releasePass() + await closing + } + expect(passFinished).toBe(true) + expect(closed).toBe(true) + }) +}) + +describe('M17.02.2 local composition portable backup/restore', () => { + test('backup refuses active work, then restores the evaluation into a fresh composition after total loss', async () => { + const sourceDirectory = await openTempDirectory('profile-recovery-local-src-') + const source = openComposition(sourceDirectory) + await source.persistence.migrate() + const evaluation = await observedEvaluationFixture() + await new SqliteEvaluationRepository(source.persistence).saveRun(evaluation) + + await expect( + exportPortableState( + new PersistencePortableStateSource({ + persistence: source.persistence, + componentVersions: { contracts: '1.0.0' }, + activeWorkIds: async () => [executionKey], + }), + { exportId: 'fenced-export', createdAt } + ) + ).rejects.toMatchObject({ code: 'PORTABLE_ACTIVE_WORK' }) + + const backup = await exportPortableState( + new PersistencePortableStateSource({ + persistence: source.persistence, + componentVersions: { contracts: '1.0.0' }, + }), + { exportId: 'local-backup', createdAt } + ) + expect(backup).toMatchObject({ + sourceProfile: 'local', + quiesced: true, + compatibility: { sourcePersistence: 'sqlite' }, + }) + expect(backup.records.length).toBeGreaterThan(0) + + // Total loss: the source composition is closed before the restore target is opened. + await disposeComposition(source) + const restoredDirectory = await openTempDirectory('profile-recovery-local-dst-') + const restored = openComposition(restoredDirectory) + await restored.persistence.migrate() + const destination = new PersistencePortableStateDestination({ + persistence: restored.persistence, + capabilities: new Set(), + secretProviders: new Set(), + }) + const plan = await planPortableImport(backup, destination) + expect(plan).toMatchObject({ applicable: true, conflicts: [] }) + await expect( + applyPortableImport(backup, plan, destination, {}, () => createdAt) + ).resolves.toMatchObject({ outcome: 'applied' }) + + // Reopen the restored composition: the retained state is read back from disk. + await disposeComposition(restored) + const reopened = openComposition(restoredDirectory) + await reopened.persistence.migrate() + await expect( + new SqliteEvaluationRepository(reopened.persistence).getRun(evaluation.evalRunId) + ).resolves.toEqual(evaluation) + }) +}) diff --git a/tests/profile-recovery-fenced-rollback.test.mjs b/tests/profile-recovery-fenced-rollback.test.mjs new file mode 100644 index 000000000..bdf8288df --- /dev/null +++ b/tests/profile-recovery-fenced-rollback.test.mjs @@ -0,0 +1,392 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { describe, expect, test } from 'bun:test' +import { SqlitePersistenceProvider } from '@control-plane/sqlite-persistence' +import { PostgresRuntimeChannelOwnershipRepository } from '../packages/database/src/runtime-channel-ownership-repository.ts' +import { evaluateRuntimeEligibility } from '../packages/runtime-sdk/src/eligibility.ts' +import { WorkflowJobStore } from '../packages/workflow-runtime/src/embedded-job-store.ts' + +/** + * Issue #1025 (M17.02.2) — fencing half, restricted to what package tests do + * not already pin. Lease-token rules, the recovery-chain branch guard, the + * cancellation identity/replay rules, portable-import rollback, and the + * exact-instance topology guard are owned by: + * - packages/workflow-runtime/src/embedded-job-store.test.mjs + * - packages/domain/src/execution-cancellation-command.test.mjs + * - packages/profile-portability/src/index.test.mjs + * - packages/profile-adapters/src/index.test.mjs + * and are not repeated here. + * + * This file adds the rollback-input binding and reconnect/generation fencing + * that no package test pins: + * - `WorkflowJobStore` rollback continuation admission binds the parent's exact + * input and checkpoint (`WORKFLOW_RECOVERY_PARENT_*`). + * - `evaluateRuntimeEligibility` reconnect revalidation for reconnecting, + * expired, revoked-grant, missing-grant, and expired-snapshot candidates. + * - `PostgresRuntimeChannelOwnershipRepository` target-generation fencing, + * driven through the repository's fake-transaction seam (no Postgres). + * + * Deterministic: disposable temp directories only, no network, no Pi, no + * production recovery, no Postgres instance. + */ + +const baseInstant = '2026-10-09T12:00:00.000Z' + +const executionInput = { + executionId: 'exe_01ARZ3NDEKTSV4RRFFQ69G5FAV', + workflowId: 'wfl_01ARZ3NDEKTSV4RRFFQ69G5FAV', + deadlineAt: '2026-10-09T13:00:00.000Z', + executionPlan: { planId: 'pln_01ARZ3NDEKTSV4RRFFQ69G5FAV', steps: [] }, +} + +const executionId = executionInput.executionId +const checkpointId = 'ckpt-rollback-v1' +const recoveryId = 'recover-ckpt-v1-0001' + +function recoveryKey(id) { + return `${executionId}:graph-recovery:${id}` +} + +async function withJobStore(run) { + const directory = await mkdtemp(join(tmpdir(), 'profile-recovery-rollback-')) + const provider = new SqlitePersistenceProvider({ path: join(directory, 'queue.sqlite') }) + try { + await provider.migrate() + await run(new WorkflowJobStore(provider)) + } finally { + try { + provider.close({ checkpoint: true }) + } finally { + await rm(directory, { recursive: true, force: true }) + } + } +} + +describe('WorkflowJobStore rollback continuation fencing (recovery graph)', () => { + async function completedParent(store) { + await store.enqueue({ workflowKey: executionId, input: executionInput, at: baseInstant }) + const [claim] = await store.claimDue({ + owner: 'attempt-owner-1', + leaseMs: 60_000, + now: baseInstant, + limit: 5, + }) + await store.complete({ + workflowKey: executionId, + owner: 'attempt-owner-1', + token: claim.lease.token, + outcome: { executionId, status: 'reconciliation_required', graphCheckpointId: checkpointId }, + at: baseInstant, + }) + } + + test('a parent continued without an explicit parent key is ambiguous to any further continuation', async () => { + await withJobStore(async (jobStore) => { + await completedParent(jobStore) + expect( + await jobStore.enqueue({ + workflowKey: recoveryKey(recoveryId), + input: executionInput, + recovery: { recoveryId, checkpointId }, + at: baseInstant, + }) + ).toMatchObject({ outcome: 'created' }) + // The legacy child named no parent key, so a later explicit continuation + // of the same checkpoint is fenced as ambiguous — never silently branched. + await expect( + jobStore.enqueue({ + workflowKey: recoveryKey('recover-ckpt-v1-0002'), + input: executionInput, + recovery: { + recoveryId: 'recover-ckpt-v1-0002', + checkpointId, + parentWorkflowKey: executionId, + }, + at: baseInstant, + }) + ).rejects.toThrow(/^WORKFLOW_RECOVERY_PARENT_AMBIGUOUS$/) + }) + }) + + test('exact input binding: the continuation must carry the parent input and checkpoint verbatim', async () => { + await withJobStore(async (jobStore) => { + await completedParent(jobStore) + + // Mutated rollback input is rejected, not silently adopted. + await expect( + jobStore.enqueue({ + workflowKey: recoveryKey(recoveryId), + input: { + ...executionInput, + workflowId: 'wfl_01ARZ3NDEKTSV4RRFFQ69G5FAV'.slice(0, -1) + 'X', + }, + recovery: { recoveryId, checkpointId, parentWorkflowKey: executionId }, + at: baseInstant, + }) + ).rejects.toThrow(/^WORKFLOW_RECOVERY_PARENT_MISMATCH$/) + + // A different target checkpoint cannot hijack the rollback. + await expect( + jobStore.enqueue({ + workflowKey: recoveryKey(recoveryId), + input: executionInput, + recovery: { + recoveryId, + checkpointId: 'ckpt-rollback-v2', + parentWorkflowKey: executionId, + }, + at: baseInstant, + }) + ).rejects.toThrow(/^WORKFLOW_RECOVERY_PARENT_MISMATCH$/) + + // A parent that has not reached reconciliation_required cannot be continued. + await jobStore.enqueue({ + workflowKey: `${executionId}-2`, + input: executionInput, + at: baseInstant, + }) + await expect( + jobStore.enqueue({ + workflowKey: `${executionId}-2:graph-recovery:${recoveryId}`, + input: executionInput, + recovery: { recoveryId, checkpointId, parentWorkflowKey: `${executionId}-2` }, + at: baseInstant, + }) + ).rejects.toThrow(/^WORKFLOW_RECOVERY_PARENT_MISMATCH$/) + expect(await jobStore.get(recoveryKey(recoveryId))).toBeUndefined() + }) + }) +}) + +describe('evaluateRuntimeEligibility reconnect revalidation (packages/runtime-sdk)', () => { + const connectionId = 'rtc_01JABCDEF0123456789ABCDEFG' + + function input(overrides = {}) { + const base = { + eligibilityVersion: 1, + evaluatedAt: '2026-10-09T12:00:30.000Z', + executionPlan: { + executionPlanId: 'pln_01JABCDEF0123456789ABCDEFG', + contentDigest: `sha256:${'a'.repeat(64)}`, + runtimeRequirements: [{ capability: 'stream.output', necessity: 'required' }], + }, + candidate: { + family: 'pi', + nodeStatus: 'online', + connection: { + runtimeConnectionId: connectionId, + identityDigest: `sha256:${'1'.repeat(64)}`, + connectionType: 'managed_local', + runtimeNodeRefId: 'rnr_01JABCDEF0123456789ABCDEFG', + runtimeDefinitionId: 'rtd_01JABCDEF0123456789ABCDEFG', + location: 'local_device', + opaqueNativeRef: 'nref_01JABCDEF0123456789ABCDEFG', + adapterVersion: '1.0.0', + driverVersion: '1.0.0', + harnessVersion: '1.0.0', + protocolVersion: '1.0.0', + status: 'connected', + health: 'healthy', + availabilityState: 'healthy', + capabilities: [{ name: 'stream.output', support: 'supported' }], + capabilitySnapshotVersion: 1, + capabilitySnapshotObservedAt: '2026-10-09T12:00:00.000Z', + capabilitySnapshotExpiresAt: '2026-10-09T12:01:00.000Z', + capabilityVerification: 'verified', + compatibilityState: 'compatible', + limitations: [], + diagnostics: [], + lastDiscoveredAt: '2026-10-09T12:00:00.000Z', + lastHeartbeatAt: '2026-10-09T12:00:00.000Z', + lastHealthCheckAt: '2026-10-09T12:00:00.000Z', + version: 2, + createdAt: '2026-10-09T11:00:00.000Z', + updatedAt: '2026-10-09T12:00:00.000Z', + }, + }, + policy: { + snapshot: { + policyId: 'workspace-standard', + version: 3, + digest: `sha256:${'b'.repeat(64)}`, + }, + allowedFamilies: ['pi', 'acp'], + allowedLocations: ['local_device'], + deniedRuntimeConnectionIds: [], + requireVerifiedCapabilities: true, + security: { status: 'allowed' }, + }, + localProjectGrant: { required: true, status: 'granted', grantRef: 'grant:project-1' }, + entitlement: { status: 'allowed', class: 'workspace' }, + preference: { runtimeConnectionId: connectionId, family: 'pi' }, + } + return { ...base, ...overrides } + } + + function withConnection(overrides) { + return input({ + candidate: { + ...input().candidate, + connection: { ...input().candidate.connection, ...overrides }, + }, + }) + } + + function reasonCodes(decision) { + return decision.reasons.map((reason) => reason.code) + } + + test('a currently granted runtime is eligible before reconnect', () => { + const decision = evaluateRuntimeEligibility(input()) + expect(decision.eligible).toBe(true) + expect(reasonCodes(decision)).toEqual([]) + }) + + test('a reconnecting or expired connection fails grant revalidation with typed reasons', () => { + const reconnecting = evaluateRuntimeEligibility( + withConnection({ availabilityState: 'reconnecting' }) + ) + expect(reconnecting.eligible).toBe(false) + expect(reasonCodes(reconnecting)).toContain('RUNTIME_OFFLINE') + + const expired = evaluateRuntimeEligibility( + withConnection({ status: 'expired', health: 'unavailable' }) + ) + expect(expired.eligible).toBe(false) + expect(reasonCodes(expired)).toContain('RUNTIME_EXPIRED') + }) + + test('a revoked or missing local project grant fails reconnect revalidation', () => { + const revokedGrant = evaluateRuntimeEligibility( + input({ + localProjectGrant: { required: true, status: 'revoked', grantRef: 'grant:project-1' }, + }) + ) + expect(revokedGrant.eligible).toBe(false) + expect(reasonCodes(revokedGrant)).toContain('LOCAL_PROJECT_GRANT_REVOKED') + + const missingGrant = evaluateRuntimeEligibility( + input({ + localProjectGrant: { required: true, status: 'missing', grantRef: 'grant:project-1' }, + }) + ) + expect(missingGrant.eligible).toBe(false) + expect(reasonCodes(missingGrant)).toContain('LOCAL_PROJECT_GRANT_MISSING') + }) + + test('an expired capability snapshot fails revalidation even while the connection looks healthy', () => { + const staleSnapshot = evaluateRuntimeEligibility( + withConnection({ capabilitySnapshotExpiresAt: '2026-10-09T12:00:00.000Z' }) + ) + expect(staleSnapshot.eligible).toBe(false) + expect(reasonCodes(staleSnapshot)).toContain('CAPABILITY_SNAPSHOT_STALE') + }) +}) + +describe('PostgresRuntimeChannelOwnershipRepository target-generation fencing (packages/database)', () => { + const nodeId = 'rnr_01DRZ3NDEKTSV4RRFFQ69G5FAV' + const workspaceId = 'wsp_01DRZ3NDEKTSV4RRFFQ69G5FAV' + const credentialFence = { credentialId: 'rgc_01DRZ3NDEKTSV4RRFFQ69G5FAV', revocationVersion: 1 } + + function record(generation, overrides = {}) { + return { + nodeId, + workspaceId, + gatewayInstanceId: `gateway-gen-${generation}`, + connectionId: `connection-gen-${generation}`, + channelGeneration: generation, + protocolVersion: { major: 1, minor: 6 }, + connectedAt: '2026-10-09T12:00:00.000Z', + lastHeartbeatAt: '2026-10-09T12:00:00.000Z', + ...overrides, + } + } + + /** Same fake-transaction seam the repository's own unit test drives. */ + function ownershipDatabase(current, fenceValid = true) { + const writes = [] + const transaction = { + execute: async () => [{ valid: fenceValid }], + select: () => ({ + from() { + return this + }, + where() { + return this + }, + async limit() { + return current === undefined ? [] : [current] + }, + }), + insert: () => ({ + values(value) { + writes.push({ kind: 'insert', value }) + return { async onConflictDoUpdate() {} } + }, + }), + update: () => ({ + set(value) { + writes.push({ kind: 'update', value }) + return { async where() {} } + }, + }), + } + return { writes, database: { transaction: (operation) => operation(transaction) } } + } + + function currentRow(generation) { + const owned = record(generation) + return { nodeId, workspaceId, generation, active: true, record: owned } + } + + test('a claim at or below the current channel generation is rejected; only a newer generation reconnects', async () => { + const repository = (current) => + new PostgresRuntimeChannelOwnershipRepository(ownershipDatabase(current).database) + + const sameGeneration = await repository(currentRow(2)).claim(record(2), credentialFence) + expect(sameGeneration.accepted).toBe(false) + expect(sameGeneration.previous.channelGeneration).toBe(2) + + const staleGeneration = await repository(currentRow(2)).claim(record(1), credentialFence) + expect(staleGeneration.accepted).toBe(false) + + // The reconnecting owner must present a strictly newer generation. + const live = ownershipDatabase(currentRow(2)) + const reconnect = await new PostgresRuntimeChannelOwnershipRepository(live.database).claim( + record(3), + credentialFence + ) + expect(reconnect).toMatchObject({ accepted: true }) + expect(reconnect.previous.channelGeneration).toBe(2) + expect(live.writes).toEqual([ + { kind: 'insert', value: expect.objectContaining({ generation: 3 }) }, + ]) + + // A foreign workspace can never take the channel over, whatever generation it claims. + await expect( + repository(currentRow(2)).claim( + record(9, { workspaceId: 'wsp_01ARZ3NDEKTSV4RRFFQ69G5FAV' }), + credentialFence + ) + ).rejects.toThrow('RUNTIME_CHANNEL_WORKSPACE_MISMATCH') + }) + + test('a superseded channel owner cannot heartbeat; only the current generation heartbeats', async () => { + const staleRepository = new PostgresRuntimeChannelOwnershipRepository( + ownershipDatabase(currentRow(3)).database + ) + expect( + await staleRepository.heartbeat(record(2, { lastHeartbeatAt: '2026-10-09T12:00:01.000Z' })) + ).toBe(false) + + const live = ownershipDatabase(currentRow(3)) + expect( + await new PostgresRuntimeChannelOwnershipRepository(live.database).heartbeat( + record(3, { lastHeartbeatAt: '2026-10-09T12:00:01.000Z' }), + credentialFence + ) + ).toBe(true) + expect(live.writes).toHaveLength(1) + }) +}) diff --git a/tests/profile-recovery-postgres-restore.integration.test.mjs b/tests/profile-recovery-postgres-restore.integration.test.mjs new file mode 100644 index 000000000..0a826836b --- /dev/null +++ b/tests/profile-recovery-postgres-restore.integration.test.mjs @@ -0,0 +1,191 @@ +// M17.02.2 (#1025): self-hosted-server (PostgreSQL) total-loss backup/restore. +// +// Integration lane only. `bun run test:integration` sets RUN_DATABASE_INTEGRATION +// and the DATABASE_* roles for the disposable local Postgres on 127.0.0.1:54329. +// A seeded hosted-server database is exported through PostgresPortableStateSource +// and restored into a freshly provisioned database through +// PostgresPortableStateDestination. The restore reads only the exported backup. +// +// The cleanup contract below runs without a database: it uses fake provisioning +// so partial setup and disposal failures are proven deterministically. + +import process from 'node:process' +import { describe, expect, test } from 'bun:test' +import { PostgresEvaluationRepository } from '@control-plane/database' +import { createIsolatedTestDatabase, integrationTestTimeout } from '@control-plane/database/testing' +import { + PostgresPortableStateDestination, + PostgresPortableStateSource, + applyPortableImport, + exportPortableState, + planPortableImport, +} from '@control-plane/profile-portability' +import { observedEvaluationFixture } from '../packages/profile-portability/src/evaluation-fixture.mjs' + +const createdAt = '2026-10-09T00:00:00.000Z' +const integrationEnabled = process.env.RUN_DATABASE_INTEGRATION === 'true' + +function isolatedCredentials() { + return { + administration: { role: 'administration', url: process.env.DATABASE_ADMIN_URL }, + migration: { role: 'migration', url: process.env.DATABASE_MIGRATION_URL }, + application: { role: 'application', url: process.env.DATABASE_URL }, + } +} + +/** Provisions `count` disposable databases, runs `run` with them, and disposes + * every database that was provisioned, including after a partial setup failure. + * Each failure is kept: a body or provisioning failure plus a disposal failure + * are reported together in one AggregateError. */ +async function withIsolatedDatabases(count, run, provision = createIsolatedTestDatabase) { + const databases = [] + let bodyFailed = false + let bodyError + try { + for (let index = 0; index < count; index += 1) { + databases.push(await provision(isolatedCredentials())) + } + await run(databases) + } catch (error) { + bodyFailed = true + bodyError = error + } + const disposals = await Promise.allSettled(databases.map((database) => database.dispose())) + const cleanupErrors = disposals.flatMap((result) => + result.status === 'rejected' ? [result.reason] : [] + ) + if (cleanupErrors.length > 0) { + throw new AggregateError( + bodyFailed ? [bodyError, ...cleanupErrors] : cleanupErrors, + 'PostgreSQL recovery test cleanup failed' + ) + } + if (bodyFailed) throw bodyError +} + +/** A provisioning fake that records disposal so cleanup can be asserted. */ +function fakeDatabase(name, { disposeError } = {}) { + const state = { name, disposed: 0 } + return { + state, + database: { + name, + dispose: async () => { + state.disposed += 1 + if (disposeError !== undefined) throw disposeError + }, + }, + } +} + +describe('M17.02.2 PostgreSQL disposable database cleanup (no database required)', () => { + test('partial replacement setup disposes the database already provisioned and surfaces the setup failure', async () => { + const first = fakeDatabase('control_plane_test_first') + let calls = 0 + const provision = async () => { + calls += 1 + if (calls === 1) return first.database + throw new Error('SIMULATED_REPLACEMENT_SETUP_FAILURE') + } + + await expect( + withIsolatedDatabases( + 2, + async () => { + throw new Error('BODY_MUST_NOT_RUN') + }, + provision + ) + ).rejects.toThrow('SIMULATED_REPLACEMENT_SETUP_FAILURE') + expect(first.state.disposed).toBe(1) + }) + + test('a disposal failure surfaces even when the body succeeded and every database is still attempted', async () => { + const failing = fakeDatabase('control_plane_test_failing', { + disposeError: new Error('SIMULATED_DISPOSE_FAILURE'), + }) + const healthy = fakeDatabase('control_plane_test_healthy') + const queue = [failing.database, healthy.database] + const provision = async () => queue.shift() + + const outcome = await withIsolatedDatabases(2, async () => undefined, provision).catch( + (error) => error + ) + expect(outcome).toBeInstanceOf(AggregateError) + expect(outcome.errors.map((error) => error.message)).toEqual(['SIMULATED_DISPOSE_FAILURE']) + expect(failing.state.disposed).toBe(1) + expect(healthy.state.disposed).toBe(1) + }) + + test('a body failure and a disposal failure are both reported', async () => { + const failing = fakeDatabase('control_plane_test_body', { + disposeError: new Error('SIMULATED_DISPOSE_FAILURE'), + }) + const outcome = await withIsolatedDatabases( + 1, + async () => { + throw new Error('SIMULATED_BODY_FAILURE') + }, + async () => failing.database + ).catch((error) => error) + expect(outcome).toBeInstanceOf(AggregateError) + expect(outcome.errors.map((error) => error.message)).toEqual([ + 'SIMULATED_BODY_FAILURE', + 'SIMULATED_DISPOSE_FAILURE', + ]) + }) +}) + +describe.skipIf(!integrationEnabled)( + 'M17.02.2 self-hosted-server backup/restore (PostgreSQL)', + () => { + test( + 'hosted-server database restores a seeded evaluation into a fresh server after total loss', + async () => { + const evaluation = await observedEvaluationFixture() + + await withIsolatedDatabases(2, async ([lost, replacement]) => { + await lost.migrate() + await replacement.migrate() + await new PostgresEvaluationRepository(lost.application).saveRun(evaluation) + + const backup = await exportPortableState( + new PostgresPortableStateSource({ + database: lost.application, + profile: 'hosted-server', + objectStore: 's3-compatible', + componentVersions: { contracts: '1.0.0' }, + }), + { exportId: 'hosted-server-backup', createdAt } + ) + expect(backup).toMatchObject({ + sourceProfile: 'hosted-server', + quiesced: true, + compatibility: { sourcePersistence: 'postgresql' }, + }) + + const destination = new PostgresPortableStateDestination({ + database: replacement.application, + profile: 'hosted-server', + capabilities: new Set(), + secretProviders: new Set(), + }) + const plan = await planPortableImport(backup, destination) + expect(plan).toMatchObject({ applicable: true, conflicts: [] }) + await expect( + applyPortableImport(backup, plan, destination, {}, () => createdAt) + ).resolves.toMatchObject({ outcome: 'applied' }) + await expect( + new PostgresEvaluationRepository(replacement.application).getRun(evaluation.evalRunId) + ).resolves.toEqual(evaluation) + + const replayPlan = await planPortableImport(backup, destination) + await expect( + applyPortableImport(backup, replayPlan, destination, {}, () => createdAt) + ).resolves.toMatchObject({ outcome: 'replayed' }) + }) + }, + integrationTestTimeout(120_000) + ) + } +) diff --git a/tests/profile-recovery-rebind-fencing.test.mjs b/tests/profile-recovery-rebind-fencing.test.mjs new file mode 100644 index 000000000..09b74e3a6 --- /dev/null +++ b/tests/profile-recovery-rebind-fencing.test.mjs @@ -0,0 +1,390 @@ +// M17.02.2 (#1025): profile rebinding and fenced rollback at the +// bindProfileRuntime composition boundary (consolidated from PR #1046). +// +// A rebound binding revalidates trusted topology, current authority, and +// residency, then resumes the retained upstream attempt through the selected +// adapter's supported idempotent semantics. It never re-issues the physical +// start and never converts a checkpoint across harnesses. Once the server-owned +// rollback fence closes, every operation on the previous owner fails closed +// before the upstream runtime is reached, and exactly one new owner at the next +// generation resumes the retained handle. Synthetic binding evidence only: no +// live runtime and no competing recovery engine. Storage-profile restore +// bindings are owned by packages/profile-adapters/src/index.test.mjs. + +import { executionConstraintFixtures } from '@control-plane/domain' +import { ManagedPiAdapter, ManagedPiDriver } from '@control-plane/managed-pi-adapter' +import { + DirectLocalRuntimeTransport, + MockRuntimeAdapter, + TransportedRuntimeAdapter, +} from '@control-plane/runtime-sdk' +import { + bindProfileRuntime, + ExecutionProfiles, + ProfileAdapterError, +} from '../packages/profile-adapters/src/index.ts' + +const component = (profile) => ({ + profile, + start: async () => undefined, + stop: async () => undefined, + health: async () => ({ ready: true, component: 'fixture', version: '1.0.0' }), +}) + +const persistence = (profile, dialect) => ({ + profile, + dialect, + migrate: async () => undefined, + health: async () => ({ ready: true, component: 'fixture', version: '1.0.0' }), + transaction: async () => undefined, + close: async () => undefined, +}) + +const composition = ( + profile, + dialect = profile === 'hosted-server' || profile === 'cloud' ? 'postgresql' : 'sqlite' +) => ({ + profile, + persistence: persistence(profile, dialect), + workflow: component(profile), + objectStore: { + put: async () => ({}), + get: async () => ({}), + head: async () => ({}), + delete: async () => undefined, + close: async () => undefined, + }, + secrets: { + resolve: async () => ({}), + health: async () => ({ ready: true, component: 'fixture', version: '1.0.0' }), + close: async () => undefined, + }, + coordination: { acquire: async () => undefined, close: async () => undefined }, + processes: { launch: async () => undefined, close: async () => undefined }, + discovery: { discover: async () => [] }, + observability: { write: () => undefined, close: async () => undefined }, +}) + +const localPlacement = Object.freeze({ + controlPlaneHostId: 'local-host', + runtimeHostId: 'local-host', + runtimeLocation: 'local_device', + coLocated: true, +}) + +const attemptId = 'att_01JABCDEF0123456789ABCDEFG' +const planId = 'pln_01JABCDEF0123456789ABCDEFG' +const digest = (character) => `sha256:${character.repeat(64)}` + +function startRequest(idempotencyKey) { + return { + attemptId, + idempotencyKey, + executionPlan: { + schemaVersion: 1, + executionPlanId: planId, + contentDigest: digest('a'), + runtimeRequirements: [ + { capability: 'stream.output', necessity: 'required', minimumSupport: 'supported' }, + ], + }, + } +} + +function managedExecutionPlan() { + return { + schemaVersion: 1, + executionPlanId: planId, + contentDigest: digest('a'), + profile: { + profileId: 'prf_01JABCDEF0123456789ABCDEFG', + profileVersionId: 'pfv_01JABCDEF0123456789ABCDEFG', + version: 3, + revision: 2, + schemaVersion: 1, + contentDigest: digest('b'), + }, + skills: [], + contextPackage: { + contextPackageId: 'ctx_01JABCDEF0123456789ABCDEFG', + contentDigest: digest('d'), + schemaVersion: 1, + compilerVersion: '1.0.0', + }, + runtimeRequirements: [ + { capability: 'stream.output', necessity: 'required', minimumSupport: 'supported' }, + { capability: 'execution.cancel', necessity: 'required', minimumSupport: 'supported' }, + ], + constraints: globalThis.structuredClone(executionConstraintFixtures.write), + policySnapshot: globalThis.structuredClone(executionConstraintFixtures.write.policySnapshot), + outputContract: { contractRef: 'contract://execution-result/v1' }, + } +} + +const allowedGuards = (calls = []) => ({ + authority: { + assertCurrent: async (context) => { + calls.push(['authority', context]) + }, + }, + residency: { + assertCurrent: async (context) => { + calls.push(['residency', context]) + }, + }, +}) + +const trustedTopology = (adapter, transport) => ({ + assertCurrent: async (_context, binding) => { + if (binding.adapter !== adapter || binding.transport !== transport) { + throw new ProfileAdapterError('PROFILE_RUNTIME_BINDING_MISMATCH', { + reason: 'TRUSTED_TOPOLOGY_REJECTED', + }) + } + }, +}) + +async function collect(iterable) { + for await (const _event of iterable) { + // Drain progress; the guard refusal rejects the iteration. + } +} + +async function failureOf(run) { + try { + await run() + } catch (error) { + return error + } + return undefined +} + +const managedPiNow = '2026-08-25T12:00:00.000Z' + +/** Minimal managed-Pi client over the canonical driver contract. */ +class RecordingManagedPiClient { + starts = [] + executions = new Map() + + async inspect() { + return { + driverVersion: '1.0.0', + runtimeVersion: '0.52.1', + protocolVersion: '1.0.0', + health: 'healthy', + capabilities: [ + { name: 'stream.output', support: 'supported' }, + { name: 'execution.cancel', support: 'supported' }, + { name: 'interaction.user-input', support: 'supported' }, + { name: 'interaction.approval', support: 'supported' }, + ], + limitations: [], + observedAt: managedPiNow, + } + } + + async start(command) { + this.starts.push(globalThis.structuredClone(command)) + const handle = { + handleId: `managed-pi:${command.attemptId}`, + attemptId: command.attemptId, + startedAt: managedPiNow, + } + this.executions.set(handle.handleId, { handle, state: 'running' }) + return handle + } + + async *progress(_handle, afterSequence = 0) { + if (afterSequence < 1) + yield { sequence: 1, occurredAt: managedPiNow, kind: 'status', state: 'running' } + } + + async submitInput(handle) { + return this.status(handle) + } + + async submitApproval(handle) { + return this.status(handle) + } + + async cancel(handle, request) { + const execution = this.executions.get(handle.handleId) + execution.state = 'cancelled' + execution.observedAt = request.requestedAt + return this.status(handle) + } + + async status(handle) { + const execution = this.executions.get(handle.handleId) + return { state: execution.state, observedAt: execution.observedAt ?? managedPiNow } + } + + async reconcile(handle) { + return this.status(handle) + } + + async cleanup(handle) { + this.executions.delete(handle.handleId) + } +} + +describe('profile recovery and fenced rollback', () => { + test('once the rollback fence closes every operation on the old owner fails before upstream effects', async () => { + const calls = [] + let authorityApproved = true + const driver = new MockRuntimeAdapter() + const transport = new DirectLocalRuntimeTransport(driver) + const adapter = new TransportedRuntimeAdapter(transport, 'mock') + const binding = await bindProfileRuntime({ + profile: ExecutionProfiles.local, + deployment: composition('local'), + candidate: { adapter, transport, placement: localPlacement }, + guards: { + authority: { + assertCurrent: async (context) => { + calls.push(['authority', context]) + if (!authorityApproved) throw new ProfileAdapterError('PROFILE_AUTHORITY_REJECTED') + }, + }, + residency: { + assertCurrent: async (context) => { + calls.push(['residency', context]) + }, + }, + }, + topology: trustedTopology(adapter, transport), + requiredCapabilities: [{ capability: 'stream.output', necessity: 'required' }], + }) + const request = startRequest('recovery:fence') + const handle = await binding.adapter.start(request) + + // The server-owned rollback fence closes: this owner's authority is no + // longer approved, exactly as a revoked grant or drained generation. + authorityApproved = false + const operations = [ + () => binding.adapter.status(handle), + () => binding.adapter.reconcile(handle), + () => collect(binding.adapter.progress(handle)), + () => + binding.adapter.submitInput(handle, { + interactionId: 'int_01JABCDEF0123456789ABCDEFG', + idempotencyKey: 'recovery:input', + text: 'hello', + }), + () => + binding.adapter.submitApproval(handle, { + interactionId: 'int_01JABCDEF0123456789ABCDEFG', + idempotencyKey: 'recovery:approval', + decision: 'approve', + }), + () => + binding.adapter.cancel(handle, { + idempotencyKey: 'recovery:cancel', + requestedAt: managedPiNow, + }), + () => binding.adapter.session({ operation: 'create', idempotencyKey: 'recovery:session' }), + () => binding.adapter.cleanup(handle), + ] + for (const operation of operations) { + const failure = await failureOf(operation) + expect(failure?.code).toBe('PROFILE_AUTHORITY_REJECTED') + } + + // Nothing reached the upstream runtime: the execution is untouched and + // the retained attempt is still running with the same handle. + expect((await driver.status(handle)).state).toBe('running') + expect(calls.some(([guard]) => guard === 'authority')).toBe(true) + }) + + test('a new owner at the next generation resumes while the fenced owner stays refused', async () => { + const driver = new MockRuntimeAdapter() + const transport = new DirectLocalRuntimeTransport(driver) + const adapter = new TransportedRuntimeAdapter(transport, 'mock') + const oldCalls = [] + let oldApproved = true + const oldBinding = await bindProfileRuntime({ + profile: ExecutionProfiles.local, + deployment: composition('local'), + candidate: { adapter, transport, placement: localPlacement }, + guards: { + authority: { + assertCurrent: async (context) => { + oldCalls.push(context) + if (!oldApproved) throw new ProfileAdapterError('PROFILE_AUTHORITY_REJECTED') + }, + }, + residency: { assertCurrent: async () => undefined }, + }, + topology: trustedTopology(adapter, transport), + }) + const handle = await oldBinding.adapter.start(startRequest('recovery:new-owner')) + + // Rollback: the old owner is fenced; the next owner binds the retained + // handle through a fresh approved authority and resumes it. + oldApproved = false + const newCalls = [] + const newBinding = await bindProfileRuntime({ + profile: ExecutionProfiles.local, + deployment: composition('local'), + candidate: { adapter, transport, placement: localPlacement }, + guards: allowedGuards(newCalls), + topology: trustedTopology(adapter, transport), + }) + const resumed = await newBinding.adapter.reconcile(handle) + expect(resumed.state).toBe('running') + expect(resumed.handle.handleId).toBe(handle.handleId) + expect(resumed.handle.attemptId).toBe(handle.attemptId) + // The new owner's guard saw the exact retained handle target. + const handleContext = newCalls.find( + ([kind, context]) => kind === 'authority' && context.operation === 'runtime.reconcile' + ) + expect(handleContext?.[1].target).toMatchObject({ + kind: 'handle', + handleId: handle.handleId, + attemptId: handle.attemptId, + startedAt: handle.startedAt, + }) + // The fenced owner is not resurrected by the new owner's success. + const refused = await failureOf(() => oldBinding.adapter.status(handle)) + expect(refused?.code).toBe('PROFILE_AUTHORITY_REJECTED') + }) + + test('a rebound profile resumes the retained managed-Pi attempt without resending the physical start', async () => { + const client = new RecordingManagedPiClient() + const transport = new DirectLocalRuntimeTransport( + new ManagedPiDriver({ client, adapterVersion: '1.0.0' }) + ) + const adapter = new ManagedPiAdapter({ transport }) + const binding = await bindProfileRuntime({ + profile: ExecutionProfiles.local, + deployment: composition('local'), + candidate: { adapter, transport, placement: localPlacement }, + guards: allowedGuards(), + topology: trustedTopology(adapter, transport), + requiredCapabilities: [{ capability: 'stream.output', necessity: 'required' }], + }) + const handle = await binding.adapter.start({ + attemptId, + idempotencyKey: 'recovery:managed-pi', + executionPlan: managedExecutionPlan(), + }) + expect(client.starts).toHaveLength(1) + + // Crash/rebind: the guarded binding is discarded and a fresh binding + // revalidates topology, authority and residency. The supported upstream + // retained-handle semantics resume the exact attempt. + const recovered = await bindProfileRuntime({ + profile: ExecutionProfiles.local, + deployment: composition('local'), + candidate: { adapter, transport, placement: localPlacement }, + guards: allowedGuards(), + topology: trustedTopology(adapter, transport), + }) + const status = await recovered.adapter.reconcile(handle) + expect(status.state).toBe('running') + expect(status.handle.handleId).toBe(handle.handleId) + expect(status.handle.attemptId).toBe(attemptId) + // Recovery never re-issues the physical start. + expect(client.starts).toHaveLength(1) + }) +}) diff --git a/tests/repository.test.mjs b/tests/repository.test.mjs index bc591b77b..066391495 100644 --- a/tests/repository.test.mjs +++ b/tests/repository.test.mjs @@ -240,6 +240,7 @@ test('discovers disjoint Bun test groups for Code Foundry', async () => { 'packages/testing/src/postgres.integration.test.mjs', 'tests/langgraph-retirement-inventory-pg.integration.test.mjs', 'tests/memory-process-loss.integration.test.mjs', + 'tests/profile-recovery-postgres-restore.integration.test.mjs', ]) const portabilityManifest = await readJson('packages/profile-portability/package.json') assert.match( @@ -310,6 +311,9 @@ test('discovers disjoint Bun test groups for Code Foundry', async () => { 'tests/m11-security-probes.test.mjs', 'tests/m11-sqlite-benchmark.test.mjs', 'tests/neon-workflow.test.mjs', + 'tests/profile-recovery-crash-restore.test.mjs', + 'tests/profile-recovery-fenced-rollback.test.mjs', + 'tests/profile-recovery-rebind-fencing.test.mjs', 'tests/skill-library.test.mjs', 'tests/m11-recovery-rpo-rto.test.mjs', 'tests/m11-retention-apply-cli.test.mjs',