feat(planner): W1b/W1c live — WAL batch writer implemented, 4 probes green; M15 MulGateDecision rename

W1b: BTreeMap board (monotonic CastId = cast order), ack(cast,
LanceVersion) with acked_version getter (the WAL<->temporal join,
Addendum-8), delegation cache, never-refuses stacking. All 4 probes
un-ignored and passing.

M15 (blocking-before-W2): planner-local GateDecision{Proceed,Sandbox,
Compass} -> MulGateDecision + deprecated alias; contract kanban gate
{Flow,Hold,Block} keeps the name; zero internal alias usage. Kills the
false friend before any planner->kanban emission.

Planner lib 204 green; clippy clean on all six touched files
(pre-existing nars_engine.rs:492 deprecated-CausalEdge64 debt noted,
untouched).

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

Author: claude

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

@@ -74,7 +74,7 @@ use crate::selector::StrategySelector;
 use crate::traits::*;
 
 // Re-export key types for ergonomic API
-pub use crate::mul::gate::GateDecision as Gate;
+pub use crate::mul::gate::MulGateDecision as Gate;
 pub use crate::mul::SituationInput;
 pub use crate::thinking::style::{FieldModulation, ScanParams, ThinkingCluster, ThinkingStyle};
 pub use crate::thinking::ThinkingContext;

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

@@ -1,6 +1,8 @@
 //! 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;
+//! `cast()` records intent moves AHEAD of any storage ack; `ack()` confirms
+//! at the `LanceVersion` the sink assigned (the CastId↔LanceVersion join
+//! wiring the WAL into the temporal classifier; see `crate::temporal`);
 //! `unacked()` is the crash-replay surface. Payload-generic: the writer never
 //! inspects `P` (DTO purity — ownership rides the cast pairing, never the DTO).
 //!
@@ -22,13 +24,19 @@
 //! mint a parallel `KanbanMove`; see the D-MBX-A6 Outcome adapter context in
 //! `crate::strategy::style_strategy`.
 
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 
 use lance_graph_contract::collapse_gate::MailboxId;
 use lance_graph_contract::kanban::KanbanMove;
 
+use crate::temporal::LanceVersion;
+
 /// Identity of one `cast()` — a write-ahead intent record on the kanban board.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+///
+/// `next_id` is monotonically increasing, so `CastId`'s derived `Ord` gives
+/// insertion order — this is what lets `board`/`acked` use `BTreeMap` and
+/// have `unacked()` iterate in cast order for free.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct CastId(pub u64);
 
 /// Ahead-firing batch writer: intent (`cast`) is visible on the board before
