fix(store-journal): run_recovery_gate enforces pre-migration ordering

GAP 1: Add store_recovery_failed guard at top of mutate_store — returns
Err immediately if the boot recovery flag is set, blocking all journaled
mutations after a recovery failure.

GAP 2: recover_interrupted_file_commits now returns usize (unresolved
commit count) instead of (). Every skip branch increments the counter;
lib.rs sets store_recovery_failed and early-returns when count > 0 OR Err
— unresolved commits are no longer log-and-proceed.

GAP 3: Extract run_recovery_gate(anchor, store_work_closure) — runs
file-commit recovery, fails closed on Err or unresolved count > 0, and
only then invokes the closure. lib.rs calls it with the migrations entry
inside the closure; tests drive the same function. The ordering invariant
lives in one place and cannot drift.

Tests: test_setup_order_recovery_completes_before_migration_reads_canonical
drives run_recovery_gate and asserts the closure sees the repaired
canonical; test_recovery_gate_failure_skips_closure_and_returns_err asserts
the closure never runs on unresolved recovery; GAP 2 tests assert
hash-mismatch and rename-failure scenarios return unresolved count > 0.

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,6 +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::{
     backfill_persona_snapshots, ensure_nest, list_managed_agent_runtimes,
     put_managed_agent_runtime_lifecycle, reconcile_managed_agent_runtimes,
@@ -340,31 +341,32 @@ pub fn run() {
                 return Ok(());
             }
 
-            // File-commit recovery MUST run before any migration, backfill, or
-            // canonical-store reader/writer.  A stale stage from a pre-crash
-            // commit must not overwrite migration output.  One owner, one site.
-            // Fail closed: a repair error sets store_recovery_failed and skips
-            // all store-touching setup.
+            // run_recovery_gate: file-commit recovery before migrations; fail closed.
             {
                 let state = app_handle.state::<AppState>();
-                if let Err(e) =
-                    crate::managed_agents::store_journal::run_file_commit_recovery(&app_handle)
-                {
+                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}");
-                    state
-                        .store_recovery_failed
-                        .store(true, std::sync::atomic::Ordering::Release);
+                    state.store_recovery_failed.store(true, Ordering::Release);
                     return Ok(());
                 }
             }
 
-            // Run all pre-identity data migrations before state loads from disk.
-            if reset_outcome.completed {
-                migration::run_boot_migrations_after_reset(&app_handle);
-            } else {
-                migration::run_boot_migrations(&app_handle);
-            }
-
             // Resolve persisted identity key (env var → file → generate+save).
             // This is fatal — the app should not start with an ephemeral identity
             // that will be lost on restart, as that silently breaks channel

desktop/src-tauri/src/managed_agents/store_journal/mod.rs

@@ -62,8 +62,8 @@ pub use schema::apply_journal_schema_pub;
 pub use schema::open_journal;
 
 pub use txn::{
-    advance_to_committed, mutate_store, prepare_publication, run_boot_recovery,
-    run_file_commit_recovery, StoreState,
+    advance_to_committed, mutate_store, prepare_publication, run_boot_recovery, run_recovery_gate,
+    StoreState,
 };
 #[cfg(test)]
 #[allow(unused_imports)]

desktop/src-tauri/src/managed_agents/store_journal/txn.rs

@@ -26,6 +26,7 @@ use std::sync::MutexGuard;
 
 use rusqlite::params;
 use tauri::AppHandle;
+use tauri::Manager;
 
 use crate::managed_agents::{ManagedAgentRecord, TeamRecord};
 
@@ -96,6 +97,29 @@ fn advance_file_commit_phase(
     Ok(())
 }
 
+/// Run file-commit recovery and, only if it succeeds with zero unresolved
+/// commits, execute `store_work`.
+///
+/// This is the structural seam that enforces "recovery before canonical-store
+/// access".  Both production setup and tests drive the same function, so the
+/// ordering invariant lives in one place and cannot drift.
+///
+/// Returns `Err` if recovery fails (`Err` or unresolved count > 0) or if
+/// `store_work` returns `Err`.  On recovery failure `store_work` is never
+/// called.
+pub fn run_recovery_gate<F>(anchor: &std::path::Path, store_work: F) -> Result<(), String>
+where
+    F: FnOnce() -> Result<(), String>,
+{
+    let unresolved = file_commit_recovery_at(anchor)?;
+    if unresolved > 0 {
+        return Err(format!(
+            "file-commit recovery: {unresolved} nonterminal commit(s) could not be resolved"
+        ));
+    }
+    store_work()
+}
+
 /// Mutate the store under the full lock sequence.
 ///
 /// The closure runs inside a real `rusqlite::Transaction` — if it returns
