From c74c689ed46fbf635318e38975f048fea2f9f821 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 08:05:09 +0000 Subject: [PATCH 1/6] fix(server): transfer native child routing between runs --- .../RunExecutionService.test.ts | 135 ++++++++++++++++++ .../orchestration-v2/RunExecutionService.ts | 41 +++++- 2 files changed, 169 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 8cf9a4f28a439..126b52e40a6f3 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -327,6 +327,141 @@ it("routes only enrolled workflow members after the root turn ends", () => { ); }); +it("moves a child thread to the run that shows its subagent now", () => { + const threadId = ThreadId.make("thread:handoff"); + const rootProviderTurnId = ProviderTurnId.make("provider-turn:handoff"); + const identity: RunExecutionService.ProviderEventRouteIdentity = { + threadId, + runId: RunId.make("run:handoff:1"), + attemptId: RunAttemptId.make("attempt:handoff:1"), + providerThreadId: ProviderThreadId.make("provider-thread:handoff"), + }; + const resumedChild = ThreadId.make("thread:handoff:resumed"); + const siblingChild = ThreadId.make("thread:handoff:sibling"); + const providerThreadOf = (childThreadId: ThreadId) => + ProviderThreadId.make(`provider-thread:${childThreadId}`); + const subagent = (childThreadId: ThreadId, runId: RunId): OrchestrationV2Subagent => ({ + id: NodeId.make(`node:${childThreadId}`), + threadId, + runId, + parentNodeId: NodeId.make("node:handoff"), + origin: "provider_native", + createdBy: "agent", + driver, + providerInstanceId: ProviderInstanceId.make("codex"), + providerThreadId: providerThreadOf(childThreadId), + childThreadId, + nativeTaskRef: null, + prompt: "Work", + title: "Work", + model: null, + status: "running", + result: null, + startedAt: null, + completedAt: null, + updatedAt: DateTime.makeUnsafe(0), + }); + const childMessage = (childThreadId: ThreadId) => + ({ + type: "message.updated", + driver, + message: { + id: MessageId.make(`message:${childThreadId}`), + threadId: childThreadId, + runId: null, + }, + }) as ProviderAdapter.ProviderAdapterV2Event; + const childTurn = (childThreadId: ThreadId, ordinal: number) => + ({ + type: "provider_turn.updated", + driver, + providerTurn: { + id: ProviderTurnId.make(`provider-turn:${childThreadId}:${ordinal}`), + providerThreadId: providerThreadOf(childThreadId), + runAttemptId: null, + status: "running", + }, + }) as ProviderAdapter.ProviderAdapterV2Event; + const ended = RunExecutionService.routeProviderEvent( + { + type: "turn.terminal", + driver, + providerThreadId: identity.providerThreadId, + providerTurnId: rootProviderTurnId, + runOrdinal: 1, + status: "completed", + failure: null, + threadDisposition: "reusable", + }, + identity, + RunExecutionService.makeProviderEventRoutingState({ + identity, + providerTurnId: rootProviderTurnId, + relatedThreadIds: [resumedChild, siblingChild], + relatedProviderThreadIds: [providerThreadOf(resumedChild), providerThreadOf(siblingChild)], + }), + )[1]; + const [firstTurnAccepted, withChildTurn] = RunExecutionService.routeProviderEvent( + childTurn(resumedChild, 1), + identity, + ended, + ); + assert.isTrue(firstTurnAccepted); + const [movedAccepted, afterMove] = RunExecutionService.routeProviderEvent( + { + type: "subagent.updated", + driver, + subagent: subagent(resumedChild, RunId.make("run:handoff:2")), + }, + identity, + withChildTurn, + ); + assert.isFalse(movedAccepted); + assert.isFalse( + RunExecutionService.routeProviderEvent(childMessage(resumedChild), identity, afterMove)[0], + ); + assert.isTrue( + RunExecutionService.routeProviderEvent(childMessage(siblingChild), identity, afterMove)[0], + ); + assert.isFalse( + RunExecutionService.routeProviderEvent(childTurn(resumedChild, 1), identity, afterMove)[0], + ); + assert.isFalse( + RunExecutionService.routeProviderEvent(childTurn(resumedChild, 2), identity, afterMove)[0], + ); + assert.isTrue( + RunExecutionService.routeProviderEvent(childTurn(siblingChild, 1), identity, afterMove)[0], + ); + assert.isTrue( + RunExecutionService.routeProviderEvent( + { type: "subagent.updated", driver, subagent: subagent(siblingChild, identity.runId) }, + identity, + afterMove, + )[0], + ); + const resuming: RunExecutionService.ProviderEventRouteIdentity = { + ...identity, + runId: RunId.make("run:handoff:2"), + attemptId: RunAttemptId.make("attempt:handoff:2"), + }; + const resumingState = RunExecutionService.makeProviderEventRoutingState({ + identity: resuming, + providerTurnId: ProviderTurnId.make("provider-turn:handoff:2"), + }); + assert.isFalse( + RunExecutionService.routeProviderEvent(childMessage(resumedChild), resuming, resumingState)[0], + ); + const [moveAccepted, afterTake] = RunExecutionService.routeProviderEvent( + { type: "subagent.updated", driver, subagent: subagent(resumedChild, resuming.runId) }, + resuming, + resumingState, + ); + assert.isTrue(moveAccepted); + assert.isTrue( + RunExecutionService.routeProviderEvent(childMessage(resumedChild), resuming, afterTake)[0], + ); +}); + it("does not route a superseded attempt through a reused provider thread", () => { const threadId = ThreadId.make("thread:shared-runtime:restart"); const providerThreadId = ProviderThreadId.make("provider-thread:shared-runtime:restart"); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 9abe812c3d125..fece1667b4ca4 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -60,7 +60,7 @@ export interface ProviderEventRoutingState { readonly rootTurnEnded: boolean; readonly workflowMemberParents: ReadonlyMap; readonly ownedProviderThreadIds: ReadonlySet; - readonly ownedProviderTurnIds: ReadonlySet; + readonly ownedProviderTurns: ReadonlyMap; readonly inheritedBackgroundTurnItems: ReadonlyMap; readonly rootProviderTurnId: ProviderTurnId | null; } @@ -337,8 +337,10 @@ export function makeProviderEventRoutingState(input: { input.identity.providerThreadId, ...(input.relatedProviderThreadIds ?? []), ]), - ownedProviderTurnIds: - input.providerTurnId === null ? new Set() : new Set([input.providerTurnId]), + ownedProviderTurns: + input.providerTurnId === null + ? new Map() + : new Map([[input.providerTurnId, input.identity.providerThreadId]]), inheritedBackgroundTurnItems: new Map( (input.inheritedBackgroundTurnItems ?? []).map((item) => [item.id, item.runId]), ), @@ -361,10 +363,11 @@ export function routeProviderEvent( }); const addProviderTurn = ( providerTurnId: ProviderTurnId, + providerThreadId: ProviderThreadId, root: boolean, ): ProviderEventRoutingState => ({ ...state, - ownedProviderTurnIds: new Set([...state.ownedProviderTurnIds, providerTurnId]), + ownedProviderTurns: new Map(state.ownedProviderTurns).set(providerTurnId, providerThreadId), rootProviderTurnId: root ? providerTurnId : state.rootProviderTurnId, }); @@ -405,9 +408,14 @@ export function routeProviderEvent( isRoot || (event.providerTurn.providerThreadId !== input.providerThreadId && state.ownedProviderThreadIds.has(event.providerTurn.providerThreadId)) || - state.ownedProviderTurnIds.has(event.providerTurn.id) || + state.ownedProviderTurns.has(event.providerTurn.id) || (event.threadId !== undefined && ownsChildThread(event.threadId)); - return belongs ? [true, addProviderTurn(event.providerTurn.id, isRoot)] : [false, state]; + return belongs + ? [ + true, + addProviderTurn(event.providerTurn.id, event.providerTurn.providerThreadId, isRoot), + ] + : [false, state]; } case "node.updated": { const belongs = ownsRun(event.node.runId) || ownsChildThread(event.node.threadId); @@ -417,6 +425,25 @@ export function routeProviderEvent( return [true, addProviderThread(event.node.providerThreadId)]; } case "subagent.updated": { + const moved = event.subagent; + if ( + moved.origin === "provider_native" && + moved.threadId === input.threadId && + moved.runId !== null && + !ownsRun(moved.runId) && + moved.childThreadId !== null && + ownsChildThread(moved.childThreadId) + ) { + const ownedThreadIds = new Set(state.ownedThreadIds); + ownedThreadIds.delete(moved.childThreadId); + const ownedProviderThreadIds = new Set(state.ownedProviderThreadIds); + if (moved.providerThreadId !== null) ownedProviderThreadIds.delete(moved.providerThreadId); + const ownedProviderTurns = new Map(state.ownedProviderTurns); + for (const [turnId, threadId] of ownedProviderTurns) { + if (threadId === moved.providerThreadId) ownedProviderTurns.delete(turnId); + } + return [false, { ...state, ownedThreadIds, ownedProviderThreadIds, ownedProviderTurns }]; + } const belongs = ownsRun(event.subagent.runId) || ownsChildThread(event.subagent.threadId); const coordinator = event.subagent; // A provider-native subagent's child thread follows the live run it is @@ -478,7 +505,7 @@ export function routeProviderEvent( return [ (event.threadId !== undefined && ownsChildThread(event.threadId)) || (event.runtimeRequest.providerTurnId !== null && - state.ownedProviderTurnIds.has(event.runtimeRequest.providerTurnId)), + state.ownedProviderTurns.has(event.runtimeRequest.providerTurnId)), state, ]; case "command_output.delta": From 1dfcd0cd7eadb2c1270650a172b2a4f30c4215f3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 08:18:14 +0000 Subject: [PATCH 2/6] fix(server): release subscriptions after native child ownership moves --- .../RunExecutionService.test.ts | 49 +++++++++++- .../orchestration-v2/RunExecutionService.ts | 74 ++++++++++++++++++- 2 files changed, 119 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 126b52e40a6f3..a8851d0f8e5e2 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1875,6 +1875,48 @@ it.effect("keeps ingesting a child thread's late background item completion", () }), ); +it.effect("releases a settled run when its live child transfers to another run", () => + Effect.gen(function* () { + const consumed = yield* Deferred.make(); + const result = yield* Effect.raceFirst( + runBackgroundItemScenario( + "child-transfer-release", + (ids) => { + 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: null, + }, + }; + return [ + childThreadCreatedEvent(ids), + linked, + childBackgroundTurnItemEvent(ids, "running", 1), + rootTerminalEvent(ids, "completed"), + { ...linked, subagent: { ...linked.subagent, runId: RunId.make("run:replacement") } }, + ]; + }, + { + keepEventStreamOpen: true, + onEventsConsumed: Deferred.succeed(consumed, undefined).pipe(Effect.asVoid), + }, + ).pipe(Effect.map(Option.some)), + Deferred.await(consumed).pipe(Effect.as(Option.none())), + ); + assert.isTrue( + Option.isSome(result), + "the transferred child must release its prior subscription", + ); + if (Option.isSome(result)) + assert.deepEqual(result.value, ["subagent:running", "turn_item:running", "root-finalized"]); + }), +); + it.effect("keeps ingesting until the last of several background items terminalizes", () => Effect.gen(function* () { const secondItemId = TurnItemId.make("turn-item:bg-multi:second"); @@ -4512,6 +4554,7 @@ function runBackgroundItemScenario( ReadonlyArray<{ readonly id: TurnItemId; readonly runId: RunId }> >; readonly onSubscribe?: Effect.Effect; + readonly onEventsConsumed?: Effect.Effect; readonly onFinalized?: ( events: ReadonlyArray, ) => Effect.Effect; @@ -4579,7 +4622,11 @@ function runBackgroundItemScenario( events: Stream.empty, subscribeEvents: Effect.gen(function* () { yield* options?.onSubscribe ?? Effect.void; - const events = Stream.fromIterable(makeEvents(ids)); + const events = Stream.fromIterable(makeEvents(ids)).pipe( + Stream.concat( + Stream.fromEffect(options?.onEventsConsumed ?? Effect.void).pipe(Stream.drain), + ), + ); return { events: options?.keepEventStreamOpen === true diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index fece1667b4ca4..b23aaa97f251c 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1242,11 +1242,79 @@ export const layer: Layer.Layer< // when finally written, behind every item that began after it. const reservedHeldTurnItemIds = new Set(); const providerEventFiber = yield* eventSubscription.events.pipe( - Stream.filterEffect((event) => - Ref.modify(eventRouting, (state) => routeProviderEvent(event, routeIdentity, state)), + Stream.mapEffect((event) => + Effect.gen(function* () { + const previous = yield* Ref.get(eventRouting); + const [accepted, routing] = routeProviderEvent(event, routeIdentity, previous); + yield* Ref.set(eventRouting, routing); + if ( + event.type === "subagent.updated" && + event.subagent.childThreadId !== null && + previous.ownedThreadIds.has(event.subagent.childThreadId) && + !routing.ownedThreadIds.has(event.subagent.childThreadId) + ) { + const moved = event.subagent; + const detachedItems = yield* Ref.modify(openRunOwnedSubagents, (current) => { + const childTurnItems = new Map( + [...current.childTurnItems].filter( + ([, item]) => item.threadId !== moved.childThreadId, + ), + ); + const turnItems = new Map(current.turnItems); + turnItems.delete(moved.id); + const detachedItems = new Set( + [...current.childTurnItems.values(), ...current.turnItems.values()] + .filter( + (item) => + item.threadId === moved.childThreadId || item.nodeId === moved.id, + ) + .map((item) => item.id), + ); + return [ + detachedItems, + { + subagents: new Map( + [...current.subagents].filter( + ([id, subagent]) => + id !== moved.id && subagent.threadId !== moved.childThreadId, + ), + ), + turnItems, + childTurnItems, + nodes: new Map( + [...current.nodes].filter( + ([id, node]) => + id !== moved.id && node.threadId !== moved.childThreadId, + ), + ), + linkedChildThreadIds: new Set( + [...current.linkedChildThreadIds].filter( + (id) => id !== moved.childThreadId, + ), + ), + }, + ]; + }); + yield* Ref.update( + activeChildSubagents, + (current) => new Set([...current].filter((id) => id !== moved.id)), + ); + yield* Ref.update( + activeChildProviderTurns, + (current) => + new Set([...current].filter((id) => routing.ownedProviderTurns.has(id))), + ); + yield* Ref.update( + activeBackgroundTurnItems, + (current) => new Set([...current].filter((id) => !detachedItems.has(id))), + ); + } + return { event, accepted }; + }), ), - Stream.tap((event) => + Stream.tap(({ event, accepted }) => Effect.gen(function* () { + if (!accepted) return; let storedEventCount = 0; const deliveredEvent = filterAssistantEvent( event, From 9077e208dd0dcd536e979c64a7aff433d8b4a518 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 09:57:02 +0000 Subject: [PATCH 3/6] fix(server): release detached nested subagent subscriptions --- .../RunExecutionService.test.ts | 17 ++++++++++++++++- .../orchestration-v2/RunExecutionService.ts | 18 ++++++++++-------- 2 files changed, 26 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index a8851d0f8e5e2..a3f3735fba43b 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1896,6 +1896,16 @@ it.effect("releases a settled run when its live child transfers to another run", return [ childThreadCreatedEvent(ids), linked, + { + ...linked, + subagent: { + ...linked.subagent, + id: NodeId.make("node:child-transfer-release:nested"), + threadId: ids.childThreadId, + runId: null, + childThreadId: null, + }, + }, childBackgroundTurnItemEvent(ids, "running", 1), rootTerminalEvent(ids, "completed"), { ...linked, subagent: { ...linked.subagent, runId: RunId.make("run:replacement") } }, @@ -1913,7 +1923,12 @@ it.effect("releases a settled run when its live child transfers to another run", "the transferred child must release its prior subscription", ); if (Option.isSome(result)) - assert.deepEqual(result.value, ["subagent:running", "turn_item:running", "root-finalized"]); + assert.deepEqual(result.value, [ + "subagent:running", + "subagent:running", + "turn_item:running", + "root-finalized", + ]); }), ); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index b23aaa97f251c..1d0dde59e0071 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1254,7 +1254,11 @@ export const layer: Layer.Layer< !routing.ownedThreadIds.has(event.subagent.childThreadId) ) { const moved = event.subagent; - const detachedItems = yield* Ref.modify(openRunOwnedSubagents, (current) => { + const detached = yield* Ref.modify(openRunOwnedSubagents, (current) => { + const subagentIds = new Set([moved.id]); + for (const subagent of current.subagents.values()) { + if (subagent.threadId === moved.childThreadId) subagentIds.add(subagent.id); + } const childTurnItems = new Map( [...current.childTurnItems].filter( ([, item]) => item.threadId !== moved.childThreadId, @@ -1271,13 +1275,10 @@ export const layer: Layer.Layer< .map((item) => item.id), ); return [ - detachedItems, + { itemIds: detachedItems, subagentIds }, { subagents: new Map( - [...current.subagents].filter( - ([id, subagent]) => - id !== moved.id && subagent.threadId !== moved.childThreadId, - ), + [...current.subagents].filter(([id]) => !subagentIds.has(id)), ), turnItems, childTurnItems, @@ -1297,7 +1298,8 @@ export const layer: Layer.Layer< }); yield* Ref.update( activeChildSubagents, - (current) => new Set([...current].filter((id) => id !== moved.id)), + (current) => + new Set([...current].filter((id) => !detached.subagentIds.has(id))), ); yield* Ref.update( activeChildProviderTurns, @@ -1306,7 +1308,7 @@ export const layer: Layer.Layer< ); yield* Ref.update( activeBackgroundTurnItems, - (current) => new Set([...current].filter((id) => !detachedItems.has(id))), + (current) => new Set([...current].filter((id) => !detached.itemIds.has(id))), ); } return { event, accepted }; From 14a4145513f35c824f612522f5d4f5f8c6a120a4 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 20 Oct 2026 03:05:47 +0000 Subject: [PATCH 4/6] fix(mobile): import thread screen callback hook --- apps/mobile/src/features/threads/ThreadDetailScreen.tsx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 6d354e57c508b..ac22a8b7d9e10 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -58,7 +58,7 @@ import type { QueuedRunEdit } from "../../state/queued-run-edit"; import type { FollowUpBehavior } from "../../lib/followUpBehavior"; import type { ActiveTurnComposerAction } from "@t3tools/client-runtime/state/composer-dispatch"; import * as Haptics from "expo-haptics"; -import { useContext, useEffect, useLayoutEffect, useRef, useState } from "react"; +import { useCallback, useContext, useEffect, useLayoutEffect, useRef, useState } from "react"; import { AppState, Keyboard, From 6f05f3f471c217e1419a5d5c70f551ae040bc48a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 20 Oct 2026 03:07:06 +0000 Subject: [PATCH 5/6] style(server): format merged child routing --- apps/server/src/orchestration-v2/RunExecutionService.ts | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index fbcf9334d537f..c3df5228c6b24 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1263,8 +1263,7 @@ export const layer: Layer.Layer< const detachedItems = new Set( [...current.childTurnItems.values(), ...current.turnItems.values()] .filter( - (item) => - item.threadId === moved.childThreadId || item.nodeId === moved.id, + (item) => item.threadId === moved.childThreadId || item.nodeId === moved.id, ) .map((item) => item.id), ); @@ -1278,8 +1277,7 @@ export const layer: Layer.Layer< childTurnItems, nodes: new Map( [...current.nodes].filter( - ([id, node]) => - id !== moved.id && node.threadId !== moved.childThreadId, + ([id, node]) => id !== moved.id && node.threadId !== moved.childThreadId, ), ), linkedChildThreadIds: new Set( @@ -1292,8 +1290,7 @@ export const layer: Layer.Layer< }); yield* Ref.update( activeChildSubagents, - (current) => - new Set([...current].filter((id) => !detached.subagentIds.has(id))), + (current) => new Set([...current].filter((id) => !detached.subagentIds.has(id))), ); yield* Ref.update( activeChildProviderTurns, From 9e330f727e219bd0ae2d252b6d108c39bedc6988 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 20 Oct 2026 03:20:32 +0000 Subject: [PATCH 6/6] fix(server): release nested SDK tasks on child transfer --- .../RunExecutionService.test.ts | 108 +++++++++--------- .../orchestration-v2/RunExecutionService.ts | 14 ++- 2 files changed, 65 insertions(+), 57 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index a3f3735fba43b..1281f578e34c0 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1875,61 +1875,63 @@ it.effect("keeps ingesting a child thread's late background item completion", () }), ); -it.effect("releases a settled run when its live child transfers to another run", () => - Effect.gen(function* () { - const consumed = yield* Deferred.make(); - const result = yield* Effect.raceFirst( - runBackgroundItemScenario( - "child-transfer-release", - (ids) => { - 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: null, - }, - }; - return [ - childThreadCreatedEvent(ids), - linked, - { - ...linked, +it.effect.each([null, "run:nested-sdk"] as const)( + "releases a settled run when its live child transfers to another run (%s)", + (nestedRunId) => + Effect.gen(function* () { + const consumed = yield* Deferred.make(); + const result = yield* Effect.raceFirst( + runBackgroundItemScenario( + "child-transfer-release", + (ids) => { + const running = subagentEvent(ids, "running"); + if (running.type !== "subagent.updated") throw new Error("Missing subagent"); + const linked = { + ...running, subagent: { - ...linked.subagent, - id: NodeId.make("node:child-transfer-release:nested"), - threadId: ids.childThreadId, - runId: null, - childThreadId: null, + ...running.subagent, + origin: "provider_native" as const, + childThreadId: ids.childThreadId, + providerThreadId: null, }, - }, - childBackgroundTurnItemEvent(ids, "running", 1), - rootTerminalEvent(ids, "completed"), - { ...linked, subagent: { ...linked.subagent, runId: RunId.make("run:replacement") } }, - ]; - }, - { - keepEventStreamOpen: true, - onEventsConsumed: Deferred.succeed(consumed, undefined).pipe(Effect.asVoid), - }, - ).pipe(Effect.map(Option.some)), - Deferred.await(consumed).pipe(Effect.as(Option.none())), - ); - assert.isTrue( - Option.isSome(result), - "the transferred child must release its prior subscription", - ); - if (Option.isSome(result)) - assert.deepEqual(result.value, [ - "subagent:running", - "subagent:running", - "turn_item:running", - "root-finalized", - ]); - }), + }; + return [ + childThreadCreatedEvent(ids), + linked, + { + ...linked, + subagent: { + ...linked.subagent, + id: NodeId.make("node:child-transfer-release:nested"), + threadId: ids.childThreadId, + runId: nestedRunId === null ? null : RunId.make(nestedRunId), + childThreadId: null, + }, + }, + childBackgroundTurnItemEvent(ids, "running", 1), + rootTerminalEvent(ids, "completed"), + { ...linked, subagent: { ...linked.subagent, runId: RunId.make("run:replacement") } }, + ]; + }, + { + keepEventStreamOpen: true, + onEventsConsumed: Deferred.succeed(consumed, undefined).pipe(Effect.asVoid), + }, + ).pipe(Effect.map(Option.some)), + Deferred.await(consumed).pipe(Effect.as(Option.none())), + ); + assert.isTrue( + Option.isSome(result), + "the transferred child must release its prior subscription", + ); + if (Option.isSome(result)) + assert.deepEqual(result.value, [ + "subagent:running", + "subagent:running", + "turn_item:running", + "root-finalized", + ]); + }), ); it.effect("keeps ingesting until the last of several background items terminalizes", () => diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index c3df5228c6b24..07c19e73e9cba 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1001,7 +1001,7 @@ export const layer: Layer.Layer< const rootRunFinalized = yield* Ref.make(false); const providerThreadOwnerLost = yield* Ref.make(false); const activeChildProviderTurns = yield* Ref.make>(new Set()); - const activeChildSubagents = yield* Ref.make>(new Set()); + const activeChildSubagents = yield* Ref.make>(new Map()); const activeBackgroundTurnItems = yield* Ref.make< ReadonlySet >(new Set(inheritedBackgroundTurnItemsById.keys())); @@ -1077,11 +1077,11 @@ export const layer: Layer.Layer< routing.ownedThreadIds.has(event.subagent.threadId); if (belongsToRootRun || belongsToOwnedChildThread) { yield* Ref.update(activeChildSubagents, (current) => { - const next = new Set(current); + const next = new Map(current); if (isSettledSubagentStatus(event.subagent.status)) { next.delete(event.subagent.id); } else { - next.add(event.subagent.id); + next.set(event.subagent.id, event.subagent.threadId); } return next; }); @@ -1290,7 +1290,13 @@ export const layer: Layer.Layer< }); yield* Ref.update( activeChildSubagents, - (current) => new Set([...current].filter((id) => !detached.subagentIds.has(id))), + (current) => + new Map( + [...current].filter( + ([id, threadId]) => + !detached.subagentIds.has(id) && threadId !== moved.childThreadId, + ), + ), ); yield* Ref.update( activeChildProviderTurns,