jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: move daemon registry into host-runtime store

Browse Source
This commit is contained in:
Mohamed Boudra
2026-03-12 19:02:06 +07:00
parent 9f089be946
commit 554017b0b8
36 changed files with 1487 additions and 1194 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/src/runtime/host-runtime.test.ts
+239 -81
View File
@@ -5,7 +5,7 @@ import type {
FetchAgentsEntry,
FetchAgentsOptions,
} from "@server/client/daemon-client";
import type { HostConnection, HostProfile } from "@/contexts/daemon-registry-context";
import type { HostConnection, HostProfile } from "@/types/host-connection";
import { useSessionStore } from "@/stores/session-store";
import {
HostRuntimeController,
@@ -72,6 +72,10 @@ class FakeDaemonClient {
});
}
async ping(): Promise<{ rttMs: number }> {
return { rttMs: 0 };
}
setConnectionState(next: ConnectionState): void {
this.state = next;
if (next.status === "disconnected") {
@@ -194,7 +198,7 @@ function makeDeps(
createdClients.push(client);
return client as unknown as DaemonClient;
},
measureLatency: async ({ connection }) => {
connectToDaemon: async ({ host, connection }) => {
const value = latencyByConnectionId[connection.id];
if (value instanceof Error) {
throw value;
@@ -202,7 +206,16 @@ function makeDeps(
if (typeof value !== "number") {
throw new Error(`missing latency for ${connection.id}`);
}
return value;
const client = new FakeDaemonClient();
client.connectCalls = 1;
client.setConnectionState({ status: "connected" });
client.ping = async () => ({ rttMs: value });
createdClients.push(client);
return {
client: client as unknown as DaemonClient,
serverId: host.serverId,
hostname: host.label ?? null,
};
},
getClientId: async () => "cid_test_runtime",
};
@@ -222,8 +235,24 @@ function createDeferred<T>() {
};
}
function makeConnectedProbeClient(latencyMs: number): FakeDaemonClient {
const client = new FakeDaemonClient();
client.connectCalls = 1;
client.setConnectionState({ status: "connected" });
client.ping = async () => ({ rttMs: latencyMs });
return client;
}
function clearProbeBackoff(controller: HostRuntimeController): void {
(
controller as unknown as {
connectionLastProbedAt: Map<string, number>;
}
).connectionLastProbedAt.clear();
}
describe("HostRuntimeController", () => {
it("keeps known hosts in connecting when client reports idle during connect", async () => {
it("keeps known hosts in connecting when a created client reports idle during connect", async () => {
const host = makeHost({
connections: [
{
@@ -236,7 +265,7 @@ describe("HostRuntimeController", () => {
const idleClient = new FakeDaemonClient();
const deps: HostRuntimeControllerDeps = {
createClient: () => idleClient as unknown as DaemonClient,
measureLatency: async () => {
connectToDaemon: async () => {
throw new Error("probe unavailable");
},
getClientId: async () => "cid_test_runtime",
@@ -251,7 +280,11 @@ describe("HostRuntimeController", () => {
// Intentionally do not emit a connected state; stay in idle.
};
await controller.start({ autoProbe: false });
await (
controller as unknown as {
switchToConnection: (input: { connectionId: string }) => Promise<void>;
}
).switchToConnection({ connectionId: "direct:lan:6767" });
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
expect(controller.getSnapshot().connectionStatus).toBe("connecting");
@@ -277,18 +310,24 @@ describe("HostRuntimeController", () => {
seenClientIds.push(clientId);
return fakeClient as unknown as DaemonClient;
},
measureLatency: async () => 10,
connectToDaemon: async () => {
throw new Error("probe unavailable");
},
getClientId: async () => "cid_runtime_stable",
},
});
await controller.start({ autoProbe: false });
await (
controller as unknown as {
switchToConnection: (input: { connectionId: string }) => Promise<void>;
}
).switchToConnection({ connectionId: "direct:lan:6767" });
expect(seenClientIds).toEqual(["cid_runtime_stable"]);
expect(controller.getSnapshot().connectionStatus).toBe("online");
});
it("selects the lowest-latency connection on startup", async () => {
it("adopts the first successful probe on startup", async () => {
const host = makeHost({ preferredConnectionId: "direct:lan:6767" });
const clients: FakeDaemonClient[] = [];
const latencies: Record<string, number | Error> = {
@@ -303,10 +342,62 @@ describe("HostRuntimeController", () => {
await controller.start({ autoProbe: false });
const snapshot = controller.getSnapshot();
expect(snapshot.activeConnectionId).toBe("relay:relay.paseo.sh:443");
expect(snapshot.activeConnectionId).toBe("direct:lan:6767");
expect(snapshot.connectionStatus).toBe("online");
expect(clients).toHaveLength(1);
expect(clients).toHaveLength(2);
expect(snapshot.client).toBe(clients[0] as unknown as DaemonClient);
expect(clients[0]?.connectCalls).toBe(1);
expect(clients[1]?.closeCalls).toBe(1);
});
it("activates the first successful probe without waiting for slower probes", async () => {
const host = makeHost({ preferredConnectionId: "direct:lan:6767" });
const slowPing = createDeferred<number>();
const clients: FakeDaemonClient[] = [];
const controller = new HostRuntimeController({
host,
deps: {
createClient: () => {
throw new Error("should adopt the probe client");
},
connectToDaemon: async ({ host, connection }) => {
const client = makeConnectedProbeClient(
connection.id === "direct:lan:6767" ? 12 : 30
);
if (connection.id === "relay:relay.paseo.sh:443") {
client.ping = async () => ({ rttMs: await slowPing.promise });
}
clients.push(client);
return {
client: client as unknown as DaemonClient,
serverId: host.serverId,
hostname: host.label ?? null,
};
},
getClientId: async () => "cid_test_runtime",
},
});
const probeCycle = controller.runProbeCycleNow();
const timeoutAt = Date.now() + 200;
while (Date.now() < timeoutAt) {
const snapshot = controller.getSnapshot();
if (
snapshot.activeConnectionId === "direct:lan:6767" &&
snapshot.connectionStatus === "online"
) {
break;
}
await new Promise((resolve) => setTimeout(resolve, 0));
}
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
expect(controller.getSnapshot().connectionStatus).toBe("online");
slowPing.resolve(30);
await probeCycle;
});
it("fails over when active connection becomes unavailable", async () => {
@@ -323,17 +414,19 @@ describe("HostRuntimeController", () => {
await controller.start({ autoProbe: false });
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
expect(clients).toHaveLength(1);
const initialClient = controller.getSnapshot().client;
expect(initialClient).toBeTruthy();
latencies["direct:lan:6767"] = new Error("direct unavailable");
latencies["relay:relay.paseo.sh:443"] = 42;
clearProbeBackoff(controller);
await controller.runProbeCycleNow();
const snapshot = controller.getSnapshot();
expect(snapshot.activeConnectionId).toBe("relay:relay.paseo.sh:443");
expect(snapshot.connectionStatus).toBe("online");
expect(clients).toHaveLength(2);
expect(clients[0]?.closeCalls).toBe(1);
expect(snapshot.client).not.toBe(initialClient);
expect((initialClient as unknown as FakeDaemonClient | null)?.closeCalls).toBe(1);
});
it("switches only after the faster alternative wins consecutive probes", async () => {
@@ -353,15 +446,24 @@ describe("HostRuntimeController", () => {
latencies["direct:lan:6767"] = 95;
latencies["relay:relay.paseo.sh:443"] = 30;
clearProbeBackoff(controller);
await controller.runProbeCycleNow();
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
clearProbeBackoff(controller);
await controller.runProbeCycleNow();
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
await controller.runProbeCycleNow();
expect(controller.getSnapshot().activeConnectionId).toBe("relay:relay.paseo.sh:443");
expect(clients).toHaveLength(2);
let switched = controller.getSnapshot().activeConnectionId === "relay:relay.paseo.sh:443";
for (let index = 0; index < 6 && !switched; index += 1) {
clearProbeBackoff(controller);
await controller.runProbeCycleNow();
switched =
controller.getSnapshot().activeConnectionId ===
"relay:relay.paseo.sh:443";
}
expect(switched).toBe(true);
expect(controller.getSnapshot().client).not.toBeNull();
});
it("does not switch on a transient latency spike", async () => {
@@ -381,24 +483,35 @@ describe("HostRuntimeController", () => {
latencies["direct:lan:6767"] = 100;
latencies["relay:relay.paseo.sh:443"] = 20;
clearProbeBackoff(controller);
await controller.runProbeCycleNow();
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
latencies["direct:lan:6767"] = 20;
latencies["relay:relay.paseo.sh:443"] = 90;
clearProbeBackoff(controller);
await controller.runProbeCycleNow();
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
latencies["direct:lan:6767"] = 100;
latencies["relay:relay.paseo.sh:443"] = 20;
clearProbeBackoff(controller);
await controller.runProbeCycleNow();
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
clearProbeBackoff(controller);
await controller.runProbeCycleNow();
expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767");
await controller.runProbeCycleNow();
expect(controller.getSnapshot().activeConnectionId).toBe("relay:relay.paseo.sh:443");
let switched = controller.getSnapshot().activeConnectionId === "relay:relay.paseo.sh:443";
for (let index = 0; index < 6 && !switched; index += 1) {
clearProbeBackoff(controller);
await controller.runProbeCycleNow();
switched =
controller.getSnapshot().activeConnectionId ===
"relay:relay.paseo.sh:443";
}
expect(switched).toBe(true);
});
it("exposes one snapshot with active connection and status from same source", async () => {
@@ -604,7 +717,11 @@ describe("HostRuntimeController", () => {
createdClients.push(client);
return client as unknown as DaemonClient;
},
measureLatency: async () => 10,
connectToDaemon: async ({ host }) => ({
client: makeConnectedProbeClient(10) as unknown as DaemonClient,
serverId: host.serverId,
hostname: host.label ?? null,
}),
getClientId: async () => "cid_test_runtime",
};
const controller = new HostRuntimeController({
@@ -681,21 +798,32 @@ describe("HostRuntimeController", () => {
host,
deps: {
createClient: () => new FakeDaemonClient() as unknown as DaemonClient,
measureLatency: async () => {
connectToDaemon: async ({ host }) => {
probeCalls += 1;
if (probeCalls === 1) {
return await slowProbe.promise;
}
if (probeCalls === 2) {
return await fastProbe.promise;
}
throw new Error("unexpected probe call");
const client = new FakeDaemonClient();
client.connectCalls = 1;
client.setConnectionState({ status: "connected" });
client.ping = async () => {
if (probeCalls === 1) {
return { rttMs: await slowProbe.promise };
}
if (probeCalls === 2) {
return { rttMs: await fastProbe.promise };
}
throw new Error("unexpected probe call");
};
return {
client: client as unknown as DaemonClient,
serverId: host.serverId,
hostname: host.label ?? null,
};
},
getClientId: async () => "cid_test_runtime",
},
});
const first = controller.runProbeCycleNow();
clearProbeBackoff(controller);
const second = controller.runProbeCycleNow();
fastProbe.resolve(12);
@@ -719,7 +847,7 @@ describe("HostRuntimeController", () => {
});
});
it("keeps active client generation stable while overlapping probes run", async () => {
it("keeps active client generation stable during background probe cycles", async () => {
const host = makeHost({
connections: [
{
@@ -729,10 +857,7 @@ describe("HostRuntimeController", () => {
},
],
});
const slowProbe = createDeferred<number>();
const fastProbe = createDeferred<number>();
const createdClients: FakeDaemonClient[] = [];
let probeCalls = 0;
const controller = new HostRuntimeController({
host,
@@ -742,18 +867,13 @@ describe("HostRuntimeController", () => {
createdClients.push(client);
return client as unknown as DaemonClient;
},
measureLatency: async () => {
probeCalls += 1;
if (probeCalls === 1) {
return 10;
}
if (probeCalls === 2) {
return await slowProbe.promise;
}
if (probeCalls === 3) {
return await fastProbe.promise;
}
return 10;
connectToDaemon: async ({ host }) => {
const client = makeConnectedProbeClient(10);
return {
client: client as unknown as DaemonClient,
serverId: host.serverId,
hostname: host.label ?? null,
};
},
getClientId: async () => "cid_test_runtime",
},
@@ -763,24 +883,13 @@ describe("HostRuntimeController", () => {
const activeClientBeforeProbes = controller.getSnapshot().client;
const generationBeforeProbes = controller.getSnapshot().clientGeneration;
const first = controller.runProbeCycleNow();
const second = controller.runProbeCycleNow();
fastProbe.resolve(12);
await second;
clearProbeBackoff(controller);
await controller.runProbeCycleNow();
expect(controller.getSnapshot().client).toBe(activeClientBeforeProbes);
expect(controller.getSnapshot().clientGeneration).toBe(
generationBeforeProbes
);
slowProbe.resolve(999);
await first;
expect(controller.getSnapshot().client).toBe(activeClientBeforeProbes);
expect(controller.getSnapshot().clientGeneration).toBe(
generationBeforeProbes
);
expect(createdClients).toHaveLength(1);
expect(createdClients[0]?.closeCalls).toBe(0);
expect(createdClients).toHaveLength(0);
});
});
@@ -796,10 +905,15 @@ describe("HostRuntimeStore", () => {
],
});
const fakeClient = new FakeDaemonClient();
fakeClient.setConnectionState({ status: "connected" });
const store = new HostRuntimeStore({
deps: {
createClient: () => fakeClient as unknown as DaemonClient,
measureLatency: async () => 5,
connectToDaemon: async ({ host }) => ({
client: fakeClient as unknown as DaemonClient,
serverId: host.serverId,
hostname: host.label ?? null,
}),
getClientId: async () => "cid_test_runtime",
},
});
@@ -832,9 +946,9 @@ describe("HostRuntimeStore", () => {
useSessionStore.getState().clearSession(host.serverId);
});
it("defers directory bootstrap until session store is initialized for that server", async () => {
it("bootstraps agent directory immediately when connection goes online (no session required)", async () => {
const host = makeHost({
serverId: "srv_deferred",
serverId: "srv_no_session",
connections: [
{
id: "direct:lan:6767",
@@ -844,26 +958,22 @@ describe("HostRuntimeStore", () => {
],
});
const fakeClient = new FakeDaemonClient();
fakeClient.setConnectionState({ status: "connected" });
const store = new HostRuntimeStore({
deps: {
createClient: () => fakeClient as unknown as DaemonClient,
measureLatency: async () => 5,
connectToDaemon: async ({ host }) => ({
client: fakeClient as unknown as DaemonClient,
serverId: host.serverId,
hostname: host.label ?? null,
}),
getClientId: async () => "cid_test_runtime",
},
});
store.syncHosts([host]);
await new Promise((resolve) => setTimeout(resolve, 250));
expect(fakeClient.fetchAgentsCalls).toHaveLength(0);
useSessionStore.getState().initializeSession(
host.serverId,
fakeClient as unknown as DaemonClient,
null as any
);
const timeoutAt = Date.now() + 600;
const timeoutAt = Date.now() + 200;
while (fakeClient.fetchAgentsCalls.length === 0 && Date.now() < timeoutAt) {
await new Promise((resolve) => setTimeout(resolve, 0));
}
@@ -872,12 +982,11 @@ describe("HostRuntimeStore", () => {
expect(fakeClient.fetchAgentsCalls[0]).toEqual({
filter: { includeArchived: true },
sort: [{ key: "updated_at", direction: "desc" }],
subscribe: { subscriptionId: "app:srv_deferred" },
subscribe: { subscriptionId: "app:srv_no_session" },
page: { limit: 200 },
});
store.syncHosts([]);
useSessionStore.getState().clearSession(host.serverId);
});
it("fetches all pages during bootstrap so older workspace agents are present", async () => {
@@ -892,6 +1001,7 @@ describe("HostRuntimeStore", () => {
],
});
const fakeClient = new FakeDaemonClient();
fakeClient.setConnectionState({ status: "connected" });
fakeClient.fetchAgentsResponses.push(
makeFetchAgentsPayload({
entries: [
@@ -923,7 +1033,11 @@ describe("HostRuntimeStore", () => {
const store = new HostRuntimeStore({
deps: {
createClient: () => fakeClient as unknown as DaemonClient,
measureLatency: async () => 5,
connectToDaemon: async ({ host }) => ({
client: fakeClient as unknown as DaemonClient,
serverId: host.serverId,
hostname: host.label ?? null,
}),
getClientId: async () => "cid_test_runtime",
},
});
@@ -991,10 +1105,15 @@ describe("HostRuntimeStore", () => {
],
});
const fakeClient = new FakeDaemonClient();
fakeClient.setConnectionState({ status: "connected" });
const store = new HostRuntimeStore({
deps: {
createClient: () => fakeClient as unknown as DaemonClient,
measureLatency: async () => 5,
connectToDaemon: async ({ host }) => ({
client: fakeClient as unknown as DaemonClient,
serverId: host.serverId,
hostname: host.label ?? null,
}),
getClientId: async () => "cid_test_runtime",
},
});
@@ -1041,7 +1160,7 @@ describe("HostRuntimeStore", () => {
useSessionStore.getState().clearSession(host.serverId);
});
it("surfaces startup failures as error instead of leaving host idle", async () => {
it("records unavailable startup probes when no connection can be established", async () => {
const host = makeHost({
connections: [
{
@@ -1056,7 +1175,7 @@ describe("HostRuntimeStore", () => {
createClient: () => {
throw new Error("create client failed");
},
measureLatency: async () => {
connectToDaemon: async () => {
throw new Error("probe unavailable");
},
getClientId: async () => "cid_test_runtime",
@@ -1066,12 +1185,51 @@ describe("HostRuntimeStore", () => {
store.syncHosts([host]);
let snapshot = store.getSnapshot(host.serverId);
const timeoutAt = Date.now() + 100;
while (snapshot?.connectionStatus !== "error" && Date.now() < timeoutAt) {
while (
snapshot?.probeByConnectionId.get("direct:lan:6767")?.status !== "unavailable" &&
Date.now() < timeoutAt
) {
await new Promise((resolve) => setTimeout(resolve, 0));
snapshot = store.getSnapshot(host.serverId);
}
expect(snapshot?.connectionStatus).toBe("error");
expect(snapshot?.lastError).toBe("create client failed");
expect(snapshot?.connectionStatus).toBe("connecting");
expect(snapshot?.lastError).toBeNull();
expect(snapshot?.probeByConnectionId.get("direct:lan:6767")).toEqual({
status: "unavailable",
latencyMs: null,
});
});
it("renameHost updates label in memory", async () => {
const store = new HostRuntimeStore({
deps: {
createClient: () => new FakeDaemonClient() as unknown as DaemonClient,
connectToDaemon: async ({ host }) => ({
client: makeConnectedProbeClient(5) as unknown as DaemonClient,
serverId: host.serverId,
hostname: host.label ?? null,
}),
getClientId: async () => "cid_test_runtime",
},
});
// upsertDirectConnection goes through setHostsAndSync, which both sets
// this.hosts and syncs controllers — matching the real init path.
await store.upsertDirectConnection({
serverId: "srv_rename",
endpoint: "lan:6767",
label: "old name",
});
expect(store.getHosts().find((h) => h.serverId === "srv_rename")?.label).toBe("old name");
// persistHosts may throw in test env (no AsyncStorage/window), but the
// in-memory state should still be updated by setHostsAndSync.
await store.renameHost("srv_rename", "new name").catch(() => undefined);
const renamed = store.getHosts().find((h) => h.serverId === "srv_rename");
expect(renamed?.label).toBe("new name");
store.syncHosts([]);
});
});
packages/app/src/runtime/host-runtime.ts
+612 -106
View File
@@ -1,16 +1,30 @@
import { useSyncExternalStore } from "react";
import { useSyncExternalStore, useCallback, useMemo } from "react";
import AsyncStorage from "@react-native-async-storage/async-storage";
import {
DaemonClient,
type ConnectionState,
type DaemonClientDiagnosticsEvent,
type FetchAgentsOptions,
} from "@server/client/daemon-client";
import type { HostConnection, HostProfile } from "@/contexts/daemon-registry-context";
import {
connectionFromListen,
normalizeStoredHostProfile,
upsertHostConnectionInProfiles,
registryHasDirectEndpoint,
type HostConnection,
type HostProfile,
} from "@/types/host-connection";
import { decodeOfferFragmentPayload, normalizeHostPort } from "@/utils/daemon-endpoints";
import { ConnectionOfferSchema, type ConnectionOffer } from "@server/shared/connection-offer";
import {
shouldUseManagedDesktopDaemon,
startManagedDaemon,
} from "@/desktop/managed-runtime/managed-runtime";
import { connectToDaemon } from "@/utils/test-daemon-connection";
import {
buildDaemonWebSocketUrl,
buildRelayWebSocketUrl,
} from "@/utils/daemon-endpoints";
import { measureConnectionLatency } from "@/utils/test-daemon-connection";
import { getOrCreateClientId } from "@/utils/client-id";
import {
selectBestConnection,
@@ -132,15 +146,23 @@ export type HostRuntimeControllerDeps = {
clientId: string;
runtimeGeneration: number;
}) => DaemonClient;
measureLatency: (input: {
connectToDaemon: (input: {
host: HostProfile;
connection: HostConnection;
}) => Promise<number>;
}) => Promise<{
client: DaemonClient;
serverId: string;
hostname: string | null;
}>;
getClientId: () => Promise<string>;
};
export type HostRuntimeStartOptions = {
autoProbe?: boolean;
initialConnection?: {
connectionId: string;
existingClient: DaemonClient;
};
};
const PROBE_TICK_MS = 2_000;
@@ -149,7 +171,7 @@ const PROBE_MAX_BACKOFF_MS = 30_000;
const ADAPTIVE_SWITCH_THRESHOLD_MS = 40;
const ADAPTIVE_SWITCH_CONSECUTIVE_PROBES = 3;
const DEFAULT_AGENT_DIRECTORY_PAGE_LIMIT = 200;
const AGENT_DIRECTORY_SESSION_RETRY_MS = 150;
const DEFAULT_AGENT_DIRECTORY_SORT: NonNullable<FetchAgentsOptions["sort"]> = [
{ key: "updated_at", direction: "desc" },
];
@@ -493,8 +515,8 @@ function createDefaultDeps(): HostRuntimeControllerDeps {
},
});
},
measureLatency: ({ host, connection }) =>
measureConnectionLatency(connection, { serverId: host.serverId }),
connectToDaemon: ({ host, connection }) =>
connectToDaemon(connection, { serverId: host.serverId }),
getClientId: () => getOrCreateClientId(),
};
}
@@ -560,6 +582,12 @@ export class HostRuntimeController {
}
this.started = true;
this.trackConnectionFirstSeen();
if (options?.initialConnection) {
await this.switchToConnection({
connectionId: options.initialConnection.connectionId,
existingClient: options.initialConnection.existingClient,
});
}
await this.runProbeCycleNow();
if (options?.autoProbe !== false) {
this.probeIntervalHandle = setInterval(() => {
@@ -645,6 +673,13 @@ export class HostRuntimeController {
});
}
async activateConnection(input: {
connectionId: string;
existingClient: DaemonClient;
}): Promise<void> {
await this.switchToConnection(input);
}
async runProbeCycleNow(): Promise<void> {
const requestVersion = ++this.probeRequestVersion;
if (this.host.connections.length === 0) {
@@ -679,73 +714,103 @@ export class HostRuntimeController {
}
const probeByConnectionId = new Map(this.snapshot.probeByConnectionId);
await Promise.all(
connectionsToProbe.map(async (connection) => {
this.connectionLastProbedAt.set(connection.id, performance.now());
try {
const latencyMs = await this.deps.measureLatency({
host: this.host,
connection,
for (const connection of connectionsToProbe) {
this.connectionLastProbedAt.set(connection.id, performance.now());
probeByConnectionId.set(connection.id, {
status: "pending",
latencyMs: null,
});
}
this.updateSnapshot({ probeByConnectionId: new Map(probeByConnectionId) });
let remaining = connectionsToProbe.length;
let activationLock: Promise<void> | null = null;
const publishProbeState = (): void => {
if (!this.isCurrentProbeRequest(requestVersion)) {
return;
}
this.updateSnapshot({ probeByConnectionId: new Map(probeByConnectionId) });
};
const maybeActivateFirstAvailable = async (
connectionId: string,
client: DaemonClient
): Promise<boolean> => {
while (!this.snapshot.activeConnectionId) {
if (!activationLock) {
activationLock = this.switchToConnection({
connectionId,
expectedProbeVersion: requestVersion,
existingClient: client,
}).finally(() => {
activationLock = null;
});
probeByConnectionId.set(connection.id, {
status: "available",
latencyMs,
});
} catch {
probeByConnectionId.set(connection.id, {
status: "unavailable",
latencyMs: null,
await activationLock;
return this.snapshot.activeConnectionId === connectionId;
}
await activationLock;
}
return false;
};
const finalizeProbeCycle = async (): Promise<void> => {
if (remaining > 0 || !this.isCurrentProbeRequest(requestVersion)) {
return;
}
const currentActiveConnectionId = this.snapshot.activeConnectionId;
const activeProbe = currentActiveConnectionId
? probeByConnectionId.get(currentActiveConnectionId)
: null;
if (
!currentActiveConnectionId ||
!findConnectionById(this.host, currentActiveConnectionId)
) {
const nextConnectionId = selectBestConnection({
candidates: buildConnectionCandidates(this.host),
probeByConnectionId,
});
if (nextConnectionId) {
await this.switchToConnection({
connectionId: nextConnectionId,
expectedProbeVersion: requestVersion,
});
}
})
);
if (!this.isCurrentProbeRequest(requestVersion)) {
return;
}
this.updateSnapshot({ probeByConnectionId });
const currentActiveConnectionId = this.snapshot.activeConnectionId;
const activeProbe = currentActiveConnectionId
? probeByConnectionId.get(currentActiveConnectionId)
: null;
if (!currentActiveConnectionId || !findConnectionById(this.host, currentActiveConnectionId)) {
const nextConnectionId = selectBestConnection({
candidates: buildConnectionCandidates(this.host),
probeByConnectionId,
});
if (nextConnectionId) {
await this.switchToConnection({
connectionId: nextConnectionId,
expectedProbeVersion: requestVersion,
});
return;
}
return;
}
if (activeProbe?.status === "unavailable") {
const nextConnectionId = selectBestConnection({
candidates: buildConnectionCandidates(this.host),
probeByConnectionId,
});
if (nextConnectionId && nextConnectionId !== currentActiveConnectionId) {
await this.switchToConnection({
connectionId: nextConnectionId,
expectedProbeVersion: requestVersion,
if (activeProbe?.status === "unavailable") {
const nextConnectionId = selectBestConnection({
candidates: buildConnectionCandidates(this.host),
probeByConnectionId,
});
if (nextConnectionId && nextConnectionId !== currentActiveConnectionId) {
await this.switchToConnection({
connectionId: nextConnectionId,
expectedProbeVersion: requestVersion,
});
}
this.switchCandidateConnectionId = null;
this.switchCandidateHitCount = 0;
return;
}
if (!activeProbe || activeProbe.status !== "available") {
return;
}
this.switchCandidateConnectionId = null;
this.switchCandidateHitCount = 0;
return;
}
if (activeProbe && activeProbe.status === "available") {
const available = Array.from(probeByConnectionId.entries())
.filter(([, probe]) => probe.status === "available")
.filter(
(
entry
): entry is [string, Extract<ConnectionProbeState, { status: "available" }>] =>
entry[1].status === "available"
)
.map(([connectionId, probe]) => ({
connectionId,
latencyMs: (probe as { status: "available"; latencyMs: number }).latencyMs,
latencyMs: probe.latencyMs,
}))
.sort((left, right) => left.latencyMs - right.latencyMs);
@@ -756,8 +821,7 @@ export class HostRuntimeController {
return;
}
const activeLatency = activeProbe.latencyMs;
const improvement = activeLatency - fastest.latencyMs;
const improvement = activeProbe.latencyMs - fastest.latencyMs;
if (improvement < ADAPTIVE_SWITCH_THRESHOLD_MS) {
this.switchCandidateConnectionId = null;
this.switchCandidateHitCount = 0;
@@ -772,8 +836,7 @@ export class HostRuntimeController {
}
if (
this.switchCandidateHitCount >=
ADAPTIVE_SWITCH_CONSECUTIVE_PROBES
this.switchCandidateHitCount >= ADAPTIVE_SWITCH_CONSECUTIVE_PROBES
) {
this.switchCandidateConnectionId = null;
this.switchCandidateHitCount = 0;
@@ -782,7 +845,63 @@ export class HostRuntimeController {
expectedProbeVersion: requestVersion,
});
}
}
};
await new Promise<void>((resolve) => {
const settleProbe = (): void => {
remaining -= 1;
void finalizeProbeCycle().finally(() => {
if (remaining === 0) {
resolve();
}
});
};
for (const connection of connectionsToProbe) {
void (async () => {
let connectedClient: DaemonClient | null = null;
let handedOffClient = false;
try {
const { client } = await this.deps.connectToDaemon({
host: this.host,
connection,
});
connectedClient = client;
if (!this.isCurrentProbeRequest(requestVersion)) {
return;
}
const activated = await maybeActivateFirstAvailable(connection.id, client);
handedOffClient = activated;
const { rttMs } = await client.ping({ timeoutMs: 5000 });
if (!this.isCurrentProbeRequest(requestVersion)) {
return;
}
probeByConnectionId.set(connection.id, {
status: "available",
latencyMs: rttMs,
});
publishProbeState();
} catch {
if (this.isCurrentProbeRequest(requestVersion)) {
probeByConnectionId.set(connection.id, {
status: "unavailable",
latencyMs: null,
});
publishProbeState();
}
} finally {
if (connectedClient && !handedOffClient) {
await connectedClient.close().catch(() => undefined);
}
settleProbe();
}
})();
}
});
}
private updateSnapshot(
@@ -880,13 +999,20 @@ export class HostRuntimeController {
private async switchToConnection(input: {
connectionId: string;
expectedProbeVersion?: number;
existingClient?: DaemonClient;
}): Promise<void> {
if (!this.canProceedForProbe(input.expectedProbeVersion)) {
if (input.existingClient) {
await input.existingClient.close().catch(() => undefined);
}
return;
}
const { connectionId, expectedProbeVersion } = input;
const { connectionId, expectedProbeVersion, existingClient } = input;
const connection = findConnectionById(this.host, connectionId);
if (!connection) {
if (existingClient) {
await existingClient.close().catch(() => undefined);
}
return;
}
const requestVersion = ++this.switchRequestVersion;
@@ -895,6 +1021,9 @@ export class HostRuntimeController {
try {
clientId = await this.resolveClientId();
} catch (error) {
if (existingClient) {
await existingClient.close().catch(() => undefined);
}
if (!this.isCurrentSwitchRequest(requestVersion)) {
return;
}
@@ -910,9 +1039,15 @@ export class HostRuntimeController {
}
if (!this.isCurrentSwitchRequest(requestVersion)) {
if (existingClient) {
await existingClient.close().catch(() => undefined);
}
return;
}
if (!this.canProceedForProbe(expectedProbeVersion)) {
if (existingClient) {
await existingClient.close().catch(() => undefined);
}
return;
}
@@ -926,14 +1061,20 @@ export class HostRuntimeController {
await previousClient.close().catch(() => undefined);
}
if (!this.isCurrentSwitchRequest(requestVersion)) {
if (existingClient) {
await existingClient.close().catch(() => undefined);
}
return;
}
if (!this.canProceedForProbe(expectedProbeVersion)) {
if (existingClient) {
await existingClient.close().catch(() => undefined);
}
return;
}
const nextGeneration = this.snapshot.clientGeneration + 1;
const client = this.deps.createClient({
const client = existingClient ?? this.deps.createClient({
host: this.host,
connection,
clientId,
@@ -1000,7 +1141,9 @@ export class HostRuntimeController {
});
try {
await client.connect();
if (!existingClient) {
await client.connect();
}
} catch (error) {
if (
!this.isCurrentSwitchRequest(requestVersion) ||
@@ -1030,18 +1173,25 @@ export class HostRuntimeController {
}
}
const REGISTRY_STORAGE_KEY = "@paseo:daemon-registry";
const DEFAULT_LOCALHOST_ENDPOINT = "localhost:6767";
const DEFAULT_LOCALHOST_BOOTSTRAP_KEY = "@paseo:default-localhost-bootstrap-v1";
const DEFAULT_LOCALHOST_BOOTSTRAP_TIMEOUT_MS = 2500;
const E2E_STORAGE_KEY = "@paseo:e2e";
export class HostRuntimeStore {
private controllers = new Map<string, HostRuntimeController>();
private serverListeners = new Map<string, Set<() => void>>();
private globalListeners = new Set<() => void>();
private hostListListeners = new Set<() => void>();
private version = 0;
private hostListVersion = 0;
private hosts: HostProfile[] = [];
private deps: HostRuntimeControllerDeps;
private lastConnectionStatusByServer = new Map<string, HostRuntimeConnectionStatus>();
private agentDirectoryBootstrapInFlight = new Map<string, Promise<void>>();
private agentDirectorySessionRetryTimerByServer = new Map<
string,
ReturnType<typeof setTimeout>
>();
private storageLoaded = false;
private bootstrapAttempted = false;
constructor(input?: {
deps?: HostRuntimeControllerDeps;
@@ -1049,7 +1199,335 @@ export class HostRuntimeStore {
this.deps = input?.deps ?? createDefaultDeps();
}
syncHosts(hosts: HostProfile[]): void {
// --- Host registry ---
getHosts(): HostProfile[] {
return this.hosts;
}
subscribeHostList(listener: () => void): () => void {
this.hostListListeners.add(listener);
return () => {
this.hostListListeners.delete(listener);
};
}
getHostListVersion(): number {
return this.hostListVersion;
}
async loadFromStorage(): Promise<void> {
if (this.storageLoaded) {
return;
}
this.storageLoaded = true;
try {
const stored = await AsyncStorage.getItem(REGISTRY_STORAGE_KEY);
if (!stored) {
return;
}
const parsed = JSON.parse(stored) as unknown;
if (!Array.isArray(parsed)) {
return;
}
const profiles = parsed
.map((entry) => normalizeStoredHostProfile(entry))
.filter((entry): entry is HostProfile => entry !== null);
this.hosts = profiles;
this.syncHosts(profiles);
this.emitHostList();
} catch (error) {
console.error("[HostRuntime] Failed to load host registry from storage", error);
}
}
async bootstrap(): Promise<void> {
if (this.bootstrapAttempted) {
return;
}
this.bootstrapAttempted = true;
try {
const isE2E = await AsyncStorage.getItem(E2E_STORAGE_KEY);
if (isE2E) {
return;
}
} catch {
return;
}
if (shouldUseManagedDesktopDaemon()) {
await this.bootstrapDesktop();
} else {
await this.bootstrapLocalhost();
}
}
private async bootstrapDesktop(): Promise<void> {
const t0 = performance.now();
const elapsed = () => Math.round(performance.now() - t0);
try {
console.info("[HostRuntime] desktop bootstrap: starting", { elapsedMs: elapsed() });
await Promise.allSettled([
(async () => {
console.info("[HostRuntime] desktop bootstrap: startManagedDaemon", { elapsedMs: elapsed() });
const daemon = await startManagedDaemon();
console.info("[HostRuntime] desktop bootstrap: managed daemon ready", { elapsedMs: elapsed(), serverId: daemon.serverId });
if (!daemon.serverId) {
return;
}
const connection = connectionFromListen(daemon.listen);
if (!connection) {
return;
}
await this.upsertHostConnection({
serverId: daemon.serverId,
label: daemon.hostname ?? undefined,
connection,
});
console.info("[HostRuntime] desktop bootstrap: host connection upserted", { elapsedMs: elapsed() });
})().catch((error) => {
console.warn("[HostRuntime] Failed to bootstrap desktop daemon connection", { elapsedMs: elapsed() }, error);
}),
(async () => {
console.info("[HostRuntime] desktop bootstrap: probing localhost", { elapsedMs: elapsed() });
const { client, serverId, hostname } = await connectToDaemon(
{
id: `bootstrap:${DEFAULT_LOCALHOST_ENDPOINT}`,
type: "directTcp",
endpoint: DEFAULT_LOCALHOST_ENDPOINT,
},
{ timeoutMs: DEFAULT_LOCALHOST_BOOTSTRAP_TIMEOUT_MS }
);
console.info("[HostRuntime] desktop bootstrap: localhost probe succeeded", { elapsedMs: elapsed(), serverId });
await this.upsertDirectConnection({
serverId,
endpoint: DEFAULT_LOCALHOST_ENDPOINT,
label: hostname ?? undefined,
existingClient: client,
});
console.info("[HostRuntime] desktop bootstrap: direct connection upserted", { elapsedMs: elapsed() });
})().catch(() => {
console.info("[HostRuntime] desktop bootstrap: localhost probe failed", { elapsedMs: elapsed() });
}),
]);
} catch (error) {
console.warn("[HostRuntime] Failed to bootstrap desktop startup host connections", error);
}
}
private async bootstrapLocalhost(): Promise<void> {
try {
const alreadyHandled = await AsyncStorage.getItem(DEFAULT_LOCALHOST_BOOTSTRAP_KEY);
if (alreadyHandled) {
return;
}
if (registryHasDirectEndpoint(this.hosts, DEFAULT_LOCALHOST_ENDPOINT)) {
await AsyncStorage.setItem(DEFAULT_LOCALHOST_BOOTSTRAP_KEY, "1");
return;
}
try {
const { client, serverId, hostname } = await connectToDaemon(
{
id: `bootstrap:${DEFAULT_LOCALHOST_ENDPOINT}`,
type: "directTcp",
endpoint: DEFAULT_LOCALHOST_ENDPOINT,
},
{ timeoutMs: DEFAULT_LOCALHOST_BOOTSTRAP_TIMEOUT_MS }
);
await this.upsertDirectConnection({
serverId,
endpoint: DEFAULT_LOCALHOST_ENDPOINT,
label: hostname ?? undefined,
existingClient: client,
});
await AsyncStorage.setItem(DEFAULT_LOCALHOST_BOOTSTRAP_KEY, "1");
} catch {
// Best-effort bootstrap only
}
} catch (error) {
console.warn("[HostRuntime] Failed to bootstrap host connections", error);
}
}
async upsertDirectConnection(input: {
serverId: string;
endpoint: string;
label?: string;
existingClient?: DaemonClient;
}): Promise<HostProfile> {
const endpoint = normalizeHostPort(input.endpoint);
return this.upsertHostConnection({
serverId: input.serverId,
label: input.label,
connection: {
id: `direct:${endpoint}`,
type: "directTcp",
endpoint,
},
existingClient: input.existingClient,
});
}
async upsertRelayConnection(input: {
serverId: string;
relayEndpoint: string;
daemonPublicKeyB64: string;
label?: string;
}): Promise<HostProfile> {
const relayEndpoint = normalizeHostPort(input.relayEndpoint);
const daemonPublicKeyB64 = input.daemonPublicKeyB64.trim();
if (!daemonPublicKeyB64) {
throw new Error("daemonPublicKeyB64 is required");
}
return this.upsertHostConnection({
serverId: input.serverId,
label: input.label,
connection: {
id: `relay:${relayEndpoint}`,
type: "relay",
relayEndpoint,
daemonPublicKeyB64,
},
});
}
async upsertConnectionFromOffer(offer: ConnectionOffer): Promise<HostProfile> {
return this.upsertRelayConnection({
serverId: offer.serverId,
relayEndpoint: offer.relay.endpoint,
daemonPublicKeyB64: offer.daemonPublicKeyB64,
});
}
async upsertConnectionFromOfferUrl(offerUrlOrFragment: string): Promise<HostProfile> {
const marker = "#offer=";
const idx = offerUrlOrFragment.indexOf(marker);
if (idx === -1) {
throw new Error("Missing #offer= fragment");
}
const encoded = offerUrlOrFragment.slice(idx + marker.length).trim();
if (!encoded) {
throw new Error("Offer payload is empty");
}
const payload = decodeOfferFragmentPayload(encoded);
const offer = ConnectionOfferSchema.parse(payload);
return this.upsertConnectionFromOffer(offer);
}
async renameHost(serverId: string, label: string): Promise<void> {
const next = this.hosts.map((h) =>
h.serverId === serverId
? { ...h, label, updatedAt: new Date().toISOString() }
: h
);
this.setHostsAndSync(next);
await this.persistHosts();
}
async removeHost(serverId: string): Promise<void> {
const remaining = this.hosts.filter((daemon) => daemon.serverId !== serverId);
this.setHostsAndSync(remaining);
await this.persistHosts();
}
async removeConnection(serverId: string, connectionId: string): Promise<void> {
const now = new Date().toISOString();
const next = this.hosts
.map((daemon) => {
if (daemon.serverId !== serverId) return daemon;
const remaining = daemon.connections.filter((conn) => conn.id !== connectionId);
if (remaining.length === 0) {
return null;
}
const preferred =
daemon.preferredConnectionId === connectionId
? (remaining[0]?.id ?? null)
: daemon.preferredConnectionId;
return {
...daemon,
connections: remaining,
preferredConnectionId: preferred,
updatedAt: now,
} satisfies HostProfile;
})
.filter((entry): entry is HostProfile => entry !== null);
this.setHostsAndSync(next);
await this.persistHosts();
}
private async upsertHostConnection(input: {
serverId: string;
label?: string;
connection: HostConnection;
existingClient?: DaemonClient;
}): Promise<HostProfile> {
const now = new Date().toISOString();
const next = upsertHostConnectionInProfiles({
profiles: this.hosts,
serverId: input.serverId,
label: input.label,
connection: input.connection,
now,
});
this.setHostsAndSync(next, {
initialConnectionByServerId: input.existingClient
? new Map([
[
input.serverId,
{
connectionId: input.connection.id,
existingClient: input.existingClient,
},
],
])
: undefined,
});
void this.persistHosts();
return next.find((daemon) => daemon.serverId === input.serverId) as HostProfile;
}
private setHostsAndSync(
hosts: HostProfile[],
options?: {
initialConnectionByServerId?: Map<
string,
{ connectionId: string; existingClient: DaemonClient }
>;
}
): void {
this.hosts = hosts;
this.syncHosts(hosts, options);
this.emitHostList();
}
private async persistHosts(): Promise<void> {
try {
await AsyncStorage.setItem(REGISTRY_STORAGE_KEY, JSON.stringify(this.hosts));
} catch (error) {
console.error("[HostRuntime] Failed to persist host registry", error);
}
}
private emitHostList(): void {
this.hostListVersion += 1;
for (const listener of this.hostListListeners) {
listener();
}
}
syncHosts(
hosts: HostProfile[],
options?: {
initialConnectionByServerId?: Map<
string,
{ connectionId: string; existingClient: DaemonClient }
>;
}
): void {
const nextIds = new Set(hosts.map((host) => host.serverId));
for (const [serverId, controller] of this.controllers) {
if (nextIds.has(serverId)) {
@@ -1058,15 +1536,20 @@ export class HostRuntimeStore {
this.controllers.delete(serverId);
this.lastConnectionStatusByServer.delete(serverId);
this.agentDirectoryBootstrapInFlight.delete(serverId);
this.clearAgentDirectorySessionRetry(serverId);
void controller.stop();
this.emit(serverId);
}
for (const host of hosts) {
const initialConnection = options?.initialConnectionByServerId?.get(host.serverId);
const existing = this.controllers.get(host.serverId);
if (existing) {
void existing.updateHost(host);
if (initialConnection) {
void existing.activateConnection(initialConnection).catch(() => {
void initialConnection.existingClient.close().catch(() => undefined);
});
}
continue;
}
const controller = new HostRuntimeController({
@@ -1082,7 +1565,13 @@ export class HostRuntimeStore {
this.maybeAutoBootstrapAgentDirectory(host.serverId);
this.emit(host.serverId);
});
void controller.start().catch((error) => {
void controller.start({
...(initialConnection
? {
initialConnection,
}
: {}),
}).catch((error) => {
const message = error instanceof Error ? error.message : String(error);
controller.markStartupError(message);
});
@@ -1109,21 +1598,14 @@ export class HostRuntimeStore {
// Runtime owns directory bootstrap policy, including reconnect and delayed
// session initialization races.
if (snapshot.connectionStatus !== "online") {
this.clearAgentDirectorySessionRetry(serverId);
return;
}
if (!didTransitionOnline && snapshot.hasEverLoadedAgentDirectory) {
this.clearAgentDirectorySessionRetry(serverId);
return;
}
if (this.agentDirectoryBootstrapInFlight.has(serverId)) {
return;
}
if (!useSessionStore.getState().sessions[serverId]) {
this.scheduleAgentDirectorySessionRetry(serverId);
return;
}
this.clearAgentDirectorySessionRetry(serverId);
const bootstrap = Promise.resolve()
.then(() =>
@@ -1150,26 +1632,6 @@ export class HostRuntimeStore {
this.agentDirectoryBootstrapInFlight.set(serverId, bootstrap);
}
private scheduleAgentDirectorySessionRetry(serverId: string): void {
if (this.agentDirectorySessionRetryTimerByServer.has(serverId)) {
return;
}
const handle = setTimeout(() => {
this.agentDirectorySessionRetryTimerByServer.delete(serverId);
this.maybeAutoBootstrapAgentDirectory(serverId);
}, AGENT_DIRECTORY_SESSION_RETRY_MS);
this.agentDirectorySessionRetryTimerByServer.set(serverId, handle);
}
private clearAgentDirectorySessionRetry(serverId: string): void {
const handle = this.agentDirectorySessionRetryTimerByServer.get(serverId);
if (!handle) {
return;
}
clearTimeout(handle);
this.agentDirectorySessionRetryTimerByServer.delete(serverId);
}
getSnapshot(serverId: string): HostRuntimeSnapshot | null {
return this.controllers.get(serverId)?.getSnapshot() ?? null;
}
@@ -1384,3 +1846,47 @@ export function useHostRuntimeSession(serverId: string): {
isConnected: isHostRuntimeConnected(snapshot),
};
}
export function useHosts(): HostProfile[] {
const store = getHostRuntimeStore();
return useSyncExternalStore(
(onStoreChange) => store.subscribeHostList(onStoreChange),
() => store.getHosts(),
() => store.getHosts()
);
}
export interface HostMutations {
upsertDirectConnection: (input: {
serverId: string;
endpoint: string;
label?: string;
}) => Promise<HostProfile>;
upsertRelayConnection: (input: {
serverId: string;
relayEndpoint: string;
daemonPublicKeyB64: string;
label?: string;
}) => Promise<HostProfile>;
upsertConnectionFromOffer: (offer: ConnectionOffer) => Promise<HostProfile>;
upsertConnectionFromOfferUrl: (offerUrlOrFragment: string) => Promise<HostProfile>;
renameHost: (serverId: string, label: string) => Promise<void>;
removeHost: (serverId: string) => Promise<void>;
removeConnection: (serverId: string, connectionId: string) => Promise<void>;
}
export function useHostMutations(): HostMutations {
const store = getHostRuntimeStore();
return useMemo(
() => ({
upsertDirectConnection: (input) => store.upsertDirectConnection(input),
upsertRelayConnection: (input) => store.upsertRelayConnection(input),
upsertConnectionFromOffer: (offer) => store.upsertConnectionFromOffer(offer),
upsertConnectionFromOfferUrl: (url) => store.upsertConnectionFromOfferUrl(url),
renameHost: (serverId, label) => store.renameHost(serverId, label),
removeHost: (serverId) => store.removeHost(serverId),
removeConnection: (serverId, connectionId) => store.removeConnection(serverId, connectionId),
}),
[store]
);
}