diff --git a/packages/app/src/components/multi-daemon-session-host.tsx b/packages/app/src/components/multi-daemon-session-host.tsx index 2ce0d1896..98bffd36c 100644 --- a/packages/app/src/components/multi-daemon-session-host.tsx +++ b/packages/app/src/components/multi-daemon-session-host.tsx @@ -1,155 +1,38 @@ +import { useEffect } from "react"; +import { AppState } from "react-native"; import { SessionProvider } from "@/contexts/session-context"; +import { useDaemonRegistry, type HostProfile } from "@/contexts/daemon-registry-context"; +import { useDaemonConnections } from "@/contexts/daemon-connections-context"; import { - useDaemonRegistry, - type HostConnection, - type HostProfile, -} from "@/contexts/daemon-registry-context"; -import { - buildDaemonWebSocketUrl, - buildRelayWebSocketUrl, -} from "@/utils/daemon-endpoints"; -import { measureConnectionLatency } from "@/utils/test-daemon-connection"; -import { - selectBestConnection, - type ConnectionCandidate, - type ConnectionProbeState, -} from "@/utils/connection-selection"; -import { useQueries } from "@tanstack/react-query"; -import { useMemo } from "react"; -import type { ActiveConnection } from "@/contexts/daemon-connections-context"; - -type Candidate = { - connection: HostConnection; - connectionId: string; - url: string; - activeConnection: ActiveConnection; - daemonPublicKeyB64?: string; -}; - -function sortConnectionsByPreference( - connections: T[], - preferredId: string | null -): T[] { - if (!preferredId) return connections; - const idx = connections.findIndex((c) => c.id === preferredId); - if (idx === -1) return connections; - return [connections[idx]!, ...connections.slice(0, idx), ...connections.slice(idx + 1)]; -} - -function buildCandidates(host: HostProfile): Candidate[] { - const preferred = host.preferredConnectionId ?? null; - const connections = sortConnectionsByPreference(host.connections, preferred); - - const out: Candidate[] = []; - - for (const conn of connections) { - if (conn.type === "direct") { - out.push({ - connection: conn, - connectionId: conn.id, - url: buildDaemonWebSocketUrl(conn.endpoint), - activeConnection: { - type: "direct", - endpoint: conn.endpoint, - display: conn.endpoint, - }, - }); - continue; - } - - out.push({ - connection: conn, - connectionId: conn.id, - url: buildRelayWebSocketUrl({ - endpoint: conn.relayEndpoint, - serverId: host.serverId, - }), - activeConnection: { type: "relay", endpoint: conn.relayEndpoint, display: "relay" }, - daemonPublicKeyB64: conn.daemonPublicKeyB64, - }); - } - - return out; -} + getHostRuntimeStore, + useHostRuntimeSession, +} from "@/runtime/host-runtime"; +import { toDaemonConnectionUpdateFromRuntime } from "@/runtime/host-runtime-bridge"; function ManagedDaemonSession({ daemon }: { daemon: HostProfile }) { - const candidates = useMemo(() => buildCandidates(daemon), [daemon]); - const latencyQueries = useQueries({ - queries: candidates.map((candidate) => ({ - queryKey: ["connection-selection-latency", daemon.serverId, candidate.connectionId], - queryFn: () => - measureConnectionLatency(candidate.connection, { - serverId: daemon.serverId, - }), - refetchInterval: 10_000, - staleTime: 9_000, - gcTime: 60_000, - retry: 1, - })), - }); + const { snapshot, client } = useHostRuntimeSession(daemon.serverId); + const { connectionStates, updateConnectionStatus } = useDaemonConnections(); + const hasConnectionRecord = connectionStates.has(daemon.serverId); - const probeByConnectionId = useMemo(() => { - const next = new Map(); + useEffect(() => { + if (!hasConnectionRecord) { + return; + } + updateConnectionStatus( + daemon.serverId, + toDaemonConnectionUpdateFromRuntime(snapshot) + ); + }, [daemon.serverId, hasConnectionRecord, snapshot, updateConnectionStatus]); - candidates.forEach((candidate, index) => { - const query = latencyQueries[index]; - if (!query) { - next.set(candidate.connectionId, { status: "pending", latencyMs: null }); - return; - } - - if (query.isSuccess && typeof query.data === "number") { - next.set(candidate.connectionId, { - status: "available", - latencyMs: query.data, - }); - return; - } - - if (query.isError) { - next.set(candidate.connectionId, { status: "unavailable", latencyMs: null }); - return; - } - - next.set(candidate.connectionId, { status: "pending", latencyMs: null }); - }); - - return next; - }, [candidates, latencyQueries]); - - const candidateInputs = useMemo( - () => - candidates.map((candidate) => ({ - connectionId: candidate.connectionId, - connection: candidate.connection, - })), - [candidates] - ); - - const activeConnectionId = useMemo( - () => - selectBestConnection({ - candidates: candidateInputs, - preferredConnectionId: daemon.preferredConnectionId, - probeByConnectionId, - }), - [candidateInputs, daemon.preferredConnectionId, probeByConnectionId] - ); - const active = - candidates.find((candidate) => candidate.connectionId === activeConnectionId) ?? null; - const activeUrl = active?.url ?? null; - - if (!activeUrl) { + if (!snapshot?.activeConnection || !client) { return null; } return ( {null} @@ -158,6 +41,26 @@ function ManagedDaemonSession({ daemon }: { daemon: HostProfile }) { export function MultiDaemonSessionHost() { const { daemons } = useDaemonRegistry(); + + useEffect(() => { + const runtime = getHostRuntimeStore(); + runtime.syncHosts(daemons); + }, [daemons]); + + useEffect(() => { + const runtime = getHostRuntimeStore(); + const subscription = AppState.addEventListener("change", (nextState) => { + if (nextState !== "active") { + return; + } + runtime.ensureConnectedAll(); + void runtime.runProbeCycleNow(); + }); + return () => { + subscription.remove(); + }; + }, []); + if (daemons.length === 0) { return null; } diff --git a/packages/app/src/contexts/daemon-connections-context.tsx b/packages/app/src/contexts/daemon-connections-context.tsx index d519d1e64..a3cf2f283 100644 --- a/packages/app/src/contexts/daemon-connections-context.tsx +++ b/packages/app/src/contexts/daemon-connections-context.tsx @@ -15,7 +15,7 @@ export type ConnectionState = export type ConnectionStatus = ConnectionState["status"]; -type ConnectionStateUpdate = +export type ConnectionStateUpdate = | { status: "idle" } | { status: "connecting"; activeConnection?: ActiveConnection | null; lastOnlineAt?: string | null } | { status: "online"; activeConnection?: ActiveConnection | null; lastOnlineAt: string; agentListReady?: boolean } diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 770198397..b0c5014c0 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -29,7 +29,6 @@ import type { AgentLifecycleStatus } from "@server/shared/agent-lifecycle"; import type { DaemonClient, ConnectionState } from "@server/client/daemon-client"; import { File } from "expo-file-system"; import { useDaemonConnections } from "./daemon-connections-context"; -import type { ActiveConnection } from "./daemon-connections-context"; import { useSessionStore, type Agent, @@ -204,24 +203,66 @@ const pushHistory = (history: string[], path: string): string[] => { return [...normalizedHistory, path]; }; -interface SessionProviderProps { +interface SessionProviderSharedProps { children: ReactNode; - serverUrl: string; serverId: string; - activeConnection: ActiveConnection | null; +} + +interface SessionProviderUrlProps extends SessionProviderSharedProps { + serverUrl: string; daemonPublicKeyB64?: string; } -// SessionProvider: Daemon client message handler that updates Zustand store -export function SessionProvider({ +interface SessionProviderClientProps extends SessionProviderSharedProps { + client: DaemonClient; +} + +export type SessionProviderProps = SessionProviderUrlProps | SessionProviderClientProps; + +function SessionProviderWithUrl({ children, - serverUrl, serverId, - activeConnection, + serverUrl, daemonPublicKeyB64, -}: SessionProviderProps) { - const queryClient = useQueryClient(); +}: SessionProviderUrlProps) { const client = useDaemonClient(serverUrl, { daemonPublicKeyB64 }); + return ( + + ); +} + +function SessionProviderWithClient({ + children, + serverId, + client, +}: SessionProviderClientProps) { + return ( + + ); +} + +// SessionProvider: Daemon client message handler that updates Zustand store +export function SessionProvider(props: SessionProviderProps) { + if ("client" in props) { + return ; + } + return ; +} + +function SessionProviderInternal({ + children, + serverId, + client, +}: SessionProviderClientProps) { + const queryClient = useQueryClient(); const [connectionSnapshot, setConnectionSnapshot] = useState(() => mapConnectionState(client.getConnectionState(), client.lastError) @@ -442,41 +483,6 @@ export function SessionProvider({ updateSessionClient(serverId, client); }, [serverId, client, updateSessionClient]); - // Connection status tracking - useEffect(() => { - if (connectionSnapshot.isConnected) { - updateConnectionStatus(serverId, { - status: "online", - activeConnection, - lastOnlineAt: new Date().toISOString(), - }); - return; - } - - if (connectionSnapshot.isConnecting) { - updateConnectionStatus(serverId, { status: "connecting", activeConnection }); - return; - } - - if (connectionSnapshot.lastError) { - updateConnectionStatus(serverId, { - status: "error", - activeConnection, - lastError: connectionSnapshot.lastError, - }); - return; - } - - updateConnectionStatus(serverId, { status: "offline", activeConnection }); - }, [ - serverId, - updateConnectionStatus, - connectionSnapshot.isConnected, - connectionSnapshot.isConnecting, - connectionSnapshot.lastError, - activeConnection, - ]); - // If the client drops mid-initialization, clear pending flags useEffect(() => { if (!connectionSnapshot.isConnected) { @@ -616,12 +622,6 @@ export function SessionProvider({ ] ); - useEffect(() => { - return () => { - updateConnectionStatus(serverId, { status: "offline", activeConnection, lastError: null }); - }; - }, [serverId, updateConnectionStatus, activeConnection]); - const updateExplorerState = useCallback( (agentId: string, updater: (state: any) => any) => { setFileExplorer(serverId, (prev) => { diff --git a/packages/app/src/runtime/host-runtime-bridge.test.ts b/packages/app/src/runtime/host-runtime-bridge.test.ts new file mode 100644 index 000000000..b0cf4a19b --- /dev/null +++ b/packages/app/src/runtime/host-runtime-bridge.test.ts @@ -0,0 +1,70 @@ +import { describe, expect, it } from "vitest"; +import type { HostRuntimeSnapshot } from "./host-runtime"; +import { toDaemonConnectionUpdateFromRuntime } from "./host-runtime-bridge"; + +const DIRECT_CONNECTION = { + type: "direct" as const, + endpoint: "lan:6767", + display: "lan:6767", +}; + +function makeSnapshot( + input: Partial +): HostRuntimeSnapshot { + return { + serverId: "srv_test", + activeConnectionId: "direct:lan:6767", + activeConnection: DIRECT_CONNECTION, + connectionStatus: "idle", + lastError: null, + lastOnlineAt: null, + probeByConnectionId: new Map(), + clientGeneration: 0, + ...input, + }; +} + +describe("toDaemonConnectionUpdateFromRuntime", () => { + it("maps online snapshots to online updates with the same active connection", () => { + const snapshot = makeSnapshot({ + connectionStatus: "online", + lastOnlineAt: "2026-02-19T20:00:00.000Z", + }); + + const update = toDaemonConnectionUpdateFromRuntime(snapshot); + + expect(update).toEqual({ + status: "online", + activeConnection: DIRECT_CONNECTION, + lastOnlineAt: "2026-02-19T20:00:00.000Z", + }); + }); + + it("maps error snapshots to error updates with the same error message", () => { + const snapshot = makeSnapshot({ + connectionStatus: "error", + lastError: "transport closed", + lastOnlineAt: "2026-02-19T20:00:00.000Z", + }); + + const update = toDaemonConnectionUpdateFromRuntime(snapshot); + + expect(update).toEqual({ + status: "error", + activeConnection: DIRECT_CONNECTION, + lastError: "transport closed", + lastOnlineAt: "2026-02-19T20:00:00.000Z", + }); + }); + + it("maps missing runtime snapshots to offline updates", () => { + const update = toDaemonConnectionUpdateFromRuntime(null); + + expect(update).toEqual({ + status: "offline", + activeConnection: null, + lastError: null, + lastOnlineAt: null, + }); + }); +}); diff --git a/packages/app/src/runtime/host-runtime-bridge.ts b/packages/app/src/runtime/host-runtime-bridge.ts new file mode 100644 index 000000000..e53d8019b --- /dev/null +++ b/packages/app/src/runtime/host-runtime-bridge.ts @@ -0,0 +1,47 @@ +import type { ConnectionStateUpdate } from "@/contexts/daemon-connections-context"; +import type { HostRuntimeSnapshot } from "./host-runtime"; + +export function toDaemonConnectionUpdateFromRuntime( + snapshot: HostRuntimeSnapshot | null +): ConnectionStateUpdate { + if (!snapshot) { + return { + status: "offline", + activeConnection: null, + lastError: null, + lastOnlineAt: null, + }; + } + + const activeConnection = snapshot.activeConnection; + switch (snapshot.connectionStatus) { + case "idle": + return { status: "idle" }; + case "connecting": + return { + status: "connecting", + activeConnection, + lastOnlineAt: snapshot.lastOnlineAt, + }; + case "online": + return { + status: "online", + activeConnection, + lastOnlineAt: snapshot.lastOnlineAt ?? new Date().toISOString(), + }; + case "offline": + return { + status: "offline", + activeConnection, + lastError: null, + lastOnlineAt: snapshot.lastOnlineAt, + }; + case "error": + return { + status: "error", + activeConnection, + lastError: snapshot.lastError ?? "Connection error", + lastOnlineAt: snapshot.lastOnlineAt, + }; + } +} diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts new file mode 100644 index 000000000..8914ee999 --- /dev/null +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -0,0 +1,253 @@ +import { describe, expect, it } from "vitest"; +import type { DaemonClient, ConnectionState } from "@server/client/daemon-client"; +import type { HostConnection, HostProfile } from "@/contexts/daemon-registry-context"; +import { + HostRuntimeController, + type HostRuntimeControllerDeps, +} from "./host-runtime"; + +class FakeDaemonClient { + private state: ConnectionState = { status: "idle" }; + private listeners = new Set<(status: ConnectionState) => void>(); + private error: string | null = null; + public connectCalls = 0; + public closeCalls = 0; + public ensureConnectedCalls = 0; + + async connect(): Promise { + this.connectCalls += 1; + this.setConnectionState({ status: "connected" }); + } + + async close(): Promise { + this.closeCalls += 1; + this.setConnectionState({ status: "disconnected", reason: "client_closed" }); + } + + ensureConnected(): void { + this.ensureConnectedCalls += 1; + if (this.state.status !== "connected") { + this.setConnectionState({ status: "connected" }); + } + } + + getConnectionState(): ConnectionState { + return this.state; + } + + subscribeConnectionStatus( + listener: (status: ConnectionState) => void + ): () => void { + this.listeners.add(listener); + listener(this.state); + return () => { + this.listeners.delete(listener); + }; + } + + get lastError(): string | null { + return this.error; + } + + setConnectionState(next: ConnectionState): void { + this.state = next; + if (next.status === "disconnected") { + this.error = next.reason ?? this.error; + } + for (const listener of this.listeners) { + listener(next); + } + } +} + +function makeHost(input?: Partial): HostProfile { + const direct: HostConnection = { + id: "direct:lan:6767", + type: "direct", + endpoint: "lan:6767", + }; + const relay: HostConnection = { + id: "relay:relay.paseo.sh:443", + type: "relay", + relayEndpoint: "relay.paseo.sh:443", + daemonPublicKeyB64: "pk_test", + }; + + return { + serverId: input?.serverId ?? "srv_test", + label: input?.label ?? "test host", + connections: input?.connections ?? [direct, relay], + preferredConnectionId: input?.preferredConnectionId ?? direct.id, + createdAt: input?.createdAt ?? new Date(0).toISOString(), + updatedAt: input?.updatedAt ?? new Date(0).toISOString(), + }; +} + +function makeDeps( + latencyByConnectionId: Record, + createdClients: FakeDaemonClient[] +): HostRuntimeControllerDeps { + return { + createClient: () => { + const client = new FakeDaemonClient(); + createdClients.push(client); + return client as unknown as DaemonClient; + }, + measureLatency: async ({ connection }) => { + const value = latencyByConnectionId[connection.id]; + if (value instanceof Error) { + throw value; + } + if (typeof value !== "number") { + throw new Error(`missing latency for ${connection.id}`); + } + return value; + }, + }; +} + +describe("HostRuntimeController", () => { + it("selects the lowest-latency connection on startup", async () => { + const host = makeHost({ preferredConnectionId: "direct:lan:6767" }); + const clients: FakeDaemonClient[] = []; + const latencies: Record = { + "direct:lan:6767": 82, + "relay:relay.paseo.sh:443": 18, + }; + const controller = new HostRuntimeController({ + host, + deps: makeDeps(latencies, clients), + }); + + await controller.start({ autoProbe: false }); + + const snapshot = controller.getSnapshot(); + expect(snapshot.activeConnectionId).toBe("relay:relay.paseo.sh:443"); + expect(snapshot.connectionStatus).toBe("online"); + expect(clients).toHaveLength(1); + expect(clients[0]?.connectCalls).toBe(1); + }); + + it("fails over when active connection becomes unavailable", async () => { + const host = makeHost({ preferredConnectionId: "direct:lan:6767" }); + const clients: FakeDaemonClient[] = []; + const latencies: Record = { + "direct:lan:6767": 15, + "relay:relay.paseo.sh:443": 55, + }; + const controller = new HostRuntimeController({ + host, + deps: makeDeps(latencies, clients), + }); + + await controller.start({ autoProbe: false }); + expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); + expect(clients).toHaveLength(1); + + latencies["direct:lan:6767"] = new Error("direct unavailable"); + latencies["relay:relay.paseo.sh:443"] = 42; + 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); + }); + + it("switches only after the faster alternative wins consecutive probes", async () => { + const host = makeHost({ preferredConnectionId: "direct:lan:6767" }); + const clients: FakeDaemonClient[] = []; + const latencies: Record = { + "direct:lan:6767": 15, + "relay:relay.paseo.sh:443": 60, + }; + const controller = new HostRuntimeController({ + host, + deps: makeDeps(latencies, clients), + }); + + await controller.start({ autoProbe: false }); + expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); + + latencies["direct:lan:6767"] = 95; + latencies["relay:relay.paseo.sh:443"] = 30; + await controller.runProbeCycleNow(); + expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); + + 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); + }); + + it("does not switch on a transient latency spike", async () => { + const host = makeHost({ preferredConnectionId: "direct:lan:6767" }); + const clients: FakeDaemonClient[] = []; + const latencies: Record = { + "direct:lan:6767": 15, + "relay:relay.paseo.sh:443": 80, + }; + const controller = new HostRuntimeController({ + host, + deps: makeDeps(latencies, clients), + }); + + await controller.start({ autoProbe: false }); + expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); + + latencies["direct:lan:6767"] = 100; + latencies["relay:relay.paseo.sh:443"] = 20; + await controller.runProbeCycleNow(); + expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); + + latencies["direct:lan:6767"] = 20; + latencies["relay:relay.paseo.sh:443"] = 90; + await controller.runProbeCycleNow(); + expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); + + latencies["direct:lan:6767"] = 100; + latencies["relay:relay.paseo.sh:443"] = 20; + await controller.runProbeCycleNow(); + expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); + + await controller.runProbeCycleNow(); + expect(controller.getSnapshot().activeConnectionId).toBe("direct:lan:6767"); + + await controller.runProbeCycleNow(); + expect(controller.getSnapshot().activeConnectionId).toBe("relay:relay.paseo.sh:443"); + }); + + it("exposes one snapshot with active connection and status from same source", async () => { + const host = makeHost(); + const clients: FakeDaemonClient[] = []; + const latencies: Record = { + "direct:lan:6767": 12, + "relay:relay.paseo.sh:443": 65, + }; + const controller = new HostRuntimeController({ + host, + deps: makeDeps(latencies, clients), + }); + + const observed = new Array>(); + const unsubscribe = controller.subscribe(() => { + observed.push(controller.getSnapshot()); + }); + + await controller.start({ autoProbe: false }); + + clients[0]?.setConnectionState({ + status: "disconnected", + reason: "transport closed", + }); + + const latest = observed[observed.length - 1]; + expect(latest?.activeConnectionId).toBe("direct:lan:6767"); + expect(latest?.connectionStatus).toBe("error"); + expect(latest?.lastError).toBe("transport closed"); + unsubscribe(); + }); +}); diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts new file mode 100644 index 000000000..60401913e --- /dev/null +++ b/packages/app/src/runtime/host-runtime.ts @@ -0,0 +1,577 @@ +import { useMemo, useSyncExternalStore } from "react"; +import { DaemonClient, type ConnectionState } from "@server/client/daemon-client"; +import type { HostConnection, HostProfile } from "@/contexts/daemon-registry-context"; +import type { ActiveConnection } from "@/contexts/daemon-connections-context"; +import { + buildDaemonWebSocketUrl, + buildRelayWebSocketUrl, +} from "@/utils/daemon-endpoints"; +import { measureConnectionLatency } from "@/utils/test-daemon-connection"; +import { + selectBestConnection, + type ConnectionCandidate, + type ConnectionProbeState, +} from "@/utils/connection-selection"; +import { createTauriWebSocketTransportFactory } from "@/utils/tauri-daemon-transport"; + +export type HostRuntimeConnectionStatus = + | "idle" + | "connecting" + | "online" + | "offline" + | "error"; + +export type HostRuntimeSnapshot = { + serverId: string; + activeConnectionId: string | null; + activeConnection: ActiveConnection | null; + connectionStatus: HostRuntimeConnectionStatus; + lastError: string | null; + lastOnlineAt: string | null; + probeByConnectionId: Map; + clientGeneration: number; +}; + +export type HostRuntimeControllerDeps = { + createClient: (input: { + host: HostProfile; + connection: HostConnection; + }) => DaemonClient; + measureLatency: (input: { + host: HostProfile; + connection: HostConnection; + }) => Promise; +}; + +export type HostRuntimeStartOptions = { + autoProbe?: boolean; +}; + +const PROBE_INTERVAL_MS = 10_000; +const ADAPTIVE_SWITCH_THRESHOLD_MS = 40; +const ADAPTIVE_SWITCH_CONSECUTIVE_PROBES = 3; + +function toActiveConnection(connection: HostConnection): ActiveConnection { + if (connection.type === "direct") { + return { + type: "direct", + endpoint: connection.endpoint, + display: connection.endpoint, + }; + } + return { + type: "relay", + endpoint: connection.relayEndpoint, + display: "relay", + }; +} + +function mapClientConnectionState(input: { + state: ConnectionState; + lastError: string | null; +}): Pick { + const { state, lastError } = input; + if (state.status === "connected") { + return { + connectionStatus: "online", + lastError: null, + lastOnlineAt: new Date().toISOString(), + }; + } + if (state.status === "connecting") { + return { + connectionStatus: "connecting", + lastError: null, + lastOnlineAt: null, + }; + } + if (state.status === "idle") { + return { + connectionStatus: "idle", + lastError: null, + lastOnlineAt: null, + }; + } + + const reason = state.reason ?? lastError ?? null; + if (!reason || reason === "client_closed") { + return { + connectionStatus: "offline", + lastError: null, + lastOnlineAt: null, + }; + } + return { + connectionStatus: "error", + lastError: reason, + lastOnlineAt: null, + }; +} + +function buildConnectionCandidates(host: HostProfile): ConnectionCandidate[] { + return host.connections.map((connection) => ({ + connectionId: connection.id, + connection, + })); +} + +function findConnectionById( + host: HostProfile, + connectionId: string | null +): HostConnection | null { + if (!connectionId) { + return null; + } + return host.connections.find((connection) => connection.id === connectionId) ?? null; +} + +function selectInitialConnectionId(host: HostProfile): string | null { + if (host.connections.length === 0) { + return null; + } + return host.connections[0]?.id ?? null; +} + +function createDefaultDeps(): HostRuntimeControllerDeps { + return { + createClient: ({ host, connection }) => { + const tauriTransportFactory = createTauriWebSocketTransportFactory(); + const base = { + suppressSendErrors: true, + ...(tauriTransportFactory + ? { transportFactory: tauriTransportFactory } + : {}), + }; + if (connection.type === "direct") { + return new DaemonClient({ + ...base, + url: buildDaemonWebSocketUrl(connection.endpoint), + }); + } + return new DaemonClient({ + ...base, + url: buildRelayWebSocketUrl({ + endpoint: connection.relayEndpoint, + serverId: host.serverId, + }), + e2ee: { + enabled: true, + daemonPublicKeyB64: connection.daemonPublicKeyB64, + }, + }); + }, + measureLatency: ({ host, connection }) => + measureConnectionLatency(connection, { serverId: host.serverId }), + }; +} + +export class HostRuntimeController { + private host: HostProfile; + private deps: HostRuntimeControllerDeps; + private snapshot: HostRuntimeSnapshot; + private listeners = new Set<() => void>(); + private activeClient: DaemonClient | null = null; + private unsubscribeClientStatus: (() => void) | null = null; + private probeIntervalHandle: ReturnType | null = null; + private started = false; + private switchCandidateConnectionId: string | null = null; + private switchCandidateHitCount = 0; + + constructor(input: { + host: HostProfile; + deps?: HostRuntimeControllerDeps; + }) { + this.host = input.host; + this.deps = input.deps ?? createDefaultDeps(); + this.snapshot = { + serverId: this.host.serverId, + activeConnectionId: null, + activeConnection: null, + connectionStatus: "idle", + lastError: null, + lastOnlineAt: null, + probeByConnectionId: new Map(), + clientGeneration: 0, + }; + } + + getSnapshot(): HostRuntimeSnapshot { + return this.snapshot; + } + + getClient(): DaemonClient | null { + return this.activeClient; + } + + subscribe(listener: () => void): () => void { + this.listeners.add(listener); + return () => { + this.listeners.delete(listener); + }; + } + + async start(options?: HostRuntimeStartOptions): Promise { + if (this.started) { + return; + } + this.started = true; + await this.runProbeCycleNow(); + + if (!this.snapshot.activeConnectionId) { + const fallbackConnectionId = selectInitialConnectionId(this.host); + if (fallbackConnectionId) { + await this.switchToConnection(fallbackConnectionId); + } + } + + if (options?.autoProbe !== false && PROBE_INTERVAL_MS > 0) { + this.probeIntervalHandle = setInterval(() => { + void this.runProbeCycleNow(); + }, PROBE_INTERVAL_MS); + } + } + + async stop(): Promise { + this.started = false; + if (this.probeIntervalHandle) { + clearInterval(this.probeIntervalHandle); + this.probeIntervalHandle = null; + } + if (this.unsubscribeClientStatus) { + this.unsubscribeClientStatus(); + this.unsubscribeClientStatus = null; + } + if (this.activeClient) { + const prev = this.activeClient; + this.activeClient = null; + await prev.close().catch(() => undefined); + } + this.updateSnapshot({ + activeConnectionId: null, + activeConnection: null, + connectionStatus: "offline", + lastError: null, + lastOnlineAt: null, + }); + } + + async updateHost(host: HostProfile): Promise { + this.host = host; + const activeConnectionId = this.snapshot.activeConnectionId; + const activeConnection = findConnectionById(host, activeConnectionId); + if (!activeConnection) { + await this.runProbeCycleNow(); + if (this.snapshot.activeConnectionId) { + return; + } + const fallbackConnectionId = selectInitialConnectionId(host); + if (!fallbackConnectionId) { + await this.stop(); + return; + } + await this.switchToConnection(fallbackConnectionId); + } + } + + ensureConnected(): void { + this.activeClient?.ensureConnected(); + } + + async runProbeCycleNow(): Promise { + if (this.host.connections.length === 0) { + this.updateSnapshot({ + probeByConnectionId: new Map(), + }); + return; + } + + const probeByConnectionId = new Map(); + await Promise.all( + this.host.connections.map(async (connection) => { + try { + const latencyMs = await this.deps.measureLatency({ + host: this.host, + connection, + }); + probeByConnectionId.set(connection.id, { + status: "available", + latencyMs, + }); + } catch { + probeByConnectionId.set(connection.id, { + status: "unavailable", + latencyMs: null, + }); + } + }) + ); + + this.updateSnapshot({ probeByConnectionId }); + + const activeConnectionId = this.snapshot.activeConnectionId; + const activeProbe = activeConnectionId + ? probeByConnectionId.get(activeConnectionId) + : null; + + if (!activeConnectionId || !findConnectionById(this.host, activeConnectionId)) { + const nextConnectionId = selectBestConnection({ + candidates: buildConnectionCandidates(this.host), + probeByConnectionId, + }); + if (nextConnectionId) { + await this.switchToConnection(nextConnectionId); + } + return; + } + + if (activeProbe?.status === "unavailable") { + const nextConnectionId = selectBestConnection({ + candidates: buildConnectionCandidates(this.host), + probeByConnectionId, + }); + if (nextConnectionId && nextConnectionId !== activeConnectionId) { + await this.switchToConnection(nextConnectionId); + } + this.switchCandidateConnectionId = null; + this.switchCandidateHitCount = 0; + return; + } + + if (activeProbe && activeProbe.status === "available") { + const available = Array.from(probeByConnectionId.entries()) + .filter(([, probe]) => probe.status === "available") + .map(([connectionId, probe]) => ({ + connectionId, + latencyMs: (probe as { status: "available"; latencyMs: number }).latencyMs, + })) + .sort((left, right) => left.latencyMs - right.latencyMs); + + const fastest = available[0] ?? null; + if (!fastest || fastest.connectionId === activeConnectionId) { + this.switchCandidateConnectionId = null; + this.switchCandidateHitCount = 0; + return; + } + + const activeLatency = activeProbe.latencyMs; + const improvement = activeLatency - fastest.latencyMs; + if (improvement < ADAPTIVE_SWITCH_THRESHOLD_MS) { + this.switchCandidateConnectionId = null; + this.switchCandidateHitCount = 0; + return; + } + + if (this.switchCandidateConnectionId === fastest.connectionId) { + this.switchCandidateHitCount += 1; + } else { + this.switchCandidateConnectionId = fastest.connectionId; + this.switchCandidateHitCount = 1; + } + + if ( + this.switchCandidateHitCount >= + ADAPTIVE_SWITCH_CONSECUTIVE_PROBES + ) { + this.switchCandidateConnectionId = null; + this.switchCandidateHitCount = 0; + await this.switchToConnection(fastest.connectionId); + } + } + } + + private updateSnapshot( + patch: Partial> + ): void { + const next: HostRuntimeSnapshot = { + ...this.snapshot, + ...patch, + serverId: this.host.serverId, + }; + this.snapshot = next; + for (const listener of this.listeners) { + listener(); + } + } + + private async switchToConnection(connectionId: string): Promise { + const connection = findConnectionById(this.host, connectionId); + if (!connection) { + return; + } + + if (this.unsubscribeClientStatus) { + this.unsubscribeClientStatus(); + this.unsubscribeClientStatus = null; + } + if (this.activeClient) { + const previousClient = this.activeClient; + this.activeClient = null; + await previousClient.close().catch(() => undefined); + } + + const client = this.deps.createClient({ + host: this.host, + connection, + }); + this.activeClient = client; + this.snapshot = { + ...this.snapshot, + serverId: this.host.serverId, + activeConnectionId: connection.id, + activeConnection: toActiveConnection(connection), + connectionStatus: "connecting", + lastError: null, + lastOnlineAt: null, + clientGeneration: this.snapshot.clientGeneration + 1, + }; + for (const listener of this.listeners) { + listener(); + } + + this.unsubscribeClientStatus = client.subscribeConnectionStatus((state) => { + const mapped = mapClientConnectionState({ + state, + lastError: client.lastError, + }); + this.updateSnapshot(mapped); + }); + + try { + await client.connect(); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + this.updateSnapshot({ + connectionStatus: "error", + lastError: message, + }); + } + } +} + +export class HostRuntimeStore { + private controllers = new Map(); + private serverListeners = new Map void>>(); + private deps: HostRuntimeControllerDeps; + + constructor(input?: { + deps?: HostRuntimeControllerDeps; + }) { + this.deps = input?.deps ?? createDefaultDeps(); + } + + syncHosts(hosts: HostProfile[]): void { + const nextIds = new Set(hosts.map((host) => host.serverId)); + for (const [serverId, controller] of this.controllers) { + if (nextIds.has(serverId)) { + continue; + } + this.controllers.delete(serverId); + void controller.stop(); + this.emit(serverId); + } + + for (const host of hosts) { + const existing = this.controllers.get(host.serverId); + if (existing) { + void existing.updateHost(host); + continue; + } + const controller = new HostRuntimeController({ + host, + deps: this.deps, + }); + this.controllers.set(host.serverId, controller); + controller.subscribe(() => { + this.emit(host.serverId); + }); + void controller.start(); + this.emit(host.serverId); + } + } + + getSnapshot(serverId: string): HostRuntimeSnapshot | null { + return this.controllers.get(serverId)?.getSnapshot() ?? null; + } + + getClient(serverId: string): DaemonClient | null { + return this.controllers.get(serverId)?.getClient() ?? null; + } + + subscribe(serverId: string, listener: () => void): () => void { + const existing = this.serverListeners.get(serverId) ?? new Set<() => void>(); + existing.add(listener); + this.serverListeners.set(serverId, existing); + return () => { + const set = this.serverListeners.get(serverId); + if (!set) { + return; + } + set.delete(listener); + if (set.size === 0) { + this.serverListeners.delete(serverId); + } + }; + } + + ensureConnectedAll(): void { + for (const controller of this.controllers.values()) { + controller.ensureConnected(); + } + } + + runProbeCycleNow(serverId?: string): Promise { + if (serverId) { + return this.controllers.get(serverId)?.runProbeCycleNow() ?? Promise.resolve(); + } + return Promise.all( + Array.from(this.controllers.values(), (controller) => + controller.runProbeCycleNow() + ) + ).then(() => undefined); + } + + private emit(serverId: string): void { + const listeners = this.serverListeners.get(serverId); + if (!listeners) { + return; + } + for (const listener of listeners) { + listener(); + } + } +} + +let singletonHostRuntimeStore: HostRuntimeStore | null = null; + +export function getHostRuntimeStore(): HostRuntimeStore { + if (!singletonHostRuntimeStore) { + singletonHostRuntimeStore = new HostRuntimeStore(); + } + return singletonHostRuntimeStore; +} + +export function useHostRuntimeSnapshot( + serverId: string +): HostRuntimeSnapshot | null { + const store = getHostRuntimeStore(); + return useSyncExternalStore( + (onStoreChange) => store.subscribe(serverId, onStoreChange), + () => store.getSnapshot(serverId), + () => store.getSnapshot(serverId) + ); +} + +export function useHostRuntimeSession(serverId: string): { + snapshot: HostRuntimeSnapshot | null; + client: DaemonClient | null; +} { + const store = getHostRuntimeStore(); + const snapshot = useHostRuntimeSnapshot(serverId); + return useMemo( + () => ({ + snapshot, + client: store.getClient(serverId), + }), + [serverId, snapshot, store] + ); +} diff --git a/packages/app/src/utils/connection-selection.test.ts b/packages/app/src/utils/connection-selection.test.ts index a3569a245..70e2af722 100644 --- a/packages/app/src/utils/connection-selection.test.ts +++ b/packages/app/src/utils/connection-selection.test.ts @@ -25,7 +25,7 @@ function probes( } describe("selectBestConnection", () => { - it("picks the available connection with lowest latency regardless of transport type", () => { + it("picks the available connection with lowest latency regardless of transport", () => { const candidates: ConnectionCandidate[] = [ { connectionId: "direct:a", connection: makeDirect("direct:a", "a:6767") }, { @@ -36,7 +36,6 @@ describe("selectBestConnection", () => { const selected = selectBestConnection({ candidates, - preferredConnectionId: "direct:a", probeByConnectionId: probes({ "direct:a": { status: "available", latencyMs: 84 }, "relay:b": { status: "available", latencyMs: 34 }, @@ -46,6 +45,28 @@ describe("selectBestConnection", () => { expect(selected).toBe("relay:b"); }); + it("picks the lowest-latency connection among mixed transport candidates", () => { + const candidates: ConnectionCandidate[] = [ + { connectionId: "direct:a", connection: makeDirect("direct:a", "a:6767") }, + { connectionId: "direct:c", connection: makeDirect("direct:c", "c:6767") }, + { + connectionId: "relay:b", + connection: makeRelay("relay:b", "relay.example:443"), + }, + ]; + + const selected = selectBestConnection({ + candidates, + probeByConnectionId: probes({ + "direct:a": { status: "available", latencyMs: 84 }, + "direct:c": { status: "available", latencyMs: 41 }, + "relay:b": { status: "available", latencyMs: 12 }, + }), + }); + + expect(selected).toBe("relay:b"); + }); + it("ignores unavailable and pending probes", () => { const candidates: ConnectionCandidate[] = [ { connectionId: "direct:a", connection: makeDirect("direct:a", "a:6767") }, @@ -58,7 +79,6 @@ describe("selectBestConnection", () => { const selected = selectBestConnection({ candidates, - preferredConnectionId: "direct:a", probeByConnectionId: probes({ "direct:a": { status: "pending", latencyMs: null }, "relay:b": { status: "unavailable", latencyMs: null }, @@ -69,7 +89,7 @@ describe("selectBestConnection", () => { expect(selected).toBe("direct:c"); }); - it("falls back to preferred connection when no candidates are available", () => { + it("returns null when no candidates are available", () => { const candidates: ConnectionCandidate[] = [ { connectionId: "direct:a", connection: makeDirect("direct:a", "a:6767") }, { @@ -80,17 +100,16 @@ describe("selectBestConnection", () => { const selected = selectBestConnection({ candidates, - preferredConnectionId: "relay:b", probeByConnectionId: probes({ "direct:a": { status: "pending", latencyMs: null }, "relay:b": { status: "unavailable", latencyMs: null }, }), }); - expect(selected).toBe("relay:b"); + expect(selected).toBeNull(); }); - it("falls back to first candidate when preferred is missing and none are available", () => { + it("returns null when no probes are available even if candidates exist", () => { const candidates: ConnectionCandidate[] = [ { connectionId: "direct:a", connection: makeDirect("direct:a", "a:6767") }, { @@ -101,13 +120,12 @@ describe("selectBestConnection", () => { const selected = selectBestConnection({ candidates, - preferredConnectionId: "missing", probeByConnectionId: probes({ "direct:a": { status: "unavailable", latencyMs: null }, "relay:b": { status: "pending", latencyMs: null }, }), }); - expect(selected).toBe("direct:a"); + expect(selected).toBeNull(); }); }); diff --git a/packages/app/src/utils/connection-selection.ts b/packages/app/src/utils/connection-selection.ts index b768aa6d4..c4bb1949f 100644 --- a/packages/app/src/utils/connection-selection.ts +++ b/packages/app/src/utils/connection-selection.ts @@ -12,14 +12,13 @@ export type ConnectionProbeState = export type SelectBestConnectionInput = { candidates: ConnectionCandidate[]; - preferredConnectionId: string | null; probeByConnectionId: Map; }; export function selectBestConnection( input: SelectBestConnectionInput ): string | null { - const { candidates, preferredConnectionId, probeByConnectionId } = input; + const { candidates, probeByConnectionId } = input; if (candidates.length === 0) { return null; } @@ -29,8 +28,9 @@ export function selectBestConnection( .filter(({ candidate }) => { const probe = probeByConnectionId.get(candidate.connectionId); return probe?.status === "available"; - }) - .sort((left, right) => { + }); + + const byLatency = (left: { candidate: ConnectionCandidate; index: number }, right: { candidate: ConnectionCandidate; index: number }) => { const leftProbe = probeByConnectionId.get(left.candidate.connectionId); const rightProbe = probeByConnectionId.get(right.candidate.connectionId); @@ -48,18 +48,12 @@ export function selectBestConnection( } return leftLatency - rightLatency; - }); + }; - if (available.length > 0) { - return available[0]!.candidate.connectionId; + if (available.length === 0) { + return null; } - const preferred = candidates.find( - (candidate) => candidate.connectionId === preferredConnectionId - ); - if (preferred) { - return preferred.connectionId; - } - - return candidates[0]!.connectionId; + const sorted = available.sort(byLatency); + return sorted[0]!.candidate.connectionId; }