authorgravatar for 97427849+flamboh@users.noreply.github.comoliver <97427849+flamboh@users.noreply.github.com> 2026-10-14 20:17:08-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-15 03:17:08+00:00
log924fb0c4d4941f9d186e46030ab358ebbdb064f8
treead45a14f30b13acea93af6042e5f158c48b02f22
parent755accf590bec6e81d1b2484d8501c95798c1cbe
signature Signed by PGP key B5690EEEBB952194

fix(server): a delegated task's question ends the parent's blocking wait (#19538)

Co-authored-by: maria-rcks <maria@kuuro.net>

9 files changed, 119 insertions(+), 14 deletions(-)

apps/server/src/mcp/OrchestratorMcpService.test.ts+9
......@@ -96,6 +96,7 @@ describe("OrchestratorMcpService", () => {
9696 messages: [],
9797 subagents: [],
9898 providerThreads: [],
99 runtimeRequests: [],
99100 } as unknown as OrchestrationV2ThreadProjection;
100101 let hasNestedWork = true;
101102 const layerDependencies = Layer.mergeAll(
......@@ -223,6 +224,7 @@ describe("OrchestratorMcpService", () => {
223224 messages: [],
224225 subagents: [],
225226 providerThreads: [],
227 runtimeRequests: [],
226228 turnItems: [],
227229 } as unknown as OrchestrationV2ThreadProjection;
228230 const layerDependencies = Layer.mergeAll(
......@@ -308,6 +310,7 @@ describe("OrchestratorMcpService", () => {
308310 messages: [],
309311 subagents: [],
310312 providerThreads: [],
313 runtimeRequests: [],
311314 } as unknown as OrchestrationV2ThreadProjection;
312315 const layerDependencies = Layer.mergeAll(
313316 NodeServices.layer,
......@@ -385,6 +388,7 @@ describe("OrchestratorMcpService", () => {
385388 messages: [],
386389 subagents: [],
387390 providerThreads: [],
391 runtimeRequests: [],
388392 } as unknown as OrchestrationV2ThreadProjection;
389393 const layerDependencies = Layer.mergeAll(
390394 NodeServices.layer,
......@@ -465,6 +469,7 @@ describe("OrchestratorMcpService", () => {
465469 messages: [],
466470 subagents: [],
467471 providerThreads: [],
472 runtimeRequests: [],
468473 } as unknown as OrchestrationV2ThreadProjection;
469474 const layerDependencies = Layer.mergeAll(
470475 NodeServices.layer,
......@@ -552,6 +557,7 @@ describe("OrchestratorMcpService", () => {
552557 messages: [],
553558 subagents: [],
554559 providerThreads: [],
560 runtimeRequests: [],
555561 } as unknown as OrchestrationV2ThreadProjection;
556562 const layerDependencies = Layer.mergeAll(
557563 NodeServices.layer,
......@@ -655,6 +661,7 @@ describe("OrchestratorMcpService", () => {
655661 appOwnedTask("node:mcp-cancel-grandchild-below", childThreadId, grandchildThreadId),
656662 ],
657663 providerThreads: [],
664 runtimeRequests: [],
658665 },
659666 ],
660667 [
......@@ -666,6 +673,7 @@ describe("OrchestratorMcpService", () => {
666673 messages: [],
667674 subagents: [],
668675 providerThreads: [],
676 runtimeRequests: [],
669677 },
670678 ],
671679 ]) as unknown as ReadonlyMap<ThreadId, OrchestrationV2ThreadProjection>;
......@@ -883,6 +891,7 @@ describe("OrchestratorMcpService provider resolution", () => {
883891 messages: [],
884892 subagents: [],
885893 providerThreads: [],
894 runtimeRequests: [],
886895 turnItems: [],
887896 } as unknown as OrchestrationV2ThreadProjection;
888897
apps/server/src/mcp/OrchestratorMcpService.ts+30-12
......@@ -101,13 +101,14 @@ import * as SecretRequests from "../secrets/SecretRequests.ts";
101101
102102const DEFAULT_WAIT_TIMEOUT_MS = 10 * 60 * 1_000;
103103const MAX_WAIT_TIMEOUT_MS = 60 * 60 * 1_000;
104// Events that can make a delegated task terminal: the parent's task record,
105// and the child's runs, nested tasks, and pending provider background work.
104// Events that can end a delegated task wait: the parent's task record,
105// and the child's runs, nested tasks, provider background work, and questions.
106106const TASK_WAKE_EVENTS = [
107107 { thread: "parent", eventType: "subagent.updated" },
108108 { thread: "child", eventType: "run.updated" },
109109 { thread: "child", eventType: "subagent.updated" },
110110 { thread: "child", eventType: "provider-thread.updated" },
111 { thread: "child", eventType: "runtime-request.updated" },
111112] as const;
112113/** A person answers the card, so a slower poll is plenty. */
113114const SECRET_REQUEST_POLL_INTERVAL_MS = 500;
......@@ -1351,7 +1352,14 @@ const make = Effect.gen(function* () {
13511352 const childControls = yield* threadManagement
13521353 .getThreadRecords(
13531354 task.childThreadId,
1354 ["runs", "messages", "contextTransfers", "subagents", "providerThreads"],
1355 [
1356 "runs",
1357 "messages",
1358 "contextTransfers",
1359 "subagents",
1360 "providerThreads",
1361 "runtimeRequests",
1362 ],
13551363 { messageRoles: ["user"] },
13561364 )
13571365 .pipe(Effect.mapError(threadManagementFailure));
......@@ -1456,6 +1464,15 @@ const make = Effect.gen(function* () {
14561464 : subagentResultForRun(childProjection, terminalRun).text
14571465 : null,
14581466 latestTerminalResultContextTransferId: resultTransferForRun(terminalRun)?.id ?? null,
1467 // A finished task has no question of its own; later child runs belong to other tasks.
1468 hasPendingQuestion:
1469 !isTerminalTaskStatus(status) &&
1470 childControls.runtimeRequests.some(
1471 (request) =>
1472 request.kind === "user_input" &&
1473 request.status === "pending" &&
1474 request.responseCapability.type === "live",
1475 ),
14591476 waitTimedOut,
14601477 } satisfies OrchestratorMcpDelegateTaskResult;
14611478 if (
......@@ -1508,7 +1525,9 @@ const make = Effect.gen(function* () {
15081525 .getThreadEventSequence(scope.thread.threadId)
15091526 .pipe(Effect.mapError(streamError));
15101527 const initial = yield* readTask(scope, taskId, false, true);
1511 if (isTerminalTaskStatus(initial.status)) return Option.some(initial);
1528 if (initial.hasPendingQuestion || isTerminalTaskStatus(initial.status)) {
1529 return Option.some(initial);
1530 }
15121531 // One stream per event type, so transcript events never fill a buffer.
15131532 return yield* Stream.mergeAll(
15141533 TASK_WAKE_EVENTS.map(({ thread, eventType }) =>
......@@ -1522,7 +1541,7 @@ const make = Effect.gen(function* () {
15221541 ).pipe(
15231542 Stream.mapError(streamError),
15241543 Stream.mapEffect(() => readTask(scope, taskId, false, true)),
1525 Stream.filter((result) => isTerminalTaskStatus(result.status)),
1544 Stream.filter((result) => result.hasPendingQuestion || isTerminalTaskStatus(result.status)),
15261545 Stream.runHead,
15271546 );
15281547 }).pipe(Effect.timeoutOption(Duration.millis(timeoutMs)), Effect.map(Option.flatten));
......@@ -2083,13 +2102,12 @@ const make = Effect.gen(function* () {
20832102 Math.max(1, input.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS),
20842103 );
20852104 const waited = yield* waitForTask(scope, taskId, timeoutMs);
2086 if (Option.isSome(waited)) {
2105 if (Option.isSome(waited) && isTerminalTaskStatus(waited.value.status)) {
20872106 return waited.value;
20882107 }
2089 // The blocking wait timed out, so it no longer owns delivery: upgrade
2090 // the task so a later terminal wakes the parent even mid-turn. Best
2091 // effort; on failure the settled_only policy still wakes a settled
2092 // parent.
2108 // An incomplete wait gives up delivery: upgrade so a later terminal
2109 // wakes the parent even mid-turn. Best effort; settled_only still
2110 // wakes a settled parent if the upgrade fails.
20932111 yield* threadManagement
20942112 .dispatch({
20952113 type: "delegated_task.wake-policy",
......@@ -2103,7 +2121,7 @@ const make = Effect.gen(function* () {
21032121 completionWake: "always",
21042122 })
21052123 .pipe(
2106 // The tool result is the timed-out task either way, so failures
2124 // The tool result still returns after a timeout or question, so failures
21072125 // stay warnings. Keep the two shapes apart: a rejected receipt
21082126 // means this exact command id already failed (a replay of a
21092127 // no-op upgrade), while anything else is a fresh dispatch fault.
......@@ -2125,7 +2143,7 @@ const make = Effect.gen(function* () {
21252143 }),
21262144 ),
21272145 );
2128 return yield* readTask(scope, taskId, true, true);
2146 return yield* readTask(scope, taskId, Option.isNone(waited), true);
21292147 }),
21302148 taskStatus: (callerScope, taskId) =>
21312149 Effect.gen(function* () {
apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts+61
......@@ -30,6 +30,7 @@ import {
3030 ProviderThreadId,
3131 ProviderTurnId,
3232 type ScheduledTask,
33 RuntimeRequestId,
3334 ScheduledTaskId,
3435 type ScheduledTaskUpsertInput,
3536 type ServerProvider,
......@@ -2722,6 +2723,66 @@ describe("orchestrator MCP toolkit", () => {
27222723 ).toMatchObject({ state: "disposed" });
27232724 yield* expectOffersToStay(0);
27242725
2726 // A live question ends the wait and leaves completion delivery armed.
2727 const questionWait = yield* invoke("delegate_task", {
2728 task: cancellationPrompt,
2729 title: "Wait question",
2730 target: { providerInstanceId: codexInstanceId, model: codexModel },
2731 mode: "wait",
2732 timeoutMs: 30_000,
2733 clientRequestId: "delegate-wait-question-1",
2734 }).pipe(Effect.forkChild);
2735 const question = (yield* waitForProjection(orchestrator, parentThreadId, (p) =>
2736 p.subagents.some(
2737 (task) => task.title === "Wait question" && task.childThreadId !== null,
2738 ),
2739 )).subagents.find((task) => task.title === "Wait question")!;
2740 const asking = yield* waitForProjection(orchestrator, question.childThreadId!, (p) =>
2741 p.providerTurns.some((turn) => turn.status === "running"),
2742 );
2743 const askedAt = yield* DateTime.now;
2744 yield* (yield* EventSink.EventSinkV2).write({
2745 events: [
2746 {
2747 id: EventId.make("event:wait-question:live"),
2748 type: "runtime-request.updated",
2749 threadId: question.childThreadId!,
2750 occurredAt: askedAt,
2751 payload: {
2752 id: RuntimeRequestId.make("request:wait-question"),
2753 nodeId: question.id,
2754 providerTurnId: null,
2755 nativeRequestRef: null,
2756 kind: "user_input",
2757 status: "pending",
2758 responseCapability: {
2759 type: "live",
2760 providerSessionId: asking.providerSessions[0]!.id,
2761 },
2762 createdAt: askedAt,
2763 resolvedAt: null,
2764 },
2765 },
2766 ],
2767 });
2768 expect((yield* Fiber.join(questionWait)).structuredContent).toMatchObject({
2769 status: "running",
2770 hasPendingQuestion: true,
2771 waitTimedOut: false,
2772 });
2773 expect(
2774 (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find(
2775 (task) => task.id === question.id,
2776 )?.completionWake,
2777 ).toBe("always");
2778 yield* invoke("task_cancel", {
2779 taskId: question.id,
2780 clientRequestId: "cancel-wait-question-1",
2781 });
2782 yield* waitForProjection(orchestrator, parentThreadId, (p) =>
2783 p.subagents.some((task) => task.id === question.id && task.status === "interrupted"),
2784 );
2785
27252786 // The MCP tool cannot force the reverse interleaving (child
27262787 // terminal before the upgrade lands), so dispatch the command
27272788 // directly. The first wait-mode delegation completed while the
apps/server/src/mcp/toolkits/orchestrator/tools.ts+1-1
......@@ -65,7 +65,7 @@ const OrchestratorCapabilitiesTool = Tool.make("orchestrator_capabilities", {
6565
6666export const DelegateTaskTool = Tool.make("delegate_task", {
6767 description:
68 "Needs an agent running inside a T3 thread. Delegate one task to a T3-owned child agent/subagent of THIS thread and run it with only the supplied task prompt, without copying parent conversation history. Choose providers and models from orchestrator_capabilities, which uses the same live catalog as the composer. Prefer native subagent tools for same-provider work only when they support the chosen model. Use this for any model missing from the native tool, including same-provider work, for cross-provider work, or for explicitly T3-owned child tasks. For every T3 delegated review round, call delegate_task again with the original brief, prior findings, responses, and unresolved objections in the task prompt. Track each round by its own taskId and use a distinct clientRequestId per round, stable across retries of that round. The childThreadId is backing storage, not the target for starting another delegated review round through t3_thread_send. Set isolation to worktree for a separate checkout based on the parent local commit, with project setup completed before the child starts. Uncommitted edits are not copied; omitted isolation shares the parent checkout. Provider, model, model options (see orchestrator_capabilities), runtime mode, and interaction mode inherit unless target overrides them. A requested runtimeMode must be one the target provider lists in orchestrator_capabilities runtimeModes; an inherited one steps down to the broadest mode it offers. Leave runtimeMode unset unless the user asks for a supervised child: approval-required and auto-accept-edits usually make the child wait for a person at each action that needs approval. Prefer mode='async' for long work; mode='wait' blocks until completion or timeout. timeoutMs on mode=wait is only the parent's wait budget and does not cancel the child. waitTimedOut on that wait call means the timeout fired; keep that taskId and read status on later task_status. An async child's completion wakes this thread through a notification, steered into active turns where supported or queued otherwise, so end the turn instead of polling or spawning watchers; use task_status only when the result is needed mid-turn.",
68 "Needs an agent running inside a T3 thread. Delegate one task to a T3-owned child agent/subagent of THIS thread and run it with only the supplied task prompt, without copying parent conversation history. Choose providers and models from orchestrator_capabilities, which uses the same live catalog as the composer. Prefer native subagent tools for same-provider work only when they support the chosen model. Use this for any model missing from the native tool, including same-provider work, for cross-provider work, or for explicitly T3-owned child tasks. For every T3 delegated review round, call delegate_task again with the original brief, prior findings, responses, and unresolved objections in the task prompt. Track each round by its own taskId and use a distinct clientRequestId per round, stable across retries of that round. The childThreadId is backing storage, not the target for starting another delegated review round through t3_thread_send. Set isolation to worktree for a separate checkout based on the parent local commit, with project setup completed before the child starts. Uncommitted edits are not copied; omitted isolation shares the parent checkout. Provider, model, model options (see orchestrator_capabilities), runtime mode, and interaction mode inherit unless target overrides them. A requested runtimeMode must be one the target provider lists in orchestrator_capabilities runtimeModes; an inherited one steps down to the broadest mode it offers. Leave runtimeMode unset unless the user asks for a supervised child: approval-required and auto-accept-edits usually make the child wait for a person at each action that needs approval. Prefer mode='async' for long work; mode='wait' blocks until completion, a blocking question, or timeout. If hasPendingQuestion, pass threadId=childThreadId to t3_pending_request_list, t3_pending_request_read (with requestId) and t3_pending_request_respond, or ask the user; the task's completion still wakes this thread. timeoutMs on mode=wait is only the parent's wait budget and does not cancel the child. waitTimedOut on that wait call means the timeout fired; keep that taskId and read status on later task_status. An async child's completion wakes this thread through a notification, steered into active turns where supported or queued otherwise, so end the turn instead of polling or spawning watchers; use task_status only when the result is needed mid-turn.",
6969 parameters: OrchestratorMcpDelegateTaskInput,
7070 success: OrchestratorMcpDelegateTaskResult,
7171 failure: OrchestratorMcpFailure,
apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts+1
......@@ -670,6 +670,7 @@ describe("CodexAdapterV2 process spawning", () => {
670670 mcp_servers: {
671671 "t3-code": {
672672 url: "http://127.0.0.1:43123/mcp",
673 tool_timeout_sec: 3_900,
673674 http_headers: {
674675 Authorization: "Bearer secret-codex-token",
675676 },
apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts+5
......@@ -1324,6 +1324,10 @@ export const CODEX_THREAD_CONFIG = {
13241324 mcp_optional_startup_grace_ms: 10_000,
13251325} as const;
13261326
1327// T3 wait tools can block for up to an hour. Keep Codex's MCP timeout above
1328// that limit so the server's own wait timeout ends a long call.
1329const CODEX_T3_MCP_TOOL_TIMEOUT_SEC = 65 * 60;
1330
13271331export function codexThreadRuntimeParams(input: {
13281332 readonly mcpSession: McpProviderSession.McpProviderSessionConfig | undefined;
13291333 readonly modelSelection?: { readonly model: string };
......@@ -1345,6 +1349,7 @@ export function codexThreadRuntimeParams(input: {
13451349 mcp_servers: {
13461350 "t3-code": {
13471351 url: mcpSession.endpoint,
1352 tool_timeout_sec: CODEX_T3_MCP_TOOL_TIMEOUT_SEC,
13481353 http_headers: {
13491354 Authorization: mcpSession.authorizationHeader,
13501355 },
docs/orchestration-v2/orchestrator-mcp-server.md+7-1
......@@ -259,7 +259,12 @@ provider session. The request becomes the V2 command
259259`mode: "async"` returns the current durable state immediately.
260260`mode: "wait"` waits for the task result, including nested work and completion follow-ups, or until
261261the timeout expires. A wait timeout does not cancel the child; the result sets
262`waitTimedOut: true`, and the caller can continue with `task_status`.
262`waitTimedOut: true`, and the caller can continue with `task_status`. A pending blocking
263question also ends the wait with `hasPendingQuestion: true` and `waitTimedOut: false`.
264If `hasPendingQuestion`, pass `threadId=childThreadId` to `t3_pending_request_list`,
265`t3_pending_request_read` (with `requestId`) and `t3_pending_request_respond`, or ask
266the user; the task's completion still wakes this thread. Async questions and approvals
267do not end the wait.
263268
264269```ts
265270type DelegateTaskResult = {
......@@ -278,6 +283,7 @@ type DelegateTaskResult = {
278283 latestTerminalStatus: "completed" | "failed" | "cancelled" | "interrupted" | null;
279284 latestTerminalSummary: string | null;
280285 latestTerminalResultContextTransferId: string | null;
286 hasPendingQuestion: boolean;
281287 waitTimedOut: boolean;
282288};
283289```
packages/contracts/src/orchestratorMcp.test.ts+1
......@@ -58,6 +58,7 @@ describe("orchestrator MCP contracts", () => {
5858 latestTerminalStatus: "completed",
5959 latestTerminalSummary: "Workspace inspected.",
6060 latestTerminalResultContextTransferId: "context-transfer-result-1",
61 hasPendingQuestion: false,
6162 waitTimedOut: false,
6263 });
6364
packages/contracts/src/orchestratorMcp.ts+4
......@@ -217,6 +217,10 @@ export const OrchestratorMcpDelegateTaskResult = Schema.Struct({
217217 model: Schema.NullOr(Schema.String),
218218 summary: Schema.NullOr(Schema.String),
219219 resultContextTransferId: Schema.NullOr(ContextTransferId),
220 hasPendingQuestion: Schema.Boolean.annotate({
221 description:
222 "Blocking question ends mode=wait. Pass threadId=childThreadId to t3_pending_request_list, t3_pending_request_read (with requestId) and t3_pending_request_respond, or ask the user; the task's completion still wakes this thread.",
223 }),
220224 waitTimedOut: Schema.Boolean.annotate({
221225 description:
222226 "True only on that mode=wait call when timeoutMs elapsed. The timeout does not cancel the child. Later task_status reads return false and use status for liveness.",