lab(drf-w4): hand flushed row images to the real CycleWriter (W4-W7)

.claude/board/entries/2026-10-25-drf-w4-handoff.md

@@ -0,0 +1,28 @@
+# D-DRF-W4 — W4–W7: flushed row images handed to the real CycleWriter (lab)
+
+**Status:** TEST-PINNED (lab, `cargo test --features handoff`; no CI job runs this crate).
+**Code:** `crates/drf-focus-probe/src/handoff.rs` (`RowImage::into_slot`, `into_slots`).
+**Tests:** `tests/integration/handoff.rs`, driving `lance-graph-supervisor::cycle_writer`
+over a reconciliation-first fake WAL. Plan: `.claude/plans/d-dr64-focus-cache-falsifiers-v1.md` §4.
+
+- **W4 PASS:** 64 writes over 37 rows → 37 owned casts. The focus and the borrow end at
+  flush; the source is then changed while the WAL holds the commit. Nothing is durable until
+  the WAL releases (`durable().through == None`, ticket unresolved); what lands is the flushed
+  snapshot. Stream positions continue across cycles (37 → 74).
+- **W2 through the writer PASS:** two facets of one row land together. Contrast arm: one cast
+  per raw patch loses the first facet, because the writer's freeze keeps the later payload
+  whole. The per-row coalesce in `flush` is what makes the handoff safe.
+- **W6 PASS:** lost acknowledgement → `ResubmitFrozen`, handed-back casts equal the frozen
+  batch byte for byte, re-submit on a fresh writer → `Reconciled`, one append. Regenerating
+  cycle 1 from the changed source → `HashConflict` → `Escalate`. A full queue hands the casts
+  back unchanged; a poisoned writer refuses new submissions.
+- **W7 PASS:** a cycle with no dirty row gives zero casts → `NoChange`, the WAL is never
+  called, the head stays; the next dirty cycle publishes exactly one version.
+- **Disables, each red:** one image per patch (W1–W3 tests and the through-writer W2 test);
+  `into_slots` not advancing the stream position (W4).
+- **Change:** `RowImage::row` is now `Vec<u8>` (was `[u8; 512]`) so the move into the
+  cast's payload is not a second copy.
+
+OPEN: W5 (descriptor-only cast pointing into live SoA, lease held until capture) needs a
+descriptor payload the writer does not have; W8 (eviction under memory pressure with live
+readers) is not run. The fake WAL stands in for Lance; a real `WalSink` run is not done.

.claude/board/entries/README.md

@@ -25,10 +25,11 @@ 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.
 
-468 entries, 2026-08-06 .. 2026-10-25.
+469 entries, 2026-08-06 .. 2026-10-25.
 
 | date | entry id | finding | file |
 |---|---|---|---|
+| 2026-10-25 | `D-DRF-W4` |  | [2026-10-25-drf-w4-handoff.md](2026-10-25-drf-w4-handoff.md) |
 | 2026-10-25 | `D-DRF-W3` |  | [2026-10-25-drf-w3-ordered-apply.md](2026-10-25-drf-w3-ordered-apply.md) |
 | 2026-10-25 | `D-DRF-W1` |  | [2026-10-25-drf-w1-writeback.md](2026-10-25-drf-w1-writeback.md) |
 | 2026-10-24 | `tanh-precision-symmetric-i8` |  | [2026-10-24-tanh-precision-symmetric-i8.md](2026-10-24-tanh-precision-symmetric-i8.md) |

crates/drf-focus-probe/Cargo.toml

@@ -2,7 +2,8 @@
 # stale/foreign-key refusal, intra-focus prefetch lookahead. Timings only in
 # examples/p2_breakeven.rs, reported, not gated.
 # Own [workspace] root, excluded from the lance-graph workspace. Run from here:
-#   cargo test
+#   cargo test                      (W1-W3, focus, prefetch)
+#   cargo test --features handoff   (W4-W7 against the real CycleWriter)
 [package]
 name = "drf-focus-probe"
 version = "0.0.0"
@@ -14,6 +15,17 @@ description = "Lab probe D-DRF-P1: sealed 64x64 u16 map, full-witness FocusKey,
 
 [dependencies]
 ndarray = { git = "https://github.com/AdaWorldAPI/ndarray", branch = "master", default-features = false, features = ["std"] }
