fix: flare-docs stale-item reconciliation + doc history-skip check

crates/agentflare-store/src/documents.rs

@@ -151,7 +151,9 @@ impl Store {
 
         let existing = tx
             .query_row(
-                "SELECT id, rowid, content, version, blob_hash, mime, metadata, size FROM store_documents
+                "SELECT id, rowid, content, version, blob_hash, mime, metadata, size,
+                        title, doc_type, tags, session_id, source
+                 FROM store_documents
                  WHERE project_id = ?1 AND path = ?2",
                 params![project_id, path],
                 |row| {
@@ -164,6 +166,11 @@ impl Store {
                         row.get::<_, String>(5)?,
                         row.get::<_, String>(6)?,
                         row.get::<_, i64>(7)?,
+                        row.get::<_, String>(8)?,
+                        row.get::<_, String>(9)?,
+                        row.get::<_, String>(10)?,
+                        row.get::<_, Option<String>>(11)?,
+                        row.get::<_, String>(12)?,
                     ))
                 },
             )
@@ -178,6 +185,11 @@ impl Store {
             old_mime,
             old_metadata,
             old_size,
+            old_title,
+            old_doc_type,
+            old_tags_json,
+            old_session_id,
+            old_source,
         )) = existing
         {
             let new_version = old_version + 1;
@@ -190,7 +202,27 @@ impl Store {
             // detect a change and history would never record; also check
             // `blob_hash` so a no-op re-upsert still skips (both unchanged)
             // while a genuine new blob still snapshots (content stays "").
-            let unchanged = old_content == content && old_blob_hash == opts.blob_hash;
+            //
+            // A caller can also re-upsert with unchanged content/blob_hash
+            // but a different title/doc_type/tags/mime/metadata/size --
+            // that's still a real change to the document's persisted state,
+            // so every `Some(...)`-provided opts field must match its
+            // stored value too, not just content, for this to count as a
+            // true no-op.
+            let old_tags: Vec<String> = serde_json::from_str(&old_tags_json).unwrap_or_default();
+            let unchanged = old_content == content
+                && old_blob_hash == opts.blob_hash
+                && opts.title.as_deref().is_none_or(|v| v == old_title)
+                && opts.doc_type.as_deref().is_none_or(|v| v == old_doc_type)
+                && opts.mime.as_deref().is_none_or(|v| v == old_mime)
+                && opts.metadata.as_deref().is_none_or(|v| v == old_metadata)
+                && opts.size.is_none_or(|v| v == old_size)
+                && opts
+                    .session_id
+                    .as_deref()
+                    .is_none_or(|v| old_session_id.as_deref() == Some(v))
+                && opts.source.as_deref().is_none_or(|v| v == old_source)
+                && opts.tags.as_ref().is_none_or(|v| *v == old_tags);
             if opts.track_history && !unchanged {
                 let history_id = db_kit::ids::new_id();
                 tx.execute(
@@ -839,6 +871,63 @@ mod tests {
         );
     }
 
+    #[test]
+    fn unchanged_content_with_changed_title_still_writes_history() {
+        let s = store();
+        let doc = s
+            .doc_upsert_with_opts(
+                "p",
+                "/title-change.md",
+                "same content",
+                DocUpsertOpts {
+                    title: Some("Old Title".into()),
+                    ..Default::default()
+                },
+            )
+            .unwrap();
+        let updated = s
+            .doc_upsert_with_opts(
+                "p",
+                "/title-change.md",
+                "same content",
+                DocUpsertOpts {
+                    title: Some("New Title".into()),
+                    ..Default::default()
+                },
+            )
+            .unwrap();
+        assert_eq!(updated.id, doc.id);
+        assert_eq!(updated.title, "New Title");
+        let history = s.doc_history(&updated.id).unwrap();
+        assert_eq!(
+            history.len(),
+            1,
+            "a title-only change must still write a history row even though content didn't change"
+        );
+        assert_eq!(history[0].title, "Old Title");
+    }
+
+    #[test]
+    fn identical_content_and_opts_reupsert_skips_history_row() {
+        let s = store();
+        let opts = || DocUpsertOpts {
+            title: Some("Same Title".into()),
+            tags: Some(vec!["rust".into()]),
+            ..Default::default()
+        };
+        let doc = s
+            .doc_upsert_with_opts("p", "/no-op.md", "same content", opts())
+            .unwrap();
+        let updated = s
+            .doc_upsert_with_opts("p", "/no-op.md", "same content", opts())
+            .unwrap();
+        assert_eq!(updated.id, doc.id);
+        assert!(
+            s.doc_history(&updated.id).unwrap().is_empty(),
+            "a true no-op re-upsert (same content and same opts) must not write history"
+        );
+    }
+
     #[test]
     fn project_path_unique_index_rejects_duplicate_raw_insert() {
         let s = store();

crates/flare-docs/src/rustdoc.rs

@@ -125,9 +125,13 @@ pub fn store_fetched(
 /// Parses the full rustdoc-JSON `index`/`paths` maps and upserts one
 /// document per item that has both a docstring and a public path (see
 /// [`indexed_items`]). Runs as a single batched transaction via
-/// [`DocsStore::upsert_batch`] — a crate can have thousands of items, and
-/// upserting them one at a time would turn a single fetch into thousands of
-/// individual commits.
+/// [`DocsStore::upsert_batch_reconciled`] — a crate can have thousands of
+/// items, and upserting them one at a time would turn a single fetch into
+/// thousands of individual commits. Reconciling against `crate_id_path`'s
+/// existing items also soft-deletes any item that was present in a
+/// previous fetch but isn't in this one (renamed, removed, or made
+/// private), so search can't keep surfacing docs for an item that no
+/// longer exists in the crate's current index.
 fn index_items(
     store: &DocsStore,
     decompressed_json: &[u8],
@@ -137,10 +141,11 @@ fn index_items(
     let index_crate: IndexCrate = serde_json::from_slice(decompressed_json)
         .map_err(|e| RustdocError::InvalidJson(e.to_string()))?;
 
+    let item_prefix = format!("{crate_id_path}/item/");
     let batch: Vec<BatchItem> = indexed_items(&index_crate)
         .into_iter()
         .map(|item| BatchItem {
-            path: format!("{crate_id_path}/item/{}", item.fq_path),
+            path: format!("{item_prefix}{}", item.fq_path),
             content: item.docs,
             title: item.name,
             doc_type: "rust-item".to_string(),
@@ -149,7 +154,7 @@ fn index_items(
         })
         .collect();
 
-    Ok(store.upsert_batch(&batch)?)
+    Ok(store.upsert_batch_reconciled(&item_prefix, &batch)?)
 }
 
 fn store_raw_json_blob(store: &DocsStore, decompressed_json: &[u8]) -> Result<String, StoreError> {
@@ -319,6 +324,74 @@ mod tests {
         );
     }
 
+    #[test]
+    fn refetch_removes_items_no_longer_present_in_the_crate_index() {
+        // Regression for the staleness gap: item A and B are indexed by the
+        // first fetch, then a refetch of the same crate_id_path ("latest")
+        // only has item A -- B must no longer be gettable or searchable
+        // afterward, not linger from the first fetch forever.
+        let first_json = br#"{
+            "root": 0,
+            "index": {
+                "0": { "docs": "A fake crate for testing." },
+                "1": { "name": "State", "docs": "Extractor for shared state." },
+                "2": { "name": "Router", "docs": "The router type." }
+            },
+            "paths": {
+                "1": { "crate_id": 0, "path": ["fake_crate", "extract", "State"], "kind": "struct" },
+                "2": { "crate_id": 0, "path": ["fake_crate", "Router"], "kind": "struct" }
+            }
+        }"#;
+        let store = DocsStore::open_memory().unwrap();
+        let fetcher = FakeFetcher {
+            response: zstd::stream::encode_all(&first_json[..], 0).unwrap(),
+        };
+        fetch_and_store(&fetcher, &store, "fake-crate", "latest").unwrap();
+        assert!(
+            store
+                .get_by_path("docsrs/fake-crate/latest/item/fake_crate::Router")
+                .unwrap()
+                .is_some()
+        );
+
+        // "latest" moved to a new version where Router was removed.
+        let second_json = br#"{
+            "root": 0,
+            "index": {
+                "0": { "docs": "A fake crate for testing." },
+                "1": { "name": "State", "docs": "Extractor for shared state." }
+            },
+            "paths": {
+                "1": { "crate_id": 0, "path": ["fake_crate", "extract", "State"], "kind": "struct" }
+            }
+        }"#;
+        let fetcher = FakeFetcher {
+            response: zstd::stream::encode_all(&second_json[..], 0).unwrap(),
+        };
+        fetch_and_store(&fetcher, &store, "fake-crate", "latest").unwrap();
+
+        assert!(
+            store
+                .get_by_path("docsrs/fake-crate/latest/item/fake_crate::Router")
+                .unwrap()
+                .is_none(),
+            "removed item must not be gettable by path after refetch"
+        );
+        let hits = store.search("The router type", 10).unwrap();
+        assert!(
+            hits.is_empty(),
+            "removed item must not be findable via search after refetch"
+        );
+
+        // The item still present in the new fetch must be unaffected.
+        assert!(
+            store
+                .get_by_path("docsrs/fake-crate/latest/item/fake_crate::extract::State")
+                .unwrap()
+                .is_some()
+        );
+    }
+
     #[test]
     fn fetch_and_store_survives_unparseable_item_index() {
         // Regression: an item whose "kind" is a value the pinned

crates/flare-docs/src/store.rs

@@ -105,6 +105,74 @@ impl DocsStore {
         let conn = self.inner.conn();
         let tx =
             rusqlite::Transaction::new_unchecked(&conn, rusqlite::TransactionBehavior::Immediate)?;
+        let written = Self::write_batch_items(&tx, items)?;
+        tx.commit()?;
+        Ok(written)
+    }
+
+    /// Like [`Self::upsert_batch`], but also reconciles the store against
+    /// the fresh fetch: any existing, non-deleted document whose path
+    /// starts with `path_prefix` and is *not* in `items` gets soft-deleted
+    /// (`deleted_at` set), in the same transaction as the batch upsert.
+    ///
+    /// This is for cache-refresh callers like rustdoc per-item indexing,
+    /// where a refetch's item set is authoritative for everything under
+    /// `path_prefix` — an item that disappeared from the new fetch (renamed,
+    /// removed, made private) must stop being findable, not linger from the
+    /// previous fetch forever.
+    pub fn upsert_batch_reconciled(
+        &self,
+        path_prefix: &str,
+        items: &[BatchItem],
+    ) -> Result<usize, Error> {
+        let conn = self.inner.conn();
+        let tx =
+            rusqlite::Transaction::new_unchecked(&conn, rusqlite::TransactionBehavior::Immediate)?;
+        let written = Self::write_batch_items(&tx, items)?;
+
+        let now = db_kit::ids::now();
+        let fresh_paths: std::collections::HashSet<&str> =
+            items.iter().map(|i| i.path.as_str()).collect();
+        let like_pattern = format!("{}%", escape_like(path_prefix));
+        let mut stmt = tx.prepare(
+            "SELECT path FROM store_documents
+             WHERE project_id = ?1 AND path LIKE ?2 ESCAPE '\\' AND deleted_at IS NULL",
+        )?;
+        let stored_paths: Vec<String> = stmt
+            .query_map(rusqlite::params![PROJECT_ID, like_pattern], |row| {
+                row.get(0)
+            })?
+            .collect::<rusqlite::Result<_>>()?;
+        drop(stmt);
+
+        for stored_path in stored_paths {
+            if !fresh_paths.contains(stored_path.as_str()) {
+                tx.execute(
+                    "UPDATE store_documents SET deleted_at = ?1 WHERE project_id = ?2 AND path = ?3",
+                    rusqlite::params![now, PROJECT_ID, stored_path],
+                )?;
+            }
+        }
+
+        tx.commit()?;
+        Ok(written)
+    }
+
+    /// Upserts `items` within an already-open transaction. Skips any item
+    /// whose content is byte-identical to what's already stored (no row
+    /// write, no FTS resync, no version bump) — a crate whose docs haven't
+    /// changed since the last fetch would otherwise still pay the full
+    /// per-item write cost (and grow `version` unboundedly) on every
+    /// refresh, the same amplification problem
+    /// [`agentflare_store::Store::doc_upsert_with_opts`]'s content-hash
+    /// short-circuit exists to avoid for the single-doc path.
+    ///
+    /// Returns the number of items actually written (inserted or changed),
+    /// not the total number of items passed in.
+    fn write_batch_items(
+        tx: &rusqlite::Transaction,
+        items: &[BatchItem],
+    ) -> rusqlite::Result<usize> {
         let now = db_kit::ids::now();
         let mut written = 0usize;
 
@@ -161,11 +229,19 @@ impl DocsStore {
             written += 1;
         }
 
-        tx.commit()?;
         Ok(written)
     }
 }
 
+/// Escapes `%`, `_`, and `\` in a LIKE prefix so it matches only literal
+/// text (used with `ESCAPE '\\'`) — crate names commonly contain `_`
+/// (e.g. `serde_json`), which would otherwise match any single character.
+fn escape_like(s: &str) -> String {
+    s.replace('\\', "\\\\")
+        .replace('%', "\\%")
+        .replace('_', "\\_")
+}
+
 /// One document to write via [`DocsStore::upsert_batch`].
 #[derive(Debug, Clone)]
 pub struct BatchItem {