From 6c580a1bfbd7b437d2451175d902dc3c9258daa4 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 8 Nov 2026 05:09:04 +0000 Subject: [PATCH 01/14] test(server): reproduce #32829, OpenCode 2 background shells not tracked These tests reproduce #32829: a background shell (`shell` with `background: true`) still running after the turn is missing from the provider thread's pendingBackgroundTasks, and a user Stop never removes it through OpenCode's shell API. The testkit maps DELETE /api/shell/:id to `shell.remove` so the replay can expect it. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/OpenCode2AdapterV2.test.ts | 105 ++++++++++++++++++ .../Adapters/OpenCode2AdapterV2.testkit.ts | 4 + 2 files changed, 109 insertions(+) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index e4e0851e82c29..5eb0b7056d76e 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -903,6 +903,111 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + const SHELL_COMMAND = "sleep 300; echo done"; + const backgroundShell = (shellID: string): ReadonlyArray => { + const tool = { sessionID: SESSION, assistantMessageID: "msg_assistant", id: `call-${shellID}` }; + return [ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + event("session.execution.started", { sessionID: SESSION }), + event("session.tool.input.started", { ...tool, name: "shell" }), + event("session.tool.called", { + ...tool, + name: "shell", + input: { command: SHELL_COMMAND, background: true }, + executed: false, + }), + { + type: "emit_inbound", + frame: { + type: "sdk.event", + event: { + id: `evt_toolsuccess_${tool.id}`, + created: 1, + type: "session.tool.success", + data: { + ...tool, + content: [ + { type: "text", text: `Command moved to the background (shell ID: ${shellID}).` }, + ], + metadata: { status: "running", truncated: false, shellID }, + executed: false, + }, + durable: { aggregateID: SESSION, seq: 1, version: 2 }, + }, + }, + }, + ]; + }; + + it.effect("lists a background shell still running after the turn as pending work", () => + Effect.gen(function* () { + const { runtime, thread } = yield* resumed([ + ...backgroundShell("sh_1"), + event("session.execution.succeeded", { sessionID: SESSION }), + ]); + const events = yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + Effect.forkScoped, + ); + yield* runtime.startTurn(turnInput(thread)); + const roster = (yield* Fiber.join(events)) + .flatMap((event) => + event.type === "provider_thread.updated" && event.providerThread.id === thread.id + ? [event.providerThread.pendingBackgroundTasks] + : [], + ) + .at(-1); + assert.deepEqual(roster ?? [], [ + { taskId: "sh_1", kind: "command", description: SHELL_COMMAND }, + ]); + assert.isTrue(yield* runtime.hasPendingBackgroundWork!); + }).pipe(Effect.scoped), + ); + + it.effect("removes a running background shell on a user Stop", () => + Effect.gen(function* () { + const { runtime, thread } = yield* resumed([ + ...backgroundShell("sh_1"), + out("shell.remove", { id: "sh_1" }), + reply("shell.remove", null), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: SESSION }), + ]); + const launched = yield* Deferred.make(); + const terminal = yield* runtime.events.pipe( + Stream.tap((event) => + event.type === "turn_item.updated" && + event.turnItem.nativeItemRef?.nativeId?.endsWith("call-sh_1") === true && + event.turnItem.type === "command_execution" && + event.turnItem.output !== undefined + ? Deferred.succeed(launched, undefined) + : Effect.void, + ), + Stream.filter( + ( + event, + ): event is Extract => + event.type === "turn.terminal", + ), + Stream.runHead, + Effect.map(Option.getOrUndefined), + Effect.forkScoped, + ); + yield* runtime.startTurn(turnInput(thread)); + yield* Deferred.await(launched); + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + assert.equal((yield* Fiber.join(terminal))?.status, "interrupted"); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + }).pipe(Effect.scoped), + ); + it.effect("gives a background subagent the rules of a mode changed while it runs", () => 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 46421aa5c5c86..9ab68864edad3 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts @@ -78,6 +78,10 @@ const operationOf = ( }; } if (method === "GET" && path === "/api/skill") return { type: "skill.list", input: query }; + const shell = /^\/api\/shell\/([^/]+)$/.exec(path); + if (method === "DELETE" && shell !== null) { + return { type: "shell.remove", input: { id: decodeURIComponent(shell[1] ?? ""), ...query } }; + } if (session !== null) { const [, sessionID, rest = ""] = session; // The client sends absent optional fields as `null`; recordings omit them. From 7db4f03f2a355cea484047a0fe8ca2fb392f20a9 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 8 Nov 2026 05:18:57 +0000 Subject: [PATCH 02/14] fix(opencode): count running background shells as pending work OpenCode 2 background shells (`shell` with `background: true`) were not part of the thread's pending background work, so a thread with a running shell looked idle and a user Stop left the shell running. The adapter now tracks each session's running shells from the launch result (and from backfilled tool parts after a stream gap), drops them when OpenCode announces their end, and verifies the tracked ones with `shell.get` when it reconnects. The roster lists them as `command` tasks after the subagents, with `parentTaskId` naming the subagent call when a child session started the shell. A user Stop removes every running shell of the thread and its subagents before interrupting; steering leaves them running. Fixes pingdotgg/t3code#32829 Co-Authored-By: Claude Opus 5.5 --- .../Adapters/OpenCode2AdapterV2.test.ts | 59 ++++++++- .../src/server/v2/adapter.ts | 117 ++++++++++++++++-- 2 files changed, 167 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 5eb0b7056d76e..f57ff23d9b4a3 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -970,7 +970,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { Effect.gen(function* () { const { runtime, thread } = yield* resumed([ ...backgroundShell("sh_1"), - out("shell.remove", { id: "sh_1" }), + out("shell.remove", { id: "sh_1", "location[directory]": WORK }), reply("shell.remove", null), out("session.interrupt", { sessionID: SESSION }), reply("session.interrupt", { interrupted: true }), @@ -1008,6 +1008,63 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + it.effect("lists a subagent's background shell under the subagent that started it", () => + Effect.gen(function* () { + const tool = { sessionID: CHILD, assistantMessageID: "msg_child", id: "call-child-shell" }; + const { runtime, thread } = yield* resumed([ + ...backgroundLaunch(CHILD), + event("session.execution.started", { sessionID: CHILD }), + event("session.tool.input.started", { ...tool, name: "shell" }), + event("session.tool.called", { + ...tool, + name: "shell", + input: { command: SHELL_COMMAND, background: true }, + executed: false, + }), + { + type: "emit_inbound", + frame: { + type: "sdk.event", + event: { + id: "evt_toolsuccess_child_shell", + created: 1, + type: "session.tool.success", + data: { + ...tool, + content: [{ type: "text", text: "Command moved to the background." }], + metadata: { status: "running", truncated: false, shellID: "sh_child" }, + executed: false, + }, + durable: { aggregateID: CHILD, seq: 1, version: 2 }, + }, + }, + }, + event("session.execution.succeeded", { sessionID: SESSION }), + ]); + const events = yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + Effect.forkScoped, + ); + yield* runtime.startTurn(withLineage(thread)); + const roster = (yield* Fiber.join(events)) + .flatMap((event) => + event.type === "provider_thread.updated" && event.providerThread.id === thread.id + ? [event.providerThread.pendingBackgroundTasks] + : [], + ) + .at(-1); + const [subagent, shell] = roster ?? []; + assert.equal(subagent?.kind, "subagent"); + assert.deepEqual(shell, { + taskId: "sh_child", + kind: "command", + description: SHELL_COMMAND, + parentTaskId: subagent?.taskId, + }); + }).pipe(Effect.scoped), + ); + it.effect("gives a background subagent the rules of a mode changed while it runs", () => Effect.gen(function* () { const { runtime, thread } = yield* resumed([ diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index 3cbd4d893b6f1..ce36de0b8a76b 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -14,12 +14,14 @@ import { type OpenCodeEvent, } from "@opencode/client/effect"; import { Mcp } from "@opencode/schema/mcp"; +import { Shell } from "@opencode/schema/shell"; import { isOrchestrationV2WorkActive, type ChatAttachment, type OrchestrationV2AppThread, type OrchestrationV2ConversationMessage, type OrchestrationV2ExecutionNode, + type OrchestrationV2PendingBackgroundTask, type OrchestrationV2ProviderCapabilities, type OrchestrationV2ProviderFailure, type OrchestrationV2ProviderRetry, @@ -312,6 +314,13 @@ interface ThreadState { readonly text: string; } >; + /** Running background shells by id; a shell stays at its launch directory when the session moves. */ + readonly shells: Map< + string, + { readonly directory: string; readonly command: string | undefined } + >; + /** Shells whose end arrived before their launch result; OpenCode does not wait for the launch to return. */ + readonly endedShells: Set; readonly stoppedChildren: Set; readonly strandedSteers: Set; mcp: @@ -769,6 +778,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr subagent, wakes: [], reports: new Map(), + shells: new Map(), + endedShells: new Set(), stoppedChildren: new Set(), strandedSteers: new Set(), mcp: undefined, @@ -829,6 +840,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const owesWork = (state: ThreadState) => state.wakes.length > 0 || state.reports.size > 0 || + state.shells.size > 0 || (state.subagent !== undefined && busy.has(state.sessionId)) || [...state.calls.values()].some((call) => call.background); @@ -1257,7 +1269,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr }); const syncRoster = Effect.fnUntraced(function* (state: ThreadState) { - const roster = runningCalls(state) + const subagents = runningCalls(state) .filter((call) => call.background) .map((call) => ({ taskId: call.nativeId, @@ -1267,17 +1279,35 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ? {} : { childThreadId: call.child.subagent.appThread.id }), })); + // Restart recovery reads a subagent before the shells naming it as their parent. + const shells = sessionsOf(state).flatMap((session) => + [...session.shells].map(([taskId, shell]) => ({ + taskId, + kind: "command" as const, + ...(shell.command === undefined ? {} : { description: shell.command }), + ...(session === state || session.subagent === undefined + ? {} + : { parentTaskId: session.subagent.call.nativeId }), + })), + ); + const roster: ReadonlyArray = [...subagents, ...shells]; + const childThreadOf = (task: OrchestrationV2PendingBackgroundTask) => + task.kind === "subagent" ? task.childThreadId : undefined; const current = state.providerThread.pendingBackgroundTasks ?? []; if ( state.publishedBackgroundWork === hasBackground(state) && current.length === roster.length && - current.every( - (task, index) => - task.taskId === roster[index]?.taskId && - task.description === roster[index]?.description && - (task.kind === "subagent" ? task.childThreadId : undefined) === - roster[index]?.childThreadId, - ) + current.every((task, index) => { + const next = roster[index]; + return ( + next !== undefined && + task.kind === next.kind && + task.taskId === next.taskId && + task.description === next.description && + task.parentTaskId === next.parentTaskId && + childThreadOf(task) === childThreadOf(next) + ); + }) ) { return; } @@ -2101,12 +2131,18 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr return yield* settleCall(call, "completed", subagentOutput(textOf(event.data.content))); } const output = textOf(event.data.content); + const launched = trackShellLaunch( + state, + event.data.metadata, + turn.tools.get(event.data.id)?.input, + ); yield* emitTool(state, turn, event.data.id, "completed", { output, files: event.data.content.flatMap((part) => (part.type === "file" ? [part] : [])), metadata: event.data.metadata, }); turn.tools.delete(event.data.id); + if (launched) yield* syncRosterAncestors(state); return; } case "session.tool.failed": { @@ -2296,6 +2332,30 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ? finishTurn(state, turn.heldEnd) : Effect.void; + const trackShellLaunch = ( + state: ThreadState, + metadata: Readonly> | undefined, + input: Readonly> | undefined, + ) => { + const shellId = stringField(metadata, "shellID"); + if (metadata?.["status"] !== "running" || shellId === undefined) return false; + if (state.endedShells.delete(shellId) || state.shells.has(shellId)) return false; + state.shells.set(shellId, { + directory: state.directory, + command: stringField(input, "command")?.trim(), + }); + return true; + }; + + const trackShellEnd = ( + state: ThreadState, + metadata: Readonly> | undefined, + ) => { + const shellId = stringField(metadata, "shellID"); + if (stringField(metadata, "source") !== "shell" || shellId === undefined) return; + if (!state.shells.delete(shellId)) state.endedShells.add(shellId); + }; + const handleEvent = Effect.fnUntraced(function* (event: OpenCode2Client.OpenCode2StreamEvent) { const sessionId = sessionOfEvent(event); const cleared = sessionId === undefined ? undefined : clearing.get(sessionId); @@ -2433,6 +2493,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr state.subagent.queued ??= event.data.inboxID; } if (event.type === "session.inbox.enqueued" && event.data.item.type === "synthetic") { + trackShellEnd(state, event.data.item.payload.metadata); return yield* onReport(state, event.data.inboxID, event.data.item.payload); } if (event.type === "session.inbox.cancelled" || event.type === "session.inbox.delivered") { @@ -2489,6 +2550,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr for (const call of runningCalls(state)) yield* settleCall(call, "failed"); for (const wake of state.wakes.splice(0)) wake.dropped = true; state.reports.clear(); + state.shells.clear(); + state.endedShells.clear(); } yield* setSessionStatus("error", message); yield* Queue.end(events); @@ -2560,6 +2623,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr tool.input = part.state.input as Record; } if (part.state.status === "completed") { + // A launch lost with the stream still runs; reconcile verifies tracked shells after this. + trackShellLaunch(state, part.state.metadata, tool.input); yield* emitTool(state, turn, part.id, "completed", { output: textOf(part.state.content), files: part.state.content.flatMap((content) => @@ -2643,6 +2708,23 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr } for (const session of sessions.toReversed()) yield* syncRoster(session); } + // A shell's end may have been lost with the stream, and backfill may re-add an ended one. + for (const state of threads.values()) { + if (state.shells.size === 0) continue; + for (const [id, shell] of state.shells) { + const running = yield* client.shell + .get({ id: Shell.ID.make(id), location: { directory: shell.directory } }) + .pipe( + Effect.map(({ data }) => data.status === "running"), + Effect.catchTags({ + ShellNotFoundError: () => Effect.succeed(false), + LocationNotFoundError: () => Effect.succeed(false), + }), + ); + if (!running) state.shells.delete(id); + } + yield* syncRosterAncestors(state); + } // 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); @@ -3303,6 +3385,21 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const children = new Map( calls.flatMap((call) => childrenOf(call).map((childId) => [childId, call.state] as const)), ); + const shells = callers.flatMap((caller) => + [...caller.shells].map(([id, shell]) => ({ caller, id, directory: shell.directory })), + ); + const removed = yield* Effect.filter(shells, ({ id, directory }) => + client.shell.remove({ id: Shell.ID.make(id), location: { directory } }).pipe( + // OpenCode declares no not-found error for remove; only the directory can be gone. + Effect.catchTags({ LocationNotFoundError: () => Effect.void }), + Effect.timeout(INTERRUPT_TIMEOUT), + Effect.tapCause((cause) => + Effect.logWarning("Could not stop an OpenCode background shell.", cause), + ), + Effect.exit, + Effect.map(Exit.isSuccess), + ), + ); const unreached = new Set(); for (const [childId, caller] of children) { caller.stoppedChildren.add(childId); @@ -3339,6 +3436,10 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr caller.stoppedChildren.add(report.childId); } } + for (const { caller, id } of removed) caller.shells.delete(id); + for (const caller of new Set(removed.map(({ caller }) => caller))) { + yield* syncRosterAncestors(caller); + } const running = state.wakes.some((wake) => wake.running); discardWakes(state, state.wakes.splice(0)); yield* syncRosterAncestors(state); From 760f1bc11082cfd653baa999e83f7dd2fd1c30e9 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 8 Nov 2026 05:45:30 +0000 Subject: [PATCH 03/14] fix(opencode): Stop starts no work from a removed shell's report Removing a background shell makes OpenCode report it to the shell's session and wake that session. Shells a user Stop removes are now marked stopped before the remove, so their reports are handled like a stopped subagent's: the wake is interrupted with no continuation and no child turn. A stream that cannot reconnect now publishes the cleared roster, and a shell lists its subagent as parent only while that subagent call still runs. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/OpenCode2AdapterV2.test.ts | 190 ++++++++++++------ .../src/server/v2/adapter.ts | 74 ++++--- 2 files changed, 174 insertions(+), 90 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index f57ff23d9b4a3..686da275da6e4 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -904,12 +904,12 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { ); const SHELL_COMMAND = "sleep 300; echo done"; - const backgroundShell = (shellID: string): ReadonlyArray => { - const tool = { sessionID: SESSION, assistantMessageID: "msg_assistant", id: `call-${shellID}` }; + const shellLaunched = ( + sessionID: string, + shellID: string, + ): ReadonlyArray => { + const tool = { sessionID, assistantMessageID: "msg_assistant", id: `call-${shellID}` }; return [ - out("session.prompt", { sessionID: SESSION, text: "" }), - promptAccepted, - event("session.execution.started", { sessionID: SESSION }), event("session.tool.input.started", { ...tool, name: "shell" }), event("session.tool.called", { ...tool, @@ -933,12 +933,43 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { metadata: { status: "running", truncated: false, shellID }, executed: false, }, - durable: { aggregateID: SESSION, seq: 1, version: 2 }, + durable: { aggregateID: sessionID, seq: 1, version: 2 }, }, }, }, ]; }; + const backgroundShell = (shellID: string): ReadonlyArray => [ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + event("session.execution.started", { sessionID: SESSION }), + ...shellLaunched(SESSION, shellID), + ]; + // OpenCode reports a removed shell to its session and wakes it before the remove returns. + const shellRemoved = (sessionID: string, shellID: string): ReadonlyArray => [ + out("shell.remove", { id: shellID, "location[directory]": WORK }), + event("session.inbox.enqueued", { + inboxID: `msg_removed_${shellID}`, + sessionID, + item: { + type: "synthetic", + payload: { + text: `Shell.NotFoundError`, + metadata: { source: "shell", shellID, state: "error" }, + }, + delivery: "steer", + }, + }), + event("session.execution.started", { sessionID }), + reply("shell.remove", null), + out("session.interrupt", { sessionID }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID }), + ]; + const drained = event("session.model.selected", { + sessionID: SESSION, + model: { providerID: "opencode", id: "big-pickle", variant: "drained" }, + }); it.effect("lists a background shell still running after the turn as pending work", () => Effect.gen(function* () { @@ -966,79 +997,111 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); - it.effect("removes a running background shell on a user Stop", () => + it.effect("clears a background shell from the roster when the stream cannot reconnect", () => Effect.gen(function* () { const { runtime, thread } = yield* resumed([ ...backgroundShell("sh_1"), - out("shell.remove", { id: "sh_1", "location[directory]": WORK }), - reply("shell.remove", null), - out("session.interrupt", { sessionID: SESSION }), - reply("session.interrupt", { interrupted: true }), - event("session.execution.interrupted", { sessionID: SESSION }), + event("session.execution.succeeded", { sessionID: SESSION }), + { type: "runtime_exit", status: "success" } as const, ]); - const launched = yield* Deferred.make(); - const terminal = yield* runtime.events.pipe( - Stream.tap((event) => - event.type === "turn_item.updated" && - event.turnItem.nativeItemRef?.nativeId?.endsWith("call-sh_1") === true && - event.turnItem.type === "command_execution" && - event.turnItem.output !== undefined - ? Deferred.succeed(launched, undefined) - : Effect.void, - ), - Stream.filter( - ( - event, - ): event is Extract => - event.type === "turn.terminal", - ), - Stream.runHead, - Effect.map(Option.getOrUndefined), - Effect.forkScoped, + const events = yield* runtime.events.pipe(Stream.runCollect, Effect.forkScoped); + const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped); + yield* runtime.startTurn(turnInput(thread)); + assert.equal((yield* Fiber.join(terminal))?.status, "completed"); + yield* TestClock.adjust("1 minute"); + const last = (yield* Fiber.join(events)) + .filter( + (event) => + event.type === "provider_thread.updated" && event.providerThread.id === thread.id, + ) + .at(-1); + assert.deepEqual( + last?.type === "provider_thread.updated" + ? [last.providerThread.pendingBackgroundTasks, last.pendingBackgroundWork] + : undefined, + [[], false], ); + }).pipe(Effect.scoped), + ); + + it.effect("removes a running background shell on a user Stop", () => + Effect.gen(function* () { + const offered = yield* Deferred.make(); + const { runtime, thread } = yield* resumed([ + ...backgroundShell("sh_1"), + event("session.execution.succeeded", { sessionID: SESSION }), + ...shellRemoved(SESSION, "sh_1"), + drained, + ]).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid), + take: Effect.never, + }), + ); + const watch = yield* watchCalls(runtime, thread); + const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped); yield* runtime.startTurn(turnInput(thread)); - yield* Deferred.await(launched); + assert.equal((yield* Fiber.join(terminal))?.status, "completed"); yield* runtime.interruptTurn({ providerThread: thread, providerTurnId: yield* providerTurnId, requestRuntimeRestart: true, }); - assert.equal((yield* Fiber.join(terminal))?.status, "interrupted"); + yield* Effect.raceFirst( + watch.until(() => watch.state.variant === "drained"), + Deferred.await(offered), + ); + assert.isFalse(yield* Deferred.isDone(offered)); + assert.deepEqual(watch.state.roster, []); + assert.isFalse(watch.state.pending); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + }).pipe(Effect.scoped), + ); + + it.effect("lists a finished subagent's background shell as a plain command a Stop removes", () => + Effect.gen(function* () { + const { runtime, thread } = yield* resumed([ + ...backgroundLaunch(CHILD), + returnedRunning("call-background"), + event("session.execution.started", { sessionID: CHILD }), + ...shellLaunched(CHILD, "sh_child"), + event("session.execution.succeeded", { sessionID: CHILD }), + childReport("msg_report", "CHILD_OK"), + event("session.inbox.delivered", { sessionID: SESSION, inboxID: "msg_report" }), + event("session.execution.succeeded", { sessionID: SESSION }), + ...shellRemoved(CHILD, "sh_child"), + drained, + ]); + const watch = yield* watchCalls(runtime, thread); + const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped); + yield* runtime.startTurn(withLineage(thread)); + assert.equal((yield* Fiber.join(terminal))?.status, "completed"); + yield* watch.until(() => watch.calls.get("Sleep")?.status === "completed"); + assert.deepEqual(watch.state.roster, [ + { taskId: "sh_child", kind: "command", description: SHELL_COMMAND }, + ]); + const childTurns = watch.childTurns.size; + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + yield* watch.until( + () => watch.state.variant === "drained" || watch.childTurns.size > childTurns, + ); + assert.equal(watch.childTurns.size, childTurns); + assert.deepEqual(watch.state.roster, []); + assert.isFalse(watch.state.pending); assert.isFalse(yield* runtime.hasPendingBackgroundWork!); }).pipe(Effect.scoped), ); it.effect("lists a subagent's background shell under the subagent that started it", () => Effect.gen(function* () { - const tool = { sessionID: CHILD, assistantMessageID: "msg_child", id: "call-child-shell" }; const { runtime, thread } = yield* resumed([ ...backgroundLaunch(CHILD), event("session.execution.started", { sessionID: CHILD }), - event("session.tool.input.started", { ...tool, name: "shell" }), - event("session.tool.called", { - ...tool, - name: "shell", - input: { command: SHELL_COMMAND, background: true }, - executed: false, - }), - { - type: "emit_inbound", - frame: { - type: "sdk.event", - event: { - id: "evt_toolsuccess_child_shell", - created: 1, - type: "session.tool.success", - data: { - ...tool, - content: [{ type: "text", text: "Command moved to the background." }], - metadata: { status: "running", truncated: false, shellID: "sh_child" }, - executed: false, - }, - durable: { aggregateID: CHILD, seq: 1, version: 2 }, - }, - }, - }, + ...shellLaunched(CHILD, "sh_child"), event("session.execution.succeeded", { sessionID: SESSION }), ]); const events = yield* runtime.events.pipe( @@ -1258,7 +1321,11 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { string, Record >(); - const state: { roster: ReadonlyArray | undefined } = { roster: undefined }; + const state: { + roster: ReadonlyArray | undefined; + pending: boolean | undefined; + variant: unknown; + } = { roster: undefined, pending: undefined, variant: undefined }; let wake = yield* Deferred.make(); yield* runtime.events.pipe( Stream.tap((event) => @@ -1271,6 +1338,9 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { } if (event.type === "provider_thread.updated" && event.providerThread.id === thread.id) { state.roster = event.providerThread.pendingBackgroundTasks; + state.pending = event.pendingBackgroundWork; + state.variant = + event.providerThread.nativeMetadata?.modelSelection?.options?.[0]?.value; } const id = event.type === "provider_turn.updated" diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index ce36de0b8a76b..4fb830135ac7b 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -309,7 +309,8 @@ interface ThreadState { string, { readonly inboxId: string; - readonly childId: string; + /** The child session or background shell that reported. */ + readonly sourceId: string; readonly report: BackgroundWorkReport; readonly text: string; } @@ -321,7 +322,8 @@ interface ThreadState { >; /** Shells whose end arrived before their launch result; OpenCode does not wait for the launch to return. */ readonly endedShells: Set; - readonly stoppedChildren: Set; + /** Child sessions and shells the user stopped; a wake their reports alone cause runs nothing. */ + readonly stopped: Set; readonly strandedSteers: Set; mcp: | { readonly name: string; readonly directory: string; readonly credential: string } @@ -780,7 +782,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr reports: new Map(), shells: new Map(), endedShells: new Set(), - stoppedChildren: new Set(), + stopped: new Set(), strandedSteers: new Set(), mcp: undefined, instructions: undefined, @@ -1269,25 +1271,24 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr }); const syncRoster = Effect.fnUntraced(function* (state: ThreadState) { - const subagents = runningCalls(state) - .filter((call) => call.background) - .map((call) => ({ - taskId: call.nativeId, - kind: "subagent" as const, - ...(call.title?.trim() ? { description: call.title.trim() } : {}), - ...(call.child?.subagent === undefined - ? {} - : { childThreadId: call.child.subagent.appThread.id }), - })); + const background = runningCalls(state).filter((call) => call.background); + const subagents = background.map((call) => ({ + taskId: call.nativeId, + kind: "subagent" as const, + ...(call.title?.trim() ? { description: call.title.trim() } : {}), + ...(call.child?.subagent === undefined + ? {} + : { childThreadId: call.child.subagent.appThread.id }), + })); // Restart recovery reads a subagent before the shells naming it as their parent. const shells = sessionsOf(state).flatMap((session) => [...session.shells].map(([taskId, shell]) => ({ taskId, kind: "command" as const, ...(shell.command === undefined ? {} : { description: shell.command }), - ...(session === state || session.subagent === undefined - ? {} - : { parentTaskId: session.subagent.call.nativeId }), + ...(session.subagent !== undefined && background.includes(session.subagent.call) + ? { parentTaskId: session.subagent.call.nativeId } + : {}), })), ); const roster: ReadonlyArray = [...subagents, ...shells]; @@ -2228,9 +2229,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const delivered = [...state.reports.values()]; state.reports.clear(); const stopped = - delivered.length > 0 && - delivered.every((entry) => state.stoppedChildren.has(entry.childId)); - for (const entry of delivered) state.stoppedChildren.delete(entry.childId); + delivered.length > 0 && delivered.every((entry) => state.stopped.has(entry.sourceId)); + for (const entry of delivered) state.stopped.delete(entry.sourceId); if (stopped) { markUnsettled(state); yield* client.session @@ -2285,7 +2285,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const outcome = reportOutcome(stringField(payload.metadata, "state")); state.reports.set(inboxId, { inboxId, - childId, + sourceId: childId, text: payload.text, report: { kind: "subagent", @@ -2294,7 +2294,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr outcome, }, }); - const status = state.stoppedChildren.has(childId) + const status = state.stopped.has(childId) ? "interrupted" : outcome === "failed" ? "failed" @@ -2349,11 +2349,24 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const trackShellEnd = ( state: ThreadState, - metadata: Readonly> | undefined, + inboxId: string, + payload: { + readonly text: string; + readonly metadata?: Readonly> | undefined; + }, ) => { - const shellId = stringField(metadata, "shellID"); - if (stringField(metadata, "source") !== "shell" || shellId === undefined) return; + const shellId = stringField(payload.metadata, "shellID"); + if (stringField(payload.metadata, "source") !== "shell" || shellId === undefined) return; + const shell = state.shells.get(shellId); if (!state.shells.delete(shellId)) state.endedShells.add(shellId); + // Removing a shell reports it to its session, which wakes that session. + if (!state.stopped.has(shellId)) return; + state.reports.set(inboxId, { + inboxId, + sourceId: shellId, + text: payload.text, + report: { kind: "command", label: shell?.command, outcome: "cancelled" }, + }); }; const handleEvent = Effect.fnUntraced(function* (event: OpenCode2Client.OpenCode2StreamEvent) { @@ -2493,7 +2506,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr state.subagent.queued ??= event.data.inboxID; } if (event.type === "session.inbox.enqueued" && event.data.item.type === "synthetic") { - trackShellEnd(state, event.data.item.payload.metadata); + trackShellEnd(state, event.data.inboxID, event.data.item.payload); return yield* onReport(state, event.data.inboxID, event.data.item.payload); } if (event.type === "session.inbox.cancelled" || event.type === "session.inbox.delivered") { @@ -2553,6 +2566,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr state.shells.clear(); state.endedShells.clear(); } + for (const state of threads.values()) yield* syncRoster(state); yield* setSessionStatus("error", message); yield* Queue.end(events); }); @@ -3295,7 +3309,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr childOwners.delete(childId); strays.set(childId, caller); retired.add(childId); - threads.get(info.parentID ?? "")?.stoppedChildren.add(childId); + threads.get(info.parentID ?? "")?.stopped.add(childId); }; for (const wake of wakes) { wake.dropped = true; @@ -3388,6 +3402,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const shells = callers.flatMap((caller) => [...caller.shells].map(([id, shell]) => ({ caller, id, directory: shell.directory })), ); + // OpenCode reports and wakes the shell's session before the remove returns. + for (const { caller, id } of shells) caller.stopped.add(id); const removed = yield* Effect.filter(shells, ({ id, directory }) => client.shell.remove({ id: Shell.ID.make(id), location: { directory } }).pipe( // OpenCode declares no not-found error for remove; only the directory can be gone. @@ -3402,7 +3418,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ); const unreached = new Set(); for (const [childId, caller] of children) { - caller.stoppedChildren.add(childId); + caller.stopped.add(childId); const reached = yield* client.session .interrupt({ sessionID: Session.ID.make(childId) }) .pipe( @@ -3432,9 +3448,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr yield* settleCall(call, "interrupted"); } for (const caller of callers) { - for (const report of caller.reports.values()) { - caller.stoppedChildren.add(report.childId); - } + for (const report of caller.reports.values()) caller.stopped.add(report.sourceId); } for (const { caller, id } of removed) caller.shells.delete(id); for (const caller of new Set(removed.map(({ caller }) => caller))) { From d5f8c8b52b6e3d1e3876ed4277c9e22dbf92e7bb Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 8 Nov 2026 06:21:14 +0000 Subject: [PATCH 04/14] fix(opencode): count background shells as their events arrive Stop could leave pending work stuck, miss a shell launched by a wake still waiting for its continuation, and swallow a natural shell result that shared a wake with a stopped one. Shell launches, ends and reports are now tracked when OpenCode sends them, including events held for a pending continuation, so Stop sees those shells and a replay counts nothing twice. Natural shell reports are recorded with their outcome, so a wake they share with a stopped source continues. A report from a source the user stopped no longer counts as owed work but still suppresses the wake it causes on its own. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/OpenCode2AdapterV2.test.ts | 168 ++++++++++++++++-- .../src/server/v2/adapter.ts | 66 ++++--- 2 files changed, 199 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 686da275da6e4..a0ff360be47ce 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -945,21 +945,25 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { event("session.execution.started", { sessionID: SESSION }), ...shellLaunched(SESSION, shellID), ]; - // OpenCode reports a removed shell to its session and wakes it before the remove returns. - const shellRemoved = (sessionID: string, shellID: string): ReadonlyArray => [ - out("shell.remove", { id: shellID, "location[directory]": WORK }), + const shellReport = (sessionID: string, shellID: string, state: string, text: string) => event("session.inbox.enqueued", { - inboxID: `msg_removed_${shellID}`, + inboxID: `msg_${state}_${shellID}`, sessionID, item: { type: "synthetic", payload: { - text: `Shell.NotFoundError`, - metadata: { source: "shell", shellID, state: "error" }, + text: `${text}`, + metadata: { source: "shell", shellID, state }, }, delivery: "steer", }, - }), + }); + const removeShell = (shellID: string) => + out("shell.remove", { id: shellID, "location[directory]": WORK }); + // OpenCode reports a removed shell to its session and wakes it before the remove returns. + const shellRemoved = (sessionID: string, shellID: string): ReadonlyArray => [ + removeShell(shellID), + shellReport(sessionID, shellID, "error", "Shell.NotFoundError"), event("session.execution.started", { sessionID }), reply("shell.remove", null), out("session.interrupt", { sessionID }), @@ -1039,9 +1043,9 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }), ); const watch = yield* watchCalls(runtime, thread); - const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped); yield* runtime.startTurn(turnInput(thread)); - assert.equal((yield* Fiber.join(terminal))?.status, "completed"); + yield* watch.until(() => watch.terminals.length > 0); + assert.deepEqual(watch.terminals, ["completed"]); yield* runtime.interruptTurn({ providerThread: thread, providerTurnId: yield* providerTurnId, @@ -1058,6 +1062,144 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + it.effect("leaves no work pending when Stop removes a shell during a turn", () => + Effect.gen(function* () { + const offered = yield* Deferred.make(); + const gate = makeProviderReplayGate(["wake"]); + const { runtime, thread } = yield* resumed( + [ + ...backgroundShell("sh_1"), + // The session is executing, so the report waits in its inbox and Stop ends the run first. + removeShell("sh_1"), + shellReport(SESSION, "sh_1", "error", "Shell.NotFoundError"), + reply("shell.remove", null), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: SESSION }), + event("session.execution.started", { sessionID: SESSION }, "wake"), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: SESSION }), + drained, + ], + { replayGate: gate }, + ).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid), + take: Effect.never, + }), + ); + const watch = yield* watchCalls(runtime, thread); + yield* runtime.startTurn(turnInput(thread)); + yield* watch.until(() => watch.state.roster?.length === 1); + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + yield* watch.until(() => watch.terminals.length > 0); + assert.isTrue(yield* Effect.promise(() => gate.waitForReached("wake"))); + assert.deepEqual(watch.terminals, ["interrupted"]); + assert.deepEqual(watch.state.roster, []); + assert.isFalse(watch.state.pending); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + gate.release("wake"); + yield* Effect.raceFirst( + watch.until(() => watch.state.variant === "drained"), + Deferred.await(offered), + ); + assert.isFalse(yield* Deferred.isDone(offered)); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + }).pipe(Effect.scoped), + ); + + it.effect("removes a shell launched by a wake still waiting for its continuation", () => + Effect.gen(function* () { + const { runtime, thread } = yield* resumed([ + event("session.execution.started", { sessionID: SESSION }), + ...shellLaunched(SESSION, "sh_1"), + removeShell("sh_1"), + shellReport(SESSION, "sh_1", "error", "Shell.NotFoundError"), + reply("shell.remove", null), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: SESSION }), + drained, + ]).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: () => Effect.void, + take: Effect.never, + }), + ); + const watch = yield* watchCalls(runtime, thread); + yield* watch.until(() => watch.state.roster?.length === 1); + assert.deepEqual(watch.state.roster, [{ taskId: "sh_1", kind: "command" }]); + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + yield* watch.until(() => watch.state.variant === "drained"); + assert.deepEqual(watch.state.roster, []); + assert.isFalse(watch.state.pending); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + }).pipe(Effect.scoped), + ); + + it.effect("continues a wake that a stopped shell shares with a finished one", () => + Effect.gen(function* () { + const offers: Array = []; + const continued = yield* Deferred.make(); + const gate = makeProviderReplayGate(["ended"]); + const { runtime, thread } = yield* resumed( + [ + ...backgroundShell("sh_a"), + event("session.execution.succeeded", { sessionID: SESSION }), + removeShell("sh_a"), + reply("shell.remove", { status: 500, body: { name: "UnknownError", data: {} } }), + event("session.execution.started", { sessionID: SESSION }), + ...shellLaunched(SESSION, "sh_b"), + event("session.execution.succeeded", { sessionID: SESSION }, "ended"), + shellReport(SESSION, "sh_a", "completed", "A_DONE"), + shellReport(SESSION, "sh_b", "completed", "B_DONE"), + event("session.execution.started", { sessionID: SESSION }), + ], + { replayGate: gate }, + ).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: (request) => + Effect.suspend(() => { + offers.push(request); + return offers.length === 2 ? Deferred.succeed(continued, undefined) : Effect.void; + }).pipe(Effect.asVoid), + take: Effect.never, + }), + ); + const watch = yield* watchCalls(runtime, thread); + yield* runtime.startTurn(turnInput(thread)); + yield* watch.until(() => watch.terminals.length > 0); + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + yield* watch.until(() => offers.length === 1 && watch.state.roster?.length === 2); + yield* runtime.startTurn(continuationTurn(thread, offers[0]!)); + // Replaying the held launch names the shell without counting it again. + yield* watch.until(() => + (watch.state.roster ?? []).every((task) => (task as { description?: string }).description), + ); + assert.deepEqual(watch.state.roster, [ + { taskId: "sh_a", kind: "command", description: SHELL_COMMAND }, + { taskId: "sh_b", kind: "command", description: SHELL_COMMAND }, + ]); + gate.release("ended"); + yield* Deferred.await(continued); + assert.include(offers[1]?.detail ?? "", "B_DONE"); + assert.deepEqual(watch.state.roster, []); + }).pipe(Effect.scoped), + ); + it.effect("lists a finished subagent's background shell as a plain command a Stop removes", () => Effect.gen(function* () { const { runtime, thread } = yield* resumed([ @@ -1073,9 +1215,9 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { drained, ]); const watch = yield* watchCalls(runtime, thread); - const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped); yield* runtime.startTurn(withLineage(thread)); - assert.equal((yield* Fiber.join(terminal))?.status, "completed"); + yield* watch.until(() => watch.terminals.length > 0); + assert.deepEqual(watch.terminals, ["completed"]); yield* watch.until(() => watch.calls.get("Sleep")?.status === "completed"); assert.deepEqual(watch.state.roster, [ { taskId: "sh_child", kind: "command", description: SHELL_COMMAND }, @@ -1326,6 +1468,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { pending: boolean | undefined; variant: unknown; } = { roster: undefined, pending: undefined, variant: undefined }; + const terminals: Array = []; let wake = yield* Deferred.make(); yield* runtime.events.pipe( Stream.tap((event) => @@ -1336,6 +1479,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { result: event.subagent.result, }); } + if (event.type === "turn.terminal") terminals.push(event.status); if (event.type === "provider_thread.updated" && event.providerThread.id === thread.id) { state.roster = event.providerThread.pendingBackgroundTasks; state.pending = event.pendingBackgroundWork; @@ -1371,7 +1515,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { yield* until(check); }); }); - return { calls, callsAtChildTurnStart, childTurns, state, until }; + return { calls, callsAtChildTurnStart, childTurns, state, terminals, until }; }); const launchedAndRunning = [ ...backgroundLaunch(CHILD), diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index 4fb830135ac7b..e3383310bc175 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -316,10 +316,7 @@ interface ThreadState { } >; /** Running background shells by id; a shell stays at its launch directory when the session moves. */ - readonly shells: Map< - string, - { readonly directory: string; readonly command: string | undefined } - >; + readonly shells: Map; /** Shells whose end arrived before their launch result; OpenCode does not wait for the launch to return. */ readonly endedShells: Set; /** Child sessions and shells the user stopped; a wake their reports alone cause runs nothing. */ @@ -526,6 +523,9 @@ const reportOutcome = (state: string | undefined) => ? ("cancelled" as const) : ("unknown" as const); +const shellCommandOf = (input: Readonly> | undefined) => + stringField(input, "command")?.trim(); + const subagentOutput = (text: string) => /^]*>\n?([\s\S]*?)\n?<\/subagent>$/.exec(text.trim())?.[1] ?? text; @@ -841,7 +841,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const owesWork = (state: ThreadState) => state.wakes.length > 0 || - state.reports.size > 0 || + [...state.reports.values()].some((entry) => !state.stopped.has(entry.sourceId)) || state.shells.size > 0 || (state.subagent !== undefined && busy.has(state.sessionId)) || [...state.calls.values()].some((call) => call.background); @@ -2132,18 +2132,16 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr return yield* settleCall(call, "completed", subagentOutput(textOf(event.data.content))); } const output = textOf(event.data.content); - const launched = trackShellLaunch( - state, - event.data.metadata, - turn.tools.get(event.data.id)?.input, - ); + // A launch held with its wake arrived before its turn could name it. + const shell = state.shells.get(stringField(event.data.metadata, "shellID") ?? ""); + if (shell !== undefined) + shell.command ??= shellCommandOf(turn.tools.get(event.data.id)?.input); yield* emitTool(state, turn, event.data.id, "completed", { output, files: event.data.content.flatMap((part) => (part.type === "file" ? [part] : [])), metadata: event.data.metadata, }); turn.tools.delete(event.data.id); - if (launched) yield* syncRosterAncestors(state); return; } case "session.tool.failed": { @@ -2340,10 +2338,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const shellId = stringField(metadata, "shellID"); if (metadata?.["status"] !== "running" || shellId === undefined) return false; if (state.endedShells.delete(shellId) || state.shells.has(shellId)) return false; - state.shells.set(shellId, { - directory: state.directory, - command: stringField(input, "command")?.trim(), - }); + state.shells.set(shellId, { directory: state.directory, command: shellCommandOf(input) }); return true; }; @@ -2356,17 +2351,40 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr }, ) => { const shellId = stringField(payload.metadata, "shellID"); - if (stringField(payload.metadata, "source") !== "shell" || shellId === undefined) return; + if (stringField(payload.metadata, "source") !== "shell" || shellId === undefined) + return false; const shell = state.shells.get(shellId); - if (!state.shells.delete(shellId)) state.endedShells.add(shellId); - // Removing a shell reports it to its session, which wakes that session. - if (!state.stopped.has(shellId)) return; + const ended = state.shells.delete(shellId); + if (!ended) state.endedShells.add(shellId); + // Removing a shell reports it to its session too, which wakes that session. state.reports.set(inboxId, { inboxId, sourceId: shellId, text: payload.text, - report: { kind: "command", label: shell?.command, outcome: "cancelled" }, + report: { + kind: "command", + label: shell?.command, + outcome: state.stopped.has(shellId) + ? "cancelled" + : reportOutcome(stringField(payload.metadata, "state")), + }, }); + return ended; + }; + + // Counted on arrival, so Stop sees what a held wake started and a replay counts nothing twice. + const trackShells = (state: ThreadState, event: OpenCode2Client.OpenCode2StreamEvent) => { + if (event.type === "session.tool.success") { + const input = state.active?.tools.get(event.data.id)?.input; + return trackShellLaunch(state, event.data.metadata, input); + } + if (event.type === "session.inbox.enqueued" && event.data.item.type === "synthetic") { + return trackShellEnd(state, event.data.inboxID, event.data.item.payload); + } + if (event.type === "session.inbox.delivered" || event.type === "session.inbox.cancelled") { + state.reports.delete(event.data.inboxID); + } + return false; }; const handleEvent = Effect.fnUntraced(function* (event: OpenCode2Client.OpenCode2StreamEvent) { @@ -2399,6 +2417,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr busy.delete(sessionId); } } + const tracked = threads.get(sessionId ?? ""); + const shellsChanged = tracked !== undefined && trackShells(tracked, event); if (event.type === "session.created" && event.data.parentID !== undefined) { const strayCaller = strays.get(event.data.parentID); if (strayCaller !== undefined) { @@ -2423,9 +2443,10 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ) { held.running = false; } - return; + } else { + yield* route(event, sessionId); } - return yield* route(event, sessionId); + if (shellsChanged) yield* syncRosterAncestors(tracked); }); const route = Effect.fnUntraced(function* ( @@ -2506,7 +2527,6 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr state.subagent.queued ??= event.data.inboxID; } if (event.type === "session.inbox.enqueued" && event.data.item.type === "synthetic") { - trackShellEnd(state, event.data.inboxID, event.data.item.payload); return yield* onReport(state, event.data.inboxID, event.data.item.payload); } if (event.type === "session.inbox.cancelled" || event.type === "session.inbox.delivered") { From 4b2a6e87841b2a0119fd4e86cbccabf2a874c814 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 8 Nov 2026 06:36:10 +0000 Subject: [PATCH 05/14] test(server): read the reconnect case's events through one consumer Co-Authored-By: Claude Opus 5.5 --- .../Adapters/OpenCode2AdapterV2.test.ts | 23 +++++++------------ 1 file changed, 8 insertions(+), 15 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index a0ff360be47ce..57856a3fdf6dd 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -1008,23 +1008,16 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { event("session.execution.succeeded", { sessionID: SESSION }), { type: "runtime_exit", status: "success" } as const, ]); - const events = yield* runtime.events.pipe(Stream.runCollect, Effect.forkScoped); - const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped); + const watch = yield* watchCalls(runtime, thread); yield* runtime.startTurn(turnInput(thread)); - assert.equal((yield* Fiber.join(terminal))?.status, "completed"); + yield* watch.until(() => watch.terminals.length > 0); + assert.deepEqual(watch.terminals, ["completed"]); + assert.deepEqual(watch.state.roster, [ + { taskId: "sh_1", kind: "command", description: SHELL_COMMAND }, + ]); yield* TestClock.adjust("1 minute"); - const last = (yield* Fiber.join(events)) - .filter( - (event) => - event.type === "provider_thread.updated" && event.providerThread.id === thread.id, - ) - .at(-1); - assert.deepEqual( - last?.type === "provider_thread.updated" - ? [last.providerThread.pendingBackgroundTasks, last.pendingBackgroundWork] - : undefined, - [[], false], - ); + yield* watch.until(() => watch.state.roster?.length === 0); + assert.isFalse(watch.state.pending); }).pipe(Effect.scoped), ); From 94e070a79d8ad527319224aee90f3a0523dbf4a4 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 8 Nov 2026 06:59:55 +0000 Subject: [PATCH 06/14] fix(opencode): count a held subagent's shells and remove shells launched during Stop A subagent's shell launched during a held autonomous wake arrived before the subagent had state, so it was never counted: replay listed only the subagent, and Stop neither removed it nor kept it from waking a continuation. Held events whose session had no state are now marked and counted on replay, and Stop removes their shells when it discards the wake. Stop snapshotted the shells before its removes awaited, so a shell launched meanwhile survived Stop. Stop now keeps removing until, under the lock, no unremoved shell remains. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/OpenCode2AdapterV2.test.ts | 180 ++++++++++++++++-- .../src/server/v2/adapter.ts | 141 ++++++++++---- 2 files changed, 268 insertions(+), 53 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 57856a3fdf6dd..0a00a33008e41 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -945,19 +945,29 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { event("session.execution.started", { sessionID: SESSION }), ...shellLaunched(SESSION, shellID), ]; - const shellReport = (sessionID: string, shellID: string, state: string, text: string) => - event("session.inbox.enqueued", { - inboxID: `msg_${state}_${shellID}`, - sessionID, - item: { - type: "synthetic", - payload: { - text: `${text}`, - metadata: { source: "shell", shellID, state }, + const shellReport = ( + sessionID: string, + shellID: string, + state: string, + text: string, + label?: string, + ) => + event( + "session.inbox.enqueued", + { + inboxID: `msg_${state}_${shellID}`, + sessionID, + item: { + type: "synthetic", + payload: { + text: `${text}`, + metadata: { source: "shell", shellID, state }, + }, + delivery: "steer", }, - delivery: "steer", }, - }); + label, + ); const removeShell = (shellID: string) => out("shell.remove", { id: shellID, "location[directory]": WORK }); // OpenCode reports a removed shell to its session and wakes it before the remove returns. @@ -1139,6 +1149,154 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + const heldVariant = (variant: string, label?: string) => + event( + "session.model.selected", + { sessionID: SESSION, model: { providerID: "opencode", id: "big-pickle", variant } }, + label, + ); + // The wake holds the call that would name the subagent, so its shell arrives for an unknown session. + const heldSubagentShell = (): ReadonlyArray => [ + event("session.execution.started", { sessionID: SESSION }), + ...backgroundLaunch(CHILD).slice(3), + returnedRunning("call-background"), + event("session.execution.started", { sessionID: CHILD }), + ...shellLaunched(CHILD, "sh_child"), + event("session.execution.succeeded", { sessionID: SESSION }), + heldVariant("held"), + ]; + const recordOffers = (offers: Array) => + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: (request) => + Effect.sync(() => { + offers.push(request); + }), + take: Effect.never, + }); + + it.effect("lists a subagent's shell from a held wake under the subagent once the wake runs", () => + Effect.gen(function* () { + const offers: Array = []; + const gate = makeProviderReplayGate(["after"]); + const { runtime, thread } = yield* resumed( + [...heldSubagentShell(), heldVariant("after", "after")], + { replayGate: gate }, + ).pipe(recordOffers(offers)); + const watch = yield* watchCalls(runtime, thread); + yield* watch.until(() => watch.state.variant === "held"); + assert.lengthOf(offers, 1); + yield* runtime.startTurn(continuationTurn(thread, offers[0]!)); + gate.release("after"); + yield* watch.until(() => watch.state.variant === "after"); + const [subagent, shell] = watch.state.roster ?? []; + assert.include(subagent, { kind: "subagent", taskId: "call-background" }); + assert.deepEqual(shell, { + taskId: "sh_child", + kind: "command", + description: SHELL_COMMAND, + parentTaskId: "call-background", + }); + }).pipe(Effect.scoped), + ); + + it.effect("removes a subagent's shell from a held wake that Stop discards", () => + Effect.gen(function* () { + const offers: Array = []; + // Released up front: it only records that the remove's report was emitted. + const gate = makeProviderReplayGate(["removing"]); + gate.release("removing"); + const { runtime, thread } = yield* resumed( + [ + ...heldSubagentShell(), + removeShell("sh_child"), + shellReport(CHILD, "sh_child", "error", "Shell.NotFoundError", "removing"), + event("session.execution.started", { sessionID: CHILD }), + reply("shell.remove", null), + out("session.interrupt", { sessionID: CHILD }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: CHILD }), + drained, + ], + { replayGate: gate }, + ).pipe(recordOffers(offers)); + const watch = yield* watchCalls(runtime, thread); + yield* watch.until(() => watch.state.variant === "held"); + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + assert.isTrue(gate.hasReached("removing")); + yield* watch.until(() => watch.state.variant === "drained"); + assert.lengthOf(offers, 1); + assert.isFalse(watch.state.pending); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + }).pipe(Effect.scoped), + ); + + it.effect("removes a shell launched while Stop removes another", () => + Effect.gen(function* () { + const offered = yield* Deferred.make(); + const gate = makeProviderReplayGate(["removed", "wake"]); + const { runtime, thread } = yield* resumed( + [ + ...backgroundShell("sh_1"), + removeShell("sh_1"), + shellReport(SESSION, "sh_1", "error", "Shell.NotFoundError"), + // The running turn launches another shell before the first remove returns. + ...shellLaunched(SESSION, "sh_2"), + // Gates hold only events; this one holds the reply until the test saw the launch. + heldVariant("launched", "removed"), + reply("shell.remove", null), + removeShell("sh_2"), + shellReport(SESSION, "sh_2", "error", "Shell.NotFoundError"), + reply("shell.remove", null), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: SESSION }), + event("session.execution.started", { sessionID: SESSION }, "wake"), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: SESSION }), + drained, + ], + { replayGate: gate }, + ).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid), + take: Effect.never, + }), + ); + const watch = yield* watchCalls(runtime, thread); + yield* runtime.startTurn(turnInput(thread)); + yield* watch.until(() => watch.state.roster?.length === 1); + const stop = yield* runtime + .interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }) + .pipe(Effect.forkScoped); + yield* watch.until(() => + (watch.state.roster ?? []).some((task) => (task as { taskId: string }).taskId === "sh_2"), + ); + gate.release("removed"); + yield* Fiber.join(stop); + yield* watch.until(() => watch.terminals.length > 0); + assert.deepEqual(watch.terminals, ["interrupted"]); + assert.deepEqual(watch.state.roster, []); + assert.isFalse(watch.state.pending); + assert.isTrue(yield* Effect.promise(() => gate.waitForReached("wake"))); + gate.release("wake"); + yield* Effect.raceFirst( + watch.until(() => watch.state.variant === "drained"), + Deferred.await(offered), + ); + assert.isFalse(yield* Deferred.isDone(offered)); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + }).pipe(Effect.scoped), + ); + it.effect("continues a wake that a stopped shell shares with a finished one", () => Effect.gen(function* () { const offers: Array = []; diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index e3383310bc175..c438415808ac0 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -735,6 +735,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const threads = new Map(); const childOwners = new Map(); const announced = new Map["data"]>(); + /** Held events whose session had no state yet, so their shells were not counted on arrival. */ + const untracked = new WeakSet(); const strays = new Map(); const busy = new Set(); const pending = new Map(); @@ -2437,6 +2439,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const held = holder?.wakes.at(-1); if (holder !== undefined && held !== undefined) { held.events.push(event); + if (tracked === undefined) untracked.add(event); if ( sessionId === holder.sessionId && (event.type === "unreadable.execution.ended" || executionEnd(event.type)) @@ -3317,9 +3320,37 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const replay = Effect.fnUntraced(function* (wake: Wake) { wake.dropped = true; - for (const event of wake.events) yield* route(event, sessionOfEvent(event), true); + for (const event of wake.events) { + const sessionId = sessionOfEvent(event); + // A subagent's state exists only once its held call is replayed. + const state = untracked.has(event) ? threads.get(sessionId ?? "") : undefined; + if (state !== undefined) trackShells(state, event); + yield* route(event, sessionId, true); + } }); + // Shells a held wake started in sessions it had not named yet; replay would count them. + const heldShells = (caller: ThreadState) => { + const running = new Set(); + for (const event of caller.wakes.flatMap((wake) => wake.events)) { + if (!untracked.has(event)) continue; + if ( + event.type === "session.tool.success" && + event.data.metadata?.["status"] === "running" + ) { + const shellId = stringField(event.data.metadata, "shellID"); + if (shellId !== undefined) running.add(shellId); + } + if (event.type === "session.inbox.enqueued" && event.data.item.type === "synthetic") { + const { metadata } = event.data.item.payload; + if (stringField(metadata, "source") === "shell") { + running.delete(stringField(metadata, "shellID") ?? ""); + } + } + } + return running; + }; + const discardWakes = (caller: ThreadState, wakes: ReadonlyArray) => { const retired = new Set(); const retire = (childId: string) => { @@ -3419,23 +3450,43 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const children = new Map( calls.flatMap((call) => childrenOf(call).map((childId) => [childId, call.state] as const)), ); - const shells = callers.flatMap((caller) => - [...caller.shells].map(([id, shell]) => ({ caller, id, directory: shell.directory })), - ); - // OpenCode reports and wakes the shell's session before the remove returns. - for (const { caller, id } of shells) caller.stopped.add(id); - const removed = yield* Effect.filter(shells, ({ id, directory }) => - client.shell.remove({ id: Shell.ID.make(id), location: { directory } }).pipe( - // OpenCode declares no not-found error for remove; only the directory can be gone. - Effect.catchTags({ LocationNotFoundError: () => Effect.void }), - Effect.timeout(INTERRUPT_TIMEOUT), - Effect.tapCause((cause) => - Effect.logWarning("Could not stop an OpenCode background shell.", cause), - ), - Effect.exit, - Effect.map(Exit.isSuccess), - ), - ); + const attempted = new Set(); + const unremoved = () => + sessionsOf(state) + .flatMap((caller) => [ + ...[...caller.shells].map(([id, shell]) => ({ + caller, + id, + directory: shell.directory, + })), + ...[...heldShells(caller)].map((id) => ({ caller, id, directory: caller.directory })), + ]) + .filter(({ id }) => !attempted.has(id)); + const removed: Array<{ readonly caller: ThreadState; readonly id: string }> = []; + // Sessions still running can launch shells while a remove waits. + const removeShells = Effect.fnUntraced(function* () { + for (let shells = unremoved(); shells.length > 0; shells = unremoved()) { + // OpenCode reports and wakes the shell's session before the remove returns. + for (const { caller, id } of shells) { + attempted.add(id); + caller.stopped.add(id); + } + const reached = yield* Effect.filter(shells, ({ id, directory }) => + client.shell.remove({ id: Shell.ID.make(id), location: { directory } }).pipe( + // OpenCode declares no not-found error for remove; only the directory can be gone. + Effect.catchTags({ LocationNotFoundError: () => Effect.void }), + Effect.timeout(INTERRUPT_TIMEOUT), + Effect.tapCause((cause) => + Effect.logWarning("Could not stop an OpenCode background shell.", cause), + ), + Effect.exit, + Effect.map(Exit.isSuccess), + ), + ); + removed.push(...reached); + } + }); + yield* removeShells(); const unreached = new Set(); for (const [childId, caller] of children) { caller.stopped.add(childId); @@ -3461,30 +3512,36 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr Effect.ignore({ log: true }), ); } - yield* lock.withPermit( - Effect.gen(function* () { - for (const call of calls) { - if (childrenOf(call).some((childId) => unreached.has(childId))) continue; - yield* settleCall(call, "interrupted"); - } - for (const caller of callers) { - for (const report of caller.reports.values()) caller.stopped.add(report.sourceId); - } - for (const { caller, id } of removed) caller.shells.delete(id); - for (const caller of new Set(removed.map(({ caller }) => caller))) { - yield* syncRosterAncestors(caller); - } - const running = state.wakes.some((wake) => wake.running); - discardWakes(state, state.wakes.splice(0)); - yield* syncRosterAncestors(state); - if (running) { - markUnsettled(state); - yield* client.session - .interrupt({ sessionID: Session.ID.make(state.sessionId) }) - .pipe(Effect.timeout(INTERRUPT_TIMEOUT), Effect.ignore({ log: true })); - } - }), - ); + // Checked under the lock, so no shell launched before Stop settles outlives it. + for (let settled = false; !settled;) { + yield* removeShells(); + settled = yield* lock.withPermit( + Effect.gen(function* () { + if (unremoved().length > 0) return false; + for (const call of calls) { + if (childrenOf(call).some((childId) => unreached.has(childId))) continue; + yield* settleCall(call, "interrupted"); + } + for (const caller of callers) { + for (const report of caller.reports.values()) caller.stopped.add(report.sourceId); + } + for (const { caller, id } of removed) caller.shells.delete(id); + for (const caller of new Set(removed.map(({ caller }) => caller))) { + yield* syncRosterAncestors(caller); + } + const running = state.wakes.some((wake) => wake.running); + discardWakes(state, state.wakes.splice(0)); + yield* syncRosterAncestors(state); + if (running) { + markUnsettled(state); + yield* client.session + .interrupt({ sessionID: Session.ID.make(state.sessionId) }) + .pipe(Effect.timeout(INTERRUPT_TIMEOUT), Effect.ignore({ log: true })); + } + return true; + }), + ); + } yield* stopStrays(state); }); From 1a7c4451ae2b82e634a1f8a014ffd6bc252fbb3b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 8 Nov 2026 07:05:20 +0000 Subject: [PATCH 07/14] fix(opencode): a failed shell check keeps the shell, not the reconnect Co-Authored-By: Claude Opus 5.5 --- .../Adapters/OpenCode2AdapterV2.test.ts | 23 +++++++++++++++++++ .../Adapters/OpenCode2AdapterV2.testkit.ts | 3 +++ .../src/server/v2/adapter.ts | 7 ++++++ 3 files changed, 33 insertions(+) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 0a00a33008e41..3f41293ba44e7 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -1011,6 +1011,29 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + it.effect("keeps a background shell tracked when its check fails after a reconnect", () => + Effect.gen(function* () { + const { runtime, thread } = yield* resumed([ + ...backgroundShell("sh_1"), + event("session.execution.succeeded", { sessionID: SESSION }), + { type: "runtime_exit", status: "success" } as const, + out("event.subscribe"), + out("session.active"), + replyData("session.active", {}), + out("shell.get", { id: "sh_1", "location[directory]": WORK }), + reply("shell.get", { status: 500, body: { name: "UnknownError", data: {} } }), + drained, + ]); + const watch = yield* watchCalls(runtime, thread); + yield* runtime.startTurn(turnInput(thread)); + yield* watch.until(() => watch.state.variant === "drained"); + assert.deepEqual(watch.state.roster, [ + { taskId: "sh_1", kind: "command", description: SHELL_COMMAND }, + ]); + assert.isTrue(watch.state.pending); + }).pipe(Effect.scoped), + ); + it.effect("clears a background shell from the roster when the stream cannot reconnect", () => 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 9ab68864edad3..308d956f04f92 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts @@ -82,6 +82,9 @@ const operationOf = ( if (method === "DELETE" && shell !== null) { return { type: "shell.remove", input: { id: decodeURIComponent(shell[1] ?? ""), ...query } }; } + if (method === "GET" && shell !== null) { + return { type: "shell.get", input: { id: decodeURIComponent(shell[1] ?? ""), ...query } }; + } if (session !== null) { const [, sessionID, rest = ""] = session; // The client sends absent optional fields as `null`; recordings omit them. diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index c438415808ac0..43d8a3c10a9e9 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -2757,6 +2757,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ShellNotFoundError: () => Effect.succeed(false), LocationNotFoundError: () => Effect.succeed(false), }), + Effect.timeout(ACTIVE_CHECK_TIMEOUT), + // Only prunes stale shells, so a failed check keeps the shell rather than the stream. + Effect.catch((cause) => + Effect.logWarning("Could not check an OpenCode background shell.", { cause }).pipe( + Effect.as(true), + ), + ), ); if (!running) state.shells.delete(id); } From 5cd4fc18f32bd91fc9226366ab35bb25b3f12c16 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 8 Nov 2026 07:12:19 +0000 Subject: [PATCH 08/14] fix(opencode): check tracked shells concurrently on reconnect Co-Authored-By: Claude Opus 5.5 --- .../src/server/v2/adapter.ts | 22 +++++++++++-------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index 43d8a3c10a9e9..d16bdfd4b4894 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -2746,10 +2746,12 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr for (const session of sessions.toReversed()) yield* syncRoster(session); } // A shell's end may have been lost with the stream, and backfill may re-add an ended one. - for (const state of threads.values()) { - if (state.shells.size === 0) continue; - for (const [id, shell] of state.shells) { - const running = yield* client.shell + const tracked = [...threads.values()].filter((state) => state.shells.size > 0); + // Concurrent, so stalled checks cost one timeout, not one each, within RECONCILE_TIMEOUT. + yield* Effect.forEach( + tracked.flatMap((state) => [...state.shells].map(([id, shell]) => ({ state, id, shell }))), + ({ state, id, shell }) => + client.shell .get({ id: Shell.ID.make(id), location: { directory: shell.directory } }) .pipe( Effect.map(({ data }) => data.status === "running"), @@ -2764,11 +2766,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr Effect.as(true), ), ), - ); - if (!running) state.shells.delete(id); - } - yield* syncRosterAncestors(state); - } + Effect.map((running) => { + if (!running) state.shells.delete(id); + }), + ), + { concurrency: "unbounded", discard: true }, + ); + for (const state of tracked) yield* syncRosterAncestors(state); // 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); From c88510a57629353853b405fbb6f0ca578e98fb40 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 11:40:38 +0000 Subject: [PATCH 09/14] fix(opencode): Stop ends work launched while it runs A user Stop drained background shells and stopped subagents before it interrupted the foreground turn, and stopped only the subagents it saw when it began. A shell the turn launched while its interrupt was in flight, or a subagent launched while a shell remove waited, outlived Stop. Stop now interrupts the foreground turn before its final drain, waits for that execution to end so every launch it made has been seen, and re-derives the subagent sessions to stop on each pass until no new shell or session appears. Steering interrupts are unchanged. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/OpenCode2AdapterV2.test.ts | 67 ++++++ .../src/server/v2/adapter.ts | 190 +++++++++++------- 2 files changed, 179 insertions(+), 78 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 3f41293ba44e7..39b30bbfe8d09 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -1320,6 +1320,73 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + it.effect("removes a shell launched while Stop interrupts the turn", () => + Effect.gen(function* () { + const { runtime, thread } = yield* resumed([ + ...backgroundShell("sh_1"), + removeShell("sh_1"), + shellReport(SESSION, "sh_1", "error", "Shell.NotFoundError"), + reply("shell.remove", null), + out("session.interrupt", { sessionID: SESSION }), + // The turn launches another shell before the interrupt returns. + ...shellLaunched(SESSION, "sh_2"), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: SESSION }), + removeShell("sh_2"), + shellReport(SESSION, "sh_2", "error", "Shell.NotFoundError"), + reply("shell.remove", null), + drained, + ]); + const watch = yield* watchCalls(runtime, thread); + yield* runtime.startTurn(turnInput(thread)); + yield* watch.until(() => watch.state.roster?.length === 1); + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + yield* watch.until(() => watch.terminals.length > 0); + assert.deepEqual(watch.terminals, ["interrupted"]); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + yield* watch.until(() => watch.state.variant === "drained"); + assert.deepEqual(watch.state.roster, []); + }).pipe(Effect.scoped), + ); + + it.effect("stops a subagent launched while Stop removes a shell", () => + Effect.gen(function* () { + const { runtime, thread } = yield* resumed([ + ...backgroundShell("sh_1"), + removeShell("sh_1"), + shellReport(SESSION, "sh_1", "error", "Shell.NotFoundError"), + // The turn launches a background subagent before the remove returns. + ...backgroundLaunch(CHILD).slice(3), + returnedRunning("call-background"), + event("session.execution.started", { sessionID: CHILD }), + reply("shell.remove", null), + out("session.interrupt", { sessionID: CHILD }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: CHILD }), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: SESSION }), + drained, + ]); + const watch = yield* watchCalls(runtime, thread); + yield* runtime.startTurn(withLineage(thread)); + yield* watch.until(() => watch.state.roster?.length === 1); + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + yield* watch.until(() => watch.state.variant === "drained"); + assert.deepEqual(watch.terminals, ["interrupted"]); + assert.deepEqual(watch.state.roster, []); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + }).pipe(Effect.scoped), + ); + it.effect("continues a wake that a stopped shell shares with a finished one", () => Effect.gen(function* () { const offers: Array = []; diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index d16bdfd4b4894..c6a6a3468be5b 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -739,6 +739,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const untracked = new WeakSet(); const strays = new Map(); const busy = new Set(); + const idleWaiters = new Map>(); const pending = new Map(); const lock = yield* Semaphore.make(1); const clearing = new Map>(); @@ -754,6 +755,19 @@ 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); + const markIdle = Effect.fnUntraced(function* (sessionId: string) { + busy.delete(sessionId); + const waiter = idleWaiters.get(sessionId); + idleWaiters.delete(sessionId); + if (waiter !== undefined) yield* Deferred.succeed(waiter, undefined); + }); + // An execution's events precede its end, so once it ends everything it launched has been seen. + const awaitIdle = Effect.fnUntraced(function* (sessionId: string) { + if (!busy.has(sessionId)) return; + const waiter = idleWaiters.get(sessionId) ?? (yield* Deferred.make()); + idleWaiters.set(sessionId, waiter); + yield* Deferred.await(waiter).pipe(Effect.timeout(INTERRUPT_TIMEOUT), Effect.ignore); + }); const markUnsettled = (state: ThreadState) => { state.unsettled = true; state.unsettledStarted = busy.has(state.sessionId); @@ -2416,7 +2430,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr } busy.add(sessionId); } else if (event.type === "unreadable.execution.ended" || executionEnd(event.type)) { - busy.delete(sessionId); + yield* markIdle(sessionId); } } const tracked = threads.get(sessionId ?? ""); @@ -2704,7 +2718,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr return; } const active = yield* client.session.active(); - for (const sessionId of busy) if (!(sessionId in active)) busy.delete(sessionId); + for (const sessionId of busy) if (!(sessionId in active)) yield* markIdle(sessionId); for (const state of threads.values()) { if (state.unsettled && !(state.sessionId in active)) state.unsettledStarted = true; } @@ -3441,16 +3455,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ); }); - const stopBackground = Effect.fnUntraced(function* (state: ThreadState) { - const calls = runningCalls(state).filter((call) => call.background); - const callers = sessionsOf(state); - // Ended subagent calls can run again to answer their own background reports. - const followUps = callers.filter( - (caller) => - caller.subagent !== undefined && - busy.has(caller.sessionId) && - !isOrchestrationV2WorkActive(caller.subagent.call.status), - ); + // The foreground stop runs before the final drain and succeeds with whether it interrupted a run, + // which can launch more work until it ends. + const stopBackground = Effect.fnUntraced(function* ( + state: ThreadState, + stopForeground: Effect.Effect, + ) { + const backgroundCalls = () => runningCalls(state).filter((call) => call.background); // Child sessions are announced before call progress names them; use announcements to stop unnamed children. const announcedTo = (caller: ThreadState) => [...announced.values()].flatMap((info) => @@ -3458,9 +3469,44 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ); const childrenOf = (call: SubagentCall) => call.child === undefined ? announcedTo(call.state) : [call.child.sessionId]; - const children = new Map( - calls.flatMap((call) => childrenOf(call).map((childId) => [childId, call.state] as const)), - ); + const interrupted = new Set(); + const unstopped = () => + [ + ...backgroundCalls().flatMap((call) => + childrenOf(call).map((sessionId) => ({ sessionId, caller: call.state })), + ), + // Ended subagent calls can run again to answer their own background reports. + ...sessionsOf(state) + .filter( + (caller) => + caller.subagent !== undefined && + busy.has(caller.sessionId) && + !isOrchestrationV2WorkActive(caller.subagent.call.status), + ) + .map((caller) => ({ sessionId: caller.sessionId, caller: undefined })), + ].filter(({ sessionId }) => !interrupted.has(sessionId)); + const unreached = new Set(); + const stopSessions = Effect.fnUntraced(function* () { + for (let sessions = unstopped(); sessions.length > 0; sessions = unstopped()) { + for (const { sessionId, caller } of sessions) { + if (interrupted.has(sessionId)) continue; + interrupted.add(sessionId); + caller?.stopped.add(sessionId); + const reached = yield* client.session + .interrupt({ sessionID: Session.ID.make(sessionId) }) + .pipe( + Effect.catchTags({ SessionNotFoundError: () => Effect.void }), + Effect.timeout(INTERRUPT_TIMEOUT), + Effect.tapCause((cause) => + Effect.logWarning("Could not stop an OpenCode subagent.", cause), + ), + Effect.exit, + Effect.map(Exit.isSuccess), + ); + if (!reached && caller !== undefined) unreached.add(sessionId); + } + } + }); const attempted = new Set(); const unremoved = () => sessionsOf(state) @@ -3498,42 +3544,21 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr } }); yield* removeShells(); - const unreached = new Set(); - for (const [childId, caller] of children) { - caller.stopped.add(childId); - const reached = yield* client.session - .interrupt({ sessionID: Session.ID.make(childId) }) - .pipe( - Effect.catchTags({ SessionNotFoundError: () => Effect.void }), - Effect.timeout(INTERRUPT_TIMEOUT), - Effect.tapCause((cause) => - Effect.logWarning("Could not stop an OpenCode subagent.", cause), - ), - Effect.exit, - Effect.map(Exit.isSuccess), - ); - if (!reached) unreached.add(childId); - } - for (const followUp of followUps) { - yield* client.session - .interrupt({ sessionID: Session.ID.make(followUp.sessionId) }) - .pipe( - Effect.catchTags({ SessionNotFoundError: () => Effect.void }), - Effect.timeout(INTERRUPT_TIMEOUT), - Effect.ignore({ log: true }), - ); - } + yield* stopSessions(); + const foreground = yield* Effect.exit(stopForeground); + if (Exit.isSuccess(foreground) && foreground.value) yield* awaitIdle(state.sessionId); // Checked under the lock, so no shell launched before Stop settles outlives it. for (let settled = false; !settled;) { yield* removeShells(); + yield* stopSessions(); settled = yield* lock.withPermit( Effect.gen(function* () { - if (unremoved().length > 0) return false; - for (const call of calls) { + if (unremoved().length > 0 || unstopped().length > 0) return false; + for (const call of backgroundCalls()) { if (childrenOf(call).some((childId) => unreached.has(childId))) continue; yield* settleCall(call, "interrupted"); } - for (const caller of callers) { + for (const caller of sessionsOf(state)) { for (const report of caller.reports.values()) caller.stopped.add(report.sourceId); } for (const { caller, id } of removed) caller.shells.delete(id); @@ -3554,6 +3579,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ); } yield* stopStrays(state); + yield* foreground; }); const listSkills = (location: { readonly directory: string }) => @@ -4024,42 +4050,50 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const state = threads.get(sessionId); if (state === undefined) return; const turn = state.active; + const stopTurn = Effect.gen(function* () { + if (turn === undefined || turn.providerTurn.id !== interruptInput.providerTurnId) { + return false; + } + turn.interrupted = true; + const reply = yield* client.session + .interrupt({ sessionID: Session.ID.make(sessionId) }) + .pipe( + Effect.timeoutOption(INTERRUPT_TIMEOUT), + Effect.tapError(() => + Effect.sync(() => { + turn.interrupted = false; + }), + ), + ); + if (reply._tag === "None") { + markUnsettled(state); + yield* lock.withPermit(finishTurn(state, { status: "interrupted" })); + return false; + } + // Stop between steer executions retracts pending steers and skips any run they already started. + if (!reply.value.interrupted && state.active === turn && turn.heldEnd !== undefined) { + markUnsettled(state); + yield* lock.withPermit(finishTurn(state, { status: "interrupted" })); + yield* cancelStrandedSteers(state); + return false; + } + if (!reply.value.interrupted && state.active === turn && turn.unsent) { + yield* lock.withPermit(finishTurn(state, { status: "interrupted" })); + return false; + } + if (!reply.value.interrupted && state.active === turn) { + turn.interrupted = false; + return yield* new ProviderAdapter.ProviderAdapterProtocolError({ + driver: OPENCODE_PROVIDER, + detail: `OpenCode session ${sessionId} had nothing running to stop`, + }); + } + return reply.value.interrupted; + }); // User Stop must end background children and their report execution; steering leaves them running. - if (interruptInput.requestRuntimeRestart === true) yield* stopBackground(state); - if (turn === undefined || turn.providerTurn.id !== interruptInput.providerTurnId) { - return; - } - turn.interrupted = true; - const reply = yield* client.session - .interrupt({ sessionID: Session.ID.make(sessionId) }) - .pipe( - Effect.timeoutOption(INTERRUPT_TIMEOUT), - Effect.tapError(() => - Effect.sync(() => { - turn.interrupted = false; - }), - ), - ); - if (reply._tag === "None") { - markUnsettled(state); - return yield* lock.withPermit(finishTurn(state, { status: "interrupted" })); - } - // Stop between steer executions retracts pending steers and skips any run they already started. - if (!reply.value.interrupted && state.active === turn && turn.heldEnd !== undefined) { - markUnsettled(state); - yield* lock.withPermit(finishTurn(state, { status: "interrupted" })); - return yield* cancelStrandedSteers(state); - } - if (!reply.value.interrupted && state.active === turn && turn.unsent) { - return yield* lock.withPermit(finishTurn(state, { status: "interrupted" })); - } - if (!reply.value.interrupted && state.active === turn) { - turn.interrupted = false; - return yield* new ProviderAdapter.ProviderAdapterProtocolError({ - driver: OPENCODE_PROVIDER, - detail: `OpenCode session ${sessionId} had nothing running to stop`, - }); - } + return yield* interruptInput.requestRuntimeRestart === true + ? stopBackground(state, stopTurn) + : Effect.asVoid(stopTurn); }, (effect, interruptInput) => effect.pipe( From 975c0819bcf72251867f53f62d0b96e572844f6f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 12:14:43 +0000 Subject: [PATCH 10/14] fix(opencode): Stop waits for every interrupted session before its last shell sweep Stop waited for the foreground turn's execution to end before its final shell sweep, but not for the background subagents and follow-up sessions it interrupted, and it interrupted a running autonomous wake only after that sweep. A shell one of them launched after acknowledging the interrupt outlived Stop. Every execution Stop interrupts now goes through the same step: interrupt it, then wait (bounded by the interrupt timeout, concurrently) for its end before the next sweep. The running wake is interrupted there too, after the foreground stop, instead of inside the settle step. Shell removes in one pass run concurrently, so stalled removes cost one timeout. Steering is unchanged. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/OpenCode2AdapterV2.test.ts | 71 +++++++++++++++ .../src/server/v2/adapter.ts | 87 ++++++++++++------- 2 files changed, 125 insertions(+), 33 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 39b30bbfe8d09..d8597007206cc 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -1353,6 +1353,77 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + it.effect("removes a shell a background subagent launched while Stop interrupted it", () => + Effect.gen(function* () { + const launch = shellLaunched(CHILD, "sh_late"); + const { runtime, thread } = yield* resumed([ + ...backgroundLaunch(CHILD), + returnedRunning("call-background"), + event("session.execution.started", { sessionID: CHILD }), + ...launch.slice(0, -1), + event("session.execution.succeeded", { sessionID: SESSION }), + out("session.interrupt", { sessionID: CHILD }), + reply("session.interrupt", { interrupted: true }), + // The child's shell starts after the interrupt returns, before its execution ends. + launch.at(-1)!, + event("session.execution.interrupted", { sessionID: CHILD }), + removeShell("sh_late"), + shellReport(CHILD, "sh_late", "error", "Shell.NotFoundError"), + reply("shell.remove", null), + drained, + ]); + const watch = yield* watchCalls(runtime, thread); + yield* runtime.startTurn(withLineage(thread)); + yield* watch.until(() => watch.terminals.length > 0); + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + yield* watch.until(() => watch.state.variant === "drained"); + assert.deepEqual(watch.state.roster, []); + }).pipe(Effect.scoped), + ); + + it.effect("removes a shell a held wake launched while Stop interrupted it", () => + Effect.gen(function* () { + const launch = shellLaunched(SESSION, "sh_late"); + const { runtime, thread } = yield* resumed([ + event("session.execution.started", { sessionID: SESSION }), + ...shellLaunched(SESSION, "sh_first"), + ...launch.slice(0, -1), + removeShell("sh_first"), + shellReport(SESSION, "sh_first", "error", "Shell.NotFoundError"), + reply("shell.remove", null), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + // The wake's shell starts after the interrupt returns, before its execution ends. + launch.at(-1)!, + event("session.execution.interrupted", { sessionID: SESSION }), + removeShell("sh_late"), + shellReport(SESSION, "sh_late", "error", "Shell.NotFoundError"), + reply("shell.remove", null), + drained, + ]).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: () => Effect.void, + take: Effect.never, + }), + ); + const watch = yield* watchCalls(runtime, thread); + yield* watch.until(() => watch.state.roster?.length === 1); + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + yield* watch.until(() => watch.state.variant === "drained"); + assert.deepEqual(watch.state.roster, []); + }).pipe(Effect.scoped), + ); + it.effect("stops a subagent launched while Stop removes a shell", () => Effect.gen(function* () { const { runtime, thread } = yield* resumed([ diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index c6a6a3468be5b..9a8d6bcb9a4dd 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -3469,11 +3469,16 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ); const childrenOf = (call: SubagentCall) => call.child === undefined ? announcedTo(call.state) : [call.child.sessionId]; - const interrupted = new Set(); - const unstopped = () => + // Keyed by session, or by wake since each wake is its own execution of the thread's session. + const interrupted = new Set(); + const unstopped = (wakes: boolean) => [ ...backgroundCalls().flatMap((call) => - childrenOf(call).map((sessionId) => ({ sessionId, caller: call.state })), + childrenOf(call).map((sessionId) => ({ + key: sessionId, + sessionId, + caller: call.state, + })), ), // Ended subagent calls can run again to answer their own background reports. ...sessionsOf(state) @@ -3483,14 +3488,24 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr busy.has(caller.sessionId) && !isOrchestrationV2WorkActive(caller.subagent.call.status), ) - .map((caller) => ({ sessionId: caller.sessionId, caller: undefined })), - ].filter(({ sessionId }) => !interrupted.has(sessionId)); + .map((caller) => ({ + key: caller.sessionId, + sessionId: caller.sessionId, + caller: undefined, + })), + ...(wakes ? state.wakes : []) + .filter((wake) => wake.running) + .map((wake) => ({ key: wake, sessionId: state.sessionId, caller: undefined })), + ].filter(({ key }) => !interrupted.has(key)); const unreached = new Set(); - const stopSessions = Effect.fnUntraced(function* () { - for (let sessions = unstopped(); sessions.length > 0; sessions = unstopped()) { - for (const { sessionId, caller } of sessions) { - if (interrupted.has(sessionId)) continue; - interrupted.add(sessionId); + // Succeeds with the sessions it interrupted. + const stopSessions = Effect.fnUntraced(function* (wakes: boolean) { + const stopped = new Set(); + for (let sessions = unstopped(wakes); sessions.length > 0; sessions = unstopped(wakes)) { + for (const { key, sessionId, caller } of sessions) { + if (interrupted.has(key)) continue; + interrupted.add(key); + stopped.add(sessionId); caller?.stopped.add(sessionId); const reached = yield* client.session .interrupt({ sessionID: Session.ID.make(sessionId) }) @@ -3498,7 +3513,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr Effect.catchTags({ SessionNotFoundError: () => Effect.void }), Effect.timeout(INTERRUPT_TIMEOUT), Effect.tapCause((cause) => - Effect.logWarning("Could not stop an OpenCode subagent.", cause), + Effect.logWarning("Could not stop an OpenCode session.", cause), ), Effect.exit, Effect.map(Exit.isSuccess), @@ -3506,7 +3521,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr if (!reached && caller !== undefined) unreached.add(sessionId); } } + return stopped; }); + // An interrupted execution can launch work until it ends; its end means all of it has been seen. + const awaitEnds = (sessionIds: Iterable) => + Effect.forEach(sessionIds, awaitIdle, { concurrency: "unbounded", discard: true }); const attempted = new Set(); const unremoved = () => sessionsOf(state) @@ -3528,32 +3547,39 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr attempted.add(id); caller.stopped.add(id); } - const reached = yield* Effect.filter(shells, ({ id, directory }) => - client.shell.remove({ id: Shell.ID.make(id), location: { directory } }).pipe( - // OpenCode declares no not-found error for remove; only the directory can be gone. - Effect.catchTags({ LocationNotFoundError: () => Effect.void }), - Effect.timeout(INTERRUPT_TIMEOUT), - Effect.tapCause((cause) => - Effect.logWarning("Could not stop an OpenCode background shell.", cause), + const reached = yield* Effect.filter( + shells, + ({ id, directory }) => + client.shell.remove({ id: Shell.ID.make(id), location: { directory } }).pipe( + // OpenCode declares no not-found error for remove; only the directory can be gone. + Effect.catchTags({ LocationNotFoundError: () => Effect.void }), + Effect.timeout(INTERRUPT_TIMEOUT), + Effect.tapCause((cause) => + Effect.logWarning("Could not stop an OpenCode background shell.", cause), + ), + Effect.exit, + Effect.map(Exit.isSuccess), ), - Effect.exit, - Effect.map(Exit.isSuccess), - ), + { concurrency: "unbounded" }, ); removed.push(...reached); } }); yield* removeShells(); - yield* stopSessions(); + // Wakes wait for the foreground stop, which may end the same execution of the thread's session. + const stopped = yield* stopSessions(false); const foreground = yield* Effect.exit(stopForeground); - if (Exit.isSuccess(foreground) && foreground.value) yield* awaitIdle(state.sessionId); - // Checked under the lock, so no shell launched before Stop settles outlives it. + if (Exit.isSuccess(foreground) && foreground.value) stopped.add(state.sessionId); + yield* awaitEnds(stopped); + // Checked under the lock, so no shell launched before Stop settles outlives it. A pass that does + // not settle found a shell or execution it had not stopped yet; both sets only grow, and stopped + // executions launch nothing after the end awaited here, so the loop ends. for (let settled = false; !settled;) { yield* removeShells(); - yield* stopSessions(); + yield* awaitEnds(yield* stopSessions(true)); settled = yield* lock.withPermit( Effect.gen(function* () { - if (unremoved().length > 0 || unstopped().length > 0) return false; + if (unremoved().length > 0 || unstopped(true).length > 0) return false; for (const call of backgroundCalls()) { if (childrenOf(call).some((childId) => unreached.has(childId))) continue; yield* settleCall(call, "interrupted"); @@ -3565,15 +3591,10 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr for (const caller of new Set(removed.map(({ caller }) => caller))) { yield* syncRosterAncestors(caller); } - const running = state.wakes.some((wake) => wake.running); + // Interrupted above but not yet ended: its end must not finish the thread's next turn. + if (state.wakes.some((wake) => wake.running)) markUnsettled(state); discardWakes(state, state.wakes.splice(0)); yield* syncRosterAncestors(state); - if (running) { - markUnsettled(state); - yield* client.session - .interrupt({ sessionID: Session.ID.make(state.sessionId) }) - .pipe(Effect.timeout(INTERRUPT_TIMEOUT), Effect.ignore({ log: true })); - } return true; }), ); From 4b557984448fd3d2e5bb506b3d9ff180e32f7748 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 13:39:25 +0000 Subject: [PATCH 11/14] fix(opencode): shells in held wakes survive neither Stop nor a reconnect A held autonomous wake owned the shells and subagents its unnamed sessions started only until replay, so a grandchild's events were dropped, a reconnect or a failed remove discarded running shells, and a held subagent's late launch slipped past Stop's last sweep. Held events now go to the nearest ancestor with state at any depth. Discarding a wake hands its running shells to the caller's tracked shells, and Stop deletes only those it removed. Stop interrupts and awaits the subagents a held wake announced in the same barrier as its other producers. Launches whose shellID is not a valid Shell.ID are ignored. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/OpenCode2AdapterV2.test.ts | 198 +++++++++++++++++- .../src/server/v2/adapter.ts | 78 +++++-- 2 files changed, 253 insertions(+), 23 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index d8597007206cc..3892337627c52 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -1238,6 +1238,9 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { out("session.interrupt", { sessionID: CHILD }), reply("session.interrupt", { interrupted: true }), event("session.execution.interrupted", { sessionID: CHILD }), + // Its wake is discarded once it ended; the stray pass interrupts what a removal restarted. + out("session.interrupt", { sessionID: CHILD }), + reply("session.interrupt", { interrupted: true }), drained, ], { replayGate: gate }, @@ -1257,6 +1260,188 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + const GRANDCHILD = "ses_f1485c529ffeGGGGGGGGGGGGGG"; + // The held subagent starts its own background subagent, whose shell arrives before either call is named. + const heldGrandchildShell = (): ReadonlyArray => { + const tool = { sessionID: CHILD, assistantMessageID: "msg_child", id: "call-nested" }; + return [ + event("session.execution.started", { sessionID: SESSION }), + ...backgroundLaunch(CHILD).slice(3), + returnedRunning("call-background"), + event("session.execution.started", { sessionID: CHILD }), + event("session.tool.input.started", { ...tool, name: "subagent" }), + event("session.tool.called", { + ...tool, + name: "subagent", + input: { description: "Nested", prompt: "sleep", background: true }, + executed: false, + }), + event("session.created", { ...childCreated(GRANDCHILD), parentID: CHILD }), + event("session.tool.progress", { + ...tool, + metadata: { sessionID: GRANDCHILD, status: "running" }, + }), + event("session.tool.success", { + ...tool, + content: [{ type: "text", text: "The subagent is working in the background." }], + metadata: { sessionID: GRANDCHILD, status: "running", truncated: false }, + executed: false, + }), + event("session.execution.started", { sessionID: GRANDCHILD }), + ...shellLaunched(GRANDCHILD, "sh_grandchild"), + event("session.execution.succeeded", { sessionID: SESSION }), + heldVariant("held"), + ]; + }; + + it.effect("lists a grandchild's shell from a held wake once the wake runs", () => + Effect.gen(function* () { + const offers: Array = []; + const gate = makeProviderReplayGate(["after"]); + const { runtime, thread } = yield* resumed( + [...heldGrandchildShell(), heldVariant("after", "after")], + { replayGate: gate }, + ).pipe(recordOffers(offers)); + const watch = yield* watchCalls(runtime, thread); + yield* watch.until(() => watch.state.variant === "held"); + yield* runtime.startTurn(continuationTurn(thread, offers[0]!)); + gate.release("after"); + yield* watch.until(() => watch.state.variant === "after"); + const [, , shell] = watch.state.roster ?? []; + assert.deepEqual(shell, { + taskId: "sh_grandchild", + kind: "command", + description: SHELL_COMMAND, + parentTaskId: `${CHILD}:call-nested`, + }); + }).pipe(Effect.scoped), + ); + + it.effect("keeps a held subagent's shell tracked when a reconnect discards its wake", () => + Effect.gen(function* () { + const offers: Array = []; + const { runtime, thread } = yield* resumed([ + ...heldSubagentShell(), + { type: "runtime_exit", status: "success" } as const, + out("event.subscribe"), + out("session.active"), + replyData("session.active", { [CHILD]: { type: "running" } }), + // A failed check keeps the shell, so this reply needs no full shell. + out("shell.get", { id: "sh_child", "location[directory]": WORK }), + reply("shell.get", { status: 500, body: { name: "UnknownError", data: {} } }), + out("session.interrupt", { sessionID: CHILD }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: CHILD }), + drained, + ]).pipe(recordOffers(offers)); + const watch = yield* watchCalls(runtime, thread); + yield* watch.until(() => watch.state.variant === "held"); + // The held wake keeps work pending until the reconnect discards it and lists what it kept. + yield* watch.until(() => watch.state.pending === false || watch.state.roster?.length === 1); + assert.deepEqual(watch.state.roster, [ + { taskId: "sh_child", kind: "command", description: SHELL_COMMAND }, + ]); + yield* watch.until(() => watch.state.variant === "drained"); + assert.isTrue(yield* runtime.hasPendingBackgroundWork!); + }).pipe(Effect.scoped), + ); + + it.effect("retries a held subagent's shell whose remove failed on the next Stop", () => + Effect.gen(function* () { + const offers: Array = []; + // Released up front: they only record which entries were emitted. + const gate = makeProviderReplayGate(["first", "second"]); + gate.release("first"); + gate.release("second"); + const { runtime, thread } = yield* resumed( + [ + ...heldSubagentShell(), + removeShell("sh_child"), + reply("shell.remove", { status: 500, body: { name: "UnknownError", data: {} } }), + out("session.interrupt", { sessionID: CHILD }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: CHILD }, "first"), + out("session.interrupt", { sessionID: CHILD }), + reply("session.interrupt", { interrupted: true }), + removeShell("sh_child"), + shellReport(CHILD, "sh_child", "error", "Shell.NotFoundError", "second"), + event("session.execution.started", { sessionID: CHILD }), + reply("shell.remove", null), + out("session.interrupt", { sessionID: CHILD }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: CHILD }), + drained, + ], + { replayGate: gate }, + ).pipe(recordOffers(offers)); + const watch = yield* watchCalls(runtime, thread); + yield* watch.until(() => watch.state.variant === "held"); + const stop = runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + yield* stop; + assert.isTrue(gate.hasReached("first")); + assert.isTrue(yield* runtime.hasPendingBackgroundWork!); + yield* watch.until(() => watch.state.roster?.length === 1); + assert.deepEqual(watch.state.roster, [ + { taskId: "sh_child", kind: "command", description: SHELL_COMMAND }, + ]); + yield* stop; + assert.isTrue(gate.hasReached("second")); + yield* watch.until(() => watch.state.variant === "drained"); + assert.deepEqual(watch.state.roster, []); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + }).pipe(Effect.scoped), + ); + + it.effect("removes a shell a held subagent launched while Stop interrupted it", () => + Effect.gen(function* () { + const launch = shellLaunched(CHILD, "sh_late"); + const offers: Array = []; + // Released up front: it only records that the remove's report was emitted. + const gate = makeProviderReplayGate(["removing"]); + gate.release("removing"); + const { runtime, thread } = yield* resumed( + [ + event("session.execution.started", { sessionID: SESSION }), + ...backgroundLaunch(CHILD).slice(3), + returnedRunning("call-background"), + event("session.execution.started", { sessionID: CHILD }), + ...launch.slice(0, -1), + event("session.execution.succeeded", { sessionID: SESSION }), + heldVariant("held"), + out("session.interrupt", { sessionID: CHILD }), + reply("session.interrupt", { interrupted: true }), + // The subagent's shell starts after the interrupt returns, before its execution ends. + launch.at(-1)!, + event("session.execution.interrupted", { sessionID: CHILD }), + removeShell("sh_late"), + shellReport(CHILD, "sh_late", "error", "Shell.NotFoundError", "removing"), + event("session.execution.started", { sessionID: CHILD }), + reply("shell.remove", null), + out("session.interrupt", { sessionID: CHILD }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: CHILD }), + drained, + ], + { replayGate: gate }, + ).pipe(recordOffers(offers)); + const watch = yield* watchCalls(runtime, thread); + yield* watch.until(() => watch.state.variant === "held"); + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + assert.isTrue(gate.hasReached("removing")); + yield* watch.until(() => watch.state.variant === "drained"); + assert.isFalse(watch.state.pending); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + }).pipe(Effect.scoped), + ); + it.effect("removes a shell launched while Stop removes another", () => Effect.gen(function* () { const offered = yield* Deferred.make(); @@ -5310,11 +5495,14 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { 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", { - status: 500, - body: { _tag: "UnknownError", message: "interrupt failed" }, - }), + // Stop interrupts the held wake's subagent, then retries it as a stray. + ...[1, 2].flatMap(() => [ + out("session.interrupt", { sessionID: SPAWNED }), + reply("session.interrupt", { + status: 500, + body: { _tag: "UnknownError", message: "interrupt failed" }, + }), + ]), out("session.prompt", { sessionID: SESSION, text: "" }), promptAccepted, event("session.execution.started", { sessionID: SESSION }), diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index 9a8d6bcb9a4dd..f117e8121e186 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -526,6 +526,13 @@ const reportOutcome = (state: string | undefined) => const shellCommandOf = (input: Readonly> | undefined) => stringField(input, "command")?.trim(); +const isShellId = Schema.is(Shell.ID); +// Any tool can report a shellID; only one OpenCode accepts back can be checked or removed. +const launchedShell = (metadata: Readonly> | undefined) => { + const shellId = metadata?.["status"] === "running" ? metadata["shellID"] : undefined; + return isShellId(shellId) ? shellId : undefined; +}; + const subagentOutput = (text: string) => /^]*>\n?([\s\S]*?)\n?<\/subagent>$/.exec(text.trim())?.[1] ?? text; @@ -2321,7 +2328,9 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const holderOf = (sessionId: string) => { const own = threads.get(sessionId); if (own !== undefined) return own.wakes.at(-1)?.running === true ? own : undefined; - const parent = announced.get(sessionId)?.parentID; + // A held subagent's own subagents have no state either; the nearest ancestor with state holds. + let parent = announced.get(sessionId)?.parentID; + while (parent !== undefined && !threads.has(parent)) parent = announced.get(parent)?.parentID; const holder = parent === undefined ? undefined : threads.get(parent); return holder !== undefined && holder.active === undefined && holder.wakes.length > 0 ? holder @@ -2351,8 +2360,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr metadata: Readonly> | undefined, input: Readonly> | undefined, ) => { - const shellId = stringField(metadata, "shellID"); - if (metadata?.["status"] !== "running" || shellId === undefined) return false; + const shellId = launchedShell(metadata); + if (shellId === undefined) return false; if (state.endedShells.delete(shellId) || state.shells.has(shellId)) return false; state.shells.set(shellId, { directory: state.directory, command: shellCommandOf(input) }); return true; @@ -2403,6 +2412,22 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr return false; }; + // A shell a discarded wake handed to its caller still reports its end to the session that launched it. + const releaseAdoptedShell = Effect.fnUntraced(function* ( + event: OpenCode2Client.OpenCode2StreamEvent, + ) { + if (event.type !== "session.inbox.enqueued" || event.data.item.type !== "synthetic") return; + const { metadata } = event.data.item.payload; + if (stringField(metadata, "source") !== "shell") return; + const shellId = stringField(metadata, "shellID") ?? ""; + for (const state of threads.values()) { + if (!state.shells.delete(shellId)) continue; + // The report wakes that session; Stop's stray pass ends what a removal started. + if (state.stopped.has(shellId)) strays.set(event.data.sessionID, state); + yield* syncRosterAncestors(state); + } + }); + const handleEvent = Effect.fnUntraced(function* (event: OpenCode2Client.OpenCode2StreamEvent) { const sessionId = sessionOfEvent(event); const cleared = sessionId === undefined ? undefined : clearing.get(sessionId); @@ -2461,6 +2486,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr held.running = false; } } else { + if (tracked === undefined) yield* releaseAdoptedShell(event); yield* route(event, sessionId); } if (shellsChanged) yield* syncRosterAncestors(tracked); @@ -3355,16 +3381,16 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr }); // Shells a held wake started in sessions it had not named yet; replay would count them. - const heldShells = (caller: ThreadState) => { - const running = new Set(); - for (const event of caller.wakes.flatMap((wake) => wake.events)) { + const heldShells = (wakes: ReadonlyArray) => { + const inputs = new Map>>(); + const running = new Map(); + for (const event of wakes.flatMap((wake) => wake.events)) { if (!untracked.has(event)) continue; - if ( - event.type === "session.tool.success" && - event.data.metadata?.["status"] === "running" - ) { - const shellId = stringField(event.data.metadata, "shellID"); - if (shellId !== undefined) running.add(shellId); + if (event.type === "session.tool.called") inputs.set(event.data.id, event.data.input); + if (event.type === "session.tool.success") { + const shellId = launchedShell(event.data.metadata); + if (shellId !== undefined) + running.set(shellId, shellCommandOf(inputs.get(event.data.id))); } if (event.type === "session.inbox.enqueued" && event.data.item.type === "synthetic") { const { metadata } = event.data.item.payload; @@ -3376,7 +3402,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr return running; }; + // Shells a dropped wake started keep running, so its caller tracks them until they end or are removed. const discardWakes = (caller: ThreadState, wakes: ReadonlyArray) => { + for (const [id, command] of heldShells(wakes)) { + if (!caller.shells.has(id)) caller.shells.set(id, { directory: caller.directory, command }); + } const retired = new Set(); const retire = (childId: string) => { const info = announced.get(childId); @@ -3496,6 +3526,14 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ...(wakes ? state.wakes : []) .filter((wake) => wake.running) .map((wake) => ({ key: wake, sessionId: state.sessionId, caller: undefined })), + // Subagents a held wake started have no call yet; their wake holds what they launch until they end. + ...(wakes ? state.wakes : []).flatMap((wake) => + [...wake.children].map((sessionId) => ({ + key: sessionId, + sessionId, + caller: undefined, + })), + ), ].filter(({ key }) => !interrupted.has(key)); const unreached = new Set(); // Succeeds with the sessions it interrupted. @@ -3535,7 +3573,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr id, directory: shell.directory, })), - ...[...heldShells(caller)].map((id) => ({ caller, id, directory: caller.directory })), + ...[...heldShells(caller.wakes).keys()].map((id) => ({ + caller, + id, + directory: caller.directory, + })), ]) .filter(({ id }) => !attempted.has(id)); const removed: Array<{ readonly caller: ThreadState; readonly id: string }> = []; @@ -3587,14 +3629,14 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr for (const caller of sessionsOf(state)) { for (const report of caller.reports.values()) caller.stopped.add(report.sourceId); } - for (const { caller, id } of removed) caller.shells.delete(id); - for (const caller of new Set(removed.map(({ caller }) => caller))) { - yield* syncRosterAncestors(caller); - } // Interrupted above but not yet ended: its end must not finish the thread's next turn. if (state.wakes.some((wake) => wake.running)) markUnsettled(state); + // Hands held shells to the thread first, so one whose remove failed stays tracked. discardWakes(state, state.wakes.splice(0)); - yield* syncRosterAncestors(state); + for (const { caller, id } of removed) caller.shells.delete(id); + for (const caller of new Set([state, ...removed.map(({ caller }) => caller)])) { + yield* syncRosterAncestors(caller); + } return true; }), ); From 8f0c659786b670edd623e277f3a269348ec7b08a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 15:05:30 +0000 Subject: [PATCH 12/14] fix(opencode): Stop keeps discarded subagents' shells tracked and stops the turn on time A discarded wake's subagent kept running past Stop's final stray interrupt or a reconnect, and shells it launched then were ignored, so Stop reported no pending work while they ran. Those sessions now stay owned by their caller, which tracks their shells after the interrupt. A held wake's shell whose end arrived before its launch result was adopted as running; held shells now keep end tombstones like normal tracking. A stalled shell removal or child interrupt delayed the foreground interrupt past the caller's 15-second deadline. Cleanup now gets a 3-second head start, after which the foreground interrupt is sent while cleanup continues. AI-assisted; verified with the tests and tools listed above. --- .../Adapters/OpenCode2AdapterV2.test.ts | 133 ++++++++++++++++++ .../src/server/v2/adapter.ts | 39 +++-- 2 files changed, 164 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 3892337627c52..65fe49ecdf6bd 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -1346,6 +1346,66 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + it.effect( + "tracks a shell a reconnect-discarded subagent launches after its kept shell ends", + () => + Effect.gen(function* () { + const offers: Array = []; + const { runtime, thread } = yield* resumed([ + ...heldSubagentShell(), + { type: "runtime_exit", status: "success" } as const, + out("event.subscribe"), + out("session.active"), + replyData("session.active", { [CHILD]: { type: "running" } }), + out("shell.get", { id: "sh_child", "location[directory]": WORK }), + reply("shell.get", { status: 500, body: { name: "UnknownError", data: {} } }), + out("session.interrupt", { sessionID: CHILD }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: CHILD }), + // The kept shell's report wakes the subagent, which launches another shell. + shellReport(CHILD, "sh_child", "completed", "DONE"), + event("session.execution.started", { sessionID: CHILD }), + ...shellLaunched(CHILD, "sh_followup"), + event("session.execution.succeeded", { sessionID: CHILD }), + drained, + ]).pipe(recordOffers(offers)); + const watch = yield* watchCalls(runtime, thread); + yield* watch.until(() => watch.state.variant === "drained"); + assert.deepEqual(watch.state.roster, [{ taskId: "sh_followup", kind: "command" }]); + assert.isTrue(yield* runtime.hasPendingBackgroundWork!); + }).pipe(Effect.scoped), + ); + + it.effect("keeps no held shell whose end arrived before its launch result", () => + Effect.gen(function* () { + const offers: Array = []; + const launch = shellLaunched(CHILD, "sh_child"); + const { runtime, thread } = yield* resumed([ + event("session.execution.started", { sessionID: SESSION }), + ...backgroundLaunch(CHILD).slice(3), + returnedRunning("call-background"), + event("session.execution.started", { sessionID: CHILD }), + ...launch.slice(0, -1), + shellReport(CHILD, "sh_child", "completed", "DONE"), + launch.at(-1)!, + event("session.execution.succeeded", { sessionID: CHILD }), + event("session.execution.succeeded", { sessionID: SESSION }), + heldVariant("held"), + { type: "runtime_exit", status: "success" } as const, + out("event.subscribe"), + out("session.active"), + replyData("session.active", {}), + out("session.interrupt", { sessionID: CHILD }), + reply("session.interrupt", { interrupted: false }), + drained, + ]).pipe(recordOffers(offers)); + const watch = yield* watchCalls(runtime, thread); + yield* watch.until(() => watch.state.variant === "drained"); + assert.deepEqual(watch.state.roster, []); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + }).pipe(Effect.scoped), + ); + it.effect("retries a held subagent's shell whose remove failed on the next Stop", () => Effect.gen(function* () { const offers: Array = []; @@ -1442,6 +1502,50 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + it.effect("tracks a shell a held subagent launches while Stop's last pass interrupts it", () => + Effect.gen(function* () { + const late = shellLaunched(CHILD, "sh_late"); + const offers: Array = []; + const { runtime, thread } = yield* resumed([ + event("session.execution.started", { sessionID: SESSION }), + ...backgroundLaunch(CHILD).slice(3), + returnedRunning("call-background"), + event("session.execution.started", { sessionID: CHILD }), + ...shellLaunched(CHILD, "sh_first"), + ...late.slice(0, -1), + event("session.execution.succeeded", { sessionID: SESSION }), + heldVariant("held"), + removeShell("sh_first"), + shellReport(CHILD, "sh_first", "error", "Shell.NotFoundError"), + reply("shell.remove", null), + out("session.interrupt", { sessionID: CHILD }), + reply("session.interrupt", { interrupted: true }), + late.at(-1)!, + event("session.execution.interrupted", { sessionID: CHILD }), + removeShell("sh_late"), + shellReport(CHILD, "sh_late", "error", "Shell.NotFoundError"), + event("session.execution.started", { sessionID: CHILD }), + reply("shell.remove", null), + out("session.interrupt", { sessionID: CHILD }), + reply("session.interrupt", { interrupted: true }), + // The removal woke the subagent again; it launches a shell before that execution ends. + ...shellLaunched(CHILD, "sh_escape"), + event("session.execution.interrupted", { sessionID: CHILD }), + drained, + ]).pipe(recordOffers(offers)); + const watch = yield* watchCalls(runtime, thread); + yield* watch.until(() => watch.state.variant === "held"); + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + yield* watch.until(() => watch.state.variant === "drained"); + assert.deepEqual(watch.state.roster, [{ taskId: "sh_escape", kind: "command" }]); + assert.isTrue(yield* runtime.hasPendingBackgroundWork!); + }).pipe(Effect.scoped), + ); + it.effect("removes a shell launched while Stop removes another", () => Effect.gen(function* () { const offered = yield* Deferred.make(); @@ -1643,6 +1747,35 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + it.effect("stops the turn within the outer Stop deadline while a shell removal stalls", () => + Effect.gen(function* () { + const { runtime, thread } = yield* resumed([ + ...backgroundShell("sh_1"), + removeShell("sh_1"), + reply("shell.remove", ""), + heldVariant("removing"), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + event("session.execution.interrupted", { sessionID: SESSION }), + ]); + const watch = yield* watchCalls(runtime, thread); + yield* runtime.startTurn(turnInput(thread)); + yield* watch.until(() => watch.state.roster?.length === 1); + yield* runtime + .interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }) + .pipe(Effect.forkScoped); + yield* watch.until(() => watch.state.variant === "removing"); + // Short of the removal's own 10-second timeout, which on its own would use up the deadline. + yield* TestClock.adjust("3 seconds"); + yield* watch.until(() => watch.terminals.length > 0); + assert.deepEqual(watch.terminals, ["interrupted"]); + }).pipe(Effect.scoped), + ); + it.effect("continues a wake that a stopped shell shares with a finished one", () => Effect.gen(function* () { const offers: Array = []; diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index f117e8121e186..92d2456edba08 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -46,6 +46,7 @@ import * as Digest from "effect/crypto/Digest"; import * as Effect from "effect/Effect"; import * as Hex from "effect/encoding/Hex"; import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Queue from "effect/Queue"; import * as Schedule from "effect/Schedule"; @@ -550,6 +551,7 @@ const claudeUsageLimit = (message: string | undefined) => : undefined; const INTERRUPT_TIMEOUT = "10 seconds"; +const FOREGROUND_STOP_DELAY = "3 seconds"; const INSTRUCTIONS_KEY = "t3-code"; const RECONNECT_ATTEMPTS = 5; const RECONNECT_DELAY = "2 seconds"; @@ -745,6 +747,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr /** Held events whose session had no state yet, so their shells were not counted on arrival. */ const untracked = new WeakSet(); const strays = new Map(); + /** Sessions of discarded wakes by the caller that tracks their shells, kept after their interrupt. */ + const adopted = new Map(); const busy = new Set(); const idleWaiters = new Map>(); const pending = new Map(); @@ -2412,20 +2416,30 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr return false; }; - // A shell a discarded wake handed to its caller still reports its end to the session that launched it. - const releaseAdoptedShell = Effect.fnUntraced(function* ( + // A discarded wake's sessions stay owned by its caller, which tracks the shells they launch. + const trackAdoptedShells = Effect.fnUntraced(function* ( event: OpenCode2Client.OpenCode2StreamEvent, ) { + const caller = adopted.get(sessionOfEvent(event) ?? ""); + if (event.type === "session.tool.success") { + if (caller !== undefined && trackShellLaunch(caller, event.data.metadata, undefined)) { + yield* syncRosterAncestors(caller); + } + return; + } if (event.type !== "session.inbox.enqueued" || event.data.item.type !== "synthetic") return; const { metadata } = event.data.item.payload; if (stringField(metadata, "source") !== "shell") return; const shellId = stringField(metadata, "shellID") ?? ""; + let released = false; for (const state of threads.values()) { if (!state.shells.delete(shellId)) continue; + released = true; // The report wakes that session; Stop's stray pass ends what a removal started. if (state.stopped.has(shellId)) strays.set(event.data.sessionID, state); yield* syncRosterAncestors(state); } + if (!released) caller?.endedShells.add(shellId); }); const handleEvent = Effect.fnUntraced(function* (event: OpenCode2Client.OpenCode2StreamEvent) { @@ -2464,6 +2478,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const strayCaller = strays.get(event.data.parentID); if (strayCaller !== undefined) { strays.set(event.data.sessionID, strayCaller); + adopted.set(event.data.sessionID, strayCaller); return; } const owner = ownerOf(event.data.parentID); @@ -2486,7 +2501,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr held.running = false; } } else { - if (tracked === undefined) yield* releaseAdoptedShell(event); + if (tracked === undefined) yield* trackAdoptedShells(event); yield* route(event, sessionId); } if (shellsChanged) yield* syncRosterAncestors(tracked); @@ -3384,18 +3399,21 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const heldShells = (wakes: ReadonlyArray) => { const inputs = new Map>>(); const running = new Map(); + // As in endedShells, an end can arrive before its launch result. + const ended = new Set(); for (const event of wakes.flatMap((wake) => wake.events)) { if (!untracked.has(event)) continue; if (event.type === "session.tool.called") inputs.set(event.data.id, event.data.input); if (event.type === "session.tool.success") { const shellId = launchedShell(event.data.metadata); - if (shellId !== undefined) + if (shellId !== undefined && !ended.delete(shellId)) running.set(shellId, shellCommandOf(inputs.get(event.data.id))); } if (event.type === "session.inbox.enqueued" && event.data.item.type === "synthetic") { const { metadata } = event.data.item.payload; - if (stringField(metadata, "source") === "shell") { - running.delete(stringField(metadata, "shellID") ?? ""); + const shellId = stringField(metadata, "shellID") ?? ""; + if (stringField(metadata, "source") === "shell" && !running.delete(shellId)) { + ended.add(shellId); } } } @@ -3414,6 +3432,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr announced.delete(childId); childOwners.delete(childId); strays.set(childId, caller); + adopted.set(childId, caller); retired.add(childId); threads.get(info.parentID ?? "")?.stopped.add(childId); }; @@ -3607,10 +3626,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr removed.push(...reached); } }); - yield* removeShells(); // Wakes wait for the foreground stop, which may end the same execution of the thread's session. - const stopped = yield* stopSessions(false); + const cleanup = yield* Effect.forkChild(Effect.andThen(removeShells(), stopSessions(false))); + // Cleanup normally goes first, but a stalled request must not hold the foreground stop past + // the caller's 15-second deadline. + yield* Fiber.await(cleanup).pipe(Effect.timeout(FOREGROUND_STOP_DELAY), Effect.ignore); const foreground = yield* Effect.exit(stopForeground); + const stopped = yield* Fiber.join(cleanup); if (Exit.isSuccess(foreground) && foreground.value) stopped.add(state.sessionId); yield* awaitEnds(stopped); // Checked under the lock, so no shell launched before Stop settles outlives it. A pass that does @@ -4189,6 +4211,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr childOwners.delete(child); threads.delete(child); } + for (const [child, caller] of adopted) if (rootOf(caller) === state) adopted.delete(child); if (state.mcp !== undefined) yield* removeMcp(state.mcp); }), respondToRuntimeRequest: Effect.fn("OpenCode2Adapter.respondToRuntimeRequest")( From dca900efbf60a471a0bd4aa0a923cfb0db4b2570 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 15:07:10 +0000 Subject: [PATCH 13/14] fix(opencode): Stop does not wait on a session that went idle while it registered AI-assisted; verified with the tests and tools listed above. --- packages/provider-opencode/src/server/v2/adapter.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index 92d2456edba08..1901965b0a13b 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -776,6 +776,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const awaitIdle = Effect.fnUntraced(function* (sessionId: string) { if (!busy.has(sessionId)) return; const waiter = idleWaiters.get(sessionId) ?? (yield* Deferred.make()); + if (!busy.has(sessionId)) return; idleWaiters.set(sessionId, waiter); yield* Deferred.await(waiter).pipe(Effect.timeout(INTERRUPT_TIMEOUT), Effect.ignore); }); From 153c326b70669131d2472b0c63aa31d3dbc6fcfe Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 15:20:15 +0000 Subject: [PATCH 14/14] fix(opencode): background shell Details show the command AI-assisted; verified with the tests and tools listed above. --- .../Adapters/OpenCode2AdapterV2.test.ts | 19 +++++++++++-------- .../src/server/v2/adapter.ts | 4 +++- 2 files changed, 14 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 65fe49ecdf6bd..b53493c0e1140 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -1005,7 +1005,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { ) .at(-1); assert.deepEqual(roster ?? [], [ - { taskId: "sh_1", kind: "command", description: SHELL_COMMAND }, + { taskId: "sh_1", kind: "command", command: SHELL_COMMAND, description: SHELL_COMMAND }, ]); assert.isTrue(yield* runtime.hasPendingBackgroundWork!); }).pipe(Effect.scoped), @@ -1028,7 +1028,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { yield* runtime.startTurn(turnInput(thread)); yield* watch.until(() => watch.state.variant === "drained"); assert.deepEqual(watch.state.roster, [ - { taskId: "sh_1", kind: "command", description: SHELL_COMMAND }, + { taskId: "sh_1", kind: "command", command: SHELL_COMMAND, description: SHELL_COMMAND }, ]); assert.isTrue(watch.state.pending); }).pipe(Effect.scoped), @@ -1046,7 +1046,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { yield* watch.until(() => watch.terminals.length > 0); assert.deepEqual(watch.terminals, ["completed"]); assert.deepEqual(watch.state.roster, [ - { taskId: "sh_1", kind: "command", description: SHELL_COMMAND }, + { taskId: "sh_1", kind: "command", command: SHELL_COMMAND, description: SHELL_COMMAND }, ]); yield* TestClock.adjust("1 minute"); yield* watch.until(() => watch.state.roster?.length === 0); @@ -1216,6 +1216,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { assert.deepEqual(shell, { taskId: "sh_child", kind: "command", + command: SHELL_COMMAND, description: SHELL_COMMAND, parentTaskId: "call-background", }); @@ -1311,6 +1312,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { assert.deepEqual(shell, { taskId: "sh_grandchild", kind: "command", + command: SHELL_COMMAND, description: SHELL_COMMAND, parentTaskId: `${CHILD}:call-nested`, }); @@ -1339,7 +1341,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { // The held wake keeps work pending until the reconnect discards it and lists what it kept. yield* watch.until(() => watch.state.pending === false || watch.state.roster?.length === 1); assert.deepEqual(watch.state.roster, [ - { taskId: "sh_child", kind: "command", description: SHELL_COMMAND }, + { taskId: "sh_child", kind: "command", command: SHELL_COMMAND, description: SHELL_COMMAND }, ]); yield* watch.until(() => watch.state.variant === "drained"); assert.isTrue(yield* runtime.hasPendingBackgroundWork!); @@ -1446,7 +1448,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { assert.isTrue(yield* runtime.hasPendingBackgroundWork!); yield* watch.until(() => watch.state.roster?.length === 1); assert.deepEqual(watch.state.roster, [ - { taskId: "sh_child", kind: "command", description: SHELL_COMMAND }, + { taskId: "sh_child", kind: "command", command: SHELL_COMMAND, description: SHELL_COMMAND }, ]); yield* stop; assert.isTrue(gate.hasReached("second")); @@ -1820,8 +1822,8 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { (watch.state.roster ?? []).every((task) => (task as { description?: string }).description), ); assert.deepEqual(watch.state.roster, [ - { taskId: "sh_a", kind: "command", description: SHELL_COMMAND }, - { taskId: "sh_b", kind: "command", description: SHELL_COMMAND }, + { taskId: "sh_a", kind: "command", command: SHELL_COMMAND, description: SHELL_COMMAND }, + { taskId: "sh_b", kind: "command", command: SHELL_COMMAND, description: SHELL_COMMAND }, ]); gate.release("ended"); yield* Deferred.await(continued); @@ -1850,7 +1852,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { assert.deepEqual(watch.terminals, ["completed"]); yield* watch.until(() => watch.calls.get("Sleep")?.status === "completed"); assert.deepEqual(watch.state.roster, [ - { taskId: "sh_child", kind: "command", description: SHELL_COMMAND }, + { taskId: "sh_child", kind: "command", command: SHELL_COMMAND, description: SHELL_COMMAND }, ]); const childTurns = watch.childTurns.size; yield* runtime.interruptTurn({ @@ -1894,6 +1896,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { assert.deepEqual(shell, { taskId: "sh_child", kind: "command", + command: SHELL_COMMAND, description: SHELL_COMMAND, parentTaskId: subagent?.taskId, }); diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index 1901965b0a13b..fbcdca16d3b49 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -1313,7 +1313,9 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr [...session.shells].map(([taskId, shell]) => ({ taskId, kind: "command" as const, - ...(shell.command === undefined ? {} : { description: shell.command }), + ...(shell.command === undefined + ? {} + : { command: shell.command, description: shell.command }), ...(session.subagent !== undefined && background.includes(session.subagent.call) ? { parentTaskId: session.subagent.call.nativeId } : {}),