Skip to content

fix: stop stalled Claude turns after inactivity #327

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
LucaLow wants to merge 1 commit into
base: main
Choose a base branch
from
+217 −22
Open

fix: stop stalled Claude turns after inactivity #327

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
108 changes: 90 additions & 18 deletions src/harness/claude-harness.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@ export interface ClaudeHarnessOptions {
reachExec?: boolean;
controlTools?: boolean;
turnWallClockMs?: number;
modelInactivityMs?: number;
execTimeoutMs?: number;
execTimeoutCeilingMs?: number;
backgroundJobTtlMs?: number;
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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();
Expand All @@ -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);
Expand All @@ -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();
}
});
});
Expand All @@ -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<string, { callId: string; status: TaskStatus; resultEmitted: boolean }>();
const taskStates = new Map<
string,
{ callId: string; status: TaskStatus; resultEmitted: boolean; active: boolean }
>();
const callUsage = new Map<string, { input: number; output: number; cacheRead: number; cacheWrite: number }>();
let stepCallIds = new Set<string>();
let recordedSteps = 0;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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<never>((_, 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,
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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",
Expand All @@ -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") {
Expand All @@ -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({
Expand Down Expand Up @@ -699,18 +769,19 @@ export function createClaudeHarness(opts: ClaudeHarnessOptions = {}): Harness {
}
})();
try {
await (wallMs > 0
? Promise.race([
consume,
new Promise<never>((_, 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<void> | Promise<never>> = [consume, inactivity];
if (wallMs > 0)
completion.push(
new Promise<never>((_, 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();
Expand Down Expand Up @@ -790,6 +861,7 @@ export function createClaudeHarness(opts: ClaudeHarnessOptions = {}): Harness {
} finally {
settled = true;
if (timer) clearTimeout(timer);
stopInactivity();
if (recordedSteps === 0) {
recordedSteps++;
try {
Expand Down
131 changes: 127 additions & 4 deletions test/claude-harness-turn.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,19 +6,31 @@ import type { NewEntry } from "../src/sessions/session-store.ts";
import type { ScopeId, SessionEntry } from "../src/types.ts";

type FakeSdkMessage = Record<string, unknown>;
type Script = (prompts: AsyncIterable<{ message: { content: unknown } }>) => AsyncGenerator<FakeSdkMessage>;
type Script = (
prompts: AsyncIterable<{ message: { content: unknown } }>,
signal: AbortSignal,
) => AsyncGenerator<FakeSdkMessage>;

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() {
Expand Down Expand Up @@ -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();
Comment on lines +314 to +326

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<void>((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");
Expand Down