From ccbcfc486663e61705febfa4dffd686a28f9f23d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 07:41:38 +0000 Subject: [PATCH 1/2] test(server): a follow-up stays open while its child's nested subagent runs Failing regression for #21837: a t3_thread_send follow-up publishes its interim reply while the child's own delegate_task still runs, and the nested-result wake's final reply never becomes the follow-up's result. AI-assisted; verified with the tests and tools listed above. --- .../DelegatedCompletionDelivery.test.ts | 245 ++++++++++++++++++ 1 file changed, 245 insertions(+) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 5facb16bf26eb..83a2afd57ea2a 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -2936,6 +2936,251 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( }; }); + it.effect("holds a follow-up open while the child's own subagent still runs", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const parentThreadId = ThreadId.make("thread:followup-grandchild-parent"); + const projectId = ProjectId.make("project:followup-grandchild"); + const parentRunId = RunId.make("run:followup-grandchild-parent"); + const rootNodeId = NodeId.make("node:followup-grandchild-parent"); + yield* seedParentWithTerminalTask({ + threadId: parentThreadId, + projectId, + runId: parentRunId, + rootNodeId, + taskId: NodeId.make("node:followup-grandchild-seed"), + deliveryState: "acknowledged", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId, + projectId, + parentRunId, + rootNodeId, + name: "followup-grandchild-child", + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + yield* orchestrator.recoverDelegatedTasks; + assert.equal( + (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (task) => task.id === child.taskId, + )?.status, + "completed", + ); + + const followupMessageId = MessageId.make("message:followup-grandchild-send"); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make("command:followup-grandchild-send"), + threadId: child.childThreadId, + senderThreadId: parentThreadId, + messageId: followupMessageId, + text: "Now delegate the next step.", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + }); + const followupRun = (yield* orchestrator.getThreadProjection( + child.childThreadId, + )).runs.find((run) => run.userMessageId === followupMessageId)!; + assert.isDefined(followupRun.delegatedTaskId); + const followupTaskId = followupRun.delegatedTaskId!; + + const grandchildId = NodeId.make("node:followup-grandchild-grandchild"); + const grandchild = { + id: grandchildId, + threadId: child.childThreadId, + runId: followupRun.id, + parentNodeId: NodeId.make("node:followup-grandchild-child-root"), + origin: "app_owned" as const, + createdBy: "agent" as const, + driver, + providerInstanceId: modelSelection.instanceId, + providerThreadId: null, + childThreadId: ThreadId.make("thread:followup-grandchild-grandchild"), + nativeTaskRef: null, + prompt: "Do the next step.", + title: null, + model: null, + completionWake: "always" as const, + status: "running" as const, + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }; + yield* eventSink.write({ + commandId: reconcileCommandId("followup-grandchild-turn-ended"), + events: [ + { + id: EventId.make("event:followup-grandchild-started"), + type: "subagent.updated", + threadId: child.childThreadId, + runId: followupRun.id, + nodeId: grandchildId, + driver, + providerInstanceId: modelSelection.instanceId, + occurredAt: now, + payload: grandchild, + }, + { + id: EventId.make("event:followup-grandchild-run-completed"), + type: "run.updated", + threadId: child.childThreadId, + runId: followupRun.id, + occurredAt: now, + payload: { ...followupRun, status: "completed", startedAt: now, completedAt: now }, + }, + ], + }); + yield* orchestrator.recoverDelegatedTasks; + + const waiting = yield* orchestrator.getThreadProjection(parentThreadId); + const heldFollowup = waiting.subagents.find((task) => task.id === followupTaskId); + assert.isNull( + heldFollowup?.result ?? null, + "the follow-up closed while the child's own subagent was still running", + ); + assert.equal(heldFollowup?.status, "running"); + assert.equal( + waiting.contextTransfers.filter( + (transfer) => + transfer.type === "subagent_result" && + transfer.sourceThreadId === child.childThreadId, + ).length, + 1, + ); + + // Another turn in the child does not release the hold while the grandchild runs. + for (const status of ["running", "completed"] as const) { + yield* eventSink.write({ + commandId: reconcileCommandId(`followup-grandchild-user-${status}`), + events: [ + runEvent({ + threadId: child.childThreadId, + runId: RunId.make("run:followup-grandchild-user"), + ordinal: followupRun.ordinal + 1, + status, + now, + }), + ], + }); + yield* orchestrator.recoverDelegatedTasks; + assert.isNull( + (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (task) => task.id === followupTaskId, + )?.result ?? null, + `the follow-up closed while another child run was ${status}`, + ); + } + + // The grandchild's completion wakes the child in a run without a delegatedTaskId. + const wakeRunId = RunId.make("run:followup-grandchild-wake"); + const wakeRun = (status: "running" | "completed") => + runEvent({ + threadId: child.childThreadId, + runId: wakeRunId, + ordinal: followupRun.ordinal + 2, + status, + now, + }); + yield* eventSink.write({ + commandId: reconcileCommandId("followup-grandchild-settled"), + events: [ + { + id: EventId.make("event:followup-grandchild-completed"), + type: "subagent.updated", + threadId: child.childThreadId, + runId: followupRun.id, + nodeId: grandchildId, + driver, + providerInstanceId: modelSelection.instanceId, + occurredAt: now, + payload: { + ...grandchild, + status: "completed", + result: "grandchild finished", + completionDelivery: { state: "acknowledged", observedByRunId: wakeRunId }, + completedAt: now, + }, + }, + { + id: EventId.make("event:followup-grandchild-wake-message"), + type: "message.updated", + threadId: child.childThreadId, + runId: wakeRunId, + occurredAt: now, + payload: { + id: MessageId.make(`message:${wakeRunId}`), + threadId: child.childThreadId, + runId: wakeRunId, + nodeId: null, + role: "user", + text: "Delegated task finished.", + attachments: [], + streaming: false, + createdBy: "agent", + creationSource: "server", + delegatedCompletion: { + parentRunId: followupRun.id, + generation: 1, + taskIds: [grandchildId], + }, + createdAt: now, + updatedAt: now, + }, + }, + wakeRun("running"), + ], + }); + yield* orchestrator.recoverDelegatedTasks; + const waking = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (task) => task.id === followupTaskId, + ); + assert.isNull(waking?.result ?? null, "the follow-up closed before the child's wake ended"); + + yield* eventSink.write({ + commandId: reconcileCommandId("followup-grandchild-wake-finished"), + events: [ + { + id: EventId.make("event:followup-grandchild-wake-result"), + type: "message.updated", + threadId: child.childThreadId, + runId: wakeRunId, + occurredAt: now, + payload: { + id: MessageId.make("message:followup-grandchild-wake-result"), + threadId: child.childThreadId, + runId: wakeRunId, + nodeId: null, + role: "assistant", + text: "The grandchild finished the next step.", + attachments: [], + streaming: false, + createdBy: "agent", + creationSource: "server", + createdAt: now, + updatedAt: now, + }, + }, + wakeRun("completed"), + ], + }); + yield* orchestrator.recoverDelegatedTasks; + const settled = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (task) => task.id === followupTaskId, + ); + assert.equal(settled?.status, "completed"); + assert.equal(settled?.result, "The grandchild finished the next step."); + }), + ); + it.effect("waits for a run the follow-up's child queued on its own thread", () => Effect.gen(function* () { const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; From ac7bb5f22813d5cf7c55d6690849af4811661ac5 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 07:54:59 +0000 Subject: [PATCH 2/2] fix(server): a follow-up delegated task waits for the subagents its child starts A t3_thread_send follow-up settled as soon as its own tagged run ended, so the parent got the child's interim reply while a subagent the child started in that turn was still running. The server's wake for that subagent ran untagged, so the child's final reply never reached the parent. A follow-up's runs now include the untagged delegatedCompletion wakes for subagents its runs started. It stays open while those subagents or their deliveries are pending, and settles with the last of its runs. Finalization, delegatedTaskResultPending and task_status share this attribution. Fixes pingdotgg/t3code#21837 AI-assisted; verified with the tests and tools listed above. --- apps/server/src/mcp/OrchestratorMcpService.ts | 25 +++++---- .../src/orchestration-v2/Orchestrator.ts | 19 +++++-- .../orchestration-v2/RestartBackgroundNote.ts | 53 +++++++++++++++++-- .../src/server/subagentProjection.ts | 23 ++++---- 4 files changed, 93 insertions(+), 27 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 844930371fa4f..7d15160bccbf0 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -101,7 +101,12 @@ import type { ProjectionRecordField, ProjectionRecordFilter, } from "../orchestration-v2/ProjectionStore.ts"; -import { delegatedTaskResultRun, runOutcome } from "../orchestration-v2/RestartBackgroundNote.ts"; +import { + delegatedTaskAwaitsSubagents, + delegatedTaskResultRun, + delegatedTaskRuns, + runOutcome, +} from "../orchestration-v2/RestartBackgroundNote.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; @@ -1501,16 +1506,18 @@ const make = Effect.gen(function* () { const childRun = delegatedTaskRun(childControls, task); const terminalRun = latestTerminalResultRun(childControls, childRun); const progress = delegatedTaskProgress(childControls); + const followupRuns = + childRun?.delegatedTaskId === task.id ? delegatedTaskRuns(childControls, task.id) : undefined; const taskResultRun = - childRun?.delegatedTaskId === task.id - ? delegatedTaskResultRun(childControls.runs, task.id) - : progress.resultRun; + followupRuns === undefined ? progress.resultRun : delegatedTaskResultRun(followupRuns); const taskWorkState = - childRun?.delegatedTaskId === task.id - ? isTerminalTaskStatus(taskStatusForRun(taskResultRun)) - ? "result_available" - : "working" - : progress.state; + followupRuns === undefined + ? progress.state + : !isTerminalTaskStatus(taskStatusForRun(taskResultRun)) + ? "working" + : delegatedTaskAwaitsSubagents(followupRuns, childControls.subagents) + ? "waiting_for_children" + : "result_available"; const resultRunIds = [ ...new Set( [taskResultRun?.id, terminalRun?.id].filter((id): id is RunId => id !== undefined), diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index e32f9c0ed2175..a4c2d7a6bcd03 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -126,7 +126,9 @@ import { cancelledBeforeStart, continuationFailedBeforePrompt, continuationStartRetryDelayMs, + delegatedTaskAwaitsSubagents, delegatedTaskResultRun, + delegatedTaskRuns, isRestartContinuationSource, runOutcome, } from "./RestartBackgroundNote.ts"; @@ -13509,13 +13511,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) { return; } - const taskRuns = childControls.runs.filter((run) => run.delegatedTaskId === options.task.id); + const taskRuns = delegatedTaskRuns(childControls, options.task.id); const childRun = taskRuns.length === 0 ? options.task.id === forkedFrom.nodeId ? options.resultRun : undefined - : delegatedTaskResultRun(taskRuns, options.task.id); + : delegatedTaskResultRun(taskRuns); if (childRun === undefined) return; let outcome = runOutcome(childRun); let terminalStatus = delegatedTaskTerminalStatus(outcome); @@ -13575,6 +13577,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ]); return; } + // A follow-up ends with the subagents its turns started, as the original task does. + if (delegatedTaskAwaitsSubagents(taskRuns, childControls.subagents)) return; if ( yield* childAwaitsRestartContinuation( taskRuns.length === 0 ? childControls.runs : taskRuns, @@ -15136,9 +15140,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio { messageRoles: ["user"] }, ); if (taskId !== undefined) { - const runs = child.runs.filter((run) => run.delegatedTaskId === taskId); - const run = delegatedTaskResultRun(runs, taskId); - if (run === undefined || delegatedTaskTerminalStatus(run.status) === null) return true; + const runs = delegatedTaskRuns(child, taskId); + const run = delegatedTaskResultRun(runs); + if ( + run === undefined || + delegatedTaskTerminalStatus(run.status) === null || + delegatedTaskAwaitsSubagents(runs, child.subagents) + ) + return true; return yield* childAwaitsRestartContinuation(runs, run); } const progress = delegatedTaskProgress(child); diff --git a/apps/server/src/orchestration-v2/RestartBackgroundNote.ts b/apps/server/src/orchestration-v2/RestartBackgroundNote.ts index 48029bdc75fd1..3de10f4e944ad 100644 --- a/apps/server/src/orchestration-v2/RestartBackgroundNote.ts +++ b/apps/server/src/orchestration-v2/RestartBackgroundNote.ts @@ -1,12 +1,15 @@ import type { NodeId, + OrchestrationV2ConversationMessage, OrchestrationV2PendingBackgroundTask, OrchestrationV2ProviderTurn, OrchestrationV2RestartCancelledBackgroundWork, OrchestrationV2Run, OrchestrationV2RunAttempt, + OrchestrationV2Subagent, OrchestrationV2TurnItem, } from "@t3tools/contracts"; +import { subagentHoldsCompletion } from "@t3tools/provider-core/server/subagentProjection"; import { latestRootProviderFailure, runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError"; type Work = OrchestrationV2RestartCancelledBackgroundWork; @@ -187,14 +190,56 @@ export function cancelledBeforeStart( return run.status === "cancelled" && neverStarted(run); } +/** A follow-up task's runs: those tagged with it, and the untagged wakes for subagents they started. */ +export function delegatedTaskRuns( + projection: { + readonly runs: ReadonlyArray; + readonly messages: ReadonlyArray< + Pick + >; + }, + taskId: NodeId, +): ReadonlyArray { + const owned = new Set( + projection.runs.filter((run) => run.delegatedTaskId === taskId).map((run) => run.id), + ); + const untagged = new Set( + projection.runs.filter((run) => run.delegatedTaskId === undefined).map((run) => run.id), + ); + for (let grew = owned.size > 0; grew;) { + grew = false; + for (const { runId, delegatedCompletion } of projection.messages) { + if ( + runId === null || + !untagged.has(runId) || + owned.has(runId) || + delegatedCompletion === undefined || + !owned.has(delegatedCompletion.parentRunId) + ) + continue; + owned.add(runId); + grew = true; + } + } + return projection.runs.filter((run) => owned.has(run.id)); +} + +/** A follow-up task's runs still owe a subagent they started its result. */ +export function delegatedTaskAwaitsSubagents( + runs: ReadonlyArray, + subagents: ReadonlyArray< + Pick + >, +): boolean { + const runIds = new Set(runs.map((run) => run.id)); + return subagents.some((task) => runIds.has(task.runId) && subagentHoldsCompletion(task)); +} + /** A follow-up task's latest run, passing over queued work it no longer waits on, as the original task does. */ export function delegatedTaskResultRun( runs: ReadonlyArray, - taskId: NodeId, ): OrchestrationV2Run | undefined { - const taskRuns = runs - .filter((run) => run.delegatedTaskId === taskId) - .toSorted((a, b) => b.ordinal - a.ordinal); + const taskRuns = runs.toSorted((a, b) => b.ordinal - a.ordinal); const awaited = taskRuns.filter( (run) => !cancelledBeforeStart(run) && !(run.status === "queued" && run.queueHeld === true), ); diff --git a/packages/provider-core/src/server/subagentProjection.ts b/packages/provider-core/src/server/subagentProjection.ts index 62145567f6651..ffbbb959563ff 100644 --- a/packages/provider-core/src/server/subagentProjection.ts +++ b/packages/provider-core/src/server/subagentProjection.ts @@ -248,6 +248,19 @@ export function subagentResultForRun( }; } +export function subagentHoldsCompletion( + task: Pick, +): boolean { + return ( + isOrchestrationV2WorkActive(task.status) || + // Publishing a child's result precedes scheduling its parent's wake. + // The parent still owes that follow-up even between those transactions. + (task.status !== "idle" && + (task.completionDelivery?.state === "pending" || + task.completionDelivery?.state === "claimed")) + ); +} + export function delegatedTaskProgress(projection: { readonly runs: OrchestrationV2ThreadProjection["runs"]; readonly messages: ReadonlyArray< @@ -274,15 +287,7 @@ export function delegatedTaskProgress(projection: { (run) => !terminal(run.status) && !(run.status === "queued" && run.queueHeld === true), ); const children = - projection.subagents.some( - (task) => - isOrchestrationV2WorkActive(task.status) || - // Publishing a child's result precedes scheduling its parent's wake. - // The parent still owes that follow-up even between those transactions. - (task.status !== "idle" && - (task.completionDelivery?.state === "pending" || - task.completionDelivery?.state === "claimed")), - ) || + projection.subagents.some(subagentHoldsCompletion) || projection.providerThreads.some((thread) => backgroundWorkHoldsCompletion(thread.pendingBackgroundTasks ?? []), );