feat(server): route agent tools through linked environments

apps/server/src/mcp/McpHttpServer.ts

@@ -1048,7 +1048,7 @@ export const layerEnvironmentToolkit = toolkitRegistration(
   EnvironmentHandlers.layer,
 ).pipe(Layer.provide(ThreadCommandExecutor.layer));
 
-const layerProjectRegistration = toolkitRegistration(ProjectToolkit, ProjectHandlers.layer);
+export const layerProjectRegistration = toolkitRegistration(ProjectToolkit, ProjectHandlers.layer);
 
 export const layerAttachmentToolkit = toolkitRegistration(
   AttachmentToolkit,

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

@@ -1,6 +1,7 @@
 import { expect, it } from "@effect/vitest";
 import {
   EnvironmentId,
+  OrchestratorMcpEnvironmentTarget,
   OrchestratorMcpFailure,
   ProviderInstanceId,
   ProviderInteractionMode,
@@ -13,6 +14,7 @@ import * as Schema from "effect/Schema";
 import { McpSchema, McpServer, Tool, Toolkit } from "effect/ai";
 
 import { OrchestratorProjectionError } from "../orchestration-v2/Orchestrator.ts";
+import * as PeerForwarding from "../peer/PeerForwarding.ts";
 import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
 import * as McpHttpServer from "./McpHttpServer.ts";
 import * as McpInvocationContext from "./McpInvocationContext.ts";
@@ -345,3 +347,57 @@ it.effect("refuses a change when the calling thread cannot be read", () =>
     ),
   ),
 );
+
+const LinkedProbeToolkit = Toolkit.make(
+  Tool.make("writes_threads", {
+    ...probe,
+    parameters: Schema.Struct({
+      environmentId: OrchestratorMcpEnvironmentTarget,
+      threadId: Schema.optional(ThreadId),
+    }),
+  }),
+);
+const linkedProbeServer = McpHttpServer.toolkitRegistration(
+  LinkedProbeToolkit,
+  McpToolAccess.toLayer(LinkedProbeToolkit, {
+    writes_threads: McpToolAccess.writesThreads(
+      (input) => [input.threadId],
+      (params) => Effect.succeed({ ran: "environmentId" in params ? "leaked target" : "here" }),
+    ),
+  }),
+).pipe(
+  Layer.provideMerge(McpServer.McpServer.layer),
+  Layer.provideMerge(
+    Layer.mock(ThreadManagement.ThreadManagementService)({
+      getThreadShell: (threadId) => Effect.succeed(shells.get(threadId) ?? null),
+    }),
+  ),
+  Layer.provide(
+    Layer.mock(PeerForwarding.PeerForwarding)({
+      call: () => Effect.succeed({ ran: "there" } as never),
+    }),
+  ),
+);
+
+it.effect(
+  "checks the caller here and the target where it lives, even when thread IDs collide",
+  () =>
+    Effect.gen(function* () {
+      const there = {
+        environmentId: EnvironmentId.make("environment-box"),
+        threadId: fullAccessThreadId,
+      };
+      expect(yield* call("writes_threads", supervised, there)).toBe("there");
+      expect(yield* call("writes_threads", readOnlyClient, there)).toBe("capability_denied");
+      expect(yield* call("writes_threads", ended, there)).toBe("parent_not_active");
+
+      const here = { ...there, environmentId: supervised.environmentId };
+      expect(yield* call("writes_threads", supervised, here)).toBe(
+        "runtime_mode_escalation_denied",
+      );
+      expect(yield* call("writes_threads", fullAccess, here)).toBe("here");
+      expect(yield* call("writes_threads", fullAccess, { threadId: fullAccessThreadId })).toBe(
+        "here",
+      );
+    }).pipe(Effect.provide(linkedProbeServer)),
+);

apps/server/src/mcp/McpToolAccess.ts

@@ -1,10 +1,14 @@
 import {
+  type EnvironmentId,
+  OrchestratorMcpEnvironmentTarget,
   OrchestratorMcpFailure,
   type ProviderInteractionMode,
   type RuntimeMode,
   type ThreadId,
 } from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
+import * as Option from "effect/Option";
+import * as Predicate from "effect/Predicate";
 import * as Ref from "effect/Ref";
 import * as Struct from "effect/Struct";
 import type * as Layer from "effect/Layer";
