# Files changed · PR \#10554 · pingdotgg/t3code

[View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/10554/diffs)

Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code)

Visibility: public

[feat\(mcp\): expose thread organization commands](https://git.cafe/pingdotgg/t3code/pull/10554.md)

Version: 1

Head commit: c03cefcfbabc1310d8c3f824f47d3e3b1115a0d7

Base commit: 4cfe3581d49e2e0e43ff30bae928ca53c3a4bcaa

Comparison base: 4cfe3581d49e2e0e43ff30bae928ca53c3a4bcaa

Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing.

## apps/server/src/mcp/McpHttpServer.ts

Status: modified; +7; −0

```diff
@@ -12,6 +12,8 @@
 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 @@
   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 @@
 export const layer = Layer.mergeAll(
   PreviewToolkitRegistrationLive,
   OrchestratorToolkitRegistrationLive,
+  ThreadToolkitRegistrationLive,
   WorktreeToolkitRegistrationLive,
 ).pipe(Layer.provideMerge(McpTransportLive));
```

## apps/server/src/mcp/OrchestratorMcpService.ts

Status: modified; +2; −2

```diff
@@ -420,7 +420,7 @@
   return mode === "plan" ? 0 : 1;
 }
 
-function resolveRuntimeMode(
+export function resolveRuntimeMode(
   parentMode: RuntimeMode,
   requested: OrchestratorMcpRuntimeMode | undefined,
 ): Effect.Effect<RuntimeMode, OrchestratorMcpFailure> {
@@ -435,7 +435,7 @@
     : Effect.succeed(resolved);
 }
 
-function resolveInteractionMode(
+export function resolveInteractionMode(
   parentMode: ProviderInteractionMode,
   requested: OrchestratorMcpInteractionMode | undefined,
 ): Effect.Effect<ProviderInteractionMode, OrchestratorMcpFailure> {
```

## apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts

Status: modified; +19; −1

```diff
@@ -613,7 +613,10 @@
               runNow: () => Effect.die("ScheduledTaskService.runNow is unused in this test"),
             }),
           );
+          const testLayer = Layer.merge(
+            McpHttpServer.OrchestratorToolkitRegistrationLive,
+            McpHttpServer.ThreadToolkitRegistrationLive,
-          const testLayer = McpHttpServer.OrchestratorToolkitRegistrationLive.pipe(
+          ).pipe(
             Layer.provideMerge(McpServer.McpServer.layer),
             Layer.provideMerge(orchestrationLayer),
             Layer.provide(providerRegistryLayer),
@@ -683,6 +686,12 @@
             const invoke = (name: string, args: Record<string, unknown>) =>
               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."));
             }
@@ -2125,7 +2134,16 @@
               interactionMode: "default",
               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,
             });
```

## apps/server/src/mcp/threadAccess.ts

Status: added; +100; −0

```diff
@@ -1,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)}`);
+});
```

## apps/server/src/mcp/toolkits/core.test.ts

Status: added; +103; −0

```diff
@@ -1,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<string>();
+  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<never>() }),
+        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"),
+                }),
+              ),
+          }),
+        ),
+      ),
+    ),
+  ),
+);
```

## apps/server/src/mcp/toolkits/thread/handlers.ts

Status: added; +36; −0

```diff
@@ -1,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 };
+    }),
+});
```

## apps/server/src/mcp/toolkits/thread/tools.ts

Status: added; +40; −0

```diff
@@ -1,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);
```

## packages/shared/src/t3McpToolPresentation.ts

Status: modified; +1; −0

```diff
@@ -48,6 +48,7 @@
   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" },
```