From a905f33000e262f9e19002cec99e35cd047deef2 Mon Sep 17 00:00:00 2001 From: Chris Deeming Date: Wed, 5 Aug 2026 15:38:20 +0100 Subject: [PATCH] fix(server): recover provider sessions when resume history is missing --- .../Drivers/ClaudeSessionRecovery.test.ts | 241 ++++++++++++++++++ .../provider/Drivers/ClaudeSessionRecovery.ts | 130 ++++++++++ .../src/provider/Layers/ClaudeAdapter.test.ts | 202 ++++++++++++++- .../src/provider/Layers/ClaudeAdapter.ts | 171 ++++++++++++- .../src/provider/Layers/CodexAdapter.test.ts | 24 ++ .../src/provider/Layers/CodexAdapter.ts | 10 + .../Layers/CodexSessionRuntime.test.ts | 1 + .../provider/Layers/CodexSessionRuntime.ts | 24 +- .../src/provider/Layers/CursorAdapter.ts | 12 + .../server/src/provider/Layers/GrokAdapter.ts | 12 + .../provider/Layers/OpenCodeAdapter.test.ts | 13 + .../src/provider/Layers/OpenCodeAdapter.ts | 26 +- .../provider/acp/AcpJsonRpcConnection.test.ts | 36 ++- .../src/provider/acp/AcpSessionRuntime.ts | 64 +++-- .../provider/acp/SessionLoadRecovery.test.ts | 21 ++ 15 files changed, 952 insertions(+), 35 deletions(-) create mode 100644 apps/server/src/provider/Drivers/ClaudeSessionRecovery.test.ts create mode 100644 apps/server/src/provider/Drivers/ClaudeSessionRecovery.ts create mode 100644 apps/server/src/provider/acp/SessionLoadRecovery.test.ts diff --git a/apps/server/src/provider/Drivers/ClaudeSessionRecovery.test.ts b/apps/server/src/provider/Drivers/ClaudeSessionRecovery.test.ts new file mode 100644 index 000000000000..f3da276f882d --- /dev/null +++ b/apps/server/src/provider/Drivers/ClaudeSessionRecovery.test.ts @@ -0,0 +1,241 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, describe, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; + +import { recoverClaudeSession } from "./ClaudeSessionRecovery.ts"; +const encodeTranscriptEntry = Schema.encodeSync( + Schema.fromJsonString(Schema.Struct({ type: Schema.String, cwd: Schema.String })), +); + +describe("recoverClaudeSession", () => { + it.effect("copies an orphaned transcript into the current project directory", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fileSystem.makeTempDirectoryScoped(); + const configDirectory = path.join(root, ".claude"); + const projectsDirectory = path.join(configDirectory, "projects"); + const sourceDirectory = path.join(projectsDirectory, "deleted-worktree"); + const cwd = path.join(root, "project"); + const missingCwd = path.join(root, "deleted-worktree", "apps", "server"); + const sessionId = "11111111-1111-4111-8111-111111111111"; + const sourceTranscript = path.join(sourceDirectory, `${sessionId}.jsonl`); + const transcript = `${encodeTranscriptEntry({ type: "user", cwd: missingCwd })}\n`; + + yield* fileSystem.makeDirectory(sourceDirectory, { recursive: true }); + yield* fileSystem.makeDirectory(cwd, { recursive: true }); + yield* fileSystem.writeFileString(sourceTranscript, transcript); + + const result = yield* recoverClaudeSession({ + environment: { CLAUDE_CONFIG_DIR: configDirectory }, + cwd, + sessionId, + }); + const targetTranscript = path.join( + projectsDirectory, + cwd.replace(/[^a-zA-Z0-9]/g, "-"), + `${sessionId}.jsonl`, + ); + + assert.equal(result, "rehomed"); + assert.equal(yield* fileSystem.readFileString(targetTranscript), transcript); + assert.equal(yield* fileSystem.readFileString(sourceTranscript), transcript); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("leaves a transcript in a live working directory in place", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fileSystem.makeTempDirectoryScoped(); + const configDirectory = path.join(root, ".claude"); + const sourceDirectory = path.join(configDirectory, "projects", "live-worktree"); + const cwd = path.join(root, "project"); + const liveCwd = path.join(root, "live-worktree"); + const sessionId = "22222222-2222-4222-8222-222222222222"; + + yield* fileSystem.makeDirectory(sourceDirectory, { recursive: true }); + yield* fileSystem.makeDirectory(liveCwd, { recursive: true }); + yield* fileSystem.writeFileString( + path.join(sourceDirectory, `${sessionId}.jsonl`), + `${encodeTranscriptEntry({ type: "assistant", cwd: liveCwd })}\n`, + ); + + const result = yield* recoverClaudeSession({ + environment: { CLAUDE_CONFIG_DIR: configDirectory }, + cwd, + sessionId, + }); + + assert.equal(result, "available"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("ignores non-directory entries in the projects directory", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fileSystem.makeTempDirectoryScoped(); + const configDirectory = path.join(root, ".claude"); + const projectsDirectory = path.join(configDirectory, "projects"); + const sourceDirectory = path.join(projectsDirectory, "deleted-worktree"); + const cwd = path.join(root, "project"); + const missingCwd = path.join(root, "deleted-worktree"); + const sessionId = "44444444-4444-4444-8444-444444444444"; + + yield* fileSystem.makeDirectory(sourceDirectory, { recursive: true }); + yield* fileSystem.makeDirectory(cwd, { recursive: true }); + yield* fileSystem.writeFileString(path.join(projectsDirectory, ".DS_Store"), ""); + yield* fileSystem.writeFileString( + path.join(sourceDirectory, `${sessionId}.jsonl`), + `${encodeTranscriptEntry({ type: "user", cwd: missingCwd })}\n`, + ); + + const result = yield* recoverClaudeSession({ + environment: { CLAUDE_CONFIG_DIR: configDirectory }, + cwd, + sessionId, + }); + + assert.equal(result, "rehomed"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("rehomes when the working directory path is now a file", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fileSystem.makeTempDirectoryScoped(); + const configDirectory = path.join(root, ".claude"); + const sourceDirectory = path.join(configDirectory, "projects", "replaced-worktree"); + const cwd = path.join(root, "project"); + const replacedCwd = path.join(root, "replaced-worktree"); + const sessionId = "77777777-7777-4777-8777-777777777777"; + + yield* fileSystem.makeDirectory(sourceDirectory, { recursive: true }); + yield* fileSystem.makeDirectory(cwd, { recursive: true }); + yield* fileSystem.writeFileString(replacedCwd, ""); + yield* fileSystem.writeFileString( + path.join(sourceDirectory, `${sessionId}.jsonl`), + `${encodeTranscriptEntry({ type: "user", cwd: replacedCwd })}\n`, + ); + + const result = yield* recoverClaudeSession({ + environment: { CLAUDE_CONFIG_DIR: configDirectory }, + cwd, + sessionId, + }); + + assert.equal(result, "rehomed"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("rehomes the most recently modified orphaned transcript", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fileSystem.makeTempDirectoryScoped(); + const configDirectory = path.join(root, ".claude"); + const projectsDirectory = path.join(configDirectory, "projects"); + const cwd = path.join(root, "project"); + const sessionId = "88888888-8888-4888-8888-888888888888"; + const olderDirectory = path.join(projectsDirectory, "aaa-old-worktree"); + const newerDirectory = path.join(projectsDirectory, "zzz-new-worktree"); + + yield* fileSystem.makeDirectory(olderDirectory, { recursive: true }); + yield* fileSystem.makeDirectory(newerDirectory, { recursive: true }); + yield* fileSystem.makeDirectory(cwd, { recursive: true }); + yield* fileSystem.writeFileString( + path.join(olderDirectory, `${sessionId}.jsonl`), + `${encodeTranscriptEntry({ type: "user", cwd: path.join(root, "gone-old") })}\n`, + ); + yield* fileSystem.writeFileString( + path.join(newerDirectory, `${sessionId}.jsonl`), + `${encodeTranscriptEntry({ type: "user", cwd: path.join(root, "gone-new") })}\n`, + ); + yield* fileSystem.utimes(path.join(olderDirectory, `${sessionId}.jsonl`), 1_000, 1_000); + + const result = yield* recoverClaudeSession({ + environment: { CLAUDE_CONFIG_DIR: configDirectory }, + cwd, + sessionId, + }); + + assert.equal(result, "rehomed"); + const rehomed = yield* fileSystem.readFileString( + path.join(projectsDirectory, cwd.replace(/[^a-zA-Z0-9]/g, "-"), `${sessionId}.jsonl`), + ); + assert.equal(rehomed.includes("gone-new"), true); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("leaves an undecodable transcript alone rather than rehoming it", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fileSystem.makeTempDirectoryScoped(); + const configDirectory = path.join(root, ".claude"); + const sourceDirectory = path.join(configDirectory, "projects", "unknown-worktree"); + const cwd = path.join(root, "project"); + const sessionId = "66666666-6666-4666-8666-666666666666"; + + yield* fileSystem.makeDirectory(sourceDirectory, { recursive: true }); + yield* fileSystem.makeDirectory(cwd, { recursive: true }); + yield* fileSystem.writeFileString( + path.join(sourceDirectory, `${sessionId}.jsonl`), + "not json\n", + ); + + const result = yield* recoverClaudeSession({ + environment: { CLAUDE_CONFIG_DIR: configDirectory }, + cwd, + sessionId, + }); + + assert.equal(result, "missing"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("reports a missing transcript when no project holds the session", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fileSystem.makeTempDirectoryScoped(); + const configDirectory = path.join(root, ".claude"); + const otherDirectory = path.join(configDirectory, "projects", "other-project"); + const cwd = path.join(root, "project"); + + yield* fileSystem.makeDirectory(otherDirectory, { recursive: true }); + yield* fileSystem.makeDirectory(cwd, { recursive: true }); + yield* fileSystem.writeFileString( + path.join(otherDirectory, "99999999-9999-4999-8999-999999999999.jsonl"), + `${encodeTranscriptEntry({ type: "user", cwd })}\n`, + ); + + const result = yield* recoverClaudeSession({ + environment: { CLAUDE_CONFIG_DIR: configDirectory }, + cwd, + sessionId: "55555555-5555-4555-8555-555555555555", + }); + + assert.equal(result, "missing"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("reports a missing transcript", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const root = yield* fileSystem.makeTempDirectoryScoped(); + const result = yield* recoverClaudeSession({ + environment: { CLAUDE_CONFIG_DIR: root }, + cwd: root, + sessionId: "33333333-3333-4333-8333-333333333333", + }); + + assert.equal(result, "missing"); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); +}); diff --git a/apps/server/src/provider/Drivers/ClaudeSessionRecovery.ts b/apps/server/src/provider/Drivers/ClaudeSessionRecovery.ts new file mode 100644 index 000000000000..f21cbd9ab751 --- /dev/null +++ b/apps/server/src/provider/Drivers/ClaudeSessionRecovery.ts @@ -0,0 +1,130 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeOS from "node:os"; + +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +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 * as Stream from "effect/Stream"; + +const decodeTranscriptEntry = Schema.decodeUnknownExit( + Schema.fromJsonString( + Schema.Struct({ + cwd: Schema.optional(Schema.NullOr(Schema.String)), + }), + ), +); + +export type ClaudeSessionRecoveryResult = "available" | "rehomed" | "missing"; + +function latestTranscriptCwd(transcript: string): { readonly cwd: string | undefined } { + let end = transcript.length; + while (end > 0) { + const newline = transcript.lastIndexOf("\n", end - 1); + const decoded = decodeTranscriptEntry(transcript.slice(newline + 1, end)); + if (Exit.isSuccess(decoded) && typeof decoded.value.cwd === "string") { + return { cwd: decoded.value.cwd }; + } + end = newline; + } + return { cwd: undefined }; +} + +function claudeProjectDirectoryName(cwd: string): string { + return cwd.replace(/[^a-zA-Z0-9]/g, "-"); +} + +const isDirectory = Effect.fn("isDirectory")(function* (candidate: string) { + const fileSystem = yield* FileSystem.FileSystem; + return yield* fileSystem.stat(candidate).pipe( + Effect.map((info) => info.type === "Directory"), + Effect.orElseSucceed(() => false), + ); +}); + +const modifiedAtMillis = Effect.fn("modifiedAtMillis")(function* (candidate: string) { + const fileSystem = yield* FileSystem.FileSystem; + return yield* fileSystem.stat(candidate).pipe( + Effect.map((info) => + Option.match(info.mtime, { onNone: () => 0, onSome: (at) => at.getTime() }), + ), + Effect.orElseSucceed(() => 0), + ); +}); + +const readTranscriptTail = Effect.fn("readTranscriptTail")(function* (transcriptPath: string) { + const fileSystem = yield* FileSystem.FileSystem; + const size = (yield* fileSystem.stat(transcriptPath)).size; + const bytesToRead = size < FileSystem.MiB(8) ? size : FileSystem.MiB(8); + return yield* fileSystem + .stream(transcriptPath, { + offset: size - bytesToRead, + bytesToRead, + }) + .pipe( + Stream.decodeText(), + Stream.runFold( + () => "", + (content, chunk) => content + chunk, + ), + ); +}); + +export const recoverClaudeSession = Effect.fn("recoverClaudeSession")(function* (input: { + readonly environment: NodeJS.ProcessEnv; + readonly cwd: string; + readonly sessionId: string; +}) { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const configured = input.environment.CLAUDE_CONFIG_DIR?.trim(); + const configDirectory = configured + ? path.resolve(input.cwd, configured) + : path.join(NodeOS.homedir(), ".claude"); + const projectsDirectory = path.join(configDirectory, "projects"); + const targetDirectory = path.join( + projectsDirectory, + claudeProjectDirectoryName(path.resolve(input.cwd)), + ); + const targetTranscript = path.join(targetDirectory, `${input.sessionId}.jsonl`); + + if (yield* fileSystem.exists(targetTranscript)) { + return "available"; + } + if (!(yield* fileSystem.exists(projectsDirectory))) { + return "missing"; + } + + const orphanedTranscripts: Array<{ readonly path: string; readonly modifiedAtMillis: number }> = + []; + for (const entry of yield* fileSystem.readDirectory(projectsDirectory)) { + const transcriptPath = path.join(projectsDirectory, entry, `${input.sessionId}.jsonl`); + if (!(yield* fileSystem.exists(transcriptPath).pipe(Effect.orElseSucceed(() => false)))) { + continue; + } + const { cwd: transcriptCwd } = latestTranscriptCwd(yield* readTranscriptTail(transcriptPath)); + if (transcriptCwd === undefined) { + continue; + } + if (yield* isDirectory(transcriptCwd)) { + return "available"; + } + orphanedTranscripts.push({ + path: transcriptPath, + modifiedAtMillis: yield* modifiedAtMillis(transcriptPath), + }); + } + + const sourceTranscript = orphanedTranscripts.sort( + (left, right) => right.modifiedAtMillis - left.modifiedAtMillis, + )[0]?.path; + if (!sourceTranscript) { + return "missing"; + } + + yield* fileSystem.makeDirectory(targetDirectory, { recursive: true }); + yield* fileSystem.copyFile(sourceTranscript, targetTranscript); + return "rehomed"; +}); diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 760f0e7fbab2..d57313658890 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -37,7 +37,12 @@ import { ServerConfig } from "../../config.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; import { ProviderAdapterProcessError, ProviderAdapterValidationError } from "../Errors.ts"; import type { ClaudeAdapterShape } from "../Services/ClaudeAdapter.ts"; -import { makeClaudeAdapter, type ClaudeAdapterLiveOptions } from "./ClaudeAdapter.ts"; +import { + isMissingResumePathError, + workspaceMovedNotice, + makeClaudeAdapter, + type ClaudeAdapterLiveOptions, +} from "./ClaudeAdapter.ts"; const decodeClaudeSettings = Schema.decodeSync(ClaudeSettings); // Test-local service tag so the rest of the file can keep using `yield* ClaudeAdapter`. @@ -156,8 +161,10 @@ function makeHarness(config?: { readonly baseDir?: string; readonly claudeConfig?: Partial; readonly instanceId?: ProviderInstanceId; + readonly failCreateQueryAfter?: number; }) { const query = new FakeClaudeQuery(); + let createQueryCalls = 0; let createInput: | { readonly prompt: AsyncIterable; @@ -169,6 +176,13 @@ function makeHarness(config?: { ...(config?.instanceId ? { instanceId: config.instanceId } : {}), createQuery: (input) => { createInput = input; + createQueryCalls += 1; + if ( + config?.failCreateQueryAfter !== undefined && + createQueryCalls > config.failCreateQueryAfter + ) { + throw new Error("mock createQuery failure"); + } return query; }, ...(config?.nativeEventLogger @@ -202,6 +216,7 @@ function makeHarness(config?: { ), query, getLastCreateQueryInput: () => createInput, + getCreateQueryCalls: () => createQueryCalls, }; } @@ -462,6 +477,36 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("starts fresh and warns when a persisted Claude transcript is missing", () => { + const root = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "claude-session-recovery-")); + const cwd = NodePath.join(root, "project"); + const configDirectory = NodePath.join(root, ".claude"); + NodeFS.mkdirSync(cwd, { recursive: true }); + const harness = makeHarness({ cwd, claudeConfig: { homePath: configDirectory } }); + const staleSessionId = "44444444-4444-4444-8444-444444444444"; + + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + cwd, + runtimeMode: "full-access", + resumeCursor: { resume: staleSessionId }, + }); + const events = Array.from(yield* Stream.runCollect(Stream.take(adapter.streamEvents, 2))); + const resumedSessionId = (session.resumeCursor as { resume?: string }).resume; + + assert.isUndefined(harness.getLastCreateQueryInput()?.options.resume); + assert.notEqual(resumedSessionId, staleSessionId); + assert.equal(events[1]?.type, "runtime.warning"); + }).pipe( + Effect.ensuring(Effect.sync(() => NodeFS.rmSync(root, { recursive: true, force: true }))), + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("maps the Claude Opus 4.7 default effort to the SDK-supported max value", () => { const harness = makeHarness(); return Effect.gen(function* () { @@ -3959,4 +4004,159 @@ describe("ClaudeAdapterLive", () => { Effect.provide(harness.layer), ); }); + it.effect( + "restarts against the original transcript and replays the prompt when a session path is missing", + () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const runtimeEventsFiber = yield* Stream.take(adapter.streamEvents, 10).pipe( + Stream.runCollect, + Effect.forkChild, + ); + + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "auto", + resumeCursor: { + threadId: String(THREAD_ID), + resume: "11111111-1111-4111-8111-111111111111", + }, + }); + + const turn = yield* adapter.sendTurn({ + threadId: session.threadId, + input: "hello", + attachments: [], + }); + + assert.equal(harness.getCreateQueryCalls(), 1); + const startedResumeId = harness.getLastCreateQueryInput()?.options.resume; + assert.equal( + harness.getLastCreateQueryInput()?.options.resume, + "11111111-1111-4111-8111-111111111111", + ); + + harness.query.emit({ + type: "result", + subtype: "error_during_execution", + is_error: true, + errors: ['Path "/tmp/gone-worktree" does not exist'], + session_id: "22222222-2222-4222-8222-222222222222", + uuid: "result-0", + } as unknown as SDKMessage); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + + assert.equal(harness.getCreateQueryCalls(), 2); + assert.equal(harness.getLastCreateQueryInput()?.options.resume, startedResumeId); + assert.deepEqual( + runtimeEvents + .filter((event) => event.type === "runtime.warning") + .map((event) => event.payload.message), + [ + "A directory from the previous session was deleted. Reopened this session in the current workspace.", + ], + ); + assert.deepEqual( + runtimeEvents + .filter((event) => event.type === "turn.started") + .map((event) => event.turnId), + [turn.turnId], + ); + assert.deepEqual( + runtimeEvents + .filter((event) => event.type === "turn.started") + .map((event) => event.turnId), + [turn.turnId], + ); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }, + ); + + it.effect("completes the replayed turn when the restart fails", () => { + const harness = makeHarness({ failCreateQueryAfter: 1 }); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const runtimeEventsFiber = yield* Stream.take(adapter.streamEvents, 8).pipe( + Stream.runCollect, + Effect.forkChild, + ); + + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "auto", + resumeCursor: { + threadId: String(THREAD_ID), + resume: "11111111-1111-4111-8111-111111111111", + }, + }); + + const turn = yield* adapter.sendTurn({ + threadId: session.threadId, + input: "hello", + attachments: [], + }); + + harness.query.emit({ + type: "result", + subtype: "error_during_execution", + is_error: true, + errors: ['Path "/tmp/gone-worktree" does not exist'], + session_id: "22222222-2222-4222-8222-222222222222", + uuid: "result-0", + } as unknown as SDKMessage); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + + assert.deepEqual( + runtimeEvents + .filter((event) => event.type === "turn.completed") + .map((event) => event.turnId), + [turn.turnId], + ); + assert.equal( + runtimeEvents.some((event) => event.type === "runtime.error"), + true, + ); + assert.deepEqual( + runtimeEvents + .filter((event) => event.type === "session.exited") + .map((event) => event.payload.exitKind), + ["error"], + ); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); +}); + +describe("isMissingResumePathError", () => { + it("matches the Claude CLI missing path failure", () => { + assert.equal( + isMissingResumePathError('Path "/Users/me/Sites/main/.claude/worktrees/gone" does not exist'), + true, + ); + }); + + it("ignores unrelated failures", () => { + assert.equal(isMissingResumePathError("Claude turn failed."), false); + assert.equal(isMissingResumePathError("Thread does not exist"), false); + }); +}); + +describe("workspaceMovedNotice", () => { + it("names the current workspace and flags stale paths", () => { + const notice = workspaceMovedNotice("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/repo/main"); + + assert.equal(notice.includes("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/repo/main"), true); + assert.equal(notice.includes("no longer exists"), true); + assert.equal(notice.includes("may be stale"), true); + }); }); diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 548ffe6a896d..939321e2e400 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -36,6 +36,7 @@ import { type ProviderRuntimeTurnStatus, type ProviderSendTurnInput, type ProviderSession, + type ProviderSessionStartInput, type ThreadTokenUsageSnapshot, type ProviderUserInputAnswers, type RuntimeContentStreamKind, @@ -72,6 +73,10 @@ import { ServerConfig } from "../../config.ts"; import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; import { resolveClaudeSdkExecutablePath } from "../Drivers/ClaudeExecutable.ts"; import { makeClaudeEnvironment } from "../Drivers/ClaudeHome.ts"; +import { + recoverClaudeSession, + type ClaudeSessionRecoveryResult, +} from "../Drivers/ClaudeSessionRecovery.ts"; import { getClaudeModelCapabilities, isClaudeUltracodeEffort, @@ -201,6 +206,12 @@ interface ClaudeSessionContext { lastKnownTotalProcessedTokens: number | undefined; lastAssistantUuid: string | undefined; lastThreadStartedId: string | undefined; + lastPrompt: SDKUserMessage | undefined; + readonly startInput: ProviderSessionStartInput; + readonly startResumeCursor: unknown; + readonly sessionRecovery: ClaudeSessionRecoveryResult; + resumePathRetried: boolean; + restarting: boolean; stopped: boolean; } @@ -232,6 +243,36 @@ function isSyntheticClaudeThreadId(value: string): boolean { return value.startsWith("claude-thread-"); } +export function isMissingResumePathError(message: string): boolean { + return /^Path ".+" does not exist$/.test(message.trim()); +} + +function restartTurnState(turnId: TurnId, startedAt: string): ClaudeTurnState { + return { + turnId, + startedAt, + items: [], + assistantTextBlocks: new Map(), + assistantTextBlockOrder: [], + capturedProposedPlanKeys: new Set(), + nextSyntheticAssistantBlockIndex: -1, + }; +} + +export function workspaceMovedNotice(cwd: string): string { + return [ + "[t3code] The working directory used earlier in this session no longer exists.", + `This session now runs in ${cwd}.`, + "Paths, branches, and worktrees mentioned earlier may be stale — check the current state before relying on them.", + ].join(" "); +} + +const SESSION_DIRECTORY_MISSING_MESSAGE = + "A directory from the previous session was deleted. Reopened this session in the current workspace."; + +const SESSION_HISTORY_LOST_MESSAGE = + "Previous session history is unavailable. Started a fresh session without earlier context."; + function hasDurableClaudeSessionId(message: SDKMessage): boolean { if (message.type !== "system") { return true; @@ -2545,6 +2586,86 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( yield* updateResumeCursor(context); }); + const restartAfterMissingPath = Effect.fn("restartAfterMissingPath")(function* ( + context: ClaudeSessionContext, + errorMessage: string, + replayTurnId: TurnId | undefined, + ) { + const abandonReplayedTurn = Effect.fn("abandonReplayedTurn")(function* ( + failure: Cause.Cause | undefined, + ) { + yield* emitRuntimeError( + context, + errorMessage, + failure !== undefined ? { cause: failure } : undefined, + ); + if (replayTurnId !== undefined) { + context.turnState = restartTurnState(replayTurnId, yield* nowIso); + yield* completeTurn(context, "failed", errorMessage); + } + context.restarting = false; + yield* stopSessionInternal(context, { emitExitEvent: false }); + + const exitStamp = yield* makeEventStamp(); + yield* offerRuntimeEvent({ + type: "session.exited", + eventId: exitStamp.eventId, + provider: PROVIDER, + createdAt: exitStamp.createdAt, + threadId: context.session.threadId, + payload: { + reason: errorMessage, + exitKind: "error", + }, + providerRefs: {}, + }); + }); + + const prompt = context.lastPrompt; + if (prompt === undefined || replayTurnId === undefined) { + yield* abandonReplayedTurn(undefined); + return; + } + + yield* stopSessionInternal(context, { emitExitEvent: false }); + + const restarted = yield* startSession({ + ...context.startInput, + resumeCursor: context.startResumeCursor, + }).pipe(Effect.catchCause((cause) => abandonReplayedTurn(cause).pipe(Effect.as(undefined)))); + if (restarted === undefined) { + return; + } + const next = sessions.get(restarted.threadId); + if (next === undefined) { + yield* abandonReplayedTurn(undefined); + return; + } + next.resumePathRetried = true; + next.lastPrompt = prompt; + if (next.sessionRecovery !== "missing") { + yield* emitRuntimeWarning(next, SESSION_DIRECTORY_MISSING_MESSAGE); + } + + next.turnState = restartTurnState(replayTurnId, yield* nowIso); + next.session = { + ...next.session, + status: "running", + activeTurnId: replayTurnId, + updatedAt: yield* nowIso, + }; + const workspaceNotice = next.session.cwd; + if (workspaceNotice !== undefined) { + yield* Queue.offer(next.promptQueue, { + type: "message", + message: buildUserMessage({ + sdkContent: [{ type: "text", text: workspaceMovedNotice(workspaceNotice) }], + }), + }); + } + yield* Queue.offer(next.promptQueue, { type: "message", message: prompt }); + }); + const handleResultMessage = Effect.fn("handleResultMessage")(function* ( context: ClaudeSessionContext, message: SDKMessage, @@ -2556,6 +2677,21 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const status = turnStatusFromResult(message); const errorMessage = message.subtype === "success" ? undefined : message.errors[0]; + if ( + status === "failed" && + errorMessage !== undefined && + isMissingResumePathError(errorMessage) && + !context.resumePathRetried && + context.lastPrompt !== undefined + ) { + context.resumePathRetried = true; + context.restarting = true; + const replayTurnId = context.turnState?.turnId; + context.turnState = undefined; + yield* restartAfterMissingPath(context, errorMessage, replayTurnId).pipe(Effect.forkDetach); + return; + } + if (status === "failed") { yield* emitRuntimeError(context, errorMessage ?? "Claude turn failed."); } @@ -2997,7 +3133,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( context: ClaudeSessionContext, exit: Exit.Exit, ) { - if (context.stopped) { + if (context.stopped || context.restarting) { return; } @@ -3170,7 +3306,25 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const startedAt = yield* nowIso; const resumeState = readClaudeResumeState(input.resumeCursor); const threadId = input.threadId; - const existingResumeSessionId = resumeState?.resume; + const requestedResumeSessionId = resumeState?.resume; + const sessionRecovery = + requestedResumeSessionId && input.cwd + ? yield* recoverClaudeSession({ + environment: claudeEnvironment, + cwd: input.cwd, + sessionId: requestedResumeSessionId, + }).pipe( + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + Effect.catchCause((cause) => + Effect.logWarning("claude.session.recovery.failed", { threadId, cause }).pipe( + Effect.as("available" as const), + ), + ), + ) + : "available"; + const existingResumeSessionId = + sessionRecovery === "missing" ? undefined : requestedResumeSessionId; const newSessionId = existingResumeSessionId === undefined ? yield* randomUUIDv4 : undefined; const sessionId = existingResumeSessionId ?? newSessionId; @@ -3640,6 +3794,12 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( lastKnownTotalProcessedTokens: undefined, lastAssistantUuid: resumeState?.resumeSessionAt, lastThreadStartedId: undefined, + lastPrompt: undefined, + startInput: input, + startResumeCursor: session.resumeCursor, + sessionRecovery, + resumePathRetried: false, + restarting: false, stopped: false, }; yield* Ref.set(contextRef, context); @@ -3652,10 +3812,14 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( provider: PROVIDER, createdAt: sessionStartedStamp.createdAt, threadId, - payload: input.resumeCursor !== undefined ? { resume: input.resumeCursor } : {}, + payload: existingResumeSessionId ? { resume: input.resumeCursor } : {}, providerRefs: {}, }); + if (sessionRecovery === "missing" && input.resumeCursor !== undefined) { + yield* emitRuntimeWarning(context, SESSION_HISTORY_LOST_MESSAGE); + } + const configuredStamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "session.configured", @@ -3808,6 +3972,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( boundInstanceId, }); + context.lastPrompt = message; yield* Queue.offer(context.promptQueue, { type: "message", message, diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 7b8fbec56660..d1a57ede8458 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -795,6 +795,30 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { }), ); + it.effect("maps a session resume fallback to runtime.warning", () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + + yield* runtime.emit({ + id: asEventId("evt-session-resume-fallback"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:00.000Z", + method: "t3/session-resume-fallback", + message: + "Codex session could not be resumed. A fresh session was started without its previous in-session context.", + }); + + const firstEvent = yield* Fiber.join(firstEventFiber); + NodeAssert.equal(firstEvent._tag, "Some"); + if (firstEvent._tag === "Some") { + NodeAssert.equal(firstEvent.value.type, "runtime.warning"); + } + }), + ); + it.effect("maps realtime started notifications with upstream realtime session ids", () => Effect.gen(function* () { const { adapter, runtime } = yield* startLifecycleRuntime(); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 4146121b1474..34058d10d119 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -498,6 +498,16 @@ function mapToRuntimeEvents( event: ProviderEvent, canonicalThreadId: ThreadId, ): ReadonlyArray { + if (event.method === "t3/session-resume-fallback" && event.message) { + return [ + { + ...runtimeEventBase(event, canonicalThreadId), + type: "runtime.warning", + payload: { message: event.message }, + }, + ]; + } + if (event.kind === "error") { if (!event.message) { return []; diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index d7346a0e0dbe..8d364cc4f185 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -426,6 +426,7 @@ describe("openCodexThread", () => { }); NodeAssert.equal(opened.thread.id, "fresh-thread"); + NodeAssert.equal(opened.resumeFallback, true); NodeAssert.deepStrictEqual( calls.map((call) => call.method), ["thread/resume", "thread/start"], diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 67108dd4dbba..c75d7729caaa 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -444,6 +444,9 @@ export function isRecoverableThreadResumeError(error: unknown): boolean { type CodexThreadOpenResponse = | CodexRpc.ClientRequestResponsesByMethod["thread/start"] | CodexRpc.ClientRequestResponsesByMethod["thread/resume"]; +type CodexThreadOpenResult = CodexThreadOpenResponse & { + readonly resumeFallback: boolean; +}; type CodexThreadOpenMethod = "thread/start" | "thread/resume"; @@ -462,7 +465,7 @@ export const openCodexThread = (input: { readonly requestedModel: string | undefined; readonly serviceTier: CodexServiceTier | undefined; readonly resumeThreadId: string | undefined; -}): Effect.Effect => { +}): Effect.Effect => { const resumeThreadId = input.resumeThreadId; const startParams = buildThreadStartParams({ cwd: input.cwd, @@ -472,7 +475,9 @@ export const openCodexThread = (input: { }); if (resumeThreadId === undefined) { - return input.client.request("thread/start", startParams); + return input.client + .request("thread/start", startParams) + .pipe(Effect.map((opened) => ({ ...opened, resumeFallback: false }))); } return input.client @@ -481,6 +486,7 @@ export const openCodexThread = (input: { ...startParams, }) .pipe( + Effect.map((opened) => ({ ...opened, resumeFallback: false })), Effect.catchIf(isRecoverableThreadResumeError, (error) => Effect.logWarning("codex app-server thread resume fell back to fresh start", { threadId: input.threadId, @@ -488,7 +494,10 @@ export const openCodexThread = (input: { resumeThreadId, recoverable: true, cause: error, - }).pipe(Effect.andThen(input.client.request("thread/start", startParams))), + }).pipe( + Effect.andThen(input.client.request("thread/start", startParams)), + Effect.map((opened) => ({ ...opened, resumeFallback: true })), + ), ), ); }; @@ -1239,6 +1248,15 @@ export const makeCodexSessionRuntime = ( updatedAt: yield* nowIso, } satisfies ProviderSession; yield* Ref.set(sessionRef, session); + if (opened.resumeFallback) { + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + method: "t3/session-resume-fallback", + message: + "Codex session could not be resumed. A fresh session was started without its previous in-session context.", + }); + } yield* emitSessionEvent("session/ready", "Codex App Server session ready."); return session; }); diff --git a/apps/server/src/provider/Layers/CursorAdapter.ts b/apps/server/src/provider/Layers/CursorAdapter.ts index 80475a5c2695..54c824faf85c 100644 --- a/apps/server/src/provider/Layers/CursorAdapter.ts +++ b/apps/server/src/provider/Layers/CursorAdapter.ts @@ -888,6 +888,18 @@ export function makeCursorAdapter( threadId: input.threadId, payload: { resume: started.initializeResult }, }); + if (started.resumeFallback) { + yield* offerRuntimeEvent({ + type: "runtime.warning", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + payload: { + message: + "Cursor session could not be resumed. A fresh session was started without its previous in-session context.", + }, + }); + } yield* offerRuntimeEvent({ type: "session.state.changed", ...(yield* makeEventStamp()), diff --git a/apps/server/src/provider/Layers/GrokAdapter.ts b/apps/server/src/provider/Layers/GrokAdapter.ts index 977cc8caaddb..64d6c163d565 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.ts @@ -890,6 +890,18 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte threadId: input.threadId, payload: { resume: started.initializeResult }, }); + if (started.resumeFallback) { + yield* offerRuntimeEvent({ + type: "runtime.warning", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + payload: { + message: + "Grok session could not be resumed. A fresh session was started without its previous in-session context.", + }, + }); + } yield* offerRuntimeEvent({ type: "session.state.changed", ...(yield* makeEventStamp()), diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index 1385ccbaabec..2bc1fadb28c2 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -409,6 +409,19 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { schemaVersion: 1, sessionId: "http://127.0.0.1:9999/session", }); + const events = Array.from( + yield* Stream.runCollect( + adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.takeUntil((event) => event.type === "thread.started"), + ), + ), + ); + const warning = events.find((event) => event.type === "runtime.warning"); + NodeAssert.equal(warning?.type, "runtime.warning"); + if (warning?.type === "runtime.warning") { + NodeAssert.match(warning.payload.message, /fresh session/i); + } yield* adapter.stopSession(threadId); }), diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index 73c23b77e686..6d0196fa379f 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -1265,7 +1265,7 @@ export function makeOpenCodeAdapter( permission: buildOpenCodePermissionRules(input.runtimeMode), }), ); - return { openCodeSession: reusable, created: false }; + return { openCodeSession: reusable, created: false, resumeFallback: false }; } // The session lives under a different cwd (e.g. the thread @@ -1292,14 +1292,9 @@ export function makeOpenCodeAdapter( permission: buildOpenCodePermissionRules(input.runtimeMode), }), ); - return { openCodeSession: forked, created: true }; + return { openCodeSession: forked, created: true, resumeFallback: false }; } - if (resumeSessionId) { - yield* Effect.logWarning( - `OpenCode session '${resumeSessionId}' no longer exists; starting a fresh session.`, - ); - } const createdSession = yield* runOpenCodeSdk("session.create", () => client.session.create({ permission: buildOpenCodePermissionRules(input.runtimeMode), @@ -1311,7 +1306,11 @@ export function makeOpenCodeAdapter( detail: "OpenCode session.create returned no session payload.", }); } - return { openCodeSession: createdSession.data, created: true }; + return { + openCodeSession: createdSession.data, + created: true, + resumeFallback: resumeSessionId !== undefined, + }; }); return { @@ -1320,6 +1319,7 @@ export function makeOpenCodeAdapter( client, openCodeSession: resolved.openCodeSession, created: resolved.created, + resumeFallback: resolved.resumeFallback, }; }).pipe(Effect.provideService(Scope.Scope, sessionScope)), ); @@ -1397,6 +1397,16 @@ export function makeOpenCodeAdapter( message: "OpenCode session started", }, }); + if (started.resumeFallback) { + yield* emit({ + ...(yield* buildEventBase({ threadId: input.threadId })), + type: "runtime.warning", + payload: { + message: + "OpenCode session could not be resumed. A fresh session was started without its previous in-session context.", + }, + }); + } yield* emit({ ...(yield* buildEventBase({ threadId: input.threadId })), type: "thread.started", diff --git a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts index b1ef0d3e5953..f8cb5382a31d 100644 --- a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts +++ b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts @@ -7,6 +7,7 @@ import * as NodeFS from "node:fs"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { it } from "@effect/vitest"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Option from "effect/Option"; import * as TestClock from "effect/testing/TestClock"; @@ -502,12 +503,13 @@ describe("AcpSessionRuntime", () => { ); }); - it.effect("fails session startup when session/load returns an error", () => + it.effect("starts a fresh session when session/load returns an error", () => Effect.gen(function* () { const runtime = yield* AcpSessionRuntime.AcpSessionRuntime; - const error = yield* runtime.start().pipe(Effect.flip); + const started = yield* runtime.start(); - expect(error._tag).toBe("AcpRequestError"); + expect(started.sessionId).toBe("mock-session-1"); + expect(started.resumeFallback).toBe(true); }).pipe( Effect.provide( AcpSessionRuntime.layer({ @@ -529,6 +531,34 @@ describe("AcpSessionRuntime", () => { ), ); + it.effect("does not fall back to a fresh session when the request logger dies", () => + Effect.gen(function* () { + const runtime = yield* AcpSessionRuntime.AcpSessionRuntime; + const exit = yield* Effect.exit(runtime.start()); + + expect(Exit.isFailure(exit)).toBe(true); + }).pipe( + Effect.provide( + AcpSessionRuntime.layer({ + authMethodId: "test", + spawn: { + command: mockAgentCommand, + args: mockAgentArgs, + }, + cwd: process.cwd(), + resumeSessionId: "stale-session-id", + clientInfo: { name: "t3-test", version: "0.0.0" }, + requestLogger: (event) => + event.method === "session/load" && event.status === "succeeded" + ? Effect.die(new Error("logger exploded")) + : Effect.void, + }), + ), + Effect.scoped, + Effect.provide(NodeServices.layer), + ), + ); + it.effect("ignores session/update replay notifications during session/load", () => Effect.gen(function* () { const runtime = yield* AcpSessionRuntime.AcpSessionRuntime; diff --git a/apps/server/src/provider/acp/AcpSessionRuntime.ts b/apps/server/src/provider/acp/AcpSessionRuntime.ts index 09fce6d56f9d..4df09ba0fc57 100644 --- a/apps/server/src/provider/acp/AcpSessionRuntime.ts +++ b/apps/server/src/provider/acp/AcpSessionRuntime.ts @@ -47,6 +47,17 @@ export interface AcpSessionEventStreamBarrier { export type AcpSessionRuntimeEvent = AcpParsedSessionEvent | AcpSessionEventStreamBarrier; +export function sessionLoadFailureIsRecoverable( + cause: Cause.Cause, +): boolean { + if (Cause.hasInterruptsOnly(cause)) { + return false; + } + return !cause.reasons.some( + (reason) => Cause.isFailReason(reason) && reason.error._tag === "AcpTransportError", + ); +} + const defaultSessionLoadTimeout = Duration.seconds(90); const defaultSessionLoadReplayIdleGap = Duration.seconds(2); @@ -88,6 +99,7 @@ export interface AcpSessionRequestLogEvent { export interface AcpSessionRuntimeStartResult { readonly sessionId: string; + readonly resumeFallback: boolean; readonly initializeResult: EffectAcpSchema.InitializeResponse; readonly sessionSetupResult: | EffectAcpSchema.LoadSessionResponse @@ -556,6 +568,18 @@ export const make = ( | EffectAcpSchema.LoadSessionResponse | EffectAcpSchema.NewSessionResponse | EffectAcpSchema.ResumeSessionResponse; + let resumeFallback = false; + const createSession = Effect.fn("AcpSessionRuntime.createSession")(function* () { + const createPayload = { + cwd: options.cwd, + mcpServers: options.mcpServers ?? [], + } satisfies EffectAcpSchema.NewSessionRequest; + return yield* runLoggedRequest( + "session/new", + createPayload, + acp.agent.createSession(createPayload), + ); + }); if (options.resumeSessionId) { const loadPayload = { sessionId: options.resumeSessionId, @@ -610,14 +634,6 @@ export const make = ( onSome: Effect.succeed, }), ), - Effect.tap((result) => - logRequest({ - method: "session/load", - payload: loadPayload, - status: "succeeded", - result, - }), - ), Effect.onError((cause) => logRequest({ method: "session/load", @@ -626,20 +642,33 @@ export const make = ( cause, }), ), + Effect.catchCause((cause) => + sessionLoadFailureIsRecoverable(cause) + ? Effect.gen(function* () { + yield* Effect.logWarning("ACP session load failed; starting a fresh session", { + sessionId: options.resumeSessionId, + cause, + }); + const created = yield* createSession(); + sessionId = created.sessionId; + resumeFallback = true; + return created; + }) + : Effect.failCause(cause), + ), ); + yield* logRequest({ + method: "session/load", + payload: loadPayload, + status: "succeeded", + result: loaded, + }); + return loaded; }).pipe(Effect.ensuring(Ref.set(sessionLoadGateRef, Option.none()))); } else { - const createPayload = { - cwd: options.cwd, - mcpServers: options.mcpServers ?? [], - } satisfies EffectAcpSchema.NewSessionRequest; - const created = yield* runLoggedRequest( - "session/new", - createPayload, - acp.agent.createSession(createPayload), - ); + const created = yield* createSession(); sessionId = created.sessionId; sessionSetupResult = created; } @@ -649,6 +678,7 @@ export const make = ( const nextState = { sessionId, + resumeFallback, initializeResult, sessionSetupResult, modelConfigId: extractModelConfigId(sessionSetupResult), diff --git a/apps/server/src/provider/acp/SessionLoadRecovery.test.ts b/apps/server/src/provider/acp/SessionLoadRecovery.test.ts new file mode 100644 index 000000000000..d698fd7674fd --- /dev/null +++ b/apps/server/src/provider/acp/SessionLoadRecovery.test.ts @@ -0,0 +1,21 @@ +import { assert, describe, it } from "@effect/vitest"; +import * as Cause from "effect/Cause"; +import * as EffectAcpErrors from "effect-acp/errors"; + +import { sessionLoadFailureIsRecoverable } from "./AcpSessionRuntime.ts"; + +describe("sessionLoadFailureIsRecoverable", () => { + it("does not recover from the session/load timeout", () => { + const timeout = new EffectAcpErrors.AcpTransportError({ + operation: "call-rpc", + method: "session/load", + detail: "session/load timed out waiting for RPC response or replay idle gap", + cause: undefined, + }); + assert.equal(sessionLoadFailureIsRecoverable(Cause.fail(timeout)), false); + }); + + it("recovers from an agent defect", () => { + assert.equal(sessionLoadFailureIsRecoverable(Cause.die(new Error("Invalid params"))), true); + }); +});