feat(planner): D-V3-W1e probe-first — batch_writer skeleton + 3 red probes

WAL-shaped BatchWriter<P> skeleton (M24: cast = intent record, ack =
confirmation, unacked = replay surface, resolve_owner = W1c delegation
cache) with todo! bodies. Three #[ignore]d probes fail as designed
(ahead-update ordering, kill-after-cast replay, delegation miss-then-hit)
— the W1b implementation commit un-ignores them.

Uses the SHIPPED contract types (KanbanMove kanban.rs:151, KanbanColumn,
MailboxId collapse_gate.rs:121) — no mint needed; the brief's
conditional mint did not fire.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MLBnPuScZy6w9di2QEjsXM

Author: claude

crates/lance-graph-planner/src/batch_writer.rs

@@ -0,0 +1,106 @@
+//! W1b ahead-firing batch writer — the kanban board IS the write-ahead log (M24).
+//!
+//! `cast()` records intent moves AHEAD of any storage ack; `ack()` confirms;
+//! `unacked()` is the crash-replay surface. Payload-generic: the writer never
+//! inspects `P` (DTO purity — ownership rides the cast pairing, never the DTO).
+//!
+//! Uses the REAL shipped kanban contract types
+//! ([`lance_graph_contract::kanban::KanbanMove`],
+//! [`lance_graph_contract::kanban::KanbanColumn`],
+//! [`lance_graph_contract::collapse_gate::MailboxId`]) — this module does not
+//! mint a parallel `KanbanMove`; see the D-MBX-A6 Outcome adapter context in
+//! `crate::strategy::style_strategy`.
+
+use std::collections::HashMap;
+
+use lance_graph_contract::collapse_gate::MailboxId;
+use lance_graph_contract::kanban::KanbanMove;
+
+/// Identity of one `cast()` — a write-ahead intent record on the kanban board.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct CastId(pub u64);
+
+/// Ahead-firing batch writer: intent (`cast`) is visible on the board before
+/// any ack; `unacked()` is the M24 crash-replay surface; `resolve_owner` is
+/// the W1c delegation cache (resolve-once, cache-hit thereafter).
+pub struct BatchWriter<P> {
+    /// Monotonic id generator for the next cast.
+    next_id: u64,
+    /// Board: intent moves recorded per cast, keyed by `CastId`, alongside the
+    /// mailbox the cast was recorded on behalf of. Visible between `cast()`
+    /// and `ack()` (and beyond, for crash-replay via `unacked()`).
+    board: HashMap<CastId, (MailboxId, Vec<KanbanMove>)>,
+    /// Casts that have been confirmed (acked). A cast present in `board` but
+    /// absent from `acked` is the crash-replay surface (`unacked()`).
+    acked: std::collections::HashSet<CastId>,
+    /// W1c delegation cache: `on_behalf` mailbox -> resolved owner mailbox.
+    delegation_cache: HashMap<MailboxId, MailboxId>,
+    /// Payloads recorded per cast (payload-generic; the writer never inspects `P`).
+    pending_payloads: Vec<(CastId, P)>,
+}
+
+impl<P> Default for BatchWriter<P> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl<P> BatchWriter<P> {
+    /// Construct an empty batch writer.
+    pub fn new() -> Self {
+        Self {
+            next_id: 0,
+            board: HashMap::new(),
+            acked: std::collections::HashSet::new(),
+            delegation_cache: HashMap::new(),
+            pending_payloads: Vec::new(),
+        }
+    }
+
+    /// AHEAD: records the intent (moves visible on the board) BEFORE any ack.
+    /// Returns the cast id.
+    ///
+    /// Probe-first skeleton (D-V3-W1e): body pending — see
+    /// `tests/w1_probes.rs::probe_ahead_update_ordering` /
+    /// `probe_kill_after_cast_replay`.
+    pub fn cast(&mut self, _on_behalf: MailboxId, _moves: Vec<KanbanMove>, _payload: P) -> CastId {
+        todo!("W1b")
+    }
+
+    /// Confirmation — marks the cast acked.
+    ///
+    /// Probe-first skeleton (D-V3-W1e): body pending — see
+    /// `tests/w1_probes.rs::probe_ahead_update_ordering`.
+    pub fn ack(&mut self, _cast: CastId) {
+        todo!("W1b")
+    }
+
+    /// Crash-replay surface (M24): casts recorded but not yet acked.
+    ///
+    /// Probe-first skeleton (D-V3-W1e): body pending — see
+    /// `tests/w1_probes.rs::probe_kill_after_cast_replay`.
+    pub fn unacked(&self) -> Vec<CastId> {
+        todo!("W1b")
+    }
+
+    /// Board read: intent moves recorded for a cast (visible between cast and ack).
+    ///
+    /// Probe-first skeleton (D-V3-W1e): body pending — see
+    /// `tests/w1_probes.rs::probe_ahead_update_ordering` /
+    /// `probe_kill_after_cast_replay`.
+    pub fn intent_moves(&self, _cast: CastId) -> Option<&[KanbanMove]> {
+        todo!("W1b")
+    }
+
+    /// W1c delegation cache: resolve an owner once, cache; returns `(owner, was_cache_hit)`.
+    ///
+    /// Probe-first skeleton (D-V3-W1e): body pending — see
+    /// `tests/w1_probes.rs::probe_delegation_miss_then_hit`.
+    pub fn resolve_owner(
+        &mut self,
+        _on_behalf: MailboxId,
+        _resolver: impl FnOnce(MailboxId) -> MailboxId,
+    ) -> (MailboxId, bool) {
+        todo!("W1c")
+    }
+}

