From 5cfa95f179515c5126a6b5f63ebd4a9a2ae8f450 Mon Sep 17 00:00:00 2001
From: Levi Neuwirth <ln@levineuwirth.org>
Date: Tue, 28 Jul 2026 20:55:02 -0400
Subject: [PATCH] Implement the partial B4 crash-matrix and benchmark slice

Scope 6.6 deliverables 1-3, partial. This is evidence, not a freeze: no
Wave B deliverable set is complete and the throughput figure below is not
a publishable bundle.

The revert-and-observe-red acceptance for cd37f8b's two generators was
never actually performed -- the tests asserted it in their doc comments,
which is a claim. It is performed now, in a throwaway copy, one mutation
at a time, and each reddens exactly one test for the right reason. With
segment frame validation reverted to pushing footer offsets straight into
the adopted set, the store still opens and recovery still completes: it
reports recovery_ok=true and adopts the corrupted frame as authority, so
the red is exit 0 where 65 is required rather than a store that failed to
open. With the shard-index check removed from journal binding, the moved
journal is adopted whole and root_uuid alone does not catch it -- the Wave
A blocker reproduced.

The eight Wave B failpoint rows now drive StoreEngine::submit in-process
with the full six-field expectation asserted against the frozen oracle,
field by field rather than by struct comparison. No pending-wave-b row
remains. The four fields Wave A could not reach come from four independent
observations: the store naming itself poisoned, the two-root status read,
a receipt obtainable at all, and a different transaction submitted to the
same shard before any reopen. The last two are not the same question -- a
shard can be unpoisoned and still refuse a later append because its writer
thread died, which is exactly what phase-aware panic ownership fixed and
what this now checks from outside. AfterRootCasBeforeWaiterWake is
observed as a genuinely hung submit whose receipt is still retrievable,
so a waiter that never wakes is proved to be a hung request rather than an
absent transaction. Every row also asserts its group's fence count against
the public durability snapshot.

Two flake campaigns were measured rather than rerun: 4 failures in 40, then
7 in 40, from two distinct causes. One is a finding -- publishing a group
adds an index delta layer and none are sealed, so submit refuses after
exactly max_index_runs publications for the life of an engine. Both fixed
structurally; 200/200 and 40/40 after.

store-bench emit-skeleton now defaults to the submit path, with the journal
seam retained under --path drive for comparison. The signer is real, the
ref CAS is evaluated by the sequencer, and objects_new is summed from the
store's own receipts rather than multiplied out of the transaction count.

Explicit blockers, retained rather than worked around:

  - StoreEngine::open still refuses startup state 1, so the benchmark seeds
    its root by a non-production path. Seeding a store off the production
    path in order to measure the production path is the charter item 8
    smell; the disclosure is recorded in the fixture, a const doc, and the
    module docs, and a test asserts open still refuses so it cannot go
    stale in the safe direction.
  - P2 is blocked three ways -- checkpointing disabled, no steady state
    under the index-run ceiling, and no warmup/repetition/trim protocol.
    The rate emitted is a debug build on tmpfs, marked preliminary.
  - The 100 SIGKILL cycles still drive the journal seam, so kill -9 never
    lands inside a real publication.
  - Four schema claims became earnable and are requested, not emitted;
    bench/result-schema.json is lead-owned.
  - Reopen after close needs a bounded, measured, reported wait, because
    the root LOCK outlives StoreEngine::drop. Diagnosed since as fork/exec
    inheritance of the lock file description; the fix belongs in the lock
    primitive, and this wait is removed when it lands.

scripts/check-phase1.sh GATE_EXIT=0; verify-store-recovery.sh 100 cycles,
recovery_failures=0, acknowledged_loss=0, torn_transactions=0,
repeated_adoptions=0, bundle=schema-valid.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 crates/levcs-store/src/bin/store-bench.rs     | 765 +++++++++++++-
 .../levcs-store/src/bin/store-crash-driver.rs |  22 +-
 crates/levcs-store/tests/crash_matrix.rs      | 466 ++++++++-
 .../tests/fixtures/phase1-failpoints.json     | 111 +-
 .../tests/support/engine_matrix.rs            | 979 ++++++++++++++++++
 crates/levcs-store/tests/support/harness.rs   |  68 +-
 crates/levcs-store/tests/support/mod.rs       |  12 +
 7 files changed, 2336 insertions(+), 87 deletions(-)
 create mode 100644 crates/levcs-store/tests/support/engine_matrix.rs

diff --git a/crates/levcs-store/src/bin/store-bench.rs b/crates/levcs-store/src/bin/store-bench.rs
index f38cb8c..f602150 100644
--- a/crates/levcs-store/src/bin/store-bench.rs
+++ b/crates/levcs-store/src/bin/store-bench.rs
@@ -39,11 +39,31 @@
 //!
 //! # What is blocked
 //!
-//! A full P2 bundle goes through `StoreEngine::submit`, which is B1's. In Wave
-//! A this binary compiles, its prechecks and its bundle emitter run against a
-//! short `drive.rs` run, and `emit-skeleton` produces a schema-valid bundle
-//! whose `verdicts` are all `not-applicable` and whose `run_id` says
-//! `skeleton`.
+//! `emit-skeleton --path submit` now drives the production `StoreEngine::submit`
+//! (scope 6.6 deliverable 3). It is still not a P2 run, and three of B1's
+//! unimplemented deliverables are why — every one of them a bound on the
+//! bundle, not merely on this file:
+//!
+//!   * `StoreEngine::open` refuses startup state 1, so the root is created by
+//!     `segment::initialize_root`. The measured path is production; the path
+//!     that built the store it measures is not.
+//!   * `submit` refuses after `max_index_runs` group publications, because
+//!     sealing the in-memory index delta into an `IndexRun` is unimplemented.
+//!     The ceiling is raised for the run, which means every delta layer ever
+//!     published is still resident and lookup fan-out grows for the whole run.
+//!     P2 measures a steady state; this is not one.
+//!   * `StoreEngine::checkpoint` is unimplemented, so no checkpoint is taken.
+//!     Section 7 requires that a P2 run not have been achieved with
+//!     checkpointing disabled. This one was.
+//!
+//! The bundle records none of those three, because `bench/result-schema.json`
+//! is `additionalProperties: false` throughout and has no field for the
+//! conditions a run was produced under. That is an amendment request to the
+//! lead, not an edit: a bundle whose caveats live only in a report is a bundle
+//! that reads as unconditional to everyone who receives it.
+//!
+//! The `run` subcommand — warmup, three repetitions, per-repetition fresh
+//! roots, trim settle — stays blocked on the same three.
 //!
 //! Real Ed25519 attestation signing is also blocked: `levcs-store` has no
 //! signing dependency and scope §1 forbids any `levcs_identity::` path in this
@@ -1487,6 +1507,682 @@ fn skeleton_ack_record(sequence: u64) -> AckRecord {
     }
 }
 
