fix(home): Home reaches environments it has only a peer link to

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

@@ -9,14 +9,24 @@ import {
   ThreadId,
 } from "@t3tools/contracts";
 import * as Cause from "effect/Cause";
+import * as DateTime from "effect/DateTime";
 import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
 import * as Layer from "effect/Layer";
+import * as Option from "effect/Option";
 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 PeerLinks from "../peer/PeerLinks.ts";
+import * as FleetBroker from "../home/FleetBroker.ts";
+import * as FleetService from "../home/FleetService.ts";
+import * as HomeWorkspace from "../home/HomeWorkspace.ts";
+import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
+import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
+import { HomeToolkit } from "./toolkits/home/tools.ts";
+import * as HomeHandlers from "./toolkits/home/handlers.ts";
 import * as HomeService from "../home/HomeService.ts";
 import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
 import * as McpHttpServer from "./McpHttpServer.ts";
@@ -523,3 +533,133 @@ it.effect("keeps Home fleet operations on their existing route", () => {
     );
   }).pipe(Effect.provide(layer));
 });
+
+const LIVE = "2030-01-01T00:00:00Z";
+const EXPIRED = "1960-01-01T00:00:00Z";
+it.effect.each([
+  ["t3_thread_read", "desktop reaches it", true, LIVE, "saw environmentId"],
+  [
+    "t3_thread_read",
+    "only a peer link reaches it",
+    false,
+    LIVE,
+    "t3_thread_read in environment-box",
+  ],
+  ["t3_thread_read", "neither reaches it", false, null, "saw environmentId"],
+  ["t3_thread_read", "its only link expired", false, EXPIRED, "saw environmentId"],
+  // Changes keep the fleet route, which registers Home's report and checks its mode.
+  ["t3_thread_send", "only a peer link reaches it", false, LIVE, "saw environmentId"],
+] as const)(
+  "Home's %s in another environment when %s",
+  ([tool, , desktopReaches, linkExpiresAt, expected]) => {
+    const toolkit = Toolkit.make(
+      Tool.make("t3_thread_read", linkedProbe),
+      Tool.make("t3_thread_send", linkedProbe),
+    );
+    const layer = McpHttpServer.toolkitRegistration(
+      toolkit,
+      McpToolAccess.toLayer(toolkit, {
+        t3_thread_read: McpToolAccess.reads(ranHere),
+        t3_thread_send: McpToolAccess.writes(ranHere),
+      }),
+    ).pipe(
+      Layer.provideMerge(McpHttpServer.layerInProcessMcpServer),
+      Layer.provide(
+        Layer.mock(HomeService.HomeService)({
+          available: true,
+          isHome: () => Effect.succeed(true),
+        }),
+      ),
+      Layer.provide(
+        Layer.mock(FleetBroker.FleetBroker)({
+          reach: Effect.succeed({
+            hostConnected: true,
+            mcpRelay: true,
+            environments: desktopReaches
+              ? [{ environmentId: box, label: "box", connected: true }]
+              : [],
+          }),
+        }),
+      ),
+      Layer.provide(
+        Layer.mock(PeerLinks.PeerLinks)({
+          enabled: Effect.succeed(true),
+          get: (environmentId) =>
+            Effect.succeed(
+              linkExpiresAt !== null && environmentId === box
+                ? Option.some({ expiresAt: DateTime.makeUnsafe(linkExpiresAt) } as never)
+                : Option.none(),
+            ),
+        }),
+      ),
+      Layer.provide(
+        Layer.mock(PeerForwarding.PeerForwarding)({
+          call: (_scope, tool, environmentId) =>
+            Effect.succeed({ ran: `${tool.name} in ${environmentId}` } as never),
+        }),
+      ),
+      Layer.provideMerge(
+        Layer.mock(ThreadManagement.ThreadManagementService)({
+          getThreadShell: (id) => Effect.succeed(shells.get(id) ?? null),
+        }),
+      ),
+    );
+    return call(tool, fullAccess, { environmentId: box }).pipe(
+      Effect.map((outcome) => expect(outcome).toBe(expected)),
+      Effect.provide(layer),
+    );
+  },
+);
+
+it.effect("Home lists environments it reaches only through a peer link", () => {
+  const windows = EnvironmentId.make("environment-windows");
+  const layer = McpHttpServer.toolkitRegistration(HomeToolkit, HomeHandlers.layer).pipe(
+    Layer.provideMerge(McpHttpServer.layerInProcessMcpServer),
+    Layer.provide(NodeCrypto.layer),
+    Layer.provide(Layer.mock(FleetService.FleetService)({})),
+    Layer.provide(Layer.mock(HomeWorkspace.HomeWorkspace)({ root: "/home" })),
+    Layer.provide(
+      Layer.mock(HomeService.HomeService)({ available: true, isHome: () => Effect.succeed(true) }),
+    ),
+    Layer.provide(
+      Layer.mock(ServerEnvironment.ServerEnvironment)({
+        getDescriptor: Effect.succeed({ label: "this mac" } as never),
+      }),
+    ),
+    Layer.provide(
+      Layer.mock(FleetBroker.FleetBroker)({
+        reach: Effect.succeed({
+          hostConnected: true,
+          mcpRelay: true,
+          environments: [{ environmentId: box, label: "box", connected: true }],
+        }),
+      }),
+    ),
+    Layer.provide(
+      Layer.mock(PeerLinks.PeerLinks)({
+        enabled: Effect.succeed(true),
+        list: Effect.succeed([
+          { environmentId: windows, label: "windows", status: "reachable" } as never,
+        ]),
+      }),
+    ),
+    Layer.provideMerge(
+      Layer.mock(ThreadManagement.ThreadManagementService)({
+        getThreadShell: (id) => Effect.succeed(shells.get(id) ?? null),
+      }),
+    ),
+  );
+  return McpServer.McpServer.pipe(
+    Effect.flatMap((server) => server.callTool({ name: "t3_environment_list", arguments: {} })),
+    Effect.map(({ structuredContent }) =>
+      expect(
+        (
+          structuredContent as { environments: ReadonlyArray<{ environmentId: string }> }
+        ).environments.map((entry) => entry.environmentId),
+      ).toEqual([here, box, windows]),
+    ),
+    Effect.provideService(McpInvocationContext.McpInvocationContext, fullAccess),
+    Effect.provideService(McpSchema.McpServerClient, mcpClient),
+    Effect.provide(layer),
+  );
+});

