mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-25 18:52:39 +00:00
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.
This commit is contained in:
Mohamed Boudra
@@ -29,6 +29,22 @@ principal. An ordinary direct B consumer does not become that producer by supply
|
||||
The optional `server_info.sourceId` reports this producer identity separately from the authenticated admission principal.
|
||||
Trusted server plugins retain the authority of their runtime; they are not sandboxed by this feature.
|
||||
|
||||
## Delegating agents
|
||||
|
||||
Agent MCP discovers destination IDs with `list_hosts`. `create_workspace`, `list_workspaces`, and
|
||||
`create_agent` accept `serverId`; omission selects the calling daemon. Remote creation requires a
|
||||
destination workspace and explicit provider/model. Caller filesystem defaults apply only on the
|
||||
calling daemon.
|
||||
|
||||
The parent’s Subagents track shows qualified children across connected hosts. Opening a child uses
|
||||
the ordinary conversation. The app retains one route per destination; a request through another
|
||||
origin reports a conflict. It does not substitute an existing direct connection.
|
||||
|
||||
Finish notifications use the existing idle wait, including a pending initial run or an already idle
|
||||
child. They name the server and agent. A daemon restart or route loss ends an outstanding notification
|
||||
wait; inspect or wait for the child again rather than resending its prompt. Concurrent prompts gain
|
||||
no additional turn attribution or delivery guarantee.
|
||||
|
||||
## Connections and recovery
|
||||
|
||||
The shared Node connector owns the target grammar and each connection attempt, including a fresh
|
||||
|
||||
@@ -1,10 +1,17 @@
|
||||
import { submitMessage } from "../support/helpers/composer";
|
||||
import { writeFile } from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { expect, test, type Page } from "../support/fixtures";
|
||||
import { runWorkspaceActionFromCommandCenter } from "../support/helpers/command-center-workspace-actions";
|
||||
import { openAgentRoute, seedMockAgentWorkspace } from "../support/helpers/mock-agent";
|
||||
import { seedWorkspace } from "../support/helpers/seed-client";
|
||||
import { openSubagentsTrack, seedParentWithSubagent } from "../support/helpers/subagents";
|
||||
import {
|
||||
delegateFleetThroughMcp,
|
||||
openSubagentsTrack,
|
||||
seedParentWithSubagent,
|
||||
} from "../support/helpers/subagents";
|
||||
|
||||
test.use({ e2eInjectPaseoTools: true });
|
||||
|
||||
const SETTINGS_KEY = "@paseo:app-settings";
|
||||
|
||||
@@ -147,3 +154,100 @@ test("a preferred chat file link still opens beside its right-pane agent", async
|
||||
await workspace.cleanup();
|
||||
}
|
||||
});
|
||||
|
||||
test("MCP delegates to self and B and opens both qualified children in ordinary conversations", async ({
|
||||
page,
|
||||
}) => {
|
||||
const fleet = await delegateFleetThroughMcp();
|
||||
let appSockets = 0;
|
||||
let routeOpens = 0;
|
||||
const activeDaemonSockets = new Set<unknown>();
|
||||
const observationRequests: Record<string, number> = {};
|
||||
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();
|
||||
}
|
||||
});
|
||||
|
||||
@@ -1,3 +1,7 @@
|
||||
import { connectDaemonClient } from "./daemon-client-loader";
|
||||
import { startIsolatedHostDaemon } from "./isolated-host-daemon";
|
||||
import { seedWorkspace } from "./seed-client";
|
||||
import { getE2EDaemonPort } from "./daemon-port";
|
||||
import { PARENT_AGENT_ID_LABEL } from "@getpaseo/protocol/agent-labels";
|
||||
import { expect, type Page } from "@playwright/test";
|
||||
import { daemonWsRoutePattern } from "./daemon-port";
|
||||
@@ -320,3 +324,144 @@ export async function detachSubagentFromTrack(page: Page, childId: string): Prom
|
||||
await expect(detachButton).toBeVisible({ timeout: 30_000 });
|
||||
await detachButton.click();
|
||||
}
|
||||
|
||||
/** The primary children come from the real agent MCP endpoint on A. */
|
||||
export async function delegateFleetThroughMcp() {
|
||||
const { Client } = await import("@modelcontextprotocol/sdk/client/index.js");
|
||||
const { StreamableHTTPClientTransport } =
|
||||
await import("@modelcontextprotocol/sdk/client/streamableHttp.js");
|
||||
const { z } = await import("zod");
|
||||
type DaemonClient = import("@getpaseo/client/internal/daemon-client").DaemonClient;
|
||||
const workspace = await seedWorkspace({ repoPrefix: "fleet-mcp-" });
|
||||
const destination = await startIsolatedHostDaemon(`fleet_${crypto.randomUUID()}`);
|
||||
const a = await connectDaemonClient<DaemonClient>({ clientIdPrefix: "fleet-seed-A" });
|
||||
const b = await connectDaemonClient<DaemonClient>({
|
||||
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<string, unknown> = {}) => {
|
||||
const response = await mcp.callTool({ name, arguments: args });
|
||||
expect(response, name).not.toMatchObject({ isError: true });
|
||||
return z.record(z.string(), z.unknown()).parse(response.structuredContent);
|
||||
};
|
||||
try {
|
||||
const parent = await a.createAgent({
|
||||
provider: "mock",
|
||||
model: "e2e-fast-stream",
|
||||
cwd: workspace.repoPath,
|
||||
workspaceId: workspace.workspaceId,
|
||||
title: "Fleet parent",
|
||||
modeId: "load-test",
|
||||
});
|
||||
const registration = await a.addHost({
|
||||
connection: `127.0.0.1:${destination.port}`,
|
||||
label: "Fleet B",
|
||||
});
|
||||
await expect
|
||||
.poll(
|
||||
async () =>
|
||||
(await a.listHosts()).hosts.find(
|
||||
(row) => row.registrationId === registration.registrationId,
|
||||
)?.status,
|
||||
)
|
||||
.toBe("online");
|
||||
await mcp.connect(
|
||||
new StreamableHTTPClientTransport(
|
||||
new URL(`http://127.0.0.1:${getE2EDaemonPort()}/mcp/agents?callerAgentId=${parent.id}`),
|
||||
),
|
||||
);
|
||||
expect(await call("list_hosts")).toMatchObject({
|
||||
hosts: expect.arrayContaining([
|
||||
expect.objectContaining({ serverId: destination.serverId, label: "Fleet B" }),
|
||||
]),
|
||||
});
|
||||
const target = await call("create_workspace", {
|
||||
serverId: destination.serverId,
|
||||
isolation: "local",
|
||||
path: workspace.repoPath,
|
||||
title: "B workspace",
|
||||
});
|
||||
expect(await call("list_workspaces", { serverId: destination.serverId })).toMatchObject({
|
||||
workspaces: expect.arrayContaining([
|
||||
expect.objectContaining({ workspaceId: target.workspaceId }),
|
||||
]),
|
||||
});
|
||||
const createChild = async (targetServerId: string, workspaceId: string, title: string) => {
|
||||
const result = await call("create_agent", {
|
||||
serverId: targetServerId,
|
||||
workspaceId,
|
||||
title,
|
||||
provider: "mock/e2e-fast-stream",
|
||||
settings: { modeId: "load-test" },
|
||||
initialPrompt: `Finish ${title}`,
|
||||
});
|
||||
return { serverId: targetServerId, agentId: z.string().parse(result.agentId), title };
|
||||
};
|
||||
const local = await createChild(serverId, workspace.workspaceId, "MCP child on A");
|
||||
const remote = await createChild(
|
||||
destination.serverId,
|
||||
z.string().parse(target.workspaceId),
|
||||
"MCP child on B",
|
||||
);
|
||||
// Existing SDK reservation creates unrelated equal-ID fixtures; primary children remain MCP-created.
|
||||
await b.createAgent({
|
||||
agentId: local.agentId,
|
||||
provider: "mock",
|
||||
model: "e2e-fast-stream",
|
||||
cwd: workspace.repoPath,
|
||||
workspaceId: String(target.workspaceId),
|
||||
title: "Unrelated on B",
|
||||
modeId: "load-test",
|
||||
});
|
||||
await a.createAgent({
|
||||
agentId: remote.agentId,
|
||||
provider: "mock",
|
||||
model: "e2e-fast-stream",
|
||||
cwd: workspace.repoPath,
|
||||
workspaceId: workspace.workspaceId,
|
||||
title: "Unrelated on A",
|
||||
modeId: "load-test",
|
||||
});
|
||||
return {
|
||||
workspace,
|
||||
parent,
|
||||
serverId,
|
||||
local,
|
||||
remote,
|
||||
a,
|
||||
b,
|
||||
close,
|
||||
async expectCompletion(child: typeof local) {
|
||||
await expect
|
||||
.poll(async () => {
|
||||
const timeline = await a.fetchAgentTimeline(parent.id, {
|
||||
direction: "tail",
|
||||
limit: 100,
|
||||
});
|
||||
return timeline.entries
|
||||
.flatMap(({ item }) =>
|
||||
item.type === "tool_call" && item.agentMessage ? [item.agentMessage] : [],
|
||||
)
|
||||
.filter(
|
||||
(message) =>
|
||||
message.event === "finished" &&
|
||||
message.sender.id === child.agentId &&
|
||||
message.text.includes(`Server: ${child.serverId}`),
|
||||
).length;
|
||||
})
|
||||
.toBe(1);
|
||||
},
|
||||
};
|
||||
} catch (error) {
|
||||
await close();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ function connectionListen(connection: HostConnection): string | null {
|
||||
case "directPipe":
|
||||
return `pipe://${connection.path}`;
|
||||
case "remoteSsh":
|
||||
case "routed":
|
||||
case "relay":
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -367,6 +367,11 @@ function useArchiveAgentPendingQuery() {
|
||||
});
|
||||
}
|
||||
|
||||
export function usePendingArchiveAgentKeys(): ReadonlySet<string> {
|
||||
const pendingQuery = useArchiveAgentPendingQuery();
|
||||
return useMemo(() => new Set(Object.keys(pendingQuery.data ?? {})), [pendingQuery.data]);
|
||||
}
|
||||
|
||||
export function usePendingArchiveAgentIds(serverId: string): ReadonlySet<string> {
|
||||
const pendingQuery = useArchiveAgentPendingQuery();
|
||||
return useMemo(
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
import type { AgentAddress } from "@getpaseo/protocol/addresses";
|
||||
import { getHostRuntimeStore } from "@/runtime/host-runtime";
|
||||
import { useToast } from "@/contexts/toast-context";
|
||||
import { memo, useCallback, type ReactElement } from "react";
|
||||
import { WorkspaceDiffStatPill } from "@/composer/diff-stat-pill";
|
||||
import { useWorkspaceHasDiffStat } from "@/composer/workspace-diff-stat";
|
||||
@@ -70,18 +73,28 @@ export const AgentTracks = memo(function AgentTracks({
|
||||
},
|
||||
[openInSidePane, tabId, workspaceKey],
|
||||
);
|
||||
const toast = useToast();
|
||||
const handleOpenSubagent = useCallback(
|
||||
(subagentId: string) => {
|
||||
const session = useSessionStore.getState().sessions[serverId];
|
||||
({ serverId: targetServerId, agentId: subagentId }: AgentAddress) => {
|
||||
try {
|
||||
getHostRuntimeStore().requireHostRoute(targetServerId, serverId);
|
||||
} catch (error) {
|
||||
toast.error(error instanceof Error ? error.message : String(error));
|
||||
return;
|
||||
}
|
||||
const session = useSessionStore.getState().sessions[targetServerId];
|
||||
const agent = session?.agents.get(subagentId) ?? session?.agentDetails.get(subagentId);
|
||||
// A subagent in another workspace lives in that workspace's tabs.
|
||||
if (agent?.workspaceId && agent.workspaceId !== workspaceId) {
|
||||
navigateToAgent({ serverId, agentId: subagentId });
|
||||
if (
|
||||
targetServerId !== serverId ||
|
||||
(agent?.workspaceId && agent.workspaceId !== workspaceId)
|
||||
) {
|
||||
navigateToAgent({ serverId: targetServerId, agentId: subagentId });
|
||||
return;
|
||||
}
|
||||
openSubagentTarget({ kind: "agent", agentId: subagentId });
|
||||
},
|
||||
[openSubagentTarget, serverId, workspaceId],
|
||||
[openSubagentTarget, serverId, workspaceId, toast],
|
||||
);
|
||||
const handleOpenProviderSubagent = useCallback(
|
||||
(parentAgentId: string, subagentId: string) => {
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
import { createHostChannelTransport } from "@getpaseo/client/internal/hosts";
|
||||
import type { OwnedSubscription } from "@getpaseo/client";
|
||||
import type { HostDirectory } from "@getpaseo/protocol/hosts";
|
||||
import { useSyncExternalStore, useMemo } from "react";
|
||||
import AsyncStorage from "@react-native-async-storage/async-storage";
|
||||
import equal from "fast-deep-equal/es6";
|
||||
@@ -113,6 +116,7 @@ export type ConnectionLinkImport =
|
||||
| { status: "cancelled" };
|
||||
|
||||
export type ActiveConnection =
|
||||
| { type: "routed"; endpoint: string; display: string }
|
||||
| { type: "directTcp"; endpoint: string; display: string }
|
||||
| { type: "directSocket"; endpoint: string; display: "socket" }
|
||||
| { type: "directPipe"; endpoint: string; display: "pipe" }
|
||||
@@ -234,6 +238,12 @@ const ADAPTIVE_SWITCH_CONSECUTIVE_PROBES = 3;
|
||||
const CONFIGURED_OVERRIDE_BOOTSTRAP_RETRY_MS = 1_000;
|
||||
|
||||
function toActiveConnection(connection: HostConnection): ActiveConnection {
|
||||
if (connection.type === "routed")
|
||||
return {
|
||||
type: "routed",
|
||||
endpoint: connection.originServerId,
|
||||
display: `via ${connection.originServerId}`,
|
||||
};
|
||||
if (connection.type === "directSocket") {
|
||||
return {
|
||||
type: "directSocket",
|
||||
@@ -518,7 +528,9 @@ function probeIntervalForConnection(
|
||||
return PROBE_MAX_BACKOFF_MS;
|
||||
}
|
||||
|
||||
function createDefaultDeps(): HostRuntimeControllerDeps {
|
||||
function createDefaultDeps(
|
||||
resolveOrigin: (serverId: string) => DaemonClient | null = () => null,
|
||||
): HostRuntimeControllerDeps {
|
||||
const browserHostAvailable =
|
||||
typeof getDesktopHost()?.browser?.executeAutomationCommand === "function";
|
||||
const browserAutomationCapabilities = browserHostAvailable
|
||||
@@ -549,6 +561,15 @@ function createDefaultDeps(): HostRuntimeControllerDeps {
|
||||
...(host.password ? { password: host.password } : {}),
|
||||
localCredential: () => readDesktopManagedLocalCredential(connection),
|
||||
} satisfies Omit<DaemonClientConfig, "url">;
|
||||
if (connection.type === "routed") {
|
||||
const origin = resolveOrigin(connection.originServerId);
|
||||
if (!origin) throw new Error(`Origin host ${connection.originServerId} is unavailable`);
|
||||
return new DaemonClient({
|
||||
...base,
|
||||
url: "ws://routed-host/ws",
|
||||
transportFactory: createHostChannelTransport(origin, host.serverId, clientId, null),
|
||||
});
|
||||
}
|
||||
if (connection.type === "directSocket" || connection.type === "directPipe") {
|
||||
return new DaemonClient({
|
||||
...base,
|
||||
@@ -597,15 +618,40 @@ function createDefaultDeps(): HostRuntimeControllerDeps {
|
||||
},
|
||||
});
|
||||
},
|
||||
connectToDaemon: ({ host, connection, timeoutMs }) =>
|
||||
connectToDaemon(connection, {
|
||||
connectToDaemon: async ({ host, connection, timeoutMs }) => {
|
||||
if (connection.type === "routed") {
|
||||
const origin = resolveOrigin(connection.originServerId);
|
||||
if (!origin?.isConnected)
|
||||
throw new Error(`Origin host ${connection.originServerId} is unavailable`);
|
||||
const client = new DaemonClient({
|
||||
url: "ws://routed-host/ws",
|
||||
clientId: crypto.randomUUID(),
|
||||
transportFactory: createHostChannelTransport(
|
||||
origin,
|
||||
host.serverId,
|
||||
crypto.randomUUID(),
|
||||
null,
|
||||
),
|
||||
reconnect: { enabled: false },
|
||||
appVersion: resolveAppVersion() ?? undefined,
|
||||
});
|
||||
try {
|
||||
await client.connect();
|
||||
return { client, serverId: host.serverId, hostname: null };
|
||||
} catch (error) {
|
||||
await client.close();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
return connectToDaemon(connection, {
|
||||
...(host.serverId ? { serverId: host.serverId } : {}),
|
||||
...(host.password ? { password: host.password } : {}),
|
||||
localCredential: () => readDesktopManagedLocalCredential(connection),
|
||||
...(timeoutMs !== undefined ? { timeoutMs } : {}),
|
||||
capabilities: appCapabilities,
|
||||
trace: nativePerformanceTrace,
|
||||
}),
|
||||
});
|
||||
},
|
||||
getClientId: () => getOrCreateClientId(),
|
||||
mountClientHandlers: ({ client, host }) => {
|
||||
const unmountServerData = mountServerDataPushRouter({
|
||||
@@ -1488,6 +1534,11 @@ interface AgentDirectoryRefreshInput {
|
||||
|
||||
export class HostRuntimeStore {
|
||||
private controllers = new Map<string, HostRuntimeController>();
|
||||
private hostObservations = new Map<
|
||||
string,
|
||||
{ client: DaemonClient; observation: OwnedSubscription<HostDirectory>; unsubscribe: () => void }
|
||||
>();
|
||||
private fleetDirectoryDemand = new Map<string, () => void>();
|
||||
private serverListeners = new Map<string, Set<() => void>>();
|
||||
private globalListeners = new Set<() => void>();
|
||||
private hostListListeners = new Set<() => void>();
|
||||
@@ -1518,7 +1569,7 @@ export class HostRuntimeStore {
|
||||
replicaRowStore?: ReplicaRowStore;
|
||||
revokePushNotifications?: typeof revokePushNotifications;
|
||||
}) {
|
||||
this.deps = input?.deps ?? createDefaultDeps();
|
||||
this.deps = input?.deps ?? createDefaultDeps((serverId) => this.getClient(serverId));
|
||||
this.storage = input?.storage ?? AsyncStorage;
|
||||
this.replicaCache = new ReplicaCache(input?.replicaRowStore ?? createReplicaRowStore());
|
||||
this.revokePushNotifications = input?.revokePushNotifications ?? revokePushNotifications;
|
||||
@@ -2185,6 +2236,12 @@ export class HostRuntimeStore {
|
||||
connection: HostConnection;
|
||||
existingClient?: DaemonClient;
|
||||
}): Promise<HostProfile> {
|
||||
const retained = this.hosts.find((host) => host.serverId === input.serverId);
|
||||
if (
|
||||
retained?.connections.some((connection) => connection.type === "routed") &&
|
||||
input.connection.type !== "routed"
|
||||
)
|
||||
throw new Error(`Host ${input.serverId} already has a retained routed connection`);
|
||||
const now = new Date().toISOString();
|
||||
const next = upsertHostConnectionInProfiles({
|
||||
profiles: this.hosts,
|
||||
@@ -2258,6 +2315,11 @@ export class HostRuntimeStore {
|
||||
if (nextIds.has(serverId)) {
|
||||
continue;
|
||||
}
|
||||
this.hostObservations.get(serverId)?.unsubscribe();
|
||||
void this.hostObservations.get(serverId)?.observation.release();
|
||||
this.hostObservations.delete(serverId);
|
||||
this.fleetDirectoryDemand.get(serverId)?.();
|
||||
this.fleetDirectoryDemand.delete(serverId);
|
||||
this.controllers.delete(serverId);
|
||||
this.lastConnectionStatusByServer.delete(serverId);
|
||||
this.connectionStatusStartedAtByServer.delete(serverId);
|
||||
@@ -2314,6 +2376,7 @@ export class HostRuntimeStore {
|
||||
const snapshot = controller.getSnapshot();
|
||||
this.syncSessionReplica(snapshot.serverId, snapshot);
|
||||
this.syncDirectoryConnection(snapshot.serverId);
|
||||
this.syncHostDiscovery(snapshot.serverId);
|
||||
this.emit(snapshot.serverId);
|
||||
});
|
||||
void controller
|
||||
@@ -2332,6 +2395,73 @@ export class HostRuntimeStore {
|
||||
}
|
||||
}
|
||||
|
||||
/** A retained destination uses its selected route; another route is an explicit conflict. */
|
||||
requireHostRoute(serverId: string, originServerId: string): void {
|
||||
if (serverId === originServerId) return;
|
||||
const connection = this.hosts.find((host) => host.serverId === serverId)?.connections[0];
|
||||
if (connection?.type !== "routed" || connection.originServerId !== originServerId) {
|
||||
throw new Error(
|
||||
`Host ${serverId} is already connected through another route. Requested via ${originServerId}.`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private syncHostDiscovery(serverId: string): void {
|
||||
const client = this.getClient(serverId);
|
||||
const previous = this.hostObservations.get(serverId);
|
||||
if (previous?.client === client) return;
|
||||
previous?.unsubscribe();
|
||||
void previous?.observation.release();
|
||||
this.hostObservations.delete(serverId);
|
||||
const host = this.hosts.find((candidate) => candidate.serverId === serverId);
|
||||
if (
|
||||
!client?.isConnected ||
|
||||
!client.getLastServerInfoMessage()?.features?.connectedHosts ||
|
||||
host?.connections.some((connection) => connection.type === "routed")
|
||||
)
|
||||
return;
|
||||
const observation = client.observeHosts();
|
||||
const apply = (directory: HostDirectory) => {
|
||||
if (this.hostObservations.get(serverId)?.client !== client) return;
|
||||
for (const target of directory.hosts) {
|
||||
if (!target.serverId || target.serverId === serverId) continue;
|
||||
if (!this.hosts.some((candidate) => candidate.serverId === target.serverId)) {
|
||||
const now = new Date().toISOString();
|
||||
const connection = {
|
||||
id: `via:${serverId}`,
|
||||
type: "routed" as const,
|
||||
originServerId: serverId,
|
||||
};
|
||||
const profile: HostProfile = {
|
||||
serverId: target.serverId,
|
||||
label: target.label ?? target.serverId,
|
||||
appearance: defaultHostAppearance(),
|
||||
lifecycle: {},
|
||||
connections: [connection],
|
||||
preferredConnectionId: connection.id,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
};
|
||||
this.setHostsAndSync([...this.hosts, profile]);
|
||||
void this.persistHosts();
|
||||
}
|
||||
if (!this.fleetDirectoryDemand.has(target.serverId))
|
||||
this.fleetDirectoryDemand.set(
|
||||
target.serverId,
|
||||
this.acquireDirectoryDemand(target.serverId),
|
||||
);
|
||||
}
|
||||
};
|
||||
const unsubscribe = observation.subscribe({
|
||||
snapshot: apply,
|
||||
update: (message) => {
|
||||
if (message.type === "hosts.changed") apply(message.payload.directory);
|
||||
},
|
||||
});
|
||||
this.hostObservations.set(serverId, { client, observation, unsubscribe });
|
||||
void observation.ready.then(apply).catch(() => undefined);
|
||||
}
|
||||
|
||||
private syncSessionReplica(serverId: string, snapshot: HostRuntimeSnapshot): void {
|
||||
if (!snapshot.client) {
|
||||
return;
|
||||
|
||||
@@ -106,6 +106,7 @@ const removeProfileIcon = <ThemedTrash2 size={ICON_SIZE.sm} uniProps={destructiv
|
||||
const addProfileIcon = <ThemedPlus size={ICON_SIZE.sm} uniProps={mutedColorMapping} />;
|
||||
|
||||
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})`;
|
||||
}
|
||||
|
||||
@@ -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"));
|
||||
},
|
||||
|
||||
@@ -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<Agent, "id" | "status" | "parentAgent
|
||||
|
||||
export interface ArchiveFinishedSubagentsDeps {
|
||||
parentAgentId: string;
|
||||
getManagedSubagent: (id: string) => ManagedSubagentSnapshot | undefined;
|
||||
archiveManagedSubagent: (id: string) => Promise<void>;
|
||||
getManagedSubagent: (address: AgentAddress) => ManagedSubagentSnapshot | undefined;
|
||||
archiveManagedSubagent: (address: AgentAddress) => Promise<void>;
|
||||
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<string> {
|
||||
@@ -169,7 +170,7 @@ async function runArchiveFinished(
|
||||
reportProgress: (completedCount: number) => void,
|
||||
): Promise<{ outcome: ArchiveFinishedOutcome; retryableFailureIds: Set<string> }> {
|
||||
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<string>();
|
||||
@@ -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 {
|
||||
|
||||
@@ -53,7 +53,14 @@ const AGENT_DEFAULTS: Agent = {
|
||||
};
|
||||
|
||||
function makeAgent(input: Partial<Agent> & Pick<Agent, "id">): 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"]);
|
||||
|
||||
@@ -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<string>,
|
||||
): 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<string>,
|
||||
): (state: SessionStoreSnapshot) => SubagentRow[] {
|
||||
let previousAgents: Map<string, Agent> | undefined;
|
||||
let previousAgents: Array<[string, Map<string, Agent>]> = [];
|
||||
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<string, Agent>],
|
||||
);
|
||||
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],
|
||||
|
||||
@@ -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}`,
|
||||
|
||||
@@ -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) => (
|
||||
<SubagentsTrackRow
|
||||
key={row.id}
|
||||
key={row.kind === "paseo" ? `${row.address.serverId}:${row.id}` : row.id}
|
||||
row={row}
|
||||
serverId={serverId}
|
||||
onOpenSubagent={onOpenSubagent}
|
||||
@@ -180,11 +182,11 @@ function ArchiveFinishedRow({
|
||||
interface SubagentsTrackRowProps {
|
||||
serverId: string;
|
||||
row: SubagentRow;
|
||||
onOpenSubagent: (id: string) => 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}
|
||||
</Text>
|
||||
{row.kind === "paseo" ? (
|
||||
<PaseoSubagentFacts serverId={serverId} agentId={row.id} />
|
||||
<>
|
||||
<SubagentFacts rowId={row.id} text={rowServerId !== serverId ? hostLabel : ""} />
|
||||
<PaseoSubagentFacts serverId={rowServerId} agentId={row.id} />
|
||||
</>
|
||||
) : (
|
||||
<SubagentFacts rowId={row.id} text={presentation.subtitle} />
|
||||
)}
|
||||
@@ -253,6 +260,8 @@ function SubagentsTrackRow({
|
||||
row.kind,
|
||||
row.id,
|
||||
serverId,
|
||||
rowServerId,
|
||||
hostLabel,
|
||||
],
|
||||
);
|
||||
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { getHostRuntimeStore } from "@/runtime/host-runtime";
|
||||
import { getParentAddressFromLabels } from "@getpaseo/protocol/agent-labels";
|
||||
import { useEffect, useMemo, useSyncExternalStore } from "react";
|
||||
import { useArchiveAgent } from "@/hooks/use-archive-agent";
|
||||
import { useSessionStore } from "@/stores/session-store";
|
||||
@@ -32,8 +34,19 @@ export function useArchiveFinishedSubagents({
|
||||
() =>
|
||||
createArchiveFinishedSubagents([], {
|
||||
parentAgentId,
|
||||
getManagedSubagent: (id) => useSessionStore.getState().sessions[serverId]?.agents.get(id),
|
||||
archiveManagedSubagent: (id) => archiveAgent({ serverId, agentId: id }),
|
||||
getManagedSubagent: (address) => {
|
||||
const agent = useSessionStore
|
||||
.getState()
|
||||
.sessions[address.serverId]?.agents.get(address.agentId);
|
||||
if (!agent) return undefined;
|
||||
const parent = getParentAddressFromLabels(agent.labels, address.serverId);
|
||||
if (parent?.serverId !== serverId || parent.agentId !== parentAgentId) return undefined;
|
||||
return { ...agent, parentAgentId: parent.agentId };
|
||||
},
|
||||
archiveManagedSubagent: async (address) => {
|
||||
getHostRuntimeStore().requireHostRoute(address.serverId, serverId);
|
||||
await archiveAgent(address);
|
||||
},
|
||||
dismissProviderSubagents: (ids) => {
|
||||
useProviderSubagentStore.getState().hideFromTrack(serverId, parentAgentId, ids);
|
||||
},
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { getHostRuntimeStore } from "@/runtime/host-runtime";
|
||||
import type { AgentAddress } from "@getpaseo/protocol/addresses";
|
||||
import { useCallback } from "react";
|
||||
import { useToast } from "@/contexts/toast-context";
|
||||
import { useArchiveAgent } from "@/hooks/use-archive-agent";
|
||||
@@ -17,26 +19,30 @@ export interface UseArchiveSubagentInput {
|
||||
serverId: string;
|
||||
}
|
||||
|
||||
export function useArchiveSubagent(input: UseArchiveSubagentInput): (subagentId: string) => void {
|
||||
export function useArchiveSubagent({
|
||||
serverId: originServerId,
|
||||
}: UseArchiveSubagentInput): (address: AgentAddress) => void {
|
||||
const { archiveAgent } = useArchiveAgent();
|
||||
const { serverId } = input;
|
||||
const toast = useToast();
|
||||
|
||||
return useCallback(
|
||||
(subagentId: string) => {
|
||||
({ serverId, agentId: subagentId }: AgentAddress) => {
|
||||
void requestArchiveSubagent(
|
||||
{ serverId, subagentId },
|
||||
{
|
||||
getSubagent: (id): ResolveArchiveSubagentDialogInput | undefined =>
|
||||
useSessionStore.getState().sessions[serverId]?.agents?.get(id),
|
||||
confirm: confirmDialog,
|
||||
archiveAgent,
|
||||
archiveAgent: async (address) => {
|
||||
getHostRuntimeStore().requireHostRoute(address.serverId, originServerId);
|
||||
await archiveAgent(address);
|
||||
},
|
||||
reportError: (error) => {
|
||||
toast.error(toErrorMessage(error));
|
||||
},
|
||||
},
|
||||
);
|
||||
},
|
||||
[archiveAgent, serverId, toast],
|
||||
[archiveAgent, originServerId, toast],
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { getHostRuntimeStore } from "@/runtime/host-runtime";
|
||||
import type { AgentAddress } from "@getpaseo/protocol/addresses";
|
||||
import { useCallback } from "react";
|
||||
import { useToast } from "@/contexts/toast-context";
|
||||
import { i18n } from "@/i18n/i18next";
|
||||
@@ -18,12 +20,13 @@ export interface UseDetachSubagentInput {
|
||||
serverId: string;
|
||||
}
|
||||
|
||||
export function useDetachSubagent(input: UseDetachSubagentInput): (subagentId: string) => void {
|
||||
const { serverId } = input;
|
||||
export function useDetachSubagent({
|
||||
serverId: originServerId,
|
||||
}: UseDetachSubagentInput): (address: AgentAddress) => void {
|
||||
const toast = useToast();
|
||||
|
||||
return useCallback(
|
||||
(subagentId: string) => {
|
||||
({ serverId, agentId: subagentId }: AgentAddress) => {
|
||||
void requestDetachSubagent(
|
||||
{ serverId, subagentId },
|
||||
{
|
||||
@@ -31,6 +34,7 @@ export function useDetachSubagent(input: UseDetachSubagentInput): (subagentId: s
|
||||
useSessionStore.getState().sessions[serverId]?.agents?.get(id),
|
||||
confirm: confirmDialog,
|
||||
detachAgent: async ({ serverId: targetServerId, agentId }) => {
|
||||
getHostRuntimeStore().requireHostRoute(targetServerId, originServerId);
|
||||
const client = useSessionStore.getState().sessions[targetServerId]?.client;
|
||||
if (!client) {
|
||||
throw new Error(i18n.t("workspaceSetup.errors.hostDisconnected"));
|
||||
@@ -46,6 +50,6 @@ export function useDetachSubagent(input: UseDetachSubagentInput): (subagentId: s
|
||||
},
|
||||
);
|
||||
},
|
||||
[serverId, toast],
|
||||
[originServerId, toast],
|
||||
);
|
||||
}
|
||||
|
||||
@@ -51,7 +51,14 @@ export interface RelayHostConnection {
|
||||
daemonPublicKeyB64: string;
|
||||
}
|
||||
|
||||
export interface RoutedHostConnection {
|
||||
id: string;
|
||||
type: "routed";
|
||||
originServerId: string;
|
||||
}
|
||||
|
||||
export type HostConnection =
|
||||
| RoutedHostConnection
|
||||
| DirectTcpHostConnection
|
||||
| DirectSocketHostConnection
|
||||
| DirectPipeHostConnection
|
||||
@@ -128,6 +135,8 @@ function hostConnectionEquals(left: HostConnection, right: HostConnection): bool
|
||||
return false;
|
||||
}
|
||||
|
||||
if (left.type === "routed" && right.type === "routed")
|
||||
return left.originServerId === right.originServerId;
|
||||
if (left.type === "directTcp" && right.type === "directTcp") {
|
||||
return left.endpoint === right.endpoint && (left.useTls ?? false) === (right.useTls ?? false);
|
||||
}
|
||||
@@ -399,6 +408,7 @@ export function createRemoteSshHostConnection(input: {
|
||||
}
|
||||
|
||||
const StoredHostConnectionSchema = z.discriminatedUnion("type", [
|
||||
z.strictObject({ id: z.string(), type: z.literal("routed"), originServerId: z.string().min(1) }),
|
||||
z.strictObject({
|
||||
id: z.string().optional(),
|
||||
type: z.literal("directTcp"),
|
||||
@@ -446,6 +456,7 @@ export const StoredHostRegistrySchema = z.array(StoredHostProfileSchema);
|
||||
type StoredHostConnection = z.infer<typeof StoredHostConnectionSchema>;
|
||||
|
||||
function normalizeStoredConnection(connection: StoredHostConnection): HostConnection | null {
|
||||
if (connection.type === "routed") return connection;
|
||||
if (connection.type === "directTcp") {
|
||||
try {
|
||||
const endpoint = normalizeLoopbackToLocalhost(normalizeHostPort(connection.endpoint));
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -146,6 +146,8 @@ export async function buildClientConfig(
|
||||
| "buildDesktopTransportUrl"
|
||||
> = defaultDaemonConnectionDependencies,
|
||||
): Promise<DaemonClientConfig> {
|
||||
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 = {
|
||||
|
||||
@@ -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": {
|
||||
|
||||
@@ -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" });
|
||||
});
|
||||
|
||||
@@ -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<CreateAgentRequestOptions, "callerAgentId" | "labels"> {
|
||||
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 } {
|
||||
|
||||
@@ -12,6 +12,10 @@
|
||||
"./*": {
|
||||
"types": "./dist/*.d.ts",
|
||||
"default": "./dist/*.js"
|
||||
},
|
||||
"./addresses": {
|
||||
"types": "./dist/addresses.d.ts",
|
||||
"default": "./dist/addresses.js"
|
||||
}
|
||||
},
|
||||
"publishConfig": {
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
/** Agent identity is qualified by its daemon; transport and reachability are separate. */
|
||||
export interface AgentAddress {
|
||||
serverId: string;
|
||||
agentId: string;
|
||||
}
|
||||
@@ -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<string, unknown> | null | undefin
|
||||
([label, value]) => isOpenAgentTabLabel(label) && value === "true",
|
||||
);
|
||||
}
|
||||
|
||||
// COMPAT(localParentLabels): added in v0.12.0; remove after 2027-04-20 once stored local parent labels are qualified.
|
||||
export function getParentAddressFromLabels(
|
||||
labels: Record<string, unknown> | 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<string, string> {
|
||||
return { [PARENT_AGENT_ID_LABEL]: parent.agentId, [PARENT_SERVER_ID_LABEL]: parent.serverId };
|
||||
}
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<AgentManager, "notifyAgentState" | "unarchiveSnapshot">;
|
||||
@@ -292,7 +296,7 @@ async function resolvePromptSource(
|
||||
manager: Pick<AgentManager, "getAgent">,
|
||||
storage: AgentStorage,
|
||||
): Promise<AgentPromptSource | undefined> {
|
||||
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<PromptDispatchDisposition | false>;
|
||||
@@ -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.",
|
||||
`<permission-request>\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<string, Set<ArmedFinishNotification>>
|
||||
>();
|
||||
|
||||
/** 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<WaitForFinishResult>;
|
||||
afterPermission(requestId: string, signal: AbortSignal): Promise<void>;
|
||||
logger: Logger;
|
||||
}): () => void {
|
||||
const controller = new AbortController();
|
||||
const byManager =
|
||||
armedFinishNotifications.get(params.agentManager) ??
|
||||
new Map<string, Set<ArmedFinishNotification>>();
|
||||
armedFinishNotifications.set(params.agentManager, byManager);
|
||||
const key = JSON.stringify([
|
||||
params.childServerId ?? null,
|
||||
params.childAgentId,
|
||||
params.callerAgentId,
|
||||
]);
|
||||
const registrations = byManager.get(key) ?? new Set<ArmedFinishNotification>();
|
||||
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<void> {
|
||||
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<ArmedFinishNotification>();
|
||||
armedByManager.set(armedKey, registrations);
|
||||
const previous = [...registrations];
|
||||
|
||||
@@ -0,0 +1,323 @@
|
||||
import { parentAddressLabels } from "@getpaseo/protocol/agent-labels";
|
||||
import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { z } from "zod";
|
||||
import {
|
||||
DaemonClient,
|
||||
type FetchWorkspacesEntry,
|
||||
type WaitForFinishResult,
|
||||
} from "@getpaseo/client/internal/daemon-client";
|
||||
import type { DaemonTransportFactory } from "@getpaseo/client/internal/daemon-client-transport-types";
|
||||
import type { HostDirectory } from "@getpaseo/protocol/hosts";
|
||||
import { OWNER_PERMISSIONS } from "../../authorization/index.js";
|
||||
import type { ConnectedHosts } from "../../connected-hosts/index.js";
|
||||
import { resolveRequiredProviderModel } from "../mcp-shared.js";
|
||||
import type { PaseoToolExecutionContext, PaseoToolResult } from "../tools/types.js";
|
||||
|
||||
export type HostOperation = (
|
||||
input: Record<string, unknown>,
|
||||
context: PaseoToolExecutionContext,
|
||||
) => Promise<PaseoToolResult>;
|
||||
export interface HostOperations {
|
||||
createAgent: HostOperation;
|
||||
createWorkspace: HostOperation;
|
||||
listWorkspaces: HostOperation;
|
||||
wait(agentId: string, signal: AbortSignal): Promise<WaitForFinishResult>;
|
||||
observe(
|
||||
agentId: string,
|
||||
handler: (snapshot: AgentSnapshotPayload | null) => void,
|
||||
signal: AbortSignal,
|
||||
): Promise<() => Promise<void>>;
|
||||
}
|
||||
export interface ToolHosts {
|
||||
directory(): Promise<HostDirectory>;
|
||||
use<T>(serverId: string, action: (client: DaemonClient) => Promise<T>): Promise<T>;
|
||||
}
|
||||
|
||||
/** 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<ReturnType<ConnectedHosts["openRoute"]>> | 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<T>(
|
||||
serverId: string | undefined,
|
||||
action: (host: HostOperations) => Promise<T>,
|
||||
): Promise<T> {
|
||||
const destination = serverId ?? input.serverId;
|
||||
if (destination === input.serverId) return action(input.local);
|
||||
if (input.lockedCwd)
|
||||
throw new Error("This caller is restricted to its local working directory");
|
||||
return input.hosts().use(destination, async (client) =>
|
||||
action({
|
||||
async createAgent(args) {
|
||||
const parsed = remoteCreate.parse(args);
|
||||
const { provider, model } = resolveRequiredProviderModel(parsed.provider);
|
||||
const workspace = (await client.fetchWorkspaces({})).entries.find(
|
||||
(row) => row.id === parsed.workspaceId,
|
||||
);
|
||||
if (!workspace) throw new Error(`Workspace ${parsed.workspaceId} not found`);
|
||||
const labels = {
|
||||
...parsed.labels,
|
||||
...(input.callerAgentId
|
||||
? parentAddressLabels({ agentId: input.callerAgentId, serverId: input.serverId })
|
||||
: {}),
|
||||
};
|
||||
const snapshot = await client.createAgent({
|
||||
config: {
|
||||
provider,
|
||||
model,
|
||||
cwd: workspace.workspaceDirectory,
|
||||
title: parsed.title,
|
||||
modeId: parsed.settings?.modeId,
|
||||
thinkingOptionId: parsed.settings?.thinkingOptionId,
|
||||
featureValues: parsed.settings?.features,
|
||||
},
|
||||
workspaceId: parsed.workspaceId,
|
||||
labels,
|
||||
initialPrompt: parsed.initialPrompt,
|
||||
idempotencyKey: parsed.idempotencyKey,
|
||||
});
|
||||
return result({
|
||||
agentId: snapshot.id,
|
||||
type: snapshot.provider,
|
||||
status: snapshot.status,
|
||||
cwd: snapshot.cwd,
|
||||
workspaceId: snapshot.workspaceId,
|
||||
currentModeId: snapshot.currentModeId,
|
||||
availableModes: snapshot.availableModes,
|
||||
});
|
||||
},
|
||||
async listWorkspaces(args) {
|
||||
const { entries } = await client.fetchWorkspaces({
|
||||
includeBackground: args.includeBackground === true,
|
||||
});
|
||||
return result({ workspaces: entries.map(workspaceSummary) });
|
||||
},
|
||||
async createWorkspace(args) {
|
||||
const isolation = z.enum(["local", "worktree"]).parse(args.isolation);
|
||||
const path = z.string().min(1).parse(args.path);
|
||||
const response = await client.createWorkspace({
|
||||
source:
|
||||
isolation === "local"
|
||||
? { kind: "directory", path, projectId: args.projectId as string | undefined }
|
||||
: {
|
||||
kind: "worktree",
|
||||
cwd: path,
|
||||
projectId: args.projectId as string | undefined,
|
||||
action:
|
||||
args.mode === "checkout-branch" || args.mode === "checkout-pr"
|
||||
? "checkout"
|
||||
: "branch-off",
|
||||
baseBranch: args.baseBranch as string | undefined,
|
||||
branchName: args.branchName as string | undefined,
|
||||
refName: args.branch as string | undefined,
|
||||
githubPrNumber: args.prNumber as number | undefined,
|
||||
worktreeSlug: args.worktreeSlug as string | undefined,
|
||||
},
|
||||
title: args.title as string | undefined,
|
||||
background: args.background as boolean | undefined,
|
||||
});
|
||||
if (response.error || !response.workspace)
|
||||
throw new Error(response.error ?? "Workspace creation failed");
|
||||
return result(workspaceSummary(response.workspace));
|
||||
},
|
||||
async observe(agentId, handler, signal) {
|
||||
const observation = client.observeTimeline([agentId], { signal });
|
||||
const refresh = async () => handler((await client.fetchAgent(agentId))?.agent ?? null);
|
||||
const unsubscribe = observation.subscribe({
|
||||
snapshot: () => {},
|
||||
update: (message) => {
|
||||
if (message.type !== "agent_stream" || message.payload.agentId !== agentId) return;
|
||||
const event = message.payload.event;
|
||||
if (
|
||||
event.type === "permission_resolved" ||
|
||||
event.type === "turn_started" ||
|
||||
event.type === "turn_completed" ||
|
||||
event.type === "turn_canceled" ||
|
||||
event.type === "turn_failed"
|
||||
)
|
||||
void refresh().catch(() => handler(null));
|
||||
},
|
||||
error: () => handler(null),
|
||||
});
|
||||
try {
|
||||
await observation.ready;
|
||||
await refresh();
|
||||
} catch (error) {
|
||||
unsubscribe();
|
||||
await observation.release();
|
||||
throw error;
|
||||
}
|
||||
return async () => {
|
||||
unsubscribe();
|
||||
await observation.release();
|
||||
};
|
||||
},
|
||||
wait: (agentId, signal) => {
|
||||
const close = () => void client.close();
|
||||
signal.addEventListener("abort", close, { once: true });
|
||||
return client
|
||||
.waitForFinish(agentId, 0)
|
||||
.finally(() => signal.removeEventListener("abort", close));
|
||||
},
|
||||
}),
|
||||
);
|
||||
},
|
||||
};
|
||||
}
|
||||
function workspaceSummary(workspace: FetchWorkspacesEntry) {
|
||||
return {
|
||||
workspaceId: workspace.id,
|
||||
projectId: workspace.projectId,
|
||||
cwd: workspace.workspaceDirectory,
|
||||
isolation: workspace.workspaceKind === "worktree" ? "worktree" : "local",
|
||||
kind: workspace.workspaceKind === "checkout" ? "local_checkout" : workspace.workspaceKind,
|
||||
title: workspace.title,
|
||||
background: workspace.background,
|
||||
};
|
||||
}
|
||||
|
||||
/** Wait pauses at permissions; observe only until that request clears before re-arming it. */
|
||||
export async function waitForPermissionChange(
|
||||
host: HostOperations,
|
||||
agentId: string,
|
||||
requestId: string,
|
||||
signal: AbortSignal,
|
||||
): Promise<void> {
|
||||
signal.throwIfAborted();
|
||||
let changed!: () => void;
|
||||
let failed!: (error: unknown) => void;
|
||||
const ready = new Promise<void>((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<void>) | 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?.();
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,9 @@
|
||||
import {
|
||||
createHostResolver,
|
||||
waitForPermissionChange,
|
||||
type HostOperation,
|
||||
type ToolHosts,
|
||||
} from "../host-operations/index.js";
|
||||
import { AgentDirectory } from "../directory.js";
|
||||
import { stat } from "node:fs/promises";
|
||||
import { z } from "zod";
|
||||
@@ -72,6 +78,7 @@ import {
|
||||
type PromptDispatchDisposition,
|
||||
sendPromptToAgent,
|
||||
setupFinishNotification,
|
||||
setupFinishWait,
|
||||
waitForAgentRunStartWithTimeout,
|
||||
} from "../agent-prompt.js";
|
||||
import { respondToAgentPermission } from "../permission-response.js";
|
||||
@@ -110,6 +117,8 @@ import type { ProviderPaseoToolsPolicy } from "@getpaseo/protocol/provider-confi
|
||||
import { isPaseoToolEnabled } from "../paseo-tool-policy.js";
|
||||
|
||||
export interface PaseoToolHostDependencies {
|
||||
serverId?: string;
|
||||
hosts?: () => ToolHosts;
|
||||
agentManager: AgentManager;
|
||||
agentStorage: AgentStorage;
|
||||
terminalManager?: TerminalManager | null;
|
||||
@@ -165,6 +174,10 @@ export interface PaseoToolHostDependencies {
|
||||
logger: Logger;
|
||||
}
|
||||
|
||||
function settledStatus(status: string): "error" | "idle" {
|
||||
return status === "error" ? "error" : "idle";
|
||||
}
|
||||
|
||||
function parseTimestamp(value: string | null | undefined): number {
|
||||
if (!value) {
|
||||
return 0;
|
||||
@@ -685,7 +698,58 @@ function createToolCatalog(
|
||||
return result.data;
|
||||
};
|
||||
|
||||
const localHostOperations = new Map<string, HostOperation>();
|
||||
const tools = new Map<string, PaseoToolDefinition>();
|
||||
const hostTools = new Map([
|
||||
["create_agent", "createAgent"],
|
||||
["create_workspace", "createWorkspace"],
|
||||
["list_workspaces", "listWorkspaces"],
|
||||
] as const);
|
||||
const hostResolver = createHostResolver({
|
||||
serverId: options.serverId ?? "self",
|
||||
callerAgentId,
|
||||
lockedCwd: callerContext?.lockedCwd,
|
||||
hosts: () => {
|
||||
if (!options.hosts) throw new Error("Connected hosts are not configured");
|
||||
return options.hosts();
|
||||
},
|
||||
local: {
|
||||
createAgent: (args, context) => localHostOperations.get("create_agent")!(args, context),
|
||||
createWorkspace: (args, context) =>
|
||||
localHostOperations.get("create_workspace")!(args, context),
|
||||
listWorkspaces: (args, context) => localHostOperations.get("list_workspaces")!(args, context),
|
||||
async observe(agentId, handler, signal) {
|
||||
const unsubscribe = agentManager.subscribe(
|
||||
(event) => {
|
||||
if (event.type === "agent_state") handler(toAgentPayload(event.agent));
|
||||
},
|
||||
{ agentId, replayState: true },
|
||||
);
|
||||
const stop = () => unsubscribe();
|
||||
signal.addEventListener("abort", stop, { once: true });
|
||||
return async () => {
|
||||
signal.removeEventListener("abort", stop);
|
||||
unsubscribe();
|
||||
};
|
||||
},
|
||||
async wait(agentId, signal) {
|
||||
const settled = await agentManager.waitForAgentEvent(agentId, {
|
||||
signal,
|
||||
waitForActive: true,
|
||||
});
|
||||
const snapshot = agentManager.getAgent(agentId);
|
||||
const final = snapshot
|
||||
? await serializeSnapshotWithMetadata(agentStorage, snapshot, childLogger)
|
||||
: null;
|
||||
return {
|
||||
status: settled.permission ? "permission" : settledStatus(settled.status),
|
||||
final,
|
||||
lastMessage: settled.lastMessage,
|
||||
error: final?.lastError ?? null,
|
||||
};
|
||||
},
|
||||
},
|
||||
});
|
||||
const registerTool = (
|
||||
name: string,
|
||||
config: PaseoToolConfig,
|
||||
@@ -695,6 +759,73 @@ function createToolCatalog(
|
||||
if (!isPaseoToolEnabled(options.paseoToolPolicy, name)) {
|
||||
return;
|
||||
}
|
||||
const operation = hostTools.get(
|
||||
name as "create_agent" | "create_workspace" | "list_workspaces",
|
||||
);
|
||||
if (operation) {
|
||||
localHostOperations.set(name, handler);
|
||||
const original = config.inputSchema;
|
||||
const selector = {
|
||||
serverId: z
|
||||
.string()
|
||||
.min(1)
|
||||
.optional()
|
||||
.describe("Destination server ID from list_hosts; omit for this host."),
|
||||
};
|
||||
config = {
|
||||
...config,
|
||||
inputSchema:
|
||||
original instanceof z.ZodObject
|
||||
? original.extend(selector)
|
||||
: { ...(original as z.ZodRawShape), ...selector },
|
||||
};
|
||||
handler = async ({ serverId, ...input }, context) => {
|
||||
const destination = serverId ?? options.serverId;
|
||||
try {
|
||||
const response = await hostResolver.use(serverId, (host) =>
|
||||
host[operation](input, context),
|
||||
);
|
||||
if (name === "create_agent" && callerAgentId && input.notifyOnFinish !== false) {
|
||||
const created = response.structuredContent as { agentId: string };
|
||||
setupFinishWait({
|
||||
agentManager,
|
||||
agentStorage,
|
||||
originServerId: options.serverId,
|
||||
childServerId: destination,
|
||||
childAgentId: created.agentId,
|
||||
callerAgentId,
|
||||
logger: childLogger,
|
||||
wait: (signal) =>
|
||||
hostResolver.use(destination, (host) => host.wait(created.agentId, signal)),
|
||||
afterPermission: (requestId, signal) =>
|
||||
hostResolver.use(destination, (host) =>
|
||||
waitForPermissionChange(host, created.agentId, requestId, signal),
|
||||
),
|
||||
});
|
||||
}
|
||||
return {
|
||||
...response,
|
||||
structuredContent: {
|
||||
...(response.structuredContent as object),
|
||||
...(destination ? { serverId: destination } : {}),
|
||||
},
|
||||
};
|
||||
} catch (error) {
|
||||
if (!options.serverId) throw error;
|
||||
const failure = {
|
||||
serverId: destination,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
...(error instanceof Error && "code" in error ? { code: error.code } : {}),
|
||||
...(error instanceof Error && "outcome" in error ? { outcome: error.outcome } : {}),
|
||||
};
|
||||
return {
|
||||
isError: true,
|
||||
content: [{ type: "text", text: JSON.stringify(failure) }],
|
||||
structuredContent: failure,
|
||||
};
|
||||
}
|
||||
};
|
||||
}
|
||||
tools.set(name, {
|
||||
name,
|
||||
title: config.title,
|
||||
@@ -1332,6 +1463,19 @@ function createToolCatalog(
|
||||
});
|
||||
}
|
||||
|
||||
registerTool(
|
||||
"list_hosts",
|
||||
{
|
||||
title: "List hosts",
|
||||
description: "Discover this host and registered destination hosts for fleet tools.",
|
||||
inputSchema: {},
|
||||
},
|
||||
async () => ({
|
||||
content: [],
|
||||
structuredContent: ensureValidJson(await hostResolver.directory()),
|
||||
}),
|
||||
);
|
||||
|
||||
registerTool(
|
||||
"create_workspace",
|
||||
{
|
||||
@@ -1593,10 +1737,15 @@ function createToolCatalog(
|
||||
workspaceId: resolvedArgs.workspaceId,
|
||||
thinking: parsedArgs.settings?.thinkingOptionId,
|
||||
features: parsedArgs.settings?.features,
|
||||
labels: parsedArgs.labels,
|
||||
labels: {
|
||||
...parsedArgs.labels,
|
||||
...(callerAgentId && !resolvedArgs.detached && options.serverId
|
||||
? { "paseo.parent-server-id": options.serverId }
|
||||
: {}),
|
||||
},
|
||||
mode: parsedArgs.settings?.modeId,
|
||||
promptFailure: "throw",
|
||||
notifyOnFinish,
|
||||
notifyOnFinish: false,
|
||||
detached: resolvedArgs.detached,
|
||||
callerAgentId,
|
||||
callerContext,
|
||||
@@ -2003,6 +2152,7 @@ function createToolCatalog(
|
||||
agentManager,
|
||||
agentStorage,
|
||||
childAgentId: agentId,
|
||||
childServerId: options.serverId,
|
||||
callerAgentId,
|
||||
waitForTurnStart: false,
|
||||
dispatchResult,
|
||||
@@ -2023,6 +2173,7 @@ function createToolCatalog(
|
||||
agentManager,
|
||||
agentStorage,
|
||||
childAgentId: agentId,
|
||||
childServerId: options.serverId,
|
||||
callerAgentId,
|
||||
waitForTurnStart: true,
|
||||
logger: childLogger,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<unknown>)[] = [];
|
||||
afterEach(async () => {
|
||||
for (const close of cleanup.splice(0).toReversed()) await close();
|
||||
});
|
||||
async function daemon(): Promise<TestPaseoDaemon> {
|
||||
const instance = await createTestPaseoDaemon();
|
||||
async function daemon(
|
||||
options: Parameters<typeof createTestPaseoDaemon>[0] = {},
|
||||
): Promise<TestPaseoDaemon> {
|
||||
const instance = await createTestPaseoDaemon(options);
|
||||
cleanup.push(() => instance.close());
|
||||
return instance;
|
||||
}
|
||||
@@ -1303,3 +1309,79 @@ test("cancelling a logical scope during admission settles it and releases its ow
|
||||
endpoint.close();
|
||||
await delivery.close();
|
||||
});
|
||||
|
||||
test("agent MCP discovers hosts, creates children on self and B, and notifies their parent", async () => {
|
||||
const parentPrompts: string[] = [];
|
||||
const a = await daemon({
|
||||
agentClients: createTestAgentClients({
|
||||
onStartTurn: (prompt, config) => {
|
||||
if (config.title === "Fleet parent") parentPrompts.push(JSON.stringify(prompt));
|
||||
},
|
||||
}),
|
||||
});
|
||||
const b = await daemon();
|
||||
const { paseo } = await api(a);
|
||||
await paseo.hosts.add({ connection: `127.0.0.1:${b.port}`, label: "Fleet B" });
|
||||
const directory = await online(paseo);
|
||||
const workspace = await paseo.workspaces.create({
|
||||
source: { kind: "directory", path: a.staticDir },
|
||||
});
|
||||
const parent = await workspace.agents.create({
|
||||
config: { provider: "claude/claude-test" },
|
||||
title: "Fleet parent",
|
||||
});
|
||||
const mcp = new Client({ name: "fleet-delegation", version: "1" });
|
||||
cleanup.push(() => mcp.close());
|
||||
await mcp.connect(
|
||||
new StreamableHTTPClientTransport(
|
||||
new URL(`http://127.0.0.1:${a.port}/mcp/agents?callerAgentId=${parent.id}`),
|
||||
),
|
||||
);
|
||||
const result = await mcp.callTool({ name: "list_hosts", arguments: {} });
|
||||
expect(result.isError).not.toBe(true);
|
||||
expect(result.structuredContent).toMatchObject({
|
||||
self: { serverId: directory.self.serverId },
|
||||
hosts: [expect.objectContaining({ serverId: directory.hosts[0]!.serverId, label: "Fleet B" })],
|
||||
});
|
||||
const call = async (name: string, args: Record<string, unknown>) => {
|
||||
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}`);
|
||||
});
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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. |
|
||||
|
||||
Reference in New Issue
Block a user