# Files changed · PR \#16684 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/16684/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [feat\(server\): agents use linked environments through any T3 tool](https://git.cafe/pingdotgg/t3code/pull/16684.md) Version: 31 Head commit: 641ee467aa5ceb9126f7c52e2b8b002bf258e0a4 Base commit: 197930788c33a086de927374a30e02da4c6336c4 Comparison base: 197930788c33a086de927374a30e02da4c6336c4 Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/server/src/mcp/McpHttpServer.ts Status: modified; +1; −1 ```diff @@ -1063,7 +1063,7 @@ EnvironmentHandlers.layer, ).pipe(Layer.provide(ThreadCommandExecutor.layer)); -const layerProjectRegistration = toolkitRegistration(ProjectToolkit, ProjectHandlers.layer); +export const layerProjectRegistration = toolkitRegistration(ProjectToolkit, ProjectHandlers.layer); export const layerAttachmentToolkit = toolkitRegistration( AttachmentToolkit, ``` ## apps/server/src/mcp/McpToolAccess.test.ts Status: modified; +113; −0 ```diff @@ -1,18 +1,22 @@ import { expect, it } from "@effect/vitest"; import { EnvironmentId, + OrchestratorMcpEnvironmentTarget, OrchestratorMcpFailure, ProviderInstanceId, ProviderInteractionMode, RuntimeMode, ThreadId, } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; import { McpSchema, McpServer, Tool, Toolkit } from "effect/ai"; import { OrchestratorProjectionError } from "../orchestration-v2/Orchestrator.ts"; +import * as PeerForwarding from "../peer/PeerForwarding.ts"; import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; import * as McpHttpServer from "./McpHttpServer.ts"; import * as McpInvocationContext from "./McpInvocationContext.ts"; @@ -340,3 +344,112 @@ ), ), ); + +const box = EnvironmentId.make("environment-box"); +const here = EnvironmentId.make("environment"); +const linkedProbe = { + ...probe, + parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, + threadId: Schema.optional(ThreadId), + runtimeMode: Schema.optional(RuntimeMode), + interactionMode: Schema.optional(ProviderInteractionMode), + }), +}; +const LinkedProbeToolkit = Toolkit.make( + Tool.make("reads", linkedProbe), + Tool.make("writes", linkedProbe), + Tool.make("writes_with_capability", linkedProbe), + Tool.make("writes_threads", linkedProbe), + Tool.make("starts_threads", linkedProbe), + Tool.make("writes_environment", linkedProbe), +); +/** Runs here, and tells whether `environmentId` reached the handler. */ +const ranHere =

(params: P) => + Effect.succeed({ ran: "environmentId" in params ? "saw environmentId" : "here" }); + +const linkedProbeServer = McpHttpServer.toolkitRegistration( + LinkedProbeToolkit, + McpToolAccess.toLayer(LinkedProbeToolkit, { + reads: McpToolAccess.reads(ranHere), + writes: McpToolAccess.writes(ranHere), + writes_with_capability: McpToolAccess.writesWithCapability(ranHere), + writes_threads: McpToolAccess.writesThreads((input) => [input.threadId], ranHere), + starts_threads: McpToolAccess.startsThreads((input) => input, ranHere), + writes_environment: McpToolAccess.writesEnvironment(ranHere), + }), +).pipe( + Layer.provideMerge(McpServer.McpServer.layer), + Layer.provideMerge( + Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadShell: (threadId) => Effect.succeed(shells.get(threadId) ?? null), + }), + ), + // Stands in for the linked environment, answering with where the call went. + Layer.provide( + Layer.mock(PeerForwarding.PeerForwarding)({ + call: (_scope, tool, environmentId) => + Effect.succeed({ ran: `${tool.name} in ${environmentId}` } as never), + }), + ), +); + +it.effect.each([ + ["reads", supervised, { environmentId: box }, "reads in environment-box"], + ["reads", supervised, { environmentId: here }, "here"], + ["reads", supervised, {}, "here"], + + // The caller is checked here before anything goes. + ["writes", readOnlyClient, { environmentId: box }, "capability_denied"], + ["writes", ended, { environmentId: box }, "parent_not_active"], + ["writes", supervised, { environmentId: box }, "writes in environment-box"], + ["writes_with_capability", readOnlyClient, { environmentId: box }, "capability_denied"], + [ + "starts_threads", + plan, + { environmentId: box, interactionMode: "default" }, + "interaction_mode_escalation_denied", + ], + ["starts_threads", plan, { environmentId: box }, "starts_threads in environment-box"], + ["writes_environment", supervised, { environmentId: box }, "capability_denied"], + [ + "writes_environment", + fullAccess, + { environmentId: box }, + "writes_environment in environment-box", + ], + ["writes_environment", fullAccess, { environmentId: here }, "here"], + + // A thread id names a thread where the call runs, and is checked there. + [ + "writes_threads", + supervised, + { environmentId: box, threadId: fullAccessThreadId }, + "writes_threads in environment-box", + ], + [ + "writes_threads", + supervised, + { environmentId: here, threadId: fullAccessThreadId }, + "runtime_mode_escalation_denied", + ], +] as const)("a linked %s from %o with %o: %s", ([tool, scope, args, expected]) => + call(tool, scope, args).pipe( + Effect.map((outcome) => expect(outcome).toBe(expected)), + Effect.provide(linkedProbeServer), + ), +); + +it.effect("refuses a linked environment to a tool that acts as its calling thread", () => + Effect.gen(function* () { + const CallerToolkit = Toolkit.make(Tool.make("acts_as_caller", linkedProbe)); + const built = yield* Layer.build( + McpToolAccess.HandlersLayer.layer( + McpToolAccess.toLayer(CallerToolkit, { + acts_as_caller: McpToolAccess.actsAsCaller(ranHere), + }), + ), + ).pipe(Effect.scoped, Effect.exit); + expect(Exit.isFailure(built) && Cause.hasDies(built.cause)).toBe(true); + }), +); ``` ## apps/server/src/mcp/McpToolAccess.ts Status: modified; +155; −55 ```diff @@ -1,10 +1,14 @@ import { + type EnvironmentId, + OrchestratorMcpEnvironmentTarget, OrchestratorMcpFailure, type ProviderInteractionMode, type RuntimeMode, type ThreadId, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Predicate from "effect/Predicate"; import * as Ref from "effect/Ref"; import * as Struct from "effect/Struct"; import type * as Layer from "effect/Layer"; @@ -15,6 +19,7 @@ DispatchModeLimit, type DispatchModeRefusal, } from "../orchestration-v2/DispatchModeLimit.ts"; +import * as PeerForwarding from "../peer/PeerForwarding.ts"; import * as McpInvocationContext from "./McpInvocationContext.ts"; import { resolveInteractionMode, resolveRuntimeMode } from "./OrchestratorMcpService.ts"; import { @@ -41,8 +46,19 @@ let declare: ( handle: (params: P) => Effect.Effect, + remote: RemoteCheck

| undefined, ) => Declaration<(params: P) => Effect.Effect>; +/** + * What a call bound for a linked environment checks here before it goes: the + * caller, not the thread or project it names, which live there and are + * checked there. `undefined` for tools that act as the calling thread itself. + */ +type RemoteCheck

