diff --git a/apps/server/src/mcp/McpToolAccess.test.ts b/apps/server/src/mcp/McpToolAccess.test.ts index 99554b13f0f1e..ab7beda5cd4c6 100644 --- a/apps/server/src/mcp/McpToolAccess.test.ts +++ b/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), + ); +}); diff --git a/apps/server/src/mcp/McpToolAccess.ts b/apps/server/src/mcp/McpToolAccess.ts index f6067f55d3226..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"; @@ -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 = 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, @@ -384,6 +396,26 @@ 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 +438,30 @@ const routed = Effect.fn("McpToolAccess.routed")(function* " }], + 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)({})), ), ), diff --git a/apps/server/src/mcp/toolkits/home/handlers.ts b/apps/server/src/mcp/toolkits/home/handlers.ts index 13a6be79fbaf2..3d42b2db92926 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,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, + })), ], }; }),