From 874c131c0da5e4f7ed1e34b9a628ca22289b65e3 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Fri, 2 Oct 2026 00:29:10 -0700 Subject: [PATCH 1/4] fix(server): OpenCode 2 continuations replay only their own background reply OpenCode runs the parent's reply to a background subagent on its own, and the adapter holds it for the continuation turn it offers. That turn replayed whichever held reply was oldest. When a continuation was cancelled while queued, or dropped before it started, its reply stayed at the head of the queue, and the next continuation showed it under its own notification while its own reply stayed held. Each held reply now gets the message id its continuation run starts from. The continuation request carries it, and `ProviderContinuationService` dispatches under it instead of allocating one. The continuation turn replays the reply its message names, and drops older ones whose continuation will never run. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Adapters/OpenCode2AdapterV2.test.ts | 151 +++++++++++++++--- .../Adapters/OpenCode2AdapterV2.ts | 27 +++- .../ProviderContinuationRequests.ts | 6 + .../ProviderContinuationService.test.ts | 19 +++ .../ProviderContinuationService.ts | 10 +- 5 files changed, 178 insertions(+), 35 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 46792897e789..3d82833413e1 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -610,6 +610,28 @@ describe("OpenCode2 adapter", () => { lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, } as OrchestrationV2AppThread, }); + /** + * The continuation turn the orchestrator starts for an offered wake: its + * message is the one the offer named, as `ProviderContinuationService` dispatches it. + */ + const continuationTurn = ( + thread: OrchestrationV2ProviderThread, + offer: ProviderContinuationRequest, + suffix = "wake", + ) => ({ + ...withLineage(thread), + runId: RunId.make(`run:opencode2-adapter:${suffix}`), + runOrdinal: 2, + providerTurnOrdinal: 2, + attemptId: RunAttemptId.make(`attempt:opencode2-adapter:${suffix}`), + message: { + ...turnInput(thread).message, + messageId: offer.messageId!, + text: offer.detail ?? "", + createdBy: "agent" as const, + creationSource: "provider" as const, + }, + }); const CHILD = "ses_f1485c529ffea4URrYruwEg0Ja"; /** * The single reader of the runtime's events: resolves `attached` once the @@ -3762,19 +3784,7 @@ describe("OpenCode2 adapter", () => { yield* Effect.gen(function* () { while (offers.length === 0) yield* Effect.yieldNow; }).pipe(Effect.timeout("2 seconds"), Effect.orDie); - yield* runtime.startTurn({ - ...withLineage(thread), - runId: RunId.make("run:opencode2-adapter:wake"), - runOrdinal: 2, - providerTurnOrdinal: 2, - attemptId: RunAttemptId.make("attempt:opencode2-adapter:wake"), - message: { - ...turnInput(thread).message, - messageId: MessageId.make("message:opencode2-adapter:wake"), - createdBy: "agent" as const, - creationSource: "provider" as const, - }, - }); + yield* runtime.startTurn(continuationTurn(thread, offers[0]!)); yield* Deferred.await(bothEnded); const [first, second] = turns; return { runtime, thread, first: first!, second: second! }; @@ -3841,6 +3851,107 @@ describe("OpenCode2 adapter", () => { }).pipe(Effect.scoped), ); + it.effect("replays only its own background reply into a continuation turn", () => + Effect.gen(function* () { + const CHILD_B = "ses_f1485c529ffeBBBBBBBBBBBBBB"; + const toolB = { sessionID: SESSION, assistantMessageID: "msg_assistant", id: "call-b" }; + const report = (inboxID: string, child: string, text: string) => + event("session.inbox.enqueued", { + inboxID, + sessionID: SESSION, + item: { + type: "synthetic", + payload: { + text: `\n${text}\n`, + description: "Sleep", + metadata: { + source: "subagent", + childID: child, + agent: "General", + state: "completed", + }, + }, + delivery: "steer", + }, + }); + const reply = (assistantMessageID: string, text: string) => + event("session.text.ended", { sessionID: SESSION, assistantMessageID, ordinal: 0, text }); + const offers: Array = []; + const { runtime, thread } = yield* resumed([ + ...backgroundLaunch(CHILD), + // The same turn starts a second background subagent. + event("session.tool.input.started", { ...toolB, name: "subagent" }), + event("session.tool.called", { + ...toolB, + name: "subagent", + input: { description: "Sleep", prompt: "sleep", background: true }, + executed: false, + }), + event("session.created", { ...childCreated(CHILD_B), sessionID: CHILD_B }), + event("session.tool.progress", { + ...toolB, + metadata: { sessionID: CHILD_B, status: "running" }, + }), + event("session.execution.succeeded", { sessionID: SESSION }), + // Each subagent's report wakes the parent into a reply OpenCode runs on its own. + event("session.execution.succeeded", { sessionID: CHILD }), + report("msg_report_a", CHILD, "A_OK"), + event("session.execution.started", { sessionID: SESSION }), + event("session.inbox.delivered", { sessionID: SESSION, inboxID: "msg_report_a" }), + reply("msg_reply_a", "REPLY_A"), + event("session.execution.succeeded", { sessionID: SESSION }), + event("session.execution.succeeded", { sessionID: CHILD_B }), + report("msg_report_b", CHILD_B, "B_OK"), + event("session.execution.started", { sessionID: SESSION }), + event("session.inbox.delivered", { sessionID: SESSION, inboxID: "msg_report_b" }), + reply("msg_reply_b", "REPLY_B"), + event("session.execution.succeeded", { sessionID: SESSION }), + ]).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: (request) => Effect.sync(() => void offers.push(request)), + take: Effect.never, + }), + ); + const collected: Array = []; + const ended = yield* Deferred.make(); + yield* runtime.events.pipe( + Stream.tap((event) => + Effect.gen(function* () { + collected.push(event); + if (collected.filter((entry) => entry.type === "turn.terminal").length === 2) { + yield* Deferred.succeed(ended, undefined); + } + }), + ), + Stream.runDrain, + Effect.forkScoped, + ); + yield* runtime.startTurn(withLineage(thread)); + yield* Effect.gen(function* () { + while (offers.length < 2) yield* Effect.yieldNow; + }).pipe(Effect.timeout("2 seconds"), Effect.orDie); + // A's continuation was cancelled before it started; B's runs. + yield* runtime.startTurn(continuationTurn(thread, offers[1]!, "wake-b")); + yield* Deferred.await(ended); + const continuationId = (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: OPENCODE_PROVIDER, + nativeTurnId: `${SESSION}:attempt:attempt:opencode2-adapter:wake-b`, + }); + assert.deepEqual( + collected.flatMap((event) => + event.type === "turn_item.updated" && + event.turnItem.type === "assistant_message" && + event.turnItem.providerTurnId === continuationId + ? [event.turnItem.text] + : [], + ), + ["REPLY_B"], + ); + // Nothing is left held for a continuation that will never run. + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + }).pipe(Effect.scoped, Effect.provide(IdAllocator.layer)), + ); + it.effect("ends a follow-up turn a steer joined exactly once", () => Effect.gen(function* () { const reportText = `\nCHILD_OK\n`; @@ -3918,19 +4029,7 @@ describe("OpenCode2 adapter", () => { yield* Effect.gen(function* () { while (offers.length === 0) yield* Effect.yieldNow; }).pipe(Effect.timeout("2 seconds"), Effect.orDie); - const followUp = { - ...withLineage(thread), - runId: RunId.make("run:opencode2-adapter:wake"), - runOrdinal: 2, - providerTurnOrdinal: 2, - attemptId: RunAttemptId.make("attempt:opencode2-adapter:wake"), - message: { - ...turnInput(thread).message, - messageId: MessageId.make("message:opencode2-adapter:wake"), - createdBy: "agent" as const, - creationSource: "provider" as const, - }, - }; + const followUp = continuationTurn(thread, offers[0]!); yield* runtime.startTurn(followUp); const followUpTurnId = (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ driver: OPENCODE_PROVIDER, diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts index 7716b75baaf9..cd117771923c 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts @@ -49,6 +49,7 @@ import { type OrchestrationV2Subagent, type OrchestrationV2TurnItem, type OrchestrationV2UserInputQuestion, + type MessageId, type ModelSelection, type ProviderApprovalDecision, type ProviderInstanceId, @@ -303,6 +304,8 @@ interface SubagentCall { * the continuation turn T3 opens for it takes them. */ interface Wake { + /** The message its continuation run starts from, which names the wake in that turn's input. */ + readonly messageId: MessageId | undefined; readonly events: Array; running: boolean; /** The background subagents whose end it answers, as its turn's notification names them. */ @@ -1658,13 +1661,14 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr */ const offerWake = Effect.fnUntraced(function* (state: ThreadState, wake: Wake) { const route = state.providerThread.appThreadId; - if (route === null) return; + if (route === null || wake.messageId === undefined) return; const notification = backgroundWorkNotification(wake.reports); yield* continuationRequests.offer({ threadId: route, providerThreadId: state.providerThread.id, driver, detail: wake.detail, + messageId: wake.messageId, ...(notification === null ? {} : { notification }), dispatchIfCurrent: (dispatch) => wake.dropped ? Effect.succeed(Option.none()) : Effect.map(dispatch, Option.some), @@ -2343,7 +2347,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const onWake = Effect.fnUntraced(function* (state: ThreadState) { const { delivered, stopped } = yield* takeReports(state); if (stopped) return; + const route = state.providerThread.appThreadId; const wake: Wake = { + // Only a thread's wake is offered a continuation; the ordinal is display metadata. + messageId: + route === null + ? undefined + : yield* idAllocator.allocate.message({ threadId: route, ordinal: 0 }), events: [], running: true, reports: delivered.map((entry) => entry.report), @@ -3377,8 +3387,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr /** * The continuation turn for an execution OpenCode started on its own: it - * takes the oldest one held and ends with it. One already taken (a user - * turn joined it) leaves nothing to run, so the turn ends at once. + * takes the wake it was dispatched for, named by its message, and ends + * with it. Wakes held before that one had their continuation cancelled or + * dropped before it started, so nothing will replay them: they are + * dropped. One already taken (a user turn joined it) leaves nothing to + * run, so the turn ends at once. */ const runWake = Effect.fnUntraced(function* ( state: ThreadState, @@ -3387,8 +3400,12 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr yield* beginTurn(state, turnInput, false); yield* lock.withPermit( Effect.gen(function* () { - const wake = state.wakes.shift(); - if (wake === undefined) return yield* finishTurn(state, { status: "completed" }); + const index = state.wakes.findIndex( + (held) => held.messageId === turnInput.message.messageId, + ); + if (index < 0) return yield* finishTurn(state, { status: "completed" }); + for (const skipped of state.wakes.splice(0, index)) skipped.dropped = true; + const wake = state.wakes.shift()!; const turn = state.active; if (turn !== undefined && wake.after !== undefined) turn.before = wake.after; // Its first report is where its history begins, so fork and rollback cut there. diff --git a/apps/server/src/orchestration-v2/ProviderContinuationRequests.ts b/apps/server/src/orchestration-v2/ProviderContinuationRequests.ts index e6af7fa791b5..8a5375f15c4b 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationRequests.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationRequests.ts @@ -41,6 +41,12 @@ export interface ProviderContinuationRequest { * reach the provider as a real prompt. */ readonly delivery?: "adapter_buffered" | "message_text"; + /** + * The message the continuation run starts from, when the adapter picks it: + * its turn input carries this id, so the adapter can tell which of its held + * wakes a run was dispatched for. Otherwise the service allocates one. + */ + readonly messageId?: MessageId; readonly dispatchIfCurrent?: ( effect: Effect.Effect, ) => Effect.Effect, E, R>; diff --git a/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts index 0c2df4e9cf44..78713299298c 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts @@ -180,6 +180,25 @@ describe("ProviderContinuationService", () => { }); }); + it.effect("dispatches a wake under the message its adapter named", () => { + return Effect.gen(function* () { + const dispatched = yield* Queue.unbounded(); + const messageId = MessageId.make("message-provider-continuation-adapter-named"); + yield* Effect.gen(function* () { + const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; + yield* requests.offer({ ...request(), messageId }); + const command = (yield* Queue.take(dispatched)) as { readonly messageId: MessageId }; + // The adapter matches the continuation run's message to the wake it holds. + assert.equal(command.messageId, messageId); + }).pipe( + Effect.provide( + layerTest({ dispatched, getThreadRecords: () => Effect.succeed(projection) }), + ), + Effect.scoped, + ); + }); + }); + it.effect("delivers a message_text wake as a real prompt, not a buffered wake", () => { return Effect.gen(function* () { const dispatched = yield* Queue.unbounded(); diff --git a/apps/server/src/orchestration-v2/ProviderContinuationService.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.ts index 2520847c7748..e1e157e245d9 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationService.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationService.ts @@ -147,10 +147,12 @@ export const layer = Layer.effectDiscard( } // The ordinal is display metadata only: allocate.message appends a // random UUID, so a stale projection read here cannot collide ids. - const messageId = yield* ids.allocate.message({ - threadId: request.threadId, - ordinal: projection.messages.length + 1, - }); + const messageId = + request.messageId ?? + (yield* ids.allocate.message({ + threadId: request.threadId, + ordinal: projection.messages.length + 1, + })); const commandId = CommandId.make(`provider-continuation:${messageId}`); const dispatch = threads.dispatch({ type: "message.dispatch", From b7f139ddfcb8b24b0bf93d1c00e1228732e147e6 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Tue, 6 Oct 2026 16:04:25 -0700 Subject: [PATCH 2/4] test(server): OpenCode 2 continuation-identity test waits on its offers, not a poll The test polled the offer count under a 2 s timeout. It now awaits a Deferred the second offer completes. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Adapters/OpenCode2AdapterV2.test.ts | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 3d82833413e1..087662378c2a 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -3877,6 +3877,7 @@ describe("OpenCode2 adapter", () => { const reply = (assistantMessageID: string, text: string) => event("session.text.ended", { sessionID: SESSION, assistantMessageID, ordinal: 0, text }); const offers: Array = []; + const bothOffered = yield* Deferred.make(); const { runtime, thread } = yield* resumed([ ...backgroundLaunch(CHILD), // The same turn starts a second background subagent. @@ -3908,7 +3909,12 @@ describe("OpenCode2 adapter", () => { event("session.execution.succeeded", { sessionID: SESSION }), ]).pipe( Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { - offer: (request) => Effect.sync(() => void offers.push(request)), + offer: (request) => + Effect.suspend(() => + offers.push(request) === 2 + ? Deferred.succeed(bothOffered, undefined).pipe(Effect.asVoid) + : Effect.void, + ), take: Effect.never, }), ); @@ -3927,9 +3933,7 @@ describe("OpenCode2 adapter", () => { Effect.forkScoped, ); yield* runtime.startTurn(withLineage(thread)); - yield* Effect.gen(function* () { - while (offers.length < 2) yield* Effect.yieldNow; - }).pipe(Effect.timeout("2 seconds"), Effect.orDie); + yield* Deferred.await(bothOffered); // A's continuation was cancelled before it started; B's runs. yield* runtime.startTurn(continuationTurn(thread, offers[1]!, "wake-b")); yield* Deferred.await(ended); From e9ef2581baa46bbc74eefee8f056ac9af1a6e5f1 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Tue, 6 Oct 2026 16:04:27 -0700 Subject: [PATCH 3/4] docs(server): note what an OpenCode 2 continuation drops when it skips earlier replies A continuation drops every reply held before its own, so cancelled continuations never pin the session. A continuation the user reordered ahead of an earlier one drops that one's reply too; the comment says so. Co-Authored-By: Claude Opus 5.5 (1M context) --- apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts index cd117771923c..c67735ad8af1 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts @@ -3404,6 +3404,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr (held) => held.messageId === turnInput.message.messageId, ); if (index < 0) return yield* finishTurn(state, { status: "completed" }); + // Trade-off: a continuation the user reordered ahead of an earlier one + // drops that one's reply too, so cancelled ones never pin the session. for (const skipped of state.wakes.splice(0, index)) skipped.dropped = true; const wake = state.wakes.shift()!; const turn = state.active; From a7034aab47b23e97fb58dfac2b37074113940475 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 10 Oct 2026 22:35:28 -0700 Subject: [PATCH 4/4] fix(server): OpenCode 2 stops subagents a dropped background reply started (#14760) Co-authored-by: Claude Opus 5.5 (1M context) --- .../Adapters/OpenCode2AdapterV2.test.ts | 901 +++++++++++++++++- .../Adapters/OpenCode2AdapterV2.testkit.ts | 6 +- .../src/server/v2/adapter.ts | 176 +++- 3 files changed, 1062 insertions(+), 21 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 589cf1c5c187..548ff5cb6643 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -48,6 +48,10 @@ import { t3McpServerName, } from "@t3tools/provider-opencode/testing"; import { openCode2ReplayRuntime } from "./OpenCode2AdapterV2.testkit.ts"; +import { + makeProviderReplayGate, + type ProviderReplayGate, +} from "@t3tools/provider-testing/replayGate"; const SESSION = "ses_f148ca2deffeJcwCnRQtb0YFNX"; const WORK = "/work/opencode2"; @@ -64,12 +68,18 @@ const reply = (operation: string, data: unknown): ProviderReplayEntry => ({ frame: { type: "sdk.response", operation, data }, }); const replyData = (operation: string, data: unknown) => reply(operation, { data }); -const event = (type: string, data: Record): ProviderReplayEntry => ({ +const event = ( + type: string, + data: Record, + /** Holds the event until a test's replay gate releases this label. */ + label?: string, +): ProviderReplayEntry => ({ type: "emit_inbound", frame: { type: "sdk.event", event: { id: `evt_${type.replaceAll(".", "")}0000`, created: 1, type, data, ...durable }, }, + ...(label === undefined ? {} : { label }), }); const durable = { durable: { aggregateID: SESSION, seq: 1, version: 1 } }; @@ -316,7 +326,11 @@ const colorForm = { */ const resumed = ( entries: ReadonlyArray, - options?: { readonly external?: boolean; readonly supervised?: boolean }, + options?: { + readonly external?: boolean; + readonly supervised?: boolean; + readonly replayGate?: ProviderReplayGate; + }, ) => Effect.gen(function* () { const runtime = yield* openCode2ReplayRuntime( @@ -335,7 +349,10 @@ const resumed = ( : []), ...entries, ]), - options?.external === undefined ? undefined : { external: options.external }, + { + ...(options?.external === undefined ? {} : { external: options.external }), + ...(options?.replayGate === undefined ? {} : { replayGate: options.replayGate }), + }, ); const thread = yield* runtime.resumeThread({ providerThread: providerThread(yield* DateTime.now), @@ -1746,6 +1763,88 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), ); + it.effect("still checks the server when a failed prompt's execution starts late", () => + Effect.gen(function* () { + const offered = yield* Deferred.make(); + const { runtime, thread } = yield* resumed([ + out("session.prompt", { sessionID: SESSION, text: "" }), + // The request failed, but the server took the prompt: its execution + // starts only after the turn already failed. + reply("session.prompt", { + status: 502, + body: { _tag: "UnknownError", message: "bad gateway" }, + }), + event("session.execution.started", { sessionID: SESSION }), + // That run is still going, so the next turn stops it and fails. + out("session.active"), + reply("session.active", { data: { [SESSION]: { type: "running" } } }), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + ]).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid), + take: Effect.never, + }), + ); + const ended = yield* terminals(runtime, 2); + yield* runtime.startTurn(turnInput(thread)).pipe(Effect.ignore); + // The late start has been read. + yield* Deferred.await(offered); + yield* runtime.startTurn(secondTurn(thread)); + const [first, second] = yield* Fiber.join(ended); + assert.equal(first?.status, "failed"); + assert.equal(second?.failure?.message, OPENCODE_2_STILL_STOPPING); + }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), + ); + + it.effect("still checks the server when a failed prompt starts after a reconnect", () => + Effect.gen(function* () { + const offered = yield* Deferred.make(); + const gate = makeProviderReplayGate(["drop"]); + const { runtime, thread } = yield* resumed( + [ + out("session.prompt", { sessionID: SESSION, text: "" }), + reply("session.prompt", { + status: 502, + body: { _tag: "UnknownError", message: "bad gateway" }, + }), + // The stream drops before the prompt's execution starts, and the + // reconnect finds the session active: the prompt's run is still to come. + event("session.usage.updated", { sessionID: SESSION }, "drop"), + { type: "runtime_exit", status: "success" } as const, + out("event.subscribe"), + out("session.active"), + replyData("session.active", { [SESSION]: { type: "running" } }), + out("permission.list", { sessionID: SESSION }), + replyData("permission.list", []), + out("session.form.list", { sessionID: SESSION }), + replyData("session.form.list", []), + // Its own start: that run is still going, so the next turn stops it and fails. + event("session.execution.started", { sessionID: SESSION }), + out("session.active"), + reply("session.active", { data: { [SESSION]: { type: "running" } } }), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + ], + { replayGate: gate }, + ).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid), + take: Effect.never, + }), + ); + const ended = yield* terminals(runtime, 2); + yield* runtime.startTurn(turnInput(thread)).pipe(Effect.ignore); + gate.release("drop"); + // The late start has been read. + yield* Deferred.await(offered); + yield* runtime.startTurn(secondTurn(thread)); + const [first, second] = yield* Fiber.join(ended); + assert.equal(first?.status, "failed"); + assert.equal(second?.failure?.message, OPENCODE_2_STILL_STOPPING); + }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), + ); + it.effect("prompts again without a check after the server refused a prompt", () => Effect.gen(function* () { const { runtime, thread } = yield* resumed([ @@ -1786,6 +1885,357 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), ); + it.effect("ends a background reply's continuation after a lost stream took a Stop's end", () => + Effect.gen(function* () { + const offered = + yield* Deferred.make(); + const gate = makeProviderReplayGate(["drop", "reply"]); + const { runtime, thread } = yield* resumed( + [ + ...backgroundLaunch(CHILD), + // Restarting the turn stops only the parent, and the server never answers. + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", ""), + // The restarted turn finds the parent still stopping and fails. + out("session.active"), + replyData("session.active", { [SESSION]: { type: "running" } }), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + // The stream drops before the parent's end; the reconnect finds it idle. + event("session.usage.updated", { sessionID: SESSION }, "drop"), + { type: "runtime_exit", status: "success" } as const, + out("event.subscribe"), + out("session.active"), + replyData("session.active", { [CHILD]: { type: "running" } }), + out("permission.list", { sessionID: CHILD }), + replyData("permission.list", []), + out("session.form.list", { sessionID: CHILD }), + replyData("session.form.list", []), + // The subagent ends and OpenCode answers its report on its own. + event("session.execution.succeeded", { sessionID: CHILD }), + event("session.inbox.enqueued", { + inboxID: "msg_report", + sessionID: SESSION, + item: { + type: "synthetic", + payload: { + text: `\nCHILD_OK\n`, + description: "Sleep", + metadata: { + source: "subagent", + childID: CHILD, + agent: "General", + state: "completed", + }, + }, + delivery: "steer", + }, + }), + event("session.execution.started", { sessionID: SESSION }), + event( + "session.text.ended", + { + sessionID: SESSION, + assistantMessageID: "msg_followup", + ordinal: 0, + text: "CHILD_OK", + }, + "reply", + ), + event("session.execution.succeeded", { sessionID: SESSION }), + // The subagent's session runs again: its turn marks that the reply's end was handled. + event("session.execution.started", { sessionID: CHILD }), + ], + { replayGate: gate }, + ).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: (request) => Deferred.succeed(offered, request).pipe(Effect.asVoid), + take: Effect.never, + }), + ); + // The single reader of the runtime's events. + const attached = yield* Deferred.make(); + const collected = yield* runtime.events.pipe( + Stream.tap((event) => + event.type === "subagent.updated" && event.subagent.childThreadId !== null + ? Deferred.succeed(attached, undefined) + : Effect.void, + ), + Stream.takeUntil( + (event) => + event.type === "provider_turn.updated" && + event.providerTurn.nativeTurnRef?.nativeId?.startsWith(`${CHILD}:turn:`) === true, + ), + Stream.runCollect, + Effect.forkScoped, + ); + yield* runtime.startTurn(withLineage(thread)); + yield* Deferred.await(attached); + const interrupt = yield* runtime + .interruptTurn({ providerThread: thread, providerTurnId: yield* providerTurnId }) + .pipe(Effect.forkScoped); + yield* TestClock.adjust("11 seconds"); + yield* Fiber.join(interrupt); + yield* runtime.startTurn({ ...secondTurn(thread), appThread: withLineage(thread).appThread }); + gate.release("drop"); + yield* runtime.startTurn({ + ...continuationTurn(thread, yield* Deferred.await(offered)), + runOrdinal: 3, + providerTurnOrdinal: 3, + }); + gate.release("reply"); + // The reply's own end finishes its continuation: the lost Stop's end does not take it. + const events = [...(yield* Fiber.join(collected))]; + assert.deepEqual( + events.flatMap((event) => (event.type === "turn.terminal" ? [event.status] : [])), + ["interrupted", "failed", "completed"], + ); + // The continuation replayed the reply. + assert.include( + events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message" + ? [event.turnItem.text] + : [], + ), + "CHILD_OK", + ); + }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), + ); + + it.effect("lets a prompt join a background reply after a lost stream took a Stop's end", () => + Effect.gen(function* () { + const offered = yield* Deferred.make(); + const gate = makeProviderReplayGate(["drop"]); + const { runtime, thread } = yield* resumed( + [ + ...backgroundLaunch(CHILD), + // Restarting the turn stops only the parent, and the server never answers. + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", ""), + // The restarted turn finds the parent still stopping and fails. + out("session.active"), + replyData("session.active", { [SESSION]: { type: "running" } }), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + // The stream drops before the parent's end; the reconnect finds it idle. + event("session.usage.updated", { sessionID: SESSION }, "drop"), + { type: "runtime_exit", status: "success" } as const, + out("event.subscribe"), + out("session.active"), + replyData("session.active", { [CHILD]: { type: "running" } }), + out("permission.list", { sessionID: CHILD }), + replyData("permission.list", []), + out("session.form.list", { sessionID: CHILD }), + replyData("session.form.list", []), + // The subagent ends and OpenCode answers its report on its own. + event("session.execution.succeeded", { sessionID: CHILD }), + event("session.inbox.enqueued", { + inboxID: "msg_report", + sessionID: SESSION, + item: { + type: "synthetic", + payload: { + text: `\nCHILD_OK\n`, + description: "Sleep", + metadata: { + source: "subagent", + childID: CHILD, + agent: "General", + state: "completed", + }, + }, + delivery: "steer", + }, + }), + event("session.execution.started", { sessionID: SESSION }), + // A prompt sent while that reply runs joins it, with no check first. + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + event("session.text.ended", { + sessionID: SESSION, + assistantMessageID: "msg_followup", + ordinal: 0, + text: "CHILD_OK", + }), + event("session.execution.succeeded", { sessionID: SESSION }), + ], + { replayGate: gate }, + ).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid), + take: Effect.never, + }), + ); + // The single reader of the runtime's events. + const attached = yield* Deferred.make(); + const ended = yield* runtime.events.pipe( + Stream.tap((event) => + event.type === "subagent.updated" && event.subagent.childThreadId !== null + ? Deferred.succeed(attached, undefined) + : Effect.void, + ), + Stream.filter( + ( + event, + ): event is Extract => + event.type === "turn.terminal", + ), + Stream.take(3), + Stream.runCollect, + Effect.forkScoped, + ); + yield* runtime.startTurn(withLineage(thread)); + yield* Deferred.await(attached); + const interrupt = yield* runtime + .interruptTurn({ providerThread: thread, providerTurnId: yield* providerTurnId }) + .pipe(Effect.forkScoped); + yield* TestClock.adjust("11 seconds"); + yield* Fiber.join(interrupt); + yield* runtime.startTurn({ ...secondTurn(thread), appThread: withLineage(thread).appThread }); + gate.release("drop"); + // The reply has started. + yield* Deferred.await(offered); + yield* runtime.startTurn({ + ...withLineage(thread), + runId: RunId.make("run:opencode2-adapter:3"), + runOrdinal: 3, + providerTurnOrdinal: 3, + attemptId: RunAttemptId.make("attempt:opencode2-adapter:3"), + }); + // The reply's start cleared the Stop's mark, so its end finishes the joined turn. + assert.deepEqual( + [...(yield* Fiber.join(ended))].map((terminal) => terminal.status), + ["interrupted", "failed", "completed"], + ); + }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), + ); + + /** + * A prompt fails without a clear rejection, OpenCode never took it, and a + * background subagent's report then wakes the parent into a reply. With + * `endedFirst` that reply ends before its continuation takes it. + */ + const replyAfterUnclearFailure = (endedFirst: boolean) => + Effect.gen(function* () { + const offered = + yield* Deferred.make(); + const gate = makeProviderReplayGate(endedFirst ? ["wake"] : ["wake", "reply"]); + const { runtime, thread } = yield* resumed( + [ + ...backgroundLaunch(CHILD), + event("session.execution.succeeded", { sessionID: SESSION }), + out("session.update", { sessionID: CHILD, permissions: t3Rules }), + reply("session.update", null), + out("session.prompt", { sessionID: SESSION, text: "" }), + reply("session.prompt", { + status: 502, + body: { _tag: "UnknownError", message: "bad gateway" }, + }), + // The subagent ends and OpenCode answers its report on its own. + event("session.execution.succeeded", { sessionID: CHILD }, "wake"), + event("session.inbox.enqueued", { + inboxID: "msg_report", + sessionID: SESSION, + item: { + type: "synthetic", + payload: { + text: `\nCHILD_OK\n`, + description: "Sleep", + metadata: { + source: "subagent", + childID: CHILD, + agent: "General", + state: "completed", + }, + }, + delivery: "steer", + }, + }), + event("session.execution.started", { sessionID: SESSION }), + event( + "session.text.ended", + { + sessionID: SESSION, + assistantMessageID: "msg_followup", + ordinal: 0, + text: "CHILD_OK", + }, + endedFirst ? undefined : "reply", + ), + event("session.execution.succeeded", { sessionID: SESSION }), + // The subagent's session runs again: a turn of its own, after the reply's end. + event("session.execution.started", { sessionID: CHILD }), + ], + { replayGate: gate }, + ).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: (request) => Deferred.succeed(offered, request).pipe(Effect.asVoid), + take: Effect.never, + }), + ); + // The single reader of the runtime's events. + const launched = yield* Deferred.make(); + const childRan = yield* Deferred.make(); + const ended: Array = []; + const allEnded = yield* Deferred.make(); + yield* runtime.events.pipe( + Stream.tap((event) => + Effect.gen(function* () { + if ( + event.type === "provider_turn.updated" && + event.providerTurn.nativeTurnRef?.nativeId?.startsWith(`${CHILD}:turn:`) === true + ) { + yield* Deferred.succeed(childRan, undefined); + } + if (event.type !== "turn_item.updated" && event.type !== "turn.terminal") return; + ended.push(event); + if (event.type === "turn.terminal") yield* Deferred.succeed(launched, undefined); + if (ended.filter((entry) => entry.type === "turn.terminal").length === 3) { + yield* Deferred.succeed(allEnded, undefined); + } + }), + ), + Stream.runDrain, + Effect.forkScoped, + ); + yield* runtime.startTurn(withLineage(thread)); + yield* Deferred.await(launched); + yield* runtime + .startTurn({ ...secondTurn(thread), appThread: withLineage(thread).appThread }) + .pipe(Effect.ignore); + gate.release("wake"); + // Everything up to the subagent's next run has been read: the reply ended. + if (endedFirst) yield* Deferred.await(childRan); + yield* runtime.startTurn({ + ...continuationTurn(thread, yield* Deferred.await(offered)), + runOrdinal: 3, + providerTurnOrdinal: 3, + }); + if (!endedFirst) gate.release("reply"); + // The reply's own end finishes its continuation: the failed prompt's mark does not take it. + yield* Deferred.await(allEnded); + assert.deepEqual( + ended.flatMap((event) => (event.type === "turn.terminal" ? [event.status] : [])), + ["completed", "failed", "completed"], + ); + assert.include( + ended.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message" + ? [event.turnItem.text] + : [], + ), + "CHILD_OK", + ); + }).pipe(Effect.scoped, Effect.provide(TestClock.layer())); + + it.effect("ends a background reply's continuation after a prompt failed unclearly", () => + replyAfterUnclearFailure(false), + ); + + it.effect("ends a finished background reply's continuation after a prompt failed unclearly", () => + replyAfterUnclearFailure(true), + ); + it.effect("does not report a Stop the server says did nothing", () => Effect.gen(function* () { const { runtime, thread } = yield* resumed([ @@ -4117,6 +4567,451 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + it.effect("keeps a subagent a stopped background reply started off the next turn", () => + Effect.gen(function* () { + const CHILD_B = "ses_f1485c529ffeBBBBBBBBBBBBBB"; + const SPAWNED = "ses_f1485c529ffeDDDDDDDDDDDDDD"; + const NESTED = "ses_f1485c529ffeNNNNNNNNNNNNNN"; + const toolB = { sessionID: SESSION, assistantMessageID: "msg_assistant", id: "call-b" }; + const spawn = { sessionID: SESSION, assistantMessageID: "msg_reply", id: "call-spawn" }; + const offers: Array = []; + const gate = makeProviderReplayGate(["settled"]); + const { runtime, thread } = yield* resumed( + [ + ...backgroundLaunch(CHILD), + // The same turn starts a second background subagent, which runs on. + event("session.tool.input.started", { ...toolB, name: "subagent" }), + event("session.tool.called", { + ...toolB, + name: "subagent", + input: { description: "Sleep", prompt: "sleep", background: true }, + executed: false, + }), + event("session.created", { ...childCreated(CHILD_B), sessionID: CHILD_B }), + event("session.tool.progress", { + ...toolB, + metadata: { sessionID: CHILD_B, status: "running" }, + }), + event("session.execution.succeeded", { sessionID: SESSION }), + event("session.execution.started", { sessionID: CHILD_B }), + // The first subagent's report wakes the parent into a reply T3 holds + // for its continuation. + event("session.execution.succeeded", { sessionID: CHILD }), + event("session.inbox.enqueued", { + inboxID: "msg_report", + sessionID: SESSION, + item: { + type: "synthetic", + payload: { + text: `\nCHILD_OK\n`, + description: "Sleep", + metadata: { + source: "subagent", + childID: CHILD, + agent: "General", + state: "completed", + }, + }, + delivery: "steer", + }, + }), + event("session.execution.started", { sessionID: SESSION }), + // The reply starts a background subagent of its own, named only by + // the reply's held events. + event("session.tool.input.started", { ...spawn, name: "subagent" }), + event("session.tool.called", { + ...spawn, + name: "subagent", + input: { description: "Sleep", prompt: "sleep", background: true }, + executed: false, + }), + event("session.created", { ...childCreated(SPAWNED), sessionID: SPAWNED }), + event("session.tool.progress", { + ...spawn, + metadata: { sessionID: SPAWNED, status: "running" }, + }), + event("session.execution.succeeded", { sessionID: SESSION }), + // The second subagent's text, shown live, marks that the reply has ended. + event("session.text.ended", { + sessionID: CHILD_B, + assistantMessageID: "msg_child_b", + ordinal: 0, + text: "B_WORKING", + }), + // The user's Stop stops the second subagent and drops the held reply. + out("session.interrupt", { sessionID: CHILD_B }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: CHILD_B }), + // The subagent only the dropped reply named is stopped too; that Stop + // fails, so it runs on. + out("session.interrupt", { sessionID: SPAWNED }), + reply("session.interrupt", { + status: 500, + body: { _tag: "UnknownError", message: "interrupt failed" }, + }), + // An unrelated prompt runs while that subagent, and a subagent it + // starts, ask for permissions, which no turn shows: they are refused. + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + event("session.execution.started", { sessionID: SESSION }), + event("permission.asked", { ...shellAsk.data, id: "per_spawned", sessionID: SPAWNED }), + out("permission.reply", { + sessionID: SPAWNED, + requestID: "per_spawned", + decision: "reject", + }), + reply("permission.reply", null), + event("session.created", { + ...childCreated(NESTED), + sessionID: NESTED, + parentID: SPAWNED, + }), + event("permission.asked", { ...shellAsk.data, id: "per_nested", sessionID: NESTED }), + out("permission.reply", { + sessionID: NESTED, + requestID: "per_nested", + decision: "reject", + }), + reply("permission.reply", null), + event("session.text.ended", { + sessionID: SESSION, + assistantMessageID: "msg_unrelated", + ordinal: 0, + text: "UNRELATED", + }), + event("session.execution.succeeded", { sessionID: SESSION }), + // The next Stop tries the subagent again, and the one it started, and reaches both. + out("session.interrupt", { sessionID: SPAWNED }), + reply("session.interrupt", { interrupted: true }), + out("session.interrupt", { sessionID: NESTED }), + reply("session.interrupt", { interrupted: true }), + // OpenCode reports the stopped subagent to the thread's session and + // wakes it; that execution answers no turn, so it is stopped and no + // continuation is offered for it. + event("session.execution.interrupted", { sessionID: SPAWNED }), + event("session.inbox.enqueued", { + inboxID: "msg_report_spawned", + sessionID: SESSION, + item: { + type: "synthetic", + payload: { + text: `\n`, + description: "Sleep", + metadata: { + source: "subagent", + childID: SPAWNED, + agent: "General", + state: "cancelled", + }, + }, + delivery: "steer", + }, + }), + event("session.execution.started", { sessionID: SESSION }), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: SESSION }, "settled"), + ], + { replayGate: gate }, + ).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: (request) => Effect.sync(() => void offers.push(request)), + take: Effect.never, + }), + ); + const collected: Array = []; + const replyEnded = yield* Deferred.make(); + const bothEnded = yield* Deferred.make(); + yield* runtime.events.pipe( + Stream.tap((event) => + Effect.gen(function* () { + collected.push(event); + if (event.type === "turn_item.updated" && event.turnItem.type === "assistant_message") { + if (event.turnItem.text === "B_WORKING") + yield* Deferred.succeed(replyEnded, undefined); + } + if (collected.filter((entry) => entry.type === "turn.terminal").length === 2) { + yield* Deferred.succeed(bothEnded, undefined); + } + }), + ), + Stream.runDrain, + Effect.forkScoped, + ); + yield* runtime.startTurn(withLineage(thread)); + yield* Deferred.await(replyEnded); + assert.lengthOf(offers, 1); + // The user stops the thread before the reply's continuation runs. + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + yield* runtime.startTurn({ ...secondTurn(thread), appThread: withLineage(thread).appThread }); + yield* Deferred.await(bothEnded); + // A later Stop reaches both subagents; once stopped they are not tried + // again (the replay fails on any further request). + for (let stops = 0; stops < 2; stops++) { + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + } + // The stopped subagent's report wakes nothing T3 runs. + yield* Effect.promise(() => gate.waitForReached("settled")); + gate.release("settled"); + assert.lengthOf(offers, 1); + // The unrelated turn shows no request it never made. + assert.deepEqual( + collected.flatMap((event) => + event.type === "runtime_request.updated" + ? [`${event.runtimeRequest.nativeRequestRef?.nativeId}:${event.runtimeRequest.status}`] + : [], + ), + [], + ); + }).pipe(Effect.scoped), + ); + + it.effect( + "keeps every subagent a stopped background reply's subagent started off the next turn", + () => + Effect.gen(function* () { + const CHILD_B = "ses_f1485c529ffeBBBBBBBBBBBBBB"; + const SPAWNED = "ses_f1485c529ffeDDDDDDDDDDDDDD"; + const NESTED = "ses_f1485c529ffeNNNNNNNNNNNNNN"; + const DEEP = "ses_f1485c529ffeEEEEEEEEEEEEEE"; + const toolB = { sessionID: SESSION, assistantMessageID: "msg_assistant", id: "call-b" }; + const spawn = { sessionID: SESSION, assistantMessageID: "msg_reply", id: "call-spawn" }; + const { runtime, thread } = yield* resumed([ + ...backgroundLaunch(CHILD), + // The same turn starts a second background subagent, which runs on. + event("session.tool.input.started", { ...toolB, name: "subagent" }), + event("session.tool.called", { + ...toolB, + name: "subagent", + input: { description: "Sleep", prompt: "sleep", background: true }, + executed: false, + }), + event("session.created", { ...childCreated(CHILD_B), sessionID: CHILD_B }), + event("session.tool.progress", { + ...toolB, + metadata: { sessionID: CHILD_B, status: "running" }, + }), + event("session.execution.succeeded", { sessionID: SESSION }), + event("session.execution.started", { sessionID: CHILD_B }), + // The first subagent's report wakes the parent into a reply T3 holds. + event("session.execution.succeeded", { sessionID: CHILD }), + event("session.inbox.enqueued", { + inboxID: "msg_report", + sessionID: SESSION, + item: { + type: "synthetic", + payload: { + text: `\nCHILD_OK\n`, + description: "Sleep", + metadata: { + source: "subagent", + childID: CHILD, + agent: "General", + state: "completed", + }, + }, + delivery: "steer", + }, + }), + event("session.execution.started", { sessionID: SESSION }), + // The held reply starts a subagent, which starts one, which starts another. + event("session.tool.input.started", { ...spawn, name: "subagent" }), + event("session.tool.called", { + ...spawn, + name: "subagent", + input: { description: "Sleep", prompt: "sleep", background: true }, + executed: false, + }), + event("session.created", { ...childCreated(SPAWNED), sessionID: SPAWNED }), + event("session.tool.progress", { + ...spawn, + metadata: { sessionID: SPAWNED, status: "running" }, + }), + event("session.created", { + ...childCreated(NESTED), + sessionID: NESTED, + parentID: SPAWNED, + }), + event("session.created", { ...childCreated(DEEP), sessionID: DEEP, parentID: NESTED }), + event("session.execution.succeeded", { sessionID: SESSION }), + // The second subagent's text, shown live, marks that the reply has ended. + event("session.text.ended", { + sessionID: CHILD_B, + assistantMessageID: "msg_child_b", + ordinal: 0, + text: "B_WORKING", + }), + // The user's Stop stops the second subagent and drops the held reply, + // then each subagent under the reply, the deepest of which runs on. + out("session.interrupt", { sessionID: CHILD_B }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: CHILD_B }), + out("session.interrupt", { sessionID: SPAWNED }), + reply("session.interrupt", { interrupted: true }), + out("session.interrupt", { sessionID: NESTED }), + reply("session.interrupt", { interrupted: true }), + out("session.interrupt", { sessionID: DEEP }), + reply("session.interrupt", { + status: 500, + body: { _tag: "UnknownError", message: "interrupt failed" }, + }), + // An unrelated prompt runs while the deepest asks for a permission, + // which no turn shows: it is refused. + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + event("session.execution.started", { sessionID: SESSION }), + event("permission.asked", { ...shellAsk.data, id: "per_deep", sessionID: DEEP }), + out("permission.reply", { sessionID: DEEP, requestID: "per_deep", decision: "reject" }), + reply("permission.reply", null), + event("session.execution.succeeded", { sessionID: SESSION }), + ]); + const collected: Array = []; + const replyEnded = yield* Deferred.make(); + const bothEnded = yield* Deferred.make(); + yield* runtime.events.pipe( + Stream.tap((event) => + Effect.gen(function* () { + collected.push(event); + if ( + event.type === "turn_item.updated" && + event.turnItem.type === "assistant_message" && + event.turnItem.text === "B_WORKING" + ) { + yield* Deferred.succeed(replyEnded, undefined); + } + if (collected.filter((entry) => entry.type === "turn.terminal").length === 2) { + yield* Deferred.succeed(bothEnded, undefined); + } + }), + ), + Stream.runDrain, + Effect.forkScoped, + ); + yield* runtime.startTurn(withLineage(thread)); + yield* Deferred.await(replyEnded); + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + yield* runtime.startTurn({ + ...secondTurn(thread), + appThread: withLineage(thread).appThread, + }); + yield* Deferred.await(bothEnded); + // The unrelated turn shows no request it never made. + assert.deepEqual( + collected.flatMap((event) => + event.type === "runtime_request.updated" + ? [ + `${event.runtimeRequest.nativeRequestRef?.nativeId}:${event.runtimeRequest.status}`, + ] + : [], + ), + [], + ); + }).pipe(Effect.scoped), + ); + + it.effect("ends a continuation while a skipped reply's subagent does not answer its Stop", () => + Effect.gen(function* () { + const CHILD_B = "ses_f1485c529ffeBBBBBBBBBBBBBB"; + const SPAWNED = "ses_f1485c529ffeDDDDDDDDDDDDDD"; + const toolB = { sessionID: SESSION, assistantMessageID: "msg_assistant", id: "call-b" }; + const spawn = { sessionID: SESSION, assistantMessageID: "msg_reply_a", id: "call-spawn" }; + const report = (inboxID: string, child: string) => + event("session.inbox.enqueued", { + inboxID, + sessionID: SESSION, + item: { + type: "synthetic", + payload: { + text: `\nOK\n`, + description: "Sleep", + metadata: { + source: "subagent", + childID: child, + agent: "General", + state: "completed", + }, + }, + delivery: "steer", + }, + }); + const offers: Array = []; + const bothOffered = yield* Deferred.make(); + const { runtime, thread } = yield* resumed([ + ...backgroundLaunch(CHILD), + event("session.tool.input.started", { ...toolB, name: "subagent" }), + event("session.tool.called", { + ...toolB, + name: "subagent", + input: { description: "Sleep", prompt: "sleep", background: true }, + executed: false, + }), + event("session.created", { ...childCreated(CHILD_B), sessionID: CHILD_B }), + event("session.tool.progress", { + ...toolB, + metadata: { sessionID: CHILD_B, status: "running" }, + }), + event("session.execution.succeeded", { sessionID: SESSION }), + // Reply A starts a subagent of its own, named only by its held events. + event("session.execution.succeeded", { sessionID: CHILD }), + report("msg_report_a", CHILD), + event("session.execution.started", { sessionID: SESSION }), + event("session.tool.input.started", { ...spawn, name: "subagent" }), + event("session.tool.called", { + ...spawn, + name: "subagent", + input: { description: "Sleep", prompt: "sleep", background: true }, + executed: false, + }), + event("session.created", { ...childCreated(SPAWNED), sessionID: SPAWNED }), + event("session.execution.succeeded", { sessionID: SESSION }), + event("session.execution.succeeded", { sessionID: CHILD_B }), + report("msg_report_b", CHILD_B), + event("session.execution.started", { sessionID: SESSION }), + event("session.text.ended", { + sessionID: SESSION, + assistantMessageID: "msg_reply_b", + ordinal: 0, + text: "REPLY_B", + }), + event("session.execution.succeeded", { sessionID: SESSION }), + // B's continuation skips reply A and stops its subagent, which never answers. + out("session.interrupt", { sessionID: SPAWNED }), + reply("session.interrupt", ""), + ]).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: (request) => + Effect.suspend(() => + offers.push(request) === 2 + ? Deferred.succeed(bothOffered, undefined).pipe(Effect.asVoid) + : Effect.void, + ), + take: Effect.never, + }), + ); + const ended = yield* terminals(runtime, 2); + yield* runtime.startTurn(withLineage(thread)); + yield* Deferred.await(bothOffered); + yield* runtime.startTurn(continuationTurn(thread, offers[1]!, "wake-b")); + // B's reply replays and ends its turn while that Stop still waits. + assert.deepEqual( + [...(yield* Fiber.join(ended))].map((terminal) => terminal.status), + ["completed", "completed"], + ); + }).pipe(Effect.scoped), + ); + it.effect("replays only its own background reply into a continuation turn", () => Effect.gen(function* () { const CHILD_B = "ses_f1485c529ffeBBBBBBBBBBBBBB"; diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts index 7eb295997483..db251e005d79 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts @@ -286,7 +286,11 @@ function layerRegistry( */ export const openCode2ReplayRuntime = ( entries: ReadonlyArray, - options?: { readonly external?: boolean; readonly borrowers?: { current: number } }, + options?: { + readonly external?: boolean; + readonly borrowers?: { current: number }; + readonly replayGate?: ProviderReplayGate; + }, ) => Effect.gen(function* () { const adapter = yield* makeReplayAdapter( diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index 53dc63aed451..785bf1a06e70 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -320,6 +320,11 @@ interface Wake { readonly after: string | undefined; /** The first report it delivered, where fork and rollback cut before its turn. */ readonly first: string | undefined; + /** + * Subagent sessions OpenCode announced while it was held: their calls are + * in its events, so only its replay names them. + */ + readonly children: Set; } interface OpenBlock { @@ -343,6 +348,12 @@ interface ThreadState { * so the next execution end belongs to that run; it clears this and ends no turn. */ unsettled: boolean; + /** + * Whether the run `unsettled` waits on had started when it was marked. A + * prompt whose request failed may start its run late; that start is the + * run's own, and only a start after it means its end was lost. + */ + unsettledStarted: boolean; /** The session's location, where its agents' path rules are read. */ directory: string; /** @@ -904,6 +915,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const childOwners = new Map(); // What OpenCode announced about a subagent's session, until its call names it. const announced = new Map["data"]>(); + // Subagent sessions a dropped held execution started, and the subagents + // they start, by the session the first of them was called from. Nothing + // names them, so they no longer route to its thread; they stay here until + // a Stop of that thread stops them or they are gone. + const strays = new Map(); // Sessions of these threads with an execution running, seen on the stream. const busy = new Set(); const pending = new Map(); @@ -926,6 +942,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const emit = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); const ownerOf = (sessionId: string) => childOwners.get(sessionId) ?? threads.get(sessionId); + /** Marks a run T3 ended while OpenCode may still run it (see `unsettled`). */ + const markUnsettled = (state: ThreadState) => { + state.unsettled = true; + state.unsettledStarted = busy.has(state.sessionId); + }; const newThreadState = ( sessionId: string, @@ -939,6 +960,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr active: undefined, model: undefined, unsettled: false, + unsettledStarted: false, directory, agent: "build", title: undefined, @@ -1842,7 +1864,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr reason, }); if (state.active === undefined) return; - state.unsettled = true; + markUnsettled(state); yield* finishTurn(state, { status: "failed", failure: makeProviderFailure({ @@ -1939,7 +1961,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const [permissions, forms] = listed.value; if (permissions.length === 0 && forms.length === 0) return; // The stopped run's end is not the next turn's. - state.unsettled = true; + markUnsettled(state); for (const request of permissions) { yield* stopStaleRequest(state.sessionId, { type: "permission", id: request.id }); } @@ -2124,7 +2146,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr // goes on. A subagent's session has no Stop end to skip. const asker = threads.get(form.sessionID) ?? state; if (!cancelled) return yield* abandonRequest(asker, "form cancel failed"); - if (asker.subagent === undefined) asker.unsettled = true; + if (asker.subagent === undefined) markUnsettled(asker); yield* finishTurn(asker, { status: "failed", failure: makeProviderFailure({ @@ -2145,6 +2167,15 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr | { readonly type: "form"; readonly form: NativeForm }, ) { const asking = asked.type === "permission" ? asked.request.sessionID : asked.form.sessionID; + // A stray belongs to no turn: its request is refused, which ends its run. + if (strays.has(asking)) { + return yield* stopStaleRequest( + asking, + asked.type === "permission" + ? { type: "permission", id: asked.request.id } + : { type: "form", id: asked.form.id }, + ); + } const state = ownerOf(asking); if (state === undefined) return; const target = requestTurn(asking); @@ -2362,7 +2393,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr delivered.every((entry) => state.stoppedChildren.has(entry.childId)); for (const entry of delivered) state.stoppedChildren.delete(entry.childId); if (stopped) { - state.unsettled = true; + markUnsettled(state); yield* client.session .interrupt({ sessionID: Session.ID.make(state.sessionId) }) .pipe(Effect.timeout(INTERRUPT_TIMEOUT), Effect.ignore({ log: true })); @@ -2387,6 +2418,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr dropped: false, after: delivered.at(-1)?.inboxId, first: delivered[0]?.inboxId, + children: new Set(), }; state.wakes.push(wake); yield* Effect.logInfo("OpenCode started a turn on its own; asking for a continuation.", { @@ -2406,6 +2438,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ) { const childId = stringField(payload.metadata, "childID"); if (stringField(payload.metadata, "source") !== "subagent" || childId === undefined) return; + // A stray that reports has ended: no Stop needs to reach it. + strays.delete(childId); const call = [...state.calls.values()].find( (candidate) => candidate.child?.sessionId === childId, ); @@ -2494,6 +2528,16 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr event.type === "session.execution.started" || event.type === "unreadable.execution.started" ) { + // A session runs one execution at a time, each opening with `started` + // on this ordered stream. Once the run `unsettled` waits on has + // started, a later start means that run is over: an end not seen by + // now was lost with a dropped stream, and the next end is this + // execution's own. A run not yet started when it was marked starts here. + const state = threads.get(sessionId); + if (state?.unsettled === true) { + if (state.unsettledStarted) state.unsettled = false; + else state.unsettledStarted = true; + } busy.add(sessionId); } else if (event.type === "unreadable.execution.ended" || executionEnd(event.type)) { busy.delete(sessionId); @@ -2501,10 +2545,17 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr } // A subagent's session: named on its parent's `subagent` call just after. if (event.type === "session.created" && event.data.parentID !== undefined) { + // A stray's subagent is a stray too: no call of the thread's names it. + const strayCaller = strays.get(event.data.parentID); + if (strayCaller !== undefined) { + strays.set(event.data.sessionID, strayCaller); + return; + } const owner = ownerOf(event.data.parentID); if (owner !== undefined) { childOwners.set(event.data.sessionID, owner); announced.set(event.data.sessionID, event.data); + holderOf(event.data.parentID)?.wakes.at(-1)?.children.add(event.data.sessionID); } return; } @@ -2525,14 +2576,20 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr return yield* route(event, sessionId); }); - /** Handles one event for its session, as the turn running there sees it. */ + /** + * Handles one event for its session, as the turn running there sees it. + * `held` is set for a held execution's events as its turn replays them. + */ const route = Effect.fnUntraced(function* ( event: OpenCode2Client.OpenCode2StreamEvent, sessionId: string | undefined, + held = false, ) { // The end of the run a timed-out Stop left behind; no turn is its own. + // A held execution's end is its own: any such run was over, or was this + // execution, before it started. const ended = - event.type === "unreadable.execution.ended" || executionEnd(event.type) + !held && (event.type === "unreadable.execution.ended" || executionEnd(event.type)) ? threads.get(sessionId ?? "") : undefined; if (ended?.unsettled === true) { @@ -2787,10 +2844,28 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const background = [...threads.values()].filter( (state) => state.subagent === undefined && hasBackground(state), ); - if (running.length === 0 && background.length === 0 && busy.size === 0) return; + const marked = [...threads.values()].some((state) => state.unsettled); + if ( + running.length === 0 && + background.length === 0 && + busy.size === 0 && + !marked && + strays.size === 0 + ) { + return; + } const active = yield* client.session.active(); // An execution that ended while the stream was down never said so. for (const sessionId of busy) if (!(sessionId in active)) busy.delete(sessionId); + // A run `unsettled` waits on is over once its session is idle: any start + // from here is a new execution's, even if the run's own start was lost in + // the gap. A session still active may not have started that run yet, so + // its next start may still be the run's own; its end clears the mark. + for (const state of threads.values()) { + if (state.unsettled && !(state.sessionId in active)) state.unsettledStarted = true; + } + // A stray that stopped meanwhile runs nothing to stop. + for (const childId of strays.keys()) if (!(childId in active)) strays.delete(childId); for (const [sessionId, state] of running) { const turn = state.active; if (turn === undefined) continue; @@ -2822,10 +2897,14 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr } // A subagent's session holds reports of its own background subagents. for (const session of sessionsOf(state)) { - for (const wake of session.wakes.splice(0)) wake.dropped = true; + 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); + } // A request asked while the stream was down was never shown, and the // run waits on it; one T3 shows that OpenCode no longer lists was // answered elsewhere or dropped with its execution. A session that @@ -3413,9 +3492,65 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr /** Feeds a held execution's events to the turn now running on its session. */ const replay = Effect.fnUntraced(function* (wake: Wake) { wake.dropped = true; - for (const event of wake.events) yield* route(event, sessionOfEvent(event)); + for (const event of wake.events) yield* route(event, sessionOfEvent(event), true); }); + /** + * Drops held executions no turn will take. The subagents one started are + * named only by its replay, so they stop routing to the thread at once and + * become strays for {@link stopStrays}. A subagent that registered its own + * call outside the held execution is the thread's, and is left alone. + */ + const discardWakes = (caller: ThreadState, wakes: ReadonlyArray) => { + const retired = new Set(); + const retire = (childId: string) => { + const info = announced.get(childId); + if (info === undefined) return; + announced.delete(childId); + childOwners.delete(childId); + strays.set(childId, caller); + retired.add(childId); + // OpenCode reports its end to the session that called it; that + // report answers no turn, so the execution it wakes is stopped. A + // deeper one reports to a stray, which T3 does not track. + threads.get(info.parentID ?? "")?.stoppedChildren.add(childId); + }; + for (const wake of wakes) { + wake.dropped = true; + for (const childId of wake.children) retire(childId); + } + // Their own subagents, at any depth, were announced live but no call + // of the thread's names them either. + for (let grew = retired.size > 0; grew;) { + grew = false; + for (const [childId, info] of announced) { + if (info.parentID === undefined || !retired.has(info.parentID)) continue; + retire(childId); + grew = true; + } + } + }; + + /** + * Stops a thread's strays one by one, as a Stop stops background + * subagents, and never under `lock`: one that does not answer would + * otherwise hold up every event. A stray is forgotten once stopped or + * gone; one still running waits for the next Stop. + */ + const stopStrays = (thread: ThreadState) => + Effect.forEach( + [...strays].flatMap(([childId, caller]) => (rootOf(caller) === thread ? [childId] : [])), + (childId) => + client.session.interrupt({ sessionID: Session.ID.make(childId) }).pipe( + Effect.asVoid, + Effect.catchTags({ SessionNotFoundError: () => Effect.void }), + Effect.timeout(INTERRUPT_TIMEOUT), + Effect.tap(() => Effect.sync(() => strays.delete(childId))), + Effect.ignore({ log: true }), + ), + { discard: true }, + ); + /** * A user turn that starts while OpenCode runs an execution on its own: the * prompt joins that execution, so the turn takes it, and its continuation @@ -3447,9 +3582,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr (held) => held.messageId === turnInput.message.messageId, ); if (index < 0) return yield* finishTurn(state, { status: "completed" }); - // Trade-off: a continuation the user reordered ahead of an earlier one - // drops that one's reply too, so cancelled ones never pin the session. - for (const skipped of state.wakes.splice(0, index)) skipped.dropped = true; + // Queued continuations can't be reordered (they carry a notification), + // so an earlier wake here lost its continuation to a cancel or a failed + // dispatch. Dropping it keeps that wake from pinning the session. + discardWakes(state, state.wakes.splice(0, index)); + yield* Effect.forkIn(stopStrays(state), sessionScope); const wake = state.wakes.shift()!; const turn = state.active; if (turn !== undefined && wake.after !== undefined) turn.before = wake.after; @@ -3457,6 +3594,9 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr if (turn !== undefined && wake.first !== undefined) { turn.providerTurn = { ...turn.providerTurn, nativeTurnRef: ref(wake.first, "weak") }; } + // The execution still runs, so the run a mark waits on is this one or + // was over before it started: its end, still to come, is this turn's. + if (wake.running) state.unsettled = false; yield* replay(wake); }), ); @@ -3537,15 +3677,17 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr } // A held wake no turn will take: its execution is stopped, not replayed. const running = state.wakes.some((wake) => wake.running); - for (const wake of state.wakes.splice(0)) wake.dropped = true; + discardWakes(state, state.wakes.splice(0)); if (running) { - state.unsettled = true; + markUnsettled(state); yield* client.session .interrupt({ sessionID: Session.ID.make(state.sessionId) }) .pipe(Effect.timeout(INTERRUPT_TIMEOUT), Effect.ignore({ log: true })); } }), ); + // Subagents of held executions this or an earlier Stop dropped. + yield* stopStrays(state); }); /** @@ -3915,7 +4057,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ? Effect.gen(function* () { // Without a clear rejection the server may have taken the // prompt, so the next turn checks before it prompts again. - if (!CLEAR_PROMPT_REJECTIONS.has(cause._tag)) state.unsettled = true; + if (!CLEAR_PROMPT_REJECTIONS.has(cause._tag)) markUnsettled(state); yield* finishTurn(state, { status: "failed", failure: makeProviderFailure({ cause, class: "provider_error" }), @@ -4023,13 +4165,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ), ); if (reply._tag === "None") { - state.unsettled = true; + markUnsettled(state); return yield* lock.withPermit(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. if (!reply.value.interrupted && state.active === turn && turn.heldEnd !== undefined) { - state.unsettled = true; + markUnsettled(state); yield* lock.withPermit(finishTurn(state, { status: "interrupted" })); return yield* cancelStrandedSteers(state); }