Core trait surface: Store, TxHandle, mechanism handles, receivers, with_tx (ADR-007 §with_tx, ADR-008 §1–§3/§8, ADR-009 §1/§6, ADR-014, ADR-015 §2, ADR-019 §1–§6, ADR-021 §1/§4/§5)

Cargo.lock

@@ -9,6 +9,7 @@ dependencies = [
  "serde",
  "serde_json",
  "thiserror",
+ "tokio",
 ]
 
 [[package]]

alkstore/Cargo.toml

@@ -9,3 +9,6 @@ repository.workspace = true
 serde = "1"
 serde_json = "1"
 thiserror = "2"
+
+[dev-dependencies]
+tokio = { version = "1", features = ["macros", "rt", "time"] }

alkstore/src/error.rs

@@ -21,6 +21,21 @@ pub enum Error {
     /// A name failed entry-point validation.
     #[error("invalid name `{name}`")]
     InvalidName { name: String },
+    /// A schedule spec that doesn't parse under the v1 grammar
+    /// (`@every <n><unit>` with `<n>` a positive integer and `<unit>`
+    /// in `s|m|h|d`).
+    ///
+    /// Rejected by `schedule()` before storage, with no engine round
+    /// trip; the caller fixes the spec string (ADR-009 §2, §6).
+    #[error("invalid schedule spec `{spec}`")]
+    InvalidSpec { spec: String },
+    /// The `run_schedules` runner's leadership lock was lost mid-run.
+    ///
+    /// A distinct, matchable outcome from the clean `Ok(())` a stop
+    /// token produces: the caller acts on it by respawning the runner
+    /// (ADR-009 §1, §6).
+    #[error("scheduler leadership lost")]
+    LeadershipLost,
     /// An operation against a closed receiver/source.
     ///
     /// The `try_recv`/`recv_timeout` form of wake close (never

alkstore/src/event_receiver.rs

@@ -0,0 +1,58 @@
+//! The durable stream subscription receiver (ADR-008 §8; ADR-021 §5;
+//! ADR-019 §6).
+//!
+//! Explicit saves only — no `save_every`, no save-on-drop (the
+//! auto-checkpoint ambiguity is not inherited); checkpointing is the
+//! consumer's call.
+
+use crate::Result;
+use crate::future::BoxedFuture;
+use crate::stream::StreamEvent;
+
+/// The durable stream subscription handle `StreamHandle::subscribe`
+/// returns.
+///
+/// # Close and error arms (ADR-021 §5)
+///
+/// `recv()`'s `Err` arm carries `Database` only — payloads cross
+/// raw, and `Codec` lives at `payload_as<T>` on the yielded
+/// [`StreamEvent`]; `Closed` is the `None` arm by construction. Close
+/// is terminal; what causes it is the documented engine asymmetry
+/// (SQLite: watcher death; Postgres: engine shutdown — reconnect
+/// wakes arrive on the receiver). A closed receiver never reopens;
+/// recovery is a fresh `subscribe(consumer)` resuming from the saved
+/// offset. No `recv_timeout` exists here — deliberate absence (a
+/// stream consumer's steady state is wake-driven re-read; polling
+/// needs are served by `try_recv` and the offset-anchored
+/// `read_since`).
+pub trait EventReceiver: Send {
+    /// Wait for the next event. `None` = the stream receiver closed
+    /// (terminal). `Some(Err(...))` carries `Database` only.
+    fn recv<'a>(&'a mut self) -> BoxedFuture<'a, Option<Result<StreamEvent>>>;
+
+    /// Poll without waiting. `Ok(None)` = no event pending right now;
+    /// `Err` carries `Database` or `Closed` (closed source).
+    fn try_recv(&mut self) -> Result<Option<StreamEvent>>;
+
+    /// Offset-anchored read from the stream, independent of this
+    /// receiver's position (saved checkpoints do not gate reads —
+    /// replay is a read concern).
+    fn read_since<'a>(
+        &'a mut self,
+        offset: i64,
+        limit: i64,
+    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>>;
+
+    /// Checkpoint the last-yielded event's offset through the same
+    /// monotone op the direct `save_offset` form uses — the forms
+    /// interleave freely and cannot regress one another (ADR-019 §6).
+    /// A save before any event has been yielded is a no-op (the
+    /// stored checkpoint untouched). The `Err` arm carries `Database`
+    /// only (ADR-021 §5).
+    fn save_offset(&mut self) -> Result<()>;
+
+    /// The receiver's current offset (the last-yielded event's
+    /// offset), used when a consumer wants to interleave the direct
+    /// save form with the receiver's position.
+    fn offset(&self) -> i64;
+}

alkstore/src/future.rs

@@ -0,0 +1,17 @@
+//! The boxed-future alias (ADR-008 §2's "desugared boxed-future form"
+//! — the family-standard async trait posture): every async trait
+//! method crosses the contract as a named future type, keeping the
+//! traits object-safe without the `async_trait` macro's erasure.
+//!
+//! `Send` bounds the future: handles are constructed and polled across
+//! `await` points on multi-threaded tokio runtimes, and
+//! `Store::begin_tx` returns `Box<dyn TxHandle + Send>` — the futures
+//! its methods produce must be equally portable across the runtime's
+//! worker threads (SQLite's engine realizes this by round-tripping
+//! `spawn_blocking` to the writer thread behind the lease, ADR-003).
+
+use std::future::Future;
+use std::pin::Pin;
+
+/// The future type every async contract method returns.
+pub type BoxedFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;

alkstore/src/job_handle.rs

@@ -0,0 +1,57 @@
+//! The boxed job handle a claim returns (ADR-019 §3; ADR-010 §2).
+//!
+//! One-shot/repeatable split, deliberate — do not "fix" it at
+//! implementation: `ack`/`retry`/`fail` consume the handle
+//! (`self: Box<Self>` — each is a terminal transition of the claim);
+//! `heartbeat` takes `&self` and is repeatable (renewal is inherently
+//! repeated within one claim).
+
+use crate::Result;
+use crate::future::BoxedFuture;
+use crate::job::Job;
+
+/// The claim handle: the row value plus its ops.
+///
+/// # The uniform validity predicate
+///
+/// Every op succeeds only while the row is `processing` **and** the
+/// caller's claim deadline is unexpired (ADR-010 §2) — deadline lapse
+/// refuses all ops (heartbeats, acks included) and a reclaim does
+/// too. `false`, not an error: the dual-execution window's worker may
+/// still complete its work, but its op will not land — at-least-once,
+/// as documented.
+pub trait JobHandle: Send {
+    /// The job row, as a value (the [`Job`] the claim returned).
+    fn job(&self) -> &Job;
+
+    /// Ack: deletes the row (completed work leaves no row, ADR-010
+    /// §1). `true` = the op landed under the validity predicate.
+    fn ack(self: Box<Self>) -> BoxedFuture<'static, Result<bool>>;
+
+    /// Retry: return the row to `pending` for redelivery. `None`
+    /// delay = the queue's equal-jitter exponential curve computed
+    /// engine-side; `Some(d)` overrides it (ADR-010 §3). Exhausting
+    /// the attempt budget dead-letters instead. `false` = refused
+    /// under the validity predicate — the row keeps its pre-call
+    /// state (at-least-once redelivery governs).
+    fn retry(
+        self: Box<Self>,
+        err: Option<String>,
+        delay: Option<i64>,
+    ) -> BoxedFuture<'static, Result<bool>>;
+
+    /// Fail: immediate dead-letter — the "stop retrying this"
+    /// operator (ADR-010 §3). `None` err records the engine's
+    /// documented-default string `"failed"` (row content, not a
+    /// matched error variant). `false` = refused under the validity
+    /// predicate.
+    fn fail(self: Box<Self>, err: Option<String>) -> BoxedFuture<'static, Result<bool>>;
+
+    /// Heartbeat: renewal — an **absolute reset** of the claim
+    /// deadline from now (`extend` is the new full deadline in
+    /// seconds, *not* additive to elapsed time; ADR-010 §2). Repeatable
+    /// (`&self`); late heartbeat refused (`false`) — it can never
+    /// steal the job back from a reclaimer; a reclaim consumes an
+    /// attempt.
+    fn heartbeat<'a>(&'a self, extend: i64) -> BoxedFuture<'a, Result<bool>>;
+}

