From 9dce1110c07f46ae65fc049e8e84b6f05a39ab66 Mon Sep 17 00:00:00 2001 From: Nikita Koynov <43469098+nkoynov@users.noreply.github.com> Date: Thu, 8 Oct 2026 12:01:55 +0300 Subject: [PATCH 1/2] fix(server): settle every OpenCode subagent call one report answers Calling a running OpenCode subagent again joins its run, which OpenCode reports once. Settle the calls that joined it with that report, and treat a call that returned while the subagent runs on as background. --- .../Adapters/OpenCode2AdapterV2.test.ts | 186 ++++++++++++++++++ .../src/server/v2/adapter.ts | 43 ++-- 2 files changed, 215 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 ff87c347641e..1614b5dff759 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -881,6 +881,192 @@ describe("OpenCode2 adapter", () => { }).pipe(Effect.scoped), ); + /** A `subagent` call's result while its subagent runs on (`session.tool.success` is at version 2). */ + const returnedRunning = (id: string): ProviderReplayEntry => ({ + type: "emit_inbound", + frame: { + type: "sdk.event", + event: { + id: `evt_toolsuccess_${id}`, + created: 1, + type: "session.tool.success", + data: { + sessionID: SESSION, + assistantMessageID: "msg_assistant", + id, + content: [{ type: "text", text: "The subagent is working in the background." }], + metadata: { sessionID: CHILD, status: "running" }, + executed: true, + }, + durable: { aggregateID: SESSION, seq: 1, version: 2 }, + }, + }, + }); + /** The model calls the subagent again by its session; the tool returns while it runs on. */ + const callAgain = (input: Record): ReadonlyArray => { + const tool = { sessionID: SESSION, assistantMessageID: "msg_assistant", id: "call-again" }; + return [ + event("session.tool.input.started", { ...tool, name: "subagent" }), + event("session.tool.called", { + ...tool, + name: "subagent", + input: { description: "Again", prompt: "also this", sessionID: CHILD, ...input }, + executed: false, + }), + event("session.tool.progress", { + ...tool, + metadata: { sessionID: CHILD, status: "running" }, + }), + // A session called again gets the thread's rules. + out("session.update", { sessionID: CHILD, permissions: "" }), + reply("session.update", null), + returnedRunning(tool.id), + ]; + }; + const childReport = (inboxID: string, text: string) => + event("session.inbox.enqueued", { + inboxID, + sessionID: SESSION, + item: { + type: "synthetic", + payload: { + text: `\n${text}\n`, + description: "Sleep", + metadata: { source: "subagent", childID: CHILD, agent: "General", state: "completed" }, + }, + delivery: "steer", + }, + }); + /** The subagent rows by title, the thread's background roster and the child's turns, as emitted. */ + const watchCalls = ( + runtime: ProviderAdapterV2SessionRuntime, + thread: OrchestrationV2ProviderThread, + ) => + Effect.gen(function* () { + const calls = new Map(); + const childTurns = new Map(); + const state: { roster: ReadonlyArray | undefined } = { roster: undefined }; + let wake = yield* Deferred.make(); + yield* runtime.events.pipe( + Stream.tap((event) => + Effect.gen(function* () { + if (event.type === "subagent.updated" && event.subagent.title !== null) { + calls.set(event.subagent.title, { + status: event.subagent.status, + result: event.subagent.result, + }); + } + if (event.type === "provider_thread.updated" && event.providerThread.id === thread.id) { + state.roster = event.providerThread.pendingBackgroundTasks; + } + const id = + event.type === "provider_turn.updated" + ? event.providerTurn.nativeTurnRef?.nativeId + : undefined; + if ( + event.type === "provider_turn.updated" && + id?.startsWith(`${CHILD}:turn:`) === true + ) { + childTurns.set(id, event.providerTurn.status); + } + yield* Deferred.succeed(wake, undefined); + }), + ), + Stream.runDrain, + Effect.forkScoped, + ); + const until = (check: () => boolean): Effect.Effect => + Effect.suspend(() => { + if (check()) return Effect.void; + return Effect.gen(function* () { + wake = yield* Deferred.make(); + if (check()) return; + yield* Deferred.await(wake); + yield* until(check); + }); + }); + return { calls, childTurns, state, until }; + }); + const launchedAndRunning = [ + ...backgroundLaunch(CHILD), + returnedRunning("call-background"), + event("session.execution.started", { sessionID: CHILD }), + ]; + + it.effect("settles every call that joined a running subagent with its one report", () => + Effect.gen(function* () { + const { runtime, thread } = yield* resumed([ + ...launchedAndRunning, + // OpenCode steers the running subagent and joins its run: one report answers both calls. + ...callAgain({ background: true }), + event("session.execution.succeeded", { sessionID: SESSION }), + event("session.execution.succeeded", { sessionID: CHILD }), + childReport("msg_report", "CHILD_OK"), + ]); + const watch = yield* watchCalls(runtime, thread); + yield* runtime.startTurn(withLineage(thread)); + yield* watch.until( + () => + watch.calls.get("Again")?.status === "completed" && + watch.calls.get("Sleep")?.status === "completed" && + watch.state.roster?.length === 0, + ); + assert.deepEqual(Object.fromEntries(watch.calls), { + Sleep: { status: "completed", result: "CHILD_OK" }, + Again: { status: "completed", result: "CHILD_OK" }, + }); + }).pipe(Effect.scoped), + ); + + it.effect("keeps a foreground call that joined a background run for that run's report", () => + Effect.gen(function* () { + const { runtime, thread } = yield* resumed([ + ...launchedAndRunning, + // Without `background`, the call joins the run in the background all the same. + ...callAgain({}), + event("session.execution.succeeded", { sessionID: SESSION }), + event("session.execution.succeeded", { sessionID: CHILD }), + childReport("msg_report", "CHILD_OK"), + ]); + const watch = yield* watchCalls(runtime, thread); + yield* runtime.startTurn(withLineage(thread)); + yield* watch.until( + () => + watch.calls.get("Again")?.status === "completed" && + watch.calls.get("Sleep")?.status === "completed" && + watch.state.roster?.length === 0, + ); + // Not settled empty when the turn that made it ended. + assert.deepEqual(watch.calls.get("Again"), { status: "completed", result: "CHILD_OK" }); + }).pipe(Effect.scoped), + ); + + it.effect("keeps a call to a finished subagent for its own run's report", () => + Effect.gen(function* () { + const { runtime, thread } = yield* resumed([ + ...launchedAndRunning, + event("session.execution.succeeded", { sessionID: CHILD }), + // Called again after it ended but before its report: the call starts a new run. + ...callAgain({ background: true }), + childReport("msg_report_1", "FIRST_OK"), + event("session.execution.started", { sessionID: CHILD }), + event("session.execution.succeeded", { sessionID: CHILD }), + event("session.execution.succeeded", { sessionID: SESSION }), + childReport("msg_report_2", "SECOND_OK"), + ]); + const watch = yield* watchCalls(runtime, thread); + yield* runtime.startTurn(withLineage(thread)); + // The new run's turn starts after the first report was handled. + yield* watch.until(() => watch.childTurns.has(`${CHILD}:turn:2`)); + assert.deepEqual(Object.fromEntries(watch.calls), { + Sleep: { status: "completed", result: "FIRST_OK" }, + Again: { status: "running", result: null }, + }); + yield* watch.until(() => watch.calls.get("Again")?.status === "completed"); + assert.deepEqual(watch.calls.get("Again"), { status: "completed", result: "SECOND_OK" }); + }).pipe(Effect.scoped), + ); + it.effect("leaves no background work pending once a reconnect finds its subagent gone", () => Effect.gen(function* () { // The subagent ends while the stream is down: its end, its report and diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index 97429ddfde85..f69e6311dd48 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -291,6 +291,8 @@ interface SubagentCall { model: string | null; /** The tool returned while the subagent runs on; the report OpenCode gives its parent settles it. */ background: boolean; + /** The running call whose subagent run this call joined, so that run's one report settles both. */ + joined: SubagentCall | undefined; child: ThreadState | undefined; status: OrchestrationV2Subagent["status"]; result: string | null; @@ -1489,6 +1491,14 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr // ran (each one's native id is its own), its grants and rules, and the // background work an earlier call left running, which Stop must reach. const previous = threads.get(childId); + // Called again while it runs, the subagent takes the prompt as a steer + // and OpenCode reports that run once, for every call that joined it. + if (previous?.active !== undefined) { + call.joined = [...call.state.calls.values()].findLast( + (candidate) => + candidate !== call && candidate.child === previous && candidate.joined === undefined, + ); + } const subagent = { call, appThread, @@ -2175,6 +2185,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr agent: undefined, model: null, background: false, + joined: undefined, child: undefined, status: "running", result: null, @@ -2240,8 +2251,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr if (call !== undefined) { const childId = stringField(event.data.metadata, "sessionID"); if (childId !== undefined) yield* attachChild(call, childId); - // A background call returns at launch; its report settles it. - if (event.data.metadata?.["status"] === "running") return; + // A background call returns at launch, and so does one that joined a + // background run; the report settles it. + if (event.data.metadata?.["status"] === "running") { + if (call.background) return; + call.background = true; + return yield* emitSubagent(call); + } return yield* settleCall(call, "completed", subagentOutput(textOf(event.data.content))); } const output = textOf(event.data.content); @@ -2379,6 +2395,9 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const call = [...state.calls.values()].find( (candidate) => candidate.child?.sessionId === childId, ); + const calls = [...state.calls.values()].filter( + (candidate) => candidate === call || (call !== undefined && candidate.joined === call), + ); const outcome = reportOutcome(stringField(payload.metadata, "state")); state.reports.set(inboxId, { inboxId, @@ -2391,18 +2410,14 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr outcome, }, }); - if (call === undefined) return; - yield* settleCall( - call, - state.stoppedChildren.has(childId) - ? "interrupted" - : outcome === "failed" - ? "failed" - : outcome === "cancelled" - ? "cancelled" - : "completed", - subagentOutput(payload.text), - ); + const status = state.stoppedChildren.has(childId) + ? "interrupted" + : outcome === "failed" + ? "failed" + : outcome === "cancelled" + ? "cancelled" + : "completed"; + for (const each of calls) yield* settleCall(each, status, subagentOutput(payload.text)); }); /** From 621896cd20bf2b7beb97223c4f241f71771c9f61 Mon Sep 17 00:00:00 2001 From: Nikita Koynov <43469098+nkoynov@users.noreply.github.com> Date: Fri, 9 Oct 2026 08:31:13 +0300 Subject: [PATCH 2/2] test(server): check the calls as the subagent's second turn starts --- .../Adapters/OpenCode2AdapterV2.test.ts | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 1614b5dff759..6fa24e51a463 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -945,6 +945,10 @@ describe("OpenCode2 adapter", () => { Effect.gen(function* () { const calls = new Map(); const childTurns = new Map(); + const callsAtChildTurnStart = new Map< + string, + Record + >(); const state: { roster: ReadonlyArray | undefined } = { roster: undefined }; let wake = yield* Deferred.make(); yield* runtime.events.pipe( @@ -967,6 +971,9 @@ describe("OpenCode2 adapter", () => { event.type === "provider_turn.updated" && id?.startsWith(`${CHILD}:turn:`) === true ) { + if (!childTurns.has(id)) { + callsAtChildTurnStart.set(id, Object.fromEntries(calls)); + } childTurns.set(id, event.providerTurn.status); } yield* Deferred.succeed(wake, undefined); @@ -985,7 +992,7 @@ describe("OpenCode2 adapter", () => { yield* until(check); }); }); - return { calls, childTurns, state, until }; + return { calls, callsAtChildTurnStart, childTurns, state, until }; }); const launchedAndRunning = [ ...backgroundLaunch(CHILD), @@ -1058,7 +1065,7 @@ describe("OpenCode2 adapter", () => { yield* runtime.startTurn(withLineage(thread)); // The new run's turn starts after the first report was handled. yield* watch.until(() => watch.childTurns.has(`${CHILD}:turn:2`)); - assert.deepEqual(Object.fromEntries(watch.calls), { + assert.deepEqual(watch.callsAtChildTurnStart.get(`${CHILD}:turn:2`), { Sleep: { status: "completed", result: "FIRST_OK" }, Again: { status: "running", result: null }, });