crates/lance-graph-planner/src/lib.rs

@@ -72,6 +72,9 @@ pub mod pipeline;
 // === Autocomplete Cache (VSA superposition KV-cache) ===
 pub mod cache;
 
+// === W1b ahead-firing batch writer (D-V3-W1e probe-first skeleton) ===
+pub mod batch_writer;
+
 // === Internal API (same-binary, zero-serde) ===
 pub mod api;
 

crates/lance-graph-planner/tests/w1_probes.rs

@@ -0,0 +1,105 @@
+//! D-V3-W1e probe-first: three failing probes for the W1b ahead-firing batch
+//! writer + W1c delegation cache, pinned against
+//! `lance_graph_planner::batch_writer::BatchWriter`.
+//!
+//! All three are `#[ignore]`d — the writer's methods are `todo!()` stubs.
+//! Un-ignore in the W1b implementation commit once the bodies are filled in.
+//!
+//! Uses the REAL shipped kanban contract types
+//! (`lance_graph_contract::kanban::{KanbanColumn, KanbanMove, ExecTarget}`,
+//! `lance_graph_contract::collapse_gate::MailboxId`) — no hand-rolled
+//! composites (F12).
+
+use lance_graph_contract::kanban::{ExecTarget, KanbanColumn, KanbanMove};
+use lance_graph_planner::batch_writer::BatchWriter;
+
+/// Build a `KanbanMove` from a `mailbox`/`from`/`to` triple using only public
+/// constructors/fields on the shipped contract type — no hand-rolled bit math.
+fn make_move(mailbox: u32, from: KanbanColumn, to: KanbanColumn, witness: u32) -> KanbanMove {
+    KanbanMove {
+        mailbox,
+        from,
+        to,
+        witness_chain_position: witness,
+        libet_offset_us: 0,
+        exec: ExecTarget::Native,
+    }
+}
+
+/// Probe 1 (W1b): cast() makes intent moves visible on the board AHEAD of any
+/// ack; ack() then removes the cast from unacked().
+#[test]
+#[ignore = "probe-first: W1b mechanism pending — un-ignore in the W1b implementation commit"]
+fn probe_ahead_update_ordering() {
+    let mut writer: BatchWriter<()> = BatchWriter::new();
+
+    let moves = vec![
+        make_move(7, KanbanColumn::Planning, KanbanColumn::CognitiveWork, 0),
+        make_move(7, KanbanColumn::CognitiveWork, KanbanColumn::Evaluation, 1),
+        make_move(7, KanbanColumn::Evaluation, KanbanColumn::Commit, 2),
+    ];
+
+    let cast = writer.cast(7, moves.clone(), ());
+
+    // AHEAD: intent is visible on the board BEFORE any ack.
+    assert_eq!(writer.intent_moves(cast), Some(moves.as_slice()));
+
+    writer.ack(cast);
+
+    // After ack, the cast is no longer in the unacked (crash-replay) surface.
+    assert!(!writer.unacked().contains(&cast));
+}
+
+/// Probe 2 (M24): a cast that is never acked stays on the crash-replay
+/// surface (`unacked()`), and its intent moves remain replayable.
+#[test]
+#[ignore = "probe-first: W1b mechanism pending — un-ignore in the W1b implementation commit"]
+fn probe_kill_after_cast_replay() {
+    let mut writer: BatchWriter<()> = BatchWriter::new();
+
+    let moves = vec![make_move(
+        11,
+        KanbanColumn::Planning,
+        KanbanColumn::CognitiveWork,
+        0,
+    )];
+
+    let cast = writer.cast(11, moves.clone(), ());
+    // Deliberately no ack() — simulates a crash between cast and ack.
+
+    let unacked = writer.unacked();
+    assert_eq!(unacked, vec![cast]);
+
+    let replayed = writer
+        .intent_moves(cast)
+        .expect("unacked cast must still have replayable intent moves");
+    assert!(!replayed.is_empty());
+    assert_eq!(replayed, moves.as_slice());
+}
+
+/// Probe 3 (W1c): resolve_owner() calls the resolver on the first lookup for
+/// a mailbox (cache miss) and skips it on the second lookup for the same
+/// mailbox (cache hit), returning the same owner both times.
+#[test]
+#[ignore = "probe-first: W1b mechanism pending — un-ignore in the W1b implementation commit"]
+fn probe_delegation_miss_then_hit() {
+    let mut writer: BatchWriter<()> = BatchWriter::new();
+
+    let on_behalf: u32 = 3;
+    let mut resolver_calls = 0u32;
+
+    let (owner_first, was_hit_first) = writer.resolve_owner(on_behalf, |mailbox| {
+        resolver_calls += 1;
+        mailbox + 1000 // arbitrary deterministic "resolved owner" transform
+    });
+    assert!(!was_hit_first, "first resolve for a mailbox must be a cache miss");
+    assert_eq!(resolver_calls, 1);
+
+    let (owner_second, was_hit_second) = writer.resolve_owner(on_behalf, |mailbox| {
+        resolver_calls += 1;
+        mailbox + 1000
+    });
+    assert!(was_hit_second, "second resolve for the same mailbox must be a cache hit");
+    assert_eq!(resolver_calls, 1, "resolver must not be called again on cache hit");
+    assert_eq!(owner_first, owner_second);
+}