diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index aeb47996e690..dc8edc58a17f 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -12,6 +12,8 @@ import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstab import packageJson from "../../package.json" with { type: "json" }; import * as McpInvocationContext from "./McpInvocationContext.ts"; import * as OrchestratorMcpService from "./OrchestratorMcpService.ts"; +import { ThreadToolkit } from "./toolkits/thread/tools.ts"; +import { ThreadToolkitHandlersLive } from "./toolkits/thread/handlers.ts"; import * as ThreadMetadataMcpService from "./ThreadMetadataMcpService.ts"; import * as McpSessionRegistry from "./McpSessionRegistry.ts"; import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts"; @@ -228,6 +230,10 @@ export const OrchestratorToolkitRegistrationLive = McpServer.toolkit(Orchestrato Layer.provide(ThreadMetadataMcpService.layer), ); +export const ThreadToolkitRegistrationLive = McpServer.toolkit(ThreadToolkit).pipe( + Layer.provide(ThreadToolkitHandlersLive), +); + export const WorktreeToolkitRegistrationLive = McpServer.toolkit(WorktreeToolkit).pipe( Layer.provide(WorktreeToolkitHandlersLive), Layer.provide(WorktreeMcpService.layer), @@ -243,5 +249,6 @@ const McpTransportLive = McpServer.layerHttp({ export const layer = Layer.mergeAll( PreviewToolkitRegistrationLive, OrchestratorToolkitRegistrationLive, + ThreadToolkitRegistrationLive, WorktreeToolkitRegistrationLive, ).pipe(Layer.provideMerge(McpTransportLive)); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index ea0abc1e9785..f74072573ff6 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -420,7 +420,7 @@ function interactionModeRank(mode: ProviderInteractionMode): number { return mode === "plan" ? 0 : 1; } -function resolveRuntimeMode( +export function resolveRuntimeMode( parentMode: RuntimeMode, requested: OrchestratorMcpRuntimeMode | undefined, ): Effect.Effect { @@ -435,7 +435,7 @@ function resolveRuntimeMode( : Effect.succeed(resolved); } -function resolveInteractionMode( +export function resolveInteractionMode( parentMode: ProviderInteractionMode, requested: OrchestratorMcpInteractionMode | undefined, ): Effect.Effect { diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index d9e06a5bacb6..595d7c72ed97 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -613,7 +613,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.ThreadToolkitRegistrationLive, + ).pipe( Layer.provideMerge(McpServer.McpServer.layer), Layer.provideMerge(orchestrationLayer), Layer.provide(providerRegistryLayer), @@ -683,6 +686,12 @@ describe("orchestrator MCP toolkit", () => { const invoke = (name: string, args: Record) => invokeAs(invocation, name, args); + const pinned = yield* invoke("t3_thread_organize", { action: "pin" }); + expect(pinned.structuredContent).toHaveProperty("sequence"); + expect((yield* orchestrator.getThreadShell(parentThreadId))?.pinnedAt).not.toBeNull(); + yield* invoke("t3_thread_organize", { action: "unpin" }); + expect((yield* orchestrator.getThreadShell(parentThreadId))?.pinnedAt).toBeNull(); + if (parentRun === undefined || parentRun.rootNodeId === null) { return yield* Effect.die(new Error("Parent run missing.")); } @@ -2126,6 +2135,15 @@ describe("orchestrator MCP toolkit", () => { branch: null, worktreePath: cwd, }); + const foreignOrganizeCall = yield* invoke("t3_thread_organize", { + threadId: foreignThreadId, + action: "pin", + }); + expect(foreignOrganizeCall.structuredContent).toMatchObject({ + code: "thread_not_found", + }); + expect((yield* orchestrator.getThreadShell(foreignThreadId))?.pinnedAt).toBeNull(); + const foreignReadCall = yield* invoke("t3_thread_read", { threadId: foreignThreadId, }); diff --git a/apps/server/src/mcp/threadAccess.ts b/apps/server/src/mcp/threadAccess.ts new file mode 100644 index 000000000000..4b3841b41a3f --- /dev/null +++ b/apps/server/src/mcp/threadAccess.ts @@ -0,0 +1,100 @@ +import { + CommandId, + OrchestratorMcpFailure, + type ThreadId, + type OrchestrationV2ThreadShell, +} from "@t3tools/contracts"; +import * as Crypto from "effect/Crypto"; +import * as Effect from "effect/Effect"; + +import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; +import * as OrchestrationMcp from "./OrchestratorMcpService.ts"; +import { type McpInvocationScope, McpInvocationContext } from "./McpInvocationContext.ts"; + +export const unavailable = () => + new OrchestratorMcpFailure({ + code: "orchestration_error", + message: "The operation could not be completed.", + }); + +export const readCaller = Effect.fn("mcp.readCaller")(function* () { + const scope = yield* McpInvocationContext; + if (!scope.capabilities.has("orchestration")) { + return yield* new OrchestratorMcpFailure({ + code: "capability_denied", + message: "This credential cannot control threads.", + }); + } + const threads = yield* ThreadManagement.ThreadManagementService; + const caller = yield* threads.getThreadShell(scope.threadId).pipe(Effect.mapError(unavailable)); + if (caller === null || caller.deletedAt !== null) { + return yield* new OrchestratorMcpFailure({ + code: "thread_not_found", + message: "The calling thread was not found.", + }); + } + return { scope, threads, caller }; +}); + +function assertLiveCaller({ + caller, + scope, +}: { + caller: OrchestrationV2ThreadShell; + scope: McpInvocationScope; +}) { + return caller.archivedAt !== null || + caller.activeRunId === null || + caller.providerInstanceId !== scope.providerInstanceId + ? Effect.fail( + new OrchestratorMcpFailure({ + code: "parent_not_active", + message: "The calling provider no longer owns an active thread run.", + }), + ) + : Effect.void; +} +export const readMutationCaller = Effect.fn("mcp.readMutationCaller")(function* () { + const context = yield* readCaller(); + yield* assertLiveCaller(context); + return context; +}); + +/** Resolve the credential's project before looking up a caller-supplied thread. */ +export const readThread = Effect.fn("mcp.readThread")(function* (threadId?: ThreadId) { + const { scope, threads, caller } = yield* readCaller(); + const projection = yield* threads + .getProjectThread({ projectId: caller.projectId, threadId: threadId ?? caller.id }) + .pipe( + Effect.mapError((error) => + error._tag === "ThreadManagementThreadNotFoundError" + ? new OrchestratorMcpFailure({ + code: "thread_not_found", + message: "The thread was not found in the calling project.", + }) + : unavailable(), + ), + ); + return { scope, threads, caller, projection }; +}); + +export const readWritableThread = Effect.fn("mcp.readWritableThread")(function* ( + threadId?: ThreadId, +) { + const context = yield* readThread(threadId); + yield* assertLiveCaller(context); + yield* OrchestrationMcp.resolveRuntimeMode( + context.caller.runtimeMode, + context.projection.thread.runtimeMode, + ); + yield* OrchestrationMcp.resolveInteractionMode( + context.caller.interactionMode, + context.projection.thread.interactionMode, + ); + return context; +}); + +export const newCommandId = Effect.fn("mcp.newCommandId")(function* () { + const crypto = yield* Crypto.Crypto; + return CommandId.make(`mcp:${yield* crypto.randomUUIDv4.pipe(Effect.orDie)}`); +}); diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts new file mode 100644 index 000000000000..d4e06ea0f2a4 --- /dev/null +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -0,0 +1,103 @@ +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import { expect, it } from "@effect/vitest"; +import { EnvironmentId, ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import { McpSchema, McpServer, Tool } from "effect/unstable/ai"; + +import { OrchestratorProjectionError } from "../../orchestration-v2/Orchestrator.ts"; +import * as ThreadManagement from "../../orchestration-v2/ThreadManagementService.ts"; +import * as McpHttpServer from "../McpHttpServer.ts"; +import { McpInvocationContext, type McpInvocationScope } from "../McpInvocationContext.ts"; +import { OrchestratorToolkit } from "./orchestrator/tools.ts"; +import { PreviewToolkit } from "./preview/tools.ts"; +import { ThreadToolkit } from "./thread/tools.ts"; +import { WorktreeToolkit } from "./worktree/tools.ts"; + +it("publishes unique tool names with object-root inputs", () => { + const names = new Set(); + for (const toolkit of [OrchestratorToolkit, PreviewToolkit, WorktreeToolkit, ThreadToolkit]) { + for (const tool of Object.values(toolkit.tools)) { + expect(names.has(tool.name)).toBe(false); + names.add(tool.name); + expect(Tool.getJsonSchema(tool)).toMatchObject({ type: "object" }); + } + } +}); + +const threadId = ThreadId.make("mcp-core-thread"); +const scope: McpInvocationScope = { + environmentId: EnvironmentId.make("mcp-core-environment"), + threadId, + providerSessionId: "mcp-core-session", + providerInstanceId: ProviderInstanceId.make("codex"), + issuedAt: 0, + capabilities: new Set(["orchestration"]), +}; +const client = McpSchema.McpServerClient.of({ + clientId: 1, + protocolVersion: "2025-06-18", + initializePayload: { + protocolVersion: "2025-06-18", + capabilities: {}, + clientInfo: { name: "mcp-core", version: "1" }, + }, + getClient: Effect.die("unused"), +}); + +it.effect("checks capability before accessing services through the production registration", () => + Effect.gen(function* () { + const server = yield* McpServer.McpServer; + expect(server.tools.some(({ tool }) => tool.name === "t3_thread_organize")).toBe(true); + const result = yield* server + .callTool({ name: "t3_thread_organize", arguments: { action: "pin" } }) + .pipe( + Effect.provideService(McpInvocationContext, { ...scope, capabilities: new Set() }), + Effect.provideService(McpSchema.McpServerClient, client), + ); + expect(result.structuredContent).toMatchObject({ code: "capability_denied" }); + }).pipe( + Effect.provide( + McpHttpServer.ThreadToolkitRegistrationLive.pipe( + Layer.provideMerge(McpServer.McpServer.layer), + Layer.provide(NodeCrypto.layer), + Layer.provide(Layer.mock(ThreadManagement.ThreadManagementService)({})), + ), + ), + ), +); + +it.effect("returns a bounded public failure without serializing storage causes", () => + Effect.gen(function* () { + const server = yield* McpServer.McpServer; + const result = yield* server + .callTool({ name: "t3_thread_organize", arguments: { action: "pin" } }) + .pipe( + Effect.provideService(McpInvocationContext, scope), + Effect.provideService(McpSchema.McpServerClient, client), + ); + expect(result.structuredContent).toEqual({ + _tag: "OrchestratorMcpFailure", + code: "orchestration_error", + message: "The operation could not be completed.", + }); + }).pipe( + Effect.provide( + McpHttpServer.ThreadToolkitRegistrationLive.pipe( + Layer.provideMerge(McpServer.McpServer.layer), + Layer.provide(NodeCrypto.layer), + Layer.provide( + Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadShell: () => + Effect.fail( + new OrchestratorProjectionError({ + threadId, + cause: new Error("private-storage-path"), + }), + ), + }), + ), + ), + ), + ), +); diff --git a/apps/server/src/mcp/toolkits/thread/handlers.ts b/apps/server/src/mcp/toolkits/thread/handlers.ts new file mode 100644 index 000000000000..d0bd35e70bf8 --- /dev/null +++ b/apps/server/src/mcp/toolkits/thread/handlers.ts @@ -0,0 +1,36 @@ +import { OrchestratorMcpFailure, type OrchestrationV2Command } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; + +import { newCommandId, readWritableThread, unavailable } from "../../threadAccess.ts"; +import { ThreadToolkit } from "./tools.ts"; + +export const ThreadToolkitHandlersLive = ThreadToolkit.toLayer({ + t3_thread_organize: (input) => + Effect.gen(function* () { + const { threads, projection } = yield* readWritableThread(input.threadId); + const common = { commandId: yield* newCommandId(), threadId: projection.thread.id }; + let command: OrchestrationV2Command; + switch (input.action) { + case "snooze": + if (input.snoozedUntil === undefined) { + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "snooze requires snoozedUntil.", + }); + } + command = { ...common, type: "thread.snooze", snoozedUntil: input.snoozedUntil }; + break; + case "unsnooze": + case "unsettle": + command = { ...common, type: `thread.${input.action}`, reason: "user" }; + break; + case "mark_unread": + command = { ...common, type: "thread.mark-unread" }; + break; + default: + command = { ...common, type: `thread.${input.action}` }; + } + const result = yield* threads.dispatch(command).pipe(Effect.mapError(unavailable)); + return { sequence: result.sequence }; + }), +}); diff --git a/apps/server/src/mcp/toolkits/thread/tools.ts b/apps/server/src/mcp/toolkits/thread/tools.ts new file mode 100644 index 000000000000..f9b3df9ed68a --- /dev/null +++ b/apps/server/src/mcp/toolkits/thread/tools.ts @@ -0,0 +1,40 @@ +import { + IsoDateTime, + OrchestratorMcpFailure, + OrchestrationV2DispatchCommandResult, + ThreadId, +} from "@t3tools/contracts"; +import * as Crypto from "effect/Crypto"; +import * as Schema from "effect/Schema"; +import { Tool, Toolkit } from "effect/unstable/ai"; + +import { ThreadManagementService } from "../../../orchestration-v2/ThreadManagementService.ts"; +import { McpInvocationContext } from "../../McpInvocationContext.ts"; + +export const ThreadOrganizeTool = Tool.make("t3_thread_organize", { + description: + "Pin, snooze, settle, archive, or mark a thread unread in the calling project. Omit threadId for this thread. snooze requires snoozedUntil. Existing thread lifecycle rules apply; this does not schedule a future action.", + parameters: Schema.Struct({ + threadId: Schema.optional(ThreadId), + action: Schema.Literals([ + "pin", + "unpin", + "snooze", + "unsnooze", + "settle", + "unsettle", + "archive", + "unarchive", + "mark_unread", + ]), + snoozedUntil: Schema.optional(IsoDateTime), + }), + success: OrchestrationV2DispatchCommandResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies: [McpInvocationContext, ThreadManagementService, Crypto.Crypto], +}) + .annotate(Tool.Title, "Organize a thread") + .annotate(Tool.Destructive, true); + +export const ThreadToolkit = Toolkit.make(ThreadOrganizeTool); diff --git a/packages/shared/src/t3McpToolPresentation.ts b/packages/shared/src/t3McpToolPresentation.ts index a9dcd685cc0c..1f14e2a9b3e9 100644 --- a/packages/shared/src/t3McpToolPresentation.ts +++ b/packages/shared/src/t3McpToolPresentation.ts @@ -48,6 +48,7 @@ const T3_MCP_TOOLS: Record< t3_thread_start: { displayName: "Start a T3 thread", summaryAction: "thread-create" }, t3_thread_list: { displayName: "List T3 threads", summaryAction: "thread-list" }, t3_thread_read: { displayName: "Read a T3 thread", summaryAction: "thread-read" }, + t3_thread_organize: { displayName: "Organize a thread" }, t3_thread_update: { displayName: "Update T3 thread metadata" }, t3_thread_send: { displayName: "Send to a T3 thread", summaryAction: "thread-send" }, t3_thread_wait: { displayName: "Wait for a T3 thread", summaryAction: "thread-wait" },