feat(agent): single-writer lock, read-only open, recoverable WAL

- HDF5Memory::create/open take an exclusive advisory lock on <store>.h5.lock
  (std File::try_lock, no new dependency). The store lives in memory and is
  rewritten wholesale at each checkpoint, so two handles on one store used to
  silently destroy each other's data; a second writer now gets
  MemoryError::Locked. The OS drops the lock with the descriptor, so a crash
  never leaves a stale lock. Acquisition retries for ~250 ms to absorb a
  previous owner that is mid-teardown; AsyncHDF5Memory::shutdown releases the
  lock once its writer task has stopped.
- HDF5Memory::open_read_only: a lock-free, point-in-time view (checkpoint +
  current WAL contents, replayed in memory) that never writes — it does not
  repair, upgrade or move the WAL, and anything that would persist returns an
  error. The CLI's recall/stats/agents-md/export use it, so a store can be
  inspected while an agent has it open. Tests that reopened a store purely to
  verify on-disk state now use it.
- open() no longer fails on a WAL that cannot possibly be replayed (torn
  header, bad magic): it is moved to <store>.h5.wal.corrupt-<ts>, reported via
  HDF5Memory::quarantined_wal(), and the healthy .h5 opens from its last
  checkpoint. A well-formed header with an unknown version still fails and is
  left untouched — most likely a newer build's WAL, which must not be
  discarded.

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

crates/clawhdf5-agent/src/async_memory.rs

@@ -408,6 +408,10 @@ impl AsyncHDF5Memory {
         let (tx, rx) = oneshot::channel();
         let _ = self.write_tx.send(WriteCmd::Shutdown(tx)).await;
         let _ = rx.await;
+        // The writer task has stopped, so nothing can write through this
+        // handle any more: release the single-writer lock now rather than at
+        // drop, so the store can be reopened while `self` is still in scope.
+        self.inner.lock().await.release_store_lock();
         Ok(())
     }
 }

crates/clawhdf5-agent/src/lib.rs

@@ -37,6 +37,7 @@ pub mod schema;
 pub mod search;
 pub mod session;
 pub mod storage;
+mod store_lock;
 pub mod temporal;
 pub mod wal;
 
@@ -86,6 +87,8 @@ pub enum MemoryError {
     Hdf5(String),
     Schema(String),
     NotFound(String),
+    /// Another `HDF5Memory` (in this or another process) has the store open.
+    Locked(String),
 }
 
 impl std::fmt::Display for MemoryError {
@@ -95,6 +98,7 @@ impl std::fmt::Display for MemoryError {
             MemoryError::Hdf5(e) => write!(f, "HDF5 error: {e}"),
             MemoryError::Schema(e) => write!(f, "schema error: {e}"),
             MemoryError::NotFound(e) => write!(f, "not found: {e}"),
+            MemoryError::Locked(e) => write!(f, "store is locked: {e}"),
         }
     }
 }
