1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import {
4 EnvironmentHttpApi,
5 EnvironmentId,
6 type AuthMcpClientAccess,
7 type ExecutionEnvironmentDescriptor,
8} from "@t3tools/contracts";
9import * as Context from "effect/Context";
10import * as Deferred from "effect/Deferred";
11import * as Effect from "effect/Effect";
12import * as Layer from "effect/Layer";
13import * as Queue from "effect/Queue";
14import * as Ref from "effect/Ref";
15import * as FetchHttpClient from "effect/http/FetchHttpClient";
16import * as HttpRouter from "effect/http/HttpRouter";
17import * as HttpServer from "effect/http/HttpServer";
18import * as HttpServerRequest from "effect/http/HttpServerRequest";
19import * as HttpApi from "effect/http-api/HttpApi";
20import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
21
22import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts";
23import * as AuthHttp from "../auth/http.ts";
24import * as McpOAuth from "../auth/McpOAuth.ts";
25import * as McpOAuthHttp from "../auth/mcpOAuthHttp.ts";
26import type * as SqlClient from "effect/sql/SqlClient";
27import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
28import * as CloudLink from "../cloud/CloudLink.ts";
29import * as ServerConfig from "../config.ts";
30import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
31import * as ServerHttp from "../http.ts";
32import * as McpHttpServer from "../mcp/McpHttpServer.ts";
33import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts";
34import * as Sqlite from "../persistence/Sqlite.ts";
35import * as PeerLinks from "./PeerLinks.ts";
36import * as ServerSettings from "../serverSettings.ts";
37import * as PeerMcpClient from "./PeerMcpClient.ts";
38import * as ManagedPolicy from "../policy/ManagedPolicy.ts";
39
40class PeerTestApi extends HttpApi.make("environment")
41 .add(EnvironmentHttpApi.groups.metadata)
42 .add(EnvironmentHttpApi.groups.mcpOAuth) {}
43
44export const descriptorOf = (
45 environmentId: string,
46 label: string,
47 capabilities: Partial<ExecutionEnvironmentDescriptor["capabilities"]> = {},
48): ExecutionEnvironmentDescriptor => ({
49 environmentId: EnvironmentId.make(environmentId),
50 label,
51 platform: { os: "linux", arch: "x64" },
52 serverVersion: "0.0.0-test",
53 capabilities: {
54 repositoryIdentity: true,
55 mcpModeLimitHeader: true,
56 linkFence: true,
57 ...capabilities,
58 },
59});
60
61/**
62 * The environment a link points at, as it runs: its descriptor, MCP OAuth,
63 * and `/mcp` with `tools` behind the real client authenticator, on a real
64 * socket. The descriptor can be swapped, as when its address comes to answer
65 * as another environment, and every bearer it receives is recorded.
66 */
67export const servePeer = <E, R>(
68 initial: ExecutionEnvironmentDescriptor,
69 tools: Layer.Layer<never, E, R>,
70) =>
71 Effect.gen(function* () {
72 const descriptor = yield* Ref.make(initial);
73 const bearers = yield* Ref.make<ReadonlyArray<string>>([]);
74 const probes = yield* Ref.make(0);
75 const asleep = yield* Ref.make(false);
76 const unanswered = yield* Queue.unbounded<void>();
77 const wake = yield* Deferred.make<void>();
78 const layerEnvironment = Layer.succeed(ServerEnvironment.ServerEnvironment, {
79 getEnvironmentId: Ref.get(descriptor).pipe(Effect.map((current) => current.environmentId)),
80 getDescriptor: Ref.get(descriptor),
81 });
82 const authContext = yield* EnvironmentAuth.layer.pipe(
83 Layer.provide(ManagedPolicy.layerTest()),
84 Layer.provideMerge(ServerSettings.layerTest({ linkedEnvironmentsEnabled: true })),
85 Layer.provide(Sqlite.layerMemory),
86 Layer.provideMerge(ServerSecretStore.layer),
87 Layer.provideMerge(ServerEnvironment.layerIdentity),
88 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "t3-peer-link-peer-" })),
89 Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
90 Layer.fresh,
91 Layer.build,
92 );
93 const layerRecordBearers = HttpRouter.middleware(
94 (httpEffect) =>
95 Effect.gen(function* () {
96 const request = yield* HttpServerRequest.HttpServerRequest;
97 if (request.url.startsWith("/.well-known/t3/environment")) {
98 yield* Ref.update(probes, (count) => count + 1);
99 if (yield* Ref.get(asleep)) {
100 yield* Queue.offer(unanswered, undefined);
101 yield* Deferred.await(wake);
102 return yield* Effect.die("the peer was closed while asleep");
103 }
104 }
105 const authorization = request.headers.authorization;
106 if (authorization !== undefined) {
107 yield* Ref.update(bearers, (seen) => [...seen, authorization]);
108 }
109 return yield* httpEffect;
110 }),
111 { global: true },
112 );
113 const layerRoutes = Layer.mergeAll(
114 HttpApiBuilder.layer(PeerTestApi).pipe(
115 Layer.provide(
116 McpOAuthHttp.layer.pipe(
117 Layer.provide(McpOAuth.layer),
118 Layer.provide(
119 Layer.mock(CloudLink.CloudLink)({
120 linkState: () =>
121 Effect.succeed({
122 linked: false,
123 cloudUserId: null,
124 relayUrl: null,
125 relayIssuer: null,
126 publishAgentActivity: false,
127 }),
128 }),
129 ),
130 ),
131 ),
132 Layer.provide(ServerHttp.layerServerEnvironmentHttpApi),
133 Layer.provide(AuthHttp.layerAuthenticatedAuth),
134 ),
135 McpHttpServer.layerMcpTransport(tools).pipe(
136 Layer.provide(
137 Layer.mock(McpSessionRegistry.McpSessionRegistry)({
138 resolve: () => Effect.undefined,
139 }),
140 ),
141 Layer.provide(McpOAuth.layerMcpClientAuthenticator),
142 ),
143 ).pipe(
144 Layer.provide(layerRecordBearers),
145 Layer.provide(layerEnvironment),
146 Layer.provide(Layer.succeedContext(authContext)),
147 );
148 // Fresh, so the peer never shares a toolkit or MCP server with the
149 // environment that links to it in the same test.
150 yield* HttpRouter.serve(layerRoutes, { disableListenLog: true, disableLogger: true }).pipe(
151 Layer.fresh,
152 Layer.build,
153 );
154 // An unanswered request would hold the server open past the test.
155 yield* Effect.addFinalizer(() => Deferred.succeed(wake, undefined));
156 const address = (yield* HttpServer.HttpServer).address;
157 if (address._tag === "UnixPathAddress") return yield* Effect.die("expected a TCP address");
158 const url = `http://127.0.0.1:${address.port}`;
159 const auth = Context.get(authContext, EnvironmentAuth.EnvironmentAuth);
160 return {
161 url,
162 auth,
163 /** Its settings, as for turning its Linked environments beta off. */
164 settings: Context.get(authContext, ServerSettings.ServerSettingsService),
165 descriptor,
166 bearers,
167 /** How often its descriptor was asked for, as a probe of whether it answers. */
168 probes,
169 /** While set, its descriptor never answers, as a machine that is asleep. */
170 asleep,
171 /** One entry per descriptor request it left unanswered while asleep. */
172 unanswered,
173 /** The sessions linking created there, which its Connections lists. */
174 linkedSessions: auth
175 .listSessions()
176 .pipe(
177 Effect.map((sessions) =>
178 sessions.filter((session) => session.client.label?.startsWith("T3 Code · ")),
179 ),
180 ),
181 };
182 });
183
184export type ServedPeer = Effect.Success<ReturnType<typeof servePeer>>;
185
186/**
187 * The environment that links: its own database and secret store, reaching
188 * peers over plain HTTP. Provide it to anything built on its links.
189 */
190export const layerLinkingEnvironment = <E = never>(
191 descriptor: ExecutionEnvironmentDescriptor,
192 secrets: typeof ServerSecretStore.layer = ServerSecretStore.layer,
193 database: Layer.Layer<
194 SqlClient.SqlClient,
195 E | Layer.Error<typeof Sqlite.layerMemory>
196 > = Sqlite.layerMemory,
197) =>
198 PeerMcpClient.layer.pipe(
199 Layer.provideMerge(PeerLinks.layer),
200 Layer.provideMerge(ServerSettings.layerTest({ linkedEnvironmentsEnabled: true })),
201 Layer.provideMerge(database),
202 Layer.provide(Layer.mergeAll(secrets, FetchHttpClient.layer)),
203 // Peers here are served on 127.0.0.1, so only `localhost` stands for this machine.
204 Layer.provide(Layer.succeed(PeerLinks.LoopbackHost, (host) => host === "localhost")),
205 // Services that act for this environment, such as delegation, name it too.
206 Layer.provideMerge(
207 Layer.succeed(ServerEnvironment.ServerEnvironment, {
208 getEnvironmentId: Effect.succeed(descriptor.environmentId),
209 getDescriptor: Effect.succeed(descriptor),
210 }),
211 ),
212 Layer.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-peer-link-here-" })),
213 Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
214 Layer.fresh,
215 );
216
217/** Links `links` to `peer` with a fresh pairing code from it. */
218export const linkTo = (
219 links: PeerLinks.PeerLinks["Service"],
220 peer: ServedPeer,
221 access: AuthMcpClientAccess = "auto",
222) =>
223 peer.auth
224 .issuePairingCredential()
225 .pipe(
226 Effect.flatMap((pairing) =>
227 links.link({ url: peer.url, pairingCode: pairing.credential, access }),
228 ),
229 );