From 044ef53a5dce3a1cc9085b998658146ec10ab64d Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Tue, 10 Nov 2026 20:56:22 -0600 Subject: [PATCH 1/4] feat(opencode): track native shell jobs across turns - Recover owned background shells and retain their output - Expose shell jobs in pending task tracking and support stopping them --- .../Adapters/OpenCode2AdapterV2.test.ts | 208 ++++++++++- .../Adapters/OpenCode2AdapterV2.testkit.ts | 10 + .../src/server/v2/OpenCode2Client.ts | 25 ++ .../src/server/v2/adapter.ts | 341 +++++++++++++++++- 4 files changed, 570 insertions(+), 14 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index aae63afa9962e..0497b3641b7cf 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -142,6 +142,11 @@ const promptAccepted = replyData("session.prompt", { delivery: "steer", }); +const emptyShellInventory: ReadonlyArray = [ + out("shell.list", ""), + reply("shell.list", { location: { directory: WORK }, data: [] }), +]; + const noOpenRequests: ReadonlyArray = [ out("permission.list", { sessionID: SESSION }), replyData("permission.list", []), @@ -317,7 +322,8 @@ const colorForm = { const resumed = ( entries: ReadonlyArray, options?: { - readonly version?: "2.2.5"; + readonly version?: "2.2.1" | "2.2.5" | "2.2.6"; + readonly shellRecovery?: ReadonlyArray; readonly goal?: OrchestrationV2ProviderThread["goal"]; readonly external?: boolean; readonly supervised?: boolean; @@ -335,9 +341,16 @@ const resumed = ( out("session.get", { sessionID: SESSION }), replyData("session.get", sessionInfo()), ...noOpenRequests, - ...(options?.version === "2.2.5" + ...(options?.version === "2.2.5" || options?.version === "2.2.6" ? [out("rpc.call", ""), reply("rpc.call", { output: null })] : []), + ...(options?.shellRecovery ?? + (options?.version === undefined + ? [] + : [ + out("shell.list", ""), + reply("shell.list", { location: { directory: WORK }, data: [] }), + ])), ...withInstructions([ ...(options?.supervised === true ? [ @@ -884,6 +897,193 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { ); return { attached: Deferred.await(attached), terminal: Fiber.join(terminal) }; }); + it.effect.each([false, true, "thread"] as const)( + "tracks native background shells through launch, retained output, stop=%s", + (stop) => + Effect.gen(function* () { + const shellID = "sh_background"; + const tool = { + sessionID: SESSION, + assistantMessageID: "msg_assistant", + id: "call-shell", + }; + const gate = makeProviderReplayGate(["shell-exit"]); + const { runtime, thread } = yield* resumed( + [ + 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: "echo retained", background: true }, + executed: false, + }), + event( + "session.tool.success", + { + ...tool, + content: [{ type: "text", text: "Launched" }], + metadata: { shellID }, + executed: false, + }, + undefined, + 2, + ), + out("shell.output", ""), + reply("shell.output", { + location: { directory: WORK }, + data: { output: "retained\n", cursor: 9, size: 9, truncated: false }, + }), + event("session.execution.succeeded", { sessionID: SESSION }), + ...(stop + ? [] + : [event("shell.exited", { id: shellID, exit: 0, status: "exited" }, "shell-exit")]), + out("shell.output", ""), + reply("shell.output", { + location: { directory: WORK }, + data: { output: "", cursor: 9, size: 9, truncated: false }, + }), + ...(stop + ? [ + out("shell.remove", { id: shellID, "location[directory]": WORK }), + event("shell.exited", { id: shellID, exit: 143, status: "exited" }), + reply("shell.remove", null), + ] + : []), + ], + { replayGate: gate }, + ); + const items: Array = []; + const terminal = yield* runtime.events.pipe( + Stream.tap((e) => + Effect.sync(() => { + if (e.type === "turn_item.updated") items.push(e.turnItem); + }), + ), + Stream.filter((e) => e.type === "turn.terminal"), + Stream.runHead, + Effect.forkScoped, + ); + yield* runtime.startTurn(withLineage(thread)); + yield* Fiber.join(terminal); + yield* Effect.logWarning("shell terminal received"); + assert.isTrue(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + const settled = yield* runtime.events.pipe( + Stream.tap((e) => + Effect.sync(() => { + if (e.type === "turn_item.updated") items.push(e.turnItem); + }), + ), + Stream.filter( + (e) => + e.type === "provider_thread.updated" && + e.providerThread.pendingBackgroundTasks?.length === 0, + ), + Stream.runHead, + Effect.forkScoped, + ); + if (stop === "thread") + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + requestRuntimeRestart: true, + }); + else if (stop) yield* runtime.stopTask!({ providerThread: thread, taskId: shellID }); + else gate.release("shell-exit"); + yield* Effect.logWarning("shell settling"); + yield* Fiber.join(settled); + yield* Effect.logWarning("shell settled"); + assert.isFalse(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + assert.deepInclude(items.filter((i) => i.type === "command_execution").at(-1), { + type: "command_execution", + status: stop ? "interrupted" : "completed", + output: "retained\n", + }); + assert.isFalse(items.some((i) => i.type === "subagent")); + }).pipe(Effect.scoped), + ); + + it.effect.each(["shell", "job", "fallback", "descendant"] as const)( + "recovers only owned native shells using %s inventory", + (discovery) => + Effect.gen(function* () { + const id = "sh_recovered"; + const info = { + id, + command: "sleep 600", + status: "running", + cwd: WORK, + shell: "/bin/sh", + file: "/remote/output", + metadata: { sessionID: SESSION }, + time: { started: 1 }, + }; + const response = (data: unknown) => ({ location: { directory: WORK }, data }); + const inventory: Array = + discovery === "shell" + ? [] + : [ + out("job.list", { sessionID: SESSION, kind: "shell", descendants: "true" }), + reply( + "job.list", + discovery === "fallback" + ? { status: 404, body: {} } + : { + data: [ + { + kind: "shell", + shellID: id, + sessionID: discovery === "descendant" ? "ses_descendant" : SESSION, + command: info.command, + status: "running", + }, + ], + }, + ), + ]; + if (discovery === "shell" || discovery === "fallback") + inventory.push( + out("shell.list", ""), + reply( + "shell.list", + response([info, { ...info, id: "sh_other", metadata: { sessionID: "ses_other" } }]), + ), + ); + inventory.push( + out("shell.get", { id }), + reply("shell.get", response(info)), + out("shell.output", { id, cursor: "0" }), + reply( + "shell.output", + response({ output: "retained", cursor: 8, size: 8, truncated: false }), + ), + ); + const { runtime, thread } = yield* resumed( + [ + out("shell.output", { id, cursor: "8" }), + reply("shell.output", response({ output: "", cursor: 8, size: 8, truncated: false })), + out("shell.remove", { id }), + reply("shell.remove", null), + ], + { + version: discovery === "shell" ? "2.2.1" : "2.2.6", + shellRecovery: inventory, + external: true, + }, + ); + assert.deepEqual(thread.pendingBackgroundTasks, [ + { taskId: id, kind: "command", description: info.command }, + ]); + assert.isTrue(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + yield* runtime.stopTask!({ providerThread: thread, taskId: "sh_other" }); + assert.isTrue(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + yield* runtime.stopTask!({ providerThread: thread, taskId: id }); + assert.isFalse(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + }).pipe(Effect.scoped), + ); + it.effect("keeps background subagents running when a turn is interrupted to restart it", () => Effect.gen(function* () { const { runtime, thread } = yield* resumed([ @@ -2422,6 +2622,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { ...(delivery === "pending" ? [out("rpc.call", ""), reply("rpc.call", { output: { unreadable: true } })] : []), + ...(delivery === "pending" ? emptyShellInventory : []), ...reconnect.slice(2), ...(delivery === "pending" ? [ @@ -5480,6 +5681,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { input: { sessionID: SESSION }, }), reply("rpc.call", { output: nativeGoal("paused") }), + ...emptyShellInventory, ], { version: "2.2.5" }, ); @@ -5524,6 +5726,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { ...recovered.slice(0, 2), out("rpc.call", ""), reply("rpc.call", { output: nativeGoal("active") }), + ...emptyShellInventory, ...recovered.slice(2), ...noOpenRequests, ] @@ -5655,6 +5858,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { ), ] : []), + ...emptyShellInventory, ...reconnect.slice(2), ], { version: "2.2.5", replayGate: gate }, diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts index e4ad66e76539f..ef222ce42f849 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts @@ -60,6 +60,16 @@ const operationOf = ( type: "rpc.call", input: { rpcID: rpc[1], method: rpc[2], ...query, ...(body as object) }, }; + if (method === "GET" && path === "/api/experimental/job") + return { type: "job.list", input: query }; + const shell = /^\/api\/shell(?:\/([^/]+)(\/output)?)?$/.exec(path); + if (shell !== null) { + const input = { ...(shell[1] === undefined ? {} : { id: shell[1] }), ...query }; + return { + type: `shell.${method === "DELETE" ? "remove" : shell[2] ? "output" : shell[1] ? "get" : "list"}`, + input, + }; + } const session = /^\/api\/session\/([^/]+)(\/.*)?$/.exec(path); if (method === "GET" && path === "/api/event") return { type: "event.subscribe" }; if (method === "GET" && path === "/api/model") return { type: "model.list", input: query }; diff --git a/packages/provider-opencode/src/server/v2/OpenCode2Client.ts b/packages/provider-opencode/src/server/v2/OpenCode2Client.ts index 71fb1d70ce7d4..53095a70c3e94 100644 --- a/packages/provider-opencode/src/server/v2/OpenCode2Client.ts +++ b/packages/provider-opencode/src/server/v2/OpenCode2Client.ts @@ -50,9 +50,24 @@ export type OpenCode2StreamEvent = | OpenCode2UnreadableTerminal | OpenCode2UnreadableStart; +const ShellJobs = Schema.Struct({ + data: Schema.Array( + Schema.Struct({ + kind: Schema.Literal("shell"), + shellID: Schema.String, + sessionID: Schema.String, + command: Schema.String, + status: Schema.Literals(["running", "completed", "error", "cancelled"]), + }), + ), +}); + /** events resolves on subscription acceptance; subscribe before work because the stream is volatile. */ export interface OpenCode2Api { readonly client: OpenCodeClient; + readonly shellJobs?: ( + sessionID: string, + ) => Effect.Effect; readonly events: Effect.Effect< Stream.Stream< OpenCode2StreamEvent, @@ -212,6 +227,16 @@ export const make = Effect.gen(function* () { Effect.provideService(HttpClient.HttpClient, fillCompactionRecents(authenticated)), Effect.map((client) => ({ client, + shellJobs: Effect.fn("OpenCode2Client.shellJobs")(function (sessionID: string) { + return authenticated + .get(`${baseUrl}/api/experimental/job`, { + urlParams: { sessionID, kind: "shell", descendants: "true" }, + }) + .pipe( + Effect.flatMap(HttpClientResponse.filterStatusOk), + Effect.flatMap(HttpClientResponse.schemaBodyJson(ShellJobs)), + ); + }), events: readEvents.pipe( Effect.provideService( HttpClient.HttpClient, diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index bbd70cc108cdc..ed9c14d9d3da9 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -13,6 +13,7 @@ import { Skill, type OpenCodeEvent, } from "@opencode/client/effect"; +import * as Shell from "@opencode/schema/shell"; import { Mcp } from "@opencode/schema/mcp"; import { isOrchestrationV2WorkActive, @@ -869,7 +870,24 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr return state.active === undefined ? undefined : { state, turn: state.active }; }; + const shellInfo = new Map(); + const shells = new Map< + string, + { + state: ThreadState; + turn?: ActiveTurn; + toolId?: string; + tool?: { name: string; input: Record }; + command: string; + running: boolean; + stopping?: boolean; + output: string; + cursor: number; + } + >(); + const owesWork = (state: ThreadState) => + [...shells.values()].some((shell) => shell.state === state && shell.running) || state.wakes.length > 0 || state.reports.size > 0 || (state.subagent !== undefined && busy.has(state.sessionId)) || @@ -1140,6 +1158,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr status: "running" | "completed" | "failed" | "interrupted", result?: { readonly output: string | undefined; + readonly nativeTaskId?: string; readonly metadata: unknown; readonly files?: ReadonlyArray<{ readonly uri: string; readonly mime: string }>; }, @@ -1174,7 +1193,12 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr type: "turn_item.updated", driver, turnItem: openCodeToolTurnItem( - itemBase(state, turn, nativeId, settled, startedAt, completedAt, updatedAt), + { + ...itemBase(state, turn, nativeId, settled, startedAt, completedAt, updatedAt), + ...(result?.nativeTaskId === undefined + ? {} + : { nativeItemRef: ref(result.nativeTaskId) }), + }, { name: tool.name, input: tool.input, @@ -1415,16 +1439,29 @@ 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 - ? {} - : { childThreadId: call.child.subagent.appThread.id }), - })); + const roster: Array = + 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 }), + })); + for (const [taskId, shell] of shells) { + if (shell.running && sessionsOf(state).includes(shell.state)) { + roster.push({ + taskId, + kind: "command", + description: shell.command, + ...(shell.state.subagent === undefined + ? {} + : { parentTaskId: shell.state.subagent.call.nativeId }), + }); + } + } const current = state.providerThread.pendingBackgroundTasks ?? []; if ( state.publishedBackgroundWork === hasBackground(state) && @@ -1434,7 +1471,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr task.taskId === roster[index]?.taskId && task.description === roster[index]?.description && (task.kind === "subagent" ? task.childThreadId : undefined) === - roster[index]?.childThreadId, + (roster[index]?.kind === "subagent" ? roster[index]?.childThreadId : undefined), ) ) { return; @@ -1455,6 +1492,130 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr } }); + const readShellOutput = Effect.fn("OpenCode2Adapter.readShellOutput")(function* (id: string) { + const shell = shells.get(id); + if (shell === undefined) return; + // Bound each read and retained UI text; the native byte cursor must never use JS string lengths. + for (let page = 0; page < 4; page++) { + const { data } = yield* client.shell.output({ + id: Shell.ID.make(id), + location: { directory: shell.state.directory }, + cursor: shell.cursor, + limit: 65536, + }); + shell.output = (shell.output + data.output).slice(-262144); + const previous = shell.cursor; + shell.cursor = data.cursor; + if (data.cursor >= data.size || data.cursor <= previous) break; + } + }); + + const emitShell = Effect.fnUntraced(function* ( + id: string, + status: "running" | "completed" | "failed" | "interrupted", + exit?: number, + ) { + const shell = shells.get(id); + if (shell?.turn === undefined || shell.toolId === undefined || shell.tool === undefined) + return; + shell.turn.tools.set(shell.toolId, shell.tool); + yield* emitTool(shell.state, shell.turn, shell.toolId, status, { + output: shell.output, + nativeTaskId: id, + metadata: { shellID: id, exit }, + }); + shell.turn.tools.delete(shell.toolId); + }); + + const stopShell = Effect.fn("OpenCode2Adapter.stopShell")(function* (id: string) { + const shell = shells.get(id); + if (shell === undefined || !shell.running) return; + yield* readShellOutput(id).pipe(Effect.ignore({ log: true })); + shell.stopping = true; + yield* client.shell + .remove({ + id: Shell.ID.make(id), + location: { directory: shell.state.directory }, + }) + .pipe( + Effect.onError(() => + Effect.sync(() => { + shell.stopping = false; + }), + ), + ); + shell.stopping = false; + shell.running = false; + shell.state.stoppedChildren.add(id); + yield* emitShell(id, "interrupted"); + shells.delete(id); + shellInfo.delete(id); + yield* syncRosterAncestors(shell.state); + }); + + const recoverShells = Effect.fn("OpenCode2Adapter.recoverShells")(function* ( + state: ThreadState, + ) { + if (compareSemverVersions(connection.version, "2.2.1") < 0) return; + const jobs = + compareSemverVersions(connection.version, "2.2.6") >= 0 && + connection.shellJobs !== undefined + ? yield* connection.shellJobs(state.sessionId).pipe(Effect.option) + : undefined; + const data = + jobs?._tag === "Some" + ? jobs.value.data.map((job) => ({ + id: Shell.ID.make(job.shellID), + command: job.command, + metadata: { sessionID: job.sessionID }, + })) + : (yield* client.shell.list({ location: { directory: state.directory } })).data; + for (const info of data) { + const sessionId = stringField(info.metadata, "sessionID"); + const owner = + sessionId === undefined + ? undefined + : (threads.get(sessionId) ?? + ownerOf(sessionId) ?? + (jobs?._tag === "Some" ? state : undefined)); + if (owner === undefined || !sessionsOf(state).includes(owner)) continue; + if (!shells.has(info.id)) + shells.set(info.id, { + state: owner, + command: info.command, + running: true, + output: "", + cursor: 0, + }); + } + for (const [id, shell] of shells) { + if (!shell.running || !sessionsOf(state).includes(shell.state)) continue; + const found = yield* client.shell + .get({ id: Shell.ID.make(id), location: { directory: shell.state.directory } }) + .pipe(Effect.catchTags({ ShellNotFoundError: () => Effect.undefined })); + if (found === undefined) { + shell.running = false; + yield* emitShell(id, "interrupted"); + continue; + } + const info = found.data; + yield* readShellOutput(id); + shell.running = info.status === "running"; + yield* emitShell( + id, + shell.running + ? "running" + : info.status === "killed" + ? "interrupted" + : info.exit === 0 + ? "completed" + : "failed", + info.exit, + ); + } + yield* syncRosterAncestors(state); + }); + const withReportedModel = ( providerThread: OrchestrationV2ProviderThread, model: ModelRef | undefined, @@ -2260,6 +2421,44 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr } return yield* settleCall(call, "completed", subagentOutput(textOf(event.data.content))); } + const shellId = stringField(event.data.metadata, "shellID"); + const tool = turn.tools.get(event.data.id); + if ( + shellId !== undefined && + tool?.name === "shell" && + event.data.metadata?.["exit"] === undefined && + event.data.metadata?.["signal"] === undefined && + event.data.metadata?.["timeout"] !== true + ) { + shells.set(shellId, { + state, + turn, + toolId: event.data.id, + tool, + command: stringField(tool.input, "command") ?? "Shell command", + running: true, + output: "", + cursor: 0, + }); + yield* readShellOutput(shellId).pipe(Effect.ignore({ log: true })); + const known = shellInfo.get(shellId); + const tracked = shells.get(shellId)!; + tracked.running = known === undefined || known.status === "running"; + yield* emitShell( + shellId, + tracked.running + ? "running" + : known?.status === "killed" + ? "interrupted" + : known?.exit === 0 + ? "completed" + : "failed", + known?.exit, + ); + yield* syncRosterAncestors(state); + return; + } + if (shellId !== undefined) shellInfo.delete(shellId); const output = textOf(event.data.content); yield* emitTool(state, turn, event.data.id, "completed", { output, @@ -2394,9 +2593,43 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr inboxId: string, payload: { readonly text: string; + readonly description?: string | undefined; readonly metadata?: Readonly> | undefined; }, ) { + if (stringField(payload.metadata, "source") === "shell") { + const id = stringField(payload.metadata, "shellID"); + const shell = id === undefined ? undefined : shells.get(id); + if (shell !== undefined && shell.state !== state) return; + state.reports.set(inboxId, { + inboxId, + childId: id ?? inboxId, + text: payload.text, + report: { + kind: "command", + label: shell?.command ?? payload.description, + outcome: reportOutcome(stringField(payload.metadata, "state")), + }, + }); + if (id !== undefined && shell?.stopping !== true) { + if (shell?.running === true && shell.state === state) { + yield* readShellOutput(id).pipe(Effect.ignore({ log: true })); + shell.running = false; + const outcome = reportOutcome(stringField(payload.metadata, "state")); + yield* emitShell( + id, + outcome === "failed" + ? "failed" + : outcome === "cancelled" + ? "interrupted" + : "completed", + ); + } + shells.delete(id); + shellInfo.delete(id); + } + return; + } const childId = stringField(payload.metadata, "childID"); if (stringField(payload.metadata, "source") !== "subagent" || childId === undefined) return; strays.delete(childId); @@ -2457,6 +2690,41 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr : Effect.void; const handleEvent = Effect.fnUntraced(function* (event: OpenCode2Client.OpenCode2StreamEvent) { + if (event.type === "shell.created") { + const sessionId = stringField(event.data.info.metadata, "sessionID"); + if (sessionId !== undefined && ownerOf(sessionId) !== undefined) { + shellInfo.set(event.data.info.id, event.data.info); + if (shellInfo.size > 128) shellInfo.delete(shellInfo.keys().next().value!); + } + return; + } + if (event.type === "shell.exited" || event.type === "shell.deleted") { + const known = shellInfo.get(event.data.id); + if (known !== undefined) + shellInfo.set(event.data.id, { + ...known, + status: event.type === "shell.deleted" ? "killed" : event.data.status, + exit: "exit" in event.data ? event.data.exit : undefined, + }); + const shell = shells.get(event.data.id); + if (shell === undefined || !shell.running || shell.stopping === true) return; + if (event.type === "shell.exited") + yield* readShellOutput(event.data.id).pipe(Effect.ignore({ log: true })); + shell.running = false; + yield* emitShell( + event.data.id, + event.type === "shell.deleted" || event.data.status === "killed" + ? "interrupted" + : event.data.exit === 0 + ? "completed" + : "failed", + "exit" in event.data ? event.data.exit : undefined, + ); + shells.delete(event.data.id); + shellInfo.delete(event.data.id); + yield* syncRosterAncestors(shell.state); + return; + } const sessionId = sessionOfEvent(event); const cleared = sessionId === undefined ? undefined : clearing.get(sessionId); if (cleared !== undefined) { @@ -2909,6 +3177,12 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr { concurrency: 4, discard: true }, ).pipe(Effect.timeout(INVENTORY_TIMEOUT), Effect.ignore({ log: true })); } + for (const state of threads.values()) { + yield* recoverShells(state).pipe( + Effect.timeout(INVENTORY_TIMEOUT), + Effect.ignore({ log: true }), + ); + } yield* lock.withPermit(reconcile()).pipe(Effect.timeout(RECONCILE_TIMEOUT)); return stream; }).pipe( @@ -2970,6 +3244,22 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr // Subscribed before any session or prompt call, so no event of theirs is missed. yield* follow(yield* connection.events).pipe(Effect.forkScoped); + yield* lock + .withPermit( + Effect.gen(function* () { + for (const [id, shell] of shells) { + if (!shell.running) continue; + const before = shell.cursor; + yield* readShellOutput(id).pipe( + Effect.timeout(INVENTORY_TIMEOUT), + Effect.ignore({ log: true }), + ); + if (shell.cursor !== before) yield* emitShell(id, "running"); + } + }), + ) + .pipe(Effect.repeat(Schedule.spaced("2 seconds")), Effect.forkScoped); + // External servers outlive T3; stop their turns before losing observation. if (connection.external) { yield* Effect.addFinalizer(() => @@ -3508,6 +3798,9 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr }); const stopBackground = Effect.fnUntraced(function* (state: ThreadState) { + for (const [id, shell] of shells) { + if (shell.running && sessionsOf(state).includes(shell.state)) yield* stopShell(id); + } const calls = runningCalls(state).filter((call) => call.background); const callers = sessionsOf(state); // Ended subagent calls can run again to answer their own background reports. @@ -3789,6 +4082,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr Effect.ignore({ log: true }), ); } + if (!loaded) + yield* recoverShells(state).pipe( + Effect.timeout(INVENTORY_TIMEOUT), + Effect.ignore({ log: true }), + ); yield* writeRules(state, threadInput.runtimePolicy ?? state.policy); if (cwd != null && native.location.directory !== cwd) { yield* client.session.move({ @@ -4117,6 +4415,25 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ), ), ), + stopTask: Effect.fn("OpenCode2Adapter.stopTask")(function* ({ providerThread, taskId }) { + const shell = shells.get(taskId); + const state = + providerThread.nativeThreadRef == null + ? undefined + : threads.get(providerThread.nativeThreadRef.nativeId ?? ""); + if (state === undefined || shell === undefined || !sessionsOf(state).includes(shell.state)) + return; + yield* lock.withPermit(stopShell(taskId)).pipe( + Effect.mapError( + (cause) => + new ProviderAdapter.ProviderAdapterProtocolError({ + driver, + detail: "Could not stop the OpenCode shell.", + cause, + }), + ), + ); + }), unloadThread: Effect.fn("OpenCode2Adapter.unloadThread")(function* ({ providerThread }) { const nativeId = providerThread.nativeThreadRef?.nativeId; const state = nativeId == null ? undefined : threads.get(nativeId); From 70559ade73213b7b8533e4111834368980d74f89 Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Tue, 10 Nov 2026 21:39:28 -0600 Subject: [PATCH 2/4] fix(opencode): bind recovered shell activity to its owning run --- .../Adapters/OpenCode2AdapterV2.test.ts | 91 ++++++++++++++----- .../src/server/v2/adapter.ts | 61 +++++++------ 2 files changed, 100 insertions(+), 52 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index a46ff99633629..198269f697e79 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -49,6 +49,7 @@ import { t3McpServerName, } from "@t3tools/provider-opencode/testing"; import { resolveAttachmentPath } from "../../attachmentStore.ts"; +import { makeProviderEventRoutingState, routeProviderEvent } from "../RunExecutionService.ts"; import { openCode2ReplayRuntime } from "./OpenCode2AdapterV2.testkit.ts"; import { makeProviderReplayGate, @@ -1030,7 +1031,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); - it.effect.each(["shell", "job", "fallback", "descendant"] as const)( + it.effect.each(["shell", "job", "fallback", "descendant", "settled"] as const)( "recovers only owned native shells using %s inventory", (discovery) => Effect.gen(function* () { @@ -1038,7 +1039,8 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { const info = { id, command: "sleep 600", - status: "running", + status: discovery === "settled" ? "exited" : "running", + ...(discovery === "settled" ? { exit: 0 } : {}), cwd: WORK, shell: "/bin/sh", file: "/remote/output", @@ -1062,7 +1064,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { shellID: id, sessionID: discovery === "descendant" ? "ses_descendant" : SESSION, command: info.command, - status: "running", + status: discovery === "settled" ? "completed" : "running", }, ], }, @@ -1087,10 +1089,21 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { ); const { runtime, thread } = yield* resumed( [ - out("shell.output", { id, cursor: "8" }), - reply("shell.output", response({ output: "", cursor: 8, size: 8, truncated: false })), - out("shell.remove", { id }), - reply("shell.remove", null), + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + event("session.execution.started", { sessionID: SESSION }), + event("session.execution.succeeded", { sessionID: SESSION }), + ...(discovery === "settled" + ? [] + : [ + out("shell.output", { id, cursor: "8" }), + reply( + "shell.output", + response({ output: "", cursor: 8, size: 8, truncated: false }), + ), + out("shell.remove", { id }), + reply("shell.remove", null), + ]), ], { version: discovery === "shell" ? "2.2.1" : "2.2.6", @@ -1098,41 +1111,75 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { external: true, }, ); - const recovered = yield* runtime.events.pipe( - Stream.filter( - (e) => e.type === "turn_item.updated" && e.turnItem.type === "command_execution", - ), - Stream.takeUntil( - (e) => e.type === "turn_item.updated" && e.turnItem.status === "interrupted", + const items: Array = []; + const rootEnded = yield* Deferred.make(); + const shellEnded = yield* Deferred.make(); + yield* runtime.events.pipe( + Stream.runForEach((event) => + Effect.gen(function* () { + if (event.type === "turn.terminal") yield* Deferred.succeed(rootEnded, undefined); + if (event.type !== "turn_item.updated" || event.turnItem.type !== "command_execution") + return; + items.push(event); + if (event.turnItem.status !== "running") + yield* Deferred.succeed(shellEnded, undefined); + }), ), - Stream.runCollect, Effect.forkScoped, ); - assert.deepEqual(thread.pendingBackgroundTasks, [ - { taskId: id, kind: "command", description: info.command }, - ]); - assert.isTrue(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + assert.deepEqual( + thread.pendingBackgroundTasks ?? [], + discovery === "settled" + ? [] + : [{ taskId: id, kind: "command", description: info.command }], + ); + assert.equal( + yield* runtime.hasPendingBackgroundWorkForThread!(thread), + discovery !== "settled", + ); + yield* runtime.startTurn(withLineage(thread)); + yield* Deferred.await(rootEnded); yield* runtime.stopTask!({ providerThread: thread, taskId: "sh_other" }); - assert.isTrue(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + assert.equal( + yield* runtime.hasPendingBackgroundWorkForThread!(thread), + discovery !== "settled", + ); yield* runtime.stopTask!({ providerThread: thread, taskId: id }); assert.isFalse(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); - const item = yield* Fiber.join(recovered); + yield* Deferred.await(shellEnded); + const item = items; const projected = item.at(0); const stopped = item.at(-1); if (stopped?.type === "turn_item.updated") { - assert.equal(stopped.turnItem.status, "interrupted"); + assert.equal( + stopped.turnItem.status, + discovery === "settled" ? "completed" : "interrupted", + ); assert.equal( stopped.turnItem.id, projected?.type === "turn_item.updated" ? projected.turnItem.id : undefined, ); } assert.equal(projected?.type, "turn_item.updated"); - if (projected?.type === "turn_item.updated") + if (projected?.type === "turn_item.updated") { + const input = turnInput(thread); + const identity = { + threadId: input.threadId, + runId: input.runId, + attemptId: input.attemptId, + providerThreadId: thread.id, + }; + const routing = makeProviderEventRoutingState({ identity, providerTurnId: null }); + assert.isTrue( + item.every((event) => routeProviderEvent(event, identity, routing)[0]), + "recovered output and terminal state must pass server event routing", + ); assert.deepInclude(projected.turnItem, { type: "command_execution", input: info.command, output: "retained", }); + } }).pipe(Effect.scoped), ); diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index 7b71d603c77bd..46638c4dc14ed 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -904,6 +904,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr output: string; cursor: number; startedAt?: DateTime.Utc; + projectionState?: ThreadState; + settlement?: { status: "completed" | "failed" | "interrupted"; exit: number | undefined }; } >(); @@ -1589,35 +1591,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const shell = shells.get(id); if (shell === undefined) return; if (shell.turn === undefined || shell.toolId === undefined || shell.tool === undefined) { - const updatedAt = yield* DateTime.now; - yield* emit({ - type: "turn_item.updated", - driver, - turnItem: { - id: idAllocator.derive.turnItemFromProviderItem({ driver, nativeItemId: id }), - threadId: shell.state.providerThread.appThreadId ?? input.threadId, - runId: null, - nodeId: null, - providerThreadId: shell.state.providerThread.id, - providerTurnId: null, - nativeItemRef: ref(id), - parentItemId: null, - ordinal: 0, - status, - title: null, - startedAt: shell.startedAt ?? updatedAt, - completedAt: status === "running" ? null : updatedAt, - updatedAt, - type: "command_execution", - input: shell.command, - output: shell.output, - ...(exit === undefined ? {} : { exitCode: exit }), - }, - }); + if (status !== "running") shell.settlement = { status, exit }; return; } shell.turn.tools.set(shell.toolId, shell.tool); - yield* emitTool(shell.state, shell.turn, shell.toolId, status, { + yield* emitTool(shell.projectionState ?? shell.state, shell.turn, shell.toolId, status, { output: shell.output, nativeTaskId: id, metadata: { shellID: id, exit }, @@ -1625,6 +1603,23 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr shell.turn.tools.delete(shell.toolId); }); + const bindRecoveredShells = Effect.fn("OpenCode2Adapter.bindRecoveredShells")(function* ( + state: ThreadState, + turn: ActiveTurn, + ) { + for (const [id, shell] of shells) { + if (shell.turn !== undefined || !sessionsOf(state).includes(shell.state)) continue; + // Root command events need a real run owner to pass server routing. + shell.turn = turn; + shell.projectionState = state; + shell.toolId = id; + shell.tool = { name: "shell", input: { command: shell.command } }; + turn.startedAt.set(id, shell.startedAt ?? (yield* DateTime.now)); + yield* emitShell(id, shell.settlement?.status ?? "running", shell.settlement?.exit); + if (shell.settlement !== undefined) shells.delete(id); + } + }); + const stopShell = Effect.fn("OpenCode2Adapter.stopShell")(function* (id: string) { const shell = shells.get(id); if (shell === undefined || !shell.running) return; @@ -1647,7 +1642,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr shell.running = false; shell.state.stoppedChildren.add(id); yield* emitShell(id, "interrupted"); - shells.delete(id); + if (shell.turn !== undefined) shells.delete(id); shellInfo.delete(id); yield* syncRosterAncestors(shell.state); }); @@ -1730,7 +1725,12 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr lock.withPermit( effect.pipe( Effect.timeout(INVENTORY_TIMEOUT), - Effect.ensuring(syncRosterAncestors(state)), + Effect.ensuring( + Effect.gen(function* () { + if (state.active !== undefined) yield* bindRecoveredShells(state, state.active); + yield* syncRosterAncestors(state); + }), + ), ), ), ); @@ -2830,7 +2830,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr : "completed", ); } - shells.delete(id); + if (shell?.turn !== undefined) shells.delete(id); shellInfo.delete(id); } return; @@ -2928,7 +2928,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr : "failed", "exit" in event.data ? event.data.exit : undefined, ); - shells.delete(event.data.id); + if (shell.turn !== undefined) shells.delete(event.data.id); shellInfo.delete(event.data.id); yield* syncRosterAncestors(shell.state); return; @@ -3961,6 +3961,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr }; yield* emitProviderThread(state); yield* setSessionStatus("running", null); + yield* lock.withPermit(bindRecoveredShells(state, turn)); return turn; }); From dfc152bd4ac52f2ce6c042be28a8dd3fd6d7556f Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Tue, 10 Nov 2026 22:00:36 -0600 Subject: [PATCH 3/4] fix(opencode): persist recovered shell activity while idle --- .../Adapters/OpenCode2AdapterV2.test.ts | 323 ++++++++++-------- .../ProviderSessionManager.test.ts | 276 ++++++++++----- .../ProviderSessionManager.ts | 25 +- .../src/server/ProviderAdapter.ts | 4 + .../src/server/v2/adapter.ts | 185 ++++++---- 5 files changed, 523 insertions(+), 290 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 198269f697e79..75d7f2584fb84 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -49,7 +49,6 @@ import { t3McpServerName, } from "@t3tools/provider-opencode/testing"; import { resolveAttachmentPath } from "../../attachmentStore.ts"; -import { makeProviderEventRoutingState, routeProviderEvent } from "../RunExecutionService.ts"; import { openCode2ReplayRuntime } from "./OpenCode2AdapterV2.testkit.ts"; import { makeProviderReplayGate, @@ -325,6 +324,7 @@ const resumed = ( options?: { readonly version?: "2.2.1" | "2.2.5" | "2.2.6"; readonly shellRecovery?: ReadonlyArray; + readonly lastRunOrdinal?: number; readonly goal?: OrchestrationV2ProviderThread["goal"]; readonly external?: boolean; readonly supervised?: boolean; @@ -384,6 +384,9 @@ const resumed = ( const thread = yield* runtime.resumeThread({ providerThread: { ...providerThread(yield* DateTime.now), + ...(options?.lastRunOrdinal === undefined + ? {} + : { lastRunOrdinal: options.lastRunOrdinal }), ...(options?.goal === undefined ? {} : { goal: options.goal }), }, threadId, @@ -901,11 +904,11 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { ); return { attached: Deferred.await(attached), terminal: Fiber.join(terminal) }; }); - it.effect.each([false, true, "thread", "failed", "burst"] as const)( + it.effect.each([false, true, "thread", "failed", "burst", "poll"] as const)( "tracks native background shells through launch, retained output, stop=%s", (stop) => Effect.gen(function* () { - const stopped = stop === true || stop === "thread"; + const stopped = stop === true || stop === "thread" || stop === "poll"; const exitCode = stop === "failed" ? 7 : 0; const shellID = "sh_background"; const tool = { @@ -942,6 +945,9 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { location: { directory: WORK }, data: { output: "retained\n", cursor: 9, size: 9, truncated: false }, }), + ...(stop === "poll" + ? [out("shell.output", ""), reply("shell.output", "")] + : []), event("session.execution.succeeded", { sessionID: SESSION }), ...(stopped ? [] @@ -984,10 +990,14 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { { replayGate: gate }, ); const items: Array = []; + const shellStarted = yield* Deferred.make(); const terminal = yield* runtime.events.pipe( Stream.tap((e) => - Effect.sync(() => { - if (e.type === "turn_item.updated") items.push(e.turnItem); + Effect.gen(function* () { + if (e.type !== "turn_item.updated") return; + items.push(e.turnItem); + if (e.turnItem.type === "command_execution" && e.turnItem.output === "retained\n") + yield* Deferred.succeed(shellStarted, undefined); }), ), Stream.filter((e) => e.type === "turn.terminal"), @@ -995,6 +1005,10 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { Effect.forkScoped, ); yield* runtime.startTurn(withLineage(thread)); + if (stop === "poll") { + yield* Deferred.await(shellStarted); + yield* TestClock.adjust("2 seconds"); + } yield* Fiber.join(terminal); assert.isTrue(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); const settled = yield* runtime.events.pipe( @@ -1031,156 +1045,173 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); - it.effect.each(["shell", "job", "fallback", "descendant", "settled"] as const)( - "recovers only owned native shells using %s inventory", - (discovery) => - Effect.gen(function* () { - const id = "sh_recovered"; - const info = { - id, - command: "sleep 600", - status: discovery === "settled" ? "exited" : "running", - ...(discovery === "settled" ? { exit: 0 } : {}), - cwd: WORK, - shell: "/bin/sh", - file: "/remote/output", - metadata: { sessionID: SESSION }, - time: { started: 1 }, - }; - const response = (data: unknown) => ({ location: { directory: WORK }, data }); - const inventory: Array = - discovery === "shell" - ? [] - : [ - out("job.list", { sessionID: SESSION, kind: "shell", descendants: "true" }), - reply( - "job.list", - discovery === "fallback" - ? { status: 404, body: {} } - : { - data: [ - { - kind: "shell", - shellID: id, - sessionID: discovery === "descendant" ? "ses_descendant" : SESSION, - command: info.command, - status: discovery === "settled" ? "completed" : "running", - }, - ], - }, - ), - ]; - if (discovery === "shell" || discovery === "fallback") - inventory.push( - out("shell.list", ""), - reply( - "shell.list", - response([info, { ...info, id: "sh_other", metadata: { sessionID: "ses_other" } }]), - ), - ); + it.effect.each([ + "shell", + "job", + "fallback", + "descendant", + "settled", + "exited", + "anchored", + ] as const)("recovers only owned native shells using %s inventory", (discovery) => + Effect.gen(function* () { + const idAllocator = yield* IdAllocator.IdAllocatorV2.pipe(Effect.provide(IdAllocator.layer)); + const id = "sh_recovered"; + const info = { + id, + command: "sleep 600", + status: discovery === "settled" ? "exited" : "running", + ...(discovery === "settled" ? { exit: 0 } : {}), + cwd: WORK, + shell: "/bin/sh", + file: "/remote/output", + metadata: { sessionID: SESSION }, + time: { started: 1 }, + }; + const response = (data: unknown) => ({ location: { directory: WORK }, data }); + const inventory: Array = + discovery === "shell" + ? [] + : [ + out("job.list", { sessionID: SESSION, kind: "shell", descendants: "true" }), + reply( + "job.list", + discovery === "fallback" + ? { status: 404, body: {} } + : { + data: [ + { + kind: "shell", + shellID: id, + sessionID: discovery === "descendant" ? "ses_descendant" : SESSION, + command: info.command, + status: discovery === "settled" ? "completed" : "running", + }, + ], + }, + ), + ]; + if (discovery === "shell" || discovery === "fallback") inventory.push( - out("shell.get", { id }), - reply("shell.get", response(info)), - out("shell.output", { id, cursor: "0" }), + out("shell.list", ""), reply( - "shell.output", - response({ output: "retained", cursor: 8, size: 8, truncated: false }), + "shell.list", + response([info, { ...info, id: "sh_other", metadata: { sessionID: "ses_other" } }]), ), ); - const { runtime, thread } = yield* resumed( - [ - out("session.prompt", { sessionID: SESSION, text: "" }), - promptAccepted, - event("session.execution.started", { sessionID: SESSION }), - event("session.execution.succeeded", { sessionID: SESSION }), - ...(discovery === "settled" - ? [] - : [ - out("shell.output", { id, cursor: "8" }), - reply( - "shell.output", - response({ output: "", cursor: 8, size: 8, truncated: false }), - ), - out("shell.remove", { id }), - reply("shell.remove", null), - ]), - ], - { - version: discovery === "shell" ? "2.2.1" : "2.2.6", - shellRecovery: inventory, - external: true, - }, - ); - const items: Array = []; - const rootEnded = yield* Deferred.make(); - const shellEnded = yield* Deferred.make(); - yield* runtime.events.pipe( - Stream.runForEach((event) => - Effect.gen(function* () { - if (event.type === "turn.terminal") yield* Deferred.succeed(rootEnded, undefined); - if (event.type !== "turn_item.updated" || event.turnItem.type !== "command_execution") - return; - items.push(event); - if (event.turnItem.status !== "running") - yield* Deferred.succeed(shellEnded, undefined); - }), - ), - Effect.forkScoped, - ); - assert.deepEqual( - thread.pendingBackgroundTasks ?? [], - discovery === "settled" - ? [] - : [{ taskId: id, kind: "command", description: info.command }], + inventory.push( + out("shell.get", { id }), + reply("shell.get", response(info)), + out("shell.output", { id, cursor: "0" }), + reply( + "shell.output", + response({ output: "retained", cursor: 8, size: 8, truncated: false }), + ), + ); + const gate = makeProviderReplayGate(["recovered-exit"]); + const { runtime, thread } = yield* resumed( + discovery === "settled" + ? [] + : [ + ...(discovery === "exited" + ? [event("shell.exited", { id, exit: 0, status: "exited" }, "recovered-exit")] + : []), + out("shell.output", { id, cursor: "8" }), + reply( + "shell.output", + response({ + output: discovery === "exited" ? " finished" : "", + cursor: discovery === "exited" ? 17 : 8, + size: discovery === "exited" ? 17 : 8, + truncated: false, + }), + ), + ...(discovery === "exited" + ? [] + : [out("shell.remove", { id }), reply("shell.remove", null)]), + ], + { + version: discovery === "shell" ? "2.2.1" : "2.2.6", + shellRecovery: inventory, + external: true, + ...(discovery === "anchored" ? { lastRunOrdinal: 7 } : {}), + replayGate: gate, + }, + ); + const items: Array = []; + const shellEnded = yield* Deferred.make(); + const rosterEnded = yield* Deferred.make(); + yield* runtime.events.pipe( + Stream.runForEach((event) => + Effect.gen(function* () { + if ( + event.type === "provider_thread.updated" && + event.sessionScoped === true && + event.providerThread.pendingBackgroundTasks?.length === 0 + ) + yield* Deferred.succeed(rosterEnded, undefined); + if (event.type !== "turn_item.updated" || event.turnItem.type !== "command_execution") + return; + items.push(event); + if (event.turnItem.status !== "running") yield* Deferred.succeed(shellEnded, undefined); + }), + ), + Effect.forkScoped, + ); + assert.deepEqual( + thread.pendingBackgroundTasks ?? [], + discovery === "settled" ? [] : [{ taskId: id, kind: "command", description: info.command }], + ); + assert.equal( + yield* runtime.hasPendingBackgroundWorkForThread!(thread), + discovery !== "settled", + ); + yield* runtime.stopTask!({ providerThread: thread, taskId: "sh_other" }); + assert.equal( + yield* runtime.hasPendingBackgroundWorkForThread!(thread), + discovery !== "settled", + ); + if (discovery === "exited") gate.release("recovered-exit"); + else yield* runtime.stopTask!({ providerThread: thread, taskId: id }); + yield* Deferred.await(shellEnded); + yield* Deferred.await(rosterEnded); + assert.isFalse(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + const item = items; + const projected = item.at(0); + const stopped = item.at(-1); + if (stopped?.type === "turn_item.updated" && stopped.turnItem.type === "command_execution") { + assert.equal( + stopped.turnItem.status, + discovery === "settled" || discovery === "exited" ? "completed" : "interrupted", ); assert.equal( - yield* runtime.hasPendingBackgroundWorkForThread!(thread), - discovery !== "settled", + stopped.turnItem.output, + discovery === "exited" ? "retained finished" : "retained", ); - yield* runtime.startTurn(withLineage(thread)); - yield* Deferred.await(rootEnded); - yield* runtime.stopTask!({ providerThread: thread, taskId: "sh_other" }); assert.equal( - yield* runtime.hasPendingBackgroundWorkForThread!(thread), - discovery !== "settled", + stopped.turnItem.id, + projected?.type === "turn_item.updated" ? projected.turnItem.id : undefined, ); - yield* runtime.stopTask!({ providerThread: thread, taskId: id }); - assert.isFalse(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); - yield* Deferred.await(shellEnded); - const item = items; - const projected = item.at(0); - const stopped = item.at(-1); - if (stopped?.type === "turn_item.updated") { - assert.equal( - stopped.turnItem.status, - discovery === "settled" ? "completed" : "interrupted", - ); - assert.equal( - stopped.turnItem.id, - projected?.type === "turn_item.updated" ? projected.turnItem.id : undefined, - ); - } - assert.equal(projected?.type, "turn_item.updated"); - if (projected?.type === "turn_item.updated") { - const input = turnInput(thread); - const identity = { - threadId: input.threadId, - runId: input.runId, - attemptId: input.attemptId, - providerThreadId: thread.id, - }; - const routing = makeProviderEventRoutingState({ identity, providerTurnId: null }); - assert.isTrue( - item.every((event) => routeProviderEvent(event, identity, routing)[0]), - "recovered output and terminal state must pass server event routing", - ); - assert.deepInclude(projected.turnItem, { - type: "command_execution", - input: info.command, - output: "retained", - }); - } - }).pipe(Effect.scoped), + } + assert.equal(projected?.type, "turn_item.updated"); + if (projected?.type === "turn_item.updated") { + assert.isTrue( + item.every( + (event) => + event.type === "turn_item.updated" && + event.turnItem.runId === + (discovery === "anchored" + ? idAllocator.derive.run({ threadId, ordinal: 7 }) + : null), + ), + ); + assert.deepInclude(projected.turnItem, { + type: "command_execution", + input: info.command, + output: "retained", + }); + } + }).pipe(Effect.scoped), ); it.effect.each(["remove", "lookup", "output"] as const)( diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index b4fc69727c1de..25ae18f7b6690 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -3660,89 +3660,205 @@ it.effect( }), ); -it.effect("ProviderSessionManagerV2 persists session-scoped runtime requests without a run", () => - Effect.gen(function* () { - const state = yield* Ref.make(emptyState); - const effect = Effect.gen(function* () { - const eventSink = yield* EventSink.EventSinkV2; - const idAllocator = yield* IdAllocator.IdAllocatorV2; - const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; - const projectionStore = yield* ProjectionStore.ProjectionStoreV2; - const now = yield* DateTime.now; - const projectId = yield* idAllocator.allocate.project({ - fixtureName: "provider-session-manager-session-request", - }); - const threadId = yield* idAllocator.allocate.thread({ - fixtureName: "provider-session-manager-session-request", - projectId, - }); - const providerSessionId = yield* idAllocator.allocate.providerSession({ - providerInstanceId: modelSelection.instanceId, - threadId, - }); - const providerThread = makeProviderThread({ - idAllocator, - threadId, - providerSessionId, - now, - }); +it.effect( + "ProviderSessionManagerV2 persists session-scoped requests and recovered commands without a run", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const effect = Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const projectId = yield* idAllocator.allocate.project({ + fixtureName: "provider-session-manager-session-request", + }); + const threadId = yield* idAllocator.allocate.thread({ + fixtureName: "provider-session-manager-session-request", + projectId, + }); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const providerThread = makeProviderThread({ + idAllocator, + threadId, + providerSessionId, + now, + }); - yield* eventSink.write({ - events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], - }); - yield* manager.open({ - threadId, - providerSessionId, - modelSelection, - runtimePolicy, - }); - const pendingRequest = yield* makePendingRuntimeRequestEvents({ - idAllocator, - threadId, - providerSessionId, - providerThread, - now, - }); - const afterSequence = yield* eventSink.latestSequence({ threadId }); - const persistedFiber = yield* eventSink.stream({ threadId, afterSequence }).pipe( - Stream.filter( - (stored) => - stored.event.type === "runtime-request.updated" || - stored.event.type === "node.updated" || - stored.event.type === "turn-item.updated", - ), - Stream.take(3), - Stream.runCollect, - Effect.forkScoped, - ); - const adapterEvents = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); - assert.isDefined(adapterEvents); - yield* Queue.offerAll(adapterEvents!, pendingRequest.providerEvents); - const persisted = Array.from(yield* Fiber.join(persistedFiber)); + const otherThreadId = yield* idAllocator.allocate.thread({ + fixtureName: "unattached-shell-owner", + projectId, + }); + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator, threadId, now }), + yield* makeThreadCreatedEvent({ idAllocator, threadId: otherThreadId, now }), + ], + }); + yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + const pendingRequest = yield* makePendingRuntimeRequestEvents({ + idAllocator, + threadId, + providerSessionId, + providerThread, + now, + }); + const shellId = "sh_recovered"; + const command = { + id: idAllocator.derive.turnItemFromProviderItem({ + driver: CODEX_DRIVER, + nativeItemId: shellId, + }), + threadId, + runId: null, + nodeId: null, + providerThreadId: providerThread.id, + providerTurnId: null, + nativeItemRef: { driver: CODEX_DRIVER, nativeId: shellId, strength: "strong" as const }, + parentItemId: null, + ordinal: 0, + status: "running" as const, + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + type: "command_execution" as const, + input: "echo retained", + output: "retained", + }; + const anchored = { + ...command, + id: idAllocator.derive.turnItemFromProviderItem({ + driver: CODEX_DRIVER, + nativeItemId: "sh_anchored", + }), + runId: idAllocator.derive.run({ threadId, ordinal: 1 }), + }; + const commandEvents = [ + { + type: "turn_item.updated" as const, + driver: CODEX_DRIVER, + turnItem: anchored, + sessionScoped: true, + }, + { + type: "turn_item.updated" as const, + driver: CODEX_DRIVER, + turnItem: command, + sessionScoped: true, + }, + { + type: "provider_thread.updated" as const, + driver: CODEX_DRIVER, + providerThread: { + ...providerThread, + pendingBackgroundTasks: [{ taskId: shellId, kind: "command" as const }], + }, + sessionScoped: true, + }, + { + type: "turn_item.updated" as const, + driver: CODEX_DRIVER, + sessionScoped: true, + turnItem: { + ...command, + status: "completed" as const, + output: "retained\nfinished", + completedAt: now, + }, + }, + { + type: "provider_thread.updated" as const, + driver: CODEX_DRIVER, + providerThread: { ...providerThread, pendingBackgroundTasks: [] }, + sessionScoped: true, + }, + ]; + const afterSequence = yield* eventSink.latestSequence({ threadId }); + const persistedFiber = yield* eventSink.stream({ threadId, afterSequence }).pipe( + Stream.filter( + (stored) => + stored.event.type === "runtime-request.updated" || + stored.event.type === "node.updated" || + (stored.event.type === "turn-item.updated" && + stored.event.payload.type === "approval_request"), + ), + Stream.take(3), + Stream.runCollect, + Effect.forkScoped, + ); + const adapterEvents = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); + assert.isDefined(adapterEvents); + yield* Queue.offerAll(adapterEvents!, [ + ...commandEvents, + { + type: "turn_item.updated", + driver: CODEX_DRIVER, + sessionScoped: true, + turnItem: { ...command, threadId: otherThreadId }, + }, + { + type: "provider_thread.updated", + driver: CODEX_DRIVER, + providerThread: { ...providerThread, appThreadId: otherThreadId }, + sessionScoped: true, + }, + ...pendingRequest.providerEvents, + ]); + const persisted = Array.from(yield* Fiber.join(persistedFiber)); - assert.sameMembers( - persisted.map((stored) => stored.event.type), - ["runtime-request.updated", "node.updated", "turn-item.updated"], - ); - const projection = yield* projectionStore.getThreadProjection(threadId); - const request = projection.runtimeRequests.find( - (candidate) => candidate.id === pendingRequest.requestId, - ); - const node = projection.nodes.find((candidate) => candidate.id === pendingRequest.nodeId); - const turnItem = projection.turnItems.find( - (candidate) => - candidate.type === "approval_request" && candidate.requestId === pendingRequest.requestId, - ); - assert.equal(request?.status, "pending"); - assert.equal(request?.providerTurnId, null); - assert.equal(node?.runId, null); - assert.equal(node?.status, "waiting"); - assert.equal(turnItem?.runId, null); - assert.equal(turnItem?.status, "waiting"); - }); + assert.sameMembers( + persisted.map((stored) => stored.event.type), + ["runtime-request.updated", "node.updated", "turn-item.updated"], + ); + const projection = yield* projectionStore.getThreadProjection(threadId); + const request = projection.runtimeRequests.find( + (candidate) => candidate.id === pendingRequest.requestId, + ); + const node = projection.nodes.find((candidate) => candidate.id === pendingRequest.nodeId); + const turnItem = projection.turnItems.find( + (candidate) => + candidate.type === "approval_request" && + candidate.requestId === pendingRequest.requestId, + ); + const recovered = projection.turnItems.find((item) => item.id === command.id); + assert.deepInclude(recovered, { + runId: null, + status: "completed", + output: "retained\nfinished", + }); + assert.deepEqual( + projection.providerThreads.find((item) => item.id === providerThread.id) + ?.pendingBackgroundTasks, + [], + ); + assert.deepInclude( + projection.turnItems.find((item) => item.id === anchored.id), + { runId: anchored.runId, output: "retained" }, + ); + assert.isEmpty(projection.runs); + const unrelated = yield* projectionStore.getThreadProjection(otherThreadId); + assert.isEmpty(unrelated.turnItems); + assert.isEmpty(unrelated.providerThreads); + assert.equal(request?.status, "pending"); + assert.equal(request?.providerTurnId, null); + assert.equal(node?.runId, null); + assert.equal(node?.status, "waiting"); + assert.equal(turnItem?.runId, null); + assert.equal(turnItem?.status, "waiting"); + }); - yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 }))); - }), + yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 }))); + }), ); it.effect( diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 0fb31e81a0732..84a64b10c81f2 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -281,10 +281,14 @@ function sessionKey(providerSessionId: ProviderSessionId): string { return String(providerSessionId); } -function sessionScopedRuntimeRequestThreadId( +function sessionScopedArtifactThreadId( event: ProviderAdapter.ProviderAdapterV2Event, ): ThreadId | undefined { switch (event.type) { + case "provider_thread.updated": + return event.sessionScoped === true + ? (event.providerThread.appThreadId ?? undefined) + : undefined; case "runtime_request.updated": return event.runtimeRequest.providerTurnId === null ? event.threadId : undefined; case "node.updated": @@ -292,6 +296,8 @@ function sessionScopedRuntimeRequestThreadId( ? event.node.threadId : undefined; case "turn_item.updated": + if (event.turnItem.type === "command_execution") + return event.sessionScoped === true ? event.turnItem.threadId : undefined; return event.turnItem.runId === null && (event.turnItem.type === "approval_request" || event.turnItem.type === "user_input_request") ? event.turnItem.threadId @@ -2274,17 +2280,21 @@ export const layerWithOptions = ( }); }); } - // Some providers can block before a run subscriber exists - // (project trust, login, or session-switch hooks). Persist - // their runless request artifacts directly so the normal T3 - // request UI can answer them and unblock session setup. - const threadId = sessionScopedRuntimeRequestThreadId(event); + // Requests and recovered background commands can arrive before a run subscribes. + const threadId = sessionScopedArtifactThreadId(event); if (threadId !== undefined) { yield* Effect.gen(function* () { const current = (yield* Ref.get(sessions)).get( sessionKey(entry.runtime.providerSessionId), ); - if (current?.runtime !== entry.runtime) return; + if ( + current?.runtime !== entry.runtime || + ((event.type === "provider_thread.updated" || + (event.type === "turn_item.updated" && + event.turnItem.type === "command_execution")) && + !current.attachedThreadIds.has(threadId)) + ) + return; yield* providerEventIngestor .ingestNormalized({ providerSessionId: entry.runtime.providerSessionId, @@ -2303,6 +2313,7 @@ export const layerWithOptions = ( ), ); }).pipe(entry.requestEventPermit.withPermits(1)); + yield* releaseLiveOutput; return; } yield* publishToSubscribers(entry.eventSubscribers, { type: "event", event }); diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 0c43deb4f7804..e925f10c8f154 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -98,6 +98,8 @@ export const ProviderAdapterV2Event = Schema.Union([ providerThread: OrchestrationV2ProviderThread, /** Work remaining at this point in the event stream, including native continuations. */ pendingBackgroundWork: Schema.optional(Schema.Boolean), + /** Persist idle background snapshots before a run subscriber exists. */ + sessionScoped: Schema.optional(Schema.Boolean), retainedNativeTurnIds: Schema.optional(Schema.Array(Schema.String)), }), Schema.Struct({ @@ -125,6 +127,8 @@ export const ProviderAdapterV2Event = Schema.Union([ type: Schema.Literal("turn_item.updated"), driver: ProviderDriverKind, turnItem: OrchestrationV2TurnItem, + /** Persist recovered activity independently of a run subscription. */ + sessionScoped: Schema.optional(Schema.Boolean), }), /** A steer the provider queued stays `pending` until the agent reads it; `cancelled` if dropped unread. */ Schema.Struct({ diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index 46638c4dc14ed..e053347e7519a 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -308,6 +308,7 @@ interface ThreadState { readonly sessionId: string; providerThread: OrchestrationV2ProviderThread; publishedBackgroundWork?: boolean; + sessionScopedBackground?: boolean; readonly imageJobs: Map< string, { @@ -904,8 +905,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr output: string; cursor: number; startedAt?: DateTime.Utc; - projectionState?: ThreadState; - settlement?: { status: "completed" | "failed" | "interrupted"; exit: number | undefined }; + recoveredRunId?: RunId | null; } >(); @@ -927,6 +927,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr driver, providerThread: state.providerThread, pendingBackgroundWork, + ...(state.sessionScopedBackground === true ? { sessionScoped: true } : {}), }); }); @@ -1558,31 +1559,50 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr } }); - const readShellOutput = Effect.fn("OpenCode2Adapter.readShellOutput")(function* ( + const fetchShellOutput = Effect.fn("OpenCode2Adapter.fetchShellOutput")(function* ( id: string, + snapshot: { directory: string; cursor: number; output: string }, settled = false, ) { - const shell = shells.get(id); - if (shell === undefined) return; + let { cursor, output } = snapshot; // Bound each read and retained UI text; the native byte cursor must never use JS string lengths. for (let page = 0; page < (settled ? 5 : 4); page++) { const { data } = yield* client.shell.output({ id: Shell.ID.make(id), - location: { directory: shell.state.directory }, - cursor: shell.cursor, + location: { directory: snapshot.directory }, + cursor, limit: 65536, }); - shell.output = (shell.output + data.output).slice(-262144); - const previous = shell.cursor; - shell.cursor = data.cursor; + output = (output + data.output).slice(-262144); + const previous = cursor; + cursor = data.cursor; if (data.cursor >= data.size || data.cursor <= previous) break; if (settled && data.size - data.cursor > 262144) { - shell.cursor = data.size - 262144; - shell.output = ""; + cursor = data.size - 262144; + output = ""; } } + return { cursor, output }; }, Effect.timeout(INVENTORY_TIMEOUT)); + const readShellOutput = Effect.fn("OpenCode2Adapter.readShellOutput")(function* ( + id: string, + settled = false, + ) { + const shell = shells.get(id); + if (shell === undefined) return; + const output = yield* fetchShellOutput( + id, + { + directory: shell.state.directory, + cursor: shell.cursor, + output: shell.output, + }, + settled, + ); + Object.assign(shell, output); + }); + const emitShell = Effect.fnUntraced(function* ( id: string, status: "running" | "completed" | "failed" | "interrupted", @@ -1591,11 +1611,36 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const shell = shells.get(id); if (shell === undefined) return; if (shell.turn === undefined || shell.toolId === undefined || shell.tool === undefined) { - if (status !== "running") shell.settlement = { status, exit }; + const updatedAt = yield* DateTime.now; + yield* emit({ + type: "turn_item.updated", + driver, + sessionScoped: true, + turnItem: { + id: idAllocator.derive.turnItemFromProviderItem({ driver, nativeItemId: id }), + threadId: shell.state.providerThread.appThreadId ?? input.threadId, + runId: shell.recoveredRunId ?? null, + nodeId: null, + providerThreadId: shell.state.providerThread.id, + providerTurnId: null, + nativeItemRef: ref(id), + parentItemId: null, + ordinal: 0, + status, + title: null, + startedAt: shell.startedAt ?? updatedAt, + completedAt: status === "running" ? null : updatedAt, + updatedAt, + type: "command_execution", + input: shell.command, + output: shell.output, + ...(exit === undefined ? {} : { exitCode: exit }), + }, + }); return; } shell.turn.tools.set(shell.toolId, shell.tool); - yield* emitTool(shell.projectionState ?? shell.state, shell.turn, shell.toolId, status, { + yield* emitTool(shell.state, shell.turn, shell.toolId, status, { output: shell.output, nativeTaskId: id, metadata: { shellID: id, exit }, @@ -1603,23 +1648,6 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr shell.turn.tools.delete(shell.toolId); }); - const bindRecoveredShells = Effect.fn("OpenCode2Adapter.bindRecoveredShells")(function* ( - state: ThreadState, - turn: ActiveTurn, - ) { - for (const [id, shell] of shells) { - if (shell.turn !== undefined || !sessionsOf(state).includes(shell.state)) continue; - // Root command events need a real run owner to pass server routing. - shell.turn = turn; - shell.projectionState = state; - shell.toolId = id; - shell.tool = { name: "shell", input: { command: shell.command } }; - turn.startedAt.set(id, shell.startedAt ?? (yield* DateTime.now)); - yield* emitShell(id, shell.settlement?.status ?? "running", shell.settlement?.exit); - if (shell.settlement !== undefined) shells.delete(id); - } - }); - const stopShell = Effect.fn("OpenCode2Adapter.stopShell")(function* (id: string) { const shell = shells.get(id); if (shell === undefined || !shell.running) return; @@ -1642,7 +1670,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr shell.running = false; shell.state.stoppedChildren.add(id); yield* emitShell(id, "interrupted"); - if (shell.turn !== undefined) shells.delete(id); + shells.delete(id); shellInfo.delete(id); yield* syncRosterAncestors(shell.state); }); @@ -1672,14 +1700,35 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ownerOf(sessionId) ?? (jobs?._tag === "Some" ? state : undefined)); if (owner === undefined || !sessionsOf(state).includes(owner)) continue; - if (!shells.has(info.id)) + if (!shells.has(info.id)) { + if (state.active === undefined && !hasBackground(state)) { + state.sessionScopedBackground = true; + delete state.publishedBackgroundWork; + } + const target = requestTurn(owner.sessionId); shells.set(info.id, { state: owner, + ...(target === undefined + ? { + recoveredRunId: + owner.providerThread.lastRunOrdinal === null + ? null + : idAllocator.derive.run({ + threadId: owner.providerThread.appThreadId ?? input.threadId, + ordinal: owner.providerThread.lastRunOrdinal, + }), + } + : { + turn: target.turn, + toolId: info.id, + tool: { name: "shell", input: { command: info.command } }, + }), command: info.command, running: true, output: "", cursor: 0, }); + } } for (const [id, shell] of shells) { if (!shell.running || !sessionsOf(state).includes(shell.state)) continue; @@ -1693,6 +1742,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr if (found === undefined) { shell.running = false; yield* emitShell(id, "interrupted"); + shells.delete(id); return; } const info = found.data; @@ -1712,6 +1762,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr : "failed", info.exit, ); + if (!shell.running) shells.delete(id); }).pipe( Effect.catch((cause) => Effect.logWarning("Could not recover an OpenCode shell.", cause).pipe( @@ -1725,12 +1776,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr lock.withPermit( effect.pipe( Effect.timeout(INVENTORY_TIMEOUT), - Effect.ensuring( - Effect.gen(function* () { - if (state.active !== undefined) yield* bindRecoveredShells(state, state.active); - yield* syncRosterAncestors(state); - }), - ), + Effect.ensuring(syncRosterAncestors(state)), ), ), ); @@ -2830,7 +2876,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr : "completed", ); } - if (shell?.turn !== undefined) shells.delete(id); + shells.delete(id); shellInfo.delete(id); } return; @@ -2928,7 +2974,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr : "failed", "exit" in event.data ? event.data.exit : undefined, ); - if (shell.turn !== undefined) shells.delete(event.data.id); + shells.delete(event.data.id); shellInfo.delete(event.data.id); yield* syncRosterAncestors(shell.state); return; @@ -3495,21 +3541,46 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr // Subscribed before any session or prompt call, so no event of theirs is missed. yield* follow(yield* connection.events).pipe(Effect.forkScoped); - yield* lock - .withPermit( - Effect.gen(function* () { - for (const [id, shell] of shells) { - if (!shell.running) continue; - const before = shell.cursor; - yield* readShellOutput(id).pipe( - Effect.timeout(INVENTORY_TIMEOUT), - Effect.ignore({ log: true }), - ); - if (shell.cursor !== before) yield* emitShell(id, "running"); - } - }), - ) - .pipe(Effect.repeat(Schedule.spaced("2 seconds")), Effect.forkScoped); + yield* Effect.gen(function* () { + const snapshots = yield* lock.withPermit( + Effect.sync(() => + [...shells] + .filter(([, shell]) => shell.running) + .map(([id, shell]) => ({ + id, + shell, + cursor: shell.cursor, + output: shell.output, + directory: shell.state.directory, + })), + ), + ); + yield* Effect.forEach( + snapshots, + (snapshot) => + fetchShellOutput(snapshot.id, snapshot).pipe( + Effect.flatMap((next) => + lock.withPermit( + Effect.gen(function* () { + const shell = shells.get(snapshot.id); + // An exit, Stop, or recovery may have consumed this cursor while the request was in flight. + if ( + shell !== snapshot.shell || + !shell.running || + shell.cursor !== snapshot.cursor + ) + return; + if (next.cursor <= shell.cursor) return; + Object.assign(shell, next); + yield* emitShell(snapshot.id, "running"); + }), + ), + ), + Effect.ignore({ log: true }), + ), + { concurrency: 4, discard: true }, + ); + }).pipe(Effect.repeat(Schedule.spaced("2 seconds")), Effect.forkScoped); // External servers outlive T3; stop their turns before losing observation. if (connection.external) { @@ -3951,6 +4022,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr }); } state.active = turn; + state.sessionScopedBackground = false; yield* emitProviderTurn(state, turn, providerTurn); state.providerThread = { ...state.providerThread, @@ -3961,7 +4033,6 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr }; yield* emitProviderThread(state); yield* setSessionStatus("running", null); - yield* lock.withPermit(bindRecoveredShells(state, turn)); return turn; }); From b368b819c98130cc7009d7d8cc21597e9f16012e Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Tue, 10 Nov 2026 22:24:32 -0600 Subject: [PATCH 4/4] fix(opencode): preserve shell projections after recovery timeouts --- .../Adapters/OpenCode2AdapterV2.test.ts | 217 ++++++++++++++---- .../src/server/v2/adapter.ts | 20 +- 2 files changed, 197 insertions(+), 40 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 75d7f2584fb84..160bcc6f32699 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -1214,7 +1214,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); - it.effect.each(["remove", "lookup", "output"] as const)( + it.effect.each(["remove", "lookup", "output", "timeout"] as const)( "continues recovering and stopping native work when shell %s fails", (failure) => Effect.gen(function* () { @@ -1234,7 +1234,8 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { out("shell.output", { id }), reply("shell.output", response({ output: "", cursor: 0, size: 0, truncated: false })), ]; - const { runtime, thread } = yield* resumed( + const gate = makeProviderReplayGate(["recovery-pending", "polled"]); + const recovery = yield* resumed( [ ...backgroundLaunch(CHILD), ...output(ids[0]!), @@ -1258,29 +1259,74 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { ], { version: "2.2.1", + replayGate: gate, shellRecovery: [ out("shell.list", ""), reply("shell.list", response(ids.map(info))), - ...ids.flatMap((id) => [ - out("shell.get", { id }), - reply( - "shell.get", - id === ids[0] && failure === "lookup" - ? { status: 500, body: {} } - : response(info(id)), - ), - ...(id === ids[0] && failure === "lookup" - ? [] - : id === ids[0] && failure === "output" - ? [ - out("shell.output", { id }), - reply("shell.output", { status: 500, body: {} }), - ] - : output(id)), - ]), + ...(failure === "timeout" + ? [ + out("shell.get", { id: ids[0] }), + reply("shell.get", ""), + event( + "session.execution.succeeded", + { sessionID: "ses_unrelated" }, + "recovery-pending", + ), + ...ids.map((id) => out("shell.output", { id })), + ...ids.map(() => + reply( + "shell.output", + response({ output: "", cursor: 0, size: 0, truncated: false }), + ), + ), + event("session.execution.succeeded", { sessionID: "ses_unrelated" }, "polled"), + ] + : ids.flatMap((id) => [ + out("shell.get", { id }), + reply( + "shell.get", + id === ids[0] && failure === "lookup" + ? { status: 500, body: {} } + : response(info(id)), + ), + ...(id === ids[0] && failure === "lookup" + ? [] + : id === ids[0] && failure === "output" + ? [ + out("shell.output", { id }), + reply("shell.output", { status: 500, body: {} }), + ] + : output(id)), + ])), ], }, + ).pipe(Effect.forkScoped); + if (failure === "timeout") { + yield* Effect.promise(() => gate.waitForReached("recovery-pending")); + yield* TestClock.adjust("5 seconds"); + } + const { runtime, thread } = yield* Fiber.join(recovery); + const recovered = yield* runtime.events.pipe( + Stream.takeUntil( + (event) => + event.type === "provider_thread.updated" && + event.providerThread.pendingBackgroundTasks?.length === 2, + ), + Stream.runCollect, ); + assert.sameMembers( + recovered.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "command_execution" + ? [event.turnItem.nativeItemRef?.nativeId] + : [], + ), + ids, + ); + if (failure === "timeout") { + gate.release("recovery-pending"); + yield* Effect.promise(() => gate.waitForReached("polled")); + gate.release("polled"); + } const watch = yield* watchBackgroundTurn(runtime); yield* runtime.startTurn(withLineage(thread)); yield* watch.attached; @@ -1587,26 +1633,119 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { event("session.execution.started", { sessionID: CHILD }), ]; - it.effect("settles a background subagent that fails, so no work is left pending", () => - Effect.gen(function* () { - const { runtime, thread } = yield* resumed([ - ...launchedAndRunning, - // The parent turn ends, and the background child fails: it reports - // nothing back, so its call has to end with the failure. - event("session.execution.succeeded", { sessionID: SESSION }), - event("session.execution.failed", { - sessionID: CHILD, - error: { type: "provider.quota", message: "Rate limit exceeded." }, - }), - ]); - const watch = yield* watchCalls(runtime, thread); - yield* runtime.startTurn(withLineage(thread)); - yield* watch.until( - () => watch.calls.get("Sleep")?.status === "failed" && watch.state.roster?.length === 0, - ); - assert.deepEqual(watch.state.roster, []); - assert.isFalse(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); - }).pipe(Effect.scoped), + it.effect.each([false, true])( + "settles a failed background subagent with recovered child=%s", + (recoveredChild) => + Effect.gen(function* () { + const ids = yield* IdAllocator.IdAllocatorV2.pipe(Effect.provide(IdAllocator.layer)); + const childThreadId = ids.derive.threadFromProviderThread({ + driver: OPENCODE_PROVIDER, + nativeThreadId: CHILD, + providerInstanceId: instanceId, + }); + const childProviderThread = { + ...providerThread(yield* DateTime.now), + id: ids.derive.providerThread({ + driver: OPENCODE_PROVIDER, + nativeThreadId: CHILD, + providerInstanceId: instanceId, + }), + appThreadId: childThreadId, + nativeThreadRef: { + driver: OPENCODE_PROVIDER, + nativeId: CHILD, + strength: "strong" as const, + }, + }; + const shell = { + id: "sh_old_child", + command: "echo done", + status: "exited", + exit: 0, + cwd: WORK, + shell: "/bin/sh", + file: "/remote/output", + metadata: { sessionID: CHILD }, + time: { started: 1, completed: 2 }, + }; + const response = (data: unknown) => ({ location: { directory: WORK }, data }); + const { runtime, thread } = yield* resumed( + [ + ...(recoveredChild + ? [ + out("session.get", { sessionID: CHILD }), + replyData("session.get", sessionInfo({ id: CHILD })), + out("permission.list", { sessionID: CHILD }), + replyData("permission.list", []), + out("session.form.list", { sessionID: CHILD }), + replyData("session.form.list", []), + out("shell.list", ""), + reply("shell.list", response([shell])), + out("shell.get", { id: shell.id }), + reply("shell.get", response(shell)), + out("shell.output", { id: shell.id }), + reply( + "shell.output", + response({ output: "done", cursor: 4, size: 4, truncated: false }), + ), + out("session.update", { sessionID: CHILD, permissions: "" }), + reply("session.update", null), + ] + : []), + ...launchedAndRunning, + // The parent turn ends, and the background child fails: it reports + // nothing back, so its call has to end with the failure. + event("session.execution.succeeded", { sessionID: SESSION }), + event("session.execution.failed", { + sessionID: CHILD, + error: { type: "provider.quota", message: "Rate limit exceeded." }, + }), + ], + recoveredChild ? { version: "2.2.1" } : undefined, + ); + if (recoveredChild) + yield* runtime.resumeThread({ + providerThread: childProviderThread, + threadId: childThreadId, + modelSelection: bigPickle, + runtimePolicy: policy(), + }); + const childUpdates: Array< + Extract + > = []; + const watch = yield* watchCalls( + { + ...runtime, + events: runtime.events.pipe( + Stream.tap((event) => + Effect.sync(() => { + if ( + event.type === "provider_thread.updated" && + event.providerThread.id === childProviderThread.id && + event.providerThread.ownerNodeId !== null + ) + childUpdates.push(event); + }), + ), + ), + }, + thread, + ); + yield* runtime.startTurn(withLineage(thread)); + yield* watch.until( + () => watch.calls.get("Sleep")?.status === "failed" && watch.state.roster?.length === 0, + ); + assert.includeMembers( + childUpdates.map((event) => event.providerThread.status), + ["active", "idle"], + ); + assert.isFalse( + childUpdates.some((event) => event.sessionScoped === true), + "reattached child status must route through the parent run", + ); + assert.deepEqual(watch.state.roster, []); + assert.isFalse(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + }).pipe(Effect.scoped), ); // regression: #21979 a failed background subagent pinned its thread as busy diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index e053347e7519a..080b1c2b1d69c 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -906,6 +906,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr cursor: number; startedAt?: DateTime.Utc; recoveredRunId?: RunId | null; + published?: boolean; } >(); @@ -1637,6 +1638,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ...(exit === undefined ? {} : { exitCode: exit }), }, }); + shell.published = true; return; } shell.turn.tools.set(shell.toolId, shell.tool); @@ -1646,6 +1648,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr metadata: { shellID: id, exit }, }); shell.turn.tools.delete(shell.toolId); + shell.published = true; }); const stopShell = Effect.fn("OpenCode2Adapter.stopShell")(function* (id: string) { @@ -1776,7 +1779,19 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr lock.withPermit( effect.pipe( Effect.timeout(INVENTORY_TIMEOUT), - Effect.ensuring(syncRosterAncestors(state)), + Effect.ensuring( + Effect.gen(function* () { + for (const [id, shell] of shells) { + if ( + shell.running && + shell.published !== true && + sessionsOf(state).includes(shell.state) + ) + yield* emitShell(id, "running"); + } + yield* syncRosterAncestors(state); + }), + ), ), ), ); @@ -1879,6 +1894,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const child = previous ?? newThreadState(childId, providerThread, call.state.directory, subagent); child.subagent = subagent; + child.sessionScopedBackground = false; child.model = info?.model ?? child.model; child.providerThread = withReportedModel(providerThread, child.model); child.directory = call.state.directory; @@ -1918,6 +1934,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr if (child.active !== undefined && status !== "completed") { yield* finishTurn(child, { status: "interrupted" }); } + child.sessionScopedBackground = false; child.providerThread = { ...child.providerThread, status: "idle", @@ -1965,6 +1982,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr turn.before = subagent.queued; subagent.queued = undefined; child.active = turn; + child.sessionScopedBackground = false; yield* emit({ type: "node.updated", driver,