diff --git a/app/src/hooks/useSubconscious.ts b/app/src/hooks/useSubconscious.ts index 7c67c7ada..12bf84ef4 100644 --- a/app/src/hooks/useSubconscious.ts +++ b/app/src/hooks/useSubconscious.ts @@ -66,11 +66,19 @@ export function useSubconscious(): UseSubconsciousResult { setLoading(true); setError(null); try { + // Each RPC is bounded by RPC_TIMEOUT_MS so Promise.all is guaranteed + // to settle. Without this, a single hung request (e.g. sidecar held + // in a long-running tick) would leave fetchingRef.current === true + // forever, and every subsequent 3s poll would silently no-op at the + // early-return above — freezing the Intelligence page on a stale + // snapshot. withTimeout returns null on timeout, matching the + // existing `.catch(() => null)` failure contract, so downstream + // setState calls just skip that slice for this tick. const [tasksRes, escalationsRes, logRes, statusRes] = await Promise.all([ - subconsciousTasksList().catch(() => null), - subconsciousEscalationsList('pending').catch(() => null), - subconsciousLogList(undefined, 30).catch(() => null), - subconsciousStatus().catch(() => null), + withTimeout(subconsciousTasksList()), + withTimeout(subconsciousEscalationsList('pending')), + withTimeout(subconsciousLogList(undefined, 30)), + withTimeout(subconsciousStatus()), ]); if (tasksRes) setTasks(unwrap(tasksRes) ?? []); @@ -167,10 +175,19 @@ export function useSubconscious(): UseSubconsciousResult { // Poll every 3s while the hook is mounted (user is on Subconscious tab). // Picks up all state changes: in_progress → act/noop/escalate/failed, // new escalations, background tick completions, etc. + // + // On unmount we also clear fetchingRef — otherwise a request that times + // out or resolves after the component has been torn down would leave the + // ref stuck `true` for the next mount (React Strict Mode double-mount in + // dev, or tab navigation back to Intelligence), silently wedging the + // poller exactly as before. useEffect(() => { refresh(); const interval = setInterval(refresh, 3000); - return () => clearInterval(interval); + return () => { + clearInterval(interval); + fetchingRef.current = false; + }; }, [refresh]); return { @@ -191,6 +208,26 @@ export function useSubconscious(): UseSubconsciousResult { }; } +/** + * Per-RPC client-side timeout for the polling refresh. Must be strictly + * less than the 3s poll interval so a hung call can't stack up across + * ticks. 2500ms leaves a 500ms safety margin. + */ +const RPC_TIMEOUT_MS = 2500; + +/** + * Race a promise against a timeout. Resolves to `null` on timeout or + * rejection — matching the prior `.catch(() => null)` contract used by + * the refresh logic so downstream code can treat "no data this tick" and + * "RPC failed this tick" identically. + */ +function withTimeout(promise: Promise, ms: number = RPC_TIMEOUT_MS): Promise { + return Promise.race([ + promise.catch(() => null), + new Promise(resolve => setTimeout(() => resolve(null), ms)), + ]); +} + /** * Unwrap a CommandResponse — callCoreRpc returns `{ result: T, logs: [...] }`. */ diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index f8a1dfc5f..a1d861a41 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -703,6 +703,37 @@ async fn run_server_inner( // Initialize screen intelligence engine if enabled in config. crate::openhuman::screen_intelligence::server::start_if_enabled(&config).await; crate::openhuman::autocomplete::start_if_enabled(&config).await; + + // Subconscious engine + heartbeat bootstrap is now gated on + // login so seed_default_tasks() runs against the per-user + // workspace (`~/.openhuman/users//workspace/`) instead + // of the pre-login global workspace. The login handler in + // `credentials::ops` triggers the same bootstrap after it + // writes `active_user.toml`. + // + // If the user is already logged in from a previous session + // (active_user.toml exists on disk), kick the bootstrap now + // so the heartbeat loop starts without waiting for the user + // to re-authenticate. + if !config.heartbeat.enabled { + log::info!("[subconscious] disabled by config (heartbeat.enabled = false)"); + } else { + let already_logged_in = crate::openhuman::config::default_root_openhuman_dir() + .ok() + .and_then(|root| crate::openhuman::config::read_active_user_id(&root)) + .is_some(); + if already_logged_in { + match crate::openhuman::subconscious::global::bootstrap_after_login().await + { + Ok(()) => log::info!( + "[subconscious] bootstrapped on startup (existing session)" + ), + Err(e) => log::warn!("[subconscious] startup bootstrap failed: {e}"), + } + } else { + log::info!("[subconscious] bootstrap deferred — waiting for login"); + } + } } Err(err) => { log::warn!("[core] config load failed, skipping local-ai and overlay: {err}"); diff --git a/src/openhuman/credentials/ops.rs b/src/openhuman/credentials/ops.rs index 7fc0cac67..3ea269918 100644 --- a/src/openhuman/credentials/ops.rs +++ b/src/openhuman/credentials/ops.rs @@ -135,6 +135,19 @@ pub async fn store_session( .map_err(|e| e.to_string())?; logs.push("session stored".to_string()); + + // Now that active_user.toml exists and config.workspace_dir resolves to + // the per-user path, seed the subconscious defaults and spawn the + // heartbeat loop. Idempotent — no-op on subsequent logins of the same + // process. Bootstrap failures are non-fatal: the session itself is + // already stored above, so we only warn. + if let Err(e) = crate::openhuman::subconscious::global::bootstrap_after_login().await { + tracing::warn!(error = %e, "[subconscious] post-login bootstrap failed"); + logs.push(format!("subconscious bootstrap warning: {e}")); + } else { + logs.push("subconscious engine bootstrapped".to_string()); + } + Ok(RpcOutcome::new(summarize_auth_profile(&profile), logs)) } @@ -152,6 +165,12 @@ pub async fn clear_session(config: &Config) -> Result/workspace/`) instead of the +//! pre-login global default. See `load.rs::resolve_runtime_config_dirs` for +//! how `active_user.toml` drives `config.workspace_dir`. use super::engine::SubconsciousEngine; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, OnceLock}; use tokio::sync::Mutex; +use tokio::task::JoinHandle; static ENGINE: OnceLock>>> = OnceLock::new(); +/// True once [`bootstrap_after_login`] has successfully seeded the engine and +/// spawned the heartbeat loop for the current active user. +static BOOTSTRAPPED: AtomicBool = AtomicBool::new(false); + +/// Heartbeat loop handle so logout / user switch can abort it cleanly. +static HEARTBEAT_HANDLE: OnceLock>>> = OnceLock::new(); + fn engine_lock() -> &'static Arc>> { ENGINE.get_or_init(|| Arc::new(Mutex::new(None))) } +fn heartbeat_slot() -> &'static Mutex>> { + HEARTBEAT_HANDLE.get_or_init(|| Mutex::new(None)) +} + /// Get or initialize the global engine. Both heartbeat loop and RPC use this. pub async fn get_or_init_engine() -> Result>>, String> { let lock = engine_lock(); @@ -42,3 +61,84 @@ pub async fn get_or_init_engine() -> Result Ok(Arc::clone(lock)) } + +/// Construct the engine (which seeds defaults into the per-user workspace) +/// and spawn the heartbeat loop. Idempotent per-process via [`BOOTSTRAPPED`]. +/// +/// Call this: +/// - after a successful login writes `active_user.toml`, OR +/// - at sidecar startup **iff** `active_user.toml` already exists. +/// +/// Calling before login would seed into the global pre-login workspace and +/// then silently diverge from the per-user workspace the UI reads from. +pub async fn bootstrap_after_login() -> Result<(), String> { + if BOOTSTRAPPED.swap(true, Ordering::SeqCst) { + tracing::debug!("[subconscious] bootstrap already ran — skipping"); + return Ok(()); + } + + let config = crate::openhuman::config::Config::load_or_init() + .await + .map_err(|e| { + BOOTSTRAPPED.store(false, Ordering::SeqCst); + format!("load config: {e}") + })?; + + if !config.heartbeat.enabled { + tracing::info!("[subconscious] heartbeat disabled in config — bootstrap skipped"); + return Ok(()); + } + + // Build the engine against the NOW-correct per-user workspace_dir. + // SubconsciousEngine::new calls seed_default_tasks() inside the + // constructor, so by the time this returns the 3 system defaults are + // present in `/subconscious/subconscious.db`. + get_or_init_engine().await.map_err(|e| { + BOOTSTRAPPED.store(false, Ordering::SeqCst); + e + })?; + tracing::info!( + workspace = %config.workspace_dir.display(), + "[subconscious] engine initialized against per-user workspace" + ); + + // Spawn the heartbeat loop and keep the JoinHandle so we can cancel it + // on logout. Without this the task would leak: tokio::spawn returns a + // detached task that drops on handle-drop but keeps running. + let heartbeat = crate::openhuman::heartbeat::engine::HeartbeatEngine::new( + config.heartbeat.clone(), + config.workspace_dir.clone(), + ); + let handle = tokio::spawn(async move { + if let Err(e) = heartbeat.run().await { + tracing::warn!("[heartbeat] loop exited with error: {e}"); + } + }); + *heartbeat_slot().lock().await = Some(handle); + tracing::info!( + "[heartbeat] periodic loop spawned ({}min interval)", + config.heartbeat.interval_minutes + ); + + Ok(()) +} + +/// Tear down the engine + heartbeat loop so the next login rebuilds them +/// against the new user's workspace. Call on logout or account switch. +/// +/// Without this, the engine `OnceLock` would stay frozen on the previous +/// user's `workspace_dir` and subsequent ticks / RPC queries would leak +/// into the wrong DB. +pub async fn reset_engine_for_user_switch() { + if let Some(handle) = heartbeat_slot().lock().await.take() { + handle.abort(); + tracing::info!("[heartbeat] loop aborted for user switch"); + } + + let lock = engine_lock(); + let mut guard = lock.lock().await; + *guard = None; + + BOOTSTRAPPED.store(false, Ordering::SeqCst); + tracing::info!("[subconscious] engine reset for user switch"); +} diff --git a/src/openhuman/subconscious/integration_test.rs b/src/openhuman/subconscious/integration_test.rs index d65f73330..7bdcd0829 100644 --- a/src/openhuman/subconscious/integration_test.rs +++ b/src/openhuman/subconscious/integration_test.rs @@ -222,4 +222,66 @@ mod tests { }) .unwrap(); } + + /// Regression test for the "empty task list on fresh install" bug. + /// + /// The core server's startup path calls `get_or_init_engine()` to + /// eagerly construct a `SubconsciousEngine`, relying on the constructor + /// to seed the 3 default system tasks. This test locks in that + /// invariant: constructing the engine alone — with no tick, no + /// trigger RPC, and no explicit seed call — must leave the 3 defaults + /// in the SQLite store. + #[test] + fn engine_construction_seeds_default_tasks() { + use crate::openhuman::config::HeartbeatConfig; + use crate::openhuman::subconscious::SubconsciousEngine; + + let dir = tempfile::tempdir().unwrap(); + let workspace = dir.path().to_path_buf(); + + // Construct the engine via the same path the core server uses at + // startup. Memory client is not required for seeding. + let _engine = SubconsciousEngine::from_heartbeat_config( + &HeartbeatConfig::default(), + workspace.clone(), + None, + ); + + // The 3 default system tasks must now exist in the store. + store::with_connection(&workspace, |conn| { + let tasks = store::list_tasks(conn, false)?; + assert_eq!( + tasks.len(), + 3, + "engine construction must seed the 3 default system tasks" + ); + assert!(tasks.iter().all(|t| t.source == TaskSource::System)); + assert!(tasks + .iter() + .all(|t| t.recurrence == TaskRecurrence::Pending)); + + Ok(()) + }) + .unwrap(); + + // Reconstructing the engine on the same workspace must not + // duplicate the defaults — seed_default_tasks is idempotent. + + let _engine2 = SubconsciousEngine::from_heartbeat_config( + &HeartbeatConfig::default(), + workspace.clone(), + None, + ); + + store::with_connection(&workspace, |conn| { + let tasks = store::list_tasks(conn, false)?; + assert_eq!( + tasks.len(), + 3, + "repeat engine construction must not duplicate default tasks" + ); + Ok(()) + }) + .unwrap(); + } } diff --git a/src/openhuman/subconscious/schemas.rs b/src/openhuman/subconscious/schemas.rs index dd094b644..587e89879 100644 --- a/src/openhuman/subconscious/schemas.rs +++ b/src/openhuman/subconscious/schemas.rs @@ -438,9 +438,15 @@ fn handle_escalations_dismiss(params: Map) -> ControllerFuture { // ── Helpers ────────────────────────────────────────────────────────────────── async fn load_config() -> Result { - crate::openhuman::config::Config::load_or_init() - .await - .map_err(|e| format!("load config: {e}")) + // Use the same 30s-bounded loader every other JSON-RPC domain uses + // (see cron/schemas.rs, webhooks/schemas.rs, etc.). Raw + // `Config::load_or_init()` can stall on `SecretStore::new` plus a chain + // of `decrypt_optional_secret` calls that may IPC to an OS keychain, + // so the subconscious handlers used to be the only unbounded outlier + // in the entire JSON-RPC surface. Under the Intelligence page's 3s + // poll that chokepoint let a slow keychain call pin the frontend's + // `Promise.all` and freeze the activity log on a stale snapshot. + crate::openhuman::config::load_config_with_timeout().await } fn field(name: &'static str, ty: TypeSchema, comment: &'static str) -> FieldSchema {