+# W4-W7 handoff to the real CycleWriter (`cargo test --features handoff`).
+lance-graph-planner = { path = "../lance-graph-planner", optional = true }
+
+[features]
+handoff = ["dep:lance-graph-planner"]
+
+[dev-dependencies]
+lance-graph-supervisor = { path = "../lance-graph-supervisor", features = ["cycle-driver"] }
+lance-graph-planner = { path = "../lance-graph-planner" }
+lance-graph-contract = { path = "../lance-graph-contract" }
+tokio = { version = "1", features = ["rt", "macros", "sync"] }
 
 [patch."https://github.com/AdaWorldAPI/ndarray"]
 ndarray = { path = "../../../ndarray" }

crates/drf-focus-probe/src/handoff.rs

@@ -0,0 +1,52 @@
+//! D-DRF W4–W7: hand flushed row images to the real `CycleWriter` as owned
+//! casts (feature `handoff`).
+//!
+//! A [`RowImage`] owns its bytes, so once [`Writeback::flush`] returns, the
+//! source rows and the focus can be dropped or changed while the writer is
+//! still sealing. The compiler checks that; the tests check that what lands is
+//! the flushed snapshot.
+//!
+//! One cast per distinct row. The writer's freeze folds same-row casts by
+//! replacing the whole payload (later stream position wins), so two casts for
+//! one row would drop one of them. [`Writeback::flush`] already coalesces per
+//! row, which is why the handoff takes its output and nothing else.
+//!
+//! [`Writeback::flush`]: crate::Writeback::flush
+
+use lance_graph_planner::persist_sink::{CycleId, MailboxId, SweepSlot};
+
+use crate::RowImage;
+
+impl RowImage {
+    /// This image as one cast for `owner`. The row bytes are moved, not copied.
+    pub fn into_slot(self, cycle: CycleId, stream_position: u64, owner: MailboxId) -> SweepSlot {
+        SweepSlot {
+            cycle,
+            stream_position,
+            owner,
+            row: u64::from(self.ord),
+            paired_move: None,
+            payload: self.row,
+        }
+    }
+}
+
+/// One cast per image, in the images' ordinal order, at stream positions
+/// `next_position..next_position + images.len()`. Returns the casts and the
+/// next free position. Positions must keep increasing across cycles (the
+/// writer's recovery watermark depends on it), so pass the returned value to
+/// the next cycle.
+pub fn into_slots(
+    images: Vec<RowImage>,
+    cycle: CycleId,
+    owner: MailboxId,
+    next_position: u64,
+) -> (Vec<SweepSlot>, u64) {
+    let n = images.len() as u64;
+    let slots = images
+        .into_iter()
+        .zip(next_position..)
+        .map(|(img, pos)| img.into_slot(cycle, pos, owner))
+        .collect();
+    (slots, next_position + n)
+}

crates/drf-focus-probe/src/lib.rs

@@ -8,6 +8,9 @@ use std::collections::HashMap;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex, OnceLock};
 
+#[cfg(feature = "handoff")]
+pub mod handoff;
+
 /// Canonical SoA row ABI.
 pub const ROW: usize = 512;
 /// One SoA row.
@@ -301,9 +304,11 @@ struct Patch {
 }
 
 /// The final image of one dirty canonical row, owned for handoff to a writer.
+/// `row` is exactly [`ROW`] bytes. It is a `Vec` so the handoff can move it
+/// into the writer's payload without a second copy.
 pub struct RowImage {
     pub ord: u16,
-    pub row: Row,
+    pub row: Vec<u8>,
 }
 
 /// Writes staged against one focus, coalesced per canonical row at flush
@@ -410,13 +415,14 @@ impl Writeback<'_> {
             if out.last().is_none_or(|img| img.ord != p.ord) {
                 // The focus was built from this same source, so every ordinal
                 // was range-checked then; refuse rather than index if not.
-                let row = *soa
+                let row = soa
                     .rows
                     .get(p.ord as usize)
                     .ok_or(Refusal::OrdinalOutOfRange {
                         slot: usize::MAX,
                         ord: p.ord,
-                    })?;
+                    })?
+                    .to_vec();
                 out.push(RowImage { ord: p.ord, row });
             }
             let img = out.last_mut().expect("pushed above");

crates/drf-focus-probe/tests/integration/focus.rs

@@ -408,7 +408,7 @@ fn w2_two_aliasing_slots_writing_different_facets_both_survive_in_any_order() {
         }
         let imgs = wb.flush(&soa).unwrap();
         assert_eq!(imgs.len(), 1);
-        imgs[0].row
+        imgs.into_iter().next().unwrap().row
     };
     let (a, b) = (image(true), image(false));
     assert_eq!(a, b, "order must not matter");