diff --git a/.claude/board/AGENT_LOG.md b/.claude/board/AGENT_LOG.md
index f88d1413e..4b0d56433 100644
--- a/.claude/board/AGENT_LOG.md
+++ b/.claude/board/AGENT_LOG.md
@@ -9,6 +9,11 @@
 - Card `substrate-mapper` (run as a general-purpose agent with the card pasted; the card is not registered in this session), six units: `hyperon-atom`, `hyperon-space`, the minimal-MeTTa interpreter, the grounded stdlib, lance-graph `nal`/`r2il`/`relalg`, `quack`.
 - Ledger: spot-checks `Space::query` L175, `complex_query` L340, `GroundingSpace` L17, interpreter dispatch L417/L453, `query` L604, `FoldDialect` L260 — all correct; 0 wrong. 1 missed: no map recorded that `ModuleSpace` splits conjunctions and requires `ModuleSpace` dependencies (`space/module.rs:33-48`); the orchestrator found it by reading the file in full.
 - Cost: about 515k–646k subagent tokens per run. No conclusion came from a card.
+## 2026-10-14 — WAL shootout harvest: three substrate-mapper runs
+
+- Card `substrate-mapper`, three units: lance 13 `dataset/mem_wal` (5 files, 187 items), lancedb 0.40 MemWAL/LSM (4 files, ~301 items), lance-graph persist sink (`persist_sink.rs`, `batch_writer.rs`, `blw_rows.rs`, `blw_tenant.rs`, 76 items).
+- Ledger: spot-checks 2/2 per map against source (`set_lsm_write_spec` :73, `execute_lsm_merge_insert` :666, `WalSink` :534, `recover_and_apply` :677, and the `LsmWriteSpec` public API); 0 wrong. One stale upstream doc found by the orchestrator, not by a card: `LsmWriteSpec`'s doc calls the `merge_insert` dispatch a follow-up, but `table/merge.rs:320` already dispatches it.
+- Cost: about 587k, 591k and 606k subagent tokens. The maps fed the pre-registration; no conclusion came from a card.
 
 ## 2026-10-14 — D-HOLO H6b harvest: four substrate-mapper runs, one reply each
 
diff --git a/.claude/board/INTEGRATION_PLANS.md b/.claude/board/INTEGRATION_PLANS.md
index 666580bb7..7b846772e 100644
--- a/.claude/board/INTEGRATION_PLANS.md
+++ b/.claude/board/INTEGRATION_PLANS.md
@@ -5,6 +5,13 @@
 - MeTTa is a declaration IR, not an execution IR: the Datalog subset lowers through `lance-graph-relalg` → R2IL → mask-risc; everything else stays in Hyperon's interpreter.
 - Seam: a named `LanceSpace` implementing `hyperon_space::Space`; `ModuleSpace` splits conjunctions, so `&self` is not the seam.
 - Measured PeTTa (SWI-Prolog) near-linear, Hyperon superlinear (anc_800: 0.42 s vs 417 s); new trie panic at ≥ 1,024 children.
+## 2026-10-14 (2) — wal-shootout-v1 — `LanceCycleWriter` vs the lancedb 0.40 MemWAL write path → `.claude/plans/wal-shootout-v1.md`
+
+**Status:** MEASURED (D-WAL-1, D-WAL-2, spec cross-check; D-WAL-3 truncated at 27.5 min by a container restart). Feeds D-LNC-7 without deciding it.
+
+- Same logical cycle on both arms (frame + landing + image rows); M adds a `pk` column for MemWAL.
+- D-WAL-1 validation: SIGKILL atomicity (sweep must bracket the boundary), acknowledged ⇒ durable (read before and after writer replay), idempotent resubmit, read horizon.
+- D-WAL-2 cost per cycle and per row at N = 64 / 1,024 / 8,192; D-WAL-3 60-minute soak per arm with a CSV sample every 30 s.
 
 ## 2026-10-14 — d-holo-address-indexed-algebra-v1 — falsifiers H1, H5, H6 for the address-indexed algebra → `.claude/plans/d-holo-address-indexed-algebra-v1.md`
 
