From bb9ef76017a488a592e7d7076970bb744b7fc05e Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 3 Apr 2026 11:15:02 +0700 Subject: [PATCH] Fix OpenCode interrupt tool-call terminal state parity --- .../server/agent/providers/opencode-agent.ts | 134 +++++- .../opencode-send-interrupt.real.e2e.test.ts | 424 ++++++++++++++++++ 2 files changed, 538 insertions(+), 20 deletions(-) create mode 100644 packages/server/src/server/daemon-e2e/opencode-send-interrupt.real.e2e.test.ts diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index e6d35cf5f..9f6a26749 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -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(); constructor( config: OpenCodeAgentConfig, @@ -1112,11 +1118,19 @@ class OpenCodeAgentSession implements AgentSession { } async interrupt(): Promise { - 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 { + private async consumeEventStream( + turnId: string, + turnAbortController: AbortController, + ): Promise { 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, + 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 { diff --git a/packages/server/src/server/daemon-e2e/opencode-send-interrupt.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/opencode-send-interrupt.real.e2e.test.ts new file mode 100644 index 000000000..4d4ceb1fa --- /dev/null +++ b/packages/server/src/server/daemon-e2e/opencode-send-interrupt.real.e2e.test.ts @@ -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 { + 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>, +): string[] { + return timeline.entries + .filter((entry) => entry.item.type === "assistant_message") + .map((entry) => entry.item.text); +} + +function findSleepToolCall( + timeline: Awaited>, +): { 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 { + 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, +): Promise { + 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, + agentId: string, + timeoutMs = 120_000, +): Promise { + const deadline = Date.now() + timeoutMs; + const handledPermissionIds = new Set(); + + 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 { + const deadline = Date.now() + timeoutMs; + const handledPermissionIds = new Set(); + + 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>; +}> { + 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, + ); +});