apps/server/src/mcp/McpToolAccess.ts

@@ -7,6 +7,7 @@ import {
   type RuntimeMode,
   type ThreadId,
 } from "@t3tools/contracts";
+import * as DateTime from "effect/DateTime";
 import * as Effect from "effect/Effect";
 import * as Option from "effect/Option";
 import * as Predicate from "effect/Predicate";
@@ -21,7 +22,9 @@ import {
   type DispatchModeRefusal,
 } from "../orchestration-v2/DispatchModeLimit.ts";
 import * as PeerForwarding from "../peer/PeerForwarding.ts";
+import * as FleetBroker from "../home/FleetBroker.ts";
 import * as HomeService from "../home/HomeService.ts";
+import * as PeerLinks from "../peer/PeerLinks.ts";
 import { assertNotLinked, assertSameLink, callerLinkOrigin } from "./linkOrigin.ts";
 import * as McpInvocationContext from "./McpInvocationContext.ts";
 import { resolveInteractionMode, resolveRuntimeMode } from "./delegatedTaskModes.ts";
@@ -376,6 +379,15 @@ const linksUnavailable = new OrchestratorMcpFailure({
  * environment, which checks the rest. The handler here never sees
  * `environmentId`, so it cannot leak into what it writes.
  */
+const HOME_PEER_READS: ReadonlySet<string> = new Set([
+  "orchestrator_capabilities",
+  "t3_project_list",
+  "t3_thread_search",
+  "t3_thread_list",
+  "t3_thread_read",
+  "t3_thread_wait",
+]);
+
 const routed = Effect.fn("McpToolAccess.routed")(function* <Tools extends Record<string, Tool.Any>>(
   toolkit: Toolkit.Toolkit<Tools>,
   declarations: Handlers<Tools>,
@@ -384,6 +396,26 @@ const routed = Effect.fn("McpToolAccess.routed")(function* <Tools extends Record
   // Optional so toolkits that never leave this environment build without it.
   const forwarding = yield* Effect.serviceOption(PeerForwarding.PeerForwarding);
   const home = yield* Effect.serviceOption(HomeService.HomeService);
+  const desktop = yield* Effect.serviceOption(FleetBroker.FleetBroker);
+  const peerLinks = yield* Effect.serviceOption(PeerLinks.PeerLinks);
+  // Home's fleet route reaches only the desktop's connections. Home's reads also use a live
+  // peer link; its changes stay on the fleet route, which keeps its reports and mode checks.
+  const homeReadsThroughPeerLink = Effect.fn("McpToolAccess.homeReadsThroughPeerLink")(function* (
+    name: string,
+    environmentId: EnvironmentId,
+  ) {
+    if (!HOME_PEER_READS.has(name) || Option.isNone(peerLinks)) return false;
+    const link = yield* peerLinks.value.get(environmentId).pipe(Effect.orElseSucceed(Option.none));
+    if (Option.isNone(link) || DateTime.isLessThan(link.value.expiresAt, yield* DateTime.now))
+      return false;
+    if (!(yield* peerLinks.value.enabled)) return false;
+    if (Option.isNone(desktop)) return true;
+    const reach = yield* desktop.value.reach;
+    return !(
+      reach.hostConnected &&
+      reach.environments.some((entry) => entry.environmentId === environmentId && entry.connected)
+    );
+  });
   const handlers: Record<string, unknown> = { ...checked };
   for (const [name, declaration] of Object.entries<Declaration<unknown>>(declarations)) {
     const tool = toolkit.tools[name];
@@ -406,28 +438,30 @@ const routed = Effect.fn("McpToolAccess.routed")(function* <Tools extends Record
           scope.capabilities.has("orchestration") &&
           (yield* home.value.isHome(scope.thread.threadId))
         ) {
+          const fleetTool = [
+            "orchestrator_capabilities",
+            "t3_project_list",
+            "t3_project_create",
+            "t3_project_update",
+            "t3_project_clone",
+            "t3_thread_search",
+            "t3_thread_list",
+            "t3_thread_read",
+            "t3_thread_wait",
+            "t3_thread_launch",
+            "t3_thread_send",
+            "t3_thread_interrupt",
+            "t3_thread_organize",
+            "t3_thread_update",
+            "t3_pending_request",
+          ].includes(name);
           if (
-            [
-              "orchestrator_capabilities",
-              "t3_project_list",
-              "t3_project_create",
-              "t3_project_update",
-              "t3_project_clone",
-              "t3_thread_search",
-              "t3_thread_list",
-              "t3_thread_read",
-              "t3_thread_wait",
-              "t3_thread_launch",
-              "t3_thread_send",
-              "t3_thread_interrupt",
-              "t3_thread_organize",
-              "t3_thread_update",
-              "t3_pending_request",
-            ].includes(name)
+            fleetTool &&
+            (target === undefined || !(yield* homeReadsThroughPeerLink(name, target)))
           ) {
             return yield* here({ ...params, environmentId });
           }
-          if (target !== undefined)
+          if (!fleetTool && target !== undefined)
             return yield* new OrchestratorMcpFailure({
               code: "capability_denied",
               message: "Home cannot use this tool in another environment.",

apps/server/src/mcp/toolkits/core.test.ts

@@ -1079,7 +1079,7 @@ it.effect("thread search links identify the environment that returned them", ()
         Effect.provideService(McpSchema.McpServerClient, client),
       );
     expect(result.structuredContent).toMatchObject({
-      matches: [{ link: "<t3-thread://v2/mcp-core-environment/worker>" }],
+      matches: [{ link: "[Work](t3-thread://v2/mcp-core-environment/worker)" }],
     });
   }).pipe(
     Effect.provide(
@@ -1161,7 +1161,7 @@ it.effect("Home waits for a remote thread through the fleet", () => {
         Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
         Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
         Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})),
-        Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})),
+        Layer.provide(Layer.mock(PeerLinks.PeerLinks)({ get: () => Effect.succeedNone })),
         Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})),
       ),
     ),

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