@@ -40,10 +48,15 @@ pub struct BatchWriter<P> {
     /// 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>,
+    /// `BTreeMap` (not `HashMap`) so iteration is in ascending `CastId` order
+    /// — i.e. cast (insertion) order, matching `unacked()`'s ordering contract.
+    board: BTreeMap<CastId, (MailboxId, Vec<KanbanMove>)>,
+    /// Casts that have been confirmed (acked), mapped to the `LanceVersion`
+    /// the sink assigned when it flushed — the CastId↔LanceVersion join
+    /// wiring the WAL into the temporal classifier (see `crate::temporal`).
+    /// A cast present in `board` but absent from `acked` is the crash-replay
+    /// surface (`unacked()`).
+    acked: BTreeMap<CastId, LanceVersion>,
     /// 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`).
@@ -61,8 +74,8 @@ impl<P> BatchWriter<P> {
     pub fn new() -> Self {
         Self {
             next_id: 0,
-            board: HashMap::new(),
-            acked: std::collections::HashSet::new(),
+            board: BTreeMap::new(),
+            acked: BTreeMap::new(),
             delegation_cache: HashMap::new(),
             pending_payloads: Vec::new(),
         }
@@ -71,47 +84,58 @@ impl<P> BatchWriter<P> {
     /// 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")
+    /// "Melden macht frei" (plan Addendum-7): casting is REPORTING, never
+    /// refused because earlier casts on the same mailbox are still unacked —
+    /// stacked casts are stacked WAL entries, each with its own `CastId`.
+    pub fn cast(&mut self, on_behalf: MailboxId, moves: Vec<KanbanMove>, payload: P) -> CastId {
+        let cast = CastId(self.next_id);
+        self.next_id += 1;
+        self.board.insert(cast, (on_behalf, moves));
+        self.pending_payloads.push((cast, payload));
+        cast
     }
 
-    /// 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")
+    /// Confirmation — marks the cast acked at the Lance version the sink
+    /// assigned (the CastId↔LanceVersion join wiring the WAL into the
+    /// temporal classifier; see planner `temporal.rs`).
+    pub fn ack(&mut self, cast: CastId, version: LanceVersion) {
+        self.acked.insert(cast, version);
     }
 
-    /// 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`.
+    /// The `LanceVersion` a cast was acked at, if it has been acked.
+    #[must_use]
+    pub fn acked_version(&self, cast: CastId) -> Option<LanceVersion> {
+        self.acked.get(&cast).copied()
+    }
+
+    /// Crash-replay surface (M24): casts recorded but not yet acked, in
+    /// ascending `CastId` (cast) order.
+    #[must_use]
     pub fn unacked(&self) -> Vec<CastId> {
-        todo!("W1b")
+        self.board
+            .keys()
+            .filter(|cast| !self.acked.contains_key(cast))
+            .copied()
+            .collect()
     }
 
     /// 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")
+    #[must_use]
+    pub fn intent_moves(&self, cast: CastId) -> Option<&[KanbanMove]> {
+        self.board.get(&cast).map(|(_, moves)| moves.as_slice())
     }
 
     /// 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,
+        on_behalf: MailboxId,
+        resolver: impl FnOnce(MailboxId) -> MailboxId,
     ) -> (MailboxId, bool) {
-        todo!("W1c")
+        if let Some(&owner) = self.delegation_cache.get(&on_behalf) {
+            return (owner, true);
+        }
+        let owner = resolver(on_behalf);
+        self.delegation_cache.insert(on_behalf, owner);
+        (owner, false)
     }
 }

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

@@ -93,7 +93,7 @@ pub mod temporal;
 mod orchestration_impl;
 
 use ir::LogicalPlan;
-use mul::{GateDecision, MulAssessment};
+use mul::{MulAssessment, MulGateDecision};
 use selector::StrategySelector;
 use thinking::ThinkingContext;
 use traits::{PlanContext, PlanStrategy, QueryFeatures};
@@ -181,7 +181,7 @@ impl PlannerAwareness {
         let gate = mul::gate_check(&mul_assessment);
 
         match gate {
-            GateDecision::Proceed { free_will_modifier } => {
+            MulGateDecision::Proceed { free_will_modifier } => {
                 // === LAYER 2: THINKING ORCHESTRATION ===
                 let thinking_ctx =
                     thinking::orchestrate(query, &mul_assessment, &plan::PlannerConfig::default());
@@ -219,8 +219,8 @@ impl PlannerAwareness {
                     emitted_edges: Vec::new(),
                 })
             }
-            GateDecision::Sandbox { reason } => Err(PlanError::GateBlocked { reason }),
-            GateDecision::Compass => {
+            MulGateDecision::Sandbox { reason } => Err(PlanError::GateBlocked { reason }),
+            MulGateDecision::Compass => {
                 let compass = mul::compass::navigate(query, &mul_assessment);
                 match compass.decision {
                     mul::compass::CompassDecision::SurfaceToMeta => Err(PlanError::SurfaceToMeta {

crates/lance-graph-planner/src/mul/gate.rs

@@ -10,8 +10,13 @@
 use super::{dk::DkPosition, homeostasis::FlowState, trust::TrustTexture, MulAssessment};
 
 /// Gate decision.
+///
+/// Renamed from `GateDecision` (M15) — this is the planner's
+/// Meta-Uncertainty-Layer verdict (Proceed/Sandbox/Compass), NOT the
+/// contract's kanban gate (`lance_graph_contract::mul::GateDecision
+/// {Flow, Hold, Block}`) which `KanbanColumn::advance_on_gate` consumes.
 #[derive(Debug, Clone)]
-pub enum GateDecision {
+pub enum MulGateDecision {
     /// All checks pass. Proceed with free will modifier applied.
     Proceed { free_will_modifier: f64 },
     /// Gate blocked. Need sandbox or human assistance.
@@ -20,36 +25,43 @@ pub enum GateDecision {
     Compass,
 }
 
+/// Renamed to `MulGateDecision` (M15); the unqualified name collided with
+/// `lance_graph_contract::mul::GateDecision`.
+#[deprecated(
+    note = "renamed to MulGateDecision (M15); the unqualified name collided with contract mul::GateDecision"
+)]
+pub type GateDecision = MulGateDecision;
+
 /// Check the gate conditions.
-pub fn check(assessment: &MulAssessment) -> GateDecision {
+pub fn check(assessment: &MulAssessment) -> MulGateDecision {
     // Check 1: Not Mount Stupid
     if assessment.dk_position == DkPosition::MountStupid {
-        return GateDecision::Sandbox {
+        return MulGateDecision::Sandbox {
             reason: "Dunning-Kruger: Mount Stupid detected. Learn first.".into(),
         };
     }
 
     // Check 2: Complexity mapped
     if !assessment.complexity_mapped {
-        return GateDecision::Sandbox {
+        return MulGateDecision::Sandbox {
             reason: "Complexity not mapped. Map dimensions before proceeding.".into(),
         };
     }
 
     // Check 3: Not depleted
     if assessment.homeostasis.needs_recovery {
-        return GateDecision::Sandbox {
+        return MulGateDecision::Sandbox {
             reason: "Depleted. Recover homeostasis first.".into(),
         };
     }
 
     // Check 4: Trust not murky/dissonant
     match assessment.trust.texture {
         TrustTexture::Murky => {
-            return GateDecision::Compass; // Can navigate with compass
+            return MulGateDecision::Compass; // Can navigate with compass
         }
         TrustTexture::Dissonant => {
-            return GateDecision::Sandbox {
+            return MulGateDecision::Sandbox {
                 reason: "Trust dissonant. Resolve dissonance before proceeding.".into(),
             };
         }
@@ -58,11 +70,11 @@ pub fn check(assessment: &MulAssessment) -> GateDecision {
 
     // Check 5: If trust is fuzzy, use compass for navigation
     if assessment.trust.texture == TrustTexture::Fuzzy {
-        return GateDecision::Compass;
+        return MulGateDecision::Compass;
     }
 
     // All checks pass
-    GateDecision::Proceed {
+    MulGateDecision::Proceed {
         free_will_modifier: assessment.free_will_modifier,
     }
 }

crates/lance-graph-planner/src/mul/mod.rs

@@ -23,7 +23,7 @@ pub mod trust;
 
 pub use compass::{CompassDecision, CompassResult};
 pub use dk::{DkDetector, DkPosition};
-pub use gate::GateDecision;
+pub use gate::MulGateDecision;
 pub use homeostasis::{FlowState, Homeostasis};
 pub use trust::{TrustQualia, TrustTexture};
 
@@ -123,7 +123,7 @@ pub fn assess(input: &SituationInput) -> MulAssessment {
 }
 
 /// Gate check: determine whether to proceed, sandbox, or use compass.
-pub fn gate_check(assessment: &MulAssessment) -> GateDecision {
+pub fn gate_check(assessment: &MulAssessment) -> MulGateDecision {
     gate::check(assessment)
 }
 
@@ -152,7 +152,7 @@ mod tests {
         let gate = gate_check(&assessment);
 
         // Mount Stupid should block
-        matches!(gate, GateDecision::Sandbox { .. });
+        matches!(gate, MulGateDecision::Sandbox { .. });
     }
 
     #[test]
@@ -171,6 +171,6 @@ mod tests {
         let assessment = assess(&input);
         let gate = gate_check(&assessment);
 
-        matches!(gate, GateDecision::Proceed { .. });
+        matches!(gate, MulGateDecision::Proceed { .. });
     }
 }

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

@@ -1,9 +1,8 @@
-//! D-V3-W1e probe-first: three failing probes for the W1b ahead-firing batch
-//! writer + W1c delegation cache, pinned against
+//! D-V3-W1e probe-first: four 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.
+//! Live (W1b implementation landed) — all four run and pass.
 //!
 //! Uses the REAL shipped kanban contract types
 //! (`lance_graph_contract::kanban::{KanbanColumn, KanbanMove, ExecTarget}`,
@@ -29,7 +28,6 @@ fn make_move(mailbox: u32, from: KanbanColumn, to: KanbanColumn, witness: u32) -
 /// 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();
 
@@ -44,16 +42,18 @@ fn probe_ahead_update_ordering() {
     // AHEAD: intent is visible on the board BEFORE any ack.
     assert_eq!(writer.intent_moves(cast), Some(moves.as_slice()));
 
-    writer.ack(cast);
+    writer.ack(cast, 1);
 
     // After ack, the cast is no longer in the unacked (crash-replay) surface.
     assert!(!writer.unacked().contains(&cast));
+
+    // The CastId<->LanceVersion join: the ack recorded the assigned version.
+    assert_eq!(writer.acked_version(cast), Some(1));
 }
 
 /// 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();
 
@@ -81,7 +81,6 @@ fn probe_kill_after_cast_replay() {
 /// 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();
 
@@ -113,7 +112,6 @@ fn probe_delegation_miss_then_hit() {
 /// for a row — is sink-side behavior, exercised in the W1b implementation
 /// tests, not at this API surface.)
 #[test]
-#[ignore = "probe-first: W1b mechanism pending — un-ignore in the W1b implementation commit"]
 fn probe_stacked_casts_never_refused() {
     let mut writer: BatchWriter<()> = BatchWriter::new();
 
@@ -135,9 +133,9 @@ fn probe_stacked_casts_never_refused() {
     assert!(writer.intent_moves(c3).is_some());
 
     // Acks retire independently and in any order.
-    writer.ack(c2);
+    writer.ack(c2, 1);
     assert_eq!(writer.unacked(), vec![c1, c3]);
-    writer.ack(c1);
-    writer.ack(c3);
+    writer.ack(c1, 2);
+    writer.ack(c3, 3);
     assert!(writer.unacked().is_empty());
 }