From 9764a875e810d0bf5184eb785ca8b1fcc44b6ab4 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Thu, 9 Apr 2026 22:10:55 +0530 Subject: [PATCH] fix(subconscious): seed defaults into per-user workspace + fix Intelligence page stale log (#462) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(subconscious): seed defaults and spawn heartbeat on startup The subconscious engine was only constructed lazily on the first engine-routed RPC (trigger, tasks_add, status). Because handle_tasks_list bypasses the engine and reads the store directly, a fresh install showed an empty Subconscious panel until the user clicked "Run now", even though SubconsciousEngine::new() seeds the 3 default system tasks on construction. Separately, HeartbeatEngine::run() — the periodic tick loop — was never spawned in production code. The only callers of HeartbeatEngine were tests, so ticks never fired automatically; users had to trigger each evaluation manually. Both issues are fixed together in run_server_inner, following the existing start_if_enabled pattern used by voice, screen_intelligence, and autocomplete: 1. Call get_or_init_engine() at startup to construct the SubconsciousEngine eagerly, which runs seed_default_tasks via from_heartbeat_config. Construction is idempotent via OnceLock; seeding is idempotent by title match, so repeat startups do not duplicate the defaults. 2. Construct HeartbeatEngine with the heartbeat config and workspace_dir, then tokio::spawn heartbeat.run() so the periodic tick loop runs for the process lifetime. The loop re-acquires the shared engine via get_or_init_engine() on each tick. Guarded by config.heartbeat.enabled so users who disable the heartbeat get neither startup seeding nor the background loop. Add engine_construction_seeds_default_tasks integration test that locks in the invariant: constructing SubconsciousEngine on a fresh workspace_dir must leave the 3 default system tasks in the store, with no tick, trigger, or explicit seed call. Also asserts that reconstructing the engine on the same workspace does not duplicate the defaults. Co-Authored-By: Claude Opus 4.6 (1M context) * fix(subconscious): defer engine bootstrap until after login Default system tasks seeded at sidecar startup into the pre-login global workspace (`~/.openhuman/workspace/`) instead of the per-user workspace (`~/.openhuman/users//workspace/`) the UI reads from after login. The engine singleton is built lazily via `get_or_init_engine()` and cached in a `OnceLock`. `Config::load_or_init` resolves `workspace_dir` from `active_user.toml` — which does not exist until after login. When the engine was constructed on startup it therefore seeded into the global default, then the frozen singleton kept pointing at that path for the rest of the session while RPC handlers like `tasks_list` re-loaded config per call and read from the correct per-user path, silently returning an empty list. Fix: - `subconscious/global.rs`: add `bootstrap_after_login()` (idempotent via `BOOTSTRAPPED: AtomicBool`) which builds the engine against the now-correct per-user workspace and spawns the heartbeat loop. Track the heartbeat `JoinHandle` in a static so it can be aborted cleanly. Add `reset_engine_for_user_switch()` that aborts the heartbeat, clears the engine option, and resets the bootstrap flag. - `core/jsonrpc.rs`: replace the unconditional eager init on startup with a conditional one that only bootstraps if `active_user.toml` already exists (so a user logged in from a previous session still gets the engine up immediately after restart). - `credentials/ops.rs`: call `bootstrap_after_login()` at the end of `verify_and_store_session` so a fresh login triggers seeding against the per-user workspace. Call `reset_engine_for_user_switch()` in `clear_session` so logout tears down the engine + heartbeat loop and a subsequent login rebuilds them against the new user. Verified locally: sidecar restart with no `active_user.toml` logs "bootstrap deferred — waiting for login"; post-login logs "seeded 3 tasks on init" + "heartbeat periodic loop spawned"; and `subconscious.tasks_list` returns the 3 system defaults from the per-user DB. Co-Authored-By: Claude Opus 4.6 (1M context) * fix(subconscious): bound config load + guard frontend poll Two related fixes for the Intelligence page freezing on a stale subconscious activity-log snapshot while ticks kept progressing in the sidecar. Root cause (backend): the subconscious RPC handlers were the only outlier in the entire JSON-RPC surface that called the raw `Config::load_or_init()` instead of the shared `load_config_with_timeout()` wrapper that every other domain schemas.rs uses (cron, webhooks, voice, team, skills, service, referral, doctor, …). `load_or_init` constructs a fresh `SecretStore` and runs a chain of `decrypt_optional_secret` calls on every invocation, which may IPC to the OS keychain — slow, unbounded, no caching. Under the Intelligence page's 3-second poll (4 parallel RPCs × ~7 keychain round-trips each = ~28 keychain calls every 3s), this pileup was enough to pin the frontend's `Promise.all` past the poll interval. Root cause (frontend): `useSubconscious.refresh()` uses `fetchingRef` as an in-flight guard. The ref is only cleared inside the `finally` block that runs after `Promise.all` settles. With no per-RPC timeout on the client side either, a single slow backend call would leave the ref stuck `true`, and every subsequent 3s `setInterval` tick would silently early-return at the top of `refresh`. The poller kept firing, but every call was a no-op — so the UI froze on whatever snapshot it last successfully fetched, even though the backend was still ticking through new decisions. Backend fix (`src/openhuman/subconscious/schemas.rs`): - Replace the local `load_config()` helper body to delegate to `crate::openhuman::config::load_config_with_timeout()`. Matches the 28 other domain schemas.rs files and brings subconscious handlers under the same 30s bound used everywhere else. Frontend fix (`app/src/hooks/useSubconscious.ts`): - Add a `withTimeout` helper (2.5s per-RPC, strictly less than the 3s poll interval) that races each of the 4 parallel RPCs against a timeout and resolves `null` on timeout — matching the existing `.catch(() => null)` contract so downstream setState logic is unchanged. - Clear `fetchingRef.current = false` in the useEffect cleanup so a late-returning request or a React Strict Mode double-mount in dev can't leave the ref stuck `true` for the next mount. Defense in depth: the backend bound prevents a permanent hang and matches repo conventions, while the frontend bound guarantees the 3s poll loop can never be pinned beyond one tick regardless of server-side latency. Verified locally — `cargo check` clean, `tsc --noEmit` clean, all 18 pre-existing warnings in unrelated modules. Co-Authored-By: Claude Opus 4.6 (1M context) * style(jsonrpc): cargo fmt the startup bootstrap block CI ran `cargo fmt --all -- --check` and flagged the conditional bootstrap block in `run_server_inner` — `let already_logged_in` should fold onto one line, the `.and_then` closure body should inline, the `match ... .await` chain should fold, and the short log!() calls should not break across lines. No behavior change. Fixes three jobs on PR #462 that were all failing at the same `cargo fmt --all -- --check` step (Rust Quality, Rust Tests, Type Check TypeScript — the last one chains cargo fmt after its prettier check). Co-Authored-By: Claude Opus 4.6 (1M context) --------- Co-authored-by: Claude Opus 4.6 (1M context) --- app/src/hooks/useSubconscious.ts | 47 +++++++- src/core/jsonrpc.rs | 31 ++++++ src/openhuman/credentials/ops.rs | 19 ++++ src/openhuman/subconscious/global.rs | 100 ++++++++++++++++++ .../subconscious/integration_test.rs | 62 +++++++++++ src/openhuman/subconscious/schemas.rs | 12 ++- 6 files changed, 263 insertions(+), 8 deletions(-) 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 {