@@ -247,11 +247,16 @@ policy allows it: the memory store standalone evicts any unpinned node, the
 disk store evicts any unpinned node, and the composite restricts the memory
 store to durable nodes.
 
-Each tier drops a non-root record's row once its only child commits, keeping
-the pages; readers and spills delay removal until they release the row. DRAM
-releases row slots, disk truncates at the row segment, and later spills omit it.
-A rewind or later fork restores the nearest row-bearing ancestor and re-prefills
-the remaining tokens.
+Disk drops a non-root record's row once its only child commits, keeping the
+pages. DRAM retains these redundant rows while its free retention space exceeds
+two continuation rows. Reserved transfer intake does not count as retention
+headroom. When that threshold is reached, or an incoming record needs more
+space than is free, DRAM reclaims eligible redundant rows before evicting other
+records. Readers and spills delay row removal until their pins are released.
+DRAM releases row slots, disk truncates at the row segment, and later spills
+omit removed rows. A rewind or later fork restores its retained endpoint when
+available; otherwise it restores the nearest row-bearing ancestor and
+re-prefills the remaining tokens.
 With lazy write-behind, a crash or an incomplete shutdown drain can lose the
 DRAM-only descendants of a page-only disk prefix, leaving recovery to an older
 row-bearing ancestor. Startup prunes childless page-only records and any
@@ -393,11 +398,18 @@ the storage namespace identity, the layout fingerprint, the key (consumed count
 digest), the parent key, the bucket digest, the page token count, the tail
 page count, the rank count, the per-rank plane sizes and row bytes, the page
 segment offset, the row segment offset, the row checksum, and the creation
-time. Version 2 uses flag bit 31 for a replicated page segment and bit 30 for a
-replicated row. Page offsets use one fixed stride for the whole record. Version-1
-records still recover, and the layout fingerprint is unchanged. An older binary
-treats replicated version-2 files as invalid and **deletes them during recovery**;
-downgrading therefore starts those records cold. A checksum segment follows,
+time. Versions 1 and 2 encode each plane size as `u64`; version 3 uses `u32`
+plane sizes so wider rank/plane catalogs fit the same header block. Row sizes
+remain `u64`. New records use version 3 when every plane fits `u32`, otherwise
+the `u64` format is retained. All encodings still require the header to fit 4 KiB.
+Versions 2 and 3 use flag bit 31 for a replicated page segment and bit 30 for a
+replicated row; version 3 also permits non-replicated records. Page offsets use
+one fixed stride for the whole record. Version-1 and version-2 records still
+recover and preserve their encoding on rewrite; cross-rank agreement compares
+canonical header fields independently of the file's encoding version. The
+layout fingerprint is unchanged. A binary that does not support a record's
+version treats it as invalid and **deletes it during recovery**; downgrading
+therefore starts those records cold. A checksum segment follows,
 one 64-bit xxh3 over each stored page record, padded to
 4 KiB, so a restore that reads only a suffix of the pages verifies each page
 it read. The page segment follows, page-major: page `i`, then rank `r`, then
 
