diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 255362b21..2f3547c53 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -62,7 +62,7 @@ use huddle::{ start_huddle, start_stt_pipeline, HuddlePhase, }; use initial_window::*; -use managed_agents::store_journal::{run_recovery_gate, store_anchor_dir}; +use managed_agents::store_journal::run_boot_recovery_gate; use managed_agents::{ backfill_persona_snapshots, ensure_nest, list_managed_agent_runtimes, put_managed_agent_runtime_lifecycle, reconcile_managed_agent_runtimes, @@ -341,27 +341,14 @@ pub fn run() { return Ok(()); } - // run_recovery_gate: file-commit recovery before migrations; fail closed. + // Pre-admission recovery gate: file-commit recovery runs here — + // before migrations, backfill, and every canonical-store reader/writer. + // Fails closed: anchor error, unresolved commits, or migration error + // all set store_recovery_failed and skip all store-touching setup. { let state = app_handle.state::(); - let anchor = store_anchor_dir(&app_handle).unwrap_or_else(|_| { - app_handle - .path() - .app_data_dir() - .unwrap_or_default() - .join("agents") - }); - let _ = std::fs::create_dir_all(&anchor); - let reset_done = reset_outcome.completed; - if let Err(e) = run_recovery_gate(&anchor, || { - if reset_done { - migration::run_boot_migrations_after_reset(&app_handle); - } else { - migration::run_boot_migrations(&app_handle); - } - Ok(()) - }) { - eprintln!("buzz-desktop: file-commit-recovery: {e}"); + if let Err(e) = run_boot_recovery_gate(&app_handle, reset_outcome.completed) { + eprintln!("buzz-desktop: boot-recovery-gate: {e}"); state.store_recovery_failed.store(true, Ordering::Release); return Ok(()); } diff --git a/desktop/src-tauri/src/managed_agents/store_journal/boot.rs b/desktop/src-tauri/src/managed_agents/store_journal/boot.rs new file mode 100644 index 000000000..45c4402be --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/store_journal/boot.rs @@ -0,0 +1,40 @@ +//! Pre-admission recovery gate: resolve anchor, run file-commit recovery, then +//! invoke boot migrations — all before any canonical-store reader or writer. +//! +//! Called once per process from `lib.rs` setup, immediately after reset +//! handling succeeds. A recovery failure (or unresolved interrupted commits) +//! propagates as `Err`; the caller sets `store_recovery_failed` and returns +//! early, keeping every journaled mutation path closed via `mutate_store`'s +//! entry guard. + +use tauri::AppHandle; + +use super::{run_recovery_gate, store_anchor_dir}; + +/// Run the pre-admission recovery gate for this process boot. +/// +/// 1. Resolves the store-family anchor directory — fails closed on error +/// (no `unwrap_or_default` fallback; a guessed path would inspect the wrong +/// journal and certify a store it never repaired). +/// 2. Creates the anchor directory if absent. +/// 3. Runs `run_recovery_gate`: file-commit recovery completes with zero +/// unresolved commits, then `store_work` executes. +/// 4. `store_work` runs the appropriate boot-migration sequence based on +/// `reset_completed`. +/// +/// Returns `Err` on any anchor, recovery, or migration failure. The caller +/// (`lib.rs`) sets `store_recovery_failed` and early-returns. +pub fn run_boot_recovery_gate(app: &AppHandle, reset_completed: bool) -> Result<(), String> { + let anchor = store_anchor_dir(app).map_err(|e| format!("resolve store anchor: {e}"))?; + + std::fs::create_dir_all(&anchor).map_err(|e| format!("create anchor dir: {e}"))?; + + run_recovery_gate(&anchor, || { + if reset_completed { + crate::migration::run_boot_migrations_after_reset(app); + } else { + crate::migration::run_boot_migrations(app); + } + Ok(()) + }) +} diff --git a/desktop/src-tauri/src/managed_agents/store_journal/mod.rs b/desktop/src-tauri/src/managed_agents/store_journal/mod.rs index abcb46c59..2e548aac0 100644 --- a/desktop/src-tauri/src/managed_agents/store_journal/mod.rs +++ b/desktop/src-tauri/src/managed_agents/store_journal/mod.rs @@ -20,6 +20,7 @@ //! writers, sign-out/reset races, or concurrent bundles. mod anchor; +mod boot; mod codec; mod events; mod generations; @@ -36,6 +37,8 @@ mod writer; pub use anchor::canonical_dev_anchor_pub; pub use anchor::store_anchor_dir; +pub use boot::run_boot_recovery_gate; + #[cfg(test)] #[allow(unused_imports)] pub use codec::StoreDecodeError; @@ -67,7 +70,9 @@ pub use txn::{ }; #[cfg(test)] #[allow(unused_imports)] -pub(crate) use txn::{file_commit_recovery_at_pub, read_store, run_boot_recovery_at}; +pub(crate) use txn::{ + file_commit_recovery_at_pub, read_store, reject_if_recovery_failed, run_boot_recovery_at, +}; pub use util::new_operation_id; diff --git a/desktop/src-tauri/src/managed_agents/store_journal/txn.rs b/desktop/src-tauri/src/managed_agents/store_journal/txn.rs index 43f490d91..d1646979c 100644 --- a/desktop/src-tauri/src/managed_agents/store_journal/txn.rs +++ b/desktop/src-tauri/src/managed_agents/store_journal/txn.rs @@ -120,6 +120,24 @@ where store_work() } +/// Reject a store mutation when the boot recovery flag is set. +/// +/// Extracted for direct unit testing — `mutate_store` delegates to this +/// function so the admission guard is testable without a full AppHandle. +#[cfg_attr(test, allow(dead_code))] +pub(crate) fn reject_if_recovery_failed( + flag: &std::sync::atomic::AtomicBool, +) -> Result<(), String> { + if flag.load(std::sync::atomic::Ordering::Acquire) { + return Err( + "store mutation rejected: boot file-commit recovery failed; \ + relaunch to retry" + .to_string(), + ); + } + Ok(()) +} + /// Mutate the store under the full lock sequence. /// /// The closure runs inside a real `rusqlite::Transaction` — if it returns @@ -142,17 +160,8 @@ where { // Guard: a boot recovery failure means the store is in an uncertain state. // Reject all mutations until the user relaunches and recovery succeeds. - { - use std::sync::atomic::Ordering; - let state = app.state::(); - if state.store_recovery_failed.load(Ordering::Acquire) { - return Err( - "store mutation rejected: boot file-commit recovery failed; \ - relaunch to retry" - .to_string(), - ); - } - } + let state = app.state::(); + reject_if_recovery_failed(&state.store_recovery_failed)?; let anchor = store_anchor_dir(app)?; std::fs::create_dir_all(&anchor).map_err(|e| format!("create anchor dir: {e}"))?; diff --git a/desktop/src-tauri/src/managed_agents/store_journal_fix_tests.rs b/desktop/src-tauri/src/managed_agents/store_journal_fix_tests.rs index fd0649b6d..780b6bb0e 100644 --- a/desktop/src-tauri/src/managed_agents/store_journal_fix_tests.rs +++ b/desktop/src-tauri/src/managed_agents/store_journal_fix_tests.rs @@ -6,7 +6,7 @@ use super::operations::insert_operation; use super::{ apply_journal_schema_pub, file_commit_recovery_at_pub, insert_outbox_event, open_journal, - run_boot_recovery_at, run_recovery_gate, Generation, + reject_if_recovery_failed, run_boot_recovery_at, run_recovery_gate, Generation, }; use crate::managed_agents::retention::{ get_pending_sync, open_retention_db, tombstone_retention_d_tag, @@ -840,31 +840,27 @@ fn v2_journal() -> rusqlite::Connection { conn } +/// Helper: returns column names of `table` via `PRAGMA table_info`. +fn table_cols(conn: &rusqlite::Connection, table: &str) -> Vec { + let mut stmt = conn + .prepare(&format!("PRAGMA table_info({table})")) + .unwrap(); + stmt.query_map([], |r| r.get::<_, String>(1)) + .unwrap() + .collect::, _>>() + .unwrap() +} + /// Fix B: upgrade from v2 (no hash columns, no d-tag) adds all three columns /// and stamps version 4. #[test] fn test_schema_migration_from_v2_adds_all_columns() { let conn = v2_journal(); apply_journal_schema_pub(&conn).unwrap(); - // All three columns must exist after migration. - let fcp_cols: Vec = { - let mut stmt = conn - .prepare("PRAGMA table_info(file_commit_phases)") - .unwrap(); - stmt.query_map([], |r| r.get::<_, String>(1)) - .unwrap() - .collect::, _>>() - .unwrap() - }; + let fcp_cols = table_cols(&conn, "file_commit_phases"); assert!(fcp_cols.contains(&"agents_content_hash".to_string())); assert!(fcp_cols.contains(&"teams_content_hash".to_string())); - let oe_cols: Vec = { - let mut stmt = conn.prepare("PRAGMA table_info(outbox_events)").unwrap(); - stmt.query_map([], |r| r.get::<_, String>(1)) - .unwrap() - .collect::, _>>() - .unwrap() - }; + let oe_cols = table_cols(&conn, "outbox_events"); assert!(oe_cols.contains(&"retention_d_tag".to_string())); let ver: u32 = conn .pragma_query_value(None, "user_version", |r| r.get(0)) @@ -887,13 +883,7 @@ fn test_schema_migration_from_partial_v3_adds_missing_d_tag() { ) .unwrap(); apply_journal_schema_pub(&conn).unwrap(); - let oe_cols: Vec = { - let mut stmt = conn.prepare("PRAGMA table_info(outbox_events)").unwrap(); - stmt.query_map([], |r| r.get::<_, String>(1)) - .unwrap() - .collect::, _>>() - .unwrap() - }; + let oe_cols = table_cols(&conn, "outbox_events"); assert!( oe_cols.contains(&"retention_d_tag".to_string()), "d-tag must be added from partial v3" @@ -944,24 +934,10 @@ fn test_schema_migration_failure_does_not_advance_version() { apply_journal_schema_pub(&conn).unwrap(); // All three columns must be present and version must reach 4. - let fcp_cols: Vec = { - let mut stmt = conn - .prepare("PRAGMA table_info(file_commit_phases)") - .unwrap(); - stmt.query_map([], |r| r.get::<_, String>(1)) - .unwrap() - .collect::, _>>() - .unwrap() - }; + let fcp_cols = table_cols(&conn, "file_commit_phases"); assert!(fcp_cols.contains(&"agents_content_hash".to_string())); assert!(fcp_cols.contains(&"teams_content_hash".to_string())); - let oe_cols: Vec = { - let mut stmt = conn.prepare("PRAGMA table_info(outbox_events)").unwrap(); - stmt.query_map([], |r| r.get::<_, String>(1)) - .unwrap() - .collect::, _>>() - .unwrap() - }; + let oe_cols = table_cols(&conn, "outbox_events"); assert!(oe_cols.contains(&"retention_d_tag".to_string())); let ver2: u32 = conn .pragma_query_value(None, "user_version", |r| r.get(0)) @@ -987,3 +963,31 @@ fn test_schema_migration_idempotent_on_v4() { .unwrap(); assert_eq!(ver2, 4); } + +// ── Round-13: admission-guard unit tests ───────────────────────────────────── + +/// GAP A: `reject_if_recovery_failed` returns `Ok` when the flag is clear. +#[test] +fn test_reject_if_recovery_failed_clear_returns_ok() { + let flag = std::sync::atomic::AtomicBool::new(false); + assert!( + reject_if_recovery_failed(&flag).is_ok(), + "must return Ok when store_recovery_failed is false" + ); +} + +/// GAP A: `reject_if_recovery_failed` returns `Err` when the flag is set, +/// proving the admission guard closes the mutation path on recovery failure. +#[test] +fn test_reject_if_recovery_failed_set_returns_err() { + let flag = std::sync::atomic::AtomicBool::new(true); + let result = reject_if_recovery_failed(&flag); + assert!( + result.is_err(), + "must return Err when store_recovery_failed is true" + ); + assert!( + result.unwrap_err().contains("relaunch to retry"), + "error message must mention relaunch" + ); +}