From 888b54bb44c72a43702002178e0fdeaebb5e3e85 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Tue, 20 Oct 2026 09:32:17 +0200 Subject: [PATCH] feat(hosts): resolve qualified agent addresses through shared runtimes Keep agent identity and origin authority stable through disconnects, destination changes, and scoped disposal. Share the resolver with trusted daemon consumers and plugins, using ordinary sessions and per-admission permission ceilings. Expose MCP host discovery through that boundary as the first checkpoint toward durable fleet subagents. --- packages/client/src/daemon-client.test.ts | 3 +- packages/client/src/daemon-client.ts | 14 + packages/client/src/hosts/index.ts | 13 +- packages/client/src/hosts/runtime.ts | 83 ++++++ packages/client/src/index.ts | 37 ++- packages/protocol/src/addresses.ts | 14 + .../src/server/agent/tools/paseo-tools.ts | 21 ++ packages/server/src/server/bootstrap.ts | 26 ++ .../server/connected-hosts/hosts.e2e.test.ts | 251 +++++++++++++++++- .../src/server/connected-hosts/index.ts | 8 +- .../server/src/server/connected-hosts/link.ts | 57 +--- .../server/src/server/host-runtime/index.ts | 123 +++++++++ .../plugins/connection-demand.e2e.test.ts | 14 +- packages/server/src/server/session.ts | 2 + .../server/src/server/websocket-server.ts | 94 +++++-- .../websocket/virtual-session-socket.ts | 47 ++++ public-docs/sdk/reference.md | 25 ++ 17 files changed, 751 insertions(+), 81 deletions(-) create mode 100644 packages/client/src/hosts/runtime.ts create mode 100644 packages/protocol/src/addresses.ts create mode 100644 packages/server/src/server/host-runtime/index.ts create mode 100644 packages/server/src/server/websocket/virtual-session-socket.ts 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.