Skip to content

Commit ac10d11

Browse files
fix(server): held queued wakes no longer keep delegated tasks running
A child thread's queued runs that Stop, a restart, or a provider failure held for the user were still counted as active work. Delegated task finalization, startup subagent-result recovery, and task_status hasPendingChildRuns all treated the child as busy, so the parent's task row stayed "running" with no result. Held queued runs now wait for the user instead of counting as owed work, in delegatedTaskProgress, the recovery predicate (TS and SQL), and hasPendingChildRuns. handleTerminalRun also makes the queue decision before finalizing the child's task, so a provider failure holds the queue first and the failed run becomes the result; a queue failure is logged and no longer skips finalization. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
1 parent d720210 commit ac10d11

9 files changed

Lines changed: 265 additions & 18 deletions

‎apps/server/src/mcp/OrchestratorMcpService.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -372,9 +372,11 @@ export function hasPendingChildRuns(
372372
childProjection: Pick<OrchestrationV2ThreadProjection, "runs">,
373373
delegatedRun: OrchestrationV2Run | undefined,
374374
): boolean {
375+
// Held queued runs wait for the user to resume the child; task_cancel holds them.
375376
return childProjection.runs.some(
376377
(run) =>
377378
!ThreadManagementService.isTerminalRunStatus(run.status) &&
379+
!(run.status === "queued" && run.queueHeld === true) &&
378380
(delegatedRun === undefined || run.ordinal > delegatedRun.ordinal),
379381
);
380382
}

‎apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -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‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,7 @@ export const DelegateTaskTool = Tool.make("delegate_task", {
7676

7777
const 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‎

Lines changed: 192 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ import {
1414
ProviderThreadId,
1515
RunId,
1616
ThreadId,
17+
TurnItemId,
1718
} from "@t3tools/contracts";
1819
import * as DateTime from "effect/DateTime";
1920
import * 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‎

Lines changed: 20 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -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‎

Lines changed: 16 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -231,20 +231,30 @@ 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}`) },
238238
archivedAt: name === "archived" ? now : null,
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‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -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) =>
@@ -3497,9 +3497,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
34973497
json_extract(child.payload_json, '$.lineage.relationshipToParent') = 'subagent'
34983498
AND json_extract(child.payload_json, '$.lineage.parentThreadId') IS NOT NULL
34993499
AND json_extract(child.payload_json, '$.forkedFrom.type') = 'node'
3500+
-- A held queue waits for the user, so the newest unheld run
3501+
-- decides, matching latestUnheldRun.
35003502
AND (
35013503
SELECT status FROM orchestration_v2_projection_runs
35023504
WHERE thread_id = child.thread_id
3505+
AND NOT (
3506+
status = 'queued'
3507+
AND json_extract(payload_json, '$.queueHeld') IS 1
3508+
)
35033509
ORDER BY ordinal DESC LIMIT 1
35043510
) IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back')
35053511
AND NOT EXISTS (

‎apps/server/src/orchestration-v2/SubagentProjection.test.ts‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -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‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -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) =>

0 commit comments

Comments
 (0)