test(e2e): keep transient developer toasts from stalling checks

Concurrent locator handlers can wait for a dismiss button another handler removed. Dismiss existing toast DOM at document lifetime, including its open shadow root, without intercepting app actions.

CLAUDE.md

@@ -46,6 +46,7 @@ At the start of non-trivial work, list `docs/` and skim anything relevant to the
 | [docs/rpc-namespacing.md](docs/rpc-namespacing.md)                   | WebSocket RPC naming convention — dotted namespaces and `.request`/`.response` pairs                                           |
 | [docs/protocol-compatibility.md](docs/protocol-compatibility.md)     | Why app/daemon versions drift, protocol vs feature contract, capability gating, COMPAT tagging                                 |
 | [docs/protocol-validation.md](docs/protocol-validation.md)           | zod-aot generated inbound WebSocket validation, patched compiler regressions, schema-purity rules                              |
+| [docs/connected-hosts.md](docs/connected-hosts.md)                   | Daemon-owned host registrations, routed SDK scopes, identity and authority boundaries                                          |
 | [docs/permissions.md](docs/permissions.md)                           | Semantic daemon permissions, principals, credentials, pairing invitations, and Hub authority                                   |
 | [docs/terminal-performance.md](docs/terminal-performance.md)         | Terminal latency pipeline, coalescing/backpressure invariants, benchmark + perf spec usage                                     |
 | [docs/agent-stream-performance.md](docs/agent-stream-performance.md) | Assistant text pipeline — coalescing window, paced reveal, why arrival lumps are smoothed at render                            |

docs/architecture.md

@@ -4,6 +4,9 @@ Paseo is a client-server system for monitoring and controlling local AI coding a
 
 Your code never leaves your machine. Paseo is local-first.
 
+Daemon-owned outbound connections and routed API scope ownership are described in
+[Connected hosts](connected-hosts.md).
+
 ## System overview
 
 ```

docs/connected-hosts.md

@@ -0,0 +1,69 @@
+# Connected hosts
+
+A saved host belongs to the daemon that stores it. The app's saved connections remain a separate list.
+Connect A to B through `paseo --host A host` or `paseo.hosts`. Consumers select B by its learned `serverId`; management uses the stable
+`registrationId` created before B is reachable.
+
+## Ownership and authority
+
+Credentials live in A's `hosts.json`, written atomically with mode `0600` on POSIX systems. Discovery returns a safe
+target description. Replacement inputs never come back through a read API. Keep this file private
+when collecting diagnostics or backing up daemon state.
+
+The authenticated producer owns its source key. Repeating an unchanged add returns the saved entry,
+including a user's disabled state. Changed input requires an explicit update. Display labels and
+access can be changed by users with `tunnel.manage`; connection and metadata changes require the
+producer or `access.manage`. Removing a producer-managed entry allows that producer to add it again.
+Disabling retains the opt-out.
+
+Each enabled registration owns one physical daemon link, shared by status monitoring and every
+routed API scope. Each scope owns an isolated virtual session on B. A derives its identity from the
+admitted origin session, producer, and logical SDK scope. B admits virtual sessions under the actual
+physical link principal, then intersects forwarded caller grants with that link's current authority.
+Changing or resuming a parent's grants closes its existing channels; reconnect re-admits demand with
+current authority. A routed API cannot select a second hop, and saving a host grants no reverse access.
+
+A producer that registers a host through a routed API is qualified on B as
+`route:<B principal>:<A serverId>:<A producer>`, with each component percent-encoded. This namespace belongs to the authenticated link
+principal. An ordinary direct B consumer does not become that producer by supplying a source key.
+The optional `server_info.sourceId` reports this producer identity separately from the authenticated admission principal.
+Trusted server plugins retain the authority of their runtime; they are not sandboxed by this feature.
+
+## Connections and recovery
+
+The shared Node connector owns the target grammar and each connection attempt, including a fresh
+SSH process. SSH reaches an already running daemon. The shared client connection lifecycle owns
+physical reconnect and keepalive; the registry retains configuration, identity acceptance, and recovery
+status. Unreachable registrations remain saved. Identity is pinned when discovered. Self and duplicate
+registrations remain inspectable; a changed destination requires explicit identity replacement.
+
+Both daemons advertise `connectedHosts`; the shared link additionally requires `hostMultiplexing`.
+A completes capability and pinned-identity checks on each physical recovery before admitting channels. There is no local fallback. An older daemon continues to serve existing local
+operations; requesting this feature gives update guidance.
+
+Disposing a routed API releases its observations and session. It does not remove the saved host or
+affect sibling API scopes. Disposing the origin cascades to its derived scopes. Observations restore
+through the usual owned-subscription contract. The ordinary client queues new requests while connecting
+within its existing deadline and rejects requests while disconnected. Connection loss clears old
+waiters and queued sends. Submitted requests with no response fail with `DaemonConnectionError`
+(`DAEMON_CONNECTION_LOST`), optional `outcome: "unknown"` and known `serverId`. Inspect the destination
+before retrying a mutation.
+Ordinary non-recoverable mutations are not automatically replayed. Existing acknowledged/idempotent
+creation recovery retains its request identity and resumes through the ordinary creation lifecycle. Multiplexing changes only the transport: ordinary permission
+RPC codes pass through unchanged, and connection/recovery behavior matches direct clients.
+
+Text session envelopes carry a channel identifier. Binary routing adds an opcode, a bounded UTF-8
+channel identifier, and the original terminal or file-transfer bytes. It never converts those bytes to
+base64 JSON. The existing direct binary opcodes and 100 MiB message limit remain unchanged; routed
+payloads reserve space for the routing header inside that limit. Relay encryption retains its own
+existing frame limit.
+
+Host links cap saved registrations, simultaneous channels, handshake lifetime, frames, physical
+output, and aggregate queued bytes. Capacity checks use the actual escaped outer JSON or binary
+frame size. Awaitable output completes when the physical send completes. An overflowing channel
+closes without closing siblings or the saved link. Routed sessions release resources on disconnect;
+subscription demand is rebuilt by its owning client on reconnect.
+
+See the [SDK reference](../public-docs/sdk/reference.md#clienthosts),
+[CLI reference](../public-docs/cli.md#connected-hosts), and
+[plugin reference](../public-docs/plugins/reference.md#connected-hosts).

docs/permissions.md

@@ -31,6 +31,10 @@ A pairing invitation is neither. It is an expiring, single-use exchange that cre
 
 Agents and terminals use workspace authority. Both can execute code and mutate the workspace, so separate write permissions would claim an isolation boundary the daemon cannot enforce.
 
+Connected-host management uses `tunnel.manage`; source-owned connection and metadata replacement
+requires the producer or `access.manage`. Routed operations retain both caller and link authority,
+see [Connected hosts](connected-hosts.md#ownership-and-authority).
+
 Owner, operator, and viewer are UI presets expanded into explicit permissions. Do not persist them as roles. Adding a permission must not silently widen an existing principal.
 
 Permissions are additive allows. Missing authority denies the operation. Do not add deny precedence.

package-lock.json

@@ -38178,6 +38178,7 @@
       "dependencies": {
         "@getpaseo/protocol": "0.12.0",
         "@getpaseo/relay": "0.12.0",
+        "ws": "^8.20.0",
         "zod": "^4.4.3"
       },
       "devDependencies": {

packages/app/e2e/browser/developer-ui.spec.ts

@@ -0,0 +1,58 @@
+import { metroTest as test, expect } from "../support/fixtures";
+import { installDeveloperToastDismissal } from "../support/developer-ui.mjs";
+import type { Page } from "@playwright/test";
+
+async function showTransientDeveloperToast(page: Page, shadow: boolean) {
+  await page.evaluate((useShadow) => {
+    const toast = document.createElement("div");
+    toast.id = "error-toast";
+    document.body.append(toast);
+    const root = useShadow ? toast.attachShadow({ mode: "open" }) : toast;
+    const button = document.createElement("button");
+    button.style.cssText = "width:32px;height:32px";
+    button.addEventListener("click", () => button.remove());
+    root.append(button);
+  }, shadow);
+}
+
+async function openTransientDeveloperToast(page: Page) {
+  await page.setContent("<button>Continue</button>");
+  await showTransientDeveloperToast(page, false);
+  await installDeveloperToastDismissal(page);
+}
+
+async function navigateWithDeveloperToastDismissal(page: Page) {
+  await installDeveloperToastDismissal(page);
+  await page.goto("data:text/html,<button>Continue</button>");
+  await showTransientDeveloperToast(page, true);
+}
+
+async function inspectLayoutWhileToastDisappears(page: Page) {
+  const action = page.getByRole("button", { name: "Continue", exact: true });
+  const boxes = await Promise.all(Array.from({ length: 4 }, () => action.boundingBox()));
+  for (const box of boxes) expect(box).not.toBeNull();
+  await expect(page.locator("#error-toast").getByRole("button")).toHaveCount(0);
+  await action.click();
+}
+
+test("a disappearing developer toast does not stall concurrent layout checks", async ({ page }) => {
+  test.setTimeout(10000);
+  await openTransientDeveloperToast(page);
+  await inspectLayoutWhileToastDisappears(page);
+});
+
+test("developer toast dismissal survives navigation and observes Expo shadow-root updates", async ({
+  page,
+}) => {
+  test.setTimeout(10000);
+  await navigateWithDeveloperToastDismissal(page);
+  await inspectLayoutWhileToastDisappears(page);
+  await page.evaluate(() => {
+    const root = document.getElementById("error-toast")!.shadowRoot!;
+    const replacement = document.createElement("button");
+    replacement.style.cssText = "width:32px;height:32px";
+    replacement.addEventListener("click", () => replacement.remove());
+    root.append(replacement);
+  });
+  await inspectLayoutWhileToastDisappears(page);
+});

packages/app/e2e/support/developer-ui.mjs

@@ -1,15 +1,27 @@
-const HIDE_DEVELOPER_TOASTS = "#error-toast { display: none !important; }";
-
 /** @param {import('@playwright/test').Page} page */
-export async function hideDeveloperToasts(page) {
-  // Developer chrome must not cover app actions or move the test's pointer/focus.
-  // Console errors remain in the runner's logs. Hiding the toast, instead of clicking it away,
-  // leaves no race with a toast that closes or is replaced between matching and clicking.
-  await page.addInitScript((css) => {
-    const style = document.createElement("style");
-    style.textContent = css;
-    document.addEventListener("DOMContentLoaded", () => document.head.append(style));
-  }, HIDE_DEVELOPER_TOASTS);
-  // A page attached after it loaded (Electron over CDP) runs no init script until it navigates.
-  await page.addStyleTag({ content: HIDE_DEVELOPER_TOASTS });
+export async function installDeveloperToastDismissal(page) {
+  // Keep developer chrome out of app actions without pointer/focus changes.
+  // Locator handlers can deadlock when concurrent checks dismiss the same toast.
+  const install = () => {
+    let observedRoot;
+    const options = { childList: true, subtree: true };
+    const dismiss = () => {
+      const toast = document.getElementById("error-toast");
+      const root = toast?.shadowRoot ?? toast;
+      if (root && root !== observedRoot) {
+        observer.disconnect();
+        observer.observe(document, options);
+        observer.observe(root, options);
+        observedRoot = root;
+      }
+      // Expo's toast root contains dismiss buttons; its overlay has a different ID.
+      for (const button of root?.querySelectorAll("button") ?? [])
+        button.dispatchEvent(new MouseEvent("click", { bubbles: true }));
+    };
+    const observer = new MutationObserver(dismiss);
+    observer.observe(document, options);
+    dismiss();
+  };
+  await page.addInitScript(install);
+  await page.evaluate(install);
 }

packages/app/e2e/support/fixtures.ts

@@ -1,5 +1,5 @@
 import { test as base, expect, type Page } from "@playwright/test";
-import { hideDeveloperToasts } from "./developer-ui.mjs";
+import { installDeveloperToastDismissal } from "./developer-ui.mjs";
 import { startOutdatedDaemon, type OutdatedDaemon } from "./helpers/daemon-update";
 import { startE2EWorker } from "./helpers/e2e-worker";
 import { getE2EDaemonPort } from "./helpers/daemon-port";
@@ -131,7 +131,7 @@ const test = daemonTest.extend<{
         await ws.close({ code: 1008, reason: "Blocked connection to localhost:6767 during e2e." });
       });
 
-      await hideDeveloperToasts(page);
+      await installDeveloperToastDismissal(page);
 
       const entries: string[] = [];
 

packages/cli/src/cli.ts

@@ -1,3 +1,4 @@
+import { createHostCommand } from "./commands/host/index.js";
 import { pairCommand } from "./commands/daemon/pair.js";
 import { Command, Option } from "commander";
 import { createAgentCommand } from "./commands/agent/index.js";
@@ -58,6 +59,8 @@ export function createCli(): Command {
     .option("--no-color", "disable colored output");
   addDaemonHostOption(program);
 
+  program.addCommand(createHostCommand());
+
   // Primary agent commands (top-level)
   addJsonAndDaemonHostOptions(addLsOptions(program.command("ls"))).action(withOutput(runLsCommand));
 

packages/cli/src/commands/host/index.ts

@@ -0,0 +1,116 @@
+import { Command } from "commander";
+import type { HostSummary, HostUpdate } from "@getpaseo/client";
+import {
+  withOutput,
+  type CommandOptions,
+  type OutputSchema,
+  type CommandResult,
+} from "../../output/index.js";
+import { addJsonAndDaemonHostOptions, collectMultiple } from "../../utils/command-options.js";
+import { connectToDaemon } from "../../utils/client.js";
+
+const schema: OutputSchema<HostSummary> = {
+  idField: "registrationId",
+  columns: [
+    { header: "ID", field: "registrationId" },
+    { header: "LABEL", field: "label" },
+    { header: "SERVER", field: "serverId" },
+    { header: "ENABLED", field: "enabled" },
+    { header: "STATUS", field: "status" },
+    { header: "SOURCE", field: (row) => `${row.source.id}/${row.source.key}` },
+  ],
+};
+function metadata(values: string[] = []): Record<string, string> {
+  return Object.fromEntries(
+    values.map((entry) => {
+      const split = entry.indexOf("=");
+      if (split < 1) throw { code: "invalid_metadata", message: "Metadata must use key=value" };
+      return [entry.slice(0, split), entry.slice(split + 1)];
+    }),
+  );
+}
+interface Options extends CommandOptions {
+  label?: string;
+  meta?: string[];
+  removeMeta?: string[];
+  connection?: string;
+  replaceIdentity?: boolean;
+}
+async function run(
+  action: string,
+  id: string | undefined,
+  options: Options,
+): Promise<CommandResult<HostSummary>> {
+  const client = await connectToDaemon({ target: options.daemonTarget });
+  try {
+    if (action === "add")
+      return {
+        type: "single",
+        data: await client.addHost({
+          connection: id!,
+          label: options.label,
+          metadata: metadata(options.meta),
+        }),
+        schema,
+      };
+    if (action === "ls") return { type: "list", data: (await client.listHosts()).hosts, schema };
+    const host = (await client.listHosts()).hosts.find((entry) => entry.registrationId === id);
+    if (!host) throw { code: "host_unknown", message: `Unknown registration ${id}` };
+    if (action === "inspect")
+      return {
+        type: "single",
+        data: host,
+        schema: { ...schema, renderHuman: () => JSON.stringify(host, null, 2) },
+      };
+    if (action === "rm") {
+      await client.removeHost(id!);
+      return { type: "single", data: host, schema };
+    }
+    const patch: HostUpdate = {};
+    if (action === "enable" || action === "disable") patch.enabled = action === "enable";
+    if (options.label !== undefined) patch.label = options.label || null;
+    if (options.connection !== undefined) patch.connection = options.connection;
+    if (options.replaceIdentity) patch.replaceIdentity = true;
+    if (options.meta?.length || options.removeMeta?.length) {
+      patch.metadata = { ...host.metadata, ...metadata(options.meta) };
+      for (const key of options.removeMeta ?? []) delete patch.metadata[key];
+    }
+    return { type: "single", data: await client.updateHost(id!, patch), schema };
+  } finally {
+    await client.close();
+  }
+}
+export function createHostCommand(): Command {
+  const host = new Command("host").description("Manage hosts connected to the selected daemon");
+  addJsonAndDaemonHostOptions(
+    host.command("ls").description("List saved hosts, including disabled and failed entries"),
+  ).action(withOutput((options: Options, _command: Command) => run("ls", undefined, options)));
+  addJsonAndDaemonHostOptions(
+    host
+      .command("add")
+      .argument("<connection>")
+      .option("--label <name>")
+      .option("--meta <key=value>", "Set metadata", collectMultiple, []),
+  ).action(
+    withOutput((connection: string, options: Options, _command: Command) =>
+      run("add", connection, options),
+    ),
+  );
+  addJsonAndDaemonHostOptions(
+    host
+      .command("update")
+      .argument("<registration-id>")
+      .option("--label <name>")
+      .option("--connection <connection>")
+      .option("--replace-identity", "Accept a new destination identity")
+      .option("--meta <key=value>", "Replace supplied metadata keys", collectMultiple, [])
+      .option("--remove-meta <key>", "Remove a metadata key", collectMultiple, []),
+  ).action(
+    withOutput((id: string, options: Options, _command: Command) => run("update", id, options)),
+  );
+  for (const action of ["inspect", "enable", "disable", "rm"])
+    addJsonAndDaemonHostOptions(host.command(action).argument("<registration-id>")).action(
+      withOutput((id: string, options: Options, _command: Command) => run(action, id, options)),
+    );
+  return host;
+}

packages/cli/src/ssh/ssh-tunnel.ts

@@ -1,90 +1,2 @@
-import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
-import { createServer, type Server, type Socket } from "node:net";
-import { buildSshTunnelArgs, type SshTransportTarget } from "@getpaseo/protocol/ssh-transport";
-
-const SSH_STDERR_LIMIT = 8192;
-
-export interface SshTunnel {
-  endpoint: string;
-  close(): void;
-  failureDetail(): string | null;
-}
-
-function formatSshFailure(
-  stderr: string,
-  code: number | null,
-  signal: NodeJS.Signals | null,
-): string {
-  const detail = stderr.trim();
-  if (detail) return detail;
-  if (signal) return `ssh exited with signal ${signal}`;
-  return `ssh exited with code ${code ?? "unknown"}`;
-}
-
-export function resolveSshFailureDetail(failure: string | null, stderr: string): string | null {
-  return failure ?? (stderr.trim() || null);
-}
-
-export function createSshTunnel(target: SshTransportTarget): Promise<SshTunnel> {
-  let server: Server | null = null;
-  let socket: Socket | null = null;
-  let child: ChildProcessWithoutNullStreams | null = null;
-  let stderr = "";
-  let failure: string | null = null;
-
-  function close(): void {
-    server?.close();
-    server = null;
-    socket?.destroy();
-    socket = null;
-    if (child && !child.killed) child.kill();
-    child = null;
-  }
-
-  return new Promise((resolve, reject) => {
-    server = createServer((acceptedSocket) => {
-      socket = acceptedSocket;
-      server?.close();
-      server = null;
-
-      child = spawn("ssh", buildSshTunnelArgs(target), {
-        stdio: ["pipe", "pipe", "pipe"],
-        windowsHide: true,
-      });
-      child.stderr.on("data", (chunk: Buffer | string) => {
-        stderr = `${stderr}${chunk.toString()}`.slice(-SSH_STDERR_LIMIT);
-      });
-      child.on("error", (error) => {
-        failure = error.message;
-        acceptedSocket.destroy(error);
-      });
-      child.on("exit", (code, signal) => {
-        if (code !== 0 || signal) failure = formatSshFailure(stderr, code, signal);
-        acceptedSocket.destroy(failure ? new Error(failure) : undefined);
-      });
-      acceptedSocket.on("error", () => undefined);
-      acceptedSocket.on("close", () => {
-        if (child && !child.killed) child.kill();
-      });
-      acceptedSocket.pipe(child.stdin);
-      child.stdout.pipe(acceptedSocket);
-    });
-    server.once("error", (error) => {
-      close();
-      reject(error);
-    });
-    server.listen(0, "127.0.0.1", () => {
-      const address = server?.address();
-      if (!address || typeof address === "string") {
-        close();
-        reject(new Error("Failed to allocate the SSH tunnel port"));
-        return;
-      }
-      resolve({
-        endpoint: `127.0.0.1:${address.port}`,
-        close,
-        failureDetail: () => resolveSshFailureDetail(failure, stderr),
-      });
-    });
-  });
-}
+export { createSshTunnel, resolveSshFailureDetail } from "@getpaseo/client/node";
+export type { SshTunnel } from "@getpaseo/client/node";

packages/cli/src/utils/client.ssh.test.ts

@@ -1,42 +1,8 @@
-import { beforeEach, describe, expect, it, vi } from "vitest";
+import { describe, expect, it } from "vitest";
 import { resolveSshFailureDetail } from "../ssh/ssh-tunnel.js";
-import { connectToDaemon } from "./client.js";
-
-const mocks = vi.hoisted(() => ({
-  configs: [] as Array<Record<string, unknown>>,
-  createSshTunnel: vi.fn(),
-}));
-
-vi.mock("@getpaseo/client/internal/daemon-client", () => ({
-  DaemonClient: class {
-    lastError = null;
-
-    constructor(config: Record<string, unknown>) {
-      mocks.configs.push(config);
-    }
-
-    async connect() {}
-    async close() {}
-  },
-}));
-
-vi.mock("../ssh/ssh-tunnel.js", async (importOriginal) => ({
-  ...(await importOriginal<typeof import("../ssh/ssh-tunnel.js")>()),
-  createSshTunnel: mocks.createSshTunnel,
-}));
-vi.mock("./client-id.js", () => ({ getOrCreateCliClientId: async () => "cli-test-id" }));
+import { parseHostTarget } from "@getpaseo/client/node";
 
 describe("CLI SSH transport", () => {
-  beforeEach(() => {
-    mocks.configs.length = 0;
-    mocks.createSshTunnel.mockReset();
-    mocks.createSshTunnel.mockResolvedValue({
-      endpoint: "127.0.0.1:4567",
-      close: vi.fn(),
-      failureDetail: () => null,
-    });
-  });
-
   it("surfaces SSH stderr before the child exit event settles", () => {
     expect(resolveSshFailureDetail(null, "Host key verification failed.\n")).toBe(
       "Host key verification failed.",
@@ -45,21 +11,14 @@ describe("CLI SSH transport", () => {
       "ssh exited with code 255",
     );
   });
-
-  it("routes an SSH host through a local tunnel", async () => {
-    await connectToDaemon({
-      target: { kind: "endpoint", host: "ssh://deploy@build-box:2222?daemonPort=7777" },
-    });
-
-    expect(mocks.createSshTunnel).toHaveBeenCalledWith({
-      host: "deploy@build-box",
-      sshPort: 2222,
-      daemonPort: 7777,
-    });
-    expect(mocks.configs[0]).toMatchObject({
-      url: "ws://127.0.0.1:4567/ws",
-      clientId: "cli-test-id",
-      clientType: "cli",
+  it("uses the shared SSH target grammar and excludes reusable credentials from discovery", () => {
+    expect(parseHostTarget("ssh://deploy@build-box:2222?daemonPort=7777")).toEqual({
+      kind: "ssh",
+      ssh: { host: "deploy@build-box", sshPort: 2222, daemonPort: 7777 },
+      description: "ssh://build-box:2222 (daemon port 7777)",
     });
+    expect(() => parseHostTarget("ssh://deploy:secret@build-box")).toThrow(
+      "Invalid host connection",
+    );
   });
 });

packages/cli/src/utils/client.ts

@@ -1,3 +1,4 @@
+import { createHostClient, parseHostTarget } from "@getpaseo/client/node";
 import {
   waitForDaemonReady,
   resolvePaseoHome,
@@ -8,21 +9,12 @@ import { describeDaemonTarget, type DaemonTarget } from "./daemon-target.js";
 export type { DaemonTarget } from "./daemon-target.js";
 import {
   buildDaemonWebSocketUrl,
-  buildRelayWebSocketUrl,
   normalizeHostPort,
   parseConnectionUri,
-  shouldUseTlsForDefaultHostedRelay,
 } from "@getpaseo/protocol/daemon-endpoints";
-import {
-  parseConnectionOfferFromUrl,
-  type ConnectionOffer,
-} from "@getpaseo/protocol/connection-offer";
-import { parseSshTransportUri } from "@getpaseo/protocol/ssh-transport";
-import { DaemonClient, type WebSocketLike } from "@getpaseo/client/internal/daemon-client";
-import { WebSocket } from "ws";
+import { DaemonClient } from "@getpaseo/client/internal/daemon-client";
 import { getOrCreateCliClientId } from "./client-id.js";
 import { resolveCliVersion } from "../version.js";
-import { createSshTunnel } from "../ssh/ssh-tunnel.js";
 
 export interface ConnectOptions {
   target: DaemonTarget;
@@ -50,15 +42,15 @@ export function buildDaemonConnectionCommandError(options: ConnectOptions & { er
   if (error && typeof error === "object" && "message" in error) message = String(error.message);
   if (options.target.kind === "endpoint")
     message = message.replaceAll(options.target.host, describeDaemonTarget(options.target));
-  if (message.startsWith("Cannot connect to daemon at "))
-    return error as { code: string; message: string; details: string };
   let code = "DAEMON_UNREACHABLE";
   if (typeof error === "object" && error !== null && "code" in error) code = String(error.code);
   else if (message === "Password required") code = "AUTH_REQUIRED";
   else if (message === "Incorrect password") code = "AUTH_FAILED";
   return {
     code,
-    message: `Cannot connect to daemon at ${describeDaemonTarget(options.target)}: ${message}`,
+    message: message.startsWith("Cannot connect to daemon at ")
+      ? message
+      : `Cannot connect to daemon at ${describeDaemonTarget(options.target)}: ${message}`,
     details: describeConnectionRemedy(code, options.target),
   };
 }
@@ -170,11 +162,8 @@ export function resolveDaemonTarget(host: string): TransportTarget {
 }
 
 export function resolveDaemonPassword(host: string): string | undefined {
-  const trimmed = host.trim();
-  if (trimmed.startsWith("tcp://")) {
-    const fromUri = parseConnectionUri(trimmed).password;
-    if (fromUri) return fromUri;
-  }
+  const fromUri = parseHostTarget(host).password;
+  if (fromUri) return fromUri;
   const fromEnv = process.env.PASEO_PASSWORD;
   return fromEnv && fromEnv.length > 0 ? fromEnv : undefined;
 }
@@ -189,117 +178,9 @@ export function resolveDaemonCredential(
   return token ? { kind: "localCredential", token } : null;
 }
 
-/**
- * Create a WebSocket factory that works in Node.js
- */
-function createNodeWebSocketFactory() {
-  return (
-    url: string,
-    options?: { headers?: Record<string, string>; protocols?: string[]; socketPath?: string },
-  ): WebSocketLike => {
-    return new WebSocket(url, options?.protocols, {
-      headers: options?.headers,
-      ...(options?.socketPath ? { socketPath: options.socketPath } : {}),
-    }) as unknown as WebSocketLike;
-  };
-}
-
-/**
- * Create and connect a daemon client
- * Returns the connected client or throws if connection fails
- */
-async function tryConnectHost(
-  host: string,
-  credential: ReturnType<typeof resolveDaemonCredential>,
-  home: string,
-  clientId: string,
-  timeout: number,
-  nodeWebSocketFactory: ReturnType<typeof createNodeWebSocketFactory>,
-): Promise<{ client: DaemonClient } | { error: unknown }> {
-  const target = resolveDaemonTarget(host);
-  const client = new DaemonClient({
-    url: target.url,
-    clientId,
-    clientType: "cli",
-    appVersion: resolveCliVersion(),
-    ...(credential?.kind === "password" ? { password: credential.password } : {}),
-    ...(credential?.kind === "localCredential"
-      ? { localCredential: () => readLocalCredentialForTarget(home, host) ?? undefined }
-      : {}),
-    connectTimeoutMs: timeout,
-    webSocketFactory: (
-      url: string,
-      config?: { headers?: Record<string, string>; protocols?: string[] },
-    ) =>
-      nodeWebSocketFactory(url, {
-        headers: config?.headers,
-        protocols: config?.protocols,
-        ...(target.type === "ipc" ? { socketPath: target.socketPath } : {}),
-      }),
-    reconnect: { enabled: false },
-  });
-
-  try {
-    await client.connect();
-    return { client };
-  } catch (error) {
-    await client.close().catch(() => {});
-    return { error };
-  }
-}
-
-async function connectViaRelayOffer(
-  offer: ConnectionOffer,
-  clientId: string,
-  timeout: number,
-  nodeWebSocketFactory: ReturnType<typeof createNodeWebSocketFactory>,
-): Promise<DaemonClient> {
-  const url = buildRelayWebSocketUrl({
-    endpoint: offer.relay.endpoint,
-    serverId: offer.serverId,
-    role: "client",
-    useTls: offer.relay.useTls ?? shouldUseTlsForDefaultHostedRelay(offer.relay.endpoint),
-  });
-
-  const client = new DaemonClient({
-    url,
-    clientId,
-    clientType: "cli",
-    appVersion: resolveCliVersion(),
-    connectTimeoutMs: timeout,
-    webSocketFactory: (
-      target: string,
-      config?: { headers?: Record<string, string>; protocols?: string[] },
-    ) => nodeWebSocketFactory(target, { headers: config?.headers, protocols: config?.protocols }),
-    e2ee: { enabled: true, daemonPublicKeyB64: offer.daemonPublicKeyB64 },
-    reconnect: { enabled: false },
-  });
-
-  try {
-    await client.connect();
-    return client;
-  } catch (error) {
-    await client.close().catch(() => {});
-    const message = error instanceof Error ? error.message : String(error);
-    const lastError = client.lastError ? ` (${client.lastError})` : "";
-    throw new Error(`Failed to connect via relay offer: ${message}${lastError}`, { cause: error });
-  }
-}
-
-function parseHostOfferOrNull(host: string | undefined): ConnectionOffer | null {
-  if (!host) return null;
-  try {
-    return parseConnectionOfferFromUrl(host);
-  } catch (error) {
-    const message = error instanceof Error ? error.message : String(error);
-    throw new Error(`Invalid pairing offer URL: ${message}`, { cause: error });
-  }
-}
-
 async function connectSelectedDaemon(options: ConnectOptions): Promise<DaemonClient> {
   const timeout = options.timeout ?? DEFAULT_TIMEOUT;
-  const deadline = Date.now() + timeout;
-  const explicitHost =
+  const host =
     options.target.kind === "endpoint"
       ? options.target.host
       : (
@@ -309,58 +190,25 @@ async function connectSelectedDaemon(options: ConnectOptions): Promise<DaemonCli
           })
         ).listen;
   const home = resolveClientPaseoHome(options.target);
-  const clientId = await getOrCreateCliClientId(home);
-  const nodeWebSocketFactory = createNodeWebSocketFactory();
-
-  if (explicitHost?.trim().startsWith("ssh://")) {
-    const target = parseSshTransportUri(explicitHost.trim());
-    const tunnel = await createSshTunnel(target);
-    const password = resolveDaemonPassword(explicitHost);
-    const result = await tryConnectHost(
-      tunnel.endpoint,
-      password ? { kind: "password", password } : null,
-      home,
-      clientId,
-      Math.max(1, deadline - Date.now()),
-      nodeWebSocketFactory,
-    );
-    if ("client" in result) {
-      const close = result.client.close.bind(result.client);
-      result.client.close = async () => {
-        try {
-          await close();
-        } finally {
-          tunnel.close();
-        }
-      };
-      return result.client;
-    }
-
-    const failure = tunnel.failureDetail();
-    tunnel.close();
-    if (failure) throw new Error(`SSH connection failed: ${failure}`, { cause: result.error });
-    throw result.error;
+  const credential = resolveDaemonCredential(host, home);
+  const client = createHostClient(host, {
+    clientId: await getOrCreateCliClientId(home),
+    clientType: "cli",
+    appVersion: resolveCliVersion(),
+    ...(credential?.kind === "password" ? { password: credential.password } : {}),
+    ...(credential?.kind === "localCredential"
+      ? { localCredential: () => readLocalCredentialForTarget(home, host) ?? undefined }
+      : {}),
+    connectTimeoutMs: timeout,
+    reconnect: { enabled: false },
+  });
+  try {
+    await client.connect();
+    return client;
+  } catch (error) {
+    await client.close();
+    throw error;
   }
-  const offer = parseHostOfferOrNull(explicitHost);
-  if (offer) {
-    return connectViaRelayOffer(
-      offer,
-      clientId,
-      Math.max(1, deadline - Date.now()),
-      nodeWebSocketFactory,
-    );
-  }
-
-  const result = await tryConnectHost(
-    explicitHost,
-    resolveDaemonCredential(explicitHost, home),
-    home,
-    clientId,
-    Math.max(1, deadline - Date.now()),
-    nodeWebSocketFactory,
-  );
-  if ("client" in result) return result.client;
-  throw result.error;
 }
 
 export async function connectToDaemon(options: ConnectOptions): Promise<DaemonClient> {

packages/cli/src/utils/daemon-target.test.ts

@@ -1,6 +1,11 @@
+import { parseHostTarget } from "@getpaseo/protocol/host-target";
 import { test, expect } from "vitest";
 import { selectDaemonTarget, describeDaemonTarget } from "./daemon-target.js";
-import { resolveDaemonCredential, resolveClientPaseoHome } from "./client.js";
+import {
+  resolveDaemonCredential,
+  resolveClientPaseoHome,
+  buildDaemonConnectionCommandError,
+} from "./client.js";
 import { mkdtemp, rm, writeFile } from "node:fs/promises";
 import { join } from "node:path";
 import { tmpdir } from "node:os";
@@ -44,6 +49,31 @@ test("endpoint descriptions redact pairing material and credentials", () => {
   ).not.toContain("private");
 });
 
+test("a pairing offer without a URL prefix still redacts its key", () => {
+  const offer = {
+    v: 2,
+    serverId: "srv_example",
+    daemonPublicKeyB64: "example-secret",
+    relay: { endpoint: "localhost:6767", useTls: false },
+  };
+  const host = `#offer=${Buffer.from(JSON.stringify(offer)).toString("base64url")}`;
+  expect(describeDaemonTarget({ kind: "endpoint", host })).toBe(
+    "relay://localhost:6767/srv_example",
+  );
+});
+
+test("malformed endpoint diagnostics never expose the supplied credential", () => {
+  const target = { kind: "endpoint" as const, host: "alice:example-secret@localhost:6767" };
+  expect(describeDaemonTarget(target)).not.toContain("example-secret");
+  for (const message of [
+    `Invalid connection ${target.host}`,
+    `Cannot connect to daemon at ${target.host}`,
+  ]) {
+    const diagnostic = buildDaemonConnectionCommandError({ target, error: new Error(message) });
+    expect(JSON.stringify(diagnostic)).not.toContain("example-secret");
+  }
+});
+
 test("CLI selects an explicit password before a matching local credential and never sends the latter remotely", async () => {
   const home = await mkdtemp(join(tmpdir(), "paseo-cli-credential-"));
   const previousPassword = process.env.PASEO_PASSWORD;
@@ -67,3 +97,34 @@ test("CLI selects an explicit password before a matching local credential and ne
     await rm(home, { recursive: true, force: true });
   }
 });