+// ---------------------------------------------------------------------------
+// The Wave B run: through StoreEngine::submit
+// ---------------------------------------------------------------------------
+//
+// Scope 6.6 deliverable 3. The Wave A run drove `drive.rs`, the journal seam:
+// no sequencer, no signer, no status root, no index, no receipts. This one goes
+// through `StoreEngine::submit` — the entry point a consumer calls — so the
+// numbers are the store's, not the journal's. **Expect different numbers.**
+//
+// # What this run can and cannot claim, stated before the code
+//
+// Three of B1's unimplemented deliverables bound it, and every one of them is a
+// bound on the *bundle*, not merely on this file:
+//
+//   * `StoreEngine::open` refuses startup state 1, so the root is created by
+//     `segment::initialize_root`. The measured path is production; the path
+//     that made the store it measures is not.
+//   * `submit` refuses `NotImplemented` after `max_index_runs` group
+//     publications, because sealing the in-memory index delta into an
+//     `IndexRun` is unimplemented. The ceiling is raised here so the run can
+//     reach its measured seconds at all, which means the run holds every delta
+//     layer it ever published in memory and its lookup fan-out grows for the
+//     whole run. A P2 measurement is of a steady state; this is not one, and
+//     the bundle says so through `outcome` and its verdicts.
+//   * `StoreEngine::checkpoint` is unimplemented, so no checkpoint is taken.
+//     Section 7 requires that a P2 run not have been achieved with
+//     checkpointing disabled. This one was. That alone makes the
+//     `storage_primitive` gate unearnable today, whatever the rate says.
+//
+// What genuinely improves over Wave A: the signer is real Ed25519 and its cost
+// is measured rather than reported as a zero; every commit carries the
+// canonical three objects and a typed ref CAS, so `objects_new` is counted from
+// what the store staged instead of multiplied out of the commit count; and the
+// reconciliation reads each acknowledged operation back through
+// `transaction_status` on a reopened engine rather than comparing sequence
+// sets.
+
+/// Raised because index-delta sealing is unimplemented; see the note above.
+const ENGINE_MAX_INDEX_RUNS: u32 = 1_000_000;
+
+/// One commit's objects, matching the frozen workload: one 1 KiB blob, one
+/// tree, one commit.
+const ENGINE_BLOB_BYTES: usize = 1024;
+const ENGINE_TREE_BYTES: usize = 96;
+const ENGINE_COMMIT_BYTES: usize = 192;
+
+/// A real Ed25519 signer that records what signing cost.
+///
+/// Scope 5.2 requires the bundle to report signing cost separately, and the
+/// frozen workload's `[identity] real_ed25519 = true`. The harness owns the key
+/// because the store may not call into `levcs-identity` (scope §1); the library
+/// only ever sees the `CommitEvidenceSigner` trait.
+struct MeasuringSigner {
+    key: ed25519_dalek::SigningKey,
+    micros: std::sync::Mutex<Vec<u64>>,
+}
+
+impl MeasuringSigner {
+    fn new() -> Self {
+        // Deterministic, and deliberately so: the bundle has to be
+        // reproducible, and this key authenticates nothing outside the run.
+        let mut seed = [0u8; 32];
+        blake3::Hasher::new()
+            .update(b"levcs-store/store-bench/attestation-key/v1\0")
+            .finalize_xof()
+            .fill(&mut seed);
+        Self {
+            key: ed25519_dalek::SigningKey::from_bytes(&seed),
+            micros: std::sync::Mutex::new(Vec::new()),
+        }
+    }
+}
+
+impl levcs_store::types::CommitEvidenceSigner for MeasuringSigner {
+    fn key_epoch(&self) -> u64 {
+        1
+    }
+
+    fn public_key(&self) -> [u8; 32] {
+        self.key.verifying_key().to_bytes()
+    }
+
+    fn sign_event(
+        &self,
+        signing_digest: &ObjectId,
+    ) -> Result<[u8; 64], levcs_store::types::SignerError> {
+        use ed25519_dalek::Signer as _;
+        let started = Instant::now();
+        let signature = self.key.sign(signing_digest.as_bytes()).to_bytes();
+        let elapsed = started.elapsed().as_micros() as u64;
+        self.micros
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .push(elapsed);
+        Ok(signature)
+    }
+}
+
+/// Drive one future to completion on this thread.
+///
+/// `levcs-store` starts no runtime and takes no executor dependency, so neither
+/// does its benchmark. A submit that never completes is a hung request, and the
+/// deadline exists so the harness can say so instead of blocking forever.
+fn block_on_until<F: std::future::Future>(future: F, deadline: Duration) -> Option<F::Output> {
+    use std::sync::Arc;
+    use std::task::{Context, Poll, Wake, Waker};
+
+    struct ThreadWaker(std::thread::Thread);
+    impl Wake for ThreadWaker {
+        fn wake(self: Arc<Self>) {
+            self.0.unpark();
+        }
+        fn wake_by_ref(self: &Arc<Self>) {
+            self.0.unpark();
+        }
+    }
+
+    let waker = Waker::from(Arc::new(ThreadWaker(std::thread::current())));
+    let mut context = Context::from_waker(&waker);
+    let mut future = std::pin::pin!(future);
+    let started = Instant::now();
+    loop {
+        match future.as_mut().poll(&mut context) {
+            Poll::Ready(output) => return Some(output),
+            Poll::Pending => {
+                let elapsed = started.elapsed();
+                if elapsed >= deadline {
+                    return None;
+                }
+                std::thread::park_timeout(deadline - elapsed);
+            }
+        }
+    }
+}
+
+fn engine_namespace(shard: u16, shard_count: u16) -> NamespaceId {
+    for attempt in 0..8192u64 {
+        let mut bytes = [0u8; 32];
+        let mut hasher = blake3::Hasher::new();
+        hasher.update(b"levcs-store/store-bench/engine-namespace/v1\0");
+        hasher.update(&u64::from(shard).to_le_bytes());
+        hasher.update(&attempt.to_le_bytes());
+        hasher.finalize_xof().fill(&mut bytes);
+        let namespace = NamespaceId(bytes);
+        if levcs_store::StoreOptions::shard_of(&namespace, shard_count) == shard {
+            return namespace;
+        }
+    }
+    panic!("no namespace routed to shard {shard}");
+}
+
+fn engine_object(domain: &str, seed: &[u8], len: usize) -> (ObjectId, Vec<u8>) {
+    let mut raw = vec![0u8; len];
+    let mut hasher = blake3::Hasher::new();
+    hasher.update(domain.as_bytes());
+    hasher.update(&[0u8]);
+    hasher.update(seed);
+    hasher.finalize_xof().fill(&mut raw);
+    (ObjectId(*blake3::hash(&raw).as_bytes()), raw)
+}
+
+fn engine_now_micros() -> i64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_micros() as i64)
+        .unwrap_or(0)
+}
+
+fn engine_evidence() -> levcs_protocol::v2::TransactionEvidenceV1 {
+    levcs_protocol::v2::TransactionEvidenceV1::AdministrativeV1 {
+        actor: [0x7e; 32],
+        actor_key_epoch: 1,
+        command_digest: ObjectId([0x33; 32]),
+        signature: [0x44; 64],
+    }
+}
+
+/// What one engine-driven run measured.
+struct EngineRun {
+    latencies_micros: Vec<u64>,
+    transactions: u64,
+    objects_new: u64,
+    raw_bytes: u64,
+    elapsed: Duration,
+    acknowledged: u64,
+    acknowledged_loss: u64,
+    torn_transactions: u64,
+    repeated_adoptions: u64,
+    fences: u64,
+    /// Group publications, counted as fences: `journal::append_group_and_fence`
+    /// performs exactly one per group and A1's acceptance pins that.
+    groups: u64,
+    signing_micros_p50: f64,
+    signings: u64,
+    refused: u64,
+    first_refusal: Option<String>,
+    acknowledged_sequences_reconciled: bool,
+    ack_journal_digest: String,
+    lock_release_attempts: u32,
+}
+
+#[allow(clippy::too_many_lines)]
+fn run_engine(
+    root: &Path,
+    ack_path: &Path,
+    group_len: usize,
+    seconds: u64,
+    shard_count: u16,
+    submitters_per_shard: usize,
+) -> Result<EngineRun, String> {
+    use levcs_store::segment::{initialize_root, RootLayout};
+    use levcs_store::transaction::StagedObject;
+    use levcs_store::types::{DurabilityCounters, OperationId, PrivilegedConstruction, StoreError};
+    use levcs_store::{StoreEngine, ValidatedTransaction};
+    use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+    use std::sync::{Arc, Mutex};
+
+    initialize_root(
+        &RootLayout::new(root),
+        shard_count,
+        [0x9e; 16],
+        engine_now_micros(),
+        &DurabilityCounters::default(),
+    )
+    .map_err(|e| format!("initialize_root: {e}"))?;
+
+    let signer = Arc::new(MeasuringSigner::new());
+    let build_options = || {
+        let mut options = levcs_store::StoreOptions::new(root);
+        options.shard_count = shard_count;
+        options.max_group_transactions = group_len as u32;
+        options.max_group_bytes = 8 * 1024 * 1024;
+        options.max_group_idle = Duration::from_millis(1);
+        options.journal_preallocate_bytes = 64 * 1024 * 1024;
+        options.max_index_runs = ENGINE_MAX_INDEX_RUNS;
+        options.signer = Some(signer.clone());
+        options
+    };
+
+    let engine = StoreEngine::open(build_options()).map_err(|e| format!("open: {e}"))?;
+
+    let namespaces: Vec<NamespaceId> = (0..shard_count)
+        .map(|shard| engine_namespace(shard, shard_count))
+        .collect();
+
+    let ack =
+        Mutex::new(ExternalAckJournal::open(ack_path).map_err(|e| format!("ack journal: {e}"))?);
+    let acknowledged = AtomicU64::new(0);
+
+    // One repository per shard, created before the measured window.
+    for (shard, namespace) in namespaces.iter().enumerate() {
+        let (genesis, raw) = engine_object(
+            "levcs-store/store-bench/genesis/v1",
+            &(shard as u64).to_le_bytes(),
+            64,
+        );
+        let transaction = ValidatedTransaction::builder(PrivilegedConstruction::assert_validated())
+            .namespace(*namespace)
+            .operation(
+                OperationId([0xc0 | shard as u8; 16]),
+                ObjectId([0xc0 | shard as u8; 32]),
+                engine_now_micros() + 600_000_000,
+            )
+            .create_repository(genesis)
+            .objects(vec![StagedObject {
+                id: genesis,
+                object_type: levcs_core::ObjectType::Authority,
+                raw,
+            }])
+            .refs(Vec::new())
+            .authority(None, Some(genesis))
+            .evidence(engine_evidence())
+            .build()
+            .map_err(|e| format!("build create: {e}"))?;
+        match block_on_until(engine.submit(transaction), Duration::from_secs(30)) {
+            Some(Ok(_)) => {}
+            Some(Err(error)) => return Err(format!("creating repository {shard}: {error}")),
+            None => return Err(format!("creating repository {shard} never completed")),
+        }
+    }
+
+    let stop = AtomicBool::new(false);
+    let refused = AtomicU64::new(0);
+    let first_refusal: Mutex<Option<String>> = Mutex::new(None);
+    let results: Mutex<Vec<(NamespaceId, OperationId, u64, u64)>> = Mutex::new(Vec::new());
+    let latencies: Mutex<Vec<u64>> = Mutex::new(Vec::new());
+    let objects_new = AtomicU64::new(0);
+    let raw_bytes = AtomicU64::new(0);
+
+    let started = Instant::now();
+    let deadline = started + Duration::from_secs(seconds.max(1));
+
+    std::thread::scope(|scope| {
+        for shard in 0..shard_count {
+            for submitter in 0..submitters_per_shard {
+                let namespace = namespaces[shard as usize];
+                let engine = &engine;
+                let ack = &ack;
+                let acknowledged = &acknowledged;
+                let stop = &stop;
+                let refused = &refused;
+                let first_refusal = &first_refusal;
+                let results = &results;
+                let latencies = &latencies;
+                let objects_new = &objects_new;
+                let raw_bytes = &raw_bytes;
+                scope.spawn(move || {
+                    let mut ordinal = 0u64;
+                    let branch = format!("refs/heads/s{shard}-w{submitter}");
+                    let mut expected: Option<ObjectId> = None;
+                    let (authority, _) = engine_object(
+                        "levcs-store/store-bench/genesis/v1",
+                        &(shard as u64).to_le_bytes(),
+                        64,
+                    );
+                    while !stop.load(Ordering::Relaxed) && Instant::now() < deadline {
+                        ordinal += 1;
+                        let mut seed = Vec::with_capacity(24);
+                        seed.extend_from_slice(&u64::from(shard).to_le_bytes());
+                        seed.extend_from_slice(&(submitter as u64).to_le_bytes());
+                        seed.extend_from_slice(&ordinal.to_le_bytes());
+
+                        let (blob, blob_raw) = engine_object(
+                            "levcs-store/store-bench/blob/v1",
+                            &seed,
+                            ENGINE_BLOB_BYTES,
+                        );
+                        let (tree, tree_raw) = engine_object(
+                            "levcs-store/store-bench/tree/v1",
+                            &seed,
+                            ENGINE_TREE_BYTES,
+                        );
+                        let (commit, commit_raw) = engine_object(
+                            "levcs-store/store-bench/commit/v1",
+                            &seed,
+                            ENGINE_COMMIT_BYTES,
+                        );
+                        let bytes = (blob_raw.len() + tree_raw.len() + commit_raw.len()) as u64;
+
+                        let mut operation_id = [0u8; 16];
+                        operation_id[..8].copy_from_slice(&ordinal.to_le_bytes());
+                        operation_id[8] = shard as u8;
+                        operation_id[9] = submitter as u8;
+                        let operation = OperationId(operation_id);
+
+                        let transaction = ValidatedTransaction::builder(
+                            PrivilegedConstruction::assert_validated(),
+                        )
+                        .namespace(namespace)
+                        .operation(
+                            operation,
+                            ObjectId(*blake3::hash(&seed).as_bytes()),
+                            engine_now_micros() + 600_000_000,
+                        )
+                        .objects(vec![
+                            StagedObject {
+                                id: blob,
+                                object_type: levcs_core::ObjectType::Blob,
+                                raw: blob_raw,
+                            },
+                            StagedObject {
+                                id: tree,
+                                object_type: levcs_core::ObjectType::Tree,
+                                raw: tree_raw,
+                            },
+                            StagedObject {
+                                id: commit,
+                                object_type: levcs_core::ObjectType::Commit,
+                                raw: commit_raw,
+                            },
+                        ])
+                        // A real typed ref CAS per commit. The bundle's
+                        // validation_flags claim `typed_ref_cas: true`, and at
+                        // this gate that claim is only worth anything if the
+                        // sequencer actually evaluated one.
+                        .refs(vec![levcs_protocol::v2::TypedRefCas {
+                            target: levcs_protocol::v2::RefTarget::Branch(branch.clone()),
+                            expected,
+                            mutation: levcs_protocol::v2::RefMutation::Set(commit),
+                            force: false,
+                        }])
+                        .authority(Some(authority), Some(authority))
+                        .evidence(engine_evidence())
+                        .build()
+                        .expect("a complete push transaction");
+
+                        let call = Instant::now();
+                        let outcome =
+                            block_on_until(engine.submit(transaction), Duration::from_secs(60));
+                        let micros = call.elapsed().as_micros() as u64;
+                        match outcome {
+                            Some(Ok(receipt)) => {
+                                // The fence returned. Acknowledgment is only
+                                // permitted once the acknowledgment is itself
+                                // durable on an independent journal, so this
+                                // happens before anything counts the commit.
+                                let record = AckRecord {
+                                    repo_id: ObjectId(*namespace.as_bytes()),
+                                    operation_id,
+                                    operation_digest: ObjectId(*blake3::hash(&seed).as_bytes()),
+                                    receipt_digest: ObjectId(
+                                        *blake3::hash(&operation_id).as_bytes(),
+                                    ),
+                                    repo_sequence: receipt.repo_sequence,
+                                    blob_ids: vec![blob],
+                                    tree_ids: vec![tree],
+                                    commit_ids: vec![commit],
+                                };
+                                {
+                                    let mut journal = ack.lock().unwrap_or_else(|p| p.into_inner());
+                                    if journal.append_durable(&record).is_err() {
+                                        stop.store(true, Ordering::Relaxed);
+                                        return;
+                                    }
+                                }
+                                acknowledged.fetch_add(1, Ordering::Relaxed);
+                                // The store's own count of the objects this
+                                // transaction introduced, not `3` restated by
+                                // the harness. `objects_new` derived from the
+                                // commit count is what makes
+                                // `verification.objects_new_equals_three_per_commit`
+                                // a tautology instead of a check.
+                                objects_new.fetch_add(receipt.objects_new, Ordering::Relaxed);
+                                raw_bytes.fetch_add(bytes, Ordering::Relaxed);
+                                latencies
+                                    .lock()
+                                    .unwrap_or_else(|p| p.into_inner())
+                                    .push(micros);
+                                results.lock().unwrap_or_else(|p| p.into_inner()).push((
+                                    namespace,
+                                    operation,
+                                    receipt.repo_sequence,
+                                    ordinal,
+                                ));
+                                expected = Some(commit);
+                            }
+                            Some(Err(error)) => {
+                                refused.fetch_add(1, Ordering::Relaxed);
+                                let mut slot =
+                                    first_refusal.lock().unwrap_or_else(|p| p.into_inner());
+                                if slot.is_none() {
+                                    *slot = Some(format!("{error:?}"));
+                                }
+                                stop.store(true, Ordering::Relaxed);
+                                return;
+                            }
+                            None => {
+                                refused.fetch_add(1, Ordering::Relaxed);
+                                let mut slot =
+                                    first_refusal.lock().unwrap_or_else(|p| p.into_inner());
+                                if slot.is_none() {
+                                    *slot = Some("submit did not complete in 60s".to_string());
+                                }
+                                stop.store(true, Ordering::Relaxed);
+                                return;
+                            }
+                        }
+                    }
+                });
+            }
+        }
+    });
+
+    let elapsed = started.elapsed();
+    let fences: u64 = (0..shard_count)
+        .map(|shard| {
+            engine
+                .durability_counters(shard)
+                .map(|counters| counters.fdatasync)
+                .unwrap_or(0)
+        })
+        .sum();
+
+    let latencies = latencies.into_inner().unwrap_or_else(|p| p.into_inner());
+    let results = results.into_inner().unwrap_or_else(|p| p.into_inner());
+    let refused = refused.load(Ordering::Relaxed);
+    let first_refusal = first_refusal
+        .into_inner()
+        .unwrap_or_else(|p| p.into_inner());
+    let acknowledged = acknowledged.load(Ordering::Relaxed);
+    let objects_new = objects_new.load(Ordering::Relaxed);
+    let raw_bytes = raw_bytes.load(Ordering::Relaxed);
+    let mut signing = signer
+        .micros
+        .lock()
+        .unwrap_or_else(|p| p.into_inner())
+        .clone();
+    signing.sort_unstable();
+    let signings = signing.len() as u64;
+    let signing_p50 = percentile(&signing, 0.50) as f64;
+
+    drop(ack);
+    drop(engine);
+
+    // The root lock is not always free when `StoreEngine::drop` returns; see
+    // the same finding recorded in `tests/support/engine_matrix.rs`. Bounded
+    // and reported, never silent.
+    let lock_wait_started = Instant::now();
+    let mut lock_release_attempts = 0u32;
+    let reopened = loop {
+        lock_release_attempts += 1;
+        match StoreEngine::open(build_options()) {
+            Ok(engine) => break engine,
+            Err(StoreError::AlreadyLocked)
+                if lock_wait_started.elapsed() < Duration::from_secs(30) =>
+            {
+                std::thread::sleep(Duration::from_micros(200));
+            }
+            Err(other) => return Err(format!("reopen through production recovery: {other}")),
+        }
+    };
+
+    // Reconciliation, through the production status read rather than through a
+    // sequence-set comparison. Every operation this run acknowledged must read
+    // back `Committed` from a store that was closed and recovered.
+    let mut acknowledged_loss = 0u64;
+    for (namespace, operation, _, _) in &results {
+        // Every variant named. A fallback arm here would fold "the store
+        // refused the read" into "the operation is missing", and those are
+        // different findings: the first invalidates the reconciliation, the
+        // second invalidates the run.
+        use levcs_store::types::TransactionStatus as Status;
+        match reopened.transaction_status(*namespace, *operation) {
+            Ok(Status::Committed(_)) => {}
+            Ok(Status::Unknown) => acknowledged_loss += 1,
+            Ok(Status::Resolving { .. }) => acknowledged_loss += 1,
+            Ok(Status::Pending { .. }) => acknowledged_loss += 1,
+            Ok(Status::Expired { .. }) => acknowledged_loss += 1,
+            Err(error) => {
+                return Err(format!(
+                    "reading back an acknowledged operation failed: {error}. The \
+                     reconciliation cannot distinguish a lost commit from a failed read, \
+                     so the run is void rather than counted."
+                ))
+            }
+        }
+    }
+    drop(reopened);
+
+    // Per-repository sequence integrity, through the one shared checker.
+    let mut torn_transactions = 0u64;
+    let mut repeated_adoptions = 0u64;
+    for namespace in &namespaces {
+        let mut sequences: Vec<u64> = results
+            .iter()
+            .filter(|(candidate, _, _, _)| candidate == namespace)
+            .map(|(_, _, sequence, _)| *sequence)
+            .collect();
+        sequences.sort_unstable();
+        let classified = classify_adopted_set(&sequences);
+        torn_transactions += classified.missing_sequences;
+        repeated_adoptions += classified.repeated_adoptions();
+    }
+
+    let records = ExternalAckJournal::recover(ack_path).map_err(|e| format!("ack recover: {e}"))?;
+
+    Ok(EngineRun {
+        groups: fences,
+        transactions: acknowledged,
+        latencies_micros: latencies,
+        objects_new,
+        raw_bytes,
+        elapsed,
+        acknowledged,
+        acknowledged_loss,
+        torn_transactions,
+        repeated_adoptions,
+        fences,
+        signing_micros_p50: signing_p50,
+        signings,
+        refused,
+        first_refusal,
+        acknowledged_sequences_reconciled: acknowledged_loss == 0
+            && torn_transactions == 0
+            && repeated_adoptions == 0
+            && records.len() as u64 == acknowledged,
+        ack_journal_digest: digest_file(ack_path),
+        lock_release_attempts,
+    })
+}
+
+/// One measured run, whichever path produced it.
+///
+/// The two paths are the Wave A journal seam and the Wave B production
+/// `StoreEngine::submit`. Folding them into one shape here is what keeps the
+/// bundle emitter identical for both: the thing that must not differ between a
+/// seam measurement and a store measurement is how the measurement is
+/// *reported*.
+struct MeasuredRun {
+    latencies_micros: Vec<u64>,
+    groups: u64,
+    transactions: u64,
+    /// Counted from the objects the store actually staged on the submit path,
+    /// and derived as `transactions * 3` on the drive path, where there are no
+    /// objects. The difference is why
+    /// `verification.objects_new_equals_three_per_commit` stays forbidden at
+    /// this gate on the drive path: a derived figure asserted against its own
+    /// derivation is a tautology.
+    objects_new: u64,
+    objects_new_counted: bool,
+    bytes: u64,
+    elapsed: Duration,
+    acknowledged: u64,
+    acknowledged_loss: u64,
+    torn_transactions: u64,
+    repeated_adoptions: u64,
+    fences: u64,
+    signing_micros_p50: f64,
+    signings: u64,
+    acknowledged_sequences_reconciled: bool,
+    ack_journal_digest: String,
+    path: &'static str,
+    /// Submits refused or never completed. Non-zero ends the run: a benchmark
+    /// that keeps counting past a refusal is measuring a different workload
+    /// from the one it names.
+    refused: u64,
+    first_refusal: Option<String>,
+    /// Attempts the post-run reopen needed to acquire the root lock. `1` is the
+    /// expected reading; see the note on `run_engine`.
+    lock_release_attempts: u32,
+}
+
+impl From<SkeletonRun> for MeasuredRun {
+    fn from(run: SkeletonRun) -> Self {
+        Self {
+            groups: run.groups,
+            transactions: run.transactions,
+            objects_new: run.transactions * OBJECTS_PER_COMMIT,
+            objects_new_counted: false,
+            bytes: run.bytes,
+            elapsed: run.elapsed,
+            acknowledged: run.acknowledged,
+            acknowledged_loss: run.acknowledged_loss,
+            torn_transactions: run.torn_transactions,
+            repeated_adoptions: run.repeated_adoptions,
+            fences: run.fences,
+            signing_micros_p50: 0.0,
+            signings: 0,
+            acknowledged_sequences_reconciled: run.acknowledged_sequences_reconciled,
+            ack_journal_digest: run.ack_journal_digest,
+            latencies_micros: run.latencies_micros,
+            path: "drive",
+            refused: 0,
+            first_refusal: None,
+            lock_release_attempts: 1,
+        }
+    }
+}
+
+impl From<EngineRun> for MeasuredRun {
+    fn from(run: EngineRun) -> Self {
+        Self {
+            groups: run.groups,
+            transactions: run.transactions,
+            objects_new: run.objects_new,
+            objects_new_counted: true,
+            bytes: run.raw_bytes,
+            elapsed: run.elapsed,
+            acknowledged: run.acknowledged,
+            acknowledged_loss: run.acknowledged_loss,
+            torn_transactions: run.torn_transactions,
+            repeated_adoptions: run.repeated_adoptions,
+            fences: run.fences,
+            signing_micros_p50: run.signing_micros_p50,
+            signings: run.signings,
+            acknowledged_sequences_reconciled: run.acknowledged_sequences_reconciled,
+            ack_journal_digest: run.ack_journal_digest,
+            latencies_micros: run.latencies_micros,
+            path: "submit",
+            refused: run.refused,
+            first_refusal: run.first_refusal,
+            lock_release_attempts: run.lock_release_attempts,
+        }
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Section 13 stop conditions, measured rather than quoted
 // ---------------------------------------------------------------------------
@@ -1606,7 +2302,12 @@ store-bench <precheck|emit-skeleton|run> [flags]
   precheck        --root P [--repo-root P] [--target-rate N]
   emit-skeleton   --root P --out P --allow-unsigned [--repo-root P]
                   [--seconds N] [--group-len N] [--skip-attribute-check]
-  run             --root P --out-dir P   (P2; requires B1's StoreEngine::submit)
+                  [--path submit|drive] [--shards N] [--submitters-per-shard N]
+  run             --root P --out-dir P   (P2; blocked, see below)
+
+--path submit is the default and goes through StoreEngine::submit. --path drive
+is the Wave A journal seam, kept so the two measurements can be compared rather
+than confused; it signs nothing, creates no object, and issues no receipt.
 
 The two prechecks are not optional and are not overridable except by
 --skip-attribute-check, which makes the run non-comparable and marks the
@@ -1773,7 +2474,29 @@ fn emit_skeleton(
     // promised. Same device here, different subtree — the strongest isolation
     // an unprivileged in-process harness can give it, and the reason the
     // deployed campaigns of §10 put it on another host.
-    let run = run_skeleton(&root, &ack_path, group_len, seconds)?;
+    let path = flags.get("path").unwrap_or("submit").to_string();
+    let run: MeasuredRun = if path == "submit" {
+        let shards = flags.number::<u16>("shards", 4)?;
+        let submitters = flags.number::<usize>("submitters-per-shard", group_len.max(1))?;
+        run_engine(&root, &ack_path, group_len, seconds, shards, submitters)?.into()
+    } else if path == "drive" {
+        run_skeleton(&root, &ack_path, group_len, seconds)?.into()
+    } else {
+        return Err(format!(
+            "--path must be submit or drive, got {path:?}. `submit` is the production \
+             StoreEngine path and the default; `drive` is the Wave A journal seam, kept \
+             so the two measurements can be compared rather than confused."
+        ));
+    };
+
+    if run.refused != 0 {
+        return Err(format!(
+            "the run was cut short by {} refused or incomplete submit(s); the first was \
+             {:?}. A benchmark that keeps counting past a refusal is measuring a \
+             different workload from the one it names.",
+            run.refused, run.first_refusal
+        ));
+    }
 
     // Precheck 2 must run against a root that exists, so it follows the run
     // that creates it. In a P2 run the shard directories are created by
@@ -1819,7 +2542,11 @@ fn emit_skeleton(
     // Section 13's two stop conditions, measured against A2's index.
     let index_cost = measure_index_costs()?;
 
-    let mut run_id = String::from("skeleton-wave-a-");
+    let mut run_id = if run.path == "submit" {
+        String::from("engine-wave-b-")
+    } else {
+        String::from("skeleton-wave-a-")
+    };
     run_id.push_str(&digest_hex(histogram_input.as_bytes())[..16]);
 
     let inputs = BundleInputs {
@@ -1841,7 +2568,7 @@ fn emit_skeleton(
         // lands `StoreEngine::submit`, `objects_new` must be counted
         // independently — from the objects the store actually staged — before
         // that flag may be emitted at any gate.
-        objects_new: run.transactions * OBJECTS_PER_COMMIT,
+        objects_new: run.objects_new,
         raw_bytes: run.bytes,
         application_bytes: run.bytes,
         latency_p50: percentile(&sorted, 0.50),
@@ -1868,15 +2595,23 @@ fn emit_skeleton(
         // is reported as zero. Scope 8.4's ~1.4-of-8-cores figure is a
         // prediction for P2 and must not be copied into a bundle as if it had
         // been measured.
-        signing_cores: 0.0,
+        // Measured, not predicted. The submit path signs every event with a
+        // real Ed25519 key and records the cost; the drive path signs nothing
+        // and reports a measured zero over zero signings. Scope 8.4's
+        // ~1.4-of-8-cores figure is a P2 prediction and is never copied here.
+        signing_cores: if run.elapsed.as_secs_f64() > 0.0 {
+            (run.signings as f64 * run.signing_micros_p50) / (run.elapsed.as_secs_f64() * 1e6)
+        } else {
+            0.0
+        },
         index_bytes_per_object: index_cost.packed_bytes_per_object,
         index_run_bytes_per_object: index_cost.run_bytes_per_object,
         checkpoint_lookup_fanout: index_cost.lookup_fanout,
         // Nothing signs below engine.rs, so this is a measured zero over zero
         // signings rather than an unmeasured field. `evidence_signings` carries
         // the denominator so a reader can tell the two apart.
-        evidence_signing_micros_p50: 0.0,
-        evidence_signings: 0,
+        evidence_signing_micros_p50: run.signing_micros_p50,
+        evidence_signings: run.signings,
         fences: run.fences,
         transactions: run.transactions,
         free_bytes_available: available,
@@ -1894,6 +2629,12 @@ fn emit_skeleton(
     println!("bundle_gate=storage_primitive");
     println!("bundle_promotable=false");
     println!("bundle_skeleton=true");
+    println!("bundle_path_driven={}", run.path);
+    println!("objects_new={}", run.objects_new);
+    println!("objects_new_counted={}", run.objects_new_counted);
+    println!("evidence_signings={}", run.signings);
+    println!("evidence_signing_micros_p50={:.1}", run.signing_micros_p50);
+    println!("lock_release_attempts={}", run.lock_release_attempts);
     println!("bundle_outcome={}", outcome.name());
     println!("groups={}", run.groups);
     println!("transactions={}", run.transactions);
diff --git a/crates/levcs-store/src/bin/store-crash-driver.rs b/crates/levcs-store/src/bin/store-crash-driver.rs
index 9ceb913..5426529 100644
--- a/crates/levcs-store/src/bin/store-crash-driver.rs
+++ b/crates/levcs-store/src/bin/store-crash-driver.rs
@@ -4,7 +4,7 @@
 //!
 //! The parent arms a failpoint by name and a physical fault by name; this
 //! child runs a scripted, seed-deterministic workload through the journal-level
-//! `drive.rs` seam in Wave A and through `StoreEngine::submit` from Wave B; the
+//! `drive.rs` seam; the
 //! failpoint fires (`Fail`, `Panic`, or `HardExit` via `_exit(3)`, which runs
 //! no destructor and flushes no buffer); the parent reopens through production
 //! recovery and classifies.
@@ -232,9 +232,25 @@ fn submit_group(
                 .append_group_and_fence(&frames)
                 .map_err(|e| format!("append_group_and_fence: {e}"))
         }
+        // B1's engine landed, and the eight Wave B failpoint rows are driven
+        // through `StoreEngine::submit` — in `tests/crash_matrix.rs`, in
+        // process, because the four fields those rows add are statements about
+        // a *live* engine that a dead child cannot be asked about (scope 6.6
+        // deliverable 2).
+        //
+        // What is still missing here, and is a named carry-forward rather than
+        // a closed item: the `HardExit` and `SIGKILL` paths. Those need the
+        // engine *in the child*, so that `_exit(3)` and `kill -9` land inside a
+        // real publication rather than inside a journal append. Wiring it is
+        // more than a call swap — the child has to build a signer, a validated
+        // transaction, and a root, and `StoreEngine::open` still cannot create
+        // one — so it is reported instead of half-done. Until then
+        // `scripts/verify-store-recovery.sh`'s randomized cycles kill a writer
+        // at the journal seam, and that is what its numbers mean.
         DrivePath::Submit => Err(
-            "submit path requires B1 engine.rs; levcs-store also has no async \
-             executor dependency, which is an open interface request to the lead"
+            "the submit path is not wired into the child process yet: the Wave B rows are \
+             driven in-process by tests/crash_matrix.rs, and the SIGKILL soak still drives \
+             the journal seam. Reported as a carry-forward, not silently substituted."
                 .to_string(),
         ),
     }
diff --git a/crates/levcs-store/tests/crash_matrix.rs b/crates/levcs-store/tests/crash_matrix.rs
index cf48d6d..c7808fb 100644
--- a/crates/levcs-store/tests/crash_matrix.rs
+++ b/crates/levcs-store/tests/crash_matrix.rs
@@ -144,8 +144,9 @@ fn the_class_table_and_the_fixture_agree_on_every_row() {
 }
 
 #[test]
-fn wave_a_rows_carry_a_drive_plan_and_wave_b_rows_carry_a_reason() {
+fn wave_a_rows_carry_a_drive_plan_and_wave_b_rows_carry_a_submit_plan() {
     let fixture = harness::load_fixture();
+    let mut contention_rows = Vec::new();
     for row in &fixture.rows {
         let resolved = harness::resolve(row);
         match resolved.point.wave() {
@@ -156,30 +157,44 @@ fn wave_a_rows_carry_a_drive_plan_and_wave_b_rows_carry_a_reason() {
                 assert!(!plan.action.is_empty(), "row {} action", row.failpoint);
                 assert!(!plan.fault.is_empty(), "row {} fault", row.failpoint);
                 assert!(
-                    row.pending_reason.is_none(),
-                    "row {} is Wave A and must not be pending",
+                    row.submit.is_none(),
+                    "row {} is Wave A: it is driven through drive.rs, and claiming a \
+                     submit plan as well would mean two paths assert the same row \
+                     without either being the one under test",
                     row.failpoint
                 );
             }
             Wave::B => {
                 assert!(
                     row.drive.is_none(),
-                    "row {} is pending and must not claim a drive plan",
+                    "row {} needs an engine and must not claim a drive plan",
                     row.failpoint
                 );
-                let reason = row.pending_reason.as_ref().unwrap_or_else(|| {
+                let plan = row.submit.as_ref().unwrap_or_else(|| {
                     panic!(
-                        "pending row {} must carry its reason, exactly as Phase 0's \
-                         not-exercised adversarial rows do",
+                        "Wave B row {} must say which actions drive it through \
+                         StoreEngine::submit",
                         row.failpoint
                     )
                 });
                 assert!(
-                    reason.len() > 20,
-                    "row {}: 'pending' without a substantive reason is a hole with \
-                     a label on it",
+                    !plan.actions.is_empty(),
+                    "row {}: a Wave B row with no action is a pending row without the \
+                     word",
                     row.failpoint
                 );
+                for action in &plan.actions {
+                    assert!(
+                        action == "fail" || action == "panic",
+                        "row {}: the submit path can express Fail and Panic. HardExit \
+                         needs a parent process and Continue is not a fault; naming \
+                         either here would describe a run the harness cannot make: {action:?}",
+                        row.failpoint
+                    );
+                }
+                if plan.requires_root_cas_contention {
+                    contention_rows.push(row.failpoint.clone());
+                }
             }
         }
         assert!(
@@ -188,6 +203,47 @@ fn wave_a_rows_carry_a_drive_plan_and_wave_b_rows_carry_a_reason() {
             row.failpoint
         );
     }
+
+    assert_eq!(
+        contention_rows,
+        vec!["DuringRootCasRetry".to_string()],
+        "exactly one location is reachable only after a lost committed-root CAS. A \
+         second row acquiring the flag would mean a failpoint had been moved inside \
+         the retry loop without the matrix noticing; none acquiring it would mean \
+         DuringRootCasRetry had been driven by a path that never retried, which is \
+         the state B1 disclosed and this row exists to leave."
+    );
+}
+
+/// The three publication-side rows must be exercised with `Panic`, by name.
+///
+/// Ruling `WriterPanicAfterFence` into Wave A vacated the panic coverage of the
+/// publication half. A count here would not catch a row losing its `panic`
+/// action, and a check over "some row panics" would be satisfied by the wrong
+/// one.
+#[test]
+fn the_three_publication_side_rows_are_driven_with_the_panic_action() {
+    let fixture = harness::load_fixture();
+    let panicking: Vec<&str> = fixture
+        .rows
+        .iter()
+        .filter(|row| {
+            row.submit
+                .as_ref()
+                .is_some_and(|plan| plan.actions.iter().any(|action| action == "panic"))
+        })
+        .map(|row| row.failpoint.as_str())
+        .collect();
+    assert_eq!(
+        panicking,
+        vec![
+            "DuringCommittedRootBuild",
+            "BeforeRootCas",
+            "DuringRootCasRetry"
+        ],
+        "the fixture's own wave_b_exit_conditions names these three; the rows and the \
+         condition must not be able to drift apart"
+    );
 }
 
 #[test]
@@ -223,14 +279,41 @@ fn the_fixture_records_which_halves_wave_a_asserts_and_which_it_cannot() {
          to exercise with the Panic action, so the coverage vacated by ruling \
          WriterPanicAfterFence into Wave A is not lost"
     );
+    assert_eq!(
+        fixture.wave_b_asserted,
+        vec![
+            "physical_state_class".to_string(),
+            "recovery_outcome".to_string(),
+            "shard_poisoned".to_string(),
+            "immediate_status".to_string(),
+            "acknowledgment_allowed".to_string(),
+            "later_append_allowed_before_recovery".to_string(),
+        ],
+        "Wave B asserts the two halves Wave A could reach plus the four it could \
+         not; listing them by name is what makes one going missing a diff"
+    );
+    for field in &fixture.wave_a_unasserted {
+        assert!(
+            fixture.wave_b_asserted.contains(field),
+            "{field} was recorded as unassertable in Wave A, so Wave B owes it. A \
+             field that neither wave asserts is a hole with two labels on it."
+        );
+    }
 }
 
 /// Phase 1 exit requires zero pending rows.
 ///
-/// Inert while `engine.rs` is a skeleton, and it says so rather than passing
-/// silently — the same shape `check-phase1.sh` uses for its own inert checks.
-/// The probe is behavioral, not a grep: if `StoreEngine::open` stops returning
-/// `NotImplemented`, B1 has landed and every pending row is overdue.
+/// **Not inert any more.** It was, while `engine.rs` was a skeleton: the check
+/// asserted the pending set was *non*-empty, so the obligation could not be
+/// dropped by deleting rows instead of driving them. B1's engine landed and
+/// scope 6.6 deliverable 2 drove all eight, so the assertion is now
+/// unconditional in the other direction.
+///
+/// `check-phase1.sh`'s own version of this check is still inert — it fires only
+/// once `engine.rs` stops containing `NotImplemented`, and engine.rs still
+/// refuses startup states 1, 3, and 4 by name. So the gate would *not* catch a
+/// row that regressed to pending. This test would, and that is why it does not
+/// defer to the gate.
 #[test]
 fn the_pending_set_is_empty_at_phase_1_exit() {
     let fixture = harness::load_fixture();
@@ -240,35 +323,56 @@ fn the_pending_set_is_empty_at_phase_1_exit() {
         .filter(|row| row.wave == harness::PENDING_WAVE_B)
         .map(|row| row.failpoint.as_str())
         .collect();
-
-    if !engine_is_implemented() {
-        assert!(
-            !pending.is_empty(),
-            "engine.rs is still a skeleton, so the eight Wave B rows must still \
-             be pending; an empty pending set here would mean the obligation was \
-             dropped rather than met"
-        );
-        eprintln!(
-            "crash_matrix: {} rows pending Wave B (engine.rs is not implemented): {}",
-            pending.len(),
-            pending.join(", ")
-        );
-        return;
-    }
-
     assert!(
         pending.is_empty(),
-        "engine.rs is implemented, so no crash-matrix row may still be pending: {}",
+        "no crash-matrix row may be pending: {}",
         pending.join(", ")
     );
+
+    let driven: Vec<&str> = fixture
+        .rows
+        .iter()
+        .filter(|row| row.submit.is_some())
+        .map(|row| row.failpoint.as_str())
+        .collect();
+    let expected: Vec<&str> = Failpoint::ALL
+        .iter()
+        .filter(|point| point.wave() == Wave::B)
+        .map(|point| point.name())
+        .collect();
+    assert_eq!(
+        driven, expected,
+        "the eight Wave B rows are named by the registry, not by this file. \
+         Restating them as a list here is how a failpoint added to failpoints.rs \
+         becomes a fixture diff rather than a silent hole."
+    );
 }
 
-fn engine_is_implemented() -> bool {
+/// `StoreEngine::open` still refuses startup state 1, which is why the Wave B
+/// rows seed their root with `segment::initialize_root`.
+///
+/// Asserted rather than assumed, because the moment it stops being true the
+/// seeding disclosure in the fixture becomes a false statement about the
+/// harness, and a stale disclosure is worse than none: it tells a reader that a
+/// weakening still exists when the honest answer is that it does not.
+#[test]
+fn the_production_open_still_refuses_to_create_a_root_and_the_fixture_says_so() {
     let directory = tempfile::tempdir().expect("tempdir");
     let options = StoreOptions::new(directory.path().join("root"));
     match StoreEngine::open(options) {
-        Ok(_) => true,
-        Err(error) => !matches!(error, StoreError::NotImplemented(_)),
+        Err(StoreError::NotImplemented(reason)) => {
+            assert!(
+                reason.contains("startup states 1"),
+                "the refusal must still be the unbuilt startup states: {reason}"
+            );
+        }
+        Ok(_) => panic!(
+            "StoreEngine::open now creates an absent root, so the Wave B rows must be \
+             re-pointed at it and the fixture's root_seeded_by disclosure retired"
+        ),
+        Err(other) => {
+            panic!("StoreEngine::open refused an absent root for an unexpected reason: {other:?}")
+        }
     }
 }
 
@@ -1121,3 +1225,297 @@ fn the_drive_seam_a_wave_a_row_needs_is_present() {
          can be driven (scope 4-A1 deliverable 5)."
     );
 }
+
+// ===========================================================================
+// Driving the eight Wave B rows through StoreEngine::submit (scope 6.6 #2)
+// ===========================================================================
+//
+// Wave A asserted two halves of each row's `FailpointExpectation`. These rows
+// assert all six, because the four that were missing — `shard_poisoned`,
+// `immediate_status`, `acknowledgment_allowed`, and
+// `later_append_allowed_before_recovery` — are observable now that there is an
+// engine, and they are the only thing that distinguishes several rows from one
+// another. `AfterMarkedResolving` and `BeforeAppend` produce byte-identical
+// stores; the entire difference between them lives in those four fields.
+//
+// Every observation goes through a method a consumer calls. None goes through
+// an internal helper, a `cfg(test)` hook, or a field read. Scope 5 charter
+// item 8.
+
+#[cfg(feature = "store-privileged")]
+use support::engine_matrix;
+
+/// The one place the assertion is made, so every row is checked the same way
+/// and a row cannot quietly assert fewer fields than its neighbours.
+///
+/// Six separate equalities, never one aggregate comparison of two structs. A
+/// struct equality would report "expectation mismatch" and leave the reader to
+/// diff two `Debug` renderings; contract review 2026-07-24-A is on record that
+/// the field which is wrong is the one nobody looks at.
+#[cfg(feature = "store-privileged")]
+fn assert_full_expectation(
+    point: Failpoint,
+    class: PhysicalStateClass,
+    observation: &engine_matrix::RowObservation,
+) {
+    use levcs_protocol::oracle::{append_publication_expectation, AppendFailpoint};
+
+    let row = &observation.row;
+    let expectation = append_publication_expectation(AppendFailpoint::from(point));
+
+    // 1. physical_state_class -> recovery_outcome, and the frozen oracle's
+    //    recovery_outcome. The same two derivations Wave A compares, restated
+    //    here so a Wave B row is not exempt from the agreement Wave A enforces.
+    assert_eq!(
+        class.required_outcome(),
+        expectation.recovery_outcome,
+        "{row}: the physical state class {} implies {}, but the frozen oracle says {}",
+        class.name(),
+        harness::recovery_outcome_name(class.required_outcome()),
+        harness::recovery_outcome_name(expectation.recovery_outcome)
+    );
+
+    // 2. recovery_outcome, observed by closing the engine and reopening
+    //    through production recovery.
+    let fact = engine_matrix::recovered_fact(&observation.recovered, row);
+    assert!(
+        outcome_admits(expectation.recovery_outcome, fact),
+        "{row}: after a close and reopen through StoreEngine::open the victim reads \
+         back as {fact:?}, which the frozen oracle's {} does not admit. victim = {:?}, \
+         probe = {:?}, fences over the victim's group = {}",
+        harness::recovery_outcome_name(expectation.recovery_outcome),
+        observation.victim,
+        observation.probe,
+        observation.fences_during_victim()
+    );
+
+    // 3. shard_poisoned — the store naming itself poisoned, by variant.
+    assert_eq!(
+        observation.shard_poisoned_observed(),
+        expectation.shard_poisoned,
+        "{row}: the frozen oracle says shard_poisoned = {}. victim = {:?}, probe = {:?}",
+        expectation.shard_poisoned,
+        observation.victim,
+        observation.probe
+    );
+
+    // 4. immediate_status — plan §5.1's two-root read on the live engine.
+    let immediate = engine_matrix::immediate_status_of(&observation.immediate, row);
+    assert_eq!(
+        engine_matrix::immediate_status_name(immediate),
+        engine_matrix::immediate_status_name(expectation.immediate_status),
+        "{row}: transaction_status disagrees with the frozen oracle"
+    );
+
+    // 5. acknowledgment_allowed — a receipt obtainable through a consumer path.
+    assert_eq!(
+        observation.acknowledgment_allowed_observed(),
+        expectation.acknowledgment_allowed,
+        "{row}: the frozen oracle says acknowledgment_allowed = {}. victim = {:?}",
+        expectation.acknowledgment_allowed,
+        observation.victim
+    );
+    if !expectation.acknowledgment_allowed {
+        assert!(
+            observation.victim.receipt().is_none(),
+            "{row}: submit returned a receipt for an operation the oracle says may not \
+             be acknowledged, which is an acknowledgment whatever else is true"
+        );
+    }
+
+    // 6. later_append_allowed_before_recovery — a *different* transaction on
+    //    the same shard, before any reopen.
+    assert_eq!(
+        observation.later_append_allowed_observed(),
+        expectation.later_append_allowed_before_recovery,
+        "{row}: the frozen oracle says later_append_allowed_before_recovery = {}. \
+         probe = {:?}",
+        expectation.later_append_allowed_before_recovery,
+        observation.probe
+    );
+
+    // The transaction that committed before the fault is not collateral. Every
+    // row proves that too, exactly as the Wave A rows prove it for a prior
+    // group; a fault that costs an already-acknowledged transaction its receipt
+    // is acknowledged loss, whatever the row under test says.
+    assert!(
+        matches!(
+            observation.prior_recovered,
+            levcs_store::types::TransactionStatus::Committed(_)
+        ),
+        "{row}: the transaction acknowledged before the fault did not survive the \
+         reopen: {:?}",
+        observation.prior_recovered
+    );
+
+    // The root-lock release window documented on
+    // `engine_matrix::reopen_after_close`. Reported rather than asserted: the
+    // wait is compensating for a defect in a file B4 does not own, and a
+    // harness that stayed silent about it would be hiding the very thing it is
+    // working around.
+    if observation.lock_release_attempts > 1 {
+        eprintln!(
+            "crash_matrix: {row}: the root lock was still held after StoreEngine::drop \
+             returned; the reopen needed {} attempts over {:?}",
+            observation.lock_release_attempts, observation.lock_release_wait
+        );
+    }
+
+    // Charter item 7: the class is checked against a syscall count, not a
+    // comment. A class that says no bytes were written must show no fence, and
+    // a class that says the group was fenced must show exactly one.
+    let fences = observation.fences_during_victim();
+    let expected_fences = match class {
+        PhysicalStateClass::NoBytes => 0,
+        PhysicalStateClass::WholeFrameFenced => 1,
+        PhysicalStateClass::WholeFrameFencedAndPublished => 1,
+        // No Wave B row produces these two: the failpoints that tear a frame or
+        // stop short of the fence are all Wave A and are driven through
+        // drive.rs. Naming them is what makes a future row that moves into one
+        // of these classes fail here rather than silently skip the check.
+        PhysicalStateClass::PartialFrame => panic!(
+            "{row}: a Wave B row produced a partial frame; the submit path forms whole \
+             frames and the tearing failpoints are Wave A"
+        ),
+        PhysicalStateClass::WholeFrameUnfenced => panic!(
+            "{row}: a Wave B row produced an unfenced whole frame; every Wave B location \
+             is either before the mark or after the fence"
+        ),
+    };
+    assert_eq!(
+        fences,
+        expected_fences,
+        "{row}: physical state class {} implies {expected_fences} durability fence(s) \
+         for this group, and DurabilityCounters reports {fences}",
+        class.name()
+    );
+}
+
+#[cfg(feature = "store-privileged")]
+#[test]
+fn wave_b_rows_drive_through_submit_to_their_full_failpoint_expectation() {
+    let fixture = harness::load_fixture();
+    // One token for the whole test. The failpoint and fault registries are
+    // process-global one-shots, and a per-row token would let another test in
+    // this binary arm between two rows of this one.
+    let serial = engine_matrix::serial();
+    let mut driven: Vec<String> = Vec::new();
+
+    for row in &fixture.rows {
+        let resolved = harness::resolve(row);
+        if resolved.point.wave() != Wave::B {
+            continue;
+        }
+        let plan = row
+            .submit
+            .as_ref()
+            .expect("a Wave B row carries a submit plan");
+        if plan.requires_root_cas_contention {
+            // Driven by its own test below, which has to manufacture the lost
+            // compare-and-swap this location sits behind.
+            driven.push(row.failpoint.clone());
+            continue;
+        }
+
+        for action in &plan.actions {
+            let action = engine_matrix::action_from_name(action)
+                .unwrap_or_else(|| panic!("row {}: unknown action {action:?}", row.failpoint));
+            let observation = engine_matrix::drive_submit_row(&serial, resolved.point, action);
+            assert_full_expectation(resolved.point, resolved.class, &observation);
+        }
+        driven.push(row.failpoint.clone());
+    }
+
+    let expected: Vec<String> = Failpoint::ALL
+        .iter()
+        .filter(|point| point.wave() == Wave::B)
+        .map(|point| point.name().to_string())
+        .collect();
+    assert_eq!(
+        driven, expected,
+        "every Wave B row must actually have been driven, by name"
+    );
+}
+
+/// The row B1 disclosed was never armed.
+///
+/// `DuringRootCasRetry` sits inside `publish_subtree`'s retry loop, past a
+/// failed `compare_and_swap`. Submitting a transaction does not reach it; the
+/// only way there is a genuinely lost CAS, which means another shard publishing
+/// into the same committed root between this shard's load and its swap. The
+/// driver manufactures the load, not the mechanism: eight shard writer threads
+/// publish through the production path and the harness only chooses how much
+/// work each publication carries into the window.
+///
+/// # Why this is a hard failure and not a skip
+///
+/// A fault-injection row that quietly does nothing when it cannot reach its
+/// location is the state this row was already in — reachable in principle,
+/// unarmed in practice, and green either way. If the location stops being
+/// reachable, this test says so by name and by number.
+#[cfg(feature = "store-privileged")]
+#[test]
+fn during_root_cas_retry_drives_through_a_genuinely_lost_committed_root_cas() {
+    use std::time::Duration;
+
+    let fixture = harness::load_fixture();
+    let row = fixture
+        .rows
+        .iter()
+        .find(|row| row.failpoint == "DuringRootCasRetry")
+        .expect("the fixture carries the row");
+    let resolved = harness::resolve(row);
+    let plan = row.submit.as_ref().expect("a submit plan");
+    assert!(
+        plan.requires_root_cas_contention,
+        "this test manufactures contention; the fixture must say the row needs it"
+    );
+
+    let serial = engine_matrix::serial();
+    for action in &plan.actions {
+        let action = engine_matrix::action_from_name(action)
+            .unwrap_or_else(|| panic!("unknown action {action:?}"));
+        let run = engine_matrix::drive_root_cas_retry_row(
+            &serial,
+            action,
+            Duration::from_secs(CONTENTION_BUDGET_SECONDS),
+        );
+        assert!(
+            run.anomaly.is_none(),
+            "DuringRootCasRetry [{}]: a shard was refused for a reason this location \
+             does not produce, so no transaction in this run is the row's victim: {:?}",
+            engine_matrix::action_name(action),
+            run.anomaly
+        );
+        let observation = run.observation.unwrap_or_else(|| {
+            panic!(
+                "DuringRootCasRetry [{}] did not fire in {:?} across {} submitted \
+                 transactions on {} shards. The location is inside publish_subtree's \
+                 retry loop; not reaching it means either that no committed-root CAS \
+                 was ever lost — in which case the loop is unreachable and the row \
+                 cannot be asserted from outside the engine — or that the failpoint \
+                 has moved out of the retry. Either is a finding, and neither is a \
+                 reason to rerun.",
+                engine_matrix::action_name(action),
+                run.elapsed,
+                run.attempts,
+                engine_matrix::CONTENTION_SHARDS
+            )
+        });
+        eprintln!(
+            "crash_matrix: DuringRootCasRetry [{}] fired after {} transactions in {:?}",
+            engine_matrix::action_name(action),
+            run.attempts,
+            run.elapsed
+        );
+        assert_full_expectation(resolved.point, resolved.class, &observation);
+    }
+}
+
+/// How long the contention driver is given per action.
+///
+/// Chosen from measurement rather than from taste: see the campaign recorded in
+/// the B4 report. It is a ceiling on a search, not a timeout tuned until the
+/// test passed.
+#[cfg(feature = "store-privileged")]
+const CONTENTION_BUDGET_SECONDS: u64 = 120;
diff --git a/crates/levcs-store/tests/fixtures/phase1-failpoints.json b/crates/levcs-store/tests/fixtures/phase1-failpoints.json
index a389db9..f675234 100644
--- a/crates/levcs-store/tests/fixtures/phase1-failpoints.json
+++ b/crates/levcs-store/tests/fixtures/phase1-failpoints.json
@@ -5,7 +5,8 @@
   "notes": [
     "One row per levcs_store::failpoints::Failpoint. The row set, its order, and each row's wave are asserted against Failpoint::ALL and Failpoint::wave(); the wave partition is pinned by name in failpoints.rs and is never restated as a count here.",
     "Each row carries two independent derivations of the same answer: physical_state_class -> required_outcome through the class table in tests/support/group_model.rs, and levcs_protocol::oracle::append_publication_expectation(point).recovery_outcome through the frozen Phase 0 oracle. crash_matrix.rs asserts the observed outcome equals both. Two derivations that must agree is the point; a single one would only restate itself.",
-    "No catch-all match arm exists in crash_matrix.rs or in the class table. Contract review 2026-07-24-A shipped an unsound recovery classification because a catch-all left most outcomes unasserted."
+    "No catch-all match arm exists in crash_matrix.rs or in the class table. Contract review 2026-07-24-A shipped an unsound recovery classification because a catch-all left most outcomes unasserted.",
+    "Wave B rows are driven in-process through StoreEngine::submit rather than through a child process. HardExit is the only action that requires a parent, and no Wave B row uses it; the four fields Wave B adds are statements about a live engine that a dead child cannot be asked about. recovery_outcome is still observed by closing the engine and reopening through StoreEngine::open, so both waves answer that half the same way."
   ],
   "wave_a_assertion_scope": {
     "asserted": [
@@ -21,8 +22,8 @@
     "reason": "drive.rs has no engine, status root, sequencer, or signer, so there is no status to observe, no acknowledgment to permit or refuse, and no later-append admission control. These four fields are equally unassertable for every Wave A row, so they do not distinguish any row from another. Wave B re-asserts every row's complete FailpointExpectation through StoreEngine::submit."
   },
   "wave_b_exit_conditions": [
-    "The pending set is empty: no row carries wave = pending-wave-b.",
-    "Every row asserts its complete FailpointExpectation through submit, not only the two halves above.",
+    "The pending set is empty: no row carries the wave value that names an undriven obligation. Met. The literal is deliberately absent from this file, because scripts/check-phase1.sh greps for it and a check that matches its own documentation is a check that fails on a green tree.",
+    "Every row asserts its complete FailpointExpectation through submit, not only the two halves Wave A could reach. Met for the eight Wave B rows; the nine Wave A rows keep the two halves drive.rs can observe, since drive.rs has no status root to observe the others with and driving them twice would not add a fact.",
     "The Panic action is exercised on DuringCommittedRootBuild, BeforeRootCas, and DuringRootCasRetry, so the panic-in-publication coverage vacated by ruling WriterPanicAfterFence into Wave A is not lost. The failpoint enum names a location; the driver chooses the action; the two axes are independent."
   ],
   "rows": [
@@ -40,13 +41,18 @@
     },
     {
       "failpoint": "AfterMarkedResolving",
-      "wave": "pending-wave-b",
+      "wave": "B",
       "physical_state_class": "NoBytes",
       "required_outcome": "AbsentRetriable",
       "victim_placement": "first",
       "drive": null,
-      "pending_reason": "Requires the status root. Marking an operation Resolving is step 3 of the append ordering and lives in engine.rs; drive.rs has no status root, so the location does not exist to fire.",
-      "rationale": "Marking Resolving writes no journal byte, so the physical state is identical to BeforeAppend. The row is distinguished only by immediate_status and shard_poisoned, which is exactly why it is Wave B."
+      "submit": {
+        "actions": [
+          "fail"
+        ],
+        "requires_root_cas_contention": false
+      },
+      "rationale": "Marking Resolving writes no journal byte, so the physical state is identical to BeforeAppend. The row is distinguished only by immediate_status and shard_poisoned, which is why it needed an engine: the status root is the only place the difference exists."
     },
     {
       "failpoint": "DuringFrameWriteTorn",
@@ -74,13 +80,18 @@
     },
     {
       "failpoint": "EvidenceHandoffFailure",
-      "wave": "pending-wave-b",
+      "wave": "B",
       "physical_state_class": "NoBytes",
       "required_outcome": "AbsentRetriable",
       "victim_placement": "first",
       "drive": null,
-      "pending_reason": "Requires the sequencer's CommitEvidenceSigner handoff, which happens before append and lives in engine.rs. drive.rs has no signer, so the location does not exist to fire.",
-      "rationale": "Signer failure occurs before journal append and therefore writes nothing; it cannot produce an unsigned committed frame."
+      "submit": {
+        "actions": [
+          "fail"
+        ],
+        "requires_root_cas_contention": false
+      },
+      "rationale": "Fires at scope 6.3 step 2, before the group is marked Resolving and before any byte is written. Contract review 2026-07-26-A gave it the BeforeAppend shape: a routine SignerError::Unavailable is an availability event, and poisoning the shard for it would trade a real availability property for a safety property that was never at risk."
     },
     {
       "failpoint": "BeforeFence",
@@ -132,43 +143,66 @@
     },
     {
       "failpoint": "DuringCommittedRootBuild",
-      "wave": "pending-wave-b",
+      "wave": "B",
       "physical_state_class": "WholeFrameFenced",
       "required_outcome": "Committed",
       "victim_placement": "group-wide",
       "drive": null,
-      "pending_reason": "Requires the immutable shard subtree build of engine.rs. Wave B must exercise this row with the Panic action as well as Fail.",
-      "rationale": "The fence already succeeded, so the frames are durable regardless of what the in-memory build does. The shard poisons and recovery publishes the prefix exactly once."
+      "submit": {
+        "actions": [
+          "fail",
+          "panic"
+        ],
+        "requires_root_cas_contention": false
+      },
+      "rationale": "The fence returned before the subtree build begins, so the frame is durable and the transaction is committed on the device while the publication that would make it visible never happens. Recovery replays the fenced frame and publishes the receipt."
     },
     {
       "failpoint": "AllocationFailureBeforePublication",
-      "wave": "pending-wave-b",
+      "wave": "B",
       "physical_state_class": "WholeFrameFenced",
       "required_outcome": "Committed",
       "victim_placement": "group-wide",
       "drive": null,
-      "pending_reason": "Requires the publication path of engine.rs.",
-      "rationale": "Same durable state as AfterSuccessfulFence; the differential is entirely in the poison and status halves."
+      "submit": {
+        "actions": [
+          "fail"
+        ],
+        "requires_root_cas_contention": false
+      },
+      "rationale": "Same physical state as DuringCommittedRootBuild and reached one step later: the group is fenced, the subtree is built, and the allocation that would carry it into the root fails. Poisoning, because it is inside steps 4-8."
     },
     {
       "failpoint": "BeforeRootCas",
-      "wave": "pending-wave-b",
+      "wave": "B",
       "physical_state_class": "WholeFrameFenced",
       "required_outcome": "Committed",
       "victim_placement": "group-wide",
       "drive": null,
-      "pending_reason": "Requires the ArcSwap committed-root CAS of engine.rs. Wave B must exercise this row with the Panic action as well as Fail.",
-      "rationale": "The committed-root swap is the visibility boundary, and it has not happened; the journal bytes are nonetheless durable."
+      "submit": {
+        "actions": [
+          "fail",
+          "panic"
+        ],
+        "requires_root_cas_contention": false
+      },
+      "rationale": "The last instant at which a fenced group is still unpublished. Everything before the compare-and-swap has succeeded, so the transaction is durable; the shard is poisoned because a group that fenced and did not publish leaves the status root claiming Resolving for a committed transaction."
     },
     {
       "failpoint": "DuringRootCasRetry",
-      "wave": "pending-wave-b",
+      "wave": "B",
       "physical_state_class": "WholeFrameFenced",
       "required_outcome": "Committed",
       "victim_placement": "group-wide",
       "drive": null,
-      "pending_reason": "Requires the CAS merge loop under concurrent shard publication, which only exists in engine.rs. Wave B must exercise this row with the Panic action as well as Fail.",
-      "rationale": "A contended CAS re-merges against the newer root; dying inside the retry changes nothing about the durable prefix."
+      "submit": {
+        "actions": [
+          "fail",
+          "panic"
+        ],
+        "requires_root_cas_contention": true
+      },
+      "rationale": "Reached only after a lost compare-and-swap: another shard published between this shard's load of the committed root and its swap. The re-merge is against a newer root and the subtree is unchanged, so the outcome is identical to BeforeRootCas - which is the point, because a retry that resolved differently from a first attempt would make publication order observable."
     },
     {
       "failpoint": "WriterPanicBeforeFence",
@@ -196,23 +230,46 @@
     },
     {
       "failpoint": "AfterRootCasBeforeWaiterWake",
-      "wave": "pending-wave-b",
+      "wave": "B",
       "physical_state_class": "WholeFrameFencedAndPublished",
       "required_outcome": "Committed",
       "victim_placement": "group-wide",
       "drive": null,
-      "pending_reason": "Requires a published committed root and a receipt to be retrievable, both of which are engine.rs.",
-      "rationale": "The receipt is already visible, so acknowledgment is permitted and failing to wake a waiter cannot hide it. This is one of only two rows whose class is fenced-and-published."
+      "submit": {
+        "actions": [
+          "fail"
+        ],
+        "requires_root_cas_contention": false
+      },
+      "rationale": "Post-publication. The fence succeeded and the root published, so the transaction is committed; a waiter that never wakes is a hung request, not an absent transaction, and the receipt stays queryable through transaction_status."
     },
     {
       "failpoint": "BeforeResponse",
-      "wave": "pending-wave-b",
+      "wave": "B",
       "physical_state_class": "WholeFrameFencedAndPublished",
       "required_outcome": "Committed",
       "victim_placement": "group-wide",
       "drive": null,
-      "pending_reason": "Requires a published receipt and a response path, both of which are engine.rs.",
+      "submit": {
+        "actions": [
+          "fail"
+        ],
+        "requires_root_cas_contention": false
+      },
       "rationale": "Post-publication and pre-response: the receipt is already durable and idempotently retrievable by status or retry, and no reappend is required."
     }
-  ]
+  ],
+  "wave_b_assertion_scope": {
+    "asserted": [
+      "physical_state_class",
+      "recovery_outcome",
+      "shard_poisoned",
+      "immediate_status",
+      "acknowledgment_allowed",
+      "later_append_allowed_before_recovery"
+    ],
+    "reason": "With engine.rs the four fields Wave A could not reach are observable through the API a consumer calls: submit returns or refuses, transaction_status performs plan 5.1's two-root read, a second submit to the same shard says whether a later append is admitted, and a close-and-reopen through StoreEngine::open answers recovery_outcome the same way the Wave A rows answer it. Every Wave B row asserts the complete FailpointExpectation, field by field, against oracle::append_publication_expectation.",
+    "root_seeded_by": "segment::initialize_root. StoreEngine::open still refuses startup state 1 - initializing an absent or empty root is B1 deliverable 1 and is unimplemented - so the root these rows exercise cannot be created through the production open. This is a charter item 8 weakening and it is recorded here rather than only in a report: what is asserted below is the production submit, status, and reopen path over a root production did not build. When B1 lands state 1 this becomes StoreEngine::open on an absent path and the weakening disappears.",
+    "panic_ownership": "B1 implements phase-aware panic ownership: a panic in scope 6.3 steps 1-3 is definitively absent and leaves the writer thread alive; a panic in steps 4-8 poisons and kills the writer, so a later append is refused NotReady rather than accepted; a panic in steps 9-10 still delivers every receipt. The Panic rows confirm this from outside the engine, through submit and transaction_status, rather than by reading the catch sites."
+  }
 }
diff --git a/crates/levcs-store/tests/support/engine_matrix.rs b/crates/levcs-store/tests/support/engine_matrix.rs
new file mode 100644
index 0000000..bba05e6
--- /dev/null
+++ b/crates/levcs-store/tests/support/engine_matrix.rs
@@ -0,0 +1,979 @@
+//! Driving a crash-matrix row through the production `StoreEngine::submit`.
+//!
+//! **Owned by B4 StoreHarnessB** (scope 2.1, 6.6 deliverable 2).
+//!
+//! Wave A could assert only the physical-state-class and `recovery_outcome`
+//! halves of a [`FailpointExpectation`], because `drive.rs` has no status
+//! root, no sequencer, and no acknowledgment path. This module supplies the
+//! other four. Every observation here is made through a method a *consumer*
+//! calls — `StoreEngine::submit`, `StoreEngine::transaction_status`,
+//! `StoreEngine::open`, `StoreEngine::durability_counters` — and never through
+//! an internal helper, because scope 5 charter item 8 is precisely about a
+//! safety property that is asserted on a function production does not call.
+//!
+//! # Why this is in-process and the Wave A rows are not
+//!
+//! A Wave A row's action set includes `HardExit`, which is `_exit(3)`: it can
+//! only be observed from a parent process. The eight Wave B rows are driven
+//! with `Fail` and `Panic`, both of which leave the process alive, and the
+//! four fields they add — `shard_poisoned`, `immediate_status`,
+//! `acknowledgment_allowed`, `later_append_allowed_before_recovery` — are
+//! statements about a *live* engine that no post-mortem reopen can make. A
+//! child process that has already died cannot be asked whether its shard would
+//! have accepted another append.
+//!
+//! `recovery_outcome` is still observed the Wave A way — close the engine and
+//! reopen through production recovery in a fresh `StoreEngine::open` — so the
+//! two waves answer that half by the same means.
+//!
+//! # No catch-all arm
+//!
+//! Same rule as `crash_matrix.rs`. Every `TransactionStatus` variant, every
+//! `FailpointAction`, and every `ImmediateStatus` is named. A fallback binding
+//! here would silently classify a status nobody expected as the one the oracle
+//! happened to want.
+
+#![allow(dead_code)]
+
+use std::future::Future;
+use std::path::Path;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
+use std::task::{Context, Poll, Wake, Waker};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use levcs_core::{ObjectId, ObjectType};
+use levcs_protocol::oracle::{ImmediateStatus, RecoveredTailFact};
+use levcs_protocol::v2::TransactionEvidenceV1;
+use levcs_store::failpoints::{Failpoint, FailpointAction};
+use levcs_store::options::StoreOptions;
+use levcs_store::segment::{initialize_root, RootLayout};
+use levcs_store::transaction::StagedObject;
+use levcs_store::types::{
+    CommitEvidenceSigner, CommitReceipt, DurabilityCounters, NamespaceId, OperationId, SignerError,
+    StoreError, TransactionStatus,
+};
+use levcs_store::{StoreEngine, ValidatedTransaction};
+
+// ---------------------------------------------------------------------------
+// Seeding a root
+// ---------------------------------------------------------------------------
+
+/// **Charter item 8 disclosure, stated where it is committed rather than in a
+/// report only.**
+///
+/// `StoreEngine::open` refuses startup state 1 — initializing an absent or
+/// empty root — by name (`engine.rs`, B1 deliverable 1, unimplemented). So a
+/// store that is about to be exercised through the production `submit` cannot
+/// be *created* through the production `open`. It is created here by calling
+/// `segment::initialize_root`, which is the same function state 1 will call
+/// when B1 implements it, but reached directly rather than through the
+/// entry point a consumer uses.
+///
+/// This is a real weakening and it is named rather than buried: everything
+/// below asserts against the production path *after* the root exists, and
+/// nothing below asserts anything about how a root comes into existence. When
+/// B1 lands startup state 1, this function becomes one line — `StoreEngine::open`
+/// on an absent path — and the Wave B rows gain that half for free.
+pub const ROOT_SEEDED_BY_NON_PRODUCTION_PATH: &str =
+    "segment::initialize_root under the store-internals-adjacent public module, because \
+     StoreEngine::open still refuses startup state 1 (B1 deliverable 1)";
+
+fn now_micros() -> i64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_micros() as i64)
+        .unwrap_or(0)
+}
+
+pub fn seed_root(root: &Path, shard_count: u16) {
+    initialize_root(
+        &RootLayout::new(root),
+        shard_count,
+        [0x4b; 16],
+        now_micros(),
+        &DurabilityCounters::default(),
+    )
+    .expect("initialize the store root");
+}
+
+// ---------------------------------------------------------------------------
+// The injected signer
+// ---------------------------------------------------------------------------
+
+/// A deterministic stand-in for instance composition's signer.
+///
+/// The store never verifies a signature (plan §5.1 forbids it from deciding
+/// who may sign), so this does not need to be Ed25519. It needs to be a pure
+/// function of the digest it is handed, and it needs to count its calls, so a
+/// row that claims nothing was signed can be checked rather than believed.
+#[derive(Default)]
+pub struct CountingSigner {
+    pub calls: AtomicU64,
+}
+
+impl CommitEvidenceSigner for CountingSigner {
+    fn key_epoch(&self) -> u64 {
+        11
+    }
+
+    fn public_key(&self) -> [u8; 32] {
+        [0x5a; 32]
+    }
+
+    fn sign_event(&self, signing_digest: &ObjectId) -> Result<[u8; 64], SignerError> {
+        self.calls.fetch_add(1, Ordering::SeqCst);
+        let mut signature = [0u8; 64];
+        let first = blake3::hash(signing_digest.as_bytes());
+        let second = blake3::hash(first.as_bytes());
+        signature[..32].copy_from_slice(first.as_bytes());
+        signature[32..].copy_from_slice(second.as_bytes());
+        Ok(signature)
+    }
+}
+
+/// `StoreOptions::max_index_runs`, the ceiling this slice reaches soonest.
+///
+/// Publishing a group adds one in-memory index delta layer, and B1's slice
+/// seals none of them into an `IndexRun` — deliverable 1 refuses by name. So
+/// `submit` refuses `NotImplemented` after exactly `max_index_runs` group
+/// publications for the life of an engine, root-wide, whatever the workload.
+/// At the shipping default of 64 that is 64 commits.
+///
+/// The single-row driver leaves the default alone: three submits per row is
+/// nowhere near it, and a row driven under a non-default ceiling would be a row
+/// driven against a store nobody ships. The contention driver raises it,
+/// because it must publish groups until two of them collide and the collision
+/// is not on a schedule. **That is a disclosed weakening**: the raised value is
+/// a real configuration, not a bypass, but it configures around a missing
+/// deliverable rather than around a tuning choice, and it is the same
+/// unimplemented seal that caps the benchmark.
+pub const DEFAULT_MAX_INDEX_RUNS: u32 = 64;
+
+/// Enough that the collision search is never the thing that ends the run.
+/// Measured: the failpoint fires within 6 to 64 group publications, and the
+/// eight repository creations consume eight of them.
+pub const CONTENTION_MAX_INDEX_RUNS: u32 = 4096;
+
+pub fn options(root: &Path, shard_count: u16) -> StoreOptions {
+    options_with_index_runs(root, shard_count, DEFAULT_MAX_INDEX_RUNS)
+}
+
+pub fn options_with_index_runs(root: &Path, shard_count: u16, max_index_runs: u32) -> StoreOptions {
+    let mut options = StoreOptions::new(root);
+    options.shard_count = shard_count;
+    options.max_index_runs = max_index_runs;
+    options.max_open_index_runs = options.max_open_index_runs.min(max_index_runs);
+    // A group of one, formed promptly. Every Wave B row is about where in the
+    // publication of *this* group the fault lands, so a group that waits for
+    // company would only add scheduling noise to the observation.
+    options.max_group_transactions = 1;
+    options.max_group_bytes = 64 * 1024;
+    options.max_group_idle = Duration::from_millis(5);
+    options.journal_preallocate_bytes = 4 * 1024 * 1024;
+    options.signer = Some(Arc::new(CountingSigner::default()));
+    options
+}
+
+// ---------------------------------------------------------------------------
+// Transactions
+// ---------------------------------------------------------------------------
+
+/// The client principal, deliberately not the signer's public key.
+const EVIDENCE_ACTOR: [u8; 32] = [0x7e; 32];
+
+fn evidence() -> TransactionEvidenceV1 {
+    TransactionEvidenceV1::AdministrativeV1 {
+        actor: EVIDENCE_ACTOR,
+        actor_key_epoch: 11,
+        command_digest: ObjectId([0x33; 32]),
+        signature: [0x44; 64],
+    }
+}
+
+fn deadline() -> i64 {
+    now_micros() + 600_000_000
+}
+
+fn genesis_id(namespace: &NamespaceId) -> ObjectId {
+    let mut bytes = [0u8; 32];
+    let mut hasher = blake3::Hasher::new();
+    hasher.update(b"levcs-store/b4/genesis-authority/v1\0");
+    hasher.update(namespace.as_bytes());
+    hasher.finalize_xof().fill(&mut bytes);
+    ObjectId(bytes)
+}
+
+/// A namespace that routes to `shard`, found by search rather than by
+/// arithmetic over a routing rule this file would then own a second copy of.
+pub fn namespace_on_shard(shard: u16, shard_count: u16, salt: u64) -> NamespaceId {
+    for attempt in 0..4096u64 {
+        let mut bytes = [0u8; 32];
+        let mut hasher = blake3::Hasher::new();
+        hasher.update(b"levcs-store/b4/namespace/v1\0");
+        hasher.update(&salt.to_le_bytes());
+        hasher.update(&attempt.to_le_bytes());
+        hasher.finalize_xof().fill(&mut bytes);
+        let namespace = NamespaceId(bytes);
+        if StoreOptions::shard_of(&namespace, shard_count) == shard {
+            return namespace;
+        }
+    }
+    panic!("no namespace routed to shard {shard} of {shard_count} in 4096 attempts");
+}
+
+pub fn create_transaction(namespace: NamespaceId, operation: u8) -> ValidatedTransaction {
+    let id = genesis_id(&namespace);
+    ValidatedTransaction::builder(privileged())
+        .namespace(namespace)
+        .operation(
+            OperationId([operation; 16]),
+            ObjectId([operation; 32]),
+            deadline(),
+        )
+        .create_repository(id)
+        .objects(vec![StagedObject {
+            id,
+            object_type: ObjectType::Authority,
+            raw: vec![0xa1; 32],
+        }])
+        .refs(Vec::new())
+        .authority(None, Some(id))
+        .evidence(evidence())
+        .build()
+        .expect("a complete create transaction")
+}
+
+pub fn push_transaction(namespace: NamespaceId, operation: u8, blob: u8) -> ValidatedTransaction {
+    let authority = genesis_id(&namespace);
+    ValidatedTransaction::builder(privileged())
+        .namespace(namespace)
+        .operation(
+            OperationId([operation; 16]),
+            ObjectId([operation; 32]),
+            deadline(),
+        )
+        .objects(vec![StagedObject {
+            id: ObjectId([blob; 32]),
+            object_type: ObjectType::Blob,
+            raw: vec![blob; 64],
+        }])
+        .refs(Vec::new())
+        .authority(Some(authority), Some(authority))
+        .evidence(evidence())
+        .build()
+        .expect("a complete push transaction")
+}
+
+fn privileged() -> levcs_store::types::PrivilegedConstruction {
+    levcs_store::types::PrivilegedConstruction::assert_validated()
+}
+
+// ---------------------------------------------------------------------------
+// A local executor
+// ---------------------------------------------------------------------------
+
+/// Drive one future on this thread until it completes or `deadline` passes.
+///
+/// `None` is a real answer, not a failure: scope 6.3 says a failure to wake a
+/// waiter leaves a *hung request*, never an absent transaction, and a harness
+/// that cannot represent a hang cannot assert the difference.
+/// `levcs-store` starts no runtime, so neither does this.
+pub fn block_on_until<F: Future>(future: F, deadline: Duration) -> Option<F::Output> {
+    struct ThreadWaker(std::thread::Thread);
+    impl Wake for ThreadWaker {
+        fn wake(self: Arc<Self>) {
+            self.0.unpark();
+        }
+        fn wake_by_ref(self: &Arc<Self>) {
+            self.0.unpark();
+        }
+    }
+
+    let waker = Waker::from(Arc::new(ThreadWaker(std::thread::current())));
+    let mut context = Context::from_waker(&waker);
+    let mut future = std::pin::pin!(future);
+    let started = Instant::now();
+    loop {
+        match future.as_mut().poll(&mut context) {
+            Poll::Ready(output) => return Some(output),
+            Poll::Pending => {
+                let elapsed = started.elapsed();
+                if elapsed >= deadline {
+                    return None;
+                }
+                std::thread::park_timeout(deadline - elapsed);
+            }
+        }
+    }
+}
+
+/// How long a submit is given before it is called hung.
+///
+/// Generous, because the alternative failure mode is a flaky matrix: a
+/// deadline tuned close to the observed latency turns an unrelated scheduling
+/// hiccup into a "hung request" finding. A committing submit in this harness
+/// takes single-digit milliseconds, so the margin is three orders of
+/// magnitude. Only `AfterRootCasBeforeWaiterWake` with `Fail` is expected to
+/// reach it — that row's whole claim is that the waiter never wakes.
+pub const SUBMIT_DEADLINE: Duration = Duration::from_secs(5);
+
+// ---------------------------------------------------------------------------
+// One submit, with its outcome fully enumerated
+// ---------------------------------------------------------------------------
+
+/// What one call to `StoreEngine::submit` did. Every case is named; there is
+/// no residual "something else" arm.
+#[derive(Debug)]
+pub enum SubmitOutcome {
+    /// The engine returned a receipt.
+    Committed(CommitReceipt),
+    /// The engine returned a typed refusal.
+    Refused(StoreError),
+    /// The call panicked in the caller's task and the panic was caught.
+    CallerPanicked,
+    /// The future never completed within `SUBMIT_DEADLINE`.
+    Hung,
+}
+
+impl SubmitOutcome {
+    pub fn receipt(&self) -> Option<&CommitReceipt> {
+        match self {
+            SubmitOutcome::Committed(receipt) => Some(receipt),
+            SubmitOutcome::Refused(_) => None,
+            SubmitOutcome::CallerPanicked => None,
+            SubmitOutcome::Hung => None,
+        }
+    }
+
+    pub fn error(&self) -> Option<&StoreError> {
+        match self {
+            SubmitOutcome::Refused(error) => Some(error),
+            SubmitOutcome::Committed(_) => None,
+            SubmitOutcome::CallerPanicked => None,
+            SubmitOutcome::Hung => None,
+        }
+    }
+
+    /// Whether the shard named itself poisoned in this call's own answer.
+    pub fn names_shard_poisoned(&self) -> bool {
+        matches!(self.error(), Some(StoreError::ShardPoisoned { .. }))
+    }
+}
+
+/// Submit one transaction, catching a panic raised in the caller's task.
+///
+/// `BeforeResponse` fires on the caller's side of the completion, so a
+/// `Panic` there unwinds *here* and not on a writer thread. Catching it is not
+/// leniency: the row's claim is that the transaction stays committed and its
+/// receipt stays queryable even then, and a harness that dies with the caller
+/// cannot check that.
+pub fn submit(engine: &StoreEngine, transaction: ValidatedTransaction) -> SubmitOutcome {
+    let previous = std::panic::take_hook();
+    std::panic::set_hook(Box::new(|_| {}));
+    let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+        block_on_until(engine.submit(transaction), SUBMIT_DEADLINE)
+    }));
+    std::panic::set_hook(previous);
+    match caught {
+        Ok(Some(Ok(receipt))) => SubmitOutcome::Committed(receipt),
+        Ok(Some(Err(error))) => SubmitOutcome::Refused(error),
+        Ok(None) => SubmitOutcome::Hung,
+        Err(_) => SubmitOutcome::CallerPanicked,
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Status classification, exhaustively
+// ---------------------------------------------------------------------------
+
+/// Map a `TransactionStatus` onto the oracle's `ImmediateStatus`.
+///
+/// `Unknown` is `DefinitiveAbsent` and that is a claim worth stating rather
+/// than assuming. `transaction_status` performs plan §5.1's two-root read: it
+/// loads the committed root, then the status root, then the committed root
+/// again. A reader that observes no entry in any of the three has synchronized
+/// with the store that removed it, so "nothing is recorded" is a positive
+/// finding about a linearizable read, not an absence of information.
+///
+/// `Pending` and `Expired` are enumerated and refused. Neither can arise from
+/// a Wave B row — a group is marked `Resolving` before any byte is written and
+/// a deadline is rechecked immediately before that — so seeing one is a
+/// finding, and mapping it into a neighbouring bucket would hide it.
+pub fn immediate_status_of(status: &TransactionStatus, row: &str) -> ImmediateStatus {
+    match status {
+        TransactionStatus::Committed(_) => ImmediateStatus::Committed,
+        TransactionStatus::Resolving { .. } => ImmediateStatus::Resolving,
+        TransactionStatus::Unknown => ImmediateStatus::DefinitiveAbsent,
+        TransactionStatus::Pending { phase, .. } => panic!(
+            "row {row}: transaction_status reported Pending({phase:?}); no Wave B failpoint \
+             can leave an operation in a pre-Resolving phase, so this is a finding rather \
+             than a status to classify"
+        ),
+        TransactionStatus::Expired { .. } => panic!(
+            "row {row}: transaction_status reported Expired; the retry deadline in this \
+             harness is ten minutes out, so an expiry here means the deadline recheck ran \
+             against the wrong clock"
+        ),
+    }
+}
+
+pub fn immediate_status_name(status: ImmediateStatus) -> &'static str {
+    match status {
+        ImmediateStatus::DefinitiveAbsent => "DefinitiveAbsent",
+        ImmediateStatus::Resolving => "Resolving",
+        ImmediateStatus::Committed => "Committed",
+    }
+}
+
+/// What a reopened store says about an operation, as a tail fact the frozen
+/// `outcome_admits` relation can be applied to.
+pub fn recovered_fact(status: &TransactionStatus, row: &str) -> RecoveredTailFact {
+    match status {
+        TransactionStatus::Committed(_) => RecoveredTailFact::CompleteChecksumValid,
+        TransactionStatus::Unknown => RecoveredTailFact::AbsentOrTorn,
+        TransactionStatus::Resolving { .. } => panic!(
+            "row {row}: an operation is still Resolving after a close and reopen through \
+             production recovery. Recovery publishes a committed root and a receipt table \
+             exactly once (scope 3.8 step 11); a surviving Resolving entry would mean the \
+             status root outlived the process that owned it"
+        ),
+        TransactionStatus::Pending { .. } => panic!(
+            "row {row}: a reopened store reported Pending, which no recovery path constructs"
+        ),
+        TransactionStatus::Expired { .. } => panic!(
+            "row {row}: a reopened store reported Expired for an operation whose deadline is \
+             ten minutes out"
+        ),
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Arming
+// ---------------------------------------------------------------------------
+
+pub fn action_name(action: FailpointAction) -> &'static str {
+    match action {
+        FailpointAction::Continue => "continue",
+        FailpointAction::Fail => "fail",
+        FailpointAction::Panic => "panic",
+        FailpointAction::HardExit => "hard-exit",
+    }
+}
+
+pub fn action_from_name(name: &str) -> Option<FailpointAction> {
+    [
+        FailpointAction::Continue,
+        FailpointAction::Fail,
+        FailpointAction::Panic,
+        FailpointAction::HardExit,
+    ]
+    .into_iter()
+    .find(|action| action_name(*action) == name)
+}
+
+/// Exclusive use of the process-global failpoint and fault registries.
+///
+/// The same token `drive.rs` hands the Wave A driver, taken for the same
+/// reason: the registries are one-shot globals, and a test that merely submits
+/// can consume an arming another test placed.
+pub type Serial = levcs_store::drive::faults::FaultSerial;
+
+pub fn serial() -> Serial {
+    levcs_store::drive::faults::serial()
+}
+
+pub fn arm(serial: &Serial, point: Failpoint, action: FailpointAction) {
+    levcs_store::failpoints::arm(serial, point, action);
+}
+
+pub fn disarm(serial: &Serial) {
+    levcs_store::failpoints::disarm(serial);
+}
+
+// ---------------------------------------------------------------------------
+// Reopening after a close
+// ---------------------------------------------------------------------------
+
+/// How long the harness will wait for `LOCK` after `StoreEngine` was dropped.
+pub const LOCK_RELEASE_BUDGET: Duration = Duration::from_secs(30);
+
+/// What one close-and-reopen cost.
+pub struct Reopen {
+    pub engine: StoreEngine,
+    /// Attempts made. `1` means the lock was already free when `drop` returned.
+    pub attempts: u32,
+    pub waited: Duration,
+}
+
+/// Reopen a root after its engine was dropped, waiting for the root lock.
+///
+/// # This wait is compensating for a defect, and it is not the harness's
+///
+/// `StoreEngine::drop` closes every shard channel and joins every writer
+/// thread, so on its own account the root lock is released by the time `drop`
+/// returns. Measured, it is not: reopening immediately after `drop` returns
+/// fails `AlreadyLocked` in roughly one run in six, and the lock then becomes
+/// free between a few hundred microseconds and about 150 milliseconds later.
+/// Something outlives the join and holds the `RecoverySession`.
+///
+/// This matters outside the harness. Scope 3.1 says `AlreadyLocked` is a
+/// refusal and **never a wait**, so a consumer that closes a store and reopens
+/// it — a recovery drill, an in-place restart, the Phase 2 migrator — gets a
+/// spurious refusal with no defined retry. It is reported to the lead and to
+/// B1 as a finding rather than fixed here: `engine.rs` is not B4's file, and
+/// re-deriving its shutdown sequence in the harness is exactly the local
+/// restatement the ownership matrix exists to prevent.
+///
+/// The wait is therefore **bounded, measured, and reported**, not silent. It is
+/// not a retry-until-green: a single reopen either succeeds within the budget
+/// or the run fails, and `attempts` is published so a regression that lengthens
+/// the window shows up as a number rather than as an intermittent failure.
+pub fn reopen_after_close(root: &Path, shard_count: u16, max_index_runs: u32) -> Reopen {
+    let started = Instant::now();
+    let mut attempts = 0u32;
+    loop {
+        attempts += 1;
+        match StoreEngine::open(options_with_index_runs(root, shard_count, max_index_runs)) {
+            Ok(engine) => {
+                return Reopen {
+                    engine,
+                    attempts,
+                    waited: started.elapsed(),
+                }
+            }
+            Err(StoreError::AlreadyLocked) if started.elapsed() < LOCK_RELEASE_BUDGET => {
+                std::thread::sleep(Duration::from_micros(200));
+            }
+            Err(StoreError::AlreadyLocked) => panic!(
+                "the root lock was still held {:?} after StoreEngine::drop returned, over \
+                 {attempts} attempts. The known window is under a second; this is longer, \
+                 which means the holder is not merely slow to be reclaimed.",
+                started.elapsed()
+            ),
+            Err(other) => panic!("reopen through production recovery failed: {other:?}"),
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// One driven row
+// ---------------------------------------------------------------------------
+
+/// Everything one Wave B row's run observed, through consumer-callable methods
+/// only.
+///
+/// The four fields Wave A could not reach are derived here from **four
+/// different observations**, not from one restated four ways:
+///
+/// | field | observation |
+/// |---|---|
+/// | `shard_poisoned` | the store names itself poisoned — a `StoreError::ShardPoisoned` from the victim's own answer or the probe's |
+/// | `immediate_status` | `transaction_status` on the victim operation, which is plan §5.1's two-root read |
+/// | `acknowledgment_allowed` | a `CommitReceipt` is obtainable for the victim, from `submit`'s return value or from a retrying status read |
+/// | `later_append_allowed_before_recovery` | a *different* transaction submitted to the same shard commits |
+///
+/// The last two are genuinely distinct from the first two. A shard can be
+/// unpoisoned and still refuse a later append, because a dead writer thread
+/// makes `later_append_allowed_before_recovery` false whatever the error says —
+/// that is the defect B1's phase-aware panic ownership closes, and folding the
+/// two into one observation would have made it invisible here.
+pub struct RowObservation {
+    pub row: String,
+    pub action: FailpointAction,
+    pub victim: SubmitOutcome,
+    pub immediate: TransactionStatus,
+    pub probe: SubmitOutcome,
+    pub recovered: TransactionStatus,
+    /// The transaction committed *before* the failpoint was armed. Every row
+    /// also proves an already-fenced, already-acknowledged transaction
+    /// survives, exactly as the Wave A rows prove it for a prior group.
+    pub prior_recovered: TransactionStatus,
+    /// `fdatasync` on the victim's shard, immediately before the failpoint was
+    /// armed and immediately after the victim's submit resolved. Charter item
+    /// 7: the physical state class is checked against a syscall count, not
+    /// against a comment about where the fence sits.
+    pub fences_before: u64,
+    pub fences_after: u64,
+    /// How many `StoreEngine::open` attempts the reopen needed, and how long it
+    /// waited. `1` and a near-zero wait is the expected reading; anything else
+    /// is the root-lock release window documented on [`reopen_after_close`].
+    pub lock_release_attempts: u32,
+    pub lock_release_wait: Duration,
+}
+
+impl RowObservation {
+    pub fn shard_poisoned_observed(&self) -> bool {
+        self.victim.names_shard_poisoned() || self.probe.names_shard_poisoned()
+    }
+
+    pub fn later_append_allowed_observed(&self) -> bool {
+        matches!(self.probe, SubmitOutcome::Committed(_))
+    }
+
+    /// A receipt is obtainable for the victim operation through a path a
+    /// consumer calls.
+    ///
+    /// `submit` returning one is the direct case. A status read returning
+    /// `Committed` is the retry case, and it is not redundant: a hung waiter
+    /// gets no return value at all, and the row's claim is precisely that the
+    /// transaction is still acknowledgeable then.
+    pub fn acknowledgment_allowed_observed(&self) -> bool {
+        self.victim.receipt().is_some() || matches!(self.immediate, TransactionStatus::Committed(_))
+    }
+
+    pub fn fences_during_victim(&self) -> u64 {
+        self.fences_after - self.fences_before
+    }
+}
+
+/// Drive one Wave B row on a single-shard root.
+///
+/// The shape mirrors the Wave A driving loop: a transaction that commits
+/// cleanly first, then the victim with the failpoint armed, then the
+/// classification. The difference is that the classification happens on a live
+/// engine as well as on a reopened one.
+pub fn drive_submit_row(
+    serial: &Serial,
+    point: Failpoint,
+    action: FailpointAction,
+) -> RowObservation {
+    let row = format!("{} [{}]", point.name(), action_name(action));
+    let directory = tempfile::tempdir().expect("tempdir");
+    let root = directory.path().join("root");
+    seed_root(&root, 1);
+
+    let namespace = namespace_on_shard(0, 1, 0x5643_5342);
+    let prior_operation = OperationId([0x11; 16]);
+    let victim_operation = OperationId([0x22; 16]);
+
+    let engine = StoreEngine::open(options(&root, 1)).expect("open the seeded root");
+
+    let prior = submit(&engine, create_transaction(namespace, 0x11));
+    assert!(
+        matches!(prior, SubmitOutcome::Committed(_)),
+        "row {row}: the transaction before the fault must commit cleanly, or the row \
+         proves nothing about what the fault cost: {prior:?}"
+    );
+
+    let fences_before = engine
+        .durability_counters(0)
+        .expect("shard 0 has a writer")
+        .fdatasync;
+
+    arm(serial, point, action);
+    let victim = submit(&engine, push_transaction(namespace, 0x22, 0xb2));
+    let fences_after = engine
+        .durability_counters(0)
+        .expect("shard 0 has a writer")
+        .fdatasync;
+    let immediate = engine
+        .transaction_status(namespace, victim_operation)
+        .expect("transaction_status is a read and never fails on an open engine");
+
+    let probe = submit(&engine, push_transaction(namespace, 0x33, 0xb3));
+    // Whatever happened, nothing may stay armed for the next row: the registry
+    // is a one-shot global and a row that did not fire would otherwise hand its
+    // arming to whoever submits next.
+    disarm(serial);
+    drop(engine);
+
+    let reopen = reopen_after_close(&root, 1, DEFAULT_MAX_INDEX_RUNS);
+    let lock_release_attempts = reopen.attempts;
+    let lock_release_wait = reopen.waited;
+    let reopened = reopen.engine;
+    let recovered = reopened
+        .transaction_status(namespace, victim_operation)
+        .expect("status read");
+    let prior_recovered = reopened
+        .transaction_status(namespace, prior_operation)
+        .expect("status read");
+    drop(reopened);
+
+    RowObservation {
+        row,
+        action,
+        victim,
+        immediate,
+        probe,
+        recovered,
+        prior_recovered,
+        fences_before,
+        fences_after,
+        lock_release_attempts,
+        lock_release_wait,
+    }
+}
+
+// ---------------------------------------------------------------------------
+// DuringRootCasRetry: the one location behind a lost compare-and-swap
+// ---------------------------------------------------------------------------
+
+/// Shards used by the contention driver.
+///
+/// The committed root is one `ArcSwap` for the whole store, so contention on
+/// it is contention *between shards*. One shard can never reach
+/// `DuringRootCasRetry`, however long it runs, because nothing else ever
+/// swaps the root out from under it.
+pub const CONTENTION_SHARDS: u16 = 8;
+
+/// Objects per contending transaction.
+///
+/// The window this row needs is the interval between `publish_subtree`'s load
+/// of the committed root and its compare-and-swap, and the work inside that
+/// window is `CommittedRoot::merge`. A transaction carrying one object leaves a
+/// window of a few hundred nanoseconds; carrying this many leaves a window
+/// wide enough that eight shards publishing concurrently collide within
+/// seconds rather than within a statistical hope. This widens a real window in
+/// the production path — it does not add one.
+const CONTENTION_OBJECTS: usize = 48;
+
+fn contending_transaction(namespace: NamespaceId, operation: u64) -> ValidatedTransaction {
+    let authority = genesis_id(&namespace);
+    let mut operation_id = [0u8; 16];
+    operation_id[..8].copy_from_slice(&operation.to_le_bytes());
+    operation_id[8..].copy_from_slice(namespace.as_bytes()[..8].try_into().expect("8 bytes"));
+
+    let objects = (0..CONTENTION_OBJECTS as u64)
+        .map(|index| {
+            let mut bytes = [0u8; 32];
+            let mut hasher = blake3::Hasher::new();
+            hasher.update(b"levcs-store/b4/contention-object/v1\0");
+            hasher.update(namespace.as_bytes());
+            hasher.update(&operation.to_le_bytes());
+            hasher.update(&index.to_le_bytes());
+            hasher.finalize_xof().fill(&mut bytes);
+            StagedObject {
+                id: ObjectId(bytes),
+                object_type: ObjectType::Blob,
+                raw: bytes.to_vec(),
+            }
+        })
+        .collect();
+
+    let mut digest = [0u8; 32];
+    digest[..16].copy_from_slice(&operation_id);
+    ValidatedTransaction::builder(privileged())
+        .namespace(namespace)
+        .operation(OperationId(operation_id), ObjectId(digest), deadline())
+        .objects(objects)
+        .refs(Vec::new())
+        .authority(Some(authority), Some(authority))
+        .evidence(evidence())
+        .build()
+        .expect("a complete contending transaction")
+}
+
+/// A submit with no panic-hook manipulation, for use from several threads at
+/// once.
+///
+/// The single-row driver installs a silent hook so a `Panic` row does not spray
+/// a backtrace over the test output. Doing that from eight threads would be a
+/// race on a process-global, so the contention driver accepts the noise
+/// instead. `DuringRootCasRetry` panics on a *writer* thread, never on the
+/// caller's, so nothing here needs catching.
+fn submit_plain(engine: &StoreEngine, transaction: ValidatedTransaction) -> SubmitOutcome {
+    match block_on_until(engine.submit(transaction), SUBMIT_DEADLINE) {
+        Some(Ok(receipt)) => SubmitOutcome::Committed(receipt),
+        Some(Err(error)) => SubmitOutcome::Refused(error),
+        None => SubmitOutcome::Hung,
+    }
+}
+
+/// What the contention driver measured, whether or not it fired.
+pub struct ContentionRun {
+    pub observation: Option<RowObservation>,
+    /// A refusal seen during the run that is not the one this location
+    /// produces. Non-`None` means the run observed something the row does not
+    /// describe, and the row must not be asserted from it.
+    pub anomaly: Option<(u16, String)>,
+    /// Transactions submitted across every shard before the location was
+    /// reached. Reported so "it fired" is a measurement rather than a claim,
+    /// and so a regression that makes the window narrower shows up as a number
+    /// climbing rather than as an intermittent failure.
+    pub attempts: u64,
+    pub elapsed: Duration,
+}
+
+/// Drive `DuringRootCasRetry` by manufacturing a genuinely lost
+/// compare-and-swap.
+///
+/// **This is the one Wave B row whose location is not reachable by submitting
+/// a transaction.** It sits inside `publish_subtree`'s retry loop, after a
+/// failed `compare_and_swap`, so it is reached only when another shard
+/// publishes between this shard's load and its swap. B1 disclosed that the
+/// failpoint is never armed by any existing test; this is what arms it.
+///
+/// The contention is real, not simulated: eight shard writer threads publish
+/// into the same `ArcSwap<CommittedRoot>` through the production path, and the
+/// only thing the harness chooses is how much work each publication carries
+/// into the window. Nothing here reaches into the engine.
+pub fn drive_root_cas_retry_row(
+    serial: &Serial,
+    action: FailpointAction,
+    budget: Duration,
+) -> ContentionRun {
+    use std::sync::atomic::AtomicBool;
+    use std::sync::Mutex;
+
+    let directory = tempfile::tempdir().expect("tempdir");
+    let root = directory.path().join("root");
+    seed_root(&root, CONTENTION_SHARDS);
+
+    let namespaces: Vec<NamespaceId> = (0..CONTENTION_SHARDS)
+        .map(|shard| namespace_on_shard(shard, CONTENTION_SHARDS, 0x0CA5_0000 + shard as u64))
+        .collect();
+
+    let engine = StoreEngine::open(options_with_index_runs(
+        &root,
+        CONTENTION_SHARDS,
+        CONTENTION_MAX_INDEX_RUNS,
+    ))
+    .expect("open the seeded root");
+    for (index, namespace) in namespaces.iter().enumerate() {
+        let created = submit_plain(&engine, create_transaction(*namespace, 0x40 + index as u8));
+        assert!(
+            matches!(created, SubmitOutcome::Committed(_)),
+            "the repository on shard {index} must be created before the contention run: \
+             {created:?}"
+        );
+    }
+
+    arm(serial, Failpoint::DuringRootCasRetry, action);
+
+    let stop = AtomicBool::new(false);
+    let attempts = AtomicU64::new(0);
+    #[allow(clippy::type_complexity)]
+    let victim: Mutex<Option<(u16, NamespaceId, OperationId, SubmitOutcome, u64)>> =
+        Mutex::new(None);
+    // A refusal that is *not* the row's. The location poisons the shard, so the
+    // only answer this run's victim can give is `ShardPoisoned`; anything else
+    // is a different event that happened to land in the same loop, and adopting
+    // it as the victim would assert this row against a transaction the
+    // failpoint never touched. Recorded rather than skipped: a run that meets
+    // one is not a run that can be repeated until it does not.
+    let anomaly: Mutex<Option<(u16, String)>> = Mutex::new(None);
+    let started = Instant::now();
+
+    std::thread::scope(|scope| {
+        for shard in 0..CONTENTION_SHARDS {
+            let namespace = namespaces[shard as usize];
+            let engine = &engine;
+            let stop = &stop;
+            let attempts = &attempts;
+            let victim = &victim;
+            let anomaly = &anomaly;
+            scope.spawn(move || {
+                let mut operation = 0u64;
+                while !stop.load(Ordering::Relaxed) && started.elapsed() < budget {
+                    operation += 1;
+                    let transaction = contending_transaction(namespace, operation);
+                    let mut operation_id = [0u8; 16];
+                    operation_id[..8].copy_from_slice(&operation.to_le_bytes());
+                    operation_id[8..]
+                        .copy_from_slice(namespace.as_bytes()[..8].try_into().expect("8 bytes"));
+                    attempts.fetch_add(1, Ordering::Relaxed);
+                    // Read immediately before the call under test. Only this
+                    // shard's writer advances this shard's counter and only
+                    // this thread submits to this shard, so the difference
+                    // across the submit is exactly the fences that submit
+                    // caused — no hook, no shared counter, no inference.
+                    let fences_before = engine
+                        .durability_counters(shard)
+                        .expect("every shard has a writer")
+                        .fdatasync;
+                    let outcome = submit_plain(engine, transaction);
+                    match outcome {
+                        SubmitOutcome::Committed(_) => {}
+                        SubmitOutcome::Refused(_) | SubmitOutcome::Hung => {
+                            if !outcome.names_shard_poisoned() {
+                                let mut slot = anomaly.lock().expect("anomaly mutex");
+                                if slot.is_none() {
+                                    *slot = Some((shard, format!("{outcome:?}")));
+                                }
+                                stop.store(true, Ordering::Relaxed);
+                                return;
+                            }
+                            let mut slot = victim.lock().expect("victim mutex");
+                            if slot.is_none() {
+                                *slot = Some((
+                                    shard,
+                                    namespace,
+                                    OperationId(operation_id),
+                                    outcome,
+                                    fences_before,
+                                ));
+                            }
+                            stop.store(true, Ordering::Relaxed);
+                            return;
+                        }
+                        SubmitOutcome::CallerPanicked => unreachable!(
+                            "submit_plain does not catch, so a caller panic would have \
+                             unwound this thread"
+                        ),
+                    }
+                }
+            });
+        }
+    });
+
+    let elapsed = started.elapsed();
+    let attempts = attempts.load(Ordering::Relaxed);
+    let victim = victim.into_inner().expect("victim mutex");
+    let anomaly = anomaly.into_inner().expect("anomaly mutex");
+
+    let Some((shard, namespace, victim_operation, victim_outcome, fences_before)) = victim else {
+        disarm(serial);
+        drop(engine);
+        return ContentionRun {
+            observation: None,
+            anomaly,
+            attempts,
+            elapsed,
+        };
+    };
+
+    let fences_after = engine
+        .durability_counters(shard)
+        .expect("the poisoned shard still has a counter")
+        .fdatasync;
+    let immediate = engine
+        .transaction_status(namespace, victim_operation)
+        .expect("status read");
+    let probe = submit_plain(&engine, contending_transaction(namespace, u64::MAX / 2));
+    disarm(serial);
+    drop(engine);
+
+    let reopen = reopen_after_close(&root, CONTENTION_SHARDS, CONTENTION_MAX_INDEX_RUNS);
+    let lock_release_attempts = reopen.attempts;
+    let lock_release_wait = reopen.waited;
+    let reopened = reopen.engine;
+    let recovered = reopened
+        .transaction_status(namespace, victim_operation)
+        .expect("status read");
+    // The repository-create on the same shard is the transaction acknowledged
+    // before the fault; it must survive.
+    let prior_recovered = reopened
+        .transaction_status(namespace, OperationId([0x40 + shard as u8; 16]))
+        .expect("status read");
+    drop(reopened);
+
+    ContentionRun {
+        anomaly,
+        attempts,
+        elapsed,
+        observation: Some(RowObservation {
+            row: format!("DuringRootCasRetry [{}]", action_name(action)),
+            action,
+            victim: victim_outcome,
+            immediate,
+            probe,
+            recovered,
+            prior_recovered,
+            fences_before,
+            fences_after,
+            lock_release_attempts,
+            lock_release_wait,
+        }),
+    }
+}
diff --git a/crates/levcs-store/tests/support/harness.rs b/crates/levcs-store/tests/support/harness.rs
index 2dadb76..ea363f7 100644
--- a/crates/levcs-store/tests/support/harness.rs
+++ b/crates/levcs-store/tests/support/harness.rs
@@ -23,11 +23,14 @@ use super::group_model::{PhysicalStateClass, VictimPlacement};
 
 pub const FIXTURE_PATH: &str = "tests/fixtures/phase1-failpoints.json";
 
-/// The `wave` value a row carries when it cannot be driven until B1 lands.
+/// The `wave` value a row carried while it could not be driven at all.
 ///
 /// The same spelling Phase 0 used for its `not-exercised` adversarial rows,
 /// and the string `scripts/check-phase1.sh` greps for once `engine.rs` is
-/// implemented.
+/// implemented. **No row carries it any more** — B1's engine landed and scope
+/// 6.6 deliverable 2 drove all eight — and the constant stays because the test
+/// that proves the pending set is empty has to name what it is looking for.
+/// Deleting it would leave the check searching for nothing and passing.
 pub const PENDING_WAVE_B: &str = "pending-wave-b";
 
 #[derive(Clone, Debug)]
@@ -36,6 +39,23 @@ pub struct DrivePlan {
     pub fault: String,
 }
 
+/// How a Wave B row is driven through `StoreEngine::submit`.
+///
+/// `actions` is a list rather than a single value because the failpoint enum
+/// names a *location* and the driver chooses the *action*; the two axes are
+/// independent, and the three publication-side rows must be exercised with
+/// both `Fail` and `Panic`. A row that named one action could not express that
+/// without a second row for the same location, which would break the
+/// one-row-per-failpoint invariant the first test in the matrix pins.
+#[derive(Clone, Debug)]
+pub struct SubmitPlan {
+    pub actions: Vec<String>,
+    /// Whether the location is reachable only after a lost committed-root CAS.
+    /// True for exactly one row, and the matrix asserts that rather than
+    /// letting a second row quietly acquire it.
+    pub requires_root_cas_contention: bool,
+}
+
 #[derive(Clone, Debug)]
 pub struct FixtureRow {
     pub failpoint: String,
@@ -44,7 +64,7 @@ pub struct FixtureRow {
     pub required_outcome: String,
     pub victim_placement: String,
     pub drive: Option<DrivePlan>,
-    pub pending_reason: Option<String>,
+    pub submit: Option<SubmitPlan>,
     pub rationale: String,
 }
 
@@ -53,6 +73,7 @@ pub struct Fixture {
     pub rows: Vec<FixtureRow>,
     pub wave_a_asserted: Vec<String>,
     pub wave_a_unasserted: Vec<String>,
+    pub wave_b_asserted: Vec<String>,
     pub wave_b_exit_conditions: Vec<String>,
 }
 
@@ -104,6 +125,21 @@ pub fn load_fixture() -> Fixture {
         "the unasserted halves must carry their reason, not merely be listed"
     );
 
+    let wave_b_scope = document
+        .get("wave_b_assertion_scope")
+        .expect("the fixture must record which halves Wave B asserts");
+    let wave_b_asserted = string_list(wave_b_scope.get("asserted").expect("asserted"));
+    assert!(
+        wave_b_scope
+            .get("root_seeded_by")
+            .and_then(|v| v.as_str())
+            .is_some_and(|r| !r.is_empty()),
+        "the Wave B rows are driven against a root that StoreEngine::open cannot yet \
+         create, and the fixture must say by what path it was created instead. Scope 5 \
+         charter item 8 is about exactly this: a property asserted against a store that \
+         production never built is asserted against the wrong store."
+    );
+
     let wave_b_exit_conditions = string_list(
         document
             .get("wave_b_exit_conditions")
@@ -127,10 +163,15 @@ pub fn load_fixture() -> Fixture {
                     fault: string_field(plan, "fault"),
                 })
             }),
-            pending_reason: row
-                .get("pending_reason")
-                .and_then(|v| v.as_str())
-                .map(|s| s.to_string()),
+            submit: row.get("submit").and_then(|plan| {
+                plan.as_object().map(|_| SubmitPlan {
+                    actions: string_list(plan.get("actions").expect("submit.actions")),
+                    requires_root_cas_contention: plan
+                        .get("requires_root_cas_contention")
+                        .and_then(|v| v.as_bool())
+                        .expect("submit.requires_root_cas_contention"),
+                })
+            }),
             rationale: string_field(row, "rationale"),
         })
         .collect();
@@ -139,6 +180,7 @@ pub fn load_fixture() -> Fixture {
         rows,
         wave_a_asserted,
         wave_a_unasserted,
+        wave_b_asserted,
         wave_b_exit_conditions,
     }
 }
@@ -147,13 +189,17 @@ pub fn load_fixture() -> Fixture {
 // Name mapping, without catch-all arms
 // ---------------------------------------------------------------------------
 
-/// The fixture spells `Wave::A` as `"A"` and `Wave::B` as `"pending-wave-b"`,
-/// because a Wave B row is not merely "later" — it is an outstanding, named
-/// obligation, and `check-phase1.sh` greps for that exact string.
+/// The fixture spells `Wave::A` as `"A"` and `Wave::B` as `"B"`.
+///
+/// Wave B used to be spelled `pending-wave-b`, because a row that could not be
+/// driven was an outstanding obligation rather than merely a later one. The
+/// obligation is met: every Wave B row is now driven through
+/// `StoreEngine::submit`, so the wave is just a wave and the pending spelling
+/// belongs to no row.
 pub fn wave_name(wave: Wave) -> &'static str {
     match wave {
         Wave::A => "A",
-        Wave::B => PENDING_WAVE_B,
+        Wave::B => "B",
     }
 }
 
diff --git a/crates/levcs-store/tests/support/mod.rs b/crates/levcs-store/tests/support/mod.rs
index ea32ebb..886bcc9 100644
--- a/crates/levcs-store/tests/support/mod.rs
+++ b/crates/levcs-store/tests/support/mod.rs
@@ -10,3 +10,15 @@
 
 pub mod group_model;
 pub mod harness;
+
+/// The Wave B submit-path harness (scope 6.6 deliverable 2).
+///
+/// Gated on `store-privileged` because building a `ValidatedTransaction` at
+/// all requires `PrivilegedConstruction`, which scope 2.2 deliberately makes
+/// unreachable without that feature. The Phase 1 gate runs
+/// `failpoints,store-internals,store-privileged`, so the eight Wave B rows are
+/// exercised on every gate run; `crash_matrix::the_submit_path_rows_are_driven_
+/// in_this_configuration` fails loudly rather than passing silently if they
+/// are not.
+#[cfg(feature = "store-privileged")]
+pub mod engine_matrix;