diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index b0688fd217bb803ce2793caa8313efd69198d776..64f2d86eda9452c58f55f68be5881f8e8316b494 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -96,6 +96,7 @@ describe("OrchestratorMcpService", () => { messages: [], subagents: [], providerThreads: [], + runtimeRequests: [], } as unknown as OrchestrationV2ThreadProjection; let hasNestedWork = true; const layerDependencies = Layer.mergeAll( @@ -223,6 +224,7 @@ describe("OrchestratorMcpService", () => { messages: [], subagents: [], providerThreads: [], + runtimeRequests: [], turnItems: [], } as unknown as OrchestrationV2ThreadProjection; const layerDependencies = Layer.mergeAll( @@ -308,6 +310,7 @@ describe("OrchestratorMcpService", () => { messages: [], subagents: [], providerThreads: [], + runtimeRequests: [], } as unknown as OrchestrationV2ThreadProjection; const layerDependencies = Layer.mergeAll( NodeServices.layer, @@ -385,6 +388,7 @@ describe("OrchestratorMcpService", () => { messages: [], subagents: [], providerThreads: [], + runtimeRequests: [], } as unknown as OrchestrationV2ThreadProjection; const layerDependencies = Layer.mergeAll( NodeServices.layer, @@ -465,6 +469,7 @@ describe("OrchestratorMcpService", () => { messages: [], subagents: [], providerThreads: [], + runtimeRequests: [], } as unknown as OrchestrationV2ThreadProjection; const layerDependencies = Layer.mergeAll( NodeServices.layer, @@ -552,6 +557,7 @@ describe("OrchestratorMcpService", () => { messages: [], subagents: [], providerThreads: [], + runtimeRequests: [], } as unknown as OrchestrationV2ThreadProjection; const layerDependencies = Layer.mergeAll( NodeServices.layer, @@ -655,6 +661,7 @@ describe("OrchestratorMcpService", () => { appOwnedTask("node:mcp-cancel-grandchild-below", childThreadId, grandchildThreadId), ], providerThreads: [], + runtimeRequests: [], }, ], [ @@ -666,6 +673,7 @@ describe("OrchestratorMcpService", () => { messages: [], subagents: [], providerThreads: [], + runtimeRequests: [], }, ], ]) as unknown as ReadonlyMap; @@ -883,6 +891,7 @@ describe("OrchestratorMcpService provider resolution", () => { messages: [], subagents: [], providerThreads: [], + runtimeRequests: [], turnItems: [], } as unknown as OrchestrationV2ThreadProjection; diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 4055544a394731d34eac33ff7cdb9dee0a1353c0..5c0a93cb1c0fc07fe73a5dff2d3c2317f1582968 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -101,13 +101,14 @@ import * as SecretRequests from "../secrets/SecretRequests.ts"; const DEFAULT_WAIT_TIMEOUT_MS = 10 * 60 * 1_000; const MAX_WAIT_TIMEOUT_MS = 60 * 60 * 1_000; -// Events that can make a delegated task terminal: the parent's task record, -// and the child's runs, nested tasks, and pending provider background work. +// Events that can end a delegated task wait: the parent's task record, +// and the child's runs, nested tasks, provider background work, and questions. const TASK_WAKE_EVENTS = [ { thread: "parent", eventType: "subagent.updated" }, { thread: "child", eventType: "run.updated" }, { thread: "child", eventType: "subagent.updated" }, { thread: "child", eventType: "provider-thread.updated" }, + { thread: "child", eventType: "runtime-request.updated" }, ] as const; /** A person answers the card, so a slower poll is plenty. */ const SECRET_REQUEST_POLL_INTERVAL_MS = 500; @@ -1351,7 +1352,14 @@ const make = Effect.gen(function* () { const childControls = yield* threadManagement .getThreadRecords( task.childThreadId, - ["runs", "messages", "contextTransfers", "subagents", "providerThreads"], + [ + "runs", + "messages", + "contextTransfers", + "subagents", + "providerThreads", + "runtimeRequests", + ], { messageRoles: ["user"] }, ) .pipe(Effect.mapError(threadManagementFailure)); @@ -1456,6 +1464,15 @@ const make = Effect.gen(function* () { : subagentResultForRun(childProjection, terminalRun).text : null, latestTerminalResultContextTransferId: resultTransferForRun(terminalRun)?.id ?? null, + // A finished task has no question of its own; later child runs belong to other tasks. + hasPendingQuestion: + !isTerminalTaskStatus(status) && + childControls.runtimeRequests.some( + (request) => + request.kind === "user_input" && + request.status === "pending" && + request.responseCapability.type === "live", + ), waitTimedOut, } satisfies OrchestratorMcpDelegateTaskResult; if ( @@ -1508,7 +1525,9 @@ const make = Effect.gen(function* () { .getThreadEventSequence(scope.thread.threadId) .pipe(Effect.mapError(streamError)); const initial = yield* readTask(scope, taskId, false, true); - if (isTerminalTaskStatus(initial.status)) return Option.some(initial); + if (initial.hasPendingQuestion || isTerminalTaskStatus(initial.status)) { + return Option.some(initial); + } // One stream per event type, so transcript events never fill a buffer. return yield* Stream.mergeAll( TASK_WAKE_EVENTS.map(({ thread, eventType }) => @@ -1522,7 +1541,7 @@ const make = Effect.gen(function* () { ).pipe( Stream.mapError(streamError), Stream.mapEffect(() => readTask(scope, taskId, false, true)), - Stream.filter((result) => isTerminalTaskStatus(result.status)), + Stream.filter((result) => result.hasPendingQuestion || isTerminalTaskStatus(result.status)), Stream.runHead, ); }).pipe(Effect.timeoutOption(Duration.millis(timeoutMs)), Effect.map(Option.flatten)); @@ -2083,13 +2102,12 @@ const make = Effect.gen(function* () { Math.max(1, input.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS), ); const waited = yield* waitForTask(scope, taskId, timeoutMs); - if (Option.isSome(waited)) { + if (Option.isSome(waited) && isTerminalTaskStatus(waited.value.status)) { return waited.value; } - // The blocking wait timed out, so it no longer owns delivery: upgrade - // the task so a later terminal wakes the parent even mid-turn. Best - // effort; on failure the settled_only policy still wakes a settled - // parent. + // An incomplete wait gives up delivery: upgrade so a later terminal + // wakes the parent even mid-turn. Best effort; settled_only still + // wakes a settled parent if the upgrade fails. yield* threadManagement .dispatch({ type: "delegated_task.wake-policy", @@ -2103,7 +2121,7 @@ const make = Effect.gen(function* () { completionWake: "always", }) .pipe( - // The tool result is the timed-out task either way, so failures + // The tool result still returns after a timeout or question, so failures // stay warnings. Keep the two shapes apart: a rejected receipt // means this exact command id already failed (a replay of a // no-op upgrade), while anything else is a fresh dispatch fault. @@ -2125,7 +2143,7 @@ const make = Effect.gen(function* () { }), ), ); - return yield* readTask(scope, taskId, true, true); + return yield* readTask(scope, taskId, Option.isNone(waited), true); }), taskStatus: (callerScope, taskId) => Effect.gen(function* () { diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index b68f3f29c8da95a02f8b12864ebaf9455f0469ba..c518fd45948144e9d73947ad9c2f65af522bca04 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -30,6 +30,7 @@ import { ProviderThreadId, ProviderTurnId, type ScheduledTask, + RuntimeRequestId, ScheduledTaskId, type ScheduledTaskUpsertInput, type ServerProvider, @@ -2722,6 +2723,66 @@ describe("orchestrator MCP toolkit", () => { ).toMatchObject({ state: "disposed" }); yield* expectOffersToStay(0); + // A live question ends the wait and leaves completion delivery armed. + const questionWait = yield* invoke("delegate_task", { + task: cancellationPrompt, + title: "Wait question", + target: { providerInstanceId: codexInstanceId, model: codexModel }, + mode: "wait", + timeoutMs: 30_000, + clientRequestId: "delegate-wait-question-1", + }).pipe(Effect.forkChild); + const question = (yield* waitForProjection(orchestrator, parentThreadId, (p) => + p.subagents.some( + (task) => task.title === "Wait question" && task.childThreadId !== null, + ), + )).subagents.find((task) => task.title === "Wait question")!; + const asking = yield* waitForProjection(orchestrator, question.childThreadId!, (p) => + p.providerTurns.some((turn) => turn.status === "running"), + ); + const askedAt = yield* DateTime.now; + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make("event:wait-question:live"), + type: "runtime-request.updated", + threadId: question.childThreadId!, + occurredAt: askedAt, + payload: { + id: RuntimeRequestId.make("request:wait-question"), + nodeId: question.id, + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input", + status: "pending", + responseCapability: { + type: "live", + providerSessionId: asking.providerSessions[0]!.id, + }, + createdAt: askedAt, + resolvedAt: null, + }, + }, + ], + }); + expect((yield* Fiber.join(questionWait)).structuredContent).toMatchObject({ + status: "running", + hasPendingQuestion: true, + waitTimedOut: false, + }); + expect( + (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (task) => task.id === question.id, + )?.completionWake, + ).toBe("always"); + yield* invoke("task_cancel", { + taskId: question.id, + clientRequestId: "cancel-wait-question-1", + }); + yield* waitForProjection(orchestrator, parentThreadId, (p) => + p.subagents.some((task) => task.id === question.id && task.status === "interrupted"), + ); + // The MCP tool cannot force the reverse interleaving (child // terminal before the upgrade lands), so dispatch the command // directly. The first wait-mode delegation completed while the diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index 28489d1dca137a73623940df023143e566cab011..36c4d78723c06c611296dc854914020220a78318 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -65,7 +65,7 @@ const OrchestratorCapabilitiesTool = Tool.make("orchestrator_capabilities", { export const DelegateTaskTool = Tool.make("delegate_task", { description: - "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.", + "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.", parameters: OrchestratorMcpDelegateTaskInput, success: OrchestratorMcpDelegateTaskResult, failure: OrchestratorMcpFailure, diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index e7eb16c378cb78592620ed2b0e81e4462b2d8000..8f6d0edb7eb8ac95288778afc9efb2a3dfe961f0 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -670,6 +670,7 @@ describe("CodexAdapterV2 process spawning", () => { mcp_servers: { "t3-code": { url: "http://127.0.0.1:43123/mcp", + tool_timeout_sec: 3_900, http_headers: { Authorization: "Bearer secret-codex-token", }, diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index aace0e6e008610db7eadee25ec4dbe2e28332de9..7f55417ea2ac0d35b6bcd87a1b9d3afd80f5b10a 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1324,6 +1324,10 @@ export const CODEX_THREAD_CONFIG = { mcp_optional_startup_grace_ms: 10_000, } as const; +// T3 wait tools can block for up to an hour. Keep Codex's MCP timeout above +// that limit so the server's own wait timeout ends a long call. +const CODEX_T3_MCP_TOOL_TIMEOUT_SEC = 65 * 60; + export function codexThreadRuntimeParams(input: { readonly mcpSession: McpProviderSession.McpProviderSessionConfig | undefined; readonly modelSelection?: { readonly model: string }; @@ -1345,6 +1349,7 @@ export function codexThreadRuntimeParams(input: { mcp_servers: { "t3-code": { url: mcpSession.endpoint, + tool_timeout_sec: CODEX_T3_MCP_TOOL_TIMEOUT_SEC, http_headers: { Authorization: mcpSession.authorizationHeader, }, diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 20f6456b1a34b98ed911082959a3581b47949588..8b701832440355d0aa74325c54485cf48d6d3256 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -259,7 +259,12 @@ provider session. The request becomes the V2 command `mode: "async"` returns the current durable state immediately. `mode: "wait"` waits for the task result, including nested work and completion follow-ups, or until the timeout expires. A wait timeout does not cancel the child; the result sets -`waitTimedOut: true`, and the caller can continue with `task_status`. +`waitTimedOut: true`, and the caller can continue with `task_status`. A pending blocking +question also ends the wait with `hasPendingQuestion: true` and `waitTimedOut: false`. +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. Async questions and approvals +do not end the wait. ```ts type DelegateTaskResult = { @@ -278,6 +283,7 @@ type DelegateTaskResult = { latestTerminalStatus: "completed" | "failed" | "cancelled" | "interrupted" | null; latestTerminalSummary: string | null; latestTerminalResultContextTransferId: string | null; + hasPendingQuestion: boolean; waitTimedOut: boolean; }; ``` diff --git a/packages/contracts/src/orchestratorMcp.test.ts b/packages/contracts/src/orchestratorMcp.test.ts index 4c4a9a4a5aac92cd68235a2ea3d69c8d1cddec8a..306044c213ad97faf49eb4dbfeb31ff3af5b2655 100644 --- a/packages/contracts/src/orchestratorMcp.test.ts +++ b/packages/contracts/src/orchestratorMcp.test.ts @@ -58,6 +58,7 @@ describe("orchestrator MCP contracts", () => { latestTerminalStatus: "completed", latestTerminalSummary: "Workspace inspected.", latestTerminalResultContextTransferId: "context-transfer-result-1", + hasPendingQuestion: false, waitTimedOut: false, }); diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 2aa36d456a37b19e9e63475e9fdb0c14091d6c12..80328d279794fb54729d8a74f3a89bdf37555b98 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -217,6 +217,10 @@ export const OrchestratorMcpDelegateTaskResult = Schema.Struct({ model: Schema.NullOr(Schema.String), summary: Schema.NullOr(Schema.String), resultContextTransferId: Schema.NullOr(ContextTransferId), + hasPendingQuestion: Schema.Boolean.annotate({ + description: + "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.", + }), waitTimedOut: Schema.Boolean.annotate({ description: "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.",