From 85ffe14e5e5552865bf010152f7c41557f543e2a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 19:38:26 +0000 Subject: [PATCH 1/6] test(server): reproduce #31087 parent shell settling before its completion wake The parent's shell drops the finished delegate from pendingBackgroundTasks as soon as finalize claims the completion, before the wake run exists, so the shell reads settled in that window. AI-assisted; verified with the tests and tools listed above. --- .../DelegatedCompletionDelivery.test.ts | 113 ++++++++++++++++++ 1 file changed, 113 insertions(+) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 3f0f9cf0342ee..d7034f99b32e5 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -1,4 +1,5 @@ import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts"; +import { backgroundWorkHoldsCompletion } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; @@ -1583,6 +1584,118 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( assert.notEqual(deliveryRun?.status, "queued"); }), ); + it.effect( + "keeps the parent shell unsettled while its delegated completion wake is pending", + () => + 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:pending-wake-shell"); + yield* seedParentWithTerminalTask({ + threadId, + projectId: ProjectId.make("project:pending-wake-shell"), + runId: RunId.make("run:pending-wake-shell-seed"), + rootNodeId: NodeId.make("node:pending-wake-shell-seed"), + taskId: NodeId.make("node:pending-wake-shell-seed-task"), + deliveryState: "acknowledged", + parentRunStatus: "completed", + now, + }); + const messageId = MessageId.make("message:pending-wake-shell"); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:pending-wake-shell-send"), + threadId, + messageId, + text: "Delegate this and end your turn", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + const parentRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.userMessageId === messageId, + )!; + const runId = parentRun.id; + yield* orchestrator.dispatch({ + type: "delegated_task.request", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make("command:pending-wake-shell-delegate"), + parentThreadId: threadId, + parentRunId: runId, + parentNodeId: parentRun.rootNodeId!, + task: "Run in the background", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + completionWake: "always", + }); + const started = yield* orchestrator.getThreadProjection(threadId); + const task = started.subagents.find((row) => row.runId === runId && row.childThreadId)!; + yield* sink.write({ + events: [ + { + ...runEvent({ + threadId, + runId, + ordinal: parentRun.ordinal, + status: "completed", + now, + }), + payload: { + ...started.runs.find((run) => run.id === runId)!, + status: "completed", + completedAt: now, + }, + }, + ], + }); + assert.isTrue( + backgroundWorkHoldsCompletion( + (yield* orchestrator.getThreadShell(threadId))?.pendingBackgroundTasks ?? [], + ), + ); + + const childThreadId = task.childThreadId!; + const childRun = (yield* orchestrator.getThreadProjection(childThreadId)).runs[0]!; + yield* sink.write({ + events: [ + { + ...runEvent({ + threadId: childThreadId, + runId: childRun.id, + ordinal: childRun.ordinal, + status: "completed", + now, + }), + payload: { ...childRun, status: "completed", completedAt: now }, + }, + ], + }); + let wake = yield* requests.take; + while (wake.threadId !== threadId) wake = yield* requests.take; + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + projection.subagents.find((row) => row.id === task.id)?.completionDelivery?.state, + "claimed", + ); + assert.isFalse( + projection.runs.some( + (run) => run.userMessageId === wake.delegatedCompletion?.messageId, + ), + ); + + const shell = yield* orchestrator.getThreadShell(threadId); + assert.isTrue( + backgroundWorkHoldsCompletion(shell?.pendingBackgroundTasks ?? []), + `parent shell reads settled (status ${shell?.status}, no held background work) while its delegated completion wake is still pending`, + ); + }), + ); it.effect("pauses queued follow-up cards and resumes them", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; From 7af21a1140ddc15debd1b0b84192e81bd5cd3a1c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 20:02:38 +0000 Subject: [PATCH 2/6] fix(server): parent threads stay waiting until a delegated result wakes them Finishing a delegated task commits its result, claims the parent's completion wake, and only then offers the wake, which queues its run in a later command. In between, the parent's shell read completed with no held background work, so clients played a false "Thread completed". Shells now keep the finished task's subagent item as held pending work while the parent run's claimed delivery has no wake run yet: no message with its id, or a steer that went unread. That predicate now lives in NotificationMailbox and is shared by the continuation service, the wake offer, and every shell producer (projection, thread shell, snapshot, awareness, settlement candidates). SQL shells find candidates through the indexed claimed delivery state, so ordinary shell reads pay one index probe. A run without its provider thread is never offered a wake, so it holds nothing. The active shell index counts claimed tasks as possibly live. Fixes pingdotgg/t3code#31087 AI-assisted; verified with the tests and tools listed above. --- .../orchestration-v2/NotificationMailbox.ts | 49 ++++ .../src/orchestration-v2/Orchestrator.ts | 33 ++- .../orchestration-v2/ProjectionStore.test.ts | 204 ++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 224 ++++++++++++++---- .../ProviderContinuationService.ts | 23 +- .../orchestrationV2PendingBackgroundWork.ts | 7 +- 6 files changed, 460 insertions(+), 80 deletions(-) diff --git a/apps/server/src/orchestration-v2/NotificationMailbox.ts b/apps/server/src/orchestration-v2/NotificationMailbox.ts index 29ba40c5777fe..27a58e2f26571 100644 --- a/apps/server/src/orchestration-v2/NotificationMailbox.ts +++ b/apps/server/src/orchestration-v2/NotificationMailbox.ts @@ -2,7 +2,9 @@ import { latestProviderTurnForAttempt, type MessageId, type OrchestrationV2ConversationMessage, + type OrchestrationV2Run, type OrchestrationV2ThreadProjection, + type OrchestrationV2TurnItem, } from "@t3tools/contracts"; export function isBufferedProviderNotification( @@ -32,3 +34,50 @@ export function isUndeliveredMailboxSteer( "completed") ); } + +type DelegatedCompletionRecords = Pick< + OrchestrationV2ThreadProjection, + "messages" | "runs" | "providerTurns" +>; + +/** The run's claimed delegated completion still owes a wake: none dispatched, or its steer went unread. */ +export function pendingDelegatedCompletionWake( + projection: DelegatedCompletionRecords, + run: Pick, +) { + const delivery = run.delegatedCompletion?.delivery; + if (run.delegatedCompletion?.disposition !== "open" || delivery == null) return undefined; + const dispatched = projection.messages.some((message) => message.id === delivery.messageId); + return dispatched && !isUndeliveredMailboxSteer(projection, delivery.messageId) + ? undefined + : delivery; +} + +/** + * Finished delegated task items whose completion wake has not started its run. A run without its + * provider thread is never offered a wake, so its tasks hold nothing. + */ +export function delegatedTaskItemsAwaitingWake( + projection: DelegatedCompletionRecords & { + readonly providerThreads: ReadonlyArray<{ readonly id: string }>; + readonly turnItems: ReadonlyArray; + }, +): ReadonlyArray { + const taskIds = new Set(); + for (const run of projection.runs) { + if ( + run.providerThreadId === null || + !projection.providerThreads.some((thread) => thread.id === run.providerThreadId) + ) { + continue; + } + for (const taskId of pendingDelegatedCompletionWake(projection, run)?.taskIds ?? []) { + taskIds.add(taskId); + } + } + if (taskIds.size === 0) return []; + return projection.turnItems.filter( + (item) => + item.type === "subagent" && item.origin === "app_owned" && taskIds.has(item.subagentId), + ); +} diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index b4bdbd705c290..1ba19ab024f59 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -128,6 +128,7 @@ import { import { isBufferedProviderNotification, isUndeliveredMailboxSteer, + pendingDelegatedCompletionWake, } from "./NotificationMailbox.ts"; import { EventSinkV2 } from "./EventSink.ts"; import { @@ -1409,18 +1410,26 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio messageIds: [delivery.messageId], }); const message = messages.messages[0]; - if (message !== undefined) { - if (message.delegatedCompletion === undefined || message.runId === null) return; - const runs = yield* projectionStore.getThreadRecords(threadId, ["runs"], { - runIds: [message.runId], - }); - const run = runs.runs[0]; - const turns = yield* projectionStore.getThreadRecords(threadId, ["providerTurns"], { - providerTurnAttemptIds: run?.activeAttemptId == null ? [] : [run.activeAttemptId], - }); - if (!isUndeliveredMailboxSteer({ ...messages, ...runs, ...turns }, delivery.messageId)) - return; - } + const runs = + message?.delegatedCompletion === undefined || message.runId === null + ? [] + : (yield* projectionStore.getThreadRecords(threadId, ["runs"], { + runIds: [message.runId], + })).runs; + const attemptId = runs[0]?.activeAttemptId; + const providerTurns = + attemptId == null + ? [] + : (yield* projectionStore.getThreadRecords(threadId, ["providerTurns"], { + providerTurnAttemptIds: [attemptId], + })).providerTurns; + if ( + pendingDelegatedCompletionWake( + { messages: messages.messages, runs, providerTurns }, + parentRun, + ) === undefined + ) + return; const projection = yield* projectionStore.getThreadRecords(threadId, ["providerThreads"], { providerThreadIds: parentRun.providerThreadId === null ? [] : [parentRun.providerThreadId], }); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index c6187a97ea8db..119f5584ffba7 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -10,6 +10,7 @@ import { MessageId, type ModelSelection, type OrchestrationV2ProviderThread, + type OrchestrationV2Run, type OrchestrationV2TurnItem, type OrchestrationV2ThreadHistoryPage, type OrchestrationV2ThreadShellSnapshot, @@ -1273,6 +1274,205 @@ const threadWorktreePathsIncludeArchivedOnRequest = Effect.gen(function* () { ]); }); +const delegatedCompletionWakeHoldsEveryShell = Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = yield* addRolledBackRecoveryCandidate("delegated-wake-hold"); + const providerThreadId = ProviderThreadId.make("provider-thread:delegated-wake-hold"); + yield* store.apply({ + id: EventId.make("event:delegated-wake-hold:provider-thread"), + type: "provider-thread.updated", + threadId, + occurredAt: now, + payload: { + id: providerThreadId, + appThreadId: threadId, + driver, + providerInstanceId, + providerSessionId: null, + ownerNodeId: null, + nativeThreadRef: null, + nativeConversationHeadRef: null, + lastRunOrdinal: null, + handoffIds: [], + forkedFrom: null, + status: "active", + firstRunOrdinal: 1, + createdAt: now, + updatedAt: now, + }, + }); + const taskId = NodeId.make("node:delegated-wake-hold:task"); + const wakeMessageId = MessageId.make("message:delegated-wake-hold:wake"); + const parentRun = { + ...(yield* store.getThreadProjection(threadId)).runs[0]!, + id: RunId.make("run:delegated-wake-hold:parent"), + ordinal: 2, + providerThreadId, + userMessageId: MessageId.make("message:delegated-wake-hold:parent"), + rootNodeId: NodeId.make("node:delegated-wake-hold:parent"), + status: "completed" as const, + completedAt: now, + delegatedCompletion: { + disposition: "open" as const, + nextGeneration: 2, + delivery: { generation: 1, messageId: wakeMessageId, taskIds: [taskId] }, + }, + }; + const applyRun = (name: string, run: OrchestrationV2Run) => + store.apply({ + id: EventId.make(`event:delegated-wake-hold:${name}`), + type: "run.updated", + threadId, + runId: run.id, + driver, + providerInstanceId, + occurredAt: now, + payload: run, + }); + yield* applyRun("parent", parentRun); + const itemId = TurnItemId.make("item:delegated-wake-hold:task"); + yield* store.apply({ + id: EventId.make("event:delegated-wake-hold:item"), + type: "turn-item.updated", + threadId, + runId: parentRun.id, + nodeId: taskId, + driver, + occurredAt: now, + payload: { + id: itemId, + type: "subagent", + threadId, + runId: parentRun.id, + nodeId: taskId, + providerThreadId, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 10, + status: "completed", + title: "Delegated task", + startedAt: now, + completedAt: now, + updatedAt: now, + subagentId: taskId, + origin: "app_owned", + driver, + providerInstanceId, + childThreadId: null, + prompt: "Do the work", + result: "done", + }, + }); + yield* store.apply({ + id: EventId.make("event:delegated-wake-hold:task"), + type: "subagent.updated", + threadId, + runId: parentRun.id, + nodeId: taskId, + driver, + providerInstanceId, + occurredAt: now, + payload: { + id: taskId, + threadId, + runId: parentRun.id, + parentNodeId: parentRun.rootNodeId, + origin: "app_owned", + createdBy: "agent", + driver, + providerInstanceId, + providerThreadId: null, + childThreadId: null, + nativeTaskRef: null, + prompt: "Do the work", + title: "Delegated task", + model: null, + completionWake: "always", + completionDelivery: { state: "claimed", observedByRunId: null }, + status: "completed", + result: "done", + startedAt: now, + completedAt: now, + updatedAt: now, + }, + }); + const heldTaskIds = Effect.gen(function* () { + yield* assertShellIndexCoversShells(); + const taskIds = ( + shell: + | { + readonly pendingBackgroundTasks?: + | ReadonlyArray<{ readonly taskId: string }> + | undefined; + } + | null + | undefined, + ) => (shell?.pendingBackgroundTasks ?? []).map((task) => task.taskId); + const producers = [ + taskIds(yield* store.getThreadShell(threadId)), + taskIds((yield* store.getShellSnapshot()).threads.find((row) => row.id === threadId)), + taskIds( + ProjectionStore.threadShellFromProjection(yield* store.getThreadProjection(threadId)), + ), + taskIds((yield* store.getSettlementCandidates(threadId))[0]), + ]; + for (const held of producers) assert.deepEqual(held, producers[0]); + return producers[0]; + }); + + assert.deepEqual(yield* heldTaskIds, [itemId]); + + // No provider thread means no wake is ever offered. + yield* applyRun("missing-provider-thread", { + ...parentRun, + providerThreadId: ProviderThreadId.make("provider-thread:delegated-wake-hold:missing"), + }); + assert.deepEqual(yield* heldTaskIds, []); + yield* applyRun("parent-restored", parentRun); + + const wakeRun = { + ...parentRun, + id: RunId.make("run:delegated-wake-hold:wake"), + ordinal: 3, + userMessageId: wakeMessageId, + rootNodeId: NodeId.make("node:delegated-wake-hold:wake"), + delegatedCompletion: { disposition: "open" as const, nextGeneration: 1, delivery: null }, + }; + yield* applyRun("wake-run", wakeRun); + yield* store.apply({ + id: EventId.make("event:delegated-wake-hold:wake-message"), + type: "message.updated", + threadId, + runId: wakeRun.id, + occurredAt: now, + payload: { + id: wakeMessageId, + threadId, + runId: wakeRun.id, + nodeId: null, + role: "user", + text: "Delegated task finished", + attachments: [], + streaming: false, + createdBy: "agent", + creationSource: "server", + delegatedCompletion: { parentRunId: parentRun.id, generation: 1, taskIds: [taskId] }, + createdAt: now, + updatedAt: now, + }, + }); + assert.deepEqual(yield* heldTaskIds, []); + + // A wake steered into a run that ended before reading it is offered again. + yield* applyRun("wake-run-steered", { + ...wakeRun, + userMessageId: MessageId.make("message:delegated-wake-hold:steered-into"), + }); + assert.deepEqual(yield* heldTaskIds, [itemId]); +}); + const commandReadsSkipSettledLeafNodes = Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; const now = yield* DateTime.now; @@ -2234,6 +2434,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { "keeps restart-cancelled work through a stale run.updated", () => restartCancelledWorkSurvivesStaleRunUpdate, ); + it.effect( + "every shell holds a finished delegated task until its completion wake starts", + () => delegatedCompletionWakeHoldsEveryShell, + ); it.effect("records restart-cancelled work without regressing a run that completed since", () => Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index b5a0f2cfb5556..9788e392ad528 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -5,6 +5,7 @@ import { projectTurnItemForWire } from "./WireProjection.ts"; import * as Stream from "effect/Stream"; import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import { makeThreadFind } from "./ThreadFind.ts"; +import { delegatedTaskItemsAwaitingWake } from "./NotificationMailbox.ts"; import type { OrchestrationV2SearchThreadInput, OrchestrationV2SearchThreadResult, @@ -994,6 +995,14 @@ type ShellPendingTurnItemRow = PayloadRow & { readonly owning_provider_instance_id: string | null; }; +type ShellAwaitingWakeRow = ShellPendingTurnItemRow & { + readonly provider_thread_id: string; + readonly run_payload_json: string; + readonly message_payload_json: string | null; + readonly message_run_payload_json: string | null; + readonly message_turn_payload_json: string | null; +}; + export type AgentAwarenessThread = ProjectThreadAwarenessV2Input["thread"] & Pick< OrchestrationV2ThreadShell, @@ -1589,6 +1598,7 @@ export function threadShellFromProjection( activeProviderThreadId: projection.thread.activeProviderThreadId, runs, pullRequests: projection.thread.pullRequests, + awaitingWakeTurnItems: delegatedTaskItemsAwaitingWake(projection), }); return { createdBy: projection.thread.createdBy, @@ -6415,6 +6425,36 @@ export const layer: Layer.Layer = } `; + // Claimed tasks are rare and indexed; the run, message and turn joins only run for them. + const selectShellAwaitingWakeRows = (threadIds: ReadonlyArray) => + sql` + SELECT s.thread_id, p.provider_thread_id, item.payload_json, + owner.provider_instance_id AS owning_provider_instance_id, + r.payload_json AS run_payload_json, + m.payload_json AS message_payload_json, + mr.payload_json AS message_run_payload_json, + CASE WHEN mr.run_id IS NULL THEN NULL ELSE ( + SELECT turn.payload_json FROM orchestration_v2_projection_provider_turns turn + WHERE turn.thread_id = s.thread_id + AND turn.run_attempt_id = json_extract(mr.payload_json, '$.activeAttemptId') + ORDER BY turn.ordinal DESC + LIMIT 1 + ) END AS message_turn_payload_json + FROM orchestration_v2_projection_subagents s + INNER JOIN orchestration_v2_projection_runs r ON r.run_id = s.run_id + INNER JOIN orchestration_v2_projection_provider_threads p + ON p.provider_thread_id = r.provider_thread_id + INNER JOIN orchestration_v2_projection_turn_items item + ON item.node_id = s.subagent_id AND item.thread_id = s.thread_id AND item.type = 'subagent' + LEFT JOIN orchestration_v2_projection_provider_threads owner + ON owner.provider_thread_id = item.provider_thread_id + LEFT JOIN orchestration_v2_projection_messages m + ON m.message_id = json_extract(r.payload_json, '$.delegatedCompletion.delivery.messageId') + LEFT JOIN orchestration_v2_projection_runs mr ON mr.run_id = m.run_id + WHERE s.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) + AND s.completion_delivery_state = 'claimed' + `; + const runMapsByThreadId = (input: { readonly runRows: ReadonlyArray; readonly itemCountRows: ReadonlyArray; @@ -6454,6 +6494,7 @@ export const layer: Layer.Layer = )(function* (input: { readonly providerThreadRows: ReadonlyArray; readonly pendingTurnItemRows: ReadonlyArray; + readonly awaitingWakeRows: ReadonlyArray; }) { const providerThreadsByThreadId = new Map< ThreadId, @@ -6472,22 +6513,63 @@ export const layer: Layer.Layer = } const pendingTurnItemsByThreadId = new Map>(); + const decodeShellTurnItem = (row: ShellPendingTurnItemRow) => + decodeTurnItemPayload(row.payload_json).pipe( + Effect.map((turnItem) => ({ + ...turnItem, + ...(row.owning_provider_instance_id === null + ? {} + : { + owningProviderInstanceId: ProviderInstanceId.make( + row.owning_provider_instance_id, + ), + }), + })), + ); for (const row of input.pendingTurnItemRows) { - const turnItem = { - ...(yield* decodeTurnItemPayload(row.payload_json)), - ...(row.owning_provider_instance_id === null - ? {} - : { - owningProviderInstanceId: ProviderInstanceId.make(row.owning_provider_instance_id), - }), - }; + const turnItem = yield* decodeShellTurnItem(row); const threadId = ThreadId.make(row.thread_id); const existing = pendingTurnItemsByThreadId.get(threadId) ?? []; existing.push(turnItem); pendingTurnItemsByThreadId.set(threadId, existing); } - return { providerThreadsByThreadId, pendingTurnItemsByThreadId }; + const awaitingWakeTurnItemsByThreadId = new Map>(); + for (const row of input.awaitingWakeRows) { + const message = + row.message_payload_json === null + ? null + : yield* decodeMessagePayload(row.message_payload_json); + const messageRun = + row.message_run_payload_json === null + ? null + : yield* decodeRunPayload(row.message_run_payload_json); + const messageTurn = + row.message_turn_payload_json === null + ? null + : yield* decodeProviderTurnPayload(row.message_turn_payload_json); + const items = delegatedTaskItemsAwaitingWake({ + runs: [ + yield* decodeRunPayload(row.run_payload_json), + ...(messageRun === null ? [] : [messageRun]), + ], + messages: message === null ? [] : [message], + providerTurns: messageTurn === null ? [] : [messageTurn], + providerThreads: [{ id: row.provider_thread_id }], + turnItems: [yield* decodeShellTurnItem(row)], + }); + if (items.length === 0) continue; + const threadId = ThreadId.make(row.thread_id); + const existing = awaitingWakeTurnItemsByThreadId.get(threadId) ?? []; + existing.push(...items); + awaitingWakeTurnItemsByThreadId.set(threadId, existing); + } + + return { + providerThreadsByThreadId, + pendingTurnItemsByThreadId, + awaitingWakeTurnItemsByThreadId, + }; }); const getSettlementCandidates: ProjectionStoreV2Shape["getSettlementCandidates"] = (threadId) => @@ -6560,11 +6642,15 @@ export const layer: Layer.Layer = `; if (rows.length === 0) return []; const threadIds = rows.map((row) => ThreadId.make(row.thread_id)); - const { providerThreadsByThreadId, pendingTurnItemsByThreadId } = - yield* pendingBackgroundDataByThreadId({ - providerThreadRows: yield* selectShellProviderThreadRows(threadIds), - pendingTurnItemRows: yield* selectShellPendingTurnItemRows(threadIds), - }); + const { + providerThreadsByThreadId, + pendingTurnItemsByThreadId, + awaitingWakeTurnItemsByThreadId, + } = yield* pendingBackgroundDataByThreadId({ + providerThreadRows: yield* selectShellProviderThreadRows(threadIds), + pendingTurnItemRows: yield* selectShellPendingTurnItemRows(threadIds), + awaitingWakeRows: yield* selectShellAwaitingWakeRows(threadIds), + }); return yield* Effect.forEach(rows, (row) => Effect.gen(function* () { const thread = yield* decodeThreadPayload(row.payload_json); @@ -6614,6 +6700,7 @@ export const layer: Layer.Layer = activeProviderThreadId: thread.activeProviderThreadId, hasActiveRun: false, pullRequests: thread.pullRequests, + awaitingWakeTurnItems: awaitingWakeTurnItemsByThreadId.get(thread.id) ?? [], }), } satisfies ProjectionSettlementCandidate; }), @@ -6710,6 +6797,10 @@ export const layer: Layer.Layer = ThreadId, ReadonlyArray >; + readonly awaitingWakeTurnItemsByThreadId: ReadonlyMap< + ThreadId, + ReadonlyArray + >; }) { const { row, @@ -6718,6 +6809,7 @@ export const layer: Layer.Layer = shellItemCountsByThreadId, providerThreadsByThreadId, pendingTurnItemsByThreadId, + awaitingWakeTurnItemsByThreadId, } = input; const thread = yield* decodeThreadPayload(row.payload_json); const latestUserAuthoredMessageAt = @@ -6817,6 +6909,7 @@ export const layer: Layer.Layer = activeProviderThreadId: thread.activeProviderThreadId, hasActiveRun: row.active_run_id !== null, pullRequests: thread.pullRequests, + awaitingWakeTurnItems: awaitingWakeTurnItemsByThreadId.get(thread.id) ?? [], }), ]; return { @@ -6914,6 +7007,7 @@ export const layer: Layer.Layer = sequenceRows, providerThreadRows, pendingTurnItemRows, + awaitingWakeRows, ] = yield* Effect.all([ readForThreadIds(selectShellRunRows, forkSourceIds), readForThreadIds(selectShellRunItemCounts, forkSourceIds), @@ -6935,6 +7029,7 @@ export const layer: Layer.Layer = ), ), ), + readForThreadIds(selectShellAwaitingWakeRows), ]); return { targetThreadIds, @@ -6945,6 +7040,7 @@ export const layer: Layer.Layer = sequenceRows, providerThreadRows, pendingTurnItemRows, + awaitingWakeRows, }; }); @@ -6957,22 +7053,25 @@ export const layer: Layer.Layer = sequenceRows, providerThreadRows, pendingTurnItemRows, + awaitingWakeRows, }: Effect.Success>) { const { runOrdinalsByThreadId, itemCountsByThreadId } = runMapsByThreadId({ runRows, itemCountRows, }); const shellItemCountsByThreadId = shellItemCountMap(shellItemCountRows); - const { providerThreadsByThreadId, pendingTurnItemsByThreadId } = - yield* pendingBackgroundDataByThreadId({ providerThreadRows, pendingTurnItemRows }); + const background = yield* pendingBackgroundDataByThreadId({ + providerThreadRows, + pendingTurnItemRows, + awaitingWakeRows, + }); const states = yield* Effect.forEach(threadRows, (row) => shellThreadStateFromRow({ row, runOrdinalsByThreadId, itemCountsByThreadId, shellItemCountsByThreadId, - providerThreadsByThreadId, - pendingTurnItemsByThreadId, + ...background, }), ); const statesByThreadId = new Map(states.map((state) => [state.thread.id, state])); @@ -7040,36 +7139,52 @@ export const layer: Layer.Layer = const getAgentAwarenessSnapshot = Effect.fn("ProjectionStore.getAgentAwarenessSnapshot")( function* (threadId?: ThreadId) { - const { threadRows, providerThreadRows, pendingTurnItemRows, sequenceRows } = - yield* sql.withTransaction( - Effect.gen(function* () { - const threadRows = yield* selectShellThreadRows( - threadId, - threadId === undefined ? { location: "active", listedOnly: true } : {}, - ); - const ids = threadRows.map((row) => ThreadId.make(row.thread_id)); - const providerThreadRows = - ids.length === 0 ? [] : yield* selectShellProviderThreadRows(ids); - const pendingTurnItemRows = - ids.length === 0 - ? [] - : yield* selectShellPendingTurnItemRows( - ids, - threadRows.flatMap((row) => - row.active_run_id !== null && row.active_run_status === "running" - ? [row.active_run_id] - : [], - ), - ); - const sequenceRows = yield* sql<{ readonly snapshot_sequence: number | null }>` + const { + threadRows, + providerThreadRows, + pendingTurnItemRows, + awaitingWakeRows, + sequenceRows, + } = yield* sql.withTransaction( + Effect.gen(function* () { + const threadRows = yield* selectShellThreadRows( + threadId, + threadId === undefined ? { location: "active", listedOnly: true } : {}, + ); + const ids = threadRows.map((row) => ThreadId.make(row.thread_id)); + const providerThreadRows = + ids.length === 0 ? [] : yield* selectShellProviderThreadRows(ids); + const pendingTurnItemRows = + ids.length === 0 + ? [] + : yield* selectShellPendingTurnItemRows( + ids, + threadRows.flatMap((row) => + row.active_run_id !== null && row.active_run_status === "running" + ? [row.active_run_id] + : [], + ), + ); + const awaitingWakeRows = + ids.length === 0 ? [] : yield* selectShellAwaitingWakeRows(ids); + const sequenceRows = yield* sql<{ readonly snapshot_sequence: number | null }>` SELECT MAX(sequence) AS snapshot_sequence FROM orchestration_events WHERE application_event_version = 2 AND aggregate_kind = 'thread' `; - return { threadRows, providerThreadRows, pendingTurnItemRows, sequenceRows }; - }), - ); - const { providerThreadsByThreadId, pendingTurnItemsByThreadId } = - yield* pendingBackgroundDataByThreadId({ providerThreadRows, pendingTurnItemRows }); + return { + threadRows, + providerThreadRows, + pendingTurnItemRows, + awaitingWakeRows, + sequenceRows, + }; + }), + ); + const background = yield* pendingBackgroundDataByThreadId({ + providerThreadRows, + pendingTurnItemRows, + awaitingWakeRows, + }); const threads = yield* Effect.forEach(threadRows, (row) => shellThreadStateFromRow({ row, @@ -7077,8 +7192,7 @@ export const layer: Layer.Layer = runOrdinalsByThreadId: new Map(), itemCountsByThreadId: new Map(), shellItemCountsByThreadId: new Map(), - providerThreadsByThreadId, - pendingTurnItemsByThreadId, + ...background, }).pipe( Effect.map((state): AgentAwarenessThread => ({ id: state.thread.id, @@ -7190,6 +7304,10 @@ export const layer: Layer.Layer = SELECT 1 FROM json_each(t.payload_json, '$.pullRequests') link WHERE json_type(link.value, '$.watch') IS NOT NULL ) + OR EXISTS ( + SELECT 1 FROM orchestration_v2_projection_subagents task + WHERE task.thread_id = t.thread_id AND task.completion_delivery_state = 'claimed' + ) ) AS may_be_live FROM orchestration_v2_projection_threads t WHERE t.deleted_at IS NULL @@ -7315,6 +7433,7 @@ export const layer: Layer.Layer = shellItemCountRows, providerThreadRows, pendingTurnItemRows, + awaitingWakeRows, ] = yield* Effect.all([ forkSourceIds.length === 0 ? Effect.succeed([] as ReadonlyArray) @@ -7332,14 +7451,18 @@ export const layer: Layer.Layer = : [], ), ), + selectShellAwaitingWakeRows(threadIds), ]); const { runOrdinalsByThreadId, itemCountsByThreadId } = runMapsByThreadId({ runRows, itemCountRows, }); const shellItemCountsByThreadId = shellItemCountMap(shellItemCountRows); - const { providerThreadsByThreadId, pendingTurnItemsByThreadId } = - yield* pendingBackgroundDataByThreadId({ providerThreadRows, pendingTurnItemRows }); + const background = yield* pendingBackgroundDataByThreadId({ + providerThreadRows, + pendingTurnItemRows, + awaitingWakeRows, + }); const states = yield* Effect.forEach([...rowsByThreadId.values()], (row) => shellThreadStateFromRow({ @@ -7347,8 +7470,7 @@ export const layer: Layer.Layer = runOrdinalsByThreadId, itemCountsByThreadId, shellItemCountsByThreadId, - providerThreadsByThreadId, - pendingTurnItemsByThreadId, + ...background, }), ); const statesByThreadId = new Map(states.map((state) => [state.thread.id, state])); diff --git a/apps/server/src/orchestration-v2/ProviderContinuationService.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.ts index 5435130e689c2..f9abcb456c48d 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationService.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationService.ts @@ -6,7 +6,7 @@ import * as Ref from "effect/Ref"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; import * as ThreadManagementService from "./ThreadManagementService.ts"; -import { isUndeliveredMailboxSteer } from "./NotificationMailbox.ts"; +import { pendingDelegatedCompletionWake } from "./NotificationMailbox.ts"; const CONTINUATION_MESSAGE_TEXT = "Background task completed."; @@ -24,21 +24,12 @@ function currentDelegatedCompletionDelivery( >, ) { const sourceRun = projection.runs.find((candidate) => candidate.id === completion.parentRunId); - const delivery = sourceRun?.delegatedCompletion?.delivery; - const alreadyDispatched = projection.messages.some( - (message) => message.id === completion.messageId, - ); - if ( - sourceRun?.delegatedCompletion?.disposition !== "open" || - delivery === null || - delivery === undefined || - delivery.generation !== completion.generation || - delivery.messageId !== completion.messageId || - (alreadyDispatched && !isUndeliveredMailboxSteer(projection, completion.messageId)) - ) { - return undefined; - } - return delivery; + const delivery = + sourceRun === undefined ? undefined : pendingDelegatedCompletionWake(projection, sourceRun); + return delivery?.generation === completion.generation && + delivery.messageId === completion.messageId + ? delivery + : undefined; } function delegatedCompletionRetryKey( diff --git a/packages/shared/src/orchestrationV2PendingBackgroundWork.ts b/packages/shared/src/orchestrationV2PendingBackgroundWork.ts index 6dc635b5153a6..9436317c137b1 100644 --- a/packages/shared/src/orchestrationV2PendingBackgroundWork.ts +++ b/packages/shared/src/orchestrationV2PendingBackgroundWork.ts @@ -234,6 +234,8 @@ export function derivePendingBackgroundWork(input: { /** Excludes rolled-back runs; optional only when the caller already filtered them, as SQL shell reads do. */ readonly runs?: ReadonlyArray; readonly pullRequests?: ReadonlyArray | undefined; + /** Finished delegated task items whose completion wake has not started; they hold like running ones. */ + readonly awaitingWakeTurnItems?: ReadonlyArray; }): ReadonlyArray { const hasActiveRun = input.hasActiveRun ?? @@ -275,7 +277,10 @@ export function derivePendingBackgroundWork(input: { } } - for (const item of pendingBackgroundTurnItems(input)) { + for (const item of [ + ...pendingBackgroundTurnItems(input), + ...(input.awaitingWakeTurnItems ?? []), + ]) { const taskId = nativeTaskIdFromTurnItem(item); if (byTaskId.has(taskId)) { continue; From 5ab6b5f5d41039388ed774b73857ef5c07a08826 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 20:23:50 +0000 Subject: [PATCH 3/6] perf(server): read claimed delegated tasks before their turn items Pin the join order so the awaiting-wake shell read starts from the claimed-task index instead of walking every turn item of each thread. --- apps/server/src/orchestration-v2/ProjectionStore.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 9788e392ad528..50607bcae244f 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6444,7 +6444,8 @@ export const layer: Layer.Layer = INNER JOIN orchestration_v2_projection_runs r ON r.run_id = s.run_id INNER JOIN orchestration_v2_projection_provider_threads p ON p.provider_thread_id = r.provider_thread_id - INNER JOIN orchestration_v2_projection_turn_items item + -- CROSS JOIN pins the claimed-task index first; unpinned, SQLite walks every turn item. + CROSS JOIN orchestration_v2_projection_turn_items item ON item.node_id = s.subagent_id AND item.thread_id = s.thread_id AND item.type = 'subagent' LEFT JOIN orchestration_v2_projection_provider_threads owner ON owner.provider_thread_id = item.provider_thread_id From d58b620b0e3489c38c7634f652777dea377aa1ed Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 20:47:44 +0000 Subject: [PATCH 4/6] fix(server): hold delegated results still waiting for a successor wake A sibling task that finished while its parent's wake run was active stays in the `pending` delivery state until the terminal-run listener reserves it a successor wake. In the gap after that wake run ended, the parent shell read completed with no pending background work, so the web gate could fire a false "Thread completed". Finished app-owned tasks left pending behind an outstanding delivery now hold like claimed ones, in the memory and SQL shell paths alike. They follow the successor reservation's rule: a live parent's blocking wait takes settled_only results, so those do not hold. The four-producer ProjectionStore test now also checks the agent awareness snapshot that feeds mobile pushes. --- .../DelegatedCompletionDelivery.test.ts | 169 ++++++++++++++++++ .../orchestration-v2/NotificationMailbox.ts | 34 +++- .../orchestration-v2/ProjectionStore.test.ts | 90 +++++++--- .../src/orchestration-v2/ProjectionStore.ts | 13 +- 4 files changed, 273 insertions(+), 33 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index d7034f99b32e5..5c816feb81d24 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -1696,6 +1696,175 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( ); }), ); + it.effect( + "keeps the parent shell unsettled while a sibling result waits for its successor wake", + () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const executor = yield* ThreadCommandExecutor.ThreadCommandExecutor; + const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:sibling-wake-shell"); + yield* seedParentWithTerminalTask({ + threadId, + projectId: ProjectId.make("project:sibling-wake-shell"), + runId: RunId.make("run:sibling-wake-shell-seed"), + rootNodeId: NodeId.make("node:sibling-wake-shell-seed"), + taskId: NodeId.make("node:sibling-wake-shell-seed-task"), + deliveryState: "acknowledged", + parentRunStatus: "completed", + now, + }); + const messageId = MessageId.make("message:sibling-wake-shell"); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:sibling-wake-shell-send"), + threadId, + messageId, + text: "Delegate two tasks and end your turn", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + const parentRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.userMessageId === messageId, + )!; + for (const name of ["a", "b"]) { + yield* orchestrator.dispatch({ + type: "delegated_task.request", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make(`command:sibling-wake-shell-delegate-${name}`), + parentThreadId: threadId, + parentRunId: parentRun.id, + parentNodeId: parentRun.rootNodeId!, + task: `Background task ${name}`, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + completionWake: "always", + }); + } + const completeRun = (runThreadId: ThreadId, run: OrchestrationV2Run) => + sink.write({ + events: [ + { + ...runEvent({ + threadId: runThreadId, + runId: run.id, + ordinal: run.ordinal, + status: "completed", + now, + }), + payload: { ...run, status: "completed", completedAt: now }, + }, + ], + }); + const started = yield* orchestrator.getThreadProjection(threadId); + const [taskA, taskB] = started.subagents + .filter((row) => row.runId === parentRun.id && row.childThreadId !== null) + .toSorted((left, right) => left.prompt.localeCompare(right.prompt)); + yield* completeRun( + threadId, + started.runs.find((run) => run.id === parentRun.id)!, + ); + + yield* completeRun( + taskA!.childThreadId!, + (yield* orchestrator.getThreadProjection(taskA!.childThreadId!)).runs[0]!, + ); + let wake = yield* requests.take; + while (wake.threadId !== threadId) wake = yield* requests.take; + const delivery = wake.delegatedCompletion!; + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "server", + commandId: CommandId.make("command:sibling-wake-shell-wake"), + threadId, + messageId: delivery.messageId, + text: "Background task completed.", + attachments: [], + delegatedCompletion: { + parentRunId: parentRun.id, + generation: delivery.generation, + taskIds: [taskA!.id], + }, + dispatchMode: { type: "queue_after_active" }, + }); + + const taskBSequence = yield* sink.latestSequence(); + yield* completeRun( + taskB!.childThreadId!, + (yield* orchestrator.getThreadProjection(taskB!.childThreadId!)).runs[0]!, + ); + yield* sink.stream({ afterSequence: taskBSequence, eventType: "subagent.updated" }).pipe( + Stream.filter( + (stored) => + stored.event.type === "subagent.updated" && + stored.event.payload.id === taskB!.id && + stored.event.payload.status === "completed", + ), + Stream.take(1), + Stream.runHead, + ); + const waking = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + waking.subagents.find((row) => row.id === taskB!.id)?.completionDelivery?.state, + "pending", + ); + const taskBItemId = waking.turnItems.find( + (item) => item.type === "subagent" && item.subagentId === taskB!.id, + )?.id; + + // Holding the parent's lock keeps the terminal-run listener from reserving B's wake. + const held = yield* Deferred.make(); + const release = yield* Deferred.make(); + const lock = yield* executor + .withLock( + threadId, + Deferred.succeed(held, undefined).pipe(Effect.andThen(Deferred.await(release))), + ) + .pipe(Effect.forkScoped({ startImmediately: true })); + yield* Deferred.await(held); + yield* completeRun( + threadId, + waking.runs.find((run) => run.userMessageId === delivery.messageId)!, + ); + const shell = yield* orchestrator.getThreadShell(threadId); + assert.isTrue( + shell?.pendingBackgroundTasks?.some((row) => row.taskId === taskBItemId), + `parent shell reads settled (status ${shell?.status}) before its sibling's successor wake is reserved`, + ); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(lock); + + let successor = yield* requests.take; + while ( + successor.threadId !== threadId || + successor.delegatedCompletion?.messageId === delivery.messageId + ) { + successor = yield* requests.take; + } + const reserved = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + reserved.runs.find((run) => run.id === parentRun.id)?.delegatedCompletion?.delivery, + { + generation: successor.delegatedCompletion!.generation, + messageId: successor.delegatedCompletion!.messageId, + taskIds: [taskB!.id], + }, + ); + assert.isTrue( + (yield* orchestrator.getThreadShell(threadId))?.pendingBackgroundTasks?.some( + (row) => row.taskId === taskBItemId, + ), + ); + }), + ); it.effect("pauses queued follow-up cards and resumes them", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/NotificationMailbox.ts b/apps/server/src/orchestration-v2/NotificationMailbox.ts index 27a58e2f26571..6842f77756bb5 100644 --- a/apps/server/src/orchestration-v2/NotificationMailbox.ts +++ b/apps/server/src/orchestration-v2/NotificationMailbox.ts @@ -3,6 +3,7 @@ import { type MessageId, type OrchestrationV2ConversationMessage, type OrchestrationV2Run, + type OrchestrationV2Subagent, type OrchestrationV2ThreadProjection, type OrchestrationV2TurnItem, } from "@t3tools/contracts"; @@ -53,13 +54,27 @@ export function pendingDelegatedCompletionWake( : delivery; } +const TERMINAL_TASK_STATUSES: ReadonlyArray = [ + "completed", + "failed", + "cancelled", + "interrupted", +]; + /** - * Finished delegated task items whose completion wake has not started its run. A run without its - * provider thread is never offered a wake, so its tasks hold nothing. + * Finished delegated task items whose completion wake has not started its run: claimed into a + * pending wake, or left pending behind a running wake, whose end reserves them a successor. A run + * without its provider thread is never offered a wake, so its tasks hold nothing. */ export function delegatedTaskItemsAwaitingWake( projection: DelegatedCompletionRecords & { readonly providerThreads: ReadonlyArray<{ readonly id: string }>; + readonly subagents: ReadonlyArray< + Pick< + OrchestrationV2Subagent, + "id" | "runId" | "origin" | "status" | "completionWake" | "completionDelivery" + > + >; readonly turnItems: ReadonlyArray; }, ): ReadonlyArray { @@ -74,6 +89,21 @@ export function delegatedTaskItemsAwaitingWake, + ) => + store.apply({ + id: EventId.make(`event:delegated-wake-hold:${name}`), + type: "subagent.updated", threadId, runId: parentRun.id, - parentNodeId: parentRun.rootNodeId, - origin: "app_owned", - createdBy: "agent", + nodeId: taskId, driver, providerInstanceId, - providerThreadId: null, - childThreadId: null, - nativeTaskRef: null, - prompt: "Do the work", - title: "Delegated task", - model: null, - completionWake: "always", - completionDelivery: { state: "claimed", observedByRunId: null }, - status: "completed", - result: "done", - startedAt: now, - completedAt: now, - updatedAt: now, - }, + occurredAt: now, + payload: { + id: taskId, + threadId, + runId: parentRun.id, + parentNodeId: parentRun.rootNodeId, + origin: "app_owned", + createdBy: "agent", + driver, + providerInstanceId, + providerThreadId: null, + childThreadId: null, + nativeTaskRef: null, + prompt: "Do the work", + title: "Delegated task", + model: null, + ...policy, + status: "completed", + result: "done", + startedAt: now, + completedAt: now, + updatedAt: now, + }, + }); + yield* applyTask("task", { + completionWake: "always", + completionDelivery: { state: "claimed", observedByRunId: null }, }); const heldTaskIds = Effect.gen(function* () { yield* assertShellIndexCoversShells(); @@ -1417,6 +1425,7 @@ const delegatedCompletionWakeHoldsEveryShell = Effect.gen(function* () { ProjectionStore.threadShellFromProjection(yield* store.getThreadProjection(threadId)), ), taskIds((yield* store.getSettlementCandidates(threadId))[0]), + taskIds((yield* store.getAgentAwarenessSnapshot(threadId)).threads[0]), ]; for (const held of producers) assert.deepEqual(held, producers[0]); return producers[0]; @@ -1471,6 +1480,31 @@ const delegatedCompletionWakeHoldsEveryShell = Effect.gen(function* () { userMessageId: MessageId.make("message:delegated-wake-hold:steered-into"), }); assert.deepEqual(yield* heldTaskIds, [itemId]); + yield* applyRun("wake-run-read", { ...wakeRun, status: "completed" }); + assert.deepEqual(yield* heldTaskIds, []); + + // A sibling that finished behind that wake waits for the successor its end reserves. + yield* applyTask("task-pending", { + completionWake: "always", + completionDelivery: { state: "pending", observedByRunId: null }, + }); + assert.deepEqual(yield* heldTaskIds, [itemId]); + + // A live parent's blocking wait takes a settled_only result instead. + yield* applyTask("task-settled-only", { + completionWake: "settled_only", + completionDelivery: { state: "pending", observedByRunId: null }, + }); + assert.deepEqual(yield* heldTaskIds, [itemId]); + yield* applyRun("parent-live", { ...parentRun, status: "running", completedAt: null }); + assert.deepEqual(yield* heldTaskIds, []); + + // With no wake outstanding, nothing reserves one. + yield* applyRun("parent-settled", { + ...parentRun, + delegatedCompletion: { disposition: "open", nextGeneration: 2, delivery: null }, + }); + assert.deepEqual(yield* heldTaskIds, []); }); const commandReadsSkipSettledLeafNodes = Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 50607bcae244f..17f234e126f5c 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -997,6 +997,7 @@ type ShellPendingTurnItemRow = PayloadRow & { type ShellAwaitingWakeRow = ShellPendingTurnItemRow & { readonly provider_thread_id: string; + readonly task_payload_json: string; readonly run_payload_json: string; readonly message_payload_json: string | null; readonly message_run_payload_json: string | null; @@ -6425,10 +6426,11 @@ export const layer: Layer.Layer = } `; - // Claimed tasks are rare and indexed; the run, message and turn joins only run for them. + // Finished tasks owed a wake are rare and indexed; the run, message and turn joins only run for them. const selectShellAwaitingWakeRows = (threadIds: ReadonlyArray) => sql` SELECT s.thread_id, p.provider_thread_id, item.payload_json, + s.payload_json AS task_payload_json, owner.provider_instance_id AS owning_provider_instance_id, r.payload_json AS run_payload_json, m.payload_json AS message_payload_json, @@ -6453,7 +6455,8 @@ export const layer: Layer.Layer = ON m.message_id = json_extract(r.payload_json, '$.delegatedCompletion.delivery.messageId') LEFT JOIN orchestration_v2_projection_runs mr ON mr.run_id = m.run_id WHERE s.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) - AND s.completion_delivery_state = 'claimed' + AND s.completion_delivery_state IN ('claimed', 'pending') + AND json_type(r.payload_json, '$.delegatedCompletion.delivery') = 'object' `; const runMapsByThreadId = (input: { @@ -6557,6 +6560,7 @@ export const layer: Layer.Layer = messages: message === null ? [] : [message], providerTurns: messageTurn === null ? [] : [messageTurn], providerThreads: [{ id: row.provider_thread_id }], + subagents: [yield* decodeSubagentPayload(row.task_payload_json)], turnItems: [yield* decodeShellTurnItem(row)], }); if (items.length === 0) continue; @@ -7307,7 +7311,10 @@ export const layer: Layer.Layer = ) OR EXISTS ( SELECT 1 FROM orchestration_v2_projection_subagents task - WHERE task.thread_id = t.thread_id AND task.completion_delivery_state = 'claimed' + INNER JOIN orchestration_v2_projection_runs task_run ON task_run.run_id = task.run_id + WHERE task.thread_id = t.thread_id + AND task.completion_delivery_state IN ('claimed', 'pending') + AND json_type(task_run.payload_json, '$.delegatedCompletion.delivery') = 'object' ) ) AS may_be_live FROM orchestration_v2_projection_threads t From 597ddb143b08ea51e3c1b757e4953db8e055d2c0 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 21:09:26 +0000 Subject: [PATCH 5/6] fix(server): held delegated results do not keep the caller's MCP session live A finished delegated task held until its completion wake starts has no provider process behind it, so it is listed without an owning provider instance and isLiveCallerShell keeps rejecting a parent whose run ended. --- packages/shared/src/orchestrationV2PendingBackgroundWork.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/shared/src/orchestrationV2PendingBackgroundWork.ts b/packages/shared/src/orchestrationV2PendingBackgroundWork.ts index 9436317c137b1..23fe250d97a62 100644 --- a/packages/shared/src/orchestrationV2PendingBackgroundWork.ts +++ b/packages/shared/src/orchestrationV2PendingBackgroundWork.ts @@ -279,7 +279,10 @@ export function derivePendingBackgroundWork(input: { for (const item of [ ...pendingBackgroundTurnItems(input), - ...(input.awaitingWakeTurnItems ?? []), + // A finished task has no provider process, so it must not keep the caller's MCP session live. + ...(input.awaitingWakeTurnItems ?? []).map( + ({ owningProviderInstanceId: _owner, ...item }) => item, + ), ]) { const taskId = nativeTaskIdFromTurnItem(item); if (byTaskId.has(taskId)) { From 1ece4c028c2377c34511cdb8f8b548ff8662f3d4 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 22:59:37 +0000 Subject: [PATCH 6/6] test(server): drop the sibling successor-wake test the projection table already covers AI-assisted; verified with the tests and tools listed above. --- .../DelegatedCompletionDelivery.test.ts | 169 ------------------ 1 file changed, 169 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index c0f0023c8202b..4c2de8698ac9d 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -2355,175 +2355,6 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( ); }), ); - it.effect( - "keeps the parent shell unsettled while a sibling result waits for its successor wake", - () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const sink = yield* EventSink.EventSinkV2; - const executor = yield* ThreadCommandExecutor.ThreadCommandExecutor; - const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; - const now = yield* DateTime.now; - const threadId = ThreadId.make("thread:sibling-wake-shell"); - yield* seedParentWithTerminalTask({ - threadId, - projectId: ProjectId.make("project:sibling-wake-shell"), - runId: RunId.make("run:sibling-wake-shell-seed"), - rootNodeId: NodeId.make("node:sibling-wake-shell-seed"), - taskId: NodeId.make("node:sibling-wake-shell-seed-task"), - deliveryState: "acknowledged", - parentRunStatus: "completed", - now, - }); - const messageId = MessageId.make("message:sibling-wake-shell"); - yield* orchestrator.dispatch({ - type: "message.dispatch", - createdBy: "user", - creationSource: "web", - commandId: CommandId.make("command:sibling-wake-shell-send"), - threadId, - messageId, - text: "Delegate two tasks and end your turn", - attachments: [], - modelSelection, - dispatchMode: { type: "start_immediately" }, - }); - const parentRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( - (run) => run.userMessageId === messageId, - )!; - for (const name of ["a", "b"]) { - yield* orchestrator.dispatch({ - type: "delegated_task.request", - createdBy: "agent", - creationSource: "mcp", - commandId: CommandId.make(`command:sibling-wake-shell-delegate-${name}`), - parentThreadId: threadId, - parentRunId: parentRun.id, - parentNodeId: parentRun.rootNodeId!, - task: `Background task ${name}`, - modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - completionWake: "always", - }); - } - const completeRun = (runThreadId: ThreadId, run: OrchestrationV2Run) => - sink.write({ - events: [ - { - ...runEvent({ - threadId: runThreadId, - runId: run.id, - ordinal: run.ordinal, - status: "completed", - now, - }), - payload: { ...run, status: "completed", completedAt: now }, - }, - ], - }); - const started = yield* orchestrator.getThreadProjection(threadId); - const [taskA, taskB] = started.subagents - .filter((row) => row.runId === parentRun.id && row.childThreadId !== null) - .toSorted((left, right) => left.prompt.localeCompare(right.prompt)); - yield* completeRun( - threadId, - started.runs.find((run) => run.id === parentRun.id)!, - ); - - yield* completeRun( - taskA!.childThreadId!, - (yield* orchestrator.getThreadProjection(taskA!.childThreadId!)).runs[0]!, - ); - let wake = yield* requests.take; - while (wake.threadId !== threadId) wake = yield* requests.take; - const delivery = wake.delegatedCompletion!; - yield* orchestrator.dispatch({ - type: "message.dispatch", - createdBy: "agent", - creationSource: "server", - commandId: CommandId.make("command:sibling-wake-shell-wake"), - threadId, - messageId: delivery.messageId, - text: "Background task completed.", - attachments: [], - delegatedCompletion: { - parentRunId: parentRun.id, - generation: delivery.generation, - taskIds: [taskA!.id], - }, - dispatchMode: { type: "queue_after_active" }, - }); - - const taskBSequence = yield* sink.latestSequence(); - yield* completeRun( - taskB!.childThreadId!, - (yield* orchestrator.getThreadProjection(taskB!.childThreadId!)).runs[0]!, - ); - yield* sink.stream({ afterSequence: taskBSequence, eventType: "subagent.updated" }).pipe( - Stream.filter( - (stored) => - stored.event.type === "subagent.updated" && - stored.event.payload.id === taskB!.id && - stored.event.payload.status === "completed", - ), - Stream.take(1), - Stream.runHead, - ); - const waking = yield* orchestrator.getThreadProjection(threadId); - assert.equal( - waking.subagents.find((row) => row.id === taskB!.id)?.completionDelivery?.state, - "pending", - ); - const taskBItemId = waking.turnItems.find( - (item) => item.type === "subagent" && item.subagentId === taskB!.id, - )?.id; - - // Holding the parent's lock keeps the terminal-run listener from reserving B's wake. - const held = yield* Deferred.make(); - const release = yield* Deferred.make(); - const lock = yield* executor - .withLock( - threadId, - Deferred.succeed(held, undefined).pipe(Effect.andThen(Deferred.await(release))), - ) - .pipe(Effect.forkScoped({ startImmediately: true })); - yield* Deferred.await(held); - yield* completeRun( - threadId, - waking.runs.find((run) => run.userMessageId === delivery.messageId)!, - ); - const shell = yield* orchestrator.getThreadShell(threadId); - assert.isTrue( - shell?.pendingBackgroundTasks?.some((row) => row.taskId === taskBItemId), - `parent shell reads settled (status ${shell?.status}) before its sibling's successor wake is reserved`, - ); - yield* Deferred.succeed(release, undefined); - yield* Fiber.join(lock); - - let successor = yield* requests.take; - while ( - successor.threadId !== threadId || - successor.delegatedCompletion?.messageId === delivery.messageId - ) { - successor = yield* requests.take; - } - const reserved = yield* orchestrator.getThreadProjection(threadId); - assert.deepEqual( - reserved.runs.find((run) => run.id === parentRun.id)?.delegatedCompletion?.delivery, - { - generation: successor.delegatedCompletion!.generation, - messageId: successor.delegatedCompletion!.messageId, - taskIds: [taskB!.id], - }, - ); - assert.isTrue( - (yield* orchestrator.getThreadShell(threadId))?.pendingBackgroundTasks?.some( - (row) => row.taskId === taskBItemId, - ), - ); - }), - ); it.effect("pauses queued follow-up cards and resumes them", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2;