From 6f6466d4cb4b61423bb6043407cb74b24d235f03 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Tue, 25 Aug 2026 15:28:06 +0200 Subject: [PATCH] Wait for authoritative provider readiness before prompting (#3821) * fix(server): trust provider connection lifecycle events Readiness now waits for the upstream connection event, and later connection events directly trigger the existing gap repair. This removes the parallel synthetic reconnection contract while preserving transport ownership and prompt gating. * test(server): model OpenCode global event envelopes --- .../providers/opencode-agent.recovery.test.ts | 1 + .../agent/providers/opencode-agent.test.ts | 44 ++++++++++++------- .../server/agent/providers/opencode-agent.ts | 14 +++--- .../providers/opencode/event-consumer.test.ts | 39 ++++++++++------ .../providers/opencode/event-consumer.ts | 17 +++---- .../test-utils/test-opencode-harness.ts | 9 +++- 6 files changed, 75 insertions(+), 49 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 f31a03738a..49993202dd 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 @@ -69,6 +69,7 @@ test("recovers missed output after EOF without failing the active turn", async ( expect(assistantTexts).toHaveLength(52); expect(assistantTexts.at(-1)).toBe("recovered output"); expect(assistantTexts).not.toContain("before boundary"); + expect(upstream.messageReads()).toBe(1); await session.close(); await fixture.manager.shutdown(); 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 f07eeda55e..154dcdb39d 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.test.ts @@ -79,11 +79,21 @@ const TEST_MODEL = "opencode/big-pickle"; function createDirectEventSource(client: OpencodeClient): OpenCodeEventSource { const listeners = new Set<(input: never) => void>(); const abort = new AbortController(); + let connected = false; void client.global .event({ signal: abort.signal, sseMaxRetryAttempts: 0 }) .then(async ({ stream }) => { for await (const event of stream) { - for (const listener of listeners) listener(event as never); + const payload = "payload" in event ? event.payload : event; + if (!connected && payload.type === "server.connected") { + connected = true; + continue; + } + for (const listener of listeners) { + listener( + ("payload" in event ? event : { directory: "/tmp/test", payload: event }) as never, + ); + } } return undefined; }); @@ -2964,7 +2974,7 @@ describe("OpenCode adapter startTurn error handling", () => { try { await session.startTurn("/summarize"); - openCode.emitEvent({ type: "reconnected" }); + openCode.emitEvent({ type: "server.connected", properties: {} }); await vi.waitFor(() => expect(countEvents(events, "turn_completed")).toBe(1)); expect(openCode.calls.sessionMessages).toHaveLength(0); @@ -3036,7 +3046,7 @@ describe("OpenCode adapter startTurn error handling", () => { return recoveredMessages; }; - openCode.emitEvent({ type: "reconnected" }); + openCode.emitEvent({ type: "server.connected", properties: {} }); await vi.waitFor(() => expect(failedRequest === "status" ? statusAttempts : messageAttempts).toBe(1), ); @@ -3073,7 +3083,7 @@ describe("OpenCode adapter startTurn error handling", () => { try { await session.startTurn("missing dispatch"); - openCode.emitEvent({ type: "reconnected" }); + openCode.emitEvent({ type: "server.connected", properties: {} }); await vi.waitFor(() => expect(countEvents(events, "turn_failed")).toBe(1)); await vi.advanceTimersByTimeAsync(5_000); @@ -3097,7 +3107,7 @@ describe("OpenCode adapter startTurn error handling", () => { try { await session.startTurn("keep live ingress moving"); - openCode.emitEvent({ type: "reconnected" }); + openCode.emitEvent({ type: "server.connected", properties: {} }); await vi.waitFor(() => expect(openCode.calls.sessionStatus).toHaveLength(1)); openCode.emitEvent({ type: "session.idle", properties: { sessionID: session.id } }); await vi.waitFor(() => expect(countEvents(events, "turn_completed")).toBe(1)); @@ -3135,7 +3145,7 @@ describe("OpenCode adapter startTurn error handling", () => { try { await session.startTurn("first turn"); - openCode.emitEvent({ type: "reconnected" }); + openCode.emitEvent({ type: "server.connected", properties: {} }); await firstStatus.promise; openCode.emitEvent({ type: "session.idle", properties: { sessionID: session.id } }); await firstCompletion.promise; @@ -3177,7 +3187,7 @@ describe("OpenCode adapter startTurn error handling", () => { try { await session.startTurn("first turn"); - openCode.emitEvent({ type: "reconnected" }); + openCode.emitEvent({ type: "server.connected", properties: {} }); await vi.waitFor(() => expect(statusAttempts).toBe(1)); await vi.advanceTimersByTimeAsync(100); await retryStarted.promise; @@ -3216,12 +3226,12 @@ describe("OpenCode adapter startTurn error handling", () => { await vi.waitFor(() => expect(session.getPendingPermissions()).toHaveLength(1)); openCode.permissionListResponse = { error: new Error("snapshot failed") }; - openCode.emitEvent({ type: "reconnected" }); + openCode.emitEvent({ type: "server.connected", properties: {} }); await vi.waitFor(() => expect(openCode.calls.permissionList).toHaveLength(1)); expect(session.getPendingPermissions()).toHaveLength(1); openCode.permissionListResponse = { data: [] }; - openCode.emitEvent({ type: "reconnected" }); + openCode.emitEvent({ type: "server.connected", properties: {} }); await vi.waitFor(() => expect(session.getPendingPermissions()).toHaveLength(0)); expect(events).toContainEqual( expect.objectContaining({ type: "permission_resolved", requestId: "permission-1" }), @@ -3316,7 +3326,7 @@ describe("OpenCode adapter startTurn error handling", () => { ], }; - openCode.emitEvent({ type: "reconnected" }); + openCode.emitEvent({ type: "server.connected", properties: {} }); await vi.waitFor(() => expect(countChildStatuses(events, recoveryCase.status, childId)).toBeGreaterThanOrEqual(1), ); @@ -3338,7 +3348,7 @@ describe("OpenCode adapter startTurn error handling", () => { openCode.permissionListResponse = { error: new Error("permission snapshot failed") }; openCode.questionListResponse = { error: new Error("question snapshot failed") }; - openCode.emitEvent({ type: "reconnected" }); + openCode.emitEvent({ type: "server.connected", properties: {} }); await vi.waitFor(() => expect(openCode.calls.permissionList.length).toBeGreaterThan(1)); expect(parent.getPendingPermissions()).toEqual([]); @@ -3361,13 +3371,13 @@ describe("OpenCode adapter startTurn error handling", () => { }, }); await vi.waitFor(() => expect(parent.getPendingPermissions()).toHaveLength(2)); - openCode.emitEvent({ type: "reconnected" }); + openCode.emitEvent({ type: "server.connected", properties: {} }); await vi.waitFor(() => expect(openCode.calls.permissionList.length).toBeGreaterThan(2)); expect(parent.getPendingPermissions()).toHaveLength(2); openCode.permissionListResponse = { data: [] }; openCode.questionListResponse = { data: [] }; - openCode.emitEvent({ type: "reconnected" }); + openCode.emitEvent({ type: "server.connected", properties: {} }); await vi.waitFor(() => expect(parent.getPendingPermissions()).toHaveLength(0)); await parent.close(); } @@ -3411,7 +3421,7 @@ describe("OpenCode adapter startTurn error handling", () => { ], }; - openCode.emitEvent({ type: "reconnected" }); + openCode.emitEvent({ type: "server.connected", properties: {} }); await vi.waitFor(() => expect(countChildStatuses(events, "failed", childId)).toBe(1)); const recoveredStatuses = events.flatMap((event) => event.type === "provider_subagent" && @@ -3470,7 +3480,7 @@ describe("OpenCode adapter startTurn error handling", () => { throw new Error("child messages unavailable"); }; - openCode.emitEvent({ type: "reconnected" }); + openCode.emitEvent({ type: "server.connected", properties: {} }); await vi.waitFor(() => expect(openCode.calls.sessionMessages).toHaveLength(1)); const recoveredStatuses = events.flatMap((event) => event.type === "provider_subagent" && @@ -3564,7 +3574,7 @@ describe("OpenCode adapter startTurn error handling", () => { try { const dispatch = session.startTurn("wait for transport"); 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", + "OpenCode server.connected event; your message was not sent. opencode-stream attempt=2 phase=first-record elapsedMs=45000 lastOutcome=watchdog", ); let settled = false; void dispatch.then( @@ -3709,7 +3719,7 @@ describe("OpenCode adapter startTurn error handling", () => { await session.startTurn("keep running"); } - openCode.emitEvent({ type: "reconnected" }); + openCode.emitEvent({ type: "server.connected", properties: {} }); await Promise.race([ blocked.promise, new Promise((_, reject) => diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index 857c24a365..3d64c891e1 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -3853,11 +3853,11 @@ class OpenCodeAgentSession implements AgentSession { await withTimeout( this.events.ready(), OPENCODE_EVENT_STREAM_READY_TIMEOUT_MS, - "OpenCode event stream first record", + "OpenCode server.connected event", ); } catch (error) { if (this.abortController === turnAbortController) this.abortController = null; - if (!(error instanceof Error) || error.message !== "OpenCode event stream first record") { + if (!(error instanceof Error) || error.message !== "OpenCode server.connected event") { throw error; } const diagnostics = this.events.diagnostics?.(); @@ -4162,11 +4162,7 @@ class OpenCodeAgentSession implements AgentSession { } private async consumeEventSourceInput(input: OpenCodeEventSourceInput): Promise { - if ("type" in input && input.type === "reconnected") { - await this.reconcileAfterGap(++this.gapRepairRevision); - return; - } - if ("type" in input && input.type === "server-exited") { + if (!("payload" in input)) { if (this.turnState.status === "stopping") return this.finishStoppingTurn(this.turnState.stop); const turnId = this.activeForegroundTurnId; if (turnId) { @@ -4177,6 +4173,10 @@ class OpenCodeAgentSession implements AgentSession { } return; } + if (input.payload.type === "server.connected") { + await this.reconcileAfterGap(++this.gapRepairRevision); + return; + } await this.consumeOpenCodeStreamEvent({ rawEvent: input, eventCount: 0 }); } 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 ff0675cfd1..c5a63bcb18 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 @@ -45,7 +45,7 @@ describe("OpenCodeEventConsumer", () => { await expect(closePromise).resolves.toBeUndefined(); }); - test("becomes ready on the first record and reconnects once after EOF", async () => { + test("waits for server.connected and reconnects once after EOF", async () => { const upstream = await createSseUpstream(); const timing = new ControlledTiming(); const processExit = deferred(); @@ -64,9 +64,12 @@ describe("OpenCodeEventConsumer", () => { await upstream.connected(1); expect(await promiseState(consumer.ready())).toBe("pending"); + upstream.send(0, arbitraryRecord("/one")); + await eventually(() => expect(inputs).toEqual([arbitraryRecord("/one")])); + expect(await promiseState(consumer.ready())).toBe("pending"); upstream.send(0, connectedRecord("/one")); await consumer.ready(); - expect(inputs).toEqual([connectedRecord("/one")]); + expect(inputs).toEqual([arbitraryRecord("/one")]); upstream.end(0); await timing.waiting(); @@ -74,12 +77,8 @@ describe("OpenCodeEventConsumer", () => { timing.advanceWait(); await upstream.connected(2); upstream.send(1, connectedRecord("/two")); - await eventually(() => expect(inputs).toHaveLength(3)); - expect(inputs).toEqual([ - connectedRecord("/one"), - { type: "reconnected" }, - connectedRecord("/two"), - ]); + await eventually(() => expect(inputs).toHaveLength(2)); + expect(inputs).toEqual([arbitraryRecord("/one"), connectedRecord("/two")]); expect(upstream.requests).toHaveLength(2); expect(upstream.requests.map((request) => request.url)).toEqual([ "/global/event", @@ -115,8 +114,8 @@ describe("OpenCodeEventConsumer", () => { timing.advanceWait(); await upstream.connected(2); upstream.send(1, connectedRecord("/two")); - await eventually(() => expect(inputs).toHaveLength(3)); - expect(inputs[1]).toEqual({ type: "reconnected" }); + await eventually(() => expect(inputs).toHaveLength(1)); + expect(inputs[0]).toEqual(connectedRecord("/two")); }); test("retries a first-record watchdog and exposes the recovery attempt", async () => { @@ -181,8 +180,8 @@ describe("OpenCodeEventConsumer", () => { await consumer.ready(); processExit.resolve(new Error("process exited")); - await eventually(() => expect(inputs).toHaveLength(2)); - expect(inputs[1]).toMatchObject({ type: "server-exited" }); + await eventually(() => expect(inputs).toHaveLength(1)); + expect(inputs[0]).toMatchObject({ type: "server-exited" }); expect(upstream.requests).toHaveLength(1); }); @@ -245,7 +244,9 @@ describe("OpenCodeEventConsumer", () => { await upstream.connected(1); upstream.send(0, connectedRecord("/one")); await consumer.ready(); - expect(inputs).toEqual([connectedRecord("/one")]); + upstream.send(0, arbitraryRecord("/one")); + await eventually(() => expect(inputs).toHaveLength(1)); + expect(inputs).toEqual([arbitraryRecord("/one")]); }); test("reconnects after a socket error with a delivered-record backoff reset", async () => { @@ -389,7 +390,7 @@ describe("OpenCodeEventConsumer", () => { await consumer.close(); - expect(inputs).toEqual([connectedRecord("/one")]); + expect(inputs).toEqual([]); }); }); @@ -438,6 +439,16 @@ function createRecordingLogger(): Pick { }; } +function arbitraryRecord(directory: string) { + return { + directory, + payload: { + type: "session.status", + properties: { sessionID: "unrelated", status: { type: "idle" } }, + }, + }; +} + 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 725eba774e..7394a3f628 100644 --- a/packages/server/src/server/agent/providers/opencode/event-consumer.ts +++ b/packages/server/src/server/agent/providers/opencode/event-consumer.ts @@ -5,10 +5,7 @@ import { } from "@opencode-ai/sdk/v2/client"; import type { Logger } from "pino"; -export type OpenCodeEventSourceInput = - | GlobalEvent - | { type: "reconnected" } - | { type: "server-exited"; error: Error }; +export type OpenCodeEventSourceInput = GlobalEvent | { type: "server-exited"; error: Error }; export interface OpenCodeEventSource { ready(): Promise; @@ -181,14 +178,14 @@ export class OpenCodeEventConsumer implements OpenCodeEventSource { return { delivered, phase, outcome: "ended" }; } armWatchdog(); - if (!delivered) { - delivered = true; - if (this.connected) this.publish({ type: "reconnected" }); - this.connected = true; - this.resolveReady(); - } + delivered = true; phase = "stream"; this.phase = phase; + if (!this.connected && event.payload.type === "server.connected") { + this.connected = true; + this.resolveReady(); + continue; + } this.publish(event); } let outcome: OpenCodeConnectionOutcome = "ended"; diff --git a/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts b/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts index 444a5a07b2..59c3f9d695 100644 --- a/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts +++ b/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts @@ -31,7 +31,14 @@ export class TestOpenCodeHarness implements OpenCodeServerManagerLike { enqueueClient(client: TestOpenCodeClient): void { client.observeEvents((event) => { - for (const listener of this.eventListeners) listener(event as OpenCodeEventSourceInput); + const input = + typeof event === "object" && + event !== null && + "type" in event && + event.type === "server-exited" + ? (event as OpenCodeEventSourceInput) + : ({ directory: "/workspace/repo", payload: event } as OpenCodeEventSourceInput); + for (const listener of this.eventListeners) listener(input); }); this.clients.push(client); }