From be21eaef404e5867ea2cec6229b7b6695e2eb8d7 Mon Sep 17 00:00:00 2001 From: Jake Leventhal Date: Mon, 5 Oct 2026 00:51:02 -0400 Subject: [PATCH] feat(mcp): agents can launch threads on another connected environment A thread's t3-code MCP server only knew its own environment, so an agent could not start work on another machine the user had connected. Environments still never talk to each other. The server hands the request to a connected app, and the app runs it over the session it already has with the target, so no credential is stored and the target enforces that app's permissions. - t3_environment_list and t3_environment_catalog discover other environments and their projects and provider instances - t3_thread_launch, t3_thread_read and t3_thread_wait take an optional environmentId; omitted means local, unchanged - the shared client runtime registers each app as a carrier with every environment that advertises the peerEnvironments capability Co-Authored-By: Claude Fable 5.1 --- apps/server/src/auth/RpcAuthorization.ts | 2 + .../src/environment/ServerEnvironment.ts | 1 + apps/server/src/mcp/McpHttpServer.ts | 3 +- apps/server/src/mcp/OrchestratorMcpService.ts | 64 +-- .../src/mcp/PeerEnvironmentBroker.test.ts | 154 ++++++ apps/server/src/mcp/PeerEnvironmentBroker.ts | 290 +++++++++++ .../src/mcp/PeerEnvironmentService.test.ts | 316 ++++++++++++ apps/server/src/mcp/PeerEnvironmentService.ts | 453 ++++++++++++++++++ apps/server/src/mcp/toolkits/core.test.ts | 2 + .../src/mcp/toolkits/environment/handlers.ts | 19 + .../src/mcp/toolkits/environment/tools.ts | 38 +- .../src/mcp/toolkits/orchestrator/handlers.ts | 9 + .../mcp/toolkits/orchestrator/tools.test.ts | 21 + .../src/mcp/toolkits/orchestrator/tools.ts | 16 +- .../src/mcp/toolkits/project/handlers.test.ts | 4 + .../src/mcp/toolkits/project/handlers.ts | 16 + apps/server/src/mcp/toolkits/project/tools.ts | 9 +- .../toolkits/worktree/registration.test.ts | 2 + .../provider/T3OrchestrationInstructions.ts | 10 + apps/server/src/server.ts | 2 + apps/server/src/ws.ts | 16 + docs/internals/remote.md | 20 + .../orchestrator-mcp-server.md | 22 + docs/user/remote-access.md | 14 + .../client-runtime/src/connection/layer.ts | 2 + .../src/connection/peerHost.test.ts | 99 ++++ .../client-runtime/src/connection/peerHost.ts | 289 +++++++++++ packages/contracts/src/environment.ts | 4 + packages/contracts/src/index.ts | 1 + packages/contracts/src/orchestratorMcp.ts | 13 + packages/contracts/src/peerEnvironment.ts | 118 +++++ packages/contracts/src/rpc.ts | 21 + packages/shared/src/t3McpToolPresentation.ts | 8 + 33 files changed, 2017 insertions(+), 41 deletions(-) create mode 100644 apps/server/src/mcp/PeerEnvironmentBroker.test.ts create mode 100644 apps/server/src/mcp/PeerEnvironmentBroker.ts create mode 100644 apps/server/src/mcp/PeerEnvironmentService.test.ts create mode 100644 apps/server/src/mcp/PeerEnvironmentService.ts create mode 100644 packages/client-runtime/src/connection/peerHost.test.ts create mode 100644 packages/client-runtime/src/connection/peerHost.ts create mode 100644 packages/contracts/src/peerEnvironment.ts diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 9a39b82b291c..ecc205827967 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -190,6 +190,8 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.previewAutomationConnect]: AuthOrchestrationOperateScope, [WS_METHODS.previewAutomationRespond]: AuthOrchestrationOperateScope, [WS_METHODS.previewAutomationFocusHost]: AuthOrchestrationOperateScope, + [WS_METHODS.peerEnvironmentsConnect]: AuthOrchestrationOperateScope, + [WS_METHODS.peerEnvironmentsRespond]: AuthOrchestrationOperateScope, [WS_METHODS.subscribePreviewEvents]: AuthOrchestrationReadScope, [WS_METHODS.subscribeDiscoveredLocalServers]: AuthOrchestrationReadScope, [WS_METHODS.deviceConfigure]: AuthOrchestrationOperateScope, diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index ef66b6d5419a..09212657262a 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -249,6 +249,7 @@ export const make = Effect.gen(function* () { serverResolvedCommandContext: true, environmentIcon: true, projectCloneTracking: true, + peerEnvironments: true, ...(serverSelfUpdate === null ? {} : { serverSelfUpdate }), ...(serverInstallation === null ? {} : { serverInstallation }), // V2 restart recovery uses the environment-owned opt-in. The old diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index 5c8f146bcadf..bcda4dab8695 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -31,6 +31,7 @@ import { AttachmentHandlersLive } from "./toolkits/attachment/handlers.ts"; import { ThreadToolkit } from "./toolkits/thread/tools.ts"; import { ThreadToolkitHandlersLive } from "./toolkits/thread/handlers.ts"; import * as ThreadMetadataMcpService from "./ThreadMetadataMcpService.ts"; +import * as PeerEnvironmentService from "./PeerEnvironmentService.ts"; import * as McpSessionRegistry from "./McpSessionRegistry.ts"; import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts"; import { OrchestratorToolkitHandlersLive } from "./toolkits/orchestrator/handlers.ts"; @@ -726,4 +727,4 @@ export const layer = Layer.mergeAll( WorktreeToolkitRegistrationLive, PullRequestsToolkitRegistrationLive, DeviceToolkitRegistrationLive, -).pipe(Layer.provideMerge(McpTransportLive)); +).pipe(Layer.provide(PeerEnvironmentService.layer), Layer.provideMerge(McpTransportLive)); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index e2a94652f2f9..ff591aaa4f2f 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -72,13 +72,13 @@ import * as ProviderRegistry from "../provider/Services/ProviderRegistry.ts"; import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; import type { McpInvocationScope } from "./McpInvocationContext.ts"; -const DEFAULT_WAIT_TIMEOUT_MS = 10 * 60 * 1_000; -const MAX_WAIT_TIMEOUT_MS = 60 * 60 * 1_000; +export const DEFAULT_WAIT_TIMEOUT_MS = 10 * 60 * 1_000; +export const MAX_WAIT_TIMEOUT_MS = 60 * 60 * 1_000; const TASK_POLL_INTERVAL_MS = 50; const DEFAULT_THREAD_LIST_LIMIT = 50; -const DEFAULT_THREAD_READ_LIMIT = 50; -const DEFAULT_THREAD_RUN_LIMIT = 10; -const DEFAULT_THREAD_ITEM_MAX_CHARS = 20_000; +export const DEFAULT_THREAD_READ_LIMIT = 50; +export const DEFAULT_THREAD_RUN_LIMIT = 10; +export const DEFAULT_THREAD_ITEM_MAX_CHARS = 20_000; interface ResolvedTarget { readonly modelSelection: ModelSelection; @@ -230,6 +230,29 @@ function providerConstraints( return constraints; } +/** One provider instance as agents see it when choosing a launch or delegation target. */ +export function providerCapability( + provider: ServerProvider, + supportsOrchestrationV2: boolean, +): OrchestratorMcpCapabilitiesResult["providers"][number] { + const constraints = providerConstraints(provider, supportsOrchestrationV2); + return { + providerInstanceId: provider.instanceId, + driverKind: provider.driver, + displayName: provider.displayName ?? null, + models: provider.models.map((model) => ({ + id: model.slug, + label: model.name ?? null, + ...(model.capabilities?.optionDescriptors === undefined + ? {} + : { options: model.capabilities.optionDescriptors }), + })), + canRunChildTask: constraints.length === 0, + canRunCrossProviderChildTask: constraints.length === 0, + constraints: [...constraints], + }; +} + /** * Checks requested option selections for duplicates and, when the model * advertises option descriptors, against those descriptors. Models without @@ -591,7 +614,7 @@ function listItemFromShell(shell: OrchestrationV2ThreadShell): OrchestratorMcpTh }; } -function threadDetail( +export function threadDetail( projection: Pick, itemCount: number, ): OrchestratorMcpThreadDetail { @@ -639,7 +662,7 @@ function threadDetail( }; } -function threadRun(run: OrchestrationV2Run): OrchestratorMcpThreadRun { +export function threadRun(run: OrchestrationV2Run): OrchestratorMcpThreadRun { return { runId: run.id, ordinal: run.ordinal, @@ -719,7 +742,7 @@ function turnItemText(item: OrchestrationV2TurnItem): string | null { } } -function timelineItem(input: { +export function timelineItem(input: { readonly row: OrchestrationV2ThreadProjection["visibleTurnItems"][number]; readonly maxChars: number; readonly textOffset?: number; @@ -1338,28 +1361,9 @@ const make = Effect.gen(function* () { inheritedModel: parent.thread.modelSelection.model, runtimeMode: parent.thread.runtimeMode, interactionMode: parent.thread.interactionMode, - providers: providers.map((provider) => { - const constraints = providerConstraints( - provider, - orchestrationCapableInstanceIds.has(provider.instanceId), - ); - return { - providerInstanceId: provider.instanceId, - driverKind: provider.driver, - displayName: provider?.displayName ?? null, - models: - provider?.models.map((model) => ({ - id: model.slug, - label: model.name ?? null, - ...(model.capabilities?.optionDescriptors === undefined - ? {} - : { options: model.capabilities.optionDescriptors }), - })) ?? [], - canRunChildTask: constraints.length === 0, - canRunCrossProviderChildTask: constraints.length === 0, - constraints: [...constraints], - }; - }), + providers: providers.map((provider) => + providerCapability(provider, orchestrationCapableInstanceIds.has(provider.instanceId)), + ), features: { appOwnedSubagents: true, asyncPolling: true, diff --git a/apps/server/src/mcp/PeerEnvironmentBroker.test.ts b/apps/server/src/mcp/PeerEnvironmentBroker.test.ts new file mode 100644 index 000000000000..30d63a71c2a1 --- /dev/null +++ b/apps/server/src/mcp/PeerEnvironmentBroker.test.ts @@ -0,0 +1,154 @@ +import { assert, it } from "@effect/vitest"; +import { + EnvironmentId, + type PeerEnvironmentOperation, + type PeerEnvironmentResponse, + type PeerEnvironmentSummary, +} from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; + +import * as PeerEnvironmentBroker from "./PeerEnvironmentBroker.ts"; + +const makeBroker = PeerEnvironmentBroker.make.pipe(Effect.provide(NodeServices.layer)); +const r2d2 = EnvironmentId.make("environment-r2d2"); +const summary = (status: PeerEnvironmentSummary["status"]): PeerEnvironmentSummary => ({ + environmentId: r2d2, + label: "r2d2", + status, +}); + +/** A connected app that answers every request with `answer`. */ +const connectApp = ( + broker: PeerEnvironmentBroker.PeerEnvironmentBroker["Service"], + clientId: string, + answer: (operation: PeerEnvironmentOperation) => PeerEnvironmentResponse["outcome"] | null, +) => + Effect.gen(function* () { + const events = yield* broker.connect({ clientId }); + const connected = yield* Deferred.make(); + yield* events.pipe( + Stream.runForEach((event) => { + if (event.type === "connected") return Deferred.succeed(connected, undefined); + const outcome = answer(event.request.operation); + return outcome === null + ? Effect.void + : broker.respond({ + clientId, + connectionId: event.connectionId, + requestId: event.request.requestId, + outcome, + }); + }), + Effect.forkScoped, + ); + yield* Deferred.await(connected); + }); + +it.effect("refuses to reach another environment when no app is connected", () => + Effect.gen(function* () { + const broker = yield* makeBroker; + const error = yield* Effect.flip(broker.invoke({ operation: "catalog", environmentId: r2d2 })); + assert.equal(error.code, "host_unavailable"); + assert.include(error.message, "connected to both"); + }), +); + +it.effect("carries a request through the connected app and returns its answer", () => + Effect.gen(function* () { + const broker = yield* makeBroker; + const seen: Array = []; + yield* connectApp(broker, "desktop", (operation) => { + seen.push(operation); + return { ok: true, result: { operation: "list", environments: [summary("connected")] } }; + }); + assert.deepEqual(yield* broker.list, [summary("connected")]); + assert.deepEqual(seen, [{ operation: "list" }]); + }).pipe(Effect.scoped), +); + +it.effect("tries another app when the newest one cannot reach the target", () => + Effect.gen(function* () { + const broker = yield* makeBroker; + yield* connectApp(broker, "desktop", () => ({ + ok: true, + result: { + operation: "catalog", + label: "r2d2", + serverVersion: "1.0.0", + providers: [], + projects: [], + }, + })); + yield* connectApp(broker, "phone", () => ({ + ok: false, + code: "environment_not_connected", + message: "not saved on this phone", + })); + const result = yield* broker.invoke({ operation: "catalog", environmentId: r2d2 }); + assert.equal(result.label, "r2d2"); + }).pipe(Effect.scoped), +); + +it.effect("does not retry elsewhere once the target itself refused the request", () => + Effect.gen(function* () { + const broker = yield* makeBroker; + let olderAppAsked = false; + yield* connectApp(broker, "desktop", () => { + olderAppAsked = true; + return null; + }); + yield* connectApp(broker, "phone", () => ({ + ok: false, + code: "environment_unauthorized", + message: "r2d2 refused the request", + })); + const error = yield* Effect.flip(broker.invoke({ operation: "catalog", environmentId: r2d2 })); + assert.equal(error.code, "environment_unauthorized"); + assert.isFalse(olderAppAsked); + }).pipe(Effect.scoped), +); + +it.effect("ignores an answer that does not come from the app that was asked", () => + Effect.gen(function* () { + const broker = yield* makeBroker; + const events = yield* broker.connect({ clientId: "desktop" }); + const fiber = yield* broker + .invoke({ operation: "catalog", environmentId: r2d2 }, 1_000) + .pipe(Effect.flip, Effect.forkScoped); + const request = yield* events.pipe( + Stream.filter((event) => event.type === "request"), + Stream.runHead, + ); + assert.isTrue(request._tag === "Some"); + if (request._tag !== "Some") return; + yield* broker.respond({ + clientId: "someone-else", + connectionId: request.value.connectionId, + requestId: request.value.request.requestId, + outcome: { ok: true, result: { operation: "list", environments: [] } }, + }); + yield* TestClock.adjust(3_000); + const error = yield* Fiber.join(fiber); + assert.equal(error.code, "host_unavailable"); + }).pipe(Effect.scoped), +); + +it.effect("reports the best status when several apps know the same environment", () => + Effect.gen(function* () { + const broker = yield* makeBroker; + yield* connectApp(broker, "desktop", () => ({ + ok: true, + result: { operation: "list", environments: [summary("connected")] }, + })); + yield* connectApp(broker, "phone", () => ({ + ok: true, + result: { operation: "list", environments: [summary("offline")] }, + })); + assert.deepEqual(yield* broker.list, [summary("connected")]); + }).pipe(Effect.scoped), +); diff --git a/apps/server/src/mcp/PeerEnvironmentBroker.ts b/apps/server/src/mcp/PeerEnvironmentBroker.ts new file mode 100644 index 000000000000..3438657cff13 --- /dev/null +++ b/apps/server/src/mcp/PeerEnvironmentBroker.ts @@ -0,0 +1,290 @@ +import { + PeerEnvironmentFailureCode, + type PeerEnvironmentHost, + type PeerEnvironmentOperation, + type PeerEnvironmentResponse, + type PeerEnvironmentResult, + type PeerEnvironmentStreamEvent, + type PeerEnvironmentSummary, +} from "@t3tools/contracts"; +import type * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; + +export class PeerEnvironmentBrokerError extends Schema.TaggedError()( + "PeerEnvironmentBrokerError", + { + /** `host_unavailable`: no connected client could carry the request. */ + code: Schema.Union([PeerEnvironmentFailureCode, Schema.Literal("host_unavailable")]), + message: Schema.String, + }, +) {} + +type OperationName = PeerEnvironmentOperation["operation"]; +type ResultFor = Extract; + +/** + * Routes a server-side request for another environment through a connected + * client. This server holds no credential for other environments: a client + * that is connected to both runs the request over its own session there, so + * the other environment enforces that client's permissions. + */ +export class PeerEnvironmentBroker extends Context.Service< + PeerEnvironmentBroker, + { + readonly connect: ( + host: PeerEnvironmentHost, + ) => Effect.Effect>; + readonly respond: (response: PeerEnvironmentResponse) => Effect.Effect; + /** Environments any connected client can reach, best status first. */ + readonly list: Effect.Effect, PeerEnvironmentBrokerError>; + readonly invoke: ( + operation: Operation, + timeoutMs?: number, + ) => Effect.Effect, PeerEnvironmentBrokerError>; + } +>()("t3/mcp/PeerEnvironmentBroker") {} + +interface HostConnection { + readonly clientId: string; + readonly connectionId: string; + readonly order: number; + readonly queue: Queue.Queue; +} + +interface PendingRequest { + readonly host: HostConnection; + readonly deferred: Deferred.Deferred; +} + +interface BrokerState { + readonly hosts: ReadonlyMap; + readonly pending: ReadonlyMap; + readonly order: number; +} + +const DEFAULT_TIMEOUT_MS = 20_000; +const LIST_TIMEOUT_MS = 5_000; + +const NO_HOST_MESSAGE = + "No T3 Code app is connected to this environment right now. Reaching another environment needs the desktop, web, or mobile app open and connected to both."; + +const STATUS_RANK: Record = { + connected: 0, + unauthorized: 1, + incompatible: 2, + offline: 3, +}; + +// Another client may reach the target where this one could not. +const canTryNextHost = (error: PeerEnvironmentBrokerError) => + error.code === "host_unavailable" || + error.code === "environment_not_connected" || + error.code === "environment_offline"; + +const ERROR_RANK: Record = { + environment_request_failed: 0, + environment_unauthorized: 1, + environment_incompatible: 2, + environment_offline: 3, + environment_not_connected: 4, + host_unavailable: 5, +}; + +export const make = Effect.gen(function* () { + const crypto = yield* Crypto.Crypto; + const state = yield* Ref.make({ hosts: new Map(), pending: new Map(), order: 0 }); + + const disconnectedError = new PeerEnvironmentBrokerError({ + code: "host_unavailable", + message: "The T3 Code app carrying this request disconnected before answering.", + }); + + /** Drops one connection generation and fails whatever it still owed. */ + const release = Effect.fn("PeerEnvironmentBroker.release")(function* (host: HostConnection) { + const orphaned = yield* Ref.modify(state, (current) => { + const hosts = new Map(current.hosts); + if (hosts.get(host.clientId)?.connectionId === host.connectionId) hosts.delete(host.clientId); + const pending = new Map(current.pending); + const orphaned: Array = []; + for (const [requestId, entry] of pending) { + if (entry.host.connectionId !== host.connectionId) continue; + pending.delete(requestId); + orphaned.push(entry); + } + return [orphaned, { ...current, hosts, pending }]; + }); + yield* Queue.end(host.queue); + yield* Effect.forEach(orphaned, (entry) => Deferred.fail(entry.deferred, disconnectedError), { + discard: true, + }); + }); + + const acquire = Effect.fn("PeerEnvironmentBroker.acquire")(function* ( + input: PeerEnvironmentHost, + ) { + const queue = yield* Queue.unbounded(); + const connectionId = yield* crypto.randomUUIDv4.pipe(Effect.orDie); + yield* Queue.offer(queue, { type: "connected", connectionId }); + const { host, replaced } = yield* Ref.modify(state, (current) => { + const order = current.order + 1; + const host: HostConnection = { clientId: input.clientId, connectionId, order, queue }; + const hosts = new Map(current.hosts); + const replaced = hosts.get(input.clientId); + hosts.set(input.clientId, host); + return [ + { host, replaced }, + { ...current, hosts, order }, + ]; + }); + if (replaced !== undefined) yield* release(replaced); + return host; + }); + + const invokeOn = Effect.fn("PeerEnvironmentBroker.invokeOn")(function* ( + host: HostConnection, + operation: PeerEnvironmentOperation, + timeoutMs: number, + ) { + const requestId = yield* crypto.randomUUIDv4.pipe(Effect.orDie); + const deferred = yield* Deferred.make(); + yield* Ref.update(state, (current) => ({ + ...current, + pending: new Map(current.pending).set(requestId, { host, deferred }), + })); + const forget = Ref.update(state, (current) => { + const pending = new Map(current.pending); + pending.delete(requestId); + return { ...current, pending }; + }); + return yield* Effect.gen(function* () { + const offered = yield* Queue.offer(host.queue, { + type: "request", + connectionId: host.connectionId, + request: { requestId, operation, timeoutMs }, + }); + if (!offered) return yield* disconnectedError; + const answer = yield* Deferred.await(deferred).pipe( + // The client enforces timeoutMs itself; the slack covers its reply. + Effect.timeoutOption(timeoutMs + 2_000), + ); + if (Option.isSome(answer)) return answer.value; + // A silent client is not retried on later requests until it reconnects. + yield* release(host); + return yield* new PeerEnvironmentBrokerError({ + code: "host_unavailable", + message: "The T3 Code app carrying this request did not answer in time.", + }); + }).pipe(Effect.ensuring(forget)); + }); + + const hostsNewestFirst = Ref.get(state).pipe( + Effect.map((current) => + Array.from(current.hosts.values()).toSorted((left, right) => right.order - left.order), + ), + ); + + const invoke = Effect.fn("PeerEnvironmentBroker.invoke")(function* ( + operation: PeerEnvironmentOperation, + timeoutMs = DEFAULT_TIMEOUT_MS, + ) { + let failure = new PeerEnvironmentBrokerError({ + code: "host_unavailable", + message: NO_HOST_MESSAGE, + }); + for (const host of yield* hostsNewestFirst) { + const attempt = yield* invokeOn(host, operation, timeoutMs).pipe( + Effect.map((result) => ({ result })), + Effect.catch((error) => Effect.succeed({ error })), + ); + if ("result" in attempt) return attempt.result; + if (ERROR_RANK[attempt.error.code] <= ERROR_RANK[failure.code]) failure = attempt.error; + if (!canTryNextHost(attempt.error)) break; + } + return yield* failure; + }); + + const list = Effect.gen(function* () { + const hosts = yield* hostsNewestFirst; + if (hosts.length === 0) { + return yield* new PeerEnvironmentBrokerError({ + code: "host_unavailable", + message: NO_HOST_MESSAGE, + }); + } + const answers = yield* Effect.forEach( + hosts, + (host) => invokeOn(host, { operation: "list" }, LIST_TIMEOUT_MS).pipe(Effect.option), + { concurrency: "unbounded" }, + ); + const merged = new Map(); + for (const answer of answers) { + if (Option.isNone(answer) || answer.value.operation !== "list") continue; + for (const environment of answer.value.environments) { + const known = merged.get(environment.environmentId); + if (known === undefined || STATUS_RANK[environment.status] < STATUS_RANK[known.status]) { + merged.set(environment.environmentId, environment); + } + } + } + return Array.from(merged.values()); + }); + + return PeerEnvironmentBroker.of({ + connect: (host) => + Effect.succeed( + Stream.unwrap( + Effect.acquireRelease(acquire(host), release).pipe( + Effect.map((connection) => Stream.fromQueue(connection.queue)), + ), + ), + ), + respond: Effect.fn("PeerEnvironmentBroker.respond")(function* (response) { + const entry = (yield* Ref.get(state)).pending.get(response.requestId); + if ( + entry === undefined || + entry.host.clientId !== response.clientId || + entry.host.connectionId !== response.connectionId + ) { + return; + } + yield* response.outcome.ok + ? Deferred.succeed(entry.deferred, response.outcome.result) + : Deferred.fail( + entry.deferred, + new PeerEnvironmentBrokerError({ + code: response.outcome.code, + message: response.outcome.message, + }), + ); + }), + list, + invoke: ( + operation: Operation, + timeoutMs?: number, + ) => + invoke(operation, timeoutMs).pipe( + Effect.flatMap((result) => + result.operation === operation.operation + ? // The check above is what narrows the union to the requested operation. + Effect.succeed(result as ResultFor) + : Effect.fail( + new PeerEnvironmentBrokerError({ + code: "environment_request_failed", + message: "The T3 Code app answered a different request.", + }), + ), + ), + ), + }); +}).pipe(Effect.withSpan("PeerEnvironmentBroker.make")); + +export const layer = Layer.effect(PeerEnvironmentBroker, make); diff --git a/apps/server/src/mcp/PeerEnvironmentService.test.ts b/apps/server/src/mcp/PeerEnvironmentService.test.ts new file mode 100644 index 000000000000..6d6344650352 --- /dev/null +++ b/apps/server/src/mcp/PeerEnvironmentService.test.ts @@ -0,0 +1,316 @@ +import { assert, it } from "@effect/vitest"; +import { + EnvironmentId, + type OrchestratorMcpFailure, + type OrchestrationV2ThreadProjection, + type OrchestrationV2ThreadShell, + type PeerEnvironmentOperation, + type PeerEnvironmentResult, + ProjectId, + ProviderInstanceId, + type ServerProvider, + ThreadId, +} from "@t3tools/contracts"; +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as TestClock from "effect/testing/TestClock"; + +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; +import type { McpInvocationScope } from "./McpInvocationContext.ts"; +import * as PeerEnvironmentBroker from "./PeerEnvironmentBroker.ts"; +import * as PeerEnvironmentService from "./PeerEnvironmentService.ts"; + +const local = EnvironmentId.make("environment-macbook"); +const r2d2 = EnvironmentId.make("environment-r2d2"); +const callerThreadId = ThreadId.make("caller-thread"); +const codex = ProviderInstanceId.make("codex"); +const codexThree = ProviderInstanceId.make("codex_three"); +const remoteProject = ProjectId.make("r2d2-project"); +const remoteThread = ThreadId.make("r2d2-thread"); + +const scope: McpInvocationScope = { + environmentId: local, + threadId: callerThreadId, + providerSessionId: "session", + providerInstanceId: codex, + issuedAt: 0, + capabilities: new Set(["orchestration" as const]), +}; + +const caller = (overrides: Partial = {}) => + ({ + id: callerThreadId, + projectId: ProjectId.make("macbook-project"), + providerInstanceId: codex, + modelSelection: { instanceId: codex, model: "gpt-5" }, + runtimeMode: "full-access", + interactionMode: "default", + activeRunId: "active-run", + archivedAt: null, + deletedAt: null, + ...overrides, + }) as OrchestrationV2ThreadShell; + +const catalog: Extract = { + operation: "catalog", + label: "r2d2", + serverVersion: "1.0.0", + projects: [{ id: remoteProject, title: "t3code", workspaceRoot: "/srv/t3code" }] as never, + providers: [ + { + instanceId: codexThree, + driver: "codex", + enabled: true, + installed: true, + status: "ready", + auth: { status: "authenticated" }, + models: [{ slug: "gpt-5", name: "GPT-5" }], + } as unknown as ServerProvider, + ], +}; + +const projection = (status: string) => + ({ + thread: { id: remoteThread, projectId: remoteProject }, + runs: [{ id: "run-1", ordinal: 1, status }], + }) as unknown as OrchestrationV2ThreadProjection; + +/** Runs `use` against a service whose broker answers from `answer` and records what it was asked. */ +const withService = ( + options: { + readonly caller?: OrchestrationV2ThreadShell; + readonly answer?: (operation: PeerEnvironmentOperation) => PeerEnvironmentResult; + }, + use: ( + service: PeerEnvironmentService.PeerEnvironmentService["Service"], + asked: ReadonlyArray, + ) => Effect.Effect, +) => { + const asked: Array = []; + const answer = options.answer ?? (() => catalog); + return Effect.gen(function* () { + return yield* use(yield* PeerEnvironmentService.PeerEnvironmentService, asked); + }).pipe( + Effect.provide( + PeerEnvironmentService.layer.pipe( + Layer.provide( + Layer.mergeAll( + NodeCrypto.layer, + Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadShell: () => Effect.succeed(options.caller ?? caller()), + }), + Layer.mock(ServerEnvironment.ServerEnvironment)({ + getDescriptor: Effect.succeed({ environmentId: local, label: "MacBook" } as never), + }), + Layer.mock(PeerEnvironmentBroker.PeerEnvironmentBroker)({ + list: Effect.fail( + new PeerEnvironmentBroker.PeerEnvironmentBrokerError({ + code: "host_unavailable", + message: "No app is connected.", + }), + ), + invoke: ((operation: PeerEnvironmentOperation) => + Effect.sync(() => { + asked.push(operation); + return answer(operation); + })) as never, + }), + ), + ), + ), + ), + ); +}; + +const launch = { + environmentId: r2d2, + projectId: remoteProject, + modelSelection: { instanceId: codexThree, model: "gpt-5" }, + title: "Remote work", + message: "Do the thing", +}; + +it.effect.each([ + ["an approval-required caller", caller({ runtimeMode: "approval-required" })], + ["a plan-mode caller", caller({ interactionMode: "plan" })], +] as const)("refuses a launch on another environment from %s", ([, denied]) => + withService({ caller: denied }, (service, asked) => + Effect.gen(function* () { + const error = yield* Effect.flip(service.launchThread(scope, launch)); + assert.equal(error.code, "capability_denied"); + assert.deepEqual(asked, []); + }), + ), +); + +it.effect("refuses a launch once the calling thread no longer owns an active run", () => + withService({ caller: caller({ activeRunId: null }) }, (service, asked) => + Effect.gen(function* () { + const error = yield* Effect.flip(service.launchThread(scope, launch)); + assert.equal(error.code, "parent_not_active"); + assert.deepEqual(asked, []); + }), + ), +); + +it.effect("refuses every peer operation for a credential without orchestration", () => + withService({}, (service, asked) => + Effect.gen(function* () { + const restricted = { ...scope, capabilities: new Set(["preview" as const]) }; + const attempts: ReadonlyArray> = [ + service.list(restricted), + service.catalog(restricted, r2d2), + service.launchThread(restricted, launch), + service.readThread(restricted, { environmentId: r2d2, threadId: remoteThread }), + service.waitForThread(restricted, { environmentId: r2d2, threadId: remoteThread }), + ]; + for (const attempt of attempts) { + assert.equal((yield* Effect.flip(attempt)).code, "capability_denied"); + } + assert.deepEqual(asked, []); + }), + ), +); + +it.effect("requires the target's own project and model instead of inheriting the caller's", () => + withService({}, (service, asked) => + Effect.gen(function* () { + const error = yield* Effect.flip( + service.launchThread(scope, { environmentId: r2d2, title: "Remote work" }), + ); + assert.equal(error.code, "invalid_request"); + assert.include(error.message, "t3_environment_catalog"); + assert.deepEqual(asked, []); + }), + ), +); + +it.effect("rejects ids that do not exist on the target before creating anything there", () => + withService({}, (service, asked) => + Effect.gen(function* () { + const wrongProject = yield* Effect.flip( + service.launchThread(scope, { ...launch, projectId: ProjectId.make("macbook-project") }), + ); + assert.equal(wrongProject.code, "invalid_request"); + const wrongProvider = yield* Effect.flip( + service.launchThread(scope, { + ...launch, + modelSelection: { instanceId: codex, model: "gpt-5" }, + }), + ); + assert.equal(wrongProvider.code, "provider_unavailable"); + assert.include(wrongProvider.message, "codex_three"); + const wrongModel = yield* Effect.flip( + service.launchThread(scope, { + ...launch, + modelSelection: { instanceId: codexThree, model: "not-a-model" }, + }), + ); + assert.equal(wrongModel.code, "model_unavailable"); + assert.isFalse(asked.some((operation) => operation.operation === "launch")); + }), + ), +); + +it.effect("launches on the target with the caller's modes and the supplied ids", () => + withService( + { + answer: (operation) => + operation.operation === "launch" + ? { + operation: "launch", + result: { + threadId: operation.input.threadId ?? remoteThread, + resumed: false, + projection: { + thread: { + id: operation.input.threadId, + projectId: operation.input.projectId, + modelSelection: operation.input.modelSelection, + }, + runs: [ + { + id: "run-1", + status: "preparing", + userMessageId: operation.input.initialMessage?.messageId, + }, + ], + } as never, + }, + } + : catalog, + }, + (service, asked) => + Effect.gen(function* () { + const result = yield* service.launchThread(scope, launch); + yield* service.launchThread(scope, launch); + // A batch of launches reads the target's catalog once. + assert.lengthOf( + asked.filter((operation) => operation.operation === "catalog"), + 1, + ); + const sent = asked.find((operation) => operation.operation === "launch"); + assert.equal(sent?.environmentId, r2d2); + assert.deepInclude(sent?.input, { + projectId: remoteProject, + modelSelection: launch.modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + workspaceStrategy: { type: "root" }, + creationSource: "mcp", + }); + assert.equal(result.projectId, remoteProject); + assert.equal(result.runId, "run-1"); + assert.equal(result.status, "preparing"); + }), + ), +); + +it.effect("lists only this environment, with the reason, when no app can carry requests", () => + withService({}, (service) => + Effect.gen(function* () { + const result = yield* service.list(scope); + assert.deepEqual(result.environments, [ + { environmentId: local, label: "MacBook", local: true, status: "connected" }, + ]); + assert.equal(result.unavailableReason, "No app is connected."); + }), + ), +); + +it.effect("waits on a remote run until it settles", () => { + let reads = 0; + return withService( + { + answer: () => ({ + operation: "thread_projection", + projection: projection(++reads < 3 ? "running" : "completed"), + }), + }, + (service) => + Effect.gen(function* () { + const fiber = yield* service + .waitForThread(scope, { environmentId: r2d2, threadId: remoteThread }) + .pipe(Effect.forkChild); + yield* TestClock.adjust(10_000); + assert.deepInclude(yield* Fiber.join(fiber), { status: "completed", timedOut: false }); + }), + ); +}); + +it.effect("reports the latest status when a remote wait times out", () => + withService( + { answer: () => ({ operation: "thread_projection", projection: projection("running") }) }, + (service) => + Effect.gen(function* () { + const fiber = yield* service + .waitForThread(scope, { environmentId: r2d2, threadId: remoteThread, timeoutMs: 5_000 }) + .pipe(Effect.forkChild); + yield* TestClock.adjust(6_000); + assert.deepInclude(yield* Fiber.join(fiber), { status: "running", timedOut: true }); + }), + ), +); diff --git a/apps/server/src/mcp/PeerEnvironmentService.ts b/apps/server/src/mcp/PeerEnvironmentService.ts new file mode 100644 index 000000000000..085f89ae23b6 --- /dev/null +++ b/apps/server/src/mcp/PeerEnvironmentService.ts @@ -0,0 +1,453 @@ +import { + EnvironmentId, + MessageId, + type ModelSelection, + OrchestratorMcpFailure, + OrchestratorMcpProviderCapability, + type OrchestratorMcpThreadReadInput, + type OrchestratorMcpThreadReadResult, + type OrchestratorMcpThreadWaitInput, + type OrchestratorMcpThreadWaitResult, + type OrchestrationV2ThreadLaunchWorkspaceStrategy, + type OrchestrationV2ThreadProjection, + type OrchestrationV2ThreadShell, + PeerEnvironmentStatus, + ProjectId, + type ProviderInteractionMode, + type RunId, + type RuntimeMode, + ThreadId, +} from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; + +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; +import * as McpInvocationContext from "./McpInvocationContext.ts"; +import { + DEFAULT_THREAD_ITEM_MAX_CHARS, + DEFAULT_THREAD_READ_LIMIT, + DEFAULT_THREAD_RUN_LIMIT, + DEFAULT_WAIT_TIMEOUT_MS, + MAX_WAIT_TIMEOUT_MS, + providerCapability, + resolveInteractionMode, + resolveRuntimeMode, + threadDetail, + threadRun, + timelineItem, +} from "./OrchestratorMcpService.ts"; +import * as PeerEnvironmentBroker from "./PeerEnvironmentBroker.ts"; +import { newCommandId, readCaller, readMutationCaller } from "./threadAccess.ts"; + +export const PeerEnvironmentListResult = Schema.Struct({ + environments: Schema.Array( + Schema.Struct({ + environmentId: EnvironmentId, + label: Schema.String, + /** True for the environment this thread runs in. */ + local: Schema.Boolean, + status: PeerEnvironmentStatus, + }), + ), + /** Set when other environments could not be looked up at all. */ + unavailableReason: Schema.NullOr(Schema.String), +}); +export type PeerEnvironmentListResult = typeof PeerEnvironmentListResult.Type; + +export const PeerEnvironmentCatalogResult = Schema.Struct({ + environmentId: EnvironmentId, + label: Schema.String, + serverVersion: Schema.String, + projects: Schema.Array( + Schema.Struct({ projectId: ProjectId, title: Schema.String, workspaceRoot: Schema.String }), + ), + providers: Schema.Array(OrchestratorMcpProviderCapability), +}); +export type PeerEnvironmentCatalogResult = typeof PeerEnvironmentCatalogResult.Type; + +export interface PeerEnvironmentLaunchInput { + readonly environmentId: EnvironmentId; + readonly projectId?: ProjectId | undefined; + readonly title: string; + readonly modelSelection?: ModelSelection | undefined; + readonly runtimeMode?: RuntimeMode | undefined; + readonly interactionMode?: ProviderInteractionMode | undefined; + readonly workspaceStrategy?: OrchestrationV2ThreadLaunchWorkspaceStrategy | undefined; + readonly message?: string | undefined; +} + +type Scope = McpInvocationContext.McpInvocationScope; + +/** + * The agent-facing operations on environments other than this one. Every call + * checks the calling thread here first; the other environment then enforces + * the permissions of the client session that carries the request. + */ +export class PeerEnvironmentService extends Context.Service< + PeerEnvironmentService, + { + readonly list: ( + scope: Scope, + ) => Effect.Effect; + readonly catalog: ( + scope: Scope, + environmentId: EnvironmentId, + ) => Effect.Effect; + readonly launchThread: ( + scope: Scope, + input: PeerEnvironmentLaunchInput, + ) => Effect.Effect< + { + readonly threadId: ThreadId; + readonly projectId: ProjectId; + readonly modelSelection: ModelSelection; + readonly runId: RunId | null; + readonly status: OrchestrationV2ThreadProjection["runs"][number]["status"] | null; + }, + OrchestratorMcpFailure + >; + readonly readThread: ( + scope: Scope, + input: OrchestratorMcpThreadReadInput & { readonly environmentId: EnvironmentId }, + ) => Effect.Effect; + readonly waitForThread: ( + scope: Scope, + input: OrchestratorMcpThreadWaitInput & { readonly environmentId: EnvironmentId }, + ) => Effect.Effect; + } +>()("t3/mcp/PeerEnvironmentService") {} + +/** Whether a tool's environment selector names somewhere other than this server. */ +export const isPeerSelector = ( + scope: Scope, + environmentId: EnvironmentId | undefined, +): environmentId is EnvironmentId => + environmentId !== undefined && environmentId !== scope.environmentId; + +const WAIT_POLL_INTERVAL_MS = 2_000; +// Reading a catalog transfers the target's whole project list through the +// carrying client, so a batch of launches validates against one read. +const LAUNCH_CATALOG_TTL_MS = 30_000; +const MESSAGE_VIEW_TYPES = new Set(["user_message", "assistant_message", "proposed_plan"]); + +function brokerFailure(error: PeerEnvironmentBroker.PeerEnvironmentBrokerError) { + switch (error.code) { + case "host_unavailable": + return new OrchestratorMcpFailure({ + code: "environment_not_connected", + message: error.message, + }); + case "environment_request_failed": + return new OrchestratorMcpFailure({ code: "orchestration_error", message: error.message }); + default: + return new OrchestratorMcpFailure({ code: error.code, message: error.message }); + } +} + +/** Remote launches hand work to another machine, so they need the same caller as local ones. */ +export function assertLaunchCaller( + caller: Pick, +) { + return caller.runtimeMode === "full-access" && caller.interactionMode === "default" + ? Effect.void + : Effect.fail( + new OrchestratorMcpFailure({ + code: "capability_denied", + message: + "Launching on another environment requires a full-access/default calling thread.", + }), + ); +} + +/** Rejects ids that belong to a different environment before anything is created there. */ +export function assertLaunchTarget( + catalog: PeerEnvironmentCatalogResult, + target: { readonly projectId: ProjectId; readonly modelSelection: ModelSelection }, +) { + if (!catalog.projects.some((project) => project.projectId === target.projectId)) { + return Effect.fail( + new OrchestratorMcpFailure({ + code: "invalid_request", + message: `Project ${target.projectId} does not exist on ${catalog.label}. Project ids are per-environment; read them with t3_environment_catalog.`, + }), + ); + } + const provider = catalog.providers.find( + (candidate) => candidate.providerInstanceId === target.modelSelection.instanceId, + ); + if (provider === undefined) { + return Effect.fail( + new OrchestratorMcpFailure({ + code: "provider_unavailable", + message: `Provider instance ${target.modelSelection.instanceId} does not exist on ${catalog.label}. It has: ${catalog.providers.map((candidate) => candidate.providerInstanceId).join(", ") || "none"}.`, + }), + ); + } + if (provider.constraints.length > 0) { + return Effect.fail( + new OrchestratorMcpFailure({ + code: "provider_unavailable", + message: `Provider instance ${provider.providerInstanceId} on ${catalog.label} cannot run: ${provider.constraints.join(" ")}`, + }), + ); + } + if ( + provider.models.length > 0 && + !provider.models.some((model) => model.id === target.modelSelection.model) + ) { + return Effect.fail( + new OrchestratorMcpFailure({ + code: "model_unavailable", + message: `Model ${target.modelSelection.model} is not offered by ${provider.providerInstanceId} on ${catalog.label}.`, + }), + ); + } + return Effect.void; +} + +/** The same page shape as a local read, taken from the window the other environment returned. */ +export function readProjection( + projection: OrchestrationV2ThreadProjection, + input: OrchestratorMcpThreadReadInput, +): OrchestratorMcpThreadReadResult { + const view = input.view ?? "messages"; + const visible = projection.visibleTurnItems.filter( + (row) => view === "activity" || MESSAGE_VIEW_TYPES.has(row.item.type), + ); + const remaining = + input.itemId === undefined + ? visible.filter((row) => row.position > (input.afterPosition ?? -1)) + : visible.filter((row) => row.sourceItemId === input.itemId); + const page = remaining.slice(0, input.limit ?? DEFAULT_THREAD_READ_LIMIT); + const messagesByThreadId = new Map([[projection.thread.id, projection.messages]]); + return { + thread: threadDetail(projection, visible.length), + recentRuns: projection.runs + .toSorted((left, right) => right.ordinal - left.ordinal) + .slice(0, input.runLimit ?? DEFAULT_THREAD_RUN_LIMIT) + .map(threadRun), + items: page.map((row) => + timelineItem({ + row, + maxChars: input.maxCharsPerItem ?? DEFAULT_THREAD_ITEM_MAX_CHARS, + messagesByThreadId, + ...(input.itemId === undefined ? {} : { textOffset: input.textOffset ?? 0 }), + }), + ), + nextPosition: page.at(-1)?.position ?? null, + hasMore: page.length < remaining.length, + }; +} + +export const make = Effect.gen(function* () { + const broker = yield* PeerEnvironmentBroker.PeerEnvironmentBroker; + const threads = yield* ThreadManagementService.ThreadManagementService; + const environment = yield* ServerEnvironment.ServerEnvironment; + const crypto = yield* Crypto.Crypto; + + const withCaller = ( + scope: Scope, + effect: Effect.Effect< + A, + E, + McpInvocationContext.McpInvocationContext | ThreadManagementService.ThreadManagementService + >, + ) => + effect.pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, scope), + Effect.provideService(ThreadManagementService.ThreadManagementService, threads), + ); + + const loadCatalog = (environmentId: EnvironmentId) => + broker.invoke({ operation: "catalog", environmentId }).pipe( + Effect.mapError(brokerFailure), + Effect.map((result): PeerEnvironmentCatalogResult => ({ + environmentId, + label: result.label, + serverVersion: result.serverVersion, + projects: result.projects.map((project) => ({ + projectId: project.id, + title: project.title, + workspaceRoot: project.workspaceRoot, + })), + // Whether an adapter is registered is only known over there; the + // provider's own status already reports whether it can run. + providers: result.providers.map((provider) => providerCapability(provider, true)), + })), + ); + + const catalogs = yield* Ref.make( + new Map< + EnvironmentId, + { readonly at: number; readonly catalog: PeerEnvironmentCatalogResult } + >(), + ); + const refreshCatalog = Effect.fn("PeerEnvironmentService.refreshCatalog")(function* ( + environmentId: EnvironmentId, + ) { + const catalog = yield* loadCatalog(environmentId); + const at = yield* Clock.currentTimeMillis; + yield* Ref.update(catalogs, (current) => new Map(current).set(environmentId, { at, catalog })); + return catalog; + }); + const recentCatalog = Effect.fn("PeerEnvironmentService.recentCatalog")(function* ( + environmentId: EnvironmentId, + ) { + const known = (yield* Ref.get(catalogs)).get(environmentId); + const now = yield* Clock.currentTimeMillis; + return known !== undefined && now - known.at < LAUNCH_CATALOG_TTL_MS + ? known.catalog + : yield* refreshCatalog(environmentId); + }); + + const loadProjection = (environmentId: EnvironmentId, threadId: ThreadId) => + broker.invoke({ operation: "thread_projection", environmentId, threadId }).pipe( + Effect.mapError(brokerFailure), + Effect.map((result) => result.projection), + ); + + return PeerEnvironmentService.of({ + list: Effect.fn("PeerEnvironmentService.list")(function* (scope) { + yield* withCaller(scope, readCaller()); + const descriptor = yield* environment.getDescriptor; + const local = { + environmentId: descriptor.environmentId, + label: descriptor.label, + local: true, + status: "connected" as const, + }; + return yield* broker.list.pipe( + Effect.map((peers) => ({ + environments: [ + local, + ...peers + .filter((peer) => peer.environmentId !== descriptor.environmentId) + .map((peer) => ({ ...peer, local: false })), + ], + unavailableReason: null, + })), + Effect.catch((error) => + Effect.succeed({ environments: [local], unavailableReason: error.message }), + ), + ); + }), + catalog: Effect.fn("PeerEnvironmentService.catalog")(function* (scope, environmentId) { + yield* withCaller(scope, readCaller()); + return yield* refreshCatalog(environmentId); + }), + launchThread: Effect.fn("PeerEnvironmentService.launchThread")(function* (scope, input) { + const { caller } = yield* withCaller(scope, readMutationCaller()); + yield* assertLaunchCaller(caller); + if (input.projectId === undefined || input.modelSelection === undefined) { + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: + "projectId and modelSelection are required with environmentId: nothing is inherited across environments. Read the target's ids with t3_environment_catalog.", + }); + } + const target = { projectId: input.projectId, modelSelection: input.modelSelection }; + const runtimeMode = yield* resolveRuntimeMode(caller.runtimeMode, input.runtimeMode); + const interactionMode = yield* resolveInteractionMode( + caller.interactionMode, + input.interactionMode, + ); + yield* assertLaunchTarget(yield* recentCatalog(input.environmentId), target); + const commandId = yield* newCommandId().pipe(Effect.provideService(Crypto.Crypto, crypto)); + const messageId = MessageId.make(commandId); + const launched = yield* broker + .invoke( + { + operation: "launch", + environmentId: input.environmentId, + input: { + commandId, + threadId: ThreadId.make(commandId), + creationSource: "mcp", + ...target, + title: input.title, + runtimeMode, + interactionMode, + workspaceStrategy: input.workspaceStrategy ?? { type: "root" }, + ...(input.message === undefined + ? {} + : { initialMessage: { messageId, text: input.message, attachments: [] } }), + }, + }, + // Covers a slow launch acknowledgement, not workspace preparation, + // which continues after the thread exists. + 60_000, + ) + .pipe(Effect.mapError(brokerFailure)); + const { thread, runs } = launched.result.projection; + const run = runs.find((candidate) => candidate.userMessageId === messageId); + return { + threadId: thread.id, + projectId: thread.projectId, + modelSelection: thread.modelSelection, + runId: run?.id ?? null, + status: run?.status ?? null, + }; + }), + readThread: Effect.fn("PeerEnvironmentService.readThread")(function* (scope, input) { + yield* withCaller(scope, readCaller()); + return readProjection(yield* loadProjection(input.environmentId, input.threadId), input); + }), + waitForThread: Effect.fn("PeerEnvironmentService.waitForThread")(function* (scope, input) { + yield* withCaller(scope, readCaller()); + const timeoutMs = Math.min( + MAX_WAIT_TIMEOUT_MS, + Math.max(1, input.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS), + ); + const first = yield* loadProjection(input.environmentId, input.threadId); + const runId = input.runId ?? ThreadManagementService.latestRun(first)?.id; + if (runId === undefined) { + return { threadId: input.threadId, runId: null, status: "idle" as const, timedOut: false }; + } + const findRun = ( + projection: OrchestrationV2ThreadProjection, + ): Effect.Effect => { + const run = projection.runs.find((candidate) => candidate.id === runId); + return run === undefined + ? Effect.fail( + new OrchestratorMcpFailure({ + code: "run_not_found", + message: `Run ${runId} was not found in thread ${input.threadId}.`, + }), + ) + : Effect.succeed(run); + }; + // The other environment cannot push to this server, so its state is + // re-read through the carrying client until the run settles. + let latest = yield* findRun(first); + const settled = ( + projection: OrchestrationV2ThreadProjection, + ): Effect.Effect => + findRun(projection).pipe( + Effect.flatMap((run) => { + latest = run; + return ThreadManagementService.isTerminalRunStatus(run.status) + ? Effect.void + : Effect.sleep(WAIT_POLL_INTERVAL_MS).pipe( + Effect.andThen(loadProjection(input.environmentId, input.threadId)), + Effect.flatMap(settled), + ); + }), + ); + const finished = yield* settled(first).pipe(Effect.timeoutOption(timeoutMs)); + return { + threadId: input.threadId, + runId, + status: latest.status, + timedOut: finished._tag === "None", + }; + }), + }); +}); + +export const layer = Layer.effect(PeerEnvironmentService, make); diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 339e5eb9b2a8..273201108c27 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -72,6 +72,8 @@ it("publishes unique tool names with reference-free object-root inputs", () => { } } expect(names.has("t3_thread_launch")).toBe(true); + expect(names.has("t3_environment_list")).toBe(true); + expect(names.has("t3_environment_catalog")).toBe(true); expect(names.has("t3_thread_start")).toBe(false); }); diff --git a/apps/server/src/mcp/toolkits/environment/handlers.ts b/apps/server/src/mcp/toolkits/environment/handlers.ts index 2c01841611dc..c4075e3e6520 100644 --- a/apps/server/src/mcp/toolkits/environment/handlers.ts +++ b/apps/server/src/mcp/toolkits/environment/handlers.ts @@ -4,6 +4,7 @@ import * as Environment from "../../../environment/ServerEnvironment.ts"; import * as ThreadCommandExecutor from "../../../orchestration-v2/ThreadCommandExecutor.ts"; import * as Settings from "../../../serverSettings.ts"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import * as PeerEnvironmentService from "../../PeerEnvironmentService.ts"; import { readCaller, readMutationCaller, unavailable } from "../../threadAccess.ts"; import { EnvironmentToolkit } from "./tools.ts"; @@ -76,4 +77,22 @@ export const EnvironmentHandlersLive = EnvironmentToolkit.toLayer({ }), ); }), + t3_environment_list: () => + Effect.gen(function* () { + const scope = yield* McpInvocationContext.McpInvocationContext; + const peers = yield* PeerEnvironmentService.PeerEnvironmentService; + return yield* peers.list(scope); + }), + t3_environment_catalog: ({ environmentId }) => + Effect.gen(function* () { + const scope = yield* McpInvocationContext.McpInvocationContext; + const peers = yield* PeerEnvironmentService.PeerEnvironmentService; + if (!PeerEnvironmentService.isPeerSelector(scope, environmentId)) + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: + "That is this environment. Use t3_project_list and orchestrator_capabilities for it.", + }); + return yield* peers.catalog(scope, environmentId); + }), }); diff --git a/apps/server/src/mcp/toolkits/environment/tools.ts b/apps/server/src/mcp/toolkits/environment/tools.ts index 76fb7ea1324d..f73a1d445eb4 100644 --- a/apps/server/src/mcp/toolkits/environment/tools.ts +++ b/apps/server/src/mcp/toolkits/environment/tools.ts @@ -1,4 +1,5 @@ import { + EnvironmentId, BackgroundActivityProfile, BackgroundActivityProfileSelection, ExecutionEnvironmentDescriptor, @@ -13,6 +14,7 @@ import * as ThreadCommandExecutor from "../../../orchestration-v2/ThreadCommandE import * as ThreadManagementService from "../../../orchestration-v2/ThreadManagementService.ts"; import * as Settings from "../../../serverSettings.ts"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import * as PeerEnvironmentService from "../../PeerEnvironmentService.ts"; const PreferenceFields = { defaultThreadEnvMode: ServerSettings.fields.defaultThreadEnvMode, @@ -40,7 +42,7 @@ const shared = { 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.", + "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. This describes only the environment the thread runs in; use t3_environment_list for the user's other connected environments.", success: Schema.Struct({ environmentId: ExecutionEnvironmentDescriptor.fields.environmentId, label: Schema.String, @@ -64,4 +66,36 @@ const EnvironmentPreferencesTool = Tool.make("t3_environment_preferences_update" }), success: Schema.Struct(PreferenceFields), }).annotate(Tool.Destructive, true); -export const EnvironmentToolkit = Toolkit.make(EnvironmentReadTool, EnvironmentPreferencesTool); +const peer = { + failure: OrchestratorMcpFailure, + failureMode: "return" as const, + dependencies: [ + McpInvocationContext.McpInvocationContext, + PeerEnvironmentService.PeerEnvironmentService, + ], +}; +const EnvironmentListTool = Tool.make("t3_environment_list", { + ...peer, + description: + "List this environment and the user's other connected environments (other machines running T3 Code). Each has an environmentId, label, and status: connected, offline, unauthorized, or incompatible (its server speaks a different protocol version). Other environments are reached through a T3 Code app that is open and connected to both; when none is, only this environment is listed and unavailableReason says why. Pass a connected environmentId to t3_environment_catalog, then to t3_thread_launch, t3_thread_read, and t3_thread_wait.", + success: PeerEnvironmentService.PeerEnvironmentListResult, +}) + .annotate(Tool.Title, "List connected environments") + .annotate(Tool.Readonly, true) + .annotate(Tool.Destructive, false); +const EnvironmentCatalogTool = Tool.make("t3_environment_catalog", { + ...peer, + description: + "Read another connected environment's projects and provider instances with their models, as the ids to use when launching a thread there. Project, provider instance, model, and thread ids are per-environment: ids from this environment (t3_project_list, orchestrator_capabilities) do not exist on another one. Get environmentId from t3_environment_list.", + parameters: Schema.Struct({ environmentId: EnvironmentId }), + success: PeerEnvironmentService.PeerEnvironmentCatalogResult, +}) + .annotate(Tool.Title, "Read another environment's projects and providers") + .annotate(Tool.Readonly, true) + .annotate(Tool.Destructive, false); +export const EnvironmentToolkit = Toolkit.make( + EnvironmentReadTool, + EnvironmentPreferencesTool, + EnvironmentListTool, + EnvironmentCatalogTool, +); diff --git a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts index 7dbbd21cb624..1ca598f53ed5 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts @@ -3,6 +3,7 @@ import * as Effect from "effect/Effect"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; import * as OrchestratorMcpService from "../../OrchestratorMcpService.ts"; +import * as PeerEnvironmentService from "../../PeerEnvironmentService.ts"; import * as ThreadMetadataMcpService from "../../ThreadMetadataMcpService.ts"; const handlers = { @@ -69,6 +70,10 @@ const handlers = { t3_thread_read: (input) => Effect.gen(function* () { const scope = yield* McpInvocationContext.McpInvocationContext; + if (PeerEnvironmentService.isPeerSelector(scope, input.environmentId)) { + const peers = yield* PeerEnvironmentService.PeerEnvironmentService; + return yield* peers.readThread(scope, { ...input, environmentId: input.environmentId }); + } const service = yield* OrchestratorMcpService.OrchestratorMcpService; return yield* service.readThread(scope, input); }), @@ -87,6 +92,10 @@ const handlers = { t3_thread_wait: (input) => Effect.gen(function* () { const scope = yield* McpInvocationContext.McpInvocationContext; + if (PeerEnvironmentService.isPeerSelector(scope, input.environmentId)) { + const peers = yield* PeerEnvironmentService.PeerEnvironmentService; + return yield* peers.waitForThread(scope, { ...input, environmentId: input.environmentId }); + } const service = yield* OrchestratorMcpService.OrchestratorMcpService; return yield* service.waitForThread(scope, input); }), diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts index 572cdcbf43e9..954754474419 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts @@ -72,6 +72,27 @@ describe("orchestrator MCP tool guidance", () => { assert.include(ScheduleTaskTool.description ?? "", "nextRunAt"); }); + it("offers the environment selector on thread read and wait as an optional field", () => { + for (const tool of [ + OrchestratorToolkit.tools.t3_thread_read, + OrchestratorToolkit.tools.t3_thread_wait, + ]) { + const schema = Tool.getJsonSchema(tool) as { + readonly properties?: Readonly>; + readonly required?: ReadonlyArray; + }; + assert.include(JSON.stringify(schema.properties?.environmentId), "t3_environment_list"); + assert.notInclude(schema.required ?? [], "environmentId"); + assert.include(tool.description ?? "", "another connected environment"); + } + // Batch creation inherits the caller's checkout, so it must not grow a selector. + const batch = Tool.getJsonSchema(CreateThreadsTool) as { + readonly properties?: Readonly>; + }; + assert.notProperty(batch.properties ?? {}, "environmentId"); + assert.include(CreateThreadsTool.description ?? "", "t3_thread_launch once per thread"); + }); + it("publishes thread metadata actions from an object-root schema", () => { const schema = Tool.getJsonSchema(ThreadUpdateTool) as { readonly type?: unknown; diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index e29b6bc144ba..885e0599367e 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -31,12 +31,14 @@ import { Tool, Toolkit } from "effect/unstable/ai"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; import * as OrchestratorMcpService from "../../OrchestratorMcpService.ts"; +import * as PeerEnvironmentService from "../../PeerEnvironmentService.ts"; import * as ThreadMetadataMcpService from "../../ThreadMetadataMcpService.ts"; const dependencies = [ McpInvocationContext.McpInvocationContext, OrchestratorMcpService.OrchestratorMcpService, ]; +const peerDependencies = [...dependencies, PeerEnvironmentService.PeerEnvironmentService]; const threadMetadataDependencies = [ McpInvocationContext.McpInvocationContext, ThreadMetadataMcpService.ThreadMetadataMcpService, @@ -44,7 +46,7 @@ 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 T3 thread. 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 T3 thread. For a separate top-level thread in a new or existing worktree, use t3_thread_launch with workspaceStrategy. These provider instances belong to this environment only; read another connected environment's with t3_environment_catalog.", success: OrchestratorMcpCapabilitiesResult, failure: OrchestratorMcpFailure, failureMode: "return", @@ -146,7 +148,7 @@ const DeleteScheduledTaskTool = Tool.make("delete_scheduled_task", { export const CreateThreadsTool = Tool.make("create_threads", { description: - "Create one or more ORDINARY TOP-LEVEL T3 conversations. This is not delegation and does not create child agents/subagents. For delegated work, choose models from orchestrator_capabilities. Prefer native subagents only when they support the chosen model; otherwise call delegate_task, including for same-provider work. Use create_threads for a batch of separate top-level threads sharing this checkout. Prefer t3_thread_launch for a single thread. Both require the user to request separate/new/top-level threads or conversations. Each entry may override provider, model, options, runtime mode, and interaction mode; omitted settings inherit. Project, branch, and worktree always inherit and cannot be overridden here. For independent implementation or a PR stack in its own worktree, use t3_thread_launch with workspaceStrategy instead of asking the agent to create a worktree in its prompt.", + "Create one or more ORDINARY TOP-LEVEL T3 conversations. This is not delegation and does not create child agents/subagents. For delegated work, choose models from orchestrator_capabilities. Prefer native subagents only when they support the chosen model; otherwise call delegate_task, including for same-provider work. Use create_threads for a batch of separate top-level threads sharing this checkout. Prefer t3_thread_launch for a single thread. Both require the user to request separate/new/top-level threads or conversations. Each entry may override provider, model, options, runtime mode, and interaction mode; omitted settings inherit. Project, branch, and worktree always inherit and cannot be overridden here, so this always creates threads in this environment; to create threads on another connected environment call t3_thread_launch once per thread with environmentId. For independent implementation or a PR stack in its own worktree, use t3_thread_launch with workspaceStrategy instead of asking the agent to create a worktree in its prompt.", parameters: OrchestratorMcpCreateThreadsInput, success: OrchestratorMcpCreateThreadsResult, failure: OrchestratorMcpFailure, @@ -159,7 +161,7 @@ export const CreateThreadsTool = Tool.make("create_threads", { const ThreadListTool = Tool.make("t3_thread_list", { description: - "List T3 threads in the calling thread's project, newest first. Filter by durable run status, title, or settled state (settled=true lists threads the user or auto-settlement moved out of the active list) and paginate with the returned cursor. Threads from other projects are never exposed.", + "List T3 threads in the calling thread's project in this environment, newest first. Filter by durable run status, title, or settled state (settled=true lists threads the user or auto-settlement moved out of the active list) and paginate with the returned cursor. Threads from other projects are never exposed.", parameters: OrchestratorMcpThreadListInput, success: OrchestratorMcpThreadListResult, failure: OrchestratorMcpFailure, @@ -173,12 +175,12 @@ const ThreadListTool = Tool.make("t3_thread_list", { const ThreadReadTool = Tool.make("t3_thread_read", { description: - "Read durable state and a paginated timeline from a T3 thread in the calling project, or from a thread the user attached to this conversation as context. The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. pendingRequestCount counts provider approval/input requests; queuedRunCount and heldQueuedRunCount report unstarted message runs independently of the latest run status. 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.", + "Read durable state and a paginated timeline from a T3 thread in the calling project, or from a thread the user attached to this conversation as context. The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. pendingRequestCount counts provider approval/input requests; queuedRunCount and heldQueuedRunCount report unstarted message runs independently of the latest run status. 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. Set environmentId to read a thread on another connected environment, such as one created there by t3_thread_launch; that read covers the thread's recent history and never acknowledges a delegated result.", parameters: OrchestratorMcpThreadReadInput, success: OrchestratorMcpThreadReadResult, failure: OrchestratorMcpFailure, failureMode: "return", - dependencies, + dependencies: peerDependencies, }) .annotate(Tool.Title, "Read a T3 thread") .annotate(Tool.Readonly, false) @@ -213,12 +215,12 @@ 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. 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. Set environmentId to wait on a thread in another connected environment.", parameters: OrchestratorMcpThreadWaitInput, success: OrchestratorMcpThreadWaitResult, failure: OrchestratorMcpFailure, failureMode: "return", - dependencies, + dependencies: peerDependencies, }) .annotate(Tool.Title, "Wait for a T3 thread") .annotate(Tool.Readonly, true) diff --git a/apps/server/src/mcp/toolkits/project/handlers.test.ts b/apps/server/src/mcp/toolkits/project/handlers.test.ts index 68dc746d2499..1216cb29ed2f 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.test.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.test.ts @@ -19,6 +19,7 @@ import * as ServerConfig from "../../../config.ts"; import * as Project from "../../../project/ProjectService.ts"; import * as ManagedProjectFolders from "../../../project/ManagedProjectFolders.ts"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import * as PeerEnvironmentService from "../../PeerEnvironmentService.ts"; import { ProjectHandlersLive } from "./handlers.ts"; import { ProjectToolkit } from "./tools.ts"; @@ -67,6 +68,7 @@ it.effect("attributes a launched thread's first message to the calling thread", }, }), Layer.mock(Project.ProjectService)({}), + Layer.mock(PeerEnvironmentService.PeerEnvironmentService)({}), Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/projects" }), NodeServices.layer, ServerConfig.layerTest(process.cwd(), { prefix: "t3-source-link-" }).pipe( @@ -129,6 +131,7 @@ it.effect("launches a scratch thread into the Scratch project", () => }, }), Layer.mock(Project.ProjectService)({}), + Layer.mock(PeerEnvironmentService.PeerEnvironmentService)({}), Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/projects", ensureScratchProject: Effect.succeed({ projectId: scratchProjectId }), @@ -215,6 +218,7 @@ it.effect("starts a project from just a title when workspaceRoot is omitted", () projectId === createdProjectId ? Option.some(createdProject) : Option.none(), ), }), + Layer.mock(PeerEnvironmentService.PeerEnvironmentService)({}), Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/projects", createNamedProject: (input) => diff --git a/apps/server/src/mcp/toolkits/project/handlers.ts b/apps/server/src/mcp/toolkits/project/handlers.ts index 346e884f3032..0bba3242d2d3 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.ts @@ -7,6 +7,8 @@ import * as Project from "../../../project/ProjectService.ts"; import * as ManagedProjectFolders from "../../../project/ManagedProjectFolders.ts"; import * as Repositories from "../../../sourceControl/SourceControlRepositoryService.ts"; import { newCommandId, readCaller, readMutationCaller, unavailable } from "../../threadAccess.ts"; +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import * as PeerEnvironmentService from "../../PeerEnvironmentService.ts"; import { ProjectToolkit } from "./tools.ts"; function projectFailure(error: Project.ProjectServiceError) { @@ -40,6 +42,20 @@ const mutation = Effect.gen(function* () { export const ProjectHandlersLive = ProjectToolkit.toLayer({ t3_thread_launch: (input) => Effect.gen(function* () { + const invocation = yield* McpInvocationContext.McpInvocationContext; + if (PeerEnvironmentService.isPeerSelector(invocation, input.environmentId)) { + if (input.scratch === true || (input.attachments ?? []).length > 0) + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: + "scratch and attachments are not supported when launching on another environment.", + }); + const peers = yield* PeerEnvironmentService.PeerEnvironmentService; + return yield* peers.launchThread(invocation, { + ...input, + environmentId: input.environmentId, + }); + } const { caller, scope } = yield* readMutationCaller(); if (caller.runtimeMode !== "full-access" || caller.interactionMode !== "default") return yield* new OrchestratorMcpFailure({ diff --git a/apps/server/src/mcp/toolkits/project/tools.ts b/apps/server/src/mcp/toolkits/project/tools.ts index 693a76854bb6..65e1e1653eb2 100644 --- a/apps/server/src/mcp/toolkits/project/tools.ts +++ b/apps/server/src/mcp/toolkits/project/tools.ts @@ -1,5 +1,6 @@ import { McpAttachmentInput } from "../attachment/input.ts"; import { + EnvironmentId, NonNegativeInt, ModelSelection, TrimmedNonEmptyString, @@ -28,6 +29,7 @@ import * as ManagedProjectFolders from "../../../project/ManagedProjectFolders.t import * as ThreadManagementService from "../../../orchestration-v2/ThreadManagementService.ts"; import * as SourceControlRepositoryService from "../../../sourceControl/SourceControlRepositoryService.ts"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import * as PeerEnvironmentService from "../../PeerEnvironmentService.ts"; const shared = { success: Project, @@ -102,8 +104,12 @@ 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. Set scratch:true instead of projectId for a thread without a project: it runs in a fresh folder of its own, outside any repository. Put the task in message. Do not ask the agent to create its own worktree via shell: that does not update the thread binding. Each call creates a new launch with no retry key; retain threadId and use t3_thread_read/t3_thread_wait to follow preparation. After errors or lost responses, inspect t3_thread_list before retrying. Attachments must be pending uploads. Requires a full-access/default caller.', + '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. Set scratch:true instead of projectId for a thread without a project: it runs in a fresh folder of its own, outside any repository. Put the task in message. Do not ask the agent to create its own worktree via shell: that does not update the thread binding. Each call creates a new launch with no retry key; retain threadId and use t3_thread_read/t3_thread_wait to follow preparation. After errors or lost responses, inspect t3_thread_list before retrying. Attachments must be pending uploads. Requires a full-access/default caller. Set environmentId to create the thread on another connected environment (another machine) instead: projectId and modelSelection are then required and must come from t3_environment_catalog for that environment, workspaceStrategy paths and branches refer to that machine, and scratch and attachments are not supported. Follow it with t3_thread_read/t3_thread_wait using the same environmentId.', parameters: Schema.Struct({ + environmentId: Schema.optional(EnvironmentId).annotateKey({ + description: + "Create the thread on this other connected environment (from t3_environment_list). Omit to launch here. Nothing is inherited across environments: supply projectId and modelSelection from t3_environment_catalog.", + }), projectId: Schema.optional(ProjectId), scratch: Schema.optional( Schema.Boolean.annotate({ @@ -142,6 +148,7 @@ const ThreadLaunchTool = Tool.make("t3_thread_launch", { ManagedProjectFolders.ManagedProjectFolders, FileSystem.FileSystem, ServerConfig.ServerConfig, + PeerEnvironmentService.PeerEnvironmentService, ], }) .annotate(Tool.Destructive, true) diff --git a/apps/server/src/mcp/toolkits/worktree/registration.test.ts b/apps/server/src/mcp/toolkits/worktree/registration.test.ts index c9d6955a634d..e0c9082d7bb0 100644 --- a/apps/server/src/mcp/toolkits/worktree/registration.test.ts +++ b/apps/server/src/mcp/toolkits/worktree/registration.test.ts @@ -24,6 +24,7 @@ import * as VcsStatusBroadcaster from "../../../vcs/VcsStatusBroadcaster.ts"; import * as McpHttpServer from "../../McpHttpServer.ts"; import * as McpSessionRegistry from "../../McpSessionRegistry.ts"; import * as PreviewAutomationBroker from "../../PreviewAutomationBroker.ts"; +import * as PeerEnvironmentBroker from "../../PeerEnvironmentBroker.ts"; const StubServicesLive = Layer.mergeAll( Layer.mock(Orchestrator.OrchestratorV2)({}), @@ -75,6 +76,7 @@ it.effect("production mcp layer lists worktree tools over http", () => }), ), Layer.provide(PreviewAutomationBroker.layer), + Layer.provide(PeerEnvironmentBroker.layer), Layer.provide(StubServicesLive), Layer.build, ); diff --git a/apps/server/src/provider/T3OrchestrationInstructions.ts b/apps/server/src/provider/T3OrchestrationInstructions.ts index a70ef35bffcd..4c90b8f47ed1 100644 --- a/apps/server/src/provider/T3OrchestrationInstructions.ts +++ b/apps/server/src/provider/T3OrchestrationInstructions.ts @@ -25,6 +25,16 @@ For stacked work, set \`baseRef\` to the intended parent branch and \`startFromO \`t3_thread_launch\` has no idempotency key. Retain its returned threadId and inspect it with \`t3_thread_read\` / \`t3_thread_wait\`; preparation can still be running after acceptance. If a launch fails or its response is lost, inspect \`t3_thread_list\` before retrying, since a thread may already exist. +### Other environments + +Everything above acts on the environment this thread runs in. The user may have other environments connected (other machines running T3 Code). Use them only when the user names another machine or environment: + +- \`t3_environment_list\` lists this environment and the others with their status. \`t3_environment_read\`, \`orchestrator_capabilities\`, \`t3_project_list\`, and \`t3_thread_list\` only ever describe this environment. +- Ids are per-environment. A project, provider instance, model, or thread id from this environment does not exist on another one. Call \`t3_environment_catalog\` with the target \`environmentId\` to get its projects and provider instances before launching there. +- \`t3_thread_launch\` with \`environmentId\` creates the thread on that environment. \`projectId\` and \`modelSelection\` are required and nothing is inherited; \`workspaceStrategy\` branches and paths refer to that machine. \`create_threads\` cannot target another environment, so launch a batch there with one \`t3_thread_launch\` call per thread. +- Follow a remote thread with \`t3_thread_read\` / \`t3_thread_wait\` and the same \`environmentId\`. Other thread tools do not reach other environments. +- Other environments are reached through a T3 Code app that is open and connected to both. \`environment_not_connected\` or \`environment_offline\` means that is not currently true: tell the user rather than retrying or working around it over SSH. \`environment_unauthorized\` means the app's access there was refused, and \`environment_incompatible\` means that server needs updating. + Tool names may include a harness-normalized MCP prefix, such as \`mcp__t3_code__delegate_task\`; the semantics are the same. Some harnesses attach optional MCP servers lazily: if an initial tool-catalog scan does not show T3 tools, do not conclude that cross-provider delegation is unavailable. Make one bounded direct attempt using the known T3 tool name on the next tool step. In Codex code mode, for example, call \`tools.mcp__t3_code__orchestrator_capabilities({})\` before reporting that the capability is absent. Keep polling/wait loops bounded, do not duplicate active work, and use stable \`clientRequestId\` values when retrying tools that accept them. ACP fallback: some ACP agents accept the injected MCP server but fail to expose its tools. When the T3 tools are absent and \`T3_ACP_MCP_NODE\` is present, call the same tools through the terminal: \`ELECTRON_RUN_AS_NODE=1 "$T3_ACP_MCP_NODE" \${T3_ACP_MCP_ENTRYPOINT:+"$T3_ACP_MCP_ENTRYPOINT"} acp-mcp-call orchestrator_capabilities '{}'\` (\`T3_ACP_MCP_ENTRYPOINT\` is unset when T3 runs as a standalone executable). Delegate with \`acp-mcp-call delegate_task '{"task":"...","target":{"providerInstanceId":"...","model":"..."},"mode":"async","clientRequestId":"..."}'\`. This is the supported T3 transport fallback, not an ordinary shell-based substitute for delegation. diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index cd64847800a2..1510ce079267 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -66,6 +66,7 @@ import * as TerminalManager from "./terminal/Manager.ts"; import * as McpHttpServer from "./mcp/McpHttpServer.ts"; import * as McpSessionRegistry from "./mcp/McpSessionRegistry.ts"; import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; +import * as PeerEnvironmentBroker from "./mcp/PeerEnvironmentBroker.ts"; import * as DeviceService from "./device/DeviceService.ts"; import { deviceHubProxyRouteLayer } from "./device/DeviceHubProxy.ts"; import * as PreviewManager from "./preview/Manager.ts"; @@ -665,6 +666,7 @@ const makeRoutesLayer = Layer.mergeAll( // and mutations observed on WebSocket invalidate patches subsequently read over HTTP. Layer.provide(PullRequestServiceLive), Layer.provide(PreviewAutomationBroker.layer), + Layer.provide(PeerEnvironmentBroker.layer), Layer.provide(ServerSelfUpdate.layer.pipe(Layer.provide(DesktopAppUpdateLayerLive))), Layer.provide(commandReadinessLayer), Layer.provide(browserApiCorsLayer), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 0082f6620bdb..8c7fb28a4af4 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -178,6 +178,7 @@ import * as ServerSettings from "./serverSettings.ts"; import * as TerminalManager from "./terminal/Manager.ts"; import { withTerminalOutputWindow } from "./terminal/OutputProtocol.ts"; import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; +import * as PeerEnvironmentBroker from "./mcp/PeerEnvironmentBroker.ts"; import * as DeviceService from "./device/DeviceService.ts"; import { remoteSshDeviceHosts } from "./device/localSshDeviceHost.ts"; import * as PreviewManager from "./preview/Manager.ts"; @@ -1181,6 +1182,7 @@ const makeWsRpcLayer = ( clientOrigin: OrchestrationClientOrigin, clientAnalyticsProps: Readonly>, previewAutomationBroker: PreviewAutomationBroker.PreviewAutomationBroker["Service"], + peerEnvironmentBroker: PeerEnvironmentBroker.PeerEnvironmentBroker["Service"], ) => ServerWsRpcGroup.toLayer( Effect.gen(function* () { @@ -3556,6 +3558,18 @@ const makeWsRpcLayer = ( previewAutomationBroker.focusHost(input), { "rpc.aggregate": "preview-automation" }, ), + [WS_METHODS.peerEnvironmentsConnect]: (input) => + observeRpcStreamEffect( + WS_METHODS.peerEnvironmentsConnect, + peerEnvironmentBroker.connect(input), + { "rpc.aggregate": "peer-environments" }, + ), + [WS_METHODS.peerEnvironmentsRespond]: (input) => + observeRpcEffect( + WS_METHODS.peerEnvironmentsRespond, + peerEnvironmentBroker.respond(input), + { "rpc.aggregate": "peer-environments" }, + ), [WS_METHODS.subscribePreviewEvents]: (_input) => observeRpcStream(WS_METHODS.subscribePreviewEvents, previewManager.events, { "rpc.aggregate": "preview", @@ -3829,6 +3843,7 @@ const makeWsRpcLayer = ( export const websocketRpcRouteLayer = Layer.unwrap( Effect.gen(function* () { const previewAutomationBroker = yield* PreviewAutomationBroker.PreviewAutomationBroker; + const peerEnvironmentBroker = yield* PeerEnvironmentBroker.PeerEnvironmentBroker; const serverSelfUpdate = yield* ServerSelfUpdate.ServerSelfUpdate; const pullRequests = yield* PullRequestService.PullRequestService; const sql = yield* SqlClient.SqlClient; @@ -3882,6 +3897,7 @@ export const websocketRpcRouteLayer = Layer.unwrap( clientOrigin, clientAnalyticsProps, previewAutomationBroker, + peerEnvironmentBroker, ).pipe( Layer.provideMerge(RpcSerialization.layerJson), Layer.provide(Layer.succeed(SqlClient.SqlClient, sql)), diff --git a/docs/internals/remote.md b/docs/internals/remote.md index 43d96f862e3b..57a1ebe19017 100644 --- a/docs/internals/remote.md +++ b/docs/internals/remote.md @@ -24,6 +24,26 @@ prove that a route works. In particular, a host's loopback address refers to a different machine when another device opens it. Endpoint selection must not silently fall back to loopback when a shareable endpoint is unavailable. +## Agents reach other environments through a client + +Environments never connect to each other and hold no credentials for one +another. When an agent asks its environment to act on a different one (launch, +read, or wait on a thread there), the +[peer broker](../../apps/server/src/mcp/PeerEnvironmentBroker.ts) hands the +request to a connected client, and the +[client host](../../packages/client-runtime/src/connection/peerHost.ts) runs it +over the session that client already has with the target. The target therefore +enforces that client's scopes, and switching an environment off in a client +removes the route. + +The consequence is that these requests work only while a client connected to +both environments is running. A stored server-to-server credential would lift +that limit, but it would outlive the user's attention and widen what a +compromised host can reach, and ordinary sessions cannot delegate one because +minting pairing credentials requires `access:write`. Results are shaped on the +requesting server from the same wire types clients read, so the target needs no +agent-specific endpoint. + ## Hosted web is a client The hosted web app stores its connection catalog in the browser and connects diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index eab33d76f5b0..ccb00625a9dc 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -329,6 +329,12 @@ its own under the environment's Scratch project. For stacked PRs, use the parent no retry key, so inspect existing threads after a failed or lost response before launching again. `create_threads` remains the batch option for a shared checkout. +With `environmentId`, the thread is created on another connected environment +instead. Nothing inherits across environments: `projectId` and `modelSelection` +are required and are checked against the target's own catalog before anything +is created there, and `scratch` and attachments are rejected. See +[other environments](#other-environments). + ### `t3_thread_list` Lists durable thread shells in the calling thread's project, newest first. @@ -394,6 +400,22 @@ Without `runId`, it selects the newest interruptible run. A terminal run is returned unchanged, and a thread with no active provider turn returns `no_active_run`. +### Other environments + +`t3_environment_list` returns this environment and the others a connected client +can reach, each with a status. `t3_environment_catalog` returns one other +environment's projects and provider instances. `t3_thread_launch`, +`t3_thread_read`, and `t3_thread_wait` accept an optional `environmentId`; every +other tool acts on the calling environment only, and ids never carry across. + +These calls are carried by a client, not by a server-to-server link; see +[remote architecture](../internals/remote.md#agents-reach-other-environments-through-a-client). +They fail with `environment_not_connected` when no client connected to both is +running, and with `environment_offline`, `environment_unauthorized`, or +`environment_incompatible` when the client's own connection to the target is in +that state. A remote read covers the recent window the target serves to +clients, and a remote wait re-reads that state until the run settles. + ## Delegated Task Lifecycle The MCP server is a command ingress into V2. It does not call provider adapters diff --git a/docs/user/remote-access.md b/docs/user/remote-access.md index f3fb96b830f9..bb4e7a7f8d06 100644 --- a/docs/user/remote-access.md +++ b/docs/user/remote-access.md @@ -80,6 +80,20 @@ on that machine. Existing threads stay where they started. If resource checks ar unavailable or all eligible machines are full, choose a machine manually to continue. Mobile keeps its manual environment selection. +### Start threads on another machine from an agent + +An agent can start threads on any other machine you have connected. Ask for it +by name, for example "start these five tasks on r2d2 with its Codex account". +The agent looks up that machine's projects and provider accounts, starts the +threads there, and can check on them afterwards. + +This works while a T3 Code app (desktop, web, or mobile) is open and connected +to both machines, because the app carries the request using its own access. +With every app closed, such as during a scheduled task overnight, the agent is +told the other machine is not connected. The thread asking must be in +Full access and Default mode, the same as for starting threads locally. To stop +agents reaching a machine, switch it off in **Settings → Connections**. + ### Tailscale HTTPS Join both devices to the same tailnet. In the desktop app, enable **Tailscale diff --git a/packages/client-runtime/src/connection/layer.ts b/packages/client-runtime/src/connection/layer.ts index 8ec233fcc097..a6f51c4987f3 100644 --- a/packages/client-runtime/src/connection/layer.ts +++ b/packages/client-runtime/src/connection/layer.ts @@ -10,6 +10,7 @@ import * as ConnectionResolver from "./resolver.ts"; import * as ConnectionDriver from "./driver.ts"; import * as EnvironmentRegistry from "./registry.ts"; import * as ConnectionOnboarding from "./onboarding.ts"; +import { runPeerEnvironmentHost } from "./peerHost.ts"; import * as PlatformConnectionSource from "../platform/source.ts"; import * as RelayEnvironmentDiscovery from "../relay/discovery.ts"; import * as RemoteEnvironmentAuthorization from "../authorization/service.ts"; @@ -84,6 +85,7 @@ export function layerWithOptions(options: RpcSession.RpcSessionOptions) { const platformSource = yield* PlatformConnectionSource.PlatformConnectionSource; yield* watchDiscoveredCompatibility().pipe(Effect.forkScoped); yield* registry.start; + yield* runPeerEnvironmentHost().pipe(Effect.forkScoped); yield* platformSource.registrations.pipe( Stream.runForEach(registry.reconcilePlatform), Effect.forkScoped, diff --git a/packages/client-runtime/src/connection/peerHost.test.ts b/packages/client-runtime/src/connection/peerHost.test.ts new file mode 100644 index 000000000000..470c685b32f5 --- /dev/null +++ b/packages/client-runtime/src/connection/peerHost.test.ts @@ -0,0 +1,99 @@ +import { EnvironmentId } from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as SubscriptionRef from "effect/SubscriptionRef"; + +import type { ConnectionCatalogEntry } from "./catalog.ts"; +import { + AVAILABLE_CONNECTION_STATE, + BearerConnectionTarget, + ConnectionBlockedError, + type SupervisorConnectionState, +} from "./model.ts"; +import { executePeerEnvironmentOperation, peerEnvironmentStatus } from "./peerHost.ts"; +import * as EnvironmentRegistry from "./registry.ts"; + +const macbook = EnvironmentId.make("environment-macbook"); +const r2d2 = EnvironmentId.make("environment-r2d2"); +const connected: SupervisorConnectionState = { ...AVAILABLE_CONNECTION_STATE, phase: "connected" }; +const blocked = (reason: ConnectionBlockedError["reason"]): SupervisorConnectionState => ({ + ...AVAILABLE_CONNECTION_STATE, + phase: "blocked", + lastFailure: new ConnectionBlockedError({ reason, detail: "blocked" }), +}); + +const entry = (environmentId: EnvironmentId, label: string): ConnectionCatalogEntry => ({ + target: new BearerConnectionTarget({ environmentId, label, connectionId: `bearer:${label}` }), + profile: { _tag: "None" } as never, + enabled: true, +}); + +/** A registry holding both machines, where r2d2 is in `state` and requests must not be sent. */ +const run = ( + state: SupervisorConnectionState, + operation: Parameters[0]["operation"], +) => + Effect.gen(function* () { + const entries = yield* SubscriptionRef.make( + new Map([ + [macbook, entry(macbook, "MacBook")], + [r2d2, entry(r2d2, "r2d2")], + ]), + ); + return yield* executePeerEnvironmentOperation({ + hostEnvironmentId: macbook, + operation, + timeoutMs: 1_000, + }).pipe( + Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, { + entries, + state: () => Effect.succeed(state), + run: () => Effect.die("a request reached an environment that is not connected"), + } as never), + ); + }); + +describe("peer environment host", () => { + it("reports how this app's connection looks to an agent", () => { + expect(peerEnvironmentStatus({}, connected)).toBe("connected"); + expect(peerEnvironmentStatus({}, AVAILABLE_CONNECTION_STATE)).toBe("offline"); + expect(peerEnvironmentStatus({}, blocked("authentication"))).toBe("unauthorized"); + expect(peerEnvironmentStatus({}, blocked("unsupported"))).toBe("incompatible"); + expect( + peerEnvironmentStatus({ unsupportedReason: "needs an update" }, AVAILABLE_CONNECTION_STATE), + ).toBe("incompatible"); + }); + + it.effect("never lists the requesting environment as its own peer", () => + Effect.gen(function* () { + expect(yield* run(connected, { operation: "list" })).toEqual({ + operation: "list", + environments: [{ environmentId: r2d2, label: "r2d2", status: "connected" }], + }); + }), + ); + + it.effect("refuses to route a request back to the environment that asked", () => + Effect.gen(function* () { + const error = yield* Effect.flip( + run(connected, { operation: "catalog", environmentId: macbook }), + ); + expect(error.code).toBe("environment_not_connected"); + }), + ); + + it.effect("names why a target cannot be reached without sending it anything", () => + Effect.gen(function* () { + const cases = [ + [AVAILABLE_CONNECTION_STATE, "environment_offline"], + [blocked("permission"), "environment_unauthorized"], + [blocked("unsupported"), "environment_incompatible"], + ] as const; + for (const [state, code] of cases) { + const error = yield* Effect.flip(run(state, { operation: "catalog", environmentId: r2d2 })); + expect(error.code).toBe(code); + expect(error.message).toContain("r2d2"); + } + }), + ); +}); diff --git a/packages/client-runtime/src/connection/peerHost.ts b/packages/client-runtime/src/connection/peerHost.ts new file mode 100644 index 000000000000..5d54c144b325 --- /dev/null +++ b/packages/client-runtime/src/connection/peerHost.ts @@ -0,0 +1,289 @@ +import { + EnvironmentAuthorizationError, + type EnvironmentId, + ORCHESTRATION_V2_WS_METHODS, + PeerEnvironmentFailureCode, + type PeerEnvironmentOperation, + type PeerEnvironmentRequest, + type PeerEnvironmentResult, + type PeerEnvironmentStatus, + WS_METHODS, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Random from "effect/Random"; +import * as Schedule from "effect/Schedule"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; + +import * as EnvironmentRpc from "../rpc/client.ts"; +import type { RpcSession } from "../rpc/session.ts"; +import type { ConnectionCatalogEntry } from "./catalog.ts"; +import type { SupervisorConnectionState } from "./model.ts"; +import * as EnvironmentRegistry from "./registry.ts"; +import * as EnvironmentSupervisor from "./supervisor.ts"; + +class PeerRequestFailure extends Schema.TaggedError()("PeerRequestFailure", { + code: PeerEnvironmentFailureCode, + message: Schema.String, +}) {} + +const isPeerRequestFailure = Schema.is(PeerRequestFailure); +const isAuthorizationError = Schema.is(EnvironmentAuthorizationError); +const isRpcUnavailableError = Schema.is(EnvironmentRpc.EnvironmentRpcUnavailableError); +const isNotRegisteredError = Schema.is(EnvironmentRegistry.EnvironmentNotRegisteredError); + +const failure = (code: PeerEnvironmentFailureCode, message: string) => + new PeerRequestFailure({ code, message }); + +/** How an agent on another environment should read this client's connection to a peer. */ +export function peerEnvironmentStatus( + entry: Pick, + state: Pick, +): PeerEnvironmentStatus { + if (state.phase === "connected") return "connected"; + if (entry.unsupportedReason !== undefined) return "incompatible"; + if (state.phase !== "blocked") return "offline"; + switch (state.lastFailure?.reason) { + case "unsupported": + return "incompatible"; + case "authentication": + case "permission": + return "unauthorized"; + default: + return "offline"; + } +} + +const STATUS_FAILURES = { + offline: ["environment_offline", "is offline or still connecting"], + unauthorized: ["environment_unauthorized", "refused this app's access"], + incompatible: [ + "environment_incompatible", + "runs an incompatible T3 Code version and must be updated", + ], +} as const; + +const describeError = (error: object) => + "message" in error && typeof error.message === "string" ? error.message : "The request failed."; + +/** + * Runs one request from `hostEnvironmentId`'s server against another + * environment, over this client's own session there. The target enforces this + * client's permissions; nothing here widens them. + */ +export const executePeerEnvironmentOperation = Effect.fn( + "clientRuntime.connection.executePeerEnvironmentOperation", +)(function* (input: { + readonly hostEnvironmentId: EnvironmentId; + readonly operation: PeerEnvironmentOperation; + readonly timeoutMs: number; +}): Effect.fn.Return< + PeerEnvironmentResult, + PeerRequestFailure, + EnvironmentRegistry.EnvironmentRegistry +> { + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + const entries = yield* SubscriptionRef.get(registry.entries); + const statusOf = (environmentId: EnvironmentId, entry: ConnectionCatalogEntry) => + registry.state(environmentId).pipe( + Effect.map((state) => peerEnvironmentStatus(entry, state)), + Effect.orElseSucceed((): PeerEnvironmentStatus => "offline"), + ); + const operation = input.operation; + if (operation.operation === "list") { + const environments = yield* Effect.forEach( + Array.from(entries).filter(([environmentId]) => environmentId !== input.hostEnvironmentId), + ([environmentId, entry]) => + statusOf(environmentId, entry).pipe( + Effect.map((status) => ({ environmentId, label: entry.target.label, status })), + ), + ); + return { operation: "list", environments }; + } + + const environmentId = operation.environmentId; + const entry = entries.get(environmentId); + if (entry === undefined || environmentId === input.hostEnvironmentId) { + return yield* failure( + "environment_not_connected", + `Environment ${environmentId} is not one of this app's other connected environments.`, + ); + } + const status = yield* statusOf(environmentId, entry); + if (status !== "connected") { + const [code, reason] = STATUS_FAILURES[status]; + return yield* failure(code, `${entry.target.label} ${reason}.`); + } + + const request = Effect.gen(function* () { + switch (operation.operation) { + case "catalog": { + const config = yield* EnvironmentRpc.request(WS_METHODS.serverGetConfig, {}); + const shell = yield* EnvironmentRpc.subscribe( + ORCHESTRATION_V2_WS_METHODS.subscribeShell, + {}, + ).pipe( + Stream.filter((item) => item.kind === "snapshot"), + Stream.runHead, + ); + return { + operation: "catalog" as const, + label: config.environment.label, + serverVersion: config.environment.serverVersion, + providers: config.providers, + projects: Option.match(shell, { + onNone: () => [], + onSome: (item) => item.snapshot.projects, + }), + }; + } + case "launch": + return { + operation: "launch" as const, + result: yield* EnvironmentRpc.request( + ORCHESTRATION_V2_WS_METHODS.launchThread, + operation.input, + ), + }; + case "thread_projection": + return { + operation: "thread_projection" as const, + projection: yield* EnvironmentRpc.request( + ORCHESTRATION_V2_WS_METHODS.getThreadProjection, + { threadId: operation.threadId }, + ), + }; + } + }); + + return yield* registry.run(environmentId, request).pipe( + Effect.timeoutOrElse({ + duration: input.timeoutMs, + orElse: () => + Effect.fail( + failure("environment_request_failed", `${entry.target.label} did not answer in time.`), + ), + }), + Effect.mapError((error) => { + if (isPeerRequestFailure(error)) return error; + if (isAuthorizationError(error)) { + return failure( + "environment_unauthorized", + `${entry.target.label} refused the request: ${describeError(error)}`, + ); + } + if ( + EnvironmentRpc.isRpcClientError(error) || + isRpcUnavailableError(error) || + isNotRegisteredError(error) + ) { + return failure("environment_offline", `${entry.target.label} is not reachable.`); + } + return failure( + "environment_request_failed", + `${entry.target.label} rejected the request: ${describeError(error)}`, + ); + }), + ); +}); + +/** + * Offers this client to one environment's server as a carrier for its agents' + * requests to the client's other environments, on each session whose server + * supports it. + */ +const servePeerRequests = (hostEnvironmentId: EnvironmentId, clientId: string) => + Stream.unwrap( + Effect.gen(function* () { + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor; + const answer = (session: RpcSession, connectionId: string, request: PeerEnvironmentRequest) => + executePeerEnvironmentOperation({ + hostEnvironmentId, + operation: request.operation, + timeoutMs: request.timeoutMs, + }).pipe( + Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, registry), + Effect.match({ + onSuccess: (result) => ({ ok: true as const, result }), + onFailure: (error) => ({ + ok: false as const, + code: error.code, + message: error.message, + }), + }), + Effect.flatMap((outcome) => + session.client[WS_METHODS.peerEnvironmentsRespond]({ + clientId, + connectionId, + requestId: request.requestId, + outcome, + }), + ), + Effect.catchCause((cause) => + Effect.logWarning("Could not answer a peer environment request.", { + hostEnvironmentId, + cause: Cause.pretty(cause), + }), + ), + ); + return SubscriptionRef.changes(supervisor.session).pipe( + Stream.switchMap( + Option.match({ + onNone: () => Stream.empty, + onSome: (session) => + Stream.unwrap( + session.initialConfig.pipe( + Effect.map((config) => + config.environment.capabilities.peerEnvironments === true + ? session.client[WS_METHODS.peerEnvironmentsConnect]({ clientId }).pipe( + // The server completes the stream to evict a silent + // client; a live one registers again. + Stream.repeat(Schedule.spaced("1 second")), + Stream.mapEffect( + (event) => + event.type === "request" + ? answer(session, event.connectionId, event.request) + : Effect.void, + { concurrency: 8, unordered: true }, + ), + ) + : Stream.empty, + ), + ), + // A failed stream waits for the next session like any other subscription. + ).pipe(Stream.catchCause(() => Stream.empty)), + }), + ), + ); + }), + ); + +/** Keeps this client registered as a peer carrier with every saved environment. */ +export const runPeerEnvironmentHost = Effect.fn("clientRuntime.connection.runPeerEnvironmentHost")( + function* () { + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + const clientId = `peer-${(yield* Random.nextInt).toString(36)}-${(yield* Random.nextInt).toString(36)}`; + yield* SubscriptionRef.changes(registry.entries).pipe( + Stream.map((entries) => Array.from(entries.keys()).sort().join("\n")), + Stream.changes, + Stream.switchMap((key) => + Stream.mergeAll( + (key === "" ? [] : (key.split("\n") as Array)).map((environmentId) => + registry.followStream( + environmentId, + servePeerRequests(environmentId, clientId).pipe( + Stream.provideService(EnvironmentRegistry.EnvironmentRegistry, registry), + ), + ), + ), + { concurrency: "unbounded" }, + ), + ), + Stream.runDrain, + ); + }, +); diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 0fdbbe6d9fd6..31dc657cd84e 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -206,6 +206,10 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ desktop servers whose app predates the remote trigger, where clients must keep telling the user to update the app on that machine. */ desktopAppUpdate: Schema.optionalKey(Schema.Boolean), + /** Server accepts `peerEnvironments.connect`, so a client connected to + several environments can carry this one's agent requests to the others. + Absent on older servers, where clients must not open that stream. */ + peerEnvironments: Schema.optionalKey(Schema.Boolean), }); export type ExecutionEnvironmentCapabilities = typeof ExecutionEnvironmentCapabilities.Type; diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 8690bb1b2390..f36734a1bca1 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -53,6 +53,7 @@ export * from "./browserProfile.ts"; export * from "./device.ts"; export * from "./preview.ts"; export * from "./previewAutomation.ts"; +export * from "./peerEnvironment.ts"; export * from "./resourceTelemetry.ts"; export * from "./usage.ts"; export * from "./scheduledTask.ts"; diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index eca3cc164fb1..31917c4e453c 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -4,6 +4,7 @@ import * as SchemaTransformation from "effect/SchemaTransformation"; import { ContextTransferId, + EnvironmentId, IsoDateTime, MessageId, NodeId, @@ -325,6 +326,10 @@ export type OrchestratorMcpThreadListResult = typeof OrchestratorMcpThreadListRe export const OrchestratorMcpThreadReadInput = Schema.Struct({ threadId: ThreadId, + environmentId: Schema.optional(EnvironmentId).annotateKey({ + description: + "Read from another connected environment (see t3_environment_list). Omit for this environment. Thread ids are per-environment.", + }), itemId: Schema.optional(TurnItemId), textOffset: Schema.optional(NonNegativeInt), view: Schema.optional(Schema.Literals(["messages", "activity"])), @@ -426,6 +431,10 @@ export type OrchestratorMcpThreadSendResult = typeof OrchestratorMcpThreadSendRe export const OrchestratorMcpThreadWaitInput = Schema.Struct({ threadId: ThreadId, + environmentId: Schema.optional(EnvironmentId).annotateKey({ + description: + "Wait on a thread in another connected environment (see t3_environment_list). Omit for this environment. Thread ids are per-environment.", + }), runId: Schema.optional(RunId), timeoutMs: Schema.optional(Schema.Number), }); @@ -585,6 +594,10 @@ export class OrchestratorMcpFailure extends Schema.TaggedError> = { ["Update", "Updating", "Updated", "environment preferences"], "environment-update", ), + t3_environment_list: tool( + ["List", "Listing", "Listed", "connected environments"], + "environment-read", + ), + t3_environment_catalog: tool( + ["Read", "Reading", "Read", "another environment's projects and providers"], + "environment-read", + ), t3_thread_launch: tool(["Launch", "Launching", "Launched", "a project thread"], "thread-create"), t3_project_list: tool(["List", "Listing", "Listed", "projects"], "project-list"), t3_project_read: tool(["Read", "Reading", "Read", "a project"], "project-read"),