fix(auth): stop 401 cascade after session expiry (OPENHUMAN-TAURI-1T) (tinyhumansai#1516)

Author: senamakel

app/src-tauri/vendor/tauri-cef

@@ -570,37 +570,62 @@ export default function CoreStateProvider({ children }: { children: ReactNode })
     });
   }, [commitState, refresh]);
 
+  // Listen for two flavours of session expiry, both routed through the
+  // same debounced `clearSession`:
+  //
+  // 1. `core-rpc-auth-expired` — emitted by `coreRpcClient` when an
+  //    individual RPC call returns 401 (usage pill, upsell banner,
+  //    threads poll, …). Multiple parallel chains can fire it in the
+  //    same frame after a token expires; the 10s debounce coalesces
+  //    them so `clearSession` only runs once.
+  // 2. `openhuman:session-expired` — emitted by `socketService` when
+  //    the core pushes `auth:session_expired` over Socket.IO (the
+  //    OpenHuman backend provider's `api_error` published
+  //    `DomainEvent::SessionExpired`, or `jsonrpc::invoke_method`
+  //    detected a 401 on a server-side method call). Without this, the
+  //    UI keeps showing a logged-in shell until the next refresh()
+  //    discovers the missing token — confusing, and a security smell
+  //    on shared devices.
+  //
+  // Depends on `clearSession` so the listener always closes over the
+  // latest closure; `clearSession`'s own deps are stable `useCallback`s,
+  // so re-registers are rare.
   useEffect(() => {
-    const onAuthExpired = (event: Event) => {
-      const customEvent = event as CustomEvent<{ method?: string; source?: string }>;
+    const runReauth = (method: string, source: string) => {
       const now = Date.now();
-      // Multiple parallel RPC chains (usage pill + upsell banner + threads
-      // poll) can each surface a 401 within the same frame after a token
-      // expires. Debounce so clearSession only runs once per real auth-loss
-      // event — the 10s window covers any straggler chains without blocking
-      // a legitimate re-login that immediately fails again.
       if (now - lastReauthAtRef.current < 10_000) {
-        log(
-          'auth-expired debounced (method=%s source=%s)',
-          customEvent.detail?.method ?? 'unknown',
-          customEvent.detail?.source ?? 'unknown'
-        );
+        log('auth-expired debounced (method=%s source=%s)', method, source);
         return;
       }
       lastReauthAtRef.current = now;
-      log(
-        'auth-expired: clearing session (method=%s source=%s)',
-        customEvent.detail?.method ?? 'unknown',
-        customEvent.detail?.source ?? 'unknown'
-      );
+      log('auth-expired: clearing session (method=%s source=%s)', method, source);
       void clearSession().catch(err => {
         log('clearSession failed after auth-expired: %O', sanitizeError(err));
       });
     };
 
-    window.addEventListener('core-rpc-auth-expired', onAuthExpired as EventListener);
+    const onRpcExpired = (event: Event) => {
+      const detail = (event as CustomEvent<{ method?: string; source?: string }>).detail;
+      runReauth(detail?.method ?? 'unknown', detail?.source ?? 'core-rpc-auth-expired');
+    };
+
+    const onSocketExpired = (event: Event) => {
+      const source =
+        event instanceof CustomEvent &&
+        event.detail &&
+        typeof event.detail === 'object' &&
+        'source' in event.detail &&
+        typeof (event.detail as { source?: unknown }).source === 'string'
+          ? (event.detail as { source: string }).source
+          : 'unknown';
+      runReauth('socket.session_expired', source);
+    };
+
+    window.addEventListener('core-rpc-auth-expired', onRpcExpired as EventListener);
+    window.addEventListener('openhuman:session-expired', onSocketExpired as EventListener);
     return () => {
-      window.removeEventListener('core-rpc-auth-expired', onAuthExpired as EventListener);
+      window.removeEventListener('core-rpc-auth-expired', onRpcExpired as EventListener);
+      window.removeEventListener('openhuman:session-expired', onSocketExpired as EventListener);
     };
   }, [clearSession]);
 

app/src/providers/CoreStateProvider.tsx

@@ -246,6 +246,25 @@ class SocketService {
     this.socket.on('channel:connection-updated', handleChannelConnectionUpdated);
     this.socket.on('channel_connection_updated', handleChannelConnectionUpdated);
 
+    // Core-side session expiry (401 from the OpenHuman backend or jsonrpc).
+    // The server has already published SessionExpired on its event bus,
+    // the credentials subscriber has cleared the JWT, and the scheduler
+    // gate is flipped to signed-out. All the UI needs to do is mirror
+    // that locally and route to onboarding. CoreStateProvider listens
+    // for the window event below and calls its own `clearSession`.
+    const handleSessionExpired = (data: unknown) => {
+      const source =
+        (data && typeof data === 'object' && 'source' in data && typeof data.source === 'string'
+          ? data.source
+          : undefined) ?? 'unknown';
+      socketLog('Session expired notification received', { source });
+      if (typeof window !== 'undefined') {
+        window.dispatchEvent(new CustomEvent('openhuman:session-expired', { detail: { source } }));
+      }
+    };
+    this.socket.on('auth:session_expired', handleSessionExpired);
+    this.socket.on('auth_session_expired', handleSessionExpired);
+
     this.socket.on('channel:managed-dm-verified', data => {
       const obj = data as Record<string, unknown> | null;
       if (!obj || typeof obj !== 'object') return;

app/src/services/socketService.ts

@@ -102,6 +102,16 @@ impl EventBus {
         Self { tx }
     }
 
+    /// Get a raw broadcast receiver. Used by bridges that need to drive
+    /// their own filtering loop (e.g. the Socket.IO `auth:session_expired`
+    /// forwarder) instead of going through the [`EventHandler`] trait.
+    /// Prefer [`Self::subscribe`] for normal subscribers — it provides
+    /// domain filtering, panic isolation, and lifetime via
+    /// [`SubscriptionHandle`].
+    pub fn raw_receiver(&self) -> broadcast::Receiver<DomainEvent> {
+        self.tx.subscribe()
+    }
+
     /// Publish an event to all active subscribers.
     ///
     /// The event is cloned and sent to each subscriber's receiving end.
@@ -242,6 +252,24 @@ mod tests {
         });
     }
 
+    #[tokio::test]
+    async fn raw_receiver_observes_published_events() {
+        // raw_receiver is the escape hatch used by the Socket.IO bridge
+        // (auth:session_expired forwarder). Confirm it sees published
+        // events the same way a trait-based subscriber would.
+        let bus = EventBus::create(16);
+        let mut rx = bus.raw_receiver();
+        bus.publish(DomainEvent::SessionExpired {
+            source: "test".into(),
+            reason: "401".into(),
+        });
+        let event = tokio::time::timeout(Duration::from_millis(200), rx.recv())
+            .await
+            .expect("recv should not hang")
+            .expect("recv should not error");
+        assert!(matches!(event, DomainEvent::SessionExpired { .. }));
+    }
+
     #[tokio::test]
     async fn single_subscriber_receives_event() {
         let bus = EventBus::create(16);

src/core/event_bus/bus.rs

@@ -421,6 +421,19 @@ pub enum DomainEvent {
     },
     /// A component restart was observed.
     HealthRestarted { component: String },
+
+    // ── Auth ────────────────────────────────────────────────────────────
+    /// The local app session is no longer valid — typically detected when
+    /// the backend returns 401 to an LLM inference call or a JSON-RPC
+    /// method. Subscribers tear down the session and pause background
+    /// LLM-bound work until the user signs back in.
+    ///
+    /// `source` is a short slug (e.g. `"llm_provider.openhuman_backend"`,
+    /// `"jsonrpc.invoke_method"`) so subscribers and logs can attribute
+    /// the trigger. `reason` is the sanitized error message that caused
+    /// detection (already redacted by the call site) — surfaced to logs,
+    /// never to Sentry or the UI verbatim.
+    SessionExpired { source: String, reason: String },
 }
 
 impl DomainEvent {
@@ -490,6 +503,8 @@ impl DomainEvent {
             | Self::SystemShutdownRequested { .. }
             | Self::HealthChanged { .. }
             | Self::HealthRestarted { .. } => "system",
+
+            Self::SessionExpired { .. } => "auth",
         }
     }
 }

src/core/event_bus/events.rs

@@ -438,6 +438,14 @@ fn all_variants_have_correct_domain() {
             },
             "learning",
         ),
