feat(mcp): expose existing queued message commands#10555

merged2

agents/mcp-thin/thread-organization

agents/mcp-thin/queue

openedSep 7, 2026byJJulius Marminge

View on GitHubTwo-way sync enabledPR last imported 3d ago

2 unresolved threads · Jump to file

Selected thread: apps/server/src/mcp/toolkits/thread/handlers.ts:28 —  `thread/handlers.ts:28` An inline `import("@t3tools/contracts").CommandId` type expression bypasses the module's import block. Consider adding `type CommandId` to the existing top-level named import from `@t3tools/contracts` (lines 1-7) and referencing it as `commandId: CommandId;` here. No longer relevant as of 90adc5c8bff1c16a8a878c1cce733d8a1e1b3e1c

0 / 6 viewed6 files changedLine totals unavailable

apps/server/src/mcp/toolkits/thread/handlers.ts · Earlier revision 2a263e94 · right 40

Original diff

@@ -1,10 +1,95 @@
-import { OrchestratorMcpFailure, type OrchestrationV2Command } from "@t3tools/contracts";
+import {
+  type ThreadId,
+  type OrchestrationV2ThreadProjection,
+  type RunId,
+  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: import("@t3tools/contracts").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) =>

macroscopeappBotSep 7, 2026

These six new handlers change backend behavior (queue paging via cursor/limit, queueEntry truncation at 1000/16000 characters, not found failure mapping, and command construction for edit/cancel/reorder/promote) but no focused tests were added — toolkits/core.test.ts only covers t3_thread_organize. Consider adding handler tests in the same style as core.test.ts, using Layer.mock(ThreadManagementService) as the test layer for the external service and asserting the real handler output (page shape/nextCursor, truncation flag, and the invalid_request path for an unknown queuedRunId).

^{Posted via Macroscope — Effect Service Conventions}

Julius MarmingeSep 7, 2026

Fixed in 90adc5c8bff1c16a8a878c1cce733d8a1e1b3e1c. CommandId is a normal top-level type import. The existing real MCP/V2 integration now follows both queue pages, verifies the 1,000/16,000 Unicode-code-point output windows, and checks an unknown queued run's invalid_request result. The fixture uses an existing queued run and cleanup path rather than adding six service mocks or duplicating queue execution tests. Integration 2/2 and server typecheck passed; the full stack core/integration batch is 6/6.

apps/server/src/mcp/toolkits/thread/handlers.ts · Earlier revision 2a263e94 · right 28

Original diff

@@ -1,10 +1,95 @@
-import { OrchestratorMcpFailure, type OrchestrationV2Command } from "@t3tools/contracts";
+import {
+  type ThreadId,
+  type OrchestrationV2ThreadProjection,
+  type RunId,
+  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: import("@t3tools/contracts").CommandId;

macroscopeappBotSep 7, 2026

thread/handlers.ts:28

An inline import("@t3tools/contracts").CommandId type expression bypasses the module's import block. Consider adding type CommandId to the existing top-level named import from @t3tools/contracts (lines 1-7) and referencing it as commandId: CommandId; here.

No longer relevant as of 90adc5c8bff1c16a8a878c1cce733d8a1e1b3e1c

Julius MarmingeSep 7, 2026