Implement the partial B1 namespace-transaction slice

A narrow vertical slice of scope 6.4: StoreEngine::open through one
retained RecoverySession, every shard recovered into the initial
CommittedRoot, and one shard writer wired GroupBuilder -> append_group_and_fence
-> ShardSubtree -> CAS publication -> completion. This closes the Wave A
carry-forward by giving GroupBuilder a production caller; the group bounds
are asserted against the sequencer as fdatasync counts on the device, not
against the builder in isolation.

§6.3's two load-bearing orderings are enforced and asserted, not assumed.
Step 8 follows step 7 with an explicit release fence, so a reader observing
the status root without an entry cannot then read a committed root older
than the publication -- verified by mutation in both directions. Waiters
are woken only after publication, and failure at steps 9-10 leaves a
queryable receipt, because a fence that succeeded and a root that published
is committed.

Three P1 findings from the B2 review are fixed here.

The committed event took its actor from the destination signer rather than
the evidence, so a frame could be signed, fenced, and published and then
fail frozen mirror verification -- first observed by another instance, long
after the bytes were durable. The test could not see it because the signer
key and the evidence actor were the same constant; they are now
deliberately different.

The deadline was rechecked in prepare, before signing and before the group
idle wait, so a slow signer could cross the retry deadline and still
append. The rule is now evaluated again over the whole group immediately
before anything is marked Resolving. The structural half matters more than
the recheck: shard_sequence is no longer consumed at prepare time, so a
dropped member leaves no hole by construction. Repair is re-derivation from
a recorded pre-image through the one sequencing function -- re-sequenced,
re-chained, re-signed -- because signing covers a digest that chains
previous_event_digest, and patching a suffix produces a durable, correctly
fenced frame whose signature verifies against nothing. The test reads every
frame back and checks both the chain and the signature; receipts alone
would not catch a partial repair.

Panic recovery had one catch around the whole writer loop, so every panic
poisoned the shard and reported every waiter as poisoned. Waiters now carry
an explicit phase and an owned reservation. A pre-append panic is
definitively absent and leaves the writer alive, since a dead writer makes
later_append_allowed_before_recovery false whatever the error says; a
request that overflows a publishing group is rolled back before that group
publishes rather than reported as its member; and a panic after publication
still delivers every receipt.

Also: the duplicated object-type and ref-kind tables are deleted in favour
of the amended shared helpers; max_objects_per_transaction and
max_refs_per_transaction are enforced at submit, the entry point a consumer
calls; status occupancy records the newly published root rather than the
one it replaced; and durability_counters/operation_status_metrics return
snapshots only, never the live counters, since a durability claim whose
auditor can write to it is not evidence.

StoreEngine::open constructs the single ProjectionStaging under the held
session before recovery -- it is recovery's projection resolver -- and
holds it exactly as long as the lock.

The remaining deliverables refuse by name: startup states 1/3/4,
coalescing, terminal retention, index sealing, journal rotation,
RepoSnapshot, and checkpoint. This is not a completed B1 deliverable set
and not freeze evidence; the gate's pending-row check stays inert while
engine.rs still returns NotImplemented.

138 library tests, 27 staging tests, 1 intentionally ignored.
scripts/check-phase1.sh GATE_EXIT=0.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

crates/levcs-store/src/transaction.rs

@@ -3,9 +3,25 @@
 //!
 //! **Shared file.** Lead owns the signatures (D0); **B1 NamespaceTxn** fills
 //! the bodies (scope 2.1, 6-B1).
+//!
+//! # What the builder settles and what the sequencer settles
+//!
+//! Everything here is fixed before the mutation lane is entered: the exact
+//! bytes, the complete ordered ref CAS set, the explicit authority CAS, the
+//! evidence, the operation identity, and the signed deadline. The sequencer
+//! then rechecks only what can have changed since — ref CAS, authority, and
+//! lifecycle — and never re-parses, re-hashes, or re-verifies any of it.
+//!
+//! `build` therefore refuses an incomplete transaction rather than filling a
+//! field in. A defaulted `expected_authority` or an inferred `new_authority`
+//! would turn an authority CAS into an unconditional overwrite, and the
+//! sequencer cannot tell the difference between a value a caller chose and one
+//! a builder supplied.
+
+use std::collections::BTreeSet;
 
 use levcs_core::{ObjectId, ObjectType};
