# Files changed · PR \#19388 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/19388/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [fix\(opencode\): retain queued native child continuation output](https://git.cafe/pingdotgg/t3code/pull/19388.md) Version: 1 Head commit: 2f388703d0bd7d5342c21d51f68fe5431d3f6d40 Base commit: f5b5f9dd7be29c4f55f4687dda15e4105cf966fc Comparison base: 519d64f6cec0a0f9ed18fbca4a3f8695cac44506 Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts Status: modified; +17; −3 ```diff @@ -31,6 +31,7 @@ import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; import * as References from "effect/References"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; @@ -975,11 +976,23 @@ event("session.usage.updated", { sessionID: SESSION }), ]); const reported = yield* Deferred.make(); + const pendingSnapshots = yield* Ref.make>([]); yield* runtime.events.pipe( Stream.tap((event) => - event.type === "subagent.updated" && event.subagent.status === "completed" - ? Deferred.succeed(reported, undefined) - : Effect.void, + Effect.gen(function* () { + if ( + event.type === "provider_thread.updated" && + event.pendingBackgroundWork !== undefined + ) { + yield* Ref.update(pendingSnapshots, (current) => [ + ...current, + event.pendingBackgroundWork!, + ]); + } + if (event.type === "subagent.updated" && event.subagent.status === "completed") { + yield* Deferred.succeed(reported, undefined); + } + }), ), Stream.runDrain, Effect.forkScoped, @@ -989,6 +1002,7 @@ // The follow-up execution OpenCode will start for the report is still to come. assert.isTrue(yield* runtime.hasPendingBackgroundWork!); assert.isTrue(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + assert.isTrue((yield* Ref.get(pendingSnapshots)).includes(true)); }).pipe(Effect.scoped), ); ``` ## apps/server/src/orchestration-v2/RunExecutionService.test.ts Status: modified; +110; −0 ```diff @@ -12,9 +12,11 @@ type OrchestrationV2DomainEvent, type OrchestrationV2ExecutionNode, type OrchestrationV2ProviderThread, + type OrchestrationV2ProviderTurn, type OrchestrationV2Run, type OrchestrationV2RunAttempt, type OrchestrationV2Subagent, + type OrchestrationV2StoredEvent, type OrchestrationV2TurnItem, ProviderDriverKind, ProviderInstanceId, @@ -1687,6 +1689,91 @@ }), ); +it.effect.each([false, true])( + "drains an owned child continuation with an idle adapter and root ownership lost=%s", + (rootOwnershipLost) => + Effect.gen(function* () { + const childEvents = yield* Ref.make>([]); + yield* runBackgroundItemScenario( + "ordered-background-continuation", + (ids) => { + const childProviderThreadId = ProviderThreadId.make( + "provider-thread:ordered-background-continuation:child", + ); + const snapshot = (pendingBackgroundWork: boolean, populated: boolean, child = false) => ({ + type: "provider_thread.updated" as const, + driver, + providerThread: { + id: child ? childProviderThreadId : ids.providerThreadId, + driver, + providerInstanceId: ProviderInstanceId.make("codex"), + providerSessionId: null, + appThreadId: child ? ids.childThreadId : ids.threadId, + ownerNodeId: null, + nativeThreadRef: null, + nativeConversationHeadRef: null, + status: "idle" as const, + firstRunOrdinal: 1, + lastRunOrdinal: 1, + handoffIds: [], + forkedFrom: null, + pendingBackgroundTasks: populated + ? [{ taskId: "child", kind: "subagent" as const, childThreadId: ids.childThreadId }] + : [], + createdAt: DateTime.makeUnsafe(0), + updatedAt: DateTime.makeUnsafe(0), + } satisfies OrchestrationV2ProviderThread, + pendingBackgroundWork, + }); + const childTurn = (status: "running" | "completed") => ({ + type: "provider_turn.updated" as const, + driver, + threadId: ids.childThreadId, + providerTurn: { + id: ProviderTurnId.make("provider-turn:ordered-background-continuation:child"), + providerThreadId: childProviderThreadId, + runAttemptId: null, + ordinal: 2, + status, + } as OrchestrationV2ProviderTurn, + }); + return [ + childThreadCreatedEvent(ids), + snapshot(true, true), + snapshot(true, true, true), + subagentEvent(ids, "running"), + rootTerminalEvent(ids, "completed"), + subagentEvent(ids, "completed"), + snapshot(true, false), + childTurn("running"), + { + type: "message.updated", + driver, + message: { threadId: ids.childThreadId, runId: null, text: "GRANDCHILD_OK" }, + } as ProviderAdapter.ProviderAdapterV2Event, + childTurn("completed"), + snapshot(false, false, true), + snapshot(rootOwnershipLost, false), + ]; + }, + { + keepEventStreamOpen: true, + hasPendingBackgroundWorkForThread: () => Effect.succeed(false), + rejectRootProviderThreadUpdates: rootOwnershipLost, + onIngest: (event) => + Ref.update(childEvents, (events) => + event.type === "provider_turn.updated" + ? [...events, event.providerTurn.status] + : event.type === "message.updated" + ? [...events, event.message.text] + : events, + ), + }, + ); + assert.deepEqual(yield* Ref.get(childEvents), ["running", "GRANDCHILD_OK", "completed"]); + }), +); + it.effect("keeps ingesting a late background command item completion after root terminal", () => Effect.gen(function* () { const observed = yield* runBackgroundItemScenario("bg-command", (ids) => [ @@ -4331,6 +4418,11 @@ ReadonlyArray<{ readonly id: TurnItemId; readonly runId: RunId }> >; readonly onSubscribe?: Effect.Effect; + readonly onIngest?: (event: ProviderAdapter.ProviderAdapterV2Event) => Effect.Effect; + readonly hasPendingBackgroundWorkForThread?: ( + thread: OrchestrationV2ProviderThread, + ) => Effect.Effect; + readonly rejectRootProviderThreadUpdates?: boolean; }, ) { return Effect.gen(function* () { @@ -4363,6 +4455,7 @@ ingestNormalized: (input) => Effect.gen(function* () { const event = input.event; + yield* options?.onIngest?.(event) ?? Effect.void; if (event.type === "turn_item.updated") { yield* Ref.update(observed, (current) => [ ...current, @@ -4375,6 +4468,18 @@ `subagent:${event.subagent.status}`, ]); } + if (event.type === "provider_thread.updated") { + if ( + options?.rejectRootProviderThreadUpdates === true && + input.writeIfProviderThreadOwner !== undefined + ) + return []; + return [ + { + event: { type: "provider-thread.updated", payload: event.providerThread }, + } as OrchestrationV2StoredEvent, + ]; + } return []; }), }), @@ -4391,6 +4496,11 @@ providerSessionId: ProviderSessionId.make(`session:${key}`), session: { events: Stream.empty, + ...(options?.hasPendingBackgroundWorkForThread === undefined + ? {} + : { + hasPendingBackgroundWorkForThread: options.hasPendingBackgroundWorkForThread, + }), subscribeEvents: Effect.gen(function* () { yield* options?.onSubscribe ?? Effect.void; const events = Stream.fromIterable(makeEvents(ids)); ``` ## apps/server/src/orchestration-v2/RunExecutionService.ts Status: modified; +21; −4 ```diff @@ -934,6 +934,9 @@ const terminalEvent = yield* Ref.make(null); const latestTurnItemOrdinal = yield* Ref.make(input.providerTurnOrdinal * 100); const latestProviderThread = yield* Ref.make(input.providerThread); + const pendingBackgroundWork = yield* Ref.make>( + new Map(), + ); const routeIdentity: ProviderEventRouteIdentity = { threadId: input.run.threadId, runId: input.run.id, @@ -1200,11 +1203,17 @@ if (backgroundItems.size > 0) { return false; } + // The adapter can already have parsed events whose SQLite writes + // are still queued. Owned children may have a continuation between + // turns; their consumed snapshots keep those events in order. + const pendingSnapshots = yield* Ref.get(pendingBackgroundWork); - // Owner loss means do not hold the stream open solely for the - // roster probe; once background sets are empty, release. - if (yield* Ref.get(providerThreadOwnerLost)) { + for (const [providerThreadId, pending] of pendingSnapshots) { + if (providerThreadId !== input.providerThread.id && pending) return false; - return true; } + // A newer root's background work must not pin this subscription. + if (yield* Ref.get(providerThreadOwnerLost)) return true; + const pendingRoot = pendingSnapshots.get(input.providerThread.id); + if (pendingRoot !== undefined) return !pendingRoot; // Claude background Bash has no turn-item projection. Keep the // stream open while this root's provider thread still reports // pending roster work so late empty updates can clear Waiting. @@ -1288,6 +1297,14 @@ } } if (event.type === "provider_thread.updated") { + const pendingSnapshot = event.pendingBackgroundWork; + if (pendingSnapshot !== undefined) { + yield* Ref.update(pendingBackgroundWork, (snapshots) => { + const next = new Map(snapshots); + next.set(event.providerThread.id, pendingSnapshot); + return next; + }); + } if (event.providerThread.id === input.providerThread.id && storedEventCount > 0) { yield* Ref.set(latestProviderThread, event.providerThread); } ``` ## packages/provider-core/src/server/ProviderAdapter.ts Status: modified; +2; −0 ```diff @@ -90,6 +90,8 @@ type: Schema.Literal("provider_thread.updated"), driver: ProviderDriverKind, providerThread: OrchestrationV2ProviderThread, + /** Work remaining at this point in the ordered event stream, including native continuations. */ + pendingBackgroundWork: Schema.optional(Schema.Boolean), /** * Set when the provider rewound its own conversation outside T3, for * example a Pi extension moving the session-tree leaf back. Lists every ``` ## packages/provider-opencode/src/server/v2/adapter.ts Status: modified; +57; −5 ```diff @@ -337,6 +337,7 @@ interface ThreadState { readonly sessionId: string; providerThread: OrchestrationV2ProviderThread; + publishedBackgroundWork?: boolean; readonly providerTurns: Map; active: ActiveTurn | undefined; /** What the native session runs now, so a changed selection is switched before prompting. */ @@ -940,7 +941,18 @@ return sessionGates.withLock(sessionId, effect); }; const emit = (event: ProviderAdapter.ProviderAdapterV2Event) => - Queue.offer(events, event).pipe(Effect.asVoid); + Effect.suspend(() => { + if (event.type === "provider_thread.updated") { + const nativeId = event.providerThread.nativeThreadRef?.nativeId; + const state = nativeId == null ? undefined : threads.get(nativeId); + if (state !== undefined) { + const pendingBackgroundWork = hasBackground(state); + state.publishedBackgroundWork = pendingBackgroundWork; + return Queue.offer(events, { ...event, pendingBackgroundWork }).pipe(Effect.asVoid); + } + } + return Queue.offer(events, event).pipe(Effect.asVoid); + }); const ownerOf = (sessionId: string) => childOwners.get(sessionId) ?? threads.get(sessionId); /** Marks a run T3 ended while OpenCode may still run it (see `unsettled`). */ const markUnsettled = (state: ThreadState) => { @@ -1393,7 +1405,8 @@ /** * Lists a thread's background subagents on its provider thread, as typed * `subagent` work: the thread shows it waits on them after its turn ends, + * and a Stop reaches them. Emits when the roster or remaining native + * continuation work changes, in order with the events that describe it. - * and a Stop reaches them. Emitted only when the list changes. */ const syncRoster = Effect.fnUntraced(function* (state: ThreadState) { const roster = runningCalls(state) @@ -1407,7 +1420,9 @@ : { childThreadId: call.child.subagent.appThread.id }), })); const current = state.providerThread.pendingBackgroundTasks ?? []; + const pendingBackgroundWork = hasBackground(state); if ( + state.publishedBackgroundWork === pendingBackgroundWork && current.length === roster.length && current.every( (task, index) => @@ -1419,6 +1434,7 @@ ) { return; } + state.publishedBackgroundWork = pendingBackgroundWork; state.providerThread = { ...state.providerThread, pendingBackgroundTasks: roster, @@ -1428,9 +1444,18 @@ type: "provider_thread.updated", driver, providerThread: state.providerThread, + pendingBackgroundWork, }); }); + 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 = ( providerThread: OrchestrationV2ProviderThread, model: ModelRef | undefined, @@ -2895,11 +2920,13 @@ yield* settleCall(call, "interrupted", LOST_BACKGROUND); } } - // A subagent's session holds reports of its own background subagents. - for (const session of sessionsOf(state)) { + // Clear every descendant's reports before publishing ancestor snapshots. + 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))) { @@ -2967,7 +2994,30 @@ stream: Stream.Stream, ): Effect.Effect => stream.pipe( + Stream.runForEach((event) => + lock.withPermit( + handleEvent(event).pipe( + Effect.andThen(() => { + const sessionId = sessionOfEvent(event); + const state = sessionId === undefined ? undefined : threads.get(sessionId); + if ( + state === undefined || + !( + 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" + ) + ) + return Effect.void; + return syncRosterAncestors(state); + }), + ), + ), + ), - Stream.runForEach((event) => lock.withPermit(handleEvent(event))), Effect.exit, Effect.flatMap(() => Effect.gen(function* () { @@ -3561,6 +3611,7 @@ if (wake?.running !== true) return; state.wakes.pop(); yield* replay(wake); + yield* syncRosterAncestors(state); }); /** @@ -3598,6 +3649,7 @@ // was over before it started: its end, still to come, is this turn's. if (wake.running) state.unsettled = false; yield* replay(wake); + yield* syncRosterAncestors(state); }), ); }); ```