feat(hosts): connect to saved hosts through a daemon

packages/server/src/server/authorization/operation-permissions.ts

@@ -6,6 +6,13 @@ type OutboundOperation = SessionOutboundMessage["type"];
 export type PermissionRequirement = DaemonPermission | readonly DaemonPermission[] | null;
 
 const INBOUND_PERMISSION = {
+  "hosts.list.request": "daemon.read",
+  "hosts.add.request": "tunnel.manage",
+  "hosts.update.request": "tunnel.manage",
+  "hosts.remove.request": "tunnel.manage",
+  "hosts.route.open.request": null,
+  "hosts.route.frame.request": null,
+  "hosts.route.close.request": null,
   abort_request: "workspace.write",
   "agent.config.apply.request": ["workspace.write", "hub.execute"],
   "agent.detach.request": "workspace.write",
@@ -216,6 +223,14 @@ const INBOUND_PERMISSION = {
 } as const satisfies Record<InboundOperation, PermissionRequirement>;
 
 const OUTBOUND_PERMISSION = {
+  "hosts.list.response": "daemon.read",
+  "hosts.changed": "daemon.read",
+  "hosts.add.response": "tunnel.manage",
+  "hosts.update.response": "tunnel.manage",
+  "hosts.remove.response": "tunnel.manage",
+  "hosts.route.open.response": null,
+  "hosts.route.frame": null,
+  "hosts.route.closed": null,
   "agent.create.response": ["workspace.write", "hub.execute"],
   "agent.create.update": ["workspace.write", "hub.execute"],
   "workspace.create.update": "workspace.manage",

packages/server/src/server/connected-hosts/hosts.e2e.test.ts

@@ -0,0 +1,688 @@
+import { afterEach, expect, test } from "vitest";
+import { createPaseoApi } from "@getpaseo/client";
+import { createHostClient } from "@getpaseo/client/node";
+import { createTestPaseoDaemon, type TestPaseoDaemon } from "../test-utils/paseo-daemon.js";
+import type { HostDirectory } from "@getpaseo/protocol/hosts";
+
+const cleanup: (() => Promise<unknown>)[] = [];
+afterEach(async () => {
+  for (const close of cleanup.splice(0).toReversed()) await close();
+});
+async function daemon(): Promise<TestPaseoDaemon> {
+  const instance = await createTestPaseoDaemon();
+  cleanup.push(() => instance.close());
+  return instance;
+}
+async function api(instance: TestPaseoDaemon) {
+  const client = createHostClient(`127.0.0.1:${instance.port}`, {
+    clientId: crypto.randomUUID(),
+    reconnect: { enabled: false },
+  });
+  await client.connect();
+  const paseo = createPaseoApi(client);
+  cleanup.push(async () => {
+    await paseo.dispose();
+    await client.close();
+  });
+  return { paseo, client };
+}
+async function online(paseo: ReturnType<typeof createPaseoApi>): Promise<HostDirectory> {
+  let directory!: HostDirectory;
+  await expect
+    .poll(
+      async () => {
+        directory = await paseo.hosts.list();
+        return directory.hosts[0]?.status;
+      },
+      { timeout: 15000 },
+    )
+    .toBe("online");
+  return directory;
+}
+
+test("SDK saves unreachable hosts, rejects syntax and preserves source identity without exposing secrets", async () => {
+  const a = await daemon();
+  const { paseo } = await api(a);
+  await expect(paseo.hosts.add({ connection: "ssh://-bad" })).rejects.toMatchObject({
+    code: "invalid_connection",
+  });
+  const saved = await paseo.hosts.add({
+    connection: "tcp://127.0.0.1:1?password=private-secret",
+    label: "Unavailable",
+    metadata: { team: "infra" },
+    sourceKey: "stable",
+  });
+  expect(saved).toMatchObject({
+    registrationId: expect.stringMatching(/^hst_/),
+    serverId: null,
+    label: "Unavailable",
+    source: { id: "owner", key: "stable" },
+    metadata: { team: "infra" },
+    enabled: true,
+  });
+  expect(JSON.stringify(await paseo.hosts.list())).not.toContain("private-secret");
+  await paseo.hosts.update(saved.registrationId, { enabled: false });
+  expect(
+    await paseo.hosts.add({
+      connection: "tcp://127.0.0.1:1?password=private-secret",
+      label: "Unavailable",
+      metadata: { team: "infra" },
+      sourceKey: "stable",
+    }),
+  ).toMatchObject({ registrationId: saved.registrationId, enabled: false, status: "disabled" });
+  await expect(
+    paseo.hosts.add({ connection: "127.0.0.1:2", sourceKey: "stable" }),
+  ).rejects.toMatchObject({ code: "source_conflict" });
+  await paseo.hosts.remove(saved.registrationId);
+  expect((await paseo.hosts.list()).hosts).toEqual([]);
+});
+
+test("ordinary routed APIs isolate scopes and enforce one-hop, disable and origin disposal", async () => {
+  const b = await daemon();
+  const a = await daemon();
+  const { paseo } = await api(a);
+  const observation = paseo.hosts.observe();
+  expect((await observation.ready).hosts).toEqual([]);
+  const saved = await paseo.hosts.add({ connection: `127.0.0.1:${b.port}`, label: "B" });
+  const directory = await online(paseo);
+  const serverId = directory.hosts[0]!.serverId!;
+  const first = await paseo.hosts.client(serverId);
+  const second = await paseo.hosts.client(serverId);
+  const workspaces = await second.workspaces.list({ subscribe: {} });
+  expect(workspaces.entries).toEqual([]);
+  await first.dispose();
+  expect((await second.workspaces.list()).entries).toEqual([]);
+  await expect(second.hosts.client(directory.self.serverId)).rejects.toMatchObject({
+    code: "unsupported_capability",
+  });
+  await paseo.hosts.update(saved.registrationId, { enabled: false });
+  await expect(paseo.hosts.client(serverId)).rejects.toMatchObject({ code: "host_disabled" });
+  await paseo.hosts.update(saved.registrationId, { enabled: true });
+  await online(paseo);
+  await expect
+    .poll(
+      async () => {
+        try {
+          return (await second.workspaces.list()).entries.length;
+        } catch {
+          return -1;
+        }
+      },
+      { timeout: 20000 },
+    )
+    .toBe(0);
+  await second.dispose();
+  await observation.release();
+  await paseo.dispose();
+});
+
+test("routed authority is bounded by the caller, with isolated simultaneous sessions", async () => {
+  const b = await daemon();
+  const a = await daemon();
+  const owner = await api(a);
+  await owner.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` });
+  const directory = await online(owner.paseo);
+  const serverId = directory.hosts[0]!.serverId!;
+  const viewerClient = createHostClient(`127.0.0.1:${a.port}`, {
+    clientId: "overlapping",
+    sessionPermissions: ["daemon.read", "workspace.read"],
+    reconnect: { enabled: false },
+  });
+  await viewerClient.connect();
+  const viewer = createPaseoApi(viewerClient);
+  cleanup.push(async () => {
+    await viewer.dispose();
+    await viewerClient.close();
+  });
+  const remote = await viewer.hosts.client(serverId);
+  expect((await remote.workspaces.list()).entries).toEqual([]);
+  await expect(
+    remote.workspaces.create({ source: { kind: "directory", path: b.paseoHome } }),
+  ).rejects.toMatchObject({ code: "permission_denied" });
+  await expect(viewer.hosts.add({ connection: "127.0.0.1:1" })).rejects.toMatchObject({
+    code: "permission_denied",
+  });
+  expect((await owner.paseo.hosts.list()).hosts).toHaveLength(1);
+  await remote.dispose();
+});
+
+test("distinct origin sessions cannot collide through a caller-selected route scope, including terminal binary output", async () => {
+  const b = await daemon();
+  const a = await daemon();
+  const admin = await api(a);
+  await admin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` });
+  const serverId = (await online(admin.paseo)).hosts[0]!.serverId!;
+  const remote = await admin.paseo.hosts.client(serverId);
+  const workspace = await remote.workspaces.create({
+    source: { kind: "directory", path: b.staticDir },
+  });
+  const one = await workspace.terminals.create({
+    command: process.execPath,
+    args: ["-e", "process.stdin.on('data', data => process.stdout.write(data))"],
+    name: "First",
+  });
+  const two = await workspace.terminals.create({
+    command: process.execPath,
+    args: ["-e", "process.stdin.on('data', data => process.stdout.write(data))"],
+    name: "Second",
+  });
+  const firstOrigin = await api(a);
+  const secondOrigin = await api(a);
+  const first = await RoutePeer.open(firstOrigin.client, serverId);
+  const second = await RoutePeer.open(secondOrigin.client, serverId);
+  await first.subscribeTerminal(one.id);
+  await second.subscribeTerminal(two.id);
+  first.input("FIRST_ONLY\n");
+  second.input("SECOND_ONLY\n");
+  await expect.poll(() => first.output()).toContain("FIRST_ONLY");
+  await expect.poll(() => second.output()).toContain("SECOND_ONLY");
+  expect(first.output()).not.toContain("SECOND_ONLY");
+  expect(second.output()).not.toContain("FIRST_ONLY");
+  await first.close();
+  second.input("STILL_SECOND\n");
+  await expect.poll(() => second.output()).toContain("STILL_SECOND");
+  await second.close();
+  await one.kill();
+  await two.kill();
+  await remote.dispose();
+});
+
+import {
+  decodeTerminalStreamFrame,
+  encodeTerminalStreamFrame,
+  TerminalStreamOpcode,
+} from "@getpaseo/protocol/binary-frames/index";
+import {
+  WSOutboundMessageSchema,
+  type SessionOutboundMessage,
+  type SessionInboundMessage,
+} from "@getpaseo/protocol/messages";
+import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
+class RoutePeer {
+  readonly messages: SessionOutboundMessage[] = [];
+  readonly bytes: Uint8Array[] = [];
+  private slot = -1;
+  private detach = () => {};
+  private constructor(
+    private readonly origin: DaemonClient,
+    private readonly routeId: string,
+  ) {
+    this.detach = origin.subscribeRawMessages((message) => {
+      if (message.type !== "hosts.route.frame" || message.payload.subscriptionId !== routeId)
+        return;
+      if (message.payload.binary) this.bytes.push(Buffer.from(message.payload.data, "base64"));
+      else {
+        const frame = WSOutboundMessageSchema.parse(JSON.parse(message.payload.data));
+        if (frame.type === "session") this.messages.push(frame.message);
+      }
+    });
+  }
+  static async open(origin: DaemonClient, serverId: string) {
+    const peer = new RoutePeer(
+      origin,
+      await origin.openHostRoute(serverId, "same-untrusted-scope"),
+    );
+    origin.sendHostFrame(
+      peer.routeId,
+      JSON.stringify({
+        type: "hello",
+        clientId: "same-untrusted-client",
+        clientType: "cli",
+        protocolVersion: 1,
+        capabilities: { owned_subscriptions: true },
+      }),
+      false,
+    );
+    await expect
+      .poll(
+        () =>
+          peer.messages.filter(
+            (message) => message.type === "status" && message.payload.status === "server_info",
+          ).length,
+      )
+      .toBe(1);
+    return peer;
+  }
+  send(message: SessionInboundMessage) {
+    this.origin.sendHostFrame(this.routeId, JSON.stringify({ type: "session", message }), false);
+  }
+  async subscribeTerminal(terminalId: string) {
+    this.send({ type: "subscribe_terminal_request", terminalId, requestId: "same-request" });
+    await expect
+      .poll(
+        () =>
+          this.messages.filter((message) => message.type === "subscribe_terminal_response").length,
+      )
+      .toBe(1);
+    const response = this.messages.find(
+      (message) => message.type === "subscribe_terminal_response",
+    );
+    if (response?.type !== "subscribe_terminal_response" || response.payload.slot === null)
+      throw new Error("Missing terminal response");
+    this.slot = response.payload.slot!;
+  }
+  input(text: string) {
+    this.origin.sendHostFrame(
+      this.routeId,
+      Buffer.from(
+        encodeTerminalStreamFrame({
+          opcode: TerminalStreamOpcode.Input,
+          slot: this.slot,
+          payload: text,
+        }),
+      ).toString("base64"),
+      true,
+    );
+  }
+  output() {
+    return this.bytes
+      .flatMap((bytes) => {
+        const frame = decodeTerminalStreamFrame(bytes);
+        return frame?.opcode === TerminalStreamOpcode.Output
+          ? [new TextDecoder().decode(frame.payload)]
+          : [];
+      })
+      .join("");
+  }
+  async close() {
+    this.detach();
+    this.origin.closeHostRoute(this.routeId);
+  }
+}
+
+import { mkdir, writeFile, stat, mkdtemp, rm } from "node:fs/promises";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
+import { resolveDaemonVersion } from "../daemon-version.js";
+
+test("identity discovery failures remain inspectable and changing a pinned destination requires explicit replacement", async () => {
+  const a = await daemon();
+  const b = await daemon();
+  const c = await daemon();
+  const { paseo } = await api(a);
+  const self = await paseo.hosts.add({ connection: `127.0.0.1:${a.port}` });
+  await expect.poll(async () => (await paseo.hosts.list()).hosts[0]!.error?.code).toBe("self_host");
+  expect((await paseo.hosts.list()).hosts[0]!.serverId).toBe(a.daemon.getServerId());
+  await paseo.hosts.remove(self.registrationId);
+  const original = await paseo.hosts.add({ connection: `127.0.0.1:${b.port}` });
+  const bId = (await online(paseo)).hosts[0]!.serverId;
+  const duplicate = await paseo.hosts.add({ connection: `ws://127.0.0.1:${b.port}/ws` });
+  await expect
+    .poll(async () => (await paseo.hosts.list()).hosts[1]!.error?.code)
+    .toBe("duplicate_host");
+  expect((await paseo.hosts.list()).hosts[1]!.serverId).toBe(bId);
+  await paseo.hosts.remove(duplicate.registrationId);
+  await paseo.hosts.update(original.registrationId, { connection: `127.0.0.1:${c.port}` });
+  await expect
+    .poll(async () => (await paseo.hosts.list()).hosts[0]!.error?.code)
+    .toBe("server_id_mismatch");
+  expect((await paseo.hosts.list()).hosts[0]!.serverId).toBe(bId);
+  await paseo.hosts.update(original.registrationId, {
+    connection: `127.0.0.1:${c.port}`,
+    replaceIdentity: true,
+  });
+  expect((await online(paseo)).hosts[0]!.serverId).toBe(c.daemon.getServerId());
+});
+
+test("disabled host configuration and source survive a fresh daemon instance", async () => {
+  const root = await mkdtemp(join(tmpdir(), "paseo-host-restart-"));
+  cleanup.push(() => rm(root, { recursive: true, force: true }));
+  const a = await createTestPaseoDaemon({ paseoHomeRoot: root, cleanup: false });
+  cleanup.push(() => a.close());
+  const { paseo } = await api(a);
+  const host = await paseo.hosts.add({
+    connection: "tcp://127.0.0.1:1?password=secret",
+    label: "Persistent",
+    metadata: { team: "infra" },
+    sourceKey: "stable",
+  });
+  await paseo.hosts.update(host.registrationId, { enabled: false });
+  expect((await stat(join(a.paseoHome, "hosts.json"))).mode & 0o777).toBe(0o600);
+  await a.close();
+  const restarted = await createTestPaseoDaemon({ paseoHomeRoot: root, cleanup: false });
+  cleanup.push(() => restarted.close());
+  const next = await api(restarted);
+  expect((await next.paseo.hosts.list()).hosts).toEqual([
+    expect.objectContaining({
+      registrationId: host.registrationId,
+      label: "Persistent",
+      metadata: { team: "infra" },
+      source: { id: "owner", key: "stable" },
+      enabled: false,
+      status: "disabled",
+    }),
+  ]);
+});
+
+test("server plugins use ordinary remote SDK operations and retain qualified source ownership on B", async () => {
+  const b = await daemon();
+  const a = await daemon();
+  const { paseo, client } = await api(a);
+  await paseo.hosts.add({ connection: `127.0.0.1:${b.port}` });
+  const serverId = (await online(paseo)).hosts[0]!.serverId!;
+  const directory = join(a.staticDir, "host-plugin");
+  await mkdir(directory);
+  await writeFile(
+    join(directory, "paseo-plugin.json"),
+    JSON.stringify({
+      id: "connected-host-test",
+      requirements: { paseo: `>=${resolveDaemonVersion(import.meta.url)}` },
+    }),
+  );
+  await writeFile(
+    join(directory, "index.server.ts"),
+    `
+import { defineRpc } from "@getpaseo/plugin";
+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);
+    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(); }
+  });
+  return () => {};
+}`,
+  );
+  await client.patchDaemonConfig({ pluginsEnabled: true });
+  await client.installDirectoryPlugin(directory);
+  const result = await client.invokePluginRpc("connected-host-test", "operate", {
+    serverId,
+    path: b.staticDir,
+  });
+  expect(result).toEqual({
+    workspaceId: expect.stringMatching(/^wks_/),
+    registrationId: expect.stringMatching(/^hst_/),
+    sourceId: `route:owner:${a.daemon.getServerId()}:plugin%3Aconnected-host-test`,
+  });
+  const direct = await api(b);
+  expect((await direct.paseo.workspaces.list()).entries).toHaveLength(1);
+  expect((await direct.paseo.hosts.list()).hosts[0]).toMatchObject({
+    source: {
+      id: `route:owner:${a.daemon.getServerId()}:plugin%3Aconnected-host-test`,
+      key: "plugin-stable",
+    },
+    metadata: { team: "plugin" },
+  });
+  await client.removePlugin("connected-host-test");
+  expect((await direct.paseo.workspaces.list()).entries).toHaveLength(1);
+});
+
+import { execFile } from "node:child_process";
+import { promisify } from "node:util";
+import { fileURLToPath } from "node:url";
+
+test("CLI host commands manage the selected daemon and emit secret-free structured output", async () => {
+  const a = await daemon();
+  const cli = async (...args: string[]) =>
+    JSON.parse(
+      (
+        await promisify(execFile)(
+          process.execPath,
+          [
+            "--import",
+            "tsx",
+            fileURLToPath(new URL("../../../../cli/src/index.ts", import.meta.url)),
+            "--host",
+            `127.0.0.1:${a.port}`,
+            "host",
+            ...args,
+            "--json",
+          ],
+          { timeout: 20000 },
+        )
+      ).stdout,
+    );
+  const saved = await cli(
+    "add",
+    "tcp://127.0.0.1:1?password=cli-secret",
+    "--label",
+    "CLI host",
+    "--meta",
+    "team=infra",
+  );
+  expect(saved).toMatchObject({
+    label: "CLI host",
+    metadata: { team: "infra" },
+    registrationId: expect.stringMatching(/^hst_/),
+  });
+  expect(JSON.stringify(await cli("ls"))).not.toContain("cli-secret");
+  expect(
+    await cli(
+      "update",
+      saved.registrationId,
+      "--meta",
+      "team=tools",
+      "--meta",
+      "other=value",
+      "--remove-meta",
+      "team",
+    ),
+  ).toMatchObject({ metadata: { other: "value" } });
+  expect(await cli("disable", saved.registrationId)).toMatchObject({
+    status: "disabled",
+    enabled: false,
+  });
+  expect(await cli("inspect", saved.registrationId)).toMatchObject({
+    enabled: false,
+    targetDescription: "tcp://127.0.0.1:1",
+  });
+  expect(await cli("enable", saved.registrationId)).toMatchObject({ enabled: true });
+  await cli("rm", saved.registrationId);
+  expect(await cli("ls")).toEqual([]);
+}, 60000);
+
+test("resuming an A session with narrower grants terminates existing routes and restores observations with current authority", async () => {
+  const b = await daemon();
+  const a = await daemon();
+  const admin = await api(a);
+  await admin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` });
+  const serverId = (await online(admin.paseo)).hosts[0]!.serverId!;
+  const origin = createHostClient(`127.0.0.1:${a.port}`, {
+    clientId: "resumed-authority",
+    reconnect: { enabled: false },
+  });
+  await origin.connect();
+  const paseo = createPaseoApi(origin);
+  cleanup.push(async () => {
+    await paseo.dispose();
+    await origin.close();
+  });
+  const remote = await paseo.hosts.client(serverId);
+  const observed = await remote.workspaces.list({ subscribe: {} });
+  const snapshots: unknown[] = [];
+  observed.subscription.subscribe({
+    snapshot: (snapshot) => snapshots.push(snapshot),
+    update() {},
+  });
+  const restricted = createHostClient(`127.0.0.1:${a.port}`, {
+    clientId: "resumed-authority",
+    sessionPermissions: ["daemon.read", "workspace.read"],
+    reconnect: { enabled: false },
+  });
+  await restricted.connect();
+  cleanup.push(() => restricted.close());
+  await expect.poll(() => snapshots.length, { timeout: 15000 }).toBe(2);
+  await expect(
+    remote.workspaces.create({ source: { kind: "directory", path: b.staticDir } }),
+  ).rejects.toMatchObject({ code: "permission_denied" });
+  expect((await remote.workspaces.list()).entries).toEqual([]);
+  await remote.dispose();
+});
+
+import { WebSocketServer } from "ws";
+import { createServer } from "node:http";
+import { once } from "node:events";
+import { hashSync } from "bcryptjs";
+
+test("old capabilities fail before execution and authentication errors recover through credential replacement", async () => {
+  const http = createServer();
+  const ws = new WebSocketServer({ server: http });
+  const operations: string[] = [];
+  ws.on("connection", (socket) =>
+    socket.on("message", (data) => {
+      const frame = JSON.parse(data.toString());
+      if (frame.type === "hello")
+        socket.send(
+          JSON.stringify({
+            type: "session",
+            message: {
+              type: "status",
+              payload: {
+                status: "server_info",
+                serverId: "srv_old",
+                hostname: "old",
+                version: "0.1.0",
+              },
+            },
+          }),
+        );
+      else if (frame.type === "session") operations.push(frame.message.type);
+    }),
+  );
+  http.listen(0, "127.0.0.1");
+  await once(http, "listening");
+  cleanup.push(async () => {
+    for (const socket of ws.clients) socket.terminate();
+    await new Promise<void>((resolve) => ws.close(() => resolve()));
+    await new Promise<void>((resolve) => http.close(() => resolve()));
+  });
+  const address = http.address();
+  if (!address || typeof address === "string") throw new Error("Missing old daemon port");
+  const old = createHostClient(`127.0.0.1:${address.port}`, {
+    clientId: "old-capability",
+    reconnect: { enabled: false },
+  });
+  await old.connect();
+  cleanup.push(() => old.close());
+  await expect(createPaseoApi(old).hosts.add({ connection: "127.0.0.1:1" })).rejects.toMatchObject({
+    code: "unsupported_capability",
+  });
+  expect(operations).toEqual([]);
+  const a = await daemon();
+  const { paseo } = await api(a);
+  const registration = await paseo.hosts.add({ connection: `127.0.0.1:${address.port}` });
+  await expect
+    .poll(async () => (await paseo.hosts.list()).hosts[0]!.error?.code)
+    .toBe("unsupported_capability");
+  await expect(paseo.hosts.client("srv_old")).rejects.toMatchObject({
+    code: "unsupported_capability",
+  });
+  expect(operations).toEqual([]);
+  await paseo.hosts.remove(registration.registrationId);
+  const b = await createTestPaseoDaemon({ auth: { password: hashSync("right-password", 4) } });
+  cleanup.push(() => b.close());
+  const saved = await paseo.hosts.add({
+    connection: `tcp://127.0.0.1:${b.port}?password=wrong-password`,
+  });
+  await expect
+    .poll(async () => (await paseo.hosts.list()).hosts[0]!.error?.code)
+    .toBe("authentication_failed");
+  expect(JSON.stringify(await paseo.hosts.list())).not.toContain("wrong-password");
+  await paseo.hosts.update(saved.registrationId, {
+    connection: `tcp://127.0.0.1:${b.port}?password=right-password`,
+  });
+  expect((await online(paseo)).hosts[0]!.serverId).toBe(b.daemon.getServerId());
+});
+
+test("an unreachable saved target becomes online after its daemon starts without being added again", async () => {
+  const reservation = createServer();
+  reservation.listen(0, "127.0.0.1");
+  await once(reservation, "listening");
+  const address = reservation.address();
+  if (!address || typeof address === "string") throw new Error("Missing reserved port");
+  await new Promise<void>((resolve) => reservation.close(() => resolve()));
+  const a = await daemon();
+  const { paseo } = await api(a);
+  const host = await paseo.hosts.add({ connection: `127.0.0.1:${address.port}` });
+  await expect.poll(async () => (await paseo.hosts.list()).hosts[0]!.status).toBe("offline");
+  const b = await createTestPaseoDaemon({ listenPort: address.port });
+  cleanup.push(() => b.close());
+  expect((await online(paseo)).hosts[0]).toMatchObject({
+    registrationId: host.registrationId,
+    serverId: b.daemon.getServerId(),
+    status: "online",
+    nextRetryAt: null,
+    error: null,
+  });
+});
+
+test("losing a sent mutation result identifies the destination and never queues unavailable operations", async () => {
+  const b = await daemon();
+  const a = await daemon();
+  const { paseo, client } = await api(a);
+  await paseo.hosts.add({ connection: `127.0.0.1:${b.port}` });
+  const serverId = (await online(paseo)).hosts[0]!.serverId!;
+  const remote = await paseo.hosts.client(serverId);
+  const pending = remote.config.patch({ pluginsEnabled: false });
+  const result = pending.catch((error: unknown) => error);
+  await client.close();
+  expect(await result).toMatchObject({ code: "outcome_unknown", serverId });
+  await expect(remote.workspaces.list()).rejects.toMatchObject({ code: "host_offline", serverId });
+  await expect(remote.terminals.list()).rejects.toMatchObject({ code: "host_offline", serverId });
+  await expect(
+    remote.workspaces.create({
+      source: { kind: "directory", path: "/tmp" },
+      workspaceId: "offline-never-created",
+    }),
+  ).rejects.toMatchObject({ code: "host_offline", serverId });
+  await remote.dispose();
+});
+
+test("disposing an origin during routed handshake settles acquisition and closes only its pending session", async () => {
+  const http = createServer();
+  const ws = new WebSocketServer({ server: http });
+  let connections = 0;
+  let routeHello = false;
+  let routeClosed = false;
+  ws.on("connection", (socket) => {
+    const monitor = ++connections === 1;
+    socket.on("close", () => {
+      if (!monitor) routeClosed = true;
+    });
+    socket.on("message", (data) => {
+      const frame = JSON.parse(data.toString());
+      if (frame.type !== "hello") return;
+      if (monitor)
+        socket.send(
+          JSON.stringify({
+            type: "session",
+            message: {
+              type: "status",
+              payload: {
+                status: "server_info",
+                serverId: "srv_pending",
+                hostname: "pending",
+                version: "0.11.0",
+                features: { connectedHosts: true, ownedSubscriptions: true },
+              },
+            },
+          }),
+        );
+      else routeHello = true;
+    });
+  });
+  http.listen(0, "127.0.0.1");
+  await once(http, "listening");
+  cleanup.push(async () => {
+    for (const socket of ws.clients) socket.terminate();
+    await new Promise<void>((resolve) => ws.close(() => resolve()));
+    await new Promise<void>((resolve) => http.close(() => resolve()));
+  });
+  const address = http.address();
+  if (!address || typeof address === "string") throw new Error("Missing pending endpoint port");
+  const a = await daemon();
+  const { paseo } = await api(a);
+  await paseo.hosts.add({ connection: `127.0.0.1:${address.port}` });
+  await online(paseo);
+  const pending = paseo.hosts.client("srv_pending");
+  const result = pending.catch((error: unknown) => error);
+  await expect.poll(() => routeHello).toBe(true);
+  await paseo.dispose();
+  expect(await result).toBeInstanceOf(Error);
+  await expect.poll(() => routeClosed).toBe(true);
+  expect(ws.clients.size).toBe(1);
+});