-use levcs_protocol::v2::{TransactionEvidenceV1, TypedRefCas};
+use levcs_protocol::v2::{RefTarget, TransactionEvidenceV1, TypedRefCas};
 
 use crate::staging::{ProjectionAdoptionOutcome, StagedProjectionAdoption};
 use crate::types::{NamespaceId, OperationId, PrivilegedConstruction, StoreError};
@@ -19,7 +35,39 @@ use crate::types::{NamespaceId, OperationId, PrivilegedConstruction, StoreError}
 /// re-hashes, re-verifies signatures, or re-evaluates policy while holding the
 /// mutation lane.
 pub struct ValidatedTransaction {
-    _private: (),
+    pub(crate) namespace: NamespaceId,
+    pub(crate) operation_id: OperationId,
+    pub(crate) operation_digest: ObjectId,
+    pub(crate) retry_until_micros: i64,
+    /// Present only for an init transaction.
+    pub(crate) create_genesis_authority: Option<ObjectId>,
+    /// Strictly ascending by `ObjectId`, which is the canonical frame order.
+    /// Sorting here rather than at encode time means the sequencer never
+    /// reorders bytes it is holding the mutation lane for.
+    pub(crate) objects: Vec<StagedObject>,
+    pub(crate) refs: Vec<TypedRefCas>,
+    pub(crate) expected_authority: Option<ObjectId>,
+    pub(crate) new_authority: Option<ObjectId>,
+    pub(crate) evidence: TransactionEvidenceV1,
+}
+
+/// Deliberately does not print object or evidence bytes. A transaction's
+/// payload is user content and a diagnostic that dumps it is a diagnostic
+/// nobody can paste into a bug report.
+impl std::fmt::Debug for ValidatedTransaction {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("ValidatedTransaction")
+            .field("namespace", &self.namespace)
+            .field("operation_id", &self.operation_id)
+            .field(
+                "creates_repository",
+                &self.create_genesis_authority.is_some(),
+            )
+            .field("objects", &self.objects.len())
+            .field("refs", &self.refs.len())
+            .finish_non_exhaustive()
+    }
 }
 
 /// One object entering namespace membership.
@@ -33,47 +81,79 @@ impl ValidatedTransaction {
     /// Begin construction. Requires a capability token that is not reachable
     /// from a `&StoreEngine`; see `PrivilegedConstruction` (scope 2.2).
     pub fn builder(_token: PrivilegedConstruction) -> ValidatedTransactionBuilder {
-        ValidatedTransactionBuilder { adoption: None }
+        ValidatedTransactionBuilder {
+            namespace: None,
+            operation: None,
+            create_genesis_authority: None,
+            objects: None,
+            refs: None,
+            authority: None,
+            evidence: None,
+            adoption: None,
+        }
     }
 }
 
 pub struct ValidatedTransactionBuilder {
+    namespace: Option<NamespaceId>,
+    operation: Option<(OperationId, ObjectId, i64)>,
+    create_genesis_authority: Option<ObjectId>,
+    objects: Option<Vec<StagedObject>>,
+    refs: Option<Vec<TypedRefCas>>,
+    /// `Some((None, None))` is a caller that explicitly moves no authority.
+    /// `None` is a caller that never said, and is refused.
+    #[allow(clippy::type_complexity)]
+    authority: Option<(Option<ObjectId>, Option<ObjectId>)>,
+    evidence: Option<TransactionEvidenceV1>,
     adoption: Option<StagedProjectionAdoption>,
 }
 