diff --git a/.claude/board/LATEST_STATE.md b/.claude/board/LATEST_STATE.md
index ba37d1969..e36bbbe28 100644
--- a/.claude/board/LATEST_STATE.md
+++ b/.claude/board/LATEST_STATE.md
@@ -1,3 +1,10 @@
+## 2026-10-15 — main after #1555 (merged); D-WAL measured on branch `ccr-9f3b8075-74u47l-wal-shootout` (unpushed)
+
+- **On `main` (#1555, merged `5366a5ab`):** `crates/lance-graph-metta` (excluded) with the D-MTP-1 join probe and the D-MTP-1b range-hop probe; plan `metta-polyfill-v1.md`. Examples only, no production change.
+- **On branch:** D-WAL-5/6/7 (decisions accepted 2026-10-15): `LanceCycleWriter` opens and writes on a Lance session with an 8 MiB metadata cache (`SESSION_METADATA_CACHE_BYTES`); index forms of `deinterlace` / `local_trajectories` in `lance-graph-planner::temporal`.
+- **Contract (new type):** `temporal_pov::{VersionClock, ClockError}` — rows of an append-only lane per version; `rows_at(&TemporalPov)` resolves a point of view to one half-open row range (v4: 4.2 ns per call; a per-row scan of 400k rows takes 127 µs). Eight bytes per version.
+- **On branch (unpushed):** D-WAL-1/2/3 for `LanceCycleWriter` vs the lancedb 0.40 MemWAL path (probe, harness, example behind `lancedb-sdk`); entry `2026-10-15-d-wal-shootout.md`. No production wiring.
+
 ## 2026-10-14 — the rail walk applies the reader's field mask on branch `ccr-0455e606-wmtsor` (unmerged)
 
 - **Contract:** `selection::walk_rails_authorized(graph, class_view, registry, view_binding, allowed, root, max_depth, visit)`. Same walk as `walk_rails`, with `allowed(class)` added to the AND at every hop: `view.mask ∩ presence ∩ allowed(class)`. A hidden field is not emitted at any depth (each hop uses its own class's mask), and a hidden rail is not followed, so a linked object cannot be read through a link the reader may not see. `allowed` returns a mask, not an `Option`, so there is no permit-all fallback. `walk_rails` is unchanged. Closes H7 for every output built on the walk (OGAR `ogar-doc-ir` resolve uses it); the caller passes the `field_mask` of its `Decision` per class. Tests: four `selection::tests::authorized_walk_*`; three fail with the mask removed, the hidden-rail test fails with the mask applied to emission only.
diff --git a/.claude/board/PR_ARC_INVENTORY.md b/.claude/board/PR_ARC_INVENTORY.md
index d48e36a6b..b69bd656c 100644
--- a/.claude/board/PR_ARC_INVENTORY.md
+++ b/.claude/board/PR_ARC_INVENTORY.md
@@ -1,3 +1,10 @@
+## 2026-10-14 — lance-graph PR #1555 (merged `5366a5ab`, branch `ccr-9f3b8075-74u47l-metta`, head `457a8c41`, 14 commits) — D-MTP: MeTTa over the masking-algebra polyfill; two-hop join and the hop as a range
+
+- **Added:** plan `metta-polyfill-v1.md` (Hyperon seams, PeTTa vs Hyperon measurements, MeTTa-as-IR verdict, datafrog fork as the exact Rust oracle). `crates/lance-graph-metta` (own workspace, excluded; datafrog from `AdaWorldAPI/datafrog` as a dev-dependency only) with `join_probe` (D-MTP-1, two-hop join through Quack `Semijoin`) and `join_range_probe` (D-MTP-1b, the hop as a range under a monotone `dst`). Entries `2026-10-14-metta-polyfill-map.md`, `-d-mtp-1-join.md`, `-d-mtp-1b-range-hop.md`; STATUS D-MTP rows.
+- **Locked:** pairs equal datafrog at every size; counts equal Hyperon and PeTTa. At 10⁶ (v4): gather count 1.2–1.8 ns/row, emit 4.0–5.7 ns/pair against datafrog's 59–71; the range path counts in 0.0001–0.0004 ns/row with runs recorded at ingest and refuses a non-monotone lane. relalg has no hop yet, so the join lowers through Quack.
+- **Deferred / open:** D-MTP-2..5 (named `LanceSpace`, recursion via the external driver, the trie width fix upstream, PeTTa in the probes); the order witness is probe-local.
+- **Confidence:** high for equality (disable runs red as pre-registered); timings are best of 5 over three runs on one VM.
+
 ## 2026-10-14 — lance-graph PR #1552 (merged `98c03dd1`, branch `ccr-9f3b8075-74u47l-haiku-harvest`) — Haiku harvest tier: six cards and their supervision rules
 
 - **Added:** cards `.claude/agents/{harvest-scout,census-counter,quote-harvester,substrate-mapper,spec-editor,guarded-executor}.md`; `tiered-agent-execution-protocol.md` § Haiku harvest tier (card table, brief sizing, supervision rules, mapping a large surface); CLAUDE.md Model Policy rewritten; BOOT files, worker template, cca2a and v3 BOOT aligned; AGENT_LOG entries for the first runs; #1551 board entries.
diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md
index 794d0326d..900808b03 100644
--- a/.claude/board/STATUS_BOARD.md
+++ b/.claude/board/STATUS_BOARD.md
@@ -1,3 +1,17 @@
+## D-WAL — `LanceCycleWriter` vs the lancedb 0.40 MemWAL write path (2026-10-14)
+
+Plan: `.claude/plans/wal-shootout-v1.md`. Feeds D-LNC-7; decides nothing.
+
+| D-id | scope | status | gate / falsifier |
+|---|---|---|---|
+| **D-WAL-1** | V1 SIGKILL atomicity, V2 acknowledged ⇒ durable, V3 idempotent resubmit, V4 read horizon; both arms | Measured (§2a): V1–V3 hold on both arms; M's acknowledged writes are invisible to other readers until a flush or a writer replay | no partial cycle, sweep brackets the boundary; no acknowledged cycle lost after the replay read; V3/V4 behaviour recorded |
+| **D-WAL-2** | commit p50/p99 and ns/row at N = 64 / 1,024 / 8,192; disk, files, versions/generations, full-scan latency after 200 cycles | Measured (§3a): M waits for the WAL ticker (~95 ms at the 100 ms default); at 10 ms faster than L from N = 1,024; M holds ~2× the bytes and its scan grows without compaction | numbers with tier; no threshold |
+| **D-WAL-3** | 60-minute soak per arm, CSV every 30 s | Measured, truncated at 27.5 min / 1,649 cycles by a container restart (§4a): M commit p50 flat 2.0–2.8 ms, L 5.8 → ~10.5 ms; M disk 1.57× L (WAL kept after each flush); both scans linear in rows; L RSS 95 → 762 MiB unexplained | drift, growth and RSS reported |
+| **D-WAL-4** | where L's resident memory goes | Measured: every Lance commit caches a full manifest clone per version (lance 13 `record_successful_commit`); append-only adds one fragment per version, so the cache grows as V²/2 to its 1 GiB default. 1,600 versions: default session 628 MiB cache / 570 MiB RSS, 8 MiB-capped session 7.8 MiB / 52 MiB | `examples/wal_rss_probe.rs`, default vs capped arm in separate processes |
+| **D-WAL-5** | cap the writer's Lance session | Done: open and both Create writes use a session with an 8 MiB metadata cache. 400 commits: 42.6 MB uncapped, under 8 MiB capped; reopen path covered (127 MB uncapped at 800) | `session_cache_stays_under_the_cap_over_many_commits`; disables on the Create write and on open each turn it red |
+| **D-WAL-6** | version clock: a point of view as one row range | Done: `contract::temporal_pov::VersionClock`; v4 4.2 ns per `rows_at` vs 127 µs per-row scan at 400k rows | row-version oracle over 200 versions with gaps; disables (gap fill, start offset, base guard) each turn it red. Not yet wired into a writer or reader; seeding from storage at open is open |
+| **D-WAL-7** | `temporal.rs` without row copies | Done: `deinterlace_indices`, `local_trajectories_indices`, `local_trajectory_indices_of`; copying forms keep their signatures as wrappers | clone counter at 0 for the index form; removing the sort turns the two ordering tests red |
+
 ## D-MTP — Hyperon/MeTTa over the masking-algebra polyfill (2026-10-14)
 
 Plan: `.claude/plans/metta-polyfill-v1.md`. Proposed crate `lance-graph-metta` (not created).
@@ -850,7 +864,7 @@ identity) is real and probe-gated.
 | D-GTM-0n / P3 | **Ternlog chaining amortization** (`ndarray/examples/ternlog_amortization_probe.rs`, re-run `--release` 2026-09-06): does chaining pay on THIS mask shape? Pre-registered pass criterion = per-constraint cost flat in K, with a **K=1 control** (nothing to chain ⇒ no win) | **GREEN, bounded.** `T3/T1` bottoms at **0.50** by K=8; T3 total flat in K (57.6 ns @ K=4 → 54.1 ns @ K=32, 16× more constraints). K=1 control reads **1.03** as it must. Two bounds ride with the number and may not be dropped when it is cited: the win is **L2-residency-contingent** (bandwidth `138→15 GB/s` past L2, ratio back to 1.03 at a 512 KiB mask), and mask loses to sparse below **0.1%** active. **Does NOT retire the standing caveat** — 2× on one of three passes still cannot account for 5×. Gates Stage-3 SPEED claims only; the shape/correctness case never depended on it |
 | D-ALPHA-1b | **`AlphaMask::zip` release-mode length guard** (`lance-graph-contract/src/alpha.rs`): the only guard was `debug_assert_eq!`, compiled out in release, so mismatched-length operands produced a mask claiming `self.len` addresses over the SHORTER operand's word count — `count`/`is_empty` under-report silently, `contains`/`materialize_ordinals` panic at a distance | **Fixed** — promoted to `assert_eq!`. Defect reproduced in release BEFORE the fix (the `should_panic` falsifier reported "test did not panic as expected"); two-sided falsifiers both disable-verified. No API change, no external callers. Stage 1b of `.claude/temporal/09-plan.md`, now DONE |
 | D-NGN-AUDIT | **`NodeGuid::new` repurpose audit** (5+3 council, `.claude/plans/nodeguid-new-repurpose-audit-v1.md`): can the V1 mint be repurposed as the V3 cascade constructor? READ-ONLY — operator: *"it's already in production, if you touch it it would change something you half understand."* | **HELD at Phase 2**; SPEC v1 SHIPPED in #1217 (`b518dbf1`). Phase 1 RETURNED (all 5 savants); consolidation is held on a VIOLATES against frozen decision F5 (*"Nobody walks anything"*) — `GET /views/graph/wave.abi` reaches `inherited_part_of` — which is the operator's to rule, not the orchestrator's to resolve. **Do NOT re-cast Phase 1, and do not cast the 3 reviewers until that ruling lands.** Draft v2 also owes F9 (the #1202 dating rule). The spec was orphaned by a branch reset before #1201 merged and recovered on 2026-09-07; #1201 itself merged hygiene-only. Spec carries a §0 retracted-readings table (4 false orchestrator readings, banked so they are not re-derived) |
-| D-LNC-7 | **One-WAL ruling**: `LanceCycleWriter`'s watermark WAL vs lance MemWAL — `applied_through` ≡ index catch-up position; two under one commit = two truths. Plan recommends keep-ours until D-LNC-3, then re-decide | **Operator decision** |
+| D-LNC-7 | **One-WAL ruling**: `LanceCycleWriter`'s watermark WAL vs lance MemWAL — `applied_through` ≡ index catch-up position; two under one commit = two truths. Plan recommends keep-ours until D-LNC-3, then re-decide | Decided 2026-10-15 after D-WAL-1..5: `LanceCycleWriter` stays the default (session metadata cache capped, D-WAL-5); MemWAL stays an option. 60-minute soak deferred |
 | D-TEMPORAL-1 | **Canonical witness identity probe** (`crates/lance-graph/tests/canonical_witness_identity_probe.rs`): does canonical address identity survive the A-store round trip on lance 11? `NodeGuid → NodeRow → NodeRowPacket::as_le_bytes → cycle commit → drop writer → reopen → scan_image`. Minted through `mint_for(classid_read_mode(c).tail_variant, …)`, never a hardcoded constructor | **GREEN, both rails.** V3 (`CLASSID_OSINT_V3`): payload bytes byte-exact across two cycles and a reopen; `payload[0..16] == G.as_bytes()`; value slab survived (so equality is not a zero-match); each cycle minted exactly `base+1`; head recovered at V2 after restart. V1 legacy control (`CLASSID_OSINT`) round-trips too. **Scope:** proves byte and address fidelity of the store; does NOT prove production reachability — `LanceCycleWriter` still has zero non-test callers (see D-MW-P2) |
 | D-TEMPORAL-1a | **Pinned false-green: the V1 accessor on a V3 key.** 256 V3 addresses varying only in `leaf` → 256 distinct 16-byte keys, 256 distinct `to_hex_v2()`, but **1** distinct `identity()`. Byte fidelity does not imply address fidelity | **Pinned as a positive assertion.** Also pins the second half, which is not a defect: `local_key_v2` yields **1** across the same set BY CONSTRUCTION — `leaf` is the 4th HHTL routing tier (bytes 10..12), part of the addressing prefix, deliberately outside the basin-local key (bytes 12..16). So "use the V3 accessor" is two choices, not one, and the wrong V3 one fails like the V1 one. Do not update the numbers if the assertion breaks — re-audit the call sites |
 | D-V3-MINT-1 | **V3 mint reachability** (`crates/lance-graph-contract/tests/v3_mint_reachability_probe.rs`): `mint_for`'s V2/V3 arm is gated on `guid-v2-tail`, which is NOT in `default`; the registry's V3 entries are gated on `guid-v3-tail`, which is. Two different features, so premise and conclusion can come apart — is a V3-registered classid silently minted into the deprecated V1 layout? | **GREEN — no defect.** `guid-v3-tail = ["guid-v2-tail"]` (`Cargo.toml:69`), so the implication holds. Measured, not inferred: `tail_variant=V3 guid-v3-tail=true guid-v2-tail=true leaf_is_live=true`, key bytes show the V2/V3 carving with `leaf` live at byte 10. Discriminator is `leaf` (the V3 arm feeds it to `new_v2`; the V1 fallback discards it), so the probe stays valid if the feature graph is ever flattened |
diff --git a/.claude/board/entries/2026-10-15-d-wal-shootout.md b/.claude/board/entries/2026-10-15-d-wal-shootout.md
new file mode 100644
index 000000000..59459657d
--- /dev/null
+++ b/.claude/board/entries/2026-10-15-d-wal-shootout.md
@@ -0,0 +1,37 @@
+# 2026-10-15 — D-WAL — `LanceCycleWriter` vs the lancedb 0.40 MemWAL write path (measured)
+
+**Status:** MEASURED. Plan `.claude/plans/wal-shootout-v1.md` §2a (D-WAL-1), §3a (D-WAL-2),
+§3b (spec cross-check against lance `v13.0.0` `docs/src/format/table/mem_wal.md`), §4a
+(D-WAL-3, truncated). Probe `crates/lance-graph/tests/wal_shootout_probe.rs`, harness
+`tests/wal_common/`, example `examples/wal_shootout.rs`, all behind `lancedb-sdk`.
+
+**Outcome:**
+- Both paths are crash-atomic per cycle and lose nothing they acknowledged (SIGKILL sweep
+  brackets the boundary on both). MemWAL needs `durable_write = true` for that; with
+  `false` an acknowledged cycle is lost (disable run).
+- MemWAL's acknowledged writes are visible to the writing handle only; another handle or
+  process sees them after a flush or after it replays the WAL by opening a writer, and the
+  base table only after compaction. The spec states this as its consistency model
+  (strong only with live memtable access, otherwise eventual).
+- Cost: with the default 100 ms flush every MemWAL write waits about 95 ms; at 10 ms it is
+  slower than `LanceCycleWriter` at 64 casts per cycle and faster from 1,024. In the soak
+  (256 casts, 10 ms) MemWAL's commit stayed flat at 2.0–2.8 ms while L's grew from 5.8 to
+  about 10.5 ms over 1,649 versions.
+- MemWAL keeps the WAL after flushing: disk grows by a WAL increment per cycle plus a
+  second copy at each memtable flush (every ~195 cycles = the 100k-row default), 1.57× L at
+  1,649 cycles. The spec describes no component that removes WAL files.
+- Neither path compacts; both full scans grow linearly in rows.
+
+- L's RSS growth is Lance's session metadata cache, not our writer (D-WAL-4,
+  `examples/wal_rss_probe.rs`): each commit inserts a clone of the new manifest keyed by
+  version, and in an append-only store manifest v lists v fragments, so the cache holds
+  about V²/2 fragment entries up to its 1 GiB default. At 1,600 versions: default session
+  628 MiB cache, 570 MiB RSS; session with an 8 MiB metadata cache 7.8 MiB, 52 MiB RSS.
+  Capping the metadata cache, or compacting so fragment count stops growing, bounds it.
+
+- Follow-up done the same day (D-WAL-5/6/7): the writer's session cap, a version clock
+  (`contract::temporal_pov::VersionClock`, 4.2 ns per point-of-view read on v4), and index
+  forms of the `temporal.rs` projections. D-LNC-7: L stays default, MemWAL an option.
+
+**Open:** a full-length soak (deferred; this one stopped at 27.5 min when the container restarted); the effect of compaction on both; D-LNC-7
+(which WAL to keep) stays a decision for the user.
diff --git a/.claude/board/entries/README.md b/.claude/board/entries/README.md
index a73d115fd..782e4b647 100644
--- a/.claude/board/entries/README.md
+++ b/.claude/board/entries/README.md
@@ -25,11 +25,12 @@ index row, (3) no duplicate entry id. Checks 1 and 2 are deliberately
 opposite directions; the stranding this convention prevents shows up in
 exactly one of them, never both.
 
-335 entries, 2026-08-06 .. 2026-10-15.
+336 entries, 2026-08-06 .. 2026-10-15.
 
 | date | entry id | finding | file |
 |---|---|---|---|
 | 2026-10-15 | `ledger-correction-wire-contract` |  | [2026-10-15-ledger-correction-wire-contract.md](2026-10-15-ledger-correction-wire-contract.md) |
+| 2026-10-15 | `D-WAL` | `LanceCycleWriter` vs the lancedb 0.40 MemWAL write path (measured) | [2026-10-15-d-wal-shootout.md](2026-10-15-d-wal-shootout.md) |
 | 2026-10-14 | `revision-layer-invariants` |  | [2026-10-14-revision-layer-invariants.md](2026-10-14-revision-layer-invariants.md) |
 | 2026-10-14 | `D-MTP-0` | where Hyperon/MeTTa attaches to the masking-algebra polyfill | [2026-10-14-metta-polyfill-map.md](2026-10-14-metta-polyfill-map.md) |
 | 2026-10-14 | `ledger-revision-after-1553` |  | [2026-10-14-ledger-revision-after-1553.md](2026-10-14-ledger-revision-after-1553.md) |
diff --git a/.claude/plans/lance-graph-as-the-modelgraph-v1.md b/.claude/plans/lance-graph-as-the-modelgraph-v1.md
index 6cd86a30e..77272eed1 100644
--- a/.claude/plans/lance-graph-as-the-modelgraph-v1.md
+++ b/.claude/plans/lance-graph-as-the-modelgraph-v1.md
@@ -323,10 +323,9 @@ accordingly to batchwriter."* Followed. The arc, and what it left:
 
 **So `Dataset::write` is real and in-tree** — two call sites, under a module doc
 saying the I/O *"is INTERNAL to this one writer"*:
-`Dataset::write` at `crates/lance-graph/src/graph/cycle_sink.rs:675` (the Create
-inside `LanceCycleWriter::bootstrap`) and `Dataset::write` at
-`crates/lance-graph/src/graph/cycle_sink.rs:710` (the Create-or-Append inside
-`LanceCycleWriter::raw_append`). The plan's G-C above — *"no `Dataset::write`"* —
+`Dataset::write` in `crates/lance-graph/src/graph/cycle_sink.rs`, function
+`LanceCycleWriter::bootstrap` (the Create), and `Dataset::write` in the same file,
+function `LanceCycleWriter::raw_append` (the Create-or-Append). The plan's G-C above — *"no `Dataset::write`"* —
 was **wrong when written.**
 
 > ⊘ The first spelling of this citation was `…rs:675` followed by a bare
diff --git a/.claude/plans/wal-shootout-v1.md b/.claude/plans/wal-shootout-v1.md
new file mode 100644
index 000000000..4d129d36d
--- /dev/null
+++ b/.claude/plans/wal-shootout-v1.md
@@ -0,0 +1,252 @@
+# WAL shootout v1 — `LanceCycleWriter` vs the lancedb 0.40 MemWAL write path
+
+**Status:** D-WAL-1 MEASURED (§2a), D-WAL-2 MEASURED (§3a), spec cross-check (§3b), D-WAL-3 MEASURED, truncated at 27.5 min by a container restart (§4a). D-WAL-4..7 DONE (§7); D-LNC-7 decided 2026-10-15 (§7). Feeds D-LNC-7 (the one-WAL
+decision, `lance-convergence-staged-migration-v1.md` §4C); this plan does not decide it.
+
+## 1. What is compared
+
+| arm | write | durable unit | read |
+|---|---|---|---|
+| **L** — ours | `LanceCycleWriter::commit_cycle(DetachedCycleBatch)` (`crates/lance-graph/src/graph/cycle_sink.rs`) | one Lance append = one `DatasetVersion` per cycle; frame + landing + image rows in one `RecordBatch` | plain Lance scan at the head version |
+| **M** — lancedb 0.40 | `Table::set_lsm_write_spec(LsmWriteSpec::unsharded())`, then `merge_insert(["pk"])` upsert, which `table/merge.rs:320` dispatches to `execute_lsm_merge_insert` → MemWAL `ShardWriter::put` | one WAL entry per `put` (`durable_write = true` by default, flush ticker 100 ms) | LSM scan (`table/query.rs:178`: memtables + flushed generations); `use_lsm(false)` reads the base table only |
+
+"Our hand-rolled batch writer" is the pair `BatchWriter` (staging, no I/O,
+`batch_writer.rs`) + `LanceCycleWriter` (the durable half). `BatchWriter` does no I/O, so
+only `LanceCycleWriter` has a MemWAL counterpart (`.claude/temporal/08-memwal-vs-batchwriter.md`).
+
+Both arms store the same logical cycle: the rows `LanceCycleWriter::build_batch` writes
+(1 frame row, N landing rows, N image rows with a 512-byte payload). M adds one column,
+`pk: Int64` (`cycle << 24 | ordinal`; lancedb rejects `UInt64` as a primary key), carrying the unenforced-primary-key marker MemWAL
+requires. Same N, same payload bytes.
+
+Prior evidence, not re-run here: D-MW-P2 (lance 11, raw `ShardWriter::put`, 5,000 rows)
+found MemWAL puts crash-atomic.
+
+## 2. Validation (D-WAL-1), each test on both arms
+
+- **V1 atomicity under SIGKILL.** A child process writes one composite cycle of
+  N = 4,096 + 4,096 + 1 rows and is killed at a sweep of delays. Recovered rows per cycle
+  ∈ {0, all}. The sweep must land on both sides of the durability boundary or the test
+  reports INCONCLUSIVE (the D-MW-P2 anti-vacuity rule). For M, "recovered" is read twice:
+  (a) by a fresh process that only reads, (b) after that process opens the shard writer
+  (replay).
+- **V2 acknowledged ⇒ durable.** The child writes cycles in a loop and prints `ACK k` after
+  each call returns; the parent SIGKILLs right after reading an ACK. Every acknowledged
+  cycle must be readable after restart. M is read the same two ways as V1. A cycle that is
+  durable but invisible to a reader until a writer replays is reported as such, not as a
+  pass or a failure.
+- **V3 idempotent resubmit.** Submitting the same frozen cycle twice: L returns
+  `Reconciled` and the row count is unchanged; M's upsert by `pk` leaves the count
+  unchanged.
+- **V4 read horizon.** While cycle n+1 is acknowledged but not flushed: what does a second
+  reader see? L: a reader at the head it opened sees cycle n only until it re-opens. M: the
+  LSM read in the writing process, an LSM read from a second handle, and a base-only read.
+  Recorded as observed behaviour; the sealed-horizon property in
+  `08-memwal-vs-batchwriter.md` §"three incompatibilities" item 2 is what this checks.
+
+Scope: process crash on a local filesystem. Not power loss (no claim about fsync), not
+object storage, not concurrent writers or fencing.
+
+## 2a. D-WAL-1 results (2026-10-14)
+
+`crates/lance-graph/tests/wal_shootout_probe.rs`, lance 13.0.0 / lancedb 0.40.0, debug build
+(correctness only; no timing is read from it), local filesystem, M flush interval 10 ms.
+
+**V1 SIGKILL atomicity — holds on both arms.** N = 4,096 casts (8,193 rows) per cycle:
+
+| kill after | 5 | 30 | 60 | 120 | 250 | 500 | 1,000 | 2,000 ms |
+|---|---|---|---|---|---|---|---|---|
+| L, fresh reader | 0 | 0 | 0 | 8,193 | 8,193 | 8,193 | 8,193 | 8,193 |
+| M, reader that has not written | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 |
+| M, same reader after its first write (replay) | 0 | 0 | 0 | 0 | 8,193 | 8,193 | 8,193 | 8,193 |
+
+Both sweeps land on both sides of the durability boundary; no partial cycle anywhere.
+
+**V2 acknowledged ⇒ durable — holds on both arms, with a visibility difference.** Kill right
+after reading ACK 2, 5 and 9 (N = 1,024): L — every acknowledged cycle is readable by a fresh
+reader. M — a reader that has not written sees **no** cycle at all; after its first write
+(which opens the shard writer and replays the WAL) every acknowledged cycle is present.
+Disable run (after commit): `durable_write = false` for M → acknowledged cycle 2 lost after
+replay, V2 red; restored.
+
+**V3 idempotent resubmit — holds on both arms.** L: `Committed` then `Reconciled`, 129 rows.
+M: two upserts of one cycle, 129 rows (dedup by `pk`).
+
+**V4 read horizon.** L: a reader opened at cycle 1 keeps seeing cycle 1 after cycle 2 commits;
+a fresh reader sees both. M, after two acknowledged cycles: the writing handle's LSM read sees
+both; a second handle in the same process (its own writer cache) sees **neither**; the
+base-only read sees neither. After `close_lsm_writers` the second handle's LSM read sees both;
+the base-only read still sees neither (no compaction ran).
+
+**What this means.** Both paths are crash-atomic per cycle and lose nothing they
+acknowledged. They differ in who can read an acknowledged write: L publishes a version every
+reader can open; M's acknowledged writes live in the WAL and the writer's memtable, readable
+by the writing handle, by any process only after a flush into a generation (here:
+`close_lsm_writers`) or after that process replays the WAL by opening a writer, and by the base
+table only after compaction. The sealed read horizon (`08-memwal-vs-batchwriter.md`) is
+therefore not lost by MemWAL for OTHER readers — they see less, not more — but a reader
+sharing the writer's handle sees in-flight state, which is the property item 2 of that doc
+warns about.
+
+## 3. Cost (D-WAL-2)
+
+Per arm, N ∈ {64, 1,024, 8,192} image rows per cycle, 200 cycles each, release build,
+program-printed tier:
+
+- commit latency per cycle: p50 / p99 (µs), and ns per row;
+- after the 200 cycles: bytes on disk, file count, Lance versions (L) or WAL entries and
+  flushed generations (M), and the latency of one full scan of all rows;
+- M twice: default flush interval (100 ms) and 10 ms, because `durable_write` waits for
+  the WAL flush and the ticker can dominate a small cycle.
+
+Direction only, no threshold before the run: M's per-call latency should track the flush
+interval for small N; L's per-cycle cost should be flat in history length while its read
+cost grows with the number of versions/fragments (never compacted).
+
+## 3a. D-WAL-2 results (2026-10-14)
+
+`crates/lance-graph/examples/wal_shootout.rs latency`, release, `-Ctarget-cpu=native`, the
+program prints `avx512f=true`; three runs, ranges span the three. Local filesystem; the cost
+is I/O and encoding, so it is reported per cycle (ms) and per stored row (µs). A cycle of N
+casts stores 2N + 1 rows (N × 512 payload bytes).
+
+| arm | N | commit p50 ms | commit p99 ms | µs per row (p50) |
+|---|---|---|---|---|
+| L | 64 | 3.5–4.8 | 5.9–8.3 | 27–37 |
+| M, flush 10 ms | 64 | 7.0–8.1 | 10.5–14.0 | 55–63 |
+| M, flush 100 ms | 64 | 96.9–97.4 | 101–103 | 751–755 |
+| L | 1,024 | 7.9–9.0 | 12.5–13.9 | 3.9–4.4 |
+| M, flush 10 ms | 1,024 | 5.3–5.6 | 15.8–17.2 | 2.6–2.7 |
+| M, flush 100 ms | 1,024 | 93.5–94.3 | 101–102 | 45.6–46.0 |
+| L | 8,192 | 72–80 | 90–106 | 4.4–4.9 |
+| M, flush 10 ms | 8,192 | 34–35 | 82–90 | 2.1–2.2 |
+| M, flush 100 ms | 8,192 | 89–91 | 172–179 | 5.4–5.6 |
+
+After the run (200 cycles at N = 64 and 1,024, 40 at N = 8,192):
+
+| N | disk L / M (MiB) | files L / M | versions L / M base | full scan L / M ms |
+|---|---|---|---|---|
+| 64 | 15.6 / 15.0 | 601 / 209 | 200 / 3 | 32–36 / 9–12 |
+| 1,024 | 205.7 / 427–429 | 601 / 234–241 | 200 / 3 | 33–43 / 91–147 |
+| 8,192 | 323.6 / 663.8 | 121 / 97 | 40 / 3 | 21–35 / 211–298 |
+
+**Against the pre-registration.**
+- "M's per-call latency tracks the flush interval for small N": confirmed and stronger than
+  expected. With the default 100 ms ticker every acknowledged write waits about 94–97 ms at
+  any N; at 10 ms, M is slower than L at N = 64 (7–8 ms vs 3.5–4.8) and faster from N = 1,024
+  (5.4 vs 8.5 ms; 34 vs 76 ms at 8,192).
+- "L's per-cycle cost flat in history": consistent over 200 cycles (checked further in the
+  soak).
+- Disk: M holds about 2× L's bytes at N ≥ 1,024. Hypothesis, not checked here: the WAL
+  entries stay on disk after their memtable is flushed into a generation, so the payload is
+  stored twice until something removes the WAL.
+- Read: L's full scan stays at 21–43 ms; M's LSM scan grows with the data (9 → 91–147 →
+  211–298 ms), because nothing compacts the flushed generations into the base table (the
+  base table stays at version 3 throughout).
+
+## 3b. Spec cross-check (2026-10-14)
+
+Source: lance `docs/src/format/table/mem_wal.md` at tag `v13.0.0`, titled "MemTable & WAL
+Specification (Experimental)". `main` differs only in the maintained-index fields and a
+per-entry `generation` value; nothing below depends on those.
+
+| property | what the spec says | what D-WAL-1/2 measured |
+|---|---|---|
+| atomicity | "Every durable WAL append creates one WAL entry" (l. 88); entry N+1 never exists before N is complete | V1: no partial cycle on M, consistent with one `put` = one entry |
+| durability | writes are "durably appended" to the shard WAL (l. 54); the spec never names `durable_write` or fsync | V2 holds with `durable_write = true` and fails with `false`: in lancedb that setting, not the format, decides whether an ack waits for the WAL |
+| recovery | "Recovery replays from `replay_after_wal_entry_position + 1`" (l. 98), done by a writer claiming the shard | V1/V2: a process sees M's rows only after its first write opens a writer and replays |
+| visibility | strong consistency needs "active in-memory MemTable access" plus the latest shard manifests; "Otherwise, reads are eventually consistent" (l. 578–579) | V4: the writing handle sees its rows; a second handle and a base-only read see nothing until a flush |
+| single writer | epoch fencing; a stale writer collides on the next WAL position (l. 110, 490–509) | not tested (§2 scope excludes concurrent writers) |
+| compaction | a separate job: merge-insert into base, `compacted_sstables` committed atomically with base (l. 518) | the base table stayed at version 3; nothing in our runs started it, so the LSM scan grew 9 → 298 ms |
+| garbage collection | removes SSTables only, after compaction and index catch-up (l. 527–532); "Deleting WAL files can weaken writer fencing" (l. 534) | M used about 2× L's disk. The spec describes no component that removes WAL entries, which supports the §3a hypothesis that the payload stays twice on disk; who may remove the WAL is not specified |
+
+Consequence for D-LNC-7: under the spec's own terms M is crash-atomic and durable per put,
+but it is eventually consistent for every reader except the writing handle, and it is only
+complete once something outside the write path runs compaction and garbage collection. L
+gives every reader a sealed version with no background job. Neither property was in doubt
+after §2a; the spec confirms they are by design, not by our configuration.
+
+## 4. Soak (D-WAL-3) — "monitor both for a while"
+
+One background process per arm, own directory, same cadence: one cycle of 256 image rows
+every second for 60 minutes (about 3,600 cycles, about 0.5 GB per arm at 512 B per row).
+Every 30 s it appends a CSV row: cycles committed, commit p50/p99 over the window, bytes
+on disk, file count, versions (L) / generations (M), full-scan latency, RSS. Questions:
+does commit latency drift with history; does read latency grow without compaction; does
+disk grow linearly; does RSS stay flat (M holds memtables).
+
+Disk budget is checked before the start (this host has about 5 GB free); the soak stops
+early if free space falls under 1 GB, and says so.
+
+## 4a. D-WAL-3 results (2026-10-15) — truncated at 27.5 minutes
+
+`crates/lance-graph/examples/wal_shootout.rs soak`, release build, one process per arm, one
+cycle of 256 casts (513 rows, 128 KiB of payload) per second, M with a 10 ms flush interval.
+Started 23:01:38 UTC. **Both runs stopped at t = 1,651 s (1,649 cycles), not 60 minutes:
+the container restarted at about 23:29 UTC** (`uptime` 0 min at the check-in, both CSVs end at
+the same second, empty logs, no "stopped early" line). Samples from t ≈ 970 s to 1,300 s
+overlap ruff builds run at low priority on the same host and are not used for latency.
+Timings are whole-call I/O on one VM, not primitive costs.
+
+| at t ≈ 30 s → 1,651 s | L (`LanceCycleWriter`) | M (MemWAL, flush 10 ms) |
+|---|---|---|
+| commit p50 | 5.8 → 10.1–11.2 ms (last 5 min); grows with history | 2.0–2.8 ms throughout; no trend |
+| commit p99 | mostly 9–35 ms; outliers 239 ms (t = 1,141, build window) and 192 ms (t = 1,591, cause unknown) | mostly 3–16 ms; one outlier 64 ms (t = 1,141, build window) |
+| disk | 562 MiB, linear, 0.34 MiB per cycle | 883 MiB (1.57× L): 0.29 MiB per cycle continuously, plus a step of about 60 MiB every ~195 cycles |
+| files | 4,948 (3 per cycle) | 1,722 |
+| versions | 1,649 (one per cycle) | base table stays at 3 |
+| full scan | 9.7 → 445 ms (846k rows, 0.53 µs/row) | 6.2 → 374 ms (0.44 µs/row; one sample 604 ms) |
+| RSS | 95 → 762 MiB, about 0.4 MiB per cycle, no plateau | 114 → 335 MiB, rising at each flush step |
+
+**What it shows.**
+- **Commit latency:** M is flat and 2–4× cheaper per cycle at this size; L's p50 nearly
+  doubles over 1,649 versions. Consistent with the D-WAL-2 crossover (M faster from about
+  1,000 casts per cycle at 10 ms flush; here 256 casts, but L pays per version).
+- **WAL not truncated, now measured over time:** M's disk grows by a WAL increment every
+  cycle and by a second copy at every flush, which comes every ~195 cycles, i.e. when the
+  100,000-row memtable default fills (195 × 513 ≈ 100k). The payload is stored twice and
+  nothing removes the WAL, as the spec cross-check (§3b) predicted.
+- **Scan:** both arms grow linearly in rows, M slightly cheaper per row. This corrects the §3a
+  reading that L's scan "stays at 21–43 ms": that held for 40–200 large cycles; with 1,649
+  small versions L pays per fragment too. Neither arm compacts.
+- **RSS:** L's process grows without a plateau and ends at 2.3× M. Not explained yet: the
+  writer, Lance's caches for 1,649 versions, or the in-process scan's `Dataset::open` each
+  sample. Needs a probe that separates them before anyone reads it as a writer leak.
+
+**Open:** a full 60-minute run (and longer) to see whether L's latency and RSS keep growing
+or level off; the RSS source; what compaction (L: `Dataset::optimize`; M: the MemWAL
+compactor) does to both curves.
+
+## 5. Deliverables
+
+| D-id | scope | gate |
+|---|---|---|
+| D-WAL-1 | V1–V4 on both arms (`crates/lance-graph/tests/wal_shootout_probe.rs`, behind `lancedb-sdk`) | V1: no partial cycle on either arm, sweep brackets the boundary; V2: no acknowledged cycle lost after the replay read; V3/V4: observed behaviour recorded |
+| D-WAL-2 | cost table (`crates/lance-graph/examples/wal_shootout.rs latency`) | numbers reported with tier; no threshold |
+| D-WAL-3 | 60-minute soak, both arms, CSV + summary | drift, growth and RSS reported per arm |
+
+## 6. Fences
+
+- No production wiring: `LanceCycleWriter` keeps zero callers outside tests; nothing here
+  enables MemWAL anywhere. D-LNC-7 stays a decision for the operator.
+- lance and lancedb come from crates.io (the upstream carve-out); no new dependency beyond
+  the existing optional `lancedb-sdk` feature.
+
+## 7. D-WAL-4..7 and the decision (2026-10-15)
+
+- **D-WAL-4 (measured).** L's resident memory is Lance's session metadata cache: each
+  commit caches a manifest clone per version, and manifest v lists v fragments, so the
+  cache grows as V^2/2 (1,600 versions: 628 MiB default vs 7.8 MiB with an 8 MiB cap;
+  RSS 570 vs 52 MiB). `examples/wal_rss_probe.rs`.
+- **D-WAL-5.** `LanceCycleWriter` opens and writes on a session with an 8 MiB metadata
+  cache (`SESSION_METADATA_CACHE_BYTES`). Old manifests are read from storage when
+  reconciliation asks for them.
+- **D-WAL-6.** `contract::temporal_pov::VersionClock`: rows of an append-only lane per
+  version, eight bytes per version; a `TemporalPov` resolves to one row range in 4.2 ns on
+  v4 (a per-row scan of 400k rows: 127 µs). Not yet wired: a writer records into it after
+  each commit; seeding it from storage at open is open.
+- **D-WAL-7.** `temporal.rs` index forms (`deinterlace_indices`,
+  `local_trajectories_indices`, `local_trajectory_indices_of`); the copying forms wrap them.
+- **D-LNC-7, decided:** `LanceCycleWriter` stays the default with the capped session;
+  MemWAL stays an option. The 60-minute soak is deferred.
+
diff --git a/crates/lance-graph-contract/examples/version_clock_ns.rs b/crates/lance-graph-contract/examples/version_clock_ns.rs
new file mode 100644
index 000000000..028b61ab1
--- /dev/null
+++ b/crates/lance-graph-contract/examples/version_clock_ns.rs
@@ -0,0 +1,69 @@
+//! D-WAL-6: cost of resolving a `TemporalPov` to a row range, per call, in ns.
+//! Compared with a per-row predicate scan over the same rows (the shape of a
+//! window that filters every row).
+//!
+//! `cargo run --release -p lance-graph-contract --example version_clock_ns`
+
+use std::hint::black_box;
+use std::time::Instant;
+
+use lance_graph_contract::temporal_pov::{TemporalPov, VersionClock, VersionRange};
+
+fn main() {
+    println!(
+        "tier: avx512f={} avx2={}",
+        cfg!(target_feature = "avx512f"),
+        cfg!(target_feature = "avx2")
+    );
+    let versions = 100_000u64;
+    let rows_per_version = 4u64;
+    let mut clock = VersionClock::new(0, 0);
+    let mut row_version = Vec::with_capacity((versions * rows_per_version) as usize);
+    for v in 1..=versions {
+        for _ in 0..rows_per_version {
+            row_version.push(v);
+        }
+        clock.record(v, row_version.len() as u64).unwrap();
+    }
+    // Pseudo-random reads so the branch predictor and cache see real spread.
+    let mut x = 0x9E37_79B9_7F4A_7C15u64;
+    let povs: Vec<TemporalPov> = (0..4096)
+        .map(|_| {
+            x ^= x << 13;
+            x ^= x >> 7;
+            x ^= x << 17;
+            let a = x % versions;
+            let b = a + (x >> 32) % 1000;
+            TemporalPov::new(VersionRange::new(a, b), 0)
+        })
+        .collect();
+
+    let iters = 20_000_000usize;
+    let t = Instant::now();
+    let mut acc = 0u64;
+    for i in 0..iters {
+        let r = clock.rows_at(black_box(&povs[i & 4095])).unwrap();
+        acc = acc.wrapping_add(r.end - r.start);
+    }
+    let clock_ns = t.elapsed().as_secs_f64() * 1e9 / iters as f64;
+    black_box(acc);
+
+    let scans = 200usize;
+    let t = Instant::now();
+    let mut acc = 0u64;
+    for i in 0..scans {
+        let pov = black_box(&povs[i & 4095]);
+        acc += row_version.iter().filter(|&&v| pov.admits(v)).count() as u64;
+    }
+    let scan_ns = t.elapsed().as_secs_f64() * 1e9 / scans as f64;
+    black_box(acc);
+
+    println!(
+        "rows {} versions {}  clock.rows_at: {clock_ns:.2} ns/call  per-row scan: {:.0} ns/call ({:.2} ns/row)  clock memory: {} B",
+        row_version.len(),
+        versions,
+        scan_ns,
+        scan_ns / row_version.len() as f64,
+        (versions + 1) * 8
+    );
+}
diff --git a/crates/lance-graph-contract/src/temporal_pov.rs b/crates/lance-graph-contract/src/temporal_pov.rs
index 98ab9b0c1..46d505420 100644
--- a/crates/lance-graph-contract/src/temporal_pov.rs
+++ b/crates/lance-graph-contract/src/temporal_pov.rs
@@ -195,10 +195,256 @@ impl TemporalPov {
     }
 }
 