@@ -104,6 +128,10 @@ fn advance_file_commit_phase(
 ///
 /// On any decode error inside this function no file is written, no journal
 /// transition occurs, and the error is propagated with `?`.
+///
+/// Returns `Err` immediately if the `store_recovery_failed` flag is set on
+/// `AppState` — a recovery failure at boot means the store is in an uncertain
+/// state and no mutations are safe until the user relaunches.
 pub fn mutate_store<'g, F, T>(
     app: &AppHandle,
     store_mutex_guard: MutexGuard<'g, ()>,
@@ -112,6 +140,20 @@ pub fn mutate_store<'g, F, T>(
 where
     F: FnOnce(StoreState<'_>) -> Result<(Vec<ManagedAgentRecord>, Vec<TeamRecord>, T), String>,
 {
+    // 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 anchor = store_anchor_dir(app)?;
     std::fs::create_dir_all(&anchor).map_err(|e| format!("create anchor dir: {e}"))?;
 
@@ -415,9 +457,9 @@ pub fn run_boot_recovery(
 /// Path-level boot recovery — **publication-only** (no file repair).
 ///
 /// File-commit recovery (phase 1) is handled exclusively by the synchronous
-/// `run_file_commit_recovery` call in app setup, which runs BEFORE this
-/// background task is spawned.  This function runs only phase 2: publication
-/// recovery (outbox re-drive into the retention DB).
+/// `run_recovery_gate` call in app setup, which runs BEFORE this background
+/// task is spawned.  This function runs only phase 2: publication recovery
+/// (outbox re-drive into the retention DB).
 ///
 /// `retention_db_path` is the path to the anchor's active scoped retention DB.
 /// When `Some`, recovery re-inserts missing `pending_sync` retention rows from
@@ -428,30 +470,19 @@ pub(crate) fn run_boot_recovery_at(
     retention_db_path: Option<&std::path::Path>,
 ) -> Result<(), String> {
     // Publication recovery only — no file I/O, no lock required.
-    // File-commit repair is the sole responsibility of run_file_commit_recovery.
+    // File-commit repair is the sole responsibility of run_recovery_gate.
     let journal = open_journal(anchor)?;
     recover_nonterminal_operations(&journal, retention_db_path)?;
     Ok(())
 }
 
-/// File-commit-only recovery: acquire the advisory lock, replay any interrupted
-/// two-phase file commits, then release the lock.
-///
-/// Called synchronously during app setup (before commands are admitted) so the
-/// store is always in a consistent state by the time the first command runs.
-pub fn run_file_commit_recovery(app: &AppHandle) -> Result<(), String> {
-    let anchor = store_anchor_dir(app)?;
-    std::fs::create_dir_all(&anchor).map_err(|e| format!("boot-recovery create anchor: {e}"))?;
-    file_commit_recovery_at(&anchor)
-}
-
 /// Path-level file-commit recovery, extracted for testing without an AppHandle.
 #[cfg(test)]
-pub(crate) fn file_commit_recovery_at_pub(anchor: &std::path::Path) -> Result<(), String> {
+pub(crate) fn file_commit_recovery_at_pub(anchor: &std::path::Path) -> Result<usize, String> {
     file_commit_recovery_at(anchor)
 }
 
-fn file_commit_recovery_at(anchor: &std::path::Path) -> Result<(), String> {
+fn file_commit_recovery_at(anchor: &std::path::Path) -> Result<usize, String> {
     let _advisory = super::lock::JournalLockGuard::acquire(anchor)?;
     let journal = open_journal(anchor)?;
     recover_interrupted_file_commits(&journal)
@@ -497,8 +528,14 @@ fn sha256_hex_of_file(path: &std::path::Path) -> Option<String> {
     Some(hex::encode(sha2::Sha256::digest(&bytes)))
 }
 
+/// Recover any interrupted two-phase file commits.
+///
+/// Returns the number of nonterminal commits that could NOT be resolved (any
+/// branch that previously `continue`d without completing the rename sequence).
+/// A non-zero count means the store is in an uncertain state and callers must
+/// treat it as a recovery failure.
 #[allow(clippy::type_complexity)]
-fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<(), String> {
+fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<usize, String> {
     // Read commit_id, phase, stage paths, AND content hashes.
     let rows: Vec<(String, String, String, String, String, String)> = {
         let mut stmt = journal
@@ -528,6 +565,8 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<()
             .map_err(|e| format!("read file_commit_phases row: {e}"))?
     };
 
+    let mut unresolved: usize = 0;
+
     for (commit_id, phase_str, agents_stage_str, teams_stage_str, agents_hash, teams_hash) in rows {
         let phase = FileCommitPhase::from_str(&phase_str).unwrap_or(FileCommitPhase::Intent);
         let agents_stage = std::path::PathBuf::from(&agents_stage_str);
@@ -540,6 +579,15 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<()
         let agents_canonical = canonical_from_stage(&agents_stage);
         let teams_canonical = canonical_from_stage(&teams_stage);
 
+        // Macro that counts a skipped commit as unresolved and moves to the next.
+        macro_rules! skip_unresolved {
+            ($msg:expr) => {{
+                eprintln!("{}", $msg);
+                unresolved += 1;
+                continue;
+            }};
+        }
+
         match phase {
             FileCommitPhase::Intent => {
                 // Both stage files should exist for a full intent commit.
@@ -554,7 +602,8 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<()
                         && sha256_hex_of_file(&teams_canonical).as_deref() == Some(&teams_hash);
                     if agents_ok && teams_ok {
                         eprintln!(
-                            "buzz-desktop: boot-recovery: {commit_id}: both stage files absent,                              canonicals verified — advancing to committed"
+                            "buzz-desktop: boot-recovery: {commit_id}: both stage files absent, \
+                             canonicals verified — advancing to committed"
                         );
                         advance_file_commit_phase(journal, &commit_id, &FileCommitPhase::Committed)
                             .unwrap_or_else(|e| {
@@ -563,25 +612,28 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<()
                                 )
                             });
                     } else {
-                        eprintln!(
-                            "buzz-desktop: boot-recovery: {commit_id}: both stage files absent                              in intent phase and canonicals do not match recorded hashes —                              failing closed; manual recovery may be needed"
-                        );
+                        skip_unresolved!(format!(
+                            "buzz-desktop: boot-recovery: {commit_id}: both stage files absent \
+                             in intent phase and canonicals do not match recorded hashes — \
+                             failing closed; manual recovery may be needed"
+                        ));
                     }
                     continue;
                 }
                 if !agents_exists {
                     // Agents stage absent, teams stage present: ambiguous —
                     // cannot determine if agents rename completed without proof.
-                    // Fail closed: skip this commit; operator intervention needed.
-                    eprintln!(
-                        "buzz-desktop: boot-recovery: {commit_id}: agents stage absent in                          intent phase (teams stage present) — cannot safely complete;                          skipping (manual recovery may be needed)"
-                    );
-                    continue;
+                    skip_unresolved!(format!(
+                        "buzz-desktop: boot-recovery: {commit_id}: agents stage absent in \
+                         intent phase (teams stage present) — cannot safely complete; \
+                         skipping (manual recovery may be needed)"
+                    ));
                 }
                 // Agents stage exists — replay agents rename.
                 if let Err(e) = rename_staged(&agents_stage, &agents_canonical) {
-                    eprintln!("buzz-desktop: boot-recovery: agents rename failed: {e}");
-                    continue;
+                    skip_unresolved!(format!(
+                        "buzz-desktop: boot-recovery: agents rename failed: {e}"
+                    ));
                 }
                 if let Err(e) =
                     advance_file_commit_phase(journal, &commit_id, &FileCommitPhase::FirstRenamed)
@@ -591,20 +643,20 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<()
                 // Teams rename.
                 if teams_exists {
                     if let Err(e) = rename_staged(&teams_stage, &teams_canonical) {
-                        eprintln!("buzz-desktop: boot-recovery: teams rename failed: {e}");
-                        continue;
+                        skip_unresolved!(format!(
+                            "buzz-desktop: boot-recovery: teams rename failed: {e}"
+                        ));
                     }
                 } else {
                     // Teams stage absent after agents rename. Verify via hash.
                     let teams_ok = !teams_hash.is_empty()
                         && sha256_hex_of_file(&teams_canonical).as_deref() == Some(&teams_hash);
                     if !teams_ok {
-                        eprintln!(
+                        skip_unresolved!(format!(
                             "buzz-desktop: boot-recovery: {commit_id}: teams stage absent \
                              in intent phase (after agents rename) and canonical does not \
                              match recorded hash — failing closed"
-                        );
-                        continue;
+                        ));
                     }
                 }
                 advance_file_commit_phase(journal, &commit_id, &FileCommitPhase::Committed)
@@ -616,10 +668,9 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<()
                 // Agents already renamed; only teams stage remains.
                 if teams_stage.exists() {
                     if let Err(e) = rename_staged(&teams_stage, &teams_canonical) {
-                        eprintln!(
+                        skip_unresolved!(format!(
                             "buzz-desktop: boot-recovery: teams rename (first_renamed) failed: {e}"
-                        );
-                        continue;
+                        ));
                     }
                     // Teams rename succeeded — advance to committed.
                 } else {
@@ -627,10 +678,11 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<()
                     let teams_ok = !teams_hash.is_empty()
                         && sha256_hex_of_file(&teams_canonical).as_deref() == Some(&teams_hash);
                     if !teams_ok {
-                        eprintln!(
-                            "buzz-desktop: boot-recovery: {commit_id}: teams stage absent in                              first_renamed phase and canonical does not match recorded hash —                              failing closed; manual recovery may be needed"
-                        );
-                        continue;
+                        skip_unresolved!(format!(
+                            "buzz-desktop: boot-recovery: {commit_id}: teams stage absent in \
+                             first_renamed phase and canonical does not match recorded hash — \
+                             failing closed; manual recovery may be needed"
+                        ));
                     }
                     // Canonical matches — teams rename already completed.
                 }
@@ -643,7 +695,7 @@ fn recover_interrupted_file_commits(journal: &rusqlite::Connection) -> Result<()
         }
     }
 
-    Ok(())
+    Ok(unresolved)
 }
 
 /// Recover nonterminal operations from the journal outbox.

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, Generation,
+    run_boot_recovery_at, run_recovery_gate, Generation,
 };
 use crate::managed_agents::retention::{
     get_pending_sync, open_retention_db, tombstone_retention_d_tag,
@@ -674,44 +674,24 @@ fn test_sync_file_recovery_repairs_dangling_intent_row() {
     );
 }
 
-/// Fix A — production-order seam test: file-commit recovery must complete
-/// BEFORE any migration or canonical-store reader observes the canonical files.
-///
-/// Seam used: `file_commit_recovery_at_pub` is the exact function that
-/// `run_file_commit_recovery` delegates to in production — not a
-/// reimplementation.  The "migration read" is a direct filesystem read of the
-/// canonical path, matching what every real migration/backfill does.
-///
-/// Scenario: a process crashed after writing a stage file but before the rename
-/// completed.  The canonical (`managed-agents.json`) still holds pre-crash
-/// ("stale") content.  The stage file holds the intended post-crash ("repaired")
-/// content.  After recovery runs, the canonical must hold the repaired content —
-/// so any subsequent migration or backfill reader sees the repaired state.
+/// Fix A — ordered seam: `run_recovery_gate` runs recovery before invoking the
+/// closure; the closure (= migrations) observes the repaired canonical.
+/// Both production and this test drive the same function.
 #[test]
 fn test_setup_order_recovery_completes_before_migration_reads_canonical() {
     let dir = tmp_dir();
     let anchor = dir.path().to_path_buf();
-
-    // "Stale" content — what canonical holds before recovery.
     let stale_content = b"[{\"pubkey\":\"stale-pre-crash\"}]";
-    // "Repaired" content — what the stage file holds (the intended commit).
     let repaired_content = b"[{\"pubkey\":\"repaired-post-crash\"}]";
     let teams_content = b"[]";
-
     let a_can = anchor.join("managed-agents.json");
     let t_can = anchor.join("teams.json");
     let a_stage = anchor.join("managed-agents.order_test.stage");
     let t_stage = anchor.join("teams.order_test.stage");
-
-    // Write stale canonical — this is what migrations would see WITHOUT recovery.
     std::fs::write(&a_can, stale_content).unwrap();
     std::fs::write(&t_can, teams_content).unwrap();
-
-    // Write stage files containing the repaired (intended) state.
     std::fs::write(&a_stage, repaired_content).unwrap();
     std::fs::write(&t_stage, teams_content).unwrap();
-
-    // Plant the intent row — recovery will detect and replay it.
     {
         let j = open_journal(&anchor).unwrap();
         insert_phase_row(
@@ -724,25 +704,124 @@ fn test_setup_order_recovery_completes_before_migration_reads_canonical() {
             &sha256_hex(teams_content),
         );
     }
-
-    // Run recovery — the exact same code path production calls via
-    // run_file_commit_recovery → file_commit_recovery_at.
-    file_commit_recovery_at_pub(&anchor).unwrap();
-
-    // Now simulate a "migration read" — read the canonical directly, as every
-    // real migration/backfill does.
-    let observed = std::fs::read(&a_can).unwrap();
+    let mut closure_saw: Option<Vec<u8>> = None;
+    let saw = &mut closure_saw;
+    run_recovery_gate(&anchor, || {
+        *saw = Some(std::fs::read(&a_can).unwrap());
+        Ok(())
+    })
+    .expect("run_recovery_gate must succeed");
+    let observed = closure_saw.expect("closure must have run");
     assert_eq!(
         observed, repaired_content,
-        "migration read must observe the repaired canonical, not the stale pre-crash state"
+        "closure must see repaired canonical"
     );
-    // Stale content must not be present.
     assert_ne!(
         observed, stale_content,
-        "stale canonical content must have been replaced by recovery"
+        "stale canonical must be replaced first"
     );
 }
 
+/// Fix A + GAP 3 — fail-closed: unresolved recovery → closure never runs,
+/// `run_recovery_gate` returns `Err`.
+#[test]
+fn test_recovery_gate_failure_skips_closure_and_returns_err() {
+    let dir = tmp_dir();
+    let anchor = dir.path().to_path_buf();
+    let a_stage = anchor.join("managed-agents.gate_fail.stage");
+    let t_stage = anchor.join("teams.gate_fail.stage");
+    // No stage files + empty hashes → hash-mismatch → 1 unresolved.
+    {
+        let j = open_journal(&anchor).unwrap();
+        insert_phase_row(
+            &j,
+            "gate_fail",
+            "intent",
+            a_stage.to_str().unwrap(),
+            t_stage.to_str().unwrap(),
+            "",
+            "",
+        );
+    }
+    let mut closure_ran = false;
+    let result = run_recovery_gate(&anchor, || {
+        closure_ran = true;
+        Ok(())
+    });
+    assert!(
+        result.is_err(),
+        "run_recovery_gate must return Err on unresolved commits"
+    );
+    assert!(
+        !closure_ran,
+        "closure must not run when recovery has unresolved commits"
+    );
+}
+
+/// GAP 2 — hash-mismatch: absent stages + non-matching hashes → unresolved count > 0.
+#[test]
+fn test_recovery_hash_mismatch_returns_unresolved_count() {
+    let dir = tmp_dir();
+    let anchor = dir.path().to_path_buf();
+    let a_stage = anchor.join("managed-agents.hmm.stage");
+    let t_stage = anchor.join("teams.hmm.stage");
+    std::fs::write(
+        anchor.join("managed-agents.json"),
+        b"[{\"pubkey\":\"actual\"}]",
+    )
+    .unwrap();
+    std::fs::write(anchor.join("teams.json"), b"[]").unwrap();
+    {
+        let j = open_journal(&anchor).unwrap();
+        insert_phase_row(
+            &j,
+            "hmm",
+            "intent",
+            a_stage.to_str().unwrap(),
+            t_stage.to_str().unwrap(),
+            "deadbeefdeadbeef",
+            "deadbeefdeadbeef",
+        );
+    }
+    let unresolved = file_commit_recovery_at_pub(&anchor).expect("must not Err");
+    assert_eq!(
+        unresolved, 1,
+        "hash-mismatch with absent stages → 1 unresolved"
+    );
+}
+
+/// GAP 2 — rename-failure: stage exists but canonical target is a directory
+/// (rename-over-directory is rejected by the OS) → unresolved count > 0.
+#[test]
+fn test_recovery_rename_failure_returns_unresolved_count() {
+    let dir = tmp_dir();
+    let anchor = dir.path().to_path_buf();
+    let repaired = b"[{\"pubkey\":\"rename-fail\"}]";
+    let teams = b"[]";
+    let a_stage = anchor.join("managed-agents.rfail.stage");
+    let t_stage = anchor.join("teams.rfail.stage");
+    let a_can = anchor.join("managed-agents.json");
+    std::fs::write(&a_stage, repaired).unwrap();
+    std::fs::write(&t_stage, teams).unwrap();
+    // canonical is a directory → rename(file, dir) → EISDIR
+    std::fs::create_dir_all(&a_can).unwrap();
+    std::fs::write(anchor.join("teams.json"), teams).unwrap();
+    {
+        let j = open_journal(&anchor).unwrap();
+        insert_phase_row(
+            &j,
+            "rfail",
+            "intent",
+            a_stage.to_str().unwrap(),
+            t_stage.to_str().unwrap(),
+            &sha256_hex(repaired),
+            &sha256_hex(teams),
+        );
+    }
+    let unresolved = file_commit_recovery_at_pub(&anchor).expect("must not Err");
+    assert_eq!(unresolved, 1, "rename failure → 1 unresolved");
+}
+
 // ── Fix B: migration tests ───────────────────────────────────────────────────
 
 /// Helper: create a v2-shaped in-memory journal (no hash/d-tag columns).