alkstore/src/lib.rs

@@ -6,29 +6,51 @@
 //! this crate and implement its traits.
 
 mod error;
+mod event_receiver;
+mod future;
 mod job;
+mod job_handle;
+mod lock;
 mod opts;
+mod outbox;
 mod payload;
+mod queue;
 mod schedule;
 mod stop_token;
+mod store;
 mod stream;
+mod stream_handle;
+mod tx;
 mod validation;
 mod wake;
+mod wake_receiver;
 
 pub use error::{Error, Result};
+pub use event_receiver::EventReceiver;
+pub use future::BoxedFuture;
 pub use job::{Job, JobState};
+pub use job_handle::JobHandle;
+pub use lock::Lock;
 pub use opts::{EnqueueOpts, QueueOpts, ScheduleOpts};
+pub use outbox::{Delivery, Outbox};
 pub use payload::{encode_payload, payload_as};
+pub use queue::Queue;
 pub use schedule::Schedule;
 pub use stop_token::StopToken;
+pub use store::{Store, WithTxClosure};
 pub use stream::StreamEvent;
+pub use stream_handle::StreamHandle;
+pub use tx::TxHandle;
 pub use validation::{
     NameKind, RESERVED_LISTENER_RECONNECTED, RESERVED_PREFIX, validate_local_name, validate_name,
     validate_shared_name,
 };
 pub use wake::Wake;
+pub use wake_receiver::WakeReceiver;
 
 #[cfg(test)]
 mod tests;
 #[cfg(test)]
+mod trait_probe_tests;
+#[cfg(test)]
 mod value_type_tests;

alkstore/src/lock.rs

@@ -0,0 +1,26 @@
+//! The named-lock handle (ADR-019 §1; ADR-008 §7's locks guarantee row).
+//!
+//! Lock ops are not part of the tx seam — there is no `lock_tx`;
+//! acquisition is a separate auto-commit op and the TTL discipline
+//! governs.
+
+use crate::Result;
+use crate::future::BoxedFuture;
+
+/// A held named lock.
+pub trait Lock: Send {
+    /// The lock name the handle was constructed with (shared
+    /// namespace — validated at the constructor).
+    fn name(&self) -> &str;
+
+    /// Renew: a new **full** TTL window from now. `false` = lost it
+    /// (expired and re-acquired elsewhere — exclusivity lapses
+    /// silently on expiry; no revocation event; ADR-008 §7). Holders
+    /// of long work must renew within TTL.
+    fn renew<'a>(&'a self, ttl: i64) -> BoxedFuture<'a, Result<bool>>;
+
+    /// Release the lock, consuming the handle (RAII shape). `true` =
+    /// it was still held (released now); `false` = already expired or
+    /// re-acquired — releasing then is a no-op, not an error.
+    fn release(self: Box<Self>) -> BoxedFuture<'static, Result<bool>>;
+}

alkstore/src/outbox.rs

@@ -0,0 +1,56 @@
+//! The outbox helper handle (ADR-019 §1; the outbox section of
+//! core-contract.md).
+//!
+//! A helper over queues, not a separate mechanism: the tx-side
+//! transactional enqueue is `TxHandle::outbox_enqueue_tx` (ADR-014);
+//! this handle carries the auto-commit enqueue and the delivery
+//! worker entry point.
+
+use serde_json::Value;
+
+use crate::Result;
+use crate::future::BoxedFuture;
+use crate::job_handle::JobHandle;
+use crate::opts::EnqueueOpts;
+
+/// The delivery callable [`Outbox::run_once`] takes.
+///
+/// A consumer-supplied async callable taking [`Box<dyn JobHandle>`],
+/// returning the boxed future of `Result<()>` (ADR-019 §5). Object
+/// safety encoded as a named one-method trait object. Outcome
+/// posture: `Ok` ⇒ the engine acks the claim; `Err(e)` ⇒ the engine
+/// retries with the queue's curve (`retry(err, None)`).
+///
+/// No engine-issued heartbeat inside delivery — the boxed
+/// [`JobHandle`]'s own `heartbeat` remains available to the closure;
+/// renewal inside the deadline is the consumer's obligation.
+pub trait Delivery: Send {
+    /// Process the claimed job's work. Returns the boxed future of
+    /// `Result<()>`.
+    fn deliver<'a>(&'a mut self, job: Box<dyn JobHandle>) -> BoxedFuture<'a, Result<()>>;
+}
+
+/// The outbox-scoped handle for one outbox name.
+///
+/// The backing queue is derived engine-side under the reserved prefix
+/// (`__alkstore_outbox:{name}`) — reachable only through this surface
+/// and `outbox_enqueue_tx`; enqueue stamps the outbox's derived
+/// `QueueOpts` set (60/5/5, ADR-010 §3).
+pub trait Outbox: Send + Sync {
+    /// The outbox name the handle was constructed with.
+    fn name(&self) -> &str;
+
+    /// Enqueue into the derived backing queue (auto-commit; the
+    /// transactional path is `TxHandle::outbox_enqueue_tx`).
+    fn enqueue<'a>(&'a self, payload: Value, opts: EnqueueOpts) -> BoxedFuture<'a, Result<i64>>;
+
+    /// Pull op: claim one job, run `delivery`, ack on `Ok`,
+    /// `retry(err, None)` (the queue curve) on `Err`. The consumer
+    /// calls it in its own loop. `true` = claimed and processed;
+    /// `false` = no work (a value, not an error).
+    fn run_once<'a>(
+        &'a mut self,
+        worker_id: &str,
+        delivery: &'a mut dyn Delivery,
+    ) -> BoxedFuture<'a, Result<bool>>;
+}

alkstore/src/queue.rs