+/// Why [`VersionClock::record`] refused a tick.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ClockError {
+    /// The version is not newer than the clock's head.
+    NotNewer {
+        /// The clock's head.
+        head: LanceVersion,
+        /// The refused version.
+        version: LanceVersion,
+    },
+    /// The row count went down: an append-only lane never loses rows.
+    RowsWentBack {
+        /// Rows at the head.
+        head_rows: u64,
+        /// The refused count.
+        rows: u64,
+    },
+}
+
+impl core::fmt::Display for ClockError {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        match self {
+            Self::NotNewer { head, version } => {
+                write!(f, "version {version} is not newer than head {head}")
+            }
+            Self::RowsWentBack { head_rows, rows } => {
+                write!(f, "row count {rows} is below the head's {head_rows}")
+            }
+        }
+    }
+}
+
+impl std::error::Error for ClockError {}
+
+/// The version clock of one append-only lane: which rows each version holds.
+///
+/// Rows are numbered in append order. Version `v` holds rows
+/// `0..rows_through(v)`, so a [`TemporalPov`] resolves to one half-open row
+/// range by two array reads. No manifest, no per-version file list and no
+/// per-row predicate: eight bytes per version, and the rows themselves stay
+/// where they are.
+///
+/// The tick is any monotone durable coordinate: a Lance version for a
+/// one-version-per-commit writer, or a WAL entry position for a MemWAL
+/// writer, whose base table does not advance per commit.
+///
+/// The clock knows versions from its `base` on. A clock seeded on an
+/// existing store at its head cannot answer for versions before that head;
+/// those reads return `None`, never a guess.
+///
+/// Rewriting rows (compaction that reorders, deletion) is out of scope:
+/// a deleted row stays inside its range and must be masked by the caller;
+/// a reordering rewrite needs a new clock.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct VersionClock {
+    base: LanceVersion,
+    /// `ends[i]` = rows durable through version `base + i`. Non-decreasing,
+    /// never empty.
+    ends: Vec<u64>,
+}
+
+impl VersionClock {
+    /// A clock that knows `base` and its row count.
+    #[must_use]
+    pub fn new(base: LanceVersion, rows_at_base: u64) -> Self {
+        Self {
+            base,
+            ends: vec![rows_at_base],
+        }
+    }
+
+    /// The first version this clock can answer for.
+    #[inline]
+    #[must_use]
+    pub fn base(&self) -> LanceVersion {
+        self.base
+    }
+
+    /// The newest recorded version.
+    #[inline]
+    #[must_use]
+    pub fn head(&self) -> LanceVersion {
+        self.base + (self.ends.len() as u64 - 1)
+    }
+
+    /// Rows durable at the head.
+    #[inline]
+    #[must_use]
+    pub fn head_rows(&self) -> u64 {
+        // `ends` is never empty.
+        self.ends[self.ends.len() - 1]
+    }
+
+    /// Record that `version` is durable with `rows` rows in total. Versions
+    /// skipped between the head and `version` added no rows to this lane
+    /// (another operation minted them) and inherit the head's count.
+    pub fn record(&mut self, version: LanceVersion, rows: u64) -> Result<(), ClockError> {
+        let head = self.head();
+        if version <= head {
+            return Err(ClockError::NotNewer { head, version });
+        }
+        let head_rows = self.head_rows();
+        if rows < head_rows {
+            return Err(ClockError::RowsWentBack { head_rows, rows });
+        }
+        let gap = (version - head - 1) as usize;
+        self.ends.extend(core::iter::repeat_n(head_rows, gap));
+        self.ends.push(rows);
+        Ok(())
+    }
+
+    /// Rows durable through `version`: `0..rows_through(version)`. A version
+    /// past the head holds the head's rows (nothing newer exists yet).
+    /// `None` for a version before [`Self::base`].
+    #[inline]
+    #[must_use]
+    pub fn rows_through(&self, version: LanceVersion) -> Option<u64> {
+        if version < self.base {
+            return None;
+        }
+        let i = (version - self.base).min(self.ends.len() as u64 - 1) as usize;
+        Some(self.ends[i])
+    }
+
+    /// The rows a point of view admits: the rows that landed in versions
+    /// `pov.range.from .. pov.range.to`, as a half-open row range. For
+    /// [`TemporalPov::at`]`(v, _)` this is `0..rows_through(v)`.
+    #[inline]
+    #[must_use]
+    pub fn rows_at(&self, pov: &TemporalPov) -> Option<core::ops::Range<u64>> {
+        let VersionRange { from, to } = pov.range;
+        if from >= to {
+            return Some(0..0);
+        }
+        let start = if from == 0 {
+            0
+        } else {
+            self.rows_through(from - 1)?
+        };
+        let end = self.rows_through(to - 1)?;
+        Some(start..end)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    fn clock() -> VersionClock {
+        // v0: 0 rows, v1: 10, v2: 10 (no rows, e.g. an index op), v3: 25.
+        let mut c = VersionClock::new(0, 0);
+        c.record(1, 10).unwrap();
+        c.record(3, 25).unwrap();
+        c
+    }
+
+    #[test]
+    fn clock_answers_at_and_ranges() {
+        let c = clock();
+        assert_eq!(c.head(), 3);
+        let at = |v| c.rows_at(&TemporalPov::at(v, 0)).unwrap();
+        assert_eq!(at(0), 0..0);
+        assert_eq!(at(1), 0..10);
+        assert_eq!(at(2), 0..10, "the skipped version inherits the head's rows");
+        assert_eq!(at(3), 0..25);
+        assert_eq!(
+            at(99),
+            0..25,
+            "a version past the head holds the head's rows"
+        );
+        assert_eq!(at(u64::MAX), 0..25, "the latest sentinel saturates");
+        // Rows landed in versions 2..=3 only.
+        let r = c
+            .rows_at(&TemporalPov::new(VersionRange::new(2, 4), 0))
+            .unwrap();
+        assert_eq!(r, 10..25);
+        let empty = c
+            .rows_at(&TemporalPov::new(VersionRange::new(4, 4), 0))
+            .unwrap();
+        assert!(empty.is_empty());
+    }
+
+    #[test]
+    fn clock_agrees_with_a_per_row_scan() {
+        // Oracle: every row carries the version it landed at; the clock range
+        // must equal the set of rows the per-row predicate admits.
+        let mut c = VersionClock::new(0, 0);
+        let mut row_version = Vec::new();
+        let mut v = 0u64;
+        for step in 1..200u64 {
+            v += 1 + step % 3; // gaps of 0..2 skipped versions
+            let added = (step * 7) % 5; // some versions add no rows
+            for _ in 0..added {
+                row_version.push(v);
+            }
+            c.record(v, row_version.len() as u64).unwrap();
+        }
+        for from in 0..=v + 2 {
+            for to in [from, from + 1, from + 3, v / 2, v + 5] {
+                let pov = TemporalPov::new(VersionRange::new(from, to), 0);
+                let want: Vec<u64> = (0..row_version.len() as u64)
+                    .filter(|&i| pov.admits(row_version[i as usize]))
+                    .collect();
+                let got: Vec<u64> = c.rows_at(&pov).unwrap().collect();
+                assert_eq!(got, want, "from {from} to {to}");
+            }
+        }
+    }
+
+    #[test]
+    fn clock_refuses_old_versions_and_shrinking_rows() {
+        let mut c = clock();
+        assert_eq!(
+            c.record(3, 30),
+            Err(ClockError::NotNewer {
+                head: 3,
+                version: 3
+            })
+        );
+        assert_eq!(
+            c.record(4, 24),
+            Err(ClockError::RowsWentBack {
+                head_rows: 25,
+                rows: 24
+            })
+        );
+        assert_eq!(c.head(), 3, "a refused record changes nothing");
+        assert!(
+            c.record(4, 25).is_ok(),
+            "an equal count is a version with no rows"
+        );
+    }
+
+    #[test]
+    fn clock_seeded_at_a_head_does_not_answer_for_earlier_versions() {
+        let c = VersionClock::new(100, 5_000);
+        assert_eq!(c.rows_through(99), None);
+        assert_eq!(c.rows_at(&TemporalPov::at(99, 0)), None);
+        assert_eq!(c.rows_at(&TemporalPov::at(100, 0)), Some(0..5_000));
+        // A range starting after the base needs rows_through(from - 1).
+        assert_eq!(
+            c.rows_at(&TemporalPov::new(VersionRange::new(101, 102), 0)),
+            Some(5_000..5_000)
+        );
+        assert_eq!(
+            c.rows_at(&TemporalPov::new(VersionRange::new(100, 101), 0)),
+            None
+        );
+    }
+
     #[test]
     fn version_range_basic_contains_and_len() {
         let r = VersionRange::new(10, 20);
diff --git a/crates/lance-graph-planner/src/temporal.rs b/crates/lance-graph-planner/src/temporal.rs
index 300e9b6f4..1e4523810 100644
--- a/crates/lance-graph-planner/src/temporal.rs
+++ b/crates/lance-graph-planner/src/temporal.rs
@@ -363,15 +363,39 @@ pub trait DeinterlaceRow {
 /// type-visible here; their bodies are trivial (`hlc_tick == None`, `NoDeps`)
 /// until the cluster bus and the SPO frontends land. No signature changes when
 /// they do.
+///
+/// **Copies every admitted row.** Prefer [`deinterlace_indices`], which returns
+/// positions into `rows` and leaves the rows where they are.
 #[must_use]
 pub fn deinterlace<R, D>(rows: &[R], v_ref: &QueryReference, deps: &D) -> Vec<R>
 where
     R: DeinterlaceRow + Clone,
     D: DependsClosure,
 {
-    let mut out: Vec<R> = rows
-        .iter()
-        .filter(|r| {
+    deinterlace_indices(rows, v_ref, deps)
+        .into_iter()
+        .map(|i| rows[i as usize].clone())
+        .collect()
+}
+
+/// [`deinterlace`] without copying rows: the positions in `rows` of the
+/// dispatchable rows, in deinterlace order. Four bytes per admitted row.
+///
+/// # Panics
+/// If `rows` holds more than `u32::MAX` rows.
+#[must_use]
+pub fn deinterlace_indices<R, D>(rows: &[R], v_ref: &QueryReference, deps: &D) -> Vec<u32>
+where
+    R: DeinterlaceRow,
+    D: DependsClosure,
+{
+    assert!(
+        u32::try_from(rows.len()).is_ok(),
+        "more than u32::MAX rows; the index forms address rows with u32"
+    );
+    let mut out: Vec<u32> = (0..rows.len() as u32)
+        .filter(|&i| {
+            let r = &rows[i as usize];
             classify_ready(
                 r.subject(),
                 r.lance_version(),
@@ -381,13 +405,14 @@ where
             )
             .dispatchable(v_ref.mode)
         })
-        .cloned()
         .collect();
     // Falls back to the row's own lance_version when no HLC tick is present
     // (single-server / legacy rows) — NOT to 0, which would force every
     // missing-HLC row ahead of all HLC rows regardless of its version (Codex
     // P2 on #468; honors the documented "fallback to lance_version").
-    out.sort_by_key(|r| {
+    // Stable sort: equal keys keep their input order, as before.
+    out.sort_by_key(|&i| {
+        let r = &rows[i as usize];
         (
             r.hlc_tick().unwrap_or_else(|| r.lance_version()),
             r.lance_version(),
@@ -441,14 +466,37 @@ pub trait LocalCausalRow {
 /// neighbours are removed from its timeline, and each timeline is put back into
 /// the owner's own cast order (the crash-replay order). Returns a `BTreeMap` so
 /// owners iterate deterministically.
+///
+/// **Copies every row.** Prefer [`local_trajectories_indices`].
 #[must_use]
 pub fn local_trajectories<R: LocalCausalRow + Clone>(global: &[R]) -> BTreeMap<MailboxId, Vec<R>> {
-    let mut by_owner: BTreeMap<MailboxId, Vec<R>> = BTreeMap::new();
-    for row in global {
-        by_owner.entry(row.owner()).or_default().push(row.clone());
+    local_trajectories_indices(global)
+        .into_iter()
+        .map(|(owner, ix)| {
+            (
+                owner,
+                ix.into_iter().map(|i| global[i as usize].clone()).collect(),
+            )
+        })
+        .collect()
+}
+
+/// [`local_trajectories`] without copying rows: per owner, the positions in
+/// `global` of that owner's rows in `cast_seq` order.
+#[must_use]
+pub fn local_trajectories_indices<R: LocalCausalRow>(
+    global: &[R],
+) -> BTreeMap<MailboxId, Vec<u32>> {
+    assert!(
+        u32::try_from(global.len()).is_ok(),
+        "more than u32::MAX rows; the index forms address rows with u32"
+    );
+    let mut by_owner: BTreeMap<MailboxId, Vec<u32>> = BTreeMap::new();
+    for (i, row) in global.iter().enumerate() {
+        by_owner.entry(row.owner()).or_default().push(i as u32);
     }
-    for rows in by_owner.values_mut() {
-        rows.sort_by_key(LocalCausalRow::cast_seq);
+    for ix in by_owner.values_mut() {
+        ix.sort_by_key(|&i| global[i as usize].cast_seq());
     }
     by_owner
 }
@@ -457,14 +505,28 @@ pub fn local_trajectories<R: LocalCausalRow + Clone>(global: &[R]) -> BTreeMap<M
 /// owner. Filters the interleaved global stream to that owner's rows and orders
 /// them by `cast_seq`. Equivalent to `local_trajectories(global).remove(&owner)`
 /// but without materialising the other owners' chains.
+///
+/// **Copies the owner's rows.** Prefer [`local_trajectory_indices_of`].
 #[must_use]
 pub fn local_trajectory_of<R: LocalCausalRow + Clone>(global: &[R], owner: MailboxId) -> Vec<R> {
-    let mut chain: Vec<R> = global
-        .iter()
-        .filter(|r| r.owner() == owner)
-        .cloned()
+    local_trajectory_indices_of(global, owner)
+        .into_iter()
+        .map(|i| global[i as usize].clone())
+        .collect()
+}
+
+/// [`local_trajectory_of`] without copying rows: the positions in `global` of
+/// `owner`'s rows in `cast_seq` order.
+#[must_use]
+pub fn local_trajectory_indices_of<R: LocalCausalRow>(global: &[R], owner: MailboxId) -> Vec<u32> {
+    assert!(
+        u32::try_from(global.len()).is_ok(),
+        "more than u32::MAX rows; the index forms address rows with u32"
+    );
+    let mut chain: Vec<u32> = (0..global.len() as u32)
+        .filter(|&i| global[i as usize].owner() == owner)
         .collect();
-    chain.sort_by_key(LocalCausalRow::cast_seq);
+    chain.sort_by_key(|&i| global[i as usize].cast_seq());
     chain
 }
 
@@ -631,6 +693,72 @@ mod tests {
         );
     }
 
+    /// D-WAL-7: the index forms select exactly the rows the copying forms
+    /// return, in the same order, and never clone a row.
+    #[test]
+    fn index_forms_match_the_copying_forms_without_cloning() {
+        use std::cell::Cell;
+        thread_local!(static CLONES: Cell<usize> = const { Cell::new(0) });
+        #[derive(Debug, PartialEq)]
+        struct Counted(Row);
+        impl Clone for Counted {
+            fn clone(&self) -> Self {
+                CLONES.with(|c| c.set(c.get() + 1));
+                Counted(self.0.clone())
+            }
+        }
+        impl DeinterlaceRow for Counted {
+            fn subject(&self) -> &str {
+                self.0.subject()
+            }
+            fn lance_version(&self) -> LanceVersion {
+                self.0.v
+            }
+            fn knowable_from(&self) -> LanceVersion {
+                self.0.knowable
+            }
+            fn hlc_tick(&self) -> Option<u64> {
+                self.0.hlc
+            }
+        }
+        let rows: Vec<Counted> = (0..40u64)
+            .map(|i| {
+                Counted(Row::new(
+                    (i * 37) % 50,
+                    i % 7,
+                    (i % 3 == 0).then_some(i * 11 % 60),
+                ))
+            })
+            .collect();
+        let v_ref = QueryReference::at(30, 0);
+        CLONES.with(|c| c.set(0));
+        let ix = deinterlace_indices(&rows, &v_ref, &NoDeps);
+        assert_eq!(CLONES.with(Cell::get), 0, "the index form clones nothing");
+        let copied = deinterlace(&rows, &v_ref, &NoDeps);
+        assert!(
+            !ix.is_empty() && ix.len() < rows.len(),
+            "the filter admits some, not all"
+        );
+        let via_ix: Vec<&Counted> = ix.iter().map(|&i| &rows[i as usize]).collect();
+        assert_eq!(via_ix, copied.iter().collect::<Vec<_>>());
+
+        let global: Vec<CausalRow> = (0..30u64)
+            .map(|i| CausalRow {
+                owner: (i % 3) as MailboxId,
+                seq: (i * 7) % 11,
+                tag: "x",
+            })
+            .collect();
+        let by_ix = local_trajectories_indices(&global);
+        let by_copy = local_trajectories(&global);
+        assert_eq!(by_ix.len(), by_copy.len());
+        for (owner, ix) in &by_ix {
+            let got: Vec<&CausalRow> = ix.iter().map(|&i| &global[i as usize]).collect();
+            assert_eq!(got, by_copy[owner].iter().collect::<Vec<_>>());
+            assert_eq!(*ix, local_trajectory_indices_of(&global, *owner));
+        }
+    }
+
     #[test]
     fn for_rung_policy() {
         assert_eq!(EpistemicMode::for_rung(0), EpistemicMode::Strict);
diff --git a/crates/lance-graph/Cargo.toml b/crates/lance-graph/Cargo.toml
index 2a909efb3..73d0a25bd 100644
--- a/crates/lance-graph/Cargo.toml
+++ b/crates/lance-graph/Cargo.toml
@@ -206,3 +206,14 @@ object_store = { workspace = true }
 name = "w0b_corpus_census"
 path = "examples/w0b_corpus_census.rs"
 test = true
+
+# D-WAL (plan `.claude/plans/wal-shootout-v1.md`): `LanceCycleWriter` against the
+# lancedb 0.40 MemWAL write path. Needs the optional lancedb dependency. The
+# example is gated here; the test `tests/wal_shootout_probe.rs` gates itself with
+# `#![cfg(feature = "lancedb-sdk")]`, because CI runs `cargo test --test '*'`
+# and cargo refuses an explicitly selected target whose required features are off.
+
+[[example]]
+name = "wal_shootout"
+path = "examples/wal_shootout.rs"
+required-features = ["lancedb-sdk"]
diff --git a/crates/lance-graph/examples/wal_rss_probe.rs b/crates/lance-graph/examples/wal_rss_probe.rs
new file mode 100644
index 000000000..46c23718a
--- /dev/null
+++ b/crates/lance-graph/examples/wal_rss_probe.rs
@@ -0,0 +1,99 @@
+//! D-WAL-4 — where the L arm's resident memory goes.
+//!
+//! Hypothesis: every Lance commit inserts a full clone of the new manifest into
+//! the session metadata cache (`record_successful_commit`, lance 13
+//! `io/commit.rs`). An append-only store adds one fragment per version, so the
+//! manifest of version v lists v fragments, and the cache holds
+//! sum(v) = V^2/2 fragment entries until it reaches its 1 GiB default cap.
+//!
+//! Two arms append the same batches. `default` uses the default session;
+//! `capped` passes a session whose metadata cache is limited to 8 MiB.
+//! Prediction if the hypothesis holds: `default` cache bytes grow
+//! super-linearly and RSS follows; `capped` cache stays at or under 8 MiB and
+//! RSS stays flat.
+//!
+//! ```text
+//! cargo run --release -p lance-graph --example wal_rss_probe -- <scratch-dir> <arm> <versions>
+//! ```
+//! Run each arm in its own process so RSS is not shared.
+
+use std::sync::Arc;
+
+use arrow_array::{Int64Array, RecordBatch, RecordBatchIterator};
+use arrow_schema::{DataType, Field, Schema};
+use lance::dataset::{Dataset, WriteMode, WriteParams};
+use lance::session::Session;
+
+fn rss_mib() -> f64 {
+    std::fs::read_to_string("/proc/self/status")
+        .ok()
+        .and_then(|s| {
+            s.lines()
+                .find(|l| l.starts_with("VmRSS:"))
+                .and_then(|l| l.split_whitespace().nth(1)?.parse::<u64>().ok())
+        })
+        .unwrap_or(0) as f64
+        / 1024.0
+}
+
+#[tokio::main(flavor = "multi_thread")]
+async fn main() {
+    let args: Vec<String> = std::env::args().collect();
+    let dir = &args[1];
+    let arm = args[2].as_str();
+    let versions: u64 = args[3].parse().expect("versions");
+    let path = format!("{dir}/rss-{arm}.lance");
+    let _ = std::fs::remove_dir_all(&path);
+
+    let session = match arm {
+        "default" => None,
+        "capped" => Some(Arc::new(Session::new(
+            0,
+            8 << 20,
+            Arc::new(Default::default()),
+        ))),
+        other => panic!("arm {other}"),
+    };
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("cycle", DataType::Int64, false),
+        Field::new("v", DataType::Int64, false),
+    ]));
+    let batch = |c: u64| {
+        RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(Int64Array::from(vec![c as i64; 256])),
+                Arc::new(Int64Array::from_iter_values(0..256)),
+            ],
+        )
+        .unwrap()
+    };
+
+    let params = |mode| WriteParams {
+        mode,
+        session: session.clone(),
+        ..Default::default()
+    };
+    let reader = RecordBatchIterator::new(vec![Ok(batch(1))], schema.clone());
+    let mut ds = Dataset::write(reader, &path, Some(params(WriteMode::Create)))
+        .await
+        .expect("create");
+
+    println!("arm,version,fragments,cache_mib,cache_items,rss_mib");
+    for c in 2..=versions {
+        let reader = RecordBatchIterator::new(vec![Ok(batch(c))], schema.clone());
+        ds.append(reader, None).await.expect("append");
+        if c.is_power_of_two() || c % 200 == 0 || c == versions {
+            let s = ds.session();
+            println!(
+                "{arm},{},{},{:.1},{},{:.1}",
+                ds.version().version,
+                ds.get_fragments().len(),
+                s.size_bytes() as f64 / (1 << 20) as f64,
+                s.approx_num_items(),
+                rss_mib(),
+            );
+        }
+    }
+    let _ = std::fs::remove_dir_all(&path);
+}
diff --git a/crates/lance-graph/examples/wal_shootout.rs b/crates/lance-graph/examples/wal_shootout.rs
new file mode 100644
index 000000000..a229936c4
--- /dev/null
+++ b/crates/lance-graph/examples/wal_shootout.rs
@@ -0,0 +1,216 @@
+//! D-WAL-2 / D-WAL-3 — cost and soak of `LanceCycleWriter` (L) against the
+//! lancedb 0.40 MemWAL write path (M). Plan: `.claude/plans/wal-shootout-v1.md`.
+//!
+//! ```text
+//! # cost table (D-WAL-2)
+//! cargo run --release -p lance-graph --features lancedb-sdk --example wal_shootout -- latency <scratch-dir>
+//! # soak, one process per arm (D-WAL-3)
+//! cargo run --release -p lance-graph --features lancedb-sdk --example wal_shootout -- \
+//!     soak <L|M> <dir> <minutes> <rows-per-cycle> <interval-ms> <csv>
+//! ```
+
+#[path = "../tests/wal_common/mod.rs"]
+mod wal_common;
+
+use std::io::Write;
+use std::path::Path;
+use std::time::{Duration, Instant};
+
+use lance_graph::graph::cycle_sink::LanceCycleWriter;
+use wal_common::*;
+
+fn pct(sorted: &[f64], p: f64) -> f64 {
+    if sorted.is_empty() {
+        return f64::NAN;
+    }
+    sorted[((sorted.len() - 1) as f64 * p).round() as usize]
+}
+
+fn rss_kib() -> u64 {
+    std::fs::read_to_string("/proc/self/status")
+        .ok()
+        .and_then(|s| {
+            s.lines()
+                .find(|l| l.starts_with("VmRSS:"))
+                .and_then(|l| l.split_whitespace().nth(1)?.parse().ok())
+        })
+        .unwrap_or(0)
+}
+
+fn free_bytes(dir: &Path) -> u64 {
+    // statvfs through `df` keeps this dependency-free.
+    let out = std::process::Command::new("df")
+        .args(["-B1", "--output=avail"])
+        .arg(dir)
+        .output()
+        .ok();
+    out.and_then(|o| String::from_utf8(o.stdout).ok())
+        .and_then(|s| s.lines().nth(1)?.trim().parse().ok())
+        .unwrap_or(u64::MAX)
+}
+
+/// One arm's writer.
+enum Arm {
+    L(Box<LanceCycleWriter>, String),
+    M(lancedb::Table),
+}
+
+impl Arm {
+    async fn open(kind: &str, dir: &str) -> Self {
+        match kind {
+            "L" => {
+                let p = format!("{dir}/cycles.lance");
+                Arm::L(
+                    Box::new(LanceCycleWriter::open(p.clone()).await.expect("open L")),
+                    p,
+                )
+            }
+            k if k.starts_with('M') => {
+                let flush: u64 = k[1..].parse().unwrap_or(100);
+                Arm::M(m_create(dir, flush).await)
+            }
+            other => panic!("arm {other}"),
+        }
+    }
+
+    async fn commit(&mut self, cycle: u64, n: usize) {
+        match self {
+            Arm::L(w, _) => l_commit(w, cycle, n).await,
+            Arm::M(t) => m_put(t, &frozen(cycle, 0, n)).await,
+        }
+    }
+
+    /// Rows readable now, and the scan time. L: a fresh open at the latest
+    /// version. M: the LSM read of this process (memtable + generations).
+    async fn scan(&self) -> (usize, Duration) {
+        let t0 = Instant::now();
+        let rows = match self {
+            Arm::L(_, p) => l_rows_by_cycle(p).await.unwrap_or_default().values().sum(),
+            Arm::M(t) => m_rows_by_cycle(t, true).await.values().sum(),
+        };
+        (rows, t0.elapsed())
+    }
+
+    /// Lance versions of the L store (M: versions of the base table, which the
+    /// MemWAL path does not advance until compaction).
+    async fn versions(&self) -> u64 {
+        match self {
+            Arm::L(w, _) => w.head().0,
+            Arm::M(t) => t.version().await.unwrap_or(0),
+        }
+    }
+}
+
+async fn latency(root: &str) {
+    println!(
+        "tier: avx512f={} | host I/O bound: numbers are per cycle (µs) and per stored row (ns)",
+        cfg!(target_feature = "avx512f")
+    );
+    println!("arm  N      cycles  p50_us    p99_us    ns/row   rows      MiB     files  versions  scan_ms");
+    for (n, cycles) in [(64usize, 200u64), (1024, 200), (8192, 40)] {
+        for kind in ["L", "M10", "M100"] {
+            let dir = format!("{root}/lat-{kind}-{n}");
+            let _ = std::fs::remove_dir_all(&dir);
+            std::fs::create_dir_all(&dir).unwrap();
+            let mut arm = Arm::open(kind, &dir).await;
+            let mut lat = Vec::with_capacity(cycles as usize);
+            for c in 1..=cycles {
+                let t0 = Instant::now();
+                arm.commit(c, n).await;
+                lat.push(t0.elapsed().as_secs_f64() * 1e6);
+            }
+            lat.sort_by(|a, b| a.partial_cmp(b).unwrap());
+            let (rows, scan) = arm.scan().await;
+            assert_eq!(rows, rows_per_cycle(n) * cycles as usize, "{kind} rows");
+            let (bytes, files) = disk_usage(Path::new(&dir));
+            let p50 = pct(&lat, 0.5);
+            println!(
+                "{kind:<4} {n:<6} {cycles:<7} {p50:<9.0} {:<9.0} {:<8.1} {rows:<9} {:<7.1} {files:<6} {:<9} {:.1}",
+                pct(&lat, 0.99),
+                p50 * 1e3 / rows_per_cycle(n) as f64,
+                bytes as f64 / (1 << 20) as f64,
+                arm.versions().await,
+                scan.as_secs_f64() * 1e3,
+            );
+            drop(arm);
+            let _ = std::fs::remove_dir_all(&dir);
+        }
+    }
+}
+
+async fn soak(kind: &str, dir: &str, minutes: u64, n: usize, interval_ms: u64, csv: &str) {
+    std::fs::create_dir_all(dir).unwrap();
+    let mut out = std::fs::OpenOptions::new()
+        .create(true)
+        .append(true)
+        .open(csv)
+        .expect("csv");
+    writeln!(
+        out,
+        "t_s,arm,cycles,commit_p50_us,commit_p99_us,commit_max_us,disk_mib,files,versions,scan_rows,scan_ms,rss_mib"
+    )
+    .unwrap();
+    let mut arm = Arm::open(kind, dir).await;
+    let start = Instant::now();
+    let end = start + Duration::from_secs(minutes * 60);
+    let mut next_sample = start + Duration::from_secs(30);
+    let mut window: Vec<f64> = Vec::new();
+    let mut cycle = 0u64;
+    while Instant::now() < end {
+        let tick = Instant::now();
+        cycle += 1;
+        arm.commit(cycle, n).await;
+        window.push(tick.elapsed().as_secs_f64() * 1e6);
+        if Instant::now() >= next_sample {
+            window.sort_by(|a, b| a.partial_cmp(b).unwrap());
+            let (bytes, files) = disk_usage(Path::new(dir));
+            let (rows, scan) = arm.scan().await;
+            writeln!(
+                out,
+                "{:.0},{kind},{cycle},{:.0},{:.0},{:.0},{:.1},{files},{},{rows},{:.1},{:.1}",
+                start.elapsed().as_secs_f64(),
+                pct(&window, 0.5),
+                pct(&window, 0.99),
+                window.last().copied().unwrap_or(f64::NAN),
+                bytes as f64 / (1 << 20) as f64,
+                arm.versions().await,
+                scan.as_secs_f64() * 1e3,
+                rss_kib() as f64 / 1024.0,
+            )
+            .unwrap();
+            out.flush().unwrap();
+            window.clear();
+            next_sample += Duration::from_secs(30);
+            if free_bytes(Path::new(dir)) < 1 << 30 {
+                writeln!(out, "# stopped early: under 1 GiB free").unwrap();
+                break;
+            }
+        }
+        let spent = tick.elapsed();
+        if spent < Duration::from_millis(interval_ms) {
+            tokio::time::sleep(Duration::from_millis(interval_ms) - spent).await;
+        }
+    }
+}
+
+#[tokio::main(flavor = "multi_thread")]
+async fn main() {
+    let args: Vec<String> = std::env::args().collect();
+    match args.get(1).map(String::as_str) {
+        Some("latency") => latency(args.get(2).expect("scratch dir")).await,
+        Some("soak") => {
+            soak(
+                &args[2],
+                &args[3],
+                args[4].parse().expect("minutes"),
+                args[5].parse().expect("rows"),
+                args[6].parse().expect("interval ms"),
+                &args[7],
+            )
+            .await
+        }
+        _ => eprintln!(
+            "usage: wal_shootout latency <dir> | soak <L|M> <dir> <min> <rows> <interval_ms> <csv>"
+        ),
+    }
+}
diff --git a/crates/lance-graph/src/graph/cycle_sink.rs b/crates/lance-graph/src/graph/cycle_sink.rs
index 5613ff8ae..d9f80ed54 100644
--- a/crates/lance-graph/src/graph/cycle_sink.rs
+++ b/crates/lance-graph/src/graph/cycle_sink.rs
@@ -86,7 +86,9 @@ use arrow_array::{
 };
 use arrow_schema::{DataType, Field, Schema, SchemaRef};
 use futures::TryStreamExt;
+use lance::dataset::builder::DatasetBuilder;
 use lance::dataset::{Dataset, WriteMode, WriteParams};
+use lance::session::Session;
 
 use lance_graph_contract::collapse_gate::MailboxId;
 use lance_graph_contract::kanban::{ExecTarget, KanbanColumn, KanbanMove};
@@ -143,6 +145,20 @@ pub fn cycle_store_schema() -> SchemaRef {
     ]))
 }
 
+/// Byte cap for the metadata cache of the writer's Lance session (D-WAL-5).
+/// Lance caches one manifest clone per committed version; without a cap that
+/// grows as V^2/2 in an append-only store (D-WAL-4). Old versions are then
+/// read from storage when asked for, which only reconciliation does.
+pub const SESSION_METADATA_CACHE_BYTES: usize = 8 << 20;
+
+/// Open `path` on the writer's capped session.
+async fn open_with(path: &str, session: &Arc<Session>) -> lance::Result<Dataset> {
+    DatasetBuilder::from_uri(path)
+        .with_session(session.clone())
+        .load()
+        .await
+}
+
 /// The sole owned application writer over one Lance cycle store.
 ///
 /// Deliberately **non-`Clone`**: constructing a second writer over the same
@@ -157,6 +173,9 @@ pub struct LanceCycleWriter {
     /// NEVER degrades back to `None` — a store that existed cannot become
     /// "empty" again (see [`reopen`](Self::reopen)).
     ds: Option<Dataset>,
+    /// The Lance session every open and write of this writer uses, with its
+    /// metadata cache capped at [`SESSION_METADATA_CACHE_BYTES`] (D-WAL-5).
+    session: Arc<Session>,
     /// `Dataset::open` count — startup + ambiguity-resolution ONLY. The
     /// normal-path invariant (zero post-success reopens) is instrumented here.
     opens: AtomicU64,
@@ -319,7 +338,12 @@ impl LanceCycleWriter {
         let dataset_path = path.into();
         let claim = WriterClaim::acquire(store_identity(&dataset_path))?;
         let opens = AtomicU64::new(0);
-        let ds = match Dataset::open(&dataset_path).await {
+        let session = Arc::new(Session::new(
+            0,
+            SESSION_METADATA_CACHE_BYTES,
+            Arc::new(Default::default()),
+        ));
+        let ds = match open_with(&dataset_path, &session).await {
             Ok(ds) => {
                 opens.fetch_add(1, Ordering::Relaxed);
                 Self::guard_schema(&dataset_path, &ds)?;
@@ -331,6 +355,7 @@ impl LanceCycleWriter {
         let mut w = Self {
             dataset_path,
             ds,
+            session,
             opens,
             reconcile_scans: AtomicU64::new(0),
             committed_through: None,
@@ -622,7 +647,7 @@ impl LanceCycleWriter {
                 Ok(())
             };
         }
-        match Dataset::open(&self.dataset_path).await {
+        match open_with(&self.dataset_path, &self.session).await {
             Ok(ds) => {
                 self.opens.fetch_add(1, Ordering::Relaxed);
                 self.ds = Some(ds);
@@ -677,6 +702,7 @@ impl LanceCycleWriter {
                 &self.dataset_path,
                 Some(WriteParams {
                     mode: WriteMode::Create,
+                    session: Some(self.session.clone()),
                     ..Default::default()
                 }),
             )
@@ -696,6 +722,13 @@ impl LanceCycleWriter {
     pub fn reconcile_scans(&self) -> u64 {
         self.reconcile_scans.load(Ordering::Relaxed)
     }
+
+    /// Bytes held by the Lance session caches of this writer's handle
+    /// (metadata and index caches). `0` before the store exists.
+    #[must_use]
+    pub fn session_cache_bytes(&self) -> u64 {
+        self.ds.as_ref().map_or(0, |d| d.session().size_bytes())
+    }
 }
 
 impl LanceCycleWriter {
@@ -712,6 +745,7 @@ impl LanceCycleWriter {
                 &self.dataset_path,
                 Some(WriteParams {
                     mode: WriteMode::Create,
+                    session: Some(self.session.clone()),
                     ..Default::default()
                 }),
             )
@@ -1322,6 +1356,58 @@ mod tests {
         );
     }
 
+    /// D-WAL-5: every Lance commit caches a clone of the new manifest, and in
+    /// this append-only store manifest v lists v fragments, so an uncapped
+    /// session grows as V^2/2 (D-WAL-4). The writer's session caps it.
+    /// Falsifier: 400 commits push the default (1 GiB) session well past the
+    /// cap; with the cap the cache stays at or under it.
+    #[tokio::test]
+    async fn session_cache_stays_under_the_cap_over_many_commits() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("cycles.lance");
+        let mut w = LanceCycleWriter::open(path.to_str().unwrap())
+            .await
+            .unwrap();
+        for c in 1..=400u64 {
+            persist_cycle(
+                &mut w,
+                CycleFrame::new(CycleId(c), DatasetVersion(c - 1)),
+                vec![artifact(c, c, 42, c)],
+            )
+            .await
+            .unwrap();
+        }
+        assert_eq!(w.head(), DatasetVersion(400));
+        let bytes = w.session_cache_bytes();
+        assert!(bytes > 0, "the session cache is in use");
+        assert!(
+            bytes <= SESSION_METADATA_CACHE_BYTES as u64,
+            "session cache {bytes} B exceeds the cap {SESSION_METADATA_CACHE_BYTES} B"
+        );
+
+        // The open path too: a reopened writer appends on the same capped
+        // session, not on a default one.
+        drop(w);
+        let mut w = LanceCycleWriter::open(path.to_str().unwrap())
+            .await
+            .unwrap();
+        for c in 401..=800u64 {
+            persist_cycle(
+                &mut w,
+                CycleFrame::new(CycleId(c), DatasetVersion(c - 1)),
+                vec![artifact(c, c, 42, c)],
+            )
+            .await
+            .unwrap();
+        }
+        assert_eq!(w.head(), DatasetVersion(800));
+        let bytes = w.session_cache_bytes();
+        assert!(
+            bytes <= SESSION_METADATA_CACHE_BYTES as u64,
+            "after reopen: session cache {bytes} B exceeds the cap"
+        );
+    }
+
     /// The one-writer topology is ENFORCED in-process: a second live writer on
     /// the same path is refused at open; dropping the first frees the path.
     /// (Cross-process exclusivity remains a deployment lease — documented.)
diff --git a/crates/lance-graph/tests/wal_common/mod.rs b/crates/lance-graph/tests/wal_common/mod.rs
new file mode 100644
index 000000000..1c3a06aa7
--- /dev/null
+++ b/crates/lance-graph/tests/wal_common/mod.rs
@@ -0,0 +1,264 @@
+//! Shared harness for the WAL shootout (plan `.claude/plans/wal-shootout-v1.md`):
+//! the same logical cycle written by two arms.
+//!
+//! - **L** — `LanceCycleWriter::commit_cycle`, one Lance append per cycle.
+//! - **M** — a lancedb 0.40 table with an `LsmWriteSpec`, written by the upsert
+//!   form of `merge_insert`, which lancedb dispatches to the MemWAL `ShardWriter`.
+//!
+//! M stores the rows `LanceCycleWriter` stores (one frame row, one landing row
+//! per cast, one image row per distinct `row` with its 512-byte payload) plus a
+//! `pk` column, because MemWAL dedups by an unenforced primary key.
+#![allow(dead_code)]
+
+use std::collections::{BTreeMap, HashMap};
+use std::sync::Arc;
+
+use arrow_array::{
+    builder::{FixedSizeBinaryBuilder, UInt32Builder, UInt8Builder},
+    Array, Int64Array, RecordBatch, RecordBatchIterator, UInt32Array, UInt64Array, UInt8Array,
+};
+use arrow_schema::{DataType, Field, Schema, SchemaRef};
+use futures::TryStreamExt;
+use lance_graph::graph::cycle_sink::{
+    cycle_store_schema, LanceCycleWriter, EPISODIC_WITNESS_BYTES,
+};
+use lance_graph_contract::scheduler::DatasetVersion;
+use lance_graph_planner::persist_sink::{CycleFrame, CycleId, DetachedCycleBatch, SweepSlot};
+use lancedb::query::{ExecutableQuery, QueryBase, Select};
+use lancedb::table::LsmWriteSpec;
+
+/// The lancedb table name used by every M directory.
+pub const M_TABLE: &str = "cycles";
+
+/// Rows one cycle of `n` casts (distinct rows) occupies on either arm:
+/// one frame row, `n` landing rows, `n` image rows.
+pub fn rows_per_cycle(n: usize) -> usize {
+    1 + 2 * n
+}
+
+/// `n` artifact casts for `cycle`, one per distinct row, each with a
+/// deterministic 512-byte payload. `stream_position` is monotonic across
+/// cycles, as `SweepSlot` requires.
+pub fn casts(cycle: u64, n: usize) -> Vec<SweepSlot> {
+    (0..n)
+        .map(|i| {
+            let mut payload = vec![0u8; EPISODIC_WITNESS_BYTES];
+            payload[..8].copy_from_slice(&cycle.to_le_bytes());
+            payload[8..16].copy_from_slice(&(i as u64).to_le_bytes());
+            for (j, b) in payload[16..].iter_mut().enumerate() {
+                *b = (cycle as usize ^ i ^ j) as u8;
+            }
+            SweepSlot {
+                cycle: CycleId(cycle),
+                stream_position: (cycle << 32) | i as u64,
+                owner: 1,
+                row: i as u64,
+                paired_move: None,
+                payload,
+            }
+        })
+        .collect()
+}
+
+/// The frozen cycle both arms write.
+pub fn frozen(cycle: u64, base_version: u64, n: usize) -> DetachedCycleBatch {
+    DetachedCycleBatch::freeze(
+        CycleFrame::new(CycleId(cycle), DatasetVersion(base_version)),
+        casts(cycle, n),
+    )
+}
+
+// ─────────────────────────────── arm L ───────────────────────────────
+
+/// Commit one cycle of `n` casts on top of the writer's current head.
+pub async fn l_commit(w: &mut LanceCycleWriter, cycle: u64, n: usize) {
+    use lance_graph_planner::persist_sink::WalSink;
+    let batch = frozen(cycle, w.head().0, n);
+    w.commit_cycle(batch).await.expect("L commit_cycle");
+}
+
+/// Rows per cycle in the L store at its latest version (`None` if the store
+/// does not exist yet).
+pub async fn l_rows_by_cycle(path: &str) -> Option<BTreeMap<u64, usize>> {
+    let ds = match lance::dataset::Dataset::open(path).await {
+        Ok(ds) => ds,
+        Err(lance::Error::DatasetNotFound { .. }) => return None,
+        Err(e) => panic!("open {path}: {e}"),
+    };
+    let mut scan = ds.scan();
+    scan.project(&["cycle"]).expect("project");
+    let mut stream = scan.try_into_stream().await.expect("scan");
+    let mut out = BTreeMap::new();
+    while let Some(b) = stream.try_next().await.expect("scan batch") {
+        count_cycles(&b, &mut out);
+    }
+    Some(out)
+}
+
+// ─────────────────────────────── arm M ───────────────────────────────
+
+/// The M schema: the cycle-store schema plus `pk`.
+pub fn m_schema() -> SchemaRef {
+    let mut fields: Vec<Field> = cycle_store_schema()
+        .fields()
+        .iter()
+        .map(|f| f.as_ref().clone())
+        .collect();
+    fields.push(Field::new("pk", DataType::Int64, false));
+    Arc::new(Schema::new(fields))
+}
+
+/// Create the M table in `dir` with its primary key and an unsharded write
+/// spec. `flush_ms` is the WAL flush interval the shard writers will use.
+pub async fn m_create(dir: &str, flush_ms: u64) -> lancedb::Table {
+    let db = lancedb::connect(dir).execute().await.expect("connect");
+    let t = db
+        .create_empty_table(M_TABLE, m_schema())
+        .execute()
+        .await
+        .expect("create_empty_table");
+    t.set_unenforced_primary_key(["pk"])
+        .await
+        .expect("primary key");
+    let defaults: HashMap<String, String> = [
+        ("durable_write".to_string(), "true".to_string()),
+        (
+            "max_wal_flush_interval_ms".to_string(),
+            flush_ms.to_string(),
+        ),
+    ]
+    .into_iter()
+    .collect();
+    t.set_lsm_write_spec(
+        LsmWriteSpec::unsharded()
+            .with_maintained_indexes(Vec::new())
+            .with_writer_config_defaults(defaults),
+    )
+    .await
+    .expect("set_lsm_write_spec");
+    t
+}
+
+/// Open the existing M table in `dir`.
+pub async fn m_open(dir: &str) -> lancedb::Table {
+    let db = lancedb::connect(dir).execute().await.expect("connect");
+    db.open_table(M_TABLE).execute().await.expect("open_table")
+}
+
+/// The M rows of one frozen cycle, in the layout `LanceCycleWriter` writes.
+pub fn m_batch(batch: &DetachedCycleBatch) -> RecordBatch {
+    let cycle = batch.frame.cycle.0;
+    let base = batch.frame.base_version.0;
+    let n = 1 + batch.landings.len() + batch.image.len();
+    let mut kind = UInt8Builder::with_capacity(n);
+    let mut stream = Vec::with_capacity(n);
+    let mut owner = UInt32Builder::with_capacity(n);
+    let mut row = Vec::with_capacity(n);
+    let mut payload = FixedSizeBinaryBuilder::with_capacity(n, EPISODIC_WITNESS_BYTES as i32);
+    // frame
+    kind.append_value(0);
+    stream.push(0);
+    owner.append_value(0);
+    row.push(0);
+    payload.append_null();
+    // landings (metadata only)
+    for s in &batch.landings {
+        kind.append_value(1);
+        stream.push(s.stream_position);
+        owner.append_value(s.owner);
+        row.push(s.row);
+        payload.append_null();
+    }
+    // image rows (final payload)
+    for (r, p) in &batch.image {
+        kind.append_value(2);
+        stream.push(0);
+        owner.append_value(0);
+        row.push(*r);
+        payload.append_value(p).expect("512-byte payload");
+    }
+    let nulls_u32 = || Arc::new(UInt32Array::from(vec![None::<u32>; n])) as Arc<dyn Array>;
+    let nulls_u8 = || Arc::new(UInt8Array::from(vec![None::<u8>; n])) as Arc<dyn Array>;
+    // lancedb accepts Int64 (not UInt64) as a primary key.
+    let pk: Vec<i64> = (0..n as u64).map(|i| ((cycle << 24) | i) as i64).collect();
+    RecordBatch::try_new(
+        m_schema(),
+        vec![
+            Arc::new(kind.finish()),
+            Arc::new(UInt64Array::from(vec![cycle; n])),
+            Arc::new(UInt64Array::from(vec![base; n])),
+            Arc::new(UInt64Array::from(vec![batch.batch_hash; n])),
+            Arc::new(UInt64Array::from(stream)),
+            Arc::new(owner.finish()),
+            Arc::new(UInt64Array::from(row)),
+            nulls_u32(),
+            nulls_u8(),
+            nulls_u8(),
+            nulls_u32(),
+            nulls_u8(),
+            Arc::new(payload.finish()),
+            Arc::new(Int64Array::from(pk)),
+        ],
+    )
+    .expect("M batch")
+}
+
+/// Write one frozen cycle through the upsert `merge_insert` (MemWAL path).
+pub async fn m_put(t: &lancedb::Table, batch: &DetachedCycleBatch) {
+    let rb = m_batch(batch);
+    let schema = rb.schema();
+    let mut mi = t.merge_insert(&["pk"]);
+    mi.when_matched_update_all(None)
+        .when_not_matched_insert_all();
+    mi.execute(Box::new(RecordBatchIterator::new(vec![Ok(rb)], schema)))
+        .await
+        .expect("M merge_insert");
+}
+
+/// Rows per cycle as an M reader sees them. `lsm = true` routes through the
+/// LSM scanner (memtables of writers open in THIS process + flushed
+/// generations); `false` reads the base table only.
+pub async fn m_rows_by_cycle(t: &lancedb::Table, lsm: bool) -> BTreeMap<u64, usize> {
+    let mut stream = t
+        .query()
+        .select(Select::columns(&["cycle"]))
+        .use_lsm(lsm)
+        .execute()
+        .await
+        .expect("M query");
+    let mut out = BTreeMap::new();
+    while let Some(b) = stream.try_next().await.expect("M batch") {
+        count_cycles(&b, &mut out);
+    }
+    out
+}
+
+fn count_cycles(b: &RecordBatch, out: &mut BTreeMap<u64, usize>) {
+    let c: &UInt64Array = b
+        .column_by_name("cycle")
+        .and_then(|c| c.as_any().downcast_ref())
+        .expect("cycle column");
+    for i in 0..c.len() {
+        *out.entry(c.value(i)).or_default() += 1;
+    }
+}
+
+/// Bytes and files under `dir`, recursively.
+pub fn disk_usage(dir: &std::path::Path) -> (u64, u64) {
+    let mut bytes = 0;
+    let mut files = 0;
+    if let Ok(rd) = std::fs::read_dir(dir) {
+        for e in rd.flatten() {
+            let p = e.path();
+            if p.is_dir() {
+                let (b, f) = disk_usage(&p);
+                bytes += b;
+                files += f;
+            } else if let Ok(m) = e.metadata() {
+                bytes += m.len();
+                files += 1;
+            }
+        }
+    }
+    (bytes, files)
+}
diff --git a/crates/lance-graph/tests/wal_shootout_probe.rs b/crates/lance-graph/tests/wal_shootout_probe.rs
new file mode 100644
index 000000000..54f451fa0
--- /dev/null
+++ b/crates/lance-graph/tests/wal_shootout_probe.rs
@@ -0,0 +1,330 @@
+//! D-WAL-1 — validation of `LanceCycleWriter` (arm L) against the lancedb 0.40
+//! MemWAL write path (arm M). Plan: `.claude/plans/wal-shootout-v1.md` §2.
+//!
+//! - **V1** SIGKILL a child mid-write: recovered rows per cycle ∈ {0, all}.
+//! - **V2** every cycle the child acknowledged before the kill is readable.
+//! - **V3** resubmitting the same frozen cycle adds no rows.
+//! - **V4** what a second reader sees while a cycle is acknowledged but not
+//!   flushed (observed behaviour, printed, not a pass/fail gate).
+//!
+//! M is read two ways after a crash: (a) a reader that has not written, (b) the
+//! same reader after its first write, which opens the shard writer and replays
+//! the WAL. Lance local filesystem only; a process crash, not a power loss.
+//!
+//! Run: `cargo test -p lance-graph --features lancedb-sdk --test wal_shootout_probe -- --nocapture --test-threads=1`
+
+#![cfg(feature = "lancedb-sdk")]
+
+mod wal_common;
+
+use std::io::{BufRead, BufReader, Write};
+use std::process::{Command, Stdio};
+use std::time::Duration;
+
+use lance_graph::graph::cycle_sink::LanceCycleWriter;
+use lance_graph_planner::persist_sink::{CommitOutcome, WalSink};
+use wal_common::*;
+
+const CHILD_ENV: &str = "WAL_SHOOTOUT_CHILD";
+/// V1 cycle size: 4,096 casts → 8,193 rows, about 2 MB of payload.
+const V1_N: usize = 4096;
+/// V2 cycle size.
+const V2_N: usize = 1024;
+/// WAL flush interval for M in these tests.
+const FLUSH_MS: u64 = 10;
+
+fn l_path(dir: &str) -> String {
+    format!("{dir}/cycles.lance")
+}
+
+// ───────────────────────────── child mode ─────────────────────────────
+
+/// Child body. `WAL_SHOOTOUT_CHILD=<v1|v2>:<L|M>:<dir>`; a no-op otherwise.
+#[tokio::test(flavor = "multi_thread")]
+async fn wal_child_body() {
+    let Ok(spec) = std::env::var(CHILD_ENV) else {
+        return;
+    };
+    let mut it = spec.splitn(3, ':');
+    let (mode, arm, dir) = (it.next().unwrap(), it.next().unwrap(), it.next().unwrap());
+    let cycles: u64 = if mode == "v1" { 1 } else { 10_000 };
+    let n = if mode == "v1" { V1_N } else { V2_N };
+    let mut out = std::io::stdout();
+    match arm {
+        "L" => {
+            let mut w = LanceCycleWriter::open(l_path(dir)).await.expect("open");
+            for c in 1..=cycles {
+                l_commit(&mut w, c, n).await;
+                writeln!(out, "ACK {c}").unwrap();
+                out.flush().unwrap();
+            }
+        }
+        "M" => {
+            let t = m_open(dir).await;
+            for c in 1..=cycles {
+                m_put(&t, &frozen(c, 0, n)).await;
+                writeln!(out, "ACK {c}").unwrap();
+                out.flush().unwrap();
+            }
+        }
+        other => panic!("arm {other}"),
+    }
+    std::thread::sleep(Duration::from_secs(30));
+}
+
+fn spawn_child(mode: &str, arm: &str, dir: &str, stdout: Stdio) -> std::process::Child {
+    Command::new(std::env::current_exe().expect("current_exe"))
+        .args(["--exact", "wal_child_body", "--nocapture"])
+        .env(CHILD_ENV, format!("{mode}:{arm}:{dir}"))
+        .stdout(stdout)
+        .stderr(Stdio::null())
+        .spawn()
+        .expect("spawn child")
+}
+
+/// After a crash: (L) rows per cycle; (M) rows per cycle before and after the
+/// reader's first write (sentinel cycle `u64::MAX >> 8`, excluded from the map).
+async fn recovered(
+    arm: &str,
+    dir: &str,
+) -> (
+    std::collections::BTreeMap<u64, usize>,
+    Option<std::collections::BTreeMap<u64, usize>>,
+) {
+    match arm {
+        "L" => (
+            l_rows_by_cycle(&l_path(dir)).await.unwrap_or_default(),
+            None,
+        ),
+        _ => {
+            let t = m_open(dir).await;
+            let before = m_rows_by_cycle(&t, true).await;
+            let sentinel = u64::MAX >> 8;
+            m_put(&t, &frozen(sentinel, 0, 1)).await;
+            let mut after = m_rows_by_cycle(&t, true).await;
+            after.remove(&sentinel);
+            (before, Some(after))
+        }
+    }
+}
+
+async fn prepare(arm: &str, dir: &str) {
+    if arm == "M" {
+        m_create(dir, FLUSH_MS).await;
+    }
+}
+
+// ───────────────────────────── V1 ─────────────────────────────
+
+#[tokio::test(flavor = "multi_thread")]
+async fn v1_sigkill_mid_write_leaves_all_or_nothing() {
+    if std::env::var(CHILD_ENV).is_ok() {
+        return;
+    }
+    let full = rows_per_cycle(V1_N);
+    for arm in ["L", "M"] {
+        let mut observed = Vec::new();
+        // 0 ms kills before the child can write; 10 s lets it finish. Both ends
+        // are needed: a sweep that lands on one side only proves nothing.
+        for delay_ms in [0u64, 5, 30, 60, 120, 250, 500, 1000, 2000, 10_000] {
+            let tmp = tempfile::tempdir().expect("tempdir");
+            let dir = tmp.path().to_string_lossy().to_string();
+            prepare(arm, &dir).await;
+            let mut child = spawn_child("v1", arm, &dir, Stdio::null());
+            std::thread::sleep(Duration::from_millis(delay_ms));
+            child.kill().expect("kill");
+            let _ = child.wait();
+            let (a, b) = recovered(arm, &dir).await;
+            let ra = a.get(&1).copied().unwrap_or(0);
+            let rb = b.as_ref().map(|m| m.get(&1).copied().unwrap_or(0));
+            eprintln!(
+                "V1 {arm} kill@{delay_ms}ms recovered: read={ra} after-replay={rb:?} (full {full})"
+            );
+            observed.push((delay_ms, ra, rb));
+        }
+        let partial: Vec<_> = observed
+            .iter()
+            .filter(|(_, a, b)| (*a != 0 && *a != full) || b.is_some_and(|b| b != 0 && b != full))
+            .collect();
+        let landed = observed
+            .iter()
+            .any(|(_, a, b)| *a > 0 || b.is_some_and(|b| b > 0));
+        let empty = observed
+            .iter()
+            .any(|(_, a, b)| *a == 0 && b.unwrap_or(0) == 0);
+        assert!(
+            landed && empty,
+            "V1 {arm}: the sweep did not land on both sides of the durability \
+             boundary (landed={landed}, empty={empty}), so it proves nothing; widen it"
+        );
+        assert!(
+            partial.is_empty(),
+            "V1 {arm}: partial cycle after SIGKILL: {partial:?}"
+        );
+    }
+}
+
+// ───────────────────────────── V2 ─────────────────────────────
+
+#[tokio::test(flavor = "multi_thread")]
+async fn v2_acknowledged_cycles_survive_sigkill() {
+    if std::env::var(CHILD_ENV).is_ok() {
+        return;
+    }
+    let full = rows_per_cycle(V2_N);
+    for arm in ["L", "M"] {
+        for target in [2u64, 5, 9] {
+            let tmp = tempfile::tempdir().expect("tempdir");
+            let dir = tmp.path().to_string_lossy().to_string();
+            prepare(arm, &dir).await;
+            let mut child = spawn_child("v2", arm, &dir, Stdio::piped());
+            let mut acked = 0u64;
+            {
+                let reader = BufReader::new(child.stdout.take().expect("stdout"));
+                for line in reader.lines() {
+                    let line = line.expect("line");
+                    if let Some(k) = line.strip_prefix("ACK ") {
+                        acked = k.trim().parse().expect("ack");
+                        if acked >= target {
+                            break;
+                        }
+                    }
+                }
+            }
+            assert!(
+                acked >= target,
+                "V2 {arm}: the child stopped at ACK {acked} before ACK {target}; nothing to recover"
+            );
+            child.kill().expect("kill");
+            let _ = child.wait();
+            let (a, b) = recovered(arm, &dir).await;
+            let lost = |m: &std::collections::BTreeMap<u64, usize>| -> Vec<u64> {
+                (1..=acked)
+                    .filter(|c| m.get(c).copied() != Some(full))
+                    .collect()
+            };
+            let partial = |m: &std::collections::BTreeMap<u64, usize>| -> Vec<(u64, usize)> {
+                m.iter()
+                    .filter(|(_, &v)| v != full)
+                    .map(|(&k, &v)| (k, v))
+                    .collect()
+            };
+            eprintln!(
+                "V2 {arm} acked={acked}: read sees cycles {:?} (missing acked {:?}); after replay {:?}",
+                a.keys().collect::<Vec<_>>(),
+                lost(&a),
+                b.as_ref().map(|m| (m.keys().copied().collect::<Vec<_>>(), lost(m))),
+            );
+            assert!(
+                partial(&a).is_empty(),
+                "V2 {arm}: partial cycle {:?}",
+                partial(&a)
+            );
+            match &b {
+                None => assert!(
+                    lost(&a).is_empty(),
+                    "V2 L lost acknowledged cycles {:?}",
+                    lost(&a)
+                ),
+                Some(b) => {
+                    assert!(
+                        partial(b).is_empty(),
+                        "V2 M: partial cycle after replay {:?}",
+                        partial(b)
+                    );
+                    assert!(
+                        lost(b).is_empty(),
+                        "V2 M lost acknowledged cycles after replay {:?}",
+                        lost(b)
+                    );
+                }
+            }
+        }
+    }
+}
+
+// ───────────────────────────── V3 ─────────────────────────────
+
+#[tokio::test(flavor = "multi_thread")]
+async fn v3_resubmitting_a_frozen_cycle_adds_no_rows() {
+    if std::env::var(CHILD_ENV).is_ok() {
+        return;
+    }
+    let n = 64;
+    // L
+    let tmp = tempfile::tempdir().unwrap();
+    let dir = tmp.path().to_string_lossy().to_string();
+    let mut w = LanceCycleWriter::open(l_path(&dir)).await.unwrap();
+    let batch = frozen(1, w.head().0, n);
+    let first = w.commit_cycle(batch.clone()).await.unwrap();
+    let second = w.commit_cycle(batch).await.unwrap();
+    let rows = l_rows_by_cycle(&l_path(&dir)).await.unwrap();
+    eprintln!("V3 L first={first:?} second={second:?} rows={rows:?}");
+    assert!(matches!(first, CommitOutcome::Committed { .. }));
+    assert!(matches!(second, CommitOutcome::Reconciled { .. }));
+    assert_eq!(rows.get(&1), Some(&rows_per_cycle(n)));
+    drop(w);
+    // M
+    let tmp = tempfile::tempdir().unwrap();
+    let dir = tmp.path().to_string_lossy().to_string();
+    let t = m_create(&dir, FLUSH_MS).await;
+    let batch = frozen(1, 0, n);
+    m_put(&t, &batch).await;
+    m_put(&t, &batch).await;
+    let rows = m_rows_by_cycle(&t, true).await;
+    eprintln!("V3 M rows after two puts of one cycle={rows:?}");
+    assert_eq!(rows.get(&1), Some(&rows_per_cycle(n)));
+}
+
+// ───────────────────────────── V4 ─────────────────────────────
+
+#[tokio::test(flavor = "multi_thread")]
+async fn v4_read_horizon_while_a_cycle_is_unflushed() {
+    if std::env::var(CHILD_ENV).is_ok() {
+        return;
+    }
+    let n = 64;
+    // L: a reader opened at cycle 1 stays there; a fresh reader sees cycle 2.
+    let tmp = tempfile::tempdir().unwrap();
+    let dir = tmp.path().to_string_lossy().to_string();
+    let path = l_path(&dir);
+    let mut w = LanceCycleWriter::open(path.clone()).await.unwrap();
+    l_commit(&mut w, 1, n).await;
+    let pinned = lance::dataset::Dataset::open(&path).await.unwrap();
+    l_commit(&mut w, 2, n).await;
+    let pinned_rows = pinned.count_rows(None).await.unwrap();
+    let fresh = l_rows_by_cycle(&path).await.unwrap();
+    eprintln!(
+        "V4 L pinned reader rows={pinned_rows} fresh reader cycles={:?}",
+        fresh.keys().collect::<Vec<_>>()
+    );
+    assert_eq!(pinned_rows, rows_per_cycle(n));
+    assert_eq!(fresh.len(), 2);
+    drop(w);
+    // M: writer handle A, second handle B in the same process (own writer cache).
+    let tmp = tempfile::tempdir().unwrap();
+    let dir = tmp.path().to_string_lossy().to_string();
+    let a = m_create(&dir, FLUSH_MS).await;
+    m_put(&a, &frozen(1, 0, n)).await;
+    m_put(&a, &frozen(2, 0, n)).await;
+    let b = m_open(&dir).await;
+    let a_lsm = m_rows_by_cycle(&a, true).await;
+    let b_lsm = m_rows_by_cycle(&b, true).await;
+    let b_base = m_rows_by_cycle(&b, false).await;
+    eprintln!(
+        "V4 M acked 2 cycles: writer LSM {:?} | second handle LSM {:?} | base-only {:?}",
+        a_lsm, b_lsm, b_base
+    );
+    a.close_lsm_writers().await.unwrap();
+    let b2 = m_open(&dir).await;
+    eprintln!(
+        "V4 M after close_lsm_writers: second handle LSM {:?} | base-only {:?}",
+        m_rows_by_cycle(&b2, true).await,
+        m_rows_by_cycle(&b2, false).await
+    );
+    assert_eq!(
+        a_lsm.len(),
+        2,
+        "the writer's own LSM read must see what it acknowledged"
+    );
+}