@@ -15,6 +19,7 @@ import {
   DispatchModeLimit,
   type DispatchModeRefusal,
 } from "../orchestration-v2/DispatchModeLimit.ts";
+import * as PeerForwarding from "../peer/PeerForwarding.ts";
 import * as McpInvocationContext from "./McpInvocationContext.ts";
 import { resolveInteractionMode, resolveRuntimeMode } from "./OrchestratorMcpService.ts";
 import {
@@ -41,8 +46,19 @@ const refuseOutsideConstruction = (token: symbol) => {
 
 let declare: <P, A, E, R>(
   handle: (params: P) => Effect.Effect<A, E, R>,
+  remote: RemoteCheck<P> | undefined,
 ) => Declaration<(params: P) => Effect.Effect<A, E, R>>;
 
+/**
+ * What a call bound for a linked environment checks here before it goes: the
+ * caller, not the thread or project it names, which live there and are
+ * checked there. `undefined` for tools that act as the calling thread itself.
+ */
+type RemoteCheck<P> = (
+  params: P,
+) => Effect.Effect<unknown, OrchestratorMcpFailure, Effect.Services<typeof writingCaller>>;
+let remoteCheckOf: <Handler>(declaration: Declaration<Handler>) => RemoteCheck<never> | undefined;
+
 type CheckedHandlerOf<D> = D extends Declaration<infer Handler> ? Handler : never;
 
 interface CheckedHandler extends Struct.Lambda {
@@ -62,13 +78,20 @@ let handlersLayer: <Tools extends Record<string, Tool.Any>, EX, RX>(
  */
 export class Declaration<out Handler> {
   readonly #handle: Handler;
-  private constructor(token: typeof builtHere, handle: Handler) {
+  readonly #remote: RemoteCheck<never> | undefined;
+  private constructor(
+    token: typeof builtHere,
+    handle: Handler,
+    remote: RemoteCheck<never> | undefined,
+  ) {
     refuseOutsideConstruction(token);
     this.#handle = handle;
+    this.#remote = remote;
   }
   static {
-    declare = (handle) => new Declaration(builtHere, handle);
+    declare = (handle, remote) => new Declaration(builtHere, handle, remote);
     checkedHandler = Struct.lambda<CheckedHandler>((declaration) => declaration.#handle);
+    remoteCheckOf = (declaration) => declaration.#remote;
   }
 }
 
@@ -101,60 +124,73 @@ const requireThreadCaller = McpInvocationContext.McpInvocationContext.pipe(
 );
 
 export const reads = <P, A, E, R>(handle: (params: P) => Effect.Effect<A, E, R>) =>
-  declare((params: P) => handle(params));
+  declare(
+    (params: P) => handle(params),
+    () => Effect.void,
+  );
 
 export const readsAsCaller = <P, A, E, R>(handle: (params: P) => Effect.Effect<A, E, R>) =>
-  declare((params: P) => requireThreadCaller.pipe(Effect.flatMap(() => handle(params))));
+  declare((params: P) => requireThreadCaller.pipe(Effect.flatMap(() => handle(params))), undefined);
 
 export const actsAsCaller = <P, A, E, R>(handle: (params: P) => Effect.Effect<A, E, R>) =>
-  declare((params: P) =>
-    requireThreadCaller.pipe(
-      Effect.flatMap(() => writingCaller),
-      Effect.flatMap(() => handle(params)),
-    ),
+  declare(
+    (params: P) =>
+      requireThreadCaller.pipe(
+        Effect.flatMap(() => writingCaller),
+        Effect.flatMap(() => handle(params)),
+      ),
+    undefined,
   );
 
 export const writes = <P, A, E, R>(handle: (params: P) => Effect.Effect<A, E, R>) =>
-  declare((params: P) => orchestratingCaller.pipe(Effect.flatMap(() => handle(params))));
+  declare(
+    (params: P) => orchestratingCaller.pipe(Effect.flatMap(() => handle(params))),
+    () => orchestratingCaller,
+  );
 
 /**
  * Unlike `writes` it does not need thread control; the handler checks its
  * capability, as the capability toolkits do.
  */
 export const writesWithCapability = <P, A, E, R>(handle: (params: P) => Effect.Effect<A, E, R>) =>
-  declare((params: P) => writingCaller.pipe(Effect.flatMap(() => handle(params))));
+  declare(
+    (params: P) => writingCaller.pipe(Effect.flatMap(() => handle(params))),
+    () => writingCaller,
+  );
 
 export const writesThreads = <P, A, E, R>(
   threads: (params: P) => ReadonlyArray<ThreadId | undefined>,
   handle: (params: P) => Effect.Effect<A, E, R>,
 ) =>
-  declare((params: P) =>
-    Effect.gen(function* () {
-      const caller = yield* writingCaller;
-      for (const threadId of threads(params)) {
-        if (threadId === undefined || threadId === caller.scope.thread?.threadId) continue;
-        const target = yield* caller.threads
-          .getThreadShell(threadId)
-          .pipe(Effect.mapError(unavailable));
-        if (target !== null && target.deletedAt === null) {
-          yield* assertTargetWithinLimits(caller.limits, target);
+  declare(
+    (params: P) =>
+      Effect.gen(function* () {
+        const caller = yield* writingCaller;
+        for (const threadId of threads(params)) {
+          if (threadId === undefined || threadId === caller.scope.thread?.threadId) continue;
+          const target = yield* caller.threads
+            .getThreadShell(threadId)
+            .pipe(Effect.mapError(unavailable));
+          if (target !== null && target.deletedAt === null) {
+            yield* assertTargetWithinLimits(caller.limits, target);
+          }
         }
-      }
-      // The target's user can raise its modes after the check above; the
-      // orchestrator checks again under the thread's lock and records its
-      // refusal here, since handlers wrap dispatch errors their own way.
-      const refused = yield* Ref.make<DispatchModeRefusal | undefined>(undefined);
-      return yield* handle(params).pipe(
-        Effect.provideService(DispatchModeLimit, { ...caller.limits, refused }),
-        Effect.catch((error) =>
-          Effect.flatMap(Ref.get(refused), (refusal) =>
-            Effect.fail<E | OrchestratorMcpFailure>(
-              refusal === undefined ? error : escalationDenied(refusal),
+        // The target's user can raise its modes after the check above; the
+        // orchestrator checks again under the thread's lock and records its
+        // refusal here, since handlers wrap dispatch errors their own way.
+        const refused = yield* Ref.make<DispatchModeRefusal | undefined>(undefined);
+        return yield* handle(params).pipe(
+          Effect.provideService(DispatchModeLimit, { ...caller.limits, refused }),
+          Effect.catch((error) =>
+            Effect.flatMap(Ref.get(refused), (refusal) =>
+              Effect.fail<E | OrchestratorMcpFailure>(
+                refusal === undefined ? error : escalationDenied(refusal),
+              ),
             ),
           ),
-        ),
-      );
-    }),
+        );
+      }),
+    () => writingCaller,
   );
 
 const escalationDenied = (refusal: DispatchModeRefusal) =>
@@ -177,21 +213,24 @@ export const startsThreads = <P, A, E, R>(
     readonly interactionMode?: ProviderInteractionMode | undefined;
   },
   handle: (params: P, modes: StartedModes) => Effect.Effect<A, E, R>,
-) =>
-  declare((params: P) =>
+) => {
+  const started = (params: P) =>
     Effect.gen(function* () {
       const { limits } = yield* writingCaller;
       const requested = modes(params);
-      const started: StartedModes = {
+      return {
         runtimeMode: yield* resolveRuntimeMode(limits.runtimeMode, requested.runtimeMode),
         interactionMode: yield* resolveInteractionMode(
           limits.interactionMode,
           requested.interactionMode,
         ),
-      };
-      return yield* handle(params, started);
-    }),
+      } satisfies StartedModes;
+    });
+  return declare(
+    (params: P) => Effect.flatMap(started(params), (modes) => handle(params, modes)),
+    started,
   );
+};
 
 const fullAccessRequired =
   "Changing projects or environment settings needs a live full-access/default calling thread or a full-access client.";
@@ -209,7 +248,10 @@ export const writesEnvironment = <P, A, E, R>(
   const check = orchestratingCaller.pipe(
     Effect.tap((caller) => assertFullAccess(caller, fullAccessRequired)),
   );
-  return declare((params: P) => check.pipe(Effect.flatMap(() => handle(params, check))));
+  return declare(
+    (params: P) => check.pipe(Effect.flatMap(() => handle(params, check))),
+    () => check,
+  );
 };
 
 type CheckServices = Effect.Services<typeof orchestratingCaller>;
@@ -246,14 +288,74 @@ export class HandlersLayer<Tools extends Record<string, Tool.Any>, EX = never, R
 const checkedHandlers = <Handlers>(declarations: Declarations<Handlers>): Handlers =>
   Struct.map(declarations, checkedHandler);
 
+/**
+ * A tool runs in a linked environment when its parameters take this target
+ * as `environmentId`. Other tools may name an environment for their own
+ * reasons, such as the one to link or to move a thread to.
+ */
+const takesLinkedEnvironment = (tool: Tool.Any) => {
+  const parameters: unknown = tool.parametersSchema;
+  return (
+    Predicate.hasProperty(parameters, "fields") &&
+    Predicate.hasProperty(parameters.fields, "environmentId") &&
+    parameters.fields.environmentId === OrchestratorMcpEnvironmentTarget
+  );
+};
+
+const linksUnavailable = new OrchestratorMcpFailure({
+  code: "orchestration_error",
+  message: "This environment cannot reach linked environments.",
+});
+
+/**
+ * Every tool that takes a linked environment runs there the same way: its
+ * declaration checks the caller here, and the same call goes to the other
+ * environment, which checks the rest. The handler here never sees
+ * `environmentId`, so it cannot leak into what it writes.
+ */
+const routed = <Tools extends Record<string, Tool.Any>>(
+  toolkit: Toolkit.Toolkit<Tools>,
+  declarations: Handlers<Tools>,
+  checked: Toolkit.HandlersFrom<Tools>,
+) =>
+  Effect.gen(function* () {
+    // Optional so toolkits that never leave this environment build without it.
+    const forwarding = yield* Effect.serviceOption(PeerForwarding.PeerForwarding);
+    const handlers: Record<string, unknown> = { ...checked };
+    for (const [name, declaration] of Object.entries<Declaration<unknown>>(declarations)) {
+      const tool = toolkit.tools[name];
+      if (tool === undefined || !takesLinkedEnvironment(tool)) continue;
+      const remote = remoteCheckOf(declaration);
+      if (remote === undefined)
+        return yield* Effect.die(
+          `${name} takes a linked environment, but it acts as its calling thread, which is here.`,
+        );
+      const here = handlers[name] as (
+        params: object,
+      ) => Effect.Effect<unknown, Tool.Failure<Tools[keyof Tools]>>;
+      handlers[name] = ({ environmentId, ...params }: { readonly environmentId?: EnvironmentId }) =>
+        Effect.gen(function* () {
+          const scope = yield* McpInvocationContext.McpInvocationContext;
+          const target = PeerForwarding.remoteTarget(scope, environmentId);
+          if (target === undefined) return yield* here(params);
+          yield* (remote as RemoteCheck<typeof params>)(params);
+          if (Option.isNone(forwarding)) return yield* linksUnavailable;
+          return yield* forwarding.value.call(scope, tool, target, { ...params, environmentId });
+        });
+    }
+    return handlers as Toolkit.HandlersFrom<Tools>;
+  });
+
 export const toLayer = <Tools extends Record<string, Tool.Any>, EX = never, RX = never>(
   toolkit: Toolkit.Toolkit<Tools>,
   build: Handlers<Tools> | Effect.Effect<Handlers<Tools>, EX, RX>,
-): HandlersLayer<Tools, EX, Exclude<RX, Scope.Scope>> =>
-  handlersLayer(
+): HandlersLayer<Tools, EX, Exclude<RX, Scope.Scope>> => {
+  // Checked as soon as they are given, so a forged declaration fails here.
+  const withDeclarations = (declarations: Handlers<Tools>) =>
+    routed(toolkit, declarations, checkedHandlers(declarations));
+  return handlersLayer(
     toolkit.toLayer(
-      Effect.isEffect(build)
-        ? Effect.map(build, checkedHandlers<Toolkit.HandlersFrom<Tools>>)
-        : checkedHandlers(build),
+      Effect.isEffect(build) ? Effect.flatMap(build, withDeclarations) : withDeclarations(build),
     ),
   );
+};