From 569b34263d356d31fe84c592cf013fd716e736be Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Tue, 20 Oct 2026 07:54:42 +0200 Subject: [PATCH] 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. --- docs/connected-hosts.md | 11 +- packages/client/src/daemon-client.test.ts | 128 ++++++++++++++++++ packages/client/src/daemon-client.ts | 111 +++++++++------ packages/client/src/hosts/index.ts | 23 ++-- .../server/connected-hosts/hosts.e2e.test.ts | 101 ++++++++++---- .../src/server/connected-hosts/index.ts | 1 - .../daemon-e2e/relay-transport.e2e.test.ts | 3 +- public-docs/sdk/reference.md | 14 +- 8 files changed, 304 insertions(+), 88 deletions(-) diff --git a/docs/connected-hosts.md b/docs/connected-hosts.md index 22ba2443f9..088a5218a7 100644 --- a/docs/connected-hosts.md +++ b/docs/connected-hosts.md @@ -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 diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index 960b19f4d7..d30a2f4aad 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -3,6 +3,7 @@ import { afterEach, expect, expectTypeOf, test, vi } from "vitest"; import { z } from "zod"; import { DaemonClient, + DaemonConnectionError, supportsUsageReports, type DaemonClientTrace, type CreateAgentRequestOptions, @@ -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, + }); +}); diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index 254c9845bf..262cca26f0 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -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 { reject(error: Error): void; timeoutHandle: ReturnType | null; requestId?: string; + submitted: boolean; } interface WaitHandle { @@ -1006,11 +1004,16 @@ type CorrelatedResponsePayload = 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 { 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); }); diff --git a/packages/client/src/hosts/index.ts b/packages/client/src/hosts/index.ts index 94a7b9b8af..1baa9ebe43 100644 --- a/packages/client/src/hosts/index.ts +++ b/packages/client/src/hosts/index.ts @@ -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; } diff --git a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts index a3f3e3fd22..0005946316 100644 --- a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts +++ b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts @@ -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[] = []; + 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 } }); }); diff --git a/packages/server/src/server/connected-hosts/index.ts b/packages/server/src/server/connected-hosts/index.ts index 83f5173fab..1a8f7c2731 100644 --- a/packages/server/src/server/connected-hosts/index.ts +++ b/packages/server/src/server/connected-hosts/index.ts @@ -384,7 +384,6 @@ export class ConnectedHosts { input.serverId, randomUUID(), null, - { failure: null }, true, )({ url: "ws://host-link/ws" }); let closed = false; diff --git a/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts b/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts index 85030a4151..facf6dffd7 100644 --- a/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts @@ -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( diff --git a/public-docs/sdk/reference.md b/public-docs/sdk/reference.md index 5f140d2aa1..6e606e35ba 100644 --- a/public-docs/sdk/reference.md +++ b/public-docs/sdk/reference.md @@ -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