@@ -0,0 +1,74 @@
+//! The queue mechanism handle (ADR-019 §1, §2; ADR-010 §1–§5).
+//!
+//! Claim/maintenance ops live on the queue-scoped handle — placement
+//! mirroring the honker-rs starting artifact. There is deliberately no
+//! `claim_tx`: a claim's visibility deadline must not be tied to a
+//! business transaction's lifetime (ADR-019 §1).
+
+use serde_json::Value;
+
+use crate::Result;
+use crate::future::BoxedFuture;
+use crate::job::Job;
+use crate::job_handle::JobHandle;
+use crate::opts::EnqueueOpts;
+
+/// The queue-scoped handle: enqueue, claims, and maintenance ops for
+/// one queue name.
+///
+/// Queues are names, not registered objects: the handle carries the
+/// name (validated at the constructor — shared-namespace rules) and
+/// the `QueueOpts` stamps resolve per enqueue, never against a live
+/// registry (ADR-010 §3a).
+pub trait Queue: Send + Sync {
+    /// The queue name the handle was constructed with.
+    fn name(&self) -> &str;
+
+    /// Enqueue a job (auto-commit; each op atomic alone) with the
+    /// handle's `QueueOpts` stamping the row (ADR-020 §3: the one
+    /// handle-carrying shape resolves over this handle's opts).
+    fn enqueue<'a>(&'a self, payload: Value, opts: EnqueueOpts) -> BoxedFuture<'a, Result<i64>>;
+
+    /// Claim one claimable row — exactly-once handout under
+    /// concurrency; `None` = no work (a value, not an error).
+    /// Claim ordering: `priority` DESC, then ready-time, then enqueue
+    /// order (ADR-010 §1). `worker_id` is the claimant identity —
+    /// consumer-local, non-empty only (`InvalidName`).
+    fn claim_one<'a>(
+        &'a self,
+        worker_id: &str,
+    ) -> BoxedFuture<'a, Result<Option<Box<dyn JobHandle>>>>;
+
+    /// Claim up to `n` rows, same ordering and exclusivity as
+    /// [`Queue::claim_one`].
+    fn claim_batch<'a>(
+        &'a self,
+        worker_id: &str,
+        n: i64,
+    ) -> BoxedFuture<'a, Result<Vec<Box<dyn JobHandle>>>>;
+
+    /// Batch ack: applies the uniform validity predicate per id
+    /// (processing state + unexpired claim deadline); lapsed,
+    /// non-claimed, acked-elsewhere, or cancelled ids silently not
+    /// counted. Returns the count that landed; partial success is
+    /// ordinary (ADR-010 §1).
+    fn ack_batch<'a>(&'a self, ids: &[i64]) -> BoxedFuture<'a, Result<i64>>;
+
+    /// Unconditionally delete a row in either `pending` or
+    /// `processing`, regardless of which worker holds it. Not an
+    /// interrupt — the holder's next handle op returns `false`, the
+    /// same shape as deadline expiry. `false` = no such live row.
+    fn cancel<'a>(&'a self, job_id: i64) -> BoxedFuture<'a, Result<bool>>;
+
+    /// Get a job row — sees dead rows too (with `last_error`/
+    /// `died_at`; ADR-010 §4). A pure data read: the returned `Job`
+    /// carries no ops (no ghost handles; a claim's `Box<dyn JobHandle>`
+    /// is the only ops carrier, ADR-019 §3).
+    fn get_job<'a>(&'a self, job_id: i64) -> BoxedFuture<'a, Result<Option<Job>>>;
+
+    /// Sweep this queue: move every past-`expires_at` row (any state —
+    /// pending and processing alike, the no-stranded-rows property,
+    /// ADR-010 §5) to dead, and delete dead rows past their retention
+    /// stamp. Returns the count moved. No leader lock required.
+    fn sweep_expired<'a>(&'a self) -> BoxedFuture<'a, Result<i64>>;
+}

alkstore/src/store.rs

@@ -0,0 +1,134 @@
+//! The `Store` trait — the unified engine-agnostic surface (ADR-008
+//! §1/§6, ADR-019 §1, ADR-009 §1).
+//!
+//! Constructors and their option structs live in the engine crates
+//! (ADR-008 §6): the engine's `open` returns this trait, and a store
+//! handed out by an engine crate honors the full v1 surface. Consumer
+//! code never branches on engine type.
+
+use serde_json::Value;
+
+use crate::Result;
+use crate::future::BoxedFuture;
+use crate::lock::Lock;
+use crate::opts::{QueueOpts, ScheduleOpts};
+use crate::outbox::Outbox;
+use crate::queue::Queue;
+use crate::schedule::Schedule;
+use crate::stop_token::StopToken;
+use crate::stream_handle::StreamHandle;
+use crate::tx::TxHandle;
+use crate::wake_receiver::WakeReceiver;
+
+/// The pinned closure shape `with_tx` takes (ADR-007): a
+/// `FnOnce(&mut dyn TxHandle) -> BoxedFuture<Result<()>>` — dyn-encoded
+/// for object safety, `Send` because the transaction's future must be.
+pub type WithTxClosure<'a> =
+    Box<dyn FnOnce(&mut dyn TxHandle) -> BoxedFuture<'_, Result<()>> + Send + 'a>;
+
+/// The store handle: one object, every mechanism behind it.
+///
+/// # Entry-point validation obligation
+///
+/// Every name-bearing method validates its names at the entry point —
+/// before any engine round trip, on auto-commit paths (and the tx
+/// path via [`TxHandle`], whose docs carry the same obligation):
+/// non-empty otherwise `InvalidName`; reserved-prefix (`__alkstore_`)
+/// rejected with `ReservedName` for the shared-namespace kinds
+/// (channels, streams, queues, outboxes, locks, schedule names, and
+/// `schedule()`'s queue argument). Consumer-local identifiers
+/// (`try_lock`'s `owner`, stream consumers, `worker_id`) take the
+/// non-empty rule only. Implemented engines share the helpers from
+/// `validation`.
+pub trait Store: Send + Sync {
+    /// Begin a transaction — the caller-held [`TxHandle`] whose `*_tx`
+    /// methods are commit-atomic with a business write. Dropping the
+    /// handle without `commit` rolls it back (ADR-021 §4).
+    fn begin_tx(&self) -> BoxedFuture<'_, Result<Box<dyn TxHandle + Send>>>;
+
+    /// Fire-and-forget notify on `channel`. No durability, no replay;
+    /// a listener attached after a commit never sees it. Payload limit
+    /// pg-side (8000 bytes, client-side checked — `PayloadTooLarge`
+    /// before any round trip); no limit on SQLite.
+    fn notify<'a>(&'a self, channel: &str, payload: Value) -> BoxedFuture<'a, Result<()>>;
+
+    /// Listen on `channel` — starts from "now" (no replay); delivers
+    /// opaque [`Wake`](crate::Wake) hints (channel only, never
+    /// payloads).
+    fn listen<'a>(&'a self, channel: &str) -> BoxedFuture<'a, Result<Box<dyn WakeReceiver>>>;
+
+    /// Take the stream-scoped handle for `name`.
+    fn stream<'a>(&'a self, name: &str) -> BoxedFuture<'a, Result<Box<dyn StreamHandle>>>;
+
+    /// Take the queue-scoped handle for `name`, with the `QueueOpts`
+    /// stamps future enqueues resolve over (ADR-010 §3a).
+    fn queue<'a>(&'a self, name: &str, opts: QueueOpts) -> BoxedFuture<'a, Result<Box<dyn Queue>>>;
+
+    /// Take the outbox-scoped handle for `name` (backing queue derived
+    /// engine-side under the reserved prefix).
+    fn outbox<'a>(&'a self, name: &str) -> BoxedFuture<'a, Result<Box<dyn Outbox>>>;
+
+    /// Acquire a named TTL lock, or fail: `None` = someone holds it
+    /// (no-work is a value, not an error). `owner` is consumer-local
+    /// (non-empty only); `ttl` is seconds. Exclusion lapses silently
+    /// at TTL expiry — holders renew within TTL (ADR-008 §7).
+    fn try_lock<'a>(
+        &'a self,
+        name: &str,
+        owner: &str,
+        ttl: i64,
+    ) -> BoxedFuture<'a, Result<Option<Box<dyn Lock>>>>;
+
+    /// Register (upsert by name) a schedule firing into `queue`.
+    /// Both the schedule name and the queue argument validate as
+    /// shared-namespace names at the entry point (ADR-021 §3 — a
+    /// schedule cannot target a reserved name). `spec` grammar v1:
+    /// `@every <n><unit>` only (s|m|h|d) — else `InvalidSpec`
+    /// (ADR-009 §2), rejected before storage. Schedules never fire
+    /// without a consumer running [`Store::run_schedules`].
+    fn schedule<'a>(
+        &'a self,
+        name: &str,
+        spec: &str,
+        queue: &str,
+        payload: Value,
+        opts: ScheduleOpts,
+    ) -> BoxedFuture<'a, Result<Schedule>>;
+
+    /// Remove a schedule registration. `true` = a row was removed;
+    /// `false` = no such schedule (a value, not an error).
+    fn unschedule<'a>(&'a self, name: &str) -> BoxedFuture<'a, Result<bool>>;
+
+    /// The opt-in scheduler runner: acquire the leadership lock (the
+    /// reserved name `__alkstore_scheduler`), loop {renew; fire due
+    /// boundaries; sleep until the next boundary or `stop`}. A
+    /// cancelled [`StopToken`] ends the sleep and returns the clean
+    /// `Ok(())`; leadership loss returns `Err(LeadershipLost)` — the
+    /// caller's recipe is to respawn (ADR-009 §1).
+    fn run_schedules<'a>(&'a self, stop: StopToken) -> BoxedFuture<'a, Result<()>>;
+
+    /// Closure-scoped transaction — a wrapper over [`Store::begin_tx`]
+    /// (provided method; ADR-007, signature pinned 2026-10-07).
+    ///
+    /// Deliberately non-generic (a `Store` trait object must stay
+    /// object-safe — ADR-008 §2): the pinned `FnOnce(&mut dyn TxHandle)
+    /// -> BoxedFuture<Result<()>>` shape crosses dyn-encoded — as
+    /// `Box<dyn FnOnce ...>` — with values returning through the
+    /// closure's captured state; no handle escapes the closure.
+    ///
+    /// Disposition: `Ok` ⇒ commit; `Err`, drop, or panic across the
+    /// closure ⇒ rollback (ADR-021 §4's drop = rollback). `with_tx`
+    /// is the convenience; `begin_tx` is the seam.
+    fn with_tx<'a>(&'a self, f: WithTxClosure<'a>) -> BoxedFuture<'a, Result<()>> {
+        Box::pin(async move {
+            let mut handle = self.begin_tx().await?;
+            match Box::pin(f(&mut *handle)).await {
+                Ok(()) => handle.commit().await,
+                Err(e) => {
+                    drop(handle);
+                    Err(e)
+                }
+            }
+        })
+    }
+}

