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..d6d393f74623 --- /dev/null +++ b/apps/server/src/mcp/PendingRequestMcpService.test.ts @@ -0,0 +1,836 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, describe, it, vi } from "@effect/vitest"; +import { + EnvironmentId, + NodeId, + PENDING_REQUEST_MCP_MAX_QUESTION_CHARS, + 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 { OrchestratorProjectionError } from "../orchestration-v2/Orchestrator.ts"; +import { ProjectionStoreThreadNotFoundError } from "../orchestration-v2/ProjectionStore.ts"; +import { + ThreadManagementProjectionLoadError, + ThreadManagementService, +} from "../orchestration-v2/ThreadManagementService.ts"; +import type { McpInvocationScope } from "./McpInvocationContext.ts"; +import * as 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?", + multiSelect: true, + 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 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"]; +}) { + return PendingRequestMcpService.layer.pipe( + Layer.provide( + Layer.mergeAll( + NodeServices.layer, + Layer.mock(ThreadManagementService)({ + getThreadProjection: + input.getThreadProjection ?? + (() => Effect.succeed(input.getParent?.() ?? parentProjection())), + 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 ?? + (() => 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.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.requests[0]?.questionPayloadStatus, "complete"); + assert.isTrue(listed.requests[0]?.answerable); + 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("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.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.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.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.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("keeps provider options without descriptions answerable", () => { + const withoutDescription = childProjectionWithRequests({ + childThreadId, + requests: [ + { + requestId, + questions: [ + { + ...questions[0]!, + options: [{ label: "Vim", description: "" }], + }, + ], + }, + ], + }); + + return Effect.gen(function* () { + const service = yield* PendingRequestMcpService.PendingRequestMcpService; + const read = yield* service.read(scope, { childThreadId, requestId }); + assert.equal(read.questionPayloadStatus, "complete"); + assert.isTrue(read.answerable); + assert.equal(read.questions[0]?.options[0]?.description, ""); + }).pipe(Effect.provide(serviceLayer({ getChild: () => withoutDescription }))); + }); + + 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.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("marks duplicate provider question IDs invalid and refuses ambiguous answers", () => { + const dispatch = vi.fn(() => Effect.die("invalid requests must not dispatch")); + const duplicateQuestions = [questions[0]!, { ...questions[0]!, header: "Second editor" }]; + const invalid = childProjectionWithRequests({ + childThreadId, + requests: [{ requestId, questions: duplicateQuestions }], + }); + + return Effect.gen(function* () { + const service = yield* PendingRequestMcpService.PendingRequestMcpService; + const read = yield* service.read(scope, { childThreadId, requestId }); + assert.equal(read.questionPayloadStatus, "invalid"); + assert.equal(read.questionCount, 2); + assert.deepEqual(read.questions, []); + assert.isFalse(read.answerable); + + const rejected = yield* service + .respond(scope, { + childThreadId, + requestId, + answers: { editor: ["Vim", "Zed"] }, + clientRequestId: "duplicate-question-ids", + }) + .pipe(Effect.flip); + assert.equal(rejected.code, "invalid_answers"); + assert.match(rejected.message, /duplicate question IDs/u); + assert.equal(dispatch.mock.calls.length, 0); + }).pipe(Effect.provide(serviceLayer({ getChild: () => invalid, dispatch }))); + }); + + it.effect("keeps projection defects out of MCP failure messages", () => + Effect.gen(function* () { + const service = yield* PendingRequestMcpService.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.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.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.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.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.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.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.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("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.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.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.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.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..29f0698ab159 --- /dev/null +++ b/apps/server/src/mcp/PendingRequestMcpService.ts @@ -0,0 +1,588 @@ +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, + 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 { + OrchestratorProjectionError, + type OrchestratorV2Error, +} from "../orchestration-v2/Orchestrator.ts"; +import { ProjectionStoreThreadNotFoundError } from "../orchestration-v2/ProjectionStore.ts"; +import { + type ThreadManagementError, + ThreadManagementService, +} from "../orchestration-v2/ThreadManagementService.ts"; +import { interactionModeWithinMcpCeiling, runtimeModeWithinMcpCeiling } from "./McpModeCeilings.ts"; +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 { + 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.")); + } + 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< + 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 { + return error instanceof Error ? error.message : String(error); +} + +const isOrchestratorProjectionError = Schema.is(OrchestratorProjectionError); +const isProjectionStoreThreadNotFound = Schema.is(ProjectionStoreThreadNotFoundError); + +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: ThreadManagementError, + threadId: ThreadId, +): PendingRequestMcpFailure { + return isMissingChild(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: 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) { + return parent.subagents + .filter( + (task): task is OrchestrationV2Subagent & { readonly childThreadId: ThreadId } => + task.origin === "app_owned" && task.childThreadId !== null, + ) + .toSorted((left, right) => left.id.localeCompare(right.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 questionPayloadStatus( + questions: Extract["questions"], +): "complete" | "too_large" | "invalid" { + if (questions.length > PENDING_REQUEST_MCP_MAX_QUESTIONS) return "too_large"; + const questionIds = new Set(); + let totalChars = 0; + for (const question of questions) { + if (questionIds.has(question.id)) return "invalid"; + questionIds.add(question.id); + 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 "too_large"; + } + 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 "too_large"; + } + totalChars += option.label.length + option.description.length; + } + if (totalChars > PENDING_REQUEST_MCP_MAX_TOTAL_QUESTION_CHARS) return "too_large"; + } + return "complete"; +} + +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.`, + ), + ); + } + const payloadStatus = questionPayloadStatus(input.item.questions); + 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", + answerable: + input.request.status === "pending" && + input.request.responseCapability.type === "live" && + payloadStatus === "complete", + questionCount: input.item.questions.length, + questionPayloadStatus: payloadStatus, + questions: payloadStatus === "complete" ? 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 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.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 })); + 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; + } + } + } + return { + requests, + nextCursor: + projectionCount >= MAX_LIST_CHILD_PROJECTIONS && + continuation !== null && + lastScannedChildIndex + 1 < children.length + ? continuation + : 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 payloadStatus = questionPayloadStatus(item.questions); + if (payloadStatus === "too_large") { + 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.`, + ); + } + if (payloadStatus === "invalid") { + return yield* failure( + "invalid_answers", + `User-input request '${request.id}' contains duplicate question IDs 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) => !Object.hasOwn(input.answers, id)); + 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..295c4bcb5cff --- /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 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, + 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. Oversized question payloads are explicitly unavailable and never partial. 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 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, + 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 35a3eb55a7e8..264b9b3496b7 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -46,6 +46,7 @@ import { formatClaudeResumeCompactionQuestion } from "@t3tools/shared/claudeComp 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, @@ -579,8 +580,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 f1ac2fbe2afd..f0bebe3d2a65 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -782,13 +782,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 66378584fd21..179346c8f70d 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; @@ -352,6 +354,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 @@ -534,6 +553,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 => @@ -4981,6 +5062,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, ); @@ -6921,6 +7003,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return { sequence: receipt.resultSequence, storedEvents, + replayed: true, } satisfies OrchestratorV2DispatchResult; } @@ -7000,11 +7083,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* () { @@ -7160,6 +7262,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return OrchestratorV2.of({ resumeQueuedRuns, dispatch: dispatchWithReceipt, + getCommandReceipt: commandReceipts.getByCommandId, getThreadProjection: (threadId) => projectionStore .getThreadProjection(threadId) @@ -7262,6 +7365,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 22741da769ad..9739e43e5385 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; @@ -653,6 +654,7 @@ const make = Effect.gen(function* () { return ThreadManagementService.of({ ensureLegacyTranscript, dispatch, + getCommandReceipt: orchestrator.getCommandReceipt, getThreadProjection, getThreadSnapshot, getThreadSnapshotWindow, diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 3aa9a6cb401a..89428ea62aec 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,44 @@ 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. 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` + +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. + +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. +Malformed provider payloads with duplicate question IDs remain discoverable in +the same bounded form with `questionPayloadStatus: "invalid"`; their ambiguous +answers are rejected. Complete payloads preserve each question's optional +`multiSelect` requirement. + +### `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 +356,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..4abb7ab716fe 100644 --- a/docs/user/permission-modes.md +++ b/docs/user/permission-modes.md @@ -26,6 +26,12 @@ 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, including multi-select 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. 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/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 d14f1dcebea3..f0c336711792 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -2217,6 +2217,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..4b0158d852ad --- /dev/null +++ b/packages/contracts/src/pendingRequestMcp.test.ts @@ -0,0 +1,146 @@ +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", () => { + 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 text, numeric, boolean, and multiple-choice answers", () => { + expect( + decodeRespond({ + childThreadId: "thread:child", + requestId: "request:questions", + 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, + }); + expect(() => + decodeRespond({ + childThreadId: "thread:child", + requestId: "request:questions", + answers: { retries: Number.POSITIVE_INFINITY }, + }), + ).toThrow(); + }); + + it("bounds list pages and requires stable target identifiers", () => { + 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), + multiSelect: true, + 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: [], + }); + }); + + it("preserves multi-select questions and rejects duplicate IDs in complete payloads", () => { + const request = { + 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: "features", + header: "Features", + question: "Which features?", + multiSelect: true, + options: [{ label: "Vim", description: "" }], + }, + ], + createdAt: "2026-08-29T12:00:00.000Z", + resolvedAt: null, + } as const; + expect(decodeReadResult(request).questions[0]?.multiSelect).toBe(true); + expect(decodeReadResult(request).questions[0]?.options[0]?.description).toBe(""); + expect(() => + decodeReadResult({ + ...request, + questionCount: 2, + questions: [request.questions[0], request.questions[0]], + }), + ).toThrow(/unique/u); + }); +}); diff --git a/packages/contracts/src/pendingRequestMcp.ts b/packages/contracts/src/pendingRequestMcp.ts new file mode 100644 index 000000000000..eb9205eb435f --- /dev/null +++ b/packages/contracts/src/pendingRequestMcp.ts @@ -0,0 +1,145 @@ +import * as Schema from "effect/Schema"; + +import { + CommandId, + IsoDateTime, + NodeId, + NonNegativeInt, + PositiveInt, + RunId, + RuntimeRequestId, + ThreadId, + TrimmedNonEmptyString, +} from "./baseSchemas.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.check(Schema.isFinite()), + Schema.Boolean, + Schema.Array(TrimmedNonEmptyString).check(Schema.isMinLength(1), Schema.isMaxLength(20)), +]); + +export const PendingRequestMcpListInput = Schema.Struct({ + cursor: Schema.optional(TrimmedNonEmptyString.check(Schema.isMaxLength(4_096))).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 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)), + multiSelect: Schema.optional(Schema.Boolean), + options: Schema.Array( + Schema.Struct({ + label: TrimmedNonEmptyString.check( + Schema.isMaxLength(PENDING_REQUEST_MCP_MAX_OPTION_LABEL_CHARS), + ), + description: Schema.String.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, + runId: Schema.NullOr(RunId), + nodeId: NodeId, + requestId: RuntimeRequestId, + providerInstanceId: ProviderInstanceId, + driverKind: ProviderDriverKind, + status: Schema.Literals(["pending", "resolved", "expired", "cancelled"]), + resumable: Schema.Boolean, + answerable: Schema.Boolean, + questionCount: NonNegativeInt, + questionPayloadStatus: Schema.Literals(["complete", "too_large", "invalid"]), + questions: Schema.Array(PendingRequestMcpQuestion).check( + Schema.isMaxLength(PENDING_REQUEST_MCP_MAX_QUESTIONS), + ), + createdAt: IsoDateTime, + resolvedAt: Schema.NullOr(IsoDateTime), +}).check( + Schema.makeFilter((request) => { + if (request.questionPayloadStatus !== "complete") return true; + const questionIds = request.questions.map((question) => question.id); + return ( + new Set(questionIds).size === questionIds.length || + "Complete pending-request question IDs must be unique." + ); + }), +); +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", + "request_payload_too_large", + "operation_rejected", + "orchestration_error", + ]), + message: Schema.String, + }, +) {} diff --git a/packages/shared/src/t3McpToolPresentation.ts b/packages/shared/src/t3McpToolPresentation.ts index 77498835757c..4e33d593928b 100644 --- a/packages/shared/src/t3McpToolPresentation.ts +++ b/packages/shared/src/t3McpToolPresentation.ts @@ -51,6 +51,9 @@ const T3_MCP_TOOLS: Record< t3_thread_send: { displayName: "Send to a T3 thread", summaryAction: "thread-send" }, t3_thread_wait: { displayName: "Wait for a T3 thread", summaryAction: "thread-wait" }, t3_thread_interrupt: { displayName: "Interrupt a T3 thread", summaryAction: "thread-interrupt" }, + t3_pending_request_list: { displayName: "List delegated user-input requests" }, + t3_pending_request_read: { displayName: "Read a delegated user-input request" }, + t3_pending_request_respond: { displayName: "Answer a delegated user-input request" }, t3_worktree_handoff: { displayName: "Hand off thread to a git worktree" }, t3_worktree_status: { displayName: "Get thread worktree status" }, preview_status: { displayName: "Get preview browser status" },