fix(agent): provenance survives compaction; bound alert/session growth; snapshot the WAL

- ProvenanceStore::remap: compaction renumbers cache indices (which are the
  provenance record ids) but nothing renumbered the ledger, so after any
  compaction — including the automatic one in delete() — every surviving
  record's hash was filed under a different record and the next
  save_or_update raised a bogus High "integrity mismatch" alert.
- Pending anomaly alerts are capped (newest 1024 kept). Alerts never block a
  save, and a session over its write limit alerts on every write, so a caller
  that didn't drain them grew the queue without bound.
- WriteAnomalyDetector tracks at most 4096 sessions, forgetting the
  least-active half on overflow instead of leaking one entry per session id
  for the life of the process.
- snapshot() copies the pending WAL next to the .h5 copy, so a snapshot is
  the store as it is now rather than as of the last checkpoint (it used to
  silently omit up to wal_max_entries recent saves).

Co-Authored-By: Claude Fable 5.1 <[email protected]>

crates/clawhdf5-agent/src/anomaly.rs

@@ -161,6 +161,9 @@ pub struct WriteEvent {
 // WriteAnomalyDetector
 // ---------------------------------------------------------------------------
 
+/// Upper bound on distinct session ids the detector tracks at once.
+const MAX_TRACKED_SESSIONS: usize = 4096;
+
 /// Tracks write events and raises alerts for suspicious behaviour.
 #[derive(Debug)]
 pub struct WriteAnomalyDetector {
@@ -189,6 +192,23 @@ impl WriteAnomalyDetector {
         if event.timestamp > self.last_timestamp {
             self.last_timestamp = event.timestamp;
         }
+        // Bound the per-session map: a long-lived process sees an unbounded
+        // number of distinct session ids. When it overflows, forget the
+        // sessions with the fewest writes (they are furthest from the limit
+        // this map exists to enforce); the current one is re-added below.
+        if self.session_counts.len() >= MAX_TRACKED_SESSIONS
+            && !self.session_counts.contains_key(&event.session_id)
+        {
+            let mut counts: Vec<u32> = self.session_counts.values().copied().collect();
+            let keep_from = counts.len() / 2;
+            counts.select_nth_unstable(keep_from);
+            let threshold = counts[keep_from];
+            self.session_counts.retain(|_, c| *c >= threshold);
+            if self.session_counts.len() >= MAX_TRACKED_SESSIONS {
+                // Every session had the same count: drop them all.
+                self.session_counts.clear();
+            }
+        }
         *self
             .session_counts
             .entry(event.session_id.clone())

crates/clawhdf5-agent/src/lib.rs

@@ -205,6 +205,9 @@ pub trait AgentMemory {
     fn get_session_summary(&self, session_id: &str) -> Result<Option<String>>;
 }
 
+/// Most anomaly alerts kept between `take_anomaly_alerts` calls.
+const MAX_PENDING_ALERTS: usize = 1024;
+
 // --- HDF5Memory ---
 
 pub struct HDF5Memory {
@@ -527,7 +530,7 @@ impl HDF5Memory {
         .into_iter()
         .flatten()
         {
-            self.anomaly_alerts.push(alert);
+            self.push_anomaly_alert(alert);
         }
     }
 
@@ -548,7 +551,7 @@ impl HDF5Memory {
             .provenance
             .verify_integrity(record_id as u64, current_chunk)
         {
-            self.anomaly_alerts.push(anomaly::AnomalyAlert {
+            self.push_anomaly_alert(anomaly::AnomalyAlert {
                 severity: anomaly::Severity::High,
                 message: format!(
                     "provenance integrity mismatch for record {record_id}: stored content no \
@@ -559,6 +562,18 @@ impl HDF5Memory {
         }
     }
 
+    /// Queue an alert, keeping only the most recent [`MAX_PENDING_ALERTS`].
+    /// Alerts never block a save, so a caller that never drains them — or a
+    /// session stuck over its write limit, which alerts on every write —
+    /// must not be able to grow this without bound.
+    fn push_anomaly_alert(&mut self, alert: anomaly::AnomalyAlert) {
+        if self.anomaly_alerts.len() >= MAX_PENDING_ALERTS {
+            let excess = self.anomaly_alerts.len() + 1 - MAX_PENDING_ALERTS;
+            self.anomaly_alerts.drain(..excess);
+        }
+        self.anomaly_alerts.push(alert);
+    }
+
     /// Alerts raised by anomaly detection / provenance checks since the last
     /// call, draining the internal queue.
     pub fn take_anomaly_alerts(&mut self) -> Vec<anomaly::AnomalyAlert> {
@@ -874,8 +889,10 @@ impl AgentMemory for HDF5Memory {
     }
 
     fn compact(&mut self) -> Result<usize> {
-        let (removed, _index_map) = self.cache.compact();
+        let (removed, index_map) = self.cache.compact();
         if removed > 0 {
+            // Record ids are cache indices, which compaction just renumbered.
+            self.provenance.remap(&index_map);
             // Compaction renumbers cache indices; rebuild the index to match.
             self.hnsw_mark_dirty();
             self.flush()?;
@@ -892,7 +909,18 @@ impl AgentMemory for HDF5Memory {
     }
 
     fn snapshot(&self, dest: &Path) -> Result<PathBuf> {
-        storage::snapshot_file(&self.config.path, dest)
+        let snapshot = storage::snapshot_file(&self.config.path, dest)?;
+        // Entries saved since the last checkpoint live only in the WAL. Copy
+        // it alongside (where `open()` looks for it) so the snapshot is the
+        // store as it is now, not as of the last checkpoint. The .h5 is
+        // copied first: if a checkpoint lands in between, the WAL copy is
+        // empty or its prefix is skipped via the checkpoint mark — never
+        // applied twice.
+        let wal_path = self.config.path.with_extension("h5.wal");
+        if self.wal.as_ref().is_some_and(|w| !w.is_empty()) && wal_path.exists() {
+            storage::snapshot_file(&wal_path, &snapshot.with_extension("h5.wal"))?;
+        }
+        Ok(snapshot)
     }
 
     fn add_session(
@@ -1508,6 +1536,65 @@ mod tests {
         assert_eq!(mem.count(), 1);
     }
 
+    #[test]
+    fn compaction_does_not_cause_false_provenance_alerts() {
+        let dir = TempDir::new().unwrap();
+        let mut mem = HDF5Memory::create(make_config(&dir)).unwrap();
+        for name in ["a", "b", "c"] {
+            let mut e = make_entry(name, &[1.0, 0.0, 0.0, 0.0]);
+            e.tags = format!("tag-{name}");
+            mem.save(e).unwrap();
+        }
+        // 1 of 3 tombstoned exceeds compact_threshold, so delete() compacts.
+        mem.delete(0).unwrap();
+        assert_eq!(mem.cache.chunks, ["b", "c"]);
+        mem.take_anomaly_alerts();
+
+        // "c" moved from id 2 to id 1. Its recorded hash must have moved too,
+        // or this update is checked against "b"'s hash and flagged.
+        let mut update = make_entry("c2", &[0.0, 1.0, 0.0, 0.0]);
+        update.tags = "tag-c".into();
+        assert_eq!(mem.save_or_update(update).unwrap(), 1);
+        let alerts = mem.take_anomaly_alerts();
+        assert!(
+            !alerts.iter().any(|a| a.message.contains("provenance")),
+            "{alerts:?}"
+        );
+    }
+
+    #[test]
+    fn pending_alerts_are_bounded() {
+        let dir = TempDir::new().unwrap();
+        let mut mem = HDF5Memory::create(make_config(&dir)).unwrap();
+        for i in 0..(MAX_PENDING_ALERTS + 50) {
+            mem.push_anomaly_alert(anomaly::AnomalyAlert {
+                severity: anomaly::Severity::Low,
+                message: format!("alert {i}"),
+                timestamp: i as f64,
+            });
+        }
+        let alerts = mem.take_anomaly_alerts();
+        assert_eq!(alerts.len(), MAX_PENDING_ALERTS);
+        assert_eq!(alerts[0].message, "alert 50", "oldest are dropped first");
+    }
+
+    #[test]
+    fn snapshot_includes_entries_still_in_the_wal() {
+        let dir = TempDir::new().unwrap();
+        let mut config = make_config(&dir);
+        config.wal_enabled = true;
+        let mut mem = HDF5Memory::create(config).unwrap();
+        mem.save(make_entry("checkpointed", &[1.0, 0.0, 0.0, 0.0]))
+            .unwrap();
+        mem.flush_wal().unwrap();
+        mem.save(make_entry("wal-only", &[0.0, 1.0, 0.0, 0.0]))
+            .unwrap();
+
+        let snap = mem.snapshot(&dir.path().join("snap.h5")).unwrap();
+        let restored = HDF5Memory::open(&snap).unwrap();
+        assert_eq!(restored.cache.chunks, ["checkpointed", "wal-only"]);
+    }
+
     #[test]
     fn store_has_a_single_writer() {
         let dir = TempDir::new().unwrap();

crates/clawhdf5-agent/src/provenance.rs

@@ -105,6 +105,23 @@ impl ProvenanceStore {
         self.records.insert(provenance.record_id, provenance);
     }
 
+    /// Renumber records after the store was compacted. `index_map[old]` is
+    /// the record's new id, or `None` if it was removed. Without this, every
+    /// surviving record's hash ends up filed under some other record's id and
+    /// the next integrity check reports a bogus mismatch.
+    pub fn remap(&mut self, index_map: &[Option<usize>]) {
+        let old = std::mem::take(&mut self.records);
+        for (old_id, mut prov) in old {
+            let new_id = usize::try_from(old_id)
+                .ok()
+                .and_then(|i| index_map.get(i).copied().flatten());
+            if let Some(new_id) = new_id {
+                prov.record_id = new_id as u64;
+                self.records.insert(new_id as u64, prov);
+            }
+        }
+    }
+
     /// Retrieve by record ID.
     pub fn get(&self, record_id: u64) -> Option<&MemoryProvenance> {
         self.records.get(&record_id)