fix(agent): log save_or_update as an Update WAL record

A save_or_update that hit an existing record was logged as a plain Save, so
replaying the WAL appended a duplicate instead of updating in place. It is now
logged as WalEntryType::Update (0x04) carrying the target index, and replay
applies it with cache.update().

The WAL header version goes 3 -> 4 for the benefit of older binaries: they
don't know record type 0x04, would read it as a torn tail and truncate it and
everything after it. An unknown header version makes them refuse the file
instead. The framing is otherwise identical, so v3 files are read by the same
code and upgraded in place on open (the header is outside the CRC chain).

Also drop the redundant WAL truncate that several callers ran straight after
flush(), which already truncates.

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

crates/clawhdf5-agent/src/lib.rs

@@ -629,7 +629,7 @@ impl HDF5Memory {
         if let Some(existing_idx) = self.cache.find_by_tags(&entry.tags) {
             if let Some(ref mut w) = self.wal {
                 let wal_entry = wal::WalEntry {
-                    entry_type: wal::WalEntryType::Save,
+                    entry_type: wal::WalEntryType::Update,
                     timestamp: entry.timestamp,
                     chunk: entry.chunk.clone(),
                     embedding: entry.embedding.clone(),
@@ -637,6 +637,7 @@ impl HDF5Memory {
                     session_id: entry.session_id.clone(),
                     tags: entry.tags.clone(),
                     tombstone_index: None,
+                    update_index: Some(existing_idx),
                 };
                 w.append_save(&wal_entry)?;
             }
@@ -668,9 +669,6 @@ impl HDF5Memory {
                 .is_none_or(|w| w.pending_count() as usize > self.config.wal_max_entries);
             if needs_flush {
                 self.flush()?;
-                if let Some(ref mut w) = self.wal {
-                    w.truncate()?;
-                }
             }
             return Ok(existing_idx);
         }
@@ -691,6 +689,7 @@ impl AgentMemory for HDF5Memory {
                 session_id: entry.session_id.clone(),
                 tags: entry.tags.clone(),
                 tombstone_index: None,
+                update_index: None,
             };
             w.append_save(&wal_entry)?;
         }
@@ -716,9 +715,6 @@ impl AgentMemory for HDF5Memory {
             .is_none_or(|w| w.pending_count() as usize > self.config.wal_max_entries);
         if needs_flush {
             self.flush()?;
-            if let Some(ref mut w) = self.wal {
-                w.truncate()?;
-            }
         }
         Ok(idx)
     }
@@ -892,9 +888,6 @@ impl HDF5Memory {
             *w *= d;
         }
         self.flush()?;
-        if let Some(ref mut w) = self.wal {
-            w.truncate()?;
-        }
         Ok(())
     }
 
@@ -906,9 +899,6 @@ impl HDF5Memory {
     /// Explicit WAL merge: flush .h5, truncate WAL.
     pub fn flush_wal(&mut self) -> Result<()> {
         self.flush()?;
-        if let Some(ref mut w) = self.wal {
-            w.truncate()?;
-        }
         Ok(())
     }
 }

crates/clawhdf5-agent/src/wal.rs

@@ -28,7 +28,18 @@ const WAL_HEADER_LEN: u64 = WAL_MAGIC.len() as u64 + 1 + 4;
 /// its target Save) — the moved/inserted entry's stored CRC was computed
 /// against a different predecessor than the one now in front of it on disk,
 /// so the chain breaks at that point and replay stops there.
-const WAL_VERSION: u8 = 3;
+const WAL_VERSION: u8 = 4;
+
+/// The chained-CRC format before [`WalEntryType::Update`] records existed.
+/// Byte-for-byte the same framing as [`WAL_VERSION`], so it is read by the
+/// same code, and `WalFile::open` upgrades it in place by rewriting the
+/// header's version byte (the header is not covered by the CRC chain).
+///
+/// The bump exists for *older binaries*: they don't know record type 0x04,
+/// would treat it as a torn tail, and would truncate it — and everything
+/// after it — away. An unknown header version makes them refuse the file
+/// with a clear error instead.
+const WAL_VERSION_CHAINED_NO_UPDATE: u8 = 3;
 
 /// The previous WAL format version: still a CRC32 per entry (so a bit-flip
 /// within one entry is caught), but not chained to the previous entry's CRC
@@ -67,6 +78,10 @@ pub enum WalEntryType {
     Save = 0x01,
     Tombstone = 0x02,
     ActivationUpdate = 0x03,
+    /// Replace the record at `update_index` in place (`save_or_update` hit).
+    /// Logged as a plain `Save` before this existed, so replay appended a
+    /// duplicate instead of updating.
+    Update = 0x04,
 }
 
 impl WalEntryType {
@@ -75,6 +90,7 @@ impl WalEntryType {
             0x01 => Some(Self::Save),
             0x02 => Some(Self::Tombstone),
             0x03 => Some(Self::ActivationUpdate),
+            0x04 => Some(Self::Update),
             _ => None,
         }
     }
@@ -91,6 +107,8 @@ pub struct WalEntry {
     pub tags: String,
     /// For tombstone entries: the index of the entry to delete.
     pub tombstone_index: Option<usize>,
+    /// For update entries: the index of the record to replace.
+    pub update_index: Option<usize>,
 }
 
 /// How many entries to accumulate before updating the header entry_count.
@@ -155,7 +173,15 @@ impl WalFile {
             let mut ver = [0u8; 1];
             f.read_exact(&mut ver)?;
             match ver[0] {
-                WAL_VERSION => {
+                WAL_VERSION | WAL_VERSION_CHAINED_NO_UPDATE => {
+                    if ver[0] == WAL_VERSION_CHAINED_NO_UPDATE {
+                        // Same framing; stamp the current version so an older
+                        // binary refuses this file rather than truncating an
+                        // Update record it can't parse. See the constant.
+                        f.seek(SeekFrom::Start(4))?;
+                        f.write_all(&[WAL_VERSION])?;
+                        f.seek(SeekFrom::Start(5))?;
+                    }
                     let mut count_buf = [0u8; 4];
                     f.read_exact(&mut count_buf)?;
                     let header_count = u32::from_le_bytes(count_buf);
@@ -257,8 +283,20 @@ impl WalFile {
             4 + entry.session_id.len() +
             4 + entry.tags.len(),
         );
+        match entry.update_index {
+            Some(index) => {
+                let index = u32::try_from(index).map_err(|_| {
+                    MemoryError::Schema(format!("WAL update index {index} exceeds u32"))
+                })?;
+                buf.push(WalEntryType::Update as u8);
+                buf.extend_from_slice(&entry.timestamp.to_le_bytes());
+                buf.extend_from_slice(&index.to_le_bytes());
+            }
+            None => {
                 buf.push(WalEntryType::Save as u8);
                 buf.extend_from_slice(&entry.timestamp.to_le_bytes());
+            }
+        }
         serialize_str(&mut buf, &entry.chunk);
         buf.extend_from_slice(&(emb_len as u32).to_le_bytes());
         for &val in &entry.embedding {
@@ -375,7 +413,7 @@ impl WalFile {
         let entry_count_hint = u32::from_le_bytes([header[5], header[6], header[7], header[8]]);
 
         match header[4] {
-            WAL_VERSION => {
+            WAL_VERSION | WAL_VERSION_CHAINED_NO_UPDATE => {
                 let (entries, _final_crc, _verified_bytes) =
                     read_chained_entries(&mut f, 0, applied);
                 Ok(entries)
@@ -492,6 +530,28 @@ pub fn replay_into_cache(entries: &[WalEntry], cache: &mut crate::cache::MemoryC
                     entry.tags.clone(),
                 );
             }
+            WalEntryType::Update => match entry.update_index {
+                // The index was valid when the record was written; if the
+                // store no longer has it, keep the data rather than drop it.
+                Some(idx) if idx < cache.len() => cache.update(
+                    idx,
+                    entry.chunk.clone(),
+                    entry.embedding.clone(),
+                    entry.source_channel.clone(),
+                    entry.timestamp,
+                    entry.session_id.clone(),
+                ),
+                _ => {
+                    cache.push(
+                        entry.chunk.clone(),
+                        entry.embedding.clone(),
+                        entry.source_channel.clone(),
+                        entry.timestamp,
+                        entry.session_id.clone(),
+                        entry.tags.clone(),
+                    );
+                }
+            },
             WalEntryType::Tombstone => {
                 if let Some(idx) = entry.tombstone_index {
                     cache.mark_deleted(idx);
@@ -676,7 +736,14 @@ fn read_one_entry<R: Read>(r: &mut R) -> Result<Option<WalEntry>, ()> {
     let timestamp = f64::from_le_bytes(ts_buf);
 
     match entry_type {
-        WalEntryType::Save => {
+        WalEntryType::Save | WalEntryType::Update => {
+            let update_index = if entry_type == WalEntryType::Update {
+                let mut idx_buf = [0u8; 4];
+                r.read_exact(&mut idx_buf).map_err(|_| ())?;
+                Some(u32::from_le_bytes(idx_buf) as usize)
+            } else {
+                None
+            };
             let chunk = read_len_prefixed_str(r).map_err(|_| ())?;
             let embedding = read_embedding(r).map_err(|_| ())?;
             let source_channel = read_len_prefixed_str(r).map_err(|_| ())?;
@@ -691,6 +758,7 @@ fn read_one_entry<R: Read>(r: &mut R) -> Result<Option<WalEntry>, ()> {
                 session_id,
                 tags,
                 tombstone_index: None,
+                update_index,
             }))
         }
         WalEntryType::Tombstone => {
@@ -706,6 +774,7 @@ fn read_one_entry<R: Read>(r: &mut R) -> Result<Option<WalEntry>, ()> {
                 session_id: String::new(),
                 tags: String::new(),
                 tombstone_index: Some(idx),
+                update_index: None,
             }))
         }
         WalEntryType::ActivationUpdate => Ok(None),
@@ -729,6 +798,7 @@ mod tests {
             session_id: "sess-001".to_string(),
             tags: "tag1,tag2".to_string(),
             tombstone_index: None,
+            update_index: None,
         }
     }
 
@@ -859,6 +929,7 @@ mod tests {
                 session_id: "sess-öö-123".to_string(),
                 tags: "α,β,γ".to_string(),
                 tombstone_index: None,
+                update_index: None,
             };
             wal.append_save(&entry).unwrap();
         }
@@ -1052,6 +1123,58 @@ mod tests {
         assert_eq!(reopen_chunks(&h5_path), ["a", "b", "c", "d"]);
     }
 
+    #[test]
+    fn save_or_update_replays_as_update_not_duplicate() {
+        let dir = TempDir::new().unwrap();
+        let config = make_config(&dir);
+        let h5_path = config.path.clone();
+        {
+            let mut mem = HDF5Memory::create(config).unwrap();
+            let mut first = make_entry("v1", &[1.0, 0.0, 0.0, 0.0]);
+            first.tags = "key".into();
+            let mut second = make_entry("v2", &[0.0, 1.0, 0.0, 0.0]);
+            second.tags = "key".into();
+            let a = mem.save_or_update(first).unwrap();
+            mem.save(make_entry("other", &[0.0, 0.0, 1.0, 0.0]))
+                .unwrap();
+            let b = mem.save_or_update(second).unwrap();
+            assert_eq!(a, b);
+            assert_eq!(mem.cache.chunks, ["v2", "other"]);
+            // Dropped without a checkpoint: all three records live in the WAL.
+        }
+        let mem = HDF5Memory::open(&h5_path).unwrap();
+        assert_eq!(mem.cache.chunks, ["v2", "other"]);
+        assert_eq!(mem.cache.embeddings[0], [0.0, 1.0, 0.0, 0.0]);
+    }
+
+    #[test]
+    fn v3_wal_is_read_and_upgraded_in_place() {
+        let dir = TempDir::new().unwrap();
+        let wal_path = dir.path().join("old.wal");
+        {
+            let mut wal = WalFile::open(&wal_path).unwrap();
+            wal.append_save(&make_wal_entry("kept", &[1.0])).unwrap();
+        }
+        // Rewrite the header as the pre-Update chained format.
+        let mut bytes = std::fs::read(&wal_path).unwrap();
+        bytes[4] = WAL_VERSION_CHAINED_NO_UPDATE;
+        std::fs::write(&wal_path, &bytes).unwrap();
+
+        assert_eq!(WalFile::read_entries(&wal_path).unwrap().len(), 1);
+        {
+            let mut wal = WalFile::open(&wal_path).unwrap();
+            assert_eq!(wal.pending_count(), 1);
+            wal.append_save(&make_wal_entry("new", &[2.0])).unwrap();
+        }
+        assert_eq!(std::fs::read(&wal_path).unwrap()[4], WAL_VERSION);
+        let chunks: Vec<_> = WalFile::read_entries(&wal_path)
+            .unwrap()
+            .into_iter()
+            .map(|e| e.chunk)
+            .collect();
+        assert_eq!(chunks, ["kept", "new"]);
+    }
+
     #[test]
     fn mark_matching_is_exact() {
         let dir = TempDir::new().unwrap();