@@ -2224,6 +2224,12 @@ impl<'a> Worker<'a> {
                 StoreEvent::Stored {
                     checkpoint, result, ..
                 } => {
+                    if let Err(error) = &result {
+                        eprintln!(
+                            "HOST_CACHE_STORE_ERROR checkpoint={} error={error}",
+                            checkpoint.value()
+                        );
+                    }
                     self.metrics.host_retry_abandoned[5]
                         .fetch_add(u64::from(result.is_err()), Ordering::Relaxed);
                     self.scheduler.unpin_checkpoint(checkpoint)?;
@@ -2304,6 +2310,12 @@ impl<'a> Worker<'a> {
         for publication in publications {
             progress = true;
             let outcome = store.store(&publication);
+            if let Err(error) = &outcome {
+                eprintln!(
+                    "HOST_CACHE_STORE_SUBMIT_ERROR checkpoint={} error={error}",
+                    publication.id.value()
+                );
+            }
             if let Ok(outcome) = outcome {
                 self.metrics.host_store_outcome(outcome);
                 match outcome {
 
@@ -396,6 +396,14 @@ impl Core {
         if !self.fits(needed) {
             return Ok(None);
         }
+        if self
+            .pool
+            .free_slots()
+            .saturating_sub(self.pool.intake_slots())
+            < needed
+        {
+            self.thin_rows_for(needed);
+        }
         if self.spill && self.retire_for(self.shortfall(needed)) {
             return Ok(None);
         }
@@ -408,6 +416,7 @@ impl Core {
                 Some(Eviction::Leaves(nodes)) => {
                     self.stats.dropped[0] += if self.flat { 0 } else { nodes.len() as u64 };
                     for node in &nodes {
+                        self.thinning.remove(&node.key);
                         self.clean_slots -= self.clean_slots(node);
                         self.release(&node.location);
                     }
@@ -674,6 +683,7 @@ impl Core {
         if !node.ready || node.pinned() {
             return false;
         }
+        self.thinning.remove(key);
         let clean = self.clean_slots(node);
         self.index.drop_row(key);
         let record = &mut self.index.node_mut(key).unwrap().location;
@@ -696,6 +706,20 @@ impl Core {
     }
 
     fn thin_rows(&mut self) {
+        self.thin_rows_for(self.layout.row_slots().saturating_mul(2));
+    }
+
+    fn thin_rows_for(&mut self, headroom: usize) {
+        // Intake slots belong to transfers, not checkpoint retention. Reclaim
+        // redundant rows before an incoming record evicts reusable endpoints.
+        if self
+            .pool
+            .free_slots()
+            .saturating_sub(self.pool.intake_slots())
+            > headroom
+        {
+            return;
+        }
         for key in std::mem::take(&mut self.thinning) {
             if !self.index.row_redundant(&key, false) {
                 continue;
@@ -728,6 +752,7 @@ impl Core {
 
     fn remove_record(&mut self, key: &PrefixKey) -> Result<()> {
         for node in self.index.remove(key)? {
+            self.thinning.remove(&node.key);
             self.clean_slots -= self.clean_slots(&node);
             self.release(&node.location);
         }
@@ -1816,7 +1841,7 @@ mod tests {
     }
 
     #[test]
-    fn conversation_rows_thin_to_root_and_tip_and_restore_through_page_carriers() {
+    fn conversation_rows_survive_with_headroom_and_restore_at_each_endpoint() {
         let geometry = [geometry(PT, &[("kv", PAGE)], ROW)];
         let (copier, gpu) = crate::testing::SimulatedGpu::shared_mapped(&geometry).unwrap();
         let mut store = MemoryStore::new(IDENTITY, host_pool(32), copier).unwrap();
@@ -1842,20 +1867,17 @@ mod tests {
             store.poll(&mut events).unwrap();
             assert_eq!(
                 store.core.index.nodes().filter(|n| n.row).count(),
-                (turn as usize).min(2)
+                turn as usize
             );
             assert_clean_slots(&store.core);
         }
-        assert_eq!(store.pool().slot_count() - store.pool().free_slots(), 10);
-        assert_eq!(store.stats().rows_thinned, [6, 0]);
-        for (tokens, expected) in [(8 * PT, 8 * PT), (7 * PT, PT)] {
+        assert_eq!(store.pool().slot_count() - store.pool().free_slots(), 16);
+        assert_eq!(store.stats().rows_thinned, [0, 0]);
+        for (tokens, expected) in [(8 * PT, 8 * PT), (7 * PT, 7 * PT)] {
             let hit = store.lookup(&history(tokens, 1), 0, u64::MAX).unwrap();
             assert_eq!(hit.consumed_tokens, expected);
-            // A rewind resumes at the root and re-prefills the remaining tokens.
-            assert_eq!(
-                tokens - hit.consumed_tokens,
-                if tokens == 7 * PT { 6 * PT } else { 0 }
-            );
+            // Every endpoint resumes directly at its retained row.
+            assert_eq!(tokens - hit.consumed_tokens, 0);
             store
                 .fetch(
                     hit.clone(),
@@ -1887,7 +1909,7 @@ mod tests {
     }
 
     #[test]
-    fn thinning_waits_for_commit_and_pins_and_preserves_existing_branches() {
+    fn rows_survive_commit_and_pins_and_existing_branches() {
         let layout = Layout::new(&[geometry(PT, &[("kv", PAGE)], 3 * PAGE)]).unwrap();
         let mut core = Core::new(IDENTITY, host_pool(64), layout, false);
         let full = history(320, 1);
@@ -1910,19 +1932,19 @@ mod tests {
         let before = core.pool.free_slots();
         core.unpin_record(&parent.key).unwrap();
         core.settle(&mut Vec::new());
-        assert_eq!(core.pool.free_slots(), before + 3);
-        assert!(core.record(&parent.key).unwrap().row.is_none());
+        assert_eq!(core.pool.free_slots(), before);
+        assert!(core.record(&parent.key).unwrap().row.is_some());
         assert_eq!(core.record(&parent.key).unwrap().pages.len(), 1);
         assert_clean_slots(&core);
 
-        // A late fork preserves the page carrier without reconstructing its lost row.
+        // A late fork retains both the page carrier and its parent's row.
         let mut branch = history(128, 1);
         branch.append(&[99; 128], &[]).unwrap();
         let sibling = published(&branch, 192, Some(128), 10, 4);
         core.store(&sibling).unwrap();
         finish(&mut core);
         assert_eq!(core.index[&parent.key].children(), 2);
-        assert!(!core.index[&parent.key].row);
+        assert!(core.index[&parent.key].row);
         assert!(core.index[&child.key].row && core.index[&sibling.key].row);
 
         // A branch formed while its row is pinned retains that row and both tips.
@@ -1944,11 +1966,130 @@ mod tests {
         finish(&mut core);
         assert_eq!(
             core.lookup(&history(256, 1), 0, u64::MAX).unwrap().key,
-            child.key
+            leaf.key
         );
         assert_clean_slots(&core);
     }
 
+    #[test]
+    fn row_thinning_uses_warm_headroom_for_each_row_geometry() {
+        for row_slots in [1, 3] {
+            for headroom in [2 * row_slots, 2 * row_slots + 1] {
+                let layout =
+                    Layout::new(&[geometry(PT, &[("kv", PAGE)], row_slots as u64 * PAGE)]).unwrap();
+                assert_eq!(layout.row_slots(), row_slots);
+                let intake = crate::offload::intake_slots(&layout, 1);
+                let warm = 3 * (1 + row_slots) + headroom;
+                let pool = SlabPool::new(PAGE, warm + intake, intake).unwrap();
+                let mut core = Core::new(IDENTITY, pool, layout, false);
+                let full = history(192, 1);
+                let checkpoints = [
+                    published(&full, 64, None, 0, 1),
+                    published(&full, 128, Some(64), 0, 2),
+                    published(&full, 192, Some(128), 0, 3),
+                ];
+                for checkpoint in &checkpoints {
+                    assert!(core.store(checkpoint).unwrap());
+                    finish(&mut core);
+                }
+                let thinned = headroom == 2 * row_slots;
+                assert_eq!(
+                    core.index[&checkpoints[1].key].row, !thinned,
+                    "row_slots={row_slots}, warm headroom={headroom}, intake={intake}"
+                );
+                assert_eq!(
+                    core.pool.free_slots(),
+                    intake + headroom + usize::from(thinned) * row_slots
+                );
+                assert!(core.index[&checkpoints[0].key].row);
+                assert!(core.index[&checkpoints[2].key].row);
+                assert_clean_slots(&core);
+            }
+        }
+    }
+
+    #[test]
+    fn pressure_thins_redundant_rows_before_allocating_another_endpoint() {
+        for (warm_slots, next_endpoint) in [(7, 256), (9, 384)] {
+            let mut core = store_core(warm_slots - 1, 1, false);
+            let full = history(next_endpoint, 1);
+            let root = published(&full, 64, None, 0, 1);
+            let middle = published(&full, 128, Some(64), 0, 2);
+            let child = published(&full, 192, Some(128), 0, 3);
+            let tip = published(&full, next_endpoint, Some(192), 0, 4);
+            for checkpoint in [&root, &middle, &child, &tip] {
+                assert!(core.store(checkpoint).unwrap());
+                finish(&mut core);
+            }
+            assert!(core.index[&root.key].row, "warm slots={warm_slots}");
+            assert!(!core.index[&middle.key].row);
+            assert_eq!(
+                core.lookup(&history(64, 1), 0, u64::MAX).map(|hit| hit.key),
+                Some(root.key)
+            );
+            assert!(core.pool.free_slots() >= core.pool.intake_slots());
+            assert_clean_slots(&core);
+        }
+    }
+
+    #[test]
+    fn pressure_thinning_waits_for_the_last_row_pin() {
+        let mut core = store_core(6, 1, false);
+        let full = history(192, 1);
+        let root = published(&full, 64, None, 0, 1);
+        let parent = published(&full, 128, Some(64), 0, 2);
+        let child = published(&full, 192, Some(128), 0, 3);
+        for checkpoint in [&root, &parent] {
+            assert!(core.store(checkpoint).unwrap());
+            finish(&mut core);
+        }
+        core.pin_record(&parent.key).unwrap();
+        assert!(core.store(&child).unwrap());
+        finish(&mut core);
+        assert!(core.index[&parent.key].row);
+        assert!(core.thinning.contains(&parent.key));
+        let before = core.pool.free_slots();
+        core.unpin_record(&parent.key).unwrap();
+        core.settle(&mut Vec::new());
+        assert!(!core.index[&parent.key].row);
+        assert_eq!(core.pool.free_slots(), before + core.layout.row_slots());
+        assert!(!core.thinning.contains(&parent.key));
+        assert_clean_slots(&core);
+    }
+
+    #[test]
+    fn thinning_queue_keeps_only_resident_records_after_removal_and_eviction() {
+        for remove_explicitly in [false, true] {
+            let mut core = store_core(7, 1, false);
+            for salt in 1..=64 {
+                let full = history(192, salt);
+                let checkpoints = [
+                    published(&full, 64, None, 0, 1),
+                    published(&full, 128, Some(64), 0, 2),
+                    published(&full, 192, Some(128), 0, 3),
+                ];
+                for checkpoint in &checkpoints {
+                    assert!(core.store(checkpoint).unwrap());
+                    finish(&mut core);
+                }
+                if remove_explicitly {
+                    // Expiration and explicit removal share this record-removal path.
+                    for checkpoint in checkpoints.iter().rev() {
+                        if core.index.contains(&checkpoint.key) {
+                            core.remove_record(&checkpoint.key).unwrap();
+                        }
+                    }
+                }
+                assert!(
+                    core.thinning.iter().all(|key| core.index.contains(key)),
+                    "stale thinning keys after chain {salt}, explicit removal={remove_explicitly}"
+                );
+                assert!(core.thinning.len() <= core.index.len());
+                assert_clean_slots(&core);
+            }
+        }
+    }
+
     #[test]
     fn a_record_one_slot_larger_than_warm_is_oversize() {
         let geometry = [geometry(PT, &[("kv", PAGE)], ROW)];
@@ -2773,8 +2914,8 @@ mod tests {
             stored(&finish(&mut core)),
             [(grandchild.key, Some(3 * PAGE + ROW))]
         );
-        // Commit thins the child: nine page slots and the tip's one row slot
-        // occupy ten of twelve warm slots. The root's row was already freed.
+        // With only one warm slot free, commit thins the redundant child row.
+        // Its pages remain available and the grandchild retains its endpoint.
         assert!(core.record(&child.key).unwrap().row.is_none());
         assert!(core.record(&grandchild.key).unwrap().row.is_some());
         assert_eq!(core.pool.free_slots(), core.pool.intake_slots() + 2);
 
@@ -1201,7 +1201,7 @@ fn shape(header: &Header) -> Result<[u8; 16]> {
     header.fingerprint = [0; 16];
     header.row_checksum = 0;
     let mut hash = Blake2b::<U16>::new();
-    hash.update(header.encode()?);
+    hash.update(header.canonical_encoding()?);
     Ok(hash.finalize().into())
 }
 
 
@@ -512,6 +512,42 @@ use std::net::TcpListener;
         assert!(agree(&zero, &peers).unwrap().is_empty());
     }
 
+    #[test]
+    fn agreement_compares_legacy_and_v3_headers_by_geometry() {
+        for (version, replication) in [
+            (1u32, record::Replication::default()),
+            (2u32, record::Replication { pages: true, row: true }),
+        ] {
+            let mut compact = scanned(key(128, 1), None, true, 0);
+            compact.header.set_replication(replication).unwrap();
+            let mut bytes = compact.header.encode().unwrap();
+            assert_eq!(u32::from_le_bytes(bytes[8..12].try_into().unwrap()), 3);
+            bytes[8..12].copy_from_slice(&version.to_le_bytes());
+            // This fixture has one rank and one plane. Legacy entries retain
+            // an eight-byte plane size followed by the eight-byte row size.
+            bytes[181..189].copy_from_slice(&compact.header.ranks[0].planes[0].to_le_bytes());
+            bytes[189..197].copy_from_slice(&compact.header.ranks[0].row_bytes.to_le_bytes());
+            bytes[197..].fill(0);
+            let mut legacy = compact.clone();
+            legacy.header = Header::decode(&bytes).unwrap();
+            assert_eq!(legacy.header, compact.header);
+            assert_eq!(legacy.header.encode().unwrap(), bytes);
+
+            for (local, peer) in [(&compact, &legacy), (&legacy, &compact)] {
+                let peers = [decode_summary(&summary(std::slice::from_ref(peer)).unwrap()[5..]).unwrap()];
+                assert_eq!(
+                    agree(std::slice::from_ref(local), &peers).unwrap(),
+                    HashMap::from([(local.header.key, true)]),
+                );
+            }
+
+            let mut different = legacy;
+            different.header.ranks[0].row_bytes += 4096;
+            let peers = [decode_summary(&summary(&[different]).unwrap()[5..]).unwrap()];
+            assert!(agree(&[compact], &peers).unwrap().is_empty());
+        }
+    }
+
     #[test]
     fn events_and_peer_file_names_round_trip_and_reject_paths() {
         let event = TransportEvent::StoreInvalidated {
Original file line number	Diff line number	Diff line change
`@@ -1201,7 +1201,7 @@ fn shape(header: &Header) -> Result<[u8; 16]> {`
`1201`	`1201`	`header.fingerprint = [0; 16];`
`1202`	`1202`	`header.row_checksum = 0;`
`1203`	`1203`	`let mut hash = Blake2b::<U16>::new();`
`1204`		`- hash.update(header.encode()?);`
	`1204`	`+ hash.update(header.canonical_encoding()?);`
`1205`	`1205`	`Ok(hash.finalize().into())`
`1206`	`1206`	`}`
`1207`	`1207`