fix(server): fail V2 turns when the OpenCode event stream ends

apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.test.ts

@@ -17,6 +17,7 @@ import {
   type OrchestrationV2ProviderTurn,
 } from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
+import * as Clock from "effect/Clock";
 import * as DateTime from "effect/DateTime";
 import * as Deferred from "effect/Deferred";
 import * as Fiber from "effect/Fiber";
@@ -1367,6 +1368,167 @@ describe("OpenCodeAdapterV2", () => {
     }).pipe(Effect.provide(idAllocatorLayer), Effect.scoped),
   );
 
+  it.effect("fails an active turn when the OpenCode event stream ends cleanly", () =>
+    Effect.gen(function* () {
+      const nativeEvents = asyncEventStream();
+      const harness = yield* makeOpenCodeRuntimeHarness(
+        "clean-event-eof",
+        "native-opencode-clean-event-eof",
+        {
+          event: {
+            subscribe: async (_input: unknown, options: { signal?: AbortSignal }) => {
+              options.signal?.addEventListener("abort", () => nativeEvents.close(), { once: true });
+              return { stream: nativeEvents.stream };
+            },
+          },
+          session: {
+            create: async () => ({
+              data: { id: "native-opencode-clean-event-eof", time: { created: 1, updated: 1 } },
+            }),
+            promptAsync: async () => ({ data: true }),
+          },
+        },
+      );
+      yield* harness.startTurn();
+      const terminalEvents = yield* harness.runtime.events.pipe(
+        Stream.runCollect,
+        Effect.forkScoped,
+      );
+
+      nativeEvents.close();
+      const received = Array.from(yield* Fiber.join(terminalEvents));
+      assert.isTrue(
+        received.some(
+          (event) =>
+            event.type === "provider_session.updated" && event.providerSession.status === "error",
+        ),
+      );
+      const terminal = received.find((event) => event.type === "turn.terminal");
+      assert.equal(terminal?.status, "failed");
+      assert.equal(terminal?.failure?.class, "transport_error");
+      assert.equal((yield* Effect.exit(harness.startTurn()))._tag, "Failure");
+    }).pipe(Effect.provide(idAllocatorLayer), Effect.scoped),
+  );
+
+  it.effect("fails compaction when its response races stream termination", () =>
+    Effect.gen(function* () {
+      const nativeEvents = asyncEventStream();
+      const summarizeStarted = promiseGate<void>();
+      const summarizeResult = promiseGate<{ data: boolean }>();
+      const baseClock = yield* Clock.Clock;
+      const eofClockRead = yield* Deferred.make<void>();
+      const releaseEofClockRead = yield* Deferred.make<void>();
+      let blockNextClockRead = false;
+      const blockingClock: Clock.Clock = {
+        ...baseClock,
+        currentTimeMillis: Effect.suspend(() => {
+          if (!blockNextClockRead) return baseClock.currentTimeMillis;
+          blockNextClockRead = false;
+          return Deferred.succeed(eofClockRead, undefined).pipe(
+            Effect.andThen(Deferred.await(releaseEofClockRead)),
+            Effect.andThen(baseClock.currentTimeMillis),
+          );
+        }),
+      };
+      const harness = yield* makeOpenCodeRuntimeHarness(
+        "compaction-eof-race",
+        "native-opencode-compaction-eof-race",
+        {
+          event: { subscribe: async () => ({ stream: nativeEvents.stream }) },
+          session: {
+            create: async () => ({
+              data: { id: "native-opencode-compaction-eof-race", time: { created: 1, updated: 1 } },
+            }),
+            summarize: () => {
+              summarizeStarted.resolve();
+              return summarizeResult.promise;
+            },
+          },
+        },
+      ).pipe(Effect.provideService(Clock.Clock, blockingClock));
+      const events = yield* harness.runtime.events.pipe(Stream.runCollect, Effect.forkScoped);
+      const start = yield* harness.startTurn("/compact").pipe(Effect.forkScoped);
+      yield* Effect.promise(() => summarizeStarted.promise);
+      blockNextClockRead = true;
+      nativeEvents.close();
+      yield* Deferred.await(eofClockRead);
+      summarizeResult.resolve({ data: true });
+      yield* Fiber.join(start);
+      yield* Deferred.succeed(releaseEofClockRead, undefined);
+      const received = Array.from(yield* Fiber.join(events));
+      const terminals = received.filter((event) => event.type === "turn.terminal");
+      assert.lengthOf(terminals, 1);
+      assert.equal(terminals[0]?.status, "failed");
+      assert.equal(terminals[0]?.failure?.class, "transport_error");
+      assert.isFalse(
+        received.some(
+          (event) =>
+            event.type === "turn_item.updated" &&
+            event.turnItem.type === "compaction" &&
+            event.turnItem.status === "completed",
+        ),
+      );
+    }).pipe(Effect.provide(idAllocatorLayer), Effect.scoped),
+  );
+
+  it.effect("does not register a turn after the OpenCode event stream ends", () =>
+    Effect.gen(function* () {
+      const nativeEvents = asyncEventStream();
+      let promptCalls = 0;
+      const harness = yield* makeOpenCodeRuntimeHarness(
+        "event-eof-start-race",
+        "native-opencode-event-eof-start-race",
+        {
+          event: {
+            subscribe: async (_input: unknown, options: { signal?: AbortSignal }) => {
+              options.signal?.addEventListener("abort", () => nativeEvents.close(), { once: true });
+              return { stream: nativeEvents.stream };
+            },
+          },
+          session: {
+            create: async () => ({
+              data: {
+                id: "native-opencode-event-eof-start-race",
+                time: { created: 1, updated: 1 },
+              },
+            }),
+            promptAsync: async () => {
+              promptCalls += 1;
+              return { data: true };
+            },
+          },
+        },
+      );
+      const baseClock = yield* Clock.Clock;
+      const startClockRead = yield* Deferred.make<void>();
+      const releaseStartClockRead = yield* Deferred.make<void>();
+      let blockNextClockRead = true;
+      const blockingClock: Clock.Clock = {
+        ...baseClock,
+        currentTimeMillis: Effect.suspend(() => {
+          if (!blockNextClockRead) return baseClock.currentTimeMillis;
+          blockNextClockRead = false;
+          return Deferred.succeed(startClockRead, undefined).pipe(
+            Effect.andThen(Deferred.await(releaseStartClockRead)),
+            Effect.andThen(baseClock.currentTimeMillis),
+          );
+        }),
+      };
+      const start = yield* harness
+        .startTurn()
+        .pipe(Effect.provideService(Clock.Clock, blockingClock), Effect.exit, Effect.forkScoped);
+      yield* Deferred.await(startClockRead);
+
+      const events = yield* harness.runtime.events.pipe(Stream.runCollect, Effect.forkScoped);
+      nativeEvents.close();
+      yield* Fiber.join(events);
+      yield* Deferred.succeed(releaseStartClockRead, undefined);
+
+      assert.isTrue(Exit.isFailure(yield* Fiber.join(start)));
+      assert.equal(promptCalls, 0);
+    }).pipe(Effect.provide(idAllocatorLayer), Effect.scoped),
+  );
+
   it("holds stale idle through prompt admission until the new user message is observed", () => {
     const admission = {
       admissionPending: true,

apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts

@@ -1039,7 +1039,8 @@ export function makeOpenCodeAdapterV2(options: OpenCodeAdapterV2Options): Provid
           updatedAt: now,
           lastError: null,
         };
-        const events = yield* Queue.unbounded<ProviderAdapterV2Event>();
+        const events = yield* Queue.unbounded<ProviderAdapterV2Event, Cause.Done>();
+        let nativeStreamFailure: OrchestrationV2ProviderFailure | null = null;
         const threads = new Map<string, OpenCodeThreadState>();
         const pendingRequests = new Map<string, PendingOpenCodeRequest>();
         const pendingRequestsByNativeId = new Map<string, PendingOpenCodeRequest>();
@@ -2113,6 +2114,10 @@ export function makeOpenCodeAdapterV2(options: OpenCodeAdapterV2Options): Provid
           },
         ) {
           if (turn.finalized) return;
+          if (nativeStreamFailure !== null) {
+            status = "failed";
+            terminal = { failure: nativeStreamFailure, threadDisposition: "broken" };
+          }
           turn.finalized = true;
           const completedAt = yield* DateTime.now;
           for (const part of turn.parts.values()) {
@@ -2795,6 +2800,10 @@ export function makeOpenCodeAdapterV2(options: OpenCodeAdapterV2Options): Provid
               const detail = Exit.isSuccess(exit)
                 ? "OpenCode event stream ended unexpectedly."
                 : openCodeRuntimeErrorDetail(Cause.squash(exit.cause));
+              nativeStreamFailure = makeProviderFailure({
+                message: detail,
+                class: "transport_error",
+              });
               yield* updateProviderSession("error", detail);
               for (const state of threads.values()) {
                 if (state.activeTurn !== null)
@@ -2803,6 +2812,7 @@ export function makeOpenCodeAdapterV2(options: OpenCodeAdapterV2Options): Provid
                     threadDisposition: "broken",
                   });
               }
