diff --git a/.agents/upstream-review.md b/.agents/upstream-review.md index 8aade218d..1a74dc25e 100644 --- a/.agents/upstream-review.md +++ b/.agents/upstream-review.md @@ -1,7 +1,7 @@ --- remote: t3code-upstream branch: main -reviewed-through: "b1e223e2b0d87124883b1410ab52dd6a1338e40d" +reviewed-through: "d1d15c67f4a5fb82fd8d5e01e5e3b288296789c3" reviewed-through-date: "2026-09-12" --- @@ -15,7 +15,7 @@ The maintainer authorized compatible catch-up and routine adaptations that prese ## Latest cycle -[Upstream integration cycle #526](https://github.com/pylon-code/pylon/issues/526) owns the maintainer-authorized reopening of DEF-7, DEF-16, #479's mobile Antigravity exception and #513's relative-provider rewind exception, plus the new source `d1d15c67f4a5fb82fd8d5e01e5e3b288296789c3` after the previous cursor. The issue tracks final-head review, CI, automatic relay deployment and remaining adapter work. The cursor stays at the previous completed cycle until this cycle's final accounting. +[Upstream integration cycle #526](https://github.com/pylon-code/pylon/issues/526) owns the maintainer-authorized reopening of DEF-7, DEF-16, #479's mobile Antigravity exception and #513's relative-provider rewind exception, plus the new source `d1d15c67f4a5fb82fd8d5e01e5e3b288296789c3` after the previous cursor. The implementation records below account for the frozen source and the reopened exceptions. The issue tracks final landing CI and automatic relay deployment. Native compatibility limits remain explicit; this decision cursor is not a universal feature-parity claim. Previous completed cycles: [#516](https://github.com/pylon-code/pylon/issues/516) through `b1e223e2b0d87124883b1410ab52dd6a1338e40d`, [#497](https://github.com/pylon-code/pylon/issues/497) through `4a4c6dd2adc350a68ba18bb28b24b5a7e4660dab`, and [#414](https://github.com/pylon-code/pylon/issues/414) through `6c583620ff7ad3235b135af7107c0543467eecfa`. Their trigger audits and exclusions describe those cycles' closure state; current dispositions below supersede them. Follow the skill's [continuation procedure](skills/review-t3-upstream/references/continuation.md), checking the issue against Git and GitHub before acting. @@ -120,18 +120,19 @@ All four sources in the exact range are accounted for in [#516](https://github.c The maintainer explicitly reopened the four remaining functional exceptions on 2026-09-12, superseding the old earliest-check dates. -| Group | Sources | Outcome and remaining scope | Pylon PR / verification | -| ---------------------------------------- | -------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| Sidebar project scope in search row | `d1d15c67f4a5fb82fd8d5e01e5e3b288296789c3` | Adopted with Pylon environment identity, narrow-layout scope choice and keyboard accessibility retained. No excluded behavior. | [#527](https://github.com/pylon-code/pylon/pull/527); independent review, 196 focused tests, web types and all ten final-head CI jobs. | -| DEF-7: anonymous macOS previews | `33b650a5b3b27382b35d2182dec6b22438c3da56` (#8243); Pylon #111 and `9d112329e` | Closed. Public unsigned previews require explicit `preview:mac` opt-in on same-repository PRs. Read-only builds feed trusted publication and cleanup; Pylon Alpha identity and production updater isolation remain. No preview was published during implementation. | [#528](https://github.com/pylon-code/pylon/pull/528), with trusted helper loading fixed in [#532](https://github.com/pylon-code/pylon/pull/532); independent review, helper/loader regressions, Actionlint and final-head CI. Actual publication/cleanup workflow verification is tracked in #526. | -| DEF-16: live Claude model quota windows | `19d8ab2ae9fc562ee7b216a0d72903fbfafa9572` (#9507) | Closed through bounded cached OAuth reconciliation after unscoped native hints. Named windows retain their account identity, timestamps and 30-minute retention; generation/token fences reject retired instances and switched accounts. No guessed model mapping or duplicate SDK probe. | [#530](https://github.com/pylon-code/pylon/pull/530); independent review, quota/registry/retention/ingestion regressions, server types and all ten final-head CI jobs. | -| Mobile Antigravity catalog and admission | `06336460c9988f29c71e839c4c9c840c4552e077`, `d487dfbf46be344e818725be70ee04be2436bfb4` | Closes #479's mobile exception. Account-owned defaults/aliases, unavailable saved choices, unknown-auth saved sends, bounded targeted catalog discovery at durable dispatch and post-await validation preserve account binding and queued content. No remaining requested catalog behavior. | [#529](https://github.com/pylon-code/pylon/pull/529); independent review, 140 focused tests, mobile types and all ten final-head CI jobs. Native rendered/live-account verification is not claimed. | -| OpenCode exact rollback | `fd5553f1afcef4f410a067687faa991743b5034c`, `efccda9ac9230db22b36990cffabdad218fa41b0` | Extends #513 with immutable native forks, exact idle source/target proof, idempotent selection, original-source compensation and durable restart recovery. Historical checkpoints lacking a proved snapshot, experimental native plan-file mode and externally managed servers with unobservable runtime flags remain unavailable for exact rewind; ordinary plan/history/resume stay usable. Claude portion remains in #526. | [#531](https://github.com/pylon-code/pylon/pull/531); independent adapter/service review, focused integration tests and isolated installed OpenCode 1.18.29 native fork verification. | -| Codex exact rollback | `fd5553f1afcef4f410a067687faa991743b5034c`, `efccda9ac9230db22b36990cffabdad218fa41b0` | Extends #513 with immutable native forks, complete JSONL and inactive goal proof, idempotent selection, compaction retention and same-incarnation recovery. Active/uninspectable goals, paginated forks, unsupported or oversized native histories and uncaptured old/root bindings remain ineligible; ordinary resume is preserved. | [#533](https://github.com/pylon-code/pylon/pull/533); independent native-source/adapter review, 360 focused tests, server types and integration capability checks. | +| Group | Sources | Outcome and remaining scope | Pylon PR / verification | +| ---------------------------------------- | -------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Sidebar project scope in search row | `d1d15c67f4a5fb82fd8d5e01e5e3b288296789c3` | Adopted with Pylon environment identity, narrow-layout scope choice and keyboard accessibility retained. No excluded behavior. | [#527](https://github.com/pylon-code/pylon/pull/527); independent review, 196 focused tests, web types and all ten final-head CI jobs. | +| DEF-7: anonymous macOS previews | `33b650a5b3b27382b35d2182dec6b22438c3da56` (#8243); Pylon #111 and `9d112329e` | Closed. Public unsigned previews require explicit `preview:mac` opt-in on same-repository PRs. Read-only builds feed trusted publication and cleanup; Pylon Alpha identity and production updater isolation remain. No preview was published during implementation. | [#528](https://github.com/pylon-code/pylon/pull/528), with trusted helper loading fixed in [#532](https://github.com/pylon-code/pylon/pull/532); independent review, helper/loader regressions, Actionlint and final-head CI. Actual publication/cleanup workflow verification is tracked in #526. | +| DEF-16: live Claude model quota windows | `19d8ab2ae9fc562ee7b216a0d72903fbfafa9572` (#9507) | Closed through bounded cached OAuth reconciliation after unscoped native hints. Named windows retain their account identity, timestamps and 30-minute retention; generation/token fences reject retired instances and switched accounts. No guessed model mapping or duplicate SDK probe. | [#530](https://github.com/pylon-code/pylon/pull/530); independent review, quota/registry/retention/ingestion regressions, server types and all ten final-head CI jobs. | +| Mobile Antigravity catalog and admission | `06336460c9988f29c71e839c4c9c840c4552e077`, `d487dfbf46be344e818725be70ee04be2436bfb4` | Closes #479's mobile exception. Account-owned defaults/aliases, unavailable saved choices, unknown-auth saved sends, bounded targeted catalog discovery at durable dispatch and post-await validation preserve account binding and queued content. No remaining requested catalog behavior. | [#529](https://github.com/pylon-code/pylon/pull/529); independent review, 140 focused tests, mobile types and all ten final-head CI jobs. Native rendered/live-account verification is not claimed. | +| OpenCode exact rollback | `fd5553f1afcef4f410a067687faa991743b5034c`, `efccda9ac9230db22b36990cffabdad218fa41b0` | Extends #513 with immutable native forks, exact idle source/target proof, idempotent selection, original-source compensation and durable restart recovery. Historical checkpoints lacking a proved snapshot, experimental native plan-file mode and externally managed servers with unobservable runtime flags remain unavailable for exact rewind; ordinary plan/history/resume stay usable. | [#531](https://github.com/pylon-code/pylon/pull/531); independent adapter/service review, focused integration tests and isolated installed OpenCode 1.18.29 native fork verification. | +| Codex exact rollback | `fd5553f1afcef4f410a067687faa991743b5034c`, `efccda9ac9230db22b36990cffabdad218fa41b0` | Extends #513 with immutable native forks, complete JSONL and inactive goal proof, idempotent selection, compaction retention and same-incarnation recovery. Active/uninspectable goals, paginated forks, unsupported or oversized native histories and uncaptured old/root bindings remain ineligible; ordinary resume is preserved. | [#533](https://github.com/pylon-code/pylon/pull/533); independent native-source/adapter review, 360 focused tests, server types and integration capability checks. | +| Claude exact rollback | `fd5553f1afcef4f410a067687faa991743b5034c`, `efccda9ac9230db22b36990cffabdad218fa41b0` | Extends #513 with immutable inclusive native forks, full bounded raw history proof, source/callback quarantine, idempotent selection and same-incarnation recovery. Empty roots, uncaptured/previous-incarnation checkpoints, unsupported histories and boundaries removed by compaction remain ineligible. Ordinary resume is preserved. | Implementation PR linked in #526; independent exact-head SDK/adapter review, 160 focused tests, server types and combined integration checks. | ## Deferred register -No open functional deferrals remain in this register. DEF-7 and DEF-16 are completed above; exact native rollback work and compatibility limits are tracked by #526. Historical reasons and revisit conditions remain in the archive and earlier cycle records. +No open functional deferrals remain in this register. DEF-7 and DEF-16 are completed above; remaining exact native rollback compatibility limits are documented above. Revisit a limit when the native provider exposes the missing proof or fork behavior; do not substitute relative counts. Historical reasons and revisit conditions remain in the archive and earlier cycle records. ## Upstream watch list diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index f57d9e5fe..5922fd633 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -3,6 +3,7 @@ import { buildRuntimeInstructions } from "../RuntimeInstructions.ts"; import * as NodeFS from "node:fs"; import * as NodeOS from "node:os"; import * as NodePath from "node:path"; +import * as NodeCrypto from "node:crypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; import type { @@ -14,6 +15,7 @@ import type { } from "@anthropic-ai/claude-agent-sdk"; import { ApprovalRequestId, + CheckpointRef, ClaudeSettings, ProviderDriverKind, ProviderItemId, @@ -52,6 +54,11 @@ import { } from "../ClaudeModelCatalog.testFixtures.ts"; import { ProviderAdapterProcessError, ProviderAdapterValidationError } from "../Errors.ts"; import type { ClaudeAdapterShape } from "../Services/ClaudeAdapter.ts"; +import { + claudeProjectDirectoryName, + readClaudeConversationAnchor, + readClaudeExactCursor, +} from "../claudeConversationHistory.ts"; import { makeClaudeAdapter, type ClaudeAdapterLiveOptions } from "./ClaudeAdapter.ts"; const decodeClaudeSettings = Schema.decodeSync(ClaudeSettings); const encodeUnknownJsonString = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); @@ -166,6 +173,7 @@ class FakeClaudeQuery implements AsyncIterable { function makeHarness(config?: { readonly nativeEventLogPath?: string; + readonly replacementQueries?: boolean; readonly nativeEventLogger?: ClaudeAdapterLiveOptions["nativeEventLogger"]; readonly cwd?: string; readonly baseDir?: string; @@ -194,7 +202,8 @@ function makeHarness(config?: { ...(config?.getSessionMessages ? { getSessionMessages: config.getSessionMessages } : {}), ...(config?.forkSession ? { forkSession: config.forkSession } : {}), createQuery: (input) => { - if (createInput && config?.getSessionMessages) queries.push(new FakeClaudeQuery()); + if (createInput && (config?.getSessionMessages || config?.replacementQueries)) + queries.push(new FakeClaudeQuery()); createInput = input; return queries.at(-1)!; }, @@ -7033,6 +7042,375 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("uses immutable full native Claude snapshots for exact retry and idle recovery", () => + Effect.gen(function* () { + const tempRoot = NodeFS.realpathSync( + NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "pylon-claude-exact-")), + ); + yield* Effect.addFinalizer(() => + Effect.sync(() => NodeFS.rmSync(tempRoot, { recursive: true, force: true })), + ); + const cwd = NodePath.join(tempRoot, "workspace"); + const configDir = NodePath.join(tempRoot, "claude"); + NodeFS.mkdirSync(cwd); + const projectDir = NodePath.join(configDir, "projects", claudeProjectDirectoryName(cwd)); + NodeFS.mkdirSync(projectDir, { recursive: true }); + const file = (id: string) => NodePath.join(projectDir, `${id}.jsonl`); + const callbackEntered = Promise.withResolvers(); + const callbackReleased = Promise.withResolvers(); + let blockCallbackUuid = false; + const harness = makeHarness({ + cwd, + baseDir: tempRoot, + claudeConfig: { homePath: configDir }, + replacementQueries: true, + crypto: (crypto) => ({ + ...crypto, + randomUUIDv4: Effect.suspend(() => { + if (!blockCallbackUuid) return crypto.randomUUIDv4; + blockCallbackUuid = false; + callbackEntered.resolve(); + return Effect.promise(() => callbackReleased.promise).pipe( + Effect.andThen(crypto.randomUUIDv4), + ); + }), + }), + }); + yield* Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const incarnation = RuntimeSessionId.make("claude-exact-incarnation"); + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + sessionIncarnationId: incarnation, + cwd, + }); + const nativeId = (session.resumeCursor as { resume: string }).resume; + const operations = adapter.absoluteConversationRollback!; + assert.isFalse(yield* operations.isAvailable(THREAD_ID)); + const rows: Array> = []; + const finishTurn = (text: string, hiddenAttachment: boolean) => + Effect.gen(function* () { + const turn = yield* adapter.sendTurn({ threadId: THREAD_ID, input: text }); + yield* Effect.promise(() => readFirstPromptMessage(harness.getLastCreateQueryInput())); + const parent = rows.at(-1)?.uuid ?? null; + const user = { + type: "user", + uuid: turn.turnId, + parentUuid: parent, + sessionId: nativeId, + cwd, + isSidechain: false, + timestamp: "2026-09-01T00:00:00.000Z", + message: { role: "user", content: text }, + }; + rows.push(user); + if (hiddenAttachment) + rows.push({ + type: "attachment", + uuid: NodeCrypto.randomUUID(), + parentUuid: user.uuid, + sessionId: nativeId, + cwd, + isSidechain: false, + timestamp: "2026-09-01T00:00:00.000Z", + attachment: { type: "context", content: "private attached context" }, + }); + const assistant = { + type: "assistant", + uuid: NodeCrypto.randomUUID(), + parentUuid: rows.at(-1)!.uuid, + sessionId: nativeId, + cwd, + isSidechain: false, + timestamp: "2026-09-01T00:00:00.000Z", + parent_tool_use_id: "preserved-tool-parent", + parent_agent_id: "preserved-agent-parent", + message: { + id: "semantic-message", + role: "assistant", + content: [{ type: "text", text: "finished" }], + }, + }; + rows.push(assistant); + NodeFS.writeFileSync( + file(nativeId), + rows.map((row) => encodeUnknownJsonString(row)).join("\n") + "\n", + ); + const completed = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => event.type === "turn.completed" && event.turnId === turn.turnId, + ), + Stream.take(1), + Stream.runDrain, + Effect.forkChild, + ); + harness.query.emit({ + type: "assistant", + session_id: nativeId, + uuid: assistant.uuid, + parent_tool_use_id: null, + message: assistant.message, + } as unknown as SDKMessage); + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + session_id: nativeId, + uuid: NodeCrypto.randomUUID(), + } as unknown as SDKMessage); + yield* Fiber.join(completed); + return turn.turnId; + }); + const firstTurn = yield* finishTurn("first prompt", true); + assert.isTrue(yield* operations.isAvailable(THREAD_ID)); + const target = yield* operations.captureAnchor({ + threadId: THREAD_ID, + binding: { + kind: "checkpoint", + checkpointTurnCount: 1, + turnId: firstTurn, + checkpointRef: CheckpointRef.make("checkpoint-one"), + checkpointOid: "oid-one", + sourceRevision: 1, + }, + }); + const targetAnchor = readClaudeConversationAnchor(target.anchor)!; + const targetFile = NodeFS.readFileSync(file(targetAnchor.snapshotSessionId), "utf8"); + const secondTurn = yield* finishTurn("second prompt", false); + const second = yield* operations.captureAnchor({ + threadId: THREAD_ID, + binding: { + kind: "checkpoint", + checkpointTurnCount: 2, + turnId: secondTurn, + checkpointRef: CheckpointRef.make("checkpoint-two"), + checkpointOid: "oid-two", + sourceRevision: 2, + }, + }); + const callbacks = harness.getLastCreateQueryInput()!.options; + const signal = yield* Effect.abortSignal; + blockCallbackUuid = true; + const suspendedQuestion = callbacks.canUseTool!( + "AskUserQuestion", + { questions: [] }, + { signal, toolUseID: "late-question", requestId: "late-question" }, + ); + yield* Effect.promise(() => callbackEntered.promise); + const source = yield* operations.captureAnchor({ + threadId: THREAD_ID, + binding: { + kind: "source", + turnId: secondTurn, + checkpointRef: CheckpointRef.make("checkpoint-two"), + checkpointOid: "oid-two", + sourceRevision: 3, + }, + }); + assert.equal( + (yield* adapter + .sendTurn({ threadId: THREAD_ID, input: "blocked after source capture" }) + .pipe(Effect.result))._tag, + "Failure", + ); + callbackReleased.resolve(); + assert.equal((yield* Effect.promise(() => suspendedQuestion))?.behavior, "deny"); + for (const toolName of ["Write", "AskUserQuestion"]) { + assert.equal( + (yield* Effect.promise(() => + callbacks.canUseTool!( + toolName, + {}, + { signal, toolUseID: "quarantined", requestId: "quarantined" }, + ), + ))?.behavior, + "deny", + ); + } + assert.equal( + (yield* Effect.promise(() => + callbacks.onUserDialog!( + { + dialogKind: "resume_return", + payload: { sessionAgeMinutes: 1, estimatedTokens: 1 }, + }, + { signal, requestId: "quarantined-resume" }, + ), + ))?.behavior, + "cancelled", + ); + const sourceAnchor = readClaudeConversationAnchor(source.anchor)!; + const sourceFile = NodeFS.readFileSync(file(sourceAnchor.snapshotSessionId), "utf8"); + NodeFS.writeFileSync( + file(targetAnchor.snapshotSessionId), + targetFile.replace("private attached context", "changed attached context"), + ); + assert.equal( + (yield* operations.applyAnchor(THREAD_ID, target.anchor).pipe(Effect.result))._tag, + "Failure", + ); + assert.equal(harness.query.closeCalls, 0); + assert.equal((yield* operations.inspectAnchor(THREAD_ID)).digest, source.digest); + NodeFS.writeFileSync(file(targetAnchor.snapshotSessionId), targetFile); + yield* operations.applyAnchor(THREAD_ID, target.anchor); + const selected = (yield* adapter.listSessions())[0]!; + const selectedResume = selected.resumeCursor as { resume: string; claudeExact: unknown }; + assert.notEqual(selectedResume.resume, targetAnchor.snapshotSessionId); + assert.equal((yield* operations.inspectAnchor(THREAD_ID)).digest, target.digest); + assert.equal(harness.queries.length, 2); + yield* operations.applyAnchor(THREAD_ID, target.anchor); + assert.equal(harness.queries.length, 2); + assert.equal((yield* adapter.listSessions())[0]?.sessionIncarnationId, incarnation); + assert.equal(NodeFS.readFileSync(file(targetAnchor.snapshotSessionId), "utf8"), targetFile); + assert.equal(NodeFS.readFileSync(file(sourceAnchor.snapshotSessionId), "utf8"), sourceFile); + assert.equal( + (yield* adapter + .sendTurn({ threadId: THREAD_ID, input: "blocked while quarantined" }) + .pipe(Effect.result))._tag, + "Failure", + ); + yield* operations.releaseAnchor(THREAD_ID, target.anchor); + const reboundAnchor = { + ...targetAnchor, + completedTurnId: "same-content-boundary", + checkpointTurnCount: 7, + }; + yield* operations.applyAnchor(THREAD_ID, reboundAnchor); + const rebound = readClaudeExactCursor( + ((yield* adapter.listSessions())[0]!.resumeCursor as { claudeExact: unknown }) + .claudeExact, + )!; + assert.equal(rebound.idle?.completedTurnId, "same-content-boundary"); + assert.equal(rebound.idle?.turnCount, 7); + assert.equal(harness.queries.length, 2); + yield* operations.applyAnchor(THREAD_ID, target.anchor); + yield* operations.releaseAnchor(THREAD_ID, target.anchor); + const durable = (yield* adapter.listSessions())[0]!; + assert.equal( + readClaudeExactCursor((durable.resumeCursor as { claudeExact: unknown }).claudeExact) + ?.idle?.turnCount, + 1, + ); + yield* adapter.stopSession(THREAD_ID); + assert.equal( + yield* adapter.recoverSession!({ + threadId: THREAD_ID, + providerInstanceId: session.providerInstanceId!, + sessionIncarnationId: RuntimeSessionId.make("wrong-incarnation"), + runtimeMode: "full-access", + cwd, + resumeCursor: durable.resumeCursor, + }), + null, + ); + const recovered = yield* adapter.recoverSession!({ + threadId: THREAD_ID, + providerInstanceId: session.providerInstanceId!, + sessionIncarnationId: incarnation, + runtimeMode: "full-access", + cwd, + resumeCursor: { + ...(durable.resumeCursor as Record), + resumeSessionAt: "550e8400-e29b-41d4-a716-446655440099", + turnCount: 99, + turnStartMessageIds: [], + }, + }); + assert.isNotNull(recovered); + const recoveredPrivate = readClaudeExactCursor( + (durable.resumeCursor as { claudeExact: unknown }).claudeExact, + )!; + assert.equal( + (recovered!.resumeCursor as { resumeSessionAt: string }).resumeSessionAt, + recoveredPrivate.idle!.lastAssistantUuid, + ); + assert.equal((recovered!.resumeCursor as { turnCount: number }).turnCount, 1); + assert.isUndefined(harness.getLastCreateQueryInput()?.options.resumeSessionAt); + yield* operations.prepareRecovery!({ + threadId: THREAD_ID, + sourceAnchor: source.anchor, + desiredAnchor: target.anchor, + expectedAnchor: target.anchor, + }); + yield* adapter.activateRecoveredSession!(THREAD_ID); + assert.equal( + (yield* adapter + .sendTurn({ threadId: THREAD_ID, input: "still quarantined" }) + .pipe(Effect.result))._tag, + "Failure", + ); + yield* operations.releaseAnchor(THREAD_ID, target.anchor); + assert.equal((yield* operations.inspectAnchor(THREAD_ID)).digest, target.digest); + const rootAttempt = yield* operations + .captureAnchor({ + threadId: THREAD_ID, + binding: { + kind: "checkpoint", + checkpointTurnCount: 0, + turnId: null, + checkpointRef: CheckpointRef.make("root"), + checkpointOid: "root-oid", + sourceRevision: 4, + }, + }) + .pipe(Effect.result); + assert.equal(rootAttempt._tag, "Failure"); + // Full JSONL proof catches context invisible to the SDK message projection. + const liveId = ((yield* adapter.listSessions())[0]!.resumeCursor as { resume: string }) + .resume; + NodeFS.writeFileSync( + file(liveId), + NodeFS.readFileSync(file(liveId), "utf8").replace( + "private attached context", + "changed attached context", + ), + ); + assert.equal( + (yield* operations.inspectAnchor(THREAD_ID).pipe(Effect.result))._tag, + "Failure", + ); + assert.notEqual(second.digest, target.digest); + // Ordinary explicit resume keeps native history even when exact proof is + // stale, and never reuses a previous incarnation's private authorization. + const staleCursor = (yield* adapter.listSessions())[0]!.resumeCursor; + yield* adapter.stopSession(THREAD_ID); + assert.equal( + yield* adapter.recoverSession!({ + threadId: THREAD_ID, + providerInstanceId: session.providerInstanceId!, + sessionIncarnationId: incarnation, + runtimeMode: "full-access", + cwd, + resumeCursor: staleCursor, + }), + null, + ); + const resumed = yield* adapter.startSession({ + threadId: THREAD_ID, + runtimeMode: "full-access", + cwd, + sessionIncarnationId: RuntimeSessionId.make("new-normal-incarnation"), + resumeCursor: staleCursor, + }); + assert.equal(harness.getLastCreateQueryInput()?.options.resume, liveId); + assert.isUndefined((resumed.resumeCursor as { claudeExact?: unknown }).claudeExact); + assert.isFalse(yield* operations.isAvailable(THREAD_ID)); + const resumedEvents = yield* adapter.streamEvents.pipe( + Stream.takeUntil( + (event) => + event.type === "session.state.changed" && + event.sessionIncarnationId === resumed.sessionIncarnationId, + ), + Stream.runCollect, + ); + assert.isFalse(encodeUnknownJsonString(resumedEvents).includes("claudeExact")); + }).pipe(Effect.provide(harness.layer)); + }), + ); + it.effect("rewinds a steered Claude turn after recovery and preserves fork boundaries", () => { const forkCalls: Array>> = []; let firstTurnId = ""; diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index a7e8e3813..ee5bd44c5 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -24,6 +24,7 @@ import { type ModelUsage, } from "@anthropic-ai/claude-agent-sdk"; import { parseCliArgs } from "@t3tools/shared/cliArgs"; +import { stableStringify } from "@t3tools/shared/relaySigning"; import { isWorkspaceImagePreviewPath } from "@t3tools/shared/filePreview"; import { ApprovalRequestId, @@ -72,6 +73,7 @@ import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Semaphore from "effect/Semaphore"; import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Fiber from "effect/Fiber"; @@ -112,7 +114,21 @@ import { type ProviderAdapterError, } from "../Errors.ts"; import { type ClaudeAdapterShape } from "../Services/ClaudeAdapter.ts"; -import { BUILT_IN_ADAPTER_CONVERSATION_ROLLBACK_MODES } from "../Services/ProviderAdapter.ts"; +import { + BUILT_IN_ADAPTER_CONVERSATION_ROLLBACK_MODES, + type ProviderAbsoluteConversationRollback, +} from "../Services/ProviderAdapter.ts"; +import { + proveClaudeHistory, + proveClaudeNativeHistory, + isVerifiedClaudeFork, + readClaudeConversationAnchor, + readClaudeExactCursor, + type ClaudeConversationAnchor, + type ClaudeExactCursor, + type ClaudeIdleHistory, +} from "../claudeConversationHistory.ts"; +import { readClaudeNativeHistoryFile } from "../claudeNativeHistoryFile.ts"; import { spawnAndCollect } from "../providerSnapshot.ts"; import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.fromJsonString(Schema.Unknown)); @@ -127,6 +143,8 @@ const decodeSessionMessages = Schema.decodeSync( Schema.Struct({ type: Schema.Literals(["user", "assistant", "system"]), uuid: Schema.String, + session_id: Schema.optionalKey(Schema.String), + parent_agent_id: Schema.optionalKey(Schema.NullOr(Schema.String)), parent_tool_use_id: Schema.NullOr(Schema.String), message: Schema.Unknown, }), @@ -370,6 +388,12 @@ interface ClaudeSessionContext { /** Limits already announced for the running turn, keyed `window:resetsAt`. */ announcedUsageLimits: { turnId: string; keys: Set } | undefined; stopped: boolean; + readonly conversationLock: Semaphore.Semaphore; + historyEpoch: number; + exactCursor: ClaudeExactCursor | undefined; + readonly completedHistories: Map; + quarantined: boolean; + recoveryHeld: boolean; } interface ClaudeQueryRuntime extends AsyncIterable { @@ -894,6 +918,12 @@ function asRuntimeRequestId(value: ApprovalRequestId): RuntimeRequestId { return RuntimeRequestId.make(value); } +function publicClaudeResumeCursor(raw: unknown): unknown { + if (typeof raw !== "object" || raw === null || Array.isArray(raw)) return raw; + const { claudeExact: _privateExact, ...publicCursor } = raw as Record; + return publicCursor; +} + function readClaudeResumeState(resumeCursor: unknown): ClaudeResumeState | undefined { if (!resumeCursor || typeof resumeCursor !== "object") { return undefined; @@ -2058,10 +2088,15 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( sessionIncarnationId: ProviderSession["sessionIncarnationId"], event: ProviderRuntimeEvent, ): Effect.Effect => - Queue.offer( - runtimeEventQueue, - sessionIncarnationId === undefined ? event : { ...event, sessionIncarnationId }, - ).pipe(Effect.asVoid); + Effect.suspend(() => { + const context = sessions.get(event.threadId); + if (context?.quarantined && context.sessionIncarnationId === sessionIncarnationId) + return Effect.void; + return Queue.offer( + runtimeEventQueue, + sessionIncarnationId === undefined ? event : { ...event, sessionIncarnationId }, + ).pipe(Effect.asVoid); + }); const logNativeSdkMessage = Effect.fnUntraced(function* ( context: ClaudeSessionContext, @@ -2120,6 +2155,230 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }; }); + const makeHistoryAccess = Effect.fn("makeHistoryAccess")(function* (context: { + readonly session: Pick; + }) { + const historyWorkerPath = yield* path + .fromFileUrl( + new URL( + import.meta.url.endsWith(".ts") + ? "../../claudeHistoryWorker.ts" + : "./claudeHistoryWorker.mjs", + import.meta.url, + ), + ) + .pipe( + Effect.mapError((cause) => + toRequestError(context.session.threadId, "thread/rollback", cause), + ), + ); + const runScopedHistoryCommand = ( + method: "getSessionMessages" | "forkSession", + args: object, + historySessionId: string, + ) => + spawnAndCollect( + process.execPath, + ChildProcess.make( + process.execPath, + [historyWorkerPath, method, historySessionId, encodeHistoryArgs(args)], + { env: { ...claudeEnvironment, ELECTRON_RUN_AS_NODE: "1" } }, + ), + ).pipe( + Effect.timeout("30 seconds"), + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + Effect.mapError((cause) => + toRequestError(context.session.threadId, "thread/rollback", cause), + ), + Effect.flatMap((result) => + result.code === 0 + ? Effect.succeed(result.stdout) + : Effect.fail( + toRequestError( + context.session.threadId, + "thread/rollback", + new Error(result.stderr || "Claude history command failed."), + ), + ), + ), + ); + const readHistory = (historySessionId: string) => { + const readOptions = { + ...(context.session.cwd ? { dir: context.session.cwd } : {}), + includeSystemMessages: true, + }; + const read = options?.getSessionMessages ?? getSessionMessages; + // Keep process ownership inside the calling Effect scope so interruption + // terminates the worker instead of leaving an independent runtime alive. + return options?.getSessionMessages || + claudeEnvironment.CLAUDE_CONFIG_DIR === process.env.CLAUDE_CONFIG_DIR + ? Effect.tryPromise({ + try: () => read(historySessionId, readOptions), + catch: (cause) => toRequestError(context.session.threadId, "thread/rollback", cause), + }) + : runScopedHistoryCommand("getSessionMessages", readOptions, historySessionId).pipe( + Effect.flatMap((source) => + Effect.try({ + try: () => decodeSessionMessages(source), + catch: (cause) => + toRequestError(context.session.threadId, "thread/rollback", cause), + }), + ), + ); + }; + const forkHistory = (sessionId: string, upToMessageId?: string) => { + const args = { + ...(context.session.cwd ? { dir: context.session.cwd } : {}), + ...(upToMessageId ? { upToMessageId } : {}), + }; + return options?.forkSession || + claudeEnvironment.CLAUDE_CONFIG_DIR === process.env.CLAUDE_CONFIG_DIR + ? Effect.tryPromise({ + try: () => (options?.forkSession ?? forkSession)(sessionId, args), + catch: (cause) => toRequestError(context.session.threadId, "thread/rollback", cause), + }) + : runScopedHistoryCommand("forkSession", args, sessionId).pipe( + Effect.flatMap((source) => + Effect.try({ + try: () => decodeHistoryFork(source), + catch: (cause) => + toRequestError(context.session.threadId, "thread/rollback", cause), + }), + ), + ); + }; + return { readHistory, forkHistory }; + }); + + const exactUnavailable = () => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "conversation/exact", + reason: "unsupported", + detail: + "Claude cannot prove this exact conversation boundary. Its native history may be missing, changed, compacted, or unsupported.", + }); + const invalidateExactHistory = (context: ClaudeSessionContext) => { + context.historyEpoch += 1; + if (context.exactCursor) { + const { idle: _idle, selected: _selected, ...cursor } = context.exactCursor; + context.exactCursor = cursor; + } + }; + const isIdle = (context: ClaudeSessionContext) => + !context.stopped && + !context.turnState && + context.inFlightTools.size === 0 && + context.liveTaskIds.size === 0 && + context.pendingApprovals.size === 0 && + context.pendingUserInputs.size === 0 && + Queue.sizeUnsafe(context.promptQueue) === 0; + const requireExactCurrent = (context: ClaudeSessionContext, epoch: number) => + Effect.suspend(() => + sessions.get(context.session.threadId) === context && + !context.stopped && + context.historyEpoch === epoch + ? Effect.void + : Effect.fail(exactUnavailable()), + ); + const readExactHistory = Effect.fn("readExactHistory")( + function* ( + context: { readonly session: Pick }, + nativeSessionId: string, + ) { + const cwd = context.session.cwd; + if (!cwd || !isUuid(nativeSessionId)) return yield* exactUnavailable(); + const canonicalCwd = yield* fileSystem + .realPath(cwd) + .pipe(Effect.mapError(() => exactUnavailable())); + const source = yield* readClaudeNativeHistoryFile({ + sessionId: nativeSessionId, + canonicalCwd, + environment: claudeEnvironment, + }).pipe( + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + Effect.mapError(() => exactUnavailable()), + ); + const native = proveClaudeNativeHistory( + source, + nativeSessionId, + new Set([cwd, canonicalCwd]), + ); + const access = yield* makeHistoryAccess(context); + const projected = proveClaudeHistory( + yield* access.readHistory(nativeSessionId).pipe( + Effect.timeout("30 seconds"), + Effect.mapError(() => exactUnavailable()), + ), + nativeSessionId, + ); + if ( + !native || + !projected || + projected.messages.length === 0 || + projected.messages.some((message) => { + const index = native.ids.get(message.uuid); + const row = index === undefined ? undefined : native.rows[index]; + return ( + !row || + row.type !== message.type || + stableStringify(row.message) !== stableStringify(message.message) + ); + }) + ) + return yield* exactUnavailable(); + return { ...native, projected: projected.messages, canonicalCwd, source }; + }, + (effect) => + effect.pipe( + Effect.timeout("30 seconds"), + Effect.mapError(() => exactUnavailable()), + ), + ); + const recordCompletedHistory = Effect.fn("recordCompletedHistory")(function* ( + context: ClaudeSessionContext, + completedTurnId: TurnId, + ) { + if ( + !context.sessionIncarnationId || + !context.resumeSessionId || + !context.lastAssistantUuid || + !isIdle(context) || + context.quarantined + ) + return; + const epoch = context.historyEpoch; + const proof = yield* readExactHistory(context, context.resumeSessionId).pipe(Effect.option); + if ( + proof._tag === "None" || + sessions.get(context.session.threadId) !== context || + !isIdle(context) || + context.historyEpoch !== epoch || + proof.value.rows.at(-1)?.uuid !== context.lastAssistantUuid + ) + return; + const idle: ClaudeIdleHistory = { + nativeSessionId: context.resumeSessionId, + digest: proof.value.digest, + messageCount: proof.value.rows.length, + lastAssistantUuid: context.lastAssistantUuid, + completedTurnId, + turnCount: context.turnStartMessageIds.length, + turnStartMessageIds: [...context.turnStartMessageIds], + }; + context.completedHistories.set(completedTurnId, idle); + context.exactCursor = { + version: 1, + providerInstanceId: boundInstanceId, + sessionIncarnationId: context.sessionIncarnationId, + threadId: context.session.threadId, + cwd: proof.value.canonicalCwd, + idle, + anchors: context.exactCursor?.anchors ?? [], + }; + }); + const updateResumeCursor = Effect.fn("updateResumeCursor")(function* ( context: ClaudeSessionContext, ) { @@ -2132,6 +2391,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ...(context.lastAssistantUuid ? { resumeSessionAt: context.lastAssistantUuid } : {}), turnCount: context.turnStartMessageIds.length, turnStartMessageIds: [...context.turnStartMessageIds], + ...(context.exactCursor ? { claudeExact: context.exactCursor } : {}), }; context.session = { @@ -2494,6 +2754,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( readonly rawSource: "claude.sdk.message" | "claude.sdk.permission"; readonly rawMethod: string; readonly rawPayload: unknown; + readonly ownsCallback?: () => boolean; }, ) { const turnState = context.turnState; @@ -2512,6 +2773,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( turnState.capturedProposedPlanKeys.add(captureKey); const stamp = yield* makeEventStamp(); + if (input.ownsCallback && !input.ownsCallback()) return; yield* offerRuntimeEvent(context.sessionIncarnationId, { type: "turn.proposed.completed", eventId: stamp.eventId, @@ -2568,7 +2830,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }); }); - const completeTurn = Effect.fn("completeTurn")(function* ( + const completeTurnUnlocked = Effect.fn("completeTurn")(function* ( context: ClaudeSessionContext, status: ProviderRuntimeTurnStatus, errorMessage?: string, @@ -2735,6 +2997,18 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( rawPayload: result ?? { status }, }); + const updatedAt = yield* nowIso; + context.turnState = undefined; + context.session = { + ...context.session, + status: "ready", + activeTurnId: undefined, + updatedAt, + ...(status === "failed" && errorMessage ? { lastError: errorMessage } : {}), + }; + if (status === "completed") yield* recordCompletedHistory(context, turnState.turnId); + yield* updateResumeCursor(context); + const stamp = yield* makeEventStamp(); yield* offerRuntimeEvent(context.sessionIncarnationId, { type: "turn.completed", @@ -2756,19 +3030,18 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }, providerRefs: nativeProviderRefs(context), }); - - const updatedAt = yield* nowIso; - context.turnState = undefined; - context.session = { - ...context.session, - status: "ready", - activeTurnId: undefined, - updatedAt, - ...(status === "failed" && errorMessage ? { lastError: errorMessage } : {}), - }; - yield* updateResumeCursor(context); }); + const completeTurn = ( + context: ClaudeSessionContext, + status: ProviderRuntimeTurnStatus, + errorMessage?: string, + result?: SDKResultMessage, + ) => + context.conversationLock.withPermits(1)( + completeTurnUnlocked(context, status, errorMessage, result), + ); + const handleStreamEvent = Effect.fn("handleStreamEvent")(function* ( context: ClaudeSessionContext, message: SDKMessage, @@ -4091,6 +4364,16 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( context: ClaudeSessionContext, message: SDKMessage, ) { + if (context.quarantined) return; + if ( + message.type === "user" || + message.type === "assistant" || + message.type === "stream_event" || + message.type === "conversation_reset" || + (message.type === "system" && message.subtype === "compact_boundary") + ) { + invalidateExactHistory(context); + } yield* logNativeSdkMessage(context, message); yield* ensureThreadId(context, message); @@ -4268,12 +4551,24 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( // Same reason as the approvals above: a request nobody can answer any more // must not stay open, or the thread can never be settled. - for (const pending of context.pendingUserInputs.values()) { + for (const [requestId, pending] of context.pendingUserInputs) { yield* pending.cancel; + const stamp = yield* makeEventStamp(); + yield* offerRuntimeEvent(context.sessionIncarnationId, { + type: "user-input.resolved", + eventId: stamp.eventId, + provider: PROVIDER, + createdAt: stamp.createdAt, + threadId: context.session.threadId, + ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), + requestId: asRuntimeRequestId(requestId), + payload: { answers: {} }, + providerRefs: nativeProviderRefs(context), + }); } if (context.turnState) { - yield* completeTurn(context, "interrupted", "Session stopped."); + yield* completeTurnUnlocked(context, "interrupted", "Session stopped."); } yield* Queue.shutdown(context.promptQueue); @@ -4343,6 +4638,11 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( input: Parameters[0], token: object, expectedContext?: ClaudeSessionContext, + exactStart?: { + readonly cursor: ClaudeExactCursor; + readonly guard: Effect.Effect; + readonly recoveryHeld?: boolean; + }, ) { const modelCatalog = yield* modelCatalogEffect; yield* requireLifecycleToken(input.threadId, token); @@ -4360,6 +4660,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }); } + if (exactStart) yield* exactStart.guard; const existingContext = sessions.get(input.threadId); if (existingContext) { yield* Effect.logWarning("claude.session.replacing", { @@ -4412,6 +4713,26 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const liveTaskIds = new Set(); const contextRef = yield* Ref.make(undefined); + const callbackCancelled = () => ({ + behavior: "deny" as const, + message: "User cancelled tool execution.", + }); + const callbackUnavailable = () => ({ + behavior: "deny" as const, + message: "Claude session is stopped, replaced, or reserved for exact conversation recovery.", + }); + // Pin the native selection rather than the streaming frame epoch: ordinary + // parallel tool frames may arrive while a user is answering an approval. + const callbackOwnership = (context: ClaudeSessionContext) => { + const nativeSessionId = context.resumeSessionId; + const selected = context.exactCursor?.selected; + return () => + sessions.get(context.session.threadId) === context && + !context.stopped && + !context.quarantined && + context.resumeSessionId === nativeSessionId && + context.exactCursor?.selected === selected; + }; /** * Handle AskUserQuestion tool calls by emitting a `user-input.requested` @@ -4425,6 +4746,8 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( readonly toolUseID?: string; }, ) { + const ownsCallback = callbackOwnership(context); + if (!ownsCallback()) return callbackUnavailable(); const requestId = ApprovalRequestId.make(yield* randomUUIDv4); // Parse questions from the SDK's AskUserQuestion input. @@ -4466,8 +4789,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( cancel: settleAsAborted, }; - // Emit user-input.requested so the UI can present the questions. + // Publish pending ownership before the event can yield to exact capture. const requestedStamp = yield* makeEventStamp(); + if (!ownsCallback()) return callbackUnavailable(); + pendingUserInputs.set(requestId, pendingInput); yield* offerRuntimeEvent(input.sessionIncarnationId, { type: "user-input.requested", eventId: requestedStamp.eventId, @@ -4494,7 +4819,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }, }); - pendingUserInputs.set(requestId, pendingInput); + if (!ownsCallback()) { + pendingUserInputs.delete(requestId); + return aborted ? callbackCancelled() : callbackUnavailable(); + } // Handle abort (e.g. turn interrupted while waiting for user input). const onAbort = () => { @@ -4516,6 +4844,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( // Emit user-input.resolved so the UI knows the interaction completed. const resolvedStamp = yield* makeEventStamp(); + if (!ownsCallback()) return aborted ? callbackCancelled() : callbackUnavailable(); yield* offerRuntimeEvent(input.sessionIncarnationId, { type: "user-input.resolved", eventId: resolvedStamp.eventId, @@ -4539,6 +4868,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }, }); + if (!ownsCallback()) return aborted ? callbackCancelled() : callbackUnavailable(); if (aborted) { return { behavior: "deny", @@ -4566,9 +4896,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( } const context = yield* Ref.get(contextRef); - if (!context) { + if (!context || !callbackOwnership(context)()) { return { behavior: "cancelled" as const }; } + const ownsCallback = callbackOwnership(context); // The question copy lives in @t3tools/shared/claudeCompaction because // the web client recognizes this exact text (and the "never" answer) @@ -4608,7 +4939,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }, ); - if (result.behavior !== "allow") { + if (result.behavior !== "allow" || !ownsCallback()) { return { behavior: "cancelled" as const }; } @@ -4633,12 +4964,9 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( callbackOptions: Parameters[2], ) { const context = yield* Ref.get(contextRef); - if (!context) { - return { - behavior: "deny", - message: "Claude session context is unavailable.", - } satisfies PermissionResult; - } + if (!context) return callbackUnavailable(); + const ownsCallback = callbackOwnership(context); + if (!ownsCallback()) return callbackUnavailable(); // Handle AskUserQuestion: surface clarifying questions to the // user via the user-input runtime event channel, regardless of @@ -4655,6 +4983,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( toolUseId: callbackOptions.toolUseID, rawSource: "claude.sdk.permission", rawMethod: "canUseTool/ExitPlanMode", + ownsCallback, rawPayload: { toolName, input: toolInput, @@ -4689,6 +5018,8 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }; const requestedStamp = yield* makeEventStamp(); + if (!ownsCallback()) return callbackUnavailable(); + pendingApprovals.set(requestId, pendingApproval); yield* offerRuntimeEvent(input.sessionIncarnationId, { type: "request.opened", eventId: requestedStamp.eventId, @@ -4719,7 +5050,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }, }); - pendingApprovals.set(requestId, pendingApproval); + if (!ownsCallback()) { + pendingApprovals.delete(requestId); + return callbackUnavailable(); + } const onAbort = () => { if (!pendingApprovals.has(requestId)) { @@ -4742,6 +5076,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( pendingApprovals.delete(requestId); const resolvedStamp = yield* makeEventStamp(); + if (!ownsCallback()) return callbackUnavailable(); yield* offerRuntimeEvent(input.sessionIncarnationId, { type: "request.resolved", eventId: resolvedStamp.eventId, @@ -4766,6 +5101,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }, }); + if (!ownsCallback()) return callbackUnavailable(); if (decision === "accept" || decision === "acceptForSession" || decision === "acceptAlways") { return { behavior: "allow", @@ -4983,8 +5319,15 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( updatedAt: startedAt, }; + const conversationLock = yield* Semaphore.make(1); const context: ClaudeSessionContext = { session, + conversationLock, + historyEpoch: 0, + exactCursor: exactStart?.cursor, + completedHistories: new Map(), + quarantined: exactStart !== undefined, + recoveryHeld: exactStart?.recoveryHeld ?? false, sessionIncarnationId: input.sessionIncarnationId, startInput: input, turnStartMessageIds: resumeState?.turnStartMessageIds @@ -5047,7 +5390,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( provider: PROVIDER, createdAt: sessionStartedStamp.createdAt, threadId, - payload: input.resumeCursor !== undefined ? { resume: input.resumeCursor } : {}, + payload: + input.resumeCursor !== undefined + ? { resume: publicClaudeResumeCursor(input.resumeCursor) } + : {}, providerRefs: {}, }); @@ -5133,8 +5479,11 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ); }); - const sendTurn: ClaudeAdapterShape["sendTurn"] = Effect.fn("sendTurn")(function* (input) { + const sendTurnUnlocked: ClaudeAdapterShape["sendTurn"] = Effect.fn("sendTurn")(function* (input) { const context = yield* requireSession(input.threadId); + if (context.quarantined) return yield* exactUnavailable(); + invalidateExactHistory(context); + yield* updateResumeCursor(context); const modelCatalog = yield* modelCatalogEffect; const selectedModel = input.modelSelection !== undefined && input.modelSelection.instanceId === boundInstanceId @@ -5155,7 +5504,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const steeringTurnState = context.turnState && context.turnState.synthetic !== true ? context.turnState : null; if (context.turnState && steeringTurnState === null) { - yield* completeTurn(context, "completed"); + yield* completeTurnUnlocked(context, "completed"); } if (modelSelection?.model) { @@ -5290,6 +5639,11 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }; }); + const sendTurn: ClaudeAdapterShape["sendTurn"] = (input) => + requireSession(input.threadId).pipe( + Effect.flatMap((context) => context.conversationLock.withPermits(1)(sendTurnUnlocked(input))), + ); + const interruptTurn: ClaudeAdapterShape["interruptTurn"] = Effect.fn("interruptTurn")( function* (threadId, _turnId) { sessionLifecycleTokens.delete(threadId); @@ -5355,67 +5709,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( detail: "Claude session id is unavailable.", }); } - const historyWorkerPath = yield* path - .fromFileUrl( - new URL( - import.meta.url.endsWith(".ts") - ? "../../claudeHistoryWorker.ts" - : "./claudeHistoryWorker.mjs", - import.meta.url, - ), - ) - .pipe(Effect.mapError((cause) => toRequestError(threadId, "thread/rollback", cause))); - const runScopedHistoryCommand = ( - method: "getSessionMessages" | "forkSession", - args: object, - historySessionId = sessionId, - ) => - spawnAndCollect( - process.execPath, - ChildProcess.make( - process.execPath, - [historyWorkerPath, method, historySessionId, encodeHistoryArgs(args)], - { env: { ...claudeEnvironment, ELECTRON_RUN_AS_NODE: "1" } }, - ), - ).pipe( - Effect.timeout("30 seconds"), - Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), - Effect.mapError((cause) => toRequestError(threadId, "thread/rollback", cause)), - Effect.flatMap((result) => - result.code === 0 - ? Effect.succeed(result.stdout) - : Effect.fail( - toRequestError( - threadId, - "thread/rollback", - new Error(result.stderr || "Claude history command failed."), - ), - ), - ), - ); - const readHistory = (historySessionId: string) => { - const readOptions = { - ...(context.session.cwd ? { dir: context.session.cwd } : {}), - includeSystemMessages: true, - }; - const read = options?.getSessionMessages ?? getSessionMessages; - // Keep process ownership inside the calling Effect scope so interruption - // terminates the worker instead of leaving an independent runtime alive. - return options?.getSessionMessages || - claudeEnvironment.CLAUDE_CONFIG_DIR === process.env.CLAUDE_CONFIG_DIR - ? Effect.tryPromise({ - try: () => read(historySessionId, readOptions), - catch: (cause) => toRequestError(threadId, "thread/rollback", cause), - }) - : runScopedHistoryCommand("getSessionMessages", readOptions, historySessionId).pipe( - Effect.flatMap((source) => - Effect.try({ - try: () => decodeSessionMessages(source), - catch: (cause) => toRequestError(threadId, "thread/rollback", cause), - }), - ), - ); - }; + const { readHistory, forkHistory } = yield* makeHistoryAccess(context); const messages = yield* readHistory(sessionId); yield* requireCurrent; // Tool results are user-role messages too. Only human prompts begin a turn. @@ -5473,26 +5767,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( } const rollbackAt = retainedCount > 0 ? messages[firstRemoved - 1]?.uuid : undefined; const retainedTurns = context.turns.slice(0, Math.max(0, context.turns.length - numTurns)); - const forkOptions = { - ...(context.session.cwd ? { dir: context.session.cwd } : {}), - ...(rollbackAt ? { upToMessageId: rollbackAt } : {}), - }; - const fork = rollbackAt - ? yield* options?.forkSession || - claudeEnvironment.CLAUDE_CONFIG_DIR === process.env.CLAUDE_CONFIG_DIR - ? Effect.tryPromise({ - try: () => (options?.forkSession ?? forkSession)(sessionId, forkOptions), - catch: (cause) => toRequestError(threadId, "thread/rollback", cause), - }) - : runScopedHistoryCommand("forkSession", forkOptions).pipe( - Effect.flatMap((source) => - Effect.try({ - try: () => decodeHistoryFork(source), - catch: (cause) => toRequestError(threadId, "thread/rollback", cause), - }), - ), - ) - : undefined; + const fork = rollbackAt ? yield* forkHistory(sessionId, rollbackAt) : undefined; yield* requireCurrent; const retainedBoundaries = boundaries.slice(0, retainedCount); if (fork) { @@ -5527,6 +5802,376 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }, ); + const requireOwnedAnchor = (context: ClaudeSessionContext, raw: unknown) => { + const anchor = readClaudeConversationAnchor(raw); + return anchor && + anchor.providerInstanceId === boundInstanceId && + anchor.sessionIncarnationId === context.sessionIncarnationId && + anchor.threadId === context.session.threadId && + anchor.cwd === context.exactCursor?.cwd + ? anchor + : undefined; + }; + const withExactContext = ( + threadId: ThreadId, + operation: (context: ClaudeSessionContext) => Effect.Effect, + ) => + requireSession(threadId).pipe( + Effect.flatMap((context) => + context.conversationLock.withPermits(1)( + Effect.suspend(() => + isIdle(context) && context.exactCursor?.idle && context.sessionIncarnationId + ? operation(context) + : Effect.fail(exactUnavailable()), + ), + ), + ), + ); + const inspectExact = Effect.fn("inspectClaudeExact")(function* (context: ClaudeSessionContext) { + const idle = context.exactCursor?.idle; + if (!idle || !isIdle(context)) return yield* exactUnavailable(); + const epoch = context.historyEpoch; + const proof = yield* readExactHistory(context, idle.nativeSessionId); + yield* requireExactCurrent(context, epoch); + if ( + proof.canonicalCwd !== context.exactCursor?.cwd || + proof.digest !== idle.digest || + proof.rows.length !== idle.messageCount || + proof.rows.at(-1)?.uuid !== idle.lastAssistantUuid + ) + return yield* exactUnavailable(); + return proof; + }); + const verifySnapshot = Effect.fn("verifyClaudeSnapshot")(function* ( + context: ClaudeSessionContext, + anchor: ClaudeConversationAnchor, + ) { + const proof = yield* readExactHistory(context, anchor.snapshotSessionId); + if ( + proof.canonicalCwd !== anchor.cwd || + proof.digest !== anchor.digest || + proof.rows.length !== anchor.messageCount + ) + return yield* exactUnavailable(); + return proof; + }); + const forkExact = Effect.fn("forkClaudeExact")(function* ( + context: ClaudeSessionContext, + sourceSessionId: string, + source: { readonly rows: ReadonlyArray; readonly digest: string }, + upToMessageId?: string, + ) { + const { forkHistory } = yield* makeHistoryAccess(context); + const fork = yield* forkHistory(sourceSessionId, upToMessageId).pipe( + Effect.timeout("30 seconds"), + Effect.mapError(() => exactUnavailable()), + ); + if (!isUuid(fork.sessionId) || fork.sessionId === sourceSessionId) + return yield* exactUnavailable(); + const proof = yield* readExactHistory(context, fork.sessionId); + const sourceIds = new Set(source.rows.map((row) => row.uuid)); + if ( + proof.digest !== source.digest || + !isVerifiedClaudeFork(sourceSessionId, source.rows, proof.rows) || + proof.rows.some((row) => row.uuid !== undefined && sourceIds.has(row.uuid)) + ) + return yield* exactUnavailable(); + return { ...proof, sessionId: fork.sessionId }; + }); + const absoluteConversationRollback: ProviderAbsoluteConversationRollback = { + isAvailable: (threadId) => + Effect.sync(() => { + const context = sessions.get(threadId); + return ( + context !== undefined && + isIdle(context) && + context.sessionIncarnationId !== undefined && + context.exactCursor?.idle !== undefined + ); + }), + captureAnchor: ({ threadId, binding }) => + withExactContext(threadId, (context) => + Effect.gen(function* () { + if (binding.turnId === null) return yield* exactUnavailable(); + const epoch = context.historyEpoch; + const cursor = context.exactCursor!; + const idle = cursor.idle!; + const current = yield* inspectExact(context); + const boundary = + context.completedHistories.get(binding.turnId) ?? + (idle.completedTurnId === binding.turnId ? idle : undefined); + if ( + !boundary || + boundary.nativeSessionId !== idle.nativeSessionId || + !boundary.lastAssistantUuid + ) + return yield* exactUnavailable(); + const prefix = proveClaudeNativeHistory( + current.source, + idle.nativeSessionId, + new Set([context.session.cwd!, cursor.cwd]), + boundary.lastAssistantUuid, + ); + if ( + !prefix || + prefix.digest !== boundary.digest || + prefix.rows.length !== boundary.messageCount + ) + return yield* exactUnavailable(); + const previous = cursor.anchors.find( + (anchor) => + anchor.completedTurnId === binding.turnId && + anchor.checkpointRef === binding.checkpointRef && + anchor.checkpointOid === binding.checkpointOid, + ); + if ( + binding.kind === "source" && + (!previous || + idle.completedTurnId !== binding.turnId || + current.digest !== boundary.digest) + ) + return yield* exactUnavailable(); + const checkpointTurnCount = + binding.kind === "checkpoint" + ? binding.checkpointTurnCount + : previous!.checkpointTurnCount; + const snapshot = yield* forkExact( + context, + idle.nativeSessionId, + prefix, + boundary.lastAssistantUuid, + ); + yield* requireExactCurrent(context, epoch); + const after = yield* inspectExact(context); + if (after.digest !== current.digest) return yield* exactUnavailable(); + const anchor: ClaudeConversationAnchor = { + version: 1, + providerInstanceId: boundInstanceId, + sessionIncarnationId: context.sessionIncarnationId!, + threadId, + cwd: cursor.cwd, + snapshotSessionId: snapshot.sessionId, + digest: snapshot.digest, + messageCount: snapshot.rows.length, + completedTurnId: binding.turnId, + checkpointTurnCount, + checkpointRef: binding.checkpointRef, + checkpointOid: binding.checkpointOid, + sourceRevision: binding.sourceRevision, + turnStartMessageIndices: boundary.turnStartMessageIds.map((id) => + id === null ? null : (prefix.ids.get(id) ?? null), + ), + }; + yield* requireExactCurrent(context, epoch); + if (!isIdle(context)) return yield* exactUnavailable(); + if (binding.kind === "source") context.quarantined = true; + context.exactCursor = { + ...cursor, + ...(current.digest === anchor.digest + ? { selected: anchor, idle: { ...idle, turnCount: checkpointTurnCount } } + : {}), + anchors: + binding.kind === "checkpoint" + ? [ + ...cursor.anchors.filter((entry) => entry.completedTurnId !== binding.turnId), + anchor, + ] + : cursor.anchors, + }; + yield* updateResumeCursor(context); + return { anchor, digest: anchor.digest }; + }), + ), + inspectAnchor: (threadId) => + withExactContext(threadId, (context) => + Effect.gen(function* () { + const proof = yield* inspectExact(context); + const anchor = requireOwnedAnchor(context, context.exactCursor?.selected); + if (!anchor || anchor.digest !== proof.digest) return yield* exactUnavailable(); + return { anchor, digest: anchor.digest }; + }), + ), + applyAnchor: (threadId, raw) => + withExactContext(threadId, (context) => + Effect.gen(function* () { + const anchor = requireOwnedAnchor(context, raw); + if (!anchor) return yield* exactUnavailable(); + const epoch = context.historyEpoch; + const current = yield* inspectExact(context); + if (current.digest === anchor.digest) { + const turnStartMessageIds = anchor.turnStartMessageIndices.map((index) => + index === null || typeof current.rows[index]?.uuid !== "string" + ? null + : (current.rows[index]!.uuid as string), + ); + context.turnStartMessageIds.splice( + 0, + context.turnStartMessageIds.length, + ...turnStartMessageIds, + ); + context.exactCursor = { + ...context.exactCursor!, + selected: anchor, + idle: { + ...context.exactCursor!.idle!, + completedTurnId: anchor.completedTurnId, + turnCount: anchor.checkpointTurnCount, + turnStartMessageIds, + }, + }; + context.quarantined = true; + yield* updateResumeCursor(context); + return; + } + const snapshot = yield* verifySnapshot(context, anchor); + const fork = yield* forkExact(context, anchor.snapshotSessionId, snapshot); + yield* requireExactCurrent(context, epoch); + yield* inspectExact(context); + const token = sessionLifecycleTokens.get(threadId); + if (!token) return yield* exactUnavailable(); + const lastAssistantUuid = fork.rows.at(-1)?.uuid; + if (typeof lastAssistantUuid !== "string") return yield* exactUnavailable(); + const turnStartMessageIds = anchor.turnStartMessageIndices.map((index) => + index === null + ? null + : typeof fork.rows[index]?.uuid === "string" + ? (fork.rows[index]!.uuid as string) + : null, + ); + const idle: ClaudeIdleHistory = { + nativeSessionId: fork.sessionId, + digest: anchor.digest, + messageCount: fork.rows.length, + lastAssistantUuid, + completedTurnId: anchor.completedTurnId, + turnCount: anchor.checkpointTurnCount, + turnStartMessageIds, + }; + const cursor: ClaudeExactCursor = { ...context.exactCursor!, idle, selected: anchor }; + const restarted = yield* startSessionInternal( + { + ...context.startInput, + runtimeMode: context.session.runtimeMode, + sessionIncarnationId: context.sessionIncarnationId, + resumeCursor: { + resume: fork.sessionId, + resumeSessionAt: lastAssistantUuid, + turnCount: anchor.checkpointTurnCount, + turnStartMessageIds, + }, + }, + token, + context, + { cursor, guard: requireExactCurrent(context, epoch) }, + ); + yield* inspectExact(restarted); + yield* updateResumeCursor(restarted); + }), + ), + releaseAnchor: (threadId, raw) => + withExactContext(threadId, (context) => + Effect.gen(function* () { + const anchor = requireOwnedAnchor(context, raw); + const proof = yield* inspectExact(context); + if ( + !anchor || + context.exactCursor?.selected?.digest !== anchor.digest || + proof.digest !== anchor.digest + ) + return yield* exactUnavailable(); + context.quarantined = false; + context.recoveryHeld = false; + yield* updateResumeCursor(context); + }), + ), + prepareRecovery: ({ threadId, sourceAnchor, desiredAnchor, expectedAnchor }) => + withExactContext(threadId, (context) => + Effect.gen(function* () { + if ( + !requireOwnedAnchor(context, sourceAnchor) || + !requireOwnedAnchor(context, desiredAnchor) + ) + return yield* exactUnavailable(); + const expected = requireOwnedAnchor(context, expectedAnchor); + const proof = yield* inspectExact(context); + if (!expected || proof.digest !== expected.digest) return yield* exactUnavailable(); + context.exactCursor = { ...context.exactCursor!, selected: expected }; + context.quarantined = true; + context.recoveryHeld = false; + yield* updateResumeCursor(context); + }), + ), + }; + const recoverSession: NonNullable = (input) => + Effect.gen(function* () { + if (input.providerInstanceId !== boundInstanceId || sessions.has(input.threadId)) return null; + const raw = input.resumeCursor as { claudeExact?: unknown } | null; + const cursor = readClaudeExactCursor(raw?.claudeExact); + const resume = readClaudeResumeState(input.resumeCursor); + if ( + !cursor?.idle || + cursor.providerInstanceId !== boundInstanceId || + cursor.sessionIncarnationId !== input.sessionIncarnationId || + cursor.threadId !== input.threadId || + resume?.resume !== cursor.idle.nativeSessionId + ) + return null; + const token = {}; + sessionLifecycleTokens.set(input.threadId, token); + const proof = yield* readExactHistory({ session: input }, cursor.idle.nativeSessionId).pipe( + Effect.option, + ); + if ( + proof._tag === "None" || + proof.value.canonicalCwd !== cursor.cwd || + proof.value.digest !== cursor.idle.digest || + proof.value.rows.length !== cursor.idle.messageCount + ) + return null; + yield* requireLifecycleToken(input.threadId, token); + const context = yield* startSessionInternal( + { + ...input, + provider: PROVIDER, + resumeCursor: { + resume: cursor.idle.nativeSessionId, + ...(cursor.idle.lastAssistantUuid + ? { resumeSessionAt: cursor.idle.lastAssistantUuid } + : {}), + turnCount: cursor.idle.turnCount, + turnStartMessageIds: cursor.idle.turnStartMessageIds, + }, + }, + token, + undefined, + { + cursor, + guard: requireLifecycleToken(input.threadId, token), + recoveryHeld: true, + }, + ); + return yield* inspectExact(context).pipe( + Effect.andThen(updateResumeCursor(context)), + Effect.map(() => ({ ...context.session })), + Effect.onError(() => + stopSessionInternal(context, { emitExitEvent: false }).pipe( + Effect.ignoreCause({ log: true }), + ), + ), + ); + }); + const activateRecoveredSession: NonNullable = ( + threadId, + ) => + withExactContext(threadId, (context) => + Effect.gen(function* () { + if (!context.recoveryHeld) return; + yield* inspectExact(context); + context.recoveryHeld = false; + context.quarantined = false; + }), + ); + const respondToRequest: ClaudeAdapterShape["respondToRequest"] = Effect.fn("respondToRequest")( function* (threadId, requestId, decision) { const context = yield* requireSession(threadId); @@ -5621,6 +6266,9 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( interruptTurn, readThread, rollbackThread, + absoluteConversationRollback, + recoverSession, + activateRecoveredSession, respondToRequest, respondToUserInput, stopSession, diff --git a/apps/server/src/provider/Services/ProviderAdapter.test.ts b/apps/server/src/provider/Services/ProviderAdapter.test.ts index 724cd2102..9dbd51f0d 100644 --- a/apps/server/src/provider/Services/ProviderAdapter.test.ts +++ b/apps/server/src/provider/Services/ProviderAdapter.test.ts @@ -7,7 +7,7 @@ describe("built-in provider conversation rollback modes", () => { expect(BUILT_IN_ADAPTER_CONVERSATION_ROLLBACK_MODES).toEqual({ antigravity: "unsupported", codex: "absolute", - claude: "relative", + claude: "absolute", cursor: "unsupported", grok: "unsupported", openCode: "absolute", diff --git a/apps/server/src/provider/Services/ProviderAdapter.ts b/apps/server/src/provider/Services/ProviderAdapter.ts index 5f6ecd013..9d49f059b 100644 --- a/apps/server/src/provider/Services/ProviderAdapter.ts +++ b/apps/server/src/provider/Services/ProviderAdapter.ts @@ -67,7 +67,7 @@ export type ProviderConversationRollbackMode = "absolute" | "relative" | "unsupp export const BUILT_IN_ADAPTER_CONVERSATION_ROLLBACK_MODES = { antigravity: "unsupported", codex: "absolute", - claude: "relative", + claude: "absolute", cursor: "unsupported", grok: "unsupported", openCode: "absolute", diff --git a/apps/server/src/provider/claudeConversationHistory.test.ts b/apps/server/src/provider/claudeConversationHistory.test.ts new file mode 100644 index 000000000..aebb05e77 --- /dev/null +++ b/apps/server/src/provider/claudeConversationHistory.test.ts @@ -0,0 +1,147 @@ +import { describe, expect, it } from "vite-plus/test"; +import { + proveClaudeHistory, + proveClaudeNativeHistory, + isVerifiedClaudeFork, +} from "./claudeConversationHistory.ts"; + +const session = "550e8400-e29b-41d4-a716-446655440010"; +const forkSession = "550e8400-e29b-41d4-a716-446655440020"; +const cwd = "/workspace"; +const uuids = [ + "550e8400-e29b-41d4-a716-446655440001", + "550e8400-e29b-41d4-a716-446655440002", + "550e8400-e29b-41d4-a716-446655440003", +]; +const forkUuids = [ + "650e8400-e29b-41d4-a716-446655440001", + "650e8400-e29b-41d4-a716-446655440002", + "650e8400-e29b-41d4-a716-446655440003", +]; +const rows = [ + { + type: "user", + uuid: uuids[0], + parentUuid: null, + sessionId: session, + cwd, + timestamp: "2026-09-01T00:00:00.000Z", + message: { role: "user", content: "prompt" }, + }, + { + type: "attachment", + uuid: uuids[1], + parentUuid: uuids[0], + sessionId: session, + cwd, + timestamp: "2026-09-01T00:00:00.000Z", + attachment: { type: "context", content: "hidden context" }, + }, + { + type: "assistant", + uuid: uuids[2], + parentUuid: uuids[1], + logicalParentUuid: uuids[0], + sessionId: session, + cwd, + timestamp: "2026-09-01T00:00:00.000Z", + parent_tool_use_id: "tool-parent", + parent_agent_id: "agent-parent", + message: { + id: "message-semantic", + role: "assistant", + content: [{ type: "tool_use", id: "tool-semantic", name: "Read", input: { path: "file" } }], + }, + }, +]; +const proof = (value: unknown, id = session) => + proveClaudeNativeHistory( + typeof value === "string" + ? value + : (value as ReadonlyArray).map((row) => JSON.stringify(row)).join("\n"), + id, + new Set([cwd]), + ); +const remap = (id: string | null | undefined) => (id == null ? id : forkUuids[uuids.indexOf(id)]); +const forkRows = rows.map((row, index) => ({ + ...row, + uuid: forkUuids[index], + sessionId: forkSession, + parentUuid: remap(row.parentUuid), + ...(row.logicalParentUuid ? { logicalParentUuid: remap(row.logicalParentUuid) } : {}), + forkedFrom: { sessionId: session, messageUuid: row.uuid }, + ...(index === rows.length - 1 ? { timestamp: "2026-09-12T00:00:00.000Z" } : {}), +})); + +describe("Claude full native history proof", () => { + it("normalizes proven fork IDs/provenance and only the final timestamp", () => { + expect(proof(forkRows, forkSession)?.digest).toBe(proof(rows)?.digest); + expect( + isVerifiedClaudeFork(session, proof(rows)!.rows, proof(forkRows, forkSession)!.rows), + ).toBe(true); + expect( + proof( + forkRows.map((row, index) => + index === 0 ? { ...row, timestamp: "2026-09-12T00:00:00.000Z" } : row, + ), + forkSession, + )?.digest, + ).not.toBe(proof(rows)?.digest); + expect( + isVerifiedClaudeFork(forkSession, proof(rows)!.rows, proof(forkRows, forkSession)!.rows), + ).toBe(false); + }); + it("preserves attachments, tool identities, parent relationships and unknown semantic data", () => { + const original = proof(rows)!.digest; + for (const [from, to] of [ + ["hidden context", "changed context"], + ["tool-parent", "different-parent"], + ["agent-parent", "other-agent"], + ["tool-semantic", "other-tool"], + ]) { + expect( + proof(rows.map((row) => JSON.parse(JSON.stringify(row).replace(from!, to!))))?.digest, + ).not.toBe(original); + } + expect( + proof(rows.map((row, index) => (index === 2 ? { ...row, logicalParentUuid: uuids[1] } : row))) + ?.digest, + ).not.toBe(original); + expect( + proof(rows.map((row, index) => (index === 2 ? { ...row, neutralizedByFork: true } : row))) + ?.digest, + ).not.toBe(original); + }); + it("rejects missing, malformed, ambiguous and foreign native content", () => { + expect(proof("")).toBeUndefined(); + expect(proof([{ type: "custom-title", customTitle: "empty" }])).toBeUndefined(); + expect(proof(`${JSON.stringify(rows[0])}\nmalformed`)).toBeUndefined(); + expect(proof([...rows, rows[0]])).toBeUndefined(); + expect(proof(rows, forkSession)).toBeUndefined(); + expect(proof(rows.map((row) => ({ ...row, cwd: "/another-workspace" })))).toBeUndefined(); + expect( + proof(rows.map((row, index) => (index === 0 ? { ...row, parentUuid: uuids[2] } : row))), + ).toBeUndefined(); + expect( + proof(rows.map((row, index) => (index === 1 ? { ...row, isSidechain: true } : row))), + ).toBeUndefined(); + }); + it("uses only explicit projected semantic fields and rejects mismatched record identity", () => { + const message = { + type: "assistant", + uuid: uuids[2], + session_id: session, + parent_tool_use_id: "parent", + parent_agent_id: "agent", + message: { content: "answer" }, + timestamp: "earlier", + }; + expect(proveClaudeHistory([message], session)?.digest).toBe( + proveClaudeHistory( + [{ ...message, timestamp: "later", uuid: forkUuids[2], session_id: forkSession }], + forkSession, + )?.digest, + ); + expect(proveClaudeHistory([message], forkSession)).toBeUndefined(); + }); +}); diff --git a/apps/server/src/provider/claudeConversationHistory.ts b/apps/server/src/provider/claudeConversationHistory.ts new file mode 100644 index 000000000..85f8fc7f8 --- /dev/null +++ b/apps/server/src/provider/claudeConversationHistory.ts @@ -0,0 +1,236 @@ +import * as NodeCrypto from "node:crypto"; +import { stableStringify } from "@t3tools/shared/relaySigning"; +import * as Exit from "effect/Exit"; +import * as Schema from "effect/Schema"; + +const Uuid = Schema.String.check( + Schema.isPattern(/^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i), +); +const Digest = Schema.String.check(Schema.isPattern(/^[0-9a-f]{64}$/)); +const NonNegativeInt = Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)); +export const ClaudeHistoryMessage = Schema.Struct({ + type: Schema.Literals(["user", "assistant", "system"]), + uuid: Uuid, + session_id: Uuid, + message: Schema.Json, + parent_tool_use_id: Schema.NullOr(Schema.String), + parent_agent_id: Schema.NullOr(Schema.String), +}); +const decodeMessages = Schema.decodeUnknownExit(Schema.Array(ClaudeHistoryMessage)); + +/** Only transport IDs regenerated by forkSession are omitted; semantic/tool lineage stays exact. */ +export function proveClaudeHistory(raw: unknown, sessionId: string) { + const decoded = decodeMessages(raw); + if (Exit.isFailure(decoded)) return undefined; + const messages = decoded.value; + if ( + messages.some((message) => message.session_id !== sessionId) || + new Set(messages.map((message) => message.uuid)).size !== messages.length + ) + return undefined; + const semantic = messages.map(({ uuid: _uuid, session_id: _sessionId, ...message }) => message); + return { + messages, + digest: NodeCrypto.createHash("sha256").update(stableStringify(semantic)).digest("hex"), + }; +} + +export const ClaudeConversationAnchor = Schema.Struct({ + version: Schema.Literal(1), + providerInstanceId: Schema.String, + sessionIncarnationId: Schema.String, + threadId: Schema.String, + cwd: Schema.String, + snapshotSessionId: Uuid, + digest: Digest, + messageCount: NonNegativeInt, + completedTurnId: Schema.NullOr(Schema.String), + checkpointTurnCount: NonNegativeInt, + checkpointRef: Schema.String, + checkpointOid: Schema.String, + sourceRevision: NonNegativeInt, + turnStartMessageIndices: Schema.Array(Schema.NullOr(NonNegativeInt)), +}); +export type ClaudeConversationAnchor = typeof ClaudeConversationAnchor.Type; +const decodeAnchor = Schema.decodeUnknownExit(ClaudeConversationAnchor); +export function readClaudeConversationAnchor(raw: unknown) { + const decoded = decodeAnchor(raw); + return Exit.isSuccess(decoded) ? decoded.value : undefined; +} + +export const ClaudeIdleHistory = Schema.Struct({ + nativeSessionId: Uuid, + digest: Digest, + messageCount: NonNegativeInt, + lastAssistantUuid: Schema.NullOr(Uuid), + completedTurnId: Schema.NullOr(Schema.String), + turnCount: NonNegativeInt, + turnStartMessageIds: Schema.Array(Schema.NullOr(Uuid)), +}); +export type ClaudeIdleHistory = typeof ClaudeIdleHistory.Type; +export const ClaudeExactCursor = Schema.Struct({ + version: Schema.Literal(1), + providerInstanceId: Schema.String, + sessionIncarnationId: Schema.String, + threadId: Schema.String, + cwd: Schema.String, + idle: Schema.optionalKey(ClaudeIdleHistory), + selected: Schema.optionalKey(ClaudeConversationAnchor), + anchors: Schema.Array(ClaudeConversationAnchor), +}); +export type ClaudeExactCursor = typeof ClaudeExactCursor.Type; +const decodeCursor = Schema.decodeUnknownExit(ClaudeExactCursor); +export function readClaudeExactCursor(raw: unknown) { + const decoded = decodeCursor(raw); + return Exit.isSuccess(decoded) ? decoded.value : undefined; +} + +const decodeNativeEntry = Schema.decodeUnknownExit(Schema.fromJsonString(Schema.JsonObject)); +const isUuid = Schema.is(Uuid); +const isJsonObject = Schema.is(Schema.JsonObject); +// SDK forks intentionally omit progress and file undo history, and append a new +// presentation title. None is provider conversation content; every other field +// and record remains in the comparison, including attachments and system data. +const OMITTED_NATIVE_METADATA = new Set([ + "progress", + "file-history-snapshot", + "custom-title", + "ai-title", + "tag", +]); +export function proveClaudeNativeHistory( + source: string, + sessionId: string, + allowedCwds: ReadonlySet, + upToMessageId?: string, +) { + const all: Array = []; + const seen = new Set(); + for (const line of source.split("\n")) { + if (line.trim().length === 0) continue; + const decoded = decodeNativeEntry(line); + if (Exit.isFailure(decoded)) return undefined; + const row = decoded.value; + if (typeof row.type !== "string") return undefined; + if (row.uuid !== undefined) { + if (!isUuid(row.uuid) || seen.has(row.uuid)) return undefined; + seen.add(row.uuid); + } + if (row.sessionId !== undefined && row.sessionId !== sessionId) return undefined; + if (row.cwd !== undefined && (typeof row.cwd !== "string" || !allowedCwds.has(row.cwd))) + return undefined; + if (row.isSidechain === true) return undefined; + if (row.forkedFrom !== undefined) { + const value = row.forkedFrom; + if ( + !isJsonObject(value) || + Object.keys(value).length !== 2 || + !isUuid(value.sessionId) || + !isUuid(value.messageUuid) + ) + return undefined; + } + all.push(row); + } + const boundary = + upToMessageId === undefined + ? all.length - 1 + : all.findIndex((row) => row.uuid === upToMessageId); + if (boundary < 0) return undefined; + const retained = all.slice(0, boundary + 1); + const progressParents = new Map( + retained + .filter((row) => row.type === "progress" && typeof row.uuid === "string") + .map((row) => [row.uuid as string, row.parentUuid]), + ); + const resolveParent = (parent: Schema.Json | undefined) => { + const visited = new Set(); + while (typeof parent === "string" && progressParents.has(parent)) { + if (visited.has(parent)) return { valid: false, parent: undefined }; + visited.add(parent); + parent = progressParents.get(parent); + } + return { valid: parent === undefined || parent === null || typeof parent === "string", parent }; + }; + const rows = retained.filter((row) => !OMITTED_NATIVE_METADATA.has(row.type as string)); + if (rows.length === 0) return undefined; + const ids = new Map(); + const normalized: Array = []; + for (const [index, row] of rows.entries()) { + const parent = resolveParent(row.parentUuid); + const logical = resolveParent(row.logicalParentUuid); + if ( + !parent.valid || + !logical.valid || + (typeof parent.parent === "string" && !ids.has(parent.parent)) || + (typeof logical.parent === "string" && !ids.has(logical.parent)) + ) + return undefined; + const { + uuid, + parentUuid: _parentUuid, + logicalParentUuid: _logicalParentUuid, + sessionId: nativeSessionId, + timestamp, + forkedFrom: _forkedFrom, + ...semantic + } = row; + if (typeof uuid === "string") ids.set(uuid, index); + normalized.push({ + ...semantic, + ...(uuid !== undefined ? { uuid: index } : {}), + ...(parent.parent !== undefined + ? { parentUuid: typeof parent.parent === "string" ? ids.get(parent.parent)! : null } + : {}), + ...(logical.parent !== undefined + ? { + logicalParentUuid: typeof logical.parent === "string" ? ids.get(logical.parent)! : null, + } + : {}), + ...(nativeSessionId !== undefined ? { sessionId: "" } : {}), + // The official fork changes only the final retained transcript timestamp. + ...(index < rows.length - 1 && timestamp !== undefined ? { timestamp } : {}), + }); + } + return { + rows, + ids, + digest: NodeCrypto.createHash("sha256").update(stableStringify(normalized)).digest("hex"), + }; +} + +export function isVerifiedClaudeFork( + sourceSessionId: string, + source: ReadonlyArray, + fork: ReadonlyArray, +) { + return ( + source.length === fork.length && + fork.every((row, index) => { + const provenance = row.forkedFrom; + return ( + provenance !== undefined && + isJsonObject(provenance) && + provenance.sessionId === sourceSessionId && + provenance.messageUuid === source[index]?.uuid + ); + }) + ); +} + +/** Native SDK0.3.260 path codec. Legacy/ambiguous locations remain ineligible. */ +export function claudeProjectDirectoryName(cwd: string) { + const encoded = cwd.replace(/[^a-zA-Z0-9]/g, "-"); + if (encoded.length <= 200) return encoded; + let hash = 0; + for (let index = 0; index < cwd.length; index += 1) + hash = ((hash << 5) - hash + cwd.charCodeAt(index)) | 0; + return `${encoded.slice(0, 200)}-${Math.abs(hash).toString(36)}`; +} +export function validClaudeProjectDirectoryOverride(value: string | undefined) { + return value !== undefined && + /^[A-Za-z0-9_-]{1,64}$/.test(value) && + !/^(con|prn|aux|nul|com[0-9]|lpt[0-9])$/i.test(value) + ? value + : undefined; +} diff --git a/apps/server/src/provider/claudeNativeHistoryFile.test.ts b/apps/server/src/provider/claudeNativeHistoryFile.test.ts new file mode 100644 index 000000000..28124988f --- /dev/null +++ b/apps/server/src/provider/claudeNativeHistoryFile.test.ts @@ -0,0 +1,72 @@ +import { symlinksSupported } from "@t3tools/shared/testing/symlinks"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; + +import { claudeProjectDirectoryName } from "./claudeConversationHistory.ts"; +import { readClaudeNativeHistoryFile } from "./claudeNativeHistoryFile.ts"; + +it.layer(NodeServices.layer)("bounded Claude native history file", (it) => { + it.effect("rejects replacement, symlinks, unknown file identity, and oversized content", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs + .makeTempDirectoryScoped({ prefix: "pylon-claude-native-reader-" }) + .pipe(Effect.flatMap(fs.realPath)); + const cwd = path.join(root, "workspace"); + yield* fs.makeDirectory(cwd); + const config = path.join(root, "config"); + const project = path.join(config, "projects", claudeProjectDirectoryName(cwd)); + yield* fs.makeDirectory(project, { recursive: true }); + const id = "550e8400-e29b-41d4-a716-446655440010"; + const filePath = path.join(project, `${id}.jsonl`); + const otherPath = path.join(root, "other.jsonl"); + yield* fs.writeFileString(filePath, "original"); + yield* fs.writeFileString(otherPath, "replaced"); + const read = readClaudeNativeHistoryFile({ + sessionId: id, + canonicalCwd: cwd, + environment: { CLAUDE_CONFIG_DIR: config }, + }); + assert.equal(yield* read, "original"); + const swapped: FileSystem.FileSystem = { + ...fs, + open: (candidate, options) => + Effect.gen(function* () { + if (candidate === filePath) yield* fs.rename(otherPath, filePath); + return yield* fs.open(candidate, options); + }), + }; + assert.equal( + (yield* read.pipe(Effect.provideService(FileSystem.FileSystem, swapped), Effect.result)) + ._tag, + "Failure", + ); + const unidentifiable: FileSystem.FileSystem = { + ...fs, + stat: (candidate) => + fs.stat(candidate).pipe(Effect.map((info) => ({ ...info, ino: Option.none() }))), + }; + assert.equal( + (yield* read.pipe( + Effect.provideService(FileSystem.FileSystem, unidentifiable), + Effect.result, + ))._tag, + "Failure", + ); + if (symlinksSupported) { + yield* fs.rename(filePath, otherPath); + yield* fs.symlink(otherPath, filePath); + assert.equal((yield* read.pipe(Effect.result))._tag, "Failure"); + yield* fs.remove(filePath); + yield* fs.rename(otherPath, filePath); + } + yield* fs.truncate(filePath, 64 * 1024 * 1024 + 1); + assert.equal((yield* read.pipe(Effect.result))._tag, "Failure"); + }), + ); +}); diff --git a/apps/server/src/provider/claudeNativeHistoryFile.ts b/apps/server/src/provider/claudeNativeHistoryFile.ts new file mode 100644 index 000000000..aae1c1feb --- /dev/null +++ b/apps/server/src/provider/claudeNativeHistoryFile.ts @@ -0,0 +1,110 @@ +import * as NodeOS from "node:os"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; + +import { + claudeProjectDirectoryName, + validClaudeProjectDirectoryOverride, +} from "./claudeConversationHistory.ts"; + +export class ClaudeNativeHistoryUnavailable extends Schema.TaggedError()( + "ClaudeNativeHistoryUnavailable", + {}, +) {} +const MAX_NATIVE_HISTORY_BYTES = 64 * 1024 * 1024; + +/** Read one current, regular provider-owned transcript through a bounded scoped handle. */ +export const readClaudeNativeHistoryFile = Effect.fn("readClaudeNativeHistoryFile")( + function* (input: { + readonly sessionId: string; + readonly canonicalCwd: string; + readonly environment: NodeJS.ProcessEnv; + }) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + if (!/^[0-9a-f-]{36}$/i.test(input.sessionId)) + return yield* new ClaudeNativeHistoryUnavailable(); + const configDir = + input.environment.CLAUDE_CONFIG_DIR ?? + path.join( + input.environment.HOME ?? input.environment.USERPROFILE ?? NodeOS.homedir(), + ".claude", + ); + if (!path.isAbsolute(configDir)) return yield* new ClaudeNativeHistoryUnavailable(); + const root = yield* fs.realPath(path.join(configDir.normalize("NFC"), "projects")); + const keys = new Set([claudeProjectDirectoryName(input.canonicalCwd)]); + const override = input.environment.CLAUDE_CONFIG_DIR + ? validClaudeProjectDirectoryOverride(input.environment.CLAUDE_CODE_PROJECT_DIR_NAME) + : undefined; + if (override) keys.add(override); + const files: string[] = []; + for (const key of keys) { + const candidate = path.join(root, key, `${input.sessionId}.jsonl`); + if (!(yield* fs.exists(candidate))) continue; + const resolved = yield* fs.realPath(candidate); + if (resolved !== candidate) return yield* new ClaudeNativeHistoryUnavailable(); + files.push(candidate); + } + if (files.length !== 1) return yield* new ClaudeNativeHistoryUnavailable(); + return yield* Effect.gen(function* () { + const filePath = files[0]!; + const pathnameBefore = yield* fs.stat(filePath); + const file = yield* fs.open(filePath, { flag: "r" }); + const before = yield* file.stat; + if ( + Option.isNone(pathnameBefore.ino) || + Option.isNone(before.ino) || + pathnameBefore.dev !== before.dev || + pathnameBefore.ino.value !== before.ino.value + ) + return yield* new ClaudeNativeHistoryUnavailable(); + if ( + before.type !== "File" || + before.size <= 0 || + before.size > MAX_NATIVE_HISTORY_BYTES || + Option.isNone(before.mtime) + ) + return yield* new ClaudeNativeHistoryUnavailable(); + const chunks: Uint8Array[] = []; + let size = 0; + while (true) { + const chunk = yield* file.readAlloc( + Math.min(64 * 1024, MAX_NATIVE_HISTORY_BYTES - size + 1), + ); + if (Option.isNone(chunk)) break; + size += chunk.value.length; + if (size > MAX_NATIVE_HISTORY_BYTES) return yield* new ClaudeNativeHistoryUnavailable(); + chunks.push(chunk.value); + } + const after = yield* file.stat; + const pathnameAfter = yield* fs.stat(filePath); + if ( + (yield* fs.realPath(filePath)) !== filePath || + Option.isNone(pathnameAfter.ino) || + pathnameAfter.dev !== before.dev || + pathnameAfter.ino.value !== before.ino.value + ) + return yield* new ClaudeNativeHistoryUnavailable(); + if ( + BigInt(size) !== before.size || + after.size !== before.size || + Option.isNone(after.mtime) || + after.mtime.value.getTime() !== before.mtime.value.getTime() + ) + return yield* new ClaudeNativeHistoryUnavailable(); + const bytes = new Uint8Array(size); + let offset = 0; + for (const chunk of chunks) { + bytes.set(chunk, offset); + offset += chunk.length; + } + return yield* Effect.try({ + try: () => new TextDecoder("utf-8", { fatal: true }).decode(bytes), + catch: () => new ClaudeNativeHistoryUnavailable(), + }); + }).pipe(Effect.scoped); + }, +); diff --git a/docs/internals/rollback-recovery.md b/docs/internals/rollback-recovery.md index 2b73b39c9..5256c7356 100644 --- a/docs/internals/rollback-recovery.md +++ b/docs/internals/rollback-recovery.md @@ -41,7 +41,7 @@ A target is published as available only after checkpoint capture proves all of t 1. the checkpoint is immutable and ready; 2. the provider adapter advertises the absolute rollback gate; -3. the live session is the managed native Prime or exact-capable OpenCode or Codex session for the projected incarnation; +3. the live session is the managed native Prime or exact-capable OpenCode, Codex or Claude session for the projected incarnation; 4. a matching exact provider anchor was stored for that checkpoint. Admission repeats the proof, checks the exact source revision, requires an idle thread and empty provider queues, and acquires the canonical workspace lease. Published availability is never admission authority. @@ -52,6 +52,8 @@ OpenCode's optional experimental plan mode stores a plan file under a name deriv Codex captures immutable full native forks for owned completed turns. Exact proof includes the complete bounded native JSONL history and inactive native goal state; fresh forks defer goal continuation until the next explicit send. Original checkpoints survive compaction separately from current source proofs. Recovery verifies the same account, workspace and incarnation before selecting a fresh deferred fork. Active or uninspectable goals, paginated forks, external history bases, non-regular files, histories over 16 MiB or 100,000 records, and lines over 1 MiB are ineligible. Imported history has no guessed root or old Pylon turn bindings. Ordinary resume remains available. +Claude captures immutable inclusive native forks and verifies full raw JSONL context against the SDK projection. Proof uses scoped regular-file reads with a 64 MiB cap and a 30-second deadline. Source capture quarantines queries and callbacks before returning a receipt; selected-history recovery requires the same persisted incarnation. Empty roots, uncaptured or previous-incarnation checkpoints, pending native work, and missing, changed or unsupported histories remain unavailable. Compaction that removes a provable mapped boundary makes that checkpoint unavailable. Ordinary resume remains usable. + ## Multi-client behavior The engine serializes admission. Requests for the same source, target and file choice join the active operation. A different target or file choice is rejected. Status is projected and streamed, so refresh, reconnect, remote clients, and multiple devices converge on the same fence and result. diff --git a/docs/user/providers-claude.md b/docs/user/providers-claude.md index fe29c6d64..190f7591b 100644 --- a/docs/user/providers-claude.md +++ b/docs/user/providers-claude.md @@ -88,6 +88,18 @@ You can also send `/compact` in an existing conversation. Web and desktop offer context meter and may suggest it when you return to a large older thread. See [commands and skills](./composer.md#commands-and-skills) for using composer commands. +## Revert completed turns + +Pylon can revert Claude conversations to completed checkpoints whose full native history it has +verified. You can keep the current files or restore the checkpoint's files. Pylon keeps immutable +conversation snapshots and resumes a fresh fork, so retries do not remove additional turns. + +Exact revert requires an idle session and a checkpoint captured in the current session identity. +It is unavailable for the empty conversation, older uncaptured checkpoints, or native history that +is missing, changed, compacted beyond proof, unsupported, or larger than 64 MiB. Background tasks and +pending inputs must finish first. Stopping and normally resuming a session preserves Claude's native +conversation but requires new completed checkpoints before exact revert becomes available again. + ## Task list Claude Code hides its task-tracking tools on its newest models. Pylon turns them back on by default, @@ -131,3 +143,7 @@ put the router's endpoint and credential variables in that instance's **Environm router must run where the environment can reach it. Follow the [Claude Code Router instructions](https://github.com/musistudio/claude-code-router) for its installation and routing configuration. + +## Rewind a conversation + +Eligible completed turns support rewinding with either restored files or your current files. Pylon verifies private native history snapshots before selecting a restored conversation. Empty roots, older turns without snapshots, and checkpoints whose boundaries were removed by compaction remain unavailable. Finish pending tools and questions before rewinding. Ordinary resume remains usable when exact history cannot be verified.