From 7272682861ba115bbf7d3e7a6160eb708cdbb299 Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Sun, 28 Jun 2026 18:08:20 +0200 Subject: [PATCH 1/7] fix: remove premature tool cancellation from text_done handler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #completeActiveTools() was called in the text_done event handler. In the real Claude Agent SDK the committed assistant message (which fires text_done) is emitted BEFORE the user/tool_result message (which fires tool_complete). This meant any tool still in #activeToolMsgIds at text_done time was immediately cancelled, then tool_complete arrived but found the correlation maps cleared — so the tool stayed permanently "cancelled — interrupted" in the UI. The prior T8c fix removed the call from tool_start but kept it in text_done, relying on the wrong assumption that the mock ordering (tool_complete before text_done) matched the real SDK. The mock tests passed while the real-world bug persisted. Fix: remove #completeActiveTools() from text_done entirely. The #consumeEvents finally block is the correct and sufficient cleanup point — it runs on every turn end (normal, interrupted, or error) and only touches tools whose tool_complete never arrived. Adds T8(d) regression guard: tool_start → text_done → tool_complete → turn_done must produce "completed", not "cancelled". Co-Authored-By: Claude Sonnet 4.6 --- src/daemon/session.ts | 8 +++-- src/tests/session-integration.test.ts | 50 +++++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 2 deletions(-) diff --git a/src/daemon/session.ts b/src/daemon/session.ts index 7a992a2..5fd44da 100644 --- a/src/daemon/session.ts +++ b/src/daemon/session.ts @@ -1636,7 +1636,11 @@ export class Session { case "text_done": { this.#accumulator.handleEvent(event); - this.#completeActiveTools(); + // NOTE: do NOT call #completeActiveTools() here. + // In the real SDK the committed assistant message (which fires text_done) + // is emitted BEFORE the user/tool_result message (which fires tool_complete). + // Calling completeActiveTools here cancels tools that are still executing. + // Cleanup is handled solely by the #consumeEvents finally block. if (this.#activeAssistantMsg) { this.#activeAssistantMsg.content = event.content; this.#activeAssistantMsg.parts = [{ kind: "text", text: event.content, markdown: true }]; @@ -1685,7 +1689,7 @@ export class Session { // NOTE: do NOT call #completeActiveTools() here. It would cancel all // currently in-flight tools, which is correct for sequential calls but // silently kills parallel tool calls from concurrent subagents. - // Cleanup is handled by text_done and the #consumeEvents finally block. + // Cleanup is handled solely by the #consumeEvents finally block. // Await the ZeroID registration fence so sub-agent identity is resolved // before we attribute this tool call. Bounded by a 5s timeout so a // hung ZeroID service can't stall the event loop indefinitely. diff --git a/src/tests/session-integration.test.ts b/src/tests/session-integration.test.ts index 7a8a94a..115b2b6 100644 --- a/src/tests/session-integration.test.ts +++ b/src/tests/session-integration.test.ts @@ -755,4 +755,54 @@ describe("T8 – regression guards: text overlap + stale status", () => { expect(finalPhases).not.toContain("cancelled"); expect(finalPhases.every((p) => p === "completed")).toBe(true); }); + + it("(d) text_done arriving before tool_complete does not cancel the in-flight tool", async () => { + // The real Claude Agent SDK emits the committed assistant message (→ text_done) + // BEFORE the user/tool_result message (→ tool_complete). A previous bug had + // #completeActiveTools() in the text_done handler which cancelled any tool + // still in #activeToolMsgIds at that moment, producing ghost + // "cancelled — interrupted" cards in the UI. + // + // This test uses the real-world SDK ordering: + // tool_start → text_done → tool_complete → turn_done + // to verify the tool ends as "completed", not "cancelled". + const toolUseId = "sdk-real-order"; + const provider = new MockSessionProvider("claude", [ + [ + { + type: "tool_start", + toolId: "t-real", + sdkToolUseId: toolUseId, + name: "Read", + input: { file_path: "/tmp/test.ts" }, + approvalId: "approval-real", + }, + // text_done fires from the committed assistant message — BEFORE tool_result. + { type: "text_done", content: "Reading the file." }, + { + type: "tool_complete", + sdkToolUseId: toolUseId, + output: "file contents", + success: true, + }, + { type: "turn_done", result: mockResult({ providerId: "claude" }) }, + ], + ]); + + const session = makeSession(provider); + const { client, received } = makeClient(); + session.attach(client); + + await session.send("run bash", TEST_AUTH); + await waitForIdle(session); + + const toolMsg = received.find((m) => m.type === "session.message" && m.role === "tool_call"); + expect(toolMsg).toBeDefined(); + const msgId = (toolMsg as { messageId?: string }).messageId!; + const lastDelta = received + .filter((d) => d.type === "session.message.delta" && (d as { messageId?: string }).messageId === msgId && (d as { toolStateUpdate?: { phase?: string } }).toolStateUpdate) + .at(-1) as { toolStateUpdate?: { phase?: string } } | undefined; + const finalPhase = lastDelta?.toolStateUpdate?.phase ?? (toolMsg as { tool?: { state?: { phase?: string } } }).tool?.state?.phase; + expect(finalPhase).toBe("completed"); + }); }); From 7f0b8ce24ece78340565dcbd508566c69800cc38 Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Sun, 28 Jun 2026 18:34:18 +0200 Subject: [PATCH 2/7] fix: consume mid-turn continuation turn after SDK interrupts current turn MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a now-priority message is pushed mid-turn, the SDK ends the current turn (emitting turn_done, often with isError) before starting the continuation turn for the injected message. Previously, #consumeEvents broke on that first turn_done and left the continuation turn unread — the session went idle/error and the user had to send "please continue" manually. The fix adds #pendingMidTurnCount (incremented in #sendInner before each pushMidTurn call, decremented in #consumeEvents when the intermediate turn_done is absorbed). When a turn_done arrives with a pending count, #consumeEvents flushes per-turn state (active tools, assistant stream, thinking, stale approvals), records partial cost, re-asserts "thinking" status, and continues the loop instead of breaking. The counter is always reset to 0 in the finally block and on explicit interrupt() so no orphaned count can block a future turn. A new T9 integration test drives a custom TurnRun with pushMidTurn support and verifies the session ends idle with the continuation message present and no "Error:" system message. Co-Authored-By: Claude Sonnet 4.6 --- src/daemon/session.ts | 39 +++++++ src/tests/session-integration.test.ts | 150 ++++++++++++++++++++++++++ 2 files changed, 189 insertions(+) diff --git a/src/daemon/session.ts b/src/daemon/session.ts index 5fd44da..ab4c242 100644 --- a/src/daemon/session.ts +++ b/src/daemon/session.ts @@ -153,6 +153,12 @@ export class Session { #seq = 0; #memory?: MemoryEngine; #chunker?: EpisodeChunker; + // Counts mid-turn messages in flight. When the SDK interrupts the current + // turn to process a pushMidTurn() injection, it emits a turn_done for the + // aborted partial turn BEFORE the continuation turn starts. This counter lets + // #consumeEvents absorb those intermediate turn_dones and keep looping instead + // of exiting the consumer and leaving the continuation turn without a reader. + #pendingMidTurnCount = 0; #indexScheduler?: IndexScheduler; #workspaceId: string; @@ -699,6 +705,7 @@ export class Session { this.#persistAndBuffer(midTurnMsg); this.#broadcastRaw(midTurnMsg); this.#accumulator.pushUserTurn(effectivePrompt); + this.#pendingMidTurnCount++; this.#activeRun.pushMidTurn(effectivePrompt, effectivePriority ?? "now"); this.#setStatus("thinking"); this.#broadcastInfoUpdate(); @@ -790,6 +797,7 @@ export class Session { */ async interrupt(sender: AuthContext): Promise { this.#store.audit(sender.sub, "session.interrupt", this.id); + this.#pendingMidTurnCount = 0; // cancel pending mid-turn continuations // Finalize any in-flight streaming messages RIGHT NOW so the UI's live // region stops spinning on content the model won't finish emitting, // before we even await the SDK — instant feedback. @@ -1578,6 +1586,36 @@ export class Session { async #consumeEvents(run: TurnRun, _sender: AuthContext): Promise { try { for await (const event of run.events) { + // When a mid-turn message interrupts the current turn, the SDK emits a + // turn_done (often with isError) for the aborted partial turn BEFORE it + // starts the continuation turn for the injected message. Absorb that + // intermediate boundary: flush per-turn state, record partial cost, and + // continue looping — no break, no error display, no status flip to idle. + // The continuation turn's events follow immediately in the same queue. + if (event.type === "turn_done" && this.#pendingMidTurnCount > 0) { + this.#pendingMidTurnCount--; + // Record history / cost for the interrupted partial turn. + this.#accumulator.handleEvent(event); + this.#recordTurnFromResult(event.result); + // Flush per-turn accumulators so the continuation turn starts clean. + this.#completeActiveTools(); + this.#flushActiveAssistant(); + this.#finalizeActiveThinking(); + this.#chunker?.onTurnEnd(); + // Dismiss any stale approval gates from the interrupted turn. + if (this.#pendingApprovals.size > 0) { + const systemAuth: AuthContext = { sub: "system", scopes: [], delegationDepth: 0, accountId: this.accountId, projectId: this.projectId }; + for (const [aid, resolveFn] of this.#pendingApprovals.entries()) { + resolveFn({ approved: false }); + this.#dismissStaleApproval(aid, systemAuth); + } + this.#pendingApprovals.clear(); + } + // Re-assert thinking status so the UI doesn't flash idle between turns. + if (this.#status !== "error") this.#setStatus("thinking"); + continue; + } + await this.#handleProviderEvent(event); if (event.type === "turn_done" || event.type === "error") break; } @@ -1590,6 +1628,7 @@ export class Session { this.#persistAndBuffer(errorMsg); this.#broadcastRaw(errorMsg); } finally { + this.#pendingMidTurnCount = 0; // safety: reset on any exit path this.#completeActiveTools(); this.#flushActiveAssistant(); this.#finalizeActiveThinking(); diff --git a/src/tests/session-integration.test.ts b/src/tests/session-integration.test.ts index 115b2b6..a5913ab 100644 --- a/src/tests/session-integration.test.ts +++ b/src/tests/session-integration.test.ts @@ -151,6 +151,33 @@ function waitForIdle(session: Session, timeoutMs = 8000): Promise { }); } +/** + * Resolve when the session broadcasts a session.status_change with the given + * status. Checks the current status first. Rejects after timeoutMs. + */ +function waitForStatus(session: Session, targetStatus: string, timeoutMs = 4000): Promise { + if (session.status === targetStatus) return Promise.resolve(); + return new Promise((resolve, reject) => { + const watcherId = randomUUID(); + const timer = setTimeout(() => { + session.detach(watcherId); + reject(new Error(`session did not reach '${targetStatus}' within ${timeoutMs}ms — status=${session.status}`)); + }, timeoutMs); + const watcher: AttachedClient = { + id: watcherId, + auth: TEST_AUTH, + send(msg) { + if (msg.type === "session.status_change" && msg.status === targetStatus) { + clearTimeout(timer); + session.detach(watcherId); + resolve(); + } + }, + }; + session.attach(watcher); + }); +} + // ── T1: Async event ordering ────────────────────────────────────────────────── describe("T1 – async event ordering", () => { @@ -806,3 +833,126 @@ describe("T8 – regression guards: text overlap + stale status", () => { expect(finalPhase).toBe("completed"); }); }); + +// ── T9: Mid-turn message handling ───────────────────────────────────────────── + +describe("T9 – mid-turn message handling", () => { + /** + * Regression test for the bug where #consumeEvents broke on the first + * turn_done (the interrupted partial turn) and left the continuation turn + * with no consumer, causing the session to stop dead with an error instead + * of processing the injected message. + * + * The test drives a custom TurnRun directly: + * 1. Emit text_delta so the session enters "thinking" status. + * 2. Wait for the mid-turn send() to arrive (via pushMidTurn). + * 3. Emit turn_done(isError, "conversation ended mid-turn") — the aborted turn. + * 4. Emit the continuation turn (text_delta + text_done + turn_done(success)). + */ + it("mid-turn message is consumed as a continuation turn, not an error stop", async () => { + const queue = new AsyncQueue(); + let notifyMidTurnReceived: (() => void) | null = null; + + // Custom SessionProvider whose TurnRun supports pushMidTurn. + const provider: import("../daemon/providers/interface.js").SessionProvider = { + id: "claude", + displayName: "MidTurnTest", + onRecoveryNeeded: undefined, + backingSessionId: "mid-turn-test-backing", + hasQueried: false, + queuedMessages: 0, + resetToNewSession() {}, + setHasQueried(_v: boolean) { }, + async teardown() { queue.close(); }, + async dispose() { queue.close(); }, + async listModels() { return []; }, + + runTurn(_opts: import("../daemon/providers/interface.js").TurnOpts): TurnRun { + void (async () => { + await Promise.resolve(); // yield so #consumeEvents loop has started + // First turn: emit a partial text_delta so the session enters "thinking", + // then stall until the mid-turn message arrives via pushMidTurn. + queue.push({ type: "text_delta", content: "Working on original task..." }); + await new Promise((r) => { notifyMidTurnReceived = r; }); + // Simulate SDK interrupting the turn when a now-priority message lands. + queue.push({ + type: "turn_done", + result: mockResult({ providerId: "claude", isError: true, errorMessage: "conversation ended mid-turn" }), + }); + // Continuation turn for the injected mid-turn message. + queue.push({ type: "text_delta", content: "Continuing with your request." }); + queue.push({ type: "text_done", content: "Continuing with your request." }); + queue.push({ type: "turn_done", result: mockResult({ providerId: "claude" }) }); + queue.close(); + })(); + + return { + events: queue, + interrupt: async () => { queue.close(); }, + pushMidTurn: (_content: string, _priority: string) => { + notifyMidTurnReceived?.(); + }, + }; + }, + }; + + // makeSession() requires MockSessionProvider; construct Session directly. + const id = randomUUID(); + store.createSession({ + id, + name: "mid-turn-integ", + workdir: tmp, + status: "idle", + createdBy: TEST_AUTH.sub, + createdAt: new Date().toISOString(), + attachedClients: 0, + accountId: TEST_AUTH.accountId!, + projectId: TEST_AUTH.projectId!, + }); + const { Session: SessionCtor } = await import("../daemon/session.js"); + const session = new SessionCtor({ + name: "mid-turn-integ", + workdir: tmp, + auth: TEST_AUTH, + store, + transcriptStore, + existingId: id, + _testProvider: provider as unknown as import("../daemon/providers/mock/session-provider.js").MockSessionProvider, + }); + + const { client, received } = makeClient(); + session.attach(client); + + // Start the first turn. + await session.send("original task", TEST_AUTH); + + // Wait until the session is "thinking" before sending the mid-turn message. + await waitForStatus(session, "thinking"); + + // Send mid-turn message — triggers pushMidTurn on the active run. + await session.send("mid-turn add something", TEST_AUTH); + + // Session must reach idle (not error) after both turns complete. + await waitForIdle(session); + + expect(session.status).toBe("idle"); + + // No "Error: conversation ended mid-turn" should appear in the scrollback. + const errorMsgs = received.filter( + (m) => + m.type === "session.message" && + (m as { role?: string }).role === "system" && + typeof (m as { content?: unknown }).content === "string" && + ((m as { content: string }).content).startsWith("Error:"), + ); + expect(errorMsgs).toHaveLength(0); + + // The continuation assistant message from the mid-turn response must exist. + const assistantMsgs = received.filter( + (m) => m.type === "session.message" && (m as { role?: string }).role === "assistant", + ); + expect(assistantMsgs.length).toBeGreaterThan(0); + const lastAssistant = assistantMsgs.at(-1) as { content?: string }; + expect(lastAssistant?.content).toContain("Continuing"); + }); +}); From f792a915ddbfc9fcd7724f5adb5d7a58cc6a1e51 Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Mon, 29 Jun 2026 02:59:09 +0200 Subject: [PATCH 3/7] fix: recover wedged turns when provider event stream stalls (#46) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A session could get stuck in "replying" forever with no messages shown (both Telegram and web UI) when a provider's event stream went silent without a terminal turn_done — e.g. a Bash command or MCP gateway call that never returns. #consumeEvents blocked in `for await (run.events)` with no watchdog, so #activeRun/status never reset; every subsequent send was then swallowed into the dead run via the wasWorking mid-turn push path, and the user never got a reply. This is a latent bug present on main as well (not introduced by the multi-provider work) — but high severity, since any single hung tool bricks the session until an explicit interrupt. Fix: - Stall watchdog in #consumeEvents: drive the iterator manually and race each pull against session.turnStallTimeoutMs (default 300s, 0 = off). On silence, force-recover via #recoverStalledRun. - #recoverStalledRun: flush streaming UI, release pending approvals, drop the run, emit a clear "Turn timed out" notice, reset to idle, and hard teardown the provider to reap the (presumed hung) subprocess. Does NOT route through #teardownProvider (which awaits the session consumer — i.e. potentially itself), avoiding self-await deadlock. - Liveness guard in #sendInner: a send onto an apparently-dead run recovers it and starts a fresh turn instead of silently queueing. - Config: session.turnStallTimeoutMs + CODEOID_TURN_STALL_TIMEOUT_MS. - MockSessionProvider: opt-in `stall` mode + close-on-teardown (mirrors ClaudeProvider) for deterministic offline testing. - Tests (T9): watchdog recovers + reaps; next send works (no permanent wedge); watchdog-off leaves the turn blocked (no false recovery). Verified: typecheck clean, biome clean, 617/617 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/config.ts | 15 +- src/daemon/providers/mock/session-provider.ts | 20 +- src/daemon/session.ts | 179 +++++++++++++++++- src/tests/session-integration.test.ts | 124 +++++++++++- 4 files changed, 331 insertions(+), 7 deletions(-) diff --git a/src/config.ts b/src/config.ts index 18b2fd6..ea68457 100644 --- a/src/config.ts +++ b/src/config.ts @@ -215,8 +215,18 @@ const SessionSchema = z .object({ defaultModel: z.string().optional(), fallbackModel: z.string().optional(), + /** + * Hard backstop against a wedged turn. If the provider event stream goes + * completely silent (no events at all) for this many ms while a turn is + * active, the turn is treated as stalled: the run is torn down, the + * subprocess reaped, status reset to idle, and a clear message shown. + * Generous by default — long-running tools still emit `tool_progress` / + * partial events, so true silence for this long is a reliable hang signal. + * Set to 0 to disable the watchdog. + */ + turnStallTimeoutMs: z.number().min(0).default(300_000), }) - .default({}); + .default({ turnStallTimeoutMs: 300_000 }); const AutoRotateSchema = z .object({ @@ -367,6 +377,8 @@ export interface CodeoidConfig { session: { defaultModel?: string; fallbackModel?: string; + /** Stall watchdog: ms of total event-stream silence before a turn is force-recovered (0 = off). Defaults to 300000 when omitted. */ + turnStallTimeoutMs?: number; }; } @@ -420,6 +432,7 @@ const ENV_OVERRIDES: readonly EnvOverride[] = [ { env: "CODEOID_AUTO_ROTATE_MIN_TURNS", path: "autoRotate.minTurnsBeforeRotate", kind: "int" }, { env: "CODEOID_DEFAULT_MODEL", path: "session.defaultModel", kind: "string" }, { env: "CODEOID_FALLBACK_MODEL", path: "session.fallbackModel", kind: "string" }, + { env: "CODEOID_TURN_STALL_TIMEOUT_MS", path: "session.turnStallTimeoutMs", kind: "int" }, ]; // ── Loading ────────────────────────────────────────────────────────────── diff --git a/src/daemon/providers/mock/session-provider.ts b/src/daemon/providers/mock/session-provider.ts index 283c0af..7a4d225 100644 --- a/src/daemon/providers/mock/session-provider.ts +++ b/src/daemon/providers/mock/session-provider.ts @@ -44,6 +44,13 @@ export class MockSessionProvider implements SessionProvider { #backingSessionId: string; #hasQueried = false; #script: ProviderEvent[][]; + /** When true, runTurn() emits its scripted events then leaves the queue OPEN + * (never closes, never emits a terminal turn_done) — simulating a provider + * whose stream has gone silent (hung tool / dead subprocess). The queue is + * only closed by teardown(), mirroring ClaudeProvider. */ + #stall: boolean; + /** Live turn queue, so teardown() can unblock a waiting consumer like the real provider. */ + #currentQueue: AsyncQueue | null = null; /** Every TurnOpts passed to runTurn() — inspect in tests. */ readonly capturedOpts: TurnOpts[] = []; @@ -51,11 +58,12 @@ export class MockSessionProvider implements SessionProvider { /** Incremented each time teardown() is called — useful for asserting cleanup. */ teardownCount = 0; - constructor(id = "mock-session", script: ProviderEvent[][] = []) { + constructor(id = "mock-session", script: ProviderEvent[][] = [], opts: { stall?: boolean } = {}) { this.id = id; this.displayName = `MockSession(${id})`; this.#backingSessionId = `${id}-backing`; this.#script = script.map((s) => [...s]); + this.#stall = opts.stall ?? false; } get backingSessionId(): string { return this.#backingSessionId; } @@ -74,6 +82,10 @@ export class MockSessionProvider implements SessionProvider { async teardown(): Promise { this.teardownCount++; this.onRecoveryNeeded = undefined; + // Mirror ClaudeProvider: closing the live turn queue unblocks any consumer + // currently awaiting the next event (e.g. a stalled run being recovered). + this.#currentQueue?.close(); + this.#currentQueue = null; } async dispose(): Promise { @@ -93,6 +105,7 @@ export class MockSessionProvider implements SessionProvider { ]; const queue = new AsyncQueue(); + this.#currentQueue = queue; // Emit events asynchronously, calling canUseTool for each tool_start // to simulate the SDK's PreToolUse hook firing before the tool runs. @@ -135,6 +148,11 @@ export class MockSessionProvider implements SessionProvider { } } + // Stall mode: emit the scripted events, then leave the queue OPEN (no + // terminal event, no close) so the consumer's next pull blocks — exactly + // what a hung provider stream looks like. Only teardown() closes it. + if (this.#stall) return; + queue.close(); } } diff --git a/src/daemon/session.ts b/src/daemon/session.ts index ab4c242..43d085c 100644 --- a/src/daemon/session.ts +++ b/src/daemon/session.ts @@ -140,6 +140,11 @@ export class Session { #provider!: SessionProvider; #activeRun: TurnRun | null = null; #eventConsumerTask: Promise | null = null; + // Wall-clock ms of the most recent provider event for the active run. The + // stall watchdog in #consumeEvents and the liveness guard in #sendInner read + // this to detect a turn whose event stream has gone silent (hung tool / dead + // subprocess) so the session can self-recover instead of wedging forever. + #lastEventAt = 0; #accumulator = new CanonicalHistoryAccumulator(); // Tracks the sender of the most recently started turn. The onRecoveryNeeded // closure reads this instead of closing over the original send()'s sender, @@ -686,6 +691,27 @@ export class Session { const effectivePriority: "now" | "next" | "later" = priority ?? (wasWorking ? "now" : "later"); + // Liveness guard: a session can look "working" while its run is actually + // wedged (provider stream went silent on a hung tool / dead subprocess). + // Queuing a mid-turn push into a dead run silently swallows the message and + // the user never gets a reply. If the active run has produced no event for + // longer than the stall window, recover it now and start a fresh turn + // instead of trusting #activeRun. + const stallMs = this.#config?.session.turnStallTimeoutMs ?? 300_000; + if ( + wasWorking && + this.#activeRun && + stallMs > 0 && + this.#lastEventAt > 0 && + Date.now() - this.#lastEventAt > stallMs + ) { + console.error( + `[codeoid/session ${this.id}] send arrived on a stalled run (${Date.now() - this.#lastEventAt}ms since last event); recovering before starting a fresh turn`, + ); + await this.#recoverStalledRun(this.#activeRun, stallMs); + // Fall through to the normal (idle) turn-start path below. + } + // For keep-warm mid-turn pushes (ClaudeProvider), inject directly into the live run. if (wasWorking && this.#activeRun?.pushMidTurn) { const hint = @@ -1584,8 +1610,39 @@ export class Session { } async #consumeEvents(run: TurnRun, _sender: AuthContext): Promise { + // Stall watchdog: drive the iterator manually so we can race each pull + // against a timeout. If the provider stream goes completely silent for + // longer than the configured window (no events at all — long-running tools + // still emit tool_progress / partial events), the turn is treated as + // wedged and force-recovered. 0 disables the watchdog. See #recoverStalledRun. + const stallMs = this.#config?.session.turnStallTimeoutMs ?? 300_000; + const iter = run.events[Symbol.asyncIterator](); + const STALL = Symbol("stall"); + this.#lastEventAt = Date.now(); try { - for await (const event of run.events) { + while (true) { + let next: IteratorResult | typeof STALL; + if (stallMs > 0) { + let timer: ReturnType | undefined; + const stall = new Promise((resolve) => { + timer = setTimeout(() => resolve(STALL), stallMs); + }); + try { + next = await Promise.race([iter.next(), stall]); + } finally { + if (timer) clearTimeout(timer); + } + } else { + next = await iter.next(); + } + + if (next === STALL) { + await this.#recoverStalledRun(run, stallMs); + return; // finally still runs; #recoverStalledRun already cleared run state + } + if (next.done) break; + const event = next.value; + this.#lastEventAt = Date.now(); // When a mid-turn message interrupts the current turn, the SDK emits a // turn_done (often with isError) for the aborted partial turn BEFORE it // starts the continuation turn for the injected message. Absorb that @@ -1652,6 +1709,71 @@ export class Session { } } + /** + * Force-recover a wedged turn whose provider event stream went silent. + * + * Called by the #consumeEvents watchdog (after `stallMs` of no events) and by + * the #sendInner liveness guard (when a send arrives on an apparently-dead + * run). Idempotent and run-scoped: if `run` is no longer the active run, the + * turn already ended and we no-op. + * + * Crucially this does NOT route through #teardownProvider (which awaits the + * session event-consumer task — i.e. potentially itself). It nulls the run + * slots up front, surfaces a clear message, resets status to idle, and hard + * tears down the PROVIDER (abort → reap the hung subprocess). The next send() + * recreates a fresh query loop. The abandoned consumer (if any) unblocks when + * teardown closes the turn queue and its finally no-ops via the run guard. + */ + async #recoverStalledRun(run: TurnRun, stallMs: number): Promise { + if (this.#activeRun !== run) return; // already ended / recovered + + console.error( + `[codeoid/session ${this.id}] turn stalled — no provider events for ${stallMs}ms; force-recovering`, + ); + + // Stop any spinning UI and release waiters BEFORE we tear down. + this.#completeActiveTools(); + this.#flushActiveAssistant(); + this.#finalizeActiveThinking(); + this.#chunker?.onTurnEnd(); + this.#pendingMidTurnCount = 0; + if (this.#pendingApprovals.size > 0) { + const systemAuth: AuthContext = { sub: "system", scopes: [], delegationDepth: 0, accountId: this.accountId, projectId: this.projectId }; + for (const [aid, resolveFn] of this.#pendingApprovals.entries()) { + resolveFn({ approved: false }); + this.#dismissStaleApproval(aid, systemAuth); + } + this.#pendingApprovals.clear(); + } + + // Drop the wedged run so a concurrent send() doesn't queue into it. + this.#activeRun = null; + this.#eventConsumerTask = null; + + const msg = this.#makeMessage( + "system", + `⚠️ Turn timed out — no activity for ${Math.round(stallMs / 1000)}s. The session was reset; send your message again to continue.`, + SYSTEM_IDENTITY, + undefined, + undefined, + { event: "turn_stalled", errorCode: "turn_stalled" }, + ); + this.#persistAndBuffer(msg); + this.#broadcastRaw(msg); + this.#setStatus("idle"); + + // Hard teardown: abort the controller and reap the (presumed hung) + // subprocess. Safe to await here — provider.teardown() awaits the + // PROVIDER's own pump, never this session's consumer. + try { + await this.#provider.teardown(); + } catch (e) { + console.error( + `[codeoid/session ${this.id}] provider teardown during stall recovery failed: ${e instanceof Error ? e.message : String(e)}`, + ); + } + } + async #handleProviderEvent(event: ProviderEvent): Promise { switch (event.type) { case "text_delta": { @@ -1687,9 +1809,10 @@ export class Session { this.#broadcastRaw(this.#activeAssistantMsg); this.#activeAssistantMsg = null; } else if (event.content) { - const msg = this.#makeMessage("assistant", event.content, this.#agentIdentity, [{ kind: "text", text: event.content, markdown: true }]); - this.#persistAndBuffer(msg); - this.#broadcastRaw(msg); + // No preceding text_delta — the SDK returned the response as a batch + // (happens for mid-turn now-priority continuations). Simulate streaming + // so the UI shows a typing animation instead of an instant text pop-in. + await this.#artificiallyStreamText(event.content); } break; } @@ -1956,6 +2079,54 @@ export class Session { this.#broadcastRaw(m); } + /** + * Emit text as artificial streaming deltas so the UI shows a typing animation + * for responses the SDK returned as a single batch (no preceding text_delta). + * This happens for mid-turn now-priority continuations where the SDK skips + * streaming and emits only an `assistant` message. + * + * Scales step size to yield ~30 animation frames at 16ms each (~480ms total), + * so the animation looks natural across any response length without adding + * meaningful latency. + * + * Interrupt safety: each loop iteration checks whether #activeAssistantMsg + * still points to the message we created; if interrupt() ran between frames + * (#flushActiveAssistant nulled it), we return early — the partial content was + * already committed by the flush. + */ + async #artificiallyStreamText(content: string): Promise { + const FRAME_MS = 16; + const steps = Math.min(30, content.length); + const charsPerStep = Math.ceil(content.length / steps); + + const msg = this.#makeMessage("assistant", "", this.#agentIdentity, []); + this.#activeAssistantMsg = msg; + this.#persistAndBuffer(msg); + this.#broadcastRaw(msg); + + for (let pos = 0; pos < content.length; pos += charsPerStep) { + if (this.#activeAssistantMsg !== msg) return; // interrupted between frames + const chunk = content.slice(pos, pos + charsPerStep); + msg.content += chunk; + const delta: SessionMessageDelta = { + type: "session.message.delta", + sessionId: this.id, + messageId: msg.messageId, + contentAppend: chunk, + timestamp: new Date().toISOString(), + }; + this.#broadcastRaw(delta); + await new Promise((r) => setTimeout(r, FRAME_MS)); + } + + if (this.#activeAssistantMsg !== msg) return; // interrupted on last frame + msg.content = content; // exact match regardless of ceiling-division rounding + msg.parts = [{ kind: "text", text: content, markdown: true }]; + this.#persistAndBuffer(msg); + this.#broadcastRaw(msg); + this.#activeAssistantMsg = null; + } + /** * Finalize the active thinking stream. Same shape as * #flushActiveAssistant — always rebroadcast with non-empty content so diff --git a/src/tests/session-integration.test.ts b/src/tests/session-integration.test.ts index a5913ab..ab11a97 100644 --- a/src/tests/session-integration.test.ts +++ b/src/tests/session-integration.test.ts @@ -45,6 +45,7 @@ import { mockResult } from "../daemon/providers/mock/index.js"; import { AsyncQueue } from "../daemon/async-queue.js"; import type { DaemonMessage, AuthContext } from "../protocol/types.js"; import type { ProviderEvent, TurnRun } from "../daemon/providers/interface.js"; +import type { CodeoidConfig } from "../config.js"; // ── Fixtures ────────────────────────────────────────────────────────────────── @@ -85,7 +86,11 @@ afterEach(async () => { * Uses existingId so the constructor skips the async saveMeta call, * eliminating ENOENT races with afterEach's rmSync. */ -function makeSession(provider: MockSessionProvider, name = "integ-test"): Session { +function makeSession( + provider: MockSessionProvider, + name = "integ-test", + config?: CodeoidConfig, +): Session { const id = randomUUID(); store.createSession({ id, @@ -106,9 +111,27 @@ function makeSession(provider: MockSessionProvider, name = "integ-test"): Sessio transcriptStore, existingId: id, _testProvider: provider, + config, }); } +/** Minimal config whose only meaningful field is a tiny stall timeout, so the + * watchdog fires in milliseconds instead of the 300s default. autoRotate is + * included (disabled) because #shouldRotate dereferences it on every send. */ +function stallConfig(turnStallTimeoutMs: number): CodeoidConfig { + return { + session: { turnStallTimeoutMs }, + autoRotate: { + enabled: false, + warnPct: 0.75, + rotatePct: 0.9, + hardRotatePct: 0.95, + minTurnsBeforeRotate: 1, + strategy: "task-anchor", + }, + } as unknown as CodeoidConfig; +} + /** Build a stub AttachedClient that records every DaemonMessage it receives. */ function makeClient(id = randomUUID()): { client: AttachedClient; received: DaemonMessage[] } { const received: DaemonMessage[] = []; @@ -956,3 +979,102 @@ describe("T9 – mid-turn message handling", () => { expect(lastAssistant?.content).toContain("Continuing"); }); }); + +// ── T9: Stall watchdog — wedged turn self-recovers ────────────────────────────── +// +// Regression guard for the "stuck in replying, no messages" wedge (#46): when a +// provider's event stream goes silent without a terminal turn_done (hung tool / +// dead subprocess), the session must NOT block forever. The watchdog force- +// recovers the turn, reaps the provider, and a subsequent send works normally. +describe("T9 – stall watchdog recovers a wedged turn", () => { + it("a turn whose stream goes silent recovers to idle, emits a timeout notice, and reaps the provider", async () => { + // stall:true → MockSessionProvider emits the delta then leaves the queue + // open forever (no turn_done, no close), exactly like a hung stream. + const provider = new MockSessionProvider( + "claude", + [[{ type: "text_delta", content: "working on it" }]], + { stall: true }, + ); + const session = makeSession(provider, "stall-watchdog", stallConfig(80)); + const { client, received } = makeClient(); + session.attach(client); + + await session.send("do the thing", TEST_AUTH); + expect(session.status).toBe("thinking"); + + // Watchdog fires at 80ms → recovery. Allow generous slack for CI. + await waitForIdle(session, 4000); + expect(session.status).toBe("idle"); + + // A clear timeout breadcrumb was surfaced (not a silent wedge). + const stalledMsg = received.find( + (m) => + m.type === "session.message" && + (m as { role?: string }).role === "system" && + /timed out/i.test((m as { content?: string }).content ?? ""), + ); + expect(stalledMsg).toBeTruthy(); + + // The presumed-hung subprocess was reaped (provider torn down). + expect(provider.teardownCount).toBeGreaterThanOrEqual(1); + }); + + it("after a stall recovery, the next send starts a fresh turn and gets a reply (no permanent wedge)", async () => { + // First turn stalls; second turn is a normal scripted reply. + const provider = new MockSessionProvider( + "claude", + [ + [{ type: "text_delta", content: "hang…" }], // stalls (queue left open) + [ + { type: "text_done", content: "Recovered and replied." }, + { type: "turn_done", result: mockResult({ providerId: "claude" }) }, + ], + ], + { stall: true }, + ); + const session = makeSession(provider, "stall-then-send", stallConfig(80)); + const { client, received } = makeClient(); + session.attach(client); + + // Turn 1 wedges → watchdog recovers it. + await session.send("first", TEST_AUTH); + await waitForIdle(session, 4000); + expect(session.status).toBe("idle"); + + // Turn 2 must behave like a normal turn (not get swallowed into the dead run). + await session.send("second", TEST_AUTH); + await waitForIdle(session, 4000); + + const assistantMsgs = received.filter( + (m) => m.type === "session.message" && (m as { role?: string }).role === "assistant", + ); + const last = assistantMsgs.at(-1) as { content?: string }; + expect(last?.content).toBe("Recovered and replied."); + // Two runTurn() calls: the stalled one + the recovered one. + expect(provider.capturedOpts.length).toBeGreaterThanOrEqual(2); + }); + + it("watchdog disabled (turnStallTimeoutMs=0) leaves a silent turn blocked (no false recovery)", async () => { + const provider = new MockSessionProvider( + "claude", + [[{ type: "text_delta", content: "indefinite" }]], + { stall: true }, + ); + const session = makeSession(provider, "stall-disabled", stallConfig(0)); + const { received } = makeClient(); + + await session.send("go", TEST_AUTH); + expect(session.status).toBe("thinking"); + + // With the watchdog off, the turn stays "thinking" — give it room to (not) recover. + await new Promise((r) => setTimeout(r, 300)); + expect(session.status).toBe("thinking"); + const stalledMsg = received.find( + (m) => m.type === "session.message" && /timed out/i.test((m as { content?: string }).content ?? ""), + ); + expect(stalledMsg).toBeUndefined(); + + // Clean up the still-open run so afterEach doesn't race a live consumer. + await session.interrupt(TEST_AUTH); + }); +}); From 808b4891a7da8706cf2ed67c698df9708f2d790d Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Mon, 29 Jun 2026 04:52:09 +0200 Subject: [PATCH 4/7] fix: address CodeRabbit review on stall watchdog MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four findings from CodeRabbit on PR #47, all valid: - config: env overrides are applied after RootSchema.safeParse, so CODEOID_TURN_STALL_TIMEOUT_MS=-1 bypassed z.number().min(0) and silently disabled the watchdog (stallMs > 0 → false). Re-validate the merged config through RootSchema after the override loop — also guards the autoRotate percentage bounds and every other constrained override. - watchdog: a pending manual tool approval is a legitimate indefinite silent period (provider blocks on canUseTool). Pause the stall race while status is waiting_approval / #pendingApprovals is non-empty so a slow human approval isn't mistaken for a hung stream. - consumer: add a run-ownership check before processing an event, so a late event from a run abandoned by #sendInner's liveness-guard recovery can't leak into the fresh turn. - message: when #recoverStalledRun is invoked from #sendInner the saved message is retried in a fresh turn — say so, instead of telling the user to resend (which would duplicate). Tests: +2 config (override applies; negative rejected) and +1 integration (watchdog stays paused through a pending approval). 620/620 pass, biome clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/config.ts | 17 ++++++++++++ src/daemon/session.ts | 26 +++++++++++++++--- src/tests/config.test.ts | 20 ++++++++++++++ src/tests/session-integration.test.ts | 38 +++++++++++++++++++++++++++ 4 files changed, 97 insertions(+), 4 deletions(-) diff --git a/src/config.ts b/src/config.ts index ea68457..a177282 100644 --- a/src/config.ts +++ b/src/config.ts @@ -494,6 +494,23 @@ export function loadConfig(opts: LoadOptions = {}): CodeoidConfig { setByPath(parsed, ov.path, parseOverride(raw, ov.kind)); } + // 3a. Re-validate after overrides. parseOverride() coerces strings to the + // declared kind but does NOT enforce schema constraints (e.g. the + // non-negative bound on session.turnStallTimeoutMs, or the 0..1 bounds on + // the autoRotate percentages). Without this, CODEOID_TURN_STALL_TIMEOUT_MS=-1 + // would slip through and silently disable the stall watchdog. Re-running + // RootSchema over the merged result fails fast on any out-of-range override. + const revalidated = RootSchema.safeParse(parsed); + if (!revalidated.success) { + const issues = revalidated.error.issues + .map((i) => ` ${i.path.join(".")}: ${i.message}`) + .join("\n"); + throw new Error( + `Invalid config after applying environment overrides:\n${issues}\n(Check the corresponding CODEOID_* env vars.)`, + ); + } + Object.assign(parsed, revalidated.data); + // 3b. Resolve the ZeroID issuer (preset name or URL → concrete base URL) and // pin the expected issuer claim. Every ZeroID deployment sets `iss` to // its base URL, so defaulting auth.issuer to the resolved URL rejects diff --git a/src/daemon/session.ts b/src/daemon/session.ts index 43d085c..96f0a88 100644 --- a/src/daemon/session.ts +++ b/src/daemon/session.ts @@ -708,7 +708,9 @@ export class Session { console.error( `[codeoid/session ${this.id}] send arrived on a stalled run (${Date.now() - this.#lastEventAt}ms since last event); recovering before starting a fresh turn`, ); - await this.#recoverStalledRun(this.#activeRun, stallMs); + await this.#recoverStalledRun(this.#activeRun, stallMs, { + continuingCurrentSend: true, + }); // Fall through to the normal (idle) turn-start path below. } @@ -1622,7 +1624,13 @@ export class Session { try { while (true) { let next: IteratorResult | typeof STALL; - if (stallMs > 0) { + // A pending manual tool approval is a legitimate indefinite silent + // period — the provider blocks on canUseTool until the user responds. + // Pause the watchdog so a slow human approval isn't mistaken for a hung + // stream and cancelled. + const waitingForApproval = + this.#status === "waiting_approval" || this.#pendingApprovals.size > 0; + if (stallMs > 0 && !waitingForApproval) { let timer: ReturnType | undefined; const stall = new Promise((resolve) => { timer = setTimeout(() => resolve(STALL), stallMs); @@ -1641,6 +1649,10 @@ export class Session { return; // finally still runs; #recoverStalledRun already cleared run state } if (next.done) break; + // Ownership guard: a concurrent #sendInner liveness-guard recovery may + // have torn down this run while we were awaiting iter.next(). Drop any + // late event from the abandoned run so it can't leak into the fresh turn. + if (this.#activeRun !== run) break; const event = next.value; this.#lastEventAt = Date.now(); // When a mid-turn message interrupts the current turn, the SDK emits a @@ -1724,7 +1736,11 @@ export class Session { * recreates a fresh query loop. The abandoned consumer (if any) unblocks when * teardown closes the turn queue and its finally no-ops via the run guard. */ - async #recoverStalledRun(run: TurnRun, stallMs: number): Promise { + async #recoverStalledRun( + run: TurnRun, + stallMs: number, + opts?: { continuingCurrentSend?: boolean }, + ): Promise { if (this.#activeRun !== run) return; // already ended / recovered console.error( @@ -1752,7 +1768,9 @@ export class Session { const msg = this.#makeMessage( "system", - `⚠️ Turn timed out — no activity for ${Math.round(stallMs / 1000)}s. The session was reset; send your message again to continue.`, + opts?.continuingCurrentSend + ? `⚠️ Previous turn timed out — no activity for ${Math.round(stallMs / 1000)}s. The session was reset and your latest message is being retried in a fresh turn.` + : `⚠️ Turn timed out — no activity for ${Math.round(stallMs / 1000)}s. The session was reset; send your message again to continue.`, SYSTEM_IDENTITY, undefined, undefined, diff --git a/src/tests/config.test.ts b/src/tests/config.test.ts index 6183510..75bc58c 100644 --- a/src/tests/config.test.ts +++ b/src/tests/config.test.ts @@ -137,6 +137,26 @@ describe("loadConfig — env precedence", () => { }), ).toThrow(/Expected integer/); }); + + it("CODEOID_TURN_STALL_TIMEOUT_MS overrides the stall watchdog timeout", () => { + const c = loadConfig({ + configPath, + env: { CODEOID_TURN_STALL_TIMEOUT_MS: "120000" }, + }); + expect(c.session.turnStallTimeoutMs).toBe(120000); + }); + + it("rejects a negative stall timeout (env override is re-validated, not just coerced)", () => { + // Regression: env overrides are applied AFTER the initial safeParse, so a + // negative value would otherwise bypass z.number().min(0) and silently + // disable the stall watchdog (stallMs > 0 guard reads false). + expect(() => + loadConfig({ + configPath, + env: { CODEOID_TURN_STALL_TIMEOUT_MS: "-1" }, + }), + ).toThrow(/environment overrides|turnStallTimeoutMs/); + }); }); describe("loadConfig — path resolution", () => { diff --git a/src/tests/session-integration.test.ts b/src/tests/session-integration.test.ts index ab11a97..e94e5db 100644 --- a/src/tests/session-integration.test.ts +++ b/src/tests/session-integration.test.ts @@ -1077,4 +1077,42 @@ describe("T9 – stall watchdog recovers a wedged turn", () => { // Clean up the still-open run so afterEach doesn't race a live consumer. await session.interrupt(TEST_AUTH); }); + + it("does NOT fire while waiting for a manual tool approval (legitimate silent period)", async () => { + // Bash is a mutation tool — in guarded (default) mode it requires approval. + // The mock blocks in canUseTool until approve(), leaving the stream silent. + // The watchdog must pause during that wait, not cancel the approval prompt. + const provider = new MockSessionProvider( + "claude", + [ + [ + { + type: "tool_start", + toolId: "bash-stall", + sdkToolUseId: "sdk-bash-stall", + name: "Bash", + input: { command: "sleep 999" }, + approvalId: "ap-stall-1", + }, + ], + ], + { stall: true }, + ); + const session = makeSession(provider, "stall-approval", stallConfig(80)); + const { received } = makeClient(); + + await session.send("run it", TEST_AUTH); + await waitForStatus(session, "waiting_approval", 4000); + + // Wait well past the 80ms stall window — the watchdog must stay paused. + await new Promise((r) => setTimeout(r, 300)); + expect(session.status).toBe("waiting_approval"); + const stalledMsg = received.find( + (m) => m.type === "session.message" && /timed out/i.test((m as { content?: string }).content ?? ""), + ); + expect(stalledMsg).toBeUndefined(); + + // Clean up the pending approval + open run. + await session.interrupt(TEST_AUTH); + }); }); From a260f6c318d4dc244b6d7e99ac00684d349e7506 Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Mon, 29 Jun 2026 05:25:16 +0200 Subject: [PATCH 5/7] fix: address 2 more CodeRabbit comments on test determinism MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - waitForStatus: close the race where the status flips to the target between the initial check and session.attach() — re-check current status after attaching so the helper can't miss the only broadcast and time out. - Attach the test client in the watchdog-disabled and approval-pause tests. `received` was never populated (client created but not attached), so the "no timeout notice" assertions passed vacuously even if the disabled / paused watchdog path had wrongly emitted one. 620/620 pass, biome clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/tests/session-integration.test.ts | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/src/tests/session-integration.test.ts b/src/tests/session-integration.test.ts index e94e5db..b4e5aca 100644 --- a/src/tests/session-integration.test.ts +++ b/src/tests/session-integration.test.ts @@ -198,6 +198,14 @@ function waitForStatus(session: Session, targetStatus: string, timeoutMs = 4000) }, }; session.attach(watcher); + // Re-check after attaching: if the status flipped to the target in the + // window between the initial check and attach(), we'd otherwise miss the + // only broadcast and time out. + if (session.status === targetStatus) { + clearTimeout(timer); + session.detach(watcherId); + resolve(); + } }); } @@ -1061,7 +1069,8 @@ describe("T9 – stall watchdog recovers a wedged turn", () => { { stall: true }, ); const session = makeSession(provider, "stall-disabled", stallConfig(0)); - const { received } = makeClient(); + const { client, received } = makeClient(); + session.attach(client); await session.send("go", TEST_AUTH); expect(session.status).toBe("thinking"); @@ -1099,7 +1108,8 @@ describe("T9 – stall watchdog recovers a wedged turn", () => { { stall: true }, ); const session = makeSession(provider, "stall-approval", stallConfig(80)); - const { received } = makeClient(); + const { client, received } = makeClient(); + session.attach(client); await session.send("run it", TEST_AUTH); await waitForStatus(session, "waiting_approval", 4000); From 2c59b2e535c1019141e9aca416da4be82c671614 Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Mon, 29 Jun 2026 09:01:02 +0200 Subject: [PATCH 6/7] fix: bound hung MCP tool calls with an SDK per-server timeout Make the SDK signal a hung MCP call instead of relying solely on the session stall watchdog. The watchdog is a coarse, provider-agnostic last-resort; the precise fix for the case we actually hit (an unresponsive MCP gateway) is the SDK's own per-server tool-call timeout. - Apply session.mcpToolTimeoutMs (default 120000) to external/user MCP servers via each server's SDK `timeout`. A hung call now returns an SDK error event the turn loop acts on, rather than going silent. Explicit per-server timeouts are preserved; the in-process memory server is left untouched; 0 disables (use SDK default). - Default (120s) sits BELOW turnStallTimeoutMs (300s) so the SDK fires first and the watchdog stays a true backstop. - Config: session.mcpToolTimeoutMs + CODEOID_MCP_TOOL_TIMEOUT_MS. - Tests: withMcpToolTimeout (inject / don't-override / no-op) + config (default ordering + env override). 625/625 pass, biome clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/config.ts | 15 ++++++++++++- src/daemon/providers/claude/index.ts | 31 ++++++++++++++++++++++++++- src/tests/config.test.ts | 14 ++++++++++++ src/tests/provider-claude.test.ts | 32 ++++++++++++++++++++++++++++ 4 files changed, 90 insertions(+), 2 deletions(-) diff --git a/src/config.ts b/src/config.ts index a177282..f9d4a43 100644 --- a/src/config.ts +++ b/src/config.ts @@ -225,8 +225,18 @@ const SessionSchema = z * Set to 0 to disable the watchdog. */ turnStallTimeoutMs: z.number().min(0).default(300_000), + /** + * Per-call wall-clock timeout (ms) applied to external (user-configured) + * MCP servers, surfaced to the SDK as each server's `timeout`. A hung MCP + * tool call (e.g. an unresponsive HTTP gateway) then returns an SDK error + * the turn loop can act on, instead of going silent. Kept BELOW + * turnStallTimeoutMs so it fires first — the stall watchdog stays a coarse + * last-resort backstop. 0 = don't set (use the SDK default). Does not apply + * to codeoid's in-process memory server. + */ + mcpToolTimeoutMs: z.number().min(0).default(120_000), }) - .default({ turnStallTimeoutMs: 300_000 }); + .default({ turnStallTimeoutMs: 300_000, mcpToolTimeoutMs: 120_000 }); const AutoRotateSchema = z .object({ @@ -379,6 +389,8 @@ export interface CodeoidConfig { fallbackModel?: string; /** Stall watchdog: ms of total event-stream silence before a turn is force-recovered (0 = off). Defaults to 300000 when omitted. */ turnStallTimeoutMs?: number; + /** Per-call timeout (ms) for external MCP servers, surfaced as the SDK's per-server `timeout`. 0 = use SDK default. Defaults to 120000 when omitted. */ + mcpToolTimeoutMs?: number; }; } @@ -433,6 +445,7 @@ const ENV_OVERRIDES: readonly EnvOverride[] = [ { env: "CODEOID_DEFAULT_MODEL", path: "session.defaultModel", kind: "string" }, { env: "CODEOID_FALLBACK_MODEL", path: "session.fallbackModel", kind: "string" }, { env: "CODEOID_TURN_STALL_TIMEOUT_MS", path: "session.turnStallTimeoutMs", kind: "int" }, + { env: "CODEOID_MCP_TOOL_TIMEOUT_MS", path: "session.mcpToolTimeoutMs", kind: "int" }, ]; // ── Loading ────────────────────────────────────────────────────────────── diff --git a/src/daemon/providers/claude/index.ts b/src/daemon/providers/claude/index.ts index 4246a2d..5711528 100644 --- a/src/daemon/providers/claude/index.ts +++ b/src/daemon/providers/claude/index.ts @@ -236,8 +236,13 @@ export class ClaudeProvider implements SessionProvider { : { sessionId: this.#claudeCodeSessionId }; // Merge user MCP servers with codeoid's in-process memory server. + // Apply a per-call wall-clock timeout to the external (user) servers so a + // hung MCP tool call surfaces as an SDK error instead of silently wedging + // the turn. Not applied to the in-process memory server. Kept below the + // session stall watchdog so the SDK signals first. + const mcpToolTimeoutMs = init.config?.session.mcpToolTimeoutMs ?? 120_000; const merged: Record = { - ...loadUserMcpServers(opts.workdir), + ...withMcpToolTimeout(loadUserMcpServers(opts.workdir), mcpToolTimeoutMs), ...(init.memory ? { codeoid_memory: buildMemoryMcpServer(init.memory, { @@ -621,6 +626,30 @@ export function translateSDKMessage( // ── Helpers ─────────────────────────────────────────────────────────────────── +/** + * Apply a per-call wall-clock `timeout` (ms) to external MCP servers so a hung + * tool call (e.g. an unresponsive HTTP gateway) returns an SDK error instead of + * silently stalling the turn. Only sets it when `ms > 0` and the server hasn't + * already declared its own `timeout`, so explicit per-server values still win. + * `timeout` is valid on every external (stdio/http/sse) McpServerConfig variant; + * these all come from JSON, so we re-cast at the same boundary parseMcpServerConfig uses. + */ +export function withMcpToolTimeout( + servers: Record, + ms: number, +): Record { + if (ms <= 0) return servers; + const out: Record = {}; + for (const [name, cfg] of Object.entries(servers)) { + const obj = cfg as unknown as Record; + out[name] = + typeof obj.timeout === "number" + ? cfg + : ({ ...obj, timeout: ms } as unknown as McpServerConfig); + } + return out; +} + function loadUserMcpServers(workdir: string): Record { try { const raw = readFileSync(join(homedir(), ".claude.json"), "utf8"); diff --git a/src/tests/config.test.ts b/src/tests/config.test.ts index 75bc58c..19940a1 100644 --- a/src/tests/config.test.ts +++ b/src/tests/config.test.ts @@ -157,6 +157,20 @@ describe("loadConfig — env precedence", () => { }), ).toThrow(/environment overrides|turnStallTimeoutMs/); }); + + it("defaults mcpToolTimeoutMs below turnStallTimeoutMs so the SDK signals first", () => { + const c = loadConfig({ configPath, env: {} }); + expect(c.session.mcpToolTimeoutMs).toBe(120000); + expect(c.session.mcpToolTimeoutMs!).toBeLessThan(c.session.turnStallTimeoutMs!); + }); + + it("CODEOID_MCP_TOOL_TIMEOUT_MS overrides the MCP tool timeout", () => { + const c = loadConfig({ + configPath, + env: { CODEOID_MCP_TOOL_TIMEOUT_MS: "30000" }, + }); + expect(c.session.mcpToolTimeoutMs).toBe(30000); + }); }); describe("loadConfig — path resolution", () => { diff --git a/src/tests/provider-claude.test.ts b/src/tests/provider-claude.test.ts index 2d21bf1..5aef4a2 100644 --- a/src/tests/provider-claude.test.ts +++ b/src/tests/provider-claude.test.ts @@ -59,6 +59,7 @@ import { translateSDKMessage, parseMcpServerConfig, extractToolResultText, + withMcpToolTimeout, } from "../daemon/providers/claude/index.js"; // ── Helpers ─────────────────────────────────────────────────────────────────── @@ -315,6 +316,37 @@ describe("parseMcpServerConfig", () => { }); }); +// ── withMcpToolTimeout ──────────────────────────────────────────────────────── + +describe("withMcpToolTimeout", () => { + it("injects timeout into external servers that don't declare one", () => { + const out = withMcpToolTimeout( + { + slack: { type: "http", url: "https://gw/mcp/slack" } as never, + local: { command: "node", args: ["x.js"] } as never, + }, + 120_000, + ); + expect((out.slack as { timeout?: number }).timeout).toBe(120_000); + expect((out.local as { timeout?: number }).timeout).toBe(120_000); + }); + + it("does not override a server's explicit timeout", () => { + const out = withMcpToolTimeout( + { slack: { type: "http", url: "https://gw", timeout: 5_000 } as never }, + 120_000, + ); + expect((out.slack as { timeout?: number }).timeout).toBe(5_000); + }); + + it("is a no-op when ms <= 0 (use the SDK default)", () => { + const servers = { slack: { type: "http", url: "https://gw" } as never }; + expect(withMcpToolTimeout(servers, 0)).toBe(servers); + const out = withMcpToolTimeout(servers, 0); + expect((out.slack as { timeout?: number }).timeout).toBeUndefined(); + }); +}); + // ── extractToolResultText ───────────────────────────────────────────────────── describe("extractToolResultText", () => { From 4f7a33b5d1c0bea202494cf41e632e2dc7022868 Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Mon, 29 Jun 2026 12:23:43 +0200 Subject: [PATCH 7/7] fix: enforce mcpToolTimeoutMs < turnStallTimeoutMs across overrides MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit: the "SDK signals first" layering was only locked in by the default values. An env override / config file could set the MCP timeout at or above the stall timeout, so the coarse watchdog would force-recover before the SDK's clean per-tool error fired — silently breaking the documented contract. Add a cross-field Zod refinement on SessionSchema. Because env overrides are re-validated through RootSchema after they're applied, this catches a bad override pair too, not just file config. Opt-out cases are exempt: turnStallTimeoutMs=0 (watchdog off → nothing to race) and mcpToolTimeoutMs=0 (use SDK default → relationship moot). The error message names the fix (lower one, or set either to 0). Tests: reject an out-of-order override pair; allow MCP >= stall when the watchdog is disabled. Adjusted the existing stall-override test to stay above the 120s MCP default. 627/627 pass, biome clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/config.ts | 19 ++++++++++++++++++- src/tests/config.test.ts | 30 ++++++++++++++++++++++++++++-- 2 files changed, 46 insertions(+), 3 deletions(-) diff --git a/src/config.ts b/src/config.ts index f9d4a43..63a93b6 100644 --- a/src/config.ts +++ b/src/config.ts @@ -236,7 +236,24 @@ const SessionSchema = z */ mcpToolTimeoutMs: z.number().min(0).default(120_000), }) - .default({ turnStallTimeoutMs: 300_000, mcpToolTimeoutMs: 120_000 }); + .default({ turnStallTimeoutMs: 300_000, mcpToolTimeoutMs: 120_000 }) + // Enforce the "SDK signals first" contract across BOTH fields — not just the + // defaults. An env override / config file could otherwise set the MCP timeout + // at or above the stall timeout, so the coarse watchdog would force-recover + // before the SDK's clean per-tool error fires. Exempt the opt-out cases: + // turnStallTimeoutMs=0 (watchdog off → nothing to race) and mcpToolTimeoutMs=0 + // (use SDK default → relationship is moot). + .refine( + (s) => + s.turnStallTimeoutMs === 0 || + s.mcpToolTimeoutMs === 0 || + s.mcpToolTimeoutMs < s.turnStallTimeoutMs, + { + message: + "must be less than session.turnStallTimeoutMs so a hung MCP call surfaces an SDK error before the stall watchdog fires (set either to 0 to opt out)", + path: ["mcpToolTimeoutMs"], + }, + ); const AutoRotateSchema = z .object({ diff --git a/src/tests/config.test.ts b/src/tests/config.test.ts index 19940a1..3fb0173 100644 --- a/src/tests/config.test.ts +++ b/src/tests/config.test.ts @@ -139,11 +139,13 @@ describe("loadConfig — env precedence", () => { }); it("CODEOID_TURN_STALL_TIMEOUT_MS overrides the stall watchdog timeout", () => { + // 200000 stays above the 120000 MCP-timeout default so the cross-field + // ordering refinement is satisfied. const c = loadConfig({ configPath, - env: { CODEOID_TURN_STALL_TIMEOUT_MS: "120000" }, + env: { CODEOID_TURN_STALL_TIMEOUT_MS: "200000" }, }); - expect(c.session.turnStallTimeoutMs).toBe(120000); + expect(c.session.turnStallTimeoutMs).toBe(200000); }); it("rejects a negative stall timeout (env override is re-validated, not just coerced)", () => { @@ -171,6 +173,30 @@ describe("loadConfig — env precedence", () => { }); expect(c.session.mcpToolTimeoutMs).toBe(30000); }); + + it("rejects an override pair where the MCP timeout is not below the stall timeout", () => { + // 400000 (MCP) >= 300000 (default stall) would let the coarse watchdog fire + // before the SDK's clean per-tool error — the cross-field refinement catches + // it even though each value is individually valid. + expect(() => + loadConfig({ + configPath, + env: { CODEOID_MCP_TOOL_TIMEOUT_MS: "400000" }, + }), + ).toThrow(/mcpToolTimeoutMs|turnStallTimeoutMs/); + }); + + it("allows MCP >= stall when the watchdog is disabled (turnStallTimeoutMs=0)", () => { + const c = loadConfig({ + configPath, + env: { + CODEOID_TURN_STALL_TIMEOUT_MS: "0", + CODEOID_MCP_TOOL_TIMEOUT_MS: "400000", + }, + }); + expect(c.session.turnStallTimeoutMs).toBe(0); + expect(c.session.mcpToolTimeoutMs).toBe(400000); + }); }); describe("loadConfig — path resolution", () => {