alkstore/src/stream_handle.rs

@@ -0,0 +1,77 @@
+//! The stream mechanism handle (ADR-019 §1; ADR-015).
+//!
+//! Publish/read/trim ops live on the stream-scoped handle — placement
+//! mirroring the honker-rs starting artifact; the transactional twins
+//! ride the `TxHandle` (ADR-021 §1).
+
+use serde_json::Value;
+
+use crate::Result;
+use crate::event_receiver::EventReceiver;
+use crate::future::BoxedFuture;
+use crate::stream::StreamEvent;
+
+/// The stream-scoped handle: publish, cursor reads, offsets, trim, and
+/// the durable subscription.
+pub trait StreamHandle: Send + Sync {
+    /// The stream name the handle was constructed with.
+    fn name(&self) -> &str;
+
+    /// Append an event to the stream's durable log (auto-commit).
+    /// Returns the assigned offset — engine-assigned, monotone per
+    /// stream, immutable.
+    fn publish<'a>(&'a self, payload: Value) -> BoxedFuture<'a, Result<i64>>;
+
+    /// Append a keyed event — [`StreamHandle::publish`]'s keyed twin.
+    /// The key is carried metadata (stored, round-tripped, no
+    /// engine-enforced behavioral role); the ordering guarantee stays
+    /// global FIFO by offset per stream (ADR-015 §1). A present key
+    /// must be non-empty (`InvalidName`). Returns the assigned
+    /// offset.
+    fn publish_with_key<'a>(
+        &'a self,
+        key: Option<String>,
+        payload: Value,
+    ) -> BoxedFuture<'a, Result<i64>>;
+
+    /// Cursor read: events with `offset > ` cursor, `offset ASC`
+    /// (global FIFO — ADR-015 §4), up to `limit`. A read from a
+    /// trimmed-away region resumes at the trim horizon's first
+    /// remaining row.
+    fn read_since<'a>(
+        &'a self,
+        offset: i64,
+        limit: i64,
+    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>>;
+
+    /// Cursor read from a consumer's stored checkpoint (absent
+    /// consumer = 0).
+    fn read_from_consumer<'a>(
+        &'a self,
+        consumer: &str,
+        limit: i64,
+    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>>;
+
+    /// Save a consumer's checkpoint — always explicit (no
+    /// auto-checkpoint cadence). Monotone: a save below the stored
+    /// checkpoint is a silent no-op (ADR-019 §6); the `Err` arm
+    /// carries `Database` only (ADR-021 §5).
+    fn save_offset<'a>(&'a self, consumer: &str, offset: i64) -> BoxedFuture<'a, Result<()>>;
+
+    /// Inspect a consumer's stored checkpoint; absent consumer = 0
+    /// (the pre-save state, ADR-019 §1).
+    fn get_offset<'a>(&'a self, consumer: &str) -> BoxedFuture<'a, Result<i64>>;
+
+    /// Bounded growth, consumer-invoked: delete events with
+    /// `offset <= horizon` (exact boundary; ADR-015 §5). Surviving
+    /// events keep their offsets (gaps legal, never renumbered); trim
+    /// emits no wake. Returns the deleted-row count.
+    fn trim_to<'a>(&'a self, horizon: i64) -> BoxedFuture<'a, Result<i64>>;
+
+    /// Durable consumption: attach a consumer that reads to the
+    /// current tail, resumes after restart from the stored offset,
+    /// and saves checkpoints explicitly (no `save_every`, no
+    /// save-on-drop). `consumer` is a consumer-local identifier
+    /// (non-empty only).
+    fn subscribe<'a>(&'a self, consumer: &str) -> BoxedFuture<'a, Result<Box<dyn EventReceiver>>>;
+}

alkstore/src/tests.rs

@@ -129,3 +129,15 @@ fn reserved_constants_exact() {
         "__alkstore_listener_reconnected__"
     );
 }
+
+#[test]
+fn scheduler_error_variants() {
+    let e = Error::InvalidSpec {
+        spec: "0 0 * * *".into(),
+    };
+    assert!(e.to_string().contains("0 0 * * *"));
+    assert_eq!(
+        Error::LeadershipLost.to_string(),
+        "scheduler leadership lost"
+    );
+}

alkstore/src/trait_probe_tests.rs

