# Files changed · PR \#21461 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/21461/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [fix\(server\): drain queued native child continuations](https://git.cafe/pingdotgg/t3code/pull/21461.md) Version: 18 Head commit: 1b052a2d9df4a7c4772ec5c690c67c234d9fa008 Base commit: 4cc74410d2b2433b50a1e02c1b725c901941557c Comparison base: 4cc74410d2b2433b50a1e02c1b725c901941557c Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/server/src/orchestration-v2/RunExecutionService.test.ts Status: modified; +264; −3 ```diff @@ -11,9 +11,11 @@ type OrchestrationV2DomainEvent, type OrchestrationV2ExecutionNode, type OrchestrationV2ProviderThread, + type OrchestrationV2ProviderTurn, type OrchestrationV2Run, type OrchestrationV2RunAttempt, type OrchestrationV2Subagent, + type OrchestrationV2StoredEvent, type OrchestrationV2TurnItem, ProviderDriverKind, ProviderInstanceId, @@ -1747,6 +1749,97 @@ }), ); +it.effect.each([false, true])( + "drains queued child output with an idle adapter and root ownership lost=%s", + (rootOwnershipLost) => + Effect.gen(function* () { + const childEvents: Array = []; + yield* runBackgroundItemScenario( + "queued-child-output", + (ids) => { + const childProviderThreadId = ProviderThreadId.make( + "provider-thread:queued-child-output:child", + ); + const snapshot = (child: boolean, pendingBackgroundWork: boolean) => ({ + 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", + firstRunOrdinal: 1, + lastRunOrdinal: 1, + handoffIds: [], + forkedFrom: null, + pendingBackgroundTasks: [], + 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:queued-child-output:child"), + providerThreadId: childProviderThreadId, + runAttemptId: null, + ordinal: 2, + status, + } as OrchestrationV2ProviderTurn, + }); + const message = (text: string) => + ({ + type: "message.updated", + driver, + message: { threadId: ids.childThreadId, runId: null, text }, + }) as ProviderAdapter.ProviderAdapterV2Event; + return [ + childThreadCreatedEvent(ids), + snapshot(false, true), + snapshot(true, true), + subagentEvent(ids, "running"), + rootTerminalEvent(ids, "completed"), + snapshot(false, true), + subagentEvent(ids, "completed"), + childTurn("running"), + message("GRANDCHILD_OK"), + childTurn("completed"), + snapshot(true, false), + snapshot(false, false), + message("after-drain"), + ]; + }, + { + keepEventStreamOpen: true, + hasPendingBackgroundWorkForThread: () => Effect.succeed(false), + onIngest: (input) => + Effect.sync(() => { + const event = input.event; + if (event.type === "provider_turn.updated") + childEvents.push(event.providerTurn.status); + if (event.type === "message.updated") childEvents.push(event.message.text); + if (event.type !== "provider_thread.updated") return []; + if (rootOwnershipLost && input.writeIfProviderThreadOwner !== undefined) return []; + return [ + { + event: { type: "provider-thread.updated", payload: event.providerThread }, + } as OrchestrationV2StoredEvent, + ]; + }), + }, + ); + assert.deepEqual(childEvents, ["running", "GRANDCHILD_OK", "completed"]); + }), +); + it.effect("keeps ingesting a child thread's late background item completion", () => Effect.gen(function* () { const observed = yield* runBackgroundItemScenario("bg-child-item", (ids) => [ @@ -1826,6 +1919,168 @@ }), ); +it.effect.each([false, true])( + "drops a transferred child's pending snapshot (independent sibling pending=%s)", + (siblingPending) => + Effect.gen(function* () { + const consumed = yield* Deferred.make(); + const result = yield* Effect.raceFirst( + runBackgroundItemScenario( + "child-transfer-pending", + (ids) => { + const childProviderThreadId = ProviderThreadId.make( + "provider-thread:child-transfer-pending:child", + ); + const siblingThreadId = ThreadId.make("thread:child-transfer-pending:sibling"); + const snapshot = ( + providerThreadId: ProviderThreadId, + appThreadId: ThreadId, + pendingBackgroundWork: boolean, + ) => ({ + type: "provider_thread.updated" as const, + driver, + providerThread: { + id: providerThreadId, + appThreadId, + status: "idle", + } as OrchestrationV2ProviderThread, + pendingBackgroundWork, + }); + const running = subagentEvent(ids, "running"); + if (running.type !== "subagent.updated") throw new Error("Missing subagent"); + const linked = { + ...running, + subagent: { + ...running.subagent, + origin: "provider_native" as const, + childThreadId: ids.childThreadId, + providerThreadId: childProviderThreadId, + }, + }; + const sibling = siblingPending + ? [ + { + ...childThreadCreatedEvent(ids), + appThread: { + id: siblingThreadId, + lineage: { + parentThreadId: ids.threadId, + relationshipToParent: "subagent", + rootThreadId: ids.threadId, + }, + }, + } as ProviderAdapter.ProviderAdapterV2Event, + snapshot( + ProviderThreadId.make("provider-thread:child-transfer-pending:sibling"), + siblingThreadId, + true, + ), + ] + : []; + return [ + childThreadCreatedEvent(ids), + linked, + snapshot(childProviderThreadId, ids.childThreadId, true), + ...sibling, + rootTerminalEvent(ids, "completed"), + { ...linked, subagent: { ...linked.subagent, runId: RunId.make("run:replacement") } }, + snapshot(childProviderThreadId, ids.childThreadId, false), + ]; + }, + { + keepEventStreamOpen: true, + hasPendingBackgroundWorkForThread: () => Effect.succeed(false), + onEventsConsumed: Deferred.succeed(consumed, undefined).pipe(Effect.asVoid), + }, + ).pipe(Effect.map(Option.some)), + Deferred.await(consumed).pipe(Effect.as(Option.none())), + ); + assert.strictEqual( + Option.isSome(result), + !siblingPending, + "only the sibling's pending work may keep the prior subscription open", + ); + }), +); + +it.effect.each([true, false])( + "keeps a moved child's nested pending work on the prior subscription (grandchild pending=%s)", + (grandchildPending) => + Effect.gen(function* () { + const consumed = yield* Deferred.make(); + const result = yield* Effect.raceFirst( + runBackgroundItemScenario( + "nested-transfer-pending", + (ids) => { + const childProviderThreadId = ProviderThreadId.make( + "provider-thread:nested-transfer-pending:child", + ); + const grandchildThreadId = ThreadId.make("thread:nested-transfer-pending:grandchild"); + const grandchildProviderThreadId = ProviderThreadId.make( + "provider-thread:nested-transfer-pending:grandchild", + ); + const snapshot = ( + providerThreadId: ProviderThreadId, + appThreadId: ThreadId, + pendingBackgroundWork: boolean, + ) => ({ + type: "provider_thread.updated" as const, + driver, + providerThread: { + id: providerThreadId, + appThreadId, + status: "idle", + } as OrchestrationV2ProviderThread, + pendingBackgroundWork, + }); + const running = subagentEvent(ids, "running"); + if (running.type !== "subagent.updated") throw new Error("Missing subagent"); + const linked = { + ...running, + subagent: { + ...running.subagent, + origin: "provider_native" as const, + childThreadId: ids.childThreadId, + providerThreadId: childProviderThreadId, + }, + }; + return [ + childThreadCreatedEvent(ids), + linked, + snapshot(childProviderThreadId, ids.childThreadId, true), + { + ...childThreadCreatedEvent(ids), + appThread: { + id: grandchildThreadId, + lineage: { + parentThreadId: ids.childThreadId, + relationshipToParent: "subagent", + rootThreadId: ids.threadId, + }, + }, + } as ProviderAdapter.ProviderAdapterV2Event, + snapshot(grandchildProviderThreadId, grandchildThreadId, true), + rootTerminalEvent(ids, "completed"), + { ...linked, subagent: { ...linked.subagent, runId: RunId.make("run:replacement") } }, + snapshot(grandchildProviderThreadId, grandchildThreadId, grandchildPending), + ]; + }, + { + keepEventStreamOpen: true, + hasPendingBackgroundWorkForThread: () => Effect.succeed(false), + onEventsConsumed: Deferred.succeed(consumed, undefined).pipe(Effect.asVoid), + }, + ).pipe(Effect.map(Option.some)), + Deferred.await(consumed).pipe(Effect.as(Option.none())), + ); + assert.strictEqual( + Option.isSome(result), + !grandchildPending, + "retained descendant work keeps the prior subscription open until its snapshot clears", + ); + }), +); + it.effect("does not pin ingestion on background items when the root turn is interrupted", () => Effect.gen(function* () { const observed = yield* runBackgroundItemScenario("bg-interrupted", (ids) => [ @@ -3251,6 +3506,10 @@ >; readonly onSubscribe?: Effect.Effect; readonly onEventsConsumed?: Effect.Effect; + readonly onIngest?: ProviderEventIngestor.ProviderEventIngestorV2["Service"]["ingestNormalized"]; + readonly hasPendingBackgroundWorkForThread?: ( + thread: OrchestrationV2ProviderThread, + ) => Effect.Effect; readonly onFinalized?: ( events: ReadonlyArray, ) => Effect.Effect; @@ -3300,7 +3559,7 @@ `subagent:${event.subagent.status}`, ]); } - return []; + return yield* options?.onIngest?.(input) ?? Effect.succeed([]); }), }), ServerSettings.layerTest(), @@ -3316,6 +3575,9 @@ 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)).pipe( @@ -3380,8 +3642,7 @@ }); }).pipe(Effect.provide(layerTest)); - const closed = yield* Deferred.await(ingestionDone).pipe(Effect.timeoutOption("2 seconds")); - assert.isTrue(Option.isSome(closed), "event ingestion fiber did not finish"); + yield* Deferred.await(ingestionDone); return yield* Ref.get(observed); }); } ``` ## apps/server/src/orchestration-v2/RunExecutionService.ts Status: modified; +17; −0 ```diff @@ -981,6 +981,7 @@ const terminalEvent = yield* Ref.make(null); const latestTurnItemOrdinal = yield* Ref.make(input.providerTurnOrdinal * 100); const latestProviderThread = yield* Ref.make(input.providerThread); + const pendingBackgroundWork = new Map(); const latestRootProviderTurn = yield* Ref.make(null); const routeIdentity: ProviderEventRouteIdentity = { threadId: input.run.threadId, @@ -1238,9 +1239,16 @@ if (backgroundItems.size > 0) { return false; } + // Live probes can run ahead of queued events. Consumed child snapshots + // retain native continuations between their terminal and next turn. + for (const [providerThreadId, pending] of pendingBackgroundWork) { + if (providerThreadId !== input.providerThread.id && pending) return false; + } if (yield* Ref.get(providerThreadOwnerLost)) { return true; } + const pendingRoot = pendingBackgroundWork.get(input.providerThread.id); + if (pendingRoot !== undefined) return !pendingRoot; // Claude background Bash has no turn-item projection. Use only the // thread-scoped probe: session-wide pending work (siblings, wake // buffers, session subagents) must not pin this root subscription. @@ -1331,6 +1339,12 @@ activeBackgroundTurnItems, (current) => new Set([...current].filter((id) => !detached.itemIds.has(id))), ); + // Routing now rejects the moved thread's snapshots, so none would clear its entry. + for (const providerThreadId of pendingBackgroundWork.keys()) { + if (!routing.ownedProviderThreadIds.has(providerThreadId)) { + pendingBackgroundWork.delete(providerThreadId); + } + } } return { event, accepted }; }), @@ -1409,6 +1423,9 @@ reservedHeldTurnItemIds.add(event.turnItem.id); } if (event.type === "provider_thread.updated") { + if (event.pendingBackgroundWork !== undefined) { + pendingBackgroundWork.set(event.providerThread.id, event.pendingBackgroundWork); + } if (event.providerThread.id === input.providerThread.id && storedEventCount > 0) { yield* Ref.set(latestProviderThread, event.providerThread); } ```