1import {
2 type EnvironmentId,
3 type OrchestrationV2LinkOrigin,
4 OrchestratorMcpEnvironmentTarget,
5 OrchestratorMcpFailure,
6 type ProviderInteractionMode,
7 type RuntimeMode,
8 type ThreadId,
9} from "@t3tools/contracts";
10import * as Effect from "effect/Effect";
11import * as Option from "effect/Option";
12import * as Predicate from "effect/Predicate";
13import * as Ref from "effect/Ref";
14import * as Struct from "effect/Struct";
15import type * as Layer from "effect/Layer";
16import type * as Scope from "effect/Scope";
17import type { Tool, Toolkit } from "effect/ai";
18
19import {
20 DispatchModeLimit,
21 type DispatchModeRefusal,
22} from "../orchestration-v2/DispatchModeLimit.ts";
23import * as PeerForwarding from "../peer/PeerForwarding.ts";
24import { assertNotLinked, assertSameLink, callerLinkOrigin } from "./linkOrigin.ts";
25import * as McpInvocationContext from "./McpInvocationContext.ts";
26import { resolveInteractionMode, resolveRuntimeMode } from "./delegatedTaskModes.ts";
27import {
28 assertFullAccess,
29 assertLiveCaller,
30 assertTargetWithinLimits,
31 type Caller,
32 loadCaller,
33 readCaller,
34 unavailable,
35} from "./threadAccess.ts";
36
37/** Private constructors only enforce types; the token also blocks runtime construction. */
38const builtHere: unique symbol = Symbol("t3/mcp/McpToolAccess/builtHere");
39
40const refuseOutsideConstruction = (token: symbol) => {
41 if (token !== builtHere) {
42 throw new TypeError("Only McpToolAccess builds MCP tool declarations and handler layers.");
43 }
44};
45
46let declare: <P, A, E, R>(
47 handle: (params: P) => Effect.Effect<A, E, R>,
48 remote: RemoteCheck<P> | undefined,
49 forwarded?: Forwarded<P, A, E, R>,
50) => Declaration<(params: P) => Effect.Effect<A, E, R>>;
51
52/**
53 * What a call bound for a linked environment checks here before it goes: the
54 * caller, not the thread or project it names, which live there and are
55 * checked there. `undefined` for tools that act as the calling thread itself.
56 */
57type RemoteCheck<P> = (
58 params: P,
59) => Effect.Effect<unknown, OrchestratorMcpFailure, Effect.Services<typeof writingCaller>>;
60let remoteCheckOf: <Handler>(declaration: Declaration<Handler>) => RemoteCheck<never> | undefined;
61
62/** What a call answered in a linked environment does here before it returns. */
63type Forwarded<P, A, E, R> = (
64 params: P,
65 environmentId: EnvironmentId,
66 result: A,
67) => Effect.Effect<A, E, R>;
68let forwardedOf: <Handler>(
69 declaration: Declaration<Handler>,
70) => Forwarded<never, unknown, unknown, unknown> | undefined;
71
72type CheckedHandlerOf<D> = D extends Declaration<infer Handler> ? Handler : never;
73
74interface CheckedHandler extends Struct.Lambda {
75 <Handler>(declaration: Declaration<Handler>): Handler;
76 readonly "~lambda.out": CheckedHandlerOf<this["~lambda.in"]>;
77}
78let checkedHandler: CheckedHandler;
79
80let handlersLayer: <Tools extends Record<string, Tool.Any>, EX, RX>(
81 layer: Layer.Layer<Tool.HandlersFor<Tools>, EX, RX>,
82) => HandlersLayer<Tools, EX, RX>;
83
84/** t3code/no-raw-mcp-registration confines Effect's unchecked registration API to McpHttpServer. */
85export class Declaration<out Handler> {
86 readonly #handle: Handler;
87 readonly #remote: RemoteCheck<never> | undefined;
88 readonly #forwarded: Forwarded<never, unknown, unknown, unknown> | undefined;
89 private constructor(
90 token: typeof builtHere,
91 handle: Handler,
92 remote: RemoteCheck<never> | undefined,
93 forwarded: Forwarded<never, unknown, unknown, unknown> | undefined,
94 ) {
95 refuseOutsideConstruction(token);
96 this.#handle = handle;
97 this.#remote = remote;
98 this.#forwarded = forwarded;
99 }
100 static {
101 declare = (handle, remote, forwarded) =>
102 new Declaration(
103 builtHere,
104 handle,
105 remote,
106 forwarded as Forwarded<never, unknown, unknown, unknown> | undefined,
107 );
108 checkedHandler = Struct.lambda<CheckedHandler>((declaration) => declaration.#handle);
109 remoteCheckOf = (declaration) => declaration.#remote;
110 forwardedOf = (declaration) => declaration.#forwarded;
111 }
112}
113
114const refuseReadOnlyClient = McpInvocationContext.McpInvocationContext.pipe(
115 Effect.flatMap((scope) =>
116 scope.client?.access === "read-only"
117 ? Effect.fail(
118 new OrchestratorMcpFailure({
119 code: "capability_denied",
120 message:
121 "This tool changes the environment, and this MCP client was approved for read-only access.",
122 }),
123 )
124 : Effect.void,
125 ),
126);
127
128const writingCaller = refuseReadOnlyClient.pipe(
129 Effect.andThen(loadCaller()),
130 Effect.tap(assertLiveCaller),
131);
132
133const orchestratingCaller = refuseReadOnlyClient.pipe(
134 Effect.andThen(readCaller()),
135 Effect.tap(assertLiveCaller),
136);
137
138const requireThreadCaller = McpInvocationContext.McpInvocationContext.pipe(
139 Effect.flatMap((scope) => McpInvocationContext.requireThreadScope(scope, "This tool")),
140);
141
142export const reads = <P, A, E, R>(handle: (params: P) => Effect.Effect<A, E, R>) =>
143 declare(
144 (params: P) => handle(params),
145 () => Effect.void,
146 );
147
148export const readsAsCaller = <P, A, E, R>(handle: (params: P) => Effect.Effect<A, E, R>) =>
149 declare((params: P) => requireThreadCaller.pipe(Effect.flatMap(() => handle(params))), undefined);
150
151export const actsAsCaller = <P, A, E, R>(handle: (params: P) => Effect.Effect<A, E, R>) =>
152 declare(
153 (params: P) =>
154 requireThreadCaller.pipe(
155 Effect.andThen(writingCaller),
156 Effect.flatMap(() => handle(params)),
157 ),
158 undefined,
159 );
160
161/**
162 * Changes something that belongs to no thread, such as a pending upload or a
163 * scheduled task. Work a linked environment started may not: none of it
164 * carries the link's origin, so none of it is that link's.
165 */
166export const writes = <P, A, E, R>(
167 handle: (params: P) => Effect.Effect<A, E, R>,
168 options?: {
169 /** Whether a linked caller may make it, as for uploads its own sends need. */
170 readonly linkedCallers?: "allowed";
171 },
172) =>
173 declare(
174 (params: P) =>
175 orchestratingCaller.pipe(
176 Effect.tap((caller) =>
177 options?.linkedCallers === "allowed"
178 ? Effect.void
179 : assertNotLinked(caller, "change this environment's scheduled tasks"),
180 ),
181 Effect.flatMap(() => handle(params)),
182 ),
183 () => orchestratingCaller,
184 );
185
186/** Requires handler capability checks, not thread control; linked work is refused the same way. */
187export const writesWithCapability = <P, A, E, R>(handle: (params: P) => Effect.Effect<A, E, R>) =>
188 declare(
189 (params: P) =>
190 writingCaller.pipe(
191 Effect.tap((caller) => assertNotLinked(caller, "start work here that is not its link's")),
192 Effect.flatMap(() => handle(params)),
193 ),
194 () => writingCaller,
195 );
196
197export const writesThreads = <P, A, E, R>(
198 threads: (params: P) => ReadonlyArray<ThreadId | undefined>,
199 handle: (params: P) => Effect.Effect<A, E, R>,
200) =>
201 declare(
202 (params: P) =>
203 Effect.gen(function* () {
204 const caller = yield* writingCaller;
205 for (const threadId of threads(params)) {
206 if (threadId === undefined || threadId === caller.scope.thread?.threadId) continue;
207 const target = yield* caller.threads
208 .getThreadShell(threadId)
209 .pipe(Effect.mapError(unavailable));
210 if (target !== null && target.deletedAt === null) {
211 yield* assertTargetWithinLimits(caller.limits, target);
212 yield* assertSameLink(caller, target);
213 }
214 }
215 // Modes can rise after this check; the orchestrator rechecks under lock and records refusal before handlers wrap errors.
216 const refused = yield* Ref.make<DispatchModeRefusal | undefined>(undefined);
217 return yield* handle(params).pipe(
218 Effect.provideService(DispatchModeLimit, { ...caller.limits, refused }),
219 Effect.catch((error) =>
220 Effect.flatMap(Ref.get(refused), (refusal) =>
221 Effect.fail<E | OrchestratorMcpFailure>(
222 refusal === undefined ? error : escalationDenied(refusal),
223 ),
224 ),
225 ),
226 );
227 }),
228 () => writingCaller,
229 );
230
231const escalationDenied = (refusal: DispatchModeRefusal) =>
232 new OrchestratorMcpFailure({
233 code:
234 refusal.mode === "runtime"
235 ? "runtime_mode_escalation_denied"
236 : "interaction_mode_escalation_denied",
237 message: `Thread ${refusal.threadId} now runs in ${refusal.runtimeMode}/${refusal.interactionMode} mode, above this caller's. Its user changed it while this call ran.`,
238 });
239
240export interface StartedModes {
241 readonly runtimeMode: RuntimeMode;
242 readonly interactionMode: ProviderInteractionMode;
243 /** The linked environment the caller acts for. The thread must carry it. */
244 readonly linkOrigin: OrchestrationV2LinkOrigin | undefined;
245}
246
247/**
248 * Starts threads with the modes `modes` requests, which may not be broader
249 * than the caller's. `linked` says what a linked caller gets: threads stamped
250 * with its `linkOrigin`, or a refusal where the tool cannot carry one.
251 */
252export const startsThreads = <P, A, E, R>(
253 modes: (params: P) => {
254 readonly runtimeMode?: RuntimeMode | undefined;
255 readonly interactionMode?: ProviderInteractionMode | undefined;
256 },
257 handle: (params: P, modes: StartedModes) => Effect.Effect<A, E, R>,
258 linked: "stamped" | { readonly refused: string },
259) => {
260 const started = Effect.fn("McpToolAccess.started")(function* (caller: Caller, params: P) {
261 const requested = modes(params);
262 return {
263 runtimeMode: yield* resolveRuntimeMode(caller.limits.runtimeMode, requested.runtimeMode),
264 interactionMode: yield* resolveInteractionMode(
265 caller.limits.interactionMode,
266 requested.interactionMode,
267 ),
268 linkOrigin: callerLinkOrigin(caller),
269 } satisfies StartedModes;
270 });
271 return declare(
272 (params: P) =>
273 Effect.gen(function* () {
274 const caller = yield* writingCaller;
275 if (linked !== "stamped") yield* assertNotLinked(caller, linked.refused);
276 return yield* handle(params, yield* started(caller, params));
277 }),
278 (params: P) => Effect.flatMap(writingCaller, (caller) => started(caller, params)),
279 );
280};
281
282/** One tool with two access rules, picked per call, such as create versus update. */
283export const either = <P, A, E, R, A2, E2, R2>(
284 useFirst: (params: P) => boolean,
285 first: Declaration<(params: P) => Effect.Effect<A, E, R>>,
286 second: Declaration<(params: P) => Effect.Effect<A2, E2, R2>>,
287) => {
288 const remotes = [remoteCheckOf(first), remoteCheckOf(second)] as ReadonlyArray<
289 RemoteCheck<P> | undefined
290 >;
291 return declare(
292 (params: P): Effect.Effect<A | A2, E | E2, R | R2> =>
293 useFirst(params) ? checkedHandler(first)(params) : checkedHandler(second)(params),
294 (params) =>
295 remotes[useFirst(params) ? 0 : 1]?.(params) ??
296 Effect.fail(
297 new OrchestratorMcpFailure({
298 code: "invalid_request",
299 message: "This action acts as the calling thread, which is here. Omit environmentId.",
300 }),
301 ),
302 );
303};
304
305/**
306 * `declaration`, whose result from a linked environment then goes through
307 * `forwarded` here, for what this environment records about the call.
308 */
309export const afterForwarding = <P, A, E, R, E2, R2>(
310 declaration: Declaration<(params: P) => Effect.Effect<A, E, R>>,
311 forwarded: Forwarded<P, A, E2, R2>,
312) =>
313 declare(
314 checkedHandler(declaration) as (params: P) => Effect.Effect<A, E | E2, R | R2>,
315 remoteCheckOf(declaration) as RemoteCheck<P> | undefined,
316 forwarded,
317 );
318
319const fullAccessRequired =
320 "Changing projects or environment settings needs a live full-access/default calling thread or a full-access client.";
321
322/** Recheck after waits before writing; caller modes may have changed. */
323export const writesEnvironment = <P, A, E, R>(
324 handle: (
325 params: P,
326 check: Effect.Effect<Caller, OrchestratorMcpFailure, CheckServices>,
327 ) => Effect.Effect<A, E, R>,
328 options?: {
329 /**
330 * Whether a linked caller may make it, as for adding a project: it leaves
331 * everything already here as it was.
332 */
333 readonly linkedCallers?: "allowed";
334 },
335) => {
336 const check = orchestratingCaller.pipe(
337 Effect.tap((caller) =>
338 options?.linkedCallers === "allowed"
339 ? Effect.void
340 : assertNotLinked(caller, "change projects or environment settings"),
341 ),
342 Effect.tap((caller) => assertFullAccess(caller, fullAccessRequired)),
343 );
344 return declare(
345 (params: P) => check.pipe(Effect.flatMap(() => handle(params, check))),
346 // Linked work never reaches a link at all; that is checked as the call goes.
347 () =>
348 orchestratingCaller.pipe(
349 Effect.tap((caller) => assertFullAccess(caller, fullAccessRequired)),
350 ),
351 );
352};
353
354type CheckServices = Effect.Services<typeof orchestratingCaller>;
355
356type Declarations<Handlers> = {
357 readonly [Name in keyof Handlers]: Declaration<Handlers[Name]> & NotCallable;
358};
359
360type NotCallable = { readonly call?: never } & { readonly apply?: never };
361
362export type Handlers<Tools extends Record<string, Tool.Any>> = Declarations<
363 Toolkit.HandlersFrom<Tools>
364>;
365
366export class HandlersLayer<Tools extends Record<string, Tool.Any>, EX = never, RX = never> {
367 readonly #layer: Layer.Layer<Tool.HandlersFor<Tools>, EX, RX>;
368 private constructor(
369 token: typeof builtHere,
370 layer: Layer.Layer<Tool.HandlersFor<Tools>, EX, RX>,
371 ) {
372 refuseOutsideConstruction(token);
373 this.#layer = layer;
374 }
375 static {
376 handlersLayer = (layer) => new HandlersLayer(builtHere, layer);
377 }
378 static layer<Tools extends Record<string, Tool.Any>, EX, RX>(
379 handlers: HandlersLayer<Tools, EX, RX>,
380 ) {
381 return handlers.#layer;
382 }
383}
384
385const checkedHandlers = <Handlers>(declarations: Declarations<Handlers>): Handlers =>
386 Struct.map(declarations, checkedHandler);
387
388/**
389 * A tool runs in a linked environment when its parameters take this target
390 * as `environmentId`. Other tools may name an environment for their own
391 * reasons, such as the one to link or to move a thread to.
392 */
393const takesLinkedEnvironment = (tool: Tool.Any) => {
394 const parameters: unknown = tool.parametersSchema;
395 return (
396 Predicate.hasProperty(parameters, "fields") &&
397 Predicate.hasProperty(parameters.fields, "environmentId") &&
398 parameters.fields.environmentId === OrchestratorMcpEnvironmentTarget
399 );
400};
401
402const linksUnavailable = new OrchestratorMcpFailure({
403 code: "orchestration_error",
404 message: "This environment cannot reach linked environments.",
405});
406
407/**
408 * Every tool that takes a linked environment runs there the same way: its
409 * declaration checks the caller here, and the same call goes to the other
410 * environment, which checks the rest. The handler here never sees
411 * `environmentId`, so it cannot leak into what it writes.
412 */
413const routed = Effect.fn("McpToolAccess.routed")(function* <Tools extends Record<string, Tool.Any>>(
414 toolkit: Toolkit.Toolkit<Tools>,
415 declarations: Handlers<Tools>,
416 checked: Toolkit.HandlersFrom<Tools>,
417) {
418 // Optional so toolkits that never leave this environment build without it.
419 const forwarding = yield* Effect.serviceOption(PeerForwarding.PeerForwarding);
420 const handlers: Record<string, unknown> = { ...checked };
421 for (const [name, declaration] of Object.entries<Declaration<unknown>>(declarations)) {
422 const tool = toolkit.tools[name];
423 if (tool === undefined || !takesLinkedEnvironment(tool)) continue;
424 const remote = remoteCheckOf(declaration);
425 const forwarded = forwardedOf(declaration) as
426 | Forwarded<object, unknown, Tool.Failure<Tools[keyof Tools]>, never>
427 | undefined;
428 if (remote === undefined)
429 return yield* Effect.die(
430 `${name} takes a linked environment, but it acts as its calling thread, which is here.`,
431 );
432 const here = handlers[name] as (
433 params: object,
434 ) => Effect.Effect<unknown, Tool.Failure<Tools[keyof Tools]>>;
435 handlers[name] = ({ environmentId, ...params }: { readonly environmentId?: EnvironmentId }) =>
436 Effect.gen(function* () {
437 const scope = yield* McpInvocationContext.McpInvocationContext;
438 const target = PeerForwarding.remoteTarget(scope, environmentId);
439 if (target === undefined) return yield* here(params);
440 if (scope.linkedEnvironments === false) return yield* PeerForwarding.linkedEnvironmentsOff;
441 yield* (remote as RemoteCheck<typeof params>)(params);
442 if (Option.isNone(forwarding)) return yield* linksUnavailable;
443 const result = yield* forwarding.value.call(scope, tool, target, {
444 ...params,
445 environmentId,
446 });
447 return forwarded === undefined ? result : yield* forwarded(params, target, result);
448 });
449 }
450 return handlers as Toolkit.HandlersFrom<Tools>;
451});
452
453export const toLayer = <Tools extends Record<string, Tool.Any>, EX = never, RX = never>(
454 toolkit: Toolkit.Toolkit<Tools>,
455 build: Handlers<Tools> | Effect.Effect<Handlers<Tools>, EX, RX>,
456): HandlersLayer<Tools, EX, Exclude<RX, Scope.Scope>> => {
457 // Checked as soon as they are given, so a forged declaration fails here.
458 const withDeclarations = (declarations: Handlers<Tools>) =>
459 routed(toolkit, declarations, checkedHandlers(declarations));
460 return handlersLayer(
461 toolkit.toLayer(
462 Effect.isEffect(build) ? Effect.flatMap(build, withDeclarations) : withDeclarations(build),
463 ),
464 );
465};