Fix OpenCode interrupt tool-call terminal state parity

2026-04-03 11:15:02 +07:00
parent d6413404e0
commit bb9ef76017
2 changed files with 538 additions and 20 deletions
@@ -29,6 +29,7 @@ import type {
  ListPersistedAgentsOptions,
  McpServerConfig,
  PersistedAgentDescriptor,
+  ToolCallTimelineItem,
 } from "../agent-sdk-types.js";
 import {
  applyProviderEnv,
@@ -201,6 +202,11 @@ function stringifyUnknownError(error: unknown): string {
  }
 }

+function normalizeTurnFailureError(error: unknown): string {
+  const normalized = stringifyUnknownError(error).trim();
+  return normalized.length > 0 ? normalized : "Unknown error";
+}
+
 function isAlreadyPresentMcpError(error: unknown): boolean {
  const normalized = stringifyUnknownError(error).toLowerCase();
  return MCP_ALREADY_PRESENT_ERROR_TOKENS.some((token) => normalized.includes(token));
@@ -955,11 +961,10 @@ export function translateOpenCodeEvent(
      if (sessionId === state.sessionId) {
        state.streamedPartKeys.clear();
        state.partTypes.clear();
-        const error = props.error as string | undefined;
        events.push({
          type: "turn_failed",
          provider: "opencode",
-          error: error ?? "Unknown error",
+          error: normalizeTurnFailureError(props.error),
        });
      }
      break;
@@ -995,6 +1000,7 @@ class OpenCodeAgentSession implements AgentSession {
  private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
  private nextTurnOrdinal = 0;
  private activeForegroundTurnId: string | null = null;
+  private readonly runningToolCalls = new Map<string, ToolCallTimelineItem>();

  constructor(
    config: OpenCodeAgentConfig,
@@ -1112,11 +1118,19 @@ class OpenCodeAgentSession implements AgentSession {
  }

  async interrupt(): Promise<void> {
-    this.abortController?.abort();
+    const turnId = this.activeForegroundTurnId;
+    const turnAbortController = this.abortController;
+    turnAbortController?.abort();
    await this.client.session.abort({
      sessionID: this.sessionId,
      directory: this.config.cwd,
    });
+    if (turnId) {
+      this.finishForegroundTurn(
+        { type: "turn_canceled", provider: "opencode", reason: "interrupted" },
+        turnId,
+      );
+    }
  }

  async startTurn(
@@ -1127,7 +1141,9 @@ class OpenCodeAgentSession implements AgentSession {
      throw new Error("A foreground turn is already active");
    }

-    this.abortController = new AbortController();
+    this.runningToolCalls.clear();
+    const turnAbortController = new AbortController();
+    this.abortController = turnAbortController;
    await this.ensureMcpServersConfigured();

    const parts = this.buildPromptParts(prompt);
@@ -1170,7 +1186,7 @@ class OpenCodeAgentSession implements AgentSession {
    }

    if (promptResponse.error) {
-      const errorMsg = JSON.stringify(promptResponse.error);
+      const errorMsg = normalizeTurnFailureError(promptResponse.error);
      this.notifySubscribers({
        type: "turn_failed",
        provider: "opencode",
@@ -1181,7 +1197,7 @@ class OpenCodeAgentSession implements AgentSession {

    const turnId = this.createTurnId();
    this.activeForegroundTurnId = turnId;
-    void this.consumeEventStream();
+    void this.consumeEventStream(turnId, turnAbortController);

    return { turnId };
  }
@@ -1193,40 +1209,118 @@ class OpenCodeAgentSession implements AgentSession {
    };
  }

-  private async consumeEventStream(): Promise<void> {
+  private async consumeEventStream(
+    turnId: string,
+    turnAbortController: AbortController,
+  ): Promise<void> {
    const eventsResult = await this.client.event.subscribe({
      directory: this.config.cwd,
    });

    try {
      for await (const event of eventsResult.stream) {
-        if (this.abortController?.signal.aborted) {
+        if (turnAbortController.signal.aborted || this.activeForegroundTurnId !== turnId) {
          break;
        }

        const translated = this.translateEvent(event);
        for (const e of translated) {
-          this.notifySubscribers(e);
-          if (e.type === "turn_completed" || e.type === "turn_failed") {
-            this.activeForegroundTurnId = null;
+          if (this.activeForegroundTurnId !== turnId) {
            return;
          }
+          if (e.type === "timeline" && e.item.type === "tool_call") {
+            this.trackToolCall(e.item);
+          }
+          if (e.type === "turn_completed" || e.type === "turn_failed" || e.type === "turn_canceled") {
+            if (e.type === "turn_failed") {
+              this.finishForegroundTurn(
+                {
+                  type: "turn_failed",
+                  provider: "opencode",
+                  error: normalizeTurnFailureError(e.error),
+                },
+                turnId,
+              );
+            } else {
+              this.finishForegroundTurn(e, turnId);
+            }
+            return;
+          }
+          this.notifySubscribers(e, turnId);
        }
      }
    } catch (error) {
-      if (!this.abortController?.signal.aborted) {
-        this.notifySubscribers({
-          type: "turn_failed",
-          provider: "opencode",
-          error: error instanceof Error ? error.message : "Stream error",
-        });
-        this.activeForegroundTurnId = null;
+      if (!turnAbortController.signal.aborted && this.activeForegroundTurnId === turnId) {
+        this.finishForegroundTurn(
+          {
+            type: "turn_failed",
+            provider: "opencode",
+            error: normalizeTurnFailureError(error),
+          },
+          turnId,
+        );
+      }
+    } finally {
+      if (turnAbortController.signal.aborted) {
+        this.finishForegroundTurn(
+          {
+            type: "turn_canceled",
+            provider: "opencode",
+            reason: "interrupted",
+          },
+          turnId,
+        );
+      }
+      if (this.abortController === turnAbortController && this.activeForegroundTurnId !== turnId) {
+        this.abortController = null;
      }
    }
  }

-  private notifySubscribers(event: AgentStreamEvent): void {
-    const turnId = this.activeForegroundTurnId;
+  private finishForegroundTurn(
+    event: Extract<AgentStreamEvent, { type: "turn_completed" | "turn_failed" | "turn_canceled" }>,
+    turnId: string,
+  ): void {
+    if (this.activeForegroundTurnId !== turnId) {
+      return;
+    }
+    if (event.type === "turn_canceled" || event.type === "turn_failed") {
+      this.synthesizeInterruptedToolCalls(turnId);
+    } else {
+      this.runningToolCalls.clear();
+    }
+    this.activeForegroundTurnId = null;
+    this.notifySubscribers(event, turnId);
+  }
+
+  private trackToolCall(item: ToolCallTimelineItem): void {
+    if (item.status === "running") {
+      this.runningToolCalls.set(item.callId, item);
+      return;
+    }
+    this.runningToolCalls.delete(item.callId);
+  }
+
+  private synthesizeInterruptedToolCalls(turnId: string): void {
+    for (const item of this.runningToolCalls.values()) {
+      this.notifySubscribers(
+        {
+          type: "timeline",
+          provider: "opencode",
+          item: {
+            ...item,
+            status: "failed",
+            error: { message: "Tool execution aborted" },
+          },
+        },
+        turnId,
+      );
+    }
+    this.runningToolCalls.clear();
+  }
+
+  private notifySubscribers(event: AgentStreamEvent, turnIdOverride?: string): void {
+    const turnId = turnIdOverride ?? this.activeForegroundTurnId;
    const tagged = turnId ? { ...event, turnId } : event;
    for (const callback of this.subscribers) {
      try {
@@ -0,0 +1,424 @@
+import { describe, expect, test } from "vitest";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import pino from "pino";
+
+import { OpenCodeAgentClient } from "../agent/providers/opencode-agent.js";
+import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
+import { DaemonClient, type WaitForFinishResult } from "../test-utils/daemon-client.js";
+import { createMessageCollector } from "../test-utils/message-collector.js";
+import { isProviderAvailable } from "./agent-configs.js";
+import type { AgentPermissionRequest } from "../agent/agent-sdk-types.js";
+import type { SessionOutboundMessage } from "../messages.js";
+
+const SYSTEM_ERROR_SNIPPET = "A foreground turn is already active";
+
+function tmpCwd(): string {
+  return mkdtempSync(path.join(tmpdir(), "daemon-real-opencode-send-interrupt-"));
+}
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+function pickOpenCodeModel(
+  models: Array<{ id: string }>,
+  preferences: string[] = [
+    "minimax-m2.5-free",
+    "kimi-k2.5-free",
+    "glm-5-free",
+    "free",
+    "mini",
+    "gpt-5-nano",
+  ],
+): string {
+  const preferred = models.find((model) =>
+    preferences.some((fragment) => model.id.includes(fragment)),
+  );
+  return preferred?.id ?? models[0]!.id;
+}
+
+function hasRunningBashToolCall(messages: SessionOutboundMessage[], agentId: string): boolean {
+  return messages.some(
+    (message) =>
+      message.type === "agent_stream" &&
+      message.payload.agentId === agentId &&
+      message.payload.event.type === "timeline" &&
+      message.payload.event.item.type === "tool_call" &&
+      message.payload.event.item.status === "running" &&
+      ["bash", "shell"].includes(message.payload.event.item.name.toLowerCase()),
+  );
+}
+
+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 findSystemErrorText(texts: string[]): string | null {
+  return texts.find((text) => text.includes("[System Error]")) ?? null;
+}
+
+function getTimelineAssistantTexts(
+  timeline: Awaited<ReturnType<DaemonClient["fetchAgentTimeline"]>>,
+): string[] {
+  return timeline.entries
+    .filter((entry) => entry.item.type === "assistant_message")
+    .map((entry) => entry.item.text);
+}
+
+function findSleepToolCall(
+  timeline: Awaited<ReturnType<DaemonClient["fetchAgentTimeline"]>>,
+): { status: "running" | "completed" | "failed" | "canceled"; callId: string } | null {
+  for (let idx = timeline.entries.length - 1; idx >= 0; idx -= 1) {
+    const entry = timeline.entries[idx];
+    if (entry?.item.type !== "tool_call") {
+      continue;
+    }
+    if (entry.item.detail.type !== "shell") {
+      continue;
+    }
+    if (!entry.item.detail.command.includes("sleep 60")) {
+      continue;
+    }
+    return {
+      status: entry.item.status,
+      callId: entry.item.callId,
+    };
+  }
+  return null;
+}
+
+async function allowPermission(
+  client: DaemonClient,
+  agentId: string,
+  permission: AgentPermissionRequest,
+): Promise<void> {
+  if (permission.kind === "question") {
+    throw new Error(
+      `Unexpected question permission while waiting for tool call: ${permission.id} ${permission.title}`,
+    );
+  }
+  await client.respondToPermission(agentId, permission.id, {
+    behavior: "allow",
+    message: "Approved by integration test",
+  });
+}
+
+async function approvePendingPermissions(
+  client: DaemonClient,
+  agentId: string,
+  handledPermissionIds: Set<string>,
+): Promise<void> {
+  const snapshot = await client.fetchAgent(agentId).catch(() => null);
+  const pending = snapshot?.agent.pendingPermissions ?? [];
+  for (const permission of pending) {
+    if (handledPermissionIds.has(permission.id)) {
+      continue;
+    }
+    handledPermissionIds.add(permission.id);
+    await allowPermission(client, agentId, permission);
+  }
+}
+
+async function waitForRunningBashToolCall(
+  client: DaemonClient,
+  collector: ReturnType<typeof createMessageCollector>,
+  agentId: string,
+  timeoutMs = 120_000,
+): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+  const handledPermissionIds = new Set<string>();
+
+  while (Date.now() < deadline) {
+    await approvePendingPermissions(client, agentId, handledPermissionIds);
+
+    const streamSystemError = findSystemErrorText(getAssistantTexts(collector.messages, agentId));
+    if (streamSystemError) {
+      throw new Error(`OpenCode failed before tool call started: ${streamSystemError}`);
+    }
+
+    if (hasRunningBashToolCall(collector.messages, agentId)) {
+      return;
+    }
+
+    const timeline = await client.fetchAgentTimeline(agentId, { limit: 120 }).catch(() => null);
+    const timelineSystemError = timeline
+      ? findSystemErrorText(getTimelineAssistantTexts(timeline).slice(-8))
+      : null;
+    if (timelineSystemError) {
+      throw new Error(`OpenCode failed before tool call started: ${timelineSystemError}`);
+    }
+    if (
+      timeline?.entries.some(
+        (entry) =>
+          entry.item.type === "tool_call" &&
+          entry.item.status === "running" &&
+          ["bash", "shell"].includes(entry.item.name.toLowerCase()),
+      )
+    ) {
+      return;
+    }
+
+    await sleep(500);
+  }
+
+  const timeline = await client.fetchAgentTimeline(agentId, { limit: 120 }).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 ? getTimelineAssistantTexts(timeline).slice(-6) : [];
+  throw new Error(
+    `Timed out waiting for running bash/shell tool call. recentToolCalls=${JSON.stringify(recentToolCalls)} recentAssistantTexts=${JSON.stringify(recentAssistantTexts)}`,
+  );
+}
+
+async function waitForSleepToolCallTerminal(
+  client: DaemonClient,
+  agentId: string,
+  timeoutMs = 30_000,
+): Promise<{ status: "completed" | "failed" | "canceled"; callId: string }> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    const timeline = await client.fetchAgentTimeline(agentId, { limit: 200 });
+    const sleepToolCall = findSleepToolCall(timeline);
+    if (sleepToolCall && sleepToolCall.status !== "running") {
+      return {
+        status: sleepToolCall.status,
+        callId: sleepToolCall.callId,
+      };
+    }
+    await sleep(300);
+  }
+
+  const timeline = await client.fetchAgentTimeline(agentId, { limit: 200 }).catch(() => null);
+  const recentToolCalls =
+    timeline?.entries
+      .filter((entry) => entry.item.type === "tool_call")
+      .slice(-10)
+      .map((entry) => ({
+        callId: entry.item.callId,
+        name: entry.item.name,
+        status: entry.item.status,
+      })) ?? [];
+  throw new Error(
+    `Timed out waiting for interrupted sleep tool call to become terminal. recentToolCalls=${JSON.stringify(recentToolCalls)}`,
+  );
+}
+
+async function waitForIdleResolvingPermissions(
+  client: DaemonClient,
+  agentId: string,
+  timeoutMs: number,
+): Promise<WaitForFinishResult> {
+  const deadline = Date.now() + timeoutMs;
+  const handledPermissionIds = new Set<string>();
+
+  while (Date.now() < deadline) {
+    const remaining = Math.max(1, deadline - Date.now());
+    const result = await client.waitForFinish(agentId, Math.min(remaining, 45_000));
+    if (result.status !== "permission") {
+      return result;
+    }
+
+    const pendingPermissions = result.final?.pendingPermissions ?? [];
+    if (pendingPermissions.length === 0) {
+      throw new Error("waitForFinish reported permission but no pending permissions were present");
+    }
+
+    let resolvedAny = false;
+    for (const permission of pendingPermissions) {
+      if (handledPermissionIds.has(permission.id)) {
+        continue;
+      }
+      handledPermissionIds.add(permission.id);
+      await allowPermission(client, agentId, permission);
+      resolvedAny = true;
+    }
+
+    if (!resolvedAny) {
+      throw new Error(
+        "Permission wait loop made no progress; all permissions were already handled",
+      );
+    }
+  }
+
+  return {
+    status: "timeout",
+    final: null,
+    error: `Timed out waiting for idle after ${timeoutMs}ms`,
+    lastMessage: null,
+  };
+}
+
+async function createHarness(): Promise<{
+  client: DaemonClient;
+  daemon: Awaited<ReturnType<typeof createTestPaseoDaemon>>;
+}> {
+  const logger = pino({ level: "silent" });
+  const daemon = await createTestPaseoDaemon({
+    agentClients: { opencode: new OpenCodeAgentClient(logger) },
+    logger,
+  });
+  const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
+  await client.connect();
+  await client.fetchAgents({ subscribe: { subscriptionId: "opencode-send-interrupt-real" } });
+  return { client, daemon };
+}
+
+describe("daemon E2E (real opencode) - send while working and interrupt", () => {
+  test.runIf(isProviderAvailable("opencode"))(
+    "send_message while sleep tool call is running starts a clean replacement turn",
+    async () => {
+      const cwd = tmpCwd();
+      const { client, daemon } = await createHarness();
+      const collector = createMessageCollector(client);
+      const followUpToken = "OPENCODE_SEND_WHILE_WORKING_OK";
+
+      try {
+        const modelList = await client.listProviderModels("opencode");
+        expect(modelList.models.length).toBeGreaterThan(0);
+
+        const agent = await client.createAgent({
+          provider: "opencode",
+          cwd,
+          title: "OpenCode send while working",
+          model: pickOpenCodeModel(modelList.models),
+          modeId: "default",
+        });
+
+        await client.sendMessage(
+          agent.id,
+          [
+            "Use the Bash tool.",
+            "Run exactly: sleep 60",
+            "Do not run it in the background.",
+            "Do not do anything after starting the command.",
+          ].join(" "),
+        );
+
+        await client.waitForAgentUpsert(
+          agent.id,
+          (snapshot) => snapshot.status === "running",
+          90_000,
+        );
+        await waitForRunningBashToolCall(client, collector, agent.id);
+
+        collector.clear();
+        await client.sendMessage(agent.id, `Reply with exactly: ${followUpToken}`);
+
+        const finish = await waitForIdleResolvingPermissions(client, agent.id, 240_000);
+        expect(finish.status).toBe("idle");
+
+        const postSendAssistantTexts = getAssistantTexts(collector.messages, agent.id);
+        expect(
+          postSendAssistantTexts.some((text) => text.includes("[System Error]")),
+        ).toBe(false);
+        expect(
+          postSendAssistantTexts.some((text) => text.includes(SYSTEM_ERROR_SNIPPET)),
+        ).toBe(false);
+
+        const timeline = await client.fetchAgentTimeline(agent.id, { limit: 160 });
+        const assistantTexts = getTimelineAssistantTexts(timeline);
+        expect(assistantTexts.some((text) => text.includes(followUpToken))).toBe(true);
+        expect(assistantTexts.some((text) => text.includes("[System Error]"))).toBe(false);
+        expect(assistantTexts.some((text) => text.includes(SYSTEM_ERROR_SNIPPET))).toBe(false);
+      } finally {
+        collector.unsubscribe();
+        await client.close().catch(() => undefined);
+        await daemon.close();
+        rmSync(cwd, { recursive: true, force: true });
+      }
+    },
+    360_000,
+  );
+
+  test.runIf(isProviderAvailable("opencode"))(
+    "explicit interrupt during sleep tool call still allows the next turn to complete",
+    async () => {
+      const cwd = tmpCwd();
+      const { client, daemon } = await createHarness();
+      const collector = createMessageCollector(client);
+      const followUpToken = "OPENCODE_INTERRUPT_FOLLOWUP_OK";
+
+      try {
+        const modelList = await client.listProviderModels("opencode");
+        expect(modelList.models.length).toBeGreaterThan(0);
+
+        const agent = await client.createAgent({
+          provider: "opencode",
+          cwd,
+          title: "OpenCode explicit interrupt",
+          model: pickOpenCodeModel(modelList.models),
+          modeId: "default",
+        });
+
+        await client.sendMessage(
+          agent.id,
+          [
+            "Use the Bash tool.",
+            "Run exactly: sleep 60",
+            "Do not run it in the background.",
+            "Do not do anything after starting the command.",
+          ].join(" "),
+        );
+
+        await client.waitForAgentUpsert(
+          agent.id,
+          (snapshot) => snapshot.status === "running",
+          90_000,
+        );
+        await waitForRunningBashToolCall(client, collector, agent.id);
+
+        await client.cancelAgent(agent.id);
+        await client.waitForAgentUpsert(
+          agent.id,
+          (snapshot) => snapshot.status === "idle" || snapshot.status === "error",
+          90_000,
+        );
+        const interruptedToolCall = await waitForSleepToolCallTerminal(client, agent.id, 45_000);
+        expect(interruptedToolCall.status).toBe("failed");
+
+        collector.clear();
+        await client.sendMessage(agent.id, `Reply with exactly: ${followUpToken}`);
+
+        const finish = await waitForIdleResolvingPermissions(client, agent.id, 240_000);
+        expect(finish.status).toBe("idle");
+
+        const postInterruptAssistantTexts = getAssistantTexts(collector.messages, agent.id);
+        expect(
+          postInterruptAssistantTexts.some((text) => text.includes("[System Error]")),
+        ).toBe(false);
+        expect(
+          postInterruptAssistantTexts.some((text) => text.includes(SYSTEM_ERROR_SNIPPET)),
+        ).toBe(false);
+
+        const timeline = await client.fetchAgentTimeline(agent.id, { limit: 200 });
+        const assistantTexts = getTimelineAssistantTexts(timeline);
+        expect(assistantTexts.some((text) => text.includes(followUpToken))).toBe(true);
+        expect(assistantTexts.some((text) => text.includes("[System Error]"))).toBe(false);
+        expect(assistantTexts.some((text) => text.includes(SYSTEM_ERROR_SNIPPET))).toBe(false);
+      } finally {
+        collector.unsubscribe();
+        await client.close().catch(() => undefined);
+        await daemon.close();
+        rmSync(cwd, { recursive: true, force: true });
+      }
+    },
+    360_000,
+  );
+});