From 00e7ac9ec8ae5ed85b98482fd9fe16749d7cac1c Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 17 May 2026 19:51:20 +0800 Subject: [PATCH] Fix OpenCode custom command hangs (#1063) * Fix OpenCode slash command turns * ci: install opencode in server-tests so real OpenCode e2e runs Server tests previously only installed the Claude Code CLI, so every opencode-*.real.e2e.test.ts self-skipped on CI. The custom-command and turn-lifecycle real tests cover the OpenCode session SSE rewrite, so add opencode-ai globally and let them run. Big Pickle is free so no provider secrets needed. * Fix OpenCode unit tests exposed by enabling opencode in CI server-tests previously skipped every OpenCodeAgentClient test because opencode-ai was not installed. Now that CI runs them, four real issues surface: - Drop unused mkdirSync import (was used on a code path 1812b1489 refactored away on main; the rebase merged imports without rechecking usage). Fixes the lint job. - Fix slash-command-timeout test passing four ctor args instead of three, which silently dropped the TestOpenCodeRuntime and ran against the real opencode runtime. - Relax the listModels contextWindowMaxTokens assertion. Some providers routed through OpenCode (OpenAI in particular) don't expose a numeric context window; assert the type only when the field is present. - Swallow late notifySubscribers calls once close() flips the closed flag and clear subscribers up front. The session-lifetime SSE stream introduced by the prior commit means a session.error arriving after the owning test has moved on can rejectCompletion on a deferred no one is awaiting, surfacing as an unhandled rejection in a downstream test file. --- .github/workflows/ci.yml | 4 +- .../opencode-agent-commands.real.e2e.test.ts | 16 +- ...code-agent-custom-command.real.e2e.test.ts | 107 ++++++++++ .../opencode-agent.full-access.test.ts | 3 +- ...encode-agent.slash-command-timeout.test.ts | 40 ++++ .../agent/providers/opencode-agent.test.ts | 96 ++++----- .../server/agent/providers/opencode-agent.ts | 201 +++++++++--------- ...pencode-assistant-message.real.e2e.test.ts | 16 +- .../opencode-reasoning-dedup.real.e2e.test.ts | 13 +- .../providers/opencode-server-manager.test.ts | 13 ++ .../providers/opencode/server-manager.ts | 5 +- .../test-utils/test-opencode-runtime.ts | 51 ++++- .../src/server/daemon-e2e/agent-configs.ts | 2 +- 13 files changed, 394 insertions(+), 173 deletions(-) create mode 100644 packages/server/src/server/agent/providers/opencode-agent-custom-command.real.e2e.test.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d8d116465..f3115f387 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -104,8 +104,8 @@ jobs: - name: Install dependencies run: npm ci - - name: Install Claude Code CLI for provider tests - run: npm install -g @anthropic-ai/claude-code + - name: Install agent CLIs for provider tests + run: npm install -g @anthropic-ai/claude-code opencode-ai - name: Build highlight dependency run: npm run build --workspace=@getpaseo/highlight diff --git a/packages/server/src/server/agent/providers/opencode-agent-commands.real.e2e.test.ts b/packages/server/src/server/agent/providers/opencode-agent-commands.real.e2e.test.ts index f3368fc30..bffaeeac1 100644 --- a/packages/server/src/server/agent/providers/opencode-agent-commands.real.e2e.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent-commands.real.e2e.test.ts @@ -1,8 +1,12 @@ -import { beforeAll, beforeEach, describe, expect, test } from "vitest"; +import { afterAll, beforeAll, beforeEach, describe, expect, test } from "vitest"; import pino from "pino"; import { isCommandAvailable } from "../../../utils/executable.js"; import { OpenCodeAgentClient } from "./opencode-agent.js"; +import { OpenCodeServerManager } from "./opencode/server-manager.js"; + +const BIG_PICKLE_MODEL = "opencode/big-pickle"; +const logger = pino({ level: "silent" }); describe("opencode agent commands contract (real)", () => { let canRun = false; @@ -17,13 +21,18 @@ describe("opencode agent commands contract (real)", () => { } }); + afterAll(async () => { + await OpenCodeServerManager.getInstance(logger).shutdown(); + }); + test("lists slash commands with the expected contract", async () => { expect(await isCommandAvailable("opencode")).toBe(true); - const client = new OpenCodeAgentClient(pino({ level: "silent" })); + const client = new OpenCodeAgentClient(logger); const session = await client.createSession({ provider: "opencode", cwd: process.cwd(), + model: BIG_PICKLE_MODEL, modeId: "plan", }); @@ -50,10 +59,11 @@ describe("opencode agent commands contract (real)", () => { test("executes a slash command without arguments", async () => { expect(await isCommandAvailable("opencode")).toBe(true); - const client = new OpenCodeAgentClient(pino({ level: "silent" })); + const client = new OpenCodeAgentClient(logger); const session = await client.createSession({ provider: "opencode", cwd: process.cwd(), + model: BIG_PICKLE_MODEL, modeId: "plan", }); diff --git a/packages/server/src/server/agent/providers/opencode-agent-custom-command.real.e2e.test.ts b/packages/server/src/server/agent/providers/opencode-agent-custom-command.real.e2e.test.ts new file mode 100644 index 000000000..a18bc212f --- /dev/null +++ b/packages/server/src/server/agent/providers/opencode-agent-custom-command.real.e2e.test.ts @@ -0,0 +1,107 @@ +import { existsSync } from "node:fs"; +import { mkdir, mkdtemp, rm, rmdir, writeFile } from "node:fs/promises"; +import { homedir, tmpdir } from "node:os"; +import path from "node:path"; + +import { beforeAll, beforeEach, describe, expect, test } from "vitest"; +import pino from "pino"; + +import { isCommandAvailable } from "../../../utils/executable.js"; +import { createDaemonTestContext, type DaemonTestContext } from "../../test-utils/index.js"; +import { OpenCodeAgentClient } from "./opencode-agent.js"; +import { OpenCodeServerManager } from "./opencode/server-manager.js"; + +const COMMAND_NAME = "paseo-issue-903-big-pickle"; +const COMMAND_FILE_NAME = `${COMMAND_NAME}.md`; +const BIG_PICKLE_MODEL = "opencode/big-pickle"; +const EXPECTED_RESPONSE = "PASEO_ISSUE_903_BIG_PICKLE_OK"; + +describe("opencode custom command Big Pickle E2E (real)", () => { + let canRun = false; + + beforeAll(async () => { + canRun = await isCommandAvailable("opencode"); + }); + + beforeEach((context) => { + if (!canRun) { + context.skip(); + } + }); + + test("executes a global custom command through Paseo using Big Pickle", async () => { + const commandDir = path.join(homedir(), ".config", "opencode", "command"); + const commandFile = path.join(commandDir, COMMAND_FILE_NAME); + const commandDirExisted = existsSync(commandDir); + if (existsSync(commandFile)) { + throw new Error(`Refusing to overwrite existing OpenCode command file: ${commandFile}`); + } + + const projectDir = await mkdtemp(path.join(tmpdir(), "paseo-opencode-big-pickle-")); + const logger = pino({ level: "silent" }); + let ctx: DaemonTestContext | undefined; + + try { + await mkdir(commandDir, { recursive: true }); + await writeFile( + commandFile, + [ + "---", + "description: Paseo issue 903 Big Pickle custom command", + "agent: build", + "---", + "", + "Reply exactly with this token and nothing else:", + EXPECTED_RESPONSE, + "", + ].join("\n"), + ); + + ctx = await createDaemonTestContext({ + logger, + agentClients: { + opencode: new OpenCodeAgentClient(logger), + }, + }); + + const agent = await ctx.client.createAgent({ + provider: "opencode", + cwd: projectDir, + model: BIG_PICKLE_MODEL, + modeId: "plan", + title: "OpenCode issue 903 Big Pickle custom command", + }); + + expect(agent.provider).toBe("opencode"); + expect(agent.model).toBe(BIG_PICKLE_MODEL); + expect(agent.status).toBe("idle"); + + const commands = await ctx.client.listCommands(agent.id); + expect(commands.error).toBeNull(); + expect(commands.commands).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + name: COMMAND_NAME, + description: "Paseo issue 903 Big Pickle custom command", + }), + ]), + ); + + await ctx.client.sendMessage(agent.id, `/${COMMAND_NAME}`); + const state = await ctx.client.waitForFinish(agent.id, 90_000); + + expect(state.status).toBe("idle"); + expect(state.error).toBeNull(); + expect(state.final?.status).toBe("idle"); + expect(state.lastMessage).toContain(EXPECTED_RESPONSE); + } finally { + await ctx?.cleanup(); + await OpenCodeServerManager.getInstance(logger).shutdown(); + await rm(commandFile, { force: true }); + if (!commandDirExisted) { + await rmdir(commandDir).catch(() => undefined); + } + await rm(projectDir, { recursive: true, force: true }); + } + }, 120_000); +}); diff --git a/packages/server/src/server/agent/providers/opencode-agent.full-access.test.ts b/packages/server/src/server/agent/providers/opencode-agent.full-access.test.ts index 3875eb0a8..75ea88441 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.full-access.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.full-access.test.ts @@ -4,7 +4,6 @@ import { createTestLogger } from "../../../test-utils/test-logger.js"; import type { AgentStreamEvent } from "../agent-sdk-types.js"; import { OpenCodeAgentClient } from "./opencode-agent.js"; import { - createEventStream, idleEvent, TestOpenCodeClient, TestOpenCodeRuntime, @@ -19,7 +18,7 @@ function mockOpenCodeClient(options: MockOpenCodeClientOptions = {}) { const runtime = new TestOpenCodeRuntime(); const openCodeClient = new TestOpenCodeClient(); openCodeClient.appAgentsResponse = { data: options.agents ?? [] }; - openCodeClient.eventStream = createEventStream(options.events ?? [idleEvent()]); + openCodeClient.sessionPromptAsyncEvents = options.events ?? [idleEvent()]; runtime.enqueueClient(openCodeClient); return { openCodeClient, runtime }; diff --git a/packages/server/src/server/agent/providers/opencode-agent.slash-command-timeout.test.ts b/packages/server/src/server/agent/providers/opencode-agent.slash-command-timeout.test.ts index 384622c32..63e2ea552 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.slash-command-timeout.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.slash-command-timeout.test.ts @@ -3,6 +3,7 @@ import { describe, expect, test } from "vitest"; import { createTestLogger } from "../../../test-utils/test-logger.js"; import { OpenCodeAgentClient } from "./opencode-agent.js"; import { + idleEvent, TestOpenCodeClient, TestOpenCodeRuntime, } from "./opencode/test-utils/test-opencode-runtime.js"; @@ -93,6 +94,41 @@ describe("OpenCodeAgentSession slash command timeout handling", () => { usage: undefined, }); }); + + test("leaves successful slash command turns open until OpenCode emits idle", async () => { + const runtime = new TestOpenCodeRuntime(); + const openCodeClient = createOpenCodeClientWithConnectedProvider(); + openCodeClient.sessionCommandEvents = []; + openCodeClient.commandListResponse = { + data: [{ name: "help", description: "Show help", hints: [] }], + }; + runtime.enqueueClient(openCodeClient); + + const client = new OpenCodeAgentClient(createTestLogger(), undefined, { runtime }); + const session = await client.createSession({ provider: "opencode", cwd: "/tmp" }); + + const runPromise = session.run("/help"); + await nextTick(); + await nextTick(); + + expect(openCodeClient.calls.sessionCommand).toHaveLength(1); + let settled = false; + void runPromise.then(() => { + settled = true; + return undefined; + }); + await nextTick(); + expect(settled).toBe(false); + + openCodeClient.emitEvent(idleEvent()); + + await expect(runPromise).resolves.toMatchObject({ + sessionId: "session-1", + finalText: "", + timeline: [], + usage: undefined, + }); + }); }); function createOpenCodeClientWithConnectedProvider(): TestOpenCodeClient { @@ -105,3 +141,7 @@ function createOpenCodeClientWithConnectedProvider(): TestOpenCodeClient { }; return openCodeClient; } + +function nextTick(): Promise { + return new Promise((resolve) => setImmediate(resolve)); +} 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 3e4e8b368..7d68c1dc9 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.test.ts @@ -1,4 +1,4 @@ -import { beforeAll, describe, expect, test, vi } from "vitest"; +import { afterAll, describe, expect, test, vi } from "vitest"; import { existsSync, mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs"; import os from "node:os"; import path from "node:path"; @@ -11,6 +11,7 @@ import { OpenCodeAgentClient, translateOpenCodeEvent, } from "./opencode-agent.js"; +import { OpenCodeServerManager } from "./opencode/server-manager.js"; import { streamSession } from "./test-utils/session-stream-adapter.js"; import { TestOpenCodeClient, @@ -33,8 +34,7 @@ function tmpCwd(): string { } } -// Dynamic model selection - will be set in beforeAll -let TEST_MODEL: string | undefined; +const TEST_MODEL = "opencode/big-pickle"; interface TurnResult { events: AgentStreamEvent[]; @@ -82,14 +82,6 @@ async function collectTurnEvents(iterator: AsyncGenerator): Pr return result; } -function createAsyncIterable(items: T[]): AsyncIterable { - return (async function* () { - for (const item of items) { - yield item; - } - })(); -} - function isBinaryInstalled(binary: string): boolean { try { const out = execFileSync("which", [binary], { encoding: "utf8" }).trim(); @@ -109,45 +101,9 @@ const hasOpenCode = isBinaryInstalled("opencode"); model: TEST_MODEL, }); - beforeAll(async () => { - const startTime = Date.now(); - logger.info("beforeAll: Starting model selection"); - - const client = new OpenCodeAgentClient(logger); - const models = await client.listModels({ cwd: os.homedir(), force: false }); - - logger.info( - { modelCount: models.length, elapsed: Date.now() - startTime }, - "beforeAll: Retrieved models", - ); - - // Prefer cheap models that support tool use (required by OpenCode agents). - // Avoid free-tier OpenRouter models — they often lack tool-use support. - const fastModel = models.find( - (m) => - m.id.includes("gpt-4.1-nano") || - m.id.includes("gpt-4.1-mini") || - m.id.includes("gpt-5-nano") || - m.id.includes("gpt-5.4-mini") || - m.id.includes("gpt-4o-mini"), - ); - - if (fastModel) { - TEST_MODEL = fastModel.id; - } else if (models.length > 0) { - // Fallback to any available model - TEST_MODEL = models[0].id; - } else { - throw new Error( - "No OpenCode models available. Please authenticate with a provider (e.g., set OPENAI_API_KEY).", - ); - } - - logger.info( - { model: TEST_MODEL, totalElapsed: Date.now() - startTime }, - "beforeAll: Selected OpenCode test model", - ); - }, 30_000); + afterAll(async () => { + await OpenCodeServerManager.getInstance(logger).shutdown(); + }); test("creates a session with valid id and provider", async () => { const cwd = tmpCwd(); @@ -200,6 +156,7 @@ const hasOpenCode = isBinaryInstalled("opencode"); // HARD ASSERT: At least one model is returned (OpenCode has connected providers) expect(models.length).toBeGreaterThan(0); + expect(models.some((model) => model.id === TEST_MODEL)).toBe(true); // HARD ASSERT: Each model has required fields with correct types for (const model of models) { @@ -214,8 +171,12 @@ const hasOpenCode = isBinaryInstalled("opencode"); expect(model.metadata).toMatchObject({ providerId: expect.any(String), modelId: expect.any(String), - contextWindowMaxTokens: expect.any(Number), }); + // contextWindowMaxTokens is upstream-provided and may be absent for some + // OpenCode-routed providers; assert the type only when present. + if (model.metadata?.contextWindowMaxTokens !== undefined) { + expect(typeof model.metadata.contextWindowMaxTokens).toBe("number"); + } } }, 60_000); @@ -623,6 +584,7 @@ describe("OpenCode adapter context-window normalization", () => { describe("OpenCode adapter startTurn error handling", () => { test("emits turn_started before live OpenCode timeline items", async () => { + const eventsGate = createTestDeferred(); const globalEvents = [ { payload: { @@ -669,10 +631,18 @@ describe("OpenCode adapter startTurn error handling", () => { ]; const fakeClient = { global: { - event: vi.fn().mockResolvedValue({ stream: createAsyncIterable(globalEvents) }), + event: vi.fn().mockResolvedValue({ + stream: (async function* () { + await eventsGate.promise; + yield* globalEvents; + })(), + }), }, session: { - promptAsync: vi.fn().mockResolvedValue({ data: {}, error: undefined }), + promptAsync: vi.fn().mockImplementation(async () => { + eventsGate.resolve(); + return { data: {}, error: undefined }; + }), }, } as never; @@ -698,6 +668,7 @@ describe("OpenCode adapter startTurn error handling", () => { }); test("unwraps OpenCode global event payloads during a turn", async () => { + const eventsGate = createTestDeferred(); const globalEvents = [ { payload: { @@ -760,10 +731,18 @@ describe("OpenCode adapter startTurn error handling", () => { subscribe: vi.fn(), }, global: { - event: vi.fn().mockResolvedValue({ stream: createAsyncIterable(globalEvents) }), + event: vi.fn().mockResolvedValue({ + stream: (async function* () { + await eventsGate.promise; + yield* globalEvents; + })(), + }), }, session: { - promptAsync: vi.fn().mockResolvedValue({ data: {}, error: undefined }), + promptAsync: vi.fn().mockImplementation(async () => { + eventsGate.resolve(); + return { data: {}, error: undefined }; + }), }, } as never; @@ -790,11 +769,13 @@ describe("OpenCode adapter startTurn error handling", () => { test("keeps a turn active while OpenCode is retrying", async () => { vi.useFakeTimers(); + const eventsGate = createTestDeferred(); const retryStream: AsyncIterable = { [Symbol.asyncIterator]: () => { let emitted = false; return { next: async () => { + await eventsGate.promise; if (!emitted) { emitted = true; return { @@ -826,7 +807,10 @@ describe("OpenCode adapter startTurn error handling", () => { session: { abort: vi.fn().mockResolvedValue({ error: null }), update: vi.fn().mockResolvedValue({ error: null }), - promptAsync: vi.fn().mockResolvedValue({ data: {}, error: undefined }), + promptAsync: vi.fn().mockImplementation(async () => { + eventsGate.resolve(); + return { data: {}, error: undefined }; + }), }, } as never; diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index 356ba54f2..3fc5d2678 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -2267,6 +2267,9 @@ class OpenCodeAgentSession implements AgentSession { private pendingChildToolPartsBySessionId = new Map(); private selectedModelContextWindowMaxTokens: number | undefined; private releaseServer: (() => void) | null; + private eventStreamAbortController: AbortController | null = null; + private eventStreamReady: Deferred | null = null; + private closed = false; private readonly persistSession: boolean; private deletedFromProvider = false; constructor( @@ -2290,6 +2293,7 @@ class OpenCodeAgentSession implements AgentSession { this.selectedModelContextWindowMaxTokens = this.resolveConfiguredModelContextWindowMaxTokens( config.model, ); + this.startEventStream(); } get id(): string | null { @@ -2422,22 +2426,17 @@ class OpenCodeAgentSession implements AgentSession { const effectiveVariant = thinkingOptionId ?? undefined; const effectiveMode = resolveOpenCodeRuntimeAgentId(this.currentMode); + try { + await this.ensureEventStreamReady(); + } catch (error) { + if (this.abortController === turnAbortController) { + this.abortController = null; + } + throw error; + } + const turnId = this.createTurnId(); this.activeForegroundTurnId = turnId; - - // 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 }; - } this.notifySubscribers({ type: "turn_started", provider: "opencode" }, turnId); const slashCommand = await this.resolveSlashCommandInvocation(prompt); @@ -2480,12 +2479,8 @@ class OpenCodeAgentSession implements AgentSession { return { turnId }; } - // command() blocks until the server finishes processing. OpenCode's SSE - // endpoint does NOT replay past events, so if the command completes before - // our SSE reader connects, we miss `session.idle` and the turn hangs. - // Handle both success and error in the response handler as a fallback — - // finishForegroundTurn's guard prevents duplicate terminal events if the - // SSE stream already delivered the event. + // command() is only dispatch acknowledgement. OpenCode session events are + // the source of truth for when the command turn becomes idle or fails. void this.client.session .command({ sessionID: this.sessionId, @@ -2514,11 +2509,6 @@ class OpenCodeAgentSession implements AgentSession { { type: "turn_failed", provider: "opencode", error: errorMsg }, turnId, ); - } else { - this.finishForegroundTurn( - { type: "turn_completed", provider: "opencode", usage: undefined }, - turnId, - ); } return; }) @@ -2608,7 +2598,6 @@ class OpenCodeAgentSession implements AgentSession { return { turnId }; } - subscribe(callback: (event: AgentStreamEvent) => void): () => void { this.subscribers.add(callback); return () => { @@ -2616,109 +2605,115 @@ class OpenCodeAgentSession implements AgentSession { }; } + private startEventStream(): void { + void this.ensureEventStreamReady().catch((error) => { + this.logger.warn({ err: error, sessionId: this.sessionId }, "OpenCode event stream failed"); + }); + } + + private ensureEventStreamReady(): Promise { + if (this.eventStreamReady) { + return this.eventStreamReady.promise; + } + + const eventStreamAbortController = new AbortController(); + const eventStreamReady = createDeferred(); + this.eventStreamAbortController = eventStreamAbortController; + this.eventStreamReady = eventStreamReady; + void this.consumeEventStream(eventStreamAbortController, eventStreamReady).finally(() => { + if (this.eventStreamAbortController === eventStreamAbortController) { + this.eventStreamAbortController = null; + this.eventStreamReady = null; + } + }); + + return eventStreamReady.promise; + } + private async consumeEventStream( - turnId: string, - turnAbortController: AbortController, - subscriptionReady: Deferred, + eventStreamAbortController: AbortController, + eventStreamReady: Deferred, ): Promise { this.traceOpenCode("provider.opencode.subscribe.start", { - turnId, sessionId: this.sessionId, cwd: this.config.cwd, }); + let eventStreamReadyResolved = false; try { const result = await this.client.global.event({ - signal: turnAbortController.signal, + signal: eventStreamAbortController.signal, sseMaxRetryAttempts: 0, }); + eventStreamReadyResolved = true; + this.traceOpenCode("provider.opencode.subscribe.ready", { + sessionId: this.sessionId, + }); + eventStreamReady.resolve(); + let eventCount = 0; - let subscriptionReadyResolved = false; for await (const rawEvent of result.stream) { eventCount += 1; - if (!subscriptionReadyResolved) { - subscriptionReadyResolved = true; - this.traceOpenCode("provider.opencode.subscribe.ready", { - turnId, - sessionId: this.sessionId, - }); - subscriptionReady.resolve(); - } - const shouldContinue = await this.consumeOpenCodeStreamEvent({ - rawEvent, - eventCount, - turnId, - turnAbortController, - }); - if (!shouldContinue) { - return; - } + await this.consumeOpenCodeStreamEvent({ rawEvent, eventCount }); } this.traceOpenCode("provider.opencode.stream.eof", { - turnId, eventCount, - aborted: turnAbortController.signal.aborted, - stillActive: this.activeForegroundTurnId === turnId, + aborted: eventStreamAbortController.signal.aborted, + activeTurnId: this.activeForegroundTurnId, }); - if (!turnAbortController.signal.aborted && this.activeForegroundTurnId === turnId) { - this.traceOpenCode("provider.opencode.turn.fail_eof", { turnId, eventCount }); - if (!subscriptionReadyResolved) { - subscriptionReady.reject(new Error("OpenCode event stream ended before it became ready")); + if (!eventStreamAbortController.signal.aborted) { + if (!eventStreamReadyResolved) { + eventStreamReady.reject(new Error("OpenCode event stream ended before it became ready")); + } + const activeTurnId = this.activeForegroundTurnId; + if (activeTurnId) { + this.traceOpenCode("provider.opencode.turn.fail_eof", { + turnId: activeTurnId, + eventCount, + }); + this.finishForegroundTurn( + { + type: "turn_failed", + provider: "opencode", + error: "OpenCode event stream ended before the turn reached a terminal state", + }, + activeTurnId, + ); } - this.finishForegroundTurn( - { - type: "turn_failed", - provider: "opencode", - error: "OpenCode event stream ended before the turn reached a terminal state", - }, - turnId, - ); } } catch (error) { this.traceOpenCode("provider.opencode.subscribe.error", { - turnId, + turnId: this.activeForegroundTurnId ?? undefined, error: error instanceof Error ? { name: error.name, message: error.message } : String(error), }); - subscriptionReady.reject(error); - if (!turnAbortController.signal.aborted && this.activeForegroundTurnId === turnId) { + if (!eventStreamReadyResolved) { + eventStreamReady.reject(error); + } + const activeTurnId = this.activeForegroundTurnId; + if (!eventStreamAbortController.signal.aborted && activeTurnId) { this.finishForegroundTurn( { type: "turn_failed", provider: "opencode", error: toDiagnosticErrorMessage(error), }, - turnId, + activeTurnId, ); } - } 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 async consumeOpenCodeStreamEvent(params: { rawEvent: unknown; eventCount: number; - turnId: string; - turnAbortController: AbortController; - }): Promise { - const { rawEvent, eventCount, turnId, turnAbortController } = params; + }): Promise { + const { rawEvent, eventCount } = params; + const turnId = this.activeForegroundTurnId; const event = unwrapOpenCodeGlobalEvent(rawEvent); this.traceOpenCode("provider.opencode.raw_event", { - turnId, + turnId: turnId ?? undefined, n: eventCount, type: event?.type, rawType: readOpenCodeRecord(rawEvent)?.type, @@ -2727,16 +2722,15 @@ class OpenCodeAgentSession implements AgentSession { properties: event?.properties, }); if (!event) { - return true; + return; } - if (turnAbortController.signal.aborted || this.activeForegroundTurnId !== turnId) { + if (!turnId) { this.traceOpenCode("provider.opencode.event.skip", { - turnId, n: eventCount, - aborted: turnAbortController.signal.aborted, - activeTurnId: this.activeForegroundTurnId, + reason: "no_active_turn", + type: event.type, }); - return false; + return; } const translated = await this.translateEvent(event); this.traceOpenCode("provider.opencode.parsed_event", { @@ -2750,7 +2744,7 @@ class OpenCodeAgentSession implements AgentSession { for (const e of translated) { if (this.activeForegroundTurnId !== turnId) { this.traceOpenCode("provider.opencode.parsed_event.skip_active", { turnId, type: e.type }); - return false; + return; } if (e.type === "timeline" && e.item.type === "tool_call") { this.trackToolCall(e.item); @@ -2762,12 +2756,10 @@ class OpenCodeAgentSession implements AgentSession { type: terminalEvent.type, }); this.finishForegroundTurn(terminalEvent, turnId); - return false; + return; } this.notifySubscribers(e, turnId); } - - return true; } private finishForegroundTurn( @@ -2790,8 +2782,6 @@ class OpenCodeAgentSession implements AgentSession { this.runningToolCalls.clear(); } this.activeForegroundTurnId = null; - // Abort the SSE connection so the SDK tears down the underlying fetch. - this.abortController?.abort(); this.abortController = null; this.notifySubscribers(event, turnId); } @@ -2833,6 +2823,9 @@ class OpenCodeAgentSession implements AgentSession { } private notifySubscribers(event: AgentStreamEvent, turnIdOverride?: string): void { + if (this.closed) { + return; + } const turnId = turnIdOverride ?? this.activeForegroundTurnId; const tagged = turnId ? { ...event, turnId } : event; this.traceOpenCode("provider.opencode.event_emit", { @@ -3003,7 +2996,16 @@ class OpenCodeAgentSession implements AgentSession { async close(): Promise { try { + // Flip closed before clearing subscribers so any event the SDK delivers + // after the abort (between here and subscribers.clear) is swallowed by + // notifySubscribers instead of bubbling through provider-runner as an + // unhandled rejection in whichever test the daemon hops to next. + this.closed = true; this.abortController?.abort(); + this.eventStreamAbortController?.abort(); + this.eventStreamAbortController = null; + this.eventStreamReady = null; + this.subscribers.clear(); await reconcileOpenCodeSessionClose({ client: this.client, sessionId: this.sessionId, @@ -3011,7 +3013,6 @@ class OpenCodeAgentSession implements AgentSession { logger: this.logger, }); await this.deleteProviderSessionIfEphemeral(); - this.subscribers.clear(); this.activeForegroundTurnId = null; } finally { this.releaseServer?.(); diff --git a/packages/server/src/server/agent/providers/opencode-assistant-message.real.e2e.test.ts b/packages/server/src/server/agent/providers/opencode-assistant-message.real.e2e.test.ts index cdb79a40e..bf6853d07 100644 --- a/packages/server/src/server/agent/providers/opencode-assistant-message.real.e2e.test.ts +++ b/packages/server/src/server/agent/providers/opencode-assistant-message.real.e2e.test.ts @@ -1,15 +1,19 @@ -import { beforeAll, beforeEach, describe, test, expect } from "vitest"; +import { afterAll, beforeAll, beforeEach, describe, test, expect } 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 "./opencode-agent.js"; +import { OpenCodeServerManager } from "./opencode/server-manager.js"; import { isProviderAvailable } from "../../daemon-e2e/agent-configs.js"; import type { AgentStreamEvent } from "../agent-sdk-types.js"; +const BIG_PICKLE_MODEL = "opencode/big-pickle"; + describe("OpenCode assistant message", () => { let canRun = false; + const logger = pino({ level: "silent" }); beforeAll(async () => { canRun = await isProviderAvailable("opencode"); @@ -21,16 +25,19 @@ describe("OpenCode assistant message", () => { } }); + afterAll(async () => { + await OpenCodeServerManager.getInstance(logger).shutdown(); + }); + test("assistant_message appears in live stream with opencode/big-pickle", async () => { const cwd = mkdtempSync(path.join(tmpdir(), "opencode-msg-")); - const logger = pino({ level: "silent" }); const client = new OpenCodeAgentClient(logger); try { const session = await client.createSession({ provider: "opencode", cwd, - model: "opencode/big-pickle", + model: BIG_PICKLE_MODEL, modeId: "build", }); @@ -46,13 +53,14 @@ describe("OpenCode assistant message", () => { test("streamHistory returns assistant_message after a completed turn", async () => { const cwd = mkdtempSync(path.join(tmpdir(), "opencode-history-")); - const logger = pino({ level: "silent" }); const client = new OpenCodeAgentClient(logger); try { const session = await client.createSession({ provider: "opencode", cwd, + model: BIG_PICKLE_MODEL, + modeId: "build", }); const result = await session.run("Say hello back in one sentence."); diff --git a/packages/server/src/server/agent/providers/opencode-reasoning-dedup.real.e2e.test.ts b/packages/server/src/server/agent/providers/opencode-reasoning-dedup.real.e2e.test.ts index a251775a5..95a426031 100644 --- a/packages/server/src/server/agent/providers/opencode-reasoning-dedup.real.e2e.test.ts +++ b/packages/server/src/server/agent/providers/opencode-reasoning-dedup.real.e2e.test.ts @@ -1,15 +1,19 @@ -import { beforeAll, beforeEach, describe, test, expect } from "vitest"; +import { afterAll, beforeAll, beforeEach, describe, test, expect } 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 "./opencode-agent.js"; +import { OpenCodeServerManager } from "./opencode/server-manager.js"; import { isProviderAvailable } from "../../daemon-e2e/agent-configs.js"; import type { AgentStreamEvent } from "../agent-sdk-types.js"; +const BIG_PICKLE_MODEL = "opencode/big-pickle"; + describe("OpenCode reasoning dedup", () => { let canRun = false; + const logger = pino({ level: "silent" }); beforeAll(async () => { canRun = await isProviderAvailable("opencode"); @@ -21,16 +25,19 @@ describe("OpenCode reasoning dedup", () => { } }); + afterAll(async () => { + await OpenCodeServerManager.getInstance(logger).shutdown(); + }); + test("reasoning content is not duplicated as assistant_message", async () => { const cwd = mkdtempSync(path.join(tmpdir(), "opencode-reasoning-dedup-")); - const logger = pino({ level: "silent" }); const client = new OpenCodeAgentClient(logger); try { const session = await client.createSession({ provider: "opencode", cwd, - model: "opencode/gpt-5-nano", + model: BIG_PICKLE_MODEL, modeId: "build", }); diff --git a/packages/server/src/server/agent/providers/opencode-server-manager.test.ts b/packages/server/src/server/agent/providers/opencode-server-manager.test.ts index 26177a082..867188233 100644 --- a/packages/server/src/server/agent/providers/opencode-server-manager.test.ts +++ b/packages/server/src/server/agent/providers/opencode-server-manager.test.ts @@ -106,6 +106,19 @@ describe("OpenCodeServerManager generations", () => { expect(second.process.kill).toHaveBeenCalledWith("SIGTERM"); }); + test("shutdown still signals a process after an earlier kill signal if it has not exited", async () => { + const manager = createTestManager(); + const first = createGeneration(4451); + stubGenerations(manager, [first]); + + await manager.acquire({ force: false }); + first.process.killed = true; + + await manager.shutdown(); + + expect(first.process.kill).toHaveBeenCalledWith("SIGTERM"); + }); + test("repeated rotations leave zero unreferenced retired servers", async () => { const manager = createTestManager(); const first = createGeneration(4501); diff --git a/packages/server/src/server/agent/providers/opencode/server-manager.ts b/packages/server/src/server/agent/providers/opencode/server-manager.ts index 97d365c7e..1705ca5fa 100644 --- a/packages/server/src/server/agent/providers/opencode/server-manager.ts +++ b/packages/server/src/server/agent/providers/opencode/server-manager.ts @@ -277,7 +277,10 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { } private async killServer(server: OpenCodeServerGeneration): Promise { - if (server.process.killed) { + if ( + (server.process.exitCode !== null && server.process.exitCode !== undefined) || + (server.process.signalCode !== null && server.process.signalCode !== undefined) + ) { return; } const result = await terminateWithTreeKill(server.process, { diff --git a/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-runtime.ts b/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-runtime.ts index f68757592..0499dbde4 100644 --- a/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-runtime.ts +++ b/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-runtime.ts @@ -65,7 +65,7 @@ export class TestOpenCodeClient { appAgentsResponse: OpenCodeResponse = { data: [] }; commandListResponse: OpenCodeResponse = { data: [] }; - eventStream: AsyncIterable = createEventStream([idleEvent()]); + eventStream: AsyncIterable; experimentalSessionListResponse: OpenCodeResponse = { data: [] }; permissionReplyResponse: OpenCodeResponse = {}; providerListResponse: OpenCodeResponse = { data: { connected: [], all: [] } }; @@ -74,13 +74,24 @@ export class TestOpenCodeClient { questionReplyResponse: OpenCodeResponse = {}; sessionAbortResponse: OpenCodeResponse = {}; sessionCommandError: unknown = null; + sessionCommandEvents: unknown[] = [idleEvent()]; sessionCommandResponse: OpenCodeResponse = {}; sessionCreateResponse: OpenCodeResponse = { data: { id: "session-1" } }; sessionDeleteResponse: OpenCodeResponse = {}; sessionMessagesResponse: OpenCodeResponse = { data: [] }; + sessionPromptAsyncEvents: unknown[] = [idleEvent()]; sessionPromptAsyncResponse: OpenCodeResponse = {}; sessionSummarizeResponse: OpenCodeResponse = { data: {} }; sessionUpdateResponse: OpenCodeResponse = {}; + private readonly queuedEventStream = createQueuedEventStream(); + + constructor() { + this.eventStream = this.queuedEventStream.stream; + } + + emitEvent(event: unknown): void { + this.queuedEventStream.emit(event); + } asSdkClient(): OpencodeClient { return { @@ -154,6 +165,9 @@ export class TestOpenCodeClient { if (this.sessionCommandError) { throw this.sessionCommandError; } + for (const event of this.sessionCommandEvents) { + this.emitEvent(event); + } return this.sessionCommandResponse; }, create: async (parameters: unknown) => { @@ -170,6 +184,9 @@ export class TestOpenCodeClient { }, promptAsync: async (parameters: unknown) => { this.calls.sessionPromptAsync.push(parameters); + for (const event of this.sessionPromptAsyncEvents) { + this.emitEvent(event); + } return this.sessionPromptAsyncResponse; }, summarize: async (parameters: unknown) => { @@ -193,6 +210,38 @@ export function createEventStream(events: unknown[]): AsyncGenerator { })(); } +function createQueuedEventStream(): { + stream: AsyncIterable; + emit: (event: unknown) => void; +} { + const queue: unknown[] = []; + const waiters: Array<(result: IteratorResult) => void> = []; + + return { + stream: { + [Symbol.asyncIterator]: () => ({ + next: () => { + const event = queue.shift(); + if (event !== undefined) { + return Promise.resolve({ done: false, value: event }); + } + return new Promise>((resolve) => { + waiters.push(resolve); + }); + }, + }), + }, + emit: (event: unknown) => { + const waiter = waiters.shift(); + if (waiter) { + waiter({ done: false, value: event }); + return; + } + queue.push(event); + }, + }; +} + export function idleEvent(): unknown { return { type: "session.idle", diff --git a/packages/server/src/server/daemon-e2e/agent-configs.ts b/packages/server/src/server/daemon-e2e/agent-configs.ts index 3a2be3dfc..762c1d09f 100644 --- a/packages/server/src/server/daemon-e2e/agent-configs.ts +++ b/packages/server/src/server/daemon-e2e/agent-configs.ts @@ -50,7 +50,7 @@ export const agentConfigs = { }, opencode: { provider: "opencode", - model: "opencode/glm-5-free", + model: "opencode/big-pickle", modes: { full: "default", ask: "default",