From 250757c66e41703a565c2bffbf392aefe2bfd57c Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Mon, 19 Oct 2026 09:54:30 +0000 Subject: [PATCH 1/4] feat(server): persist links to peer environments Pair through existing MCP OAuth, store each credential under its own name, and expose link/list/get/unlink through the environment CLI and typed RPCs. Require the linked identity and mode-limit capability before reading a token. Decode structured MCP results first, refuse HTTP redirects, and release a new credential when persistence fails or the operation is cancelled. Preserve main's migration 064 and add PeerLinks as migration 065. Rebuilt from [#16655](https://github.com/pingdotgg/t3code/pull/16655). Co-Authored-By: Claude Opus 5.5 (1M context) --- apps/server/src/auth/RpcAuthorization.test.ts | 70 ++ apps/server/src/auth/RpcAuthorization.ts | 2 + apps/server/src/binCli.ts | 2 + apps/server/src/cli/environment.ts | 189 ++++++ .../src/environment/ServerEnvironment.ts | 2 + .../src/observability/RpcInstrumentation.ts | 4 + apps/server/src/peer/PeerLinks.test.ts | 633 ++++++++++++++++++ apps/server/src/peer/PeerLinks.ts | 492 ++++++++++++++ apps/server/src/peer/PeerMcpClient.ts | 257 +++++++ apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/055_OrchestrationV2.test.ts | 4 +- .../persistence/Migrations/065_PeerLinks.ts | 24 + .../reconcileV2PreviewMigration.test.ts | 2 + apps/server/src/server.ts | 3 + apps/server/src/ws.ts | 11 + docs/user/outside-agents.md | 20 + .../src/state/commandPermissions.test.ts | 27 + .../contracts/src/clientRpcPermissions.ts | 3 + packages/contracts/src/environment.ts | 4 + packages/contracts/src/index.ts | 1 + packages/contracts/src/peerLink.ts | 71 ++ packages/contracts/src/rpc.ts | 40 ++ 22 files changed, 1862 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/cli/environment.ts create mode 100644 apps/server/src/peer/PeerLinks.test.ts create mode 100644 apps/server/src/peer/PeerLinks.ts create mode 100644 apps/server/src/peer/PeerMcpClient.ts create mode 100644 apps/server/src/persistence/Migrations/065_PeerLinks.ts create mode 100644 packages/contracts/src/peerLink.ts diff --git a/apps/server/src/auth/RpcAuthorization.test.ts b/apps/server/src/auth/RpcAuthorization.test.ts index 8b3953feb9cad..7a79c4738ce4b 100644 --- a/apps/server/src/auth/RpcAuthorization.test.ts +++ b/apps/server/src/auth/RpcAuthorization.test.ts @@ -1,4 +1,8 @@ import { + AuthAccessReadScope, + AuthAccessWriteScope, + EnvironmentId, + PeerLinkError, AuthEnvironmentMaintainScope, AuthDiagnosticsReadScope, AuthFilesystemReadScope, @@ -470,3 +474,69 @@ it.effect("denies manual cleanup before the handler without settings permission" expect(handled).toBe(false); }).pipe(Effect.scoped), ); + +it.effect( + "peer links can be read with access:read, while linking and unlinking require access:write", + () => + Effect.scoped( + Effect.gen(function* () { + const tested = [ + WS_METHODS.peerLinksList, + WS_METHODS.peerLinksGet, + WS_METHODS.peerLinksLink, + WS_METHODS.peerLinksUnlink, + ] as const; + const group = WsRpcGroup.omit( + ...[...WsRpcGroup.requests.keys()].filter( + (tag): tag is Exclude => + !(tested as ReadonlyArray).includes(tag), + ), + ); + const handled: Array = []; + const handlers = group.toLayer({ + [WS_METHODS.peerLinksList]: () => Effect.succeed({ links: [] }), + [WS_METHODS.peerLinksGet]: () => Effect.succeed(null), + [WS_METHODS.peerLinksLink]: () => + Effect.sync(() => handled.push("link")).pipe( + Effect.andThen( + new PeerLinkError({ reason: "self", message: "That address is this environment." }), + ), + ), + [WS_METHODS.peerLinksUnlink]: () => + Effect.sync(() => { + handled.push("unlink"); + return { removed: true }; + }), + }); + const reader = yield* RpcTest.makeClient(group).pipe( + Effect.provide(Layer.merge(handlers, RpcAuthorization.layer([AuthAccessReadScope]))), + ); + const peer = { environmentId: EnvironmentId.make("peer") }; + const link = { + url: "http://127.0.0.1:1", + pairingCode: "test", + access: "read-only" as const, + }; + expect(yield* reader[WS_METHODS.peerLinksList]({})).toEqual({ links: [] }); + expect(yield* reader[WS_METHODS.peerLinksGet](peer)).toBeNull(); + expect(yield* reader[WS_METHODS.peerLinksLink](link).pipe(Effect.flip)).toMatchObject({ + requiredPermission: AuthAccessWriteScope, + }); + expect(yield* reader[WS_METHODS.peerLinksUnlink](peer).pipe(Effect.flip)).toMatchObject({ + requiredPermission: AuthAccessWriteScope, + }); + expect(handled).toEqual([]); + const writer = yield* RpcTest.makeClient(group).pipe( + Effect.provide(Layer.merge(handlers, RpcAuthorization.layer([AuthAccessWriteScope]))), + ); + expect(yield* writer[WS_METHODS.peerLinksLink](link).pipe(Effect.flip)).toMatchObject({ + _tag: "PeerLinkError", + }); + expect(yield* writer[WS_METHODS.peerLinksUnlink](peer)).toEqual({ removed: true }); + expect(handled).toEqual(["link", "unlink"]); + expect(yield* writer[WS_METHODS.peerLinksGet](peer).pipe(Effect.flip)).toMatchObject({ + requiredPermission: AuthAccessReadScope, + }); + }), + ), +); diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 30e0ca31a5945..b03406d7f84d8 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -83,6 +83,8 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.serverGetStorageCleanupReport]: AuthOrchestrationReadScope, [WS_METHODS.serverGetDatabaseStorage]: AuthOrchestrationReadScope, [WS_METHODS.serverGetSettings]: AuthOrchestrationReadScope, + [WS_METHODS.peerLinksList]: AuthAccessReadScope, + [WS_METHODS.peerLinksGet]: AuthAccessReadScope, [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 b0a2ceffef8f6..8ae72f4d7425e 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 0000000000000..38f35ee3c2089 --- /dev/null +++ b/apps/server/src/cli/environment.ts @@ -0,0 +1,189 @@ +import { AuthMcpClientAccess, EnvironmentId, PeerLink, PeerLinkSummary } from "@t3tools/contracts"; +import * as Console from "effect/Console"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +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 = ( + flags: CliAuthLocationFlags, + run: (links: PeerLinks.PeerLinks["Service"]) => Effect.Effect, + options?: { readonly quietLogs?: boolean }, +) => + Effect.gen(function* () { + 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(PeerLinkSummary))); + +const encodeLink = Schema.encodeEffect(Schema.fromJsonString(Schema.NullOr(PeerLink))); + +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 HTTPS address, or HTTP over Tailscale or loopback.", + ), + ), + 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 environmentGetCommand = Command.make("get", { + ...authLocationFlags, + json: jsonFlag, + environmentId: Argument.String("environment-id"), +}).pipe( + Command.withDescription("Read a stored link without contacting the peer."), + Command.withHandler((flags) => + runWithPeerLinks( + flags, + (links) => + Effect.gen(function* () { + const linked = yield* links.get(EnvironmentId.make(flags.environmentId)); + yield* Console.log( + flags.json + ? `${yield* encodeLink(Option.getOrNull(linked))}\n` + : Option.isNone(linked) + ? `No linked environment ${flags.environmentId}.\n` + : `${linked.value.label} (${linked.value.environmentId})\n access ${linked.value.access}, expires ${DateTime.formatIso(linked.value.expiresAt)}\n ${linked.value.urls.join(", ")}\n`, + ); + }), + { 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, + environmentGetCommand, + environmentUnlinkCommand, + ]), +); diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index cef5d9358ca94..d37fa9346a6eb 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -264,6 +264,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 161694d55cb8f..1f9af53fa036a 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/apps/server/src/observability/RpcInstrumentation.ts @@ -59,6 +59,10 @@ const RPC_AGGREGATES = { [WS_METHODS.serverGetStorageCleanupReport]: "server", [WS_METHODS.serverGetDatabaseStorage]: "server", [WS_METHODS.serverGetSettings]: "server", + [WS_METHODS.peerLinksList]: "peerLinks", + [WS_METHODS.peerLinksGet]: "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 0000000000000..124a5d4807c5d --- /dev/null +++ b/apps/server/src/peer/PeerLinks.test.ts @@ -0,0 +1,633 @@ +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 Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; +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 SqlClient from "effect/sql/SqlClient"; +import * as TestClock from "effect/testing/TestClock"; +import { McpSchema, Tool, Toolkit } from "effect/ai"; +import { + FetchHttpClient, + HttpRouter, + HttpServer, + HttpServerRequest, + HttpServerResponse, +} 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 decodeMcpRequest = Schema.decodeUnknownEffect( + Schema.Struct({ method: Schema.String, id: Schema.optional(Schema.Number) }), +); + +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 redirects = yield* Ref.make>(new Map()); + const redirectedBodies = yield* Ref.make>([]); + const toolResult = yield* Ref.make(null); + 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 redirect = (yield* Ref.get(redirects)).get(request.url); + if (redirect !== undefined) { + return HttpServerResponse.redirect(redirect, { status: 307 }); + } + if (request.url === "/redirect-target") { + const body = yield* request.text; + yield* Ref.update(redirectedBodies, (bodies) => [...bodies, body]); + return HttpServerResponse.empty({ status: 400 }); + } + const authorization = request.headers.authorization; + if (authorization !== undefined) { + yield* Ref.update(bearers, (seen) => [...seen, authorization]); + } + const response = yield* httpEffect; + // Keep real authentication and dispatch, but exercise richer peer wire results. + const result = yield* Ref.get(toolResult); + if (result !== null && request.url === "/mcp" && response.status === 200) { + const rpc = yield* decodeMcpRequest(yield* request.json); + if (rpc.method === "tools/call") { + return HttpServerResponse.jsonUnsafe({ jsonrpc: "2.0", id: rpc.id, result }); + } + } + return response; + }), + { 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, + toolResult, + redirects, + redirectedBodies, + /** 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, + afterSecretWrite: Effect.Effect = Effect.void, +) => + Effect.gen(function* () { + const secretReads = yield* Ref.make(0); + const layerSecrets = Layer.effect( + ServerSecretStore.ServerSecretStore, + Effect.gen(function* () { + const secrets = yield* ServerSecretStore.ServerSecretStore; + return ServerSecretStore.ServerSecretStore.of({ + ...secrets, + set: (name, value) => secrets.set(name, value).pipe(Effect.andThen(afterSecretWrite)), + get: (name) => + Ref.update(secretReads, (count) => count + 1).pipe(Effect.andThen(secrets.get(name))), + }); + }), + ).pipe(Layer.provide(ServerSecretStore.layer)); + const context = yield* PeerMcpClient.layer.pipe( + Layer.provideMerge(PeerLinks.layer), + Layer.provideMerge( + Layer.mergeAll( + Sqlite.layerMemory, + layerSecrets, + 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, + ); + const secrets = Context.get(context, ServerSecretStore.ServerSecretStore); + const fileSystem = yield* FileSystem.FileSystem; + return { + secretReads, + sql: Context.get(context, SqlClient.SqlClient), + secretNames: fileSystem + .readDirectory(secrets.directory!) + .pipe(Effect.map((names) => names.sort())), + 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, reads and forgets a peer while preserving access limits and refusals", () => + 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"]]); + + yield* Ref.set(b.probes, 0); + expect(yield* a.links.get(box.environmentId)).toEqual(Option.some(linked)); + expect(Option.isNone(yield* a.links.get(laptop.environmentId))).toBe(true); + expect(yield* Ref.get(b.probes)).toBe(0); + + // 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." }); + + expect(yield* a.links.unlink(box.environmentId)).toBe(true); + expect(yield* a.links.unlink(box.environmentId)).toBe(false); + expect(yield* a.links.get(box.environmentId)).toEqual(Option.none()); + expect(yield* a.links.list).toEqual([]); + expect(yield* a.secretNames).toEqual([]); + yield* Ref.set(b.bearers, []); + expect((yield* probe(a).pipe(Effect.flip)).code).toBe("invalid_request"); + expect(yield* Ref.get(b.bearers)).toHaveLength(0); + expect(yield* linkedSessions(b)).toHaveLength(1); + }), + ).pipe(Effect.provide(Layer.merge(NodeHttpServer.layerTest, NodeServices.layer))), +); + +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 firstStored = yield* Deferred.make(); + const finishFirst = yield* Deferred.make(); + const writes = yield* Ref.make(0); + const a = yield* makeA( + laptop, + Effect.gen(function* () { + const write = yield* Ref.updateAndGet(writes, (count) => count + 1); + if (write === 1) { + yield* Deferred.succeed(firstStored, undefined); + yield* Deferred.await(finishFirst); + } + }), + ); + const first = yield* b.auth.issuePairingCredential(); + const second = yield* b.auth.issuePairingCredential(); + const pending = yield* a.links + .link({ + url: b.url, + pairingCode: first.credential, + access: "read-only", + }) + .pipe(Effect.forkChild); + // Let the full-access link commit between the other token write and row write. + yield* Deferred.await(firstStored); + yield* a.links.link({ url: b.url, pairingCode: second.credential, access: "full-access" }); + yield* Deferred.succeed(finishFirst, undefined); + yield* Fiber.join(pending); + const stored = yield* a.links.get(box.environmentId); + expect(Option.map(stored, (link) => link.access)).toEqual(Option.some("read-only")); + expect((yield* probe(a)).modes).toBe("approval-required/default"); + expect(yield* a.secretNames).toHaveLength(1); + expect(yield* a.links.unlink(box.environmentId)).toBe(true); + expect(yield* a.secretNames).toEqual([]); + }), + ).pipe(Effect.provide(Layer.merge(NodeHttpServer.layerTest, NodeServices.layer))), +); + +it.effect("never sends a bearer after the peer identity or mode capability changes", () => + 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); + + const readsBefore = yield* Ref.get(a.secretReads); + // 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)).toHaveLength(0); + expect((yield* a.links.list).map((link) => link.status)).toEqual(["unreachable"]); + + yield* Ref.set( + b.descriptor, + descriptorOf("environment-box", "Box", { mcpModeLimitHeader: false }), + ); + const probesBefore = yield* Ref.get(b.probes); + const downgraded = yield* probe(a, { + runtimeMode: "approval-required", + interactionMode: "plan", + }).pipe(Effect.exit); + expect(yield* Ref.get(b.bearers)).toHaveLength(0); + expect(Exit.isFailure(downgraded)).toBe(true); + expect(yield* Ref.get(a.secretReads)).toBe(readsBefore); + expect(yield* Ref.get(b.probes)).toBe(probesBefore + 1); + expect((yield* a.links.list).map((link) => link.status)).toEqual(["incompatible"]); + expect((yield* a.links.get(box.environmentId)).pipe(Option.getOrThrow).lastError).toContain( + "limits", + ); + }), + ).pipe(Effect.provide(Layer.merge(NodeHttpServer.layerTest, NodeServices.layer))), +); + +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)).toHaveLength(0); + }), + ).pipe(Effect.provide(Layer.merge(NodeHttpServer.layerTest, NodeServices.layer))), +); + +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(Layer.merge(NodeHttpServer.layerTest, NodeServices.layer))), +); + +it.effect( + "decodes structured MCP results with display text and rejects invalid result shapes", + () => + 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" }); + const result = { caller: "T3 Code · Laptop", modes: "auto/default" }; + yield* Ref.set( + b.toolResult, + new McpSchema.CallToolResult({ + content: [{ type: "text", text: JSON.stringify(result) }], + }), + ); + expect(yield* probe(a)).toEqual(result); + yield* Ref.set( + b.toolResult, + new McpSchema.CallToolResult({ + structuredContent: result, + content: [ + { type: "text", text: JSON.stringify({ url: "https://example.com" }) }, + { type: "text", text: "Snapshot text for the user." }, + ], + }), + ); + expect(yield* probe(a)).toEqual(result); + + yield* Ref.set( + b.toolResult, + new McpSchema.CallToolResult({ + structuredContent: { caller: 123, modes: "auto/default" }, + content: [{ type: "text", text: JSON.stringify(result) }], + }), + ); + expect((yield* probe(a).pipe(Effect.flip)).code).toBe("orchestration_error"); + yield* Ref.set( + b.toolResult, + new McpSchema.CallToolResult({ + isError: true, + structuredContent: { code: "thread_not_found", message: "No such thread." }, + content: [{ type: "text", text: "Display-only refusal text." }], + }), + ); + expect((yield* probe(a).pipe(Effect.flip)).code).toBe("thread_not_found"); + }), + ).pipe(Effect.provide(Layer.merge(NodeHttpServer.layerTest, NodeServices.layer))), +); + +it.effect("failed and cancelled relinks preserve the previous link and credential", () => + Effect.scoped( + Effect.gen(function* () { + const b = yield* serveB(box); + const written = yield* Deferred.make(); + let pause = false; + const a = yield* makeA( + laptop, + Effect.suspend(() => + pause + ? Deferred.succeed(written, undefined).pipe(Effect.andThen(Effect.never)) + : Effect.void, + ), + ); + const pairing = yield* b.auth.issuePairingCredential(); + const linked = yield* a.links.link({ + url: b.url, + pairingCode: pairing.credential, + access: "read-only", + }); + const secretsBefore = yield* a.secretNames; + pause = true; + const cancelledPairing = yield* b.auth.issuePairingCredential(); + const pending = yield* a.links + .link({ + url: b.url, + pairingCode: cancelledPairing.credential, + access: "full-access", + }) + .pipe(Effect.forkChild); + yield* Deferred.await(written); + yield* Fiber.interrupt(pending); + expect(yield* a.links.get(box.environmentId)).toEqual(Option.some(linked)); + expect(yield* a.secretNames).toEqual(secretsBefore); + pause = false; + yield* a.sql`CREATE TRIGGER reject_peer_link BEFORE INSERT ON peer_environment_links + BEGIN SELECT RAISE(FAIL, 'test persistence failure'); END`; + const replacement = yield* b.auth.issuePairingCredential(); + const failed = yield* a.links + .link({ url: b.url, pairingCode: replacement.credential, access: "full-access" }) + .pipe(Effect.flip); + expect(failed.reason).toBe("storage"); + expect(yield* a.links.get(box.environmentId)).toEqual(Option.some(linked)); + expect((yield* probe(a)).modes).toBe("approval-required/default"); + expect(yield* a.secretNames).toEqual(secretsBefore); + }), + ).pipe(Effect.provide(Layer.merge(NodeHttpServer.layerTest, NodeServices.layer))), +); + +it.effect("does not follow descriptor, OAuth or authenticated MCP redirects", () => + Effect.scoped( + Effect.gen(function* () { + const b = yield* serveB(box); + const a = yield* makeA(laptop); + const pairing = yield* b.auth.issuePairingCredential(); + const target = b.url.replace("127.0.0.1", "localhost") + "/redirect-target"; + yield* Ref.set(b.redirects, new Map([["/oauth/mcp/decision", target]])); + const linked = yield* a.links + .link({ + url: b.url, + pairingCode: pairing.credential, + access: "auto", + }) + .pipe(Effect.exit); + expect(Exit.isFailure(linked)).toBe(true); + expect( + (yield* Ref.get(b.redirectedBodies)).some((body) => body.includes(pairing.credential)), + ).toBe(false); + expect(yield* a.links.list).toEqual([]); + expect(yield* Ref.get(b.redirectedBodies)).toEqual([]); + yield* Ref.set(b.redirects, new Map([["/.well-known/t3/environment", target]])); + expect( + (yield* a.links + .link({ + url: b.url, + pairingCode: pairing.credential, + access: "auto", + }) + .pipe(Effect.flip)).reason, + ).toBe("not_a_t3_environment"); + expect(yield* Ref.get(b.redirectedBodies)).toEqual([]); + yield* Ref.set(b.redirects, new Map()); + yield* a.links.link({ url: b.url, pairingCode: pairing.credential, access: "auto" }); + yield* Ref.set(b.redirects, new Map([["/mcp", target]])); + expect((yield* probe(a).pipe(Effect.flip)).code).toBe("orchestration_error"); + expect(yield* Ref.get(b.redirectedBodies)).toEqual([]); + }), + ).pipe(Effect.provide(Layer.merge(NodeHttpServer.layerTest, NodeServices.layer))), +); diff --git a/apps/server/src/peer/PeerLinks.ts b/apps/server/src/peer/PeerLinks.ts new file mode 100644 index 0000000000000..2c5019b2be0c7 --- /dev/null +++ b/apps/server/src/peer/PeerLinks.ts @@ -0,0 +1,492 @@ +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 FetchHttpClient from "effect/http/FetchHttpClient"; +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); + +interface ResolvedPeerLink { + readonly link: PeerLink; + readonly url: string; + readonly token: string; +} + +/** Peer sessions use the existing MCP OAuth flow; tokens stay in the secret store. */ +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; + readonly resolve: ( + environmentId: EnvironmentId, + ) => Effect.Effect; + 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); + +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, + }), + ), + ), + ); + +// Pairing codes and bearer tokens require HTTPS outside loopback and Tailscale. +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).pipe( + HttpClient.transformResponse( + Effect.provideService(FetchHttpClient.RequestInit, { redirect: "error" }), + ), + ); + const environment = yield* ServerEnvironment.ServerEnvironment; + + 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), + ); + + const describe = (origin: string) => + clientFor(origin).pipe( + Effect.flatMap((client) => client.metadata.descriptor()), + Effect.timeout(probeTimeout(origin)), + Effect.option, + ); + + const toLink = (row: PeerLinkRow) => + Effect.gen(function* () { + 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; + }).pipe(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])), + ); + + const signIn = (origin: string, input: PeerLinkCreateInput, clientName: string) => + Effect.gen(function* () { + 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}`, + }), + ), + ), + ); + }).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"] = (input) => + Effect.scoped( + Effect.gen(function* () { + 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, + }; + // A failed or cancelled save must release only its own credential. + const secretName = yield* newSecretName; + let committed = false; + yield* Effect.addFinalizer(() => + committed + ? Effect.void + : secrets.remove(secretName).pipe(Effect.ignoreCause({ log: true })), + ); + 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* Effect.uninterruptible( + Effect.gen(function* () { + 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)); + // Commit and ownership transfer cannot be interrupted between them. + committed = true; + if (Option.isSome(previous) && previous.value.secret_name !== secretName) { + yield* secrets.remove(previous.value.secret_name).pipe(Effect.ignore); + } + }), + ); + return linked; + }), + ); + + const findPeer = (link: PeerLink) => + Effect.gen(function* () { + for (const url of link.urls) { + const found = yield* describe(url); + if (Option.isSome(found) && found.value.environmentId === link.environmentId) { + return Option.some({ + url, + compatible: found.value.capabilities.mcpModeLimitHeader === true, + }); + } + } + return Option.none(); + }); + + const statusOf = (link: PeerLink, now: DateTime.Utc) => + DateTime.isLessThanOrEqualTo(link.expiresAt, now) + ? Effect.succeed("expired") + : findPeer(link).pipe( + Effect.map((peer): PeerLinkStatus => + Option.isNone(peer) + ? "unreachable" + : peer.value.compatible + ? "reachable" + : "incompatible", + ), + ); + + 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"] = (environmentId) => + Effect.gen(function* () { + 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"] = (environmentId) => + Effect.gen(function* () { + const row = yield* readRow(environmentId); + if (Option.isNone(row)) { + return yield* new PeerLinkError({ + reason: "unknown_link", + message: `This environment is not linked to ${environmentId}. Run t3 environment list to see linked environments.`, + }); + } + 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 peer = yield* findPeer(link); + if (Option.isNone(peer)) { + return yield* new PeerLinkError({ + reason: "unreachable", + message: `${link.label} did not answer at ${link.urls.join(", ")}.`, + }); + } + if (!peer.value.compatible) { + return yield* new PeerLinkError({ + reason: "incompatible", + message: `${link.label} can no longer keep a linked agent's limits. Update it before using this link.`, + }); + } + 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.`, + }); + } + return { link, url: peer.value.url, 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 0000000000000..511836d623d00 --- /dev/null +++ b/apps/server/src/peer/PeerMcpClient.ts @@ -0,0 +1,257 @@ +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 FetchHttpClient from "effect/http/FetchHttpClient"; +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"; + +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, + { + 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 Failure = Schema.Struct({ code: OrchestratorMcpFailure.fields.code, message: Schema.String }); +const decodeFailure = Schema.decodeUnknownOption(Failure); +const decodeTextFailure = Schema.decodeUnknownOption(Schema.fromJsonString(Failure)); +const decodeTextResult = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Unknown)); + +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).pipe( + HttpClient.transformResponse( + Effect.provideService(FetchHttpClient.RequestInit, { redirect: "error" }), + ), + ); + // 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 = (url: string, token: string, limitHeader: Record) => + Effect.gen(function* () { + 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"] = (input) => + Effect.gen(function* () { + 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 result = body.result; + const text = result.content.map((part) => part.text ?? "").join(""); + if (result.isError === true) { + const refused = + result.structuredContent === undefined + ? decodeTextFailure(text) + : decodeFailure(result.structuredContent); + 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 decodeResult = Effect.gen(function* () { + const value = + result.structuredContent === undefined + ? yield* decodeTextResult(text) + : result.structuredContent; + // oxlint-disable-next-line t3code/no-inline-schema-compile -- Each caller supplies its own success schema. + return yield* Schema.decodeUnknownEffect(input.success)(value); + }); + return yield* decodeResult.pipe( + Effect.mapError( + () => + new OrchestratorMcpFailure({ + code: "orchestration_error", + message: + "The linked environment's answer did not match this version. Update both environments.", + }), + ), + ); + }).pipe( + Effect.tapError((error) => + links.recordOutcome(input.environmentId, { error: error.message }), + ), + Effect.tap(() => links.recordOutcome(input.environmentId, { error: null })), + 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 2b4981e04d1b9..88d1cc97bb45c 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -67,6 +67,7 @@ import Migration0061 from "./Migrations/061_PullRequestReviewGuides.ts"; import Migration0062 from "./Migrations/062_ThreadFailureLookupIndex.ts"; import Migration0063 from "./Migrations/063_DropUnusedV1Projections.ts"; import Migration0064 from "./Migrations/064_CanonicalizeThreadPullRequestLinks.ts"; +import Migration0065 from "./Migrations/065_PeerLinks.ts"; export const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -134,6 +135,7 @@ export const migrationEntries = [ [62, "ThreadFailureLookupIndex", Migration0062], [63, "DropUnusedV1Projections", Migration0063], [64, "CanonicalizeThreadPullRequestLinks", Migration0064], + [65, "PeerLinks", Migration0065], ] 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 447c365082f27..c5450fefb9229 100644 --- a/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts +++ b/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts @@ -13,7 +13,7 @@ layer("055_OrchestrationV2", (it) => { Effect.sync(() => { assert.deepStrictEqual( migrationEntries.map(([id]) => id), - Array.from({ length: 64 }, (_, index) => index + 1), + Array.from({ length: 65 }, (_, index) => index + 1), ); }), ); @@ -36,6 +36,7 @@ layer("055_OrchestrationV2", (it) => { [62, "ThreadFailureLookupIndex"], [63, "DropUnusedV1Projections"], [64, "CanonicalizeThreadPullRequestLinks"], + [65, "PeerLinks"], ]); assert.deepStrictEqual(yield* runMigrations(), []); @@ -66,6 +67,7 @@ layer("055_OrchestrationV2", (it) => { { migration_id: 62, name: "ThreadFailureLookupIndex" }, { migration_id: 63, name: "DropUnusedV1Projections" }, { migration_id: 64, name: "CanonicalizeThreadPullRequestLinks" }, + { migration_id: 65, name: "PeerLinks" }, ]); const tables = yield* sql<{ readonly name: string }>` diff --git a/apps/server/src/persistence/Migrations/065_PeerLinks.ts b/apps/server/src/persistence/Migrations/065_PeerLinks.ts new file mode 100644 index 0000000000000..5551ccdd07a25 --- /dev/null +++ b/apps/server/src/persistence/Migrations/065_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 802c5110490ac..d03d9d2a97fc5 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -44,6 +44,7 @@ describe("V2 preview upgrade", () => { [62, "ThreadFailureLookupIndex"], [63, "DropUnusedV1Projections"], [64, "CanonicalizeThreadPullRequestLinks"], + [65, "PeerLinks"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -131,6 +132,7 @@ describe("V2 preview upgrade", () => { [62, "ThreadFailureLookupIndex"], [63, "DropUnusedV1Projections"], [64, "CanonicalizeThreadPullRequestLinks"], + [65, "PeerLinks"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 59daedc151071..e3ea81af29d64 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -64,6 +64,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"; @@ -706,6 +708,7 @@ const layerMakeRoutes = Layer.mergeAll( Layer.provide(layerPullRequestService), Layer.provide(ServerBrowser.layer.pipe(Layer.provide(DesktopBrowserChannel.layer))), Layer.provide(PreviewBrowser.layer), + 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.ts b/apps/server/src/ws.ts index ade826e53f38b..ce1a4d5b62ff3 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"; @@ -1260,6 +1261,7 @@ 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; @@ -2350,6 +2352,13 @@ 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.peerLinksGet]: ({ environmentId }) => + peerLinks.get(environmentId).pipe(Effect.map(Option.getOrNull)), + [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]: ({ patch, providerInstanceMutation }) => @@ -3091,6 +3100,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", @@ -3181,6 +3191,7 @@ export const layer = Layer.unwrap( reviewGuides, ), ), + Layer.provide(Layer.succeed(PeerLinks.PeerLinks, peerLinks)), Layer.provide( SourceControlDiscovery.layer.pipe( Layer.provide( diff --git a/docs/user/outside-agents.md b/docs/user/outside-agents.md index 7cc6692e38f6c..8da8106637149 100644 --- a/docs/user/outside-agents.md +++ b/docs/user/outside-agents.md @@ -88,6 +88,26 @@ Any agent that supports remote MCP servers over HTTP with OAuth can connect. Giv it the MCP URL and choose OAuth, and the agent finds the sign-in settings itself. Agents without OAuth support cannot connect. +## Link from another T3 Code environment + +On the host that will make requests, link the other environment with a pairing +code from its **Settings → Connections** or `t3 auth pairing create`: + +```sh +t3 environment link https:// --pairing-code --access approval-required +``` + +Use HTTPS, or HTTP over loopback or Tailscale. Plain HTTP LAN addresses are +not supported. The link belongs to the host where you ran the command and +appears on the peer as **T3 Code · **. Calls keep both the link's +access ceiling and the calling agent's narrower modes. + +Use `t3 environment list` to check reachability and +`t3 environment get --json` to read a stored link without +contacting the peer. Run `t3 environment unlink ` to forget it +locally. To revoke its access on the peer, remove its session in that peer's +Connections settings. After a link expires, link again with a new pairing code. + ## Remove an agent Approved agents appear under **Settings → Connections** like other clients. diff --git a/packages/client-runtime/src/state/commandPermissions.test.ts b/packages/client-runtime/src/state/commandPermissions.test.ts index 58a77730a4011..dd28f6d51e9a1 100644 --- a/packages/client-runtime/src/state/commandPermissions.test.ts +++ b/packages/client-runtime/src/state/commandPermissions.test.ts @@ -7,6 +7,7 @@ import type { RpcSession } from "../rpc/session.ts"; import { describe, expect, it } from "@effect/vitest"; import { vi } from "vite-plus/test"; import { + AuthAccessWriteScope, AuthOrchestrationOperateScope, AuthSettingsWriteScope, AuthSourceControlWriteScope, @@ -59,6 +60,32 @@ const setup = Effect.gen(function* () { }); describe("command permissions", () => { + it.effect("requires the destination access grant to link or unlink environments", () => + Effect.scoped( + Effect.gen(function* () { + const registry = yield* setup; + registry.set(sessions(env), AsyncResult.success(grant(true))); + registry.set( + sessions(other), + AsyncResult.success({ + ...grant(false), + scopes: [AuthAccessWriteScope], + permissions: [AuthAccessWriteScope], + }), + ); + for (const method of [WS_METHODS.peerLinksLink, WS_METHODS.peerLinksUnlink]) { + const access = createCommandPermissions(runtime, method); + expect(registry.get(access.permissionAtom(env))).toBe(false); + expect(yield* access.authorize(registry, env).pipe(Effect.flip)).toMatchObject({ + requiredPermission: AuthAccessWriteScope, + }); + expect(registry.get(access.permissionAtom(other))).toBe(true); + yield* access.authorize(registry, other); + } + }), + ), + ); + it.effect("requires the destination settings grant to run storage cleanup", () => Effect.scoped( Effect.gen(function* () { diff --git a/packages/contracts/src/clientRpcPermissions.ts b/packages/contracts/src/clientRpcPermissions.ts index e111eb244c5b1..0aa697eae3dd7 100644 --- a/packages/contracts/src/clientRpcPermissions.ts +++ b/packages/contracts/src/clientRpcPermissions.ts @@ -1,6 +1,7 @@ import * as Schema from "effect/Schema"; import { GitPreparePullRequestThreadInput } from "./git.ts"; import { + AuthAccessWriteScope, AuthOrchestrationOperateScope, AuthSettingsWriteScope, AuthSourceControlWriteScope, @@ -11,6 +12,8 @@ import { WS_METHODS } from "./rpc.ts"; /** Incremental client enforcement; the server still authorizes every request. */ export const CLIENT_GUARDED_RPC_SCOPES = { + [WS_METHODS.peerLinksLink]: AuthAccessWriteScope, + [WS_METHODS.peerLinksUnlink]: AuthAccessWriteScope, [WS_METHODS.serverRunStorageCleanup]: AuthSettingsWriteScope, [WS_METHODS.serverCheckOtlpEndpoint]: AuthSettingsWriteScope, [WS_METHODS.pullRequestsRunAction]: AuthSourceControlWriteScope, diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index b78cf0f999718..8d69f31dffe8a 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -149,6 +149,10 @@ 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), + /** Supports persistent outgoing peer links. */ + peerLinks: Schema.optionalKey(Schema.Boolean), + /** Honors `T3-Mode-Limit`; peers require this before sending authenticated calls. */ + 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 600d503ade39b..eaf9589ea92e1 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 0000000000000..3a2f5a0f808ff --- /dev/null +++ b/packages/contracts/src/peerLink.ts @@ -0,0 +1,71 @@ +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", + "incompatible", + "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 base URL, or HTTP over Tailscale or loopback. */ + 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 b74175d23c6f6..07cb8d11df2d9 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, @@ -474,6 +482,10 @@ export const WS_METHODS = { serverRunStorageCleanup: "server.runStorageCleanup", serverGetStorageCleanupReport: "server.getStorageCleanupReport", serverGetDatabaseStorage: "server.getDatabaseStorage", + peerLinksList: "peerLinks.list", + peerLinksGet: "peerLinks.get", + peerLinksLink: "peerLinks.link", + peerLinksUnlink: "peerLinks.unlink", serverGetSettings: "server.getSettings", serverUpdateSettings: "server.updateSettings", serverDiscoverSourceControl: "server.discoverSourceControl", @@ -746,6 +758,30 @@ 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 WsPeerLinksGetRpc = Rpc.make(WS_METHODS.peerLinksGet, { + payload: Schema.Struct({ environmentId: PeerLink.fields.environmentId }), + success: Schema.NullOr(PeerLink), + 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, @@ -1882,6 +1918,10 @@ export const WsRpcGroup = RpcGroup.make( WsServerGetDatabaseStorageRpc, WsServerGetSettingsRpc, WsServerUpdateSettingsRpc, + WsPeerLinksListRpc, + WsPeerLinksGetRpc, + WsPeerLinksLinkRpc, + WsPeerLinksUnlinkRpc, WsServerDiscoverSourceControlRpc, WsServerSearchAcpRegistryRpc, WsServerPrepareAcpRegistryAgentRpc, From 3466f499f69983a8f2880f108919c8498f67ea9d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 10:10:57 +0000 Subject: [PATCH 2/4] fix(server): prefer compatible peer link addresses --- apps/server/src/cli/environment.ts | 10 +++++---- apps/server/src/peer/PeerLinks.test.ts | 29 ++++++++++++++++++++++++-- apps/server/src/peer/PeerLinks.ts | 13 +++++++----- docs/user/outside-agents.md | 20 ------------------ 4 files changed, 41 insertions(+), 31 deletions(-) diff --git a/apps/server/src/cli/environment.ts b/apps/server/src/cli/environment.ts index 38f35ee3c2089..5ecb2ffd15305 100644 --- a/apps/server/src/cli/environment.ts +++ b/apps/server/src/cli/environment.ts @@ -86,17 +86,19 @@ const environmentLinkCommand = Command.make("link", { Flag.atLeast(0), ), pairingCode: Flag.String("pairing-code").pipe( - Flag.withDescription("A pairing code from the other environment (`t3 pair` there)."), + Flag.withDescription( + "A pairing code from the other environment (`t3 auth pairing create` 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.", + "Access ceiling for this outgoing link: 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.", + "Create an outgoing link to another T3 Code environment. Revoke its session in that environment's Settings → Connections.", ), Command.withHandler((flags) => runWithPeerLinks(flags, (links) => @@ -179,7 +181,7 @@ const environmentUnlinkCommand = Command.make("unlink", { ); export const environmentCommand = Command.make("environment").pipe( - Command.withDescription("Link other T3 Code environments, so agents here can work there."), + Command.withDescription("Manage outgoing links to other T3 Code environments."), Command.withSubcommands([ environmentLinkCommand, environmentListCommand, diff --git a/apps/server/src/peer/PeerLinks.test.ts b/apps/server/src/peer/PeerLinks.test.ts index 124a5d4807c5d..3700b3933eaaa 100644 --- a/apps/server/src/peer/PeerLinks.test.ts +++ b/apps/server/src/peer/PeerLinks.test.ts @@ -93,6 +93,7 @@ const ProbeToolkit = Toolkit.make( const serveB = (initial: ExecutionEnvironmentDescriptor) => Effect.gen(function* () { const descriptor = yield* Ref.make(initial); + const alternateDescriptor = yield* Ref.make(initial); const bearers = yield* Ref.make>([]); const probes = yield* Ref.make(0); const redirects = yield* Ref.make>(new Map()); @@ -118,6 +119,9 @@ const serveB = (initial: ExecutionEnvironmentDescriptor) => const request = yield* HttpServerRequest.HttpServerRequest; if (request.url.startsWith("/.well-known/t3/environment")) { yield* Ref.update(probes, (count) => count + 1); + if (request.headers.host?.startsWith("localhost:")) { + return HttpServerResponse.jsonUnsafe(yield* Ref.get(alternateDescriptor)); + } } const redirect = (yield* Ref.get(redirects)).get(request.url); if (redirect !== undefined) { @@ -197,6 +201,7 @@ const serveB = (initial: ExecutionEnvironmentDescriptor) => url: `http://127.0.0.1:${address.port}`, auth: Context.get(authContext, EnvironmentAuth.EnvironmentAuth), descriptor, + alternateDescriptor, bearers, toolResult, redirects, @@ -387,13 +392,20 @@ it.effect("never sends a bearer after the peer identity or mode capability chang 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" }); + const alternateUrl = b.url.replace("127.0.0.1", "localhost"); + yield* a.links.link({ + url: b.url, + alternateUrls: [alternateUrl], + pairingCode: pairing.credential, + access: "auto", + }); yield* probe(a); expect((yield* Ref.get(b.bearers)).length).toBeGreaterThan(0); const readsBefore = yield* Ref.get(a.secretReads); // 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.alternateDescriptor, 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"); @@ -404,6 +416,7 @@ it.effect("never sends a bearer after the peer identity or mode capability chang b.descriptor, descriptorOf("environment-box", "Box", { mcpModeLimitHeader: false }), ); + yield* Ref.set(b.alternateDescriptor, yield* Ref.get(b.descriptor)); const probesBefore = yield* Ref.get(b.probes); const downgraded = yield* probe(a, { runtimeMode: "approval-required", @@ -412,11 +425,23 @@ it.effect("never sends a bearer after the peer identity or mode capability chang expect(yield* Ref.get(b.bearers)).toHaveLength(0); expect(Exit.isFailure(downgraded)).toBe(true); expect(yield* Ref.get(a.secretReads)).toBe(readsBefore); - expect(yield* Ref.get(b.probes)).toBe(probesBefore + 1); + expect(yield* Ref.get(b.probes)).toBe(probesBefore + 2); expect((yield* a.links.list).map((link) => link.status)).toEqual(["incompatible"]); expect((yield* a.links.get(box.environmentId)).pipe(Option.getOrThrow).lastError).toContain( "limits", ); + + yield* Ref.set(b.alternateDescriptor, box); + const beforeFallback = yield* Ref.get(b.probes); + expect((yield* a.links.resolve(box.environmentId)).url).toBe(alternateUrl); + expect(yield* Ref.get(b.probes)).toBe(beforeFallback + 2); + expect(yield* Ref.get(a.secretReads)).toBe(readsBefore + 1); + expect( + yield* probe(a, { runtimeMode: "approval-required", interactionMode: "plan" }), + ).toEqual({ caller: "T3 Code · Laptop", modes: "approval-required/plan" }); + expect(yield* Ref.get(b.probes)).toBe(beforeFallback + 4); + expect((yield* a.links.list).map((link) => link.status)).toEqual(["reachable"]); + expect(yield* Ref.get(b.probes)).toBe(beforeFallback + 6); }), ).pipe(Effect.provide(Layer.merge(NodeHttpServer.layerTest, NodeServices.layer))), ); diff --git a/apps/server/src/peer/PeerLinks.ts b/apps/server/src/peer/PeerLinks.ts index 2c5019b2be0c7..5d79d65367566 100644 --- a/apps/server/src/peer/PeerLinks.ts +++ b/apps/server/src/peer/PeerLinks.ts @@ -372,16 +372,19 @@ const make = Effect.gen(function* () { const findPeer = (link: PeerLink) => Effect.gen(function* () { + let incompatibleUrl: string | undefined; for (const url of link.urls) { const found = yield* describe(url); if (Option.isSome(found) && found.value.environmentId === link.environmentId) { - return Option.some({ - url, - compatible: found.value.capabilities.mcpModeLimitHeader === true, - }); + if (found.value.capabilities.mcpModeLimitHeader === true) { + return Option.some({ url, compatible: true }); + } + incompatibleUrl ??= url; } } - return Option.none(); + return incompatibleUrl === undefined + ? Option.none() + : Option.some({ url: incompatibleUrl, compatible: false }); }); const statusOf = (link: PeerLink, now: DateTime.Utc) => diff --git a/docs/user/outside-agents.md b/docs/user/outside-agents.md index 8da8106637149..7cc6692e38f6c 100644 --- a/docs/user/outside-agents.md +++ b/docs/user/outside-agents.md @@ -88,26 +88,6 @@ Any agent that supports remote MCP servers over HTTP with OAuth can connect. Giv it the MCP URL and choose OAuth, and the agent finds the sign-in settings itself. Agents without OAuth support cannot connect. -## Link from another T3 Code environment - -On the host that will make requests, link the other environment with a pairing -code from its **Settings → Connections** or `t3 auth pairing create`: - -```sh -t3 environment link https:// --pairing-code --access approval-required -``` - -Use HTTPS, or HTTP over loopback or Tailscale. Plain HTTP LAN addresses are -not supported. The link belongs to the host where you ran the command and -appears on the peer as **T3 Code · **. Calls keep both the link's -access ceiling and the calling agent's narrower modes. - -Use `t3 environment list` to check reachability and -`t3 environment get --json` to read a stored link without -contacting the peer. Run `t3 environment unlink ` to forget it -locally. To revoke its access on the peer, remove its session in that peer's -Connections settings. After a link expires, link again with a new pairing code. - ## Remove an agent Approved agents appear under **Settings → Connections** like other clients. From e06c97fff9849f13fcc86267ed2773615d7ca4aa Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 10:27:15 +0000 Subject: [PATCH 3/4] docs(server): clarify supported peer link addresses --- apps/server/src/cli/environment.ts | 2 +- apps/server/src/peer/PeerLinks.ts | 4 ++-- packages/contracts/src/peerLink.ts | 10 ++++------ 3 files changed, 7 insertions(+), 9 deletions(-) diff --git a/apps/server/src/cli/environment.ts b/apps/server/src/cli/environment.ts index 5ecb2ffd15305..469d52b99b315 100644 --- a/apps/server/src/cli/environment.ts +++ b/apps/server/src/cli/environment.ts @@ -78,7 +78,7 @@ const environmentLinkCommand = Command.make("link", { ...authLocationFlags, url: Argument.String("url").pipe( Argument.withDescription( - "Where the other environment answers: its HTTPS address, or HTTP over Tailscale or loopback.", + "Where the other environment answers: HTTPS, or HTTP with loopback, a Tailscale .ts.net name, or Tailscale IPv6.", ), ), alternateUrl: Flag.String("url").pipe( diff --git a/apps/server/src/peer/PeerLinks.ts b/apps/server/src/peer/PeerLinks.ts index 5d79d65367566..4a8a9208ffeb1 100644 --- a/apps/server/src/peer/PeerLinks.ts +++ b/apps/server/src/peer/PeerLinks.ts @@ -96,7 +96,7 @@ const failStorage = (cause: unknown) => ), ); -// Pairing codes and bearer tokens require HTTPS outside loopback and Tailscale. +// Only loopback and Tailscale-specific hosts can use HTTP for credentials. const peerOrigin = (url: string) => { const parsed = URL.parse(url); if (parsed === null || (parsed.protocol !== "https:" && parsed.protocol !== "http:")) { @@ -115,7 +115,7 @@ const peerOrigin = (url: string) => { 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.`, + message: `${url} is plain http, which would send the link's credentials unencrypted. Use HTTPS, or HTTP with loopback, a Tailscale .ts.net name, or Tailscale IPv6.`, }), ); } diff --git a/packages/contracts/src/peerLink.ts b/packages/contracts/src/peerLink.ts index 3a2f5a0f808ff..5bb3a21727545 100644 --- a/packages/contracts/src/peerLink.ts +++ b/packages/contracts/src/peerLink.ts @@ -4,10 +4,8 @@ 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. + * An outgoing link authenticated through the peer's MCP pairing flow. + * The peer enforces its access ceiling and can revoke its session in Connections. */ export const PeerLink = Schema.Struct({ environmentId: EnvironmentId, @@ -38,11 +36,11 @@ export const PeerLinkSummary = Schema.Struct({ export type PeerLinkSummary = typeof PeerLinkSummary.Type; export const PeerLinkCreateInput = Schema.Struct({ - /** Where the peer answers: an HTTPS base URL, or HTTP over Tailscale or loopback. */ + /** HTTPS, or HTTP with loopback, a Tailscale .ts.net name, or Tailscale IPv6. */ 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). */ + /** A pairing code from the peer (`t3 auth pairing create`, or its Connections settings). */ pairingCode: TrimmedNonEmptyString, access: AuthMcpClientAccess, }); From 84a4f73b9ec721840534d00b15084f7c997a7441 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 11:44:00 +0000 Subject: [PATCH 4/4] fix(server): handle peer response and cleanup failures --- apps/server/src/peer/PeerLinks.test.ts | 124 +++++++++++++++++- apps/server/src/peer/PeerLinks.ts | 32 +++-- apps/server/src/peer/PeerMcpClient.ts | 35 +++-- ...CanonicalizeThreadPullRequestLinks.test.ts | 6 +- 4 files changed, 170 insertions(+), 27 deletions(-) diff --git a/apps/server/src/peer/PeerLinks.test.ts b/apps/server/src/peer/PeerLinks.test.ts index 3700b3933eaaa..d910d8677584a 100644 --- a/apps/server/src/peer/PeerLinks.test.ts +++ b/apps/server/src/peer/PeerLinks.test.ts @@ -99,6 +99,10 @@ const serveB = (initial: ExecutionEnvironmentDescriptor) => const redirects = yield* Ref.make>(new Map()); const redirectedBodies = yield* Ref.make>([]); const toolResult = yield* Ref.make(null); + const sse = yield* Ref.make(false); + const replies = yield* Ref.make>( + new Map(), + ); const layerEnvironment = Layer.succeed(ServerEnvironment.ServerEnvironment, { getEnvironmentId: Ref.get(descriptor).pipe(Effect.map((current) => current.environmentId)), getDescriptor: Ref.get(descriptor), @@ -136,12 +140,30 @@ const serveB = (initial: ExecutionEnvironmentDescriptor) => if (authorization !== undefined) { yield* Ref.update(bearers, (seen) => [...seen, authorization]); } + const reply = (yield* Ref.get(replies)).get(request.url); + if (reply !== undefined) + return HttpServerResponse.jsonUnsafe(reply.body, { status: reply.status }); const response = yield* httpEffect; // Keep real authentication and dispatch, but exercise richer peer wire results. const result = yield* Ref.get(toolResult); if (result !== null && request.url === "/mcp" && response.status === 200) { const rpc = yield* decodeMcpRequest(yield* request.json); if (rpc.method === "tools/call") { + if (yield* Ref.get(sse)) { + return HttpServerResponse.text( + [ + 'data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"result":"notification"}}', + "", + 'data: {"jsonrpc":"2.0","id":999,"error":{"code":-1,"message":"other request"}}', + "", + 'data: {"jsonrpc":"2.0",', + 'data: "id":' + rpc.id + ',"result":' + JSON.stringify(result) + "}", + "", + "", + ].join("\r\n"), + { contentType: "text/event-stream" }, + ); + } return HttpServerResponse.jsonUnsafe({ jsonrpc: "2.0", id: rpc.id, result }); } } @@ -204,6 +226,8 @@ const serveB = (initial: ExecutionEnvironmentDescriptor) => alternateDescriptor, bearers, toolResult, + sse, + replies, redirects, redirectedBodies, /** How often its descriptor was asked for, as a probe of whether it answers. */ @@ -215,6 +239,7 @@ const serveB = (initial: ExecutionEnvironmentDescriptor) => const makeA = ( descriptor: ExecutionEnvironmentDescriptor, afterSecretWrite: Effect.Effect = Effect.void, + beforeSecretRemove: Effect.Effect = Effect.void, ) => Effect.gen(function* () { const secretReads = yield* Ref.make(0); @@ -225,6 +250,7 @@ const makeA = ( return ServerSecretStore.ServerSecretStore.of({ ...secrets, set: (name, value) => secrets.set(name, value).pipe(Effect.andThen(afterSecretWrite)), + remove: (name) => beforeSecretRemove.pipe(Effect.andThen(secrets.remove(name))), get: (name) => Ref.update(secretReads, (count) => count + 1).pipe(Effect.andThen(secrets.get(name))), }); @@ -289,7 +315,29 @@ it.effect("links, reads and forgets a peer while preserving access limits and re Effect.scoped( Effect.gen(function* () { const b = yield* serveB(box); - const a = yield* makeA(laptop); + const removal = yield* Ref.make<"fail" | "pause" | null>(null); + const removalEntered = yield* Deferred.make(); + const allowRemoval = yield* Deferred.make(); + const a = yield* makeA( + laptop, + Effect.void, + Ref.get(removal).pipe( + Effect.flatMap((action) => + action === "fail" + ? Effect.fail( + new ServerSecretStore.SecretStoreRemoveError({ + resource: "test credential", + cause: new Error("removal failed"), + }), + ) + : action === "pause" + ? Deferred.succeed(removalEntered, undefined).pipe( + Effect.andThen(Deferred.await(allowRemoval)), + ) + : Effect.void, + ), + ), + ); const refused = yield* a.links .link({ url: b.url, pairingCode: "not-a-code", access: "auto" }) @@ -332,6 +380,31 @@ it.effect("links, reads and forgets a peer while preserving access limits and re const notFound = yield* probe(a, undefined, "refuse").pipe(Effect.flip); expect(notFound).toMatchObject({ code: "thread_not_found", message: "No such thread." }); + const names = yield* a.secretNames; + const beforeUnlink = yield* a.links.get(box.environmentId); + yield* Ref.set(removal, "fail"); + expect((yield* a.links.unlink(box.environmentId).pipe(Effect.flip)).reason).toBe("storage"); + expect(yield* a.links.get(box.environmentId)).toEqual(beforeUnlink); + expect(yield* a.secretNames).toEqual(names); + expect(yield* probe(a)).toEqual({ caller: "T3 Code · Laptop", modes: "auto/default" }); + + yield* Ref.set(removal, "pause"); + const unlinking = yield* a.links.unlink(box.environmentId).pipe(Effect.forkScoped); + yield* Deferred.await(removalEntered); + yield* Ref.set(removal, null); + const newPairing = yield* b.auth.issuePairingCredential(); + const relinked = yield* a.links.link({ + url: b.url, + pairingCode: newPairing.credential, + access: "approval-required", + }); + yield* Deferred.succeed(allowRemoval, undefined); + expect(yield* Fiber.join(unlinking)).toBe(false); + expect(yield* a.links.get(box.environmentId)).toEqual(Option.some(relinked)); + expect(yield* probe(a)).toEqual({ + caller: "T3 Code · Laptop", + modes: "approval-required/default", + }); expect(yield* a.links.unlink(box.environmentId)).toBe(true); expect(yield* a.links.unlink(box.environmentId)).toBe(false); expect(yield* a.links.get(box.environmentId)).toEqual(Option.none()); @@ -340,7 +413,7 @@ it.effect("links, reads and forgets a peer while preserving access limits and re yield* Ref.set(b.bearers, []); expect((yield* probe(a).pipe(Effect.flip)).code).toBe("invalid_request"); expect(yield* Ref.get(b.bearers)).toHaveLength(0); - expect(yield* linkedSessions(b)).toHaveLength(1); + expect(yield* linkedSessions(b)).toHaveLength(2); }), ).pipe(Effect.provide(Layer.merge(NodeHttpServer.layerTest, NodeServices.layer))), ); @@ -506,6 +579,51 @@ it.effect( .pipe(Effect.flip); expect(lanAlternate.message).toContain("plain http"); + for (const url of [ + "file://username:password@example.test/path?token=private", + "http://username:password@192.168.1.20:3773/path?token=private", + ]) { + const invalid = yield* a.links + .link({ url, pairingCode: "unused", access: "auto" }) + .pipe(Effect.flip); + expect(invalid.message).not.toMatch(/username|password|private|example\.test/); + } + yield* Ref.set(b.descriptor, box); + const peerText = "PRIVATE-PEER-RESPONSE\n".repeat(200); + for (const [path, body, message] of [ + [ + "/oauth/mcp/register", + { error: "invalid_client_metadata", error_description: peerText }, + "The peer refused to register this environment. Check its connection settings.", + ], + [ + "/oauth/mcp/decision", + { _tag: "AuthMcpApprovalError", message: peerText }, + "The peer refused link approval. Check the pairing code and try again.", + ], + [ + "/oauth/mcp/token", + { error: "invalid_grant", error_description: peerText }, + "The peer refused the sign-in token. Create a new pairing code and try again.", + ], + [ + "/oauth/mcp/decision", + { redirectTo: "http://localhost/?error_description=" + encodeURIComponent(peerText) }, + "The peer did not approve the link. Check the pairing code and try again.", + ], + ] as const) { + yield* Ref.set( + b.replies, + new Map([[path, { status: "redirectTo" in body ? 200 : 400, body }]]), + ); + const code = yield* b.auth.issuePairingCredential(); + const failed = yield* a.links + .link({ url: b.url, pairingCode: code.credential, access: "auto" }) + .pipe(Effect.flip); + expect(failed.reason).toBe("pairing_rejected"); + expect(failed.message).toBe(message); + } + yield* Ref.set(b.replies, new Map()); yield* Ref.set(b.descriptor, laptop); const self = yield* a.links .link({ url: b.url, pairingCode: pairing.credential, access: "auto" }) @@ -545,6 +663,8 @@ it.effect( }), ); expect(yield* probe(a)).toEqual(result); + yield* Ref.set(b.sse, true); + expect(yield* probe(a)).toEqual(result); yield* Ref.set( b.toolResult, diff --git a/apps/server/src/peer/PeerLinks.ts b/apps/server/src/peer/PeerLinks.ts index 4a8a9208ffeb1..b0d2536e2a968 100644 --- a/apps/server/src/peer/PeerLinks.ts +++ b/apps/server/src/peer/PeerLinks.ts @@ -103,7 +103,7 @@ const peerOrigin = (url: string) => { return Effect.fail( new PeerLinkError({ reason: "unreachable", - message: `${url} is not an http or https address.`, + message: "The peer address must use http or https.", }), ); } @@ -115,7 +115,8 @@ const peerOrigin = (url: string) => { return Effect.fail( new PeerLinkError({ reason: "unreachable", - message: `${url} is plain http, which would send the link's credentials unencrypted. Use HTTPS, or HTTP with loopback, a Tailscale .ts.net name, or Tailscale IPv6.`, + message: + "The peer address uses plain http, which would send the link's credentials unencrypted. Use HTTPS, or HTTP with loopback, a Tailscale .ts.net name, or Tailscale IPv6.", }), ); } @@ -190,7 +191,9 @@ const make = Effect.gen(function* () { Effect.fail( new PeerLinkError({ reason: "pairing_rejected", - message: `The peer refused to register this environment: ${error.message}`, + message: + "The peer refused to register this environment. Check its connection settings.", + cause: error, }), ), ), @@ -220,7 +223,11 @@ const make = Effect.gen(function* () { Effect.catchTags({ AuthMcpApprovalError: (error) => Effect.fail( - new PeerLinkError({ reason: "pairing_rejected", message: error.message }), + new PeerLinkError({ + reason: "pairing_rejected", + message: "The peer refused link approval. Check the pairing code and try again.", + cause: error, + }), ), }), ); @@ -235,9 +242,7 @@ const make = Effect.gen(function* () { 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.", + message: "The peer did not approve the link. Check the pairing code and try again.", }); } return yield* client.mcpOAuth @@ -256,7 +261,9 @@ const make = Effect.gen(function* () { Effect.fail( new PeerLinkError({ reason: "pairing_rejected", - message: `The peer refused the sign-in: ${error.message}`, + message: + "The peer refused the sign-in token. Create a new pairing code and try again.", + cause: error, }), ), ), @@ -422,13 +429,14 @@ const make = Effect.gen(function* () { const unlink: PeerLinks["Service"]["unlink"] = (environmentId) => Effect.gen(function* () { + const row = yield* readRow(environmentId); + if (Option.isNone(row)) return false; + yield* secrets.remove(row.value.secret_name).pipe(Effect.catch(failStorage)); const deleted = yield* sql<{ secret_name: string }>` - DELETE FROM peer_environment_links WHERE environment_id = ${environmentId} + DELETE FROM peer_environment_links + WHERE environment_id = ${environmentId} AND secret_name = ${row.value.secret_name} 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; }); diff --git a/apps/server/src/peer/PeerMcpClient.ts b/apps/server/src/peer/PeerMcpClient.ts index 511836d623d00..bc92e8c67c269 100644 --- a/apps/server/src/peer/PeerMcpClient.ts +++ b/apps/server/src/peer/PeerMcpClient.ts @@ -45,6 +45,7 @@ export class PeerMcpClient extends Context.Service< >()("t3/peer/PeerMcpClient") {} const ToolCallResponse = Schema.Struct({ + id: Schema.Number, result: Schema.optional( Schema.Struct({ content: Schema.Array( @@ -56,7 +57,9 @@ const ToolCallResponse = Schema.Struct({ ), error: Schema.optional(Schema.Struct({ code: Schema.Number, message: Schema.String })), }); -const decodeToolCallResponse = Schema.decodeUnknownEffect(Schema.fromJsonString(ToolCallResponse)); +const ToolCallResponseJson = Schema.fromJsonString(ToolCallResponse); +const decodeToolCallResponse = Schema.decodeUnknownEffect(ToolCallResponseJson); +const decodeToolCallResponseOption = Schema.decodeUnknownOption(ToolCallResponseJson); const Failure = Schema.Struct({ code: OrchestratorMcpFailure.fields.code, message: Schema.String }); const decodeFailure = Schema.decodeUnknownOption(Failure); const decodeTextFailure = Schema.decodeUnknownOption(Schema.fromJsonString(Failure)); @@ -73,15 +76,25 @@ const fromLinkError = (error: PeerLinkError) => 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 responseJson = (contentType: string | undefined, text: string, requestId: number) => { + if (contentType?.includes("text/event-stream") !== true) return text; + for (const event of text.replace(/\r\n?/g, "\n").split("\n\n")) { + const data = event + .split("\n") + .filter((line) => line.startsWith("data:")) + .map((line) => line.slice("data:".length).replace(/^ /, "")) + .join("\n"); + const decoded = decodeToolCallResponseOption(data); + if ( + decoded._tag === "Some" && + decoded.value.id === requestId && + (decoded.value.result !== undefined || decoded.value.error !== undefined) + ) { + return data; + } + } + return ""; +}; const make = Effect.gen(function* () { const links = yield* PeerLinks.PeerLinks; @@ -187,7 +200,7 @@ const make = Effect.gen(function* () { }); } const body = yield* decodeToolCallResponse( - responseJson(response.headers["content-type"], yield* response.text), + responseJson(response.headers["content-type"], yield* response.text, 1), ).pipe( Effect.mapError( () => diff --git a/apps/server/src/persistence/Migrations/064_CanonicalizeThreadPullRequestLinks.test.ts b/apps/server/src/persistence/Migrations/064_CanonicalizeThreadPullRequestLinks.test.ts index 706cc02760eac..69eb8eff57027 100644 --- a/apps/server/src/persistence/Migrations/064_CanonicalizeThreadPullRequestLinks.test.ts +++ b/apps/server/src/persistence/Migrations/064_CanonicalizeThreadPullRequestLinks.test.ts @@ -123,8 +123,10 @@ layer("064_CanonicalizeThreadPullRequestLinks", (it) => { 'user', ${v1Payload}, '{}', 1) `; - assert.deepEqual(yield* runMigrations(), [[64, "CanonicalizeThreadPullRequestLinks"]]); - assert.deepEqual(yield* runMigrations(), []); + assert.deepEqual(yield* runMigrations({ toMigrationInclusive: 64 }), [ + [64, "CanonicalizeThreadPullRequestLinks"], + ]); + assert.deepEqual(yield* runMigrations({ toMigrationInclusive: 64 }), []); const rows = yield* sql<{ readonly thread_id: string; readonly payload_json: string }>` SELECT thread_id, payload_json FROM orchestration_v2_projection_threads `;