+
+test.each(["/tmp/paseo.sock", "\\\\.\\pipe\\paseo", "localhost:6767", "[::1]:6767", "6767"])(
+  "validated bare target %s keeps its spelling in command hints",
+  (host) => {
+    expect(describeDaemonTarget({ kind: "endpoint", host })).toBe(host);
+  },
+);
+
+test("validated URI command hints preserve their transport address without credentials", () => {
+  const ssh = describeDaemonTarget({
+    kind: "endpoint",
+    host: "ssh://deploy@build-box:2222?daemonPort=7777&password=example-secret",
+  });
+  expect(ssh).not.toContain("example-secret");
+  expect(parseHostTarget(ssh)).toMatchObject({
+    kind: "ssh",
+    ssh: { host: "deploy@build-box", sshPort: 2222, daemonPort: 7777 },
+  });
+  const ws = describeDaemonTarget({
+    kind: "endpoint",
+    host: "wss://example.test/custom-ws?authorization=example-secret",
+  });
+  expect(ws).not.toContain("example-secret");
+  expect(new URL(ws).pathname).toBe("/custom-ws");
+  expect(
+    describeDaemonTarget({
+      kind: "endpoint",
+      host: "tcp://localhost:6767?daemonPort=example-secret&ssl=example-secret",
+    }),
+  ).not.toContain("example-secret");
+});

packages/cli/src/utils/daemon-target.ts

@@ -1,3 +1,4 @@
+import { parseHostTarget } from "@getpaseo/protocol/host-target";
 import { resolvePaseoHome } from "@getpaseo/server/daemon-control";
 
 export type DaemonTarget = { kind: "instance"; home: string } | { kind: "endpoint"; host: string };
