Add provider thread IDs and safe cross-thread reads

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

@@ -11,6 +11,14 @@ import { HttpBody, HttpClient, HttpRouter, HttpServerResponse } from "effect/uns
 import * as McpHttpServer from "./McpHttpServer.ts";
 import * as McpInvocationContext from "./McpInvocationContext.ts";
 import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts";
+import {
+  ProjectionSnapshotQuery,
+  type ProjectionSnapshotQueryShape,
+} from "../orchestration/Services/ProjectionSnapshotQuery.ts";
+import {
+  ProviderSessionDirectory,
+  type ProviderSessionDirectoryShape,
+} from "../provider/Services/ProviderSessionDirectory.ts";
 
 const environmentId = EnvironmentId.make("environment-mcp-test");
 const threadId = ThreadId.make("thread-mcp-test");
@@ -38,6 +46,36 @@ const TestLayer = McpHttpServer.PreviewToolkitRegistrationLive.pipe(
   Layer.provideMerge(McpServer.McpServer.layer),
   Layer.provideMerge(PreviewAutomationBroker.layer.pipe(Layer.provide(NodeServices.layer))),
 );
+const unused = () => Effect.die("unused");
+const projectionSnapshotQueryStub: ProjectionSnapshotQueryShape = {
+  getCommandReadModel: unused,
+  getSnapshot: unused,
+  getShellSnapshot: unused,
+  getArchivedShellSnapshot: unused,
+  getSnapshotSequence: unused,
+  getCounts: unused,
+  getActiveProjectByWorkspaceRoot: unused,
+  getProjectShellById: unused,
+  getFirstActiveThreadIdByProjectId: unused,
+  getThreadCheckpointContext: unused,
+  getFullThreadDiffContext: unused,
+  getThreadShellById: unused,
+  getThreadDetailById: unused,
+  getThreadDetailSnapshot: unused,
+};
+const providerSessionDirectoryStub: ProviderSessionDirectoryShape = {
+  upsert: unused,
+  getProvider: unused,
+  getBinding: unused,
+  listThreadIds: unused,
+  listBindings: unused,
+};
+const CombinedToolkitTestLayer = McpHttpServer.ToolkitRegistrationLive.pipe(
+  Layer.provideMerge(McpServer.McpServer.layer),
+  Layer.provideMerge(PreviewAutomationBroker.layer.pipe(Layer.provide(NodeServices.layer))),
+  Layer.provideMerge(Layer.succeed(ProjectionSnapshotQuery, projectionSnapshotQueryStub)),
+  Layer.provideMerge(Layer.succeed(ProviderSessionDirectory, providerSessionDirectoryStub)),
+);
 
 it("normalizes empty successful notification responses to accepted", () => {
   const notificationResponse = McpHttpServer.normalizeMcpHttpResponse(
@@ -51,6 +89,21 @@ it("normalizes empty successful notification responses to accepted", () => {
   expect(resultResponse.status).toBe(200);
 });
 
+it.effect("registers cross-thread reading without replacing preview tools", () =>
+  Effect.gen(function* () {
+    const server = yield* McpServer.McpServer;
+    const toolNames = new Set(server.tools.map(({ tool }) => tool.name));
+    expect(toolNames.has("preview_status")).toBe(true);
+    expect(toolNames.has("preview_snapshot")).toBe(true);
+    expect(toolNames.has("read_thread")).toBe(true);
+
+    const readThreadTool = server.tools.find(({ tool }) => tool.name === "read_thread");
+    expect(readThreadTool?.tool.annotations?.readOnlyHint).toBe(true);
+    expect(readThreadTool?.tool.annotations?.idempotentHint).toBe(true);
+    expect(readThreadTool?.tool.annotations?.destructiveHint).toBe(false);
+  }).pipe(Effect.provide(CombinedToolkitTestLayer)),
+);
+
 it.effect("returns bounded structural preview snapshot failures", () =>
   Effect.scoped(
     Effect.gen(function* () {

apps/server/src/mcp/McpHttpServer.ts

@@ -22,6 +22,8 @@ import {
   PreviewSnapshotToolkit,
   PreviewStandardToolkit,
 } from "./toolkits/preview/tools.ts";
+import { ThreadControlToolkitHandlersLive } from "./toolkits/thread-control/handlers.ts";
+import { ThreadControlToolkit } from "./toolkits/thread-control/tools.ts";
 
 const unauthorized = HttpServerResponse.jsonUnsafe(
   {
@@ -208,10 +210,19 @@ export const PreviewToolkitRegistrationLive = Layer.mergeAll(
   PreviewSnapshotRegistrationLive,
 );
 
+export const ThreadControlToolkitRegistrationLive = McpServer.toolkit(ThreadControlToolkit).pipe(
+  Layer.provide(ThreadControlToolkitHandlersLive),
+);
+
+export const ToolkitRegistrationLive = Layer.mergeAll(
+  PreviewToolkitRegistrationLive,
+  ThreadControlToolkitRegistrationLive,
+);
+
 const McpTransportLive = McpServer.layerHttp({
   name: "T3 Code",
   version: packageJson.version,
   path: "/mcp",
 }).pipe(Layer.provide(McpAuthMiddlewareLive));
 
-export const layer = PreviewToolkitRegistrationLive.pipe(Layer.provideMerge(McpTransportLive));
+export const layer = ToolkitRegistrationLive.pipe(Layer.provideMerge(McpTransportLive));

apps/server/src/mcp/McpInvocationContext.ts

@@ -7,7 +7,7 @@ import {
 import * as Context from "effect/Context";
 import * as Effect from "effect/Effect";
 
-export type McpCapability = "preview";
+export type McpCapability = "preview" | "thread-read";
 
 export interface McpInvocationScope {
   readonly environmentId: EnvironmentId;
@@ -25,7 +25,7 @@ export class McpInvocationContext extends Context.Service<
 >()("t3/mcp/McpInvocationContext") {}
 
 export const requireMcpCapability = Effect.fn("mcp.requireCapability")(function* (
-  capability: McpCapability,
+  capability: "preview",
 ) {
   const invocation = yield* McpInvocationContext;
   if (!invocation.capabilities.has(capability)) {

apps/server/src/mcp/McpSessionRegistry.test.ts

@@ -47,6 +47,7 @@ it.effect("stores only a token hash, resolves the bearer token, and revokes by t
 
     const resolved = yield* registry.resolve(token);
     expect(resolved?.threadId).toBe(threadId);
+    expect(resolved?.capabilities).toEqual(new Set(["preview", "thread-read"]));
 
     yield* registry.revokeThread(threadId);
     expect(yield* registry.resolve(token)).toBeUndefined();

apps/server/src/mcp/McpSessionRegistry.ts

@@ -114,7 +114,7 @@ const makeWithOptions = Effect.fn("McpSessionRegistry.make")(function* (
         threadId: ThreadId.make(request.threadId),
         providerSessionId,
         providerInstanceId: ProviderInstanceId.make(request.providerInstanceId),
-        capabilities: new Set(["preview"]),
+        capabilities: new Set(["preview", "thread-read"]),
         issuedAt,
         expiresAt,
       };

apps/server/src/mcp/toolkits/thread-control/handlers.test.ts

@@ -0,0 +1,209 @@
+import { expect, it } from "@effect/vitest";
+import {
+  EventId,
+  EnvironmentId,
+  MessageId,
+  ProjectId,
+  ProviderDriverKind,
+  ProviderInstanceId,
+  ThreadId,
+  type OrchestrationProjectShell,
+  type OrchestrationThread,
+} from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+import * as Option from "effect/Option";
+
+import * as McpInvocationContext from "../../McpInvocationContext.ts";
+import {
+  ProjectionSnapshotQuery,
+  type ProjectionSnapshotQueryShape,
+} from "../../../orchestration/Services/ProjectionSnapshotQuery.ts";
+import {
+  ProviderSessionDirectory,
+  type ProviderRuntimeBindingWithMetadata,
+  type ProviderSessionDirectoryShape,
+} from "../../../provider/Services/ProviderSessionDirectory.ts";
+import { readThread } from "./handlers.ts";
+
+const NOW = "2026-01-01T00:00:00.000Z";
+const t3ThreadId = ThreadId.make("t3-thread-1");
+const providerThreadId = "019c-native-codex-thread";
+const projectId = ProjectId.make("project-1");
+
+const project: OrchestrationProjectShell = {
+  id: projectId,
+  title: "T3 Code",
+  workspaceRoot: "/workspace/t3code",
+  defaultModelSelection: null,
+  scripts: [],
+  createdAt: NOW,
+  updatedAt: NOW,
+};
+
+const thread: OrchestrationThread = {
+  id: t3ThreadId,
+  projectId,
+  title: "Cross-thread implementation",
+  modelSelection: {
+    instanceId: ProviderInstanceId.make("codex"),
+    model: "gpt-5.4",
+  },
+  runtimeMode: "full-access",
+  interactionMode: "default",
+  branch: null,
+  worktreePath: null,
+  latestTurn: null,
+  createdAt: NOW,
+  updatedAt: NOW,
+  archivedAt: null,
+  settledOverride: null,
+  settledAt: null,
+  snoozedUntil: null,
+  snoozedAt: null,
+  deletedAt: null,
+  messages: [
+    {
+      id: MessageId.make("message-1"),
+      role: "user",
+      text: "first",
+      turnId: null,
+      streaming: false,
+      createdAt: NOW,
+      updatedAt: NOW,
+    },
+    {
+      id: MessageId.make("message-2"),
+      role: "assistant",
+      text: "second",
+      turnId: null,
+      streaming: false,
+      createdAt: NOW,
+      updatedAt: NOW,
+    },
+  ],
+  proposedPlans: [],
+  activities: [
+    {
+      id: EventId.make("activity-1"),
+      tone: "info",
+      kind: "status",
+      summary: "Working",
+      payload: {},
+      turnId: null,
+      createdAt: NOW,
+    },
+  ],
+  checkpoints: [],
+  session: {
+    threadId: t3ThreadId,
+    status: "ready",
+    providerName: "codex",
+    providerInstanceId: ProviderInstanceId.make("codex"),
+    providerThreadId,
+    runtimeMode: "full-access",
+    activeTurnId: null,
+    lastError: null,
+    updatedAt: NOW,
+  },
+};
+
+const binding: ProviderRuntimeBindingWithMetadata = {
+  threadId: t3ThreadId,
+  provider: ProviderDriverKind.make("codex"),
+  providerInstanceId: ProviderInstanceId.make("codex"),
+  resumeCursor: { threadId: providerThreadId },
+  lastSeenAt: NOW,
+};
+
+const invocation = McpInvocationContext.McpInvocationContext.of({
+  environmentId: EnvironmentId.make("environment-1"),
+  threadId: ThreadId.make("requesting-thread"),
+  providerSessionId: "provider-session-1",
+  providerInstanceId: ProviderInstanceId.make("codex"),
+  capabilities: new Set(["preview", "thread-read"] as const),
+  issuedAt: 1,
+  expiresAt: Number.MAX_SAFE_INTEGER,
+});
+
+const unused = () => Effect.die("unused");
+
+function makeSnapshotQuery(): ProjectionSnapshotQueryShape {
+  return {
+    getCommandReadModel: unused,
+    getSnapshot: unused,
+    getShellSnapshot: unused,
+    getArchivedShellSnapshot: unused,
+    getSnapshotSequence: unused,
+    getCounts: unused,
+    getActiveProjectByWorkspaceRoot: unused,
+    getFirstActiveThreadIdByProjectId: unused,
+    getThreadCheckpointContext: unused,
+    getFullThreadDiffContext: unused,
+    getThreadShellById: unused,
+    getThreadDetailSnapshot: unused,
+    getProjectShellById: (id) =>
+      Effect.succeed(id === projectId ? Option.some(project) : Option.none()),
+    getThreadDetailById: (id) =>
+      Effect.succeed(id === t3ThreadId ? Option.some(thread) : Option.none()),
+  };
+}
+
+const sessionDirectory: ProviderSessionDirectoryShape = {
+  upsert: unused,
+  getProvider: unused,
+  getBinding: unused,
+  listThreadIds: unused,
+  listBindings: () => Effect.succeed([binding]),
+};
+
+function runRead(input: Parameters<typeof readThread>[0]) {
+  return readThread(input).pipe(
+    Effect.provideService(McpInvocationContext.McpInvocationContext, invocation),
+    Effect.provideService(ProjectionSnapshotQuery, makeSnapshotQuery()),
+    Effect.provideService(ProviderSessionDirectory, sessionDirectory),
+  );
+}
+
+it.effect("reads a thread by its T3 ID without modifying it", () =>
+  Effect.gen(function* () {
+    const before = structuredClone(thread);
+    const result = yield* runRead({ threadId: t3ThreadId });
+
+    expect(result.threadId).toBe(t3ThreadId);
+    expect(result.providerThreadId).toBe(providerThreadId);
+    expect(result.messages.map(({ text }) => text)).toEqual(["first", "second"]);
+    expect(thread).toEqual(before);
+  }),
+);
+
+it.effect("resolves a copied Codex thread ID and bounds the returned history", () =>
+  Effect.gen(function* () {
+    const result = yield* runRead({
+      threadId: providerThreadId,
+      messageLimit: 1,
+      activityLimit: 1,
+    });
+
+    expect(result.threadId).toBe(t3ThreadId);
+    expect(result.messages.map(({ text }) => text)).toEqual(["second"]);
+    expect(result.totals.messages).toBe(2);
+    expect(result.truncated.messages).toBe(true);
+  }),
+);
+
+it.effect("does not disclose a thread when cross-thread capability is absent", () =>
+  Effect.gen(function* () {
+    const restrictedInvocation = {
+      ...invocation,
+      capabilities: new Set(["preview"] as const),
+    };
+    const error = yield* readThread({ threadId: t3ThreadId }).pipe(
+      Effect.provideService(McpInvocationContext.McpInvocationContext, restrictedInvocation),
+      Effect.provideService(ProjectionSnapshotQuery, makeSnapshotQuery()),
+      Effect.provideService(ProviderSessionDirectory, sessionDirectory),
+      Effect.flip,
+    );
+
+    expect(error._tag).toBe("ThreadReadUnavailableError");
+  }),
+);