diff --git a/packages/app/e2e/browser/subagent-origin-pane.spec.ts b/packages/app/e2e/browser/subagent-origin-pane.spec.ts index 47b80270fe..b7258ed3de 100644 --- a/packages/app/e2e/browser/subagent-origin-pane.spec.ts +++ b/packages/app/e2e/browser/subagent-origin-pane.spec.ts @@ -3,6 +3,9 @@ import { submitMessage } from "../support/helpers/composer"; import { openSessions, clickSessionRow } from "../support/helpers/archive-tab"; import { seedFleetSubagents, + delegateFleetThroughMcp, + observeFleetRouteOwnership, + readFleetCompletion, connectFleetDirectHost, openFleetParentTrack, readFleetAuthorityState, @@ -16,6 +19,8 @@ import { openAgentRoute, seedMockAgentWorkspace } from "../support/helpers/mock- import { seedWorkspace } from "../support/helpers/seed-client"; import { openSubagentsTrack, seedParentWithSubagent } from "../support/helpers/subagents"; +test.use({ e2eInjectPaseoTools: true }); + const SETTINGS_KEY = "@paseo:app-settings"; async function openParentInRightPane(page: Page, parentId: string, workspaceId: string) { @@ -258,3 +263,66 @@ test("a fleet child retains its origin beside direct access through collision, c await fleet.close(); } }); + +test("an agent delegates through MCP to self and B, opens both ordinary conversations and receives correlated completion", async ({ + page, +}) => { + test.setTimeout(180_000); + await page.setViewportSize({ width: 1400, height: 900 }); + const counts = observeFleetRouteOwnership(page); + const fleet = await delegateFleetThroughMcp(); + try { + await openAgentRoute(page, { + workspaceId: fleet.workspace.workspaceId, + agentId: fleet.parent.id, + }); + console.log("Fleet ordinary parent ownership", await counts()); + await openSubagentsTrack(page); + await expect(page.getByTestId(`subagents-track-row-${fleet.local.id}`)).toContainText( + "MCP child on A", + ); + await expect(page.getByTestId(`subagents-track-row-${fleet.remote.id}`)).toContainText( + "MCP child on B", + ); + await expect(page.getByTestId(`subagents-track-row-${fleet.remote.id}`)).toContainText( + "Fleet B", + ); + await page.getByTestId(`subagents-track-row-${fleet.local.id}`).click(); + await expect(page).toHaveURL((url) => + url.pathname.includes(`/h/${fleet.local.serverId}/workspace/`), + ); + await expect(page.getByRole("textbox", { name: "Message agent..." })).toBeVisible(); + await fleet.expectCompletion(fleet.local); + await submitMessage(page, "UI_MCP_FLEET_LOCAL"); + await expect(page.getByText("UI_MCP_FLEET_LOCAL", { exact: true })).toBeVisible(); + expect( + (await fleet.direct.agents.ref(fleet.local.id).refresh())?.agent.lastUserMessageAt, + ).toBeNull(); + await page.getByTestId(`workspace-tab-agent_${fleet.parent.id}`).click(); + await openSubagentsTrack(page); + await page.getByTestId(`subagents-track-row-${fleet.remote.id}`).click(); + await expect(page).toHaveURL( + (url) => + url.pathname.includes(`/h/${fleet.remote.serverId}/workspace/`) && + url.searchParams.get("origin") === fleet.parent.serverId, + ); + await expect(page.getByRole("textbox", { name: "Message agent..." })).toBeVisible(); + console.log("Fleet destination conversation ownership", await counts()); + await fleet.expectCompletion(fleet.remote); + await submitMessage(page, "UI_MCP_FLEET_REMOTE"); + await expect(page.getByText("UI_MCP_FLEET_REMOTE", { exact: true })).toBeVisible(); + expect( + (await fleet.origin.agents.ref(fleet.remote.id).refresh())?.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 readFleetCompletion(page, fleet.local); + await readFleetCompletion(page, fleet.remote); + } finally { + await fleet.close(); + } +}); diff --git a/packages/app/e2e/support/helpers/subagents.ts b/packages/app/e2e/support/helpers/subagents.ts index 94ee2588d1..d660b539df 100644 --- a/packages/app/e2e/support/helpers/subagents.ts +++ b/packages/app/e2e/support/helpers/subagents.ts @@ -449,6 +449,10 @@ export async function readFleetAuthorityState( .findAll({ queryKey: ["providersSnapshot", destinationId] }) ?? []; return { status: binding?.runtime.getSnapshot(destinationId)?.connectionStatus ?? "absent", + hydrated: binding?.sessionStore.getState().sessions[destinationId]?.hasHydratedWorkspaces, + workspaceIds: [ + ...(binding?.sessionStore.getState().sessions[destinationId]?.workspaces.keys() ?? []), + ], catalogVersion: Math.max(0, ...queries.map((query) => query.state.dataUpdatedAt)), }; }; @@ -477,3 +481,192 @@ export async function expectFleetRouteUnavailable( } else await expect(page.getByRole("button", { name: "Send message", exact: true })).toHaveCount(0); } + +/** Primary children are created by the agent's real MCP endpoint, never the seed SDK. */ +export async function delegateFleetThroughMcp() { + const { Client } = await import("@modelcontextprotocol/sdk/client/index.js"); + const { StreamableHTTPClientTransport } = + await import("@modelcontextprotocol/sdk/client/streamableHttp.js"); + const { getE2EDaemonPort } = await import("./daemon-port"); + const { z } = await import("zod"); + const workspace = await seedWorkspace({ repoPrefix: "mcp-fleet-journey-" }); + const destination = await startIsolatedHostDaemon(`fleet_${randomUUID()}`); + const createApi = await loadPaseoApiFactory(); + const aClient = await connectDaemonClient({ clientIdPrefix: "mcp-fleet-origin" }); + const bClient = await connectDaemonClient({ + clientIdPrefix: "mcp-fleet-observer", + port: destination.port, + }); + const origin = createApi(aClient); + const direct = createApi(bClient); + const mcp = new Client({ name: "fleet-delegating-agent", version: "1.0.0" }); + const close = async () => { + await mcp.close(); + await Promise.allSettled([origin.dispose(), direct.dispose()]); + await Promise.allSettled([aClient.close(), bClient.close()]); + await destination.close(); + await workspace.cleanup(); + }; + const call = async (name: string, args: Record = {}) => { + const result = await mcp.callTool({ name, arguments: args }); + expect(result, `${name} result`).not.toMatchObject({ isError: true }); + return z.record(z.string(), z.unknown()).parse(result.structuredContent); + }; + try { + const config = { provider: "mock/ten-second-stream", modeId: "load-test" }; + const parent = await origin.workspaces + .ref(workspace.workspaceId) + .agents.create({ title: "Fleet MCP parent", config }); + const registered = await origin.hosts.add({ + connection: `127.0.0.1:${destination.port}`, + label: "Fleet B", + }); + await expect + .poll( + async () => + (await origin.hosts.list()).hosts.find( + (host) => host.registrationId === registered.registrationId, + )?.status, + ) + .toBe("online"); + await mcp.connect( + new StreamableHTTPClientTransport( + new URL(`http://127.0.0.1:${getE2EDaemonPort()}/mcp/agents?callerAgentId=${parent.id}`), + ), + ); + const hosts = await call("list_hosts"); + expect(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: "Fleet B workspace", + }); + const targetWorkspaceId = z.string().parse(target.workspaceId); + expect(await call("list_workspaces", { serverId: destination.serverId })).toMatchObject({ + workspaces: expect.arrayContaining([ + expect.objectContaining({ workspaceId: targetWorkspaceId }), + ]), + }); + const createChild = async (serverId: string, workspaceId: string, title: string) => { + const result = await call("create_agent", { + serverId, + workspaceId, + title, + provider: config.provider, + settings: { modeId: config.modeId }, + initialPrompt: `Complete ${title}`, + notifyOnFinish: true, + }); + const agentId = z.string().parse(result.agentId); + const submissionId = z.string().parse(result.submissionId); + expect(await call("get_agent_status", { serverId, agentId })).toMatchObject({ serverId }); + expect(await call("get_agent_activity", { serverId, agentId })).toMatchObject({ serverId }); + expect(await call("list_agents", { serverId })).toMatchObject({ + serverId, + agents: expect.arrayContaining([expect.objectContaining({ id: agentId })]), + }); + return { id: agentId, serverId, workspaceId, submissionId, title }; + }; + const local = await createChild(parent.serverId, workspace.workspaceId, "MCP child on A"); + const remote = await createChild(destination.serverId, targetWorkspaceId, "MCP child on B"); + // Opposite-host collisions are unrelated fixture agents. Neither primary child is seeded. + await direct.agents.create({ + agentId: local.id, + cwd: workspace.repoPath, + title: "Unrelated collision on B", + config, + }); + await origin.agents.create({ + agentId: remote.id, + cwd: workspace.repoPath, + title: "Unrelated collision on A", + config, + }); + expect((await parent.subagents.list()).rows.map((row) => row.child)).toEqual( + expect.arrayContaining([ + { serverId: local.serverId, agentId: local.id }, + { serverId: remote.serverId, agentId: remote.id }, + ]), + ); + return { + workspace, + destination, + parent, + local, + remote, + origin, + direct, + close, + async expectCompletion(child: typeof local) { + await expect + .poll( + async () => { + const timeline = await parent.timeline.refetch(); + return timeline.entries + .flatMap(({ item }) => + item.type === "tool_call" && item.agentMessage ? [item.agentMessage] : [], + ) + .filter( + (message) => + message.event === "finished" && + message.sender.address?.serverId === child.serverId && + message.text.includes(child.submissionId), + ); + }, + { timeout: 30000 }, + ) + .toHaveLength(1); + }, + }; + } catch (error) { + await close(); + throw error; + } +} + +export async function readFleetCompletion( + page: Page, + child: { title: string; submissionId: string }, +) { + await expect( + page.getByRole("button", { name: new RegExp(`Agent finished.*${child.title}`) }), + ).toBeVisible(); +} + +/** Records observed requests and runtime-owned sessions without opening extra scopes. */ +export function observeFleetRouteOwnership(page: Page) { + const requests: Record = {}; + let sockets = 0; + page.on("websocket", (socket) => { + sockets++; + socket.on("framesent", ({ payload }) => { + const request = inspectSessionWireRequest(payload)?.request; + if (!request || typeof request.type !== "string") return; + if (request.subscribe !== undefined || request.type === "hosts.route.open.request") + requests[request.type] = (requests[request.type] ?? 0) + 1; + }); + }); + return async () => ({ + sockets, + cumulativeRequests: { ...requests }, + bindings: await page.evaluate(() => { + const registry = Reflect.get(globalThis, "__paseoHostRuntimeStore") as HostRuntimeStore; + return registry.getAuthorityBindings().map((binding) => { + const id = binding.reference.serverId; + const session = binding.sessionStore.getState().sessions[id]; + return { + reference: binding.reference, + status: binding.runtime.getSnapshot(id)?.connectionStatus, + session: session?.client !== null && session?.client !== undefined, + workspaceReplica: session?.hasHydratedWorkspaces === true, + timelineReplica: session?.viewedTimelineSync !== undefined, + }; + }); + }), + }); +} diff --git a/packages/app/src/runtime/host-authority-provider.test.tsx b/packages/app/src/runtime/host-authority-provider.test.tsx index 6e77e7f5ad..3b894bce2e 100644 --- a/packages/app/src/runtime/host-authority-provider.test.tsx +++ b/packages/app/src/runtime/host-authority-provider.test.tsx @@ -1,13 +1,15 @@ +/** @vitest-environment jsdom */ import { createWorkspaceLayoutStore, useWorkspaceLayoutStoreApi, } from "@/stores/workspace-layout-store"; -import React from "react"; +import React, { act } from "react"; +import { useHasHydratedWorkspaces } from "@/stores/session-store-hooks"; import { createProviderSubagentStore, useProviderSubagentStoreApi, } from "@/subagents/provider-store"; -import { renderToStaticMarkup } from "react-dom/server"; +import { createRoot } from "react-dom/client"; import { expect, it } from "vitest"; import { QueryClient, useQueryClient } from "@tanstack/react-query"; import { createSessionStore, useSessionStoreApi } from "@/stores/session-store"; @@ -15,7 +17,7 @@ import { useCreateFlowStoreApi } from "@/stores/create-flow-store"; import { HostRuntimeStore, useHostRuntimeStore, type HostAuthorityBinding } from "./host-runtime"; import { HostAuthorityProvider } from "./host-authority-provider"; -it("ordinary consumer hooks retain the runtime, store and creation owner of each authority", () => { +it("ordinary consumer hooks retain the runtime, store and creation owner of each authority", async () => { const binding = (originServerId: string): HostAuthorityBinding => { const sessionStore = createSessionStore(); return { @@ -31,7 +33,12 @@ it("ordinary consumer hooks retain the runtime, store and creation owner of each }; const direct = binding("B"); const routed = binding("A"); + for (const owner of [direct, routed]) { + owner.sessionStore.getState().initializeSession("B", null); + owner.sessionStore.getState().setHasHydratedWorkspaces("B", owner === routed); + } function OrdinaryConsumer({ expected }: { expected: HostAuthorityBinding }) { + expect(useHasHydratedWorkspaces("B")).toBe(expected === routed); const runtime = useHostRuntimeStore(); const store = useSessionStoreApi(); const creates = useCreateFlowStoreApi(); @@ -43,14 +50,22 @@ it("ordinary consumer hooks retain the runtime, store and creation owner of each expect(creates).toBe(expected.sessionStore.createFlows); return null; } - renderToStaticMarkup( - <> - - - - - - - , - ); + const container = document.createElement("div"); + const root = createRoot(container); + try { + await act(async () => + root.render( + <> + + + + + + + , + ), + ); + } finally { + await act(async () => root.unmount()); + } }); diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index 8514015cc2..c6aed21296 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -1,3 +1,4 @@ +import type { HostDirectory } from "@getpaseo/protocol/hosts"; import { createWorkspaceLayoutStore, type WorkspaceLayoutStoreApi, @@ -1548,6 +1549,7 @@ export class HostRuntimeStore { { binding: HostAuthorityBinding; refs: number; close(): Promise } >(); private bindingSnapshot: readonly HostAuthorityBinding[] = []; + private fleetDirectory: HostDirectory | null = null; private serverListeners = new Map void>>(); private globalListeners = new Set<() => void>(); private hostListListeners = new Set<() => void>(); @@ -1764,6 +1766,9 @@ export class HostRuntimeStore { agentDirectoryError: failure, }), }); + // A bound workspace needs its own directory, including destinations absent + // from the global sidebar host list. DirectorySync shares all demand sources. + const releaseDirectoryDemand = scoped.acquireDirectoryDemand(serverId); let stopped = false; let origin: DaemonClient | null = null; let controller = new AbortController(); @@ -1876,11 +1881,18 @@ export class HostRuntimeStore { if (origin && origin.getConnectionState().status === "connected" && !directory) { try { directory = origin.observeHosts(); - const changed = () => { + const changed = (value: HostDirectory) => { + scoped.fleetDirectory = value; + scoped.emit(serverId); directoryRevision++; acquire(); }; - directory.subscribe({ snapshot: changed, update: changed }); + directory.subscribe({ + snapshot: changed, + update: (message) => { + if (message.type === "hosts.changed") changed(message.payload.directory); + }, + }); void directory.ready.catch(() => {}); } catch (failure) { publish(null, { status: "disconnected" }, toErrorMessage(failure)); @@ -1909,6 +1921,7 @@ export class HostRuntimeStore { stopTarget?.(); rejectReady(new Error("Host authority binding is disposed")); scoped.syncServerDataClient(serverId, null); + releaseDirectoryDemand(); scoped.directorySyncByServer.get(serverId)?.dispose(); scoped.sessionStore.getState().sessions[serverId]?.viewedTimelineSync?.dispose(); for (const agentId of scoped.sessionStore @@ -1934,6 +1947,12 @@ export class HostRuntimeStore { // --- Host registry --- + getHostLabel(serverId: string): string { + const directory = this.fleetDirectory; + if (directory?.self.serverId === serverId) return directory.self.label; + return directory?.hosts.find((host) => host.serverId === serverId)?.label ?? serverId; + } + getHosts(): HostProfile[] { return this.hosts; } diff --git a/packages/app/src/stores/session-store-hooks/index.ts b/packages/app/src/stores/session-store-hooks/index.ts index 89978811ec..acb6748d19 100644 --- a/packages/app/src/stores/session-store-hooks/index.ts +++ b/packages/app/src/stores/session-store-hooks/index.ts @@ -20,7 +20,7 @@ import { workspaceEqualityFns, type WorkspaceStructure, } from "./selectors"; -import { useSessionStore, type WorkspaceDescriptor } from "../session-store"; +import { useSessionStoreApi, type WorkspaceDescriptor } from "../session-store"; import type { DesktopBadgeWorkspaceStatus } from "@/utils/desktop-badge-state"; // These are the ONLY supported ways to read workspaces from the session store. @@ -38,7 +38,7 @@ export function useWorkspace( workspaceId: string | null, ): WorkspaceDescriptor | null { return useStoreWithEqualityFn( - useSessionStore, + useSessionStoreApi(), (state) => selectWorkspace(state, serverId, workspaceId), workspaceEqualityFns.identity, ); @@ -50,7 +50,7 @@ export function useWorkspaceFields( project: (w: WorkspaceDescriptor) => T, ): T | null { return useStoreWithEqualityFn( - useSessionStore, + useSessionStoreApi(), (state) => selectWorkspaceFields(state, serverId, workspaceId, project), workspaceEqualityFns.deep, ); @@ -58,7 +58,7 @@ export function useWorkspaceFields( export function useWorkspaceExists(serverId: string | null, workspaceId: string | null): boolean { return useStoreWithEqualityFn( - useSessionStore, + useSessionStoreApi(), (state) => selectWorkspaceExists(state, serverId, workspaceId), workspaceEqualityFns.identity, ); @@ -66,7 +66,7 @@ export function useWorkspaceExists(serverId: string | null, workspaceId: string export function useHasHydratedWorkspaces(serverId: string | null): boolean { return useStoreWithEqualityFn( - useSessionStore, + useSessionStoreApi(), (state) => selectHasHydratedWorkspaces(state, serverId), workspaceEqualityFns.identity, ); @@ -74,7 +74,7 @@ export function useHasHydratedWorkspaces(serverId: string | null): boolean { export function useHydratedWorkspaceServerIds(serverIds: string[]): string[] { return useStoreWithEqualityFn( - useSessionStore, + useSessionStoreApi(), (state) => selectHydratedWorkspaceServerIds(state, serverIds), workspaceEqualityFns.deep, ); @@ -82,7 +82,7 @@ export function useHydratedWorkspaceServerIds(serverIds: string[]): string[] { export function useWorkspaceDirectoryServerIds(serverIds: string[]): string[] { return useStoreWithEqualityFn( - useSessionStore, + useSessionStoreApi(), (state) => selectWorkspaceDirectoryServerIds(state, serverIds), workspaceEqualityFns.deep, ); @@ -93,7 +93,7 @@ export function useWorkspaceDirectory( workspaceId: string | null, ): string | null { return useStoreWithEqualityFn( - useSessionStore, + useSessionStoreApi(), (state) => selectWorkspaceDirectory(state, serverId, workspaceId), workspaceEqualityFns.identity, ); @@ -105,7 +105,7 @@ export function useWorkspaceStructure(serverIds: string[]): WorkspaceStructure { [serverIds], ); const projects = useStoreWithEqualityFn( - useSessionStore, + useSessionStoreApi(), selectProjects, workspaceEqualityFns.deep, ); @@ -133,7 +133,7 @@ export function useWorkspaceStructure(serverIds: string[]): WorkspaceStructure { export function useWorkspaceKeys(serverId: string | null): string[] { return useStoreWithEqualityFn( - useSessionStore, + useSessionStoreApi(), (state) => selectWorkspaceKeys(state, serverId), workspaceEqualityFns.deep, ); @@ -141,7 +141,7 @@ export function useWorkspaceKeys(serverId: string | null): string[] { export function useRecommendedProjectPaths(serverId: string | null): string[] { return useStoreWithEqualityFn( - useSessionStore, + useSessionStoreApi(), (state) => selectRecommendedProjectPaths(state, serverId), workspaceEqualityFns.deep, ); @@ -149,7 +149,7 @@ export function useRecommendedProjectPaths(serverId: string | null): string[] { export function useHasWorkspaces(serverId: string | null): boolean { return useStoreWithEqualityFn( - useSessionStore, + useSessionStoreApi(), (state) => selectHasWorkspaces(state, serverId), workspaceEqualityFns.identity, ); @@ -157,7 +157,7 @@ export function useHasWorkspaces(serverId: string | null): boolean { export function useWorkspaceStatusesForBadges(): DesktopBadgeWorkspaceStatus[] { return useStoreWithEqualityFn( - useSessionStore, + useSessionStoreApi(), (state) => selectWorkspaceStatusesForBadges(state), workspaceEqualityFns.deep, ); diff --git a/packages/app/src/subagents/track.tsx b/packages/app/src/subagents/track.tsx index 67c22ed048..d22b9b8aee 100644 --- a/packages/app/src/subagents/track.tsx +++ b/packages/app/src/subagents/track.tsx @@ -1,6 +1,13 @@ +import { useHostRuntimeStore } from "@/runtime/host-runtime"; import { HostAuthorityRoute } from "@/runtime/host-authority-route"; import { useAgentSnapshotFactsLine } from "@/panels/agent-facts-line"; -import { useCallback, useMemo, type ReactElement, type ReactNode } from "react"; +import { + useCallback, + useMemo, + useSyncExternalStore, + type ReactElement, + type ReactNode, +} from "react"; import { Text, View } from "react-native"; import { useTranslation } from "react-i18next"; import { Archive, Unlink } from "lucide-react-native"; @@ -297,7 +304,22 @@ function SubagentsTrackRow({ function ManagedSubagentFacts({ row }: { row: PaseoSubagentRow }) { const text = useAgentSnapshotFactsLine(row.address.serverId, row.facts); - return ; + const runtime = useHostRuntimeStore(); + const subscribe = useCallback( + (changed: () => void) => runtime.subscribe(row.address.serverId, changed), + [runtime, row.address.serverId], + ); + const read = useCallback( + () => runtime.getHostLabel(row.address.serverId), + [runtime, row.address.serverId], + ); + const host = useSyncExternalStore(subscribe, read, read); + return ( + <> + + + + ); } function SubagentFacts({ rowId, text }: { rowId: string; text: string | null }) { diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index 2edcf98ad0..a2a3a89656 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -1715,7 +1715,10 @@ export class DaemonClient { } observeAgentRun(agentId: string, submissionId: string) { this.requireFleetSubagents(); - return this.owned.observe(async (accept) => { + return this.owned.observe<{ + run: import("@getpaseo/protocol/agent-runs").AgentRunRecord | null; + subscriptionId: string; + }>(async (accept) => { const result = await this.sendCorrelatedSessionRequest({ message: { type: "agent.runs.get.request", agentId, submissionId, observe: true }, responseType: "agent.runs.get.response", diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index ba03a2f63e..b4d7e80fbe 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -364,6 +364,7 @@ export interface PaseoAgentTimelineHandle { } export type PaseoSubagentCreateOptions = Omit & { + notifyOnFinish?: boolean; serverId?: string; workspaceId?: string; workspace?: import("@getpaseo/protocol/messages").SubagentCreateInput["workspace"]; @@ -683,7 +684,7 @@ export function createPaseoApi( return daemonClient.listSubagents(parentId); }, observe: () => own(() => daemonClient.observeSubagents(parentId)), - create: async ({ serverId, idempotencyKey, workspace, ...input }) => { + create: async ({ serverId, idempotencyKey, workspace, notifyOnFinish, ...input }) => { if (lifetime.signal.aborted) throw new Error("Paseo API is disposed"); const normalized = toDaemonAgentCreateOptions( { ...input, cwd: input.cwd ?? "" }, @@ -693,6 +694,7 @@ export function createPaseoApi( serverId: serverId ?? requireServerIdentity(daemonClient.getServerId()), idempotencyKey: idempotencyKey ?? safeRandomId(), workspace, + notifyOnFinish, input: { ...normalized, labels: input.labels ?? {}, config: normalized.config! }, }); subagentRuntime ??= hosts.runtime(); diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 2fd17816d3..ea5d90d2a7 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -3269,6 +3269,7 @@ export const SubagentDirectorySchema = z.object({ }); export type SubagentDirectory = z.infer; export const SubagentCreateInputSchema = z.object({ + notifyOnFinish: z.boolean().optional(), serverId: z.string().min(1), idempotencyKey: z.string().min(1).max(512), workspace: WorkspaceCreateRequestSchema.omit({ diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 5cc050c9ec..8b9159a91a 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -8633,7 +8633,7 @@ test("archiveAgent persists archivedAt and updatedAt before emitting closed stat expect(lifecycles.slice(-2)).toEqual(["idle", "closed"]); }); -test("fires onAgentArchived for archived parent and cascaded children", async () => { +test("fires onAgentArchived for archived parent and cascaded children for an execution-owned parent", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-archived-hook-cascade-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); @@ -8655,7 +8655,10 @@ test("fires onAgentArchived for archived parent and cascaded children", async () title: "Parent", }, undefined, - { workspaceId: undefined }, + { + owner: { kind: "daemon", daemonId: "hub", executionId: "archive-test" }, + workspaceId: undefined, + }, ); const liveChild = await manager.createAgent( { provider: "codex", cwd: workdir, title: "Child" }, @@ -8884,7 +8887,7 @@ test("unarchiveSnapshot keeps the stored record archived when native unarchive f expect(client.unarchivedHandles).toHaveLength(1); }); -test("archiveAgent cascade archives in-memory children with the full archive contract", async () => { +test("archiveAgent cascade archives in-memory children with the full archive contract for an execution-owned parent", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-cascade-contract-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); @@ -8904,7 +8907,10 @@ test("archiveAgent cascade archives in-memory children with the full archive con title: "Parent", }, undefined, - { workspaceId: undefined }, + { + owner: { kind: "daemon", daemonId: "hub", executionId: "archive-test" }, + workspaceId: undefined, + }, ); const child = await manager.createAgent( { @@ -8936,7 +8942,7 @@ test("archiveAgent cascade archives in-memory children with the full archive con expect(storedUnrelated?.archivedAt).toBeUndefined(); }); -test("archiveAgent detaches an open same-workspace child instead of cascading", async () => { +test("archiveAgent detaches an open same-workspace child instead of cascading for an execution-owned parent", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-cascade-open-child-")); const storage = new AgentStorage(join(workdir, "agents"), logger); const manager = new AgentManager({ @@ -8946,9 +8952,16 @@ test("archiveAgent detaches an open same-workspace child instead of cascading", logger, }); const parent = await manager.createAgent( - { provider: "codex", cwd: workdir, title: "Parent" }, + { + provider: "codex", + cwd: workdir, + title: "Parent", + }, undefined, - { workspaceId: "workspace-a" }, + { + owner: { kind: "daemon", daemonId: "hub", executionId: "archive-test" }, + workspaceId: "workspace-a", + }, ); const child = await manager.createAgent( { provider: "codex", cwd: workdir, title: "Open child" }, @@ -8973,7 +8986,7 @@ test("archiveAgent detaches an open same-workspace child instead of cascading", expect(manager.getAgent(child.id)?.id).toBe(child.id); }); -test("archiveAgent detaches a cross-workspace child even when its tab is closed", async () => { +test("archiveAgent detaches a cross-workspace child even when its tab is closed for an execution-owned parent", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-cascade-cross-workspace-child-")); const storage = new AgentStorage(join(workdir, "agents"), logger); const manager = new AgentManager({ @@ -8983,9 +8996,16 @@ test("archiveAgent detaches a cross-workspace child even when its tab is closed" logger, }); const parent = await manager.createAgent( - { provider: "codex", cwd: workdir, title: "Parent" }, + { + provider: "codex", + cwd: workdir, + title: "Parent", + }, undefined, - { workspaceId: "workspace-a" }, + { + owner: { kind: "daemon", daemonId: "hub", executionId: "archive-test" }, + workspaceId: "workspace-a", + }, ); const child = await manager.createAgent( { provider: "codex", cwd: workdir, title: "Cross-workspace child" }, @@ -9007,7 +9027,7 @@ test("archiveAgent detaches a cross-workspace child even when its tab is closed" expect(storedChild?.labels[PARENT_AGENT_ID_LABEL]).toBeUndefined(); }); -test("archiveAgent re-reads a child before deciding whether to cascade", async () => { +test("archiveAgent re-reads a child before deciding whether to cascade for an execution-owned parent", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-cascade-fresh-child-")); class ChildOpensAfterCascadeListStorage extends AgentStorage { @@ -9037,9 +9057,16 @@ test("archiveAgent re-reads a child before deciding whether to cascade", async ( logger, }); const parent = await manager.createAgent( - { provider: "codex", cwd: workdir, title: "Parent" }, + { + provider: "codex", + cwd: workdir, + title: "Parent", + }, undefined, - { workspaceId: "workspace-a" }, + { + owner: { kind: "daemon", daemonId: "hub", executionId: "archive-test" }, + workspaceId: "workspace-a", + }, ); const child = await manager.createAgent( { provider: "codex", cwd: workdir, title: "Late-open child" }, @@ -9061,7 +9088,7 @@ test("archiveAgent re-reads a child before deciding whether to cascade", async ( expect(storedChild?.labels[PARENT_AGENT_ID_LABEL]).toBeUndefined(); }); -test("archiveAgent cannot overtake a received child open-tab update", async () => { +test("archiveAgent cannot overtake a received child open-tab update for an execution-owned parent", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-cascade-open-race-")); const markerWriteStarted = deferred(); const releaseMarkerWrite = deferred(); @@ -9089,9 +9116,16 @@ test("archiveAgent cannot overtake a received child open-tab update", async () = logger, }); const parent = await manager.createAgent( - { provider: "codex", cwd: workdir, title: "Parent" }, + { + provider: "codex", + cwd: workdir, + title: "Parent", + }, undefined, - { workspaceId: "workspace-a" }, + { + owner: { kind: "daemon", daemonId: "hub", executionId: "archive-test" }, + workspaceId: "workspace-a", + }, ); const child = await manager.createAgent( { provider: "codex", cwd: workdir, title: "Opening child" }, @@ -9116,7 +9150,7 @@ test("archiveAgent cannot overtake a received child open-tab update", async () = expect(storedChild?.labels[PARENT_AGENT_ID_LABEL]).toBeUndefined(); }); -test("archiveAgent cascade closes a running child runtime", async () => { +test("archiveAgent cascade closes a running child runtime for an execution-owned parent", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-cascade-running-child-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); @@ -9166,7 +9200,10 @@ test("archiveAgent cascade closes a running child runtime", async () => { title: "Parent", }, undefined, - { workspaceId: undefined }, + { + owner: { kind: "daemon", daemonId: "hub", executionId: "archive-test" }, + workspaceId: undefined, + }, ); const child = await manager.createAgent( { @@ -9206,7 +9243,7 @@ test("archiveAgent cascade closes a running child runtime", async () => { expect(childLifecycleEvents).toContain("closed"); }); -test("archiveAgent cascade archives off-memory children with the full archive contract", async () => { +test("archiveAgent cascade archives off-memory children with the full archive contract for an execution-owned parent", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-cascade-off-memory-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); @@ -9225,7 +9262,10 @@ test("archiveAgent cascade archives off-memory children with the full archive co title: "Parent", }, undefined, - { workspaceId: undefined }, + { + owner: { kind: "daemon", daemonId: "hub", executionId: "archive-test" }, + workspaceId: undefined, + }, ); const child = await manager.createAgent( { @@ -9246,7 +9286,7 @@ test("archiveAgent cascade archives off-memory children with the full archive co expectArchivedAgentRecord(await storage.get(child.id), "idle"); }); -test("archiveAgent cascade notifies subscribers for in-memory and off-memory children", async () => { +test("archiveAgent cascade notifies subscribers for in-memory and off-memory children for an execution-owned parent", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-cascade-notifications-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); @@ -9265,7 +9305,10 @@ test("archiveAgent cascade notifies subscribers for in-memory and off-memory chi title: "Parent", }, undefined, - { workspaceId: undefined }, + { + owner: { kind: "daemon", daemonId: "hub", executionId: "archive-test" }, + workspaceId: undefined, + }, ); const inMemoryChild = await manager.createAgent( { @@ -9314,7 +9357,7 @@ test("archiveAgent cascade notifies subscribers for in-memory and off-memory chi }); }); -test("archiveAgent cascade surfaces partial child archive failures", async () => { +test("archiveAgent cascade surfaces partial child archive failures for an execution-owned parent", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-cascade-partial-failure-")); const storagePath = join(workdir, "agents"); let failingChildId: string | null = null; @@ -9344,7 +9387,10 @@ test("archiveAgent cascade surfaces partial child archive failures", async () => title: "Parent", }, undefined, - { workspaceId: undefined }, + { + owner: { kind: "daemon", daemonId: "hub", executionId: "archive-test" }, + workspaceId: undefined, + }, ); const child = await manager.createAgent( { @@ -10242,7 +10288,7 @@ test("idle agents remain resident until an explicit lifecycle action closes them } }); -test("archiving a closed parent still cascades to its managed children", async () => { +test("archiving a closed parent still cascades to its execution-owned children", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-closed-parent-archive-")); const storage = new AgentStorage(join(workdir, "agents"), logger); const manager = new AgentManager({ @@ -10254,9 +10300,16 @@ test("archiving a closed parent still cascades to its managed children", async ( try { const parent = await manager.createAgent( - { provider: "codex", cwd: workdir, title: "Closed parent" }, + { + provider: "codex", + cwd: workdir, + title: "Closed parent", + }, undefined, - { workspaceId: undefined }, + { + owner: { kind: "daemon", daemonId: "hub", executionId: "archive-test" }, + workspaceId: undefined, + }, ); const child = await manager.createAgent( { provider: "codex", cwd: workdir, title: "Managed child" }, diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index c7ca3d22db..54c81af398 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -1900,12 +1900,10 @@ export class AgentManager { return { archivedAt }; } - // Children created via the MCP `create_agent` tool carry the parent-agent-id - // label pointing back at the caller. Archiving the parent cascades to those - // children so subagent fleets don't outlive their orchestrator. Detached - // handoff agents omit this label, so they stand outside the cascade. + // Managed membership and settlement belong to Subagents. This cascade retains + // only private/internal and execution-owned lifecycles. private async cascadeArchiveChildren( - parent: Pick, + parent: Pick, ): Promise { const registry = this.registry; if (!registry) { @@ -1914,7 +1912,7 @@ export class AgentManager { const parentAgentId = parent.id; const records = await registry.list(); for (const record of records) { - if (this.managedChildLifecycleOwner && !record.internal && !record.owner) continue; + if (!parent.internal && !parent.owner && !record.internal && !record.owner) continue; if (record.archivedAt) { continue; } diff --git a/packages/server/src/server/agent/agent-prompt.ts b/packages/server/src/server/agent/agent-prompt.ts index 0451770eb6..63e20018e8 100644 --- a/packages/server/src/server/agent/agent-prompt.ts +++ b/packages/server/src/server/agent/agent-prompt.ts @@ -439,7 +439,7 @@ interface FinishNotificationBodyInput { permissionRequest?: AgentPermissionRequest; } -function formatFinishNotificationBody(params: FinishNotificationBodyInput): string { +export function formatFinishNotificationBody(params: FinishNotificationBodyInput): string { const statusLine = `Agent ${params.childAgentId} (${params.title}) ${params.reason}.`; const sections = [statusLine]; if (params.reason === "needs permission" && params.permissionRequest) { diff --git a/packages/server/src/server/agent/create-agent/create.test.ts b/packages/server/src/server/agent/create-agent/create.test.ts index e1c77bca91..da30c72275 100644 --- a/packages/server/src/server/agent/create-agent/create.test.ts +++ b/packages/server/src/server/agent/create-agent/create.test.ts @@ -294,7 +294,7 @@ test("session create stamps the new worktree's workspaceId when a setup continua } }); -test("mcp create stamps the new worktree's workspaceId, not the parent's", async () => { +test("standalone creation stamps the new worktree's workspaceId, not the parent's", async () => { const workdir = mkdtempSync(join(tmpdir(), "create-agent-test-")); const storage = new AgentStorage(join(workdir, "agents"), logger); const agentManager = createRealAgentManager(storage); @@ -314,6 +314,21 @@ test("mcp create stamps the new worktree's workspaceId, not the parent's", async }, ); + await expect( + createAgentCommand( + { agentManager, agentStorage: storage, logger, providerSnapshotManager }, + { + kind: "mcp", + callerAgentId: parent.id, + provider: "codex/gpt-5.4", + title: "Unowned managed child", + initialPrompt: "must not start", + notifyOnFinish: false, + }, + ), + ).rejects.toThrow("Managed creation requires the parent.subagents capability"); + expect((await storage.list()).map((agent) => agent.id)).toEqual([parent.id]); + const { snapshot: child } = await createAgentCommand( { agentManager, @@ -332,6 +347,7 @@ test("mcp create stamps the new worktree's workspaceId, not the parent's", async initialPrompt: "do the thing", notifyOnFinish: false, callerAgentId: parent.id, + detached: true, worktree: { worktreeName: "feature", baseBranch: "main" }, }, ); diff --git a/packages/server/src/server/agent/create-agent/create.ts b/packages/server/src/server/agent/create-agent/create.ts index d435752925..7021da57ca 100644 --- a/packages/server/src/server/agent/create-agent/create.ts +++ b/packages/server/src/server/agent/create-agent/create.ts @@ -175,10 +175,22 @@ interface ResolvedCreateAgent { createdWorktree?: CreatePaseoWorktreeWorkflowResult; } +function assertCreationOwnership(input: CreateAgentCommandInput): void { + if ( + input.kind === "mcp" && + input.callerAgentId && + !input.detached && + !input.internal && + !input.owner + ) + throw new Error("Managed creation requires the parent.subagents capability"); +} + export async function createAgentCommand( dependencies: CreateAgentCommandDependencies, input: CreateAgentCommandInput, ): Promise { + assertCreationOwnership(input); const resolved = input.kind === "session" ? await resolveSessionCreateAgent(dependencies, input) @@ -341,7 +353,10 @@ async function resolveMcpCreateAgent( caller: parentAgent ? { id: parentAgent.id, cwd: parentAgent.cwd, workspaceId: parentAgent.workspaceId } : null, - labels: input.labels, + labels: + input.internal && parentAgent + ? { ...input.labels, "paseo.parent-agent-id": parentAgent.id } + : input.labels, childAgentDefaultLabels: input.callerContext?.childAgentDefaultLabels, legacyDetached: input.detached ?? false, resolveWorkspace: async (workspaceId) => ({ workspaceId, cwd: resolvedCwd }), diff --git a/packages/server/src/server/agent/create-agent/intent.test.ts b/packages/server/src/server/agent/create-agent/intent.test.ts index 38f390d3f8..40bedd1152 100644 --- a/packages/server/src/server/agent/create-agent/intent.test.ts +++ b/packages/server/src/server/agent/create-agent/intent.test.ts @@ -3,7 +3,7 @@ import { PARENT_AGENT_ID_LABEL } from "@getpaseo/protocol/agent-labels"; import { resolveCreateAgentIntent } from "./intent.js"; describe("resolveCreateAgentIntent", () => { - it("keeps caller parentage when an explicit workspace changes placement", async () => { + it("resolves placement without manufacturing relationship membership", async () => { const intent = await resolveCreateAgentIntent({ explicitWorkspaceId: "workspace-isolated", caller: { id: "parent-agent", cwd: "/parent", workspaceId: "workspace-parent" }, @@ -15,10 +15,8 @@ describe("resolveCreateAgentIntent", () => { expect(intent).toEqual({ workspaceId: "workspace-isolated", cwd: "/isolated", - parentAgentId: "parent-agent", labels: { purpose: "review", - [PARENT_AGENT_ID_LABEL]: "parent-agent", }, }); }); @@ -36,7 +34,6 @@ describe("resolveCreateAgentIntent", () => { expect(intent.workspaceId).toBe("workspace-parent"); expect(intent.cwd).toBe("/parent"); - expect(intent.parentAgentId).toBe("parent-agent"); expect(createCount).toBe(0); }); @@ -50,7 +47,6 @@ describe("resolveCreateAgentIntent", () => { expect(intent).toEqual({ workspaceId: "workspace-created", cwd: "/created", - parentAgentId: null, labels: {}, }); }); @@ -67,7 +63,6 @@ describe("resolveCreateAgentIntent", () => { expect(intent).toEqual({ workspaceId: "workspace-parent", cwd: "/parent", - parentAgentId: null, labels: {}, }); }); diff --git a/packages/server/src/server/agent/create-agent/intent.ts b/packages/server/src/server/agent/create-agent/intent.ts index c811144b74..24e3e58ce8 100644 --- a/packages/server/src/server/agent/create-agent/intent.ts +++ b/packages/server/src/server/agent/create-agent/intent.ts @@ -14,7 +14,6 @@ export interface CreateAgentPlacement { export interface CreateAgentIntent { workspaceId: string; cwd: string; - parentAgentId: string | null; labels: Record; } @@ -27,12 +26,10 @@ export async function resolveCreateAgentIntent(input: { createWorkspace: () => Promise; legacyDetached?: boolean; }): Promise { - const parentAgentId = input.legacyDetached ? null : (input.caller?.id ?? null); const placement = await resolvePlacement(input); const labels = { ...input.childAgentDefaultLabels, ...input.labels, - ...(parentAgentId ? { [PARENT_AGENT_ID_LABEL]: parentAgentId } : {}), }; // COMPAT(detachedCreate): legacy callers may still request detached creation. @@ -42,7 +39,7 @@ export async function resolveCreateAgentIntent(input: { delete labels[PARENT_AGENT_ID_LABEL]; } - return { ...placement, parentAgentId, labels }; + return { ...placement, labels }; } async function resolvePlacement(input: { diff --git a/packages/server/src/server/agent/mcp-parity.e2e.test.ts b/packages/server/src/server/agent/mcp-parity.e2e.test.ts index 722a01c0e4..62c8522baa 100644 --- a/packages/server/src/server/agent/mcp-parity.e2e.test.ts +++ b/packages/server/src/server/agent/mcp-parity.e2e.test.ts @@ -316,6 +316,402 @@ afterAll(async () => { await rm(tempRoot, { recursive: true, force: true }); }); +test("agent MCP delegates to self and selected host and receives qualified run completion", async () => { + const { createHostClient } = await import("@getpaseo/client/node"); + const { createPaseoApi } = await import("@getpaseo/client"); + const parentPrompts: string[] = []; + const a = await createTestPaseoDaemon({ + agentClients: createTestAgentClients({ + onStartTurn(prompt, config) { + if (config.title === "Fleet MCP parent" && typeof prompt === "string") + parentPrompts.push(prompt); + }, + }), + }); + const b = await createTestPaseoDaemon(); + const client = createHostClient(`127.0.0.1:${a.port}`, { + clientId: crypto.randomUUID(), + reconnect: { enabled: false }, + }); + await client.connect(); + const api = createPaseoApi(client); + let scoped: McpClient | undefined; + try { + const parent = await api.agents.create({ + cwd: a.staticDir, + config: { provider: "claude/claude-test-model" }, + title: "Fleet MCP parent", + }); + const registration = await api.hosts.add({ + connection: `127.0.0.1:${b.port}`, + label: "Worker B", + }); + await expect + .poll( + async () => + (await api.hosts.list()).hosts.find( + (host) => host.registrationId === registration.registrationId, + )?.status, + ) + .toBe("online"); + scoped = await createMcpClient( + buildExpectedAgentMcpUrl({ host: "127.0.0.1", port: a.port, agentId: parent.id }), + ); + const hosts = await callToolStructured(scoped, "list_hosts"); + expect(hosts).toMatchObject({ + self: { serverId: a.daemon.getServerId() }, + hosts: expect.arrayContaining([ + expect.objectContaining({ serverId: b.daemon.getServerId() }), + ]), + }); + const serverId = b.daemon.getServerId(); + const workspace = await callToolStructured(scoped, "create_workspace", { + serverId, + isolation: "local", + path: b.staticDir, + }); + expect(await callToolStructured(scoped, "list_workspaces", { serverId })).toMatchObject({ + serverId, + workspaces: expect.arrayContaining([ + expect.objectContaining({ workspaceId: workspace.workspaceId }), + ]), + }); + const children: StructuredContent[] = []; + for (const destination of [ + { serverId: a.daemon.getServerId() }, + { serverId, workspaceId: workspace.workspaceId }, + ]) { + const child = await callToolStructured(scoped, "create_agent", { + ...destination, + title: "Fleet MCP child", + provider: "claude/claude-test-model", + initialPrompt: "Finish delegated fleet work", + }); + expect(child).toMatchObject({ + serverId: destination.serverId, + submissionId: expect.any(String), + }); + children.push(child); + expect( + await callToolStructured(scoped, "get_agent_status", { + serverId: destination.serverId, + agentId: child.agentId, + }), + ).toMatchObject({ serverId: destination.serverId }); + expect( + await callToolStructured(scoped, "get_agent_activity", { + serverId: destination.serverId, + agentId: child.agentId, + }), + ).toMatchObject({ serverId: destination.serverId, content: expect.any(String) }); + expect( + await callToolStructured(scoped, "list_agents", { serverId: destination.serverId }), + ).toMatchObject({ + serverId: destination.serverId, + agents: expect.arrayContaining([expect.objectContaining({ id: child.agentId })]), + }); + } + expect((await parent.subagents.list()).rows.map((row) => row.child)).toEqual( + expect.arrayContaining( + children.map((child) => ({ serverId: child.serverId, agentId: child.agentId })), + ), + ); + await expect.poll(() => parentPrompts.length).toBe(2); + for (const child of children) { + const notification = parentPrompts + .map((text) => parseAgentMessage(text)) + .find((message) => message?.source?.serverId === child.serverId); + expect(notification?.source).toMatchObject({ + kind: "agent-notification", + serverId: child.serverId, + agentId: child.agentId, + event: "finished", + }); + expect(notification?.text).toContain(str(child.submissionId)); + expect(notification?.text).toContain("checkpoint"); + } + } finally { + await scoped?.close(); + await api.dispose(); + await client.close(); + await a.close(); + await b.close(); + } +}); + +test("fleet MCP rejects missing destination placement and disabled or offline hosts without local effects", async () => { + const { createHostClient } = await import("@getpaseo/client/node"); + const { createPaseoApi } = await import("@getpaseo/client"); + const b = await createTestPaseoDaemon(); + const sourceClient = createHostClient(`127.0.0.1:${daemonHandle.port}`, { + clientId: crypto.randomUUID(), + reconnect: { enabled: false }, + }); + const targetClient = createHostClient(`127.0.0.1:${b.port}`, { + clientId: crypto.randomUUID(), + reconnect: { enabled: false }, + }); + await Promise.all([sourceClient.connect(), targetClient.connect()]); + const source = createPaseoApi(sourceClient); + const target = createPaseoApi(targetClient); + try { + const registration = await source.hosts.add({ connection: `127.0.0.1:${b.port}` }); + await expect + .poll( + async () => + (await source.hosts.list()).hosts.find( + (host) => host.registrationId === registration.registrationId, + )?.status, + ) + .toBe("online"); + const serverId = b.daemon.getServerId(); + const workspace = await callToolStructured(agentScopedClient, "create_workspace", { + serverId, + isolation: "local", + path: b.staticDir, + }); + const input = { + serverId, + workspaceId: workspace.workspaceId, + title: "Must not fall back", + provider: "claude/claude-test-model", + initialPrompt: "Never execute rejected work", + }; + const counts = async () => [ + (await source.agents.list()).entries.length, + (await target.agents.list()).entries.length, + ]; + const before = await counts(); + const noWorkspace = await agentScopedClient.callTool({ + name: "create_agent", + args: { ...input, workspaceId: undefined }, + }); + expect(noWorkspace).toMatchObject({ + isError: true, + structuredContent: { serverId, code: "workspace_required" }, + }); + await expectToolError( + agentScopedClient, + "create_agent", + { ...input, provider: undefined }, + /provider/, + ); + expect(await counts()).toEqual(before); + const unparented = await callToolStructured(topLevelClient, "create_agent", { + ...input, + title: "Top level fleet root", + }); + const snapshot = (await target.agents.ref(str(unparented.agentId)).refresh())?.agent; + expect(snapshot?.parent ?? null).toBeNull(); + expect(snapshot?.managedRelationship).toBeNull(); + await source.hosts.update(registration.registrationId, { enabled: false }); + expect(await agentScopedClient.callTool({ name: "create_agent", args: input })).toMatchObject({ + isError: true, + structuredContent: { serverId, code: "host_disabled" }, + }); + expect(await counts()).toEqual([before[0], before[1]! + 1]); + await source.hosts.update(registration.registrationId, { enabled: true }); + await expect + .poll( + async () => + (await source.hosts.list()).hosts.find( + (host) => host.registrationId === registration.registrationId, + )?.status, + ) + .toBe("online"); + await b.close(); + await expect + .poll( + async () => + (await source.hosts.list()).hosts.find( + (host) => host.registrationId === registration.registrationId, + )?.status, + ) + .not.toBe("online"); + expect(await agentScopedClient.callTool({ name: "create_agent", args: input })).toMatchObject({ + isError: true, + structuredContent: { serverId }, + }); + expect((await source.agents.list()).entries.length).toBe(before[0]); + } finally { + await source.dispose(); + await target.dispose(); + await sourceClient.close(); + await targetClient.close(); + await b.close(); + } +}); + +test("agent MCP tool policy rejects fleet creation before effects", async () => { + const { createHostClient } = await import("@getpaseo/client/node"); + const { createPaseoApi } = await import("@getpaseo/client"); + const a = await createTestPaseoDaemon({ + providerOverrides: { + codex: { paseoTools: { disabledTools: ["create_agent", "create_workspace"] } }, + }, + }); + const b = await createTestPaseoDaemon(); + const client = createHostClient(`127.0.0.1:${a.port}`, { + clientId: crypto.randomUUID(), + reconnect: { enabled: false }, + }); + await client.connect(); + const source = createPaseoApi(client); + let restricted: McpClient | undefined; + try { + const parent = await source.agents.create({ + cwd: a.staticDir, + title: "Restricted caller", + config: { provider: "codex/gpt-5.4" }, + }); + await source.hosts.add({ connection: `127.0.0.1:${b.port}` }); + await expect.poll(async () => (await source.hosts.list()).hosts[0]?.status).toBe("online"); + restricted = await createMcpClient( + buildExpectedAgentMcpUrl({ host: "127.0.0.1", port: a.port, agentId: parent.id }), + ); + await expectToolError( + restricted, + "create_workspace", + { serverId: b.daemon.getServerId(), isolation: "local", path: b.staticDir }, + /not found|unknown/i, + ); + await expectToolError( + restricted, + "create_agent", + { + serverId: b.daemon.getServerId(), + workspaceId: "not-authority", + title: "Denied child", + provider: "claude/claude-test-model", + initialPrompt: "Must not execute", + }, + /not found|unknown/i, + ); + // The voice caller context is an injected public tool-host seam; the + // actual MCP server and target daemon remain ordinary real dependencies. + const { createAgentMcpServer } = await import("./mcp-server.js"); + const { ProviderSnapshotManager } = await import("./provider-snapshot-manager.js"); + const { InMemoryTransport } = await import("@modelcontextprotocol/sdk/inMemory.js"); + const { Client } = await import("@modelcontextprotocol/sdk/client/index.js"); + const { default: pino } = await import("pino"); + const logger = pino({ level: "silent" }); + const runtime = source.hosts.runtime(); + const server = await createAgentMcpServer({ + agentManager: a.daemon.agentManager, + agentStorage: a.daemon.agentStorage, + providerSnapshotManager: new ProviderSnapshotManager({ + logger, + extraClients: createTestAgentClients(), + }), + callerAgentId: parent.id, + resolveCallerContext: () => ({ lockedCwd: a.staticDir, allowCustomCwd: false }), + hostRuntime: async () => runtime, + logger, + }); + const locked = new Client({ name: "locked-workspace-caller", version: "1.0.0" }); + const [callerTransport, serverTransport] = InMemoryTransport.createLinkedPair(); + try { + await Promise.all([server.connect(serverTransport), locked.connect(callerTransport)]); + expect( + await locked.callTool({ + name: "create_agent", + arguments: { + serverId: b.daemon.getServerId(), + workspaceId: "must-not-resolve", + title: "Denied placement", + provider: "claude/claude-test-model", + initialPrompt: "Must not execute", + }, + }), + ).toMatchObject({ + isError: true, + structuredContent: { serverId: b.daemon.getServerId(), code: "access_denied" }, + }); + expect( + await locked.callTool({ + name: "create_workspace", + arguments: { serverId: b.daemon.getServerId(), isolation: "local", path: b.staticDir }, + }), + ).toMatchObject({ + isError: true, + structuredContent: { serverId: b.daemon.getServerId(), code: "access_denied" }, + }); + } finally { + await locked.close(); + await server.close(); + await runtime.dispose(); + } + expect((await parent.subagents.list()).rows).toEqual([]); + expect(await b.daemon.agentStorage.list()).toEqual([]); + } finally { + await restricted?.close(); + await source.dispose(); + await client.close(); + await a.close(); + await b.close(); + } +}); + +test("MCP-created managed child belongs to the parent ledger and archives with its parent", async () => { + const { createHostClient } = await import("@getpaseo/client/node"); + const { createPaseoApi } = await import("@getpaseo/client"); + const client = createHostClient(`127.0.0.1:${daemonHandle.port}`, { + clientId: crypto.randomUUID(), + reconnect: { enabled: false }, + }); + await client.connect(); + const api = createPaseoApi(client); + const archived: string[] = []; + const releaseArchived = daemonHandle.daemon.agentManager.onAgentArchived((id) => { + archived.push(id); + }); + let scoped: McpClient | undefined; + try { + const parentId = await createTopLevelAgent({ title: "Ledger MCP parent" }); + scoped = await createMcpClient( + buildExpectedAgentMcpUrl({ host: "127.0.0.1", port: daemonHandle.port, agentId: parentId }), + ); + const child = await callToolStructured(scoped, "create_agent", { + title: "Ledger MCP child", + provider: "claude/claude-test-model", + initialPrompt: "Finish the delegated task", + notifyOnFinish: false, + }); + const childId = str(child.agentId); + expect((await api.agents.ref(parentId).subagents.list()).rows).toEqual([ + expect.objectContaining({ + child: { serverId: daemonHandle.daemon.getServerId(), agentId: childId }, + }), + ]); + const childStates: string[] = []; + const releaseStates = daemonHandle.daemon.agentManager.subscribe( + (event) => { + if (event.type === "agent_state" && event.agent.id === childId) + childStates.push(event.agent.lifecycle); + }, + { agentId: childId, replayState: false }, + ); + try { + await callToolStructured(topLevelClient, "archive_agent", { agentId: parentId }); + await expect + .poll(async () => (await api.agents.ref(childId).refresh())?.agent.archivedAt) + .toBeTruthy(); + expect(archived.filter((id) => id === parentId || id === childId).sort()).toEqual( + [parentId, childId].sort(), + ); + expect(childStates).toContain("closed"); + expect(daemonHandle.daemon.agentManager.getAgent(childId)).toBeNull(); + } finally { + releaseStates(); + } + } finally { + releaseArchived(); + await scoped?.close(); + await api.dispose(); + await client.close(); + } +}); + describe("Suite A: Core Fixes", () => { test.each([ ["Stop", "cancelAgent"], @@ -402,6 +798,7 @@ describe("Suite A: Core Fixes", () => { expect(terminalEvents).toEqual(["turn_canceled"]); expect(notification.source).toEqual({ kind: "agent-notification", + serverId: client.getLastServerInfoMessage()!.serverId, agentId: childId, title: "Watched child", event: "errored", diff --git a/packages/server/src/server/agent/run-notifications/index.ts b/packages/server/src/server/agent/run-notifications/index.ts new file mode 100644 index 0000000000..ea60fa46c6 --- /dev/null +++ b/packages/server/src/server/agent/run-notifications/index.ts @@ -0,0 +1,102 @@ +import { formatFinishNotificationBody } from "../agent-prompt.js"; +import { createHash } from "node:crypto"; +import type { PaseoAgentHandle } from "@getpaseo/client"; +import type { AgentRunRecord } from "@getpaseo/protocol/agent-runs"; + +/** One completion consumer for a qualified child, independent of its transport. */ +export function observeRunCompletion(input: { + child: PaseoAgentHandle; + recipient: PaseoAgentHandle; + submissionId: string; + title: string; + active(): boolean; + failed(error: unknown): void; +}): { ready: Promise; release(): Promise } { + const observation = input.child.runs.observe(input.submissionId); + let closed = false; + const submitted = new Set(); + let delivery: Promise = Promise.resolve(); + const accept = (run: AgentRunRecord | null) => { + if (closed || !input.active() || !run || run.durability === "pending") return; + const checkpoint = run.checkpoints.at(-1); + if ( + !checkpoint || + submitted.has(checkpoint.sequence) || + !["completed", "failed", "canceled", "permission"].includes(checkpoint.kind) + ) + return; + submitted.add(checkpoint.sequence); + const identity = { + serverId: run.agent.serverId, + agentId: run.agent.agentId, + submissionId: run.submissionId, + turnId: run.turnId, + checkpoint: checkpoint.sequence, + }; + const messageId = `run-notification:${createHash("sha256") + .update(JSON.stringify([input.recipient.address, identity])) + .digest("hex")}`; + delivery = delivery + .then(async () => { + if (closed || !input.active()) return; + const permissionRequest = + checkpoint.kind === "permission" + ? (await input.child.refresh())?.agent.pendingPermissions.find( + (request) => request.id === checkpoint.permissionRequestId, + ) + : undefined; + if (checkpoint.kind === "permission" && !permissionRequest) return; + if (closed || !input.active()) return; + const reason = notificationReason(checkpoint.kind); + const text = `${formatFinishNotificationBody({ + childAgentId: run.agent.agentId, + title: input.title, + reason, + lastAssistantMessage: checkpoint.result ?? null, + permissionRequest, + })}\n${JSON.stringify(identity)}${checkpoint.error ? `\n${checkpoint.error}` : ""}`; + await input.recipient.send(text, { + messageId, + unarchive: false, + activeTurnBehavior: "steer", + source: { + agent: run.agent, + title: input.title, + event: notificationEvent(checkpoint.kind), + }, + }); + return undefined; + }) + .catch(input.failed); + }; + observation.subscribe({ + snapshot: ({ run }) => accept(run), + update: (message) => { + if (message.type === "agent.runs.changed") accept(message.payload.run); + }, + error: input.failed, + }); + return { + ready: observation.ready, + async release() { + closed = true; + await observation.release(); + await delivery; + }, + }; +} + +function notificationReason( + kind: string, +): "finished" | "errored" | "was canceled" | "needs permission" { + if (kind === "completed") return "finished"; + if (kind === "canceled") return "was canceled"; + if (kind === "permission") return "needs permission"; + return "errored"; +} + +function notificationEvent(kind: string): "permission-required" | "finished" | "errored" { + if (kind === "permission") return "permission-required"; + if (kind === "completed") return "finished"; + return "errored"; +} diff --git a/packages/server/src/server/agent/tools/host-tool.ts b/packages/server/src/server/agent/tools/host-tool.ts index f919b08cc2..bf2efa36d7 100644 --- a/packages/server/src/server/agent/tools/host-tool.ts +++ b/packages/server/src/server/agent/tools/host-tool.ts @@ -10,12 +10,14 @@ export async function executeHostTool( api: PaseoApi, identity: { serverId: string; originServerId: string }, ) => Promise, + validate?: (identity: { serverId: string; originServerId: string }) => void, ): Promise { let serverId = requestedServerId; try { const runtime = await resolveRuntime(); const originServerId = (await runtime.directory()).self.serverId; serverId ??= originServerId; + validate?.({ serverId, originServerId }); const api = await runtime.client(serverId); let result: PaseoToolResult; let payload: Record; diff --git a/packages/server/src/server/agent/tools/paseo-tools.ts b/packages/server/src/server/agent/tools/paseo-tools.ts index cfdd32a72a..79cc6949fb 100644 --- a/packages/server/src/server/agent/tools/paseo-tools.ts +++ b/packages/server/src/server/agent/tools/paseo-tools.ts @@ -1,8 +1,6 @@ import { executeHostTool } from "./host-tool.js"; -import { AgentDirectory } from "../directory.js"; import type { HostRuntime, PaseoApi } from "@getpaseo/client"; import { HostDirectorySchema } from "@getpaseo/protocol/hosts"; -import { stat } from "node:fs/promises"; import { z } from "zod"; import { ensureValidJson } from "../../json-utils.js"; import type { Logger } from "pino"; @@ -20,23 +18,18 @@ import { WorkspaceScriptPayloadSchema, } from "../../messages.js"; import type { AgentListItemPayload } from "../../messages.js"; -import { - buildStoredAgentPayload, - toAgentListItemPayload, - toAgentPayload, -} from "../agent-projections.js"; +import { toAgentListItemPayload, toAgentPayload } from "../agent-projections.js"; import { curateAgentActivity } from "../activity-curator.js"; import { selectItemsByProjectedLimit } from "../timeline-projection.js"; import type { AgentStorage } from "../agent-storage.js"; -import { AgentNotFoundError, ensureAgentLoaded } from "../agent-loading.js"; -import { isStoredAgentProviderAvailable } from "../../persistence-hooks.js"; +import { AgentNotFoundError } from "../agent-loading.js"; import { archiveByScope, killTerminalsForWorkspace, requireActiveWorkspaceForArchive, type ArchiveDependencies, } from "../../workspace-archive-service.js"; -import { createAgentCommand, type CreateAgentFromMcpInput } from "../create-agent/create.js"; +import { type CreateAgentFromMcpInput } from "../create-agent/create.js"; import type { VoiceCallerContext, VoiceSpeakHandler } from "../../voice-types.js"; import type { FirstAgentContext } from "../../messages.js"; import { everyMsToFiveFieldCron } from "@getpaseo/protocol/schedule/cadence"; @@ -68,7 +61,6 @@ import { parseDurationString, resolveRequiredProviderModel, sanitizePermissionRequest, - serializeSnapshotWithMetadata, toScheduleSummary, } from "../mcp-shared.js"; import { @@ -85,12 +77,10 @@ import type { ProjectRegistry, WorkspaceRegistry, } from "../../workspace-registry.js"; -import { resolveWorktreeSourceCwd } from "../../workspace-source.js"; import type { WorkspaceScriptsService } from "../../session/workspace-scripts/workspace-scripts-service.js"; import { type ArchiveCommandDependencies, type CreatePaseoWorktreeCommandInput, - createPaseoWorktreeCommand, } from "../../worktree/commands.js"; import { registerBrowserTools } from "../../browser-tools/tools.js"; import type { BrowserToolsBroker } from "../../browser-tools/broker.js"; @@ -200,13 +190,13 @@ const WorkspaceAutomationSummarySchema = z.object({ background: z.boolean().optional(), }); -function toWorkspaceAutomationSummary(workspace: PersistedWorkspaceRecord) { +function workspaceSummary(workspace: import("@getpaseo/client").PaseoWorkspace) { return { - workspaceId: workspace.workspaceId, + workspaceId: workspace.id, projectId: workspace.projectId, - cwd: workspace.cwd, - isolation: workspace.kind === "worktree" ? ("worktree" as const) : ("local" as const), - kind: workspace.kind, + cwd: workspace.workspaceDirectory, + isolation: workspace.workspaceKind === "worktree" ? "worktree" : "local", + kind: workspace.workspaceKind, title: workspace.title, background: workspace.background, }; @@ -238,20 +228,6 @@ function assertOptionsAbsent( } } -async function isExistingDirectory(path: string): Promise { - try { - return (await stat(path)).isDirectory(); - } catch (error) { - if (isMissingPathError(error)) return false; - throw error; - } -} - -function isMissingPathError(error: unknown): boolean { - if (typeof error !== "object" || error === null || !("code" in error)) return false; - return error.code === "ENOENT" || error.code === "ENOTDIR"; -} - function resolveWorkspaceWorktreeTarget(input: WorkspaceWorktreeOptions): WorkspaceWorktreeTarget { switch (input.mode ?? "branch-off") { case "branch-off": @@ -654,7 +630,6 @@ function createToolCatalog( terminalManager, workspaceScripts, scheduleService, - providerSnapshotManager, callerAgentId, resolveSpeakHandler, resolveCallerContext, @@ -702,11 +677,18 @@ function createToolCatalog( }; const registerHostTool = ( name: string, - config: Omit & { inputSchema: T }, + config: Omit & { + inputSchema: T; + validateTarget?: ( + input: z.output>, + identity: { serverId: string; originServerId: string }, + ) => void; + }, handler: ( input: z.output>, api: PaseoApi, context: PaseoToolExecutionContext, + identity: { serverId: string; originServerId: string }, ) => Promise, ) => { registerTool( @@ -725,17 +707,28 @@ function createToolCatalog( }, async ({ serverId, ...input }, context) => { if (!options.hostRuntime) throw new Error("Host runtime is not configured"); - return executeHostTool(options.hostRuntime, serverId, (api, identity) => { - const normalized = { ...input }; - if (identity.originServerId !== identity.serverId && callerContext?.lockedCwd) - throw Object.assign( - new PaseoToolInputError("This caller is restricted to its origin workspace"), - { code: "access_denied" }, - ); - if ("cwd" in config.inputSchema && identity.originServerId === identity.serverId) - normalized.cwd = resolveScopedCwd(normalized.cwd); - return handler(normalized as z.output>, api, context); - }); + return executeHostTool( + options.hostRuntime, + serverId, + (api, identity) => { + const normalized = { ...input }; + if ( + "cwd" in config.inputSchema && + identity.originServerId === identity.serverId && + (normalized.cwd !== undefined || callerAgentId) + ) + normalized.cwd = resolveScopedCwd(normalized.cwd); + return handler(normalized as z.output>, api, context, identity); + }, + (identity) => { + if (identity.originServerId !== identity.serverId && callerContext?.lockedCwd) + throw Object.assign( + new PaseoToolInputError("This caller is restricted to its origin workspace"), + { code: "access_denied" }, + ); + config.validateTarget?.(input as z.output>, identity); + }, + ); }, ); }; @@ -1385,7 +1378,7 @@ function createToolCatalog( }, ); - registerTool( + registerHostTool( "create_workspace", { title: "Create workspace", @@ -1440,93 +1433,63 @@ function createToolCatalog( }, outputSchema: WorkspaceAutomationSummarySchema.shape, }, - async ({ - isolation, - path, - projectId, - title, - background, - mode, - worktreeSlug, - branchName, - baseBranch, - branch, - prNumber, - forge, - }) => { - let workspace: PersistedWorkspaceRecord; - if (isolation === "local") { - const cwd = resolveScopedCwd(path, { required: true }); - if (!(await isExistingDirectory(cwd))) { - throw new PaseoToolInputError(`Directory not found: ${cwd}`); - } - assertOptionsAbsent( - [ - ["mode", mode], - ["worktreeSlug", worktreeSlug], - ["branchName", branchName], - ["baseBranch", baseBranch], - ["branch", branch], - ["prNumber", prNumber], - ["forge", forge], - ], - "Worktree options require isolation worktree", - ); - if (!options.createDirectoryWorkspace) { - throw new Error("Workspace provisioning is not configured"); - } - workspace = await options.createDirectoryWorkspace(cwd, title, projectId, { - background, - callerWorkspaceId: resolveCallerAgent()?.workspaceId, - }); - } else { - let cwd = - path !== undefined || !projectId ? resolveScopedCwd(path, { required: true }) : null; - if (!cwd) { - if (!options.projectRegistry) { - throw new Error("Project registry is not configured"); - } - cwd = await resolveWorktreeSourceCwd({ projectId }, options.projectRegistry); - } - const worktreeTarget = resolveWorkspaceWorktreeTarget({ - mode, - worktreeSlug, - branchName, - baseBranch, - branch, - prNumber, - forge, - }); - const result = await createPaseoWorktreeCommand( - { - paseoHome: options.paseoHome, - worktreesRoot: options.worktreesRoot, - createPaseoWorktreeWorkflow: options.createPaseoWorktree, - }, - { - cwd, - ...(projectId ? { projectId } : {}), - ...(worktreeSlug ? { worktreeSlug } : {}), - ...worktreeTarget, - ...(title ? { title } : {}), - background, - callerWorkspaceId: resolveCallerAgent()?.workspaceId, - }, - ); - if (!result.ok) { - throw result.cause; - } - workspace = result.createdWorktree.workspace; - } - - return { - content: [], - structuredContent: ensureValidJson(toWorkspaceAutomationSummary(workspace)), - }; + async ( + { + isolation, + path, + projectId, + title, + background, + mode, + worktreeSlug, + branchName, + baseBranch, + branch, + prNumber, + forge, + }, + api, + _context, + identity, + ) => { + const caller = resolveCallerAgent(); + const sourcePath = + identity.serverId === identity.originServerId + ? resolveScopedCwd(path, { required: !projectId }) + : path; + const source = + isolation === "local" + ? { kind: "directory" as const, path: sourcePath!, projectId } + : { + kind: "worktree" as const, + cwd: sourcePath, + projectId, + worktreeSlug, + ...resolveWorkspaceWorktreeTarget({ + mode, + worktreeSlug, + branchName, + baseBranch, + branch, + prNumber, + forge, + }), + }; + if (isolation === "local" && !sourcePath) + throw new PaseoToolInputError("path is required on the destination host"); + const workspace = await api.workspaces.create({ + source, + title, + background, + ...(identity.serverId === identity.originServerId ? { callerAgentId: caller?.id } : {}), + }); + const snapshot = workspace.current(); + if (!snapshot) throw new Error("Workspace creation omitted its snapshot"); + return { content: [], structuredContent: workspaceSummary(snapshot) }; }, ); - registerTool( + registerHostTool( "list_workspaces", { title: "List workspaces", @@ -1534,19 +1497,18 @@ function createToolCatalog( inputSchema: { includeBackground: z.boolean().optional().default(false) }, outputSchema: { workspaces: z.array(WorkspaceAutomationSummarySchema) }, }, - async ({ includeBackground = false }) => { - if (!options.workspaceRegistry) { - throw new Error("Workspace registry is not configured"); - } - const workspaces = (await options.workspaceRegistry.list()) - .filter( - (workspace) => !workspace.archivedAt && (includeBackground || !workspace.background), - ) - .map(toWorkspaceAutomationSummary); - return { - content: [], - structuredContent: ensureValidJson({ workspaces }), - }; + async ({ includeBackground = false }, api) => { + const workspaces = []; + let cursor: string | undefined; + do { + const page = await api.workspaces.list({ + filter: { includeBackground }, + page: { limit: 200, cursor }, + }); + workspaces.push(...page.entries.map(workspaceSummary)); + cursor = page.pageInfo.nextCursor ?? undefined; + } while (cursor); + return { content: [], structuredContent: { workspaces } }; }, ); @@ -1593,15 +1555,23 @@ function createToolCatalog( }, ); - registerTool( + registerHostTool( "create_agent", { title: "Create agent", description: "Create an agent. Agent-scoped creation defaults to your workspace and creates your subagent. Top-level creation without workspaceId creates a new local workspace. Requires provider/model (for example codex/gpt-5.4) and an initial prompt. Do not guess; call list_providers and list_models first if uncertain.", - inputSchema: createAgentInputSchema, + inputSchema: createAgentInputSchema.shape, + validateTarget: (input, identity) => { + if (identity.serverId !== identity.originServerId && !input.workspaceId) + throw Object.assign( + new PaseoToolInputError("workspaceId is required on the destination host"), + { code: "workspace_required" }, + ); + }, outputSchema: { agentId: z.string(), + submissionId: z.string(), type: AgentProviderEnum, status: AgentStatusEnum, cwd: z.string(), @@ -1611,75 +1581,160 @@ function createToolCatalog( guidance: z.string().optional(), }, }, - async (args: unknown) => { - const resolvedArgs = await resolveCreateAgentToolArgs(args); - const { parsedArgs, worktree } = resolvedArgs; - let notifyOnFinish: boolean; - if (resolvedArgs.kind === "agent-scoped") { - notifyOnFinish = parsedArgs.notifyOnFinish; + async (args, api, _context, identity) => { + // Compatibility placement is normalized here; parentage is always the caller's + // qualified capability and is never inferred from destination metadata. + const resolvedArgs = await normalizeAgentCreation(args, identity); + const { parsedArgs } = resolvedArgs; + const { provider, model } = resolveRequiredProviderModel(parsedArgs.provider); + const config = { + provider: `${provider}/${model}`, + ...(identity.serverId === identity.originServerId + ? { options: resolveInheritedProviderConfig(provider)?.providerOptions } + : {}), + modeId: parsedArgs.settings?.modeId, + thinkingOptionId: parsedArgs.settings?.thinkingOptionId, + featureValues: parsedArgs.settings?.features, + }; + const labels = { ...callerContext?.childAgentDefaultLabels, ...parsedArgs.labels }; + delete labels["paseo.parent-agent-id"]; + const submissionId = crypto.randomUUID(); + const input = { + clientMessageId: submissionId, + config, + title: parsedArgs.title, + prompt: parsedArgs.initialPrompt, + labels, + ...(resolvedArgs.cwd ? { cwd: resolvedArgs.cwd } : {}), + }; + let workspaceId = resolvedArgs.workspaceId; + const caller = resolveCallerAgent(); + await validateAgentPlacement(api, workspaceId, input.cwd); + const workspaceRecipe = creationWorkspaceRecipe(resolvedArgs); + let child; + let snapshot; + if (callerAgentId && !resolvedArgs.detached) { + const origin = await (await options.hostRuntime!()).client(identity.originServerId); + try { + child = await origin.agents.ref(callerAgentId).subagents.create({ + ...input, + cwd: + input.cwd ?? + (identity.serverId === identity.originServerId ? caller?.cwd : undefined), + serverId: identity.serverId, + workspaceId, + workspace: workspaceRecipe, + notifyOnFinish: parsedArgs.notifyOnFinish, + }); + snapshot = (await child.refresh())?.agent; + } finally { + await origin + .dispose() + .catch((error) => childLogger.warn({ error }, "Parent capability cleanup failed")); + } } else { - notifyOnFinish = resolvedArgs.parsedArgs.notifyOnFinish ?? false; + if (workspaceRecipe) { + const agentId = crypto.randomUUID(); + await api.workspaces.create({ + ...workspaceRecipe, + agent: { ...input, agentId, cwd: input.cwd! }, + }); + child = api.agents.ref(agentId); + } else { + if (!workspaceId) + workspaceId = ( + await api.workspaces.create({ source: { kind: "directory", path: process.cwd() } }) + ).id; + child = await api.workspaces.ref(workspaceId).agents.create(input); + } + snapshot = (await child.refresh())?.agent; } - const selectedProvider = resolveRequiredProviderModel(parsedArgs.provider).provider; - const inheritedConfig = resolveInheritedProviderConfig(selectedProvider); - const { snapshot, initialPromptStarted } = await createAgentCommand( - { - agentManager, - agentStorage, - logger: childLogger, - paseoHome: options.paseoHome, - worktreesRoot: options.worktreesRoot, - terminalManager, - providerSnapshotManager, - createPaseoWorktree: options.createPaseoWorktree, - ...(options.ensureWorkspaceForCreate - ? { ensureWorkspaceForCreate: options.ensureWorkspaceForCreate } - : {}), - }, - { - kind: "mcp", - provider: parsedArgs.provider, - title: parsedArgs.title, - initialPrompt: parsedArgs.initialPrompt, - config: inheritedConfig, - cwd: resolvedArgs.cwd, - workspaceId: resolvedArgs.workspaceId, - thinking: parsedArgs.settings?.thinkingOptionId, - features: parsedArgs.settings?.features, - labels: parsedArgs.labels, - mode: parsedArgs.settings?.modeId, - promptFailure: "throw", - notifyOnFinish, - detached: resolvedArgs.detached, - callerAgentId, - callerContext, - worktree, - }, - ); - - // Return immediately for async creation. - const currentSnapshot = agentManager.getAgent(snapshot.id) ?? snapshot; - const guidance = - callerAgentId && notifyOnFinish && initialPromptStarted - ? "You will get notified when the created agent finishes, errors, or needs permission. Do not poll for status; continue with other work until the notification arrives." - : "Creation acknowledged. Observe completion using get_agent_status/get_agent_activity or SDK/RPC subscriptions and wait APIs."; - const response = { + if (!snapshot) throw new Error("Created agent is not available"); + return { content: [], structuredContent: ensureValidJson({ - agentId: currentSnapshot.id, + agentId: child.id, + submissionId, type: snapshot.provider, - status: currentSnapshot.lifecycle, - cwd: currentSnapshot.cwd, - ...(currentSnapshot.workspaceId ? { workspaceId: currentSnapshot.workspaceId } : {}), - currentModeId: currentSnapshot.currentModeId, - availableModes: currentSnapshot.availableModes, - ...(guidance ? { guidance } : {}), + status: snapshot.status, + cwd: snapshot.cwd, + workspaceId: snapshot.workspaceId, + currentModeId: snapshot.currentModeId, + availableModes: snapshot.availableModes, + guidance: + caller && parsedArgs.notifyOnFinish + ? PROMPTED_AGENT_NOTIFICATION_GUIDANCE + : "Creation acknowledged. Observe completion using get_agent_status/get_agent_activity or SDK/RPC subscriptions and wait APIs.", }), }; - return response; }, ); + async function validateAgentPlacement( + api: PaseoApi, + workspaceId: string | undefined, + cwd: string | undefined, + ) { + // Validate placement restrictions before creation/reservation on either host. + if (workspaceId) { + const workspace = await api.workspaces.ref(workspaceId).refresh(); + if (!workspace?.workspaceDirectory) + throw new PaseoToolInputError(`Workspace ${workspaceId} not found`); + if ( + callerContext?.lockedCwd && + !isSameOrDescendantPath( + expandUserPath(callerContext.lockedCwd), + cwd ?? workspace.workspaceDirectory, + ) + ) + throw new PaseoToolInputError(`Workspace ${workspaceId} is outside the allowed cwd`); + } + } + + function creationWorkspaceRecipe(resolvedArgs: { + worktree: CreateAgentFromMcpInput["worktree"]; + cwd: string | undefined; + }) { + return resolvedArgs.worktree + ? { + source: { + kind: "worktree" as const, + cwd: resolvedArgs.cwd!, + action: resolvedArgs.worktree.action, + worktreeSlug: resolvedArgs.worktree.worktreeName, + branchName: resolvedArgs.worktree.branchName, + baseBranch: resolvedArgs.worktree.baseBranch, + refName: resolvedArgs.worktree.refName, + githubPrNumber: resolvedArgs.worktree.githubPrNumber, + }, + } + : undefined; + } + + async function normalizeAgentCreation( + args: z.output, + identity: { serverId: string; originServerId: string }, + ) { + return hasLegacyCreateAgentPlacement(args) + ? await resolveCreateAgentToolArgs(args) + : { + parsedArgs: parseToolArguments( + callerAgentId + ? agentToAgentCreateAgentArgsSchema + : canonicalTopLevelCreateAgentArgsSchema, + args, + ), + workspaceId: + args.workspaceId ?? + (identity.serverId === identity.originServerId + ? resolveCallerAgent()?.workspaceId + : undefined), + cwd: undefined, + worktree: undefined, + detached: false, + }; + } + type ResolvedCreateAgentToolArgs = | { kind: "agent-scoped"; @@ -2156,7 +2211,7 @@ function createToolCatalog( }, ); - registerTool( + registerHostTool( "list_agents", { title: "List agents", @@ -2179,39 +2234,32 @@ function createToolCatalog( agents: z.array(AgentListItemPayloadSchema), }, }, - async ({ - includeBackground = false, - includeArchived = false, - cwd, - sinceHours = 48, - statuses, - limit = 50, - }) => { - const callerCwd = callerAgentId ? resolveCallerAgent()?.cwd : undefined; - const requestedCwd = cwd?.trim() ? expandUserPath(cwd) : callerCwd; + async ( + { + includeBackground = false, + includeArchived = false, + cwd, + sinceHours = 48, + statuses, + limit = 50, + }, + api, + ) => { + const requestedCwd = cwd?.trim() ? expandUserPath(cwd) : undefined; const statusFilter = statuses && statuses.length > 0 ? new Set(statuses) : null; const sinceMs = Date.now() - sinceHours * 60 * 60 * 1000; - if (!options.workspaceRegistry) throw new Error("Workspace registry is not configured"); - const directory = new AgentDirectory({ - manager: agentManager, - storage: agentStorage, - workspaces: options.workspaceRegistry, - projectLive: (snapshot) => - serializeSnapshotWithMetadata(agentStorage, snapshot, childLogger), - projectStored: (record) => - buildStoredAgentPayload( - record, - new Set(providerSnapshotManager.listRegisteredProviderIds()), - ), - isProviderVisible: () => true, - isStoredProviderAvailable: (record) => - isStoredAgentProviderAvailable( - record, - new Set(providerSnapshotManager.listRegisteredProviderIds()), - ), - }); - const agents = (await directory.list({ includeBackground, includeArchived })) - .map(toAgentListItemPayload) + const entries = []; + let cursor: string | undefined; + do { + const page = await api.agents.list({ + filter: { includeBackground, includeArchived }, + page: { limit: 200, cursor }, + }); + entries.push(...page.entries); + cursor = page.pageInfo.nextCursor ?? undefined; + } while (cursor); + const agents = entries + .map((entry) => toAgentListItemPayload(entry.agent)) .filter((agent) => !requestedCwd || isSameOrDescendantPath(requestedCwd, agent.cwd)) .filter((agent) => !statusFilter || statusFilter.has(agent.status)) .filter((agent) => !agent.archivedAt || resolveAgentListActivityTime(agent) >= sinceMs) @@ -3135,7 +3183,7 @@ function createToolCatalog( }, ); - registerTool( + registerHostTool( "get_agent_activity", { title: "Get agent activity", @@ -3154,14 +3202,11 @@ function createToolCatalog( content: z.string(), }, }, - async ({ agentId, limit }) => { - await ensureAgentLoaded(agentId, { - agentManager, - agentStorage, - logger: childLogger, - }); - const timeline = agentManager.getTimeline(agentId); - const snapshot = agentManager.getAgent(agentId); + async ({ agentId, limit }, api) => { + const agent = api.agents.ref(agentId); + const payload = await agent.timeline.refetch(); + const timeline = payload.entries.map((entry) => entry.item); + const snapshot = (await agent.refresh())?.agent; const selection = selectItemsByProjectedLimit({ items: timeline, diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 17b435e453..e662eb8349 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -4383,7 +4383,10 @@ export class Session { ): Promise { // COMPAT(qualifiedSubagents): added in v0.12, remove after 2027-04-20. // Normalize legacy parent placement before the creation owner reserves its resource. - const legacyParentId = request.callerAgentId ?? request.labels[PARENT_AGENT_ID_LABEL]; + const legacyParentId = + request.parent === undefined + ? (request.callerAgentId ?? request.labels[PARENT_AGENT_ID_LABEL]) + : undefined; if (legacyParentId && this.subagentOptions.subagents) { const key = request.idempotencyKey ?? request.requestId; const observing = await this.creationService.subscribe("agent", key, observer ?? (() => {})); @@ -4538,7 +4541,10 @@ export class Session { ): Promise { const { requestId, type: _type, subscribe: _subscribe, idempotencyKey, ...intent } = request; const agentInput = request.agent; - const parentId = agentInput?.callerAgentId ?? agentInput?.labels[PARENT_AGENT_ID_LABEL]; + const parentId = + agentInput?.parent === undefined + ? (agentInput?.callerAgentId ?? agentInput?.labels[PARENT_AGENT_ID_LABEL]) + : undefined; const { subagents, subagentAuthority, serverId } = this.subagentOptions; if (agentInput && parentId && subagents && subagentAuthority && serverId) { const key = idempotencyKey ?? requestId; @@ -4717,7 +4723,10 @@ export class Session { ): Promise { // COMPAT(qualifiedSubagents): normalize pre-fleet caller/label creation at the session edge. // Added in v0.12; remove after 2027-04-20 once all creation callers use parent capabilities. - const legacyParentId = msg.callerAgentId ?? msg.labels[PARENT_AGENT_ID_LABEL]; + const legacyParentId = + msg.parent === undefined + ? (msg.callerAgentId ?? msg.labels[PARENT_AGENT_ID_LABEL]) + : undefined; const { subagents, subagentAuthority, serverId } = this.subagentOptions; if (legacyParentId && subagents && subagentAuthority && serverId) { const { @@ -4836,9 +4845,7 @@ export class Session { workspaceId: resolvedIntent.intent.workspaceId, worktreeName, initialPrompt, - source: msg.callerAgentId - ? { kind: "agent-message", agentId: msg.callerAgentId } - : undefined, + source: creationPromptSource(msg), clientMessageId, outputSchema, images, @@ -9272,3 +9279,16 @@ function createHostSession(options: SessionOptions, session: Session): Connected function hostSource(options: SessionOptions): string { return options.sourceId ?? "owner"; } + +function creationPromptSource( + message: CreateAgentRequestMessage, +): import("./agent/agent-messages/index.js").AgentPromptSource | undefined { + if (message.parent) + return { + kind: "agent-message", + agentId: message.parent.agentId, + serverId: message.parent.serverId, + }; + if (message.callerAgentId) return { kind: "agent-message", agentId: message.callerAgentId }; + return undefined; +} diff --git a/packages/server/src/server/subagents/index.ts b/packages/server/src/server/subagents/index.ts index 43ce412b37..2932bfdeb9 100644 --- a/packages/server/src/server/subagents/index.ts +++ b/packages/server/src/server/subagents/index.ts @@ -1,3 +1,4 @@ +import { observeRunCompletion } from "../agent/run-notifications/index.js"; import type { HostDirectory } from "@getpaseo/protocol/hosts"; import { shouldDetachFromArchivedParent } from "@getpaseo/protocol/agent-lifecycle"; import { hasOpenAgentTab, getParentAgentIdFromLabels } from "@getpaseo/protocol/agent-labels"; @@ -40,6 +41,7 @@ const AuthoritySchema = z.object({ const RelationshipStateSchema = z.object({ row: SubagentRelationshipSchema, authority: AuthoritySchema, + notificationError: z.string().optional(), confirmed: z.enum(["attached", "detached", "archived", "canceled"]).nullable(), intent: z.object({ revision: z.number().int().nonnegative(), @@ -480,6 +482,7 @@ export class Subagents { const { config, initialPrompt, ...input } = creation.input; const agentOptions = { ...input, + labels: { ...input.labels, "paseo.parent-agent-id": row.parent.agentId }, agentId: row.child.agentId, config: { ...config, @@ -696,6 +699,18 @@ export class Subagents { ) return; const observation = child.observe(); + const notification = + record.provenance === "creation" && record.creation.notifyOnFinish + ? observeRunCompletion({ + child, + recipient: await runtime.agent(record.row.parent), + submissionId: + record.creation.input.clientMessageId ?? `subagent:${record.row.id}:initial`, + title: record.creation.input.config.title ?? "Subagent", + active: () => !this.closed && record.intent === intent && observesChild(record), + failed: (error) => this.markNotificationFailure(record, intent, error), + }) + : null; let applying: Promise = Promise.resolve(); const accept = (snapshot: AgentSnapshotPayload): void => { applying = applying @@ -708,6 +723,7 @@ export class Subagents { if (!observesChild(record)) { this.observers.delete(record.row.id); await observation.release(); + await notification?.release(); await this.releaseUnusedRuntimes(); } return undefined; @@ -720,6 +736,7 @@ export class Subagents { intent, release: async () => { await observation.release(); + await notification?.release(); await applying; }, }); @@ -739,14 +756,27 @@ export class Subagents { }); try { await observation.ready; + await notification?.ready; await applying; } catch (error) { this.observers.delete(record.row.id); await observation.release(); + await notification?.release(); if (record.intent === intent) await this.markUnavailable(record, error); throw error; } } + private markNotificationFailure( + record: RelationshipRecord, + intent: RelationshipIntent, + error: unknown, + ): void { + void this.transact(async () => { + if (record.intent !== intent) return; + record.notificationError = error instanceof Error ? error.message : String(error); + await this.persist(); + }).catch(() => {}); + } private captureSnapshot(record: RelationshipRecord, snapshot: AgentSnapshotPayload | null): void { if (!snapshot) return; record.row.snapshot = snapshot;