@@ -39,13 +40,26 @@ export function selectDaemonTarget(
 export function describeDaemonTarget(target: DaemonTarget): string {
   if (target.kind === "instance") return `home ${target.home}`;
   try {
-    const url = new URL(target.host);
-    if (url.password) url.password = "REDACTED";
-    for (const key of url.searchParams.keys())
-      if (/password|token|secret/i.test(key)) url.searchParams.set(key, "REDACTED");
-    if (url.hash) url.hash = "REDACTED";
-    return url.toString();
+    const input = target.host.trim();
+    const parsed = parseHostTarget(input);
+    // Discovery descriptions are display text; command hints must preserve the address syntax.
+    if (parsed.kind === "ssh" || (!parsed.offer && !parsed.socketPath && input.includes("://"))) {
+      const url = new URL(input);
+      if (decodeURIComponent(url.username).includes(":")) url.username = "REDACTED";
+      for (const key of new Set(url.searchParams.keys())) {
+        const routingOption =
+          (parsed.kind === "ssh" && key === "daemonPort") ||
+          (key === "ssl" &&
+            url.searchParams.getAll(key).every((value) => value === "true" || value === "false"));
+        if (!routingOption) url.searchParams.set(key, "REDACTED");
+      }
+      return url.toString();
+    }
+    // Pairing offers must always use safe display text, including prefix-free fragments.
+    return parsed.kind === "socket" && !parsed.offer && !input.includes("://")
+      ? input
+      : parsed.description;
   } catch {
-    return target.host.replace(/([?&](?:password|token|secret)=)[^&]*/gi, "$1REDACTED");
+    return "<invalid endpoint>";
   }
 }

packages/client/package.json

@@ -24,6 +24,10 @@
     "./internal/daemon-client-websocket-transport": {
       "types": "./dist/daemon-client-websocket-transport.d.ts",
       "default": "./dist/daemon-client-websocket-transport.js"
+    },
+    "./node": {
+      "types": "./dist/node/index.d.ts",
+      "default": "./dist/node/index.js"
     }
   },
   "publishConfig": {
@@ -41,6 +45,7 @@
   "dependencies": {
     "@getpaseo/protocol": "0.12.0",
     "@getpaseo/relay": "0.12.0",
+    "ws": "^8.20.0",
     "zod": "^4.4.3"
   },
   "devDependencies": {

packages/client/src/daemon-client-relay-e2ee-transport.ts

@@ -24,10 +24,16 @@ export function createRelayE2eeTransportFactory(args: {
   baseFactory: DaemonTransportFactory;
   daemonPublicKeyB64: string;
   logger: TransportLogger;
+  assertOutboundCapacity?: (wireBytes: number) => void;
 }): DaemonTransportFactory {
   return ({ url }) => {
     const base = args.baseFactory({ url });
-    return createEncryptedTransport(base, args.daemonPublicKeyB64, args.logger);
+    return createEncryptedTransport(
+      base,
+      args.daemonPublicKeyB64,
+      args.logger,
+      args.assertOutboundCapacity,
+    );
   };
 }
 
@@ -42,6 +48,7 @@ export function createEncryptedTransport(
   base: DaemonTransport,
   daemonPublicKeyB64: string,
   logger: TransportLogger,
+  assertOutboundCapacity: (wireBytes: number) => void = () => {},
 ): DaemonTransport {
   let channel: EncryptedChannel | null = null;
   let opened = false;
@@ -148,6 +155,9 @@ export function createEncryptedTransport(
       if (wireBytes > MAX_RELAY_WEBSOCKET_MESSAGE_BYTES) {
         throw messageTooLargeError(wireBytes, MAX_RELAY_WEBSOCKET_MESSAGE_BYTES);
       }
+      // Encryption and its transport handoff are synchronous up to send's first await.
+      // Reject capacity failures here so a routed sender loses only its own channel.
+      assertOutboundCapacity(wireBytes);
       void channel.send(payload).catch((error) => {
         emitError(error);
       });

packages/client/src/daemon-client.test.ts

@@ -3,6 +3,7 @@ import { afterEach, expect, expectTypeOf, test, vi } from "vitest";
 import { z } from "zod";
 import {
   DaemonClient,
+  DaemonConnectionError,
   supportsUsageReports,
   type DaemonClientTrace,
   type CreateAgentRequestOptions,
@@ -873,9 +874,12 @@ test("keeps relay upgrade credentials out of the socket request", async () => {
     },
   });
   clients.push(client);
-  void client.connect();
+  const connecting = client.connect();
+  void connecting.catch(() => {});
   await vi.waitFor(() => expect(requests).toHaveLength(1));
   expect(requests[0]).toEqual({ url: "ws://relay.test/ws?role=client&serverId=srv_test&v=2" });
+  await client.close();
+  await expect(connecting).rejects.toThrow("Daemon client closed");
 });
 
 test("refuses relay password auth without an encrypted hello", async () => {
@@ -7844,3 +7848,130 @@ test("measures an outbound message in UTF-8 bytes against the daemon's message l
   expect(mock.sent).toHaveLength(1);
   expect(client.getConnectionState().status).toBe("connected");
 });
+
+test("ordinary lifecycle queues new calls, distinguishes submitted outcomes, and never resends lost requests", async () => {
+  const transports = [createMockTransport(), createMockTransport(), createMockTransport()];
+  let attempt = 0;
+  const client = new DaemonClient({
+    url: "ws://transport",
+    clientId: "ordinary-lifecycle",
+    reconnect: { enabled: false },
+    transportFactory: () => transports[attempt++]!.transport,
+  });
+  clients.push(client);
+  const first = transports[0]!;
+  const connecting = client.connect();
+  const initial = client.getDaemonConfig("initial").catch((error: unknown) => error);
+  expect(first.sent).toEqual([]);
+  first.triggerOpen();
+  await connecting;
+  expect(parseSentFrame(first.sent[0]).requestId).toBe("initial");
+  first.triggerMessage(
+    wrapSessionMessage({
+      type: "rpc_error",
+      payload: { requestId: "initial", error: "Denied", code: "access_denied" },
+    }),
+  );
+  expect(await initial).toMatchObject({ code: "access_denied", serverId: "srv_test_1" });
+  const submitted = client
+    .patchDaemonConfig({ pluginsEnabled: false }, "submitted")
+    .catch((error: unknown) => error);
+  first.triggerClose({ code: 1001, reason: "physical loss" });
+  const unknown = await submitted;
+  expect(unknown).toBeInstanceOf(DaemonConnectionError);
+  expect(unknown).toMatchObject({
+    code: "DAEMON_CONNECTION_LOST",
+    outcome: "unknown",
+    serverId: "srv_test_1",
+  });
+
+  expect(() => client.ensureConnected()).not.toThrow();
+  const second = transports[1]!;
+  const neverSent = client
+    .patchDaemonConfig({ pluginsEnabled: true }, "never-sent")
+    .catch((error: unknown) => error);
+  expect(second.sent).toEqual([]);
+  second.triggerClose({ code: 1001, reason: "handshake lost" });
+  const unavailable = await neverSent;
+  expect(unavailable).toBeInstanceOf(DaemonConnectionError);
+  expect(unavailable).toMatchObject({ code: "DAEMON_CONNECTION_LOST", serverId: "srv_test_1" });
+  expect((unavailable as DaemonConnectionError).outcome).toBeUndefined();
+
+  client.ensureConnected();
+  const third = transports[2]!;
+  const recovery = client.getDaemonConfig("recovery").catch((error: unknown) => error);
+  expect(third.sent).toEqual([]);
+  third.triggerOpen();
+  expect(third.sent.map(parseSentFrame).map((frame) => frame.requestId)).toEqual(["recovery"]);
+  third.triggerMessage(
+    wrapSessionMessage({
+      type: "rpc_error",
+      payload: { requestId: "recovery", error: "Denied", code: "access_denied" },
+    }),
+  );
+  expect(await recovery).toMatchObject({ code: "access_denied" });
+});
+
+test.each([false, true])(
+  "synchronous send rejection leaves no unknown outcome after close (close during handoff: %s)",
+  async (closeDuringSend) => {
+    const first = createMockTransport();
+    const second = createMockTransport();
+    const rejected = Object.assign(new Error("Transport rejected this send"), {
+      code: "resource_limit",
+    });
+    const originalSend = first.transport.send;
+    first.transport.send = (data) => {
+      if (typeof data === "string" && data.includes("set_daemon_config_request")) {
+        if (closeDuringSend) first.triggerClose({ code: 1001, reason: "closed during send" });
+        throw rejected;
+      }
+      originalSend(data);
+    };
+    let attempt = 0;
+    const client = new DaemonClient({
+      url: "ws://transport",
+      clientId: "rejected-send",
+      reconnect: { enabled: false },
+      transportFactory: () => [first, second][attempt++]!.transport,
+    });
+    clients.push(client);
+    const connected = client.connect();
+    first.triggerOpen();
+    await connected;
+    await expect(client.patchDaemonConfig({ pluginsEnabled: false }, "rejected")).rejects.toBe(
+      rejected,
+    );
+    first.triggerClose({ code: 1001, reason: "lost connection" });
+    client.ensureConnected();
+    const queued = client.getDaemonConfig("not-submitted").catch((error: unknown) => error);
+    second.triggerClose({ code: 1001, reason: "lost handshake" });
+    const failure = (await queued) as DaemonConnectionError;
+    expect(failure).toBeInstanceOf(DaemonConnectionError);
+    expect(failure.outcome).toBeUndefined();
+    expect((rejected as Error & { outcome?: string }).outcome).toBeUndefined();
+  },
+);
+
+test("initial acquisition preserves structured transport failures and ordinary offline errors retain their cause", async () => {
+  const transport = createMockTransport();
+  const failure = Object.assign(new Error("Destination identity changed"), {
+    code: "server_id_mismatch",
+    serverId: "srv_expected",
+  });
+  const client = new DaemonClient({
+    url: "ws://transport",
+    clientId: "structured-acquisition",
+    reconnect: { enabled: false },
+    transportFactory: () => transport.transport,
+  });
+  clients.push(client);
+  const connecting = client.connect();
+  const rejected = expect(connecting).rejects.toBe(failure);
+  transport.triggerError(failure);
+  await rejected;
+  await expect(client.getDaemonConfig()).rejects.toMatchObject({
+    code: "DAEMON_CONNECTION_LOST",
+    cause: failure,
+  });
+});

packages/client/src/daemon-client.ts

@@ -1,4 +1,11 @@
+import {
+  HOST_ROUTE_OPCODE,
+  encodeHostRouteFrame,
+  decodeHostRouteFrame,
+  type HostRouteFrame,
+} from "@getpaseo/protocol/binary-frames/index";
 import { messageTooLargeError } from "./daemon-client-transport-utils.js";
+import type { HostAdd, HostUpdate, HostDirectory } from "@getpaseo/protocol/hosts";
 import { legacyUsageIcon } from "./legacy-usage-icons.js";
 import { subscribeTimeline, type TimelineMessage } from "./timeline-subscription/index.js";
 import { ProviderSnapshotUpdates } from "./provider-snapshots/index.js";
@@ -333,7 +340,7 @@ export type ConnectionState =
   | { status: "idle" }
   | { status: "connecting"; attempt: number }
   | { status: "connected" }
-  | { status: "disconnected"; reason?: string }
+  | { status: "disconnected"; reason?: string; nextRetryAt?: string; failureCode?: string }
   | { status: "disposed" };
 
 export type DaemonEvent =
@@ -391,6 +398,7 @@ export interface DaemonClientConfig {
   /** Deliver compact bodies/hash references to a caller-owned snapshot cache.
    * The default keeps public SDK snapshot entries expanded. */
   providerSnapshots?: "wire";
+  sessionPermissions?: import("@getpaseo/protocol/messages").DaemonPermission[];
   url: string;
   clientId: string;
   clientType?: "mobile" | "browser" | "cli" | "mcp" | "hub";
@@ -938,6 +946,7 @@ interface Waiter<T> {
   reject(error: Error): void;
   timeoutHandle: ReturnType<typeof setTimeout> | null;
   requestId?: string;
+  submitted: boolean;
 }
 
 interface WaitHandle<T> {
@@ -995,11 +1004,16 @@ type CorrelatedResponsePayload<TType extends CorrelatedResponseType> =
   CorrelatedResponsePayloads[TType];
 
 export class DaemonConnectionError extends Error {
+  readonly serverId?: string;
+  readonly outcome?: "unknown";
   constructor(
     message: string,
     readonly code: "DAEMON_CONNECTION_LOST" | "DAEMON_REQUEST_TIMEOUT" = "DAEMON_CONNECTION_LOST",
+    context?: { serverId?: string; outcome?: "unknown"; cause?: Error },
   ) {
-    super(message);
+    super(message, context?.cause ? { cause: context.cause } : undefined);
+    this.serverId = context?.serverId;
+    this.outcome = context?.outcome;
     this.name = "DaemonConnectionError";
   }
 }
@@ -1008,8 +1022,16 @@ class DaemonRpcError extends Error {
   readonly requestId: string;
   readonly requestType?: string;
   readonly code?: string;
+  readonly serverId?: string;
+  readonly userMessage: string;
 
-  constructor(params: { requestId: string; error: string; requestType?: string; code?: string }) {
+  constructor(params: {
+    requestId: string;
+    error: string;
+    requestType?: string;
+    code?: string;
+    serverId?: string;
+  }) {
     const parts = [params.error];
     if (params.requestType) parts.push(`requestType=${params.requestType}`);
     if (params.code) parts.push(`code=${params.code}`);
@@ -1018,6 +1040,8 @@ class DaemonRpcError extends Error {
     this.requestId = params.requestId;
     this.requestType = params.requestType;
     this.code = params.code;
+    this.serverId = params.serverId;
+    this.userMessage = params.error;
   }
 }
 
@@ -1282,6 +1306,8 @@ export class DaemonClient {
   private connectResolve: (() => void) | null = null;
   private connectReject: ((error: Error) => void) | null = null;
   private lastErrorValue: string | null = null;
+  private destinationServerId: string | undefined;
+  private connectionFailure: Error | undefined;
   private authFailureReasonValue: DaemonAuthFailureReason | null = null;
   private connectionState: ConnectionState = { status: "idle" };
   private readonly terminalStreams = new TerminalStreamRouter();
@@ -1479,6 +1505,7 @@ export class DaemonClient {
           }
           this.scheduleReconnect({
             reason,
+            failure: event instanceof Error ? event : undefined,
             event: "TRANSPORT_CLOSE",
             reasonCode: "transport_closed",
           });
@@ -1502,6 +1529,7 @@ export class DaemonClient {
                   this.lastErrorValue = reason;
                   this.scheduleReconnect({
                     reason,
+                    failure: event instanceof Error ? event : undefined,
                     event: "TRANSPORT_ERROR",
                     reasonCode: "transport_error",
                   });
@@ -1518,6 +1546,7 @@ export class DaemonClient {
           this.lastErrorValue = reason;
           this.scheduleReconnect({
             reason,
+            failure: event instanceof Error ? event : undefined,
             event: "TRANSPORT_ERROR",
             reasonCode: "transport_error",
           });
@@ -1568,9 +1597,7 @@ export class DaemonClient {
       return;
     }
     this.shouldReconnect = false;
-    this.connectPromise = null;
-    this.connectResolve = null;
-    this.connectReject = null;
+    this.rejectConnect(new Error("Daemon client closed"));
     if (this.reconnectTimeout) {
       clearTimeout(this.reconnectTimeout);
       this.reconnectTimeout = null;
@@ -1613,7 +1640,9 @@ export class DaemonClient {
       this.attemptConnect();
       return;
     }
-    void this.connect();
+    void this.connect().catch((error: unknown) =>
+      this.logger.debug({ err: error }, "Connection attempt ended"),
+    );
   }
 
   private verifyConnection(): void {
@@ -1638,6 +1667,119 @@ export class DaemonClient {
       });
   }
 
+  requireConnectedHosts(): void {
+    if (!this.isConnected)
+      throw Object.assign(new Error("The managing daemon is offline"), {
+        code: "host_offline",
+        serverId: this.destinationServerId,
+      });
+    if (this.lastServerInfoMessage?.features?.connectedHosts !== true)
+      throw Object.assign(new Error("Update the managing daemon to use connected hosts"), {
+        code: "unsupported_capability",
+      });
+  }
+  async listHosts(): Promise<HostDirectory> {
+    this.requireConnectedHosts();
+    return (
+      await this.sendCorrelatedSessionRequest({
+        message: { type: "hosts.list.request" },
+        responseType: "hosts.list.response",
+      })
+    ).directory;
+  }
+  observeHosts(): OwnedSubscription<HostDirectory> {
+    this.requireConnectedHosts();
+    return this.owned.observe(async (accept) => {
+      const result = await this.sendCorrelatedSessionRequest({
+        message: { type: "hosts.list.request", observe: true },
+        responseType: "hosts.list.response",
+      });
+      if (!result.subscriptionId) throw new Error("Host omitted subscription identity");
+      const snapshot = { ...result.directory, subscriptionId: result.subscriptionId };
+      accept(snapshot);
+      return snapshot;
+    });
+  }
+  async addHost(input: HostAdd) {
+    this.requireConnectedHosts();
+    return (
+      await this.sendCorrelatedSessionRequest({
+        message: { type: "hosts.add.request", input },
+        responseType: "hosts.add.response",
+      })
+    ).host;
+  }
+  async updateHost(registrationId: string, patch: HostUpdate) {
+    this.requireConnectedHosts();
+    return (
+      await this.sendCorrelatedSessionRequest({
+        message: { type: "hosts.update.request", registrationId, patch },
+        responseType: "hosts.update.response",
+      })
+    ).host;
+  }
+  async removeHost(registrationId: string): Promise<void> {
+    this.requireConnectedHosts();
+    await this.sendCorrelatedSessionRequest({
+      message: { type: "hosts.remove.request", registrationId },
+      responseType: "hosts.remove.response",
+    });
+  }
+  async openHostRoute(serverId: string, scopeId: string): Promise<string> {
+    this.requireConnectedHosts();
+    try {
+      return (
+        await this.sendCorrelatedSessionRequest({
+          message: { type: "hosts.route.open.request", serverId, scopeId },
+          responseType: "hosts.route.open.response",
+        })
+      ).routeId;
+    } catch (error) {
+      throw Object.assign(error instanceof Error ? error : new Error("Host route failed"), {
+        serverId,
+      });
+    }
+  }
+  async openHostLink(channelId: string): Promise<string> {
+    this.requireConnectedHosts();
+    if (this.getLastServerInfoMessage()?.features?.hostMultiplexing !== true)
+      throw Object.assign(new Error("Update the destination daemon"), {
+        code: "unsupported_capability",
+      });
+    return (
+      await this.sendCorrelatedSessionRequest({
+        message: { type: "hosts.link.open.request", channelId },
+        responseType: "hosts.link.open.response",
+      })
+    ).routeId;
+  }
+  sendHostFrame(routeId: string, data: string | Uint8Array): void {
+    if (typeof data === "string") {
+      this.sendSessionMessageStrict({
+        type: "hosts.route.frame.request",
+        routeId,
+        data,
+        binary: false,
+      });
+    } else {
+      if (!this.isConnected || !this.transport) throw this.unavailableError();
+      this.sendTransportFrame(encodeHostRouteFrame(routeId, data));
+    }
+  }
+  private readonly hostFrameListeners = new Set<(frame: HostRouteFrame) => void>();
+  subscribeHostFrames(handler: (frame: HostRouteFrame) => void): () => void {
+    this.hostFrameListeners.add(handler);
+    return () => this.hostFrameListeners.delete(handler);
+  }
+  cancelHostRoute(scopeId: string): void {
+    if (this.isConnected && this.transport)
+      this.sendSessionMessageStrict({ type: "hosts.route.close.request", scopeId });
+  }
+  closeHostRoute(routeId: string): void {
+    if (this.isConnected && this.transport)
+      this.sendSessionMessageStrict({ type: "hosts.route.close.request", routeId });
+  }
+
   getConnectionState(): ConnectionState {
     return this.connectionState;
   }
@@ -1770,17 +1912,34 @@ export class DaemonClient {
     this.endTraceSection(isOpen);
   }
 
-  private sendJsonMessage(envelopeType: string, messageType: string, message: unknown): void {
+  private sendJsonMessage(
+    envelopeType: string,
+    messageType: string,
+    message: unknown,
+    submitted?: () => void,
+  ): void {
     this.traceInstant("paseo.ws.message.outbound", {
       envelopeType,
       messageType,
     });
-    this.sendTransportFrame(JSON.stringify(message));
+    this.sendTransportFrame(JSON.stringify(message), submitted);
   }
 
-  private sendTransportFrame(frame: string | Uint8Array | ArrayBuffer): void {
+  private sendSessionPayload(payload: SessionInboundMessage): void {
+    this.sendJsonMessage("session", payload.type, { type: "session", message: payload }, () => {
+      if (!("requestId" in payload) || typeof payload.requestId !== "string") return;
+      for (const waiter of this.waiters) {
+        if (waiter.requestId === payload.requestId) waiter.submitted = true;
+      }
+    });
+  }
+
+  private sendTransportFrame(
+    frame: string | Uint8Array | ArrayBuffer,
+    submitted?: () => void,
+  ): void {
     if (!this.transport) {
-      throw new DaemonConnectionError("Transport not connected");
+      throw this.unavailableError();
     }
     // The daemon closes the connection on an oversized message, and creations resubmit on reconnect.
     assertTransportFrameFits(frame);
@@ -1789,6 +1948,7 @@ export class DaemonClient {
       size: String(getTransportFrameSize(frame)),
     });
     try {
+      submitted?.();
       this.transport.send(frame);
     } finally {
       this.endTraceSection(isOpen);
@@ -1809,7 +1969,7 @@ export class DaemonClient {
     }
     const payload = SessionInboundMessageSchema.parse(message);
     try {
-      this.sendJsonMessage("session", payload.type, { type: "session", message: payload });
+      this.sendSessionPayload(payload);
     } catch (error) {
       if (this.config.suppressSendErrors) {
         return;
@@ -1848,7 +2008,7 @@ export class DaemonClient {
   private sendSessionMessageOrThrow(message: SessionInboundMessage): Promise<void> {
     return this.sendWhenConnected(() => {
       const payload = SessionInboundMessageSchema.parse(message);
-      this.sendJsonMessage("session", payload.type, { type: "session", message: payload });
+      this.sendSessionPayload(payload);
     });
   }
 
@@ -1888,7 +2048,7 @@ export class DaemonClient {
     }
 
     // Not connected and not connecting - fail immediately
-    return Promise.reject(new DaemonConnectionError(`Transport not connected (status: ${status})`));
+    return Promise.reject(this.unavailableError());
   }
 
   /**
@@ -1942,6 +2102,7 @@ export class DaemonClient {
             kind: "error",
             error: new DaemonRpcError({
               requestId: msg.payload.requestId,
+              serverId: this.destinationServerId,
               error: msg.payload.error,
               requestType: msg.payload.requestType,
               code: msg.payload.code,
@@ -2053,13 +2214,20 @@ export class DaemonClient {
     });
   }
 
+  private unavailableError(): DaemonConnectionError {
+    return new DaemonConnectionError("Transport not connected", "DAEMON_CONNECTION_LOST", {
+      serverId: this.destinationServerId,
+      cause: this.connectionFailure,
+    });
+  }
+
   private sendSessionMessageStrict(message: SessionInboundMessage): void {
     if (!this.transport || this.connectionState.status !== "connected") {
-      throw new DaemonConnectionError("Transport not connected");
+      throw this.unavailableError();
     }
     const payload = SessionInboundMessageSchema.parse(message);
     try {
-      this.sendJsonMessage("session", payload.type, { type: "session", message: payload });
+      this.sendSessionPayload(payload);
     } catch (error) {
       throw error instanceof Error ? error : new Error(String(error));
     }
@@ -2874,7 +3042,9 @@ export class DaemonClient {
   // ============================================================================
 
   private readonly creations = new CreationClient({
-    supports: (feature) => this.lastServerInfoMessage?.features?.[feature] === true,
+    supports: (feature) => {
+      return this.lastServerInfoMessage?.features?.[feature] === true;
+    },
     requestId: () => this.createRequestId(),
     request: (kind, input) =>
       kind === "workspace"
@@ -6310,6 +6480,9 @@ export class DaemonClient {
       const auth = this.helloAuth;
       this.sendJsonMessage("hello", "hello", {
         type: "hello",
+        ...(this.config.sessionPermissions
+          ? { sessionPermissions: this.config.sessionPermissions }
+          : {}),
         clientId: this.config.clientId,
         clientType: this.config.clientType ?? "cli",
         protocolVersion: 1,
@@ -6484,6 +6657,16 @@ export class DaemonClient {
   }
 
   private tryHandleBinaryFrame(rawBytes: Uint8Array): boolean {
+    const hostFrame = decodeHostRouteFrame(rawBytes);
+    if (rawBytes[0] === HOST_ROUTE_OPCODE && !hostFrame) {
+      this.disposeTransport(1008, "Invalid host routing frame");
+      this.scheduleReconnect({ reason: "Invalid host routing frame", event: "INVALID_HOST_FRAME" });
+      return true;
+    }
+    if (hostFrame) {
+      for (const handler of this.hostFrameListeners) handler(hostFrame);
+      return true;
+    }
     const fileFrame = decodeFileTransferFrame(rawBytes);
     if (fileFrame) {
       this.traceInstant("paseo.ws.message.inbound", {
@@ -6655,6 +6838,7 @@ export class DaemonClient {
   }
 
   private scheduleReconnect(input?: {
+    failure?: Error;
     reason?: string;
     event?: string;
     reasonCode?: string;
@@ -6675,9 +6859,18 @@ export class DaemonClient {
 
     // Clear all pending waiters and queued sends since the connection was lost
     // and responses from the previous connection will never arrive.
-    this.clearWaiters(new DaemonConnectionError(reason ?? "Connection lost"));
-    this.rejectPendingSendQueue(new DaemonConnectionError(reason ?? "Connection lost"));
-    this.rejectPingProbe(new DaemonConnectionError(reason ?? "Connection lost"));
+    const connectionError = new DaemonConnectionError(
+      reason ?? "Connection lost",
+      "DAEMON_CONNECTION_LOST",
+      {
+        serverId: this.destinationServerId,
+        cause: input?.failure,
+      },
+    );
+    this.connectionFailure = input?.failure;
+    this.clearWaiters(connectionError, true);
+    this.rejectPendingSendQueue(connectionError);
+    this.rejectPingProbe(connectionError);
     this.terminalStreams.clearSlots();
     this.lastServerInfoMessage = null;
 
@@ -6692,7 +6885,7 @@ export class DaemonClient {
       this.rejectConnect(
         this.authFailureReasonValue
           ? new DaemonAuthenticationError(this.authFailureReasonValue)
-          : new Error(reason ?? "Transport disconnected before connect"),
+          : (input?.failure ?? new Error(reason ?? "Transport disconnected before connect")),
       );
       return;
     }
@@ -6702,11 +6895,26 @@ export class DaemonClient {
 
   private emitDisconnectedStateForReconnect(
     reason: string | undefined,
-    input: { reason?: string; event?: string; reasonCode?: string } | undefined,
+    input: { reason?: string; event?: string; reasonCode?: string; failure?: Error } | undefined,
   ): void {
     this.updateConnectionState(
       {
         status: "disconnected",
+        ...(input?.failure && "code" in input.failure && typeof input.failure.code === "string"
+          ? { failureCode: input.failure.code }
+          : {}),
+        ...(this.shouldReconnect && this.config.reconnect?.enabled !== false
+          ? {
+              nextRetryAt: new Date(
+                Date.now() +
+                  Math.min(
+                    (this.config.reconnect?.baseDelayMs ?? DEFAULT_RECONNECT_BASE_DELAY_MS) *
+                      2 ** this.reconnectAttempt,
+                    this.config.reconnect?.maxDelayMs ?? DEFAULT_RECONNECT_MAX_DELAY_MS,
+                  ),
+              ).toISOString(),
+            }
+          : {}),
         ...(reason ? { reason } : {}),
       },
       {
@@ -6800,6 +7008,8 @@ export class DaemonClient {
       const serverInfo = parseServerInfoStatusPayload(consumerMessage.payload);
       if (serverInfo) {
         this.lastServerInfoMessage = serverInfo;
+        this.destinationServerId = serverInfo.serverId;
+        this.connectionFailure = undefined;
         if (this.connectionState.status === "connecting") {
           this.resetConnectTimeout();
           this.reconnectAttempt = 0;
@@ -6883,12 +7093,20 @@ export class DaemonClient {
     }
   }
 
-  private clearWaiters(error: Error): void {
+  private clearWaiters(error: Error, connectionLost = false): void {
     for (const waiter of Array.from(this.waiters)) {
       if (waiter.timeoutHandle) {
         clearTimeout(waiter.timeoutHandle);
       }
-      waiter.reject(error);
+      waiter.reject(
+        connectionLost && waiter.submitted && error instanceof DaemonConnectionError
+          ? new DaemonConnectionError(error.message, error.code, {
+              serverId: error.serverId,
+              outcome: "unknown",
+              cause: error.cause instanceof Error ? error.cause : undefined,
+            })
+          : error,
+      );
     }
     this.waiters.clear();
   }
@@ -6995,6 +7213,7 @@ export class DaemonClient {
         reject: wrappedReject,
         timeoutHandle,
         requestId: options?.requestId,
+        submitted: false,
       };
       this.waiters.add(waiter);
     });

packages/client/src/hosts/index.ts

@@ -0,0 +1,208 @@
+import { safeRandomId } from "../daemon-client-transport-utils.js";
+import { DaemonClient } from "../daemon-client.js";
+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";
+
+export interface PaseoHostActions {
+  list(): Promise<HostDirectory>;
+  observe(): OwnedSubscription<HostDirectory>;
+  add(input: HostAdd): Promise<HostSummary>;
+  update(registrationId: string, patch: HostUpdate): Promise<HostSummary>;
+  remove(registrationId: string): Promise<void>;
+  client(serverId: string): Promise<PaseoApi>;
+}
+
+export function createHostChannelTransport(
+  origin: DaemonClient,
+  serverId: string,
+  scopeId: string,
+  initialRouteId: string | null,
+  link = false,
+): DaemonTransportFactory {
+  let first: string | null = initialRouteId;
+  return () => {
+    const opens = new Set<() => void>();
+    const closes = new Set<(event?: unknown) => void>();
+    const messages = new Set<(data: unknown, binary: boolean) => void>();
+    const errors = new Set<(error: unknown) => void>();
+    let routeId: string | null = null;
+    let closed = false;
+    const cleanups: (() => void)[] = [];
+    const close = (error?: unknown) => {
+      if (closed) return;
+      closed = true;
+      if (error instanceof Error) Object.assign(error, { serverId });
+      for (const cleanup of cleanups) cleanup();
+      try {
+        if (routeId) origin.closeHostRoute(routeId);
+        else origin.cancelHostRoute(scopeId);
+      } catch {
+        /* Transport already closed. */
+      }
+      if (error) for (const handler of errors) handler(error);
+      for (const handler of closes) handler(error);
+    };
+    cleanups.push(
+      origin.subscribeRawMessages((message) => {
+        if (message.type === "hosts.route.frame" && message.payload.subscriptionId === routeId) {
+          for (const handler of messages) handler(message.payload.data, false);
+        }
+        if (message.type === "hosts.route.closed" && message.payload.subscriptionId === routeId)
+          close(
+            Object.assign(new Error(message.payload.message), {
+              code: message.payload.code,
+              serverId,
+            }),
+          );
+      }),
+    );
+    cleanups.push(
+      origin.subscribeHostFrames((frame) => {
+        if (frame.routeId === routeId) for (const handler of messages) handler(frame.payload, true);
+      }),
+    );
+    cleanups.push(
+      origin.subscribeConnectionStatus((connectionState) => {
+        if (connectionState.status !== "connected") close();
+      }),
+    );
+    let opening: Promise<string>;
+    if (first) opening = Promise.resolve(first);
+    else opening = link ? origin.openHostLink(scopeId) : origin.openHostRoute(serverId, scopeId);
+    first = null;
+    void opening
+      .then((id) => {
+        if (closed) {
+          origin.closeHostRoute(id);
+          return;
+        }
+        routeId = id;
+        for (const handler of opens) handler();
+        return undefined;
+      })
+      .catch(close);
+    return {
+      send(data) {
+        if (closed || !routeId)
+          throw Object.assign(new Error(`Host ${serverId} is offline`), {
+            code: "host_offline",
+            serverId,
+          });
+        origin.sendHostFrame(
+          routeId,
+          typeof data === "string" || data instanceof Uint8Array ? data : new Uint8Array(data),
+        );
+      },
+      close: () => close(),
+      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);
+      },
+    };
+  };
+}
+
+export function createHostActions(
+  origin: DaemonClient,
+  owner: {
+    own<T extends { release(): Promise<void> }>(create: () => T): T;
+    signal: AbortSignal;
+    api: typeof createPaseoApi;
+  },
+): PaseoHostActions {
+  const assertActive = () => {
+    if (owner.signal.aborted) throw new Error("Paseo API is disposed");
+  };
+  return {
+    list: () => {
+      assertActive();
+      return origin.listHosts();
+    },
+    observe: () => owner.own(() => origin.observeHosts()),
+    add: (input) => {
+      assertActive();
+      return origin.addHost(input);
+    },
+    update: (id, patch) => {
+      assertActive();
+      return origin.updateHost(id, patch);
+    },
+    remove: (id) => {
+      assertActive();
+      return origin.removeHost(id);
+    },
+    async client(serverId) {
+      assertActive();
+      origin.requireConnectedHosts();
+      if (origin.getLastServerInfoMessage()?.serverId === serverId) {
+        const api = owner.api(origin, { signal: owner.signal });
+        const scope = owner.own(() => ({ release: () => api.dispose() }));
+        return { ...api, dispose: () => scope.release() };
+      }
+      const scopeId = safeRandomId();
+      const cancel = () => origin.cancelHostRoute(scopeId);
+      owner.signal.addEventListener("abort", cancel, { once: true });
+      let routeId: string;
+      try {
+        routeId = await origin.openHostRoute(serverId, scopeId);
+      } catch (error) {
+        throw Object.assign(error instanceof Error ? error : new Error(String(error)), {
+          serverId,
+        });
+      } finally {
+        owner.signal.removeEventListener("abort", cancel);
+      }
+      if (owner.signal.aborted) {
+        origin.closeHostRoute(routeId);
+        assertActive();
+      }
+      const client = new DaemonClient({
+        url: "ws://routed-host/ws",
+        clientId: scopeId,
+        transportFactory: createHostChannelTransport(origin, serverId, scopeId, routeId),
+        // Initial acquisition has a deadline; established scopes restore demand on loss.
+        reconnect: { enabled: false },
+      });
+      const api = owner.api(client, { signal: owner.signal });
+      const scope = owner.own(() => ({
+        release: async () => {
+          try {
+            await api.dispose();
+          } finally {
+            await client.close();
+          }
+        },
+      }));
+      try {
+        await client.connect();
+        client.setReconnectEnabled(true);
+        assertActive();
+        return { ...api, dispose: () => scope.release() };
+      } catch (error) {
+        const failure = Object.assign(
+          error instanceof Error ? error : new Error("Host handshake failed"),
+          {
+            code: error instanceof Error && "code" in error ? error.code : "host_offline",
+            serverId,
+          },
+        );
+        await scope.release();
+        throw failure;
+      }
+    },
+  };
+}

packages/client/src/index.test.ts

@@ -267,6 +267,7 @@ test("createPaseoApi borrows daemon capabilities without exposing connection own
     "agents",
     "config",
     "dispose",
+    "hosts",
     "observeEvents",
     "projects",
     "providers",

packages/client/src/index.ts

@@ -1,3 +1,6 @@
+import { createHostActions, type PaseoHostActions } from "./hosts/index.js";
+export type { PaseoHostActions } from "./hosts/index.js";
+export type { HostAdd, HostUpdate, HostSummary, HostDirectory } from "@getpaseo/protocol/hosts";
 import type { OwnedSubscription } from "./connection/index.js";
 export type { OwnedSubscription, SubscriptionObserver } from "./connection/index.js";
 import type { DaemonClientConfig } from "./daemon-client.js";
@@ -484,6 +487,7 @@ export interface PaseoConfigActions {
 }
 
 export interface PaseoApi {
+  readonly hosts: PaseoHostActions;
   dispose(): Promise<void>;
   observeEvents: DaemonClient["observeEvents"];
   readonly terminals: PaseoTerminalActions;
@@ -695,6 +699,7 @@ export function createPaseoApi(
 
   return {
     dispose,
+    hosts: createHostActions(daemonClient, { own, signal: lifetime.signal, api: createPaseoApi }),
     observeEvents,
     terminals,
     projects: {
@@ -1034,6 +1039,7 @@ function listProviderUsage(
   daemonClient: DaemonClient,
   options?: PaseoProviderUsageOptions,
 ): Promise<PaseoProviderUsageResult> {
+  daemonClient.ensureConnected();
   // COMPAT(providerUsageList): added in v0.1.98, remove after 2027-02-28 once daemon floor >= v0.1.98.
   if (daemonClient.getLastServerInfoMessage()?.features?.providerUsageList !== true) {
     return Promise.reject(new Error("Update the host to list provider usage."));

packages/client/src/node/buffering.ts

@@ -0,0 +1,25 @@
+/** Counts bytes until the socket's send callback, shared by all links owned by a daemon. */
+export class HostTransportBudget {
+  private used = 0;
+  constructor(readonly limit = 256 * 1024 * 1024) {}
+  get bufferedBytes(): number {
+    return this.used;
+  }
+  assertCapacity(bytes: number): void {
+    if (bytes > this.limit - this.used)
+      throw Object.assign(new Error("Connected host buffer limit reached"), {
+        code: "resource_limit",
+      });
+  }
+  reserve(bytes: number): () => void {
+    this.assertCapacity(bytes);
+    this.used += bytes;
+    let released = false;
+    return () => {
+      if (!released) {
+        released = true;
+        this.used -= bytes;
+      }
+    };
+  }
+}

packages/client/src/node/connector.test.ts

@@ -0,0 +1,153 @@
+import { once } from "node:events";
+import { createServer } from "node:http";
+import { afterEach, expect, test } from "vitest";
+import { WebSocketServer, WebSocket } from "ws";
+import { createHostTransportFactory, HostTransportBudget } from "./index.js";
+const cleanup: (() => Promise<unknown>)[] = [];
+afterEach(async () => {
+  for (const close of cleanup.splice(0).toReversed()) await close();
+});
+async function endpoint() {
+  const server = createServer();
+  const wss = new WebSocketServer({ server });
+  server.listen(0, "127.0.0.1");
+  await once(server, "listening");
+  cleanup.push(async () => {
+    for (const socket of wss.clients) socket.terminate();
+    await new Promise<void>((resolve) => wss.close(() => server.close(() => resolve())));
+  });
+  const address = server.address();
+  if (!address || typeof address === "string") throw new Error("Missing port");
+  return { wss, url: `ws://127.0.0.1:${address.port}/ws` };
+}
+test("connector retains per-attempt headers and WebSocket subprotocols", async () => {
+  const server = await endpoint();
+  const connected = once(server.wss, "connection");
+  const transport = createHostTransportFactory(server.url)({
+    url: server.url,
+    headers: { Authorization: "Bearer example" },
+    protocols: ["paseo-test"],
+  });
+  cleanup.push(async () => transport.close());
+  const [socket, request] = (await connected) as [WebSocket, import("node:http").IncomingMessage];
+  expect(request.headers.authorization).toBe("Bearer example");
+  expect(socket.protocol).toBe("paseo-test");
+});
+test("a slow consumer cannot exceed the shared byte budget and close releases reserved bytes", async () => {
+  const server = await endpoint();
+  const budget = new HostTransportBudget(4 * 1024 * 1024);
+  const connected = once(server.wss, "connection");
+  const transport = createHostTransportFactory(server.url, budget)({ url: server.url });
+  cleanup.push(async () => transport.close());
+  const ready = new Promise<void>((resolve) => transport.onOpen(resolve));
+  const [socket] = (await connected) as [WebSocket];
+  socket.pause();
+  await ready;
+  const frame = new Uint8Array(1024 * 1024);
+  for (let index = 0; index < 4; index++) transport.send(frame);
+  expect(budget.bufferedBytes).toBe(4 * 1024 * 1024);
+  expect(() => transport.send(frame)).toThrow("Connected host buffer limit reached");
+  expect(budget.bufferedBytes).toBeLessThanOrEqual(budget.limit);
+  transport.close();
+  await expect.poll(() => budget.bufferedBytes).toBe(0);
+});
+test("direct connector preserves 25 MiB binary frames in both directions", async () => {
+  const server = await endpoint();
+  const connected = once(server.wss, "connection");
+  const transport = createHostTransportFactory(server.url)({ url: server.url });
+  cleanup.push(async () => transport.close());
+  const ready = new Promise<void>((resolve) => transport.onOpen(resolve));
+  const received = new Promise<Uint8Array>((resolve, reject) => {
+    transport.onMessage((bytes, binary) => binary && resolve(bytes as Uint8Array));
+    transport.onError(reject);
+  });
+  const [socket] = (await connected) as [WebSocket];
+  await ready;
+  const bytes = new Uint8Array(25 * 1024 * 1024).fill(173);
+  const echoed = once(socket, "message");
+  transport.send(bytes);
+  const [sent, binary] = await echoed;
+  expect(binary).toBe(true);
+  expect(Buffer.from(sent as Uint8Array).equals(bytes)).toBe(true);
+  socket.send(bytes);
+  expect(Buffer.from(await received).equals(bytes)).toBe(true);
+});
+
+import {
+  createDaemonChannel,
+  generateKeyPair,
+  exportPublicKey,
+  type Transport,
+} from "@getpaseo/relay/e2ee";
+test("encrypted connector rejects overload synchronously and preserves the physical link for sibling traffic", async () => {
+  const server = await endpoint();
+  const keys = generateKeyPair();
+  const received: (string | ArrayBuffer)[] = [];
+  let peer!: WebSocket;
+  let ready!: Promise<unknown>;
+  server.wss.on("connection", (socket) => {
+    peer = socket;
+    const wire: Transport = {
+      send: (data) => socket.send(data),
+      close: () => socket.close(),
+      onmessage: null,
+      onclose: null,
+      onerror: null,
+    };
+    socket.on("message", (data, isBinary) => {
+      const bytes = Buffer.from(data as Buffer);
+      wire.onmessage?.({
+        data: isBinary
+          ? bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength)
+          : bytes.toString(),
+        isBinary,
+      });
+    });
+    ready = createDaemonChannel(wire, keys, {
+      onmessage: (data) => {
+        received.push(data);
+      },
+    });
+  });
+  const port = new URL(server.url).port;
+  const budget = new HostTransportBudget(4 * 1024 * 1024);
+  const transport = createHostTransportFactory(
+    `relay://127.0.0.1:${port}/srv_slow?key=${encodeURIComponent(exportPublicKey(keys.publicKey))}`,
+    budget,
+  )({ url: server.url });
+  cleanup.push(async () => transport.close());
+  const errors: unknown[] = [];
+  let closed = false;
+  transport.onError((error) => errors.push(error));
+  transport.onClose(() => {
+    closed = true;
+  });
+  await new Promise<void>((resolve) => transport.onOpen(resolve));
+  await ready;
+  peer.pause();
+  const frame = new Uint8Array(1024 * 1024).fill(173);
+  let rejected: unknown;
+  for (let index = 0; index < 8; index++) {
+    try {
+      transport.send(frame);
+    } catch (error) {
+      rejected = error;
+      break;
+    }
+  }
+  expect(rejected).toMatchObject({ code: "resource_limit" });
+  expect(budget.bufferedBytes).toBeLessThanOrEqual(budget.limit);
+  await new Promise<void>((resolve) => setImmediate(resolve));
+  expect(errors).toEqual([]);
+  expect(closed).toBe(false);
+  transport.send("sibling control");
+  peer.resume();
+  await expect.poll(() => received.includes("sibling control")).toBe(true);
+  expect(received.filter((data) => data instanceof ArrayBuffer).length).toBeGreaterThan(0);
+  expect(
+    received
+      .filter((data) => data instanceof ArrayBuffer)
+      .every((data) => Buffer.from(data).equals(frame)),
+  ).toBe(true);
+  expect(server.wss.clients.size).toBe(1);
+});

packages/client/src/node/index.ts

@@ -0,0 +1,195 @@
+export { createHostChannelTransport } from "../hosts/index.js";
+import { HostTransportBudget } from "./buffering.js";
+export { HostTransportBudget } from "./buffering.js";
+import { homedir } from "node:os";
+import { WebSocket } from "ws";
+import { parseHostTarget, type HostTarget } from "@getpaseo/protocol/host-target";
+import { createSshTunnel, type SshTunnel } from "./ssh-tunnel.js";
+import { createWebSocketTransportFactory } from "../daemon-client-websocket-transport.js";
+import { createRelayE2eeTransportFactory } from "../daemon-client-relay-e2ee-transport.js";
+import type {
+  DaemonTransport,
+  DaemonTransportFactory,
+  WebSocketLike,
+} from "../daemon-client-transport-types.js";
+import { DaemonClient, type DaemonClientConfig } from "../daemon-client.js";
+
+export { parseHostTarget };
+export type { SshTunnel } from "./ssh-tunnel.js";
+export { createSshTunnel, resolveSshFailureDetail } from "./ssh-tunnel.js";
+
+/** A factory owns one attempt, including a fresh SSH process and its teardown. */
+export function createHostTransportFactory(
+  connection: string,
+  budget = new HostTransportBudget(),
+): DaemonTransportFactory {
+  const target = parseHostTarget(connection);
+  return (attemptOptions) => {
+    const opens = new Set<() => void>();
+    const closes = new Set<(event?: unknown) => void>();
+    const errors = new Set<(event?: unknown) => void>();
+    const messages = new Set<(data: unknown, binary: boolean) => void>();
+    let closed = false;
+    let closeEvent: unknown;
+    let transport: DaemonTransport | undefined;
+    let teardown = () => {};
+    let disposeSocket = () => {};
+    let assertOutboundCapacity = (bytes: number) => budget.assertCapacity(bytes);
+    let sshTunnel: SshTunnel | null = null;
+    const close = (code?: number, reason?: string) => {
+      if (closed) return;
+      closed = true;
+      try {
+        transport?.close(code, reason);
+      } finally {
+        try {
+          disposeSocket();
+          teardown();
+        } finally {
+          for (const handler of closes) handler(closeEvent);
+        }
+      }
+    };
+    void (async () => {
+      let socketTarget: Extract<HostTarget, { kind: "socket" }>;
+      if (target.kind === "ssh") {
+        const tunnel = await createSshTunnel(target.ssh);
+        sshTunnel = tunnel;
+        teardown = () => tunnel.close();
+        socketTarget = parseHostTarget(tunnel.endpoint) as Extract<HostTarget, { kind: "socket" }>;
+      } else socketTarget = target;
+      if (closed) {
+        teardown();
+        return;
+      }
+      const socketPath = socketTarget.socketPath?.replace(/^~(?=\/)/, homedir());
+      let factory = createWebSocketTransportFactory((url, options) => {
+        const socket = new WebSocket(
+          socketPath && url.startsWith("ws+unix:") ? `ws+unix://${socketPath}:/ws` : url,
+          options?.protocols,
+          {
+            headers: options?.headers,
+            ...(socketPath ? { socketPath } : {}),
+            maxPayload: 100 * 1024 * 1024,
+          },
+        );
+        disposeSocket = () => socket.terminate();
+        assertOutboundCapacity = (bytes) => {
+          if (socket.bufferedAmount + bytes > 128 * 1024 * 1024) {
+            throw Object.assign(new Error("Host socket buffer limit reached"), {
+              code: "resource_limit",
+            });
+          }
+          budget.assertCapacity(bytes);
+        };
+        const pending = new Set<() => void>();
+        const send = socket.send.bind(socket);
+        socket.send = (data, ...args: unknown[]) => {
+          const bytes =
+            typeof data === "string"
+              ? Buffer.byteLength(data)
+              : Buffer.byteLength(data as Uint8Array);
+          assertOutboundCapacity(bytes);
+          const release = budget.reserve(bytes);
+          pending.add(release);
+          const callback = args.find(
+            (value): value is (error?: Error) => void => typeof value === "function",
+          );
+          const sendOptions = args.find((value) => value !== null && typeof value === "object") as
+            | { binary?: boolean }
+            | undefined;
+          try {
+            send(data, sendOptions ?? {}, (error) => {
+              release();
+              pending.delete(release);
+              callback?.(error);
+            });
+          } catch (error) {
+            release();
+            pending.delete(release);
+            throw error;
+          }
+        };
+        socket.once("close", () => {
+          for (const release of pending) release();
+          pending.clear();
+        });
+        return socket as unknown as WebSocketLike;
+      });
+      if (socketTarget.offer)
+        factory = createRelayE2eeTransportFactory({
+          baseFactory: factory,
+          daemonPublicKeyB64: socketTarget.offer.daemonPublicKeyB64,
+          logger: { warn() {} },
+          assertOutboundCapacity: (bytes) => assertOutboundCapacity(bytes),
+        });
+      transport = factory({ ...attemptOptions, url: socketTarget.url });
+      transport.onOpen(() => {
+        if (!closed) for (const handler of opens) handler();
+      });
+      transport.onClose((event) => {
+        closeEvent = event;
+        close();
+      });
+      transport.onError((error) => {
+        if (closed) return;
+        const detail = sshTunnel?.failureDetail() ?? "";
+        const failure =
+          /permission denied|authentication failed|host key verification failed|REMOTE HOST IDENTIFICATION HAS CHANGED/i.test(
+            detail,
+          )
+            ? Object.assign(
+                new Error(
+                  "SSH authentication or host verification failed. Check the SSH configuration on this daemon.",
+                ),
+                { code: "authentication_failed" },
+              )
+            : error;
+        for (const handler of errors) handler(failure);
+        close();
+      });
+      transport.onMessage((data, binary) => {
+        if (!closed) for (const handler of messages) handler(data, binary);
+      });
+    })().catch(() => {
+      for (const handler of errors) handler(new Error("Host transport failed"));
+      close();
+    });
+    return {
+      send(data) {
+        if (closed || !transport) throw new Error("Host transport is unavailable");
+        transport.send(data);
+      },
+      close,
+      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);
+      },
+    };
+  };
+}
+export function createHostClient(
+  connection: string,
+  config: Omit<DaemonClientConfig, "url" | "transportFactory">,
+  budget?: HostTransportBudget,
+): DaemonClient {
+  const target = parseHostTarget(connection);
+  return new DaemonClient({
+    ...config,
+    url: "ws://host-connector/ws",
+    ...(target.password ? { password: target.password } : {}),
+    transportFactory: createHostTransportFactory(connection, budget),
+  });
+}

packages/client/src/node/ssh-tunnel.ts

@@ -0,0 +1,90 @@
+import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
+import { createServer, type Server, type Socket } from "node:net";
+import { buildSshTunnelArgs, type SshTransportTarget } from "@getpaseo/protocol/ssh-transport";
+
+const SSH_STDERR_LIMIT = 8192;
+
+export interface SshTunnel {
+  endpoint: string;
+  close(): void;
+  failureDetail(): string | null;
+}
+
+function formatSshFailure(
+  stderr: string,
+  code: number | null,
+  signal: NodeJS.Signals | null,
+): string {
+  const detail = stderr.trim();
+  if (detail) return detail;
+  if (signal) return `ssh exited with signal ${signal}`;
+  return `ssh exited with code ${code ?? "unknown"}`;
+}
+
+export function resolveSshFailureDetail(failure: string | null, stderr: string): string | null {
+  return failure ?? (stderr.trim() || null);
+}
+
+export function createSshTunnel(target: SshTransportTarget): Promise<SshTunnel> {
+  let server: Server | null = null;
+  let socket: Socket | null = null;
+  let child: ChildProcessWithoutNullStreams | null = null;
+  let stderr = "";
+  let failure: string | null = null;
+
+  function close(): void {
+    server?.close();
+    server = null;
+    socket?.destroy();
+    socket = null;
+    if (child && !child.killed) child.kill();
+    child = null;
+  }
+
+  return new Promise((resolve, reject) => {
+    server = createServer((acceptedSocket) => {
+      socket = acceptedSocket;
+      server?.close();
+      server = null;
+
+      child = spawn("ssh", buildSshTunnelArgs(target), {
+        stdio: ["pipe", "pipe", "pipe"],
+        windowsHide: true,
+      });
+      child.stderr.on("data", (chunk: Buffer | string) => {
+        stderr = `${stderr}${chunk.toString()}`.slice(-SSH_STDERR_LIMIT);
+      });
+      child.on("error", (error) => {
+        failure = error.message;
+        acceptedSocket.destroy(error);
+      });
+      child.on("exit", (code, signal) => {
+        if (code !== 0 || signal) failure = formatSshFailure(stderr, code, signal);
+        acceptedSocket.destroy(failure ? new Error(failure) : undefined);
+      });
+      acceptedSocket.on("error", () => undefined);
+      acceptedSocket.on("close", () => {
+        if (child && !child.killed) child.kill();
+      });
+      acceptedSocket.pipe(child.stdin);
+      child.stdout.pipe(acceptedSocket);
+    });
+    server.once("error", (error) => {
+      close();
+      reject(error);
+    });
+    server.listen(0, "127.0.0.1", () => {
+      const address = server?.address();
+      if (!address || typeof address === "string") {
+        close();
+        reject(new Error("Failed to allocate the SSH tunnel port"));
+        return;
+      }
+      resolve({
+        endpoint: `127.0.0.1:${address.port}`,
+        close,
+        failureDetail: () => resolveSshFailureDetail(failure, stderr),
+      });
+    });
+  });
+}

packages/desktop/e2e/browser-tabs.e2e.mjs

@@ -12,7 +12,7 @@ import { fileURLToPath } from "node:url";
 import { experimental_createMCPClient } from "ai";
 import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
 import { chromium } from "playwright";
-import { hideDeveloperToasts } from "../../app/e2e/support/developer-ui.mjs";
+import { installDeveloperToastDismissal } from "../../app/e2e/support/developer-ui.mjs";
 import { runAppearanceFontSizeRegression } from "./appearance-font-size.electron.mjs";
 import { runSettingsMemoryRegression } from "./settings-memory.electron.mjs";
 
@@ -1189,7 +1189,7 @@ async function main() {
 
     browser = await chromium.connectOverCDP(`http://127.0.0.1:${cdpPort}`);
     const page = await waitForAppPage(browser, expoPort);
-    await hideDeveloperToasts(page);
+    await installDeveloperToastDismissal(page);
     const status = await waitForDesktopStatus(page);
 
     const checkPluginLinks = () =>

packages/protocol/src/binary-frames/demux.test.ts

@@ -2,6 +2,9 @@ import { describe, expect, it } from "vitest";
 
 import {
   decodeBinaryFrame,
+  encodeHostRouteFrame,
+  decodeHostRouteFrame,
+  MAX_HOST_ROUTE_ID_BYTES,
   encodeFileTransferFrame,
   encodeTerminalStreamFrame,
   FileTransferOpcode,
@@ -51,3 +54,38 @@ describe("binary frame demux", () => {
     expect(decodeBinaryFrame(new Uint8Array([0xff, 0]))).toBeNull();
   });
 });
+
+describe("host route binary framing", () => {
+  it("wraps terminal and file frames unchanged and leaves their direct interpretation intact", () => {
+    const frames = [
+      encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Input, slot: 7, payload: "echo" }),
+      encodeFileTransferFrame({
+        opcode: FileTransferOpcode.FileChunk,
+        requestId: "upload",
+        payload: new Uint8Array([0, 255, 128, 1]),
+      }),
+    ];
+    for (const payload of frames) {
+      const wire = encodeHostRouteFrame("channel", payload);
+      const decoded = decodeBinaryFrame(wire);
+      expect(decoded).toEqual({ kind: "host_route", frame: { routeId: "channel", payload } });
+      expect(wire.subarray(10)).toEqual(payload);
+      expect(decodeBinaryFrame(decodeHostRouteFrame(wire)!.payload)).toEqual(
+        decodeBinaryFrame(payload),
+      );
+    }
+  });
+  it("rejects truncated, oversized identifiers and invalid UTF-8 without throwing", () => {
+    for (const bytes of [
+      new Uint8Array([0x30]),
+      new Uint8Array([0x30, 0, 4, 65]),
+      new Uint8Array([0x30, 0, 0]),
+      new Uint8Array([0x30, 1, 1]),
+      new Uint8Array([0x30, 0, 1, 255]),
+    ])
+      expect(decodeBinaryFrame(bytes)).toBeNull();
+    expect(() =>
+      encodeHostRouteFrame("a".repeat(MAX_HOST_ROUTE_ID_BYTES + 1), new Uint8Array()),
+    ).toThrow("Invalid host route ID");
+  });
+});

packages/protocol/src/binary-frames/demux.ts

@@ -1,3 +1,4 @@
+import { HOST_ROUTE_OPCODE, decodeHostRouteFrame, type HostRouteFrame } from "./host-route.js";
 import {
   decodeFileTransferFrame,
   FileTransferOpcode,
@@ -10,11 +11,16 @@ import {
 } from "./terminal.js";
 
 export type BinaryFrame =
+  | { kind: "host_route"; frame: HostRouteFrame }
   | { kind: "terminal"; frame: TerminalStreamFrame }
   | { kind: "file_transfer"; frame: FileTransferFrame };
 
 export function decodeBinaryFrame(bytes: Uint8Array): BinaryFrame | null {
   switch (bytes[0]) {
+    case HOST_ROUTE_OPCODE: {
+      const frame = decodeHostRouteFrame(bytes);
+      return frame ? { kind: "host_route", frame } : null;
+    }
     case TerminalStreamOpcode.Output:
     case TerminalStreamOpcode.Input:
     case TerminalStreamOpcode.Resize:

packages/protocol/src/binary-frames/host-route.ts

@@ -0,0 +1,33 @@
+import { MAX_WEBSOCKET_MESSAGE_BYTES } from "../websocket-message-limit.js";
+/** Routing adds a bounded header; the enclosed terminal/file frame is never re-encoded. */
+export const HOST_ROUTE_OPCODE = 0x30;
+export const MAX_HOST_ROUTE_ID_BYTES = 256;
+export const MAX_HOST_ROUTE_HEADER_BYTES = 3 + MAX_HOST_ROUTE_ID_BYTES;
+export const MAX_ROUTED_PAYLOAD_BYTES = MAX_WEBSOCKET_MESSAGE_BYTES - MAX_HOST_ROUTE_HEADER_BYTES;
+export interface HostRouteFrame {
+  routeId: string;
+  payload: Uint8Array;
+}
+export function encodeHostRouteFrame(routeId: string, payload: Uint8Array): Uint8Array {
+  const id = new TextEncoder().encode(routeId);
+  if (!id.length || id.length > MAX_HOST_ROUTE_ID_BYTES) throw new Error("Invalid host route ID");
+  if (payload.byteLength > MAX_ROUTED_PAYLOAD_BYTES) throw new Error("Host frame is too large");
+  const frame = new Uint8Array(3 + id.length + payload.byteLength);
+  frame[0] = HOST_ROUTE_OPCODE;
+  new DataView(frame.buffer).setUint16(1, id.length);
+  frame.set(id, 3);
+  frame.set(payload, 3 + id.length);
+  return frame;
+}
+export function decodeHostRouteFrame(bytes: Uint8Array): HostRouteFrame | null {
+  if (bytes[0] !== HOST_ROUTE_OPCODE || bytes.length < 3) return null;
+  const length = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength).getUint16(1);
+  if (!length || length > MAX_HOST_ROUTE_ID_BYTES || bytes.length < 3 + length) return null;
+  if (bytes.length - 3 - length > MAX_ROUTED_PAYLOAD_BYTES) return null;
+  try {
+    const routeId = new TextDecoder("utf-8", { fatal: true }).decode(bytes.subarray(3, 3 + length));
+    return { routeId, payload: bytes.subarray(3 + length) };
+  } catch {
+    return null;
+  }
+}

packages/protocol/src/binary-frames/index.ts

@@ -1,3 +1,4 @@
 export * from "./demux.js";
 export * from "./file-transfer.js";
 export * from "./terminal.js";
+export * from "./host-route.js";

packages/protocol/src/host-target.test.ts

@@ -0,0 +1,76 @@
+import { expect, test } from "vitest";
+import { parseHostTarget } from "./host-target.js";
+
+test.each([
+  ["127.0.0.1:6767", "tcp://127.0.0.1:6767"],
+  ["4321", "tcp://127.0.0.1:4321"],
+  ["[::1]:4321", "tcp://[::1]:4321"],
+  ["unix:///tmp/paseo.sock", "unix:///tmp/paseo.sock"],
+  ["pipe://\\\\.\\pipe\\paseo", "pipe://\\\\.\\pipe\\paseo"],
+  ["wss://example.com/custom?token=secret", "wss://example.com"],
+  ["tcp://example.com:6767?password=secret", "tcp://example.com:6767"],
+])(
+  "the shared connector accepts %s and exposes only safe display text",
+  (connection, description) => {
+    expect(parseHostTarget(connection).description).toBe(description);
+  },
+);
+test("SSH daemon credentials are separate from SSH authentication and excluded from display text", () => {
+  expect(parseHostTarget("ssh://deploy@build-box:2222?daemonPort=7777&password=secret")).toEqual({
+    kind: "ssh",
+    ssh: { host: "deploy@build-box", sshPort: 2222, daemonPort: 7777 },
+    password: "secret",
+    description: "ssh://build-box:2222 (daemon port 7777)",
+  });
+});
+test.each([
+  "",
+  "ssh://-bad",
+  "ssh://user:secret@host",
+  "ssh://host?unknown=secret",
+  "unix://",
+  "tcp://host:0",
+  "C:\\Users\\socket",
+  "alice:example-secret@localhost:6767",
+  "not-a-scheme://localhost:6767",
+  "localhost/path:6767",
+  "localhost?password=example-secret:6767",
+  "localhost#example-secret:6767",
+  "tcp://localhost/path:6767",
+  "relay://alice:example-secret@localhost:6767/server?key=private",
+])("invalid target syntax is rejected without echoing %s", (connection) => {
+  expect(() => parseHostTarget(connection)).toThrow("Invalid host connection");
+});
+
+test("SSH aliases and DNS names retain their connection semantics", () => {
+  expect(parseHostTarget("build-box.example.:6767").description).toBe(
+    "tcp://build-box.example.:6767",
+  );
+  expect(parseHostTarget("ssh://deploy@my_ssh-alias")).toMatchObject({
+    kind: "ssh",
+    ssh: { host: "deploy@my_ssh-alias" },
+  });
+});
+
+test("encoded SSH usernames cannot leak into the target description", () => {
+  expect(parseHostTarget("ssh://alice%3Aexample-secret@my-alias").description).toBe(
+    "ssh://my-alias (daemon port 6767)",
+  );
+});
+
+test("pairing offers cannot publish a credential-bearing or malformed relay authority", () => {
+  for (const endpoint of [
+    "alice:example-secret@localhost:6767",
+    "localhost/path:6767",
+    "localhost?token=example-secret:6767",
+  ]) {
+    const offer = {
+      v: 2,
+      serverId: "srv_test",
+      daemonPublicKeyB64: "private-key",
+      relay: { endpoint, useTls: false },
+    };
+    const url = `https://app.paseo.sh/#offer=${Buffer.from(JSON.stringify(offer)).toString("base64url")}`;
+    expect(() => parseHostTarget(url)).toThrow("Invalid host connection");
+  }
+});

packages/protocol/src/host-target.ts

@@ -0,0 +1,137 @@
+import { parseConnectionOfferFromUrl, type ConnectionOffer } from "./connection-offer.js";
+import {
+  buildDaemonWebSocketUrl,
+  buildRelayWebSocketUrl,
+  normalizeHostPort,
+  parseConnectionUri,
+  parseRelayConnectionUri,
+  shouldUseTlsForDefaultHostedRelay,
+} from "./daemon-endpoints.js";
+import { parseSshTransportUri, type SshTransportTarget } from "./ssh-transport.js";
+
+export type HostTarget =
+  | { kind: "ssh"; ssh: SshTransportTarget; password?: string; description: string }
+  | {
+      kind: "socket";
+      url: string;
+      socketPath?: string;
+      password?: string;
+      offer?: ConnectionOffer;
+      description: string;
+    };
+
+/** The shared saved-host and CLI grammar. Errors never echo credential-bearing input. */
+export function parseHostTarget(input: string): HostTarget {
+  try {
+    return parse(input.trim());
+  } catch {
+    throw Object.assign(
+      new Error(
+        "Invalid host connection. Use host:port, ws(s)://, tcp://, unix://, pipe://, ssh:// or a pairing URL.",
+      ),
+      { code: "invalid_connection" },
+    );
+  }
+}
+function parse(input: string): HostTarget {
+  if (!input) throw new Error("empty");
+  if (input.startsWith("ssh://")) {
+    return sshTarget(input);
+  }
+
+  if (input.startsWith("relay://") || input.includes("#connect=")) {
+    const relay = parseRelayConnectionUri(input);
+    return relayTarget(relay.offer, relay.password);
+  }
+  const offer = parseConnectionOfferFromUrl(input);
+  if (offer) return relayTarget(offer);
+  if (
+    input.startsWith("unix://") ||
+    input.startsWith("pipe://") ||
+    input.startsWith("/") ||
+    input.startsWith("~") ||
+    input.startsWith("\\\\.\\pipe\\")
+  ) {
+    const socketPath = input.replace(/^(unix|pipe):\/\//, "");
+    if (!socketPath || socketPath === "~") throw new Error("empty socket");
+    const pipe = input.startsWith("pipe://") || input.startsWith("\\\\.\\pipe\\");
+    return {
+      kind: "socket",
+      socketPath,
+      url: pipe ? "ws://localhost/ws" : `ws+unix://${socketPath}:/ws`,
+      description: `${pipe ? "pipe" : "unix"}://${socketPath}`,
+    };
+  }
+  if (input.startsWith("tcp://")) return tcpTarget(input);
+  if (/^wss?:\/\//.test(input)) return websocketTarget(input);
+  const endpoint = validatedEndpoint(/^\d+$/.test(input) ? `127.0.0.1:${input}` : input);
+  return {
+    kind: "socket",
+    url: buildDaemonWebSocketUrl(endpoint, { useTls: false }),
+    description: `tcp://${endpoint}`,
+  };
+}
+function relayTarget(offer: ConnectionOffer, password?: string): HostTarget {
+  const endpoint = validatedEndpoint(offer.relay.endpoint);
+  return {
+    kind: "socket",
+    url: buildRelayWebSocketUrl({
+      endpoint,
+      serverId: offer.serverId,
+      role: "client",
+      useTls: offer.relay.useTls ?? shouldUseTlsForDefaultHostedRelay(offer.relay.endpoint),
+    }),
+    offer,
+    password,
+    description: `relay://${endpoint}/${encodeURIComponent(offer.serverId)}`,
+  };
+}
+
+function tcpTarget(input: string): HostTarget {
+  const uri = new URL(input);
+  if (uri.pathname && uri.pathname !== "/") throw new Error("Unexpected TCP path");
+  if (uri.hash) throw new Error("Unexpected TCP fragment");
+  const parts = parseConnectionUri(input);
+  const endpoint = validatedEndpoint(
+    `${parts.isIpv6 ? `[${parts.host}]` : parts.host}:${parts.port}`,
+  );
+  return {
+    kind: "socket",
+    url: buildDaemonWebSocketUrl(endpoint, { useTls: parts.useTls }),
+    password: parts.password,
+    description: `tcp://${endpoint}`,
+  };
+}
+function websocketTarget(input: string): HostTarget {
+  const url = new URL(input);
+  if (!url.hostname || url.username || url.password || url.hash) throw new Error("invalid url");
+  const password = url.searchParams.get("password") ?? undefined;
+  // Unknown query parameters may themselves contain credentials; discovery shows only the origin.
+  if (url.pathname === "/") url.pathname = "/ws";
+  url.searchParams.delete("password");
+  return { kind: "socket", url: url.toString(), password, description: url.origin };
+}
+
+function sshTarget(input: string): HostTarget {
+  const uri = new URL(input);
+  if (uri.searchParams.getAll("password").length > 1) throw new Error("Duplicate daemon password");
+  const password = uri.searchParams.get("password") || undefined;
+  uri.searchParams.delete("password");
+  const ssh = parseSshTransportUri(uri.toString());
+  return {
+    kind: "ssh",
+    ssh,
+    ...(password ? { password } : {}),
+    description: `ssh://${uri.hostname}${ssh.sshPort ? `:${ssh.sshPort}` : ""} (daemon port ${ssh.daemonPort})`,
+  };
+}
+
+/** A bare endpoint has exactly one authority; URL syntax must use an explicit supported scheme. */
+function validatedEndpoint(input: string): string {
+  if (!/^(?:[a-zA-Z0-9_][a-zA-Z0-9_.-]*|\[[0-9a-fA-F:.]+\]):[0-9]{1,5}$/.test(input))
+    throw new Error("Invalid endpoint");
+  // URL parsing validates IPv6 literals and rejects invalid authority/port encodings.
+  const url = new URL(`http://${input}`);
+  if (!url.hostname) throw new Error("Missing endpoint host");
+  return normalizeHostPort(input);
+}

packages/protocol/src/hosts.ts

@@ -0,0 +1,123 @@
+import { z } from "zod";
+
+export const HostMetadataSchema = z.record(z.string().max(256), z.string().max(4096));
+export const HostAddSchema = z.object({
+  connection: z.string().min(1).max(16384),
+  label: z.string().max(256).optional(),
+  metadata: HostMetadataSchema.optional(),
+  sourceKey: z.string().min(1).max(256).optional(),
+});
+export const HostUpdateSchema = z.object({
+  label: z.string().max(256).nullable().optional(),
+  metadata: HostMetadataSchema.optional(),
+  enabled: z.boolean().optional(),
+  connection: z.string().min(1).max(16384).optional(),
+  replaceIdentity: z.boolean().optional(),
+});
+export const HostSummarySchema = z.object({
+  registrationId: z.string(),
+  serverId: z.string().nullable(),
+  label: z.string().nullable(),
+  source: z.object({ id: z.string(), key: z.string() }),
+  metadata: HostMetadataSchema,
+  enabled: z.boolean(),
+  status: z.enum(["connecting", "online", "offline", "error", "disabled"]),
+  targetDescription: z.string(),
+  lastOnlineAt: z.string().nullable(),
+  nextRetryAt: z.string().nullable(),
+  error: z.object({ code: z.string(), message: z.string() }).nullable(),
+});
+export const HostDirectorySchema = z.object({
+  self: z.object({ serverId: z.string(), label: z.string() }),
+  hosts: z.array(HostSummarySchema),
+});
+export type HostAdd = z.infer<typeof HostAddSchema>;
+export type HostUpdate = z.infer<typeof HostUpdateSchema>;
+export type HostSummary = z.infer<typeof HostSummarySchema>;
+export type HostDirectory = z.infer<typeof HostDirectorySchema>;
+
+export const HostsListRequestSchema = z.object({
+  type: z.literal("hosts.list.request"),
+  requestId: z.string(),
+  observe: z.boolean().optional(),
+});
+export const HostsAddRequestSchema = z.object({
+  type: z.literal("hosts.add.request"),
+  requestId: z.string(),
+  input: HostAddSchema,
+});
+export const HostsUpdateRequestSchema = z.object({
+  type: z.literal("hosts.update.request"),
+  requestId: z.string(),
+  registrationId: z.string(),
+  patch: HostUpdateSchema,
+});
+export const HostsRemoveRequestSchema = z.object({
+  type: z.literal("hosts.remove.request"),
+  requestId: z.string(),
+  registrationId: z.string(),
+});
+export const HostsRouteOpenRequestSchema = z.object({
+  type: z.literal("hosts.route.open.request"),
+  requestId: z.string(),
+  serverId: z.string(),
+  scopeId: z.string().min(1).max(256),
+});
+export const HostsRouteFrameRequestSchema = z.object({
+  type: z.literal("hosts.route.frame.request"),
+  routeId: z.string(),
+  data: z.string().max(100 * 1024 * 1024),
+  binary: z.literal(false),
+});
+export const HostsRouteCloseRequestSchema = z.object({
+  type: z.literal("hosts.route.close.request"),
+  routeId: z.string().optional(),
+  scopeId: z.string().optional(),
+});
+export const HostsListResponseSchema = z.object({
+  type: z.literal("hosts.list.response"),
+  payload: z.object({
+    requestId: z.string(),
+    directory: HostDirectorySchema,
+    subscriptionId: z.string().optional(),
+  }),
+});
+export const HostsChangedSchema = z.object({
+  type: z.literal("hosts.changed"),
+  payload: z.object({ subscriptionId: z.string(), directory: HostDirectorySchema }),
+});
+export const HostsAddResponseSchema = z.object({
+  type: z.literal("hosts.add.response"),
+  payload: z.object({ requestId: z.string(), host: HostSummarySchema }),
+});
+export const HostsUpdateResponseSchema = z.object({
+  type: z.literal("hosts.update.response"),
+  payload: z.object({ requestId: z.string(), host: HostSummarySchema }),
+});
+export const HostsRemoveResponseSchema = z.object({
+  type: z.literal("hosts.remove.response"),
+  payload: z.object({ requestId: z.string() }),
+});
+export const HostsRouteOpenResponseSchema = z.object({
+  type: z.literal("hosts.route.open.response"),
+  payload: z.object({ requestId: z.string(), routeId: z.string() }),
+});
+export const HostsRouteFrameSchema = z.object({
+  type: z.literal("hosts.route.frame"),
+  payload: z.object({ subscriptionId: z.string(), data: z.string(), binary: z.literal(false) }),
+});
+export const HostsRouteClosedSchema = z.object({
+  type: z.literal("hosts.route.closed"),
+  payload: z.object({ subscriptionId: z.string(), code: z.string(), message: z.string() }),
+});
+
+// Logical B sessions are admitted under the authenticated physical link, never hello auth.
+export const HostsLinkOpenRequestSchema = z.object({
+  type: z.literal("hosts.link.open.request"),
+  requestId: z.string(),
+  channelId: z.string().min(1).max(256),
+});
+export const HostsLinkOpenResponseSchema = z.object({
+  type: z.literal("hosts.link.open.response"),
+  payload: z.object({ requestId: z.string(), routeId: z.string() }),
+});

packages/protocol/src/messages.ts

@@ -1,4 +1,23 @@
 import { AgentMessageSchema } from "./agent-message.js";
+import {
+  HostsListRequestSchema,
+  HostsAddRequestSchema,
+  HostsUpdateRequestSchema,
+  HostsRemoveRequestSchema,
+  HostsLinkOpenRequestSchema,
+  HostsRouteOpenRequestSchema,
+  HostsRouteFrameRequestSchema,
+  HostsRouteCloseRequestSchema,
+  HostsListResponseSchema,
+  HostsChangedSchema,
+  HostsAddResponseSchema,
+  HostsUpdateResponseSchema,
+  HostsRemoveResponseSchema,
+  HostsLinkOpenResponseSchema,
+  HostsRouteOpenResponseSchema,
+  HostsRouteFrameSchema,
+  HostsRouteClosedSchema,
+} from "./hosts.js";
 import { PluginRegistryIdentitySchema } from "./plugin-registry.js";
 import { AgentProfileSchema, AgentSkillSelectionSchema } from "./agent-profile.js";
 export {
@@ -3202,6 +3221,15 @@ export const SubscriptionReleaseResponseSchema = z.object({
 });
 
 export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
+  HostsListRequestSchema,
+  HostsAddRequestSchema,
+  HostsUpdateRequestSchema,
+  HostsRemoveRequestSchema,
+  HostsLinkOpenRequestSchema,
+  HostsRouteOpenRequestSchema,
+  HostsRouteFrameRequestSchema,
+  HostsRouteCloseRequestSchema,
+
   BrowserHostRegisterRequestSchema,
   SubscriptionReleaseRequestSchema,
   SessionEventsSetSubscriptionRequestSchema,
@@ -3571,6 +3599,8 @@ export const ServerInfoStatusPayloadSchema = z
     serverId: z.string().trim().min(1),
     hostname: ServerInfoHostnameSchema.optional(),
     version: ServerInfoVersionSchema.optional(),
+    // Authenticated producer for host registration ownership, separate from admission authority.
+    sourceId: z.string().optional(),
     // COMPAT(sessionPermissions): optional while clients support older daemons.
     permissions: z.array(DaemonPermissionSchema).optional(),
     // COMPAT(desktopManaged): added in v0.1.X, remove optional parsing after 2027-01-16.
@@ -3592,6 +3622,8 @@ export const ServerInfoStatusPayloadSchema = z
         internalWorkspaces: z.boolean().optional(),
         backgroundWorkspaces: z.boolean().optional(),
         providersSnapshot: z.boolean().optional(),
+        connectedHosts: z.boolean().optional(),
+        hostMultiplexing: z.boolean().optional(),
         usageSources: z.boolean().optional(),
         // COMPAT(providersSnapshotCwd): added in v0.3.2, remove gate after 2027-02-10.
         providersSnapshotCwd: z.boolean().optional(),
@@ -6871,6 +6903,16 @@ export const AgentSkillsImportLegacySelectionResponseSchema = z.object({
 });
 
 export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
+  HostsListResponseSchema,
+  HostsChangedSchema,
+  HostsAddResponseSchema,
+  HostsUpdateResponseSchema,
+  HostsRemoveResponseSchema,
+  HostsLinkOpenResponseSchema,
+  HostsRouteOpenResponseSchema,
+  HostsRouteFrameSchema,
+  HostsRouteClosedSchema,
+
   BrowserHostRegisterResponseSchema,
   SubscriptionReleaseResponseSchema,
   SessionEventsSetSubscriptionResponseSchema,
@@ -7563,6 +7605,11 @@ export const WSPongMessageSchema = z.object({
 
 export const WSHelloMessageSchema = z.object({
   type: z.literal("hello"),
+  sessionPermissions: z.array(DaemonPermissionSchema).optional(),
+  routed: z.boolean().optional(),
+  producer: z
+    .object({ serverId: z.string().min(1).max(256), sourceId: z.string().min(1).max(1024) })
+    .optional(),
   clientId: z.string().min(1),
   clientType: z.enum(["mobile", "browser", "cli", "mcp", "hub"]),
   protocolVersion: z.number().int(),

packages/server/package.json

@@ -89,7 +89,7 @@
     "speech:transcribe:local": "tsx scripts/transcribe-local-wav.ts",
     "test": "npm run test:unit && npm run test:integration",
     "test:unit": "vitest run --fileParallelism --exclude \"**/*.e2e.test.ts\"",
-    "test:integration": "vitest run --maxWorkers=1 src/server/git-filename-diffs.e2e.test.ts src/server/daemon-e2e/checkout-diff-content.e2e.test.ts src/server/daemon-e2e/filesystem.e2e.test.ts src/server/daemon-e2e/file-download.e2e.test.ts src/server/daemon-e2e/cli-logs-follow.e2e.test.ts src/server/creation/creation.e2e.test.ts src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/workspace-create-worktree-source.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/daemon-e2e/oversized-message.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts && vitest run --maxWorkers=1 src/server/daemon-e2e/git-operations.e2e.test.ts src/server/session.create-agent-worktree-autoarchive.e2e.test.ts src/server/agent/providers/opencode-bridge.local.e2e.test.ts src/server/daemon-e2e/relay-transport.e2e.test.ts -t \"merges into|creates a worktree and auto-archives both|versioned runtime|oversized attachments\"",
+    "test:integration": "vitest run --maxWorkers=1 src/server/git-filename-diffs.e2e.test.ts src/server/daemon-e2e/checkout-diff-content.e2e.test.ts src/server/daemon-e2e/filesystem.e2e.test.ts src/server/daemon-e2e/file-download.e2e.test.ts src/server/daemon-e2e/cli-logs-follow.e2e.test.ts src/server/creation/creation.e2e.test.ts src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/workspace-create-worktree-source.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/daemon-e2e/oversized-message.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts src/server/connected-hosts/hosts.e2e.test.ts && vitest run --maxWorkers=1 src/server/daemon-e2e/git-operations.e2e.test.ts src/server/session.create-agent-worktree-autoarchive.e2e.test.ts src/server/agent/providers/opencode-bridge.local.e2e.test.ts src/server/daemon-e2e/relay-transport.e2e.test.ts -t \"merges into|creates a worktree and auto-archives both|versioned runtime|oversized attachments|connected hosts\"",
     "test:hub-cli-contract": "vitest run --maxWorkers=1 src/server/hub/hub-cli-contract.test.ts",
     "test:integration:all": "npm run test:e2e",
     "test:integration:real": "vitest run real.e2e.test.ts",

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

@@ -6,6 +6,14 @@ 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.link.open.request": null,
+  "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 +224,15 @@ 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.link.open.response": null,
+  "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/index.ts

@@ -0,0 +1,539 @@
+export { DaemonLinkEndpoint, VirtualHostSocket } from "./link.js";
+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,
+  createHostChannelTransport,
+  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 {
+  client: DaemonClient;
+  accepted: boolean;
+  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);
+      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.",
+        );
+      }
+      if (this.rows.length >= 128)
+        throw new HostError("resource_limit", "Remove a host before adding another (limit 128)");
+      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 stopped = false;
+    const client = createHostClient(
+      row.connection,
+      {
+        clientId: `host-link-${this.self.serverId}-${row.registrationId}`,
+        connectTimeoutMs: 10000,
+      },
+      this.budget,
+    );
+    const live: Live = {
+      client,
+      accepted: false,
+      stop: () => {
+        stopped = true;
+        void client.close();
+      },
+    };
+    this.live.set(row.registrationId, live);
+    const accept = async () => {
+      try {
+        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 || info.features?.hostMultiplexing !== 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)));
+          if (stopped || client.getLastServerInfoMessage() !== info || !client.isConnected) return;
+          Object.assign(row, next);
+          live.accepted = true;
+          this.changed();
+        });
+      } catch (error) {
+        if (stopped) return;
+        row.status = "error";
+        row.error = connectionFailure(error, client.authFailureReason !== null) ?? {
+          code: "host_offline",
+          message: "Host connection failed",
+        };
+        row.nextRetryAt = null;
+        this.changed();
+        await client.close();
+      }
+    };
+    client.subscribeConnectionStatus((state) => {
+      if (stopped || row.status === "error") return;
+      live.accepted = false;
+      if (state.status === "connected") {
+        void accept();
+        return;
+      }
+      for (const route of this.routes)
+        if (route.registrationId === row.registrationId) route.close();
+      row.status = state.status === "connecting" ? "connecting" : "offline";
+      row.nextRetryAt = state.status === "disconnected" ? (state.nextRetryAt ?? null) : null;
+      row.error =
+        state.status === "disconnected"
+          ? (connectionFailure(
+              Object.assign(new Error("Check the SSH configuration and daemon credentials"), {
+                code: state.failureCode,
+              }),
+              client.authFailureReason !== null,
+            ) ?? {
+              code: "host_offline",
+              message: "Cannot reach host. Check the connection; retrying automatically.",
+            })
+          : null;
+      if (row.error?.code === "authentication_failed") {
+        row.status = "error";
+        row.nextRetryAt = null;
+        client.setReconnectEnabled(false);
+        void client.close();
+      }
+      this.changed();
+    });
+    void client.connect().catch(() => {});
+  }
+  async openRoute(input: {
+    serverId: string;
+    scopeId: string;
+    sourceId: string;
+    originSessionId: string;
+    permissions: readonly DaemonPermission[];
+    signal: AbortSignal;
+    frame(data: string | Uint8Array): void;
+    ended(code: string, message: string): void;
+  }): Promise<{ send(data: string | Uint8Array): 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 live = this.live.get(row.registrationId);
+    const client = live?.client;
+    if (!live?.accepted || !client?.isConnected)
+      throw new HostError("host_offline", "Host is offline", input.serverId);
+    const transport = createHostChannelTransport(
+      client,
+      input.serverId,
+      randomUUID(),
+      null,
+      true,
+    )({ url: "ws://host-link/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 ? (data as Uint8Array) : String(data));
+    });
+    await ready;
+    if (closed) throw new HostError("host_offline", "Host closed during connection");
+    return {
+      close: () => route.close(),
+      send: (data) => {
+        const binary = typeof data !== "string";
+        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 string) as WSHelloMessage;
+          if (hello.type !== "hello") throw new HostError("invalid_message", "Expected hello");
+          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: undefined,
+            }),
+          );
+          helloSent = true;
+          return;
+        }
+        if (!verified) throw new HostError("host_offline", "Host identity has not been verified");
+        try {
+          transport.send(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/link.ts

@@ -0,0 +1,210 @@
+import type { SessionDelivery } from "../session/owned-subscriptions/index.js";
+import { EventEmitter } from "node:events";
+import { createHash } from "node:crypto";
+import { encodeHostRouteFrame, type HostRouteFrame } from "@getpaseo/protocol/binary-frames/index";
+import type {
+  DaemonPermission,
+  SessionInboundMessage,
+  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<void>,
+    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<void> {
+    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<string, { scopeId: string; socket: VirtualHostSocket }>();
+  private closed = false;
+  constructor(
+    private readonly parent: {
+      identity: string;
+      permissions(): readonly DaemonPermission[];
+      routed(): boolean;
+      bufferedAmount(): number | null;
+      delivery: SessionDelivery;
+      admit(socket: VirtualHostSocket, permissions: readonly DaemonPermission[]): Promise<void>;
+    },
+  ) {}
+  handles(message: SessionInboundMessage): boolean {
+    return (
+      message.type === "hosts.link.open.request" ||
+      (message.type === "hosts.route.frame.request" && this.channels.has(message.routeId)) ||
+      (message.type === "hosts.route.close.request" &&
+        ((!!message.routeId && this.channels.has(message.routeId)) ||
+          [...this.channels.values()].some((channel) => channel.scopeId === message.scopeId)))
+    );
+  }
+  async dispatch(message: SessionInboundMessage): Promise<void> {
+    if (message.type === "hosts.link.open.request") {
+      const error = (code: string, text: string) =>
+        this.parent.delivery.reply({
+          type: "rpc_error",
+          payload: {
+            requestId: message.requestId,
+            requestType: message.type,
+            code,
+            error: text,
+          },
+        });
+      if (this.closed || this.parent.routed()) {
+        error("unsupported_capability", "Connected hosts support one hop");
+        return;
+      }
+      if (this.channels.size >= 128) {
+        error("resource_limit", "Too many host channels");
+        return;
+      }
+      let socket: VirtualHostSocket;
+      const owner = this.parent.delivery.begin("host-link", undefined, () => {
+        this.channels.delete(owner.id);
+        socket?.close();
+      });
+      const id = owner.id;
+      // Capture actual authenticated authority. The caller hello can only reduce it.
+      const ceiling = [...this.parent.permissions()];
+      socket = new VirtualHostSocket(
+        async (data) => {
+          const output: SessionOutboundMessage | null =
+            typeof data === "string"
+              ? {
+                  type: "hosts.route.frame",
+                  payload: { subscriptionId: id, data, binary: false },
+                }
+              : null;
+          const frame = typeof data === "string" ? null : encodeHostRouteFrame(id, data);
+          const bytes = output
+            ? Buffer.byteLength(JSON.stringify({ type: "session", message: output }))
+            : frame!.byteLength;
+          const buffered = this.parent.bufferedAmount();
+          // Account for the exact escaped outer frame, retaining control-frame headroom.
+          if (buffered === null || buffered + bytes > 64 * 1024 * 1024 - 65536) {
+            socket.close(1009, "Host channel output limit reached");
+            throw new Error("Host channel output limit reached");
+          }
+          try {
+            if (output) await owner.emitAndWait(output);
+            else await owner.emitBinaryAndWait(frame!);
+          } catch (failure) {
+            socket.close();
+            throw failure;
+          }
+        },
+        (code, reason) => {
+          this.channels.delete(id);
+          let failure = "host_offline";
+          if (code === 1009) failure = "resource_limit";
+          if (code === 1008) failure = "permission_denied";
+          owner.emit({
+            type: "hosts.route.closed",
+            payload: {
+              subscriptionId: id,
+              code: failure,
+              message: reason,
+            },
+          });
+          void owner.release();
+        },
+        () => this.parent.bufferedAmount() ?? 0,
+        (data) => {
+          const hello = JSON.parse(data);
+          if (hello.type !== "hello" || typeof hello.clientId !== "string")
+            throw new Error("Expected hello");
+          return JSON.stringify({
+            ...hello,
+            auth: undefined,
+            routed: true,
+            clientId: `link-${createHash("sha256")
+              .update(JSON.stringify([this.parent.identity, hello.clientId]))
+              .digest("hex")}`,
+          });
+        },
+      );
+      this.channels.set(id, { scopeId: message.channelId, socket });
+      try {
+        await this.parent.admit(socket, ceiling);
+        if (this.closed || socket.readyState !== 1) {
+          socket.close();
+          return;
+        }
+        this.parent.delivery.reply({
+          type: "hosts.link.open.response",
+          payload: { requestId: message.requestId, routeId: id },
+        });
+      } catch {
+        socket.close();
+        error("host_offline", "Could not admit host channel");
+      }
+      return;
+    }
+    if (message.type === "hosts.route.frame.request" && this.channels.has(message.routeId)) {
+      this.channels.get(message.routeId)!.socket.feed(message.data);
+      return;
+    }
+    if (message.type === "hosts.route.close.request") {
+      for (const [id, channel] of this.channels) {
+        if (message.routeId === id || (message.scopeId && message.scopeId === channel.scopeId))
+          channel.socket.close();
+      }
+    }
+    return;
+  }
+  binary(frame: HostRouteFrame): boolean {
+    const socket = this.channels.get(frame.routeId)?.socket;
+    if (!socket) return false;
+    socket.feed(frame.payload);
+    return true;
+  }
+  close(reason: "host_offline" | "permission_denied" = "host_offline"): void {
+    if (this.closed) return;
+    this.closed = true;
+    for (const { socket } of this.channels.values())
+      socket.close(
+        reason === "permission_denied" ? 1008 : 1000,
+        "Physical link authority or connection changed",
+      );
+    this.channels.clear();
+  }
+}

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

@@ -0,0 +1,252 @@
+import { encodeHostRouteFrame, type HostRouteFrame } from "@getpaseo/protocol/binary-frames/index";
+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 | Uint8Array): 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,
+  ) {}
+  binary(frame: HostRouteFrame, source: object): void {
+    const route = this.routes.get(frame.routeId);
+    if (!route || route.source !== source) return;
+    try {
+      route.send(frame.payload);
+    } catch {
+      route.fail("resource_limit", "Host input buffer limit reached");
+    }
+  }
+  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);
+        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) => {
+          const output: SessionOutboundMessage | null =
+            typeof data === "string"
+              ? {
+                  type: "hosts.route.frame",
+                  payload: { subscriptionId: owner.id, data, binary: false },
+                }
+              : null;
+          const binary = typeof data === "string" ? null : encodeHostRouteFrame(owner.id, data);
+          const size = output
+            ? Buffer.byteLength(JSON.stringify({ type: "session", message: output }))
+            : binary!.byteLength;
+          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;
+          }
+          if (output) owner.emit(output);
+          else owner.emitBinary(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/daemon-e2e/relay-transport.e2e.test.ts

@@ -1,3 +1,14 @@
+import { createPaseoApi, type PaseoApi } from "@getpaseo/client";
+import { createHostClient } from "@getpaseo/client/node";
+import { hashSync } from "bcryptjs";
+import { once } from "node:events";
+import {
+  decodeTerminalStreamFrame,
+  encodeTerminalStreamFrame,
+  TerminalStreamOpcode,
+} from "@getpaseo/protocol/binary-frames/index";
+import type { DaemonClient as HostClient } from "@getpaseo/client/internal/daemon-client";
+import type { SessionOutboundMessage } from "@getpaseo/protocol/messages";
 import { describe, expect, test } from "vitest";
 import { withTimeout } from "../../utils/promise-timeout.js";
 import { DaemonClient } from "../test-utils/daemon-client.js";
@@ -194,9 +205,9 @@ async function waitForCapturedLog(
   let relayProcess: ChildProcess | null = null;
   let relayStdoutLines: string[] = [];
 
-  const startRelay = async (options: { useLocalRelay?: boolean } = {}) => {
+  const startRelay = async (options: { useLocalRelay?: boolean; port?: number } = {}) => {
     relayStdoutLines = [];
-    relayPort = await getAvailablePort();
+    relayPort = options.port ?? (await getAvailablePort());
     const relayDir = path.resolve(process.cwd(), "../relay");
     const relayArgs = [
       "wrangler",
@@ -255,10 +266,152 @@ async function waitForCapturedLog(
 
   const stopRelay = async () => {
     if (!relayProcess) return;
-    relayProcess.kill("SIGTERM");
+    const process = relayProcess;
     relayProcess = null;
+    const exited = once(process, "exit");
+    process.kill("SIGTERM");
+    await withTimeout(exited, 10000, "Local relay did not stop");
   };
 
+  test("connected hosts route RPC and binary traffic through local E2EE relay and recover scopes", async () => {
+    let a: Awaited<ReturnType<typeof createTestPaseoDaemon>> | undefined;
+    let b: Awaited<ReturnType<typeof createTestPaseoDaemon>> | undefined;
+    let origin: ReturnType<typeof createHostClient> | undefined;
+    let api: PaseoApi | undefined;
+    let remote: PaseoApi | undefined;
+    let sibling: PaseoApi | undefined;
+    let terminalRoute: Awaited<ReturnType<typeof openConnectedHostTerminal>> | undefined;
+    let wire: Awaited<ReturnType<typeof relayWire>> | undefined;
+    try {
+      await startRelay({ useLocalRelay: true });
+      const port = relayPort;
+      b = await createTestPaseoDaemon({
+        relayEnabled: true,
+        relayEndpoint: `127.0.0.1:${port}`,
+        relayUseTls: false,
+        relayPublicUseTls: false,
+        auth: { password: hashSync("relay-password", 4) },
+      });
+      a = await createTestPaseoDaemon();
+      origin = createHostClient(`127.0.0.1:${a.port}`, { clientId: crypto.randomUUID() });
+      await origin.connect();
+      api = createPaseoApi(origin);
+      const offerUrl = await getPairingOfferUrl({
+        paseoHome: b.paseoHome,
+        relayEnabled: true,
+        relayEndpoint: b.config.relayEndpoint,
+      });
+      const { serverId, daemonPublicKeyB64 } = decodeOfferFromFragmentUrl(offerUrl);
+      wire = await relayWire(port);
+      const connection = new URL(`relay://127.0.0.1:${wire.port}/${serverId}`);
+      connection.searchParams.set("key", daemonPublicKeyB64);
+      connection.searchParams.set("password", "wrong-password");
+      const saved = await api.hosts.add({ connection: connection.toString() });
+      await expect
+        .poll(async () => (await api!.hosts.list()).hosts[0]?.error?.code, { timeout: 20000 })
+        .toBe("authentication_failed");
+      expect(JSON.stringify(await api.hosts.list())).not.toMatch(/wrong-password|relay-password/);
+      connection.searchParams.set("password", "relay-password");
+      await api.hosts.update(saved.registrationId, { connection: connection.toString() });
+      await expect
+        .poll(async () => (await api!.hosts.list()).hosts[0]?.status, { timeout: 20000 })
+        .toBe("online");
+      const admittedConnections = wire.connections();
+      expect(wire.active()).toBe(1);
+      remote = await api.hosts.client(serverId);
+      sibling = await api.hosts.client(serverId);
+      expect(wire.connections()).toBe(admittedConnections);
+      expect(wire.active()).toBe(1);
+      expect((await remote.workspaces.list()).entries).toEqual([]);
+      const observed = await remote.workspaces.list({ subscribe: {} });
+      let snapshots = 0;
+      observed.subscription.subscribe({
+        snapshot: () => {
+          snapshots++;
+        },
+        update() {},
+      });
+      const workspace = await remote.workspaces.create({
+        source: { kind: "directory", path: b.staticDir },
+        title: "Relayed workspace",
+      });
+      const terminal = await workspace.terminals.create({
+        command: process.execPath,
+        args: ["-e", "process.stdin.on('data', data => process.stdout.write(data))"],
+      });
+      terminalRoute = await openConnectedHostTerminal(origin, serverId, terminal.id);
+      expect(wire.connections()).toBe(admittedConnections);
+      terminalRoute.write("ENCRYPTED_BINARY_BEFORE\n");
+      await expect.poll(() => terminalRoute!.output()).toContain("ENCRYPTED_BINARY_BEFORE");
+      terminalRoute.close();
+      terminalRoute = undefined;
+      const before = snapshots;
+      await stopRelay();
+      await expect
+        .poll(
+          async () => {
+            try {
+              await remote!.workspaces.list();
+              return "online";
+            } catch (error) {
+              return (error as { code: string }).code;
+            }
+          },
+          { timeout: 15000 },
+        )
+        .toBe("DAEMON_CONNECTION_LOST");
+      await startRelay({ useLocalRelay: true, port });
+      await expect
+        .poll(
+          async () => {
+            try {
+              return (await remote!.workspaces.list()).entries.length;
+            } catch {
+              return -1;
+            }
+          },
+          { timeout: 30000 },
+        )
+        .toBe(1);
+      await expect.poll(() => snapshots, { timeout: 15000 }).toBeGreaterThan(before);
+      expect(wire.active()).toBe(1);
+      terminalRoute = await openConnectedHostTerminal(origin, serverId, terminal.id);
+      terminalRoute.write("ENCRYPTED_BINARY_AFTER\n");
+      await expect.poll(() => terminalRoute!.output()).toContain("ENCRYPTED_BINARY_AFTER");
+      terminalRoute.close();
+      terminalRoute = undefined;
+      await remote.dispose();
+      remote = undefined;
+      expect((await sibling.workspaces.list()).entries).toHaveLength(1);
+      await sibling.terminals.ref(terminal.id).kill();
+      await api.dispose();
+      await expect(sibling.workspaces.list()).rejects.toThrow();
+      const direct = createHostClient(`127.0.0.1:${b.port}`, {
+        clientId: crypto.randomUUID(),
+        password: "relay-password",
+        reconnect: { enabled: false },
+      });
+      try {
+        await direct.connect();
+        const directApi = createPaseoApi(direct);
+        expect((await directApi.workspaces.list()).entries).toHaveLength(1);
+        await directApi.dispose();
+      } finally {
+        await direct.close();
+      }
+    } finally {
+      terminalRoute?.close();
+      await remote?.dispose();
+      await sibling?.dispose();
+      await api?.dispose();
+      await origin?.close();
+      await a?.close();
+      await b?.close();
+      await wire?.close();
+      await stopRelay();
+    }
+  }, 120000);
+
   test("oversized attachments fail without disconnecting or resubmitting over the relay", async () => {
     await startRelay({ useLocalRelay: true });
     const daemon = await createTestPaseoDaemon({
@@ -932,3 +1085,112 @@ async function waitForCapturedLog(
     }
   }, 90000);
 });
+
+async function openConnectedHostTerminal(origin: HostClient, serverId: string, terminalId: string) {
+  const routeId = await origin.openHostRoute(serverId, crypto.randomUUID());
+  const messages: SessionOutboundMessage[] = [];
+  let output = "";
+  const detachBinary = origin.subscribeHostFrames((routed) => {
+    if (routed.routeId !== routeId) return;
+    const frame = decodeTerminalStreamFrame(routed.payload);
+    if (frame?.opcode === TerminalStreamOpcode.Output)
+      output += new TextDecoder().decode(frame.payload);
+  });
+  const detach = origin.subscribeRawMessages((message) => {
+    if (message.type !== "hosts.route.frame" || message.payload.subscriptionId !== routeId) return;
+    const frame = WSOutboundMessageSchema.parse(JSON.parse(message.payload.data));
+    if (frame.type === "session") messages.push(frame.message);
+  });
+  const close = () => {
+    detach();
+    detachBinary();
+    origin.closeHostRoute(routeId);
+  };
+  try {
+    origin.sendHostFrame(
+      routeId,
+      JSON.stringify({
+        type: "hello",
+        clientId: "relay-binary",
+        clientType: "cli",
+        protocolVersion: 1,
+      }),
+    );
+    await expect
+      .poll(() =>
+        messages.some(
+          (message) => message.type === "status" && message.payload.status === "server_info",
+        ),
+      )
+      .toBe(true);
+    origin.sendHostFrame(
+      routeId,
+      JSON.stringify({
+        type: "session",
+        message: { type: "subscribe_terminal_request", terminalId, requestId: "binary" },
+      }),
+    );
+    await expect
+      .poll(() => messages.some((message) => message.type === "subscribe_terminal_response"))
+      .toBe(true);
+    const response = messages.find((message) => message.type === "subscribe_terminal_response");
+    if (response?.type !== "subscribe_terminal_response" || response.payload.slot == null)
+      throw new Error("Missing binary stream slot");
+    const slot = response.payload.slot;
+    return {
+      close,
+      output: () => output,
+      write: (payload: string) =>
+        origin.sendHostFrame(
+          routeId,
+          encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Input, slot, payload }),
+        ),
+    };
+  } catch (error) {
+    close();
+    throw error;
+  }
+}
+
+async function relayWire(relayPort: number) {
+  const sockets = new Set<net.Socket>();
+  let connections = 0;
+  let active = 0;
+  const server = net.createServer((source) => {
+    const target = net.connect(relayPort, "127.0.0.1");
+    sockets.add(source);
+    sockets.add(target);
+    let counted = false;
+    target.on("connect", () => {
+      counted = true;
+      connections++;
+      active++;
+    });
+    target.on("error", () => source.destroy());
+    source.on("error", () => target.destroy());
+    target.on("close", () => {
+      if (counted) active--;
+      sockets.delete(target);
+      source.destroy();
+    });
+    source.on("close", () => {
+      sockets.delete(source);
+      target.destroy();
+    });
+    source.pipe(target);
+    target.pipe(source);
+  });
+  server.listen(0, "127.0.0.1");
+  await once(server, "listening");
+  const address = server.address();
+  if (!address || typeof address === "string") throw new Error("Missing relay wire port");
+  return {
+    port: address.port,
+    connections: () => connections,
+    active: () => active,
+    close: async () => {
+      for (const socket of sockets) socket.destroy();
+      await new Promise<void>((resolve) => server.close(() => resolve()));
+    },
+  };
+}

packages/server/src/server/session.test.ts

@@ -288,6 +288,7 @@ vi.mock("./worktree-bootstrap.js", async (importOriginal) => {
 });
 
 interface SessionForTestOptions {
+  sendToSource?: SessionOptions["onMessageToSource"];
   clientId?: string;
   permissions?: readonly DaemonPermission[];
   agentManager?: { [K in keyof SessionOptions["agentManager"]]?: unknown };
@@ -441,6 +442,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session {
     daemonRuntimeConfig: options.daemonRuntimeConfig,
     permissions: options.permissions ?? OWNER_PERMISSIONS,
   };
+  if (options.sendToSource) sessionOptions.onMessageToSource = options.sendToSource;
   return new Session(sessionOptions);
 }
 
@@ -5814,3 +5816,40 @@ test("provider snapshots preserve versionless visibility while capabilities upda
   ]);
   expect(references.compactSnapshot!.entries[0]!.modes![0]!.icon).toBe("ShieldCheck");
 });
+
+test("a failing direct source observes fire-and-forget file replies without unhandled rejections", async () => {
+  const root = mkdtempSync(join(tmpdir(), "session-send-failure-"));
+  writeFileSync(join(root, "readme.txt"), "source reply");
+  let attempts = 0;
+  const unhandled: unknown[] = [];
+  const collect = (error: unknown) => unhandled.push(error);
+  process.on("unhandledRejection", collect);
+  const session = createSessionForTest({
+    sendToSource: async () => {
+      attempts++;
+      throw new Error("Physical source closed during send");
+    },
+  });
+  try {
+    const source = {};
+    session.updateClientCapabilities(null, source);
+    await session.handleMessage(
+      {
+        type: "file_explorer_request",
+        requestId: "direct-file",
+        cwd: root,
+        path: "readme.txt",
+        mode: "file",
+        acceptBinary: false,
+      },
+      source,
+    );
+    await new Promise<void>((resolve) => setTimeout(resolve, 30));
+    expect(attempts).toBeGreaterThan(0);
+    expect(unhandled).toEqual([]);
+  } finally {
+    process.off("unhandledRejection", collect);
+    await session.cleanup();
+    rmSync(root, { recursive: true, force: true });
+  }
+});

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";
@@ -442,16 +444,20 @@ const nodeSessionFileSystem: SessionFileSystem = {
 type AgentMcpTransportFactory = () => Promise<unknown>;
 
 export interface SessionOptions {
+  connectedHosts?: ConnectedHosts;
+  sourceId?: string;
+  routed?: boolean;
   browserToolsBroker?: BrowserToolsBroker | null;
   clientId: string;
   permissions: readonly DaemonPermission[];
   appVersion?: string | null;
   clientCapabilities?: Record<string, unknown> | null;
   onMessage: (msg: SessionOutboundMessage) => void;
-  onMessageToSource?: (source: object, msg: SessionOutboundMessage) => void;
+  onMessageToSource?: (source: object, msg: SessionOutboundMessage) => void | Promise<void>;
   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;
@@ -674,17 +680,31 @@ interface ClientActivity {
 }
 
 export class Session {
+  private readonly connectedHostSession: ConnectedHostSession | null;
+  private routed = false;
+  isRouted(): boolean {
+    return this.routed;
+  }
+  getHostSourceId(): string {
+    return this.hostSourceId;
+  }
+  private hostSourceId: string;
+  setHostSource(sourceId: string): void {
+    this.hostSourceId = sourceId;
+    this.connectedHostSession?.setSource(sourceId);
+  }
+  setRouted(routed: boolean): void {
+    this.routed = routed;
+  }
   readonly delivery = new SessionDelivery(
     (source, message) => {
       if (!this.authorization.allowsOutbound(message)) return;
-      if (this.onMessageToSource) this.onMessageToSource(source, message);
-      else this.onMessage(message);
+      if (this.onMessageToSource) return this.onMessageToSource(source, message);
+      this.onMessage(message);
     },
     (source, frame) => {
       if (this.onBinaryMessageToSource) {
-        void this.onBinaryMessageToSource(source, frame).catch((error) =>
-          this.sessionLogger.warn({ err: error }, "Failed to emit binary frame"),
-        );
+        return this.onBinaryMessageToSource(source, frame);
       } else this.emitBinary(frame);
     },
     (source, message) => this.workspaceSetupMessageForClient(message, source),
@@ -706,7 +726,7 @@ export class Session {
   private readonly sessionId: string;
   private readonly onMessage: (msg: SessionOutboundMessage) => void;
   private readonly onMessageToSource:
-    | ((source: object, msg: SessionOutboundMessage) => void)
+    | ((source: object, msg: SessionOutboundMessage) => void | Promise<void>)
     | null;
   private readonly onBinaryMessage: ((frame: Uint8Array) => void) | null;
   private readonly onBinaryMessageToSource:
@@ -864,13 +884,20 @@ 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.hostSourceId = hostSource(options);
+    this.connectedHostSession = createHostSession(options, this);
     this.onMessage = onMessage;
-    this.onMessageToSource = onMessageToSource ?? null;
+    this.onMessageToSource = onMessageToSource
+      ? (source, message) => this.observeSourceSend(() => onMessageToSource(source, message))
+      : null;
     this.onBinaryMessage = onBinaryMessage ?? null;
-    this.onBinaryMessageToSource = onBinaryMessageToSource ?? null;
+    this.onBinaryMessageToSource = onBinaryMessageToSource
+      ? (source, frame) => this.observeSourceSend(() => onBinaryMessageToSource(source, frame))
+      : null;
     this.getTransportBufferedAmount = getTransportBufferedAmount ?? (() => 0);
     this.onLifecycleIntent = onLifecycleIntent ?? null;
     this.onWorkspaceRecovered = onWorkspaceRecovered ?? null;
@@ -2173,7 +2200,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",
             },
           });
         }
@@ -2220,6 +2247,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")
@@ -2374,11 +2402,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) ??
@@ -2396,6 +2423,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) ??
@@ -3177,6 +3215,10 @@ export class Session {
   }
 
   public async handleBinaryFrame(binaryFrame: BinaryFrame, source: object): Promise<void> {
+    if (binaryFrame.kind === "host_route") {
+      this.connectedHostSession?.binary(binaryFrame.frame, source);
+      return;
+    }
     if (!this.authorization.allowsPermission("workspace.write")) {
       return;
     }
@@ -8617,6 +8659,20 @@ export class Session {
     this.emitBinary(frame);
   }
 
+  private observeSourceSend(send: () => void | Promise<void>): Promise<void> {
+    let pending: Promise<void>;
+    try {
+      pending = Promise.resolve(send());
+    } catch (error) {
+      pending = Promise.reject(error);
+    }
+    // Legacy producers fire and forget; owned streams await the same failure.
+    void pending.catch((error) =>
+      this.sessionLogger.warn({ err: error }, "Failed to emit source frame"),
+    );
+    return pending;
+  }
+
   private emitForSource(msg: SessionOutboundMessage, source?: object): void {
     if (source && this.onMessageToSource) {
       msg = this.delivery.authorizeReply(msg, source);
@@ -8765,3 +8821,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/session/owned-subscriptions/index.test.ts

@@ -169,3 +169,40 @@ test("failed teardown still closes the delivery and rejects late work", async ()
     delivery.request(source, { type: "ping", requestId: "late" }, async () => {}),
   ).rejects.toThrow("Session delivery is closed");
 });
+
+test("owned send completion retains source proofs and reports physical failures to awaiting producers", async () => {
+  const source = {};
+  const failure = new Error("Physical send failed");
+  const delivery = new SessionDelivery(
+    async (socket, message) => {
+      expect(socket).toBe(source);
+      expect(delivery.permits(socket, message)).toBe(true);
+      throw failure;
+    },
+    async (socket, frame) => {
+      expect(socket).toBe(source);
+      expect(delivery.permitsBinary(socket, frame)).toBe(true);
+      throw failure;
+    },
+  );
+  delivery.attach(source, true);
+  await delivery.request(
+    source,
+    { type: "hosts.link.open.request", requestId: "open", channelId: "scope" },
+    async () => {
+      const owner = delivery.begin("host-link", undefined, () => {});
+      await expect(
+        owner.emitAndWait({
+          type: "hosts.route.frame",
+          payload: { subscriptionId: owner.id, data: "text", binary: false },
+        }),
+      ).rejects.toBe(failure);
+      await expect(owner.emitBinaryAndWait(new Uint8Array([0, 255]))).rejects.toBe(failure);
+      await owner.release();
+      await expect(owner.emitBinaryAndWait(new Uint8Array([0]))).rejects.toThrow(
+        "Subscription source is closed",
+      );
+    },
+  );
+  await delivery.close();
+});

packages/server/src/server/session/owned-subscriptions/index.ts

@@ -51,6 +51,8 @@ export interface OwnedSubscription {
   readonly signal: AbortSignal;
   emit(message: SessionOutboundMessage): void;
   emitBinary(frame: Uint8Array): void;
+  emitAndWait(message: SessionOutboundMessage): Promise<void>;
+  emitBinaryAndWait(frame: Uint8Array): Promise<void>;
   release(): Promise<void>;
 }
 
@@ -67,8 +69,14 @@ export class SessionDelivery {
   private closing?: Promise<void>;
 
   constructor(
-    private readonly send: (source: object, message: SessionOutboundMessage) => void,
-    private readonly sendBinary: (source: object, frame: Uint8Array) => void = () => {},
+    private readonly send: (
+      source: object,
+      message: SessionOutboundMessage,
+    ) => void | Promise<void>,
+    private readonly sendBinary: (
+      source: object,
+      frame: Uint8Array,
+    ) => void | Promise<void> = () => {},
     private readonly project: (
       source: object,
       message: SessionOutboundMessage,
@@ -319,7 +327,18 @@ export class SessionDelivery {
       emitBinary: (frame) => {
         if (!owner.active || !source.active) return;
         this.proofs.set(frame, owner);
-        this.sendBinary(source.socket, frame);
+        void Promise.resolve(this.sendBinary(source.socket, frame)).catch(() => {});
+      },
+      emitAndWait: async (message) => {
+        if (!owner.active || !source.active) throw new Error("Subscription source is closed");
+        const projected = this.project(source.socket, withSubscriptionId(message, responseId));
+        this.proofs.set(projected, owner);
+        await this.send(source.socket, projected);
+      },
+      emitBinaryAndWait: async (frame) => {
+        if (!owner.active || !source.active) throw new Error("Subscription source is closed");
+        this.proofs.set(frame, owner);
+        await this.sendBinary(source.socket, frame);
       },
       release: () => this.releaseOwner(owner),
     };
@@ -414,7 +433,7 @@ export class SessionDelivery {
   private sendOwned(owner: DeliveryOwner, message: SessionOutboundMessage): void {
     const projected = this.project(owner.source.socket, message);
     this.proofs.set(projected, owner);
-    this.send(owner.source.socket, projected);
+    void Promise.resolve(this.send(owner.source.socket, projected)).catch(() => {});
   }
 
   private releaseOwner(owner: RetainedOwner): Promise<void> {

packages/server/src/server/websocket-server.relay-reconnect.test.ts

@@ -71,6 +71,18 @@ const sessionMock = vi.hoisted(() => {
     wantsSourceEvent = (source: object) => !this.delivery.isModern(source);
     getSessionId = vi.fn(() => "mock-session-id");
     getPermissions = vi.fn(() => this.args.permissions as string[]);
+    setPermissions = (permissions: string[]) => {
+      this.args.permissions = permissions;
+    };
+    private routed = false;
+    setRouted = (routed: boolean) => {
+      this.routed = routed;
+    };
+    isRouted = () => this.routed;
+    setHostSource = (sourceId: string) => {
+      this.args.sourceId = sourceId;
+    };
+    getHostSourceId = () => this.args.sourceId as string;
     allowsInbound = vi.fn(() => true);
     allowsPermission = vi.fn(() => true);
     publish = vi.fn((message: unknown) => {
@@ -473,16 +485,12 @@ function holdNextSessionMessage(session: (typeof sessionMock.instances)[number])
 function holdSessionCleanup(session: (typeof sessionMock.instances)[number]): {
   finish: () => void;
 } {
-  let finish = () => {};
-  session.cleanup.mockImplementationOnce(
-    () =>
-      new Promise<void>((resolve) => {
-        finish = resolve;
-      }),
-  );
-  return {
-    finish: () => finish(),
-  };
+  let finish!: () => void;
+  const pending = new Promise<void>((resolve) => {
+    finish = resolve;
+  });
+  session.cleanup.mockImplementationOnce(() => pending);
+  return { finish };
 }
 
 describe("relay external socket reconnect behavior", () => {

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

@@ -1,3 +1,6 @@
+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 { stat } from "node:fs/promises";
 import type { CreationSnapshot } from "@getpaseo/protocol/messages";
 import { CreationService } from "./creation/index.js";
@@ -437,6 +440,8 @@ export interface WebSocketLike {
 }
 
 interface SessionConnectionBase {
+  hubExecutionAgents?: HubExecutionAgents;
+  permissionCeiling?: readonly DaemonPermission[];
   session: Session;
   principalId: string;
   sessionKey: string;
@@ -460,13 +465,14 @@ interface PluginSessionConnection extends SessionConnectionBase {
 type SessionConnection = ReconnectableSessionConnection | PluginSessionConnection;
 
 interface SocketSessionOptions {
+  sourceId?: string;
   clientId: string;
   appVersion: string | null;
   clientCapabilities: Record<string, unknown> | null;
   permissions: readonly DaemonPermission[];
   connectionLogger: pino.Logger;
   onMessage: (message: SessionOutboundMessage) => void;
-  onMessageToSource?: (source: object, message: SessionOutboundMessage) => void;
+  onMessageToSource?: (source: object, message: SessionOutboundMessage) => void | Promise<void>;
   onBinaryMessage?: (frame: Uint8Array) => void;
   onBinaryMessageToSource?: (source: object, frame: Uint8Array) => Promise<void>;
   getTransportBufferedAmount?: (source?: object) => number | null;
@@ -527,6 +533,7 @@ function requireWebSocketServices(params: {
 export class VoiceAssistantWebSocketServer {
   private readonly logger: pino.Logger;
   private readonly wss: WebSocketServer;
+  private readonly hostLinks = new Map<WebSocketLike, DaemonLinkEndpoint>();
   private readonly pendingConnections: Map<WebSocketLike, PendingConnection> = new Map();
   private readonly sessions: Map<WebSocketLike, SessionConnection> = new Map();
   private readonly socketIdentities: Map<WebSocketLike, WebSocketConnectionIdentity> = new Map();
@@ -550,6 +557,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 +709,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;
@@ -1047,6 +1056,12 @@ export class VoiceAssistantWebSocketServer {
     principalId: string,
     permissions: readonly DaemonPermission[],
   ): void {
+    for (const [socket, link] of this.hostLinks) {
+      if (this.sessions.get(socket)?.principalId === principalId) {
+        link.close("permission_denied");
+        this.hostLinks.delete(socket);
+      }
+    }
     for (const pending of this.pendingConnections.values()) {
       if (pending.admission?.principalId === principalId) {
         pending.admission = { ...pending.admission, permissions };
@@ -1054,7 +1069,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 +1090,7 @@ export class VoiceAssistantWebSocketServer {
 
   public async close(): Promise<void> {
     this.prepareForShutdown();
+    await this.connectedHosts.close();
     this.unsubscribeSpeechReadiness?.();
     this.unsubscribeSpeechReadiness = null;
     this.unsubscribeDaemonConfigChange?.();
@@ -1159,6 +1179,20 @@ export class VoiceAssistantWebSocketServer {
     message: WSOutboundMessage,
     reportRejection = false,
   ): void {
+    const prepared = this.prepareJsonDelivery(sockets, message, reportRejection);
+    if (!prepared) return;
+    for (const ws of prepared.sockets) {
+      this.sendFrameToClient(ws, prepared.payload, prepared.bytes, () => {
+        this.runtimeMetrics.recordOutboundMessage(message, ws.bufferedAmount);
+      });
+    }
+  }
+
+  private prepareJsonDelivery(
+    sockets: Iterable<WebSocketLike>,
+    message: WSOutboundMessage,
+    reportRejection: boolean,
+  ): { sockets: WebSocketLike[]; payload: string; bytes: number } | null {
     const writableSockets = [...sockets].filter((ws) => {
       if (
         message.type === "session" &&
@@ -1180,7 +1214,7 @@ export class VoiceAssistantWebSocketServer {
       return this.ensureOutboundCapacity(ws, 0);
     });
     if (writableSockets.length === 0) {
-      return;
+      return null;
     }
 
     let payload: string;
@@ -1188,14 +1222,30 @@ export class VoiceAssistantWebSocketServer {
       payload = JSON.stringify(message);
     } catch (err) {
       this.logger.warn({ err }, "ws_serialize_failed");
-      return;
+      return null;
     }
 
-    const payloadBytes = outboundFrameByteLength(payload);
-    for (const ws of writableSockets) {
-      this.sendFrameToClient(ws, payload, payloadBytes, () => {
-        this.runtimeMetrics.recordOutboundMessage(message, ws.bufferedAmount);
+    return { sockets: writableSockets, payload, bytes: outboundFrameByteLength(payload) };
+  }
+
+  private async sendMessageToClientAndWait(
+    ws: WebSocketLike,
+    message: WSOutboundMessage,
+  ): Promise<void> {
+    const prepared = this.prepareJsonDelivery([ws], message, true);
+    if (!prepared) throw new Error("Source message delivery unavailable");
+    try {
+      const sent = await sendBoundedPhysicalFrameAndWait({
+        socket: ws,
+        frame: prepared.payload,
+        frameBytes: prepared.bytes,
+        onHighWater: () => this.closeAtOutboundHighWater(ws),
       });
+      if (!sent) throw new Error("Source is closed");
+      this.runtimeMetrics.recordOutboundMessage(message, ws.bufferedAmount);
+    } catch (err) {
+      this.logger.warn({ err }, "ws_send_failed");
+      throw err;
     }
   }
 
@@ -1394,6 +1444,10 @@ export class VoiceAssistantWebSocketServer {
       appVersion,
       clientCapabilities,
       permissions: admission.permissions,
+      sourceId:
+        lifecycle.kind === "ephemeral-plugin"
+          ? `plugin:${lifecycle.pluginId}`
+          : admission.principalId,
       connectionLogger,
       onMessage: (msg) => {
         if (!connection) {
@@ -1401,11 +1455,11 @@ export class VoiceAssistantWebSocketServer {
         }
         this.sendToConnection(connection, wrapSessionMessage(msg));
       },
-      onMessageToSource: (source, msg) => {
+      onMessageToSource: async (source, msg) => {
         if (!connection || !connection.sockets.has(source as WebSocketLike)) {
           return;
         }
-        this.sendToClient(source as WebSocketLike, wrapSessionMessage(msg));
+        await this.sendMessageToClientAndWait(source as WebSocketLike, wrapSessionMessage(msg));
       },
       onBinaryMessage: (frame) => {
         if (!connection) {
@@ -1446,6 +1500,7 @@ export class VoiceAssistantWebSocketServer {
     const base: SessionConnectionBase = {
       session,
       principalId: admission.principalId,
+      hubExecutionAgents: admission.hubExecutionAgents,
       sessionKey: sessionConnectionKey(admission.principalId, clientId),
       clientId,
       appVersion,
@@ -1463,6 +1518,17 @@ 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 (socket instanceof VirtualHostSocket) continue;
+          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 +1682,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 +1714,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 +1832,14 @@ export class VoiceAssistantWebSocketServer {
       clearTimeout(existing.externalDisconnectCleanupTimeout);
       existing.externalDisconnectCleanupTimeout = null;
     }
+    for (const socket of existing.sockets) {
+      this.hostLinks.get(socket)?.close("permission_denied");
+      this.hostLinks.delete(socket);
+    }
+    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 +1870,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 +1892,14 @@ export class VoiceAssistantWebSocketServer {
       serverId: this.serverId,
       hostname: getHostName(),
       version: this.daemonVersion,
+      sourceId: session.getHostSourceId(),
       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,
+        hostMultiplexing: true,
         usageSources: true,
         ownedSubscriptions: true,
         agentRequestReceipts: true,
@@ -2038,6 +2141,9 @@ export class VoiceAssistantWebSocketServer {
       error?: Error;
     },
   ): Promise<void> {
+    this.hostLinks.get(ws)?.close();
+    this.hostLinks.delete(ws);
+
     this.applicationSocketLease.release(ws);
     const identity = this.socketIdentities.get(ws);
     const identityFields = identity ? toConnectionLogFields(identity) : {};
@@ -2086,6 +2192,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);
@@ -2228,6 +2338,12 @@ export class VoiceAssistantWebSocketServer {
       return false;
     }
     const decodedFrame = decodeBinaryFrame(asBytes);
+    if (asBytes[0] === HOST_ROUTE_OPCODE && !decodedFrame) {
+      ws.close(1008, "Invalid host routing frame");
+      return true;
+    }
+    if (decodedFrame?.kind === "host_route" && this.hostLinks.get(ws)?.binary(decodedFrame.frame))
+      return true;
     if (!decodedFrame) {
       return false;
     }
@@ -2382,11 +2498,38 @@ export class VoiceAssistantWebSocketServer {
     }
   }
 
+  private hostLink(ws: WebSocketLike, connection: SessionConnection): DaemonLinkEndpoint {
+    let link = this.hostLinks.get(ws);
+    if (!link) {
+      link = new DaemonLinkEndpoint({
+        identity: JSON.stringify([connection.principalId, connection.clientId]),
+        permissions: () => connection.session.getPermissions(),
+        routed: () => connection.session.isRouted(),
+        bufferedAmount: () => ws.bufferedAmount ?? null,
+        delivery: connection.session.delivery,
+        admit: (socket, permissions) =>
+          this.attachSocket(socket, undefined, undefined, false, {
+            principalId: connection.principalId,
+            permissions,
+          }),
+      });
+      this.hostLinks.set(ws, link);
+    }
+    return link;
+  }
+
   private async dispatchSessionMessage(
     ws: WebSocketLike,
     activeConnection: SessionConnection,
     message: Extract<WSInboundMessage, { type: "session" }>,
   ): Promise<void> {
+    const link = this.hostLink(ws, activeConnection);
+    if (link.handles(message.message)) {
+      await activeConnection.session.delivery.request(ws, message.message, () =>
+        link.dispatch(message.message),
+      );
+      return;
+    }
     this.recordInboundSessionRequestType(message.message.type);
     const controlRpc = getControlRpcLogInfo(message.message);
     if (controlRpc) {

public-docs/cli.md

@@ -64,6 +64,31 @@ Entries may also include `loginErrors`. Structured output preserves numeric valu
 
 A completed query exits with status `0`, including when individual reports are unavailable or failed. Connection or request failures, unsupported hosts, unknown report IDs, and conflicting selectors exit with status `1` and write the error to stderr. Human output says **No usage reports returned** for an empty result; this does not establish whether account discovery succeeded for every source.
 
+## Connected hosts
+
+`--host` selects the daemon whose saved connections you manage. Registrations use their
+`registrationId`, not their label.
+
+```sh
+paseo --host localhost:6767 host add 'ssh://deploy@build-box' --label build-box --meta team=infra
+paseo --host localhost:6767 host ls --json
+paseo --host localhost:6767 host inspect hst_123
+paseo --host localhost:6767 host update hst_123 --meta team=tools --remove-meta obsolete
+paseo --host localhost:6767 host update hst_123 --connection 'ssh://deploy@new-address'
+paseo --host localhost:6767 host disable hst_123
+paseo --host localhost:6767 host enable hst_123
+paseo --host localhost:6767 host rm hst_123
+```
+
+Add accepts TCP/host-and-port, `ws://`/`wss://`, local socket/pipe, SSH, and relay/pairing URLs.
+SSH connects to an already running daemon. Add `?password=...` to an SSH URI when that daemon requires a password (`&password=...` after `daemonPort`). Unreachable targets remain saved and retry automatically.
+`ls` includes disabled and failed hosts. `inspect` includes safe connection details and recovery guidance.
+JSON output never includes connection credentials.
+
+`update --meta` replaces supplied keys; `--remove-meta` removes keys. Replacing a connection retains
+the pinned daemon identity. Add `--replace-identity` with `--connection` only after verifying a changed
+destination. `rm` removes the saved connection; agents and workspaces on that host keep running.
+
 ## Provider diagnostics
 
 Ask the daemon to inspect the provider environment it actually uses:

public-docs/plugins/reference.md

@@ -26,6 +26,26 @@ Local plugins are directory sources installed into one Paseo daemon. A plugin ca
 
 Plugin code is trusted and unsandboxed. Client surfaces run in the Paseo app. Backend contributions run in a subprocess with access to the daemon machine, including its files, processes, credentials, and network.
 
+## Connected hosts
+
+`usePaseo().hosts` in a client surface and `context.paseo.hosts` in a server handler or lifecycle hook
+use the same [SDK host API](/sdk/reference#clienthosts). They describe the selected daemon's saved
+connections. Existing `useHosts()` and `getPaseoClient(serverId)` keep describing app-known hosts.
+
+```ts
+const remote = await context.paseo.hosts.client("srv_build_box");
+try {
+  return await remote.workspaces.list();
+} finally {
+  await remote.dispose();
+}
+```
+
+A returned API owns its observations. Teardown of the plugin API also disposes derived scopes.
+Keep remote calls inside the handler or lifecycle scope that owns them. Calling B does not forward
+B's plugin lifecycle hooks to A. Registrations created by a routed plugin retain a source qualified
+by the authenticated link and originating plugin.
+
 ## Project files
 
 `paseo plugin init /absolute/path/to/my-plugin` creates:

public-docs/sdk/reference.md

@@ -203,6 +203,63 @@ Use `workspace.terminals.create(options?)` and `workspace.terminals.list(options
 
 `config.patch(patch, requestId?)` validates, persists, and returns an updated configuration. Use this administrative surface for host configuration, not per-agent choices. A patch affects every client and future agent using that daemon.
 
+## `client.hosts`
+
+Manage hosts connected to the selected daemon. Both the managing daemon and destination must support connected hosts.
+
+```ts
+const directory = await client.hosts.list();
+const registration = await client.hosts.add({
+  connection: "ssh://deploy@build-box",
+  label: "Build box",
+  metadata: { team: "infra" },
+  sourceKey: "build-box",
+});
+await client.hosts.update(registration.registrationId, { enabled: false });
+await client.hosts.remove(registration.registrationId);
+```
+
+`list()` returns `{ self: { serverId, label }, hosts }`. Each host includes its registration and server
+IDs, label, authenticated source, string metadata, enabled state, connection status, safe target
+description, last-online/retry timestamps, and an actionable error. Credentials are never returned.
+A valid but unreachable connection is saved and retried. Invalid syntax fails without saving.
+
+`update(registrationId, patch)` accepts `label`, `metadata`, `enabled`, `connection`, and
+`replaceIdentity`. Metadata replaces the whole map. Replacing a connection preserves the expected
+server ID unless `replaceIdentity: true` explicitly accepts a new destination.
+
+`observe()` returns an owned subscription. Await `ready`, subscribe for snapshots and `hosts.changed`
+updates, then call `release()`. Observations restore a current snapshot after reconnect.
+
+```ts
+const remote = await client.hosts.client("srv_build_box");
+try {
+  const workspaces = await remote.workspaces.list();
+  const providers = await remote.providers.listAvailable();
+  await remote.agents.ref("agent_on_build_box").send("Continue the investigation");
+} finally {
+  await remote.dispose();
+}
+```
+
+`client(serverId)` returns a caller-owned `PaseoApi` bound to that destination through the selected
+daemon. The ordinary APIs keep their signatures. Paths and IDs belong to the destination. Selecting
+the local server ID returns a local scope. Routing supports one hop.
+
+Disposing one scope releases its observations without closing siblings or deleting the saved host.
+Origin disposal cascades. Observations recover through the ordinary client lifecycle. While connecting,
+new requests use the existing bounded send queue; disconnected requests fail. Connection loss clears
+old waiters and queued sends. Ordinary non-recoverable requests are not replayed; existing
+acknowledged/idempotent creation recovery retains its request identity. A submitted request whose response
+is lost rejects with `DaemonConnectionError` (`code: "DAEMON_CONNECTION_LOST"`), optional
+`outcome: "unknown"` and known `serverId`. Inspect before retrying a mutation.
+
+Registration and host acquisition failures expose a stable `code`: `host_unknown`, `host_disabled`, `host_offline`,
+`permission_denied`, `unsupported_capability`, `authentication_failed`, `server_id_mismatch`,
+`self_host`, or `duplicate_host`. Ordinary RPC permission codes, including `access_denied`, pass through
+unchanged. Multiplexing is a transport; consumers use the same client/API operations and lifecycle.
+Discovery does not grant execution authority.
+
 ## 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.

scripts/ci-workflow.test.mjs

@@ -475,7 +475,7 @@ test("Windows units and integration run independently with complete routed cover
   assert.doesNotMatch(invocations[0], / -t /);
   assert.equal(
     invocations[1],
-    'vitest run --maxWorkers=1 src/server/daemon-e2e/git-operations.e2e.test.ts src/server/session.create-agent-worktree-autoarchive.e2e.test.ts src/server/agent/providers/opencode-bridge.local.e2e.test.ts src/server/daemon-e2e/relay-transport.e2e.test.ts -t "merges into|creates a worktree and auto-archives both|versioned runtime|oversized attachments"',
+    'vitest run --maxWorkers=1 src/server/daemon-e2e/git-operations.e2e.test.ts src/server/session.create-agent-worktree-autoarchive.e2e.test.ts src/server/agent/providers/opencode-bridge.local.e2e.test.ts src/server/daemon-e2e/relay-transport.e2e.test.ts -t "merges into|creates a worktree and auto-archives both|versioned runtime|oversized attachments|connected hosts"',
   );
   const serverConfig = readFileSync(new URL("packages/server/vitest.config.ts", repoRoot), "utf8");
   assert.doesNotMatch(serverConfig, /maxWorkers/);