@@ -0,0 +1,397 @@
+//! Object-safety compile probes and signature-conformance checks for
+//! the trait surface (the task's acceptance criterion): a trivial
+//! local impl of every trait must construct as `Box<dyn ...>` —
+//! proving the whole surface is object-safe and shape-complete.
+
+use std::time::Duration;
+
+use serde_json::{Value, json};
+
+use crate::event_receiver::EventReceiver;
+use crate::job::Job;
+use crate::job_handle::JobHandle;
+use crate::lock::Lock;
+use crate::outbox::{Delivery, Outbox};
+use crate::queue::Queue;
+use crate::stream::StreamEvent;
+use crate::stream_handle::StreamHandle;
+use crate::tx::TxHandle;
+use crate::wake_receiver::WakeReceiver;
+use crate::{
+    BoxedFuture, EnqueueOpts, Error, QueueOpts, Result, Schedule, ScheduleOpts, StopToken, Store,
+    Wake,
+};
+
+#[derive(Default)]
+struct ProbeTx;
+
+impl TxHandle for ProbeTx {
+    fn enqueue_tx<'a>(
+        &'a mut self,
+        _queue: &str,
+        _opts: EnqueueOpts,
+        _payload: Value,
+    ) -> BoxedFuture<'a, Result<i64>> {
+        Box::pin(std::future::ready(Ok(1)))
+    }
+    fn publish_tx<'a>(
+        &'a mut self,
+        _stream: &str,
+        _payload: Value,
+    ) -> BoxedFuture<'a, Result<i64>> {
+        Box::pin(std::future::ready(Ok(1)))
+    }
+    fn publish_with_key_tx<'a>(
+        &'a mut self,
+        _stream: &str,
+        _key: Option<String>,
+        _payload: Value,
+    ) -> BoxedFuture<'a, Result<i64>> {
+        Box::pin(std::future::ready(Ok(1)))
+    }
+    fn notify_tx<'a>(&'a mut self, _channel: &str, _payload: Value) -> BoxedFuture<'a, Result<()>> {
+        Box::pin(std::future::ready(Ok(())))
+    }
+    fn save_offset_tx<'a>(
+        &'a mut self,
+        _stream: &str,
+        _consumer: &str,
+        _offset: i64,
+    ) -> BoxedFuture<'a, Result<()>> {
+        Box::pin(std::future::ready(Ok(())))
+    }
+    fn get_job_tx<'a>(
+        &'a mut self,
+        _queue: &str,
+        _job_id: i64,
+    ) -> BoxedFuture<'a, Result<Option<Job>>> {
+        Box::pin(std::future::ready(Ok(None)))
+    }
+    fn get_offset_tx<'a>(
+        &'a mut self,
+        _stream: &str,
+        _consumer: &str,
+    ) -> BoxedFuture<'a, Result<i64>> {
+        Box::pin(std::future::ready(Ok(0)))
+    }
+    fn read_since_tx<'a>(
+        &'a mut self,
+        _stream: &str,
+        _offset: i64,
+        _limit: i64,
+    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
+        Box::pin(std::future::ready(Ok(Vec::new())))
+    }
+    fn read_from_consumer_tx<'a>(
+        &'a mut self,
+        _stream: &str,
+        _consumer: &str,
+        _limit: i64,
+    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
+        Box::pin(std::future::ready(Ok(Vec::new())))
+    }
+    fn outbox_enqueue_tx<'a>(
+        &'a mut self,
+        _outbox: &str,
+        _opts: EnqueueOpts,
+        _payload: Value,
+    ) -> BoxedFuture<'a, Result<i64>> {
+        Box::pin(std::future::ready(Ok(1)))
+    }
+    fn commit(self: Box<Self>) -> BoxedFuture<'static, Result<()>> {
+        Box::pin(std::future::ready(Ok(())))
+    }
+}
+
+struct ProbeJob;
+
+impl JobHandle for ProbeJob {
+    fn job(&self) -> &Job {
+        unreachable!()
+    }
+    fn ack(self: Box<Self>) -> BoxedFuture<'static, Result<bool>> {
+        Box::pin(std::future::ready(Ok(true)))
+    }
+    fn retry(
+        self: Box<Self>,
+        _err: Option<String>,
+        _delay: Option<i64>,
+    ) -> BoxedFuture<'static, Result<bool>> {
+        Box::pin(std::future::ready(Ok(true)))
+    }
+    fn fail(self: Box<Self>, _err: Option<String>) -> BoxedFuture<'static, Result<bool>> {
+        Box::pin(std::future::ready(Ok(true)))
+    }
+    fn heartbeat<'a>(&'a self, _extend: i64) -> BoxedFuture<'a, Result<bool>> {
+        Box::pin(std::future::ready(Ok(true)))
+    }
+}
+
+struct ProbeQueue;
+
+impl Queue for ProbeQueue {
+    fn name(&self) -> &str {
+        "probe"
+    }
+    fn enqueue<'a>(&'a self, _payload: Value, _opts: EnqueueOpts) -> BoxedFuture<'a, Result<i64>> {
+        Box::pin(std::future::ready(Ok(1)))
+    }
+    fn claim_one<'a>(
+        &'a self,
+        _worker_id: &str,
+    ) -> BoxedFuture<'a, Result<Option<Box<dyn JobHandle>>>> {
+        Box::pin(std::future::ready(Ok(None)))
+    }
+    fn claim_batch<'a>(
+        &'a self,
+        _worker_id: &str,
+        _n: i64,
+    ) -> BoxedFuture<'a, Result<Vec<Box<dyn JobHandle>>>> {
+        Box::pin(std::future::ready(Ok(Vec::new())))
+    }
+    fn ack_batch<'a>(&'a self, _ids: &[i64]) -> BoxedFuture<'a, Result<i64>> {
+        Box::pin(std::future::ready(Ok(0)))
+    }
+    fn cancel<'a>(&'a self, _job_id: i64) -> BoxedFuture<'a, Result<bool>> {
+        Box::pin(std::future::ready(Ok(false)))
+    }
+    fn get_job<'a>(&'a self, _job_id: i64) -> BoxedFuture<'a, Result<Option<Job>>> {
+        Box::pin(std::future::ready(Ok(None)))
+    }
+    fn sweep_expired<'a>(&'a self) -> BoxedFuture<'a, Result<i64>> {
+        Box::pin(std::future::ready(Ok(0)))
+    }
+}
+
+struct ProbeStream;
+
+impl StreamHandle for ProbeStream {
+    fn name(&self) -> &str {
+        "probe"
+    }
+    fn publish<'a>(&'a self, _payload: Value) -> BoxedFuture<'a, Result<i64>> {
+        Box::pin(std::future::ready(Ok(1)))
+    }
+    fn publish_with_key<'a>(
+        &'a self,
+        _key: Option<String>,
+        _payload: Value,
+    ) -> BoxedFuture<'a, Result<i64>> {
+        Box::pin(std::future::ready(Ok(1)))
+    }
+    fn read_since<'a>(
+        &'a self,
+        _offset: i64,
+        _limit: i64,
+    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
+        Box::pin(std::future::ready(Ok(Vec::new())))
+    }
+    fn read_from_consumer<'a>(
+        &'a self,
+        _consumer: &str,
+        _limit: i64,
+    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
+        Box::pin(std::future::ready(Ok(Vec::new())))
+    }
+    fn save_offset<'a>(&'a self, _consumer: &str, _offset: i64) -> BoxedFuture<'a, Result<()>> {
+        Box::pin(std::future::ready(Ok(())))
+    }
+    fn get_offset<'a>(&'a self, _consumer: &str) -> BoxedFuture<'a, Result<i64>> {
+        Box::pin(std::future::ready(Ok(0)))
+    }
+    fn trim_to<'a>(&'a self, _horizon: i64) -> BoxedFuture<'a, Result<i64>> {
+        Box::pin(std::future::ready(Ok(0)))
+    }
+    fn subscribe<'a>(&'a self, _consumer: &str) -> BoxedFuture<'a, Result<Box<dyn EventReceiver>>> {
+        Box::pin(std::future::ready(Ok(
+            Box::new(ProbeEventReceiver) as Box<dyn EventReceiver>
+        )))
+    }
+}
+
+struct ProbeEventReceiver;
+
+impl EventReceiver for ProbeEventReceiver {
+    fn recv<'a>(&'a mut self) -> BoxedFuture<'a, Option<Result<StreamEvent>>> {
+        Box::pin(std::future::ready(None))
+    }
+    fn try_recv(&mut self) -> Result<Option<StreamEvent>> {
+        Ok(None)
+    }
+    fn read_since<'a>(
+        &'a mut self,
+        _offset: i64,
+        _limit: i64,
+    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
+        Box::pin(std::future::ready(Ok(Vec::new())))
+    }
+    fn save_offset(&mut self) -> Result<()> {
+        Ok(())
+    }
+    fn offset(&self) -> i64 {
+        0
+    }
+}
+
+struct ProbeOutbox;
+
+impl Outbox for ProbeOutbox {
+    fn name(&self) -> &str {
+        "probe"
+    }
+    fn enqueue<'a>(&'a self, _payload: Value, _opts: EnqueueOpts) -> BoxedFuture<'a, Result<i64>> {
+        Box::pin(std::future::ready(Ok(1)))
+    }
+    fn run_once<'a>(
+        &'a mut self,
+        _worker_id: &str,
+        _delivery: &'a mut dyn Delivery,
+    ) -> BoxedFuture<'a, Result<bool>> {
+        Box::pin(std::future::ready(Ok(false)))
+    }
+}
+
+struct ProbeLock;
+
+impl Lock for ProbeLock {
+    fn name(&self) -> &str {
+        "probe"
+    }
+    fn renew<'a>(&'a self, _ttl: i64) -> BoxedFuture<'a, Result<bool>> {
+        Box::pin(std::future::ready(Ok(true)))
+    }
+    fn release(self: Box<Self>) -> BoxedFuture<'static, Result<bool>> {
+        Box::pin(std::future::ready(Ok(true)))
+    }
+}
+
+struct ProbeWakeReceiver;
+
+impl WakeReceiver for ProbeWakeReceiver {
+    fn recv<'a>(&'a mut self) -> BoxedFuture<'a, Option<Wake>> {
+        Box::pin(std::future::ready(None))
+    }
+    fn try_recv(&mut self) -> Result<Option<Wake>> {
+        Err(Error::Closed)
+    }
+    fn recv_timeout<'a>(&'a mut self, _timeout: Duration) -> BoxedFuture<'a, Result<Option<Wake>>> {
+        Box::pin(std::future::ready(Ok(None)))
+    }
+}
+
+struct ProbeStore;
+
+impl Store for ProbeStore {
+    fn begin_tx(&self) -> BoxedFuture<'_, Result<Box<dyn TxHandle + Send>>> {
+        Box::pin(std::future::ready(Ok(
+            Box::new(ProbeTx) as Box<dyn TxHandle + Send>
+        )))
+    }
+    fn notify<'a>(&'a self, _channel: &str, _payload: Value) -> BoxedFuture<'a, Result<()>> {
+        Box::pin(std::future::ready(Ok(())))
+    }
+    fn listen<'a>(&'a self, _channel: &str) -> BoxedFuture<'a, Result<Box<dyn WakeReceiver>>> {
+        Box::pin(std::future::ready(Ok(
+            Box::new(ProbeWakeReceiver) as Box<dyn WakeReceiver>
+        )))
+    }
+    fn stream<'a>(&'a self, _name: &str) -> BoxedFuture<'a, Result<Box<dyn StreamHandle>>> {
+        Box::pin(std::future::ready(Ok(
+            Box::new(ProbeStream) as Box<dyn StreamHandle>
+        )))
+    }
+    fn queue<'a>(
+        &'a self,
+        _name: &str,
+        _opts: QueueOpts,
+    ) -> BoxedFuture<'a, Result<Box<dyn Queue>>> {
+        Box::pin(std::future::ready(Ok(
+            Box::new(ProbeQueue) as Box<dyn Queue>
+        )))
+    }
+    fn outbox<'a>(&'a self, _name: &str) -> BoxedFuture<'a, Result<Box<dyn Outbox>>> {
+        Box::pin(std::future::ready(Ok(
+            Box::new(ProbeOutbox) as Box<dyn Outbox>
+        )))
+    }
+    fn try_lock<'a>(
+        &'a self,
+        _name: &str,
+        _owner: &str,
+        _ttl: i64,
+    ) -> BoxedFuture<'a, Result<Option<Box<dyn Lock>>>> {
+        Box::pin(std::future::ready(Ok(None)))
+    }
+    fn schedule<'a>(
+        &'a self,
+        _name: &str,
+        _spec: &str,
+        _queue: &str,
+        _payload: Value,
+        _opts: ScheduleOpts,
+    ) -> BoxedFuture<'a, Result<Schedule>> {
+        Box::pin(std::future::ready(Ok(Schedule {
+            name: "probe".into(),
+            spec: "@every 60s".into(),
+            queue: "q".into(),
+            opts: ScheduleOpts::default(),
+        })))
+    }
+    fn unschedule<'a>(&'a self, _name: &str) -> BoxedFuture<'a, Result<bool>> {
+        Box::pin(std::future::ready(Ok(false)))
+    }
+    fn run_schedules<'a>(&'a self, _stop: StopToken) -> BoxedFuture<'a, Result<()>> {
+        Box::pin(std::future::ready(Ok(())))
+    }
+}
+
+#[test]
+fn trait_objects_construct() {
+    let _tx: Box<dyn TxHandle + Send> = Box::new(ProbeTx);
+    let _job: Box<dyn JobHandle> = Box::new(ProbeJob);
+    let _queue: Box<dyn Queue> = Box::new(ProbeQueue);
+    let _stream: Box<dyn StreamHandle> = Box::new(ProbeStream);
+    let _receiver: Box<dyn EventReceiver> = Box::new(ProbeEventReceiver);
+    let _outbox: Box<dyn Outbox> = Box::new(ProbeOutbox);
+    let _lock: Box<dyn Lock> = Box::new(ProbeLock);
+    let _wake_receiver: Box<dyn WakeReceiver> = Box::new(ProbeWakeReceiver);
+    let _store: Box<dyn Store> = Box::new(ProbeStore);
+}
+
+#[tokio::test]
+async fn with_tx_commit_and_rollback_dispositions() {
+    let store: Box<dyn Store> = Box::new(ProbeStore);
+
+    let committed = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+    let flag = committed.clone();
+    store
+        .with_tx(Box::new(move |_tx| {
+            let flag = flag.clone();
+            Box::pin(async move {
+                flag.store(true, std::sync::atomic::Ordering::SeqCst);
+                Ok(())
+            }) as BoxedFuture<'_, Result<()>>
+        }))
+        .await
+        .unwrap_or_else(|e| panic!("with_tx commit failed: {e}"));
+    assert!(committed.load(std::sync::atomic::Ordering::SeqCst));
+
+    let rolled_back = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+    let flag = rolled_back.clone();
+    let r = store
+        .with_tx(Box::new(move |tx| {
+            let flag = flag.clone();
+            Box::pin(async move {
+                flag.store(true, std::sync::atomic::Ordering::SeqCst);
+                let _ = tx
+                    .enqueue_tx("q", EnqueueOpts::default(), json!({}))
+                    .await?;
+                Err::<(), _>(Error::InvalidName {
+                    name: "probe".into(),
+                })
+            }) as BoxedFuture<'_, Result<()>>
+        }))
+        .await;
+    assert!(matches!(r, Err(Error::InvalidName { .. })));
+    assert!(rolled_back.load(std::sync::atomic::Ordering::SeqCst));
+}