packages/server/src/server/connected-hosts/index.ts

@@ -0,0 +1,531 @@
+export { ConnectedHostSession } from "./session.js";
+import { randomUUID, createHash } from "node:crypto";
+import { mkdir, readFile, rename, writeFile, rm } from "node:fs/promises";
+import { join } from "node:path";
+import { z } from "zod";
+import {
+  createHostClient,
+  createHostTransportFactory,
+  parseHostTarget,
+  HostTransportBudget,
+} from "@getpaseo/client/node";
+import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
+import {
+  HostSummarySchema,
+  HostAddSchema,
+  HostUpdateSchema,
+  type HostDirectory,
+  type HostSummary,
+  type HostAdd,
+  type HostUpdate,
+} from "@getpaseo/protocol/hosts";
+import type { DaemonPermission, WSHelloMessage } from "@getpaseo/protocol/messages";
+
+export class HostError extends Error {
+  constructor(
+    readonly code: string,
+    message: string,
+    readonly serverId?: string,
+  ) {
+    super(message);
+  }
+}
+const SavedSchema = HostSummarySchema.extend({ connection: z.string() });
+type Saved = z.infer<typeof SavedSchema>;
+export interface HostActor {
+  sourceId: string;
+  manageSources: boolean;
+}
+interface Live {
+  stop(): void;
+}
+interface Route {
+  registrationId: string;
+  close(code?: string, message?: string): void;
+}
+
+/** Daemon-owned configuration, identity pinning, retry and routed session lifetime. */
+export class ConnectedHosts {
+  private readonly budget = new HostTransportBudget();
+  private rows: Saved[] = [];
+  private readonly live = new Map<string, Live>();
+  private readonly routes = new Set<Route>();
+  private readonly listeners = new Set<() => void>();
+  private queue: Promise<unknown> = Promise.resolve();
+  private readonly ready: Promise<void>;
+  private closed = false;
+  constructor(
+    private readonly home: string,
+    private readonly self: HostDirectory["self"],
+  ) {
+    this.ready = this.load();
+    void this.ready.catch(() => {});
+  }
+  private async load(): Promise<void> {
+    try {
+      this.rows = z
+        .array(SavedSchema)
+        .parse(JSON.parse(await readFile(join(this.home, "hosts.json"), "utf8")));
+    } catch (error) {
+      if ((error as NodeJS.ErrnoException).code !== "ENOENT")
+        throw new Error("Cannot read saved connected hosts", { cause: error });
+    }
+    if (this.closed) return;
+    for (const row of this.rows) {
+      row.status = row.enabled ? "connecting" : "disabled";
+      row.nextRetryAt = null;
+      this.start(row);
+    }
+  }
+  async list(): Promise<HostDirectory> {
+    await this.ready;
+    return {
+      self: { ...this.self },
+      hosts: this.rows.map(({ connection: _secret, ...row }) => structuredClone(row)),
+    };
+  }
+  observe(changed: () => void): () => void {
+    this.listeners.add(changed);
+    return () => this.listeners.delete(changed);
+  }
+  private changed(): void {
+    for (const listener of this.listeners) listener();
+  }
+  private async persist(rows: Saved[]): Promise<void> {
+    await mkdir(this.home, { recursive: true });
+    const temp = join(this.home, `.hosts-${randomUUID()}.tmp`);
+    try {
+      await writeFile(temp, JSON.stringify(rows), { mode: 0o600 });
+      await rename(temp, join(this.home, "hosts.json"));
+    } finally {
+      await rm(temp, { force: true });
+    }
+  }
+  private mutate<T>(run: () => Promise<T>): Promise<T> {
+    const result = this.queue.then(async () => {
+      await this.ready;
+      if (this.closed) throw new HostError("host_offline", "Daemon is stopping");
+      return run();
+    });
+    this.queue = result.catch(() => {});
+    return result;
+  }
+  private summary(row: Saved): HostSummary {
+    const { connection: _secret, ...summary } = row;
+    return structuredClone(summary);
+  }
+  add(input: HostAdd, actor: HostActor): Promise<HostSummary> {
+    return this.mutate(async () => {
+      input = HostAddSchema.parse(input);
+      const target = parseHostTarget(input.connection);
+      if (this.rows.length >= 128)
+        throw new HostError("resource_limit", "Remove a host before adding another (limit 128)");
+      const source = { id: actor.sourceId, key: input.sourceKey ?? randomUUID() };
+      const existing = this.rows.find(
+        (row) => row.source.id === source.id && row.source.key === source.key,
+      );
+      if (existing) {
+        if (
+          existing.connection === input.connection &&
+          existing.label === (input.label ?? null) &&
+          JSON.stringify(Object.entries(existing.metadata).sort()) ===
+            JSON.stringify(Object.entries(input.metadata ?? {}).sort())
+        )
+          return this.summary(existing);
+        throw new HostError(
+          "source_conflict",
+          "This source key already exists. Update its registration explicitly.",
+        );
+      }
+      const row: Saved = {
+        registrationId: `hst_${randomUUID()}`,
+        serverId: null,
+        connection: input.connection,
+        label: input.label ?? null,
+        metadata: input.metadata ?? {},
+        source,
+        enabled: true,
+        status: "connecting",
+        targetDescription: target.description,
+        lastOnlineAt: null,
+        nextRetryAt: null,
+        error: null,
+      };
+      await this.persist([...this.rows, row]);
+      this.rows.push(row);
+      this.start(row);
+      this.changed();
+      return this.summary(row);
+    });
+  }
+  update(id: string, patch: HostUpdate, actor: HostActor): Promise<HostSummary> {
+    return this.mutate(async () => {
+      patch = HostUpdateSchema.parse(patch);
+      const previous = this.require(id);
+      if (
+        (patch.connection !== undefined || patch.metadata !== undefined || patch.replaceIdentity) &&
+        actor.sourceId !== previous.source.id &&
+        !actor.manageSources
+      )
+        throw new HostError(
+          "permission_denied",
+          "Only the source or access manager can replace its connection and metadata",
+        );
+      if (patch.replaceIdentity && patch.connection === undefined)
+        throw new HostError(
+          "invalid_connection",
+          "Identity replacement requires a replacement connection",
+        );
+      const row: Saved = {
+        ...previous,
+        ...(patch.label !== undefined ? { label: patch.label } : {}),
+        ...(patch.metadata !== undefined ? { metadata: patch.metadata } : {}),
+        ...(patch.enabled !== undefined ? { enabled: patch.enabled } : {}),
+      };
+      if (patch.connection !== undefined) {
+        row.connection = patch.connection;
+        row.targetDescription = parseHostTarget(patch.connection).description;
+      }
+      if (patch.replaceIdentity) row.serverId = null;
+      const restart =
+        patch.connection !== undefined || patch.enabled !== undefined || patch.replaceIdentity;
+      if (restart) {
+        row.status = row.enabled ? "connecting" : "disabled";
+        row.error = null;
+        row.nextRetryAt = null;
+      }
+      const rows = this.rows.map((entry) => (entry === previous ? row : entry));
+      await this.persist(rows);
+      if (restart) this.stop(id, row.enabled ? "host_offline" : "host_disabled");
+      Object.assign(previous, row);
+      this.rows = rows.map((entry) => (entry === row ? previous : entry));
+      if (restart) this.start(previous);
+      this.changed();
+      return this.summary(previous);
+    });
+  }
+  remove(id: string, actor: HostActor): Promise<void> {
+    return this.mutate(async () => {
+      const row = this.require(id);
+      if (actor.sourceId !== row.source.id && !actor.manageSources)
+        throw new HostError(
+          "permission_denied",
+          "Only the source or access manager can remove this host. Disable it to stop access.",
+        );
+      const rows = this.rows.filter((entry) => entry !== row);
+      await this.persist(rows);
+      this.stop(id, "host_unknown");
+      this.rows = rows;
+      this.changed();
+    });
+  }
+  private require(id: string): Saved {
+    const row = this.rows.find((entry) => entry.registrationId === id);
+    if (!row) throw new HostError("host_unknown", "Unknown host registration");
+    return row;
+  }
+  private stop(id: string, code = "host_offline"): void {
+    this.live.get(id)?.stop();
+    this.live.delete(id);
+    for (const route of this.routes)
+      if (route.registrationId === id)
+        route.close(code, "Host access was disabled, replaced or removed");
+  }
+  private start(row: Saved): void {
+    if (!row.enabled || this.closed) return;
+    let client: DaemonClient | null = null;
+    let timer: ReturnType<typeof setTimeout> | undefined;
+    let stopped = false;
+    let connecting = false;
+    const live: Live = {
+      stop: () => {
+        stopped = true;
+        clearTimeout(timer);
+        void client?.close();
+      },
+    };
+    this.live.set(row.registrationId, live);
+    const retry = () => {
+      if (stopped || connecting || timer || row.status === "error") return;
+      row.status = "offline";
+      row.error = {
+        code: "host_offline",
+        message: "Cannot reach host. Check the connection; retrying automatically.",
+      };
+      row.nextRetryAt = new Date(Date.now() + 5000).toISOString();
+      this.changed();
+      timer = setTimeout(() => {
+        timer = undefined;
+        void attempt();
+      }, 5000);
+      timer.unref();
+    };
+    const attempt = async () => {
+      if (stopped) return;
+      await client?.close();
+      client = createHostClient(row.connection, {
+        clientId: `host-monitor-${this.self.serverId}-${row.registrationId}`,
+        reconnect: { enabled: false },
+        connectTimeoutMs: 10000,
+      });
+      client.subscribeConnectionStatus((state) => {
+        if (!stopped && state.status === "disconnected") {
+          for (const route of this.routes)
+            if (route.registrationId === row.registrationId) route.close();
+          retry();
+        }
+      });
+      connecting = true;
+      row.status = "connecting";
+      row.nextRetryAt = null;
+      this.changed();
+      try {
+        await client.connect();
+        if (stopped) return;
+        const info = client.getLastServerInfoMessage();
+        if (!info) throw new HostError("host_offline", "No daemon identity received");
+        if (row.serverId && info.serverId !== row.serverId)
+          throw new HostError(
+            "server_id_mismatch",
+            "Daemon identity changed. Verify the destination and explicitly replace its identity.",
+          );
+        if (!row.serverId)
+          await this.mutate(async () => {
+            if (stopped) return;
+            const discovered = { ...row, serverId: info.serverId };
+            await this.persist(this.rows.map((entry) => (entry === row ? discovered : entry)));
+            row.serverId = info.serverId;
+          });
+        if (info.serverId === this.self.serverId)
+          throw new HostError(
+            "self_host",
+            "This connection reaches this daemon. Remove it or replace the connection.",
+          );
+        if (this.rows.find((entry) => entry.serverId === info.serverId) !== row)
+          throw new HostError(
+            "duplicate_host",
+            "Another registration already owns this daemon. Remove the duplicate registration.",
+          );
+        if (info.features?.connectedHosts !== true)
+          throw new HostError(
+            "unsupported_capability",
+            "Update the destination daemon to use connected hosts.",
+          );
+        // Pin and persist under the same writer as management to prevent discovery races.
+        await this.mutate(async () => {
+          if (stopped) return;
+          if (this.rows.find((entry) => entry.serverId === info.serverId) !== row)
+            throw new HostError("duplicate_host", "Another registration already owns this daemon");
+          const next = {
+            ...row,
+            serverId: info.serverId,
+            status: "online" as const,
+            lastOnlineAt: new Date().toISOString(),
+            error: null,
+            nextRetryAt: null,
+          };
+          await this.persist(this.rows.map((entry) => (entry === row ? next : entry)));
+          Object.assign(row, next);
+          this.changed();
+        });
+      } catch (error) {
+        if (stopped) return;
+        const failure = connectionFailure(error, client.authFailureReason !== null);
+        if (failure) {
+          row.status = "error";
+          row.error = failure;
+          row.nextRetryAt = null;
+          this.changed();
+        }
+        await client.close();
+      } finally {
+        connecting = false;
+        if (!stopped && !client.isConnected) retry();
+      }
+    };
+    void attempt();
+  }
+  async openRoute(input: {
+    serverId: string;
+    scopeId: string;
+    sourceId: string;
+    originSessionId: string;
+    permissions: readonly DaemonPermission[];
+    signal: AbortSignal;
+    frame(data: string, binary: boolean): void;
+    ended(code: string, message: string): void;
+  }): Promise<{ send(data: string, binary: boolean): void; close(): void }> {
+    await this.ready;
+    const row = this.rows.find((entry) => entry.serverId === input.serverId);
+    if (!row) throw new HostError("host_unknown", `Unknown host ${input.serverId}`, input.serverId);
+    if (!row.enabled)
+      throw new HostError(
+        "host_disabled",
+        `Enable host ${input.serverId} before using it`,
+        input.serverId,
+      );
+    if (row.status !== "online")
+      throw new HostError(
+        row.error?.code ?? "host_offline",
+        row.error?.message ?? `Host ${input.serverId} is offline`,
+        input.serverId,
+      );
+    if (this.routes.size >= 128) throw new HostError("resource_limit", "Too many routed sessions");
+    const transport = createHostTransportFactory(
+      row.connection,
+      this.budget,
+    )({ url: "ws://host-route/ws" });
+    let closed = false;
+    let verified = false;
+    let helloSent = false;
+    const timer = setTimeout(() => route.close("host_offline", "Host handshake timed out"), 15000);
+    timer.unref();
+    const route: Route = {
+      registrationId: row.registrationId,
+      close: (code = "host_offline", message = `Host ${input.serverId} disconnected`) => {
+        if (closed) return;
+        closed = true;
+        input.signal.removeEventListener("abort", abort);
+        clearTimeout(timer);
+        this.routes.delete(route);
+        try {
+          transport.close();
+        } finally {
+          failed(new HostError(code, message, input.serverId));
+          input.ended(code, message);
+        }
+      },
+    };
+    this.routes.add(route);
+    const abort = () =>
+      route.close("host_offline", "Origin session closed or its authority changed");
+    input.signal.addEventListener("abort", abort, { once: true });
+    let opened!: () => void;
+    let failed!: (error: unknown) => void;
+    const ready = new Promise<void>((resolve, reject) => {
+      opened = resolve;
+      failed = reject;
+    });
+    if (input.signal.aborted) {
+      route.close();
+    }
+    transport.onOpen(opened);
+    transport.onError((error) => {
+      const authenticationFailed =
+        error instanceof Error && "code" in error && error.code === "authentication_failed";
+      const code = authenticationFailed ? "authentication_failed" : "host_offline";
+      const message = authenticationFailed
+        ? "Host authentication failed. Check the SSH configuration and daemon credentials."
+        : "Host transport failed";
+      failed(new HostError(code, message, input.serverId));
+      route.close(code, message);
+    });
+    transport.onClose(() => {
+      failed(new HostError("host_offline", "Host transport closed"));
+      route.close();
+    });
+    transport.onMessage((data, binary) => {
+      if (closed) return;
+      if (!verified) {
+        if (binary) {
+          route.close();
+          return;
+        }
+        try {
+          const envelope = JSON.parse(String(data));
+          if (envelope.type === "hello.rejected") {
+            route.close("authentication_failed", "Replace the host credentials");
+            return;
+          }
+          const info =
+            envelope.type === "session" && envelope.message.type === "status"
+              ? envelope.message.payload
+              : null;
+          if (info?.status !== "server_info") return;
+          if (info.serverId !== row.serverId) {
+            route.close(
+              "server_id_mismatch",
+              "Host identity changed; explicit replacement required",
+            );
+            return;
+          }
+          if (info.features?.connectedHosts !== true) {
+            route.close("unsupported_capability", "Update the destination daemon");
+            return;
+          }
+          verified = true;
+          clearTimeout(timer);
+        } catch {
+          route.close();
+          return;
+        }
+      }
+      input.frame(
+        binary ? Buffer.from(data as Uint8Array).toString("base64") : String(data),
+        binary,
+      );
+    });
+    await ready;
+    if (closed) throw new HostError("host_offline", "Host closed during connection");
+    return {
+      close: () => route.close(),
+      send: (data, binary) => {
+        if (closed) throw new HostError("host_offline", "Host is offline");
+        if (!helloSent) {
+          if (binary) throw new HostError("invalid_message", "Expected hello");
+          const hello = JSON.parse(data) as WSHelloMessage;
+          if (hello.type !== "hello") throw new HostError("invalid_message", "Expected hello");
+          const target = parseHostTarget(row.connection);
+          const clientId = `route-${createHash("sha256")
+            .update(
+              JSON.stringify([
+                this.self.serverId,
+                input.sourceId,
+                input.originSessionId,
+                input.scopeId,
+              ]),
+            )
+            .digest("hex")}`;
+          transport.send(
+            JSON.stringify({
+              ...hello,
+              clientId,
+              routed: true,
+              producer: { serverId: this.self.serverId, sourceId: input.sourceId },
+              sessionPermissions: input.permissions,
+              auth: target.password ? { kind: "password", password: target.password } : undefined,
+            }),
+          );
+          helloSent = true;
+          return;
+        }
+        if (!verified) throw new HostError("host_offline", "Host identity has not been verified");
+        try {
+          transport.send(binary ? Buffer.from(data, "base64") : data);
+        } catch {
+          route.close("resource_limit", "Host buffer limit reached or transport unavailable");
+        }
+      },
+    };
+  }
+  async close(): Promise<void> {
+    this.closed = true;
+    await this.ready;
+    await this.queue;
+    for (const id of this.live.keys()) this.stop(id);
+    for (const route of this.routes) route.close();
+    this.listeners.clear();
+  }
+}
+
+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 === "authentication_failed")
+    return { code: "authentication_failed", message: error.message };
+  if (authenticationFailed)
+    return {
+      code: "authentication_failed",
+      message: "Host rejected the credentials. Replace the connection credentials.",
+    };
+  return null;
+}

