mem-engine-seam-tx — the wave's load-bearing seam lands: the complete Store/TxHandle trait surfaces over the guarded state core, with the tx overlay as the discipline centerpiece. Wider than the sqlite/pg seam twins by design (those rode pre-ported substrate ops and left mechanism methods stubbed; mem has no substrate, and the task's acceptance pins the full trait surfaces implemented over guarded state — a Rust trait impl is all-or-nothing per method): the seam task therefore realizes the mechanisms' committed-state effect layer and every mechanism handle itself, and the wave-6 mechanism tasks now own their acceptance-criteria unit pins against this foundation. Store surface: notify (codec-typed encode as the entry check only — the wake delivers the channel alone; PayloadTooLarge never produced, the ADR-016 §5 non-occurrence arm pinned at 2 MiB), listen (WakeReceiver bridge with the pinned recv forms; attach starts from now, one close path at engine drop), stream/queue/outbox/try_lock (validated constructors → boxed core-owned handle traits: the ADR-010 queue state machine with claims/reclaim-eats-attempt/uniform validity predicate/dead-letter/sweep + mem's third-owner equal-jitter curve — std-only RandomState hash jitter, no rand; ADR-015 log+offsets+trim+wake-driven subscribe; ADR-008 §7 TTL registry with renew-past-expiry refused; ADR-014 outbox helper with run_once ack/retry), schedule/unschedule/run_schedules (ADR-009 collapse: @every-only third-owner parser + register-table with pre-parsed interval and resolved stamps; the 64-cap bounded catch-up + grid-aligned skip-forward tick fired-and-advanced under one guarded op (schedules→queues lock order); leadership through the reserved __alkstore_scheduler lock with runner-unique owner, renew-before-tick, keep-awake sleep renewing on cadence, LeadershipLost / clean-stop-Ok returns). Tx overlay: begin_tx allocates the handle-private overlay; all eleven *_tx stage with entry-point validation before any effect, Codec-typed encode, one clock read, ADR-020 §3 stamp resolution (plain 300/3/5/none, outbox 60/5/5, the per-job max_attempts override via resolution.rs's stamps_with_override); tx reads merge overlay over committed (read-your-own-writes: queue-scoped + dead-visible get_job_tx, monotone staged saves, offset-ASC merged reads, extent guards clamping empty); commit consumes the handle, merges under one guarded op (queues→streams, the canonical lock order documented in state.rs), and fires the overlay's pending wakes AFTER the merge — wake-at-commit structural from birth (the pg-fix-tx-wake failure shape cannot exist here); drop (explicit, or through an unwinding panic — the overlay is plain local data, nothing runs on discard) = rollback with no ghosts, trivially. Stage-time id/offset allocation (the pg-shaped story — rolled-back allocations gap out; claims order id ASC, offsets immutable/never renumbered). Stream subscribers ride a reserved-prefix engine-internal wake channel (__alkstore_stream_wake:{stream}) unreachable by consumer entry points, attached before the stored-offset read; notify's transport stores nothing. 89 mem unit tests: the overlay discipline rows the task pins, the numeric-domain rows (extents/durations/boundaries/grammar/validation), and the mechanism smoke pins (the defect-class catch: ack/ack_batch originally deleted before checking the predicate — fixed; a refusal must leave the row exactly as it was). Manifest: serde_json = 1 joins (the trait signatures cross Value; both engine twins carry it; not a driver dep — wasm32 gate stays green). Task file: status completed, Notes (the whole-engine scope disposition + 11 decision-of-record entries incl. lock-renewal delta vs the sqlite substrate, ScheduleRow shape, runner lease mechanics, substrate predicate parity) + Summary filled. Verified: cargo build; workspace cargo test green (core 25, suite harness 3, mem 89, SQLite 191 lib + 25 suite, pg 124 lib + 25 suite + 9 schema); clippy --all-targets -D warnings clean; cargo fmt --check clean; cargo check --target wasm32-unknown-unknown -p alkstore-mem clean

alkstore-mem/src/stream.rs

@@ -0,0 +1,704 @@
+//! The stream mechanism's mem arm — the in-memory event log +
+//! per-consumer offsets (engine-mem.md §Mapping the contract: global
+//! FIFO by monotone offset; keyed publish rides the overlay).
+//!
+//! - Publishes append to the per-stream log under its section guard:
+//!   engine-assigned offsets (the engine-wide allocator — pg-bigserial
+//!   shaped, monotone, immutable, never renumbered; a fresh stream's
+//!   first event does not carry offset 1, the ADR-015 §4 note), keys
+//!   are carried metadata (a present key must be non-empty —
+//!   `InvalidName`, ADR-015 §1), and the commit boundary fires one
+//!   wake per event on the stream's engine-internal wake channel.
+//! - Cursor reads are `offset ASC` (global FIFO, ADR-015 §4) with the
+//!   extent guard at the trait-impl entry (`limit <= 0` → empty
+//!   `Vec`, ADR-023 §2).
+//! - Checkpoints are explicit, monotone saves (a save below the
+//!   stored checkpoint is a silent no-op — ADR-019 §6; an absent
+//!   consumer reads 0, ADR-019 §1).
+//! - `trim_to` deletes `offset <= horizon` exactly (a negative horizon
+//!   deletes nothing, idempotently — the boundary argument is total,
+//!   ADR-023 §2), **wakes nothing**, and surviving events keep their
+//!   offsets (gaps legal, ADR-015 §5). Reads from a trimmed region
+//!   resume at the horizon's first remaining row.
+//! - `subscribe` attaches the receiver's wake feed **before** the
+//!   stored-offset read (no commit can fall between — a buffered wake
+//!   for a commit already re-read costs an empty re-poll, never a
+//!   missed event), drains the log to the tail, then parks on the
+//!   feed: wake-driven re-read with no polling requirement; explicit
+//!   saves only. The one close path is engine drop — the bus close
+//!   disconnects every feed and `recv()` returns the terminal `None`
+//!   (ADR-021 §5).
+
+use std::collections::VecDeque;
+use std::sync::Arc;
+
+use serde_json::Value;
+
+use alkstore::{
+    BoxedFuture, Error, EventReceiver, Result, StreamEvent, StreamHandle, validate_local_name,
+};
+
+use crate::clock::Clock;
+use crate::state::Guard;
+use crate::state::{EventRow, State, StreamState};
+use crate::wakes::WakeBus;
+
+/// How many events a wake-driven re-read fetches per page while
+/// draining to the tail.
+const SUBSCRIBE_PAGE: i64 = 256;
+
+/// The engine-internal wake channel a stream's subscribers ride:
+/// engine-derived, under the reserved prefix — consumer entry points
+/// reject the prefix, so a consumer `listen` can never attach to it
+/// (and `notify` can never fire into it). The publish-side wake feed
+/// is plumbing, not contract surface.
+pub(crate) fn stream_wake_channel(stream: &str) -> String {
+    format!("{}stream_wake:{}", alkstore::RESERVED_PREFIX, stream)
+}
+
+/// Insert one committed event under the streams guard (the auto-commit
+/// publish and the commit-merge share this effect). Returns nothing —
+/// the caller owns the wake firing at its commit boundary.
+pub(crate) fn insert_event(s: &mut StreamState, stream: &str, row: EventRow) {
+    s.logs
+        .entry(stream.to_string())
+        .or_default()
+        .insert(row.offset, row);
+}
+
+/// Read `offset ASC`, `offset > cursor`, up to `limit` (the extent
+/// guard is the caller's — this helper is contract-blind to it).
+pub(crate) fn read_events(s: &StreamState, stream: &str, from: i64, limit: i64) -> Vec<EventRow> {
+    s.logs
+        .get(stream)
+        .map(|log| {
+            log.range((from + 1)..)
+                .take(limit.max(0) as usize)
+                .map(|(_, row)| row.clone())
+                .collect()
+        })
+        .unwrap_or_default()
+}
+
+/// The monotone checkpoint save (ADR-019 §6): a save at-or-below the
+/// stored checkpoint is a silent no-op; the direct form, the tx form,
+/// and the receiver's no-arg form all land here.
+pub(crate) fn save_offset_row(s: &mut StreamState, stream: &str, consumer: &str, offset: i64) {
+    let key = (stream.to_string(), consumer.to_string());
+    match s.offsets.get(&key) {
+        Some(current) if offset <= *current => {}
+        _ => {
+            s.offsets.insert(key, offset);
+        }
+    }
+}
+
+/// The stored checkpoint; absent consumer = 0.
+pub(crate) fn get_offset_row(s: &StreamState, stream: &str, consumer: &str) -> i64 {
+    s.offsets
+        .get(&(stream.to_string(), consumer.to_string()))
+        .copied()
+        .unwrap_or(0)
+}
+
+/// `trim_to`: delete `offset <= horizon` (exact boundary); surviving
+/// events keep their offsets. Returns the deleted count. Wakes nothing.
+pub(crate) fn trim_events(s: &mut StreamState, stream: &str, horizon: i64) -> i64 {
+    let Some(log) = s.logs.get_mut(stream) else {
+        return 0;
+    };
+    let horizon_next = horizon.saturating_add(1);
+    let removed: Vec<i64> = log.range(..horizon_next).map(|(k, _)| *k).collect();
+    let count = removed.len();
+    for offset in removed {
+        log.remove(&offset);
+    }
+    count as i64
+}
+
+fn event_from_row(row: EventRow, stream: &str) -> StreamEvent {
+    StreamEvent::from_row(
+        row.offset,
+        stream.to_string(),
+        row.key,
+        row.payload,
+        row.created_at,
+    )
+}
+
+/// The stream-scoped handle (see the module docs) for one validated
+/// stream name.
+pub struct MemStreamHandle {
+    pub(crate) name: String,
+    pub(crate) state: Arc<State>,
+    pub(crate) clock: Arc<Clock>,
+    pub(crate) wakes: Arc<WakeBus>,
+}
+
+impl std::fmt::Debug for MemStreamHandle {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("MemStreamHandle")
+            .field("name", &self.name)
+            .finish_non_exhaustive()
+    }
+}
+
+impl MemStreamHandle {
+    pub(crate) fn new(
+        name: String,
+        state: Arc<State>,
+        clock: Arc<Clock>,
+        wakes: Arc<WakeBus>,
+    ) -> Self {
+        Self {
+            name,
+            state,
+            clock,
+            wakes,
+        }
+    }
+
+    /// The keyed publish is the one publish shape (the plain form is
+    /// `key = None`).
+    fn stage_publish(&self, key: Option<String>, payload: Value) -> Result<i64> {
+        if let Some(k) = &key
+            && (k.is_empty() || k.trim().is_empty())
+        {
+            return Err(Error::InvalidName { name: k.clone() });
+        }
+        let bytes = alkstore::encode_payload(&payload)?;
+        let now = self.clock.unix_now();
+        let offset = self.state.allocate_event_offset();
+        let row = EventRow {
+            offset,
+            key,
+            payload: bytes,
+            created_at: now,
+        };
+        {
+            let mut s = self.state.streams.guard();
+            insert_event(&mut s, &self.name, row);
+        }
+        // The auto-commit publish's commit boundary is its own
+        // completion: one wake per event, per-subscriber FIFO.
+        let channel = stream_wake_channel(&self.name);
+        self.wakes.fire_commit_wakes([channel]);
+        Ok(offset)
+    }
+}
+
+impl StreamHandle for MemStreamHandle {
+    fn name(&self) -> &str {
+        &self.name
+    }
+
+    fn publish<'a>(&'a self, payload: Value) -> BoxedFuture<'a, Result<i64>> {
+        match self.stage_publish(None, payload) {
+            Ok(offset) => Box::pin(async move { Ok(offset) }),
+            Err(e) => Box::pin(async move { Err(e) }),
+        }
+    }
+
+    fn publish_with_key<'a>(
+        &'a self,
+        key: Option<String>,
+        payload: Value,
+    ) -> BoxedFuture<'a, Result<i64>> {
+        match self.stage_publish(key, payload) {
+            Ok(offset) => Box::pin(async move { Ok(offset) }),
+            Err(e) => Box::pin(async move { Err(e) }),
+        }
+    }
+
+    fn read_since<'a>(
+        &'a self,
+        offset: i64,
+        limit: i64,
+    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
+        if limit <= 0 {
+            return Box::pin(async { Ok(Vec::new()) });
+        }
+        let events = {
+            let s = self.state.streams.guard();
+            read_events(&s, &self.name, offset, limit)
+        };
+        let name = self.name.clone();
+        Box::pin(async move {
+            Ok(events
+                .into_iter()
+                .map(|r| event_from_row(r, &name))
+                .collect())
+        })
+    }
+
+    fn read_from_consumer<'a>(
+        &'a self,
+        consumer: &str,
+        limit: i64,
+    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
+        if let Err(e) = validate_local_name(consumer) {
+            return Box::pin(async move { Err(e) });
+        }
+        if limit <= 0 {
+            return Box::pin(async { Ok(Vec::new()) });
+        }
+        let name = self.name.clone();
+        let state = self.state.clone();
+        let consumer = consumer.to_string();
+        Box::pin(async move {
+            let events = {
+                let s = state.streams.guard();
+                let from = get_offset_row(&s, &name, &consumer);
+                read_events(&s, &name, from, limit)
+            };
+            Ok(events
+                .into_iter()
+                .map(|r| event_from_row(r, &name))
+                .collect())
+        })
+    }
+
+    fn save_offset<'a>(&'a self, consumer: &str, offset: i64) -> BoxedFuture<'a, Result<()>> {
+        if let Err(e) = validate_local_name(consumer) {
+            return Box::pin(async move { Err(e) });
+        }
+        let name = self.name.clone();
+        let state = self.state.clone();
+        let consumer = consumer.to_string();
+        Box::pin(async move {
+            let mut s = state.streams.guard();
+            save_offset_row(&mut s, &name, &consumer, offset);
+            Ok(())
+        })
+    }
+
+    fn get_offset<'a>(&'a self, consumer: &str) -> BoxedFuture<'a, Result<i64>> {
+        if let Err(e) = validate_local_name(consumer) {
+            return Box::pin(async move { Err(e) });
+        }
+        let name = self.name.clone();
+        let state = self.state.clone();
+        let consumer = consumer.to_string();
+        Box::pin(async move {
+            let s = state.streams.guard();
+            Ok(get_offset_row(&s, &name, &consumer))
+        })
+    }
+
+    fn trim_to<'a>(&'a self, horizon: i64) -> BoxedFuture<'a, Result<i64>> {
+        let name = self.name.clone();
+        let state = self.state.clone();
+        Box::pin(async move {
+            let mut s = state.streams.guard();
+            Ok(trim_events(&mut s, &name, horizon))
+        })
+    }
+
+    fn subscribe<'a>(&'a self, consumer: &str) -> BoxedFuture<'a, Result<Box<dyn EventReceiver>>> {
+        if let Err(e) = validate_local_name(consumer) {
+            return Box::pin(async move { Err(e) });
+        }
+        let name = self.name.clone();
+        let consumer = consumer.to_string();
+        let state = self.state.clone();
+        let wakes = self.wakes.clone();
+        Box::pin(async move {
+            // Attach first (starts from "now"; attach-after-commit
+            // no-replay is structural), then read the stored offset and
+            // drain to the tail.
+            let feed = wakes.attach(&stream_wake_channel(&name));
+            let mut cursor;
+            let mut pending = VecDeque::new();
+            {
+                let s = state.streams.guard();
+                cursor = get_offset_row(&s, &name, &consumer);
+                loop {
+                    let rows = read_events(&s, &name, cursor, SUBSCRIBE_PAGE);
+                    let got = rows.len();
+                    for row in rows {
+                        cursor = row.offset;
+                        pending.push_back(event_from_row(row, &name));
+                    }
+                    if got < SUBSCRIBE_PAGE as usize {
+                        break;
+                    }
+                }
+            }
+            Ok(Box::new(MemEventReceiver {
+                stream: name,
+                consumer,
+                state,
+                feed,
+                pending,
+                cursor,
+                position: 0,
+            }) as Box<dyn EventReceiver>)
+        })
+    }
+}
+
+/// The durable subscription receiver (see the module docs). Its
+/// position is the last-*yielded* event's offset, so `save_offset()`
+/// never checkpoints an event the consumer did not observe.
+pub struct MemEventReceiver {
+    stream: String,
+    consumer: String,
+    state: Arc<State>,
+    feed: crate::wakes::WakeFeed,
+    pending: VecDeque<StreamEvent>,
+    cursor: i64,
+    position: i64,
+}
+
+impl std::fmt::Debug for MemEventReceiver {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("MemEventReceiver")
+            .field("stream", &self.stream)
+            .finish_non_exhaustive()
+    }
+}
+
+impl MemEventReceiver {
+    /// One forward read from the cursor into the pending buffer.
+    fn drain_a_page(&mut self) -> Result<()> {
+        let events = {
+            let s = self.state.streams.guard();
+            read_events(&s, &self.stream, self.cursor, SUBSCRIBE_PAGE)
+        };
+        for row in events {
+            self.cursor = row.offset;
+            self.pending.push_back(event_from_row(row, &self.stream));
+        }
+        Ok(())
+    }
+
+    fn pop_positional(&mut self) -> Option<StreamEvent> {
+        let event = self.pending.pop_front()?;
+        self.position = event.offset;
+        Some(event)
+    }
+}
+
+impl EventReceiver for MemEventReceiver {
+    fn recv<'a>(&'a mut self) -> BoxedFuture<'a, Option<Result<StreamEvent>>> {
+        Box::pin(async move {
+            loop {
+                if let Some(event) = self.pop_positional() {
+                    return Some(Ok(event));
+                }
+                // Park on the wake feed: a commit that lands after this
+                // park fires a wake that releases it; a commit that
+                // landed before it already buffered a wake (attach-first).
+                let _wake = match self.feed.rx().recv().await {
+                    Some(wake) => wake,
+                    None => return None,
+                };
+                if let Err(e) = self.drain_a_page() {
+                    return Some(Err(database_only(e)));
+                }
+            }
+        })
+    }
+
+    fn try_recv(&mut self) -> Result<Option<StreamEvent>> {
+        if let Some(event) = self.pop_positional() {
+            return Ok(Some(event));
+        }
+        // A buffered wake is the hint that a commit landed which the
+        // drain has not seen yet; the poll drains it and re-reads once.
+        let closed = matches!(
+            self.feed.rx().try_recv(),
+            Err(tokio::sync::mpsc::error::TryRecvError::Disconnected)
+        );
+        self.drain_a_page()?;
+        if let Some(event) = self.pop_positional() {
+            return Ok(Some(event));
+        }
+        if closed {
+            // The bus close (engine drop) is the one close path: once
+            // the feed is disconnected and nothing is buffered or
+            // readable, Closed is the terminal poll arm (never reopens).
+            return Err(Error::Closed);
+        }
+        Ok(None)
+    }
+
+    fn read_since<'a>(
+        &'a mut self,
+        offset: i64,
+        limit: i64,
+    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
+        if limit <= 0 {
+            return Box::pin(async { Ok(Vec::new()) });
+        }
+        let events = {
+            let s = self.state.streams.guard();
+            read_events(&s, &self.stream, offset, limit)
+        };
+        let stream = self.stream.clone();
+        Box::pin(async move {
+            Ok(events
+                .into_iter()
+                .map(|r| event_from_row(r, &stream))
+                .collect())
+        })
+    }
+
+    fn save_offset(&mut self) -> Result<()> {
+        let position = self.position;
+        if position == 0 {
+            return Ok(());
+        }
+        let mut s = self.state.streams.guard();
+        save_offset_row(&mut s, &self.stream, &self.consumer, position);
+        Ok(())
+    }
+
+    fn offset(&self) -> i64 {
+        self.position
+    }
+}
+
+/// `recv()`'s `Err` arm carries `Database` only (ADR-021 §5): a read
+/// error mapped elsewhere is remapped into the opaque fallback with
+/// the original preserved as the source (the decode-side posture
+/// guard — mem reads cannot fail in practice).
+fn database_only(e: Error) -> Error {
+    match e {
+        already @ Error::Database(_) => already,
+        other => Error::database(other),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::store::MemStore;
+    use alkstore::{EventReceiver, Store, StreamHandle};
+
+    fn handle(store: &MemStore, name: &str) -> MemStreamHandle {
+        MemStreamHandle::new(
+            name.to_string(),
+            store.state.clone(),
+            store.clock.clone(),
+            store.wakes.clone(),
+        )
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn publish_read_round_trip_is_global_fifo_by_offset() {
+        let store = MemStore::new();
+        let handle = handle(&store, "orders");
+        let a = handle.publish(serde_json::json!(1)).await.unwrap();
+        let b = handle.publish(serde_json::json!(2)).await.unwrap();
+        let keyed = handle
+            .publish_with_key(Some("cust-7".into()), serde_json::json!(3))
+            .await
+            .unwrap();
+        assert!(a < b && b < keyed, "offsets are monotone per stream");
+        let events = handle.read_since(0, 10).await.unwrap();
+        assert_eq!(events.len(), 3);
+        assert_eq!(
+            events.iter().map(|e| e.offset).collect::<Vec<_>>(),
+            vec![a, b, keyed],
+            "offset ASC — global FIFO regardless of keys"
+        );
+        assert_eq!(events[0].stream, "orders");
+        assert_eq!(events[0].key, None);
+        assert_eq!(events[2].key.as_deref(), Some("cust-7"));
+        assert_eq!(events[2].payload_as::<i64>().unwrap(), 3);
+        assert!(
+            events
+                .iter()
+                .all(|e| e.created_at == store.clock.unix_now())
+        );
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn empty_some_key_is_invalid_at_every_entry() {
+        let store = MemStore::new();
+        let handle = handle(&store, "orders");
+        let err = handle
+            .publish_with_key(Some(String::new()), serde_json::json!(1))
+            .await
+            .unwrap_err();
+        assert!(matches!(err, alkstore::Error::InvalidName { .. }));
+        let err = handle
+            .publish_with_key(Some("   ".into()), serde_json::json!(1))
+            .await
+            .unwrap_err();
+        assert!(matches!(err, alkstore::Error::InvalidName { .. }));
+        let mut tx = store.begin_tx().await.unwrap();
+        let err = tx
+            .publish_with_key_tx("orders", Some(String::new()), serde_json::json!(1))
+            .await
+            .unwrap_err();
+        assert!(matches!(err, alkstore::Error::InvalidName { .. }));
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn read_extents_clamp_empty() {
+        let store = MemStore::new();
+        let handle = handle(&store, "orders");
+        handle.publish(serde_json::json!(1)).await.unwrap();
+        assert!(handle.read_since(0, 0).await.unwrap().is_empty());
+        assert!(
+            handle.read_since(0, -3).await.unwrap().is_empty(),
+            "the LIMIT -1 dialect artifact is dead"
+        );
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn saves_are_monotone_and_absent_consumers_read_zero() {
+        let store = MemStore::new();
+        let handle = handle(&store, "orders");
+        assert_eq!(
+            handle.get_offset("c1").await.unwrap(),
+            0,
+            "an absent consumer reads 0"
+        );
+        let tail = handle.publish(serde_json::json!(1)).await.unwrap();
+        handle.save_offset("c1", tail).await.unwrap();
+        assert_eq!(handle.get_offset("c1").await.unwrap(), tail);
+        handle.save_offset("c1", tail - 1).await.unwrap();
+        assert_eq!(
+            handle.get_offset("c1").await.unwrap(),
+            tail,
+            "a below-checkpoint save is a silent no-op (monotone)"
+        );
+        handle.save_offset("c1", tail).await.unwrap();
+        assert_eq!(
+            handle.get_offset("c1").await.unwrap(),
+            tail,
+            "an equal save is also a no-op"
+        );
+        handle.save_offset("c1", tail + 5).await.unwrap();
+        assert_eq!(handle.get_offset("c1").await.unwrap(), tail + 5);
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn trim_is_exact_boundary_and_wakes_nothing() {
+        let store = MemStore::new();
+        let handle = handle(&store, "orders");
+        let a = handle.publish(serde_json::json!(1)).await.unwrap();
+        let b = handle.publish(serde_json::json!(2)).await.unwrap();
+        let c = handle.publish(serde_json::json!(3)).await.unwrap();
+        assert_eq!(
+            handle.trim_to(b - 1).await.unwrap(),
+            1,
+            "exact boundary: offset <= horizon"
+        );
+        let events = handle.read_since(0, 10).await.unwrap();
+        assert_eq!(
+            events.iter().map(|e| e.offset).collect::<Vec<_>>(),
+            vec![b, c],
+            "surviving rows keep their offsets, never renumbered"
+        );
+        assert_eq!(
+            handle.trim_to(-100).await.unwrap(),
+            0,
+            "a negative horizon deletes nothing"
+        );
+        assert_eq!(handle.trim_to(c).await.unwrap(), 2);
+        assert_eq!(handle.trim_to(c).await.unwrap(), 0, "idempotent");
+        // Reads from a trimmed region resume at the horizon's first
+        // remaining row.
+        assert!(handle.read_since(0, 10).await.unwrap().is_empty());
+        let _ = a;
+        assert_eq!(handle.get_offset("late").await.unwrap(), 0);
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn subscriber_replays_from_the_saved_offset_and_ride_wakes() {
+        let store = MemStore::new();
+        let handle = handle(&store, "orders");
+        let _a = handle.publish(serde_json::json!(1)).await.unwrap();
+        let b = handle.publish(serde_json::json!(2)).await.unwrap();
+        handle.save_offset("c1", b).await.unwrap();
+        let mut receiver = handle.subscribe("c1").await.unwrap();
+        // Everything after the checkpoint replays on attach.
+        let c = handle.publish(serde_json::json!(3)).await.unwrap();
+        let first = receiver.recv().await.unwrap().unwrap();
+        assert_eq!(
+            first.offset, c,
+            "replay starts at the stored checkpoint's forward tail"
+        );
+        assert_eq!(first.payload_as::<i64>().unwrap(), 3);
+        // Post-attach publishes deliver without polling.
+        let d = handle.publish(serde_json::json!(4)).await.unwrap();
+        let second = receiver.recv().await.unwrap().unwrap();
+        assert_eq!(second.offset, d);
+        // The explicit-save receiver form and the direct form compose.
+        receiver.save_offset().unwrap();
+        assert_eq!(receiver.offset(), d);
+        assert_eq!(
+            handle.get_offset("c1").await.unwrap(),
+            d,
+            "the receiver saves the last-yielded offset"
+        );
+        // A save before any yield is a no-op on the stored checkpoint.
+        let mut fresh = handle.subscribe("c2").await.unwrap();
+        assert_eq!(fresh.offset(), 0);
+        fresh.save_offset().unwrap();
+        assert_eq!(handle.get_offset("c2").await.unwrap(), 0);
+        let _ = _a;
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn receiver_try_recv_reads_without_await() {
+        let store = MemStore::new();
+        let handle = handle(&store, "orders");
+        let mut receiver = handle.subscribe("c1").await.unwrap();
+        assert!(try_none(receiver.as_mut()), "idle reads as Ok(None)");
+        let a = handle.publish(serde_json::json!(1)).await.unwrap();
+        let got = receiver.try_recv().unwrap().unwrap();
+        assert_eq!(got.offset, a);
+        assert!(try_none(receiver.as_mut()));
+    }
+
+    fn try_none(receiver: &mut dyn EventReceiver) -> bool {
+        matches!(receiver.try_recv(), Ok(None))
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn engine_drop_closes_the_receiver_on_both_arms() {
+        let (mut receiver, offset) = {
+            let store = MemStore::new();
+            let handle = handle(&store, "orders");
+            let offset = handle.publish(serde_json::json!(1)).await.unwrap();
+            (handle.subscribe("c1").await.unwrap(), offset)
+        };
+        // The engine is gone; the replayed pre-drop event still
+        // delivers (it was committed and visible), then the receiver
+        // is closed — terminal on both arms (never reopens).
+        let delivered = receiver
+            .try_recv()
+            .unwrap()
+            .expect("the pre-drop event replays");
+        assert_eq!(delivered.offset, offset);
+        assert!(matches!(receiver.try_recv(), Err(alkstore::Error::Closed)));
+        let closed = tokio::time::timeout(std::time::Duration::from_millis(200), receiver.recv())
+            .await
+            .expect("the closed feed parks no receiver");
+        assert!(
+            closed.is_none(),
+            "the terminal None arm (engine drop is the close cause)"
+        );
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn trim_wakes_nothing_for_an_open_subscriber() {
+        let store = MemStore::new();
+        let handle = handle(&store, "orders");
+        let a = handle.publish(serde_json::json!(1)).await.unwrap();
+        let b = handle.publish(serde_json::json!(2)).await.unwrap();
+        let mut receiver = handle.subscribe("c1").await.unwrap();
+        assert_eq!(receiver.recv().await.unwrap().unwrap().offset, a);
+        assert_eq!(receiver.recv().await.unwrap().unwrap().offset, b);
+        handle.trim_to(b).await.unwrap();
+        let quiet =
+            tokio::time::timeout(std::time::Duration::from_millis(50), receiver.recv()).await;
+        assert!(quiet.is_err(), "trim emits no wake and delivers no event");
+        let _ = receiver.read_since(0, 10).await.unwrap();
+    }
+}