refactor(store-journal): extract boot recovery gate and admission guard

Move the anchor-resolve + run_recovery_gate + migrations block from lib.rs
into store_journal/boot.rs:run_boot_recovery_gate(app, reset_completed).
lib.rs is reduced from 999 to 988 lines (within the 1000-line ratchet on
the merge tree).

Fail closed on anchor-resolution failure: store_anchor_dir Err propagates
as Err from run_boot_recovery_gate with no unwrap_or_default fallback.

Extract reject_if_recovery_failed(flag: &AtomicBool) -> Result<(), String>
from mutate_store's inline block so the admission guard is unit-testable.
mutate_store delegates to it; two new tests cover the clear and set cases.

Add table_cols helper in fix_tests to eliminate repeated PRAGMA table_info
boilerplate (keeps store_journal_fix_tests.rs at 993 lines, under 1000).

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>

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::<AppState>();
-                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(());
                 }

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(())
+    })
+}

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;
 

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::<crate::app_state::AppState>();
-        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::<crate::app_state::AppState>();
+    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}"))?;

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<String> {
+    let mut stmt = conn
+        .prepare(&format!("PRAGMA table_info({table})"))
+        .unwrap();
+    stmt.query_map([], |r| r.get::<_, String>(1))
+        .unwrap()
+        .collect::<Result<Vec<_>, _>>()
+        .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<String> = {
-        let mut stmt = conn
-            .prepare("PRAGMA table_info(file_commit_phases)")
-            .unwrap();
-        stmt.query_map([], |r| r.get::<_, String>(1))
-            .unwrap()
-            .collect::<Result<Vec<_>, _>>()
-            .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<String> = {
-        let mut stmt = conn.prepare("PRAGMA table_info(outbox_events)").unwrap();
-        stmt.query_map([], |r| r.get::<_, String>(1))
-            .unwrap()
-            .collect::<Result<Vec<_>, _>>()
-            .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<String> = {
-        let mut stmt = conn.prepare("PRAGMA table_info(outbox_events)").unwrap();
-        stmt.query_map([], |r| r.get::<_, String>(1))
-            .unwrap()
-            .collect::<Result<Vec<_>, _>>()
-            .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<String> = {
-        let mut stmt = conn
-            .prepare("PRAGMA table_info(file_commit_phases)")
-            .unwrap();
-        stmt.query_map([], |r| r.get::<_, String>(1))
-            .unwrap()
-            .collect::<Result<Vec<_>, _>>()
-            .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<String> = {
-        let mut stmt = conn.prepare("PRAGMA table_info(outbox_events)").unwrap();
-        stmt.query_map([], |r| r.get::<_, String>(1))
-            .unwrap()
-            .collect::<Result<Vec<_>, _>>()
-            .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"
+    );
+}