refactor(hosts): use the ordinary client contract over channel transports

Keep queuing, permission codes and reconnection independent of transport selection. Track submitted request outcomes generically while retaining connection error codes and structured causes.
2026-10-22 17:22:47 +00:00 · 2026-10-20 07:59:01 +02:00
parent 71cc853f7b
commit 569b34263d
8 changed files with 304 additions and 88 deletions
@@ -43,9 +43,14 @@ 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; requests made while offline fail. Losing a response
-returns `outcome_unknown` with the destination ID. Inspect the destination before retrying a mutation.
-No mutation is automatically replayed.
+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
@@ -3,6 +3,7 @@ import { afterEach, expect, expectTypeOf, test, vi } from "vitest";
 import { z } from "zod";
 import {
  DaemonClient,
+  DaemonConnectionError,
  supportsUsageReports,
  type DaemonClientTrace,
  type CreateAgentRequestOptions,
@@ -7847,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,
+  });
+});
@@ -398,9 +398,6 @@ export interface DaemonClientConfig {
  /** Deliver compact bodies/hash references to a caller-owned snapshot cache.
   * The default keeps public SDK snapshot entries expanded. */
  providerSnapshots?: "wire";
-  routedClient?: boolean;
-  routedServerId?: string;
-  routedFailure?: () => Error | null;
  sessionPermissions?: import("@getpaseo/protocol/messages").DaemonPermission[];
  url: string;
  clientId: string;
@@ -949,6 +946,7 @@ interface Waiter<T> {
  reject(error: Error): void;
  timeoutHandle: ReturnType<typeof setTimeout> | null;
  requestId?: string;
+  submitted: boolean;
 }

 interface WaitHandle<T> {
@@ -1006,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";
  }
 }
@@ -1303,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();
@@ -1500,6 +1505,7 @@ export class DaemonClient {
          }
          this.scheduleReconnect({
            reason,
+            failure: event instanceof Error ? event : undefined,
            event: "TRANSPORT_CLOSE",
            reasonCode: "transport_closed",
          });
@@ -1619,7 +1625,6 @@ export class DaemonClient {
  }

  ensureConnected(options?: { verify?: boolean }): void {
-    this.requireRoutedConnection();
    if (this.connectionState.status === "disposed") {
      return;
    }
@@ -1655,7 +1660,7 @@ export class DaemonClient {
          event: "CONNECTION_VERIFICATION_FAILED",
          reasonCode: "liveness_timeout",
        });
-        if (!this.config.routedClient) this.ensureConnected();
+        this.ensureConnected();
      })
      .finally(() => {
        if (this.connectionVerification === transport) this.connectionVerification = null;
@@ -1666,7 +1671,7 @@ export class DaemonClient {
    if (!this.isConnected)
      throw Object.assign(new Error("The managing daemon is offline"), {
        code: "host_offline",
-        serverId: this.config.routedServerId,
+        serverId: this.destinationServerId,
      });
    if (this.lastServerInfoMessage?.features?.connectedHosts !== true)
      throw Object.assign(new Error("Update the managing daemon to use connected hosts"), {
@@ -1907,15 +1912,32 @@ 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 this.unavailableError();
    }
@@ -1926,6 +1948,7 @@ export class DaemonClient {
      size: String(getTransportFrameSize(frame)),
    });
    try {
+      submitted?.();
      this.transport.send(frame);
    } finally {
      this.endTraceSection(isOpen);
@@ -1946,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;
@@ -1985,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);
    });
  }

@@ -2004,7 +2027,7 @@ export class DaemonClient {
    }

    // If connecting, queue the message to be sent once connected
