From 54299a2e6c79558b26c97fd746bbff834825f4d3 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 24 Aug 2026 19:31:11 +0200 Subject: [PATCH] fix(server): let event-stream startup recover once (#3814) The readiness wait expired at the same boundary as the producer watchdog, so a reconnect already initiated by the event consumer could never benefit the pending turn. Keep dispatch gated, allow that retry to land, and preserve the actual SDK failure details for logs and user-visible reports. --- .../agent/providers/opencode-agent.test.ts | 63 +++++++-- .../server/agent/providers/opencode-agent.ts | 53 +++++--- .../providers/opencode/event-consumer.test.ts | 107 ++++++++++++++- .../providers/opencode/event-consumer.ts | 124 ++++++++++++++++-- .../providers/opencode/server-manager.ts | 10 +- ...ode-slow-plugin-readiness.real.e2e.test.ts | 87 ++++++++++++ 6 files changed, 397 insertions(+), 47 deletions(-) create mode 100644 packages/server/src/server/daemon-e2e/opencode-slow-plugin-readiness.real.e2e.test.ts 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); +});