Remove primary host plumbing

2025-11-26 18:30:04 +01:00
parent 5807f62f1a
commit db87f331fb
17 changed files with 455 additions and 387 deletions
@@ -1,18 +1,13 @@
 import { createContext, useCallback, useContext, useEffect, useMemo, useRef, useState } from "react";
 import type { ReactNode } from "react";
-import AsyncStorage from "@react-native-async-storage/async-storage";
-import { useQuery, useQueryClient } from "@tanstack/react-query";
 import { useDaemonRegistry, type DaemonProfile } from "./daemon-registry-context";
-import { trackAnalyticsEvent } from "@/utils/analytics";
 import type { SessionContextValue } from "./session-context";

 export interface SessionDirectoryEntry {
  getSnapshot: () => SessionContextValue | null;
  subscribe: (listener: () => void) => () => void;
 }
-
-export type SessionAccessorRole = "primary" | "background";
-type SessionAccessorRegistry = Map<string, Map<SessionAccessorRole, SessionDirectoryEntry>>;
+type SessionAccessorRegistry = Map<string, SessionDirectoryEntry>;

 export type ConnectionState =
  | { status: "idle"; lastError: null; lastOnlineAt: string | null }
@@ -35,33 +30,18 @@ export type DaemonConnectionRecord = {
 } & ConnectionState;

 interface DaemonConnectionsContextValue {
-  activeDaemonId: string | null;
-  activeDaemon: DaemonProfile | null;
  connectionStates: Map<string, DaemonConnectionRecord>;
  isLoading: boolean;
-  setActiveDaemonId: (daemonId: string, options?: SetActiveDaemonOptions) => void;
  updateConnectionStatus: (daemonId: string, update: ConnectionStateUpdate) => void;
  sessionAccessors: Map<string, SessionDirectoryEntry>;
-  sessionAccessorRoles: Map<string, Set<SessionAccessorRole>>;
-  registerSessionAccessor: (
-    daemonId: string,
-    entry: SessionDirectoryEntry,
-    role?: SessionAccessorRole
-  ) => void;
-  unregisterSessionAccessor: (daemonId: string, role?: SessionAccessorRole) => void;
+  registerSessionAccessor: (daemonId: string, entry: SessionDirectoryEntry) => void;
+  unregisterSessionAccessor: (daemonId: string) => void;
  subscribeToSessionDirectory: (listener: () => void) => () => void;
  notifySessionDirectoryChange: () => void;
 }

 const DaemonConnectionsContext = createContext<DaemonConnectionsContextValue | null>(null);

