authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-10-07 21:49:54-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-07 21:49:54-07:00
log0bfd9dd19d0c9418a936190525ec8f4c4fdae09b
tree1a20bd49cd4e1d584ad577af5d8718d193cfceb8
parentb24f0fbba09d8623c896081b4ff70aa7190c8fda
signature Signed by PGP key B5690EEEBB952194

fix(server): held queued wakes no longer keep delegated tasks running (#17028)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

9 files changed, 265 insertions(+), 18 deletions(-)

apps/server/src/mcp/OrchestratorMcpService.ts+2
......@@ -370,9 +370,11 @@ export function hasPendingChildRuns(
370370 childProjection: Pick<OrchestrationV2ThreadProjection, "runs">,
371371 delegatedRun: OrchestrationV2Run | undefined,
372372): boolean {
373 // Held queued runs wait for the user to resume the child; task_cancel holds them.
373374 return childProjection.runs.some(
374375 (run) =>
375376 !ThreadManagementService.isTerminalRunStatus(run.status) &&
377 !(run.status === "queued" && run.queueHeld === true) &&
376378 (delegatedRun === undefined || run.ordinal > delegatedRun.ordinal),
377379 );
378380}
apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts+14
......@@ -1873,6 +1873,20 @@ describe("orchestrator MCP toolkit", () => {
18731873 legacyDelegatedRun,
18741874 ),
18751875 ).toBe(true);
1876 // A queue Stop held waits for the user; it is not pending child work.
1877 expect(
1878 hasPendingChildRuns(
1879 {
1880 ...legacyChildProjection,
1881 runs: legacyChildProjection.runs.map((run) =>
1882 run.id === activeChildFollowup.runId
1883 ? { ...run, status: "queued", queueHeld: true }
1884 : run,
1885 ),
1886 },
1887 legacyDelegatedRun,
1888 ),
1889 ).toBe(false);
18761890 const completedTaskCancelCall = yield* invoke("task_cancel", {
18771891 taskId: delegated.taskId,
18781892 reason: "Stop the child's later work too.",
apps/server/src/mcp/toolkits/orchestrator/tools.ts+1-1
......@@ -76,7 +76,7 @@ export const DelegateTaskTool = Tool.make("delegate_task", {
7676
7777const TaskStatusTool = Tool.make("task_status", {
7878 description:
79 "Needs an agent running inside a T3 thread. Read a T3-owned delegated task created by this parent thread. childRunId identifies the original delegated run. workState distinguishes working, waiting_for_children, and result_available; a completed turn with live nested work is not a completed task. summary is the final task result, including provider errors on failure, and remains stable after publication. hasPendingChildRuns reports later queued or executing turns in the backing child thread, even after the task is terminal; it does not reopen the task, and task_cancel stops those turns too. latestTerminal* provides later non-monitor turn results. Reading a terminal result acknowledges its automatic parent delivery.",
79 "Needs an agent running inside a T3 thread. Read a T3-owned delegated task created by this parent thread. childRunId identifies the original delegated run. workState distinguishes working, waiting_for_children, and result_available; a completed turn with live nested work is not a completed task. summary is the final task result, including provider errors on failure, and remains stable after publication. hasPendingChildRuns reports later queued or executing turns in the backing child thread, even after the task is terminal; it does not reopen the task, and task_cancel stops those turns too. Turns held in a stopped queue wait for the user and do not count. latestTerminal* provides later non-monitor turn results. Reading a terminal result acknowledges its automatic parent delivery.",
8080 parameters: OrchestratorMcpTaskStatusInput,
8181 success: OrchestratorMcpDelegateTaskResult,
8282 failure: OrchestratorMcpFailure,
apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts+192
......@@ -14,6 +14,7 @@ import {
1414 ProviderThreadId,
1515 RunId,
1616 ThreadId,
17 TurnItemId,
1718} from "@t3tools/contracts";
1819import * as DateTime from "effect/DateTime";
1920import * as Effect from "effect/Effect";
......@@ -1204,6 +1205,197 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => {
12041205 }),
12051206 );
12061207
1208 it.effect("settles a cancelled child whose held wakes wait behind it", () =>
1209 Effect.gen(function* () {
1210 const orchestrator = yield* Orchestrator.OrchestratorV2;
1211 const eventSink = yield* EventSink.EventSinkV2;
1212 const now = yield* DateTime.now;
1213 const threadId = ThreadId.make("thread:held-wake-parent");
1214 const projectId = ProjectId.make("project:held-wake-parent");
1215 const runId = RunId.make("run:held-wake-parent");
1216 const rootNodeId = NodeId.make("node:held-wake-parent-root");
1217 yield* seedParentWithTerminalTask({
1218 threadId,
1219 projectId,
1220 runId,
1221 rootNodeId,
1222 taskId: NodeId.make("node:held-wake-parent-settled"),
1223 deliveryState: "delivered",
1224 now,
1225 });
1226 const child = yield* seedRestartCancelledChild({
1227 parentThreadId: threadId,
1228 projectId,
1229 parentRunId: runId,
1230 rootNodeId,
1231 name: "held-wake-child",
1232 completionWake: "always",
1233 continuationPending: false,
1234 now,
1235 });
1236 // Pull request watch wakes queued while the child ran; the restart held them.
1237 const held = runEvent({
1238 threadId: child.childThreadId,
1239 runId: RunId.make("run:held-wake-child:2"),
1240 ordinal: 2,
1241 status: "queued",
1242 now,
1243 });
1244 yield* eventSink.write({
1245 commandId: CommandId.make("command:held-wake-child:held"),
1246 events: [
1247 {
1248 ...held,
1249 payload: { ...held.payload, startedAt: null, completedAt: null, queueHeld: true },
1250 },
1251 ],
1252 });
1253
1254 yield* orchestrator.recoverDelegatedTasks;
1255
1256 const recovered = yield* orchestrator.getThreadProjection(threadId);
1257 const task = recovered.subagents.find((row) => row.id === child.taskId);
1258 assert.equal(task?.status, "cancelled");
1259 assert.isNotNull(task?.result ?? null);
1260 assert.isTrue(
1261 recovered.contextTransfers.some(
1262 (transfer) =>
1263 transfer.type === "subagent_result" && transfer.sourceThreadId === child.childThreadId,
1264 ),
1265 );
1266 assert.isFalse(yield* orchestrator.delegatedTaskResultPending(child.childThreadId));
1267 // The held wake stays held for the user to resume or discard.
1268 const childProjection = yield* orchestrator.getThreadProjection(child.childThreadId);
1269 assert.deepEqual(
1270 childProjection.runs.map((run) => [run.status, run.queueHeld ?? false]),
1271 [
1272 ["cancelled", false],
1273 ["queued", true],
1274 ],
1275 );
1276 }),
1277 );
1278
1279 it.effect("settles a child whose provider failure holds its queued wakes", () =>
1280 Effect.gen(function* () {
1281 const orchestrator = yield* Orchestrator.OrchestratorV2;
1282 const eventSink = yield* EventSink.EventSinkV2;
1283 const now = yield* DateTime.now;
1284 const threadId = ThreadId.make("thread:failed-hold-parent");
1285 const projectId = ProjectId.make("project:failed-hold-parent");
1286 const runId = RunId.make("run:failed-hold-parent");
1287 const rootNodeId = NodeId.make("node:failed-hold-parent-root");
1288 yield* seedParentWithTerminalTask({
1289 threadId,
1290 projectId,
1291 runId,
1292 rootNodeId,
1293 taskId: NodeId.make("node:failed-hold-parent-settled"),
1294 deliveryState: "delivered",
1295 now,
1296 });
1297 const child = yield* seedRestartCancelledChild({
1298 parentThreadId: threadId,
1299 projectId,
1300 parentRunId: runId,
1301 rootNodeId,
1302 name: "failed-hold-child",
1303 completionWake: "always",
1304 continuationPending: false,
1305 now,
1306 });
1307 // The child is resumed, and a wake queues behind its running turn.
1308 const activeRun = runEvent({
1309 threadId: child.childThreadId,
1310 runId: RunId.make("run:failed-hold-child:2"),
1311 ordinal: 2,
1312 status: "running",
1313 now,
1314 });
1315 const queued = runEvent({
1316 threadId: child.childThreadId,
1317 runId: RunId.make("run:failed-hold-child:3"),
1318 ordinal: 3,
1319 status: "queued",
1320 now,
1321 });
1322 yield* eventSink.write({
1323 commandId: CommandId.make("command:failed-hold-child:resumed"),
1324 events: [
1325 activeRun,
1326 { ...queued, payload: { ...queued.payload, startedAt: null, completedAt: null } },
1327 ],
1328 });
1329 const afterSequence = yield* eventSink.latestSequence();
1330 const errorItemId = TurnItemId.make("turn-item:failed-hold-child:error");
1331 yield* eventSink.write({
1332 commandId: CommandId.make("command:failed-hold-child:failed"),
1333 events: [
1334 {
1335 id: EventId.make("event:failed-hold-child:error"),
1336 type: "turn-item.updated",
1337 threadId: child.childThreadId,
1338 runId: activeRun.runId,
1339 providerInstanceId: modelSelection.instanceId,
1340 occurredAt: now,
1341 payload: {
1342 id: errorItemId,
1343 type: "error",
1344 threadId: child.childThreadId,
1345 runId: activeRun.runId,
1346 nodeId: null,
1347 providerThreadId: null,
1348 providerTurnId: null,
1349 nativeItemRef: null,
1350 parentItemId: null,
1351 ordinal: 1,
1352 status: "failed",
1353 title: "Provider failure",
1354 startedAt: now,
1355 completedAt: now,
1356 updatedAt: now,
1357 failure: {
1358 class: "provider_error",
1359 message: "Provider failed.",
1360 code: "provider_failed",
1361 retryable: null,
1362 },
1363 },
1364 },
1365 {
1366 ...activeRun,
1367 id: EventId.make("event:failed-hold-child:failed"),
1368 payload: { ...activeRun.payload, status: "failed", completedAt: now },
1369 },
1370 ],
1371 });
1372
1373 // The failure holds the queue, which leaves the failure as the task's result.
1374 const settled = yield* eventSink
1375 .stream({ afterSequence, eventType: "subagent.updated" })
1376 .pipe(
1377 Stream.filter(
1378 (stored) =>
1379 stored.event.type === "subagent.updated" && stored.event.payload.id === child.taskId,
1380 ),
1381 Stream.take(1),
1382 Stream.runHead,
1383 );
1384 assert.isTrue(settled._tag === "Some");
1385 const recovered = yield* orchestrator.getThreadProjection(threadId);
1386 assert.equal(recovered.subagents.find((row) => row.id === child.taskId)?.status, "failed");
1387 const childProjection = yield* orchestrator.getThreadProjection(child.childThreadId);
1388 assert.deepEqual(
1389 childProjection.runs.map((run) => [run.status, run.queueHeld ?? false]),
1390 [
1391 ["cancelled", false],
1392 ["failed", false],
1393 ["queued", true],
1394 ],
1395 );
1396 }),
1397 );
1398
12071399 it.effect("wakes the parent for a settled_only task once its spawning turn ended", () =>
12081400 Effect.gen(function* () {
12091401 const orchestrator = yield* Orchestrator.OrchestratorV2;
apps/server/src/orchestration-v2/Orchestrator.ts+20-12
......@@ -10613,25 +10613,33 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1061310613 // while holding the parent lock, so nesting the parent lock inside the
1061410614 // child lock here would invert that order, and the keyed executor's
1061510615 // semaphores are neither reentrant nor deadlock-aware.
10616 const parentThreadId = yield* appOwnedSubagentParentThreadId(threadId);
10617 if (parentThreadId !== undefined) {
10618 yield* threadDispatch.withLock(parentThreadId, finalizeAppOwnedSubagent(threadId));
10619 }
1062010616 if (stored.event.type === "run.updated") {
1062110617 yield* threadDispatch.withLock(
1062210618 threadId,
1062310619 finalizeDelegatedCompletionDelivery(threadId, stored.event.payload.id),
1062410620 );
1062510621 }
10626 yield* threadDispatch.withLock(
10627 threadId,
10628 startNextQueuedRun(
10622 yield* threadDispatch
10623 .withLock(
1062910624 threadId,
10630 stored.event.type === "run.updated" && stored.event.payload.status === "failed"
10631 ? { failedRunId: stored.event.payload.id }
10632 : undefined,
10633 ),
10634 );
10625 startNextQueuedRun(
10626 threadId,
10627 stored.event.type === "run.updated" && stored.event.payload.status === "failed"
10628 ? { failedRunId: stored.event.payload.id }
10629 : undefined,
10630 ),
10631 )
10632 .pipe(
10633 Effect.catchCause((cause) =>
10634 Effect.logWarning("Failed to start the next queued V2 run", { threadId, cause }),
10635 ),
10636 );
10637 // After the queue decision: a provider failure holds the child's queued
10638 // wakes, and only then is the failed run the task's result.
10639 const parentThreadId = yield* appOwnedSubagentParentThreadId(threadId);
10640 if (parentThreadId !== undefined) {
10641 yield* threadDispatch.withLock(parentThreadId, finalizeAppOwnedSubagent(threadId));
10642 }
1063510643 }).pipe(
1063610644 Effect.catchCause((cause) =>
1063710645 Effect.logWarning("Failed to react to terminal V2 run", {
apps/server/src/orchestration-v2/ProjectionRecovery.test.ts+16-3
......@@ -231,7 +231,7 @@ it.effect("recovers terminal subagent results until their cross-thread transfer
231231 const now = yield* DateTime.now;
232232 const parent = yield* createThread("subagent-parent");
233233 const children: Array<ThreadId> = [];
234 for (const name of ["terminal", "archived", "deleted", "running"]) {
234 for (const name of ["terminal", "archived", "deleted", "running", "held", "queued"]) {
235235 const child = yield* createThread(`subagent-${name}`, {
236236 lineage: { parentThreadId: parent, relationshipToParent: "subagent", rootThreadId: parent },
237237 forkedFrom: { type: "node", nodeId: NodeId.make(`node:${name}`) },
......@@ -239,12 +239,22 @@ it.effect("recovers terminal subagent results until their cross-thread transfer
239239 deletedAt: name === "deleted" ? now : null,
240240 });
241241 yield* createRun(child, name === "running" ? "running" : "completed");
242 // A wake queued behind the result: held by Stop or a restart, or still deliverable.
243 if (name === "held" || name === "queued") {
244 yield* createRun(child, "queued", {
245 ordinal: 2,
246 startedAt: null,
247 ...(name === "held" ? { queueHeld: true } : {}),
248 });
249 }
242250 children.push(child);
243251 }
252 const terminalChildren = new Set([children[0]!, children[1]!, children[4]!]);
244253 assert.deepEqual(
245254 new Set(yield* projections.getRecoveryThreadIds("subagent-results")),
246 new Set(children.slice(0, 2)),
255 terminalChildren,
247256 );
257
248258 const completed = children[0]!;
249259 const transferId = ContextTransferId.make("transfer:recovery:subagent-result");
250260 yield* projections.apply({
......@@ -273,7 +283,10 @@ it.effect("recovers terminal subagent results until their cross-thread transfer
273283 });
274284 const archivedChild = children[1];
275285 assert.isDefined(archivedChild);
276 assert.deepEqual(yield* projections.getRecoveryThreadIds("subagent-results"), [archivedChild]);
286 assert.deepEqual(
287 new Set(yield* projections.getRecoveryThreadIds("subagent-results")),
288 new Set([archivedChild, children[4]!]),
289 );
277290 assert.deepEqual(yield* projections.getUnreadableThreadIds(), []);
278291 yield* sql`
279292 UPDATE orchestration_v2_projection_context_transfers SET payload_json = '{}'
apps/server/src/orchestration-v2/ProjectionStore.ts+7-1
......@@ -514,7 +514,7 @@ function needsRecovery(
514514 parentThreadId !== null &&
515515 projection.thread.forkedFrom?.type === "node" &&
516516 ["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes(
517 projection.runs.at(-1)?.status ?? "idle",
517 latestUnheldRun(projection.runs)?.status ?? "idle",
518518 ) &&
519519 !projection.contextTransfers.some(
520520 (transfer) =>
......@@ -3505,9 +3505,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
35053505 json_extract(child.payload_json, '$.lineage.relationshipToParent') = 'subagent'
35063506 AND json_extract(child.payload_json, '$.lineage.parentThreadId') IS NOT NULL
35073507 AND json_extract(child.payload_json, '$.forkedFrom.type') = 'node'
3508 -- A held queue waits for the user, so the newest unheld run
3509 -- decides, matching latestUnheldRun.
35083510 AND (
35093511 SELECT status FROM orchestration_v2_projection_runs
35103512 WHERE thread_id = child.thread_id
3513 AND NOT (
3514 status = 'queued'
3515 AND json_extract(payload_json, '$.queueHeld') IS 1
3516 )
35113517 ORDER BY ordinal DESC LIMIT 1
35123518 ) IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back')
35133519 AND NOT EXISTS (
apps/server/src/orchestration-v2/SubagentProjection.test.ts+8
......@@ -237,6 +237,14 @@ it("waits for nested work and retains the report across monitor acknowledgements
237237 startedAt: null,
238238 };
239239 assert.equal(delegatedTaskProgress({ ...projection, runs: [run, pending] }).state, "working");
240 // Stop or a restart holds queued wakes for the user; they are not work the task owes.
241 const cancelled = { ...run, status: "cancelled" as const };
242 const held = delegatedTaskProgress({
243 ...projection,
244 runs: [cancelled, { ...pending, queueHeld: true }],
245 });
246 assert.equal(held.state, "result_available");
247 assert.equal(held.resultRun?.id, cancelled.id);
240248 const report = { ...pending, status: "completed" as const, startedAt: parentCreatedAt };
241249 const monitor = { ...report, id: RunId.make("monitor"), ordinal: 3 };
242250 const artifacts = makeSubagentConversationArtifacts({
apps/server/src/orchestration-v2/SubagentProjection.ts+5-1
......@@ -229,7 +229,11 @@ export function delegatedTaskProgress(projection: {
229229 const workRuns = projection.runs.filter(
230230 (run) => !monitorRuns.has(run.id) && run.status !== "rolled_back",
231231 );
232 const active = workRuns.some((run) => !terminal(run.status));
232 // A held queue waits for the user to resume it (after Stop, a restart, or a
233 // provider failure), so its runs are not work the task still owes.
234 const active = workRuns.some(
235 (run) => !terminal(run.status) && !(run.status === "queued" && run.queueHeld === true),
236 );
233237 const children =
234238 projection.subagents.some(
235239 (task) =>