Migration safety: rebuild FTS after incompatible-index wipe, fail fast on embedder feature gap (#210)

* fix(history): rebuild FTS from DB when on-disk index was wiped or empty

Opening a store over a foreign/incompatible tantivy index (e.g. the retired
Python impl's index with a different schema) makes open_or_recreate wipe it to
a fresh empty index. rebuild_fts had no call sites, so all indexed history was
silently lost and never repopulated.

open_dir now surfaces whether it took the recreate path; HistoryStore::open
repopulates any index that was recreated or is empty while its source table
still holds rows, from the DB, synchronously (small corpus), logging counts.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

* fix(run): fail fast on embedder feature-gap before opening the store

create_embedder's error (e.g. fastembed without the local-embed extra) was
swallowed by .unwrap_or(None) at the async_main call site, so startup proceeded
to open and wipe the FTS and only died later in create_projectors under a
misleading Discord-token hint.

Extract a testable resolve_embedder seam and call it in run_inner BEFORE
Familiar::load_from_disk opens the store: on error, log the real message
(which names the fix) and exit 1. The embedder is threaded into async_main.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

* fix(activities): de-flake c1 wake test — raise recv_wake hang-guard to 30s

recv_wake's 1s timeout is only a hang-guard (every caller .expects Some; none
asserts None), but the runner does several spawn_blocking DB round-trips before
publishing the wake, which can outlast 1s on a loaded current-thread test
runtime — turning the guard into a false failure (~67% under CPU load). Raise
the bound to 30s; a genuinely-stuck test still fails, the race is gone.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

* style: rustfmt import ordering in commands/run.rs

Co-Authored-By: Claude Opus 4.8 <[email protected]>

---------

Co-authored-by: Claude Opus 4.8 <[email protected]>

Author: CorVous

familiar-connect/src/activities/engine.rs

@@ -2697,7 +2697,12 @@ mod tests {
     async fn recv_wake(
         sub: &mut crate::bus::in_process::Subscription,
     ) -> Option<std::sync::Arc<Event>> {
-        timeout(StdDur::from_secs(1), sub.recv())
+        // Pure hang-guard: every caller `.expect`s `Some`, so the bound only
+        // exists to fail a genuinely-stuck test instead of hanging CI forever.
+        // Keep it generous — the runner does several `spawn_blocking` DB
+        // round-trips before publishing, which can outlast a tight 1s deadline
+        // on a loaded current-thread test runtime (the `c1_*` flake).
+        timeout(StdDur::from_secs(30), sub.recv())
             .await
             .ok()
             .flatten()

familiar-connect/src/commands/run.rs

@@ -28,12 +28,13 @@ use crate::activities::engine::{
 };
 use crate::bot::{BotHandle, build_activity_presence_cb};
 use crate::budget::TierBudget;
+use crate::config::EmbeddingConfig;
 use crate::context::layers::ChannelResolver;
 use crate::context::{
     Assembler, CharacterCardLayer, ConversationSummaryLayer, LorebookLayer, OperatingModeLayer,
     PeopleDossierLayer, RagContextLayer, RecentHistoryLayer, ReflectionLayer,
 };
-use crate::embedding::Embedder;
+use crate::embedding::{Embedder, EmbeddingError};
 use crate::familiar::Familiar;
 use crate::focus::FocusManager;
 use crate::sleep::maintenance::SleepPromptText;
@@ -97,6 +98,30 @@ pub fn resolve_familiar_root(
     Ok(dir)
 }
 
+/// Resolve the startup embedder from `config`, or fail.
+///
+/// The composition root calls this BEFORE opening the history store / FTS: a
+/// misconfigured embedding backend (e.g. `fastembed` without the `local-embed`
+/// extra) must refuse to start rather than wipe the FTS and then die deep in
+/// `create_projectors` under a misleading Discord-token hint. The backend error
+/// (which already names the real fix) is propagated, never swallowed into
+/// `None`.
+///
+/// # Errors
+/// Whatever [`create_embedder`](crate::embedding::create_embedder) returns —
+/// unknown backend, bad dimensionality, or the `fastembed` `local-embed` gap.
+#[cfg_attr(
+    not(feature = "discord"),
+    allow(
+        dead_code,
+        reason = "the only non-test caller is the `discord`-gated `run_inner`; \
+                  the fail-fast contract is unit-tested under default features"
+    )
+)]
+fn resolve_embedder(config: &EmbeddingConfig) -> Result<Option<Arc<dyn Embedder>>, EmbeddingError> {
+    crate::embedding::create_embedder(config)
+}
+
 /// The two hardcoded operating-mode strings (byte-exact; Python
 /// `_default_assembler`).
 fn operating_modes() -> HashMap<String, String> {
@@ -444,6 +469,19 @@ fn run_inner(token: &str, familiar_root: &Path) -> i32 {
         None
     };
 
+    // Resolve the embedder BEFORE loading the familiar (which opens + possibly
+    // rebuilds the history store / FTS). A misconfigured backend must refuse to
+    // start here, while nothing has been mutated — not fail deep in
+    // `create_projectors` after the store is already open. Its error names the
+    // real fix (e.g. the `local-embed` extra), so surface it verbatim.
+    let embedder = match resolve_embedder(&config.embedding) {
+        Ok(embedder) => embedder,
+        Err(err) => {
+            tracing::error!("Embedding backend unavailable: {err}");
+            return 1;
+        }
+    };
+
     let familiar = match Familiar::load_from_disk(
         familiar_root,
         llm_clients,
@@ -474,7 +512,7 @@ fn run_inner(token: &str, familiar_root: &Path) -> i32 {
             return 1;
         }
     };
-    match runtime.block_on(async_main(token.to_owned(), familiar)) {
+    match runtime.block_on(async_main(token.to_owned(), familiar, embedder)) {
         Ok(()) => 0,
         Err(err) => {
             tracing::error!(
@@ -601,7 +639,11 @@ fn spawn_signal_listener(controller: Arc<ShutdownController>) {
     clippy::too_many_lines,
     reason = "the composition root wires every subsystem; splitting obscures the ordering contract"
 )]
-async fn async_main(token: String, mut familiar: Familiar) -> anyhow::Result<()> {
+async fn async_main(
+    token: String,
+    mut familiar: Familiar,
+    embedder: Option<Arc<dyn Embedder>>,
+) -> anyhow::Result<()> {
     use std::sync::Mutex;
 
     use crate::bot::{ActivityResync, AsyncBotStore, BotStore, CreateBotDeps, create_bot};
@@ -689,8 +731,8 @@ async fn async_main(token: String, mut familiar: Familiar) -> anyhow::Result<()>
     })
     .await?;
 
-    let embedder = crate::embedding::create_embedder(&familiar.config.embedding).unwrap_or(None);
-
+    // Embedder resolved at the composition root (before the store opened); a bad
+    // backend would already have aborted startup in `run_inner`.
     let voice_assembler = Arc::new(default_assembler(
         &familiar,
         familiar.config.voice_window_size,
@@ -1083,11 +1125,12 @@ async fn async_main(token: String, mut familiar: Familiar) -> anyhow::Result<()>
 mod tests {
     use super::{
         ShutdownController, ShutdownStage, build_activity_engine, default_assembler,
-        resolve_familiar_root,
+        resolve_embedder, resolve_familiar_root,
     };
     use crate::activities::engine::ActivityEngine;
     use crate::bot::{BotHandle, Presence, PresenceSink};
     use crate::budget::TierBudget;
+    use crate::config::EmbeddingConfig;
     use crate::familiar::Familiar;
     use crate::focus::FocusManager;
     use crate::processors::SendText;
@@ -1098,6 +1141,38 @@ mod tests {
     use std::sync::{Arc, Mutex};
     use tempfile::TempDir;
 
+    // --- resolve_embedder (composition-root fail-fast) ---
+
+    /// The composition root must surface the embedder feature-gap as an error
+    /// (which names the `local-embed` fix), not swallow it into `None` and let
+    /// startup proceed to mutate the store. The default test build lacks the
+    /// `local-embed` extra, so `fastembed` genuinely has no backend here.
+    #[test]
+    fn resolve_embedder_fastembed_without_extra_errors() {
+        let config = EmbeddingConfig {
+            backend: "fastembed".to_owned(),
+            ..EmbeddingConfig::default()
+        };
+        let err = resolve_embedder(&config)
+            .err()
+            .expect("fastembed without the local-embed extra must fail fast");
+        assert!(
+            err.to_string().contains("local-embed"),
+            "error must name the real fix, got: {err}"
+        );
+    }
+
+    /// A disabled backend resolves to `None` without erroring — fail-fast must
+    /// not become fail-always.
+    #[test]
+    fn resolve_embedder_off_backend_is_none() {
+        let config = EmbeddingConfig {
+            backend: "off".to_owned(),
+            ..EmbeddingConfig::default()
+        };
+        assert!(resolve_embedder(&config).unwrap().is_none());
+    }
+
     // --- resolve_familiar_root (ported from test_run_cmd.py) ---
 
     #[test]

familiar-connect/src/history/fts.rs

@@ -182,25 +182,31 @@ impl TantivyFts {
 
     /// On-disk index rooted at `dir` (created if absent). If an existing index
     /// there is unreadable/version-incompatible, the directory is wiped and a
-    /// fresh empty index is created — callers repopulate via
+    /// fresh empty index is created. The returned flag is `true` in exactly that
+    /// wipe-and-recreate case, so callers can repopulate from the source-of-truth
+    /// tables via
     /// [`HistoryStore::rebuild_fts`](super::store::HistoryStore::rebuild_fts).
-    pub fn open_dir(dir: &Path) -> Result<Self, StoreError> {
+    pub fn open_dir(dir: &Path) -> Result<(Self, bool), StoreError> {
         std::fs::create_dir_all(dir).map_err(|e| StoreError::Fts(e.to_string()))?;
-        let index = Self::open_or_recreate(dir)?;
-        Self::finish(index)
+        let (index, recreated) = Self::open_or_recreate(dir)?;
+        Ok((Self::finish(index)?, recreated))
     }
 
-    fn open_or_recreate(dir: &Path) -> Result<Index, StoreError> {
+    /// Open the on-disk index, or wipe-and-recreate it if incompatible. The bool
+    /// reports whether the recreate path was taken.
+    fn open_or_recreate(dir: &Path) -> Result<(Index, bool), StoreError> {
         let schema = build_schema();
         let mmap = MmapDirectory::open(dir).map_err(|e| StoreError::Fts(e.to_string()))?;
         if let Ok(index) = Index::open_or_create(mmap, schema.clone()) {
-            return Ok(index);
+            return Ok((index, false));
         }
         // Incompatible/corrupt on-disk index — wipe and start fresh.
         std::fs::remove_dir_all(dir).map_err(|e| StoreError::Fts(e.to_string()))?;
         std::fs::create_dir_all(dir).map_err(|e| StoreError::Fts(e.to_string()))?;
         let mmap = MmapDirectory::open(dir).map_err(|e| StoreError::Fts(e.to_string()))?;
-        Index::open_or_create(mmap, schema).map_err(|e| StoreError::Fts(e.to_string()))
+        let index =
+            Index::open_or_create(mmap, schema).map_err(|e| StoreError::Fts(e.to_string()))?;
+        Ok((index, true))
     }
 
     fn finish(index: Index) -> Result<Self, StoreError> {
@@ -233,6 +239,14 @@ impl TantivyFts {
         })
     }
 
+    /// Whether the index currently holds no documents. Used after `open_dir` to
+    /// decide whether a freshly-created (but not recreate-flagged) index needs
+    /// repopulating from the source tables.
+    #[must_use]
+    pub fn is_empty(&self) -> bool {
+        self.reader.searcher().num_docs() == 0
+    }
+
     /// Test seam: install (or clear with `None`) a fake commit that fails before
     /// the real commit runs. Mirrors monkeypatching `FtsIndex._commit_writer`.
     pub fn set_commit_fault(&self, fault: Option<CommitFault>) {
@@ -414,6 +428,41 @@ mod tests {
         assert!(hits[0].1 > 0.0, "BM25 score must be positive");
     }
 
+    #[test]
+    fn open_dir_flags_recreate_on_incompatible_schema() {
+        use tantivy::Index;
+        use tantivy::schema::{STORED, STRING, Schema};
+
+        let dir = tempfile::tempdir().unwrap();
+        // Write an index whose schema differs from `build_schema` (row_id as text
+        // rather than i64) — stands in for the Python-vs-Rust incompatibility.
+        let mut sb = Schema::builder();
+        sb.add_text_field("row_id", STRING | STORED);
+        sb.add_text_field("content", STRING | STORED);
+        Index::create_in_dir(dir.path(), sb.build()).unwrap();
+
+        let (idx, recreated) = TantivyFts::open_dir(dir.path()).unwrap();
+        assert!(
+            recreated,
+            "incompatible on-disk schema must trigger recreate"
+        );
+        assert!(idx.is_empty(), "a recreated index starts empty");
+    }
+
+    #[test]
+    fn open_dir_preserves_compatible_index() {
+        let dir = tempfile::tempdir().unwrap();
+        {
+            let (idx, recreated) = TantivyFts::open_dir(dir.path()).unwrap();
+            assert!(!recreated);
+            idx.add(1, "the quick brown fox").unwrap();
+        }
+        let (idx, recreated) = TantivyFts::open_dir(dir.path()).unwrap();
+        assert!(!recreated, "a compatible reopen must not wipe the index");
+        assert!(!idx.is_empty());
+        assert_eq!(idx.search("fox", 5).len(), 1);
+    }
+
     #[test]
     fn upsert_replaces_prior_doc() {
         let idx = TantivyFts::in_memory().unwrap();

familiar-connect/src/history/store.rs

@@ -1053,9 +1053,15 @@ impl HistoryStore {
             std::fs::create_dir_all(parent)?;
         }
         let fts_root = parent.unwrap_or_else(|| Path::new(".")).join("fts");
-        let fts_turns = Box::new(TantivyFts::open_dir(&fts_root.join("turns"))?);
-        let fts_facts = Box::new(TantivyFts::open_dir(&fts_root.join("facts"))?);
-        Self::init(Db::open(path)?, fts_turns, fts_facts)
+        let (fts_turns, turns_recreated) = TantivyFts::open_dir(&fts_root.join("turns"))?;
+        let (fts_facts, facts_recreated) = TantivyFts::open_dir(&fts_root.join("facts"))?;
+        // An index that was wiped-and-recreated, or is otherwise empty, has lost
+        // whatever the retired Python impl indexed; repopulate it from the DB.
+        let turns_stale = turns_recreated || fts_turns.is_empty();
+        let facts_stale = facts_recreated || fts_facts.is_empty();
+        let store = Self::init(Db::open(path)?, Box::new(fts_turns), Box::new(fts_facts))?;
+        store.repopulate_stale_fts(turns_stale, facts_stale)?;
+        Ok(store)
     }
 
     /// Open a store with caller-supplied FTS indexes. The DB is set up exactly
@@ -3285,13 +3291,70 @@ impl HistoryStore {
 
     /// Drop and repopulate the tantivy turns index from `turns`.
     pub fn rebuild_fts(&self) -> Result<(), StoreError> {
+        self.rebuild_turns_fts().map(|_| ())
+    }
+
+    /// Clear and repopulate the turns index from the `turns` table; returns the
+    /// number of rows indexed.
+    fn rebuild_turns_fts(&self) -> Result<usize, StoreError> {
         self.fts_turns.clear()?;
         let rows = self.db.query_map(
             "SELECT id, content FROM turns ORDER BY id ASC",
             vec![],
             |r| Ok((r.get::<_, i64>("id")?, r.get::<_, String>("content")?)),
         )?;
-        self.fts_turns.add_many(&rows)
+        let count = rows.len();
+        self.fts_turns.add_many(&rows)?;
+        Ok(count)
+    }
+
+    /// Clear and repopulate the facts index from the `facts` table; returns the
+    /// number of rows indexed.
+    fn rebuild_facts_fts(&self) -> Result<usize, StoreError> {
+        self.fts_facts.clear()?;
+        let rows =
+            self.db
+                .query_map("SELECT id, text FROM facts ORDER BY id ASC", vec![], |r| {
+                    Ok((r.get::<_, i64>("id")?, r.get::<_, String>("text")?))
+                })?;
+        let count = rows.len();
+        self.fts_facts.add_many(&rows)?;
+        Ok(count)
+    }
+
+    /// Repopulate any FTS index that was wiped/recreated or is empty while its
+    /// source table still holds rows (the Python-index migration failure mode).
+    /// Synchronous — the live corpus is ~10k rows.
+    fn repopulate_stale_fts(&self, turns_stale: bool, facts_stale: bool) -> Result<(), StoreError> {
+        if turns_stale && self.table_has_rows("turns")? {
+            let rows = self.rebuild_turns_fts()?;
+            tracing::info!(
+                target: "familiar_connect.history",
+                index = "turns",
+                rows,
+                "rebuilt FTS index from DB (on-disk index was wiped or empty)"
+            );
+        }
+        if facts_stale && self.table_has_rows("facts")? {
+            let rows = self.rebuild_facts_fts()?;
+            tracing::info!(
+                target: "familiar_connect.history",
+                index = "facts",
+                rows,
+                "rebuilt FTS index from DB (on-disk index was wiped or empty)"
+            );
+        }
+        Ok(())
+    }
+
+    /// Cheap existence probe for one of the fixed source tables.
+    fn table_has_rows(&self, table: &str) -> Result<bool, StoreError> {
+        // `table` is a fixed literal from `repopulate_stale_fts`, never user input.
+        let present = self
+            .db
+            .query_scalar_i64(format!("SELECT EXISTS(SELECT 1 FROM {table})"), vec![])?
+            .unwrap_or(0);
+        Ok(present != 0)
     }
 
     /// Highest turn id indexed for `familiar_id` (a `MAX(turns.id)` query; the