alkstore/src/tx.rs

@@ -0,0 +1,144 @@
+//! The transactional seam's caller-held handle (ADR-007, ADR-008 §2,
+//! ADR-014, ADR-015 §2, ADR-021 §1/§4).
+//!
+//! The `*_tx` methods make side effects commit-atomic with a business
+//! write: every op lands in *the caller's transaction*. The trait is
+//! core-owned — engines implement it for their concrete handle; no
+//! downcast, no enum, no generic method (object safety, ADR-008 §2).
+
+use serde_json::Value;
+
+use crate::future::BoxedFuture;
+use crate::opts::EnqueueOpts;
+use crate::{Job, Result, StreamEvent};
+
+/// The caller-held transaction handle.
+///
+/// Engine shape notes, not contract: the Postgres handle holds the
+/// pooled client directly (straight `.await`s); the SQLite handle is a
+/// writer-slot lease whose ops round-trip `spawn_blocking` (ADR-003,
+/// ADR-004).
+///
+/// # Entry-point validation obligation
+///
+/// Every name-bearing method (`enqueue_tx`, `publish_tx`,
+/// `publish_with_key_tx`, `notify_tx`, `save_offset_tx`,
+/// `get_job_tx`/`get_offset_tx`/`read_since_tx`/
+/// `read_from_consumer_tx`, `outbox_enqueue_tx`) validates its names
+/// before any engine round trip — identically on the auto-commit and
+/// tx paths: non-empty (else `InvalidName`), reserved-prefix rejected
+/// for shared-namespace kinds (else `ReservedName`). Validation must
+/// not be bypassable by the commit-atomic path (ADR-008 §4).
+///
+/// # Drop = rollback
+///
+/// A handle dropped without `commit` rolls back (ADR-021 §4): the
+/// no-ghosts property holds through RAII paths, not only the explicit
+/// ones. There is no `rollback` method — dropping *is* the rollback;
+/// a caller wanting an explicit rollback drops the handle.
+pub trait TxHandle: Send {
+    /// Enqueue a job into `queue`, commit-atomic with the caller's
+    /// business write. Returns the job id.
+    ///
+    /// No-handle-open shape: the row stamps the engine's derived
+    /// default `QueueOpts` (300/3/5/none, ADR-020 §3) — a `TxHandle`
+    /// carries no queue opts. Validates `queue` as a shared-namespace
+    /// name (`InvalidName`/`ReservedName`) before any round trip.
+    fn enqueue_tx<'a>(
+        &'a mut self,
+        queue: &str,
+        opts: EnqueueOpts,
+        payload: Value,
+    ) -> BoxedFuture<'a, Result<i64>>;
+
+    /// Publish an event to `stream`, commit-atomic. Returns the
+    /// assigned offset. Validates `stream` as a shared-namespace name.
+    fn publish_tx<'a>(&'a mut self, stream: &str, payload: Value) -> BoxedFuture<'a, Result<i64>>;
+
+    /// Publish a keyed event to `stream` — [`TxHandle::publish_tx`]'s
+    /// keyed twin; commit-atomic keyed publishes are the point
+    /// (ADR-015 §2). The key is carried metadata; a present key must
+    /// be non-empty (`InvalidName`, ADR-008 §5 annotated). Returns the
+    /// assigned offset.
+    fn publish_with_key_tx<'a>(
+        &'a mut self,
+        stream: &str,
+        key: Option<String>,
+        payload: Value,
+    ) -> BoxedFuture<'a, Result<i64>>;
+
+    /// Deliver a fire-and-forget notify, commit-atomic — delivered
+    /// only at commit; rollback drops it (ADR-007). Validates
+    /// `channel`; may fail with `PayloadTooLarge` pg-side
+    /// (client-side check before the round trip, the contract-wide
+    /// matchable engine asymmetry).
+    fn notify_tx<'a>(&'a mut self, channel: &str, payload: Value) -> BoxedFuture<'a, Result<()>>;
+
+    /// Save a stream consumer's checkpoint, commit-atomic. Saves are
+    /// monotone — a save below the stored checkpoint is a silent
+    /// no-op (ADR-019 §6); the `Err` arm carries `Database` only
+    /// (ADR-021 §5). Validates `stream`; `consumer` is a
+    /// consumer-local identifier (non-empty only).
+    fn save_offset_tx<'a>(
+        &'a mut self,
+        stream: &str,
+        consumer: &str,
+        offset: i64,
+    ) -> BoxedFuture<'a, Result<()>>;
+
+    /// Read a job row inside the caller's transaction — reads the
+    /// transaction's own writes (read-your-own-writes, ADR-021 §1).
+    /// Sees dead rows; a pure read, no ops ride it. Validates `queue`.
+    fn get_job_tx<'a>(
+        &'a mut self,
+        queue: &str,
+        job_id: i64,
+    ) -> BoxedFuture<'a, Result<Option<Job>>>;
+
+    /// Read a consumer's stored checkpoint inside the caller's
+    /// transaction; absent consumer = 0 (ADR-019 §1). Pure read.
+    /// Validates `stream`.
+    fn get_offset_tx<'a>(
+        &'a mut self,
+        stream: &str,
+        consumer: &str,
+    ) -> BoxedFuture<'a, Result<i64>>;
+
+    /// Read stream events from `offset`, inside the caller's
+    /// transaction (`offset ASC`, global FIFO — ADR-015 §4). Pure
+    /// read. Validates `stream`.
+    fn read_since_tx<'a>(
+        &'a mut self,
+        stream: &str,
+        offset: i64,
+        limit: i64,
+    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>>;
+
+    /// Read stream events from a consumer's stored checkpoint, inside
+    /// the caller's transaction. Pure read. Validates `stream`.
+    fn read_from_consumer_tx<'a>(
+        &'a mut self,
+        stream: &str,
+        consumer: &str,
+        limit: i64,
+    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>>;
+
+    /// Enqueue into an outbox's derived backing queue, commit-atomic
+    /// (ADR-014). Takes the *outbox* name — validated like
+    /// `store.outbox(name)` — and derives the queue engine-side; the
+    /// derived name (reserved-prefixed) is unreachable by
+    /// [`TxHandle::enqueue_tx`] by design, so this is the only
+    /// transactional path into the backing queue. Stamps the outbox's
+    /// derived `QueueOpts` set (60/5/5).
+    fn outbox_enqueue_tx<'a>(
+        &'a mut self,
+        outbox: &str,
+        opts: EnqueueOpts,
+        payload: Value,
+    ) -> BoxedFuture<'a, Result<i64>>;
+
+    /// Commit the transaction, consuming the handle. The alternative
+    /// disposition is drop = rollback. One-shot: callers re-`begin_tx`
+    /// for the next transaction.
+    fn commit(self: Box<Self>) -> BoxedFuture<'static, Result<()>>;
+}

