lab(drf-w1): staged writeback coalesced per canonical row (W1–W3)

.claude/board/entries/2026-10-25-drf-w1-writeback.md

@@ -0,0 +1,22 @@
+# D-DRF W1–W3 — staged writeback coalesced per canonical row (lab)
+
+**Status:** TEST-PINNED (lab). **Code:** `crates/drf-focus-probe/src/lib.rs` (`Focus::writeback`,
+`Writeback::{stage, dirty_rows, flush}`, `RowImage`). **Tests:** `w1_*`, `w2_*`, `w3_*` in
+`tests/integration/focus.rs`; 14/14 green on native AVX-512 and x86-64-v3.
+
+- **W1 PASS:** 4096 slots over 37 distinct rows, 64 thoughts → exactly 37 row images
+  (37 × 512 = 18 944 B), in ordinal order, bytes outside the written facet equal to the
+  sealed source. A1 (dirty fraction) for this workload = 37 / 4096 slots = 0.009.
+- **W2 PASS:** two slots aliasing one row write different facets; the single image keeps
+  both, identical for both staging orders.
+- **W3 PASS (refusal arm):** two different values for one facet of one row →
+  `Refusal::Conflict`; equal values coalesce. Two facets whose byte ranges overlap
+  (offsets 16 and 17) also refuse; facets that only touch both survive (codex P2 on #1765).
+  Stale source at flush → `StaleMap`.
+- **Disables, each red:** one image per patch (W1/W2/W3), last-payload-wins row replacement
+  (W2), conflict check removed (W3), overlap clause removed (W3 overlap).
+
+OPEN: W3's "declared stream position" ordering (deterministic noncommuting apply) is not
+built; this probe only refuses. W4–W8 (owned handoff to a CycleWriter, WAL, durability,
+eviction under pressure) not run. Images are owned copies by design: they are the frozen
+batch a writer takes ownership of (W4), not a read path.

.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.
 
-466 entries, 2026-08-06 .. 2026-10-24.
+467 entries, 2026-08-06 .. 2026-10-25.
 
 | date | entry id | finding | file |
 |---|---|---|---|
+| 2026-10-25 | `D-DRF` |  | [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) |
 | 2026-10-24 | `D-DRF-P2` |  | [2026-10-24-drf-p2-breakeven.md](2026-10-24-drf-p2-breakeven.md) |
 | 2026-10-24 | `D-DRF-P1` |  | [2026-10-24-drf-p1-focus-lease-probe.md](2026-10-24-drf-p1-focus-lease-probe.md) |

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

@@ -82,9 +82,25 @@ impl Facet {
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum Refusal {
-    OrdinalOutOfRange { slot: usize, ord: u16 },
-    StaleMap { map_seal: u64, source_seal: u64 },
+    OrdinalOutOfRange {
+        slot: usize,
+        ord: u16,
+    },
+    StaleMap {
+        map_seal: u64,
+        source_seal: u64,
+    },
     FacetOutOfRow,
+    /// Slot index outside the 64x64 focus.
+    SlotOutOfFocus {
+        slot: usize,
+    },
+    /// Two staged writes give one facet of one row different values, or
+    /// write overlapping byte ranges of one row (W3).
+    Conflict {
+        ord: u16,
+        offset: usize,
+    },
 }
 
 fn read_i32(row: &Row, offset: usize) -> i32 {
@@ -271,3 +287,109 @@ pub fn direct_sum_where(
 fn prefetch(p: *const u8) {
     ndarray::simd::prefetch_read_t0(p);
 }
+
+/// One staged facet write, addressed by canonical ordinal.
+#[derive(Clone, Copy)]
+struct Patch {
+    ord: u16,
+    offset: usize,
+    value: i32,
+}
+
+/// The final image of one dirty canonical row, owned for handoff to a writer.
+pub struct RowImage {
+    pub ord: u16,
+    pub row: Row,
+}
+
+/// Writes staged against one focus, coalesced per canonical row at flush
+/// (D-DRF W1-W3). The focus itself is never mutated.
+pub struct Writeback<'f> {
+    focus: &'f Focus,
+    patches: Vec<Patch>,
+}
+
+impl Focus {
+    pub fn writeback(&self) -> Writeback<'_> {
+        Writeback {
+            focus: self,
+            patches: Vec::new(),
+        }
+    }
+}
+
+impl Writeback<'_> {
+    /// Stage `value` into `facet` of the row behind `slot`.
+    pub fn stage(&mut self, slot: usize, facet: Facet, value: i32) -> Result<(), Refusal> {
+        let ord = *self
+            .focus
+            .map
+            .ord
+            .get(slot)
+            .ok_or(Refusal::SlotOutOfFocus { slot })?;
+        self.patches.push(Patch {
+            ord,
+            offset: facet.offset(),
+            value,
+        });
+        Ok(())
+    }
+
+    /// Distinct canonical rows touched so far.
+    pub fn dirty_rows(&self) -> usize {
+        let mut ords: Vec<u16> = self.patches.iter().map(|p| p.ord).collect();
+        ords.sort_unstable();
+        ords.dedup();
+        ords.len()
+    }
+
+    /// One image per distinct dirty row, in ordinal order: the sealed source row
+    /// with every staged facet applied. Writes to different facets of one row
+    /// all survive (W2); two different values for the same facet, or two
+    /// facets whose byte ranges overlap, refuse (W3).
+    pub fn flush(mut self, soa: &Soa<'_>) -> Result<Vec<RowImage>, Refusal> {
+        check_seal(soa, &self.focus.map)?;
+        if self.focus.key.source != soa.id {
+            return Err(Refusal::StaleMap {
+                map_seal: self.focus.map.seal,
+                source_seal: soa.id.index_seal,
+            });
+        }
+        self.patches.sort_by_key(|p| (p.ord, p.offset));
+        for w in self.patches.windows(2) {
+            if w[0].ord != w[1].ord {
+                continue;
+            }
+            // Same facet with a different value, or two facets whose byte
+            // ranges overlap: either would silently overwrite a staged write.
+            // Sorted by offset, so checking neighbours finds every overlap.
+            let same = w[0].offset == w[1].offset;
+            if (same && w[0].value != w[1].value)
+                || (!same && w[1].offset < w[0].offset + Facet::WIDTH)
+            {
+                return Err(Refusal::Conflict {
+                    ord: w[1].ord,
+                    offset: w[1].offset,
+                });
+            }
+        }
+        let mut out: Vec<RowImage> = Vec::new();
+        for p in &self.patches {
+            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
+                    .rows
+                    .get(p.ord as usize)
+                    .ok_or(Refusal::OrdinalOutOfRange {
+                        slot: usize::MAX,
+                        ord: p.ord,
+                    })?;
+                out.push(RowImage { ord: p.ord, row });
+            }
+            let img = out.last_mut().expect("pushed above");
+            img.row[p.offset..p.offset + Facet::WIDTH].copy_from_slice(&p.value.to_le_bytes());
+        }
+        Ok(out)
+    }
+}

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

