From dfc1edaa6ca8b0abb2492911d9c848b0c8d0e812 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 4 Oct 2026 12:29:36 +0000 Subject: [PATCH] fix(mcp): preserve thread command rejection reasons --- ...OrchestratorMcpToolkit.integration.test.ts | 24 ++++++++++ apps/server/src/mcp/threadAccess.ts | 13 ++++++ apps/server/src/mcp/toolkits/core.test.ts | 44 ++++++++++++++++++- .../src/mcp/toolkits/thread/handlers.ts | 13 +++--- 4 files changed, 87 insertions(+), 7 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index 142e0cc55f12..9bf90c464121 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -696,6 +696,18 @@ describe("orchestrator MCP toolkit", () => { const invoke = (name: string, args: Record) => invokeAs(invocation, name, args); + const refusedSettle = yield* invoke("t3_thread_organize", { action: "settle" }); + expect(refusedSettle.structuredContent).toEqual({ + _tag: "OrchestratorMcpFailure", + code: "orchestration_error", + message: `Thread ${parentThreadId} has active or blocked work and cannot be settled.`, + }); + const afterRefusedSettle = yield* orchestrator.getThreadProjection(parentThreadId); + expect(afterRefusedSettle.thread.settledOverride).not.toBe("settled"); + expect(afterRefusedSettle.runs.find((run) => run.id === parentRun?.id)?.status).toBe( + "running", + ); + const pinned = yield* invoke("t3_thread_organize", { action: "pin" }); expect(pinned.structuredContent).toHaveProperty("sequence"); expect((yield* orchestrator.getThreadShell(parentThreadId))?.pinnedAt).not.toBeNull(); @@ -705,6 +717,18 @@ describe("orchestrator MCP toolkit", () => { if (parentRun === undefined || parentRun.rootNodeId === null) { return yield* Effect.die(new Error("Parent run missing.")); } + for (const name of ["t3_queue_edit", "t3_queue_cancel"]) { + const refusedQueueMutation = yield* invoke(name, { + queuedRunId: parentRun.id, + ...(name === "t3_queue_edit" ? { text: "Keep the active turn." } : {}), + }); + expect(refusedQueueMutation.structuredContent).toEqual({ + _tag: "OrchestratorMcpFailure", + code: "orchestration_error", + message: `Run ${parentRun.id} is not queued.`, + }); + } + let parentRootNodeId = parentRun.rootNodeId; const queueAutomaticCompletion = (suffix: string, taskText: string) => Effect.gen(function* () { diff --git a/apps/server/src/mcp/threadAccess.ts b/apps/server/src/mcp/threadAccess.ts index bd3a4de60010..71a21de839fc 100644 --- a/apps/server/src/mcp/threadAccess.ts +++ b/apps/server/src/mcp/threadAccess.ts @@ -8,6 +8,7 @@ import { import * as Crypto from "effect/Crypto"; import * as Effect from "effect/Effect"; +import type { OrchestratorV2Error } from "../orchestration-v2/Orchestrator.ts"; import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; import * as OrchestrationMcp from "./OrchestratorMcpService.ts"; import * as McpInvocationContext from "./McpInvocationContext.ts"; @@ -18,6 +19,18 @@ export const unavailable = () => message: "The operation could not be completed.", }); +/** Decider string rejections are public; wrapped storage and hydration causes are not. */ +export const dispatchFailure = (error: OrchestratorV2Error) => + (error._tag === "OrchestratorDispatchError" || + error._tag === "OrchestratorCommandRejectedError") && + typeof error.cause === "string" && + error.cause.length > 0 + ? new OrchestratorMcpFailure({ + code: "orchestration_error", + message: Array.from(error.cause).slice(0, 1000).join(""), + }) + : unavailable(); + export const readCaller = Effect.fn("mcp.readCaller")(function* () { const scope = yield* McpInvocationContext.McpInvocationContext; if (!scope.capabilities.has("orchestration")) { diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 339e5eb9b2a8..6e2564eb2936 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -3,6 +3,7 @@ import { expect, it } from "@effect/vitest"; import { DEFAULT_SERVER_SETTINGS, ChatImageAttachment, + CommandId, EnvironmentId, ProviderInstanceId, ThreadId, @@ -13,10 +14,15 @@ import * as Schema from "effect/Schema"; import { McpAttachmentInput } from "./attachment/input.ts"; import { McpSchema, McpServer, Tool } from "effect/unstable/ai"; -import { OrchestratorProjectionError } from "../../orchestration-v2/Orchestrator.ts"; +import { + OrchestratorCommandRejectedError, + OrchestratorDispatchError, + OrchestratorProjectionError, +} from "../../orchestration-v2/Orchestrator.ts"; import * as ThreadManagement from "../../orchestration-v2/ThreadManagementService.ts"; import * as McpHttpServer from "../McpHttpServer.ts"; import * as McpInvocationContext from "../McpInvocationContext.ts"; +import { dispatchFailure } from "../threadAccess.ts"; import { OrchestratorToolkit } from "./orchestrator/tools.ts"; import { PreviewToolkit } from "./preview/tools.ts"; import { PreviewControlsToolkit } from "./previewControls/tools.ts"; @@ -157,6 +163,42 @@ it.effect("returns a bounded public failure without serializing storage causes", ), ); +it("bounds public command rejections and redacts internal dispatch causes", () => { + const command = { commandId: CommandId.make("mcp-core-command"), commandType: "thread.settle" }; + expect( + dispatchFailure(new OrchestratorDispatchError({ ...command, cause: "🙂".repeat(1001) })) + .message, + ).toBe("🙂".repeat(1000)); + expect( + dispatchFailure( + new OrchestratorCommandRejectedError({ ...command, cause: "Run is not queued." }), + ).message, + ).toBe("Run is not queued."); + for (const cause of [ + undefined, + "", + new Error("private-storage-path"), + { message: "private-storage-path" }, + ]) { + expect(dispatchFailure(new OrchestratorDispatchError({ ...command, cause }))).toMatchObject({ + code: "orchestration_error", + message: "The operation could not be completed.", + }); + expect( + dispatchFailure(new OrchestratorCommandRejectedError({ ...command, cause })), + ).toMatchObject({ + code: "orchestration_error", + message: "The operation could not be completed.", + }); + } + expect( + dispatchFailure(new OrchestratorProjectionError({ threadId, cause: "private-storage-path" })), + ).toMatchObject({ + code: "orchestration_error", + message: "The operation could not be completed.", + }); +}); + it("keeps MCP preference output allowlisted and Unicode-bounded", () => { const settings = { ...DEFAULT_SERVER_SETTINGS, diff --git a/apps/server/src/mcp/toolkits/thread/handlers.ts b/apps/server/src/mcp/toolkits/thread/handlers.ts index c0f5d135b6cc..0ee28877e1d2 100644 --- a/apps/server/src/mcp/toolkits/thread/handlers.ts +++ b/apps/server/src/mcp/toolkits/thread/handlers.ts @@ -11,6 +11,7 @@ import * as Effect from "effect/Effect"; import { modelSelectionCommandType } from "@t3tools/shared/model"; import { + dispatchFailure, newCommandId, readCaller, readMutationCaller, @@ -45,7 +46,7 @@ const dispatch = Effect.fn("mcp.dispatchThreadCommand")(function* ( const { threads, projection } = yield* readWritableThread(threadId); const result = yield* threads .dispatch(command({ commandId: yield* newCommandId(), threadId: projection.thread.id })) - .pipe(Effect.mapError(unavailable)); + .pipe(Effect.mapError(dispatchFailure)); return { sequence: result.sequence }; }); @@ -129,7 +130,7 @@ export const ThreadToolkitHandlersLive = ThreadToolkit.toLayer({ createdBy: "agent", creationSource: "mcp", }) - .pipe(Effect.mapError(unavailable)); + .pipe(Effect.mapError(dispatchFailure)); return { sequence: result.sequence, targetThreadId }; }), t3_thread_merge_back: (input) => @@ -145,7 +146,7 @@ export const ThreadToolkitHandlersLive = ThreadToolkit.toLayer({ createdBy: "agent", creationSource: "mcp", }) - .pipe(Effect.mapError(unavailable)); + .pipe(Effect.mapError(dispatchFailure)); return { sequence: result.sequence, targetThreadId: input.targetThreadId }; }), t3_thread_transfers: (input) => @@ -188,7 +189,7 @@ export const ThreadToolkitHandlersLive = ThreadToolkit.toLayer({ commandId: yield* newCommandId(), modelSelection: input.modelSelection, }) - .pipe(Effect.mapError(unavailable)); + .pipe(Effect.mapError(dispatchFailure)); return { sequence: result.sequence }; }), t3_pending_request_list: (input) => @@ -216,7 +217,7 @@ export const ThreadToolkitHandlersLive = ThreadToolkit.toLayer({ requestId: input.requestId, answers: input.answers, }) - .pipe(Effect.mapError(unavailable)); + .pipe(Effect.mapError(dispatchFailure)); return { sequence: result.sequence }; }), t3_queue_list: (input) => @@ -297,7 +298,7 @@ export const ThreadToolkitHandlersLive = ThreadToolkit.toLayer({ default: command = { ...common, type: `thread.${input.action}` }; } - const result = yield* threads.dispatch(command).pipe(Effect.mapError(unavailable)); + const result = yield* threads.dispatch(command).pipe(Effect.mapError(dispatchFailure)); return { sequence: result.sequence }; }), });