Skip to content

fix(server): a delegated child's answer after its monitor ends reaches the parent #31856

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
juliusmarminge wants to merge 4 commits into from
+148 −128
Closed

fix(server): a delegated child's answer after its monitor ends reaches the parent #31856

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1dc0530
test(server): reproduce #31843 monitor-end wake not delivered to parent
juliusmarminge Nov 6, 2026
405e1db
fix(server): a delegated child's monitor ending wakes its parent
juliusmarminge Nov 6, 2026
a99108b
Merge remote-tracking branch 'origin/main' into fix/issue-31843
juliusmarminge Nov 6, 2026
a61fc32
Merge remote-tracking branch 'origin/main' into fix/issue-31843
juliusmarminge Nov 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
269 changes: 143 additions & 126 deletions apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2366,136 +2366,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 = [
Expand Down
7 changes: 5 additions & 2 deletions apps/server/src/orchestration-v2/Orchestrator.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -699,14 +699,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<OrchestrationV2Command, { readonly type: "message.dispatch" }>,
): 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" ||
Expand Down
Loading