feat(store): the put seam — two internal paths, the three numbered rules, Batch, kv-only rejection

- `put(len: Option<u64>, stream: impl AsyncRead + Unpin + Send) -> (Key, Pin)`:
  known-length one-pass flow; unknown-length buffer-to-threshold flow with
  the mid-stream fresh-stage flush (ADR-003 option (d), three numbered rules:
  flush-and-continue, derivation over the eventual stored artifact as a
  whole via buffer form and StagingCursor read-back form, zero residue on
  stream error)
- staged-put seam amended (operator sign-off): `put_staged()` keyless
  handout, `StagingCursor::commit(self, key)` publishes under the
  store-core-resolved key — the canonical key completes at stream end and
  the wave-1 handout shape could not express the commit-rename flow
- pin-before-publish rides the arbitration lock on both paths and both
  tiers; dedup re-puts refcount the existing entry
- Batch seam: `Store::batch()` + `put_in_batch()`, members publish pin-
  before-publish, group pins drop together on `Batch::commit` (new public
  verb) or drop
- kv-only rejection (ADR-012 §6.2): known-length over-threshold rejects
  immediately (stream never read), unknown-length at mid-stream overflow;
  zero residue; error carrier `Io(String)` (named design point)
- declared-len mismatch rejects `Verification` zero-residue (named design
  point); stream shape generic-no-dyn (named design point)
- incremental BlobHasher (pinned byte-equal to Algorithm::derive);
  AnyStaging cursor dispatch; git-interop put-path leg against
  git hash-object; eleven put-path behavior tests

Verification: cargo test (default/--features mem/--all-features) green;
clippy --all-targets -D warnings green in all three feature shapes;
cargo fmt --check green; cargo doc --no-deps at the 24-warning baseline.

src/backend.rs

