From 62008598913d00e19e5ac125fd5cf89a2b4b48a0 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Tue, 20 Oct 2026 19:40:07 +0200 Subject: [PATCH] feat(hosts): delegate agents across connected hosts Resolve MCP host operations through in-process commands for self and the existing routed client transport for other destinations. Keep ordinary parent labels, app sessions and idle finish notifications. --- docs/connected-hosts.md | 16 + .../e2e/browser/subagent-origin-pane.spec.ts | 106 +++++- packages/app/e2e/support/helpers/subagents.ts | 145 ++++++++ .../src/desktop/daemon/local-credential.ts | 1 + packages/app/src/hooks/use-archive-agent.ts | 5 + packages/app/src/panels/agent-tracks.tsx | 23 +- packages/app/src/runtime/host-runtime.ts | 140 +++++++- .../app/src/screens/settings/host-page.tsx | 1 + .../src/subagents/archive-finished.test.ts | 27 +- .../app/src/subagents/archive-finished.ts | 24 +- packages/app/src/subagents/select.test.ts | 54 ++- packages/app/src/subagents/select.ts | 47 ++- .../src/subagents/track-presentation.test.ts | 1 + packages/app/src/subagents/track.tsx | 37 +- .../app/src/subagents/use-archive-finished.ts | 17 +- .../app/src/subagents/use-archive-subagent.ts | 16 +- .../app/src/subagents/use-detach-subagent.ts | 12 +- packages/app/src/types/host-connection.ts | 11 + packages/app/src/utils/agent-snapshots.ts | 5 +- .../app/src/utils/test-daemon-connection.ts | 2 + packages/client/package.json | 4 + packages/client/src/index.test.ts | 35 ++ packages/client/src/index.ts | 54 ++- packages/protocol/package.json | 4 + packages/protocol/src/addresses.ts | 5 + packages/protocol/src/agent-labels.ts | 22 ++ .../src/paseo-tool-presentation/catalog.ts | 1 + .../server/src/server/agent/agent-prompt.ts | 109 +++++- .../src/server/agent/host-operations/index.ts | 323 ++++++++++++++++++ .../src/server/agent/tools/paseo-tools.ts | 155 ++++++++- packages/server/src/server/bootstrap.ts | 5 + .../server/connected-hosts/hosts.e2e.test.ts | 86 ++++- .../server/src/server/websocket-server.ts | 5 + public-docs/sdk/reference.md | 3 + 34 files changed, 1403 insertions(+), 98 deletions(-) create mode 100644 packages/protocol/src/addresses.ts create mode 100644 packages/server/src/server/agent/host-operations/index.ts diff --git a/docs/connected-hosts.md b/docs/connected-hosts.md index 088a5218a7..a1cfa69ed6 100644 --- a/docs/connected-hosts.md +++ b/docs/connected-hosts.md @@ -29,6 +29,22 @@ principal. An ordinary direct B consumer does not become that producer by supply The optional `server_info.sourceId` reports this producer identity separately from the authenticated admission principal. Trusted server plugins retain the authority of their runtime; they are not sandboxed by this feature. +## Delegating agents + +Agent MCP discovers destination IDs with `list_hosts`. `create_workspace`, `list_workspaces`, and +`create_agent` accept `serverId`; omission selects the calling daemon. Remote creation requires a +destination workspace and explicit provider/model. Caller filesystem defaults apply only on the +calling daemon. + +The parent’s Subagents track shows qualified children across connected hosts. Opening a child uses +the ordinary conversation. The app retains one route per destination; a request through another +origin reports a conflict. It does not substitute an existing direct connection. + +Finish notifications use the existing idle wait, including a pending initial run or an already idle +child. They name the server and agent. A daemon restart or route loss ends an outstanding notification +wait; inspect or wait for the child again rather than resending its prompt. Concurrent prompts gain +no additional turn attribution or delivery guarantee. + ## Connections and recovery The shared Node connector owns the target grammar and each connection attempt, including a fresh diff --git a/packages/app/e2e/browser/subagent-origin-pane.spec.ts b/packages/app/e2e/browser/subagent-origin-pane.spec.ts index 47a0120da6..9c6cb4a3f2 100644 --- a/packages/app/e2e/browser/subagent-origin-pane.spec.ts +++ b/packages/app/e2e/browser/subagent-origin-pane.spec.ts @@ -1,10 +1,17 @@ +import { submitMessage } from "../support/helpers/composer"; import { writeFile } from "node:fs/promises"; import path from "node:path"; import { expect, test, type Page } from "../support/fixtures"; import { runWorkspaceActionFromCommandCenter } from "../support/helpers/command-center-workspace-actions"; import { openAgentRoute, seedMockAgentWorkspace } from "../support/helpers/mock-agent"; import { seedWorkspace } from "../support/helpers/seed-client"; -import { openSubagentsTrack, seedParentWithSubagent } from "../support/helpers/subagents"; +import { + delegateFleetThroughMcp, + openSubagentsTrack, + seedParentWithSubagent, +} from "../support/helpers/subagents"; + +test.use({ e2eInjectPaseoTools: true }); const SETTINGS_KEY = "@paseo:app-settings"; @@ -147,3 +154,100 @@ test("a preferred chat file link still opens beside its right-pane agent", async await workspace.cleanup(); } }); + +test("MCP delegates to self and B and opens both qualified children in ordinary conversations", async ({ + page, +}) => { + const fleet = await delegateFleetThroughMcp(); + let appSockets = 0; + let routeOpens = 0; + const activeDaemonSockets = new Set(); + const observationRequests: Record = {}; + page.on("websocket", (socket) => { + appSockets++; + if (new URL(socket.url()).pathname === "/ws") { + activeDaemonSockets.add(socket); + socket.on("close", () => activeDaemonSockets.delete(socket)); + } + socket.on("framesent", ({ payload }) => { + if (typeof payload !== "string" || !payload.startsWith("{")) return; + if (String(payload).includes('"hosts.route.open.request"')) routeOpens++; + const outer = JSON.parse(String(payload)); + const request = outer.type === "session" ? outer.message : null; + if ( + request?.type && + (request.subscribe !== undefined || + request.observe === true || + request.type.endsWith("set_subscription.request")) + ) + observationRequests[request.type] = (observationRequests[request.type] ?? 0) + 1; + }); + }); + try { + await openAgentRoute(page, { + workspaceId: fleet.workspace.workspaceId, + agentId: fleet.parent.id, + }); + await openSubagentsTrack(page); + await expect(page.getByTestId(`subagents-track-row-${fleet.local.agentId}`)).toContainText( + "MCP child on A", + ); + await expect(page.getByTestId(`subagents-track-row-${fleet.remote.agentId}`)).toContainText( + "MCP child on B", + ); + await expect(page.getByTestId(`subagents-track-row-${fleet.remote.agentId}`)).toContainText( + "Fleet B", + ); + await page.getByTestId(`subagents-track-row-${fleet.local.agentId}`).click(); + await expect(page.getByTestId(`workspace-tab-agent_${fleet.local.agentId}`)).toBeVisible(); + await submitMessage(page, "LOCAL_UI_TARGET"); + await expect(page.getByText("LOCAL_UI_TARGET", { exact: true })).toBeVisible(); + expect((await fleet.b.fetchAgent(fleet.local.agentId))?.agent.lastUserMessageAt).toBeNull(); + await page.getByTestId(`workspace-tab-agent_${fleet.parent.id}`).click(); + await openSubagentsTrack(page); + await page.getByTestId(`subagents-track-row-${fleet.remote.agentId}`).click(); + await expect(page).toHaveURL((url) => + url.pathname.includes(`/h/${fleet.remote.serverId}/workspace/`), + ); + await submitMessage(page, "REMOTE_UI_TARGET"); + await expect(page.getByText("REMOTE_UI_TARGET", { exact: true })).toBeVisible(); + expect((await fleet.a.fetchAgent(fleet.remote.agentId))?.agent.lastUserMessageAt).toBeNull(); + await fleet.expectCompletion(fleet.local); + await fleet.expectCompletion(fleet.remote); + await openAgentRoute(page, { + workspaceId: fleet.workspace.workspaceId, + agentId: fleet.parent.id, + }); + await expect( + page.getByRole("button", { name: /Agent finished.*MCP child on A/ }), + ).toBeVisible(); + await expect( + page.getByRole("button", { name: /Agent finished.*MCP child on B/ }), + ).toBeVisible(); + const owners = await page.evaluate(() => { + const runtime = Reflect.get(globalThis, "__paseoHostRuntimeStore") as { + getHosts(): Array<{ serverId: string }>; + getSnapshot( + serverId: string, + ): { connectionStatus: string; clientGeneration: number } | null; + }; + return runtime.getHosts().map((host) => { + const snapshot = runtime.getSnapshot(host.serverId); + return { + serverId: host.serverId, + connectionStatus: snapshot?.connectionStatus, + clientGeneration: snapshot?.clientGeneration, + }; + }); + }); + console.log("Fleet ordinary route counts", { + appSockets, + activeDaemonSockets: activeDaemonSockets.size, + routeOpens, + owners, + observationRequests, + }); + } finally { + await fleet.close(); + } +}); diff --git a/packages/app/e2e/support/helpers/subagents.ts b/packages/app/e2e/support/helpers/subagents.ts index f3a5ca7c71..7b86eef899 100644 --- a/packages/app/e2e/support/helpers/subagents.ts +++ b/packages/app/e2e/support/helpers/subagents.ts @@ -1,3 +1,7 @@ +import { connectDaemonClient } from "./daemon-client-loader"; +import { startIsolatedHostDaemon } from "./isolated-host-daemon"; +import { seedWorkspace } from "./seed-client"; +import { getE2EDaemonPort } from "./daemon-port"; import { PARENT_AGENT_ID_LABEL } from "@getpaseo/protocol/agent-labels"; import { expect, type Page } from "@playwright/test"; import { daemonWsRoutePattern } from "./daemon-port"; @@ -320,3 +324,144 @@ export async function detachSubagentFromTrack(page: Page, childId: string): Prom await expect(detachButton).toBeVisible({ timeout: 30_000 }); await detachButton.click(); } + +/** The primary children come from the real agent MCP endpoint on A. */ +export async function delegateFleetThroughMcp() { + const { Client } = await import("@modelcontextprotocol/sdk/client/index.js"); + const { StreamableHTTPClientTransport } = + await import("@modelcontextprotocol/sdk/client/streamableHttp.js"); + const { z } = await import("zod"); + type DaemonClient = import("@getpaseo/client/internal/daemon-client").DaemonClient; + const workspace = await seedWorkspace({ repoPrefix: "fleet-mcp-" }); + const destination = await startIsolatedHostDaemon(`fleet_${crypto.randomUUID()}`); + const a = await connectDaemonClient({ clientIdPrefix: "fleet-seed-A" }); + const b = await connectDaemonClient({ + clientIdPrefix: "fleet-observe-B", + port: destination.port, + }); + const serverId = a.getLastServerInfoMessage()!.serverId; + const mcp = new Client({ name: "fleet-parent", version: "1" }); + const close = async () => { + await mcp.close(); + await Promise.allSettled([a.close(), b.close()]); + await destination.close(); + await workspace.cleanup(); + }; + const call = async (name: string, args: Record = {}) => { + const response = await mcp.callTool({ name, arguments: args }); + expect(response, name).not.toMatchObject({ isError: true }); + return z.record(z.string(), z.unknown()).parse(response.structuredContent); + }; + try { + const parent = await a.createAgent({ + provider: "mock", + model: "e2e-fast-stream", + cwd: workspace.repoPath, + workspaceId: workspace.workspaceId, + title: "Fleet parent", + modeId: "load-test", + }); + const registration = await a.addHost({ + connection: `127.0.0.1:${destination.port}`, + label: "Fleet B", + }); + await expect + .poll( + async () => + (await a.listHosts()).hosts.find( + (row) => row.registrationId === registration.registrationId, + )?.status, + ) + .toBe("online"); + await mcp.connect( + new StreamableHTTPClientTransport( + new URL(`http://127.0.0.1:${getE2EDaemonPort()}/mcp/agents?callerAgentId=${parent.id}`), + ), + ); + expect(await call("list_hosts")).toMatchObject({ + hosts: expect.arrayContaining([ + expect.objectContaining({ serverId: destination.serverId, label: "Fleet B" }), + ]), + }); + const target = await call("create_workspace", { + serverId: destination.serverId, + isolation: "local", + path: workspace.repoPath, + title: "B workspace", + }); + expect(await call("list_workspaces", { serverId: destination.serverId })).toMatchObject({ + workspaces: expect.arrayContaining([ + expect.objectContaining({ workspaceId: target.workspaceId }), + ]), + }); + const createChild = async (targetServerId: string, workspaceId: string, title: string) => { + const result = await call("create_agent", { + serverId: targetServerId, + workspaceId, + title, + provider: "mock/e2e-fast-stream", + settings: { modeId: "load-test" }, + initialPrompt: `Finish ${title}`, + }); + return { serverId: targetServerId, agentId: z.string().parse(result.agentId), title }; + }; + const local = await createChild(serverId, workspace.workspaceId, "MCP child on A"); + const remote = await createChild( + destination.serverId, + z.string().parse(target.workspaceId), + "MCP child on B", + ); + // Existing SDK reservation creates unrelated equal-ID fixtures; primary children remain MCP-created. + await b.createAgent({ + agentId: local.agentId, + provider: "mock", + model: "e2e-fast-stream", + cwd: workspace.repoPath, + workspaceId: String(target.workspaceId), + title: "Unrelated on B", + modeId: "load-test", + }); + await a.createAgent({ + agentId: remote.agentId, + provider: "mock", + model: "e2e-fast-stream", + cwd: workspace.repoPath, + workspaceId: workspace.workspaceId, + title: "Unrelated on A", + modeId: "load-test", + }); + return { + workspace, + parent, + serverId, + local, + remote, + a, + b, + close, + async expectCompletion(child: typeof local) { + await expect + .poll(async () => { + const timeline = await a.fetchAgentTimeline(parent.id, { + direction: "tail", + limit: 100, + }); + return timeline.entries + .flatMap(({ item }) => + item.type === "tool_call" && item.agentMessage ? [item.agentMessage] : [], + ) + .filter( + (message) => + message.event === "finished" && + message.sender.id === child.agentId && + message.text.includes(`Server: ${child.serverId}`), + ).length; + }) + .toBe(1); + }, + }; + } catch (error) { + await close(); + throw error; + } +} diff --git a/packages/app/src/desktop/daemon/local-credential.ts b/packages/app/src/desktop/daemon/local-credential.ts index 005abdc21f..78093c8e63 100644 --- a/packages/app/src/desktop/daemon/local-credential.ts +++ b/packages/app/src/desktop/daemon/local-credential.ts @@ -10,6 +10,7 @@ function connectionListen(connection: HostConnection): string | null { case "directPipe": return `pipe://${connection.path}`; case "remoteSsh": + case "routed": case "relay": return null; } diff --git a/packages/app/src/hooks/use-archive-agent.ts b/packages/app/src/hooks/use-archive-agent.ts index 92d8e4d2d6..e3ecf685b2 100644 --- a/packages/app/src/hooks/use-archive-agent.ts +++ b/packages/app/src/hooks/use-archive-agent.ts @@ -367,6 +367,11 @@ function useArchiveAgentPendingQuery() { }); } +export function usePendingArchiveAgentKeys(): ReadonlySet { + const pendingQuery = useArchiveAgentPendingQuery(); + return useMemo(() => new Set(Object.keys(pendingQuery.data ?? {})), [pendingQuery.data]); +} + export function usePendingArchiveAgentIds(serverId: string): ReadonlySet { const pendingQuery = useArchiveAgentPendingQuery(); return useMemo( diff --git a/packages/app/src/panels/agent-tracks.tsx b/packages/app/src/panels/agent-tracks.tsx index e67a004b45..56798dbf6e 100644 --- a/packages/app/src/panels/agent-tracks.tsx +++ b/packages/app/src/panels/agent-tracks.tsx @@ -1,3 +1,6 @@ +import type { AgentAddress } from "@getpaseo/protocol/addresses"; +import { getHostRuntimeStore } from "@/runtime/host-runtime"; +import { useToast } from "@/contexts/toast-context"; import { memo, useCallback, type ReactElement } from "react"; import { WorkspaceDiffStatPill } from "@/composer/diff-stat-pill"; import { useWorkspaceHasDiffStat } from "@/composer/workspace-diff-stat"; @@ -70,18 +73,28 @@ export const AgentTracks = memo(function AgentTracks({ }, [openInSidePane, tabId, workspaceKey], ); + const toast = useToast(); const handleOpenSubagent = useCallback( - (subagentId: string) => { - const session = useSessionStore.getState().sessions[serverId]; + ({ serverId: targetServerId, agentId: subagentId }: AgentAddress) => { + try { + getHostRuntimeStore().requireHostRoute(targetServerId, serverId); + } catch (error) { + toast.error(error instanceof Error ? error.message : String(error)); + return; + } + const session = useSessionStore.getState().sessions[targetServerId]; const agent = session?.agents.get(subagentId) ?? session?.agentDetails.get(subagentId); // A subagent in another workspace lives in that workspace's tabs. - if (agent?.workspaceId && agent.workspaceId !== workspaceId) { - navigateToAgent({ serverId, agentId: subagentId }); + if ( + targetServerId !== serverId || + (agent?.workspaceId && agent.workspaceId !== workspaceId) + ) { + navigateToAgent({ serverId: targetServerId, agentId: subagentId }); return; } openSubagentTarget({ kind: "agent", agentId: subagentId }); }, - [openSubagentTarget, serverId, workspaceId], + [openSubagentTarget, serverId, workspaceId, toast], ); const handleOpenProviderSubagent = useCallback( (parentAgentId: string, subagentId: string) => { diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index 55298ce5a1..c807c2d2ac 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -1,3 +1,6 @@ +import { createHostChannelTransport } from "@getpaseo/client/internal/hosts"; +import type { OwnedSubscription } from "@getpaseo/client"; +import type { HostDirectory } from "@getpaseo/protocol/hosts"; import { useSyncExternalStore, useMemo } from "react"; import AsyncStorage from "@react-native-async-storage/async-storage"; import equal from "fast-deep-equal/es6"; @@ -113,6 +116,7 @@ export type ConnectionLinkImport = | { status: "cancelled" }; export type ActiveConnection = + | { type: "routed"; endpoint: string; display: string } | { type: "directTcp"; endpoint: string; display: string } | { type: "directSocket"; endpoint: string; display: "socket" } | { type: "directPipe"; endpoint: string; display: "pipe" } @@ -234,6 +238,12 @@ const ADAPTIVE_SWITCH_CONSECUTIVE_PROBES = 3; const CONFIGURED_OVERRIDE_BOOTSTRAP_RETRY_MS = 1_000; function toActiveConnection(connection: HostConnection): ActiveConnection { + if (connection.type === "routed") + return { + type: "routed", + endpoint: connection.originServerId, + display: `via ${connection.originServerId}`, + }; if (connection.type === "directSocket") { return { type: "directSocket", @@ -518,7 +528,9 @@ function probeIntervalForConnection( return PROBE_MAX_BACKOFF_MS; } -function createDefaultDeps(): HostRuntimeControllerDeps { +function createDefaultDeps( + resolveOrigin: (serverId: string) => DaemonClient | null = () => null, +): HostRuntimeControllerDeps { const browserHostAvailable = typeof getDesktopHost()?.browser?.executeAutomationCommand === "function"; const browserAutomationCapabilities = browserHostAvailable @@ -549,6 +561,15 @@ function createDefaultDeps(): HostRuntimeControllerDeps { ...(host.password ? { password: host.password } : {}), localCredential: () => readDesktopManagedLocalCredential(connection), } satisfies Omit; + if (connection.type === "routed") { + const origin = resolveOrigin(connection.originServerId); + if (!origin) throw new Error(`Origin host ${connection.originServerId} is unavailable`); + return new DaemonClient({ + ...base, + url: "ws://routed-host/ws", + transportFactory: createHostChannelTransport(origin, host.serverId, clientId, null), + }); + } if (connection.type === "directSocket" || connection.type === "directPipe") { return new DaemonClient({ ...base, @@ -597,15 +618,40 @@ function createDefaultDeps(): HostRuntimeControllerDeps { }, }); }, - connectToDaemon: ({ host, connection, timeoutMs }) => - connectToDaemon(connection, { + connectToDaemon: async ({ host, connection, timeoutMs }) => { + if (connection.type === "routed") { + const origin = resolveOrigin(connection.originServerId); + if (!origin?.isConnected) + throw new Error(`Origin host ${connection.originServerId} is unavailable`); + const client = new DaemonClient({ + url: "ws://routed-host/ws", + clientId: crypto.randomUUID(), + transportFactory: createHostChannelTransport( + origin, + host.serverId, + crypto.randomUUID(), + null, + ), + reconnect: { enabled: false }, + appVersion: resolveAppVersion() ?? undefined, + }); + try { + await client.connect(); + return { client, serverId: host.serverId, hostname: null }; + } catch (error) { + await client.close(); + throw error; + } + } + return connectToDaemon(connection, { ...(host.serverId ? { serverId: host.serverId } : {}), ...(host.password ? { password: host.password } : {}), localCredential: () => readDesktopManagedLocalCredential(connection), ...(timeoutMs !== undefined ? { timeoutMs } : {}), capabilities: appCapabilities, trace: nativePerformanceTrace, - }), + }); + }, getClientId: () => getOrCreateClientId(), mountClientHandlers: ({ client, host }) => { const unmountServerData = mountServerDataPushRouter({ @@ -1488,6 +1534,11 @@ interface AgentDirectoryRefreshInput { export class HostRuntimeStore { private controllers = new Map(); + private hostObservations = new Map< + string, + { client: DaemonClient; observation: OwnedSubscription; unsubscribe: () => void } + >(); + private fleetDirectoryDemand = new Map void>(); private serverListeners = new Map void>>(); private globalListeners = new Set<() => void>(); private hostListListeners = new Set<() => void>(); @@ -1518,7 +1569,7 @@ export class HostRuntimeStore { replicaRowStore?: ReplicaRowStore; revokePushNotifications?: typeof revokePushNotifications; }) { - this.deps = input?.deps ?? createDefaultDeps(); + this.deps = input?.deps ?? createDefaultDeps((serverId) => this.getClient(serverId)); this.storage = input?.storage ?? AsyncStorage; this.replicaCache = new ReplicaCache(input?.replicaRowStore ?? createReplicaRowStore()); this.revokePushNotifications = input?.revokePushNotifications ?? revokePushNotifications; @@ -2185,6 +2236,12 @@ export class HostRuntimeStore { connection: HostConnection; existingClient?: DaemonClient; }): Promise { + const retained = this.hosts.find((host) => host.serverId === input.serverId); + if ( + retained?.connections.some((connection) => connection.type === "routed") && + input.connection.type !== "routed" + ) + throw new Error(`Host ${input.serverId} already has a retained routed connection`); const now = new Date().toISOString(); const next = upsertHostConnectionInProfiles({ profiles: this.hosts, @@ -2258,6 +2315,11 @@ export class HostRuntimeStore { if (nextIds.has(serverId)) { continue; } + this.hostObservations.get(serverId)?.unsubscribe(); + void this.hostObservations.get(serverId)?.observation.release(); + this.hostObservations.delete(serverId); + this.fleetDirectoryDemand.get(serverId)?.(); + this.fleetDirectoryDemand.delete(serverId); this.controllers.delete(serverId); this.lastConnectionStatusByServer.delete(serverId); this.connectionStatusStartedAtByServer.delete(serverId); @@ -2314,6 +2376,7 @@ export class HostRuntimeStore { const snapshot = controller.getSnapshot(); this.syncSessionReplica(snapshot.serverId, snapshot); this.syncDirectoryConnection(snapshot.serverId); + this.syncHostDiscovery(snapshot.serverId); this.emit(snapshot.serverId); }); void controller @@ -2332,6 +2395,73 @@ export class HostRuntimeStore { } } + /** A retained destination uses its selected route; another route is an explicit conflict. */ + requireHostRoute(serverId: string, originServerId: string): void { + if (serverId === originServerId) return; + const connection = this.hosts.find((host) => host.serverId === serverId)?.connections[0]; + if (connection?.type !== "routed" || connection.originServerId !== originServerId) { + throw new Error( + `Host ${serverId} is already connected through another route. Requested via ${originServerId}.`, + ); + } + } + + private syncHostDiscovery(serverId: string): void { + const client = this.getClient(serverId); + const previous = this.hostObservations.get(serverId); + if (previous?.client === client) return; + previous?.unsubscribe(); + void previous?.observation.release(); + this.hostObservations.delete(serverId); + const host = this.hosts.find((candidate) => candidate.serverId === serverId); + if ( + !client?.isConnected || + !client.getLastServerInfoMessage()?.features?.connectedHosts || + host?.connections.some((connection) => connection.type === "routed") + ) + return; + const observation = client.observeHosts(); + const apply = (directory: HostDirectory) => { + if (this.hostObservations.get(serverId)?.client !== client) return; + for (const target of directory.hosts) { + if (!target.serverId || target.serverId === serverId) continue; + if (!this.hosts.some((candidate) => candidate.serverId === target.serverId)) { + const now = new Date().toISOString(); + const connection = { + id: `via:${serverId}`, + type: "routed" as const, + originServerId: serverId, + }; + const profile: HostProfile = { + serverId: target.serverId, + label: target.label ?? target.serverId, + appearance: defaultHostAppearance(), + lifecycle: {}, + connections: [connection], + preferredConnectionId: connection.id, + createdAt: now, + updatedAt: now, + }; + this.setHostsAndSync([...this.hosts, profile]); + void this.persistHosts(); + } + if (!this.fleetDirectoryDemand.has(target.serverId)) + this.fleetDirectoryDemand.set( + target.serverId, + this.acquireDirectoryDemand(target.serverId), + ); + } + }; + const unsubscribe = observation.subscribe({ + snapshot: apply, + update: (message) => { + if (message.type === "hosts.changed") apply(message.payload.directory); + }, + }); + this.hostObservations.set(serverId, { client, observation, unsubscribe }); + void observation.ready.then(apply).catch(() => undefined); + } + private syncSessionReplica(serverId: string, snapshot: HostRuntimeSnapshot): void { if (!snapshot.client) { return; diff --git a/packages/app/src/screens/settings/host-page.tsx b/packages/app/src/screens/settings/host-page.tsx index 6b308547a1..c70c4941cb 100644 --- a/packages/app/src/screens/settings/host-page.tsx +++ b/packages/app/src/screens/settings/host-page.tsx @@ -106,6 +106,7 @@ const removeProfileIcon = ; function formatHostConnectionLabel(connection: HostConnection, t: TFunction): string { + if (connection.type === "routed") return `Via ${connection.originServerId}`; if (connection.type === "relay") { return `${t("settings.host.badges.relay")} (${connection.relayEndpoint})`; } diff --git a/packages/app/src/subagents/archive-finished.test.ts b/packages/app/src/subagents/archive-finished.test.ts index 5c1615d96c..8ad93f990f 100644 --- a/packages/app/src/subagents/archive-finished.test.ts +++ b/packages/app/src/subagents/archive-finished.test.ts @@ -5,6 +5,7 @@ import type { PaseoSubagentRow, ProviderSubagentRow } from "./select"; function paseo(id: string, status: PaseoSubagentRow["status"] = "idle"): PaseoSubagentRow { return { kind: "paseo", + address: { serverId: "server-1", agentId: id }, id, provider: "codex", title: id, @@ -76,8 +77,8 @@ describe("createArchiveFinishedSubagents", () => { [provider("native"), paseo("first"), paseo("second", "error"), paseo("resumed")], { parentAgentId: "parent", - getManagedSubagent: (id) => current.get(id), - archiveManagedSubagent: (id) => { + getManagedSubagent: ({ agentId: id }) => current.get(id), + archiveManagedSubagent: ({ agentId: id }) => { events.push(`archive:${id}`); return id === "first" ? first.promise : second.promise; }, @@ -130,8 +131,8 @@ describe("createArchiveFinishedSubagents", () => { const calls: string[] = []; const archive = createArchiveFinishedSubagents([paseo("first"), paseo("second")], { parentAgentId: "parent", - getManagedSubagent: (id) => current.get(id), - archiveManagedSubagent: (id) => { + getManagedSubagent: ({ agentId: id }) => current.get(id), + archiveManagedSubagent: ({ agentId: id }) => { calls.push(id); const response = responses.shift(); if (!response) throw new Error("Unexpected archive"); @@ -204,8 +205,8 @@ describe("createArchiveFinishedSubagents", () => { ]); const archive = createArchiveFinishedSubagents([paseo("first"), paseo("second")], { parentAgentId: "parent", - getManagedSubagent: (id) => current.get(id), - archiveManagedSubagent: async (id) => { + getManagedSubagent: ({ agentId: id }) => current.get(id), + archiveManagedSubagent: async ({ agentId: id }) => { throw id === "first" ? firstError : secondError; }, dismissProviderSubagents: () => undefined, @@ -230,7 +231,7 @@ describe("createArchiveFinishedSubagents", () => { const current = new Map([["failed", managed("failed")]]); const archive = createArchiveFinishedSubagents([paseo("failed")], { parentAgentId: "parent", - getManagedSubagent: (id) => current.get(id), + getManagedSubagent: ({ agentId: id }) => current.get(id), archiveManagedSubagent: async () => { current.set("failed", managed("failed", "running")); throw new Error("archive failed"); @@ -256,7 +257,7 @@ describe("createArchiveFinishedSubagents", () => { const current = new Map([["failed", managed("failed")]]); const archive = createArchiveFinishedSubagents([paseo("failed")], { parentAgentId: "parent", - getManagedSubagent: (id) => current.get(id), + getManagedSubagent: ({ agentId: id }) => current.get(id), archiveManagedSubagent: async () => { current.set("failed", managed("failed", "initializing")); throw new Error("archive failed"); @@ -297,7 +298,7 @@ describe("createArchiveFinishedSubagents", () => { const current = new Map([["failed", managed("failed")]]); const archive = createArchiveFinishedSubagents([paseo("failed")], { parentAgentId: "parent", - getManagedSubagent: (id) => current.get(id), + getManagedSubagent: ({ agentId: id }) => current.get(id), archiveManagedSubagent: async () => { updateSnapshot(current); throw error; @@ -325,8 +326,8 @@ describe("createArchiveFinishedSubagents", () => { const calls: string[] = []; const archive = createArchiveFinishedSubagents([paseo("first"), paseo("second")], { parentAgentId: "parent", - getManagedSubagent: (id) => current.get(id), - archiveManagedSubagent: async (id) => { + getManagedSubagent: ({ agentId: id }) => current.get(id), + archiveManagedSubagent: async ({ agentId: id }) => { calls.push(id); if (calls.length === 1) throw firstError; if (calls.length === 2) { @@ -384,8 +385,8 @@ describe("createArchiveFinishedSubagents", () => { const archived: string[] = []; const archive = createArchiveFinishedSubagents(rows, { parentAgentId: "parent", - getManagedSubagent: (id) => current.get(id), - archiveManagedSubagent: async (id) => { + getManagedSubagent: ({ agentId: id }) => current.get(id), + archiveManagedSubagent: async ({ agentId: id }) => { archived.push(id); current.set("resumed", managed("resumed", "running")); }, diff --git a/packages/app/src/subagents/archive-finished.ts b/packages/app/src/subagents/archive-finished.ts index 0f3c685591..c79c3af6f9 100644 --- a/packages/app/src/subagents/archive-finished.ts +++ b/packages/app/src/subagents/archive-finished.ts @@ -1,3 +1,4 @@ +import type { AgentAddress } from "@getpaseo/protocol/addresses"; import type { Agent } from "@/stores/session-store"; import type { SubagentRow } from "./select"; @@ -22,8 +23,8 @@ export type ManagedSubagentSnapshot = Pick ManagedSubagentSnapshot | undefined; - archiveManagedSubagent: (id: string) => Promise; + getManagedSubagent: (address: AgentAddress) => ManagedSubagentSnapshot | undefined; + archiveManagedSubagent: (address: AgentAddress) => Promise; dismissProviderSubagents: (ids: string[]) => void; } @@ -55,12 +56,12 @@ function eligibleSignature(rows: readonly SubagentRow[]): string { return identitySignature(rowIdentities(rows)); } -function managedRowIdentity(id: string): string { - return `paseo:${id}`; +function managedRowIdentity(address: AgentAddress): string { + return `paseo:${address.serverId}:${address.agentId}`; } function rowIdentity(row: SubagentRow): string { - return row.kind === "paseo" ? managedRowIdentity(row.id) : `provider:${row.id}`; + return row.kind === "paseo" ? managedRowIdentity(row.address) : `provider:${row.id}`; } function rowIdentities(rows: readonly SubagentRow[]): Set { @@ -169,7 +170,7 @@ async function runArchiveFinished( reportProgress: (completedCount: number) => void, ): Promise<{ outcome: ArchiveFinishedOutcome; retryableFailureIds: Set }> { const providerIds = rows.filter((row) => row.kind === "provider").map((row) => row.id); - const paseoIds = rows.filter((row) => row.kind === "paseo").map((row) => row.id); + const paseoAddresses = rows.filter((row) => row.kind === "paseo").map((row) => row.address); let completedCount = 0; const outcome = emptyOutcome(); const retryableFailureIds = new Set(); @@ -181,15 +182,16 @@ async function runArchiveFinished( reportProgress(completedCount); } - for (const id of paseoIds) { - if (canArchiveManagedSubagent(deps.getManagedSubagent(id), deps.parentAgentId)) { + for (const address of paseoAddresses) { + const id = address.agentId; + if (canArchiveManagedSubagent(deps.getManagedSubagent(address), deps.parentAgentId)) { try { - await deps.archiveManagedSubagent(id); + await deps.archiveManagedSubagent(address); outcome.archivedPaseoIds.push(id); } catch (error) { outcome.failures.push({ id, error }); - if (canArchiveManagedSubagent(deps.getManagedSubagent(id), deps.parentAgentId)) { - retryableFailureIds.add(managedRowIdentity(id)); + if (canArchiveManagedSubagent(deps.getManagedSubagent(address), deps.parentAgentId)) { + retryableFailureIds.add(managedRowIdentity(address)); } } } else { diff --git a/packages/app/src/subagents/select.test.ts b/packages/app/src/subagents/select.test.ts index 38a905183d..c9d7ca4096 100644 --- a/packages/app/src/subagents/select.test.ts +++ b/packages/app/src/subagents/select.test.ts @@ -53,7 +53,14 @@ const AGENT_DEFAULTS: Agent = { }; function makeAgent(input: Partial & Pick): Agent { - return { ...AGENT_DEFAULTS, ...input }; + return { + ...AGENT_DEFAULTS, + ...input, + labels: { + ...(input.parentAgentId ? { "paseo.parent-agent-id": input.parentAgentId } : {}), + ...input.labels, + }, + }; } function setAgents(agents: Agent[]): void { @@ -73,6 +80,41 @@ afterEach(() => { }); describe("selectSubagentsForParent", () => { + it("keeps equal child IDs on different hosts qualified to the complete parent address", () => { + const parentId = "shared-parent"; + const childId = "shared-child"; + const local = makeAgent({ + id: childId, + parentAgentId: parentId, + labels: { "paseo.parent-agent-id": parentId, "paseo.parent-server-id": SERVER_ID }, + }); + const remote = makeAgent({ + id: childId, + serverId: "server-2", + parentAgentId: null, + labels: { "paseo.parent-agent-id": parentId, "paseo.parent-server-id": SERVER_ID }, + }); + setAgents([local]); + useSessionStore.getState().initializeSession("server-2", null as unknown as DaemonClient); + useSessionStore.getState().setAgents("server-2", new Map([[childId, remote]])); + try { + const rows = selectSubagentsForParent( + useSessionStore.getState(), + { serverId: SERVER_ID, parentAgentId: parentId }, + EMPTY_PENDING_ARCHIVE_IDS, + ); + expect(rows).toHaveLength(2); + expect(rows).toEqual( + expect.arrayContaining([ + expect.objectContaining({ address: { serverId: SERVER_ID, agentId: childId } }), + expect.objectContaining({ address: { serverId: "server-2", agentId: childId } }), + ]), + ); + } finally { + useSessionStore.getState().clearSession("server-2"); + } + }); + it("hides cached provider children when the host does not support them", () => { useProviderSubagentStore.getState().applyUpdate(SERVER_ID, { kind: "upsert", @@ -315,6 +357,7 @@ describe("selectSubagentsForParent", () => { expect(rows).toEqual([ { kind: "paseo", + address: { serverId: SERVER_ID, agentId: "child" }, id: "child", provider: "claude", title: "Review child", @@ -327,6 +370,7 @@ describe("selectSubagentsForParent", () => { }, ]); expect(Object.keys(rows[0] ?? {}).sort()).toEqual([ + "address", "createdAt", "description", "id", @@ -371,7 +415,7 @@ describe("selectSubagentsForParent", () => { setAgents([ makeAgent({ id: "parent-a" }), makeAgent({ id: "parent-b" }), - { ...child, parentAgentId: "parent-b" }, + { ...child, parentAgentId: "parent-b", labels: { "paseo.parent-agent-id": "parent-b" } }, ]); expect( @@ -409,7 +453,7 @@ describe("selectSubagentsForParent", () => { serverId: SERVER_ID, parentAgentId: "parent", }, - new Set(["child-b"]), + new Set([`${SERVER_ID}:child-b`]), ); expect(rows.map((row) => row.id)).toEqual(["child-a"]); @@ -424,7 +468,7 @@ describe("selectSubagentsForParent", () => { serverId: SERVER_ID, parentAgentId: "parent", }, - new Set(["child"]), + new Set([`${SERVER_ID}:child`]), ); expect(rows).toEqual([]); @@ -541,7 +585,7 @@ describe("createSubagentsForParentSelector", () => { const rows = createSubagentsForParentSelector( params, - new Set(["child-b"]), + new Set([`${SERVER_ID}:child-b`]), )(useSessionStore.getState()); expect(rows.map((row) => row.id)).toEqual(["child-a"]); diff --git a/packages/app/src/subagents/select.ts b/packages/app/src/subagents/select.ts index 164a4d1280..29f76e34d4 100644 --- a/packages/app/src/subagents/select.ts +++ b/packages/app/src/subagents/select.ts @@ -1,5 +1,7 @@ +import type { AgentAddress } from "@getpaseo/protocol/addresses"; +import { getParentAddressFromLabels } from "@getpaseo/protocol/agent-labels"; import { useEffect, useMemo } from "react"; -import { usePendingArchiveAgentIds } from "@/hooks/use-archive-agent"; +import { usePendingArchiveAgentKeys } from "@/hooks/use-archive-agent"; import equal from "fast-deep-equal"; import { useStoreWithEqualityFn } from "zustand/traditional"; import { useSessionStore, type Agent } from "@/stores/session-store"; @@ -8,6 +10,7 @@ import type { ProviderSubagentDescriptorPayload } from "@getpaseo/protocol/messa export interface PaseoSubagentRow { kind: "paseo"; + address: AgentAddress; id: Agent["id"]; provider: Agent["provider"]; title: Agent["title"]; @@ -55,6 +58,7 @@ const EMPTY_PROVIDER_SUBAGENT_ROWS: ProviderSubagentRow[] = []; function toSubagentRow(agent: Agent): SubagentRow { return { kind: "paseo", + address: { serverId: agent.serverId, agentId: agent.id }, id: agent.id, provider: agent.provider, title: agent.title, @@ -72,21 +76,19 @@ export function selectSubagentsForParent( params: SelectSubagentsParams, pendingArchiveIds: ReadonlySet, ): SubagentRow[] { - const agents = state.sessions[params.serverId]?.agents; - if (!agents || agents.size === 0) { - return EMPTY_SUBAGENT_ROWS; - } - const rows: SubagentRow[] = []; - for (const agent of agents.values()) { - if ( - agent.archivedAt || - pendingArchiveIds.has(agent.id) || - agent.parentAgentId !== params.parentAgentId - ) { - continue; + for (const [serverId, session] of Object.entries(state.sessions)) { + for (const agent of session.agents.values()) { + const parent = getParentAddressFromLabels(agent.labels, serverId); + if ( + agent.archivedAt || + pendingArchiveIds.has(`${serverId}:${agent.id}`) || + parent?.serverId !== params.serverId || + parent.agentId !== params.parentAgentId + ) + continue; + rows.push(toSubagentRow(agent)); } - rows.push(toSubagentRow(agent)); } if (rows.length === 0) { @@ -104,12 +106,21 @@ export function createSubagentsForParentSelector( params: SelectSubagentsParams, pendingArchiveIds: ReadonlySet, ): (state: SessionStoreSnapshot) => SubagentRow[] { - let previousAgents: Map | undefined; + let previousAgents: Array<[string, Map]> = []; let previousRows: SubagentRow[] | null = null; return (state) => { - const agents = state.sessions[params.serverId]?.agents; - if (previousRows && agents === previousAgents) { + const agents = Object.entries(state.sessions).map( + ([serverId, session]) => [serverId, session.agents] as [string, Map], + ); + if ( + previousRows && + agents.length === previousAgents.length && + agents.every( + ([serverId, map], index) => + previousAgents[index]?.[0] === serverId && previousAgents[index]?.[1] === map, + ) + ) { return previousRows; } @@ -159,7 +170,7 @@ export function selectProviderSubagentsForParent( export function useSubagentsForParent(params: SelectSubagentsParams): SubagentRow[] { const { serverId, parentAgentId } = params; - const pendingArchiveIds = usePendingArchiveAgentIds(serverId); + const pendingArchiveIds = usePendingArchiveAgentKeys(); const selectPaseoRows = useMemo( () => createSubagentsForParentSelector({ serverId, parentAgentId }, pendingArchiveIds), [serverId, parentAgentId, pendingArchiveIds], diff --git a/packages/app/src/subagents/track-presentation.test.ts b/packages/app/src/subagents/track-presentation.test.ts index df009583be..0397795541 100644 --- a/packages/app/src/subagents/track-presentation.test.ts +++ b/packages/app/src/subagents/track-presentation.test.ts @@ -13,6 +13,7 @@ function row( ): PaseoSubagentRow { return { kind: "paseo", + address: overrides.address ?? { serverId: "server-1", agentId: overrides.id }, id: overrides.id, provider: overrides.provider ?? "codex", title: overrides.title ?? `Agent ${overrides.id}`, diff --git a/packages/app/src/subagents/track.tsx b/packages/app/src/subagents/track.tsx index 43f659c534..d759da431e 100644 --- a/packages/app/src/subagents/track.tsx +++ b/packages/app/src/subagents/track.tsx @@ -1,3 +1,5 @@ +import type { AgentAddress } from "@getpaseo/protocol/addresses"; +import { useHosts } from "@/runtime/host-runtime"; import { useCallback, useMemo, type ReactElement, type ReactNode } from "react"; import { Text, View } from "react-native"; import { useTranslation } from "react-i18next"; @@ -35,17 +37,17 @@ const foregroundMutedColorMapping = (theme: Theme) => ({ export interface SubagentsTrackProps { serverId: string; rows: SubagentRow[]; - onOpenSubagent: (id: string) => void; + onOpenSubagent: (address: AgentAddress) => void; onOpenProviderSubagent: (parentAgentId: string, subagentId: string) => void; /** * Whether opening this row shows it in a sheet. A sheet stacks over the list and uncovers it on * close, so the list stays open; a tab replaces the list's pane, so choosing it ends the list. */ opensInSheet: (row: SubagentRow) => boolean; - onArchiveSubagent: (id: string) => void; + onArchiveSubagent: (address: AgentAddress) => void; onArchiveFinished?: () => void; archiveFinishedStatus?: ArchiveFinishedStatus; - onDetachSubagent?: (id: string) => void; + onDetachSubagent?: (address: AgentAddress) => void; } const IDLE_ARCHIVE_FINISHED_STATUS: ArchiveFinishedStatus = { kind: "idle" }; @@ -105,7 +107,7 @@ export function SubagentsTrack({ ) : null} {rows.map((row) => ( void; + onOpenSubagent: (address: AgentAddress) => void; onOpenProviderSubagent: (parentAgentId: string, subagentId: string) => void; opensInSheet: boolean; - onArchiveSubagent: (id: string) => void; - onDetachSubagent?: (id: string) => void; + onArchiveSubagent: (address: AgentAddress) => void; + onDetachSubagent?: (address: AgentAddress) => void; } function SubagentsTrackRow({ @@ -198,22 +200,24 @@ function SubagentsTrackRow({ }: SubagentsTrackRowProps): ReactElement { const { t } = useTranslation(); const isCompact = useIsCompactFormFactor(); - const presentation = useRowPresentation(row, serverId); + const rowServerId = row.kind === "paseo" ? row.address.serverId : serverId; + const hostLabel = useHosts().find((host) => host.serverId === rowServerId)?.label ?? rowServerId; + const presentation = useRowPresentation(row, rowServerId); const displayLabel = presentation.titleState === "loading" ? t("common.states.loading") : presentation.label; const handlePress = useCallback(() => { if (row.kind === "provider") { onOpenProviderSubagent(row.parentAgentId, row.id); } else { - onOpenSubagent(row.id); + onOpenSubagent(row.address); } }, [onOpenProviderSubagent, onOpenSubagent, row]); const handleArchivePress = useCallback(() => { - onArchiveSubagent(row.id); - }, [onArchiveSubagent, row.id]); + if (row.kind === "paseo") onArchiveSubagent(row.address); + }, [onArchiveSubagent, row]); const handleDetachPress = useCallback(() => { - onDetachSubagent?.(row.id); - }, [onDetachSubagent, row.id]); + if (row.kind === "paseo") onDetachSubagent?.(row.address); + }, [onDetachSubagent, row]); const actionsAlwaysVisible = isNative || isCompact; const renderRow = useCallback( @@ -225,7 +229,10 @@ function SubagentsTrackRow({ {displayLabel} {row.kind === "paseo" ? ( - + <> + + + ) : ( )} @@ -253,6 +260,8 @@ function SubagentsTrackRow({ row.kind, row.id, serverId, + rowServerId, + hostLabel, ], ); diff --git a/packages/app/src/subagents/use-archive-finished.ts b/packages/app/src/subagents/use-archive-finished.ts index e7653f054e..c4d6af94eb 100644 --- a/packages/app/src/subagents/use-archive-finished.ts +++ b/packages/app/src/subagents/use-archive-finished.ts @@ -1,3 +1,5 @@ +import { getHostRuntimeStore } from "@/runtime/host-runtime"; +import { getParentAddressFromLabels } from "@getpaseo/protocol/agent-labels"; import { useEffect, useMemo, useSyncExternalStore } from "react"; import { useArchiveAgent } from "@/hooks/use-archive-agent"; import { useSessionStore } from "@/stores/session-store"; @@ -32,8 +34,19 @@ export function useArchiveFinishedSubagents({ () => createArchiveFinishedSubagents([], { parentAgentId, - getManagedSubagent: (id) => useSessionStore.getState().sessions[serverId]?.agents.get(id), - archiveManagedSubagent: (id) => archiveAgent({ serverId, agentId: id }), + getManagedSubagent: (address) => { + const agent = useSessionStore + .getState() + .sessions[address.serverId]?.agents.get(address.agentId); + if (!agent) return undefined; + const parent = getParentAddressFromLabels(agent.labels, address.serverId); + if (parent?.serverId !== serverId || parent.agentId !== parentAgentId) return undefined; + return { ...agent, parentAgentId: parent.agentId }; + }, + archiveManagedSubagent: async (address) => { + getHostRuntimeStore().requireHostRoute(address.serverId, serverId); + await archiveAgent(address); + }, dismissProviderSubagents: (ids) => { useProviderSubagentStore.getState().hideFromTrack(serverId, parentAgentId, ids); }, diff --git a/packages/app/src/subagents/use-archive-subagent.ts b/packages/app/src/subagents/use-archive-subagent.ts index b00f9bb871..a72dd667be 100644 --- a/packages/app/src/subagents/use-archive-subagent.ts +++ b/packages/app/src/subagents/use-archive-subagent.ts @@ -1,3 +1,5 @@ +import { getHostRuntimeStore } from "@/runtime/host-runtime"; +import type { AgentAddress } from "@getpaseo/protocol/addresses"; import { useCallback } from "react"; import { useToast } from "@/contexts/toast-context"; import { useArchiveAgent } from "@/hooks/use-archive-agent"; @@ -17,26 +19,30 @@ export interface UseArchiveSubagentInput { serverId: string; } -export function useArchiveSubagent(input: UseArchiveSubagentInput): (subagentId: string) => void { +export function useArchiveSubagent({ + serverId: originServerId, +}: UseArchiveSubagentInput): (address: AgentAddress) => void { const { archiveAgent } = useArchiveAgent(); - const { serverId } = input; const toast = useToast(); return useCallback( - (subagentId: string) => { + ({ serverId, agentId: subagentId }: AgentAddress) => { void requestArchiveSubagent( { serverId, subagentId }, { getSubagent: (id): ResolveArchiveSubagentDialogInput | undefined => useSessionStore.getState().sessions[serverId]?.agents?.get(id), confirm: confirmDialog, - archiveAgent, + archiveAgent: async (address) => { + getHostRuntimeStore().requireHostRoute(address.serverId, originServerId); + await archiveAgent(address); + }, reportError: (error) => { toast.error(toErrorMessage(error)); }, }, ); }, - [archiveAgent, serverId, toast], + [archiveAgent, originServerId, toast], ); } diff --git a/packages/app/src/subagents/use-detach-subagent.ts b/packages/app/src/subagents/use-detach-subagent.ts index 36b2f2a560..28b145e967 100644 --- a/packages/app/src/subagents/use-detach-subagent.ts +++ b/packages/app/src/subagents/use-detach-subagent.ts @@ -1,3 +1,5 @@ +import { getHostRuntimeStore } from "@/runtime/host-runtime"; +import type { AgentAddress } from "@getpaseo/protocol/addresses"; import { useCallback } from "react"; import { useToast } from "@/contexts/toast-context"; import { i18n } from "@/i18n/i18next"; @@ -18,12 +20,13 @@ export interface UseDetachSubagentInput { serverId: string; } -export function useDetachSubagent(input: UseDetachSubagentInput): (subagentId: string) => void { - const { serverId } = input; +export function useDetachSubagent({ + serverId: originServerId, +}: UseDetachSubagentInput): (address: AgentAddress) => void { const toast = useToast(); return useCallback( - (subagentId: string) => { + ({ serverId, agentId: subagentId }: AgentAddress) => { void requestDetachSubagent( { serverId, subagentId }, { @@ -31,6 +34,7 @@ export function useDetachSubagent(input: UseDetachSubagentInput): (subagentId: s useSessionStore.getState().sessions[serverId]?.agents?.get(id), confirm: confirmDialog, detachAgent: async ({ serverId: targetServerId, agentId }) => { + getHostRuntimeStore().requireHostRoute(targetServerId, originServerId); const client = useSessionStore.getState().sessions[targetServerId]?.client; if (!client) { throw new Error(i18n.t("workspaceSetup.errors.hostDisconnected")); @@ -46,6 +50,6 @@ export function useDetachSubagent(input: UseDetachSubagentInput): (subagentId: s }, ); }, - [serverId, toast], + [originServerId, toast], ); } diff --git a/packages/app/src/types/host-connection.ts b/packages/app/src/types/host-connection.ts index 0962c02c7c..1521f7d919 100644 --- a/packages/app/src/types/host-connection.ts +++ b/packages/app/src/types/host-connection.ts @@ -51,7 +51,14 @@ export interface RelayHostConnection { daemonPublicKeyB64: string; } +export interface RoutedHostConnection { + id: string; + type: "routed"; + originServerId: string; +} + export type HostConnection = + | RoutedHostConnection | DirectTcpHostConnection | DirectSocketHostConnection | DirectPipeHostConnection @@ -128,6 +135,8 @@ function hostConnectionEquals(left: HostConnection, right: HostConnection): bool return false; } + if (left.type === "routed" && right.type === "routed") + return left.originServerId === right.originServerId; if (left.type === "directTcp" && right.type === "directTcp") { return left.endpoint === right.endpoint && (left.useTls ?? false) === (right.useTls ?? false); } @@ -399,6 +408,7 @@ export function createRemoteSshHostConnection(input: { } const StoredHostConnectionSchema = z.discriminatedUnion("type", [ + z.strictObject({ id: z.string(), type: z.literal("routed"), originServerId: z.string().min(1) }), z.strictObject({ id: z.string().optional(), type: z.literal("directTcp"), @@ -446,6 +456,7 @@ export const StoredHostRegistrySchema = z.array(StoredHostProfileSchema); type StoredHostConnection = z.infer; function normalizeStoredConnection(connection: StoredHostConnection): HostConnection | null { + if (connection.type === "routed") return connection; if (connection.type === "directTcp") { try { const endpoint = normalizeLoopbackToLocalhost(normalizeHostPort(connection.endpoint)); diff --git a/packages/app/src/utils/agent-snapshots.ts b/packages/app/src/utils/agent-snapshots.ts index c1c70c640e..13a7bcd14b 100644 --- a/packages/app/src/utils/agent-snapshots.ts +++ b/packages/app/src/utils/agent-snapshots.ts @@ -1,6 +1,6 @@ import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages"; import type { AgentPermissionRequest } from "@getpaseo/protocol/agent-types"; -import { getParentAgentIdFromLabels } from "@getpaseo/protocol/agent-labels"; +import { getParentAddressFromLabels } from "@getpaseo/protocol/agent-labels"; import { TURN_LIVENESS_IDLE, type ActiveTurnIdentity, @@ -98,7 +98,8 @@ export function normalizeAgentSnapshot(snapshot: AgentSnapshotPayload, serverId: ? new Date(snapshot.attentionTimestamp) : null; const archivedAt = snapshot.archivedAt ? new Date(snapshot.archivedAt) : null; - const parentAgentId = getParentAgentIdFromLabels(snapshot.labels); + const parent = getParentAddressFromLabels(snapshot.labels, serverId); + const parentAgentId = parent?.serverId === serverId ? parent.agentId : null; // COMPAT(agentTurnIdentity): added in v0.2.6, remove after 2027-01-31 once daemon floor >= v0.2.6. // Old daemons expose only status. Normalize that legacy signal once so the rest // of the app consumes one activity shape. diff --git a/packages/app/src/utils/test-daemon-connection.ts b/packages/app/src/utils/test-daemon-connection.ts index b882b1f30a..5942b45621 100644 --- a/packages/app/src/utils/test-daemon-connection.ts +++ b/packages/app/src/utils/test-daemon-connection.ts @@ -146,6 +146,8 @@ export async function buildClientConfig( | "buildDesktopTransportUrl" > = defaultDaemonConnectionDependencies, ): Promise { + if (connection.type === "routed") + throw new Error("Routed host connections are resolved by HostRuntime"); const clientId = await deps.getClientId(); const desktopTransportFactory = deps.createDesktopTransportFactory(); const base = { diff --git a/packages/client/package.json b/packages/client/package.json index 57c937656d..d3f5cabb63 100644 --- a/packages/client/package.json +++ b/packages/client/package.json @@ -28,6 +28,10 @@ "./node": { "types": "./dist/node/index.d.ts", "default": "./dist/node/index.js" + }, + "./internal/hosts": { + "types": "./dist/hosts/index.d.ts", + "default": "./dist/hosts/index.js" } }, "publishConfig": { diff --git a/packages/client/src/index.test.ts b/packages/client/src/index.test.ts index e6b15d06bf..a7bb17121c 100644 --- a/packages/client/src/index.test.ts +++ b/packages/client/src/index.test.ts @@ -1735,3 +1735,38 @@ test("canceled timeline handles and captured state are collectible while their A ); expect(result.stdout).toContain('"phase":"API alive"'); }, 20000); + +test("qualified parents preserve foreign identity without resolving an equal local agent ID", async () => { + const { client, ws } = await connectClient(); + const localParent = client.agents.ref("same-parent"); + expect(localParent.address).toEqual({ serverId: "srv_sdk_test", agentId: "same-parent" }); + const creating = client.agents.create({ + config: { provider: "codex/gpt-5.4" }, + cwd: "/repo/sdk", + parent: { serverId: "srv_parent_elsewhere", agentId: "same-parent" }, + labels: { task: "preserved" }, + }); + const request = parseSentSessionMessage(ws.sent.at(-1)); + expect(request).not.toHaveProperty("callerAgentId"); + expect(request).toMatchObject({ + labels: { + task: "preserved", + "paseo.parent-agent-id": "same-parent", + "paseo.parent-server-id": "srv_parent_elsewhere", + }, + }); + ws.message( + sessionMessage({ + type: "status", + payload: { + status: "agent_created", + requestId: request.requestId, + agentId: "agent_sdk", + agent: createAgent(), + }, + }), + ); + await creating; + await client.close(); + expect(localParent.address).toEqual({ serverId: "srv_sdk_test", agentId: "same-parent" }); +}); diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index c9f9c3fa9d..455d377634 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -1,3 +1,6 @@ +import type { AgentAddress } from "@getpaseo/protocol/addresses"; +import { parentAddressLabels } from "@getpaseo/protocol/agent-labels"; +export type { AgentAddress } from "@getpaseo/protocol/addresses"; import { createHostActions, type PaseoHostActions } from "./hosts/index.js"; export type { PaseoHostActions } from "./hosts/index.js"; export type { HostAdd, HostUpdate, HostSummary, HostDirectory } from "@getpaseo/protocol/hosts"; @@ -249,7 +252,7 @@ export interface PaseoAgentCreateOptions { onEvent?: (snapshot: CreationSnapshot) => void; config: PaseoAgentConfig; cwd: string; - parent?: string | PaseoAgentHandle; + parent?: string | AgentAddress | PaseoAgentHandle; title?: PaseoAgentSessionConfig["title"]; env?: CreateAgentRequestMessage["env"]; prompt?: string; @@ -343,6 +346,7 @@ export interface PaseoAgentTimelineHandle { export interface PaseoAgentHandle { readonly id: string; + readonly address: AgentAddress; /** * `workspaceId` through `archivedAt` mirror the last snapshot this handle * observed. A handle from `ref()` reads `null` for all of them until @@ -530,10 +534,12 @@ export function createPaseoClient(config: PaseoClientConfig): PaseoClient { function toDaemonAgentCreateOptions( options: PaseoAgentCreateOptions, placement?: { workspaceId: string; cwd: string }, + serverId?: string, ): CreateAgentRequestOptions { const { config: agentConfig, cwd, parent, title, prompt, ...requestOptions } = options; const { provider: providerModel, options: providerOptions, ...runtimeConfig } = agentConfig; const { provider, model } = parseProviderModel(providerModel); + const relationship = parent ? encodeParent(parent, serverId) : {}; return { ...requestOptions, config: { @@ -545,7 +551,12 @@ function toDaemonAgentCreateOptions( ...(providerOptions !== undefined ? { providerOptions } : {}), }, ...(placement ? { workspaceId: placement.workspaceId } : {}), - ...(parent ? { callerAgentId: resolveAgentId(parent) } : {}), + ...(parent + ? { + ...relationship, + labels: { ...requestOptions.labels, ...relationship.labels }, + } + : {}), ...(prompt !== undefined ? { initialPrompt: prompt } : {}), }; } @@ -592,7 +603,13 @@ export function createPaseoApi( options: PaseoAgentCreateOptions, placement?: { workspaceId: string; cwd: string }, ) => { - const agent = await daemonClient.createAgent(toDaemonAgentCreateOptions(options, placement)); + const agent = await daemonClient.createAgent( + toDaemonAgentCreateOptions( + options, + placement, + daemonClient.getLastServerInfoMessage()?.serverId, + ), + ); return createAgentHandle(agent); }; const terminals = createTerminalActions(daemonClient, async (workspaceId) => { @@ -717,7 +734,18 @@ export function createPaseoApi( openWorkspace(daemonClient, createWorkspaceHandle, input, requestId), create: async ({ requestId, agent, ...options }) => { const result = await daemonClient.createWorkspace( - { ...options, ...(agent ? { agent: toDaemonAgentCreateOptions(agent) } : {}) }, + { + ...options, + ...(agent + ? { + agent: toDaemonAgentCreateOptions( + agent, + undefined, + daemonClient.getLastServerInfoMessage()?.serverId, + ), + } + : {}), + }, requestId, ); if (result.error || !result.workspace) { @@ -870,8 +898,14 @@ function createAgentHandleFactory( const id = typeof agent === "string" ? agent : agent.id; let current = typeof agent === "string" ? null : agent; + let serverId = daemonClient.getLastServerInfoMessage()?.serverId; const handle: PaseoAgentHandle = { id, + get address() { + serverId ??= daemonClient.getLastServerInfoMessage()?.serverId; + if (!serverId) throw new Error("Connect to a host before obtaining an agent address"); + return { serverId, agentId: id }; + }, timeline: { append: (item) => daemonClient.appendAgentTimelineItem(id, item), refetch: async (options) => { @@ -1020,8 +1054,16 @@ function resolveWorkspaceId(workspace: string | PaseoWorkspaceHandle): string { return typeof workspace === "string" ? workspace : workspace.id; } -function resolveAgentId(agent: string | PaseoAgentHandle): string { - return typeof agent === "string" ? agent : agent.id; +function encodeParent( + parent: string | AgentAddress | PaseoAgentHandle, + serverId?: string, +): Pick { + if (typeof parent === "string") return { callerAgentId: parent }; + const address = "address" in parent ? parent.address : parent; + if (!serverId) throw new Error("Connect to a host before creating a child"); + return address.serverId === serverId + ? { callerAgentId: address.agentId, labels: parentAddressLabels(address) } + : { labels: parentAddressLabels(address) }; } function parseProviderModel(selection: string): { provider: string; model: string } { diff --git a/packages/protocol/package.json b/packages/protocol/package.json index 770ae5f50f..1dd6665592 100644 --- a/packages/protocol/package.json +++ b/packages/protocol/package.json @@ -12,6 +12,10 @@ "./*": { "types": "./dist/*.d.ts", "default": "./dist/*.js" + }, + "./addresses": { + "types": "./dist/addresses.d.ts", + "default": "./dist/addresses.js" } }, "publishConfig": { diff --git a/packages/protocol/src/addresses.ts b/packages/protocol/src/addresses.ts new file mode 100644 index 0000000000..6a8328b93c --- /dev/null +++ b/packages/protocol/src/addresses.ts @@ -0,0 +1,5 @@ +/** Agent identity is qualified by its daemon; transport and reachability are separate. */ +export interface AgentAddress { + serverId: string; + agentId: string; +} diff --git a/packages/protocol/src/agent-labels.ts b/packages/protocol/src/agent-labels.ts index 2bf05950dd..70cbaf59b1 100644 --- a/packages/protocol/src/agent-labels.ts +++ b/packages/protocol/src/agent-labels.ts @@ -1,3 +1,5 @@ +import type { AgentAddress } from "./addresses.js"; +export const PARENT_SERVER_ID_LABEL = "paseo.parent-server-id"; export const PARENT_AGENT_ID_LABEL = "paseo.parent-agent-id"; const OPEN_AGENT_TAB_LABEL_PREFIX = "paseo.open-agent-tab."; @@ -29,3 +31,23 @@ export function hasOpenAgentTab(labels: Record | null | undefin ([label, value]) => isOpenAgentTabLabel(label) && value === "true", ); } + +// COMPAT(localParentLabels): added in v0.12.0; remove after 2027-04-20 once stored local parent labels are qualified. +export function getParentAddressFromLabels( + labels: Record | null | undefined, + serverId: string, +): AgentAddress | null { + const agentId = getParentAgentIdFromLabels(labels); + if (!agentId) return null; + const parentServerId = labels?.[PARENT_SERVER_ID_LABEL]; + return { + serverId: + typeof parentServerId === "string" && parentServerId.trim() + ? parentServerId.trim() + : serverId, + agentId, + }; +} +export function parentAddressLabels(parent: AgentAddress): Record { + return { [PARENT_AGENT_ID_LABEL]: parent.agentId, [PARENT_SERVER_ID_LABEL]: parent.serverId }; +} diff --git a/packages/protocol/src/paseo-tool-presentation/catalog.ts b/packages/protocol/src/paseo-tool-presentation/catalog.ts index 1706134523..0c35b128b7 100644 --- a/packages/protocol/src/paseo-tool-presentation/catalog.ts +++ b/packages/protocol/src/paseo-tool-presentation/catalog.ts @@ -195,6 +195,7 @@ export const PASEO_TOOL_CATALOG = { context: workspaceContext(), detail: { inputOrder: WORKSPACE_FIELDS, outputFields: ["workspaceId", "projectId"] }, }, + list_hosts: { label: "List hosts" }, list_workspaces: { label: "List workspaces" }, rename_workspace: { label: "Rename workspace", diff --git a/packages/server/src/server/agent/agent-prompt.ts b/packages/server/src/server/agent/agent-prompt.ts index bf6d50480b..6b120abb1c 100644 --- a/packages/server/src/server/agent/agent-prompt.ts +++ b/packages/server/src/server/agent/agent-prompt.ts @@ -1,3 +1,4 @@ +import type { WaitForFinishResult } from "@getpaseo/client/internal/daemon-client"; import { prepareAgentMessage, type AgentPromptSource } from "./agent-messages/index.js"; import type { Logger } from "pino"; @@ -10,7 +11,10 @@ import type { AgentManager, ManagedAgent } from "./agent-manager.js"; import type { AgentStorage } from "./agent-storage.js"; import { ensureAgentLoaded } from "./agent-loading.js"; import { isStaleProviderSessionError } from "./stale-provider-session-error.js"; -import { getParentAgentIdFromLabels } from "@getpaseo/protocol/agent-labels"; +import { + getParentAgentIdFromLabels, + getParentAddressFromLabels, +} from "@getpaseo/protocol/agent-labels"; import type { ActiveTurnBehavior } from "@getpaseo/protocol/messages"; export type AgentUnarchiveController = Pick; @@ -292,7 +296,7 @@ async function resolvePromptSource( manager: Pick, storage: AgentStorage, ): Promise { - if (!source) return undefined; + if (!source || source.title) return source; const title = ( manager.getAgent(source.agentId)?.config.title ?? (await storage.get(source.agentId))?.title )?.trim(); @@ -401,6 +405,7 @@ export interface SetupFinishNotificationParams { agentStorage: AgentStorage; childAgentId: string; callerAgentId: string; + childServerId?: string; requireParentOwnership?: boolean; /** Retain pre-dispatch outcomes only when dispatch keeps the existing turn. */ dispatchResult?: Promise; @@ -429,6 +434,7 @@ const FINISH_NOTIFICATION_MESSAGE_LIMIT = 4000; interface FinishNotificationBodyInput { childAgentId: string; + serverId?: string; title: string; reason: FinishNotificationReason; lastAssistantMessage: string | null; @@ -438,12 +444,14 @@ interface FinishNotificationBodyInput { function formatFinishNotificationBody(params: FinishNotificationBodyInput): string { const statusLine = `Agent ${params.childAgentId} (${params.title}) ${params.reason}.`; const sections = [statusLine]; + if (params.serverId) sections.push(`Server: ${params.serverId}`); if (params.reason === "needs permission" && params.permissionRequest) { sections.push( "Respond with `respond_to_permission` using the `agentId` and `requestId` below.", `\n${JSON.stringify( { agentId: params.childAgentId, + ...(params.serverId ? { serverId: params.serverId } : {}), requestId: params.permissionRequest.id, request: params.permissionRequest, }, @@ -480,6 +488,101 @@ const armedFinishNotifications = new WeakMap< Map> >(); +/** Creation waits and accepted follow-ups share notification ownership. */ +export function setupFinishWait(params: { + agentManager: AgentManager; + agentStorage: AgentStorage; + originServerId?: string; + childServerId?: string; + childAgentId: string; + callerAgentId: string; + wait(signal: AbortSignal): Promise; + afterPermission(requestId: string, signal: AbortSignal): Promise; + logger: Logger; +}): () => void { + const controller = new AbortController(); + const byManager = + armedFinishNotifications.get(params.agentManager) ?? + new Map>(); + armedFinishNotifications.set(params.agentManager, byManager); + const key = JSON.stringify([ + params.childServerId ?? null, + params.childAgentId, + params.callerAgentId, + ]); + const registrations = byManager.get(key) ?? new Set(); + byManager.set(key, registrations); + for (const previous of registrations) previous.stop(); + byManager.set(key, registrations); + let superseded = Promise.resolve(false); + const stop = () => { + controller.abort(); + registrations.delete(registration); + if (registrations.size === 0 && byManager.get(key) === registrations) byManager.delete(key); + }; + const registration: ArmedFinishNotification = { + stop, + supersedeWhen(accepted) { + superseded = superseded.then(async (prior) => prior || (await accepted)); + }, + }; + registrations.add(registration); + async function watch(): Promise { + while (!controller.signal.aborted) { + const result = await params.wait(controller.signal); + if (controller.signal.aborted || (await superseded) || result.status === "timeout") return; + const parent = getParentAddressFromLabels( + result.final?.labels, + params.childServerId ?? "self", + ); + if ( + parent?.agentId !== params.callerAgentId || + parent.serverId !== (params.originServerId ?? "self") + ) + return; + if ((await params.agentStorage.get(params.callerAgentId))?.archivedAt) return; + const reason = { idle: "finished", error: "errored", permission: "needs permission" }[ + result.status + ] as FinishNotificationReason; + const permissionRequest = result.final?.pendingPermissions[0]; + await sendPromptToAgent({ + agentManager: params.agentManager, + agentStorage: params.agentStorage, + agentId: params.callerAgentId, + prompt: formatFinishNotificationBody({ + childAgentId: params.childAgentId, + serverId: params.childServerId, + title: result.final?.title ?? params.childAgentId, + reason, + lastAssistantMessage: result.lastMessage, + permissionRequest, + }), + source: { + kind: "agent-notification", + agentId: params.childAgentId, + title: result.final?.title ?? params.childAgentId, + event: finishNotificationEvents[reason], + }, + activeTurnBehavior: "steer", + unarchive: false, + logger: params.logger, + }); + if (result.status !== "permission" || !permissionRequest) return; + await params.afterPermission(permissionRequest.id, controller.signal); + } + } + void watch() + .catch((error) => { + if (!controller.signal.aborted) + params.logger.warn( + { err: error, agentId: params.childAgentId }, + "Child finish wait stopped", + ); + }) + .finally(stop); + return stop; +} + export function setupFinishNotification(params: SetupFinishNotificationParams): () => void { const { agentManager, @@ -499,7 +602,7 @@ export function setupFinishNotification(params: SetupFinishNotificationParams): const armedByManager = armedFinishNotifications.get(agentManager) ?? new Map(); armedFinishNotifications.set(agentManager, armedByManager); - const armedKey = JSON.stringify([childAgentId, callerAgentId]); + const armedKey = JSON.stringify([params.childServerId ?? null, childAgentId, callerAgentId]); const registrations = armedByManager.get(armedKey) ?? new Set(); armedByManager.set(armedKey, registrations); const previous = [...registrations]; diff --git a/packages/server/src/server/agent/host-operations/index.ts b/packages/server/src/server/agent/host-operations/index.ts new file mode 100644 index 0000000000..9c45ef4576 --- /dev/null +++ b/packages/server/src/server/agent/host-operations/index.ts @@ -0,0 +1,323 @@ +import { parentAddressLabels } from "@getpaseo/protocol/agent-labels"; +import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages"; +import { randomUUID } from "node:crypto"; +import { z } from "zod"; +import { + DaemonClient, + type FetchWorkspacesEntry, + type WaitForFinishResult, +} from "@getpaseo/client/internal/daemon-client"; +import type { DaemonTransportFactory } from "@getpaseo/client/internal/daemon-client-transport-types"; +import type { HostDirectory } from "@getpaseo/protocol/hosts"; +import { OWNER_PERMISSIONS } from "../../authorization/index.js"; +import type { ConnectedHosts } from "../../connected-hosts/index.js"; +import { resolveRequiredProviderModel } from "../mcp-shared.js"; +import type { PaseoToolExecutionContext, PaseoToolResult } from "../tools/types.js"; + +export type HostOperation = ( + input: Record, + context: PaseoToolExecutionContext, +) => Promise; +export interface HostOperations { + createAgent: HostOperation; + createWorkspace: HostOperation; + listWorkspaces: HostOperation; + wait(agentId: string, signal: AbortSignal): Promise; + observe( + agentId: string, + handler: (snapshot: AgentSnapshotPayload | null) => void, + signal: AbortSignal, + ): Promise<() => Promise>; +} +export interface ToolHosts { + directory(): Promise; + use(serverId: string, action: (client: DaemonClient) => Promise): Promise; +} + +/** Daemon MCP has the daemon's authority, bounded by the destination's admission of its link. */ +export function createToolHosts(hosts: ConnectedHosts): ToolHosts { + return { + directory: () => hosts.list(), + async use(serverId, action) { + const lifetime = new AbortController(); + const transportFactory: DaemonTransportFactory = () => { + const opens = new Set<() => void>(); + const closes = new Set<(event?: unknown) => void>(); + const errors = new Set<(event: unknown) => void>(); + const messages = new Set<(data: unknown, binary: boolean) => void>(); + let closed = false; + let route: Awaited> | undefined; + void hosts + .openRoute({ + serverId, + scopeId: randomUUID(), + sourceId: "mcp", + originSessionId: "mcp", + permissions: OWNER_PERMISSIONS, + signal: lifetime.signal, + frame: (data) => { + for (const listener of messages) listener(data, typeof data !== "string"); + }, + ended: (code, message) => { + for (const listener of errors) + listener(Object.assign(new Error(message), { code, serverId })); + for (const listener of closes) listener(); + }, + }) + .then((opened) => { + route = opened; + if (closed) opened.close(); + else for (const listener of opens) listener(); + return undefined; + }) + .catch((error) => { + for (const listener of errors) listener(error); + for (const listener of closes) listener(error); + }); + return { + send: (data) => { + if (!route) throw new Error("Host route is not open"); + route.send(data instanceof ArrayBuffer ? new Uint8Array(data) : data); + }, + close: () => { + closed = true; + route?.close(); + }, + onOpen: (listener) => { + opens.add(listener); + return () => opens.delete(listener); + }, + onClose: (listener) => { + closes.add(listener); + return () => closes.delete(listener); + }, + onError: (listener) => { + errors.add(listener); + return () => errors.delete(listener); + }, + onMessage: (listener) => { + messages.add(listener); + return () => messages.delete(listener); + }, + }; + }; + const client = new DaemonClient({ + url: "ws://host-route/ws", + clientId: randomUUID(), + transportFactory, + reconnect: { enabled: false }, + }); + try { + await client.connect(); + return await action(client); + } finally { + lifetime.abort(); + await client.close(); + } + }, + }; +} + +const remoteCreate = z + .object({ + provider: z.string().min(1), + title: z.string().trim().min(1).max(60), + initialPrompt: z.string().min(1), + workspaceId: z.string().min(1), + labels: z.record(z.string(), z.string()).optional(), + notifyOnFinish: z.boolean().optional(), + idempotencyKey: z.string().optional(), + settings: z + .object({ + modeId: z.string().optional(), + thinkingOptionId: z.string().optional(), + features: z.record(z.string(), z.unknown()).optional(), + }) + .optional(), + }) + .strict(); +const result = (structuredContent: unknown): PaseoToolResult => ({ + content: [], + structuredContent, +}); + +export function createHostResolver(input: { + serverId: string; + local: HostOperations; + hosts: () => ToolHosts; + callerAgentId?: string; + lockedCwd?: string; +}) { + return { + directory: () => input.hosts().directory(), + async use( + serverId: string | undefined, + action: (host: HostOperations) => Promise, + ): Promise { + const destination = serverId ?? input.serverId; + if (destination === input.serverId) return action(input.local); + if (input.lockedCwd) + throw new Error("This caller is restricted to its local working directory"); + return input.hosts().use(destination, async (client) => + action({ + async createAgent(args) { + const parsed = remoteCreate.parse(args); + const { provider, model } = resolveRequiredProviderModel(parsed.provider); + const workspace = (await client.fetchWorkspaces({})).entries.find( + (row) => row.id === parsed.workspaceId, + ); + if (!workspace) throw new Error(`Workspace ${parsed.workspaceId} not found`); + const labels = { + ...parsed.labels, + ...(input.callerAgentId + ? parentAddressLabels({ agentId: input.callerAgentId, serverId: input.serverId }) + : {}), + }; + const snapshot = await client.createAgent({ + config: { + provider, + model, + cwd: workspace.workspaceDirectory, + title: parsed.title, + modeId: parsed.settings?.modeId, + thinkingOptionId: parsed.settings?.thinkingOptionId, + featureValues: parsed.settings?.features, + }, + workspaceId: parsed.workspaceId, + labels, + initialPrompt: parsed.initialPrompt, + idempotencyKey: parsed.idempotencyKey, + }); + return result({ + agentId: snapshot.id, + type: snapshot.provider, + status: snapshot.status, + cwd: snapshot.cwd, + workspaceId: snapshot.workspaceId, + currentModeId: snapshot.currentModeId, + availableModes: snapshot.availableModes, + }); + }, + async listWorkspaces(args) { + const { entries } = await client.fetchWorkspaces({ + includeBackground: args.includeBackground === true, + }); + return result({ workspaces: entries.map(workspaceSummary) }); + }, + async createWorkspace(args) { + const isolation = z.enum(["local", "worktree"]).parse(args.isolation); + const path = z.string().min(1).parse(args.path); + const response = await client.createWorkspace({ + source: + isolation === "local" + ? { kind: "directory", path, projectId: args.projectId as string | undefined } + : { + kind: "worktree", + cwd: path, + projectId: args.projectId as string | undefined, + action: + args.mode === "checkout-branch" || args.mode === "checkout-pr" + ? "checkout" + : "branch-off", + baseBranch: args.baseBranch as string | undefined, + branchName: args.branchName as string | undefined, + refName: args.branch as string | undefined, + githubPrNumber: args.prNumber as number | undefined, + worktreeSlug: args.worktreeSlug as string | undefined, + }, + title: args.title as string | undefined, + background: args.background as boolean | undefined, + }); + if (response.error || !response.workspace) + throw new Error(response.error ?? "Workspace creation failed"); + return result(workspaceSummary(response.workspace)); + }, + async observe(agentId, handler, signal) { + const observation = client.observeTimeline([agentId], { signal }); + const refresh = async () => handler((await client.fetchAgent(agentId))?.agent ?? null); + const unsubscribe = observation.subscribe({ + snapshot: () => {}, + update: (message) => { + if (message.type !== "agent_stream" || message.payload.agentId !== agentId) return; + const event = message.payload.event; + if ( + event.type === "permission_resolved" || + event.type === "turn_started" || + event.type === "turn_completed" || + event.type === "turn_canceled" || + event.type === "turn_failed" + ) + void refresh().catch(() => handler(null)); + }, + error: () => handler(null), + }); + try { + await observation.ready; + await refresh(); + } catch (error) { + unsubscribe(); + await observation.release(); + throw error; + } + return async () => { + unsubscribe(); + await observation.release(); + }; + }, + wait: (agentId, signal) => { + const close = () => void client.close(); + signal.addEventListener("abort", close, { once: true }); + return client + .waitForFinish(agentId, 0) + .finally(() => signal.removeEventListener("abort", close)); + }, + }), + ); + }, + }; +} +function workspaceSummary(workspace: FetchWorkspacesEntry) { + return { + workspaceId: workspace.id, + projectId: workspace.projectId, + cwd: workspace.workspaceDirectory, + isolation: workspace.workspaceKind === "worktree" ? "worktree" : "local", + kind: workspace.workspaceKind === "checkout" ? "local_checkout" : workspace.workspaceKind, + title: workspace.title, + background: workspace.background, + }; +} + +/** Wait pauses at permissions; observe only until that request clears before re-arming it. */ +export async function waitForPermissionChange( + host: HostOperations, + agentId: string, + requestId: string, + signal: AbortSignal, +): Promise { + signal.throwIfAborted(); + let changed!: () => void; + let failed!: (error: unknown) => void; + const ready = new Promise((resolve, reject) => { + changed = resolve; + failed = reject; + }); + const abort = () => failed(signal.reason ?? new Error("Finish wait stopped")); + signal.addEventListener("abort", abort, { once: true }); + let release: (() => Promise) | undefined; + try { + release = await host.observe( + agentId, + (snapshot) => { + if (!snapshot) failed(new Error(`Agent ${agentId} is unavailable`)); + else if (!snapshot.pendingPermissions.some((request) => request.id === requestId)) + changed(); + }, + signal, + ); + await ready; + } finally { + signal.removeEventListener("abort", abort); + await release?.(); + } +} diff --git a/packages/server/src/server/agent/tools/paseo-tools.ts b/packages/server/src/server/agent/tools/paseo-tools.ts index 30f28426b1..e1d88179ac 100644 --- a/packages/server/src/server/agent/tools/paseo-tools.ts +++ b/packages/server/src/server/agent/tools/paseo-tools.ts @@ -1,3 +1,9 @@ +import { + createHostResolver, + waitForPermissionChange, + type HostOperation, + type ToolHosts, +} from "../host-operations/index.js"; import { AgentDirectory } from "../directory.js"; import { stat } from "node:fs/promises"; import { z } from "zod"; @@ -72,6 +78,7 @@ import { type PromptDispatchDisposition, sendPromptToAgent, setupFinishNotification, + setupFinishWait, waitForAgentRunStartWithTimeout, } from "../agent-prompt.js"; import { respondToAgentPermission } from "../permission-response.js"; @@ -110,6 +117,8 @@ import type { ProviderPaseoToolsPolicy } from "@getpaseo/protocol/provider-confi import { isPaseoToolEnabled } from "../paseo-tool-policy.js"; export interface PaseoToolHostDependencies { + serverId?: string; + hosts?: () => ToolHosts; agentManager: AgentManager; agentStorage: AgentStorage; terminalManager?: TerminalManager | null; @@ -165,6 +174,10 @@ export interface PaseoToolHostDependencies { logger: Logger; } +function settledStatus(status: string): "error" | "idle" { + return status === "error" ? "error" : "idle"; +} + function parseTimestamp(value: string | null | undefined): number { if (!value) { return 0; @@ -685,7 +698,58 @@ function createToolCatalog( return result.data; }; + const localHostOperations = new Map(); const tools = new Map(); + const hostTools = new Map([ + ["create_agent", "createAgent"], + ["create_workspace", "createWorkspace"], + ["list_workspaces", "listWorkspaces"], + ] as const); + const hostResolver = createHostResolver({ + serverId: options.serverId ?? "self", + callerAgentId, + lockedCwd: callerContext?.lockedCwd, + hosts: () => { + if (!options.hosts) throw new Error("Connected hosts are not configured"); + return options.hosts(); + }, + local: { + createAgent: (args, context) => localHostOperations.get("create_agent")!(args, context), + createWorkspace: (args, context) => + localHostOperations.get("create_workspace")!(args, context), + listWorkspaces: (args, context) => localHostOperations.get("list_workspaces")!(args, context), + async observe(agentId, handler, signal) { + const unsubscribe = agentManager.subscribe( + (event) => { + if (event.type === "agent_state") handler(toAgentPayload(event.agent)); + }, + { agentId, replayState: true }, + ); + const stop = () => unsubscribe(); + signal.addEventListener("abort", stop, { once: true }); + return async () => { + signal.removeEventListener("abort", stop); + unsubscribe(); + }; + }, + async wait(agentId, signal) { + const settled = await agentManager.waitForAgentEvent(agentId, { + signal, + waitForActive: true, + }); + const snapshot = agentManager.getAgent(agentId); + const final = snapshot + ? await serializeSnapshotWithMetadata(agentStorage, snapshot, childLogger) + : null; + return { + status: settled.permission ? "permission" : settledStatus(settled.status), + final, + lastMessage: settled.lastMessage, + error: final?.lastError ?? null, + }; + }, + }, + }); const registerTool = ( name: string, config: PaseoToolConfig, @@ -695,6 +759,73 @@ function createToolCatalog( if (!isPaseoToolEnabled(options.paseoToolPolicy, name)) { return; } + const operation = hostTools.get( + name as "create_agent" | "create_workspace" | "list_workspaces", + ); + if (operation) { + localHostOperations.set(name, handler); + const original = config.inputSchema; + const selector = { + serverId: z + .string() + .min(1) + .optional() + .describe("Destination server ID from list_hosts; omit for this host."), + }; + config = { + ...config, + inputSchema: + original instanceof z.ZodObject + ? original.extend(selector) + : { ...(original as z.ZodRawShape), ...selector }, + }; + handler = async ({ serverId, ...input }, context) => { + const destination = serverId ?? options.serverId; + try { + const response = await hostResolver.use(serverId, (host) => + host[operation](input, context), + ); + if (name === "create_agent" && callerAgentId && input.notifyOnFinish !== false) { + const created = response.structuredContent as { agentId: string }; + setupFinishWait({ + agentManager, + agentStorage, + originServerId: options.serverId, + childServerId: destination, + childAgentId: created.agentId, + callerAgentId, + logger: childLogger, + wait: (signal) => + hostResolver.use(destination, (host) => host.wait(created.agentId, signal)), + afterPermission: (requestId, signal) => + hostResolver.use(destination, (host) => + waitForPermissionChange(host, created.agentId, requestId, signal), + ), + }); + } + return { + ...response, + structuredContent: { + ...(response.structuredContent as object), + ...(destination ? { serverId: destination } : {}), + }, + }; + } catch (error) { + if (!options.serverId) throw error; + const failure = { + serverId: destination, + error: error instanceof Error ? error.message : String(error), + ...(error instanceof Error && "code" in error ? { code: error.code } : {}), + ...(error instanceof Error && "outcome" in error ? { outcome: error.outcome } : {}), + }; + return { + isError: true, + content: [{ type: "text", text: JSON.stringify(failure) }], + structuredContent: failure, + }; + } + }; + } tools.set(name, { name, title: config.title, @@ -1332,6 +1463,19 @@ function createToolCatalog( }); } + registerTool( + "list_hosts", + { + title: "List hosts", + description: "Discover this host and registered destination hosts for fleet tools.", + inputSchema: {}, + }, + async () => ({ + content: [], + structuredContent: ensureValidJson(await hostResolver.directory()), + }), + ); + registerTool( "create_workspace", { @@ -1593,10 +1737,15 @@ function createToolCatalog( workspaceId: resolvedArgs.workspaceId, thinking: parsedArgs.settings?.thinkingOptionId, features: parsedArgs.settings?.features, - labels: parsedArgs.labels, + labels: { + ...parsedArgs.labels, + ...(callerAgentId && !resolvedArgs.detached && options.serverId + ? { "paseo.parent-server-id": options.serverId } + : {}), + }, mode: parsedArgs.settings?.modeId, promptFailure: "throw", - notifyOnFinish, + notifyOnFinish: false, detached: resolvedArgs.detached, callerAgentId, callerContext, @@ -2003,6 +2152,7 @@ function createToolCatalog( agentManager, agentStorage, childAgentId: agentId, + childServerId: options.serverId, callerAgentId, waitForTurnStart: false, dispatchResult, @@ -2023,6 +2173,7 @@ function createToolCatalog( agentManager, agentStorage, childAgentId: agentId, + childServerId: options.serverId, callerAgentId, waitForTurnStart: true, logger: childLogger, diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 73e668735f..5125a0b923 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -1393,6 +1393,11 @@ export async function createPaseoDaemon( const createAgentToolHostDependencies = ( runtime: PaseoToolRuntimeContext, ): PaseoToolHostDependencies => ({ + serverId, + hosts: () => { + if (!wsServer) throw new Error("Daemon is not accepting host operations"); + return wsServer.toolHosts(); + }, agentManager, agentStorage, terminalManager, diff --git a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts index 0005946316..a4211f4226 100644 --- a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts +++ b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts @@ -1,3 +1,7 @@ +import { z } from "zod"; +import { createTestAgentClients } from "../test-utils/fake-agent-client.js"; +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; import { afterEach, expect, test } from "vitest"; import { createPaseoApi } from "@getpaseo/client"; import { createHostClient, createHostChannelTransport } from "@getpaseo/client/node"; @@ -8,8 +12,10 @@ const cleanup: (() => Promise)[] = []; afterEach(async () => { for (const close of cleanup.splice(0).toReversed()) await close(); }); -async function daemon(): Promise { - const instance = await createTestPaseoDaemon(); +async function daemon( + options: Parameters[0] = {}, +): Promise { + const instance = await createTestPaseoDaemon(options); cleanup.push(() => instance.close()); return instance; } @@ -1303,3 +1309,79 @@ test("cancelling a logical scope during admission settles it and releases its ow endpoint.close(); await delivery.close(); }); + +test("agent MCP discovers hosts, creates children on self and B, and notifies their parent", async () => { + const parentPrompts: string[] = []; + const a = await daemon({ + agentClients: createTestAgentClients({ + onStartTurn: (prompt, config) => { + if (config.title === "Fleet parent") parentPrompts.push(JSON.stringify(prompt)); + }, + }), + }); + const b = await daemon(); + const { paseo } = await api(a); + await paseo.hosts.add({ connection: `127.0.0.1:${b.port}`, label: "Fleet B" }); + const directory = await online(paseo); + const workspace = await paseo.workspaces.create({ + source: { kind: "directory", path: a.staticDir }, + }); + const parent = await workspace.agents.create({ + config: { provider: "claude/claude-test" }, + title: "Fleet parent", + }); + const mcp = new Client({ name: "fleet-delegation", version: "1" }); + cleanup.push(() => mcp.close()); + await mcp.connect( + new StreamableHTTPClientTransport( + new URL(`http://127.0.0.1:${a.port}/mcp/agents?callerAgentId=${parent.id}`), + ), + ); + const result = await mcp.callTool({ name: "list_hosts", arguments: {} }); + expect(result.isError).not.toBe(true); + expect(result.structuredContent).toMatchObject({ + self: { serverId: directory.self.serverId }, + hosts: [expect.objectContaining({ serverId: directory.hosts[0]!.serverId, label: "Fleet B" })], + }); + const call = async (name: string, args: Record) => { + const response = await mcp.callTool({ name, arguments: args }); + expect(response.isError, JSON.stringify(response)).not.toBe(true); + return z.record(z.string(), z.unknown()).parse(response.structuredContent); + }; + const serverId = directory.hosts[0]!.serverId!; + const destination = await call("create_workspace", { + serverId, + isolation: "local", + path: b.staticDir, + title: "Destination workspace", + }); + expect(await call("list_workspaces", { serverId })).toMatchObject({ + serverId, + workspaces: [expect.objectContaining({ workspaceId: destination.workspaceId })], + }); + const self = await call("create_agent", { + title: "Self child", + provider: "claude/claude-test", + initialPrompt: "Finish local delegation", + }); + const remote = await call("create_agent", { + serverId, + workspaceId: destination.workspaceId, + title: "B child", + provider: "claude/claude-test", + initialPrompt: "Finish destination delegation", + }); + expect(self.serverId).toBe(directory.self.serverId); + expect(remote.serverId).toBe(serverId); + const bApi = await api(b); + expect( + (await bApi.paseo.agents.ref(String(remote.agentId)).refresh())?.agent.labels, + ).toMatchObject({ + "paseo.parent-agent-id": parent.id, + "paseo.parent-server-id": directory.self.serverId, + }); + await expect.poll(() => parentPrompts.join("\n")).toContain(`Agent ${self.agentId}`); + await expect.poll(() => parentPrompts.join("\n")).toContain(`Agent ${remote.agentId}`); + expect(parentPrompts.join("\n")).toContain(`Server: ${serverId}`); + expect(parentPrompts.join("\n")).toContain(`Server: ${directory.self.serverId}`); +}); diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 2c8da4ed42..b985503fd7 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -1,3 +1,4 @@ +import { createToolHosts } from "./agent/host-operations/index.js"; import { HOST_ROUTE_OPCODE } from "@getpaseo/protocol/binary-frames/index"; import { DaemonLinkEndpoint, VirtualHostSocket } from "./connected-hosts/index.js"; import { ConnectedHosts } from "./connected-hosts/index.js"; @@ -558,6 +559,10 @@ export class VoiceAssistantWebSocketServer { private readonly downloadTokenStore: DownloadTokenStore; private readonly paseoHome: string; private readonly connectedHosts: ConnectedHosts; + + toolHosts() { + return createToolHosts(this.connectedHosts); + } private readonly passwordHash: string | undefined; private readonly credentialSource: DaemonAuthConfig | undefined; private readonly worktreesRoot: string | undefined; diff --git a/public-docs/sdk/reference.md b/public-docs/sdk/reference.md index 6e606e35ba..2867261c32 100644 --- a/public-docs/sdk/reference.md +++ b/public-docs/sdk/reference.md @@ -59,6 +59,8 @@ Create a new client after `close()`. Creation options include `config`, `cwd`, `parent`, `title`, `prompt`, `env`, `outputSchema`, `images`, `attachments`, `git`, `worktree`, `autoArchive`, `background`, and `labels`. +`parent` accepts a host-local agent ID, an `AgentAddress` (`{ serverId, agentId }`), or an agent handle. A handle’s `address` keeps its host identity after disconnect. When creating through `hosts.client(serverId)`, pass the parent handle or complete address; the SDK carries parent metadata to the destination. + `background` is workspace creation intent. It cannot accompany an existing `workspaceId`. See [background workspaces](./workspaces.md) for caller inheritance and inclusive discovery. @@ -80,6 +82,7 @@ See [background workspaces](./workspaces.md) for caller inheritance and inclusiv | Member | Result | Behavior | | ------------------------------ | --------------------------------- | ------------------------------------------------------------------------------------------------------- | | `id` | `string` | Stable daemon agent ID. | +| `address` | `AgentAddress` | Complete `{ serverId, agentId }` identity; requires a connected host to establish it. | | `workspaceId` | `string \| null` | Current workspace placement. | | `cwd` | `string \| null` | Current working directory. | | `status` | Agent status or `null` | Current lifecycle status. |