packages/server/src/server/connected-hosts/session.ts

@@ -0,0 +1,234 @@
+import type {
+  SessionInboundMessage,
+  SessionOutboundMessage,
+  DaemonPermission,
+} from "@getpaseo/protocol/messages";
+import { ConnectedHosts, HostError, type HostActor } from "./index.js";
+import type { SessionDelivery } from "../session/owned-subscriptions/index.js";
+
+type HostRequest<T extends SessionInboundMessage["type"]> = Extract<
+  SessionInboundMessage,
+  { type: T }
+>;
+interface SessionRoute {
+  scopeId: string;
+  source: object;
+  send(data: string, binary: boolean): void;
+  release(): Promise<void>;
+  fail(code: string, message: string): void;
+}
+
+/** Adapts host capabilities to source-owned session delivery. */
+export class ConnectedHostSession {
+  private readonly routes = new Map<string, SessionRoute>();
+  constructor(
+    private readonly hosts: ConnectedHosts,
+    private readonly delivery: SessionDelivery,
+    private readonly actor: HostActor,
+    private readonly permissions: () => DaemonPermission[],
+    private readonly emit: (message: SessionOutboundMessage) => void,
+    private readonly routed: () => boolean,
+    private readonly originSessionId: string,
+    private readonly bufferedAmount: (source: object) => number | null,
+  ) {}
+  setSource(sourceId: string): void {
+    this.actor.sourceId = sourceId;
+  }
+  dispatch(message: SessionInboundMessage): Promise<void> | undefined {
+    if (!message.type.startsWith("hosts.")) return;
+    return this.handle(message);
+  }
+  revoke(): void {
+    for (const route of this.routes.values())
+      route.fail(
+        "permission_denied",
+        "Caller authority changed. Reconnecting with current grants.",
+      );
+    void this.delivery.releaseFamily("host-route");
+  }
+  private async handle(message: SessionInboundMessage): Promise<void> {
+    try {
+      if (this.routed() && message.type.startsWith("hosts.route."))
+        throw new HostError(
+          "unsupported_capability",
+          "Connected hosts support one hop. Connect to the managing daemon to select a host.",
+        );
+      await this.execute(message);
+    } catch (error) {
+      this.fail(message, error);
+    }
+  }
+  private fail(message: SessionInboundMessage, error: unknown): void {
+    let failure = new HostError(
+      "host_error",
+      "Host operation failed. Check the connection and retry.",
+    );
+    if (error instanceof HostError) failure = error;
+    else if (error instanceof Error && "code" in error && error.code === "invalid_connection")
+      failure = new HostError("invalid_connection", error.message);
+    if (message.type === "hosts.route.frame.request") {
+      const route = this.routes.get(message.routeId);
+      if (route && route.source === this.delivery.currentSource)
+        route.fail(failure.code, failure.message);
+      return;
+    }
+    if (!("requestId" in message) || !message.requestId) return;
+    this.emit({
+      type: "rpc_error",
+      payload: {
+        requestId: message.requestId,
+        requestType: message.type,
+        code: failure.code,
+        error: failure.message,
+      },
+    });
+  }
+  private async execute(message: SessionInboundMessage): Promise<void> {
+    const actor = { ...this.actor, manageSources: this.permissions().includes("access.manage") };
+    switch (message.type) {
+      case "hosts.list.request":
+        return this.list(message);
+      case "hosts.add.request":
+        this.emit({
+          type: "hosts.add.response",
+          payload: {
+            requestId: message.requestId,
+            host: await this.hosts.add(message.input, actor),
+          },
+        });
+        return;
+      case "hosts.update.request":
+        this.emit({
+          type: "hosts.update.response",
+          payload: {
+            requestId: message.requestId,
+            host: await this.hosts.update(message.registrationId, message.patch, actor),
+          },
+        });
+        return;
+      case "hosts.remove.request":
+        await this.hosts.remove(message.registrationId, actor);
+        this.emit({ type: "hosts.remove.response", payload: { requestId: message.requestId } });
+        return;
+      case "hosts.route.open.request":
+        return this.open(message);
+      case "hosts.route.frame.request": {
+        const route = this.routes.get(message.routeId);
+        if (!route || route.source !== this.delivery.currentSource)
+          throw new HostError("host_unknown", "Unknown routed session");
+        route.send(message.data, message.binary);
+        return;
+      }
+      case "hosts.route.close.request": {
+        const route = message.routeId
+          ? this.routes.get(message.routeId)
+          : [...this.routes.values()].find(
+              (entry) =>
+                entry.scopeId === message.scopeId && entry.source === this.delivery.currentSource,
+            );
+        if (route && route.source === this.delivery.currentSource) await route.release();
+        return;
+      }
+    }
+  }
+  private async list(message: HostRequest<"hosts.list.request">): Promise<void> {
+    if (!message.observe) {
+      this.emit({
+        type: "hosts.list.response",
+        payload: { requestId: message.requestId, directory: await this.hosts.list() },
+      });
+      return;
+    }
+    let stop = () => {};
+    const owner = this.delivery.begin("hosts", undefined, () => stop());
+    try {
+      stop = this.hosts.observe(() => {
+        void this.hosts
+          .list()
+          .then((directory) => {
+            owner.emit({ type: "hosts.changed", payload: { subscriptionId: owner.id, directory } });
+            return undefined;
+          })
+          .catch(() => owner.release());
+      });
+      this.emit({
+        type: "hosts.list.response",
+        payload: {
+          requestId: message.requestId,
+          subscriptionId: owner.id,
+          directory: await this.hosts.list(),
+        },
+      });
+    } catch (error) {
+      await owner.release();
+      throw error;
+    }
+  }
+  private async open(message: HostRequest<"hosts.route.open.request">): Promise<void> {
+    if (this.delivery.subscriptionIds(this.delivery.currentSource!, "host-route").length >= 32)
+      throw new HostError("resource_limit", "Too many routed sessions for this caller");
+    let close = () => {};
+    const owner = this.delivery.begin("host-route", undefined, () => {
+      this.routes.delete(owner.id);
+      close();
+    });
+    const fail = (code: string, text: string) => {
+      owner.emit({
+        type: "hosts.route.closed",
+        payload: { subscriptionId: owner.id, code, message: text },
+      });
+      void owner.release();
+    };
+    this.routes.set(owner.id, {
+      scopeId: message.scopeId,
+      source: owner.source,
+      send: () => {
+        throw new HostError("host_offline", "Host is connecting");
+      },
+      release: owner.release,
+      fail,
+    });
+    try {
+      const route = await this.hosts.openRoute({
+        serverId: message.serverId,
+        scopeId: message.scopeId,
+        sourceId: this.actor.sourceId,
+        originSessionId: this.originSessionId,
+        permissions: this.permissions(),
+        signal: owner.signal,
+        frame: (data, binary) => {
+          const size = Buffer.byteLength(data) + 512;
+          const buffered = this.bufferedAmount(owner.source);
+          if (buffered === null || buffered + size > 32 * 1024 * 1024) {
+            fail("resource_limit", "The receiving client cannot keep up with host output");
+            return;
+          }
+          owner.emit({
+            type: "hosts.route.frame",
+            payload: { subscriptionId: owner.id, data, binary },
+          });
+        },
+        ended: fail,
+      });
+      close = route.close;
+      if (owner.signal.aborted) {
+        route.close();
+        return;
+      }
+      this.routes.set(owner.id, {
+        scopeId: message.scopeId,
+        source: owner.source,
+        send: route.send,
+        release: owner.release,
+        fail,
+      });
+      this.emit({
+        type: "hosts.route.open.response",
+        payload: { requestId: message.requestId, routeId: owner.id },
+      });
+    } catch (error) {
+      await owner.release();
+      throw error;
+    }
+  }
+}

