From 5ae8a4f4f98cc74e772da8bee62c825a744baa2d Mon Sep 17 00:00:00 2001 From: Vito Cappello Date: Sun, 26 Jul 2026 23:11:39 -0400 Subject: [PATCH] fix: preserve Claude cache during stalled CLI recovery (#113866) * Preserve Claude cache during stalled CLI recovery * fix(agents): harden Claude stall recovery * fix(agents): reject partial Claude recovery output * fix(agents): rewind Claude recovery to a safe checkpoint * fix(agents): preserve checkpointed fork retries * fix(agents): cold-reseed downgraded Claude forks --------- Co-authored-by: VACInc <3279061+VACInc@users.noreply.github.com> Co-authored-by: Peter Steinberger --- extensions/anthropic/cli-backend.ts | 3 + src/agents/cli-output.test.ts | 49 ++ src/agents/cli-output.ts | 53 +- src/agents/cli-runner.reliability.test.ts | 620 +++++++++++++++++- src/agents/cli-runner.ts | 161 ++++- ...aude-live-session.background-tasks.test.ts | 92 +++ src/agents/cli-runner/claude-live-session.ts | 35 +- src/agents/cli-runner/execute.ts | 1 + .../helpers.system-prompt-resume.test.ts | 25 +- src/agents/cli-runner/helpers.ts | 7 + src/agents/cli-runner/types.ts | 16 +- src/agents/cli-session.ts | 3 + .../command/attempt-execution.cli.test.ts | 226 ++++++- src/agents/command/attempt-execution.ts | 62 +- src/agents/command/session-store.test.ts | 11 +- src/agents/command/session-store.ts | 26 +- src/config/sessions/cli-session-binding.ts | 1 + src/config/sessions/types.ts | 2 + src/plugins/cli-backend.types.ts | 2 + 19 files changed, 1325 insertions(+), 70 deletions(-) diff --git a/extensions/anthropic/cli-backend.ts b/extensions/anthropic/cli-backend.ts index aaacc32f7766..7fc8c0cc4a94 100644 --- a/extensions/anthropic/cli-backend.ts +++ b/extensions/anthropic/cli-backend.ts @@ -166,6 +166,9 @@ export function buildAnthropicCliBackend(): CliBackendPlugin { "{sessionId}", ], forkArg: "--fork-session", + // Claude Code 2.1.209+ exposes this hidden print-mode flag, and stream-json + // emits the matching transcript UUID on assistant records. + resumeAtArg: "--resume-session-at", output: "jsonl", liveSession: "claude-stdio", input: "stdin", diff --git a/src/agents/cli-output.test.ts b/src/agents/cli-output.test.ts index 7b31d177d4d4..44098c69b591 100644 --- a/src/agents/cli-output.test.ts +++ b/src/agents/cli-output.test.ts @@ -757,6 +757,55 @@ describe("parseCliJsonl", () => { }); }); + it("captures the last Claude assistant transcript UUID as a resume checkpoint", () => { + const result = parseCliJsonl( + [ + JSON.stringify({ type: "system", subtype: "init", session_id: "session-checkpoint" }), + JSON.stringify({ + type: "assistant", + uuid: "assistant-checkpoint-1", + message: { + id: "provider-message-1", + role: "assistant", + content: [{ type: "text", text: "first" }], + }, + }), + JSON.stringify({ + type: "assistant", + uuid: "assistant-checkpoint-2", + message: { + id: "provider-message-2", + role: "assistant", + content: [{ type: "text", text: "done" }], + }, + }), + JSON.stringify({ + type: "assistant", + uuid: "subagent-checkpoint", + parent_tool_use_id: "tool-use-1", + message: { + id: "provider-subagent-message", + role: "assistant", + content: [{ type: "text", text: "nested" }], + }, + }), + JSON.stringify({ + type: "result", + session_id: "session-checkpoint", + result: "done", + }), + ].join("\n"), + { + command: "claude", + output: "jsonl", + sessionIdFields: ["session_id"], + }, + "claude-cli", + ); + + expect(result?.resumeCheckpointId).toBe("assistant-checkpoint-2"); + }); + it("preserves Claude session metadata even when the final result text is empty", () => { const result = parseCliJsonl( [ diff --git a/src/agents/cli-output.ts b/src/agents/cli-output.ts index c0ada261b5b3..02f432dd94aa 100644 --- a/src/agents/cli-output.ts +++ b/src/agents/cli-output.ts @@ -46,6 +46,8 @@ export type CliOutput = { text: string; rawText?: string; sessionId?: string; + /** Backend-owned assistant boundary that can safely anchor a later resumed fork. */ + resumeCheckpointId?: string; usage?: CliUsage; /** Terminal cumulative turn usage for diagnostics; reply accounting keeps using `usage`. */ diagnosticUsage?: CliUsage; @@ -499,6 +501,22 @@ function pickCliSessionId( return undefined; } +function pickCliResumeCheckpointId(params: { + backend: CliBackendConfig; + providerId: string; + parsed: Record; +}): string | undefined { + if ( + !isClaudeStreamJsonDialect(params) || + params.parsed.type !== "assistant" || + params.parsed.parent_tool_use_id != null + ) { + return undefined; + } + const checkpointId = typeof params.parsed.uuid === "string" ? params.parsed.uuid.trim() : ""; + return checkpointId || undefined; +} + function shouldUnwrapNestedCliResultText(params: { providerId?: string; parsed: Record; @@ -1191,6 +1209,7 @@ export function createCliJsonlStreamingParser(params: { let assistantText = ""; let pendingClaudeText = ""; let sessionId: string | undefined; + let resumeCheckpointId: string | undefined; let usage: CliUsage | undefined; let diagnosticUsage: CliUsage | undefined; let output: CliOutput | null = null; @@ -1265,6 +1284,7 @@ export function createCliJsonlStreamingParser(params: { usage = nextUsage ?? usage; } if (parsed.type === "assistant" && isRecord(parsed.message)) { + resumeCheckpointId = pickCliResumeCheckpointId({ ...params, parsed }) ?? resumeCheckpointId; params.onAssistantMessage?.(parsed.message); } const geminiErrorText = isGeminiStreamJsonDialect(params) @@ -1316,6 +1336,7 @@ export function createCliJsonlStreamingParser(params: { output = { ...result, text, + ...(resumeCheckpointId ? { resumeCheckpointId } : {}), ...(diagnosticUsage ? { diagnosticUsage } : {}), }; return; @@ -1517,10 +1538,17 @@ export function createCliJsonlStreamingParser(params: { return output; } if (isStreamJsonDialect(params) && assistantText.trim()) { - return { text: assistantText.trim(), sessionId, usage }; + return { + text: assistantText.trim(), + sessionId, + usage, + ...(resumeCheckpointId ? { resumeCheckpointId } : {}), + }; } const text = texts.join("\n").trim(); - return text ? { text, sessionId, usage } : null; + return text + ? { text, sessionId, usage, ...(resumeCheckpointId ? { resumeCheckpointId } : {}) } + : null; }, }; } @@ -1537,6 +1565,7 @@ function parseCliJsonl( return null; } let sessionId: string | undefined; + let resumeCheckpointId: string | undefined; let usage: CliUsage | undefined; const texts: string[] = []; let streamJsonText = ""; @@ -1549,6 +1578,8 @@ function parseCliJsonl( if (!sessionId && typeof parsed.thread_id === "string") { sessionId = parsed.thread_id.trim(); } + resumeCheckpointId = + pickCliResumeCheckpointId({ backend, providerId, parsed }) ?? resumeCheckpointId; const nextUsage = readCliUsage(parsed); const shouldUseUsage = !isClaudeStreamJsonResult({ backend, providerId, parsed }) || !usage; if (shouldUseUsage) { @@ -1589,11 +1620,18 @@ function parseCliJsonl( }); if (claudeResult) { if (claudeResult.text || claudeResult.errorText) { - return claudeResult; + return { + ...claudeResult, + ...(resumeCheckpointId ? { resumeCheckpointId } : {}), + }; } // Live sessions reparse the completed JSONL transcript, so preserve // streamed text here as well as in the incremental parser above. - return { ...claudeResult, text: streamJsonText.trim() || texts.join("\n").trim() }; + return { + ...claudeResult, + text: streamJsonText.trim() || texts.join("\n").trim(), + ...(resumeCheckpointId ? { resumeCheckpointId } : {}), + }; } const claudeDelta = parseClaudeCliStreamingDelta({ @@ -1622,7 +1660,12 @@ function parseCliJsonl( return { text: "", sessionId, usage, errorText: geminiErrorText }; } if (streamJsonDialect && (streamJsonText.trim() || sawGeminiStructuredOutput)) { - return { text: streamJsonText.trim(), sessionId, usage }; + return { + text: streamJsonText.trim(), + sessionId, + usage, + ...(resumeCheckpointId ? { resumeCheckpointId } : {}), + }; } if (streamJsonDialect) { return { text: "", sessionId, usage, errorText: CLI_STREAM_JSON_MISSING_RESULT_ERROR }; diff --git a/src/agents/cli-runner.reliability.test.ts b/src/agents/cli-runner.reliability.test.ts index 32e806308ffe..04ba945f47c1 100644 --- a/src/agents/cli-runner.reliability.test.ts +++ b/src/agents/cli-runner.reliability.test.ts @@ -624,6 +624,233 @@ describe("runCliAgent reliability", () => { expect(clearBeforeRetry).not.toHaveBeenCalled(); }); + it("keeps cold transcript reseed for stalled sessions without a checkpoint", async () => { + supervisorSpawnMock.mockClear(); + supervisorSpawnMock + .mockResolvedValueOnce( + createManagedRun({ + reason: "no-output-timeout", + exitCode: null, + exitSignal: "SIGKILL", + durationMs: 200, + stdout: "", + stderr: "", + timedOut: true, + noOutputTimedOut: true, + }), + ) + .mockResolvedValueOnce( + createManagedRun({ + reason: "exit", + exitCode: 0, + exitSignal: null, + durationMs: 50, + stdout: "fresh fallback", + stderr: "", + timedOut: false, + noOutputTimedOut: false, + }), + ); + const prepareForkRetry = vi.fn(async () => true); + const clearBeforeRetry = vi.fn(async () => true); + const context = buildPreparedContext({ + sessionKey: "agent:main:no-checkpoint", + runId: "run-no-checkpoint", + cliSessionId: "legacy-session", + provider: "claude-cli", + model: "opus", + openClawHistoryPrompt: CLI_RESEED_PROMPT, + }); + context.preparedBackend.backend = { + ...context.preparedBackend.backend, + resumeArgs: ["--resume", "{sessionId}"], + forkArg: "--fork-session", + resumeAtArg: "--resume-session-at", + }; + + const result = await runPreparedCliAgent({ + ...context, + params: { + ...context.params, + onBeforeForkedCliSessionRetry: prepareForkRetry, + onBeforeFreshCliSessionRetry: clearBeforeRetry, + }, + }); + + expect(result.payloads).toEqual([{ text: "fresh fallback" }]); + expect(prepareForkRetry).not.toHaveBeenCalled(); + expect(clearBeforeRetry).toHaveBeenCalledOnce(); + expect(supervisorSpawnMock).toHaveBeenCalledTimes(2); + const freshArgv = requireArray( + requireRecord( + callArg(supervisorSpawnMock, 1, 0, "fresh fallback spawn"), + "fresh fallback spawn", + ).argv, + "fresh fallback argv", + ); + expect(freshArgv).not.toContain("--fork-session"); + expect(freshArgv).not.toContain("--resume-session-at"); + }); + + it("falls back to cold reseed when Claude lacks the checkpoint flag", async () => { + supervisorSpawnMock.mockClear(); + supervisorSpawnMock + .mockResolvedValueOnce( + createManagedRun({ + reason: "no-output-timeout", + exitCode: null, + exitSignal: "SIGKILL", + durationMs: 200, + stdout: "", + stderr: "", + timedOut: true, + noOutputTimedOut: true, + }), + ) + .mockResolvedValueOnce( + createManagedRun({ + reason: "exit", + exitCode: 1, + exitSignal: null, + durationMs: 25, + stdout: "", + stderr: "error: unknown option '--resume-session-at'", + timedOut: false, + noOutputTimedOut: false, + }), + ) + .mockResolvedValueOnce( + createManagedRun({ + reason: "exit", + exitCode: 0, + exitSignal: null, + durationMs: 50, + stdout: "fresh fallback", + stderr: "", + timedOut: false, + noOutputTimedOut: false, + }), + ); + const prepareForkRetry = vi.fn(async () => true); + const claimFork = vi.fn(async () => true); + const restoreFork = vi.fn(async () => {}); + const clearBeforeRetry = vi.fn(async () => true); + const context = buildPreparedContext({ + sessionKey: "agent:main:old-claude", + runId: "run-old-claude", + cliSessionId: "old-claude-session", + provider: "claude-cli", + model: "opus", + openClawHistoryPrompt: CLI_RESEED_PROMPT, + }); + context.preparedBackend.backend = { + ...context.preparedBackend.backend, + resumeArgs: ["--resume", "{sessionId}"], + forkArg: "--fork-session", + resumeAtArg: "--resume-session-at", + }; + context.params.cliSessionBinding = { + sessionId: "old-claude-session", + resumeCheckpointId: "assistant-before-stall", + }; + + const result = await runPreparedCliAgent({ + ...context, + params: { + ...context.params, + onBeforeForkedCliSessionRetry: prepareForkRetry, + claimCliSessionFork: claimFork, + restoreCliSessionFork: restoreFork, + persistCliSessionForkSuccessor: vi.fn(async () => {}), + onBeforeFreshCliSessionRetry: clearBeforeRetry, + }, + }); + + expect(result.payloads).toEqual([{ text: "fresh fallback" }]); + expect(prepareForkRetry).toHaveBeenCalledOnce(); + expect(claimFork).toHaveBeenCalledOnce(); + expect(restoreFork).toHaveBeenCalledOnce(); + expect(clearBeforeRetry).toHaveBeenCalledWith({ + provider: "claude-cli", + reason: "timeout", + sessionId: "old-claude-session", + }); + expect(supervisorSpawnMock).toHaveBeenCalledTimes(3); + }); + + it("cold reseeds an initially armed checkpoint after a Claude downgrade", async () => { + supervisorSpawnMock.mockClear(); + supervisorSpawnMock + .mockResolvedValueOnce( + createManagedRun({ + reason: "exit", + exitCode: 1, + exitSignal: null, + durationMs: 25, + stdout: "", + stderr: "error: unknown option '--resume-session-at'", + timedOut: false, + noOutputTimedOut: false, + }), + ) + .mockResolvedValueOnce( + createManagedRun({ + reason: "exit", + exitCode: 0, + exitSignal: null, + durationMs: 50, + stdout: "fresh fallback", + stderr: "", + timedOut: false, + noOutputTimedOut: false, + }), + ); + const claimFork = vi.fn(async () => true); + const restoreFork = vi.fn(async () => {}); + const clearBeforeRetry = vi.fn(async () => true); + const context = buildPreparedContext({ + sessionKey: "agent:main:downgraded-claude", + runId: "run-downgraded-claude", + cliSessionId: "downgraded-session", + provider: "claude-cli", + model: "opus", + openClawHistoryPrompt: CLI_RESEED_PROMPT, + }); + context.preparedBackend.backend = { + ...context.preparedBackend.backend, + resumeArgs: ["--resume", "{sessionId}"], + forkArg: "--fork-session", + resumeAtArg: "--resume-session-at", + }; + context.params.cliSessionBinding = { + sessionId: "downgraded-session", + resumeCheckpointId: "assistant-before-stall", + forkNextResume: true, + }; + + const result = await runPreparedCliAgent({ + ...context, + params: { + ...context.params, + forkCliSessionOnResume: true, + claimCliSessionFork: claimFork, + restoreCliSessionFork: restoreFork, + persistCliSessionForkSuccessor: vi.fn(async () => {}), + onBeforeFreshCliSessionRetry: clearBeforeRetry, + }, + }); + + expect(result.payloads).toEqual([{ text: "fresh fallback" }]); + expect(claimFork).toHaveBeenCalledOnce(); + expect(restoreFork).toHaveBeenCalledOnce(); + expect(clearBeforeRetry).toHaveBeenCalledWith({ + provider: "claude-cli", + reason: "session_expired", + sessionId: "downgraded-session", + }); + expect(supervisorSpawnMock).toHaveBeenCalledTimes(2); + }); + it("preserves fresh retry for direct CLI callers without a pre-clear hook", async () => { supervisorSpawnMock.mockClear(); supervisorSpawnMock.mockResolvedValueOnce( @@ -1768,7 +1995,7 @@ describe("runCliAgent reliability", () => { expect(clearBeforeRetry).not.toHaveBeenCalled(); }); - it("keeps non-capture live-session artifacts through fresh recovery retry", async () => { + it("forks a synthetic-stalled resume without rebuilding its cached conversation", async () => { vi.useFakeTimers(); supervisorSpawnMock.mockClear(); const transcriptProbe = vi.fn(async () => false); @@ -1796,12 +2023,14 @@ describe("runCliAgent reliability", () => { notifyFirstSpawn = resolve; }); let spawnCount = 0; + const spawnedArgv: string[][] = []; supervisorSpawnMock.mockImplementation(async (...args: unknown[]) => { spawnCount += 1; const input = args[0] as { argv?: string[]; onStdout?: (chunk: string) => void; }; + spawnedArgv.push(input.argv ?? []); expect(resolveArg(input.argv, "--mcp-config")).toBe(mcpConfigPath); expect(resolveArg(input.argv, "--skills-plugin-dir")).toBe(skillsDir); expect(fs.existsSync(mcpConfigPath)).toBe(true); @@ -1809,6 +2038,7 @@ describe("runCliAgent reliability", () => { if (spawnCount === 1) { notifyFirstSpawn?.(); + const stdoutListener = input.onStdout; let resolveExit: ((value: RunExit) => void) | undefined; const exited = new Promise((resolve) => { resolveExit = resolve; @@ -1818,7 +2048,27 @@ describe("runCliAgent reliability", () => { pid: 3301, startedAtMs: Date.now(), stdin: { - write: vi.fn((dataValue: string, cb?: (err?: Error | null) => void) => cb?.()), + write: vi.fn((dataValue: string, cb?: (err?: Error | null) => void) => { + stdoutListener?.( + [ + JSON.stringify({ + type: "system", + subtype: "init", + session_id: "stale-live", + }), + JSON.stringify({ + type: "assistant", + session_id: "stale-live", + message: { + model: "", + role: "assistant", + content: [{ type: "text", text: "No response requested." }], + }, + }), + ].join("\n") + "\n", + ); + cb?.(); + }), end: vi.fn(), }, wait: vi.fn(() => exited), @@ -1839,15 +2089,25 @@ describe("runCliAgent reliability", () => { const stdoutListener = input.onStdout; return { - runId: "live-retry-fresh", + runId: "live-retry-fork", pid: 3302, startedAtMs: Date.now(), stdin: { write: vi.fn((dataValue: string, cb?: (err?: Error | null) => void) => { stdoutListener?.( [ - JSON.stringify({ type: "system", subtype: "init", session_id: "fresh-live" }), - JSON.stringify({ type: "result", session_id: "fresh-live", result: "fresh ok" }), + JSON.stringify({ type: "system", subtype: "init", session_id: "forked-live" }), + JSON.stringify({ + type: "assistant", + uuid: "assistant-after-recovery", + session_id: "forked-live", + message: { + model: "claude-fable-5", + role: "assistant", + content: [{ type: "text", text: "fork ok" }], + }, + }), + JSON.stringify({ type: "result", session_id: "forked-live", result: "fork ok" }), ].join("\n") + "\n", ); cb?.(); @@ -1881,6 +2141,8 @@ describe("runCliAgent reliability", () => { "--skills-plugin-dir", skillsDir, ], + forkArg: "--fork-session", + resumeAtArg: "--resume-session-at", output: "jsonl" as const, input: "stdin" as const, modelArg: "--model", @@ -1898,6 +2160,10 @@ describe("runCliAgent reliability", () => { const cleanup = vi.fn(async () => { fs.rmSync(artifactDir, { recursive: true, force: true }); }); + const prepareForkRetry = vi.fn(async () => true); + const claimFork = vi.fn(async () => true); + const persistForkSuccessor = vi.fn(async () => {}); + const restoreFork = vi.fn(async () => {}); const clearBeforeRetry = vi.fn(async () => true); const context = buildPreparedContext({ sessionKey: "agent:main:live-artifacts", @@ -1910,12 +2176,20 @@ describe("runCliAgent reliability", () => { context.preparedBackend.backend = liveBackend; context.preparedBackend.cleanup = cleanup; context.backendResolved.config = liveBackend; + context.params.cliSessionBinding = { + sessionId: "stale-live", + resumeCheckpointId: "assistant-before-stall", + }; const resultPromise = runPreparedCliAgent({ ...context, params: { ...context.params, timeoutMs: 5_000, + onBeforeForkedCliSessionRetry: prepareForkRetry, + claimCliSessionFork: claimFork, + persistCliSessionForkSuccessor: persistForkSuccessor, + restoreCliSessionFork: restoreFork, onBeforeFreshCliSessionRetry: clearBeforeRetry, }, }); @@ -1923,20 +2197,348 @@ describe("runCliAgent reliability", () => { await vi.advanceTimersByTimeAsync(1_000); const result = await resultPromise; - expect(result.payloads).toEqual([{ text: "fresh ok" }]); - expect(result.meta.finalPromptText).toContain("User: earlier context"); - expect(result.meta.agentMeta?.cliSessionBinding?.sessionId).toBe("fresh-live"); + expect(result.payloads).toEqual([{ text: "fork ok" }]); + expect(result.meta.finalPromptText).not.toContain("User: earlier context"); + expect(result.meta.agentMeta?.cliSessionBinding?.sessionId).toBe("forked-live"); + expect(result.meta.agentMeta?.cliSessionBinding?.resumeCheckpointId).toBe( + "assistant-after-recovery", + ); expect(transcriptProbe).not.toHaveBeenCalled(); expect(supervisorSpawnMock).toHaveBeenCalledTimes(2); - expect(clearBeforeRetry).toHaveBeenCalledWith({ + expect(spawnedArgv[0]).not.toContain("--fork-session"); + expect(spawnedArgv[1]).toEqual( + expect.arrayContaining([ + "--resume", + "stale-live", + "--fork-session", + "--resume-session-at", + "assistant-before-stall", + ]), + ); + expect(prepareForkRetry).toHaveBeenCalledWith({ provider: "claude-cli", reason: "timeout", sessionId: "stale-live", }); + expect(claimFork).toHaveBeenCalledOnce(); + expect(persistForkSuccessor).toHaveBeenCalledWith("forked-live"); + expect(restoreFork).not.toHaveBeenCalled(); + expect(clearBeforeRetry).not.toHaveBeenCalled(); expect(cleanup).toHaveBeenCalledOnce(); expect(fs.existsSync(artifactDir)).toBe(false); }); + it("falls back to transcript reseeding when the cache-preserving fork also stalls", async () => { + vi.useFakeTimers(); + supervisorSpawnMock.mockClear(); + const spawnedArgv: string[][] = []; + let notifyFirstSpawn: (() => void) | undefined; + const firstSpawned = new Promise((resolve) => { + notifyFirstSpawn = resolve; + }); + let notifySecondSpawn: (() => void) | undefined; + const secondSpawned = new Promise((resolve) => { + notifySecondSpawn = resolve; + }); + supervisorSpawnMock.mockImplementation(async (...args: unknown[]) => { + const input = args[0] as { + argv?: string[]; + onStdout?: (chunk: string) => void; + }; + spawnedArgv.push(input.argv ?? []); + const spawnIndex = spawnedArgv.length; + if (spawnIndex === 1) { + notifyFirstSpawn?.(); + } else if (spawnIndex === 2) { + notifySecondSpawn?.(); + } + let resolveExit: ((value: RunExit) => void) | undefined; + const exited = new Promise((resolve) => { + resolveExit = resolve; + }); + return { + runId: `live-fork-fallback-${spawnIndex}`, + pid: 3400 + spawnIndex, + startedAtMs: Date.now(), + stdin: { + write: vi.fn((dataValue: string, cb?: (err?: Error | null) => void) => { + const sessionId = + spawnIndex === 1 + ? "stalled-source" + : spawnIndex === 2 + ? "forked-before-stall" + : "fresh-after-fork-stall"; + input.onStdout?.( + [ + JSON.stringify({ type: "system", subtype: "init", session_id: sessionId }), + ...(spawnIndex < 3 + ? [ + JSON.stringify({ + type: "assistant", + session_id: sessionId, + message: { + model: "", + role: "assistant", + content: [{ type: "text", text: "No response requested." }], + }, + }), + ] + : [ + JSON.stringify({ + type: "result", + session_id: sessionId, + result: "fresh fallback ok", + }), + ]), + ].join("\n") + "\n", + ); + cb?.(); + }), + end: vi.fn(), + }, + wait: vi.fn(() => exited), + cancel: vi.fn(() => + resolveExit?.({ + reason: "manual-cancel", + exitCode: null, + exitSignal: null, + durationMs: 1, + stdout: "", + stderr: "", + timedOut: false, + noOutputTimedOut: false, + }), + ), + }; + }); + + const backend = { + command: "claude", + args: ["-p", "--output-format", "stream-json"], + resumeArgs: ["-p", "--resume", "{sessionId}", "--output-format", "stream-json"], + forkArg: "--fork-session", + resumeAtArg: "--resume-session-at", + output: "jsonl" as const, + input: "stdin" as const, + modelArg: "--model", + sessionArgs: ["--session-id", "{sessionId}"], + sessionMode: "always" as const, + liveSession: "claude-stdio" as const, + reliability: { + watchdog: { + resume: { noOutputTimeoutMs: 1_000, minMs: 1_000, maxMs: 1_000 }, + fresh: { noOutputTimeoutMs: 1_000, minMs: 1_000, maxMs: 1_000 }, + }, + }, + serialize: true, + }; + const context = buildPreparedContext({ + sessionKey: "agent:main:fork-then-fresh", + runId: "run-fork-then-fresh", + cliSessionId: "stalled-source", + provider: "claude-cli", + model: "opus", + openClawHistoryPrompt: CLI_RESEED_PROMPT, + }); + context.preparedBackend.backend = backend; + context.backendResolved.config = backend; + context.params.cliSessionBinding = { + sessionId: "stalled-source", + resumeCheckpointId: "assistant-before-stall", + }; + const prepareForkRetry = vi.fn(async () => true); + const claimFork = vi.fn(async () => true); + const persistForkSuccessor = vi.fn(async () => {}); + const restoreFork = vi.fn(async () => {}); + const clearBeforeRetry = vi.fn(async () => true); + + const resultPromise = runPreparedCliAgent({ + ...context, + params: { + ...context.params, + timeoutMs: 5_000, + onBeforeForkedCliSessionRetry: prepareForkRetry, + claimCliSessionFork: claimFork, + persistCliSessionForkSuccessor: persistForkSuccessor, + restoreCliSessionFork: restoreFork, + onBeforeFreshCliSessionRetry: clearBeforeRetry, + }, + }); + await firstSpawned; + await vi.advanceTimersByTimeAsync(1_000); + await secondSpawned; + await vi.advanceTimersByTimeAsync(1_000); + const result = await resultPromise; + + expect(result.payloads).toEqual([{ text: "fresh fallback ok" }]); + expect(result.meta.finalPromptText).toContain("User: earlier context"); + expect(spawnedArgv).toHaveLength(3); + expect(spawnedArgv[1]).toEqual( + expect.arrayContaining([ + "--resume", + "stalled-source", + "--fork-session", + "--resume-session-at", + "assistant-before-stall", + ]), + ); + expect(spawnedArgv[2]).not.toContain("--resume"); + expect(spawnedArgv[2]).not.toContain("--fork-session"); + expect(prepareForkRetry).toHaveBeenCalledOnce(); + expect(claimFork).toHaveBeenCalledOnce(); + expect(persistForkSuccessor).toHaveBeenCalledWith("forked-before-stall"); + expect(restoreFork).not.toHaveBeenCalled(); + expect(clearBeforeRetry).toHaveBeenCalledWith({ + provider: "claude-cli", + reason: "timeout", + sessionId: "forked-before-stall", + }); + }); + + it("tracks and clears a successor when the initial attempt is already a fork", async () => { + vi.useFakeTimers(); + supervisorSpawnMock.mockClear(); + const spawnedArgv: string[][] = []; + let notifyFirstSpawn: (() => void) | undefined; + const firstSpawned = new Promise((resolve) => { + notifyFirstSpawn = resolve; + }); + supervisorSpawnMock.mockImplementation(async (...args: unknown[]) => { + const input = args[0] as { + argv?: string[]; + onStdout?: (chunk: string) => void; + }; + spawnedArgv.push(input.argv ?? []); + const spawnIndex = spawnedArgv.length; + if (spawnIndex === 1) { + notifyFirstSpawn?.(); + } + let resolveExit: ((value: RunExit) => void) | undefined; + const exited = new Promise((resolve) => { + resolveExit = resolve; + }); + return { + runId: `initial-fork-fallback-${spawnIndex}`, + pid: 3500 + spawnIndex, + startedAtMs: Date.now(), + stdin: { + write: vi.fn((dataValue: string, cb?: (err?: Error | null) => void) => { + const sessionId = + spawnIndex === 1 ? "initial-fork-successor" : "fresh-after-initial-fork-stall"; + input.onStdout?.( + [ + JSON.stringify({ type: "system", subtype: "init", session_id: sessionId }), + ...(spawnIndex === 1 + ? [] + : [ + JSON.stringify({ + type: "result", + session_id: sessionId, + result: "initial fork fallback ok", + }), + ]), + ].join("\n") + "\n", + ); + cb?.(); + }), + end: vi.fn(), + }, + wait: vi.fn(() => exited), + cancel: vi.fn(() => + resolveExit?.({ + reason: "manual-cancel", + exitCode: null, + exitSignal: null, + durationMs: 1, + stdout: "", + stderr: "", + timedOut: false, + noOutputTimedOut: false, + }), + ), + }; + }); + + const backend = { + command: "claude", + args: ["-p", "--output-format", "stream-json"], + resumeArgs: ["-p", "--resume", "{sessionId}", "--output-format", "stream-json"], + forkArg: "--fork-session", + resumeAtArg: "--resume-session-at", + output: "jsonl" as const, + input: "stdin" as const, + modelArg: "--model", + sessionArgs: ["--session-id", "{sessionId}"], + sessionMode: "always" as const, + liveSession: "claude-stdio" as const, + reliability: { + watchdog: { + resume: { noOutputTimeoutMs: 1_000, minMs: 1_000, maxMs: 1_000 }, + fresh: { noOutputTimeoutMs: 1_000, minMs: 1_000, maxMs: 1_000 }, + }, + }, + serialize: true, + }; + const context = buildPreparedContext({ + sessionKey: "agent:main:initial-fork-fallback", + runId: "run-initial-fork-fallback", + cliSessionId: "initial-fork-parent", + provider: "claude-cli", + model: "opus", + openClawHistoryPrompt: CLI_RESEED_PROMPT, + }); + context.preparedBackend.backend = backend; + context.backendResolved.config = backend; + context.params.cliSessionBinding = { + sessionId: "initial-fork-parent", + resumeCheckpointId: "assistant-before-initial-fork", + forkNextResume: true, + }; + const claimFork = vi.fn(async () => true); + const persistForkSuccessor = vi.fn(async () => {}); + const restoreFork = vi.fn(async () => {}); + const clearBeforeRetry = vi.fn(async () => true); + + const resultPromise = runPreparedCliAgent({ + ...context, + params: { + ...context.params, + timeoutMs: 5_000, + forkCliSessionOnResume: true, + claimCliSessionFork: claimFork, + persistCliSessionForkSuccessor: persistForkSuccessor, + restoreCliSessionFork: restoreFork, + onBeforeFreshCliSessionRetry: clearBeforeRetry, + }, + }); + await firstSpawned; + await vi.advanceTimersByTimeAsync(1_000); + const result = await resultPromise; + + expect(result.payloads).toEqual([{ text: "initial fork fallback ok" }]); + expect(result.meta.finalPromptText).toContain("User: earlier context"); + expect(spawnedArgv).toHaveLength(2); + expect(spawnedArgv[0]).toEqual( + expect.arrayContaining([ + "--resume", + "initial-fork-parent", + "--fork-session", + "--resume-session-at", + "assistant-before-initial-fork", + ]), + ); + expect(spawnedArgv[1]).not.toContain("--resume"); + expect(spawnedArgv[1]).not.toContain("--fork-session"); + expect(claimFork).toHaveBeenCalledOnce(); + expect(persistForkSuccessor).toHaveBeenCalledWith("initial-fork-successor"); + expect(restoreFork).not.toHaveBeenCalled(); + expect(clearBeforeRetry).toHaveBeenCalledWith({ + provider: "claude-cli", + reason: "timeout", + sessionId: "initial-fork-successor", + }); + }); + it("does not fresh retry a no-output timeout after CLI diagnostic output", async () => { supervisorSpawnMock.mockClear(); enqueueSystemEventMock.mockClear(); diff --git a/src/agents/cli-runner.ts b/src/agents/cli-runner.ts index 97b5e494527a..ac7b8166269d 100644 --- a/src/agents/cli-runner.ts +++ b/src/agents/cli-runner.ts @@ -136,6 +136,20 @@ function shouldRetryFreshCliSessionAfterFailover(params: { } } +function shouldRetryForkedCliSessionAfterFailover(error: FailoverError): boolean { + return error.reason === "timeout" && error.code === "cli_no_output_timeout"; +} + +function isUnsupportedCliResumeAtError(error: unknown, resumeAtArg: string): boolean { + const message = formatErrorMessage(error).toLowerCase(); + return ( + message.includes(resumeAtArg.toLowerCase()) && + ["unknown", "unexpected", "unrecognized", "not recognized"].some((token) => + message.includes(token), + ) + ); +} + function formatCliEmptyOutputDiagnostics(output: CliOutput): string | undefined { const process = output.diagnostics?.process; if (!process) { @@ -909,15 +923,45 @@ export async function runPreparedCliAgent( throw error; }; - const executeCliAttempt = async (cliSessionIdToUse?: string, timeoutMs = params.timeoutMs) => { + const executeCliAttempt = async ( + cliSessionIdToUse?: string, + options?: { + timeoutMs?: number; + forkCliSessionOnResume?: boolean; + resumeAt?: string; + onForkSuccessorPersisted?: (sessionId: string) => void; + }, + ) => { + const timeoutMs = options?.timeoutMs ?? params.timeoutMs; + const forkCliSessionOnResume = + options?.forkCliSessionOnResume ?? context.params.forkCliSessionOnResume; + const cliSessionResumeAt = + cliSessionIdToUse && forkCliSessionOnResume + ? (options?.resumeAt ?? + context.params.cliSessionResumeAt ?? + context.params.cliSessionBinding?.resumeCheckpointId) + : undefined; + const persistCliSessionForkSuccessor = + options?.onForkSuccessorPersisted && context.params.persistCliSessionForkSuccessor + ? async (sessionId: string) => { + await context.params.persistCliSessionForkSuccessor?.(sessionId); + options.onForkSuccessorPersisted?.(sessionId); + } + : context.params.persistCliSessionForkSuccessor; const attemptContext = - timeoutMs === params.timeoutMs + timeoutMs === params.timeoutMs && + forkCliSessionOnResume === context.params.forkCliSessionOnResume && + cliSessionResumeAt === context.params.cliSessionResumeAt && + persistCliSessionForkSuccessor === context.params.persistCliSessionForkSuccessor ? context : { ...context, params: { ...context.params, timeoutMs, + forkCliSessionOnResume, + cliSessionResumeAt, + persistCliSessionForkSuccessor, }, }; diagnosticLifecycle?.setPhase("send"); @@ -1139,6 +1183,9 @@ export async function runPreparedCliAgent( ...(context.effectiveAuthProfileId ? { authProfileId: context.effectiveAuthProfileId } : {}), + ...(resultParams.output.resumeCheckpointId + ? { resumeCheckpointId: resultParams.output.resumeCheckpointId } + : {}), ...(context.authEpoch ? { authEpoch: context.authEpoch } : {}), authEpochVersion: context.authEpochVersion, ...(context.extraSystemPromptHash @@ -1340,9 +1387,20 @@ export async function runPreparedCliAgent( hookRunner, }); const reusableCliSessionId = resolveReusableCliSessionId(context.reusableCliSession); + const resumeCheckpointId = params.cliSessionBinding?.resumeCheckpointId; + let retryableSessionId = reusableCliSessionId; try { return await finishCliAttempt( - await executeCliAttempt(reusableCliSessionId), + await executeCliAttempt( + reusableCliSessionId, + params.forkCliSessionOnResume + ? { + onForkSuccessorPersisted: (sessionId) => { + retryableSessionId = sessionId; + }, + } + : undefined, + ), reusableCliSessionId, ); } catch (err) { @@ -1350,11 +1408,77 @@ export async function runPreparedCliAgent( if (deliveredFailure) { return deliveredFailure; } - if (isFailoverError(err)) { - const retryableSessionId = reusableCliSessionId; + let recoveryError = err; + if ( + params.forkCliSessionOnResume && + resumeCheckpointId && + context.preparedBackend.backend.resumeAtArg && + isUnsupportedCliResumeAtError(err, context.preparedBackend.backend.resumeAtArg) + ) { + recoveryError = new FailoverError("CLI backend cannot resume from the stored checkpoint.", { + reason: "session_expired", + provider: params.provider, + model: context.modelId, + sessionId: params.sessionId, + lane: params.lane, + status: resolveFailoverStatus("session_expired"), + cause: err, + }); + } + if (isFailoverError(recoveryError)) { if ( + !params.forkCliSessionOnResume && + shouldRetryForkedCliSessionAfterFailover(recoveryError) && + retryableSessionId && + resumeCheckpointId && + params.sessionKey && + context.preparedBackend.backend.forkArg && + context.preparedBackend.backend.resumeAtArg && + params.onBeforeForkedCliSessionRetry + ) { + try { + const retryTimeoutMs = params.timeoutMs - (Date.now() - context.started); + if (retryTimeoutMs <= 0) { + throw recoveryError; + } + const forkPrepared = await params.onBeforeForkedCliSessionRetry({ + provider: params.provider, + reason: recoveryError.reason, + sessionId: retryableSessionId, + }); + if (!forkPrepared) { + throw recoveryError; + } + cliBackendLog.warn( + `cli session recovery fork: provider=${params.provider} reason=${recoveryError.reason} sessionKey=${params.sessionKey}`, + ); + return await finishCliAttempt( + await executeCliAttempt(retryableSessionId, { + timeoutMs: retryTimeoutMs, + forkCliSessionOnResume: true, + resumeAt: resumeCheckpointId, + onForkSuccessorPersisted: (sessionId) => { + retryableSessionId = sessionId; + }, + }), + ); + } catch (forkError) { + const deliveredForkFailure = await finishDeliveredFailure(forkError); + if (deliveredForkFailure) { + return deliveredForkFailure; + } + recoveryError = isUnsupportedCliResumeAtError( + forkError, + context.preparedBackend.backend.resumeAtArg, + ) + ? err + : forkError; + } + } + if ( + isFailoverError(recoveryError) && shouldRetryFreshCliSessionAfterFailover({ - error: err, + error: recoveryError, hasHistoryPrompt: Boolean(context.openClawHistoryPrompt), }) && retryableSessionId && @@ -1363,22 +1487,27 @@ export async function runPreparedCliAgent( try { const retryTimeoutMs = params.timeoutMs - (Date.now() - context.started); if (retryTimeoutMs <= 0) { - throw err; + throw recoveryError; } if (params.onBeforeFreshCliSessionRetry) { const clearedStaleBinding = await params.onBeforeFreshCliSessionRetry({ provider: params.provider, - reason: err.reason, + reason: recoveryError.reason, sessionId: retryableSessionId, }); if (!clearedStaleBinding) { - throw err; + throw recoveryError; } } cliBackendLog.warn( - `cli session recovery retry: provider=${params.provider} reason=${err.reason} sessionKey=${params.sessionKey}`, + `cli session recovery retry: provider=${params.provider} reason=${recoveryError.reason} sessionKey=${params.sessionKey}`, + ); + return await finishCliAttempt( + await executeCliAttempt(undefined, { + timeoutMs: retryTimeoutMs, + forkCliSessionOnResume: false, + }), ); - return await finishCliAttempt(await executeCliAttempt(undefined, retryTimeoutMs)); } catch (retryErr) { const deliveredRetryFailure = await finishDeliveredFailure(retryErr); if (deliveredRetryFailure) { @@ -1393,20 +1522,22 @@ export async function runPreparedCliAgent( return toCliRunFailure(retryErr); } } + } + if (isFailoverError(recoveryError)) { await runCliAgentEndHook(params, { - event: buildFailedAgentEndEvent(formatErrorMessage(err)), + event: buildFailedAgentEndEvent(formatErrorMessage(recoveryError)), ctx: hookContext, hookRunner, }); - throw err; + throw recoveryError; } - const message = formatErrorMessage(err); + const message = formatErrorMessage(recoveryError); await runCliAgentEndHook(params, { event: buildFailedAgentEndEvent(message), ctx: hookContext, hookRunner, }); - return toCliRunFailure(err); + return toCliRunFailure(recoveryError); } }; diff --git a/src/agents/cli-runner/claude-live-session.background-tasks.test.ts b/src/agents/cli-runner/claude-live-session.background-tasks.test.ts index 00bf6dbab985..22bda8143b2f 100644 --- a/src/agents/cli-runner/claude-live-session.background-tasks.test.ts +++ b/src/agents/cli-runner/claude-live-session.background-tasks.test.ts @@ -690,6 +690,98 @@ describe("claude live session provisional results", () => { expect(driver.cancel).not.toHaveBeenCalled(); }); + it.each([ + { + label: "marks a resumed synthetic-only stall as safe for cache-preserving recovery", + useResume: true, + expectedCode: "cli_no_output_timeout", + chunk: jsonl([ + { type: "system", subtype: "init", session_id: "live-synthetic-no-result" }, + { + type: "assistant", + session_id: "live-synthetic-no-result", + message: { + model: "", + role: "assistant", + content: [{ type: "text", text: "No response requested." }], + }, + }, + ]), + }, + { + label: "marks a resumed init-only stall as safe for recovery", + useResume: true, + expectedCode: "cli_no_output_timeout", + chunk: jsonl([{ type: "system", subtype: "init", session_id: "live-init-no-result" }]), + }, + { + label: "does not mark a fresh init-only stall as safe to replay", + useResume: false, + expectedCode: undefined, + chunk: jsonl([{ type: "system", subtype: "init", session_id: "live-fresh-init-no-result" }]), + }, + { + label: "does not mark a stall as retryable after substantive assistant output", + useResume: true, + expectedCode: undefined, + chunk: jsonl([ + { type: "system", subtype: "init", session_id: "live-synthetic-substantive" }, + { + type: "assistant", + session_id: "live-synthetic-substantive", + message: { + model: "", + role: "assistant", + content: [{ type: "text", text: "No response requested." }], + }, + }, + { + type: "assistant", + session_id: "live-synthetic-substantive", + message: { + model: "claude-fable-5", + role: "assistant", + content: [{ type: "text", text: "Partial real answer" }], + }, + }, + ]), + }, + { + label: "does not mark an incomplete stdout record as safe to replay", + useResume: true, + expectedCode: undefined, + chunk: '{"type":"assistant","message":{"model":"claude-fable-5"', + }, + ])("$label", async ({ useResume, expectedCode, chunk }) => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + const driver = installLiveStdoutDriver(); + const resultPromise = startLiveTurn({ + runId: `run-replay-safe-stall-${useResume ? "resume" : "fresh"}`, + timeoutMs: 60_000, + noOutputTimeoutMs: 1_000, + useResume, + }); + await vi.advanceTimersByTimeAsync(0); + await driver.stdout.waitReady(); + driver.stdout.emit(chunk); + + const errorPromise = resultPromise.catch((error: unknown) => error); + await vi.advanceTimersByTimeAsync(1_000); + const error = (await errorPromise) as { code?: string; cliTimeout?: unknown }; + expect(error).toMatchObject({ + name: "FailoverError", + cliTimeout: { + mode: "no-output", + timeoutSeconds: 1, + observedActivity: true, + activeToolCount: 0, + backgroundTaskCount: 0, + }, + }); + expect(error.code).toBe(expectedCode); + expect(driver.cancel).toHaveBeenCalledWith("manual-cancel"); + }); + it("still aborts on the turn timeout while waiting after a synthetic placeholder", async () => { vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); const driver = installLiveStdoutDriver(); diff --git a/src/agents/cli-runner/claude-live-session.ts b/src/agents/cli-runner/claude-live-session.ts index bbabaee5a5f0..bcc5ff2fb753 100644 --- a/src/agents/cli-runner/claude-live-session.ts +++ b/src/agents/cli-runner/claude-live-session.ts @@ -83,6 +83,10 @@ type ClaudeLiveTurn = { timeoutTimer: NodeJS.Timeout | null; activeTools: Map; observedStdout: boolean; + /** Only resumed turns may replay a lifecycle-only stall through a fork. */ + useResume: boolean; + /** True after any output other than init or the exact synthetic queue placeholder. */ + hasReplayUnsafeActivity: boolean; /** * Claude consumed queued session notifications before processing this turn. * The following empty result is provisional; the same process can emit the @@ -820,14 +824,23 @@ function armNoOutputTimer(session: ClaudeLiveSession, turn: ClaudeLiveTurn, dela return; } } + const retryableResumeStall = + turn.useResume && + session.stdoutBuffer.trim().length === 0 && + !turn.hasReplayUnsafeActivity && + turn.toolEventCount === 0 && + turn.activeTools.size === 0 && + session.outstandingBackgroundTaskIds.size === 0; closeLiveSession( session, "abort", createTimeoutError( session, `CLI produced no output for ${Math.round((Date.now() - quietSinceMs) / 1000)}s and was terminated.`, - // Retryable only when the process never produced any output this turn. - turn.lastOutputAtMs === null ? "cli_no_output_timeout" : undefined, + // Claude can emit only init or a synthetic queue placeholder before a + // resumed stream wedges. No assistant/tool work has happened, so + // recovery can fork the cached session before transcript reseed. + turn.lastOutputAtMs === null || retryableResumeStall ? "cli_no_output_timeout" : undefined, { mode: "no-output", timeoutSeconds: Math.round((Date.now() - quietSinceMs) / 1000), @@ -1208,6 +1221,9 @@ function handleClaudeLiveLine(session: ClaudeLiveSession, line: string): void { turn.observedStdout = true; } if (!parsed) { + if (turn) { + turn.hasReplayUnsafeActivity = true; + } return; } const parsedSessionId = parseSessionId(parsed); @@ -1217,6 +1233,14 @@ function handleClaudeLiveLine(session: ClaudeLiveSession, line: string): void { if (!turn) { return; } + if ( + !( + (parsed.type === "system" && parsed.subtype === "init") || + isClaudeLiveProvisionalSyntheticPlaceholder(parsed) + ) + ) { + turn.hasReplayUnsafeActivity = true; + } noteClaudeLiveContinuationAfterSyntheticPlaceholder(session, turn); turn.rawChars += trimmed.length + 1; if ( @@ -1448,6 +1472,9 @@ async function createClaudeLiveSession(params: { onStderr: (chunk) => { if (session) { session.currentTurn?.onCliOutput?.(chunk, "stderr"); + if (session.currentTurn && chunk.trim()) { + session.currentTurn.hasReplayUnsafeActivity = true; + } session.stderr += chunk; if (session.stderr.length > LIVE_SESSION_LIMITS.maxStderrChars) { closeLiveSession( @@ -1506,6 +1533,7 @@ function createTurn(params: { context: PreparedCliRunContext; noOutputTimeoutMs: number; allowSyntheticContinuationGrace: boolean; + useResume: boolean; onAssistantDelta: (delta: CliStreamingDelta) => void; onThinkingDelta?: (delta: CliThinkingDelta) => void; onThinkingProgress?: (progress: CliThinkingProgress) => void; @@ -1543,6 +1571,8 @@ function createTurn(params: { timeoutTimer: null, activeTools: new Map(), observedStdout: false, + useResume: params.useResume, + hasReplayUnsafeActivity: false, pendingSyntheticPlaceholder: false, allowSyntheticContinuationGrace: params.allowSyntheticContinuationGrace, deferredSyntheticOutput: null, @@ -1886,6 +1916,7 @@ export async function runClaudeLiveSessionTurn(params: { context: params.context, noOutputTimeoutMs: params.noOutputTimeoutMs, allowSyntheticContinuationGrace: params.useResume && createdSessionForTurn, + useResume: params.useResume, onAssistantDelta: params.onAssistantDelta, onThinkingDelta: params.onThinkingDelta, onThinkingProgress: params.onThinkingProgress, diff --git a/src/agents/cli-runner/execute.ts b/src/agents/cli-runner/execute.ts index 0a4d9c981c57..c56ffda356c0 100644 --- a/src/agents/cli-runner/execute.ts +++ b/src/agents/cli-runner/execute.ts @@ -210,6 +210,7 @@ export async function executePreparedCliRun( promptArg: argsPrompt, useResume, forkResume: params.forkCliSessionOnResume, + resumeAt: params.cliSessionResumeAt, sendSystemPromptOnResume: resendSystemPromptForSoftResume, }); diff --git a/src/agents/cli-runner/helpers.system-prompt-resume.test.ts b/src/agents/cli-runner/helpers.system-prompt-resume.test.ts index 98107c9b6f97..f10ec719ee0f 100644 --- a/src/agents/cli-runner/helpers.system-prompt-resume.test.ts +++ b/src/agents/cli-runner/helpers.system-prompt-resume.test.ts @@ -185,7 +185,11 @@ describe("buildCliArgs — issue #80374", () => { }); it("appends a configured fork argument only to the marked resume", () => { - const backend = { ...BACKEND_ALWAYS, forkArg: "--fork-session" } as CliBackendConfig; + const backend = { + ...BACKEND_ALWAYS, + forkArg: "--fork-session", + resumeAtArg: "--resume-session-at", + } as CliBackendConfig; const resumed = buildCliArgs({ backend, baseArgs: ["--resume", "source-session"], @@ -193,6 +197,7 @@ describe("buildCliArgs — issue #80374", () => { sessionId: "source-session", useResume: true, forkResume: true, + resumeAt: "assistant-before-turn", }); const subsequent = buildCliArgs({ backend, @@ -203,7 +208,11 @@ describe("buildCliArgs — issue #80374", () => { forkResume: false, }); expect(resumed).toContain("--fork-session"); + expect(resumed).toEqual( + expect.arrayContaining(["--resume-session-at", "assistant-before-turn"]), + ); expect(subsequent).not.toContain("--fork-session"); + expect(subsequent).not.toContain("--resume-session-at"); }); it("rejects a marked fork when the backend has no fork argument", () => { @@ -218,6 +227,20 @@ describe("buildCliArgs — issue #80374", () => { }), ).toThrow("does not support forked session resume"); }); + + it("rejects a checkpoint when the backend has no resume-at argument", () => { + expect(() => + buildCliArgs({ + backend: { ...BACKEND_ALWAYS, forkArg: "--fork-session" } as CliBackendConfig, + baseArgs: ["--resume", "source-session"], + modelId: "claude-haiku-4-5", + sessionId: "source-session", + useResume: true, + forkResume: true, + resumeAt: "assistant-before-turn", + }), + ).toThrow("does not support checkpointed session resume"); + }); }); // ─── buildClaudeLiveArgs (Path 4: live-stdio strip guard) ─────────────────── diff --git a/src/agents/cli-runner/helpers.ts b/src/agents/cli-runner/helpers.ts index 2f911836bfd9..2198f40ff0ec 100644 --- a/src/agents/cli-runner/helpers.ts +++ b/src/agents/cli-runner/helpers.ts @@ -487,6 +487,7 @@ export function buildCliArgs(params: { promptArg?: string; useResume: boolean; forkResume?: boolean; + resumeAt?: string; sendSystemPromptOnResume?: boolean; }): string[] { const args: string[] = [...params.baseArgs]; @@ -533,6 +534,12 @@ export function buildCliArgs(params: { } args.push(params.backend.forkArg); } + if (params.resumeAt) { + if (!params.useResume || !params.backend.resumeAtArg) { + throw new Error("CLI backend does not support checkpointed session resume"); + } + args.push(params.backend.resumeAtArg, params.resumeAt); + } if (params.promptArg !== undefined) { let replacedPromptPlaceholder = false; for (let i = 0; i < args.length; i += 1) { diff --git a/src/agents/cli-runner/types.ts b/src/agents/cli-runner/types.ts index 02371acf291a..9d2dbcc565dc 100644 --- a/src/agents/cli-runner/types.ts +++ b/src/agents/cli-runner/types.ts @@ -41,6 +41,12 @@ import type { FastModeAutoProgressState } from "../fast-mode.js"; import type { ScheduledToolPolicyContext } from "../scheduled-tool-policy.js"; import type { SilentReplyPromptMode } from "../system-prompt.types.js"; +type CliSessionRetryParams = { + provider: string; + reason: FailoverReason; + sessionId: string; +}; + /** Input contract for one CLI-backed agent run. */ export type RunCliAgentParams = { sessionId: string; @@ -121,12 +127,16 @@ export type RunCliAgentParams = { cliSessionBinding?: CliSessionBinding; /** Consume the backend fork argument on this resume invocation only. */ forkCliSessionOnResume?: boolean; + /** Bound a resumed fork at this previously observed assistant checkpoint. */ + cliSessionResumeAt?: string; /** Atomically claim the persisted one-shot marker after the CLI queue admits this turn. */ claimCliSessionFork?: () => Promise; /** Re-arm a claimed marker when the CLI turn fails before producing a successor session. */ restoreCliSessionFork?: () => Promise; /** Persist the successor ID as soon as the CLI reports the forked session. */ persistCliSessionForkSuccessor?: (sessionId: string) => Promise; + /** Atomically arm a cache-preserving fork before retrying a stalled resumed session. */ + onBeforeForkedCliSessionRetry?: (params: CliSessionRetryParams) => boolean | Promise; authProfileId?: string; /** Private seam: report the credential/runtime owner only after a successful real turn. */ onSuccessfulAuthBinding?: (binding: { @@ -139,11 +149,7 @@ export type RunCliAgentParams = { runtimeArtifactId?: string; skipLocalCredential?: true; }) => void; - onBeforeFreshCliSessionRetry?: (params: { - provider: string; - reason: FailoverReason; - sessionId: string; - }) => boolean | Promise; + onBeforeFreshCliSessionRetry?: (params: CliSessionRetryParams) => boolean | Promise; bootstrapPromptWarningSignaturesSeen?: string[]; bootstrapPromptWarningSignature?: string; bootstrapContextMode?: BootstrapContextMode; diff --git a/src/agents/cli-session.ts b/src/agents/cli-session.ts index 926a122115a4..151491d4bff5 100644 --- a/src/agents/cli-session.ts +++ b/src/agents/cli-session.ts @@ -47,6 +47,9 @@ export function setCliSessionBinding( ...entry.cliSessionBindings, [normalized]: { sessionId: trimmed, + ...(normalizeOptionalString(binding.resumeCheckpointId) + ? { resumeCheckpointId: normalizeOptionalString(binding.resumeCheckpointId) } + : {}), ...(binding.forceReuse === true ? { forceReuse: true } : {}), ...(binding.forkNextResume === true ? { forkNextResume: true } : {}), ...(normalizeOptionalString(binding.authProfileId) diff --git a/src/agents/command/attempt-execution.cli.test.ts b/src/agents/command/attempt-execution.cli.test.ts index 133c886d55dc..dd8e0582ae51 100644 --- a/src/agents/command/attempt-execution.cli.test.ts +++ b/src/agents/command/attempt-execution.cli.test.ts @@ -580,18 +580,24 @@ describe("CLI attempt execution", () => { expect(persisted[sessionKey]?.cliSessionBindings?.["claude-cli"]?.sessionId).toBe(cliSessionId); }); - it("clears reused Claude CLI session IDs before a fresh retry after timeout failover", async () => { + it("atomically forks and rebinds a reused Claude CLI session after timeout failover", async () => { const sessionKey = "agent:main:direct:cli-timeout"; const cliSessionId = "timeout-poisoned-session"; + const forkedCliSessionId = "timeout-recovery-fork"; await writeClaudeCliAssistantTranscript(cliSessionId); const sessionEntry = makeClaudeCliSessionEntry("session-cli-timeout", cliSessionId); const sessionStore: Record = { [sessionKey]: sessionEntry }; await writeSessionStoreSeed(sessionStore); runCliAgentMock.mockImplementationOnce(async (args: unknown) => { - const retry = requireRecord(args, "run CLI agent argument").onBeforeFreshCliSessionRetry; - expect(retry).toBeTypeOf("function"); + const runArgs = requireRecord(args, "run CLI agent argument"); + const prepareFork = runArgs.onBeforeForkedCliSessionRetry; + const claimFork = runArgs.claimCliSessionFork; + const persistFork = runArgs.persistCliSessionForkSuccessor; + expect(prepareFork).toBeTypeOf("function"); + expect(claimFork).toBeTypeOf("function"); + expect(persistFork).toBeTypeOf("function"); await ( - retry as (params: { + prepareFork as (params: { provider: string; reason: "timeout"; sessionId: string; @@ -601,9 +607,17 @@ describe("CLI attempt execution", () => { reason: "timeout", sessionId: cliSessionId, }); - expect(sessionStore[sessionKey]?.cliSessionBindings?.["claude-cli"]).toBeUndefined(); - expect(sessionStore[sessionKey]?.cliSessionIds?.["claude-cli"]).toBeUndefined(); - expect(sessionStore[sessionKey]?.claudeCliSessionId).toBeUndefined(); + expect(sessionStore[sessionKey]?.cliSessionBindings?.["claude-cli"]?.forkNextResume).toBe( + true, + ); + await (claimFork as () => Promise)(); + expect( + sessionStore[sessionKey]?.cliSessionBindings?.["claude-cli"]?.forkNextResume, + ).toBeUndefined(); + await (persistFork as (sessionId: string) => Promise)(forkedCliSessionId); + expect(sessionStore[sessionKey]?.cliSessionBindings?.["claude-cli"]?.sessionId).toBe( + forkedCliSessionId, + ); return makeCliResult("hello after timeout"); }); @@ -617,9 +631,203 @@ describe("CLI attempt execution", () => { expect(runCliAgentMock).toHaveBeenCalledTimes(1); expect(firstRunCliAgentArg().cliSessionId).toBe(cliSessionId); + expect(sessionStore[sessionKey]?.cliSessionBindings?.["claude-cli"]?.sessionId).toBe( + forkedCliSessionId, + ); + expect(sessionStore[sessionKey]?.cliSessionIds?.["claude-cli"]).toBe(forkedCliSessionId); + expect(sessionStore[sessionKey]?.claudeCliSessionId).toBe(forkedCliSessionId); + + const persisted = readSessionStore(); + expect(persisted[sessionKey]?.cliSessionBindings?.["claude-cli"]?.sessionId).toBe( + forkedCliSessionId, + ); + }); + + it("clears a persisted fork successor before transcript fallback", async () => { + const sessionKey = "agent:main:direct:cli-fork-timeout"; + const cliSessionId = "timeout-parent-session"; + const forkedCliSessionId = "timeout-stalled-fork"; + await writeClaudeCliAssistantTranscript(cliSessionId); + const sessionEntry = makeClaudeCliSessionEntry("session-cli-fork-timeout", cliSessionId); + sessionEntry.cliSessionBindings!["claude-cli"]!.forkNextResume = true; + const sessionStore: Record = { [sessionKey]: sessionEntry }; + await writeSessionStoreSeed(sessionStore); + runCliAgentMock.mockImplementationOnce(async (args: unknown) => { + const runArgs = requireRecord(args, "run CLI agent argument"); + const claimFork = runArgs.claimCliSessionFork; + const persistFork = runArgs.persistCliSessionForkSuccessor; + const clearFork = runArgs.onBeforeFreshCliSessionRetry; + expect(runArgs.forkCliSessionOnResume).toBe(true); + expect(runArgs.onBeforeForkedCliSessionRetry).toBeUndefined(); + expect(clearFork).toBeTypeOf("function"); + await (claimFork as () => Promise)(); + await (persistFork as (sessionId: string) => Promise)(forkedCliSessionId); + await ( + clearFork as (params: { + provider: string; + reason: "timeout"; + sessionId: string; + }) => Promise + )({ + provider: "claude-cli", + reason: "timeout", + sessionId: forkedCliSessionId, + }); + return makeCliResult("hello after fork timeout"); + }); + + await runClaudeCliAttempt({ + sessionKey, + sessionEntry, + sessionStore, + body: "retry after fork timeout", + runId: "run-cli-fork-timeout", + }); + expect(sessionStore[sessionKey]?.cliSessionBindings?.["claude-cli"]).toBeUndefined(); - expect(sessionStore[sessionKey]?.cliSessionIds?.["claude-cli"]).toBeUndefined(); - expect(sessionStore[sessionKey]?.claudeCliSessionId).toBeUndefined(); + const persisted = readSessionStore(); + expect(persisted[sessionKey]?.cliSessionBindings?.["claude-cli"]).toBeUndefined(); + }); + + it("clears a persisted fork successor when recovery fails after rebinding", async () => { + const sessionKey = "agent:main:direct:cli-fork-finalization-failure"; + const cliSessionId = "finalization-parent-session"; + const forkedCliSessionId = "partial-fork-successor"; + await writeClaudeCliAssistantTranscript(cliSessionId); + const sessionEntry = makeClaudeCliSessionEntry( + "session-cli-fork-finalization-failure", + cliSessionId, + ); + sessionEntry.cliSessionBindings!["claude-cli"]!.forkNextResume = true; + const sessionStore: Record = { [sessionKey]: sessionEntry }; + await writeSessionStoreSeed(sessionStore); + const finalizationError = Object.assign(new Error("fork finalization failed"), { + name: "AbortError", + }); + runCliAgentMock.mockImplementationOnce(async (args: unknown) => { + const runArgs = requireRecord(args, "run CLI agent argument"); + await (runArgs.claimCliSessionFork as () => Promise)(); + await (runArgs.persistCliSessionForkSuccessor as (sessionId: string) => Promise)( + forkedCliSessionId, + ); + throw finalizationError; + }); + + await expect( + runClaudeCliAttempt({ + sessionKey, + sessionEntry, + sessionStore, + body: "resume and fail after fork", + runId: "run-cli-fork-finalization-failure", + }), + ).rejects.toBe(finalizationError); + + expect(sessionStore[sessionKey]?.cliSessionBindings?.["claude-cli"]).toBeUndefined(); + expect(readSessionStore()[sessionKey]?.cliSessionBindings?.["claude-cli"]).toBeUndefined(); + }); + + it("preserves a restored fork marker when recovery dies before producing a successor", async () => { + const sessionKey = "agent:main:direct:cli-fork-before-successor-failure"; + const cliSessionId = "recovery-source-session"; + await writeClaudeCliAssistantTranscript(cliSessionId); + const sessionEntry = makeClaudeCliSessionEntry( + "session-cli-fork-before-successor-failure", + cliSessionId, + ); + const sessionStore: Record = { [sessionKey]: sessionEntry }; + await writeSessionStoreSeed(sessionStore); + const recoveryError = Object.assign(new Error("fork process died before init"), { + name: "AbortError", + }); + runCliAgentMock.mockImplementationOnce(async (args: unknown) => { + const runArgs = requireRecord(args, "run CLI agent argument"); + await ( + runArgs.onBeforeForkedCliSessionRetry as (params: { + provider: string; + reason: "timeout"; + sessionId: string; + }) => Promise + )({ provider: "claude-cli", reason: "timeout", sessionId: cliSessionId }); + await (runArgs.claimCliSessionFork as () => Promise)(); + await (runArgs.restoreCliSessionFork as () => Promise)(); + throw recoveryError; + }); + + await expect( + runClaudeCliAttempt({ + sessionKey, + sessionEntry, + sessionStore, + body: "resume and fail before fork init", + runId: "run-cli-fork-before-successor-failure", + }), + ).rejects.toBe(recoveryError); + + expect(sessionStore[sessionKey]?.cliSessionBindings?.["claude-cli"]).toMatchObject({ + sessionId: cliSessionId, + forkNextResume: true, + }); + expect(readSessionStore()[sessionKey]?.cliSessionBindings?.["claude-cli"]).toMatchObject({ + sessionId: cliSessionId, + forkNextResume: true, + }); + }); + + it("does not clear a concurrent rebind after failed fork recovery", async () => { + const sessionKey = "agent:main:direct:cli-fork-concurrent-rebind"; + const cliSessionId = "concurrent-parent-session"; + const forkedCliSessionId = "failed-fork-successor"; + const concurrentCliSessionId = "newer-concurrent-session"; + await writeClaudeCliAssistantTranscript(cliSessionId); + const sessionEntry = makeClaudeCliSessionEntry( + "session-cli-fork-concurrent-rebind", + cliSessionId, + ); + const sessionStore: Record = { [sessionKey]: sessionEntry }; + await writeSessionStoreSeed(sessionStore); + const recoveryError = Object.assign(new Error("fork recovery aborted"), { + name: "AbortError", + }); + runCliAgentMock.mockImplementationOnce(async (args: unknown) => { + const runArgs = requireRecord(args, "run CLI agent argument"); + await ( + runArgs.onBeforeForkedCliSessionRetry as (params: { + provider: string; + reason: "timeout"; + sessionId: string; + }) => Promise + )({ provider: "claude-cli", reason: "timeout", sessionId: cliSessionId }); + await (runArgs.claimCliSessionFork as () => Promise)(); + await (runArgs.persistCliSessionForkSuccessor as (sessionId: string) => Promise)( + forkedCliSessionId, + ); + + const concurrentEntry = makeClaudeCliSessionEntry( + sessionEntry.sessionId, + concurrentCliSessionId, + ); + await replaceSessionEntry({ sessionKey, storePath }, concurrentEntry); + sessionStore[sessionKey] = concurrentEntry; + throw recoveryError; + }); + + await expect( + runClaudeCliAttempt({ + sessionKey, + sessionEntry, + sessionStore, + body: "resume while another turn rebinds", + runId: "run-cli-fork-concurrent-rebind", + }), + ).rejects.toBe(recoveryError); + + expect(sessionStore[sessionKey]?.cliSessionBindings?.["claude-cli"]?.sessionId).toBe( + concurrentCliSessionId, + ); + expect(readSessionStore()[sessionKey]?.cliSessionBindings?.["claude-cli"]?.sessionId).toBe( + concurrentCliSessionId, + ); }); it("does not install a stale-session clearing hook for storeless CLI attempts", async () => { diff --git a/src/agents/command/attempt-execution.ts b/src/agents/command/attempt-execution.ts index 5ea56ad377ce..1139eb85596f 100644 --- a/src/agents/command/attempt-execution.ts +++ b/src/agents/command/attempt-execution.ts @@ -714,16 +714,15 @@ export function runAgentAttempt(params: { activeCliSessionBinding = cliSessionBinding, ) => { const forkCliSessionOnResume = activeCliSessionBinding?.forkNextResume === true; - if ( - forkCliSessionOnResume && - !resolveCliBackendConfig(cliExecutionProvider, params.cfg, { - agentId: params.sessionAgentId, - })?.config.forkArg - ) { + const resolvedCliBackend = resolveCliBackendConfig(cliExecutionProvider, params.cfg, { + agentId: params.sessionAgentId, + }); + const supportsCliSessionFork = Boolean(resolvedCliBackend?.config.forkArg); + if (forkCliSessionOnResume && !supportsCliSessionFork) { throw new Error(`CLI backend "${cliExecutionProvider}" does not support session forks`); } const forkStoreParams = - forkCliSessionOnResume && nextCliSessionId && mutableCliSessionStore + supportsCliSessionFork && nextCliSessionId && mutableCliSessionStore ? { provider: cliExecutionProvider, expectedCliSessionId: nextCliSessionId, @@ -839,9 +838,9 @@ export function runAgentAttempt(params: { suppressNextUserMessagePersistence: params.suppressPromptPersistenceOnRetry === true, disableTools, allowEmptyAssistantReplyAsSilent: isSubagentAnnounceHandoff, - ...(mutableCliSessionStore && !forkCliSessionOnResume + ...(forkStoreParams && !forkCliSessionOnResume ? { - onBeforeFreshCliSessionRetry: async (retry) => { + onBeforeForkedCliSessionRetry: async (retry) => { if ( hasNewGeneratedMediaTaskForSessionKey( params.sessionKey, @@ -852,15 +851,40 @@ export function runAgentAttempt(params: { return false; } + log.warn( + `CLI session stalled, arming forked recovery: provider=${sanitizeForLog(cliExecutionProvider)} sessionKey=${forkStoreParams.sessionKey}`, + ); + + const armed = await restoreCliSessionForkInStore(forkStoreParams); + if (armed) { + params.sessionEntry = armed; + } + return Boolean(armed); + }, + } + : {}), + ...(mutableCliSessionStore + ? { + onBeforeFreshCliSessionRetry: async (retry) => { + if ( + hasNewGeneratedMediaTaskForSessionKey(params.sessionKey, mediaTaskIdsBefore) + ) { + return false; + } + log.warn( `CLI session failed, clearing before fresh retry: provider=${sanitizeForLog(cliExecutionProvider)} sessionKey=${mutableCliSessionStore.sessionKey} reason=${sanitizeForLog(retry.reason)}`, ); - params.sessionEntry = - (await clearCliSessionInStore({ - provider: cliExecutionProvider, - ...mutableCliSessionStore, - })) ?? params.sessionEntry; + const cleared = await clearCliSessionInStore({ + provider: cliExecutionProvider, + expectedCliSessionId: retry.sessionId, + ...mutableCliSessionStore, + }); + if (!cleared) { + return false; + } + params.sessionEntry = cleared; return true; }, } @@ -872,12 +896,17 @@ export function runAgentAttempt(params: { try { return await runCliWithSession(activeCliSessionBinding?.sessionId, activeCliSessionBinding); } catch (err) { + const failedCliSessionBinding = getCliSessionBinding( + params.sessionEntry, + cliExecutionProvider, + ); + const failedCliSessionId = failedCliSessionBinding?.sessionId; if ( isClaudeCliProvider(cliExecutionProvider) && - !activeCliSessionBinding?.forkNextResume && + failedCliSessionBinding?.forkNextResume !== true && shouldClearReusedCliSessionAfterError(err) && !hasNewGeneratedMediaTaskForSessionKey(params.sessionKey, mediaTaskIdsBefore) && - activeCliSessionBinding?.sessionId && + failedCliSessionId && mutableCliSessionStore ) { log.warn( @@ -887,6 +916,7 @@ export function runAgentAttempt(params: { params.sessionEntry = (await clearCliSessionInStore({ provider: cliExecutionProvider, + expectedCliSessionId: failedCliSessionId, ...mutableCliSessionStore, })) ?? params.sessionEntry; } diff --git a/src/agents/command/session-store.test.ts b/src/agents/command/session-store.test.ts index d762eaa189b2..0ceeb675b65c 100644 --- a/src/agents/command/session-store.test.ts +++ b/src/agents/command/session-store.test.ts @@ -2910,6 +2910,7 @@ describe("consumeCliSessionForkInStore", () => { cliSessionBindings: { "claude-cli": { sessionId: "claude-source-session", + resumeCheckpointId: "assistant-before-turn", forceReuse: true, forkNextResume: true, }, @@ -2930,12 +2931,17 @@ describe("consumeCliSessionForkInStore", () => { }); expect(consumed?.cliSessionBindings?.["claude-cli"]).toEqual({ sessionId: "claude-source-session", + resumeCheckpointId: "assistant-before-turn", forceReuse: true, }); expect(consumed?.label).toBe("concurrent update"); expect( loadPersistedSessionEntry(storePath, sessionKey)?.cliSessionBindings?.["claude-cli"], - ).toEqual({ sessionId: "claude-source-session", forceReuse: true }); + ).toEqual({ + sessionId: "claude-source-session", + resumeCheckpointId: "assistant-before-turn", + forceReuse: true, + }); await expect( consumeCliSessionForkInStore({ provider: "claude-cli", @@ -2986,6 +2992,7 @@ describe("consumeCliSessionForkInStore", () => { cliSessionBindings: { "claude-cli": { sessionId: "claude-source-session", + resumeCheckpointId: "assistant-before-turn", forceReuse: true, authProfileId: "claude:work", authEpoch: "epoch-1", @@ -3007,6 +3014,7 @@ describe("consumeCliSessionForkInStore", () => { expect(persisted?.cliSessionBindings?.["claude-cli"]).toEqual({ sessionId: "claude-fork-session", + resumeCheckpointId: "assistant-before-turn", forceReuse: true, authProfileId: "claude:work", authEpoch: "epoch-1", @@ -3016,6 +3024,7 @@ describe("consumeCliSessionForkInStore", () => { loadPersistedSessionEntry(storePath, sessionKey)?.cliSessionBindings?.["claude-cli"], ).toEqual({ sessionId: "claude-fork-session", + resumeCheckpointId: "assistant-before-turn", forceReuse: true, authProfileId: "claude:work", authEpoch: "epoch-1", diff --git a/src/agents/command/session-store.ts b/src/agents/command/session-store.ts index 9a0c403a4801..bb6dd324df15 100644 --- a/src/agents/command/session-store.ts +++ b/src/agents/command/session-store.ts @@ -14,7 +14,12 @@ import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { createLazyImportLoader } from "../../shared/lazy-promise.js"; import { resolveNonNegativeNumber } from "../../shared/number-coercion.js"; import { resolveDefaultAgentId } from "../agent-scope.js"; -import { clearCliSession, setCliSessionBinding, setCliSessionId } from "../cli-session.js"; +import { + clearCliSession, + getCliSessionBinding, + setCliSessionBinding, + setCliSessionId, +} from "../cli-session.js"; import { DEFAULT_CONTEXT_TOKENS } from "../defaults.js"; import { clearMainSessionRecoveryAfterAgentRun } from "../main-session-recovery-clear.js"; import { isCliProvider } from "../model-selection.js"; @@ -330,17 +335,15 @@ export async function clearCliSessionInStore(params: { sessionStore: Record; storePath: string; expectedSessionId?: string; + expectedCliSessionId?: string; }): Promise { - const { provider, sessionKey, sessionStore, storePath, expectedSessionId } = params; + const { provider, sessionKey, sessionStore, storePath, expectedSessionId, expectedCliSessionId } = + params; const entry = sessionStore[sessionKey]; if (!entry) { return undefined; } - const next = { ...entry }; - clearCliSession(next, provider); - next.updatedAt = Date.now(); - const persisted = await patchSessionEntry( { storePath, @@ -353,6 +356,15 @@ export async function clearCliSessionInStore(params: { ) { return null; } + if ( + expectedCliSessionId && + getCliSessionBinding(currentEntry, provider)?.sessionId !== expectedCliSessionId + ) { + return null; + } + const next = { ...currentEntry }; + clearCliSession(next, provider); + next.updatedAt = Date.now(); return next; }, { fallbackEntry: entry }, @@ -400,7 +412,7 @@ export async function consumeCliSessionForkInStore(params: { return persisted ?? undefined; } -/** Re-arms a claimed fork marker after a failed CLI turn. */ +/** Arms a fork marker for recovery, or re-arms one after a failed CLI turn. */ export async function restoreCliSessionForkInStore(params: { provider: string; sessionKey: string; diff --git a/src/config/sessions/cli-session-binding.ts b/src/config/sessions/cli-session-binding.ts index e65b4414d127..97d7c9957b47 100644 --- a/src/config/sessions/cli-session-binding.ts +++ b/src/config/sessions/cli-session-binding.ts @@ -75,6 +75,7 @@ export function getCliSessionBinding( if (bindingSessionId) { return { sessionId: bindingSessionId, + resumeCheckpointId: normalizeOptionalString(fromBindings?.resumeCheckpointId), ...(fromBindings?.forceReuse === true ? { forceReuse: true } : {}), ...(fromBindings?.forkNextResume === true ? { forkNextResume: true } : {}), authProfileId: normalizeOptionalString(fromBindings?.authProfileId), diff --git a/src/config/sessions/types.ts b/src/config/sessions/types.ts index 9731b92653bd..6d69f6000257 100644 --- a/src/config/sessions/types.ts +++ b/src/config/sessions/types.ts @@ -63,6 +63,8 @@ export type CliSessionReseedReceipt = { export type CliSessionBinding = { sessionId: string; + /** Last successful assistant boundary accepted by the backend's resume contract. */ + resumeCheckpointId?: string; /** Resume with the backend's fork argument once, then clear before process start. */ forkNextResume?: true; /** Trust an explicitly attached CLI session even when auth, prompt, or MCP fingerprints drift. */ diff --git a/src/plugins/cli-backend.types.ts b/src/plugins/cli-backend.types.ts index 18d8ce943e3a..52ae42ae3c9f 100644 --- a/src/plugins/cli-backend.types.ts +++ b/src/plugins/cli-backend.types.ts @@ -34,6 +34,8 @@ export type CliBackendConfig = { resumeArgs?: string[]; /** Argument appended to one explicitly forked resume invocation. */ forkArg?: string; + /** Argument followed by an assistant checkpoint id to bound one resumed fork. */ + resumeAtArg?: string; /** When to pass session ids. */ sessionMode?: "always" | "existing" | "none"; /** JSON fields to read session id from (in order). */