From 49a67c607e01db017f8c62bc98bf2f0f950fdaac Mon Sep 17 00:00:00 2001 From: tinzeljar Date: Fri, 23 Oct 2026 18:12:11 +0200 Subject: [PATCH 1/4] fix(opencode): preserve owned turns during server recovery --- .../Adapters/OpenCode2AdapterV2.test.ts | 437 ++++++++++++++++++ .../Adapters/OpenCode2AdapterV2.testkit.ts | 20 +- docs/user/providers-opencode.md | 4 +- .../src/server/v2/OpenCode2Server.test.ts | 4 + .../src/server/v2/OpenCode2Server.ts | 4 +- .../src/server/v2/adapter.ts | 271 ++++++----- 6 files changed, 630 insertions(+), 110 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 15ca24556ad6..7bcf0e534d66 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -16,6 +16,7 @@ import { ProviderTurnId, RunAttemptId, RunId, + type RuntimeRequestId, ThreadId, type ModelSelection, type OrchestrationV2AppThread, @@ -32,6 +33,7 @@ import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as Exit from "effect/Exit"; @@ -309,6 +311,9 @@ const resumed = ( options?: { readonly external?: boolean; readonly supervised?: boolean; + readonly processExit?: Effect.Effect; + readonly beforeResponse?: (path: string) => Effect.Effect; + readonly borrowers?: { current: number }; readonly replayGate?: ProviderReplayGate; readonly attachmentsDir?: string; readonly fileSystem?: FileSystem.FileSystem; @@ -336,6 +341,11 @@ const resumed = ( ...(options?.fileSystem === undefined ? {} : { fileSystem: options.fileSystem }), ...(options?.onPrompt === undefined ? {} : { onPrompt: options.onPrompt }), ...(options?.external === undefined ? {} : { external: options.external }), + ...(options?.processExit === undefined ? {} : { processExit: options.processExit }), + ...(options?.beforeResponse === undefined + ? {} + : { beforeResponse: options.beforeResponse }), + ...(options?.borrowers === undefined ? {} : { borrowers: options.borrowers }), ...(options?.replayGate === undefined ? {} : { replayGate: options.replayGate }), ...(options?.attachmentsDir === undefined ? {} @@ -2170,6 +2180,433 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + it.effect("keeps owned work through a stalled reconciliation without releasing its server", () => + Effect.gen(function* () { + const reading = yield* Deferred.make(); + const resume = yield* Deferred.make(); + const borrowers = { current: 0 }; + const { runtime, thread } = yield* resumed( + [ + out("session.prompt", { sessionID: SESSION, id: PROMPT_ID, text: "" }), + promptAccepted, + ...reconnected({ [SESSION]: { type: "running" } }), + ...openRequests([]), + event("session.execution.succeeded", { sessionID: SESSION }), + ], + { + borrowers, + beforeResponse: (path) => + path.endsWith("/message") + ? Deferred.succeed(reading, undefined).pipe(Effect.andThen(Deferred.await(resume))) + : Effect.void, + }, + ); + const collected: Array = []; + const ended = yield* runtime.events.pipe( + Stream.tap((event) => Effect.sync(() => collected.push(event))), + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runDrain, + Effect.forkScoped, + ); + yield* runtime.startTurn(turnInput(thread)); + yield* Deferred.await(reading); + yield* TestClock.adjust("3 minutes"); + assert.isFalse(collected.some((event) => event.type === "turn.terminal")); + assert.equal(borrowers.current, 1); + yield* Deferred.succeed(resume, undefined); + yield* Fiber.join(ended); + assert.deepInclude(collected.at(-1), { type: "turn.terminal", status: "completed" }); + }).pipe(Effect.scoped), + ); + + it.effect("reconciles slow sessions without spending one shared deadline", () => + Effect.gen(function* () { + const otherId = "ses_other_stalled_session"; + const otherPrompt = "msg_other_stalled_prompt"; + const reads = yield* Queue.unbounded>(); + let recovering = false; + const history = (sessionID: string, id: string) => [ + out("message.list", { sessionID, order: "desc", limit: "50" }), + reply("message.list", { + data: [{ id, time: { created: 1 }, text: "hi", type: "user" }], + cursor: {}, + }), + ]; + const requests = (sessionID: string) => [ + out("permission.list", { sessionID }), + replyData("permission.list", []), + out("session.form.list", { sessionID }), + replyData("session.form.list", []), + ]; + const { runtime, thread } = yield* resumed( + [ + out("session.get", { sessionID: otherId }), + replyData("session.get", sessionInfo({ id: otherId })), + ...requests(otherId), + out("session.prompt", { sessionID: SESSION, id: PROMPT_ID, text: "" }), + promptAccepted, + out("session.instructions.entry.put", { + sessionID: otherId, + key: "t3-code", + value: "", + }), + reply("session.instructions.entry.put", null), + out("session.prompt", { sessionID: otherId, id: otherPrompt, text: "" }), + replyData("session.prompt", { + id: otherPrompt, + sessionID: otherId, + time: { created: 1 }, + type: "user", + payload: { text: "hi" }, + delivery: "steer", + }), + { type: "runtime_exit", status: "success" }, + out("event.subscribe"), + out("session.active"), + replyData("session.active", { + [SESSION]: { type: "running" }, + [otherId]: { type: "running" }, + }), + ...history(SESSION, PROMPT_ID), + ...history(otherId, otherPrompt), + ...requests(SESSION), + ...requests(otherId), + event("session.execution.succeeded", { sessionID: SESSION }), + event("session.execution.succeeded", { sessionID: otherId }), + ], + { + beforeResponse: (path) => + recovering && + (path === "/api/session/active" || /\/(message|permission|form)$/.test(path)) + ? Effect.gen(function* () { + const release = yield* Deferred.make(); + yield* Queue.offer(reads, release); + yield* Deferred.await(release); + }) + : Effect.void, + }, + ); + const other = yield* runtime.resumeThread({ + providerThread: { + ...providerThread(yield* DateTime.now), + id: ProviderThreadId.make("provider-thread:other"), + nativeThreadRef: { driver: OPENCODE_PROVIDER, nativeId: otherId, strength: "strong" }, + }, + threadId, + modelSelection: bigPickle, + runtimePolicy: policy(), + }); + const terminals = yield* runtime.events.pipe( + Stream.filter((event) => event.type === "turn.terminal"), + Stream.take(2), + Stream.runCollect, + Effect.forkScoped, + ); + yield* runtime.startTurn(turnInput(thread)); + recovering = true; + yield* runtime.startTurn({ + ...turnInput(other), + runId: RunId.make("run:other"), + attemptId: RunAttemptId.make("attempt:other"), + }); + // Each response is within the old per-attempt budget; their sum is not. + for (let read = 0; read < 7; read += 1) { + const release = yield* Queue.take(reads); + yield* TestClock.adjust("8 seconds"); + yield* Deferred.succeed(release, undefined); + } + const ended = yield* Fiber.join(terminals); + assert.deepEqual( + ended.map((event) => event.status), + ["completed", "completed"], + ); + assert.sameMembers( + ended.map((event) => event.providerThreadId), + [thread.id, other.id], + ); + }).pipe(Effect.scoped), + ); + + it.effect("recovers an owned server after more than five failed reconciliation requests", () => + Effect.gen(function* () { + const allowRecovery = yield* Deferred.make(); + const attempts = yield* Queue.unbounded(); + let attempt = 0; + const { runtime, thread } = yield* resumed( + [ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + { type: "runtime_exit", status: "success" }, + out("event.subscribe"), + ...Array.from({ length: 6 }, () => [ + out("session.active"), + reply("session.active", { status: 503, body: { message: "Recovering" } }), + ]).flat(), + ...reconnected({ [SESSION]: { type: "running" } }).slice(2), + ...openRequests([]), + event("session.execution.succeeded", { sessionID: SESSION }), + ], + { + beforeResponse: (path) => + path === "/api/session/active" + ? Effect.gen(function* () { + attempt += 1; + yield* Queue.offer(attempts, attempt); + if (attempt === 7) yield* Deferred.await(allowRecovery); + }) + : Effect.void, + }, + ); + + const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped); + + yield* runtime.startTurn(turnInput(thread)); + + for (let expected = 1; expected <= 6; expected += 1) { + assert.equal(yield* Queue.take(attempts), expected); + + yield* TestClock.adjust("30 seconds"); + } + assert.equal(yield* Queue.take(attempts), 7); + yield* Deferred.succeed(allowRecovery, undefined); + assert.equal((yield* Fiber.join(terminal))?.status, "completed"); + }).pipe(Effect.scoped), + ); + + it.effect("bounds recovery when an external server keeps rejecting requests", () => + Effect.gen(function* () { + const attempts = yield* Queue.unbounded(); + const { runtime, thread } = yield* resumed( + [ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + { type: "runtime_exit", status: "success" }, + ...Array.from({ length: 5 }, () => [ + out("event.subscribe"), + out("session.active"), + reply("session.active", { status: 503, body: { message: "Unavailable" } }), + ]).flat(), + ], + { + external: true, + beforeResponse: (path) => + path === "/api/session/active" + ? Queue.offer(attempts, undefined).pipe(Effect.asVoid) + : Effect.void, + }, + ); + const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped); + yield* runtime.startTurn(turnInput(thread)); + for (let attempt = 0; attempt < 4; attempt += 1) { + yield* Queue.take(attempts); + yield* TestClock.adjust("2 seconds"); + } + yield* Queue.take(attempts); + assert.equal((yield* Fiber.join(terminal))?.status, "failed"); + }).pipe(Effect.scoped), + ); + + it.effect("restores approvals and questions after a slow owned-server recovery", () => + Effect.gen(function* () { + const reading = yield* Deferred.make(); + const resume = yield* Deferred.make(); + const requests = yield* Queue.unbounded(); + const { runtime, thread } = yield* resumed( + [ + out("session.prompt", { sessionID: SESSION, id: PROMPT_ID, text: "" }), + promptAccepted, + ...reconnected({ [SESSION]: { type: "running" } }), + ...openRequests([shellAsk.data], [colorForm]), + out("permission.reply", { + sessionID: SESSION, + requestID: shellAsk.data.id, + decision: "once", + }), + reply("permission.reply", null), + out("session.form.reply", { + sessionID: SESSION, + formID: colorForm.id, + answer: { q0: "Red" }, + }), + reply("session.form.reply", null), + event("session.execution.succeeded", { sessionID: SESSION }), + ], + { + supervised: true, + beforeResponse: (path) => + path.endsWith("/message") + ? Deferred.succeed(reading, undefined).pipe(Effect.andThen(Deferred.await(resume))) + : Effect.void, + }, + ); + const ended = yield* Deferred.make(); + yield* runtime.events.pipe( + Stream.runForEach((event) => + Effect.gen(function* () { + if ( + event.type === "runtime_request.updated" && + event.runtimeRequest.status === "pending" + ) { + yield* Queue.offer(requests, event.runtimeRequest.id); + } + if (event.type === "turn.terminal") { + assert.equal(event.status, "completed"); + yield* Deferred.succeed(ended, undefined); + } + }), + ), + Effect.forkScoped, + ); + yield* runtime.startTurn(turnInput(thread, bigPickle, "approval-required")); + yield* Deferred.await(reading); + yield* TestClock.adjust("2 minutes"); + yield* Deferred.succeed(resume, undefined); + yield* runtime.respondToRuntimeRequest({ + requestId: yield* Queue.take(requests), + decision: "accept", + }); + yield* runtime.respondToRuntimeRequest({ + requestId: yield* Queue.take(requests), + answers: { q0: "Red" }, + }); + yield* Deferred.await(ended); + }).pipe(Effect.scoped), + ); + + it.effect("fails owned work when the process exits during a stalled history read", () => + Effect.gen(function* () { + const reading = yield* Deferred.make(); + const cancelled = yield* Deferred.make(); + const exited = yield* Deferred.make(); + const { runtime, thread } = yield* resumed( + [ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + ...reconnected({ [SESSION]: { type: "running" } }), + ], + { + processExit: Deferred.await(exited), + beforeResponse: (path) => + path.endsWith("/message") + ? Deferred.succeed(reading, undefined).pipe( + Effect.andThen(Effect.never), + Effect.onInterrupt(() => Deferred.succeed(cancelled, undefined)), + ) + : Effect.void, + }, + ); + const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped); + yield* runtime.startTurn(turnInput(thread)); + yield* Deferred.await(reading); + yield* Deferred.succeed(exited, undefined); + assert.equal((yield* Fiber.join(terminal))?.status, "failed"); + yield* Deferred.await(cancelled); + }).pipe(Effect.scoped), + ); + + it.effect("releases a recovering session on shutdown without waiting for its HTTP read", () => + Effect.gen(function* () { + const sessionScope = yield* Effect.acquireRelease(Scope.make(), (scope) => + Scope.close(scope, Exit.void), + ); + const reading = yield* Deferred.make(); + const cancelled = yield* Deferred.make(); + const borrowers = { current: 0 }; + const { runtime, thread } = yield* resumed( + [ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + ...reconnected({ [SESSION]: { type: "running" } }), + ], + { + borrowers, + beforeResponse: (path) => + path.endsWith("/message") + ? Deferred.succeed(reading, undefined).pipe( + Effect.andThen(Effect.never), + Effect.onInterrupt(() => Deferred.succeed(cancelled, undefined)), + ) + : Effect.void, + }, + ).pipe(Effect.provideService(Scope.Scope, sessionScope)); + yield* runtime.startTurn(turnInput(thread)); + yield* Deferred.await(reading); + assert.equal(borrowers.current, 1); + yield* Scope.close(sessionScope, Exit.void); + yield* Deferred.await(cancelled); + assert.equal(borrowers.current, 0); + }).pipe(Effect.scoped), + ); + + it.effect("honors a timed-out Stop during recovery and ignores its late backfill", () => + Effect.gen(function* () { + const recovered = yield* Deferred.make(); + const reading = yield* Deferred.make(); + const resume = yield* Deferred.make(); + const { runtime, thread } = yield* resumed( + [ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + ...reconnected({ [SESSION]: { type: "running" } }), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", ""), + ...openRequests([]), + event("session.execution.interrupted", { sessionID: SESSION }), + event("session.model.selected", { + sessionID: SESSION, + model: { providerID: "opencode", id: "recovered" }, + }), + ], + { + beforeResponse: (path) => + path.endsWith("/message") + ? Deferred.succeed(reading, undefined).pipe(Effect.andThen(Deferred.await(resume))) + : Effect.void, + }, + ); + const collected: Array = []; + const stopped = yield* Deferred.make(); + yield* runtime.events.pipe( + Stream.runForEach((event) => + Effect.gen(function* () { + collected.push(event); + if ( + event.type === "provider_thread.updated" && + event.providerThread.nativeMetadata?.modelSelection?.model === "opencode/recovered" + ) { + yield* Deferred.succeed(recovered, undefined); + } + if (event.type === "turn.terminal") yield* Deferred.succeed(stopped, undefined); + }), + ), + Effect.forkScoped, + ); + yield* runtime.startTurn(turnInput(thread)); + yield* Deferred.await(reading); + const stopping = yield* runtime + .interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + }) + .pipe(Effect.forkScoped); + yield* TestClock.adjust("11 seconds"); + yield* Fiber.join(stopping); + yield* Deferred.await(stopped); + yield* Deferred.succeed(resume, undefined); + yield* Deferred.await(recovered); + assert.deepEqual( + collected.filter((event) => event.type === "turn.terminal").map((event) => event.status), + ["interrupted"], + ); + assert.isFalse( + collected.some( + (event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message", + ), + ); + }).pipe(Effect.scoped), + ); + it.effect("keeps a turn still running after a reconnect open for its next events", () => Effect.gen(function* () { const { runtime, thread } = yield* resumed([ diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts index 0b29a15d9e9b..18978c281342 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts @@ -131,6 +131,7 @@ const replayHttpClient = ( controller: OpenCodeReplayController, replayGate: ProviderReplayGate | undefined, onPrompt: ((input: unknown) => void) | undefined, + beforeResponse: ((path: string) => Effect.Effect) | undefined, ) => HttpClient.make((request, url) => Effect.tryPromise({ @@ -185,13 +186,18 @@ const replayHttpClient = ( new HttpClientError.HttpClientError({ reason: new HttpClientError.TransportError({ request, cause }), }), - }).pipe(Effect.map((response) => HttpClientResponse.fromWeb(request, response))), + }).pipe( + Effect.map((response) => HttpClientResponse.fromWeb(request, response)), + Effect.tap(() => beforeResponse?.(url.pathname) ?? Effect.void), + ), ); export const replayServer = ( transcript: ProviderReplayTranscript, options?: { readonly external?: boolean; + readonly processExit?: Effect.Effect; + readonly beforeResponse?: (path: string) => Effect.Effect; readonly onPrompt?: (input: unknown) => void; readonly replayGate?: ProviderReplayGate; readonly borrowers?: { current: number }; @@ -208,7 +214,12 @@ export const replayServer = ( const opencode = yield* OpenCode2Client.make.pipe( Effect.provideService( HttpClient.HttpClient, - replayHttpClient(controller, options?.replayGate, options?.onPrompt), + replayHttpClient( + controller, + options?.replayGate, + options?.onPrompt, + options?.beforeResponse, + ), ), ); const connection = { @@ -216,6 +227,7 @@ export const replayServer = ( url: BASE_URL, version: transcript.version, external: options?.external ?? false, + ...(options?.external ? {} : { processExit: options?.processExit ?? Effect.never }), }; const borrowers = options?.borrowers; return OpenCode2Server.OpenCode2Server.of({ @@ -239,6 +251,8 @@ const makeReplayAdapter = ( transcript: ProviderReplayTranscript, options?: { readonly external?: boolean; + readonly processExit?: Effect.Effect; + readonly beforeResponse?: (path: string) => Effect.Effect; readonly onPrompt?: (input: unknown) => void; readonly replayGate?: ProviderReplayGate; readonly borrowers?: { current: number }; @@ -282,6 +296,8 @@ export const openCode2ReplayRuntime = ( entries: ReadonlyArray, options?: { readonly external?: boolean; + readonly processExit?: Effect.Effect; + readonly beforeResponse?: (path: string) => Effect.Effect; readonly onPrompt?: (input: unknown) => void; readonly borrowers?: { current: number }; readonly replayGate?: ProviderReplayGate; diff --git a/docs/user/providers-opencode.md b/docs/user/providers-opencode.md index 9ecee2a0439c..70d17226ec47 100644 --- a/docs/user/providers-opencode.md +++ b/docs/user/providers-opencode.md @@ -34,7 +34,9 @@ settings. T3 Code uses only that configured password for an external server; it does not forward a local `OPENCODE_SERVER_PASSWORD`. If connection or version checks fail, check the URL, credentials, and OpenCode version, then refresh provider status. -After a lost connection, send another prompt to reconnect to the same OpenCode +T3 Code waits for a local server it started to recover after a lost connection. +You can stop the turn while it reconnects. If the server exits or an external +connection cannot be restored, send another prompt to resume the same OpenCode session. ## Approvals diff --git a/packages/provider-opencode/src/server/v2/OpenCode2Server.test.ts b/packages/provider-opencode/src/server/v2/OpenCode2Server.test.ts index d1e9693fb67b..fe30654b151d 100644 --- a/packages/provider-opencode/src/server/v2/OpenCode2Server.test.ts +++ b/packages/provider-opencode/src/server/v2/OpenCode2Server.test.ts @@ -207,8 +207,12 @@ describe("OpenCode2Server spawned server", () => { assert.strictEqual(first.version, "2.0.18"); assert.strictEqual(first.external, false); assert.strictEqual(second.client, first.client); + if (first.processExit === undefined) + return yield* Effect.die("Missing owned process lifetime"); + return first.processExit; }).pipe( Effect.scoped, + Effect.flatMap((exited) => exited), Effect.provide( Layer.mergeAll( OpenCode2Client.layer, diff --git a/packages/provider-opencode/src/server/v2/OpenCode2Server.ts b/packages/provider-opencode/src/server/v2/OpenCode2Server.ts index e24d550e99ee..03627fb3d908 100644 --- a/packages/provider-opencode/src/server/v2/OpenCode2Server.ts +++ b/packages/provider-opencode/src/server/v2/OpenCode2Server.ts @@ -21,6 +21,8 @@ export interface OpenCode2Connection extends OpenCode2Client.OpenCode2Api { readonly url: string; readonly version: string; readonly external: boolean; + /** Only owned processes have an authoritative lifetime signal. */ + readonly processExit?: Effect.Effect; } export class OpenCode2Server extends Context.Service< @@ -175,7 +177,7 @@ export const make = Effect.fn("OpenCode2Server.make")(function* (input: { withConnection: (use) => owner.withServer((server) => latest?.url === server.url - ? use(latest) + ? use({ ...latest, processExit: server.exitCode.pipe(Effect.asVoid) }) : Effect.die(new Error("OpenCode 2 server was lent before verification.")), ), }); diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index f837e24e2793..770b482ce584 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -40,6 +40,7 @@ import { import type * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; import * as Crypto from "effect/Crypto"; import * as Effect from "effect/Effect"; import * as Hex from "effect/encoding/Hex"; @@ -551,6 +552,11 @@ const RECONNECT_ATTEMPTS = 5; const RECONNECT_DELAY = "2 seconds"; const RECONCILE_TIMEOUT = "15 seconds"; const RECONNECT_WAIT = "30 seconds"; +const OWNED_RECONNECT_SCHEDULE = Schedule.exponential(RECONNECT_DELAY).pipe( + Schedule.modifyDelay(({ duration }) => + Effect.succeed(Duration.min(duration, Duration.seconds(30))), + ), +); const LOST_BACKGROUND = "T3 Code lost its connection to OpenCode while this subagent ran, so its result is not shown."; const INVENTORY_TIMEOUT = "5 seconds"; @@ -2535,66 +2541,72 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr Stream.takeUntil((message) => message.id === start), Stream.runCollect, ); - // Without its start the history holds only earlier turns (the stream - // dropped before the prompt landed): none of it is this turn's. - const found = start === null || read.at(-1)?.id === start; - if (!found) return undefined; - const recent = before !== undefined && before !== null ? read.slice(0, -1) : read; - const idle = recent.find((message) => message.type === "idle"); - const assistant = recent.find((message) => message.type === "assistant"); - if (assistant?.retry !== undefined) { - // Snapshots omit when the retry was scheduled, so only its remaining delay is known. - yield* updateRetry( - state, - turn, - assistant.retry, - DateTime.toEpochMillis(assistant.retry.at) - DateTime.toEpochMillis(yield* DateTime.now), - ); - } else if ( - turn.retry !== undefined && - assistant !== undefined && - assistant.error === undefined && - assistant.content.length > 0 - ) { - yield* emitRetry(state, turn, "completed"); - turn.retry = undefined; - } - for (const message of recent.toReversed()) { - if (message.type !== "assistant" || state.active !== turn) continue; - const ordinals = { text: 0, reasoning: 0 }; - for (const part of message.content) { - if (part.type === "text" || part.type === "reasoning") { - const block = { assistantMessageID: message.id, ordinal: ordinals[part.type]++ }; - yield* emitText(state, turn, block, part.type, () => part.text, true); - continue; - } - if (part.type !== "tool") continue; - if (!turn.tools.has(part.id)) { - turn.tools.set(part.id, { name: part.name, input: {} }); - turn.startedAt.set(part.id, yield* DateTime.now); - } - const tool = turn.tools.get(part.id)!; - if (typeof part.state.input === "object" && part.state.input !== null) { - tool.input = part.state.input as Record; + return yield* lock.withPermit( + Effect.gen(function* () { + if (state.active !== turn) return undefined; + // Without its start the history holds only earlier turns (the stream + // dropped before the prompt landed): none of it is this turn's. + const found = start === null || read.at(-1)?.id === start; + if (!found) return undefined; + const recent = before !== undefined && before !== null ? read.slice(0, -1) : read; + const idle = recent.find((message) => message.type === "idle"); + const assistant = recent.find((message) => message.type === "assistant"); + if (assistant?.retry !== undefined) { + // Snapshots omit when the retry was scheduled, so only its remaining delay is known. + yield* updateRetry( + state, + turn, + assistant.retry, + DateTime.toEpochMillis(assistant.retry.at) - + DateTime.toEpochMillis(yield* DateTime.now), + ); + } else if ( + turn.retry !== undefined && + assistant !== undefined && + assistant.error === undefined && + assistant.content.length > 0 + ) { + yield* emitRetry(state, turn, "completed"); + turn.retry = undefined; } - if (part.state.status === "completed") { - yield* emitTool(state, turn, part.id, "completed", { - output: textOf(part.state.content), - metadata: part.state.metadata, - }); - turn.tools.delete(part.id); - } else if (part.state.status === "error") { - yield* emitTool(state, turn, part.id, "failed", { - output: part.state.error.message, - metadata: part.state.metadata, - }); - turn.tools.delete(part.id); - } else { - yield* emitTool(state, turn, part.id, "running"); + for (const message of recent.toReversed()) { + if (message.type !== "assistant" || state.active !== turn) continue; + const ordinals = { text: 0, reasoning: 0 }; + for (const part of message.content) { + if (part.type === "text" || part.type === "reasoning") { + const block = { assistantMessageID: message.id, ordinal: ordinals[part.type]++ }; + yield* emitText(state, turn, block, part.type, () => part.text, true); + continue; + } + if (part.type !== "tool") continue; + if (!turn.tools.has(part.id)) { + turn.tools.set(part.id, { name: part.name, input: {} }); + turn.startedAt.set(part.id, yield* DateTime.now); + } + const tool = turn.tools.get(part.id)!; + if (typeof part.state.input === "object" && part.state.input !== null) { + tool.input = part.state.input as Record; + } + if (part.state.status === "completed") { + yield* emitTool(state, turn, part.id, "completed", { + output: textOf(part.state.content), + metadata: part.state.metadata, + }); + turn.tools.delete(part.id); + } else if (part.state.status === "error") { + yield* emitTool(state, turn, part.id, "failed", { + output: part.state.error.message, + metadata: part.state.metadata, + }); + turn.tools.delete(part.id); + } else { + yield* emitTool(state, turn, part.id, "running"); + } + } } - } - } - return idle === undefined ? undefined : { outcome: idle.outcome }; + return idle === undefined ? undefined : { outcome: idle.outcome }; + }), + ); }); /** @@ -2630,61 +2642,85 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const turn = state.active; if (turn === undefined) continue; const ended = yield* backfill(sessionId, state); - if (sessionId in active || state.active !== turn) continue; - yield* finishTurn( - state, - ended?.outcome === "succeeded" - ? { status: turn.interrupted ? "interrupted" : "completed" } - : ended?.outcome === "failed" - ? { - status: "failed", - failure: makeProviderFailure({ - message: - "OpenCode ended the turn with an error while T3 Code was reconnecting.", - class: "provider_error", - }), - } - : { status: "interrupted" }, + if (sessionId in active) continue; + yield* lock.withPermit( + Effect.gen(function* () { + if (state.active !== turn) return; + yield* finishTurn( + state, + ended?.outcome === "succeeded" + ? { status: turn.interrupted ? "interrupted" : "completed" } + : ended?.outcome === "failed" + ? { + status: "failed", + failure: makeProviderFailure({ + message: + "OpenCode ended the turn with an error while T3 Code was reconnecting.", + class: "provider_error", + }), + } + : { status: "interrupted" }, + ); + }), ); } - for (const state of background) { - for (const call of runningCalls(state)) { - const child = call.child?.sessionId; - if (child === undefined || !(child in active)) { - yield* settleCall(call, "interrupted", LOST_BACKGROUND); + yield* lock.withPermit( + Effect.gen(function* () { + for (const state of background) { + for (const call of runningCalls(state)) { + const child = call.child?.sessionId; + if (child === undefined || !(child in active)) { + yield* settleCall(call, "interrupted", LOST_BACKGROUND); + } + } + for (const session of sessionsOf(state)) { + discardWakes(session, session.wakes.splice(0)); + session.reports.clear(); + } } - } - for (const session of sessionsOf(state)) { - discardWakes(session, session.wakes.splice(0)); - session.reports.clear(); - } - } - // Stopped off the lock this reconcile holds; one still running waits for a Stop. - for (const thread of new Set([...strays.values()].map(rootOf))) { - yield* Effect.forkIn(stopStrays(thread), sessionScope); - } + // Stopped off the lock this reconcile holds; one still running waits for a Stop. + for (const thread of new Set([...strays.values()].map(rootOf))) { + yield* Effect.forkIn(stopStrays(thread), sessionScope); + } + }), + ); const listed = new Map>(); for (const sessionId of Object.keys(active)) { - if (ownerOf(sessionId) === undefined) continue; + const owner = ownerOf(sessionId); + if (owner === undefined) continue; + const turn = owner.active; + // Replies may settle during these reads; an older snapshot must not ask again. + const known = new Set([...pending.values()].map((entry) => entry.native.id)); const sessionID = Session.ID.make(sessionId); const [permissions, forms] = yield* Effect.all([ client.permission.list({ sessionID }), client.session.form.list({ sessionID }), ]); - listed.set(sessionId, new Set([...permissions, ...forms].map((request) => request.id))); - const shown = new Set([...pending.values()].map((entry) => entry.native.id)); - for (const request of permissions) { - if (!shown.has(request.id)) yield* onAsked({ type: "permission", request }); - } - for (const form of forms) { - if (!shown.has(form.id)) yield* onAsked({ type: "form", form }); - } - } - for (const entry of Array.from(pending.values())) { - if (listed.get(entry.sessionId)?.has(entry.native.id) !== true) { - yield* settleRequest(entry, "cancelled"); - } + yield* lock.withPermit( + Effect.gen(function* () { + if (ownerOf(sessionId) !== owner || owner.active !== turn) return; + listed.set(sessionId, new Set([...permissions, ...forms].map((request) => request.id))); + const shown = new Set([...pending.values()].map((entry) => entry.native.id)); + for (const request of permissions) { + if (!shown.has(request.id) && !known.has(request.id)) + yield* onAsked({ type: "permission", request }); + } + for (const form of forms) { + if (!shown.has(form.id) && !known.has(form.id)) + yield* onAsked({ type: "form", form }); + } + }), + ); } + yield* lock.withPermit( + Effect.gen(function* () { + for (const entry of Array.from(pending.values())) { + if (listed.get(entry.sessionId)?.has(entry.native.id) !== true) { + yield* settleRequest(entry, "cancelled"); + } + } + }), + ); }); const reconnect = Effect.gen(function* () { @@ -2702,10 +2738,21 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr yield* Scope.close(previous, Exit.void); // A restarted server forgot T3's MCP servers; the next turn adds them again. for (const state of threads.values()) state.mcp = undefined; - yield* lock.withPermit(reconcile).pipe(Effect.timeout(RECONCILE_TIMEOUT)); + // A live owned process may still be serving its turns. Recovery waits for its + // reads, outside the event lock so Stop can settle independently. External + // servers have no process lifetime signal, so retain their bounded recovery. + yield* connection.processExit === undefined + ? reconcile.pipe(Effect.timeout(RECONCILE_TIMEOUT)) + : reconcile.pipe(Effect.retry({ schedule: OWNED_RECONNECT_SCHEDULE })); return stream; }).pipe( - Effect.retry({ times: RECONNECT_ATTEMPTS - 1, schedule: Schedule.spaced(RECONNECT_DELAY) }), + Effect.retry({ + times: connection.processExit === undefined ? RECONNECT_ATTEMPTS - 1 : undefined, + schedule: + connection.processExit === undefined + ? Schedule.spaced(RECONNECT_DELAY) + : OWNED_RECONNECT_SCHEDULE, + }), ); let currentScope = initial.scope; @@ -2733,7 +2780,9 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr Effect.option, ); if (next._tag === "None") { - return yield* failAll("The OpenCode event stream was lost and could not reconnect."); + return yield* lock.withPermit( + failAll("The OpenCode event stream was lost and could not reconnect."), + ); } streamFailure = undefined; const done = reconnected; @@ -2744,7 +2793,17 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ), ); // Subscribed before any session or prompt call, so no event of theirs is missed. - yield* follow(yield* connection.events).pipe(Effect.forkScoped); + const following = follow(yield* connection.events); + yield* ( + connection.processExit === undefined + ? following + : Effect.raceFirst( + following, + connection.processExit.pipe( + Effect.andThen(lock.withPermit(failAll("The OpenCode server exited."))), + ), + ) + ).pipe(Effect.forkScoped); // A server T3 did not start keeps running after T3 stops, so stop the turns // it would otherwise finish unseen. A spawned server stops with its owner. From d86979ed510942dc6c6a044bab505c97d60c6d5e Mon Sep 17 00:00:00 2001 From: tinzeljar Date: Fri, 23 Oct 2026 18:28:31 +0200 Subject: [PATCH 2/4] fix(opencode): preserve controls and cleanup during recovery --- .../Adapters/OpenCode2AdapterV2.test.ts | 156 +++++++++++++----- .../src/server/v2/OpenCode2Client.test.ts | 33 ++++ .../src/server/v2/OpenCode2Client.ts | 14 +- .../src/server/v2/adapter.ts | 27 +-- 4 files changed, 179 insertions(+), 51 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 7bcf0e534d66..74d97bda2b3a 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -2330,7 +2330,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { it.effect("recovers an owned server after more than five failed reconciliation requests", () => Effect.gen(function* () { const allowRecovery = yield* Deferred.make(); - const attempts = yield* Queue.unbounded(); + const attempts = yield* Queue.unbounded(); let attempt = 0; const { runtime, thread } = yield* resumed( [ @@ -2351,23 +2351,19 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { path === "/api/session/active" ? Effect.gen(function* () { attempt += 1; - yield* Queue.offer(attempts, attempt); + yield* Queue.offer(attempts, undefined); if (attempt === 7) yield* Deferred.await(allowRecovery); }) : Effect.void, }, ); - const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped); - yield* runtime.startTurn(turnInput(thread)); - - for (let expected = 1; expected <= 6; expected += 1) { - assert.equal(yield* Queue.take(attempts), expected); - + for (let retry = 0; retry < 6; retry += 1) { + yield* Queue.take(attempts); yield* TestClock.adjust("30 seconds"); } - assert.equal(yield* Queue.take(attempts), 7); + yield* Queue.take(attempts); yield* Deferred.succeed(allowRecovery, undefined); assert.equal((yield* Fiber.join(terminal))?.status, "completed"); }).pipe(Effect.scoped), @@ -2538,72 +2534,158 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); - it.effect("honors a timed-out Stop during recovery and ignores its late backfill", () => + it.effect.each([ + ["timed-out", ""], + ["confirmed", { interrupted: true }], + ] as const)( + "honors a %s Stop during recovery and ignores its late backfill", + ([outcome, response]) => + Effect.gen(function* () { + const recovered = yield* Deferred.make(); + const reading = yield* Deferred.make(); + const resume = yield* Deferred.make(); + const { runtime, thread } = yield* resumed( + [ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + ...reconnected({ [SESSION]: { type: "running" } }), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", response), + ...openRequests([]), + event("session.execution.interrupted", { sessionID: SESSION }), + event("session.model.selected", { + sessionID: SESSION, + model: { providerID: "opencode", id: "recovered" }, + }), + ], + { + beforeResponse: (path) => + path.endsWith("/message") + ? Deferred.succeed(reading, undefined).pipe(Effect.andThen(Deferred.await(resume))) + : Effect.void, + }, + ); + const collected: Array = []; + const stopped = yield* Deferred.make(); + yield* runtime.events.pipe( + Stream.runForEach((event) => + Effect.gen(function* () { + collected.push(event); + if ( + event.type === "provider_thread.updated" && + event.providerThread.nativeMetadata?.modelSelection?.model === "opencode/recovered" + ) { + yield* Deferred.succeed(recovered, undefined); + } + if (event.type === "turn.terminal") yield* Deferred.succeed(stopped, undefined); + }), + ), + Effect.forkScoped, + ); + yield* runtime.startTurn(turnInput(thread)); + yield* Deferred.await(reading); + const stopping = yield* runtime + .interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + }) + .pipe(Effect.forkScoped); + if (outcome === "timed-out") yield* TestClock.adjust("11 seconds"); + yield* Fiber.join(stopping); + assert.equal(runtime.providerSession.status, "ready"); + yield* Deferred.await(stopped); + yield* Deferred.succeed(resume, undefined); + yield* Deferred.await(recovered); + assert.deepEqual( + collected.filter((event) => event.type === "turn.terminal").map((event) => event.status), + ["interrupted"], + ); + assert.isFalse( + collected.some( + (event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message", + ), + ); + }).pipe(Effect.scoped), + ); + + it.effect("keeps a background child's requests when its parent stops during reconciliation", () => Effect.gen(function* () { - const recovered = yield* Deferred.make(); const reading = yield* Deferred.make(); const resume = yield* Deferred.make(); + const recovered = yield* Deferred.make(); + const permission = { ...shellAsk.data, sessionID: CHILD }; + const form = { ...colorForm, sessionID: CHILD }; const { runtime, thread } = yield* resumed( [ - out("session.prompt", { sessionID: SESSION, text: "" }), - promptAccepted, - ...reconnected({ [SESSION]: { type: "running" } }), + ...backgroundLaunch(CHILD), + event("permission.asked", permission), + event("form.created", { form }), + ...reconnected({ [SESSION]: { type: "running" }, [CHILD]: { type: "running" } }), + ...openRequests([]), + out("permission.list", { sessionID: CHILD }), + replyData("permission.list", [permission]), + out("session.form.list", { sessionID: CHILD }), + replyData("session.form.list", [form]), out("session.interrupt", { sessionID: SESSION }), reply("session.interrupt", ""), - ...openRequests([]), - event("session.execution.interrupted", { sessionID: SESSION }), event("session.model.selected", { sessionID: SESSION, model: { providerID: "opencode", id: "recovered" }, }), + out("permission.reply", { sessionID: CHILD, requestID: permission.id, decision: "once" }), + reply("permission.reply", null), + out("session.form.reply", { sessionID: CHILD, formID: form.id, answer: { q0: "Red" } }), + reply("session.form.reply", null), ], { beforeResponse: (path) => - path.endsWith("/message") + path === `/api/session/${CHILD}/form` ? Deferred.succeed(reading, undefined).pipe(Effect.andThen(Deferred.await(resume))) : Effect.void, }, ); - const collected: Array = []; - const stopped = yield* Deferred.make(); + const requests = new Map< + string, + { readonly id: RuntimeRequestId; readonly status: string } + >(); yield* runtime.events.pipe( Stream.runForEach((event) => Effect.gen(function* () { - collected.push(event); + if (event.type === "runtime_request.updated") { + const nativeId = event.runtimeRequest.nativeRequestRef?.nativeId; + if (nativeId != null) requests.set(nativeId, event.runtimeRequest); + } if ( event.type === "provider_thread.updated" && event.providerThread.nativeMetadata?.modelSelection?.model === "opencode/recovered" ) { yield* Deferred.succeed(recovered, undefined); } - if (event.type === "turn.terminal") yield* Deferred.succeed(stopped, undefined); }), ), Effect.forkScoped, ); - yield* runtime.startTurn(turnInput(thread)); + yield* runtime.startTurn(withLineage(thread)); yield* Deferred.await(reading); const stopping = yield* runtime - .interruptTurn({ - providerThread: thread, - providerTurnId: yield* providerTurnId, - }) + .interruptTurn({ providerThread: thread, providerTurnId: yield* providerTurnId }) .pipe(Effect.forkScoped); yield* TestClock.adjust("11 seconds"); yield* Fiber.join(stopping); - yield* Deferred.await(stopped); yield* Deferred.succeed(resume, undefined); yield* Deferred.await(recovered); - assert.deepEqual( - collected.filter((event) => event.type === "turn.terminal").map((event) => event.status), - ["interrupted"], - ); - assert.isFalse( - collected.some( - (event) => - event.type === "turn_item.updated" && event.turnItem.type === "assistant_message", - ), - ); + assert.isTrue(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + assert.equal(requests.get(permission.id)?.status, "pending"); + assert.equal(requests.get(form.id)?.status, "pending"); + yield* runtime.respondToRuntimeRequest({ + requestId: requests.get(permission.id)!.id, + decision: "accept", + }); + yield* runtime.respondToRuntimeRequest({ + requestId: requests.get(form.id)!.id, + answers: { q0: "Red" }, + }); }).pipe(Effect.scoped), ); diff --git a/packages/provider-opencode/src/server/v2/OpenCode2Client.test.ts b/packages/provider-opencode/src/server/v2/OpenCode2Client.test.ts index 3b002194befe..87dd02822fe3 100644 --- a/packages/provider-opencode/src/server/v2/OpenCode2Client.test.ts +++ b/packages/provider-opencode/src/server/v2/OpenCode2Client.test.ts @@ -1,5 +1,7 @@ import { assert, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Scope from "effect/Scope"; import * as Layer from "effect/Layer"; import * as Redacted from "effect/Redacted"; import * as HttpClient from "effect/http/HttpClient"; @@ -46,3 +48,34 @@ describe("OpenCode2Client authorization", () => { }), ); }); + +it.effect("closes an event subscription that recovery has not started reading", () => + Effect.gen(function* () { + const signals: Array = []; + const http = HttpClient.make((request, _url, signal) => + Effect.sync(() => { + signals.push(signal); + return HttpClientResponse.fromWeb( + request, + new Response(": connected\n\n", { + headers: { "content-type": "text/event-stream" }, + }), + ); + }), + ); + const opencode = yield* OpenCode2Client.make.pipe( + Effect.provideService(HttpClient.HttpClient, http), + ); + const connection = yield* opencode.connect({ + baseUrl: "http://127.0.0.1:4096", + password: "test", + }); + const scope = yield* Effect.acquireRelease(Scope.make(), (scope) => + Scope.close(scope, Exit.void), + ); + yield* connection.events.pipe(Effect.provideService(Scope.Scope, scope), Effect.asVoid); + assert.isFalse(signals[0]!.aborted); + yield* Scope.close(scope, Exit.void); + assert.isTrue(signals[0]!.aborted); + }).pipe(Effect.scoped), +); diff --git a/packages/provider-opencode/src/server/v2/OpenCode2Client.ts b/packages/provider-opencode/src/server/v2/OpenCode2Client.ts index 41d11a70cbc8..cf0743c21784 100644 --- a/packages/provider-opencode/src/server/v2/OpenCode2Client.ts +++ b/packages/provider-opencode/src/server/v2/OpenCode2Client.ts @@ -7,6 +7,7 @@ import * as Option from "effect/Option"; import * as Redacted from "effect/Redacted"; import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; +import type * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as Sse from "effect/encoding/Sse"; import * as HttpClient from "effect/http/HttpClient"; @@ -61,7 +62,8 @@ export interface OpenCode2Api { OpenCode2StreamEvent, HttpClientError.HttpClientError | Sse.Retry | Sse.SseError | OpenCode2SilentStreamError >, - HttpClientError.HttpClientError + HttpClientError.HttpClientError, + Scope.Scope >; } @@ -120,9 +122,13 @@ const undecodable = Effect.fnUntraced(function* (data: string) { return Result.failVoid; }); -const readEvents = HttpClient.get("/api/event", { - headers: { accept: "text/event-stream" }, -}).pipe( +const readEvents = HttpClient.HttpClient.pipe( + // Recovery subscribes before reading history, so its stream may never be consumed. + Effect.flatMap((client) => + HttpClient.withScope(client).get("/api/event", { + headers: { accept: "text/event-stream" }, + }), + ), Effect.flatMap(HttpClientResponse.filterStatusOk), Effect.map((response) => response.stream.pipe( diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index 770b482ce584..e2992e6e2aab 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -2688,7 +2688,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr for (const sessionId of Object.keys(active)) { const owner = ownerOf(sessionId); if (owner === undefined) continue; - const turn = owner.active; + const turn = requestTurn(sessionId)?.turn; // Replies may settle during these reads; an older snapshot must not ask again. const known = new Set([...pending.values()].map((entry) => entry.native.id)); const sessionID = Session.ID.make(sessionId); @@ -2698,7 +2698,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ]); yield* lock.withPermit( Effect.gen(function* () { - if (ownerOf(sessionId) !== owner || owner.active !== turn) return; + if (ownerOf(sessionId) !== owner || requestTurn(sessionId)?.turn !== turn) return; listed.set(sessionId, new Set([...permissions, ...forms].map((request) => request.id))); const shown = new Set([...pending.values()].map((entry) => entry.native.id)); for (const request of permissions) { @@ -2729,7 +2729,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const scope = yield* Scope.make(); const { next, stream } = yield* Effect.gen(function* () { const next = yield* borrow.pipe(Effect.provideService(Scope.Scope, scope)); - return { next, stream: yield* next.events }; + return { next, stream: yield* next.events.pipe(Effect.provideService(Scope.Scope, scope)) }; }).pipe(Effect.onError(() => Scope.close(scope, Exit.void))); const previous = currentScope; connection = next; @@ -2738,9 +2738,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr yield* Scope.close(previous, Exit.void); // A restarted server forgot T3's MCP servers; the next turn adds them again. for (const state of threads.values()) state.mcp = undefined; - // A live owned process may still be serving its turns. Recovery waits for its - // reads, outside the event lock so Stop can settle independently. External - // servers have no process lifetime signal, so retain their bounded recovery. + // An owned process's exit, rather than a read deadline, bounds recovery. + // HTTP waits stay outside the event lock so Stop can still settle. yield* connection.processExit === undefined ? reconcile.pipe(Effect.timeout(RECONCILE_TIMEOUT)) : reconcile.pipe(Effect.retry({ schedule: OWNED_RECONNECT_SCHEDULE })); @@ -2793,7 +2792,9 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ), ); // Subscribed before any session or prompt call, so no event of theirs is missed. - const following = follow(yield* connection.events); + const following = follow( + yield* connection.events.pipe(Effect.provideService(Scope.Scope, currentScope)), + ); yield* ( connection.processExit === undefined ? following @@ -3894,9 +3895,15 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr }), ), ); - if (reply._tag === "None") { - markUnsettled(state); - return yield* lock.withPermit(finishTurn(state, { status: "interrupted" })); + // Recovery may hold the terminal event behind a blocked history read. + if (reply._tag === "None" || (reply.value.interrupted && streamFailure !== undefined)) { + return yield* lock.withPermit( + Effect.gen(function* () { + if (state.active !== turn) return; + markUnsettled(state); + yield* finishTurn(state, { status: "interrupted" }); + }), + ); } // Between the executions a late steer spans nothing runs yet. The // steers are taken back, and a run they already started is skipped. From 119d8995ca95168a252f1a9a196f86eb4c0a5f00 Mon Sep 17 00:00:00 2001 From: tinzeljar Date: Fri, 23 Oct 2026 18:57:51 +0200 Subject: [PATCH 3/4] fix(opencode): reconcile deleted sessions without blocking siblings --- .../Adapters/OpenCode2AdapterV2.test.ts | 39 ++++++++++++++----- .../src/server/v2/adapter.ts | 9 ++++- 2 files changed, 37 insertions(+), 11 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 74d97bda2b3a..d88872eece3c 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -2219,7 +2219,10 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); - it.effect("reconciles slow sessions without spending one shared deadline", () => + it.effect.each([ + ["slow sessions without a shared deadline", false], + ["a session deleted after the active snapshot", true], + ] as const)("reconciles %s", ([_scenario, deleted]) => Effect.gen(function* () { const otherId = "ses_other_stalled_session"; const otherPrompt = "msg_other_stalled_prompt"; @@ -2267,16 +2270,29 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { [SESSION]: { type: "running" }, [otherId]: { type: "running" }, }), - ...history(SESSION, PROMPT_ID), + ...(deleted + ? [ + out("message.list", { sessionID: SESSION, order: "desc", limit: "50" }), + reply("message.list", { + status: 404, + body: { + _tag: "SessionNotFoundError", + sessionID: SESSION, + message: "Deleted during recovery", + }, + }), + ] + : history(SESSION, PROMPT_ID)), ...history(otherId, otherPrompt), - ...requests(SESSION), + ...(deleted ? [] : requests(SESSION)), ...requests(otherId), - event("session.execution.succeeded", { sessionID: SESSION }), + ...(deleted ? [] : [event("session.execution.succeeded", { sessionID: SESSION })]), event("session.execution.succeeded", { sessionID: otherId }), ], { beforeResponse: (path) => recovering && + !deleted && (path === "/api/session/active" || /\/(message|permission|form)$/.test(path)) ? Effect.gen(function* () { const release = yield* Deferred.make(); @@ -2309,17 +2325,20 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { runId: RunId.make("run:other"), attemptId: RunAttemptId.make("attempt:other"), }); - // Each response is within the old per-attempt budget; their sum is not. - for (let read = 0; read < 7; read += 1) { - const release = yield* Queue.take(reads); - yield* TestClock.adjust("8 seconds"); - yield* Deferred.succeed(release, undefined); + if (!deleted) { + // Each response fits the old deadline; their sum does not. + for (let read = 0; read < 7; read += 1) { + const release = yield* Queue.take(reads); + yield* TestClock.adjust("8 seconds"); + yield* Deferred.succeed(release, undefined); + } } const ended = yield* Fiber.join(terminals); assert.deepEqual( ended.map((event) => event.status), - ["completed", "completed"], + [deleted ? "interrupted" : "completed", "completed"], ); + if (deleted) assert.equal(ended[0]?.threadDisposition, "broken"); assert.sameMembers( ended.map((event) => event.providerThreadId), [thread.id, other.id], diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index e2992e6e2aab..9e772bd49e9a 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -2540,10 +2540,12 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ).pipe( Stream.takeUntil((message) => message.id === start), Stream.runCollect, + Effect.catchTags({ SessionNotFoundError: () => Effect.succeed(null) }), ); return yield* lock.withPermit( Effect.gen(function* () { if (state.active !== turn) return undefined; + if (read === null) return { outcome: "missing" as const }; // Without its start the history holds only earlier turns (the stream // dropped before the prompt landed): none of it is this turn's. const found = start === null || read.at(-1)?.id === start; @@ -2632,7 +2634,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ) { return; } - const active = yield* client.session.active(); + const active = { ...(yield* client.session.active()) }; for (const sessionId of busy) if (!(sessionId in active)) busy.delete(sessionId); for (const state of threads.values()) { if (state.unsettled && !(state.sessionId in active)) state.unsettledStarted = true; @@ -2642,6 +2644,10 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const turn = state.active; if (turn === undefined) continue; const ended = yield* backfill(sessionId, state); + if (ended?.outcome === "missing") { + delete active[Session.ID.make(sessionId)]; + busy.delete(sessionId); + } if (sessionId in active) continue; yield* lock.withPermit( Effect.gen(function* () { @@ -2660,6 +2666,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr }), } : { status: "interrupted" }, + ended?.outcome === "missing" ? "broken" : "reusable", ); }), ); From 38f1cdcba137516b00fe62aa8f9f1379d3df821d Mon Sep 17 00:00:00 2001 From: tinzeljar Date: Fri, 23 Oct 2026 19:28:06 +0200 Subject: [PATCH 4/4] test(opencode): cover requests arriving during reconciliation --- .../Adapters/OpenCode2AdapterV2.test.ts | 150 ++++++++++-------- 1 file changed, 85 insertions(+), 65 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index d88872eece3c..cee73a292273 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -2421,71 +2421,91 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); - it.effect("restores approvals and questions after a slow owned-server recovery", () => - Effect.gen(function* () { - const reading = yield* Deferred.make(); - const resume = yield* Deferred.make(); - const requests = yield* Queue.unbounded(); - const { runtime, thread } = yield* resumed( - [ - out("session.prompt", { sessionID: SESSION, id: PROMPT_ID, text: "" }), - promptAccepted, - ...reconnected({ [SESSION]: { type: "running" } }), - ...openRequests([shellAsk.data], [colorForm]), - out("permission.reply", { - sessionID: SESSION, - requestID: shellAsk.data.id, - decision: "once", - }), - reply("permission.reply", null), - out("session.form.reply", { - sessionID: SESSION, - formID: colorForm.id, - answer: { q0: "Red" }, - }), - reply("session.form.reply", null), - event("session.execution.succeeded", { sessionID: SESSION }), - ], - { - supervised: true, - beforeResponse: (path) => - path.endsWith("/message") - ? Deferred.succeed(reading, undefined).pipe(Effect.andThen(Deferred.await(resume))) - : Effect.void, - }, - ); - const ended = yield* Deferred.make(); - yield* runtime.events.pipe( - Stream.runForEach((event) => - Effect.gen(function* () { - if ( - event.type === "runtime_request.updated" && - event.runtimeRequest.status === "pending" - ) { - yield* Queue.offer(requests, event.runtimeRequest.id); - } - if (event.type === "turn.terminal") { - assert.equal(event.status, "completed"); - yield* Deferred.succeed(ended, undefined); - } - }), - ), - Effect.forkScoped, - ); - yield* runtime.startTurn(turnInput(thread, bigPickle, "approval-required")); - yield* Deferred.await(reading); - yield* TestClock.adjust("2 minutes"); - yield* Deferred.succeed(resume, undefined); - yield* runtime.respondToRuntimeRequest({ - requestId: yield* Queue.take(requests), - decision: "accept", - }); - yield* runtime.respondToRuntimeRequest({ - requestId: yield* Queue.take(requests), - answers: { q0: "Red" }, - }); - yield* Deferred.await(ended); - }).pipe(Effect.scoped), + it.effect.each(["lists", "stream"] as const)( + "restores approvals and questions from %s after slow recovery", + (source) => + Effect.gen(function* () { + const gate = makeProviderReplayGate(["late-permission"]); + let holdReads = false; + const reading = yield* Deferred.make(); + const resume = yield* Deferred.make(); + const requests = yield* Queue.unbounded(); + const { runtime, thread } = yield* resumed( + [ + out("session.prompt", { sessionID: SESSION, id: PROMPT_ID, text: "" }), + promptAccepted, + ...reconnected({ [SESSION]: { type: "running" } }), + ...openRequests( + source === "lists" ? [shellAsk.data] : [], + source === "lists" ? [colorForm] : [], + ), + ...(source === "stream" + ? [ + event("permission.asked", shellAsk.data, "late-permission"), + event("form.created", { form: colorForm }), + ] + : []), + out("permission.reply", { + sessionID: SESSION, + requestID: shellAsk.data.id, + decision: "once", + }), + reply("permission.reply", null), + out("session.form.reply", { + sessionID: SESSION, + formID: colorForm.id, + answer: { q0: "Red" }, + }), + reply("session.form.reply", null), + event("session.execution.succeeded", { sessionID: SESSION }), + ], + { + supervised: true, + replayGate: gate, + beforeResponse: (path) => + holdReads && path.endsWith(source === "stream" ? "/form" : "/message") + ? Deferred.succeed(reading, undefined).pipe(Effect.andThen(Deferred.await(resume))) + : Effect.void, + }, + ); + const ended = yield* Deferred.make(); + yield* runtime.events.pipe( + Stream.runForEach((event) => + Effect.gen(function* () { + if ( + event.type === "runtime_request.updated" && + event.runtimeRequest.status === "pending" + ) { + yield* Queue.offer(requests, event.runtimeRequest.id); + } + if (event.type === "turn.terminal") { + assert.equal(event.status, "completed"); + yield* Deferred.succeed(ended, undefined); + } + }), + ), + Effect.forkScoped, + ); + holdReads = true; + yield* runtime.startTurn(turnInput(thread, bigPickle, "approval-required")); + yield* Deferred.await(reading); + if (source === "stream") { + yield* Effect.promise(() => gate.waitForReached("late-permission")); + gate.release("late-permission"); + } + yield* TestClock.adjust("2 minutes"); + assert.equal(yield* Queue.size(requests), 0); + yield* Deferred.succeed(resume, undefined); + yield* runtime.respondToRuntimeRequest({ + requestId: yield* Queue.take(requests), + decision: "accept", + }); + yield* runtime.respondToRuntimeRequest({ + requestId: yield* Queue.take(requests), + answers: { q0: "Red" }, + }); + yield* Deferred.await(ended); + }).pipe(Effect.scoped), ); it.effect("fails owned work when the process exits during a stalled history read", () =>