diff --git a/packages/provider-opencode/src/server/adapter.test.ts b/packages/provider-opencode/src/server/adapter.test.ts index c1c89b2b96263..4e547d952a756 100644 --- a/packages/provider-opencode/src/server/adapter.test.ts +++ b/packages/provider-opencode/src/server/adapter.test.ts @@ -1387,7 +1387,13 @@ describe("OpenCodeAdapterV2", () => { ); // OpenCode 1.18.32 background tasks complete at launch; children stay busy after root settlement. - it.effect("reports a background task child as pending work after the root turn settles", () => + // Event order from a live 1.19.5 run: the child reports busy only after the launch completes. + it.effect.each([ + { case: "finishes", ending: "idle", outcome: "completed" }, + { case: "is aborted", ending: "abort", outcome: "interrupted" }, + { case: "fails before its first turn", ending: "error", outcome: "failed" }, + { case: "never starts and the thread is stopped", ending: "stop", outcome: "interrupted" }, + ])("keeps a background task's subagent running until its child $case", ({ ending, outcome }) => Effect.gen(function* () { const nativeEvents = asyncEventStream(); const push = (event: unknown) => Effect.promise(() => nativeEvents.push(event)); @@ -1415,9 +1421,14 @@ describe("OpenCodeAdapterV2", () => { return yield* Effect.die("OpenCode runtime must expose hasPendingBackgroundWork."); } yield* harness.startTurn(); - const terminal = yield* harness.runtime.events.pipe( - Stream.filter((event) => event.type === "turn.terminal"), - Stream.runHead, + const received = yield* harness.runtime.events.pipe( + Stream.takeUntil( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "subagent" && + event.turnItem.status !== "running", + ), + Stream.runCollect, Effect.forkScoped, ); const taskPart = (status: "running" | "completed") => ({ @@ -1459,7 +1470,6 @@ describe("OpenCodeAdapterV2", () => { }, }); yield* push(status(root, "busy")); - yield* push(taskPart("running")); yield* push({ type: "session.created", properties: { @@ -1467,15 +1477,55 @@ describe("OpenCodeAdapterV2", () => { info: { id: child, parentID: root, time: { created: 2, updated: 2 } }, }, }); - yield* push(status(child, "busy")); + yield* push(taskPart("running")); + const starts = ending !== "stop" && ending !== "error"; + if (starts) { + yield* push({ + type: "message.updated", + properties: { + sessionID: child, + info: { id: "msg_child_prompt", sessionID: child, role: "user", time: { created: 2 } }, + }, + }); + } yield* push(taskPart("completed")); + if (starts) yield* push(status(child, "busy")); yield* push(status(root, "idle")); - assert.equal(Option.getOrUndefined(yield* Fiber.join(terminal))?.status, "completed"); - assert.isTrue(yield* hasPendingBackgroundWork, "the running child must pin idle release"); + assert.equal(yield* hasPendingBackgroundWork, starts, "a running child pins idle release"); - yield* push(status(child, "idle")); - yield* push({ type: "session.idle", properties: { sessionID: child } }); + if (ending === "stop") { + yield* harness.runtime.interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: ProviderTurnId.make("provider-turn:settled"), + requestRuntimeRestart: true, + }); + } else { + if (ending !== "idle") { + yield* push({ + type: "session.error", + properties: { + sessionID: child, + error: + ending === "abort" + ? { name: "MessageAbortedError", data: {} } + : { name: "UnknownError", data: { message: "model unavailable" } }, + }, + }); + } + yield* push(status(child, "idle")); + yield* push({ type: "session.idle", properties: { sessionID: child } }); + } assert.isFalse(yield* hasPendingBackgroundWork, "an idle child must not pin idle release"); + assert.deepEqual( + (yield* Fiber.join(received)).flatMap((event) => + event.type === "subagent.updated" + ? [`subagent ${event.subagent.status}`] + : event.type === "turn.terminal" + ? [`root turn ${event.status}`] + : [], + ), + ["subagent running", "subagent running", "root turn completed", `subagent ${outcome}`], + ); }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, McpProviderSessions.layer))), ); diff --git a/packages/provider-opencode/src/server/adapter.ts b/packages/provider-opencode/src/server/adapter.ts index 4d265ab9dc460..3fed50f96e436 100644 --- a/packages/provider-opencode/src/server/adapter.ts +++ b/packages/provider-opencode/src/server/adapter.ts @@ -391,6 +391,8 @@ interface OpenCodeSubagentContext { childProviderThreadId: OrchestrationV2ProviderThread["id"] | null; model: string | null; result: string | null; + // A background task call returns at launch; held here until its child session stops. + launch: ToolPart | null; } interface OpenCodeThreadState { @@ -1039,6 +1041,10 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function const relatedSessionOwners = new Map(); // Busy children outlive parent settlement; keep their server open. const busySessionIds = new Set(); + // A background task returns before its new child session reports busy. + const unstartedSessionIds = new Set(); + // A child can error before any turn of it is projected; its next idle settles the launch. + const sessionErrorOutcomes = new Map(); // Remember settled requests across routing retries to prevent resurrection; only recent IDs matter. const settledNativeRequestIds = new Set(); const pendingChildRequestRoutes = new Set(); @@ -1359,6 +1365,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function state: OpenCodeThreadState, turn: ActiveOpenCodeTurn, part: ToolPart, + childOutcome?: TerminalTurnStatus, ) { const now = yield* DateTime.now; const nativeItemRef = providerRef(part.id); @@ -1389,6 +1396,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function childProviderThreadId: null, model: null, result: null, + launch: null, }; subagentsByNativeItemId.set(part.id, context); } @@ -1500,8 +1508,27 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function } const output = toolOutput(part); if (part.state.status === "completed" && output !== undefined) context.result = output; - const status = toolStatus(part); - const completedAt = toolCompletedAt(part, now); + const launchStatus = toolStatus(part); + context.launch = + childOutcome === undefined && + launchStatus.item === "completed" && + context.childSessionId !== null && + (busySessionIds.has(context.childSessionId) || + unstartedSessionIds.has(context.childSessionId)) + ? part + : null; + const status = + context.launch !== null + ? ({ node: "running", item: "running" } as const) + : childOutcome === undefined + ? launchStatus + : { node: childOutcome, item: childOutcome }; + const completedAt = + context.launch !== null + ? null + : childOutcome === undefined + ? toolCompletedAt(part, now) + : now; const subagentStatus: OrchestrationV2Subagent["status"] = status.item; const subagent: OrchestrationV2Subagent = { id: nodeId, @@ -1580,6 +1607,49 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function }); }); + const settleLaunchedSubagents = Effect.fnUntraced(function* ( + childSessionId: string, + outcome?: TerminalTurnStatus, + ) { + const lastTurn = Array.from(threads.get(childSessionId)?.providerTurns.values() ?? []).at( + -1, + )?.status; + const errorOutcome = sessionErrorOutcomes.get(childSessionId); + sessionErrorOutcomes.delete(childSessionId); + const childOutcome = + outcome ?? + errorOutcome ?? + (lastTurn === "interrupted" || lastTurn === "failed" || lastTurn === "cancelled" + ? lastTurn + : "completed"); + for (const context of subagentsByNativeItemId.values()) { + if (context.childSessionId === childSessionId && context.launch !== null) { + yield* emitSubagent( + context.parentState, + context.parentTurn, + context.launch, + childOutcome, + ); + } + } + }); + + // No idle event settles a launched child that Stop aborts before it reports busy. + const interruptLaunchedSubagents = Effect.fnUntraced(function* (rootSessionId: string) { + for (const context of subagentsByNativeItemId.values()) { + let owner = context.parentState; + while (owner.parentSubagent !== null) owner = owner.parentSubagent.parentState; + if (owner.nativeSessionId === rootSessionId && context.launch !== null) { + yield* emitSubagent( + context.parentState, + context.parentTurn, + context.launch, + "interrupted", + ); + } + } + }); + const emitToolPart = Effect.fnUntraced(function* ( state: OpenCodeThreadState, turn: ActiveOpenCodeTurn, @@ -2676,6 +2746,9 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function case "session.created": case "session.updated": { const info = event.properties.info; + if (event.type === "session.created" && info.parentID !== undefined) { + unstartedSessionIds.add(info.id); + } if (info.parentID !== undefined && !threads.has(info.id)) { const owner = threads.get(info.parentID) ?? relatedSessionOwners.get(info.parentID); if (owner !== undefined) { @@ -2687,9 +2760,12 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function case "session.deleted": relatedSessionOwners.delete(event.properties.info.id); busySessionIds.delete(event.properties.info.id); + unstartedSessionIds.delete(event.properties.info.id); + yield* settleLaunchedSubagents(event.properties.info.id, "interrupted"); return; case "session.status": { const sessionId = event.properties.sessionID; + unstartedSessionIds.delete(sessionId); switch (event.properties.status.type) { case "busy": case "retry": @@ -2718,10 +2794,14 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function state.activeTurn.interrupted ? "interrupted" : "completed", ); } + if (event.properties.status.type === "idle") { + yield* settleLaunchedSubagents(sessionId); + } return; } case "session.idle": { busySessionIds.delete(event.properties.sessionID); + unstartedSessionIds.delete(event.properties.sessionID); const state = threads.get(event.properties.sessionID); if (state?.activeTurn !== null && state?.activeTurn !== undefined) { if (state.activeTurn.admissionPending) { @@ -2734,6 +2814,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function state.activeTurn.interrupted ? "interrupted" : "completed", ); } + yield* settleLaunchedSubagents(event.properties.sessionID); return; } case "session.error": { @@ -2751,6 +2832,12 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function }); return; } + if (event.properties.sessionID !== undefined) { + sessionErrorOutcomes.set( + event.properties.sessionID, + isMessageAbortedError(event) ? "interrupted" : "failed", + ); + } const states = event.properties.sessionID === undefined ? Array.from(threads.values()).filter((state) => state.activeTurn !== null) @@ -2815,6 +2902,8 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function Effect.flatMap((exit) => Effect.gen(function* () { busySessionIds.clear(); + unstartedSessionIds.clear(); + sessionErrorOutcomes.clear(); if (closing || abortController.signal.aborted) return; const detail = Exit.isSuccess(exit) ? "OpenCode event stream ended unexpectedly." @@ -2830,6 +2919,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function failure: makeProviderFailure({ message: detail, class: "transport_error" }), threadDisposition: "broken", }); + yield* settleLaunchedSubagents(state.nativeSessionId, "failed"); } yield* Queue.end(events); }), @@ -3584,6 +3674,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function for (const controller of commandControllers.get(sessionId) ?? []) controller.abort(); yield* abortDescendants(sessionId); + yield* interruptLaunchedSubagents(sessionId); return; } if ( @@ -3622,6 +3713,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function ); // Root abort leaves children running; report incomplete cleanup even after root termination. yield* abortDescendants(sessionId); + yield* interruptLaunchedSubagents(sessionId); }, (effect, interruptInput) => effect.pipe(