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

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

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

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

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<Error>();
@@ -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<Logger, "debug" | "warn"> {
   };
 }
 
+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 }> = [];

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<void>;
@@ -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";

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