-const ACTIVE_DAEMON_STORAGE_KEY = "@paseo:active-daemon-id";
-const ACTIVE_DAEMON_QUERY_KEY = ["active-daemon-id"];
-
-export type SetActiveDaemonOptions = {
-  source?: string;
-};
-
 function createDefaultConnectionState(): ConnectionState {
  return {
    status: "idle",
@@ -132,25 +112,10 @@ export function useDaemonConnections(): DaemonConnectionsContextValue {

 export function DaemonConnectionsProvider({ children }: { children: ReactNode }) {
  const { daemons, isLoading: registryLoading } = useDaemonRegistry();
-  const [activeDaemonId, setActiveDaemonIdState] = useState<string | null>(null);
  const [connectionStates, setConnectionStates] = useState<Map<string, DaemonConnectionRecord>>(new Map());
  const [sessionAccessorRegistry, setSessionAccessorRegistry] = useState<SessionAccessorRegistry>(new Map());
-  const queryClient = useQueryClient();
-  const activeDaemonPreference = useQuery({
-    queryKey: ACTIVE_DAEMON_QUERY_KEY,
-    queryFn: loadActiveDaemonPreference,
-    staleTime: Infinity,
-    gcTime: Infinity,
-  });
  const sessionDirectoryListenersRef = useRef<Set<() => void>>(new Set());

-  useEffect(() => {
-    if (activeDaemonPreference.isPending) {
-      return;
-    }
-    setActiveDaemonIdState(activeDaemonPreference.data ?? null);
-  }, [activeDaemonPreference.data, activeDaemonPreference.isPending]);
-
  // Ensure connection states stay in sync with registry entries
  useEffect(() => {
    setConnectionStates((prev) => {
@@ -172,77 +137,18 @@ export function DaemonConnectionsProvider({ children }: { children: ReactNode })
      let changed = false;
      const next: SessionAccessorRegistry = new Map();

-      for (const [daemonId, roleMap] of prev.entries()) {
+      for (const [daemonId, entry] of prev.entries()) {
        if (!validDaemonIds.has(daemonId)) {
          changed = true;
          continue;
        }
-        next.set(daemonId, roleMap);
+        next.set(daemonId, entry);
      }

      return changed ? next : prev;
    });
  }, [daemons]);

-  // Keep active daemon aligned with registry default
-  const persistActiveDaemonId = useCallback(async (daemonId: string | null) => {
-    queryClient.setQueryData<string | null>(ACTIVE_DAEMON_QUERY_KEY, daemonId);
-    try {
-      if (daemonId) {
-        await AsyncStorage.setItem(ACTIVE_DAEMON_STORAGE_KEY, daemonId);
-      } else {
-        await AsyncStorage.removeItem(ACTIVE_DAEMON_STORAGE_KEY);
-      }
-    } catch (error) {
-      console.error("[DaemonConnections] Failed to persist active daemon", error);
-    }
-  }, [queryClient]);
-
-  const setActiveDaemonId = useCallback(
-    (daemonId: string, options?: SetActiveDaemonOptions) => {
-      setActiveDaemonIdState((previous) => {
-        if (previous !== daemonId) {
-          trackAnalyticsEvent({
-            type: "daemon_active_changed",
-            daemonId,
-            previousDaemonId: previous,
-            source: options?.source,
-          });
-        }
-        return daemonId;
-      });
-      void persistActiveDaemonId(daemonId);
-    },
-    [persistActiveDaemonId]
-  );
-
-  useEffect(() => {
-    if (activeDaemonPreference.isPending) {
-      return;
-    }
-
-    if (daemons.length === 0) {
-      setActiveDaemonIdState(null);
-      void persistActiveDaemonId(null);
-      return;
-    }
-
-    if (activeDaemonId && daemons.some((daemon) => daemon.id === activeDaemonId)) {
-      return;
-    }
-
-    const fallback = daemons[0];
-    if (fallback) {
-      setActiveDaemonId(fallback.id, { source: "auto_fallback" });
-    }
-  }, [daemons, activeDaemonId, activeDaemonPreference.isPending, persistActiveDaemonId, setActiveDaemonId]);
-
-  const activeDaemon = useMemo(() => {
-    if (!activeDaemonId) {
-      return null;
-    }
-    return daemons.find((daemon) => daemon.id === activeDaemonId) ?? null;
-  }, [activeDaemonId, daemons]);

  const updateConnectionStatus = useCallback(
    (daemonId: string, update: ConnectionStateUpdate) => {
@@ -269,41 +175,28 @@ export function DaemonConnectionsProvider({ children }: { children: ReactNode })
    []
  );

-  const registerSessionAccessor = useCallback(
-    (daemonId: string, entry: SessionDirectoryEntry, role: SessionAccessorRole = "primary") => {
-      setSessionAccessorRegistry((prev) => {
-        const next = new Map(prev);
-        const existing = new Map(next.get(daemonId) ?? []);
-        existing.set(role, entry);
-        next.set(daemonId, existing);
-        return next;
-      });
-    },
-    []
-  );
+  const registerSessionAccessor = useCallback((daemonId: string, entry: SessionDirectoryEntry) => {
+    setSessionAccessorRegistry((prev) => {
+      const next = new Map(prev);
+      const hasExisting = next.has(daemonId);
+      next.set(daemonId, entry);
+      if (hasExisting) {
+        console.warn(`[DaemonConnections] Duplicate session accessor detected for "${daemonId}". Overwriting.`);
+      }
+      return next;
+    });
+  }, []);

-  const unregisterSessionAccessor = useCallback(
-    (daemonId: string, role: SessionAccessorRole = "primary") => {
-      setSessionAccessorRegistry((prev) => {
-        const existing = prev.get(daemonId);
-        if (!existing) {
-          return prev;
-        }
-
-        const nextRoleMap = new Map(existing);
-        nextRoleMap.delete(role);
-
-        const next = new Map(prev);
-        if (nextRoleMap.size === 0) {
-          next.delete(daemonId);
-        } else {
-          next.set(daemonId, nextRoleMap);
-        }
-        return next;
-      });
-    },
-    []
-  );
+  const unregisterSessionAccessor = useCallback((daemonId: string) => {
+    setSessionAccessorRegistry((prev) => {
+      if (!prev.has(daemonId)) {
+        return prev;
+      }
+      const next = new Map(prev);
+      next.delete(daemonId);
+      return next;
+    });
+  }, []);

  const subscribeToSessionDirectory = useCallback((listener: () => void) => {
    sessionDirectoryListenersRef.current.add(listener);
@@ -323,43 +216,13 @@ export function DaemonConnectionsProvider({ children }: { children: ReactNode })
    }
  }, []);

-  const { sessionAccessors, sessionAccessorRoles } = useMemo(() => {
-    const flattened = new Map<string, SessionDirectoryEntry>();
-    const roleSets = new Map<string, Set<SessionAccessorRole>>();
-
-    sessionAccessorRegistry.forEach((roleMap, daemonId) => {
-      if (roleMap.size === 0) {
-        return;
-      }
-
-      const roles = new Set<SessionAccessorRole>();
-      roleMap.forEach((_entry, role) => roles.add(role));
-      roleSets.set(daemonId, roles);
-
-      const primaryEntry = roleMap.get("primary");
-      if (primaryEntry) {
-        flattened.set(daemonId, primaryEntry);
-        return;
-      }
-
-      const backgroundEntry = roleMap.get("background");
-      if (backgroundEntry) {
-        flattened.set(daemonId, backgroundEntry);
-      }
-    });
-
-    return { sessionAccessors: flattened, sessionAccessorRoles: roleSets };
-  }, [sessionAccessorRegistry]);
+  const sessionAccessors = useMemo(() => new Map(sessionAccessorRegistry), [sessionAccessorRegistry]);

  const value: DaemonConnectionsContextValue = {
-    activeDaemonId,
-    activeDaemon,
    connectionStates,
-    isLoading: registryLoading || activeDaemonPreference.isPending,
-    setActiveDaemonId,
+    isLoading: registryLoading,
    updateConnectionStatus,
    sessionAccessors,
-    sessionAccessorRoles,
    registerSessionAccessor,
    unregisterSessionAccessor,
    subscribeToSessionDirectory,
@@ -372,13 +235,3 @@ export function DaemonConnectionsProvider({ children }: { children: ReactNode })
    </DaemonConnectionsContext.Provider>
  );
 }
-
-async function loadActiveDaemonPreference(): Promise<string | null> {
-  try {
-    const stored = await AsyncStorage.getItem(ACTIVE_DAEMON_STORAGE_KEY);
-    return stored ?? null;
-  } catch (error) {
-    console.error("[DaemonConnections] Failed to read active daemon preference", error);
-    throw error;
-  }
-}
@@ -1,6 +1,7 @@
-import { createContext, useContext, useState, ReactNode, useCallback, useEffect, useRef } from "react";
+import { createContext, useContext, useState, ReactNode, useCallback, useEffect, useRef, useMemo } from "react";
 import { useSpeechmaticsAudio } from "@/hooks/use-speechmatics-audio";
-import { useSession } from "./session-context";
+import { useSessionDirectory } from "@/hooks/use-session-directory";
+import type { SessionContextValue } from "./session-context";
 import { generateMessageId } from "@/types/stream";
 import type { WSInboundMessage } from "@server/server/messages";

@@ -11,9 +12,10 @@ interface RealtimeContextValue {
  isDetecting: boolean;
  isSpeaking: boolean;
  segmentDuration: number;
-  startRealtime: () => Promise<void>;
+  startRealtime: (serverId: string) => Promise<void>;
  stopRealtime: () => Promise<void>;
  toggleMute: () => void;
+  activeServerId: string | null;
 }

 const RealtimeContext = createContext<RealtimeContextValue | null>(null);
@@ -31,13 +33,9 @@ interface RealtimeProviderProps {
 }

 export function RealtimeProvider({ children }: RealtimeProviderProps) {
-  const {
-    ws,
-    audioPlayer,
-    isPlayingAudio,
-    setMessages,
-    setVoiceDetectionFlags,
-  } = useSession();
+  const sessionDirectory = useSessionDirectory();
+  const [activeServerId, setActiveServerId] = useState<string | null>(null);
+  const realtimeSessionRef = useRef<SessionContextValue | null>(null);
  const [isRealtimeMode, setIsRealtimeMode] = useState(false);
  const bargeInPlaybackStopRef = useRef<number | null>(null);

@@ -45,22 +43,29 @@ export function RealtimeProvider({ children }: RealtimeProviderProps) {
    onSpeechStart: () => {
      console.log("[Realtime] Speech detected");
      // Stop audio playback if playing
-      if (isPlayingAudio) {
+      const session = realtimeSessionRef.current;
+      const sessionAudioPlayer = session?.audioPlayer;
+      const sessionWs = session?.ws;
+      const sessionIsPlayingAudio = session?.isPlayingAudio ?? false;
+
+      if (sessionIsPlayingAudio && sessionAudioPlayer) {
        if (bargeInPlaybackStopRef.current === null) {
          bargeInPlaybackStopRef.current = Date.now();
        }
-        audioPlayer.stop();
+        sessionAudioPlayer.stop();
      }

      // Abort any in-flight orchestrator turn before the new speech segment streams
      try {
-        const abortMessage: WSInboundMessage = {
-          type: "session",
-          message: {
-            type: "abort_request",
-          },
-        };
-        ws.send(abortMessage);
+        if (sessionWs) {
+          const abortMessage: WSInboundMessage = {
+            type: "session",
+            message: {
+              type: "abort_request",
+            },
+          };
+          sessionWs.send(abortMessage);
+        }
        console.log("[Realtime] Sent abort_request before streaming audio");
      } catch (error) {
        console.error("[Realtime] Failed to send abort_request:", error);
@@ -78,8 +83,9 @@ export function RealtimeProvider({ children }: RealtimeProviderProps) {
      );

      // Send audio segment to server (realtime always goes to orchestrator)
+      const session = realtimeSessionRef.current;
      try {
-        ws.send({
+        session?.ws.send({
          type: "session",
          message: {
            type: "realtime_audio_chunk",
@@ -94,16 +100,19 @@ export function RealtimeProvider({ children }: RealtimeProviderProps) {
    },
    onError: (error) => {
      console.error("[Realtime] Audio error:", error);
-      setMessages((prev) => [
-        ...prev,
-        {
-          type: "activity",
-          id: generateMessageId(),
-          timestamp: Date.now(),
-          activityType: "error",
-          message: `Realtime audio error: ${error.message}`,
-        },
-      ]);
+      const session = realtimeSessionRef.current;
+      if (session) {
+        session.setMessages((prev) => [
+          ...prev,
+          {
+            type: "activity",
+            id: generateMessageId(),
+            timestamp: Date.now(),
+            activityType: "error",
+            message: `Realtime audio error: ${error.message}`,
+          },
+        ]);
+      }
    },
    volumeThreshold: 0.3,
    silenceDuration: 2000,
@@ -113,8 +122,22 @@ export function RealtimeProvider({ children }: RealtimeProviderProps) {

  // Update voice detection flags whenever they change
  useEffect(() => {
-    setVoiceDetectionFlags(realtimeAudio.isDetecting, realtimeAudio.isSpeaking);
-  }, [realtimeAudio.isDetecting, realtimeAudio.isSpeaking, setVoiceDetectionFlags]);
+    const session = realtimeSessionRef.current;
+    session?.setVoiceDetectionFlags(realtimeAudio.isDetecting, realtimeAudio.isSpeaking);
+  }, [realtimeAudio.isDetecting, realtimeAudio.isSpeaking]);
+
+  const activeSession = useMemo(() => {
+    if (!activeServerId) {
+      return null;
+    }
+    return sessionDirectory.get(activeServerId) ?? null;
+  }, [activeServerId, sessionDirectory]);
+
+  useEffect(() => {
+    realtimeSessionRef.current = activeSession;
+  }, [activeSession]);
+
+  const isPlayingAudio = activeSession?.isPlayingAudio ?? false;

  useEffect(() => {
    if (!isPlayingAudio && bargeInPlaybackStopRef.current !== null) {
@@ -128,47 +151,61 @@ export function RealtimeProvider({ children }: RealtimeProviderProps) {
    }
  }, [isPlayingAudio]);

-  const startRealtime = useCallback(async () => {
-    try {
-      await realtimeAudio.start();
-      setIsRealtimeMode(true);
-      console.log("[Realtime] Mode enabled");
+  const startRealtime = useCallback(
+    async (serverId: string) => {
+      const session = sessionDirectory.get(serverId) ?? null;
+      if (!session) {
+        throw new Error(`Host ${serverId} is not connected`);
+      }

-      // Notify server
-      const modeMessage: WSInboundMessage = {
-        type: "session",
-        message: {
-          type: "set_realtime_mode",
-          enabled: true,
-        },
-      };
-      ws.send(modeMessage);
-    } catch (error: any) {
-      console.error("[Realtime] Failed to start:", error);
-      throw error;
-    }
-  }, [realtimeAudio, ws]);
+      try {
+        realtimeSessionRef.current = session;
+        setActiveServerId(serverId);
+        await realtimeAudio.start();
+        setIsRealtimeMode(true);
+        console.log("[Realtime] Mode enabled");
+
+        const modeMessage: WSInboundMessage = {
+          type: "session",
+          message: {
+            type: "set_realtime_mode",
+            enabled: true,
+          },
+        };
+        session.ws.send(modeMessage);
+      } catch (error: any) {
+        console.error("[Realtime] Failed to start:", error);
+        setActiveServerId((current) => (current === serverId ? null : current));
+        throw error;
+      }
+    },
+    [realtimeAudio, sessionDirectory]
+  );

  const stopRealtime = useCallback(async () => {
    try {
+      const session = realtimeSessionRef.current;
+      session?.audioPlayer.stop();
      await realtimeAudio.stop();
      setIsRealtimeMode(false);
+      setActiveServerId(null);
      console.log("[Realtime] Mode disabled");

-      // Notify server
-      const modeMessage: WSInboundMessage = {
-        type: "session",
-        message: {
-          type: "set_realtime_mode",
-          enabled: false,
-        },
-      };
-      ws.send(modeMessage);
+      if (session) {
+        const modeMessage: WSInboundMessage = {
+          type: "session",
+          message: {
+            type: "set_realtime_mode",
+            enabled: false,
+          },
+        };
+        session.ws.send(modeMessage);
+      }
    } catch (error: any) {
      console.error("[Realtime] Failed to stop:", error);
      throw error;
    }
-  }, [realtimeAudio, ws]);
+  }, [realtimeAudio]);

  const value: RealtimeContextValue = {
    isRealtimeMode,
@@ -180,6 +217,7 @@ export function RealtimeProvider({ children }: RealtimeProviderProps) {
    startRealtime,
    stopRealtime,
    toggleMute: realtimeAudio.toggleMute,
+    activeServerId,
  };

  return (
@@ -29,7 +29,7 @@ import type {
 } from "@server/server/agent/agent-sdk-types";
 import { ScrollView } from "react-native";
 import * as FileSystem from 'expo-file-system';
-import { useDaemonConnections, type SessionAccessorRole } from "./daemon-connections-context";
+import { useDaemonConnections } from "./daemon-connections-context";

 const derivePendingPermissionKey = (agentId: string, request: AgentPermissionRequest) => {
  const fallbackId =
@@ -388,11 +388,9 @@ interface SessionProviderProps {
  children: ReactNode;
  serverUrl: string;
  serverId: string;
-  role?: SessionAccessorRole;
 }

-export function SessionProvider({ children, serverUrl, serverId, role }: SessionProviderProps) {
-  const sessionRole = role ?? "primary";
+export function SessionProvider({ children, serverUrl, serverId }: SessionProviderProps) {
  const ws = useWebSocket(serverUrl);
  const wsIsConnected = ws.isConnected;
  const {
@@ -1796,11 +1794,11 @@ export function SessionProvider({ children, serverUrl, serverId, role }: Session
  );

  useEffect(() => {
-    registerSessionAccessor(serverId, sessionDirectoryEntry, sessionRole);
+    registerSessionAccessor(serverId, sessionDirectoryEntry);
    return () => {
-      unregisterSessionAccessor(serverId, sessionRole);
+      unregisterSessionAccessor(serverId);
    };
-  }, [serverId, sessionRole, registerSessionAccessor, unregisterSessionAccessor, sessionDirectoryEntry]);
+  }, [serverId, registerSessionAccessor, unregisterSessionAccessor, sessionDirectoryEntry]);

  return (
    <SessionContext.Provider value={value}>