,
-) =>
- declare((params: P) =>
+) => {
+ const started = (params: P) =>
Effect.gen(function* () {
const { limits } = yield* writingCaller;
const requested = modes(params);
- const started: StartedModes = {
+ return {
runtimeMode: yield* resolveRuntimeMode(limits.runtimeMode, requested.runtimeMode),
interactionMode: yield* resolveInteractionMode(
limits.interactionMode,
requested.interactionMode,
),
- };
- return yield* handle(params, started);
- }),
+ } satisfies StartedModes;
+ });
+ return declare(
+ (params: P) => Effect.flatMap(started(params), (modes) => handle(params, modes)),
+ started,
);
+};
const fullAccessRequired =
"Changing projects or environment settings needs a live full-access/default calling thread or a full-access client.";
@@ -209,7 +248,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,
+ );
};
type CheckServices = Effect.Services;
@@ -246,14 +288,74 @@ 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 = PeerForwarding.remoteTarget(scope, environmentId);
+ if (target === undefined) return yield* here(params);
+ yield* (remote as RemoteCheck)(params);
+ if (Option.isNone(forwarding)) return yield* linksUnavailable;
+ return yield* forwarding.value.call(scope, tool, target, { ...params, environmentId });
+ });
+ }
+ return handlers as Toolkit.HandlersFrom;
+ });
+
export const toLayer = , EX = never, RX = never>(
toolkit: Toolkit.Toolkit,
build: Handlers | Effect.Effect, EX, RX>,
-): HandlersLayer> =>
- 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/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts
index b14c89b2da106..0831236569f46 100644
--- a/apps/server/src/mcp/OrchestratorMcpService.ts
+++ b/apps/server/src/mcp/OrchestratorMcpService.ts
@@ -2685,6 +2685,54 @@ const make = Effect.gen(function* () {
yield* resolveRuntimeMode(limits.runtimeMode, target.thread.runtimeMode);
yield* resolveInteractionMode(limits.interactionMode, target.thread.interactionMode);
const key = yield* requestKey(input.clientRequestId);
+ // A peer from before `stop` drops it when decoding, so it interrupts
+ // the run instead, as this call did then.
+ if (input.stop === true) {
+ const commandId = stableCommandId({ scope, requestKey: key, operation: "thread-stop" });
+ // The cause goes to the log; the caller learns only that the stop failed.
+ const stopFailed = (error: unknown) =>
+ Effect.logWarning("orchestrator-mcp.thread-stop-failed", {
+ threadId: input.threadId,
+ error,
+ }).pipe(
+ Effect.andThen(
+ Effect.fail(
+ failure("orchestration_error", `Unable to stop thread ${input.threadId}.`),
+ ),
+ ),
+ );
+ const stopped = yield* threadManagement
+ .dispatch({
+ type: "thread.stop",
+ commandId,
+ threadId: input.threadId,
+ ...(input.reason === undefined ? {} : { reason: input.reason }),
+ })
+ .pipe(Effect.catch(stopFailed));
+ yield* threadManagement
+ .stopDelegatedTasks({ threadId: input.threadId, commandId, reason: input.reason })
+ .pipe(Effect.catch(stopFailed));
+ // Queue holds and completion disposal also update runs. Only the
+ // original interrupt request identifies the run Stop reached.
+ const request = stopped.storedEvents.find(
+ (stored) =>
+ stored.event.type === "turn-item.updated" &&
+ stored.event.payload.type === "run_interrupt_request" &&
+ (stored.event.payload.providerTurnId !== null ||
+ stopped.storedEvents.some(
+ (updated) =>
+ updated.event.type === "run.updated" &&
+ updated.event.runId === stored.event.runId &&
+ updated.event.payload.status === "interrupted",
+ )),
+ );
+ const interrupted = request?.event.runId;
+ return {
+ threadId: input.threadId,
+ runId: interrupted ?? null,
+ status: interrupted === undefined ? "no_active_run" : "interrupt_requested",
+ } satisfies OrchestratorMcpThreadInterruptResult;
+ }
const result = yield* threadManagement
.interruptThread({
projectId: target.thread.projectId,
diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts
index c7b5fc245094f..94a5f4aa8f96b 100644
--- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts
+++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts
@@ -80,6 +80,7 @@ import * as SecretRequests from "../secrets/SecretRequests.ts";
import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
import * as McpHttpServer from "./McpHttpServer.ts";
import * as McpInvocationContext from "./McpInvocationContext.ts";
+import * as PeerForwarding from "../peer/PeerForwarding.ts";
import { delegatedTaskRun, hasPendingChildRuns } from "./OrchestratorMcpService.ts";
import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
@@ -697,6 +698,7 @@ describe("orchestrator MCP toolkit", () => {
Layer.provide(layerProviderRegistry),
Layer.provide(ServerSettings.layerTest()),
Layer.provide(layerScheduledTaskStub),
+ Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})),
Layer.provide(
Layer.mock(ProjectService.ProjectService)({
@@ -4035,6 +4037,7 @@ describe("orchestrator MCP toolkit", () => {
Layer.provide(layerProviderRegistry),
Layer.provide(ServerSettings.layerTest()),
Layer.provide(layerUnusedScheduledTaskStub),
+ Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
Layer.provideMerge(
SecretRequests.layer.pipe(
diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts
index aaa5b176db468..91f722cdbc98f 100644
--- a/apps/server/src/mcp/toolkits/core.test.ts
+++ b/apps/server/src/mcp/toolkits/core.test.ts
@@ -59,6 +59,7 @@ import * as EnvironmentHandlers from "./environment/handlers.ts";
import { ProjectToolkit } from "./project/tools.ts";
import { AttachmentToolkit } from "./attachment/tools.ts";
import * as AttachmentHandlers from "./attachment/handlers.ts";
+import * as PeerForwarding from "../../peer/PeerForwarding.ts";
import { ThreadToolkit } from "./thread/tools.ts";
import { WorktreeToolkit } from "./worktree/tools.ts";
import { DeviceToolkit } from "./device/tools.ts";
@@ -89,6 +90,7 @@ const decodeThreadReadResult = Schema.decodeUnknownEffect(OrchestratorMcpThreadR
const layerThreadToolkit = McpHttpServer.layerThreadToolkit.pipe(
Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})),
Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
+ Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
);
it("publishes unique tool names with reference-free object-root inputs", () => {
@@ -715,6 +717,7 @@ it.effect("refuses act-as-caller tools to a client caller", () =>
Layer.provide(Layer.mock(ProviderRegistry.ProviderRegistry)({})),
Layer.provide(Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({})),
Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
+ Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
),
@@ -770,6 +773,7 @@ it.effect("a caller cannot rewrite a scheduled task that runs above its own mode
),
Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
+ Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
),
),
),
@@ -854,6 +858,7 @@ it.effect.each(["t3_thread_interrupt", "t3_workflow_stop"])(
Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
+ Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
),
),
),
@@ -944,6 +949,7 @@ it.effect("discovers workflow stop targets and requires orchestration capability
Layer.provide(Layer.mock(ProviderRegistry.ProviderRegistry)({})),
Layer.provide(Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({})),
Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
+ Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
),
diff --git a/apps/server/src/mcp/toolkits/environment/tools.ts b/apps/server/src/mcp/toolkits/environment/tools.ts
index e441174900ce1..d2e9a51bb124a 100644
--- a/apps/server/src/mcp/toolkits/environment/tools.ts
+++ b/apps/server/src/mcp/toolkits/environment/tools.ts
@@ -2,6 +2,7 @@ import {
BackgroundActivityProfile,
BackgroundActivityProfileSelection,
ExecutionEnvironmentDescriptor,
+ OrchestratorMcpEnvironmentTarget,
OrchestratorMcpFailure,
ServerSettings,
ServerSettingsPatch,
@@ -41,6 +42,7 @@ const EnvironmentReadTool = Tool.make("t3_environment_read", {
...shared,
description:
"Read this server's identity and selected environment preferences. Provider/model availability is exposed by orchestrator_capabilities. Writing instructions are limited to 4,000 characters.",
+ parameters: Schema.Struct({ environmentId: OrchestratorMcpEnvironmentTarget }),
success: Schema.Struct({
environmentId: ExecutionEnvironmentDescriptor.fields.environmentId,
label: Schema.String,
@@ -56,6 +58,7 @@ const EnvironmentPreferencesTool = Tool.make("t3_environment_preferences_update"
description:
"Update selected environment-wide preferences through normal settings persistence and notifications. Requires a live full-access/default calling thread. Omitted fields are preserved; empty customInstructions clears them.",
parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
defaultThreadEnvMode: ServerSettingsPatch.fields.defaultThreadEnvMode,
newWorktreesStartFromOrigin: ServerSettingsPatch.fields.newWorktreesStartFromOrigin,
enableProviderUpdateChecks: ServerSettingsPatch.fields.enableProviderUpdateChecks,
diff --git a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts
index 7c05b330047cf..2d24d869695e2 100644
--- a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts
+++ b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts
@@ -2,6 +2,7 @@ import { OrchestratorMcpFailure } from "@t3tools/contracts";
import { OrchestratorToolkit } from "./tools.ts";
import * as Effect from "effect/Effect";
+import * as PeerForwarding from "../../../peer/PeerForwarding.ts";
import { readCaller } from "../../threadAccess.ts";
import * as McpInvocationContext from "../../McpInvocationContext.ts";
import * as McpToolAccess from "../../McpToolAccess.ts";
@@ -140,6 +141,13 @@ const handlers = {
return yield* service.waitForThread(scope, input);
}),
),
+ t3_environment_links: McpToolAccess.reads(() =>
+ Effect.gen(function* () {
+ const scope = yield* McpInvocationContext.McpInvocationContext;
+ const forwarding = yield* PeerForwarding.PeerForwarding;
+ return yield* forwarding.links(scope);
+ }),
+ ),
t3_thread_interrupt: McpToolAccess.writesThreads(
(input) => [input.threadId],
(input) =>
diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts
index b8f735069d31c..b587f60ffc35c 100644
--- a/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts
+++ b/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts
@@ -1,5 +1,7 @@
+import { EnvironmentId } from "@t3tools/contracts";
import { assert, describe, it } from "@effect/vitest";
import { Tool } from "effect/ai";
+import * as Schema from "effect/Schema";
import {
CreateThreadsTool,
@@ -80,6 +82,7 @@ describe("orchestrator MCP tool guidance", () => {
assert.equal(schema.type, "object");
assert.hasAllKeys(schema.properties ?? {}, [
+ "environmentId",
"threadId",
"action",
"title",
@@ -87,5 +90,14 @@ describe("orchestrator MCP tool guidance", () => {
"clientRequestId",
]);
assert.include(ThreadUpdateTool.description ?? "", "Workspace and branch changes");
+ const decode = Schema.decodeUnknownSync(ThreadUpdateTool.parametersSchema);
+ assert.throws(() => decode({ action: "regenerate_title", title: "Requested title" }));
+ assert.throws(() => decode({ action: "rename" }));
+ const environmentId = EnvironmentId.make("peer-environment");
+ assert.deepEqual(decode({ action: "rename", title: "Requested title", environmentId }), {
+ action: "rename",
+ title: "Requested title",
+ environmentId,
+ });
});
});
diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts
index f0dabc0260e37..379684e989da4 100644
--- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts
+++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts
@@ -1,12 +1,15 @@
import {
+ OrchestratorMcpCapabilitiesInput,
OrchestratorMcpCapabilitiesResult,
OrchestrationV2StopWorkflowInput,
+ OrchestratorMcpEnvironmentLinksResult,
OrchestratorMcpCreateThreadsInput,
OrchestratorMcpCreateThreadsResult,
OrchestratorMcpDelegateTaskInput,
OrchestratorMcpDelegateTaskResult,
OrchestratorMcpDeleteScheduledTaskInput,
OrchestratorMcpDeleteScheduledTaskResult,
+ OrchestratorMcpEnvironmentTarget,
OrchestratorMcpRequestSecretInput,
OrchestratorMcpRequestSecretResult,
OrchestratorMcpFailure,
@@ -35,6 +38,7 @@ import { Tool, Toolkit } from "effect/ai";
import * as Schema from "effect/Schema";
import * as ThreadManagementService from "../../../orchestration-v2/ThreadManagementService.ts";
+import * as PeerForwarding from "../../../peer/PeerForwarding.ts";
import * as McpInvocationContext from "../../McpInvocationContext.ts";
import * as OrchestratorMcpService from "../../OrchestratorMcpService.ts";
import * as ThreadMetadataMcpService from "../../ThreadMetadataMcpService.ts";
@@ -44,6 +48,7 @@ const dependencies = [
ThreadManagementService.ThreadManagementService,
OrchestratorMcpService.OrchestratorMcpService,
];
+const linkDependencies = [...dependencies, PeerForwarding.PeerForwarding];
const threadMetadataDependencies = [
McpInvocationContext.McpInvocationContext,
ThreadManagementService.ThreadManagementService,
@@ -52,7 +57,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 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. Pass environmentId to list a linked environment's providers instead.",
+ parameters: OrchestratorMcpCapabilitiesInput,
success: OrchestratorMcpCapabilitiesResult,
failure: OrchestratorMcpFailure,
failureMode: "return",
@@ -105,7 +111,10 @@ const TaskCancelTool = Tool.make("task_cancel", {
const WorkflowStopTool = Tool.make("t3_workflow_stop", {
description:
"Stop one running Claude workflow in a T3 thread. Use the workflow coordinator's subagentId returned by t3_thread_read with view='activity'. Stops the whole workflow, including its phases and agents. Does not stop the parent thread or other workflows. Individual workflow phases and agents cannot be stopped separately.",
- parameters: OrchestrationV2StopWorkflowInput,
+ parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
+ ...OrchestrationV2StopWorkflowInput.fields,
+ }),
success: Schema.Struct({}),
failure: OrchestratorMcpFailure,
failureMode: "return",
@@ -116,7 +125,7 @@ const WorkflowStopTool = Tool.make("t3_workflow_stop", {
export const ScheduleTaskTool = Tool.make("schedule_task", {
description:
- "Create persistent work in the app scheduler that runs even when no turn is active. Pass schedule as a STRUCTURED OBJECT, never JSON text. Timers: {type:'interval', everyMs:3600000} is hourly; {type:'fixed_time', timeOfDay:'09:00', weekdays:[1,2,3,4,5]} is weekday mornings; report the returned nextRunAt. Webhooks: {type:'webhook'} runs once per request to a generated URL. The run sees the request ONLY through prompt placeholders: {{body.path}} (e.g. {{body.action}}, {{body.release.tag_name}}), {{headers.name}}, {{query.name}}, {{body}}, or {{request}} (method, headers with credentials redacted, and body). Each value arrives framed as untrusted request data, and $skill, @file or /command mentions inside it do not act; without a signature, anyone with the URL can trigger a run. For a sender that signs requests, first call request_secret so the user enters the secret privately (never ask for it in chat or invent one), then set signature with the returned secretRef, e.g. GitHub: {type:'webhook', signature:{header:'x-hub-signature-256', encoding:'hex', prefix:'sha256=', secretRef}}. The result's webhookUrl is the public URL to give the user; if it is absent, this environment has no T3 Connect managed tunnel, so tell the user to enable T3 Connect remote access rather than sharing a path. Omit projectId for this thread's project. In this thread's project, runs post into THIS thread by default (bindToCurrentThread=true), which suits an orchestrator that sees every trigger, delegates work, and can dedupe against what is in flight; use false only when the user wants a fresh top-level thread per run. Elsewhere each run launches a fresh thread. Provider, model, and runtime settings inherit from this thread, or from the project default when there is no calling thread.",
+ "Create persistent work in the app scheduler that runs even when no turn is active. Pass schedule as a STRUCTURED OBJECT, never JSON text. Timers: {type:'interval', everyMs:3600000} is hourly; {type:'fixed_time', timeOfDay:'09:00', weekdays:[1,2,3,4,5]} is weekday mornings; report the returned nextRunAt. Webhooks: {type:'webhook'} runs once per request to a generated URL. The run sees the request ONLY through prompt placeholders: {{body.path}} (e.g. {{body.action}}, {{body.release.tag_name}}), {{headers.name}}, {{query.name}}, {{body}}, or {{request}} (method, headers with credentials redacted, and body). Each value arrives framed as untrusted request data, and $skill, @file or /command mentions inside it do not act; without a signature, anyone with the URL can trigger a run. Secret references work only in this environment; to configure a signed webhook in a linked environment, use a thread there. For a sender that signs requests, first call request_secret so the user enters the secret privately (never ask for it in chat or invent one), then set signature with the returned secretRef, e.g. GitHub: {type:'webhook', signature:{header:'x-hub-signature-256', encoding:'hex', prefix:'sha256=', secretRef}}. The result's webhookUrl is the public URL to give the user; if it is absent, this environment has no T3 Connect managed tunnel, so tell the user to enable T3 Connect remote access rather than sharing a path. Omit projectId for this thread's project. In this thread's project, runs post into THIS thread by default (bindToCurrentThread=true), which suits an orchestrator that sees every trigger, delegates work, and can dedupe against what is in flight; use false only when the user wants a fresh top-level thread per run. Elsewhere each run launches a fresh thread. Provider, model, and runtime settings inherit from this thread, or from the project default when there is no calling thread.",
parameters: OrchestratorMcpScheduleTaskInput,
success: OrchestratorMcpScheduleTaskResult,
failure: OrchestratorMcpFailure,
@@ -143,7 +152,7 @@ const ListScheduledTasksTool = Tool.make("list_scheduled_tasks", {
const UpdateScheduledTaskTool = Tool.make("update_scheduled_task", {
description:
- "Update an existing scheduled task by scheduledTaskId (from list_scheduled_tasks). Only the provided fields change; omit a field to leave it as-is. Use enabled=false to pause a task without deleting it. Set bindToCurrentThread to move the task between posting into this thread and launching a fresh thread per run.",
+ "Update an existing scheduled task by scheduledTaskId (from list_scheduled_tasks). Only the provided fields change; omit a field to leave it as-is. To set or replace signature.secretRef in a linked environment, use a thread there; secret references work only in this environment. Use enabled=false to pause a task without deleting it. Set bindToCurrentThread to move the task between posting into this thread and launching a fresh thread per run.",
parameters: OrchestratorMcpUpdateScheduledTaskInput,
success: OrchestratorMcpScheduleTaskResult,
failure: OrchestratorMcpFailure,
@@ -192,7 +201,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, newest first. Omit projectId for the calling thread's project. Pass environmentId to list a linked environment's threads; projectId is then required. Filter by durable run status, title, or settled state (settled=true lists threads the user or auto-settlement moved out of the active list), or snoozed state, and paginate with the returned cursor. A snoozed thread wakes early 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 in this environment for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread's current title. Name a thread in a linked environment by title instead: its link would open here.",
parameters: OrchestratorMcpThreadListInput,
success: OrchestratorMcpThreadListResult,
failure: OrchestratorMcpFailure,
@@ -206,7 +215,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 in this environment, or in a linked one with environmentId. The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. Reading an untruncated terminal assistant result from this parent thread's direct app-owned child acknowledges that child's automatic completion delivery. Continue with afterPosition=nextPosition. Recover long item text with itemId and textOffset=nextTextOffset until nextTextOffset is null; offsets count UTF-16 code units. The thread also reports its snooze state and, in pullRequests, every linked pull request; linkedPullRequest is a legacy single link and can be null while pullRequests still has entries. To link a thread in this environment for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread's current title. Name a thread in a linked environment by title instead: its link would open here.",
parameters: OrchestratorMcpThreadReadInput,
success: OrchestratorMcpThreadReadResult,
failure: OrchestratorMcpFailure,
@@ -222,7 +231,10 @@ 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.",
- parameters: ThreadMetadataMcpUpdateInput,
+ parameters: ThreadMetadataMcpUpdateInput.mapFields(
+ (fields) => ({ environmentId: OrchestratorMcpEnvironmentTarget, ...fields }),
+ { unsafePreserveChecks: true },
+ ),
success: ThreadMetadataMcpUpdateResult,
failure: OrchestratorMcpFailure,
failureMode: "return",
@@ -234,7 +246,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 in this environment, or in a linked one with environmentId. The target cannot have broader permission modes than the caller. Use delegate_task with the full review context and a new clientRequestId for independent review rounds. For a follow-up that needs the same child's conversation, send to its childThreadId. After the original task completes, a requested follow-up creates a fresh task in that child thread and returns taskId when it starts a turn or restarts a turn without its own task; its completion notifies the parent, and the original result stays unchanged. A steer that joins a running turn returns that turn's task, if it has one. mode='auto' starts an idle thread, steers a fully active turn, or queues behind a turn that is not yet steerable. Use queue for a separate follow-up turn, steer for an in-flight update, or restart to interrupt-and-restart the active turn. clientRequestId makes retries idempotent.",
parameters: OrchestratorMcpThreadSendInput,
success: OrchestratorMcpThreadSendResult,
failure: OrchestratorMcpFailure,
@@ -247,7 +259,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 to reach a terminal durable state, here or in a linked environment with environmentId. Without runId, the latest run at call time is selected; an idle thread returns immediately. Timeout does not interrupt work, so call again or use t3_thread_read/list after timedOut=true. Waiting reports status only and does not acknowledge a delegated result.",
parameters: OrchestratorMcpThreadWaitInput,
success: OrchestratorMcpThreadWaitResult,
failure: OrchestratorMcpFailure,
@@ -261,7 +273,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 in this environment, or in a linked one with environmentId. Without runId, the newest interruptible run is selected. Terminal runs and threads without an active turn return without another side effect. stop=true stops the thread as its Stop button does instead: it also holds queued messages, ends pull request watches and stops the tasks it delegated. clientRequestId makes retries idempotent.",
parameters: OrchestratorMcpThreadInterruptInput,
success: OrchestratorMcpThreadInterruptResult,
failure: OrchestratorMcpFailure,
@@ -271,6 +283,19 @@ const ThreadInterruptTool = Tool.make("t3_thread_interrupt", {
.annotate(Tool.Title, "Interrupt a T3 thread")
.annotate(Tool.Destructive, true);
+const EnvironmentLinksTool = Tool.make("t3_environment_links", {
+ description:
+ "List the other T3 Code environments this one is linked to, such as the user's other machines, and whether each answers now. Pass one's environmentId to any tool that takes environmentId, such as t3_thread_launch, t3_thread_send, t3_pending_request_respond, t3_queue_list or t3_project_list, to act there. A linked environment lets this one's agents do at most its access there, and never more than the calling agent may do here. An expired link has to be linked again from a new pairing code.",
+ success: OrchestratorMcpEnvironmentLinksResult,
+ failure: OrchestratorMcpFailure,
+ failureMode: "return",
+ dependencies: linkDependencies,
+})
+ .annotate(Tool.Title, "List linked environments")
+ .annotate(Tool.Readonly, true)
+ .annotate(Tool.Destructive, false)
+ .annotate(Tool.Idempotent, true);
+
export const OrchestratorToolkit = Toolkit.make(
OrchestratorCapabilitiesTool,
DelegateTaskTool,
@@ -289,4 +314,5 @@ export const OrchestratorToolkit = Toolkit.make(
ThreadSendTool,
ThreadWaitTool,
ThreadInterruptTool,
+ EnvironmentLinksTool,
);
diff --git a/apps/server/src/mcp/toolkits/project/handlers.test.ts b/apps/server/src/mcp/toolkits/project/handlers.test.ts
index 0aae015f25116..324f2797ca841 100644
--- a/apps/server/src/mcp/toolkits/project/handlers.test.ts
+++ b/apps/server/src/mcp/toolkits/project/handlers.test.ts
@@ -22,6 +22,7 @@ import * as ServerConfig from "../../../config.ts";
import * as ServerSettings from "../../../serverSettings.ts";
import * as Project from "../../../project/ProjectService.ts";
import * as ManagedProjectFolders from "../../../project/ManagedProjectFolders.ts";
+import * as PeerForwarding from "../../../peer/PeerForwarding.ts";
import * as GitVcsDriver from "../../../vcs/GitVcsDriver.ts";
import * as VcsProcess from "../../../vcs/VcsProcess.ts";
import * as McpInvocationContext from "../../McpInvocationContext.ts";
@@ -79,6 +80,7 @@ it.effect("attributes a launched thread's first message to the calling thread",
},
}),
Layer.mock(Project.ProjectService)({}),
+ Layer.mock(PeerForwarding.PeerForwarding)({}),
Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/projects" }),
Layer.mock(GitVcsDriver.GitVcsDriver)({}),
NodeServices.layer,
@@ -151,6 +153,7 @@ it.effect("launches a scratch thread into the Scratch project", () =>
},
}),
Layer.mock(Project.ProjectService)({}),
+ Layer.mock(PeerForwarding.PeerForwarding)({}),
Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
namedProjectsRoot: "/projects",
ensureScratchProject: Effect.succeed({ projectId: scratchProjectId }),
@@ -247,6 +250,7 @@ it.effect("starts a project from just a title when workspaceRoot is omitted", ()
projectId === createdProjectId ? Option.some(createdProject) : Option.none(),
),
}),
+ Layer.mock(PeerForwarding.PeerForwarding)({}),
Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
namedProjectsRoot: "/projects",
createNamedProject: (input) =>
@@ -357,6 +361,7 @@ const clientLaunchHarness = (input: {
: Option.none(),
),
}),
+ Layer.mock(PeerForwarding.PeerForwarding)({}),
Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/projects" }),
NodeServices.layer,
).pipe(
diff --git a/apps/server/src/mcp/toolkits/project/tools.ts b/apps/server/src/mcp/toolkits/project/tools.ts
index 07fadeeb0aa22..f16a9a8b9d3cb 100644
--- a/apps/server/src/mcp/toolkits/project/tools.ts
+++ b/apps/server/src/mcp/toolkits/project/tools.ts
@@ -14,6 +14,7 @@ import {
ProjectUpdatePayload,
ProjectId,
OrchestratorMcpClientRequestId,
+ OrchestratorMcpEnvironmentTarget,
OrchestratorMcpFailure,
SourceControlCloneRepositoryInput,
SourceControlCloneRepositoryResult,
@@ -46,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 in this environment, or in a linked one 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 }))),
}),
@@ -62,15 +64,19 @@ 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 }),
+ 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 through the existing project service. Set createWorkspaceRootIfMissing to create a directory. Omit workspaceRoot to start a new project from just its title: the app makes a Git repository for it in its own projects folder, with a README, an icon, and a first commit (commitError says why a commit failed; the project exists either way). Each call creates a new request; an existing registered workspace is rejected. Clone separately with t3_project_clone when needed. Pass environmentId to create it in a linked environment; workspaceRoot and defaultModelSelection then name that environment's paths and providers.",
parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
...ProjectCreatePayload.fields,
workspaceRoot: Schema.optional(ProjectCreatePayload.fields.workspaceRoot),
}),
@@ -81,19 +87,30 @@ 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 }),
+ 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) }),
+ 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 using the app's source-control service. This only clones; register the returned cwd with t3_project_create. An existing destination is not adopted or removed on failure. Pass environmentId to clone in a linked environment, then register it there with t3_project_create and the same environmentId.",
+ parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
+ ...SourceControlCloneRepositoryInput.fields,
+ }),
success: SourceControlCloneRepositoryResult,
dependencies: [
...shared.dependencies,
@@ -105,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. Pass clientRequestId to retry safely: the same key returns the thread the first call created instead of launching another. Retain threadId and use t3_thread_read/t3_thread_wait to follow preparation. To link a thread for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread\'s current title. Without a key, after errors or lost responses, inspect t3_thread_list before retrying. Attachments must be pending uploads. The new thread may not run with broader runtime or interaction modes than the caller: the calling T3 thread\'s own modes, or the permission mode an outside agent was approved with.',
+ 'Create an ordinary TOP-LEVEL thread with an explicit workspace binding before its agent starts. Use this when the user requests independent work, a new thread, or a PR stack in its own worktree; use delegate_task for child subagents. Set workspaceStrategy to {type:"worktree",baseRef:"parent-branch",branch:"new-branch",startFromOrigin:false} for a new worktree based on local commits, or {type:"existing_worktree",worktreePath:"/absolute/path",branch:"existing-branch"} to use an existing checkout. 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. Pass clientRequestId to retry safely: the same key returns the thread the first call created instead of launching another. Retain threadId and use t3_thread_read/t3_thread_wait to follow preparation. To link a thread in this environment for the user, write `[title](t3-thread://v1/)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread\'s current title. Name a thread in a linked environment by title instead: its link would open here. Without a key, after errors or lost responses, inspect t3_thread_list before retrying. Attachments must be pending uploads. The new thread may not run with broader runtime or interaction modes than the caller: the calling T3 thread\'s own modes, or the permission mode an outside agent was approved with. Pass environmentId to launch in a linked environment; projectId and modelSelection then name that environment\'s project and providers (t3_project_list and orchestrator_capabilities with the same environmentId list them), and attachments are not supported.',
parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
projectId: Schema.optional(ProjectId),
scratch: Schema.optional(
Schema.Boolean.annotate({
diff --git a/apps/server/src/mcp/toolkits/pullRequests/tools.ts b/apps/server/src/mcp/toolkits/pullRequests/tools.ts
index a3a75eabaa385..4595578e1e50e 100644
--- a/apps/server/src/mcp/toolkits/pullRequests/tools.ts
+++ b/apps/server/src/mcp/toolkits/pullRequests/tools.ts
@@ -1,5 +1,6 @@
import {
McpCapabilityUnavailableError,
+ OrchestratorMcpEnvironmentTarget,
OrchestratorMcpFailure,
PositiveInt,
PullRequestReviewGuide,
@@ -35,6 +36,7 @@ const REGISTER_EVERY_PR =
"Register every pull request you open for this thread, including each layer of a stack, right after creating it.";
export const PullRequestTargetInput = Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
threadId: Schema.optional(
ThreadId.annotate({
description: "Thread to act on. Omit for this thread.",
@@ -300,6 +302,7 @@ const UnlinkPullRequestTool = Tool.make("unlink_pull_request", {
const ListThreadPullRequestsTool = Tool.make("list_thread_pull_requests", {
description: `List the pull requests linked to a thread (omit threadId for this thread) with their last known host state, and how they chain into stacks (bottom to top). ${REGISTER_EVERY_PR}`,
parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
threadId: Schema.optional(
ThreadId.annotate({ description: "Thread to list. Omit for this thread." }),
),
@@ -379,6 +382,7 @@ const WritePullRequestGuidesTool = Tool.make("write_pull_request_guides", {
description:
"Write review guides for up to 20 pull requests, the tour a maintainer reads in a pull request's Guide tab. Each run reads its change in a temporary checkout and takes minutes, so this returns once the runs are under way; read a guide with read_pull_request_guide to see when it is written. A guide for the same head that is already running is joined, not restarted; a pull request whose guide is current is written again. Each pull request is read through the project in this environment whose checkout has that repository. Fails while review guides are off on this server.",
parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
pullRequests: Schema.Array(GuideTarget).check(Schema.isMinLength(1), Schema.isMaxLength(20)),
}),
success: WritePullRequestGuidesResult,
@@ -394,7 +398,10 @@ const WritePullRequestGuidesTool = Tool.make("write_pull_request_guides", {
const ReadPullRequestGuideTool = Tool.make("read_pull_request_guide", {
description:
"Read a pull request's stored review guide and its run: run.Running while one is being written, run.Failed with the reason when the last one failed, guide null when none was written. stale is true after a push the guide has not caught up with. Never starts a run.",
- parameters: GuideTarget,
+ parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
+ ...GuideTarget.fields,
+ }),
success: ReadPullRequestGuideResult,
failure: PullRequestToolError,
dependencies: guideDependencies,
diff --git a/apps/server/src/mcp/toolkits/thread/tools.ts b/apps/server/src/mcp/toolkits/thread/tools.ts
index b5fb7d6be3a68..95894b23a5962 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,
@@ -32,6 +33,7 @@ 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.",
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,
@@ -82,6 +88,7 @@ const QueueListTool = Tool.make("t3_queue_list", {
description:
"List queued messages in delivery order. Results are a live offset page; use t3_thread_read for full thread history.",
parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
threadId: Schema.optional(ThreadId),
cursor: Schema.optional(NonNegativeInt),
limit: Schema.optional(Schema.Int.check(Schema.isBetween({ minimum: 1, maximum: 100 }))),
@@ -127,7 +134,11 @@ const QueuePromoteTool = Tool.make("t3_queue_promote_to_steer", {
parameters: Schema.Struct({ ...queueTarget, targetRunId: RunId }),
}).annotate(Tool.Destructive, true);
-const requestTarget = { threadId: Schema.optional(ThreadId), requestId: RuntimeRequestId };
+const requestTarget = {
+ environmentId: OrchestratorMcpEnvironmentTarget,
+ threadId: Schema.optional(ThreadId),
+ requestId: RuntimeRequestId,
+};
const question = Schema.Struct({
id: Schema.String,
header: Schema.String,
@@ -152,7 +163,10 @@ 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) }),
+ parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
+ threadId: Schema.optional(ThreadId),
+ }),
success: Schema.Struct({ requestIds: Schema.Array(RuntimeRequestId) }),
})
.annotate(Tool.Readonly, true)
@@ -179,7 +193,10 @@ 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) }),
+ parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
+ threadId: Schema.optional(ThreadId),
+ }),
success: Schema.Struct({
threadId: ThreadId,
modelSelection: ModelSelection,
@@ -194,6 +211,7 @@ const ThreadConfigureTool = Tool.make("t3_thread_configure", {
description:
"Set a thread's provider, model and options with the existing selection command. Omit threadId for this thread. This does not change permission modes. Use orchestrator_capabilities to choose a selection.",
parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
threadId: Schema.optional(ThreadId),
modelSelection: ModelSelection,
}),
@@ -205,6 +223,7 @@ const ThreadForkTool = Tool.make("t3_thread_fork", {
description:
"Fork a thread from a stable run or checkpoint using the existing fork command. Omit threadId to fork this thread. The fork inherits the source configuration. Acceptance does not mean a provider turn has completed.",
parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
threadId: Schema.optional(ThreadId),
sourcePoint: OrchestrationV2ThreadForkSourcePoint,
title: Schema.optional(TrimmedNonEmptyString),
@@ -216,6 +235,7 @@ const ThreadMergeBackTool = Tool.make("t3_thread_merge_back", {
description:
"Merge context from a thread back to a related thread in the same project. Omit sourceThreadId to merge from this thread. Existing lineage and transfer rules apply.",
parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
sourceThreadId: Schema.optional(ThreadId),
targetThreadId: ThreadId,
sourcePoint: OrchestrationV2ThreadForkSourcePoint,
@@ -225,7 +245,10 @@ const ThreadMergeBackTool = Tool.make("t3_thread_merge_back", {
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) }),
+ parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
+ threadId: Schema.optional(ThreadId),
+ }),
success: Schema.Struct({
transfers: Schema.Array(
Schema.Struct({
@@ -245,6 +268,7 @@ const ThreadSearchTool = Tool.make("t3_thread_search", {
description:
"Search active thread titles and content with the app's existing bounded search. Matches are limited to one project (projectId, else the calling thread's project) out of the global top matches, so this may return fewer than limit. A caller outside a T3 thread that omits projectId searches every project. No pagination or exhaustive-result guarantee.",
parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
...OrchestrationSearchThreadsInput.fields,
projectId: Schema.optional(ProjectId),
}),
@@ -258,7 +282,10 @@ const ScheduledTaskRunTool = Tool.make("run_scheduled_task_now", {
...commandTool,
description:
"Run a scheduled task now through the existing scheduler. Requires a full-access/default caller. Each call is a new manual run; completion means dispatch/bookkeeping completed, not that the provider turn finished.",
- parameters: Schema.Struct({ taskId: ScheduledTaskId }),
+ parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
+ taskId: ScheduledTaskId,
+ }),
success: Schema.Struct({
taskId: ScheduledTaskId,
threadId: ScheduledTask.fields.threadId,
diff --git a/apps/server/src/mcp/toolkits/worktree/registration.test.ts b/apps/server/src/mcp/toolkits/worktree/registration.test.ts
index 1598031d50c75..4038f652caab4 100644
--- a/apps/server/src/mcp/toolkits/worktree/registration.test.ts
+++ b/apps/server/src/mcp/toolkits/worktree/registration.test.ts
@@ -34,6 +34,7 @@ import * as McpHttpServer from "../../McpHttpServer.ts";
import * as McpSessionRegistry from "../../McpSessionRegistry.ts";
import * as PreviewAutomationBroker from "../../PreviewAutomationBroker.ts";
import * as PreviewBrowser from "../../../preview/PreviewBrowser.ts";
+import * as PeerForwarding from "../../../peer/PeerForwarding.ts";
const layerStubServices = Layer.mergeAll(
Layer.mock(Orchestrator.OrchestratorV2)({}),
@@ -43,6 +44,7 @@ const layerStubServices = Layer.mergeAll(
Layer.mock(ProviderRegistry.ProviderRegistry)({}),
Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({}),
Layer.mock(ScheduledTaskService.ScheduledTaskService)({}),
+ Layer.mock(PeerForwarding.PeerForwarding)({}),
Layer.mock(SecretRequests.SecretRequests)({}),
Layer.mock(ProjectService.ProjectService)({}),
Layer.mock(PullRequestReviewGuideService.PullRequestReviewGuideService)({}),
diff --git a/apps/server/src/mcp/toolkits/worktree/tools.ts b/apps/server/src/mcp/toolkits/worktree/tools.ts
index 85bbfde03ce68..b746bf068f7f9 100644
--- a/apps/server/src/mcp/toolkits/worktree/tools.ts
+++ b/apps/server/src/mcp/toolkits/worktree/tools.ts
@@ -1,5 +1,6 @@
import {
WorktreeMcpFailure,
+ OrchestratorMcpEnvironmentTarget,
OrchestratorMcpFailure,
VcsListRefsInput,
VcsListRefsResult,
@@ -62,6 +63,7 @@ const WorktreeListTool = Tool.make("t3_worktree_list", {
description:
"List branch refs and their associated checkout paths for a thread's workspace (omit threadId for this thread) using the app's ref inventory. Detached worktrees without a branch are not included. Use t3_worktree_status for the thread binding and t3_worktree_handoff to create a new worktree.",
parameters: Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
threadId: Schema.optional(ThreadId),
query: VcsListRefsInput.fields.query,
cursor: VcsListRefsInput.fields.cursor,
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/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts
index 55c2855144a0f..c04e643c19b20 100644
--- a/apps/server/src/orchestration-v2/EventSink.ts
+++ b/apps/server/src/orchestration-v2/EventSink.ts
@@ -184,6 +184,11 @@ export interface EventSinkV2Shape {
readonly latestSequence: (input?: {
readonly threadId?: ThreadId;
}) => Effect.Effect;
+ readonly read: (input: {
+ readonly threadId: ThreadId;
+ readonly throughSequence: number;
+ readonly eventType: OrchestrationV2DomainEvent["type"];
+ }) => Stream.Stream;
readonly readByCommandId: (input: {
readonly commandId: CommandId;
}) => Stream.Stream;
@@ -901,6 +906,14 @@ const layerBase: Layer.Layer<
}),
),
),
+ read: (input) =>
+ eventStore
+ .read(input)
+ .pipe(
+ Stream.mapError(
+ (cause) => new EventSinkStreamError({ threadId: input.threadId, cause }),
+ ),
+ ),
readByCommandId: (input) =>
eventStore.readByCommandId(input).pipe(
Stream.mapError(
diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts
index 6f739b5ff8e0e..1f3ab01c9d92c 100644
--- a/apps/server/src/orchestration-v2/Orchestrator.ts
+++ b/apps/server/src/orchestration-v2/Orchestrator.ts
@@ -378,6 +378,12 @@ export interface OrchestratorV2Shape {
readonly getThreadEventSequence: (
threadId: ThreadId,
) => Effect.Effect;
+ /** Committed history at a fixed sequence, without subscribing to live events. */
+ readonly readStoredEvents: (input: {
+ readonly threadId: ThreadId;
+ readonly throughSequence: number;
+ readonly eventType: OrchestrationV2DomainEvent["type"];
+ }) => Stream.Stream;
readonly streamStoredEvents: Stream.Stream;
readonly streamStoredEventsFrom: (input?: {
readonly threadId?: ThreadId;
@@ -9717,6 +9723,26 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
),
);
const latestRun = projection.runs.findLast((run) => run.status !== "queued");
+ if (command.throughSequence !== undefined && latestRun !== undefined) {
+ // A queued run already belonged to Stop even if it started before the delegated walk.
+ const existed = yield* eventSink
+ .read({
+ threadId: command.threadId,
+ throughSequence: command.throughSequence,
+ eventType: "run.created",
+ })
+ .pipe(
+ Stream.runFold(
+ () => false,
+ (found, stored) => found || stored.event.runId === latestRun.id,
+ ),
+ Effect.mapError(
+ (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }),
+ ),
+ );
+ if (!existed) return undefined;
+ // Partial retries keep queue/watch changes thread-wide while the original run is current.
+ }
const target =
projection.runs.findLast(
(run) =>
@@ -11383,7 +11409,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
// A settle that ended nothing still records its receipt: a replayed Stop
// effect then finds it instead of settling work that appeared since.
const resultSequence = yield* Effect.gen(function* () {
- const sequence = yield* eventSink.latestSequence({ threadId: commandThreadId(command) });
+ // Internal Stop receipts also bound other threads; client commands keep their thread cursor.
+ const sequence = yield* eventSink.latestSequence(
+ command.type === "thread.stop" ? undefined : { threadId: commandThreadId(command) },
+ );
yield* commandReceipts.insertIfAbsent({
commandId: command.commandId,
threadId: commandThreadId(command),
@@ -11772,6 +11801,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
eventSink
.latestSequence({ threadId })
.pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause }))),
+ readStoredEvents: (input) =>
+ eventSink
+ .read(input)
+ .pipe(Stream.mapError((cause) => new OrchestratorDomainEventStreamError({ cause }))),
streamStoredEvents: eventSink.stream().pipe(
Stream.mapError(
(cause) =>
@@ -11926,6 +11959,12 @@ const layerUnavailable: Layer.Layer = Layer.succeed(
cause: "Orchestration V2 live runtime is not configured.",
}),
),
+ readStoredEvents: () =>
+ Stream.fail(
+ new OrchestratorDomainEventStreamError({
+ cause: "Orchestration V2 live runtime is not configured.",
+ }),
+ ),
streamStoredEventsFrom: () =>
Stream.fail(
new OrchestratorDomainEventStreamError({
diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts
index d83dd9578c279..58d58b1b465d9 100644
--- a/apps/server/src/orchestration-v2/ThreadManagementService.ts
+++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts
@@ -43,6 +43,7 @@ import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Ref from "effect/Ref";
+import * as Result from "effect/Result";
import * as Schema from "effect/Schema";
import * as Stream from "effect/Stream";
@@ -1092,20 +1093,64 @@ const make = Effect.gen(function* () {
const stopDelegatedTasks: ThreadManagementServiceShape["stopDelegatedTasks"] = (input) =>
Effect.gen(function* () {
- const { subagents } = yield* orchestrator.getThreadRecords(input.threadId, ["subagents"]);
+ const receipt = yield* receipts
+ .getByCommandId(input.commandId)
+ .pipe(
+ Effect.mapError(
+ (cause) =>
+ new Orchestrator.OrchestratorProjectionError({ threadId: input.threadId, cause }),
+ ),
+ );
+ return yield* stopDelegatedTasksThrough(
+ input,
+ Option.isSome(receipt) ? receipt.value.resultSequence : undefined,
+ );
+ });
+
+ const stopDelegatedTasksThrough = (
+ input: Parameters[0],
+ throughSequence: number | undefined,
+ ): ReturnType =>
+ Effect.gen(function* () {
+ // A retry must not stop tasks delegated after the original Stop, including
+ // when an earlier walk stopped some children and failed on another.
+ const childThreadIds = new Set(
+ throughSequence === undefined
+ ? (yield* orchestrator.getThreadRecords(input.threadId, ["subagents"])).subagents.flatMap(
+ (task) =>
+ task.origin === "app_owned" && task.childThreadId !== null
+ ? [task.childThreadId]
+ : [],
+ )
+ : yield* orchestrator
+ .readStoredEvents({
+ threadId: input.threadId,
+ throughSequence,
+ eventType: "subagent.updated",
+ })
+ .pipe(
+ Stream.filterMap((stored) =>
+ stored.event.type === "subagent.updated" &&
+ stored.event.payload.origin === "app_owned" &&
+ stored.event.payload.childThreadId !== null
+ ? Result.succeed(stored.event.payload.childThreadId)
+ : Result.failVoid,
+ ),
+ Stream.runCollect,
+ ),
+ );
const failures: Array<
Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError
> = [];
- for (const task of subagents) {
- if (task.origin !== "app_owned" || task.childThreadId === null) continue;
- const threadId = task.childThreadId;
+ for (const threadId of childThreadIds) {
yield* dispatch({
type: "thread.stop",
commandId: CommandId.make(`${input.commandId}:stop:${threadId}`),
threadId,
+ ...(throughSequence === undefined ? {} : { throughSequence }),
...(input.reason === undefined ? {} : { reason: input.reason }),
}).pipe(
- Effect.andThen(stopDelegatedTasks({ ...input, threadId })),
+ Effect.andThen(stopDelegatedTasksThrough({ ...input, threadId }, throughSequence)),
Effect.catch((error) =>
Effect.logWarning("Unable to stop a delegated task", {
parentThreadId: input.threadId,
diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts
index 12db927d9c6d0..3c7ef3400728c 100644
--- a/apps/server/src/orchestration-v2/ThreadStop.test.ts
+++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts
@@ -1,8 +1,13 @@
import { assert, it } from "@effect/vitest";
+import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import * as FileSystem from "effect/FileSystem";
+import * as Path from "effect/Path";
import {
CommandId,
ContextTransferId,
EventId,
+ EnvironmentId,
MessageId,
NodeId,
type OrchestrationV2DomainEvent,
@@ -24,13 +29,23 @@ import * as Schema from "effect/Schema";
import * as SqlClient from "effect/sql/SqlClient";
import * as Stream from "effect/Stream";
import * as EventStore from "./EventStore.ts";
+import * as EventSink from "./EventSink.ts";
import * as TestClock from "effect/testing/TestClock";
import * as SqlitePersistence from "../persistence/Sqlite.ts";
import * as ServerSettings from "../serverSettings.ts";
+import * as OrchestratorMcpService from "../mcp/OrchestratorMcpService.ts";
+import type * as McpInvocationContext from "../mcp/McpInvocationContext.ts";
+import * as ThreadLaunch from "./ThreadLaunchService.ts";
+import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
+import * as ProviderRegistry from "../provider/ProviderRegistry.ts";
+import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts";
+import * as ProjectService from "../project/ProjectService.ts";
+import * as SecretRequests from "../secrets/SecretRequests.ts";
import { CLAUDE_PROVIDER } from "./Adapters/ClaudeAdapterV2.ts";
import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
import * as CommandReceiptStore from "./CommandReceiptStore.ts";
+import { DispatchModeLimit } from "./DispatchModeLimit.ts";
import { OrchestrationEffectRequestV2 } from "./EffectOutbox.ts";
import * as Orchestrator from "./Orchestrator.ts";
import * as EffectWorker from "./EffectWorker.ts";
@@ -50,24 +65,42 @@ const adapter = {
planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }),
openSession: () => Effect.die("Runs here never reach a provider"),
} as ProviderAdapter.ProviderAdapterV2["Service"];
-const layerDatabase = SqlitePersistence.layerMemory;
-const layerTest = ThreadManagementService.layer.pipe(
- Layer.provideMerge(
- Layer.mergeAll(
- layerDatabase,
- ProjectionStore.layer.pipe(Layer.provide(layerDatabase)),
- EventStore.layer.pipe(Layer.provide(layerDatabase)),
- EffectOutbox.layer.pipe(Layer.provide(layerDatabase)),
- ProviderReplayHarness.layerWithRegistry(
- { name: "thread-stop" },
- ProviderAdapterRegistry.layerFromAdapters([adapter]),
- { databaseLayer: layerDatabase, runEffectWorker: false },
+const fileDatabase = (dbPath: string) =>
+ SqlitePersistence.layerFromPath(dbPath).pipe(Layer.provide(NodeServices.layer));
+const makeLayerTest = (
+ layerDatabase: ReturnType = SqlitePersistence.layerMemory,
+) =>
+ ThreadManagementService.layer.pipe(
+ Layer.provideMerge(
+ Layer.mergeAll(
+ layerDatabase,
+ ProjectionStore.layer.pipe(Layer.provide(layerDatabase)),
+ EventStore.layer.pipe(Layer.provide(layerDatabase)),
+ EffectOutbox.layer.pipe(Layer.provide(layerDatabase)),
+ ProviderReplayHarness.layerWithRegistry(
+ { name: "thread-stop" },
+ ProviderAdapterRegistry.layerFromAdapters([adapter]),
+ { databaseLayer: layerDatabase, runEffectWorker: false },
+ ),
),
),
- ),
- Layer.provideMerge(ServerSettings.layerTest()),
- Layer.provideMerge(CommandReceiptStore.layer.pipe(Layer.provide(layerDatabase))),
-);
+ Layer.provideMerge(ServerSettings.layerTest()),
+ Layer.provideMerge(CommandReceiptStore.layer.pipe(Layer.provide(layerDatabase))),
+ );
+const layerTest = makeLayerTest();
+
+const layerMcp = (layerDatabase: ReturnType) =>
+ OrchestratorMcpService.layer.pipe(
+ Layer.provideMerge(makeLayerTest(layerDatabase)),
+ Layer.provide(NodeCrypto.layer),
+ Layer.provide(ProviderAdapterRegistry.layerFromAdapters([adapter])),
+ Layer.provide(Layer.mock(ThreadLaunch.ThreadLaunchService)({})),
+ Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})),
+ Layer.provide(Layer.mock(ProviderRegistry.ProviderRegistry)({})),
+ Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
+ Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
+ Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
+ );
const encodeEffectRequest = Schema.encodeSync(Schema.fromJsonString(OrchestrationEffectRequestV2));
@@ -424,7 +457,7 @@ it.effect.each([
it.effect("Stop ends watches, holds queues, and stops the delegated tasks under the thread", () =>
Effect.gen(function* () {
const orchestrator = yield* Orchestrator.OrchestratorV2;
- const projections = yield* ProjectionStore.ProjectionStoreV2;
+ const sink = yield* EventSink.EventSinkV2;
const threads = yield* ThreadManagementService.ThreadManagementService;
const sql = yield* SqlClient.SqlClient;
const parentThreadId = ThreadId.make("thread:stop-parent");
@@ -434,16 +467,21 @@ it.effect("Stop ends watches, holds queues, and stops the delegated tasks under
assert.isTrue(Exit.isFailure(yield* Effect.exit(watch(childThreadId, 2))));
const grandchildThreadId = yield* delegate(childThreadId, "grandchild task");
yield* send(childThreadId, "child follow-up", "queue_after_active");
+ yield* send(childThreadId, "child queued after follow-up", "queue_after_active");
const now = yield* DateTime.now;
const firstRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!;
- yield* projections.apply({
- id: EventId.make("event:stop-parent:first-run-completed"),
- type: "run.updated",
- threadId: parentThreadId,
- runId: firstRun.id,
- occurredAt: now,
- payload: { ...firstRun, status: "completed", completedAt: now },
+ yield* sink.write({
+ events: [
+ {
+ id: EventId.make("event:stop-parent:first-run-completed"),
+ type: "run.updated",
+ threadId: parentThreadId,
+ runId: firstRun.id,
+ occurredAt: now,
+ payload: { ...firstRun, status: "completed", completedAt: now },
+ },
+ ],
});
yield* send(parentThreadId, "second", "start_immediately");
const secondRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.at(-1)!;
@@ -471,15 +509,44 @@ it.effect("Stop ends watches, holds queues, and stops the delegated tasks under
"delegated-tasks.stop",
);
+ const childRun = (yield* orchestrator.getThreadProjection(childThreadId)).runs[0]!;
+ yield* sink.write({
+ events: [
+ {
+ id: EventId.make("event:stop-child:first-run-completed"),
+ type: "run.updated",
+ threadId: childThreadId,
+ runId: childRun.id,
+ occurredAt: now,
+ payload: { ...childRun, status: "completed", completedAt: now },
+ },
+ ],
+ });
+ yield* orchestrator.dispatch({
+ type: "queue.resume",
+ commandId: CommandId.make("resume-child-before-stop-walk"),
+ threadId: childThreadId,
+ });
+ assert.deepEqual((yield* threadState(childThreadId)).runs, ["completed", "starting", "queued"]);
+
yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId: stopCommandId });
assert.deepEqual(yield* threadState(childThreadId), {
- runs: ["interrupted", "queued:held"],
+ runs: ["completed", "interrupted", "queued:held"],
watched: [],
});
assert.deepEqual((yield* threadState(grandchildThreadId)).runs, ["interrupted"]);
yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId: stopCommandId });
- assert.deepEqual((yield* threadState(childThreadId)).runs, ["interrupted", "queued:held"]);
+ assert.deepEqual((yield* threadState(childThreadId)).runs, [
+ "completed",
+ "interrupted",
+ "queued:held",
+ ]);
+
+ yield* send(parentThreadId, "new work", "start_immediately");
+ const newChild = yield* delegate(parentThreadId, "new task after Stop");
+ yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId: stopCommandId });
+ assert.deepEqual((yield* threadState(newChild)).runs, ["starting"]);
}).pipe(Effect.provide(layerTest)),
);
@@ -951,40 +1018,75 @@ it.effect.each([
}).pipe(Effect.provide(layerTest)),
);
-it.effect("a delegated task that cannot be stopped fails the walk after its siblings stop", () =>
+it.effect("a partial delegated Stop retries only the original children that failed", () =>
Effect.gen(function* () {
const orchestrator = yield* Orchestrator.OrchestratorV2;
- const projections = yield* ProjectionStore.ProjectionStoreV2;
const threads = yield* ThreadManagementService.ThreadManagementService;
const parentThreadId = ThreadId.make("thread:stop-partial");
yield* createWatchingThread(parentThreadId, 6);
yield* send(parentThreadId, "work", "start_immediately");
- const childThreadId = yield* delegate(parentThreadId, "stoppable task");
- const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!;
- yield* projections.apply({
- id: EventId.make("event:stop-partial:missing-task"),
- type: "subagent.updated",
- threadId: parentThreadId,
- runId: task.runId!,
- nodeId: NodeId.make("node:stop-partial:missing"),
- providerInstanceId: instanceId,
- occurredAt: yield* DateTime.now,
- payload: {
- ...task,
- id: NodeId.make("node:stop-partial:missing"),
- childThreadId: ThreadId.make("thread:stop-partial:missing"),
- startedAt: DateTime.subtract(task.startedAt!, { hours: 1 }),
- },
- });
-
- const walked = yield* Effect.exit(
- threads.stopDelegatedTasks({
+ const rootRefused = yield* threads
+ .dispatch({
+ type: "thread.stop",
+ commandId: CommandId.make("stop-root-above-limit"),
threadId: parentThreadId,
- commandId: CommandId.make("stop-partial"),
+ })
+ .pipe(
+ Effect.provideService(DispatchModeLimit, {
+ runtimeMode: "approval-required",
+ interactionMode: "default",
+ }),
+ Effect.exit,
+ );
+ assert.isTrue(Exit.isFailure(rootRefused));
+ assert.deepEqual((yield* threadState(parentThreadId)).runs, ["starting"]);
+ const blockedChild = yield* delegate(parentThreadId, "default task");
+ const advancedChild = yield* delegate(parentThreadId, "default task that advances");
+ const stoppedChild = yield* delegate(parentThreadId, "plan task");
+ const blockedGrandchild = yield* delegate(stoppedChild, "broader grandchild");
+ yield* orchestrator.dispatch({
+ type: "thread.runtime-mode.set",
+ commandId: CommandId.make("approval-child"),
+ threadId: stoppedChild,
+ runtimeMode: "approval-required",
+ });
+ yield* orchestrator.dispatch({
+ type: "thread.interaction-mode.set",
+ commandId: CommandId.make("plan-child"),
+ threadId: stoppedChild,
+ interactionMode: "plan",
+ });
+ const commandId = CommandId.make("stop-partial");
+ yield* orchestrator.dispatch({ type: "thread.stop", commandId, threadId: parentThreadId });
+ const walked = yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId }).pipe(
+ Effect.provideService(DispatchModeLimit, {
+ runtimeMode: "approval-required",
+ interactionMode: "plan",
}),
+ Effect.exit,
);
assert.isTrue(Exit.isFailure(walked));
- assert.deepEqual((yield* threadState(childThreadId)).runs, ["interrupted"]);
+ assert.deepEqual((yield* threadState(stoppedChild)).runs, ["interrupted"]);
+ assert.deepEqual((yield* threadState(blockedChild)).runs, ["starting"]);
+ assert.deepEqual((yield* threadState(blockedGrandchild)).runs, ["starting"]);
+
+ yield* orchestrator.dispatch({
+ type: "thread.stop",
+ commandId: CommandId.make("independent-child-stop"),
+ threadId: advancedChild,
+ });
+ yield* send(advancedChild, "new work before failed Stop retry", "start_immediately");
+ yield* send(stoppedChild, "new child work", "start_immediately");
+ const newGrandchild = yield* delegate(stoppedChild, "new grandchild");
+ yield* send(parentThreadId, "new parent work", "start_immediately");
+ const newChild = yield* delegate(parentThreadId, "new child");
+ yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId });
+ assert.deepEqual((yield* threadState(blockedChild)).runs, ["interrupted"]);
+ assert.deepEqual((yield* threadState(blockedGrandchild)).runs, ["interrupted"]);
+ assert.deepEqual((yield* threadState(stoppedChild)).runs, ["interrupted", "starting"]);
+ assert.deepEqual((yield* threadState(advancedChild)).runs, ["interrupted", "starting"]);
+ assert.deepEqual((yield* threadState(newChild)).runs, ["starting"]);
+ assert.deepEqual((yield* threadState(newGrandchild)).runs, ["starting"]);
}).pipe(Effect.provide(layerTest)),
);
@@ -1334,3 +1436,119 @@ it.effect.each(["missing session", "completed", "failed", "cancelled"] as const)
);
}).pipe(Effect.provide(layerTest)),
);
+
+it.effect("an MCP Stop replays its original result after the database and services reopen", () =>
+ Effect.gen(function* () {
+ const fs = yield* FileSystem.FileSystem;
+ const path = yield* Path.Path;
+ const dir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-mcp-stop-replay-" });
+ const database = fileDatabase(path.join(dir, "statev2.sqlite"));
+ const scope: McpInvocationContext.McpInvocationScope = {
+ environmentId: EnvironmentId.make("stop-replay"),
+ requestNamespace: "client:stop-replay",
+ capabilities: new Set(["orchestration"]),
+ issuedAt: 1,
+ thread: undefined,
+ client: { sessionId: "stop-replay", label: "Stop replay", access: "full-access" },
+ };
+ const threadId = ThreadId.make("thread:mcp-stop-replay");
+ const stop = { threadId, stop: true, clientRequestId: "stop" };
+ const first = yield* Effect.gen(function* () {
+ const mcp = yield* OrchestratorMcpService.OrchestratorMcpService;
+ const idle = ThreadId.make("thread:mcp-stop-idle");
+ yield* createWatchingThread(idle, 102);
+ yield* send(idle, "finished work", "start_immediately");
+ yield* send(idle, "held queue", "queue_after_active");
+ const idleProjection = yield* (yield* Orchestrator.OrchestratorV2).getThreadProjection(idle);
+ const idleRun = idleProjection.runs[0]!;
+ const now = yield* DateTime.now;
+ yield* (yield* EventSink.EventSinkV2).write({
+ events: [
+ {
+ id: EventId.make("event:mcp-stop-idle:completed"),
+ type: "run.updated",
+ threadId: idle,
+ runId: idleRun.id,
+ occurredAt: now,
+ payload: { ...idleRun, status: "completed", completedAt: now },
+ },
+ {
+ id: EventId.make("event:mcp-stop-idle:held"),
+ type: "run.updated",
+ threadId: idle,
+ runId: idleProjection.runs[1]!.id,
+ occurredAt: now,
+ payload: { ...idleProjection.runs[1]!, queueHeld: true },
+ },
+ ],
+ });
+ assert.deepEqual(
+ yield* mcp.interruptThread(scope, {
+ ...stop,
+ threadId: idle,
+ clientRequestId: "idle-stop",
+ }),
+ {
+ threadId: idle,
+ runId: null,
+ status: "no_active_run",
+ },
+ );
+ yield* createWatchingThread(threadId, 101);
+ const sent = yield* mcp.sendToThread(scope, {
+ threadId,
+ message: "first",
+ clientRequestId: "send",
+ });
+ const childThreadId = yield* delegate(threadId, "original child");
+ yield* delegate(childThreadId, "original grandchild");
+ const stopped = yield* mcp.interruptThread(scope, stop);
+ assert.equal(stopped.runId, sent.runId);
+ yield* send(childThreadId, "child work after parent stopped", "start_immediately");
+ const store = yield* EventStore.EventStoreV2;
+ const childSequence = yield* store.latestSequence({ threadId: childThreadId });
+ const noopStop = { ...stop, clientRequestId: "already-stopped-parent" };
+ const noopStopped = yield* mcp.interruptThread(scope, noopStop);
+ assert.deepEqual(noopStopped, { threadId, runId: null, status: "no_active_run" });
+ assert.deepEqual((yield* threadState(childThreadId)).runs, ["interrupted", "interrupted"]);
+ const sql = yield* SqlClient.SqlClient;
+ const [noopReceipt] = yield* sql<{ command_id: string; result_sequence: number }>`
+ SELECT command_id, result_sequence FROM orchestration_command_receipts
+ WHERE aggregate_id = ${threadId} AND command_type = 'thread.stop'
+ ORDER BY result_sequence DESC LIMIT 1
+ `;
+ assert.isAtLeast(noopReceipt!.result_sequence, childSequence);
+ assert.deepEqual(
+ yield* store
+ .readByCommandId({ commandId: CommandId.make(noopReceipt!.command_id) })
+ .pipe(Stream.runCollect),
+ [],
+ );
+ return { sent, stopped, childThreadId, noopStop, noopStopped };
+ }).pipe(Effect.provide(layerMcp(database).pipe(Layer.fresh)));
+
+ yield* Effect.gen(function* () {
+ const mcp = yield* OrchestratorMcpService.OrchestratorMcpService;
+ const orchestrator = yield* Orchestrator.OrchestratorV2;
+ const replay = yield* mcp.sendToThread(scope, {
+ threadId,
+ message: "first",
+ clientRequestId: "send",
+ });
+ assert.equal(replay.runId, first.sent.runId);
+ assert.equal((yield* orchestrator.getThreadProjection(threadId)).runs.length, 1);
+ yield* send(threadId, "after restart", "start_immediately");
+ const newChild = yield* delegate(threadId, "child after restart");
+ yield* send(first.childThreadId, "child work after restart", "start_immediately");
+ assert.deepEqual(yield* mcp.interruptThread(scope, first.noopStop), first.noopStopped);
+ assert.deepEqual((yield* threadState(first.childThreadId)).runs, [
+ "interrupted",
+ "interrupted",
+ "starting",
+ ]);
+ assert.deepEqual(yield* mcp.interruptThread(scope, stop), first.stopped);
+ assert.deepEqual((yield* threadState(threadId)).runs, ["interrupted", "starting"]);
+ assert.deepEqual((yield* threadState(newChild)).runs, ["starting"]);
+ }).pipe(Effect.provide(layerMcp(database).pipe(Layer.fresh)));
+ }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
+);
diff --git a/apps/server/src/peer/PeerForwarding.test.ts b/apps/server/src/peer/PeerForwarding.test.ts
new file mode 100644
index 0000000000000..b0aa2dbfeb863
--- /dev/null
+++ b/apps/server/src/peer/PeerForwarding.test.ts
@@ -0,0 +1,654 @@
+import { NodeHttpServer } from "@effect/platform-node";
+import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import {
+ EnvironmentId,
+ type OrchestrationV2Run,
+ type ModelSelection,
+ ProjectId,
+ ProviderInstanceId,
+ RunId,
+ ThreadId,
+} from "@t3tools/contracts";
+import { expect, it } from "@effect/vitest";
+import * as Context from "effect/Context";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as Option from "effect/Option";
+import * as Ref from "effect/Ref";
+import { McpSchema, McpServer } from "effect/ai";
+
+import * as ServerConfig from "../config.ts";
+import * as ThreadLaunch from "../orchestration-v2/ThreadLaunchService.ts";
+import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
+import * as ThreadSearch from "../orchestration-v2/ThreadSearch.ts";
+import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts";
+import * as SourceControlRepositoryService from "../sourceControl/SourceControlRepositoryService.ts";
+import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
+import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts";
+import * as ProjectService from "../project/ProjectService.ts";
+import * as ProviderRegistry from "../provider/ProviderRegistry.ts";
+import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts";
+import * as SecretRequests from "../secrets/SecretRequests.ts";
+import * as ServerSettings from "../serverSettings.ts";
+import * as McpHttpServer from "../mcp/McpHttpServer.ts";
+import * as McpInvocationContext from "../mcp/McpInvocationContext.ts";
+import { liveThreadShell } from "../mcp/McpToolAccess.testkit.ts";
+import * as PeerForwarding from "./PeerForwarding.ts";
+import * as PeerLinks from "./PeerLinks.ts";
+import { descriptorOf, layerLinkingEnvironment, linkTo, servePeer } from "./PeerLinks.testkit.ts";
+
+const laptop = descriptorOf("environment-laptop", "Laptop");
+const box = descriptorOf("environment-box", "Box");
+const boxProject = ProjectId.make("project:box");
+const boxThread = ThreadId.make("thread:box-auto");
+const boxRun = RunId.make("run:box-auto");
+
+/** What B's thread service was asked to do. */
+interface Seen {
+ readonly sends: ReadonlyArray;
+ readonly waits: ReadonlyArray;
+ readonly launches: ReadonlyArray;
+ readonly projects: ReadonlyArray;
+ /** Commands dispatched and delegated-task stops, as ` `. */
+ readonly stops: ReadonlyArray;
+ readonly selections: ReadonlyArray;
+}
+const nothingSeen: Seen = {
+ sends: [],
+ waits: [],
+ launches: [],
+ projects: [],
+ stops: [],
+ selections: [],
+};
+
+/**
+ * B's threads: one `auto` thread in `boxProject`, whose run finishes on the
+ * third wait. Every send and wait is recorded.
+ */
+const boxThreads = (seen: Ref.Ref) => {
+ const shell = {
+ ...liveThreadShell(boxThread, { runtimeMode: "auto" }),
+ projectId: boxProject,
+ modelSelection: {
+ instanceId: ProviderInstanceId.make("codex"),
+ model: "gpt-5",
+ options: [{ id: "reasoning", value: "high" }],
+ },
+ };
+ const run = (status: OrchestrationV2Run["status"]) =>
+ ({ id: boxRun, status }) as unknown as OrchestrationV2Run;
+ const projection = { thread: shell, runs: [run("running")], runtimeRequests: [] } as never;
+ return Layer.mock(ThreadManagement.ThreadManagementService)({
+ getThreadShell: (threadId) => Effect.succeed(threadId === boxThread ? shell : null),
+ getProjectThreadRecords: () => Effect.succeed(projection),
+ getThreadRecords: () => Effect.succeed(projection),
+ listProjectThreads: () => Effect.succeed([shell] as never),
+ sendToThread: (input) =>
+ Ref.update(seen, (current) => ({ ...current, sends: [...current.sends, input] })).pipe(
+ Effect.as({ run: run("running"), delivery: "started" } as never),
+ ),
+ dispatch: (command) =>
+ Ref.update(seen, (current) => ({
+ ...current,
+ selections:
+ command.type === "thread.model-selection.set"
+ ? [...current.selections, command.modelSelection]
+ : current.selections,
+ stops: [
+ ...current.stops,
+ `${command.type} ${"threadId" in command ? command.threadId : ""}`,
+ ],
+ })).pipe(
+ // The stop interrupts the thread's running run.
+ Effect.as({
+ sequence: 0,
+ storedEvents: [
+ {
+ sequence: 1,
+ commandId: command.commandId,
+ event: {
+ type: "turn-item.updated",
+ runId: boxRun,
+ payload: { type: "run_interrupt_request", providerTurnId: "turn:box" },
+ },
+ },
+ ],
+ } as never),
+ ),
+ stopDelegatedTasks: (input) =>
+ Ref.update(seen, (current) => ({
+ ...current,
+ stops: [...current.stops, `delegated-tasks.stop ${input.threadId}`],
+ })),
+ waitForThread: (input) =>
+ Ref.updateAndGet(seen, (current) => ({ ...current, waits: [...current.waits, input] })).pipe(
+ Effect.map(({ waits }) => ({
+ threadId: input.threadId,
+ run: run(waits.length < 3 ? "running" : "completed"),
+ timedOut: waits.length < 3,
+ })),
+ ),
+ });
+};
+
+/** B launches whatever it is asked to, and records it. */
+const boxLaunches = (seen: Ref.Ref) =>
+ Layer.mock(ThreadLaunch.ThreadLaunchService)({
+ getAcceptedLaunch: () => Effect.succeed(Option.none()),
+ launch: (input) =>
+ Ref.update(seen, (current) => ({ ...current, launches: [...current.launches, input] })).pipe(
+ Effect.as({
+ threadId: input.threadId,
+ projection: {
+ thread: {
+ id: input.threadId,
+ projectId: input.projectId,
+ modelSelection: input.modelSelection,
+ title: input.title,
+ },
+ runs: [],
+ },
+ resumed: false,
+ } as unknown as ThreadLaunch.ThreadLaunchResult),
+ ),
+ });
+
+const boxProjects = (seen: Ref.Ref) =>
+ Layer.mock(ProjectService.ProjectService)({
+ create: (input) =>
+ Ref.update(seen, (current) => ({ ...current, projects: [...current.projects, input] })).pipe(
+ Effect.as({
+ id: input.projectId,
+ title: input.title,
+ workspaceRoot: input.workspaceRoot,
+ defaultModelSelection: null,
+ scripts: [],
+ createdAt: "2026-10-18T00:00:00.000Z",
+ updatedAt: "2026-10-18T00:00:00.000Z",
+ deletedAt: null,
+ }),
+ ),
+ });
+
+/** The orchestrator and project toolkits on B's real `/mcp`, behind its real OAuth. */
+const serveBox = (seen: Ref.Ref) =>
+ servePeer(
+ box,
+ Layer.mergeAll(
+ McpHttpServer.layerOrchestratorToolkit,
+ McpHttpServer.layerProjectRegistration,
+ McpHttpServer.layerThreadToolkit,
+ ).pipe(
+ Layer.provide(NodeCrypto.layer),
+ Layer.provide(boxThreads(seen)),
+ Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})),
+ Layer.provide(boxLaunches(seen)),
+ Layer.provide(
+ Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/projects" }),
+ ),
+ Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})),
+ Layer.provide(boxProjects(seen)),
+ Layer.provide(Layer.mock(SourceControlRepositoryService.SourceControlRepositoryService)({})),
+ Layer.provide(
+ ServerConfig.layerTest(process.cwd(), { prefix: "t3-peer-forwarding-box-" }).pipe(
+ Layer.provide(NodeServices.layer),
+ ),
+ ),
+ Layer.provide(NodeServices.layer),
+ Layer.provide(Layer.mock(ProviderRegistry.ProviderRegistry)({})),
+ Layer.provide(Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({})),
+ Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
+ Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
+ Layer.provide(ServerSettings.layerTest()),
+ Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
+ ),
+ );
+
+const mcpClient = McpSchema.McpServerClient.of({
+ clientId: 1,
+ protocolVersion: "2025-06-18",
+ clientCapabilities: {},
+ clientInfo: { name: "peer-forwarding-test", version: "1" },
+ initializePayload: {
+ protocolVersion: "2025-06-18",
+ capabilities: {},
+ clientInfo: { name: "peer-forwarding-test", version: "1" },
+ },
+ getClient: Effect.die("unused"),
+});
+
+/**
+ * Environment A: its own orchestrator toolkit with forwarding, and a thread
+ * caller whose modes the test picks. Its local threads never matter here.
+ */
+const makeLaptop = Effect.gen(function* () {
+ const linking = yield* layerLinkingEnvironment(laptop).pipe(Layer.build);
+ const layerHere = Layer.mergeAll(
+ McpHttpServer.layerOrchestratorToolkit,
+ McpHttpServer.layerProjectRegistration,
+ McpHttpServer.layerThreadToolkit,
+ ).pipe(
+ Layer.provideMerge(McpServer.McpServer.layer),
+ Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})),
+ Layer.provide(Layer.mock(ThreadLaunch.ThreadLaunchService)({})),
+ Layer.provide(
+ Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/projects" }),
+ ),
+ Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})),
+ Layer.provide(Layer.mock(SourceControlRepositoryService.SourceControlRepositoryService)({})),
+ Layer.provide(
+ ServerConfig.layerTest(process.cwd(), { prefix: "t3-peer-forwarding-laptop-" }).pipe(
+ Layer.provide(NodeServices.layer),
+ ),
+ ),
+ Layer.provide(NodeServices.layer),
+ Layer.provide(PeerForwarding.layer),
+ Layer.provide(Layer.succeedContext(linking)),
+ Layer.provide(NodeCrypto.layer),
+ Layer.provideMerge(
+ Layer.mock(ThreadManagement.ThreadManagementService)({
+ getThreadShell: (threadId) =>
+ Effect.succeed(
+ liveThreadShell(
+ threadId,
+ threadId === ThreadId.make("thread:laptop-plan")
+ ? { runtimeMode: "auto", interactionMode: "plan" }
+ : { runtimeMode: "full-access" },
+ ),
+ ),
+ getProjectThreadRecords: ({ threadId }) =>
+ Effect.succeed({
+ thread: liveThreadShell(threadId, { runtimeMode: "full-access" }),
+ } as never),
+ }),
+ ),
+ Layer.provide(Layer.mock(ProviderRegistry.ProviderRegistry)({})),
+ Layer.provide(Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({})),
+ Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
+ Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
+ Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
+ Layer.provide(ServerSettings.layerTest()),
+ Layer.fresh,
+ );
+ const here = yield* Layer.build(layerHere);
+ const server = Context.get(here, McpServer.McpServer);
+ const scopeOf = (
+ caller: string,
+ capabilities: ReadonlySet = new Set(["orchestration"]),
+ ): McpInvocationContext.McpInvocationScope => ({
+ environmentId: laptop.environmentId,
+ requestNamespace: `provider-session:${caller}`,
+ thread: {
+ threadId: ThreadId.make(caller),
+ providerSessionId: `provider-session:${caller}`,
+ providerInstanceId: ProviderInstanceId.make("codex"),
+ },
+ client: undefined,
+ capabilities,
+ issuedAt: 1,
+ });
+ const call = (
+ caller: string,
+ name: string,
+ args: Record,
+ capabilities?: ReadonlySet,
+ ) =>
+ server
+ .callTool({ name, arguments: args })
+ .pipe(
+ Effect.provideService(
+ McpInvocationContext.McpInvocationContext,
+ scopeOf(caller, capabilities),
+ ),
+ Effect.provideService(McpSchema.McpServerClient, mcpClient),
+ );
+ return { links: Context.get(linking, PeerLinks.PeerLinks), call };
+});
+
+const failureOf = (result: McpSchema.CallToolResult) => {
+ const text = result.content[0];
+ return result.isError === true && text?.type === "text" ? JSON.parse(text.text) : undefined;
+};
+
+it.effect("an agent here works on a linked environment's threads within its own modes", () =>
+ Effect.scoped(
+ Effect.gen(function* () {
+ const seen = yield* Ref.make(nothingSeen);
+ const b = yield* serveBox(seen);
+ const a = yield* makeLaptop;
+ yield* linkTo(a.links, b, "auto");
+
+ const listed = yield* a.call("thread:laptop-full", "t3_environment_links", {});
+ expect(listed.structuredContent).toMatchObject({
+ environmentId: laptop.environmentId,
+ links: [
+ { environmentId: box.environmentId, label: "Box", status: "reachable", access: "auto" },
+ ],
+ });
+
+ const threads = yield* a.call("thread:laptop-full", "t3_thread_list", {
+ environmentId: box.environmentId,
+ projectId: boxProject,
+ });
+ expect(threads.isError).toBe(false);
+ expect(threads.structuredContent).toMatchObject({
+ projectId: boxProject,
+ threads: [{ threadId: boxThread, runtimeMode: "auto" }],
+ });
+
+ // A full-access agent here is held to the link's `auto` there, which
+ // the box's `auto` thread is within.
+ const sent = yield* a.call("thread:laptop-full", "t3_thread_send", {
+ environmentId: box.environmentId,
+ threadId: boxThread,
+ message: "Carry on there.",
+ clientRequestId: "send-1",
+ });
+ expect(sent.isError).toBe(false);
+ expect(sent.structuredContent).toMatchObject({ threadId: boxThread, delivery: "started" });
+
+ // A plan-mode agent here may not steer that thread there: the box
+ // refuses it, because the call carries the agent's own modes.
+ const planned = yield* a.call("thread:laptop-plan", "t3_thread_send", {
+ environmentId: box.environmentId,
+ threadId: boxThread,
+ message: "Change course.",
+ });
+ expect(failureOf(planned)).toMatchObject({ code: "interaction_mode_escalation_denied" });
+ expect((yield* Ref.get(seen)).sends).toHaveLength(1);
+
+ // stop=true stops the thread there as its Stop button does: the whole
+ // thread and the tasks it delegated, not just the running turn.
+ const stopped = yield* a.call("thread:laptop-full", "t3_thread_interrupt", {
+ environmentId: box.environmentId,
+ threadId: boxThread,
+ stop: true,
+ clientRequestId: "stop-1",
+ });
+ expect(stopped.structuredContent).toMatchObject({
+ threadId: boxThread,
+ runId: boxRun,
+ status: "interrupt_requested",
+ });
+ expect((yield* Ref.get(seen)).stops).toEqual([
+ `thread.stop ${boxThread}`,
+ `delegated-tasks.stop ${boxThread}`,
+ ]);
+
+ yield* Ref.set(
+ b.descriptor,
+ descriptorOf("environment-box", "Box", {
+ mcpModeLimitHeader: false,
+ }),
+ );
+ const incompatible = yield* a.call("thread:laptop-full", "t3_environment_links", {});
+ expect(incompatible.isError).toBe(false);
+ expect(incompatible.structuredContent).toMatchObject({
+ links: [{ environmentId: box.environmentId, status: "incompatible" }],
+ });
+ }),
+ ).pipe(Effect.provide(NodeHttpServer.layerTest)),
+);
+
+it.effect("keeps a forwarded wait short and its retries scoped to the caller", () =>
+ Effect.scoped(
+ Effect.gen(function* () {
+ const seen = yield* Ref.make(nothingSeen);
+ const b = yield* serveBox(seen);
+ const a = yield* makeLaptop;
+ yield* linkTo(a.links, b, "auto");
+
+ const waited = yield* a.call("thread:laptop-full", "t3_thread_wait", {
+ environmentId: box.environmentId,
+ threadId: boxThread,
+ timeoutMs: 110_000,
+ });
+ expect(waited.structuredContent).toEqual({
+ threadId: boxThread,
+ runId: boxRun,
+ status: "completed",
+ timedOut: false,
+ });
+ // Three pieces of at most 50 s each, the later ones pinned to the run
+ // the first one found.
+ expect(
+ (yield* Ref.get(seen)).waits.map(({ timeoutMs, runId }) => [timeoutMs, runId]),
+ ).toEqual([
+ [50_000, undefined],
+ [50_000, boxRun],
+ [10_000, boxRun],
+ ]);
+
+ const send = (caller: string, clientRequestId: string) =>
+ a.call(caller, "t3_thread_send", {
+ environmentId: box.environmentId,
+ threadId: boxThread,
+ message: "Once.",
+ clientRequestId,
+ });
+ yield* send("thread:laptop-full", "same-key");
+ yield* send("thread:laptop-full", "same-key");
+ yield* send("thread:laptop-other", "same-key");
+ const [first, retried, other] = (yield* Ref.get(seen)).sends.map((input) => input.commandId);
+ // The box sees every caller here as the one link, so the key is scoped
+ // to its caller before it leaves: a retry matches, another agent's same
+ // key does not.
+ expect(retried).toBe(first);
+ expect(other).not.toBe(first);
+ expect(first).not.toContain("same-key");
+ }),
+ ).pipe(Effect.provide(NodeHttpServer.layerTest)),
+);
+
+it.effect("launches in a linked environment with the caller's modes, retry-safe", () =>
+ Effect.scoped(
+ Effect.gen(function* () {
+ const seen = yield* Ref.make(nothingSeen);
+ const b = yield* serveBox(seen);
+ const a = yield* makeLaptop;
+ yield* linkTo(a.links, b, "full-access");
+ const codex = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" };
+ const launch = (caller: string, args: Record = {}) =>
+ a.call(caller, "t3_thread_launch", {
+ environmentId: box.environmentId,
+ projectId: boxProject,
+ title: "Run the suite there",
+ modelSelection: codex,
+ message: "Run the suite.",
+ clientRequestId: "launch-1",
+ ...args,
+ });
+
+ const first = yield* launch("thread:laptop-plan");
+ expect(first.isError).toBe(false);
+ expect(first.structuredContent).toMatchObject({
+ projectId: boxProject,
+ modelSelection: codex,
+ });
+ const retried = yield* launch("thread:laptop-plan");
+ const [launched, relaunched] = (yield* Ref.get(seen)).launches;
+ // The thread there runs with the caller's own modes here, below the
+ // link's full access, and a retry asks for the same thread.
+ expect(launched).toMatchObject({ runtimeMode: "auto", interactionMode: "plan" });
+ expect(relaunched?.threadId).toBe(launched?.threadId);
+ expect(retried.structuredContent).toEqual(first.structuredContent);
+
+ // An escalation past the caller's own modes is refused here.
+ yield* Ref.set(b.bearers, []);
+ const escalated = yield* launch("thread:laptop-plan", { interactionMode: "default" });
+ expect(failureOf(escalated)).toMatchObject({ code: "interaction_mode_escalation_denied" });
+ expect(yield* Ref.get(b.bearers)).toEqual([]);
+ // A pending upload lives here, so it cannot go along.
+ yield* Ref.set(b.bearers, []);
+ const withUpload = yield* launch("thread:laptop-full", {
+ clientRequestId: "launch-2",
+ attachments: [
+ { type: "image", id: "upload-1", name: "a.png", mimeType: "image/png", sizeBytes: 1 },
+ ],
+ });
+ expect(failureOf(withUpload)).toMatchObject({ code: "invalid_request" });
+ expect(yield* Ref.get(b.bearers)).toEqual([]);
+ expect((yield* Ref.get(seen)).launches).toHaveLength(2);
+ }),
+ ).pipe(Effect.provide(NodeHttpServer.layerTest)),
+);
+
+it.effect("creates a remote project only when both caller and link allow full access", () =>
+ Effect.scoped(
+ Effect.gen(function* () {
+ const seen = yield* Ref.make(nothingSeen);
+ const b = yield* serveBox(seen);
+ const a = yield* makeLaptop;
+ const register = { title: "t3code", workspaceRoot: "/home/box/t3code" };
+
+ yield* linkTo(a.links, b, "full-access");
+ const created = yield* a.call("thread:laptop-full", "t3_project_create", {
+ environmentId: box.environmentId,
+ ...register,
+ });
+ expect(created.structuredContent).toMatchObject(register);
+ expect((yield* Ref.get(seen)).projects).toMatchObject([register]);
+
+ // A plan-mode agent here may not, and nothing leaves.
+ yield* Ref.set(b.bearers, []);
+ const planned = yield* a.call("thread:laptop-plan", "t3_project_create", {
+ environmentId: box.environmentId,
+ ...register,
+ });
+ expect(failureOf(planned)).toMatchObject({ code: "capability_denied" });
+ expect(yield* Ref.get(b.bearers)).toEqual([]);
+
+ // Nor may a full-access agent through a link with less: the box holds it.
+ yield* linkTo(a.links, b, "auto");
+ const refused = yield* a.call("thread:laptop-full", "t3_project_create", {
+ environmentId: box.environmentId,
+ ...register,
+ });
+ expect(failureOf(refused)).toMatchObject({ code: "capability_denied" });
+ expect((yield* Ref.get(seen)).projects).toHaveLength(1);
+ }),
+ ).pipe(Effect.provide(NodeHttpServer.layerTest)),
+);
+
+it.effect("runs any tool that takes a linked environment there", () =>
+ Effect.scoped(
+ Effect.gen(function* () {
+ const seen = yield* Ref.make(nothingSeen);
+ const b = yield* serveBox(seen);
+ const a = yield* makeLaptop;
+ yield* linkTo(a.links, b, "auto");
+
+ const there = yield* a.call("thread:laptop-full", "t3_thread_configuration", {
+ environmentId: box.environmentId,
+ threadId: boxThread,
+ });
+ expect(there.structuredContent).toMatchObject({ threadId: boxThread, runtimeMode: "auto" });
+ const archived = yield* a.call("thread:laptop-full", "t3_thread_organize", {
+ environmentId: box.environmentId,
+ threadId: boxThread,
+ action: "archive",
+ });
+ expect(archived.isError).toBe(false);
+ expect((yield* Ref.get(seen)).stops).toEqual([`thread.archive ${boxThread}`]);
+
+ const selection = {
+ instanceId: "codex",
+ model: "gpt-5",
+ options: [{ id: "reasoning", value: "high" }],
+ };
+ const configured = yield* a.call("thread:laptop-full", "t3_thread_configure", {
+ environmentId: box.environmentId,
+ threadId: boxThread,
+ modelSelection: { provider: "codex", model: "gpt-5", options: { reasoning: "high" } },
+ });
+ expect(configured.isError).toBe(false);
+ expect((yield* Ref.get(seen)).selections).toEqual([selection]);
+ expect(there.structuredContent).toHaveProperty("modelSelection", selection);
+
+ // This environment's own id keeps the call here.
+ const own = yield* a.call("thread:laptop-full", "t3_thread_configuration", {
+ environmentId: laptop.environmentId,
+ });
+ expect(own.structuredContent).toMatchObject({
+ threadId: "thread:laptop-full",
+ runtimeMode: "full-access",
+ });
+ }),
+ ).pipe(Effect.provide(NodeHttpServer.layerTest)),
+);
+
+it.effect("refuses what a linked environment cannot take before anything leaves", () =>
+ Effect.scoped(
+ Effect.gen(function* () {
+ const seen = yield* Ref.make(nothingSeen);
+ const b = yield* serveBox(seen);
+ const a = yield* makeLaptop;
+ yield* linkTo(a.links, b, "auto");
+ yield* Ref.set(b.bearers, []);
+
+ const probes = yield* Ref.get(b.probes);
+ for (const name of ["schedule_task", "update_scheduled_task"]) {
+ const args =
+ name === "schedule_task"
+ ? { projectId: boxProject, prompt: "Handle {{body.action}}." }
+ : { scheduledTaskId: "scheduled-task:box" };
+ const signed = yield* a.call("thread:laptop-full", name, {
+ ...args,
+ environmentId: box.environmentId,
+ schedule: {
+ type: "webhook",
+ signature: {
+ header: "x-hub-signature-256",
+ encoding: "hex",
+ prefix: "sha256=",
+ secretRef: "secret:local-only",
+ },
+ },
+ });
+ expect(yield* Ref.get(b.probes)).toBe(probes);
+ expect(yield* Ref.get(b.bearers)).toEqual([]);
+ expect(failureOf(signed)).toMatchObject({
+ code: "invalid_request",
+ message: expect.stringContaining("signature.secretRef"),
+ });
+ expect(failureOf(signed).message).toContain("destination environment");
+
+ const ordinary = yield* a.call("thread:laptop-full", name, {
+ ...args,
+ environmentId: EnvironmentId.make("environment-elsewhere"),
+ schedule: { type: "interval", everyMs: 3_600_000 },
+ });
+ expect(failureOf(ordinary)).toMatchObject({
+ code: "invalid_request",
+ message: expect.stringContaining("not linked"),
+ });
+ }
+
+ const unknown = yield* a.call("thread:laptop-full", "t3_thread_read", {
+ environmentId: EnvironmentId.make("environment-elsewhere"),
+ threadId: boxThread,
+ });
+ expect(failureOf(unknown)).toMatchObject({ code: "invalid_request" });
+ expect(yield* Ref.get(b.bearers)).toEqual([]);
+
+ // The box sees only the link's credential, so one that may not control
+ // threads here may not reach them there either.
+ const previewOnly = new Set(["preview"]);
+ const links = yield* a.call("thread:laptop-full", "t3_environment_links", {}, previewOnly);
+ expect(failureOf(links)).toMatchObject({ code: "capability_denied" });
+ const read = yield* a.call(
+ "thread:laptop-full",
+ "t3_thread_configuration",
+ {
+ environmentId: box.environmentId,
+ threadId: boxThread,
+ },
+ previewOnly,
+ );
+ expect(failureOf(read)).toMatchObject({ code: "capability_denied" });
+ expect(yield* Ref.get(b.bearers)).toEqual([]);
+ }),
+ ).pipe(Effect.provide(NodeHttpServer.layerTest)),
+);
diff --git a/apps/server/src/peer/PeerForwarding.ts b/apps/server/src/peer/PeerForwarding.ts
new file mode 100644
index 0000000000000..e508af3d8ef2a
--- /dev/null
+++ b/apps/server/src/peer/PeerForwarding.ts
@@ -0,0 +1,195 @@
+import {
+ type EnvironmentId,
+ OrchestratorMcpFailure,
+ type OrchestratorMcpEnvironmentLinksResult,
+ OrchestratorMcpThreadWaitInput,
+ OrchestratorMcpThreadWaitResult,
+} from "@t3tools/contracts";
+import * as Context from "effect/Context";
+import * as Crypto from "effect/Crypto";
+import * as DateTime from "effect/DateTime";
+import * as Effect from "effect/Effect";
+import * as Hex from "effect/encoding/Hex";
+import * as Layer from "effect/Layer";
+import * as Schema from "effect/Schema";
+import type * as Tool from "effect/ai/Tool";
+
+import * as McpInvocationContext from "../mcp/McpInvocationContext.ts";
+import { readCaller, unavailable } from "../mcp/threadAccess.ts";
+import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
+import * as PeerLinks from "./PeerLinks.ts";
+import * as PeerMcpClient from "./PeerMcpClient.ts";
+
+/**
+ * A peer's `/mcp` holds a wait open for its whole timeout without sending
+ * anything, and T3 Connect's edge drops a request idle for about 100 s, so a
+ * forwarded wait goes out in pieces no longer than this.
+ */
+const FORWARDED_WAIT_CHUNK_MS = 50_000;
+const DEFAULT_WAIT_TIMEOUT_MS = 10 * 60 * 1_000;
+const MAX_WAIT_TIMEOUT_MS = 60 * 60 * 1_000;
+
+const hasScheduleSecretRef = Schema.is(
+ Schema.Struct({
+ schedule: Schema.Struct({
+ signature: Schema.Struct({ secretRef: Schema.String }),
+ }),
+ }),
+);
+
+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") {}
+
+export const remoteTarget = (scope: Scope, environmentId: EnvironmentId | undefined) =>
+ environmentId === scope.environmentId ? undefined : environmentId;
+
+const make = Effect.gen(function* () {
+ const peers = yield* PeerMcpClient.PeerMcpClient;
+ const links = yield* PeerLinks.PeerLinks;
+ const crypto = yield* Crypto.Crypto;
+ const threads = yield* ThreadManagement.ThreadManagementService;
+
+ /**
+ * The caller's own limits here, which travel with every call it forwards.
+ * The peer sees only the link's credential, so a credential that may not
+ * control threads here is refused here.
+ */
+ const callerLimits = (scope: Scope) =>
+ readCaller().pipe(
+ Effect.map((caller) => caller.limits),
+ Effect.provideService(McpInvocationContext.McpInvocationContext, scope),
+ Effect.provideService(ThreadManagement.ThreadManagementService, threads),
+ );
+
+ /**
+ * Every caller here reaches the peer as the one link session, so a retry
+ * key is scoped to its caller before it leaves: two agents here reusing a
+ * key never collide there, and one agent retrying gets its first result.
+ */
+ const scopedRequestId = (namespace: string, clientRequestId: string) =>
+ crypto.digest("SHA-256", new TextEncoder().encode(`${namespace}\n${clientRequestId}`)).pipe(
+ Effect.map((digest) => `fwd:${Hex.encode(digest)}`),
+ Effect.orDie,
+ );
+
+ const forward = (
+ scope: Scope,
+ toolName: string,
+ parameters: Schema.Top,
+ success: Schema.Top,
+ environmentId: EnvironmentId,
+ params: unknown,
+ ) =>
+ Effect.gen(function* () {
+ const limits = yield* callerLimits(scope);
+ const jsonParameters = Schema.toCodecJson(parameters as Schema.Encoder);
+ const encoded = (yield* Schema.encodeUnknownEffect(jsonParameters)(params).pipe(
+ Effect.mapError(unavailable),
+ )) as Record;
+ const { environmentId: _target, ...args } = encoded;
+ // A pending upload lives here, so it cannot go with the call.
+ if (Array.isArray(args.attachments) && args.attachments.length > 0) {
+ return yield* new OrchestratorMcpFailure({
+ code: "invalid_request",
+ message: "A launch in a linked environment cannot carry attachments yet.",
+ });
+ }
+ if (
+ (toolName === "schedule_task" || toolName === "update_scheduled_task") &&
+ hasScheduleSecretRef(args)
+ ) {
+ return yield* new OrchestratorMcpFailure({
+ code: "invalid_request",
+ message:
+ "A schedule in a linked environment cannot use signature.secretRef: secret references are local to this server and work once. Configure the webhook signing secret from a thread in the destination environment instead.",
+ });
+ }
+ if (typeof args.clientRequestId === "string") {
+ args.clientRequestId = yield* scopedRequestId(scope.requestNamespace, args.clientRequestId);
+ }
+ return yield* peers.call({
+ environmentId,
+ tool: toolName,
+ arguments: args,
+ limits,
+ success: Schema.toCodecJson(success as Schema.Decoder),
+ });
+ });
+
+ const waitForThread = (
+ scope: Scope,
+ environmentId: EnvironmentId,
+ input: OrchestratorMcpThreadWaitInput,
+ ) =>
+ Effect.gen(function* () {
+ let remaining = Math.min(
+ MAX_WAIT_TIMEOUT_MS,
+ Math.max(1, input.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS),
+ );
+ let runId = input.runId;
+ while (true) {
+ const chunk = Math.min(remaining, FORWARDED_WAIT_CHUNK_MS);
+ const result = yield* forward(
+ scope,
+ "t3_thread_wait",
+ OrchestratorMcpThreadWaitInput,
+ OrchestratorMcpThreadWaitResult,
+ environmentId,
+ { threadId: input.threadId, ...(runId === undefined ? {} : { runId }), timeoutMs: chunk },
+ );
+ remaining -= chunk;
+ if (!result.timedOut || remaining <= 0) return result;
+ // Later pieces wait for the run the first one picked, not a newer one.
+ runId = result.runId ?? undefined;
+ }
+ });
+
+ const call: PeerForwarding["Service"]["call"] = (scope, tool, environmentId, params) =>
+ tool.name === "t3_thread_wait"
+ ? (waitForThread(
+ scope,
+ environmentId,
+ params as OrchestratorMcpThreadWaitInput,
+ ) as Effect.Effect, OrchestratorMcpFailure>)
+ : forward(scope, tool.name, tool.parametersSchema, tool.successSchema, environmentId, params);
+
+ const listLinks: PeerForwarding["Service"]["links"] = (scope) =>
+ Effect.gen(function* () {
+ yield* callerLimits(scope);
+ const listed = yield* links.list.pipe(
+ Effect.mapError(
+ (error) =>
+ new OrchestratorMcpFailure({ code: "orchestration_error", message: error.message }),
+ ),
+ );
+ return {
+ environmentId: scope.environmentId,
+ links: listed.map((link) => ({
+ environmentId: link.environmentId,
+ label: link.label,
+ status: link.status,
+ access: link.access,
+ expiresAt: DateTime.formatIso(link.expiresAt),
+ lastError: link.lastError,
+ })),
+ };
+ });
+
+ return PeerForwarding.of({ call, links: listLinks });
+});
+
+export const layer = Layer.effect(PeerForwarding, make);
diff --git a/apps/server/src/peer/PeerLinks.test.ts b/apps/server/src/peer/PeerLinks.test.ts
index d910d8677584a..b226fee115d35 100644
--- a/apps/server/src/peer/PeerLinks.test.ts
+++ b/apps/server/src/peer/PeerLinks.test.ts
@@ -1,11 +1,6 @@
import { NodeHttpServer } from "@effect/platform-node";
import * as NodeServices from "@effect/platform-node/NodeServices";
-import {
- EnvironmentHttpApi,
- EnvironmentId,
- type ExecutionEnvironmentDescriptor,
- OrchestratorMcpFailure,
-} from "@t3tools/contracts";
+import { type ExecutionEnvironmentDescriptor, OrchestratorMcpFailure } from "@t3tools/contracts";
import { expect, it } from "@effect/vitest";
import * as Context from "effect/Context";
import * as Deferred from "effect/Deferred";
@@ -21,54 +16,21 @@ import * as Schema from "effect/Schema";
import * as SqlClient from "effect/sql/SqlClient";
import * as TestClock from "effect/testing/TestClock";
import { McpSchema, Tool, Toolkit } from "effect/ai";
-import {
- FetchHttpClient,
- HttpRouter,
- HttpServer,
- HttpServerRequest,
- HttpServerResponse,
-} from "effect/http";
-import * as HttpApi from "effect/http-api/HttpApi";
-import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
+import * as FetchHttpClient from "effect/http/FetchHttpClient";
-import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts";
-import * as AuthHttp from "../auth/http.ts";
-import * as McpOAuth from "../auth/McpOAuth.ts";
-import * as McpOAuthHttp from "../auth/mcpOAuthHttp.ts";
import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
import * as ServerConfig from "../config.ts";
import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
-import * as ServerHttp from "../http.ts";
import * as McpHttpServer from "../mcp/McpHttpServer.ts";
import * as McpInvocationContext from "../mcp/McpInvocationContext.ts";
-import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts";
import * as McpToolAccess from "../mcp/McpToolAccess.ts";
import * as Sqlite from "../persistence/Sqlite.ts";
import * as PeerLinks from "./PeerLinks.ts";
+import { descriptorOf, servePeer } from "./PeerLinks.testkit.ts";
import * as PeerMcpClient from "./PeerMcpClient.ts";
-class PeerTestApi extends HttpApi.make("environment")
- .add(EnvironmentHttpApi.groups.metadata)
- .add(EnvironmentHttpApi.groups.mcpOAuth) {}
-
-const descriptorOf = (
- environmentId: string,
- label: string,
- capabilities: Partial = {},
-): ExecutionEnvironmentDescriptor => ({
- environmentId: EnvironmentId.make(environmentId),
- label,
- platform: { os: "linux", arch: "x64" },
- serverVersion: "0.0.0-test",
- capabilities: { repositoryIdentity: true, mcpModeLimitHeader: true, ...capabilities },
-});
-
/** What B's probe saw: the calling client and the modes it may start work with. */
const Probe = Schema.Struct({ caller: Schema.String, modes: Schema.String });
-const decodeMcpRequest = Schema.decodeUnknownEffect(
- Schema.Struct({ method: Schema.String, id: Schema.optional(Schema.Number) }),
-);
-
const ProbeToolkit = Toolkit.make(
Tool.make("probe", {
description: "Reports the calling client and the modes it may start work with.",
@@ -85,155 +47,34 @@ const ProbeToolkit = Toolkit.make(
}),
);
-/**
- * Environment B as it runs: its descriptor, MCP OAuth, and `/mcp` behind the
- * real client authenticator, on a real socket. The descriptor can be swapped,
- * as when B's address comes to answer as another environment.
- */
const serveB = (initial: ExecutionEnvironmentDescriptor) =>
- Effect.gen(function* () {
- const descriptor = yield* Ref.make(initial);
- const alternateDescriptor = yield* Ref.make(initial);
- const bearers = yield* Ref.make>([]);
- const probes = yield* Ref.make(0);
- const redirects = yield* Ref.make>(new Map());
- const redirectedBodies = yield* Ref.make>([]);
- const toolResult = yield* Ref.make(null);
- const sse = yield* Ref.make(false);
- const replies = yield* Ref.make>(
- new Map(),
- );
- const layerEnvironment = Layer.succeed(ServerEnvironment.ServerEnvironment, {
- getEnvironmentId: Ref.get(descriptor).pipe(Effect.map((current) => current.environmentId)),
- getDescriptor: Ref.get(descriptor),
- });
- const authContext = yield* EnvironmentAuth.layer.pipe(
- Layer.provide(Sqlite.layerMemory),
- Layer.provideMerge(ServerSecretStore.layer),
- Layer.provideMerge(ServerEnvironment.layerIdentity),
- Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "t3-peer-link-b-" })),
- Layer.provideMerge(NodeServices.layer),
- Layer.fresh,
- Layer.build,
- );
- // Every bearer B receives, on any route, so a test can show one was never sent.
- const layerRecordBearers = HttpRouter.middleware(
- (httpEffect) =>
- Effect.gen(function* () {
- const request = yield* HttpServerRequest.HttpServerRequest;
- if (request.url.startsWith("/.well-known/t3/environment")) {
- yield* Ref.update(probes, (count) => count + 1);
- if (request.headers.host?.startsWith("localhost:")) {
- return HttpServerResponse.jsonUnsafe(yield* Ref.get(alternateDescriptor));
- }
- }
- const redirect = (yield* Ref.get(redirects)).get(request.url);
- if (redirect !== undefined) {
- return HttpServerResponse.redirect(redirect, { status: 307 });
- }
- if (request.url === "/redirect-target") {
- const body = yield* request.text;
- yield* Ref.update(redirectedBodies, (bodies) => [...bodies, body]);
- return HttpServerResponse.empty({ status: 400 });
- }
- const authorization = request.headers.authorization;
- if (authorization !== undefined) {
- yield* Ref.update(bearers, (seen) => [...seen, authorization]);
- }
- const reply = (yield* Ref.get(replies)).get(request.url);
- if (reply !== undefined)
- return HttpServerResponse.jsonUnsafe(reply.body, { status: reply.status });
- const response = yield* httpEffect;
- // Keep real authentication and dispatch, but exercise richer peer wire results.
- const result = yield* Ref.get(toolResult);
- if (result !== null && request.url === "/mcp" && response.status === 200) {
- const rpc = yield* decodeMcpRequest(yield* request.json);
- if (rpc.method === "tools/call") {
- if (yield* Ref.get(sse)) {
- return HttpServerResponse.text(
- [
- 'data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"result":"notification"}}',
- "",
- 'data: {"jsonrpc":"2.0","id":999,"error":{"code":-1,"message":"other request"}}',
- "",
- 'data: {"jsonrpc":"2.0",',
- 'data: "id":' + rpc.id + ',"result":' + JSON.stringify(result) + "}",
- "",
- "",
- ].join("\r\n"),
- { contentType: "text/event-stream" },
- );
- }
- return HttpServerResponse.jsonUnsafe({ jsonrpc: "2.0", id: rpc.id, result });
- }
- }
- return response;
- }),
- { global: true },
- );
- const layerRoutes = Layer.mergeAll(
- HttpApiBuilder.layer(PeerTestApi).pipe(
- Layer.provide(McpOAuthHttp.layer.pipe(Layer.provide(McpOAuth.layer))),
- Layer.provide(ServerHttp.layerServerEnvironmentHttpApi),
- Layer.provide(AuthHttp.layerAuthenticatedAuth),
- ),
- McpHttpServer.toolkitRegistration(
- ProbeToolkit,
- McpToolAccess.toLayer(ProbeToolkit, {
- probe: McpToolAccess.reads(() =>
- McpInvocationContext.McpInvocationContext.pipe(
- Effect.map((scope) => {
- const modes = McpInvocationContext.clientModeCeiling(scope.client);
- return {
- caller: scope.client?.label ?? "none",
- modes: `${modes.runtimeMode}/${modes.interactionMode}`,
- };
- }),
- ),
+ servePeer(
+ initial,
+ McpHttpServer.toolkitRegistration(
+ ProbeToolkit,
+ McpToolAccess.toLayer(ProbeToolkit, {
+ probe: McpToolAccess.reads(() =>
+ McpInvocationContext.McpInvocationContext.pipe(
+ Effect.map((scope) => {
+ const modes = McpInvocationContext.clientModeCeiling(scope.client);
+ return {
+ caller: scope.client?.label ?? "none",
+ modes: `${modes.runtimeMode}/${modes.interactionMode}`,
+ };
+ }),
),
- refuse: McpToolAccess.reads(() =>
- Effect.fail(
- new OrchestratorMcpFailure({
- code: "thread_not_found",
- message: "No such thread.",
- }),
- ),
+ ),
+ refuse: McpToolAccess.reads(() =>
+ Effect.fail(
+ new OrchestratorMcpFailure({
+ code: "thread_not_found",
+ message: "No such thread.",
+ }),
),
- }),
- ).pipe(
- Layer.provideMerge(McpHttpServer.layerMcpTransport),
- Layer.provide(
- Layer.mock(McpSessionRegistry.McpSessionRegistry)({
- resolve: () => Effect.succeed(undefined),
- }),
),
- Layer.provide(McpOAuth.layerMcpClientAuthenticator),
- ),
- ).pipe(
- Layer.provide(layerRecordBearers),
- Layer.provide(layerEnvironment),
- Layer.provide(Layer.succeedContext(authContext)),
- );
- yield* HttpRouter.serve(layerRoutes, { disableListenLog: true, disableLogger: true }).pipe(
- Layer.build,
- );
- const address = (yield* HttpServer.HttpServer).address;
- if (address._tag === "UnixPathAddress") return yield* Effect.die("expected a TCP address");
- return {
- url: `http://127.0.0.1:${address.port}`,
- auth: Context.get(authContext, EnvironmentAuth.EnvironmentAuth),
- descriptor,
- alternateDescriptor,
- bearers,
- toolResult,
- sse,
- replies,
- redirects,
- redirectedBodies,
- /** How often its descriptor was asked for, as a probe of whether it answers. */
- probes,
- };
- });
+ }),
+ ),
+ );
/** Environment A: its own database and secret store, reaching B over plain HTTP. */
const makeA = (
diff --git a/apps/server/src/peer/PeerLinks.testkit.ts b/apps/server/src/peer/PeerLinks.testkit.ts
new file mode 100644
index 0000000000000..f8e6a539e3fa1
--- /dev/null
+++ b/apps/server/src/peer/PeerLinks.testkit.ts
@@ -0,0 +1,227 @@
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import {
+ EnvironmentHttpApi,
+ EnvironmentId,
+ type AuthMcpClientAccess,
+ type ExecutionEnvironmentDescriptor,
+} from "@t3tools/contracts";
+import * as Context from "effect/Context";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as Schema from "effect/Schema";
+import * as McpSchema from "effect/ai/McpSchema";
+import * as Ref from "effect/Ref";
+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 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 McpSessionRegistry from "../mcp/McpSessionRegistry.ts";
+import * as Sqlite from "../persistence/Sqlite.ts";
+import * as PeerLinks from "./PeerLinks.ts";
+import * as PeerMcpClient from "./PeerMcpClient.ts";
+
+const decodeMcpRequest = Schema.decodeUnknownEffect(
+ Schema.Struct({ method: Schema.String, id: Schema.optional(Schema.Number) }),
+);
+const encodeToolResult = Schema.encodeSync(Schema.fromJsonString(McpSchema.CallToolResult));
+
+class PeerTestApi extends HttpApi.make("environment")
+ .add(EnvironmentHttpApi.groups.metadata)
+ .add(EnvironmentHttpApi.groups.mcpOAuth) {}
+
+export const descriptorOf = (
+ environmentId: string,
+ label: string,
+ capabilities: Partial = {},
+): ExecutionEnvironmentDescriptor => ({
+ environmentId: EnvironmentId.make(environmentId),
+ label,
+ platform: { os: "linux", arch: "x64" },
+ serverVersion: "0.0.0-test",
+ capabilities: { repositoryIdentity: true, mcpModeLimitHeader: true, ...capabilities },
+});
+
+/**
+ * The environment a link points at, as it runs: its descriptor, MCP OAuth,
+ * and `/mcp` with `tools` behind the real client authenticator, on a real
+ * socket. The descriptor can be swapped, as when its address comes to answer
+ * as another environment, and every bearer it receives is recorded.
+ */
+export const servePeer = (
+ initial: ExecutionEnvironmentDescriptor,
+ tools: Layer.Layer,
+) =>
+ Effect.gen(function* () {
+ const descriptor = yield* Ref.make(initial);
+ const alternateDescriptor = yield* Ref.make(initial);
+ const bearers = yield* Ref.make>([]);
+ const probes = yield* Ref.make(0);
+ const redirects = yield* Ref.make>(new Map());
+ const redirectedBodies = yield* Ref.make>([]);
+ const toolResult = yield* Ref.make(null);
+ const sse = yield* Ref.make(false);
+ const replies = yield* Ref.make>(
+ new Map(),
+ );
+ const layerEnvironment = Layer.succeed(ServerEnvironment.ServerEnvironment, {
+ getEnvironmentId: Ref.get(descriptor).pipe(Effect.map((current) => current.environmentId)),
+ getDescriptor: Ref.get(descriptor),
+ });
+ const authContext = yield* EnvironmentAuth.layer.pipe(
+ Layer.provide(Sqlite.layerMemory),
+ Layer.provideMerge(ServerSecretStore.layer),
+ Layer.provideMerge(ServerEnvironment.layerIdentity),
+ Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "t3-peer-link-peer-" })),
+ Layer.provideMerge(NodeServices.layer),
+ Layer.fresh,
+ Layer.build,
+ );
+ // Every bearer B receives, on any route, so a test can show one was never sent.
+ const layerRecordBearers = HttpRouter.middleware(
+ (httpEffect) =>
+ Effect.gen(function* () {
+ const request = yield* HttpServerRequest.HttpServerRequest;
+ if (request.url.startsWith("/.well-known/t3/environment")) {
+ yield* Ref.update(probes, (count) => count + 1);
+ if (request.headers.host?.startsWith("localhost:")) {
+ return HttpServerResponse.jsonUnsafe(yield* Ref.get(alternateDescriptor));
+ }
+ }
+ const redirect = (yield* Ref.get(redirects)).get(request.url);
+ if (redirect !== undefined) {
+ return HttpServerResponse.redirect(redirect, { status: 307 });
+ }
+ if (request.url === "/redirect-target") {
+ const body = yield* request.text;
+ yield* Ref.update(redirectedBodies, (bodies) => [...bodies, body]);
+ return HttpServerResponse.empty({ status: 400 });
+ }
+ const authorization = request.headers.authorization;
+ if (authorization !== undefined) {
+ yield* Ref.update(bearers, (seen) => [...seen, authorization]);
+ }
+ const reply = (yield* Ref.get(replies)).get(request.url);
+ if (reply !== undefined)
+ return HttpServerResponse.jsonUnsafe(reply.body, { status: reply.status });
+ const response = yield* httpEffect;
+ // Keep real authentication and dispatch, but exercise richer peer wire results.
+ const result = yield* Ref.get(toolResult);
+ if (result !== null && request.url === "/mcp" && response.status === 200) {
+ const rpc = yield* decodeMcpRequest(yield* request.json);
+ if (rpc.method === "tools/call") {
+ if (yield* Ref.get(sse)) {
+ return HttpServerResponse.text(
+ [
+ 'data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"result":"notification"}}',
+ "",
+ 'data: {"jsonrpc":"2.0","id":999,"error":{"code":-1,"message":"other request"}}',
+ "",
+ 'data: {"jsonrpc":"2.0",',
+ 'data: "id":' + rpc.id + ',"result":' + encodeToolResult(result) + "}",
+ "",
+ "",
+ ].join("\r\n"),
+ { contentType: "text/event-stream" },
+ );
+ }
+ return HttpServerResponse.jsonUnsafe({ jsonrpc: "2.0", id: rpc.id, result });
+ }
+ }
+ return response;
+ }),
+ { global: true },
+ );
+ const layerRoutes = Layer.mergeAll(
+ HttpApiBuilder.layer(PeerTestApi).pipe(
+ Layer.provide(McpOAuthHttp.layer.pipe(Layer.provide(McpOAuth.layer))),
+ Layer.provide(ServerHttp.layerServerEnvironmentHttpApi),
+ Layer.provide(AuthHttp.layerAuthenticatedAuth),
+ ),
+ tools.pipe(
+ Layer.provideMerge(McpHttpServer.layerMcpTransport),
+ Layer.provide(
+ Layer.mock(McpSessionRegistry.McpSessionRegistry)({
+ resolve: () => Effect.succeed(undefined),
+ }),
+ ),
+ Layer.provide(McpOAuth.layerMcpClientAuthenticator),
+ ),
+ ).pipe(
+ Layer.provide(layerRecordBearers),
+ Layer.provide(layerEnvironment),
+ Layer.provide(Layer.succeedContext(authContext)),
+ );
+ // Keep the peer's toolkit and MCP server separate from the linking environment.
+ yield* HttpRouter.serve(layerRoutes, { disableListenLog: true, disableLogger: true }).pipe(
+ Layer.fresh,
+ Layer.build,
+ );
+ const address = (yield* HttpServer.HttpServer).address;
+ if (address._tag === "UnixPathAddress") return yield* Effect.die("expected a TCP address");
+ return {
+ url: `http://127.0.0.1:${address.port}`,
+ auth: Context.get(authContext, EnvironmentAuth.EnvironmentAuth),
+ descriptor,
+ alternateDescriptor,
+ bearers,
+ toolResult,
+ sse,
+ replies,
+ redirects,
+ redirectedBodies,
+ /** How often its descriptor was asked for, as a probe of whether it answers. */
+ probes,
+ };
+ });
+
+export type ServedPeer = Effect.Success>;
+
+/**
+ * The environment that links: its own database and secret store, reaching
+ * peers over plain HTTP. Provide it to anything built on its links.
+ */
+export const layerLinkingEnvironment = (descriptor: ExecutionEnvironmentDescriptor) =>
+ PeerMcpClient.layer.pipe(
+ Layer.provideMerge(PeerLinks.layer),
+ Layer.provide(
+ Layer.mergeAll(
+ Sqlite.layerMemory,
+ ServerSecretStore.layer,
+ FetchHttpClient.layer,
+ Layer.succeed(ServerEnvironment.ServerEnvironment, {
+ getEnvironmentId: Effect.succeed(descriptor.environmentId),
+ getDescriptor: Effect.succeed(descriptor),
+ }),
+ ),
+ ),
+ Layer.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-peer-link-here-" })),
+ Layer.provideMerge(NodeServices.layer),
+ Layer.fresh,
+ );
+
+/** Links `links` to `peer` with a fresh pairing code from it. */
+export const linkTo = (
+ links: PeerLinks.PeerLinks["Service"],
+ peer: ServedPeer,
+ access: AuthMcpClientAccess = "auto",
+) =>
+ peer.auth
+ .issuePairingCredential()
+ .pipe(
+ Effect.flatMap((pairing) =>
+ links.link({ url: peer.url, pairingCode: pairing.credential, access }),
+ ),
+ );
diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts
index e3ea81af29d64..d1abdea82e0ef 100644
--- a/apps/server/src/server.ts
+++ b/apps/server/src/server.ts
@@ -64,6 +64,7 @@ import * as ProviderInstanceRegistryHydration from "./provider/ProviderInstanceR
import * as ProjectSettingsOverrideSweep from "./provider/ProjectSettingsOverrideSweep.ts";
import * as TerminalManager from "./terminal/Manager.ts";
import * as McpHttpServer from "./mcp/McpHttpServer.ts";
+import * as PeerForwarding from "./peer/PeerForwarding.ts";
import * as PeerLinks from "./peer/PeerLinks.ts";
import * as PeerMcpClient from "./peer/PeerMcpClient.ts";
import * as McpSessionRegistry from "./mcp/McpSessionRegistry.ts";
@@ -708,7 +709,12 @@ const layerMakeRoutes = Layer.mergeAll(
Layer.provide(layerPullRequestService),
Layer.provide(ServerBrowser.layer.pipe(Layer.provide(DesktopBrowserChannel.layer))),
Layer.provide(PreviewBrowser.layer),
- Layer.provide(PeerMcpClient.layer.pipe(Layer.provideMerge(PeerLinks.layer))),
+ Layer.provide(
+ PeerForwarding.layer.pipe(
+ Layer.provideMerge(PeerMcpClient.layer),
+ Layer.provideMerge(PeerLinks.layer),
+ ),
+ ),
Layer.provide(PreviewAutomationBroker.layer),
Layer.provide(ServerSelfUpdate.layer.pipe(Layer.provide(layerDesktopAppUpdate))),
Layer.provide(layerCommandReadiness),
diff --git a/docs/user/outside-agents.md b/docs/user/outside-agents.md
index 7cc6692e38f6c..ebbaf7ad9dcfe 100644
--- a/docs/user/outside-agents.md
+++ b/docs/user/outside-agents.md
@@ -88,6 +88,44 @@ Any agent that supports remote MCP servers over HTTP with OAuth can connect. Giv
it the MCP URL and choose OAuth, and the agent finds the sign-in settings itself.
Agents without OAuth support cannot connect.
+## Link from another T3 Code environment
+
+On the host that will make requests, link the other environment with a pairing
+code from its **Settings → Connections** or `t3 auth pairing create`:
+
+```sh
+t3 environment link https:// --pairing-code --access approval-required
+```
+
+Use HTTPS, or HTTP with a loopback address, a Tailscale `.ts.net` name, or a
+Tailscale IPv6 address. Plain HTTP LAN and Tailscale IPv4 addresses are not
+supported. The link belongs to the host where you ran the command and
+appears on the peer as **T3 Code · **. Calls keep both the link's
+access ceiling and the calling agent's narrower modes.
+
+Use `t3 environment list` to check reachability and
+`t3 environment get --json` to read a stored link without
+contacting the peer. Run `t3 environment unlink ` to forget it
+locally. To revoke its access on the peer, remove its session in that peer's
+Connections settings. After a link expires, link again with a new pairing code.
+
+Agents can call `t3_environment_links`, then pass a listed `environmentId` to
+tools that accept it. Projects, threads, runs, providers and models must come
+from that destination; supply thread or project IDs explicitly. This includes
+project creation, thread launch/read/send/wait, queues, pending requests and
+ordinary scheduling. `delegate_task` stays local.
+
+Remote launches cannot carry attachments. Remote schedules cannot use a
+`signature.secretRef` created here; configure signed webhooks from a thread
+in the destination instead. Waits use requests of at most 50 seconds and keep
+following the originally selected run.
+
+`t3_thread_interrupt` with `stop:true` also holds the queue, ends pull request
+watches and stops delegated work. A partial retry can hold the current queue
+and end current watches on the original still-current run; it does not stop
+later-created runs or descendants. `interrupt_requested` confirms acceptance,
+not termination. Older peers that ignore `stop` only interrupt the run.
+
## Remove an agent
Approved agents appear under **Settings → Connections** like other clients.
diff --git a/packages/client-runtime/src/t3ToolSummary.ts b/packages/client-runtime/src/t3ToolSummary.ts
index 25878d4ff5bc7..338118aafbf81 100644
--- a/packages/client-runtime/src/t3ToolSummary.ts
+++ b/packages/client-runtime/src/t3ToolSummary.ts
@@ -333,6 +333,9 @@ export function summarizeT3ToolCalls(
case "environment-update":
label = phrase("Updated", "update", `environment preferences ${times}`);
break;
+ case "environment-links":
+ label = phrase("Listed", "list", `linked environments ${times}`);
+ break;
case "attachment-prepare":
label = phrase(
"Prepared",
diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts
index 94a9eca331d13..7baf5c05a713b 100644
--- a/packages/contracts/src/orchestrationV2.ts
+++ b/packages/contracts/src/orchestrationV2.ts
@@ -3034,6 +3034,7 @@ const OrchestrationV2InternalCommand = Schema.Union([
commandId: CommandId,
threadId: ThreadId,
reason: Schema.optional(Schema.String),
+ throughSequence: Schema.optional(NonNegativeInt),
}),
/**
* Internal so no client can mark a request saved without the value being
diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts
index 271e585eabd63..74f7b40479114 100644
--- a/packages/contracts/src/orchestratorMcp.ts
+++ b/packages/contracts/src/orchestratorMcp.ts
@@ -2,8 +2,10 @@ import * as Effect from "effect/Effect";
import * as Schema from "effect/Schema";
import * as SchemaTransformation from "effect/SchemaTransformation";
+import { AuthMcpClientAccess } from "./auth.ts";
import {
ContextTransferId,
+ EnvironmentId,
IsoDateTime,
MessageId,
NodeId,
@@ -24,6 +26,7 @@ import {
} from "./scheduledTask.ts";
import { ProviderInteractionMode, RuntimeMode } from "./providerPolicy.ts";
import { PullRequestState } from "./pullRequest.ts";
+import { PeerLinkStatus } from "./peerLink.ts";
import {
ThreadLinkedPullRequest,
ThreadPullRequestKey,
@@ -295,7 +298,20 @@ const OrchestratorMcpProjectTarget = Schema.optional(
}),
);
+/**
+ * A linked environment to act in instead of this one. The tools that take it
+ * forward the call there, where the link's access and the caller's own modes
+ * both apply.
+ */
+export const OrchestratorMcpEnvironmentTarget = Schema.optional(
+ EnvironmentId.annotate({
+ description:
+ "A linked environment to do this in; t3_environment_links lists them. Omit for this environment. Ids (projects, threads, runs) then belong to that environment.",
+ }),
+);
+
export const OrchestratorMcpThreadListInput = Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
projectId: OrchestratorMcpProjectTarget,
statuses: Schema.optional(
Schema.Array(OrchestratorMcpThreadStatus).check(Schema.isMaxLength(10)),
@@ -354,6 +370,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 +458,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 +477,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,9 +493,16 @@ 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))),
+ /**
+ * Stop the thread as its Stop button does instead: interrupt its running
+ * turn whichever run that is, hold its queued messages, end its pull request
+ * watches and stop every task it delegated. `runId` is then not used.
+ */
+ stop: Schema.optional(Schema.Boolean),
clientRequestId: Schema.optional(OrchestratorMcpClientRequestId),
});
export type OrchestratorMcpThreadInterruptInput = typeof OrchestratorMcpThreadInterruptInput.Type;
@@ -510,6 +536,29 @@ export const OrchestratorMcpProviderCapability = Schema.Struct({
});
export type OrchestratorMcpProviderCapability = typeof OrchestratorMcpProviderCapability.Type;
+export const OrchestratorMcpCapabilitiesInput = Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
+});
+export type OrchestratorMcpCapabilitiesInput = typeof OrchestratorMcpCapabilitiesInput.Type;
+
+export const OrchestratorMcpEnvironmentLink = Schema.Struct({
+ environmentId: EnvironmentId,
+ label: Schema.String,
+ status: PeerLinkStatus,
+ /** The most the linked environment lets this one's agents do there. */
+ access: AuthMcpClientAccess,
+ expiresAt: IsoDateTime,
+ lastError: Schema.NullOr(Schema.String),
+});
+export type OrchestratorMcpEnvironmentLink = typeof OrchestratorMcpEnvironmentLink.Type;
+
+export const OrchestratorMcpEnvironmentLinksResult = Schema.Struct({
+ environmentId: EnvironmentId,
+ links: Schema.Array(OrchestratorMcpEnvironmentLink),
+});
+export type OrchestratorMcpEnvironmentLinksResult =
+ typeof OrchestratorMcpEnvironmentLinksResult.Type;
+
export const OrchestratorMcpCapabilitiesResult = Schema.Struct({
parentThreadId: Schema.NullOr(ThreadId),
inheritedProviderInstanceId: Schema.NullOr(ProviderInstanceId),
@@ -531,6 +580,7 @@ export const OrchestratorMcpCapabilitiesResult = Schema.Struct({
export type OrchestratorMcpCapabilitiesResult = typeof OrchestratorMcpCapabilitiesResult.Type;
export const OrchestratorMcpScheduleTaskInput = Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
projectId: OrchestratorMcpProjectTarget,
prompt: OrchestratorMcpPrompt.annotate({
description: "Prompt executed on every scheduled run.",
@@ -574,6 +624,7 @@ export const OrchestratorMcpScheduleTaskResult = OrchestratorMcpScheduledTask;
export type OrchestratorMcpScheduleTaskResult = typeof OrchestratorMcpScheduleTaskResult.Type;
export const OrchestratorMcpListScheduledTasksInput = Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
projectId: Schema.optional(
ProjectId.annotate({
description:
@@ -591,6 +642,7 @@ export type OrchestratorMcpListScheduledTasksResult =
typeof OrchestratorMcpListScheduledTasksResult.Type;
export const OrchestratorMcpUpdateScheduledTaskInput = Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
scheduledTaskId: ScheduledTaskId,
prompt: Schema.optional(OrchestratorMcpPrompt),
title: Schema.optional(OrchestratorMcpTitle),
@@ -640,6 +692,7 @@ export const OrchestratorMcpRequestSecretResult = Schema.Union([
export type OrchestratorMcpRequestSecretResult = typeof OrchestratorMcpRequestSecretResult.Type;
export const OrchestratorMcpDeleteScheduledTaskInput = Schema.Struct({
+ environmentId: OrchestratorMcpEnvironmentTarget,
scheduledTaskId: ScheduledTaskId,
});
export type OrchestratorMcpDeleteScheduledTaskInput =
diff --git a/packages/provider-core/src/server/orchestrationInstructions.ts b/packages/provider-core/src/server/orchestrationInstructions.ts
index c93c730ba094d..767cd99061178 100644
--- a/packages/provider-core/src/server/orchestrationInstructions.ts
+++ b/packages/provider-core/src/server/orchestrationInstructions.ts
@@ -14,8 +14,9 @@ The \`t3-code\` MCP server provides app-owned orchestration. Treat these concept
- \`t3_thread_launch\` and \`create_threads\` create ordinary top-level T3 conversations. Use them only when the user explicitly asks for separate/new/top-level threads or conversations. Never use them merely because the user said "subagent" or requested parallel delegated work.
- For every T3 delegated review round, call \`delegate_task\` again. Include the original brief, prior findings, responses, and unresolved objections in each new task prompt. Track each round by its own \`taskId\`. Use a distinct \`clientRequestId\` per round, stable across retries of that round. Do not use \`t3_thread_send\` on \`childThreadId\` to continue a delegated review.`,
`- \`schedule_task\` creates persistent recurring work in the app scheduler. Pass \`schedule\` as a structured object, never as JSON text: \`{"type":"interval","everyMs":3600000}\` for an interval, or \`{"type":"fixed_time","timeOfDay":"09:00","weekdays":[1,2,3,4,5]}\` for a wall-clock schedule, or \`{"type":"webhook"}\` to run on each request to the returned \`webhookUrl\` (the run sees the request only through \`{{body.path}}\`-style placeholders in the prompt). By default runs return to the current thread, which suits orchestrating: each trigger arrives here and you delegate or dedupe; set \`bindToCurrentThread=false\` only when the user wants a fresh thread for every run. After scheduling a timer, report the returned cadence and next run time; for a webhook, report its \`webhookUrl\`, or say T3 Connect remote access is needed if it is missing.
+- \`t3_environment_links\` lists the user's other T3 Code environments that this one is linked to, such as another machine. To work there, pass that \`environmentId\` to any T3 tool that takes one, such as \`t3_thread_launch\`, \`t3_thread_send\`, \`t3_pending_request_respond\`, \`t3_queue_list\`, or \`t3_project_list\`; project, thread, run, and model ids then belong to that environment, so pass \`threadId\` explicitly. Work there runs within your own modes and the link's access.
- When you need a secret from the user (a token, API key, or webhook signing secret), call \`request_secret\` so they enter it privately, then pass the returned \`secretRef\` to the tool that needs it, e.g. \`signature.secretRef\` on a webhook task for a sender that signs requests such as GitHub. A \`secretRef\` works once. Never ask for a secret in chat, never invent one, and never repeat one.
-- To mention another thread to the user, link it as \`[title](t3-thread://v1/)\` with its exact \`threadId\`, not URL-encoded. T3 Code opens the thread in the app and shows its current title.`,
+- To mention another thread to the user, link it as \`[title](t3-thread://v1/)\` with its exact \`threadId\`, not URL-encoded. T3 Code opens the thread in the app and shows its current title. Name a thread in a linked environment by title instead: its link would open here.`,
`### Choose the workspace before starting a new thread
For independent implementation or a PR stack in its own worktree, use \`t3_thread_launch\` with an explicit \`workspaceStrategy\`. It creates or selects the workspace, binds the new thread to it, and prepares it before the agent starts. Put the task in \`message\`, not \`prompt\`:
diff --git a/packages/shared/src/t3McpToolPresentation.ts b/packages/shared/src/t3McpToolPresentation.ts
index b18351cdd2c49..3fb210e16791b 100644
--- a/packages/shared/src/t3McpToolPresentation.ts
+++ b/packages/shared/src/t3McpToolPresentation.ts
@@ -50,6 +50,7 @@ export type T3McpToolSummaryAction =
| "project-delete"
| "project-clone"
| "environment-read"
+ | "environment-links"
| "environment-update"
| "attachment-prepare"
| "attachment-discard"
@@ -309,6 +310,10 @@ const T3_MCP_TOOLS: Readonly> = {
["Read", "Reading", "Read", "environment preferences"],
"environment-read",
),
+ t3_environment_links: tool(
+ ["List", "Listing", "Listed", "linked environments"],
+ "environment-links",
+ ),
t3_environment_preferences_update: tool(
["Update", "Updating", "Updated", "environment preferences"],
"environment-update",