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

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

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

Visibility: public

[feat\(mcp\): expose existing queued message commands](https://git.cafe/pingdotgg/t3code/pull/10555.md)

Version: 4

Head commit: 0a74c9341cb5342e0e5f1d5697eebc5a7dc95a7e

Base commit: c03cefcfbabc1310d8c3f824f47d3e3b1115a0d7

Comparison base: c03cefcfbabc1310d8c3f824f47d3e3b1115a0d7

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

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

Status: modified; +19; −1

```diff
@@ -1047,7 +1047,7 @@
               commandId: CommandId.make("command:mcp-parent:queue-race:user"),
               threadId: parentThreadId,
               messageId: queuedUserMessageId,
-              text: "Queue user follow-up work.",
+              text: "🙂".repeat(16001),
               attachments: [],
               modelSelection: codexSelection,
               dispatchMode: { type: "queue_after_active" },
@@ -1067,6 +1067,24 @@
             if (queuedUserRun === undefined) {
               return yield* Effect.die(new Error("Queued user follow-up missing."));
             }
+            const queueFirstPage = yield* invoke("t3_queue_list", { limit: 1 });
+            expect(queueFirstPage.structuredContent).toMatchObject({
+              items: [{ queuedRunId: queueRace.queuedRun.id }],
+              nextCursor: 1,
+            });
+            const queueSecondPage = yield* invoke("t3_queue_list", { cursor: 1, limit: 1 });
+            expect(queueSecondPage.structuredContent).toEqual({
+              items: [{ queuedRunId: queuedUserRun.id, text: "🙂".repeat(1000), truncated: true }],
+              nextCursor: null,
+            });
+            const queueRead = yield* invoke("t3_queue_read", { queuedRunId: queuedUserRun.id });
+            expect(queueRead.structuredContent).toEqual({
+              queuedRunId: queuedUserRun.id,
+              text: "🙂".repeat(16000),
+              truncated: true,
+            });
+            const missingQueueRead = yield* invoke("t3_queue_read", { queuedRunId: parentRun.id });
+            expect(missingQueueRead.structuredContent).toMatchObject({ code: "invalid_request" });
             const queueRaceStatus = yield* invoke("task_status", { taskId: queueRace.task.id });
             expect(queueRaceStatus.isError).toBe(false);
             yield* waitForProjection(
```

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

Status: modified; +85; −2

```diff
@@ -1,10 +1,93 @@
+import {
+  type CommandId,
+  type ThreadId,
+  type OrchestrationV2ThreadProjection,
+  type RunId,
+  OrchestratorMcpFailure,
+  type OrchestrationV2Command,
+} from "@t3tools/contracts";
-import { OrchestratorMcpFailure, type OrchestrationV2Command } from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
 
-import { newCommandId, readWritableThread, unavailable } from "../../threadAccess.ts";
+import { newCommandId, readThread, readWritableThread, unavailable } from "../../threadAccess.ts";
+import { queuedRunsInDeliveryOrder } from "../../../orchestration-v2/QueuedRunOrder.ts";
 import { ThreadToolkit } from "./tools.ts";
 
+function queueEntry(projection: OrchestrationV2ThreadProjection, runId: RunId, limit: number) {
+  const run = projection.runs.find((run) => run.id === runId && run.status === "queued");
+  const message = projection.messages.find((message) => message.id === run?.userMessageId);
+  if (run === undefined || message === undefined) return undefined;
+  const characters = Array.from(message.text);
+  return {
+    queuedRunId: run.id,
+    text: characters.slice(0, limit).join(""),
+    truncated: characters.length > limit,
+  };
+}
+const dispatch = Effect.fn("mcp.dispatchThreadCommand")(function* (
+  threadId: ThreadId | undefined,
+  command: (common: { commandId: CommandId; threadId: ThreadId }) => OrchestrationV2Command,
+) {
+  const { threads, projection } = yield* readWritableThread(threadId);
+  const result = yield* threads
+    .dispatch(command({ commandId: yield* newCommandId(), threadId: projection.thread.id }))
+    .pipe(Effect.mapError(unavailable));
+  return { sequence: result.sequence };
+});
+
 export const ThreadToolkitHandlersLive = ThreadToolkit.toLayer({
+  t3_queue_list: (input) =>
+    Effect.gen(function* () {
+      const { projection } = yield* readThread(input.threadId);
+      const runs = queuedRunsInDeliveryOrder(projection);
+      const cursor = input.cursor ?? 0;
+      const end = cursor + (input.limit ?? 20);
+      return {
+        items: runs.slice(cursor, end).flatMap((run) => {
+          const entry = queueEntry(projection, run.id, 1000);
+          return entry === undefined ? [] : [entry];
+        }),
+        nextCursor: end < runs.length ? end : null,
+      };
+    }),
+  t3_queue_read: (input) =>
+    Effect.gen(function* () {
+      const { projection } = yield* readThread(input.threadId);
+      const entry = queueEntry(projection, input.queuedRunId, 16000);
+      return (
+        entry ??
+        (yield* new OrchestratorMcpFailure({
+          code: "invalid_request",
+          message: "The queued message was not found.",
+        }))
+      );
+    }),
+  t3_queue_edit: (input) =>
+    dispatch(input.threadId, (common) => ({
+      ...common,
+      type: "queued-run.edit",
+      runId: input.queuedRunId,
+      text: input.text,
+    })),
+  t3_queue_cancel: (input) =>
+    dispatch(input.threadId, (common) => ({
+      ...common,
+      type: "queued-run.cancel",
+      runId: input.queuedRunId,
+    })),
+  t3_queue_reorder: (input) =>
+    dispatch(input.threadId, (common) => ({
+      ...common,
+      type: "queued-run.reorder",
+      runId: input.queuedRunId,
+      beforeRunId: input.beforeRunId,
+    })),
+  t3_queue_promote_to_steer: (input) =>
+    dispatch(input.threadId, (common) => ({
+      ...common,
+      type: "queued-message.promote-to-steer",
+      queuedRunId: input.queuedRunId,
+      targetRunId: input.targetRunId,
+    })),
   t3_thread_organize: (input) =>
     Effect.gen(function* () {
       const { threads, projection } = yield* readWritableThread(input.threadId);
```

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

Status: modified; +74; −2

```diff
@@ -3,6 +3,8 @@
   OrchestratorMcpFailure,
   OrchestrationV2DispatchCommandResult,
   ThreadId,
+  RunId,
+  NonNegativeInt,
 } from "@t3tools/contracts";
 import * as Crypto from "effect/Crypto";
 import * as Schema from "effect/Schema";
@@ -31,10 +33,80 @@
   }),
   success: OrchestrationV2DispatchCommandResult,
   failure: OrchestratorMcpFailure,
+  failureMode: "return" as const,
-  failureMode: "return",
   dependencies: [McpInvocationContext, ThreadManagementService, Crypto.Crypto],
 })
   .annotate(Tool.Title, "Organize a thread")
   .annotate(Tool.Destructive, true);
 
+const queueTarget = { threadId: Schema.optional(ThreadId), queuedRunId: RunId };
+const commandTool = {
+  success: OrchestrationV2DispatchCommandResult,
+  failure: OrchestratorMcpFailure,
+  failureMode: "return" as const,
+  dependencies: [McpInvocationContext, ThreadManagementService, Crypto.Crypto],
+};
+const queueEntry = Schema.Struct({
+  queuedRunId: RunId,
+  text: Schema.String,
+  truncated: Schema.Boolean,
-export const ThreadToolkit = Toolkit.make(ThreadOrganizeTool);
+});
+export const QueueListTool = Tool.make("t3_queue_list", {
+  ...commandTool,
+  description:
+    "List queued messages in delivery order. Results are a live offset page; use t3_thread_read for full thread history.",
+  parameters: Schema.Struct({
+    threadId: Schema.optional(ThreadId),
+    cursor: Schema.optional(NonNegativeInt),
+    limit: Schema.optional(Schema.Int.check(Schema.isBetween({ minimum: 1, maximum: 100 }))),
+  }),
+  success: Schema.Struct({
+    items: Schema.Array(queueEntry),
+    nextCursor: Schema.NullOr(NonNegativeInt),
+  }),
+})
+  .annotate(Tool.Readonly, true)
+  .annotate(Tool.Destructive, false);
+export const QueueReadTool = Tool.make("t3_queue_read", {
+  ...commandTool,
+  description: "Read up to 16,000 characters of a queued message in the calling project.",
+  parameters: Schema.Struct(queueTarget),
+  success: queueEntry,
+})
+  .annotate(Tool.Readonly, true)
+  .annotate(Tool.Destructive, false);
+export const QueueEditTool = Tool.make("t3_queue_edit", {
+  ...commandTool,
+  description:
+    "Replace a queued message's text, preserving its attachments. The service rejects runs that are no longer queued.",
+  parameters: Schema.Struct({
+    ...queueTarget,
+    text: Schema.String.check(Schema.isMaxLength(100000)),
+  }),
+}).annotate(Tool.Destructive, true);
+export const QueueCancelTool = Tool.make("t3_queue_cancel", {
+  ...commandTool,
+  description: "Cancel a queued run using the existing queue command.",
+  parameters: Schema.Struct(queueTarget),
+}).annotate(Tool.Destructive, true);
+export const QueueReorderTool = Tool.make("t3_queue_reorder", {
+  ...commandTool,
+  description: "Move a queued run before another queued run, or to the end with beforeRunId=null.",
+  parameters: Schema.Struct({ ...queueTarget, beforeRunId: Schema.NullOr(RunId) }),
+}).annotate(Tool.Destructive, true);
+export const QueuePromoteTool = Tool.make("t3_queue_promote_to_steer", {
+  ...commandTool,
+  description:
+    "Deliver a queued message as steering to the specified active run. Existing provider and run-state rules apply.",
+  parameters: Schema.Struct({ ...queueTarget, targetRunId: RunId }),
+}).annotate(Tool.Destructive, true);
+
+export const ThreadToolkit = Toolkit.make(
+  ThreadOrganizeTool,
+  QueueListTool,
+  QueueReadTool,
+  QueueEditTool,
+  QueueCancelTool,
+  QueueReorderTool,
+  QueuePromoteTool,
+);
```

## apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts

Status: modified; +5; −1

```diff
@@ -45,6 +45,7 @@
 
 import { attachmentRelativePath } from "../../attachmentStore.ts";
 import * as McpProviderSession from "../../mcp/McpProviderSession.ts";
+import { ThreadToolkit } from "../../mcp/toolkits/thread/tools.ts";
 import { OrchestratorToolkit } from "../../mcp/toolkits/orchestrator/tools.ts";
 import type { EventNdjsonLogger } from "../../provider/Layers/EventNdjsonLogger.ts";
 import {
@@ -580,7 +581,10 @@
   });
 
   it("matches the read-only allowlist to the orchestrator toolkit annotations", () => {
+    const readOnlyToolNames = [
-    const readOnlyToolNames = Object.values(OrchestratorToolkit.tools)
+      ...Object.values(OrchestratorToolkit.tools),
+      ...Object.values(ThreadToolkit.tools),
+    ]
       .filter((tool) => Context.get(tool.annotations, Tool.Readonly))
       .map((tool) => `mcp__t3-code__${tool.name}`)
       .sort();
```

## apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts

Status: modified; +2; −0

```diff
@@ -808,6 +808,8 @@
   "mcp__t3-code__list_scheduled_tasks",
   "mcp__t3-code__t3_thread_list",
   "mcp__t3-code__t3_thread_wait",
+  "mcp__t3-code__t3_queue_list",
+  "mcp__t3-code__t3_queue_read",
 ];
 
 // The SDK's `allowedTools` only pre-approves tool calls; availability is the
```

## packages/shared/src/t3McpToolPresentation.ts

Status: modified; +6; −0

```diff
@@ -48,6 +48,12 @@
   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_queue_list: { displayName: "List queued messages" },
+  t3_queue_read: { displayName: "Read a queued message" },
+  t3_queue_edit: { displayName: "Edit a queued message" },
+  t3_queue_cancel: { displayName: "Cancel a queued run" },
+  t3_queue_reorder: { displayName: "Reorder a queued run" },
+  t3_queue_promote_to_steer: { displayName: "Steer with a queued message" },
   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" },
```