diff --git a/crates/didbot-pds/tests/record_heap.rs b/crates/didbot-pds/tests/record_heap.rs
new file mode 100644
index 00000000..22717541
--- /dev/null
+++ b/crates/didbot-pds/tests/record_heap.rs
@@ -0,0 +1,510 @@
+//! What a record's body being on disk rather than in memory changes.
+//!
+//! A record write is now two appends and not one: the body goes into
+//! `records/pds.heap` as a framed block, and the journal gets a slot naming
+//! it. The order of those two is the whole of the crash argument — **bytes
+//! reach the medium before the fact that names them does** — and these are
+//! the four states a power cut between them can leave, driven by cutting the
+//! two files where a crash would have.
+//!
+//! Every fixture is small on purpose. What is under test is which of the two
+//! files is ahead of the other, never how much either holds.
+
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+use didbot_dns::LoopbackDns;
+use didbot_identity::Zone;
+use didbot_pds::{
+    Durable, FileAccountStore, ListParams, Precondition, ProvisionRequest, Provisioner, Registry,
+    Swap,
+};
+use serde_json::json;
+use time::Duration;
+
+const ZONE_HOST: &str = "agents.localhost";
+const THING: &str = "com.example.thing";
+
+fn scratch(name: &str) -> PathBuf {
+    let dir = std::env::temp_dir().join(format!("didbot-heap-{name}-{}", std::process::id()));
+    let _ = std::fs::remove_dir_all(&dir);
+    dir
+}
+
+fn zone() -> Zone {
+    Zone::delegated("localhost", ZONE_HOST).expect("test zone should be constructible")
+}
+
+type Pds = Provisioner<LoopbackDns, Arc<FileAccountStore>>;
+
+fn boot(dir: &Path) -> (Durable, Pds) {
+    let durable = Durable::open(dir, Duration::days(30)).expect("the directory should open");
+    let pds = Provisioner::new(
+        "did:web:owner.example",
+        zone(),
+        "http://localhost:3600".to_string(),
+        LoopbackDns::new(),
+        durable.accounts(),
+    )
+    .with_record_store(durable.records())
+    .with_blob_store(durable.blobs())
+    .with_commit_history(durable.history());
+    (durable, pds)
+}
+
+fn heap_path(dir: &Path) -> PathBuf {
+    dir.join(didbot_pds::wal::RECORD_DIR)
+        .join(didbot_pds::wal::HEAP_FILE)
+}
+
+fn write(pds: &Pds, did: &str, text: &str) -> (String, String) {
+    let written = pds
+        .put_record(
+            did,
+            THING,
+            None,
+            json!({"text": text, "emoji": "\u{1f9ff}", "createdAt": "2026-01-01T00:00:00Z"}),
+            &Swap::default(),
+        )
+        .expect("the record should write");
+    (written.rkey, written.cid.to_string())
+}
+
+/// Every record key the repository serves, in key order.
+fn keys(pds: &Pds, did: &str) -> Vec<String> {
+    let mut out: Vec<String> = pds
+        .list_records(did, THING, &ListParams::new(50))
+        .expect("a listing")
+        .into_iter()
+        .map(|(rkey, _)| rkey)
+        .collect();
+    out.sort();
+    out
+}
+
+/// The text a record holds, for a read that has to prove it got the value and
+/// not merely a key.
+fn text(pds: &Pds, did: &str, rkey: &str) -> Option<String> {
+    pds.get_record(did, THING, rkey)
+        .ok()
+        .flatten()
+        .and_then(|record| record["text"].as_str().map(str::to_owned))
+}
+
+/// A deployment with one agent and three records. Returns the directory's
+/// DID and the three keys.
+fn seeded(dir: &Path) -> (String, Vec<(String, String)>) {
+    let (durable, pds) = boot(dir);
+    let did = pds
+        .provision(ProvisionRequest::new("kestrel", None))
+        .expect("provision")
+        .account
+        .did
+        .as_str()
+        .to_owned();
+    let written = vec![
+        write(&pds, &did, "quernstone"),
+        write(&pds, &did, "marlpit"),
+        write(&pds, &did, "sillion"),
+    ];
+    durable.wal().sync().expect("flush the journal");
+    drop(durable);
+    (did, written)
+}
+
+/// The frame boundaries in a file this module's framing wrote.
+///
+/// Read out of the headers rather than computed, so the walk stops at the
+/// first frame that is not whole instead of running off the end.
+fn frame_ends(path: &Path) -> Vec<u64> {
+    let bytes = std::fs::read(path).expect("the file reads");
+    let mut ends = Vec::new();
+    let mut at = 0usize;
+    while at + 8 <= bytes.len() {
+        let len = u32::from_le_bytes(bytes[at + 4..at + 8].try_into().expect("four")) as usize;
+        let end = at + 8 + len;
+        if end > bytes.len() {
+            break;
+        }
+        ends.push(end as u64);
+        at = end;
+    }
+    ends
+}
+
+/// Flips a bit inside the frame whose payload holds `needle`.
+///
+/// Located by content rather than by offset: the heap holds the records the
+/// server authors as well as the ones a test writes, so "the first frame" is
+/// not the first record a test can name.
+fn damage_frame_holding(path: &Path, needle: &str) {
+    let mut bytes = std::fs::read(path).expect("the heap reads");
+    let at = bytes
+        .windows(needle.len())
+        .position(|window| window == needle.as_bytes())
+        .unwrap_or_else(|| panic!("the heap holds no body carrying {needle:?}"));
+    bytes[at] ^= 0b0010_0000;
+    std::fs::write(path, &bytes).expect("damage it");
+}
+
+fn truncate(path: &Path, to: u64) {
+    std::fs::OpenOptions::new()
+        .write(true)
+        .open(path)
+        .expect("open the file")
+        .set_len(to)
+        .expect("truncate it");
+}
+
+// ---------------------------------------------------------------------------
+// The crash between the two appends, in both directions
+// ---------------------------------------------------------------------------
+
+/// The body reached the medium and the fact naming it did not.
+///
+/// The harmless direction, and the one the ordering is chosen to leave: the
+/// journal is cut back to before the last record's slot while the heap still
+/// holds that record's body. The deployment opens, serves the records the
+/// journal does name, and the body nothing names is reclaimed — which is the
+/// same posture `.incoming/` has in the blob store.
+#[test]
+fn a_body_the_journal_never_named_is_reclaimed_and_costs_nothing_else() {
+    let dir = scratch("body-ahead");
+    let (did, written) = seeded(&dir);
+
+    let log = dir.join(didbot_pds::wal::LOG_FILE);
+    let ends = frame_ends(&log);
+    // One frame back: the last thing written is the commit over the third
+    // record, and the frame before it is that record's slot.
+    let cut = ends[ends.len() - 3];
+    truncate(&log, cut);
+    let heap_before = std::fs::metadata(heap_path(&dir))
+        .expect("the heap is there")
+        .len();
+
+    let (durable, pds) = boot(&dir);
+    assert_eq!(
+        keys(&pds, &did),
+        vec![written[0].0.clone(), written[1].0.clone()],
+        "the records the journal still names came back, and the one it lost did not"
+    );
+    assert_eq!(
+        text(&pds, &did, &written[0].0).as_deref(),
+        Some("quernstone")
+    );
+    assert_eq!(text(&pds, &did, &written[1].0).as_deref(), Some("marlpit"));
+
+    let heap_after = std::fs::metadata(heap_path(&dir))
+        .expect("the heap is there")
+        .len();
+    assert!(
+        heap_after < heap_before,
+        "the body no journal entry names is still in the heap: {heap_before} bytes became \
+         {heap_after}"
+    );
+    // And it stays reclaimed: a second boot finds nothing left to sweep.
+    drop(durable);
+    let (durable, _) = boot(&dir);
+    assert_eq!(
+        std::fs::metadata(heap_path(&dir))
+            .expect("the heap is there")
+            .len(),
+        heap_after,
+        "the reclaim is not a fixed point"
+    );
+
+    drop(durable);
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
+/// The fact reached the medium and the body naming it did not.
+///
+/// The direction the ordering exists to make rare, driven directly by cutting
+/// the heap back while the journal still names what was cut. The slot lies
+/// past the end of the heap, which is the torn tail: the record is dropped
+/// with a line naming it, the deployment opens, and every record whose body
+/// is still there serves. Nothing is served *from* the missing body, which is
+/// the failure this is really about.
+#[test]
+fn a_slot_past_the_end_of_the_heap_drops_its_record_and_serves_the_rest() {
+    let dir = scratch("fact-ahead");
+    let (did, written) = seeded(&dir);
+
+    let heap = heap_path(&dir);
+    let ends = frame_ends(&heap);
+    // Back to the end of the second record's body, so the third record's
+    // slot names bytes that are not there.
+    truncate(&heap, ends[ends.len() - 2]);
+
+    let (durable, pds) = boot(&dir);
+    assert_eq!(
+        keys(&pds, &did),
+        vec![written[0].0.clone(), written[1].0.clone()],
+        "the record whose body never landed was not dropped"
+    );
+    assert_eq!(
+        text(&pds, &did, &written[0].0).as_deref(),
+        Some("quernstone")
+    );
+    assert_eq!(text(&pds, &did, &written[1].0).as_deref(), Some("marlpit"));
+    assert_eq!(
+        text(&pds, &did, &written[2].0),
+        None,
+        "a record whose body is gone was served anyway"
+    );
+    // The repository still signs, which is what makes the dropped record a
+    // loss rather than an inconsistency.
+    pds.export_repo(did.as_str(), None)
+        .expect("the repository should still export");
+
+    drop(durable);
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
+/// A heap cut one byte into the last body is the same answer.
+///
+/// The clean cut above is the easy case: the slot is wholly past the end.
+/// This one leaves the frame's header and part of its payload, so the slot is
+/// *partly* inside the heap — and the rule is a bound on the whole frame,
+/// which is what makes a half-arrived body indistinguishable from an absent
+/// one rather than something to guess about.
+#[test]
+fn a_body_that_half_arrived_is_the_same_as_one_that_did_not() {
+    let dir = scratch("half-body");
+    let (did, written) = seeded(&dir);
+
+    let heap = heap_path(&dir);
+    let ends = frame_ends(&heap);
+    truncate(&heap, ends[ends.len() - 1] - 1);
+
+    let (durable, pds) = boot(&dir);
+    assert_eq!(
+        keys(&pds, &did),
+        vec![written[0].0.clone(), written[1].0.clone()]
+    );
+    assert_eq!(text(&pds, &did, &written[2].0), None);
+
+    drop(durable);
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
+/// Bodies the journal never named, written by nothing, go at the next boot.
+///
+/// The append that landed and whose entry did not, in its purest form: bytes
+/// on the end of the heap that no slot reaches. They are dead space and the
+/// boot is where it is swept, so a deployment that crashes mid-write a
+/// thousand times does not carry a thousand orphaned bodies forever.
+#[test]
+fn heap_bytes_no_entry_names_are_swept_at_the_next_boot() {
+    let dir = scratch("orphan-bodies");
+    let (did, written) = seeded(&dir);
+
+    let heap = heap_path(&dir);
+    let named = std::fs::metadata(&heap).expect("the heap is there").len();
+    // A frame nothing points at, appended the way an interrupted write would
+    // have left it.
+    {
+        use std::io::Write as _;
+        let mut file = std::fs::OpenOptions::new()
+            .append(true)
+            .open(&heap)
+            .expect("open the heap");
+        // Header and payload, the framing this file is written in; the
+        // checksum does not matter because nothing will ever read it.
+        file.write_all(&[0u8; 8]).expect("a header");
+        file.write_all(b"farthingale").expect("a payload");
+    }
+    assert!(std::fs::metadata(&heap).expect("the heap").len() > named);
+
+    let (durable, pds) = boot(&dir);
+    assert_eq!(
+        std::fs::metadata(&heap).expect("the heap").len(),
+        named,
+        "the bytes no slot names were kept"
+    );
+    assert_eq!(
+        keys(&pds, &did).len(),
+        written.len(),
+        "and nothing else went"
+    );
+
+    drop(durable);
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
+/// A body inside the heap that does not check out is refused, not served.
+///
+/// The middle of the file rather than its tail, which is the case
+/// `pds.wal`'s replay says plainly cannot be told apart from a torn write by
+/// guessing. The frame's CRC is what answers it: the read that wanted those
+/// bytes is refused, the record reads as absent with a line saying why, and
+/// every other record in the repository is untouched. What must not happen is
+/// the bytes being handed out as a record.
+#[test]
+fn a_damaged_body_is_refused_rather_than_served_as_a_record() {
+    let dir = scratch("damaged-body");
+    let (did, written) = seeded(&dir);
+
+    // Inside the first record's own frame, which is in the middle of the
+    // heap by construction: two more bodies were appended after it.
+    damage_frame_holding(&heap_path(&dir), "quernstone");
+
+    let (durable, pds) = boot(&dir);
+    assert_eq!(
+        text(&pds, &did, &written[0].0),
+        None,
+        "a frame that does not check out was served as a record"
+    );
+    assert_eq!(text(&pds, &did, &written[1].0).as_deref(), Some("marlpit"));
+    assert_eq!(text(&pds, &did, &written[2].0).as_deref(), Some("sillion"));
+
+    drop(durable);
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
+// ---------------------------------------------------------------------------
+// What the slot answers on its own
+// ---------------------------------------------------------------------------
+
+/// A compare-and-swap is decided by the slot, with the body never read.
+///
+/// `Entry::RecordStored` carries the content identifier so that a
+/// precondition is a comparison against the slot rather than a read of the
+/// body and a rehash of it. The way to prove the body is not read is to make
+/// reading it impossible: the record's frame is damaged, so any read of it is
+/// refused — and the swap is still decided correctly in both directions.
+#[test]
+fn a_precondition_is_answered_from_the_slot_with_the_body_unreadable() {
+    let dir = scratch("swap-from-slot");
+    let (did, written) = seeded(&dir);
+    let (rkey, cid) = written[0].clone();
+
+    damage_frame_holding(&heap_path(&dir), "quernstone");
+
+    let (durable, pds) = boot(&dir);
+    // The body really is unreadable, so nothing below can have read it.
+    assert_eq!(text(&pds, &did, &rkey), None);
+
+    // A swap naming the wrong version is refused, and it is refused with the
+    // version the slot holds rather than with "no record".
+    let wrong = pds.put_record(
+        did.as_str(),
+        THING,
+        Some(&rkey),
+        json!({"text": "weftling", "emoji": "\u{1f9ff}", "createdAt": "2026-01-01T00:00:00Z"}),
+        &Swap {
+            record: Precondition::Version(
+                didbot_data::Cid::parse(&written[1].1).expect("a content identifier"),
+            ),
+            commit: None,
+        },
+    );
+    match wrong {
+        Err(didbot_pds::ProvisionError::Record(didbot_pds::RecordError::SwapFailed {
+            found,
+            ..
+        })) => assert_eq!(found, cid, "the swap was decided against the wrong version"),
+        other => panic!("a swap naming the wrong version was not refused: {other:?}"),
+    }
+
+    // And the one naming the version the slot holds goes through.
+    pds.put_record(
+        did.as_str(),
+        THING,
+        Some(&rkey),
+        json!({"text": "weftling", "emoji": "\u{1f9ff}", "createdAt": "2026-01-01T00:00:00Z"}),
+        &Swap {
+            record: Precondition::Version(
+                didbot_data::Cid::parse(&cid).expect("a content identifier"),
+            ),
+            commit: None,
+        },
+    )
+    .expect("a swap naming the version the slot holds should be accepted");
+    // The replacement is readable, because its body is a frame that landed.
+    assert_eq!(text(&pds, &did, &rkey).as_deref(), Some("weftling"));
+
+    drop(durable);
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
+// ---------------------------------------------------------------------------
+// The whole drill, over the second file
+// ---------------------------------------------------------------------------
+
+/// Every cut a crash could leave in the heap opens or is refused.
+///
+/// `tests/restore.rs` walks this over `pds.wal`; this is the same walk over
+/// the file the bodies are in. Cut at each frame boundary and one byte before
+/// it — a clean cut and a torn one — and require every single one to open and
+/// hold together: the records whose bodies survived serve their own values,
+/// no record serves anything else's, and the repository still signs.
+#[test]
+fn every_truncation_of_the_heap_opens_to_something_self_consistent() {
+    let dir = scratch("heap-drill");
+    let copy = scratch("heap-drill-copy");
+    let (did, written) = seeded(&dir);
+
+    let ends = frame_ends(&heap_path(&dir));
+    assert!(
+        ends.len() >= 4,
+        "the fixture should produce a real spread of cuts, not {}",
+        ends.len()
+    );
+    let bytes = std::fs::read(heap_path(&dir)).expect("the heap reads");
+    let mut cuts: Vec<u64> = Vec::new();
+    for end in &ends {
+        cuts.push(*end);
+        cuts.push(end.saturating_sub(1));
+    }
+
+    let mut kept_all = 0usize;
+    for cut in cuts {
+        let _ = std::fs::remove_dir_all(&copy);
+        copy_tree(&dir, &copy);
+        std::fs::write(heap_path(&copy), &bytes[..cut as usize]).expect("cut the heap");
+
+        let at = format!("a heap cut at {cut} of {}", bytes.len());
+        let (durable, pds) = boot(&copy);
+        // Whatever survived, every record that reads back reads back as its
+        // own value. A slot resolved against the wrong frame would show up
+        // here as a key holding somebody else's text.
+        let served = keys(&pds, &did);
+        for (rkey, expected) in [
+            (&written[0].0, "quernstone"),
+            (&written[1].0, "marlpit"),
+            (&written[2].0, "sillion"),
+        ] {
+            if let Some(found) = text(&pds, &did, rkey) {
+                assert_eq!(found, expected, "{at}: {rkey} came back as another record");
+            }
+        }
+        pds.export_repo(did.as_str(), None)
+            .unwrap_or_else(|error| panic!("{at}: the repository will not sign: {error}"));
+        if served.len() == written.len() {
+            kept_all += 1;
+        }
+        drop(durable);
+    }
+    assert!(
+        kept_all > 0,
+        "every cut lost a record, so the drill never exercised keeping them all"
+    );
+
+    let _ = std::fs::remove_dir_all(&dir);
+    let _ = std::fs::remove_dir_all(&copy);
+}
+
+/// A recursive copy, which is what a block snapshot hands back.
+fn copy_tree(from: &Path, into: &Path) {
+    std::fs::create_dir_all(into).expect("the copy's directory");
+    for entry in std::fs::read_dir(from).expect("the source").flatten() {
+        let path = entry.path();
+        let target = into.join(entry.file_name());
+        if path.is_dir() {
+            copy_tree(&path, &target);
+        } else {
+            std::fs::copy(&path, &target).expect("copy a file");
+        }
+    }
+}
diff --git a/crates/didbot-pds/tests/repo_cost.rs b/crates/didbot-pds/tests/repo_cost.rs
index 32e020c6..0609a628 100644
--- a/crates/didbot-pds/tests/repo_cost.rs
+++ b/crates/didbot-pds/tests/repo_cost.rs
@@ -457,3 +457,76 @@ fn rebuild_cost_by_repository_size() {
         );
     }
 }
+
+/// What a write costs on a deployment whose records are on disk.
+///
+/// The other benchmark on this page times a write against a store that holds
+/// every record in memory. This one times the same write against a
+/// `didbot_pds::Durable` over a real data directory, which is what a
+/// deployment runs, and it walks the buffer in front of the bodies: a write
+/// signs a commit over `RecordStore::snapshot`, so a repository's whole
+/// contents are read back on every write and the buffer is what decides
+/// whether that read reaches the disk.
+///
+/// Three columns, and the middle one is the number that matters. `buffered`
+/// is the default budget, which holds a repository this size whole;
+/// `unbuffered` is the same deployment with the budget set to zero, which is
+/// the cost of the disk read this exists to avoid; `fill` is what building
+/// the fixture cost per record, so a reader can see the write path at a size
+/// where the repository was still growing.
+#[test]
+#[ignore = "a benchmark: run with --ignored --nocapture"]
+fn durable_write_cost_by_repository_size() {
+    println!(
+        "{:>8}  {:>12}  {:>12}  {:>12}",
+        "records", "fill", "buffered", "unbuffered"
+    );
+    for size in bench_sizes() {
+        let dir = std::env::temp_dir().join(format!(
+            "didbot-durable-bench-{size}-{}",
+            std::process::id()
+        ));
+        let _ = std::fs::remove_dir_all(&dir);
+        let durable = didbot_pds::Durable::open(&dir, time::Duration::days(30))
+            .expect("a data directory opens");
+        let pds = Provisioner::new(
+            "did:web:owner.example",
+            zone(),
+            PDS_ENDPOINT.to_string(),
+            LoopbackDns::new(),
+            durable.accounts(),
+        )
+        .with_record_store(durable.records())
+        .with_blob_store(durable.blobs())
+        .with_commit_history(durable.history());
+        let did = account(&pds, &format!("bench{size}"));
+
+        let started = Instant::now();
+        fill(&pds, &did, size);
+        let per_fill = started.elapsed() / size.max(1) as u32;
+
+        let buffered = timed_writes(&pds, &did);
+        durable.records().heap().set_buffer_bytes(0);
+        let unbuffered = timed_writes(&pds, &did);
+
+        println!("{size:>8}  {per_fill:>12?}  {buffered:>12?}  {unbuffered:>12?}");
+        drop(durable);
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+}
+
+/// Eight writes into `did`, averaged.
+fn timed_writes(pds: &impl Registry, did: &str) -> std::time::Duration {
+    let started = Instant::now();
+    for _ in 0..8 {
+        pds.put_record(
+            did,
+            "com.example.thing",
+            None,
+            thing("timed"),
+            &Swap::default(),
+        )
+        .expect("a write");
+    }
+    started.elapsed() / 8
+}
diff --git a/crates/didbot-pds/tests/store_cost.rs b/crates/didbot-pds/tests/store_cost.rs
index d6bddb93..fea4b51d 100644
--- a/crates/didbot-pds/tests/store_cost.rs
+++ b/crates/didbot-pds/tests/store_cost.rs
@@ -100,6 +100,20 @@ fn live() -> usize {
 /// The value `build` returns is dropped *after* the reading is taken, so what
 /// is measured is what the thing costs while it is being held — which is the
 /// question, because a store is held for the life of the process.
+/// Held across every measurement on this page.
+///
+/// The counter above is one number for the whole process, so two tests
+/// allocating at once measure each other. The test harness runs them in
+/// parallel by default, so the serialization has to be here.
+static MEASURING: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
+/// Takes the measurement lock, recovering from a poisoned mutex.
+fn measuring() -> std::sync::MutexGuard<'static, ()> {
+    MEASURING
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+}
+
 fn resident<T>(build: impl FnOnce() -> T) -> (T, usize) {
     let before = live();
     let held = build();
@@ -185,6 +199,7 @@ fn row(what: &str, count: usize, bytes: usize, wire: Option<usize>) {
 #[test]
 #[ignore = "a benchmark: run with --ignored --nocapture"]
 fn resident_bytes_by_stored_thing() {
+    let _measuring = measuring();
     let count = samples();
     let record = thing("quernstone");
     let json = serde_json::to_vec(&record).expect("json").len();
@@ -280,6 +295,7 @@ fn resident_bytes_by_stored_thing() {
 /// keeps the number in that file honest.
 #[test]
 fn a_record_costs_a_bounded_multiple_of_its_wire_form() {
+    let _measuring = measuring();
     let count = 512;
     let record = thing("quernstone");
     let cbor = didbot_data::dag_cbor::encode(
@@ -312,3 +328,95 @@ fn a_record_costs_a_bounded_multiple_of_its_wire_form() {
          which is {multiple:.1}x; `plan/store-scale.md` records a smaller number"
     );
 }
+
+/// What a record costs resident once its value is in a heap, and what that
+/// claim actually is.
+///
+/// The claim is narrower than "not resident" and this is written to prove
+/// exactly the narrow one. A record still costs its three keys, their share
+/// of three `BTreeMap` nodes, a slot and a content identifier — so the cost
+/// per record is not zero and nothing here says it is. What leaves is the
+/// *value*, and the test for that is that the per-record cost stops moving
+/// when the value gets bigger: two stores holding the same number of records
+/// under the same keys, one with short values and one with values an order of
+/// magnitude longer, cost the same.
+///
+/// The buffer in front of the bodies is set to zero for the measurement,
+/// because the buffer is a knob and what is being measured is the floor
+/// underneath it. `plan/store-scale.md`'s exit criterion is that capacity
+/// stops being a wall at total record *bytes* and becomes a slope in key
+/// *count*; the slope is what the first assertion bounds and the flatness in
+/// value size is what the second one is.
+#[test]
+fn a_record_in_a_heap_costs_its_key_and_not_its_value() {
+    let _measuring = measuring();
+    let count = 512;
+
+    let short = heap_store_cost(count, 8);
+    let long = heap_store_cost(count, 800);
+
+    let per_short = short as f64 / count as f64;
+    let per_long = long as f64 / count as f64;
+
+    // The slope in key count. A slot, a content identifier, three keys and
+    // their nodes; loose, because this is a regression guard rather than a
+    // target, and far under the resident store's own ceiling above.
+    assert!(
+        per_short < 400.0,
+        "a record with its value in a heap costs {per_short:.0} resident bytes, which is more \
+         than its keys and its slot"
+    );
+
+    // And the claim itself, as the one-directional thing it is: a hundred
+    // times the value does not cost more resident bytes. The comparison is
+    // one-sided because the other side is allocator arithmetic — two runs
+    // with different key text land in different size classes — and the
+    // question is whether the value is in there, not whether two runs agree
+    // to the byte.
+    assert!(
+        per_long < per_short * 1.5,
+        "a record's resident cost grew with its value: {per_short:.0} bytes for an 8 byte \
+         value against {per_long:.0} for an 800 byte one, against the {:.0}x growth a resident \
+         value would be",
+        800.0 / 8.0
+    );
+}
+
+/// Resident bytes for `count` records of `value` bytes each, in a heap-backed
+/// store with nothing buffered.
+fn heap_store_cost(count: usize, value: usize) -> usize {
+    let dir = std::env::temp_dir().join(format!(
+        "didbot-heap-cost-{count}-{value}-{}",
+        std::process::id()
+    ));
+    let _ = std::fs::remove_dir_all(&dir);
+    let heap = std::sync::Arc::new(didbot_pds::Heap::open(&dir).expect("a heap opens"));
+    heap.set_buffer_bytes(0);
+
+    let (store, bytes) = resident(|| {
+        let store = didbot_pds::HeapRecordStore::new(heap.clone());
+        for n in 0..count {
+            store
+                .put(
+                    "did:web:one.agents.localhost",
+                    COLLECTION,
+                    None,
+                    thing(&format!("{n:0width$}", width = value)),
+                    &Precondition::Unconditional,
+                )
+                .expect("a write");
+        }
+        store
+    });
+    assert_eq!(store.stats().records, count);
+    row(
+        &format!("record in a heap, {value} byte value"),
+        count,
+        bytes,
+        None,
+    );
+    drop(store);
+    drop(heap);
+    let _ = std::fs::remove_dir_all(&dir);
+    bytes
+}
diff --git a/crates/didbot-serve/tests/record_heap.rs b/crates/didbot-serve/tests/record_heap.rs
new file mode 100644
index 00000000..3f42c093
--- /dev/null
+++ b/crates/didbot-serve/tests/record_heap.rs
@@ -0,0 +1,354 @@
+//! A record written before a restart reads back after one, through XRPC.
+//!
+//! The bodies of records are no longer in the journal: a write appends the
+//! record's canonical DAG-CBOR to `records/pds.heap` and the journal carries
+//! a slot naming it. Whether that survives a restart is a fact about the
+//! binary a deployment starts, over a data directory it keeps, so this asks
+//! the way a client would — `com.atproto.repo.createRecord` into one run,
+//! `com.atproto.repo.getRecord` out of the next — rather than through any
+//! store this workspace could have handed itself.
+//!
+//! The harness is `credential_durability.rs`'s, for the same reason that file
+//! gives: the property lives in `assemble`, and a test that does not run the
+//! binary cannot see it.
+
+#![cfg(unix)]
+
+use std::net::{Ipv6Addr, SocketAddr, TcpListener};
+use std::path::{Path, PathBuf};
+use std::process::{Child, Command, Stdio};
+use std::time::{Duration, Instant};
+
+use serde_json::{json, Value};
+
+/// What `serve_router` logs once it is actually serving.
+const LISTENING: &str = "listening addr=";
+
+/// What the binary prints when the port it was handed is already taken.
+const BIND_FAILED: &str = "could not bind";
+
+/// How long a run gets to reach its listening state.
+const BOOT: Duration = Duration::from_secs(60);
+
+/// How long a run gets to exit once signalled.
+const SHUTDOWN: Duration = Duration::from_secs(30);
+
+/// How many ports a start will try before giving up.
+const PORT_ATTEMPTS: usize = 5;
+
+/// The sample collection, the same one `write_gates.rs` writes into.
+const THING: &str = "com.example.thing";
+
+/// A directory that removes itself, so a failed assertion leaves no
+/// write-ahead log behind in the temp directory.
+struct TempDir(PathBuf);
+
+impl TempDir {
+    fn new(tag: &str) -> Self {
+        let dir = TempDir(std::env::temp_dir().join(format!(
+            "didbot-credential-{tag}-{}-{}",
+            std::process::id(),
+            std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .expect("the clock is after 1970")
+                .as_nanos()
+        )));
+        std::fs::create_dir_all(&dir.0).expect("mkdir");
+        dir
+    }
+
+    fn join(&self, name: &str) -> PathBuf {
+        self.0.join(name)
+    }
+}
+
+impl Drop for TempDir {
+    fn drop(&mut self) {
+        let _ = std::fs::remove_dir_all(&self.0);
+    }
+}
+
+/// One `didbot-pds`, on a port of its own, with the log it writes.
+///
+/// Killed on drop, so an assertion that unwinds does not leave a server
+/// holding the data directory's lock against the next run in the test.
+struct Run {
+    child: Option<Child>,
+    port: u16,
+    log: PathBuf,
+}
+
+impl Drop for Run {
+    fn drop(&mut self) {
+        if let Some(child) = self.child.as_mut() {
+            let _ = child.kill();
+            let _ = child.wait();
+        }
+    }
+}
+
+impl Run {
+    fn log(&self) -> String {
+        std::fs::read_to_string(&self.log).unwrap_or_default()
+    }
+
+    fn wait_until_listening(&mut self) -> bool {
+        let deadline = Instant::now() + BOOT;
+        while Instant::now() < deadline {
+            if self.log().contains(LISTENING) {
+                return true;
+            }
+            let child = self.child.as_mut().expect("still held");
+            if child.try_wait().expect("try_wait").is_some() {
+                return self.log().contains(LISTENING);
+            }
+            std::thread::sleep(Duration::from_millis(50));
+        }
+        false
+    }
+
+    /// Stops the run the way a deployment does, and waits for the flush.
+    ///
+    /// `SIGTERM` rather than a kill because the point of the restart is that
+    /// the next run reads what this one wrote: a killed process leaves the
+    /// question of whether the log was flushed mixed into the answer, and
+    /// `graceful_shutdown.rs` already owns the question of whether the
+    /// signal is handled at all.
+    fn stop(&mut self) {
+        let mut child = self.child.take().expect("still running");
+        let killed = Command::new("kill")
+            .args(["-TERM", &child.id().to_string()])
+            .status()
+            .expect("run kill");
+        assert!(killed.success(), "could not signal the server");
+        let deadline = Instant::now() + SHUTDOWN;
+        loop {
+            match child.try_wait().expect("wait on the server") {
+                Some(_) => break,
+                None if Instant::now() < deadline => {
+                    std::thread::sleep(Duration::from_millis(50));
+                }
+                None => {
+                    let _ = child.kill();
+                    panic!("the server was still running {SHUTDOWN:?} after SIGTERM");
+                }
+            }
+        }
+    }
+}
+
+/// A port nothing was listening on a moment ago. See `graceful_shutdown.rs`
+/// for the window this trades for and how a lost race is detected.
+fn reserve_port() -> TcpListener {
+    let addr = SocketAddr::from((Ipv6Addr::UNSPECIFIED, 0));
+    TcpListener::bind(addr).expect("the kernel has an ephemeral port to spare")
+}
+
+/// Starts the real binary, retrying on a fresh port if it loses the race for
+/// one. `data` is what `--data` is given; `None` runs without the flag.
+fn start(dir: &TempDir, tag: &str, data: Option<&Path>) -> Run {
+    for attempt in 1..=PORT_ATTEMPTS {
+        let mut run = spawn(dir, &format!("{tag}-{attempt}"), data);
+        if run.wait_until_listening() {
+            return run;
+        }
+        let log = run.log();
+        assert!(
+            log.contains(BIND_FAILED),
+            "the server never reached {LISTENING:?}; log:\n{log}"
+        );
+        assert!(
+            attempt < PORT_ATTEMPTS,
+            "lost the race for a reserved port {PORT_ATTEMPTS} times running; last log:\n{log}"
+        );
+    }
+    unreachable!("the loop either returns or asserts")
+}
+
+fn spawn(dir: &TempDir, tag: &str, data: Option<&Path>) -> Run {
+    // A log file per attempt, so a restart's boot line is never confused
+    // with the previous run's.
+    let log_path = dir.join(&format!("{tag}.log"));
+    let log = std::fs::File::create(&log_path).expect("create the log");
+    let errors = log.try_clone().expect("dup the log");
+
+    let reserved = reserve_port();
+    let port = reserved
+        .local_addr()
+        .expect("the reserved socket has an address")
+        .port();
+    let mut command = Command::new(env!("CARGO_BIN_EXE_didbot-pds"));
+    command
+        .args(["--port", &port.to_string()])
+        // No e-stop socket, so this does not contend for the shared default
+        // path with anything else on the machine.
+        .args(["--estop-socket", "none"])
+        .env("NO_COLOR", "1")
+        .stdout(Stdio::from(log))
+        .stderr(Stdio::from(errors));
+    if let Some(data) = data {
+        command.args(["--data", data.to_str().expect("utf-8 path")]);
+    }
+    drop(reserved);
+    let child = command.spawn().expect("spawn didbot-pds");
+
+    Run {
+        child: Some(child),
+        port,
+        log: log_path,
+    }
+}
+
+/// One XRPC answer: the HTTP status and the parsed JSON body.
+struct Answer {
+    status: u16,
+    body: Value,
+}
+
+/// `POST /xrpc/<nsid>` against a run, optionally bearing an agent token.
+async fn post(run: &Run, nsid: &str, token: Option<&str>, body: Value) -> Answer {
+    let mut request = reqwest::Client::new()
+        .post(format!("http://127.0.0.1:{}/xrpc/{nsid}", run.port))
+        .json(&body);
+    if let Some(token) = token {
+        request = request.bearer_auth(token);
+    }
+    let response = request.send().await.expect("the server answered");
+    let status = response.status().as_u16();
+    let text = response.text().await.expect("read the body");
+    let body = serde_json::from_str(&text).unwrap_or_else(|err| {
+        panic!("{nsid} answered {status} with a body that is not JSON ({err}): {text}")
+    });
+    Answer { status, body }
+}
+
+/// An agent account and the one copy of its write credential.
+struct Agent {
+    did: String,
+    token: String,
+}
+
+async fn provision(run: &Run, agent_id: &str) -> Agent {
+    let answer = post(
+        run,
+        "bot.did.provisionAgent",
+        None,
+        json!({ "agentId": agent_id }),
+    )
+    .await;
+    assert_eq!(
+        answer.status, 200,
+        "provisioning was refused: {}",
+        answer.body
+    );
+    Agent {
+        did: answer.body["did"]
+            .as_str()
+            .expect("the response names the account")
+            .to_owned(),
+        token: answer.body["agentToken"]
+            .as_str()
+            .expect("the response carries the write credential")
+            .to_owned(),
+    }
+}
+
+/// `GET /xrpc/<nsid>?<query>` against a run.
+async fn get(run: &Run, nsid: &str, query: &[(&str, &str)]) -> Answer {
+    let response = reqwest::Client::new()
+        .get(format!("http://127.0.0.1:{}/xrpc/{nsid}", run.port))
+        .query(query)
+        .send()
+        .await
+        .expect("the server answered");
+    let status = response.status().as_u16();
+    let text = response.text().await.expect("read the body");
+    let body = serde_json::from_str(&text).unwrap_or_else(|err| {
+        panic!("{nsid} answered {status} with a body that is not JSON ({err}): {text}")
+    });
+    Answer { status, body }
+}
+
+/// Writes one record as `agent`, and hands back what the server said.
+async fn write_as(run: &Run, agent: &Agent, text: &str) -> Answer {
+    post(
+        run,
+        "com.atproto.repo.createRecord",
+        Some(&agent.token),
+        json!({
+            "repo": agent.did,
+            "collection": THING,
+            "record": {
+                "text": text,
+                "emoji": "\u{1f9ff}",
+                "createdAt": "2026-08-27T10:00:00Z",
+            },
+        }),
+    )
+    .await
+}
+
+/// A record written into one run of the server reads back out of the next.
+///
+/// The value, not merely the key: the whole of what moved is where the bytes
+/// live, so a test that only checked the record was listed would pass over a
+/// slot resolving to nothing. The CID is asserted too, because it is what a
+/// client pins and quotes back — and it is now the identifier of bytes on
+/// disk as well as of the record that was handed in.
+#[tokio::test]
+async fn a_record_written_before_a_restart_reads_back_after_one() {
+    let dir = TempDir::new("restart");
+    let data = dir.join("pds");
+
+    let mut first = start(&dir, "first", Some(&data));
+    let agent = provision(&first, "quernstone").await;
+    let written = write_as(&first, &agent, "weftling").await;
+    assert_eq!(
+        written.status, 200,
+        "the record did not write: {}",
+        written.body
+    );
+    let uri = written.body["uri"]
+        .as_str()
+        .expect("the response names the record")
+        .to_owned();
+    let cid = written.body["cid"]
+        .as_str()
+        .expect("the response names the version")
+        .to_owned();
+    let rkey = uri.rsplit('/').next().expect("a key").to_owned();
+    first.stop();
+
+    let second = start(&dir, "second", Some(&data));
+    let read = get(
+        &second,
+        "com.atproto.repo.getRecord",
+        &[("repo", &agent.did), ("collection", THING), ("rkey", &rkey)],
+    )
+    .await;
+    assert_eq!(
+        read.status, 200,
+        "the record did not read back after the restart: {}",
+        read.body
+    );
+    assert_eq!(read.body["uri"].as_str(), Some(uri.as_str()));
+    assert_eq!(
+        read.body["cid"].as_str(),
+        Some(cid.as_str()),
+        "the record came back under a different version"
+    );
+    assert_eq!(
+        read.body["value"]["text"].as_str(),
+        Some("weftling"),
+        "the record came back without the value it was written with"
+    );
+
+    // And the deployment is still writable, which is what proves the heap
+    // was reopened for appending rather than merely read.
+    let again = write_as(&second, &agent, "farthingale").await;
+    assert_eq!(
+        again.status, 200,
+        "the run that read the record could not write another: {}",
+        again.body
+    );
+}