Fix stale abort result contaminating replacement turns after interrupt

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.
2026-03-25 20:12:18 +07:00
parent 36327cad00
commit 08f25f065a
7 changed files with 388 additions and 34 deletions
@@ -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<void>();
+    const allowSecondStartToResolve = deferred<void>();
+    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<void> {
+        this.pushEvent({
+          type: "turn_canceled",
+          provider: this.provider,
+          reason: "Interrupted",
+          turnId: "turn-1",
+        });
+      }
+    }
+
+    class StaleReplacementClient extends TestAgentClient {
+      override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
+        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");
@@ -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;
@@ -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<unknown> }) => {
+      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", () => {
@@ -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<AgentStreamEvent, { type: "turn_failed" }> {
@@ -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);
@@ -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;
@@ -16,6 +16,10 @@ function tmpCwd(): string {
  return mkdtempSync(path.join(tmpdir(), "daemon-real-tool-interrupt-"));
 }

+function sleep(ms: number): Promise<void> {
+  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<ReturnType<DaemonClient["fetchAgentTimeline"]>>) {
+  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<typeof createMessageCollector>,
+  agentId: string,
+  timeoutMs = 90_000,
+): Promise<void> {
+  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<void>((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"),
@@ -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";