From b2d5051b10d508381ca44d3e4ad86ffd4941432c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:14:59 +0000 Subject: [PATCH 1/4] fix(opencode): publish ordered pending-work snapshots Publish native continuation work with provider-thread events and refresh every ancestor when calls settle, including direct nested Stop. Source: https://github.com/pingdotgg/t3code/pull/19388 Co-authored-by: Exotic209093 <134711311+Exotic209093@users.noreply.github.com> --- .../Adapters/OpenCode2AdapterV2.test.ts | 227 ++++++++++++------ .../src/server/ProviderAdapter.ts | 2 + .../src/server/v2/adapter.ts | 87 ++++--- 3 files changed, 208 insertions(+), 108 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 4b3c72b11e17a..5f4e09d2f6bee 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -891,47 +891,85 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); - it.effect("keeps a finished background subagent's queued report as pending work", () => - Effect.gen(function* () { - const { runtime, thread } = yield* resumed([ - ...backgroundLaunch(CHILD), - event("session.execution.succeeded", { sessionID: SESSION }), - event("session.execution.succeeded", { sessionID: CHILD }), - event("session.inbox.enqueued", { - inboxID: "msg_report", - sessionID: SESSION, - item: { - type: "synthetic", - payload: { - text: `\nCHILD_OK\n`, - description: "Sleep", - metadata: { - source: "subagent", - childID: CHILD, - agent: "General", - state: "completed", - }, - }, - delivery: "steer", - }, - }), - event("session.usage.updated", { sessionID: SESSION }), - ]); - const reported = yield* Deferred.make(); - yield* runtime.events.pipe( - Stream.tap((event) => - event.type === "subagent.updated" && event.subagent.status === "completed" - ? Deferred.succeed(reported, undefined) - : Effect.void, - ), - Stream.runDrain, - Effect.forkScoped, - ); - yield* runtime.startTurn(withLineage(thread)); - yield* Deferred.await(reported); - assert.isTrue(yield* runtime.hasPendingBackgroundWork!); - assert.isTrue(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); - }).pipe(Effect.scoped), + it.effect( + "keeps a finished background subagent's queued report pending until its native continuation ends", + () => + Effect.gen(function* () { + const gate = makeProviderReplayGate(["continue"]); + const { runtime, thread } = yield* resumed( + [ + ...nestedLaunch, + event("session.execution.succeeded", { sessionID: DEEP }), + deepReport("completed"), + event("session.model.selected", { + sessionID: SESSION, + model: { providerID: "opencode", id: "big-pickle", variant: "thinking" }, + }), + event("session.execution.started", { sessionID: MIDDLE }, "continue"), + event("session.inbox.delivered", { sessionID: MIDDLE, inboxID: "msg_deep_report" }), + event("session.text.ended", { + sessionID: MIDDLE, + assistantMessageID: "msg_native_continuation", + ordinal: 0, + text: "GRANDCHILD_OK", + }), + event("session.execution.succeeded", { sessionID: MIDDLE }), + event("session.model.selected", { + sessionID: SESSION, + model: { providerID: "opencode", id: "big-pickle", variant: "none" }, + }), + ], + { replayGate: gate }, + ); + const queued = yield* Deferred.make(); + const drained = yield* Deferred.make(); + const timeline: Array = []; + yield* runtime.events.pipe( + Stream.runForEach((event) => + Effect.gen(function* () { + if (event.type === "provider_thread.updated") { + if ( + event.providerThread.nativeThreadRef?.nativeId === MIDDLE && + event.pendingBackgroundWork !== undefined + ) { + timeline.push(event.pendingBackgroundWork); + } + if (event.providerThread.id === thread.id) { + const variant = + event.providerThread.nativeMetadata?.modelSelection?.options?.[0]?.value; + if (variant === "thinking") yield* Deferred.succeed(queued, undefined); + if (variant === "none") yield* Deferred.succeed(drained, undefined); + } + } + if ( + event.type === "provider_turn.updated" && + event.providerTurn.nativeTurnRef?.nativeId === `${MIDDLE}:turn:2` + ) { + timeline.push(event.providerTurn.status); + } + if (event.type === "message.updated" && event.message.text === "GRANDCHILD_OK") { + timeline.push(event.message.text); + } + }), + ), + Effect.forkScoped, + ); + yield* runtime.startTurn(withLineage(thread)); + yield* Deferred.await(queued); + assert.isTrue(yield* runtime.hasPendingBackgroundWork!); + assert.isTrue(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + assert.strictEqual(timeline.at(-1), true); + timeline.length = 0; + gate.release("continue"); + yield* Deferred.await(drained); + assert.deepEqual( + timeline.filter((entry) => typeof entry === "string"), + ["running", "GRANDCHILD_OK", "completed"], + ); + assert.isAbove(timeline.indexOf(false), timeline.indexOf("completed")); + assert.strictEqual(timeline.at(-1), false); + assert.isFalse(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + }).pipe(Effect.scoped), ); const returnedRunning = (id: string): ProviderReplayEntry => ({ @@ -1334,6 +1372,8 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { Effect.gen(function* () { const deep: Array = []; const middleTurns = new Map(); + const pendingSnapshots = new Map(); + let rootVariant: unknown; const terminals: Array = []; const changed = yield* Deferred.make(); let wake = changed; @@ -1341,6 +1381,15 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { Stream.tap((event) => Effect.gen(function* () { if (event.type === "turn.terminal") terminals.push(event.status); + if (event.type === "provider_thread.updated") { + const nativeId = event.providerThread.nativeThreadRef?.nativeId; + if (nativeId != null && event.pendingBackgroundWork !== undefined) { + pendingSnapshots.set(nativeId, event.pendingBackgroundWork); + } + if (nativeId === SESSION) + rootVariant = + event.providerThread.nativeMetadata?.modelSelection?.options?.[0]?.value; + } if (event.type === "subagent.updated" && event.subagent.title === "Deep") { deep.push(event.subagent.status); } @@ -1365,7 +1414,16 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { yield* until(check); }); }); - return { deep, middleTurns, terminals, until }; + return { + deep, + middleTurns, + terminals, + pendingSnapshots, + get rootVariant() { + return rootVariant; + }, + until, + }; }); it.effect("stops a foreground subagent's background subagent after the turn ended", () => @@ -1408,38 +1466,52 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { it.effect("keeps a reused subagent's background subagent tracked and stops it", () => Effect.gen(function* () { const again = "call-middle-again"; - const { runtime, thread } = yield* resumed([ - ...nestedLaunch, - out("session.update", { sessionID: DEEP, permissions: "" }), - reply("session.update", null), - out("session.prompt", { sessionID: SESSION, text: "" }), - promptAccepted, - event("session.execution.started", { sessionID: SESSION }), - event("session.tool.input.started", { ...toolOf(SESSION, again), name: "subagent" }), - event("session.tool.called", { - ...toolOf(SESSION, again), - name: "subagent", - input: { description: "Middle", prompt: "again", sessionID: MIDDLE }, - executed: false, - }), - event("session.tool.progress", { - ...toolOf(SESSION, again), - metadata: { sessionID: MIDDLE, status: "running" }, - }), - out("session.update", { sessionID: MIDDLE, permissions: "" }), - reply("session.update", null), - event("session.execution.started", { sessionID: MIDDLE }), - event("session.execution.succeeded", { sessionID: MIDDLE }), - event("session.tool.success", { - ...toolOf(SESSION, again), - content: [{ type: "text", text: `` }], - metadata: { sessionID: MIDDLE, status: "completed", truncated: false }, - executed: false, - }), - event("session.execution.succeeded", { sessionID: SESSION }), - out("session.interrupt", { sessionID: DEEP }), - reply("session.interrupt", { interrupted: true }), - ]); + const gate = makeProviderReplayGate(["stopped"]); + const { runtime, thread } = yield* resumed( + [ + ...nestedLaunch, + out("session.update", { sessionID: DEEP, permissions: "" }), + reply("session.update", null), + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + event("session.execution.started", { sessionID: SESSION }), + event("session.tool.input.started", { ...toolOf(SESSION, again), name: "subagent" }), + event("session.tool.called", { + ...toolOf(SESSION, again), + name: "subagent", + input: { description: "Middle", prompt: "again", sessionID: MIDDLE }, + executed: false, + }), + event("session.tool.progress", { + ...toolOf(SESSION, again), + metadata: { sessionID: MIDDLE, status: "running" }, + }), + out("session.update", { sessionID: MIDDLE, permissions: "" }), + reply("session.update", null), + event("session.execution.started", { sessionID: MIDDLE }), + event("session.execution.succeeded", { sessionID: MIDDLE }), + event("session.tool.success", { + ...toolOf(SESSION, again), + content: [{ type: "text", text: `` }], + metadata: { sessionID: MIDDLE, status: "completed", truncated: false }, + executed: false, + }), + // Stop lands after native execution ends but before its report settles the call. + event("session.execution.succeeded", { sessionID: DEEP }), + event("session.execution.succeeded", { sessionID: SESSION }), + out("session.interrupt", { sessionID: DEEP }), + reply("session.interrupt", { interrupted: false }), + event( + "session.model.selected", + { + sessionID: SESSION, + model: { providerID: "opencode", id: "big-pickle", variant: "none" }, + }, + "stopped", + ), + ], + { replayGate: gate }, + ); const watch = yield* watchNested(runtime); yield* runtime.startTurn(withLineage(thread)); yield* watch.until(() => watch.terminals.length === 1); @@ -1448,12 +1520,17 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { assert.deepEqual(watch.terminals, ["completed", "completed"]); assert.equal(watch.deep.at(-1), "running"); assert.isTrue(yield* runtime.hasPendingBackgroundWork!); + assert.strictEqual(watch.pendingSnapshots.get(MIDDLE), true); yield* runtime.interruptTurn({ providerThread: thread, providerTurnId: yield* providerTurnId, requestRuntimeRestart: true, }); - yield* watch.until(() => watch.deep.at(-1) === "interrupted"); + gate.release("stopped"); + yield* watch.until(() => watch.rootVariant === "none"); + assert.equal(watch.deep.at(-1), "interrupted"); + assert.isFalse(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + assert.strictEqual(watch.pendingSnapshots.get(MIDDLE), false); }).pipe(Effect.scoped), ); diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index d9a0f2fc95e59..d471f0e2f05d0 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -91,6 +91,8 @@ export const ProviderAdapterV2Event = Schema.Union([ type: Schema.Literal("provider_thread.updated"), driver: ProviderDriverKind, providerThread: OrchestrationV2ProviderThread, + /** Work remaining at this point in the event stream, including native continuations. */ + pendingBackgroundWork: Schema.optional(Schema.Boolean), retainedNativeTurnIds: Schema.optional(Schema.Array(Schema.String)), }), Schema.Struct({ diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index c4cc88ec7bfef..03865cc55e1e5 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -257,6 +257,7 @@ interface OpenBlock { interface ThreadState { readonly sessionId: string; providerThread: OrchestrationV2ProviderThread; + publishedBackgroundWork?: boolean; readonly providerTurns: Map; active: ActiveTurn | undefined; model: ModelRef | undefined; @@ -821,6 +822,17 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const hasBackground = (thread: ThreadState) => sessionsOf(thread).some(owesWork); + const emitProviderThread = Effect.fnUntraced(function* (state: ThreadState) { + const pendingBackgroundWork = hasBackground(state); + state.publishedBackgroundWork = pendingBackgroundWork; + yield* emit({ + type: "provider_thread.updated", + driver, + providerThread: state.providerThread, + pendingBackgroundWork, + }); + }); + const setSessionStatus = ( status: OrchestrationV2ProviderSession["status"], lastError: string | null, @@ -1163,7 +1175,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr result: call.result, }, }); - yield* syncRoster(rootOf(call.state)); + yield* syncRosterAncestors(call.state); }); const syncRoster = Effect.fnUntraced(function* (state: ThreadState) { @@ -1179,6 +1191,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr })); const current = state.providerThread.pendingBackgroundTasks ?? []; if ( + state.publishedBackgroundWork === hasBackground(state) && current.length === roster.length && current.every( (task, index) => @@ -1195,11 +1208,15 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr pendingBackgroundTasks: roster, updatedAt: yield* DateTime.now, }; - yield* emit({ - type: "provider_thread.updated", - driver, - providerThread: state.providerThread, - }); + yield* emitProviderThread(state); + }); + + const syncRosterAncestors = Effect.fnUntraced(function* (state: ThreadState) { + let current: ThreadState | undefined = state; + while (current !== undefined) { + yield* syncRoster(current); + current = current.subagent?.call.state; + } }); const withReportedModel = ( @@ -1310,11 +1327,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr childOwners.set(childId, rootOf(call.state)); call.child = child; yield* emit({ type: "app_thread.created", driver, appThread }); - yield* emit({ - type: "provider_thread.updated", - driver, - providerThread: child.providerThread, - }); + yield* emitProviderThread(child); yield* emitSubagent(call); // A session called again was not made now, so it may hold the rules of // a mode the thread has left. OpenCode applies a rules change to the @@ -1351,11 +1364,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr status: "idle", updatedAt: call.completedAt, }; - yield* emit({ - type: "provider_thread.updated", - driver, - providerThread: child.providerThread, - }); + yield* emitProviderThread(child); } yield* emitSubagent(call); }); @@ -1556,11 +1565,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr status: threadDisposition === "broken" ? "error" : "idle", updatedAt: completedAt, }; - yield* emit({ - type: "provider_thread.updated", - driver, - providerThread: state.providerThread, - }); + yield* emitProviderThread(state); const anyActive = [...threads.values()].some( (candidate) => candidate.subagent === undefined && candidate.active !== undefined, ); @@ -2322,11 +2327,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const providerThread = withReportedModel(state.providerThread, event.data.model); if (providerThread !== state.providerThread) { state.providerThread = { ...providerThread, updatedAt: yield* DateTime.now }; - yield* emit({ - type: "provider_thread.updated", - driver, - providerThread: state.providerThread, - }); + yield* emitProviderThread(state); } } if ( @@ -2514,10 +2515,12 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr yield* settleCall(call, "interrupted", LOST_BACKGROUND); } } - for (const session of sessionsOf(state)) { + const sessions = sessionsOf(state); + for (const session of sessions) { discardWakes(session, session.wakes.splice(0)); session.reports.clear(); } + for (const session of sessions.toReversed()) yield* syncRoster(session); } // Stopped off the lock this reconcile holds; one still running waits for a Stop. for (const thread of new Set([...strays.values()].map(rootOf))) { @@ -2574,7 +2577,26 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr stream: Stream.Stream, ): Effect.Effect => stream.pipe( - Stream.runForEach((event) => lock.withPermit(handleEvent(event))), + Stream.runForEach((event) => + lock.withPermit( + Effect.gen(function* () { + yield* handleEvent(event); + if ( + event.type === "session.execution.started" || + event.type === "unreadable.execution.started" || + executionEnd(event.type) || + event.type === "unreadable.execution.ended" || + event.type === "session.inbox.enqueued" || + event.type === "session.inbox.delivered" || + event.type === "session.inbox.cancelled" + ) { + const sessionId = sessionOfEvent(event); + const state = sessionId === undefined ? undefined : threads.get(sessionId); + if (state !== undefined) yield* syncRosterAncestors(state); + } + }), + ), + ), Effect.exit, Effect.flatMap(() => Effect.gen(function* () { @@ -3041,11 +3063,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr lastRunOrdinal: turnInput.runOrdinal, updatedAt: startedAt, }; - yield* emit({ - type: "provider_thread.updated", - driver, - providerThread: state.providerThread, - }); + yield* emitProviderThread(state); yield* setSessionStatus("running", null); return turn; }); @@ -3108,6 +3126,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr if (wake?.running !== true) return; state.wakes.pop(); yield* replay(wake); + yield* syncRosterAncestors(state); }); const runWake = Effect.fnUntraced(function* ( @@ -3135,6 +3154,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr } if (wake.running) state.unsettled = false; yield* replay(wake); + yield* syncRosterAncestors(state); }), ); }); @@ -3200,6 +3220,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr } const running = state.wakes.some((wake) => wake.running); discardWakes(state, state.wakes.splice(0)); + yield* syncRosterAncestors(state); if (running) { markUnsettled(state); yield* client.session From 359f713fcad9131bb1beaffd1f423a6ccfce8905 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 23:31:54 +0000 Subject: [PATCH 2/4] fix(opencode): refresh descendant snapshots after wake replay --- .../Adapters/OpenCode2AdapterV2.test.ts | 72 ++++++++++++++++++- .../src/server/v2/adapter.ts | 4 +- 2 files changed, 73 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index bcee8b5034c79..b115dffd887d8 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -74,11 +74,18 @@ const event = ( type: string, data: Record, label?: string, + version = 1, ): ProviderReplayEntry => ({ type: "emit_inbound", frame: { type: "sdk.event", - event: { id: `evt_${type.replaceAll(".", "")}0000`, created: 1, type, data, ...durable }, + event: { + id: `evt_${type.replaceAll(".", "")}0000`, + created: 1, + type, + data, + durable: { ...durable.durable, version }, + }, }, ...(label === undefined ? {} : { label }), }); @@ -1469,6 +1476,69 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }; }); + it.effect("clears a nested report cancelled during a held wake", () => + Effect.gen(function* () { + const offers: Array = []; + const gate = makeProviderReplayGate(["drained"]); + const { runtime, thread } = yield* resumed( + [ + ...backgroundLaunch(CHILD), + event("session.execution.succeeded", { sessionID: SESSION }), + event("session.execution.succeeded", { sessionID: CHILD }), + childReport("msg_report", "CHILD_OK"), + event("session.execution.started", { sessionID: SESSION }), + ...nestedLaunch.slice(3, -2), + event( + "session.tool.success", + { + ...toolOf(SESSION, "call-middle"), + content: [{ type: "text", text: "MIDDLE_OK" }], + metadata: { sessionID: MIDDLE, status: "completed", truncated: false }, + executed: false, + }, + undefined, + 2, + ), + event("session.execution.succeeded", { sessionID: DEEP }), + deepReport("completed"), + event("session.inbox.cancelled", { sessionID: MIDDLE, inboxID: "msg_deep_report" }), + event("session.execution.succeeded", { sessionID: SESSION }), + event("session.model.selected", { + sessionID: SESSION, + model: { providerID: "opencode", id: "big-pickle", variant: "thinking" }, + }), + event( + "session.model.selected", + { + sessionID: SESSION, + model: { providerID: "opencode", id: "big-pickle", variant: "none" }, + }, + "drained", + ), + ], + { replayGate: gate }, + ).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: (request) => + Effect.sync(() => { + offers.push(request); + }), + take: Effect.never, + }), + ); + const watch = yield* watchNested(runtime); + yield* runtime.startTurn(withLineage(thread)); + assert.isTrue(yield* Effect.promise(() => gate.waitForReached("drained"))); + yield* watch.until(() => watch.rootVariant === "thinking"); + assert.lengthOf(offers, 1); + yield* runtime.startTurn(continuationTurn(thread, offers[0]!)); + gate.release("drained"); + yield* watch.until(() => watch.rootVariant === "none"); + assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + assert.strictEqual(watch.pendingSnapshots.get(MIDDLE), false); + }).pipe(Effect.scoped), + ); + it.effect("stops a foreground subagent's background subagent after the turn ended", () => Effect.gen(function* () { const offered = yield* Deferred.make(); diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index 099c258d9d9da..bfb1f812cccc5 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -3156,7 +3156,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr if (wake?.running !== true) return; state.wakes.pop(); yield* replay(wake); - yield* syncRosterAncestors(state); + for (const session of sessionsOf(state).toReversed()) yield* syncRoster(session); }); const runWake = Effect.fnUntraced(function* ( @@ -3184,7 +3184,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr } if (wake.running) state.unsettled = false; yield* replay(wake); - yield* syncRosterAncestors(state); + for (const session of sessionsOf(state).toReversed()) yield* syncRoster(session); }), ); }); From 4cc74410d2b2433b50a1e02c1b725c901941557c Mon Sep 17 00:00:00 2001 From: Issue triage verification Date: Thu, 22 Oct 2026 10:45:32 +0000 Subject: [PATCH 3/4] fix(opencode): refresh ancestor snapshots after wake replay --- .../Adapters/OpenCode2AdapterV2.test.ts | 91 ++++++++++++++++++- .../src/server/v2/adapter.ts | 2 + 2 files changed, 90 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 703731d3202f7..7bae7e1c69828 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -1221,6 +1221,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { const deep: Array = []; const middleTurns = new Map(); const pendingSnapshots = new Map(); + const variants = new Map(); let rootVariant: unknown; const terminals: Array = []; const changed = yield* Deferred.make(); @@ -1234,9 +1235,10 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { if (nativeId != null && event.pendingBackgroundWork !== undefined) { pendingSnapshots.set(nativeId, event.pendingBackgroundWork); } - if (nativeId === SESSION) - rootVariant = - event.providerThread.nativeMetadata?.modelSelection?.options?.[0]?.value; + const variant = + event.providerThread.nativeMetadata?.modelSelection?.options?.[0]?.value; + if (nativeId != null) variants.set(nativeId, variant); + if (nativeId === SESSION) rootVariant = variant; } if (event.type === "subagent.updated" && event.subagent.title === "Deep") { deep.push(event.subagent.status); @@ -1267,6 +1269,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { middleTurns, terminals, pendingSnapshots, + variants, get rootVariant() { return rootVariant; }, @@ -1337,6 +1340,88 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + it.effect("clears the caller's pending work when a called session's held wake runs", () => + Effect.gen(function* () { + const OTHER = "ses_other00000000000000000000"; + const offers: Array = []; + const gate = makeProviderReplayGate(["drained"]); + const { runtime, thread } = yield* resumed( + [ + out("session.get", { sessionID: OTHER }), + replyData("session.get", sessionInfo({ id: OTHER })), + out("permission.list", { sessionID: OTHER }), + replyData("permission.list", []), + out("session.form.list", { sessionID: OTHER }), + replyData("session.form.list", []), + event("session.execution.started", { sessionID: OTHER }), + event("session.execution.succeeded", { sessionID: OTHER }), + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + event("session.execution.started", { sessionID: SESSION }), + event("session.tool.input.started", { + ...toolOf(SESSION, "call-other"), + name: "subagent", + }), + event("session.tool.called", { + ...toolOf(SESSION, "call-other"), + name: "subagent", + input: { description: "Other", prompt: "again" }, + executed: false, + }), + event("session.tool.progress", { + ...toolOf(SESSION, "call-other"), + metadata: { sessionID: OTHER, status: "running" }, + }), + event("session.tool.success", { + ...toolOf(SESSION, "call-other"), + content: [{ type: "text", text: "OTHER_OK" }], + metadata: { sessionID: OTHER, status: "completed", truncated: false }, + executed: false, + }), + event("session.execution.succeeded", { sessionID: SESSION }), + event( + "session.model.selected", + { + sessionID: OTHER, + model: { providerID: "opencode", id: "big-pickle", variant: "none" }, + }, + "drained", + ), + ], + { replayGate: gate }, + ).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: (request) => + Effect.sync(() => { + offers.push(request); + }), + take: Effect.never, + }), + ); + const other = yield* runtime.resumeThread({ + providerThread: { + ...thread, + id: ProviderThreadId.make("provider-thread:opencode2-adapter:other"), + nativeThreadRef: { driver: OPENCODE_PROVIDER, nativeId: OTHER, strength: "strong" }, + }, + threadId, + modelSelection: bigPickle, + runtimePolicy: policy(), + }); + const watch = yield* watchNested(runtime); + yield* watch.until(() => offers.length === 1); + yield* runtime.startTurn(withLineage(thread)); + assert.isTrue(yield* Effect.promise(() => gate.waitForReached("drained"))); + // The called session's held wake is its caller's pending work too. + yield* watch.until(() => watch.pendingSnapshots.get(SESSION) === true); + yield* runtime.startTurn(continuationTurn(other, offers[0]!)); + gate.release("drained"); + yield* watch.until(() => watch.variants.get(OTHER) === "none"); + assert.isFalse(watch.pendingSnapshots.get(OTHER)); + assert.isFalse(watch.pendingSnapshots.get(SESSION)); + }).pipe(Effect.scoped), + ); + it.effect("stops a foreground subagent's background subagent after the turn ended", () => Effect.gen(function* () { const offered = yield* Deferred.make(); diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index 3c79f8605c73e..29ed0eed0fbca 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -3284,6 +3284,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr state.wakes.pop(); yield* replay(wake); for (const session of sessionsOf(state).toReversed()) yield* syncRoster(session); + yield* syncRosterAncestors(state); }); const runWake = Effect.fnUntraced(function* ( @@ -3312,6 +3313,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr if (wake.running) state.unsettled = false; yield* replay(wake); for (const session of sessionsOf(state).toReversed()) yield* syncRoster(session); + yield* syncRosterAncestors(state); }), ); }); From 9b97de29a6bf8bbe3d10c29c6fb6eb80e129f37c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 22 Oct 2026 11:37:52 +0000 Subject: [PATCH 4/4] test(opencode): add pending-snapshot evidence driver for #19228 Replays recorded OpenCode 2 fixtures through OrchestratorV2 and records the adapter's provider_thread.updated snapshots per native thread, since pendingBackgroundWork does not reach domain events at this head. Co-Authored-By: Claude Opus 5.5 --- ...OpenCode2PendingSnapshots.evidence.test.ts | 159 ++++++++++++++++++ 1 file changed, 159 insertions(+) create mode 100644 apps/server/src/orchestration-v2/testkit/OpenCode2PendingSnapshots.evidence.test.ts diff --git a/apps/server/src/orchestration-v2/testkit/OpenCode2PendingSnapshots.evidence.test.ts b/apps/server/src/orchestration-v2/testkit/OpenCode2PendingSnapshots.evidence.test.ts new file mode 100644 index 0000000000000..1d7477cc37dc9 --- /dev/null +++ b/apps/server/src/orchestration-v2/testkit/OpenCode2PendingSnapshots.evidence.test.ts @@ -0,0 +1,159 @@ +import { assert, describe, it } from "@effect/vitest"; +import type { ProviderInstanceId } from "@t3tools/contracts"; +import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as Config from "effect/Config"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Stream from "effect/Stream"; + +import { OpenCode2OrchestratorReplayHarness } from "../Adapters/OpenCode2AdapterV2.testkit.ts"; +import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts"; +import { findReplayFixture, runReplayFixture } from "./ProviderReplayFixtures.ts"; + +// Evidence driver for #19228: replays recorded OpenCode 2 fixtures through +// OrchestratorV2 and records the adapter's own event stream, because the +// `pendingBackgroundWork` flag is dropped before it reaches domain events. +// Set EV_TRANSCRIPT to a file path to append the transcript there. + +type Row = { + readonly id: string; + readonly fixture: string; + /** True when every native thread must end on a `pendingBackgroundWork: false` snapshot. */ + readonly requireSettledSnapshots: boolean; +}; + +const ROWS: ReadonlyArray = [ + { id: "R1", fixture: "opencode2_nested_background", requireSettledSnapshots: true }, + { id: "R1-compat", fixture: "opencode2_background", requireSettledSnapshots: true }, + { id: "R2c-compat", fixture: "opencode2_resume_after_restart", requireSettledSnapshots: false }, + { id: "R2c-compat", fixture: "opencode2_interrupt", requireSettledSnapshots: false }, + { id: "R3", fixture: "opencode2_simple", requireSettledSnapshots: false }, + { id: "R3", fixture: "opencode2_tool_call", requireSettledSnapshots: false }, +]; + +const tasksOf = (tasks: ReadonlyArray<{ readonly taskId: string }> | undefined) => + `[${(tasks ?? []).map((task) => task.taskId).join(",")}]`; + +const describeEvent = (event: ProviderAdapter.ProviderAdapterV2Event): string | undefined => { + switch (event.type) { + case "provider_thread.updated": + return `provider_thread.updated native=${event.providerThread.nativeThreadRef?.nativeId ?? "?"} pendingBackgroundWork=${String(event.pendingBackgroundWork)} pendingBackgroundTasks=${tasksOf(event.providerThread.pendingBackgroundTasks)}`; + case "provider_turn.updated": + return `provider_turn.updated native=${event.providerTurn.nativeTurnRef?.nativeId ?? "?"} status=${event.providerTurn.status}`; + case "subagent.updated": + return `subagent.updated title=${event.subagent.title ?? "?"} status=${event.subagent.status}`; + case "turn.terminal": + return `turn.terminal status=${event.status}`; + default: + return undefined; + } +}; + +const tappedHarness = ( + record: (event: ProviderAdapter.ProviderAdapterV2Event) => void, +): typeof OpenCode2OrchestratorReplayHarness => ({ + ...OpenCode2OrchestratorReplayHarness, + makeProviderAdapterRegistryLayer: ( + ...args: Parameters + ) => + Layer.effect( + ProviderAdapterRegistry.ProviderAdapterRegistryV2, + Effect.gen(function* () { + const registry = yield* ProviderAdapterRegistry.ProviderAdapterRegistryV2; + return ProviderAdapterRegistry.ProviderAdapterRegistryV2.of({ + ...registry, + get: (instanceId: ProviderInstanceId) => + registry.get(instanceId).pipe( + Effect.map((adapter) => + ProviderAdapter.ProviderAdapterV2.of({ + ...adapter, + openSession: (input) => + adapter.openSession(input).pipe( + Effect.map((runtime) => ({ + ...runtime, + events: runtime.events.pipe( + Stream.tap((event) => Effect.sync(() => record(event))), + ), + })), + ), + }), + ), + ), + }); + }), + ).pipe( + Layer.provide(OpenCode2OrchestratorReplayHarness.makeProviderAdapterRegistryLayer(...args)), + ), +}); + +const runRow = (row: Row) => + Effect.gen(function* () { + const fixture = findReplayFixture(row.fixture); + const driver = fixture?.providers.find((provider) => provider.driver === "opencode"); + assert.isDefined(fixture); + assert.isDefined(driver); + const events: Array = []; + const lines: Array = []; + const emit = (line: string) => lines.push(line); + emit(`== ${row.id} ${row.fixture} (recorded fixture) ==`); + const exit = yield* runReplayFixture({ + fixtureName: row.fixture, + buildInput: fixture.buildInput, + driver, + harness: tappedHarness((event) => events.push(event)), + }).pipe(Effect.exit); + let index = 0; + for (const event of events) { + const line = describeEvent(event); + if (line !== undefined) emit(` adapter[${index}] ${line}`); + index += 1; + } + const lastSnapshot = new Map(); + let everPending = false; + for (const event of events) { + if (event.type !== "provider_thread.updated") continue; + const nativeId = event.providerThread.nativeThreadRef?.nativeId ?? "?"; + lastSnapshot.set(nativeId, event.pendingBackgroundWork); + if (event.pendingBackgroundWork === true) everPending = true; + } + const finals = [...lastSnapshot].map(([id, flag]) => `${id}=${String(flag)}`).join(" "); + emit(` final snapshot per native thread: ${finals}`); + if (exit._tag === "Success") { + for (const domain of exit.value.domainEvents) { + if (domain.type !== "provider-thread.updated") continue; + emit( + ` domain provider-thread.updated native=${domain.payload.nativeThreadRef?.nativeId ?? "?"} pendingBackgroundTasks=${tasksOf(domain.payload.pendingBackgroundTasks)}`, + ); + } + } + const failures: Array = []; + if (exit._tag === "Failure") failures.push("fixture output assertions failed"); + if (row.requireSettledSnapshots) { + if (!everPending) failures.push("no pendingBackgroundWork=true snapshot was published"); + for (const [id, flag] of lastSnapshot) { + if (flag !== false) failures.push(`${id} ended on pendingBackgroundWork=${String(flag)}`); + } + } else if ([...lastSnapshot.values()].includes(true)) { + failures.push("a native thread ended with pending work"); + } + emit( + ` ${row.id} ${row.fixture}: ${failures.length === 0 ? "PASS" : `FAIL (${failures.join("; ")})`}`, + ); + const transcript = yield* Config.String("EV_TRANSCRIPT").pipe(Config.withDefault("")); + if (transcript !== "") { + const fs = yield* FileSystem.FileSystem; + yield* fs.writeFileString(transcript, `${lines.join("\n")}\n`, { flag: "a" }); + } + return failures; + }).pipe(Effect.provide(NodeServices.layer)); + +describe("OpenCode 2 pending-work snapshots (#19228 evidence)", () => { + it.effect.each(ROWS.map((row) => [`${row.id} ${row.fixture}`, row] as const))("%s", ([, row]) => + runRow(row).pipe( + Effect.map((failures) => assert.deepEqual(failures, [])), + Effect.scoped, + ), + ); +});