@@ -344,3 +344,149 @@ fn p2_mispredicted_focus_and_64_fast_switches_leave_no_lease_and_no_corruption()
     assert_eq!(mgr.evict_unused(), 2);
     assert!(mgr.is_empty());
 }
+
+const OFF2: usize = 64;
+
+#[test]
+fn w1_dirty_rows_coalesce_to_one_image_per_distinct_row() {
+    let r = rows();
+    let soa = Soa {
+        rows: &r,
+        id: sid(1, 1, 1, 9),
+    };
+    // 4096 slots over 37 distinct rows: 37 * 512 B of final images.
+    let m = map_from(300, 9, |s| (s % 37) as u16 * 11);
+    let mgr = FocusManager::new();
+    let f = mgr.load(&soa, &m, Scope(1), facet()).unwrap();
+    let mut wb = f.writeback();
+    // 64 thoughts, each touching every slot of its row with the same value.
+    for t in 0..64usize {
+        for s in (t..SLOTS).step_by(64) {
+            let ord = m.ord[s] as i32;
+            wb.stage(s, facet(), ord * 2).unwrap();
+        }
+    }
+    assert_eq!(wb.dirty_rows(), 37);
+    let imgs = wb.flush(&soa).unwrap();
+    assert_eq!(imgs.len(), 37);
+    assert_eq!(imgs.len() * ROW, 18_944);
+    for w in imgs.windows(2) {
+        assert!(w[0].ord < w[1].ord, "images in ordinal order");
+    }
+    for img in &imgs {
+        let src = &r[img.ord as usize];
+        let v = i32::from_le_bytes(img.row[OFF..OFF + 4].try_into().unwrap());
+        assert_eq!(v, img.ord as i32 * 2);
+        // Bytes outside the written facet equal the sealed source row.
+        assert_eq!(&img.row[..OFF], &src[..OFF]);
+        assert_eq!(&img.row[OFF + 4..], &src[OFF + 4..]);
+    }
+    // The focus's own gathered column is untouched.
+    assert_eq!(f.sum_where(&[u64::MAX; 64]), oracle(&m, &[u64::MAX; 64]));
+}
+
+#[test]
+fn w2_two_aliasing_slots_writing_different_facets_both_survive_in_any_order() {
+    let r = rows();
+    let soa = Soa {
+        rows: &r,
+        id: sid(1, 1, 1, 9),
+    };
+    // Slots 0 and 1 both point at row 500.
+    let m = map_from(301, 9, |s| if s < 2 { 500 } else { (s % NROWS) as u16 });
+    let mgr = FocusManager::new();
+    let f = mgr.load(&soa, &m, Scope(1), facet()).unwrap();
+    let other = Facet::new(OFF2).unwrap();
+    let image = |first: bool| {
+        let mut wb = f.writeback();
+        if first {
+            wb.stage(0, facet(), 7).unwrap();
+            wb.stage(1, other, -9).unwrap();
+        } else {
+            wb.stage(1, other, -9).unwrap();
+            wb.stage(0, facet(), 7).unwrap();
+        }
+        let imgs = wb.flush(&soa).unwrap();
+        assert_eq!(imgs.len(), 1);
+        imgs[0].row
+    };
+    let (a, b) = (image(true), image(false));
+    assert_eq!(a, b, "order must not matter");
+    assert_eq!(i32::from_le_bytes(a[OFF..OFF + 4].try_into().unwrap()), 7);
+    assert_eq!(
+        i32::from_le_bytes(a[OFF2..OFF2 + 4].try_into().unwrap()),
+        -9
+    );
+}
+
+#[test]
+fn w3_same_facet_different_values_refuse_and_equal_values_coalesce() {
+    let r = rows();
+    let soa = Soa {
+        rows: &r,
+        id: sid(1, 1, 1, 9),
+    };
+    let m = map_from(302, 9, |s| if s < 2 { 500 } else { (s % NROWS) as u16 });
+    let mgr = FocusManager::new();
+    let f = mgr.load(&soa, &m, Scope(1), facet()).unwrap();
+    let mut wb = f.writeback();
+    wb.stage(0, facet(), 7).unwrap();
+    wb.stage(1, facet(), 8).unwrap();
+    assert_eq!(
+        wb.flush(&soa).err(),
+        Some(Refusal::Conflict {
+            ord: 500,
+            offset: OFF
+        })
+    );
+    let mut wb = f.writeback();
+    wb.stage(0, facet(), 7).unwrap();
+    wb.stage(1, facet(), 7).unwrap();
+    assert_eq!(wb.flush(&soa).unwrap().len(), 1);
+    // Out-of-focus slot and a flush against another source are refused.
+    let mut wb = f.writeback();
+    assert_eq!(
+        wb.stage(SLOTS, facet(), 1),
+        Err(Refusal::SlotOutOfFocus { slot: SLOTS })
+    );
+    wb.stage(0, facet(), 1).unwrap();
+    let moved = Soa {
+        rows: &r,
+        id: sid(1, 1, 2, 9),
+    };
+    assert!(matches!(wb.flush(&moved), Err(Refusal::StaleMap { .. })));
+}
+
+#[test]
+fn w3_overlapping_facets_at_different_offsets_refuse_and_adjacent_facets_both_survive() {
+    // Codex P2 on #1765: offsets 16 and 17 share bytes 17..20, so the second
+    // write would silently corrupt the first. Refuse it.
+    let r = rows();
+    let soa = Soa {
+        rows: &r,
+        id: sid(1, 1, 1, 9),
+    };
+    let m = map_from(303, 9, |s| if s < 2 { 500 } else { (s % NROWS) as u16 });
+    let mgr = FocusManager::new();
+    let f = mgr.load(&soa, &m, Scope(1), facet()).unwrap();
+    let mut wb = f.writeback();
+    wb.stage(0, facet(), 1).unwrap();
+    wb.stage(1, Facet::new(OFF + 1).unwrap(), 2).unwrap();
+    assert_eq!(
+        wb.flush(&soa).err(),
+        Some(Refusal::Conflict {
+            ord: 500,
+            offset: OFF + 1
+        })
+    );
+    // Silence twin: facets that only touch (OFF, OFF + 4) do not overlap.
+    let mut wb = f.writeback();
+    wb.stage(0, facet(), 1).unwrap();
+    wb.stage(1, Facet::new(OFF + Facet::WIDTH).unwrap(), 2)
+        .unwrap();
+    let imgs = wb.flush(&soa).unwrap();
+    assert_eq!(imgs.len(), 1);
+    let row = &imgs[0].row;
+    assert_eq!(row[OFF..OFF + 4], 1i32.to_le_bytes());
+    assert_eq!(row[OFF + 4..OFF + 8], 2i32.to_le_bytes());
+}