From 1dc0530d966d5d1a3b933f78cc767e9ead363121 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 6 Nov 2026 06:43:38 +0000 Subject: [PATCH 1/2] test(server): reproduce #31843 monitor-end wake not delivered to parent --- .../DelegatedCompletionDelivery.test.ts | 269 ++++++++++-------- 1 file changed, 143 insertions(+), 126 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index db3a2759e3ac5..8aa2df3087eff 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -1962,136 +1962,153 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( }), ); - it.effect("delivers the run a background command's wake starts after the original result", () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const eventSink = yield* EventSink.EventSinkV2; - const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; - const now = yield* DateTime.now; - const parentThreadId = ThreadId.make("thread:background-wake-parent"); - const projectId = ProjectId.make("project:background-wake"); - const parentRunId = RunId.make("run:background-wake-parent"); - const rootNodeId = NodeId.make("node:background-wake-parent"); - yield* seedParentWithTerminalTask({ - threadId: parentThreadId, - projectId, - runId: parentRunId, - rootNodeId, - taskId: NodeId.make("node:background-wake-seed"), - deliveryState: "acknowledged", - now, - }); - const child = yield* seedRestartCancelledChild({ - parentThreadId, - projectId, - parentRunId, - rootNodeId, - name: "background-wake-child", - completionWake: "always", - continuationPending: false, - runStatus: "completed", - now, - }); - yield* orchestrator.recoverDelegatedTasks; - let firstWake = yield* requests.take; - while (firstWake.threadId !== parentThreadId) firstWake = yield* requests.take; - yield* orchestrator.dispatch({ - type: "delegated_task.completion-delivery.acknowledge", - commandId: CommandId.make("command:background-wake-ack"), - parentThreadId, - taskId: child.taskId, - observedByRunId: parentRunId, - }); - const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.find( - (run) => run.id === parentRunId, - )!; - yield* eventSink.write({ - commandId: reconcileCommandId("background-wake-parent-finished"), - events: [ - { - id: EventId.make("event:background-wake-parent-finished"), - type: "run.updated", - threadId: parentThreadId, - runId: parentRunId, - occurredAt: now, - payload: { ...parentRun, status: "completed", completedAt: now }, - }, - ], - }); + const wakeSources = [ + { + id: "background", + name: "background command", + source: { kind: "command" }, + text: "Background activity updated", + summary: 'Command "sleep 60" finished', + }, + { + id: "monitor-end", + name: "monitor", + source: { kind: "monitor" }, + text: 'Monitor "sleep 60" stream ended', + summary: 'Monitor "sleep 60" finished', + }, + ] as const; + it.effect.each(wakeSources)( + "delivers the run a $name's ending wake starts after the original result", + (source) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; + const now = yield* DateTime.now; + const parentThreadId = ThreadId.make(`thread:${source.id}-wake-parent`); + const projectId = ProjectId.make(`project:${source.id}-wake`); + const parentRunId = RunId.make(`run:${source.id}-wake-parent`); + const rootNodeId = NodeId.make(`node:${source.id}-wake-parent`); + yield* seedParentWithTerminalTask({ + threadId: parentThreadId, + projectId, + runId: parentRunId, + rootNodeId, + taskId: NodeId.make(`node:${source.id}-wake-seed`), + deliveryState: "acknowledged", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId, + projectId, + parentRunId, + rootNodeId, + name: `${source.id}-wake-child`, + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + yield* orchestrator.recoverDelegatedTasks; + let firstWake = yield* requests.take; + while (firstWake.threadId !== parentThreadId) firstWake = yield* requests.take; + yield* orchestrator.dispatch({ + type: "delegated_task.completion-delivery.acknowledge", + commandId: CommandId.make(`command:${source.id}-wake-ack`), + parentThreadId, + taskId: child.taskId, + observedByRunId: parentRunId, + }); + const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.find( + (run) => run.id === parentRunId, + )!; + yield* eventSink.write({ + commandId: reconcileCommandId(`${source.id}-wake-parent-finished`), + events: [ + { + id: EventId.make(`event:${source.id}-wake-parent-finished`), + type: "run.updated", + threadId: parentThreadId, + runId: parentRunId, + occurredAt: now, + payload: { ...parentRun, status: "completed", completedAt: now }, + }, + ], + }); - // The shape ProviderContinuationService dispatches when the provider - // reports a background command (wire kind `background_command`) ended. - const wakeMessageId = MessageId.make("message:background-wake"); - yield* orchestrator.dispatch({ - type: "message.dispatch", - createdBy: "agent", - creationSource: "provider", - commandId: CommandId.make("command:background-wake"), - threadId: child.childThreadId, - messageId: wakeMessageId, - text: "Background activity updated", - notification: { - source: { kind: "command" }, - outcome: "completed", - summary: 'Command "sleep 60" finished', - }, - attachments: [], - dispatchMode: { type: "queue_after_active" }, - }); - const run = (yield* orchestrator.getThreadProjection(child.childThreadId)).runs.find( - (row) => row.userMessageId === wakeMessageId, - )!; - assert.isDefined(run.delegatedTaskId); - yield* eventSink.write({ - commandId: reconcileCommandId("background-wake-finished"), - events: [ - { - id: EventId.make("event:background-wake-result"), - type: "message.updated", - threadId: child.childThreadId, - runId: run.id, - occurredAt: now, - payload: { - id: MessageId.make("message:background-wake-result"), + // The shape ProviderContinuationService dispatches when the provider reports + // a background command (wire kind `background_command`) or monitor ended. + const wakeMessageId = MessageId.make(`message:${source.id}-wake`); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "provider", + commandId: CommandId.make(`command:${source.id}-wake`), + threadId: child.childThreadId, + messageId: wakeMessageId, + text: source.text, + notification: { source: source.source, outcome: "completed", summary: source.summary }, + attachments: [], + dispatchMode: { type: "queue_after_active" }, + }); + const run = (yield* orchestrator.getThreadProjection(child.childThreadId)).runs.find( + (row) => row.userMessageId === wakeMessageId, + )!; + assert.isDefined( + run.delegatedTaskId, + `the run woken by the child's ${source.name} ending should be tracked as a follow-up delegated task for the parent`, + ); + yield* eventSink.write({ + commandId: reconcileCommandId(`${source.id}-wake-finished`), + events: [ + { + id: EventId.make(`event:${source.id}-wake-result`), + type: "message.updated", threadId: child.childThreadId, runId: run.id, - nodeId: null, - role: "assistant", - text: "The background command finished cleanly.", - attachments: [], - streaming: false, - createdBy: "agent", - creationSource: "server", - createdAt: now, - updatedAt: now, + occurredAt: now, + payload: { + id: MessageId.make(`message:${source.id}-wake-result`), + threadId: child.childThreadId, + runId: run.id, + nodeId: null, + role: "assistant", + text: "The background command finished cleanly.", + attachments: [], + streaming: false, + createdBy: "agent", + creationSource: "server", + createdAt: now, + updatedAt: now, + }, }, - }, - { - id: EventId.make("event:background-wake-finished"), - type: "run.updated", - threadId: child.childThreadId, - runId: run.id, - occurredAt: now, - payload: { ...run, status: "completed", startedAt: now, completedAt: now }, - }, - ], - }); - yield* orchestrator.recoverDelegatedTasks; - const followup = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( - (task) => task.id === run.delegatedTaskId, - )!; - assert.equal(followup.status, "completed"); - assert.equal(followup.result, "The background command finished cleanly."); - assert.equal(followup.completionDelivery?.state, "claimed"); - let wake = yield* requests.take; - while ( - wake.threadId !== parentThreadId || - wake.delegatedCompletion?.messageId === firstWake.delegatedCompletion?.messageId - ) { - wake = yield* requests.take; - } - assert.equal(wake.delegatedCompletion?.parentRunId, parentRunId); - }), + { + id: EventId.make(`event:${source.id}-wake-finished`), + type: "run.updated", + threadId: child.childThreadId, + runId: run.id, + occurredAt: now, + payload: { ...run, status: "completed", startedAt: now, completedAt: now }, + }, + ], + }); + yield* orchestrator.recoverDelegatedTasks; + const followup = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (task) => task.id === run.delegatedTaskId, + )!; + assert.equal(followup.status, "completed"); + assert.equal(followup.result, "The background command finished cleanly."); + assert.equal(followup.completionDelivery?.state, "claimed"); + let wake = yield* requests.take; + while ( + wake.threadId !== parentThreadId || + wake.delegatedCompletion?.messageId === firstWake.delegatedCompletion?.messageId + ) { + wake = yield* requests.take; + } + assert.equal(wake.delegatedCompletion?.parentRunId, parentRunId); + }), ); const outcomes = [ From 405e1dbb573356312bec8727f0a9203b399a4ede Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 6 Nov 2026 06:48:14 +0000 Subject: [PATCH 2/2] fix(server): a delegated child's monitor ending wakes its parent A delegated child woken by a monitor's terminal notification (completed, failed, cancelled, unknown) answered without starting a subagent follow-up, so its new result never reached the parent. Monitor notifications now gate on outcome: only "updated" progress events skip the follow-up, matching how a background command's ending wake works. AI-assisted; verified with the tests and tools listed above. --- apps/server/src/orchestration-v2/Orchestrator.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index bcec0a13cf20a..cea7fdb8a3958 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -695,14 +695,17 @@ function cancelledAppOwnedTask( }; } -// Provider wakes can start follow-up work; monitor updates only report progress. +// Provider wakes and monitor ends can start follow-up work; monitor updates only report progress. // Dispatch takes the parent's lock for follow-ups because intake writes the parent. function mayStartSubagentFollowup( command: Extract, ): boolean { if (command.scheduledTaskId !== undefined || command.restartContinuationOfRunId !== undefined) return false; - if (command.notification !== undefined) return command.notification.source.kind !== "monitor"; + if (command.notification !== undefined) + return ( + command.notification.source.kind !== "monitor" || command.notification.outcome !== "updated" + ); return ( command.createdBy === "user" || command.creationSource === "mcp" ||