feat(server): route agent work to linked environments

apps/server/src/mcp/McpToolAccess.ts

@@ -1,10 +1,15 @@
 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 PeerForwarding from "../peer/PeerForwarding.ts";
 import * as Ref from "effect/Ref";
 import * as Struct from "effect/Struct";
 import type * as Layer from "effect/Layer";
@@ -41,8 +46,14 @@ const refuseOutsideConstruction = (token: symbol) => {
 
 let declare: <P, A, E, R>(
   handle: (params: P) => Effect.Effect<A, E, R>,
+  remote?: RemoteCheck<P>,
 ) => Declaration<(params: P) => Effect.Effect<A, E, R>>;
 
+type RemoteCheck<P> = (
+  params: P,
+) => Effect.Effect<void, 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,12 +73,19 @@ 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);
+    remoteCheckOf = (declaration) => declaration.#remote;
     checkedHandler = Struct.lambda<CheckedHandler>((declaration) => declaration.#handle);
   }
 }
@@ -101,7 +119,10 @@ 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))));
@@ -115,7 +136,10 @@ export const actsAsCaller = <P, A, E, R>(handle: (params: P) => Effect.Effect<A,
   );
 
 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.pipe(Effect.asVoid),
+  );
 
 /**
  * Unlike `writes` it does not need thread control; the handler checks its
@@ -128,33 +152,35 @@ 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.pipe(Effect.asVoid),
   );
 
 const escalationDenied = (refusal: DispatchModeRefusal) =>
@@ -178,19 +204,27 @@ export const startsThreads = <P, A, E, R>(
   },
   handle: (params: P, modes: StartedModes) => Effect.Effect<A, E, R>,
 ) =>
-  declare((params: P) =>
-    Effect.gen(function* () {
-      const { limits } = yield* writingCaller;
-      const requested = modes(params);
-      const started: StartedModes = {
-        runtimeMode: yield* resolveRuntimeMode(limits.runtimeMode, requested.runtimeMode),
-        interactionMode: yield* resolveInteractionMode(
-          limits.interactionMode,
-          requested.interactionMode,
-        ),
-      };
-      return yield* handle(params, started);
-    }),
+  declare(
+    (params: P) =>
+      Effect.gen(function* () {
+        const { limits } = yield* writingCaller;
+        const requested = modes(params);
+        const started: StartedModes = {
+          runtimeMode: yield* resolveRuntimeMode(limits.runtimeMode, requested.runtimeMode),
+          interactionMode: yield* resolveInteractionMode(
+            limits.interactionMode,
+            requested.interactionMode,
+          ),
+        };
+        return yield* handle(params, started);
+      }),
+    (params: P) =>
+      Effect.gen(function* () {
+        const { limits } = yield* writingCaller;
+        const requested = modes(params);
+        yield* resolveRuntimeMode(limits.runtimeMode, requested.runtimeMode);
+        yield* resolveInteractionMode(limits.interactionMode, requested.interactionMode);
+      }),
   );
 
 const fullAccessRequired =
@@ -209,7 +243,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.pipe(Effect.asVoid),
+  );
 };
 
 type CheckServices = Effect.Services<typeof orchestratingCaller>;
@@ -246,14 +283,77 @@ 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 = environmentId === scope.environmentId ? undefined : 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),
     ),
   );
+};

apps/server/src/mcp/toolkits/environment/handlers.ts

@@ -1,4 +1,6 @@
 import { OrchestratorMcpFailure, type ServerSettings } from "@t3tools/contracts";
+import * as PeerForwarding from "../../../peer/PeerForwarding.ts";
+import * as Option from "effect/Option";
 import * as Effect from "effect/Effect";
 import * as Environment from "../../../environment/ServerEnvironment.ts";
 import * as ThreadCommandExecutor from "../../../orchestration-v2/ThreadCommandExecutor.ts";
@@ -41,6 +43,14 @@ const access = Effect.gen(function* () {
   return { ...context, descriptor, settings: yield* Settings.ServerSettingsService };
 });
 export const layer = McpToolAccess.toLayer(EnvironmentToolkit, {
+  t3_environment_links: McpToolAccess.reads(() =>
+    Effect.gen(function* () {
+      const scope = yield* McpInvocationContext.McpInvocationContext;
+      const forwarding = yield* Effect.serviceOption(PeerForwarding.PeerForwarding);
+      if (Option.isNone(forwarding)) return yield* unavailable();
+      return yield* forwarding.value.links(scope);
+    }),
+  ),
   t3_environment_read: McpToolAccess.reads(() =>
     Effect.gen(function* () {
       const { descriptor, settings } = yield* access;

apps/server/src/mcp/toolkits/environment/tools.ts

@@ -1,4 +1,5 @@
 import {
+  OrchestratorMcpEnvironmentLinksResult,
   BackgroundActivityProfile,
   BackgroundActivityProfileSelection,
   ExecutionEnvironmentDescriptor,
@@ -64,4 +65,17 @@ const EnvironmentPreferencesTool = Tool.make("t3_environment_preferences_update"
   }),
   success: Schema.Struct(PreferenceFields),
 }).annotate(Tool.Destructive, true);
-export const EnvironmentToolkit = Toolkit.make(EnvironmentReadTool, EnvironmentPreferencesTool);
+const EnvironmentLinksTool = Tool.make("t3_environment_links", {
+  ...shared,
+  description:
+    "List this environment's paired destinations without credentials. Reachable means the public descriptor answers with the expected identity, not that access is still authorized.",
+  success: OrchestratorMcpEnvironmentLinksResult,
+})
+  .annotate(Tool.Readonly, true)
+  .annotate(Tool.Destructive, false);
+
+export const EnvironmentToolkit = Toolkit.make(
+  EnvironmentReadTool,
+  EnvironmentPreferencesTool,
+  EnvironmentLinksTool,
+);

apps/server/src/mcp/toolkits/orchestrator/tools.test.ts

@@ -1,5 +1,6 @@
 import { assert, describe, it } from "@effect/vitest";
 import { Tool } from "effect/ai";
+import * as Schema from "effect/Schema";
 
 import {
   CreateThreadsTool,
@@ -9,6 +10,8 @@ import {
   ThreadUpdateTool,
 } from "./tools.ts";
 
+const decodeThreadUpdate = Schema.decodeUnknownResult(ThreadUpdateTool.parametersSchema);
+
 describe("orchestrator MCP tool guidance", () => {
   it("directs subagent requests to delegation instead of ordinary threads", () => {
     assert.include(DelegateTaskTool.description ?? "", "child agent/subagent");
@@ -80,12 +83,22 @@ describe("orchestrator MCP tool guidance", () => {
 
     assert.equal(schema.type, "object");
     assert.hasAllKeys(schema.properties ?? {}, [
+      "environmentId",
       "threadId",
       "action",
       "title",
       "pullRequest",
       "clientRequestId",
     ]);
     assert.include(ThreadUpdateTool.description ?? "", "Workspace and branch changes");
+    assert.deepEqual(
+      [
+        { action: "rename" },
+        { action: "link_pull_request" },
+        { action: "rename", title: "Renamed locally" },
+        { action: "rename", title: "Renamed remotely", environmentId: "destination" },
+      ].map((input) => decodeThreadUpdate(input)._tag),
+      ["Failure", "Failure", "Success", "Success"],
+    );
   });
 });