ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-27 21:07:01 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix: preserve Claude cache during stalled CLI recovery (#113866)

Browse Source 
* 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 <steipete@gmail.com>
This commit is contained in:
Vito Cappello
2026-07-26 23:11:39 -04:00
committed by GitHub
parent e71e4ac83e
commit 5ae8a4f4f9
19 changed files with 1325 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/anthropic/cli-backend.ts
+3
View File
@@ -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",
src/agents/cli-output.test.ts
+49
View File
@@ -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(
[
src/agents/cli-output.ts
+48 -5
View File
@@ -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, unknown>;
}): 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<string, unknown>;
@@ -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 };
src/agents/cli-runner.reliability.test.ts
+611 -9
View File
@@ -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<RunExit>((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: "<synthetic>",
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<void>((resolve) => {
notifyFirstSpawn = resolve;
});
let notifySecondSpawn: (() => void) | undefined;
const secondSpawned = new Promise<void>((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<RunExit>((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: "<synthetic>",
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<void>((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<RunExit>((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();
src/agents/cli-runner.ts
+146 -15
View File
@@ -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);
}
};
src/agents/cli-runner/claude-live-session.background-tasks.test.ts
+92
View File
@@ -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: "<synthetic>",
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: "<synthetic>",
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();
src/agents/cli-runner/claude-live-session.ts
+33 -2
View File
@@ -83,6 +83,10 @@ type ClaudeLiveTurn = {
timeoutTimer: NodeJS.Timeout | null;
activeTools: Map<string, ClaudeLiveActiveTool>;
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,
src/agents/cli-runner/execute.ts
+1
View File
@@ -210,6 +210,7 @@ export async function executePreparedCliRun(
promptArg: argsPrompt,
useResume,
forkResume: params.forkCliSessionOnResume,
resumeAt: params.cliSessionResumeAt,
sendSystemPromptOnResume: resendSystemPromptForSoftResume,
});
src/agents/cli-runner/helpers.system-prompt-resume.test.ts
+24 -1
View File
@@ -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) ───────────────────
src/agents/cli-runner/helpers.ts
+7
View File
@@ -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) {
src/agents/cli-runner/types.ts
+11 -5
View File
@@ -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<boolean>;
/** Re-arm a claimed marker when the CLI turn fails before producing a successor session. */
restoreCliSessionFork?: () => Promise<void>;
/** Persist the successor ID as soon as the CLI reports the forked session. */
persistCliSessionForkSuccessor?: (sessionId: string) => Promise<void>;
/** Atomically arm a cache-preserving fork before retrying a stalled resumed session. */
onBeforeForkedCliSessionRetry?: (params: CliSessionRetryParams) => boolean | Promise<boolean>;
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<boolean>;
onBeforeFreshCliSessionRetry?: (params: CliSessionRetryParams) => boolean | Promise<boolean>;
bootstrapPromptWarningSignaturesSeen?: string[];
bootstrapPromptWarningSignature?: string;
bootstrapContextMode?: BootstrapContextMode;
src/agents/cli-session.ts
+3
View File
@@ -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)
src/agents/command/attempt-execution.cli.test.ts
+217 -9
View File
@@ -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<string, SessionEntry> = { [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<boolean>)();
expect(
sessionStore[sessionKey]?.cliSessionBindings?.["claude-cli"]?.forkNextResume,
).toBeUndefined();
await (persistFork as (sessionId: string) => Promise<void>)(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<string, SessionEntry> = { [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<boolean>)();
await (persistFork as (sessionId: string) => Promise<void>)(forkedCliSessionId);
await (
clearFork as (params: {
provider: string;
reason: "timeout";
sessionId: string;
}) => Promise<boolean>
)({
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<string, SessionEntry> = { [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<boolean>)();
await (runArgs.persistCliSessionForkSuccessor as (sessionId: string) => Promise<void>)(
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<string, SessionEntry> = { [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<boolean>
)({ provider: "claude-cli", reason: "timeout", sessionId: cliSessionId });
await (runArgs.claimCliSessionFork as () => Promise<boolean>)();
await (runArgs.restoreCliSessionFork as () => Promise<void>)();
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<string, SessionEntry> = { [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<boolean>
)({ provider: "claude-cli", reason: "timeout", sessionId: cliSessionId });
await (runArgs.claimCliSessionFork as () => Promise<boolean>)();
await (runArgs.persistCliSessionForkSuccessor as (sessionId: string) => Promise<void>)(
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 () => {
src/agents/command/attempt-execution.ts
+46 -16
View File
@@ -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;
}
src/agents/command/session-store.test.ts
+10 -1
View File
@@ -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",
src/agents/command/session-store.ts
+19 -7
View File
@@ -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<string, SessionEntry>;
storePath: string;
expectedSessionId?: string;
expectedCliSessionId?: string;
}): Promise<SessionEntry | undefined> {
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;
src/config/sessions/cli-session-binding.ts
+1
View File
@@ -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),
src/config/sessions/types.ts
+2
View File
@@ -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. */
src/plugins/cli-backend.types.ts
+2
View File
@@ -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). */