@@ -240,6 +244,15 @@ pub struct HDF5Memory {
     /// via [`HDF5Memory::take_anomaly_alerts`]. Saves are never blocked on
     /// these — surfacing is opt-in for callers that want to act on them.
     anomaly_alerts: Vec<anomaly::AnomalyAlert>,
+    /// Opened with [`HDF5Memory::open_read_only`]: nothing may reach the disk.
+    read_only: bool,
+    /// A WAL that `open()` could not read and moved aside; see
+    /// [`HDF5Memory::quarantined_wal`].
+    quarantined_wal: Option<PathBuf>,
+    /// Single-writer guard. Declared last so it is released only after the
+    /// WAL and everything else has been dropped. `None` once a wrapper that
+    /// has stopped all writes released it early (see `release_store_lock`).
+    _lock: Option<store_lock::StoreLock>,
 }
 
 impl std::fmt::Debug for HDF5Memory {
@@ -251,6 +264,7 @@ impl std::fmt::Debug for HDF5Memory {
 impl HDF5Memory {
     /// Create a new HDF5 memory file with the given configuration.
     pub fn create(config: MemoryConfig) -> Result<Self> {
+        let lock = store_lock::StoreLock::acquire(&config.path)?;
         let cache = MemoryCache::new(config.embedding_dim);
         let sessions = SessionCache::new();
         let knowledge = KnowledgeCache::new();
@@ -282,17 +296,103 @@ impl HDF5Memory {
             provenance: provenance::ProvenanceStore::new(),
             anomaly: anomaly::WriteAnomalyDetector::new(anomaly::AnomalyConfig::default()),
             anomaly_alerts: Vec::new(),
+            read_only: false,
+            quarantined_wal: None,
+            _lock: Some(lock),
         })
     }
 
     /// Open an existing HDF5 memory file.
+    /// If the WAL at `wal_path` can't possibly be replayed — its header is
+    /// torn (crash while the file was being created) or isn't a WAL header at
+    /// all — move it aside so a healthy `.h5` still opens, and return where it
+    /// went. A well-formed header with an *unknown version* is left alone and
+    /// still fails `open()`: that WAL was most likely written by a newer
+    /// build, and discarding it would lose data this binary merely can't read.
+    fn quarantine_unreadable_wal(wal_path: &Path) -> Result<Option<PathBuf>> {
+        if !wal_path.exists() {
+            return Ok(None);
+        }
+        let reason = match wal::wal_header_status(wal_path)? {
+            wal::WalHeaderStatus::Readable | wal::WalHeaderStatus::UnknownVersion(_) => {
+                return Ok(None);
+            }
+            wal::WalHeaderStatus::Torn => "truncated header",
+            wal::WalHeaderStatus::BadMagic => "bad magic bytes",
+        };
+        let ts = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_secs();
+        let dest = wal_path.with_extension(format!("wal.corrupt-{ts}"));
+        std::fs::rename(wal_path, &dest)?;
+        eprintln!(
+            "clawhdf5-agent: WAL {} is unreadable ({reason}); moved to {} and continuing \
+             from the last checkpoint",
+            wal_path.display(),
+            dest.display()
+        );
+        Ok(Some(dest))
+    }
+
+    /// Give up the single-writer lock before this value is dropped. Only for
+    /// wrappers that have already stopped every write path but keep the handle
+    /// alive (`AsyncHDF5Memory::shutdown`), so the store can be reopened.
+    #[cfg_attr(not(feature = "async"), allow(dead_code))]
+    pub(crate) fn release_store_lock(&mut self) {
+        self._lock = None;
+    }
+
+    /// Where `open()` moved an unreadable WAL, if it had to. Entries that were
+    /// only in that WAL are not in this store; the file is kept for forensics.
+    pub fn quarantined_wal(&self) -> Option<&Path> {
+        self.quarantined_wal.as_deref()
+    }
+
     pub fn open(path: &Path) -> Result<Self> {
+        Self::open_impl(path, false)
+    }
+
+    /// Open a store for reading only, without taking the single-writer lock —
+    /// so it works while another `HDF5Memory` (in this or another process)
+    /// has the store open for writing, e.g. to inspect what is on disk.
+    ///
+    /// It loads the last checkpoint plus whatever the WAL held at that
+    /// moment; it is a point-in-time view and does not follow later writes.
+    /// Nothing is written: the WAL file is not repaired, upgraded or moved,
+    /// and every operation that would persist state returns an error.
+    pub fn open_read_only(path: &Path) -> Result<Self> {
+        Self::open_impl(path, true)
+    }
+
+    fn open_impl(path: &Path, read_only: bool) -> Result<Self> {
+        let lock = if read_only {
+            None
+        } else {
+            Some(store_lock::StoreLock::acquire(path)?)
+        };
         let ((config, mut cache, sessions, knowledge), wal_applied) =
             storage::read_from_disk_with_mark(path)?;
 
         // Replay WAL if present
         let wal_path = path.with_extension("h5.wal");
-        let wal = if wal_path.exists() {
+        let quarantined_wal = if read_only {
+            None
+        } else {
+            Self::quarantine_unreadable_wal(&wal_path)?
+        };
+        let wal = if read_only {
+            // Replay in memory only. `WalFile::open` would truncate a torn
+            // tail and may rewrite the header — both belong to the writer. An
+            // unreadable WAL is simply skipped: the writer will deal with it.
+            if wal_path.exists()
+                && let Ok(entries) =
+                    wal::WalFile::read_entries_for_migration(&wal_path, wal_applied)
+            {
+                wal::replay_into_cache(&entries, &mut cache);
+            }
+            None
+        } else if wal_path.exists() {
             // Uses the migration-only reader since this is the one legitimate
             // path that may need to read a legacy (pre-CRC) WAL file — see
             // WalFile::read_entries_for_migration.
@@ -331,6 +431,9 @@ impl HDF5Memory {
             provenance: provenance::ProvenanceStore::new(),
             anomaly: anomaly::WriteAnomalyDetector::new(anomaly::AnomalyConfig::default()),
             anomaly_alerts: Vec::new(),
+            read_only,
+            quarantined_wal,
+            _lock: lock,
         })
     }
 
@@ -340,6 +443,12 @@ impl HDF5Memory {
     /// also clear the WAL, otherwise `open()` will replay stale entries
     /// on top of the already-persisted data, duplicating them.
     fn flush(&mut self) -> Result<()> {
+        if self.read_only {
+            return Err(MemoryError::Io(std::io::Error::new(
+                std::io::ErrorKind::PermissionDenied,
+                "store was opened read-only",
+            )));
+        }
         // Record which WAL prefix this checkpoint contains, so a crash before
         // the truncate below can't replay those entries a second time.
         let wal_applied = self.wal.as_ref().map(|w| w.mark());
@@ -1399,6 +1508,90 @@ mod tests {
         assert_eq!(mem.count(), 1);
     }
 
+    #[test]
+    fn store_has_a_single_writer() {
+        let dir = TempDir::new().unwrap();
+        let config = make_config(&dir);
+        let path = config.path.clone();
+        let mem = HDF5Memory::create(config).unwrap();
+        assert!(matches!(
+            HDF5Memory::open(&path),
+            Err(MemoryError::Locked(_))
+        ));
+        drop(mem);
+        HDF5Memory::open(&path).unwrap();
+    }
+
+    #[test]
+    fn read_only_open_coexists_with_a_writer_and_never_writes() {
+        let dir = TempDir::new().unwrap();
+        let mut config = make_config(&dir);
+        config.wal_enabled = true;
+        let path = config.path.clone();
+        let wal_path = path.with_extension("h5.wal");
+        let mut writer = HDF5Memory::create(config).unwrap();
+        writer
+            .save(make_entry("pending", &[1.0, 0.0, 0.0, 0.0]))
+            .unwrap();
+        let wal_before = std::fs::read(&wal_path).unwrap();
+        let h5_before = std::fs::read(&path).unwrap();
+
+        // Sees the checkpoint plus the writer's un-checkpointed WAL entry.
+        let mut reader = HDF5Memory::open_read_only(&path).unwrap();
+        assert_eq!(reader.cache.chunks, ["pending"]);
+        assert!(reader.save(make_entry("nope", &[0.0; 4])).is_err());
+        assert!(reader.flush_wal().is_err());
+        drop(reader);
+
+        assert_eq!(std::fs::read(&wal_path).unwrap(), wal_before);
+        assert_eq!(std::fs::read(&path).unwrap(), h5_before);
+        // The writer is unaffected.
+        writer
+            .save(make_entry("more", &[0.0, 1.0, 0.0, 0.0]))
+            .unwrap();
+    }
+
+    #[test]
+    fn unreadable_wal_is_quarantined_not_fatal() {
+        let dir = TempDir::new().unwrap();
+        let config = make_config(&dir);
+        let path = config.path.clone();
+        let wal_path = path.with_extension("h5.wal");
+        {
+            let mut mem = HDF5Memory::create(config).unwrap();
+            mem.save(make_entry("kept", &[1.0, 0.0, 0.0, 0.0])).unwrap();
+            mem.flush_wal().unwrap();
+        }
+        std::fs::write(&wal_path, b"not a wal at all").unwrap();
+
+        let mem = HDF5Memory::open(&path).unwrap();
+        assert_eq!(mem.cache.chunks, ["kept"]);
+        let moved = mem.quarantined_wal().expect("WAL should be quarantined");
+        assert_eq!(std::fs::read(moved).unwrap(), b"not a wal at all");
+        // A fresh, valid WAL took its place.
+        assert!(wal::WalFile::read_entries(&wal_path).unwrap().is_empty());
+    }
+
+    #[test]
+    fn wal_from_a_newer_build_is_refused_not_discarded() {
+        let dir = TempDir::new().unwrap();
+        let mut config = make_config(&dir);
+        config.wal_enabled = true;
+        let path = config.path.clone();
+        let wal_path = path.with_extension("h5.wal");
+        drop(HDF5Memory::create(config).unwrap());
+        let mut bytes = std::fs::read(&wal_path).unwrap();
+        bytes[4] = 200; // a version this build has never heard of
+        std::fs::write(&wal_path, &bytes).unwrap();
+
+        assert!(HDF5Memory::open(&path).is_err());
+        assert_eq!(
+            std::fs::read(&wal_path).unwrap(),
+            bytes,
+            "WAL left untouched"
+        );
+    }
+
     #[test]
     fn empty_file_operations() {
         let dir = TempDir::new().unwrap();
@@ -1407,6 +1600,7 @@ mod tests {
         let mem = HDF5Memory::create(config).unwrap();
         assert_eq!(mem.count(), 0);
         assert_eq!(mem.count_active(), 0);
+        drop(mem); // a store has a single writer; release it before reopening
 
         let mem2 = HDF5Memory::open(&path).unwrap();
         assert_eq!(mem2.count(), 0);

crates/clawhdf5-agent/src/store_lock.rs

@@ -0,0 +1,79 @@
+//! Single-writer guard for a memory store.
+//!
+//! `HDF5Memory` keeps the whole store in memory and rewrites the `.h5` file at
+//! every checkpoint, so two handles on one store (two processes, or two opens
+//! in one process) silently destroy each other's data: whoever checkpoints
+//! last wins, and both append to the same WAL with independent CRC chains.
+//! The lock turns that into an immediate, explicit error.
+
+use std::fs::{File, OpenOptions, TryLockError};
+use std::path::{Path, PathBuf};
+
+use crate::MemoryError;
+
+const LOCK_RETRIES: u32 = 25;
+const LOCK_RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(10);
+
+/// An exclusive advisory lock on `<store>.h5.lock`, held for the lifetime of
+/// the owning `HDF5Memory` and released when it is dropped (or when the
+/// process dies — the OS drops the lock with the file descriptor, so a crash
+/// never leaves a stale lock behind; the empty lock file itself is harmless).
+#[derive(Debug)]
+pub(crate) struct StoreLock {
+    _file: File,
+}
+
+impl StoreLock {
+    pub(crate) fn lock_path(store: &Path) -> PathBuf {
+        store.with_extension("h5.lock")
+    }
+
+    pub(crate) fn acquire(store: &Path) -> Result<Self, MemoryError> {
+        let path = Self::lock_path(store);
+        let file = OpenOptions::new()
+            .create(true)
+            .truncate(false)
+            .write(true)
+            .open(&path)?;
+        // A previous owner may be mid-teardown (e.g. an `AsyncHDF5Memory`
+        // dropped without `shutdown()`: its background task releases the
+        // store a moment later), so give the lock a short, bounded grace
+        // period before reporting a genuine second writer.
+        let mut attempts_left = LOCK_RETRIES;
+        loop {
+            match file.try_lock() {
+                Ok(()) => return Ok(Self { _file: file }),
+                Err(TryLockError::WouldBlock) if attempts_left > 0 => {
+                    attempts_left -= 1;
+                    std::thread::sleep(LOCK_RETRY_DELAY);
+                }
+                Err(TryLockError::WouldBlock) => {
+                    return Err(MemoryError::Locked(format!(
+                        "{} is already open in this or another process (lock file {})",
+                        store.display(),
+                        path.display()
+                    )));
+                }
+                Err(TryLockError::Error(e)) => return Err(MemoryError::Io(e)),
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn second_acquire_fails_until_first_is_dropped() {
+        let dir = tempfile::TempDir::new().unwrap();
+        let store = dir.path().join("s.h5");
+        let first = StoreLock::acquire(&store).unwrap();
+        assert!(matches!(
+            StoreLock::acquire(&store),
+            Err(MemoryError::Locked(_))
+        ));
+        drop(first);
+        StoreLock::acquire(&store).unwrap();
+    }
+}

crates/clawhdf5-agent/src/wal.rs

@@ -135,6 +135,45 @@ pub struct WalFile {
     chain_len: u64,
 }
 
+/// What a WAL file's 9-byte header looks like, without reading any entries.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum WalHeaderStatus {
+    /// A version this build can read (current or legacy).
+    Readable,
+    /// Shorter than a header — e.g. a crash while the file was being created.
+    /// It cannot contain entries.
+    Torn,
+    /// Not a WAL file at all.
+    BadMagic,
+    /// Well-formed header from a version this build doesn't know — most
+    /// likely written by a *newer* build. Never discard this: the entries are
+    /// probably fine, this binary just can't read them.
+    UnknownVersion(u8),
+}
+
+/// Classify the header of the WAL at `path`.
+pub fn wal_header_status(path: &Path) -> std::io::Result<WalHeaderStatus> {
+    let mut header = [0u8; WAL_HEADER_LEN as usize];
+    let mut f = File::open(path)?;
+    let mut filled = 0;
+    while filled < header.len() {
+        match f.read(&mut header[filled..])? {
+            0 => return Ok(WalHeaderStatus::Torn),
+            n => filled += n,
+        }
+    }
+    if header[0..4] != WAL_MAGIC {
+        return Ok(WalHeaderStatus::BadMagic);
+    }
+    Ok(match header[4] {
+        WAL_VERSION
+        | WAL_VERSION_CHAINED_NO_UPDATE
+        | WAL_VERSION_CRC_UNCHAINED
+        | WAL_VERSION_LEGACY_NO_CRC => WalHeaderStatus::Readable,
+        v => WalHeaderStatus::UnknownVersion(v),
+    })
+}
+
 /// A position in a WAL's CRC chain: `len` bytes of entries after the header,
 /// whose chained CRC is `crc`.
 ///

crates/clawhdf5-agent/tests/e2e_tests.rs

@@ -196,7 +196,7 @@ fn test_migration_round_trip() {
     mem.add_relation(e1, e2, "discusses", 0.8).unwrap();
 
     // Verify all data transferred by reopening
-    let reopened = HDF5Memory::open(&path).unwrap();
+    let reopened = HDF5Memory::open_read_only(&path).unwrap();
     assert_eq!(reopened.count(), 500);
 
     // Verify sessions
@@ -266,7 +266,7 @@ fn test_knowledge_graph_workflow() {
     assert_eq!(entity.entity_type, "library");
 
     // Persistence
-    let reopened = HDF5Memory::open(&path).unwrap();
+    let reopened = HDF5Memory::open_read_only(&path).unwrap();
     assert_eq!(reopened.knowledge().entities.len(), 4);
     assert_eq!(reopened.knowledge().relations.len(), 4);
 
@@ -316,7 +316,7 @@ fn test_multi_session_workflow() {
     assert_eq!(mem.count(), 100); // 5 sessions * 20 entries
 
     // Reopen and verify sessions
-    let reopened = HDF5Memory::open(&path).unwrap();
+    let reopened = HDF5Memory::open_read_only(&path).unwrap();
     for sess in 0..5 {
         let summary = reopened
             .get_session_summary(&format!("sess_{sess}"))
@@ -460,7 +460,7 @@ fn test_snapshot_and_continue() {
     assert_eq!(snap_mem.count(), 50);
 
     // Original should have 100
-    let orig_mem = HDF5Memory::open(&path).unwrap();
+    let orig_mem = HDF5Memory::open_read_only(&path).unwrap();
     assert_eq!(orig_mem.count(), 100);
 }
 
@@ -483,7 +483,7 @@ fn test_config_persistence_across_ops() {
     mem.add_session("s1", 0, 0, "ch", "summary").unwrap();
     mem.add_entity("Entity", "type", -1).unwrap();
 
-    let reopened = HDF5Memory::open(&path).unwrap();
+    let reopened = HDF5Memory::open_read_only(&path).unwrap();
     assert_eq!(reopened.config().embedding_dim, 128);
     assert_eq!(reopened.config().embedder, "custom:my-embedder-v2");
     assert_eq!(reopened.config().chunk_size, 2048);
@@ -695,7 +695,7 @@ fn test_large_text_chunks() {
     mem.save_batch(entries).unwrap();
 
     // Reopen and verify
-    let reopened = HDF5Memory::open(&path).unwrap();
+    let reopened = HDF5Memory::open_read_only(&path).unwrap();
     assert_eq!(reopened.count(), 10);
 
     let (_, cache, _, _) = read_cache(&path);
@@ -752,7 +752,7 @@ fn test_interleaved_sessions_entries() {
     mem.flush_wal().unwrap();
 
     // Verify
-    let reopened = HDF5Memory::open(&path).unwrap();
+    let reopened = HDF5Memory::open_read_only(&path).unwrap();
     assert_eq!(reopened.count(), 6);
     assert_eq!(
         reopened.get_session_summary("s1").unwrap().as_deref(),
@@ -806,7 +806,7 @@ fn test_knowledge_graph_with_embeddings() {
     mem.add_relation(e_python, e_hdf5, "reads", 0.9).unwrap();
 
     // Verify entity-embedding linkage persists
-    let reopened = HDF5Memory::open(&path).unwrap();
+    let reopened = HDF5Memory::open_read_only(&path).unwrap();
     let rust_entity = reopened.knowledge().get_entity(e_rust).unwrap();
     assert_eq!(rust_entity.embedding_idx, idx0 as i64);
 

crates/clawhdf5-agent/tests/stress_tests.rs

@@ -105,7 +105,7 @@ fn test_heavy_tombstoning() {
     assert_eq!(mem.count_active(), 5000);
 
     // Verify persistence
-    let reopened = HDF5Memory::open(&path).unwrap();
+    let reopened = HDF5Memory::open_read_only(&path).unwrap();
     assert_eq!(reopened.count(), 5000);
 }
 
@@ -163,7 +163,7 @@ fn test_large_embeddings_1536() {
     assert_eq!(mem.count(), 10_000);
 
     // Verify persistence
-    let reopened = HDF5Memory::open(&path).unwrap();
+    let reopened = HDF5Memory::open_read_only(&path).unwrap();
     assert_eq!(reopened.count(), 10_000);
 
     // Verify search works on large dims
@@ -545,7 +545,7 @@ fn test_delete_all_entries() {
     assert_eq!(mem.count(), 0);
 
     // Verify persistence
-    let reopened = HDF5Memory::open(&path).unwrap();
+    let reopened = HDF5Memory::open_read_only(&path).unwrap();
     assert_eq!(reopened.count(), 0);
 }
 
@@ -639,7 +639,7 @@ fn test_unicode_content() {
     ];
     mem.save_batch(entries).unwrap();
 
-    let reopened = HDF5Memory::open(&path).unwrap();
+    let reopened = HDF5Memory::open_read_only(&path).unwrap();
     assert_eq!(reopened.count(), 3);
 
     let (_, cache, _, _) = clawhdf5_agent::storage::read_from_disk(&path).unwrap();
@@ -685,6 +685,6 @@ fn test_rapid_save_delete_cycles() {
     assert_eq!(removed, 250);
     assert_eq!(mem.count(), 250);
 
-    let reopened = HDF5Memory::open(&path).unwrap();
+    let reopened = HDF5Memory::open_read_only(&path).unwrap();
     assert_eq!(reopened.count(), 250);
 }