+              yield* Queue.end(events);
             }),
           ),
           Effect.forkIn(scope),
@@ -3082,6 +3092,11 @@ export function makeOpenCodeAdapterV2(options: OpenCodeAdapterV2Options): Provid
             }),
           startTurn: (turnInput) =>
             Effect.gen(function* () {
+              if (nativeStreamFailure !== null) {
+                return yield* protocolError(
+                  "OpenCode event stream has ended; reconnect the provider session before starting another turn.",
+                );
+              }
               const sessionId = nativeThreadId(turnInput.providerThread);
               const state = threads.get(sessionId);
               if (state === undefined) {
@@ -3119,6 +3134,21 @@ export function makeOpenCodeAdapterV2(options: OpenCodeAdapterV2Options): Provid
                 startedAt,
                 completedAt: null,
               };
+              const admissionMessageId = yield* makeOpenCodeMessageId();
+              // No Effect may be yielded between this check and installing the
+              // turn. If the event stream ended while IDs were being prepared,
+              // registering afterward would leave a running turn that the EOF
+              // handler had already finished scanning.
+              if (nativeStreamFailure !== null) {
+                return yield* protocolError(
+                  "OpenCode event stream has ended; reconnect the provider session before starting another turn.",
+                );
+              }
+              if (state.activeTurn !== null) {
+                return yield* protocolError(
+                  `OpenCode provider thread ${turnInput.providerThread.id} already has an active turn`,
+                );
+              }
               const turn: ActiveOpenCodeTurn = {
                 isRoot: true,
                 threadId: turnInput.threadId,
@@ -3140,7 +3170,7 @@ export function makeOpenCodeAdapterV2(options: OpenCodeAdapterV2Options): Provid
                 providerTurn,
                 nextItemOrdinal: turnInput.providerTurnOrdinal * 100 + 1,
                 nativeUserMessageId: null,
-                admissionMessageId: yield* makeOpenCodeMessageId(),
+                admissionMessageId,
                 interrupted: false,
                 finalized: false,
                 planId: null,
@@ -3178,7 +3208,9 @@ export function makeOpenCodeAdapterV2(options: OpenCodeAdapterV2Options): Provid
                     ),
                 ).pipe(
                   Effect.tap(() =>
-                    turn.interrupted ? Effect.void : emitCompactionItem(state, turn),
+                    turn.interrupted || turn.finalized || nativeStreamFailure !== null
+                      ? Effect.void
+                      : emitCompactionItem(state, turn),
                   ),
                   Effect.tap(() =>
                     finalizeTurn(state, turn, turn.interrupted ? "interrupted" : "completed"),