diff --git a/packages/server/src/server/agent/providers/opencode-agent.test.ts b/packages/server/src/server/agent/providers/opencode-agent.test.ts index 4fbea8854..77f467028 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.test.ts @@ -555,3 +555,47 @@ describe("OpenCode adapter context-window normalization", () => { ]); }); }); + +describe("OpenCode adapter startTurn error handling", () => { + test("emits turn_failed when client.session.promptAsync throws synchronously", async () => { + // Async iterable that never yields and never resolves. The IIFE in + // startTurn synchronously hits the promptAsync throw and finishes the + // turn before this iterator is ever pulled, so the never-resolving + // promise inside next() is fine and gets garbage-collected. + const neverYieldingStream: AsyncIterable = { + [Symbol.asyncIterator]: () => ({ + next: () => new Promise(() => {}), + }), + }; + + const fakeClient = { + event: { + subscribe: vi.fn().mockResolvedValue({ stream: neverYieldingStream }), + }, + session: { + promptAsync: vi.fn(() => { + throw new Error("boom: synchronous throw"); + }), + }, + } as never; + + const session = new __openCodeInternals.OpenCodeAgentSession( + { provider: "opencode", cwd: "/tmp/test" }, + fakeClient, + "ses_unit_test", + createTestLogger(), + ); + + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + await session.startTurn("hello"); + + const failed = events.find((event) => event.type === "turn_failed"); + expect(failed).toBeDefined(); + expect(failed?.type).toBe("turn_failed"); + if (failed?.type === "turn_failed") { + expect(failed.error).toContain("boom: synchronous throw"); + } + }); +}); diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index 4be4a5547..44ce051ba 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -728,6 +728,9 @@ export const __openCodeInternals = { reconcileOpenCodeSessionClose, resolveOpenCodeModelLookupKeyFromAssistantMessage, resolveOpenCodeSelectedModelContextWindow, + get OpenCodeAgentSession() { + return OpenCodeAgentSession; + }, }; export class OpenCodeServerManager { @@ -1864,6 +1867,22 @@ function appendOpenCodeSessionStatus( // "retry" and "busy" are transient — no terminal event. } +interface Deferred { + promise: Promise; + resolve: (value: T) => void; + reject: (error: unknown) => void; +} + +function createDeferred(): Deferred { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + class OpenCodeAgentSession implements AgentSession { readonly provider = "opencode" as const; readonly capabilities = OPENCODE_CAPABILITIES; @@ -2058,7 +2077,20 @@ class OpenCodeAgentSession implements AgentSession { const turnId = this.createTurnId(); this.activeForegroundTurnId = turnId; - void this.consumeEventStream(turnId, turnAbortController); + + // OpenCode's /event SSE endpoint does NOT replay past events. If we send + // the prompt before our reader is connected, terminal events fired early + // by the server (e.g. session.error / session.idle for invalid model or + // mode) are missed and the turn hangs forever. Wait for the subscription + // to be established before sending anything. + const subscriptionReady = createDeferred(); + void this.consumeEventStream(turnId, turnAbortController, subscriptionReady); + try { + await subscriptionReady.promise; + } catch { + // consumeEventStream already finished the turn with the subscription error. + return { turnId }; + } const slashCommand = await this.resolveSlashCommandInvocation(prompt); if (slashCommand) { @@ -2160,25 +2192,28 @@ class OpenCodeAgentSession implements AgentSession { ); }); } else { - void this.client.session - .promptAsync({ - sessionID: this.sessionId, - directory: this.config.cwd, - parts, - ...(options?.outputSchema - ? { - format: { - type: "json_schema" as const, - schema: options.outputSchema as Record, - }, - } - : {}), - ...(this.config.systemPrompt ? { system: this.config.systemPrompt } : {}), - ...(model ? { model } : {}), - ...(effectiveMode ? { agent: effectiveMode } : {}), - ...(effectiveVariant ? { variant: effectiveVariant } : {}), - }) - .then((promptResponse) => { + // Wrap in an async IIFE so a synchronous throw from promptAsync (e.g. + // SDK input validation) is caught alongside async rejections. A plain + // `.then().catch()` chain would let a sync throw escape unhandled. + void (async () => { + try { + const promptResponse = await this.client.session.promptAsync({ + sessionID: this.sessionId, + directory: this.config.cwd, + parts, + ...(options?.outputSchema + ? { + format: { + type: "json_schema" as const, + schema: options.outputSchema as Record, + }, + } + : {}), + ...(this.config.systemPrompt ? { system: this.config.systemPrompt } : {}), + ...(model ? { model } : {}), + ...(effectiveMode ? { agent: effectiveMode } : {}), + ...(effectiveVariant ? { variant: effectiveVariant } : {}), + }); if (promptResponse.error) { this.finishForegroundTurn( { @@ -2189,9 +2224,7 @@ class OpenCodeAgentSession implements AgentSession { turnId, ); } - return; - }) - .catch((error) => { + } catch (error) { this.finishForegroundTurn( { type: "turn_failed", @@ -2200,7 +2233,8 @@ class OpenCodeAgentSession implements AgentSession { }, turnId, ); - }); + } + })(); } return { turnId }; @@ -2216,12 +2250,14 @@ class OpenCodeAgentSession implements AgentSession { private async consumeEventStream( turnId: string, turnAbortController: AbortController, + subscriptionReady: Deferred, ): Promise { try { const result = await this.client.event.subscribe( { directory: this.config.cwd }, { signal: turnAbortController.signal, sseMaxRetryAttempts: 0 }, ); + subscriptionReady.resolve(); for await (const event of result.stream) { if (turnAbortController.signal.aborted || this.activeForegroundTurnId !== turnId) { @@ -2256,6 +2292,7 @@ class OpenCodeAgentSession implements AgentSession { ); } } catch (error) { + subscriptionReady.reject(error); if (!turnAbortController.signal.aborted && this.activeForegroundTurnId === turnId) { this.finishForegroundTurn( { diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index e68a5f11f..c8c43df55 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -850,6 +850,14 @@ export async function createPaseoDaemon( if (wsServer) { await wsServer.close(); } + // Force-drop remaining sockets so httpServer.close() resolves promptly. + // We've already closed wsServer (which sent ws-layer close frames) and + // stopped every other service, so anything still attached is a TCP + // socket whose higher-level shutdown hasn't fully released it (e.g. + // upgraded WS sockets in the closing handshake, or HTTP keep-alive + // sockets in CLOSE_WAIT). closeIdleConnections() does not catch + // upgraded sockets, so we use closeAllConnections() here. + httpServer.closeAllConnections(); await new Promise((resolve) => { httpServer.close(() => resolve()); }); diff --git a/packages/server/src/server/daemon-e2e/opencode-invalid-mode.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/opencode-invalid-mode.real.e2e.test.ts new file mode 100644 index 000000000..6c431d1e1 --- /dev/null +++ b/packages/server/src/server/daemon-e2e/opencode-invalid-mode.real.e2e.test.ts @@ -0,0 +1,69 @@ +import { beforeAll, beforeEach, 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 } from "../test-utils/daemon-client.js"; +import { isProviderAvailable } from "./agent-configs.js"; + +function tmpCwd(): string { + return mkdtempSync(path.join(tmpdir(), "daemon-real-opencode-invalid-mode-")); +} + +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-invalid-mode" } }); + return { client, daemon }; +} + +describe("daemon E2E (real opencode) - invalid mode handling", () => { + let canRun = false; + + beforeAll(async () => { + canRun = await isProviderAvailable("opencode"); + }); + + beforeEach((context) => { + if (!canRun) { + context.skip(); + } + }); + + test("initial prompt with a nonexistent OpenCode mode fails instead of hanging forever", async () => { + const cwd = tmpCwd(); + const { client, daemon } = await createHarness(); + + try { + const agent = await client.createAgent({ + provider: "opencode", + cwd, + title: "OpenCode invalid mode regression", + modeId: "definitely-not-a-real-mode", + initialPrompt: "hello", + }); + + const finish = await client.waitForFinish(agent.id, 30_000); + expect(finish.status).toBe("error"); + expect(finish.error).toBeTruthy(); + + const snapshot = await client.fetchAgent(agent.id); + expect(snapshot.agent?.status).toBe("error"); + } finally { + await client.close().catch(() => undefined); + await daemon.close(); + rmSync(cwd, { recursive: true, force: true }); + } + }, 60_000); +});