-    if (status === "connecting" && !this.config.routedClient) {
+    if (status === "connecting") {
      return new Promise((resolve, reject) => {
        const timeoutHandle = setTimeout(() => {
          // Remove from queue
@@ -2079,13 +2102,10 @@ export class DaemonClient {
            kind: "error",
            error: new DaemonRpcError({
              requestId: msg.payload.requestId,
-              serverId: this.config.routedServerId,
+              serverId: this.destinationServerId,
              error: msg.payload.error,
              requestType: msg.payload.requestType,
-              code:
-                this.config.routedClient && msg.payload.code === "access_denied"
-                  ? "permission_denied"
-                  : msg.payload.code,
+              code: msg.payload.code,
            }),
          };
        }
@@ -2194,18 +2214,11 @@ export class DaemonClient {
    });
  }

-  private requireRoutedConnection(): void {
-    if (this.config.routedClient && !this.isConnected) throw this.unavailableError();
-  }
-
-  private unavailableError(): Error {
-    if (this.config.routedFailure?.()) return this.config.routedFailure()!;
-    if (this.config.routedClient)
-      return Object.assign(new Error(`Host ${this.config.routedServerId} is offline`), {
-        code: "host_offline",
-        serverId: this.config.routedServerId,
-      });
-    return new DaemonConnectionError("Transport not connected");
+  private unavailableError(): DaemonConnectionError {
+    return new DaemonConnectionError("Transport not connected", "DAEMON_CONNECTION_LOST", {
+      serverId: this.destinationServerId,
+      cause: this.connectionFailure,
+    });
  }

  private sendSessionMessageStrict(message: SessionInboundMessage): void {
@@ -2214,7 +2227,7 @@ export class DaemonClient {
    }
    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));
    }
@@ -3030,7 +3043,6 @@ export class DaemonClient {

  private readonly creations = new CreationClient({
    supports: (feature) => {
-      this.requireRoutedConnection();
      return this.lastServerInfoMessage?.features?.[feature] === true;
    },
    requestId: () => this.createRequestId(),
@@ -6847,16 +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(
-      this.config.routedClient
-        ? Object.assign(
-            new Error("Connection lost before result; inspect the destination before retrying"),
-            { code: "outcome_unknown", serverId: this.config.routedServerId },
-          )
-        : new DaemonConnectionError(reason ?? "Connection lost"),
+    const connectionError = new DaemonConnectionError(
+      reason ?? "Connection lost",
+      "DAEMON_CONNECTION_LOST",
+      {
+        serverId: this.destinationServerId,
+        cause: input?.failure,
+      },
    );
-    this.rejectPendingSendQueue(new DaemonConnectionError(reason ?? "Connection lost"));
-    this.rejectPingProbe(new DaemonConnectionError(reason ?? "Connection lost"));
+    this.connectionFailure = input?.failure;
+    this.clearWaiters(connectionError, true);
+    this.rejectPendingSendQueue(connectionError);
+    this.rejectPingProbe(connectionError);
    this.terminalStreams.clearSlots();
    this.lastServerInfoMessage = null;

@@ -6994,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;
@@ -7077,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();
  }
@@ -7189,6 +7213,7 @@ export class DaemonClient {
        reject: wrappedReject,
        timeoutHandle,
        requestId: options?.requestId,
+        submitted: false,
      };
      this.waiters.add(waiter);
    });
@@ -19,7 +19,6 @@ export function createHostChannelTransport(
  serverId: string,
  scopeId: string,
  initialRouteId: string | null,
-  state: { failure: Error | null },
  link = false,
 ): DaemonTransportFactory {
  let first: string | null = initialRouteId;
@@ -34,7 +33,7 @@ export function createHostChannelTransport(
    const close = (error?: unknown) => {
      if (closed) return;
      closed = true;
-      if (error instanceof Error) state.failure = Object.assign(error, { serverId });
+      if (error instanceof Error) Object.assign(error, { serverId });
      for (const cleanup of cleanups) cleanup();
      try {
        if (routeId) origin.closeHostRoute(routeId);
@@ -80,7 +79,6 @@ export function createHostChannelTransport(
          return;
        }
        routeId = id;
-        state.failure = null;
        for (const handler of opens) handler();
        return undefined;
      })
@@ -161,6 +159,10 @@ export function createHostActions(
      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);
      }
@@ -168,16 +170,12 @@ export function createHostActions(
        origin.closeHostRoute(routeId);
        assertActive();
      }
-      const state: { failure: Error | null } = { failure: null };
      const client = new DaemonClient({
        url: "ws://routed-host/ws",
        clientId: scopeId,
-        transportFactory: createHostChannelTransport(origin, serverId, scopeId, routeId, state),
+        transportFactory: createHostChannelTransport(origin, serverId, scopeId, routeId),
        // Initial acquisition has a deadline; established scopes restore demand on loss.
        reconnect: { enabled: false },
-        routedClient: true,
-        routedServerId: serverId,
-        routedFailure: () => state.failure,
      });
      const api = owner.api(client, { signal: owner.signal });
      const scope = owner.own(() => ({
@@ -195,12 +193,13 @@ export function createHostActions(
        assertActive();
        return { ...api, dispose: () => scope.release() };
      } catch (error) {
-        const failure =
-          state.failure ??
-          Object.assign(error instanceof Error ? error : new Error("Host handshake failed"), {
+        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;
      }
@@ -157,7 +157,7 @@ test("routed authority is bounded by the caller, with isolated simultaneous sess
  expect((await remote.workspaces.list()).entries).toEqual([]);
  await expect(
    remote.workspaces.create({ source: { kind: "directory", path: b.paseoHome } }),
-  ).rejects.toMatchObject({ code: "permission_denied" });
+  ).rejects.toMatchObject({ code: "access_denied" });
  await expect(viewer.hosts.add({ connection: "127.0.0.1:1" })).rejects.toMatchObject({
    code: "permission_denied",
  });
@@ -165,6 +165,42 @@ test("routed authority is bounded by the caller, with isolated simultaneous sess
  await remote.dispose();
 });

+test("SDK host clients preserve ordinary permission RPC errors", async () => {
+  const b = await daemon();
+  const a = await daemon();
+  const owner = await api(a);
+  await owner.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` });
+  const serverId = (await online(owner.paseo)).hosts[0]!.serverId!;
+  const clients = [a, b].map((host) =>
+    createHostClient(`127.0.0.1:${host.port}`, {
+      clientId: crypto.randomUUID(),
+      sessionPermissions: ["daemon.read", "workspace.read"],
+      reconnect: { enabled: false },
+    }),
+  );
+  for (const client of clients) {
+    cleanup.push(() => client.close());
+    await client.connect();
+  }
+  const origin = createPaseoApi(clients[0]!);
+  const direct = createPaseoApi(clients[1]!);
+  const remote = await origin.hosts.client(serverId);
+  cleanup.push(async () => {
+    await remote.dispose();
+    await direct.dispose();
+    await origin.dispose();
+  });
+  const input = { source: { kind: "directory" as const, path: b.staticDir } };
+  const directError = (await direct.workspaces.create(input).catch((error: unknown) => error)) as {
+    code: string;
+  };
+  const remoteError = (await remote.workspaces.create(input).catch((error: unknown) => error)) as {
+    code: string;
+  };
+  expect(directError.code).toBe("access_denied");
+  expect(remoteError.code).toBe(directError.code);
+});
+
 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);
@@ -619,7 +655,7 @@ test("resuming an A session with narrower grants terminates existing routes and
  await expect.poll(() => snapshots.length, { timeout: 15000 }).toBe(2);
  await expect(
    remote.workspaces.create({ source: { kind: "directory", path: b.staticDir } }),
-  ).rejects.toMatchObject({ code: "permission_denied" });
+  ).rejects.toMatchObject({ code: "access_denied" });
  expect((await remote.workspaces.list()).entries).toEqual([]);
  await remote.dispose();
 });
@@ -722,7 +758,7 @@ test("an unreachable saved target becomes online after its daemon starts without
  });
 });

-test("losing a sent mutation result identifies the destination and never queues unavailable operations", async () => {
+test("losing a sent mutation result retains ordinary errors, identifies the destination and never replays it", async () => {
  const b = await daemon();
  const a = await daemon();
  const { paseo, client } = await api(a);
@@ -732,15 +768,29 @@ test("losing a sent mutation result identifies the destination and never queues
  const pending = remote.config.patch({ pluginsEnabled: false });
  const result = pending.catch((error: unknown) => error);
  await client.close();
-  expect(await result).toMatchObject({ code: "outcome_unknown", serverId });
-  await expect(remote.workspaces.list()).rejects.toMatchObject({ code: "host_offline", serverId });
-  await expect(remote.terminals.list()).rejects.toMatchObject({ code: "host_offline", serverId });
-  await expect(
-    remote.workspaces.create({
-      source: { kind: "directory", path: "/tmp" },
-      workspaceId: "offline-never-created",
-    }),
-  ).rejects.toMatchObject({ code: "host_offline", serverId });
+  expect(await result).toMatchObject({
+    code: "DAEMON_CONNECTION_LOST",
+    outcome: "unknown",
+    serverId,
+  });
+  await expect(remote.workspaces.list()).rejects.toMatchObject({
+    code: "DAEMON_CONNECTION_LOST",
+    serverId,
+  });
+  const direct = await api(b);
+  await direct.client.close();
+  const terminalFailure = (await direct.paseo.terminals
+    .list()
+    .catch((error: unknown) => error)) as Error;
+  await expect(remote.terminals.list()).rejects.toThrow(terminalFailure.message);
+  const creation = {
+    source: { kind: "directory" as const, path: "/tmp" },
+    workspaceId: "offline-never-created",
+  };
+  const creationFailure = (await direct.paseo.workspaces
+    .create(creation)
+    .catch((error: unknown) => error)) as Error;
+  await expect(remote.workspaces.create(creation)).rejects.toThrow(creationFailure.message);
  await remote.dispose();
 });

@@ -910,21 +960,22 @@ test("a resumed physical link re-admits every logical session under its current
  const logical = new DaemonClient({
    url: "ws://logical/ws",
    clientId: "logical",
-    transportFactory: createHostChannelTransport(
-      parent,
-      serverId,
-      "logical",
-      null,
-      { failure: null },
-      true,
-    ),
-    routedClient: true,
-    routedServerId: serverId,
+    transportFactory: createHostChannelTransport(parent, serverId, "logical", null, true),
  });
  cleanup.push(() => logical.close());
-  await logical.connect();
+  const connecting = logical.connect();
+  expect(() => logical.ensureConnected()).not.toThrow();
  const scoped = createPaseoApi(logical);
  cleanup.push(() => scoped.dispose());
+  const initialCall = scoped.workspaces.list();
+  await connecting;
+  expect((await initialCall).entries).toEqual([]);
+  const recoveryCalls: Promise<unknown>[] = [];
+  const stopRecovery = logical.subscribeConnectionStatus((state) => {
+    if (state.status === "connecting")
+      recoveryCalls.push(scoped.workspaces.list().catch((error: unknown) => error));
+  });
+  cleanup.push(async () => stopRecovery());
  const bytes = Uint8Array.from({ length: 256 * 1024 + 123 }, (_, index) => index % 251);
  const upload = await logical.uploadFile({
    fileName: "routed.bin",
@@ -955,7 +1006,7 @@ test("a resumed physical link re-admits every logical session under its current
  ]);
  await expect(
    scoped.workspaces.create({ source: { kind: "directory", path: b.staticDir } }),
-  ).rejects.toMatchObject({ code: "permission_denied" });
+  ).rejects.toMatchObject({ code: "access_denied" });
  const restored = createHostClient(`127.0.0.1:${b.port}`, {
    clientId: "physical-grants",
    reconnect: { enabled: false },
@@ -964,6 +1015,8 @@ test("a resumed physical link re-admits every logical session under its current
  await restored.connect();
  await expect.poll(() => snapshots.length, { timeout: 15000 }).toBe(3);
  expect(logical.getLastServerInfoMessage()?.permissions).toContain("workspace.write");
+  expect(recoveryCalls.length).toBeGreaterThanOrEqual(2);
+  for (const call of recoveryCalls) expect(await call).toMatchObject({ entries: [] });
  await scoped.workspaces.create({ source: { kind: "directory", path: b.staticDir } });
 });

@@ -384,7 +384,6 @@ export class ConnectedHosts {
      input.serverId,
      randomUUID(),
      null,
-      { failure: null },
      true,
    )({ url: "ws://host-link/ws" });
    let closed = false;
@@ -329,6 +329,7 @@ async function waitForCapturedLog(
        snapshot: () => {
          snapshots++;
        },
+        update() {},
      });
      const workspace = await remote.workspaces.create({
        source: { kind: "directory", path: b.staticDir },
@@ -358,7 +359,7 @@ async function waitForCapturedLog(
          },
          { timeout: 15000 },
        )
-        .toBe("host_offline");
+        .toBe("DAEMON_CONNECTION_LOST");
      await startRelay({ useLocalRelay: true, port });
      await expect
        .poll(
@@ -247,12 +247,18 @@ daemon. The ordinary APIs keep their signatures. Paths and IDs belong to the des
 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. Offline operations fail; observations recover. A lost result reports
-`outcome_unknown` and the destination server ID. Inspect before retrying a mutation.
+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.

-Host failures expose a stable `code`: `host_unknown`, `host_disabled`, `host_offline`,
+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`. Discovery does not grant execution authority.
+`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