packages/server/src/server/session.ts

@@ -1,4 +1,6 @@
 import { AgentDirectory } from "./agent/directory.js";
+import { ConnectedHostSession } from "./connected-hosts/index.js";
+import type { ConnectedHosts } from "./connected-hosts/index.js";
 import { searchTimeline } from "./agent/chat-search/index.js";
 import type { BrowserToolsBroker } from "./browser-tools/broker.js";
 import { BrowserAutomationHostCapabilitySchema } from "@getpaseo/protocol/browser-automation/capabilities";
@@ -434,6 +436,9 @@ const nodeSessionFileSystem: SessionFileSystem = {
 type AgentMcpTransportFactory = () => Promise<unknown>;
 
 export interface SessionOptions {
+  connectedHosts?: ConnectedHosts;
+  sourceId?: string;
+  routed?: boolean;
   browserToolsBroker?: BrowserToolsBroker | null;
   clientId: string;
   permissions: readonly DaemonPermission[];
@@ -444,6 +449,7 @@ export interface SessionOptions {
   onBinaryMessage?: (frame: Uint8Array) => void;
   onBinaryMessageToSource?: (source: object, frame: Uint8Array) => Promise<void>;
   getTransportBufferedAmount?: (source?: object) => number | null;
+  getConnectedHostBufferedAmount?: () => number | null;
   onLifecycleIntent?: (intent: SessionLifecycleIntent) => void;
   onWorkspaceRecovered?: (workspace: PersistedWorkspaceRecord) => Promise<void>;
   publishScriptStatusUpdate?: (message: ScriptStatusUpdateMessage) => void;
@@ -666,6 +672,22 @@ interface ClientActivity {
 }
 
 export class Session {
+  private readonly connectedHostSession: ConnectedHostSession | null;
+  private routed = false;
+  isRouted(): boolean {
+    return this.routed;
+  }
+  getPrincipalId(): string {
+    return this.hostPrincipalId;
+  }
+  private hostPrincipalId: string;
+  setHostSource(sourceId: string): void {
+    this.hostPrincipalId = sourceId;
+    this.connectedHostSession?.setSource(sourceId);
+  }
+  setRouted(routed: boolean): void {
+    this.routed = routed;
+  }
   readonly delivery = new SessionDelivery(
     (source, message) => {
       if (!this.authorization.allowsOutbound(message)) return;
@@ -856,9 +878,12 @@ export class Session {
     this.browserToolsBroker = options.browserToolsBroker;
     this.clientId = clientId;
     this.authorization = new SessionAuthorization(permissions);
+
     this.appVersion = appVersion ?? null;
     this.clientCapabilities = parseClientCapabilities(clientCapabilities);
     this.sessionId = uuidv4();
+    this.hostPrincipalId = hostSource(options);
+    this.connectedHostSession = createHostSession(options, this);
     this.onMessage = onMessage;
     this.onMessageToSource = onMessageToSource ?? null;
     this.onBinaryMessage = onBinaryMessage ?? null;
@@ -2165,7 +2190,7 @@ export class Session {
               requestId,
               requestType: msg.type,
               error: `Session is not authorized for ${msg.type}`,
-              code: "access_denied",
+              code: msg.type.startsWith("hosts.") ? "permission_denied" : "access_denied",
             },
           });
         }
@@ -2212,6 +2237,7 @@ export class Session {
 
   public setPermissions(permissions: readonly DaemonPermission[]): void {
     this.authorization.replacePermissions(permissions);
+    this.connectedHostSession?.revoke();
     if (!this.authorization.allowsPermission("workspace.write")) {
       void this.delivery
         .releaseFamily("browser-host")
@@ -2366,11 +2392,10 @@ export class Session {
 
   private async dispatchInboundMessage(msg: SessionInboundMessage, source?: object): Promise<void> {
     const promise =
+      this.connectedHostSession?.dispatch(msg) ??
       this.dispatchSubscriptionMessage(msg, source) ??
       this.dispatchVoiceAndControlMessage(msg) ??
-      this.dispatchAgentRewindMessage(msg, source) ??
-      this.dispatchAgentRelationshipMessage(msg) ??
-      this.dispatchAgentTimelineMessage(msg, source) ??
+      this.dispatchAgentObservationMessage(msg, source) ??
       this.dispatchHubExecutionMessage(msg) ??
       this.dispatchCreationMessage(msg, source) ??
       this.dispatchAgentLifecycleMessage(msg) ??
@@ -2388,6 +2413,17 @@ export class Session {
     if (promise) await promise;
   }
 
+  private dispatchAgentObservationMessage(
+    msg: SessionInboundMessage,
+    source?: object,
+  ): Promise<void> | undefined {
+    return (
+      this.dispatchAgentRewindMessage(msg, source) ??
+      this.dispatchAgentRelationshipMessage(msg) ??
+      this.dispatchAgentTimelineMessage(msg, source)
+    );
+  }
+
   private dispatchWorkspaceLifecycleMessage(msg: SessionInboundMessage): Promise<void> | undefined {
     return (
       this.dispatchWorkspaceStateMessage(msg) ??
@@ -8691,3 +8727,21 @@ function legacyWantsEvent(
       return true;
   }
 }
+
+function createHostSession(options: SessionOptions, session: Session): ConnectedHostSession | null {
+  if (!options.connectedHosts) return null;
+  return new ConnectedHostSession(
+    options.connectedHosts,
+    session.delivery,
+    { sourceId: options.sourceId ?? "owner", manageSources: false },
+    () => session.getPermissions(),
+    (message) => session.publish(message),
+    () => session.isRouted(),
+    session.getSessionId(),
+    () => options.getConnectedHostBufferedAmount?.() ?? 0,
+  );
+}
+
+function hostSource(options: SessionOptions): string {
+  return options.sourceId ?? "owner";
+}

packages/server/src/server/websocket-server.ts

@@ -1,3 +1,4 @@
+import { ConnectedHosts } from "./connected-hosts/index.js";
 import { stat } from "node:fs/promises";
 import type { CreationSnapshot } from "@getpaseo/protocol/messages";
 import { CreationService } from "./creation/index.js";
@@ -437,6 +438,7 @@ export interface WebSocketLike {
 }
 
 interface SessionConnectionBase {
+  permissionCeiling?: readonly DaemonPermission[];
   session: Session;
   principalId: string;
   sessionKey: string;
@@ -460,6 +462,7 @@ interface PluginSessionConnection extends SessionConnectionBase {
 type SessionConnection = ReconnectableSessionConnection | PluginSessionConnection;
 
 interface SocketSessionOptions {
+  sourceId?: string;
   clientId: string;
   appVersion: string | null;
   clientCapabilities: Record<string, unknown> | null;
@@ -550,6 +553,7 @@ export class VoiceAssistantWebSocketServer {
   private readonly workspaceAutoName: WorkspaceAutoName;
   private readonly downloadTokenStore: DownloadTokenStore;
   private readonly paseoHome: string;
+  private readonly connectedHosts: ConnectedHosts;
   private readonly passwordHash: string | undefined;
   private readonly credentialSource: DaemonAuthConfig | undefined;
   private readonly worktreesRoot: string | undefined;
@@ -701,6 +705,7 @@ export class VoiceAssistantWebSocketServer {
     this.workspaceAutoName = workspaceAutoName;
     this.downloadTokenStore = downloadTokenStore;
     this.paseoHome = paseoHome;
+    this.connectedHosts = new ConnectedHosts(paseoHome, { serverId, label: getHostName() });
     this.passwordHash = auth?.password;
     this.worktreesRoot = daemonRuntimeConfig?.worktreesRoot;
     this.daemonConfigStore = daemonConfigStore;
@@ -1054,7 +1059,11 @@ export class VoiceAssistantWebSocketServer {
     }
     for (const connection of new Set(this.externalSessionsByKey.values())) {
       if (connection.principalId === principalId) {
-        connection.session.setPermissions(permissions);
+        connection.session.setPermissions(
+          connection.permissionCeiling
+            ? permissions.filter((permission) => connection.permissionCeiling!.includes(permission))
+            : permissions,
+        );
       }
     }
   }
@@ -1071,6 +1080,7 @@ export class VoiceAssistantWebSocketServer {
 
   public async close(): Promise<void> {
     this.prepareForShutdown();
+    await this.connectedHosts.close();
     this.unsubscribeSpeechReadiness?.();
     this.unsubscribeSpeechReadiness = null;
     this.unsubscribeDaemonConfigChange?.();
@@ -1394,6 +1404,10 @@ export class VoiceAssistantWebSocketServer {
       appVersion,
       clientCapabilities,
       permissions: admission.permissions,
+      sourceId:
+        lifecycle.kind === "ephemeral-plugin"
+          ? `plugin:${lifecycle.pluginId}`
+          : admission.principalId,
       connectionLogger,
       onMessage: (msg) => {
         if (!connection) {
@@ -1463,6 +1477,16 @@ export class VoiceAssistantWebSocketServer {
 
   private createSocketSession(options: SocketSessionOptions): Session {
     return new Session({
+      connectedHosts: this.connectedHosts,
+      getConnectedHostBufferedAmount: () => {
+        let total = 0;
+        for (const socket of this.sessions.keys()) {
+          if (typeof socket.bufferedAmount !== "number") return null;
+          total += socket.bufferedAmount;
+        }
+        return total;
+      },
+      sourceId: options.sourceId,
       browserToolsBroker: this.browserToolsBroker,
       clientId: options.clientId,
       appVersion: options.appVersion,
@@ -1616,8 +1640,16 @@ export class VoiceAssistantWebSocketServer {
     }
 
     this.clearPendingConnection(ws);
-    const admitted = pending.admission;
+    let admitted = pending.admission;
     if (!admitted) throw new Error("Admitted hello has no session admission");
+    if (message.sessionPermissions)
+      admitted = {
+        ...admitted,
+        permissions: admitted.permissions.filter((permission) =>
+          message.sessionPermissions!.includes(permission),
+        ),
+      };
+    pending.admission = admitted;
     pending.identity.clientId = clientId;
     if (message.appVersion) {
       pending.identity.appVersion = message.appVersion;
@@ -1640,6 +1672,9 @@ export class VoiceAssistantWebSocketServer {
       lifecycle: pluginId ? { kind: "ephemeral-plugin", pluginId } : { kind: "reconnectable" },
       admission: admitted,
     });
+    connection.permissionCeiling = message.sessionPermissions;
+    connection.session.setRouted(message.routed === true);
+    this.assignHostSource(connection, message, admitted.principalId, pluginId);
     this.sessions.set(ws, connection);
     if (connection.lifecycle === "reconnectable") {
       this.externalSessionsByKey.set(sessionKey, connection);
@@ -1755,6 +1790,10 @@ export class VoiceAssistantWebSocketServer {
       clearTimeout(existing.externalDisconnectCleanupTimeout);
       existing.externalDisconnectCleanupTimeout = null;
     }
+    existing.permissionCeiling = message.sessionPermissions;
+    existing.session.setPermissions(pending.admission!.permissions);
+    existing.session.setRouted(message.routed === true);
+    this.assignHostSource(existing, message, pending.admission!.principalId);
     const newAppVersion = message.appVersion ?? null;
     if (newAppVersion && newAppVersion !== existing.appVersion) {
       existing.appVersion = newAppVersion;
@@ -1785,6 +1824,21 @@ export class VoiceAssistantWebSocketServer {
     );
   }
 
+  private assignHostSource(
+    connection: SessionConnection,
+    hello: WSHelloMessage,
+    principalId: string,
+    pluginId?: string,
+  ): void {
+    if (pluginId) return;
+    const producer = hello.producer;
+    const sourceId =
+      hello.routed && producer
+        ? `route:${[principalId, producer.serverId, producer.sourceId].map(encodeURIComponent).join(":")}`
+        : principalId;
+    connection.session.setHostSource(sourceId);
+  }
+
   private buildServerInfoStatusPayload(session: Session): ServerInfoStatusPayload {
     return {
       status: "server_info",
@@ -1792,11 +1846,13 @@ export class VoiceAssistantWebSocketServer {
       serverId: this.serverId,
       hostname: getHostName(),
       version: this.daemonVersion,
+      principalId: session.getPrincipalId(),
       permissions: session.getPermissions(),
       // COMPAT(desktopManaged): added in v0.1.X, remove optional parsing after 2027-01-16.
       desktopManaged: this.daemonRuntimeConfig?.desktopManaged === true,
       ...(this.serverCapabilities ? { capabilities: this.serverCapabilities } : {}),
       features: {
+        connectedHosts: true,
         usageSources: true,
         ownedSubscriptions: true,
         agentRequestReceipts: true,
@@ -2088,6 +2144,10 @@ export class VoiceAssistantWebSocketServer {
         this.finishPluginSocketCleanup(ws);
         return;
       }
+      if (connection.session.isRouted()) {
+        await this.cleanupConnection(connection, "Routed session disconnected");
+        return;
+      }
       this.incrementRuntimeCounter("sessionDisconnectedWaitingReconnect");
       if (connection.externalDisconnectCleanupTimeout) {
         clearTimeout(connection.externalDisconnectCleanupTimeout);