= ( + params: P, +) => Effect.Effect>; +let remoteCheckOf: (declaration: Declaration) => RemoteCheck | undefined; + type CheckedHandlerOf = D extends Declaration ? Handler : never; interface CheckedHandler extends Struct.Lambda { @@ -62,13 +78,20 @@ */ export class Declaration { readonly #handle: Handler; + readonly #remote: RemoteCheck | undefined; + private constructor( + token: typeof builtHere, + handle: Handler, + remote: RemoteCheck | undefined, - private constructor(token: typeof builtHere, handle: Handler) { + ) { refuseOutsideConstruction(token); this.#handle = handle; + this.#remote = remote; } static { - declare = (handle) => new Declaration(builtHere, handle); + declare = (handle, remote) => new Declaration(builtHere, handle, remote); checkedHandler = Struct.lambda((declaration) => declaration.#handle); + remoteCheckOf = (declaration) => declaration.#remote; } } @@ -101,60 +124,73 @@ ); export const reads = (handle: (params: P) => Effect.Effect) => + declare( + (params: P) => handle(params), + () => Effect.void, + ); - declare((params: P) => handle(params)); export const readsAsCaller = (handle: (params: P) => Effect.Effect) => + declare((params: P) => requireThreadCaller.pipe(Effect.flatMap(() => handle(params))), undefined); - declare((params: P) => requireThreadCaller.pipe(Effect.flatMap(() => handle(params)))); export const actsAsCaller = (handle: (params: P) => Effect.Effect) => - declare((params: P) => - requireThreadCaller.pipe( - Effect.flatMap(() => writingCaller), - Effect.flatMap(() => handle(params)), + declare( - ), + (params: P) => + requireThreadCaller.pipe( + Effect.flatMap(() => writingCaller), + Effect.flatMap(() => handle(params)), + ), + undefined, ); export const writes = (handle: (params: P) => Effect.Effect) => + declare( + (params: P) => orchestratingCaller.pipe(Effect.flatMap(() => handle(params))), + () => orchestratingCaller, - declare((params: P) => orchestratingCaller.pipe(Effect.flatMap(() => handle(params)))); + ); /** * Unlike `writes` it does not need thread control; the handler checks its * capability, as the capability toolkits do. */ export const writesWithCapability = (handle: (params: P) => Effect.Effect) => + declare( + (params: P) => writingCaller.pipe(Effect.flatMap(() => handle(params))), - declare((params: P) => writingCaller.pipe(Effect.flatMap(() => handle(params)))); + () => writingCaller, + ); export const writesThreads = ( threads: (params: P) => ReadonlyArray, handle: (params: P) => Effect.Effect, ) => - declare((params: P) => - Effect.gen(function* () { - const caller = yield* writingCaller; - for (const threadId of threads(params)) { - if (threadId === undefined || threadId === caller.scope.thread?.threadId) continue; - const target = yield* caller.threads - .getThreadShell(threadId) - .pipe(Effect.mapError(unavailable)); - if (target !== null && target.deletedAt === null) { - yield* assertTargetWithinLimits(caller.limits, target); + declare( + (params: P) => + Effect.gen(function* () { + const caller = yield* writingCaller; + for (const threadId of threads(params)) { + if (threadId === undefined || threadId === caller.scope.thread?.threadId) continue; + const target = yield* caller.threads + .getThreadShell(threadId) + .pipe(Effect.mapError(unavailable)); + if (target !== null && target.deletedAt === null) { + yield* assertTargetWithinLimits(caller.limits, target); + } } - } - // The target's user can raise its modes after the check above; the + // The target's user can raise its modes after the check above; the + // orchestrator checks again under the thread's lock and records its + // refusal here, since handlers wrap dispatch errors their own way. - // orchestrator checks again under the thread's lock and records its - // refusal here, since handlers wrap dispatch errors their own way. - const refused = yield* Ref.make(undefined); - return yield* handle(params).pipe( - Effect.provideService(DispatchModeLimit, { ...caller.limits, refused }), - Effect.catch((error) => - Effect.flatMap(Ref.get(refused), (refusal) => - Effect.fail( - refusal === undefined ? error : escalationDenied(refusal), + const refused = yield* Ref.make(undefined); + return yield* handle(params).pipe( + Effect.provideService(DispatchModeLimit, { ...caller.limits, refused }), + Effect.catch((error) => + Effect.flatMap(Ref.get(refused), (refusal) => + Effect.fail( + refusal === undefined ? error : escalationDenied(refusal), + ), ), ), - ), - ); + ); + }), - }), + () => writingCaller, ); const escalationDenied = (refusal: DispatchModeRefusal) => @@ -177,21 +213,23 @@ readonly interactionMode?: ProviderInteractionMode | undefined; }, handle: (params: P, modes: StartedModes) => Effect.Effect, +) => { + const started = Effect.fn("McpToolAccess.started")(function* (params: P) { + const { limits } = yield* writingCaller; + const requested = modes(params); + return { + runtimeMode: yield* resolveRuntimeMode(limits.runtimeMode, requested.runtimeMode), + interactionMode: yield* resolveInteractionMode( + limits.interactionMode, + requested.interactionMode, + ), + } satisfies StartedModes; + }); + return declare( -) => - declare((params: P) => + (params: P) => Effect.flatMap(started(params), (modes) => handle(params, modes)), + started, - Effect.gen(function* () { - const { limits } = yield* writingCaller; - const requested = modes(params); - const started: StartedModes = { - runtimeMode: yield* resolveRuntimeMode(limits.runtimeMode, requested.runtimeMode), - interactionMode: yield* resolveInteractionMode( - limits.interactionMode, - requested.interactionMode, - ), - }; - return yield* handle(params, started); - }), ); +}; const fullAccessRequired = "Changing projects or environment settings needs a live full-access/default calling thread or a full-access client."; @@ -209,7 +247,10 @@ const check = orchestratingCaller.pipe( Effect.tap((caller) => assertFullAccess(caller, fullAccessRequired)), ); + return declare( + (params: P) => check.pipe(Effect.flatMap(() => handle(params, check))), + () => check, - return declare((params: P) => check.pipe(Effect.flatMap(() => handle(params, check)))); + ); }; type CheckServices = Effect.Services; @@ -246,14 +287,73 @@ const checkedHandlers = (declarations: Declarations): Handlers => Struct.map(declarations, checkedHandler); +/** + * A tool runs in a linked environment when its parameters take this target + * as `environmentId`. Other tools may name an environment for their own + * reasons, such as the one to link or to move a thread to. + */ +const takesLinkedEnvironment = (tool: Tool.Any) => { + const parameters: unknown = tool.parametersSchema; + return ( + Predicate.hasProperty(parameters, "fields") && + Predicate.hasProperty(parameters.fields, "environmentId") && + parameters.fields.environmentId === OrchestratorMcpEnvironmentTarget + ); +}; + +const linksUnavailable = new OrchestratorMcpFailure({ + code: "orchestration_error", + message: "This environment cannot reach linked environments.", +}); + +/** + * Every tool that takes a linked environment runs there the same way: its + * declaration checks the caller here, and the same call goes to the other + * environment, which checks the rest. The handler here never sees + * `environmentId`, so it cannot leak into what it writes. + */ +const routed = Effect.fn("McpToolAccess.routed")(function* >( + toolkit: Toolkit.Toolkit, + declarations: Handlers, + checked: Toolkit.HandlersFrom, +) { + // Optional so toolkits that never leave this environment build without it. + const forwarding = yield* Effect.serviceOption(PeerForwarding.PeerForwarding); + const handlers: Record = { ...checked }; + for (const [name, declaration] of Object.entries>(declarations)) { + const tool = toolkit.tools[name]; + if (tool === undefined || !takesLinkedEnvironment(tool)) continue; + const remote = remoteCheckOf(declaration); + if (remote === undefined) + return yield* Effect.die( + `${name} takes a linked environment, but it acts as its calling thread, which is here.`, + ); + const here = handlers[name] as ( + params: object, + ) => Effect.Effect>; + handlers[name] = ({ environmentId, ...params }: { readonly environmentId?: EnvironmentId }) => + Effect.gen(function* () { + const scope = yield* McpInvocationContext.McpInvocationContext; + const target = PeerForwarding.remoteTarget(scope, environmentId); + if (target === undefined) return yield* here(params); + yield* (remote as RemoteCheck)(params); + if (Option.isNone(forwarding)) return yield* linksUnavailable; + return yield* forwarding.value.call(scope, tool, target, { ...params, environmentId }); + }); + } + return handlers as Toolkit.HandlersFrom; +}); + export const toLayer = , EX = never, RX = never>( toolkit: Toolkit.Toolkit, build: Handlers | Effect.Effect, EX, RX>, +): HandlersLayer> => { + // Checked as soon as they are given, so a forged declaration fails here. + const withDeclarations = (declarations: Handlers) => + routed(toolkit, declarations, checkedHandlers(declarations)); + return handlersLayer( -): HandlersLayer> => - handlersLayer( toolkit.toLayer( - Effect.isEffect(build) - ? Effect.map(build, checkedHandlers>) - : checkedHandlers(build), + Effect.isEffect(build) ? Effect.flatMap(build, withDeclarations) : withDeclarations(build), ), ); +}; ``` ## apps/server/src/mcp/OrchestratorMcpService.ts Status: modified; +39; −0 ```diff @@ -2893,6 +2893,45 @@ yield* resolveRuntimeMode(limits.runtimeMode, target.thread.runtimeMode); yield* resolveInteractionMode(limits.interactionMode, target.thread.interactionMode); const key = yield* requestKey(input.clientRequestId); + // A peer from before `stop` drops it when decoding, so it interrupts + // the run instead, as this call did then. + if (input.stop === true) { + const commandId = stableCommandId({ scope, requestKey: key, operation: "thread-stop" }); + // The cause goes to the log; the caller learns only that the stop failed. + const stopFailed = (error: unknown) => + Effect.logWarning("orchestrator-mcp.thread-stop-failed", { + threadId: input.threadId, + error, + }).pipe( + Effect.andThen( + Effect.fail( + failure("orchestration_error", `Unable to stop thread ${input.threadId}.`), + ), + ), + ); + const stopped = yield* threadManagement + .dispatch({ + type: "thread.stop", + commandId, + threadId: input.threadId, + ...(input.reason === undefined ? {} : { reason: input.reason }), + }) + .pipe(Effect.catch(stopFailed)); + // A retry with the same clientRequestId repeats only the stops that failed. + yield* threadManagement + .stopDelegatedTasks({ threadId: input.threadId, commandId, reason: input.reason }) + .pipe(Effect.catch(stopFailed)); + // The run this stop interrupted, from its own events, so a retry + // reports the same run even if another has started since. + const interrupted = stopped.storedEvents.find( + (stored) => stored.event.type === "run.updated", + )?.event.runId; + return { + threadId: input.threadId, + runId: interrupted ?? null, + status: interrupted === undefined ? "no_active_run" : "interrupt_requested", + } satisfies OrchestratorMcpThreadInterruptResult; + } const result = yield* threadManagement .interruptThread({ projectId: target.thread.projectId, ``` ## apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts Status: modified; +2; −0 ```diff @@ -46,6 +46,7 @@ import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; import * as McpHttpServer from "./McpHttpServer.ts"; import * as McpInvocationContext from "./McpInvocationContext.ts"; +import * as PeerForwarding from "../peer/PeerForwarding.ts"; const layerOrchestratorToolkit = McpHttpServer.layerOrchestratorToolkit.pipe( Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})), @@ -220,6 +221,7 @@ Layer.provide(layerProviderRegistry), Layer.provide(ServerSettings.layerTest()), Layer.provide(layerUnusedScheduledTaskStub), + Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(ProjectService.ProjectService)({})), Layer.provideMerge( SecretRequests.layer.pipe( ``` ## apps/server/src/mcp/toolkits/core.test.ts Status: modified; +6; −0 ```diff @@ -51,6 +51,7 @@ import { dispatchFailure } from "../threadAccess.ts"; import * as EnvironmentHandlers from "./environment/handlers.ts"; import * as AttachmentHandlers from "./attachment/handlers.ts"; +import * as PeerForwarding from "../../peer/PeerForwarding.ts"; const layerOrchestratorToolkit = McpHttpServer.layerOrchestratorToolkit.pipe( Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})), @@ -71,6 +72,7 @@ const layerThreadToolkit = McpHttpServer.layerThreadToolkit.pipe( Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})), Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})), + Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), ); const threadId = ThreadId.make("mcp-core-thread"); @@ -542,6 +544,7 @@ Layer.provide(Layer.mock(ProviderRegistry.ProviderRegistry)({})), Layer.provide(Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({})), Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})), + Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(ProjectService.ProjectService)({})), Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})), ), @@ -597,6 +600,7 @@ ), Layer.provide(Layer.mock(ProjectService.ProjectService)({})), Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})), + Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), ), ), ), @@ -681,6 +685,7 @@ Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})), Layer.provide(Layer.mock(ProjectService.ProjectService)({})), Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})), + Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), ), ), ), @@ -771,6 +776,7 @@ Layer.provide(Layer.mock(ProviderRegistry.ProviderRegistry)({})), Layer.provide(Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({})), Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})), + Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(ProjectService.ProjectService)({})), Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})), ), ``` ## apps/server/src/mcp/toolkits/environment/tools.ts Status: modified; +3; −0 ```diff @@ -2,6 +2,7 @@ BackgroundActivityProfile, BackgroundActivityProfileSelection, ExecutionEnvironmentDescriptor, + OrchestratorMcpEnvironmentTarget, OrchestratorMcpFailure, ServerSettings, ServerSettingsPatch, @@ -41,6 +42,7 @@ ...shared, description: "Read this server's identity and selected environment preferences. Provider/model availability is exposed by orchestrator_capabilities. Writing instructions are limited to 4,000 characters.", + parameters: Schema.Struct({ environmentId: OrchestratorMcpEnvironmentTarget }), success: Schema.Struct({ environmentId: ExecutionEnvironmentDescriptor.fields.environmentId, label: Schema.String, @@ -56,6 +58,7 @@ description: "Update selected environment-wide preferences through normal settings persistence and notifications. Requires a live full-access/default calling thread. Omitted fields are preserved; empty customInstructions clears them.", parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, defaultThreadEnvMode: ServerSettingsPatch.fields.defaultThreadEnvMode, newWorktreesStartFromOrigin: ServerSettingsPatch.fields.newWorktreesStartFromOrigin, enableProviderUpdateChecks: ServerSettingsPatch.fields.enableProviderUpdateChecks, ``` ## apps/server/src/mcp/toolkits/orchestrator/handlers.ts Status: modified; +8; −0 ```diff @@ -2,6 +2,7 @@ import { OrchestratorToolkit } from "./tools.ts"; import * as Effect from "effect/Effect"; +import * as PeerForwarding from "../../../peer/PeerForwarding.ts"; import { readCaller } from "../../threadAccess.ts"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; import * as McpToolAccess from "../../McpToolAccess.ts"; @@ -140,6 +141,13 @@ return yield* service.waitForThread(scope, input); }), ), + t3_environment_links: McpToolAccess.reads(() => + Effect.gen(function* () { + const scope = yield* McpInvocationContext.McpInvocationContext; + const forwarding = yield* PeerForwarding.PeerForwarding; + return yield* forwarding.links(scope); + }), + ), t3_thread_interrupt: McpToolAccess.writesThreads( (input) => [input.threadId], (input) => ``` ## apps/server/src/mcp/toolkits/orchestrator/tools.test.ts Status: modified; +13; −0 ```diff @@ -1,4 +1,6 @@ import { expect, it } from "@effect/vitest"; +import * as Exit from "effect/Exit"; +import * as Schema from "effect/Schema"; import { Tool } from "effect/ai"; import { OrchestratorToolkit } from "./tools.ts"; @@ -14,3 +16,14 @@ expect(schema.properties[name]?.description, name).toContain(maximum); } }); + +it("refuses a t3_thread_update whose fields do not fit its action", () => { + const decode = Schema.decodeUnknownExit( + OrchestratorToolkit.tools.t3_thread_update.parametersSchema, + ); + expect(Exit.isFailure(decode({ action: "rename" }))).toBe(true); + expect(Exit.isFailure(decode({ action: "regenerate_title", title: "New" }))).toBe(true); + expect(Exit.isSuccess(decode({ action: "rename", title: "New", environmentId: undefined }))).toBe( + true, + ); +}); ``` ## apps/server/src/mcp/toolkits/orchestrator/tools.ts Status: modified; +35; −8 ```diff @@ -2,12 +2,14 @@ OrchestratorMcpCapabilitiesInput, OrchestratorMcpCapabilitiesResult, OrchestrationV2StopWorkflowInput, + OrchestratorMcpEnvironmentLinksResult, OrchestratorMcpCreateThreadsInput, OrchestratorMcpCreateThreadsResult, OrchestratorMcpDelegateTaskInput, OrchestratorMcpDelegateTaskResult, OrchestratorMcpDeleteScheduledTaskInput, OrchestratorMcpDeleteScheduledTaskResult, + OrchestratorMcpEnvironmentTarget, OrchestratorMcpRequestSecretInput, OrchestratorMcpRequestSecretResult, OrchestratorMcpFailure, @@ -34,8 +36,10 @@ } from "@t3tools/contracts"; import { Tool, Toolkit } from "effect/ai"; import * as Schema from "effect/Schema"; +import * as Struct from "effect/Struct"; import * as ThreadManagementService from "../../../orchestration-v2/ThreadManagementService.ts"; +import * as PeerForwarding from "../../../peer/PeerForwarding.ts"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; import * as OrchestratorMcpService from "../../OrchestratorMcpService.ts"; import * as ThreadMetadataMcpService from "../../ThreadMetadataMcpService.ts"; @@ -45,6 +49,8 @@ ThreadManagementService.ThreadManagementService, OrchestratorMcpService.OrchestratorMcpService, ]; +/** Listing this environment's links. */ +const linkDependencies = [...dependencies, PeerForwarding.PeerForwarding]; const threadMetadataDependencies = [ McpInvocationContext.McpInvocationContext, ThreadManagementService.ThreadManagementService, @@ -53,7 +59,7 @@ const OrchestratorCapabilitiesTool = Tool.make("orchestrator_capabilities", { description: - "List the V2 provider instances, a short list of their models from the same live catalog as the composer, inherited runtime settings, and app-owned orchestration features available to this caller. The list has the user's favorite and custom models, your own model, every current Codex, Claude, Muse and Grok model, and the first 8 current models of other providers. Other legacy models are left out. omittedModels counts models left out; pass search, such as 'sonnet 4.6', to find them. delegate_task accepts any catalog model, listed or not. For a separate top-level thread in a new or existing worktree, use t3_thread_launch with workspaceStrategy.", + "List the V2 provider instances, a short list of their models from the same live catalog as the composer, inherited runtime settings, and app-owned orchestration features available to this caller. The list has the user's favorite and custom models, your own model, every current Codex, Claude, Muse and Grok model, and the first 8 current models of other providers. Other legacy models are left out. omittedModels counts models left out; pass search, such as 'sonnet 4.6', to find them. delegate_task accepts any catalog model, listed or not. For a separate top-level thread in a new or existing worktree, use t3_thread_launch with workspaceStrategy. Pass environmentId to list a linked environment's providers instead.", parameters: OrchestratorMcpCapabilitiesInput, success: OrchestratorMcpCapabilitiesResult, failure: OrchestratorMcpFailure, @@ -107,7 +113,10 @@ const WorkflowStopTool = Tool.make("t3_workflow_stop", { description: "Stop one running Claude workflow in a T3 thread. Use the workflow coordinator's subagentId returned by t3_thread_read with view='activity'. Stops the whole workflow, including its phases and agents. Does not stop the parent thread or other workflows. Individual workflow phases and agents cannot be stopped separately.", + parameters: Schema.Struct({ - parameters: OrchestrationV2StopWorkflowInput, + environmentId: OrchestratorMcpEnvironmentTarget, + ...OrchestrationV2StopWorkflowInput.fields, + }), success: Schema.Struct({}), failure: OrchestratorMcpFailure, failureMode: "return", @@ -194,7 +203,7 @@ const ThreadListTool = Tool.make("t3_thread_list", { description: - "List T3 threads in a project, newest first. Omit projectId for the calling thread's project. Filter by durable run status, title, or settled state (settled=true lists threads the user or auto-settlement moved out of the active list), or snoozed state, and paginate with the returned cursor. A snoozed thread wakes early only when it asks for something or fails after snoozing; completed runs and background wakes keep it snoozed. pullRequests lists every pull request linked to each thread; linkedPullRequest is a legacy single link and can be null while pullRequests still has entries. To link a thread for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread's current title.", + "List T3 threads in a project, newest first. Omit projectId for the calling thread's project. Pass environmentId to list a linked environment's threads; projectId is then required. Filter by durable run status, title, or settled state (settled=true lists threads the user or auto-settlement moved out of the active list), or snoozed state, and paginate with the returned cursor. A snoozed thread wakes early only when it asks for something or fails after snoozing; completed runs and background wakes keep it snoozed. pullRequests lists every pull request linked to each thread; linkedPullRequest is a legacy single link and can be null while pullRequests still has entries. To link a thread in this environment for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread's current title. Name a thread in a linked environment by title instead: its link would open here.", parameters: OrchestratorMcpThreadListInput, success: OrchestratorMcpThreadListResult, failure: OrchestratorMcpFailure, @@ -208,7 +217,7 @@ const ThreadReadTool = Tool.make("t3_thread_read", { description: + "Read durable state and a paginated timeline from any T3 thread in this environment, or in a linked one with environmentId. The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. Reading an untruncated terminal assistant result from this parent thread's direct app-owned child acknowledges that child's automatic completion delivery. Continue with afterPosition=nextPosition. Recover long item text with itemId and textOffset=nextTextOffset until nextTextOffset is null; offsets count UTF-16 code units. The thread also reports its snooze state and, in pullRequests, every linked pull request; linkedPullRequest is a legacy single link and can be null while pullRequests still has entries. To link a thread in this environment for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread's current title. Name a thread in a linked environment by title instead: its link would open here.", - "Read durable state and a paginated timeline from any T3 thread in this environment. The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. Reading an untruncated terminal assistant result from this parent thread's direct app-owned child acknowledges that child's automatic completion delivery. Continue with afterPosition=nextPosition. Recover long item text with itemId and textOffset=nextTextOffset until nextTextOffset is null; offsets count UTF-16 code units. The thread also reports its snooze state and, in pullRequests, every linked pull request; linkedPullRequest is a legacy single link and can be null while pullRequests still has entries. To link a thread for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread's current title.", parameters: OrchestratorMcpThreadReadInput, success: OrchestratorMcpThreadReadResult, failure: OrchestratorMcpFailure, @@ -224,7 +233,11 @@ const ThreadUpdateTool = Tool.make("t3_thread_update", { description: "Update metadata for a thread. Omit threadId to update this thread. Use action='rename' with title, action='regenerate_title' with no extra field, action='link_pull_request' with pullRequest, or action='unlink_pull_request'. Workspace and branch changes are intentionally not supported. clientRequestId makes retries idempotent.", + // Keeps the input's action/field check, which rebuilding the struct would drop. + parameters: ThreadMetadataMcpUpdateInput.mapFields( + Struct.assign({ environmentId: OrchestratorMcpEnvironmentTarget }), + { unsafePreserveChecks: true }, + ), - parameters: ThreadMetadataMcpUpdateInput, success: ThreadMetadataMcpUpdateResult, failure: OrchestratorMcpFailure, failureMode: "return", @@ -236,7 +249,7 @@ const ThreadSendTool = Tool.make("t3_thread_send", { description: + "Send a message to any T3 thread in this environment, or in a linked one with environmentId. The target cannot have broader permission modes than the caller. Use delegate_task with the full review context and a new clientRequestId for independent review rounds. For a follow-up that needs the same child's conversation, send to its childThreadId. After the original task completes, a requested follow-up creates a fresh task in that child thread and returns taskId when it starts a turn or restarts a turn without its own task; its completion notifies the parent, and the original result stays unchanged. A steer that joins a running turn returns that turn's task, if it has one. mode='auto' starts an idle thread, steers a fully active turn, or queues behind a turn that is not yet steerable. Use queue for a separate follow-up turn, steer for an in-flight update, or restart to interrupt-and-restart the active turn. clientRequestId makes retries idempotent.", - "Send a message to any T3 thread in this environment. The target cannot have broader permission modes than the caller. Use delegate_task with the full review context and a new clientRequestId for independent review rounds. For a follow-up that needs the same child's conversation, send to its childThreadId. After the original task completes, a requested follow-up creates a fresh task in that child thread and returns taskId when it starts a turn or restarts a turn without its own task; its completion notifies the parent, and the original result stays unchanged. A steer that joins a running turn returns that turn's task, if it has one. mode='auto' starts an idle thread, steers a fully active turn, or queues behind a turn that is not yet steerable. Use queue for a separate follow-up turn, steer for an in-flight update, or restart to interrupt-and-restart the active turn. clientRequestId makes retries idempotent.", parameters: OrchestratorMcpThreadSendInput, success: OrchestratorMcpThreadSendResult, failure: OrchestratorMcpFailure, @@ -249,7 +262,7 @@ const ThreadWaitTool = Tool.make("t3_thread_wait", { description: - "Wait for a T3 thread run to reach a terminal durable state. Without runId, the latest run at call time is selected; an idle thread returns immediately. Timeout does not interrupt work, so call again or use t3_thread_read/list after timedOut=true. Waiting reports status only and does not acknowledge a delegated result.", + "Wait for a T3 thread run to reach a terminal durable state, here or in a linked environment with environmentId. Without runId, the latest run at call time is selected; an idle thread returns immediately. Timeout does not interrupt work, so call again or use t3_thread_read/list after timedOut=true. Waiting reports status only and does not acknowledge a delegated result.", parameters: OrchestratorMcpThreadWaitInput, success: OrchestratorMcpThreadWaitResult, failure: OrchestratorMcpFailure, @@ -263,7 +276,7 @@ const ThreadInterruptTool = Tool.make("t3_thread_interrupt", { description: + "Request interruption of a running turn in any T3 thread in this environment, or in a linked one with environmentId. Without runId, the newest interruptible run is selected. Terminal runs and threads without an active turn return without another side effect. stop=true stops the thread as its Stop button does instead: it also holds queued messages, ends pull request watches and stops the tasks it delegated. clientRequestId makes retries idempotent.", - "Request interruption of a running turn in any T3 thread in this environment. Without runId, the newest interruptible run is selected. Terminal runs and threads without an active turn return without another side effect. clientRequestId makes retries idempotent.", parameters: OrchestratorMcpThreadInterruptInput, success: OrchestratorMcpThreadInterruptResult, failure: OrchestratorMcpFailure, @@ -273,6 +286,19 @@ .annotate(Tool.Title, "Interrupt a T3 thread") .annotate(Tool.Destructive, true); +const EnvironmentLinksTool = Tool.make("t3_environment_links", { + description: + "List the other T3 Code environments this one is linked to, such as the user's other machines, and whether each answers now. Pass one's environmentId to any tool that takes environmentId, such as t3_thread_launch, t3_thread_send, t3_pending_request_respond, t3_queue_list or t3_project_list, to act there; delegate_task takes it as target.environmentId. A linked environment lets this one's agents do at most its access there, and never more than the calling agent may do here. An expired link has to be linked again from a new pairing code.", + success: OrchestratorMcpEnvironmentLinksResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies: linkDependencies, +}) + .annotate(Tool.Title, "List linked environments") + .annotate(Tool.Readonly, true) + .annotate(Tool.Destructive, false) + .annotate(Tool.Idempotent, true); + export const OrchestratorToolkit = Toolkit.make( OrchestratorCapabilitiesTool, DelegateTaskTool, @@ -291,4 +317,5 @@ ThreadSendTool, ThreadWaitTool, ThreadInterruptTool, + EnvironmentLinksTool, ); ``` ## apps/server/src/mcp/toolkits/project/handlers.test.ts Status: modified; +2; −0 ```diff @@ -23,6 +23,7 @@ import * as ServerSettings from "../../../serverSettings.ts"; import * as Project from "../../../project/ProjectService.ts"; import * as ManagedProjectFolders from "../../../project/ManagedProjectFolders.ts"; +import * as PeerForwarding from "../../../peer/PeerForwarding.ts"; import * as GitVcsDriver from "../../../vcs/GitVcsDriver.ts"; import * as VcsProcess from "../../../vcs/VcsProcess.ts"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; @@ -121,6 +122,7 @@ : Option.none(), ), }), + Layer.mock(PeerForwarding.PeerForwarding)({}), Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/projects" }), NodeServices.layer, ).pipe( ``` ## apps/server/src/mcp/toolkits/project/handlers.ts Status: modified; +3; −1 ```diff @@ -18,6 +18,7 @@ import * as ServerSettings from "../../../serverSettings.ts"; import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { resolveRuntimeMode } from "../../OrchestratorMcpService.ts"; +import * as McpInvocationContext from "../../McpInvocationContext.ts"; import * as McpToolAccess from "../../McpToolAccess.ts"; import { newCommandId, readCaller, resolveProjectId, unavailable } from "../../threadAccess.ts"; import { ProjectToolkit } from "./tools.ts"; @@ -64,8 +65,9 @@ (input) => input, (input, { runtimeMode, interactionMode }) => Effect.gen(function* () { + const scope = yield* McpInvocationContext.McpInvocationContext; const context = yield* readCaller(); - const { caller, scope } = context; + const { caller } = context; // A retry with the same key replays the first launch: the ids derive // from the caller and the key, and the launch service replays a // command it already accepted. ``` ## apps/server/src/mcp/toolkits/project/tools.ts Status: modified; +26; −8 ```diff @@ -14,6 +14,7 @@ ProjectUpdatePayload, ProjectId, OrchestratorMcpClientRequestId, + OrchestratorMcpEnvironmentTarget, OrchestratorMcpFailure, SourceControlCloneRepositoryInput, SourceControlCloneRepositoryResult, @@ -46,8 +47,9 @@ const ProjectListTool = Tool.make("t3_project_list", { ...shared, description: + "List registered projects in this environment, or in a linked one with environmentId. Pages use the current project snapshot and may shift between calls.", - "List registered projects in this environment. Pages use the current project snapshot and may shift between calls.", parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, cursor: Schema.optional(NonNegativeInt), limit: Schema.optional(Schema.Int.check(Schema.isBetween({ minimum: 1, maximum: 100 }))), }), @@ -62,15 +64,19 @@ ...shared, description: "Read a registered project in this environment, including its workspace and saved scripts.", + parameters: Schema.Struct({ - parameters: Schema.Struct({ projectId: ProjectId }), + environmentId: OrchestratorMcpEnvironmentTarget, + projectId: ProjectId, + }), }) .annotate(Tool.Readonly, true) .annotate(Tool.Destructive, false); const ProjectCreateTool = Tool.make("t3_project_create", { ...shared, description: - "Register a project directory through the existing project service. Set createWorkspaceRootIfMissing to create a directory. Omit workspaceRoot to start a new project from just its title: the app makes a Git repository for it in its own projects folder, with a README, an icon, and a first commit (commitError says why a commit failed; the project exists either way). Each call creates a new request; an existing registered workspace is rejected. Clone separately with t3_project_clone when needed.", + "Register a project directory through the existing project service. Set createWorkspaceRootIfMissing to create a directory. Omit workspaceRoot to start a new project from just its title: the app makes a Git repository for it in its own projects folder, with a README, an icon, and a first commit (commitError says why a commit failed; the project exists either way). Each call creates a new request; an existing registered workspace is rejected. Clone separately with t3_project_clone when needed. Pass environmentId to create it in a linked environment; workspaceRoot and defaultModelSelection then name that environment's paths and providers.", parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, ...ProjectCreatePayload.fields, workspaceRoot: Schema.optional(ProjectCreatePayload.fields.workspaceRoot), }), @@ -81,19 +87,30 @@ ...shared, description: "Update a registered project's settings. Omitted fields are preserved. Uses the same project service as the app.", - parameters: Schema.Struct({ projectId: ProjectId, ...ProjectUpdatePayload.fields }), + parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, + projectId: ProjectId, + ...ProjectUpdatePayload.fields, + }), }).annotate(Tool.Destructive, true); const ProjectDeleteTool = Tool.make("t3_project_delete", { ...shared, description: "Delete a project using the existing project deletion lifecycle. Nonempty projects require force=true. This does not delete the repository directory or promise a deleted-thread count.", + parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, + projectId: ProjectId, + force: Schema.optionalKey(Schema.Boolean), - parameters: Schema.Struct({ projectId: ProjectId, force: Schema.optionalKey(Schema.Boolean) }), + }), }).annotate(Tool.Destructive, true); const ProjectCloneTool = Tool.make("t3_project_clone", { ...shared, description: + "Clone a repository using the app's source-control service. This only clones; register the returned cwd with t3_project_create. An existing destination is not adopted or removed on failure. Pass environmentId to clone in a linked environment, then register it there with t3_project_create and the same environmentId.", + parameters: Schema.Struct({ - "Clone a repository using the app's source-control service. This only clones; register the returned cwd with t3_project_create. An existing destination is not adopted or removed on failure.", - parameters: SourceControlCloneRepositoryInput, + environmentId: OrchestratorMcpEnvironmentTarget, + ...SourceControlCloneRepositoryInput.fields, + }), success: SourceControlCloneRepositoryResult, dependencies: [ ...shared.dependencies, @@ -105,8 +122,9 @@ const ThreadLaunchTool = Tool.make("t3_thread_launch", { ...shared, description: - 'Create an ordinary TOP-LEVEL thread with an explicit workspace binding before its agent starts. Use this when the user requests independent work, a new thread, or a PR stack in its own worktree; use delegate_task for child subagents. Set workspaceStrategy to {type:"worktree",baseRef:"parent-branch",branch:"new-branch",startFromOrigin:false} for a new worktree based on local commits, or {type:"existing_worktree",worktreePath:"/absolute/path",branch:"existing-branch"} to use an existing checkout. To fetch and start from the base branch\'s remote, set startFromOrigin:true. Omitted workspaceStrategy means the project root, NOT the caller\'s worktree. Omit projectId/modelSelection/modes to inherit those settings from the calling thread; a caller outside a T3 thread must pass projectId and gets the project\'s default model. Set scratch:true instead of projectId for a thread without a project: it runs in a fresh folder of its own, outside any repository. Put the task in message. Do not ask the agent to create its own worktree via shell: that does not update the thread binding. Pass clientRequestId to retry safely: the same key returns the thread the first call created instead of launching another. Retain threadId and use t3_thread_read/t3_thread_wait to follow preparation. To link a thread for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread\'s current title. Without a key, after errors or lost responses, inspect t3_thread_list before retrying. Attachments must be pending uploads. The new thread may not run with broader runtime or interaction modes than the caller: the calling T3 thread\'s own modes, or the permission mode an outside agent was approved with.', + 'Create an ordinary TOP-LEVEL thread with an explicit workspace binding before its agent starts. Use this when the user requests independent work, a new thread, or a PR stack in its own worktree; use delegate_task for child subagents. Set workspaceStrategy to {type:"worktree",baseRef:"parent-branch",branch:"new-branch",startFromOrigin:false} for a new worktree based on local commits, or {type:"existing_worktree",worktreePath:"/absolute/path",branch:"existing-branch"} to use an existing checkout. To fetch and start from the base branch\'s remote, set startFromOrigin:true. Omitted workspaceStrategy means the project root, NOT the caller\'s worktree. Omit projectId/modelSelection/modes to inherit those settings from the calling thread; a caller outside a T3 thread must pass projectId and gets the project\'s default model. Set scratch:true instead of projectId for a thread without a project: it runs in a fresh folder of its own, outside any repository. Put the task in message. Do not ask the agent to create its own worktree via shell: that does not update the thread binding. Pass clientRequestId to retry safely: the same key returns the thread the first call created instead of launching another. Retain threadId and use t3_thread_read/t3_thread_wait to follow preparation. To link a thread in this environment for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread\'s current title. Name a thread in a linked environment by title instead: its link would open here. Without a key, after errors or lost responses, inspect t3_thread_list before retrying. Attachments must be pending uploads. The new thread may not run with broader runtime or interaction modes than the caller: the calling T3 thread\'s own modes, or the permission mode an outside agent was approved with. Pass environmentId to launch in a linked environment; projectId and modelSelection then name that environment\'s project and providers (t3_project_list and orchestrator_capabilities with the same environmentId list them), and attachments are not supported.', parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, projectId: Schema.optional(ProjectId), scratch: Schema.optional( Schema.Boolean.annotate({ ``` ## apps/server/src/mcp/toolkits/pullRequests/tools.ts Status: modified; +8; −1 ```diff @@ -1,5 +1,6 @@ import { McpCapabilityUnavailableError, + OrchestratorMcpEnvironmentTarget, OrchestratorMcpFailure, PositiveInt, PullRequestReviewGuide, @@ -35,6 +36,7 @@ "Register every pull request you open for this thread, including each layer of a stack, right after creating it."; export const PullRequestTargetInput = Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, threadId: Schema.optional( ThreadId.annotate({ description: "Thread to act on. Omit for this thread.", @@ -300,6 +302,7 @@ const ListThreadPullRequestsTool = Tool.make("list_thread_pull_requests", { description: `List the pull requests linked to a thread (omit threadId for this thread) with their last known host state, and how they chain into stacks (bottom to top). ${REGISTER_EVERY_PR}`, parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, threadId: Schema.optional( ThreadId.annotate({ description: "Thread to list. Omit for this thread." }), ), @@ -379,6 +382,7 @@ description: "Write review guides for up to 20 pull requests, the tour a maintainer reads in a pull request's Guide tab. Each run reads its change in a temporary checkout and takes minutes, so this returns once the runs are under way; read a guide with read_pull_request_guide to see when it is written. A guide for the same head that is already running is joined, not restarted; a pull request whose guide is current is written again. Each pull request is read through the project in this environment whose checkout has that repository. Fails while review guides are off on this server.", parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, pullRequests: Schema.Array(GuideTarget).check(Schema.isMinLength(1), Schema.isMaxLength(20)), }), success: WritePullRequestGuidesResult, @@ -394,7 +398,10 @@ const ReadPullRequestGuideTool = Tool.make("read_pull_request_guide", { description: "Read a pull request's stored review guide and its run: run.Running while one is being written, run.Failed with the reason when the last one failed, guide null when none was written. stale is true after a push the guide has not caught up with. Never starts a run.", + parameters: Schema.Struct({ - parameters: GuideTarget, + environmentId: OrchestratorMcpEnvironmentTarget, + ...GuideTarget.fields, + }), success: ReadPullRequestGuideResult, failure: PullRequestToolError, dependencies: guideDependencies, ``` ## apps/server/src/mcp/toolkits/thread/tools.ts Status: modified; +33; −6 ```diff @@ -1,4 +1,5 @@ import { + OrchestratorMcpEnvironmentTarget, ScheduledTaskId, ScheduledTask, OrchestrationSearchThreadsInput, @@ -32,6 +33,7 @@ description: "Pin, snooze, settle, archive, or mark a thread unread. Omit threadId for this thread. snooze requires snoozedUntil. Existing thread lifecycle rules apply. Settling this thread takes effect when your turn completes, returning settlesWhenTurnEnds=true; a turn that fails or is interrupted, or a queued message, leaves it active.", parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, threadId: Schema.optional(ThreadId), action: Schema.Literals([ "pin", @@ -61,7 +63,11 @@ .annotate(Tool.Title, "Organize a thread") .annotate(Tool.Destructive, true); +const queueTarget = { -const queueTarget = { threadId: Schema.optional(ThreadId), queuedRunId: RunId }; + environmentId: OrchestratorMcpEnvironmentTarget, + threadId: Schema.optional(ThreadId), + queuedRunId: RunId, +}; const commandTool = { success: OrchestrationV2DispatchCommandResult, failure: OrchestratorMcpFailure, @@ -82,6 +88,7 @@ description: "List queued messages in delivery order. Results are a live offset page; use t3_thread_read for full thread history.", parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, threadId: Schema.optional(ThreadId), cursor: Schema.optional(NonNegativeInt), limit: Schema.optional(Schema.Int.check(Schema.isBetween({ minimum: 1, maximum: 100 }))), @@ -127,7 +134,11 @@ parameters: Schema.Struct({ ...queueTarget, targetRunId: RunId }), }).annotate(Tool.Destructive, true); -const requestTarget = { threadId: Schema.optional(ThreadId), requestId: RuntimeRequestId }; +const requestTarget = { + environmentId: OrchestratorMcpEnvironmentTarget, + threadId: Schema.optional(ThreadId), + requestId: RuntimeRequestId, +}; const question = Schema.Struct({ id: Schema.String, header: Schema.String, @@ -152,7 +163,10 @@ ...commandTool, description: "List pending user questions in a thread. Omit threadId for this thread. Approval requests are not included.", + parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, + threadId: Schema.optional(ThreadId), - parameters: Schema.Struct({ threadId: Schema.optional(ThreadId) }), + }), success: Schema.Struct({ requestIds: Schema.Array(RuntimeRequestId) }), }) .annotate(Tool.Readonly, true) @@ -179,7 +193,10 @@ ...commandTool, description: "Read a thread's provider/model selection and modes. Omit threadId for this thread. orchestrator_capabilities lists available providers and models.", + parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, + threadId: Schema.optional(ThreadId), - parameters: Schema.Struct({ threadId: Schema.optional(ThreadId) }), + }), success: Schema.Struct({ threadId: ThreadId, modelSelection: ModelSelection, @@ -194,6 +211,7 @@ description: "Set a thread's provider, model and options with the existing selection command. Omit threadId for this thread. This does not change permission modes. Use orchestrator_capabilities to choose a selection.", parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, threadId: Schema.optional(ThreadId), modelSelection: ModelSelection, }), @@ -205,6 +223,7 @@ description: "Fork a thread from a stable run or checkpoint using the existing fork command. Omit threadId to fork this thread. The fork inherits the source configuration. Acceptance does not mean a provider turn has completed.", parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, threadId: Schema.optional(ThreadId), sourcePoint: OrchestrationV2ThreadForkSourcePoint, title: Schema.optional(TrimmedNonEmptyString), @@ -216,6 +235,7 @@ description: "Merge context from a thread back to a related thread in the same project. Omit sourceThreadId to merge from this thread. Existing lineage and transfer rules apply.", parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, sourceThreadId: Schema.optional(ThreadId), targetThreadId: ThreadId, sourcePoint: OrchestrationV2ThreadForkSourcePoint, @@ -225,7 +245,10 @@ const ThreadTransfersTool = Tool.make("t3_thread_transfers", { ...commandTool, description: "Read context transfer status for a thread. Omit threadId for this thread.", + parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, + threadId: Schema.optional(ThreadId), + }), - parameters: Schema.Struct({ threadId: Schema.optional(ThreadId) }), success: Schema.Struct({ transfers: Schema.Array( Schema.Struct({ @@ -245,6 +268,7 @@ description: "Search active thread titles and content with the app's existing bounded search. Matches are limited to one project (projectId, else the calling thread's project) out of the global top matches, so this may return fewer than limit. A caller outside a T3 thread that omits projectId searches every project. No pagination or exhaustive-result guarantee.", parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, ...OrchestrationSearchThreadsInput.fields, projectId: Schema.optional(ProjectId), }), @@ -258,7 +282,10 @@ ...commandTool, description: "Run a scheduled task now through the existing scheduler. Requires a full-access/default caller. Returns the thread this run dispatched to, or null if dispatch failed. Each call is a new manual run; completion means dispatch/bookkeeping completed, not that the provider turn finished.", + parameters: Schema.Struct({ - parameters: Schema.Struct({ taskId: ScheduledTaskId }), + environmentId: OrchestratorMcpEnvironmentTarget, + taskId: ScheduledTaskId, + }), success: Schema.Struct({ taskId: ScheduledTaskId, threadId: ScheduledTask.fields.threadId, ``` ## apps/server/src/mcp/toolkits/worktree/tools.ts Status: modified; +2; −0 ```diff @@ -1,5 +1,6 @@ import { WorktreeMcpFailure, + OrchestratorMcpEnvironmentTarget, OrchestratorMcpFailure, VcsListRefsInput, VcsListRefsResult, @@ -62,6 +63,7 @@ description: "List branch refs and their associated checkout paths for a thread's workspace (omit threadId for this thread) using the app's ref inventory. Detached worktrees without a branch are not included. Use t3_worktree_status for the thread binding and t3_worktree_handoff to create a new worktree.", parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, threadId: Schema.optional(ThreadId), query: VcsListRefsInput.fields.query, cursor: VcsListRefsInput.fields.cursor, ``` ## apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts Status: modified; +1; −0 ```diff @@ -1452,6 +1452,7 @@ "mcp__t3-code__t3_thread_search", "mcp__t3-code__t3_preview_list", "mcp__t3-code__t3_environment_read", + "mcp__t3-code__t3_environment_links", "mcp__t3-code__t3_queue_list", "mcp__t3-code__t3_queue_read", "mcp__t3-code__html_preview", ``` ## apps/server/src/peer/PeerForwarding.test.ts Status: added; +616; −0 ```diff @@ -1,0 +1,616 @@ +import { NodeHttpServer } from "@effect/platform-node"; +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { + EnvironmentId, + type OrchestrationV2Run, + type SourceControlCloneRepositoryInput, + ProjectId, + ProviderInstanceId, + RunId, + ThreadId, +} from "@t3tools/contracts"; +import { expect, it } from "@effect/vitest"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Layer from "effect/Layer"; +import * as Ref from "effect/Ref"; +import { McpSchema, McpServer } from "effect/ai"; + +import * as ServerConfig from "../config.ts"; +import * as ThreadLaunch from "../orchestration-v2/ThreadLaunchService.ts"; +import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; +import * as ThreadSearch from "../orchestration-v2/ThreadSearch.ts"; +import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts"; +import * as SourceControlRepositoryService from "../sourceControl/SourceControlRepositoryService.ts"; +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; +import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts"; +import * as ProjectService from "../project/ProjectService.ts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; +import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; +import * as SecretRequests from "../secrets/SecretRequests.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as McpHttpServer from "../mcp/McpHttpServer.ts"; +import * as McpInvocationContext from "../mcp/McpInvocationContext.ts"; +import { liveThreadShell } from "../mcp/McpToolAccess.testkit.ts"; +import * as PeerForwarding from "./PeerForwarding.ts"; +import * as PeerLinks from "./PeerLinks.ts"; +import { descriptorOf, layerLinkingEnvironment, linkTo, servePeer } from "./PeerLinks.testkit.ts"; + +const laptop = descriptorOf("environment-laptop", "Laptop"); +const box = descriptorOf("environment-box", "Box"); +const boxProject = ProjectId.make("project:box"); +const boxThread = ThreadId.make("thread:box-auto"); +const boxRun = RunId.make("run:box-auto"); + +/** What B's thread service was asked to do. */ +interface Seen { + readonly sends: ReadonlyArray; + readonly waits: ReadonlyArray; + readonly launches: ReadonlyArray; + readonly projects: ReadonlyArray; + readonly clones: ReadonlyArray; + /** Commands dispatched and delegated-task stops, as ` `. */ + readonly stops: ReadonlyArray; +} +const nothingSeen: Seen = { + sends: [], + waits: [], + launches: [], + projects: [], + clones: [], + stops: [], +}; + +/** + * B's threads: one `auto` thread in `boxProject`, whose run finishes on the + * third wait. Every send and wait is recorded. + */ +const boxThreads = (seen: Ref.Ref) => { + const shell = { ...liveThreadShell(boxThread, { runtimeMode: "auto" }), projectId: boxProject }; + const run = (status: OrchestrationV2Run["status"]) => + ({ id: boxRun, status }) as unknown as OrchestrationV2Run; + const projection = { thread: shell, runs: [run("running")], runtimeRequests: [] } as never; + return Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadShell: (threadId) => Effect.succeed(threadId === boxThread ? shell : null), + getProjectThreadRecords: () => Effect.succeed(projection), + getThreadRecords: () => Effect.succeed(projection), + listProjectThreads: () => Effect.succeed([shell] as never), + sendToThread: (input) => + Ref.update(seen, (current) => ({ ...current, sends: [...current.sends, input] })).pipe( + Effect.as({ run: run("running"), delivery: "started" } as never), + ), + dispatch: (command) => + Ref.update(seen, (current) => ({ + ...current, + stops: [ + ...current.stops, + `${command.type} ${"threadId" in command ? command.threadId : ""}`, + ], + })).pipe( + // The stop interrupts the thread's running run. + Effect.as({ + sequence: 0, + storedEvents: [ + { + sequence: 1, + commandId: command.commandId, + event: { type: "run.updated", runId: boxRun }, + }, + ], + } as never), + ), + stopDelegatedTasks: (input) => + Ref.update(seen, (current) => ({ + ...current, + stops: [...current.stops, `delegated-tasks.stop ${input.threadId}`], + })), + waitForThread: (input) => + Ref.updateAndGet(seen, (current) => ({ ...current, waits: [...current.waits, input] })).pipe( + Effect.map(({ waits }) => ({ + threadId: input.threadId, + run: run(waits.length < 3 ? "running" : "completed"), + timedOut: waits.length < 3, + })), + ), + }); +}; + +/** B launches whatever it is asked to, and records it. */ +const boxLaunches = (seen: Ref.Ref) => + Layer.mock(ThreadLaunch.ThreadLaunchService)({ + launch: (input) => + Ref.update(seen, (current) => ({ ...current, launches: [...current.launches, input] })).pipe( + Effect.as({ + threadId: input.threadId, + projection: { + thread: { + id: input.threadId, + projectId: input.projectId, + modelSelection: input.modelSelection, + title: input.title, + }, + runs: [], + }, + resumed: false, + } as unknown as ThreadLaunch.ThreadLaunchResult), + ), + }); + +/** B clones and registers whatever it is asked to, and records it. */ +const boxProjects = (seen: Ref.Ref) => + Layer.merge( + Layer.mock(ProjectService.ProjectService)({ + getById: () => Effect.succeed(Option.none()), + create: (input) => + Ref.update(seen, (current) => ({ + ...current, + projects: [...current.projects, input], + })).pipe( + Effect.as({ + id: input.projectId, + title: input.title, + workspaceRoot: input.workspaceRoot, + defaultModelSelection: null, + scripts: [], + createdAt: "2026-10-18T00:00:00.000Z", + updatedAt: "2026-10-18T00:00:00.000Z", + deletedAt: null, + }), + ), + }), + Layer.mock(SourceControlRepositoryService.SourceControlRepositoryService)({ + cloneRepository: (input) => + Ref.update(seen, (current) => ({ ...current, clones: [...current.clones, input] })).pipe( + Effect.as({ + cwd: input.destinationPath, + remoteUrl: "https://github.com/pingdotgg/t3code.git", + repository: null, + }), + ), + }), + ); + +/** The orchestrator and project toolkits on B's real `/mcp`, behind its real OAuth. */ +const serveBox = (seen: Ref.Ref) => + servePeer( + box, + Layer.mergeAll( + McpHttpServer.layerOrchestratorToolkit, + McpHttpServer.layerProjectRegistration, + McpHttpServer.layerThreadToolkit, + ).pipe( + Layer.provide(NodeCrypto.layer), + Layer.provide(boxThreads(seen)), + Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})), + Layer.provide(boxLaunches(seen)), + Layer.provide( + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/projects" }), + ), + Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})), + Layer.provide(boxProjects(seen)), + Layer.provide( + ServerConfig.layerTest(process.cwd(), { prefix: "t3-peer-forwarding-box-" }).pipe( + Layer.provide(NodeServices.layer), + ), + ), + Layer.provide(NodeServices.layer), + Layer.provide(Layer.mock(ProviderRegistry.ProviderRegistry)({})), + Layer.provide(Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({})), + Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})), + Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), + ), + ); + +const mcpClient = McpSchema.McpServerClient.of({ + clientId: 1, + protocolVersion: "2025-06-18", + clientCapabilities: {}, + clientInfo: { name: "peer-forwarding-test", version: "1" }, + initializePayload: { + protocolVersion: "2025-06-18", + capabilities: {}, + clientInfo: { name: "peer-forwarding-test", version: "1" }, + }, + getClient: Effect.die("unused"), +}); + +/** + * Environment A: its own orchestrator toolkit with forwarding, and a thread + * caller whose modes the test picks. Its local threads never matter here. + */ +const makeLaptop = Effect.gen(function* () { + const linking = yield* layerLinkingEnvironment(laptop).pipe(Layer.build); + const layerHere = Layer.mergeAll( + McpHttpServer.layerOrchestratorToolkit, + McpHttpServer.layerProjectRegistration, + McpHttpServer.layerThreadToolkit, + ).pipe( + Layer.provideMerge(McpServer.McpServer.layer), + Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})), + Layer.provide(Layer.mock(ThreadLaunch.ThreadLaunchService)({})), + Layer.provide( + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/projects" }), + ), + Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})), + Layer.provide(Layer.mock(SourceControlRepositoryService.SourceControlRepositoryService)({})), + Layer.provide( + ServerConfig.layerTest(process.cwd(), { prefix: "t3-peer-forwarding-laptop-" }).pipe( + Layer.provide(NodeServices.layer), + ), + ), + Layer.provide(NodeServices.layer), + Layer.provide(PeerForwarding.layer), + Layer.provide(Layer.succeedContext(linking)), + Layer.provide(NodeCrypto.layer), + Layer.provideMerge( + Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadShell: (threadId) => + Effect.succeed( + liveThreadShell( + threadId, + threadId === ThreadId.make("thread:laptop-plan") + ? { runtimeMode: "full-access", interactionMode: "plan" } + : { runtimeMode: "full-access" }, + ), + ), + getProjectThreadRecords: ({ threadId }) => + Effect.succeed({ + thread: liveThreadShell(threadId, { runtimeMode: "full-access" }), + } as never), + }), + ), + Layer.provide(Layer.mock(ProviderRegistry.ProviderRegistry)({})), + Layer.provide(Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({})), + Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})), + Layer.provide(Layer.mock(ProjectService.ProjectService)({})), + Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})), + Layer.provide(ServerSettings.layerTest()), + Layer.fresh, + ); + const here = yield* Layer.build(layerHere); + const server = Context.get(here, McpServer.McpServer); + const scopeOf = ( + caller: string, + capabilities: ReadonlySet = new Set(["orchestration"]), + ): McpInvocationContext.McpInvocationScope => ({ + environmentId: laptop.environmentId, + requestNamespace: `provider-session:${caller}`, + thread: { + threadId: ThreadId.make(caller), + providerSessionId: `provider-session:${caller}`, + providerInstanceId: ProviderInstanceId.make("codex"), + }, + client: undefined, + capabilities, + issuedAt: 1, + }); + const call = ( + caller: string, + name: string, + args: Record, + capabilities?: ReadonlySet, + ) => + server + .callTool({ name, arguments: args }) + .pipe( + Effect.provideService( + McpInvocationContext.McpInvocationContext, + scopeOf(caller, capabilities), + ), + Effect.provideService(McpSchema.McpServerClient, mcpClient), + ); + return { links: Context.get(linking, PeerLinks.PeerLinks), call }; +}); + +const failureOf = (result: McpSchema.CallToolResult) => { + const text = result.content[0]; + return result.isError === true && text?.type === "text" ? JSON.parse(text.text) : undefined; +}; + +it.effect("an agent here works on a linked environment's threads within its own modes", () => + Effect.scoped( + Effect.gen(function* () { + const seen = yield* Ref.make(nothingSeen); + const b = yield* serveBox(seen); + const a = yield* makeLaptop; + yield* linkTo(a.links, b, "auto"); + + const listed = yield* a.call("thread:laptop-full", "t3_environment_links", {}); + expect(listed.structuredContent).toMatchObject({ + environmentId: laptop.environmentId, + links: [ + { environmentId: box.environmentId, label: "Box", status: "reachable", access: "auto" }, + ], + }); + + const threads = yield* a.call("thread:laptop-full", "t3_thread_list", { + environmentId: box.environmentId, + projectId: boxProject, + }); + expect(threads.isError).toBe(false); + expect(threads.structuredContent).toMatchObject({ + projectId: boxProject, + threads: [{ threadId: boxThread, runtimeMode: "auto" }], + }); + + // A full-access agent here is held to the link's `auto` there, which + // the box's `auto` thread is within. + const sent = yield* a.call("thread:laptop-full", "t3_thread_send", { + environmentId: box.environmentId, + threadId: boxThread, + message: "Carry on there.", + clientRequestId: "send-1", + }); + expect(sent.isError).toBe(false); + expect(sent.structuredContent).toMatchObject({ threadId: boxThread, delivery: "started" }); + + // A plan-mode agent here may not steer that thread there: the box + // refuses it, because the call carries the agent's own modes. + const planned = yield* a.call("thread:laptop-plan", "t3_thread_send", { + environmentId: box.environmentId, + threadId: boxThread, + message: "Change course.", + }); + expect(failureOf(planned)).toMatchObject({ code: "interaction_mode_escalation_denied" }); + expect((yield* Ref.get(seen)).sends).toHaveLength(1); + + // stop=true stops the thread there as its Stop button does: the whole + // thread and the tasks it delegated, not just the running turn. + const stopped = yield* a.call("thread:laptop-full", "t3_thread_interrupt", { + environmentId: box.environmentId, + threadId: boxThread, + stop: true, + clientRequestId: "stop-1", + }); + expect(stopped.structuredContent).toMatchObject({ + threadId: boxThread, + runId: boxRun, + status: "interrupt_requested", + }); + expect((yield* Ref.get(seen)).stops).toEqual([ + `thread.stop ${boxThread}`, + `delegated-tasks.stop ${boxThread}`, + ]); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect("keeps a forwarded wait short and its retries scoped to the caller", () => + Effect.scoped( + Effect.gen(function* () { + const seen = yield* Ref.make(nothingSeen); + const b = yield* serveBox(seen); + const a = yield* makeLaptop; + yield* linkTo(a.links, b, "auto"); + + const waited = yield* a.call("thread:laptop-full", "t3_thread_wait", { + environmentId: box.environmentId, + threadId: boxThread, + timeoutMs: 600_000, + }); + expect(waited.structuredContent).toEqual({ + threadId: boxThread, + runId: boxRun, + status: "completed", + timedOut: false, + }); + // Three pieces of at most 50 s each, the later ones pinned to the run + // the first one found. + expect( + (yield* Ref.get(seen)).waits.map(({ timeoutMs, runId }) => [timeoutMs, runId]), + ).toEqual([ + [PeerForwarding.FORWARDED_WAIT_CHUNK_MS, undefined], + [PeerForwarding.FORWARDED_WAIT_CHUNK_MS, boxRun], + [PeerForwarding.FORWARDED_WAIT_CHUNK_MS, boxRun], + ]); + + const send = (caller: string, clientRequestId: string) => + a.call(caller, "t3_thread_send", { + environmentId: box.environmentId, + threadId: boxThread, + message: "Once.", + clientRequestId, + }); + yield* send("thread:laptop-full", "same-key"); + yield* send("thread:laptop-full", "same-key"); + yield* send("thread:laptop-other", "same-key"); + const [first, retried, other] = (yield* Ref.get(seen)).sends.map((input) => input.commandId); + // The box sees every caller here as the one link, so the key is scoped + // to its caller before it leaves: a retry matches, another agent's same + // key does not. + expect(retried).toBe(first); + expect(other).not.toBe(first); + expect(first).not.toContain("same-key"); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect("launches in a linked environment with the caller's modes, retry-safe", () => + Effect.scoped( + Effect.gen(function* () { + const seen = yield* Ref.make(nothingSeen); + const b = yield* serveBox(seen); + const a = yield* makeLaptop; + yield* linkTo(a.links, b, "full-access"); + const codex = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" }; + const launch = (caller: string, args: Record = {}) => + a.call(caller, "t3_thread_launch", { + environmentId: box.environmentId, + projectId: boxProject, + title: "Run the suite there", + modelSelection: codex, + message: "Run the suite.", + clientRequestId: "launch-1", + ...args, + }); + + const first = yield* launch("thread:laptop-plan"); + expect(first.isError).toBe(false); + expect(first.structuredContent).toMatchObject({ + projectId: boxProject, + modelSelection: codex, + }); + const retried = yield* launch("thread:laptop-plan"); + const [launched, relaunched] = (yield* Ref.get(seen)).launches; + // The thread there runs with the caller's own modes here, below the + // link's full access, and a retry asks for the same thread. + expect(launched).toMatchObject({ runtimeMode: "full-access", interactionMode: "plan" }); + expect(relaunched?.threadId).toBe(launched?.threadId); + expect(retried.structuredContent).toEqual(first.structuredContent); + + // An escalation past the caller's own modes is refused here. + const escalated = yield* launch("thread:laptop-plan", { interactionMode: "default" }); + expect(failureOf(escalated)).toMatchObject({ code: "interaction_mode_escalation_denied" }); + // A pending upload lives here, so it cannot go along. + yield* Ref.set(b.bearers, []); + const withUpload = yield* launch("thread:laptop-full", { + clientRequestId: "launch-2", + attachments: [ + { type: "image", id: "upload-1", name: "a.png", mimeType: "image/png", sizeBytes: 1 }, + ], + }); + expect(failureOf(withUpload)).toMatchObject({ code: "invalid_request" }); + expect(yield* Ref.get(b.bearers)).toEqual([]); + expect((yield* Ref.get(seen)).launches).toHaveLength(2); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect("brings a repository to a linked environment, only through full access", () => + Effect.scoped( + Effect.gen(function* () { + const seen = yield* Ref.make(nothingSeen); + const b = yield* serveBox(seen); + const a = yield* makeLaptop; + const clone = { repository: "pingdotgg/t3code", destinationPath: "/home/box/t3code" }; + const register = { title: "t3code", workspaceRoot: "/home/box/t3code" }; + + yield* linkTo(a.links, b, "full-access"); + const cloned = yield* a.call("thread:laptop-full", "t3_project_clone", { + environmentId: box.environmentId, + ...clone, + }); + expect(cloned.structuredContent).toMatchObject({ cwd: clone.destinationPath }); + const created = yield* a.call("thread:laptop-full", "t3_project_create", { + environmentId: box.environmentId, + ...register, + }); + expect(created.structuredContent).toMatchObject(register); + expect((yield* Ref.get(seen)).clones).toEqual([clone]); + expect((yield* Ref.get(seen)).projects).toMatchObject([register]); + + // A plan-mode agent here may not, and nothing leaves. + yield* Ref.set(b.bearers, []); + const planned = yield* a.call("thread:laptop-plan", "t3_project_create", { + environmentId: box.environmentId, + ...register, + }); + expect(failureOf(planned)).toMatchObject({ code: "capability_denied" }); + expect(yield* Ref.get(b.bearers)).toEqual([]); + + // Nor may a full-access agent through a link with less: the box holds it. + yield* linkTo(a.links, b, "auto"); + for (const [name, args] of [ + ["t3_project_clone", clone], + ["t3_project_create", register], + ] as const) { + const refused = yield* a.call("thread:laptop-full", name, { + environmentId: box.environmentId, + ...args, + }); + expect(failureOf(refused), name).toMatchObject({ code: "capability_denied" }); + } + expect((yield* Ref.get(seen)).clones).toHaveLength(1); + expect((yield* Ref.get(seen)).projects).toHaveLength(1); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect("runs any tool that takes a linked environment there", () => + Effect.scoped( + Effect.gen(function* () { + const seen = yield* Ref.make(nothingSeen); + const b = yield* serveBox(seen); + const a = yield* makeLaptop; + yield* linkTo(a.links, b, "auto"); + + const there = yield* a.call("thread:laptop-full", "t3_thread_configuration", { + environmentId: box.environmentId, + threadId: boxThread, + }); + expect(there.structuredContent).toMatchObject({ threadId: boxThread, runtimeMode: "auto" }); + const archived = yield* a.call("thread:laptop-full", "t3_thread_organize", { + environmentId: box.environmentId, + threadId: boxThread, + action: "archive", + }); + expect(archived.isError).toBe(false); + expect((yield* Ref.get(seen)).stops).toEqual([`thread.archive ${boxThread}`]); + + // This environment's own id keeps the call here. + const own = yield* a.call("thread:laptop-full", "t3_thread_configuration", { + environmentId: laptop.environmentId, + }); + expect(own.structuredContent).toMatchObject({ + threadId: "thread:laptop-full", + runtimeMode: "full-access", + }); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect("refuses what a linked environment cannot take before anything leaves", () => + Effect.scoped( + Effect.gen(function* () { + const seen = yield* Ref.make(nothingSeen); + const b = yield* serveBox(seen); + const a = yield* makeLaptop; + yield* linkTo(a.links, b, "auto"); + yield* Ref.set(b.bearers, []); + + const unknown = yield* a.call("thread:laptop-full", "t3_thread_read", { + environmentId: EnvironmentId.make("environment-elsewhere"), + threadId: boxThread, + }); + expect(failureOf(unknown)).toMatchObject({ code: "invalid_request" }); + expect(yield* Ref.get(b.bearers)).toEqual([]); + + // The box sees only the link's credential, so one that may not control + // threads here may not reach them there either. + const previewOnly = new Set(["preview"]); + for (const [name, args] of [ + ["t3_environment_links", {}], + ["t3_project_list", { environmentId: box.environmentId }], + ["t3_thread_list", { environmentId: box.environmentId, projectId: boxProject }], + ["t3_thread_read", { environmentId: box.environmentId, threadId: boxThread }], + ["t3_thread_wait", { environmentId: box.environmentId, threadId: boxThread }], + [ + "t3_thread_send", + { environmentId: box.environmentId, threadId: boxThread, message: "Carry on there." }, + ], + [ + "t3_thread_organize", + { environmentId: box.environmentId, threadId: boxThread, action: "archive" }, + ], + [ + "t3_thread_launch", + { + environmentId: box.environmentId, + projectId: boxProject, + title: "Start there", + modelSelection: { instanceId: "codex", model: "gpt-5" }, + message: "Start there.", + }, + ], + ] as const) { + const refused = yield* a.call("thread:laptop-full", name, args, previewOnly); + expect(failureOf(refused), name).toMatchObject({ code: "capability_denied" }); + } + expect(yield* Ref.get(b.bearers)).toEqual([]); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); ``` ## apps/server/src/peer/PeerForwarding.ts Status: added; +192; −0 ```diff @@ -1,0 +1,192 @@ +import { + type EnvironmentId, + OrchestratorMcpFailure, + type OrchestratorMcpEnvironmentLinksResult, + OrchestratorMcpThreadWaitInput, + OrchestratorMcpThreadWaitResult, +} from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Hex from "effect/encoding/Hex"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import type * as Tool from "effect/ai/Tool"; + +import * as McpInvocationContext from "../mcp/McpInvocationContext.ts"; +import { readCaller, unavailable } from "../mcp/threadAccess.ts"; +import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; +import * as PeerLinks from "./PeerLinks.ts"; +import * as PeerMcpClient from "./PeerMcpClient.ts"; + +/** + * A peer's `/mcp` holds a wait open for its whole timeout without sending + * anything, and T3 Connect's edge drops a request idle for about 100 s, so a + * forwarded wait goes out in pieces no longer than this. + */ +export const FORWARDED_WAIT_CHUNK_MS = 50_000; +const DEFAULT_WAIT_TIMEOUT_MS = 10 * 60 * 1_000; +const MAX_WAIT_TIMEOUT_MS = 60 * 60 * 1_000; + +type Scope = McpInvocationContext.McpInvocationScope; + +/** + * Runs T3 MCP tools in a linked environment for an agent here. The tool's own + * declaration checks the agent here first; the linked environment then holds + * the call to the link's access, narrowed to the agent's modes, which every + * forwarded call carries. + */ +export class PeerForwarding extends Context.Service< + PeerForwarding, + { + /** + * Runs `tool` in the linked environment `environmentId` with `params`. A + * `t3_thread_wait` goes in pieces short enough to survive any route. + */ + readonly call: ( + scope: Scope, + tool: T, + environmentId: EnvironmentId, + params: Tool.Parameters, + ) => Effect.Effect, OrchestratorMcpFailure>; + /** `t3_thread_wait` there, in pieces short enough to survive any route. */ + readonly waitForThread: ( + scope: Scope, + environmentId: EnvironmentId, + input: OrchestratorMcpThreadWaitInput, + ) => Effect.Effect; + /** The environments this one links to, as an agent sees them. */ + readonly links: ( + scope: Scope, + ) => Effect.Effect; + } +>()("t3/peer/PeerForwarding") {} + +/** The linked environment a call names, or nothing when it names this one. */ +export const remoteTarget = (scope: Scope, environmentId: EnvironmentId | undefined) => + environmentId === scope.environmentId ? undefined : environmentId; + +const make = Effect.gen(function* () { + const peers = yield* PeerMcpClient.PeerMcpClient; + const links = yield* PeerLinks.PeerLinks; + const crypto = yield* Crypto.Crypto; + const threads = yield* ThreadManagement.ThreadManagementService; + + /** + * The caller's own limits here, which travel with every call it forwards. + * The peer sees only the link's credential, so a credential that may not + * control threads here is refused here. + */ + const callerLimits = (scope: Scope) => + readCaller().pipe( + Effect.map((caller) => caller.limits), + Effect.provideService(McpInvocationContext.McpInvocationContext, scope), + Effect.provideService(ThreadManagement.ThreadManagementService, threads), + ); + + /** + * Every caller here reaches the peer as the one link session, so a retry + * key is scoped to its caller before it leaves: two agents here reusing a + * key never collide there, and one agent retrying gets its first result. + */ + const scopedRequestId = (namespace: string, clientRequestId: string) => + crypto.digest("SHA-256", new TextEncoder().encode(`${namespace}\n${clientRequestId}`)).pipe( + Effect.map((digest) => `fwd:${Hex.encode(digest)}`), + Effect.orDie, + ); + + /** Tool and contract schemas are plain data: they encode and decode without services. */ + const forward = Effect.fn("PeerForwarding.forward")(function* ( + scope: Scope, + toolName: string, + parameters: Schema.Top, + success: Schema.Top, + environmentId: EnvironmentId, + params: unknown, + ) { + const limits = yield* callerLimits(scope); + const encoded = (yield* Schema.encodeUnknownEffect(parameters as Schema.Encoder)( + params, + ).pipe(Effect.mapError(unavailable))) as Record; + const { environmentId: _target, ...args } = encoded; + // A pending upload lives here, so it cannot go with the call. + if (Array.isArray(args.attachments) && args.attachments.length > 0) { + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "A launch in a linked environment cannot carry attachments yet.", + }); + } + if (typeof args.clientRequestId === "string") { + args.clientRequestId = yield* scopedRequestId(scope.requestNamespace, args.clientRequestId); + } + return yield* peers.call({ + environmentId, + tool: toolName, + arguments: args, + limits, + success: success as Schema.Decoder, + }); + }); + + const waitForThread: PeerForwarding["Service"]["waitForThread"] = Effect.fn( + "PeerForwarding.waitForThread", + )(function* (scope, environmentId, input) { + let remaining = Math.min( + MAX_WAIT_TIMEOUT_MS, + Math.max(1, input.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS), + ); + let runId = input.runId; + while (true) { + const chunk = Math.min(remaining, FORWARDED_WAIT_CHUNK_MS); + const result = yield* forward( + scope, + "t3_thread_wait", + OrchestratorMcpThreadWaitInput, + OrchestratorMcpThreadWaitResult, + environmentId, + { threadId: input.threadId, ...(runId === undefined ? {} : { runId }), timeoutMs: chunk }, + ); + remaining -= chunk; + if (!result.timedOut || remaining <= 0) return result; + // Later pieces wait for the run the first one picked, not a newer one. + runId = result.runId ?? undefined; + } + }); + + const call: PeerForwarding["Service"]["call"] = (scope, tool, environmentId, params) => + tool.name === "t3_thread_wait" + ? (waitForThread( + scope, + environmentId, + params as OrchestratorMcpThreadWaitInput, + ) as Effect.Effect, OrchestratorMcpFailure>) + : forward(scope, tool.name, tool.parametersSchema, tool.successSchema, environmentId, params); + + const listLinks: PeerForwarding["Service"]["links"] = Effect.fn("PeerForwarding.listLinks")( + function* (scope) { + yield* callerLimits(scope); + const listed = yield* links.list.pipe( + Effect.mapError( + (error) => + new OrchestratorMcpFailure({ code: "orchestration_error", message: error.message }), + ), + ); + return { + environmentId: scope.environmentId, + links: listed.map((link) => ({ + environmentId: link.environmentId, + label: link.label, + status: link.status, + access: link.access, + expiresAt: DateTime.formatIso(link.expiresAt), + lastError: link.lastError, + })), + }; + }, + ); + + return PeerForwarding.of({ call, waitForThread, links: listLinks }); +}); + +export const layer = Layer.effect(PeerForwarding, make); ``` ## apps/server/src/peer/PeerLinks.test.ts Status: modified; +35; −160 ```diff @@ -1,11 +1,5 @@ import { NodeHttpServer } from "@effect/platform-node"; -import * as NodeServices from "@effect/platform-node/NodeServices"; -import { - EnvironmentHttpApi, +import { type ExecutionEnvironmentDescriptor, OrchestratorMcpFailure } from "@t3tools/contracts"; - EnvironmentId, - type ExecutionEnvironmentDescriptor, - OrchestratorMcpFailure, -} from "@t3tools/contracts"; import { expect, it } from "@effect/vitest"; import * as Context from "effect/Context"; import * as Duration from "effect/Duration"; @@ -19,46 +13,23 @@ import { FetchHttpClient, HttpClient, - 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 { + descriptorOf, + layerLinkingEnvironment, + servePeer, + type ServedPeer, +} from "./PeerLinks.testkit.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 }); @@ -78,126 +49,37 @@ }), ); -/** - * 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) => +const probeHandlers = 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.gen(function* () { - const descriptor = yield* Ref.make(initial); - const bearers = yield* Ref.make>([]); + Effect.fail( + new OrchestratorMcpFailure({ code: "thread_not_found", message: "No such thread." }), - const probes = yield* Ref.make(0); + ), + ), +}); + +const serveB = (descriptor: ExecutionEnvironmentDescriptor) => + servePeer(descriptor, McpHttpServer.toolkitRegistration(ProbeToolkit, probeHandlers)); - const layerEnvironment = Layer.succeed(ServerEnvironment.ServerEnvironment, { - getEnvironmentId: Ref.get(descriptor).pipe(Effect.map((current) => current.environmentId)), - getDescriptor: Ref.get(descriptor), - }); - const authContext = yield* EnvironmentAuth.layer.pipe( - Layer.provide(Sqlite.layerMemory), - Layer.provideMerge(ServerSecretStore.layer), - Layer.provideMerge(ServerEnvironment.layerIdentity), - Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "t3-peer-link-b-" })), - Layer.provideMerge(NodeServices.layer), - Layer.fresh, - Layer.build, - ); - // Every bearer B receives, on any route, so a test can show one was never sent. - const layerRecordBearers = HttpRouter.middleware( - (httpEffect) => - Effect.gen(function* () { - const request = yield* HttpServerRequest.HttpServerRequest; - if (request.url.startsWith("/.well-known/t3/environment")) { - yield* Ref.update(probes, (count) => count + 1); - } - const authorization = request.headers.authorization; - if (authorization !== undefined) { - yield* Ref.update(bearers, (seen) => [...seen, authorization]); - } - return yield* httpEffect; - }), - { global: true }, - ); - const layerRoutes = Layer.mergeAll( - HttpApiBuilder.layer(PeerTestApi).pipe( - Layer.provide(McpOAuthHttp.layer.pipe(Layer.provide(McpOAuth.layer))), - Layer.provide(ServerHttp.layerServerEnvironmentHttpApi), - Layer.provide(AuthHttp.layerAuthenticatedAuth), - ), - McpHttpServer.toolkitRegistration( - ProbeToolkit, - McpToolAccess.toLayer(ProbeToolkit, { - probe: McpToolAccess.reads(() => - McpInvocationContext.McpInvocationContext.pipe( - Effect.map((scope) => { - const modes = McpInvocationContext.clientModeCeiling(scope.client); - return { - caller: scope.client?.label ?? "none", - modes: `${modes.runtimeMode}/${modes.interactionMode}`, - }; - }), - ), - ), - refuse: McpToolAccess.reads(() => - Effect.fail( - new OrchestratorMcpFailure({ code: "thread_not_found", message: "No such thread." }), - ), - ), - }), - ).pipe( - Layer.provideMerge(McpHttpServer.layerMcpTransport), - Layer.provide( - Layer.mock(McpSessionRegistry.McpSessionRegistry)({ - resolve: () => Effect.succeed(undefined), - }), - ), - Layer.provide(McpOAuth.layerMcpClientAuthenticator), - ), - ).pipe( - Layer.provide(layerRecordBearers), - Layer.provide(layerEnvironment), - Layer.provide(Layer.succeedContext(authContext)), - ); - yield* HttpRouter.serve(layerRoutes, { disableListenLog: true, disableLogger: true }).pipe( - Layer.build, - ); - const address = (yield* HttpServer.HttpServer).address; - if (address._tag === "UnixPathAddress") return yield* Effect.die("expected a TCP address"); - return { - url: `http://127.0.0.1:${address.port}`, - auth: Context.get(authContext, EnvironmentAuth.EnvironmentAuth), - descriptor, - bearers, - /** How often its descriptor was asked for, as a probe of whether it answers. */ - probes, - }; - }); +/** Environment A, which links to B. */ -/** Environment A: its own database and secret store, reaching B over plain HTTP. */ const makeA = (descriptor: ExecutionEnvironmentDescriptor) => - Effect.gen(function* () { - const context = yield* PeerMcpClient.layer.pipe( - Layer.provideMerge(PeerLinks.layer), - Layer.provide( - Layer.mergeAll( - Sqlite.layerMemory, - ServerSecretStore.layer, - FetchHttpClient.layer, - Layer.succeed(ServerEnvironment.ServerEnvironment, { - getEnvironmentId: Effect.succeed(descriptor.environmentId), - getDescriptor: Effect.succeed(descriptor), - }), - ), - ), - Layer.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-peer-link-a-" })), - Layer.provide(NodeServices.layer), - Layer.fresh, - Layer.build, - ); + layerLinkingEnvironment(descriptor).pipe( - return { + Layer.build, + Effect.map((context) => ({ links: Context.get(context, PeerLinks.PeerLinks), peer: Context.get(context, PeerMcpClient.PeerMcpClient), - }; + })), + ); - }); const laptop = descriptorOf("environment-laptop", "Laptop"); const box = descriptorOf("environment-box", "Box"); @@ -214,14 +96,7 @@ success: Probe, }); -const linkedSessions = (b: Effect.Success>) => - b.auth - .listSessions() - .pipe( - Effect.map((sessions) => - sessions.filter((session) => session.client.label?.startsWith("T3 Code · ")), - ), +const linkedSessions = (b: ServedPeer) => b.linkedSessions; - ); it.effect("links with a pairing code, and the peer holds its agents to the limits they carry", () => Effect.scoped( ``` [Next page](https://git.cafe/pingdotgg/t3code/pull/16684/diffs?after=ImFwcHMvc2VydmVyL3NyYy9wZWVyL1BlZXJMaW5rcy50ZXN0LnRzIg&expectedVersion=31&format=markdown)