From 73415db0985c659da09b2e785aa08e372b3a7bab Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 16:20:06 +0000 Subject: [PATCH 1/3] feat(server): route agent work to linked environments --- apps/server/src/mcp/McpToolAccess.ts | 194 ++- .../src/mcp/toolkits/environment/handlers.ts | 10 + .../src/mcp/toolkits/environment/tools.ts | 16 +- .../mcp/toolkits/orchestrator/tools.test.ts | 13 + .../src/mcp/toolkits/orchestrator/tools.ts | 16 +- .../src/mcp/toolkits/project/handlers.ts | 21 +- apps/server/src/mcp/toolkits/project/tools.ts | 42 +- apps/server/src/mcp/toolkits/thread/tools.ts | 85 +- .../Adapters/ClaudeAdapterV2.ts | 1 + .../src/orchestration-v2/DispatchModeLimit.ts | 14 + .../ThreadLaunchService.test.ts | 140 ++ .../orchestration-v2/ThreadLaunchService.ts | 11 +- apps/server/src/peer/PeerForwarding.test.ts | 1182 +++++++++++++++++ apps/server/src/peer/PeerForwarding.ts | 180 +++ apps/server/src/peer/PeerLinks.ts | 15 +- apps/server/src/peer/PeerMcpClient.ts | 314 +++++ apps/server/src/server.ts | 3 + .../orchestrator-mcp-server.md | 5 +- packages/client-runtime/src/t3ToolSummary.ts | 3 + packages/contracts/src/environment.ts | 7 + packages/contracts/src/orchestratorMcp.ts | 29 +- packages/contracts/src/peerLink.ts | 2 +- packages/contracts/src/threadMetadataMcp.ts | 2 + packages/shared/src/t3McpToolPresentation.ts | 5 + 24 files changed, 2212 insertions(+), 98 deletions(-) create mode 100644 apps/server/src/peer/PeerForwarding.test.ts create mode 100644 apps/server/src/peer/PeerForwarding.ts create mode 100644 apps/server/src/peer/PeerMcpClient.ts diff --git a/apps/server/src/mcp/McpToolAccess.ts b/apps/server/src/mcp/McpToolAccess.ts index 5cc8dc0c38b8f..c98bfa33013f4 100644 --- a/apps/server/src/mcp/McpToolAccess.ts +++ b/apps/server/src/mcp/McpToolAccess.ts @@ -1,10 +1,15 @@ 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 PeerForwarding from "../peer/PeerForwarding.ts"; import * as Ref from "effect/Ref"; import * as Struct from "effect/Struct"; import type * as Layer from "effect/Layer"; @@ -41,8 +46,14 @@ const refuseOutsideConstruction = (token: symbol) => { let declare: ( handle: (params: P) => Effect.Effect, + remote?: RemoteCheck

, ) => Declaration<(params: P) => Effect.Effect>; +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,12 +73,19 @@ let handlersLayer: , EX, RX>( */ export class Declaration { readonly #handle: Handler; - private constructor(token: typeof builtHere, handle: Handler) { + readonly #remote: RemoteCheck | undefined; + private constructor( + token: typeof builtHere, + handle: Handler, + remote: RemoteCheck | undefined, + ) { refuseOutsideConstruction(token); this.#handle = handle; + this.#remote = remote; } static { - declare = (handle) => new Declaration(builtHere, handle); + declare = (handle, remote) => new Declaration(builtHere, handle, remote); + remoteCheckOf = (declaration) => declaration.#remote; checkedHandler = Struct.lambda((declaration) => declaration.#handle); } } @@ -101,7 +119,10 @@ const requireThreadCaller = McpInvocationContext.McpInvocationContext.pipe( ); export const reads = (handle: (params: P) => Effect.Effect) => - declare((params: P) => handle(params)); + declare( + (params: P) => handle(params), + () => Effect.void, + ); export const readsAsCaller = (handle: (params: P) => Effect.Effect) => declare((params: P) => requireThreadCaller.pipe(Effect.flatMap(() => handle(params)))); @@ -115,7 +136,10 @@ export const actsAsCaller = (handle: (params: P) => Effect.Effect(handle: (params: P) => Effect.Effect) => - declare((params: P) => orchestratingCaller.pipe(Effect.flatMap(() => handle(params)))); + declare( + (params: P) => orchestratingCaller.pipe(Effect.flatMap(() => handle(params))), + () => orchestratingCaller.pipe(Effect.asVoid), + ); /** * Unlike `writes` it does not need thread control; the handler checks its @@ -128,33 +152,35 @@ 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 - // 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), + // 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. + 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.pipe(Effect.asVoid), ); const escalationDenied = (refusal: DispatchModeRefusal) => @@ -178,19 +204,27 @@ export const startsThreads = ( }, handle: (params: P, modes: StartedModes) => Effect.Effect, ) => - declare((params: P) => - 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); - }), + declare( + (params: P) => + 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); + }), + (params: P) => + Effect.gen(function* () { + const { limits } = yield* writingCaller; + const requested = modes(params); + yield* resolveRuntimeMode(limits.runtimeMode, requested.runtimeMode); + yield* resolveInteractionMode(limits.interactionMode, requested.interactionMode); + }), ); const fullAccessRequired = @@ -209,7 +243,10 @@ export const writesEnvironment = ( const check = orchestratingCaller.pipe( Effect.tap((caller) => assertFullAccess(caller, fullAccessRequired)), ); - return declare((params: P) => check.pipe(Effect.flatMap(() => handle(params, check)))); + return declare( + (params: P) => check.pipe(Effect.flatMap(() => handle(params, check))), + () => check.pipe(Effect.asVoid), + ); }; type CheckServices = Effect.Services; @@ -246,14 +283,77 @@ export class HandlersLayer, EX = never, R 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 = >( + toolkit: Toolkit.Toolkit, + declarations: Handlers, + checked: Toolkit.HandlersFrom, +) => + Effect.gen(function* () { + // 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 = environmentId === scope.environmentId ? undefined : 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> => - handlersLayer( +): 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( toolkit.toLayer( - Effect.isEffect(build) - ? Effect.map(build, checkedHandlers>) - : checkedHandlers(build), + Effect.isEffect(build) ? Effect.flatMap(build, withDeclarations) : withDeclarations(build), ), ); +}; diff --git a/apps/server/src/mcp/toolkits/environment/handlers.ts b/apps/server/src/mcp/toolkits/environment/handlers.ts index fb1716885d6e8..9ab1713e4ae0b 100644 --- a/apps/server/src/mcp/toolkits/environment/handlers.ts +++ b/apps/server/src/mcp/toolkits/environment/handlers.ts @@ -1,4 +1,6 @@ import { OrchestratorMcpFailure, type ServerSettings } from "@t3tools/contracts"; +import * as PeerForwarding from "../../../peer/PeerForwarding.ts"; +import * as Option from "effect/Option"; import * as Effect from "effect/Effect"; import * as Environment from "../../../environment/ServerEnvironment.ts"; import * as ThreadCommandExecutor from "../../../orchestration-v2/ThreadCommandExecutor.ts"; @@ -41,6 +43,14 @@ const access = Effect.gen(function* () { return { ...context, descriptor, settings: yield* Settings.ServerSettingsService }; }); export const layer = McpToolAccess.toLayer(EnvironmentToolkit, { + t3_environment_links: McpToolAccess.reads(() => + Effect.gen(function* () { + const scope = yield* McpInvocationContext.McpInvocationContext; + const forwarding = yield* Effect.serviceOption(PeerForwarding.PeerForwarding); + if (Option.isNone(forwarding)) return yield* unavailable(); + return yield* forwarding.value.links(scope); + }), + ), t3_environment_read: McpToolAccess.reads(() => Effect.gen(function* () { const { descriptor, settings } = yield* access; diff --git a/apps/server/src/mcp/toolkits/environment/tools.ts b/apps/server/src/mcp/toolkits/environment/tools.ts index e441174900ce1..b3e25551f31b4 100644 --- a/apps/server/src/mcp/toolkits/environment/tools.ts +++ b/apps/server/src/mcp/toolkits/environment/tools.ts @@ -1,4 +1,5 @@ import { + OrchestratorMcpEnvironmentLinksResult, BackgroundActivityProfile, BackgroundActivityProfileSelection, ExecutionEnvironmentDescriptor, @@ -64,4 +65,17 @@ const EnvironmentPreferencesTool = Tool.make("t3_environment_preferences_update" }), success: Schema.Struct(PreferenceFields), }).annotate(Tool.Destructive, true); -export const EnvironmentToolkit = Toolkit.make(EnvironmentReadTool, EnvironmentPreferencesTool); +const EnvironmentLinksTool = Tool.make("t3_environment_links", { + ...shared, + description: + "List this environment's paired destinations without credentials. Reachable means the public descriptor answers with the expected identity, not that access is still authorized.", + success: OrchestratorMcpEnvironmentLinksResult, +}) + .annotate(Tool.Readonly, true) + .annotate(Tool.Destructive, false); + +export const EnvironmentToolkit = Toolkit.make( + EnvironmentReadTool, + EnvironmentPreferencesTool, + EnvironmentLinksTool, +); diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts index b8f735069d31c..41715e372b52f 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts @@ -1,5 +1,6 @@ import { assert, describe, it } from "@effect/vitest"; import { Tool } from "effect/ai"; +import * as Schema from "effect/Schema"; import { CreateThreadsTool, @@ -9,6 +10,8 @@ import { ThreadUpdateTool, } from "./tools.ts"; +const decodeThreadUpdate = Schema.decodeUnknownResult(ThreadUpdateTool.parametersSchema); + describe("orchestrator MCP tool guidance", () => { it("directs subagent requests to delegation instead of ordinary threads", () => { assert.include(DelegateTaskTool.description ?? "", "child agent/subagent"); @@ -80,6 +83,7 @@ describe("orchestrator MCP tool guidance", () => { assert.equal(schema.type, "object"); assert.hasAllKeys(schema.properties ?? {}, [ + "environmentId", "threadId", "action", "title", @@ -87,5 +91,14 @@ describe("orchestrator MCP tool guidance", () => { "clientRequestId", ]); assert.include(ThreadUpdateTool.description ?? "", "Workspace and branch changes"); + assert.deepEqual( + [ + { action: "rename" }, + { action: "link_pull_request" }, + { action: "rename", title: "Renamed locally" }, + { action: "rename", title: "Renamed remotely", environmentId: "destination" }, + ].map((input) => decodeThreadUpdate(input)._tag), + ["Failure", "Failure", "Success", "Success"], + ); }); }); diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index f0dabc0260e37..4b67069bd28f5 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -1,5 +1,6 @@ import { OrchestratorMcpCapabilitiesResult, + OrchestratorMcpEnvironmentTarget, OrchestrationV2StopWorkflowInput, OrchestratorMcpCreateThreadsInput, OrchestratorMcpCreateThreadsResult, @@ -52,7 +53,8 @@ const threadMetadataDependencies = [ const OrchestratorCapabilitiesTool = Tool.make("orchestrator_capabilities", { description: - "List the V2 provider instances and their current models from the same live catalog as the composer, including configured custom models, inherited runtime settings, and app-owned orchestration features available to this caller. For a separate top-level thread in a new or existing worktree, use t3_thread_launch with workspaceStrategy.", + "List the V2 provider instances and their current models locally or in a linked environment with environmentId, from the same live catalog as that environment's composer, including configured custom models, inherited runtime settings, and app-owned orchestration features available to this caller. For a separate top-level thread in a new or existing worktree, use t3_thread_launch with workspaceStrategy.", + parameters: Schema.Struct({ environmentId: OrchestratorMcpEnvironmentTarget }), success: OrchestratorMcpCapabilitiesResult, failure: OrchestratorMcpFailure, failureMode: "return", @@ -192,7 +194,7 @@ export const CreateThreadsTool = Tool.make("create_threads", { 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 when it asks for something, fails, or completes. 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 locally or in a linked environment with environmentId, newest first. Locally, 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 when it asks for something, fails, or completes. 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.", parameters: OrchestratorMcpThreadListInput, success: OrchestratorMcpThreadListResult, failure: OrchestratorMcpFailure, @@ -206,7 +208,7 @@ const ThreadListTool = Tool.make("t3_thread_list", { const ThreadReadTool = Tool.make("t3_thread_read", { description: - "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.", + "Read durable state and a paginated timeline from any T3 thread locally or in a linked environment 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 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, @@ -221,7 +223,7 @@ const ThreadReadTool = Tool.make("t3_thread_read", { export 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.", + "Update metadata for a thread locally or in a linked environment with environmentId. Locally, 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.", parameters: ThreadMetadataMcpUpdateInput, success: ThreadMetadataMcpUpdateResult, failure: OrchestratorMcpFailure, @@ -234,7 +236,7 @@ export const ThreadUpdateTool = Tool.make("t3_thread_update", { const ThreadSendTool = Tool.make("t3_thread_send", { description: - "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.", + "Send a message to any T3 thread locally or in a linked environment 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.", parameters: OrchestratorMcpThreadSendInput, success: OrchestratorMcpThreadSendResult, failure: OrchestratorMcpFailure, @@ -247,7 +249,7 @@ const ThreadSendTool = Tool.make("t3_thread_send", { 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 locally or in a linked environment with environmentId 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.", parameters: OrchestratorMcpThreadWaitInput, success: OrchestratorMcpThreadWaitResult, failure: OrchestratorMcpFailure, @@ -261,7 +263,7 @@ const ThreadWaitTool = Tool.make("t3_thread_wait", { const ThreadInterruptTool = Tool.make("t3_thread_interrupt", { description: - "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.", + "Request interruption of a running turn in any T3 thread locally or in a linked environment with environmentId. 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, diff --git a/apps/server/src/mcp/toolkits/project/handlers.ts b/apps/server/src/mcp/toolkits/project/handlers.ts index ca47192c98e4b..aa7038ba648d0 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.ts @@ -1,4 +1,10 @@ -import { MessageId, ThreadId, OrchestratorMcpFailure, ProjectId } from "@t3tools/contracts"; +import { + CommandId, + MessageId, + ThreadId, + OrchestratorMcpFailure, + ProjectId, +} from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Option from "effect/Option"; @@ -57,7 +63,18 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { Effect.gen(function* () { const context = yield* readCaller(); const { caller } = context; - const commandId = yield* newCommandId(); + const commandId = + input.clientRequestId === undefined + ? yield* newCommandId() + : CommandId.make( + [ + "command", + "mcp", + encodeURIComponent(context.scope.requestNamespace), + "launch", + encodeURIComponent(input.clientRequestId), + ].join(":"), + ); const threadId = ThreadId.make(commandId); const messageId = MessageId.make(commandId); const attachments = input.attachments ?? []; diff --git a/apps/server/src/mcp/toolkits/project/tools.ts b/apps/server/src/mcp/toolkits/project/tools.ts index e455c041c4c31..36460ae0ee972 100644 --- a/apps/server/src/mcp/toolkits/project/tools.ts +++ b/apps/server/src/mcp/toolkits/project/tools.ts @@ -1,6 +1,8 @@ import { McpAttachmentInput } from "../attachment/input.ts"; import { NonNegativeInt, + OrchestratorMcpEnvironmentTarget, + OrchestratorMcpClientRequestId, ModelSelection, TrimmedNonEmptyString, ThreadId, @@ -45,8 +47,9 @@ const shared = { const ProjectListTool = Tool.make("t3_project_list", { ...shared, description: - "List registered projects in this environment. Pages use the current project snapshot and may shift between calls.", + "List registered projects locally or in a linked environment with environmentId. 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 }))), }), @@ -60,16 +63,20 @@ const ProjectListTool = Tool.make("t3_project_list", { const ProjectReadTool = Tool.make("t3_project_read", { ...shared, description: - "Read a registered project in this environment, including its workspace and saved scripts.", - parameters: Schema.Struct({ projectId: ProjectId }), + "Read a registered project locally or in a linked environment with environmentId, including its workspace and saved scripts.", + parameters: Schema.Struct({ + 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 locally or in a linked environment with environmentId 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.", parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, ...ProjectCreatePayload.fields, workspaceRoot: Schema.optional(ProjectCreatePayload.fields.workspaceRoot), }), @@ -79,20 +86,31 @@ const ProjectCreateTool = Tool.make("t3_project_create", { const ProjectUpdateTool = Tool.make("t3_project_update", { ...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 }), + "Update a registered project's settings locally or in a linked environment with environmentId. Omitted fields are preserved. Uses the same project service as the app.", + 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({ projectId: ProjectId, force: Schema.optionalKey(Schema.Boolean) }), + "Delete a project locally or in a linked environment with environmentId 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), + }), }).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.", - parameters: SourceControlCloneRepositoryInput, + "Clone a repository locally or in a linked environment with environmentId 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: Schema.Struct({ + ...SourceControlCloneRepositoryInput.fields, + environmentId: OrchestratorMcpEnvironmentTarget, + }), success: SourceControlCloneRepositoryResult, dependencies: [ ...shared.dependencies, @@ -104,8 +122,9 @@ const ProjectCloneTool = Tool.make("t3_project_clone", { 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. For upstream commits, 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. Each call creates a new launch with no retry key; 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. 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 locally or in a linked environment with environmentId, 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. For upstream commits, 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. Reuse clientRequestId when retrying a launch after a lost response; omit it for a fresh launch. 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. After errors or lost responses, inspect t3_thread_list before retrying. For a linked destination, pass its projectId or scratch:true; omitted modelSelection uses that project\'s default. 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.', parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, projectId: Schema.optional(ProjectId), scratch: Schema.optional( Schema.Boolean.annotate({ @@ -114,6 +133,7 @@ const ThreadLaunchTool = Tool.make("t3_thread_launch", { }), ), title: TrimmedNonEmptyString, + clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), modelSelection: Schema.optional(ModelSelection), runtimeMode: Schema.optional(RuntimeMode), interactionMode: Schema.optional(ProviderInteractionMode), diff --git a/apps/server/src/mcp/toolkits/thread/tools.ts b/apps/server/src/mcp/toolkits/thread/tools.ts index b5fb7d6be3a68..6b605dc9b5e7b 100644 --- a/apps/server/src/mcp/toolkits/thread/tools.ts +++ b/apps/server/src/mcp/toolkits/thread/tools.ts @@ -1,4 +1,5 @@ import { + OrchestratorMcpEnvironmentTarget, ScheduledTaskId, ScheduledTask, OrchestrationSearchThreadsInput, @@ -30,8 +31,9 @@ import * as McpInvocationContext from "../../McpInvocationContext.ts"; const ThreadOrganizeTool = Tool.make("t3_thread_organize", { 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.", + "Pin, snooze, settle, archive, or mark a thread unread locally or in a linked environment with environmentId. Locally, 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 @@ const ThreadOrganizeTool = Tool.make("t3_thread_organize", { .annotate(Tool.Title, "Organize a thread") .annotate(Tool.Destructive, true); -const queueTarget = { threadId: Schema.optional(ThreadId), queuedRunId: RunId }; +const queueTarget = { + environmentId: OrchestratorMcpEnvironmentTarget, + threadId: Schema.optional(ThreadId), + queuedRunId: RunId, +}; const commandTool = { success: OrchestrationV2DispatchCommandResult, failure: OrchestratorMcpFailure, @@ -80,8 +86,9 @@ const queueEntry = Schema.Struct({ const QueueListTool = Tool.make("t3_queue_list", { ...commandTool, description: - "List queued messages in delivery order. Results are a live offset page; use t3_thread_read for full thread history.", + "List queued messages in delivery order locally or in a linked environment with environmentId. 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 }))), @@ -95,7 +102,8 @@ const QueueListTool = Tool.make("t3_queue_list", { .annotate(Tool.Destructive, false); const QueueReadTool = Tool.make("t3_queue_read", { ...commandTool, - description: "Read up to 16,000 characters of a queued message. Omit threadId for this thread.", + description: + "Read up to 16,000 characters of a queued message locally or in a linked environment with environmentId. Locally, omit threadId for this thread.", parameters: Schema.Struct(queueTarget), success: queueEntry, }) @@ -104,7 +112,7 @@ const QueueReadTool = Tool.make("t3_queue_read", { const QueueEditTool = Tool.make("t3_queue_edit", { ...commandTool, description: - "Replace a queued message's text, preserving its attachments. The service rejects runs that are no longer queued.", + "Replace a queued message's text locally or in a linked environment with environmentId, preserving its attachments. The service rejects runs that are no longer queued.", parameters: Schema.Struct({ ...queueTarget, text: Schema.String.check(Schema.isMaxLength(100000)), @@ -112,22 +120,34 @@ const QueueEditTool = Tool.make("t3_queue_edit", { }).annotate(Tool.Destructive, true); const QueueCancelTool = Tool.make("t3_queue_cancel", { ...commandTool, - description: "Cancel a queued run using the existing queue command.", + description: + "Cancel a queued run locally or in a linked environment with environmentId using the existing queue command.", parameters: Schema.Struct(queueTarget), }).annotate(Tool.Destructive, true); const QueueReorderTool = Tool.make("t3_queue_reorder", { ...commandTool, - description: "Move a queued run before another queued run, or to the end with beforeRunId=null.", - parameters: Schema.Struct({ ...queueTarget, beforeRunId: Schema.NullOr(RunId) }), + description: + "Move a queued run before another queued run, or to the end with beforeRunId=null, locally or in a linked environment with environmentId.", + parameters: Schema.Struct({ + ...queueTarget, + beforeRunId: Schema.NullOr(RunId), + }), }).annotate(Tool.Destructive, true); const QueuePromoteTool = Tool.make("t3_queue_promote_to_steer", { ...commandTool, description: - "Deliver a queued message as steering to the specified active run. Existing provider and run-state rules apply.", - parameters: Schema.Struct({ ...queueTarget, targetRunId: RunId }), + "Deliver a queued message as steering to the specified active run locally or in a linked environment with environmentId. Existing provider and run-state rules apply.", + 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, @@ -151,8 +171,11 @@ const pendingRequest = Schema.Struct({ const PendingRequestListTool = Tool.make("t3_pending_request_list", { ...commandTool, description: - "List pending user questions in a thread. Omit threadId for this thread. Approval requests are not included.", - parameters: Schema.Struct({ threadId: Schema.optional(ThreadId) }), + "List pending user questions in a thread locally or in a linked environment with environmentId. Locally, omit threadId for this thread. Approval requests are not included.", + parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, + threadId: Schema.optional(ThreadId), + }), success: Schema.Struct({ requestIds: Schema.Array(RuntimeRequestId) }), }) .annotate(Tool.Readonly, true) @@ -160,7 +183,7 @@ const PendingRequestListTool = Tool.make("t3_pending_request_list", { const PendingRequestReadTool = Tool.make("t3_pending_request_read", { ...commandTool, description: - "Read a pending user question. Answer with t3_pending_request_respond; existing live or message response handling is used.", + "Read a pending user question locally or in a linked environment with environmentId. Answer with t3_pending_request_respond; existing live or message response handling is used.", parameters: Schema.Struct(requestTarget), success: pendingRequest, }) @@ -169,8 +192,11 @@ const PendingRequestReadTool = Tool.make("t3_pending_request_read", { const PendingRequestRespondTool = Tool.make("t3_pending_request_respond", { ...commandTool, description: - "Answer a pending user-input request using the existing runtime response command. This cannot approve a permission request.", - parameters: Schema.Struct({ ...requestTarget, answers: ProviderUserInputAnswers }), + "Answer a pending user-input request locally or in a linked environment with environmentId using the existing runtime response command. This cannot approve a permission request.", + parameters: Schema.Struct({ + ...requestTarget, + answers: ProviderUserInputAnswers, + }), }) .annotate(Tool.Destructive, true) .annotate(Tool.OpenWorld, true); @@ -178,8 +204,11 @@ const PendingRequestRespondTool = Tool.make("t3_pending_request_respond", { const ThreadConfigurationTool = Tool.make("t3_thread_configuration", { ...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({ threadId: Schema.optional(ThreadId) }), + "Read a thread's provider/model selection and modes locally or in a linked environment with environmentId. Locally, omit threadId for this thread. orchestrator_capabilities lists available providers and models.", + parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, + threadId: Schema.optional(ThreadId), + }), success: Schema.Struct({ threadId: ThreadId, modelSelection: ModelSelection, @@ -192,8 +221,9 @@ const ThreadConfigurationTool = Tool.make("t3_thread_configuration", { const ThreadConfigureTool = Tool.make("t3_thread_configure", { ...commandTool, 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.", + "Set a thread's provider, model and options locally or in a linked environment with environmentId using the existing selection command. Locally, 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, }), @@ -203,8 +233,9 @@ const transferResult = Schema.Struct({ sequence: NonNegativeInt, targetThreadId: const ThreadForkTool = Tool.make("t3_thread_fork", { ...commandTool, 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.", + "Fork a thread from a stable run or checkpoint locally or in a linked environment with environmentId using the existing fork command. Locally, 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), @@ -214,8 +245,9 @@ const ThreadForkTool = Tool.make("t3_thread_fork", { const ThreadMergeBackTool = Tool.make("t3_thread_merge_back", { ...commandTool, 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.", + "Merge context from a thread back to a related thread in the same project locally or in a linked environment with environmentId. Locally, 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, @@ -224,8 +256,12 @@ const ThreadMergeBackTool = Tool.make("t3_thread_merge_back", { }).annotate(Tool.Destructive, true); const ThreadTransfersTool = Tool.make("t3_thread_transfers", { ...commandTool, - description: "Read context transfer status for a thread. Omit threadId for this thread.", - parameters: Schema.Struct({ threadId: Schema.optional(ThreadId) }), + description: + "Read context transfer status for a thread locally or in a linked environment with environmentId. Locally, omit threadId for this thread.", + parameters: Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, + threadId: Schema.optional(ThreadId), + }), success: Schema.Struct({ transfers: Schema.Array( Schema.Struct({ @@ -243,8 +279,9 @@ const ThreadTransfersTool = Tool.make("t3_thread_transfers", { const ThreadSearchTool = Tool.make("t3_thread_search", { ...commandTool, 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.", + "Search active thread titles and content locally or in a linked environment with environmentId using the app's existing bounded search. Matches are limited to one project (projectId, else the local 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), }), diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index be276af5f0edf..5bf7edaa27a38 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -1231,6 +1231,7 @@ export const CLAUDE_READ_ONLY_T3_MCP_ALLOWED_TOOLS: ReadonlyArray = [ "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", diff --git a/apps/server/src/orchestration-v2/DispatchModeLimit.ts b/apps/server/src/orchestration-v2/DispatchModeLimit.ts index ed255570d8ad5..11887e8926af5 100644 --- a/apps/server/src/orchestration-v2/DispatchModeLimit.ts +++ b/apps/server/src/orchestration-v2/DispatchModeLimit.ts @@ -43,3 +43,17 @@ export const exceededDispatchModeLimit = ( : interactionModeRank[modes.interactionMode] > interactionModeRank[limit.interactionMode] ? "interaction" : undefined; + +// A launch must honor both its requested modes and any narrower caller limit. +export const narrowDispatchModeLimit = ( + requested: DispatchModes, + ambient: DispatchModeLimitValue | undefined, +): DispatchModeLimitValue => ({ + ...ambient, + runtimeMode: + ambient !== undefined && + runtimeModeRank[ambient.runtimeMode] < runtimeModeRank[requested.runtimeMode] + ? ambient.runtimeMode + : requested.runtimeMode, + interactionMode: ambient?.interactionMode === "plan" ? "plan" : requested.interactionMode, +}); diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index 8bdf7230fd06f..ec4aafce6431d 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -9,6 +9,12 @@ import { createPendingAttachmentId, resolveAttachmentPath } from "../attachmentS import * as ThreadMessageIntake from "./ThreadMessageIntake.ts"; import { assert, it, vi } from "@effect/vitest"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import { McpSchema, McpServer } from "effect/ai"; +import * as McpHttpServer from "../mcp/McpHttpServer.ts"; +import * as McpInvocationContext from "../mcp/McpInvocationContext.ts"; +import * as ProjectHandlers from "../mcp/toolkits/project/handlers.ts"; +import { ProjectToolkit } from "../mcp/toolkits/project/tools.ts"; +import * as SourceControlRepositories from "../sourceControl/SourceControlRepositoryService.ts"; import { ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE, ChatAttachmentId, @@ -70,6 +76,9 @@ import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAda const projectId = ProjectId.make("project:launch-test"); const otherProjectId = ProjectId.make("project:launch-other"); const encodeThreadProjection = Schema.encodeEffect(OrchestrationV2ThreadProjectionJson); +const decodeMcpLaunchResult = Schema.decodeUnknownEffect( + ProjectToolkit.tools.t3_thread_launch.successSchema, +); const modelSelection = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.1-codex", @@ -2033,6 +2042,137 @@ it.effect("replays a server-allocated launch", () => }), ); +it.effect("replays MCP launches durably within each caller's request namespace", () => + Effect.gen(function* () { + const setupEntered = yield* Deferred.make(); + const allowSetup = yield* Deferred.make(); + const harness = makeHarness({ + hasCommit: () => Effect.succeed(true), + runSetup: () => + Deferred.succeed(setupEntered, undefined).pipe( + Effect.andThen(Deferred.await(allowSetup)), + Effect.as({ status: "no-script" as const }), + ), + }); + const layerDependencies = Layer.mergeAll( + harness.layer, + NodeCrypto.layer, + NodeServices.layer, + ServerConfig.layerTest(process.cwd(), { prefix: "t3-mcp-launch-retry-" }).pipe( + Layer.provide(NodeServices.layer), + ), + Layer.mock(GitVcsDriver.GitVcsDriver)({}), + Layer.mock(SourceControlRepositories.SourceControlRepositoryService)({}), + ); + const layerToolkit = McpHttpServer.toolkitRegistration( + ProjectToolkit, + ProjectHandlers.layer, + ).pipe(Layer.provideMerge(McpServer.McpServer.layer), Layer.provideMerge(layerDependencies)); + yield* Effect.gen(function* () { + const server = yield* McpServer.McpServer; + const threads = yield* ThreadManagement.ThreadManagementService; + const client = McpSchema.McpServerClient.of({ + clientId: 1, + protocolVersion: "2025-06-18", + clientCapabilities: {}, + clientInfo: { name: "launch-retry", version: "1" }, + initializePayload: { + protocolVersion: "2025-06-18", + capabilities: {}, + clientInfo: { name: "launch-retry", version: "1" }, + }, + getClient: Effect.die("unused"), + }); + const launch = (sessionId: string) => + server + .callTool({ + name: "t3_thread_launch", + arguments: { + projectId, + title: "Retry safely", + message: "Only one initial message per caller", + workspaceStrategy: { type: "worktree", baseRef: "main" }, + clientRequestId: "same-request", + }, + }) + .pipe( + Effect.provideService(McpSchema.McpServerClient, client), + Effect.provideService(McpInvocationContext.McpInvocationContext, { + environmentId: EnvironmentId.make("launch-environment"), + requestNamespace: `client:${sessionId}`, + thread: undefined, + client: { sessionId, label: sessionId, access: "auto" }, + issuedAt: 0, + capabilities: new Set(["orchestration"] as const), + }), + Effect.flatMap((result) => { + assert.isFalse(result.isError, JSON.stringify(result)); + return decodeMcpLaunchResult(result.structuredContent); + }), + ); + const first = yield* launch("caller-a"); + yield* Deferred.await(setupEntered); + const replay = yield* launch("caller-a"); + const other = yield* launch("caller-b"); + const otherReplay = yield* launch("caller-b"); + assert.equal(replay.threadId, first.threadId); + assert.equal(replay.runId, first.runId); + assert.notEqual(other.threadId, first.threadId); + assert.equal(otherReplay.threadId, other.threadId); + assert.equal(otherReplay.runId, other.runId); + for (const result of [first, other]) { + const projection = yield* threads.getThreadProjection(result.threadId); + assert.equal(projection.messages.length, 1); + assert.equal(projection.runs.length, 1); + assert.equal(projection.runs[0]?.id, result.runId); + assert.equal(projection.thread.projectId, projectId); + assert.equal(projection.thread.runtimeMode, "auto"); + assert.equal(projection.thread.interactionMode, "default"); + assert.deepEqual(projection.thread.modelSelection, modelSelection); + } + }).pipe(Effect.provide(layerToolkit)); + }), +); + +it.effect("refuses a narrowed agent launch replay before persisting its first message", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + for (const modes of [ + { runtimeMode: "auto", interactionMode: "default" }, + { runtimeMode: "full-access", interactionMode: "plan" }, + ] as const) { + const input = { + ...launchInput({ + command: `command:launch:narrowed:${modes.runtimeMode}`, + thread: `thread:launch:narrowed:${modes.runtimeMode}`, + }), + createdBy: "agent" as const, + creationSource: "mcp" as const, + }; + const created = yield* launches.launch(input); + // A retry can carry the first message after an earlier create-only request won. + const result = yield* launches + .launch({ + ...input, + ...modes, + initialMessage: { text: "Must not start above my modes", attachments: [] }, + }) + .pipe(Effect.exit); + const projection = yield* threads.getThreadProjection(created.threadId); + assert.equal(projection.messages.length, 0); + assert.equal(projection.runs.length, 0); + assert.isTrue(Exit.isFailure(result)); + if (Exit.isFailure(result)) { + const error = Cause.squash(result.cause); + assert.instanceOf(error, ThreadLaunch.ThreadLaunchError); + assert.equal((error as ThreadLaunch.ThreadLaunchError).operation, "dispatch-message"); + } + } + }).pipe(Effect.provide(harness.layer)); +}); + it.effect("refuses a launch when the project disables the model's provider instance", () => { const harness = makeHarness({ serverSettings: { diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index 757feba982107..819a96e1d6493 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -1,3 +1,4 @@ +import { DispatchModeLimit, narrowDispatchModeLimit } from "./DispatchModeLimit.ts"; import * as WorktreeSetupTracker from "../project/WorktreeSetupTracker.ts"; import * as ProjectCloneTracker from "../project/ProjectCloneTracker.ts"; import * as TerminalManager from "../terminal/Manager.ts"; @@ -956,6 +957,11 @@ const make = Effect.gen(function* () { (yield* ids.allocate .message({ threadId, ordinal: 1 }) .pipe(Effect.mapError(mapError(input, "dispatch-message", threadId)))); + const ambientLimit = yield* DispatchModeLimit; + const launchLimit = + input.createdBy === "agent" + ? narrowDispatchModeLimit(input, ambientLimit) + : ambientLimit; const dispatched = yield* threads .dispatch({ type: "message.dispatch", @@ -977,7 +983,10 @@ const make = Effect.gen(function* () { createdBy: input.createdBy, creationSource: input.creationSource, }) - .pipe(Effect.mapError(mapError(input, "dispatch-message", threadId))); + .pipe( + Effect.provideService(DispatchModeLimit, launchLimit), + Effect.mapError(mapError(input, "dispatch-message", threadId)), + ); const runCreated = dispatched.storedEvents.find( (stored) => stored.event.type === "run.created", ); diff --git a/apps/server/src/peer/PeerForwarding.test.ts b/apps/server/src/peer/PeerForwarding.test.ts new file mode 100644 index 0000000000000..45a3f6403ab4d --- /dev/null +++ b/apps/server/src/peer/PeerForwarding.test.ts @@ -0,0 +1,1182 @@ +import { expect, it } from "@effect/vitest"; +import { NodeHttpServer } from "@effect/platform-node"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { + ChatAttachmentId, + AuthSessionId, + CommandId, + EnvironmentHttpApi, + EnvironmentId, + OrchestratorMcpFailure, + MessageId, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + RunId, + ThreadId, + type ExecutionEnvironmentDescriptor, + type OrchestratorMcpThreadWaitResult, +} from "@t3tools/contracts"; +import { createPendingAttachmentId, resolveAttachmentPath } from "../attachmentStore.ts"; +import * as Clock from "effect/Clock"; +import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Queue from "effect/Queue"; +import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as TestClock from "effect/testing/TestClock"; +import { McpSchema, McpServer } from "effect/ai"; +import type * as Tool from "effect/ai/Tool"; +import * as FetchHttpClient from "effect/http/FetchHttpClient"; +import * as HttpRouter from "effect/http/HttpRouter"; +import * as HttpServer from "effect/http/HttpServer"; +import * as HttpServerRequest from "effect/http/HttpServerRequest"; +import * as HttpServerResponse from "effect/http/HttpServerResponse"; +import * as HttpApi from "effect/http-api/HttpApi"; +import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder"; +import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +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 McpSessionRegistryTestkit from "../mcp/McpSessionRegistry.testkit.ts"; +import * as OrchestratorMcpService from "../mcp/OrchestratorMcpService.ts"; +import * as ThreadMetadataMcpService from "../mcp/ThreadMetadataMcpService.ts"; +import * as ProjectHandlers from "../mcp/toolkits/project/handlers.ts"; +import { ProjectToolkit } from "../mcp/toolkits/project/tools.ts"; +import { OrchestratorToolkit } from "../mcp/toolkits/orchestrator/tools.ts"; +import { ThreadToolkit } from "../mcp/toolkits/thread/tools.ts"; +import { EnvironmentToolkit } from "../mcp/toolkits/environment/tools.ts"; +import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts"; +import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts"; +import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexAdapterV2.ts"; +import * as ProviderReplayHarness from "../orchestration-v2/testkit/ProviderReplayHarness.ts"; +import * as RuntimeLayer from "../orchestration-v2/runtimeLayer.ts"; +import * as ThreadCommandExecutor from "../orchestration-v2/ThreadCommandExecutor.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 Sqlite from "../persistence/Sqlite.ts"; +import * as ProjectService from "../project/ProjectService.ts"; +import * as ProjectEnrichment from "../project/ProjectEnrichmentService.ts"; +import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts"; +import * as ProjectFaviconResolver from "../project/ProjectFaviconResolver.ts"; +import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts"; +import * as ProjectCloneTracker from "../project/ProjectCloneTracker.ts"; +import * as WorktreeSetupTracker from "../project/WorktreeSetupTracker.ts"; +import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts"; +import * as ProviderRegistryMock from "../provider/testUtils/providerRegistryMock.ts"; +import * as GitWorkflow from "../git/GitWorkflowService.ts"; +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; +import * as Repositories from "../sourceControl/SourceControlRepositoryService.ts"; +import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts"; +import * as ScheduledTasks from "../scheduledTasks/ScheduledTaskService.ts"; +import * as SecretRequests from "../secrets/SecretRequests.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as TerminalManager from "../terminal/Manager.ts"; +import * as TextGeneration from "../textGeneration/TextGeneration.ts"; +import * as WorkspacePaths from "../workspace/WorkspacePaths.ts"; +import * as PeerLinks from "./PeerLinks.ts"; +import * as PeerMcpClient from "./PeerMcpClient.ts"; +import * as PeerForwarding from "./PeerForwarding.ts"; + +class PeerApi extends HttpApi.make("environment") + .add(EnvironmentHttpApi.groups.metadata) + .add(EnvironmentHttpApi.groups.mcpOAuth) {} +const destinationId = EnvironmentId.make("destination"); +const sourceId = EnvironmentId.make("source"); +const modelSelection = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.1-codex" }; +const descriptor = (environmentId: EnvironmentId): ExecutionEnvironmentDescriptor => ({ + environmentId, + label: environmentId, + platform: { os: "linux", arch: "x64" }, + serverVersion: "0.0.0-test", + // Only the fixture advertises compatibility with the future receiving fence. + capabilities: { repositoryIdentity: true, mcpModeLimitHeader: true, linkFence: true }, +}); +const environmentLayer = (value: Effect.Effect) => + Layer.succeed(ServerEnvironment.ServerEnvironment, { + getDescriptor: value, + getEnvironmentId: value.pipe(Effect.map((d) => d.environmentId)), + }); +const decodeFailure = Schema.decodeUnknownSync(Schema.fromJsonString(OrchestratorMcpFailure)); +const client = McpSchema.McpServerClient.of({ + clientId: 1, + clientCapabilities: {}, + clientInfo: { name: "test", version: "1" }, + protocolVersion: "2025-06-18", + initializePayload: { + protocolVersion: "2025-06-18", + capabilities: {}, + clientInfo: { name: "test", version: "1" }, + }, + getClient: Effect.die("unused"), +}); +const scope = ( + name = "A", + access: "full-access" | "auto" | "read-only" = "full-access", +): McpInvocationContext.McpInvocationScope => ({ + environmentId: sourceId, + requestNamespace: `client:${name}`, + thread: undefined, + client: { sessionId: name, label: name, access }, + capabilities: new Set(["orchestration"]), + issuedAt: 1, +}); +const registration = Layer.mergeAll( + McpHttpServer.layerOrchestratorToolkit, + McpHttpServer.layerThreadToolkit, + McpHttpServer.layerEnvironmentToolkit, + McpHttpServer.toolkitRegistration(ProjectToolkit, ProjectHandlers.layer), +); + +const makeEnvironment = Effect.fnUntraced(function* (id: EnvironmentId) { + const fs = yield* FileSystem.FileSystem; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3-forwarding-" }); + const described = yield* Ref.make(descriptor(id)); + const database = Sqlite.layerMemory; + const config = ServerConfig.layerTest(root, { prefix: "t3-forwarding-state-" }); + const registry = ProviderAdapterRegistry.layerFromAdapters([ + { + instanceId: modelSelection.instanceId, + driver: ProviderDriverKind.make("codex"), + getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }), + openSession: () => Effect.die("provider execution is disabled"), + } as Parameters[0][number], + ]); + // Persist accepted runs/outbox work, but do not execute external provider turns. + const orchestrator = ProviderReplayHarness.layerWithRegistry({ name: `peer-${id}` }, registry, { + databaseLayer: database, + runEffectWorker: false, + }); + const settings = ServerSettings.layerTest(); + const receipts = CommandReceiptStore.layer.pipe(Layer.provide(database)); + const threads = ThreadManagement.layer.pipe( + Layer.provide(Layer.mergeAll(orchestrator, settings, receipts)), + ); + const projects = RuntimeLayer.layerProjectService.pipe( + Layer.provideMerge(ProjectEnrichment.layer), + Layer.provide( + Layer.mergeAll( + WorkspacePaths.layer, + Layer.succeed(RepositoryIdentityResolver.RepositoryIdentityResolver, { + resolve: () => Effect.succeed(null), + }), + Layer.succeed(ProjectFaviconResolver.ProjectFaviconResolver, { + resolvePath: () => Effect.succeed(null), + }), + ThreadCommandExecutor.layer, + ), + ), + Layer.provide(settings), + Layer.provide(database), + Layer.provide(config), + Layer.provide(NodeServices.layer), + ); + const external = Layer.mergeAll( + WorktreeSetupTracker.layer, + ProviderRegistryMock.layer([ + { + instanceId: modelSelection.instanceId, + driver: ProviderDriverKind.make("codex"), + displayName: `${id} Codex`, + enabled: true, + installed: true, + version: "test", + status: "ready", + auth: { status: "authenticated" }, + checkedAt: "2026-09-13T00:00:00.000Z", + models: [ + { + slug: modelSelection.model, + name: modelSelection.model, + isCustom: false, + capabilities: null, + }, + { slug: `${id}-custom`, name: `${id} custom`, isCustom: true, capabilities: null }, + ], + slashCommands: [], + skills: [], + }, + ]), + Layer.mock(ProjectCloneTracker.ProjectCloneTracker)({ get: () => Effect.succeed(null) }), + Layer.mock(TerminalManager.TerminalManager)({ close: () => Effect.void }), + Layer.mock(GitWorkflow.GitWorkflowService)({ hasCommit: () => Effect.succeed(true) }), + Layer.succeed(ProjectSetupScriptRunner.ProjectSetupScriptRunner, { + runForThread: () => Effect.succeed({ status: "no-script" as const }), + }), + Layer.mock(TextGeneration.TextGeneration)({}), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + namedProjectsRoot: `${root}/projects`, + folderForThread: () => Effect.succeed(Option.none()), + }), + Layer.mock(GitVcsDriver.GitVcsDriver)({ + execute: (input) => + fs.makeDirectory(`${input.cwd}/${input.args.at(-1)}`, { recursive: true }).pipe( + Effect.andThen( + fs.writeFileString( + `${input.cwd}/${input.args.at(-1)}/cloned.txt`, + "controlled git clone", + ), + ), + Effect.as({ + stdout: "", + stderr: "", + exitCode: ChildProcessSpawner.ExitCode(0), + stdoutTruncated: false, + stderrTruncated: false, + }), + Effect.orDie, + ), + }), + Layer.mock(SourceControlProviderRegistry.SourceControlProviderRegistry)({ + resolveLink: () => undefined, + }), + Layer.mock(ScheduledTasks.ScheduledTaskService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(ThreadSearch.ThreadSearch)({}), + ); + const base = Layer.mergeAll( + threads, + projects, + registry, + settings, + receipts, + external, + IdAllocator.layer, + ThreadCommandExecutor.layer, + ); + const launch = ThreadLaunch.layer.pipe(Layer.provide(base)); + const business = Layer.mergeAll( + OrchestratorMcpService.layer.pipe(Layer.provide(launch), Layer.provide(base)), + ThreadMetadataMcpService.layer.pipe(Layer.provide(threads)), + Repositories.layer.pipe(Layer.provide(external)), + launch, + base, + McpOAuth.layerMcpClientAuthenticator.pipe(Layer.provideMerge(EnvironmentAuth.layer)), + ).pipe( + Layer.provideMerge(ServerSecretStore.layer), + Layer.provideMerge(ServerEnvironment.layerIdentity), + Layer.provideMerge(database), + Layer.provideMerge(environmentLayer(Ref.get(described))), + Layer.provideMerge(config), + Layer.provideMerge(NodeServices.layer), + Layer.fresh, + ); + const context = yield* Layer.build(business); + const projectService = Context.get(context, ProjectService.ProjectService); + yield* projectService.create({ + commandId: CommandId.make(`${id}:project`), + projectId: ProjectId.make(`${id}:project`), + workspaceRoot: root, + title: `${id} project`, + }); + const project = yield* projectService.update({ + commandId: CommandId.make(`${id}:defaults`), + projectId: ProjectId.make(`${id}:project`), + defaultModelSelection: modelSelection, + }); + const threadService = Context.get(context, ThreadManagement.ThreadManagementService); + const seed = Effect.fnUntraced(function* ( + name: string, + runtimeMode: "auto" | "full-access" = "full-access", + interactionMode: "default" | "plan" = "default", + ) { + const threadId = ThreadId.make(`${id}:${name}`); + yield* threadService.dispatch({ + type: "thread.create", + commandId: CommandId.make(`${threadId}:create`), + threadId, + projectId: project.id, + title: name, + modelSelection, + runtimeMode, + interactionMode, + branch: null, + worktreePath: root, + createdBy: "user", + creationSource: "web", + }); + return threadId; + }); + const caller = Effect.fnUntraced(function* ( + name: string, + runtimeMode: "auto" | "full-access" = "full-access", + interactionMode: "default" | "plan" = "default", + ) { + const threadId = yield* seed(name, runtimeMode, interactionMode); + yield* threadService.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`${threadId}:start`), + threadId, + messageId: MessageId.make(`${threadId}:start`), + text: "source stays here", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + return { + ...scope(name), + environmentId: id, + client: undefined, + thread: { + threadId, + providerInstanceId: modelSelection.instanceId, + providerSessionId: `live:${threadId}`, + }, + } satisfies McpInvocationContext.McpInvocationScope; + }); + return { + caller, + context, + root, + described, + project, + projects: projectService, + threads: threadService, + seed, + auth: Context.get(context, EnvironmentAuth.EnvironmentAuth), + sql: Context.get(context, SqlClient.SqlClient), + }; +}); + +type RequestRecord = { + method: string | undefined; + session: string | undefined; + limit: string | undefined; + body: unknown; +}; +const serveDestination = Effect.fnUntraced(function* ( + destination: Effect.Success>, +) { + const requests = yield* Ref.make>([]); + const bearers = yield* Ref.make(0); + const probes = yield* Ref.make(0); + const faults = yield* Ref.make>([]); + const middleware = HttpRouter.middleware( + (effect) => + Effect.gen(function* () { + const request = yield* HttpServerRequest.HttpServerRequest; + if (request.headers.authorization !== undefined) yield* Ref.update(bearers, (n) => n + 1); + if (request.url.startsWith("/.well-known/t3/environment")) + yield* Ref.update(probes, (n) => n + 1); + if (request.url !== "/mcp") return yield* effect; + const body = yield* request.json; + const method = (body as { method?: string }).method; + yield* Ref.update(requests, (seen) => [ + ...seen, + { + method, + body, + session: request.headers["mcp-session-id"], + limit: request.headers["t3-mode-limit"], + }, + ]); + if (method === "tools/call") { + const fault = yield* Ref.modify(faults, (items) => [items[0], items.slice(1)]); + if (fault === "404") return HttpServerResponse.empty({ status: 404 }); + if (fault === "malformed") + return HttpServerResponse.text("private upstream secret", { + contentType: "application/json", + }); + if (fault === "rpc-error") + return HttpServerResponse.jsonUnsafe({ + jsonrpc: "2.0", + id: (body as { id: string | number }).id, + error: { code: -32603, message: "private upstream secret" }, + }); + } + return yield* effect; + }), + { global: true }, + ); + const routes = Layer.mergeAll( + registration.pipe( + Layer.provideMerge(McpHttpServer.layerMcpTransport), + Layer.provide(McpSessionRegistryTestkit.layer), + ), + HttpApiBuilder.layer(PeerApi).pipe( + Layer.provide(McpOAuthHttp.layer.pipe(Layer.provide(McpOAuth.layer))), + Layer.provide(ServerHttp.layerServerEnvironmentHttpApi), + Layer.provide(AuthHttp.layerAuthenticatedAuth), + ), + ).pipe(Layer.provide(middleware), Layer.provide(Layer.succeedContext(destination.context))); + yield* HttpRouter.serve(routes, { disableListenLog: true, disableLogger: true }).pipe( + Layer.build, + ); + const address = (yield* HttpServer.HttpServer).address; + if (address._tag === "UnixPathAddress") return yield* Effect.die("TCP expected"); + return { url: `http://127.0.0.1:${address.port}`, requests, faults, bearers, probes }; +}); + +const paired = Effect.gen(function* () { + const destination = yield* makeEnvironment(destinationId); + const http = yield* serveDestination(destination); + const source = yield* makeEnvironment(sourceId); + const linksContext = yield* PeerLinks.layer.pipe( + Layer.provide(FetchHttpClient.layer), + Layer.provide(Layer.succeedContext(source.context)), + Layer.build, + ); + const links = Context.get(linksContext, PeerLinks.PeerLinks); + const pair = Effect.fnUntraced(function* ( + access: "full-access" | "auto" | "read-only" = "full-access", + ) { + return yield* links.link({ + url: http.url, + access, + pairingCode: (yield* destination.auth.issuePairingCredential()).credential, + }); + }); + yield* pair(); + const forwarding = PeerForwarding.layer.pipe( + Layer.provide(PeerMcpClient.layer.pipe(Layer.provide(FetchHttpClient.layer))), + Layer.provide(Layer.succeedContext(linksContext)), + Layer.provide(Layer.succeedContext(source.context)), + ); + const registered = yield* registration.pipe( + Layer.provideMerge(McpServer.McpServer.layer), + Layer.provideMerge(forwarding), + Layer.provide(Layer.succeedContext(source.context)), + Layer.build, + ); + const server = Context.get(registered, McpServer.McpServer); + const invoke = (tool: T, args: Record, caller = scope()) => + server + .callTool({ + name: tool.name, + arguments: args, + }) + .pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, caller), + Effect.provideService(McpSchema.McpServerClient, client), + Effect.map((result) => + result.isError === true + ? { + ...result, + structuredContent: decodeFailure( + result.content + .flatMap((part) => (part.type === "text" ? [part.text] : [])) + .join("\n"), + ), + } + : result, + ), + ); + const call = (tool: T, args: Tool.Parameters, caller = scope()) => + invoke(tool, args as Record, caller).pipe( + Effect.flatMap((result) => { + expect(result.isError, JSON.stringify(result)).not.toBe(true); + // oxlint-disable-next-line t3code/no-inline-schema-compile -- Each actual tool supplies its result schema. + return Schema.decodeUnknownEffect(tool.successSchema as Schema.Decoder>)( + result.structuredContent, + ); + }), + ); + return { source, destination, http, links, pair, call, invoke }; +}); + +it.effect( + "routes registered tools over paired HTTP with durable caller namespaces and destination effects", + () => + Effect.scoped( + Effect.gen(function* () { + const f = yield* paired; + const target = yield* f.destination.seed("target"); + const liveCaller = yield* f.source.caller("local"); + const otherCaller = yield* f.source.caller("other"); + const local = liveCaller.thread.threadId; + const before = yield* f.source.threads.getThreadProjection(local); + const links = yield* f.call(EnvironmentToolkit.tools.t3_environment_links, {}, liveCaller); + expect(links).toMatchObject({ + environmentId: sourceId, + links: [{ environmentId: destinationId, status: "reachable" }], + }); + const localCapabilities = yield* f.call( + OrchestratorToolkit.tools.orchestrator_capabilities, + {}, + ); + expect(localCapabilities.providers).toMatchObject([ + { + models: expect.arrayContaining([{ id: "source-custom", label: "source custom" }]), + canRunChildTask: true, + }, + ]); + const remoteCapabilities = yield* f + .invoke( + OrchestratorToolkit.tools.orchestrator_capabilities, + { environmentId: destinationId }, + liveCaller, + ) + .pipe(Effect.result); + const capabilityRequests = (yield* Ref.get(f.http.requests)).filter( + (request) => + request.method === "tools/call" && + (request.body as { params: { name: string } }).params.name === + "orchestrator_capabilities", + ); + const listed = yield* f.call(OrchestratorToolkit.tools.t3_thread_list, { + environmentId: destinationId, + projectId: f.destination.project.id, + }); + expect(JSON.stringify(listed)).toContain(target); + const read = yield* f.call( + OrchestratorToolkit.tools.t3_thread_read, + { environmentId: destinationId, threadId: target }, + liveCaller, + ); + expect(JSON.stringify(read)).toContain(target); + const args = { + environmentId: destinationId, + threadId: target, + message: "remote message", + clientRequestId: "same", + }; + const a = yield* f.call(OrchestratorToolkit.tools.t3_thread_send, args, liveCaller); + expect(yield* f.call(OrchestratorToolkit.tools.t3_thread_send, args, liveCaller)).toEqual( + a, + ); + const b = yield* f.call(OrchestratorToolkit.tools.t3_thread_send, args, otherCaller); + expect(b.messageId).not.toBe(a.messageId); + const projection = yield* f.destination.threads.getThreadProjection(target); + expect(projection.messages.map((m) => m.id)).toEqual([a.messageId, b.messageId]); + expect(projection.runs).toHaveLength(2); + yield* f.call(OrchestratorToolkit.tools.t3_thread_update, { + environmentId: destinationId, + threadId: target, + action: "rename", + title: "Renamed remotely", + }); + const queued = yield* f.call(OrchestratorToolkit.tools.t3_thread_send, { + ...args, + clientRequestId: "queued", + message: "before edit", + mode: "queue", + }); + yield* f.call(ThreadToolkit.tools.t3_queue_edit, { + environmentId: destinationId, + threadId: target, + queuedRunId: queued.runId, + text: "after edit", + }); + expect( + yield* f.call(ThreadToolkit.tools.t3_queue_list, { + environmentId: destinationId, + threadId: target, + }), + ).toMatchObject({ + items: expect.arrayContaining([ + { queuedRunId: queued.runId, text: "after edit", truncated: false }, + ]), + }); + const edited = yield* f.destination.threads.getThreadProjection(target); + expect(edited.thread.title).toBe("Renamed remotely"); + expect(edited.messages.find((m) => m.id === queued.messageId)?.text).toBe("after edit"); + + const launchArgs = { + environmentId: destinationId, + projectId: f.destination.project.id, + title: "remote launch", + message: "only once", + clientRequestId: "launch", + }; + const launched = yield* f.call( + ProjectToolkit.tools.t3_thread_launch, + launchArgs, + scope("A", "auto"), + ); + const replay = yield* f.call( + ProjectToolkit.tools.t3_thread_launch, + launchArgs, + scope("A", "auto"), + ); + expect(replay.threadId).toBe(launched.threadId); + expect(replay.runId).toBe(launched.runId); + const otherLaunch = yield* f.call( + ProjectToolkit.tools.t3_thread_launch, + launchArgs, + scope("B", "auto"), + ); + expect(otherLaunch.threadId).not.toBe(launched.threadId); + for (const launch of [launched, otherLaunch]) { + const saved = yield* f.destination.threads.getThreadProjection(launch.threadId); + expect(saved.messages).toHaveLength(1); + expect(saved.runs).toHaveLength(1); + expect(saved.thread).toMatchObject({ + projectId: f.destination.project.id, + modelSelection, + runtimeMode: "auto", + interactionMode: "default", + }); + } + expect(yield* f.source.threads.getThreadProjection(local)).toEqual(before); + + yield* Ref.set(f.http.requests, []); + yield* Ref.set(f.http.bearers, 0); + yield* f.call(OrchestratorToolkit.tools.t3_thread_read, { threadId: local }); + yield* f.call(OrchestratorToolkit.tools.t3_thread_read, { + environmentId: sourceId, + threadId: local, + }); + expect(yield* Ref.get(f.http.requests)).toEqual([]); + expect(yield* Ref.get(f.http.bearers)).toBe(0); + expect( + yield* f.invoke(OrchestratorToolkit.tools.t3_thread_read, { + environmentId: EnvironmentId.make("unknown"), + threadId: target, + }), + ).toMatchObject({ isError: true, structuredContent: { code: "invalid_request" } }); + expect( + yield* f.invoke( + OrchestratorToolkit.tools.t3_thread_send, + { + environmentId: sourceId, + threadId: local, + message: "local escalation", + }, + scope("auto", "auto"), + ), + ).toMatchObject({ + isError: true, + structuredContent: { code: "runtime_mode_escalation_denied" }, + }); + const destinationIds = (yield* f.destination.threads.listProjectThreads({ + projectId: f.destination.project.id, + includeSubagents: false, + })).map((thread) => thread.id); + const deniedSend = { ...args, clientRequestId: "denied" }; + for (const caller of [ + scope("reader", "read-only"), + { ...scope(), capabilities: new Set() }, + { ...scope(), environmentId: EnvironmentId.make("wrong-source") }, + ]) { + expect( + yield* f.invoke(OrchestratorToolkit.tools.t3_thread_send, deniedSend, caller), + ).toMatchObject({ isError: true, structuredContent: { code: "capability_denied" } }); + } + const planCaller = yield* f.source.caller("plan", "full-access", "plan"); + expect( + yield* f.invoke( + ProjectToolkit.tools.t3_thread_launch, + { ...launchArgs, interactionMode: "default" }, + planCaller, + ), + ).toMatchObject({ + isError: true, + structuredContent: { code: "interaction_mode_escalation_denied" }, + }); + const fs = yield* FileSystem.FileSystem; + const config = Context.get(f.source.context, ServerConfig.ServerConfig); + const attachment = { + type: "image" as const, + id: ChatAttachmentId.make(createPendingAttachmentId()), + name: "local.png", + mimeType: "image/png", + sizeBytes: 4, + }; + const pendingPath = resolveAttachmentPath({ + attachmentsDir: config.attachmentsDir, + attachment, + })!; + yield* fs.makeDirectory(config.attachmentsDir, { recursive: true }); + yield* fs.writeFile(pendingPath, new Uint8Array([1, 2, 3, 4])); + expect( + yield* f.invoke(ProjectToolkit.tools.t3_thread_launch, { + ...launchArgs, + clientRequestId: "attachment", + attachments: [attachment], + }), + ).toMatchObject({ + isError: true, + structuredContent: { + code: "invalid_request", + message: expect.stringContaining("local attachment"), + }, + }); + expect(Array.from(yield* fs.readFile(pendingPath))).toEqual([1, 2, 3, 4]); + expect( + (yield* f.destination.threads.listProjectThreads({ + projectId: f.destination.project.id, + includeSubagents: false, + })).map((thread) => thread.id), + ).toEqual(destinationIds); + expect(yield* Ref.get(f.http.requests)).toEqual([]); + expect(yield* Ref.get(f.http.bearers)).toBe(0); + + // These refusals must reach the destination: the source cannot know its target modes. + expect( + yield* f.invoke(OrchestratorToolkit.tools.t3_thread_send, deniedSend, planCaller), + ).toMatchObject({ + isError: true, + structuredContent: { code: "interaction_mode_escalation_denied" }, + }); + expect( + yield* f.invoke( + OrchestratorToolkit.tools.t3_thread_send, + deniedSend, + scope("auto", "auto"), + ), + ).toMatchObject({ + isError: true, + structuredContent: { code: "runtime_mode_escalation_denied" }, + }); + expect( + (yield* Ref.get(f.http.requests)) + .filter((r) => r.method === "tools/call") + .map((r) => r.limit), + ).toEqual(["full-access/plan", "auto/default"]); + yield* f.pair("auto"); + const narrowLaunch = yield* f.invoke( + ProjectToolkit.tools.t3_thread_launch, + { ...launchArgs, clientRequestId: "narrow-link" }, + scope(), + ); + expect(yield* f.invoke(OrchestratorToolkit.tools.t3_thread_send, deniedSend)).toMatchObject( + { isError: true, structuredContent: { code: "runtime_mode_escalation_denied" } }, + ); + expect(yield* f.destination.threads.getThreadProjection(target)).toEqual(edited); + expect.soft(remoteCapabilities, JSON.stringify(remoteCapabilities)).toMatchObject({ + _tag: "Success", + success: { + structuredContent: { + providers: [ + { + providerInstanceId: modelSelection.instanceId, + displayName: "destination Codex", + models: expect.arrayContaining([ + { id: "destination-custom", label: "destination custom" }, + ]), + canRunChildTask: true, + }, + ], + }, + }, + }); + + expect.soft(capabilityRequests).toHaveLength(1); + + expect(narrowLaunch.isError, JSON.stringify(narrowLaunch)).not.toBe(true); + // oxlint-disable-next-line t3code/no-inline-schema-compile -- Decode the real launch response before reading its persisted thread. + const narrowLaunched = yield* Schema.decodeUnknownEffect( + ProjectToolkit.tools.t3_thread_launch.successSchema, + )(narrowLaunch.structuredContent); + const narrowSaved = yield* f.destination.threads.getThreadProjection( + narrowLaunched.threadId, + ); + expect(narrowSaved.thread).toMatchObject({ + runtimeMode: "auto", + interactionMode: "default", + }); + }), + ).pipe(Effect.provide(Layer.mergeAll(NodeHttpServer.layerTest, NodeServices.layer))), +); + +it.effect( + "bounds forwarded waits by elapsed time, pins the first run, and interrupts pending requests", + () => + Effect.scoped( + Effect.gen(function* () { + const requests = yield* Queue.unbounded<{ + args: Record; + reply: Deferred.Deferred; + finalized: Deferred.Deferred; + }>(); + const calls = yield* Ref.make(0); + const peers = PeerMcpClient.PeerMcpClient.of({ + call: (input) => + Effect.gen(function* () { + const reply = yield* Deferred.make(); + const finalized = yield* Deferred.make(); + yield* Ref.update(calls, (n) => n + 1); + yield* Queue.offer(requests, { args: input.arguments, reply, finalized }); + return yield* Deferred.await(reply).pipe( + Effect.flatMap(Schema.decodeUnknownEffect(input.success)), + Effect.orDie, + Effect.ensuring(Deferred.succeed(finalized, undefined)), + ); + }), + }); + const context = yield* PeerForwarding.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed(PeerMcpClient.PeerMcpClient, peers), + Layer.mock(PeerLinks.PeerLinks)({}), + Layer.mock(ThreadManagement.ThreadManagementService)({}), + environmentLayer(Effect.succeed(descriptor(sourceId))), + NodeServices.layer, + ), + ), + Layer.build, + ); + const forwarding = Context.get(context, PeerForwarding.PeerForwarding); + const threadId = ThreadId.make("waiting"); + const runId = RunId.make("first-run"); + const response: OrchestratorMcpThreadWaitResult = { + threadId, + runId, + status: "running", + timedOut: true, + }; + const wait = (timeoutMs: number) => + forwarding.call(scope(), OrchestratorToolkit.tools.t3_thread_wait, destinationId, { + threadId, + timeoutMs, + }); + const startedAt = yield* Clock.currentTimeMillis; + const waiting = yield* wait(120_000).pipe(Effect.forkScoped); + const first = yield* Queue.take(requests); + expect(first.args.threadId).toBe(threadId); + expect(Number(first.args.timeoutMs)).toBeGreaterThan(0); + expect(Number(first.args.timeoutMs)).toBeLessThanOrEqual(50_000); + yield* TestClock.adjust(60_000); + yield* Deferred.succeed(first.reply, response); + yield* TestClock.adjust(1); + const second = yield* Queue.take(requests); + expect(second.args.runId).toBe(runId); + expect(Number(second.args.timeoutMs)).toBeGreaterThan(0); + expect(Number(second.args.timeoutMs)).toBeLessThanOrEqual(50_000); + yield* TestClock.adjust(55_000); + yield* Deferred.succeed(second.reply, { ...response, runId: RunId.make("successor") }); + yield* TestClock.adjust(1); + const third = yield* Queue.take(requests); + expect(third.args.runId).toBe(runId); + const finalRemaining = 120_000 - ((yield* Clock.currentTimeMillis) - startedAt); + expect(finalRemaining).toBe(4_998); + const finalTimeout = Number(third.args.timeoutMs); + // The destination starts its timer after probe/init/transit, not at the local deadline. + const overhead = 100; + yield* TestClock.adjust(finalTimeout + overhead); + yield* Deferred.succeed(third.reply, { ...response, status: "completed", timedOut: false }); + const finalAnswer = yield* Fiber.join(waiting); + yield* Deferred.await(third.finalized); + + const quick = yield* wait(10).pipe(Effect.forkScoped); + const quickFirst = yield* Queue.take(requests); + yield* Deferred.succeed(quickFirst.reply, response); + yield* Deferred.await(quickFirst.finalized); + yield* TestClock.adjust(1); + const quickSecond = yield* Queue.take(requests); + expect(quickSecond.args.runId).toBe(runId); + expect(Number(quickSecond.args.timeoutMs)).toBeGreaterThan(0); + expect(Number(quickSecond.args.timeoutMs)).toBeLessThanOrEqual(9); + yield* TestClock.adjust(9); + expect(yield* Fiber.join(quick)).toEqual(response); + yield* Deferred.await(quickSecond.finalized); + + const noResponse = yield* wait(20).pipe(Effect.result, Effect.forkScoped); + const pending = yield* Queue.take(requests); + yield* TestClock.adjust(20); + expect(yield* Fiber.join(noResponse)).toMatchObject({ + _tag: "Failure", + failure: { code: "orchestration_error" }, + }); + yield* Deferred.await(pending.finalized); + + const cancelled = yield* wait(120_000).pipe(Effect.forkScoped); + const cancelling = yield* Queue.take(requests); + const count = yield* Ref.get(calls); + yield* Fiber.interrupt(cancelled); + yield* Deferred.await(cancelling.finalized); + yield* TestClock.adjust(120_000); + expect(yield* Ref.get(calls)).toBe(count); + expect(yield* Queue.size(requests)).toBe(0); + + const short = yield* wait(10_000).pipe(Effect.result, Effect.forkScoped); + const shortRequest = yield* Queue.take(requests); + const shortTimeout = Number(shortRequest.args.timeoutMs); + yield* TestClock.adjust(shortTimeout + overhead); + yield* Deferred.succeed(shortRequest.reply, response); + yield* Deferred.await(shortRequest.finalized); + yield* TestClock.adjust(Math.max(0, 10_000 - shortTimeout - overhead)); + const shortAnswer = yield* Fiber.join(short); + // Continuing after a fresh timeout is allowed, but any last request must honor the deadline. + for (const remaining of yield* Queue.clear(requests)) { + yield* Deferred.await(remaining.finalized); + } + expect + .soft(shortAnswer, JSON.stringify(shortAnswer)) + .toMatchObject({ _tag: "Success", success: response }); + expect + .soft(finalAnswer, JSON.stringify(finalAnswer)) + .toMatchObject({ runId, status: "completed", timedOut: false }); + for (const [requested, remaining] of [ + [shortTimeout, 10_000], + [finalTimeout, finalRemaining], + ] as const) { + expect.soft(requested).toBeLessThanOrEqual(50_000); + expect.soft(requested).toBeLessThan(remaining); + expect.soft(requested).toBeGreaterThan(0); + expect.soft(remaining - requested).toBeLessThanOrEqual(5_000); + } + }), + ), +); + +it.effect( + "renews MCP sessions once, revalidates repaired credentials, and persists safe best-effort outcomes", + () => + Effect.scoped( + Effect.gen(function* () { + const f = yield* paired; + const target = yield* f.destination.seed("protocol"); + const outcome = () => f.source.sql<{ + last_reached_at: string | null; + last_error: string | null; + }>` + SELECT last_reached_at, last_error FROM peer_environment_links WHERE environment_id = ${destinationId}`; + const pairedAt = (yield* outcome())[0]!.last_reached_at; + yield* TestClock.adjust(1); + const read = { environmentId: destinationId, threadId: target }; + yield* Ref.set(f.http.requests, []); + yield* Ref.set(f.http.bearers, 0); + yield* f.call(OrchestratorToolkit.tools.t3_thread_read, read); + yield* f.call(OrchestratorToolkit.tools.t3_thread_read, read, scope("narrow", "auto")); + let requests = yield* Ref.get(f.http.requests); + expect(requests.filter((r) => r.method === "initialize")).toHaveLength(1); + const reused = requests.filter((r) => r.method === "tools/call"); + expect(reused.map((r) => r.limit)).toEqual(["full-access/default", "auto/default"]); + expect(reused[1]?.session).toBe(reused[0]?.session); + const reachedAt = (yield* outcome())[0]!.last_reached_at; + expect(reachedAt).not.toBe(pairedAt); + yield* TestClock.adjust(1); + expect( + yield* f.invoke( + OrchestratorToolkit.tools.t3_thread_send, + { + ...read, + message: "must refuse", + clientRequestId: "narrow", + }, + scope("narrow", "auto"), + ), + ).toMatchObject({ + isError: true, + structuredContent: { code: "runtime_mode_escalation_denied" }, + }); + expect((yield* f.destination.threads.getThreadProjection(target)).messages).toEqual([]); + const refused = (yield* outcome())[0]!; + expect(refused.last_reached_at).toBe(reachedAt); + expect(refused.last_error).toBe("Peer call failed (runtime_mode_escalation_denied)."); + + yield* Ref.set(f.http.requests, []); + yield* Ref.set(f.http.bearers, 0); + yield* Ref.set(f.http.faults, ["404"]); + const send = { ...read, message: "renew once", clientRequestId: "renew" }; + const accepted = yield* f.call(OrchestratorToolkit.tools.t3_thread_send, send); + requests = yield* Ref.get(f.http.requests); + expect(requests.filter((r) => r.method === "initialize")).toHaveLength(1); + const retried = requests.filter((r) => r.method === "tools/call"); + expect(retried).toHaveLength(2); + expect(retried[0]?.body).toEqual(retried[1]?.body); + expect(retried[0]?.session).not.toBe(retried[1]?.session); + expect(yield* f.call(OrchestratorToolkit.tools.t3_thread_send, send)).toEqual(accepted); + expect((yield* f.destination.threads.getThreadProjection(target)).messages).toHaveLength(1); + expect((yield* outcome())[0]?.last_error).toBeNull(); + + yield* Ref.set(f.http.requests, []); + yield* Ref.set(f.http.bearers, 0); + yield* Ref.set(f.http.faults, ["404", "404"]); + expect( + yield* f.invoke(OrchestratorToolkit.tools.t3_thread_send, { + ...send, + clientRequestId: "two404", + }), + ).toMatchObject({ isError: true, structuredContent: { code: "orchestration_error" } }); + requests = yield* Ref.get(f.http.requests); + expect(requests.filter((r) => r.method === "initialize")).toHaveLength(1); + expect(requests.filter((r) => r.method === "tools/call")).toHaveLength(2); + expect((yield* f.destination.threads.getThreadProjection(target)).messages).toHaveLength(1); + for (const fault of ["malformed", "rpc-error"] as const) { + yield* Ref.set(f.http.faults, [fault]); + const failed = yield* f.invoke(OrchestratorToolkit.tools.t3_thread_read, read); + expect(failed).toMatchObject({ + isError: true, + structuredContent: { code: "orchestration_error" }, + }); + expect(JSON.stringify(failed)).not.toContain("private upstream secret"); + expect((yield* outcome())[0]?.last_error).toBe("Peer call failed (orchestration_error)."); + } + yield* f.call(OrchestratorToolkit.tools.t3_thread_read, read); + expect((yield* outcome())[0]?.last_error).toBeNull(); + + // A real SQLite trigger fails only diagnostic updates, after destination acceptance. + yield* f.source + .sql`CREATE TRIGGER fail_peer_outcome BEFORE UPDATE OF last_reached_at, last_error + ON peer_environment_links BEGIN SELECT RAISE(FAIL, 'private outcome storage failure'); END`; + const durable = yield* f.call(OrchestratorToolkit.tools.t3_thread_send, { + ...send, + message: "accepted despite diagnostic failure", + clientRequestId: "outcome-fails", + }); + yield* f.source.sql`DROP TRIGGER fail_peer_outcome`; + const persisted = yield* f.destination.threads.getThreadProjection(target); + expect(persisted.messages.map((m) => m.id)).toContain(durable.messageId); + expect(persisted.runs).toHaveLength(2); + + const sessions = yield* f.destination.auth.listSessions(); + expect(sessions).toHaveLength(1); + yield* f.destination.auth.revokeSession(AuthSessionId.make(sessions[0]!.sessionId)); + expect( + yield* f.invoke(OrchestratorToolkit.tools.t3_thread_send, { + ...send, + clientRequestId: "revoked", + }), + ).toMatchObject({ + isError: true, + structuredContent: { + code: "capability_denied", + message: expect.stringMatching(/link.*again/i), + }, + }); + yield* f.pair("read-only"); + yield* Ref.set(f.http.requests, []); + yield* Ref.set(f.http.bearers, 0); + yield* f.call(OrchestratorToolkit.tools.t3_thread_read, read); + expect( + (yield* Ref.get(f.http.requests)).filter((r) => r.method === "initialize"), + ).toHaveLength(1); + expect( + yield* f.invoke(OrchestratorToolkit.tools.t3_thread_send, { + ...send, + clientRequestId: "repaired-narrow", + }), + ).toMatchObject({ isError: true, structuredContent: { code: "capability_denied" } }); + expect(yield* f.destination.threads.getThreadProjection(target)).toEqual(persisted); + for (const changed of [ + { + ...descriptor(destinationId), + capabilities: { repositoryIdentity: true, linkFence: true }, + }, + { + ...descriptor(destinationId), + capabilities: { repositoryIdentity: true, mcpModeLimitHeader: true }, + }, + descriptor(EnvironmentId.make("different-peer")), + ]) { + yield* Ref.set(f.destination.described, changed); + yield* Ref.set(f.http.probes, 0); + yield* Ref.set(f.http.requests, []); + yield* Ref.set(f.http.bearers, 0); + expect((yield* f.invoke(OrchestratorToolkit.tools.t3_thread_read, read)).isError).toBe( + true, + ); + expect(yield* Ref.get(f.http.probes)).toBe(1); + expect(yield* Ref.get(f.http.requests)).toEqual([]); + expect(yield* Ref.get(f.http.bearers)).toBe(0); + } + yield* Ref.set(f.destination.described, descriptor(destinationId)); + yield* f.call(OrchestratorToolkit.tools.t3_thread_read, read); + expect((yield* outcome())[0]?.last_error).toBeNull(); + const reusedIds = reused.map((request) => (request.body as { id: string | number }).id); + expect(reusedIds).not.toContain(0); + expect( + new Set(reusedIds).size, + "sequential calls in one MCP session need distinct IDs", + ).toBe(2); + }), + ).pipe(Effect.provide(Layer.mergeAll(NodeHttpServer.layerTest, NodeServices.layer))), +); + +it.effect( + "clones and registers projects only with full-access default authority at both ends", + () => + Effect.scoped( + Effect.gen(function* () { + const f = yield* paired; + const fs = yield* FileSystem.FileSystem; + const sourceBefore = yield* f.source.projects.snapshot; + const remoteUrl = "https://example.com/team/repository.git"; + const clone = yield* f.call(ProjectToolkit.tools.t3_project_clone, { + environmentId: destinationId, + destinationPath: `${f.destination.root}/clone`, + remoteUrl, + }); + expect(yield* fs.readFileString(`${clone.cwd}/cloned.txt`)).toBe("controlled git clone"); + const project = yield* f.call(ProjectToolkit.tools.t3_project_create, { + environmentId: destinationId, + workspaceRoot: clone.cwd, + title: "Cloned project", + }); + yield* f.call(ProjectToolkit.tools.t3_project_update, { + environmentId: destinationId, + projectId: project.id, + defaultModelSelection: modelSelection, + }); + expect( + yield* f.call(ProjectToolkit.tools.t3_project_read, { + environmentId: destinationId, + projectId: project.id, + }), + ).toMatchObject({ + id: project.id, + workspaceRoot: clone.cwd, + defaultModelSelection: modelSelection, + }); + expect( + Option.getOrThrow(yield* f.destination.projects.getById(project.id)).workspaceRoot, + ).toBe(clone.cwd); + expect(yield* f.source.projects.snapshot).toEqual(sourceBefore); + const projectCount = (yield* f.destination.projects.snapshot).projects.length; + const forbidden = `${f.destination.root}/forbidden`; + const createArgs = { + environmentId: destinationId, + workspaceRoot: forbidden, + title: "Forbidden", + createWorkspaceRootIfMissing: true, + }; + const cloneArgs = { environmentId: destinationId, destinationPath: forbidden, remoteUrl }; + yield* Ref.set(f.http.requests, []); + yield* Ref.set(f.http.bearers, 0); + const plan = { + ...scope(), + client: { + ...scope().client!, + narrowedTo: { runtimeMode: "full-access" as const, interactionMode: "plan" as const }, + }, + }; + for (const caller of [scope("auto", "auto"), plan]) { + expect( + yield* f.invoke(ProjectToolkit.tools.t3_project_create, createArgs, caller), + ).toMatchObject({ isError: true, structuredContent: { code: "capability_denied" } }); + expect( + yield* f.invoke(ProjectToolkit.tools.t3_project_clone, cloneArgs, caller), + ).toMatchObject({ isError: true, structuredContent: { code: "capability_denied" } }); + } + expect(yield* Ref.get(f.http.requests)).toEqual([]); + expect(yield* Ref.get(f.http.bearers)).toBe(0); + expect(yield* fs.exists(forbidden)).toBe(false); + yield* f.pair("auto"); + yield* Ref.set(f.http.requests, []); + yield* Ref.set(f.http.bearers, 0); + expect(yield* f.invoke(ProjectToolkit.tools.t3_project_create, createArgs)).toMatchObject({ + isError: true, + structuredContent: { code: "capability_denied" }, + }); + expect(yield* f.invoke(ProjectToolkit.tools.t3_project_clone, cloneArgs)).toMatchObject({ + isError: true, + structuredContent: { code: "capability_denied" }, + }); + expect( + (yield* Ref.get(f.http.requests)).filter((r) => r.method === "tools/call"), + ).toHaveLength(2); + expect(yield* fs.exists(forbidden)).toBe(false); + expect((yield* f.destination.projects.snapshot).projects).toHaveLength(projectCount); + }), + ).pipe(Effect.provide(Layer.mergeAll(NodeHttpServer.layerTest, NodeServices.layer))), +); diff --git a/apps/server/src/peer/PeerForwarding.ts b/apps/server/src/peer/PeerForwarding.ts new file mode 100644 index 0000000000000..d30212590a41b --- /dev/null +++ b/apps/server/src/peer/PeerForwarding.ts @@ -0,0 +1,180 @@ +import { + type EnvironmentId, + OrchestratorMcpFailure, + type OrchestratorMcpEnvironmentLinksResult, + OrchestratorMcpThreadWaitInput, + OrchestratorMcpThreadWaitResult, +} from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; +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 Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import type * as Tool from "effect/ai/Tool"; +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import * as McpInvocationContext from "../mcp/McpInvocationContext.ts"; +import { assertLiveCaller, 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"; + +type Scope = McpInvocationContext.McpInvocationScope; + +export class PeerForwarding extends Context.Service< + PeerForwarding, + { + readonly call: ( + scope: Scope, + tool: T, + environmentId: EnvironmentId, + params: Tool.Parameters, + ) => Effect.Effect, OrchestratorMcpFailure>; + readonly links: ( + scope: Scope, + ) => Effect.Effect; + } +>()("t3/peer/PeerForwarding") {} + +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; + const environment = yield* ServerEnvironment.ServerEnvironment; + + const callerLimits = (scope: Scope) => + Effect.gen(function* () { + const descriptor = yield* environment.getDescriptor; + if (scope.environmentId !== descriptor.environmentId) + return yield* new OrchestratorMcpFailure({ + code: "capability_denied", + message: "This credential belongs to another environment.", + }); + const caller = yield* readCaller(); + yield* assertLiveCaller(caller); + return caller.limits; + }).pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, scope), + Effect.provideService(ThreadManagement.ThreadManagementService, threads), + ); + + const forward = ( + scope: Scope, + tool: string, + parameters: Schema.Top, + success: Schema.Top, + environmentId: EnvironmentId, + params: unknown, + ) => + Effect.gen(function* () { + const limits = yield* callerLimits(scope); + const encoded = yield* Schema.encodeUnknownEffect(parameters as Schema.Encoder)( + params, + ).pipe(Effect.mapError(unavailable)); + const { environmentId: _target, ...args } = encoded as Record; + 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 local attachment uploads.", + }); + // The destination sees one paired session for all callers here. + if (typeof args.clientRequestId === "string") + args.clientRequestId = yield* crypto + .digest( + "SHA-256", + new TextEncoder().encode( + `${encodeURIComponent(scope.requestNamespace)}:${encodeURIComponent(args.clientRequestId)}`, + ), + ) + .pipe( + Effect.map((digest) => `fwd:${Hex.encode(digest)}`), + Effect.orDie, + ); + return yield* peers.call({ + environmentId, + tool, + arguments: args, + limits, + success: success as Schema.Decoder, + }); + }); + + const waitForThread = ( + scope: Scope, + environmentId: EnvironmentId, + input: OrchestratorMcpThreadWaitInput, + ) => + Effect.gen(function* () { + const budget = Math.min(3_600_000, Math.max(1, input.timeoutMs ?? 600_000)); + const deadline = (yield* Clock.currentTimeMillis) + budget; + let pinnedRun = input.runId; + let last: OrchestratorMcpThreadWaitResult | undefined; + const expired = () => + last === undefined + ? Effect.fail( + new OrchestratorMcpFailure({ + code: "orchestration_error", + message: "The linked environment did not answer before the wait deadline.", + }), + ) + : Effect.succeed({ ...last, timedOut: true }); + while (true) { + const remaining = deadline - (yield* Clock.currentTimeMillis); + if (remaining <= 0) return yield* expired(); + // Reserve time for probes and transport before the absolute deadline. + const headroom = Math.min(3_000, Math.ceil(remaining / 10)); + const result = yield* forward( + scope, + "t3_thread_wait", + OrchestratorMcpThreadWaitInput, + OrchestratorMcpThreadWaitResult, + environmentId, + { + threadId: input.threadId, + runId: pinnedRun, + timeoutMs: Math.max(1, Math.min(50_000, remaining - headroom)), + }, + ).pipe(Effect.timeoutOption(remaining)); + if (Option.isNone(result)) return yield* expired(); + last = result.value; + if (pinnedRun === undefined && last.runId !== null) pinnedRun = last.runId; + if (!last.timedOut || deadline - (yield* Clock.currentTimeMillis) <= headroom) return last; + // A peer may answer a timeout immediately. Yield until time advances + // rather than spin indefinitely while spending no deadline budget. + yield* Effect.sleep(1); + } + }); + + 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"] = (scope) => + Effect.gen(function* () { + yield* callerLimits(scope); + const listed = yield* links.list.pipe(Effect.mapError(unavailable)); + 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, links: listLinks }); +}); + +export const layer = Layer.effect(PeerForwarding, make); diff --git a/apps/server/src/peer/PeerLinks.ts b/apps/server/src/peer/PeerLinks.ts index cfa567d178f65..b4ad2df00b2e5 100644 --- a/apps/server/src/peer/PeerLinks.ts +++ b/apps/server/src/peer/PeerLinks.ts @@ -50,6 +50,10 @@ export class PeerLinks extends Context.Service< environmentId: EnvironmentId, ) => Effect.Effect, PeerLinkError>; readonly unlink: (environmentId: EnvironmentId) => Effect.Effect; + readonly recordOutcome: ( + environmentId: EnvironmentId, + outcome: { readonly error: string | null }, + ) => Effect.Effect; readonly resolve: ( environmentId: EnvironmentId, ) => Effect.Effect; @@ -491,7 +495,16 @@ const make = Effect.gen(function* () { return { link, url, token: new TextDecoder().decode(stored.value) }; }); - return PeerLinks.of({ link, list, get, unlink, resolve }); + const recordOutcome: PeerLinks["Service"]["recordOutcome"] = Effect.fn("PeerLinks.recordOutcome")( + function* (environmentId, outcome) { + const now = DateTime.formatIso(yield* DateTime.now); + yield* sql`UPDATE peer_environment_links SET + last_reached_at = CASE WHEN ${outcome.error} IS NULL THEN ${now} ELSE last_reached_at END, + last_error = ${outcome.error === null ? null : outcome.error.slice(0, 256)} + WHERE environment_id = ${environmentId}`.pipe(Effect.catch(failStorage)); + }, + ); + return PeerLinks.of({ link, list, get, unlink, resolve, recordOutcome }); }); export const layer = Layer.effect(PeerLinks, make); diff --git a/apps/server/src/peer/PeerMcpClient.ts b/apps/server/src/peer/PeerMcpClient.ts new file mode 100644 index 0000000000000..07ca3d0092b50 --- /dev/null +++ b/apps/server/src/peer/PeerMcpClient.ts @@ -0,0 +1,314 @@ +import { + type EnvironmentId, + OrchestratorMcpFailure, + type PeerLinkError, + type ProviderInteractionMode, + type RuntimeMode, +} from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as Effect from "effect/Effect"; +import * as Hex from "effect/encoding/Hex"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; +import { McpSchema } from "effect/ai"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientRequest from "effect/http/HttpClientRequest"; +import * as McpInvocationContext from "../mcp/McpInvocationContext.ts"; +import * as PeerLinks from "./PeerLinks.ts"; + +const PROTOCOL_VERSION = "2025-06-18"; + +export class PeerMcpClient extends Context.Service< + PeerMcpClient, + { + readonly call: >(input: { + readonly environmentId: EnvironmentId; + readonly tool: string; + readonly arguments: Record; + readonly limits: { + readonly runtimeMode: RuntimeMode; + readonly interactionMode: ProviderInteractionMode; + }; + readonly success: S; + }) => Effect.Effect; + } +>()("t3/peer/PeerMcpClient") {} + +const RpcResponse = Schema.Union([ + Schema.Struct({ + jsonrpc: Schema.Literal("2.0"), + id: Schema.Number, + result: Schema.Unknown, + error: Schema.optional(Schema.Never), + }), + Schema.Struct({ + jsonrpc: Schema.Literal("2.0"), + id: Schema.Number, + result: Schema.optional(Schema.Never), + error: Schema.Struct({ + code: Schema.Number, + message: Schema.String, + data: Schema.optional(Schema.Unknown), + }), + }), +]); +const decodeRpcResponse = Schema.decodeUnknownOption(Schema.fromJsonString(RpcResponse)); +const decodeInitializeResult = Schema.decodeUnknownEffect(McpSchema.InitializeResult); +const decodeToolResult = Schema.decodeUnknownEffect(McpSchema.CallToolResult); +const isMcpFailure = Schema.is(OrchestratorMcpFailure); +const Refusal = Schema.Struct({ + code: OrchestratorMcpFailure.fields.code, + message: Schema.String, +}); +const decodeRefusal = Schema.decodeUnknownOption(Refusal); +const decodeTextRefusal = Schema.decodeUnknownOption(Schema.fromJsonString(Refusal)); +const protocolFailure = () => + new OrchestratorMcpFailure({ + code: "orchestration_error", + message: "The linked environment answered in an unexpected shape. Update both environments.", + }); +const revoked = () => + new OrchestratorMcpFailure({ + code: "capability_denied", + message: + "The linked environment no longer accepts this link. It may have been revoked there; link it again.", + }); +const fromLinkError = (error: PeerLinkError) => + new OrchestratorMcpFailure({ + code: + error.reason === "unknown_link" + ? "invalid_request" + : error.reason === "expired" || error.reason === "incompatible" + ? "capability_denied" + : "orchestration_error", + message: error.message, + }); + +const make = Effect.gen(function* () { + const links = yield* PeerLinks.PeerLinks; + const http = yield* HttpClient.HttpClient; + const crypto = yield* Crypto.Crypto; + const nextRequestId = yield* Ref.make(1); + const sessions = yield* Ref.make< + ReadonlyMap< + string, + { + readonly credential: string; + readonly sessionId: string; + } + > + >(new Map()); + + // Consume every body, including errors and the initialized notification. + const post = Effect.fnUntraced(function* ( + url: string, + token: string, + body: unknown, + headers: Record, + ) { + const response = yield* http.execute( + HttpClientRequest.post(`${url}/mcp`).pipe( + HttpClientRequest.setHeaders({ + accept: "application/json, text/event-stream", + authorization: `Bearer ${token}`, + ...headers, + }), + HttpClientRequest.bodyJsonUnsafe(body), + ), + ); + const text = yield* response.text; + return { status: response.status, headers: response.headers, text }; + }); + + const rpcResult = Effect.fnUntraced(function* ( + response: { readonly headers: Record; readonly text: string }, + id: number, + ) { + // Each SSE event has its own JSON-RPC message. Notifications may precede + // the response; data lines within an event are joined with newlines. + const messages = response.headers["content-type"]?.includes("text/event-stream") + ? response.text + .replaceAll("\r\n", "\n") + .split("\n\n") + .flatMap((event) => { + const data = event + .split("\n") + .filter((line) => line.startsWith("data:")) + .map((line) => line.slice(5).replace(/^ /, "")) + .join("\n"); + return data.length === 0 ? [] : [data]; + }) + : [response.text]; + for (const text of messages) { + const decoded = decodeRpcResponse(text); + if (Option.isNone(decoded) || decoded.value.id !== id) continue; + if (decoded.value.error !== undefined) { + const refusal = decodeRefusal(decoded.value.error.data); + if (Option.isSome(refusal)) + return yield* new OrchestratorMcpFailure({ + code: refusal.value.code, + message: refusal.value.message.slice(0, 1000), + }); + return yield* new OrchestratorMcpFailure({ + code: decoded.value.error.code === -32602 ? "invalid_request" : "orchestration_error", + message: "The linked environment refused the MCP request.", + }); + } + return decoded.value.result; + } + return yield* protocolFailure(); + }); + + const assertOk = (status: number) => + status === 200 + ? Effect.void + : Effect.fail( + status === 401 || status === 403 + ? revoked() + : new OrchestratorMcpFailure({ + code: "orchestration_error", + message: `The linked environment answered HTTP ${status}.`, + }), + ); + + const openSession = Effect.fnUntraced(function* ( + url: string, + token: string, + headers: Record, + ) { + const response = yield* post( + url, + token, + { + jsonrpc: "2.0", + id: 0, + method: "initialize", + params: { + protocolVersion: PROTOCOL_VERSION, + capabilities: {}, + clientInfo: { name: "t3code-peer", version: "1" }, + }, + }, + headers, + ); + yield* assertOk(response.status); + const initialized = yield* rpcResult(response, 0); + const result = yield* decodeInitializeResult(initialized).pipe( + Effect.mapError(protocolFailure), + ); + const sessionId = response.headers["mcp-session-id"]; + if (sessionId === undefined || result.protocolVersion !== PROTOCOL_VERSION) + return yield* protocolFailure(); + const notified = yield* post( + url, + token, + { + jsonrpc: "2.0", + method: "notifications/initialized", + }, + { ...headers, "mcp-session-id": sessionId, "mcp-protocol-version": PROTOCOL_VERSION }, + ); + if (notified.status !== 202 && notified.status !== 200) yield* assertOk(notified.status); + return sessionId; + }); + + const call: PeerMcpClient["Service"]["call"] = (input) => + Effect.gen(function* () { + // Even a cached MCP session must revalidate the peer before bearer use. + const resolved = yield* links + .resolve(input.environmentId) + .pipe(Effect.mapError(fromLinkError)); + const headers = { + [McpInvocationContext.MODE_LIMIT_HEADER]: `${input.limits.runtimeMode}/${input.limits.interactionMode}`, + }; + const key = `${encodeURIComponent(input.environmentId)}:${encodeURIComponent(resolved.url)}`; + const credential = yield* crypto + .digest("SHA-256", new TextEncoder().encode(resolved.token)) + .pipe(Effect.map(Hex.encode), Effect.orDie); + const cached = (yield* Ref.get(sessions)).get(key); + const sessionId = + cached?.credential === credential + ? cached.sessionId + : yield* openSession(resolved.url, resolved.token, headers); + const saveSession = (sessionId: string) => + Ref.update(sessions, (current) => new Map(current).set(key, { credential, sessionId })); + yield* saveSession(sessionId); + const request = { + jsonrpc: "2.0", + id: yield* Ref.getAndUpdate(nextRequestId, (id) => id + 1), + method: "tools/call", + params: { name: input.tool, arguments: input.arguments }, + }; + const callOnce = (sessionId: string) => + post(resolved.url, resolved.token, request, { + ...headers, + "mcp-session-id": sessionId, + "mcp-protocol-version": PROTOCOL_VERSION, + }); + let response = yield* callOnce(sessionId); + // A missing in-memory MCP session is the only safe transport replay. + if (response.status === 404) { + const renewed = yield* openSession(resolved.url, resolved.token, headers); + yield* saveSession(renewed); + response = yield* callOnce(renewed); + } + yield* assertOk(response.status); + const result = yield* rpcResult(response, request.id).pipe( + Effect.flatMap(decodeToolResult), + Effect.mapError((error) => (isMcpFailure(error) ? error : protocolFailure())), + ); + const text = result.content + .flatMap((part) => (part.type === "text" ? [part.text] : [])) + .join("\n"); + if (result.isError === true) { + const refusal = + result.structuredContent === undefined + ? decodeTextRefusal(text) + : decodeRefusal(result.structuredContent); + return yield* Option.isSome(refusal) + ? new OrchestratorMcpFailure({ + code: refusal.value.code, + message: refusal.value.message.slice(0, 1000), + }) + : new OrchestratorMcpFailure({ + code: "orchestration_error", + message: "The linked environment refused the call.", + }); + } + return yield* ( + result.structuredContent === undefined + ? // oxlint-disable-next-line t3code/no-inline-schema-compile -- The destination tool supplies its result schema per call. + Schema.decodeUnknownEffect(Schema.fromJsonString(input.success))(text) + : // oxlint-disable-next-line t3code/no-inline-schema-compile -- The destination tool supplies its result schema per call. + Schema.decodeUnknownEffect(input.success)(result.structuredContent) + ).pipe(Effect.mapError(protocolFailure)); + }).pipe( + Effect.catchTags({ + HttpClientError: () => + Effect.fail( + new OrchestratorMcpFailure({ + code: "orchestration_error", + message: "The linked environment stopped answering.", + }), + ), + }), + // Diagnostics must not turn a durable success into a failed mutation. + Effect.tapError((error) => + links + .recordOutcome(input.environmentId, { + error: `Peer call failed (${error.code}).`, + }) + .pipe(Effect.ignoreCause), + ), + Effect.tap(() => + links.recordOutcome(input.environmentId, { error: null }).pipe(Effect.ignoreCause), + ), + ); + + return PeerMcpClient.of({ call }); +}); + +export const layer = Layer.effect(PeerMcpClient, make); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 60e0267c38fb1..9b15a67653a5f 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -63,6 +63,8 @@ import * as TextGeneration from "./textGeneration/TextGeneration.ts"; import * as ProviderInstanceRegistryHydration from "./provider/ProviderInstanceRegistryHydration.ts"; import * as ProjectSettingsOverrideSweep from "./provider/ProjectSettingsOverrideSweep.ts"; import * as TerminalManager from "./terminal/Manager.ts"; +import * as PeerMcpClient from "./peer/PeerMcpClient.ts"; +import * as PeerForwarding from "./peer/PeerForwarding.ts"; import * as PeerLinks from "./peer/PeerLinks.ts"; import * as McpHttpServer from "./mcp/McpHttpServer.ts"; import * as McpSessionRegistry from "./mcp/McpSessionRegistry.ts"; @@ -707,6 +709,7 @@ const layerMakeRoutes = Layer.mergeAll( Layer.provide(layerPullRequestService), Layer.provide(ServerBrowser.layer.pipe(Layer.provide(DesktopBrowserChannel.layer))), Layer.provide(PreviewBrowser.layer), + Layer.provide(PeerForwarding.layer.pipe(Layer.provide(PeerMcpClient.layer))), Layer.provide(PeerLinks.layer), Layer.provide(PreviewAutomationBroker.layer), Layer.provide(ServerSelfUpdate.layer.pipe(Layer.provide(layerDesktopAppUpdate))), diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 9b9412af65420..77f551fa69e74 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -348,8 +348,9 @@ Pass the task in `message`. Project, model, and modes inherit when omitted; workspace does not. `scratch: true` launches without a project, in a folder of its own under the environment's Scratch project. For stacked PRs, use the parent branch as `baseRef` with `startFromOrigin: false`. The new thread may not run with broader runtime or -interaction modes than the caller. Launch has no retry key, so inspect existing threads after a failed or lost response before -launching again. `create_threads` remains the batch option for a shared checkout. +interaction modes than the caller. Pass a bounded `clientRequestId` for durable +retries; reuse the same key after a lost response, or omit it for a fresh launch. +`create_threads` remains the batch option for a shared checkout. ### `t3_thread_list` diff --git a/packages/client-runtime/src/t3ToolSummary.ts b/packages/client-runtime/src/t3ToolSummary.ts index 25878d4ff5bc7..fa4291eb0ec2e 100644 --- a/packages/client-runtime/src/t3ToolSummary.ts +++ b/packages/client-runtime/src/t3ToolSummary.ts @@ -327,6 +327,9 @@ export function summarizeT3ToolCalls( quantity(countEntities(entityIds("cwd")), "repository", "repositories"), ); break; + case "environment-links": + label = phrase("Listed", "list", `linked environments ${times}`); + break; case "environment-read": label = phrase("Checked", "check", `environment preferences ${times}`); break; diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 76511d12278b8..ad0d90db9d6ff 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -9,6 +9,13 @@ import { TrimmedNonEmptyString, } from "./baseSchemas.ts"; +export const OrchestratorMcpEnvironmentTarget = Schema.optional( + EnvironmentId.annotate({ + description: + "Linked destination environment from t3_environment_links. Omit or name this environment to act locally. Remote IDs belong to the destination; local caller targets are not inherited.", + }), +); + export const ORCHESTRATION_PROTOCOL_VERSION = 2; export const ORCHESTRATION_PROTOCOL_VERSION_TEXT = "2"; export const ORCHESTRATION_PROTOCOL_QUERY_PARAM = "orchestrationProtocol"; diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 778712ad23999..2b045e856c57c 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -4,6 +4,7 @@ import * as SchemaTransformation from "effect/SchemaTransformation"; import { ContextTransferId, + EnvironmentId, IsoDateTime, MessageId, NodeId, @@ -22,6 +23,7 @@ import { ScheduledTaskSchedule, ScheduledTaskUpsertSchedule, } from "./scheduledTask.ts"; +import { PeerLinkSummary } from "./peerLink.ts"; import { ProviderInteractionMode, RuntimeMode } from "./providerPolicy.ts"; import { PullRequestState } from "./pullRequest.ts"; import { @@ -43,13 +45,33 @@ import { } from "./model.ts"; import { ProviderDriverKind, ProviderInstanceId } from "./providerInstance.ts"; +import { OrchestratorMcpEnvironmentTarget } from "./environment.ts"; + +export { OrchestratorMcpEnvironmentTarget } from "./environment.ts"; + +export const OrchestratorMcpEnvironmentLinksResult = Schema.Struct({ + environmentId: EnvironmentId, + links: Schema.Array( + Schema.Struct({ + environmentId: PeerLinkSummary.fields.environmentId, + label: PeerLinkSummary.fields.label, + status: PeerLinkSummary.fields.status, + access: PeerLinkSummary.fields.access, + expiresAt: IsoDateTime, + lastError: PeerLinkSummary.fields.lastError, + }), + ), +}); +export type OrchestratorMcpEnvironmentLinksResult = + typeof OrchestratorMcpEnvironmentLinksResult.Type; + const OrchestratorMcpPrompt = TrimmedNonEmptyString.check(Schema.isMaxLength(120_000)).annotate({ description: "Complete task or message text for the target agent.", }); const OrchestratorMcpTitle = TrimmedNonEmptyString.check(Schema.isMaxLength(512)).annotate({ description: "Optional concise display title.", }); -const OrchestratorMcpClientRequestId = TrimmedNonEmptyString.check( +export const OrchestratorMcpClientRequestId = TrimmedNonEmptyString.check( Schema.isMaxLength(256), ).annotate({ description: "Stable idempotency key to reuse when retrying this mutation." }); @@ -296,6 +318,7 @@ const OrchestratorMcpProjectTarget = Schema.optional( ); export const OrchestratorMcpThreadListInput = Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, projectId: OrchestratorMcpProjectTarget, statuses: Schema.optional( Schema.Array(OrchestratorMcpThreadStatus).check(Schema.isMaxLength(10)), @@ -354,6 +377,7 @@ export const OrchestratorMcpThreadListResult = Schema.Struct({ export type OrchestratorMcpThreadListResult = typeof OrchestratorMcpThreadListResult.Type; export const OrchestratorMcpThreadReadInput = Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, threadId: ThreadId, itemId: Schema.optional(TurnItemId), textOffset: Schema.optional(NonNegativeInt), @@ -441,6 +465,7 @@ export const OrchestratorMcpThreadReadResult = Schema.Struct({ export type OrchestratorMcpThreadReadResult = typeof OrchestratorMcpThreadReadResult.Type; export const OrchestratorMcpThreadSendInput = Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, threadId: ThreadId, message: OrchestratorMcpPrompt, mode: Schema.optional(Schema.Literals(["auto", "queue", "steer", "restart"])), @@ -459,6 +484,7 @@ export const OrchestratorMcpThreadSendResult = Schema.Struct({ export type OrchestratorMcpThreadSendResult = typeof OrchestratorMcpThreadSendResult.Type; export const OrchestratorMcpThreadWaitInput = Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, threadId: ThreadId, runId: Schema.optional(RunId), timeoutMs: Schema.optional(Schema.Finite), @@ -474,6 +500,7 @@ export const OrchestratorMcpThreadWaitResult = Schema.Struct({ export type OrchestratorMcpThreadWaitResult = typeof OrchestratorMcpThreadWaitResult.Type; export const OrchestratorMcpThreadInterruptInput = Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, threadId: ThreadId, runId: Schema.optional(RunId), reason: Schema.optional(Schema.String.check(Schema.isMaxLength(2_000))), diff --git a/packages/contracts/src/peerLink.ts b/packages/contracts/src/peerLink.ts index a78069b08d2e4..ce06856312c22 100644 --- a/packages/contracts/src/peerLink.ts +++ b/packages/contracts/src/peerLink.ts @@ -16,7 +16,7 @@ export const PeerLink = Schema.Struct({ }); export type PeerLink = typeof PeerLink.Type; -/** Whether a link can be used now, as worked out when it is listed. */ +/** Descriptor reachability and local expiry; this does not authenticate the peer session. */ export const PeerLinkStatus = Schema.Literals(["reachable", "unreachable", "expired"]); export type PeerLinkStatus = typeof PeerLinkStatus.Type; diff --git a/packages/contracts/src/threadMetadataMcp.ts b/packages/contracts/src/threadMetadataMcp.ts index 7c132e4d45f4b..ff9bb4ec54a9e 100644 --- a/packages/contracts/src/threadMetadataMcp.ts +++ b/packages/contracts/src/threadMetadataMcp.ts @@ -8,6 +8,7 @@ import { ThreadId, TrimmedNonEmptyString, } from "./baseSchemas.ts"; +import { OrchestratorMcpEnvironmentTarget } from "./environment.ts"; import { ThreadLinkedPullRequest } from "./threadPullRequest.ts"; import { ThreadTitleRegeneration } from "./threadTitle.ts"; @@ -79,6 +80,7 @@ export const ThreadMetadataMcpPullRequest = Schema.Struct({ export type ThreadMetadataMcpPullRequest = typeof ThreadMetadataMcpPullRequest.Type; export const ThreadMetadataMcpUpdateInput = Schema.Struct({ + environmentId: OrchestratorMcpEnvironmentTarget, threadId: Schema.optional(ThreadId).annotate({ description: "Thread to update. Omit to update the calling thread.", }), diff --git a/packages/shared/src/t3McpToolPresentation.ts b/packages/shared/src/t3McpToolPresentation.ts index b18351cdd2c49..4096ac66e4db8 100644 --- a/packages/shared/src/t3McpToolPresentation.ts +++ b/packages/shared/src/t3McpToolPresentation.ts @@ -49,6 +49,7 @@ export type T3McpToolSummaryAction = | "project-update" | "project-delete" | "project-clone" + | "environment-links" | "environment-read" | "environment-update" | "attachment-prepare" @@ -305,6 +306,10 @@ const T3_MCP_TOOLS: Readonly> = { t3_worktree_list: tool(["List", "Listing", "Listed", "workspace branches"], "worktree-list"), t3_preview_list: tool(["List", "Listing", "Listed", "preview tabs"], "browser", "browser"), t3_preview_close: tool(["Close", "Closing", "Closed", "a preview tab"], "browser", "browser"), + t3_environment_links: tool( + ["List", "Listing", "Listed", "linked environments"], + "environment-links", + ), t3_environment_read: tool( ["Read", "Reading", "Read", "environment preferences"], "environment-read", From 887c1dfc38f9477b7ca6a4d92209e60424fea1f6 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 18:20:42 +0000 Subject: [PATCH 2/3] fix(server): preserve retry keys and refuse peer redirects --- .../src/mcp/toolkits/project/handlers.test.ts | 27 +++++- .../src/mcp/toolkits/project/handlers.ts | 5 ++ apps/server/src/peer/PeerForwarding.test.ts | 86 ++++++++++++++++++- apps/server/src/peer/PeerForwarding.ts | 3 +- apps/server/src/peer/PeerMcpClient.ts | 7 +- 5 files changed, 124 insertions(+), 4 deletions(-) diff --git a/apps/server/src/mcp/toolkits/project/handlers.test.ts b/apps/server/src/mcp/toolkits/project/handlers.test.ts index 29a8305358011..99b9e5a6582de 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.test.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.test.ts @@ -388,7 +388,12 @@ it.effect("a client launches at its ceiling with the project's default model", ( .handle("t3_thread_launch", params) .pipe(Stream.unwrap, Stream.runCollect, Effect.provide(dependencies)); - const result = yield* handle({ title: "Fix", projectId, message: "Fix the bug" }); + const result = yield* handle({ + title: "Fix", + projectId, + message: "Fix the bug", + clientRequestId: "fix-😀", + }); expect(result.at(-1)?.result).toMatchObject({ projectId, modelSelection }); expect(launched[0]?.runtimeMode).toBe("auto-accept-edits"); expect(launched[0]?.initialMessage?.senderThreadId).toBeUndefined(); @@ -399,6 +404,26 @@ it.effect("a client launches at its ceiling with the project's default model", ( const untargeted = yield* handle({ title: "Fix" }); expect(untargeted.at(-1)?.result).toMatchObject({ code: "target_required" }); expect(launched).toHaveLength(1); + + const invalidKey = yield* handle({ + title: "Fix", + projectId, + clientRequestId: "fix-\ud800", + }).pipe(Effect.exit); + expect(launched).toHaveLength(1); + expect(invalidKey).toMatchObject({ + _tag: "Success", + value: [ + { + isFailure: true, + result: { + _tag: "OrchestratorMcpFailure", + code: "invalid_request", + message: "clientRequestId must contain valid paired Unicode surrogates.", + }, + }, + ], + }); }), ); diff --git a/apps/server/src/mcp/toolkits/project/handlers.ts b/apps/server/src/mcp/toolkits/project/handlers.ts index aa7038ba648d0..cd32a83c58fb9 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.ts @@ -63,6 +63,11 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { Effect.gen(function* () { const context = yield* readCaller(); const { caller } = context; + if (input.clientRequestId !== undefined && !input.clientRequestId.isWellFormed()) + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "clientRequestId must contain valid paired Unicode surrogates.", + }); const commandId = input.clientRequestId === undefined ? yield* newCommandId() diff --git a/apps/server/src/peer/PeerForwarding.test.ts b/apps/server/src/peer/PeerForwarding.test.ts index 45a3f6403ab4d..f80d9c884fca0 100644 --- a/apps/server/src/peer/PeerForwarding.test.ts +++ b/apps/server/src/peer/PeerForwarding.test.ts @@ -368,7 +368,9 @@ const serveDestination = Effect.fnUntraced(function* ( const requests = yield* Ref.make>([]); const bearers = yield* Ref.make(0); const probes = yield* Ref.make(0); - const faults = yield* Ref.make>([]); + const faults = yield* Ref.make< + ReadonlyArray<"404" | "malformed" | "rpc-error" | { redirect: string }> + >([]); const middleware = HttpRouter.middleware( (effect) => Effect.gen(function* () { @@ -390,6 +392,8 @@ const serveDestination = Effect.fnUntraced(function* ( ]); if (method === "tools/call") { const fault = yield* Ref.modify(faults, (items) => [items[0], items.slice(1)]); + if (typeof fault === "object") + return HttpServerResponse.redirect(fault.redirect, { status: 307 }); if (fault === "404") return HttpServerResponse.empty({ status: 404 }); if (fault === "malformed") return HttpServerResponse.text("private upstream secret", { @@ -783,6 +787,43 @@ it.effect( runtimeMode: "auto", interactionMode: "default", }); + + const unicodeThreads: Array = []; + const threadsBeforeUnicode = yield* f.destination.threads.listProjectThreads({ + projectId: f.destination.project.id, + includeSubagents: false, + }); + // JSON strings preserve lone UTF-16 surrogates; replacing them with U+FFFD aliases keys. + for (const [caller, clientRequestId] of [ + [liveCaller, "\ud800"], + [otherCaller, "\ud800"], + [liveCaller, "\udc00"], + [liveCaller, "\ud801"], + [liveCaller, "\ufffd"], + ] as const) { + const input = { + environmentId: destinationId, + projectId: f.destination.project.id, + title: "Unicode retry key", + clientRequestId, + }; + const first = yield* f.call(ProjectToolkit.tools.t3_thread_launch, input, caller); + const saved = yield* f.destination.threads.getThreadProjection(first.threadId); + expect(yield* f.call(ProjectToolkit.tools.t3_thread_launch, input, caller)).toEqual( + first, + ); + expect(yield* f.destination.threads.getThreadProjection(first.threadId)).toEqual(saved); + expect(saved.messages).toEqual([]); + expect(saved.runs).toEqual([]); + expect(unicodeThreads).not.toContain(first.threadId); + unicodeThreads.push(first.threadId); + } + expect( + yield* f.destination.threads.listProjectThreads({ + projectId: f.destination.project.id, + includeSubagents: false, + }), + ).toHaveLength(threadsBeforeUnicode.length + unicodeThreads.length); }), ).pipe(Effect.provide(Layer.mergeAll(NodeHttpServer.layerTest, NodeServices.layer))), ); @@ -974,6 +1015,49 @@ it.effect( expect(refused.last_reached_at).toBe(reachedAt); expect(refused.last_error).toBe("Peer call failed (runtime_mode_escalation_denied)."); + const receiverContext = yield* Layer.build(Layer.fresh(NodeHttpServer.layerTest)); + const receiver = Context.get(receiverContext, HttpServer.HttpServer); + const received = yield* Ref.make< + ReadonlyArray<{ method: string; body: string; session: string | undefined }> + >([]); + yield* receiver.serve( + Effect.gen(function* () { + const request = yield* HttpServerRequest.HttpServerRequest; + const body = yield* request.text; + yield* Ref.update(received, (seen) => [ + ...seen, + { method: request.method, body, session: request.headers["mcp-session-id"] }, + ]); + return HttpServerResponse.empty({ status: 502 }); + }), + ); + if (receiver.address._tag === "UnixPathAddress") return yield* Effect.die("TCP expected"); + const beforeRedirect = yield* f.destination.threads.getThreadProjection(target); + expect(requests.map((request) => request.method)).toContain("notifications/initialized"); + yield* Ref.set(f.http.requests, []); + yield* Ref.set(f.http.probes, 0); + // This native listener is reachable, but its plain-HTTP origin is not an allowed peer. + yield* Ref.set(f.http.faults, [ + { redirect: `http://0.0.0.0:${receiver.address.port}/untrusted` }, + ]); + expect( + yield* f.invoke(OrchestratorToolkit.tools.t3_thread_send, { + ...read, + message: "synthetic body must stay at the paired origin", + clientRequestId: "redirected-write", + }), + ).toMatchObject({ isError: true, structuredContent: { code: "orchestration_error" } }); + expect(yield* Ref.get(f.http.probes)).toBe(1); + expect((yield* Ref.get(f.http.requests)).map((request) => request.method)).toEqual([ + "tools/call", + ]); + expect(yield* f.destination.threads.getThreadProjection(target)).toEqual(beforeRedirect); + expect + .soft(yield* Ref.get(received), "a tool POST must not follow a peer redirect") + .toEqual([]); + yield* f.call(OrchestratorToolkit.tools.t3_thread_read, read); + expect((yield* outcome())[0]?.last_error).toBeNull(); + yield* Ref.set(f.http.requests, []); yield* Ref.set(f.http.bearers, 0); yield* Ref.set(f.http.faults, ["404"]); diff --git a/apps/server/src/peer/PeerForwarding.ts b/apps/server/src/peer/PeerForwarding.ts index d30212590a41b..a6bc0cbab77f5 100644 --- a/apps/server/src/peer/PeerForwarding.ts +++ b/apps/server/src/peer/PeerForwarding.ts @@ -87,7 +87,8 @@ const make = Effect.gen(function* () { .digest( "SHA-256", new TextEncoder().encode( - `${encodeURIComponent(scope.requestNamespace)}:${encodeURIComponent(args.clientRequestId)}`, + // @effect-diagnostics-next-line preferSchemaOverJson:off -- Hash known strings without replacing lone UTF-16 surrogates. + JSON.stringify([scope.requestNamespace, args.clientRequestId]), ), ) .pipe( diff --git a/apps/server/src/peer/PeerMcpClient.ts b/apps/server/src/peer/PeerMcpClient.ts index 07ca3d0092b50..4c32409b51a66 100644 --- a/apps/server/src/peer/PeerMcpClient.ts +++ b/apps/server/src/peer/PeerMcpClient.ts @@ -14,6 +14,7 @@ import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import { McpSchema } from "effect/ai"; +import * as FetchHttpClient from "effect/http/FetchHttpClient"; import * as HttpClient from "effect/http/HttpClient"; import * as HttpClientRequest from "effect/http/HttpClientRequest"; import * as McpInvocationContext from "../mcp/McpInvocationContext.ts"; @@ -89,7 +90,11 @@ const fromLinkError = (error: PeerLinkError) => const make = Effect.gen(function* () { const links = yield* PeerLinks.PeerLinks; - const http = yield* HttpClient.HttpClient; + const http = (yield* HttpClient.HttpClient).pipe( + HttpClient.transform((request) => + request.pipe(Effect.provideService(FetchHttpClient.RequestInit, { redirect: "manual" })), + ), + ); const crypto = yield* Crypto.Crypto; const nextRequestId = yield* Ref.make(1); const sessions = yield* Ref.make< From 1e82c98b68a1c3191dd74ce341836c2438371495 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 19:09:38 +0000 Subject: [PATCH 3/3] fix(server): refuse ambiguous peer retries and pace waits --- apps/server/src/peer/PeerForwarding.test.ts | 168 +++++++++++++++----- apps/server/src/peer/PeerForwarding.ts | 9 +- apps/server/src/peer/PeerMcpClient.ts | 13 +- 3 files changed, 145 insertions(+), 45 deletions(-) diff --git a/apps/server/src/peer/PeerForwarding.test.ts b/apps/server/src/peer/PeerForwarding.test.ts index f80d9c884fca0..e57be48ba1474 100644 --- a/apps/server/src/peer/PeerForwarding.test.ts +++ b/apps/server/src/peer/PeerForwarding.test.ts @@ -36,6 +36,8 @@ import { McpSchema, McpServer } from "effect/ai"; import type * as Tool from "effect/ai/Tool"; import * as FetchHttpClient from "effect/http/FetchHttpClient"; import * as HttpRouter from "effect/http/HttpRouter"; +import * as HttpEffect from "effect/http/HttpEffect"; +import * as HttpClient from "effect/http/HttpClient"; import * as HttpServer from "effect/http/HttpServer"; import * as HttpServerRequest from "effect/http/HttpServerRequest"; import * as HttpServerResponse from "effect/http/HttpServerResponse"; @@ -369,7 +371,17 @@ const serveDestination = Effect.fnUntraced(function* ( const bearers = yield* Ref.make(0); const probes = yield* Ref.make(0); const faults = yield* Ref.make< - ReadonlyArray<"404" | "malformed" | "rpc-error" | { redirect: string }> + ReadonlyArray< + | "404" + | "malformed" + | "rpc-error" + | { redirect: string } + | { + afterDispatch: Deferred.Deferred; + resume: Deferred.Deferred; + status: Deferred.Deferred; + } + > >([]); const middleware = HttpRouter.middleware( (effect) => @@ -378,7 +390,7 @@ const serveDestination = Effect.fnUntraced(function* ( if (request.headers.authorization !== undefined) yield* Ref.update(bearers, (n) => n + 1); if (request.url.startsWith("/.well-known/t3/environment")) yield* Ref.update(probes, (n) => n + 1); - if (request.url !== "/mcp") return yield* effect; + if (request.url !== "/mcp" || request.method !== "POST") return yield* effect; const body = yield* request.json; const method = (body as { method?: string }).method; yield* Ref.update(requests, (seen) => [ @@ -392,8 +404,18 @@ const serveDestination = Effect.fnUntraced(function* ( ]); if (method === "tools/call") { const fault = yield* Ref.modify(faults, (items) => [items[0], items.slice(1)]); - if (typeof fault === "object") - return HttpServerResponse.redirect(fault.redirect, { status: 307 }); + if (typeof fault === "object") { + if ("redirect" in fault) + return HttpServerResponse.redirect(fault.redirect, { status: 307 }); + const response = yield* effect; + // Hold the committed reply before the transport's session check writes HTTP. + yield* HttpEffect.appendPreResponseHandler((_, response) => + Deferred.succeed(fault.status, response.status).pipe(Effect.as(response)), + ); + yield* Deferred.succeed(fault.afterDispatch, request.headers["mcp-session-id"]!); + yield* Deferred.await(fault.resume); + return response; + } if (fault === "404") return HttpServerResponse.empty({ status: 404 }); if (fault === "malformed") return HttpServerResponse.text("private upstream secret", { @@ -887,18 +909,18 @@ it.effect( expect(Number(first.args.timeoutMs)).toBeLessThanOrEqual(50_000); yield* TestClock.adjust(60_000); yield* Deferred.succeed(first.reply, response); - yield* TestClock.adjust(1); + yield* TestClock.adjust(100); const second = yield* Queue.take(requests); expect(second.args.runId).toBe(runId); expect(Number(second.args.timeoutMs)).toBeGreaterThan(0); expect(Number(second.args.timeoutMs)).toBeLessThanOrEqual(50_000); yield* TestClock.adjust(55_000); yield* Deferred.succeed(second.reply, { ...response, runId: RunId.make("successor") }); - yield* TestClock.adjust(1); + yield* TestClock.adjust(100); const third = yield* Queue.take(requests); expect(third.args.runId).toBe(runId); const finalRemaining = 120_000 - ((yield* Clock.currentTimeMillis) - startedAt); - expect(finalRemaining).toBe(4_998); + expect(finalRemaining).toBe(4_800); const finalTimeout = Number(third.args.timeoutMs); // The destination starts its timer after probe/init/transit, not at the local deadline. const overhead = 100; @@ -907,18 +929,43 @@ it.effect( const finalAnswer = yield* Fiber.join(waiting); yield* Deferred.await(third.finalized); - const quick = yield* wait(10).pipe(Effect.forkScoped); + const quickStartedAt = yield* Clock.currentTimeMillis; + const quickCompletedAt = yield* Deferred.make(); + const quick = yield* wait(10).pipe( + Effect.tap(() => Deferred.complete(quickCompletedAt, Clock.currentTimeMillis)), + Effect.forkScoped, + ); const quickFirst = yield* Queue.take(requests); + const quickCount = yield* Ref.get(calls); yield* Deferred.succeed(quickFirst.reply, response); yield* Deferred.await(quickFirst.finalized); - yield* TestClock.adjust(1); - const quickSecond = yield* Queue.take(requests); - expect(quickSecond.args.runId).toBe(runId); - expect(Number(quickSecond.args.timeoutMs)).toBeGreaterThan(0); - expect(Number(quickSecond.args.timeoutMs)).toBeLessThanOrEqual(9); yield* TestClock.adjust(9); + expect.soft(yield* Ref.get(calls), "no retry inside a 10ms budget").toBe(quickCount); + yield* TestClock.adjust(1); expect(yield* Fiber.join(quick)).toEqual(response); - yield* Deferred.await(quickSecond.finalized); + expect((yield* Deferred.await(quickCompletedAt)) - quickStartedAt).toBe(10); + expect.soft(yield* Ref.get(calls)).toBe(quickCount); + for (const remaining of yield* Queue.clear(requests)) { + yield* Deferred.await(remaining.finalized); + } + + const repeated = yield* wait(1_000).pipe(Effect.forkScoped); + let timedOutRequest = yield* Queue.take(requests); + for (let retry = 0; retry < 2; retry++) { + const beforeRetry = yield* Ref.get(calls); + yield* Deferred.succeed(timedOutRequest.reply, response); + yield* Deferred.await(timedOutRequest.finalized); + yield* TestClock.adjust(99); + expect + .soft(yield* Ref.get(calls), "quick timeouts need at least 100ms backoff") + .toBe(beforeRetry); + yield* TestClock.adjust(1); + timedOutRequest = yield* Queue.take(requests); + expect(yield* Ref.get(calls)).toBe(beforeRetry + 1); + expect(timedOutRequest.args.runId).toBe(runId); + } + yield* Deferred.succeed(timedOutRequest.reply, { ...response, timedOut: false }); + expect(yield* Fiber.join(repeated)).toEqual({ ...response, timedOut: false }); const noResponse = yield* wait(20).pipe(Effect.result, Effect.forkScoped); const pending = yield* Queue.take(requests); @@ -970,7 +1017,7 @@ it.effect( ); it.effect( - "renews MCP sessions once, revalidates repaired credentials, and persists safe best-effort outcomes", + "refuses ambiguous session loss, recovers on explicit retry, and persists safe best-effort outcomes", () => Effect.scoped( Effect.gen(function* () { @@ -1058,34 +1105,83 @@ it.effect( yield* f.call(OrchestratorToolkit.tools.t3_thread_read, read); expect((yield* outcome())[0]?.last_error).toBeNull(); + const terminatingTarget = yield* f.destination.seed("terminated-session"); + const afterDispatch = yield* Deferred.make(); + const resume = yield* Deferred.make(); + const status = yield* Deferred.make(); yield* Ref.set(f.http.requests, []); - yield* Ref.set(f.http.bearers, 0); - yield* Ref.set(f.http.faults, ["404"]); - const send = { ...read, message: "renew once", clientRequestId: "renew" }; + yield* Ref.set(f.http.faults, [{ afterDispatch, resume, status }]); + const inFlight = yield* f + .invoke(OrchestratorToolkit.tools.t3_thread_send, { + environmentId: destinationId, + threadId: terminatingTarget, + message: "unkeyed mutation survives session termination", + }) + .pipe(Effect.forkScoped); + const terminatedSession = yield* Deferred.await(afterDispatch); + const committed = yield* f.destination.threads.getThreadProjection(terminatingTarget); + expect(committed.messages).toHaveLength(1); + expect(committed.runs).toHaveLength(1); + const resolved = yield* f.links.resolve(destinationId); + const deleted = yield* HttpClient.del(`${f.http.url}/mcp`, { + headers: { + authorization: `Bearer ${resolved.token}`, + "mcp-session-id": terminatedSession, + "mcp-protocol-version": "2025-06-18", + }, + }).pipe(Effect.provide(FetchHttpClient.layer)); + expect(deleted.status).toBe(204); + yield* Deferred.succeed(resume, undefined); + expect(yield* Deferred.await(status), "native transport response after DELETE").toBe(404); + const terminatedResult = yield* Fiber.join(inFlight); + expect.soft(terminatedResult).toMatchObject({ + isError: true, + structuredContent: { code: "orchestration_error" }, + }); + requests = yield* Ref.get(f.http.requests); + expect + .soft( + requests.filter((r) => r.method === "tools/call"), + "a dispatched mutation must not be silently replayed", + ) + .toHaveLength(1); + expect.soft(requests.filter((r) => r.method === "initialize")).toHaveLength(0); + expect + .soft(yield* f.destination.threads.getThreadProjection(terminatingTarget)) + .toEqual(committed); + + yield* Ref.set(f.http.requests, []); + yield* f.call(OrchestratorToolkit.tools.t3_thread_read, { + environmentId: destinationId, + threadId: terminatingTarget, + }); + requests = yield* Ref.get(f.http.requests); + expect + .soft( + requests.filter((r) => r.method === "initialize"), + "the next explicit call must open a fresh session", + ) + .toHaveLength(1); + expect(requests.filter((r) => r.method === "tools/call")).toHaveLength(1); + expect(requests.find((r) => r.method === "tools/call")?.session).not.toBe( + terminatedSession, + ); + + const send = { ...read, message: "explicit keyed retry", clientRequestId: "renew" }; const accepted = yield* f.call(OrchestratorToolkit.tools.t3_thread_send, send); + yield* Ref.set(f.http.requests, []); + yield* Ref.set(f.http.faults, ["404"]); + expect + .soft(yield* f.invoke(OrchestratorToolkit.tools.t3_thread_send, send)) + .toMatchObject({ isError: true, structuredContent: { code: "orchestration_error" } }); requests = yield* Ref.get(f.http.requests); - expect(requests.filter((r) => r.method === "initialize")).toHaveLength(1); - const retried = requests.filter((r) => r.method === "tools/call"); - expect(retried).toHaveLength(2); - expect(retried[0]?.body).toEqual(retried[1]?.body); - expect(retried[0]?.session).not.toBe(retried[1]?.session); + expect.soft(requests.filter((r) => r.method === "initialize")).toHaveLength(0); + expect.soft(requests.filter((r) => r.method === "tools/call")).toHaveLength(1); + expect((yield* f.destination.threads.getThreadProjection(target)).messages).toHaveLength(1); expect(yield* f.call(OrchestratorToolkit.tools.t3_thread_send, send)).toEqual(accepted); expect((yield* f.destination.threads.getThreadProjection(target)).messages).toHaveLength(1); expect((yield* outcome())[0]?.last_error).toBeNull(); - yield* Ref.set(f.http.requests, []); - yield* Ref.set(f.http.bearers, 0); - yield* Ref.set(f.http.faults, ["404", "404"]); - expect( - yield* f.invoke(OrchestratorToolkit.tools.t3_thread_send, { - ...send, - clientRequestId: "two404", - }), - ).toMatchObject({ isError: true, structuredContent: { code: "orchestration_error" } }); - requests = yield* Ref.get(f.http.requests); - expect(requests.filter((r) => r.method === "initialize")).toHaveLength(1); - expect(requests.filter((r) => r.method === "tools/call")).toHaveLength(2); - expect((yield* f.destination.threads.getThreadProjection(target)).messages).toHaveLength(1); for (const fault of ["malformed", "rpc-error"] as const) { yield* Ref.set(f.http.faults, [fault]); const failed = yield* f.invoke(OrchestratorToolkit.tools.t3_thread_read, read); diff --git a/apps/server/src/peer/PeerForwarding.ts b/apps/server/src/peer/PeerForwarding.ts index a6bc0cbab77f5..f179188f21af4 100644 --- a/apps/server/src/peer/PeerForwarding.ts +++ b/apps/server/src/peer/PeerForwarding.ts @@ -143,10 +143,11 @@ const make = Effect.gen(function* () { if (Option.isNone(result)) return yield* expired(); last = result.value; if (pinnedRun === undefined && last.runId !== null) pinnedRun = last.runId; - if (!last.timedOut || deadline - (yield* Clock.currentTimeMillis) <= headroom) return last; - // A peer may answer a timeout immediately. Yield until time advances - // rather than spin indefinitely while spending no deadline budget. - yield* Effect.sleep(1); + if (!last.timedOut) return last; + const remainingAfterReply = deadline - (yield* Clock.currentTimeMillis); + if (remainingAfterReply <= headroom) return last; + // A peer may time out immediately; pace continuations within the absolute deadline. + yield* Effect.sleep(Math.min(100, remainingAfterReply)); } }); diff --git a/apps/server/src/peer/PeerMcpClient.ts b/apps/server/src/peer/PeerMcpClient.ts index 4c32409b51a66..8c65437ed76f5 100644 --- a/apps/server/src/peer/PeerMcpClient.ts +++ b/apps/server/src/peer/PeerMcpClient.ts @@ -253,12 +253,15 @@ const make = Effect.gen(function* () { "mcp-session-id": sessionId, "mcp-protocol-version": PROTOCOL_VERSION, }); - let response = yield* callOnce(sessionId); - // A missing in-memory MCP session is the only safe transport replay. + const response = yield* callOnce(sessionId); if (response.status === 404) { - const renewed = yield* openSession(resolved.url, resolved.token, headers); - yield* saveSession(renewed); - response = yield* callOnce(renewed); + yield* Ref.update(sessions, (current) => { + const cached = current.get(key); + if (cached?.credential !== credential || cached.sessionId !== sessionId) return current; + const updated = new Map(current); + updated.delete(key); + return updated; + }); } yield* assertOk(response.status); const result = yield* rpcResult(response, request.id).pipe(