From a2d68963e803e13050f5a687c82ca786b07a8b26 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 29 Aug 2026 18:07:02 -0700 Subject: [PATCH 1/5] feat(mcp): answer delegated user-input requests --- apps/server/src/mcp/McpHttpServer.ts | 9 + apps/server/src/mcp/McpModeCeilings.ts | 32 ++ apps/server/src/mcp/OrchestratorMcpService.ts | 23 +- ...OrchestratorMcpToolkit.integration.test.ts | 239 ++++++++- .../src/mcp/PendingRequestMcpService.test.ts | 427 ++++++++++++++++ .../src/mcp/PendingRequestMcpService.ts | 462 ++++++++++++++++++ .../mcp/toolkits/pendingRequest/handlers.ts | 20 + .../src/mcp/toolkits/pendingRequest/tools.ts | 61 +++ .../toolkits/worktree/registration.test.ts | 9 + .../Adapters/ClaudeAdapterV2.test.ts | 8 +- .../Adapters/ClaudeAdapterV2.ts | 6 +- .../src/orchestration-v2/Orchestrator.ts | 106 +++- .../ThreadManagementService.ts | 2 + .../orchestrator-mcp-server.md | 29 +- docs/user/permission-modes.md | 5 + packages/contracts/src/index.ts | 1 + packages/contracts/src/orchestrationV2.ts | 8 + packages/contracts/src/orchestratorMcp.ts | 1 + .../contracts/src/pendingRequestMcp.test.ts | 50 ++ packages/contracts/src/pendingRequestMcp.ts | 103 ++++ packages/shared/src/t3McpToolPresentation.ts | 3 + 21 files changed, 1577 insertions(+), 27 deletions(-) create mode 100644 apps/server/src/mcp/McpModeCeilings.ts create mode 100644 apps/server/src/mcp/PendingRequestMcpService.test.ts create mode 100644 apps/server/src/mcp/PendingRequestMcpService.ts create mode 100644 apps/server/src/mcp/toolkits/pendingRequest/handlers.ts create mode 100644 apps/server/src/mcp/toolkits/pendingRequest/tools.ts create mode 100644 packages/contracts/src/pendingRequestMcp.test.ts create mode 100644 packages/contracts/src/pendingRequestMcp.ts diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index 03492c3ef366..d139a6829ed7 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -14,6 +14,7 @@ import * as McpInvocationContext from "./McpInvocationContext.ts"; import * as OrchestratorMcpService from "./OrchestratorMcpService.ts"; import * as McpSessionRegistry from "./McpSessionRegistry.ts"; import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts"; +import * as PendingRequestMcpService from "./PendingRequestMcpService.ts"; import { OrchestratorToolkitHandlersLive } from "./toolkits/orchestrator/handlers.ts"; import { OrchestratorToolkit } from "./toolkits/orchestrator/tools.ts"; import { @@ -25,6 +26,8 @@ import { PreviewSnapshotToolkit, PreviewStandardToolkit, } from "./toolkits/preview/tools.ts"; +import { PendingRequestToolkitHandlersLive } from "./toolkits/pendingRequest/handlers.ts"; +import { PendingRequestToolkit } from "./toolkits/pendingRequest/tools.ts"; import { WorktreeToolkitHandlersLive } from "./toolkits/worktree/handlers.ts"; import { WorktreeToolkit } from "./toolkits/worktree/tools.ts"; import * as WorktreeMcpService from "./WorktreeMcpService.ts"; @@ -227,6 +230,11 @@ export const OrchestratorToolkitRegistrationLive = McpServer.toolkit(Orchestrato Layer.provide(OrchestratorMcpService.layer), ); +export const PendingRequestToolkitRegistrationLive = McpServer.toolkit(PendingRequestToolkit).pipe( + Layer.provide(PendingRequestToolkitHandlersLive), + Layer.provide(PendingRequestMcpService.layer), +); + export const WorktreeToolkitRegistrationLive = McpServer.toolkit(WorktreeToolkit).pipe( Layer.provide(WorktreeToolkitHandlersLive), Layer.provide(WorktreeMcpService.layer), @@ -242,5 +250,6 @@ const McpTransportLive = McpServer.layerHttp({ export const layer = Layer.mergeAll( PreviewToolkitRegistrationLive, OrchestratorToolkitRegistrationLive, + PendingRequestToolkitRegistrationLive, WorktreeToolkitRegistrationLive, ).pipe(Layer.provideMerge(McpTransportLive)); diff --git a/apps/server/src/mcp/McpModeCeilings.ts b/apps/server/src/mcp/McpModeCeilings.ts new file mode 100644 index 000000000000..4ff0fbfd8765 --- /dev/null +++ b/apps/server/src/mcp/McpModeCeilings.ts @@ -0,0 +1,32 @@ +import type { ProviderInteractionMode, RuntimeMode } from "@t3tools/contracts"; + +function runtimeModeRank(mode: RuntimeMode): number { + switch (mode) { + case "approval-required": + return 0; + case "auto-accept-edits": + return 1; + case "auto": + return 2; + case "full-access": + return 3; + } +} + +function interactionModeRank(mode: ProviderInteractionMode): number { + return mode === "plan" ? 0 : 1; +} + +export function runtimeModeWithinMcpCeiling( + callerMode: RuntimeMode, + targetMode: RuntimeMode, +): boolean { + return runtimeModeRank(targetMode) <= runtimeModeRank(callerMode); +} + +export function interactionModeWithinMcpCeiling( + callerMode: ProviderInteractionMode, + targetMode: ProviderInteractionMode, +): boolean { + return interactionModeRank(targetMode) <= interactionModeRank(callerMode); +} diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 7d7bd24f02de..341d79943750 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -71,6 +71,7 @@ import { } from "../orchestration-v2/ThreadManagementService.ts"; import { ProviderRegistry } from "../provider/Services/ProviderRegistry.ts"; import { ScheduledTaskService } from "../scheduledTasks/ScheduledTaskService.ts"; +import { interactionModeWithinMcpCeiling, runtimeModeWithinMcpCeiling } from "./McpModeCeilings.ts"; import type { McpInvocationScope } from "./McpInvocationContext.ts"; const DEFAULT_WAIT_TIMEOUT_MS = 10 * 60 * 1_000; @@ -371,29 +372,12 @@ function pageIncludesTerminalTaskResult(input: { }); } -function runtimeModeRank(mode: RuntimeMode): number { - switch (mode) { - case "approval-required": - return 0; - case "auto-accept-edits": - return 1; - case "auto": - return 2; - case "full-access": - return 3; - } -} - -function interactionModeRank(mode: ProviderInteractionMode): number { - return mode === "plan" ? 0 : 1; -} - function resolveRuntimeMode( parentMode: RuntimeMode, requested: OrchestratorMcpRuntimeMode | undefined, ): Effect.Effect { const resolved = requested === undefined || requested === "inherit" ? parentMode : requested; - return runtimeModeRank(resolved) > runtimeModeRank(parentMode) + return !runtimeModeWithinMcpCeiling(parentMode, resolved) ? Effect.fail( failure( "runtime_mode_escalation_denied", @@ -408,7 +392,7 @@ function resolveInteractionMode( requested: OrchestratorMcpInteractionMode | undefined, ): Effect.Effect { const resolved = requested === undefined || requested === "inherit" ? parentMode : requested; - return interactionModeRank(resolved) > interactionModeRank(parentMode) + return !interactionModeWithinMcpCeiling(parentMode, resolved) ? Effect.fail( failure( "interaction_mode_escalation_denied", @@ -1113,6 +1097,7 @@ const make = Effect.gen(function* () { threadManagement: true, incrementalThreadRead: true, scheduledTasks: true, + delegatedUserInputRequests: true, maxBatchThreads: 20, }, }; diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index ad6ea73c7688..21cb618e6415 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -6,6 +6,7 @@ import { IsoDateTime, MessageId, type ModelSelection, + NodeId, type OrchestrationV2ProviderCapabilities, type OrchestrationV2ProviderSession, type OrchestrationV2ProviderThread, @@ -19,12 +20,16 @@ import { OrchestratorMcpThreadReadResult, OrchestratorMcpThreadSendResult, OrchestratorMcpThreadWaitResult, + PendingRequestMcpListResult, + PendingRequestMcpReadResult, + PendingRequestMcpRespondResult, ProjectId, ProviderDriverKind, ProviderInstanceId, type ProviderOptionDescriptor, ProviderThreadId, ProviderTurnId, + RuntimeRequestId, type ScheduledTask, ScheduledTaskId, type ScheduledTaskUpsertInput, @@ -49,6 +54,7 @@ import { layer as threadManagementServiceLayer } from "../orchestration-v2/Threa import { type ProviderAdapterV2Event, ProviderAdapterProtocolError, + type ProviderAdapterV2RuntimeRequestResponseInput, type ProviderAdapterV2Shape, type ProviderAdapterV2TurnInput, } from "../orchestration-v2/ProviderAdapter.ts"; @@ -75,6 +81,7 @@ const delegatedPrompt = "Inspect the delegated API boundary and return the resul const delegatedResult = "Delegated API boundary inspected."; const cancellationPrompt = "Remain active until the parent cancels this delegated task."; const createdThreadPrompt = "Complete the newly created ordinary thread."; +const delegatedQuestionPrompt = "Ask which editor to configure, then finish after the answer."; const decodeCreateThreadsResult = Schema.decodeUnknownEffect(OrchestratorMcpCreateThreadsResult); const decodeCreatedThread = Schema.decodeUnknownEffect(OrchestratorMcpCreatedThread); @@ -87,6 +94,11 @@ const decodeThreadListResult = Schema.decodeUnknownEffect(OrchestratorMcpThreadL const decodeThreadReadResult = Schema.decodeUnknownEffect(OrchestratorMcpThreadReadResult); const decodeThreadSendResult = Schema.decodeUnknownEffect(OrchestratorMcpThreadSendResult); const decodeThreadWaitResult = Schema.decodeUnknownEffect(OrchestratorMcpThreadWaitResult); +const decodePendingRequestListResult = Schema.decodeUnknownEffect(PendingRequestMcpListResult); +const decodePendingRequestReadResult = Schema.decodeUnknownEffect(PendingRequestMcpReadResult); +const decodePendingRequestRespondResult = Schema.decodeUnknownEffect( + PendingRequestMcpRespondResult, +); const codexSelection = { instanceId: codexInstanceId, @@ -146,6 +158,10 @@ function makeDeterministicAdapter(input: { readonly capturedTurns: Ref.Ref>; readonly shouldComplete: (turn: ProviderAdapterV2TurnInput) => boolean; readonly terminalGate?: (turn: ProviderAdapterV2TurnInput) => Deferred.Deferred | undefined; + readonly userInput?: { + readonly shouldAsk: (turn: ProviderAdapterV2TurnInput) => boolean; + readonly response: Deferred.Deferred; + }; readonly response: (turn: ProviderAdapterV2TurnInput) => string; }): ProviderAdapterV2Shape { return { @@ -247,6 +263,96 @@ function makeDeterministicAdapter(input: { }, }, ]); + if (input.userInput?.shouldAsk(turnInput) === true) { + const runtimeRequestId = RuntimeRequestId.make( + `request:${input.instanceId}:${turnInput.threadId}:${turnInput.runOrdinal}:editor`, + ); + const requestNodeId = NodeId.make( + `node:${input.instanceId}:${turnInput.threadId}:${turnInput.runOrdinal}:editor`, + ); + yield* publish([ + { + type: "node.updated", + driver: input.driver, + node: { + id: requestNodeId, + threadId: turnInput.threadId, + runId: turnInput.runId, + parentNodeId: turnInput.rootNodeId, + rootNodeId: turnInput.rootNodeId, + kind: "user_input_request", + status: "waiting", + countsForRun: false, + providerThreadId: turnInput.providerThread.id, + providerTurnId, + nativeItemRef: null, + runtimeRequestId, + checkpointScopeId: null, + startedAt: eventTime, + completedAt: null, + }, + }, + { + type: "runtime_request.updated", + driver: input.driver, + threadId: turnInput.threadId, + runtimeRequest: { + id: runtimeRequestId, + nodeId: requestNodeId, + providerTurnId, + nativeRequestRef: null, + kind: "user_input", + status: "pending", + responseCapability: { + type: "live", + providerSessionId: sessionInput.providerSessionId, + }, + createdAt: eventTime, + resolvedAt: null, + }, + }, + { + type: "turn_item.updated", + driver: input.driver, + turnItem: { + id: TurnItemId.make( + `turn-item:${input.instanceId}:${turnInput.threadId}:${turnInput.runOrdinal}:editor`, + ), + threadId: turnInput.threadId, + runId: turnInput.runId, + nodeId: requestNodeId, + providerThreadId: turnInput.providerThread.id, + providerTurnId, + nativeItemRef: null, + parentItemId: null, + ordinal: turnInput.runOrdinal * 100 + 1, + status: "waiting", + title: null, + startedAt: eventTime, + completedAt: null, + updatedAt: eventTime, + type: "user_input_request", + requestId: runtimeRequestId, + questions: [ + { + id: "editor", + header: "Editor", + question: "Which editor should the delegated task configure?", + options: [ + { label: "Vim", description: "Configure Vim." }, + { label: "Zed", description: "Configure Zed." }, + ], + }, + ], + }, + }, + ]); + // A real adapter returns after starting the provider turn; its + // notification stream remains live while the provider waits. + // Keeping the deterministic adapter's effect worker free is + // essential because runtime-request.respond is a later effect. + return; + } const terminalGate = input.terminalGate?.(turnInput); if (terminalGate !== undefined) { yield* Deferred.await(terminalGate); @@ -355,7 +461,10 @@ function makeDeterministicAdapter(input: { }, ]); }), - respondToRuntimeRequest: () => Effect.void, + respondToRuntimeRequest: (response) => + input.userInput === undefined + ? Effect.void + : Deferred.succeed(input.userInput.response, response).pipe(Effect.asVoid), readThreadSnapshot: () => unsupported(input.driver, "readThreadSnapshot is unused in this test"), rollbackThread: () => unsupported(input.driver, "rollbackThread is unused in this test"), @@ -430,6 +539,8 @@ describe("orchestrator MCP toolkit", () => { Effect.gen(function* () { const cwd = yield* checkpointWorkspace("orchestrator-mcp-toolkit"); const capturedTurns = yield* Ref.make>([]); + const delegatedQuestionResponse = + yield* Deferred.make(); const parentTerminalGates = new Map>(); const deliveryTerminalGates = new Map>(); const registryLayer = makeProviderAdapterRegistryLayer([ @@ -440,6 +551,10 @@ describe("orchestrator MCP toolkit", () => { capturedTurns, shouldComplete: (turn) => turn.threadId !== parentThreadId && turn.message.text !== cancellationPrompt, + userInput: { + shouldAsk: (turn) => turn.message.text === delegatedQuestionPrompt, + response: delegatedQuestionResponse, + }, terminalGate: (turn) => turn.message.text.startsWith("Delegated task") || turn.message.text.startsWith("Delegated tasks") @@ -570,7 +685,10 @@ describe("orchestrator MCP toolkit", () => { runNow: () => Effect.die("ScheduledTaskService.runNow is unused in this test"), }), ); - const testLayer = McpHttpServer.OrchestratorToolkitRegistrationLive.pipe( + const testLayer = Layer.merge( + McpHttpServer.OrchestratorToolkitRegistrationLive, + McpHttpServer.PendingRequestToolkitRegistrationLive, + ).pipe( Layer.provideMerge(McpServer.McpServer.layer), Layer.provideMerge(orchestrationLayer), Layer.provide(providerRegistryLayer), @@ -1267,6 +1385,123 @@ describe("orchestrator MCP toolkit", () => { }); expect(yield* Ref.get(scheduledStore)).toHaveLength(0); + const questionTaskCall = yield* invoke("delegate_task", { + task: delegatedQuestionPrompt, + target: { + providerInstanceId: codexInstanceId, + model: codexModel, + }, + mode: "async", + clientRequestId: "delegate-codex-question-1", + }); + expect(questionTaskCall.isError).toBe(false); + const questionTask = yield* decodeDelegateTaskResult( + questionTaskCall.structuredContent, + ).pipe(Effect.orDie); + const pendingRequestStored = yield* orchestrator + .streamStoredEventsFrom({ threadId: questionTask.childThreadId }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "runtime-request.updated" && + stored.event.payload.kind === "user_input" && + stored.event.payload.status === "pending", + ), + Stream.runHead, + ); + if ( + pendingRequestStored._tag === "None" || + pendingRequestStored.value.event.type !== "runtime-request.updated" + ) { + return yield* Effect.die(new Error("Delegated user-input request was not stored.")); + } + const delegatedRequestId = pendingRequestStored.value.event.payload.id; + + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-parent:question-ceiling:downgrade"), + threadId: parentThreadId, + runtimeMode: "approval-required", + }); + const deniedByFreshCallerMode = yield* orchestrator + .dispatch({ + type: "runtime-request.respond", + commandId: CommandId.make("command:mcp-parent:question-ceiling:denied"), + threadId: questionTask.childThreadId, + requestId: delegatedRequestId, + answers: { editor: "Vim" }, + policyCeiling: { + callerThreadId: parentThreadId, + runtimeMode: "full-access", + interactionMode: "default", + }, + }) + .pipe(Effect.flip); + expect(deniedByFreshCallerMode._tag).toBe("OrchestratorDispatchError"); + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-parent:question-ceiling:restore"), + threadId: parentThreadId, + runtimeMode: "full-access", + }); + + const pendingListCall = yield* invoke("t3_pending_request_list", {}); + expect(pendingListCall.isError).toBe(false); + const pendingList = yield* decodePendingRequestListResult( + pendingListCall.structuredContent, + ).pipe(Effect.orDie); + expect(pendingList.requests).toEqual([ + expect.objectContaining({ + taskId: questionTask.taskId, + childThreadId: questionTask.childThreadId, + requestId: delegatedRequestId, + status: "pending", + resumable: true, + }), + ]); + + const pendingReadCall = yield* invoke("t3_pending_request_read", { + childThreadId: questionTask.childThreadId, + requestId: delegatedRequestId, + }); + expect(pendingReadCall.isError).toBe(false); + const pendingRead = yield* decodePendingRequestReadResult( + pendingReadCall.structuredContent, + ).pipe(Effect.orDie); + expect(pendingRead.questions).toEqual([expect.objectContaining({ id: "editor" })]); + + const pendingRespondCall = yield* invoke("t3_pending_request_respond", { + childThreadId: questionTask.childThreadId, + requestId: delegatedRequestId, + answers: { editor: "Vim" }, + clientRequestId: "answer-delegated-editor-1", + }); + expect(pendingRespondCall.isError).toBe(false); + const pendingRespond = yield* decodePendingRequestRespondResult( + pendingRespondCall.structuredContent, + ).pipe(Effect.orDie); + expect(pendingRespond).toMatchObject({ + replayed: false, + request: { requestId: delegatedRequestId, status: "resolved" }, + }); + expect(yield* Deferred.await(delegatedQuestionResponse)).toEqual({ + requestId: delegatedRequestId, + answers: { editor: "Vim" }, + }); + + const replayedPendingRespondCall = yield* invoke("t3_pending_request_respond", { + childThreadId: questionTask.childThreadId, + requestId: delegatedRequestId, + answers: { editor: "Vim" }, + clientRequestId: "answer-delegated-editor-1", + }); + expect(replayedPendingRespondCall.isError).toBe(false); + expect(replayedPendingRespondCall.structuredContent).toMatchObject({ + commandId: pendingRespond.commandId, + receiptSequence: pendingRespond.receiptSequence, + replayed: true, + }); + const delegatedCall = yield* invoke("delegate_task", { task: delegatedPrompt, target: { diff --git a/apps/server/src/mcp/PendingRequestMcpService.test.ts b/apps/server/src/mcp/PendingRequestMcpService.test.ts new file mode 100644 index 000000000000..d3b5ddd88314 --- /dev/null +++ b/apps/server/src/mcp/PendingRequestMcpService.test.ts @@ -0,0 +1,427 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, describe, it, vi } from "@effect/vitest"; +import { + EnvironmentId, + NodeId, + ProjectId, + ProviderInstanceId, + ProviderSessionId, + ProviderThreadId, + RuntimeRequestId, + ThreadId, + TurnItemId, + type OrchestrationV2RuntimeRequest, + type OrchestrationV2ThreadProjection, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; + +import type { CommandReceiptV2 } from "../orchestration-v2/CommandReceiptStore.ts"; +import { ThreadManagementService } from "../orchestration-v2/ThreadManagementService.ts"; +import type { McpInvocationScope } from "./McpInvocationContext.ts"; +import { layer, PendingRequestMcpService } from "./PendingRequestMcpService.ts"; + +const parentThreadId = ThreadId.make("thread:pending-request-parent"); +const childThreadId = ThreadId.make("thread:pending-request-child"); +const unrelatedThreadId = ThreadId.make("thread:pending-request-unrelated"); +const projectId = ProjectId.make("project:pending-request"); +const taskId = NodeId.make("node:pending-request-task"); +const requestNodeId = NodeId.make("node:pending-request-question"); +const requestId = RuntimeRequestId.make("request:pending-request-question"); +const providerThreadId = ProviderThreadId.make("provider-thread:pending-request"); +const providerSessionId = ProviderSessionId.make("provider-session:pending-request-child"); +const now = DateTime.makeUnsafe("2026-08-29T12:00:00.000Z"); + +const scope: McpInvocationScope = { + environmentId: EnvironmentId.make("environment:pending-request"), + threadId: parentThreadId, + providerSessionId: "provider-session:mcp-pending-request", + providerInstanceId: ProviderInstanceId.make("codex"), + capabilities: new Set(["orchestration"]), + issuedAt: 1, +}; + +const questions = [ + { + id: "editor", + header: "Editor", + question: "Which editor should the delegated task configure?", + options: [ + { label: "Vim", description: "Use Vim." }, + { label: "Zed", description: "Use Zed." }, + ], + }, +]; + +function parentProjection( + origin: "app_owned" | "provider_native" = "app_owned", + modes: { + readonly runtimeMode?: "approval-required" | "auto-accept-edits" | "auto" | "full-access"; + readonly interactionMode?: "plan" | "default"; + } = {}, +) { + return { + thread: { + id: parentThreadId, + projectId, + runtimeMode: modes.runtimeMode ?? "auto-accept-edits", + interactionMode: modes.interactionMode ?? "default", + deletedAt: null, + }, + subagents: [ + { + id: taskId, + threadId: parentThreadId, + origin, + childThreadId, + updatedAt: now, + }, + ], + } as unknown as OrchestrationV2ThreadProjection; +} + +function childProjection( + input: { + readonly kind?: OrchestrationV2RuntimeRequest["kind"]; + readonly status?: OrchestrationV2RuntimeRequest["status"]; + readonly resumable?: boolean; + readonly runtimeMode?: "approval-required" | "auto-accept-edits" | "auto" | "full-access"; + readonly interactionMode?: "plan" | "default"; + } = {}, +) { + const status = input.status ?? "pending"; + return { + thread: { + id: childThreadId, + projectId, + runtimeMode: input.runtimeMode ?? "auto-accept-edits", + interactionMode: input.interactionMode ?? "default", + deletedAt: null, + }, + providerThreads: [ + { + id: providerThreadId, + driver: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + providerSessionId, + }, + ], + runtimeRequests: [ + { + id: requestId, + nodeId: requestNodeId, + providerTurnId: null, + nativeRequestRef: null, + kind: input.kind ?? "user_input", + status, + responseCapability: + input.resumable === false + ? { type: "not_resumable", reason: "The provider session ended." } + : { type: "live", providerSessionId }, + createdAt: now, + resolvedAt: status === "pending" ? null : now, + }, + ], + turnItems: [ + { + id: TurnItemId.make("turn-item:pending-request-question"), + threadId: childThreadId, + runId: null, + nodeId: requestNodeId, + providerThreadId, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: status === "pending" ? "waiting" : "completed", + title: null, + startedAt: now, + completedAt: status === "pending" ? null : now, + updatedAt: now, + type: "user_input_request", + requestId, + questions, + }, + ], + } as unknown as OrchestrationV2ThreadProjection; +} + +function serviceLayer(input: { + readonly getParent?: () => OrchestrationV2ThreadProjection; + readonly getChild: () => OrchestrationV2ThreadProjection; + readonly getReceipt?: ThreadManagementService["Service"]["getCommandReceipt"]; + readonly dispatch?: ThreadManagementService["Service"]["dispatch"]; +}) { + return layer.pipe( + Layer.provide( + Layer.mergeAll( + NodeServices.layer, + Layer.mock(ThreadManagementService)({ + getThreadProjection: () => Effect.succeed(input.getParent?.() ?? parentProjection()), + getProjectThread: ({ threadId }) => + threadId === childThreadId + ? Effect.succeed(input.getChild()) + : Effect.die(new Error(`Unexpected child projection read: ${threadId}`)), + getCommandReceipt: input.getReceipt ?? (() => Effect.succeed(Option.none())), + dispatch: + input.dispatch ?? + (() => Effect.succeed({ sequence: 1, storedEvents: [], replayed: false })), + }), + ), + ), + ); +} + +describe("PendingRequestMcpService", () => { + it.effect( + "lists and reads only structured user-input questions on direct app-owned children", + () => + Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const listed = yield* service.list(scope, {}); + assert.equal(listed.requests.length, 1); + assert.equal(listed.requests[0]?.taskId, taskId); + assert.equal(listed.requests[0]?.childThreadId, childThreadId); + assert.equal(listed.requests[0]?.requestId, requestId); + assert.deepEqual(listed.requests[0]?.questions, questions); + assert.equal(listed.nextCursor, null); + + const read = yield* service.read(scope, { childThreadId, requestId }); + assert.equal(read.providerInstanceId, "codex"); + assert.equal(read.driverKind, "codex"); + assert.equal(read.status, "pending"); + assert.isTrue(read.resumable); + }).pipe(Effect.provide(serviceLayer({ getChild: () => childProjection() }))), + ); + + it.effect("rejects provider-native children and non-user-input request kinds", () => + Effect.gen(function* () { + const providerOwnedService = yield* PendingRequestMcpService; + const wrongChild = yield* providerOwnedService + .read(scope, { childThreadId, requestId }) + .pipe(Effect.flip); + assert.equal(wrongChild.code, "child_not_found"); + }).pipe( + Effect.provide( + serviceLayer({ + getParent: () => parentProjection("provider_native"), + getChild: () => childProjection(), + }), + ), + Effect.andThen( + Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const wrongKind = yield* service + .read(scope, { childThreadId, requestId }) + .pipe(Effect.flip); + assert.equal(wrongKind.code, "wrong_request_kind"); + }).pipe( + Effect.provide( + serviceLayer({ getChild: () => childProjection({ kind: "file-change" }) }), + ), + ), + ), + ), + ); + + it.effect("answers every question once and replays the accepted durable receipt", () => + Effect.gen(function* () { + const projection = yield* Ref.make(childProjection()); + const acceptedReceipt = yield* Ref.make>(Option.none()); + const dispatch = vi.fn((command) => + Effect.gen(function* () { + assert.equal(command.type, "runtime-request.respond"); + if (command.type !== "runtime-request.respond") return assert.fail("wrong command"); + assert.equal(command.decision, undefined); + assert.deepEqual(command.answers, { editor: "Vim" }); + yield* Ref.set(projection, childProjection({ status: "resolved" })); + yield* Ref.set( + acceptedReceipt, + Option.some({ + commandId: command.commandId, + threadId: childThreadId, + commandType: command.type, + acceptedAt: now, + resultSequence: 7, + status: "accepted", + error: null, + }), + ); + return { sequence: 7, storedEvents: [], replayed: false }; + }), + ); + const testLayer = serviceLayer({ + getChild: () => Ref.getUnsafe(projection), + getReceipt: () => Ref.get(acceptedReceipt), + dispatch, + }); + + yield* Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const input = { + childThreadId, + requestId, + answers: { editor: "Vim" }, + clientRequestId: "answer-editor-once", + } as const; + const first = yield* service.respond(scope, input); + assert.equal(first.receiptSequence, 7); + assert.isFalse(first.replayed); + assert.equal(first.request.status, "resolved"); + + const replay = yield* service.respond(scope, input); + assert.equal(replay.commandId, first.commandId); + assert.equal(replay.receiptSequence, first.receiptSequence); + assert.isTrue(replay.replayed); + assert.equal(dispatch.mock.calls.length, 1); + }).pipe(Effect.provide(testLayer)); + }), + ); + + it.effect("reloads the child projection after an overlapping response becomes accepted", () => + Effect.gen(function* () { + const childReads = vi + .fn<() => OrchestrationV2ThreadProjection>() + .mockReturnValueOnce(childProjection()) + .mockReturnValue(childProjection({ status: "resolved" })); + const dispatch = vi.fn(() => Effect.die("accepted replay must not dispatch again")); + const acceptedReceipt: CommandReceiptV2 = { + commandId: "command:accepted-overlap" as CommandReceiptV2["commandId"], + threadId: childThreadId, + commandType: "runtime-request.respond", + acceptedAt: now, + resultSequence: 11, + status: "accepted", + error: null, + }; + + yield* Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const replay = yield* service.respond(scope, { + childThreadId, + requestId, + answers: { editor: "Vim" }, + clientRequestId: "accepted-overlap", + }); + assert.isTrue(replay.replayed); + assert.equal(replay.receiptSequence, 11); + assert.equal(replay.request.status, "resolved"); + assert.equal(childReads.mock.calls.length, 2); + assert.equal(dispatch.mock.calls.length, 0); + }).pipe( + Effect.provide( + serviceLayer({ + getChild: childReads, + getReceipt: () => Effect.succeed(Option.some(acceptedReceipt)), + dispatch, + }), + ), + ); + }), + ); + + it.effect("rejects incomplete answers, stale requests, and ended provider sessions", () => + Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const invalid = yield* service + .respond(scope, { + childThreadId, + requestId, + answers: { unknown: "value" }, + clientRequestId: "invalid-answers", + }) + .pipe(Effect.flip); + assert.equal(invalid.code, "invalid_answers"); + }).pipe( + Effect.provide(serviceLayer({ getChild: () => childProjection() })), + Effect.andThen( + Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const stale = yield* service + .respond(scope, { + childThreadId, + requestId, + answers: { editor: "Vim" }, + clientRequestId: "stale-request", + }) + .pipe(Effect.flip); + assert.equal(stale.code, "request_not_pending"); + }).pipe( + Effect.provide(serviceLayer({ getChild: () => childProjection({ status: "resolved" }) })), + ), + ), + Effect.andThen( + Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const unavailable = yield* service + .respond(scope, { + childThreadId, + requestId, + answers: { editor: "Vim" }, + clientRequestId: "ended-provider-session", + }) + .pipe(Effect.flip); + assert.equal(unavailable.code, "request_not_resumable"); + }).pipe( + Effect.provide(serviceLayer({ getChild: () => childProjection({ resumable: false }) })), + ), + ), + ), + ); + + it.effect("keeps delegated answers within the caller's runtime and interaction ceilings", () => + Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const runtimeError = yield* service + .respond(scope, { + childThreadId, + requestId, + answers: { editor: "Vim" }, + clientRequestId: "runtime-ceiling", + }) + .pipe(Effect.flip); + assert.equal(runtimeError.code, "runtime_mode_escalation_denied"); + }).pipe( + Effect.provide( + serviceLayer({ + getParent: () => parentProjection("app_owned", { runtimeMode: "approval-required" }), + getChild: () => childProjection({ runtimeMode: "full-access" }), + }), + ), + Effect.andThen( + Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const interactionError = yield* service + .respond(scope, { + childThreadId, + requestId, + answers: { editor: "Vim" }, + clientRequestId: "interaction-ceiling", + }) + .pipe(Effect.flip); + assert.equal(interactionError.code, "interaction_mode_escalation_denied"); + }).pipe( + Effect.provide( + serviceLayer({ + getParent: () => parentProjection("app_owned", { interactionMode: "plan" }), + getChild: () => childProjection({ interactionMode: "default" }), + }), + ), + ), + ), + ), + ); + + it.effect("never treats an unrelated thread id as an authorized child", () => + Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const error = yield* service + .read(scope, { childThreadId: unrelatedThreadId, requestId }) + .pipe(Effect.flip); + assert.equal(error.code, "child_not_found"); + }).pipe(Effect.provide(serviceLayer({ getChild: () => childProjection() }))), + ); +}); diff --git a/apps/server/src/mcp/PendingRequestMcpService.ts b/apps/server/src/mcp/PendingRequestMcpService.ts new file mode 100644 index 000000000000..395879f407d5 --- /dev/null +++ b/apps/server/src/mcp/PendingRequestMcpService.ts @@ -0,0 +1,462 @@ +import { + CommandId, + type OrchestrationV2RuntimeRequest, + type OrchestrationV2Subagent, + type OrchestrationV2ThreadProjection, + type OrchestrationV2TurnItem, + PendingRequestMcpFailure, + type PendingRequestMcpListInput, + type PendingRequestMcpListResult, + type PendingRequestMcpReadInput, + type PendingRequestMcpReadResult, + type PendingRequestMcpRequest, + type PendingRequestMcpRespondInput, + type PendingRequestMcpRespondResult, + type RuntimeRequestId, + type ThreadId, +} from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; + +import { + ThreadManagementService, + ThreadManagementThreadNotFoundError, +} from "../orchestration-v2/ThreadManagementService.ts"; +import { interactionModeWithinMcpCeiling, runtimeModeWithinMcpCeiling } from "./McpModeCeilings.ts"; +import type { McpInvocationScope } from "./McpInvocationContext.ts"; + +const DEFAULT_LIST_LIMIT = 20; + +function parseListCursor( + cursor: string | undefined, +): Effect.Effect< + { readonly childIndex: number; readonly requestIndex: number }, + PendingRequestMcpFailure +> { + if (cursor === undefined) return Effect.succeed({ childIndex: 0, requestIndex: 0 }); + const match = /^(0|[1-9]\d*):(0|[1-9]\d*)$/.exec(cursor); + if (match === null) { + return Effect.fail(failure("invalid_request", "The pending-request cursor is invalid.")); + } + const childIndex = Number(match[1]); + const requestIndex = Number(match[2]); + return Number.isSafeInteger(childIndex) && Number.isSafeInteger(requestIndex) + ? Effect.succeed({ childIndex, requestIndex }) + : Effect.fail(failure("invalid_request", "The pending-request cursor is invalid.")); +} + +export class PendingRequestMcpService extends Context.Service< + PendingRequestMcpService, + { + readonly list: ( + scope: McpInvocationScope, + input: PendingRequestMcpListInput, + ) => Effect.Effect; + readonly read: ( + scope: McpInvocationScope, + input: PendingRequestMcpReadInput, + ) => Effect.Effect; + readonly respond: ( + scope: McpInvocationScope, + input: PendingRequestMcpRespondInput, + ) => Effect.Effect; + } +>()("t3/mcp/PendingRequestMcpService") {} + +function failure( + code: PendingRequestMcpFailure["code"], + message: string, +): PendingRequestMcpFailure { + return new PendingRequestMcpFailure({ code, message }); +} + +function errorMessage(error: unknown): string { + if (typeof error === "object" && error !== null && "detail" in error) { + return String((error as { readonly detail: unknown }).detail); + } + if (typeof error === "object" && error !== null && "cause" in error) { + const cause = (error as { readonly cause: unknown }).cause; + if (typeof cause === "string") return cause; + if (cause instanceof Error) return cause.message; + } + return error instanceof Error ? error.message : String(error); +} + +const isThreadNotFound = Schema.is(ThreadManagementThreadNotFoundError); + +function projectionFailure(error: unknown, threadId: ThreadId): PendingRequestMcpFailure { + return isThreadNotFound(error) + ? failure("child_not_found", `Delegated child thread '${threadId}' was not found.`) + : failure( + "orchestration_error", + `Unable to load delegated child thread '${threadId}': ${errorMessage(error)}`, + ); +} + +function dispatchFailure(error: unknown): PendingRequestMcpFailure { + const tag = + typeof error === "object" && error !== null && "_tag" in error + ? String((error as { readonly _tag: unknown })._tag) + : ""; + return tag === "OrchestratorDispatchError" || + tag === "OrchestratorCommandPreviouslyRejectedError" || + tag === "OrchestratorCommandIdConflictError" + ? failure("operation_rejected", errorMessage(error)) + : failure("orchestration_error", errorMessage(error)); +} + +function directAppOwnedChildren(parent: OrchestrationV2ThreadProjection) { + return parent.subagents + .filter( + (task): task is OrchestrationV2Subagent & { readonly childThreadId: ThreadId } => + task.origin === "app_owned" && task.childThreadId !== null, + ) + .toSorted( + (left, right) => + DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt) || + right.id.localeCompare(left.id), + ); +} + +function findUserInputItem( + projection: OrchestrationV2ThreadProjection, + requestId: RuntimeRequestId, +): Extract | undefined { + return projection.turnItems.find( + (item): item is Extract => + item.type === "user_input_request" && item.requestId === requestId, + ); +} + +function requestSummary(input: { + readonly task: OrchestrationV2Subagent & { readonly childThreadId: ThreadId }; + readonly projection: OrchestrationV2ThreadProjection; + readonly request: OrchestrationV2RuntimeRequest; + readonly item: Extract; +}): Effect.Effect { + const providerThread = + input.item.providerThreadId === null + ? undefined + : input.projection.providerThreads.find( + (candidate) => candidate.id === input.item.providerThreadId, + ); + if (providerThread === undefined) { + return Effect.fail( + failure( + "orchestration_error", + `User-input request '${input.request.id}' has no durable provider thread.`, + ), + ); + } + return Effect.succeed({ + taskId: input.task.id, + childThreadId: input.task.childThreadId, + runId: input.item.runId, + nodeId: input.request.nodeId, + requestId: input.request.id, + providerInstanceId: providerThread.providerInstanceId, + driverKind: providerThread.driver, + status: input.request.status, + resumable: input.request.responseCapability.type === "live", + questions: input.item.questions, + createdAt: DateTime.formatIso(input.request.createdAt), + resolvedAt: + input.request.resolvedAt === null ? null : DateTime.formatIso(input.request.resolvedAt), + }); +} + +function stablePart(value: string): Effect.Effect { + try { + return Effect.succeed(encodeURIComponent(value)); + } catch { + return Effect.fail( + failure( + "invalid_request", + "clientRequestId contains invalid Unicode and cannot be used for retry identity.", + ), + ); + } +} + +export const make = Effect.gen(function* () { + const crypto = yield* Crypto.Crypto; + const threadManagement = yield* ThreadManagementService; + + const requireCapability = (scope: McpInvocationScope) => + scope.capabilities.has("orchestration") + ? Effect.void + : Effect.fail( + failure( + "capability_denied", + "This MCP credential does not grant orchestration capabilities.", + ), + ); + + const loadParent = (scope: McpInvocationScope) => + Effect.gen(function* () { + yield* requireCapability(scope); + const parent = yield* threadManagement + .getThreadProjection(scope.threadId) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to load calling thread '${scope.threadId}': ${errorMessage(error)}`, + ), + ), + ); + if (parent.thread.deletedAt !== null) { + return yield* failure("child_not_found", "The calling thread is no longer active."); + } + return parent; + }); + + const loadChild = (parent: OrchestrationV2ThreadProjection, childThreadId: ThreadId) => + Effect.gen(function* () { + const task = directAppOwnedChildren(parent).find( + (candidate) => candidate.childThreadId === childThreadId, + ); + if (task === undefined) { + return yield* failure( + "child_not_found", + `Thread '${childThreadId}' is not a direct app-owned delegated child of '${parent.thread.id}'.`, + ); + } + const projection = yield* threadManagement + .getProjectThread({ projectId: parent.thread.projectId, threadId: childThreadId }) + .pipe(Effect.mapError((error) => projectionFailure(error, childThreadId))); + return { task, projection } as const; + }); + + const findRequest = (input: { + readonly task: OrchestrationV2Subagent & { readonly childThreadId: ThreadId }; + readonly projection: OrchestrationV2ThreadProjection; + readonly requestId: RuntimeRequestId; + }) => + Effect.gen(function* () { + const request = input.projection.runtimeRequests.find( + (candidate) => candidate.id === input.requestId, + ); + if (request === undefined) { + return yield* failure( + "request_not_found", + `Runtime request '${input.requestId}' was not found on delegated child '${input.task.childThreadId}'.`, + ); + } + if (request.kind !== "user_input") { + return yield* failure( + "wrong_request_kind", + `Runtime request '${request.id}' is '${request.kind}', not a user-input question.`, + ); + } + const item = findUserInputItem(input.projection, request.id); + if (item === undefined) { + return yield* failure( + "orchestration_error", + `User-input request '${request.id}' has no durable question details.`, + ); + } + return { request, item } as const; + }); + + const commandId = (input: { + readonly scope: McpInvocationScope; + readonly childThreadId: ThreadId; + readonly requestId: RuntimeRequestId; + readonly clientRequestId: string | undefined; + }) => + Effect.gen(function* () { + const requestKey = input.clientRequestId ?? (yield* crypto.randomUUIDv4.pipe(Effect.orDie)); + const parts = yield* Effect.all( + [input.scope.providerSessionId, input.childThreadId, input.requestId, requestKey].map( + stablePart, + ), + ); + return CommandId.make(["command", "mcp", "pending-request", ...parts].join(":")); + }); + + const requireResponseCeiling = ( + caller: OrchestrationV2ThreadProjection, + target: OrchestrationV2ThreadProjection, + ) => { + if (!runtimeModeWithinMcpCeiling(caller.thread.runtimeMode, target.thread.runtimeMode)) { + return Effect.fail( + failure( + "runtime_mode_escalation_denied", + `Target runtime mode ${target.thread.runtimeMode} is broader than caller mode ${caller.thread.runtimeMode}.`, + ), + ); + } + return interactionModeWithinMcpCeiling( + caller.thread.interactionMode, + target.thread.interactionMode, + ) + ? Effect.void + : Effect.fail( + failure( + "interaction_mode_escalation_denied", + `Target interaction mode ${target.thread.interactionMode} is broader than caller mode ${caller.thread.interactionMode}.`, + ), + ); + }; + + const resultFor = (input: { + readonly task: OrchestrationV2Subagent & { readonly childThreadId: ThreadId }; + readonly projection: OrchestrationV2ThreadProjection; + readonly requestId: RuntimeRequestId; + }) => + findRequest(input).pipe( + Effect.flatMap(({ request, item }) => + requestSummary({ task: input.task, projection: input.projection, request, item }), + ), + ); + + return PendingRequestMcpService.of({ + list: (scope, input) => + Effect.gen(function* () { + const parent = yield* loadParent(scope); + const children = directAppOwnedChildren(parent); + const limit = input.limit ?? DEFAULT_LIST_LIMIT; + const cursor = yield* parseListCursor(input.cursor); + const requests: Array = []; + let childIndex = cursor.childIndex; + let requestIndex = cursor.requestIndex; + while (childIndex < children.length && requests.length < limit) { + const task = children[childIndex]; + if (task === undefined) break; + const projection = yield* threadManagement + .getProjectThread({ + projectId: parent.thread.projectId, + threadId: task.childThreadId, + }) + .pipe(Effect.mapError((error) => projectionFailure(error, task.childThreadId))); + const pending = projection.runtimeRequests.filter( + (request) => request.kind === "user_input" && request.status === "pending", + ); + while (requestIndex < pending.length && requests.length < limit) { + const request = pending[requestIndex]; + if (request === undefined) break; + requests.push(yield* resultFor({ task, projection, requestId: request.id })); + requestIndex += 1; + } + if (requestIndex < pending.length) break; + childIndex += 1; + requestIndex = 0; + } + return { + requests, + nextCursor: childIndex < children.length ? `${childIndex}:${requestIndex}` : null, + } satisfies PendingRequestMcpListResult; + }), + read: (scope, input) => + Effect.gen(function* () { + const parent = yield* loadParent(scope); + const child = yield* loadChild(parent, input.childThreadId); + return yield* resultFor({ ...child, requestId: input.requestId }); + }), + respond: (scope, input) => + Effect.gen(function* () { + const parent = yield* loadParent(scope); + let child = yield* loadChild(parent, input.childThreadId); + const id = yield* commandId({ + scope, + childThreadId: input.childThreadId, + requestId: input.requestId, + clientRequestId: input.clientRequestId, + }); + const receipt = yield* threadManagement + .getCommandReceipt(id) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to inspect response receipt '${id}': ${errorMessage(error)}`, + ), + ), + ); + if (Option.isSome(receipt) && receipt.value.status === "accepted") { + if ( + receipt.value.threadId !== input.childThreadId || + receipt.value.commandType !== "runtime-request.respond" + ) { + return yield* failure( + "operation_rejected", + `Command '${id}' was already used for another operation.`, + ); + } + child = yield* loadChild(parent, input.childThreadId); + const request = yield* resultFor({ ...child, requestId: input.requestId }); + return { + commandId: id, + receiptSequence: receipt.value.resultSequence, + replayed: true, + request, + } satisfies PendingRequestMcpRespondResult; + } + if (Option.isSome(receipt)) { + return yield* failure( + "operation_rejected", + receipt.value.error ?? `Command '${id}' was previously rejected.`, + ); + } + + yield* requireResponseCeiling(parent, child.projection); + + const { request, item } = yield* findRequest({ ...child, requestId: input.requestId }); + if (request.status !== "pending") { + return yield* failure( + "request_not_pending", + `User-input request '${request.id}' is ${request.status}.`, + ); + } + if (request.responseCapability.type !== "live") { + return yield* failure("request_not_resumable", request.responseCapability.reason); + } + const expectedIds = new Set(item.questions.map((question) => question.id)); + const suppliedIds = Object.keys(input.answers); + const missing = [...expectedIds].filter((id) => !(id in input.answers)); + const unknown = suppliedIds.filter((answerId) => !expectedIds.has(answerId)); + if (missing.length > 0 || unknown.length > 0) { + return yield* failure( + "invalid_answers", + [ + missing.length === 0 ? "" : `Missing question IDs: ${missing.join(", ")}.`, + unknown.length === 0 ? "" : `Unknown question IDs: ${unknown.join(", ")}.`, + ] + .filter(Boolean) + .join(" "), + ); + } + + const dispatch = yield* threadManagement + .dispatch({ + type: "runtime-request.respond", + commandId: id, + threadId: input.childThreadId, + requestId: input.requestId, + answers: input.answers, + policyCeiling: { + callerThreadId: parent.thread.id, + runtimeMode: parent.thread.runtimeMode, + interactionMode: parent.thread.interactionMode, + }, + }) + .pipe(Effect.mapError(dispatchFailure)); + child = yield* loadChild(parent, input.childThreadId); + const resolved = yield* resultFor({ ...child, requestId: input.requestId }); + return { + commandId: id, + receiptSequence: dispatch.sequence, + replayed: dispatch.replayed ?? false, + request: resolved, + } satisfies PendingRequestMcpRespondResult; + }), + }); +}); + +export const layer = Layer.effect(PendingRequestMcpService, make); diff --git a/apps/server/src/mcp/toolkits/pendingRequest/handlers.ts b/apps/server/src/mcp/toolkits/pendingRequest/handlers.ts new file mode 100644 index 000000000000..07b7f98d41cc --- /dev/null +++ b/apps/server/src/mcp/toolkits/pendingRequest/handlers.ts @@ -0,0 +1,20 @@ +import { McpInvocationContext } from "../../McpInvocationContext.ts"; +import { PendingRequestMcpService } from "../../PendingRequestMcpService.ts"; +import { PendingRequestToolkit } from "./tools.ts"; + +const handlers = { + t3_pending_request_list: (input) => + PendingRequestMcpService.use((service) => + McpInvocationContext.use((scope) => service.list(scope, input)), + ), + t3_pending_request_read: (input) => + PendingRequestMcpService.use((service) => + McpInvocationContext.use((scope) => service.read(scope, input)), + ), + t3_pending_request_respond: (input) => + PendingRequestMcpService.use((service) => + McpInvocationContext.use((scope) => service.respond(scope, input)), + ), +} satisfies Parameters[0]; + +export const PendingRequestToolkitHandlersLive = PendingRequestToolkit.toLayer(handlers); diff --git a/apps/server/src/mcp/toolkits/pendingRequest/tools.ts b/apps/server/src/mcp/toolkits/pendingRequest/tools.ts new file mode 100644 index 000000000000..963ce34f7ef8 --- /dev/null +++ b/apps/server/src/mcp/toolkits/pendingRequest/tools.ts @@ -0,0 +1,61 @@ +import { + PendingRequestMcpFailure, + PendingRequestMcpListInput, + PendingRequestMcpListResult, + PendingRequestMcpReadInput, + PendingRequestMcpReadResult, + PendingRequestMcpRespondInput, + PendingRequestMcpRespondResult, +} from "@t3tools/contracts"; +import { Tool, Toolkit } from "effect/unstable/ai"; + +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import { PendingRequestMcpService } from "../../PendingRequestMcpService.ts"; + +const dependencies = [McpInvocationContext.McpInvocationContext, PendingRequestMcpService]; + +export const PendingRequestListTool = Tool.make("t3_pending_request_list", { + description: + "List pending user-input questions from direct app-owned delegated children of the calling thread. Results are bounded and never include approval requests or unrelated threads.", + parameters: PendingRequestMcpListInput, + success: PendingRequestMcpListResult, + failure: PendingRequestMcpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "List delegated user-input requests") + .annotate(Tool.Readonly, true) + .annotate(Tool.Destructive, false) + .annotate(Tool.Idempotent, true); + +export const PendingRequestReadTool = Tool.make("t3_pending_request_read", { + description: + "Read one structured user-input question from a direct app-owned delegated child. Resolved and stale requests remain readable; this tool never responds or approves anything.", + parameters: PendingRequestMcpReadInput, + success: PendingRequestMcpReadResult, + failure: PendingRequestMcpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "Read a delegated user-input request") + .annotate(Tool.Readonly, true) + .annotate(Tool.Destructive, false) + .annotate(Tool.Idempotent, true); + +export const PendingRequestRespondTool = Tool.make("t3_pending_request_respond", { + description: + "Answer a pending user-input question from a direct app-owned delegated child. Supply every exact question ID. This cannot answer approval requests, grant permissions, or target the calling agent or an unrelated thread. Acceptance is reported by a durable V2 command receipt; provider delivery may complete asynchronously.", + parameters: PendingRequestMcpRespondInput, + success: PendingRequestMcpRespondResult, + failure: PendingRequestMcpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "Answer a delegated user-input request") + .annotate(Tool.Destructive, true); + +export const PendingRequestToolkit = Toolkit.make( + PendingRequestListTool, + PendingRequestReadTool, + PendingRequestRespondTool, +); diff --git a/apps/server/src/mcp/toolkits/worktree/registration.test.ts b/apps/server/src/mcp/toolkits/worktree/registration.test.ts index 3300a869fe67..68723fec49c8 100644 --- a/apps/server/src/mcp/toolkits/worktree/registration.test.ts +++ b/apps/server/src/mcp/toolkits/worktree/registration.test.ts @@ -114,6 +114,9 @@ it.effect("production mcp layer lists worktree tools over http", () => // than replacing them. expect(toolNames).toContain("preview_status"); expect(toolNames).toContain("delegate_task"); + expect(toolNames).toContain("t3_pending_request_list"); + expect(toolNames).toContain("t3_pending_request_read"); + expect(toolNames).toContain("t3_pending_request_respond"); // The handoff tool mutates thread state, reaches the network (origin // fetch), and runs project setup scripts, so its MCP hints must not @@ -125,6 +128,12 @@ it.effect("production mcp layer lists worktree tools over http", () => const status = tools.find((tool) => tool.name === "t3_worktree_status"); expect(status?.annotations?.readOnlyHint).toBe(true); expect(status?.annotations?.destructiveHint).toBe(false); + const pendingList = tools.find((tool) => tool.name === "t3_pending_request_list"); + expect(pendingList?.annotations?.readOnlyHint).toBe(true); + expect(pendingList?.annotations?.destructiveHint).toBe(false); + const pendingRespond = tools.find((tool) => tool.name === "t3_pending_request_respond"); + expect(pendingRespond?.annotations?.readOnlyHint).toBe(false); + expect(pendingRespond?.annotations?.destructiveHint).toBe(true); // MCP requires every tool input schema to be a top-level object schema. // A non-object schema (e.g. the anyOf produced by an empty diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 5ebbec7a2155..5ab667c52769 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -40,6 +40,7 @@ import { Tool } from "effect/unstable/ai"; import { attachmentRelativePath } from "../../attachmentStore.ts"; import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; import { OrchestratorToolkit } from "../../mcp/toolkits/orchestrator/tools.ts"; +import { PendingRequestToolkit } from "../../mcp/toolkits/pendingRequest/tools.ts"; import type { EventNdjsonLogger } from "../../provider/Layers/EventNdjsonLogger.ts"; import { ProviderAdapterV2RuntimePolicy, @@ -492,8 +493,11 @@ describe("ClaudeAdapterV2 MCP query overrides", () => { }); }); - it("matches the read-only allowlist to the orchestrator toolkit annotations", () => { - const readOnlyToolNames = Object.values(OrchestratorToolkit.tools) + it("matches the read-only allowlist to the registered toolkit annotations", () => { + const readOnlyToolNames = [ + ...Object.values(OrchestratorToolkit.tools), + ...Object.values(PendingRequestToolkit.tools), + ] .filter((tool) => Context.get(tool.annotations, Tool.Readonly)) .map((tool) => `mcp__t3-code__${tool.name}`) .sort(); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index a52f220ee4b2..ad89ccc50a8a 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -722,13 +722,15 @@ export function makeClaudeQueryOptions(input: { export const CLAUDE_T3_MCP_TOOL_WILDCARD = "mcp__t3-code__*"; -// Must stay in sync with the Tool.Readonly annotations on OrchestratorToolkit; -// ClaudeAdapterV2.test.ts cross-checks this list against the toolkit. +// Must stay in sync with the Tool.Readonly annotations on the toolkits below; +// ClaudeAdapterV2.test.ts cross-checks this list against those toolkits. export const CLAUDE_READ_ONLY_T3_MCP_ALLOWED_TOOLS: ReadonlyArray = [ "mcp__t3-code__orchestrator_capabilities", "mcp__t3-code__list_scheduled_tasks", "mcp__t3-code__t3_thread_list", "mcp__t3-code__t3_thread_wait", + "mcp__t3-code__t3_pending_request_list", + "mcp__t3-code__t3_pending_request_read", ]; // The SDK's `allowedTools` only pre-approves tool calls; availability is the diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 6c5ae7bc7b58..83e6cc0bfcbb 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -169,6 +169,7 @@ export type OrchestratorV2Error = typeof OrchestratorV2Error.Type; export interface OrchestratorV2DispatchResult { readonly sequence: number; readonly storedEvents: ReadonlyArray; + readonly replayed?: boolean; } export interface OrchestratorV2Shape { @@ -176,6 +177,7 @@ export interface OrchestratorV2Shape { readonly dispatch: ( command: OrchestrationV2Command, ) => Effect.Effect; + readonly getCommandReceipt: CommandReceiptStoreV2["Service"]["getByCommandId"]; readonly getThreadProjection: ( threadId: ThreadId, ) => Effect.Effect; @@ -341,6 +343,23 @@ function nextQueuedRun( return queuedRunsInDeliveryOrder(projection)[0]; } +function runtimeModeRank(mode: OrchestrationV2AppThread["runtimeMode"]): number { + switch (mode) { + case "approval-required": + return 0; + case "auto-accept-edits": + return 1; + case "auto": + return 2; + case "full-access": + return 3; + } +} + +function interactionModeRank(mode: OrchestrationV2AppThread["interactionMode"]): number { + return mode === "plan" ? 0 : 1; +} + function latestStableRun(projection: OrchestrationV2ThreadProjection): OrchestrationV2Run | null { return ( projection.runs @@ -523,6 +542,68 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const threadForkService = yield* ThreadForkServiceV2; const threadDispatch = yield* makeKeyedSerialExecutor(); + const enforceRuntimeRequestPolicyCeiling = ( + command: Extract, + targetProjection: OrchestrationV2ThreadProjection, + ) => + Effect.gen(function* () { + const ceiling = command.policyCeiling; + if (ceiling === undefined) return; + const callerProjection = + ceiling.callerThreadId === command.threadId + ? targetProjection + : yield* projectionStore.getThreadProjection(ceiling.callerThreadId).pipe( + Effect.mapError( + (cause) => + new OrchestratorProjectionError({ + threadId: ceiling.callerThreadId, + cause, + }), + ), + ); + const authorizedChild = callerProjection.subagents.some( + (task) => + task.origin === "app_owned" && + task.threadId === ceiling.callerThreadId && + task.childThreadId === command.threadId, + ); + if ( + callerProjection.thread.deletedAt !== null || + callerProjection.thread.projectId !== targetProjection.thread.projectId || + !authorizedChild + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Caller thread ${ceiling.callerThreadId} does not own the target as an active app-owned delegated child.`, + }); + } + if ( + runtimeModeRank(targetProjection.thread.runtimeMode) > + runtimeModeRank(ceiling.runtimeMode) || + runtimeModeRank(targetProjection.thread.runtimeMode) > + runtimeModeRank(callerProjection.thread.runtimeMode) + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Target runtime mode ${targetProjection.thread.runtimeMode} exceeds the caller ceiling.`, + }); + } + if ( + interactionModeRank(targetProjection.thread.interactionMode) > + interactionModeRank(ceiling.interactionMode) || + interactionModeRank(targetProjection.thread.interactionMode) > + interactionModeRank(callerProjection.thread.interactionMode) + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Target interaction mode ${targetProjection.thread.interactionMode} exceeds the caller ceiling.`, + }); + } + }); + const mapDispatchError = (command: OrchestrationV2Command) => (effect: Effect.Effect): Effect.Effect => @@ -4964,6 +5045,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio .pipe( Effect.mapError(() => new OrchestratorProjectionError({ threadId: command.threadId })), ); + yield* enforceRuntimeRequestPolicyCeiling(command, projection); const runtimeRequest = projection.runtimeRequests.find( (candidate) => candidate.id === command.requestId, ); @@ -6904,6 +6986,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return { sequence: receipt.resultSequence, storedEvents, + replayed: true, } satisfies OrchestratorV2DispatchResult; } @@ -6983,11 +7066,30 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return { sequence: committed.receipt.resultSequence, storedEvents: committed.storedEvents, + replayed: !committed.committed, } satisfies OrchestratorV2DispatchResult; }); + const dispatchLockKeys = (command: OrchestrationV2Command): ReadonlyArray => { + const keys = + command.type === "runtime-request.respond" && command.policyCeiling !== undefined + ? [command.threadId, command.policyCeiling.callerThreadId] + : [commandThreadId(command)]; + return [...new Set(keys)].toSorted((left, right) => (left < right ? -1 : left > right ? 1 : 0)); + }; + + const withDispatchLocks = ( + keys: ReadonlyArray, + effect: Effect.Effect, + ): Effect.Effect => { + const [key, ...remaining] = keys; + return key === undefined + ? effect + : threadDispatch.withLock(key, withDispatchLocks(remaining, effect)); + }; + const dispatchWithReceipt = (command: OrchestrationV2Command) => - threadDispatch.withLock(commandThreadId(command), dispatchWithReceiptEffect(command)); + withDispatchLocks(dispatchLockKeys(command), dispatchWithReceiptEffect(command)); const handleTerminalRun = (stored: OrchestrationV2StoredEvent) => Effect.gen(function* () { @@ -7143,6 +7245,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return OrchestratorV2.of({ resumeQueuedRuns, dispatch: dispatchWithReceipt, + getCommandReceipt: commandReceipts.getByCommandId, getThreadProjection: (threadId) => projectionStore .getThreadProjection(threadId) @@ -7241,6 +7344,7 @@ export const layerUnavailable: Layer.Layer = Layer.succeed( cause: "Orchestration V2 live runtime is not configured.", }), ), + getCommandReceipt: () => Effect.succeed(Option.none()), getThreadProjection: (threadId) => Effect.fail( new OrchestratorProjectionError({ diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 77e52bffbbbe..42a1291f5e4d 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -271,6 +271,7 @@ export interface ThreadManagementServiceShape { readonly dispatch: ( command: OrchestrationV2Command, ) => Effect.Effect; + readonly getCommandReceipt: OrchestratorV2["Service"]["getCommandReceipt"]; readonly getThreadProjection: ( threadId: ThreadId, ) => Effect.Effect; @@ -645,6 +646,7 @@ const make = Effect.gen(function* () { return ThreadManagementService.of({ ensureLegacyTranscript, dispatch, + getCommandReceipt: orchestrator.getCommandReceipt, getThreadProjection, getThreadSnapshot, getProjectThread, diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 3aa9a6cb401a..9237f2d08871 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -140,7 +140,7 @@ selection model-visible without allowing a request that cannot run. ## Tool Surface -The server exposes eleven orchestration tools. +The server exposes orchestration, thread, and delegated-request tools. ### `orchestrator_capabilities` @@ -218,6 +218,32 @@ Interrupts the active child run through the normal V2 `run.interrupt` command. It is idempotent for terminal tasks and accepts an optional cancellation reason. +### `t3_pending_request_list` + +Lists pending structured user-input questions from direct app-owned delegated +children of the calling thread. The bounded page follows the parent's durable +sub-agent records and never scans unrelated or provider-native child threads. +Approval and permission requests are excluded. + +### `t3_pending_request_read` + +Reads one structured user-input request by exact child-thread and request IDs. +Unlike the list path, read can return resolved, expired, or cancelled requests +so callers can distinguish a stale response attempt from a missing request. +The read path is non-mutating. + +### `t3_pending_request_respond` + +Answers a pending, live `user_input` request on a direct app-owned delegated +child. The input is an answers record keyed by every question ID returned from +list/read. It dispatches the normal V2 `runtime-request.respond` command with +answers only and returns its durable receipt sequence. It cannot accept or +deny permission requests, approve a tool call, answer a provider-native child, +or target an unrelated thread. `clientRequestId` makes an accepted response +safe to replay without calling the provider twice. The child's runtime and +interaction modes must remain within the caller's captured and current mode +ceilings when the serialized response decision commits. + ### `create_threads` Creates between one and twenty ordinary top-level T3 threads: @@ -318,6 +344,7 @@ provider model -> parent/child execution nodes -> consumed subagent_spawn context transfer -> normal provider effect and runtime ingestion + -> optional user-input request projected for parent list/read/respond -> child run reaches a terminal state -> parent subagent/node/turn item finalized -> consumed subagent_result context transfer diff --git a/docs/user/permission-modes.md b/docs/user/permission-modes.md index 9bf9c10b20f5..17baed8077c3 100644 --- a/docs/user/permission-modes.md +++ b/docs/user/permission-modes.md @@ -26,6 +26,11 @@ unattended until it finishes or asks a question of its own. Approvals appear inline in the conversation. Approve or reject one and the agent continues from there. +When an agent delegates work to a T3-owned child conversation, it can relay that child's +structured questions and send your answers back. This applies only to user-input questions from +that direct child. It does not let either agent approve permission requests or expand the child's +permission mode. + For Grok, **Always allow this session** remembers the matching command or tool input. Other actions still ask for approval. It does not change the thread to **Full access**. diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index cfe24f5ffd22..b87829defbec 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -28,6 +28,7 @@ export * from "./orchestrationProject.ts"; export * from "./orchestrationV2.ts"; export * from "./applicationEvent.ts"; export * from "./orchestratorMcp.ts"; +export * from "./pendingRequestMcp.ts"; export * from "./orchestration.ts"; export * from "./t3ProjectFile.ts"; export * from "./editor.ts"; diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 801ed5f4d29a..2f3ca77008e6 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -2211,6 +2211,14 @@ export const OrchestrationV2Command = Schema.Union([ requestId: RuntimeRequestId, decision: Schema.optional(ProviderApprovalDecision), answers: Schema.optional(ProviderUserInputAnswers), + /** Optional MCP caller ceiling enforced against fresh caller and target projections. */ + policyCeiling: Schema.optional( + Schema.Struct({ + callerThreadId: ThreadId, + runtimeMode: RuntimeMode, + interactionMode: ProviderInteractionMode, + }), + ), }), Schema.Struct({ type: Schema.Literal("checkpoint.rollback"), diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 7e62e9351d31..b1c76db040d6 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -460,6 +460,7 @@ export const OrchestratorMcpCapabilitiesResult = Schema.Struct({ threadManagement: Schema.Boolean, incrementalThreadRead: Schema.Boolean, scheduledTasks: Schema.Boolean, + delegatedUserInputRequests: Schema.optional(Schema.Boolean), maxBatchThreads: Schema.Number, }), }); diff --git a/packages/contracts/src/pendingRequestMcp.test.ts b/packages/contracts/src/pendingRequestMcp.test.ts new file mode 100644 index 000000000000..166f94b3fd39 --- /dev/null +++ b/packages/contracts/src/pendingRequestMcp.test.ts @@ -0,0 +1,50 @@ +import { describe, expect, it } from "@effect/vitest"; +import * as Schema from "effect/Schema"; + +import { + PendingRequestMcpListInput, + PendingRequestMcpReadInput, + PendingRequestMcpRespondInput, +} from "./pendingRequestMcp.ts"; + +const decodeList = Schema.decodeUnknownSync(PendingRequestMcpListInput); +const decodeRead = Schema.decodeUnknownSync(PendingRequestMcpReadInput); +const decodeRespond = Schema.decodeUnknownSync(PendingRequestMcpRespondInput); + +describe("pending-request MCP contracts", () => { + it("keeps every operation discoverable as a root object schema", () => { + for (const schema of [ + PendingRequestMcpListInput, + PendingRequestMcpReadInput, + PendingRequestMcpRespondInput, + ]) { + expect(Schema.toJsonSchemaDocument(schema).schema.type).toBe("object"); + } + }); + + it("accepts structured single and multiple-choice answers", () => { + expect( + decodeRespond({ + childThreadId: "thread:child", + requestId: "request:questions", + answers: { + editor: "vim", + features: ["queue controls", "attachments"], + }, + clientRequestId: "answer-questions-1", + }).answers, + ).toEqual({ + editor: "vim", + features: ["queue controls", "attachments"], + }); + }); + + it("bounds list pages and requires stable target identifiers", () => { + expect(decodeList({ limit: 50, cursor: "2:1" })).toEqual({ limit: 50, cursor: "2:1" }); + expect(() => decodeList({ limit: 51 })).toThrow(); + expect(decodeRead({ childThreadId: "thread:child", requestId: "request:questions" })).toEqual({ + childThreadId: "thread:child", + requestId: "request:questions", + }); + }); +}); diff --git a/packages/contracts/src/pendingRequestMcp.ts b/packages/contracts/src/pendingRequestMcp.ts new file mode 100644 index 000000000000..56972387e78d --- /dev/null +++ b/packages/contracts/src/pendingRequestMcp.ts @@ -0,0 +1,103 @@ +import * as Schema from "effect/Schema"; + +import { + CommandId, + IsoDateTime, + NodeId, + NonNegativeInt, + PositiveInt, + RunId, + RuntimeRequestId, + ThreadId, + TrimmedNonEmptyString, +} from "./baseSchemas.ts"; +import { OrchestrationV2UserInputQuestion } from "./orchestrationV2.ts"; +import { ProviderDriverKind, ProviderInstanceId } from "./providerInstance.ts"; + +const PendingRequestMcpLimit = PositiveInt.check(Schema.isLessThanOrEqualTo(50)); +const PendingRequestMcpClientRequestId = TrimmedNonEmptyString.check( + Schema.isMaxLength(256), +).annotate({ description: "Stable idempotency key to reuse when retrying this response." }); +const PendingRequestMcpAnswer = Schema.Union([ + TrimmedNonEmptyString, + Schema.Array(TrimmedNonEmptyString).check(Schema.isMinLength(1), Schema.isMaxLength(20)), +]); + +export const PendingRequestMcpListInput = Schema.Struct({ + cursor: Schema.optional(TrimmedNonEmptyString.check(Schema.isMaxLength(64))).annotate({ + description: "Opaque cursor returned by the previous list page.", + }), + limit: Schema.optional(PendingRequestMcpLimit), +}); +export type PendingRequestMcpListInput = typeof PendingRequestMcpListInput.Type; + +export const PendingRequestMcpReadInput = Schema.Struct({ + childThreadId: ThreadId, + requestId: RuntimeRequestId, +}); +export type PendingRequestMcpReadInput = typeof PendingRequestMcpReadInput.Type; + +export const PendingRequestMcpRespondInput = Schema.Struct({ + childThreadId: ThreadId, + requestId: RuntimeRequestId, + answers: Schema.Record(TrimmedNonEmptyString, PendingRequestMcpAnswer).annotate({ + description: + "Answers keyed by the exact question IDs returned by list/read. Every question must be answered; values may be one string or a non-empty string array.", + }), + clientRequestId: Schema.optional(PendingRequestMcpClientRequestId), +}); +export type PendingRequestMcpRespondInput = typeof PendingRequestMcpRespondInput.Type; + +export const PendingRequestMcpRequest = Schema.Struct({ + taskId: NodeId, + childThreadId: ThreadId, + runId: Schema.NullOr(RunId), + nodeId: NodeId, + requestId: RuntimeRequestId, + providerInstanceId: ProviderInstanceId, + driverKind: ProviderDriverKind, + status: Schema.Literals(["pending", "resolved", "expired", "cancelled"]), + resumable: Schema.Boolean, + questions: Schema.Array(OrchestrationV2UserInputQuestion), + createdAt: IsoDateTime, + resolvedAt: Schema.NullOr(IsoDateTime), +}); +export type PendingRequestMcpRequest = typeof PendingRequestMcpRequest.Type; + +export const PendingRequestMcpListResult = Schema.Struct({ + requests: Schema.Array(PendingRequestMcpRequest), + nextCursor: Schema.NullOr(TrimmedNonEmptyString), +}); +export type PendingRequestMcpListResult = typeof PendingRequestMcpListResult.Type; + +export const PendingRequestMcpReadResult = PendingRequestMcpRequest; +export type PendingRequestMcpReadResult = typeof PendingRequestMcpReadResult.Type; + +export const PendingRequestMcpRespondResult = Schema.Struct({ + commandId: CommandId, + receiptSequence: NonNegativeInt, + replayed: Schema.Boolean, + request: PendingRequestMcpRequest, +}); +export type PendingRequestMcpRespondResult = typeof PendingRequestMcpRespondResult.Type; + +export class PendingRequestMcpFailure extends Schema.TaggedErrorClass()( + "PendingRequestMcpFailure", + { + code: Schema.Literals([ + "capability_denied", + "child_not_found", + "request_not_found", + "wrong_request_kind", + "request_not_pending", + "request_not_resumable", + "runtime_mode_escalation_denied", + "interaction_mode_escalation_denied", + "invalid_request", + "invalid_answers", + "operation_rejected", + "orchestration_error", + ]), + message: Schema.String, + }, +) {} diff --git a/packages/shared/src/t3McpToolPresentation.ts b/packages/shared/src/t3McpToolPresentation.ts index eedebd0a3aa1..d06ed6a97808 100644 --- a/packages/shared/src/t3McpToolPresentation.ts +++ b/packages/shared/src/t3McpToolPresentation.ts @@ -23,6 +23,9 @@ const T3_MCP_TOOL_DISPLAY_NAMES: Record = { t3_thread_send: "Send to a T3 thread", t3_thread_wait: "Wait for a T3 thread", t3_thread_interrupt: "Interrupt a T3 thread", + t3_pending_request_list: "List delegated user-input requests", + t3_pending_request_read: "Read a delegated user-input request", + t3_pending_request_respond: "Answer a delegated user-input request", t3_worktree_handoff: "Hand off thread to a git worktree", t3_worktree_status: "Get thread worktree status", preview_status: "Get preview browser status", From f7571890e7fbdfc08af9507f2762b31be96d220c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 29 Aug 2026 18:12:20 -0700 Subject: [PATCH 2/5] fix(mcp): redact pending request defects --- .../src/mcp/PendingRequestMcpService.test.ts | 32 ++++++++++++++++++- .../src/mcp/PendingRequestMcpService.ts | 1 - 2 files changed, 31 insertions(+), 2 deletions(-) diff --git a/apps/server/src/mcp/PendingRequestMcpService.test.ts b/apps/server/src/mcp/PendingRequestMcpService.test.ts index d3b5ddd88314..9cbf3d77831d 100644 --- a/apps/server/src/mcp/PendingRequestMcpService.test.ts +++ b/apps/server/src/mcp/PendingRequestMcpService.test.ts @@ -20,6 +20,7 @@ import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import type { CommandReceiptV2 } from "../orchestration-v2/CommandReceiptStore.ts"; +import { OrchestratorProjectionError } from "../orchestration-v2/Orchestrator.ts"; import { ThreadManagementService } from "../orchestration-v2/ThreadManagementService.ts"; import type { McpInvocationScope } from "./McpInvocationContext.ts"; import { layer, PendingRequestMcpService } from "./PendingRequestMcpService.ts"; @@ -152,6 +153,7 @@ function childProjection( function serviceLayer(input: { readonly getParent?: () => OrchestrationV2ThreadProjection; readonly getChild: () => OrchestrationV2ThreadProjection; + readonly getThreadProjection?: ThreadManagementService["Service"]["getThreadProjection"]; readonly getReceipt?: ThreadManagementService["Service"]["getCommandReceipt"]; readonly dispatch?: ThreadManagementService["Service"]["dispatch"]; }) { @@ -160,7 +162,9 @@ function serviceLayer(input: { Layer.mergeAll( NodeServices.layer, Layer.mock(ThreadManagementService)({ - getThreadProjection: () => Effect.succeed(input.getParent?.() ?? parentProjection()), + getThreadProjection: + input.getThreadProjection ?? + (() => Effect.succeed(input.getParent?.() ?? parentProjection())), getProjectThread: ({ threadId }) => threadId === childThreadId ? Effect.succeed(input.getChild()) @@ -197,6 +201,32 @@ describe("PendingRequestMcpService", () => { }).pipe(Effect.provide(serviceLayer({ getChild: () => childProjection() }))), ); + it.effect("keeps projection defects out of MCP failure messages", () => + Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const failure = yield* service.list(scope, {}).pipe(Effect.flip); + assert.equal(failure.code, "orchestration_error"); + assert.equal( + failure.message, + `Unable to load calling thread '${parentThreadId}': Failed to load orchestration projection for thread ${parentThreadId}.`, + ); + assert.notInclude(failure.message, "database credentials leaked"); + }).pipe( + Effect.provide( + serviceLayer({ + getChild: () => childProjection(), + getThreadProjection: () => + Effect.fail( + new OrchestratorProjectionError({ + threadId: parentThreadId, + cause: new Error("database credentials leaked"), + }), + ), + }), + ), + ), + ); + it.effect("rejects provider-native children and non-user-input request kinds", () => Effect.gen(function* () { const providerOwnedService = yield* PendingRequestMcpService; diff --git a/apps/server/src/mcp/PendingRequestMcpService.ts b/apps/server/src/mcp/PendingRequestMcpService.ts index 395879f407d5..41a020fe37eb 100644 --- a/apps/server/src/mcp/PendingRequestMcpService.ts +++ b/apps/server/src/mcp/PendingRequestMcpService.ts @@ -82,7 +82,6 @@ function errorMessage(error: unknown): string { if (typeof error === "object" && error !== null && "cause" in error) { const cause = (error as { readonly cause: unknown }).cause; if (typeof cause === "string") return cause; - if (cause instanceof Error) return cause.message; } return error instanceof Error ? error.message : String(error); } From 4b59d8274cf4454d1ba4bc5a8fd80f7225b5fbf6 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 29 Aug 2026 18:21:05 -0700 Subject: [PATCH 3/5] fix(mcp): bound pending request discovery --- .../src/mcp/PendingRequestMcpService.test.ts | 298 +++++++++++++++++- .../src/mcp/PendingRequestMcpService.ts | 200 +++++++++--- .../src/mcp/toolkits/pendingRequest/tools.ts | 6 +- .../orchestrator-mcp-server.md | 10 +- docs/user/permission-modes.md | 3 +- .../contracts/src/pendingRequestMcp.test.ts | 55 +++- packages/contracts/src/pendingRequestMcp.ts | 40 ++- 7 files changed, 548 insertions(+), 64 deletions(-) diff --git a/apps/server/src/mcp/PendingRequestMcpService.test.ts b/apps/server/src/mcp/PendingRequestMcpService.test.ts index 9cbf3d77831d..97ac3c49def9 100644 --- a/apps/server/src/mcp/PendingRequestMcpService.test.ts +++ b/apps/server/src/mcp/PendingRequestMcpService.test.ts @@ -3,6 +3,7 @@ import { assert, describe, it, vi } from "@effect/vitest"; import { EnvironmentId, NodeId, + PENDING_REQUEST_MCP_MAX_QUESTION_CHARS, ProjectId, ProviderInstanceId, ProviderSessionId, @@ -21,7 +22,11 @@ import * as Ref from "effect/Ref"; import type { CommandReceiptV2 } from "../orchestration-v2/CommandReceiptStore.ts"; import { OrchestratorProjectionError } from "../orchestration-v2/Orchestrator.ts"; -import { ThreadManagementService } from "../orchestration-v2/ThreadManagementService.ts"; +import { ProjectionStoreThreadNotFoundError } from "../orchestration-v2/ProjectionStore.ts"; +import { + ThreadManagementProjectionLoadError, + ThreadManagementService, +} from "../orchestration-v2/ThreadManagementService.ts"; import type { McpInvocationScope } from "./McpInvocationContext.ts"; import { layer, PendingRequestMcpService } from "./PendingRequestMcpService.ts"; @@ -150,10 +155,72 @@ function childProjection( } as unknown as OrchestrationV2ThreadProjection; } +function parentProjectionWithTasks( + tasks: ReadonlyArray<{ + readonly taskId: ReturnType; + readonly childThreadId: ReturnType; + readonly updatedAt?: typeof now; + }>, +) { + const parent = parentProjection(); + const baseTask = parent.subagents[0]; + if (baseTask === undefined) throw new Error("Pending-request parent fixture is incomplete."); + return { + ...parent, + subagents: tasks.map((task) => ({ + ...baseTask, + id: task.taskId, + threadId: parentThreadId, + origin: "app_owned" as const, + childThreadId: task.childThreadId, + updatedAt: task.updatedAt ?? now, + })), + } as OrchestrationV2ThreadProjection; +} + +function childProjectionWithRequests(input: { + readonly childThreadId: ReturnType; + readonly requests: ReadonlyArray<{ + readonly requestId: ReturnType; + readonly status?: OrchestrationV2RuntimeRequest["status"]; + readonly questions?: typeof questions; + }>; +}) { + const base = childProjection(); + const providerThread = base.providerThreads[0]; + const item = base.turnItems.find((candidate) => candidate.type === "user_input_request"); + if (providerThread === undefined || item?.type !== "user_input_request") { + throw new Error("Pending-request test fixture is incomplete."); + } + return { + ...base, + thread: { ...base.thread, id: input.childThreadId }, + runtimeRequests: input.requests.map((request) => ({ + ...base.runtimeRequests[0], + id: request.requestId, + nodeId: NodeId.make(`node:pending-request:${request.requestId}`), + status: request.status ?? "pending", + resolvedAt: request.status === undefined || request.status === "pending" ? null : now, + })), + turnItems: input.requests.map((request, index) => ({ + ...item, + id: TurnItemId.make(`turn-item:pending-request:${index}`), + threadId: input.childThreadId, + nodeId: NodeId.make(`node:pending-request:${request.requestId}`), + requestId: request.requestId, + questions: request.questions ?? questions, + status: + request.status === undefined || request.status === "pending" ? "waiting" : "completed", + completedAt: request.status === undefined || request.status === "pending" ? null : now, + })), + } as OrchestrationV2ThreadProjection; +} + function serviceLayer(input: { readonly getParent?: () => OrchestrationV2ThreadProjection; readonly getChild: () => OrchestrationV2ThreadProjection; readonly getThreadProjection?: ThreadManagementService["Service"]["getThreadProjection"]; + readonly getProjectThread?: ThreadManagementService["Service"]["getProjectThread"]; readonly getReceipt?: ThreadManagementService["Service"]["getCommandReceipt"]; readonly dispatch?: ThreadManagementService["Service"]["dispatch"]; }) { @@ -165,10 +232,12 @@ function serviceLayer(input: { getThreadProjection: input.getThreadProjection ?? (() => Effect.succeed(input.getParent?.() ?? parentProjection())), - getProjectThread: ({ threadId }) => - threadId === childThreadId - ? Effect.succeed(input.getChild()) - : Effect.die(new Error(`Unexpected child projection read: ${threadId}`)), + getProjectThread: + input.getProjectThread ?? + (({ threadId }) => + threadId === childThreadId + ? Effect.succeed(input.getChild()) + : Effect.die(new Error(`Unexpected child projection read: ${threadId}`))), getCommandReceipt: input.getReceipt ?? (() => Effect.succeed(Option.none())), dispatch: input.dispatch ?? @@ -191,6 +260,8 @@ describe("PendingRequestMcpService", () => { assert.equal(listed.requests[0]?.childThreadId, childThreadId); assert.equal(listed.requests[0]?.requestId, requestId); assert.deepEqual(listed.requests[0]?.questions, questions); + assert.equal(listed.requests[0]?.questionPayloadStatus, "complete"); + assert.isTrue(listed.requests[0]?.answerable); assert.equal(listed.nextCursor, null); const read = yield* service.read(scope, { childThreadId, requestId }); @@ -201,6 +272,223 @@ describe("PendingRequestMcpService", () => { }).pipe(Effect.provide(serviceLayer({ getChild: () => childProjection() }))), ); + it.effect("continues by stable task and request identities after prior results change", () => + Effect.gen(function* () { + const taskA = NodeId.make("node:pending-request-task-a"); + const taskB = NodeId.make("node:pending-request-task-b"); + const childA = ThreadId.make("thread:pending-request-child-a"); + const childB = ThreadId.make("thread:pending-request-child-b"); + const requestA = RuntimeRequestId.make("request:pending-request-a"); + const requestB = RuntimeRequestId.make("request:pending-request-b"); + const requestC = RuntimeRequestId.make("request:pending-request-c"); + let parent = parentProjectionWithTasks([ + { taskId: taskB, childThreadId: childB }, + { taskId: taskA, childThreadId: childA }, + ]); + let projectionA = childProjectionWithRequests({ + childThreadId: childA, + requests: [{ requestId: requestA }, { requestId: requestB }], + }); + const projectionB = childProjectionWithRequests({ + childThreadId: childB, + requests: [{ requestId: requestC }], + }); + + yield* Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const first = yield* service.list(scope, { limit: 1 }); + assert.deepEqual( + first.requests.map((request) => request.requestId), + [requestA], + ); + assert.isNotNull(first.nextCursor); + + parent = parentProjectionWithTasks([ + { + taskId: taskA, + childThreadId: childA, + updatedAt: DateTime.makeUnsafe("2026-08-29T12:02:00.000Z"), + }, + { + taskId: taskB, + childThreadId: childB, + updatedAt: DateTime.makeUnsafe("2026-08-29T12:01:00.000Z"), + }, + ]); + projectionA = childProjectionWithRequests({ + childThreadId: childA, + requests: [{ requestId: requestA, status: "resolved" }, { requestId: requestB }], + }); + + const second = yield* service.list(scope, { + cursor: first.nextCursor ?? undefined, + limit: 2, + }); + assert.deepEqual( + second.requests.map((request) => request.requestId), + [requestB, requestC], + ); + assert.isNull(second.nextCursor); + }).pipe( + Effect.provide( + serviceLayer({ + getParent: () => parent, + getChild: () => projectionA, + getProjectThread: ({ threadId }) => + Effect.succeed(threadId === childA ? projectionA : projectionB), + }), + ), + ); + }), + ); + + it.effect("bounds empty child scans and returns a continuation for unscanned children", () => + Effect.gen(function* () { + const tasks = Array.from({ length: 22 }, (_, index) => ({ + taskId: NodeId.make(`node:pending-request-empty-${String(index).padStart(2, "0")}`), + childThreadId: ThreadId.make( + `thread:pending-request-empty-${String(index).padStart(2, "0")}`, + ), + })); + const getProjectThread = vi.fn(({ threadId }) => + Effect.succeed(childProjectionWithRequests({ childThreadId: threadId, requests: [] })), + ); + + yield* Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const first = yield* service.list(scope, { limit: 50 }); + assert.deepEqual(first.requests, []); + assert.isNotNull(first.nextCursor); + assert.equal(getProjectThread.mock.calls.length, 20); + + const second = yield* service.list(scope, { cursor: first.nextCursor ?? undefined }); + assert.deepEqual(second.requests, []); + assert.isNull(second.nextCursor); + assert.equal(getProjectThread.mock.calls.length, 22); + }).pipe( + Effect.provide( + serviceLayer({ + getParent: () => parentProjectionWithTasks(tasks), + getChild: () => childProjection(), + getProjectThread, + }), + ), + ); + }), + ); + + it.effect("skips only typed missing children while surfacing projection storage failures", () => + Effect.gen(function* () { + const missingTask = NodeId.make("node:pending-request-missing-a"); + const validTask = NodeId.make("node:pending-request-valid-b"); + const missingChild = ThreadId.make("thread:pending-request-missing-a"); + const validChild = ThreadId.make("thread:pending-request-valid-b"); + const validRequest = RuntimeRequestId.make("request:pending-request-valid"); + const parent = parentProjectionWithTasks([ + { taskId: missingTask, childThreadId: missingChild }, + { taskId: validTask, childThreadId: validChild }, + ]); + const missingCause = new ThreadManagementProjectionLoadError({ + projectId, + threadId: missingChild, + cause: new OrchestratorProjectionError({ + threadId: missingChild, + cause: new ProjectionStoreThreadNotFoundError({ threadId: missingChild }), + }), + }); + const storageCause = new ThreadManagementProjectionLoadError({ + projectId, + threadId: missingChild, + cause: new OrchestratorProjectionError({ + threadId: missingChild, + cause: new Error("sqlite credentials leaked"), + }), + }); + + yield* Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const listed = yield* service.list(scope, {}); + assert.deepEqual( + listed.requests.map((request) => request.requestId), + [validRequest], + ); + }).pipe( + Effect.provide( + serviceLayer({ + getParent: () => parent, + getChild: () => childProjection(), + getProjectThread: ({ threadId }) => + threadId === missingChild + ? Effect.fail(missingCause) + : Effect.succeed( + childProjectionWithRequests({ + childThreadId: validChild, + requests: [{ requestId: validRequest }], + }), + ), + }), + ), + ); + + yield* Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const failed = yield* service.list(scope, {}).pipe(Effect.flip); + assert.equal(failed.code, "orchestration_error"); + assert.notInclude(failed.message, "sqlite credentials leaked"); + }).pipe( + Effect.provide( + serviceLayer({ + getParent: () => + parentProjectionWithTasks([{ taskId: missingTask, childThreadId: missingChild }]), + getChild: () => childProjection(), + getProjectThread: () => Effect.fail(storageCause), + }), + ), + ); + }), + ); + + it.effect("marks oversized provider questions as unavailable and refuses partial answers", () => { + const dispatch = vi.fn(() => Effect.die("oversized requests must not dispatch")); + const oversizedQuestions = [ + { + ...questions[0]!, + question: "q".repeat(PENDING_REQUEST_MCP_MAX_QUESTION_CHARS + 1), + }, + ]; + const oversized = childProjectionWithRequests({ + childThreadId, + requests: [{ requestId, questions: oversizedQuestions }], + }); + + return Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const read = yield* service.read(scope, { childThreadId, requestId }); + assert.equal(read.questionPayloadStatus, "too_large"); + assert.equal(read.questionCount, 1); + assert.deepEqual(read.questions, []); + assert.isFalse(read.answerable); + + const rejected = yield* service + .respond(scope, { + childThreadId, + requestId, + answers: { editor: "Vim" }, + clientRequestId: "oversized-question", + }) + .pipe(Effect.flip); + assert.equal(rejected.code, "request_payload_too_large"); + assert.equal(dispatch.mock.calls.length, 0); + }).pipe( + Effect.provide( + serviceLayer({ + getChild: () => oversized, + dispatch, + }), + ), + ); + }); + it.effect("keeps projection defects out of MCP failure messages", () => Effect.gen(function* () { const service = yield* PendingRequestMcpService; diff --git a/apps/server/src/mcp/PendingRequestMcpService.ts b/apps/server/src/mcp/PendingRequestMcpService.ts index 41a020fe37eb..e86c6ffc9e9d 100644 --- a/apps/server/src/mcp/PendingRequestMcpService.ts +++ b/apps/server/src/mcp/PendingRequestMcpService.ts @@ -1,5 +1,13 @@ import { CommandId, + PENDING_REQUEST_MCP_MAX_HEADER_CHARS, + PENDING_REQUEST_MCP_MAX_OPTION_DESCRIPTION_CHARS, + PENDING_REQUEST_MCP_MAX_OPTION_LABEL_CHARS, + PENDING_REQUEST_MCP_MAX_OPTIONS_PER_QUESTION, + PENDING_REQUEST_MCP_MAX_QUESTION_CHARS, + PENDING_REQUEST_MCP_MAX_QUESTION_ID_CHARS, + PENDING_REQUEST_MCP_MAX_QUESTIONS, + PENDING_REQUEST_MCP_MAX_TOTAL_QUESTION_CHARS, type OrchestrationV2RuntimeRequest, type OrchestrationV2Subagent, type OrchestrationV2ThreadProjection, @@ -24,6 +32,12 @@ import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import { + OrchestratorProjectionError, + type OrchestratorV2Error, +} from "../orchestration-v2/Orchestrator.ts"; +import { ProjectionStoreThreadNotFoundError } from "../orchestration-v2/ProjectionStore.ts"; +import { + ThreadManagementProjectionLoadError, ThreadManagementService, ThreadManagementThreadNotFoundError, } from "../orchestration-v2/ThreadManagementService.ts"; @@ -31,23 +45,34 @@ import { interactionModeWithinMcpCeiling, runtimeModeWithinMcpCeiling } from "./ import type { McpInvocationScope } from "./McpInvocationContext.ts"; const DEFAULT_LIST_LIMIT = 20; +const MAX_LIST_CHILD_PROJECTIONS = 20; + +interface ListCursor { + readonly taskId: string; + readonly requestId: string | null; +} function parseListCursor( cursor: string | undefined, -): Effect.Effect< - { readonly childIndex: number; readonly requestIndex: number }, - PendingRequestMcpFailure -> { - if (cursor === undefined) return Effect.succeed({ childIndex: 0, requestIndex: 0 }); - const match = /^(0|[1-9]\d*):(0|[1-9]\d*)$/.exec(cursor); +): Effect.Effect { + if (cursor === undefined) return Effect.succeed(null); + const match = /^v1:([^:]+):(.*)$/.exec(cursor); if (match === null) { return Effect.fail(failure("invalid_request", "The pending-request cursor is invalid.")); } - const childIndex = Number(match[1]); - const requestIndex = Number(match[2]); - return Number.isSafeInteger(childIndex) && Number.isSafeInteger(requestIndex) - ? Effect.succeed({ childIndex, requestIndex }) - : Effect.fail(failure("invalid_request", "The pending-request cursor is invalid.")); + try { + const taskId = decodeURIComponent(match[1] ?? ""); + const requestId = decodeURIComponent(match[2] ?? ""); + return taskId.length === 0 + ? Effect.fail(failure("invalid_request", "The pending-request cursor is invalid.")) + : Effect.succeed({ taskId, requestId: requestId.length === 0 ? null : requestId }); + } catch { + return Effect.fail(failure("invalid_request", "The pending-request cursor is invalid.")); + } +} + +function listCursor(taskId: string, requestId: string | null): string { + return `v1:${encodeURIComponent(taskId)}:${requestId === null ? "" : encodeURIComponent(requestId)}`; } export class PendingRequestMcpService extends Context.Service< @@ -76,20 +101,22 @@ function failure( } function errorMessage(error: unknown): string { - if (typeof error === "object" && error !== null && "detail" in error) { - return String((error as { readonly detail: unknown }).detail); - } - if (typeof error === "object" && error !== null && "cause" in error) { - const cause = (error as { readonly cause: unknown }).cause; - if (typeof cause === "string") return cause; - } return error instanceof Error ? error.message : String(error); } const isThreadNotFound = Schema.is(ThreadManagementThreadNotFoundError); +const isProjectionLoadError = Schema.is(ThreadManagementProjectionLoadError); +const isOrchestratorProjectionError = Schema.is(OrchestratorProjectionError); +const isProjectionStoreThreadNotFound = Schema.is(ProjectionStoreThreadNotFoundError); + +function isMissingChild(error: unknown): boolean { + if (isThreadNotFound(error)) return true; + if (!isProjectionLoadError(error) || !isOrchestratorProjectionError(error.cause)) return false; + return isProjectionStoreThreadNotFound(error.cause.cause); +} function projectionFailure(error: unknown, threadId: ThreadId): PendingRequestMcpFailure { - return isThreadNotFound(error) + return isMissingChild(error) ? failure("child_not_found", `Delegated child thread '${threadId}' was not found.`) : failure( "orchestration_error", @@ -97,16 +124,15 @@ function projectionFailure(error: unknown, threadId: ThreadId): PendingRequestMc ); } -function dispatchFailure(error: unknown): PendingRequestMcpFailure { - const tag = - typeof error === "object" && error !== null && "_tag" in error - ? String((error as { readonly _tag: unknown })._tag) - : ""; - return tag === "OrchestratorDispatchError" || - tag === "OrchestratorCommandPreviouslyRejectedError" || - tag === "OrchestratorCommandIdConflictError" - ? failure("operation_rejected", errorMessage(error)) - : failure("orchestration_error", errorMessage(error)); +function dispatchFailure(error: OrchestratorV2Error): PendingRequestMcpFailure { + switch (error._tag) { + case "OrchestratorDispatchError": + case "OrchestratorCommandPreviouslyRejectedError": + case "OrchestratorCommandIdConflictError": + return failure("operation_rejected", error.message); + default: + return failure("orchestration_error", error.message); + } } function directAppOwnedChildren(parent: OrchestrationV2ThreadProjection) { @@ -115,11 +141,7 @@ function directAppOwnedChildren(parent: OrchestrationV2ThreadProjection) { (task): task is OrchestrationV2Subagent & { readonly childThreadId: ThreadId } => task.origin === "app_owned" && task.childThreadId !== null, ) - .toSorted( - (left, right) => - DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt) || - right.id.localeCompare(left.id), - ); + .toSorted((left, right) => left.id.localeCompare(right.id)); } function findUserInputItem( @@ -132,6 +154,35 @@ function findUserInputItem( ); } +function questionPayloadFits( + questions: Extract["questions"], +): boolean { + if (questions.length > PENDING_REQUEST_MCP_MAX_QUESTIONS) return false; + let totalChars = 0; + for (const question of questions) { + if ( + question.id.length > PENDING_REQUEST_MCP_MAX_QUESTION_ID_CHARS || + question.header.length > PENDING_REQUEST_MCP_MAX_HEADER_CHARS || + question.question.length > PENDING_REQUEST_MCP_MAX_QUESTION_CHARS || + question.options.length > PENDING_REQUEST_MCP_MAX_OPTIONS_PER_QUESTION + ) { + return false; + } + totalChars += question.id.length + question.header.length + question.question.length; + for (const option of question.options) { + if ( + option.label.length > PENDING_REQUEST_MCP_MAX_OPTION_LABEL_CHARS || + option.description.length > PENDING_REQUEST_MCP_MAX_OPTION_DESCRIPTION_CHARS + ) { + return false; + } + totalChars += option.label.length + option.description.length; + } + if (totalChars > PENDING_REQUEST_MCP_MAX_TOTAL_QUESTION_CHARS) return false; + } + return true; +} + function requestSummary(input: { readonly task: OrchestrationV2Subagent & { readonly childThreadId: ThreadId }; readonly projection: OrchestrationV2ThreadProjection; @@ -152,6 +203,7 @@ function requestSummary(input: { ), ); } + const payloadFits = questionPayloadFits(input.item.questions); return Effect.succeed({ taskId: input.task.id, childThreadId: input.task.childThreadId, @@ -162,7 +214,13 @@ function requestSummary(input: { driverKind: providerThread.driver, status: input.request.status, resumable: input.request.responseCapability.type === "live", - questions: input.item.questions, + answerable: + input.request.status === "pending" && + input.request.responseCapability.type === "live" && + payloadFits, + questionCount: input.item.questions.length, + questionPayloadStatus: payloadFits ? "complete" : "too_large", + questions: payloadFits ? input.item.questions : [], createdAt: DateTime.formatIso(input.request.createdAt), resolvedAt: input.request.resolvedAt === null ? null : DateTime.formatIso(input.request.resolvedAt), @@ -323,33 +381,73 @@ export const make = Effect.gen(function* () { const limit = input.limit ?? DEFAULT_LIST_LIMIT; const cursor = yield* parseListCursor(input.cursor); const requests: Array = []; - let childIndex = cursor.childIndex; - let requestIndex = cursor.requestIndex; - while (childIndex < children.length && requests.length < limit) { + let projectionCount = 0; + let continuation: string | null = null; + let lastScannedChildIndex = -1; + + for (let childIndex = 0; childIndex < children.length; childIndex += 1) { const task = children[childIndex]; if (task === undefined) break; + const taskComparison = cursor === null ? 1 : task.id.localeCompare(cursor.taskId); + if (taskComparison < 0 || (taskComparison === 0 && cursor?.requestId === null)) continue; + if (projectionCount >= MAX_LIST_CHILD_PROJECTIONS) { + break; + } + projectionCount += 1; + lastScannedChildIndex = childIndex; const projection = yield* threadManagement .getProjectThread({ projectId: parent.thread.projectId, threadId: task.childThreadId, }) - .pipe(Effect.mapError((error) => projectionFailure(error, task.childThreadId))); - const pending = projection.runtimeRequests.filter( - (request) => request.kind === "user_input" && request.status === "pending", - ); - while (requestIndex < pending.length && requests.length < limit) { + .pipe( + Effect.matchEffect({ + onFailure: (error) => + isMissingChild(error) + ? Effect.void + : Effect.fail(projectionFailure(error, task.childThreadId)), + onSuccess: Effect.succeed, + }), + ); + continuation = listCursor(task.id, null); + if (projection === undefined) continue; + + const pending = projection.runtimeRequests + .filter((request) => request.kind === "user_input" && request.status === "pending") + .toSorted((left, right) => left.id.localeCompare(right.id)); + for (let requestIndex = 0; requestIndex < pending.length; requestIndex += 1) { const request = pending[requestIndex]; if (request === undefined) break; + if ( + cursor !== null && + taskComparison === 0 && + cursor.requestId !== null && + request.id.localeCompare(cursor.requestId) <= 0 + ) { + continue; + } requests.push(yield* resultFor({ task, projection, requestId: request.id })); - requestIndex += 1; + continuation = listCursor(task.id, request.id); + if (requests.length === limit) { + const hasMoreRequests = pending + .slice(requestIndex + 1) + .some((candidate) => candidate.id.localeCompare(request.id) > 0); + const hasMoreChildren = childIndex + 1 < children.length; + return { + requests, + nextCursor: hasMoreRequests || hasMoreChildren ? continuation : null, + } satisfies PendingRequestMcpListResult; + } } - if (requestIndex < pending.length) break; - childIndex += 1; - requestIndex = 0; } return { requests, - nextCursor: childIndex < children.length ? `${childIndex}:${requestIndex}` : null, + nextCursor: + projectionCount >= MAX_LIST_CHILD_PROJECTIONS && + continuation !== null && + lastScannedChildIndex + 1 < children.length + ? continuation + : null, } satisfies PendingRequestMcpListResult; }), read: (scope, input) => @@ -416,6 +514,12 @@ export const make = Effect.gen(function* () { if (request.responseCapability.type !== "live") { return yield* failure("request_not_resumable", request.responseCapability.reason); } + if (!questionPayloadFits(item.questions)) { + return yield* failure( + "request_payload_too_large", + `User-input request '${request.id}' exceeds the bounded MCP question payload and cannot be answered through this tool.`, + ); + } const expectedIds = new Set(item.questions.map((question) => question.id)); const suppliedIds = Object.keys(input.answers); const missing = [...expectedIds].filter((id) => !(id in input.answers)); diff --git a/apps/server/src/mcp/toolkits/pendingRequest/tools.ts b/apps/server/src/mcp/toolkits/pendingRequest/tools.ts index 963ce34f7ef8..295c4bcb5cff 100644 --- a/apps/server/src/mcp/toolkits/pendingRequest/tools.ts +++ b/apps/server/src/mcp/toolkits/pendingRequest/tools.ts @@ -16,7 +16,7 @@ const dependencies = [McpInvocationContext.McpInvocationContext, PendingRequestM export const PendingRequestListTool = Tool.make("t3_pending_request_list", { description: - "List pending user-input questions from direct app-owned delegated children of the calling thread. Results are bounded and never include approval requests or unrelated threads.", + "List pending user-input questions from direct app-owned delegated children of the calling thread. Results and child scans are bounded; an empty page can have a continuation cursor. Oversized question payloads are explicitly unavailable and never partial. Approval requests and unrelated threads are excluded.", parameters: PendingRequestMcpListInput, success: PendingRequestMcpListResult, failure: PendingRequestMcpFailure, @@ -30,7 +30,7 @@ export const PendingRequestListTool = Tool.make("t3_pending_request_list", { export const PendingRequestReadTool = Tool.make("t3_pending_request_read", { description: - "Read one structured user-input question from a direct app-owned delegated child. Resolved and stale requests remain readable; this tool never responds or approves anything.", + "Read one structured user-input question from a direct app-owned delegated child. Resolved and stale requests remain readable. Oversized question payloads are explicitly unavailable and never partial. This tool never responds or approves anything.", parameters: PendingRequestMcpReadInput, success: PendingRequestMcpReadResult, failure: PendingRequestMcpFailure, @@ -44,7 +44,7 @@ export const PendingRequestReadTool = Tool.make("t3_pending_request_read", { export const PendingRequestRespondTool = Tool.make("t3_pending_request_respond", { description: - "Answer a pending user-input question from a direct app-owned delegated child. Supply every exact question ID. This cannot answer approval requests, grant permissions, or target the calling agent or an unrelated thread. Acceptance is reported by a durable V2 command receipt; provider delivery may complete asynchronously.", + "Answer a pending user-input question from a direct app-owned delegated child. Supply every exact question ID from a complete bounded payload; oversized payloads cannot be answered through MCP. This cannot answer approval requests, grant permissions, or target the calling agent or an unrelated thread. Acceptance is reported by a durable V2 command receipt; provider delivery may complete asynchronously.", parameters: PendingRequestMcpRespondInput, success: PendingRequestMcpRespondResult, failure: PendingRequestMcpFailure, diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 9237f2d08871..029ba51c1b1c 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -223,7 +223,10 @@ reason. Lists pending structured user-input questions from direct app-owned delegated children of the calling thread. The bounded page follows the parent's durable sub-agent records and never scans unrelated or provider-native child threads. -Approval and permission requests are excluded. +Approval and permission requests are excluded. Cursors continue from stable +task/request identities, so resolving an earlier page does not shift later +results. A page also caps child projections inspected and can therefore return +an empty result with a continuation cursor. ### `t3_pending_request_read` @@ -232,6 +235,11 @@ Unlike the list path, read can return resolved, expired, or cancelled requests so callers can distinguish a stale response attempt from a missing request. The read path is non-mutating. +Question payloads are bounded by count, field length, and total text. A request +that exceeds those bounds remains discoverable with its stable IDs, +`questionPayloadStatus: "too_large"`, `answerable: false`, and no partial +question set. Such a request cannot be answered through MCP. + ### `t3_pending_request_respond` Answers a pending, live `user_input` request on a direct app-owned delegated diff --git a/docs/user/permission-modes.md b/docs/user/permission-modes.md index 17baed8077c3..c212fa92a5e0 100644 --- a/docs/user/permission-modes.md +++ b/docs/user/permission-modes.md @@ -29,7 +29,8 @@ there. When an agent delegates work to a T3-owned child conversation, it can relay that child's structured questions and send your answers back. This applies only to user-input questions from that direct child. It does not let either agent approve permission requests or expand the child's -permission mode. +permission mode. Exceptionally large question sets remain visible as unavailable instead of being +partially shown or answered. For Grok, **Always allow this session** remembers the matching command or tool input. Other actions still ask for approval. It does not change the thread to **Full access**. diff --git a/packages/contracts/src/pendingRequestMcp.test.ts b/packages/contracts/src/pendingRequestMcp.test.ts index 166f94b3fd39..ef5999947133 100644 --- a/packages/contracts/src/pendingRequestMcp.test.ts +++ b/packages/contracts/src/pendingRequestMcp.test.ts @@ -2,13 +2,16 @@ import { describe, expect, it } from "@effect/vitest"; import * as Schema from "effect/Schema"; import { + PENDING_REQUEST_MCP_MAX_QUESTION_CHARS, PendingRequestMcpListInput, + PendingRequestMcpReadResult, PendingRequestMcpReadInput, PendingRequestMcpRespondInput, } from "./pendingRequestMcp.ts"; const decodeList = Schema.decodeUnknownSync(PendingRequestMcpListInput); const decodeRead = Schema.decodeUnknownSync(PendingRequestMcpReadInput); +const decodeReadResult = Schema.decodeUnknownSync(PendingRequestMcpReadResult); const decodeRespond = Schema.decodeUnknownSync(PendingRequestMcpRespondInput); describe("pending-request MCP contracts", () => { @@ -22,7 +25,7 @@ describe("pending-request MCP contracts", () => { } }); - it("accepts structured single and multiple-choice answers", () => { + it("accepts structured text, numeric, boolean, and multiple-choice answers", () => { expect( decodeRespond({ childThreadId: "thread:child", @@ -30,21 +33,69 @@ describe("pending-request MCP contracts", () => { answers: { editor: "vim", features: ["queue controls", "attachments"], + retries: 3, + confirmed: true, }, clientRequestId: "answer-questions-1", }).answers, ).toEqual({ editor: "vim", features: ["queue controls", "attachments"], + retries: 3, + confirmed: true, }); }); it("bounds list pages and requires stable target identifiers", () => { - expect(decodeList({ limit: 50, cursor: "2:1" })).toEqual({ limit: 50, cursor: "2:1" }); + expect(decodeList({ limit: 50, cursor: "v1:node%3Atask:request%3Aquestion" })).toEqual({ + limit: 50, + cursor: "v1:node%3Atask:request%3Aquestion", + }); expect(() => decodeList({ limit: 51 })).toThrow(); expect(decodeRead({ childThreadId: "thread:child", requestId: "request:questions" })).toEqual({ childThreadId: "thread:child", requestId: "request:questions", }); }); + + it("rejects oversized complete question payloads but represents them explicitly", () => { + const result = { + taskId: "node:task", + childThreadId: "thread:child", + runId: null, + nodeId: "node:request", + requestId: "request:questions", + providerInstanceId: "codex", + driverKind: "codex", + status: "pending", + resumable: true, + answerable: true, + questionCount: 1, + questionPayloadStatus: "complete", + questions: [ + { + id: "editor", + header: "Editor", + question: "q".repeat(PENDING_REQUEST_MCP_MAX_QUESTION_CHARS + 1), + options: [], + }, + ], + createdAt: "2026-08-29T12:00:00.000Z", + resolvedAt: null, + }; + expect(() => decodeReadResult(result)).toThrow(); + expect( + decodeReadResult({ + ...result, + answerable: false, + questionPayloadStatus: "too_large", + questions: [], + }), + ).toMatchObject({ + requestId: "request:questions", + questionCount: 1, + questionPayloadStatus: "too_large", + questions: [], + }); + }); }); diff --git a/packages/contracts/src/pendingRequestMcp.ts b/packages/contracts/src/pendingRequestMcp.ts index 56972387e78d..40f916372889 100644 --- a/packages/contracts/src/pendingRequestMcp.ts +++ b/packages/contracts/src/pendingRequestMcp.ts @@ -11,20 +11,30 @@ import { ThreadId, TrimmedNonEmptyString, } from "./baseSchemas.ts"; -import { OrchestrationV2UserInputQuestion } from "./orchestrationV2.ts"; import { ProviderDriverKind, ProviderInstanceId } from "./providerInstance.ts"; +export const PENDING_REQUEST_MCP_MAX_QUESTIONS = 20; +export const PENDING_REQUEST_MCP_MAX_OPTIONS_PER_QUESTION = 20; +export const PENDING_REQUEST_MCP_MAX_QUESTION_ID_CHARS = 256; +export const PENDING_REQUEST_MCP_MAX_HEADER_CHARS = 256; +export const PENDING_REQUEST_MCP_MAX_QUESTION_CHARS = 4_000; +export const PENDING_REQUEST_MCP_MAX_OPTION_LABEL_CHARS = 256; +export const PENDING_REQUEST_MCP_MAX_OPTION_DESCRIPTION_CHARS = 2_000; +export const PENDING_REQUEST_MCP_MAX_TOTAL_QUESTION_CHARS = 32_000; + const PendingRequestMcpLimit = PositiveInt.check(Schema.isLessThanOrEqualTo(50)); const PendingRequestMcpClientRequestId = TrimmedNonEmptyString.check( Schema.isMaxLength(256), ).annotate({ description: "Stable idempotency key to reuse when retrying this response." }); const PendingRequestMcpAnswer = Schema.Union([ TrimmedNonEmptyString, + Schema.Number, + Schema.Boolean, Schema.Array(TrimmedNonEmptyString).check(Schema.isMinLength(1), Schema.isMaxLength(20)), ]); export const PendingRequestMcpListInput = Schema.Struct({ - cursor: Schema.optional(TrimmedNonEmptyString.check(Schema.isMaxLength(64))).annotate({ + cursor: Schema.optional(TrimmedNonEmptyString.check(Schema.isMaxLength(4_096))).annotate({ description: "Opaque cursor returned by the previous list page.", }), limit: Schema.optional(PendingRequestMcpLimit), @@ -42,12 +52,28 @@ export const PendingRequestMcpRespondInput = Schema.Struct({ requestId: RuntimeRequestId, answers: Schema.Record(TrimmedNonEmptyString, PendingRequestMcpAnswer).annotate({ description: - "Answers keyed by the exact question IDs returned by list/read. Every question must be answered; values may be one string or a non-empty string array.", + "Answers keyed by the exact question IDs returned by list/read. Every question must be answered; values may be a string, number, boolean, or non-empty string array.", }), clientRequestId: Schema.optional(PendingRequestMcpClientRequestId), }); export type PendingRequestMcpRespondInput = typeof PendingRequestMcpRespondInput.Type; +const PendingRequestMcpQuestion = Schema.Struct({ + id: TrimmedNonEmptyString.check(Schema.isMaxLength(PENDING_REQUEST_MCP_MAX_QUESTION_ID_CHARS)), + header: TrimmedNonEmptyString.check(Schema.isMaxLength(PENDING_REQUEST_MCP_MAX_HEADER_CHARS)), + question: TrimmedNonEmptyString.check(Schema.isMaxLength(PENDING_REQUEST_MCP_MAX_QUESTION_CHARS)), + options: Schema.Array( + Schema.Struct({ + label: TrimmedNonEmptyString.check( + Schema.isMaxLength(PENDING_REQUEST_MCP_MAX_OPTION_LABEL_CHARS), + ), + description: TrimmedNonEmptyString.check( + Schema.isMaxLength(PENDING_REQUEST_MCP_MAX_OPTION_DESCRIPTION_CHARS), + ), + }), + ).check(Schema.isMaxLength(PENDING_REQUEST_MCP_MAX_OPTIONS_PER_QUESTION)), +}); + export const PendingRequestMcpRequest = Schema.Struct({ taskId: NodeId, childThreadId: ThreadId, @@ -58,7 +84,12 @@ export const PendingRequestMcpRequest = Schema.Struct({ driverKind: ProviderDriverKind, status: Schema.Literals(["pending", "resolved", "expired", "cancelled"]), resumable: Schema.Boolean, - questions: Schema.Array(OrchestrationV2UserInputQuestion), + answerable: Schema.Boolean, + questionCount: NonNegativeInt, + questionPayloadStatus: Schema.Literals(["complete", "too_large"]), + questions: Schema.Array(PendingRequestMcpQuestion).check( + Schema.isMaxLength(PENDING_REQUEST_MCP_MAX_QUESTIONS), + ), createdAt: IsoDateTime, resolvedAt: Schema.NullOr(IsoDateTime), }); @@ -95,6 +126,7 @@ export class PendingRequestMcpFailure extends Schema.TaggedErrorClass Date: Sat, 29 Aug 2026 18:23:12 -0700 Subject: [PATCH 4/5] fix(mcp): require finite request answers --- packages/contracts/src/pendingRequestMcp.test.ts | 7 +++++++ packages/contracts/src/pendingRequestMcp.ts | 2 +- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/packages/contracts/src/pendingRequestMcp.test.ts b/packages/contracts/src/pendingRequestMcp.test.ts index ef5999947133..c05ce26f71d1 100644 --- a/packages/contracts/src/pendingRequestMcp.test.ts +++ b/packages/contracts/src/pendingRequestMcp.test.ts @@ -44,6 +44,13 @@ describe("pending-request MCP contracts", () => { retries: 3, confirmed: true, }); + expect(() => + decodeRespond({ + childThreadId: "thread:child", + requestId: "request:questions", + answers: { retries: Number.POSITIVE_INFINITY }, + }), + ).toThrow(); }); it("bounds list pages and requires stable target identifiers", () => { diff --git a/packages/contracts/src/pendingRequestMcp.ts b/packages/contracts/src/pendingRequestMcp.ts index 40f916372889..90543c2acbf0 100644 --- a/packages/contracts/src/pendingRequestMcp.ts +++ b/packages/contracts/src/pendingRequestMcp.ts @@ -28,7 +28,7 @@ const PendingRequestMcpClientRequestId = TrimmedNonEmptyString.check( ).annotate({ description: "Stable idempotency key to reuse when retrying this response." }); const PendingRequestMcpAnswer = Schema.Union([ TrimmedNonEmptyString, - Schema.Number, + Schema.Number.check(Schema.isFinite()), Schema.Boolean, Schema.Array(TrimmedNonEmptyString).check(Schema.isMinLength(1), Schema.isMaxLength(20)), ]); From 88e4840319c73cab63f8350f473528be1f182cc1 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 29 Aug 2026 18:29:31 -0700 Subject: [PATCH 5/5] fix(mcp): validate pending answer ownership --- .../src/mcp/PendingRequestMcpService.test.ts | 35 +++++++++++++++++++ .../src/mcp/PendingRequestMcpService.ts | 33 +++++++++++------ 2 files changed, 58 insertions(+), 10 deletions(-) diff --git a/apps/server/src/mcp/PendingRequestMcpService.test.ts b/apps/server/src/mcp/PendingRequestMcpService.test.ts index 97ac3c49def9..b8cdc2333a9c 100644 --- a/apps/server/src/mcp/PendingRequestMcpService.test.ts +++ b/apps/server/src/mcp/PendingRequestMcpService.test.ts @@ -690,6 +690,41 @@ describe("PendingRequestMcpService", () => { ), ); + it.effect("requires own answers for prototype-named question IDs", () => { + const dispatch = vi.fn(() => Effect.die("missing answers must not dispatch")); + const prototypeQuestion = childProjectionWithRequests({ + childThreadId, + requests: [ + { + requestId, + questions: [{ ...questions[0]!, id: "toString" }], + }, + ], + }); + + return Effect.gen(function* () { + const service = yield* PendingRequestMcpService; + const rejected = yield* service + .respond(scope, { + childThreadId, + requestId, + answers: {}, + clientRequestId: "prototype-question-id", + }) + .pipe(Effect.flip); + assert.equal(rejected.code, "invalid_answers"); + assert.equal(rejected.message, "Missing question IDs: toString."); + assert.equal(dispatch.mock.calls.length, 0); + }).pipe( + Effect.provide( + serviceLayer({ + getChild: () => prototypeQuestion, + dispatch, + }), + ), + ); + }); + it.effect("keeps delegated answers within the caller's runtime and interaction ceilings", () => Effect.gen(function* () { const service = yield* PendingRequestMcpService; diff --git a/apps/server/src/mcp/PendingRequestMcpService.ts b/apps/server/src/mcp/PendingRequestMcpService.ts index e86c6ffc9e9d..58ec997b3eb2 100644 --- a/apps/server/src/mcp/PendingRequestMcpService.ts +++ b/apps/server/src/mcp/PendingRequestMcpService.ts @@ -37,9 +37,8 @@ import { } from "../orchestration-v2/Orchestrator.ts"; import { ProjectionStoreThreadNotFoundError } from "../orchestration-v2/ProjectionStore.ts"; import { - ThreadManagementProjectionLoadError, + type ThreadManagementError, ThreadManagementService, - ThreadManagementThreadNotFoundError, } from "../orchestration-v2/ThreadManagementService.ts"; import { interactionModeWithinMcpCeiling, runtimeModeWithinMcpCeiling } from "./McpModeCeilings.ts"; import type { McpInvocationScope } from "./McpInvocationContext.ts"; @@ -104,18 +103,32 @@ function errorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } -const isThreadNotFound = Schema.is(ThreadManagementThreadNotFoundError); -const isProjectionLoadError = Schema.is(ThreadManagementProjectionLoadError); const isOrchestratorProjectionError = Schema.is(OrchestratorProjectionError); const isProjectionStoreThreadNotFound = Schema.is(ProjectionStoreThreadNotFoundError); -function isMissingChild(error: unknown): boolean { - if (isThreadNotFound(error)) return true; - if (!isProjectionLoadError(error) || !isOrchestratorProjectionError(error.cause)) return false; - return isProjectionStoreThreadNotFound(error.cause.cause); +function isMissingChild(error: ThreadManagementError): boolean { + switch (error._tag) { + case "ThreadManagementThreadNotFoundError": + return true; + case "ThreadManagementProjectionLoadError": + return ( + isOrchestratorProjectionError(error.cause) && + isProjectionStoreThreadNotFound(error.cause.cause) + ); + case "ThreadManagementRunNotFoundError": + case "ThreadManagementThreadArchivedError": + case "ThreadManagementNoSteerableRunError": + case "ThreadManagementThreadNotInterruptibleError": + case "ThreadManagementProjectThreadsListError": + case "ThreadManagementDurableRunProjectionError": + return false; + } } -function projectionFailure(error: unknown, threadId: ThreadId): PendingRequestMcpFailure { +function projectionFailure( + error: ThreadManagementError, + threadId: ThreadId, +): PendingRequestMcpFailure { return isMissingChild(error) ? failure("child_not_found", `Delegated child thread '${threadId}' was not found.`) : failure( @@ -522,7 +535,7 @@ export const make = Effect.gen(function* () { } const expectedIds = new Set(item.questions.map((question) => question.id)); const suppliedIds = Object.keys(input.answers); - const missing = [...expectedIds].filter((id) => !(id in input.answers)); + const missing = [...expectedIds].filter((id) => !Object.hasOwn(input.answers, id)); const unknown = suppliedIds.filter((answerId) => !expectedIds.has(answerId)); if (missing.length > 0 || unknown.length > 0) { return yield* failure(