diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index e4e0851e82c29..b53493c0e1140 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -903,6 +903,1006 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + const SHELL_COMMAND = "sleep 300; echo done"; + const shellLaunched = ( + sessionID: string, + shellID: string, + ): ReadonlyArray => { + const tool = { sessionID, assistantMessageID: "msg_assistant", id: `call-${shellID}` }; + return [ + 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: 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), + ]; + 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", + }, + }, + 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. + 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 }), + 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* () { + 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", command: SHELL_COMMAND, description: SHELL_COMMAND }, + ]); + assert.isTrue(yield* runtime.hasPendingBackgroundWork!); + }).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", command: SHELL_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([ + ...backgroundShell("sh_1"), + event("session.execution.succeeded", { sessionID: SESSION }), + { type: "runtime_exit", status: "success" } as const, + ]); + const watch = yield* watchCalls(runtime, thread); + yield* runtime.startTurn(turnInput(thread)); + yield* watch.until(() => watch.terminals.length > 0); + assert.deepEqual(watch.terminals, ["completed"]); + assert.deepEqual(watch.state.roster, [ + { taskId: "sh_1", kind: "command", command: SHELL_COMMAND, description: SHELL_COMMAND }, + ]); + yield* TestClock.adjust("1 minute"); + yield* watch.until(() => watch.state.roster?.length === 0); + assert.isFalse(watch.state.pending); + }).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); + yield* runtime.startTurn(turnInput(thread)); + yield* watch.until(() => watch.terminals.length > 0); + assert.deepEqual(watch.terminals, ["completed"]); + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + 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("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), + ); + + 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", + command: SHELL_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 }), + // 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 }, + ).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), + ); + + 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", + command: SHELL_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", command: SHELL_COMMAND, description: SHELL_COMMAND }, + ]); + yield* watch.until(() => watch.state.variant === "drained"); + assert.isTrue(yield* runtime.hasPendingBackgroundWork!); + }).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 = []; + // 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", command: SHELL_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("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(); + 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("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("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([ + ...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("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 = []; + 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", command: SHELL_COMMAND, description: SHELL_COMMAND }, + { taskId: "sh_b", kind: "command", command: SHELL_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([ + ...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); + yield* runtime.startTurn(withLineage(thread)); + 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", command: SHELL_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 { runtime, thread } = yield* resumed([ + ...backgroundLaunch(CHILD), + event("session.execution.started", { sessionID: CHILD }), + ...shellLaunched(CHILD, "sh_child"), + 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", + command: SHELL_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([ @@ -1096,7 +2096,12 @@ 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 }; + const terminals: Array = []; let wake = yield* Deferred.make(); yield* runtime.events.pipe( Stream.tap((event) => @@ -1107,8 +2112,12 @@ 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; + state.variant = + event.providerThread.nativeMetadata?.modelSelection?.options?.[0]?.value; } const id = event.type === "provider_turn.updated" @@ -1139,7 +2148,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), @@ -4622,11 +5631,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/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts index 46421aa5c5c86..308d956f04f92 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts @@ -78,6 +78,13 @@ 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 (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 3cbd4d893b6f1..fbcdca16d3b49 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, @@ -44,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"; @@ -307,12 +310,18 @@ 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; } >; - readonly stoppedChildren: Set; + /** Running background shells by id; a shell stays at its launch directory when the session moves. */ + 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. */ + readonly stopped: Set; readonly strandedSteers: Set; mcp: | { readonly name: string; readonly directory: string; readonly credential: string } @@ -515,6 +524,16 @@ const reportOutcome = (state: string | undefined) => ? ("cancelled" as const) : ("unknown" as const); +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; @@ -532,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"; @@ -724,8 +744,13 @@ 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(); + /** 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(); const lock = yield* Semaphore.make(1); const clearing = new Map>(); @@ -741,6 +766,20 @@ 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()); + if (!busy.has(sessionId)) return; + 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); @@ -769,7 +808,9 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr subagent, wakes: [], reports: new Map(), - stoppedChildren: new Set(), + shells: new Map(), + endedShells: new Set(), + stopped: new Set(), strandedSteers: new Set(), mcp: undefined, instructions: undefined, @@ -828,7 +869,8 @@ 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); @@ -1257,27 +1299,46 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr }); const syncRoster = Effect.fnUntraced(function* (state: ThreadState) { - const roster = 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 + 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 ? {} - : { childThreadId: call.child.subagent.appThread.id }), - })); + : { command: shell.command, description: shell.command }), + ...(session.subagent !== undefined && background.includes(session.subagent.call) + ? { 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,6 +2162,10 @@ 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); + // 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] : [])), @@ -2192,9 +2257,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 @@ -2249,7 +2313,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", @@ -2258,7 +2322,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" @@ -2271,7 +2335,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 @@ -2296,6 +2362,89 @@ 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 = 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; + }; + + const trackShellEnd = ( + state: ThreadState, + inboxId: string, + payload: { + readonly text: string; + readonly metadata?: Readonly> | undefined; + }, + ) => { + const shellId = stringField(payload.metadata, "shellID"); + if (stringField(payload.metadata, "source") !== "shell" || shellId === undefined) + return false; + const shell = state.shells.get(shellId); + 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: 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; + }; + + // 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) { const sessionId = sessionOfEvent(event); const cleared = sessionId === undefined ? undefined : clearing.get(sessionId); @@ -2323,13 +2472,16 @@ 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 ?? ""); + 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) { strays.set(event.data.sessionID, strayCaller); + adopted.set(event.data.sessionID, strayCaller); return; } const owner = ownerOf(event.data.parentID); @@ -2344,15 +2496,18 @@ 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)) ) { held.running = false; } - return; + } else { + if (tracked === undefined) yield* trackAdoptedShells(event); + yield* route(event, sessionId); } - return yield* route(event, sessionId); + if (shellsChanged) yield* syncRosterAncestors(tracked); }); const route = Effect.fnUntraced(function* ( @@ -2489,7 +2644,10 @@ 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(); } + for (const state of threads.values()) yield* syncRoster(state); yield* setSessionStatus("error", message); yield* Queue.end(events); }); @@ -2560,6 +2718,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) => @@ -2602,7 +2762,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; } @@ -2643,6 +2803,34 @@ 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. + 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"), + Effect.catchTags({ + 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), + ), + ), + 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); @@ -3201,10 +3389,45 @@ 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 = (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 && !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; + const shellId = stringField(metadata, "shellID") ?? ""; + if (stringField(metadata, "source") === "shell" && !running.delete(shellId)) { + ended.add(shellId); + } + } + } + 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); @@ -3212,8 +3435,9 @@ 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 ?? "")?.stoppedChildren.add(childId); + threads.get(info.parentID ?? "")?.stopped.add(childId); }; for (const wake of wakes) { wake.dropped = true; @@ -3283,16 +3507,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) => @@ -3300,57 +3521,153 @@ 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)), - ); + // 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) => ({ + key: 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) => ({ + key: caller.sessionId, + sessionId: caller.sessionId, + caller: undefined, + })), + ...(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(); - for (const [childId, caller] of children) { - caller.stoppedChildren.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* 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.stoppedChildren.add(report.childId); - } + // 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) }) + .pipe( + Effect.catchTags({ SessionNotFoundError: () => Effect.void }), + Effect.timeout(INTERRUPT_TIMEOUT), + Effect.tapCause((cause) => + Effect.logWarning("Could not stop an OpenCode session.", cause), + ), + Effect.exit, + Effect.map(Exit.isSuccess), + ); + if (!reached && caller !== undefined) unreached.add(sessionId); } - 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 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) + .flatMap((caller) => [ + ...[...caller.shells].map(([id, shell]) => ({ + caller, + id, + directory: shell.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 }> = []; + // 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), + ), + { concurrency: "unbounded" }, + ); + removed.push(...reached); + } + }); + // Wakes wait for the foreground stop, which may end the same execution of the thread's session. + 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 + // 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* awaitEnds(yield* stopSessions(true)); + settled = yield* lock.withPermit( + Effect.gen(function* () { + 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"); + } + for (const caller of sessionsOf(state)) { + for (const report of caller.reports.values()) caller.stopped.add(report.sourceId); + } + // 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)); + 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; + }), + ); + } yield* stopStrays(state); + yield* foreground; }); const listSkills = (location: { readonly directory: string }) => @@ -3821,42 +4138,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( @@ -3889,6 +4214,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")(