From 8fe7b265684387d4b0e1f427e5f1af8ffc86c2cf Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Mon, 2 Nov 2026 19:21:18 -0800 Subject: [PATCH] fix(server): subagent questions reach a busy parent agent A delegated task's question woke its parent with a queued message. A parent whose turn never ends (an orchestrator kept alive by child messages) never saw it, so the questions sat unanswered or landed on the user. Question notices now steer into the running parent turn, the same way background task completions already do. Retries stay queued and keep the user's saved model selection. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/EffectWorker.ts | 9 +- .../src/orchestration-v2/Orchestrator.ts | 25 ++-- .../SteeringCompletion.integration.test.ts | 115 ++++++++++++++++-- .../orchestrator-mcp-server.md | 8 +- 4 files changed, 132 insertions(+), 25 deletions(-) diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index 534043aeb4eb8..c5b5ea9b383a9 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -278,6 +278,10 @@ export const layerExecutor: Layer.Layer< const message = projection.messages.find((item) => item.id === messageId); const run = projection.runs.find((item) => item.id === message?.runId); if (message === undefined || run === undefined) return yield* error; + // Notifications must be queued messages. + const queued = + message.delegatedCompletion !== undefined || + message.notification !== undefined; // Reuse message ids and receipts so retries cannot duplicate messages or follow-ups. yield* threads .dispatch({ @@ -292,10 +296,7 @@ export const layerExecutor: Layer.Layer< ? {} : { modelSelection: run.modelSelection }), dispatchMode: { - type: - message.delegatedCompletion === undefined - ? "start_immediately" - : "queue_after_active", + type: queued ? "queue_after_active" : "start_immediately", }, createdBy: message.createdBy, creationSource: message.creationSource, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 3dc0a65ae03a3..600764704e57f 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -4685,6 +4685,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio readonly senderThreadId?: OrchestrationV2ConversationMessage["senderThreadId"]; readonly idleCompaction?: OrchestrationV2ConversationMessage["idleCompaction"]; readonly delegatedCompletion?: OrchestrationV2ConversationMessage["delegatedCompletion"]; + readonly notification?: OrchestrationV2ConversationMessage["notification"]; readonly forceRestart: boolean; }) { const targetRun = input.projection.runs.find((candidate) => candidate.id === input.targetRunId); @@ -4838,6 +4839,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(input.delegatedCompletion === undefined ? {} : { delegatedCompletion: input.delegatedCompletion }), + ...(input.notification === undefined ? {} : { notification: input.notification }), ...(input.scheduledTaskId === undefined ? {} : { scheduledTaskId: input.scheduledTaskId }), @@ -4939,6 +4941,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio input.projection.thread.providerInstanceId !== input.modelSelection.instanceId; if ( input.delegatedCompletion === undefined && + input.notification === undefined && (instanceChanged || !modelSelectionsEqual(input.projection.thread.modelSelection, input.modelSelection)) ) { @@ -5729,13 +5732,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio taskIds: delivery.taskIds, }; } - // Route durable mailbox deliveries under the thread lock, using the live - // session's capabilities. Never interrupt/restart a turn for a notification. + // Route durable mailbox deliveries and delegated task questions under the + // thread lock, using the live session's capabilities. A question queued + // behind a long parent turn is never seen. Never interrupt/restart a turn + // for a notification. if ( - delegatedCompletion !== undefined && - delegatedCompletion.taskIds.every( - (id) => projection.subagents.find((task) => task.id === id)?.completionWake === "always", - ) + delegatedCompletion !== undefined + ? delegatedCompletion.taskIds.every( + (id) => + projection.subagents.find((task) => task.id === id)?.completionWake === "always", + ) + : command.notification?.source.kind === "delegated_task" ) { const active = projection.runs.find((run) => run.status === "running"); const providerThread = projection.providerThreads.find( @@ -5759,7 +5766,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if ( Option.isSome(session) && session.value.providerSession.capabilities.turns.supportsActiveSteering && - session.value.providerSession.capabilities.turns.activeSteeringInterruptsTools !== true + session.value.providerSession.capabilities.turns.activeSteeringInterruptsTools !== true && + !(yield* stopReachedRun(command, command.threadId, active.id)) ) { dispatchMode = { type: "steer_active", targetRunId: active.id }; } @@ -5826,11 +5834,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio effects, projection, modelSelection: - delegatedCompletion === undefined + delegatedCompletion === undefined && command.notification === undefined ? modelSelection : (projection.runs.find((run) => run.id === dispatchMode.targetRunId)?.modelSelection ?? modelSelection), delegatedCompletion, + ...(command.notification === undefined ? {} : { notification: command.notification }), targetRunId: dispatchMode.targetRunId, messageId: command.messageId, text: dispatchText, diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index e570c3ad81055..bd3f4e5b2e7d8 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -44,13 +44,16 @@ it.effect.each( [true, "with interrupting native steering"], [true, "settled only"], [false, "during continuation"], + ["question", "during delivery"], ] as const - ).map(([mailbox, timing]) => ({ - mailbox, + ).map(([kind, timing]) => ({ + mailbox: kind === true, + question: kind === "question", timing, - label: mailbox ? "mailbox notification" : "steering", + label: + kind === "question" ? "delegated task question" : kind ? "mailbox notification" : "steering", })), -)("delivers $label when completion wins $timing", ({ mailbox, timing }) => +)("delivers $label when completion wins $timing", ({ mailbox, question, timing }) => Effect.scoped( Effect.gen(function* () { const cwd = yield* checkpointWorkspace(`steering-completion-${timing.replaceAll(" ", "-")}`); @@ -255,6 +258,21 @@ it.effect.each( ], }); } + if (question) { + const { thread } = yield* orchestrator.getThreadProjection(threadId); + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make("event:composer-selection"), + type: "thread.model-selection-updated", + threadId, + providerInstanceId: instanceId, + occurredAt: yield* DateTime.now, + payload: { ...thread, modelSelection: composerSelection }, + }, + ], + }); + } const dispatchSteer = orchestrator.dispatch({ type: "message.dispatch", commandId: CommandId.make("steer"), @@ -270,11 +288,21 @@ it.effect.each( sizeBytes: 10, }, ], - dispatchMode: mailbox - ? { type: "queue_after_active" } - : { type: "steer_active", targetRunId: first.runId }, - createdBy: mailbox ? "agent" : "user", - creationSource: mailbox ? "server" : "web", + dispatchMode: + mailbox || question + ? { type: "queue_after_active" } + : { type: "steer_active", targetRunId: first.runId }, + createdBy: mailbox || question ? "agent" : "user", + creationSource: mailbox || question ? "server" : "web", + ...(question + ? { + notification: { + source: { kind: "delegated_task", taskIds: [taskId] }, + outcome: "updated", + summary: "Delegated task has a question", + }, + } + : {}), ...(mailbox ? { delegatedCompletion: { @@ -340,6 +368,8 @@ it.effect.each( if (delivery !== null) { yield* Deferred.succeed(rejectSteer, undefined); yield* Fiber.join(delivery); + } + if (delivery !== null && !question) { const requeued = yield* orchestrator.getThreadProjection(threadId); const item = requeued.turnItems.find( (item) => item.type === "user_message" && item.messageId === messageId, @@ -399,7 +429,11 @@ it.effect.each( ]); assert.equal(steerCalls, timing === "during delivery" ? 1 : 0); const final = yield* orchestrator.getThreadProjection(threadId); - if (!mailbox) { + if (question) { + assert.deepEqual(final.thread.modelSelection, composerSelection); + assert.deepEqual(started[1]?.modelSelection, composerSelection); + } + if (!mailbox && !question) { assert.equal( final.turnItems.find( (item) => item.type === "user_message" && item.messageId === messageId, @@ -414,7 +448,7 @@ it.effect.each( ); assert.equal( final.turnItems.filter((item) => - mailbox + mailbox || question ? item.type === "notification" : item.type === "user_message" && item.messageId === messageId, ).length, @@ -617,6 +651,65 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* return { cwd, events, started, steered, interrupted, layer, startFirstTurn }; }); +it.effect("steers a delegated task question notice into the running parent turn", () => + Effect.scoped( + Effect.gen(function* () { + const { started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( + "delegated-question-steer", + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + const { thread } = yield* orchestrator.getThreadProjection(threadId); + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make("event:composer-selection"), + type: "thread.model-selection-updated", + threadId, + providerInstanceId: instanceId, + occurredAt: yield* DateTime.now, + payload: { ...thread, modelSelection: composerSelection }, + }, + ], + }); + const messageId = MessageId.make("message:delegated-task-question:child-question"); + const notification = { + source: { + kind: "delegated_task" as const, + taskIds: [NodeId.make("task:question")], + childThreadId: ThreadId.make("thread:question-child"), + }, + outcome: "updated" as const, + summary: "Delegated task has a question", + }; + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("question-wake"), + threadId, + messageId, + text: "Delegated task asked a question", + notification, + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "server", + }); + yield* worker.drain(); + const after = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual(steered, ["Delegated task asked a question"]); + assert.lengthOf(started, 1); + assert.isFalse(after.runs.some((run) => run.status === "queued")); + const message = after.messages.find((candidate) => candidate.id === messageId); + assert.equal(message?.runId, started[0]?.runId); + assert.deepEqual(message?.notification, notification); + assert.deepEqual(after.thread.modelSelection, composerSelection); + }).pipe(Effect.provide(layer)); + }), + ), +); + it.effect("sends a selected held message once and preserves the remaining queue order", () => Effect.scoped( Effect.gen(function* () { diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 915fa59ab22d0..41146da1ed4c1 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -515,8 +515,12 @@ Users rarely watch child threads, so a child question would otherwise go unanswered. When a child opens a pending `user_input` request, the parent gets a notification message telling it to answer with the pending request tools or ask the user. The message says whether the child is blocked or, for a Codex -async question, still working. If the question is answered or dismissed before -the notification starts a turn, that queued turn is cancelled. Approvals are not +async question, still working. If the parent's turn is running and its provider +can steer without interrupting tools, the notification steers into that turn, +like a background task's completion. A queued notice would wait behind a +long-running orchestrator turn that may never end. Otherwise it queues. If the +question is answered or dismissed before a queued notification starts a turn, +that queued turn is cancelled. Approvals are not routed, because those tools cannot answer them. A parent still blocked in a `delegate_task` wait gets the blocking question through that wait instead.