From 7d092fecb6b0a12ce0bdbd1f4eca665700f9b119 Mon Sep 17 00:00:00 2001 From: Kayzel Date: Thu, 5 Nov 2026 09:29:48 +0100 Subject: [PATCH 01/13] test(server): parent shell stays held until a delegated wake is dispatched --- .../DelegatedCompletionDelivery.test.ts | 180 ++++++++++++++++++ 1 file changed, 180 insertions(+) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index db3a2759e3ac5..9567dd1986209 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -1628,6 +1628,186 @@ it.layer(layerContinuationWorkerTest)("interrupted completion wakes", (it) => { it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( "subagent follow-ups", (it) => { + it.effect("keeps an idle parent waiting until a finished task's wake is dispatched", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:wake-handoff-parent"); + const projectId = ProjectId.make("project:wake-handoff"); + const runId = RunId.make("run:wake-handoff-parent"); + const rootNodeId = NodeId.make("node:wake-handoff-parent-root"); + const taskId = NodeId.make("node:wake-handoff-task"); + const childThreadId = ThreadId.make("thread:wake-handoff-child"); + yield* seedParentWithTerminalTask({ + threadId, + projectId, + runId, + rootNodeId, + taskId: NodeId.make("node:wake-handoff-seed"), + deliveryState: "acknowledged", + parentRunStatus: "completed", + now, + }); + const childRun = runEvent({ + threadId: childThreadId, + runId: RunId.make("run:wake-handoff-child"), + ordinal: 1, + status: "running", + now, + }); + yield* sink.write({ + events: [ + { + id: EventId.make("event:wake-handoff-child-thread"), + type: "thread.created", + threadId: childThreadId, + occurredAt: now, + payload: { + createdBy: "agent", + creationSource: "server", + id: childThreadId, + projectId, + title: "Review the diff", + providerInstanceId: modelSelection.instanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { + parentThreadId: threadId, + relationshipToParent: "subagent", + rootThreadId: threadId, + }, + forkedFrom: { type: "node", nodeId: taskId }, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }, + { + id: EventId.make("event:wake-handoff-task"), + type: "subagent.updated", + threadId, + runId, + nodeId: taskId, + driver, + providerInstanceId: modelSelection.instanceId, + occurredAt: now, + payload: { + id: taskId, + threadId, + runId, + parentNodeId: rootNodeId, + origin: "app_owned", + createdBy: "agent", + driver, + providerInstanceId: modelSelection.instanceId, + providerThreadId: null, + childThreadId, + nativeTaskRef: null, + prompt: "Review the diff.", + title: null, + model: null, + completionWake: "always", + status: "running", + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }, + }, + { + id: EventId.make("event:wake-handoff-turn-item"), + type: "turn-item.updated", + threadId, + runId, + nodeId: taskId, + driver, + occurredAt: now, + payload: { + id: TurnItemId.make("turn-item:wake-handoff-task"), + threadId, + runId, + nodeId: taskId, + providerThreadId: parentProviderThreadId(threadId), + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "running", + title: "Review the diff", + startedAt: now, + completedAt: null, + updatedAt: now, + type: "subagent", + subagentId: taskId, + origin: "app_owned", + driver, + providerInstanceId: modelSelection.instanceId, + childThreadId, + prompt: "Review the diff.", + result: null, + }, + }, + childRun, + ], + }); + const roster = Effect.map( + orchestrator.getThreadShell(threadId), + (shell) => + shell?.pendingBackgroundTasks?.map((task) => + task.kind === "subagent" ? task.childThreadId : task.kind, + ) ?? [], + ); + assert.deepEqual(yield* roster, [childThreadId]); + + yield* sink.write({ + events: [ + { + ...childRun, + id: EventId.make("event:wake-handoff-child-completed"), + payload: { ...childRun.payload, status: "completed", completedAt: now }, + }, + ], + }); + let wake = yield* requests.take; + while (wake.threadId !== threadId) wake = yield* requests.take; + const delivery = wake.delegatedCompletion!; + // The result is committed and the wake is queued for the continuation worker, but + // no wake run exists yet. Clients read a completed run with an empty roster as done. + const handoff = yield* orchestrator.getThreadShell(threadId); + assert.equal(handoff?.status, "completed"); + assert.deepEqual(yield* roster, [childThreadId]); + + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("command:wake-handoff-dispatch"), + threadId, + messageId: delivery.messageId, + text: "Delegated task finished.", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "server", + delegatedCompletion: { + parentRunId: delivery.parentRunId, + generation: delivery.generation, + taskIds: [taskId], + }, + }); + const woken = yield* orchestrator.getThreadShell(threadId); + assert.notEqual(woken?.latestRunId, runId); + assert.deepEqual(woken?.pendingBackgroundTasks, []); + }), + ); it.effect("recovers a reserved delivery without loading conversation history", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; From 8a7fc6c140373e13d19e9fca1c7a0647d2dc4768 Mon Sep 17 00:00:00 2001 From: Kayzel Date: Thu, 5 Nov 2026 09:36:56 +0100 Subject: [PATCH 02/13] fix(server): a parent no longer reads as done while its delegate's wake is queued Closes #31346 --- .../src/orchestration-v2/ProjectionStore.ts | 23 +++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index dea68ee7723d1..8877cb0361c1b 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6479,6 +6479,29 @@ export const layer: Layer.Layer = AND i.status NOT IN ('completed', 'interrupted', 'failed', 'cancelled') AND (i.run_id IS NULL OR r.status <> 'rolled_back') ${threadIds === undefined ? sql`` : sql`AND i.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)}))`} + -- A finished delegate's result is committed before the continuation worker dispatches + -- the parent's wake, so its task holds the parent until the wake message exists. + UNION ALL + SELECT i.thread_id, json_set(i.payload_json, '$.status', 'waiting') AS payload_json, + p.provider_instance_id AS owning_provider_instance_id + FROM orchestration_v2_projection_subagents task + INNER JOIN orchestration_v2_projection_runs r ON r.run_id = task.run_id + INNER JOIN orchestration_v2_projection_turn_items i + ON i.node_id = task.subagent_id AND i.thread_id = task.thread_id AND i.type = 'subagent' + LEFT JOIN orchestration_v2_projection_provider_threads p + ON p.provider_thread_id = i.provider_thread_id + WHERE task.completion_delivery_state = 'claimed' + AND task.origin = 'app_owned' + ${threadIds === undefined ? sql`` : sql`AND task.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)}))`} + AND json_extract(r.payload_json, '$.delegatedCompletion.disposition') = 'open' + AND EXISTS ( + SELECT 1 FROM json_each(r.payload_json, '$.delegatedCompletion.delivery.taskIds') owed + WHERE owed.value = task.subagent_id + ) + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_messages wake + WHERE wake.message_id = json_extract(r.payload_json, '$.delegatedCompletion.delivery.messageId') + ) ${ activeRunIds.length === 0 ? sql`` From fcb910b35f5a855a9cc3e89daa69903e4e5d75fa Mon Sep 17 00:00:00 2001 From: Kayzel Date: Thu, 5 Nov 2026 10:01:06 +0100 Subject: [PATCH 03/13] fix(server): the held wake task keeps the parent waiting without reviving its caller access --- apps/server/src/orchestration-v2/ProjectionStore.ts | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 8877cb0361c1b..bf6da76a77da6 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6479,17 +6479,16 @@ export const layer: Layer.Layer = AND i.status NOT IN ('completed', 'interrupted', 'failed', 'cancelled') AND (i.run_id IS NULL OR r.status <> 'rolled_back') ${threadIds === undefined ? sql`` : sql`AND i.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)}))`} - -- A finished delegate's result is committed before the continuation worker dispatches - -- the parent's wake, so its task holds the parent until the wake message exists. + -- A finished delegate's result commits before the continuation worker dispatches the + -- parent's wake run, so its task holds the parent until a newer run exists. Without an + -- owning instance it holds completion only, not the ended provider's caller access. UNION ALL SELECT i.thread_id, json_set(i.payload_json, '$.status', 'waiting') AS payload_json, - p.provider_instance_id AS owning_provider_instance_id + NULL AS owning_provider_instance_id FROM orchestration_v2_projection_subagents task INNER JOIN orchestration_v2_projection_runs r ON r.run_id = task.run_id INNER JOIN orchestration_v2_projection_turn_items i ON i.node_id = task.subagent_id AND i.thread_id = task.thread_id AND i.type = 'subagent' - LEFT JOIN orchestration_v2_projection_provider_threads p - ON p.provider_thread_id = i.provider_thread_id WHERE task.completion_delivery_state = 'claimed' AND task.origin = 'app_owned' ${threadIds === undefined ? sql`` : sql`AND task.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)}))`} @@ -6499,8 +6498,8 @@ export const layer: Layer.Layer = WHERE owed.value = task.subagent_id ) AND NOT EXISTS ( - SELECT 1 FROM orchestration_v2_projection_messages wake - WHERE wake.message_id = json_extract(r.payload_json, '$.delegatedCompletion.delivery.messageId') + SELECT 1 FROM orchestration_v2_projection_runs newer + WHERE newer.thread_id = r.thread_id AND newer.ordinal > r.ordinal ) ${ activeRunIds.length === 0 From ed17905ba52ee9dcd10131a56989d5670637605d Mon Sep 17 00:00:00 2001 From: Kayzel Date: Thu, 5 Nov 2026 10:20:55 +0100 Subject: [PATCH 04/13] fix(server): later delegates in a cohort also hold the parent until their wake is dispatched --- .../DelegatedCompletionDelivery.test.ts | 311 ++++++++++-------- .../src/orchestration-v2/ProjectionStore.ts | 6 +- 2 files changed, 178 insertions(+), 139 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 9567dd1986209..dfa8f1da569c7 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -1628,7 +1628,7 @@ it.layer(layerContinuationWorkerTest)("interrupted completion wakes", (it) => { it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( "subagent follow-ups", (it) => { - it.effect("keeps an idle parent waiting until a finished task's wake is dispatched", () => + it.effect("keeps an idle parent waiting until each finished task's wake is dispatched", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const sink = yield* EventSink.EventSinkV2; @@ -1638,8 +1638,6 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( const projectId = ProjectId.make("project:wake-handoff"); const runId = RunId.make("run:wake-handoff-parent"); const rootNodeId = NodeId.make("node:wake-handoff-parent-root"); - const taskId = NodeId.make("node:wake-handoff-task"); - const childThreadId = ThreadId.make("thread:wake-handoff-child"); yield* seedParentWithTerminalTask({ threadId, projectId, @@ -1650,162 +1648,207 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( parentRunStatus: "completed", now, }); - const childRun = runEvent({ - threadId: childThreadId, - runId: RunId.make("run:wake-handoff-child"), - ordinal: 1, - status: "running", - now, - }); - yield* sink.write({ - events: [ - { - id: EventId.make("event:wake-handoff-child-thread"), - type: "thread.created", - threadId: childThreadId, - occurredAt: now, - payload: { - createdBy: "agent", - creationSource: "server", - id: childThreadId, - projectId, - title: "Review the diff", - providerInstanceId: modelSelection.instanceId, - modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: null, - activeProviderThreadId: null, - lineage: { - parentThreadId: threadId, - relationshipToParent: "subagent", - rootThreadId: threadId, + const startChild = Effect.fn(function* (name: string) { + const taskId = NodeId.make(`node:wake-handoff-${name}`); + const childThreadId = ThreadId.make(`thread:wake-handoff-${name}`); + const childRun = runEvent({ + threadId: childThreadId, + runId: RunId.make(`run:wake-handoff-${name}`), + ordinal: 1, + status: "running", + now, + }); + yield* sink.write({ + events: [ + { + id: EventId.make(`event:wake-handoff-${name}-thread`), + type: "thread.created", + threadId: childThreadId, + occurredAt: now, + payload: { + createdBy: "agent", + creationSource: "server", + id: childThreadId, + projectId, + title: "Review the diff", + providerInstanceId: modelSelection.instanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { + parentThreadId: threadId, + relationshipToParent: "subagent", + rootThreadId: threadId, + }, + forkedFrom: { type: "node", nodeId: taskId }, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, }, - forkedFrom: { type: "node", nodeId: taskId }, - createdAt: now, - updatedAt: now, - archivedAt: null, - settledOverride: null, - settledAt: null, - lastVisitedAt: null, - deletedAt: null, }, - }, - { - id: EventId.make("event:wake-handoff-task"), - type: "subagent.updated", - threadId, - runId, - nodeId: taskId, - driver, - providerInstanceId: modelSelection.instanceId, - occurredAt: now, - payload: { - id: taskId, + { + id: EventId.make(`event:wake-handoff-${name}-task`), + type: "subagent.updated", threadId, runId, - parentNodeId: rootNodeId, - origin: "app_owned", - createdBy: "agent", + nodeId: taskId, driver, providerInstanceId: modelSelection.instanceId, - providerThreadId: null, - childThreadId, - nativeTaskRef: null, - prompt: "Review the diff.", - title: null, - model: null, - completionWake: "always", - status: "running", - result: null, - startedAt: now, - completedAt: null, - updatedAt: now, + occurredAt: now, + payload: { + id: taskId, + threadId, + runId, + parentNodeId: rootNodeId, + origin: "app_owned", + createdBy: "agent", + driver, + providerInstanceId: modelSelection.instanceId, + providerThreadId: null, + childThreadId, + nativeTaskRef: null, + prompt: "Review the diff.", + title: null, + model: null, + completionWake: "always", + status: "running", + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }, }, - }, - { - id: EventId.make("event:wake-handoff-turn-item"), - type: "turn-item.updated", - threadId, - runId, - nodeId: taskId, - driver, - occurredAt: now, - payload: { - id: TurnItemId.make("turn-item:wake-handoff-task"), + { + id: EventId.make(`event:wake-handoff-${name}-turn-item`), + type: "turn-item.updated", threadId, runId, nodeId: taskId, - providerThreadId: parentProviderThreadId(threadId), - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: 1, - status: "running", - title: "Review the diff", - startedAt: now, - completedAt: null, - updatedAt: now, - type: "subagent", - subagentId: taskId, - origin: "app_owned", driver, - providerInstanceId: modelSelection.instanceId, - childThreadId, - prompt: "Review the diff.", - result: null, + occurredAt: now, + payload: { + id: TurnItemId.make(`turn-item:wake-handoff-${name}`), + threadId, + runId, + nodeId: taskId, + providerThreadId: parentProviderThreadId(threadId), + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "running", + title: "Review the diff", + startedAt: now, + completedAt: null, + updatedAt: now, + type: "subagent", + subagentId: taskId, + origin: "app_owned", + driver, + providerInstanceId: modelSelection.instanceId, + childThreadId, + prompt: "Review the diff.", + result: null, + }, }, - }, - childRun, - ], + childRun, + ], + }); + return { taskId, childThreadId, childRun }; }); - const roster = Effect.map( - orchestrator.getThreadShell(threadId), - (shell) => + const roster = Effect.map(orchestrator.getThreadShell(threadId), (shell) => + ( shell?.pendingBackgroundTasks?.map((task) => task.kind === "subagent" ? task.childThreadId : task.kind, - ) ?? [], + ) ?? [] + ).toSorted(), ); - assert.deepEqual(yield* roster, [childThreadId]); - - yield* sink.write({ - events: [ - { - ...childRun, - id: EventId.make("event:wake-handoff-child-completed"), - payload: { ...childRun.payload, status: "completed", completedAt: now }, - }, - ], + const finishChild = Effect.fn(function* ( + child: Effect.Success>, + ) { + yield* sink.write({ + events: [ + { + ...child.childRun, + id: EventId.make(`event:${child.childThreadId}-completed`), + payload: { ...child.childRun.payload, status: "completed", completedAt: now }, + }, + ], + }); + let wake = yield* requests.take; + while (wake.threadId !== threadId) wake = yield* requests.take; + return wake.delegatedCompletion!; }); - let wake = yield* requests.take; - while (wake.threadId !== threadId) wake = yield* requests.take; - const delivery = wake.delegatedCompletion!; + const dispatchWake = ( + delivery: Effect.Success>, + taskId: NodeId, + ) => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`command:${delivery.messageId}`), + threadId, + messageId: delivery.messageId, + text: "Delegated task finished.", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "server", + delegatedCompletion: { + parentRunId: delivery.parentRunId, + generation: delivery.generation, + taskIds: [taskId], + }, + }); + + const first = yield* startChild("first"); + const second = yield* startChild("second"); + assert.deepEqual(yield* roster, [first.childThreadId, second.childThreadId]); + + const firstDelivery = yield* finishChild(first); // The result is committed and the wake is queued for the continuation worker, but // no wake run exists yet. Clients read a completed run with an empty roster as done. const handoff = yield* orchestrator.getThreadShell(threadId); assert.equal(handoff?.status, "completed"); - assert.deepEqual(yield* roster, [childThreadId]); + assert.deepEqual(yield* roster, [first.childThreadId, second.childThreadId]); - yield* orchestrator.dispatch({ - type: "message.dispatch", - commandId: CommandId.make("command:wake-handoff-dispatch"), - threadId, - messageId: delivery.messageId, - text: "Delegated task finished.", - attachments: [], - dispatchMode: { type: "queue_after_active" }, - createdBy: "agent", - creationSource: "server", - delegatedCompletion: { - parentRunId: delivery.parentRunId, - generation: delivery.generation, - taskIds: [taskId], - }, - }); + yield* dispatchWake(firstDelivery, first.taskId); const woken = yield* orchestrator.getThreadShell(threadId); assert.notEqual(woken?.latestRunId, runId); assert.deepEqual(woken?.pendingBackgroundTasks, []); + + const wakeRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.userMessageId === firstDelivery.messageId, + )!; + yield* sink.write({ + events: [ + { + id: EventId.make("event:wake-handoff-first-wake-completed"), + type: "run.updated", + threadId, + runId: wakeRun.id, + providerInstanceId: modelSelection.instanceId, + occurredAt: now, + payload: { ...wakeRun, status: "completed", completedAt: now }, + }, + ], + }); + assert.deepEqual(yield* roster, [second.childThreadId]); + + // The spawning run already has a newer run, so the hold must key on the open delivery. + const secondDelivery = yield* finishChild(second); + assert.deepEqual(yield* roster, [second.childThreadId]); + yield* dispatchWake(secondDelivery, second.taskId); + assert.deepEqual( + (yield* orchestrator.getThreadShell(threadId))?.pendingBackgroundTasks, + [], + ); }), ); it.effect("recovers a reserved delivery without loading conversation history", () => diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index bf6da76a77da6..03fd4d6774d9f 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6480,7 +6480,7 @@ export const layer: Layer.Layer = AND (i.run_id IS NULL OR r.status <> 'rolled_back') ${threadIds === undefined ? sql`` : sql`AND i.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)}))`} -- A finished delegate's result commits before the continuation worker dispatches the - -- parent's wake run, so its task holds the parent until a newer run exists. Without an + -- parent's wake run, so its task holds the parent until that wake run settles it. Without an -- owning instance it holds completion only, not the ended provider's caller access. UNION ALL SELECT i.thread_id, json_set(i.payload_json, '$.status', 'waiting') AS payload_json, @@ -6497,10 +6497,6 @@ export const layer: Layer.Layer = SELECT 1 FROM json_each(r.payload_json, '$.delegatedCompletion.delivery.taskIds') owed WHERE owed.value = task.subagent_id ) - AND NOT EXISTS ( - SELECT 1 FROM orchestration_v2_projection_runs newer - WHERE newer.thread_id = r.thread_id AND newer.ordinal > r.ordinal - ) ${ activeRunIds.length === 0 ? sql`` From 53dd334030b683fd09249ef68883e87cd0519773 Mon Sep 17 00:00:00 2001 From: Kayzel Date: Thu, 5 Nov 2026 10:39:30 +0100 Subject: [PATCH 05/13] test(server): a finished task awaiting the next delivery keeps the parent waiting --- .../DelegatedCompletionDelivery.test.ts | 68 +++++++++++++++++++ 1 file changed, 68 insertions(+) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index dfa8f1da569c7..e65a2c528db8e 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -322,6 +322,74 @@ const seedParentWithTerminalTask = (input: { }); it.layer(layerTest)("delegated completion delivery repairs", (it) => { + it.effect("keeps an idle parent waiting while a finished task waits for the next delivery", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:wake-handoff-successor"); + const runId = RunId.make("run:wake-handoff-successor"); + const deliveredTaskId = NodeId.make("node:wake-handoff-successor-delivered"); + const taskId = NodeId.make("node:wake-handoff-successor-pending"); + // The delivery run has ended, but the successor delivery is not claimed yet. + yield* seedParentWithTerminalTask({ + threadId, + projectId: ProjectId.make("project:wake-handoff-successor"), + runId, + rootNodeId: NodeId.make("node:wake-handoff-successor-root"), + taskId, + deliveryState: "pending", + deliveryTaskIds: [deliveredTaskId], + siblingTasks: [{ id: deliveredTaskId, deliveryState: "delivered" }], + parentRunStatus: "completed", + completionWake: "always", + now, + }); + yield* sink.write({ + events: [ + { + id: EventId.make("event:wake-handoff-successor-turn-item"), + type: "turn-item.updated", + threadId, + runId, + nodeId: taskId, + driver, + occurredAt: now, + payload: { + id: TurnItemId.make("turn-item:wake-handoff-successor"), + threadId, + runId, + nodeId: taskId, + providerThreadId: parentProviderThreadId(threadId), + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: "Review the diff", + startedAt: now, + completedAt: now, + updatedAt: now, + type: "subagent", + subagentId: taskId, + origin: "app_owned", + driver, + providerInstanceId: modelSelection.instanceId, + childThreadId: null, + prompt: "Review the diff.", + result: "child finished", + }, + }, + ], + }); + const shell = yield* orchestrator.getThreadShell(threadId); + assert.equal(shell?.status, "completed"); + assert.deepEqual( + shell?.pendingBackgroundTasks?.map((task) => task.kind), + ["subagent"], + ); + }), + ); it.effect( "interrupting a completion wake disposes both cohorts without reviving them on late results", () => From 8e2e5357a72ab38798ce64f23e91a993deea3b81 Mon Sep 17 00:00:00 2001 From: Kayzel Date: Thu, 5 Nov 2026 10:49:06 +0100 Subject: [PATCH 06/13] fix(server): a finished task awaiting the next delivery keeps the parent waiting --- .../src/orchestration-v2/ProjectionStore.ts | 21 ++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 03fd4d6774d9f..734d284ae3b42 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6480,7 +6480,8 @@ export const layer: Layer.Layer = AND (i.run_id IS NULL OR r.status <> 'rolled_back') ${threadIds === undefined ? sql`` : sql`AND i.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)}))`} -- A finished delegate's result commits before the continuation worker dispatches the - -- parent's wake run, so its task holds the parent until that wake run settles it. Without an + -- parent's wake run, so its task holds the parent until that wake run settles it. A task + -- left pending behind an open delivery holds until the next delivery claims it. Without an -- owning instance it holds completion only, not the ended provider's caller access. UNION ALL SELECT i.thread_id, json_set(i.payload_json, '$.status', 'waiting') AS payload_json, @@ -6489,13 +6490,23 @@ export const layer: Layer.Layer = INNER JOIN orchestration_v2_projection_runs r ON r.run_id = task.run_id INNER JOIN orchestration_v2_projection_turn_items i ON i.node_id = task.subagent_id AND i.thread_id = task.thread_id AND i.type = 'subagent' - WHERE task.completion_delivery_state = 'claimed' + WHERE task.completion_delivery_state IN ('claimed', 'pending') AND task.origin = 'app_owned' ${threadIds === undefined ? sql`` : sql`AND task.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)}))`} AND json_extract(r.payload_json, '$.delegatedCompletion.disposition') = 'open' - AND EXISTS ( - SELECT 1 FROM json_each(r.payload_json, '$.delegatedCompletion.delivery.taskIds') owed - WHERE owed.value = task.subagent_id + AND ( + ( + task.completion_delivery_state = 'claimed' + AND EXISTS ( + SELECT 1 FROM json_each(r.payload_json, '$.delegatedCompletion.delivery.taskIds') owed + WHERE owed.value = task.subagent_id + ) + ) + OR ( + task.completion_delivery_state = 'pending' + AND i.status IN ('completed', 'interrupted', 'failed', 'cancelled') + AND json_extract(r.payload_json, '$.delegatedCompletion.delivery.messageId') IS NOT NULL + ) ) ${ activeRunIds.length === 0 From efa5b372d60b33c3b471f4d1224bf63072c0441a Mon Sep 17 00:00:00 2001 From: Kayzel Date: Thu, 5 Nov 2026 11:09:34 +0100 Subject: [PATCH 07/13] refactor(server): one hold predicate for owed delegated tasks --- .../DelegatedCompletionDelivery.test.ts | 485 ++++++++---------- .../src/orchestration-v2/ProjectionStore.ts | 26 +- 2 files changed, 209 insertions(+), 302 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index e65a2c528db8e..256e3f982a630 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -347,39 +347,14 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { }); yield* sink.write({ events: [ - { - id: EventId.make("event:wake-handoff-successor-turn-item"), - type: "turn-item.updated", + subagentTurnItemEvent({ threadId, runId, - nodeId: taskId, - driver, - occurredAt: now, - payload: { - id: TurnItemId.make("turn-item:wake-handoff-successor"), - threadId, - runId, - nodeId: taskId, - providerThreadId: parentProviderThreadId(threadId), - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: 1, - status: "completed", - title: "Review the diff", - startedAt: now, - completedAt: now, - updatedAt: now, - type: "subagent", - subagentId: taskId, - origin: "app_owned", - driver, - providerInstanceId: modelSelection.instanceId, - childThreadId: null, - prompt: "Review the diff.", - result: "child finished", - }, - }, + taskId, + childThreadId: null, + status: "completed", + now, + }), ], }); const shell = yield* orchestrator.getThreadShell(threadId); @@ -1330,6 +1305,153 @@ const runEvent = (input: { }, }); +const subagentTurnItemEvent = (input: { + readonly threadId: ThreadId; + readonly runId: RunId; + readonly taskId: NodeId; + readonly childThreadId: ThreadId | null; + readonly status: "running" | "completed"; + readonly now: DateTime.Utc; +}) => ({ + id: EventId.make(`event:${input.taskId}:turn-item:${input.status}`), + type: "turn-item.updated" as const, + threadId: input.threadId, + runId: input.runId, + nodeId: input.taskId, + driver, + occurredAt: input.now, + payload: { + id: TurnItemId.make(`turn-item:${input.taskId}`), + threadId: input.threadId, + runId: input.runId, + nodeId: input.taskId, + providerThreadId: parentProviderThreadId(input.threadId), + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: input.status, + title: "Review the diff", + startedAt: input.now, + completedAt: input.status === "running" ? null : input.now, + updatedAt: input.now, + type: "subagent" as const, + subagentId: input.taskId, + origin: "app_owned" as const, + driver, + providerInstanceId: modelSelection.instanceId, + childThreadId: input.childThreadId, + prompt: "Review the diff.", + result: input.status === "running" ? null : "child finished", + }, +}); + +// A delegated child with a running task on the parent and a running run of its own. +const delegateChild = (input: { + readonly threadId: ThreadId; + readonly projectId: ProjectId; + readonly runId: RunId; + readonly rootNodeId: NodeId; + readonly name: string; + readonly turnItem?: boolean; + readonly now: DateTime.Utc; +}) => + Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const taskId = NodeId.make(`node:${input.name}`); + const childThreadId = ThreadId.make(`thread:${input.name}`); + const run = runEvent({ + threadId: childThreadId, + runId: RunId.make(`run:${input.name}`), + ordinal: 1, + status: "running", + now: input.now, + }); + yield* sink.write({ + events: [ + { + id: EventId.make(`event:${input.name}:thread`), + type: "thread.created", + threadId: childThreadId, + occurredAt: input.now, + payload: { + createdBy: "agent", + creationSource: "server", + id: childThreadId, + projectId: input.projectId, + title: input.name, + providerInstanceId: modelSelection.instanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { + parentThreadId: input.threadId, + relationshipToParent: "subagent", + rootThreadId: input.threadId, + }, + forkedFrom: { type: "node", nodeId: taskId }, + createdAt: input.now, + updatedAt: input.now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }, + { + id: EventId.make(`event:${input.name}:task`), + type: "subagent.updated", + threadId: input.threadId, + runId: input.runId, + nodeId: taskId, + driver, + providerInstanceId: modelSelection.instanceId, + occurredAt: input.now, + payload: { + id: taskId, + threadId: input.threadId, + runId: input.runId, + parentNodeId: input.rootNodeId, + origin: "app_owned", + createdBy: "agent", + driver, + providerInstanceId: modelSelection.instanceId, + providerThreadId: null, + childThreadId, + nativeTaskRef: null, + prompt: `Run ${input.name}.`, + title: null, + model: null, + completionWake: "always", + status: "running", + result: null, + startedAt: input.now, + completedAt: null, + updatedAt: input.now, + }, + }, + ...(input.turnItem === true + ? [ + subagentTurnItemEvent({ + threadId: input.threadId, + runId: input.runId, + taskId, + childThreadId, + status: "running", + now: input.now, + }), + ] + : []), + run, + ], + }); + return { task: { id: taskId }, childThreadId, run: run.payload }; + }); + const seedRestartCancelledChild = (input: { readonly parentThreadId: ThreadId; readonly projectId: ProjectId; @@ -1716,121 +1838,6 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( parentRunStatus: "completed", now, }); - const startChild = Effect.fn(function* (name: string) { - const taskId = NodeId.make(`node:wake-handoff-${name}`); - const childThreadId = ThreadId.make(`thread:wake-handoff-${name}`); - const childRun = runEvent({ - threadId: childThreadId, - runId: RunId.make(`run:wake-handoff-${name}`), - ordinal: 1, - status: "running", - now, - }); - yield* sink.write({ - events: [ - { - id: EventId.make(`event:wake-handoff-${name}-thread`), - type: "thread.created", - threadId: childThreadId, - occurredAt: now, - payload: { - createdBy: "agent", - creationSource: "server", - id: childThreadId, - projectId, - title: "Review the diff", - providerInstanceId: modelSelection.instanceId, - modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: null, - activeProviderThreadId: null, - lineage: { - parentThreadId: threadId, - relationshipToParent: "subagent", - rootThreadId: threadId, - }, - forkedFrom: { type: "node", nodeId: taskId }, - createdAt: now, - updatedAt: now, - archivedAt: null, - settledOverride: null, - settledAt: null, - lastVisitedAt: null, - deletedAt: null, - }, - }, - { - id: EventId.make(`event:wake-handoff-${name}-task`), - type: "subagent.updated", - threadId, - runId, - nodeId: taskId, - driver, - providerInstanceId: modelSelection.instanceId, - occurredAt: now, - payload: { - id: taskId, - threadId, - runId, - parentNodeId: rootNodeId, - origin: "app_owned", - createdBy: "agent", - driver, - providerInstanceId: modelSelection.instanceId, - providerThreadId: null, - childThreadId, - nativeTaskRef: null, - prompt: "Review the diff.", - title: null, - model: null, - completionWake: "always", - status: "running", - result: null, - startedAt: now, - completedAt: null, - updatedAt: now, - }, - }, - { - id: EventId.make(`event:wake-handoff-${name}-turn-item`), - type: "turn-item.updated", - threadId, - runId, - nodeId: taskId, - driver, - occurredAt: now, - payload: { - id: TurnItemId.make(`turn-item:wake-handoff-${name}`), - threadId, - runId, - nodeId: taskId, - providerThreadId: parentProviderThreadId(threadId), - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: 1, - status: "running", - title: "Review the diff", - startedAt: now, - completedAt: null, - updatedAt: now, - type: "subagent", - subagentId: taskId, - origin: "app_owned", - driver, - providerInstanceId: modelSelection.instanceId, - childThreadId, - prompt: "Review the diff.", - result: null, - }, - }, - childRun, - ], - }); - return { taskId, childThreadId, childRun }; - }); const roster = Effect.map(orchestrator.getThreadShell(threadId), (shell) => ( shell?.pendingBackgroundTasks?.map((task) => @@ -1838,27 +1845,33 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( ) ?? [] ).toSorted(), ); - const finishChild = Effect.fn(function* ( - child: Effect.Success>, - ) { - yield* sink.write({ + const complete = (run: OrchestrationV2Run) => + sink.write({ events: [ { - ...child.childRun, - id: EventId.make(`event:${child.childThreadId}-completed`), - payload: { ...child.childRun.payload, status: "completed", completedAt: now }, + ...runEvent({ + threadId: run.threadId, + runId: run.id, + ordinal: run.ordinal, + status: "completed", + now, + }), + payload: { ...run, status: "completed" as const, completedAt: now }, }, ], }); + // Returns the roster read after the child's result commits and before its wake run + // exists, then runs that wake the way the continuation worker does. + const finish = Effect.fn(function* ( + child: Effect.Success>, + ) { + yield* complete(child.run); let wake = yield* requests.take; while (wake.threadId !== threadId) wake = yield* requests.take; - return wake.delegatedCompletion!; - }); - const dispatchWake = ( - delivery: Effect.Success>, - taskId: NodeId, - ) => - orchestrator.dispatch({ + const delivery = wake.delegatedCompletion!; + assert.equal((yield* orchestrator.getThreadShell(threadId))?.status, "completed"); + const held = yield* roster; + yield* orchestrator.dispatch({ type: "message.dispatch", commandId: CommandId.make(`command:${delivery.messageId}`), threadId, @@ -1871,52 +1884,33 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( delegatedCompletion: { parentRunId: delivery.parentRunId, generation: delivery.generation, - taskIds: [taskId], + taskIds: [child.task.id], }, }); - - const first = yield* startChild("first"); - const second = yield* startChild("second"); - assert.deepEqual(yield* roster, [first.childThreadId, second.childThreadId]); - - const firstDelivery = yield* finishChild(first); - // The result is committed and the wake is queued for the continuation worker, but - // no wake run exists yet. Clients read a completed run with an empty roster as done. - const handoff = yield* orchestrator.getThreadShell(threadId); - assert.equal(handoff?.status, "completed"); - assert.deepEqual(yield* roster, [first.childThreadId, second.childThreadId]); - - yield* dispatchWake(firstDelivery, first.taskId); - const woken = yield* orchestrator.getThreadShell(threadId); - assert.notEqual(woken?.latestRunId, runId); - assert.deepEqual(woken?.pendingBackgroundTasks, []); - - const wakeRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( - (run) => run.userMessageId === firstDelivery.messageId, - )!; - yield* sink.write({ - events: [ - { - id: EventId.make("event:wake-handoff-first-wake-completed"), - type: "run.updated", - threadId, - runId: wakeRun.id, - providerInstanceId: modelSelection.instanceId, - occurredAt: now, - payload: { ...wakeRun, status: "completed", completedAt: now }, - }, - ], + const wakeRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.userMessageId === delivery.messageId, + )!; + yield* complete(wakeRun); + return held; }); - assert.deepEqual(yield* roster, [second.childThreadId]); + const child = (name: string) => + delegateChild({ + threadId, + projectId, + runId, + rootNodeId, + name: `wake-handoff-${name}`, + turnItem: true, + now, + }); - // The spawning run already has a newer run, so the hold must key on the open delivery. - const secondDelivery = yield* finishChild(second); + const first = yield* child("first"); + const second = yield* child("second"); + assert.deepEqual(yield* finish(first), [first.childThreadId, second.childThreadId]); assert.deepEqual(yield* roster, [second.childThreadId]); - yield* dispatchWake(secondDelivery, second.taskId); - assert.deepEqual( - (yield* orchestrator.getThreadShell(threadId))?.pendingBackgroundTasks, - [], - ); + // The spawning run already has a newer run, so the hold must key on the open delivery. + assert.deepEqual(yield* finish(second), [second.childThreadId]); + assert.deepEqual(yield* roster, []); }), ); it.effect("recovers a reserved delivery without loading conversation history", () => @@ -3795,89 +3789,14 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { deliveryState: "acknowledged", now, }); - // A delegated child with a running task on the parent and a running run of its own. const delegate = (name: string) => - Effect.gen(function* () { - const taskId = NodeId.make(`node:late-settle-${key}-${name}`); - const childThreadId = ThreadId.make(`thread:late-settle-${key}-${name}`); - const run = runEvent({ - threadId: childThreadId, - runId: RunId.make(`run:late-settle-${key}-${name}`), - ordinal: 1, - status: "running", - now, - }); - yield* sink.write({ - events: [ - { - id: EventId.make(`event:late-settle-${key}-${name}:thread`), - type: "thread.created", - threadId: childThreadId, - occurredAt: now, - payload: { - createdBy: "agent", - creationSource: "server", - id: childThreadId, - projectId: ProjectId.make(`late-settle-project-${key}`), - title: name, - providerInstanceId: modelSelection.instanceId, - modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: null, - activeProviderThreadId: null, - lineage: { - parentThreadId: threadId, - relationshipToParent: "subagent", - rootThreadId: threadId, - }, - forkedFrom: { type: "node", nodeId: taskId }, - createdAt: now, - updatedAt: now, - archivedAt: null, - settledOverride: null, - settledAt: null, - lastVisitedAt: null, - deletedAt: null, - }, - }, - { - id: EventId.make(`event:late-settle-${key}-${name}:task`), - type: "subagent.updated", - threadId, - runId, - nodeId: taskId, - driver, - providerInstanceId: modelSelection.instanceId, - occurredAt: now, - payload: { - id: taskId, - threadId, - runId, - parentNodeId: rootNodeId, - origin: "app_owned", - createdBy: "agent", - driver, - providerInstanceId: modelSelection.instanceId, - providerThreadId: null, - childThreadId, - nativeTaskRef: null, - prompt: `Run ${name}.`, - title: null, - model: null, - completionWake: "always", - status: "running", - result: null, - startedAt: now, - completedAt: null, - updatedAt: now, - }, - }, - run, - ], - }); - return { task: { id: taskId }, childThreadId, run: run.payload }; + delegateChild({ + threadId, + projectId: ProjectId.make(`late-settle-project-${key}`), + runId, + rootNodeId, + name: `late-settle-${key}-${name}`, + now, }); const settled = (taskId: NodeId, afterSequence: number) => sink.stream({ afterSequence, eventType: "subagent.updated" }).pipe( diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 734d284ae3b42..f7747950ff984 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6479,10 +6479,9 @@ export const layer: Layer.Layer = AND i.status NOT IN ('completed', 'interrupted', 'failed', 'cancelled') AND (i.run_id IS NULL OR r.status <> 'rolled_back') ${threadIds === undefined ? sql`` : sql`AND i.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)}))`} - -- A finished delegate's result commits before the continuation worker dispatches the - -- parent's wake run, so its task holds the parent until that wake run settles it. A task - -- left pending behind an open delivery holds until the next delivery claims it. Without an - -- owning instance it holds completion only, not the ended provider's caller access. + -- A finished delegate's result commits before its wake run exists, and a delivery run ends + -- before the next delivery is claimed, so an owed task holds the parent while a delivery + -- is reserved. Without an owning instance it holds completion, not caller access. UNION ALL SELECT i.thread_id, json_set(i.payload_json, '$.status', 'waiting') AS payload_json, NULL AS owning_provider_instance_id @@ -6490,24 +6489,13 @@ export const layer: Layer.Layer = INNER JOIN orchestration_v2_projection_runs r ON r.run_id = task.run_id INNER JOIN orchestration_v2_projection_turn_items i ON i.node_id = task.subagent_id AND i.thread_id = task.thread_id AND i.type = 'subagent' - WHERE task.completion_delivery_state IN ('claimed', 'pending') + WHERE task.status IN ('completed', 'interrupted', 'failed', 'cancelled') + AND task.completion_delivery_state IN ('claimed', 'pending') AND task.origin = 'app_owned' ${threadIds === undefined ? sql`` : sql`AND task.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)}))`} + AND i.status IN ('completed', 'interrupted', 'failed', 'cancelled') AND json_extract(r.payload_json, '$.delegatedCompletion.disposition') = 'open' - AND ( - ( - task.completion_delivery_state = 'claimed' - AND EXISTS ( - SELECT 1 FROM json_each(r.payload_json, '$.delegatedCompletion.delivery.taskIds') owed - WHERE owed.value = task.subagent_id - ) - ) - OR ( - task.completion_delivery_state = 'pending' - AND i.status IN ('completed', 'interrupted', 'failed', 'cancelled') - AND json_extract(r.payload_json, '$.delegatedCompletion.delivery.messageId') IS NOT NULL - ) - ) + AND json_extract(r.payload_json, '$.delegatedCompletion.delivery.messageId') IS NOT NULL ${ activeRunIds.length === 0 ? sql`` From 67f46ffe39faf3cc1a72ec8d31df178ef189a514 Mon Sep 17 00:00:00 2001 From: Kayzel Date: Thu, 5 Nov 2026 14:14:18 +0100 Subject: [PATCH 08/13] test(server): a pending task with no reserved delivery does not hold the parent --- .../DelegatedCompletionDelivery.test.ts | 90 ++++++++++--------- 1 file changed, 48 insertions(+), 42 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 256e3f982a630..750bb03128ccc 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -322,48 +322,54 @@ const seedParentWithTerminalTask = (input: { }); it.layer(layerTest)("delegated completion delivery repairs", (it) => { - it.effect("keeps an idle parent waiting while a finished task waits for the next delivery", () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const sink = yield* EventSink.EventSinkV2; - const now = yield* DateTime.now; - const threadId = ThreadId.make("thread:wake-handoff-successor"); - const runId = RunId.make("run:wake-handoff-successor"); - const deliveredTaskId = NodeId.make("node:wake-handoff-successor-delivered"); - const taskId = NodeId.make("node:wake-handoff-successor-pending"); - // The delivery run has ended, but the successor delivery is not claimed yet. - yield* seedParentWithTerminalTask({ - threadId, - projectId: ProjectId.make("project:wake-handoff-successor"), - runId, - rootNodeId: NodeId.make("node:wake-handoff-successor-root"), - taskId, - deliveryState: "pending", - deliveryTaskIds: [deliveredTaskId], - siblingTasks: [{ id: deliveredTaskId, deliveryState: "delivered" }], - parentRunStatus: "completed", - completionWake: "always", - now, - }); - yield* sink.write({ - events: [ - subagentTurnItemEvent({ - threadId, - runId, - taskId, - childThreadId: null, - status: "completed", - now, - }), - ], - }); - const shell = yield* orchestrator.getThreadShell(threadId); - assert.equal(shell?.status, "completed"); - assert.deepEqual( - shell?.pendingBackgroundTasks?.map((task) => task.kind), - ["subagent"], - ); - }), + it.effect.each([ + // The delivery run has ended, but the successor delivery is not claimed yet. + { name: "reserved", reserved: true, held: ["subagent"] }, + // No delivery will claim it, so it must not hold the parent forever. + { name: "unreserved", reserved: false, held: [] }, + ])( + "holds an idle parent on a pending finished task only while $name", + ({ name, reserved, held }) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make(`thread:wake-handoff-successor-${name}`); + const runId = RunId.make(`run:wake-handoff-successor-${name}`); + const deliveredTaskId = NodeId.make(`node:wake-handoff-successor-${name}-delivered`); + const taskId = NodeId.make(`node:wake-handoff-successor-${name}-pending`); + yield* seedParentWithTerminalTask({ + threadId, + projectId: ProjectId.make(`project:wake-handoff-successor-${name}`), + runId, + rootNodeId: NodeId.make(`node:wake-handoff-successor-${name}-root`), + taskId, + deliveryState: "pending", + ...(reserved ? { deliveryTaskIds: [deliveredTaskId] } : {}), + siblingTasks: [{ id: deliveredTaskId, deliveryState: "delivered" }], + parentRunStatus: "completed", + completionWake: "always", + now, + }); + yield* sink.write({ + events: [ + subagentTurnItemEvent({ + threadId, + runId, + taskId, + childThreadId: null, + status: "completed", + now, + }), + ], + }); + const shell = yield* orchestrator.getThreadShell(threadId); + assert.equal(shell?.status, "completed"); + assert.deepEqual( + shell?.pendingBackgroundTasks?.map((task) => task.kind), + held, + ); + }), ); it.effect( "interrupting a completion wake disposes both cohorts without reviving them on late results", From da4008c2f5c07af44116ae5ca572fc959dd85974 Mon Sep 17 00:00:00 2001 From: Kayzel Date: Thu, 5 Nov 2026 14:21:18 +0100 Subject: [PATCH 09/13] docs(server): the wake hold comment names the open cohort --- apps/server/src/orchestration-v2/ProjectionStore.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index f7747950ff984..507422b52da9e 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6480,8 +6480,9 @@ export const layer: Layer.Layer = AND (i.run_id IS NULL OR r.status <> 'rolled_back') ${threadIds === undefined ? sql`` : sql`AND i.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)}))`} -- A finished delegate's result commits before its wake run exists, and a delivery run ends - -- before the next delivery is claimed, so an owed task holds the parent while a delivery - -- is reserved. Without an owning instance it holds completion, not caller access. + -- before the next delivery is claimed, so an owed task holds the parent while its open + -- cohort has a delivery reserved. Without an owning instance it holds completion, not + -- caller access. UNION ALL SELECT i.thread_id, json_set(i.payload_json, '$.status', 'waiting') AS payload_json, NULL AS owning_provider_instance_id From 040cd0161066e8dc88c3033a249b580aabc4e315 Mon Sep 17 00:00:00 2001 From: Kayzel Date: Sun, 8 Nov 2026 14:31:45 +0100 Subject: [PATCH 10/13] test(server): a rolled-back run's owed task does not hold the parent --- .../DelegatedCompletionDelivery.test.ts | 21 +++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 750bb03128ccc..a8b539974e9f0 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -171,7 +171,7 @@ const seedParentWithTerminalTask = (input: { readonly deliveryState: "pending" | "delivered" | "claimed" | "acknowledged" | "disposed"; readonly completionWake?: "always" | "settled_only"; readonly deliveryTaskIds?: ReadonlyArray; - readonly parentRunStatus?: "running" | "completed"; + readonly parentRunStatus?: "running" | "completed" | "rolled_back"; readonly siblingTasks?: ReadonlyArray<{ readonly id: NodeId; readonly deliveryState: "delivered" | "claimed"; @@ -261,7 +261,7 @@ const seedParentWithTerminalTask = (input: { status: input.parentRunStatus ?? "running", requestedAt: input.now, startedAt: input.now, - completedAt: input.parentRunStatus === "completed" ? input.now : null, + completedAt: (input.parentRunStatus ?? "running") === "running" ? null : input.now, checkpointId: null, contextHandoffId: null, delegatedCompletion: { @@ -327,9 +327,11 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { { name: "reserved", reserved: true, held: ["subagent"] }, // No delivery will claim it, so it must not hold the parent forever. { name: "unreserved", reserved: false, held: [] }, + // A checkpoint rollback hides the run's items but keeps its cohort open, and a later run settles. + { name: "rolled back", reserved: true, held: [], parentRunStatus: "rolled_back" as const }, ])( "holds an idle parent on a pending finished task only while $name", - ({ name, reserved, held }) => + ({ name, reserved, held, parentRunStatus = "completed" }) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const sink = yield* EventSink.EventSinkV2; @@ -347,7 +349,7 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { deliveryState: "pending", ...(reserved ? { deliveryTaskIds: [deliveredTaskId] } : {}), siblingTasks: [{ id: deliveredTaskId, deliveryState: "delivered" }], - parentRunStatus: "completed", + parentRunStatus, completionWake: "always", now, }); @@ -361,6 +363,17 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { status: "completed", now, }), + ...(parentRunStatus === "rolled_back" + ? [ + runEvent({ + threadId, + runId: RunId.make(`${runId}-later`), + ordinal: 2, + status: "completed", + now, + }), + ] + : []), ], }); const shell = yield* orchestrator.getThreadShell(threadId); From 985aaf2e9647b5ed14dfa73407d2196f616a59eb Mon Sep 17 00:00:00 2001 From: Kayzel Date: Sun, 8 Nov 2026 14:33:16 +0100 Subject: [PATCH 11/13] fix(server): a rolled-back run's owed task no longer holds the parent --- apps/server/src/orchestration-v2/ProjectionStore.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 507422b52da9e..e331e5f549f4d 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6491,6 +6491,7 @@ export const layer: Layer.Layer = INNER JOIN orchestration_v2_projection_turn_items i ON i.node_id = task.subagent_id AND i.thread_id = task.thread_id AND i.type = 'subagent' WHERE task.status IN ('completed', 'interrupted', 'failed', 'cancelled') + AND r.status <> 'rolled_back' AND task.completion_delivery_state IN ('claimed', 'pending') AND task.origin = 'app_owned' ${threadIds === undefined ? sql`` : sql`AND task.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)}))`} From e24a9b87031380ce38298f3cc47d5b467aef2ee4 Mon Sep 17 00:00:00 2001 From: Kayzel Date: Sun, 8 Nov 2026 14:37:42 +0100 Subject: [PATCH 12/13] test(server): type the rollback row's parent run status --- .../orchestration-v2/DelegatedCompletionDelivery.test.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index a8b539974e9f0..5425407efa65d 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -324,14 +324,14 @@ const seedParentWithTerminalTask = (input: { it.layer(layerTest)("delegated completion delivery repairs", (it) => { it.effect.each([ // The delivery run has ended, but the successor delivery is not claimed yet. - { name: "reserved", reserved: true, held: ["subagent"] }, + { name: "reserved", reserved: true, held: ["subagent"], parentRunStatus: "completed" as const }, // No delivery will claim it, so it must not hold the parent forever. - { name: "unreserved", reserved: false, held: [] }, + { name: "unreserved", reserved: false, held: [], parentRunStatus: "completed" as const }, // A checkpoint rollback hides the run's items but keeps its cohort open, and a later run settles. { name: "rolled back", reserved: true, held: [], parentRunStatus: "rolled_back" as const }, ])( "holds an idle parent on a pending finished task only while $name", - ({ name, reserved, held, parentRunStatus = "completed" }) => + ({ name, reserved, held, parentRunStatus }) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const sink = yield* EventSink.EventSinkV2; From 4e78cb72e355a0fbcd1606c878001fe2db3c92d8 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 03:45:42 +0000 Subject: [PATCH 13/13] perf(server): the owed-task hold starts from settled delegated tasks Without planner stats, the hold arm of the shell pending read walked every turn item of each requested thread. Pin it to the settled subagents index, then runs and turn items by key, and assert the plan. AI-assisted; verified with the tests and tools listed above. --- .../orchestration-v2/ProjectionStore.test.ts | 23 +++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 6 +++-- 2 files changed, 27 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 40b7709870585..74d62916d7244 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1733,6 +1733,29 @@ const commandReadsSkipSettledLeafNodes = Effect.gen(function* () { [...(yield* store.listAppOwnedChildThreadIds(threadId))], [childThreadIds.delivered, childThreadIds["no-delivery"]], ); + // The owed-task hold starts from settled tasks, not from every turn item of the thread. + const { statements, tracer } = traceSqlStatements(); + yield* store.getThreadShell(threadId).pipe(Effect.withTracer(tracer)); + const shellPendingQuery = statements.find((statement) => + statement.includes("FROM orchestration_v2_projection_subagents task"), + ); + assert.isDefined(shellPendingQuery); + const sql = yield* SqlClient.SqlClient; + const plan = (yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${shellPendingQuery}`, + )).flatMap((row) => (row.detail.startsWith("SEARCH ") ? [row.detail] : [])); + const holdStart = plan.findIndex((detail) => detail.startsWith("SEARCH task ")); + assert.isAtLeast(holdStart, 0); + const [taskLookup, runLookup, itemLookup] = plan.slice(holdStart, holdStart + 3); + assert.include(taskLookup, "orchestration_v2_projection_subagents_settled_idx"); + assert.include( + runLookup, + "SEARCH r USING INDEX sqlite_autoindex_orchestration_v2_projection_runs_1", + ); + assert.include( + itemLookup, + "SEARCH i USING INDEX orchestration_v2_projection_turn_items_node_ordinal_idx", + ); }); const backgroundThreadUpdatesKeepTimestamp = Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index bdf3f0d602887..d8c576792038f 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6603,9 +6603,11 @@ export const layer: Layer.Layer = UNION ALL SELECT i.thread_id, json_set(i.payload_json, '$.status', 'waiting') AS payload_json, NULL AS owning_provider_instance_id + -- Pinned: without stats the planner walks every turn item of the thread instead. FROM orchestration_v2_projection_subagents task - INNER JOIN orchestration_v2_projection_runs r ON r.run_id = task.run_id - INNER JOIN orchestration_v2_projection_turn_items i + INDEXED BY orchestration_v2_projection_subagents_settled_idx + CROSS JOIN orchestration_v2_projection_runs r ON r.run_id = task.run_id + CROSS JOIN orchestration_v2_projection_turn_items i ON i.node_id = task.subagent_id AND i.thread_id = task.thread_id AND i.type = 'subagent' WHERE task.status IN ('completed', 'interrupted', 'failed', 'cancelled') AND r.status <> 'rolled_back'