alkstore/src/wake_receiver.rs

@@ -0,0 +1,38 @@
+//! The wake receiver (ADR-008 §3, §8): the `listen(channel)` return.
+//!
+//! Wakes are best-effort hints, not per-commit delivery promises —
+//! possibly coalesced (SQLite) or per-notify (Postgres), possibly
+//! repeated after reconnect, possibly absent entirely — recovery is
+//! the consumer's re-read, never a guaranteed deliver. Consumers must
+//! be idempotent on wake.
+
+use std::time::Duration;
+
+use crate::Result;
+use crate::future::BoxedFuture;
+use crate::wake::Wake;
+
+/// The wake receiver: the durable-subscription-free listen side of
+/// the notify mechanism.
+///
+/// Close semantics (ADR-008 §3): `recv() -> None` means closed
+/// (source death — SQLite's watcher death; terminal). The synthetic
+/// reconnect-wake (the reserved channel) covers Postgres connection
+/// gaps and arrives as a normal [`Wake`].
+pub trait WakeReceiver: Send {
+    /// Wait for the next wake. `None` = the source closed (terminal;
+    /// a closed receiver never reopens — recovery is a fresh
+    /// `listen`).
+    fn recv<'a>(&'a mut self) -> BoxedFuture<'a, Option<Wake>>;
+
+    /// Poll without waiting. `Ok(None)` = no wake pending right now
+    /// (idle); `Err(Closed)` = the source closed. Closed and idle are
+    /// distinguishable states.
+    fn try_recv(&mut self) -> Result<Option<Wake>>;
+
+    /// Bounded wait for the next wake. Timeout expiry without a wake
+    /// returns `Ok(None)` — idle is the documented `Ok(None)` arm;
+    /// the `Err` arm carries `Database` (storage failure) or `Closed`
+    /// (source death), never a timeout signal (ADR-008 §3).
+    fn recv_timeout<'a>(&'a mut self, timeout: Duration) -> BoxedFuture<'a, Result<Option<Wake>>>;
+}

