From 08f25f065a1a4724f4f8f15cd90f78587d676cb2 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 25 Mar 2026 20:12:18 +0700 Subject: [PATCH] Fix stale abort result contaminating replacement turns after interrupt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When interrupting a Claude agent mid-tool-call and sending a replacement message, the SDK's abort error result was being attributed to the new foreground turn, causing [System Error] banners and displaced replies. Root cause: pendingInterruptAbort was cleared by visible activity from the new turn before the stale result arrived (timing race), and the stale result then poisoned the replacement turn. Fix: - Suppress stale non-success results at the top of routeSdkMessageFromPump before any turn attribution, using both flag-based (pendingInterruptAbort) and content-based (isAbortError) detection - Stop clearing pendingInterruptAbort on visible activity — only consume it when a result message arrives - Prevent idle flash during replacement by checking pendingReplacement in agent-manager turn_completed/turn_canceled handlers - Fix vitest env loading to use import.meta.url instead of process.cwd() - Load .env.test eagerly in agent-configs.ts for collection-time availability Includes regression tests covering the exact f160a2a3 daemon log ordering. --- .../src/server/agent/agent-manager.test.ts | 114 ++++++++++ .../server/src/server/agent/agent-manager.ts | 6 +- ...agent.interrupt-restart-regression.test.ts | 67 ++++++ .../server/agent/providers/claude-agent.ts | 26 ++- .../src/server/daemon-e2e/agent-configs.ts | 8 +- ...d-during-tool-call-claude.real.e2e.test.ts | 195 +++++++++++++++--- .../server/src/test-utils/vitest-setup.ts | 6 +- 7 files changed, 388 insertions(+), 34 deletions(-) diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index e22cabb44..00866731b 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -1328,6 +1328,120 @@ describe("AgentManager", () => { unsubscribe(); }); + test("replaceAgentRun stays running when a stale old terminal arrives before the replacement turn is current", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-replace-stale-terminal-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + const secondStartEntered = deferred(); + const allowSecondStartToResolve = deferred(); + let capturedSession: StaleReplacementSession | null = null; + + class StaleReplacementSession extends TestAgentSession { + private localTurnCounter = 0; + + override async startTurn(): Promise<{ turnId: string }> { + const turnId = `turn-${++this.localTurnCounter}`; + const turnNum = this.localTurnCounter; + + if (turnNum === 1) { + setTimeout(() => { + this.pushEvent({ type: "turn_started", provider: this.provider, turnId }); + }, 0); + return { turnId }; + } + + secondStartEntered.resolve(); + await allowSecondStartToResolve.promise; + setTimeout(() => { + this.pushEvent({ type: "turn_started", provider: this.provider, turnId }); + this.pushEvent({ type: "turn_completed", provider: this.provider, turnId }); + }, 0); + return { turnId }; + } + + override async interrupt(): Promise { + this.pushEvent({ + type: "turn_canceled", + provider: this.provider, + reason: "Interrupted", + turnId: "turn-1", + }); + } + } + + class StaleReplacementClient extends TestAgentClient { + override async createSession(config: AgentSessionConfig): Promise { + capturedSession = new StaleReplacementSession(config); + return capturedSession; + } + } + + const manager = new AgentManager({ + clients: { + codex: new StaleReplacementClient(), + }, + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000126", + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workdir, + }); + + const lifecycleUpdates: string[] = []; + const unsubscribe = manager.subscribe( + (event) => { + if (event.type !== "agent_state" || event.agent.id !== snapshot.id) { + return; + } + lifecycleUpdates.push(event.agent.lifecycle); + }, + { agentId: snapshot.id, replayState: false }, + ); + + const firstRun = manager.streamAgent(snapshot.id, "first run"); + const firstRunDrain = (async () => { + for await (const _event of firstRun) { + // Drain events so lifecycle updates are applied. + } + })(); + + await manager.waitForAgentRunStart(snapshot.id); + + const secondRun = manager.replaceAgentRun(snapshot.id, "replacement run"); + const secondRunDrain = (async () => { + for await (const _event of secondRun) { + // Drain replacement run. + } + })(); + + await secondStartEntered.promise; + + const replaceGapSnapshot = manager.getAgent(snapshot.id) as + | ({ pendingReplacement: boolean; activeForegroundTurnId: string | null; lifecycle: string }) + | undefined; + expect(replaceGapSnapshot?.pendingReplacement).toBe(false); + expect(replaceGapSnapshot?.activeForegroundTurnId).toBeNull(); + expect(replaceGapSnapshot?.lifecycle).toBe("running"); + + capturedSession!.pushEvent({ type: "turn_completed", provider: "codex", turnId: "turn-1" }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + // eslint-disable-next-line no-console + console.log("replace-gap-lifecycle", lifecycleUpdates, manager.getAgent(snapshot.id)); + expect(manager.getAgent(snapshot.id)?.lifecycle).toBe("running"); + expect(lifecycleUpdates.at(-1)).toBe("running"); + + allowSecondStartToResolve.resolve(); + + await manager.waitForAgentRunStart(snapshot.id); + await firstRunDrain; + await secondRunDrain; + unsubscribe(); + }); + test("applies live autonomous events while no foreground run is active", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-live-events-")); const storagePath = join(workdir, "agents"); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 819e74a5b..e0bb8ce35 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -2038,7 +2038,8 @@ export class AgentManager { agent.lastUsage = event.usage; agent.lastError = undefined; // For autonomous turns (not foreground), transition to idle - if (!isForegroundEvent && agent.lifecycle !== "idle") { + // unless a replacement is pending (avoid idle flash during replace) + if (!isForegroundEvent && agent.lifecycle !== "idle" && !agent.pendingReplacement) { (agent as ActiveManagedAgent).lifecycle = "idle"; this.emitState(agent); } @@ -2094,7 +2095,8 @@ export class AgentManager { "handleStreamEvent: turn_canceled", ); // For autonomous turns, transition to idle - if (!isForegroundEvent) { + // unless a replacement is pending (avoid idle flash during replace) + if (!isForegroundEvent && !agent.pendingReplacement) { (agent as ActiveManagedAgent).lifecycle = "idle"; } agent.lastError = undefined; diff --git a/packages/server/src/server/agent/providers/claude-agent.interrupt-restart-regression.test.ts b/packages/server/src/server/agent/providers/claude-agent.interrupt-restart-regression.test.ts index 92ba1f5c7..b33d82758 100644 --- a/packages/server/src/server/agent/providers/claude-agent.interrupt-restart-regression.test.ts +++ b/packages/server/src/server/agent/providers/claude-agent.interrupt-restart-regression.test.ts @@ -428,6 +428,73 @@ describe("ClaudeAgentSession interrupt regression", () => { await session.close(); }); + + test("stale abort result after replacement start does not poison the new foreground turn", async () => { + const logger = createTestLogger(); + let queryRef: ScriptedQuery | null = null; + + sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable }) => { + queryRef = createScriptedQuery({ + prompt, + sessionId: "interrupt-stale-result-session", + }); + return queryRef; + }); + + const client = new ClaudeAgentClient({ logger }); + const session = await client.createSession({ + provider: "claude", + cwd: process.cwd(), + }); + + const firstTurn = streamSession(session, "first prompt"); + const firstStarted = await firstTurn.next(); + await waitFor(() => queryRef?.prompts.length === 1); + + await session.interrupt(); + const firstTurnEvents = [firstStarted.value!, ...(await collectUntilTerminal(firstTurn))]; + expect(firstTurnEvents.some((event) => event.type === "turn_canceled")).toBe(true); + + const observedSecondTurnEvents: AgentStreamEvent[] = []; + const unsubscribe = session.subscribe((event) => { + observedSecondTurnEvents.push(event); + }); + + const secondTurn = streamSession(session, "second prompt"); + const secondStarted = await secondTurn.next(); + await waitFor(() => queryRef?.prompts.length === 2); + + queryRef?.emit({ + type: "result", + subtype: "error_during_execution", + errors: ["Request was aborted."], + session_id: "interrupt-stale-result-session", + }); + queryRef?.emit({ + type: "assistant", + message: { content: "SECOND_PROMPT_RESPONSE" }, + session_id: "interrupt-stale-result-session", + }); + queryRef?.emit(buildSuccessResult("interrupt-stale-result-session")); + + const secondTurnEvents = [secondStarted.value!, ...(await collectUntilTerminal(secondTurn))]; + unsubscribe(); + + expect(secondTurnEvents.some((event) => event.type === "turn_failed")).toBe(false); + expect(secondTurnEvents.some((event) => event.type === "turn_canceled")).toBe(false); + expect(secondTurnEvents.some((event) => event.type === "turn_completed")).toBe(true); + expect(collectAssistantText(secondTurnEvents)).toContain("SECOND_PROMPT_RESPONSE"); + expect( + observedSecondTurnEvents.filter((event) => event.type === "turn_started").length, + ).toBe(1); + expect( + observedSecondTurnEvents.some( + (event) => event.type === "turn_failed" || event.type === "turn_canceled", + ), + ).toBe(false); + + await session.close(); + }); }); describe("ClaudeAgentSession autonomous turns", () => { diff --git a/packages/server/src/server/agent/providers/claude-agent.ts b/packages/server/src/server/agent/providers/claude-agent.ts index 2ad8351e0..e38b2379a 100644 --- a/packages/server/src/server/agent/providers/claude-agent.ts +++ b/packages/server/src/server/agent/providers/claude-agent.ts @@ -1977,6 +1977,12 @@ class ClaudeAgentSession implements AgentSession { this.transitionTurnState("idle", reason); } + private isAbortError(message: SDKMessage): boolean { + const errors = + "errors" in message && Array.isArray(message.errors) ? message.errors : []; + return errors.some((e: string) => /\baborted\b/i.test(e)); + } + private buildTurnFailedEvent( errorMessage: string, ): Extract { @@ -2247,6 +2253,25 @@ class ClaudeAgentSession implements AgentSession { } private routeSdkMessageFromPump(message: SDKMessage): void { + // Suppress stale results from interrupted requests. The cancel path already + // emitted the terminal event; this result is leftover from the killed API + // request. Consume the flag on ANY result so it doesn't linger. + if (message.type === "result" && this.pendingInterruptAbort) { + this.pendingInterruptAbort = false; + if (message.subtype !== "success") { + this.logger.debug("Suppressing stale non-success result from interrupted request"); + return; + } + } + if ( + message.type === "result" && + message.subtype !== "success" && + this.isAbortError(message) + ) { + this.logger.debug("Suppressing abort result by content"); + return; + } + const isForeground = Boolean(this.activeForegroundTurnId); const assistantishMessage = message.type === "assistant" || @@ -2333,7 +2358,6 @@ class ClaudeAgentSession implements AgentSession { ) ) { this.foregroundHasVisibleActivity = true; - this.pendingInterruptAbort = false; } this.dispatchEvents(events); diff --git a/packages/server/src/server/daemon-e2e/agent-configs.ts b/packages/server/src/server/daemon-e2e/agent-configs.ts index d95e81875..d4de333be 100644 --- a/packages/server/src/server/daemon-e2e/agent-configs.ts +++ b/packages/server/src/server/daemon-e2e/agent-configs.ts @@ -3,10 +3,16 @@ * Enables running the same tests against Claude, Codex, and OpenCode providers. */ import { existsSync } from "node:fs"; -import { join } from "node:path"; +import { join, dirname, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; import { homedir } from "node:os"; +import dotenv from "dotenv"; import { isCommandAvailable } from "../agent/provider-launch-config.js"; +// Load .env.test eagerly so isProviderAvailable() has credentials at collection time. +const serverRoot = resolve(dirname(fileURLToPath(import.meta.url)), "../../.."); +dotenv.config({ path: resolve(serverRoot, ".env.test"), override: true }); + export interface AgentTestConfig { provider: "claude" | "codex" | "opencode"; model: string; diff --git a/packages/server/src/server/daemon-e2e/send-during-tool-call-claude.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/send-during-tool-call-claude.real.e2e.test.ts index 0fb177929..b616a487b 100644 --- a/packages/server/src/server/daemon-e2e/send-during-tool-call-claude.real.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/send-during-tool-call-claude.real.e2e.test.ts @@ -16,6 +16,10 @@ function tmpCwd(): string { return mkdtempSync(path.join(tmpdir(), "daemon-real-tool-interrupt-")); } +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + function hasRunningToolCall(messages: SessionOutboundMessage[], agentId: string): boolean { for (const m of messages) { if ( @@ -31,9 +35,136 @@ function hasRunningToolCall(messages: SessionOutboundMessage[], agentId: string) return false; } +function getAssistantTexts(messages: SessionOutboundMessage[], agentId: string): string[] { + return messages + .filter( + (message) => + message.type === "agent_stream" && + message.payload.agentId === agentId && + message.payload.event.type === "timeline" && + message.payload.event.item.type === "assistant_message", + ) + .map((message) => message.payload.event.item.text); +} + +function hasProviderLimitText(text: string): boolean { + return /hit your limit|rate limit|quota|credits/i.test(text); +} + +function countTurnStarted(messages: SessionOutboundMessage[], agentId: string): number { + return messages.filter( + (message) => + message.type === "agent_stream" && + message.payload.agentId === agentId && + message.payload.event.type === "turn_started", + ).length; +} + +function getAgentStatuses(messages: SessionOutboundMessage[], agentId: string): string[] { + return messages + .filter( + (message) => + message.type === "agent_update" && + message.payload.kind === "upsert" && + message.payload.agent.id === agentId, + ) + .map((message) => message.payload.agent.status); +} + +function getStatusesBeforeFirstAssistant(messages: SessionOutboundMessage[], agentId: string): string[] { + const firstAssistantIndex = messages.findIndex( + (message) => + message.type === "agent_stream" && + message.payload.agentId === agentId && + message.payload.event.type === "timeline" && + message.payload.event.item.type === "assistant_message", + ); + if (firstAssistantIndex < 0) { + return []; + } + return getAgentStatuses(messages.slice(0, firstAssistantIndex), agentId); +} + +function summarizeTimelineItems(timeline: Awaited>) { + return timeline.entries.slice(-15).map((entry) => { + const item = entry.item; + if (item.type === "assistant_message") { + return { type: item.type, text: item.text }; + } + if (item.type === "tool_call") { + return { + type: item.type, + name: item.name, + status: item.status, + callId: item.callId, + }; + } + if (item.type === "user_message") { + return { type: item.type, text: item.text }; + } + return { type: item.type }; + }); +} + +async function waitForRunningToolCall( + client: DaemonClient, + collector: ReturnType, + agentId: string, + timeoutMs = 90_000, +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (hasRunningToolCall(collector.messages, agentId)) { + return; + } + + const timeline = await client.fetchAgentTimeline(agentId, { limit: 100 }).catch(() => null); + const assistantTexts = + timeline?.entries + .filter((entry) => entry.item.type === "assistant_message") + .slice(-5) + .map((entry) => entry.item.text) ?? []; + const limitText = assistantTexts.find((text) => hasProviderLimitText(text)); + if (limitText) { + throw new Error(`Claude could not reach the tool call because the provider rejected the run: ${limitText}`); + } + if ( + timeline?.entries.some( + (entry) => + entry.item.type === "tool_call" && + entry.item.name.toLowerCase() === "bash" && + entry.item.status === "running", + ) + ) { + return; + } + + await sleep(500); + } + + const timeline = await client.fetchAgentTimeline(agentId, { limit: 100 }).catch(() => null); + const recentToolCalls = + timeline?.entries + .filter((entry) => entry.item.type === "tool_call") + .slice(-10) + .map((entry) => ({ + name: entry.item.name, + status: entry.item.status, + callId: entry.item.callId, + })) ?? []; + const recentAssistantTexts = + timeline?.entries + .filter((entry) => entry.item.type === "assistant_message") + .slice(-5) + .map((entry) => entry.item.text) ?? []; + throw new Error( + `Timed out waiting for running tool call. Recent tool_calls=${JSON.stringify(recentToolCalls)} recent assistant text=${JSON.stringify(recentAssistantTexts)}`, + ); +} + describe("daemon E2E (real claude) - send message during tool call", () => { test.runIf(isProviderAvailable("claude"))( - "sending a message while a tool call is running starts a new turn", + "sending a message while a tool call is running replaces the turn without error, idle flash, or autonomous fallback", async () => { const logger = pino({ level: "silent" }); const cwd = tmpCwd(); @@ -59,7 +190,12 @@ describe("daemon E2E (real claude) - send message during tool call", () => { // Step 1: Ask Claude to run sleep 60 in the foreground await client.sendMessage( agent.id, - "Run the bash command `sleep 60` and wait for it to complete. Do not run it in the background.", + [ + "Use the Bash tool.", + "Run exactly: sleep 60", + "Do not use a background task.", + "Do not do anything after starting the command.", + ].join(" "), ); // Step 2: Wait for the agent to be running @@ -70,31 +206,9 @@ describe("daemon E2E (real claude) - send message during tool call", () => { ); // Step 3: Wait for a tool call to appear as "running" in the stream - await new Promise((resolve, reject) => { - const timeout = setTimeout(() => { - reject(new Error("Timed out waiting for running tool call")); - }, 90_000); + await waitForRunningToolCall(client, collector, agent.id); - if (hasRunningToolCall(collector.messages, agent.id)) { - clearTimeout(timeout); - resolve(); - return; - } - - const unsub = client.subscribeRawMessages((message) => { - if ( - message.type === "agent_stream" && - message.payload.agentId === agent.id && - message.payload.event.type === "timeline" && - message.payload.event.item.type === "tool_call" && - message.payload.event.item.status === "running" - ) { - clearTimeout(timeout); - unsub(); - resolve(); - } - }); - }); + collector.clear(); // Step 4: Send a second message while the tool call is still running await client.sendMessage(agent.id, "Reply with exactly: INTERRUPT_RECEIVED"); @@ -102,10 +216,34 @@ describe("daemon E2E (real claude) - send message during tool call", () => { // Step 5: Wait for the agent to finish — this is the critical assertion. // If the bug is present, the agent will stop and never start a new turn. const finish = await client.waitForFinish(agent.id, 120_000); - expect(finish.status).toBe("idle"); + const postSendMessages = [...collector.messages]; + const postSendAssistantTexts = getAssistantTexts(postSendMessages, agent.id); + const postSendStatuses = getAgentStatuses(postSendMessages, agent.id); + const statusesBeforeFirstAssistant = getStatusesBeforeFirstAssistant( + postSendMessages, + agent.id, + ); + const timeline = await client.fetchAgentTimeline(agent.id, { limit: 100 }); + + if (finish.status !== "idle") { + const snapshot = await client.fetchAgent(agent.id); + throw new Error( + `Expected idle after replacement, got ${finish.status}. postSendStatuses=${JSON.stringify(postSendStatuses)} statusesBeforeFirstAssistant=${JSON.stringify(statusesBeforeFirstAssistant)} postSendAssistantTexts=${JSON.stringify(postSendAssistantTexts)} turnStarted=${countTurnStarted(postSendMessages, agent.id)} agentStatus=${snapshot?.agent.status ?? null} recentTimeline=${JSON.stringify(summarizeTimelineItems(timeline))}`, + ); + } + + // Replacement should create exactly one new turn. A second turn_started here + // means the reply got displaced onto a later autonomous wake. + expect(countTurnStarted(postSendMessages, agent.id)).toBe(1); + + // The replacement path should not surface as agent error state. + expect(postSendStatuses).not.toContain("error"); + + // The agent should not flash idle before the replacement produces visible output. + expect(statusesBeforeFirstAssistant).not.toContain("idle"); + expect(statusesBeforeFirstAssistant).not.toContain("error"); // Step 6: Verify the agent actually responded to our second message - const timeline = await client.fetchAgentTimeline(agent.id, { limit: 100 }); const assistantTexts = timeline.entries .filter((entry) => entry.item.type === "assistant_message") .map((entry) => { @@ -118,6 +256,7 @@ describe("daemon E2E (real claude) - send message during tool call", () => { text.includes("[System Error]"), ); expect(hasSystemError).toBe(false); + expect(postSendAssistantTexts.some((text) => text.includes("[System Error]"))).toBe(false); const responded = assistantTexts.some((text) => text.toUpperCase().includes("INTERRUPT_RECEIVED"), diff --git a/packages/server/src/test-utils/vitest-setup.ts b/packages/server/src/test-utils/vitest-setup.ts index 8f0beb4ad..8b4be14e1 100644 --- a/packages/server/src/test-utils/vitest-setup.ts +++ b/packages/server/src/test-utils/vitest-setup.ts @@ -1,9 +1,11 @@ import path from "node:path"; +import { fileURLToPath } from "node:url"; import dotenv from "dotenv"; // Load package-local .env.test first for integration/E2E credentials, then repo-root .env fallback. -dotenv.config({ path: path.resolve(process.cwd(), ".env.test"), override: true }); -dotenv.config({ path: path.resolve(process.cwd(), "../.env") }); +const serverRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../.."); +dotenv.config({ path: path.resolve(serverRoot, ".env.test"), override: true }); +dotenv.config({ path: path.resolve(serverRoot, "../.env") }); process.env.GIT_TERMINAL_PROMPT = "0"; process.env.GIT_SSH_COMMAND = "ssh -oBatchMode=yes";