@@ -1,9 +1,11 @@
 import { OrchestratorMcpFailure } from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
+import * as Option from "effect/Option";
 
 import * as ServerEnvironment from "../../../environment/ServerEnvironment.ts";
 import * as FleetBroker from "../../../home/FleetBroker.ts";
 import * as HomeService from "../../../home/HomeService.ts";
+import * as PeerLinks from "../../../peer/PeerLinks.ts";
 import * as McpToolAccess from "../../McpToolAccess.ts";
 import { callerIsHome, readHomeChangeCaller } from "../../homeRouting.ts";
 import { readCaller, unavailable } from "../../threadAccess.ts";
@@ -29,14 +31,41 @@ export const layer = McpToolAccess.toLayer(HomeToolkit, {
         };
       }
       const reach = yield* (yield* FleetBroker.FleetBroker).reach;
+      const desktop = reach.environments.filter(
+        (candidate) => candidate.environmentId !== scope.environmentId,
+      );
+      // Home's thread tools also reach these through their peer links.
+      const peerLinks = yield* Effect.serviceOption(PeerLinks.PeerLinks);
+      // With linked environments off, every peer call is refused, so none is listed.
+      const linked =
+        Option.isNone(peerLinks) || !(yield* peerLinks.value.enabled)
+          ? []
+          : yield* peerLinks.value.list.pipe(Effect.orElseSucceed(() => []));
       return {
         currentEnvironmentId: scope.environmentId,
         relayConnected: reach.hostConnected,
         environments: [
           current,
-          ...reach.environments
-            .filter((candidate) => candidate.environmentId !== scope.environmentId)
-            .map((candidate) => ({ ...candidate, current: false })),
+          ...desktop.map((candidate) => {
+            const link = linked.find((entry) => entry.environmentId === candidate.environmentId);
+            return {
+              ...candidate,
+              connected: candidate.connected || link?.status === "reachable",
+              current: false,
+            };
+          }),
+          ...linked
+            .filter(
+              (link) =>
+                link.environmentId !== scope.environmentId &&
+                !desktop.some((candidate) => candidate.environmentId === link.environmentId),
+            )
+            .map((link) => ({
+              environmentId: link.environmentId,
+              label: link.label,
+              connected: link.status === "reachable",
+              current: false,
+            })),
         ],
       };
     }),