diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 18919cea55ec..30ccb6fd5210 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -5,6 +5,7 @@ import { authScopeRequiredResponse, AssetCreateUrlInput, AuthAccessReadScope, + AuthAccessWriteScope, ServerSettingsPatch, ProviderInstanceMutation, requiredScopesForServerSettingsPatch, @@ -83,6 +84,9 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.serverGetStorageCleanupReport]: AuthOrchestrationReadScope, [WS_METHODS.serverGetDatabaseStorage]: AuthOrchestrationReadScope, [WS_METHODS.serverGetSettings]: AuthOrchestrationReadScope, + [WS_METHODS.peerLinksList]: AuthAccessReadScope, + [WS_METHODS.peerLinksLink]: AuthAccessWriteScope, + [WS_METHODS.peerLinksUnlink]: AuthAccessWriteScope, [WS_METHODS.serverUpdateSettings]: AuthSettingsWriteScope, [WS_METHODS.serverSearchAcpRegistry]: AuthOrchestrationReadScope, [WS_METHODS.serverPrepareAcpRegistryAgent]: AuthProvidersManageScope, diff --git a/apps/server/src/binCli.ts b/apps/server/src/binCli.ts index b0a2ceffef8f..8ae72f4d7425 100644 --- a/apps/server/src/binCli.ts +++ b/apps/server/src/binCli.ts @@ -10,6 +10,7 @@ import * as NetService from "@t3tools/shared/Net"; import packageJson from "../package.json" with { type: "json" }; import { acpMcpBridgeCommand, acpMcpCallCommand } from "./cli/acpMcpBridge.ts"; import { authCommand } from "./cli/auth.ts"; +import { environmentCommand } from "./cli/environment.ts"; import { appCommand } from "./cli/app.ts"; import { browserCommand } from "./cli/browser.ts"; import { connectCommand } from "./cli/connect.ts"; @@ -74,6 +75,7 @@ export const makeCli = ({ cloudEnabled = hasCloudPublicConfig } = {}) => appCommand, pairCommand, authCommand, + environmentCommand, browserCommand, projectCommand, serviceCommand, diff --git a/apps/server/src/cli/environment.ts b/apps/server/src/cli/environment.ts new file mode 100644 index 000000000000..d831bc44e828 --- /dev/null +++ b/apps/server/src/cli/environment.ts @@ -0,0 +1,158 @@ +import { AuthMcpClientAccess, EnvironmentId, type PeerLinkSummary } from "@t3tools/contracts"; +import * as Console from "effect/Console"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as References from "effect/References"; +import * as Schema from "effect/Schema"; +import * as Argument from "effect/cli/Argument"; +import * as Command from "effect/cli/Command"; +import * as Flag from "effect/cli/Flag"; +import * as GlobalFlag from "effect/cli/GlobalFlag"; +import * as FetchHttpClient from "effect/http/FetchHttpClient"; + +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerConfig from "../config.ts"; +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import * as PeerLinks from "../peer/PeerLinks.ts"; +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import { authLocationFlags, type CliAuthLocationFlags, resolveCliAuthConfig } from "./config.ts"; + +/** + * Links live in this environment's database and secret store, so these + * commands work whether or not its server is running. + */ +const runWithPeerLinks = Effect.fn("environment.runWithPeerLinks")(function* ( + flags: CliAuthLocationFlags, + run: (links: PeerLinks.PeerLinks["Service"]) => Effect.Effect, + options?: { readonly quietLogs?: boolean }, +) { + const logLevel = yield* GlobalFlag.LogLevel; + const config = yield* resolveCliAuthConfig(flags, logLevel); + return yield* PeerLinks.PeerLinks.pipe( + Effect.flatMap(run), + Effect.provide( + PeerLinks.layer.pipe( + Layer.provide( + Layer.mergeAll( + ServerSecretStore.layer, + SqlitePersistence.layerConfig, + FetchHttpClient.layer, + ), + ), + Layer.provide(ServerEnvironment.layer.pipe(Layer.provide(ServerSecretStore.layer))), + Layer.provide(ServerConfig.layer(config)), + Layer.provide( + Layer.succeed(References.MinimumLogLevel, options?.quietLogs ? "Error" : config.logLevel), + ), + ), + ), + ); +}); + +const jsonFlag = Flag.Boolean("json").pipe( + Flag.withDescription("Emit JSON instead of human-readable output."), + Flag.withDefault(false), +); + +const encodeLinks = Schema.encodeEffect( + Schema.fromJsonString(Schema.Array(Schema.toEncoded(Schema.Unknown))), +); + +const formatLinks = (links: ReadonlyArray) => + links.length === 0 + ? "No linked environments.\n" + : `${links + .map( + (link) => + `${link.label} (${link.environmentId})\n ${link.status}, access ${link.access}, expires ${DateTime.formatIso(link.expiresAt)}\n ${link.urls.join(", ")}${link.lastError === null ? "" : `\n last error: ${link.lastError}`}`, + ) + .join("\n\n")}\n`; + +const environmentLinkCommand = Command.make("link", { + ...authLocationFlags, + url: Argument.String("url").pipe( + Argument.withDescription( + "Where the other environment answers: its T3 Connect, Tailscale, LAN or loopback address.", + ), + ), + alternateUrl: Flag.String("url").pipe( + Flag.withDescription("Another address for the same environment, tried after the first."), + Flag.atLeast(0), + ), + pairingCode: Flag.String("pairing-code").pipe( + Flag.withDescription("A pairing code from the other environment (`t3 pair` there)."), + ), + access: Flag.Literals("access", AuthMcpClientAccess.literals).pipe( + Flag.withDescription( + "What this environment's agents may do there: read-only, or the broadest runtime mode.", + ), + Flag.withDefault("approval-required" as const), + ), +}).pipe( + Command.withDescription( + "Link another T3 Code environment, so agents here can work there. It shows up there under Settings → Connections, where it can be revoked.", + ), + Command.withHandler((flags) => + runWithPeerLinks(flags, (links) => + Effect.gen(function* () { + const linked = yield* links.link({ + url: flags.url, + alternateUrls: flags.alternateUrl, + pairingCode: flags.pairingCode, + access: flags.access, + }); + yield* Console.log( + `Linked ${linked.label} (${linked.environmentId}) with ${linked.access} access, until ${DateTime.formatIso(linked.expiresAt)}.\n`, + ); + }), + ), + ), +); + +const environmentListCommand = Command.make("list", { ...authLocationFlags, json: jsonFlag }).pipe( + Command.withDescription("List linked environments and whether each answers now."), + Command.withHandler((flags) => + runWithPeerLinks( + flags, + (links) => + Effect.gen(function* () { + const listed = yield* links.list; + yield* Console.log(flags.json ? `${yield* encodeLinks(listed)}\n` : formatLinks(listed)); + }), + { quietLogs: flags.json }, + ), + ), +); + +const environmentUnlinkCommand = Command.make("unlink", { + ...authLocationFlags, + environmentId: Argument.String("environment-id").pipe( + Argument.withDescription("The linked environment's id, from `t3 environment list`."), + ), +}).pipe( + Command.withDescription( + "Forget a linked environment here. Revoke it in that environment's Connections to end its session there too.", + ), + Command.withHandler((flags) => + runWithPeerLinks(flags, (links) => + Effect.gen(function* () { + const removed = yield* links.unlink(EnvironmentId.make(flags.environmentId)); + yield* Console.log( + removed + ? `Forgot ${flags.environmentId}. Revoke it in that environment's Connections to end its session there.\n` + : `No linked environment ${flags.environmentId}.\n`, + ); + }), + ), + ), +); + +export const environmentCommand = Command.make("environment").pipe( + Command.withDescription("Link other T3 Code environments, so agents here can work there."), + Command.withSubcommands([ + environmentLinkCommand, + environmentListCommand, + environmentUnlinkCommand, + ]), +); diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index fbb5970560bc..00fa09008513 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -266,6 +266,8 @@ export const make = Effect.gen(function* () { : {}), ...(desktopAppUpdate ? { desktopAppUpdate: true } : {}), serverBrowser: true, + peerLinks: true, + mcpModeLimitHeader: true, }, }; diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts index 3f3942e75d08..43554490dc98 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/apps/server/src/observability/RpcInstrumentation.ts @@ -59,6 +59,9 @@ const RPC_AGGREGATES = { [WS_METHODS.serverGetStorageCleanupReport]: "server", [WS_METHODS.serverGetDatabaseStorage]: "server", [WS_METHODS.serverGetSettings]: "server", + [WS_METHODS.peerLinksList]: "peerLinks", + [WS_METHODS.peerLinksLink]: "peerLinks", + [WS_METHODS.peerLinksUnlink]: "peerLinks", [WS_METHODS.serverUpdateSettings]: "server", [WS_METHODS.serverSearchAcpRegistry]: "server", [WS_METHODS.serverPrepareAcpRegistryAgent]: "server", diff --git a/apps/server/src/peer/PeerLinks.test.ts b/apps/server/src/peer/PeerLinks.test.ts new file mode 100644 index 000000000000..47b872b3772c --- /dev/null +++ b/apps/server/src/peer/PeerLinks.test.ts @@ -0,0 +1,397 @@ +import { NodeHttpServer } from "@effect/platform-node"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { + EnvironmentHttpApi, + EnvironmentId, + type ExecutionEnvironmentDescriptor, + OrchestratorMcpFailure, +} from "@t3tools/contracts"; +import { expect, it } from "@effect/vitest"; +import * as Context from "effect/Context"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; +import * as TestClock from "effect/testing/TestClock"; +import { Tool, Toolkit } from "effect/ai"; +import { FetchHttpClient, HttpRouter, HttpServer, HttpServerRequest } from "effect/http"; +import * as HttpApi from "effect/http-api/HttpApi"; +import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder"; + +import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts"; +import * as AuthHttp from "../auth/http.ts"; +import * as McpOAuth from "../auth/McpOAuth.ts"; +import * as McpOAuthHttp from "../auth/mcpOAuthHttp.ts"; +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerConfig from "../config.ts"; +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import * as ServerHttp from "../http.ts"; +import * as McpHttpServer from "../mcp/McpHttpServer.ts"; +import * as McpInvocationContext from "../mcp/McpInvocationContext.ts"; +import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts"; +import * as McpToolAccess from "../mcp/McpToolAccess.ts"; +import * as Sqlite from "../persistence/Sqlite.ts"; +import * as PeerLinks from "./PeerLinks.ts"; +import * as PeerMcpClient from "./PeerMcpClient.ts"; + +class PeerTestApi extends HttpApi.make("environment") + .add(EnvironmentHttpApi.groups.metadata) + .add(EnvironmentHttpApi.groups.mcpOAuth) {} + +const descriptorOf = ( + environmentId: string, + label: string, + capabilities: Partial = {}, +): ExecutionEnvironmentDescriptor => ({ + environmentId: EnvironmentId.make(environmentId), + label, + platform: { os: "linux", arch: "x64" }, + serverVersion: "0.0.0-test", + capabilities: { repositoryIdentity: true, mcpModeLimitHeader: true, ...capabilities }, +}); + +/** What B's probe saw: the calling client and the modes it may start work with. */ +const Probe = Schema.Struct({ caller: Schema.String, modes: Schema.String }); + +const ProbeToolkit = Toolkit.make( + Tool.make("probe", { + description: "Reports the calling client and the modes it may start work with.", + success: Probe, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies: [McpInvocationContext.McpInvocationContext], + }), + Tool.make("refuse", { + description: "Always refuses.", + success: Probe, + failure: OrchestratorMcpFailure, + failureMode: "return", + }), +); + +/** + * Environment B as it runs: its descriptor, MCP OAuth, and `/mcp` behind the + * real client authenticator, on a real socket. The descriptor can be swapped, + * as when B's address comes to answer as another environment. + */ +const serveB = (initial: ExecutionEnvironmentDescriptor) => + Effect.gen(function* () { + const descriptor = yield* Ref.make(initial); + const bearers = yield* Ref.make>([]); + const probes = yield* Ref.make(0); + const layerEnvironment = Layer.succeed(ServerEnvironment.ServerEnvironment, { + getEnvironmentId: Ref.get(descriptor).pipe(Effect.map((current) => current.environmentId)), + getDescriptor: Ref.get(descriptor), + }); + const authContext = yield* EnvironmentAuth.layer.pipe( + Layer.provide(Sqlite.layerMemory), + Layer.provideMerge(ServerSecretStore.layer), + Layer.provideMerge(ServerEnvironment.layerIdentity), + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "t3-peer-link-b-" })), + Layer.provideMerge(NodeServices.layer), + Layer.fresh, + Layer.build, + ); + // Every bearer B receives, on any route, so a test can show one was never sent. + const layerRecordBearers = HttpRouter.middleware( + (httpEffect) => + Effect.gen(function* () { + const request = yield* HttpServerRequest.HttpServerRequest; + if (request.url.startsWith("/.well-known/t3/environment")) { + yield* Ref.update(probes, (count) => count + 1); + } + const authorization = request.headers.authorization; + if (authorization !== undefined) { + yield* Ref.update(bearers, (seen) => [...seen, authorization]); + } + return yield* httpEffect; + }), + { global: true }, + ); + const layerRoutes = Layer.mergeAll( + HttpApiBuilder.layer(PeerTestApi).pipe( + Layer.provide(McpOAuthHttp.layer.pipe(Layer.provide(McpOAuth.layer))), + Layer.provide(ServerHttp.layerServerEnvironmentHttpApi), + Layer.provide(AuthHttp.layerAuthenticatedAuth), + ), + McpHttpServer.toolkitRegistration( + ProbeToolkit, + McpToolAccess.toLayer(ProbeToolkit, { + probe: McpToolAccess.reads(() => + McpInvocationContext.McpInvocationContext.pipe( + Effect.map((scope) => { + const modes = McpInvocationContext.clientModeCeiling(scope.client); + return { + caller: scope.client?.label ?? "none", + modes: `${modes.runtimeMode}/${modes.interactionMode}`, + }; + }), + ), + ), + refuse: McpToolAccess.reads(() => + Effect.fail( + new OrchestratorMcpFailure({ code: "thread_not_found", message: "No such thread." }), + ), + ), + }), + ).pipe( + Layer.provideMerge(McpHttpServer.layerMcpTransport), + Layer.provide( + Layer.mock(McpSessionRegistry.McpSessionRegistry)({ + resolve: () => Effect.succeed(undefined), + }), + ), + Layer.provide(McpOAuth.layerMcpClientAuthenticator), + ), + ).pipe( + Layer.provide(layerRecordBearers), + Layer.provide(layerEnvironment), + Layer.provide(Layer.succeedContext(authContext)), + ); + yield* HttpRouter.serve(layerRoutes, { disableListenLog: true, disableLogger: true }).pipe( + Layer.build, + ); + const address = (yield* HttpServer.HttpServer).address; + if (address._tag === "UnixPathAddress") return yield* Effect.die("expected a TCP address"); + return { + url: `http://127.0.0.1:${address.port}`, + auth: Context.get(authContext, EnvironmentAuth.EnvironmentAuth), + descriptor, + bearers, + /** How often its descriptor was asked for, as a probe of whether it answers. */ + probes, + }; + }); + +/** Environment A: its own database and secret store, reaching B over plain HTTP. */ +const makeA = (descriptor: ExecutionEnvironmentDescriptor) => + Effect.gen(function* () { + const context = yield* PeerMcpClient.layer.pipe( + Layer.provideMerge(PeerLinks.layer), + Layer.provide( + Layer.mergeAll( + Sqlite.layerMemory, + ServerSecretStore.layer, + FetchHttpClient.layer, + Layer.succeed(ServerEnvironment.ServerEnvironment, { + getEnvironmentId: Effect.succeed(descriptor.environmentId), + getDescriptor: Effect.succeed(descriptor), + }), + ), + ), + Layer.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-peer-link-a-" })), + Layer.provide(NodeServices.layer), + Layer.fresh, + Layer.build, + ); + return { + links: Context.get(context, PeerLinks.PeerLinks), + peer: Context.get(context, PeerMcpClient.PeerMcpClient), + }; + }); + +const laptop = descriptorOf("environment-laptop", "Laptop"); +const box = descriptorOf("environment-box", "Box"); +const probe = ( + a: Effect.Success>, + limits: PeerMcpClient.PeerCallLimits = { runtimeMode: "full-access", interactionMode: "default" }, + tool = "probe", +) => + a.peer.call({ + environmentId: box.environmentId, + tool, + arguments: {}, + limits, + success: Probe, + }); + +const linkedSessions = (b: Effect.Success>) => + b.auth + .listSessions() + .pipe( + Effect.map((sessions) => + sessions.filter((session) => session.client.label?.startsWith("T3 Code · ")), + ), + ); + +it.effect("links with a pairing code, and the peer holds its agents to the limits they carry", () => + Effect.scoped( + Effect.gen(function* () { + const b = yield* serveB(box); + const a = yield* makeA(laptop); + + const refused = yield* a.links + .link({ url: b.url, pairingCode: "not-a-code", access: "auto" }) + .pipe(Effect.flip); + expect(refused.reason).toBe("pairing_rejected"); + expect(yield* linkedSessions(b)).toEqual([]); + + const pairing = yield* b.auth.issuePairingCredential(); + const linked = yield* a.links.link({ + url: b.url, + pairingCode: pairing.credential, + access: "auto", + }); + expect(linked).toMatchObject({ + environmentId: box.environmentId, + label: "Box", + urls: [b.url], + access: "auto", + }); + // B lists the link under A's name, where it can be revoked. + expect((yield* linkedSessions(b)).map((session) => session.client.label)).toEqual([ + "T3 Code · Laptop", + ]); + expect( + (yield* a.links.list).map(({ environmentId, status }) => [environmentId, status]), + ).toEqual([[box.environmentId, "reachable"]]); + + // The caller's limits narrow the link's access; they never widen it. + expect( + yield* probe(a, { runtimeMode: "approval-required", interactionMode: "plan" }), + ).toEqual({ caller: "T3 Code · Laptop", modes: "approval-required/plan" }); + expect(yield* probe(a)).toEqual({ caller: "T3 Code · Laptop", modes: "auto/default" }); + + // A refusal keeps the peer's reason. + const notFound = yield* probe(a, undefined, "refuse").pipe(Effect.flip); + expect(notFound).toMatchObject({ code: "thread_not_found", message: "No such thread." }); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect("two links racing to one peer leave the stored access matching the token", () => + Effect.scoped( + Effect.gen(function* () { + const b = yield* serveB(box); + const a = yield* makeA(laptop); + const [first, second] = yield* Effect.all( + [b.auth.issuePairingCredential(), b.auth.issuePairingCredential()], + { concurrency: 2 }, + ); + yield* Effect.all( + [ + a.links.link({ url: b.url, pairingCode: first.credential, access: "read-only" }), + a.links.link({ url: b.url, pairingCode: second.credential, access: "full-access" }), + ], + { concurrency: 2 }, + ); + const [stored] = yield* a.links.list; + // The token sent is the one issued with the access the row records. + const { modes } = yield* probe(a); + expect(modes).toBe( + stored!.access === "read-only" ? "approval-required/default" : "full-access/default", + ); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect("never sends a link's token to an address that stops answering as that peer", () => + Effect.scoped( + Effect.gen(function* () { + const b = yield* serveB(box); + const a = yield* makeA(laptop); + const pairing = yield* b.auth.issuePairingCredential(); + yield* a.links.link({ url: b.url, pairingCode: pairing.credential, access: "auto" }); + yield* probe(a); + expect((yield* Ref.get(b.bearers)).length).toBeGreaterThan(0); + + // The address now answers as another environment, as a reused LAN address would. + yield* Ref.set(b.descriptor, descriptorOf("environment-impostor", "Impostor")); + yield* Ref.set(b.bearers, []); + const failed = yield* probe(a).pipe(Effect.flip); + expect(failed.message).toContain("Box did not answer"); + expect(yield* Ref.get(b.bearers)).toEqual([]); + expect((yield* a.links.list).map((link) => link.status)).toEqual(["unreachable"]); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect("reads one stored link without asking the peer", () => + Effect.scoped( + Effect.gen(function* () { + const b = yield* serveB(box); + const a = yield* makeA(laptop); + const pairing = yield* b.auth.issuePairingCredential(); + yield* a.links.link({ url: b.url, pairingCode: pairing.credential, access: "auto" }); + yield* Ref.set(b.probes, 0); + + const stored = yield* a.links.get(box.environmentId); + expect(Option.map(stored, (link) => link.label)).toEqual(Option.some("Box")); + expect(Option.isNone(yield* a.links.get(laptop.environmentId))).toBe(true); + expect(yield* Ref.get(b.probes)).toBe(0); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect("tells the agent to link again once the peer revokes the link or it expires", () => + Effect.scoped( + Effect.gen(function* () { + const b = yield* serveB(box); + const a = yield* makeA(laptop); + const pairing = yield* b.auth.issuePairingCredential(); + yield* a.links.link({ url: b.url, pairingCode: pairing.credential, access: "auto" }); + yield* probe(a); + + const [session] = yield* linkedSessions(b); + yield* b.auth.revokeSession(session!.sessionId); + const revoked = yield* probe(a).pipe(Effect.flip); + expect(revoked.code).toBe("capability_denied"); + expect(revoked.message).toContain("revoked there"); + + yield* TestClock.adjust(Duration.days(31)); + expect((yield* a.links.list).map((link) => link.status)).toEqual(["expired"]); + yield* Ref.set(b.bearers, []); + const expired = yield* probe(a).pipe(Effect.flip); + expect(expired.code).toBe("capability_denied"); + expect(expired.message).toContain("expired"); + expect(yield* Ref.get(b.bearers)).toEqual([]); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect( + "refuses to link itself, plain http off this machine and the tailnet, or a peer that cannot hold a linked agent to its limits", + () => + Effect.scoped( + Effect.gen(function* () { + const b = yield* serveB( + descriptorOf("environment-box", "Box", { mcpModeLimitHeader: false }), + ); + const a = yield* makeA(laptop); + const pairing = yield* b.auth.issuePairingCredential(); + const older = yield* a.links + .link({ url: b.url, pairingCode: pairing.credential, access: "auto" }) + .pipe(Effect.flip); + expect(older.reason).toBe("incompatible"); + + // Plain http to a LAN address would carry the code and token in clear. + const lan = yield* a.links + .link({ + url: "http://192.168.1.20:3773", + pairingCode: pairing.credential, + access: "auto", + }) + .pipe(Effect.flip); + expect(lan.message).toContain("plain http"); + const lanAlternate = yield* a.links + .link({ + url: b.url, + alternateUrls: ["http://192.168.1.20:3773"], + pairingCode: pairing.credential, + access: "auto", + }) + .pipe(Effect.flip); + expect(lanAlternate.message).toContain("plain http"); + + yield* Ref.set(b.descriptor, laptop); + const self = yield* a.links + .link({ url: b.url, pairingCode: pairing.credential, access: "auto" }) + .pipe(Effect.flip); + expect(self.reason).toBe("self"); + // Neither attempt spent the pairing code on a session. + expect(yield* linkedSessions(b)).toEqual([]); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); diff --git a/apps/server/src/peer/PeerLinks.ts b/apps/server/src/peer/PeerLinks.ts new file mode 100644 index 000000000000..0586e2462365 --- /dev/null +++ b/apps/server/src/peer/PeerLinks.ts @@ -0,0 +1,482 @@ +import { + AuthMcpClientAccess, + AuthMcpRegistrationError, + AuthMcpTokenError, + EnvironmentHttpApi, + EnvironmentId, + type PeerLink, + type PeerLinkCreateInput, + PeerLinkError, + type PeerLinkStatus, + type PeerLinkSummary, +} from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Base64Url from "effect/encoding/Base64Url"; +import * as Hex from "effect/encoding/Hex"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpApiClient from "effect/http-api/HttpApiClient"; +import { isLocalLoopbackHost, isTailnetHost } from "@t3tools/shared/hostClassification"; + +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; + +/** The redirect registered for a link. Nothing ever listens there: the code comes back in the response. */ +const LINK_REDIRECT_URI = "http://127.0.0.1/t3code-peer-link"; +/** + * How long a peer's descriptor may take before a route counts as unreachable. + * A tailnet or relayed route to a machine that is waking up takes longer. + */ +const probeTimeout = (origin: string) => + isLocalLoopbackHost(new URL(origin).hostname) ? Duration.seconds(3) : Duration.seconds(10); + +/** A usable link: where to reach the peer and the token to send there. */ +export interface ResolvedPeerLink { + readonly link: PeerLink; + /** The first address that answered as this peer. */ + readonly url: string; + readonly token: string; +} + +/** + * Links from this environment to others. Linking signs this environment in to + * the peer's `/mcp` as an outside agent with a pairing code from the peer, the + * same MCP OAuth flow an outside agent uses, without a browser. The peer's + * session token is kept in the server secret store; the table keeps the rest. + */ +export class PeerLinks extends Context.Service< + PeerLinks, + { + readonly link: (input: PeerLinkCreateInput) => Effect.Effect; + readonly list: Effect.Effect, PeerLinkError>; + /** The stored link, without asking the peer whether it answers. */ + readonly get: ( + environmentId: EnvironmentId, + ) => Effect.Effect, PeerLinkError>; + /** Forgets the link here. The peer still lists the session until it is revoked there. */ + readonly unlink: (environmentId: EnvironmentId) => Effect.Effect; + /** An address that answers as the linked peer, with its token, ready to call. */ + readonly resolve: ( + environmentId: EnvironmentId, + ) => Effect.Effect; + /** Records the outcome of a call, so listing shows whether the peer answered last time. */ + readonly recordOutcome: ( + environmentId: EnvironmentId, + outcome: { readonly error: string | null }, + ) => Effect.Effect; + } +>()("t3/peer/PeerLinks") {} + +interface PeerLinkRow { + readonly environment_id: string; + readonly secret_name: string; + readonly label: string; + readonly urls_json: string; + readonly access: string; + readonly linked_at: string; + readonly expires_at: string; + readonly last_reached_at: string | null; + readonly last_error: string | null; +} + +const decodeUrls = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Array(Schema.String))); +const encodeUrls = Schema.encodeEffect(Schema.fromJsonString(Schema.Array(Schema.String))); +const decodeAccess = Schema.decodeUnknownEffect(AuthMcpClientAccess); + +/** Fails with a fixed storage error that keeps its cause; the log gets the details. */ +const failStorage = (cause: unknown) => + Effect.logWarning("Peer link storage failed", cause).pipe( + Effect.andThen( + Effect.fail( + new PeerLinkError({ + reason: "storage", + message: + "Linked environments could not be read or saved here. The server log has details.", + cause, + }), + ), + ), + ); + +/** + * The base URL a peer is reached at, normalised to its origin. The pairing + * code and then the link's token travel to it, so plain http is accepted only + * where the network already encrypts: this machine, or a Tailscale address. + */ +const peerOrigin = (url: string) => { + const parsed = URL.parse(url); + if (parsed === null || (parsed.protocol !== "https:" && parsed.protocol !== "http:")) { + return Effect.fail( + new PeerLinkError({ + reason: "unreachable", + message: `${url} is not an http or https address.`, + }), + ); + } + if ( + parsed.protocol === "http:" && + !isLocalLoopbackHost(parsed.hostname) && + !isTailnetHost(parsed.hostname) + ) { + return Effect.fail( + new PeerLinkError({ + reason: "unreachable", + message: `${url} is plain http, which would send the link's credentials unencrypted. Use its https address, or its Tailscale address.`, + }), + ); + } + return Effect.succeed(parsed.origin); +}; + +const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const secrets = yield* ServerSecretStore.ServerSecretStore; + const crypto = yield* Crypto.Crypto; + const httpClient = yield* HttpClient.HttpClient; + const environment = yield* ServerEnvironment.ServerEnvironment; + + /** A fresh name for one link's token; the row records it. */ + const newSecretName = crypto.randomBytes(16).pipe( + Effect.map((bytes) => `peer-link-${Hex.encode(bytes)}`), + Effect.orDie, + ); + + const clientFor = (origin: string) => + HttpApiClient.make(EnvironmentHttpApi, { baseUrl: origin }).pipe( + Effect.provideService(HttpClient.HttpClient, httpClient), + ); + + /** The environment answering at `origin`, or nothing when it does not answer as T3 Code. */ + const describe = (origin: string) => + clientFor(origin).pipe( + Effect.flatMap((client) => client.metadata.descriptor()), + Effect.timeout(probeTimeout(origin)), + Effect.option, + ); + + const toLink = Effect.fn("PeerLinks.toLink")(function* (row: PeerLinkRow) { + return { + environmentId: EnvironmentId.make(row.environment_id), + label: row.label, + urls: yield* decodeUrls(row.urls_json), + access: yield* decodeAccess(row.access), + linkedAt: DateTime.makeUnsafe(row.linked_at), + expiresAt: DateTime.makeUnsafe(row.expires_at), + lastReachedAt: row.last_reached_at === null ? null : DateTime.makeUnsafe(row.last_reached_at), + lastError: row.last_error, + } satisfies PeerLink; + }, Effect.catch(failStorage)); + + const readRows = sql` + SELECT * FROM peer_environment_links ORDER BY label, environment_id + `.pipe(Effect.catch(failStorage)); + + const readRow = (environmentId: EnvironmentId) => + sql` + SELECT * FROM peer_environment_links WHERE environment_id = ${environmentId} + `.pipe( + Effect.catch(failStorage), + Effect.map((rows) => Option.fromNullishOr(rows[0])), + ); + + /** Signs in to the peer's `/mcp` with a pairing code: register, decide, exchange. */ + const signIn = Effect.fn("PeerLinks.signIn")( + function* (origin: string, input: PeerLinkCreateInput, clientName: string) { + const client = yield* clientFor(origin); + // Only the peer's own refusals are rejections. A dropped connection or + // an answer this version cannot read is classified below. + const registered = yield* client.mcpOAuth + .register({ payload: { client_name: clientName, redirect_uris: [LINK_REDIRECT_URI] } }) + .pipe( + // A plain Schema.Error has no tag to catch by. + Effect.catchIf(Schema.is(AuthMcpRegistrationError), (error) => + Effect.fail( + new PeerLinkError({ + reason: "pairing_rejected", + message: `The peer refused to register this environment: ${error.message}`, + }), + ), + ), + ); + const verifier = Base64Url.encode(yield* crypto.randomBytes(32).pipe(Effect.orDie)); + const challenge = Base64Url.encode( + yield* crypto.digest("SHA-256", new TextEncoder().encode(verifier)).pipe(Effect.orDie), + ); + const state = Base64Url.encode(yield* crypto.randomBytes(16).pipe(Effect.orDie)); + const resource = `${origin}/mcp`; + const decided = yield* client.mcpOAuth + .decision({ + payload: { + authorization: { + response_type: "code", + client_id: registered.client_id, + redirect_uri: LINK_REDIRECT_URI, + code_challenge: challenge, + code_challenge_method: "S256", + state, + resource, + }, + decision: { _tag: "pairing-code", access: input.access, code: input.pairingCode }, + }, + }) + .pipe( + Effect.catchTags({ + AuthMcpApprovalError: (error) => + Effect.fail( + new PeerLinkError({ reason: "pairing_rejected", message: error.message }), + ), + }), + ); + const redirect = URL.parse(decided.redirectTo); + if (redirect === null) { + return yield* new PeerLinkError({ + reason: "incompatible", + message: `${origin} answered the sign-in with an address this version cannot read. Update both environments.`, + }); + } + const code = redirect.searchParams.get("code"); + if (code === null || redirect.searchParams.get("state") !== state) { + return yield* new PeerLinkError({ + reason: "pairing_rejected", + message: + redirect.searchParams.get("error_description") ?? + "The peer did not approve the link. Check the pairing code and try again.", + }); + } + return yield* client.mcpOAuth + .token({ + payload: { + grant_type: "authorization_code", + code, + redirect_uri: LINK_REDIRECT_URI, + client_id: registered.client_id, + code_verifier: verifier, + resource, + }, + }) + .pipe( + Effect.catchIf(Schema.is(AuthMcpTokenError), (error) => + Effect.fail( + new PeerLinkError({ + reason: "pairing_rejected", + message: `The peer refused the sign-in: ${error.message}`, + }), + ), + ), + ); + }, + (effect, origin, input, clientName) => + effect.pipe( + // Endpoint errors are mapped above; what is left is the transport or an + // answer this version cannot decode. + Effect.catchTags({ + SchemaError: (cause) => + Effect.fail( + new PeerLinkError({ + reason: "incompatible", + message: `${origin} answered in a shape this version does not understand. Update both environments.`, + cause, + }), + ), + HttpClientError: (cause) => + Effect.fail( + new PeerLinkError({ + reason: "unreachable", + message: `${origin} stopped answering while linking.`, + cause, + }), + ), + }), + ), + ); + + const link: PeerLinks["Service"]["link"] = Effect.fn("PeerLinks.link")(function* (input) { + const origins = yield* Effect.forEach([input.url, ...(input.alternateUrls ?? [])], peerOrigin); + const primary = origins[0]!; + const descriptor = yield* describe(primary); + if (Option.isNone(descriptor)) { + return yield* new PeerLinkError({ + reason: "not_a_t3_environment", + message: `Nothing at ${primary} answered as a T3 Code environment.`, + }); + } + const peer = descriptor.value; + const self = yield* environment.getDescriptor; + if (peer.environmentId === self.environmentId) { + return yield* new PeerLinkError({ + reason: "self", + message: "That address is this environment.", + }); + } + if (peer.capabilities.mcpModeLimitHeader !== true) { + return yield* new PeerLinkError({ + reason: "incompatible", + message: `${peer.label} runs a T3 Code version that cannot keep a linked agent's limits. Update it, then link again.`, + }); + } + const issued = yield* signIn(primary, input, `T3 Code · ${self.label}`); + const now = yield* DateTime.now; + const linked: PeerLink = { + environmentId: peer.environmentId, + label: peer.label, + urls: origins, + access: input.access, + linkedAt: now, + expiresAt: DateTime.add(now, { seconds: issued.expires_in }), + lastReachedAt: now, + lastError: null, + }; + // The token goes under a name of its own before the row points at it, + // so the row and its token are replaced together. + const secretName = yield* newSecretName; + yield* secrets + .set(secretName, new TextEncoder().encode(issued.access_token)) + .pipe(Effect.catch(failStorage)); + const urlsJson = yield* encodeUrls(origins).pipe(Effect.catch(failStorage)); + const previous = yield* readRow(peer.environmentId); + yield* sql` + INSERT INTO peer_environment_links ( + environment_id, secret_name, label, urls_json, access, linked_at, expires_at, + last_reached_at, last_error + ) VALUES ( + ${linked.environmentId}, ${secretName}, ${linked.label}, ${urlsJson}, ${linked.access}, + ${DateTime.formatIso(linked.linkedAt)}, ${DateTime.formatIso(linked.expiresAt)}, + ${DateTime.formatIso(now)}, NULL + ) + ON CONFLICT (environment_id) DO UPDATE SET + secret_name = excluded.secret_name, + label = excluded.label, + urls_json = excluded.urls_json, + access = excluded.access, + linked_at = excluded.linked_at, + expires_at = excluded.expires_at, + last_reached_at = excluded.last_reached_at, + last_error = NULL + `.pipe(Effect.catch(failStorage)); + // Linking again replaces the old token. A racing link may leave its own + // behind; it is unused either way. + if (Option.isSome(previous) && previous.value.secret_name !== secretName) { + yield* secrets.remove(previous.value.secret_name).pipe(Effect.ignore); + } + return linked; + }); + + const statusOf = (link: PeerLink, now: DateTime.Utc) => + DateTime.isLessThanOrEqualTo(link.expiresAt, now) + ? Effect.succeed("expired") + : Effect.findFirst(link.urls, (url) => + describe(url).pipe( + Effect.map( + (found) => Option.isSome(found) && found.value.environmentId === link.environmentId, + ), + ), + ).pipe( + Effect.map((answered): PeerLinkStatus => + Option.isSome(answered) ? "reachable" : "unreachable", + ), + ); + + const list: PeerLinks["Service"]["list"] = Effect.gen(function* () { + const links = yield* Effect.forEach(yield* readRows, toLink); + const now = yield* DateTime.now; + return yield* Effect.forEach( + links, + (link) => statusOf(link, now).pipe(Effect.map((status) => ({ ...link, status }))), + { concurrency: "unbounded" }, + ); + }); + + const get: PeerLinks["Service"]["get"] = (environmentId) => + readRow(environmentId).pipe( + Effect.flatMap( + Option.match({ + onNone: () => Effect.succeed(Option.none()), + onSome: (row) => Effect.map(toLink(row), Option.some), + }), + ), + ); + + const unlink: PeerLinks["Service"]["unlink"] = Effect.fn("PeerLinks.unlink")( + function* (environmentId) { + const deleted = yield* sql<{ secret_name: string }>` + DELETE FROM peer_environment_links WHERE environment_id = ${environmentId} + RETURNING secret_name + `.pipe(Effect.catch(failStorage)); + for (const row of deleted) { + yield* secrets.remove(row.secret_name).pipe(Effect.catch(failStorage)); + } + return deleted.length > 0; + }, + ); + + const resolve: PeerLinks["Service"]["resolve"] = Effect.fn("PeerLinks.resolve")( + function* (environmentId) { + const row = yield* readRow(environmentId); + if (Option.isNone(row)) { + return yield* new PeerLinkError({ + reason: "unknown_link", + message: `This environment is not linked to ${environmentId}. t3_environment_links lists the linked ones.`, + }); + } + const link = yield* toLink(row.value); + if (DateTime.isLessThanOrEqualTo(link.expiresAt, yield* DateTime.now)) { + return yield* new PeerLinkError({ + reason: "expired", + message: `The link to ${link.label} expired. Link it again with a new pairing code from ${link.label}.`, + }); + } + const stored = yield* secrets.get(row.value.secret_name).pipe(Effect.catch(failStorage)); + if (Option.isNone(stored)) { + return yield* new PeerLinkError({ + reason: "unknown_link", + message: `The link to ${link.label} lost its credential. Link it again.`, + }); + } + // The token goes only to an address that proves it is this peer, so a + // reused LAN address or a moved hostname never receives it. + const url = yield* Effect.findFirst(link.urls, (candidate) => + describe(candidate).pipe( + Effect.map( + (found) => Option.isSome(found) && found.value.environmentId === environmentId, + ), + ), + ); + if (Option.isNone(url)) { + return yield* new PeerLinkError({ + reason: "unreachable", + message: `${link.label} did not answer at ${link.urls.join(", ")}.`, + }); + } + return { link, url: url.value, token: new TextDecoder().decode(stored.value) }; + }, + ); + + const recordOutcome: PeerLinks["Service"]["recordOutcome"] = (environmentId, outcome) => + DateTime.now.pipe( + Effect.flatMap((now) => + outcome.error === null + ? sql` + UPDATE peer_environment_links + SET last_reached_at = ${DateTime.formatIso(now)}, last_error = NULL + WHERE environment_id = ${environmentId} + ` + : sql` + UPDATE peer_environment_links SET last_error = ${outcome.error} + WHERE environment_id = ${environmentId} + `, + ), + Effect.asVoid, + Effect.ignoreCause({ log: true }), + ); + + return PeerLinks.of({ link, list, get, unlink, resolve, recordOutcome }); +}); + +export const layer = Layer.effect(PeerLinks, make); diff --git a/apps/server/src/peer/PeerMcpClient.ts b/apps/server/src/peer/PeerMcpClient.ts new file mode 100644 index 000000000000..11b222e570d1 --- /dev/null +++ b/apps/server/src/peer/PeerMcpClient.ts @@ -0,0 +1,253 @@ +import { + type EnvironmentId, + OrchestratorMcpFailure, + type PeerLinkError, + type ProviderInteractionMode, + type RuntimeMode, +} from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; +import * as HttpBody from "effect/http/HttpBody"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientRequest from "effect/http/HttpClientRequest"; + +import * as McpInvocationContext from "../mcp/McpInvocationContext.ts"; +import * as PeerLinks from "./PeerLinks.ts"; + +/** The only MCP protocol version a T3 Code `/mcp` registers. */ +const PROTOCOL_VERSION = "2025-06-18"; + +/** The limits a call carries to the peer: the calling agent's modes here. */ +export interface PeerCallLimits { + readonly runtimeMode: RuntimeMode; + readonly interactionMode: ProviderInteractionMode; +} + +/** + * Calls a linked environment's T3 MCP tools as this environment's outside + * agent. Each call carries the calling agent's limits in `T3-Mode-Limit`, so + * the peer enforces the narrower of those and the link's own access. + */ +export class PeerMcpClient extends Context.Service< + PeerMcpClient, + { + /** Calls `tool` on the peer and decodes its structured result with `success`. */ + readonly call: >(input: { + readonly environmentId: EnvironmentId; + readonly tool: string; + readonly arguments: Record; + readonly limits: PeerCallLimits; + readonly success: S; + }) => Effect.Effect; + } +>()("t3/peer/PeerMcpClient") {} + +const ToolCallResponse = Schema.Struct({ + result: Schema.optional( + Schema.Struct({ + content: Schema.Array( + Schema.Struct({ type: Schema.String, text: Schema.optional(Schema.String) }), + ), + structuredContent: Schema.optional(Schema.Unknown), + isError: Schema.optional(Schema.Boolean), + }), + ), + error: Schema.optional(Schema.Struct({ code: Schema.Number, message: Schema.String })), +}); +const decodeToolCallResponse = Schema.decodeUnknownEffect(Schema.fromJsonString(ToolCallResponse)); +const decodeFailure = Schema.decodeUnknownOption( + Schema.fromJsonString( + Schema.Struct({ code: OrchestratorMcpFailure.fields.code, message: Schema.String }), + ), +); + +/** A peer link problem, as the agent that asked hears it. */ +const fromLinkError = (error: PeerLinkError) => + new OrchestratorMcpFailure({ + code: + error.reason === "unknown_link" + ? "invalid_request" + : error.reason === "expired" + ? "capability_denied" + : "orchestration_error", + message: error.message, + }); + +/** The body of an MCP response: JSON, or a single server-sent event carrying JSON. */ +const responseJson = (contentType: string | undefined, text: string) => + contentType?.includes("text/event-stream") === true + ? text + .split("\n") + .filter((line) => line.startsWith("data:")) + .map((line) => line.slice("data:".length).trim()) + .join("") + : text; + +const make = Effect.gen(function* () { + const links = yield* PeerLinks.PeerLinks; + const httpClient = yield* HttpClient.HttpClient; + // One MCP session per peer and address, reused across calls. The peer keeps + // sessions in memory, so a restart there answers 404 and we start a new one. + const sessions = yield* Ref.make>(new Map()); + + const post = (url: string, token: string, body: unknown, headers: Record) => + httpClient.execute( + HttpClientRequest.post(`${url}/mcp`).pipe( + HttpClientRequest.setHeaders({ + accept: "application/json, text/event-stream", + authorization: `Bearer ${token}`, + ...headers, + }), + HttpClientRequest.setBody(HttpBody.text(JSON.stringify(body), "application/json")), + ), + ); + + const openSession = Effect.fn("PeerMcpClient.openSession")(function* ( + url: string, + token: string, + limitHeader: Record, + ) { + const initialized = yield* post( + url, + token, + { + jsonrpc: "2.0", + id: 0, + method: "initialize", + params: { + protocolVersion: PROTOCOL_VERSION, + capabilities: {}, + clientInfo: { name: "t3code-peer", version: "1" }, + }, + }, + limitHeader, + ); + const sessionId = initialized.headers["mcp-session-id"]; + if (initialized.status !== 200 || sessionId === undefined) { + return yield* new OrchestratorMcpFailure({ + code: initialized.status === 401 ? "capability_denied" : "orchestration_error", + message: + initialized.status === 401 + ? "The linked environment no longer accepts this link. It may have been revoked there; link it again." + : `The linked environment refused to start an MCP session (HTTP ${initialized.status}).`, + }); + } + yield* post( + url, + token, + { jsonrpc: "2.0", method: "notifications/initialized" }, + { ...limitHeader, "mcp-session-id": sessionId, "mcp-protocol-version": PROTOCOL_VERSION }, + ); + return sessionId; + }); + + const call: PeerMcpClient["Service"]["call"] = Effect.fn("PeerMcpClient.call")( + function* (input) { + const resolved = yield* links + .resolve(input.environmentId) + .pipe(Effect.mapError(fromLinkError)); + const limitHeader = { + [McpInvocationContext.MODE_LIMIT_HEADER]: `${input.limits.runtimeMode}/${input.limits.interactionMode}`, + }; + const sessionKey = `${input.environmentId} ${resolved.url}`; + const callOnce = (sessionId: string) => + post( + resolved.url, + resolved.token, + { + jsonrpc: "2.0", + id: 1, + method: "tools/call", + params: { name: input.tool, arguments: input.arguments }, + }, + { ...limitHeader, "mcp-session-id": sessionId, "mcp-protocol-version": PROTOCOL_VERSION }, + ); + const existing = (yield* Ref.get(sessions)).get(sessionKey); + const sessionId = existing ?? (yield* openSession(resolved.url, resolved.token, limitHeader)); + let response = yield* callOnce(sessionId); + if (response.status === 404) { + const renewed = yield* openSession(resolved.url, resolved.token, limitHeader); + yield* Ref.update(sessions, (current) => new Map(current).set(sessionKey, renewed)); + response = yield* callOnce(renewed); + } else if (existing === undefined) { + yield* Ref.update(sessions, (current) => new Map(current).set(sessionKey, sessionId)); + } + if (response.status === 401) { + return yield* new OrchestratorMcpFailure({ + code: "capability_denied", + message: + "The linked environment no longer accepts this link. It may have been revoked there; link it again.", + }); + } + if (response.status !== 200) { + return yield* new OrchestratorMcpFailure({ + code: "orchestration_error", + message: `The linked environment answered HTTP ${response.status}.`, + }); + } + const body = yield* decodeToolCallResponse( + responseJson(response.headers["content-type"], yield* response.text), + ).pipe( + Effect.mapError( + () => + new OrchestratorMcpFailure({ + code: "orchestration_error", + message: "The linked environment answered in an unexpected shape.", + }), + ), + ); + if (body.error !== undefined || body.result === undefined) { + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: body.error?.message ?? "The linked environment returned no result.", + }); + } + const text = body.result.content.map((part) => part.text ?? "").join(""); + if (body.result.isError === true) { + // A refusal keeps the peer's own code, so the agent sees why it was refused. + const refused = decodeFailure(text); + return yield* refused._tag === "Some" + ? new OrchestratorMcpFailure({ code: refused.value.code, message: refused.value.message }) + : new OrchestratorMcpFailure({ + code: "orchestration_error", + message: text || "The linked environment refused the call.", + }); + } + const success = Schema.fromJsonString(input.success); + return yield* Schema.decodeUnknownEffect(success)(text).pipe( + Effect.mapError( + () => + new OrchestratorMcpFailure({ + code: "orchestration_error", + message: + "The linked environment's answer did not match this version. Update both environments.", + }), + ), + ); + }, + (effect, input) => + effect.pipe( + Effect.tapError((error) => + links.recordOutcome(input.environmentId, { error: error.message }), + ), + Effect.tap(() => links.recordOutcome(input.environmentId, { error: null })), + // Recorded above with the transport's own message; the agent hears it plainly. + Effect.catchTags({ + HttpClientError: () => + Effect.fail( + new OrchestratorMcpFailure({ + code: "orchestration_error", + message: "The linked environment stopped answering.", + }), + ), + }), + ), + ); + + return PeerMcpClient.of({ call }); +}); + +export const layer = Layer.effect(PeerMcpClient, make); diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 8dfbfe712a7d..e3f7583fd439 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -70,6 +70,7 @@ import Migration0064 from "./Migrations/064_CanonicalizeThreadPullRequestLinks.t import Migration0065 from "./Migrations/065_CodexUsageNativeTurnIndex.ts"; import Migration0066 from "./Migrations/066_RepairClientSettlementTimestamps.ts"; import Migration0067 from "./Migrations/067_WebhookDedupe.ts"; +import Migration0068 from "./Migrations/068_PeerLinks.ts"; export const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -140,6 +141,7 @@ export const migrationEntries = [ [65, "CodexUsageNativeTurnIndex", Migration0065], [66, "RepairClientSettlementTimestamps", Migration0066], [67, "WebhookDedupe", Migration0067], + [68, "PeerLinks", Migration0068], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts b/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts index 5bbf3cd9a7e6..57f2dc939603 100644 --- a/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts +++ b/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts @@ -15,7 +15,7 @@ layer("055_OrchestrationV2", (it) => { Effect.sync(() => { assert.deepStrictEqual( migrationEntries.map(([id]) => id), - Array.from({ length: 67 }, (_, index) => index + 1), + Array.from({ length: 68 }, (_, index) => index + 1), ); }), ); @@ -41,6 +41,7 @@ layer("055_OrchestrationV2", (it) => { [65, "CodexUsageNativeTurnIndex"], [66, "RepairClientSettlementTimestamps"], [67, "WebhookDedupe"], + [68, "PeerLinks"], ]); assert.deepStrictEqual(yield* runMigrations(), []); @@ -74,6 +75,7 @@ layer("055_OrchestrationV2", (it) => { { migration_id: 65, name: "CodexUsageNativeTurnIndex" }, { migration_id: 66, name: "RepairClientSettlementTimestamps" }, { migration_id: 67, name: "WebhookDedupe" }, + { migration_id: 68, name: "PeerLinks" }, ]); const tables = yield* sql<{ readonly name: string }>` diff --git a/apps/server/src/persistence/Migrations/068_PeerLinks.ts b/apps/server/src/persistence/Migrations/068_PeerLinks.ts new file mode 100644 index 000000000000..5551ccdd07a2 --- /dev/null +++ b/apps/server/src/persistence/Migrations/068_PeerLinks.ts @@ -0,0 +1,24 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Other environments this one signed in to as an outside MCP agent. The + // session token itself is kept in the server secret store under + // `secret_name`, which is unique to each link, so a row always names its + // own token even when two links to one environment race. + yield* sql` + CREATE TABLE IF NOT EXISTS peer_environment_links ( + environment_id TEXT PRIMARY KEY, + secret_name TEXT NOT NULL, + label TEXT NOT NULL, + urls_json TEXT NOT NULL, + access TEXT NOT NULL, + linked_at TEXT NOT NULL, + expires_at TEXT NOT NULL, + last_reached_at TEXT, + last_error TEXT + ) + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 990d450cde6b..edaca42042f2 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -47,6 +47,7 @@ describe("V2 preview upgrade", () => { [65, "CodexUsageNativeTurnIndex"], [66, "RepairClientSettlementTimestamps"], [67, "WebhookDedupe"], + [68, "PeerLinks"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -137,6 +138,7 @@ describe("V2 preview upgrade", () => { [65, "CodexUsageNativeTurnIndex"], [66, "RepairClientSettlementTimestamps"], [67, "WebhookDedupe"], + [68, "PeerLinks"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 48d701103c07..e3809450589d 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -65,6 +65,8 @@ import * as ProviderInstanceRegistryHydration from "./provider/ProviderInstanceR import * as ProjectSettingsOverrideSweep from "./provider/ProjectSettingsOverrideSweep.ts"; import * as TerminalManager from "./terminal/Manager.ts"; import * as McpHttpServer from "./mcp/McpHttpServer.ts"; +import * as PeerLinks from "./peer/PeerLinks.ts"; +import * as PeerMcpClient from "./peer/PeerMcpClient.ts"; import * as McpSessionRegistry from "./mcp/McpSessionRegistry.ts"; import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; import * as DeviceService from "./device/DeviceService.ts"; @@ -722,6 +724,8 @@ const layerMakeRoutes = Layer.mergeAll( Layer.provide(layerPullRequestService), Layer.provide(ServerBrowser.layer.pipe(Layer.provide(DesktopBrowserChannel.layer))), Layer.provide(PreviewBrowser.layer), + // Links to other environments, shared by Settings (WebSocket) and agents (/mcp). + Layer.provide(PeerMcpClient.layer.pipe(Layer.provideMerge(PeerLinks.layer))), Layer.provide(PreviewAutomationBroker.layer), Layer.provide(ServerSelfUpdate.layer.pipe(Layer.provide(layerDesktopAppUpdate))), Layer.provide(layerCommandReadiness), diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 5d9f09260fb5..fa0838ca9308 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -55,6 +55,7 @@ import * as ModelManifest from "./provider/ModelManifest.ts"; import * as Orchestrator from "./orchestration-v2/Orchestrator.ts"; import * as OtlpEndpointCheck from "./observability/OtlpEndpointCheck.ts"; import * as PairingGrantStore from "./auth/PairingGrantStore.ts"; +import * as PeerLinks from "./peer/PeerLinks.ts"; import * as PortScanner from "./preview/PortScanner.ts"; import * as PreviewManager from "./preview/Manager.ts"; import * as ProcessDiagnostics from "./diagnostics/ProcessDiagnostics.ts"; @@ -472,6 +473,7 @@ const layerUnusedConfigRpcServices = Layer.mergeAll( Layer.mock(Orchestrator.OrchestratorV2)({}), Layer.mock(OtlpEndpointCheck.OtlpEndpointCheck)({}), Layer.mock(PairingGrantStore.PairingGrantStore)({}), + Layer.mock(PeerLinks.PeerLinks)({}), Layer.mock(PortScanner.PortDiscovery)({}), Layer.mock(PreviewManager.PreviewManager)({}), Layer.mock(ProcessDiagnostics.ProcessDiagnostics)({}), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 90dc72b0d8a0..5d6890841ac2 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1,6 +1,7 @@ import { OrchestrationDispatchCommandError } from "@t3tools/contracts"; import * as Crypto from "effect/Crypto"; import * as Orchestrator from "./orchestration-v2/Orchestrator.ts"; +import * as PeerLinks from "./peer/PeerLinks.ts"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; @@ -1284,6 +1285,7 @@ export const layerWsRpc = ( const storageCleanup = yield* StorageCleanup.StorageCleanup; const databaseStorage = yield* DatabaseStorage.DatabaseStorage; const serverSettings = yield* ServerSettings.ServerSettingsService; + const peerLinks = yield* PeerLinks.PeerLinks; const startup = yield* ServerRuntimeStartup.ServerRuntimeStartup; const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries; const workspaceFileSystem = yield* WorkspaceFileSystem.WorkspaceFileSystem; @@ -2380,6 +2382,11 @@ export const layerWsRpc = ( [WS_METHODS.serverRunStorageCleanup]: () => storageCleanup.runNow, [WS_METHODS.serverGetStorageCleanupReport]: () => storageCleanup.reports, [WS_METHODS.serverGetDatabaseStorage]: () => databaseStorage.status, + [WS_METHODS.peerLinksList]: (_input) => + peerLinks.list.pipe(Effect.map((links) => ({ links }))), + [WS_METHODS.peerLinksLink]: (input) => peerLinks.link(input), + [WS_METHODS.peerLinksUnlink]: ({ environmentId }) => + peerLinks.unlink(environmentId).pipe(Effect.map((removed) => ({ removed }))), [WS_METHODS.serverGetSettings]: (_input) => serverSettings.getSettings.pipe(Effect.map(ServerSettings.redactServerSettingsForClient)), [WS_METHODS.serverUpdateSettings]: @@ -3138,6 +3145,7 @@ export const layer = Layer.unwrap( const serverSelfUpdate = yield* ServerSelfUpdate.ServerSelfUpdate; const pullRequests = yield* PullRequestService.PullRequestService; const reviewGuides = yield* PullRequestReviewGuideService.PullRequestReviewGuideService; + const peerLinks = yield* PeerLinks.PeerLinks; const sql = yield* SqlClient.SqlClient; return HttpRouter.add( "GET", @@ -3229,6 +3237,7 @@ export const layer = Layer.unwrap( reviewGuides, ), ), + Layer.provide(Layer.succeed(PeerLinks.PeerLinks, peerLinks)), Layer.provide( SourceControlDiscovery.layer.pipe( Layer.provide( diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 4272c9d12d94..2c62bf548bad 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -152,6 +152,14 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ "server"`) and streams them over `/api/preview-stream`. Clients without a local browser runtime open server tabs here. */ serverBrowser: Schema.optionalKey(Schema.Boolean), + /** Server links to other environments (`PeerLink`) and its agents can work + in them. Absent on servers from before linking, which clients must not + offer linking for. */ + peerLinks: Schema.optionalKey(Schema.Boolean), + /** Server's `/mcp` narrows an OAuth client's modes from `T3-Mode-Limit`, so + a linked environment's calls keep its agent's limits. Another environment + links to this one only when it is set. */ + mcpModeLimitHeader: Schema.optionalKey(Schema.Boolean), }); export type ExecutionEnvironmentCapabilities = typeof ExecutionEnvironmentCapabilities.Type; diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 7f38a38dbbae..d9680e959287 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -39,6 +39,7 @@ export * from "./orchestrationProject.ts"; export * from "./orchestrationV2.ts"; export * from "./applicationEvent.ts"; export * from "./orchestratorMcp.ts"; +export * from "./peerLink.ts"; export * from "./threadMetadataMcp.ts"; export * from "./threadPullRequest.ts"; export * from "./threadSearch.ts"; diff --git a/packages/contracts/src/peerLink.ts b/packages/contracts/src/peerLink.ts new file mode 100644 index 000000000000..022b1b9b9f8c --- /dev/null +++ b/packages/contracts/src/peerLink.ts @@ -0,0 +1,66 @@ +import * as Schema from "effect/Schema"; + +import { AuthMcpClientAccess } from "./auth.ts"; +import { EnvironmentId, TrimmedNonEmptyString } from "./baseSchemas.ts"; + +/** + * A link from this environment to another one (the peer). This environment + * signed in to the peer's `/mcp` as an outside agent, with a pairing code from + * the peer, so its agents can work there. The peer enforces the access chosen + * at linking, and revokes the link from its own Connections. + */ +export const PeerLink = Schema.Struct({ + environmentId: EnvironmentId, + label: Schema.String, + /** Addresses the peer was linked at, tried in order. */ + urls: Schema.Array(TrimmedNonEmptyString), + access: AuthMcpClientAccess, + linkedAt: Schema.DateTimeUtc, + expiresAt: Schema.DateTimeUtc, + lastReachedAt: Schema.NullOr(Schema.DateTimeUtc), + lastError: Schema.NullOr(Schema.String), +}); +export type PeerLink = typeof PeerLink.Type; + +/** Whether a link can be used now, as worked out when it is listed. */ +export const PeerLinkStatus = Schema.Literals(["reachable", "unreachable", "expired"]); +export type PeerLinkStatus = typeof PeerLinkStatus.Type; + +export const PeerLinkSummary = Schema.Struct({ + ...PeerLink.fields, + status: PeerLinkStatus, +}); +export type PeerLinkSummary = typeof PeerLinkSummary.Type; + +export const PeerLinkCreateInput = Schema.Struct({ + /** Where the peer answers: an https, Tailscale, LAN or loopback base URL. */ + url: TrimmedNonEmptyString, + /** More addresses for the same peer, tried after `url`. */ + alternateUrls: Schema.optionalKey(Schema.Array(TrimmedNonEmptyString)), + /** A pairing code from the peer (`t3 pair` there, or its Connections settings). */ + pairingCode: TrimmedNonEmptyString, + access: AuthMcpClientAccess, +}); +export type PeerLinkCreateInput = typeof PeerLinkCreateInput.Type; + +export const PeerLinkRemoveInput = Schema.Struct({ environmentId: EnvironmentId }); +export type PeerLinkRemoveInput = typeof PeerLinkRemoveInput.Type; + +export const PeerLinkRemoveResult = Schema.Struct({ removed: Schema.Boolean }); +export type PeerLinkRemoveResult = typeof PeerLinkRemoveResult.Type; + +/** Why linking or calling a peer failed, in words a user can act on. */ +export class PeerLinkError extends Schema.TaggedError()("PeerLinkError", { + reason: Schema.Literals([ + "unreachable", + "not_a_t3_environment", + "incompatible", + "self", + "pairing_rejected", + "expired", + "unknown_link", + "storage", + ]), + message: Schema.String, + cause: Schema.optionalKey(Schema.Defect()), +}) {} diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index d7a11234c446..2973a5486237 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -14,6 +14,14 @@ import { McpAppToolInfoInput, McpAppUpdateModelContextInput, } from "./mcpApps.ts"; +import { + PeerLink, + PeerLinkCreateInput, + PeerLinkError, + PeerLinkRemoveInput, + PeerLinkRemoveResult, + PeerLinkSummary, +} from "./peerLink.ts"; import { ChatGptReconnectProfileInput, ChatGptReconnectProfile, @@ -482,6 +490,9 @@ export const WS_METHODS = { serverRunStorageCleanup: "server.runStorageCleanup", serverGetStorageCleanupReport: "server.getStorageCleanupReport", serverGetDatabaseStorage: "server.getDatabaseStorage", + peerLinksList: "peerLinks.list", + peerLinksLink: "peerLinks.link", + peerLinksUnlink: "peerLinks.unlink", serverGetSettings: "server.getSettings", serverUpdateSettings: "server.updateSettings", serverDiscoverSourceControl: "server.discoverSourceControl", @@ -755,6 +766,24 @@ const WsServerGetDatabaseStorageRpc = Rpc.make(WS_METHODS.serverGetDatabaseStora error: EnvironmentAuthorizationError, }); +const WsPeerLinksListRpc = Rpc.make(WS_METHODS.peerLinksList, { + payload: Schema.Struct({}), + success: Schema.Struct({ links: Schema.Array(PeerLinkSummary) }), + error: Schema.Union([PeerLinkError, EnvironmentAuthorizationError]), +}); + +const WsPeerLinksLinkRpc = Rpc.make(WS_METHODS.peerLinksLink, { + payload: PeerLinkCreateInput, + success: PeerLink, + error: Schema.Union([PeerLinkError, EnvironmentAuthorizationError]), +}); + +const WsPeerLinksUnlinkRpc = Rpc.make(WS_METHODS.peerLinksUnlink, { + payload: PeerLinkRemoveInput, + success: PeerLinkRemoveResult, + error: Schema.Union([PeerLinkError, EnvironmentAuthorizationError]), +}); + const WsServerGetSettingsRpc = Rpc.make(WS_METHODS.serverGetSettings, { payload: Schema.Struct({}), success: ServerSettings, @@ -1909,6 +1938,9 @@ export const WsRpcGroup = RpcGroup.make( WsServerGetDatabaseStorageRpc, WsServerGetSettingsRpc, WsServerUpdateSettingsRpc, + WsPeerLinksListRpc, + WsPeerLinksLinkRpc, + WsPeerLinksUnlinkRpc, WsServerDiscoverSourceControlRpc, WsServerSearchAcpRegistryRpc, WsServerPrepareAcpRegistryAgentRpc,