+        // Auth
+        (
+            DomainEvent::SessionExpired {
+                source: "test".into(),
+                reason: "401".into(),
+            },
+            "auth",
+        ),
     ];
 
     for (event, expected_domain) in cases {

src/core/event_bus/events_tests.rs

@@ -100,17 +100,28 @@ pub async fn rpc_handler(State(state): State<AppState>, Json(req): Json<RpcReque
 pub async fn invoke_method(state: AppState, method: &str, params: Value) -> Result<Value, String> {
     let result = invoke_method_inner(state, method, params).await;
 
-    // Session auto-cleanup: If the backend says we're unauthorized,
-    // we should reflect that locally by clearing the stored token.
+    // Session auto-cleanup: if the backend says we're unauthorized, publish
+    // a `SessionExpired` event. The credentials subscriber clears the stored
+    // token, flips the scheduler-gate signed-out override so background
+    // workers stand down, and (eventually) pushes a sign-out to the UI.
+    // Centralising via the event bus means 401 detection from any path
+    // (this one, `llm_provider.api_error`, …) gets the same teardown.
     if let Err(ref msg) = result {
         if is_session_expired_error(msg) {
             log::warn!(
-                "[jsonrpc] backend returned 401 for method '{}' — clearing stored session",
+                "[jsonrpc] backend returned 401 for method '{}' — publishing SessionExpired",
                 method
             );
-            if let Ok(config) = crate::openhuman::config::rpc::load_config_with_timeout().await {
-                let _ = crate::openhuman::credentials::rpc::clear_session(&config).await;
-            }
+            // Scrub before publishing — subscribers log `reason`, and the
+            // upstream error string could include API keys / tokens from
+            // pasted-through provider replies. `sanitize_api_error` runs
+            // `scrub_secret_patterns` and truncates.
+            crate::core::event_bus::publish_global(
+                crate::core::event_bus::DomainEvent::SessionExpired {
+                    source: format!("jsonrpc.invoke_method:{method}"),
+                    reason: crate::openhuman::providers::ops::sanitize_api_error(msg),
+                },
+            );
         }
     }
 
@@ -944,6 +955,42 @@ fn register_domain_subscribers(
         // (otherwise they fall back to `Policy::Normal` and miss the
         // initial throttle decision on battery-powered hosts).
         crate::openhuman::scheduler_gate::init_global(&config);
+
+        // Seed the scheduler-gate signed-out override from the on-disk
+        // session. Without this, a sidecar that boots with no stored JWT
+        // would happily spin up cron / channel loops and fire LLM requests
+        // that all 401 immediately.
+        match crate::api::jwt::get_session_token(&config) {
+            Ok(Some(_)) => {
+                crate::openhuman::scheduler_gate::set_signed_out(false);
+            }
+            Ok(None) => {
+                log::info!(
+                    "[auth] no session token at startup — scheduler gate set to signed_out"
+                );
+                crate::openhuman::scheduler_gate::set_signed_out(true);
+            }
+            Err(err) => {
+                log::warn!(
+                    "[auth] failed to read session token at startup ({err}) — assuming signed_out"
+                );
+                crate::openhuman::scheduler_gate::set_signed_out(true);
+            }
+        }
+
+        // Register the SessionExpired handler before any subscribers that
+        // might publish 401-derived events, so the very first 401 is
+        // routed through `clear_session` + the scheduler-gate override.
+        if let Some(handle) = crate::core::event_bus::subscribe_global(Arc::new(
+            crate::openhuman::credentials::bus::SessionExpiredSubscriber::new(),
+        )) {
+            std::mem::forget(handle);
+        } else {
+            log::warn!(
+                "[event_bus] failed to register SessionExpired subscriber — bus not initialized"
+            );
+        }
+
         crate::openhuman::memory::tree::jobs::start(config.clone());
 
         // Restart requests go through a subscriber so every trigger path shares
@@ -970,7 +1017,7 @@ fn register_domain_subscribers(
         crate::openhuman::agent::bus::register_agent_handlers();
 
         log::info!(
-            "[event_bus] domain subscribers registered (webhook, channel, health, conversation, composio, restart, proactive, agent)"
+            "[event_bus] domain subscribers registered (webhook, channel, health, conversation, composio, restart, proactive, agent, session_expired)"
         );
     });
 }

src/core/jsonrpc.rs

@@ -333,6 +333,7 @@ pub fn spawn_web_channel_bridge(io: SocketIo) {
     let io_overlay = io.clone();
     let io_notify = io.clone();
     let io_transcription = io.clone();
+    let io_auth = io.clone();
 
     // 2. Dictation hotkey events → broadcast to all connected clients.
     tokio::spawn(async move {
@@ -420,6 +421,69 @@ pub fn spawn_web_channel_bridge(io: SocketIo) {
         log::debug!("[socketio] core_notification bridge stopped");
     });
 
+    // 6. SessionExpired events → broadcast to all clients so the UI can
+    //    proactively tear down user-scoped state and route to onboarding
+    //    instead of waiting for the next poll to discover the JWT is gone.
+    //    Subscribes to the global event bus and filters for
+    //    `DomainEvent::SessionExpired`; ignores everything else.
+    tokio::spawn(async move {
+        // Poll until `event_bus::init_global` has run. Socket.IO bridges
+        // spawn from `spawn_web_channel_bridge`, which on some startup
+        // paths runs before `register_domain_subscribers` initialises
+        // the bus. A one-shot check would silently no-op for the rest
+        // of the process; a short polling loop with a hard cap retries
+        // without spinning forever if init genuinely never happens
+        // (e.g. tests that drive the socket layer in isolation).
+        let bus = {
+            const RETRY_INTERVAL_MS: u64 = 250;
+            const MAX_WAIT_SECS: u64 = 30;
+            let max_attempts = (MAX_WAIT_SECS * 1000) / RETRY_INTERVAL_MS;
+            let mut attempts: u64 = 0;
+            loop {
+                if let Some(bus) = crate::core::event_bus::global() {
+                    break bus;
+                }
+                attempts += 1;
+                if attempts > max_attempts {
+                    log::warn!(
+                        "[socketio] event_bus not initialised after {}s — SessionExpired bridge giving up",
+                        MAX_WAIT_SECS
+                    );
+                    return;
+                }
+                tokio::time::sleep(std::time::Duration::from_millis(RETRY_INTERVAL_MS)).await;
+            }
+        };
+        let mut rx = bus.raw_receiver();
+        loop {
+            let event = match rx.recv().await {
+                Ok(event) => event,
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
+                    log::warn!(
+                        "[socketio] dropped {} event_bus events due to lag (auth bridge)",
+                        skipped
+                    );
+                    continue;
+                }
+                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+            };
+            if let crate::core::event_bus::DomainEvent::SessionExpired { source, reason } = event {
+                log::info!(
+                    "[socketio] broadcast auth:session_expired source={} reason_len={}",
+                    source,
+                    reason.len()
+                );
+                // The UI doesn't need the raw reason (already logged
+                // server-side and we don't want auth-error strings in the
+                // renderer console). Just send the source slug.
+                let payload = serde_json::json!({ "source": source });
+                let _ = io_auth.emit("auth:session_expired", &payload);
+                let _ = io_auth.emit("auth_session_expired", &payload);
+            }
+        }
+        log::debug!("[socketio] auth session_expired bridge stopped");
+    });
+
     // 5. Transcription results → broadcast to all connected clients.
     tokio::spawn(async move {
         let mut rx = crate::openhuman::voice::dictation_listener::subscribe_transcription_results();