From 2972db56497a70b1d2f6a280aff69ea289f6b87b Mon Sep 17 00:00:00 2001 From: Lu Wang Date: Fri, 17 Jul 2026 08:30:59 +0800 Subject: [PATCH] fix: preserve Claude resumed synthetic turns (#90799) Preserve Claude CLI replies that continue after a resumed-session synthetic placeholder while retaining bounded fallback for terminal no-output cases. Fixes #99131. Related #90789. Prepared head SHA: 08cd27aaeb7bdc9f723724ceffb175f99a027948 Co-authored-by: Lu Wang <7668944+wangwllu@users.noreply.github.com> Co-authored-by: Shakker <165377636+shakkernerd@users.noreply.github.com> Reviewed-by: @shakkernerd --- ...aude-live-session.background-tasks.test.ts | 348 +++++++++++++++++- src/agents/cli-runner/claude-live-session.ts | 127 +++++++ 2 files changed, 471 insertions(+), 4 deletions(-) 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 ec8bf239af06..cedccd53d910 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 @@ -1,4 +1,4 @@ -/** Claude live session: interim result while native background subagents run. */ +/** Claude live session: provisional results while native or queued work continues. */ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { setDiagnosticsEnabledForProcess, @@ -158,7 +158,12 @@ function jsonl(lines: unknown[]): string { return lines.map((line) => JSON.stringify(line)).join("\n") + "\n"; } -function startLiveTurn(params: { runId: string; timeoutMs?: number; noOutputTimeoutMs?: number }) { +function startLiveTurn(params: { + runId: string; + timeoutMs?: number; + noOutputTimeoutMs?: number; + useResume?: boolean; +}) { const context = buildPreparedCliRunContext({ runId: params.runId, timeoutMs: params.timeoutMs, @@ -168,7 +173,7 @@ function startLiveTurn(params: { runId: string; timeoutMs?: number; noOutputTime args: context.preparedBackend.backend.args ?? [], env: {}, prompt: "hi", - useResume: false, + useResume: params.useResume ?? false, noOutputTimeoutMs: params.noOutputTimeoutMs ?? 5_000, getProcessSupervisor: getProcessSupervisorForTest, onAssistantDelta: () => {}, @@ -176,7 +181,7 @@ function startLiveTurn(params: { runId: string; timeoutMs?: number; noOutputTime }); } -describe("claude live session background tasks", () => { +describe("claude live session provisional results", () => { it.each([ { taskType: "local_agent", label: "subagent" }, { taskType: "local_workflow", label: "workflow" }, @@ -371,6 +376,341 @@ describe("claude live session background tasks", () => { expect(driver.cancel).not.toHaveBeenCalled(); }); + it("keeps the turn open after a synthetic placeholder until the real result arrives", async () => { + const driver = installLiveStdoutDriver(); + const resultPromise = startLiveTurn({ + runId: "run-synthetic-placeholder", + useResume: true, + }); + await driver.stdout.waitReady(); + + driver.stdout.emit( + jsonl([ + { type: "system", subtype: "init", session_id: "live-synthetic" }, + { + type: "assistant", + session_id: "live-synthetic", + message: { + model: "", + role: "assistant", + content: [{ type: "text", text: "No response requested." }], + }, + }, + { + type: "result", + subtype: "success", + session_id: "live-synthetic", + result: "", + }, + ]), + ); + + let settled = false; + void resultPromise.then( + () => { + settled = true; + }, + () => { + settled = true; + }, + ); + await Promise.resolve(); + expect(settled).toBe(false); + expect(driver.cancel).not.toHaveBeenCalled(); + await waitForDiagnosticEventsDrained(); + expect( + getDiagnosticSessionActivitySnapshot({ sessionKey: "agent:main:bg" }).lastProgressReason, + ).toBe("cli_live:result_deferred_synthetic_placeholder"); + + driver.stdout.emit( + jsonl([ + { + type: "assistant", + session_id: "live-synthetic", + message: { + model: "claude-fable-5", + role: "assistant", + content: [{ type: "text", text: "The background work is complete." }], + }, + }, + { + type: "result", + subtype: "success", + session_id: "live-synthetic", + result: "The background work is complete.", + }, + ]), + ); + + const result = await resultPromise; + expect(result.output.text).toBe("The background work is complete."); + expect(driver.cancel).not.toHaveBeenCalled(); + }); + + it("does not defer ordinary or non-empty results that resemble a synthetic placeholder", async () => { + const ordinaryDriver = installLiveStdoutDriver({ + onWrite: (stdout) => { + stdout( + jsonl([ + { type: "system", subtype: "init", session_id: "live-ordinary-placeholder" }, + { + type: "assistant", + session_id: "live-ordinary-placeholder", + message: { + model: "claude-fable-5", + role: "assistant", + content: [{ type: "text", text: "No response requested." }], + }, + }, + { + type: "result", + subtype: "success", + session_id: "live-ordinary-placeholder", + result: "", + }, + ]), + ); + }, + }); + const ordinary = await startLiveTurn({ runId: "run-ordinary-placeholder" }); + expect(ordinary.output.text).toBe(""); + expect(ordinaryDriver.cancel).not.toHaveBeenCalled(); + + resetClaudeLiveSessionsForTest(); + const nonEmptyDriver = installLiveStdoutDriver({ + onWrite: (stdout) => { + stdout( + jsonl([ + { type: "system", subtype: "init", session_id: "live-synthetic-nonempty" }, + { + type: "assistant", + session_id: "live-synthetic-nonempty", + message: { + model: "", + role: "assistant", + content: [{ type: "text", text: "No response requested." }], + }, + }, + { + type: "result", + subtype: "success", + session_id: "live-synthetic-nonempty", + result: "real answer", + }, + ]), + ); + }, + }); + const nonEmpty = await startLiveTurn({ + runId: "run-synthetic-nonempty", + useResume: true, + }); + expect(nonEmpty.output.text).toBe("real answer"); + expect(nonEmptyDriver.cancel).not.toHaveBeenCalled(); + }); + + it("does not defer a synthetic placeholder on a fresh live process", async () => { + const driver = installLiveStdoutDriver({ + onWrite: (stdout) => { + stdout( + jsonl([ + { type: "system", subtype: "init", session_id: "live-synthetic-fresh" }, + { + type: "assistant", + session_id: "live-synthetic-fresh", + message: { + model: "", + role: "assistant", + content: [{ type: "text", text: "No response requested." }], + }, + }, + { + type: "result", + subtype: "success", + session_id: "live-synthetic-fresh", + result: "", + }, + ]), + ); + }, + }); + + const result = await startLiveTurn({ runId: "run-synthetic-fresh" }); + expect(result.output.text).toBe(""); + expect(driver.cancel).not.toHaveBeenCalled(); + }); + + it("expires a terminal resumed placeholder through the existing empty-result path", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + const driver = installLiveStdoutDriver(); + const resultPromise = startLiveTurn({ + runId: "run-synthetic-grace-expiry", + timeoutMs: 60_000, + noOutputTimeoutMs: 60_000, + useResume: true, + }); + await vi.advanceTimersByTimeAsync(0); + await driver.stdout.waitReady(); + + driver.stdout.emit( + jsonl([ + { type: "system", subtype: "init", session_id: "live-synthetic-expiry" }, + { + type: "assistant", + session_id: "live-synthetic-expiry", + message: { + model: "", + role: "assistant", + content: [{ type: "text", text: "No response requested." }], + }, + }, + { + type: "result", + subtype: "success", + session_id: "live-synthetic-expiry", + result: "", + }, + ]), + ); + + let settled = false; + void resultPromise.then(() => { + settled = true; + }); + await vi.advanceTimersByTimeAsync(29_999); + expect(settled).toBe(false); + + await vi.advanceTimersByTimeAsync(1); + const result = await resultPromise; + expect(result.output.text).toBe(""); + expect(driver.cancel).not.toHaveBeenCalled(); + }); + + it("expires the synthetic grace before a matching short no-output watchdog", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + const driver = installLiveStdoutDriver(); + const resultPromise = startLiveTurn({ + runId: "run-synthetic-short-watchdog", + timeoutMs: 60_000, + noOutputTimeoutMs: 1_000, + useResume: true, + }); + await vi.advanceTimersByTimeAsync(0); + await driver.stdout.waitReady(); + + driver.stdout.emit( + jsonl([ + { type: "system", subtype: "init", session_id: "live-synthetic-short-watchdog" }, + { + type: "assistant", + session_id: "live-synthetic-short-watchdog", + message: { + model: "", + role: "assistant", + content: [{ type: "text", text: "No response requested." }], + }, + }, + { + type: "result", + subtype: "success", + session_id: "live-synthetic-short-watchdog", + result: "", + }, + ]), + ); + + await vi.advanceTimersByTimeAsync(999); + let settled = false; + void resultPromise.then(() => { + settled = true; + }); + await Promise.resolve(); + expect(settled).toBe(false); + + await vi.advanceTimersByTimeAsync(1); + const result = await resultPromise; + expect(result.output.text).toBe(""); + expect(driver.cancel).not.toHaveBeenCalled(); + }); + + it("still aborts on the turn timeout while waiting after a synthetic placeholder", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + const driver = installLiveStdoutDriver(); + const resultPromise = startLiveTurn({ + runId: "run-synthetic-timeout", + timeoutMs: 5_000, + noOutputTimeoutMs: 60_000, + useResume: true, + }); + await vi.advanceTimersByTimeAsync(0); + await driver.stdout.waitReady(); + + driver.stdout.emit( + jsonl([ + { type: "system", subtype: "init", session_id: "live-synthetic-timeout" }, + { + type: "assistant", + session_id: "live-synthetic-timeout", + message: { + model: "", + role: "assistant", + content: [{ type: "text", text: "Continue from where you left off." }], + }, + }, + { + type: "result", + subtype: "success", + session_id: "live-synthetic-timeout", + result: "", + }, + ]), + ); + + const rejection = expect(resultPromise).rejects.toMatchObject({ + name: "FailoverError", + message: expect.stringMatching(/exceeded timeout/i), + }); + await vi.advanceTimersByTimeAsync(5_000); + await rejection; + expect(driver.cancel).toHaveBeenCalledWith("manual-cancel"); + }); + + it("fails immediately when an error result follows a synthetic placeholder", async () => { + const driver = installLiveStdoutDriver(); + const resultPromise = startLiveTurn({ + runId: "run-synthetic-error", + useResume: true, + }); + await driver.stdout.waitReady(); + + driver.stdout.emit( + jsonl([ + { type: "system", subtype: "init", session_id: "live-synthetic-error" }, + { + type: "assistant", + session_id: "live-synthetic-error", + message: { + model: "", + role: "assistant", + content: [{ type: "text", text: "No response requested." }], + }, + }, + { + type: "result", + subtype: "error_during_execution", + is_error: true, + session_id: "live-synthetic-error", + result: "provider failed", + }, + ]), + ); + + await expect(resultPromise).rejects.toMatchObject({ + name: "FailoverError", + rawError: expect.stringMatching(/provider failed/i), + }); + }); + it("fails the turn on an error result even when background tasks are outstanding", async () => { const driver = installLiveStdoutDriver(); const resultPromise = startLiveTurn({ runId: "run-bg-error" }); diff --git a/src/agents/cli-runner/claude-live-session.ts b/src/agents/cli-runner/claude-live-session.ts index fb4ae71ae134..30465a6e3a93 100644 --- a/src/agents/cli-runner/claude-live-session.ts +++ b/src/agents/cli-runner/claude-live-session.ts @@ -68,6 +68,15 @@ type ClaudeLiveTurn = { timeoutTimer: NodeJS.Timeout | null; activeTools: Map; observedStdout: boolean; + /** + * Claude consumed queued session notifications before processing this turn. + * The following empty result is provisional; the same process can emit the + * real answer later, so a bounded grace observes whether output continues. + */ + pendingSyntheticPlaceholder: boolean; + allowSyntheticContinuationGrace: boolean; + deferredSyntheticOutput: CliOutput | null; + syntheticContinuationTimer: NodeJS.Timeout | null; completedToolCallIds: Set; toolEventCount: number; streamingParser: ReturnType; @@ -129,6 +138,17 @@ type ClaudeLiveToolTerminalOutcome = | { outcome: "cancelled" | "failed" | "timed_out" | "unknown" }; const CLAUDE_LIVE_IDLE_TIMEOUT_MS = 10 * 60 * 1_000; const CLAUDE_LIVE_CLOSE_WAIT_TIMEOUT_MS = 5_000; +// The observed queued-notification resume emits new process activity within +// seconds. Cap this below the normal resumed no-output watchdog so terminal +// placeholders still reach existing empty-response handling promptly. +const CLAUDE_LIVE_SYNTHETIC_CONTINUATION_GRACE_MS = 30_000; +// Claude Code uses these exact messages while draining internal +// session work. Matching both the model sentinel and full text avoids treating +// user-authored lookalikes as lifecycle signals. +const CLAUDE_LIVE_PROVISIONAL_SYNTHETIC_PLACEHOLDERS = new Set([ + "No response requested.", + "Continue from where you left off.", +]); const liveSessions = new Map(); const liveSessionCreates = new Map(); @@ -437,6 +457,11 @@ function clearTurnTimers(turn: ClaudeLiveTurn): void { clearTimeout(turn.timeoutTimer); turn.timeoutTimer = null; } + if (turn.syntheticContinuationTimer) { + clearTimeout(turn.syntheticContinuationTimer); + turn.syntheticContinuationTimer = null; + } + turn.deferredSyntheticOutput = null; } function clearOutstandingBackgroundTasks(session: ClaudeLiveSession): void { @@ -817,6 +842,87 @@ function applyBackgroundTasksChanged( } } +function isClaudeLiveProvisionalSyntheticPlaceholder(parsed: Record): boolean { + if (parsed.type !== "assistant" || !isRecord(parsed.message)) { + return false; + } + const message = parsed.message; + if (message.model !== "") { + return false; + } + const content = Array.isArray(message.content) ? message.content : []; + const text = content + .flatMap((block) => + isRecord(block) && block.type === "text" && typeof block.text === "string" + ? [block.text] + : [], + ) + .join("") + .trim(); + return CLAUDE_LIVE_PROVISIONAL_SYNTHETIC_PLACEHOLDERS.has(text); +} + +function isClaudeLiveSubstantiveAssistantProgress(parsed: Record): boolean { + if (parsed.type === "assistant" && isRecord(parsed.message)) { + return parsed.message.model !== ""; + } + if (parsed.type !== "stream_event" || !isRecord(parsed.event)) { + return false; + } + const event = parsed.event; + return ( + event.type === "content_block_delta" && + isRecord(event.delta) && + event.delta.type === "text_delta" && + typeof event.delta.text === "string" && + event.delta.text.length > 0 + ); +} + +function deferClaudeLiveSyntheticResult( + session: ClaudeLiveSession, + turn: ClaudeLiveTurn, + output: CliOutput, +): void { + turn.pendingSyntheticPlaceholder = false; + turn.deferredSyntheticOutput = output; + if (turn.noOutputTimer) { + clearTimeout(turn.noOutputTimer); + turn.noOutputTimer = null; + } + if (turn.syntheticContinuationTimer) { + clearTimeout(turn.syntheticContinuationTimer); + } + const graceMs = Math.min(CLAUDE_LIVE_SYNTHETIC_CONTINUATION_GRACE_MS, session.noOutputTimeoutMs); + turn.syntheticContinuationTimer = setTimeout(() => { + if (session.currentTurn !== turn || !turn.deferredSyntheticOutput) { + return; + } + const terminalOutput = turn.deferredSyntheticOutput; + turn.syntheticContinuationTimer = null; + turn.deferredSyntheticOutput = null; + emitClaudeLiveProgress(turn, "cli_live:synthetic_placeholder_grace_expired"); + finishTurn(session, terminalOutput); + }, graceMs); + emitClaudeLiveProgress(turn, "cli_live:result_deferred_synthetic_placeholder"); +} + +function noteClaudeLiveContinuationAfterSyntheticPlaceholder( + session: ClaudeLiveSession, + turn: ClaudeLiveTurn, +): void { + if (!turn.deferredSyntheticOutput) { + return; + } + if (turn.syntheticContinuationTimer) { + clearTimeout(turn.syntheticContinuationTimer); + turn.syntheticContinuationTimer = null; + } + turn.deferredSyntheticOutput = null; + armNoOutputTimer(session, turn, session.noOutputTimeoutMs); + emitClaudeLiveProgress(turn, "cli_live:synthetic_placeholder_continuation"); +} + function resetNoOutputTimer(session: ClaudeLiveSession): void { const turn = session.currentTurn; if (!turn) { @@ -974,6 +1080,7 @@ function handleClaudeLiveLine(session: ClaudeLiveSession, line: string): void { if (!turn) { return; } + noteClaudeLiveContinuationAfterSyntheticPlaceholder(session, turn); turn.rawChars += trimmed.length + 1; if ( turn.rawChars > turn.outputLimits.maxTurnRawChars || @@ -988,6 +1095,11 @@ function handleClaudeLiveLine(session: ClaudeLiveSession, line: string): void { } turn.rawLines.push(trimmed); applyBackgroundTasksChanged(session, parsed); + if (turn.allowSyntheticContinuationGrace && isClaudeLiveProvisionalSyntheticPlaceholder(parsed)) { + turn.pendingSyntheticPlaceholder = true; + } else if (turn.pendingSyntheticPlaceholder && isClaudeLiveSubstantiveAssistantProgress(parsed)) { + turn.pendingSyntheticPlaceholder = false; + } const toolEventCountBefore = turn.toolEventCount; turn.streamingParser.push(`${trimmed}\n`); turn.sessionId = parsedSessionId ?? turn.sessionId; @@ -1029,6 +1141,13 @@ function handleClaudeLiveLine(session: ClaudeLiveSession, line: string): void { emitClaudeLiveProgress(turn, "cli_live:result_deferred_background_tasks"); return; } + // A resumed Claude session can first consume queued task notifications and + // emit an empty synthetic result, then continue the same user turn. Keep the + // live process and watchdogs authoritative instead of racing it with fallback. + if (turn.pendingSyntheticPlaceholder && !output.text.trim()) { + deferClaudeLiveSyntheticResult(session, turn, output); + return; + } finishTurn(session, output); } @@ -1242,6 +1361,7 @@ async function createClaudeLiveSession(params: { function createTurn(params: { context: PreparedCliRunContext; noOutputTimeoutMs: number; + allowSyntheticContinuationGrace: boolean; onAssistantDelta: (delta: CliStreamingDelta) => void; onThinkingDelta?: (delta: CliThinkingDelta) => void; onThinkingProgress?: (progress: CliThinkingProgress) => void; @@ -1275,6 +1395,10 @@ function createTurn(params: { timeoutTimer: null, activeTools: new Map(), observedStdout: false, + pendingSyntheticPlaceholder: false, + allowSyntheticContinuationGrace: params.allowSyntheticContinuationGrace, + deferredSyntheticOutput: null, + syntheticContinuationTimer: null, completedToolCallIds: new Set(), toolEventCount: 0, streamingParser: createCliJsonlStreamingParser({ @@ -1402,6 +1526,7 @@ export async function runClaudeLiveSessionTurn(params: { env: params.env, }); let cleanupDone = false; + let createdSessionForTurn = false; const cleanup = async () => { if (cleanupDone) { return; @@ -1552,6 +1677,7 @@ export async function runClaudeLiveSessionTurn(params: { liveSessionCreates.set(key, { generation, promise: createSession }); try { session = await createSession; + createdSessionForTurn = true; } catch (error) { await cleanup(); throw error; @@ -1593,6 +1719,7 @@ export async function runClaudeLiveSessionTurn(params: { liveSession.currentTurn = createTurn({ context: params.context, noOutputTimeoutMs: params.noOutputTimeoutMs, + allowSyntheticContinuationGrace: params.useResume && createdSessionForTurn, onAssistantDelta: params.onAssistantDelta, onThinkingDelta: params.onThinkingDelta, onThinkingProgress: params.onThinkingProgress,