tasks/core-trait-surface.md

@@ -1,7 +1,7 @@
 ---
 id: core-trait-surface
 name: Core trait surface (Store, TxHandle, mechanism handles, receivers)
-status: pending
+status: completed
 depends_on: [core-value-types]
 scope: moderate
 risk: medium
@@ -87,17 +87,17 @@ drop=rollback, no-claim_tx absence rationale).
 
 ## Acceptance Criteria
 
-- [ ] Every trait above with the exact pinned method sets and
+- [x] Every trait above with the exact pinned method sets and
       signatures; compiles as object-safe (`Box<dyn ...>` everywhere
       the contract says)
-- [ ] `with_tx` provided method present with the pinned signature
+- [x] `with_tx` provided method present with the pinned signature
-- [ ] A compile-probe test constructs `Box<dyn TxHandle>`,
+- [x] A compile-probe test constructs `Box<dyn TxHandle>`,
       `Box<dyn Queue>`, etc. from a trivial local impl, proving object
       safety of the whole surface
-- [ ] Trait docs state: entry-point validation obligation, drop =
+- [x] Trait docs state: entry-point validation obligation, drop =
       rollback, the uniform handle-op validity predicate reference,
       heartbeat's absolute-reset semantics, no `claim_tx` rationale
-- [ ] `cargo test -p alkstore`, clippy `-D warnings`, fmt clean
+- [x] `cargo test -p alkstore`, clippy `-D warnings`, fmt clean
 
 ## References
 
@@ -110,8 +110,73 @@ drop=rollback, no-claim_tx absence rationale).
 
 ## Notes
 
-> To be filled by implementation agent
+- **Async posture**: the "desugared boxed-future form" (ADR-008 §2) is
+  realized as a named alias, `BoxedFuture<'a, T> =
+  Pin<Box<dyn Future<Output = T> + Send + 'a>>` in
+  `src/future.rs` — no `async_trait` macro, no core dependency beyond
+  the pinned posture (Send bounds the futures: `begin_tx` returns
+  `Box<dyn TxHandle + Send>`, so its futures must be equally
+  portable). Sync methods (`try_recv`/`save_offset`/`offset` on
+  receivers, `job(&self)`, `name(&self)`) stay sync as pinned.
+- **`with_tx` dyn encoding**: the pinned signature is
+  non-generic — a `Store` trait object must stay object-safe — so the
+  generic-`F` form does not compile on the trait; the pinned
+  `FnOnce(&mut dyn TxHandle) -> BoxedFuture<Result<()>>` shape crosses
+  dyn-encoded as `Box<dyn FnOnce ...>`, aliased `WithTxClosure<'a>`
+  (re-exported) to keep the complex type out of signatures. Values
+  return through captured state; the provided default body commits on
+  `Ok`, drops (rolls back) on `Err`.
+- **Delivery callable encoding** (ADR-019 §5's implementer's choice):
+  the named one-method trait object — `trait Delivery { fn
+  deliver(&mut self, Box<dyn JobHandle>) -> BoxedFuture<Result<()>>;
+  }` — with `run_once(worker_id, delivery: &mut dyn Delivery)`.
+  `&mut dyn FnMut` cannot return boxed futures of borrowed data under
+  object safety, so the named-trait form is the workable answer; the
+  engine acks on the closure's `Ok` and retries on its `Err`.
+- **Fallibility**: the ADR sketches elide `Result<>` for brevity
+  (ADR-008 §2's note: every fallible op returns the §5 taxonomy);
+  entry points validating names must err, so constructors
+  (`stream`/`queue`/`outbox`/`listen`) are `Result`-returning as is
+  `notify` (PayloadTooLarge); `begin_tx` returns `Result` (SQLite
+  writer-slot acquisition can fail); no-claim and no-lock outcomes
+  stay values (`Option`), handle-op outcomes stay `bool`s per
+  ADR-008 §5's act-differently rule.
+- **Scheduler error variants** (`InvalidSpec { spec }`,
+  `LeadershipLost` — ADR-009 §6) joined `Error` here per
+  `core-errors-and-validation`'s note (exactly-six list was that
+  task's scope; their signatures pin in this task). Unit-tested.
+- **`tokio` added as a *dev*-dependency only** (test to drive the
+  boxed-future probes) — runtime deps remain exactly
+  serde/serde_json/thiserror (ADR-020 §4, ADR-001: no driver deps in
+  core).
+- `Wake`'s `#[non_exhaustive]` prevented the probe's struct-literal
+  construction — the probe re-exports or constructs nothing of the
+  read types beyond the existing value-type tests; trait-object
+  construction is the probe's whole point.
+- Every trait doc carries the pinned semantics text: entry-point
+  validation obligations (Store; TxHandle's own mirror — name-bearing
+  `*_tx` methods validate like their auto-commit counterparts),
+  drop=rollback (TxHandle), no-`claim_tx` rationale (Queue), the
+  uniform validity predicate + absolute-reset heartbeat (JobHandle),
+  monotone saves (StreamHandle/EventReceiver), terminal-close arms
+  and the deliberate no-`recv_timeout` absence (receivers).
 
 ## Summary
 
-> To be filled on completion
+The core trait surface — contract v1 in code — is implemented and
+re-exported from `alkstore/src/lib.rs`: `BoxedFuture` +
+`TxHandle` (the eleven pinned methods incl. ADR-021 §1's four tx
+reads, `commit(self: Box<Self>)`, drop=rollback docs);
+`Store` (the pinned method set + the `with_tx` provided method in the
+pinned dyn-encoded signature, with commit/rollback dispositions —
+unit-tested); mechanism handles `Queue`/`JobHandle`/`StreamHandle`/
+`Outbox` (+ `Delivery`)/`Lock` with the exact ADR-019 §1/§3/§5 method
+sets; receivers `WakeReceiver` (timeout expiry = `Ok(None)`; `Err` =
+Database/Closed only) and `EventReceiver` (no `recv_timeout`,
+deliberate); `Error::InvalidSpec`/`LeadershipLost` added (ADR-009
+§6). Object-safety compile probes construct every trait object from
+trivial local impls (`trait_objects_construct`), and the `with_tx`
+dispositions run against a probe store (`with_tx_commit_and_
+rollback_dispositions`) — 23 tests green; `cargo test -p alkstore`,
+workspace `cargo build`, clippy `-D warnings` (workspace), and
+`cargo fmt --check` all clean.