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