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);