From 8342a95367b587decb43d252856e0c9ec2c1dd4a Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 21 Oct 2026 13:52:10 +0700 Subject: [PATCH] Create, prompt, and monitor agents across connected hosts (#7507) * 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. * test(app): normalize subagent fixtures through snapshots * docs(protocol): clarify same-host parent labels * fix(hosts): resolve subagents through destination sessions A child address identifies its destination regardless of the connection used. Reuse the existing host candidate selection for discovered routes and direct connections, and keep navigation and row actions independent of the parent transport. * fix(hosts): keep routed destinations distinct during discovery A routed candidate names only the origin daemon. Match it within the named destination profile so two destinations discovered through the same origin retain separate identities and connections, including existing saved routes. * feat(hosts): prompt and query agents on selected hosts Extend the existing MCP host operation boundary with prompt, status, activity and list operations. Local calls retain their handlers and notification owner; selected destinations use ordinary client RPCs and the existing idle finish wait. --- docs/connected-hosts.md | 20 + .../e2e/browser/subagent-origin-pane.spec.ts | 197 ++++++++- packages/app/e2e/support/helpers/subagents.ts | 155 ++++++- .../src/desktop/daemon/local-credential.ts | 1 + packages/app/src/hooks/use-archive-agent.ts | 5 + packages/app/src/panels/agent-tracks.tsx | 16 +- packages/app/src/runtime/host-runtime.test.ts | 117 +++++- packages/app/src/runtime/host-runtime.ts | 117 +++++- .../app/src/screens/settings/host-page.tsx | 1 + .../workspace-subagents-integration.test.ts | 18 +- .../src/subagents/archive-finished.test.ts | 27 +- .../app/src/subagents/archive-finished.ts | 24 +- packages/app/src/subagents/index.ts | 4 +- .../app/src/subagents/opens-in-sheet.test.tsx | 12 +- 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 | 13 +- .../app/src/subagents/use-archive-subagent.ts | 12 +- .../app/src/subagents/use-detach-subagent.ts | 12 +- .../app/src/types/host-connection.test.ts | 51 +++ packages/app/src/types/host-connection.ts | 19 +- 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 + .../src/server/agent/activity-curator.ts | 20 +- .../src/server/agent/agent-projections.ts | 61 +++ .../server/src/server/agent/agent-prompt.ts | 109 ++++- .../src/server/agent/host-operations/index.ts | 393 ++++++++++++++++++ .../src/server/agent/tools/paseo-tools.ts | 241 +++++++---- packages/server/src/server/bootstrap.ts | 5 + .../server/connected-hosts/hosts.e2e.test.ts | 181 +++++++- .../server/src/server/websocket-server.ts | 5 + public-docs/sdk/reference.md | 3 + 41 files changed, 1919 insertions(+), 191 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..1844456007 100644 --- a/docs/connected-hosts.md +++ b/docs/connected-hosts.md @@ -29,6 +29,26 @@ 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`, +`create_agent`, `send_agent_prompt`, `get_agent_status`, `get_agent_activity`, and `list_agents` +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. Prompting a remote child acknowledges acceptance and uses the existing idle wait for +finish notifications. Remote agent lists have no caller-directory default; pass `cwd` to filter them. + +The parent’s Subagents track shows qualified children across connected hosts. Opening a child uses +the ordinary conversation on the destination's session. The runtime selects among that host's +connection candidates, including direct connections and routes discovered through another daemon. +Adding a direct connection to a discovered host adds another candidate. If no candidate connects, +the destination workspace keeps its existing disconnected or reconnecting behavior. + +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..30b19e5099 100644 --- a/packages/app/e2e/browser/subagent-origin-pane.spec.ts +++ b/packages/app/e2e/browser/subagent-origin-pane.spec.ts @@ -1,10 +1,19 @@ +import { buildSeededHost } from "../support/helpers/daemon-registry"; +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, + expectManagedSubagentArchived, + openSubagentsTrack, + seedParentWithSubagent, +} from "../support/helpers/subagents"; + +test.use({ e2eInjectPaseoTools: true }); const SETTINGS_KEY = "@paseo:app-settings"; @@ -147,3 +156,189 @@ 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(); + } +}); + +async function openFleetParentWithDirectB( + page: Page, + fleet: Awaited>, +) { + await page.goto("/", { waitUntil: "commit" }); + const host = buildSeededHost({ + serverId: fleet.remote.serverId, + endpoint: `127.0.0.1:${fleet.destination.port}`, + label: "Fleet B", + nowIso: new Date().toISOString(), + }); + await page.evaluate((extraHost) => { + localStorage.setItem("@paseo:e2e-extra-hosts", JSON.stringify([extraHost])); + }, host); + await openAgentRoute(page, { + workspaceId: fleet.workspace.workspaceId, + agentId: fleet.parent.id, + }); + await openSubagentsTrack(page); + await expect(page.getByTestId(`subagents-track-row-${fleet.remote.agentId}`)).toContainText( + "MCP child on B", + ); +} + +async function openFleetChild( + page: Page, + fleet: Awaited>, +) { + await page.getByTestId(`subagents-track-row-${fleet.remote.agentId}`).click(); + await expect(page).toHaveURL( + (url) => url.pathname === `/h/${fleet.remote.serverId}/workspace/${fleet.remote.workspaceId}`, + ); +} + +test("a directly connected B opens and archives its MCP child from A's track", async ({ page }) => { + const fleet = await delegateFleetThroughMcp(); + try { + await openFleetParentWithDirectB(page, fleet); + await openFleetChild(page, fleet); + await expect(page.getByRole("button", { name: "MCP child on B", exact: true })).toBeVisible(); + await submitMessage(page, "DIRECT_B_TARGET"); + await expect(page.getByText("DIRECT_B_TARGET", { exact: true })).toBeVisible(); + expect((await fleet.a.fetchAgent(fleet.remote.agentId))?.agent.lastUserMessageAt).toBeNull(); + await openFleetParentWithDirectB(page, fleet); + await page.getByTestId(`subagents-track-row-${fleet.remote.agentId}`).hover(); + page.once("dialog", (dialog) => { + void dialog.accept(); + }); + await page.getByRole("button", { name: "Archive MCP child on B", exact: true }).click(); + await expectManagedSubagentArchived({ client: fleet.b }, fleet.remote.agentId); + await expect(page.getByTestId(`subagents-track-row-${fleet.remote.agentId}`)).toHaveCount(0); + expect((await fleet.a.fetchAgent(fleet.remote.agentId))?.agent.archivedAt).toBeNull(); + await expect(page.getByText(/already connected through another route/)).toHaveCount(0); + } finally { + await fleet.close(); + } +}); + +test("an unavailable B child opens its cached workspace with existing reconnecting behavior", async ({ + page, +}) => { + const fleet = await delegateFleetThroughMcp(); + try { + await openFleetParentWithDirectB(page, fleet); + await fleet.destination.close(); + await expect + .poll(() => + page.evaluate((serverId) => { + const runtime = Reflect.get(globalThis, "__paseoHostRuntimeStore") as { + getSnapshot(id: string): { serverId: string; connectionStatus: string } | null; + }; + return runtime.getSnapshot(serverId); + }, fleet.remote.serverId), + ) + .toMatchObject({ + serverId: fleet.remote.serverId, + connectionStatus: expect.stringMatching(/^(offline|connecting|error)$/), + }); + await openFleetChild(page, fleet); + await expect(page.getByTestId("workspace-header-title").filter({ visible: true })).toHaveText( + "B workspace", + ); + await expect(page.getByRole("button", { name: "MCP child on B", exact: true })).toBeVisible(); + await expect(page.getByText(/already connected through another route/)).toHaveCount(0); + } finally { + await fleet.close(); + } +}); diff --git a/packages/app/e2e/support/helpers/subagents.ts b/packages/app/e2e/support/helpers/subagents.ts index f3a5ca7c71..3354bf3dd8 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"; @@ -166,7 +170,7 @@ export async function expectSubagentRowGone(page: Page, childId: string): Promis } export async function expectManagedSubagentUnarchived( - workspace: Pick, + workspace: { client: Pick }, childId: string, ): Promise { await expect(workspace.client.fetchAgent({ agentId: childId })).resolves.toMatchObject({ @@ -175,7 +179,7 @@ export async function expectManagedSubagentUnarchived( } export async function expectManagedSubagentArchived( - workspace: Pick, + workspace: { client: Pick }, childId: string, ): Promise { await expect @@ -320,3 +324,150 @@ 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), + workspaceId, + 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, + destination, + 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..3fc42030a7 100644 --- a/packages/app/src/panels/agent-tracks.tsx +++ b/packages/app/src/panels/agent-tracks.tsx @@ -1,3 +1,4 @@ +import type { AgentAddress } from "@getpaseo/protocol/addresses"; import { memo, useCallback, type ReactElement } from "react"; import { WorkspaceDiffStatPill } from "@/composer/diff-stat-pill"; import { useWorkspaceHasDiffStat } from "@/composer/workspace-diff-stat"; @@ -56,8 +57,8 @@ export const AgentTracks = memo(function AgentTracks({ const canDetachSubagents = useSessionStore( (state) => state.sessions[serverId]?.serverInfo?.features?.agentDetach === true, ); - const archiveSubagent = useArchiveSubagent({ serverId }); - const detachSubagent = useDetachSubagent({ serverId }); + const archiveSubagent = useArchiveSubagent(); + const detachSubagent = useDetachSubagent(); const openSubagentTarget = useCallback( (target: WorkspaceTabTarget) => { openPreferredWorkspaceTarget({ @@ -71,12 +72,15 @@ export const AgentTracks = memo(function AgentTracks({ [openInSidePane, tabId, workspaceKey], ); const handleOpenSubagent = useCallback( - (subagentId: string) => { - const session = useSessionStore.getState().sessions[serverId]; + ({ serverId: targetServerId, agentId: subagentId }: AgentAddress) => { + 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 }); diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index 61285eb2c6..1425dc3560 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -11,6 +11,7 @@ import type { FetchAgentsOptions, } from "@getpaseo/client/internal/daemon-client"; import { DaemonAuthenticationError } from "@getpaseo/client/internal/daemon-client"; +import type { HostDirectory } from "@getpaseo/protocol/hosts"; import type { ConnectionOffer } from "@getpaseo/protocol/connection-offer"; import type { SessionOutboundMessage } from "@getpaseo/protocol/messages"; import type { AgentPermissionRequest } from "@getpaseo/protocol/agent-types"; @@ -157,10 +158,24 @@ class FakeDaemonClient { } } + get isConnected(): boolean { + return this.state.status === "connected"; + } + getConnectionState(): ConnectionState { return this.state; } + public connectedHosts = false; + public hostDirectory: Promise = Promise.resolve({ + self: { serverId: "srv_test", label: "test" }, + hosts: [], + }); + + observeHosts() { + return subscriptionFixture(this.hostDirectory, () => () => {}); + } + public ownedSubscriptions = true; private daemonVersion: string | null = "0.8.0"; @@ -171,7 +186,10 @@ class FakeDaemonClient { serverId: "srv_test", hostname: "test", version: this.daemonVersion, - features: { ownedSubscriptions: this.ownedSubscriptions }, + features: { + ownedSubscriptions: this.ownedSubscriptions, + connectedHosts: this.connectedHosts, + }, }; } @@ -1658,6 +1676,103 @@ describe("HostRuntimeController", () => { }); describe("HostRuntimeStore", () => { + it.each([false, true])( + "keeps B and C distinct with routed and direct candidates (B direct first: %s)", + async (directFirst) => { + const directory = new Deferred(); + const origin = new FakeDaemonClient(); + origin.setConnectionState({ status: "connected" }); + origin.connectedHosts = true; + origin.hostDirectory = directory.promise; + const store = new HostRuntimeStore({ + storage: createMemoryHostRuntimeStorage(), + deps: { + createClient: ({ host }) => + (host.serverId === "srv_a" + ? origin + : new FakeDaemonClient()) as unknown as DaemonClient, + connectToDaemon: async ({ host }) => ({ + client: (host.serverId === "srv_a" + ? origin + : makeConnectedProbeClient(5)) as unknown as DaemonClient, + serverId: host.serverId, + hostname: host.label, + }), + getClientId: async () => "cid_fleet_candidates", + }, + }); + const addDirectB = () => + store.upsertDirectConnection({ + serverId: "srv_b", + endpoint: "b:6767", + label: "My B", + }); + try { + if (directFirst) await addDirectB(); + await store.upsertDirectConnection({ + serverId: "srv_c", + endpoint: "c:6767", + label: "My C", + }); + await store.upsertDirectConnection({ serverId: "srv_a", endpoint: "a:6767" }); + await waitForHostOnline(store, "srv_a"); + directory.resolve({ + self: { serverId: "srv_a", label: "A" }, + hosts: ["b", "c"].map((target) => ({ + registrationId: `reg_${target}`, + serverId: `srv_${target}`, + label: `Fleet ${target.toUpperCase()}`, + source: { id: "user", key: target }, + metadata: {}, + enabled: true, + status: "online" as const, + targetDescription: `${target}:6767`, + lastOnlineAt: null, + nextRetryAt: null, + error: null, + })), + }); + await vi.waitFor(() => + expect(store.getHosts()).toContainEqual( + expect.objectContaining({ + serverId: "srv_b", + connections: expect.arrayContaining([ + { id: "via:srv_a", type: "routed", originServerId: "srv_a" }, + ]), + }), + ), + ); + await waitForHostOnline(store, "srv_b"); + if (!directFirst) await addDirectB(); + const targets = store.getHosts().filter((host) => host.serverId === "srv_b"); + expect(targets).toHaveLength(1); + expect(targets[0]?.connections).toEqual( + expect.arrayContaining([ + { id: "via:srv_a", type: "routed", originServerId: "srv_a" }, + { id: "direct:b:6767", type: "directTcp", endpoint: "b:6767", useTls: false }, + ]), + ); + expect(targets[0]?.connections).toHaveLength(2); + expect(targets[0]?.label).toBe(directFirst ? "My B" : "Fleet B"); + expect(store.getSnapshot("srv_b")?.connectionStatus).toBe("online"); + expect(store.getHosts()).toHaveLength(3); + const targetC = store.getHosts().find((host) => host.serverId === "srv_c"); + expect(targetC).toMatchObject({ + serverId: "srv_c", + label: "My C", + connections: [ + { id: "direct:c:6767", type: "directTcp", endpoint: "c:6767", useTls: false }, + { id: "via:srv_a", type: "routed", originServerId: "srv_a" }, + ], + }); + await waitForHostOnline(store, "srv_c"); + expect(store.getSnapshot("srv_c")?.connectionStatus).toBe("online"); + } finally { + store.syncHosts([]); + } + }, + ); + it.each(["active", "inactive", "background"] as const)( "keeps reconnect enabled through inactive/background and resumes immediately (mounted %s)", async (currentState) => { diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index 85d077ee0a..152e3331e5 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -1,4 +1,7 @@ import { watchdog } from "@/watchdog"; +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"; @@ -114,6 +117,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" } @@ -235,6 +239,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", @@ -519,7 +529,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 @@ -550,6 +562,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, @@ -598,15 +619,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: createPerformanceTrace({ host: host.serverId, connection: connection.id }), - }), + }); + }, getClientId: () => getOrCreateClientId(), mountClientHandlers: ({ client, host }) => { const unmountServerData = mountServerDataPushRouter({ @@ -1495,6 +1541,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>(); @@ -1525,7 +1576,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; @@ -2265,6 +2316,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); @@ -2321,6 +2377,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 @@ -2339,6 +2396,56 @@ export class HostRuntimeStore { } } + 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; + const next = upsertHostConnectionInProfiles({ + profiles: this.hosts, + serverId: target.serverId, + label: target.label ?? undefined, + connection: { + id: `via:${serverId}`, + type: "routed", + originServerId: serverId, + }, + }); + if (next !== this.hosts) { + this.setHostsAndSync(next); + 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/stores/workspace-subagents-integration.test.ts b/packages/app/src/stores/workspace-subagents-integration.test.ts index 5f04a39bd9..ff2c8b2f4b 100644 --- a/packages/app/src/stores/workspace-subagents-integration.test.ts +++ b/packages/app/src/stores/workspace-subagents-integration.test.ts @@ -1,4 +1,5 @@ import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; +import { parentAddressLabels } from "@getpaseo/protocol/agent-labels"; import { afterEach, describe, expect, it, vi } from "vitest"; import { buildWorkspaceTabSnapshot, @@ -10,6 +11,7 @@ import { selectSubagentsForParent } from "@/subagents/select"; import { findPaneById, findPaneContainingTab } from "./workspace-layout-actions"; import { useWorkspaceLayoutStore } from "./workspace-layout-store"; import { useSessionStore, type Agent } from "./session-store"; +import { normalizeAgentSnapshot, projectAgentSnapshot } from "@/utils/agent-snapshots"; vi.mock("@react-native-async-storage/async-storage", () => { const storage = new Map(); @@ -73,7 +75,11 @@ const AGENT_DEFAULTS: Agent = { }; function makeAgent(input: Partial & Pick): Agent { - return { ...AGENT_DEFAULTS, ...input }; + const agent = { ...AGENT_DEFAULTS, ...input }; + return { + ...normalizeAgentSnapshot(projectAgentSnapshot(agent), SERVER_ID), + projectPlacement: agent.projectPlacement, + }; } function initializeAgents(agents: Agent[]): void { @@ -142,7 +148,7 @@ describe("workspace subagents integration", () => { const child = makeAgent({ id: "child-agent", - parentAgentId: "parent-agent", + labels: parentAddressLabels({ serverId: SERVER_ID, agentId: "parent-agent" }), title: "Child agent", }); const parent = makeAgent({ @@ -186,7 +192,7 @@ describe("workspace subagents integration", () => { }); const child = makeAgent({ id: "child-agent", - parentAgentId: "parent-agent", + labels: parentAddressLabels({ serverId: SERVER_ID, agentId: "parent-agent" }), title: "Child agent", }); @@ -205,7 +211,7 @@ describe("workspace subagents integration", () => { ).map((row) => row.id), ).toEqual(["child-agent"]); - appendAgent({ ...child, parentAgentId: null, labels: {} }); + appendAgent(makeAgent({ ...child, labels: {} })); reconcileWorkspaceTabs(workspaceKey!, deriveVisibilityFromSession()); expect(getWorkspaceAgentTabIds(workspaceKey!)).toEqual([ @@ -238,7 +244,7 @@ describe("workspace subagents integration", () => { }); const child = makeAgent({ id: "child-agent", - parentAgentId: parent.id, + labels: parentAddressLabels({ serverId: SERVER_ID, agentId: parent.id }), title: "Cross-workspace child", }); @@ -271,7 +277,7 @@ describe("workspace subagents integration", () => { }); const child = makeAgent({ id: "child-agent", - parentAgentId: parent.id, + labels: parentAddressLabels({ serverId: SERVER_ID, agentId: parent.id }), title: "Child agent", }); 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/index.ts b/packages/app/src/subagents/index.ts index 9c6227a34f..6c539931e1 100644 --- a/packages/app/src/subagents/index.ts +++ b/packages/app/src/subagents/index.ts @@ -1,7 +1,7 @@ export type { SubagentRow } from "./select"; export { selectSubagentsForParent, useSubagentsForParent } from "./select"; -export { useArchiveSubagent, type UseArchiveSubagentInput } from "./use-archive-subagent"; -export { useDetachSubagent, type UseDetachSubagentInput } from "./use-detach-subagent"; +export { useArchiveSubagent } from "./use-archive-subagent"; +export { useDetachSubagent } from "./use-detach-subagent"; export { resolveCloseAgentTabPolicy, type CloseAgentTabPolicy } from "./close-tab-policy"; export { isWorkspaceRootAgent } from "./workspace-root-policy"; export { diff --git a/packages/app/src/subagents/opens-in-sheet.test.tsx b/packages/app/src/subagents/opens-in-sheet.test.tsx index 0792091135..c9b915f787 100644 --- a/packages/app/src/subagents/opens-in-sheet.test.tsx +++ b/packages/app/src/subagents/opens-in-sheet.test.tsx @@ -2,9 +2,11 @@ * @vitest-environment jsdom */ import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; +import { parentAddressLabels } from "@getpaseo/protocol/agent-labels"; import { act, renderHook } from "@testing-library/react"; import { afterEach, describe, expect, it, vi } from "vitest"; import { useSessionStore, type Agent } from "@/stores/session-store"; +import { normalizeAgentSnapshot, projectAgentSnapshot } from "@/utils/agent-snapshots"; import { useWorkspaceLayoutStore } from "@/stores/workspace-layout-store"; import { buildWorkspaceTabPersistenceKey } from "@/workspace-tabs/model"; import { useSubagentOpensInSheet } from "./opens-in-sheet"; @@ -77,7 +79,15 @@ if (!workspaceKey) throw new Error("workspace key"); function seedParentWithChild(): SubagentRow { const parent: Agent = { ...AGENT_DEFAULTS, id: "parent", title: "Parent" }; - const child: Agent = { ...AGENT_DEFAULTS, id: "child", title: "Child", parentAgentId: "parent" }; + const child = normalizeAgentSnapshot( + projectAgentSnapshot({ + ...AGENT_DEFAULTS, + id: "child", + title: "Child", + labels: parentAddressLabels({ serverId: SERVER_ID, agentId: parent.id }), + }), + SERVER_ID, + ); useSessionStore.getState().initializeSession(SERVER_ID, null as unknown as DaemonClient); useSessionStore.getState().setAgents( SERVER_ID, 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..0152dc9ba3 100644 --- a/packages/app/src/subagents/use-archive-finished.ts +++ b/packages/app/src/subagents/use-archive-finished.ts @@ -1,3 +1,4 @@ +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 +33,16 @@ 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: archiveAgent, 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..e8189c1830 100644 --- a/packages/app/src/subagents/use-archive-subagent.ts +++ b/packages/app/src/subagents/use-archive-subagent.ts @@ -1,3 +1,4 @@ +import type { AgentAddress } from "@getpaseo/protocol/addresses"; import { useCallback } from "react"; import { useToast } from "@/contexts/toast-context"; import { useArchiveAgent } from "@/hooks/use-archive-agent"; @@ -13,17 +14,12 @@ export type { ResolveArchiveSubagentDialogInput, } from "./archive-subagent"; -export interface UseArchiveSubagentInput { - serverId: string; -} - -export function useArchiveSubagent(input: UseArchiveSubagentInput): (subagentId: string) => void { +export function useArchiveSubagent(): (address: AgentAddress) => void { const { archiveAgent } = useArchiveAgent(); - const { serverId } = input; const toast = useToast(); return useCallback( - (subagentId: string) => { + ({ serverId, agentId: subagentId }: AgentAddress) => { void requestArchiveSubagent( { serverId, subagentId }, { @@ -37,6 +33,6 @@ export function useArchiveSubagent(input: UseArchiveSubagentInput): (subagentId: }, ); }, - [archiveAgent, serverId, toast], + [archiveAgent, toast], ); } diff --git a/packages/app/src/subagents/use-detach-subagent.ts b/packages/app/src/subagents/use-detach-subagent.ts index 36b2f2a560..120aedb872 100644 --- a/packages/app/src/subagents/use-detach-subagent.ts +++ b/packages/app/src/subagents/use-detach-subagent.ts @@ -1,3 +1,4 @@ +import type { AgentAddress } from "@getpaseo/protocol/addresses"; import { useCallback } from "react"; import { useToast } from "@/contexts/toast-context"; import { i18n } from "@/i18n/i18next"; @@ -14,16 +15,11 @@ export type { ResolveDetachSubagentDialogInput, } from "./detach-subagent"; -export interface UseDetachSubagentInput { - serverId: string; -} - -export function useDetachSubagent(input: UseDetachSubagentInput): (subagentId: string) => void { - const { serverId } = input; +export function useDetachSubagent(): (address: AgentAddress) => void { const toast = useToast(); return useCallback( - (subagentId: string) => { + ({ serverId, agentId: subagentId }: AgentAddress) => { void requestDetachSubagent( { serverId, subagentId }, { @@ -46,6 +42,6 @@ export function useDetachSubagent(input: UseDetachSubagentInput): (subagentId: s }, ); }, - [serverId, toast], + [toast], ); } diff --git a/packages/app/src/types/host-connection.test.ts b/packages/app/src/types/host-connection.test.ts index a476c32089..0f0590bb9e 100644 --- a/packages/app/src/types/host-connection.test.ts +++ b/packages/app/src/types/host-connection.test.ts @@ -205,6 +205,57 @@ describe("upsertHostConnectionInProfiles", () => { path: "/tmp/paseo.sock", }; + it.each([false, true])( + "keeps destinations distinct when a persisted route shares an origin (C already saved: %s)", + (savedC) => { + const route: HostConnection = { id: "via:srv_a", type: "routed", originServerId: "srv_a" }; + const directB: HostConnection = { + id: "direct:b:6767", + type: "directTcp", + endpoint: "b:6767", + }; + const directC: HostConnection = { + id: "direct:c:6767", + type: "directTcp", + endpoint: "c:6767", + }; + const b: HostProfile = { + ...makeHost("srv_b"), + label: "My B", + connections: [route, directB], + preferredConnectionId: route.id, + }; + const c: HostProfile = { + ...makeHost("srv_c"), + label: "My C", + connections: [directC], + preferredConnectionId: directC.id, + }; + const profiles = upsertHostConnectionInProfiles({ + profiles: savedC ? [b, c] : [b], + serverId: "srv_c", + label: "Fleet C", + connection: route, + }); + expect(profiles).toHaveLength(2); + expect(profiles[0]).toEqual(b); + expect(profiles[1]).toMatchObject({ + serverId: "srv_c", + label: savedC ? "My C" : "Fleet C", + connections: savedC ? [directC, route] : [route], + preferredConnectionId: savedC ? directC.id : route.id, + }); + expect( + upsertHostConnectionInProfiles({ + profiles, + serverId: "srv_b", + label: "Fleet B", + connection: route, + }), + ).toBe(profiles); + }, + ); + it("gives a newly discovered host the default appearance", () => { const [profile] = upsertHostConnectionInProfiles({ profiles: [], diff --git a/packages/app/src/types/host-connection.ts b/packages/app/src/types/host-connection.ts index 0962c02c7c..fb3150a168 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); } @@ -236,9 +245,11 @@ export function upsertHostConnectionInProfiles(input: { const matchingIndexes = existing.reduce((matches, daemon, index) => { if ( daemon.serverId === serverId || - daemon.connections.some((existingConnection) => - hostConnectionEquals(existingConnection, normalizedConnection), - ) + // A routed candidate names its origin; its destination belongs to the host profile. + (normalizedConnection.type !== "routed" && + daemon.connections.some((existingConnection) => + hostConnectionEquals(existingConnection, normalizedConnection), + )) ) { matches.push(index); } @@ -399,6 +410,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 +458,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..e2eb008093 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", ); } + +// Absent parent server label means the same host. +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/activity-curator.ts b/packages/server/src/server/agent/activity-curator.ts index ec5bceb1cb..488867c657 100644 --- a/packages/server/src/server/agent/activity-curator.ts +++ b/packages/server/src/server/agent/activity-curator.ts @@ -5,7 +5,7 @@ import type { AgentTimelineRow } from "./agent-timeline-store-types.js"; import { isLikelyExternalToolName } from "@getpaseo/protocol/tool-name-normalization"; import { buildAgentMessageDisplay } from "@getpaseo/protocol/paseo-tool-presentation/index"; import { buildToolCallDisplayModel } from "@getpaseo/protocol/tool-call-display"; -import { projectTimelineRows } from "./timeline-projection.js"; +import { projectTimelineRows, selectItemsByProjectedLimit } from "./timeline-projection.js"; const DEFAULT_MAX_ITEMS = 0; const MAX_TOOL_INPUT_CHARS = 400; @@ -210,6 +210,24 @@ export function curateAgentActivity( : "No activity to display."; } +export function summarizeAgentActivity(timeline: AgentTimelineItem[], limit?: number) { + const selection = selectItemsByProjectedLimit({ + items: timeline, + direction: "tail", + limit: limit ?? 0, + }); + const { totalProjected, shownProjected } = selection; + const noun = totalProjected === 1 ? "activity" : "activities"; + const countHeader = + limit && shownProjected < totalProjected + ? `Showing ${shownProjected} of ${totalProjected} ${noun} (limited to ${limit})` + : `Showing all ${totalProjected} ${noun}`; + return { + updateCount: timeline.length, + content: `${countHeader}\n\n${curateAgentActivity(selection.items)}`, + }; +} + interface ForkCursorBoundary { timelineEpoch: string; cursor: { epoch: string; seq: number }; diff --git a/packages/server/src/server/agent/agent-projections.ts b/packages/server/src/server/agent/agent-projections.ts index dbad9586db..7d154e9c14 100644 --- a/packages/server/src/server/agent/agent-projections.ts +++ b/packages/server/src/server/agent/agent-projections.ts @@ -1,3 +1,4 @@ +import { isSameOrDescendantPath } from "../path-utils.js"; import type { AgentListItemPayload, AgentSnapshotPayload, @@ -286,6 +287,66 @@ export function toAgentListItemPayload(agent: AgentSnapshotPayload): AgentListIt }; } +function parseTimestamp(value: string | null | undefined): number { + if (!value) { + return 0; + } + const parsed = Date.parse(value); + return Number.isNaN(parsed) ? 0 : parsed; +} + +function resolveAgentListActivityTime(agent: AgentListItemPayload): number { + return Math.max( + parseTimestamp(agent.updatedAt), + parseTimestamp(agent.lastUserMessageAt), + parseTimestamp(agent.attentionTimestamp), + parseTimestamp(agent.archivedAt), + parseTimestamp(agent.createdAt), + ); +} + +function compareAgentListItems(a: AgentListItemPayload, b: AgentListItemPayload): number { + const attentionDelta = + Number(b.requiresAttention ?? false) - Number(a.requiresAttention ?? false); + if (attentionDelta !== 0) { + return attentionDelta; + } + + const statusOrder = { + running: 0, + initializing: 1, + idle: 2, + error: 3, + closed: 4, + } as Record; + const statusDelta = (statusOrder[a.status] ?? 999) - (statusOrder[b.status] ?? 999); + if (statusDelta !== 0) { + return statusDelta; + } + + return resolveAgentListActivityTime(b) - resolveAgentListActivityTime(a); +} + +export function selectAgentListItems( + agents: readonly AgentSnapshotPayload[], + options: { + cwd?: string; + statuses?: AgentSnapshotPayload["status"][]; + sinceHours?: number; + limit?: number; + } = {}, +): AgentListItemPayload[] { + const statusFilter = options.statuses?.length ? new Set(options.statuses) : null; + const sinceMs = Date.now() - (options.sinceHours ?? 48) * 60 * 60 * 1000; + return agents + .map(toAgentListItemPayload) + .filter((agent) => !options.cwd || isSameOrDescendantPath(options.cwd, agent.cwd)) + .filter((agent) => !statusFilter || statusFilter.has(agent.status)) + .filter((agent) => !agent.archivedAt || resolveAgentListActivityTime(agent) >= sinceMs) + .sort(compareAgentListItems) + .slice(0, options.limit ?? 50); +} + export function toRecentProviderSessionDescriptorPayload( session: ImportableProviderSession & { provider: string }, options: RecentProviderSessionProjectionOptions, 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..b69f0775a9 --- /dev/null +++ b/packages/server/src/server/agent/host-operations/index.ts @@ -0,0 +1,393 @@ +import { parentAddressLabels } from "@getpaseo/protocol/agent-labels"; +import type { ActiveTurnBehavior, 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 { selectAgentListItems } from "../agent-projections.js"; +import { summarizeAgentActivity } from "../activity-curator.js"; +import { expandUserPath } from "../../path-utils.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; + sendPrompt: HostOperation; + getStatus: HostOperation; + getActivity: HostOperation; + listAgents: 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 sendPrompt(args) { + const agentId = args.agentId as string; + if (args.sessionMode !== undefined) + await client.setAgentMode(agentId, args.sessionMode as string); + await client.sendMessage(agentId, args.prompt as string, { + sourceAgentId: input.callerAgentId, + activeTurnBehavior: + (args.activeTurnBehavior as ActiveTurnBehavior | undefined) ?? + (input.callerAgentId ? "steer" : "interrupt"), + }); + const agent = (await client.fetchAgent(agentId))?.agent; + if (!agent) throw new Error(`Agent ${agentId} not found`); + return result({ + success: true, + status: agent.status, + guidance: + input.callerAgentId && args.notifyOnFinish !== false + ? "You will get notified when the prompted agent finishes, errors, or needs permission. Do not poll for status; continue with other work until the notification arrives." + : "Dispatch acknowledged. Observe completion using get_agent_status/get_agent_activity or SDK/RPC subscriptions and wait APIs.", + }); + }, + async getStatus(args) { + const agent = (await client.fetchAgent(args.agentId as string))?.agent; + if (!agent) throw new Error(`Agent ${args.agentId} not found`); + return result({ status: agent.status, snapshot: agent }); + }, + async getActivity(args) { + const agentId = args.agentId as string; + const limit = args.limit as number | undefined; + const timeline = await client.fetchAgentTimeline(agentId, { + direction: "tail", + limit: limit ?? 0, + }); + const agent = (await client.fetchAgent(agentId))?.agent; + if (!agent) throw new Error(`Agent ${agentId} not found`); + return result({ + agentId, + ...summarizeAgentActivity( + timeline.entries.map((entry) => entry.item), + limit, + ), + currentModeId: agent.currentModeId, + }); + }, + async listAgents(args) { + const { entries } = await client.fetchAgents({ + includeBackground: args.includeBackground === true, + filter: { includeArchived: args.includeArchived === true }, + }); + return result({ + agents: selectAgentListItems( + entries.map((entry) => entry.agent), + { + cwd: (args.cwd as string | undefined)?.trim() + ? expandUserPath(args.cwd as string) + : undefined, + statuses: args.statuses as AgentSnapshotPayload["status"][] | undefined, + sinceHours: args.sinceHours as number | undefined, + limit: args.limit as number | undefined, + }, + ), + }); + }, + 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 a5117b36b1..97ded06e69 100644 --- a/packages/server/src/server/agent/tools/paseo-tools.ts +++ b/packages/server/src/server/agent/tools/paseo-tools.ts @@ -1,3 +1,10 @@ +import { + createHostResolver, + waitForPermissionChange, + type HostOperation, + type HostOperations, + type ToolHosts, +} from "../host-operations/index.js"; import { AgentDirectory } from "../directory.js"; import { stat } from "node:fs/promises"; import { z } from "zod"; @@ -16,14 +23,12 @@ import { AgentSnapshotPayloadSchema, WorkspaceScriptPayloadSchema, } from "../../messages.js"; -import type { AgentListItemPayload } from "../../messages.js"; import { buildStoredAgentPayload, - toAgentListItemPayload, + selectAgentListItems, toAgentPayload, } from "../agent-projections.js"; -import { curateAgentActivity } from "../activity-curator.js"; -import { selectItemsByProjectedLimit } from "../timeline-projection.js"; +import { summarizeAgentActivity } from "../activity-curator.js"; import type { AgentStorage } from "../agent-storage.js"; import { AgentNotFoundError, ensureAgentLoaded } from "../agent-loading.js"; import { isStoredAgentProviderAvailable } from "../../persistence-hooks.js"; @@ -72,6 +77,7 @@ import { type PromptDispatchDisposition, sendPromptToAgent, setupFinishNotification, + setupFinishWait, waitForAgentRunStartWithTimeout, } from "../agent-prompt.js"; import { respondToAgentPermission } from "../permission-response.js"; @@ -110,6 +116,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,22 +173,8 @@ export interface PaseoToolHostDependencies { logger: Logger; } -function parseTimestamp(value: string | null | undefined): number { - if (!value) { - return 0; - } - const parsed = Date.parse(value); - return Number.isNaN(parsed) ? 0 : parsed; -} - -function resolveAgentListActivityTime(agent: AgentListItemPayload): number { - return Math.max( - parseTimestamp(agent.updatedAt), - parseTimestamp(agent.lastUserMessageAt), - parseTimestamp(agent.attentionTimestamp), - parseTimestamp(agent.archivedAt), - parseTimestamp(agent.createdAt), - ); +function settledStatus(status: string): "error" | "idle" { + return status === "error" ? "error" : "idle"; } interface ProviderSummary { @@ -328,28 +322,6 @@ function toProviderSummary(entry: { }; } -function compareAgentListItems(a: AgentListItemPayload, b: AgentListItemPayload): number { - const attentionDelta = - Number(b.requiresAttention ?? false) - Number(a.requiresAttention ?? false); - if (attentionDelta !== 0) { - return attentionDelta; - } - - const statusOrder = { - running: 0, - initializing: 1, - idle: 2, - error: 3, - closed: 4, - } as Record; - const statusDelta = (statusOrder[a.status] ?? 999) - (statusOrder[b.status] ?? 999); - if (statusDelta !== 0) { - return statusDelta; - } - - return resolveAgentListActivityTime(b) - resolveAgentListActivityTime(a); -} - function resolveScheduleProviderAndModel(params: { provider?: string; defaultProvider: AgentProvider; @@ -685,7 +657,66 @@ 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"], + ["send_agent_prompt", "sendPrompt"], + ["get_agent_status", "getStatus"], + ["get_agent_activity", "getActivity"], + ["list_agents", "listAgents"], + ] as const satisfies ReadonlyArray); + 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), + sendPrompt: (args, context) => localHostOperations.get("send_agent_prompt")!(args, context), + getStatus: (args, context) => localHostOperations.get("get_agent_status")!(args, context), + getActivity: (args, context) => localHostOperations.get("get_agent_activity")!(args, context), + listAgents: (args, context) => localHostOperations.get("list_agents")!(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 +726,76 @@ function createToolCatalog( if (!isPaseoToolEnabled(options.paseoToolPolicy, name)) { return; } + const operation = hostTools.get(name as Parameters[0]); + 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), + ); + let childAgentId: string | undefined; + if (name === "create_agent") { + childAgentId = (response.structuredContent as { agentId: string }).agentId; + } else if (name === "send_agent_prompt" && destination !== options.serverId) { + childAgentId = input.agentId as string; + } + if (childAgentId && callerAgentId && input.notifyOnFinish !== false) { + setupFinishWait({ + agentManager, + agentStorage, + originServerId: options.serverId, + childServerId: destination, + childAgentId, + callerAgentId, + logger: childLogger, + wait: (signal) => + hostResolver.use(destination, (host) => host.wait(childAgentId, signal)), + afterPermission: (requestId, signal) => + hostResolver.use(destination, (host) => + waitForPermissionChange(host, childAgentId, 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, @@ -1259,6 +1360,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", { @@ -1520,10 +1634,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, @@ -1849,6 +1968,7 @@ function createToolCatalog( agentManager, agentStorage, childAgentId: agentId, + childServerId: options.serverId, callerAgentId, waitForTurnStart: false, dispatchResult, @@ -1869,6 +1989,7 @@ function createToolCatalog( agentManager, agentStorage, childAgentId: agentId, + childServerId: options.serverId, callerAgentId, waitForTurnStart: true, logger: childLogger, @@ -2004,8 +2125,6 @@ function createToolCatalog( }) => { const callerCwd = callerAgentId ? resolveCallerAgent()?.cwd : undefined; const requestedCwd = cwd?.trim() ? expandUserPath(cwd) : callerCwd; - const statusFilter = statuses && statuses.length > 0 ? new Set(statuses) : null; - const sinceMs = Date.now() - sinceHours * 60 * 60 * 1000; if (!options.workspaceRegistry) throw new Error("Workspace registry is not configured"); const directory = new AgentDirectory({ manager: agentManager, @@ -2025,13 +2144,10 @@ function createToolCatalog( new Set(providerSnapshotManager.listRegisteredProviderIds()), ), }); - const agents = (await directory.list({ includeBackground, includeArchived })) - .map(toAgentListItemPayload) - .filter((agent) => !requestedCwd || isSameOrDescendantPath(requestedCwd, agent.cwd)) - .filter((agent) => !statusFilter || statusFilter.has(agent.status)) - .filter((agent) => !agent.archivedAt || resolveAgentListActivityTime(agent) >= sinceMs) - .sort(compareAgentListItems) - .slice(0, limit); + const agents = selectAgentListItems( + await directory.list({ includeBackground, includeArchived }), + { cwd: requestedCwd, statuses, sinceHours, limit }, + ); return { content: [], @@ -3028,29 +3144,12 @@ function createToolCatalog( const timeline = agentManager.getTimeline(agentId); const snapshot = agentManager.getAgent(agentId); - const selection = selectItemsByProjectedLimit({ - items: timeline, - direction: "tail", - limit: limit ?? 0, - }); - const curatedContent = curateAgentActivity(selection.items); - const { totalProjected, shownProjected } = selection; - - const noun = totalProjected === 1 ? "activity" : "activities"; - const countHeader = - limit && shownProjected < totalProjected - ? `Showing ${shownProjected} of ${totalProjected} ${noun} (limited to ${limit})` - : `Showing all ${totalProjected} ${noun}`; - - const contentWithCount = `${countHeader}\n\n${curatedContent}`; - return { content: [], structuredContent: ensureValidJson({ agentId, - updateCount: timeline.length, + ...summarizeAgentActivity(timeline, limit), currentModeId: snapshot?.currentModeId ?? null, - content: contentWithCount, }), }; }, 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..0efd8d86d6 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,174 @@ 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, client: aClient } = 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}`); + + await aClient.createAgent({ + agentId: String(remote.agentId), + provider: "claude", + model: "claude-test", + cwd: a.staticDir, + workspaceId: workspace.id, + title: "Unrelated equal ID on A", + }); + expect(await call("get_agent_status", { serverId, agentId: remote.agentId })).toMatchObject({ + serverId, + snapshot: { id: remote.agentId, title: "B child", cwd: b.staticDir }, + }); + expect(await call("get_agent_status", { agentId: remote.agentId })).toMatchObject({ + serverId: directory.self.serverId, + snapshot: { title: "Unrelated equal ID on A" }, + }); + const beforeFollowup = parentPrompts.filter((prompt) => + prompt.includes(`Agent ${remote.agentId}`), + ).length; + expect(beforeFollowup).toBe(1); + expect( + await call("send_agent_prompt", { + serverId, + agentId: remote.agentId, + prompt: "REMOTE_FOLLOWUP_MARKER", + sessionMode: "full-access", + }), + ).toMatchObject({ success: true, serverId, guidance: expect.stringContaining("notified") }); + expect(await call("get_agent_activity", { serverId, agentId: remote.agentId })).toMatchObject({ + serverId, + agentId: remote.agentId, + currentModeId: "full-access", + content: expect.stringContaining("REMOTE_FOLLOWUP_MARKER"), + }); + expect(await call("list_agents", { serverId })).toMatchObject({ + serverId, + agents: expect.arrayContaining([ + expect.objectContaining({ id: remote.agentId, title: "B child", cwd: b.staticDir }), + ]), + }); + expect(await call("list_agents", {})).toMatchObject({ + serverId: directory.self.serverId, + agents: expect.arrayContaining([ + expect.objectContaining({ + id: remote.agentId, + title: "Unrelated equal ID on A", + cwd: a.staticDir, + }), + ]), + }); + await expect + .poll(() => parentPrompts.filter((prompt) => prompt.includes(`Agent ${remote.agentId}`)).length) + .toBe(2); + expect(parentPrompts.findLast((prompt) => prompt.includes(`Agent ${remote.agentId}`))).toContain( + `Server: ${serverId}`, + ); + expect((await aClient.fetchAgent(String(remote.agentId)))?.agent.lastUserMessageAt).toBeNull(); + expect(JSON.stringify(await aClient.fetchAgentTimeline(String(remote.agentId)))).not.toContain( + "REMOTE_FOLLOWUP_MARKER", + ); + + for (const selector of [{}, { serverId: directory.self.serverId }]) { + expect(await call("get_agent_status", { ...selector, agentId: self.agentId })).toMatchObject({ + serverId: directory.self.serverId, + snapshot: { title: "Self child" }, + }); + const previous = parentPrompts.filter((prompt) => + prompt.includes(`Agent ${self.agentId}`), + ).length; + expect( + await call("send_agent_prompt", { + ...selector, + agentId: self.agentId, + prompt: "SELF_FOLLOWUP_MARKER", + }), + ).toMatchObject({ + success: true, + serverId: directory.self.serverId, + disposition: "turn_started", + }); + expect(await call("get_agent_activity", { ...selector, agentId: self.agentId })).toMatchObject({ + serverId: directory.self.serverId, + content: expect.stringContaining("SELF_FOLLOWUP_MARKER"), + }); + expect(await call("list_agents", selector)).toMatchObject({ + serverId: directory.self.serverId, + agents: expect.arrayContaining([ + expect.objectContaining({ id: self.agentId, title: "Self child" }), + ]), + }); + await expect + .poll(() => parentPrompts.filter((prompt) => prompt.includes(`Agent ${self.agentId}`)).length) + .toBe(previous + 1); + } +}); diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 8d14e27e9f..8f36e1a2b8 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. |