diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index d30a2f4aad..75f793c4b4 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -65,7 +65,6 @@ function createMockTransport() { let onOpen: () => void = () => {}; let onClose: (_event?: unknown) => void = () => {}; let onError: (_event?: unknown) => void = () => {}; - let serverInfoOrdinal = 1; const transport: DaemonTransport = { send: (data) => { @@ -113,7 +112,7 @@ function createMockTransport() { type: "status", payload: { status: "server_info", - serverId: `srv_test_${serverInfoOrdinal++}`, + serverId: "srv_test_1", hostname: null, version: null, features: options?.features ?? { ownedSubscriptions: true }, diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index 262cca26f0..9bdd7a9e4c 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -6446,6 +6446,11 @@ export class DaemonClient { return this.lastServerInfoMessage; } + /** Accepted destination identity survives connection gaps. */ + getServerId(): string | null { + return this.destinationServerId ?? null; + } + private requireHubRelationshipSupport(): void { // COMPAT(hubRelationship): added in v0.1.X, drop the gate when floor >= v0.1.X. if (this.lastServerInfoMessage?.features?.hubRelationship !== true) { @@ -7007,6 +7012,15 @@ export class DaemonClient { if (consumerMessage.type === "status") { const serverInfo = parseServerInfoStatusPayload(consumerMessage.payload); if (serverInfo) { + if (this.destinationServerId && serverInfo.serverId !== this.destinationServerId) { + const failure = Object.assign(new Error("The connection reached a different host"), { + code: "server_id_mismatch", + serverId: this.destinationServerId, + }); + this.disposeTransport(1008, failure.message); + this.scheduleReconnect({ reason: failure.message, failure }); + return; + } this.lastServerInfoMessage = serverInfo; this.destinationServerId = serverInfo.serverId; this.connectionFailure = undefined; diff --git a/packages/client/src/hosts/index.ts b/packages/client/src/hosts/index.ts index 1baa9ebe43..fc39403e1f 100644 --- a/packages/client/src/hosts/index.ts +++ b/packages/client/src/hosts/index.ts @@ -4,6 +4,7 @@ import type { DaemonTransportFactory } from "../daemon-client-transport-types.js import type { OwnedSubscription } from "../connection/index.js"; import type { PaseoApi, createPaseoApi } from "../index.js"; import type { HostAdd, HostUpdate, HostSummary, HostDirectory } from "@getpaseo/protocol/hosts"; +import { HostRuntime } from "./runtime.js"; export interface PaseoHostActions { list(): Promise; @@ -12,6 +13,7 @@ export interface PaseoHostActions { update(registrationId: string, patch: HostUpdate): Promise; remove(registrationId: string): Promise; client(serverId: string): Promise; + runtime(): HostRuntime; } export function createHostChannelTransport( @@ -127,7 +129,7 @@ export function createHostActions( const assertActive = () => { if (owner.signal.aborted) throw new Error("Paseo API is disposed"); }; - return { + const actions: PaseoHostActions = { list: () => { assertActive(); return origin.listHosts(); @@ -204,5 +206,14 @@ export function createHostActions( throw failure; } }, + runtime() { + assertActive(); + const runtime = new HostRuntime({ hosts: actions }); + const dispose = runtime.dispose.bind(runtime); + const scope = owner.own(() => ({ release: dispose })); + runtime.dispose = () => scope.release(); + return runtime; + }, }; + return actions; } diff --git a/packages/client/src/hosts/runtime.ts b/packages/client/src/hosts/runtime.ts new file mode 100644 index 0000000000..451a6ff663 --- /dev/null +++ b/packages/client/src/hosts/runtime.ts @@ -0,0 +1,83 @@ +import { AgentAddressSchema, type AgentAddress } from "@getpaseo/protocol/addresses"; +import type { HostDirectory } from "@getpaseo/protocol/hosts"; +import type { PaseoApi, PaseoAgentHandle } from "../index.js"; + +/** One origin authority owns resolution and the lifetime of its destination scopes. */ +export class HostRuntime { + private readonly agentClients = new Map>(); + private readonly pending = new Set>(); + private readonly scopes = new Set(); + private disposed = false; + private disposal: Promise | null = null; + + constructor(private readonly origin: Pick) {} + + async directory(): Promise { + if (this.disposed) throw new Error("Host runtime is disposed"); + return this.origin.hosts.list(); + } + + /** Caller-owned scope. Disposing it never invalidates another consumer's scope. */ + async client(serverId: string): Promise { + if (this.disposed) throw new Error("Host runtime is disposed"); + const acquiring = this.origin.hosts.client(serverId).then(async (api) => { + if (this.disposed) { + await api.dispose(); + throw new Error("Host runtime is disposed"); + } + let disposal: Promise | null = null; + const scope = { + ...api, + dispose: () => { + if (!disposal) { + this.scopes.delete(scope); + disposal = api.dispose(); + } + return disposal; + }, + }; + this.scopes.add(scope); + return scope; + }); + this.pending.add(acquiring); + void acquiring.then( + () => this.pending.delete(acquiring), + () => this.pending.delete(acquiring), + ); + return acquiring; + } + + /** Handles borrow a runtime-owned scope and remain bound to its origin authority. */ + async agent(address: AgentAddress): Promise { + if (this.disposed) throw new Error("Host runtime is disposed"); + const { serverId, agentId } = AgentAddressSchema.parse(address); + let client = this.agentClients.get(serverId); + if (!client) { + client = this.client(serverId); + this.agentClients.set(serverId, client); + const acquiring = client; + void acquiring.catch(() => { + if (this.agentClients.get(serverId) === acquiring) this.agentClients.delete(serverId); + }); + } + const api = await client; + if (this.disposed) throw new Error("Host runtime is disposed"); + return api.agents.ref(agentId); + } + + dispose(): Promise { + if (this.disposal) return this.disposal; + this.disposed = true; + this.disposal = (async () => { + const results = await Promise.allSettled([...this.scopes].map((scope) => scope.dispose())); + // Acquisitions that finish after disposal release themselves before rejecting callers. + await Promise.allSettled(this.pending); + this.agentClients.clear(); + const failures = results.flatMap((result) => + result.status === "rejected" ? [result.reason] : [], + ); + if (failures.length) throw new AggregateError(failures, "Failed to dispose host runtime"); + })(); + return this.disposal; + } +} diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index c9f9c3fa9d..a26c535bb2 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -1,4 +1,7 @@ import { createHostActions, type PaseoHostActions } from "./hosts/index.js"; +import type { AgentAddress } from "@getpaseo/protocol/addresses"; +export type { AgentAddress, WorkspaceAddress } from "@getpaseo/protocol/addresses"; +export { HostRuntime } from "./hosts/runtime.js"; export type { PaseoHostActions } from "./hosts/index.js"; export type { HostAdd, HostUpdate, HostSummary, HostDirectory } from "@getpaseo/protocol/hosts"; import type { OwnedSubscription } from "./connection/index.js"; @@ -343,6 +346,8 @@ export interface PaseoAgentTimelineHandle { export interface PaseoAgentHandle { readonly id: string; + readonly serverId: 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 @@ -587,6 +592,7 @@ export function createPaseoApi( daemonClient, listenAgents, (agentId, handler) => own(() => daemonClient.subscribeAgentTimeline(agentId, handler)), + lifetime.signal, ); const createAgent = async ( options: PaseoAgentCreateOptions, @@ -865,16 +871,32 @@ function createAgentHandleFactory( daemonClient: DaemonClient, listen: (handler: PaseoAgentUpdateHandler) => () => void, subscribeTimeline: DaemonClient["subscribeAgentTimeline"], + signal: AbortSignal, ): AgentHandleFactory { + const assertActive = () => { + if (signal.aborted) throw new Error("Paseo API is disposed"); + }; return (agent) => { const id = typeof agent === "string" ? agent : agent.id; let current = typeof agent === "string" ? null : agent; const handle: PaseoAgentHandle = { id, + get serverId() { + const serverId = daemonClient.getServerId(); + if (!serverId) throw new Error("Connect to the host before reading an agent address"); + return serverId; + }, + get address() { + return { serverId: handle.serverId, agentId: id }; + }, timeline: { - append: (item) => daemonClient.appendAgentTimelineItem(id, item), + append: async (item) => { + assertActive(); + return daemonClient.appendAgentTimelineItem(id, item); + }, refetch: async (options) => { + assertActive(); const result = await daemonClient.fetchAgentTimeline(id, options); if (result.agent) { current = result.agent; @@ -943,19 +965,24 @@ function createAgentHandleFactory( }, current: () => current, refresh: async (requestId) => { + assertActive(); const result = await daemonClient.fetchAgent({ agentId: id, requestId }); current = result?.agent ?? null; return result; }, send: async (text, options) => { + assertActive(); await daemonClient.sendAgentMessage(id, text, options); }, respondToPermission: async ({ requestId, response }) => { + assertActive(); await daemonClient.respondToPermission(id, requestId, response); }, run: async (text, options) => { + assertActive(); const { timeoutMs, ...sendOptions } = options ?? {}; await daemonClient.sendAgentMessage(id, text, sendOptions); + assertActive(); const result = await daemonClient.waitForFinish( id, timeoutMs ?? DEFAULT_WAIT_FOR_FINISH_MS, @@ -966,6 +993,7 @@ function createAgentHandleFactory( return result; }, waitForFinish: async (timeoutMs) => { + assertActive(); const result = await daemonClient.waitForFinish( id, timeoutMs ?? DEFAULT_WAIT_FOR_FINISH_MS, @@ -975,8 +1003,12 @@ function createAgentHandleFactory( } return result; }, - commands: (options) => daemonClient.listCommands({ agentId: id, ...options }), + commands: async (options) => { + assertActive(); + return daemonClient.listCommands({ agentId: id, ...options }); + }, archive: async () => { + assertActive(); const result = await daemonClient.archiveAgent(id); if (current) { current = { ...current, archivedAt: result.archivedAt }; @@ -984,6 +1016,7 @@ function createAgentHandleFactory( return result; }, detach: async () => { + assertActive(); await daemonClient.detachAgent(id); }, subscribe: (handler) => diff --git a/packages/protocol/src/addresses.ts b/packages/protocol/src/addresses.ts new file mode 100644 index 0000000000..c97078f6be --- /dev/null +++ b/packages/protocol/src/addresses.ts @@ -0,0 +1,14 @@ +import { z } from "zod"; + +export const AgentAddressSchema = z.object({ + serverId: z.string().min(1), + agentId: z.string().min(1), +}); + +export const WorkspaceAddressSchema = z.object({ + serverId: z.string().min(1), + workspaceId: z.string().min(1), +}); + +export type AgentAddress = z.infer; +export type WorkspaceAddress = z.infer; diff --git a/packages/server/src/server/agent/tools/paseo-tools.ts b/packages/server/src/server/agent/tools/paseo-tools.ts index 30f28426b1..30ac658f3e 100644 --- a/packages/server/src/server/agent/tools/paseo-tools.ts +++ b/packages/server/src/server/agent/tools/paseo-tools.ts @@ -1,4 +1,6 @@ import { AgentDirectory } from "../directory.js"; +import type { HostRuntime } 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"; @@ -110,6 +112,7 @@ import type { ProviderPaseoToolsPolicy } from "@getpaseo/protocol/provider-confi import { isPaseoToolEnabled } from "../paseo-tool-policy.js"; export interface PaseoToolHostDependencies { + hostRuntime?: () => Promise; agentManager: AgentManager; agentStorage: AgentStorage; terminalManager?: TerminalManager | null; @@ -1332,6 +1335,24 @@ function createToolCatalog( }); } + registerTool( + "list_hosts", + { + title: "List hosts", + description: + "Discover this host and its saved one-hop destinations without connection credentials.", + inputSchema: {}, + outputSchema: HostDirectorySchema.shape, + }, + async () => { + if (!options.hostRuntime) throw new Error("Host runtime is not configured"); + return { + content: [], + structuredContent: ensureValidJson(await (await options.hostRuntime()).directory()), + }; + }, + ); + registerTool( "create_workspace", { diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 73e668735f..ee117615c8 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -1,4 +1,6 @@ import type { PluginRegistries } from "@getpaseo/protocol/plugin-registry"; +import type { HostRuntime } from "@getpaseo/client"; +import { OWNER_PERMISSIONS } from "./authorization/index.js"; import { describeHookWorkspace } from "./plugins/lifecycle/index.js"; import express from "express"; import { createServer as createHTTPServer, type IncomingMessage, type ServerResponse } from "http"; @@ -119,6 +121,7 @@ export async function fanOutReconciledWorkspaceUpdates(input: { } import { VoiceAssistantWebSocketServer } from "./websocket-server.js"; +import type { HostRuntimeAuthority } from "./host-runtime/index.js"; import { WorkspaceSetupRuntime } from "./workspace-setup-runtime.js"; import { createWorkspaceLabelService } from "./workspace-labels/index.js"; import { createGitHubService } from "../services/github-service.js"; @@ -469,6 +472,7 @@ export interface PaseoDaemonConfig { export interface PaseoDaemon { config: PaseoDaemonConfig; + createHostRuntime(authority: HostRuntimeAuthority): Promise; agentManager: AgentManager; agentStorage: AgentStorage; terminalManager: TerminalManager; @@ -1390,9 +1394,30 @@ export async function createPaseoDaemon( ); logger.info({ elapsed: elapsed() }, "Preparing voice and MCP runtime"); + let agentToolHostRuntime: Promise | null = null; + const createHostRuntime = (authority: HostRuntimeAuthority) => { + if (!wsServer) return Promise.reject(new Error("WebSocket server is not running")); + return wsServer.createHostRuntime(authority); + }; + const getAgentToolHostRuntime = (): Promise => { + if (!wsServer) return Promise.reject(new Error("WebSocket server is not running")); + if (!agentToolHostRuntime) { + // Agent MCP already admits the daemon capability token or owner password. + const acquiring = createHostRuntime({ + permissionCeiling: OWNER_PERMISSIONS, + resolveAdmission: async () => ({ principalId: "owner", permissions: OWNER_PERMISSIONS }), + }); + agentToolHostRuntime = acquiring; + void acquiring.catch(() => { + if (agentToolHostRuntime === acquiring) agentToolHostRuntime = null; + }); + } + return agentToolHostRuntime; + }; const createAgentToolHostDependencies = ( runtime: PaseoToolRuntimeContext, ): PaseoToolHostDependencies => ({ + hostRuntime: getAgentToolHostRuntime, agentManager, agentStorage, terminalManager, @@ -1864,6 +1889,7 @@ export async function createPaseoDaemon( return { config, + createHostRuntime, agentManager, agentStorage, terminalManager, diff --git a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts index 0005946316..77813bc01d 100644 --- a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts +++ b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts @@ -1,5 +1,5 @@ import { afterEach, expect, test } from "vitest"; -import { createPaseoApi } from "@getpaseo/client"; +import { createPaseoApi, HostRuntime } from "@getpaseo/client"; import { createHostClient, createHostChannelTransport } from "@getpaseo/client/node"; import { createTestPaseoDaemon, type TestPaseoDaemon } from "../test-utils/paseo-daemon.js"; import type { HostDirectory } from "@getpaseo/protocol/hosts"; @@ -516,12 +516,13 @@ import { z } from "zod"; const operate = defineRpc({ name: "operate", input: z.object({ serverId: z.string(), path: z.string() }), output: z.object({ workspaceId: z.string(), registrationId: z.string(), sourceId: z.string() }) }); export default function contribute(server) { server.handle(operate, async (input, context) => { - const remote = await context.paseo.hosts.client(input.serverId); + const runtime = context.paseo.hosts.runtime(); + const remote = await runtime.client(input.serverId); try { const workspace = await remote.workspaces.create({ source: { kind: "directory", path: input.path }, title: "Plugin remote workspace" }); const host = await remote.hosts.add({ connection: "127.0.0.1:1", sourceKey: "plugin-stable", metadata: { team: "plugin" } }); return { workspaceId: workspace.id, registrationId: host.registrationId, sourceId: host.source.id }; - } finally { await remote.dispose(); } + } finally { await runtime.dispose(); } }); return () => {}; }`, @@ -905,13 +906,14 @@ test("source-key registration remains idempotent at directory capacity", async ( /** Real WebSocket proxy records the A–B physical wire without replacing either daemon. */ async function hostWire(b: TestPaseoDaemon) { + let destination = b; const http = createServer(); const wss = new WebSocketServer({ server: http }); const frames: { binary: boolean; bytes: Buffer; direction: "toB" | "fromB" }[] = []; let connections = 0; wss.on("connection", (downstream) => { connections++; - const upstream = new WebSocket(`ws://127.0.0.1:${b.port}/ws`); + const upstream = new WebSocket(`ws://127.0.0.1:${destination.port}/ws`); const pending: (() => void)[] = []; upstream.on("open", () => { for (const send of pending.splice(0)) send(); @@ -948,6 +950,10 @@ async function hostWire(b: TestPaseoDaemon) { frames, connections: () => connections, active: () => wss.clients.size, + retarget(next: TestPaseoDaemon) { + destination = next; + for (const socket of wss.clients) socket.close(); + }, }; } @@ -1303,3 +1309,240 @@ test("cancelling a logical scope during admission settles it and releases its ow endpoint.close(); await delivery.close(); }); + +test("agent handles keep server identity when agent IDs collide across hosts", async () => { + const a = await daemon(); + const b = await daemon(); + const origin = await api(a); + await origin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` }); + const directory = await online(origin.paseo); + const destination = directory.hosts[0]!.serverId!; + const remote = await origin.paseo.hosts.client(destination); + const id = crypto.randomUUID(); + const localAgent = await origin.paseo.agents.create({ + agentId: id, + cwd: a.staticDir, + config: { provider: "codex/test" }, + title: "On A", + }); + const remoteAgent = await remote.agents.create({ + agentId: id, + cwd: b.staticDir, + config: { provider: "codex/test" }, + title: "On B", + }); + expect(localAgent.address).toEqual({ serverId: directory.self.serverId, agentId: id }); + expect(remoteAgent.address).toEqual({ serverId: destination, agentId: id }); + expect(localAgent.serverId).toBe(directory.self.serverId); + expect(remoteAgent.serverId).toBe(destination); + expect(localAgent.current()?.title).toBe("On A"); + expect(remoteAgent.current()?.title).toBe("On B"); + const runtime = origin.paseo.hosts.runtime(); + cleanup.push(() => runtime.dispose()); + const direct = await api(b); + const directRuntime = direct.paseo.hosts.runtime(); + cleanup.push(() => directRuntime.dispose()); + const [firstScope, secondScope] = await Promise.all([ + runtime.client(destination), + runtime.client(destination), + ]); + expect(firstScope).not.toBe(secondScope); + await firstScope.dispose(); + expect((await secondScope.workspaces.list()).entries).toHaveLength(1); + for (const [owner, address, title] of [ + [runtime, localAgent.address, "On A"], + [runtime, remoteAgent.address, "On B"], + [directRuntime, remoteAgent.address, "On B"], + ] as const) { + const agent = await owner.agent(address); + expect(agent.address).toEqual(address); + expect((await agent.refresh())?.agent.title).toBe(title); + } + await origin.paseo.hosts.update(directory.hosts[0]!.registrationId, { enabled: false }); + expect(remoteAgent.address).toEqual({ serverId: destination, agentId: id }); + const unavailable = await runtime.agent(remoteAgent.address); + await expect(unavailable.refresh()).rejects.toThrow(); + const stillDirect = await directRuntime.agent(remoteAgent.address); + expect((await stillDirect.refresh())?.agent.title).toBe("On B"); + await runtime.dispose(); + await expect(runtime.agent(localAgent.address)).rejects.toThrow("disposed"); +}); + +test("established agent identity survives disconnect and rejects a different server on reconnect", async () => { + const a = await daemon(); + const b = await daemon(); + const wire = await hostWire(a); + const client = createHostClient(wire.connection, { + clientId: crypto.randomUUID(), + reconnect: { enabled: false }, + }); + cleanup.push(() => client.close()); + await client.connect(); + const paseo = createPaseoApi(client); + cleanup.push(() => paseo.dispose()); + const agent = paseo.agents.ref(crypto.randomUUID()); + const address = agent.address; + wire.retarget(b); + await expect.poll(() => client.getConnectionState().status).toBe("disconnected"); + expect(client.getLastServerInfoMessage()).toBeNull(); + expect(agent.address).toEqual(address); + await expect(client.connect()).rejects.toMatchObject({ + code: "server_id_mismatch", + serverId: address.serverId, + }); + expect(agent.address).toEqual(address); + expect(client.getConnectionState().status).toBe("disconnected"); +}); + +test("runtime disposal rejects every pending acquisition and releases the resulting scopes", async () => { + const a = await daemon(); + const b = await daemon(); + const origin = await api(a); + await origin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` }); + const directory = await online(origin.paseo); + const serverId = directory.hosts[0]!.serverId!; + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const acquired: ReturnType[] = []; + // Delay an ordinary public acquisition; every connection and scope is real. + const runtime = new HostRuntime({ + hosts: { + ...origin.paseo.hosts, + async client(target) { + const client = await origin.paseo.hosts.client(target); + acquired.push(client); + await gate; + return client; + }, + }, + }); + cleanup.push(() => runtime.dispose()); + const calls = [ + runtime.client(serverId), + runtime.client(serverId), + runtime.agent({ serverId, agentId: crypto.randomUUID() }), + ]; + const settled = Promise.allSettled(calls); + await expect.poll(() => acquired.length).toBe(3); + const disposed = runtime.dispose(); + release(); + const results = await settled; + expect( + results.every( + (result) => result.status === "rejected" && /disposed/.test(String(result.reason)), + ), + ).toBe(true); + await disposed; + for (const client of acquired) await expect(client.workspaces.list()).rejects.toThrow(); +}); + +test("MCP list_hosts uses the daemon host runtime and returns the same secret-free directory as SDK", async () => { + const a = await daemon(); + const b = await daemon(); + const origin = await api(a); + await origin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}`, label: "Build box" }); + await online(origin.paseo); + const { Client } = await import("@modelcontextprotocol/sdk/client/index.js"); + const { StreamableHTTPClientTransport } = + await import("@modelcontextprotocol/sdk/client/streamableHttp.js"); + const mcp = new Client({ name: "fleet-regression", version: "1.0.0" }); + cleanup.push(() => mcp.close()); + await mcp.connect( + new StreamableHTTPClientTransport(new URL(`http://127.0.0.1:${a.port}/mcp/agents`)), + ); + const result = await mcp.callTool({ name: "list_hosts", arguments: {} }); + expect(result.isError).not.toBe(true); + expect(result.structuredContent).toEqual(await origin.paseo.hosts.list()); +}); + +test("a changed daemon on physical reconnect remains an inspectable identity failure", async () => { + const a = await daemon(); + const b = await daemon(); + const c = await daemon(); + const wire = await hostWire(b); + const origin = await api(a); + await origin.paseo.hosts.add({ connection: wire.connection }); + const directory = await online(origin.paseo); + const registration = directory.hosts[0]!; + wire.retarget(c); + await expect + .poll(async () => (await origin.paseo.hosts.list()).hosts[0]?.error?.code, { timeout: 10000 }) + .toBe("server_id_mismatch"); + expect((await origin.paseo.hosts.list()).hosts[0]).toMatchObject({ + serverId: registration.serverId, + status: "error", + nextRetryAt: null, + }); + await origin.paseo.hosts.update(registration.registrationId, { + connection: wire.connection, + replaceIdentity: true, + }); + expect((await online(origin.paseo)).hosts[0]?.serverId).toBe(c.daemon.getServerId()); +}, 20000); + +test("runtime disposal ends borrowed agent operations while preserving identity and origin ownership", async () => { + const a = await daemon(); + const origin = await api(a); + const local = await origin.paseo.agents.create({ + cwd: a.staticDir, + config: { provider: "codex/test" }, + }); + const runtime = origin.paseo.hosts.runtime(); + cleanup.push(() => runtime.dispose()); + const borrowed = await runtime.agent(local.address); + await borrowed.refresh(); + await runtime.dispose(); + expect(borrowed.address).toEqual(local.address); + expect(borrowed.current()?.id).toBe(local.id); + await expect(borrowed.refresh()).rejects.toThrow("disposed"); + await expect(borrowed.send("must not dispatch")).rejects.toThrow("disposed"); + await expect(borrowed.archive()).rejects.toThrow("disposed"); + expect((await local.refresh())?.agent.archivedAt).toBeNull(); +}); + +test("disposing a daemon-owned runtime closes its source session and failed admission leaves none", async () => { + const a = await daemon(); + const origin = await api(a); + const sessions = async () => { + const { diagnostic } = await origin.client.collectDiagnostics(); + const runtimes = /Host runtimes: (\d+)/.exec(diagnostic); + const ownedSessions = /Owned sessions: (\d+)/.exec(diagnostic); + if (!runtimes || !ownedSessions) throw new Error("Missing host runtime diagnostics"); + return [Number(runtimes[1]), Number(ownedSessions[1])]; + }; + const before = await sessions(); + const runtime = await a.daemon.createHostRuntime({ + permissionCeiling: ["daemon.read", "workspace.read"], + resolveAdmission: async () => ({ + principalId: "owner", + permissions: ["daemon.read", "workspace.read", "workspace.manage"], + }), + }); + cleanup.push(() => runtime.dispose()); + expect((await runtime.directory()).self.serverId).toBe(a.daemon.getServerId()); + const scoped = await runtime.client(a.daemon.getServerId()); + await expect( + scoped.workspaces.create({ source: { kind: "directory", path: a.staticDir } }), + ).rejects.toMatchObject({ code: "access_denied" }); + await scoped.dispose(); + const active = await sessions(); + expect(active[0]).toBe(before[0]! + 1); + expect(active[1]).toBe(before[1]! + 1); + await runtime.dispose(); + await expect.poll(sessions).toEqual(before); + await expect(runtime.directory()).rejects.toThrow("disposed"); + const revoked = Object.assign(new Error("Runtime authority was revoked"), { + code: "access_denied", + }); + await expect( + a.daemon.createHostRuntime({ + permissionCeiling: ["daemon.read"], + resolveAdmission: async () => { + throw revoked; + }, + }), + ).rejects.toMatchObject({ code: "access_denied" }); + await expect.poll(sessions).toEqual(before); +}); diff --git a/packages/server/src/server/connected-hosts/index.ts b/packages/server/src/server/connected-hosts/index.ts index 1a8f7c2731..45f4408d4a 100644 --- a/packages/server/src/server/connected-hosts/index.ts +++ b/packages/server/src/server/connected-hosts/index.ts @@ -339,7 +339,7 @@ export class ConnectedHosts { message: "Cannot reach host. Check the connection; retrying automatically.", }) : null; - if (row.error?.code === "authentication_failed") { + if (row.error?.code === "authentication_failed" || row.error?.code === "server_id_mismatch") { row.status = "error"; row.nextRetryAt = null; client.setReconnectEnabled(false); @@ -528,6 +528,12 @@ export class ConnectedHosts { function connectionFailure(error: unknown, authenticationFailed: boolean): HostSummary["error"] { if (error instanceof HostError) return { code: error.code, message: error.message }; + if (error instanceof Error && "code" in error && error.code === "server_id_mismatch") + return { + code: "server_id_mismatch", + message: + "Daemon identity changed. Verify the destination and explicitly replace its identity.", + }; if (error instanceof Error && "code" in error && error.code === "authentication_failed") return { code: "authentication_failed", message: error.message }; if (authenticationFailed) diff --git a/packages/server/src/server/connected-hosts/link.ts b/packages/server/src/server/connected-hosts/link.ts index 85fe06ed4e..183ae4c220 100644 --- a/packages/server/src/server/connected-hosts/link.ts +++ b/packages/server/src/server/connected-hosts/link.ts @@ -1,5 +1,6 @@ import type { SessionDelivery } from "../session/owned-subscriptions/index.js"; -import { EventEmitter } from "node:events"; +import { VirtualSessionSocket } from "../websocket/virtual-session-socket.js"; +export { VirtualSessionSocket as VirtualHostSocket } from "../websocket/virtual-session-socket.js"; import { createHash } from "node:crypto"; import { encodeHostRouteFrame, type HostRouteFrame } from "@getpaseo/protocol/binary-frames/index"; import type { @@ -8,55 +9,9 @@ import type { SessionOutboundMessage, } from "@getpaseo/protocol/messages"; -/** A logical socket has no network resources or authority of its own. */ -export class VirtualHostSocket extends EventEmitter { - readyState = 1; - constructor( - private readonly output: (data: string | Uint8Array) => Promise, - private readonly ended: (code: number, reason: string) => void, - private readonly buffered: () => number, - private readonly transformHello: (data: string) => string, - ) { - super(); - } - get bufferedAmount(): number { - return this.buffered(); - } - private helloPending = true; - feed(data: string | Uint8Array): void { - if (this.readyState !== 1) return; - if (this.helloPending) { - if (typeof data !== "string") { - this.close(1008, "Expected hello"); - return; - } - try { - data = this.transformHello(data); - } catch { - this.close(1008, "Invalid hello"); - return; - } - this.helloPending = false; - } - this.emit("message", typeof data === "string" ? data : Buffer.from(data)); - } - async send(data: string | Uint8Array | ArrayBuffer): Promise { - if (this.readyState !== 1) throw new Error("Host channel closed"); - await this.output( - typeof data === "string" || data instanceof Uint8Array ? data : new Uint8Array(data), - ); - } - close(code = 1000, reason = "Host channel closed"): void { - if (this.readyState !== 1) return; - this.readyState = 3; - this.ended(code, reason); - this.emit("close", 1000, "Host channel closed"); - } -} - /** One authenticated physical source owns all its channels and their admission ceiling. */ export class DaemonLinkEndpoint { - private readonly channels = new Map(); + private readonly channels = new Map(); private closed = false; constructor( private readonly parent: { @@ -65,7 +20,7 @@ export class DaemonLinkEndpoint { routed(): boolean; bufferedAmount(): number | null; delivery: SessionDelivery; - admit(socket: VirtualHostSocket, permissions: readonly DaemonPermission[]): Promise; + admit(socket: VirtualSessionSocket, permissions: readonly DaemonPermission[]): Promise; }, ) {} handles(message: SessionInboundMessage): boolean { @@ -97,7 +52,7 @@ export class DaemonLinkEndpoint { error("resource_limit", "Too many host channels"); return; } - let socket: VirtualHostSocket; + let socket: VirtualSessionSocket; const owner = this.parent.delivery.begin("host-link", undefined, () => { this.channels.delete(owner.id); socket?.close(); @@ -105,7 +60,7 @@ export class DaemonLinkEndpoint { const id = owner.id; // Capture actual authenticated authority. The caller hello can only reduce it. const ceiling = [...this.parent.permissions()]; - socket = new VirtualHostSocket( + socket = new VirtualSessionSocket( async (data) => { const output: SessionOutboundMessage | null = typeof data === "string" diff --git a/packages/server/src/server/host-runtime/index.ts b/packages/server/src/server/host-runtime/index.ts new file mode 100644 index 0000000000..866eedbdda --- /dev/null +++ b/packages/server/src/server/host-runtime/index.ts @@ -0,0 +1,123 @@ +import { createPaseoApi, HostRuntime, type PaseoApi } from "@getpaseo/client"; +import { DaemonClient } from "@getpaseo/client/internal/daemon-client"; +import type { DaemonTransportFactory } from "@getpaseo/client/internal/daemon-client-transport-types"; +import { VirtualSessionSocket } from "../websocket/virtual-session-socket.js"; +import type { DaemonPermission } from "@getpaseo/protocol/messages"; +import type { SessionAdmission } from "../websocket-server.js"; + +export interface HostRuntimeAuthority { + permissionCeiling: readonly DaemonPermission[]; + resolveAdmission(): Promise; +} + +/** Admission is supplied by the host; this adapter never manufactures authority. */ +export async function createDaemonHostRuntime(input: { + attach(socket: VirtualSessionSocket): Promise; + clientId: string; + permissionCeiling: readonly DaemonPermission[]; + disposed(runtime: HostRuntime): void; +}): Promise { + const transportFactory: DaemonTransportFactory = () => { + const opens = new Set<() => void>(); + const closes = new Set<(event?: unknown) => void>(); + const errors = new Set<(error: unknown) => void>(); + const messages = new Set<(data: unknown, binary: boolean) => void>(); + const socket = new VirtualSessionSocket( + async (data) => { + for (const handler of messages) handler(data, typeof data !== "string"); + }, + (code, reason) => { + for (const handler of closes) handler({ code, reason }); + }, + () => 0, + (hello) => hello, + ); + void input.attach(socket).then( + () => { + if (socket.readyState === 1) for (const handler of opens) handler(); + return undefined; + }, + (error: unknown) => { + for (const handler of errors) handler(error); + socket.close(); + }, + ); + return { + send: (data) => socket.feed(data instanceof ArrayBuffer ? new Uint8Array(data) : data), + close: (code, reason) => socket.close(code, reason), + onOpen(handler) { + opens.add(handler); + return () => { + opens.delete(handler); + }; + }, + onClose(handler) { + closes.add(handler); + return () => { + closes.delete(handler); + }; + }, + onError(handler) { + errors.add(handler); + return () => { + errors.delete(handler); + }; + }, + onMessage(handler) { + messages.add(handler); + return () => { + messages.delete(handler); + }; + }, + }; + }; + const client = new DaemonClient({ + url: "ws://host-runtime/ws", + clientId: input.clientId, + sessionPermissions: [...input.permissionCeiling], + transportFactory, + reconnect: { enabled: false }, + }); + const origin = createPaseoApi(client); + const runtime = new OwnedHostRuntime(origin, client, input.disposed); + try { + await client.connect(); + client.setReconnectEnabled(true); + return runtime; + } catch (error) { + await runtime.dispose(); + throw error; + } +} + +class OwnedHostRuntime extends HostRuntime { + private ownedDisposal: Promise | null = null; + + constructor( + private readonly originApi: PaseoApi, + private readonly originClient: DaemonClient, + private readonly onDisposed: (runtime: HostRuntime) => void, + ) { + super(originApi); + } + + override dispose(): Promise { + if (this.ownedDisposal) return this.ownedDisposal; + this.ownedDisposal = (async () => { + try { + await super.dispose(); + } finally { + try { + await this.originApi.dispose(); + } finally { + try { + await this.originClient.close(); + } finally { + this.onDisposed(this); + } + } + } + })(); + return this.ownedDisposal; + } +} diff --git a/packages/server/src/server/plugins/connection-demand.e2e.test.ts b/packages/server/src/server/plugins/connection-demand.e2e.test.ts index 7c6cbfca28..ae79be56f6 100644 --- a/packages/server/src/server/plugins/connection-demand.e2e.test.ts +++ b/packages/server/src/server/plugins/connection-demand.e2e.test.ts @@ -5,10 +5,17 @@ import { expect, test } from "vitest"; import { createPaseoApi } from "@getpaseo/client"; import { DaemonClient } from "../test-utils/daemon-client.js"; import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js"; +import { resolveDaemonVersion } from "../daemon-version.js"; test("an RPC-only plugin receives no agent, project or provider data", async () => { const directory = await mkdtemp(path.join(tmpdir(), "paseo-quiet-plugin-")); - await writeFile(path.join(directory, "paseo-plugin.json"), JSON.stringify({ id: "quiet" })); + await writeFile( + path.join(directory, "paseo-plugin.json"), + JSON.stringify({ + id: "quiet", + requirements: { paseo: `>=${resolveDaemonVersion(import.meta.url)}` }, + }), + ); await writeFile( path.join(directory, "index.server.ts"), ` @@ -33,6 +40,7 @@ export default function contribute(server) { const legacy = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, capabilities: { + owned_subscriptions: false, selective_agent_timeline: false, explicit_event_subscriptions: false, provider_snapshot_references: false, @@ -65,7 +73,9 @@ export default function contribute(server) { await client.waitForFinish(agent.id); await client.refreshProvidersSnapshot({ cwd: directory }); const counts = await client.invokePluginRpc("quiet", "counts", {}); - expect(counts).toEqual({ status: 1 }); + expect( + Object.keys(counts as Record).filter((type) => type !== "status"), + ).toEqual([]); await legacy.getDaemonConfig(); expect(legacyTypes).toEqual( expect.arrayContaining(["project.update", "providers_snapshot_update", "agent_stream"]), diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 6fe3e8bf2d..00d90c7729 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -568,6 +568,7 @@ export interface SessionOptions { daemonVersion?: string; daemonRuntimeConfig?: DaemonRuntimeConfig; getWebSocketRuntimeMetrics?: () => DaemonWebSocketRuntimeDiagnosticSnapshot | null; + getHostObservationMetrics?: () => Record; } export type SessionLifecycleIntent = @@ -1075,6 +1076,7 @@ export class Session { daemonRuntimeConfig, getWebSocketRuntimeMetrics, getObservationMetrics: () => ({ + ...options.getHostObservationMetrics?.(), Registrations: this.delivery.registrationCount, "Source registrations": this.delivery.sourceRegistrationCount(), "Producer listeners": [ diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 2c8da4ed42..c89f35aa69 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -1,6 +1,7 @@ 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"; +import { createDaemonHostRuntime, type HostRuntimeAuthority } from "./host-runtime/index.js"; import { stat } from "node:fs/promises"; import type { CreationSnapshot } from "@getpaseo/protocol/messages"; import { CreationService } from "./creation/index.js"; @@ -457,12 +458,11 @@ interface ReconnectableSessionConnection extends SessionConnectionBase { externalDisconnectCleanupTimeout: ReturnType | null; } -interface PluginSessionConnection extends SessionConnectionBase { - lifecycle: "ephemeral-plugin"; - pluginId: string; +interface OwnedSessionConnection extends SessionConnectionBase { + lifecycle: "ephemeral"; } -type SessionConnection = ReconnectableSessionConnection | PluginSessionConnection; +type SessionConnection = ReconnectableSessionConnection | OwnedSessionConnection; interface SocketSessionOptions { sourceId?: string; @@ -531,6 +531,37 @@ function requireWebSocketServices(params: { * WebSocket server that only accepts sockets + parses/forwards messages to the session layer. */ export class VoiceAssistantWebSocketServer { + private readonly hostRuntimes = new Set>>(); + private readonly ownedSockets = new WeakSet(); + + public async createHostRuntime(authority: HostRuntimeAuthority) { + const runtime = await createDaemonHostRuntime({ + clientId: `host-runtime:${randomUUID()}`, + permissionCeiling: authority.permissionCeiling, + disposed: (finishedRuntime) => { + this.hostRuntimes.delete(finishedRuntime); + }, + attach: async (socket) => { + const admission = await authority.resolveAdmission(); + if (admission.hubExecutionAgents) { + throw Object.assign( + new Error("Host runtime does not support execution-scoped admission"), + { + code: "unsupported_capability", + }, + ); + } + this.ownedSockets.add(socket); + await this.attachExternalSocket(socket, undefined, admission); + }, + }); + if (this.connectionLifecycle === "stopping") { + await runtime.dispose(); + throw new Error("Cannot create host runtime while shutting down"); + } + this.hostRuntimes.add(runtime); + return runtime; + } private readonly logger: pino.Logger; private readonly wss: WebSocketServer; private readonly hostLinks = new Map(); @@ -1041,6 +1072,7 @@ export class VoiceAssistantWebSocketServer { resolve = finish; }); this.pluginSocketIds.set(ws, pluginId); + this.ownedSockets.add(ws); this.pluginSocketCleanup.set(ws, resolve); try { await this.attachSocket(ws, undefined, undefined, true, OWNER_SESSION_ADMISSION); @@ -1067,7 +1099,10 @@ export class VoiceAssistantWebSocketServer { pending.admission = { ...pending.admission, permissions }; } } - for (const connection of new Set(this.externalSessionsByKey.values())) { + for (const connection of new Set([ + ...this.externalSessionsByKey.values(), + ...this.sessions.values(), + ])) { if (connection.principalId === principalId) { connection.session.setPermissions( connection.permissionCeiling @@ -1090,6 +1125,8 @@ export class VoiceAssistantWebSocketServer { public async close(): Promise { this.prepareForShutdown(); + await Promise.allSettled([...this.hostRuntimes].map((runtime) => runtime.dispose())); + this.hostRuntimes.clear(); await this.connectedHosts.close(); this.unsubscribeSpeechReadiness?.(); this.unsubscribeSpeechReadiness = null; @@ -1432,11 +1469,20 @@ export class VoiceAssistantWebSocketServer { appVersion: string | null; clientCapabilities: Record | null; connectionLogger: pino.Logger; - lifecycle: { kind: "reconnectable" } | { kind: "ephemeral-plugin"; pluginId: string }; + lifecycle: "reconnectable" | "ephemeral"; + sourceId: string; admission: SessionAdmission; }): SessionConnection { - const { ws, clientId, appVersion, clientCapabilities, connectionLogger, lifecycle, admission } = - params; + const { + ws, + clientId, + appVersion, + clientCapabilities, + connectionLogger, + lifecycle, + admission, + sourceId, + } = params; let connection: SessionConnection | null = null; const session = this.createSocketSession({ @@ -1444,10 +1490,7 @@ export class VoiceAssistantWebSocketServer { appVersion, clientCapabilities, permissions: admission.permissions, - sourceId: - lifecycle.kind === "ephemeral-plugin" - ? `plugin:${lifecycle.pluginId}` - : admission.principalId, + sourceId, connectionLogger, onMessage: (msg) => { if (!connection) { @@ -1509,8 +1552,8 @@ export class VoiceAssistantWebSocketServer { sockets: new Set([ws]), }; connection = - lifecycle.kind === "ephemeral-plugin" - ? { ...base, lifecycle: "ephemeral-plugin", pluginId: lifecycle.pluginId } + lifecycle === "ephemeral" + ? { ...base, lifecycle: "ephemeral" } : { ...base, lifecycle: "reconnectable", externalDisconnectCleanupTimeout: null }; session.updateClientCapabilities(clientCapabilities, ws, appVersion); return connection; @@ -1618,6 +1661,14 @@ export class VoiceAssistantWebSocketServer { daemonVersion: this.daemonVersion, daemonRuntimeConfig: this.daemonRuntimeConfig, getWebSocketRuntimeMetrics: () => this.lastRuntimeMetricsSnapshot, + getHostObservationMetrics: () => ({ + "Host runtimes": this.hostRuntimes.size, + "Owned sessions": new Set( + [...this.sessions.entries()] + .filter(([socket]) => this.ownedSockets.has(socket)) + .map(([, connection]) => connection), + ).size, + }), }); } @@ -1697,7 +1748,9 @@ export class VoiceAssistantWebSocketServer { pending.identity.appVersion = message.appVersion; } const sessionKey = sessionConnectionKey(admitted.principalId, clientId); - const existing = pluginId ? undefined : this.externalSessionsByKey.get(sessionKey); + const existing = this.ownedSockets.has(ws) + ? undefined + : this.externalSessionsByKey.get(sessionKey); if (existing) { this.resumeSession({ ws, message, pending, existing }); return; @@ -1711,7 +1764,8 @@ export class VoiceAssistantWebSocketServer { appVersion: message.appVersion ?? null, clientCapabilities: message.capabilities ?? null, connectionLogger, - lifecycle: pluginId ? { kind: "ephemeral-plugin", pluginId } : { kind: "reconnectable" }, + lifecycle: this.socketLifetime(ws), + sourceId: pluginId ? `plugin:${pluginId}` : admitted.principalId, admission: admitted, }); connection.permissionCeiling = message.sessionPermissions; @@ -1733,6 +1787,10 @@ export class VoiceAssistantWebSocketServer { ); } + private socketLifetime(ws: WebSocketLike): SessionConnection["lifecycle"] { + return this.ownedSockets.has(ws) ? "ephemeral" : "reconnectable"; + } + private handleHelloSafely(params: { ws: WebSocketLike; message: WSHelloMessage; @@ -2186,9 +2244,9 @@ export class VoiceAssistantWebSocketServer { this.socketIdentities.delete(ws); if (connection.sockets.size === 0) { - if (connection.lifecycle === "ephemeral-plugin") { + if (connection.lifecycle === "ephemeral") { this.pluginSocketIds.delete(ws); - await this.cleanupConnection(connection, "Plugin session disconnected"); + await this.cleanupConnection(connection, "Owned session disconnected"); this.finishPluginSocketCleanup(ws); return; } diff --git a/packages/server/src/server/websocket/virtual-session-socket.ts b/packages/server/src/server/websocket/virtual-session-socket.ts new file mode 100644 index 0000000000..292c264c3f --- /dev/null +++ b/packages/server/src/server/websocket/virtual-session-socket.ts @@ -0,0 +1,47 @@ +import { EventEmitter } from "node:events"; + +/** A logical socket has no network resources or authority of its own. */ +export class VirtualSessionSocket extends EventEmitter { + readyState = 1; + constructor( + private readonly output: (data: string | Uint8Array) => Promise, + private readonly ended: (code: number, reason: string) => void, + private readonly buffered: () => number, + private readonly transformHello: (data: string) => string, + ) { + super(); + } + get bufferedAmount(): number { + return this.buffered(); + } + private helloPending = true; + feed(data: string | Uint8Array): void { + if (this.readyState !== 1) return; + if (this.helloPending) { + if (typeof data !== "string") { + this.close(1008, "Expected hello"); + return; + } + try { + data = this.transformHello(data); + } catch { + this.close(1008, "Invalid hello"); + return; + } + this.helloPending = false; + } + this.emit("message", typeof data === "string" ? data : Buffer.from(data)); + } + async send(data: string | Uint8Array | ArrayBuffer): Promise { + if (this.readyState !== 1) throw new Error("Host channel closed"); + await this.output( + typeof data === "string" || data instanceof Uint8Array ? data : new Uint8Array(data), + ); + } + close(code = 1000, reason = "Host channel closed"): void { + if (this.readyState !== 1) return; + this.readyState = 3; + this.ended(code, reason); + this.emit("close", 1000, "Host channel closed"); + } +} diff --git a/public-docs/sdk/reference.md b/public-docs/sdk/reference.md index 6e606e35ba..7d852ef094 100644 --- a/public-docs/sdk/reference.md +++ b/public-docs/sdk/reference.md @@ -80,6 +80,8 @@ See [background workspaces](./workspaces.md) for caller inheritance and inclusiv | Member | Result | Behavior | | ------------------------------ | --------------------------------- | ------------------------------------------------------------------------------------------------------- | | `id` | `string` | Stable daemon agent ID. | +| `serverId` | `string` | Accepted daemon identity, retained across disconnects. Connect before reading it. | +| `address` | `{ serverId, agentId }` | Complete agent identity, independent of the connection route. | | `workspaceId` | `string \| null` | Current workspace placement. | | `cwd` | `string \| null` | Current working directory. | | `status` | Agent status or `null` | Current lifecycle status. | @@ -260,6 +262,29 @@ Registration and host acquisition failures expose a stable `code`: `host_unknown unchanged. Multiplexing is a transport; consumers use the same client/API operations and lifecycle. Discovery does not grant execution authority. +### Resolve agent addresses + +Use `HostRuntime` when a workflow holds complete agent addresses. Each runtime resolves through its +origin API, preserving that route's authority even when another direct connection exists. + +```ts +const runtime = client.hosts.runtime(); +try { + const agent = await runtime.agent({ serverId: "srv_build_box", agentId: "agent_on_build_box" }); + await agent.refresh(); + await agent.send("Continue the investigation"); +} finally { + await runtime.dispose(); +} +``` + +`directory()` returns the origin's host directory. `client(serverId)` acquires an independent, +caller-owned API scope; dispose it after use. `agent(address)` borrows a runtime-owned scope. +Disposing the runtime releases its scopes and pending acquisitions, leaving the origin API open. +Agent addresses and accepted snapshots remain readable; operations on disposed handles reject. +A connection that reaches a different daemon on reconnect fails with `server_id_mismatch` before +queued work executes. Create a new client after explicitly choosing the new destination. + ## Errors and cleanup Connection, validation, rejection, and timeout failures reject their promise. Turn outcomes are returned through `PaseoAgentRunResult.status` because permission and provider errors are expected agent states.