mem-engine-foundations — the mem engine's foundations land (alkstore-mem, wave 6 task 1 of the ADR-024 chain): fourth workspace member with the manifest surface ADR-024 §1 pins — alkstore versioned path pin (ADR-017 §4.1) plus tokio restricted to the wasm-supported subset (sync + time; dev-deps rt/macros for the current-thread-flavor tests), no driver deps, no parking_lot (the description's prose named it, but the ADR + acceptance criterion pin the manifest tighter — short-held internal guards are std::sync::Mutex behind a state::Guard trait that folds poisoning away structurally, guards never held across awaits or user code, tokio mpsc the cross-await primitive). Five modules re-exported from lib.rs whose crate docs carry the ADR-016 posture statement (honest-ephemeral, full contract v1, wasm32-compile-clean identity, architecture sketch with the wave-6 construction line). MemStore::new() is the engine-native isolated constructor (ADR-024 §4's documented open-prose exception): fresh guarded state core (state.rs — empty named per-mechanism sections QueueState/StreamState/LockState/ScheduleState, shapes owned by the mechanism tasks), fresh wake bus, fresh clock per open; teardown is drop driving WakeBus::close() — no explicit close form. Wake substrate (wakes.rs): WakeBus::attach → WakeFeed (one unbounded mpsc feed per subscriber, detach-on-drop, attach under the registry guard so attach-after-commit no-replay is structural), fire_commit_wakes(channels) as the commit-ordered wake source's entry point (guard makes watermark bump + per-channel fanout one operation — per-subscriber FIFO by commit order, never coalesced), close empties senders so engine drop disconnects every feed; commit_watermark is a cfg(test) ordering observable only. Clock (clock.rs): one accessor Clock::unix_now over SystemTime, freeze/advance/unfreeze as the cfg(test) deterministic-seam; MemStore clock/wakes cfg(test) accessors. Validation (validation.rs) carries no new rule — pin module only (empty/whitespace InvalidName, prefix ReservedName, RESERVED_LISTENER_RECONNECTED rejected though unused here); mechanisms route core's validate_shared_name/local_name directly. Targeted allow(dead_code) on not-yet-wired scaffold surfaces so clippy -D warnings stays green pre-consumer (mechanism tasks own exhausting the allows); tokio time pinned unexercised (lock-TTL/scheduler consumers arrive with their tasks). Unit pins: 17 mem tests — two-instances-share-nothing (Arc pointer identity + behavioral: frozen clock isolation, cross-instance wake silence), state section independence/relock, wake routing (channel-scoped + foreign silence, per-notify never coalesced, watermark-ordered FIFO, no-replay, close disconnects all + post-close inert, dropped feed detaches), engine-drop closes feeds, clock determinism, validation pin row. Task file: status completed, Notes (decisions of record: parking_lot rejection, empty sections, substrate API shape, no-close, scaffold allow posture, unexercised time feature) + Summary filled. Verified: cargo build; workspace cargo test green (pg 124/25/9, sqlite 191/25, core 25, suite 3, mem 17); clippy --all-targets -D warnings clean; cargo fmt --check clean; cargo check --target wasm32-unknown-unknown -p alkstore-mem clean — the wave-6 wasm gate starts green from this task onward

alkstore-mem/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "alkstore-mem"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[dependencies]
+alkstore = { version = "0.1", path = "../alkstore" }
+tokio = { version = "1", features = ["sync", "time"] }
+
+[dev-dependencies]
+tokio = { version = "1", features = ["rt", "macros"] }

alkstore-mem/src/clock.rs

@@ -0,0 +1,110 @@
+//! The engine's clock — every timestamp the mem engine produces (unix
+//! seconds stamps, visibility deadlines, TTLs, scheduler boundaries)
+//! reads this one accessor ([`Clock::unix_now`]); the process clock is
+//! all there is (engine-mem.md: no server clock to defer to).
+//!
+//! The single point exists so tests can observe timestamps
+//! deterministically the way the SQL engines observe row shapes: a
+//! `#[cfg(test)]` freeze/advance seam pins a wall value and every
+//! accessor call returns it (see the unit pins below). Production
+//! reads `std::time::SystemTime`.
+
+#[cfg(test)]
+use std::sync::Mutex;
+
+#[cfg(test)]
+use crate::state::Guard;
+
+// Mechanism task code stamps every timestamp through this accessor —
+// scaffolding reads as dead until the wave-6 mechanism tasks wire it.
+#[allow(dead_code)]
+fn process_unix_now() -> i64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|since| since.as_secs() as i64)
+        .unwrap_or(0)
+}
+
+/// The engine's single clock.
+#[derive(Debug)]
+pub(crate) struct Clock {
+    #[cfg(test)]
+    frozen: Mutex<Option<i64>>,
+}
+
+impl Default for Clock {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Clock {
+    pub(crate) fn new() -> Self {
+        Self {
+            #[cfg(test)]
+            frozen: Mutex::new(None),
+        }
+    }
+
+    /// Unix-seconds "now" — the one timestamp source of the engine.
+    #[allow(dead_code)]
+    pub(crate) fn unix_now(&self) -> i64 {
+        #[cfg(test)]
+        let frozen = { *self.frozen.guard() };
+        #[cfg(test)]
+        if let Some(unix) = frozen {
+            return unix;
+        }
+        process_unix_now()
+    }
+
+    #[cfg(test)]
+    pub(crate) fn freeze(&self, at_unix: i64) {
+        *self.frozen.guard() = Some(at_unix);
+    }
+
+    #[cfg(test)]
+    pub(crate) fn advance(&self, seconds: i64) {
+        let mut guard = self.frozen.guard();
+        *guard = Some(frozen_now(&guard).saturating_add(seconds));
+    }
+
+    #[cfg(test)]
+    pub(crate) fn unfreeze(&self) {
+        *self.frozen.guard() = None;
+    }
+}
+
+#[cfg(test)]
+fn frozen_now(frozen: &Option<i64>) -> i64 {
+    frozen.unwrap_or_else(process_unix_now)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn unfrozen_reads_a_plausible_process_time() {
+        let clock = Clock::new();
+        let now = clock.unix_now();
+        assert!(
+            now > 1_700_000_000,
+            "the accessor reads the process clock when unfrozen"
+        );
+    }
+
+    #[test]
+    fn frozen_reads_deterministically_and_advances() {
+        let clock = Clock::new();
+        clock.freeze(1_234_567);
+        assert_eq!(clock.unix_now(), 1_234_567);
+        clock.advance(10);
+        assert_eq!(clock.unix_now(), 1_234_577);
+        clock.unfreeze();
+        assert!(
+            clock.unix_now() > 1_700_000_000,
+            "unfreeze restores the process clock"
+        );
+    }
+}

alkstore-mem/src/lib.rs

@@ -0,0 +1,52 @@
+//! alkstore-mem — the mem engine: the core contract implemented on
+//! guarded in-memory state, process-owned (ADR-001, ADR-024).
+//!
+//! # Posture
+//!
+//! **Honest-ephemeral** (ADR-016 carriers, ADR-024 §2): no durability,
+//! single-process, ephemeral by design — a legitimate production
+//! choice for workloads that accept ephemerality, and never fleet-valid
+//! (the shared-pool predicate fails structurally — there is no pool).
+//! The dependency on `alkstore-mem` *is* the ephemerality declaration;
+//! compile-time engine identity (this crate) carries the honesty —
+//! no runtime surface is fabricated to detect it. Each
+//! `MemStore::new` yields a fully isolated instance sharing nothing;
+//! teardown is drop, and engine drop closes open receivers.
+//!
+//! **Full contract v1** (ADR-024 §3): every mechanism, every guarantee
+//! row — the queue state machine, the scheduler collapse, streams with
+//! per-consumer offsets and trim, TTL locks, the tx seam, wakes. Born
+//! pinned: the contract suite's third `StoreFactory` column (ADR-022)
+//! runs this engine under a current-thread runtime flavor.
+//!
+//! **wasm32** ([ADR-024] §5): this crate compiles clean for
+//! `wasm32-unknown-unknown` — tokio restricted to the wasm-supported
+//! subset (sync + time), no driver dependencies, no `spawn_blocking`
+//! anywhere; internals are current-thread-runtime-clean by
+//! construction. Every timestamp reads the process clock through one
+//! internal accessor (no server clock to defer to).
+//!
+//! # Architecture
+//!
+//! The store instance *is* the state: guarded per-mechanism sections
+//! (`crate::state`) behind short-held synchronous internal locks,
+//! tokio primitives for cross-await coordination. The tx seam is an
+//! overlay (stage → commit-merge → wake-at-commit; drop = discard);
+//! wakes route through in-process channels (`crate::wakes`) —
+//! delivered at the commit boundary directly to each open receiver,
+//! per-subscriber FIFO by commit order, never coalesced (no batching
+//! boundary exists).
+//!
+//! Wave 6 is under construction (ADR-024): the foundations —
+//! `MemStore::new`, the guarded state core, the wake-routing
+//! substrate, the clock accessor, entry-point validation — are laid;
+//! the `Store`/`TxHandle` trait surfaces and the mechanism handles
+//! land with the wave's seam and mechanism tasks.
+
+mod clock;
+mod state;
+mod store;
+mod validation;
+mod wakes;
+
+pub use store::MemStore;

alkstore-mem/src/state.rs

@@ -0,0 +1,133 @@
+//! The guarded per-mechanism state core — the store instance *is* the
+//! state (engine-mem.md §Architecture): per-mechanism tables owned by
+//! the engine instance behind the family's locking discipline
+//! (short-held synchronous internal locks; tokio primitives for
+//! cross-await coordination — [`crate::wakes`]' channels and the
+//! scheduler's boundary waits arrive with their tasks).
+//!
+//! The idiom: each mechanism's committed state sits behind one
+//! short-held [`std::sync::Mutex`] section, accessed through
+//! [`Guard::guard`]. Guards are never held across an `.await`, never
+//! across user-provided code (delivery closures, payload callbacks) —
+//! staging happens engine-side, so the mutex cannot observe a panic
+//! while held and poisoning is structurally unreachable; the
+//! [`Guard::guard`] fold into the inner value keeps the type honest
+//! anyway.
+//!
+//! This module owns the shared scaffolding only: the container, the
+//! section shapes' names, and the guard idiom. Each mechanism's table
+//! shape is owned by its own wave-6 task (`mem-engine-queues`,
+//! `mem-engine-streams`, `mem-engine-locks`,
+//! `mem-engine-scheduler-outbox`) — the sections here start empty and
+//! grow there.
+
+use std::sync::Mutex;
+use std::sync::MutexGuard;
+
+/// A short-held synchronous internal lock: `guard()` is the one access
+/// idiom (see the module docs for the discipline).
+pub(crate) trait Guard<T> {
+    fn guard(&self) -> MutexGuard<'_, T>;
+}
+
+impl<T> Guard<T> for Mutex<T> {
+    fn guard(&self) -> MutexGuard<'_, T> {
+        self.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+}
+
+/// The queue mechanism's committed section: the jobs table and the
+/// dead-letter storage (`mem-engine-queues` owns the shapes).
+#[derive(Debug, Default)]
+pub(crate) struct QueueState {}
+
+/// The stream mechanism's committed section: event logs and
+/// per-consumer offsets (`mem-engine-streams` owns the shapes).
+#[derive(Debug, Default)]
+pub(crate) struct StreamState {}
+
+/// The named-locks mechanism's committed section: the TTL registry
+/// (`mem-engine-locks` owns the shapes).
+#[derive(Debug, Default)]
+pub(crate) struct LockState {}
+
+/// The scheduler mechanism's committed section: the schedule table
+/// (`mem-engine-scheduler-outbox` owns the shapes; the outbox rides
+/// queues).
+#[derive(Debug, Default)]
+pub(crate) struct ScheduleState {}
+
+/// The store's committed state: one guarded section per mechanism.
+/// Sections are independent — a mechanism task locks only its own.
+// The sections are read by their owning wave-6 mechanism tasks; the
+// foundations build alone sees them as dead.
+#[allow(dead_code)]
+pub(crate) struct State {
+    pub(crate) queues: Mutex<QueueState>,
+    pub(crate) streams: Mutex<StreamState>,
+    pub(crate) locks: Mutex<LockState>,
+    pub(crate) schedules: Mutex<ScheduleState>,
+}
+
+impl State {
+    pub(crate) fn new() -> Self {
+        Self {
+            queues: Mutex::new(QueueState::default()),
+            streams: Mutex::new(StreamState::default()),
+            locks: Mutex::new(LockState::default()),
+            schedules: Mutex::new(ScheduleState::default()),
+        }
+    }
+}
+
+impl std::fmt::Debug for State {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("State").finish_non_exhaustive()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn sections_are_per_mechanism_and_independent() {
+        let state = State::new();
+        let queued_ptr = {
+            let mut queues = state.queues.guard();
+            std::ptr::from_mut(&mut *queues)
+        };
+        let streams_ptr = {
+            let mut streams = state.streams.guard();
+            std::ptr::from_mut(&mut *streams)
+        };
+        let queues_ptr_again = {
+            let mut queues = state.queues.guard();
+            std::ptr::from_mut(&mut *queues)
+        };
+        assert_eq!(queued_ptr, queues_ptr_again, "the same section relocks");
+        assert_ne!(
+            queued_ptr.cast::<()>(),
+            streams_ptr.cast::<()>(),
+            "sections are distinct storage"
+        );
+    }
+
+    #[test]
+    fn sections_relock_after_guard_drop() {
+        let state = std::sync::Arc::new(State::new());
+        std::thread::scope(|scope| {
+            let held = scope.spawn(|| {
+                for _ in 0..100 {
+                    let _guard = state.streams.guard();
+                    std::thread::yield_now();
+                }
+            });
+            for _ in 0..100 {
+                let _guard = state.queues.guard();
+                std::thread::yield_now();
+            }
+            held.join().expect("the lock churn task never panics");
+        });
+    }
+}

alkstore-mem/src/store.rs

@@ -0,0 +1,148 @@
+//! The mem engine's store: `MemStore::new`, the engine-native
+//! constructor (ADR-024 §4 — the one documented exception to the
+//! open prose), and the guarded state it owns.
+//!
+//! Isolation per open: each `new()` yields a fully isolated instance
+//! sharing nothing — a fresh `State` (the guarded per-mechanism
+//! core), a fresh `WakeBus` (the wake-routing substrate), a fresh
+//! `Clock` — exactly the suite factory's isolation contract;
+//! two instances never contend and never observe each other. There is
+//! no connection, no file, no pool; teardown is drop — the
+//! engine-wide `Drop` runs the close path (the wake bus empties, every
+//! open feed disconnects), and the state is gone with the process's
+//! references, honestly ephemeral ADR-016-shaped posture:
+//! the dependency on `alkstore-mem` *is* the ephemerality
+//! declaration).
+//!
+//! The `Store` trait implementation is the seam task's artifact
+//! (`mem-engine-seam-tx` — the tx overlay is the load-bearing half);
+//! this type is the trait boundary's substrate.
+
+use std::sync::Arc;
+
+use crate::clock::Clock;
+use crate::state::State;
+use crate::wakes::WakeBus;
+
+/// The mem engine's store handle. Consumers take it from
+/// `MemStore::new`; once the seam task lands, they hold
+/// `alkstore::Store` — the trait is the contract surface, never the
+/// engine type.
+pub struct MemStore {
+    // Read by the wave-6 mechanism task code (the `Store` seam and the
+    // handles); the foundations build alone sees `state`/`clock` as dead.
+    #[allow(dead_code)]
+    pub(crate) state: Arc<State>,
+    pub(crate) wakes: Arc<WakeBus>,
+    #[allow(dead_code)]
+    pub(crate) clock: Arc<Clock>,
+}
+
+impl Default for MemStore {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl std::fmt::Debug for MemStore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("MemStore").finish_non_exhaustive()
+    }
+}
+
+impl MemStore {
+    /// Construct a fully isolated mem engine instance. Engine-native:
+    /// no connection string, no file path (ADR-024 §4's
+    /// documented exception to the open prose). Teardown is drop;
+    /// each instance shares nothing with any other.
+    pub fn new() -> Self {
+        Self {
+            state: Arc::new(State::new()),
+            wakes: Arc::new(WakeBus::new()),
+            clock: Arc::new(Clock::new()),
+        }
+    }
+
+    #[cfg(test)]
+    pub(crate) fn wakes(&self) -> &Arc<WakeBus> {
+        &self.wakes
+    }
+
+    #[cfg(test)]
+    pub(crate) fn clock(&self) -> &Arc<Clock> {
+        &self.clock
+    }
+}
+
+impl Drop for MemStore {
+    fn drop(&mut self) {
+        self.wakes.close();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn two_new_instances_share_nothing() {
+        let a = MemStore::new();
+        let b = MemStore::new();
+        assert!(
+            !std::sync::Arc::ptr_eq(&a.state, &b.state),
+            "the state cores are distinct instances"
+        );
+        assert!(
+            !std::sync::Arc::ptr_eq(&a.wakes, &b.wakes),
+            "the wake substrates are distinct instances"
+        );
+        assert!(
+            !std::sync::Arc::ptr_eq(&a.clock, &b.clock),
+            "the clocks are distinct instances"
+        );
+    }
+
+    #[test]
+    fn instances_observed_through_their_futures_share_nothing() {
+        let a = MemStore::new();
+        let b = MemStore::new();
+        a.clock().freeze(1_000_000);
+        assert_eq!(a.clock().unix_now(), 1_000_000, "instance a's clock froze");
+        assert!(
+            b.clock().unix_now() > 1_700_000_000,
+            "instance b's clock kept the process time"
+        );
+    }
+
+    #[test]
+    fn wake_substrates_do_not_cross_instances() {
+        let a = MemStore::new();
+        let b = MemStore::new();
+        a.wakes().fire_commit_wakes(vec!["orders".into()]);
+        let mut late_feed = b.wakes().attach("orders");
+        assert!(
+            late_feed.rx().try_recv().is_err(),
+            "instance a's commit fired before the feed attached — instance b's substrate never replayed it, and instance a's substrate is unreachable from b"
+        );
+        let mut a_feed = a.wakes().attach("orders");
+        a.wakes().fire_commit_wakes(vec!["orders".into()]);
+        assert!(
+            matches!(
+                a_feed.rx().try_recv(),
+                Ok(wake) if wake.channel == "orders"
+            ),
+            "instance a's substrate routes its own attaches"
+        );
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn engine_drop_closes_open_feeds() {
+        let store = MemStore::new();
+        let mut feed = store.wakes().attach("orders");
+        std::mem::drop(store);
+        assert!(
+            feed.rx().recv().await.is_none(),
+            "teardown is drop: the engine's close path disconnects every feed"
+        );
+    }
+}

alkstore-mem/src/validation.rs

@@ -0,0 +1,49 @@
+//! The entry-point name validation is core's ([`alkstore`]'s
+//! `validate_shared_name` / `validate_local_name` — ADR-008 §4), same
+//! as every engine; this module carries no new rule, only the mem
+//! engine's pin that the rule it routes every future entry point
+//! through is the family rule: empty and whitespace-only names are
+//! `InvalidName`, the reserved prefix `__alkstore_` is `ReservedName`,
+//! and the one reserved string (`__alkstore_listener_reconnected__`
+//! — the Postgres reconnect channel, unused on this engine) is still
+//! rejected because it carries the prefix.
+
+#[cfg(test)]
+mod tests {
+    use alkstore::Error;
+
+    #[test]
+    fn empty_and_whitespace_only_names_are_invalid() {
+        for name in ["", "   "] {
+            assert!(
+                matches!(
+                    alkstore::validate_shared_name(name),
+                    Err(Error::InvalidName { .. })
+                ),
+                "`{name}` validates as `InvalidName` on the shared kind"
+            );
+        }
+    }
+
+    #[test]
+    fn reserved_prefix_names_are_rejected() {
+        assert!(matches!(
+            alkstore::validate_shared_name("__alkstore_scheduler"),
+            Err(Error::ReservedName { .. })
+        ));
+    }
+
+    #[test]
+    fn the_reserved_listener_string_is_rejected_even_though_unused_here() {
+        assert!(
+            alkstore::validate_shared_name(alkstore::RESERVED_LISTENER_RECONNECTED).is_err(),
+            "the reserved string is unused on this engine but still rejected"
+        );
+    }
+
+    #[test]
+    fn ordinary_names_validate() {
+        assert!(alkstore::validate_shared_name("orders").is_ok());
+        assert!(alkstore::validate_local_name("__alkstore_-prefixed-local-ids-are-legal").is_ok());
+    }
+}

alkstore-mem/src/wakes.rs

@@ -0,0 +1,334 @@
+//! The wake-routing substrate: in-process channels carrying wakes at
+//! the commit boundary (engine-mem.md §Architecture).
+//!
+//! There is no batching boundary — wakes deliver at the commit
+//! boundary directly to each open receiver, per-subscriber FIFO by
+//! commit order; coalescing is not applicable. The substrate realizes
+//! the shape structurally:
+//!
+//! - **Commit-ordered source**: a mechanism (notify now — streams and
+//!   queue wakes arrive with their tasks) fires its commit's wakes
+//!   through [`WakeBus::fire_commit_wakes`], under one guard: the
+//!   commit sequence bumps and the per-channel fanout completes as a
+//!   single operation, so every subscriber's receipt order is exactly
+//!   commit order. Two concurrent commits serialize on the guard; a
+//!   subscriber's wake stream is FIFO by commit order.
+//! - **Per-channel attach**: [`WakeBus::attach`] registers under the
+//!   same guard and returns an unbounded [`tokio::sync::mpsc`] feed —
+//!   attach never blocks, and a receiver attached after a commit never
+//!   sees it (no replay; the attach either happens before that commit's
+//!   fanout or after it — the guard admits no mid-fanout attach).
+//! - **Subscriber registry**: unbounded senders live only in the bus,
+//!   so the sole close path is the engine's drop. [`WakeBus::close`]
+//!   empties the registry under the same guard — every live feed's
+//!   sender drops and each receiver drains to `Disconnected`; the
+//!   consumer-visible `recv() -> None` terminal arm arrives with
+//!   `mem-engine-notify-listen`'s [`alkstore::WakeReceiver`] bridge
+//!   over this substrate. A dropped feed detaches itself
+//!   ([`WakeFeed`]'s `Drop`).
+//!
+//! The `Wake` value crosses exactly as the contract pins it — channel
+//! only, never payloads or ids (ADR-008 §3). The fanout is per-notify:
+//! one commit's notify on a channel is one wake per subscribed feed,
+//! never coalesced (no batching boundary exists to coalesce across).
+
+// The substrate routes every wave-6 mechanism's commit-boundary
+// wakes; until the mechanism tasks wire it, the foundations-only
+// build sees the attach/fire surface as dead.
+#![allow(dead_code)]
+
+use std::sync::Arc;
+use std::sync::Mutex;
+
+use alkstore::Wake;
+
+use crate::state::Guard;
+
+/// A subscriber's registry row: its id (the detach key), the channel
+/// it attached on (exact-match deliverance), and its feed's sender.
+/// Senders live only here — a subscriber holds no bus reference of its
+/// own beyond its [`WakeFeed`] (which carries the bus Arc for
+/// detach-on-drop, never for delivery), so the engine's close empties
+/// the senders and the feeds disconnect regardless of who still holds
+/// a feed.
+#[derive(Debug)]
+struct Subscription {
+    id: u64,
+    channel: String,
+    tx: tokio::sync::mpsc::UnboundedSender<Wake>,
+}
+
+/// The internal guarded registry: next-subscriber id, next commit
+/// sequence, and the live subscriptions in first-attached order.
+#[derive(Debug, Default)]
+struct Registry {
+    next_subscriber_id: u64,
+    next_commit_seq: u64,
+    subscriptions: Vec<Subscription>,
+}
+
+/// The in-process wake bus: the commit-ordered wake source's entry
+/// point ([`WakeBus::fire_commit_wakes`]) and the per-channel receiver
+/// attach ([`WakeBus::attach`]) over one guarded registry.
+pub(crate) struct WakeBus {
+    registry: Mutex<Registry>,
+    closed: std::sync::atomic::AtomicBool,
+}
+
+impl Default for WakeBus {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl std::fmt::Debug for WakeBus {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("WakeBus").finish_non_exhaustive()
+    }
+}
+
+impl WakeBus {
+    #[allow(dead_code)]
+    pub(crate) fn new() -> Self {
+        Self {
+            registry: Mutex::new(Registry::default()),
+            closed: std::sync::atomic::AtomicBool::new(false),
+        }
+    }
+
+    /// Attach a feed for `channel` — starts from "now": the attach
+    /// lands under the registry guard, so commits that fired before it
+    /// are never replayed into it. Returns the feed plus its detach
+    /// handle (drop the [`WakeFeed`] to detach).
+    pub(crate) fn attach(self: &Arc<Self>, channel: &str) -> WakeFeed {
+        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
+        let mut registry = self.registry.guard();
+        if self.closed.load(std::sync::atomic::Ordering::Acquire) {
+            return WakeFeed {
+                id: 0,
+                bus: Arc::clone(self),
+                rx: dead_feed_receiver(),
+            };
+        }
+        registry.next_subscriber_id += 1;
+        let id = registry.next_subscriber_id;
+        registry.subscriptions.push(Subscription {
+            id,
+            channel: channel.to_string(),
+            tx,
+        });
+        WakeFeed {
+            id,
+            bus: Arc::clone(self),
+            rx,
+        }
+    }
+
+    /// Fire the commit boundary's wakes — the mechanism's wake source:
+    /// under one guard, the commit sequence bumps once and every feed
+    /// attached on a fired channel receives one [`Wake`] per commit-notify,
+    /// in commit order. A channel with no subscribers delivers nothing.
+    /// A closed bus is inert.
+    pub(crate) fn fire_commit_wakes(&self, channels: impl IntoIterator<Item = String>) {
+        if self.closed.load(std::sync::atomic::Ordering::Acquire) {
+            return;
+        }
+        let mut registry = self.registry.guard();
+        registry.next_commit_seq += 1;
+        let fired: Vec<String> = channels.into_iter().collect();
+        for channel in &fired {
+            for subscription in &registry.subscriptions {
+                if subscription.channel == *channel {
+                    let wake = Wake::new(channel.clone());
+                    let _ = subscription.tx.send(wake);
+                }
+            }
+        }
+    }
+
+    /// The commit watermark: how many commit boundaries have fired
+    /// through this bus. The substrate's ordering observable — the
+    /// consumer contract never surfaces it (wakes carry the channel
+    /// only); unit tests pin FIFO-by-commit-order against it.
+    #[cfg(test)]
+    pub(crate) fn commit_watermark(&self) -> u64 {
+        self.registry.guard().next_commit_seq
+    }
+
+    /// Close the bus — the engine's drop path. Empties the registry
+    /// under one guard: every feed's sender drops, receivers drain to
+    /// disconnected, and attach/fire after it are no-ops. Idempotent.
+    pub(crate) fn close(&self) {
+        self.closed
+            .store(true, std::sync::atomic::Ordering::Release);
+        let mut registry = self.registry.guard();
+        registry.subscriptions.clear();
+    }
+
+    fn detach(&self, id: u64) {
+        let mut registry = self.registry.guard();
+        registry.subscriptions.retain(|sub| sub.id != id);
+    }
+}
+
+/// A feed whose sender never existed — the attach-on-closed shape (the
+/// engine is gone; the feed reads as closed immediately, the same
+/// terminal arm an engine drop produces).
+fn dead_feed_receiver() -> tokio::sync::mpsc::UnboundedReceiver<Wake> {
+    let (_tx, rx) = tokio::sync::mpsc::unbounded_channel::<Wake>();
+    rx
+}
+
+/// A receiver's handle into the substrate: the raw unbounded feed of
+/// [`Wake`] values (per-subscriber FIFO by commit order — see the
+/// module docs) plus its own-detach-on-drop. This is the substrate
+/// primitive; the consumer-visible `WakeReceiver` bridge (the
+/// `recv`/`try_recv`/`recv_timeout` forms and the terminal `None`
+/// arm) is `mem-engine-notify-listen`'s surface over it.
+pub(crate) struct WakeFeed {
+    id: u64,
+    bus: Arc<WakeBus>,
+    rx: tokio::sync::mpsc::UnboundedReceiver<Wake>,
+}
+
+impl std::fmt::Debug for WakeFeed {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("WakeFeed")
+            .field("id", &self.id)
+            .finish_non_exhaustive()
+    }
+}
+
+impl WakeFeed {
+    pub(crate) fn rx(&mut self) -> &mut tokio::sync::mpsc::UnboundedReceiver<Wake> {
+        &mut self.rx
+    }
+}
+
+impl Drop for WakeFeed {
+    fn drop(&mut self) {
+        if self.id != 0 {
+            self.bus.detach(self.id);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use tokio::sync::mpsc::error::TryRecvError;
+
+    fn delivered(feed: &mut WakeFeed) -> Vec<Wake> {
+        let mut wakes = Vec::new();
+        loop {
+            match feed.rx().try_recv() {
+                Ok(wake) => wakes.push(wake),
+                Err(TryRecvError::Empty) => break,
+                Err(TryRecvError::Disconnected) => {
+                    panic!("the feed disconnected unexpectedly")
+                }
+            }
+        }
+        wakes
+    }
+
+    #[test]
+    fn delivery_is_channel_scoped_and_never_coalesced() {
+        let bus = Arc::new(WakeBus::new());
+        let mut a = bus.attach("orders");
+        let mut b = bus.attach("orders");
+        let mut foreign = bus.attach("payments");
+        bus.fire_commit_wakes(vec!["orders".into(), "orders".into(), "payments".into()]);
+        let a_wakes = delivered(&mut a);
+        let b_wakes = delivered(&mut b);
+        let foreign_wakes = delivered(&mut foreign);
+        assert_eq!(
+            a_wakes, b_wakes,
+            "subscribers on the same channel receive the same wakes"
+        );
+        assert!(
+            a_wakes.iter().all(|w| w.channel == "orders"),
+            "each wake carries its channel"
+        );
+        assert_eq!(
+            foreign_wakes[0].channel, "payments",
+            "a foreign-channel subscriber receives only its own channel"
+        );
+        assert_eq!(
+            a_wakes.len(),
+            2,
+            "per-notify delivery — one wake per commit, never coalesced"
+        );
+    }
+
+    #[test]
+    fn per_subscriber_fifo_by_commit_order() {
+        let bus = Arc::new(WakeBus::new());
+        let mut feed = bus.attach("orders");
+        for seq in 0..8 {
+            assert_eq!(
+                bus.commit_watermark(),
+                seq,
+                "the watermark advances one per fired commit boundary"
+            );
+            bus.fire_commit_wakes(vec![format!("orders-{seq}"), "orders".into()]);
+        }
+        assert_eq!(bus.commit_watermark(), 8);
+        let wakes = delivered(&mut feed);
+        assert_eq!(
+            wakes,
+            vec![Wake::new("orders".into()); 8],
+            "each commit's wake delivered, in commit order"
+        );
+    }
+
+    #[test]
+    fn attach_after_commit_sees_nothing_before_it() {
+        let bus = Arc::new(WakeBus::new());
+        bus.fire_commit_wakes(vec!["orders".into()]);
+        let mut late = bus.attach("orders");
+        assert!(
+            late.rx().try_recv().is_err(),
+            "a listener attached after a commit never sees it (no replay)"
+        );
+        bus.fire_commit_wakes(vec!["orders".into()]);
+        assert_eq!(
+            delivered(&mut late),
+            vec![Wake::new("orders".into())],
+            "post-attach commits deliver"
+        );
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn close_disconnects_every_live_feed() {
+        let bus = Arc::new(WakeBus::new());
+        let mut a = bus.attach("orders");
+        let mut b = bus.attach("payments");
+        bus.close();
+        bus.fire_commit_wakes(vec!["orders".into()]);
+        let mut late = bus.attach("orders");
+        for feed in [&mut a, &mut b, &mut late] {
+            assert!(
+                feed.rx().recv().await.is_none(),
+                "engine close is the terminal None arm for every open receiver"
+            );
+        }
+        assert_eq!(bus.commit_watermark(), 0, "a fire on a closed bus is inert");
+    }
+
+    #[test]
+    fn dropped_feed_detaches() {
+        let bus = Arc::new(WakeBus::new());
+        let feed = bus.attach("orders");
+        let mut kept = bus.attach("orders");
+        std::mem::drop(feed);
+        bus.fire_commit_wakes(vec!["orders".into()]);
+        let wakes = delivered(&mut kept);
+        assert_eq!(
+            wakes.len(),
+            1,
+            "the detached feed received nothing, the live one the wake"
+        );
+    }
+}