fix(hosts): share daemon links across isolated routed sessions

Multiplex logical sessions over the monitored daemon connection and wrap original binary bytes in routing headers. Admit channels under their authenticated parent and current grants; preserve scoped delivery and physical send completion.

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

@@ -10,6 +10,7 @@ const INBOUND_PERMISSION = {
   "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,
@@ -228,6 +229,7 @@ const OUTBOUND_PERMISSION = {
   "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,

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

@@ -1,6 +1,6 @@
 import { afterEach, expect, test } from "vitest";
 import { createPaseoApi } from "@getpaseo/client";
-import { createHostClient } from "@getpaseo/client/node";
+import { createHostClient, createHostChannelTransport } from "@getpaseo/client/node";
 import { createTestPaseoDaemon, type TestPaseoDaemon } from "../test-utils/paseo-daemon.js";
 import type { HostDirectory } from "@getpaseo/protocol/hosts";
 
@@ -91,18 +91,22 @@ test("SDK saves unreachable hosts, rejects syntax and preserves source identity
 
 test("ordinary routed APIs isolate scopes and enforce one-hop, disable and origin disposal", async () => {
   const b = await daemon();
+  const wire = await hostWire(b);
   const a = await daemon();
   const { paseo } = await api(a);
   const observation = paseo.hosts.observe();
   expect((await observation.ready).hosts).toEqual([]);
-  const saved = await paseo.hosts.add({ connection: `127.0.0.1:${b.port}`, label: "B" });
+  const saved = await paseo.hosts.add({ connection: wire.connection, label: "B" });
   const directory = await online(paseo);
   const serverId = directory.hosts[0]!.serverId!;
   const first = await paseo.hosts.client(serverId);
   const second = await paseo.hosts.client(serverId);
+  expect(wire.connections()).toBe(1);
+  expect(wire.active()).toBe(1);
   const workspaces = await second.workspaces.list({ subscribe: {} });
   expect(workspaces.entries).toEqual([]);
   await first.dispose();
+  expect(wire.active()).toBe(1);
   expect((await second.workspaces.list()).entries).toEqual([]);
   await expect(second.hosts.client(directory.self.serverId)).rejects.toMatchObject({
     code: "unsupported_capability",
@@ -123,7 +127,10 @@ test("ordinary routed APIs isolate scopes and enforce one-hop, disable and origi
       { timeout: 20000 },
     )
     .toBe(0);
+  expect(wire.connections()).toBe(2);
+  expect(wire.active()).toBe(1);
   await second.dispose();
+  expect(wire.active()).toBe(1);
   await observation.release();
   await paseo.dispose();
 });
@@ -160,9 +167,10 @@ test("routed authority is bounded by the caller, with isolated simultaneous sess
 
 test("distinct origin sessions cannot collide through a caller-selected route scope, including terminal binary output", async () => {
   const b = await daemon();
+  const wire = await hostWire(b);
   const a = await daemon();
   const admin = await api(a);
-  await admin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` });
+  await admin.paseo.hosts.add({ connection: wire.connection });
   const serverId = (await online(admin.paseo)).hosts[0]!.serverId!;
   const remote = await admin.paseo.hosts.client(serverId);
   const workspace = await remote.workspaces.create({
@@ -184,12 +192,45 @@ test("distinct origin sessions cannot collide through a caller-selected route sc
   const second = await RoutePeer.open(secondOrigin.client, serverId);
   await first.subscribeTerminal(one.id);
   await second.subscribeTerminal(two.id);
+  expect(wire.connections()).toBe(1);
   first.input("FIRST_ONLY\n");
   second.input("SECOND_ONLY\n");
   await expect.poll(() => first.output()).toContain("FIRST_ONLY");
   await expect.poll(() => second.output()).toContain("SECOND_ONLY");
   expect(first.output()).not.toContain("SECOND_ONLY");
   expect(second.output()).not.toContain("FIRST_ONLY");
+  const firstBytes = new Uint8Array([0, 255, 128, 173]);
+  const secondBytes = new Uint8Array([173, 128, 255, 0]);
+  const paths = await Promise.all([
+    first.upload("first.bin", firstBytes),
+    second.upload("second.bin", secondBytes),
+  ]);
+  expect((await readFile(paths[0]!)).equals(firstBytes)).toBe(true);
+  expect((await readFile(paths[1]!)).equals(secondBytes)).toBe(true);
+
+  const binary = wire.frames.filter((frame) => frame.binary);
+  expect(binary.length).toBeGreaterThan(0);
+  expect(binary.every((frame) => decodeHostRouteFrame(frame.bytes) !== null)).toBe(true);
+  expect(
+    binary.some((frame) => {
+      const payload = decodeHostRouteFrame(frame.bytes)!.payload;
+      return Buffer.from(payload).equals(
+        encodeTerminalStreamFrame({
+          opcode: TerminalStreamOpcode.Input,
+          slot: 0,
+          payload: "FIRST_ONLY\n",
+        }),
+      );
+    }),
+  ).toBe(true);
+  const json = wire.frames
+    .filter((frame) => !frame.binary)
+    .map((frame) => JSON.parse(frame.bytes.toString()));
+  expect(json.filter((frame) => frame.type === "hello")).toHaveLength(1);
+  expect(json.some((frame) => frame.message?.type === "hosts.link.open.request")).toBe(true);
+  expect(
+    json.some((frame) => frame.message?.binary === true || frame.message?.payload?.binary === true),
+  ).toBe(false);
   await first.close();
   second.input("STILL_SECOND\n");
   await expect.poll(() => second.output()).toContain("STILL_SECOND");
@@ -200,6 +241,9 @@ test("distinct origin sessions cannot collide through a caller-selected route sc
 });
 
 import {
+  encodeFileTransferFrame,
+  FileTransferOpcode,
+  decodeHostRouteFrame,
   decodeTerminalStreamFrame,
   encodeTerminalStreamFrame,
   TerminalStreamOpcode,
@@ -209,7 +253,7 @@ import {
   type SessionOutboundMessage,
   type SessionInboundMessage,
 } from "@getpaseo/protocol/messages";
-import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
+import { DaemonClient } from "@getpaseo/client/internal/daemon-client";
 class RoutePeer {
   readonly messages: SessionOutboundMessage[] = [];
   readonly bytes: Uint8Array[] = [];
@@ -219,15 +263,21 @@ class RoutePeer {
     private readonly origin: DaemonClient,
     private readonly routeId: string,
   ) {
-    this.detach = origin.subscribeRawMessages((message) => {
+    const binary = origin.subscribeHostFrames((frame) => {
+      if (frame.routeId === routeId) this.bytes.push(frame.payload);
+    });
+    const text = origin.subscribeRawMessages((message) => {
       if (message.type !== "hosts.route.frame" || message.payload.subscriptionId !== routeId)
         return;
-      if (message.payload.binary) this.bytes.push(Buffer.from(message.payload.data, "base64"));
-      else {
+      {
         const frame = WSOutboundMessageSchema.parse(JSON.parse(message.payload.data));
         if (frame.type === "session") this.messages.push(frame.message);
       }
     });
+    this.detach = () => {
+      binary();
+      text();
+    };
   }
   static async open(origin: DaemonClient, serverId: string) {
     const peer = new RoutePeer(
@@ -243,7 +293,6 @@ class RoutePeer {
         protocolVersion: 1,
         capabilities: { owned_subscriptions: true },
       }),
-      false,
     );
     await expect
       .poll(
@@ -264,7 +313,7 @@ class RoutePeer {
     return peer;
   }
   send(message: SessionInboundMessage) {
-    this.origin.sendHostFrame(this.routeId, JSON.stringify({ type: "session", message }), false);
+    this.origin.sendHostFrame(this.routeId, JSON.stringify({ type: "session", message }));
   }
   async subscribeTerminal(terminalId: string) {
     this.send({ type: "subscribe_terminal_request", terminalId, requestId: "same-request" });
@@ -284,16 +333,48 @@ class RoutePeer {
   input(text: string) {
     this.origin.sendHostFrame(
       this.routeId,
-      Buffer.from(
-        encodeTerminalStreamFrame({
-          opcode: TerminalStreamOpcode.Input,
-          slot: this.slot,
-          payload: text,
-        }),
-      ).toString("base64"),
-      true,
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.Input,
+        slot: this.slot,
+        payload: text,
+      }),
     );
   }
+  async upload(fileName: string, bytes: Uint8Array) {
+    const requestId = "same-upload-request";
+    const modifiedAt = new Date().toISOString();
+    this.send({
+      type: "file.upload.request",
+      requestId,
+      fileName,
+      mimeType: "application/octet-stream",
+      size: bytes.length,
+      modifiedAt,
+    });
+    for (const frame of [
+      encodeFileTransferFrame({
+        opcode: FileTransferOpcode.FileBegin,
+        requestId,
+        metadata: {
+          mime: "application/octet-stream",
+          size: bytes.length,
+          encoding: "binary",
+          modifiedAt,
+          fileName,
+        },
+      }),
+      encodeFileTransferFrame({ opcode: FileTransferOpcode.FileChunk, requestId, payload: bytes }),
+      encodeFileTransferFrame({ opcode: FileTransferOpcode.FileEnd, requestId }),
+    ])
+      this.origin.sendHostFrame(this.routeId, frame);
+    await expect
+      .poll(() => this.messages.some((message) => message.type === "file.upload.response"))
+      .toBe(true);
+    const response = this.messages.find((message) => message.type === "file.upload.response");
+    if (response?.type !== "file.upload.response" || !response.payload.file)
+      throw new Error("Upload failed");
+    return response.payload.file.path;
+  }
   output() {
     return this.bytes
       .flatMap((bytes) => {
@@ -310,9 +391,9 @@ class RoutePeer {
   }
 }
 
-import { mkdir, writeFile, stat, mkdtemp, rm } from "node:fs/promises";
-import { join } from "node:path";
-import { tmpdir } from "node:os";
+import { mkdir, writeFile, readFile, chmod, stat, mkdtemp, rm } from "node:fs/promises";
+import { join, dirname, basename } from "node:path";
+import { tmpdir, userInfo } from "node:os";
 import { resolveDaemonVersion } from "../daemon-version.js";
 
 test("identity discovery failures remain inspectable and changing a pinned destination requires explicit replacement", async () => {
@@ -433,7 +514,7 @@ export default function contribute(server) {
   expect((await direct.paseo.workspaces.list()).entries).toHaveLength(1);
 });
 
-import { execFile } from "node:child_process";
+import { execFile, spawn } from "node:child_process";
 import { promisify } from "node:util";
 import { fileURLToPath } from "node:url";
 
@@ -543,7 +624,7 @@ test("resuming an A session with narrower grants terminates existing routes and
   await remote.dispose();
 });
 
-import { WebSocketServer } from "ws";
+import { WebSocketServer, WebSocket } from "ws";
 import { createServer } from "node:http";
 import { once } from "node:events";
 import { hashSync } from "bcryptjs";
@@ -669,14 +750,10 @@ test("disposing an origin during routed handshake settles acquisition and closes
   let routeHello = false;
   let routeClosed = false;
   ws.on("connection", (socket) => {
-    const monitor = ++connections === 1;
-    socket.on("close", () => {
-      if (!monitor) routeClosed = true;
-    });
+    connections++;
     socket.on("message", (data) => {
       const frame = JSON.parse(data.toString());
-      if (frame.type !== "hello") return;
-      if (monitor)
+      if (frame.type === "hello")
         socket.send(
           JSON.stringify({
             type: "session",
@@ -686,13 +763,28 @@ test("disposing an origin during routed handshake settles acquisition and closes
                 status: "server_info",
                 serverId: "srv_pending",
                 hostname: "pending",
-                version: "0.11.0",
-                features: { connectedHosts: true, ownedSubscriptions: true },
+                version: "0.12.0",
+                features: {
+                  connectedHosts: true,
+                  hostMultiplexing: true,
+                  ownedSubscriptions: true,
+                },
               },
             },
           }),
         );
-      else routeHello = true;
+      if (frame.message?.type === "hosts.link.open.request")
+        socket.send(
+          JSON.stringify({
+            type: "session",
+            message: {
+              type: "hosts.link.open.response",
+              payload: { requestId: frame.message.requestId, routeId: "pending-channel" },
+            },
+          }),
+        );
+      if (frame.message?.type === "hosts.route.frame.request") routeHello = true;
+      if (frame.message?.type === "hosts.route.close.request") routeClosed = true;
     });
   });
   http.listen(0, "127.0.0.1");
@@ -715,4 +807,427 @@ test("disposing an origin during routed handshake settles acquisition and closes
   expect(await result).toBeInstanceOf(Error);
   await expect.poll(() => routeClosed).toBe(true);
   expect(ws.clients.size).toBe(1);
+  expect(connections).toBe(1);
+});
+
+test("source-key registration remains idempotent at directory capacity", async () => {
+  const { ConnectedHosts } = await import("./index.js");
+  const home = await mkdtemp(join(tmpdir(), "hosts-capacity-"));
+  cleanup.push(() => rm(home, { recursive: true, force: true }));
+  const rows = Array.from({ length: 128 }, (_, index) => ({
+    registrationId: `saved-${index}`,
+    serverId: null,
+    label: null,
+    source: { id: "plugin:capacity", key: `key-${index}` },
+    metadata: {},
+    enabled: false,
+    status: "disabled",
+    targetDescription: "127.0.0.1:1",
+    lastOnlineAt: null,
+    nextRetryAt: null,
+    error: null,
+    connection: "127.0.0.1:1",
+  }));
+  await writeFile(join(home, "hosts.json"), JSON.stringify(rows));
+  const registry = new ConnectedHosts(home, { serverId: "self", label: "self" });
+  cleanup.push(() => registry.close());
+  const actor = { sourceId: "plugin:capacity", manageSources: false };
+  await expect(
+    registry.add({ connection: "127.0.0.1:1", sourceKey: "key-0" }, actor),
+  ).resolves.toMatchObject({ registrationId: "saved-0", enabled: false });
+  await expect(
+    registry.add({ connection: "127.0.0.1:1", sourceKey: "new" }, actor),
+  ).rejects.toMatchObject({ code: "resource_limit" });
+});
+
+/** Real WebSocket proxy records the A–B physical wire without replacing either daemon. */
+async function hostWire(b: TestPaseoDaemon) {
+  const http = createServer();
+  const wss = new WebSocketServer({ server: http });
+  const frames: { binary: boolean; bytes: Buffer; direction: "toB" | "fromB" }[] = [];
+  let connections = 0;
+  wss.on("connection", (downstream) => {
+    connections++;
+    const upstream = new WebSocket(`ws://127.0.0.1:${b.port}/ws`);
+    const pending: (() => void)[] = [];
+    upstream.on("open", () => {
+      for (const send of pending.splice(0)) send();
+    });
+    downstream.on("message", (data, binary) => {
+      const bytes = Buffer.from(data as Buffer);
+      frames.push({ binary, bytes, direction: "toB" });
+      const send = () => {
+        if (upstream.readyState === 1) upstream.send(bytes, { binary });
+      };
+      if (upstream.readyState === 0) pending.push(send);
+      else send();
+    });
+    upstream.on("message", (data, binary) => {
+      const bytes = Buffer.from(data as Buffer);
+      frames.push({ binary, bytes, direction: "fromB" });
+      if (downstream.readyState === 1) downstream.send(bytes, { binary });
+    });
+    upstream.on("error", () => downstream.close());
+    upstream.on("close", () => downstream.close());
+    downstream.on("close", () => upstream.close());
+  });
+  http.listen(0, "127.0.0.1");
+  await once(http, "listening");
+  cleanup.push(async () => {
+    for (const socket of wss.clients) socket.terminate();
+    await new Promise<void>((resolve) => wss.close(() => resolve()));
+    await new Promise<void>((resolve) => http.close(() => resolve()));
+  });
+  const address = http.address();
+  if (!address || typeof address === "string") throw new Error("Missing wire port");
+  return {
+    connection: `127.0.0.1:${address.port}`,
+    frames,
+    connections: () => connections,
+    active: () => wss.clients.size,
+  };
+}
+
+test("a resumed physical link re-admits every logical session under its current grant ceiling", async () => {
+  const b = await daemon();
+  const parent = createHostClient(`127.0.0.1:${b.port}`, { clientId: "physical-grants" });
+  cleanup.push(() => parent.close());
+  await parent.connect();
+  const serverId = parent.getLastServerInfoMessage()!.serverId;
+  const logical = new DaemonClient({
+    url: "ws://logical/ws",
+    clientId: "logical",
+    transportFactory: createHostChannelTransport(
+      parent,
+      serverId,
+      "logical",
+      null,
+      { failure: null },
+      true,
+    ),
+    routedClient: true,
+    routedServerId: serverId,
+  });
+  cleanup.push(() => logical.close());
+  await logical.connect();
+  const scoped = createPaseoApi(logical);
+  cleanup.push(() => scoped.dispose());
+  const bytes = Uint8Array.from({ length: 256 * 1024 + 123 }, (_, index) => index % 251);
+  const upload = await logical.uploadFile({
+    fileName: "routed.bin",
+    mimeType: "application/octet-stream",
+    bytes,
+  });
+  expect(upload.error).toBeNull();
+  const downloaded = await logical.readFile(
+    dirname(upload.file!.path),
+    basename(upload.file!.path),
+  );
+  expect(Buffer.from(downloaded.bytes).equals(bytes)).toBe(true);
+
+  const observed = await scoped.workspaces.list({ subscribe: {} });
+  const snapshots: unknown[] = [];
+  observed.subscription.subscribe({ snapshot: (value) => snapshots.push(value), update() {} });
+  const narrowed = createHostClient(`127.0.0.1:${b.port}`, {
+    clientId: "physical-grants",
+    sessionPermissions: ["daemon.read", "workspace.read"],
+    reconnect: { enabled: false },
+  });
+  cleanup.push(() => narrowed.close());
+  await narrowed.connect();
+  await expect.poll(() => snapshots.length, { timeout: 15000 }).toBe(2);
+  expect(logical.getLastServerInfoMessage()?.permissions).toEqual([
+    "daemon.read",
+    "workspace.read",
+  ]);
+  await expect(
+    scoped.workspaces.create({ source: { kind: "directory", path: b.staticDir } }),
+  ).rejects.toMatchObject({ code: "permission_denied" });
+  const restored = createHostClient(`127.0.0.1:${b.port}`, {
+    clientId: "physical-grants",
+    reconnect: { enabled: false },
+  });
+  cleanup.push(() => restored.close());
+  await restored.connect();
+  await expect.poll(() => snapshots.length, { timeout: 15000 }).toBe(3);
+  expect(logical.getLastServerInfoMessage()?.permissions).toContain("workspace.write");
+  await scoped.workspaces.create({ source: { kind: "directory", path: b.staticDir } });
+});
+
+import { DaemonLinkEndpoint, VirtualHostSocket } from "./index.js";
+import { SessionDelivery } from "../session/owned-subscriptions/index.js";
+import { sendBoundedPhysicalFrameAndWait } from "../websocket/physical-socket.js";
+
+test("channel output accounts for escaped text, awaits physical drain, and preserves its sibling", async () => {
+  const http = createServer();
+  const wss = new WebSocketServer({ server: http });
+  http.listen(0, "127.0.0.1");
+  await once(http, "listening");
+  const address = http.address();
+  if (!address || typeof address === "string") throw new Error("Missing port");
+  const accepted = once(wss, "connection");
+  const receiver = new WebSocket(`ws://127.0.0.1:${address.port}`);
+  const received: { binary: boolean; bytes: Buffer }[] = [];
+  receiver.on("message", (bytes, binary) =>
+    received.push({ bytes: Buffer.from(bytes as Buffer), binary }),
+  );
+  const [physical] = (await accepted) as [WebSocket];
+  await once(receiver, "open");
+  cleanup.push(async () => {
+    receiver.terminate();
+    physical.terminate();
+    await new Promise<void>((resolve) => wss.close(() => resolve()));
+    await new Promise<void>((resolve) => http.close(() => resolve()));
+  });
+  const delivery = new SessionDelivery(
+    async (source, message) => {
+      expect(delivery.permits(source, message)).toBe(true);
+      await sendBoundedPhysicalFrameAndWait({
+        socket: physical,
+        frame: JSON.stringify({ type: "session", message }),
+        onHighWater: () => {
+          throw new Error("Shared link overflowed");
+        },
+      });
+    },
+    async (source, frame) => {
+      expect(delivery.permitsBinary(source, frame)).toBe(true);
+      await sendBoundedPhysicalFrameAndWait({
+        socket: physical,
+        frame,
+        onHighWater: () => {
+          throw new Error("Shared link overflowed");
+        },
+      });
+    },
+  );
+  delivery.attach(physical, true);
+  const channels: VirtualHostSocket[] = [];
+  const endpoint = new DaemonLinkEndpoint({
+    identity: "authenticated",
+    permissions: () => ["daemon.read"],
+    routed: () => false,
+    bufferedAmount: () => physical.bufferedAmount,
+    delivery,
+    admit: async (socket) => {
+      channels.push(socket);
+    },
+  });
+  cleanup.push(async () => {
+    endpoint.close();
+    await delivery.close();
+  });
+  for (const channelId of ["slow", "sibling"]) {
+    const request = { type: "hosts.link.open.request" as const, requestId: channelId, channelId };
+    await delivery.request(physical, request, () => endpoint.dispatch(request));
+  }
+  await expect(channels[0]!.send('"'.repeat(34 * 1024 * 1024))).rejects.toThrow(
+    "Host channel output limit reached",
+  );
+  expect(channels[0]!.readyState).toBe(3);
+  expect(physical.readyState).toBe(1);
+  await channels[1]!.send("sibling-before");
+  await expect
+    .poll(() => received.some((frame) => !frame.binary && frame.bytes.includes("sibling-before")))
+    .toBe(true);
+  receiver.pause();
+  let completed = false;
+  const bytes = new Uint8Array(8 * 1024 * 1024).fill(173);
+  const pending = channels[1]!.send(bytes).then(() => {
+    completed = true;
+    return undefined;
+  });
+  await expect.poll(() => physical.bufferedAmount).toBeGreaterThan(0);
+  expect(completed).toBe(false);
+  expect(physical.bufferedAmount).toBeLessThan(64 * 1024 * 1024);
+  receiver.resume();
+  await pending;
+  await channels[1]!.send("sibling-after");
+  await expect
+    .poll(() => received.some((frame) => !frame.binary && frame.bytes.includes("sibling-after")))
+    .toBe(true);
+  const binary = received.find((frame) => frame.binary)!;
+  expect(Buffer.from(decodeHostRouteFrame(binary.bytes)!.payload).equals(bytes)).toBe(true);
+  expect(wss.clients.size).toBe(1);
+});
+
+import { existsSync } from "node:fs";
+test.skipIf(process.platform !== "linux" || !existsSync("/usr/sbin/sshd"))(
+  "real SSH shares one tunnel across scopes and owns fresh attempts after recovery",
+  async () => {
+    const root = await mkdtemp(join(tmpdir(), "hosts-ssh-"));
+    cleanup.push(() => rm(root, { recursive: true, force: true }));
+    const run = promisify(execFile);
+    for (const name of ["host", "client"])
+      await run("ssh-keygen", ["-t", "ed25519", "-N", "", "-f", join(root, name)]);
+    const authorized = await readFile(join(root, "client.pub"));
+    await writeFile(join(root, "authorized_keys"), authorized);
+    const portServer = createServer();
+    portServer.listen(0, "127.0.0.1");
+    await once(portServer, "listening");
+    const address = portServer.address();
+    if (!address || typeof address === "string") throw new Error("Missing SSH port");
+    await new Promise<void>((resolve) => portServer.close(() => resolve()));
+    await writeFile(
+      join(root, "sshd_config"),
+      `Port ${address.port}\nListenAddress 127.0.0.1\nHostKey ${root}/host\nPidFile ${root}/pid\nAuthorizedKeysFile ${root}/authorized_keys\nStrictModes no\nPasswordAuthentication no\nKbdInteractiveAuthentication no\nUsePAM no\nAllowTcpForwarding yes\n`,
+    );
+    await writeFile(
+      join(root, "ssh"),
+      `#!/bin/sh\nprintf 'attempt\\n' >> '${root}/attempts'\nexec /usr/bin/ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile='${root}/known_hosts' -i '${root}/client' "$@"\n`,
+    );
+    await chmod(join(root, "ssh"), 0o700);
+    const originalPath = process.env.PATH;
+    process.env.PATH = `${root}:${originalPath}`;
+    cleanup.push(async () => {
+      process.env.PATH = originalPath;
+    });
+    const b = await createTestPaseoDaemon({ auth: { password: hashSync("ssh-password", 4) } });
+    cleanup.push(() => b.close());
+    const a = await daemon();
+    const { paseo, client } = await api(a);
+    const host = await paseo.hosts.add({
+      connection: `ssh://${userInfo().username}@127.0.0.1:${address.port}?daemonPort=${b.port}&password=ssh-password`,
+    });
+    await expect.poll(async () => (await paseo.hosts.list()).hosts[0]?.status).toBe("offline");
+    const sshd = spawn("/usr/sbin/sshd", [
+      "-D",
+      "-f",
+      join(root, "sshd_config"),
+      "-E",
+      join(root, "sshd.log"),
+    ]);
+    cleanup.push(async () => {
+      sshd.kill();
+    });
+    await online(paseo);
+    const attempts = async () =>
+      (await readFile(join(root, "attempts"), "utf8")).trim().split("\n").length;
+    const before = await attempts();
+    const first = await paseo.hosts.client(b.daemon.getServerId());
+    const sibling = await paseo.hosts.client(b.daemon.getServerId());
+    expect((await sibling.workspaces.list()).entries).toEqual([]);
+    expect(await attempts()).toBe(before);
+    await first.dispose();
+    expect((await sibling.workspaces.list()).entries).toEqual([]);
+    expect(await attempts()).toBe(before);
+    const workspace = await sibling.workspaces.create({
+      source: { kind: "directory", path: b.staticDir },
+    });
+    const terminal = await workspace.terminals.create({
+      command: process.execPath,
+      args: ["-e", "process.stdin.on('data', data => process.stdout.write(data))"],
+    });
+    const stream = await RoutePeer.open(client, b.daemon.getServerId());
+    await stream.subscribeTerminal(terminal.id);
+    stream.input("SSH_BINARY_BEFORE\n");
+    await expect.poll(() => stream.output()).toContain("SSH_BINARY_BEFORE");
+    await stream.close();
+    expect(await attempts()).toBe(before);
+
+    await paseo.hosts.update(host.registrationId, { enabled: false });
+    await paseo.hosts.update(host.registrationId, { enabled: true });
+    await online(paseo);
+    await expect
+      .poll(
+        async () => {
+          try {
+            return (await sibling.workspaces.list()).entries.length;
+          } catch {
+            return -1;
+          }
+        },
+        { timeout: 15000 },
+      )
+      .toBe(1);
+    expect(await attempts()).toBe(before + 1);
+    const resumed = await RoutePeer.open(client, b.daemon.getServerId());
+    await resumed.subscribeTerminal(terminal.id);
+    resumed.input("SSH_BINARY_AFTER\n");
+    await expect.poll(() => resumed.output()).toContain("SSH_BINARY_AFTER");
+    await resumed.close();
+    await terminal.kill();
+    expect(await attempts()).toBe(before + 1);
+    await sibling.dispose();
+    await paseo.hosts.update(host.registrationId, { enabled: false });
+    await writeFile(join(root, "authorized_keys"), "");
+    await paseo.hosts.update(host.registrationId, { enabled: true });
+    await expect
+      .poll(async () => (await paseo.hosts.list()).hosts[0]?.error?.code, { timeout: 15000 })
+      .toBe("authentication_failed");
+    expect((await paseo.hosts.list()).hosts[0]?.nextRetryAt).toBeNull();
+    await writeFile(join(root, "authorized_keys"), authorized);
+    await paseo.hosts.update(host.registrationId, { enabled: true });
+    await online(paseo);
+  },
+  60000,
+);
+
+test("malformed routing headers terminate their physical source and leave the daemon usable", async () => {
+  const b = await daemon();
+  const socket = new WebSocket(`ws://127.0.0.1:${b.port}/ws`);
+  cleanup.push(async () => socket.terminate());
+  await once(socket, "open");
+  const info = once(socket, "message");
+  socket.send(
+    JSON.stringify({
+      type: "hello",
+      clientId: "malformed-route",
+      clientType: "cli",
+      protocolVersion: 1,
+      capabilities: { owned_subscriptions: true },
+    }),
+  );
+  await info;
+  const opened = once(socket, "message");
+  socket.send(
+    JSON.stringify({
+      type: "session",
+      message: { type: "hosts.link.open.request", requestId: "channel", channelId: "channel" },
+    }),
+  );
+  const [response] = await opened;
+  expect(JSON.parse(String(response)).message.type).toBe("hosts.link.open.response");
+  const closed = once(socket, "close");
+  socket.send(new Uint8Array([0x30, 0, 3, 97]));
+  expect((await closed)[0]).toBe(1008);
+  expect((await (await api(b)).paseo.workspaces.list()).entries).toEqual([]);
+});
+
+test("cancelling a logical scope during admission settles it and releases its owned channel", async () => {
+  const source = {};
+  const messages: SessionOutboundMessage[] = [];
+  const delivery = new SessionDelivery((_source, message) => {
+    messages.push(message);
+  });
+  delivery.attach(source, true);
+  let pendingSocket: VirtualHostSocket | undefined;
+  const endpoint = new DaemonLinkEndpoint({
+    identity: "physical",
+    permissions: () => ["daemon.read"],
+    routed: () => false,
+    bufferedAmount: () => 0,
+    delivery,
+    admit: (socket) =>
+      new Promise<void>((resolve) => {
+        pendingSocket = socket;
+        socket.once("close", resolve);
+      }),
+  });
+  const open = {
+    type: "hosts.link.open.request" as const,
+    requestId: "pending",
+    channelId: "logical-scope",
+  };
+  const acquiring = delivery.request(source, open, () => endpoint.dispatch(open));
+  expect(pendingSocket).toBeDefined();
+  const cancel = { type: "hosts.route.close.request" as const, scopeId: "logical-scope" };
+  expect(endpoint.handles(cancel)).toBe(true);
+  await delivery.request(source, cancel, () => endpoint.dispatch(cancel));
+  await acquiring;
+  expect(pendingSocket!.readyState).toBe(3);
+  expect(messages.some((message) => message.type === "hosts.link.open.response")).toBe(false);
+  expect(messages.filter((message) => message.type === "hosts.route.closed")).toHaveLength(1);
+  endpoint.close();
+  await delivery.close();
 });

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

@@ -1,3 +1,4 @@
+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";
@@ -5,7 +6,7 @@ import { join } from "node:path";
 import { z } from "zod";
 import {
   createHostClient,
-  createHostTransportFactory,
+  createHostChannelTransport,
   parseHostTarget,
   HostTransportBudget,
 } from "@getpaseo/client/node";
@@ -37,6 +38,8 @@ export interface HostActor {
   manageSources: boolean;
 }
 interface Live {
+  client: DaemonClient;
+  accepted: boolean;
   stop(): void;
 }
 interface Route {
@@ -118,8 +121,6 @@ export class ConnectedHosts {
     return this.mutate(async () => {
       input = HostAddSchema.parse(input);
       const target = parseHostTarget(input.connection);
-      if (this.rows.length >= 128)
-        throw new HostError("resource_limit", "Remove a host before adding another (limit 128)");
       const source = { id: actor.sourceId, key: input.sourceKey ?? randomUUID() };
       const existing = this.rows.find(
         (row) => row.source.id === source.id && row.source.key === source.key,
@@ -137,6 +138,8 @@ export class ConnectedHosts {
           "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,
@@ -233,55 +236,26 @@ export class ConnectedHosts {
   }
   private start(row: Saved): void {
     if (!row.enabled || this.closed) return;
-    let client: DaemonClient | null = null;
-    let timer: ReturnType<typeof setTimeout> | undefined;
     let stopped = false;
-    let connecting = false;
+    const 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;
-        clearTimeout(timer);
-        void client?.close();
+        void client.close();
       },
     };
     this.live.set(row.registrationId, live);
-    const retry = () => {
-      if (stopped || connecting || timer || row.status === "error") return;
-      row.status = "offline";
-      row.error = {
-        code: "host_offline",
-        message: "Cannot reach host. Check the connection; retrying automatically.",
-      };
-      row.nextRetryAt = new Date(Date.now() + 5000).toISOString();
-      this.changed();
-      timer = setTimeout(() => {
-        timer = undefined;
-        void attempt();
-      }, 5000);
-      timer.unref();
-    };
-    const attempt = async () => {
-      if (stopped) return;
-      await client?.close();
-      client = createHostClient(row.connection, {
-        clientId: `host-monitor-${this.self.serverId}-${row.registrationId}`,
-        reconnect: { enabled: false },
-        connectTimeoutMs: 10000,
-      });
-      client.subscribeConnectionStatus((state) => {
-        if (!stopped && state.status === "disconnected") {
-          for (const route of this.routes)
-            if (route.registrationId === row.registrationId) route.close();
-          retry();
-        }
-      });
-      connecting = true;
-      row.status = "connecting";
-      row.nextRetryAt = null;
-      this.changed();
+    const accept = async () => {
       try {
-        await client.connect();
-        if (stopped) return;
         const info = client.getLastServerInfoMessage();
         if (!info) throw new HostError("host_offline", "No daemon identity received");
         if (row.serverId && info.serverId !== row.serverId)
@@ -306,7 +280,7 @@ export class ConnectedHosts {
             "duplicate_host",
             "Another registration already owns this daemon. Remove the duplicate registration.",
           );
-        if (info.features?.connectedHosts !== true)
+        if (info.features?.connectedHosts !== true || info.features?.hostMultiplexing !== true)
           throw new HostError(
             "unsupported_capability",
             "Update the destination daemon to use connected hosts.",
@@ -325,25 +299,55 @@ export class ConnectedHosts {
             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;
-        const failure = connectionFailure(error, client.authFailureReason !== null);
-        if (failure) {
-          row.status = "error";
-          row.error = failure;
-          row.nextRetryAt = null;
-          this.changed();
-        }
+        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();
-      } finally {
-        connecting = false;
-        if (!stopped && !client.isConnected) retry();
       }
     };
-    void attempt();
+    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;
@@ -352,9 +356,9 @@ export class ConnectedHosts {
     originSessionId: string;
     permissions: readonly DaemonPermission[];
     signal: AbortSignal;
-    frame(data: string, binary: boolean): void;
+    frame(data: string | Uint8Array): void;
     ended(code: string, message: string): void;
-  }): Promise<{ send(data: string, binary: boolean): void; close(): 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);
@@ -371,10 +375,18 @@ export class ConnectedHosts {
         input.serverId,
       );
     if (this.routes.size >= 128) throw new HostError("resource_limit", "Too many routed sessions");
-    const transport = createHostTransportFactory(
-      row.connection,
-      this.budget,
-    )({ url: "ws://host-route/ws" });
+    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,
+      { failure: null },
+      true,
+    )({ url: "ws://host-link/ws" });
     let closed = false;
     let verified = false;
     let helloSent = false;
@@ -460,22 +472,19 @@ export class ConnectedHosts {
           return;
         }
       }
-      input.frame(
-        binary ? Buffer.from(data as Uint8Array).toString("base64") : String(data),
-        binary,
-      );
+      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, binary) => {
+      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 WSHelloMessage;
+          const hello = JSON.parse(data as string) as WSHelloMessage;
           if (hello.type !== "hello") throw new HostError("invalid_message", "Expected hello");
-          const target = parseHostTarget(row.connection);
           const clientId = `route-${createHash("sha256")
             .update(
               JSON.stringify([
@@ -493,7 +502,7 @@ export class ConnectedHosts {
               routed: true,
               producer: { serverId: this.self.serverId, sourceId: input.sourceId },
               sessionPermissions: input.permissions,
-              auth: target.password ? { kind: "password", password: target.password } : undefined,
+              auth: undefined,
             }),
           );
           helloSent = true;
@@ -501,7 +510,7 @@ export class ConnectedHosts {
         }
         if (!verified) throw new HostError("host_offline", "Host identity has not been verified");
         try {
-          transport.send(binary ? Buffer.from(data, "base64") : data);
+          transport.send(data);
         } catch {
           route.close("resource_limit", "Host buffer limit reached or transport unavailable");
         }

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

@@ -1,3 +1,4 @@
+import { encodeHostRouteFrame, type HostRouteFrame } from "@getpaseo/protocol/binary-frames/index";
 import type {
   SessionInboundMessage,
   SessionOutboundMessage,
@@ -13,7 +14,7 @@ type HostRequest<T extends SessionInboundMessage["type"]> = Extract<
 interface SessionRoute {
   scopeId: string;
   source: object;
-  send(data: string, binary: boolean): void;
+  send(data: string | Uint8Array): void;
   release(): Promise<void>;
   fail(code: string, message: string): void;
 }
@@ -31,6 +32,15 @@ export class ConnectedHostSession {
     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;
   }
@@ -116,7 +126,7 @@ export class ConnectedHostSession {
         const route = this.routes.get(message.routeId);
         if (!route || route.source !== this.delivery.currentSource)
           throw new HostError("host_unknown", "Unknown routed session");
-        route.send(message.data, message.binary);
+        route.send(message.data);
         return;
       }
       case "hosts.route.close.request": {
@@ -196,17 +206,25 @@ export class ConnectedHostSession {
         originSessionId: this.originSessionId,
         permissions: this.permissions(),
         signal: owner.signal,
-        frame: (data, binary) => {
-          const size = Buffer.byteLength(data) + 512;
+        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;
           }
-          owner.emit({
-            type: "hosts.route.frame",
-            payload: { subscriptionId: owner.id, data, binary },
-          });
+          if (output) owner.emit(output);
+          else owner.emitBinary(binary!);
         },
         ended: fail,
       });

packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts

@@ -281,6 +281,7 @@ async function waitForCapturedLog(
     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;
@@ -301,7 +302,8 @@ async function waitForCapturedLog(
         relayEndpoint: b.config.relayEndpoint,
       });
       const { serverId, daemonPublicKeyB64 } = decodeOfferFromFragmentUrl(offerUrl);
-      const connection = new URL(`relay://127.0.0.1:${port}/${serverId}`);
+      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() });
@@ -314,8 +316,12 @@ async function waitForCapturedLog(
       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;
@@ -333,6 +339,7 @@ async function waitForCapturedLog(
         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();
@@ -366,6 +373,7 @@ async function waitForCapturedLog(
         )
         .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");
@@ -398,6 +406,7 @@ async function waitForCapturedLog(
       await origin?.close();
       await a?.close();
       await b?.close();
+      await wire?.close();
       await stopRelay();
     }
   }, 120000);
@@ -1080,19 +1089,20 @@ async function openConnectedHostTerminal(origin: HostClient, serverId: string, t
   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;
-    if (message.payload.binary) {
-      const frame = decodeTerminalStreamFrame(Buffer.from(message.payload.data, "base64"));
-      if (frame?.opcode === TerminalStreamOpcode.Output)
-        output += new TextDecoder().decode(frame.payload);
-    } else {
-      const frame = WSOutboundMessageSchema.parse(JSON.parse(message.payload.data));
-      if (frame.type === "session") messages.push(frame.message);
-    }
+    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 {
@@ -1104,7 +1114,6 @@ async function openConnectedHostTerminal(origin: HostClient, serverId: string, t
         clientType: "cli",
         protocolVersion: 1,
       }),
-      false,
     );
     await expect
       .poll(() =>
@@ -1119,7 +1128,6 @@ async function openConnectedHostTerminal(origin: HostClient, serverId: string, t
         type: "session",
         message: { type: "subscribe_terminal_request", terminalId, requestId: "binary" },
       }),
-      false,
     );
     await expect
       .poll(() => messages.some((message) => message.type === "subscribe_terminal_response"))
@@ -1134,10 +1142,7 @@ async function openConnectedHostTerminal(origin: HostClient, serverId: string, t
       write: (payload: string) =>
         origin.sendHostFrame(
           routeId,
-          Buffer.from(
-            encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Input, slot, payload }),
-          ).toString("base64"),
-          true,
+          encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Input, slot, payload }),
         ),
     };
   } catch (error) {
@@ -1145,3 +1150,46 @@ async function openConnectedHostTerminal(origin: HostClient, serverId: string, t
     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

@@ -453,7 +453,7 @@ export interface SessionOptions {
   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;
@@ -699,14 +699,12 @@ export class Session {
   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),
@@ -728,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:
@@ -893,9 +891,13 @@ export class Session {
     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;
@@ -3213,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;
     }
@@ -8653,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);

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.ts

@@ -1,3 +1,5 @@
+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";
@@ -438,6 +440,7 @@ export interface WebSocketLike {
 }
 
 interface SessionConnectionBase {
+  hubExecutionAgents?: HubExecutionAgents;
   permissionCeiling?: readonly DaemonPermission[];
   session: Session;
   principalId: string;
@@ -469,7 +472,7 @@ interface SocketSessionOptions {
   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;
@@ -530,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();
@@ -1052,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 };
@@ -1169,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" &&
@@ -1190,7 +1214,7 @@ export class VoiceAssistantWebSocketServer {
       return this.ensureOutboundCapacity(ws, 0);
     });
     if (writableSockets.length === 0) {
-      return;
+      return null;
     }
 
     let payload: string;
@@ -1198,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;
     }
   }
 
@@ -1415,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) {
@@ -1460,6 +1500,7 @@ export class VoiceAssistantWebSocketServer {
     const base: SessionConnectionBase = {
       session,
       principalId: admission.principalId,
+      hubExecutionAgents: admission.hubExecutionAgents,
       sessionKey: sessionConnectionKey(admission.principalId, clientId),
       clientId,
       appVersion,
@@ -1481,6 +1522,7 @@ export class VoiceAssistantWebSocketServer {
       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;
         }
@@ -1790,6 +1832,10 @@ 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);
@@ -1853,6 +1899,7 @@ export class VoiceAssistantWebSocketServer {
       ...(this.serverCapabilities ? { capabilities: this.serverCapabilities } : {}),
       features: {
         connectedHosts: true,
+        hostMultiplexing: true,
         usageSources: true,
         ownedSubscriptions: true,
         agentRequestReceipts: true,
@@ -2094,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) : {};
@@ -2288,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;
     }
@@ -2442,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) {