From 5cd15907402350a57ecec123b90164f930c795e1 Mon Sep 17 00:00:00 2001 From: eyeveil <64274427+eyeveil@users.noreply.github.com> Date: Tue, 10 Nov 2026 21:24:51 +0100 Subject: [PATCH 1/3] fix(home): Home reaches environments it has only a peer link to Home's thread, project and capability tools went to the desktop fleet route, which knows only the desktop's connections, so an environment reachable only through a peer link failed with "is not connected to this desktop". Those tools now use the peer link when the desktop cannot reach the environment, and t3_environment_list lists peer-linked environments. --- apps/server/src/mcp/McpToolAccess.test.ts | 106 ++++++++++++++++++ apps/server/src/mcp/McpToolAccess.ts | 57 ++++++---- apps/server/src/mcp/toolkits/core.test.ts | 5 +- apps/server/src/mcp/toolkits/home/handlers.ts | 33 +++++- 4 files changed, 177 insertions(+), 24 deletions(-) diff --git a/apps/server/src/mcp/McpToolAccess.test.ts b/apps/server/src/mcp/McpToolAccess.test.ts index 99554b13f0f1e..37ea1bd79df84 100644 --- a/apps/server/src/mcp/McpToolAccess.test.ts +++ b/apps/server/src/mcp/McpToolAccess.test.ts @@ -12,11 +12,17 @@ import * as Cause from "effect/Cause"; 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 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 +529,103 @@ it.effect("keeps Home fleet operations on their existing route", () => { ); }).pipe(Effect.provide(layer)); }); + +it.effect.each([ + ["desktop reaches it", true, true, "saw environmentId"], + ["only a peer link reaches it", false, true, "t3_thread_read in environment-box"], + ["neither reaches it", false, false, "saw environmentId"], +] as const)("Home reads another environment when %s", ([, desktopReaches, linked, expected]) => { + const toolkit = Toolkit.make(Tool.make("t3_thread_read", linkedProbe)); + const layer = McpHttpServer.toolkitRegistration( + toolkit, + McpToolAccess.toLayer(toolkit, { t3_thread_read: McpToolAccess.reads(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)({ + get: (environmentId) => + Effect.succeed( + linked && environmentId === box ? Option.some({} 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("t3_thread_read", 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( + 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)({ + 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), + ); +}); diff --git a/apps/server/src/mcp/McpToolAccess.ts b/apps/server/src/mcp/McpToolAccess.ts index f6067f55d3226..f6cbf3520cc84 100644 --- a/apps/server/src/mcp/McpToolAccess.ts +++ b/apps/server/src/mcp/McpToolAccess.ts @@ -21,7 +21,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"; @@ -384,6 +386,22 @@ const routed = Effect.fn("McpToolAccess.routed")(function* entry.environmentId === environmentId && entry.connected) + ); + }); const handlers: Record = { ...checked }; for (const [name, declaration] of Object.entries>(declarations)) { const tool = toolkit.tools[name]; @@ -406,28 +424,27 @@ const routed = Effect.fn("McpToolAccess.routed")(function* { 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.succeed(Option.none()) }), + ), Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), ), ), diff --git a/apps/server/src/mcp/toolkits/home/handlers.ts b/apps/server/src/mcp/toolkits/home/handlers.ts index 13a6be79fbaf2..509e7a8e050cc 100644 --- a/apps/server/src/mcp/toolkits/home/handlers.ts +++ b/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,39 @@ 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); + const linked = Option.isNone(peerLinks) + ? [] + : 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, + })), ], }; }), From decbc07a12f7e0845188af64380e560ebfc32906 Mon Sep 17 00:00:00 2001 From: eyeveil <64274427+eyeveil@users.noreply.github.com> Date: Tue, 10 Nov 2026 21:31:06 +0100 Subject: [PATCH 2/3] fix(home): only Home's reads use a peer link, and only a live one Review follow-up: launches, sends and other changes keep the fleet route, which registers Home's reports and checks its mode. An expired link or linked environments turned off keep the fleet route's error. --- apps/server/src/mcp/McpToolAccess.test.ts | 123 +++++++++++++--------- apps/server/src/mcp/McpToolAccess.ts | 27 ++++- 2 files changed, 97 insertions(+), 53 deletions(-) diff --git a/apps/server/src/mcp/McpToolAccess.test.ts b/apps/server/src/mcp/McpToolAccess.test.ts index 37ea1bd79df84..ecfefb249be78 100644 --- a/apps/server/src/mcp/McpToolAccess.test.ts +++ b/apps/server/src/mcp/McpToolAccess.test.ts @@ -9,6 +9,7 @@ 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"; @@ -530,56 +531,82 @@ 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([ - ["desktop reaches it", true, true, "saw environmentId"], - ["only a peer link reaches it", false, true, "t3_thread_read in environment-box"], - ["neither reaches it", false, false, "saw environmentId"], -] as const)("Home reads another environment when %s", ([, desktopReaches, linked, expected]) => { - const toolkit = Toolkit.make(Tool.make("t3_thread_read", linkedProbe)); - const layer = McpHttpServer.toolkitRegistration( - toolkit, - McpToolAccess.toLayer(toolkit, { t3_thread_read: McpToolAccess.reads(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)({ - get: (environmentId) => - Effect.succeed( - linked && environmentId === box ? Option.some({} 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), + ["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), }), - ), - ); - return call("t3_thread_read", fullAccess, { environmentId: box }).pipe( - Effect.map((outcome) => expect(outcome).toBe(expected)), - Effect.provide(layer), - ); -}); + ).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"); diff --git a/apps/server/src/mcp/McpToolAccess.ts b/apps/server/src/mcp/McpToolAccess.ts index f6cbf3520cc84..85f94be49cbec 100644 --- a/apps/server/src/mcp/McpToolAccess.ts +++ b/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"; @@ -378,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 = 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* >( toolkit: Toolkit.Toolkit, declarations: Handlers, @@ -388,13 +398,17 @@ const routed = Effect.fn("McpToolAccess.routed")(function* Date: Tue, 10 Nov 2026 21:34:12 +0100 Subject: [PATCH 3/3] fix(home): Home's environment list hides peers while linked environments are off Also provides the Home services the environment list test needs, and updates the fork's thread search link expectation to the titled link upstream now returns. --- apps/server/src/mcp/McpToolAccess.test.ts | 7 +++++++ apps/server/src/mcp/toolkits/core.test.ts | 7 ++----- apps/server/src/mcp/toolkits/home/handlers.ts | 8 +++++--- 3 files changed, 14 insertions(+), 8 deletions(-) diff --git a/apps/server/src/mcp/McpToolAccess.test.ts b/apps/server/src/mcp/McpToolAccess.test.ts index ecfefb249be78..ab7beda5cd4c6 100644 --- a/apps/server/src/mcp/McpToolAccess.test.ts +++ b/apps/server/src/mcp/McpToolAccess.test.ts @@ -21,6 +21,9 @@ 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"; @@ -612,6 +615,9 @@ 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) }), ), @@ -631,6 +637,7 @@ it.effect("Home lists environments it reaches only through a peer link", () => { ), Layer.provide( Layer.mock(PeerLinks.PeerLinks)({ + enabled: Effect.succeed(true), list: Effect.succeed([ { environmentId: windows, label: "windows", status: "reachable" } as never, ]), diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 9275694741b8c..e7bbd908f5a41 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -25,7 +25,6 @@ import { import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; -import * as Option from "effect/Option"; import * as TestClock from "effect/testing/TestClock"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; @@ -1080,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: "" }], + matches: [{ link: "[Work](t3-thread://v2/mcp-core-environment/worker)" }], }); }).pipe( Effect.provide( @@ -1162,9 +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)({ get: () => Effect.succeed(Option.none()) }), - ), + Layer.provide(Layer.mock(PeerLinks.PeerLinks)({ get: () => Effect.succeedNone })), Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), ), ), diff --git a/apps/server/src/mcp/toolkits/home/handlers.ts b/apps/server/src/mcp/toolkits/home/handlers.ts index 509e7a8e050cc..3d42b2db92926 100644 --- a/apps/server/src/mcp/toolkits/home/handlers.ts +++ b/apps/server/src/mcp/toolkits/home/handlers.ts @@ -36,9 +36,11 @@ export const layer = McpToolAccess.toLayer(HomeToolkit, { ); // Home's thread tools also reach these through their peer links. const peerLinks = yield* Effect.serviceOption(PeerLinks.PeerLinks); - const linked = Option.isNone(peerLinks) - ? [] - : yield* peerLinks.value.list.pipe(Effect.orElseSucceed(() => [])); + // 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,