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

fix: preserve Claude resumed synthetic turns (#90799)

Browse Source 
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: 08cd27aaeb
Co-authored-by: Lu Wang <7668944+wangwllu@users.noreply.github.com>
Co-authored-by: Shakker <165377636+shakkernerd@users.noreply.github.com>
Reviewed-by: @shakkernerd
This commit is contained in:
Lu Wang
2026-07-17 08:30:59 +08:00
committed by GitHub
parent d71c1fe596
commit 2972db5649
2 changed files with 471 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agents/cli-runner/claude-live-session.background-tasks.test.ts
+344 -4
View File
@@ -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: "<synthetic>",
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: "<synthetic>",
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: "<synthetic>",
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: "<synthetic>",
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: "<synthetic>",
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: "<synthetic>",
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: "<synthetic>",
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" });
src/agents/cli-runner/claude-live-session.ts
+127
View File
@@ -68,6 +68,15 @@ type ClaudeLiveTurn = {
timeoutTimer: NodeJS.Timeout | null;
activeTools: Map<string, ClaudeLiveActiveTool>;
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<string>;
toolEventCount: number;
streamingParser: ReturnType<typeof createCliJsonlStreamingParser>;
@@ -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 <synthetic> 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<string, ClaudeLiveSession>();
const liveSessionCreates = new Map<string, ClaudeLiveSessionCreate>();
@@ -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<string, unknown>): boolean {
if (parsed.type !== "assistant" || !isRecord(parsed.message)) {
return false;
}
const message = parsed.message;
if (message.model !== "<synthetic>") {
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<string, unknown>): boolean {
if (parsed.type === "assistant" && isRecord(parsed.message)) {
return parsed.message.model !== "<synthetic>";
}
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,