authorgravatar for saiansh2525@gmail.comYash Singh <saiansh2525@gmail.com> 2026-10-10 00:10:10-05:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-10 00:10:10-05:00
logc572cb71d2711d8cdb4ff7e0ba88af5e9a23aa81
tree0685f7d7fdcd522f63d0eb460ab56c2ddc377b99
parentfa9f936ef5cbbe4bd4f9890b638766fc92f45075
signature Signed by PGP key B5690EEEBB952194

fix(server): track resumed subagent follow-ups as separate tasks (#17696)

14 files changed, 954 insertions(+), 122 deletions(-)

apps/server/src/mcp/OrchestratorMcpService.test.ts+10-1
......@@ -200,7 +200,16 @@ describe("OrchestratorMcpService", () => {
200200 } as unknown as OrchestrationV2ThreadProjection;
201201 const childProjection = {
202202 thread: { id: childThreadId },
203 runs: [{ id: RunId.make("run:mcp-restart-child"), ordinal: 1, status: "cancelled" }],
203 runs: [
204 {
205 id: RunId.make("run:mcp-restart-child"),
206 ordinal: 1,
207 status: "cancelled",
208 delegatedTaskId: taskId,
209 },
210 // An unrelated later run cannot supply this follow-up's result.
211 { id: RunId.make("run:mcp-restart-later"), ordinal: 2, status: "running" },
212 ],
204213 contextTransfers: [],
205214 messages: [],
206215 subagents: [],
apps/server/src/mcp/OrchestratorMcpService.ts+109-54
......@@ -351,6 +351,8 @@ export function delegatedTaskRun(
351351 childProjection: Pick<OrchestrationV2ThreadProjection, "runs" | "contextTransfers">,
352352 task: OrchestrationV2Subagent,
353353): OrchestrationV2Run | undefined {
354 const taskRun = childProjection.runs.find((run) => run.delegatedTaskId === task.id);
355 if (taskRun !== undefined) return taskRun;
354356 const spawnTransfer = childProjection.contextTransfers.find(
355357 (transfer) =>
356358 transfer.type === "subagent_spawn" &&
......@@ -390,22 +392,62 @@ function isTerminalTaskStatus(
390392 );
391393}
392394
393function directAppOwnedChildTask(
395function directAppOwnedChildTasks(
394396 parent: Pick<OrchestrationV2ThreadProjection, "thread" | "subagents">,
395 target: Pick<OrchestrationV2ThreadProjection, "thread">,
396): OrchestrationV2Subagent | undefined {
397 target: Pick<OrchestrationV2ThreadProjection, "thread" | "runs">,
398 returnedRunIds: ReadonlySet<RunId | null>,
399): ReadonlyArray<OrchestrationV2Subagent> {
397400 if (
398401 target.thread.lineage.parentThreadId !== parent.thread.id ||
399402 target.thread.lineage.relationshipToParent !== "subagent"
400403 ) {
401 return undefined;
404 return [];
402405 }
403 return parent.subagents.find(
406 const original = parent.subagents.find(
404407 (task) =>
405408 task.origin === "app_owned" &&
406409 task.threadId === parent.thread.id &&
410 task.id ===
411 (target.thread.forkedFrom?.type === "node" ? target.thread.forkedFrom.nodeId : undefined) &&
407412 task.childThreadId === target.thread.id,
408413 );
414 const taskIds = new Set(
415 target.runs
416 .filter((run) => returnedRunIds.has(run.id))
417 .map((run) => run.delegatedTaskId ?? original?.id),
418 );
419 return parent.subagents.filter(
420 (task) =>
421 taskIds.has(task.id) &&
422 task.origin === "app_owned" &&
423 task.childThreadId === target.thread.id,
424 );
425}
426
427function taskResultTransfer(
428 parent: Pick<OrchestrationV2ThreadProjection, "thread" | "contextTransfers">,
429 child: Pick<OrchestrationV2ThreadProjection, "runs" | "contextTransfers">,
430 task: OrchestrationV2Subagent,
431) {
432 const originalRun = delegatedTaskRun(child, task);
433 const resultRunIds = new Set(
434 child.runs
435 .filter(
436 (run) =>
437 run.delegatedTaskId === task.id ||
438 (originalRun?.delegatedTaskId === undefined && run.delegatedTaskId === undefined),
439 )
440 .map((run) => run.id),
441 );
442 if (originalRun !== undefined) resultRunIds.add(originalRun.id);
443 return parent.contextTransfers.find(
444 (transfer) =>
445 transfer.type === "subagent_result" &&
446 transfer.sourceThreadId === task.childThreadId &&
447 transfer.targetThreadId === parent.thread.id &&
448 ((transfer.sourcePoint.runId !== undefined && resultRunIds.has(transfer.sourcePoint.runId)) ||
449 (transfer.sourcePoint.runId === undefined && originalRun?.delegatedTaskId === undefined)),
450 );
409451}
410452
411453function pageIncludesTerminalTaskResult(input: {
......@@ -418,12 +460,7 @@ function pageIncludesTerminalTaskResult(input: {
418460 >;
419461 readonly maxChars: number;
420462}): boolean {
421 const transfer = input.parent.contextTransfers.find(
422 (transfer) =>
423 transfer.type === "subagent_result" &&
424 transfer.sourceThreadId === input.target.thread.id &&
425 transfer.targetThreadId === input.parent.thread.id,
426 );
463 const transfer = taskResultTransfer(input.parent, input.target, input.task);
427464 if (transfer === undefined) return false;
428465 const run =
429466 transfer.sourcePoint.runId === undefined
......@@ -1241,9 +1278,21 @@ const make = Effect.gen(function* () {
12411278 const childRun = delegatedTaskRun(childControls, task);
12421279 const terminalRun = latestTerminalResultRun(childControls, childRun);
12431280 const progress = delegatedTaskProgress(childControls);
1281 const taskResultRun =
1282 childRun?.delegatedTaskId === task.id
1283 ? childControls.runs
1284 .filter((run) => run.delegatedTaskId === task.id)
1285 .toSorted((a, b) => b.ordinal - a.ordinal)[0]
1286 : progress.resultRun;
1287 const taskWorkState =
1288 childRun?.delegatedTaskId === task.id
1289 ? isTerminalTaskStatus(taskStatusForRun(taskResultRun))
1290 ? "result_available"
1291 : "working"
1292 : progress.state;
12441293 const resultRunIds = [
12451294 ...new Set(
1246 [progress.resultRun?.id, terminalRun?.id].filter((id): id is RunId => id !== undefined),
1295 [taskResultRun?.id, terminalRun?.id].filter((id): id is RunId => id !== undefined),
12471296 ),
12481297 ];
12491298 const resultRecords = yield* threadManagement
......@@ -1263,12 +1312,12 @@ const make = Effect.gen(function* () {
12631312 // the child started working again after this read.
12641313 const heldForRestart =
12651314 task.result === null &&
1266 progress.state === "result_available" &&
1315 taskWorkState === "result_available" &&
12671316 (yield* threadManagement
1268 .delegatedTaskResultPending(task.childThreadId)
1317 .delegatedTaskResultPending(task.childThreadId, childRun?.delegatedTaskId)
12691318 .pipe(Effect.mapError(threadManagementFailure)));
12701319 const workState =
1271 task.result !== null ? "result_available" : heldForRestart ? "working" : progress.state;
1320 task.result !== null ? "result_available" : heldForRestart ? "working" : taskWorkState;
12721321 const status =
12731322 task.result !== null
12741323 ? taskStatusForRun(
......@@ -1280,15 +1329,15 @@ const make = Effect.gen(function* () {
12801329 : childRun,
12811330 )
12821331 : workState === "result_available"
1283 ? taskStatusForRun(progress.resultRun ?? childRun)
1284 : taskStatusForRun(childRun) === "queued"
1332 ? taskStatusForRun(taskResultRun ?? childRun)
1333 : taskStatusForRun(taskResultRun ?? childRun) === "queued"
12851334 ? "queued"
12861335 : "running";
12871336 const derivedResult =
12881337 task.result !== null
12891338 ? task.result
1290 : progress.resultRun !== undefined && isTerminalTaskStatus(status)
1291 ? subagentResultForRun(childProjection, progress.resultRun).text
1339 : taskResultRun !== undefined && isTerminalTaskStatus(status)
1340 ? subagentResultForRun(childProjection, taskResultRun).text
12921341 : null;
12931342 const resultTransfers = parentProjection.contextTransfers.filter(
12941343 (transfer) =>
......@@ -1304,7 +1353,7 @@ const make = Effect.gen(function* () {
13041353 ? resultTransfers.find((transfer) => transfer.sourcePoint.runId === undefined)
13051354 : undefined) ??
13061355 null);
1307 const resultTransfer = resultTransfers[0] ?? null;
1356 const resultTransfer = taskResultTransfer(parentProjection, childProjection, task) ?? null;
13081357 const terminalStatus = terminalRun === undefined ? null : taskStatusForRun(terminalRun);
13091358 const response = {
13101359 taskId: task.id,
......@@ -2315,45 +2364,48 @@ const make = Effect.gen(function* () {
23152364 { concurrency: 1 },
23162365 );
23172366 const messagesByThreadId = new Map(sourceMessages);
2318 const task = parent === undefined ? undefined : directAppOwnedChildTask(parent, target);
2367 const returnedRunIds = new Set(
2368 page
2369 .filter((row) => row.sourceThreadId === target.thread.id)
2370 .map((row) => row.item.runId),
2371 );
2372 const tasks =
2373 parent === undefined ? [] : directAppOwnedChildTasks(parent, target, returnedRunIds);
23192374 if (
23202375 parent !== undefined &&
23212376 scope.thread !== undefined &&
2322 task !== undefined &&
2377 tasks.length > 0 &&
23232378 (input.textOffset ?? 0) === 0
23242379 ) {
2325 const transfer = parent.contextTransfers.find(
2326 (transfer) =>
2327 transfer.type === "subagent_result" &&
2328 transfer.sourceThreadId === target.thread.id &&
2329 transfer.targetThreadId === parent.thread.id,
2330 );
2331 const resultRunId = transfer?.sourcePoint.runId ?? delegatedTaskRun(target, task)?.id;
2332 const resultRunIds = resultRunId === undefined ? [] : [resultRunId];
2333 const resultRecords = yield* threadManagement
2334 .getThreadRecords(target.thread.id, ["messages", "turnItems"], {
2335 messageRoles: ["assistant"],
2336 messageRunIds: resultRunIds,
2337 turnItemRunIds: resultRunIds,
2338 turnItemTypes: ["assistant_message", "error"],
2339 })
2340 .pipe(Effect.mapError(threadManagementFailure));
2341 if (
2342 pageIncludesTerminalTaskResult({
2343 parent,
2344 page,
2345 task,
2346 target: { ...target, ...resultRecords },
2347 maxChars,
2348 })
2349 ) {
2350 yield* readTask(
2351 scope as McpThreadInvocationScope,
2352 task.id,
2353 false,
2354 true,
2355 "thread-read-acknowledge",
2356 );
2380 for (const task of tasks) {
2381 const transfer = taskResultTransfer(parent, target, task);
2382 const resultRunId = transfer?.sourcePoint.runId ?? delegatedTaskRun(target, task)?.id;
2383 const resultRunIds = resultRunId === undefined ? [] : [resultRunId];
2384 const resultRecords = yield* threadManagement
2385 .getThreadRecords(target.thread.id, ["messages", "turnItems"], {
2386 messageRoles: ["assistant"],
2387 messageRunIds: resultRunIds,
2388 turnItemRunIds: resultRunIds,
2389 turnItemTypes: ["assistant_message", "error"],
2390 })
2391 .pipe(Effect.mapError(threadManagementFailure));
2392 if (
2393 pageIncludesTerminalTaskResult({
2394 parent,
2395 page,
2396 task,
2397 target: { ...target, ...resultRecords },
2398 maxChars,
2399 })
2400 ) {
2401 yield* readTask(
2402 scope as McpThreadInvocationScope,
2403 task.id,
2404 false,
2405 true,
2406 "thread-read-acknowledge",
2407 );
2408 }
23572409 }
23582410 }
23592411 return {
......@@ -2419,6 +2471,9 @@ const make = Effect.gen(function* () {
24192471 threadId: input.threadId,
24202472 messageId,
24212473 runId: result.run.id,
2474 ...(result.run.delegatedTaskId === undefined
2475 ? {}
2476 : { taskId: result.run.delegatedTaskId }),
24222477 status: result.run.status,
24232478 delivery: result.delivery,
24242479 } satisfies OrchestratorMcpThreadSendResult;
apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts+75-5
......@@ -1782,6 +1782,9 @@ describe("orchestrator MCP toolkit", () => {
17821782 expect(delegatedStatus.resultContextTransferId).not.toBeNull();
17831783 expect(delegatedStatus.latestTerminalResultContextTransferId).not.toBeNull();
17841784
1785 // Wait-mode completion already returned its result while the parent was live.
1786 yield* expectOffersToStay(0);
1787
17851788 const childFollowupCall = yield* invoke("t3_thread_send", {
17861789 threadId: delegated.childThreadId,
17871790 message: "Confirm the delegated API boundary remains inspected.",
......@@ -1924,10 +1927,14 @@ describe("orchestrator MCP toolkit", () => {
19241927 latestTerminalStatus: "interrupted",
19251928 });
19261929
1927 // A wait-mode child (completionWake settled_only) that completes
1928 // while the parent run is live does not offer a wake: the
1929 // blocking delegate_task call above already returned the result.
1930 yield* expectOffersToStay(0);
1930 // Later requested work has its own delivery, including an interrupted follow-up.
1931 expect(childFollowup.taskId).toBeDefined();
1932 expect(activeChildFollowup.taskId).toBeDefined();
1933 yield* waitForContinuationOffers(2);
1934 yield* expectOffersToStay(2);
1935 yield* invoke("task_status", { taskId: childFollowup.taskId });
1936 yield* invoke("task_status", { taskId: activeChildFollowup.taskId });
1937 yield* Ref.set(continuationOffers, []);
19311938
19321939 const repeatedDelegatedCall = yield* invoke("delegate_task", {
19331940 task: delegatedPrompt,
......@@ -4007,6 +4014,18 @@ describe("orchestrator MCP toolkit", () => {
40074014 const pendingProjection = yield* orchestrator.getThreadProjection(
40084015 delegated.childThreadId,
40094016 );
4017 const followupTasks = (yield* orchestrator.getThreadProjection(
4018 parentThreadId,
4019 )).subagents.filter(
4020 (task) =>
4021 task.childThreadId === delegated.childThreadId && task.id !== delegated.taskId,
4022 );
4023 expect(followupTasks).toHaveLength(2);
4024 expect(followupTasks.every((task) => task.result === null)).toBe(true);
4025 expect(followupTasks.map((task) => task.id)).toEqual([
4026 runningFollowup.taskId,
4027 queuedFollowup.taskId,
4028 ]);
40104029 expect(
40114030 pendingProjection.runs.find((run) => run.id === delegated.childRunId)?.status,
40124031 ).toBe("completed");
......@@ -4082,6 +4101,47 @@ describe("orchestrator MCP toolkit", () => {
40824101 expect(finalProjection.runs.find((run) => run.id === queuedFollowup.runId)?.status).toBe(
40834102 "completed",
40844103 );
4104 yield* orchestrator
4105 .streamStoredEventsFrom({
4106 threadId: parentThreadId,
4107 afterSequence: finalSequence,
4108 })
4109 .pipe(
4110 Stream.filter(
4111 (stored) =>
4112 stored.event.type === "subagent.updated" &&
4113 stored.event.payload.id === followupTasks[1]!.id &&
4114 stored.event.payload.status === "completed",
4115 ),
4116 Stream.runHead,
4117 Effect.flatMap(
4118 Option.match({
4119 onNone: () => Effect.die("Follow-up task did not settle."),
4120 onSome: () => Effect.void,
4121 }),
4122 ),
4123 );
4124 const finishedParent = yield* orchestrator.getThreadProjection(parentThreadId);
4125 expect(
4126 finishedParent.subagents.find((task) => task.id === followupTasks[0]!.id)?.status,
4127 ).toBe("interrupted");
4128 expect(
4129 finishedParent.subagents.find((task) => task.id === followupTasks[1]!.id),
4130 ).toMatchObject({
4131 status: "completed",
4132 result: queuedFollowupResult,
4133 });
4134 expect(
4135 finishedParent.contextTransfers.filter(
4136 (transfer) =>
4137 transfer.type === "subagent_result" &&
4138 transfer.sourceThreadId === delegated.childThreadId,
4139 ),
4140 ).toHaveLength(3);
4141 expect(
4142 finishedParent.subagents.find((task) => task.id === followupTasks[1]!.id)
4143 ?.completionDelivery?.state,
4144 ).toBe("claimed");
40854145 const finalStatusCall = yield* invoke("task_status", {
40864146 taskId: delegated.taskId,
40874147 });
......@@ -4097,8 +4157,18 @@ describe("orchestrator MCP toolkit", () => {
40974157 latestTerminalRunId: queuedFollowup.runId,
40984158 latestTerminalStatus: "completed",
40994159 latestTerminalSummary: queuedFollowupResult,
4100 latestTerminalResultContextTransferId: null,
41014160 });
4161 expect(finalStatus.latestTerminalResultContextTransferId).not.toBeNull();
4162 const read = yield* invoke("t3_thread_read", {
4163 threadId: delegated.childThreadId,
4164 limit: 100,
4165 });
4166 expect(read.isError).not.toBe(true);
4167 const afterRead = yield* orchestrator.getThreadProjection(parentThreadId);
4168 expect(
4169 afterRead.subagents.find((task) => task.id === queuedFollowup.taskId)
4170 ?.completionDelivery?.state,
4171 ).toBe("acknowledged");
41024172 }).pipe(Effect.provide(layerTest));
41034173 }),
41044174 ),
apps/server/src/mcp/toolkits/orchestrator/tools.test.ts+1-1
......@@ -22,7 +22,7 @@ describe("orchestrator MCP tool guidance", () => {
2222 assert.include(DelegateTaskTool.description ?? "", "childThreadId is backing storage");
2323 assert.include(
2424 OrchestratorToolkit.tools.t3_thread_send.description ?? "",
25 "Do not use a delegated task's childThreadId to start another review round",
25 "a requested follow-up creates a fresh task",
2626 );
2727 assert.include(
2828 OrchestratorToolkit.tools.task_cancel.description ?? "",
apps/server/src/mcp/toolkits/orchestrator/tools.ts+1-1
......@@ -219,7 +219,7 @@ export const ThreadUpdateTool = Tool.make("t3_thread_update", {
219219
220220const ThreadSendTool = Tool.make("t3_thread_send", {
221221 description:
222 "Send a message to any T3 thread in this environment. The target cannot have broader permission modes than the caller. Do not use a delegated task's childThreadId to start another review round here; use delegate_task with the full review context and a new clientRequestId for that round. Thread messages do not create a new delegated task or reopen a completed task. mode='auto' starts an idle thread, steers a fully active turn, or queues behind a turn that is not yet steerable. Use queue for a separate follow-up turn, steer for an in-flight update, or restart to interrupt-and-restart the active turn. clientRequestId makes retries idempotent.",
222 "Send a message to any T3 thread in this environment. The target cannot have broader permission modes than the caller. Use delegate_task with the full review context and a new clientRequestId for independent review rounds. For a follow-up that needs the same child's conversation, send to its childThreadId. After the original task completes, a requested follow-up creates a fresh task in that child thread and returns taskId; its completion notifies the parent, and the original result stays unchanged. mode='auto' starts an idle thread, steers a fully active turn, or queues behind a turn that is not yet steerable. Use queue for a separate follow-up turn, steer for an in-flight update, or restart to interrupt-and-restart the active turn. clientRequestId makes retries idempotent.",
223223 parameters: OrchestratorMcpThreadSendInput,
224224 success: OrchestratorMcpThreadSendResult,
225225 failure: OrchestratorMcpFailure,
apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts+313
......@@ -41,6 +41,7 @@ import { continueRestartedRun } from "./RestartContinuation.ts";
4141import * as RuntimeLayer from "./runtimeLayer.ts";
4242import * as ProviderTurnStartServiceTestkit from "./ProviderTurnStartService.testkit.ts";
4343import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
44import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
4445
4546const layerPlatformTest = Layer.merge(
4647 NodeServices.layer,
......@@ -904,6 +905,318 @@ const seedRestartCancelledChild = (input: {
904905 return { taskId, childThreadId, childRunId };
905906 });
906907
908it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))(
909 "subagent follow-ups",
910 (it) => {
911 const outcomes = [
912 "completed",
913 "failed",
914 "interrupted",
915 "restarted",
916 "stopped",
917 "overlapping",
918 ] as const;
919 it.effect.each(outcomes)(
920 "delivers a %s follow-up independently of the original task",
921 (outcome) =>
922 Effect.gen(function* () {
923 const orchestrator = yield* Orchestrator.OrchestratorV2;
924 const eventSink = yield* EventSink.EventSinkV2;
925 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
926 const now = yield* DateTime.now;
927 const parentThreadId = ThreadId.make(`thread:followup-${outcome}`);
928 const projectId = ProjectId.make(`project:followup-${outcome}`);
929 const parentRunId = RunId.make(`run:followup-${outcome}`);
930 const rootNodeId = NodeId.make(`node:followup-${outcome}`);
931 yield* seedParentWithTerminalTask({
932 threadId: parentThreadId,
933 projectId,
934 runId: parentRunId,
935 rootNodeId,
936 taskId: NodeId.make(`node:seed-${outcome}`),
937 deliveryState: "acknowledged",
938 now,
939 });
940 const child = yield* seedRestartCancelledChild({
941 parentThreadId,
942 projectId,
943 parentRunId,
944 rootNodeId,
945 name: `followup-child-${outcome}`,
946 completionWake: "always",
947 continuationPending: false,
948 runStatus: "completed",
949 now,
950 });
951 yield* orchestrator.recoverDelegatedTasks;
952 let firstWake = yield* requests.take;
953 while (firstWake.threadId !== parentThreadId) firstWake = yield* requests.take;
954 assert.equal(firstWake.delegatedCompletion?.parentRunId, parentRunId);
955 yield* orchestrator.dispatch({
956 type: "delegated_task.completion-delivery.acknowledge",
957 commandId: CommandId.make(`command:followup-ack-${outcome}`),
958 parentThreadId,
959 taskId: child.taskId,
960 observedByRunId: parentRunId,
961 });
962 const original = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find(
963 (task) => task.id === child.taskId,
964 )!;
965 const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.find(
966 (run) => run.id === parentRunId,
967 )!;
968 yield* eventSink.write({
969 commandId: reconcileCommandId(`followup-parent-finished-${outcome}`),
970 events: [
971 {
972 id: EventId.make(`event:followup-parent-finished-${outcome}`),
973 type: "run.updated",
974 threadId: parentThreadId,
975 runId: parentRunId,
976 occurredAt: now,
977 payload: { ...parentRun, status: "completed", completedAt: now },
978 },
979 ],
980 });
981 const messageId = MessageId.make(`message:followup-${outcome}`);
982 yield* orchestrator.dispatch({
983 type: "message.dispatch",
984 createdBy: outcome === "failed" ? "user" : "agent",
985 creationSource: outcome === "failed" ? "web" : "mcp",
986 commandId: CommandId.make(`command:followup-send-${outcome}`),
987 threadId: child.childThreadId,
988 senderThreadId: parentThreadId,
989 messageId,
990 text: "Check the same work again.",
991 attachments: [],
992 dispatchMode: { type: "start_immediately" },
993 });
994 const childProjection = yield* orchestrator.getThreadProjection(child.childThreadId);
995 let run = childProjection.runs.find((row) => row.userMessageId === messageId)!;
996 assert.isDefined(run.delegatedTaskId);
997 const workingParent = yield* orchestrator.getShellSnapshot();
998 const backgroundTask = workingParent.threads
999 .find((thread) => thread.id === parentThreadId)
1000 ?.pendingBackgroundTasks?.find((task) => task.kind === "subagent");
1001 assert.equal(backgroundTask?.childThreadId, child.childThreadId);
1002 if (outcome === "overlapping") {
1003 yield* orchestrator.dispatch({
1004 type: "message.dispatch",
1005 createdBy: "agent",
1006 creationSource: "mcp",
1007 commandId: CommandId.make("command:overlapping-followup"),
1008 threadId: child.childThreadId,
1009 senderThreadId: parentThreadId,
1010 messageId: MessageId.make("message:overlapping-followup"),
1011 text: "Keep working on another follow-up.",
1012 attachments: [],
1013 dispatchMode: { type: "queue_after_active" },
1014 });
1015 }
1016 if (outcome === "restarted") {
1017 const cut = { ...run, status: "cancelled" as const, startedAt: now, completedAt: now };
1018 yield* eventSink.writeWithEffects({
1019 commandId: reconcileCommandId("followup-restart-cut"),
1020 events: [
1021 {
1022 id: EventId.make("event:followup-restart-cut"),
1023 type: "run.updated",
1024 threadId: child.childThreadId,
1025 runId: run.id,
1026 occurredAt: now,
1027 payload: cut,
1028 },
1029 ],
1030 effects: [
1031 {
1032 id: `effect:restart-continuation:${run.id}`,
1033 commandId: reconcileCommandId("followup-restart-cut"),
1034 threadId: child.childThreadId,
1035 request: { type: "provider-runtime.continue", sourceRunId: run.id },
1036 },
1037 ],
1038 });
1039 yield* orchestrator.recoverDelegatedTasks;
1040 assert.isNull(
1041 (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find(
1042 (task) => task.id === run.delegatedTaskId,
1043 )!.result,
1044 );
1045 const continuationMessageId = MessageId.make("message:followup-restart-continuation");
1046 yield* orchestrator.dispatch({
1047 type: "message.dispatch",
1048 createdBy: "agent",
1049 creationSource: "server",
1050 commandId: CommandId.make("command:followup-restart-continuation"),
1051 threadId: child.childThreadId,
1052 restartContinuationOfRunId: run.id,
1053 messageId: continuationMessageId,
1054 text: "Continue where you left off.",
1055 attachments: [],
1056 dispatchMode: { type: "start_immediately" },
1057 });
1058 const continuation = (yield* orchestrator.getThreadProjection(
1059 child.childThreadId,
1060 )).runs.find((row) => row.userMessageId === continuationMessageId)!;
1061 assert.equal(continuation.delegatedTaskId, run.delegatedTaskId);
1062 run = continuation;
1063 }
1064 if (outcome === "stopped") {
1065 yield* orchestrator.dispatch({
1066 type: "thread.stop",
1067 threadId: parentThreadId,
1068 commandId: CommandId.make(`command:followup-stop-${outcome}`),
1069 });
1070 }
1071 // Reconciliation skips the live terminal listener; recovery must finish this recorded task.
1072 yield* eventSink.write({
1073 commandId: reconcileCommandId(`followup-finish-${outcome}`),
1074 events: [
1075 {
1076 id: EventId.make(`event:followup-finish-${outcome}`),
1077 type: "run.updated",
1078 threadId: child.childThreadId,
1079 runId: run.id,
1080 occurredAt: now,
1081 payload: {
1082 ...run,
1083 status:
1084 outcome === "stopped"
1085 ? "interrupted"
1086 : outcome === "restarted" || outcome === "overlapping"
1087 ? "completed"
1088 : outcome,
1089 startedAt: now,
1090 completedAt: now,
1091 },
1092 },
1093 ],
1094 });
1095 yield* orchestrator.recoverDelegatedTasks;
1096 const finished = yield* orchestrator.getThreadProjection(parentThreadId);
1097 assert.deepEqual(
1098 finished.subagents.find((task) => task.id === child.taskId),
1099 original,
1100 );
1101 const followup = finished.subagents.find((task) => task.id === run.delegatedTaskId)!;
1102 if (outcome === "overlapping") {
1103 const later = finished.subagents.find(
1104 (task) =>
1105 task.childThreadId === child.childThreadId &&
1106 task.id !== child.taskId &&
1107 task.id !== followup.id,
1108 )!;
1109 assert.equal(later.status, "pending");
1110 assert.isNull(later.result);
1111 }
1112 assert.equal(
1113 followup.status,
1114 outcome === "stopped"
1115 ? "interrupted"
1116 : outcome === "restarted" || outcome === "overlapping"
1117 ? "completed"
1118 : outcome,
1119 );
1120 assert.equal(
1121 followup.completionDelivery?.state,
1122 outcome === "stopped" ? "disposed" : "claimed",
1123 );
1124 if (outcome !== "stopped") {
1125 let wake = yield* requests.take;
1126 while (
1127 wake.threadId !== parentThreadId ||
1128 wake.delegatedCompletion?.messageId === firstWake.delegatedCompletion?.messageId
1129 ) {
1130 wake = yield* requests.take;
1131 }
1132 assert.equal(wake.delegatedCompletion?.parentRunId, parentRunId);
1133 assert.notEqual(
1134 wake.delegatedCompletion?.messageId,
1135 firstWake.delegatedCompletion?.messageId,
1136 );
1137 }
1138 assert.equal(
1139 finished.contextTransfers.filter(
1140 (transfer) =>
1141 transfer.type === "subagent_result" &&
1142 transfer.sourceThreadId === child.childThreadId,
1143 ).length,
1144 2,
1145 );
1146 yield* orchestrator.recoverDelegatedTasks;
1147 const recovered = yield* orchestrator.getThreadProjection(parentThreadId);
1148 assert.equal(recovered.contextTransfers.length, finished.contextTransfers.length);
1149 if (outcome === "completed") {
1150 const automaticMessageId = MessageId.make("message:followup-automatic-wake");
1151 yield* orchestrator.dispatch({
1152 type: "message.dispatch",
1153 createdBy: "agent",
1154 creationSource: "server",
1155 commandId: CommandId.make("command:followup-automatic-wake"),
1156 threadId: child.childThreadId,
1157 messageId: automaticMessageId,
1158 text: "A monitor reported new output.",
1159 attachments: [],
1160 notification: {
1161 source: { kind: "monitor" },
1162 outcome: "updated",
1163 summary: "Monitor updated",
1164 },
1165 dispatchMode: { type: "queue_after_active" },
1166 });
1167 const automatic = (yield* orchestrator.getThreadProjection(
1168 child.childThreadId,
1169 )).runs.find((run) => run.userMessageId === automaticMessageId)!;
1170 assert.isUndefined(automatic.delegatedTaskId);
1171 yield* eventSink.write({
1172 commandId: reconcileCommandId("followup-automatic-finished"),
1173 events: [
1174 {
1175 id: EventId.make("event:followup-automatic-finished"),
1176 type: "run.updated",
1177 threadId: child.childThreadId,
1178 runId: automatic.id,
1179 occurredAt: now,
1180 payload: { ...automatic, status: "completed", startedAt: now, completedAt: now },
1181 },
1182 ],
1183 });
1184 yield* orchestrator.recoverDelegatedTasks;
1185 const afterAutomatic = yield* orchestrator.getThreadProjection(parentThreadId);
1186 assert.deepEqual(afterAutomatic.subagents, recovered.subagents);
1187 assert.equal(afterAutomatic.contextTransfers.length, recovered.contextTransfers.length);
1188 const control = yield* seedRestartCancelledChild({
1189 parentThreadId,
1190 projectId,
1191 parentRunId,
1192 rootNodeId,
1193 name: "followup-control-child",
1194 completionWake: "always",
1195 continuationPending: false,
1196 runStatus: "completed",
1197 now,
1198 });
1199 yield* orchestrator.recoverDelegatedTasks;
1200 const afterControl = yield* orchestrator.getThreadProjection(parentThreadId);
1201 assert.equal(
1202 afterControl.subagents.find((task) => task.id === control.taskId)?.completionDelivery
1203 ?.state,
1204 "claimed",
1205 );
1206 assert.equal(
1207 afterControl.contextTransfers.filter(
1208 (transfer) =>
1209 transfer.type === "subagent_result" &&
1210 transfer.sourceThreadId === control.childThreadId,
1211 ).length,
1212 1,
1213 );
1214 }
1215 }),
1216 );
1217 },
1218);
1219
9071220it.layer(layerTest)("delegated tasks across a server restart", (it) => {
9081221 it.effect("holds a restart-cancelled child for its continuation's result", () =>
9091222 Effect.gen(function* () {
apps/server/src/orchestration-v2/Orchestrator.ts+262-16
......@@ -22,6 +22,7 @@ import {
2222 CommandId,
2323 isProviderNativeSubagentThread,
2424 MessageId,
25 type NodeId,
2526 type ModelSelection,
2627 OrchestrationV2Command,
2728 type OrchestrationV2InternalCommand,
......@@ -279,9 +280,11 @@ export interface OrchestratorV2Shape {
279280 /**
280281 * Whether a delegated child's apparent result is not final yet: a restart
281282 * continuation is pending, or the child is working again.
283 * A taskId limits the check to that follow-up's runs.
282284 */
283285 readonly delegatedTaskResultPending: (
284286 childThreadId: ThreadId,
287 taskId?: NodeId,
285288 ) => Effect.Effect<boolean, OrchestratorProjectionError>;
286289 readonly dispatch: (
287290 command: OrchestrationV2ServerCommand,
......@@ -4561,6 +4564,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
45614564 }
45624565 }
45634566
4567 const continuedTaskId = projection.runs.find(
4568 (run) => run.id === command.restartContinuationOfRunId,
4569 )?.delegatedTaskId;
4570
45644571 if (projection.thread.settledOverride !== null) {
45654572 const now = yield* DateTime.now;
45664573 const thread: OrchestrationV2AppThread = {
......@@ -4980,6 +4987,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
49804987 ...(command.restartContinuationOfRunId === undefined
49814988 ? {}
49824989 : { restartContinuationOfRunId: command.restartContinuationOfRunId }),
4990 ...(continuedTaskId === undefined ? {} : { delegatedTaskId: continuedTaskId }),
49834991 };
49844992 const attempt: OrchestrationV2RunAttempt = {
49854993 id: attemptId,
......@@ -5322,6 +5330,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
53225330 ...(dispatchMode.type === "defer_start" && dispatchMode.workspaceStrategy !== undefined
53235331 ? { workspacePreparation: dispatchMode.workspaceStrategy }
53245332 : {}),
5333 ...(continuedTaskId === undefined ? {} : { delegatedTaskId: continuedTaskId }),
53255334 ...wakeWorkStartedAt(projection.runs, command),
53265335 };
53275336 const attempt: OrchestrationV2RunAttempt = {
......@@ -6015,6 +6024,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
60156024 ...(command.restartContinuationOfRunId === undefined
60166025 ? {}
60176026 : { restartContinuationOfRunId: command.restartContinuationOfRunId }),
6027 ...(continuedTaskId === undefined ? {} : { delegatedTaskId: continuedTaskId }),
60186028 ...wakeWorkStartedAt(projection.runs, command),
60196029 };
60206030 const attempt: OrchestrationV2RunAttempt = {
......@@ -6738,6 +6748,165 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
67386748 },
67396749 );
67406750
6751 // A follow-up is new requested work in the same child conversation. Give it
6752 // its own task so results, acknowledgments, Stop, and recovery use the same
6753 // lifecycle as a fresh delegation without replacing the original result.
6754 const recordSubagentFollowup = Effect.fnUntraced(function* (
6755 command: Extract<OrchestrationV2Command, { readonly type: "message.dispatch" }>,
6756 events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
6757 ) {
6758 if (
6759 command.scheduledTaskId !== undefined ||
6760 command.notification !== undefined ||
6761 command.restartContinuationOfRunId !== undefined ||
6762 (command.createdBy !== "user" && command.creationSource !== "mcp")
6763 )
6764 return;
6765 const pending = yield* Ref.get(events);
6766 const message = pending
6767 .flatMap((event) =>
6768 event.type === "message.updated" && event.payload.id === command.messageId
6769 ? [event.payload]
6770 : [],
6771 )
6772 .at(-1);
6773 if (message?.runId == null) return;
6774 const child = yield* projectionStore.getThread(command.threadId);
6775 const parentThreadId = child.lineage.parentThreadId;
6776 if (
6777 child.lineage.relationshipToParent !== "subagent" ||
6778 parentThreadId === null ||
6779 (command.createdBy !== "user" && command.senderThreadId !== parentThreadId)
6780 )
6781 return;
6782 const run =
6783 pending
6784 .flatMap((event) =>
6785 (event.type === "run.created" || event.type === "run.updated") &&
6786 event.payload.id === message.runId
6787 ? [event.payload]
6788 : [],
6789 )
6790 .at(-1) ??
6791 (yield* projectionStore.getThreadRecords(command.threadId, ["runs"])).runs.find(
6792 (row) => row.id === message.runId,
6793 );
6794 if (run === undefined || run.delegatedTaskId !== undefined) return;
6795 const parent = yield* projectionStore
6796 .getThreadRecords(parentThreadId, ["runs", "subagents", "providerTurns", "attempts"])
6797 .pipe(mapDispatchError(command));
6798 const original = parent.subagents.find(
6799 (task) =>
6800 task.origin === "app_owned" &&
6801 task.childThreadId === child.id &&
6802 task.id === (child.forkedFrom?.type === "node" ? child.forkedFrom.nodeId : undefined),
6803 );
6804 // Before first publication, follow-ups remain part of the original task.
6805 if (original === undefined || original.result === null) return;
6806 const owner =
6807 parent.runs.filter(isBlockingRun).toSorted((a, b) => b.ordinal - a.ordinal)[0] ??
6808 latestExecutedRun(parent.runs);
6809 if (owner === null || owner.rootNodeId === null) return;
6810 const now = message.createdAt;
6811 const adapter = yield* providerAdapters.get(run.providerInstanceId);
6812 const taskId = idAllocator.derive.delegatedTaskNode({ commandId: command.commandId });
6813 const task: OrchestrationV2Subagent = {
6814 ...original,
6815 createdBy: command.createdBy,
6816 driver: adapter.driver,
6817 id: taskId,
6818 runId: owner.id,
6819 parentNodeId: owner.rootNodeId,
6820 providerInstanceId: run.providerInstanceId,
6821 providerThreadId: run.providerThreadId,
6822 model: run.modelSelection.model,
6823 modelSelection: run.modelSelection,
6824 prompt: command.text,
6825 title: child.title,
6826 completionWake: "always",
6827 completionDelivery: { state: "pending", observedByRunId: null },
6828 status: run.status === "queued" ? "pending" : "running",
6829 progress: undefined,
6830 result: null,
6831 startedAt: now,
6832 completedAt: null,
6833 updatedAt: now,
6834 };
6835 const node: OrchestrationV2ExecutionNode = {
6836 id: taskId,
6837 threadId: parentThreadId,
6838 runId: owner.id,
6839 parentNodeId: owner.rootNodeId,
6840 rootNodeId: owner.rootNodeId,
6841 kind: "subagent",
6842 status: task.status,
6843 countsForRun: false,
6844 providerThreadId: run.providerThreadId,
6845 providerTurnId: null,
6846 nativeItemRef: null,
6847 runtimeRequestId: null,
6848 checkpointScopeId: null,
6849 startedAt: now,
6850 completedAt: null,
6851 };
6852 const turnItem: OrchestrationV2TurnItem = {
6853 id: idAllocator.derive.delegatedTaskTurnItem({ commandId: command.commandId }),
6854 threadId: parentThreadId,
6855 runId: owner.id,
6856 nodeId: taskId,
6857 providerThreadId: owner.providerThreadId,
6858 providerTurnId: providerTurnForRun(parent, owner)?.id ?? null,
6859 nativeItemRef: null,
6860 parentItemId: null,
6861 ordinal: yield* nextTurnItemOrdinal(parent),
6862 status: task.status,
6863 title: child.title,
6864 startedAt: now,
6865 completedAt: null,
6866 updatedAt: now,
6867 type: "subagent",
6868 subagentId: taskId,
6869 origin: "app_owned",
6870 driver: task.driver,
6871 providerInstanceId: task.providerInstanceId,
6872 childThreadId: child.id,
6873 prompt: command.text,
6874 result: null,
6875 };
6876 const emitEvent = emit(events, command);
6877 yield* emitEvent({
6878 type: "run.updated",
6879 threadId: child.id,
6880 runId: run.id,
6881 occurredAt: now,
6882 payload: { ...run, delegatedTaskId: taskId },
6883 });
6884 yield* emitEvent({
6885 type: "node.updated",
6886 threadId: parentThreadId,
6887 runId: owner.id,
6888 nodeId: taskId,
6889 occurredAt: now,
6890 payload: node,
6891 });
6892 yield* emitEvent({
6893 type: "subagent.updated",
6894 threadId: parentThreadId,
6895 runId: owner.id,
6896 nodeId: taskId,
6897 occurredAt: now,
6898 payload: task,
6899 });
6900 yield* emitEvent({
6901 type: "turn-item.updated",
6902 threadId: parentThreadId,
6903 runId: owner.id,
6904 nodeId: taskId,
6905 occurredAt: now,
6906 payload: turnItem,
6907 });
6908 });
6909
67416910 // Rewrites a delegated task's completionWake after creation. The wait path
67426911 // uses this when its blocking window ends without a terminal (timeout), so
67436912 // a child that later terminalizes mid-parent-turn still wakes the parent.
......@@ -9581,16 +9750,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
95819750 * running). `settledContinuationOf` names the restart continuation that just
95829751 * declined or failed, so its own still-running effect does not hold the child.
95839752 */
9584 const finalizeAppOwnedSubagent = (
9585 childThreadId: ThreadId,
9586 options?: { readonly settledContinuationOf?: RunId },
9753 const finalizeAppOwnedSubagentTask = (
9754 childControls: ProjectionRecords<
9755 "runs" | "messages" | "subagents" | "providerThreads" | "providerTurns" | "attempts"
9756 >,
9757 options: {
9758 readonly settledContinuationOf?: RunId;
9759 readonly taskId: NodeId;
9760 readonly resultRun?: OrchestrationV2Run;
9761 },
95879762 ) =>
95889763 Effect.gen(function* () {
9589 const childControls = yield* projectionStore.getThreadRecords(
9590 childThreadId,
9591 ["runs", "messages", "subagents", "providerThreads", "providerTurns", "attempts"],
9592 { messageRoles: ["user"] },
9593 );
9764 const childThreadId = childControls.thread.id;
95949765 const forkedFrom = childControls.thread.forkedFrom;
95959766 if (
95969767 childControls.thread.lineage.relationshipToParent !== "subagent" ||
......@@ -9599,9 +9770,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
95999770 ) {
96009771 return;
96019772 }
9602 const progress = delegatedTaskProgress(childControls);
9603 if (progress.state !== "result_available") return;
9604 const childRun = progress.resultRun;
9773 const taskRuns = childControls.runs.filter((run) => run.delegatedTaskId === options.taskId);
9774 const childRun =
9775 taskRuns.length === 0
9776 ? options.taskId === forkedFrom.nodeId
9777 ? options.resultRun
9778 : undefined
9779 : taskRuns.toSorted((a, b) => b.ordinal - a.ordinal)[0];
96059780 if (childRun === undefined) return;
96069781 const terminalStatus = delegatedTaskTerminalStatus(childRun.status);
96079782 if (terminalStatus === null) {
......@@ -9609,7 +9784,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
96099784 }
96109785 if (
96119786 yield* childAwaitsRestartContinuation(
9612 childControls.runs,
9787 taskRuns.length === 0 ? childControls.runs : taskRuns,
96139788 childRun,
96149789 options?.settledContinuationOf,
96159790 )
......@@ -9649,7 +9824,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
96499824 );
96509825 const task = parentProjection.subagents.find(
96519826 (candidate) =>
9652 candidate.id === forkedFrom.nodeId &&
9827 candidate.id === options.taskId &&
96539828 candidate.origin === "app_owned" &&
96549829 candidate.childThreadId === childThreadId,
96559830 );
......@@ -9660,7 +9835,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
96609835 (transfer) =>
96619836 transfer.type === "subagent_result" &&
96629837 transfer.sourceThreadId === childThreadId &&
9663 transfer.targetThreadId === parentThreadId,
9838 transfer.targetThreadId === parentThreadId &&
9839 (transfer.sourcePoint.runId === childRun.id ||
9840 (options.taskId === forkedFrom.nodeId && transfer.sourcePoint.runId === undefined)),
96649841 );
96659842 if (existingResultTransfer !== undefined) {
96669843 return;
......@@ -9873,6 +10050,49 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
987310050 }
987410051 });
987510052
10053 const finalizeAppOwnedSubagent = (
10054 childThreadId: ThreadId,
10055 options?: { readonly settledContinuationOf?: RunId },
10056 ) =>
10057 Effect.gen(function* () {
10058 const childControls = yield* projectionStore.getThreadRecords(
10059 childThreadId,
10060 ["runs", "messages", "subagents", "providerThreads", "providerTurns", "attempts"],
10061 { messageRoles: ["user"] },
10062 );
10063 const parentThreadId = childControls.thread.lineage.parentThreadId;
10064 if (
10065 childControls.thread.lineage.relationshipToParent !== "subagent" ||
10066 parentThreadId === null ||
10067 childControls.thread.forkedFrom?.type !== "node"
10068 )
10069 return;
10070 const progress = delegatedTaskProgress(childControls);
10071 const parent = yield* projectionStore.getThreadRecords(parentThreadId, [
10072 "subagents",
10073 "contextTransfers",
10074 ]);
10075 for (const task of parent.subagents) {
10076 if (
10077 task.origin !== "app_owned" ||
10078 task.childThreadId !== childThreadId ||
10079 (task.id === childControls.thread.forkedFrom.nodeId &&
10080 progress.state !== "result_available") ||
10081 (task.result !== null &&
10082 parent.contextTransfers.some(
10083 (transfer) =>
10084 transfer.type === "subagent_result" && transfer.sourceThreadId === childThreadId,
10085 ))
10086 )
10087 continue;
10088 yield* finalizeAppOwnedSubagentTask(childControls, {
10089 ...options,
10090 taskId: task.id,
10091 ...(progress.resultRun === undefined ? {} : { resultRun: progress.resultRun }),
10092 });
10093 }
10094 });
10095
987610096 const finalizeDelegatedCompletionDelivery = (threadId: ThreadId, runId: RunId) =>
987710097 Effect.gen(function* () {
987810098 // Ordinary turns do not own a delegated delivery. Read just their input
......@@ -10266,6 +10486,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1026610486 });
1026710487 }
1026810488 yield* dispatchMessage(command, events, effects);
10489 yield* recordSubagentFollowup(command, events).pipe(mapDispatchError(command));
1026910490 break;
1027010491 }
1027110492 case "notification.delivery.accept":
......@@ -10627,7 +10848,26 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1062710848 });
1062810849
1062910850 const dispatchWithReceipt = (command: OrchestrationV2ServerCommand) =>
10630 threadDispatch.withLock(commandThreadId(command), dispatchWithReceiptEffect(command));
10851 Effect.gen(function* () {
10852 const dispatch = threadDispatch.withLock(
10853 commandThreadId(command),
10854 dispatchWithReceiptEffect(command),
10855 );
10856 if (
10857 command.type !== "message.dispatch" ||
10858 (command.createdBy !== "user" && command.creationSource !== "mcp")
10859 )
10860 return yield* dispatch;
10861 const parentThreadId = yield* appOwnedSubagentParentThreadId(command.threadId).pipe(
10862 Effect.mapError(
10863 (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }),
10864 ),
10865 );
10866 // Follow-up intake and Stop serialize on the parent before taking the child lock.
10867 return yield* parentThreadId === undefined
10868 ? dispatch
10869 : threadDispatch.withLock(parentThreadId, dispatch);
10870 });
1063110871
1063210872 const handleTerminalRun = (stored: OrchestrationV2StoredEvent) =>
1063310873 Effect.gen(function* () {
......@@ -10792,13 +11032,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1079211032 );
1079311033 });
1079411034
10795 const delegatedTaskResultPending = (childThreadId: ThreadId) =>
11035 const delegatedTaskResultPending = (childThreadId: ThreadId, taskId?: NodeId) =>
1079611036 Effect.gen(function* () {
1079711037 const child = yield* projectionStore.getThreadRecords(
1079811038 childThreadId,
1079911039 ["runs", "messages", "subagents", "providerThreads", "providerTurns", "attempts"],
1080011040 { messageRoles: ["user"] },
1080111041 );
11042 if (taskId !== undefined) {
11043 const runs = child.runs.filter((run) => run.delegatedTaskId === taskId);
11044 const run = runs.toSorted((a, b) => b.ordinal - a.ordinal)[0];
11045 if (run === undefined || delegatedTaskTerminalStatus(run.status) === null) return true;
11046 return yield* childAwaitsRestartContinuation(runs, run);
11047 }
1080211048 const progress = delegatedTaskProgress(child);
1080311049 // A caller's older read saw a result; newer work since then means it is not final.
1080411050 if (progress.state !== "result_available") return true;
apps/server/src/orchestration-v2/ProjectionRecovery.test.ts+83
......@@ -299,6 +299,89 @@ it.effect("recovers terminal subagent results until their cross-thread transfer
299299 }).pipe(Effect.provide(layerTest)),
300300);
301301
302it.effect.each([
303 { name: "SQLite", layer: layerTest },
304 { name: "memory", layer: ProjectionStore.layerMemory },
305])("recovers only recorded follow-up results with $name projections", ({ layer }) =>
306 Effect.gen(function* () {
307 const projections = yield* ProjectionStore.ProjectionStoreV2;
308 const now = yield* DateTime.now;
309 const parent = yield* createThread("followup-parent");
310 const child = yield* createThread("followup-child", {
311 lineage: { parentThreadId: parent, relationshipToParent: "subagent", rootThreadId: parent },
312 forkedFrom: { type: "node", nodeId: NodeId.make("node:original") },
313 });
314 const originalRunId = yield* createRun(child, "completed");
315 yield* projections.apply({
316 id: EventId.make("event:original-result"),
317 type: "context-transfer.created",
318 threadId: parent,
319 occurredAt: now,
320 payload: {
321 id: ContextTransferId.make("transfer:original-result"),
322 type: "subagent_result",
323 sourceThreadId: child,
324 targetThreadId: parent,
325 sourcePoint: { threadId: child, runId: originalRunId },
326 basePoint: null,
327 sourceProviderInstanceId: providerInstanceId,
328 targetProviderInstanceId: providerInstanceId,
329 targetRunId: null,
330 status: "pending",
331 resolution: null,
332 createdBy: "system",
333 error: null,
334 createdAt: now,
335 updatedAt: now,
336 consumedAt: null,
337 },
338 });
339 // Historical untracked work does not acquire a notification on restart.
340 yield* createRun(child, "completed", { ordinal: 2 });
341 assert.deepEqual(yield* projections.getRecoveryThreadIds("subagent-results"), []);
342 const taskId = NodeId.make("node:followup");
343 yield* createRun(child, "completed", { ordinal: 3, delegatedTaskId: taskId });
344 yield* createRun(child, "running", { ordinal: 4 });
345 const task = {
346 id: taskId,
347 threadId: parent,
348 runId: null,
349 parentNodeId: NodeId.make("node:parent"),
350 origin: "app_owned" as const,
351 createdBy: "agent" as const,
352 driver,
353 providerInstanceId,
354 providerThreadId: null,
355 childThreadId: child,
356 nativeTaskRef: null,
357 prompt: "Check the follow-up.",
358 title: null,
359 model: null,
360 status: "running" as const,
361 result: null,
362 startedAt: now,
363 completedAt: null,
364 updatedAt: now,
365 };
366 yield* projections.apply({
367 id: EventId.make("event:followup-task"),
368 type: "subagent.updated",
369 threadId: parent,
370 occurredAt: now,
371 payload: task,
372 });
373 assert.deepEqual(yield* projections.getRecoveryThreadIds("subagent-results"), [child]);
374 yield* projections.apply({
375 id: EventId.make("event:followup-settled"),
376 type: "subagent.updated",
377 threadId: parent,
378 occurredAt: now,
379 payload: { ...task, status: "completed", result: "Follow-up result.", completedAt: now },
380 });
381 assert.deepEqual(yield* projections.getRecoveryThreadIds("subagent-results"), []);
382 }).pipe(Effect.provide(layer)),
383);
384
302385it.effect("includes shared sessions and provider-owned background rosters in recovery", () =>
303386 Effect.gen(function* () {
304387 const projections = yield* ProjectionStore.ProjectionStoreV2;
apps/server/src/orchestration-v2/ProjectionStore.ts+50-23
......@@ -537,6 +537,7 @@ export const ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION = 2;
537537function needsRecovery(
538538 projection: OrchestrationV2ThreadProjection,
539539 kind: ProjectionRecoveryKind,
540 parent?: Pick<OrchestrationV2ThreadProjection, "subagents" | "contextTransfers">,
540541): boolean {
541542 if (projection.thread.deletedAt !== null) return false;
542543 switch (kind) {
......@@ -556,15 +557,28 @@ function needsRecovery(
556557 projection.thread.lineage.relationshipToParent === "subagent" &&
557558 parentThreadId !== null &&
558559 projection.thread.forkedFrom?.type === "node" &&
559 ["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes(
560 latestUnheldRun(projection.runs)?.status ?? "idle",
561 ) &&
562 !projection.contextTransfers.some(
563 (transfer) =>
564 transfer.type === "subagent_result" &&
565 transfer.sourceThreadId === projection.thread.id &&
566 transfer.targetThreadId === parentThreadId,
567 )
560 ((parent?.subagents ?? []).some(
561 (task) =>
562 task.origin === "app_owned" &&
563 task.result === null &&
564 task.childThreadId === projection.thread.id &&
565 projection.runs.some(
566 (run) =>
567 run.delegatedTaskId === task.id &&
568 ["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes(
569 run.status,
570 ),
571 ),
572 ) ||
573 (["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes(
574 latestUnheldRun(projection.runs)?.status ?? "idle",
575 ) &&
576 ![...projection.contextTransfers, ...(parent?.contextTransfers ?? [])].some(
577 (transfer) =>
578 transfer.type === "subagent_result" &&
579 transfer.sourceThreadId === projection.thread.id &&
580 transfer.targetThreadId === parentThreadId,
581 )))
568582 );
569583 }
570584 case "runtime":
......@@ -3576,23 +3590,28 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
35763590 json_extract(child.payload_json, '$.lineage.relationshipToParent') = 'subagent'
35773591 AND json_extract(child.payload_json, '$.lineage.parentThreadId') IS NOT NULL
35783592 AND json_extract(child.payload_json, '$.forkedFrom.type') = 'node'
3579 -- A held queue waits for the user, so the newest unheld run
3580 -- decides, matching latestUnheldRun.
3581 AND (
3582 SELECT status FROM orchestration_v2_projection_runs
3583 WHERE thread_id = child.thread_id
3584 AND NOT (
3585 status = 'queued'
3586 AND json_extract(payload_json, '$.queueHeld') IS 1
3587 )
3588 ORDER BY ordinal DESC LIMIT 1
3589 ) IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back')
3590 AND NOT EXISTS (
3593 AND (EXISTS (
3594 SELECT 1 FROM orchestration_v2_projection_runs AS requested
3595 JOIN orchestration_v2_projection_subagents AS task
3596 ON task.subagent_id = json_extract(requested.payload_json, '$.delegatedTaskId')
3597 WHERE requested.thread_id = child.thread_id
3598 AND requested.status IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back')
3599 AND task.child_thread_id = child.thread_id
3600 AND json_extract(task.payload_json, '$.origin') = 'app_owned'
3601 AND json_extract(task.payload_json, '$.result') IS NULL
3602 ) OR (
3603 -- The original task still waits for the latest unheld work.
3604 (SELECT status FROM orchestration_v2_projection_runs
3605 WHERE thread_id = child.thread_id
3606 AND NOT (status = 'queued' AND json_extract(payload_json, '$.queueHeld') IS 1)
3607 ORDER BY ordinal DESC LIMIT 1
3608 ) IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back')
3609 AND NOT EXISTS (
35913610 SELECT 1 FROM orchestration_v2_projection_context_transfers
35923611 WHERE source_thread_id = child.thread_id
35933612 AND target_thread_id = json_extract(child.payload_json, '$.lineage.parentThreadId')
35943613 AND type = 'subagent_result'
3595 )
3614 )))
35963615 ELSE 0 END
35973616 `;
35983617 case "runtime":
......@@ -6121,7 +6140,15 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(
61216140 Effect.gen(function* () {
61226141 const projections = (yield* Ref.get(replayState)).projections;
61236142 return [...projections.values()]
6124 .filter((projection) => needsRecovery(projection, kind))
6143 .filter((projection) =>
6144 needsRecovery(
6145 projection,
6146 kind,
6147 projection.thread.lineage.parentThreadId === null
6148 ? undefined
6149 : projections.get(projection.thread.lineage.parentThreadId),
6150 ),
6151 )
61256152 .toSorted(
61266153 (left, right) =>
61276154 DateTime.toEpochMillis(left.thread.updatedAt) -
apps/web/src/components/chat/ThreadRelationshipsControl.tsx+23-11
......@@ -170,23 +170,35 @@ function relationshipThreadTitle(input: {
170170}
171171
172172/**
173 * A delegated task settles with its first run, but the parent can keep sending
174 * the child follow-ups. While the child thread has a live run, the row's timer
175 * and hover card follow that run instead of the settled task.
173 * The row's timer and hover card follow current child work, including queued
174 * follow-ups whose provider turn has not started yet.
176175 */
177function liveSubagent<Agent extends RuntimeSubagent>(
176function currentSubagent<Agent extends RuntimeSubagent>(
178177 agent: Agent | undefined,
179178 childThread: OrchestrationV2ThreadShell | null | undefined,
180179): Agent | undefined {
181 const liveStatus = childThread?.activityRunStatus;
182 if (!agent || !liveStatus) return agent;
183 const startedAt = childThread.activityRunStartedAt;
180 const liveStatus =
181 childThread?.activityRunStatus ?? (childThread?.status === "queued" ? "queued" : null);
182 if (!agent || !childThread) return agent;
183 const newerRun =
184 childThread.latestRunRequestedAt &&
185 DateTime.toEpochMillis(childThread.latestRunRequestedAt) >
186 Date.parse(agent.completedAt ?? agent.startedAt ?? agent.updatedAt);
187 if (!liveStatus && !newerRun) return agent;
188 const status = liveStatus ?? childThread.status;
189 const startedAt = liveStatus ? childThread.activityRunStartedAt : childThread.latestRunStartedAt;
190 const completedAt = liveStatus ? null : childThread.latestRunCompletedAt;
184191 return {
185192 ...agent,
186 status: liveStatus === "running" || liveStatus === "waiting" ? liveStatus : "pending",
193 status:
194 status === "preparing" || status === "starting" || status === "queued"
195 ? "pending"
196 : status === "rolled_back"
197 ? "interrupted"
198 : status,
187199 startedAt: startedAt ? DateTime.formatIso(startedAt) : null,
188 completedAt: null,
189 // The settled task's output belongs to its first run, not this one.
200 completedAt: completedAt ? DateTime.formatIso(completedAt) : null,
201 // The task's output belongs to its recorded run, not newer child work.
190202 progress: null,
191203 result: null,
192204 error: null,
......@@ -380,7 +392,7 @@ export function ThreadRelationshipsPanel(props: {
380392 ? BotIcon
381393 : GitForkIcon;
382394 const relationship = relationshipLabel(edge, props.threadId);
383 const agent = liveSubagent(
395 const agent = currentSubagent(
384396 isSubagent && !isParent ? subagentsByThreadId.get(threadId) : undefined,
385397 node?.thread,
386398 );
packages/client-runtime/src/state/threadRelationships.test.ts+12-5
......@@ -188,11 +188,17 @@ describe("thread relationships", () => {
188188 });
189189
190190 it.each([
191 ["running", "running"],
192 [null, "completed"],
191 ["running", "running", "running"],
192 ["waiting", "completed", "waiting"],
193 [null, "queued", "queued"],
194 [null, "failed", "failed"],
195 [null, "interrupted", "interrupted"],
196 [null, "cancelled", "cancelled"],
197 [null, "completed", "completed"],
198 [null, "rolled_back", "interrupted"],
193199 ])(
194 "shows a subagent whose child activity is %s as %s after its delegated task settled",
195 (childActivity, expected) => {
200 "shows child activity %s and latest status %s as %s after its delegated task settled",
201 (childActivity, childStatus, expected) => {
196202 const parent = ThreadId.make("thread-parent");
197203 const child = ThreadId.make("thread-child");
198204 const graph = deriveThreadRelationshipGraph({
......@@ -200,8 +206,9 @@ describe("thread relationships", () => {
200206 { id: parent, status: "completed", forkedFrom: null, lineage: { parentThreadId: null } },
201207 {
202208 id: child,
203 status: childActivity ?? "completed",
209 status: childStatus,
204210 activityRunStatus: childActivity,
211 latestRunId: "run-child-followup",
205212 forkedFrom: null,
206213 lineage: { parentThreadId: parent, relationshipToParent: "subagent" },
207214 },
packages/client-runtime/src/state/threadRelationships.ts+11-5
......@@ -71,7 +71,12 @@ export function deriveThreadRelationshipGraph(input: {
7171 const addEdge = (edge: ThreadRelationshipEdge) => {
7272 ensureNode(edge.sourceThreadId);
7373 ensureNode(edge.targetThreadId);
74 edgesByKey.set(edgeKey(edge), edge);
74 edgesByKey.set(
75 edgeKey(edge),
76 edge.kind === "subagent" && edge.status === "rolled_back"
77 ? { ...edge, status: "interrupted" }
78 : edge,
79 );
7580 };
7681
7782 for (const thread of threads) {
......@@ -92,14 +97,15 @@ export function deriveThreadRelationshipGraph(input: {
9297 const ownerThreadId = input.projection.thread.id;
9398 for (const subagent of input.projection.subagents) {
9499 if (subagent.childThreadId === null) continue;
95 // The subagent record settles with the delegated task's first run, but the
96 // parent can keep sending the child follow-ups. A live run on the child
97 // thread outranks that settled status.
100 const child = threadsById.get(subagent.childThreadId);
101 // The row describes the child conversation, including queued follow-ups
102 // and their outcomes. A child without an app run still uses its task.
98103 addEdge({
99104 sourceThreadId: ownerThreadId,
100105 targetThreadId: subagent.childThreadId,
101106 kind: "subagent",
102 status: threadsById.get(subagent.childThreadId)?.activityRunStatus ?? subagent.status,
107 status:
108 child?.activityRunStatus ?? (child?.latestRunId != null ? child.status : subagent.status),
103109 });
104110 }
105111 for (const transfer of input.projection.contextTransfers) {
packages/contracts/src/orchestrationV2.ts+2
......@@ -561,6 +561,8 @@ export const OrchestrationV2Run = Schema.Struct({
561561 contextHandoffId: Schema.NullOr(ContextHandoffId),
562562 /** Links server-generated restart continuations to the interrupted run. */
563563 restartContinuationOfRunId: Schema.optional(RunId),
564 /** The delegated task this run answers, including a follow-up in an existing child thread. */
565 delegatedTaskId: Schema.optional(NodeId),
564566 /**
565567 * Set on wake runs (background notifications, delegated task results,
566568 * restart continuations): when the work they continue started. Read it
packages/contracts/src/orchestratorMcp.ts+2
......@@ -434,6 +434,8 @@ export const OrchestratorMcpThreadSendResult = Schema.Struct({
434434 threadId: ThreadId,
435435 messageId: MessageId,
436436 runId: RunId,
437 /** Fresh task for a follow-up to an app-owned child; the previous task stays terminal. */
438 taskId: Schema.optional(NodeId),
437439 status: OrchestrationV2RunStatus,
438440 delivery: Schema.Literals(["started", "queued", "steered", "restarted"]),
439441});