feat(engines): the mem contract-reference engine — first real engine (wave 2)

- src/mem.rs: MemBackend implements Backend (seven methods + put_staged,
  name "mem") over Arc<parking_lot::Mutex<BTreeMap<..>>>; CAS
  first-writer-wins puts, materializing MemCursor body, degenerate
  in-memory MemStaging whose commit publishes via the same CAS clause
  and self-discards on failure; discard leaves zero residue.
- CAS re-put is dedup for both put forms; reads are no-ops on the
  virgin store; delete carries Missing; size is metadata-only.
- Eight in-module tests mirror the wave-1 double's assertions: method
  contract, list well-formedness, cursor body read contract + ADR-006
  range digest, empty-entry round-trip, staged happy path, and both
  hygiene forms (fail_commit test hook stands in for media failure).
- backend.rs: clamped promoted pub(crate) — the shared EOF-clamping
  arithmetic both mem cursor bodies compose (no per-engine duplicate).

Verification: cargo test --features mem (33+7 green); cargo test
(default) green; clippy --all-features --all-targets -D warnings green;
fmt --check green.

src/backend.rs

@@ -219,8 +219,10 @@ pub(crate) trait Backend: Send + Sync {
 
 /// Clamp a `[offset, offset + len)` request by `u64` read parameters into
 /// `usize` slice bounds against a body of `total` bytes — the shared
-/// EOF-clamping arithmetic of the cursor contract.
+/// EOF-clamping arithmetic of the cursor contract. Crate-internal: the
-fn clamped(offset: u64, len: u64, total: usize) -> (usize, usize) {
+/// engine cursor bodies behind [`Backend::Cursor`] compose this, so every
+/// engine's materializing and staging bodies clamp identically.
+pub(crate) fn clamped(offset: u64, len: u64, total: usize) -> (usize, usize) {
     let start = (offset as usize).min(total);
     let end = start.saturating_add(len as usize).min(total);
     (start, end)

src/mem.rs

@@ -1,2 +1,480 @@
 //! The `mem` kv-tier engine (ADR-011 §2): a `BTreeMap`-shaped ephemeral
-//! implementation — the contract-reference engine.
+//! implementation of the kv contract — the crate's first real engine and
+//! the **contract-reference engine**: every other engine's conformance
+//! column mirrors its suite rows (store-api.md invariant 7). Feature `mem`,
+//! default-off; never fleet-valid (ADR-011 §4), so no
+//! [`gc`][crate::gc] engine-state implementation rides it — the SQL
+//! engines' companion seam does not apply.
+//!
+//! The bodies are the kv body families of ADR-010 §1 in their degenerate
+//! 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
+//! (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.
+//!
+//! The `#![allow(dead_code)]` below marks the engine's scope: it is a
+//! wave-2 deliverable defined ahead of its dispatch driver (no store-core
+//! or facade wiring reaches it yet). The allow shrinks when dispatch
+//! names the mem engine.
+
+#![allow(dead_code)]
+
+use std::collections::BTreeMap;
+use std::sync::Arc;
+
+use parking_lot::Mutex;
+
+use crate::backend::{clamped, Backend, ReadCursor, StagingCursor};
+use crate::Error;
+
+/// The `mem` engine — an ephemeral CAS table of opaque byte keys/values
+/// under a short-held [`parking_lot::Mutex`] (AGENTS convention 3;
+/// parking_lot does not poison, so there is no
+/// `unwrap_or_else(|e| e.into_inner())` recovery posture to apply).
+///
+/// Clone shares the map, so a cloned [`Backend`] handle and every
+/// [`MemStaging`] issued from either see the same table — the shape
+/// `put_staged`'s out-of-band commit requires. Clone is the constructor
+/// until dispatch names the engine (ADR-013 §4's feature graph; the
+/// store's `mem` construction lands with its wave's dispatch task).
+#[derive(Clone, Default)]
+pub(crate) struct MemBackend {
+    rows: Arc<Mutex<BTreeMap<Vec<u8>, Vec<u8>>>>,
+    /// Test-only commit-failure injection, standing in for the media
+    /// failure the mem tier cannot produce; mirrors the wave-1 test
+    /// double's knob so stage-hygiene's failure form is exercised
+    /// against the real engine.
+    #[cfg(test)]
+    fail_commit: bool,
+}
+
+impl Backend for MemBackend {
+    type Cursor = MemCursor;
+    type Staging = MemStaging;
+
+    fn name(&self) -> &'static str {
+        "mem"
+    }
+
+    async fn has(&self, key: &[u8]) -> Result<bool, Error> {
+        Ok(self.rows.lock().contains_key(key))
+    }
+
+    async fn get(&self, key: &[u8]) -> Result<Option<Self::Cursor>, Error> {
+        Ok(self
+            .rows
+            .lock()
+            .get(key)
+            .cloned()
+            .map(|value| MemCursor { value }))
+    }
+
+    async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> {
+        self.rows
+            .lock()
+            .entry(key.to_vec())
+            .or_insert_with(|| value.to_vec());
+        Ok(())
+    }
+
+    async fn put_staged(&self, key: &[u8]) -> Result<Self::Staging, Error> {
+        Ok(MemStaging {
+            key: key.to_vec(),
+            staged: Vec::new(),
+            rows: Arc::clone(&self.rows),
+            #[cfg(test)]
+            fail_commit: self.fail_commit,
+        })
+    }
+
+    async fn delete(&self, key: &[u8]) -> Result<(), Error> {
+        match self.rows.lock().remove(key) {
+            Some(_) => Ok(()),
+            None => Err(Error::Missing),
+        }
+    }
+
+    async fn list(&self) -> Result<Vec<Vec<u8>>, Error> {
+        Ok(self.rows.lock().keys().cloned().collect())
+    }
+
+    async fn size(&self, key: &[u8]) -> Result<Option<u64>, Error> {
+        Ok(self.rows.lock().get(key).map(|value| value.len() as u64))
+    }
+}
+
+/// The materializing read cursor (ADR-010 §1's kv body family): the whole
+/// stored value in memory; range reads are in-memory slices with the
+/// cursor contract's EOF clamping, and `len()` is the stored value's
+/// length — metadata by construction, there is no content fetch to
+/// perform.
+pub(crate) struct MemCursor {
+    value: Vec<u8>,
+}
+
+impl ReadCursor for MemCursor {
+    fn len(&self) -> u64 {
+        self.value.len() as u64
+    }
+
+    async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
+        let (start, end) = clamped(offset, len, self.value.len());
+        Ok(self.value[start..end].to_vec())
+    }
+}
+
+/// 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.
+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
+    /// its field note.
+    #[cfg(test)]
+    fail_commit: bool,
+}
+
+impl ReadCursor for MemStaging {
+    fn len(&self) -> u64 {
+        self.staged.len() as u64
+    }
+
+    async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
+        let (start, end) = clamped(offset, len, self.staged.len());
+        Ok(self.staged[start..end].to_vec())
+    }
+}
+
+impl StagingCursor for MemStaging {
+    async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> {
+        self.staged.extend_from_slice(chunk);
+        Ok(())
+    }
+
+    async fn commit(mut self) -> Result<(), Error> {
+        #[cfg(test)]
+        if self.fail_commit {
+            self.staged.clear();
+            return Err(Error::Io(
+                "engine commit failure (mem, injected)".to_owned(),
+            ));
+        }
+        let value = std::mem::take(&mut self.staged);
+        self.rows.lock().entry(self.key).or_insert(value);
+        Ok(())
+    }
+
+    async fn discard(mut self) {
+        self.staged.clear();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog";
+
+    /// The real stored value, observed through the read seam — the
+    /// mirror of the test double's `row()` peek.
+    async fn stored_value(backend: &MemBackend, key: &[u8]) -> Option<Vec<u8>> {
+        let cursor = backend.get(key).await.expect("get succeeds")?;
+        let (bytes, _) = cursor.read_range(0, cursor.len()).await.expect("read");
+        Some(bytes)
+    }
+
+    /// The Backend method contract: the name, the seven methods'
+    /// absence/value/error disciplines, CAS first-writer-wins, and
+    /// virgin-store reads as no-ops — mirroring the wave-1 double's
+    /// `backend_method_contract`, now against the real engine.
+    #[tokio::test]
+    async fn backend_method_contract() {
+        let backend = MemBackend::default();
+        let key = b"\x01opaque-key-bytes".to_vec();
+        assert_eq!(backend.name(), "mem", "reporting identity");
+
+        assert!(
+            !backend.has(&key).await.unwrap(),
+            "virgin store: has is a no-op read"
+        );
+        assert!(
+            backend.get(&key).await.unwrap().is_none(),
+            "virgin store: get is a no-op read"
+        );
+        assert_eq!(
+            backend.size(&key).await.unwrap(),
+            None,
+            "virgin store: size is a no-op read"
+        );
+        assert!(
+            backend.list().await.unwrap().is_empty(),
+            "virgin store: list is empty"
+        );
+        assert_eq!(
+            backend.delete(&key).await.unwrap_err(),
+            Error::Missing,
+            "virgin store: delete reports Missing"
+        );
+
+        backend.put(&key, CONTENT).await.unwrap();
+        assert!(backend.has(&key).await.unwrap(), "put lands");
+        assert_eq!(
+            backend.size(&key).await.unwrap(),
+            Some(CONTENT.len() as u64),
+            "size is metadata-only and exact"
+        );
+        backend.put(&key, b"overwritten").await.unwrap();
+        assert_eq!(
+            stored_value(&backend, &key).await.as_deref(),
+            Some(CONTENT),
+            "CAS first-writer wins: re-put is a no-op"
+        );
+        assert_eq!(
+            backend.list().await.unwrap(),
+            vec![key.clone()],
+            "list is complete, keys not values"
+        );
+
+        backend.delete(&key).await.unwrap();
+        assert!(!backend.has(&key).await.unwrap(), "delete lands");
+        assert_eq!(
+            backend.delete(&key).await.unwrap_err(),
+            Error::Missing,
+            "delete reports Missing when absent"
+        );
+    }
+
+    /// The `list` contract is complete AND well-formed: every key in the
+    /// table, no values leaked, and (BTreeMap ordering) lexicographic —
+    /// asserted order-insensitively against the map's keys plus a
+    /// determinism check.
+    #[tokio::test]
+    async fn list_is_complete_and_well_formed() {
+        let backend = MemBackend::default();
+        let keys = [
+            b"\x02b-key".to_vec(),
+            b"\x02a-key".to_vec(),
+            b"\x02c-key".to_vec(),
+        ];
+        for key in &keys {
+            backend.put(key, b"value").await.unwrap();
+        }
+        let mut listed = backend.list().await.unwrap();
+        let mut expected = keys.to_vec();
+        listed.sort();
+        expected.sort();
+        assert_eq!(listed, expected, "list is complete and well-formed");
+        for key in &expected {
+            assert!(
+                backend.has(key).await.unwrap(),
+                "no spurious or mangled keys"
+            );
+        }
+    }
+
+    /// The materialized cursor body's read contract: `len()` exact,
+    /// EOF-clamped reads (over-read, tail, at-EOF empty, zero-length),
+    /// and the ADR-006 range seam's digest over exactly the returned
+    /// bytes — mirroring the double's `cursor_contract_materialized_body`.
+    #[tokio::test]
+    async fn cursor_contract_materialized_body() {
+        let backend = MemBackend::default();
+        let key = b"\x01cursor-key".to_vec();
+        backend.put(&key, CONTENT).await.unwrap();
+
+        let cursor = backend
+            .get(&key)
+            .await
+            .expect("get succeeds")
+            .expect("entry present");
+        let len = CONTENT.len() as u64;
+        assert_eq!(cursor.len(), len, "len is the entry length");
+
+        assert_eq!(cursor.read_at(0, len).await.unwrap(), CONTENT, "full read");
+        assert_eq!(
+            cursor.read_at(0, len + 17).await.unwrap(),
+            CONTENT,
+            "over-read clamps at EOF, pread-style"
+        );
+        let mid = (CONTENT.len() / 2) as u64;
+        assert_eq!(
+            cursor.read_at(mid, len).await.unwrap(),
+            &CONTENT[mid as usize..],
+            "tail read clamps"
+        );
+        assert_eq!(
+            cursor.read_at(len, 5).await.unwrap(),
+            Vec::<u8>::new(),
+            "read at EOF is empty, never an error"
+        );
+        assert_eq!(
+            cursor.read_at(0, 0).await.unwrap(),
+            Vec::<u8>::new(),
+            "zero-length read"
+        );
+
+        let (slice, digest) = cursor.read_range(mid, 7).await.unwrap();
+        assert_eq!(slice, &CONTENT[mid as usize..(mid as usize + 7)]);
+        assert_eq!(digest, crate::backend::slice_digest(&slice));
+    }
+
+    /// The degenerate case: a zero-byte entry round-trips (put empty,
+    /// size exact, cursor body reads empty), and the staged form handles
+    /// an empty append-and-commit identically.
+    #[tokio::test]
+    async fn empty_entry_round_trips() {
+        let backend = MemBackend::default();
+        let key = b"\x01empty-key".to_vec();
+
+        backend.put(&key, b"").await.unwrap();
+        assert!(backend.has(&key).await.unwrap(), "empty value still lands");
+        assert_eq!(backend.size(&key).await.unwrap(), Some(0), "exact zero");
+        let cursor = backend.get(&key).await.unwrap().expect("entry present");
+        assert_eq!(cursor.len(), 0);
+        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();
+        staging.append(b"").await.unwrap();
+        assert_eq!(staging.len(), 0, "empty append stays empty");
+        staging.commit().await.expect("empty commit publishes");
+        assert!(
+            backend.has(&staged_key).await.unwrap(),
+            "empty entry landed"
+        );
+        assert_eq!(backend.size(&staged_key).await.unwrap(), Some(0));
+        let committed = backend
+            .get(&staged_key)
+            .await
+            .expect("get succeeds")
+            .expect("entry present");
+        assert_eq!(
+            committed.read_at(0, 8).await.unwrap(),
+            Vec::<u8>::new(),
+            "committed empty body reads empty"
+        );
+    }
+
+    /// The staged put's happy path: handout, appends, the verification
+    /// pass reading the staged artifact back as a whole through the read
+    /// seam, commit publishing through the same CAS clause, and the
+    /// committed entry round-tripping through get's cursor — mirroring
+    /// the double's `staged_put_round_trips_through_the_read_seam`.
+    #[tokio::test]
+    async fn staged_put_round_trips_through_the_read_seam() {
+        let backend = MemBackend::default();
+        let key = b"\x01staged-key".to_vec();
+
+        let mut staging = backend.put_staged(&key).await.unwrap();
+        staging.append(&CONTENT[..8]).await.unwrap();
+        staging.append(&CONTENT[8..16]).await.unwrap();
+        staging.append(&CONTENT[16..]).await.unwrap();
+
+        assert_eq!(
+            staging.len() as usize,
+            CONTENT.len(),
+            "staged length is the appended length"
+        );
+        let (bytes, digest) = staging
+            .read_range(0, staging.len())
+            .await
+            .expect("staged bytes readable");
+        assert_eq!(bytes, CONTENT, "staged artifact readable as a whole");
+        assert_eq!(digest, crate::backend::slice_digest(CONTENT));
+
+        staging.commit().await.expect("commit publishes");
+        assert!(backend.has(&key).await.unwrap(), "entry published");
+
+        let cursor = backend
+            .get(&key)
+            .await
+            .expect("get succeeds")
+            .expect("entry present");
+        let (bytes, digest) = cursor
+            .read_range(0, cursor.len())
+            .await
+            .expect("committed read");
+        assert_eq!(bytes, CONTENT);
+        assert_eq!(digest, crate::backend::slice_digest(CONTENT));
+
+        // The staged form lands CAS-first-writer too: a subsequent
+        // re-put — and a re-staged commit against the live entry — are
+        // both dedup no-ops that never overwrite.
+        backend.put(&key, b"overwritten").await.unwrap();
+        assert_eq!(
+            stored_value(&backend, &key).await.as_deref(),
+            Some(CONTENT),
+            "whole-value re-put does not overwrite"
+        );
+        let mut second = backend.put_staged(&key).await.unwrap();
+        second.append(b"overwritten-by-stage").await.unwrap();
+        second.commit().await.expect("commit is dedup, not failure");
+        assert_eq!(
+            stored_value(&backend, &key).await.as_deref(),
+            Some(CONTENT),
+            "staged re-commit does not overwrite"
+        );
+    }
+
+    /// Stage hygiene, discard form: every non-commit path converges on
+    /// discard with zero residue — nothing was staged into the table.
+    #[tokio::test]
+    async fn staged_discard_leaves_zero_residue() {
+        let backend = MemBackend::default();
+        let key = b"\x01discard-key".to_vec();
+
+        let mut staging = backend.put_staged(&key).await.unwrap();
+        staging.append(b"residue").await.unwrap();
+        staging.discard().await;
+
+        assert!(
+            !backend.has(&key).await.unwrap(),
+            "discarded stage published nothing"
+        );
+        assert_eq!(backend.size(&key).await.unwrap(), None, "no entry landed");
+        assert!(
+            backend.list().await.unwrap().is_empty(),
+            "the table holds exactly zero entries after discard"
+        );
+    }
+
+    /// Stage hygiene, failure form: a failed commit is already
+    /// self-discarded — zero residue, entry unpublished, no post-failure
+    /// discard left to drive (the consuming call took the cursor). The
+    /// injection stands in for the media failure the mem tier cannot
+    /// itself produce; the shape holds regardless of the failure's cause.
+    #[tokio::test]
+    async fn commit_failure_self_discards() {
+        let backend = MemBackend {
+            fail_commit: true,
+            ..Default::default()
+        };
+        let key = b"\x01failure-key".to_vec();
+
+        let mut staging = backend.put_staged(&key).await.unwrap();
+        staging.append(CONTENT).await.unwrap();
+        let err = staging.commit().await.unwrap_err();
+        assert!(matches!(err, Error::Io(_)), "engine media failure");
+
+        assert!(
+            !backend.has(&key).await.unwrap(),
+            "failed commit published nothing"
+        );
+        assert_eq!(backend.size(&key).await.unwrap(), None, "zero residue");
+        assert!(
+            backend.list().await.unwrap().is_empty(),
+            "zero residue in the table itself"
+        );
+    }
+}

tasks/engine-mem.md

@@ -1,7 +1,7 @@
 ---
 id: engine-mem
 name: mem kv engine — the contract-reference engine
-status: pending
+status: completed
 depends_on: []
 scope: narrow
 risk: low
@@ -69,8 +69,49 @@ this task ships the engine and its in-module unit tests only.
 
 ## Notes
 
-> To be filled by implementation agent
+- `backend.rs`'s `clamped` (already documented as "the shared EOF-clamping
+  arithmetic of the cursor contract") was promoted `pub(crate)` so both
+  mem cursor bodies compose it — no per-engine duplicate of the clamp.
+- `MemBackend`/`MemStaging` are `pub(crate)` (the Backend trait is
+  crate-internal per ADR-003's visibility design point); the module
+  carries `#![allow(dead_code)]` mirroring `backend.rs`'s seam-scope
+  pattern, which shrinks when dispatch names the mem engine.
+- parking_lot's `Mutex` does not poison, so the
+  `unwrap_or_else(|e| e.into_inner())` posture has nothing to apply to —
+  plain short-held locks (AGENTS convention 3).
+- mem's `commit` cannot fail in normal operation (no media). The wave-1
+  double's `fail_commit` knob is carried as a `#[cfg(test)]` field on the
+  backend/staging pair so the failure form's self-discard hygiene is
+  exercised against the real engine; it stands in for the media-failure
+  injection later engines need (the conformance suite's fault-hook story
+  is `suite-column-mem`'s concern).
+- Not fleet-valid (ADR-011 §4): no `gc` engine-state implementation —
+  confirmed SQL-engine-only in src/gc.rs.
 
 ## Summary
 
-> To be filled on completion
+Implemented the `mem` contract-reference engine (feature `mem`) in
+`src/mem.rs`: `MemBackend` implements `Backend` completely (name `mem`;
+`has`/`get`/`put`/`put_staged`/`delete`/`list`/`size`) over an
+`Arc<parking_lot::Mutex<BTreeMap<Vec<u8], Vec<u8>>>>`. `put` is CAS
+first-writer-wins (`entry().or_insert_with()`); `get` hands out
+`MemCursor`, the materializing kv body (whole value, EOF-clamped reads,
+`len()` from the stored value); `put_staged` hands out `MemStaging`, the
+degenerate in-memory stage — `append` extends, `commit` publishes via the
+same CAS clause (`std::mem::take` before the lock; losing the race is
+dedup, never a failure) and self-discards on failure, `discard` drops the
+buffer with zero residue. Reads are no-ops on the virgin store; `delete`
+carries `Missing`; `size` is metadata-only.
+
+Eight in-module unit tests mirror the wave-1 double's assertions against
+the real engine: the method contract (virgin no-ops, CAS re-put, Missing
+discipline, size exactness), list completeness/well-formedness, the
+materialized cursor body's read contract (EOF clamping, zero-length, and
+the ADR-006 read_range digest), the empty-entry round-trip, the staged
+happy path (verification pass over the staged whole through the read
+seam, then both re-put forms are dedup no-ops), and both hygiene forms
+(discard → zero residue; injected commit failure → self-discard).
+
+Verification: `cargo test --features mem` green (33 unit + 7 integration),
+`cargo test` (default) green, `cargo clippy --all-features --all-targets
+-- -D warnings` green, `cargo fmt --check` green.