From d517b3314e2cab49a21b4ada689bc52eca235ef2 Mon Sep 17 00:00:00 2001 From: James Hu Date: Sun, 4 Oct 2026 17:18:32 -0700 Subject: [PATCH] fix(opencode): wait for the server to listen before opening the event stream (#5873) * fix(opencode): wait for the server to listen before opening the event stream OpenCode can accept TCP connections before its HTTP handler is ready. Starting the event consumer at spawn lets a retry land in that window and hang without receiving server.connected. Gate the first request on the listening promise. Use a five-second watchdog before connection readiness and abandon stalled attempts independently of fetch abort, ignoring any late records. Keep the existing thirty-second watchdog after readiness. * fix(opencode): forward the listening gate from the agent client The first startup fix passed the listening gate from the server manager, but the agent client factory dropped it before building the consumer. Lab logs showed four generations retrying before listening and then hitting the first-record watchdog. Forward the gate through that factory. Exercise the client-owned manager with a real spawned process in the existing provider dispatch test, and assert that no event request arrives before listening. --- .../providers/opencode-agent.recovery.test.ts | 88 +++++++++--- .../server/agent/providers/opencode-agent.ts | 3 +- .../providers/opencode/event-consumer.test.ts | 64 ++++++++- .../providers/opencode/event-consumer.ts | 134 +++++++++++------- .../providers/opencode/server-manager.ts | 21 ++- 5 files changed, 229 insertions(+), 81 deletions(-) diff --git a/packages/server/src/server/agent/providers/opencode-agent.recovery.test.ts b/packages/server/src/server/agent/providers/opencode-agent.recovery.test.ts index d0e8e663fc..8a4f43426e 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.recovery.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.recovery.test.ts @@ -15,29 +15,63 @@ import { OpenCodeServerManager } from "./opencode/server-manager.js"; test("dispatches ascending OpenCode message identifiers through the public provider path", async () => { const upstream = await createRecoveryUpstream(); - const fixture = await createPublicRecoverySession(upstream, new RecoveryTiming()); - const observed: AgentStreamEvent[] = []; - fixture.session.subscribe((event) => observed.push(event)); - await upstream.connected(1); - upstream.send(0, connectedRecord()); - - await fixture.session.startTurn("first"); - await upstream.dispatched(1); - upstream.send(0, idleRecord()); - await eventually(() => - expect(observed.filter((event) => event.type === "turn_completed")).toHaveLength(1), + const client = new OpenCodeAgentClient( + createTestLogger(), + { + command: { + mode: "replace", + argv: [ + process.execPath, + "-e", + `setInterval(() => {}, 1000); fetch(${JSON.stringify( + `${upstream.url}/test/listening`, + )}).then(() => console.log("listening on test server"));`, + "--", + ], + }, + }, + { + resolveHomeDir: () => process.cwd(), + createClient: ({ directory }) => createOpencodeClient({ baseUrl: upstream.url, directory }), + }, ); - await fixture.session.startTurn("second"); - const ids = await upstream.dispatched(2); + try { + const creating = client.createSession({ + provider: "opencode", + cwd: "/workspace", + }); + void creating.catch(() => undefined); + await upstream.awaitingListening(); + // The process is alive but has not reported listening, so even a reachable endpoint must stay untouched. + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(upstream.connectionCount()).toBe(0); + upstream.reportListening(); + const session = await creating; + const observed: AgentStreamEvent[] = []; + session.subscribe((event) => observed.push(event)); + await upstream.connected(1); + upstream.send(0, connectedRecord()); - expect(ids).toHaveLength(2); - expect(ids[0]).toMatch(/^msg_[0-9a-f]{12}[0-9A-Za-z]{14}$/); - expect(ids[1]).toMatch(/^msg_[0-9a-f]{12}[0-9A-Za-z]{14}$/); - expect((ids[0] as string) < (ids[1] as string)).toBe(true); + await session.startTurn("first"); + await upstream.dispatched(1); + upstream.send(0, idleRecord()); + await eventually(() => + expect(observed.filter((event) => event.type === "turn_completed")).toHaveLength(1), + ); + await session.startTurn("second"); + const ids = await upstream.dispatched(2); - await fixture.session.close(); - await fixture.manager.shutdown(); - await upstream.close(); + expect(ids).toHaveLength(2); + expect(ids[0]).toMatch(/^msg_[0-9a-f]{12}[0-9A-Za-z]{14}$/); + expect(ids[1]).toMatch(/^msg_[0-9a-f]{12}[0-9A-Za-z]{14}$/); + expect((ids[0] as string) < (ids[1] as string)).toBe(true); + + await session.close(); + } finally { + upstream.reportListening(); + await client.shutdown(); + await upstream.close(); + } }); test("recovers missed output after EOF without failing the active turn", async () => { @@ -306,7 +340,10 @@ async function createPublicRecoverySession( serverProcess?.exit(); return "terminated"; }, - createEventSource: (options) => new OpenCodeEventConsumer({ ...options, timing }), + createEventSource: (options) => { + expect(options.listening).toBeInstanceOf(Promise); + return new OpenCodeEventConsumer({ ...options, timing }); + }, }); const client = new OpenCodeAgentClient(createTestLogger(), undefined, { serverManager: manager, @@ -338,6 +375,7 @@ class RecoveryServerProcess extends EventEmitter { async function createRecoveryUpstream() { const streams: ServerResponse[] = []; + let listeningResponse: ServerResponse | null = null; const dispatchIds: string[] = []; const mcpAddNames: string[] = []; let messages: unknown[] = []; @@ -345,6 +383,10 @@ async function createRecoveryUpstream() { let status: "busy" | "idle" = "idle"; const server = createServer(async (request, response) => { const pathname = new URL(request.url ?? "/", "http://127.0.0.1").pathname; + if (pathname === "/test/listening") { + listeningResponse = response; + return; + } if (request.url?.startsWith("/global/event")) { response.writeHead(200, { "content-type": "text/event-stream" }); response.flushHeaders(); @@ -378,6 +420,10 @@ async function createRecoveryUpstream() { return { port: address.port, url: `http://127.0.0.1:${address.port}`, + awaitingListening: async () => + expect.poll(() => listeningResponse, { timeout: 3_000 }).not.toBeNull(), + reportListening: () => listeningResponse?.end(), + connectionCount: () => streams.length, connected: async (count: number) => eventually(() => expect(streams).toHaveLength(count)), send(index: number, value: unknown) { streams[index]?.write(`data: ${JSON.stringify(value)}\n\n`); diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index 5b75733f33..7e49bc50d3 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -1429,10 +1429,11 @@ export class OpenCodeAgentClient implements AgentClient { OpenCodeServerManager.getInstance(this.logger, runtimeSettings, { managedProcesses: deps.managedProcesses, resolveHomeDir: deps.resolveHomeDir, - createEventSource: ({ serverUrl, processExit, logger: eventLogger }) => + createEventSource: ({ serverUrl, processExit, listening, logger: eventLogger }) => new OpenCodeEventConsumer({ serverUrl, processExit, + listening, logger: eventLogger, createClient: (baseUrl) => this.createOpenCodeClient({ baseUrl, directory: "" }), }), 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 aad9702ad0..2a0999f74e 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 @@ -49,9 +49,11 @@ describe("OpenCodeEventConsumer", () => { const upstream = await createSseUpstream(); const timing = new ControlledTiming(); const processExit = deferred(); + const listening = deferred(); const consumer = new OpenCodeEventConsumer({ serverUrl: upstream.url, processExit: processExit.promise, + listening: listening.promise, logger: createRecordingLogger(), timing, }); @@ -62,6 +64,10 @@ describe("OpenCodeEventConsumer", () => { const inputs: OpenCodeEventSourceInput[] = []; consumer.subscribe((input) => inputs.push(input)); + await new Promise((resolve) => setImmediate(resolve)); + expect(upstream.requests).toHaveLength(0); + expect(consumer.diagnostics().attempt).toBe(0); + listening.resolve(); await upstream.connected(1); expect(await promiseState(consumer.ready())).toBe("pending"); upstream.send(0, arbitraryRecord("/one")); @@ -106,10 +112,7 @@ describe("OpenCodeEventConsumer", () => { await consumer.ready(); timing.expireWatchdog(); - expect(timing.watchdogDelays).toEqual([ - EXPECTED_STREAM_WATCHDOG_MS, - EXPECTED_STREAM_WATCHDOG_MS, - ]); + expect(timing.watchdogDelays).toEqual([5_000, EXPECTED_STREAM_WATCHDOG_MS]); await timing.waiting(); timing.advanceWait(); await upstream.connected(2); @@ -120,6 +123,7 @@ describe("OpenCodeEventConsumer", () => { test("retries a first-record watchdog and exposes the recovery attempt", async () => { const upstream = await createSseUpstream(); + upstream.stallNext(1); const timing = new ControlledTiming(); const logger = createRecordingLogger(); const consumer = new OpenCodeEventConsumer({ @@ -127,6 +131,12 @@ describe("OpenCodeEventConsumer", () => { processExit: new Promise(() => undefined), logger, timing, + createClient: (baseUrl) => + createOpencodeClient({ + baseUrl, + // Keep the abandoned socket open to reproduce a fetch that does not unwind on abort. + fetch: (request) => fetch(new Request(request, { signal: null })), + }), }); cleanups.push(async () => { await consumer.close(); @@ -134,6 +144,7 @@ describe("OpenCodeEventConsumer", () => { }); await upstream.connected(1); + expect(timing.watchdogDelays).toEqual([5_000]); timing.expireWatchdog(); await timing.waiting(); expect(consumer.diagnostics()).toMatchObject({ @@ -156,7 +167,18 @@ describe("OpenCodeEventConsumer", () => { await upstream.connected(2); upstream.send(1, connectedRecord("/recovered")); await consumer.ready(); - expect(consumer.diagnostics()).toMatchObject({ attempt: 2, phase: "stream" }); + expect(consumer.diagnostics()).toMatchObject({ + attempt: 2, + phase: "stream", + }); + expect(upstream.isOpen(0)).toBe(true); + upstream.resume(0); + upstream.send(0, connectedRecord("/abandoned")); + upstream.send(0, arbitraryRecord("/abandoned")); + const inputs: OpenCodeEventSourceInput[] = []; + consumer.subscribe((input) => inputs.push(input)); + await new Promise((resolve) => setImmediate(resolve)); + expect(inputs).toEqual([]); }); test("publishes one terminal and stops reconnecting on process exit", async () => { @@ -422,6 +444,16 @@ describe("OpenCodeEventConsumer", () => { await consumer.close(); expect(inputs).toEqual([]); + + const waitingConsumer = new OpenCodeEventConsumer({ + serverUrl: upstream.url, + processExit: new Promise(() => undefined), + listening: new Promise(() => undefined), + logger: createRecordingLogger(), + }); + await waitingConsumer.close(); + await expect(waitingConsumer.ready()).rejects.toThrow("OpenCode event source closed"); + expect(upstream.requests).toHaveLength(1); }); }); @@ -499,8 +531,14 @@ async function createSseUpstream() { const responses: ServerResponse[] = []; const requests: Array<{ url: string | undefined }> = []; let failuresRemaining = 0; + let stallsRemaining = 0; const server = createServer((request, response) => { requests.push({ url: request.url }); + if (stallsRemaining > 0) { + stallsRemaining -= 1; + responses.push(response); + return; + } if (failuresRemaining > 0) { failuresRemaining -= 1; response.writeHead(503).end(); @@ -533,7 +571,21 @@ async function createSseUpstream() { failNext(count: number) { failuresRemaining = count; }, - close: async () => new Promise((resolve) => server.close(() => resolve())), + stallNext(count: number) { + stallsRemaining = count; + }, + isOpen(index: number) { + return responses[index]?.socket?.destroyed === false; + }, + resume(index: number) { + responses[index]?.writeHead(200, { "content-type": "text/event-stream" }); + responses[index]?.flushHeaders(); + }, + close: async () => + new Promise((resolve) => { + server.close(() => resolve()); + server.closeAllConnections(); + }), }; } 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 064d6d6905..9471a28745 100644 --- a/packages/server/src/server/agent/providers/opencode/event-consumer.ts +++ b/packages/server/src/server/agent/providers/opencode/event-consumer.ts @@ -29,11 +29,13 @@ export interface OpenCodeEventConsumerTiming { export interface OpenCodeEventConsumerOptions { serverUrl: string; processExit: Promise; + listening?: Promise; logger: Pick; createClient?: (baseUrl: string) => OpencodeClient; timing?: OpenCodeEventConsumerTiming; } +const FIRST_RECORD_WATCHDOG_MS = 5_000; const WATCHDOG_MS = 30_000; const MAX_BACKOFF_MS = 5_000; const FAILURE_WARNING_ATTEMPT = 4; @@ -98,7 +100,7 @@ export class OpenCodeEventConsumer implements OpenCodeEventSource { this.rejectReady = reject; }); void this.readyPromise.catch(() => undefined); - this.connectionTask = this.consume(options.processExit); + this.connectionTask = this.consume(options.processExit, options.listening); void this.connectionTask.catch(() => undefined); } @@ -131,8 +133,23 @@ export class OpenCodeEventConsumer implements OpenCodeEventSource { await this.connectionTask.catch(() => undefined); } - private async consume(processExit: Promise): Promise { + private async consume(processExit: Promise, listening?: Promise): Promise { void processExit.then((error) => this.exit(error)); + if (listening) { + const signal = this.connectionAbort.signal; + let stopWaiting!: () => void; + const stopped = new Promise((resolve) => { + stopWaiting = resolve; + }); + signal.addEventListener("abort", stopWaiting, { once: true }); + try { + await Promise.race([listening, stopped]); + } catch { + return; + } finally { + signal.removeEventListener("abort", stopWaiting); + } + } let reconnectAttempt = 0; while (!this.closed) { this.attempt += 1; @@ -150,62 +167,83 @@ export class OpenCodeEventConsumer implements OpenCodeEventSource { private async consumeConnection(signal: AbortSignal): Promise { const requestAbort = new AbortController(); - const abortRequest = () => requestAbort.abort(signal.reason); + const attemptToken = { active: true }; + let resolveAbandoned!: (result: OpenCodeConnectionResult) => void; + const abandoned = new Promise((resolve) => { + resolveAbandoned = resolve; + }); + const abortRequest = () => { + attemptToken.active = false; + resolveAbandoned({ delivered, phase, outcome: "ended" }); + 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; + cancelWatchdog = this.timing.arm( + this.connected ? WATCHDOG_MS : FIRST_RECORD_WATCHDOG_MS, + () => { + const error = new Error(`OpenCode event stream ${phase} watchdog expired`); + // Abandon a stalled request even when fetch ignores abort, for example during OpenCode startup. + attemptToken.active = false; + resolveAbandoned({ delivered, phase, outcome: "watchdog", error }); + requestAbort.abort(error); }, - }); - armWatchdog(); - for await (const event of result.stream) { - if (this.closed) { - return { delivered, phase, outcome: "ended" }; + ); + }; + const consumeStream = async (): Promise => { + try { + const result = await this.client.global.event({ + signal: requestAbort.signal, + sseMaxRetryAttempts: 0, + onSseError: (error) => { + sseError = error; + }, + }); + for await (const event of result.stream) { + // Ignore late records from abandoned requests, for example a delayed server.connected event. + if (this.closed || !attemptToken.active) { + return { delivered, phase, outcome: "ended" }; + } + delivered = true; + phase = "stream"; + this.phase = phase; + if (!this.connected && event.payload.type === "server.connected") { + this.connected = true; + this.resolveReady(); + armWatchdog(); + continue; + } + armWatchdog(); + this.logPluginFailure(event); + this.publish(event); } - armWatchdog(); - delivered = true; - phase = "stream"; - this.phase = phase; - if (!this.connected && event.payload.type === "server.connected") { - this.connected = true; - this.resolveReady(); - continue; - } - this.logPluginFailure(event); - this.publish(event); + let outcome: OpenCodeConnectionOutcome = "ended"; + if (sseError !== undefined) outcome = "error"; + return { + delivered, + phase, + outcome, + ...(sseError === undefined ? {} : { error: sseError }), + }; + } catch (error) { + return { + delivered, + phase, + outcome: "error", + error, + }; } - 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, - }; + }; + armWatchdog(); + try { + return await Promise.race([consumeStream(), abandoned]); } finally { + attemptToken.active = false; cancelWatchdog(); signal.removeEventListener("abort", abortRequest); requestAbort.abort(); @@ -285,5 +323,5 @@ function containsPluginError(error: unknown): boolean { } 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 cd560af7ba..9ff5dc6c96 100644 --- a/packages/server/src/server/agent/providers/opencode/server-manager.ts +++ b/packages/server/src/server/agent/providers/opencode/server-manager.ts @@ -356,6 +356,12 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { const processExit = new Promise((resolve) => { resolveProcessExit = resolve; }); + let resolveListening!: () => void; + let rejectListening!: (error: Error) => void; + const ready = new Promise((resolve, reject) => { + resolveListening = resolve; + rejectListening = reject; + }); const server: OpenCodeServerGeneration = { environment, process: serverProcess, @@ -364,7 +370,12 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { refCount: 0, retired: false, ready: Promise.resolve(), - events: this.createEventSource({ serverUrl: url, processExit, logger: this.logger }), + events: this.createEventSource({ + serverUrl: url, + processExit, + logger: this.logger, + listening: ready, + }), managedProcessRecord, }; this.logger.info( @@ -403,7 +414,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { return sections.join("\n"); }; - const ready = new Promise((resolve, reject) => { + { let timeout: ReturnType; const failStartup = (error: Error) => { if (settled) { @@ -411,7 +422,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { } settled = true; clearTimeout(timeout); - reject(error); + rejectListening(error); }; timeout = setTimeout(() => { if (!started) { @@ -426,7 +437,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { started = true; settled = true; clearTimeout(timeout); - resolve(); + resolveListening(); } }); @@ -462,7 +473,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { } } }); - }); + } server.ready = ready.catch(async (error) => { await this.killServer(server);