From e47092227e6a9bdb18f20cdee1be8c4405c12e67 Mon Sep 17 00:00:00 2001 From: Luca Lowndes Date: Tue, 11 Aug 2026 12:03:33 +1000 Subject: [PATCH] fix: stop stalled Claude turns after inactivity --- src/harness/claude-harness.ts | 108 ++++++++++++++++++++----- test/claude-harness-turn.test.ts | 131 ++++++++++++++++++++++++++++++- 2 files changed, 217 insertions(+), 22 deletions(-) diff --git a/src/harness/claude-harness.ts b/src/harness/claude-harness.ts index 0d4bf651..a6b31ed0 100644 --- a/src/harness/claude-harness.ts +++ b/src/harness/claude-harness.ts @@ -52,6 +52,7 @@ export interface ClaudeHarnessOptions { reachExec?: boolean; controlTools?: boolean; turnWallClockMs?: number; + modelInactivityMs?: number; execTimeoutMs?: number; execTimeoutCeilingMs?: number; backgroundJobTtlMs?: number; @@ -100,6 +101,7 @@ type BridgedTool = { const CHILD_TOOL_NAMES = new Set(["execute", "read", "write", "publish", "memory", "history", "background"]); const CLAUDE_CHILD_AGENT_TYPES = new Set(["research", "code", "consult"]); +const CLAUDE_MODEL_INACTIVITY_MS = 5 * 60_000; const CLAUDE_ENV_PASSTHROUGH = [ "PATH", "TMPDIR", @@ -348,6 +350,9 @@ export function createClaudeHarness(opts: ClaudeHarnessOptions = {}): Harness { consult: { description: "Provide an independent expert analysis.", prompt: childPolicy, tools: childToolNames }, }; const queue = new MessageQueue(); + let toolStarted = () => {}; + let toolFinished = () => {}; + let stopInactivity = () => {}; let terminateProvider = () => { queue.close(); controller.abort(); @@ -358,6 +363,7 @@ export function createClaudeHarness(opts: ClaudeHarnessOptions = {}): Harness { const callId = String( (extra as { toolUseId?: string } | undefined)?.toolUseId ?? randomBytes(8).toString("hex"), ); + toolStarted(); try { const result = await definition.execute(callId, args); if (result.terminate || ref.pausedOnApproval || ref.silentRequested) setImmediate(terminateProvider); @@ -367,6 +373,8 @@ export function createClaudeHarness(opts: ClaudeHarnessOptions = {}): Harness { content: [{ type: "text", text: error instanceof Error ? error.message : String(error) }], isError: true, }; + } finally { + toolFinished(); } }); }); @@ -391,7 +399,10 @@ export function createClaudeHarness(opts: ClaudeHarnessOptions = {}): Harness { for (const value of thinking.splice(0)) await turn.emit({ type: "thinking", payload: { thinking: value }, scopeLabel: turn.scopeLabel }); }; - const taskStates = new Map(); + const taskStates = new Map< + string, + { callId: string; status: TaskStatus; resultEmitted: boolean; active: boolean } + >(); const callUsage = new Map(); let stepCallIds = new Set(); let recordedSteps = 0; @@ -480,9 +491,10 @@ export function createClaudeHarness(opts: ClaudeHarnessOptions = {}): Harness { active.add(sdkQuery); const interrupt = async (fromUser: boolean) => { stopped ||= fromUser; + stopInactivity(); queue.close(); - await sdkQuery.interrupt().catch(() => undefined); controller.abort(); + await sdkQuery.interrupt().catch(() => undefined); }; terminateProvider = () => { void interrupt(false); @@ -518,6 +530,43 @@ export function createClaudeHarness(opts: ClaudeHarnessOptions = {}): Harness { : null; const wallMs = turn.turnWallClockMs ?? defaultTurnWallClockMs; let timer: NodeJS.Timeout | undefined; + let inactivityTimer: NodeJS.Timeout | undefined; + let activeTools = 0; + let inactivityStopped = false; + let waitingForModel = false; + let inactivityError: Error | undefined; + let rejectInactivity = (_error: Error) => {}; + const inactivity = new Promise((_, reject) => { + rejectInactivity = reject; + }); + const resetInactivity = () => { + if (inactivityTimer) clearTimeout(inactivityTimer); + if (settled || inactivityStopped || !waitingForModel || activeTools > 0) return; + inactivityTimer = setTimeout(() => { + inactivityError = new Error( + `Claude model produced no activity for ${Math.round((opts.modelInactivityMs ?? CLAUDE_MODEL_INACTIVITY_MS) / 1000)}s`, + ); + rejectInactivity(inactivityError); + void interrupt(false); + }, opts.modelInactivityMs ?? CLAUDE_MODEL_INACTIVITY_MS); + }; + const pauseInactivity = () => { + if (inactivityTimer) clearTimeout(inactivityTimer); + }; + toolStarted = () => { + if (settled || inactivityStopped) return; + activeTools++; + pauseInactivity(); + }; + toolFinished = () => { + if (settled || inactivityStopped) return; + activeTools = Math.max(0, activeTools - 1); + resetInactivity(); + }; + stopInactivity = () => { + inactivityStopped = true; + pauseInactivity(); + }; let signalsStopped = false; const recordedRequest = { system: turn.systemPrompt, @@ -572,11 +621,23 @@ export function createClaudeHarness(opts: ClaudeHarnessOptions = {}): Harness { } }; try { - await sdkQuery.initializationResult(); - await appendTape(initial, true); - queue.push(initial); const consume = (async () => { - for await (const message of sdkQuery) { + waitingForModel = true; + resetInactivity(); + await sdkQuery.initializationResult(); + waitingForModel = false; + pauseInactivity(); + await appendTape(initial, true); + queue.push(initial); + const iterator = sdkQuery[Symbol.asyncIterator](); + while (!settled) { + waitingForModel = true; + resetInactivity(); + const next = await iterator.next(); + waitingForModel = false; + pauseInactivity(); + if (next.done) break; + const message = next.value; if (settled) break; if (message.type === "assistant") { const usage = message.message.usage; @@ -620,7 +681,8 @@ export function createClaudeHarness(opts: ClaudeHarnessOptions = {}): Harness { title: message.description || message.prompt || "subagent task", status: "in_progress", }); - taskStates.set(message.task_id, { callId, status: "in_progress", resultEmitted: false }); + taskStates.set(message.task_id, { callId, status: "in_progress", resultEmitted: false, active: true }); + toolStarted(); if (!message.skip_transcript) { await turn.emit({ type: "tool_call", @@ -644,6 +706,10 @@ export function createClaudeHarness(opts: ClaudeHarnessOptions = {}): Harness { if (tracked && next && next !== tracked.status) { await transitionTask(opts.tasks, message.task_id, tracked.status, next, turn.runId ?? turn.session.id); tracked.status = next; + if (next !== "in_progress" && tracked.active) { + tracked.active = false; + toolFinished(); + } } } if (message.type === "system" && message.subtype === "task_notification") { @@ -654,6 +720,10 @@ export function createClaudeHarness(opts: ClaudeHarnessOptions = {}): Harness { await transitionTask(opts.tasks, message.task_id, tracked.status, next, turn.runId ?? turn.session.id); tracked.status = next; } + if (tracked.active) { + tracked.active = false; + toolFinished(); + } if (!tracked.resultEmitted && !message.skip_transcript) { tracked.resultEmitted = true; await turn.emit({ @@ -699,18 +769,19 @@ export function createClaudeHarness(opts: ClaudeHarnessOptions = {}): Harness { } })(); try { - await (wallMs > 0 - ? Promise.race([ - consume, - new Promise((_, reject) => { - timer = setTimeout(() => { - void interrupt(false); - reject(new NonRetryableTurnError(`Claude turn exceeded ${Math.round(wallMs / 1000)}s wall clock`)); - }, wallMs); - }), - ]) - : consume); + const completion: Array | Promise> = [consume, inactivity]; + if (wallMs > 0) + completion.push( + new Promise((_, reject) => { + timer = setTimeout(() => { + void interrupt(false); + reject(new NonRetryableTurnError(`Claude turn exceeded ${Math.round(wallMs / 1000)}s wall clock`)); + }, wallMs); + }), + ); + await Promise.race(completion); } catch (error) { + if (error === inactivityError) throw error; if (!controller.signal.aborted || error instanceof NonRetryableTurnError) throw error; const reply = streamedText.trim(); await flushThinking(); @@ -790,6 +861,7 @@ export function createClaudeHarness(opts: ClaudeHarnessOptions = {}): Harness { } finally { settled = true; if (timer) clearTimeout(timer); + stopInactivity(); if (recordedSteps === 0) { recordedSteps++; try { diff --git a/test/claude-harness-turn.test.ts b/test/claude-harness-turn.test.ts index 5b5f4654..fa3d028d 100644 --- a/test/claude-harness-turn.test.ts +++ b/test/claude-harness-turn.test.ts @@ -6,19 +6,31 @@ import type { NewEntry } from "../src/sessions/session-store.ts"; import type { ScopeId, SessionEntry } from "../src/types.ts"; type FakeSdkMessage = Record; -type Script = (prompts: AsyncIterable<{ message: { content: unknown } }>) => AsyncGenerator; +type Script = ( + prompts: AsyncIterable<{ message: { content: unknown } }>, + signal: AbortSignal, +) => AsyncGenerator; let currentScript: Script = async function* () {}; +let currentInitialization = async () => ({}); +let interruptHangs = false; mock.module("@anthropic-ai/claude-agent-sdk", { namedExports: { - query: ({ prompt }: { prompt: AsyncIterable<{ message: { content: unknown } }> }) => { - const generator = currentScript(prompt); + query: ({ + prompt, + options, + }: { + prompt: AsyncIterable<{ message: { content: unknown } }>; + options: { abortController: AbortController }; + }) => { + const generator = currentScript(prompt, options.abortController.signal); return { async initializationResult() { - return {}; + return currentInitialization(); }, async interrupt() { + if (interruptHangs) await new Promise(() => {}); await generator.return?.(undefined as never); }, close() { @@ -271,6 +283,117 @@ test("a turn that dies before its first result still records exactly one request assert.equal((llmRequests[0]!.request as { system: string }).system, "be brief"); }); +test("a Claude turn fails when the model stream stays inactive", async () => { + currentScript = async function* (prompts) { + await prompts[Symbol.asyncIterator]().next(); + await new Promise(() => {}); + yield resultMessage("unreachable"); + }; + + const harness = createClaudeHarness({ modelInactivityMs: 20 }); + const { turn } = harnessTurn(); + + await assert.rejects(() => harness.turns.runTurn(turn), /Claude model produced no activity/); +}); + +test("Claude initialization must produce activity before the inactivity deadline", async () => { + currentInitialization = async () => new Promise(() => {}); + const harness = createClaudeHarness({ modelInactivityMs: 20 }); + const { turn } = harnessTurn(); + + try { + await assert.rejects(() => harness.turns.runTurn(turn), /Claude model produced no activity/); + } finally { + currentInitialization = async () => ({}); + } +}); + +test("Claude stream activity resets the inactivity deadline", async () => { + currentScript = async function* (prompts) { + await prompts[Symbol.asyncIterator]().next(); + await new Promise((resolve) => setTimeout(resolve, 15)); + yield assistantMessage("msg_A", "working", { + input_tokens: 1, + output_tokens: 1, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + }); + await new Promise((resolve) => setTimeout(resolve, 15)); + yield resultMessage("done"); + }; + + const harness = createClaudeHarness({ modelInactivityMs: 25 }); + const { turn } = harnessTurn(); + + assert.equal((await harness.turns.runTurn(turn)).reply, "done"); +}); + +test("slow local message persistence does not count as model inactivity", async () => { + currentScript = async function* (prompts) { + await prompts[Symbol.asyncIterator]().next(); + yield assistantMessage("msg_A", "working", { + input_tokens: 1, + output_tokens: 1, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + }); + yield resultMessage("done"); + }; + + const harness = createClaudeHarness({ modelInactivityMs: 25 }); + const { turn } = harnessTurn({ tape: async () => new Promise((resolve) => setTimeout(resolve, 40)) }); + + assert.equal((await harness.turns.runTurn(turn)).reply, "done"); +}); + +test("cancellation aborts a silent model even when the SDK interrupt hangs", async () => { + const cancel = new AbortController(); + interruptHangs = true; + currentScript = async function* (prompts, signal) { + await prompts[Symbol.asyncIterator]().next(); + await new Promise((resolve) => signal.addEventListener("abort", () => resolve(), { once: true })); + if (!signal.aborted) yield resultMessage("unreachable"); + }; + + const harness = createClaudeHarness({ modelInactivityMs: 25 }); + const { turn } = harnessTurn({ cancel: cancel.signal }); + setTimeout(() => cancel.abort(), 5); + + try { + assert.equal((await harness.turns.runTurn(turn)).stopped, true); + } finally { + interruptHangs = false; + } +}); + +test("Claude Agent work pauses the inactivity deadline", async () => { + currentScript = async function* (prompts) { + await prompts[Symbol.asyncIterator]().next(); + yield { + type: "system", + subtype: "task_started", + task_id: "task-1", + tool_use_id: "tool-1", + description: "long-running child task", + prompt: "work", + }; + await new Promise((resolve) => setTimeout(resolve, 40)); + yield { + type: "system", + subtype: "task_notification", + task_id: "task-1", + status: "completed", + summary: "finished", + }; + yield resultMessage("done"); + }; + + const harness = createClaudeHarness({ modelInactivityMs: 25 }); + const { turn } = harnessTurn(); + + assert.equal((await harness.turns.runTurn(turn)).reply, "done"); +}); + test("the claude harness offers compaction and detection so a utility role cannot silently disable them", async () => { const harness = createClaudeHarness({}); assert.equal(typeof harness.models.compactHistory, "function");