From 9535964b28a7bbe41639f7dff48dcf3a9abcb702 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 6 Nov 2026 23:38:09 +0000 Subject: [PATCH 1/6] test(server): a finished child's running native subagent keeps its row working after Stop, reproduces #32235 A delegated child whose run completed but still records a provider-native subagent as running holds the parent's task in waiting_for_children. The row's Stop (subagent.stop on the parent, thread.stop on the child) and a recovery pass leave that task running. AI-assisted; verified with the tests and tools listed above. --- .../src/orchestration-v2/ThreadStop.test.ts | 70 +++++++++++++++++++ 1 file changed, 70 insertions(+) diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index 6b3ea08a86449..1d4cd62845e7b 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -1742,3 +1742,73 @@ it.effect.each(["Stop", "Force stop"] as const)( ); }).pipe(Effect.provide(layerTest)), ); + +it.effect( + "Stop settles a delegated row whose finished child still records a running native subagent", + () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const parentThreadId = ThreadId.make("thread:held-row"); + yield* createWatchingThread(parentThreadId, 30); + yield* send(parentThreadId, "parent", "start_immediately"); + const childThreadId = yield* delegate(parentThreadId, "held child"); + const childRun = (yield* orchestrator.getThreadProjection(childThreadId)).runs[0]!; + const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!; + const now = yield* DateTime.now; + // A native subagent the provider never reported ending, with no open turn item. + yield* projections.apply({ + id: EventId.make("event:held-row:native"), + type: "subagent.updated", + threadId: childThreadId, + runId: childRun.id, + occurredAt: now, + payload: { + ...task, + id: NodeId.make("node:held-row-native"), + threadId: childThreadId, + runId: childRun.id, + parentNodeId: childRun.rootNodeId!, + origin: "provider_native", + childThreadId: null, + status: "running", + completionDelivery: undefined, + }, + }); + yield* projections.apply({ + id: EventId.make("event:held-row:child-completed"), + type: "run.updated", + threadId: childThreadId, + runId: childRun.id, + occurredAt: now, + payload: { ...childRun, status: "completed", startedAt: now, completedAt: now }, + }); + yield* orchestrator.recoverDelegatedTasks; + assert.equal( + (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]?.status, + "running", + ); + + // The row's Stop: the parent's task, then the child thread. + yield* orchestrator.dispatch({ + type: "subagent.stop", + commandId: CommandId.make("held-row:stop-task"), + threadId: parentThreadId, + runId: task.runId!, + subagentId: task.id, + }); + yield* orchestrator.dispatch({ + type: "thread.stop", + commandId: CommandId.make("held-row:stop-child"), + threadId: childThreadId, + }); + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + yield* worker.drain(); + yield* orchestrator.recoverDelegatedTasks; + assert.notInclude( + ["pending", "running", "waiting"], + (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]?.status, + "Stop leaves the finished child's row working", + ); + }).pipe(Effect.provide(layerTest)), +); From 60aac5a0842cab2cf7082d8f24545c0316b044c2 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 6 Nov 2026 23:43:26 +0000 Subject: [PATCH 2/6] fix(server): Stop ends a finished child's stale native subagent so its row settles A delegated child that finished while still recording a provider-native subagent as active held its parent's task forever: delegatedTaskProgress treats the record as live work, and neither the row's Stop nor task_cancel had anything to interrupt, so the parent's row stayed Working. When thread.stop finds no live run or background work to interrupt, it now ends that thread's provider-native subagent records that are still active but have no live turn item, through the same endOrphanedNativeSubagent settlement the interrupt path uses. The settled subagent.updated event then drives the existing finalizer, which completes the parent's task with the child's result. Records with a live run or turn item stay with the interrupt path. Fixes pingdotgg/t3code#32235 AI-assisted; verified with the tests and tools listed above. --- .../src/orchestration-v2/Orchestrator.ts | 30 +++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index bcec0a13cf20a..222fc4cbc4298 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -12247,6 +12247,36 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio for (const run of reached.values()) { yield* markStoppedRun({ command, events, thread: projection.thread, run, now }); } + // With nothing left to interrupt, a native subagent still recorded as active (no live + // turn item) would hold this thread's delegated result forever, so Stop ends it. + const staleNative = (target === undefined ? projection.subagents : []).filter( + (subagent) => + subagent.origin === "provider_native" && + isOrchestrationV2WorkActive(subagent.status) && + !projection.turnItems.some( + (item) => item.type === "subagent" && item.subagentId === subagent.id, + ), + ); + if (staleNative.length > 0) { + const { nodes } = yield* projectionStore + .getThreadRecords(command.threadId, ["nodes"]) + .pipe(mapDispatchError(command)); + for (const subagent of staleNative) { + const ended = yield* endOrphanedNativeSubagent({ + projection: { nodes, subagents: projection.subagents, turnItems: projection.turnItems }, + subagentId: subagent.id, + status: "interrupted", + now, + emitted: yield* Ref.get(events), + allocateEventId: () => + idAllocator.allocate.event({ + threadId: command.threadId, + commandId: command.commandId, + }), + }).pipe(mapDispatchError(command)); + yield* Ref.update(events, (existing) => [...existing, ...ended]); + } + } yield* holdStoppedThread({ command, events, projection, cohortRunIds: [], now }); yield* cancelIdleOrphans({ command, events, projection }); yield* stopDelegatedTasks; From 4adccc6af930d8fcf98d22c30ff14f698861e4f2 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 7 Nov 2026 00:03:40 +0000 Subject: [PATCH 3/6] fix(server): Stop also ends a stale native subagent's own thread The stale native record Stop ends can own a child thread whose runless root turn still shows Working; end it the way the other native-stop paths do. The test now settles through the production reactor and checks that thread. AI-assisted; verified with the tests and tools listed above. --- .../src/orchestration-v2/Orchestrator.ts | 32 +++++++++-- .../src/orchestration-v2/ThreadStop.test.ts | 53 +++++++++++++++++-- 2 files changed, 76 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 222fc4cbc4298..7259dbd96297f 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -12261,6 +12261,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const { nodes } = yield* projectionStore .getThreadRecords(command.threadId, ["nodes"]) .pipe(mapDispatchError(command)); + const allocateEventId = () => + idAllocator.allocate.event({ + threadId: command.threadId, + commandId: command.commandId, + }); for (const subagent of staleNative) { const ended = yield* endOrphanedNativeSubagent({ projection: { nodes, subagents: projection.subagents, turnItems: projection.turnItems }, @@ -12268,13 +12273,30 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio status: "interrupted", now, emitted: yield* Ref.get(events), - allocateEventId: () => - idAllocator.allocate.event({ - threadId: command.threadId, - commandId: command.commandId, - }), + allocateEventId, }).pipe(mapDispatchError(command)); yield* Ref.update(events, (existing) => [...existing, ...ended]); + if (subagent.childThreadId === null) continue; + const child = yield* projectionStore + .getThreadRecords(subagent.childThreadId, ["nodes", "providerTurns", "turnItems"], { + turnItemRunIds: [null], + turnItemStatuses: ["pending", "running", "waiting"], + }) + .pipe( + Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }), + mapDispatchError(command), + ); + if (child === null) continue; + const childEnded = yield* endRunlessRootTurns({ + threadId: subagent.childThreadId, + providerInstanceId: subagent.providerInstanceId, + projection: child, + status: "interrupted", + now, + emitted: yield* Ref.get(events), + allocateEventId, + }).pipe(mapDispatchError(command)); + yield* Ref.update(events, (existing) => [...existing, ...childEnded]); } } yield* holdStoppedThread({ command, events, projection, cohortRunIds: [], now }); diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index 1d4cd62845e7b..bb03316ae05ac 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -1756,6 +1756,46 @@ it.effect( const childRun = (yield* orchestrator.getThreadProjection(childThreadId)).runs[0]!; const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!; const now = yield* DateTime.now; + const nativeThreadId = ThreadId.make("thread:held-row-native"); + yield* createWatchingThread(nativeThreadId, 31); + const nativeRoot = { + ...(yield* orchestrator.getThreadProjection(childThreadId)).nodes.find( + (node) => node.id === childRun.rootNodeId, + )!, + id: NodeId.make("node:held-row-native-root"), + threadId: nativeThreadId, + runId: null, + parentNodeId: null, + status: "running" as const, + countsForRun: false, + completedAt: null, + }; + yield* projections.apply({ + id: EventId.make("event:held-row:native-root"), + type: "node.updated", + threadId: nativeThreadId, + nodeId: nativeRoot.id, + occurredAt: now, + payload: nativeRoot, + }); + yield* projections.apply({ + id: EventId.make("event:held-row:native-turn"), + type: "provider-turn.updated", + threadId: nativeThreadId, + nodeId: nativeRoot.id, + occurredAt: now, + payload: { + id: ProviderTurnId.make("turn:held-row-native"), + providerThreadId: ProviderThreadId.make("provider-thread:held-row-native"), + nodeId: nativeRoot.id, + runAttemptId: null, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }); // A native subagent the provider never reported ending, with no open turn item. yield* projections.apply({ id: EventId.make("event:held-row:native"), @@ -1770,7 +1810,7 @@ it.effect( runId: childRun.id, parentNodeId: childRun.rootNodeId!, origin: "provider_native", - childThreadId: null, + childThreadId: nativeThreadId, status: "running", completionDelivery: undefined, }, @@ -1804,11 +1844,16 @@ it.effect( }); const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; yield* worker.drain(); - yield* orchestrator.recoverDelegatedTasks; - assert.notInclude( - ["pending", "running", "waiting"], + assert.equal( (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]?.status, + "completed", "Stop leaves the finished child's row working", ); + assert.equal( + (yield* orchestrator.getThreadProjection(nativeThreadId)).nodes.find( + (node) => node.id === nativeRoot.id, + )?.status, + "interrupted", + ); }).pipe(Effect.provide(layerTest)), ); From 40dfa08c962e3fc63ea260fd1808cdb00a32d836 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 7 Nov 2026 02:59:50 +0000 Subject: [PATCH 4/6] fix(server): Stop spares live native subagents and ends nested ones Stop's cleanup of a finished child's stale native subagent now skips a record whose provider session is still live, since that session reports the end itself, and only considers records from the thread's runs, which leaves runless workflow members alone. It ends the native agent's nested subagents and their roots through the same walk settleBackgroundWork uses, now shared as endNativeSubagentTrees, with hasLiveSession hoisted beside it. AI-assisted; verified with the tests and tools listed above. --- .../src/orchestration-v2/Orchestrator.ts | 305 ++++++++++-------- .../src/orchestration-v2/ThreadStop.test.ts | 291 ++++++++++++----- 2 files changed, 366 insertions(+), 230 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index afd7e642d56bf..b8df43340cde0 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1094,6 +1094,21 @@ function lastDeliveredRunForProviderThread( ); } +type DyingNativeSubagent = { + readonly projection: Pick; + readonly subagent: Pick< + Extract, + "subagentId" | "origin" | "childThreadId" | "providerInstanceId" + >; +}; + +const nativeSubagentRef = (subagent: OrchestrationV2Subagent): DyingNativeSubagent["subagent"] => ({ + subagentId: subagent.id, + origin: subagent.origin, + childThreadId: subagent.childThreadId, + providerInstanceId: subagent.providerInstanceId, +}); + const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(function* () { const checkpointService = yield* CheckpointServiceV2; const commandPolicy = yield* CommandPolicyV2; @@ -10628,6 +10643,93 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } }); + const hasLiveSession = Effect.fn("Orchestrator.hasLiveSession")(function* ( + providerThreads: ReadonlyArray, + providerThreadId: OrchestrationV2ProviderThread["id"], + ) { + const sessionId = providerThreads.find( + (candidate) => candidate.id === providerThreadId, + )?.providerSessionId; + return ( + sessionId !== null && + sessionId !== undefined && + Option.isSome( + yield* providerSessions.get(sessionId).pipe(Effect.orElseSucceed(() => Option.none())), + ) + ); + }); + + // Ends each native subagent's record, its own thread's runless root turns, and the + // subagents it started there, which a provider can record on that thread rather than its parent's. + const endNativeSubagentTrees = Effect.fn("Orchestrator.endNativeSubagentTrees")( + function* (input: { + readonly command: Extract< + OrchestrationV2ServerCommand, + { readonly type: "run.interrupt" | "thread.background-work.settle" | "thread.stop" } + >; + readonly events: Ref.Ref>; + readonly dying: ReadonlyArray; + readonly now: DateTime.Utc; + }) { + const allocateEventId = () => + idAllocator.allocate.event({ + threadId: input.command.threadId, + commandId: input.command.commandId, + }); + const dying = [...input.dying]; + const childThreadIds = new Set(); + for (const { projection, subagent } of dying) { + const ended = yield* endOrphanedNativeSubagent({ + projection, + subagentId: subagent.subagentId, + status: "interrupted", + now: input.now, + emitted: yield* Ref.get(input.events), + allocateEventId, + }).pipe(mapDispatchError(input.command)); + yield* Ref.update(input.events, (events) => [...events, ...ended]); + const childThreadId = subagent.childThreadId; + if ( + subagent.origin !== "provider_native" || + childThreadId === null || + childThreadIds.has(childThreadId) + ) { + continue; + } + childThreadIds.add(childThreadId); + const child = yield* projectionStore + .getThreadRecords(childThreadId, ["nodes", "providerTurns", "subagents", "turnItems"], { + // Runless items are the subagent's; any run there is the user's own. + turnItemRunIds: [null], + turnItemStatuses: ["pending", "running", "waiting"], + }) + .pipe( + Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }), + mapDispatchError(input.command), + ); + if (child === null) continue; + const childEnded = yield* endRunlessRootTurns({ + threadId: childThreadId, + providerInstanceId: subagent.providerInstanceId, + projection: child, + status: "interrupted", + now: input.now, + emitted: yield* Ref.get(input.events), + allocateEventId, + }).pipe(mapDispatchError(input.command)); + yield* Ref.update(input.events, (events) => [...events, ...childEnded]); + for (const nested of child.turnItems) { + if (nested.type === "subagent") dying.push({ projection: child, subagent: nested }); + } + for (const nested of child.subagents) { + if (nested.origin === "provider_native" && nested.runId === null) { + dying.push({ projection: child, subagent: nativeSubagentRef(nested) }); + } + } + } + }, + ); + const settleBackgroundWork = Effect.fn("Orchestrator.settleBackgroundWork")(function* (input: { readonly command: Extract< OrchestrationV2ServerCommand, @@ -10644,30 +10746,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }) { const emitEvent = emit(input.events, input.command); const runOrdinals = new Map(input.projection.runs.map((run) => [run.id, run.ordinal])); - const allocateEventId = () => - idAllocator.allocate.event({ - threadId: input.command.threadId, - commandId: input.command.commandId, - }); - const childThreadIds = new Set(); - const liveness = new Map(); - const hasLiveSession = Effect.fn("Orchestrator.hasLiveSession")(function* ( - providerThreadId: OrchestrationV2ProviderThread["id"], - ) { - const known = liveness.get(providerThreadId); - if (known !== undefined) return known; - const sessionId = input.projection.providerThreads.find( - (candidate) => candidate.id === providerThreadId, - )?.providerSessionId; - const live = - sessionId !== null && - sessionId !== undefined && - Option.isSome( - yield* providerSessions.get(sessionId).pipe(Effect.orElseSucceed(() => Option.none())), - ); - liveness.set(providerThreadId, live); - return live; - }); + const dying: Array = []; const stillReportsBackgroundWork = Effect.fnUntraced(function* ( providerThread: OrchestrationV2ProviderThread, ) { @@ -10693,7 +10772,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if ( providerThreadId !== null && providerThreadId !== input.stoppedProviderThreadId && - (yield* hasLiveSession(providerThreadId)) + (yield* hasLiveSession(input.projection.providerThreads, providerThreadId)) ) { continue; } @@ -10712,85 +10791,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio updatedAt: input.now, }, }); - if (item.type !== "subagent") continue; - // A native subagent died with the process too: end its record, its own - // thread's runless root turn, and the subagents it started there, which - // a provider can record on that thread rather than this one. - const dying: Array<{ - readonly projection: Pick< - OrchestrationV2ThreadProjection, - "nodes" | "subagents" | "turnItems" - >; - readonly subagent: Pick< - Extract, - "subagentId" | "origin" | "childThreadId" | "providerInstanceId" - >; - }> = [{ projection: input.projection, subagent: item }]; - for (const { projection, subagent } of dying) { - const ended = yield* endOrphanedNativeSubagent({ - projection, - subagentId: subagent.subagentId, - status: "interrupted", - now: input.now, - emitted: yield* Ref.get(input.events), - allocateEventId, - }).pipe(mapDispatchError(input.command)); - yield* Ref.update(input.events, (events) => [...events, ...ended]); - const childThreadId = subagent.childThreadId; - if ( - subagent.origin !== "provider_native" || - childThreadId === null || - childThreadIds.has(childThreadId) - ) { - continue; - } - childThreadIds.add(childThreadId); - const child = yield* projectionStore - .getThreadRecords(childThreadId, ["nodes", "providerTurns", "subagents", "turnItems"], { - // Runless items are the subagent's; any run there is the user's own. - turnItemRunIds: [null], - turnItemStatuses: ["pending", "running", "waiting"], - }) - .pipe( - Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }), - mapDispatchError(input.command), - ); - if (child === null) continue; - const childEnded = yield* endRunlessRootTurns({ - threadId: childThreadId, - providerInstanceId: subagent.providerInstanceId, - projection: child, - status: "interrupted", - now: input.now, - emitted: yield* Ref.get(input.events), - allocateEventId, - }).pipe(mapDispatchError(input.command)); - yield* Ref.update(input.events, (events) => [...events, ...childEnded]); - for (const nested of child.turnItems) { - if (nested.type === "subagent") dying.push({ projection: child, subagent: nested }); - } - for (const nested of child.subagents) { - if (nested.origin === "provider_native" && nested.runId === null) { - dying.push({ - projection: child, - subagent: { - subagentId: nested.id, - origin: nested.origin, - childThreadId: nested.childThreadId, - providerInstanceId: nested.providerInstanceId, - }, - }); - } - } - } + // A native subagent died with the process too. + if (item.type === "subagent") dying.push({ projection: input.projection, subagent: item }); } + yield* endNativeSubagentTrees({ + command: input.command, + events: input.events, + dying, + now: input.now, + }); for (const providerThread of input.projection.providerThreads) { if ((providerThread.pendingBackgroundTasks?.length ?? 0) === 0) continue; if (providerThread.id === input.stoppedProviderThreadId) { // A refused newer start can take ownership and reject the stopped // turn's roster clear. Reconcile it without hiding surviving work. if (yield* stillReportsBackgroundWork(providerThread)) continue; - } else if (yield* hasLiveSession(providerThread.id)) { + } else if (yield* hasLiveSession(input.projection.providerThreads, providerThread.id)) { continue; } yield* emitEvent({ @@ -12256,57 +12272,60 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio for (const run of reached.values()) { yield* markStoppedRun({ command, events, thread: projection.thread, run, now }); } - // With nothing left to interrupt, a native subagent still recorded as active (no live - // turn item) would hold this thread's delegated result forever, so Stop ends it. - const staleNative = (target === undefined ? projection.subagents : []).filter( - (subagent) => - subagent.origin === "provider_native" && - isOrchestrationV2WorkActive(subagent.status) && - !projection.turnItems.some( + // With nothing left to interrupt, a native subagent of this thread's run still recorded + // as active (no live turn item) would hold its delegated result forever, so Stop ends it. + const staleNative: Array = []; + for (const subagent of target === undefined ? projection.subagents : []) { + if ( + subagent.origin !== "provider_native" || + subagent.runId === null || + !isOrchestrationV2WorkActive(subagent.status) || + projection.turnItems.some( (item) => item.type === "subagent" && item.subagentId === subagent.id, - ), - ); + ) + ) { + continue; + } + // A live session still reports the end itself. + if (subagent.providerThreadId !== null) { + const nativeProviderThreads = + subagent.childThreadId === null + ? [] + : yield* projectionStore + .getThreadRecords(subagent.childThreadId, ["providerThreads"]) + .pipe( + Effect.map((child) => child.providerThreads), + Effect.catchTags({ + ProjectionStoreThreadNotFoundError: () => Effect.succeed([]), + }), + mapDispatchError(command), + ); + if ( + yield* hasLiveSession( + [...projection.providerThreads, ...nativeProviderThreads], + subagent.providerThreadId, + ) + ) { + continue; + } + } + staleNative.push(nativeSubagentRef(subagent)); + } if (staleNative.length > 0) { const { nodes } = yield* projectionStore .getThreadRecords(command.threadId, ["nodes"]) .pipe(mapDispatchError(command)); - const allocateEventId = () => - idAllocator.allocate.event({ - threadId: command.threadId, - commandId: command.commandId, - }); - for (const subagent of staleNative) { - const ended = yield* endOrphanedNativeSubagent({ - projection: { nodes, subagents: projection.subagents, turnItems: projection.turnItems }, - subagentId: subagent.id, - status: "interrupted", - now, - emitted: yield* Ref.get(events), - allocateEventId, - }).pipe(mapDispatchError(command)); - yield* Ref.update(events, (existing) => [...existing, ...ended]); - if (subagent.childThreadId === null) continue; - const child = yield* projectionStore - .getThreadRecords(subagent.childThreadId, ["nodes", "providerTurns", "turnItems"], { - turnItemRunIds: [null], - turnItemStatuses: ["pending", "running", "waiting"], - }) - .pipe( - Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }), - mapDispatchError(command), - ); - if (child === null) continue; - const childEnded = yield* endRunlessRootTurns({ - threadId: subagent.childThreadId, - providerInstanceId: subagent.providerInstanceId, - projection: child, - status: "interrupted", - now, - emitted: yield* Ref.get(events), - allocateEventId, - }).pipe(mapDispatchError(command)); - yield* Ref.update(events, (existing) => [...existing, ...childEnded]); - } + const threadProjection = { + nodes, + subagents: projection.subagents, + turnItems: projection.turnItems, + }; + yield* endNativeSubagentTrees({ + command, + events, + dying: staleNative.map((subagent) => ({ projection: threadProjection, subagent })), + now, + }); } yield* holdStoppedThread({ command, events, projection, cohortRunIds: [], now }); yield* cancelIdleOrphans({ command, events, projection }); diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index bb03316ae05ac..5558654c643ce 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -25,6 +25,7 @@ import * as Schema from "effect/Schema"; import * as SqlClient from "effect/sql/SqlClient"; import * as Stream from "effect/Stream"; import * as EventStore from "./EventStore.ts"; +import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as TestClock from "effect/testing/TestClock"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; @@ -49,7 +50,38 @@ const adapter = { driver: ProviderDriverKind.make("codex"), getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }), - openSession: () => Effect.die("Runs here never reach a provider"), + openSession: (input) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const unused = () => Effect.die("Runs here never reach a provider"); + return { + instanceId, + driver: ProviderDriverKind.make("codex"), + providerSessionId: input.providerSessionId, + providerSession: { + id: input.providerSessionId, + driver: ProviderDriverKind.make("codex"), + providerInstanceId: instanceId, + status: "ready", + cwd: process.cwd(), + model: modelSelection.model, + capabilities: CodexProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }, + events: Stream.never, + ensureThread: unused, + resumeThread: unused, + startTurn: unused, + steerTurn: unused, + interruptTurn: unused, + respondToRuntimeRequest: unused, + readThreadSnapshot: unused, + rollbackThread: unused, + forkThread: unused, + }; + }), } as ProviderAdapter.ProviderAdapterV2["Service"]; const layerDatabase = SqlitePersistence.layerMemory; const layerTest = ThreadManagementService.layer.pipe( @@ -1743,91 +1775,158 @@ it.effect.each(["Stop", "Force stop"] as const)( }).pipe(Effect.provide(layerTest)), ); -it.effect( - "Stop settles a delegated row whose finished child still records a running native subagent", - () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const projections = yield* ProjectionStore.ProjectionStoreV2; - const parentThreadId = ThreadId.make("thread:held-row"); - yield* createWatchingThread(parentThreadId, 30); - yield* send(parentThreadId, "parent", "start_immediately"); - const childThreadId = yield* delegate(parentThreadId, "held child"); - const childRun = (yield* orchestrator.getThreadProjection(childThreadId)).runs[0]!; - const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!; - const now = yield* DateTime.now; - const nativeThreadId = ThreadId.make("thread:held-row-native"); - yield* createWatchingThread(nativeThreadId, 31); - const nativeRoot = { - ...(yield* orchestrator.getThreadProjection(childThreadId)).nodes.find( - (node) => node.id === childRun.rootNodeId, - )!, - id: NodeId.make("node:held-row-native-root"), - threadId: nativeThreadId, - runId: null, - parentNodeId: null, - status: "running" as const, - countsForRun: false, - completedAt: null, - }; - yield* projections.apply({ - id: EventId.make("event:held-row:native-root"), - type: "node.updated", - threadId: nativeThreadId, - nodeId: nativeRoot.id, - occurredAt: now, - payload: nativeRoot, +// A finished delegated child whose native subagent the provider never reported ending, +// with no open turn item, and a nested native subagent on that agent's own thread. +const finishedChildWithNativeSubagent = (liveSessionId?: ProviderSessionId) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const parentThreadId = ThreadId.make("thread:held-row"); + yield* createWatchingThread(parentThreadId, 30); + yield* send(parentThreadId, "parent", "start_immediately"); + const childThreadId = yield* delegate(parentThreadId, "held child"); + const child = yield* orchestrator.getThreadProjection(childThreadId); + const childRun = child.runs[0]!; + const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!; + const now = yield* DateTime.now; + const nativeThreadId = ThreadId.make("thread:held-row-native"); + const nestedThreadId = ThreadId.make("thread:held-row-nested"); + yield* createWatchingThread(nativeThreadId, 31); + yield* createWatchingThread(nestedThreadId, 32); + const nativeProviderThreadId = ProviderThreadId.make("provider-thread:held-row-native"); + if (liveSessionId !== undefined) { + const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* sessions.open({ + threadId: childThreadId, + providerSessionId: liveSessionId, + modelSelection, + runtimePolicy: { runtimeMode: "full-access", interactionMode: "default", cwd: null }, }); yield* projections.apply({ - id: EventId.make("event:held-row:native-turn"), - type: "provider-turn.updated", + id: EventId.make("event:held-row:native-provider-thread"), + type: "provider-thread.updated", threadId: nativeThreadId, - nodeId: nativeRoot.id, occurredAt: now, payload: { - id: ProviderTurnId.make("turn:held-row-native"), - providerThreadId: ProviderThreadId.make("provider-thread:held-row-native"), - nodeId: nativeRoot.id, - runAttemptId: null, - nativeTurnRef: null, - ordinal: 1, - status: "running", - startedAt: now, - completedAt: null, - }, - }); - // A native subagent the provider never reported ending, with no open turn item. - yield* projections.apply({ - id: EventId.make("event:held-row:native"), - type: "subagent.updated", - threadId: childThreadId, - runId: childRun.id, - occurredAt: now, - payload: { - ...task, - id: NodeId.make("node:held-row-native"), - threadId: childThreadId, - runId: childRun.id, - parentNodeId: childRun.rootNodeId!, - origin: "provider_native", - childThreadId: nativeThreadId, - status: "running", - completionDelivery: undefined, + ...child.providerThreads[0]!, + id: nativeProviderThreadId, + appThreadId: nativeThreadId, + providerSessionId: liveSessionId, + status: "active", }, }); + } + const runningRoot = (threadId: ThreadId, id: NodeId) => ({ + ...child.nodes.find((node) => node.id === childRun.rootNodeId)!, + id, + rootNodeId: id, + threadId, + runId: null, + parentNodeId: null, + status: "running" as const, + countsForRun: false, + completedAt: null, + }); + const nativeRoot = runningRoot(nativeThreadId, NodeId.make("node:held-row-native-root")); + const nestedRoot = runningRoot(nestedThreadId, NodeId.make("node:held-row-nested-root")); + for (const root of [nativeRoot, nestedRoot]) { yield* projections.apply({ - id: EventId.make("event:held-row:child-completed"), - type: "run.updated", - threadId: childThreadId, - runId: childRun.id, + id: EventId.make(`event:held-row:${root.id}`), + type: "node.updated", + threadId: root.threadId, + nodeId: root.id, occurredAt: now, - payload: { ...childRun, status: "completed", startedAt: now, completedAt: now }, + payload: root, }); - yield* orchestrator.recoverDelegatedTasks; - assert.equal( - (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]?.status, - "running", - ); + } + yield* projections.apply({ + id: EventId.make("event:held-row:native-turn"), + type: "provider-turn.updated", + threadId: nativeThreadId, + nodeId: nativeRoot.id, + occurredAt: now, + payload: { + id: ProviderTurnId.make("turn:held-row-native"), + providerThreadId: nativeProviderThreadId, + nodeId: nativeRoot.id, + runAttemptId: null, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }); + const native = { + ...task, + id: NodeId.make("node:held-row-native"), + threadId: childThreadId, + runId: childRun.id, + parentNodeId: childRun.rootNodeId!, + origin: "provider_native" as const, + providerThreadId: nativeProviderThreadId, + childThreadId: nativeThreadId, + status: "running" as const, + completionDelivery: undefined, + }; + yield* projections.apply({ + id: EventId.make("event:held-row:native"), + type: "subagent.updated", + threadId: childThreadId, + runId: childRun.id, + occurredAt: now, + payload: native, + }); + yield* projections.apply({ + id: EventId.make("event:held-row:nested"), + type: "subagent.updated", + threadId: nativeThreadId, + occurredAt: now, + payload: { + ...native, + id: NodeId.make("node:held-row-nested"), + threadId: nativeThreadId, + runId: null, + parentNodeId: nativeRoot.id, + childThreadId: nestedThreadId, + }, + }); + yield* projections.apply({ + id: EventId.make("event:held-row:child-completed"), + type: "run.updated", + threadId: childThreadId, + runId: childRun.id, + occurredAt: now, + payload: { ...childRun, status: "completed", startedAt: now, completedAt: now }, + }); + yield* orchestrator.recoverDelegatedTasks; + assert.equal( + (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]?.status, + "running", + ); + const statuses = Effect.gen(function* () { + const nativeAfter = yield* orchestrator.getThreadProjection(nativeThreadId); + const nestedAfter = yield* orchestrator.getThreadProjection(nestedThreadId); + return { + row: (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]?.status, + native: (yield* orchestrator.getThreadProjection(childThreadId)).subagents.find( + (subagent) => subagent.id === native.id, + )?.status, + nativeRoot: nativeAfter.nodes.find((node) => node.id === nativeRoot.id)?.status, + nested: nativeAfter.subagents[0]?.status, + nestedRoot: nestedAfter.nodes.find((node) => node.id === nestedRoot.id)?.status, + }; + }); + return { parentThreadId, childThreadId, task, statuses }; + }); + +it.effect( + "Stop settles a delegated row whose finished child still records a running native subagent", + () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const { parentThreadId, childThreadId, task, statuses } = + yield* finishedChildWithNativeSubagent(); // The row's Stop: the parent's task, then the child thread. yield* orchestrator.dispatch({ @@ -1844,16 +1943,34 @@ it.effect( }); const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; yield* worker.drain(); - assert.equal( - (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]?.status, - "completed", - "Stop leaves the finished child's row working", - ); - assert.equal( - (yield* orchestrator.getThreadProjection(nativeThreadId)).nodes.find( - (node) => node.id === nativeRoot.id, - )?.status, - "interrupted", - ); + assert.deepEqual(yield* statuses, { + row: "completed", + native: "interrupted", + nativeRoot: "interrupted", + nested: "interrupted", + nestedRoot: "interrupted", + }); }).pipe(Effect.provide(layerTest)), ); + +it.effect("Stop leaves a native subagent running while its provider session is live", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const { childThreadId, statuses } = yield* finishedChildWithNativeSubagent( + ProviderSessionId.make("session:held-row-live"), + ); + + yield* orchestrator.dispatch({ + type: "thread.stop", + commandId: CommandId.make("held-row:stop-live-child"), + threadId: childThreadId, + }); + assert.deepEqual(yield* statuses, { + row: "running", + native: "running", + nativeRoot: "running", + nested: "running", + nestedRoot: "running", + }); + }).pipe(Effect.provide(layerTest)), +); From 5c05c9c31404f1aafcafc61b7d69923485990a0b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 7 Nov 2026 03:18:45 +0000 Subject: [PATCH 5/6] fix(server): Stop ends stale native subagents even with a live session A native subagent record that is still active but has no live turn item lost its end, since adapters write the record and its item together. A live provider session cannot tell such a stuck agent from a running one: for Codex the stale agent itself keeps the session pinned. So a thread Stop with nothing else to interrupt now ends these records and their native subtree without checking session liveness. settleBackgroundWork keeps its own per-call liveness cache and again ends each interrupted subagent's tree right after that item's update, sharing one visited set across items. Tests: ThreadStop (regression fails on main with the row still running), BackgroundWorkStop, ProjectionSettlement, ProviderRuntimeRecoveryService, RestartBackgroundNote, RestartContinuation, DelegatedCompletionDelivery, and vp run check:changed. AI-assisted; verified with the tests and tools listed above. --- .../src/orchestration-v2/Orchestrator.ts | 85 ++--- .../src/orchestration-v2/ThreadStop.test.ts | 304 +++++++----------- 2 files changed, 143 insertions(+), 246 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index b8df43340cde0..08ce94e82eadb 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -10643,22 +10643,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } }); - const hasLiveSession = Effect.fn("Orchestrator.hasLiveSession")(function* ( - providerThreads: ReadonlyArray, - providerThreadId: OrchestrationV2ProviderThread["id"], - ) { - const sessionId = providerThreads.find( - (candidate) => candidate.id === providerThreadId, - )?.providerSessionId; - return ( - sessionId !== null && - sessionId !== undefined && - Option.isSome( - yield* providerSessions.get(sessionId).pipe(Effect.orElseSucceed(() => Option.none())), - ) - ); - }); - // Ends each native subagent's record, its own thread's runless root turns, and the // subagents it started there, which a provider can record on that thread rather than its parent's. const endNativeSubagentTrees = Effect.fn("Orchestrator.endNativeSubagentTrees")( @@ -10669,6 +10653,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio >; readonly events: Ref.Ref>; readonly dying: ReadonlyArray; + readonly childThreadIds: Set; readonly now: DateTime.Utc; }) { const allocateEventId = () => @@ -10677,7 +10662,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio commandId: input.command.commandId, }); const dying = [...input.dying]; - const childThreadIds = new Set(); + const childThreadIds = input.childThreadIds; for (const { projection, subagent } of dying) { const ended = yield* endOrphanedNativeSubagent({ projection, @@ -10746,7 +10731,25 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }) { const emitEvent = emit(input.events, input.command); const runOrdinals = new Map(input.projection.runs.map((run) => [run.id, run.ordinal])); - const dying: Array = []; + const childThreadIds = new Set(); + const liveness = new Map(); + const hasLiveSession = Effect.fn("Orchestrator.hasLiveSession")(function* ( + providerThreadId: OrchestrationV2ProviderThread["id"], + ) { + const known = liveness.get(providerThreadId); + if (known !== undefined) return known; + const sessionId = input.projection.providerThreads.find( + (candidate) => candidate.id === providerThreadId, + )?.providerSessionId; + const live = + sessionId !== null && + sessionId !== undefined && + Option.isSome( + yield* providerSessions.get(sessionId).pipe(Effect.orElseSucceed(() => Option.none())), + ); + liveness.set(providerThreadId, live); + return live; + }); const stillReportsBackgroundWork = Effect.fnUntraced(function* ( providerThread: OrchestrationV2ProviderThread, ) { @@ -10772,7 +10775,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if ( providerThreadId !== null && providerThreadId !== input.stoppedProviderThreadId && - (yield* hasLiveSession(input.projection.providerThreads, providerThreadId)) + (yield* hasLiveSession(providerThreadId)) ) { continue; } @@ -10791,22 +10794,23 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio updatedAt: input.now, }, }); + if (item.type !== "subagent") continue; // A native subagent died with the process too. - if (item.type === "subagent") dying.push({ projection: input.projection, subagent: item }); + yield* endNativeSubagentTrees({ + command: input.command, + events: input.events, + dying: [{ projection: input.projection, subagent: item }], + childThreadIds, + now: input.now, + }); } - yield* endNativeSubagentTrees({ - command: input.command, - events: input.events, - dying, - now: input.now, - }); for (const providerThread of input.projection.providerThreads) { if ((providerThread.pendingBackgroundTasks?.length ?? 0) === 0) continue; if (providerThread.id === input.stoppedProviderThreadId) { // A refused newer start can take ownership and reject the stopped // turn's roster clear. Reconcile it without hiding surviving work. if (yield* stillReportsBackgroundWork(providerThread)) continue; - } else if (yield* hasLiveSession(input.projection.providerThreads, providerThread.id)) { + } else if (yield* hasLiveSession(providerThread.id)) { continue; } yield* emitEvent({ @@ -12272,8 +12276,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio for (const run of reached.values()) { yield* markStoppedRun({ command, events, thread: projection.thread, run, now }); } - // With nothing left to interrupt, a native subagent of this thread's run still recorded - // as active (no live turn item) would hold its delegated result forever, so Stop ends it. + // Record and item are written together, so an active record without a live item lost its end. const staleNative: Array = []; for (const subagent of target === undefined ? projection.subagents : []) { if ( @@ -12286,29 +12289,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) { continue; } - // A live session still reports the end itself. - if (subagent.providerThreadId !== null) { - const nativeProviderThreads = - subagent.childThreadId === null - ? [] - : yield* projectionStore - .getThreadRecords(subagent.childThreadId, ["providerThreads"]) - .pipe( - Effect.map((child) => child.providerThreads), - Effect.catchTags({ - ProjectionStoreThreadNotFoundError: () => Effect.succeed([]), - }), - mapDispatchError(command), - ); - if ( - yield* hasLiveSession( - [...projection.providerThreads, ...nativeProviderThreads], - subagent.providerThreadId, - ) - ) { - continue; - } - } staleNative.push(nativeSubagentRef(subagent)); } if (staleNative.length > 0) { @@ -12324,6 +12304,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio command, events, dying: staleNative.map((subagent) => ({ projection: threadProjection, subagent })), + childThreadIds: new Set(), now, }); } diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index 5558654c643ce..628a6a30cc2f9 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -25,7 +25,6 @@ import * as Schema from "effect/Schema"; import * as SqlClient from "effect/sql/SqlClient"; import * as Stream from "effect/Stream"; import * as EventStore from "./EventStore.ts"; -import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as TestClock from "effect/testing/TestClock"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; @@ -50,38 +49,7 @@ const adapter = { driver: ProviderDriverKind.make("codex"), getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }), - openSession: (input) => - Effect.gen(function* () { - const now = yield* DateTime.now; - const unused = () => Effect.die("Runs here never reach a provider"); - return { - instanceId, - driver: ProviderDriverKind.make("codex"), - providerSessionId: input.providerSessionId, - providerSession: { - id: input.providerSessionId, - driver: ProviderDriverKind.make("codex"), - providerInstanceId: instanceId, - status: "ready", - cwd: process.cwd(), - model: modelSelection.model, - capabilities: CodexProviderCapabilitiesV2, - createdAt: now, - updatedAt: now, - lastError: null, - }, - events: Stream.never, - ensureThread: unused, - resumeThread: unused, - startTurn: unused, - steerTurn: unused, - interruptTurn: unused, - respondToRuntimeRequest: unused, - readThreadSnapshot: unused, - rollbackThread: unused, - forkThread: unused, - }; - }), + openSession: () => Effect.die("Runs here never reach a provider"), } as ProviderAdapter.ProviderAdapterV2["Service"]; const layerDatabase = SqlitePersistence.layerMemory; const layerTest = ThreadManagementService.layer.pipe( @@ -1777,156 +1745,126 @@ it.effect.each(["Stop", "Force stop"] as const)( // A finished delegated child whose native subagent the provider never reported ending, // with no open turn item, and a nested native subagent on that agent's own thread. -const finishedChildWithNativeSubagent = (liveSessionId?: ProviderSessionId) => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const projections = yield* ProjectionStore.ProjectionStoreV2; - const parentThreadId = ThreadId.make("thread:held-row"); - yield* createWatchingThread(parentThreadId, 30); - yield* send(parentThreadId, "parent", "start_immediately"); - const childThreadId = yield* delegate(parentThreadId, "held child"); - const child = yield* orchestrator.getThreadProjection(childThreadId); - const childRun = child.runs[0]!; - const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!; - const now = yield* DateTime.now; - const nativeThreadId = ThreadId.make("thread:held-row-native"); - const nestedThreadId = ThreadId.make("thread:held-row-nested"); - yield* createWatchingThread(nativeThreadId, 31); - yield* createWatchingThread(nestedThreadId, 32); - const nativeProviderThreadId = ProviderThreadId.make("provider-thread:held-row-native"); - if (liveSessionId !== undefined) { - const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2; - yield* sessions.open({ - threadId: childThreadId, - providerSessionId: liveSessionId, - modelSelection, - runtimePolicy: { runtimeMode: "full-access", interactionMode: "default", cwd: null }, +it.effect( + "Stop settles a delegated row whose finished child still records a running native subagent", + () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const parentThreadId = ThreadId.make("thread:held-row"); + yield* createWatchingThread(parentThreadId, 30); + yield* send(parentThreadId, "parent", "start_immediately"); + const childThreadId = yield* delegate(parentThreadId, "held child"); + const child = yield* orchestrator.getThreadProjection(childThreadId); + const childRun = child.runs[0]!; + const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!; + const now = yield* DateTime.now; + const nativeThreadId = ThreadId.make("thread:held-row-native"); + const nestedThreadId = ThreadId.make("thread:held-row-nested"); + yield* createWatchingThread(nativeThreadId, 31); + yield* createWatchingThread(nestedThreadId, 32); + const nativeProviderThreadId = ProviderThreadId.make("provider-thread:held-row-native"); + const runningRoot = (threadId: ThreadId, id: NodeId) => ({ + ...child.nodes.find((node) => node.id === childRun.rootNodeId)!, + id, + rootNodeId: id, + threadId, + runId: null, + parentNodeId: null, + status: "running" as const, + countsForRun: false, + completedAt: null, }); + const nativeRoot = runningRoot(nativeThreadId, NodeId.make("node:held-row-native-root")); + const nestedRoot = runningRoot(nestedThreadId, NodeId.make("node:held-row-nested-root")); + for (const root of [nativeRoot, nestedRoot]) { + yield* projections.apply({ + id: EventId.make(`event:held-row:${root.id}`), + type: "node.updated", + threadId: root.threadId, + nodeId: root.id, + occurredAt: now, + payload: root, + }); + } yield* projections.apply({ - id: EventId.make("event:held-row:native-provider-thread"), - type: "provider-thread.updated", + id: EventId.make("event:held-row:native-turn"), + type: "provider-turn.updated", threadId: nativeThreadId, + nodeId: nativeRoot.id, occurredAt: now, payload: { - ...child.providerThreads[0]!, - id: nativeProviderThreadId, - appThreadId: nativeThreadId, - providerSessionId: liveSessionId, - status: "active", + id: ProviderTurnId.make("turn:held-row-native"), + providerThreadId: nativeProviderThreadId, + nodeId: nativeRoot.id, + runAttemptId: null, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, }, }); - } - const runningRoot = (threadId: ThreadId, id: NodeId) => ({ - ...child.nodes.find((node) => node.id === childRun.rootNodeId)!, - id, - rootNodeId: id, - threadId, - runId: null, - parentNodeId: null, - status: "running" as const, - countsForRun: false, - completedAt: null, - }); - const nativeRoot = runningRoot(nativeThreadId, NodeId.make("node:held-row-native-root")); - const nestedRoot = runningRoot(nestedThreadId, NodeId.make("node:held-row-nested-root")); - for (const root of [nativeRoot, nestedRoot]) { + const native = { + ...task, + id: NodeId.make("node:held-row-native"), + threadId: childThreadId, + runId: childRun.id, + parentNodeId: childRun.rootNodeId!, + origin: "provider_native" as const, + providerThreadId: nativeProviderThreadId, + childThreadId: nativeThreadId, + status: "running" as const, + completionDelivery: undefined, + }; yield* projections.apply({ - id: EventId.make(`event:held-row:${root.id}`), - type: "node.updated", - threadId: root.threadId, - nodeId: root.id, + id: EventId.make("event:held-row:native"), + type: "subagent.updated", + threadId: childThreadId, + runId: childRun.id, occurredAt: now, - payload: root, + payload: native, }); - } - yield* projections.apply({ - id: EventId.make("event:held-row:native-turn"), - type: "provider-turn.updated", - threadId: nativeThreadId, - nodeId: nativeRoot.id, - occurredAt: now, - payload: { - id: ProviderTurnId.make("turn:held-row-native"), - providerThreadId: nativeProviderThreadId, - nodeId: nativeRoot.id, - runAttemptId: null, - nativeTurnRef: null, - ordinal: 1, - status: "running", - startedAt: now, - completedAt: null, - }, - }); - const native = { - ...task, - id: NodeId.make("node:held-row-native"), - threadId: childThreadId, - runId: childRun.id, - parentNodeId: childRun.rootNodeId!, - origin: "provider_native" as const, - providerThreadId: nativeProviderThreadId, - childThreadId: nativeThreadId, - status: "running" as const, - completionDelivery: undefined, - }; - yield* projections.apply({ - id: EventId.make("event:held-row:native"), - type: "subagent.updated", - threadId: childThreadId, - runId: childRun.id, - occurredAt: now, - payload: native, - }); - yield* projections.apply({ - id: EventId.make("event:held-row:nested"), - type: "subagent.updated", - threadId: nativeThreadId, - occurredAt: now, - payload: { - ...native, - id: NodeId.make("node:held-row-nested"), + yield* projections.apply({ + id: EventId.make("event:held-row:nested"), + type: "subagent.updated", threadId: nativeThreadId, - runId: null, - parentNodeId: nativeRoot.id, - childThreadId: nestedThreadId, - }, - }); - yield* projections.apply({ - id: EventId.make("event:held-row:child-completed"), - type: "run.updated", - threadId: childThreadId, - runId: childRun.id, - occurredAt: now, - payload: { ...childRun, status: "completed", startedAt: now, completedAt: now }, - }); - yield* orchestrator.recoverDelegatedTasks; - assert.equal( - (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]?.status, - "running", - ); - const statuses = Effect.gen(function* () { - const nativeAfter = yield* orchestrator.getThreadProjection(nativeThreadId); - const nestedAfter = yield* orchestrator.getThreadProjection(nestedThreadId); - return { - row: (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]?.status, - native: (yield* orchestrator.getThreadProjection(childThreadId)).subagents.find( - (subagent) => subagent.id === native.id, - )?.status, - nativeRoot: nativeAfter.nodes.find((node) => node.id === nativeRoot.id)?.status, - nested: nativeAfter.subagents[0]?.status, - nestedRoot: nestedAfter.nodes.find((node) => node.id === nestedRoot.id)?.status, - }; - }); - return { parentThreadId, childThreadId, task, statuses }; - }); - -it.effect( - "Stop settles a delegated row whose finished child still records a running native subagent", - () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const { parentThreadId, childThreadId, task, statuses } = - yield* finishedChildWithNativeSubagent(); + occurredAt: now, + payload: { + ...native, + id: NodeId.make("node:held-row-nested"), + threadId: nativeThreadId, + runId: null, + parentNodeId: nativeRoot.id, + childThreadId: nestedThreadId, + }, + }); + yield* projections.apply({ + id: EventId.make("event:held-row:child-completed"), + type: "run.updated", + threadId: childThreadId, + runId: childRun.id, + occurredAt: now, + payload: { ...childRun, status: "completed", startedAt: now, completedAt: now }, + }); + yield* orchestrator.recoverDelegatedTasks; + assert.equal( + (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]?.status, + "running", + ); + const statuses = Effect.gen(function* () { + const nativeAfter = yield* orchestrator.getThreadProjection(nativeThreadId); + const nestedAfter = yield* orchestrator.getThreadProjection(nestedThreadId); + return { + row: (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]?.status, + native: (yield* orchestrator.getThreadProjection(childThreadId)).subagents.find( + (subagent) => subagent.id === native.id, + )?.status, + nativeRoot: nativeAfter.nodes.find((node) => node.id === nativeRoot.id)?.status, + nested: nativeAfter.subagents[0]?.status, + nestedRoot: nestedAfter.nodes.find((node) => node.id === nestedRoot.id)?.status, + }; + }); // The row's Stop: the parent's task, then the child thread. yield* orchestrator.dispatch({ @@ -1952,25 +1890,3 @@ it.effect( }); }).pipe(Effect.provide(layerTest)), ); - -it.effect("Stop leaves a native subagent running while its provider session is live", () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const { childThreadId, statuses } = yield* finishedChildWithNativeSubagent( - ProviderSessionId.make("session:held-row-live"), - ); - - yield* orchestrator.dispatch({ - type: "thread.stop", - commandId: CommandId.make("held-row:stop-live-child"), - threadId: childThreadId, - }); - assert.deepEqual(yield* statuses, { - row: "running", - native: "running", - nativeRoot: "running", - nested: "running", - nestedRoot: "running", - }); - }).pipe(Effect.provide(layerTest)), -); From 240f3678ecfe39c3506ceeccf22ebf4526f2c689 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 20:50:28 +0000 Subject: [PATCH 6/6] refactor(server): filter stale native subagents in Stop directly AI-assisted; verified with the tests and tools listed above. --- .../src/orchestration-v2/Orchestrator.ts | 28 +++++++++---------- 1 file changed, 13 insertions(+), 15 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 254ea1007fa2b..ea276af21fc24 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -10864,7 +10864,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio commandId: input.command.commandId, }); const dying = [...input.dying]; - const childThreadIds = input.childThreadIds; + const { childThreadIds } = input; for (const { projection, subagent } of dying) { const ended = yield* endOrphanedNativeSubagent({ projection, @@ -12479,20 +12479,15 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* markStoppedRun({ command, events, thread: projection.thread, run, now }); } // Record and item are written together, so an active record without a live item lost its end. - const staleNative: Array = []; - for (const subagent of target === undefined ? projection.subagents : []) { - if ( - subagent.origin !== "provider_native" || - subagent.runId === null || - !isOrchestrationV2WorkActive(subagent.status) || - projection.turnItems.some( + const staleNative = (target === undefined ? projection.subagents : []).filter( + (subagent) => + subagent.origin === "provider_native" && + subagent.runId !== null && + isOrchestrationV2WorkActive(subagent.status) && + !projection.turnItems.some( (item) => item.type === "subagent" && item.subagentId === subagent.id, - ) - ) { - continue; - } - staleNative.push(nativeSubagentRef(subagent)); - } + ), + ); if (staleNative.length > 0) { const { nodes } = yield* projectionStore .getThreadRecords(command.threadId, ["nodes"]) @@ -12505,7 +12500,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* endNativeSubagentTrees({ command, events, - dying: staleNative.map((subagent) => ({ projection: threadProjection, subagent })), + dying: staleNative.map((subagent) => ({ + projection: threadProjection, + subagent: nativeSubagentRef(subagent), + })), childThreadIds: new Set(), now, });