From 49d5625c1e401c0dd9ca44b0d674363cf4edbe54 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sat, 5 Sep 2026 14:25:00 +1000 Subject: [PATCH] fix(server): fail V2 turns when the OpenCode event stream ends --- .../Adapters/OpenCodeAdapterV2.test.ts | 162 ++++++++++++++++++ .../Adapters/OpenCodeAdapterV2.ts | 38 +++- 2 files changed, 197 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.test.ts index efe9a24c7663..695e726a9ba2 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.test.ts +++ b/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(); + const summarizeResult = promiseGate<{ data: boolean }>(); + const baseClock = yield* Clock.Clock; + const eofClockRead = yield* Deferred.make(); + const releaseEofClockRead = yield* Deferred.make(); + 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(); + const releaseStartClockRead = yield* Deferred.make(); + 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, diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts index d01839006bd7..63e43210248d 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts +++ b/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(); + const events = yield* Queue.unbounded(); + let nativeStreamFailure: OrchestrationV2ProviderFailure | null = null; const threads = new Map(); const pendingRequests = new Map(); const pendingRequestsByNativeId = new Map(); @@ -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"),