@@ -108,6 +108,17 @@ pub(crate) trait ReadCursor: Send {
 /// on discard with zero residue (the stage-hygiene invariant,
 /// store-api.md §Write path; exact-count tested per engine).
 ///
+/// **The key resolves at commit, not at handout (named design point,
+/// the wave-2 amendment of the wave-1 handout shape).** The canonical
+/// key is the derivation over preamble + content, and it completes at
+/// stream end — after the staging cursor already exists (put paths
+/// stream content once into the cursor while hashing; ADR-003's put
+/// paths). So `commit` carries the store-core-resolved key and the
+/// stage's identity before that point is only the engine-internal
+/// stage file/tx (unique per `put_staged` handout). Publish is the
+/// CAS first-writer-wins entry under that committed key exactly as
+/// [`Backend::put`].
+///
 /// Every engine implements the staged form: kv-tier bodies are the
 /// degenerate in-memory stage committed through the engine's whole-value
 /// row write (the store core selects the form by tier policy, ADR-010
@@ -121,15 +132,18 @@ pub(crate) trait StagingCursor: ReadCursor {
     /// "restart the stream" requirement anywhere).
     async fn append(&mut self, chunk: &[u8]) -> Result<(), Error>;
 
-    /// Publish the entry: the stage's bytes become the CAS entry in the
+    /// Publish the entry: the stage's bytes become the CAS entry under
+    /// `key` — the store-core-resolved canonical key (the derivation
+    /// completes at stream end; see the type's design point) — in the
     /// engine's put shape (commit-rename on `local`, the in-tx `lowrite`
-    /// batch on pg-lo, the row write on kv bodies). This is the staged
-    /// form's terminal step — the store core never re-uses the cursor
-    /// after the call. On failure the stage is **already discarded** —
-    /// zero residue by the stage-hygiene invariant; there is no
-    /// post-failure discard to drive (the consuming call took the
-    /// cursor).
-    async fn commit(self) -> Result<(), Error>;
+    /// batch on pg-lo, the row write on kv bodies). CAS first-writer-wins:
+    /// a losing race against an existing entry is dedup (`Ok`), never a
+    /// failure. This is the staged form's terminal step — the store core
+    /// never re-uses the cursor after the call. On failure the stage is
+    /// **already discarded** — zero residue by the stage-hygiene
+    /// invariant; there is no post-failure discard to drive (the
+    /// consuming call took the cursor).
+    async fn commit(self, key: &[u8]) -> Result<(), Error>;
 
     /// Discard the stage: every non-commit path converges here, zero
     /// residue. Driven explicitly by the store core (stream error,
@@ -189,11 +203,12 @@ pub(crate) trait Backend: Send + Sync {
     async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error>;
 
     /// The staged put's cursor handout (ADR-010 §1). The store core
-    /// selects the form before content flows: the large tier's
-    /// stage-then-commit-rename contract, kv engines' degenerate
-    /// in-memory stage. See [`StagingCursor`] for the contract every
-    /// body converges on.
-    async fn put_staged(&self, key: &[u8]) -> Result<Self::Staging, Error>;
+    /// selects the form before content flows (the cursor's identity
+    /// before commit is the engine-internal stage only — the canonical
+    /// key resolves at [`StagingCursor::commit`], the type's design
+    /// point). See [`StagingCursor`] for the contract every body
+    /// converges on.
+    async fn put_staged(&self) -> Result<Self::Staging, Error>;
 
     /// Unconditional delete — arbitration has already happened above the
     /// trait (the facade decides and owns the refusal: `GcRefuse` never
@@ -442,7 +457,6 @@ mod tests {
     }
 
     struct TestStaging {
-        key: Vec<u8>,
         staged: Vec<u8>,
         rows: Arc<Mutex<BTreeMap<Vec<u8>, Vec<u8>>>>,
         fail_commit: bool,
@@ -464,13 +478,13 @@ mod tests {
             Ok(())
         }
 
-        async fn commit(mut self) -> Result<(), Error> {
+        async fn commit(mut self, key: &[u8]) -> Result<(), Error> {
             if self.fail_commit {
                 self.staged.clear();
                 return Err(Error::Io("engine commit failure (test double)".to_owned()));
             }
             let value = std::mem::take(&mut self.staged);
-            self.rows.lock().entry(self.key.clone()).or_insert(value);
+            self.rows.lock().entry(key.to_vec()).or_insert(value);
             Ok(())
         }
 
@@ -508,9 +522,8 @@ mod tests {
             Ok(())
         }
 
-        async fn put_staged(&self, key: &[u8]) -> Result<Self::Staging, Error> {
+        async fn put_staged(&self) -> Result<Self::Staging, Error> {
             Ok(TestStaging {
-                key: key.to_vec(),
                 staged: Vec::new(),
                 rows: self.rows.clone(),
                 fail_commit: self.fail_commit,
@@ -603,7 +616,7 @@ mod tests {
         let backend = TestBackend::default();
         let key = b"\x01staged-key".to_vec();
 
-        let mut staging = backend.put_staged(&key).await.unwrap();
+        let mut staging = backend.put_staged().await.unwrap();
         staging.append(&CONTENT[..8]).await.unwrap();
         staging.append(&CONTENT[8..16]).await.unwrap();
         staging.append(&CONTENT[16..]).await.unwrap();
@@ -620,7 +633,7 @@ mod tests {
         assert_eq!(bytes, CONTENT, "staged artifact readable as a whole");
         assert_eq!(digest, slice_digest(CONTENT));
 
-        staging.commit().await.expect("commit publishes");
+        staging.commit(&key).await.expect("commit publishes");
         assert!(backend.has(&key).await.unwrap(), "entry published");
 
         let cursor = backend
@@ -650,7 +663,7 @@ mod tests {
         let backend = TestBackend::default();
         let key = b"\x01discard-key".to_vec();
 
-        let mut staging = backend.put_staged(&key).await.unwrap();
+        let mut staging = backend.put_staged().await.unwrap();
         staging.append(b"residue").await.unwrap();
         staging.discard().await;
 
@@ -676,9 +689,9 @@ mod tests {
         };
         let key = b"\x01failure-key".to_vec();
 
-        let mut staging = backend.put_staged(&key).await.unwrap();
+        let mut staging = backend.put_staged().await.unwrap();
         staging.append(CONTENT).await.unwrap();
-        let err = staging.commit().await.unwrap_err();
+        let err = staging.commit(&key).await.unwrap_err();
         assert!(matches!(err, Error::Io(_)), "engine media failure");
 
         assert!(

src/facade.rs

@@ -178,6 +178,15 @@ impl Pin {
     }
 }
 
+impl fmt::Debug for Pin {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Pin")
+            .field("key", &self.key.digest())
+            .field("armed", &self.armed)
+            .finish()
+    }
+}
+
 impl Drop for Pin {
     fn drop(&mut self) {
         if self.armed {
@@ -239,6 +248,17 @@ impl Batch {
         self.keys.is_empty()
     }
 
+    /// **Batch commit**: the group's pins drop together in one ordered
+    /// step — protection spans the whole batch or none of it, and ends
+    /// when the batch ends (ADR-005 §Marks 2; the same release batch
+    /// drop drives). A committed batch's members are no longer
+    /// batch-protected; durable protection beyond the write group is
+    /// the consumer's liveness seam (`Store::register_liveness_source`
+    /// or the pin-conversion ordering).
+    pub fn commit(mut self) {
+        self.release_members();
+    }
+
     /// The one-step release: every member refcount decremented inside
     /// a single arbitration-lock section, sorted into the keys' total
     /// order. Driven by [`Drop::drop`] (batch drop) and available for

src/hashing.rs

@@ -27,7 +27,7 @@
 //! is absent from the crate entirely (ADR-002 Decision class 3).
 
 use sha1::Sha1;
-use sha2::Sha256;
+use sha2::{Digest as _, Sha256};
 
 use crate::Error;
 
@@ -129,6 +129,61 @@ where
     hasher.finalize().to_vec()
 }
 
+/// The streaming form of the derivation: one hasher fed the preamble
+/// at initialization (`"blob <len>\0"`, the length known up front on
+/// the known-length put path) and the content as it flows. The finish
+/// is byte-identical to [`Algorithm::derive`] over the same input —
+/// this is the same derivation, driven incrementally so content can
+/// stream once (ADR-003's put paths; the wave-2 facade logic this
+/// module's doc anticipated).
+///
+/// Crate-internal: the derivation is a store-core concern; this type
+/// is the put path's tool, not public API.
+pub(crate) struct BlobHasher {
+    algorithm: Algorithm,
+    inner: InnerHasher,
+}
+
+enum InnerHasher {
+    Sha256(Sha256),
+    Sha1(Sha1),
+}
+
+impl BlobHasher {
+    /// A hasher initialized with the preamble for an artifact of `len`
+    /// bytes — the known-length path's one-pass form (the preamble is
+    /// known before content flows).
+    pub(crate) fn new(algorithm: Algorithm, len: u64) -> BlobHasher {
+        let mut hasher = BlobHasher {
+            algorithm,
+            inner: match algorithm {
+                Algorithm::GitBlobSha256 => InnerHasher::Sha256(Sha256::new()),
+                Algorithm::GitBlobSha1 => InnerHasher::Sha1(Sha1::new()),
+            },
+        };
+        hasher.update(&blob_preamble(len));
+        hasher
+    }
+
+    /// Feed the next chunk of content.
+    pub(crate) fn update(&mut self, chunk: &[u8]) {
+        match &mut self.inner {
+            InnerHasher::Sha256(inner) => inner.update(chunk),
+            InnerHasher::Sha1(inner) => inner.update(chunk),
+        }
+    }
+
+    /// The derivation's answer over preamble + everything fed so far —
+    /// pinned equal to [`Algorithm::derive`] by the in-module tests.
+    pub(crate) fn finish(self) -> Digest {
+        let bytes = match self.inner {
+            InnerHasher::Sha256(inner) => inner.finalize().to_vec(),
+            InnerHasher::Sha1(inner) => inner.finalize().to_vec(),
+        };
+        Digest::from_parts(self.algorithm, bytes)
+    }
+}
+
 /// An algorithm + digest pair — the hash of a blob under one of this
 /// crate's derivations (ADR-002; hashing-and-keys.md §The
 /// Key/Digest surface).
@@ -294,6 +349,54 @@ mod tests {
         );
     }
 
+    /// The streaming form is the same derivation: a `BlobHasher` fed
+    /// the content in arbitrary chunk boundaries finishes byte-identical
+    /// to [`Algorithm::derive`] — both algorithms, boundaries around the
+    /// known-answer vectors, and a multi-chunk bulk body.
+    #[test]
+    fn streaming_hasher_matches_known_length_derivation() {
+        for algorithm in [Algorithm::GitBlobSha256, Algorithm::GitBlobSha1] {
+            let bodies: [&[u8]; 4] = [b"", b"hello", &[0u8; 100_003], &[7u8; 9]];
+
+            for body in bodies {
+                let expected = algorithm.derive(body, body.len() as u64);
+
+                // whole-body update.
+                let mut hasher = BlobHasher::new(algorithm, body.len() as u64);
+                hasher.update(body);
+                assert_eq!(hasher.finish(), expected, "whole-body, {algorithm:?}");
+
+                // byte-at-a-time.
+                let mut hasher = BlobHasher::new(algorithm, body.len() as u64);
+                for byte in body {
+                    hasher.update(std::slice::from_ref(byte));
+                }
+                assert_eq!(hasher.finish(), expected, "byte-at-a-time, {algorithm:?}");
+
+                // awkward fixed chunks (7, including past the end).
+                let mut hasher = BlobHasher::new(algorithm, body.len() as u64);
+                for chunk in body.chunks(7) {
+                    hasher.update(chunk);
+                }
+                assert_eq!(hasher.finish(), expected, "chunks of 7, {algorithm:?}");
+            }
+
+            // The canonical 'hello' and empty vectors, streamed.
+            let expected = algorithm.derive(b"hello", 5).bytes().to_vec();
+            let mut hasher = BlobHasher::new(algorithm, 5);
+            hasher.update(b"hel");
+            hasher.update(b"lo");
+            assert_eq!(hasher.finish().bytes().to_vec(), expected);
+
+            let hasher = BlobHasher::new(algorithm, 0);
+            assert_eq!(
+                hasher.finish().bytes().to_vec(),
+                algorithm.derive(b"", 0).bytes().to_vec(),
+                "empty content streams with no update"
+            );
+        }
+    }
+
     /// Algorithm ordering equals registry-byte ordering, so
     /// digest-derived sort is the keys' byte total order.
     #[test]

src/local.rs

@@ -266,16 +266,18 @@ impl ReadCursor for LocalCursor {
 /// The staged put's staging cursor (ADR-010 §1): backed by a stage file
 /// in the staging area, unique per stage — appends write through, the
 /// staged length tracks completed appends (the sole-writer protocol),
-/// and `commit` renames the stage file into its sharded committed path —
-/// publish is the rename. Every failure path converges on discard with
-/// zero residue: a commit that cannot complete (media failure, CAS
-/// dedup, parent creation) removes the stage file itself; [`StagingCursor::discard`]
-/// removes it; and a cursor dropped without either is best-effort
-/// removed by the [`Drop`] body — the engine's cancellation story,
-/// covered by the in-module hygiene tests.
+/// and `commit(key)` renames the stage file into its sharded committed
+/// path under the store-core-resolved canonical key (the staged-put
+/// seam's key-at-commit design point: the derivation completes at stream
+/// end, so until commit the stage's identity is the engine-internal
+/// stage file alone) — publish is the rename. Every failure path
+/// converges on discard with zero residue: a commit that cannot complete
+/// (media failure, CAS dedup, parent creation) removes the stage file
+/// itself; [`StagingCursor::discard`] removes it; and a cursor dropped
+/// without either is best-effort removed by the [`Drop`] body — the
+/// engine's cancellation story, covered by the in-module hygiene tests.
 pub(crate) struct LocalStaging {
     root: Arc<PathBuf>,
-    key: Vec<u8>,
     stage_path: Option<PathBuf>,
     file: File,
     staged_len: AtomicU64,
@@ -321,7 +323,7 @@ impl StagingCursor for LocalStaging {
         Ok(())
     }
 
-    async fn commit(mut self) -> Result<(), Error> {
+    async fn commit(mut self, key: &[u8]) -> Result<(), Error> {
         let stage = match self.stage_path.take() {
             Some(stage) => stage,
             None => {
@@ -329,7 +331,7 @@ impl StagingCursor for LocalStaging {
             }
         };
         let root = Arc::clone(&self.root);
-        let key = std::mem::take(&mut self.key);
+        let key = key.to_vec();
         let cleanup = stage.clone();
         let result = blocking(move || {
             let committed = committed_path(&root, &key)?;
@@ -454,10 +456,8 @@ impl Backend for LocalBackend {
         .await
     }
 
-    async fn put_staged(&self, key: &[u8]) -> Result<Self::Staging, Error> {
-        let root = Arc::clone(&self.root);
-        let key = key.to_vec();
-        let inner_root = Arc::clone(&root);
+    async fn put_staged(&self) -> Result<Self::Staging, Error> {
+        let inner_root = Arc::clone(&self.root);
         let name = format!("stage-{}", stage_id());
         #[cfg(test)]
         let fail_commit = self.fail_commit;
@@ -473,7 +473,6 @@ impl Backend for LocalBackend {
                 .map_err(io_err)?;
             Ok(LocalStaging {
                 root: inner_root,
-                key,
                 stage_path: Some(stage_path),
                 file,
                 staged_len: AtomicU64::new(0),
@@ -590,9 +589,9 @@ mod tests {
     }
 
     async fn staged(backend: &LocalBackend, key: &[u8], content: &[u8]) -> Result<(), Error> {
-        let mut staging = backend.put_staged(key).await?;
+        let mut staging = backend.put_staged().await?;
         staging.append(content).await?;
-        staging.commit().await
+        staging.commit(key).await
     }
 
     /// The Backend method contract: the name, the seven methods'
@@ -770,10 +769,13 @@ mod tests {
         assert_eq!(cursor.read_at(0, 8).await.unwrap(), Vec::<u8>::new());
 
         let staged_key = key_for(b"empty-staged");
-        let mut staging = backend.put_staged(&staged_key).await.unwrap();
+        let mut staging = backend.put_staged().await.unwrap();
         staging.append(b"").await.unwrap();
         assert_eq!(staging.len(), 0, "empty append stays empty");
-        staging.commit().await.expect("empty commit publishes");
+        staging
+            .commit(&staged_key)
+            .await
+            .expect("empty commit publishes");
         assert!(
             backend.has(&staged_key).await.unwrap(),
             "empty entry landed"
@@ -800,7 +802,7 @@ mod tests {
         let (_root, backend) = new_backend().await;
         let key = key_for(b"staged-round-trip");
 
-        let mut staging = backend.put_staged(&key).await.unwrap();
+        let mut staging = backend.put_staged().await.unwrap();
         staging.append(&CONTENT[..8]).await.unwrap();
         staging.append(&CONTENT[8..24]).await.unwrap();
         staging.append(&CONTENT[24..]).await.unwrap();
@@ -819,7 +821,7 @@ mod tests {
         let tail = staging.read_at(CONTENT.len() as u64, 32).await.unwrap();
         assert_eq!(tail, Vec::<u8>::new(), "staged read clamps at staged EOF");
 
-        staging.commit().await.expect("commit publishes");
+        staging.commit(&key).await.expect("commit publishes");
         assert!(backend.has(&key).await.unwrap(), "entry published");
 
         let cursor = backend
@@ -857,7 +859,7 @@ mod tests {
         let (root, backend) = new_backend().await;
         let key = key_for(b"discard-key");
 
-        let mut staging = backend.put_staged(&key).await.unwrap();
+        let mut staging = backend.put_staged().await.unwrap();
         staging.append(b"residue").await.unwrap();
         staging.discard().await;
 
@@ -887,9 +889,9 @@ mod tests {
         };
         let key = key_for(b"failure-key");
 
-        let mut staging = backend.put_staged(&key).await.unwrap();
+        let mut staging = backend.put_staged().await.unwrap();
         staging.append(CONTENT).await.unwrap();
-        let err = staging.commit().await.unwrap_err();
+        let err = staging.commit(&key).await.unwrap_err();
         assert!(matches!(err, Error::Io(_)), "engine media failure");
 
         assert!(
@@ -913,7 +915,7 @@ mod tests {
         let key = key_for(b"cancelled-key");
 
         {
-            let mut staging = backend.put_staged(&key).await.unwrap();
+            let mut staging = backend.put_staged().await.unwrap();
             staging.append(b"never published").await.unwrap();
             // Dropped without commit or discard.
         }
@@ -953,21 +955,23 @@ mod tests {
         for i in 0..N {
             let content = format!("discard-{i}");
             let key = key_for(content.as_bytes());
-            let mut staging = backend.put_staged(&key).await.unwrap();
+            let _ = &key;
+            let mut staging = backend.put_staged().await.unwrap();
             staging.append(content.as_bytes()).await.unwrap();
             staging.discard().await;
         }
         for i in 0..N {
             let content = format!("commit-failure-{i}");
             let key = key_for(content.as_bytes());
-            let mut staging = failure_backend.put_staged(&key).await.unwrap();
+            let mut staging = failure_backend.put_staged().await.unwrap();
             staging.append(content.as_bytes()).await.unwrap();
-            staging.commit().await.unwrap_err();
+            staging.commit(&key).await.unwrap_err();
         }
         for i in 0..N {
             let content = format!("cancelled-{i}");
             let key = key_for(content.as_bytes());
-            let mut staging = backend.put_staged(&key).await.unwrap();
+            let _ = &key;
+            let mut staging = backend.put_staged().await.unwrap();
             staging.append(content.as_bytes()).await.unwrap();
             drop(staging);
         }

src/mem.rs

@@ -10,9 +10,12 @@
 //! form: [`get`][Backend::get] hands out a read cursor that materializes
 //! the whole value (in-memory slices; `len()` from the stored value), and
 //! [`put_staged`][Backend::put_staged] hands out the degenerate in-memory
-//! stage whose `commit` publishes through the same whole-value CAS row
-//! write as `put` — `entry().or_insert()`, first writer wins, a re-put is
-//! dedup and never an overwrite. `commit` cannot fail in normal operation
+//! stage whose `commit(key)` publishes through the same whole-value CAS
+//! row write as `put` — `entry().or_insert()`, first writer wins, a re-put
+//! is dedup and never an overwrite. The key resolves at commit (the
+//! staged-put seam's wave-2 amendment: the canonical derivation completes
+//! at stream end, so the stage's identity before that point is the
+//! engine-internal buffer alone). `commit` cannot fail in normal operation
 //! (no media can fail here); the test-only hook noted on [`MemStaging`]
 //! exercises the failure form's self-discard hygiene, standing in for the
 //! media-failure injection later engines need.
@@ -75,9 +78,8 @@ impl Backend for MemBackend {
         Ok(())
     }
 
-    async fn put_staged(&self, key: &[u8]) -> Result<Self::Staging, Error> {
+    async fn put_staged(&self) -> Result<Self::Staging, Error> {
         Ok(MemStaging {
-            key: key.to_vec(),
             staged: Vec::new(),
             rows: Arc::clone(&self.rows),
             #[cfg(test)]
@@ -122,14 +124,15 @@ impl ReadCursor for MemCursor {
 }
 
 /// The degenerate in-memory staging cursor (ADR-010 §1): appends extend
-/// the buffer; `commit` publishes the whole staged bytes through the same
-/// CAS clause as [`Backend::put`] — first writer wins, losing the race is
-/// dedup (`Ok`), never a failure — and every non-commit path, including
-/// the failure form, converges on zero residue. `commit` takes the
-/// buffer via `std::mem::take` before the table lock, so a failed publish
-/// cannot strand the staged bytes in the map.
+/// the buffer; `commit(key)` publishes the whole staged bytes through the
+/// same CAS clause as [`Backend::put`] — under the store-core-resolved
+/// canonical key (the staged-put seam's key-at-commit design point),
+/// first writer wins, losing the race is dedup (`Ok`), never a failure —
+/// and every non-commit path, including the failure form, converges on
+/// zero residue. `commit` takes the buffer via `std::mem::take` before
+/// the table lock, so a failed publish cannot strand the staged bytes
+/// in the map.
 pub(crate) struct MemStaging {
-    key: Vec<u8>,
     staged: Vec<u8>,
     rows: Arc<Mutex<BTreeMap<Vec<u8>, Vec<u8>>>>,
     /// Test-only injection, shared with the issuing [`MemBackend`] — see
@@ -155,7 +158,7 @@ impl StagingCursor for MemStaging {
         Ok(())
     }
 
-    async fn commit(mut self) -> Result<(), Error> {
+    async fn commit(mut self, key: &[u8]) -> Result<(), Error> {
         #[cfg(test)]
         if self.fail_commit {
             self.staged.clear();
@@ -164,7 +167,7 @@ impl StagingCursor for MemStaging {
             ));
         }
         let value = std::mem::take(&mut self.staged);
-        self.rows.lock().entry(self.key).or_insert(value);
+        self.rows.lock().entry(key.to_vec()).or_insert(value);
         Ok(())
     }
 
@@ -338,10 +341,13 @@ mod tests {
         assert_eq!(cursor.read_at(0, 8).await.unwrap(), Vec::<u8>::new());
 
         let staged_key = b"\x01empty-staged-key".to_vec();
-        let mut staging = backend.put_staged(&staged_key).await.unwrap();
+        let mut staging = backend.put_staged().await.unwrap();
         staging.append(b"").await.unwrap();
         assert_eq!(staging.len(), 0, "empty append stays empty");
-        staging.commit().await.expect("empty commit publishes");
+        staging
+            .commit(&staged_key)
+            .await
+            .expect("empty commit publishes");
         assert!(
             backend.has(&staged_key).await.unwrap(),
             "empty entry landed"
@@ -369,7 +375,7 @@ mod tests {
         let backend = MemBackend::default();
         let key = b"\x01staged-key".to_vec();
 
-        let mut staging = backend.put_staged(&key).await.unwrap();
+        let mut staging = backend.put_staged().await.unwrap();
         staging.append(&CONTENT[..8]).await.unwrap();
         staging.append(&CONTENT[8..16]).await.unwrap();
         staging.append(&CONTENT[16..]).await.unwrap();
@@ -386,7 +392,7 @@ mod tests {
         assert_eq!(bytes, CONTENT, "staged artifact readable as a whole");
         assert_eq!(digest, crate::backend::slice_digest(CONTENT));
 
-        staging.commit().await.expect("commit publishes");
+        staging.commit(&key).await.expect("commit publishes");
         assert!(backend.has(&key).await.unwrap(), "entry published");
 
         let cursor = backend
@@ -410,9 +416,12 @@ mod tests {
             Some(CONTENT),
             "whole-value re-put does not overwrite"
         );
-        let mut second = backend.put_staged(&key).await.unwrap();
+        let mut second = backend.put_staged().await.unwrap();
         second.append(b"overwritten-by-stage").await.unwrap();
-        second.commit().await.expect("commit is dedup, not failure");
+        second
+            .commit(&key)
+            .await
+            .expect("commit is dedup, not failure");
         assert_eq!(
             stored_value(&backend, &key).await.as_deref(),
             Some(CONTENT),
@@ -427,7 +436,7 @@ mod tests {
         let backend = MemBackend::default();
         let key = b"\x01discard-key".to_vec();
 
-        let mut staging = backend.put_staged(&key).await.unwrap();
+        let mut staging = backend.put_staged().await.unwrap();
         staging.append(b"residue").await.unwrap();
         staging.discard().await;
 
@@ -455,9 +464,9 @@ mod tests {
         };
         let key = b"\x01failure-key".to_vec();
 
-        let mut staging = backend.put_staged(&key).await.unwrap();
+        let mut staging = backend.put_staged().await.unwrap();
         staging.append(CONTENT).await.unwrap();
-        let err = staging.commit().await.unwrap_err();
+        let err = staging.commit(&key).await.unwrap_err();
         assert!(matches!(err, Error::Io(_)), "engine media failure");
 
         assert!(