+fn incomplete(field: &str) -> StoreError {
+    StoreError::InvalidConfiguration(format!(
+        "validated transaction is incomplete: {field} was never set"
+    ))
+}
+
 impl ValidatedTransactionBuilder {
-    pub fn namespace(self, _namespace: NamespaceId) -> Self {
+    pub fn namespace(mut self, namespace: NamespaceId) -> Self {
+        self.namespace = Some(namespace);
         self
     }
 
-    pub fn operation(self, _id: OperationId, _digest: ObjectId, _retry_until_micros: i64) -> Self {
+    pub fn operation(mut self, id: OperationId, digest: ObjectId, retry_until_micros: i64) -> Self {
+        self.operation = Some((id, digest, retry_until_micros));
         self
     }
 
     /// Repository-create metadata. Present only for an init transaction, which
     /// permanently binds `repo_id` and the genesis authority hash in the
     /// catalog (plan §4 identity invariant 2).
-    pub fn create_repository(self, _genesis_authority: ObjectId) -> Self {
+    pub fn create_repository(mut self, genesis_authority: ObjectId) -> Self {
+        self.create_genesis_authority = Some(genesis_authority);
         self
     }
 
-    pub fn objects(self, _objects: Vec<StagedObject>) -> Self {
+    pub fn objects(mut self, objects: Vec<StagedObject>) -> Self {
+        self.objects = Some(objects);
         self
     }
 
     /// The complete typed Set/Delete set. Multi-ref updates are entirely old
     /// or entirely new, including after crashes (plan §4 transaction
     /// invariant 2).
-    pub fn refs(self, _refs: Vec<TypedRefCas>) -> Self {
+    pub fn refs(mut self, refs: Vec<TypedRefCas>) -> Self {
+        self.refs = Some(refs);
         self
     }
 
     /// Explicit expected and new current authority. Never inferred.
-    pub fn authority(self, _expected: Option<ObjectId>, _new: Option<ObjectId>) -> Self {
+    pub fn authority(mut self, expected: Option<ObjectId>, new: Option<ObjectId>) -> Self {
+        self.authority = Some((expected, new));
         self
     }
 
-    pub fn evidence(self, _evidence: TransactionEvidenceV1) -> Self {
+    pub fn evidence(mut self, evidence: TransactionEvidenceV1) -> Self {
+        self.evidence = Some(evidence);
         self
     }
 
@@ -107,15 +187,79 @@ impl ValidatedTransactionBuilder {
 
     pub fn build(mut self) -> Result<ValidatedTransaction, StoreError> {
         if let Some(adoption) = self.adoption.take() {
-            // D0-B freezes the lifecycle while B1 still owns the successful
-            // builder body. `NotImplemented` is definitive and pre-append.
+            // Adoption is the B1/B3 seam of scope 6.2 item 9 and is not part
+            // of this slice. The pin is released with a definitive pre-append
+            // outcome before the refusal, because a dropped pin with no
+            // outcome is the leak §6.5 declares a bug — the refusal must not
+            // create one.
             adoption
                 .handle
                 .finish(ProjectionAdoptionOutcome::DefinitivePreAppendFailure)?;
+            return Err(StoreError::NotImplemented(
+                "ValidatedTransactionBuilder::adopt_projection — B1 NamespaceTxn, \
+                 scope 6-B1 deliverable 3 and scope 6.2 item 9",
+            ));
         }
-        Err(StoreError::NotImplemented(
-            "ValidatedTransactionBuilder::build — B1 NamespaceTxn, scope 6-B1 deliverable 3",
-        ))
+
+        let namespace = self.namespace.ok_or_else(|| incomplete("namespace"))?;
+        let (operation_id, operation_digest, retry_until_micros) =
+            self.operation.ok_or_else(|| incomplete("operation"))?;
+        let evidence = self.evidence.ok_or_else(|| incomplete("evidence"))?;
+        let (expected_authority, new_authority) =
+            self.authority.ok_or_else(|| incomplete("authority"))?;
+        let mut objects = self.objects.ok_or_else(|| incomplete("objects"))?;
+        let refs = self.refs.ok_or_else(|| incomplete("refs"))?;
+
+        // Canonical frame order, established once. A duplicate `ObjectId` is
+        // refused rather than deduplicated: two records for one object make
+        // `objects_new` and the index disagree about what the transaction
+        // introduced, and only the caller knows which one it meant.
+        objects.sort_by(|left, right| left.id.as_bytes().cmp(right.id.as_bytes()));
+        if objects.windows(2).any(|pair| pair[0].id == pair[1].id) {
+            return Err(StoreError::Conflict(
+                "a transaction may introduce one object at most once".into(),
+            ));
+        }
+
+        let mut targets: BTreeSet<&RefTarget> = BTreeSet::new();
+        for update in &refs {
+            if !targets.insert(&update.target) {
+                return Err(StoreError::Conflict(
+                    "a transaction may name one typed ref at most once; a duplicate target \
+                     makes the same-ref winner ambiguous inside a single frame"
+                        .into(),
+                ));
+            }
+        }
+
+        // An init transaction binds a genesis authority permanently, so the
+        // authority object itself has to be in this transaction's bytes. The
+        // frame records its exact length and hash, and neither can be invented
+        // from a bare `ObjectId`.
+        if let Some(genesis) = self.create_genesis_authority {
+            let present = objects.iter().any(|object| {
+                object.id == genesis && matches!(object.object_type, ObjectType::Authority)
+            });
+            if !present {
+                return Err(StoreError::Conflict(
+                    "a repository-create transaction must carry its genesis authority object"
+                        .into(),
+                ));
+            }
+        }
+
+        Ok(ValidatedTransaction {
+            namespace,
+            operation_id,
+            operation_digest,
+            retry_until_micros,
+            create_genesis_authority: self.create_genesis_authority,
+            objects,
+            refs,
+            expected_authority,
+            new_authority,
+            evidence,
+        })
     }
 }
 
@@ -165,6 +309,170 @@ mod tests {
         }
     }
 
+    pub(crate) fn administrative_evidence() -> TransactionEvidenceV1 {
+        TransactionEvidenceV1::AdministrativeV1 {
+            actor: [9; 32],
+            actor_key_epoch: 1,
+            command_digest: ObjectId([4; 32]),
+            signature: [5; 64],
+        }
+    }
+
+    fn complete() -> ValidatedTransactionBuilder {
+        ValidatedTransaction::builder(PrivilegedConstruction::internal())
+            .namespace(NamespaceId([1; 32]))
+            .operation(OperationId([2; 16]), ObjectId([3; 32]), 1_000)
+            .objects(Vec::new())
+            .refs(Vec::new())
+            .authority(None, None)
+            .evidence(administrative_evidence())
+    }
+
+    /// Every omitted field is refused by name. An enumerated list rather than
+    /// one representative case: the defect this guards against is a builder
+    /// that quietly defaults exactly one field, and a single-case test cannot
+    /// see which field that is.
+    #[test]
+    fn an_incomplete_builder_is_refused_field_by_field() {
+        complete().build().expect("the complete builder builds");
+
+        let bare = || ValidatedTransaction::builder(PrivilegedConstruction::internal());
+        let cases: Vec<(&str, ValidatedTransactionBuilder)> = vec![
+            (
+                "namespace",
+                bare()
+                    .operation(OperationId([2; 16]), ObjectId([3; 32]), 1_000)
+                    .objects(Vec::new())
+                    .refs(Vec::new())
+                    .authority(None, None)
+                    .evidence(administrative_evidence()),
+            ),
+            (
+                "operation",
+                bare()
+                    .namespace(NamespaceId([1; 32]))
+                    .objects(Vec::new())
+                    .refs(Vec::new())
+                    .authority(None, None)
+                    .evidence(administrative_evidence()),
+            ),
+            (
+                "evidence",
+                bare()
+                    .namespace(NamespaceId([1; 32]))
+                    .operation(OperationId([2; 16]), ObjectId([3; 32]), 1_000)
+                    .objects(Vec::new())
+                    .refs(Vec::new())
+                    .authority(None, None),
+            ),
+            (
+                "authority",
+                bare()
+                    .namespace(NamespaceId([1; 32]))
+                    .operation(OperationId([2; 16]), ObjectId([3; 32]), 1_000)
+                    .objects(Vec::new())
+                    .refs(Vec::new())
+                    .evidence(administrative_evidence()),
+            ),
+            (
+                "objects",
+                bare()
+                    .namespace(NamespaceId([1; 32]))
+                    .operation(OperationId([2; 16]), ObjectId([3; 32]), 1_000)
+                    .refs(Vec::new())
+                    .authority(None, None)
+                    .evidence(administrative_evidence()),
+            ),
+            (
+                "refs",
+                bare()
+                    .namespace(NamespaceId([1; 32]))
+                    .operation(OperationId([2; 16]), ObjectId([3; 32]), 1_000)
+                    .objects(Vec::new())
+                    .authority(None, None)
+                    .evidence(administrative_evidence()),
+            ),
+        ];
+
+        for (field, builder) in cases {
+            match builder.build() {
+                Err(StoreError::InvalidConfiguration(message)) => assert!(
+                    message.contains(field),
+                    "refusal must name the missing field {field}, got {message}"
+                ),
+                other => panic!("omitting {field} must be refused, got {other:?}"),
+            }
+        }
+    }
+
+    #[test]
+    fn a_duplicate_object_or_ref_target_is_refused() {
+        let object = |byte: u8| StagedObject {
+            id: ObjectId([byte; 32]),
+            object_type: ObjectType::Blob,
+            raw: vec![byte],
+        };
+        match complete().objects(vec![object(7), object(7)]).build() {
+            Err(StoreError::Conflict(message)) => {
+                assert!(message.contains("at most once"), "{message}")
+            }
+            other => panic!("a duplicate object must be refused, got {other:?}"),
+        }
+
+        let update = || TypedRefCas {
+            target: RefTarget::Branch("main".into()),
+            expected: None,
+            mutation: levcs_protocol::v2::RefMutation::Set(ObjectId([8; 32])),
+            force: false,
+        };
+        match complete().refs(vec![update(), update()]).build() {
+            Err(StoreError::Conflict(message)) => {
+                assert!(message.contains("same-ref winner"), "{message}")
+            }
+            other => panic!("a duplicate ref target must be refused, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn a_create_without_its_genesis_authority_object_is_refused() {
+        let genesis = ObjectId([12; 32]);
+        match complete().create_repository(genesis).build() {
+            Err(StoreError::Conflict(message)) => {
+                assert!(message.contains("genesis authority object"), "{message}")
+            }
+            other => panic!("a create with no genesis object must be refused, got {other:?}"),
+        }
+
+        complete()
+            .create_repository(genesis)
+            .objects(vec![StagedObject {
+                id: genesis,
+                object_type: ObjectType::Authority,
+                raw: vec![1, 2, 3],
+            }])
+            .build()
+            .expect("a create carrying its genesis authority builds");
+    }
+
+    #[test]
+    fn a_projection_adoption_is_refused_and_releases_its_pin() {
+        let lifecycle = Arc::new(Lifecycle::default());
+        let result = complete()
+            .adopt_projection(
+                descriptor([3; 16]),
+                ProjectionAdoption::new(lifecycle.clone()),
+            )
+            .expect("one adoption is admitted")
+            .build();
+
+        assert!(matches!(result, Err(StoreError::NotImplemented(_))));
+        assert_eq!(
+            &*lifecycle.outcomes.lock().unwrap(),
+            &[ProjectionAdoptionOutcome::DefinitivePreAppendFailure]
+        );
+        assert_eq!(*lifecycle.dropped.lock().unwrap(), 0);
+    }
+
     #[test]
     fn duplicate_projection_adoption_releases_both_pins() {
         let first = Arc::new(Lifecycle::default());