From 937b3010c180eadfaa4f6308f59322f24fa85f1e Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Tue, 1 Sep 2026 04:06:16 -0600 Subject: [PATCH] feat(prime): add exact checkpoint rollback --- .../server/src/git/GitWorkflowService.test.ts | 2 + .../orchestration/Layers/CheckpointReactor.ts | 51 +- .../Layers/RollbackAdmissionAtomic.test.ts | 2 + .../Layers/RollbackReconciliation.test.ts | 2 + .../persistence/Layers/RollbackSagas.test.ts | 8 + .../src/persistence/Layers/RollbackSagas.ts | 18 +- .../Migrations/051_DurableRollbackSagas.ts | 2 + .../src/persistence/Services/RollbackSagas.ts | 5 + .../provider/Layers/ProviderService.test.ts | 105 +++- .../src/provider/Layers/ProviderService.ts | 100 +++- .../src/provider/Services/ProviderAdapter.ts | 36 +- .../src/provider/Services/ProviderService.ts | 24 +- .../prime/PrimeAgentDaemonAdapter.test.ts | 220 ++++++++ .../provider/prime/PrimeAgentDaemonAdapter.ts | 512 +++++++++++++++++- .../provider/prime/PrimeAgentDaemonBridge.ts | 4 + .../PrimeAgentDaemonSessionRuntime.test.ts | 127 +++++ .../prime/PrimeAgentDaemonSessionRuntime.ts | 243 +++++++++ .../prime/PrimeAgentRecoveryLedger.test.ts | 20 + .../prime/PrimeAgentRecoveryLedger.ts | 19 + .../src/rollback/RollbackAdmission.test.ts | 6 + apps/server/src/rollback/RollbackAdmission.ts | 12 +- .../src/rollback/RollbackSagaRunner.test.ts | 71 ++- .../server/src/rollback/RollbackSagaRunner.ts | 107 +++- apps/server/src/serverRuntimeStartup.ts | 40 +- docs/internals/prime-agent-daemon-parity.md | 81 +-- docs/internals/providers.md | 10 +- 26 files changed, 1717 insertions(+), 110 deletions(-) diff --git a/apps/server/src/git/GitWorkflowService.test.ts b/apps/server/src/git/GitWorkflowService.test.ts index 8455cf50e..b8052b424 100644 --- a/apps/server/src/git/GitWorkflowService.test.ts +++ b/apps/server/src/git/GitWorkflowService.test.ts @@ -218,6 +218,8 @@ describe("GitWorkflowService", () => { workspaceCwd: "/repo", sourceRevision: 2, targetRevision: 1, + sourceTurnId: null, + targetTurnId: null, sourceCheckpointRef: "refs/source" as never, sourceCheckpointOid: "a".repeat(40), targetCheckpointRef: "refs/target" as never, diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.ts index e0804e6d5..b649189b4 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.ts @@ -247,6 +247,7 @@ export const make = Effect.gen(function* () { readonly threadId: ThreadId; readonly cwd: string; readonly checkpointTurnCount: number; + readonly turnId: TurnId | null; readonly checkpointRef: ReturnType; readonly capturedAt: string; }) { @@ -273,10 +274,22 @@ export const make = Effect.gen(function* () { cwd: input.cwd, checkpointRef: input.checkpointRef, }); - const anchor = yield* providerService.captureConversationAnchor(input.threadId); + const anchor = yield* providerService.captureConversationAnchor({ + threadId: input.threadId, + binding: { + kind: "checkpoint", + checkpointTurnCount: input.checkpointTurnCount, + turnId: input.turnId, + checkpointRef: input.checkpointRef, + checkpointOid: checkpoint.oid, + sourceRevision: input.checkpointTurnCount, + }, + }); yield* rollbackRepository.value.putCheckpointAnchor({ threadId: input.threadId, checkpointTurnCount: input.checkpointTurnCount, + turnId: input.turnId, + sourceRevision: input.checkpointTurnCount, providerInstanceId: session.value.providerInstanceId, sessionIncarnationId: session.value.sessionIncarnationId, checkpointRef: input.checkpointRef, @@ -331,6 +344,7 @@ export const make = Effect.gen(function* () { threadId: input.threadId, cwd: input.cwd, checkpointTurnCount: input.turnCount, + turnId: input.turnId, checkpointRef: targetCheckpointRef, capturedAt: input.createdAt, }); @@ -590,14 +604,24 @@ export const make = Effect.gen(function* () { cwd: checkpointCwd, checkpointRef: baselineCheckpointRef, }); - if (baselineExists) { - return; + if (!baselineExists) { + yield* checkpointStore.captureCheckpoint({ + cwd: checkpointCwd, + checkpointRef: baselineCheckpointRef, + }); } - - yield* checkpointStore.captureCheckpoint({ + yield* capturePrivateCheckpointAnchor({ + threadId: thread.id, cwd: checkpointCwd, + checkpointTurnCount: currentTurnCount, + turnId: + thread.checkpoints.find( + (checkpoint) => checkpoint.checkpointTurnCount === currentTurnCount, + )?.turnId ?? null, checkpointRef: baselineCheckpointRef, + capturedAt: event.createdAt, }); + if (baselineExists) return; yield* receiptBus.publish({ type: "checkpoint.baseline.captured", threadId: thread.id, @@ -749,14 +773,23 @@ export const make = Effect.gen(function* () { cwd: checkpointCwd, checkpointRef: baselineCheckpointRef, }); - if (baselineExists) { - return; + if (!baselineExists) { + yield* checkpointStore.captureCheckpoint({ + cwd: checkpointCwd, + checkpointRef: baselineCheckpointRef, + }); } - - yield* checkpointStore.captureCheckpoint({ + yield* capturePrivateCheckpointAnchor({ + threadId, cwd: checkpointCwd, + checkpointTurnCount: currentTurnCount, + turnId: + thread.checkpoints.find((checkpoint) => checkpoint.checkpointTurnCount === currentTurnCount) + ?.turnId ?? null, checkpointRef: baselineCheckpointRef, + capturedAt: event.occurredAt, }); + if (baselineExists) return; yield* receiptBus.publish({ type: "checkpoint.baseline.captured", threadId, diff --git a/apps/server/src/orchestration/Layers/RollbackAdmissionAtomic.test.ts b/apps/server/src/orchestration/Layers/RollbackAdmissionAtomic.test.ts index bb4f3f7a5..9beafed81 100644 --- a/apps/server/src/orchestration/Layers/RollbackAdmissionAtomic.test.ts +++ b/apps/server/src/orchestration/Layers/RollbackAdmissionAtomic.test.ts @@ -53,6 +53,8 @@ const admission = Layer.succeed(RollbackAdmission, { workspaceCwd: "/workspace/atomic", sourceRevision: 2, targetRevision: 1, + sourceTurnId: null, + targetTurnId: null, sourceCheckpointRef: checkpointRefForThreadTurn(threadId, 2), sourceCheckpointOid: "2".repeat(40), targetCheckpointRef: checkpointRefForThreadTurn(threadId, 1), diff --git a/apps/server/src/orchestration/Layers/RollbackReconciliation.test.ts b/apps/server/src/orchestration/Layers/RollbackReconciliation.test.ts index 571d89f33..a1b938af5 100644 --- a/apps/server/src/orchestration/Layers/RollbackReconciliation.test.ts +++ b/apps/server/src/orchestration/Layers/RollbackReconciliation.test.ts @@ -41,6 +41,8 @@ const pending = { workspaceCwd: "/startup/workspace", sourceRevision: 2, targetRevision: 1, + sourceTurnId: null, + targetTurnId: null, sourceCheckpointRef: "refs/t3/checkpoints/source" as never, sourceCheckpointOid: "a".repeat(40), targetCheckpointRef: "refs/t3/checkpoints/target" as never, diff --git a/apps/server/src/persistence/Layers/RollbackSagas.test.ts b/apps/server/src/persistence/Layers/RollbackSagas.test.ts index 28b402adb..af247053f 100644 --- a/apps/server/src/persistence/Layers/RollbackSagas.test.ts +++ b/apps/server/src/persistence/Layers/RollbackSagas.test.ts @@ -37,6 +37,8 @@ const makeState = ( workspaceCwd: "/private/workspace/canary", sourceRevision: 2, targetRevision: 1, + sourceTurnId: null, + targetTurnId: null, sourceCheckpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-rollback-a/turn/2"), sourceCheckpointOid: "a".repeat(40), targetCheckpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-rollback-a/turn/1"), @@ -160,6 +162,8 @@ layer("RollbackSagaRepository", (it) => { yield* repository.putCheckpointAnchor({ threadId: threadA, checkpointTurnCount, + turnId: null, + sourceRevision: checkpointTurnCount, providerInstanceId, sessionIncarnationId, checkpointRef: CheckpointRef.make( @@ -175,6 +179,8 @@ layer("RollbackSagaRepository", (it) => { yield* repository.putCheckpointAnchor({ threadId: threadA, checkpointTurnCount: 1, + turnId: null, + sourceRevision: 1, providerInstanceId, sessionIncarnationId, checkpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-rollback-a/turn/1"), @@ -188,6 +194,8 @@ layer("RollbackSagaRepository", (it) => { .putCheckpointAnchor({ threadId: threadA, checkpointTurnCount: 1, + turnId: null, + sourceRevision: 1, providerInstanceId, sessionIncarnationId, checkpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-rollback-a/turn/1"), diff --git a/apps/server/src/persistence/Layers/RollbackSagas.ts b/apps/server/src/persistence/Layers/RollbackSagas.ts index d759ca52a..614bc5d64 100644 --- a/apps/server/src/persistence/Layers/RollbackSagas.ts +++ b/apps/server/src/persistence/Layers/RollbackSagas.ts @@ -5,7 +5,7 @@ import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import type { SqlError } from "effect/unstable/sql/SqlError"; -import { NonNegativeInt, ProjectId, ThreadId } from "@t3tools/contracts"; +import { NonNegativeInt, ProjectId, ThreadId, TurnId } from "@t3tools/contracts"; import { PersistenceDecodeError, toPersistenceDecodeError, @@ -41,6 +41,8 @@ const LeaseDbRow = Schema.Struct({ const AnchorDbRow = Schema.Struct({ threadId: ThreadId, checkpointTurnCount: NonNegativeInt, + turnId: Schema.NullOr(TurnId), + sourceRevision: NonNegativeInt, providerInstanceId: Schema.String, sessionIncarnationId: Schema.String, checkpointRef: Schema.String, @@ -299,16 +301,19 @@ const make = Effect.gen(function* () { }); const rows = yield* sql<{ readonly anchorDigest: string }>` INSERT INTO rollback_checkpoint_anchors ( - thread_id, checkpoint_turn_count, provider_instance_id, session_incarnation_id, - checkpoint_ref, checkpoint_oid, anchor_json, anchor_digest, captured_at + thread_id, checkpoint_turn_count, turn_id, source_revision, provider_instance_id, + session_incarnation_id, checkpoint_ref, checkpoint_oid, anchor_json, anchor_digest, captured_at ) VALUES ( - ${anchor.threadId}, ${anchor.checkpointTurnCount}, ${anchor.providerInstanceId}, + ${anchor.threadId}, ${anchor.checkpointTurnCount}, ${anchor.turnId}, ${anchor.sourceRevision}, + ${anchor.providerInstanceId}, ${anchor.sessionIncarnationId}, ${anchor.checkpointRef}, ${anchor.checkpointOid}, ${anchorJson}, ${anchor.anchorDigest}, ${anchor.capturedAt} ) ON CONFLICT (thread_id, checkpoint_turn_count, provider_instance_id, session_incarnation_id) DO UPDATE SET captured_at = rollback_checkpoint_anchors.captured_at - WHERE rollback_checkpoint_anchors.checkpoint_ref = excluded.checkpoint_ref + WHERE rollback_checkpoint_anchors.turn_id IS excluded.turn_id + AND rollback_checkpoint_anchors.source_revision = excluded.source_revision + AND rollback_checkpoint_anchors.checkpoint_ref = excluded.checkpoint_ref AND rollback_checkpoint_anchors.checkpoint_oid = excluded.checkpoint_oid AND rollback_checkpoint_anchors.anchor_digest = excluded.anchor_digest RETURNING anchor_digest AS "anchorDigest" @@ -324,6 +329,7 @@ const make = Effect.gen(function* () { const getCheckpointAnchor: RollbackSagaRepositoryShape["getCheckpointAnchor"] = (input) => sql` SELECT thread_id AS "threadId", checkpoint_turn_count AS "checkpointTurnCount", + turn_id AS "turnId", source_revision AS "sourceRevision", provider_instance_id AS "providerInstanceId", session_incarnation_id AS "sessionIncarnationId", checkpoint_ref AS "checkpointRef", checkpoint_oid AS "checkpointOid", anchor_json AS "anchorJson", anchor_digest AS "anchorDigest", captured_at AS "capturedAt" @@ -353,6 +359,8 @@ const make = Effect.gen(function* () { decodeAnchor({ threadId: row.threadId, checkpointTurnCount: row.checkpointTurnCount, + turnId: row.turnId, + sourceRevision: row.sourceRevision, providerInstanceId: row.providerInstanceId, sessionIncarnationId: row.sessionIncarnationId, checkpointRef: row.checkpointRef, diff --git a/apps/server/src/persistence/Migrations/051_DurableRollbackSagas.ts b/apps/server/src/persistence/Migrations/051_DurableRollbackSagas.ts index 08fc1bf76..235a98ff1 100644 --- a/apps/server/src/persistence/Migrations/051_DurableRollbackSagas.ts +++ b/apps/server/src/persistence/Migrations/051_DurableRollbackSagas.ts @@ -56,6 +56,8 @@ export default Effect.gen(function* () { CREATE TABLE IF NOT EXISTS rollback_checkpoint_anchors ( thread_id TEXT NOT NULL, checkpoint_turn_count INTEGER NOT NULL, + turn_id TEXT, + source_revision INTEGER NOT NULL, provider_instance_id TEXT NOT NULL, session_incarnation_id TEXT NOT NULL, checkpoint_ref TEXT NOT NULL, diff --git a/apps/server/src/persistence/Services/RollbackSagas.ts b/apps/server/src/persistence/Services/RollbackSagas.ts index ab494b5ad..36ad5ef45 100644 --- a/apps/server/src/persistence/Services/RollbackSagas.ts +++ b/apps/server/src/persistence/Services/RollbackSagas.ts @@ -10,6 +10,7 @@ import { ProviderInstanceId, RuntimeSessionId, ThreadId, + TurnId, } from "@t3tools/contracts"; import type { PersistenceDecodeError, PersistenceSqlError } from "../Errors.ts"; @@ -43,6 +44,8 @@ export const RollbackSagaState = Schema.Struct({ workspaceCwd: Schema.String, sourceRevision: NonNegativeInt, targetRevision: NonNegativeInt, + sourceTurnId: Schema.NullOr(TurnId), + targetTurnId: Schema.NullOr(TurnId), sourceCheckpointRef: CheckpointRef, sourceCheckpointOid: Schema.String, targetCheckpointRef: CheckpointRef, @@ -87,6 +90,8 @@ export type RollbackSagaRecord = typeof RollbackSagaRecord.Type; export const RollbackCheckpointAnchor = Schema.Struct({ threadId: ThreadId, checkpointTurnCount: NonNegativeInt, + turnId: Schema.NullOr(TurnId), + sourceRevision: NonNegativeInt, providerInstanceId: ProviderInstanceId, sessionIncarnationId: RuntimeSessionId, checkpointRef: CheckpointRef, diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index ebd9e8b55..f997e6aa4 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -2720,6 +2720,79 @@ fanout.layer("ProviderServiceLive fanout", (it) => { }), ); + it.effect("rejects an exact anchor result after its provider generation retires", () => + Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + const threadId = asThreadId("thread-generation-replaced-absolute-anchor"); + yield* provider.startSession(threadId, { + provider: CODEX_DRIVER, + providerInstanceId: codexInstanceId, + threadId, + runtimeMode: "full-access", + }); + + const originalCapabilities = fanout.codex.adapter.capabilities; + const originalAbsoluteRollback = fanout.codex.adapter.absoluteConversationRollback; + const originalRuntimeFence = fanout.codex.adapter.runtimeFence; + const current = yield* Ref.make(true); + const captureEntered = yield* Deferred.make(); + const releaseCapture = yield* Deferred.make(); + Object.assign(fanout.codex.adapter, { + capabilities: { ...originalCapabilities, conversationRollback: "absolute" }, + absoluteConversationRollback: { + isAvailable: () => Effect.succeed(true), + captureAnchor: () => + Deferred.succeed(captureEntered, undefined).pipe( + Effect.andThen(Deferred.await(releaseCapture)), + Effect.as({ anchor: { privateLeaf: "stale-private" }, digest: "stale-digest" }), + ), + inspectAnchor: () => Effect.succeed({ anchor: {}, digest: "unused" }), + applyAnchor: () => Effect.void, + releaseAnchor: () => Effect.void, + }, + runtimeFence: { + generation: {}, + configRevision: "private-anchor-test-revision", + isCurrent: Ref.get(current), + }, + }); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + Object.assign(fanout.codex.adapter, { capabilities: originalCapabilities }); + if (originalAbsoluteRollback === undefined) { + delete (fanout.codex.adapter as { absoluteConversationRollback?: unknown }) + .absoluteConversationRollback; + } else { + Object.assign(fanout.codex.adapter, { + absoluteConversationRollback: originalAbsoluteRollback, + }); + } + if (originalRuntimeFence === undefined) { + delete (fanout.codex.adapter as { runtimeFence?: unknown }).runtimeFence; + } else { + Object.assign(fanout.codex.adapter, { runtimeFence: originalRuntimeFence }); + } + }), + ); + + const anchorFiber = yield* provider.captureConversationAnchor!({ + threadId, + binding: { + kind: "source", + sourceRevision: 1, + checkpointRef: "refs/t3/checkpoints/private/source" as never, + checkpointOid: "a".repeat(40), + turnId: null, + }, + }).pipe(Effect.forkChild); + yield* Deferred.await(captureEntered); + yield* Ref.set(current, false); + yield* Deferred.succeed(releaseCapture, undefined); + + assert.isTrue(Exit.isFailure(yield* Fiber.await(anchorFiber))); + }), + ); + it.effect("keeps Stop authoritative while an account transition is starting", () => Effect.gen(function* () { const provider = yield* ProviderService.ProviderService; @@ -3883,6 +3956,23 @@ describe("agent browser access", () => { let recoveredSession: ProviderSession | undefined; const recoveryAdapter: ProviderAdapterShape = { ...codex.adapter, + capabilities: { + ...codex.adapter.capabilities, + conversationRollback: "absolute", + }, + absoluteConversationRollback: { + isAvailable: () => Effect.succeed(true), + captureAnchor: () => Effect.succeed({ anchor: {}, digest: "unused" }), + inspectAnchor: () => Effect.succeed({ anchor: {}, digest: "unused" }), + applyAnchor: () => Effect.void, + releaseAnchor: () => Effect.void, + prepareRecovery: (input) => + Effect.sync(() => { + assert.deepEqual(input.sourceAnchor, { privateLeaf: "restart-source-private" }); + assert.deepEqual(input.desiredAnchor, { privateLeaf: "restart-target-private" }); + order.push("quarantine"); + }), + }, recoverSession: (input) => Effect.gen(function* () { assert.isDefined(McpProviderSession.readMcpProviderSession(threadId)); @@ -3943,10 +4033,21 @@ describe("agent browser access", () => { ); yield* Effect.yieldNow; - yield* provider.recoverRestartSessions!(); + yield* provider.recoverRestartSessions!({ + pendingAbsoluteRollbacks: new Map([ + [ + threadId, + { + sourceAnchor: { privateLeaf: "restart-source-private" }, + desiredAnchor: { privateLeaf: "restart-target-private" }, + expectedAnchor: { privateLeaf: "restart-source-private" }, + }, + ], + ]), + }); yield* Fiber.join(consumer); - assert.deepEqual(order, ["mcp", "recover", "activate"]); + assert.deepEqual(order, ["mcp", "recover", "quarantine", "activate"]); const [recoveredEvent] = yield* Ref.get(recoveredEvents); assert.equal(recoveredEvent?.eventId, asEventId("evt-restart-adopted-output")); assert.equal(recoveredEvent?.turnId, recoveredTurnId); diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 7e5c52315..7268e9d0a 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -1603,7 +1603,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( const recoverRestartSessions: ProviderServiceMethod<"recoverRestartSessions"> = Effect.fn( "recoverRestartSessions", - )(function* () { + )(function* (input) { const bindings = yield* directory.listBindings(); for (const binding of bindings) { let adoptedAdapter: ProviderAdapterShape | undefined; @@ -1675,6 +1675,30 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( }, ); yield* requireAdapterGenerationCurrent(adapter, "ProviderService.recoverRestartSessions"); + if (input?.unrecoverableAbsoluteRollbacks?.has(binding.threadId) === true) { + return yield* toValidationError( + "ProviderService.recoverRestartSessions", + "Pending exact rollback authority is incomplete for restart adoption.", + ); + } + const pendingRollback = input?.pendingAbsoluteRollbacks?.get(binding.threadId); + if (pendingRollback !== undefined) { + const operations = adapter.absoluteConversationRollback; + if ( + adapter.capabilities.conversationRollback !== "absolute" || + operations?.prepareRecovery === undefined + ) { + return yield* toValidationError( + "ProviderService.recoverRestartSessions", + "Pending exact rollback quarantine is unavailable for the recovered session.", + ); + } + yield* operations.prepareRecovery({ + threadId: binding.threadId, + ...pendingRollback, + }); + yield* requireAdapterGenerationCurrent(adapter, "ProviderService.recoverRestartSessions"); + } yield* adapter.activateRecoveredSession(binding.threadId); }).pipe( Effect.catchCause((cause) => @@ -2892,6 +2916,10 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( operation: "ProviderService.absoluteConversationRollback", allowRecovery: true, }); + yield* requireAdapterGenerationCurrent( + routed.adapter, + "ProviderService.absoluteConversationRollback", + ); const operations = routed.adapter.absoluteConversationRollback; if ( routed.adapter.capabilities.conversationRollback !== "absolute" || @@ -2913,17 +2941,31 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( operation: "ProviderService.hasAbsoluteConversationRollback", allowRecovery: true, }); - return ( - routed.adapter.capabilities.conversationRollback === "absolute" && - routed.adapter.absoluteConversationRollback !== undefined + if ( + routed.adapter.capabilities.conversationRollback !== "absolute" || + routed.adapter.absoluteConversationRollback === undefined + ) { + return false; + } + yield* requireAdapterGenerationCurrent( + routed.adapter, + "ProviderService.hasAbsoluteConversationRollback", + ); + const available = yield* routed.adapter.absoluteConversationRollback + .isAvailable(threadId) + .pipe(Effect.orElseSucceed(() => false)); + yield* requireAdapterGenerationCurrent( + routed.adapter, + "ProviderService.hasAbsoluteConversationRollback", ); + return available; }); const captureConversationAnchor: NonNullable> = - Effect.fn("captureConversationAnchor")(function* (threadId) { - const { operations } = yield* resolveAbsoluteConversationRollback(threadId); - return yield* operations - .captureAnchor(threadId) + Effect.fn("captureConversationAnchor")(function* (input) { + const { routed, operations } = yield* resolveAbsoluteConversationRollback(input.threadId); + const anchor = yield* operations + .captureAnchor(input) .pipe( Effect.mapError(() => toValidationError( @@ -2932,12 +2974,17 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ), ), ); + yield* requireAdapterGenerationCurrent( + routed.adapter, + "ProviderService.captureConversationAnchor", + ); + return anchor; }); const inspectConversationAnchor: NonNullable> = Effect.fn("inspectConversationAnchor")(function* (threadId) { - const { operations } = yield* resolveAbsoluteConversationRollback(threadId); - return yield* operations + const { routed, operations } = yield* resolveAbsoluteConversationRollback(threadId); + const anchor = yield* operations .inspectAnchor(threadId) .pipe( Effect.mapError(() => @@ -2947,12 +2994,17 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ), ), ); + yield* requireAdapterGenerationCurrent( + routed.adapter, + "ProviderService.inspectConversationAnchor", + ); + return anchor; }); const applyConversationAnchor: NonNullable> = Effect.fn("applyConversationAnchor")(function* (input) { - const { operations } = yield* resolveAbsoluteConversationRollback(input.threadId); - return yield* operations + const { routed, operations } = yield* resolveAbsoluteConversationRollback(input.threadId); + yield* operations .applyAnchor(input.threadId, input.anchor) .pipe( Effect.mapError(() => @@ -2962,6 +3014,29 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ), ), ); + yield* requireAdapterGenerationCurrent( + routed.adapter, + "ProviderService.applyConversationAnchor", + ); + }); + + const releaseConversationAnchor: NonNullable> = + Effect.fn("releaseConversationAnchor")(function* (input) { + const { routed, operations } = yield* resolveAbsoluteConversationRollback(input.threadId); + yield* operations + .releaseAnchor(input.threadId, input.anchor) + .pipe( + Effect.mapError(() => + toValidationError( + "ProviderService.releaseConversationAnchor", + "The provider could not prove the final conversation anchor.", + ), + ), + ); + yield* requireAdapterGenerationCurrent( + routed.adapter, + "ProviderService.releaseConversationAnchor", + ); }); const rollbackConversation: ProviderServiceMethod<"rollbackConversation"> = Effect.fn( @@ -3213,6 +3288,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( captureConversationAnchor, inspectConversationAnchor, applyConversationAnchor, + releaseConversationAnchor, uploadFeedback, // Each access creates a fresh PubSub subscription so that multiple // consumers (ProviderRuntimeIngestion, CheckpointReactor, etc.) each diff --git a/apps/server/src/provider/Services/ProviderAdapter.ts b/apps/server/src/provider/Services/ProviderAdapter.ts index d116bf7d2..e0ce4c5f0 100644 --- a/apps/server/src/provider/Services/ProviderAdapter.ts +++ b/apps/server/src/provider/Services/ProviderAdapter.ts @@ -33,6 +33,7 @@ import type { ProviderSetSessionInputQueueModeInput, ProviderSession, ProviderSessionStartInput, + CheckpointRef, ProviderUploadFeedbackInput, ProviderUploadFeedbackResult, ThreadId, @@ -65,14 +66,43 @@ export interface ProviderConversationAnchorReceipt { readonly digest: string; } +export type ProviderConversationAnchorBinding = + | { + readonly kind: "checkpoint"; + readonly checkpointTurnCount: number; + readonly turnId: TurnId | null; + readonly checkpointRef: CheckpointRef; + readonly checkpointOid: string; + readonly sourceRevision: number; + } + | { + readonly kind: "source"; + readonly sourceRevision: number; + readonly checkpointRef: CheckpointRef; + readonly checkpointOid: string; + readonly turnId: TurnId | null; + }; + export interface ProviderAbsoluteConversationRollback { - readonly captureAnchor: ( - threadId: ThreadId, - ) => Effect.Effect; + /** Exact per-thread gates. Static adapter capability alone is not enough. */ + readonly isAvailable: (threadId: ThreadId) => Effect.Effect; + readonly captureAnchor: (input: { + readonly threadId: ThreadId; + readonly binding: ProviderConversationAnchorBinding; + }) => Effect.Effect; readonly inspectAnchor: ( threadId: ThreadId, ) => Effect.Effect; readonly applyAnchor: (threadId: ThreadId, anchor: Json) => Effect.Effect; + /** Holds native output until the saga commits and proves the final leaf. */ + readonly releaseAnchor: (threadId: ThreadId, anchor: Json) => Effect.Effect; + /** Installs quarantine before retained restart-adoption frames are released. */ + readonly prepareRecovery?: (input: { + readonly threadId: ThreadId; + readonly sourceAnchor: Json; + readonly desiredAnchor: Json; + readonly expectedAnchor: Json; + }) => Effect.Effect; } export interface ProviderAdapterCapabilities { diff --git a/apps/server/src/provider/Services/ProviderService.ts b/apps/server/src/provider/Services/ProviderService.ts index 84cf4ae04..55dedb23c 100644 --- a/apps/server/src/provider/Services/ProviderService.ts +++ b/apps/server/src/provider/Services/ProviderService.ts @@ -64,6 +64,7 @@ import type { ProviderServiceError } from "../Errors.ts"; import type { ProviderAdapterCapabilities, ProviderConversationAnchorReceipt, + ProviderConversationAnchorBinding, } from "./ProviderAdapter.ts"; import type { ProviderInstanceRoutingInfo } from "./ProviderAdapterRegistry.ts"; @@ -95,7 +96,17 @@ export interface ProviderServiceShape { ) => Effect.Effect; /** Adopt eligible surviving Prime executions before startup orphan reconciliation. */ - readonly recoverRestartSessions?: () => Effect.Effect; + readonly recoverRestartSessions?: (input?: { + readonly unrecoverableAbsoluteRollbacks?: ReadonlySet; + readonly pendingAbsoluteRollbacks?: ReadonlyMap< + ThreadId, + { + readonly sourceAnchor: Json; + readonly desiredAnchor: Json; + readonly expectedAnchor: Json; + } + >; + }) => Effect.Effect; /** * Interrupt a running provider turn. @@ -259,9 +270,10 @@ export interface ProviderServiceShape { readonly hasAbsoluteConversationRollback?: ( threadId: ThreadId, ) => Effect.Effect; - readonly captureConversationAnchor?: ( - threadId: ThreadId, - ) => Effect.Effect; + readonly captureConversationAnchor?: (input: { + readonly threadId: ThreadId; + readonly binding: ProviderConversationAnchorBinding; + }) => Effect.Effect; readonly inspectConversationAnchor?: ( threadId: ThreadId, ) => Effect.Effect; @@ -269,6 +281,10 @@ export interface ProviderServiceShape { readonly threadId: ThreadId; readonly anchor: Json; }) => Effect.Effect; + readonly releaseConversationAnchor?: (input: { + readonly threadId: ThreadId; + readonly anchor: Json; + }) => Effect.Effect; /** * Upload a thread and return the provider's shareable feedback identifier. diff --git a/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.test.ts b/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.test.ts index 3de3b1eb0..0aa37b691 100644 --- a/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.test.ts +++ b/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.test.ts @@ -5,6 +5,8 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { describe, expect, it } from "@effect/vitest"; import { ApprovalRequestId, + CheckpointRef, + CommandId, defaultInstanceIdForDriver, EnvironmentId, PrimeAgentSettings, @@ -18,6 +20,7 @@ import { RuntimeTaskId, SessionInteractionRequestId, ThreadId, + TurnId, } from "@t3tools/contracts"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; @@ -36,6 +39,7 @@ import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; import * as ProviderSessionRuntime from "../../persistence/ProviderSessionRuntime.ts"; import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; import * as ServerSettings from "../../serverSettings.ts"; +import type { PrimeAgentRecoveryLedgerShape } from "./PrimeAgentRecoveryLedger.ts"; import * as AnalyticsService from "../../telemetry/AnalyticsService.ts"; import type { ProviderAdapterError } from "../Errors.ts"; import * as ProviderAdapterRegistry from "../Services/ProviderAdapterRegistry.ts"; @@ -538,11 +542,22 @@ function fakeRuntimeFactory( captures.queue = queue; captures.promptObserved = promptObserved; for (const event of captures.startupEvents) yield* Queue.offer(queue, event); + let conversationLeafId = "prime-root-leaf"; const runtime: PrimeAgentDaemonSessionRuntime = { resumeCursor: PRIME_AGENT_DAEMON_RESUME_CURSOR, sessionId: "native-session-secret", sessionFile: `${input.sessionDir}/native-session-secret.jsonl`, activeSessionId: "native-active-secret", + conversationRuntimeGeneration: "native-runtime-generation-secret", + initialConversationLeafId: conversationLeafId, + conversationRollbackAvailable: true, + inspectConversationLeaf: Effect.sync(() => conversationLeafId), + navigateConversationLeaf: ({ desiredLeafId }) => + Effect.sync(() => { + conversationLeafId = desiredLeafId; + }), + prepareConversationRollback: () => Effect.void, + releaseConversationRollback: () => Effect.void, initialSnapshot: { ...initialSnapshot(), children: captures.agentRoster }, initialResources: { available: true, skills: [], prompts: [], commands: [] }, sideQuestionsAvailable: captures.sideQuestionsAvailable, @@ -9473,4 +9488,209 @@ describe("PrimeAgentDaemonAdapter", () => { }), ).pipe(Effect.provide(testLayer)), ); + + it.effect("exposes exact private rollback only on the managed public Prime API path", () => + Effect.scoped( + Effect.gen(function* () { + class ExactDaemonConnection { + getState() {} + navigateTree() {} + } + const exactManager = { + bridge: { DaemonAgentConnection: ExactDaemonConnection }, + recoveryEnabled: false, + platform: "darwin", + architecture: "arm64", + } as unknown as PrimeAgentDaemonManager; + const captures = makeCaptures(); + const adapter = yield* makePrimeAgentDaemonAdapter(decodeSettings({}), exactManager, { + instanceId, + runtimeFactory: fakeRuntimeFactory(captures), + recoveryManagedBuildId: "managed-prime-test-build", + }); + const subscription = yield* subscribe(adapter); + const sessionIncarnationId = RuntimeSessionId.make("private-rollback-incarnation"); + yield* adapter.startSession({ + threadId, + cwd: process.cwd(), + runtimeMode: "full-access", + sessionIncarnationId, + }); + yield* awaitObservedType(subscription.observed, "thread.started"); + + expect(adapter.capabilities.conversationRollback).toBe("absolute"); + const rollback = adapter.absoluteConversationRollback!; + expect(yield* rollback.isAvailable(threadId)).toBe(true); + const source = yield* rollback.captureAnchor({ + threadId, + binding: { + kind: "source", + sourceRevision: 0, + checkpointRef: CheckpointRef.make("refs/t3/checkpoints/private/source"), + checkpointOid: "a".repeat(40), + turnId: TurnId.make("private-source-turn"), + }, + }); + const sourceAnchor = source.anchor as Readonly>; + const desiredAnchor = { + ...sourceAnchor, + leafId: "prime-target-leaf-private", + } as Schema.Json; + + yield* rollback.applyAnchor(threadId, desiredAnchor); + const quarantinedEventCount = subscription.events.length; + yield* offer(captures, { + _tag: "GoalUpdated", + goal: { + available: true, + active: false, + status: "complete", + objective: "must stay private while rollback is pending", + tokensUsed: 1, + timeUsedSeconds: 1, + continuationsUsed: 0, + }, + }); + yield* Effect.yieldNow; + yield* Effect.yieldNow; + expect(subscription.events).toHaveLength(quarantinedEventCount); + expect((yield* rollback.inspectAnchor(threadId)).digest).not.toBe(source.digest); + + yield* rollback.releaseAnchor(threadId, desiredAnchor); + yield* offer(captures, { + _tag: "GoalUpdated", + goal: { + available: true, + active: false, + status: "complete", + objective: "public after exact release", + tokensUsed: 1, + timeUsedSeconds: 1, + continuationsUsed: 0, + }, + }); + yield* awaitObservedType(subscription.observed, "session.goal.updated"); + + const stale = { + ...sourceAnchor, + runtimeGeneration: "stale-private-generation", + } as Schema.Json; + const staleResult = yield* rollback.applyAnchor(threadId, stale).pipe(Effect.result); + expect(staleResult._tag).toBe("Failure"); + const publicEncoding = encodeUnknownJson(subscription.events); + expect(publicEncoding).not.toContain("prime-target-leaf-private"); + expect(publicEncoding).not.toContain("private-source-turn"); + }), + ).pipe(Effect.provide(testLayer)), + ); + + it.effect("retains a settled recoverable Prime session for exact rollback", () => + Effect.scoped( + Effect.gen(function* () { + class ExactDaemonConnection { + getState() {} + navigateTree() {} + } + const exactManager = { + bridge: { DaemonAgentConnection: ExactDaemonConnection }, + recoveryEnabled: true, + platform: "darwin", + architecture: "arm64", + } as unknown as PrimeAgentDaemonManager; + const captures = makeCaptures(); + const markAdmittedCalls: string[] = []; + const markIdleObserved = yield* Deferred.make(); + const markIdleCalls: Array<{ readonly threadId: string; readonly ownerToken: string }> = []; + const recoveryLedger = { + putPrepared: () => Effect.void, + get: () => Effect.succeed(Option.none()), + listActive: () => Effect.succeed([]), + markAdmitted: () => + Effect.sync(() => { + markAdmittedCalls.push("admitted"); + return true; + }), + markIdle: (input: { readonly threadId: string; readonly ownerToken: string }) => + Effect.sync(() => { + markIdleCalls.push(input); + return true; + }).pipe(Effect.tap(() => Deferred.succeed(markIdleObserved, undefined))), + discardPrepared: () => Effect.succeed(true), + updateTranscriptProgress: () => Effect.succeed(true), + claim: () => Effect.succeed(Option.none()), + releaseClaim: () => Effect.succeed(true), + commitAdoption: () => Effect.succeed(true), + markNativeCleanup: () => Effect.succeed(true), + markTerminalProjected: () => Effect.void, + markCheckpointQuiesced: () => Effect.void, + deleteIfSettled: () => Effect.succeed(false), + } as unknown as PrimeAgentRecoveryLedgerShape; + const adapter = yield* makePrimeAgentDaemonAdapter(decodeSettings({}), exactManager, { + instanceId, + runtimeFactory: fakeRuntimeFactory(captures), + recoveryManagedBuildId: "managed-prime-test-build", + recoveryLedger, + }); + const subscription = yield* subscribe(adapter); + const sessionIncarnationId = RuntimeSessionId.make("retained-rollback-incarnation"); + yield* adapter.startSession({ + threadId, + cwd: process.cwd(), + runtimeMode: "full-access", + sessionIncarnationId, + }); + const turnInput = { + threadId, + input: "retain exact rollback authority", + sessionIncarnationId, + admissionRequestId: CommandId.make("retained-admission"), + } as const; + yield* adapter.prepareTurnRecovery!(turnInput); + expect(adapter.capabilities.conversationRollback).toBe("absolute"); + expect(captures.runtimeInputs).toHaveLength(2); + expect(captures.runtimeInputs.at(-1)?.recovery?.kind).toBe("create"); + const turn = yield* adapter.sendTurn(turnInput).pipe(Effect.forkChild); + yield* Queue.take(captures.promptObserved!); + yield* offer(captures, { _tag: "RunCompleted", messages: [] }); + yield* Fiber.join(turn); + yield* awaitObservedType(subscription.observed, "turn.completed"); + + expect(markAdmittedCalls).toHaveLength(1); + yield* Deferred.await(markIdleObserved); + const retainedSessions = yield* adapter.listSessions(); + expect(retainedSessions).toHaveLength(1); + expect(retainedSessions[0]).toMatchObject({ status: "ready" }); + expect(markIdleCalls).toHaveLength(1); + expect(markIdleCalls[0]?.threadId).toBe(threadId); + expect( + (yield* adapter.listSessions()).find((session) => session.threadId === threadId), + ).toMatchObject({ + status: "ready", + sessionIncarnationId, + }); + expect(yield* adapter.absoluteConversationRollback!.isAvailable(threadId)).toBe(true); + const held = yield* adapter.prepareTurnRecovery!({ + ...turnInput, + admissionRequestId: CommandId.make("next-admission-before-checkpoint-hold"), + }).pipe(Effect.result); + expect(held._tag).toBe("Failure"); + if (held._tag === "Failure") expect(held.failure).toMatchObject({ reason: "busy" }); + expect(captures.runtimeInputs).toHaveLength(2); + }), + ).pipe(Effect.provide(testLayer)), + ); + + it.effect("keeps the absolute capability unsupported without the managed build proof", () => + Effect.scoped( + Effect.gen(function* () { + const captures = makeCaptures(); + const adapter = yield* makePrimeAgentDaemonAdapter(decodeSettings({}), manager, { + instanceId, + runtimeFactory: fakeRuntimeFactory(captures), + }); + expect(adapter.capabilities.conversationRollback).toBe("unsupported"); + expect(adapter.absoluteConversationRollback).toBeUndefined(); + }), + ).pipe(Effect.provide(testLayer)), + ); }); diff --git a/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.ts b/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.ts index 525d4e6fd..71cee7b84 100644 --- a/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.ts +++ b/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.ts @@ -75,7 +75,11 @@ import { type ProviderAdapterError, } from "../Errors.ts"; import type { PrimeAgentAdapterShape } from "../Services/PrimeAgentAdapter.ts"; -import { BUILT_IN_ADAPTER_CONVERSATION_ROLLBACK_MODES } from "../Services/ProviderAdapter.ts"; +import { + BUILT_IN_ADAPTER_CONVERSATION_ROLLBACK_MODES, + type ProviderConversationAnchorReceipt, + type ProviderConversationAnchorBinding, +} from "../Services/ProviderAdapter.ts"; import { type EventNdjsonLogger, makeEventNdjsonLogger } from "../Layers/EventNdjsonLogger.ts"; import { primeAgentSessionDirectory } from "../Layers/PrimeAgentAdapter.ts"; import type { @@ -172,6 +176,36 @@ function crashAtPrimeAgentRecoveryTestBarrier(stage: PrimeAgentRecoveryTestCrash } export const PRIME_AGENT_SIDE_QUESTION_TIMEOUT_MS = 2 * 60_000; const PRIME_AGENT_SIDE_QUESTION_MAX_ACTIVE = 4; +const primeConversationLeafId = Schema.String.check(Schema.isMinLength(1), Schema.isMaxLength(256)); +const primeConversationAnchorBinding = Schema.Union([ + Schema.Struct({ + kind: Schema.Literal("checkpoint"), + checkpointTurnCount: Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)), + turnId: Schema.NullOr(Schema.String), + checkpointRef: Schema.String, + checkpointOid: Schema.String, + sourceRevision: Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)), + }), + Schema.Struct({ + kind: Schema.Literal("source"), + sourceRevision: Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)), + checkpointRef: Schema.String, + checkpointOid: Schema.String, + turnId: Schema.NullOr(Schema.String), + }), +]); +const primeConversationAnchor = Schema.Struct({ + version: Schema.Literal(1), + provider: Schema.Literal("prime-agent"), + providerInstanceId: Schema.String, + runtimeGeneration: Schema.String, + sessionIncarnationId: Schema.String, + nativeSessionId: Schema.String, + leafId: primeConversationLeafId, + binding: Schema.optional(primeConversationAnchorBinding), +}).annotate({ parseOptions: { onExcessProperty: "error" } }); +type PrimeConversationAnchor = typeof primeConversationAnchor.Type; +const decodePrimeConversationAnchor = Schema.decodeUnknownOption(primeConversationAnchor); const unavailableSessionGoal: SessionGoalUpdatedPayload = { available: false, active: false, @@ -364,6 +398,12 @@ interface PrimeAgentDaemonSessionContext { session: ProviderSession; readonly scope: Scope.Closeable; readonly runtime: PrimeAgentDaemonSessionRuntime; + readonly rootConversationLeafId: string | undefined; + rootConversationCheckpointTurnCount: number | undefined; + readonly settledConversationLeafIds: Map; + rollbackSourceLeafId: string | undefined; + rollbackTargetLeafId: string | undefined; + rollbackQuarantined: boolean; readonly managedExtensionPath: string; readonly managedExtensionSource: string; managedPlanProjectionEnabled: boolean; @@ -816,6 +856,10 @@ export function makePrimeAgentDaemonAdapter( new Error("The Prime Agent runtime context does not own this daemon adapter."), ); } + const managedAbsoluteRollbackAvailable = + options?.recoveryManagedBuildId !== undefined && + typeof manager.bridge.DaemonAgentConnection.prototype.getState === "function" && + typeof manager.bridge.DaemonAgentConnection.prototype.navigateTree === "function"; const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; const serverConfig = yield* ServerConfig; @@ -855,6 +899,11 @@ export function makePrimeAgentDaemonAdapter( rawRecoveryLedger.markAdmitted(input, { commitGuard }).pipe(Effect.orDie), false, ), + markIdle: (input: Parameters[0]) => + guardGeneration( + rawRecoveryLedger.markIdle(input, { commitGuard }).pipe(Effect.orDie), + false, + ), updateTranscriptProgress: ( input: Parameters[0], ) => @@ -1126,12 +1175,14 @@ export function makePrimeAgentDaemonAdapter( context: PrimeAgentDaemonSessionContext, event: ProviderRuntimeEvent, ) => - offerRuntimeEvent({ - ...event, - // The immutable context owns the event even after its session map entry - // is deleted or replaced. Never infer incarnation from mutable routing. - sessionIncarnationId: context.sessionIncarnationId, - }); + context.rollbackQuarantined + ? Effect.void + : offerRuntimeEvent({ + ...event, + // The immutable context owns the event even after its session map entry + // is deleted or replaced. Never infer incarnation from mutable routing. + sessionIncarnationId: context.sessionIncarnationId, + }); const getThreadSemaphore = (threadId: string) => SynchronizedRef.modifyEffect(threadLocksRef, (current) => { @@ -1987,6 +2038,16 @@ export function makePrimeAgentDaemonAdapter( (turn.correlationId === undefined && turn.cancellationRequested)) ? { state: "cancelled" } : outcome; + if ( + managedAbsoluteRollbackAvailable && + context.session.runtimeMode === "full-access" && + context.runtime.conversationRollbackAvailable + ) { + const terminalLeaf = yield* context.runtime.inspectConversationLeaf.pipe(Effect.result); + if (terminalLeaf._tag === "Success") { + context.settledConversationLeafIds.set(turn.id, terminalLeaf.success); + } + } turn.pendingRunCompletionHandoff = undefined; if ( effectiveOutcome.state === "failed" && @@ -2066,15 +2127,25 @@ export function makePrimeAgentDaemonAdapter( status: "ready", updatedAt: yield* nowIso, }; - yield* Deferred.succeed(turn.completed, undefined).pipe(Effect.ignore); if (context.recoveryOwnerToken !== undefined && !context.stopRequested) { - context.stopRequested = true; - yield* Effect.forkDetach( - Effect.yieldNow.pipe( - Effect.andThen(withThreadLock(context.threadId, stopSessionInternal(context))), - ), - ); + const retainedForRollback = + managedAbsoluteRollbackAvailable && + context.runtime.conversationRollbackAvailable && + (yield* recoveryLedger!.markIdle({ + threadId: context.threadId, + ownerToken: context.recoveryOwnerToken, + updatedAt: yield* nowIso, + })); + if (!retainedForRollback) { + context.stopRequested = true; + yield* Effect.forkDetach( + Effect.yieldNow.pipe( + Effect.andThen(withThreadLock(context.threadId, stopSessionInternal(context))), + ), + ); + } } + yield* Deferred.succeed(turn.completed, undefined).pipe(Effect.ignore); return true; }); @@ -4559,14 +4630,18 @@ export function makePrimeAgentDaemonAdapter( snapshotMessageCount: runtime.initialSnapshot.state.messageCount, snapshotMessages: runtime.initialSnapshot.messages, }); - if (authority.turnId === null || !replay.valid) { + if ( + !replay.valid || + (authority.turnId === null && + (replay.backlog.length > 0 || runtime.inputAdmissionBusy)) + ) { return yield* new ProviderAdapterProcessError({ provider: PROVIDER, threadId: input.threadId, detail: "Prime Agent restart recovery could not prove complete event continuity.", }); } - recoveryBacklog = replay.backlog; + if (authority.turnId !== null) recoveryBacklog = replay.backlog; } const now = yield* nowIso; @@ -4575,7 +4650,10 @@ export function makePrimeAgentDaemonAdapter( const session: ProviderSession = { provider: PROVIDER, providerInstanceId: boundInstanceId, - status: recoveryStart?.kind === "adopt" ? "running" : "ready", + status: + recoveryStart?.kind === "adopt" && recoveryStart.authority.turnId !== null + ? "running" + : "ready", runtimeMode: input.runtimeMode, cwd, model, @@ -4598,6 +4676,13 @@ export function makePrimeAgentDaemonAdapter( session, scope: sessionScope, runtime, + rootConversationLeafId: + recoveryStart === undefined ? runtime.initialConversationLeafId : undefined, + rootConversationCheckpointTurnCount: undefined, + settledConversationLeafIds: new Map(), + rollbackSourceLeafId: undefined, + rollbackTargetLeafId: undefined, + rollbackQuarantined: false, managedExtensionPath, managedExtensionSource, managedPlanProjectionEnabled: true, @@ -4875,6 +4960,62 @@ export function makePrimeAgentDaemonAdapter( input: ProviderSendTurnInput, ) { if (!recoveryPlatformEligible) return; + const retained = sessions.get(input.threadId); + if ( + managedAbsoluteRollbackAvailable && + retained !== undefined && + retained.recoveryOwnerToken !== undefined && + !retained.stopped && + !retained.stopRequested && + retained.session.status === "ready" && + retained.activeTurn === undefined + ) { + const retainedAuthority = Option.getOrUndefined(yield* recoveryLedger!.get(input.threadId)); + if ( + retainedAuthority === undefined || + retainedAuthority.ownerToken !== retained.recoveryOwnerToken || + retainedAuthority.turnId !== null || + !retainedAuthority.terminalProjected || + !retainedAuthority.checkpointQuiesced + ) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "prepareTurnRecovery", + reason: "busy", + issue: "Prime Agent is still retaining the previous turn for exact recovery.", + }); + } + const restartInput = { + threadId: retained.threadId, + provider: PROVIDER, + providerInstanceId: boundInstanceId, + runtimeMode: retained.session.runtimeMode, + ...(retained.session.cwd === undefined ? {} : { cwd: retained.session.cwd }), + ...(retained.session.model === undefined + ? {} + : { + modelSelection: { + instanceId: boundInstanceId, + model: retained.session.model, + }, + }), + resumeCursor: retained.session.resumeCursor, + sessionIncarnationId: retained.sessionIncarnationId, + } as const; + const completion = yield* withThreadMutationLock( + input.threadId, + stopSessionInternal(retained), + ); + yield* Deferred.await(completion); + if (Option.isSome(yield* recoveryLedger!.get(input.threadId))) { + return yield* new ProviderAdapterProcessError({ + provider: PROVIDER, + threadId: input.threadId, + detail: "Prime Agent could not retire the previous exact recovery authority.", + }); + } + yield* startSession(restartInput); + } const plan = yield* withThreadMutationLock( input.threadId, Effect.gen(function* () { @@ -4965,7 +5106,6 @@ export function makePrimeAgentDaemonAdapter( let authority = Option.getOrUndefined(yield* recoveryLedger!.get(input.threadId)); const authorityMatches = (candidate: PrimeAgentRecoveryAuthority) => candidate.threadId === input.threadId && - candidate.turnId !== null && candidate.providerInstanceId === input.providerInstanceId && candidate.sessionIncarnationId === input.sessionIncarnationId && candidate.packageRoot === manager.bridge.packageRoot && @@ -5367,15 +5507,20 @@ export function makePrimeAgentDaemonAdapter( if (context === undefined || context.stopped || !context.recoveryPendingActivation) return; const turn = context.activeTurn; - if (turn === undefined) { + if ( + turn === undefined && + (context.session.status !== "ready" || context.recoveryBacklog.length > 0) + ) { return yield* new ProviderAdapterProcessError({ provider: PROVIDER, threadId, detail: "Recovered Prime Agent execution lost its admitted turn.", }); } - for (const message of context.recoveryBacklog) { - yield* publishDrafts(context, { _tag: "MessageCompleted", message }, turn); + if (turn !== undefined) { + for (const message of context.recoveryBacklog) { + yield* publishDrafts(context, { _tag: "MessageCompleted", message }, turn); + } } context.recoveryPendingActivation = false; context.eventFiber = yield* context.runtime.events.pipe( @@ -7665,6 +7810,326 @@ export function makePrimeAgentDaemonAdapter( .filter((context) => !context.stopRequested && !context.stopped) .map((context) => ({ ...context.session })), ); + const rollbackValidationError = (operation: string, issue: string) => + new ProviderAdapterValidationError({ + provider: PROVIDER, + operation, + issue, + }); + + const rollbackContext = ( + threadId: ThreadId, + operation: string, + ): Effect.Effect => + Effect.gen(function* () { + const context = sessions.get(threadId); + const generationCurrent = yield* commitGuard; + if ( + !generationCurrent || + !managedAbsoluteRollbackAvailable || + context === undefined || + context.stopped || + context.stopRequested || + context.session.runtimeMode !== "full-access" || + !context.runtime.conversationRollbackAvailable || + context.runtime.conversationRuntimeGeneration === undefined || + context.runtime.inputAdmissionBusy || + context.activeTurn !== undefined || + context.nativeRunActive || + context.nativeBashActive || + context.backgroundQuiescencePending || + context.nativeQueueActionActive || + context.inputQueue.steeringCount !== 0 || + context.inputQueue.followUpCount !== 0 || + context.pendingApprovals.size !== 0 || + context.pendingInteractions.size !== 0 || + context.activeNativeChildren.size !== 0 + ) { + return yield* rollbackValidationError( + operation, + "Exact Prime Agent conversation rollback is unavailable for this session.", + ); + } + return context; + }); + + const requireRollbackContextCurrent = Effect.fn( + "PrimeAgentDaemonAdapter.requireRollbackContextCurrent", + )(function* (context: PrimeAgentDaemonSessionContext, operation: string) { + const current = yield* rollbackContext(context.threadId, operation); + if (current !== context) { + return yield* rollbackValidationError( + operation, + "The exact Prime Agent conversation owner changed during rollback.", + ); + } + }); + + const anchorReceipt = ( + context: PrimeAgentDaemonSessionContext, + leafId: string, + binding?: ProviderConversationAnchorBinding, + ): ProviderConversationAnchorReceipt => { + const identity = { + version: 1 as const, + provider: "prime-agent" as const, + providerInstanceId: String(boundInstanceId), + runtimeGeneration: context.runtime.conversationRuntimeGeneration!, + sessionIncarnationId: String(context.sessionIncarnationId), + nativeSessionId: context.runtime.sessionId, + leafId, + }; + const anchor: Schema.Json = + binding === undefined ? identity : { ...identity, binding: { ...binding } }; + const digest = NodeCrypto.createHash("sha256") + .update(JSON.stringify(identity), "utf8") + .digest("hex"); + return { anchor, digest }; + }; + + const decodeBoundAnchor = ( + context: PrimeAgentDaemonSessionContext, + raw: Schema.Json, + operation: string, + ): Effect.Effect => { + const decoded = decodePrimeConversationAnchor(raw); + return Option.isSome(decoded) && + decoded.value.providerInstanceId === boundInstanceId && + decoded.value.runtimeGeneration === context.runtime.conversationRuntimeGeneration && + decoded.value.sessionIncarnationId === context.sessionIncarnationId && + decoded.value.nativeSessionId === context.runtime.sessionId + ? Effect.succeed(decoded.value) + : Effect.fail( + rollbackValidationError( + operation, + "The private Prime Agent conversation anchor is stale or invalid.", + ), + ); + }; + + const isAbsoluteRollbackAvailable = (threadId: ThreadId) => + rollbackContext(threadId, "absoluteConversationRollback.isAvailable").pipe( + Effect.as(true), + Effect.orElseSucceed(() => false), + ); + + const captureAbsoluteAnchor = Effect.fn("PrimeAgentDaemonAdapter.captureAbsoluteAnchor")( + function* (input: { + readonly threadId: ThreadId; + readonly binding: ProviderConversationAnchorBinding; + }) { + const context = yield* rollbackContext( + input.threadId, + "absoluteConversationRollback.captureAnchor", + ); + let leafId: string | undefined; + if (input.binding.kind === "checkpoint") { + if ( + input.binding.sourceRevision !== input.binding.checkpointTurnCount || + (input.binding.checkpointTurnCount === 0 && input.binding.turnId !== null) || + (input.binding.checkpointTurnCount > 0 && input.binding.turnId === null) + ) { + return yield* rollbackValidationError( + "absoluteConversationRollback.captureAnchor", + "The exact checkpoint binding is invalid.", + ); + } + leafId = + input.binding.turnId === null + ? undefined + : context.settledConversationLeafIds.get(TurnId.make(input.binding.turnId)); + if (leafId === undefined && context.rootConversationLeafId !== undefined) { + if (context.rootConversationCheckpointTurnCount === undefined) { + context.rootConversationCheckpointTurnCount = input.binding.checkpointTurnCount; + } + if (context.rootConversationCheckpointTurnCount === input.binding.checkpointTurnCount) { + leafId = context.rootConversationLeafId; + } + } + } else { + leafId = yield* context.runtime.inspectConversationLeaf.pipe( + Effect.mapError((error) => + runtimeOperationError(input.threadId, "capture-conversation-anchor", error), + ), + ); + } + if (leafId === undefined) { + return yield* rollbackValidationError( + "absoluteConversationRollback.captureAnchor", + "The exact Prime Agent conversation leaf is unavailable.", + ); + } + if (input.binding.kind === "source") { + context.rollbackSourceLeafId = leafId; + context.rollbackQuarantined = true; + yield* context.runtime + .prepareConversationRollback({ + desiredLeafId: leafId, + allowedSourceLeafId: leafId, + }) + .pipe( + Effect.mapError((error) => + runtimeOperationError(input.threadId, "prepare-conversation-rollback", error), + ), + ); + yield* requireRollbackContextCurrent( + context, + "absoluteConversationRollback.captureAnchor", + ); + } + return anchorReceipt(context, leafId, input.binding); + }, + ); + + const inspectAbsoluteAnchor = Effect.fn("PrimeAgentDaemonAdapter.inspectAbsoluteAnchor")( + function* (threadId: ThreadId) { + const context = yield* rollbackContext( + threadId, + "absoluteConversationRollback.inspectAnchor", + ); + const leafId = yield* context.runtime.inspectConversationLeaf.pipe( + Effect.mapError((error) => + runtimeOperationError(threadId, "inspect-conversation-anchor", error), + ), + ); + yield* requireRollbackContextCurrent(context, "absoluteConversationRollback.inspectAnchor"); + if ( + context.rollbackQuarantined && + leafId !== context.rollbackSourceLeafId && + leafId !== context.rollbackTargetLeafId + ) { + return yield* rollbackValidationError( + "absoluteConversationRollback.inspectAnchor", + "Prime Agent conversation state is outside the exact rollback boundary.", + ); + } + return anchorReceipt(context, leafId); + }, + ); + + const applyAbsoluteAnchor = Effect.fn("PrimeAgentDaemonAdapter.applyAbsoluteAnchor")(function* ( + threadId: ThreadId, + rawAnchor: Schema.Json, + ) { + const context = yield* rollbackContext(threadId, "absoluteConversationRollback.applyAnchor"); + const anchor = yield* decodeBoundAnchor( + context, + rawAnchor, + "absoluteConversationRollback.applyAnchor", + ); + const applyingSource = anchor.leafId === context.rollbackSourceLeafId; + const allowedSourceLeafId = applyingSource + ? context.rollbackTargetLeafId + : context.rollbackSourceLeafId; + if (allowedSourceLeafId === undefined) { + return yield* rollbackValidationError( + "absoluteConversationRollback.applyAnchor", + "The exact Prime Agent source leaf is unavailable.", + ); + } + if (!applyingSource) context.rollbackTargetLeafId = anchor.leafId; + context.rollbackQuarantined = true; + yield* context.runtime + .navigateConversationLeaf({ + desiredLeafId: anchor.leafId, + allowedSourceLeafId, + }) + .pipe( + Effect.mapError((error) => + runtimeOperationError(threadId, "apply-conversation-anchor", error), + ), + ); + yield* requireRollbackContextCurrent(context, "absoluteConversationRollback.applyAnchor"); + }); + + const prepareAbsoluteRecovery = Effect.fn("PrimeAgentDaemonAdapter.prepareAbsoluteRecovery")( + function* (input: { + readonly threadId: ThreadId; + readonly sourceAnchor: Schema.Json; + readonly desiredAnchor: Schema.Json; + readonly expectedAnchor: Schema.Json; + }) { + const context = yield* rollbackContext( + input.threadId, + "absoluteConversationRollback.prepareRecovery", + ); + const source = yield* decodeBoundAnchor( + context, + input.sourceAnchor, + "absoluteConversationRollback.prepareRecovery", + ); + const desired = yield* decodeBoundAnchor( + context, + input.desiredAnchor, + "absoluteConversationRollback.prepareRecovery", + ); + const expected = yield* decodeBoundAnchor( + context, + input.expectedAnchor, + "absoluteConversationRollback.prepareRecovery", + ); + if (expected.leafId !== source.leafId && expected.leafId !== desired.leafId) { + return yield* rollbackValidationError( + "absoluteConversationRollback.prepareRecovery", + "The recovered Prime Agent rollback boundary is invalid.", + ); + } + context.rollbackSourceLeafId = source.leafId; + context.rollbackTargetLeafId = desired.leafId; + context.rollbackQuarantined = true; + yield* context.runtime + .prepareConversationRollback({ + desiredLeafId: expected.leafId, + allowedSourceLeafId: expected.leafId === source.leafId ? desired.leafId : source.leafId, + }) + .pipe( + Effect.mapError((error) => + runtimeOperationError(input.threadId, "prepare-conversation-rollback", error), + ), + ); + yield* requireRollbackContextCurrent( + context, + "absoluteConversationRollback.prepareRecovery", + ); + }, + ); + + const releaseAbsoluteAnchor = Effect.fn("PrimeAgentDaemonAdapter.releaseAbsoluteAnchor")( + function* (threadId: ThreadId, rawAnchor: Schema.Json) { + const context = yield* rollbackContext( + threadId, + "absoluteConversationRollback.releaseAnchor", + ); + const anchor = yield* decodeBoundAnchor( + context, + rawAnchor, + "absoluteConversationRollback.releaseAnchor", + ); + yield* context.runtime + .releaseConversationRollback(anchor.leafId) + .pipe( + Effect.mapError((error) => + runtimeOperationError(threadId, "release-conversation-anchor", error), + ), + ); + yield* requireRollbackContextCurrent(context, "absoluteConversationRollback.releaseAnchor"); + context.rollbackQuarantined = false; + context.rollbackSourceLeafId = undefined; + context.rollbackTargetLeafId = undefined; + }, + ); + + const absoluteConversationRollback = managedAbsoluteRollbackAvailable + ? { + isAvailable: isAbsoluteRollbackAvailable, + captureAnchor: captureAbsoluteAnchor, + inspectAnchor: inspectAbsoluteAnchor, + applyAnchor: applyAbsoluteAnchor, + releaseAnchor: releaseAbsoluteAnchor, + prepareRecovery: prepareAbsoluteRecovery, + } + : undefined; + const hasSession: PrimeAgentAdapterShape["hasSession"] = (threadId) => Effect.sync(() => { const context = sessions.get(threadId); @@ -7791,8 +8256,11 @@ export function makePrimeAgentDaemonAdapter( provider: PROVIDER, capabilities: { sessionModelSwitch: "in-session", - conversationRollback: BUILT_IN_ADAPTER_CONVERSATION_ROLLBACK_MODES.primeDaemon, + conversationRollback: managedAbsoluteRollbackAvailable + ? "absolute" + : BUILT_IN_ADAPTER_CONVERSATION_ROLLBACK_MODES.primeDaemon, }, + ...(absoluteConversationRollback === undefined ? {} : { absoluteConversationRollback }), startSession, prepareTurnRecovery, recoverSession, diff --git a/apps/server/src/provider/prime/PrimeAgentDaemonBridge.ts b/apps/server/src/provider/prime/PrimeAgentDaemonBridge.ts index d0fcea275..3518c0069 100644 --- a/apps/server/src/provider/prime/PrimeAgentDaemonBridge.ts +++ b/apps/server/src/provider/prime/PrimeAgentDaemonBridge.ts @@ -189,6 +189,10 @@ export interface PrimeAgentDaemonAgentConnection { readonly getInitialSnapshot: () => Promise; readonly getRlmChildSnapshots?: () => Promise; readonly getState?: () => Promise; + readonly navigateTree?: ( + targetId: string, + options?: { readonly summarize?: boolean }, + ) => Promise; readonly promptAndWait: ( message: string, options?: PrimeAgentDaemonPromptOptions, diff --git a/apps/server/src/provider/prime/PrimeAgentDaemonSessionRuntime.test.ts b/apps/server/src/provider/prime/PrimeAgentDaemonSessionRuntime.test.ts index 5602ee605..2d2b97111 100644 --- a/apps/server/src/provider/prime/PrimeAgentDaemonSessionRuntime.test.ts +++ b/apps/server/src/provider/prime/PrimeAgentDaemonSessionRuntime.test.ts @@ -269,6 +269,11 @@ function fixture(options?: { readonly omitQueueMutation?: boolean; readonly queueMutationCapability?: boolean; readonly getStateImpl?: () => Promise; + readonly navigateTreeImpl?: ( + targetId: string, + options?: { readonly summarize?: boolean }, + ) => Promise; + readonly omitNavigateTree?: boolean; readonly setSteeringModeImpl?: (mode: "all" | "one-at-a-time") => Promise; readonly setFollowUpModeImpl?: (mode: "all" | "one-at-a-time") => Promise; readonly compactImpl?: () => Promise; @@ -475,6 +480,9 @@ function fixture(options?: { if (options?.omitRefine === true) { Object.defineProperty(this, "refine", { value: undefined }); } + if (options?.omitNavigateTree === true) { + Object.defineProperty(this, "navigateTree", { value: undefined }); + } if (options?.omitModelCatalog === true) { Object.defineProperty(this, "getModelCatalog", { value: undefined }); } @@ -549,6 +557,19 @@ function fixture(options?: { : undefined, ); } + navigateTree( + targetId: string, + navigateOptions?: { readonly summarize?: boolean }, + ): Promise { + captures.connectionCalls.push({ + method: "navigateTree", + args: [targetId, navigateOptions], + }); + return ( + options?.navigateTreeImpl?.(targetId, navigateOptions) ?? + Promise.resolve({ cancelled: false }) + ); + } promptAndWait( message: string, promptOptions?: PrimeAgentDaemonPromptOptions, @@ -11631,6 +11652,112 @@ describe("PrimeAgentDaemonSessionRuntime", () => { }); }), ); + + it.effect("navigates exact private leaves without summary and proves release", () => + Effect.gen(function* () { + let leafId = "leaf-source-private"; + const test = fixture({ + rawSnapshot: { + ...snapshot(), + state: { ...snapshot().state, leafId }, + }, + getStateImpl: () => + Promise.resolve({ + sessionId: "session-1", + activeSessionId: "active-secret-1", + leafId, + }), + navigateTreeImpl: (targetId) => { + leafId = targetId; + return Promise.resolve({ cancelled: false }); + }, + }); + yield* Effect.scoped( + Effect.gen(function* () { + const runtime = yield* test.make(); + expect(runtime.conversationRollbackAvailable).toBe(true); + expect(runtime.initialConversationLeafId).toBe("leaf-source-private"); + yield* runtime.navigateConversationLeaf({ + desiredLeafId: "leaf-target-private", + allowedSourceLeafId: "leaf-source-private", + }); + expect(yield* runtime.inspectConversationLeaf).toBe("leaf-target-private"); + yield* runtime.releaseConversationRollback("leaf-target-private"); + }), + ); + expect( + test.captures.connectionCalls.filter((call) => call.method === "navigateTree"), + ).toEqual([ + { + method: "navigateTree", + args: ["leaf-target-private", { summarize: false }], + }, + ]); + }), + ); + + it.effect("rejects a third leaf and keeps response-loss target proof inspectable", () => + Effect.gen(function* () { + let leafId = "leaf-source-private"; + const test = fixture({ + rawSnapshot: { + ...snapshot(), + state: { ...snapshot().state, leafId }, + }, + getStateImpl: () => + Promise.resolve({ + sessionId: "session-1", + activeSessionId: "active-secret-1", + leafId, + }), + navigateTreeImpl: (targetId) => { + leafId = targetId; + return Promise.reject(new Error("private daemon response was lost")); + }, + }); + yield* Effect.scoped( + Effect.gen(function* () { + const runtime = yield* test.make(); + const lostResponse = yield* runtime + .navigateConversationLeaf({ + desiredLeafId: "leaf-target-private", + allowedSourceLeafId: "leaf-source-private", + }) + .pipe(Effect.result); + expect(lostResponse._tag).toBe("Failure"); + expect(yield* runtime.inspectConversationLeaf).toBe("leaf-target-private"); + leafId = "leaf-third-private"; + const fenced = yield* runtime + .prepareConversationRollback({ + desiredLeafId: "leaf-target-private", + allowedSourceLeafId: "leaf-source-private", + }) + .pipe(Effect.result); + expect(fenced._tag).toBe("Failure"); + }), + ); + }), + ); + + it.effect("keeps exact navigation unavailable without the public navigateTree method", () => + Effect.gen(function* () { + const test = fixture({ + rawSnapshot: { + ...snapshot(), + state: { ...snapshot().state, leafId: "leaf-private" }, + }, + omitNavigateTree: true, + }); + yield* Effect.scoped( + Effect.gen(function* () { + const runtime = yield* test.make(); + expect(runtime.conversationRollbackAvailable).toBe(false); + const result = yield* runtime.inspectConversationLeaf.pipe(Effect.result); + expect(result._tag).toBe("Failure"); + }), + ); + }), + ); }); describe("Prime Agent live activity privacy boundary", () => { diff --git a/apps/server/src/provider/prime/PrimeAgentDaemonSessionRuntime.ts b/apps/server/src/provider/prime/PrimeAgentDaemonSessionRuntime.ts index 9b5bbca95..0b05e6d38 100644 --- a/apps/server/src/provider/prime/PrimeAgentDaemonSessionRuntime.ts +++ b/apps/server/src/provider/prime/PrimeAgentDaemonSessionRuntime.ts @@ -616,6 +616,22 @@ const refinementResultSchema = Schema.Struct({ appliedEdits: Schema.Array(Schema.Struct({ applied: Schema.Boolean })), scope: Schema.optional(Schema.Literals(["local", "global"])), }); +const privateConversationLeafIdSchema = Schema.String.check( + Schema.isMinLength(1), + Schema.isMaxLength(256), +); +const privateConversationLeafStateSchema = Schema.Struct({ + activeSessionId: Schema.optional(Schema.String), + sessionId: Schema.String, + leafId: Schema.NullOr(privateConversationLeafIdSchema), +}); +const privateConversationLeafSnapshotSchema = Schema.Struct({ + state: privateConversationLeafStateSchema, +}); +const privateConversationNavigationResultSchema = Schema.Struct({ + cancelled: Schema.Boolean, + aborted: Schema.optional(Schema.Boolean), +}); const decodeThinkingLevel = Schema.decodeUnknownOption(thinkingLevelSchema); const decodeServiceTier = Schema.decodeUnknownOption(serviceTierSchema); @@ -639,6 +655,15 @@ const decodeSessionStats = Schema.decodeUnknownOption(sessionStatsSchema); const decodeRlmMaxDepthStatus = Schema.decodeUnknownOption(rlmMaxDepthStatusSchema); const decodeAgentMessageReceipt = Schema.decodeUnknownOption(agentMessageReceiptSchema); const decodeRefinementResult = Schema.decodeUnknownOption(refinementResultSchema); +const decodePrivateConversationLeafState = Schema.decodeUnknownOption( + privateConversationLeafStateSchema, +); +const decodePrivateConversationLeafSnapshot = Schema.decodeUnknownOption( + privateConversationLeafSnapshotSchema, +); +const decodePrivateConversationNavigationResult = Schema.decodeUnknownOption( + privateConversationNavigationResultSchema, +); function managedPlanToolDefinitionMatches(value: unknown): boolean { const decoded = decodeManagedPlanToolDefinition(value); @@ -820,6 +845,9 @@ const runtimeErrorOperation = Schema.Literals([ "remove-only-input-queue-item", "set-input-queue-mode", "get-compaction-state", + "inspect-conversation-leaf", + "navigate-conversation-leaf", + "release-conversation-leaf", "compact", "refine-local-harness", "abort-compaction", @@ -1044,7 +1072,24 @@ export interface PrimeAgentDaemonSessionRuntime { readonly sessionId: string; readonly sessionFile: string; readonly activeSessionId: string; + /** Stable only within one compatible daemon supervisor generation. */ + readonly conversationRuntimeGeneration?: string; readonly initialSnapshot: PrimeAgentDaemonCanonicalSnapshot; + /** Private immutable root selected from the raw initial snapshot. */ + readonly initialConversationLeafId?: string; + readonly conversationRollbackAvailable: boolean; + readonly inspectConversationLeaf: Effect.Effect; + readonly navigateConversationLeaf: (input: { + readonly desiredLeafId: string; + readonly allowedSourceLeafId: string; + }) => Effect.Effect; + readonly prepareConversationRollback: (input: { + readonly desiredLeafId: string; + readonly allowedSourceLeafId: string; + }) => Effect.Effect; + readonly releaseConversationRollback: ( + expectedLeafId: string, + ) => Effect.Effect; readonly initialResources: PrimeAgentDaemonSessionResources; readonly initialAgentDepth: PrimeAgentDaemonAgentDepth; readonly initialInputQueue: PrimeAgentDaemonInputQueue; @@ -1557,6 +1602,13 @@ export const makePrimeAgentDaemonSessionRuntime = Effect.fn("makePrimeAgentDaemo >(); let needsResumeAfterAbort = false; let connectionGeneration = 0; + let conversationRollbackFence: + | { + readonly desiredLeafId: string; + readonly allowedSourceLeafId: string; + compromised: boolean; + } + | undefined; type RouteRetirementSignal = { readonly listeners: Set<() => void>; retired: boolean; @@ -2579,6 +2631,10 @@ export const makePrimeAgentDaemonSessionRuntime = Effect.fn("makePrimeAgentDaemo weight, ...(recoveryCursor === undefined ? {} : { recoveryCursor }), } satisfies QueuedRuntimeEvent; + if (conversationRollbackFence !== undefined) { + onCommit(); + return Effect.void; + } if (event._tag === "SessionClosed") { runtimeEventIngressFailed = true; settleReconnectResolution(connectionGeneration, false); @@ -4825,6 +4881,14 @@ export const makePrimeAgentDaemonSessionRuntime = Effect.fn("makePrimeAgentDaemo }), ), ); + const privateInitialLeaf = decodePrivateConversationLeafSnapshot(rawSnapshot); + const initialConversationLeafId = Option.flatMap(privateInitialLeaf, (snapshot) => + snapshot.state.sessionId === sessionId && + (snapshot.state.activeSessionId === undefined || + snapshot.state.activeSessionId === activeSessionId) + ? Option.fromNullishOr(snapshot.state.leafId) + : Option.none(), + ); const rawSnapshotWeight = boundedCorrelatedProofRouteWeight(rawSnapshot); if (rawSnapshotWeight > MAX_CORRELATED_PROOF_ROUTE_WEIGHT - initializationAcceptedEventWeight) { initializationOverflow = true; @@ -5161,6 +5225,176 @@ export const makePrimeAgentDaemonSessionRuntime = Effect.fn("makePrimeAgentDaemo ), ); + const conversationRuntimeGeneration = client.hello?.supervisorGeneration?.trim(); + const conversationRollbackAvailable = + input.requiredExtension === undefined && + conversationRuntimeGeneration !== undefined && + conversationRuntimeGeneration.length > 0 && + Predicate.isFunction(connection!.getState) && + Predicate.isFunction(connection!.navigateTree); + + const readConversationLeaf = Effect.fn( + "PrimeAgentDaemonSessionRuntime.inspectConversationLeaf", + )(function* () { + yield* ensureOpen("inspect-conversation-leaf"); + if (!conversationRollbackAvailable) { + return yield* runtimeError( + "inspect-conversation-leaf", + "incompatible-api", + "The installed Prime Agent connection does not support exact conversation navigation.", + ); + } + const getState = yield* requireMethod("inspect-conversation-leaf", connection!.getState); + const output = yield* Effect.tryPromise({ + try: () => getState.call(connection), + catch: () => + runtimeError( + "inspect-conversation-leaf", + "request-failed", + "Prime Agent conversation state could not be inspected.", + ), + }).pipe( + Effect.timeoutOrElse({ + duration: COMMAND_TIMEOUT_MS, + orElse: () => + runtimeError( + "inspect-conversation-leaf", + "request-timed-out", + "Prime Agent conversation inspection timed out.", + ), + }), + ); + const decoded = decodePrivateConversationLeafState(output); + if ( + Option.isNone(decoded) || + decoded.value.sessionId !== sessionId || + (decoded.value.activeSessionId !== undefined && + decoded.value.activeSessionId !== activeSessionId) || + decoded.value.leafId === null + ) { + return yield* runtimeError( + "inspect-conversation-leaf", + "invalid-response", + "Prime Agent returned an invalid or mismatched conversation state.", + ); + } + return decoded.value.leafId; + }); + + const prepareConversationRollback: PrimeAgentDaemonSessionRuntime["prepareConversationRollback"] = + Effect.fn("PrimeAgentDaemonSessionRuntime.prepareConversationRollback")(function* (input) { + const desiredLeafId = yield* validateNonEmpty( + "navigate-conversation-leaf", + "desiredLeafId", + input.desiredLeafId, + ); + const allowedSourceLeafId = yield* validateNonEmpty( + "navigate-conversation-leaf", + "allowedSourceLeafId", + input.allowedSourceLeafId, + ); + conversationRollbackFence = { + desiredLeafId, + allowedSourceLeafId, + compromised: false, + }; + const current = yield* readConversationLeaf(); + if (current !== desiredLeafId && current !== allowedSourceLeafId) { + conversationRollbackFence.compromised = true; + return yield* runtimeError( + "navigate-conversation-leaf", + "invalid-response", + "Prime Agent conversation state is outside the exact rollback boundary.", + ); + } + }); + + const navigateConversationLeaf: PrimeAgentDaemonSessionRuntime["navigateConversationLeaf"] = + Effect.fn("PrimeAgentDaemonSessionRuntime.navigateConversationLeaf")(function* (input) { + yield* prepareConversationRollback(input); + if (conversationRollbackFence?.compromised === true) { + return yield* runtimeError( + "navigate-conversation-leaf", + "invalid-response", + "Prime Agent conversation rollback authority is compromised.", + ); + } + const current = yield* readConversationLeaf(); + if (current === input.desiredLeafId) return; + if (current !== input.allowedSourceLeafId) { + conversationRollbackFence!.compromised = true; + return yield* runtimeError( + "navigate-conversation-leaf", + "invalid-response", + "Prime Agent conversation state is outside the exact rollback boundary.", + ); + } + const navigate = yield* requireMethod( + "navigate-conversation-leaf", + connection!.navigateTree, + ); + const output = yield* Effect.tryPromise({ + try: () => navigate.call(connection, input.desiredLeafId, { summarize: false }), + catch: () => + runtimeError( + "navigate-conversation-leaf", + "request-failed", + "Prime Agent conversation navigation outcome is unavailable.", + ), + }).pipe( + Effect.timeoutOrElse({ + duration: COMMAND_TIMEOUT_MS, + orElse: () => + runtimeError( + "navigate-conversation-leaf", + "request-timed-out", + "Prime Agent conversation navigation timed out.", + ), + }), + ); + const result = decodePrivateConversationNavigationResult(output); + if (Option.isNone(result) || result.value.cancelled || result.value.aborted === true) { + return yield* runtimeError( + "navigate-conversation-leaf", + "invalid-response", + "Prime Agent did not confirm exact conversation navigation.", + ); + } + const inspected = yield* readConversationLeaf(); + if (inspected !== input.desiredLeafId) { + if (inspected !== input.allowedSourceLeafId) + conversationRollbackFence!.compromised = true; + return yield* runtimeError( + "navigate-conversation-leaf", + "invalid-response", + "Prime Agent did not reach the exact requested conversation state.", + ); + } + }); + + const releaseConversationRollback: PrimeAgentDaemonSessionRuntime["releaseConversationRollback"] = + Effect.fn("PrimeAgentDaemonSessionRuntime.releaseConversationRollback")( + function* (expectedLeafId) { + yield* validateNonEmpty("release-conversation-leaf", "expectedLeafId", expectedLeafId); + if (conversationRollbackFence?.compromised === true) { + return yield* runtimeError( + "release-conversation-leaf", + "invalid-response", + "Prime Agent conversation rollback authority is compromised.", + ); + } + const current = yield* readConversationLeaf(); + if (current !== expectedLeafId) { + return yield* runtimeError( + "release-conversation-leaf", + "invalid-response", + "Prime Agent did not preserve the committed conversation state.", + ); + } + conversationRollbackFence = undefined; + }, + ); + const agentMessageAvailable = input.requiredExtension === undefined && Predicate.isFunction(connection!.sendAgentMessage); const rlmQuiescenceAvailable = Predicate.isFunction(connection!.waitForHeadlessCompletion); @@ -8311,7 +8545,16 @@ export const makePrimeAgentDaemonSessionRuntime = Effect.fn("makePrimeAgentDaemo sessionId, sessionFile, activeSessionId, + ...(conversationRuntimeGeneration === undefined ? {} : { conversationRuntimeGeneration }), initialSnapshot: initialEvent, + ...(Option.isNone(initialConversationLeafId) + ? {} + : { initialConversationLeafId: initialConversationLeafId.value }), + conversationRollbackAvailable, + inspectConversationLeaf: readConversationLeaf(), + navigateConversationLeaf, + prepareConversationRollback, + releaseConversationRollback, initialResources, initialAgentDepth, initialInputQueue, diff --git a/apps/server/src/provider/prime/PrimeAgentRecoveryLedger.test.ts b/apps/server/src/provider/prime/PrimeAgentRecoveryLedger.test.ts index 0c9c960f0..dd139cda6 100644 --- a/apps/server/src/provider/prime/PrimeAgentRecoveryLedger.test.ts +++ b/apps/server/src/provider/prime/PrimeAgentRecoveryLedger.test.ts @@ -95,6 +95,26 @@ const admit = (ledger: PrimeAgentRecoveryLedgerShape) => }); layer("PrimeAgentRecoveryLedger", (it) => { + it.effect("retains settled exact rollback authority as idle and adoptable", () => + Effect.gen(function* () { + yield* resetLedger; + const ledger = yield* make; + yield* ledger.putPrepared(authority); + assert.isTrue(yield* admit(ledger)); + assert.isTrue( + yield* ledger.markIdle({ + threadId: authority.threadId, + ownerToken: authority.ownerToken, + updatedAt: "2026-01-01T00:00:01.500Z", + }), + ); + assert.isNull(Option.getOrThrow(yield* ledger.get(authority.threadId)).turnId); + const retained = yield* ledger.listActive(); + assert.equal(retained.length, 1); + assert.isNull(retained[0]?.turnId); + }), + ); + it.effect( "keeps prior authority while one stable adoption route advances through every phase", () => diff --git a/apps/server/src/provider/prime/PrimeAgentRecoveryLedger.ts b/apps/server/src/provider/prime/PrimeAgentRecoveryLedger.ts index 03566bf2e..05599ed6c 100644 --- a/apps/server/src/provider/prime/PrimeAgentRecoveryLedger.ts +++ b/apps/server/src/provider/prime/PrimeAgentRecoveryLedger.ts @@ -166,6 +166,15 @@ export interface PrimeAgentRecoveryLedgerShape { }, options?: PrimeAgentRecoveryCommitOptions, ) => Effect.Effect; + /** Retains exact native ownership after a settled turn for private rollback. */ + readonly markIdle: ( + input: { + readonly threadId: string; + readonly ownerToken: string; + readonly updatedAt: string; + }, + options?: PrimeAgentRecoveryCommitOptions, + ) => Effect.Effect; readonly discardPrepared: ( input: { readonly threadId: string; readonly ownerToken: string }, options?: PrimeAgentRecoveryCommitOptions, @@ -566,6 +575,15 @@ export const make = Effect.gen(function* () { options, ); + const markIdle: PrimeAgentRecoveryLedgerShape["markIdle"] = (input, options) => + conditionalUpdate( + "markIdle", + `UPDATE prime_agent_recovery_ledger SET turn_id=NULL, state='active', updated_at=? + WHERE thread_id=? AND owner_token=? AND state='active' RETURNING thread_id`, + [input.updatedAt, input.threadId, input.ownerToken], + options, + ); + const discardPrepared: PrimeAgentRecoveryLedgerShape["discardPrepared"] = (input, options) => conditionalUpdate( "discardPrepared", @@ -819,6 +837,7 @@ export const make = Effect.gen(function* () { get, listActive, markAdmitted, + markIdle, discardPrepared, updateTranscriptProgress, claim, diff --git a/apps/server/src/rollback/RollbackAdmission.test.ts b/apps/server/src/rollback/RollbackAdmission.test.ts index 4dbcef8b2..322402763 100644 --- a/apps/server/src/rollback/RollbackAdmission.test.ts +++ b/apps/server/src/rollback/RollbackAdmission.test.ts @@ -124,6 +124,7 @@ const makeHarness = (options: HarnessOptions = {}) => { captureConversationAnchor: () => Effect.succeed({ anchor: {}, digest: "source" }), inspectConversationAnchor: () => Effect.succeed({ anchor: {}, digest: "source" }), applyConversationAnchor: () => Effect.void, + releaseConversationAnchor: () => Effect.void, getSessionInputQueue: () => Effect.succeed({ steeringCount: options.queueCount ?? 0, @@ -154,6 +155,11 @@ const makeHarness = (options: HarnessOptions = {}) => { Option.some({ threadId, checkpointTurnCount: input.checkpointTurnCount, + turnId: + input.checkpointTurnCount === 0 + ? null + : TurnId.make(`turn-${input.checkpointTurnCount}`), + sourceRevision: input.checkpointTurnCount, providerInstanceId, sessionIncarnationId, checkpointRef: input.checkpointTurnCount === 0 ? baselineRef : turnOneRef, diff --git a/apps/server/src/rollback/RollbackAdmission.ts b/apps/server/src/rollback/RollbackAdmission.ts index 6e4cc2d86..54a76e3b6 100644 --- a/apps/server/src/rollback/RollbackAdmission.ts +++ b/apps/server/src/rollback/RollbackAdmission.ts @@ -79,6 +79,7 @@ export const make = Effect.gen(function* () { provider.captureConversationAnchor === undefined || provider.inspectConversationAnchor === undefined || provider.applyConversationAnchor === undefined || + provider.releaseConversationAnchor === undefined || !(yield* provider .hasAbsoluteConversationRollback(command.threadId) .pipe( @@ -218,7 +219,14 @@ export const make = Effect.gen(function* () { .pipe( Effect.mapError(() => invariant("The private target provider anchor is unavailable.")), ); - if (Option.isNone(desired) || desired.value.checkpointOid !== targetIdentity.oid) { + const targetTurnId = targetSummary?.turnId ?? null; + if ( + Option.isNone(desired) || + desired.value.sourceRevision !== command.turnCount || + desired.value.turnId !== targetTurnId || + desired.value.checkpointRef !== targetCheckpointRef || + desired.value.checkpointOid !== targetIdentity.oid + ) { return yield* invariant( "The private target provider anchor does not match the immutable checkpoint.", ); @@ -256,6 +264,8 @@ export const make = Effect.gen(function* () { workspaceCwd: sessionIdentity.cwd, sourceRevision, targetRevision: command.turnCount, + sourceTurnId: sourceSummary.turnId, + targetTurnId, sourceCheckpointRef, sourceCheckpointOid: sourceIdentity.oid, targetCheckpointRef, diff --git a/apps/server/src/rollback/RollbackSagaRunner.test.ts b/apps/server/src/rollback/RollbackSagaRunner.test.ts index 0943809f6..8f9434377 100644 --- a/apps/server/src/rollback/RollbackSagaRunner.test.ts +++ b/apps/server/src/rollback/RollbackSagaRunner.test.ts @@ -51,6 +51,8 @@ const makeState = (operationId: string): RollbackSagaState => ({ workspaceCwd: "/workspace/fake", sourceRevision: 2, targetRevision: 1, + sourceTurnId: null, + targetTurnId: null, sourceCheckpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-runner/turn/2"), sourceCheckpointOid: "a".repeat(40), targetCheckpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-runner/turn/1"), @@ -97,6 +99,7 @@ const makeEnvironment = ( }; let lease = true; let providerDigest = "provider-source"; + let currentProviderMode = providerMode; let workspaceDigest = "workspace-source"; let preimageCleaned = false; let anchorsDeleted = false; @@ -211,6 +214,7 @@ const makeEnvironment = ( anchor: { leafId: providerDigest }, digest: providerDigest, })), + releaseConversationAnchor: () => Effect.void, applyConversationAnchor: (input: { readonly anchor: unknown }) => Effect.suspend(() => { const isTarget = @@ -219,7 +223,7 @@ const makeEnvironment = ( providerDigest = "provider-source"; return Effect.void; } - switch (providerMode) { + switch (currentProviderMode) { case "success": providerDigest = "provider-target"; return Effect.void; @@ -336,6 +340,12 @@ const makeEnvironment = ( return { repository, makeRunner, + setProviderDigest: (digest: string) => { + providerDigest = digest; + }, + setProviderMode: (mode: ProviderMode) => { + currentProviderMode = mode; + }, snapshot: () => ({ record, lease, @@ -413,6 +423,65 @@ it.effect("reconciles an unknown provider result by inspecting the exact target" }), ); +it.effect("reapplies the target after a compatible reconnect returns to source", () => + Effect.gen(function* () { + const operationId = "operation-reconnect-source"; + const environment = makeEnvironment(operationId); + const interrupted = yield* environment.makeRunner("persisted:provider-applied"); + yield* runInterrupted(interrupted, operationId); + assert.equal(environment.snapshot().record.state.phase, "provider-applied"); + + environment.setProviderDigest("provider-source"); + yield* environment.repository.clearOwnersForStartup(); + const recovered = yield* environment.makeRunner(); + yield* recovered.run(operationId, true); + const snapshot = environment.snapshot(); + assert.equal(snapshot.providerDigest, "provider-target"); + assert.equal(snapshot.record.state.phase, "complete"); + assert.equal(snapshot.projectionCommits, 1); + }), +); + +it.effect("fails closed when reconnect inspection finds a third provider leaf", () => + Effect.gen(function* () { + const operationId = "operation-reconnect-third-leaf"; + const environment = makeEnvironment(operationId); + const interrupted = yield* environment.makeRunner("persisted:provider-applied"); + yield* runInterrupted(interrupted, operationId); + environment.setProviderDigest("provider-third"); + yield* environment.repository.clearOwnersForStartup(); + + const recovered = yield* environment.makeRunner(); + yield* recovered.run(operationId, true); + const snapshot = environment.snapshot(); + assert.equal(snapshot.record.state.phase, "manual-recovery"); + assert.isFalse(snapshot.projectionCommitted); + }), +); + +it.effect("enters manual recovery when the target cannot be reproved after projection", () => + Effect.gen(function* () { + const operationId = "operation-post-projection-source"; + const environment = makeEnvironment(operationId); + const interrupted = yield* environment.makeRunner("persisted:projection-committed"); + yield* runInterrupted(interrupted, operationId); + assert.equal(environment.snapshot().record.state.phase, "projection-committed"); + assert.equal(environment.snapshot().projectionCommits, 1); + + environment.setProviderDigest("provider-source"); + environment.setProviderMode("stayed-source"); + yield* environment.repository.clearOwnersForStartup(); + const recovered = yield* environment.makeRunner(); + yield* recovered.run(operationId, true); + + const snapshot = environment.snapshot(); + assert.equal(snapshot.record.state.phase, "manual-recovery"); + assert.equal(snapshot.record.state.attempt, 2); + assert.equal(snapshot.projectionCommits, 1); + assert.isFalse(snapshot.record.terminal); + }), +); + it.effect("compensates workspace and provider when the provider stays at source", () => Effect.gen(function* () { const environment = makeEnvironment("operation-compensate", "stayed-source"); diff --git a/apps/server/src/rollback/RollbackSagaRunner.ts b/apps/server/src/rollback/RollbackSagaRunner.ts index f4efecaf9..481f94f87 100644 --- a/apps/server/src/rollback/RollbackSagaRunner.ts +++ b/apps/server/src/rollback/RollbackSagaRunner.ts @@ -47,6 +47,7 @@ export const make = Effect.gen(function* () { const captureConversationAnchor = provider.captureConversationAnchor; const inspectConversationAnchor = provider.inspectConversationAnchor; const applyConversationAnchor = provider.applyConversationAnchor; + const releaseConversationAnchor = provider.releaseConversationAnchor; const ownerId = yield* (yield* Crypto.Crypto).randomUUIDv4; const nowIso = Effect.map(DateTime.now, DateTime.formatIso); @@ -193,6 +194,18 @@ export const make = Effect.gen(function* () { ); return; } + if (releaseConversationAnchor === undefined || record.state.sourceAnchor === null) { + yield* manual(record, "provider-quarantine-release-unavailable"); + return; + } + const releasedProvider = yield* releaseConversationAnchor({ + threadId: record.state.threadId, + anchor: record.state.sourceAnchor, + }).pipe(Effect.result); + if (releasedProvider._tag === "Failure") { + yield* manual(record, "provider-quarantine-release-unproved"); + return; + } const preimageForCleanup = privatePreimage(record.state); if (preimageForCleanup !== null) { const cleaned = yield* workspace.cleanupPreimage(preimageForCleanup).pipe(Effect.result); @@ -223,7 +236,16 @@ export const make = Effect.gen(function* () { const state = record.state; switch (state.phase) { case "source-anchor-capture-started": { - const source = yield* captureConversationAnchor!(state.threadId).pipe(Effect.result); + const source = yield* captureConversationAnchor!({ + threadId: state.threadId, + binding: { + kind: "source", + sourceRevision: state.sourceRevision, + checkpointRef: state.sourceCheckpointRef, + checkpointOid: state.sourceCheckpointOid, + turnId: state.sourceTurnId, + }, + }).pipe(Effect.result); yield* after("side-effect:source-anchor-captured", state.operationId); if (source._tag === "Failure") return yield* compensate(record, "source-anchor-capture-failed"); @@ -358,16 +380,48 @@ export const make = Effect.gen(function* () { checkpointOid: state.targetCheckpointOid, }) .pipe(Effect.result); - const providerReceipt = yield* inspectConversationAnchor!(state.threadId).pipe( + let providerReceipt = yield* inspectConversationAnchor!(state.threadId).pipe( Effect.result, ); + if ( + providerReceipt._tag === "Success" && + providerReceipt.success.digest === state.sourceAnchorDigest && + state.desiredAnchor !== null + ) { + yield* applyConversationAnchor!({ + threadId: state.threadId, + anchor: state.desiredAnchor, + }).pipe(Effect.result); + yield* after("side-effect:provider-target-reapplied", state.operationId); + providerReceipt = yield* inspectConversationAnchor!(state.threadId).pipe(Effect.result); + } + if ( + providerReceipt._tag === "Success" && + providerReceipt.success.digest !== state.desiredAnchorDigest && + providerReceipt.success.digest !== state.sourceAnchorDigest + ) { + return yield* manual(record, "provider-anchor-neither-source-nor-target"); + } if ( workspaceReceipt._tag === "Failure" || workspaceReceipt.success.digest !== state.workspaceReceiptDigest || - providerReceipt._tag === "Failure" || - providerReceipt.success.digest !== state.desiredAnchorDigest - ) + providerReceipt._tag === "Failure" + ) { return yield* manual(record, "precommit-postcondition-lost"); + } + if (providerReceipt.success.digest !== state.desiredAnchorDigest) { + if (state.attempt + 1 < MAX_PROVIDER_TARGET_ATTEMPTS) { + const next = yield* update(record, { + phase: "provider-applied", + attempt: state.attempt + 1, + lastErrorCode: "provider-target-stayed-source-after-reconnect", + }); + if (Option.isNone(next)) return; + record = next.value; + continue; + } + return yield* compensate(record, "provider-target-retry-exhausted"); + } const next = yield* update(record, { phase: "projection-commit-started" }); if (Option.isNone(next)) return; record = next.value; @@ -399,6 +453,44 @@ export const make = Effect.gen(function* () { continue; } case "projection-committed": { + let providerReceipt = yield* inspectConversationAnchor!(state.threadId).pipe( + Effect.result, + ); + if ( + providerReceipt._tag === "Success" && + providerReceipt.success.digest === state.sourceAnchorDigest && + state.desiredAnchor !== null + ) { + yield* applyConversationAnchor!({ + threadId: state.threadId, + anchor: state.desiredAnchor, + }).pipe(Effect.result); + yield* after("side-effect:provider-target-reapplied", state.operationId); + providerReceipt = yield* inspectConversationAnchor!(state.threadId).pipe(Effect.result); + } + if ( + providerReceipt._tag === "Success" && + providerReceipt.success.digest !== state.desiredAnchorDigest && + providerReceipt.success.digest !== state.sourceAnchorDigest + ) { + return yield* manual(record, "provider-anchor-neither-source-nor-target"); + } + if ( + providerReceipt._tag !== "Success" || + providerReceipt.success.digest !== state.desiredAnchorDigest + ) { + if (state.attempt + 1 < MAX_PROVIDER_TARGET_ATTEMPTS) { + const retried = yield* update(record, { + phase: "projection-committed", + attempt: state.attempt + 1, + lastErrorCode: "provider-target-unproved-after-projection", + }); + if (Option.isNone(retried)) return; + record = retried.value; + continue; + } + return yield* manual(record, "provider-target-unproved-after-projection"); + } const next = yield* update(record, { phase: "cleanup-started", cleanup: "running" }); if (Option.isNone(next)) return; record = next.value; @@ -423,6 +515,11 @@ export const make = Effect.gen(function* () { threadId: state.threadId, checkpointTurnCount: state.targetRevision, }); + if (state.desiredAnchor === null) return yield* Effect.die("provider anchor missing"); + yield* releaseConversationAnchor!({ + threadId: state.threadId, + anchor: state.desiredAnchor, + }); const preimage = privatePreimage(state); if (preimage !== null) yield* workspace.cleanupPreimage(preimage); }).pipe(Effect.result); diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index bbcc66466..908b38c44 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -21,6 +21,7 @@ import * as Path from "effect/Path"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; +import type { Json } from "effect/Schema"; import * as Scope from "effect/Scope"; import * as ServerConfig from "./config.ts"; @@ -37,6 +38,7 @@ import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; import * as ProviderService from "./provider/Services/ProviderService.ts"; import * as ProviderSessionDirectory from "./provider/Services/ProviderSessionDirectory.ts"; import * as ProviderSessionReaper from "./provider/Services/ProviderSessionReaper.ts"; +import { RollbackSagaRepository } from "./persistence/Services/RollbackSagas.ts"; import { forkParked } from "./serverActivation.ts"; import * as ServiceLauncherClient from "./cloud/serviceLauncherClient.ts"; import { @@ -302,11 +304,43 @@ export const reconcileProviderSessions = Effect.gen(function* () { const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService; const providerService = yield* ProviderService.ProviderService; const query = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; + const rollbackRepository = yield* Effect.serviceOption(RollbackSagaRepository); - // Prime restart adoption must install exact incarnation fencing and release retained - // replay before generic orphan settlement can observe the thread as dead. + // Prime restart adoption installs rollback quarantine before exact-incarnation + // fencing releases any retained native frames. if (providerService.recoverRestartSessions !== undefined) { - yield* providerService.recoverRestartSessions(); + const unrecoverableAbsoluteRollbacks = new Set(); + const pendingAbsoluteRollbacks = new Map< + ThreadId, + { readonly sourceAnchor: Json; readonly desiredAnchor: Json; readonly expectedAnchor: Json } + >(); + const pendingRecords = Option.isSome(rollbackRepository) + ? yield* rollbackRepository.value.listNonterminal() + : []; + for (const record of pendingRecords) { + const { sourceAnchor, desiredAnchor } = record.state; + if (sourceAnchor === null || desiredAnchor === null) { + unrecoverableAbsoluteRollbacks.add(record.state.threadId); + continue; + } + const expectTarget = [ + "workspace-applied", + "provider-apply-started", + "provider-applied", + "projection-commit-started", + "projection-committed", + "cleanup-started", + ].includes(record.state.phase); + pendingAbsoluteRollbacks.set(record.state.threadId, { + sourceAnchor, + desiredAnchor, + expectedAnchor: expectTarget ? desiredAnchor : sourceAnchor, + }); + } + yield* providerService.recoverRestartSessions({ + pendingAbsoluteRollbacks, + unrecoverableAbsoluteRollbacks, + }); } const liveThreadIds = new Set( (yield* providerService.listSessions()).map((session) => session.threadId), diff --git a/docs/internals/prime-agent-daemon-parity.md b/docs/internals/prime-agent-daemon-parity.md index 15c273a7d..6e2587cdf 100644 --- a/docs/internals/prime-agent-daemon-parity.md +++ b/docs/internals/prime-agent-daemon-parity.md @@ -9,6 +9,8 @@ raw method tunnel. Pylon dynamically probes optional methods on the installed package. All native identifiers, paths, private prompts, diagnostics, and result envelopes terminate at the Prime adapter boundary. +Exact checkpoint rollback is a server-private exception to the otherwise deferred history surface. The managed native adapter uses only public `getState()` and `navigateTree()` calls. It stores opaque leaf anchors in private rollback tables, never in contracts or public events. Availability is per thread and requires an idle, quiescent, full-access session with matching provider, runtime-generation, session-incarnation, and native-session identity. Navigation never summarizes an abandoned branch. A nonterminal saga quarantines Prime output, accepts only its source or target leaf after reconnect, and proves the exact committed target before release. Managed recoverable ownership remains idle and adoptable after a settled turn; the next turn cannot rotate that owner until terminal projection and checkpoint quiescence are durable. + Prime Agent 0.8.1 keeps daemon protocol 7 and schema 22 unchanged from 0.8.0, which advanced the schema from 16. Pylon supplies the fresh owner runtime configuration required to recover a client-owned worker, refreshes the RLM roster from the authoritative snapshot method when available, and retains @@ -292,45 +294,46 @@ both emit a fixed provider-neutral status before an authoritative terminal event assistant text follows the latest tool boundary. Reasoning, tool data, native errors, and identifiers are never used to synthesize assistant prose. -| Public API outcome | Pylon status | Decision | -| -------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -| `attach`, root `subscribe`, `getState`, `getInitialSnapshot`, `dispose` | Integrated internally | Own one private daemon session per active Pylon thread, resume exact verified identity, reconnect through public snapshots/events, and close with the thread scope. On 0.8.1 recovery Pylon resupplies the owner `cwd`, session/agent directories, execution policy, and current Pylon-selected model and thinking level; successful model and thinking mutations update that transient recovery context, which Prime never persists. The root connection's `getMessages` transcript API is never called; Pylon's event-sourced transcript remains authoritative. These are lifecycle primitives, not client RPCs. | -| `promptAndWait`, `steer`, `followUp`, `abort` | Integrated | Typed turn admission, active steering, explicit follow-up, queue modes, and interruption remain under Pylon turn ownership. On 0.8.1 the prompt boundary is followed by the public RLM-quiescence barrier so descendant-triggered parent continuations remain in the same canonical turn. Every admitted steer or follow-up advances a correlation generation and rearms that boundary. An in-process transport reconnect preserves the same Pylon turn only after complete native replay or exact public-snapshot reconciliation; ambiguity fails the turn and disposes the native session. Admission is never retried. | -| `supportsNegotiatedCapability`, `submitCorrelatedPrompt`, `cancelPromptLifecycle`, `getPromptLifecycles` | Integrated behind frozen SDK feature and post-attach proof | The exact frozen `negotiated_daemon_session_capabilities_v1` feature enables the proof accessor check only after attach. A false initial proof selects typed-busy ordinary admission; a true proof latches strict correlated mode. Consumer proof epochs fence every asynchronous lifecycle call, worker snapshot, and resync. Proof loss is terminal and never resubmits or downgrades an owned prompt. Direct replacement currently invalidates proof before publication and therefore closes the strict session. Method presence, version, schema, hello offers, and mutable feature registries never enable this path. | -| `prompt`, `waitForIdle` | Intentionally redundant | Pylon uses the cancellable `promptAndWait` admission path and daemon events for exact turn settlement. A second fire-and-observe prompt path or an unscoped idle waiter would weaken turn and checkpoint ownership. | -| `getQueue`, `clearQueue`, `setSteeringMode`, `setFollowUpMode` | Integrated | Expose only counts and delivery modes. Queued text stays private. | -| `mutateQueuedMessage` | Partially integrated | Pylon exposes sole-lane deletion only. Preview text stays server-private. A serialized, non-recovering compare-delete is followed by reconciliation; ambiguous mutations are never retried. Multi-item delete, move, and replace remain unavailable without opaque IDs or revisions. | -| `abortAndClearQueue` | Intentionally folded into Stop | The public operation combines interruption and queue deletion. Pylon keeps non-interrupting Clear and authoritative Stop separate so the reverse state is unambiguous. Prime Agent 0.8.1 suspends queued-input admission when aborting, so Pylon explicitly resumes that scheduler before reusing the session. | -| `setModel`, `setThinkingLevel`, `setServiceTier` | Integrated | Exact selection is owned by the durable thread model projection and reconciled against sanitized session state. | -| `getAvailableModels`, `getModelCatalog` | Integrated provider-catalog and auth-readiness enrichment | Attached sessions use `getModelCatalog`, with `getAvailableModels` as fallback. Strict decoding maps configured models into the existing provider snapshot and reports authenticated readiness only from a healthy current catalog containing a configured native provider. Empty catalogs and non-ready probes leave authentication unknown. This does not verify live network access or expose credential data. Failed or late reads keep the last good list without letting it override probe health; discovery never creates a session. | -| `cycleModel`, `cycleThinkingLevel`, `setScopedModels` | Intentionally redundant | Pylon already has an exact multi-client model picker. Prime's scoped list is memory-only and native cycling cannot atomically update Pylon's durable selection, so exposing both would create split-brain state. | -| `setTransport` | Intentionally excluded | Transport is environment/provider plumbing owned by Pylon, not a per-thread user setting. Prime does not expose authoritative current transport in session state. | -| `getSessionStats` | Integrated | Only bounded context usage and finite reported turn-cost outcomes cross the boundary. | -| `compact`, `abortCompaction`, `setAutoCompactionEnabled` | Integrated | Manual compaction is argument-free; summaries, instructions, paths, and native results are discarded. Automatic state is reconciled before publication. Prime suspends queued session input when compaction aborts the active run, including when compaction is declined, so the next prompt first performs the same exact-session queue resume used after Stop and fails before admission if that resume cannot be proved. | -| `refine({ global: false })` | Integrated | Explicit local-only refinement accepts no instructions or rollback identity. RPC success contains aggregate edit counts only. Timeout/rejection is outcome-unknown and is never retried. | -| `abortBranchSummary` | Intentionally folded into Stop | Pylon does not start a standalone native branch-summary operation; stopping the owning turn remains authoritative. | -| `setAutoRetryEnabled`, `abortRetry` | Deferred | Retry lifecycle is observed safely, but enabled state has no authoritative readback and the setter writes shared provider settings. Stop already cancels the owning turn. A distinct retry control needs truthful session state and receipts. | -| `reload`, `getCommands` | Integrated | Full-access sessions can reload while idle and show bounded safe command metadata. Before and after reload, Pylon verifies its generated source plus the managed extension path, marker, and exact plan-tool definition. Supervised sessions fail closed. | -| `acquireSessionInputPause` | Intentionally redundant | Every Pylon prompt and resource reload is serialized by the per-thread adapter lock, and reload is admitted only while the owned session is idle. Pylon's scoped MCP server is attached before the first snapshot and released only after turn ownership ends, so it is never replaced during live input. Holding a native lease would add reconnect failure modes without fencing additional work. Revisit if Pylon supports live MCP configuration changes. | -| `supportsAcpMcpServers`, `replaceAcpMcpServers`, `releaseAcpMcpServers` | Integrated with scoped Pylon ownership | `McpProviderSession` is the single per-thread source of truth. Before the first daemon snapshot, Pylon replaces one `t3-code` HTTP server under the stable owner `pylon:` and fails closed if Prime cannot own it. After a daemon reconnect, Pylon reclaims that ownership before publishing the resynced session; if it cannot, the session closes instead of continuing without browser tools. Session teardown releases only that owner and server name before disposing the connection. ACP fallback sends the same scoped server in `session/new`. Browser-disabled sessions send nothing, and Prime-owned MCP settings and catalogs remain private. | -| `getResourceSnapshot` | Partially integrated by safe outcome | Commands and safe skill/prompt metadata are decoded internally. Native paths, diagnostics, extensions, themes, packages, and MCP configuration are not sent to clients. | -| `respondToExtensionUiRequest` | Partially integrated by safe outcome | Select, confirm, and input dialogs plus bounded notifications, status, and widgets are correlated without exposing native request envelopes. Submitted free-form input uses a transient provider RPC and is redacted from durable activities. Editor replacement is cancelled because its prefill may contain sensitive model or tool material that cannot safely enter Pylon's synchronized event stream. | -| `getRlmChildSnapshots`, `getRlmMaxDepthStatus`, `setRlmMaxDepth`, `cancelRlmChild`, `sendAgentMessage` | Integrated | On 0.8.1 the authoritative roster atomically replaces Pylon's private cache after strict bounded decoding; older versions retain the event-derived roster. Canonical Pylon task IDs resolve through that private live roster. Messaging is ephemeral. | -| `watchSession`; watcher `subscribe`, `getMessages`, `close` | Integrated for bounded child live activity in ordinary sessions | A short-lived watcher attaches only to an explicitly selected active descendant. Its `getMessages` supplies one bounded, sanitized committed-message snapshot with assistant text and a coarse safe-label tool skeleton, then watcher message and tool lifecycle events maintain active-only live activity until the watcher closes. Native tool IDs are immediately reduced to attachment-salted in-memory correlation digests; arguments, results, reasoning, paths, timestamps, metadata, and error text never cross the boundary. Strict correlated sessions reject this API before attachment because the same shared-client attachment would invalidate the root proof. This is distinct from, and does not enable, root transcript reads. | -| `getAgentMessageStatus`, `pauseAgentMessages`, `resumeAgentMessages`, `clearAgentMessages` | Intentionally excluded | These controls are daemon-global and can change or clear traffic belonging to unrelated sessions. | -| `startSideQuestion`, `abortSideQuestion` | Integrated as constrained transient quick questions | Supervised, fresh sessions may run one bounded tool-free question through a requester-owned unary RPC. Pylon uses separate public/native IDs, returns only one temporary answer, requests one bounded abort on cancellation, timeout, or disconnect, and never retries or persists the prompt, answer, native errors, or lifecycle. Full-access extension hooks, restored sessions, ACP, follow-up transcripts, and reconnect recovery fail closed. | -| `getHeartbeat`, `setHeartbeat`, `updateHeartbeat`, `listHeartbeats`, `listCronJobs`, `addCronJob`, `cancelCronJob`, `manageHeartbeat` | Deferred on lifecycle ownership | Scheduling promotes work to resident ownership, but public APIs do not provide Pylon an authoritative autonomous-turn/checkpoint identity, demotion, reattachment, or fail-safe delete flow. Shipping now could leave invisible mutations or orphaned work. | -| `getContextTree` | Blocked by upstream execution safety | Prime 0.8.1 synchronously follows and recursively scans unbounded `sub-*` directories before Pylon can decode or time out the result. Until Prime adds intrinsic symlink, cycle, depth, node, and byte bounds, Pylon uses bounded session stats and its own observed agent usage instead; native labels, IDs, model metadata, costs, and history remain private. | -| `getSessionContext`, `getSessionTree`, `getUserMessagesForForking`, `getLastAssistantText` | Intentionally redundant/sensitive | Pylon's event-sourced transcript and checkpoints are authoritative; mirroring Prime's private transcript/tree would create a second history source and expose hidden context. | -| `getSystemPrompt` | Intentionally excluded | Hidden instructions and prompt internals are never read for verification or copied across the provider boundary. | -| `getToolDefinition("pylon_update_plan")` | Integrated internally for one exact verifier | Pylon queries only its own managed tool by exact name, strictly decodes and compares the expected public definition, then discards the result. No other tool definition is queried or exposed. | -| `listSavedSessions`, `newSession`, `switchSession`, `fork`, `navigateTree`, `importFromJsonl`, `exportToHtml`, `exportToJsonl`, `renameSavedSession`, `deleteSavedSession` | Deferred on history coordination | Public DTOs are filesystem/path-shaped and can enumerate unrelated Prime history. Native history mutation must first coordinate atomically with Pylon threads, worktrees, and checkpoints. | -| `setSessionName`, `setSessionEntryLabel` | Intentionally redundant | Pylon thread titles and durable activities are the user-visible source of truth. | -| `executeBash`, `executeBashAndWait`, `abortBash` | Intentionally redundant | Pylon's terminal and an agent turn have separate ownership and audit semantics; a raw session bash tunnel would bypass both. | -| `waitForHeadlessCompletion({ waitForRlmQuiescence })` | Integrated as an ordinary-turn barrier; autonomous ownership deferred | Daemon mode uses only the method's authoritative RLM ordering boundary after a Pylon-owned prompt and discards the autonomous-status result. This keeps descendants and their parent continuations inside the canonical Pylon turn without claiming native headless/resident turns. Those autonomous turns remain deferred until they have checkpoint identity, reattachment, stop, and deletion semantics. The valid boundary also supplies the cumulative usage delta used for terminal root, child, and continued-parent billing. ACP compatibility mode uses 0.8.1's terminal standard prompt boundary plus the correlated metadata, while retaining metadata-driven settlement for 0.8.0. | -| `getSessionHeader` | Intentionally redundant/sensitive | The header can repeat native saved-session identity and metadata. Pylon uses its private verified resume sidecar plus the durable thread projection instead of exposing or persisting a second header source. | -| `onBeforeSessionInvalidate` | Unavailable in daemon mode 0.8.1 | The public daemon implementation is a no-op returning only an unsubscribe function, so there is no lifecycle outcome to integrate. Pylon tears down its owned connection scope explicitly. | -| `promoteToResident` | Intentionally excluded until automation ownership exists | Client-owned workers must remain stoppable and reapable by their Pylon thread. | +| Public API outcome | Pylon status | Decision | +| ---------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| `attach`, root `subscribe`, `getState`, `getInitialSnapshot`, `dispose` | Integrated internally | Own one private daemon session per active Pylon thread, resume exact verified identity, reconnect through public snapshots/events, and close with the thread scope. On 0.8.1 recovery Pylon resupplies the owner `cwd`, session/agent directories, execution policy, and current Pylon-selected model and thinking level; successful model and thinking mutations update that transient recovery context, which Prime never persists. The root connection's `getMessages` transcript API is never called; Pylon's event-sourced transcript remains authoritative. These are lifecycle primitives, not client RPCs. | +| `promptAndWait`, `steer`, `followUp`, `abort` | Integrated | Typed turn admission, active steering, explicit follow-up, queue modes, and interruption remain under Pylon turn ownership. On 0.8.1 the prompt boundary is followed by the public RLM-quiescence barrier so descendant-triggered parent continuations remain in the same canonical turn. Every admitted steer or follow-up advances a correlation generation and rearms that boundary. An in-process transport reconnect preserves the same Pylon turn only after complete native replay or exact public-snapshot reconciliation; ambiguity fails the turn and disposes the native session. Admission is never retried. | +| `supportsNegotiatedCapability`, `submitCorrelatedPrompt`, `cancelPromptLifecycle`, `getPromptLifecycles` | Integrated behind frozen SDK feature and post-attach proof | The exact frozen `negotiated_daemon_session_capabilities_v1` feature enables the proof accessor check only after attach. A false initial proof selects typed-busy ordinary admission; a true proof latches strict correlated mode. Consumer proof epochs fence every asynchronous lifecycle call, worker snapshot, and resync. Proof loss is terminal and never resubmits or downgrades an owned prompt. Direct replacement currently invalidates proof before publication and therefore closes the strict session. Method presence, version, schema, hello offers, and mutable feature registries never enable this path. | +| `prompt`, `waitForIdle` | Intentionally redundant | Pylon uses the cancellable `promptAndWait` admission path and daemon events for exact turn settlement. A second fire-and-observe prompt path or an unscoped idle waiter would weaken turn and checkpoint ownership. | +| `getQueue`, `clearQueue`, `setSteeringMode`, `setFollowUpMode` | Integrated | Expose only counts and delivery modes. Queued text stays private. | +| `mutateQueuedMessage` | Partially integrated | Pylon exposes sole-lane deletion only. Preview text stays server-private. A serialized, non-recovering compare-delete is followed by reconciliation; ambiguous mutations are never retried. Multi-item delete, move, and replace remain unavailable without opaque IDs or revisions. | +| `abortAndClearQueue` | Intentionally folded into Stop | The public operation combines interruption and queue deletion. Pylon keeps non-interrupting Clear and authoritative Stop separate so the reverse state is unambiguous. Prime Agent 0.8.1 suspends queued-input admission when aborting, so Pylon explicitly resumes that scheduler before reusing the session. | +| `setModel`, `setThinkingLevel`, `setServiceTier` | Integrated | Exact selection is owned by the durable thread model projection and reconciled against sanitized session state. | +| `getAvailableModels`, `getModelCatalog` | Integrated provider-catalog and auth-readiness enrichment | Attached sessions use `getModelCatalog`, with `getAvailableModels` as fallback. Strict decoding maps configured models into the existing provider snapshot and reports authenticated readiness only from a healthy current catalog containing a configured native provider. Empty catalogs and non-ready probes leave authentication unknown. This does not verify live network access or expose credential data. Failed or late reads keep the last good list without letting it override probe health; discovery never creates a session. | +| `cycleModel`, `cycleThinkingLevel`, `setScopedModels` | Intentionally redundant | Pylon already has an exact multi-client model picker. Prime's scoped list is memory-only and native cycling cannot atomically update Pylon's durable selection, so exposing both would create split-brain state. | +| `setTransport` | Intentionally excluded | Transport is environment/provider plumbing owned by Pylon, not a per-thread user setting. Prime does not expose authoritative current transport in session state. | +| `getSessionStats` | Integrated | Only bounded context usage and finite reported turn-cost outcomes cross the boundary. | +| `compact`, `abortCompaction`, `setAutoCompactionEnabled` | Integrated | Manual compaction is argument-free; summaries, instructions, paths, and native results are discarded. Automatic state is reconciled before publication. Prime suspends queued session input when compaction aborts the active run, including when compaction is declined, so the next prompt first performs the same exact-session queue resume used after Stop and fails before admission if that resume cannot be proved. | +| `refine({ global: false })` | Integrated | Explicit local-only refinement accepts no instructions or rollback identity. RPC success contains aggregate edit counts only. Timeout/rejection is outcome-unknown and is never retried. | +| `abortBranchSummary` | Intentionally folded into Stop | Pylon does not start a standalone native branch-summary operation; stopping the owning turn remains authoritative. | +| `setAutoRetryEnabled`, `abortRetry` | Deferred | Retry lifecycle is observed safely, but enabled state has no authoritative readback and the setter writes shared provider settings. Stop already cancels the owning turn. A distinct retry control needs truthful session state and receipts. | +| `reload`, `getCommands` | Integrated | Full-access sessions can reload while idle and show bounded safe command metadata. Before and after reload, Pylon verifies its generated source plus the managed extension path, marker, and exact plan-tool definition. Supervised sessions fail closed. | +| `acquireSessionInputPause` | Intentionally redundant | Every Pylon prompt and resource reload is serialized by the per-thread adapter lock, and reload is admitted only while the owned session is idle. Pylon's scoped MCP server is attached before the first snapshot and released only after turn ownership ends, so it is never replaced during live input. Holding a native lease would add reconnect failure modes without fencing additional work. Revisit if Pylon supports live MCP configuration changes. | +| `supportsAcpMcpServers`, `replaceAcpMcpServers`, `releaseAcpMcpServers` | Integrated with scoped Pylon ownership | `McpProviderSession` is the single per-thread source of truth. Before the first daemon snapshot, Pylon replaces one `t3-code` HTTP server under the stable owner `pylon:` and fails closed if Prime cannot own it. After a daemon reconnect, Pylon reclaims that ownership before publishing the resynced session; if it cannot, the session closes instead of continuing without browser tools. Session teardown releases only that owner and server name before disposing the connection. ACP fallback sends the same scoped server in `session/new`. Browser-disabled sessions send nothing, and Prime-owned MCP settings and catalogs remain private. | +| `getResourceSnapshot` | Partially integrated by safe outcome | Commands and safe skill/prompt metadata are decoded internally. Native paths, diagnostics, extensions, themes, packages, and MCP configuration are not sent to clients. | +| `respondToExtensionUiRequest` | Partially integrated by safe outcome | Select, confirm, and input dialogs plus bounded notifications, status, and widgets are correlated without exposing native request envelopes. Submitted free-form input uses a transient provider RPC and is redacted from durable activities. Editor replacement is cancelled because its prefill may contain sensitive model or tool material that cannot safely enter Pylon's synchronized event stream. | +| `getRlmChildSnapshots`, `getRlmMaxDepthStatus`, `setRlmMaxDepth`, `cancelRlmChild`, `sendAgentMessage` | Integrated | On 0.8.1 the authoritative roster atomically replaces Pylon's private cache after strict bounded decoding; older versions retain the event-derived roster. Canonical Pylon task IDs resolve through that private live roster. Messaging is ephemeral. | +| `watchSession`; watcher `subscribe`, `getMessages`, `close` | Integrated for bounded child live activity in ordinary sessions | A short-lived watcher attaches only to an explicitly selected active descendant. Its `getMessages` supplies one bounded, sanitized committed-message snapshot with assistant text and a coarse safe-label tool skeleton, then watcher message and tool lifecycle events maintain active-only live activity until the watcher closes. Native tool IDs are immediately reduced to attachment-salted in-memory correlation digests; arguments, results, reasoning, paths, timestamps, metadata, and error text never cross the boundary. Strict correlated sessions reject this API before attachment because the same shared-client attachment would invalidate the root proof. This is distinct from, and does not enable, root transcript reads. | +| `getAgentMessageStatus`, `pauseAgentMessages`, `resumeAgentMessages`, `clearAgentMessages` | Intentionally excluded | These controls are daemon-global and can change or clear traffic belonging to unrelated sessions. | +| `startSideQuestion`, `abortSideQuestion` | Integrated as constrained transient quick questions | Supervised, fresh sessions may run one bounded tool-free question through a requester-owned unary RPC. Pylon uses separate public/native IDs, returns only one temporary answer, requests one bounded abort on cancellation, timeout, or disconnect, and never retries or persists the prompt, answer, native errors, or lifecycle. Full-access extension hooks, restored sessions, ACP, follow-up transcripts, and reconnect recovery fail closed. | +| `getHeartbeat`, `setHeartbeat`, `updateHeartbeat`, `listHeartbeats`, `listCronJobs`, `addCronJob`, `cancelCronJob`, `manageHeartbeat` | Deferred on lifecycle ownership | Scheduling promotes work to resident ownership, but public APIs do not provide Pylon an authoritative autonomous-turn/checkpoint identity, demotion, reattachment, or fail-safe delete flow. Shipping now could leave invisible mutations or orphaned work. | +| `getContextTree` | Blocked by upstream execution safety | Prime 0.8.1 synchronously follows and recursively scans unbounded `sub-*` directories before Pylon can decode or time out the result. Until Prime adds intrinsic symlink, cycle, depth, node, and byte bounds, Pylon uses bounded session stats and its own observed agent usage instead; native labels, IDs, model metadata, costs, and history remain private. | +| `getSessionContext`, `getSessionTree`, `getUserMessagesForForking`, `getLastAssistantText` | Intentionally redundant/sensitive | Pylon's event-sourced transcript and checkpoints are authoritative; mirroring Prime's private transcript/tree would create a second history source and expose hidden context. | +| `getSystemPrompt` | Intentionally excluded | Hidden instructions and prompt internals are never read for verification or copied across the provider boundary. | +| `getToolDefinition("pylon_update_plan")` | Integrated internally for one exact verifier | Pylon queries only its own managed tool by exact name, strictly decodes and compares the expected public definition, then discards the result. No other tool definition is queried or exposed. | +| `listSavedSessions`, `newSession`, `switchSession`, `fork`, `importFromJsonl`, `exportToHtml`, `exportToJsonl`, `renameSavedSession`, `deleteSavedSession` | Deferred on history coordination | Public DTOs are filesystem/path-shaped and can enumerate unrelated Prime history. Native history mutation must first coordinate atomically with Pylon threads, worktrees, and checkpoints. | +| `getState().leafId`, `navigateTree(leafId, { summarize: false })` | Integrated privately for exact checkpoint rollback | Only the pinned managed full-access daemon path can advertise absolute rollback. Pylon binds immutable checkpoint leaves to provider/runtime/session identity, fences input, quarantines native output, permits only the recorded source or target leaf, proves navigation by rereading `getState()`, and releases only after the committed target is exact. Leaf ids and native identities never cross the adapter-private persistence boundary. Missing methods, approval mode, active work, stale identity, or a third leaf fail closed. | +| `setSessionName`, `setSessionEntryLabel` | Intentionally redundant | Pylon thread titles and durable activities are the user-visible source of truth. | +| `executeBash`, `executeBashAndWait`, `abortBash` | Intentionally redundant | Pylon's terminal and an agent turn have separate ownership and audit semantics; a raw session bash tunnel would bypass both. | +| `waitForHeadlessCompletion({ waitForRlmQuiescence })` | Integrated as an ordinary-turn barrier; autonomous ownership deferred | Daemon mode uses only the method's authoritative RLM ordering boundary after a Pylon-owned prompt and discards the autonomous-status result. This keeps descendants and their parent continuations inside the canonical Pylon turn without claiming native headless/resident turns. Those autonomous turns remain deferred until they have checkpoint identity, reattachment, stop, and deletion semantics. The valid boundary also supplies the cumulative usage delta used for terminal root, child, and continued-parent billing. ACP compatibility mode uses 0.8.1's terminal standard prompt boundary plus the correlated metadata, while retaining metadata-driven settlement for 0.8.0. | +| `getSessionHeader` | Intentionally redundant/sensitive | The header can repeat native saved-session identity and metadata. Pylon uses its private verified resume sidecar plus the durable thread projection instead of exposing or persisting a second header source. | +| `onBeforeSessionInvalidate` | Unavailable in daemon mode 0.8.1 | The public daemon implementation is a no-op returning only an unsubscribe function, so there is no lifecycle outcome to integrate. Pylon tears down its owned connection scope explicitly. | +| `promoteToResident` | Intentionally excluded until automation ownership exists | Client-owned workers must remain stoppable and reapable by their Pylon thread. | Prime Agent 0.8.1 retains the goal-continuation and durable refinement-message changes as native session behavior. Pylon continues to project only bounded goal state and aggregate refinement lifecycle; it does diff --git a/docs/internals/providers.md b/docs/internals/providers.md index 37351c968..1b823d411 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -26,9 +26,11 @@ transport, config, and event shapes are mapped. ### Absolute conversation rollback -The optional `absoluteConversationRollback` adapter boundary captures, inspects, and applies a private JSON anchor with a stable equality digest. The durable rollback saga accepts only adapters that declare `conversationRollback: "absolute"` and implement all three operations. Relative turn counts and uninspectable no-op paths always fail closed. +The optional `absoluteConversationRollback` adapter boundary captures, inspects, applies, and releases a private JSON anchor with a stable equality digest. The durable saga accepts only an adapter that declares `conversationRollback: "absolute"`, implements every operation, and reports the exact thread available. Relative turn counts and uninspectable no-op paths fail closed. -All built-in production adapters currently omit this boundary and remain `unsupported`. The foundation provides only the provider-neutral contract, private persistence, leases, compensation, and reconciliation. A later provider-specific phase must prove its immutable anchor semantics before changing that declaration. Anchors, native session identities, receipts, and recovery paths must not enter orchestration events, logs, shell projections, or client payloads. +The managed Prime daemon adapter is the only built-in absolute implementation. It is enabled only when the pinned managed build exposes the public `getState()` and `navigateTree()` methods. A full-access session must be idle, quiescent, on the same provider instance, runtime generation, session incarnation, and native session. Prime `leafId`, native identities, and anchors stay in private rollback tables and adapter memory. They never enter orchestration events, receipts, logs, telemetry, shell projections, or client payloads. + +Each ready checkpoint anchor binds one exact leaf to its checkpoint ref, object id, turn id, and source revision. A conflicting recapture cannot overwrite it. Navigation calls `navigateTree(leafId, { summarize: false })`, then proves the result with `getState().leafId`. While a saga is nonterminal, the provider admission fence and Prime event quarantine block new input and public output. Source or target is safe to reconcile after response loss or compatible reconnect; any third leaf enters manual recovery. Cleanup releases quarantine only after the public projection commit and a final exact-target proof. A managed recoverable Prime owner becomes an idle retained authority after settlement, stays adoptable across server restart, and cannot rotate into the next turn until terminal projection and checkpoint-quiescence holds are durable. Prime ACP, approval-required sessions, unmanaged or incompatible builds, and every other built-in provider remain unsupported. Prime Agent uses its public detached-daemon APIs as the primary runtime on macOS, Linux, and WSL2 (which reports itself as Linux). `PrimeAgentDriver.create` rejects a native `win32` server before @@ -362,8 +364,8 @@ synchronization. commands. 2. [`ProviderCommandReactor`][cmd] reacts to orchestration intent events and dispatches provider calls. -3. [`CheckpointReactor`][checkpoint] captures workspace checkpoints on turn start and completion. It - rejects coordinated rollback requests while rollback is disabled. +3. [`CheckpointReactor`][checkpoint] captures workspace checkpoints and private provider anchors on + turn start and completion. It dispatches coordinated rollback only after strict admission. ### Turn-start admission and reconciliation