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 8e4cdd9f6e..f07eeda55e 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.test.ts @@ -30,6 +30,7 @@ import type { // guard: deriving the boundary from OPENCODE_SERVER_STARTUP_TIMEOUT_MS would keep the // readiness tests green if that constant were shortened below the required budget. const EXPECTED_STARTUP_BUDGET_MS = 30_000; +const EXPECTED_EVENT_READINESS_BUDGET_MS = 45_000; function tmpCwd(): string { const dir = mkdtempSync(path.join(os.tmpdir(), "opencode-agent-test-")); @@ -3540,7 +3541,7 @@ describe("OpenCode adapter startTurn error handling", () => { await session.close(); }); - test("bounds dispatch readiness at the server startup budget without sending a prompt", async () => { + test("bounds dispatch readiness without sending a prompt", async () => { vi.useFakeTimers(); const openCode = new TestOpenCodeClient(); const session = new __openCodeInternals.OpenCodeAgentSession( @@ -3552,20 +3553,32 @@ describe("OpenCode adapter startTurn error handling", () => { { ready: () => new Promise(() => undefined), subscribe: () => () => undefined, + diagnostics: () => ({ + attempt: 2, + phase: "first-record", + elapsedMs: EXPECTED_EVENT_READINESS_BUDGET_MS, + lastOutcome: "watchdog", + }), }, ); try { const dispatch = session.startTurn("wait for transport"); - const rejection = expect(dispatch).rejects.toThrow("OpenCode event stream first record"); + const rejection = expect(dispatch).rejects.toThrow( + "OpenCode event stream first record; your message was not sent. opencode-stream attempt=2 phase=first-record elapsedMs=45000 lastOutcome=watchdog", + ); let settled = false; - const markSettled = () => { - settled = true; - }; - void dispatch.then(markSettled, markSettled); + void dispatch.then( + () => { + settled = true; + return undefined; + }, + () => { + settled = true; + return undefined; + }, + ); - await vi.advanceTimersByTimeAsync(EXPECTED_STARTUP_BUDGET_MS - 1); - // Still waiting one tick before the budget: a shorter readiness timeout would have - // already failed the dispatch here. + await vi.advanceTimersByTimeAsync(EXPECTED_EVENT_READINESS_BUDGET_MS - 1); expect(settled).toBe(false); expect(openCode.calls.sessionPromptAsync).toHaveLength(0); @@ -3597,8 +3610,6 @@ describe("OpenCode adapter startTurn error handling", () => { ); try { const dispatch = session.startTurn("wait for a slow transport"); - // OpenCode installs with plugins have been observed taking well past ten seconds - // to emit their first SSE record. await vi.advanceTimersByTimeAsync(20_000); expect(openCode.calls.sessionPromptAsync).toHaveLength(0); @@ -3613,6 +3624,36 @@ describe("OpenCode adapter startTurn error handling", () => { } }); + test("allows the shared stream to recover once after its thirty-second watchdog", async () => { + vi.useFakeTimers(); + const openCode = new TestOpenCodeClient(); + openCode.sessionPromptAsyncEvents = []; + const streamReady = createTestDeferred(); + const session = new __openCodeInternals.OpenCodeAgentSession( + { provider: "opencode", cwd: "/workspace/repo" }, + openCode.asSdkClient(), + "ses_readiness_retry", + createTestLogger(), + new Map(), + { + ready: () => streamReady.promise, + subscribe: () => () => undefined, + }, + ); + try { + const dispatch = session.startTurn("wait for the producer retry"); + await vi.advanceTimersByTimeAsync(35_000); + streamReady.resolve(); + await dispatch; + await vi.advanceTimersByTimeAsync(0); + expect(openCode.calls.sessionPromptAsync).toHaveLength(1); + } finally { + vi.useRealTimers(); + streamReady.resolve(); + await session.close(); + } + }); + test("close aborts every in-flight reconciliation request class", async () => { const cases = [ "child discovery", diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index e48266a8b3..857c24a365 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -76,6 +76,7 @@ import { withTimeout } from "../../../utils/promise-timeout.js"; import { execCommand } from "../../../utils/spawn.js"; import { mapOpencodeToolCall } from "./opencode/tool-call-mapper.js"; import { + OPENCODE_EVENT_STREAM_READY_TIMEOUT_MS, OpenCodeServerManager, OPENCODE_SERVER_STARTUP_TIMEOUT_MS, type OpenCodeServerAcquisition, @@ -83,6 +84,7 @@ import { } from "./opencode/server-manager.js"; import { OpenCodeEventConsumer, + type OpenCodeEventStreamDiagnostics, type OpenCodeEventSource, type OpenCodeEventSourceInput, } from "./opencode/event-consumer.js"; @@ -112,6 +114,17 @@ import { type OpenCodeProviderOptions, } from "./opencode/options.js"; +function formatOpenCodeEventStreamDiagnostics(diagnostics: OpenCodeEventStreamDiagnostics): string { + return [ + "opencode-stream", + `attempt=${diagnostics.attempt}`, + `phase=${diagnostics.phase}`, + `elapsedMs=${diagnostics.elapsedMs}`, + ...(diagnostics.lastOutcome ? [`lastOutcome=${diagnostics.lastOutcome}`] : []), + ...(diagnostics.lastError ? [`lastError=${JSON.stringify(diagnostics.lastError)}`] : []), + ].join(" "); +} + const OPENCODE_CAPABILITIES: AgentCapabilityFlags = { supportsStreaming: true, supportsSessionPersistence: true, @@ -1370,10 +1383,11 @@ export class OpenCodeAgentClient implements AgentClient { OpenCodeServerManager.getInstance(this.logger, runtimeSettings, { managedProcesses: deps.managedProcesses, resolveHomeDir: deps.resolveHomeDir, - createEventSource: ({ serverUrl, processExit }) => + createEventSource: ({ serverUrl, processExit, logger: eventLogger }) => new OpenCodeEventConsumer({ serverUrl, processExit, + logger: eventLogger, createClient: (baseUrl) => this.createOpenCodeClient({ baseUrl, directory: "" }), }), }); @@ -3654,21 +3668,7 @@ class OpenCodeAgentSession implements AgentSession { const effectiveVariant = thinkingOptionId ?? undefined; const effectiveMode = resolveOpenCodeRuntimeAgentId(this.currentMode); - try { - // The stream cannot deliver its first record before OpenCode finished booting, so - // this wait gets the same budget as server startup instead of a shorter one that - // fails turns on slow (plugin-heavy or cold) starts. - await withTimeout( - this.events.ready(), - OPENCODE_SERVER_STARTUP_TIMEOUT_MS, - "OpenCode event stream first record", - ); - } catch (error) { - if (this.abortController === turnAbortController) { - this.abortController = null; - } - throw error; - } + await this.awaitEventStreamReady(turnAbortController); const turnId = this.createTurnId(); this.materializedParts.clear(); @@ -3848,6 +3848,27 @@ class OpenCodeAgentSession implements AgentSession { return { turnId }; } + private async awaitEventStreamReady(turnAbortController: AbortController): Promise { + try { + await withTimeout( + this.events.ready(), + OPENCODE_EVENT_STREAM_READY_TIMEOUT_MS, + "OpenCode event stream first record", + ); + } catch (error) { + if (this.abortController === turnAbortController) this.abortController = null; + if (!(error instanceof Error) || error.message !== "OpenCode event stream first record") { + throw error; + } + const diagnostics = this.events.diagnostics?.(); + if (!diagnostics) throw error; + throw new Error( + `${error.message}; your message was not sent. ${formatOpenCodeEventStreamDiagnostics(diagnostics)}`, + { cause: error }, + ); + } + } + private rethrowRunnerWaitError(error: unknown): never { if (this.closed) throw new Error("OpenCode session is closed", { cause: error }); throw error; diff --git a/packages/server/src/server/agent/providers/opencode/event-consumer.test.ts b/packages/server/src/server/agent/providers/opencode/event-consumer.test.ts index a6e572a7e2..ff0675cfd1 100644 --- a/packages/server/src/server/agent/providers/opencode/event-consumer.test.ts +++ b/packages/server/src/server/agent/providers/opencode/event-consumer.test.ts @@ -1,6 +1,7 @@ import { createServer, type ServerResponse } from "node:http"; import { createOpencodeClient, type OpencodeClient } from "@opencode-ai/sdk/v2/client"; -import { afterEach, describe, expect, test } from "vitest"; +import type { Logger } from "pino"; +import { afterEach, describe, expect, test, vi } from "vitest"; import { OpenCodeEventConsumer, @@ -8,6 +9,8 @@ import { type OpenCodeEventSourceInput, } from "./event-consumer.js"; +const EXPECTED_STREAM_WATCHDOG_MS = 30_000; + describe("OpenCodeEventConsumer", () => { const cleanups: Array<() => Promise> = []; afterEach(async () => { @@ -33,6 +36,7 @@ describe("OpenCodeEventConsumer", () => { consumer = new OpenCodeEventConsumer({ serverUrl: upstream.url, processExit: new Promise(() => undefined), + logger: createRecordingLogger(), timing, }); cleanups.push(upstream.close); @@ -48,6 +52,7 @@ describe("OpenCodeEventConsumer", () => { const consumer = new OpenCodeEventConsumer({ serverUrl: upstream.url, processExit: processExit.promise, + logger: createRecordingLogger(), timing, }); cleanups.push(async () => { @@ -88,6 +93,7 @@ describe("OpenCodeEventConsumer", () => { const consumer = new OpenCodeEventConsumer({ serverUrl: upstream.url, processExit: new Promise(() => undefined), + logger: createRecordingLogger(), timing, }); cleanups.push(async () => { @@ -101,7 +107,10 @@ describe("OpenCodeEventConsumer", () => { await consumer.ready(); timing.expireWatchdog(); - expect(timing.watchdogDelays).toEqual([30_000, 30_000]); + expect(timing.watchdogDelays).toEqual([ + EXPECTED_STREAM_WATCHDOG_MS, + EXPECTED_STREAM_WATCHDOG_MS, + ]); await timing.waiting(); timing.advanceWait(); await upstream.connected(2); @@ -110,6 +119,47 @@ describe("OpenCodeEventConsumer", () => { expect(inputs[1]).toEqual({ type: "reconnected" }); }); + test("retries a first-record watchdog and exposes the recovery attempt", async () => { + const upstream = await createSseUpstream(); + const timing = new ControlledTiming(); + const logger = createRecordingLogger(); + const consumer = new OpenCodeEventConsumer({ + serverUrl: upstream.url, + processExit: new Promise(() => undefined), + logger, + timing, + }); + cleanups.push(async () => { + await consumer.close(); + await upstream.close(); + }); + + await upstream.connected(1); + timing.expireWatchdog(); + await timing.waiting(); + expect(consumer.diagnostics()).toMatchObject({ + attempt: 1, + phase: "first-record", + lastOutcome: "watchdog", + lastError: "OpenCode event stream first-record watchdog expired", + }); + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ + attempt: 1, + phase: "first-record", + outcome: "watchdog", + everReady: false, + }), + "OpenCode event stream connection failed; retrying", + ); + + timing.advanceWait(); + await upstream.connected(2); + upstream.send(1, connectedRecord("/recovered")); + await consumer.ready(); + expect(consumer.diagnostics()).toMatchObject({ attempt: 2, phase: "stream" }); + }); + test("publishes one terminal and stops reconnecting on process exit", async () => { const upstream = await createSseUpstream(); const timing = new ControlledTiming(); @@ -117,6 +167,7 @@ describe("OpenCodeEventConsumer", () => { const consumer = new OpenCodeEventConsumer({ serverUrl: upstream.url, processExit: processExit.promise, + logger: createRecordingLogger(), timing, }); cleanups.push(async () => { @@ -138,9 +189,11 @@ describe("OpenCodeEventConsumer", () => { test("backs off repeated zero-record EOFs exponentially up to the cap", async () => { const upstream = await createSseUpstream(); const timing = new ControlledTiming(); + const logger = createRecordingLogger(); const consumer = new OpenCodeEventConsumer({ serverUrl: upstream.url, processExit: new Promise(() => undefined), + logger, timing, }); cleanups.push(async () => { @@ -156,6 +209,18 @@ describe("OpenCodeEventConsumer", () => { } expect(timing.waitDelays).toEqual([100, 200, 400, 800, 1_600, 3_200, 5_000, 5_000]); + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ + phase: "first-record", + outcome: "ended", + attempt: 4, + consecutiveFailures: 4, + elapsedMs: expect.any(Number), + retryDelayMs: 800, + everReady: false, + }), + "OpenCode event stream connection failed; retrying", + ); }); test("isolates a throwing subscriber from the shared transport", async () => { @@ -164,6 +229,7 @@ describe("OpenCodeEventConsumer", () => { const consumer = new OpenCodeEventConsumer({ serverUrl: upstream.url, processExit: new Promise(() => undefined), + logger: createRecordingLogger(), timing, }); cleanups.push(async () => { @@ -188,6 +254,7 @@ describe("OpenCodeEventConsumer", () => { const consumer = new OpenCodeEventConsumer({ serverUrl: upstream.url, processExit: new Promise(() => undefined), + logger: createRecordingLogger(), timing, }); cleanups.push(async () => { @@ -209,6 +276,7 @@ describe("OpenCodeEventConsumer", () => { const consumer = new OpenCodeEventConsumer({ serverUrl: upstream.url, processExit: new Promise(() => undefined), + logger: createRecordingLogger(), timing, }); cleanups.push(async () => { @@ -236,6 +304,7 @@ describe("OpenCodeEventConsumer", () => { const consumer = new OpenCodeEventConsumer({ serverUrl: upstream.url, processExit: new Promise(() => undefined), + logger: createRecordingLogger(), createClient: () => ({ ...realClient, @@ -257,12 +326,38 @@ describe("OpenCodeEventConsumer", () => { expect(requestOptions).toMatchObject({ sseMaxRetryAttempts: 0 }); }); + test("retains the SDK error behind a zero-record stream", async () => { + const upstream = await createSseUpstream(); + upstream.failNext(1); + const timing = new ControlledTiming(); + const logger = createRecordingLogger(); + const consumer = new OpenCodeEventConsumer({ + serverUrl: upstream.url, + processExit: new Promise(() => undefined), + logger, + timing, + }); + cleanups.push(async () => { + await consumer.close(); + await upstream.close(); + }); + + await timing.waiting(); + expect(consumer.diagnostics()).toMatchObject({ + attempt: 1, + phase: "first-record", + lastOutcome: "error", + lastError: "SSE failed: 503 Service Unavailable", + }); + }); + test("rejects readiness and publishes terminal when the process exits before a record", async () => { const upstream = await createSseUpstream(); const processExit = deferred(); const consumer = new OpenCodeEventConsumer({ serverUrl: upstream.url, processExit: processExit.promise, + logger: createRecordingLogger(), }); cleanups.push(async () => { await consumer.close(); @@ -283,6 +378,7 @@ describe("OpenCodeEventConsumer", () => { const consumer = new OpenCodeEventConsumer({ serverUrl: upstream.url, processExit: new Promise(() => undefined), + logger: createRecordingLogger(), }); cleanups.push(upstream.close); const inputs: OpenCodeEventSourceInput[] = []; @@ -335,6 +431,13 @@ function connectedRecord(directory: string) { return { directory, payload: { type: "server.connected", properties: {} } }; } +function createRecordingLogger(): Pick { + return { + debug: vi.fn() as unknown as Logger["debug"], + warn: vi.fn() as unknown as Logger["warn"], + }; +} + async function createSseUpstream() { const responses: ServerResponse[] = []; const requests: Array<{ url: string | undefined }> = []; diff --git a/packages/server/src/server/agent/providers/opencode/event-consumer.ts b/packages/server/src/server/agent/providers/opencode/event-consumer.ts index cbce3d7a21..725eba774e 100644 --- a/packages/server/src/server/agent/providers/opencode/event-consumer.ts +++ b/packages/server/src/server/agent/providers/opencode/event-consumer.ts @@ -3,6 +3,7 @@ import { type GlobalEvent, type OpencodeClient, } from "@opencode-ai/sdk/v2/client"; +import type { Logger } from "pino"; export type OpenCodeEventSourceInput = | GlobalEvent @@ -12,6 +13,15 @@ export type OpenCodeEventSourceInput = export interface OpenCodeEventSource { ready(): Promise; subscribe(listener: (input: OpenCodeEventSourceInput) => void): () => void; + diagnostics?(): OpenCodeEventStreamDiagnostics; +} + +export interface OpenCodeEventStreamDiagnostics { + attempt: number; + phase: "first-record" | "stream"; + elapsedMs: number; + lastOutcome?: "ended" | "error" | "watchdog"; + lastError?: string; } export interface OpenCodeEventConsumerTiming { @@ -22,12 +32,24 @@ export interface OpenCodeEventConsumerTiming { export interface OpenCodeEventConsumerOptions { serverUrl: string; processExit: Promise; + logger: Pick; createClient?: (baseUrl: string) => OpencodeClient; timing?: OpenCodeEventConsumerTiming; } const WATCHDOG_MS = 30_000; const MAX_BACKOFF_MS = 5_000; +const FAILURE_WARNING_ATTEMPT = 4; + +type OpenCodeEventStreamPhase = "first-record" | "stream"; +type OpenCodeConnectionOutcome = "ended" | "error" | "watchdog"; + +interface OpenCodeConnectionResult { + delivered: boolean; + phase: OpenCodeEventStreamPhase; + outcome: OpenCodeConnectionOutcome; + error?: unknown; +} const systemTiming: OpenCodeEventConsumerTiming = { arm(delayMs, callback) { @@ -53,12 +75,18 @@ const systemTiming: OpenCodeEventConsumerTiming = { export class OpenCodeEventConsumer implements OpenCodeEventSource { private readonly listeners = new Set<(input: OpenCodeEventSourceInput) => void>(); private readonly client: OpencodeClient; + private readonly logger: Pick; private readonly timing: OpenCodeEventConsumerTiming; + private readonly startedAt = Date.now(); private readonly readyPromise: Promise; private resolveReady!: () => void; private rejectReady!: (error: Error) => void; private connectionAbort = new AbortController(); private connectionTask: Promise; + private attempt = 0; + private phase: OpenCodeEventStreamPhase = "first-record"; + private lastOutcome?: OpenCodeConnectionOutcome; + private lastError?: string; private connected = false; private closed = false; @@ -66,6 +94,7 @@ export class OpenCodeEventConsumer implements OpenCodeEventSource { this.client = options.createClient?.(options.serverUrl) ?? createOpencodeClient({ baseUrl: options.serverUrl }); + this.logger = options.logger; this.timing = options.timing ?? systemTiming; this.readyPromise = new Promise((resolve, reject) => { this.resolveReady = resolve; @@ -85,6 +114,16 @@ export class OpenCodeEventConsumer implements OpenCodeEventSource { return () => this.listeners.delete(listener); } + diagnostics(): OpenCodeEventStreamDiagnostics { + return { + attempt: this.attempt, + phase: this.phase, + elapsedMs: Date.now() - this.startedAt, + ...(this.lastOutcome === undefined ? {} : { lastOutcome: this.lastOutcome }), + ...(this.lastError === undefined ? {} : { lastError: this.lastError }), + }; + } + async close(): Promise { if (this.closed) return; this.closed = true; @@ -99,34 +138,48 @@ export class OpenCodeEventConsumer implements OpenCodeEventSource { void processExit.then((error) => this.exit(error)); let reconnectAttempt = 0; while (!this.closed) { - const delivered = await this.consumeConnection(this.connectionAbort.signal).catch( - () => false, - ); + this.attempt += 1; + this.phase = "first-record"; + const result = await this.consumeConnection(this.connectionAbort.signal); if (this.closed) return; - reconnectAttempt = delivered ? 0 : reconnectAttempt + 1; + this.lastOutcome = result.outcome; + this.lastError = result.error === undefined ? undefined : errorMessage(result.error); + reconnectAttempt = result.delivered ? 0 : reconnectAttempt + 1; const delayMs = Math.min(100 * 2 ** Math.max(0, reconnectAttempt - 1), MAX_BACKOFF_MS); + this.logConnectionFailure(result, this.attempt, reconnectAttempt, delayMs); await this.timing.wait(delayMs, this.connectionAbort.signal).catch(() => undefined); } } - private async consumeConnection(signal: AbortSignal): Promise { + private async consumeConnection(signal: AbortSignal): Promise { const requestAbort = new AbortController(); const abortRequest = () => requestAbort.abort(signal.reason); signal.addEventListener("abort", abortRequest, { once: true }); let cancelWatchdog: () => void = () => undefined; let delivered = false; + let phase: OpenCodeEventStreamPhase = "first-record"; + let watchdogPhase: OpenCodeEventStreamPhase | null = null; + let sseError: unknown; + const armWatchdog = () => { + cancelWatchdog(); + cancelWatchdog = this.timing.arm(WATCHDOG_MS, () => { + watchdogPhase = phase; + requestAbort.abort(new Error(`OpenCode event stream ${phase} watchdog expired`)); + }); + }; try { const result = await this.client.global.event({ signal: requestAbort.signal, sseMaxRetryAttempts: 0, + onSseError: (error) => { + sseError = error; + }, }); - const armWatchdog = () => { - cancelWatchdog(); - cancelWatchdog = this.timing.arm(WATCHDOG_MS, () => requestAbort.abort()); - }; armWatchdog(); for await (const event of result.stream) { - if (this.closed) return delivered; + if (this.closed) { + return { delivered, phase, outcome: "ended" }; + } armWatchdog(); if (!delivered) { delivered = true; @@ -134,9 +187,26 @@ export class OpenCodeEventConsumer implements OpenCodeEventSource { this.connected = true; this.resolveReady(); } + phase = "stream"; + this.phase = phase; this.publish(event); } - return delivered; + let outcome: OpenCodeConnectionOutcome = "ended"; + if (watchdogPhase) outcome = "watchdog"; + else if (sseError !== undefined) outcome = "error"; + return { + delivered, + phase: watchdogPhase ?? phase, + outcome, + ...(sseError === undefined ? {} : { error: sseError }), + }; + } catch (error) { + return { + delivered, + phase: watchdogPhase ?? phase, + outcome: watchdogPhase ? "watchdog" : "error", + error, + }; } finally { cancelWatchdog(); signal.removeEventListener("abort", abortRequest); @@ -144,6 +214,32 @@ export class OpenCodeEventConsumer implements OpenCodeEventSource { } } + private logConnectionFailure( + result: OpenCodeConnectionResult, + attempt: number, + consecutiveFailures: number, + retryDelayMs: number, + ): void { + const elapsedMs = Date.now() - this.startedAt; + const details = { + ...(result.error === undefined ? {} : { err: result.error }), + phase: result.phase, + outcome: result.outcome, + attempt, + consecutiveFailures, + elapsedMs, + retryDelayMs, + everReady: this.connected, + }; + const log = + result.outcome === "watchdog" || + this.connected || + consecutiveFailures >= FAILURE_WARNING_ATTEMPT + ? this.logger.warn.bind(this.logger) + : this.logger.debug.bind(this.logger); + log(details, "OpenCode event stream connection failed; retrying"); + } + private exit(error: Error): void { if (this.closed) return; this.closed = true; @@ -164,6 +260,10 @@ export class OpenCodeEventConsumer implements OpenCodeEventSource { } } +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + export type OpenCodeEventConsumerFactory = ( - options: Pick, + options: Pick, ) => OpenCodeEventConsumer; 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 e98c8fbd7e..b95c4e0b7e 100644 --- a/packages/server/src/server/agent/providers/opencode/server-manager.ts +++ b/packages/server/src/server/agent/providers/opencode/server-manager.ts @@ -21,12 +21,10 @@ import { type OpenCodeEventSource, } from "./event-consumer.js"; -/** - * Budget for an OpenCode server to become usable after spawn. Plugin-heavy installs - * routinely need well over ten seconds on a cold start, so every wait that depends on - * the server finishing its boot shares this budget. - */ +/** Budget for the OpenCode HTTP server to become usable after spawn. */ export const OPENCODE_SERVER_STARTUP_TIMEOUT_MS = 30_000; +/** One stalled SSE attempt plus enough time for the consumer's retry. */ +export const OPENCODE_EVENT_STREAM_READY_TIMEOUT_MS = 45_000; const OPENCODE_SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_MS = 5_000; const OPENCODE_SERVER_FORCE_SHUTDOWN_TIMEOUT_MS = 1_000; @@ -345,7 +343,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { refCount: 0, retired: false, ready: Promise.resolve(), - events: this.createEventSource({ serverUrl: url, processExit }), + events: this.createEventSource({ serverUrl: url, processExit, logger: this.logger }), managedProcessRecord, }; void managedProcessRecord.then((record) => { diff --git a/packages/server/src/server/daemon-e2e/opencode-slow-plugin-readiness.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/opencode-slow-plugin-readiness.real.e2e.test.ts new file mode 100644 index 0000000000..3645b018f1 --- /dev/null +++ b/packages/server/src/server/daemon-e2e/opencode-slow-plugin-readiness.real.e2e.test.ts @@ -0,0 +1,87 @@ +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { pathToFileURL } from "node:url"; + +import pino from "pino"; +import { beforeAll, beforeEach, describe, expect, test } from "vitest"; + +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 { canRunRealProvider } from "./real-provider-test-config.js"; + +const SLOW_PLUGIN_MS = 20_000; + +describe("daemon E2E (real opencode) - slow plugin readiness", () => { + let canRun = false; + + beforeAll(async () => { + canRun = await canRunRealProvider("opencode"); + }); + + beforeEach((context) => { + if (!canRun) context.skip(); + }); + + test("a fresh dedicated generation starts while a global plugin initializes slowly", async () => { + const root = mkdtempSync(path.join(tmpdir(), "paseo-opencode-slow-plugin-")); + const cwd = path.join(root, "worktree"); + const configHome = path.join(root, "config"); + const dataHome = path.join(root, "data"); + const cacheHome = path.join(root, "cache"); + const pluginPath = path.join(root, "slow-plugin.js"); + const configDir = path.join(configHome, "opencode"); + mkdirSync(cwd, { recursive: true }); + mkdirSync(configDir, { recursive: true }); + mkdirSync(dataHome, { recursive: true }); + mkdirSync(cacheHome, { recursive: true }); + writeFileSync( + pluginPath, + [ + "export default async () => {", + ` await new Promise((resolve) => setTimeout(resolve, ${SLOW_PLUGIN_MS}))`, + " return {}", + "}", + "", + ].join("\n"), + ); + writeFileSync( + path.join(configDir, "opencode.json"), + JSON.stringify({ plugin: [pathToFileURL(pluginPath).href] }, null, 2), + ); + + const logger = pino({ level: "silent" }); + const openCode = new OpenCodeAgentClient(logger, { + env: { + XDG_CONFIG_HOME: configHome, + XDG_DATA_HOME: dataHome, + XDG_CACHE_HOME: cacheHome, + OPENCODE_DISABLE_AUTO_UPDATE: "1", + }, + }); + const daemon = await createTestPaseoDaemon({ + agentClients: { opencode: openCode }, + logger, + }); + const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + await client.connect(); + await client.fetchAgents({ subscribe: { subscriptionId: "opencode-slow-plugin" } }); + + try { + const agent = await client.createAgent({ + provider: "opencode", + cwd, + title: "OpenCode slow plugin readiness regression", + model: "diagnostic/missing-model", + initialPrompt: "diagnostic first turn", + }); + const finish = await client.waitForFinish(agent.id, 90_000); + expect(finish.error).not.toContain("OpenCode event stream first record"); + } finally { + await client.close().catch(() => undefined); + await daemon.close(); + rmSync(root, { recursive: true, force: true }); + } + }, 120_000); +});