From f5dcc6187ddf401a3a5234e362a59cacd71abb30 Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Thu, 1 Oct 2026 00:34:34 -0600 Subject: [PATCH 1/4] fix(server): Claude threads recover when a restart interrupts the first turn --- .../src/provider/Layers/ClaudeAdapter.test.ts | 146 +++++++++++++++++- .../src/provider/Layers/ClaudeAdapter.ts | 50 +++++- 2 files changed, 193 insertions(+), 3 deletions(-) diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index f7e331304..3848f594b 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -127,8 +127,11 @@ class FakeClaudeQuery implements AsyncIterable { this.setModelCalls.push(model); }; + public permissionModeError: unknown = undefined; + readonly setPermissionMode = async (mode: PermissionMode): Promise => { this.setPermissionModeCalls.push(mode); + if (this.permissionModeError !== undefined) throw this.permissionModeError; }; readonly setMaxThinkingTokens = async (maxThinkingTokens: number | null): Promise => { @@ -4022,6 +4025,10 @@ describe("ClaudeAdapterLive", () => { Effect.gen(function* () { const claudeConfig = decodeClaudeSettings({}); return yield* makeClaudeAdapter(claudeConfig, { + // The replacement resumes the first session, so its prompt is saved. + getSessionMessages: async (sessionId) => [ + claudeHistoryMessage({ type: "user", uuid: "user-1", sessionId }), + ], createQuery: () => { const query = new FakeClaudeQuery(); if (queries.length === 0) { @@ -5261,6 +5268,10 @@ describe("ClaudeAdapterLive", () => { Effect.gen(function* () { const claudeConfig = decodeClaudeSettings({}); return yield* makeClaudeAdapter(claudeConfig, { + // The replacement resumes the first session, so its prompt is saved. + getSessionMessages: async (sessionId) => [ + claudeHistoryMessage({ type: "user", uuid: "user-1", sessionId }), + ], createQuery: () => { const query = new FakeClaudeQuery(); queries.push(query); @@ -7484,6 +7495,135 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("starts fresh when a first-turn resume id was never saved by Claude", () => { + const unsavedSessionId = "550e8400-e29b-41d4-a716-446655440020"; + const historyReads: Array = []; + const harness = makeHarness({ + getSessionMessages: async (sessionId) => { + historyReads.push(sessionId); + return []; + }, + }); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + + const session = yield* adapter.startSession({ + threadId: RESUME_THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + resumeCursor: { resume: unsavedSessionId, turnCount: 0, turnStartMessageIds: [] }, + runtimeMode: "full-access", + }); + + assert.deepEqual(historyReads, [unsavedSessionId]); + const options = harness.getLastCreateQueryInput()?.options; + assert.equal(options?.resume, undefined); + assert.equal(typeof options?.sessionId, "string"); + assert.notEqual(options?.sessionId, unsavedSessionId); + // The persisted cursor must name the session Claude will actually save. + assert.deepEqual(session.resumeCursor, { + threadId: RESUME_THREAD_ID, + resume: options?.sessionId, + turnCount: 0, + }); + + yield* adapter.sendTurn({ + threadId: session.threadId, + input: "hello again", + interactionMode: "default", + attachments: [], + }); + assert.deepEqual(harness.query.setPermissionModeCalls, ["bypassPermissions"]); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + + it.effect("resumes a first-turn session once Claude has saved the prompt", () => { + const savedSessionId = "550e8400-e29b-41d4-a716-446655440021"; + const harness = makeHarness({ + getSessionMessages: async (sessionId) => [ + claudeHistoryMessage({ type: "user", uuid: "user-1", sessionId, content: "hello" }), + ], + }); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + + const session = yield* adapter.startSession({ + threadId: RESUME_THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + resumeCursor: { resume: savedSessionId, turnCount: 0, turnStartMessageIds: [] }, + runtimeMode: "full-access", + }); + + const options = harness.getLastCreateQueryInput()?.options; + assert.equal(options?.resume, savedSessionId); + assert.equal(options?.sessionId, undefined); + assert.equal( + (session.resumeCursor as { readonly resume?: string } | undefined)?.resume, + savedSessionId, + ); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + + it.effect("keeps resuming when the first-turn history check fails", () => { + const sessionId = "550e8400-e29b-41d4-a716-446655440022"; + const harness = makeHarness({ + getSessionMessages: async () => { + throw new Error("history unavailable"); + }, + }); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + + yield* adapter.startSession({ + threadId: RESUME_THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + resumeCursor: { resume: sessionId, turnCount: 0, turnStartMessageIds: [] }, + runtimeMode: "full-access", + }); + + const options = harness.getLastCreateQueryInput()?.options; + assert.equal(options?.resume, sessionId); + assert.equal(options?.sessionId, undefined); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + + it.effect("reports a missing Claude conversation as a session-not-found error", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + harness.query.permissionModeError = new Error( + "No conversation found with session ID: 550e8400-e29b-41d4-a716-446655440023", + ); + + const error = yield* adapter + .sendTurn({ + threadId: session.threadId, + input: "hello", + interactionMode: "default", + attachments: [], + }) + .pipe(Effect.flip); + + assert.equal(error._tag, "ProviderAdapterSessionNotFoundError"); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("preserves durable resume ids across Claude resume hooks", () => { const harness = makeHarness(); return Effect.gen(function* () { @@ -9990,7 +10130,11 @@ describe("ClaudeAdapterLive", () => { }); it.effect("routes Claude resume compaction through the shared user-input UI", () => { - const harness = makeHarness(); + const harness = makeHarness({ + getSessionMessages: async (sessionId) => [ + claudeHistoryMessage({ type: "user", uuid: "user-1", sessionId }), + ], + }); return Effect.gen(function* () { const adapter = yield* ClaudeAdapter; const session = yield* adapter.startSession({ diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 41cbb8407..9092aa412 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -402,6 +402,8 @@ interface ClaudeTaskAgentState { const PENDING_TASK_ENTRY_CAP = 256; /** How long Stop waits for Claude to abort a turn before killing the process. */ const CLAUDE_INTERRUPT_GRACE = "3 seconds"; +/** Bounds the check for an unsaved session; a slow check keeps the resume. */ +const CLAUDE_UNSAVED_RESUME_PROBE_TIMEOUT = "10 seconds"; /** * Buffers a value that a later task_started reads by tool_use_id (a racing @@ -1992,7 +1994,12 @@ function toSessionError( cause: unknown, ): ProviderAdapterSessionNotFoundError | ProviderAdapterSessionClosedError | undefined { const normalized = toMessage(cause, "").toLowerCase(); - if (normalized.includes("unknown session") || normalized.includes("not found")) { + if ( + normalized.includes("unknown session") || + normalized.includes("not found") || + // Claude Code's reply when resuming a session id it never saved. + normalized.includes("no conversation found") + ) { return new ProviderAdapterSessionNotFoundError({ provider: PROVIDER, threadId, @@ -4857,8 +4864,46 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( } const startedAt = yield* nowIso; - const resumeState = readClaudeResumeState(input.resumeCursor); + const persistedResumeState = readClaudeResumeState(input.resumeCursor); const threadId = input.threadId; + // Pylon generates the Claude session id and records it before Claude has + // written a transcript. A restart during the first turn can leave a cursor + // for a conversation Claude never saved, and resuming it fails every turn + // with "No conversation found". Without a completed turn or recorded + // message id, check the history and start fresh only when it is + // confirmed empty; a failed check keeps the resume. + const unprovenResumeSessionId = + exactStart === undefined && + persistedResumeState?.resume !== undefined && + (persistedResumeState.turnCount ?? 0) === 0 && + persistedResumeState.resumeSessionAt === undefined && + !(persistedResumeState.turnStartMessageIds ?? []).some((id) => id !== null) + ? persistedResumeState.resume + : undefined; + const resumeUnsaved = + unprovenResumeSessionId !== undefined && + (yield* makeHistoryAccess({ session: input }).pipe( + Effect.flatMap(({ readHistory }) => readHistory(unprovenResumeSessionId)), + Effect.timeout(CLAUDE_UNSAVED_RESUME_PROBE_TIMEOUT), + Effect.matchEffect({ + onFailure: (cause) => + Effect.logWarning("claude.session.resume-probe-failed", { + threadId, + sessionId: unprovenResumeSessionId, + cause, + }).pipe(Effect.as(false)), + onSuccess: (messages) => + messages.length > 0 + ? Effect.succeed(false) + : Effect.logInfo("claude.session.resume-unsaved", { + threadId, + sessionId: unprovenResumeSessionId, + }).pipe(Effect.as(true)), + }), + )); + const unsavedResumeSessionId = resumeUnsaved ? unprovenResumeSessionId : undefined; + // An unsaved session carries no turns, so it restarts as a new session. + const resumeState = unsavedResumeSessionId === undefined ? persistedResumeState : undefined; const existingResumeSessionId = resumeState?.resume; const newSessionId = existingResumeSessionId === undefined ? yield* randomUUIDv4 : undefined; const sessionId = existingResumeSessionId ?? newSessionId; @@ -5445,6 +5490,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( existingResumeSessionId !== undefined ? "resume-session" : "generated-session", "claude.resume.thread_id": resumeState?.threadId ?? "", "claude.resume.session_id": existingResumeSessionId ?? "", + "claude.resume.unsaved_session_id": unsavedResumeSessionId ?? "", "claude.resume.session_at": resumeState?.resumeSessionAt ?? "", "claude.resume.turn_count": resumeState?.turnCount ?? -1, "claude.query.cwd": input.cwd ?? "", From e5a873272166c5f731e658c99745746092bd1310 Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Thu, 1 Oct 2026 08:03:12 -0600 Subject: [PATCH 2/4] fix(server): probe Claude history whenever no assistant message proves the session --- .../src/provider/Layers/ClaudeAdapter.test.ts | 109 ++++++++++++++++++ .../src/provider/Layers/ClaudeAdapter.ts | 28 +++-- .../serverRuntimeStartup.reconcile.test.ts | 71 ++++++++++++ apps/server/src/serverRuntimeStartup.ts | 9 ++ packages/contracts/src/provider.ts | 6 + 5 files changed, 212 insertions(+), 11 deletions(-) diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 3848f594b..2b78d81fd 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -7524,7 +7524,9 @@ describe("ClaudeAdapterLive", () => { threadId: RESUME_THREAD_ID, resume: options?.sessionId, turnCount: 0, + turnStartMessageIds: [], }); + assert.equal(session.conversationReset, true); yield* adapter.sendTurn({ threadId: session.threadId, @@ -7539,6 +7541,106 @@ describe("ClaudeAdapterLive", () => { ); }); + const restartDuringFirstTurn = ( + history: ( + turnId: string, + ) => Awaited>>, + ) => { + let recordedTurnId: string | undefined; + const harness = makeHarness({ + getSessionMessages: async () => (recordedTurnId === undefined ? [] : history(recordedTurnId)), + }); + const run = Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const first = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + const firstSessionId = harness.getLastCreateQueryInput()?.options.sessionId; + // ProviderService persists the cursor sendTurn returns, before Claude + // has written the prompt to its transcript. + const turn = yield* adapter.sendTurn({ + threadId: first.threadId, + input: "hello", + attachments: [], + }); + recordedTurnId = String(turn.turnId); + const persistedCursor = turn.resumeCursor; + yield* adapter.stopSession(first.threadId); + + const restarted = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + resumeCursor: persistedCursor, + }); + return { + adapter, + firstSessionId, + persistedCursor, + restarted, + options: harness.getLastCreateQueryInput()?.options, + turnId: recordedTurnId, + }; + }); + return { harness, run }; + }; + + it.effect("starts fresh from the cursor a restart-interrupted first turn persisted", () => { + const { harness, run } = restartDuringFirstTurn(() => []); + return Effect.gen(function* () { + const { adapter, firstSessionId, persistedCursor, restarted, options, turnId } = yield* run; + const restartedQuery = harness.queries.at(-1)!; + + assert.deepEqual(persistedCursor, { + threadId: THREAD_ID, + resume: firstSessionId, + turnCount: 1, + turnStartMessageIds: [turnId], + }); + assert.equal(options?.resume, undefined); + assert.equal(typeof options?.sessionId, "string"); + assert.notEqual(options?.sessionId, firstSessionId); + assert.equal(restarted.conversationReset, true); + // Turn ids carry over so rollback boundaries still match Pylon's turns. + assert.deepEqual(restarted.resumeCursor, { + threadId: THREAD_ID, + resume: options?.sessionId, + turnCount: 1, + turnStartMessageIds: [turnId], + }); + + yield* adapter.sendTurn({ + threadId: restarted.threadId, + input: "hello again", + interactionMode: "default", + attachments: [], + }); + assert.equal(harness.queries.length, 2); + assert.deepEqual(restartedQuery.setPermissionModeCalls, ["bypassPermissions"]); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + + it.effect("resumes the restart-interrupted first turn once Claude saved its prompt", () => { + const { harness, run } = restartDuringFirstTurn((turnId) => [ + claudeHistoryMessage({ type: "user", uuid: turnId, content: "hello" }), + ]); + return Effect.gen(function* () { + const { firstSessionId, restarted, options } = yield* run; + + assert.equal(options?.resume, firstSessionId); + assert.equal(options?.sessionId, undefined); + assert.equal(restarted.conversationReset, undefined); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("resumes a first-turn session once Claude has saved the prompt", () => { const savedSessionId = "550e8400-e29b-41d4-a716-446655440021"; const harness = makeHarness({ @@ -8355,8 +8457,11 @@ describe("ClaudeAdapterLive", () => { entered.resolve(); await released.promise; }; + // Startup checks the unproven cursor's history; hold only the rollback's reads. + let rollbackStarted = false; const harness = makeHarness({ getSessionMessages: async (sessionId) => { + if (!rollbackStarted) return history(sessionId); if ( (stage === "read" && sessionId === originalId) || (stage === "fork-read" && sessionId === forkId) @@ -8381,6 +8486,7 @@ describe("ClaudeAdapterLive", () => { turnStartMessageIds: ["user-1", "user-2"], }, }); + rollbackStarted = true; const rollback = yield* adapter .rollbackThread(THREAD_ID, 1) .pipe(Effect.result, Effect.forkChild); @@ -8693,6 +8799,9 @@ describe("ClaudeAdapterLive", () => { runtimeMode: "full-access", resumeCursor: { resume: "550e8400-e29b-41d4-a716-446655440010", + // An observed assistant message proves the session, so startup + // skips its history check and only the rollback spawns the worker. + resumeSessionAt: "assistant-2", turnCount: 2, turnStartMessageIds: ["user-1", "user-2"], }, diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 9092aa412..6306b0822 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -4866,18 +4866,18 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const startedAt = yield* nowIso; const persistedResumeState = readClaudeResumeState(input.resumeCursor); const threadId = input.threadId; - // Pylon generates the Claude session id and records it before Claude has - // written a transcript. A restart during the first turn can leave a cursor - // for a conversation Claude never saved, and resuming it fails every turn - // with "No conversation found". Without a completed turn or recorded - // message id, check the history and start fresh only when it is - // confirmed empty; a failed check keeps the resume. + // Pylon generates the Claude session id and records it, along with each + // turn's prompt id, before Claude has written a transcript. A restart + // during the first turn can leave a cursor for a conversation Claude never + // saved, and resuming it fails every turn with "No conversation found". + // Turn counters are recorded before the prompt reaches Claude, so they + // prove nothing; only an observed assistant message does. Without one, + // check the history and start fresh only when it is confirmed empty. A + // failed or slow check keeps the resume. const unprovenResumeSessionId = exactStart === undefined && persistedResumeState?.resume !== undefined && - (persistedResumeState.turnCount ?? 0) === 0 && - persistedResumeState.resumeSessionAt === undefined && - !(persistedResumeState.turnStartMessageIds ?? []).some((id) => id !== null) + persistedResumeState.resumeSessionAt === undefined ? persistedResumeState.resume : undefined; const resumeUnsaved = @@ -4902,8 +4902,13 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }), )); const unsavedResumeSessionId = resumeUnsaved ? unprovenResumeSessionId : undefined; - // An unsaved session carries no turns, so it restarts as a new session. - const resumeState = unsavedResumeSessionId === undefined ? persistedResumeState : undefined; + // Claude starts a new session for an unsaved one. The recorded turn ids + // stay so Pylon's turn count and rollback boundaries still line up. + let resumeState = persistedResumeState; + if (unsavedResumeSessionId !== undefined && persistedResumeState !== undefined) { + const { resume: _unsaved, ...carried } = persistedResumeState; + resumeState = carried; + } const existingResumeSessionId = resumeState?.resume; const newSessionId = existingResumeSessionId === undefined ? yield* randomUUIDv4 : undefined; const sessionId = existingResumeSessionId ?? newSessionId; @@ -5537,6 +5542,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ...(input.cwd ? { cwd: input.cwd } : {}), ...(modelSelection?.model ? { model: modelSelection.model } : {}), ...(threadId ? { threadId } : {}), + ...(unsavedResumeSessionId !== undefined ? { conversationReset: true } : {}), resumeCursor: { ...(threadId ? { threadId } : {}), ...(sessionId ? { resume: sessionId } : {}), diff --git a/apps/server/src/serverRuntimeStartup.reconcile.test.ts b/apps/server/src/serverRuntimeStartup.reconcile.test.ts index 1151f2508..840711fb7 100644 --- a/apps/server/src/serverRuntimeStartup.reconcile.test.ts +++ b/apps/server/src/serverRuntimeStartup.reconcile.test.ts @@ -1158,3 +1158,74 @@ it.effect("settles failed opt-in recovery without retrying the provider turn", ( }); }), ); + +it.effect("does not send a continuation when the provider restarted the conversation", () => + Effect.gen(function* () { + const turnId = TurnId.make("turn-conversation-reset"); + const thread = makeThread("thread-conversation-reset", "running", turnId); + const settled = yield* Deferred.make(); + const sends: ProviderSendTurnInput[] = []; + const dispatched: OrchestrationCommand[] = []; + let binding: ProviderSessionDirectory.ProviderRuntimeBinding = { + threadId: thread.id, + provider: ProviderDriverKind.make("codex"), + providerInstanceId, + status: "running", + resumeCursor: { threadId: thread.id }, + runtimePayload: { activeTurnId: turnId }, + }; + yield* runReconciliation({ + threads: [thread], + continueAfterRestart: true, + providerService: { + ...makeProviderService(), + startSession: (threadId, input) => + fakeStartSession(threadId, input).pipe( + Effect.map((session) => ({ ...session, conversationReset: true })), + ), + sendTurn: (input) => + Effect.sync(() => { + sends.push(input); + return { threadId: input.threadId, turnId: TurnId.make("unexpected") }; + }), + }, + directory: { + getBinding: () => Effect.sync(() => Option.some(binding)), + upsert: (next) => + Effect.sync(() => { + binding = next; + }), + removeExact: () => Effect.die("unused"), + recordImportedTranscript: () => Effect.die("unused"), + getProvider: () => Effect.die("unused"), + listThreadIds: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), + }, + dispatch: (command) => + Effect.gen(function* () { + dispatched.push(command); + if (command.type === "thread.session.set" && command.session.status === "error") { + yield* Deferred.succeed(settled, undefined); + } + return { sequence: dispatched.length }; + }), + }); + yield* Deferred.await(settled); + + // A "continue" prompt would reach a conversation with none of the + // interrupted turn's context, so the thread asks the user to resend. + assert.equal(sends.length, 0); + const errorSettle = dispatched.at(-1); + assert.equal(errorSettle?.type, "thread.session.set"); + if (errorSettle?.type === "thread.session.set") { + assert.equal(errorSettle.session.status, "error"); + assert.equal(errorSettle.session.sessionIncarnationId, recoveredIncarnationId(thread.id)); + assert.match(errorSettle.session.lastError ?? "", /Send your message again/); + } + assert.deepStrictEqual(binding.runtimePayload, { + activeTurnId: null, + continueAfterServerUpdate: null, + continueAfterServerUpdatePrepared: null, + }); + }), +); diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index da9b130f3..e04c796d6 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -348,6 +348,8 @@ const runStartupPhase = (phase: string, effect: Effect.Effect) const ORPHANED_PROVIDER_SESSION_ERROR = "Provider session did not survive a server restart. Send a new message to continue."; +const CONVERSATION_RESET_AFTER_RESTART_ERROR = + "The server restarted before this conversation was saved, so its context was lost. Send your message again to continue."; const SERVER_UPDATE_CONTINUATION_KEY = "continueAfterServerUpdate"; const SERVER_UPDATE_CONTINUATION_PROMPT = "Continue where you left off."; @@ -792,6 +794,13 @@ export const reconcileProviderSessions = Effect.gen(function* () { createdAt: boundAt, }); boundSession = recoveredBinding; + // The provider lost the interrupted conversation and started a new + // one. A continuation prompt would reach a model with no context, + // so leave the fresh runtime idle and ask the user to resend. + if (recovered.conversationReset === true) { + yield* settleAsError(CONVERSATION_RESET_AFTER_RESTART_ERROR); + return; + } const capabilities = yield* providerService.getCapabilities(providerInstanceId); yield* providerService.sendTurn({ threadId: thread.id, diff --git a/packages/contracts/src/provider.ts b/packages/contracts/src/provider.ts index 42c849e40..5e68d58a9 100644 --- a/packages/contracts/src/provider.ts +++ b/packages/contracts/src/provider.ts @@ -53,6 +53,12 @@ export const ProviderSession = Schema.Struct({ resumeCursor: Schema.optional(Schema.Unknown), /** True when the provider attached this runtime from a durable continuation. */ restored: Schema.optional(Schema.Boolean), + /** + * True when the provider could not find the saved conversation and started + * a new one, so earlier context is gone and a continuation has nothing to + * continue. + */ + conversationReset: Schema.optional(Schema.Boolean), activeTurnId: Schema.optional(TurnId), /** Admission request owning the active provider turn, when known. */ activeTurnRequestId: Schema.optional(CommandId), From 92bb4009d37253be90d700ba2a6c6bd9f2aa8960 Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Thu, 1 Oct 2026 13:02:24 -0600 Subject: [PATCH 3/4] fix(server): reset Claude sessions only when no project holds the transcript --- .../src/provider/Layers/ClaudeAdapter.test.ts | 121 +++++++++++++++++- .../src/provider/Layers/ClaudeAdapter.ts | 38 ++++-- .../provider/claudeNativeHistoryFile.test.ts | 35 ++++- .../src/provider/claudeNativeHistoryFile.ts | 39 ++++-- .../serverRuntimeStartup.reconcile.test.ts | 45 +++++-- apps/server/src/serverRuntimeStartup.ts | 83 +++++++----- 6 files changed, 301 insertions(+), 60 deletions(-) diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 2b78d81fd..61508f5a8 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -232,7 +232,21 @@ function makeHarness(config?: { layer: Layer.effect( ClaudeAdapter, Effect.gen(function* () { - const claudeConfig = decodeClaudeSettings(config?.claudeConfig ?? {}); + // A history hook stands in for Claude's transcripts; give the + // unsaved-session scan an empty config home instead of ~/.claude. + const isolatedHome = + config?.getSessionMessages && !config.claudeConfig?.homePath && !config.environment + ? yield* Effect.acquireRelease( + Effect.sync(() => + NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "claude-config-home-")), + ), + (home) => Effect.sync(() => NodeFS.rmSync(home, { recursive: true, force: true })), + ) + : undefined; + const claudeConfig = decodeClaudeSettings({ + ...(isolatedHome ? { homePath: isolatedHome } : {}), + ...config?.claudeConfig, + }); const crypto = yield* Crypto.Crypto; const adapter = makeClaudeAdapter(claudeConfig, adapterOptions).pipe( Effect.provideService(Crypto.Crypto, config?.crypto?.(crypto) ?? crypto), @@ -7671,6 +7685,111 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("keeps resuming a session Claude saved under another project", () => { + // The SDK lookup is scoped to the current cwd, but Claude Code resumes a + // session id from any project, as after a moved workspace. + const sessionId = "550e8400-e29b-41d4-a716-446655440024"; + const readOptions: Array = []; + const home = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "claude-config-home-")); + const project = NodePath.join(home, "projects", "-previous-workspace"); + NodeFS.mkdirSync(project, { recursive: true }); + NodeFS.writeFileSync(NodePath.join(project, `${sessionId}.jsonl`), "{}\n"); + const harness = makeHarness({ + claudeConfig: { homePath: home }, + getSessionMessages: async (_sessionId, options) => { + readOptions.push(options); + return []; + }, + }); + return Effect.gen(function* () { + yield* Effect.addFinalizer(() => + Effect.sync(() => NodeFS.rmSync(home, { recursive: true, force: true })), + ); + const adapter = yield* ClaudeAdapter; + + const session = yield* adapter.startSession({ + threadId: RESUME_THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + cwd: NodeOS.tmpdir(), + resumeCursor: { resume: sessionId, turnCount: 1, turnStartMessageIds: ["turn-1"] }, + runtimeMode: "full-access", + }); + + // The check searches every project rather than only the cwd's. + assert.deepEqual(readOptions, [{ includeSystemMessages: true }]); + const options = harness.getLastCreateQueryInput()?.options; + assert.equal(options?.resume, sessionId); + assert.equal(options?.sessionId, undefined); + assert.equal(session.conversationReset, undefined); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + + it.effect("resumes an imported session without checking its history", () => { + // Imported cursors carry no turn count, so the session came from Claude + // Code itself rather than an id Pylon generated. + const sessionId = "550e8400-e29b-41d4-a716-446655440025"; + const historyReads: Array = []; + const harness = makeHarness({ + getSessionMessages: async (id) => { + historyReads.push(id); + return []; + }, + }); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + + yield* adapter.startSession({ + threadId: RESUME_THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + resumeCursor: { threadId: RESUME_THREAD_ID, resume: sessionId }, + runtimeMode: "full-access", + }); + + assert.deepEqual(historyReads, []); + assert.equal(harness.getLastCreateQueryInput()?.options.resume, sessionId); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + + it.effect("keeps resuming when the first-turn history check times out", () => { + const sessionId = "550e8400-e29b-41d4-a716-446655440026"; + const historyRead = Promise.withResolvers(); + const harness = makeHarness({ + getSessionMessages: () => { + historyRead.resolve(); + return new Promise(() => undefined); + }, + }); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + + const start = yield* adapter + .startSession({ + threadId: RESUME_THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + resumeCursor: { resume: sessionId, turnCount: 1, turnStartMessageIds: ["turn-1"] }, + runtimeMode: "full-access", + }) + .pipe(Effect.forkChild); + yield* Effect.promise(() => historyRead.promise); + yield* TestClock.adjust("10 seconds"); + const session = yield* Fiber.join(start); + + const options = harness.getLastCreateQueryInput()?.options; + assert.equal(options?.resume, sessionId); + assert.equal(options?.sessionId, undefined); + assert.equal(session.conversationReset, undefined); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("keeps resuming when the first-turn history check fails", () => { const sessionId = "550e8400-e29b-41d4-a716-446655440022"; const harness = makeHarness({ diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 6306b0822..2a28db23a 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -130,7 +130,7 @@ import { type ClaudeExactCursor, type ClaudeIdleHistory, } from "../claudeConversationHistory.ts"; -import { readClaudeNativeHistoryFile } from "../claudeNativeHistoryFile.ts"; +import { claudeTranscriptExists, readClaudeNativeHistoryFile } from "../claudeNativeHistoryFile.ts"; import { spawnAndCollect } from "../providerSnapshot.ts"; import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.fromJsonString(Schema.Unknown)); @@ -2329,9 +2329,11 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ), ), ); - const readHistory = (historySessionId: string) => { + // The SDK scopes a lookup to `dir` when given; "all" searches every project + // the way Claude Code's resume does. + const readHistory = (historySessionId: string, scope: "project" | "all" = "project") => { const readOptions = { - ...(context.session.cwd ? { dir: context.session.cwd } : {}), + ...(scope === "project" && context.session.cwd ? { dir: context.session.cwd } : {}), includeSystemMessages: true, }; const read = options?.getSessionMessages ?? getSessionMessages; @@ -2377,6 +2379,21 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( return { readHistory, forkHistory }; }); + // Proves Claude never saved a session: neither the SDK's all-project lookup + // nor a scan of every project directory finds it. Any read failure fails + // the check so the caller keeps resuming. + const isUnsavedClaudeSession = Effect.fn("isUnsavedClaudeSession")(function* ( + session: Pick, + sessionId: string, + ) { + const { readHistory } = yield* makeHistoryAccess({ session }); + if ((yield* readHistory(sessionId, "all")).length > 0) return false; + return !(yield* claudeTranscriptExists({ sessionId, environment: claudeEnvironment }).pipe( + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + )); + }); + const exactUnavailable = () => new ProviderAdapterRequestError({ provider: PROVIDER, @@ -4874,16 +4891,17 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( // prove nothing; only an observed assistant message does. Without one, // check the history and start fresh only when it is confirmed empty. A // failed or slow check keeps the resume. + // Only cursors Pylon wrote carry a turn count; imported sessions do not. const unprovenResumeSessionId = exactStart === undefined && persistedResumeState?.resume !== undefined && + persistedResumeState.turnCount !== undefined && persistedResumeState.resumeSessionAt === undefined ? persistedResumeState.resume : undefined; const resumeUnsaved = unprovenResumeSessionId !== undefined && - (yield* makeHistoryAccess({ session: input }).pipe( - Effect.flatMap(({ readHistory }) => readHistory(unprovenResumeSessionId)), + (yield* isUnsavedClaudeSession(input, unprovenResumeSessionId).pipe( Effect.timeout(CLAUDE_UNSAVED_RESUME_PROBE_TIMEOUT), Effect.matchEffect({ onFailure: (cause) => @@ -4892,13 +4910,13 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( sessionId: unprovenResumeSessionId, cause, }).pipe(Effect.as(false)), - onSuccess: (messages) => - messages.length > 0 - ? Effect.succeed(false) - : Effect.logInfo("claude.session.resume-unsaved", { + onSuccess: (unsaved) => + unsaved + ? Effect.logInfo("claude.session.resume-unsaved", { threadId, sessionId: unprovenResumeSessionId, - }).pipe(Effect.as(true)), + }).pipe(Effect.as(true)) + : Effect.succeed(false), }), )); const unsavedResumeSessionId = resumeUnsaved ? unprovenResumeSessionId : undefined; diff --git a/apps/server/src/provider/claudeNativeHistoryFile.test.ts b/apps/server/src/provider/claudeNativeHistoryFile.test.ts index 28124988f..afc134c4d 100644 --- a/apps/server/src/provider/claudeNativeHistoryFile.test.ts +++ b/apps/server/src/provider/claudeNativeHistoryFile.test.ts @@ -7,9 +7,42 @@ import * as Option from "effect/Option"; import * as Path from "effect/Path"; import { claudeProjectDirectoryName } from "./claudeConversationHistory.ts"; -import { readClaudeNativeHistoryFile } from "./claudeNativeHistoryFile.ts"; +import { claudeTranscriptExists, readClaudeNativeHistoryFile } from "./claudeNativeHistoryFile.ts"; it.layer(NodeServices.layer)("bounded Claude native history file", (it) => { + it.effect("finds a transcript in any project and fails when projects cannot be read", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "pylon-claude-transcript-" }); + const config = path.join(root, "config"); + const id = "550e8400-e29b-41d4-a716-446655440010"; + const exists = claudeTranscriptExists({ + sessionId: id, + environment: { CLAUDE_CONFIG_DIR: config }, + }); + // No projects directory: Claude has saved nothing. + assert.equal(yield* exists, false); + yield* fs.makeDirectory(path.join(config, "projects", "-current"), { recursive: true }); + assert.equal(yield* exists, false); + yield* fs.makeDirectory(path.join(config, "projects", "-elsewhere")); + yield* fs.writeFileString(path.join(config, "projects", "-elsewhere", `${id}.jsonl`), "{}"); + assert.equal(yield* exists, true); + + const unreadable: FileSystem.FileSystem = { + ...fs, + readDirectory: () => fs.readDirectory(path.join(root, "missing")), + }; + assert.equal( + (yield* exists.pipe( + Effect.provideService(FileSystem.FileSystem, unreadable), + Effect.result, + ))._tag, + "Failure", + ); + }), + ); + it.effect("rejects replacement, symlinks, unknown file identity, and oversized content", () => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; diff --git a/apps/server/src/provider/claudeNativeHistoryFile.ts b/apps/server/src/provider/claudeNativeHistoryFile.ts index aae1c1feb..992877d15 100644 --- a/apps/server/src/provider/claudeNativeHistoryFile.ts +++ b/apps/server/src/provider/claudeNativeHistoryFile.ts @@ -16,6 +16,34 @@ export class ClaudeNativeHistoryUnavailable extends Schema.TaggedError(); + const freshCursor = { threadId: thread.id, resume: "fresh-session", turnCount: 1 }; + const cleared = yield* Deferred.make(); const sends: ProviderSendTurnInput[] = []; const dispatched: OrchestrationCommand[] = []; let binding: ProviderSessionDirectory.ProviderRuntimeBinding = { @@ -1179,9 +1180,28 @@ it.effect("does not send a continuation when the provider restarted the conversa continueAfterRestart: true, providerService: { ...makeProviderService(), + // Like ProviderService, persist the new session's cursor and + // incarnation into the binding, merging the runtime payload. startSession: (threadId, input) => fakeStartSession(threadId, input).pipe( - Effect.map((session) => ({ ...session, conversationReset: true })), + Effect.map((session) => ({ + ...session, + resumeCursor: freshCursor, + conversationReset: true, + })), + Effect.tap((session) => + Effect.sync(() => { + binding = { + ...binding, + status: "running", + resumeCursor: session.resumeCursor, + runtimePayload: { + ...(binding.runtimePayload as Record), + sessionIncarnationId: session.sessionIncarnationId, + }, + }; + }), + ), ), sendTurn: (input) => Effect.sync(() => { @@ -1192,8 +1212,15 @@ it.effect("does not send a continuation when the provider restarted the conversa directory: { getBinding: () => Effect.sync(() => Option.some(binding)), upsert: (next) => - Effect.sync(() => { + Effect.gen(function* () { binding = next; + const payload = next.runtimePayload as Record | undefined; + if ( + payload?.sessionIncarnationId !== undefined && + payload.continueAfterServerUpdate === null + ) { + yield* Deferred.succeed(cleared, undefined); + } }), removeExact: () => Effect.die("unused"), recordImportedTranscript: () => Effect.die("unused"), @@ -1202,15 +1229,12 @@ it.effect("does not send a continuation when the provider restarted the conversa listBindings: () => Effect.succeed([]), }, dispatch: (command) => - Effect.gen(function* () { + Effect.sync(() => { dispatched.push(command); - if (command.type === "thread.session.set" && command.session.status === "error") { - yield* Deferred.succeed(settled, undefined); - } return { sequence: dispatched.length }; }), }); - yield* Deferred.await(settled); + yield* Deferred.await(cleared); // A "continue" prompt would reach a conversation with none of the // interrupted turn's context, so the thread asks the user to resend. @@ -1222,10 +1246,15 @@ it.effect("does not send a continuation when the provider restarted the conversa assert.equal(errorSettle.session.sessionIncarnationId, recoveredIncarnationId(thread.id)); assert.match(errorSettle.session.lastError ?? "", /Send your message again/); } + // The new session's cursor and incarnation survive the settlement, so a + // second restart recovers the new session rather than the lost one. + assert.deepStrictEqual(binding.resumeCursor, freshCursor); + assert.equal(binding.status, "running"); assert.deepStrictEqual(binding.runtimePayload, { activeTurnId: null, continueAfterServerUpdate: null, continueAfterServerUpdatePrepared: null, + sessionIncarnationId: recoveredIncarnationId(thread.id), }); }), ); diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index e04c796d6..5268c7d5b 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -458,6 +458,8 @@ export const markRunningProviderSessionsForContinuation = Effect.gen(function* ( const clearContinuationMarkers = ( directory: ProviderSessionDirectory.ProviderSessionDirectory["Service"], threadIds: ReadonlyArray, + // Leave a binding a newer session incarnation already owns untouched. + expectedIncarnationId?: string, ) => Effect.forEach( threadIds, @@ -466,15 +468,26 @@ const clearContinuationMarkers = ( Effect.flatMap( Option.match({ onNone: () => Effect.void, - onSome: (binding) => - directory.upsert({ + onSome: (binding) => { + const boundIncarnationId = readRuntimePayload( + binding.runtimePayload, + ).sessionIncarnationId; + if ( + expectedIncarnationId !== undefined && + typeof boundIncarnationId === "string" && + boundIncarnationId !== expectedIncarnationId + ) { + return Effect.void; + } + return directory.upsert({ ...binding, runtimePayload: { ...readRuntimePayload(binding.runtimePayload), [SERVER_UPDATE_CONTINUATION_KEY]: null, continueAfterServerUpdatePrepared: null, }, - }), + }); + }, }), ), ), @@ -635,6 +648,33 @@ export const reconcileProviderSessions = Effect.gen(function* () { // must keep that id: the recovered runtime stays live, and reverting the // projection to the pre-restart incarnation would fence every later turn. let boundSession: OrchestrationSession = session; + const settleProjectionAsError = (lastError: string) => + Effect.gen(function* () { + const reconciledAt = DateTime.formatIso(yield* DateTime.now); + yield* orchestrationEngine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make(yield* crypto.randomUUIDv4), + threadId: thread.id, + session: { + ...boundSession, + status: "error", + activeTurnId: null, + lastError, + updatedAt: reconciledAt, + }, + createdAt: reconciledAt, + }); + }).pipe( + Effect.retry({ times: 1 }), + Effect.catchCauseIf( + (cause) => !Cause.hasInterrupts(cause), + (cause) => + Effect.logWarning("failed to settle orphaned provider session projection", { + threadId: thread.id, + cause, + }), + ), + ); const settleAsError = (lastError: string) => Effect.gen(function* () { yield* Effect.gen(function* () { @@ -665,32 +705,7 @@ export const reconcileProviderSessions = Effect.gen(function* () { ), ); - yield* Effect.gen(function* () { - const reconciledAt = DateTime.formatIso(yield* DateTime.now); - yield* orchestrationEngine.dispatch({ - type: "thread.session.set", - commandId: CommandId.make(yield* crypto.randomUUIDv4), - threadId: thread.id, - session: { - ...boundSession, - status: "error", - activeTurnId: null, - lastError, - updatedAt: reconciledAt, - }, - createdAt: reconciledAt, - }); - }).pipe( - Effect.retry({ times: 1 }), - Effect.catchCauseIf( - (cause) => !Cause.hasInterrupts(cause), - (cause) => - Effect.logWarning("failed to settle orphaned provider session projection", { - threadId: thread.id, - cause, - }), - ), - ); + yield* settleProjectionAsError(lastError); }); if ( @@ -797,9 +812,12 @@ export const reconcileProviderSessions = Effect.gen(function* () { // The provider lost the interrupted conversation and started a new // one. A continuation prompt would reach a model with no context, // so leave the fresh runtime idle and ask the user to resend. + // ProviderService already persisted the new session's cursor, so + // settle only the projection; the markers clear below against the + // current binding. if (recovered.conversationReset === true) { - yield* settleAsError(CONVERSATION_RESET_AFTER_RESTART_ERROR); - return; + yield* settleProjectionAsError(CONVERSATION_RESET_AFTER_RESTART_ERROR); + return recovered.sessionIncarnationId; } const capabilities = yield* providerService.getCapabilities(providerInstanceId); yield* providerService.sendTurn({ @@ -809,11 +827,12 @@ export const reconcileProviderSessions = Effect.gen(function* () { : { input: SERVER_UPDATE_CONTINUATION_PROMPT }), interactionMode: thread.interactionMode, }); + return recovered.sessionIncarnationId; }); const continuationExit = yield* Effect.exit(continuation); if (Exit.isSuccess(continuationExit) || Cause.hasInterrupts(continuationExit.cause)) { if (Exit.isSuccess(continuationExit)) { - yield* clearContinuationMarkers(directory, [thread.id]).pipe( + yield* clearContinuationMarkers(directory, [thread.id], continuationExit.value).pipe( Effect.uninterruptible, Effect.catchCause((cause) => Effect.logWarning("failed to clear completed provider session continuation", { From a6d3b9663fcdda43ae6f6246239c48b05450caa0 Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Thu, 1 Oct 2026 13:29:20 -0600 Subject: [PATCH 4/4] fix(server): continuation settlement patches only bindings recovery still owns --- .../src/provider/Layers/ClaudeAdapter.ts | 4 +- .../Layers/ProviderSessionDirectory.test.ts | 47 ++++ .../src/provider/claudeNativeHistoryFile.ts | 7 +- .../serverRuntimeStartup.reconcile.test.ts | 236 +++++++++++++++++- apps/server/src/serverRuntimeStartup.ts | 133 +++++++--- 5 files changed, 377 insertions(+), 50 deletions(-) diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 2a28db23a..247961254 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -4891,7 +4891,9 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( // prove nothing; only an observed assistant message does. Without one, // check the history and start fresh only when it is confirmed empty. A // failed or slow check keeps the resume. - // Only cursors Pylon wrote carry a turn count; imported sessions do not. + // Imported cursors carry no turn count, so they are never checked. Any + // cursor the adapter has rewritten has one, so the count only narrows the + // check; the transcript scan is what keeps a saved session resuming. const unprovenResumeSessionId = exactStart === undefined && persistedResumeState?.resume !== undefined && diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts index 593644b40..1e8cd7603 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts @@ -576,6 +576,53 @@ it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryL }), ); + it.effect("lets a commit guard read the binding and patch only the given fields", () => + Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory; + const threadId = ThreadId.make("thread-guarded-patch"); + yield* directory.upsert({ + provider: ProviderDriverKind.make("claudeAgent"), + providerInstanceId: ProviderInstanceId.make("claudeAgent"), + threadId, + status: "running", + resumeCursor: { resume: "current-session" }, + runtimePayload: { sessionIncarnationId: "current", continueAfterServerUpdate: "turn-1" }, + }); + // Startup recovery checks ownership inside the write; the guard's read + // must not wait on the directory's own mutation permit. + const ownedBy = (incarnation: string) => + directory.getBinding(threadId).pipe( + Effect.map( + (binding) => + Option.isSome(binding) && + (binding.value.runtimePayload as Record).sessionIncarnationId === + incarnation, + ), + Effect.orElseSucceed(() => false), + ); + const clearMarker = (incarnation: string) => + directory.upsert( + { + provider: ProviderDriverKind.make("claudeAgent"), + threadId, + runtimePayload: { continueAfterServerUpdate: null }, + }, + { commitGuard: ownedBy(incarnation) }, + ); + + yield* clearMarker("retired"); + expect(Option.getOrThrow(yield* directory.getBinding(threadId))).toMatchObject({ + runtimePayload: { continueAfterServerUpdate: "turn-1" }, + }); + yield* clearMarker("current"); + expect(Option.getOrThrow(yield* directory.getBinding(threadId))).toMatchObject({ + status: "running", + resumeCursor: { resume: "current-session" }, + runtimePayload: { sessionIncarnationId: "current", continueAfterServerUpdate: null }, + }); + }), + ); + it.effect("skips a session binding when its private commit guard retires", () => Effect.gen(function* () { const directory = yield* ProviderSessionDirectory; diff --git a/apps/server/src/provider/claudeNativeHistoryFile.ts b/apps/server/src/provider/claudeNativeHistoryFile.ts index 992877d15..f7d99fbaf 100644 --- a/apps/server/src/provider/claudeNativeHistoryFile.ts +++ b/apps/server/src/provider/claudeNativeHistoryFile.ts @@ -15,6 +15,7 @@ export class ClaudeNativeHistoryUnavailable extends Schema.TaggedError MAX_TRANSCRIPT_SCAN_PROJECTS) + return yield* new ClaudeNativeHistoryUnavailable(); + for (const project of projects) { if (yield* fs.exists(path.join(projectsDir, project, `${input.sessionId}.jsonl`))) return true; } return false; diff --git a/apps/server/src/serverRuntimeStartup.reconcile.test.ts b/apps/server/src/serverRuntimeStartup.reconcile.test.ts index 66432dd91..76ff91c01 100644 --- a/apps/server/src/serverRuntimeStartup.reconcile.test.ts +++ b/apps/server/src/serverRuntimeStartup.reconcile.test.ts @@ -98,6 +98,29 @@ const makeProviderService = ( streamEvents: Stream.empty, }) satisfies ProviderService.ProviderService["Service"]; +// Mirrors ProviderSessionDirectory.upsert: unset fields keep their stored +// values and the runtime payload merges into the stored one. +const mergeBinding = ( + current: ProviderSessionDirectory.ProviderRuntimeBinding | undefined, + next: ProviderSessionDirectory.ProviderRuntimeBinding, +): ProviderSessionDirectory.ProviderRuntimeBinding => { + const payload = (value: unknown) => + value !== null && typeof value === "object" ? (value as Record) : {}; + const status = next.status ?? current?.status; + const resumeCursor = next.resumeCursor !== undefined ? next.resumeCursor : current?.resumeCursor; + const runtimePayload = + next.runtimePayload === undefined + ? current?.runtimePayload + : { ...payload(current?.runtimePayload), ...payload(next.runtimePayload) }; + return { + ...current, + ...next, + ...(status !== undefined ? { status } : {}), + ...(resumeCursor !== undefined ? { resumeCursor } : {}), + ...(runtimePayload !== undefined ? { runtimePayload } : {}), + }; +}; + // A restarted server has no live runtime, so recovery mints a fresh // incarnation for the thread. Continuation must bind this exact id. const recoveredIncarnationId = (threadId: ThreadId) => @@ -328,7 +351,7 @@ it.effect.each( }), upsert: (binding) => Effect.sync(() => { - bindings.set(binding.threadId, binding); + bindings.set(binding.threadId, mergeBinding(bindings.get(binding.threadId), binding)); upserts.push(binding); const clearedCount = upserts.filter((candidate) => { const payload = candidate.runtimePayload; @@ -644,7 +667,8 @@ it.effect("reconciles multiple active and archived orphans but skips live sessio Effect.tap(() => Effect.sync(() => { const orphanIds = [starting.id, running.id, staleActiveTurn.id, archived.id]; - assert.deepStrictEqual(bindingReads, orphanIds); + // Each settle re-reads the binding inside the directory's write guard. + assert.deepStrictEqual([...new Set(bindingReads)], orphanIds); assert.deepStrictEqual( dispatched.map((command) => command.type === "thread.session.set" && command.threadId), orphanIds, @@ -661,6 +685,8 @@ it.effect("reconciles multiple active and archived orphans but skips live sessio orphanIds.map(() => ({ status: "error" as const, activeTurnId: null })), ); assert.equal(upserts.length, orphanIds.length); + // Settling patches only the stopped status and cleared fields; the + // directory merges them, so the cursor and other payload keys stay. for (const binding of upserts) { assert.equal(binding.status, "stopped"); assert.deepStrictEqual( @@ -668,13 +694,12 @@ it.effect("reconciles multiple active and archived orphans but skips live sessio binding.threadId === staleActiveTurn.id ? { activeTurnId: null, - unrelated: binding.threadId, continueAfterServerUpdate: null, continueAfterServerUpdatePrepared: null, } - : { activeTurnId: null, unrelated: binding.threadId }, + : { activeTurnId: null }, ); - assert.deepStrictEqual(binding.resumeCursor, { cursor: binding.threadId }); + assert.equal(binding.resumeCursor, undefined); } }), ), @@ -990,7 +1015,7 @@ for (const preparedStatus of [ getBinding: () => Effect.sync(() => Option.some(binding)), upsert: (next: ProviderSessionDirectory.ProviderRuntimeBinding) => Effect.gen(function* () { - binding = next; + binding = mergeBinding(binding, next); if (binding.status !== "starting" || sends.length === 0) return; yield* Deferred.succeed(cleared, undefined); }), @@ -1098,7 +1123,7 @@ it.effect("settles failed opt-in recovery without retrying the provider turn", ( getBinding: () => Effect.sync(() => Option.some(binding)), upsert: (next) => Effect.sync(() => { - binding = next; + binding = mergeBinding(binding, next); }), removeExact: () => Effect.die("unused"), recordImportedTranscript: () => Effect.die("unused"), @@ -1211,10 +1236,11 @@ it.effect("does not send a continuation when the provider restarted the conversa }, directory: { getBinding: () => Effect.sync(() => Option.some(binding)), - upsert: (next) => + upsert: (next, options) => Effect.gen(function* () { - binding = next; - const payload = next.runtimePayload as Record | undefined; + if (options?.commitGuard !== undefined && !(yield* options.commitGuard)) return; + binding = mergeBinding(binding, next); + const payload = binding.runtimePayload as Record | undefined; if ( payload?.sessionIncarnationId !== undefined && payload.continueAfterServerUpdate === null @@ -1258,3 +1284,193 @@ it.effect("does not send a continuation when the provider restarted the conversa }); }), ); + +// Recovery whose provider start persists a new cursor and incarnation, the way +// ProviderService.startSession does, through a directory that honors write +// guards. `interleave` runs another writer right after the next binding read. +const runRecoveryWithPersistedStart = (input: { + readonly threadName: string; + readonly onSendTurn?: ( + interleave: ( + update: ( + binding: ProviderSessionDirectory.ProviderRuntimeBinding, + ) => ProviderSessionDirectory.ProviderRuntimeBinding, + ) => void, + ) => void; + readonly failRecoveredBinding?: boolean; +}) => + Effect.gen(function* () { + const turnId = TurnId.make(`turn-${input.threadName}`); + const thread = makeThread(input.threadName, "running", turnId); + const freshCursor = { threadId: thread.id, resume: "fresh-session" }; + const done = yield* Deferred.make(); + const errorSettled = yield* Deferred.make(); + const dispatched: OrchestrationCommand[] = []; + let pendingInterleave: + | (( + binding: ProviderSessionDirectory.ProviderRuntimeBinding, + ) => ProviderSessionDirectory.ProviderRuntimeBinding) + | undefined; + let binding: ProviderSessionDirectory.ProviderRuntimeBinding = { + threadId: thread.id, + provider: ProviderDriverKind.make("codex"), + providerInstanceId, + status: "running", + resumeCursor: { threadId: thread.id, resume: "lost-session" }, + runtimePayload: { activeTurnId: turnId }, + }; + const readBinding = () => + Effect.sync(() => { + const snapshot = binding; + const update = pendingInterleave; + pendingInterleave = undefined; + if (update) binding = update(binding); + return Option.some(snapshot); + }); + yield* runReconciliation({ + threads: [thread], + continueAfterRestart: true, + providerService: { + ...makeProviderService(), + getCapabilities: () => + Effect.succeed({ sessionModelSwitch: "in-session", promptlessTurnContinuation: true }), + startSession: (threadId, startInput) => + fakeStartSession(threadId, startInput).pipe( + Effect.map((session) => ({ ...session, resumeCursor: freshCursor })), + Effect.tap((session) => + Effect.sync(() => { + binding = mergeBinding(binding, { + threadId, + provider: session.provider, + status: "running", + resumeCursor: session.resumeCursor, + runtimePayload: { sessionIncarnationId: session.sessionIncarnationId }, + }); + }), + ), + ), + sendTurn: (sendInput) => + Effect.sync(() => { + input.onSendTurn?.((update) => { + pendingInterleave = update; + }); + return { threadId: sendInput.threadId, turnId: TurnId.make("turn-continued") }; + }), + }, + directory: { + getBinding: readBinding, + upsert: (next, options) => + Effect.gen(function* () { + const committed = options?.commitGuard === undefined || (yield* options.commitGuard); + if (committed) binding = mergeBinding(binding, next); + // Signal once a marker clear was attempted, written or guarded off. + const payload = next.runtimePayload as Record | undefined; + if (payload?.continueAfterServerUpdatePrepared === null) { + yield* Deferred.succeed(done, undefined); + } + }), + removeExact: () => Effect.die("unused"), + recordImportedTranscript: () => Effect.die("unused"), + getProvider: () => Effect.die("unused"), + listThreadIds: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), + }, + dispatch: (command) => + Effect.gen(function* () { + dispatched.push(command); + if ( + input.failRecoveredBinding === true && + command.type === "thread.session.set" && + command.session.status === "running" + ) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: "projection unavailable", + }); + } + if (command.type === "thread.session.set" && command.session.status === "error") { + yield* Deferred.succeed(errorSettled, undefined); + } + return { sequence: dispatched.length }; + }), + }); + // A failed recovery settles the projection after its binding patch. + yield* Deferred.await(input.failRecoveredBinding === true ? errorSettled : done); + return { thread, freshCursor, dispatched, binding: () => binding }; + }); + +it.effect("clearing continuation markers keeps a turn that advanced the cursor meanwhile", () => + Effect.gen(function* () { + const advancedCursor = { threadId: "advanced", resume: "fresh-session", turnCount: 2 }; + // A user turn on the recovered incarnation persists between the clear's + // read and its write. + const run = yield* runRecoveryWithPersistedStart({ + threadName: "thread-advanced-during-clear", + onSendTurn: (interleave) => + interleave((current) => + mergeBinding(current, { + ...current, + resumeCursor: advancedCursor, + runtimePayload: { activeTurnId: "turn-user" }, + }), + ), + }); + const binding = run.binding(); + assert.deepStrictEqual(binding.resumeCursor, advancedCursor); + assert.deepStrictEqual(binding.runtimePayload, { + activeTurnId: "turn-user", + continueAfterServerUpdate: null, + continueAfterServerUpdatePrepared: null, + sessionIncarnationId: recoveredIncarnationId(run.thread.id), + }); + }), +); + +it.effect("clearing continuation markers leaves a replacement session's binding alone", () => + Effect.gen(function* () { + const replacementCursor = { threadId: "replacement", resume: "replacement-session" }; + // A replacement session persists its own incarnation and cursor between + // the clear's read and its write. + const run = yield* runRecoveryWithPersistedStart({ + threadName: "thread-replaced-during-clear", + onSendTurn: (interleave) => + interleave((current) => + mergeBinding(current, { + ...current, + resumeCursor: replacementCursor, + runtimePayload: { sessionIncarnationId: "replacement-incarnation" }, + }), + ), + }); + const binding = run.binding(); + assert.deepStrictEqual(binding.resumeCursor, replacementCursor); + assert.equal( + (binding.runtimePayload as Record).sessionIncarnationId, + "replacement-incarnation", + ); + // The recovered incarnation's clear was guarded off. + assert.equal( + (binding.runtimePayload as Record).continueAfterServerUpdatePrepared, + true, + ); + }), +); + +it.effect("a failed recovery settle keeps the cursor the new session persisted", () => + Effect.gen(function* () { + const run = yield* runRecoveryWithPersistedStart({ + threadName: "thread-failed-after-start", + failRecoveredBinding: true, + }); + const binding = run.binding(); + // The recovered runtime is live; the pre-restart cursor must not return. + assert.deepStrictEqual(binding.resumeCursor, run.freshCursor); + assert.equal(binding.status, "running"); + const payload = binding.runtimePayload as Record; + assert.equal(payload.sessionIncarnationId, recoveredIncarnationId(run.thread.id)); + assert.equal(payload.continueAfterServerUpdate, null); + assert.equal(payload.continueAfterServerUpdatePrepared, null); + const settle = run.dispatched.at(-1); + assert.equal(settle?.type === "thread.session.set" && settle.session.status, "error"); + }), +); diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index 5268c7d5b..b4b5769ad 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -455,41 +455,88 @@ export const markRunningProviderSessionsForContinuation = Effect.gen(function* ( ); }).pipe(Effect.mapError(toServerUpdateThreadContinuationError)); +const bindingIncarnationId = (binding: ProviderSessionDirectory.ProviderRuntimeBinding) => { + const value = readRuntimePayload(binding.runtimePayload).sessionIncarnationId; + return typeof value === "string" ? value : undefined; +}; + +const CLEARED_CONTINUATION_MARKERS = { + [SERVER_UPDATE_CONTINUATION_KEY]: null, + continueAfterServerUpdatePrepared: null, +} as const; + +/** + * Writes only the given fields onto the thread's current binding, and only + * while `owns` holds for it. The ownership check repeats inside the + * directory's mutation permit, so a concurrent turn's cursor or a replacement + * session's binding is never overwritten with a stale copy; unlisted fields + * keep their current values and the runtime payload merges. + */ +const patchOwnedBinding = ( + directory: ProviderSessionDirectory.ProviderSessionDirectory["Service"], + threadId: ThreadId, + patch: { + readonly status?: ProviderSessionDirectory.ProviderRuntimeBinding["status"]; + readonly runtimePayload: Record; + }, + owns: (binding: ProviderSessionDirectory.ProviderRuntimeBinding) => boolean, +) => + directory.getBinding(threadId).pipe( + Effect.flatMap( + Option.match({ + onNone: () => Effect.void, + onSome: (current) => + owns(current) + ? directory.upsert( + { + threadId, + provider: current.provider, + ...(current.providerInstanceId !== undefined + ? { providerInstanceId: current.providerInstanceId } + : {}), + ...(patch.status !== undefined ? { status: patch.status } : {}), + runtimePayload: patch.runtimePayload, + }, + { + commitGuard: directory.getBinding(threadId).pipe( + Effect.map( + (latest) => + Option.isSome(latest) && + latest.value.provider === current.provider && + owns(latest.value), + ), + Effect.orElseSucceed(() => false), + ), + }, + ) + : Effect.void, + }), + ), + ); + const clearContinuationMarkers = ( directory: ProviderSessionDirectory.ProviderSessionDirectory["Service"], threadIds: ReadonlyArray, - // Leave a binding a newer session incarnation already owns untouched. + // Leave a binding a newer session incarnation already owns untouched. Every + // provider start records its incarnation, so a binding without one was not + // replaced. expectedIncarnationId?: string, ) => Effect.forEach( threadIds, (threadId) => - directory.getBinding(threadId).pipe( - Effect.flatMap( - Option.match({ - onNone: () => Effect.void, - onSome: (binding) => { - const boundIncarnationId = readRuntimePayload( - binding.runtimePayload, - ).sessionIncarnationId; - if ( - expectedIncarnationId !== undefined && - typeof boundIncarnationId === "string" && - boundIncarnationId !== expectedIncarnationId - ) { - return Effect.void; - } - return directory.upsert({ - ...binding, - runtimePayload: { - ...readRuntimePayload(binding.runtimePayload), - [SERVER_UPDATE_CONTINUATION_KEY]: null, - continueAfterServerUpdatePrepared: null, - }, - }); - }, - }), - ), + patchOwnedBinding( + directory, + threadId, + { runtimePayload: CLEARED_CONTINUATION_MARKERS }, + (binding) => { + const boundIncarnationId = bindingIncarnationId(binding); + return ( + expectedIncarnationId === undefined || + boundIncarnationId === undefined || + boundIncarnationId === expectedIncarnationId + ); + }, ), { concurrency: "unbounded", discard: true }, ); @@ -678,21 +725,31 @@ export const reconcileProviderSessions = Effect.gen(function* () { const settleAsError = (lastError: string) => Effect.gen(function* () { yield* Effect.gen(function* () { - if (Option.isSome(binding)) { - yield* directory.upsert({ - ...binding.value, + if (Option.isNone(binding)) return; + const clearsMarkers = continuationMarkerPresent || interruptedByRestart; + // Recovery may already have started a new incarnation and persisted + // its cursor. Stop only the pre-restart binding; on a replacement, + // clear just the markers and leave the live session's state alone. + const preRestartIncarnationId = bindingIncarnationId(binding.value); + yield* patchOwnedBinding( + directory, + thread.id, + { status: "stopped", runtimePayload: { - ...readRuntimePayload(binding.value.runtimePayload), activeTurnId: null, - ...(continuationMarkerPresent || interruptedByRestart - ? { - [SERVER_UPDATE_CONTINUATION_KEY]: null, - continueAfterServerUpdatePrepared: null, - } - : {}), + ...(clearsMarkers ? CLEARED_CONTINUATION_MARKERS : {}), }, - }); + }, + (current) => bindingIncarnationId(current) === preRestartIncarnationId, + ); + if (clearsMarkers) { + yield* patchOwnedBinding( + directory, + thread.id, + { runtimePayload: CLEARED_CONTINUATION_MARKERS }, + (current) => bindingIncarnationId(current) !== preRestartIncarnationId, + ); } }).pipe( Effect.catchCauseIf(