SQLite engine: transactional seam — begin_tx, writer-slot lease, all eleven *_tx methods (task sqlite-engine-seam-tx)

alkstore-sqlite/src/lib.rs

@@ -18,9 +18,12 @@
 //! `Send`-across-await; the family REQ-TTY-01 posture).
 
 mod opts;
+mod resolution;
 mod seam;
 mod store;
 mod substrate;
+mod tx;
 
 pub use opts::{DEFAULT_MAX_READERS, SqliteOpts};
 pub use store::{SqliteStore, open};
+pub use tx::SqliteTxHandle;

alkstore-sqlite/src/resolution.rs

@@ -0,0 +1,87 @@
+//! The engine-side opts-resolution arithmetic (ADR-020 §1–§3) and the
+//! derived stamp sets — the shared formulas the tx seam stamps rows
+//! with and the auto-commit mechanism tasks reuse one owner per
+//! formula (ADR-012 §2 — resolution lives engine-side, the substrate
+//! takes primitives).
+
+use alkstore::{EnqueueOpts, QueueOpts};
+use rusqlite::Connection;
+
+use crate::substrate::Stamps;
+
+/// The engine's second-precision clock — the single `unixepoch()` read
+/// the resolution and row write share (ADR-020 §1: "one enqueue, one
+/// clock read").
+///
+/// Mirrors the substrate's own `now_unix` shape; the mapping into the
+/// contract taxonomy is engine-side (the substrate is contract-blind).
+pub(crate) fn now_unix(conn: &Connection) -> alkstore::Result<i64> {
+    conn.query_row("SELECT unixepoch()", [], |r| r.get(0))
+        .map_err(crate::seam::sqlite_error)
+}
+
+/// The plain queue's derived `QueueOpts` defaults (300/3/5/none,
+/// honker parities — ADR-010 §3) — the stamp source for the
+/// no-handle-open shapes (`enqueue_tx`, scheduler boundary fires,
+/// ADR-020 §3 / §3a note).
+///
+/// Named (not an inline `QueueOpts::default()` call) so the "derived
+/// defaults" formula has one owner: a future default change lands
+/// here, once, engine-wide.
+pub(crate) fn plain_queue_default_stamps() -> Stamps {
+    let QueueOpts {
+        visibility_timeout_s,
+        max_attempts,
+        backoff_base_s,
+        dead_letter_retention_s,
+    } = QueueOpts::default();
+    Stamps {
+        max_attempts,
+        visibility_timeout_s,
+        backoff_base_s,
+        dead_letter_retention_s,
+    }
+}
+
+/// The outbox backing queue's derived `QueueOpts` set (60/5/5 — ADR-010
+/// §3, visibility 60 s, max_attempts 5, backoff base 5 s, retention
+/// forever). The stamp source for `outbox_enqueue_tx` /
+/// `Outbox::enqueue`.
+pub(crate) fn outbox_default_stamps() -> Stamps {
+    Stamps {
+        visibility_timeout_s: 60,
+        max_attempts: 5,
+        backoff_base_s: 5,
+        dead_letter_retention_s: None,
+    }
+}
+
+/// The derived backing-queue name for an outbox (the reserved
+/// namespace — ADR-008 §4). Engine-internal derivation: the reserved
+/// prefix makes the name unreachable by `enqueue_tx` by design
+/// (ADR-014 §1) — outbox enqueue is the only path into it.
+pub(crate) fn outbox_backing_queue_name(outbox: &str) -> String {
+    format!("{}outbox:{outbox}", alkstore::RESERVED_PREFIX)
+}
+
+/// Resolve [`EnqueueOpts`] against the enqueue instant: the row's ready
+/// time (`delay` → `now + delay`, wins over `run_at` (absolute);
+/// neither → `now`) and the absolute row expiry (`expires` relative
+/// seconds from enqueue; `None` = never) — ADR-020 §1/§2.
+pub(crate) struct ResolvedEnqueue {
+    pub(crate) run_at: i64,
+    pub(crate) expires_at: Option<i64>,
+}
+
+pub(crate) fn resolve_enqueue_opts(
+    conn: &Connection,
+    opts: &EnqueueOpts,
+) -> alkstore::Result<ResolvedEnqueue> {
+    let now = now_unix(conn)?;
+    let run_at = match opts.delay {
+        Some(delay) => now.saturating_add(delay),
+        None => opts.run_at.unwrap_or(now),
+    };
+    let expires_at = opts.expires.map(|s| now.saturating_add(s));
+    Ok(ResolvedEnqueue { run_at, expires_at })
+}

alkstore-sqlite/src/seam.rs

@@ -15,10 +15,7 @@ use alkstore::Error;
 
 /// The blocking bridge every trait method's substrate round trip goes
 /// through. Private: the seam is engine-internal — consumers meet it
-/// only through the async trait surface. `cfg(test)`-gated until the
-/// seam task (`sqlite-engine-seam-tx`) wires the trait impls; the
-/// helper's shape and error path are pinned by this wave's smoke test.
-#[cfg(test)]
+/// only through the async trait surface.
 pub(crate) async fn blocking<T, F>(f: F) -> Result<T, Error>
 where
     T: Send + 'static,

alkstore-sqlite/src/store.rs

@@ -112,7 +112,12 @@ impl Store for SqliteStore {
     fn begin_tx(
         &self,
     ) -> alkstore::BoxedFuture<'_, alkstore::Result<Box<dyn alkstore::TxHandle + Send>>> {
-        Box::pin(async { Err(database_error("begin_tx wiring lands with the seam task")) })
+        let writer = self.writer.clone();
+        let db_path = self.db_path.to_string_lossy().into_owned();
+        let reopen: std::sync::Arc<
+            dyn Fn() -> alkstore::Result<rusqlite::Connection> + Send + Sync,
+        > = std::sync::Arc::new(move || crate::substrate::open_writer_connection(&db_path));
+        crate::tx::SqliteTxHandle::begin(writer, reopen)
     }
 
     fn notify<'a>(
@@ -218,3 +223,5 @@ impl Store for SqliteStore {
 
 #[cfg(test)]
 mod open_tests;
+#[cfg(test)]
+mod tx_tests;

alkstore-sqlite/src/substrate/mod.rs

@@ -40,10 +40,10 @@
 //! ops (`fork-rederive-queue-ops`, register D-12) live in
 //! `queue_ops.rs` — owned contract-v1 code, not lineage body.
 //!
-//! Port state (wave 2): the machinery below is ported but not yet
-//! wired by the engine layer (wave 3 wires `open`, the trait impl, the
-//! seam). A subset of the ported surface is therefore unreachable
-//! until that wiring lands.
+//! Port state (wave 2): the machinery below is ported; wave 3 wires
+//! `open` and the trait/seam impl. A subset of the ported surface
+//! remains unreachable until the mechanism tasks wire it (the
+//! `#![allow(dead_code)]` posture stays until the last wiring task).
 
 #![allow(dead_code)]
 #![allow(unused_imports)]
@@ -53,8 +53,27 @@ mod queue_ops;
 mod schema;
 mod watcher;
 
+pub(crate) use ops::{
+    arg_i64, arg_opt_i64, attach_alkstore_functions, in_savepoint, lock_acquire, lock_release,
+    lock_renew, stream_get_offset, stream_publish, stream_read_since, stream_save_offset,
+};
+pub(crate) use queue_ops::{
+    FireOpts, Stamps, ack, ack_batch, cancel, claim_batch, enqueue, fail, get_job, heartbeat,
+    queue_next_claim_at, retry, scheduler_register, scheduler_soonest, scheduler_tick,
+    scheduler_unregister, sweep_expired,
+};
 pub(crate) use schema::{
     Error as SchemaError, Readers, Writer, apply_default_pragmas, attach_notify, bootstrap_schema,
     open_conn, open_conn_bootstrapped,
 };
+
+/// Provision a fresh connection into the writer slot after a
+/// cancellation-path connection was consumed. Private helper over the
+/// substrate's bootstrapped open; the engine's tx handle captures it
+/// as its `reopen` closure's body.
+pub(crate) fn open_writer_connection(path: &str) -> Result<rusqlite::Connection, alkstore::Error> {
+    open_conn_bootstrapped(path).map_err(|e| match e {
+        SchemaError::Sqlite(inner) => alkstore::Error::database(inner),
+    })
+}
 pub(crate) use watcher::{DEFAULT_WATCHER_POLL_INTERVAL, SharedUpdateWatcher, WatcherConfig};

alkstore-sqlite/src/tx.rs

@@ -0,0 +1,526 @@
+//! The transactional seam's SQLite arm — `begin_tx`, the
+//! [`SqliteTxHandle`] (the writer-slot lease, ADR-007), and all eleven
+//! `*_tx` methods (`sqlite-engine-seam-tx`).
+//!
+//! `begin_tx` acquires the writer slot and opens `BEGIN IMMEDIATE` on
+//! the acquired connection; the handle owns connection + slot. Every
+//! op round-trips the `spawn_blocking` seam to that connection —
+//! holding the handle across `await` points parks the writer, which
+//! is the design (WAL single-writer, honest).
+//!
+//! Drop = rollback (ADR-021 §4): dropping the handle without commit
+//! issues `ROLLBACK` (ignoring failure — no panic may escape drop) and
+//! releases the writer slot; a panic unwinding through any scope
+//! holding the handle reaches the same rollback. The no-ghosts
+//! property holds through the RAII path.
+//!
+//! Errors: substrate errors map to
+//! [`Error::Database`](alkstore::Error::Database) with the source chain
+//! preserved (ADR-008 §5); `encode_payload`'s typed `Codec` error
+//! propagates from the enqueue/publish paths (ADR-023 §1). Every
+//! name-bearing method validates at the entry point, before any round
+//! trip (ADR-008 §4) — the commit-atomic path cannot bypass it.
+
+use serde_json::Value;
+use std::sync::Arc;
+
+use alkstore::{BoxedFuture, EnqueueOpts, Error, Job, JobState, Result, StreamEvent, TxHandle};
+use alkstore::{validate_local_name, validate_shared_name};
+
+use crate::resolution::{
+    outbox_backing_queue_name, outbox_default_stamps, plain_queue_default_stamps,
+    resolve_enqueue_opts,
+};
+use crate::seam::{blocking, sqlite_error};
+use crate::substrate::{
+    Stamps, Writer, enqueue, get_job, stream_get_offset, stream_publish, stream_read_since,
+    stream_save_offset,
+};
+
+/// The caller-held SQLite transaction handle: the writer-slot lease.
+/// Constructed only through [`crate::SqliteStore`]'s `begin_tx` —
+/// never held across an engine `Store` trait object (the trait is what
+/// consumers see).
+pub struct SqliteTxHandle {
+    conn: Option<rusqlite::Connection>,
+    writer: Arc<Writer>,
+    reopen: Arc<dyn Fn() -> Result<rusqlite::Connection> + Send + Sync>,
+}
+
+impl std::fmt::Debug for SqliteTxHandle {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("SqliteTxHandle")
+            .field("open", &self.conn.is_some())
+            .finish_non_exhaustive()
+    }
+}
+
+impl SqliteTxHandle {
+    pub(crate) fn begin(
+        writer: Arc<Writer>,
+        reopen: Arc<dyn Fn() -> Result<rusqlite::Connection> + Send + Sync>,
+    ) -> BoxedFuture<'static, Result<Box<dyn TxHandle + Send>>> {
+        Box::pin(async move {
+            let writer2 = writer.clone();
+            let conn = blocking(move || {
+                let conn = writer2.acquire().ok_or_else(|| {
+                    alkstore::Error::database(std::io::Error::other(
+                        "the store is closed: writer slot unavailable",
+                    ))
+                })?;
+                conn.execute_batch("BEGIN IMMEDIATE")
+                    .map_err(sqlite_error)
+                    .map(|_| conn)
+            })
+            .await?;
+            Ok(Box::new(SqliteTxHandle {
+                conn: Some(conn),
+                writer,
+                reopen,
+            }) as _)
+        })
+    }
+
+    /// The private bridge: every op routes through the same connection
+    /// via `spawn_blocking`. The connection crosses the seam inside an
+    /// RAII [`ConnLease`]: a future cancelled mid-op drops the lease,
+    /// which closes the connection (SQLite rolls back the uncommitted
+    /// transaction) and replenishes the writer slot with a fresh
+    /// connection — the lease cannot leak, and the slot stays
+    /// grantable. If `commit`/drop already consumed the handle (`conn`
+    /// taken), the op fails closed with `Closed` — one-shot handles
+    /// cannot op past their terminal transition.
+    fn with_conn<T, F>(&mut self, f: F) -> BoxedFuture<'_, Result<T>>
+    where
+        T: Send + 'static,
+        F: FnOnce(&rusqlite::Connection) -> Result<T> + Send + 'static,
+    {
+        let Some(conn) = self.conn.take() else {
+            return Box::pin(async { Err(Error::Closed) });
+        };
+        let writer = self.writer.clone();
+        let reopen = self.reopen.clone();
+        let fut = blocking(move || {
+            let lease = ConnLease {
+                conn: Some(conn),
+                writer: writer.clone(),
+                reopen: reopen.clone(),
+            };
+            let out = match lease.conn() {
+                Some(conn) => f(conn),
+                None => Err(Error::database(std::io::Error::other(
+                    "the store is closed: writer slot unavailable",
+                ))),
+            };
+            Ok((out, lease))
+        });
+        Box::pin(async move {
+            let (out, lease) = fut.await?;
+            self.conn = lease.release();
+            out
+        })
+    }
+}
+
+/// The writer connection inside the `with_conn` bridge: an RAII guard
+/// so the connection crossing `spawn_blocking` always has an owner
+/// that accounts for the writer slot. Handled normally, the future's
+/// completion defuses the guard (`release`) and the connection returns
+/// to the handle; on a mid-op-future cancellation the guard drops
+/// instead — dropping the connection (its uncommitted transaction
+/// rolls back at the SQLite layer) and replenishing the writer slot
+/// with a fresh connection, so the next lease proceeds.
+struct ConnLease {
+    conn: Option<rusqlite::Connection>,
+    writer: std::sync::Arc<Writer>,
+    reopen: Arc<dyn Fn() -> Result<rusqlite::Connection> + Send + Sync>,
+}
+
+impl ConnLease {
+    fn conn(&self) -> Option<&rusqlite::Connection> {
+        self.conn.as_ref()
+    }
+
+    fn release(mut self) -> Option<rusqlite::Connection> {
+        self.conn.take()
+    }
+}
+
+impl Drop for ConnLease {
+    fn drop(&mut self) {
+        if let Some(conn) = self.conn.take() {
+            if let Err(rollback) = conn.execute_batch("ROLLBACK") {
+                eprintln!(
+                    "alkstore: error: rollback on a cancelled transaction op \
+                     left the connection in an unknown state: {rollback}"
+                );
+            }
+            drop(conn);
+            if let Ok(fresh) = (self.reopen)() {
+                self.writer.release(fresh);
+            }
+        }
+    }
+}
+
+/// Stamp resolution for the tx path over the handle-less default
+/// shapes: `EnqueueOpts::max_attempts` is the one per-job override,
+/// resolved over the base stamp set (ADR-010 §3a; the other stamps
+/// take the base untouched — the derived defaults carry
+/// visibility/backoff/retention).
+fn stamps_with_override(base: Stamps, opts: &EnqueueOpts) -> Stamps {
+    Stamps {
+        max_attempts: opts.max_attempts.unwrap_or(base.max_attempts),
+        ..base
+    }
+}
+
+impl TxHandle for SqliteTxHandle {
+    fn enqueue_tx<'a>(
+        &'a mut self,
+        queue: &str,
+        opts: EnqueueOpts,
+        payload: Value,
+    ) -> BoxedFuture<'a, Result<i64>> {
+        if let Err(e) = validate_shared_name(queue) {
+            return Box::pin(async move { Err(e) });
+        }
+        let queue = queue.to_string();
+        self.with_conn(move |conn| {
+            let bytes = alkstore::encode_payload(&payload)?;
+            let resolved = resolve_enqueue_opts(conn, &opts)?;
+            let stamps = stamps_with_override(plain_queue_default_stamps(), &opts);
+            enqueue(
+                conn,
+                &queue,
+                std::str::from_utf8(&bytes)
+                    .map_err(|e| Error::Codec(format!("row payload must be utf-8: {e}")))?,
+                resolved.run_at,
+                opts.priority,
+                resolved.expires_at,
+                stamps,
+            )
+            .map_err(sqlite_error)
+        })
+    }
+
+    fn publish_tx<'a>(&'a mut self, stream: &str, payload: Value) -> BoxedFuture<'a, Result<i64>> {
+        let stream = stream.to_string();
+        Box::pin(async move { self.publish_with_key_tx(&stream, None, payload).await })
+    }
+
+    fn publish_with_key_tx<'a>(
+        &'a mut self,
+        stream: &str,
+        key: Option<String>,
+        payload: Value,
+    ) -> BoxedFuture<'a, Result<i64>> {
+        if let Err(e) = validate_shared_name(stream) {
+            return Box::pin(async move { Err(e) });
+        }
+        let key = match key {
+            Some(k) if k.is_empty() || k.trim().is_empty() => {
+                return Box::pin(async move { Err(Error::InvalidName { name: k }) });
+            }
+            other => other,
+        };
+        let stream = stream.to_string();
+        self.with_conn(move |conn| {
+            let bytes = alkstore::encode_payload(&payload)?;
+            stream_publish(
+                conn,
+                &stream,
+                key.as_deref(),
+                std::str::from_utf8(&bytes)
+                    .map_err(|e| Error::Codec(format!("row payload must be utf-8: {e}")))?,
+            )
+            .map_err(sqlite_error)
+        })
+    }
+
+    fn notify_tx<'a>(&'a mut self, channel: &str, payload: Value) -> BoxedFuture<'a, Result<()>> {
+        if let Err(e) = validate_shared_name(channel) {
+            return Box::pin(async move { Err(e) });
+        }
+        let channel = channel.to_string();
+        self.with_conn(move |conn| {
+            let bytes = alkstore::encode_payload(&payload)?;
+            conn.query_row(
+                "SELECT notify(?1, ?2)",
+                rusqlite::params![
+                    channel,
+                    String::from_utf8(bytes).map_err(|e| Error::Codec(format!(
+                        "notify payload must be utf-8 text: {e}"
+                    )))?,
+                ],
+                |_| Ok(()),
+            )
+            .map_err(sqlite_error)
+        })
+    }
+
+    fn save_offset_tx<'a>(
+        &'a mut self,
+        stream: &str,
+        consumer: &str,
+        offset: i64,
+    ) -> BoxedFuture<'a, Result<()>> {
+        if let Err(e) = validate_shared_name(stream) {
+            return Box::pin(async move { Err(e) });
+        }
+        if let Err(e) = validate_local_name(consumer) {
+            return Box::pin(async move { Err(e) });
+        }
+        let stream = stream.to_string();
+        let consumer = consumer.to_string();
+        self.with_conn(move |conn| {
+            stream_save_offset(conn, &consumer, &stream, offset)
+                .map(|_| ())
+                .map_err(sqlite_error)
+        })
+    }
+
+    fn get_job_tx<'a>(
+        &'a mut self,
+        queue: &str,
+        job_id: i64,
+    ) -> BoxedFuture<'a, Result<Option<Job>>> {
+        if let Err(e) = validate_shared_name(queue) {
+            return Box::pin(async move { Err(e) });
+        }
+        let queue = queue.to_string();
+        self.with_conn(move |conn| {
+            let row = get_job(conn, job_id).map_err(sqlite_error)?;
+            let Some(json) = row else {
+                return Ok(None);
+            };
+            let value: serde_json::Value = serde_json::from_str(&json)
+                .map_err(|e| Error::Codec(format!("job row must be json: {e}")))?;
+            if value["queue"].as_str() != Some(queue.as_str()) {
+                return Ok(None);
+            }
+            job_from_json(&value)
+        })
+    }
+
+    fn get_offset_tx<'a>(
+        &'a mut self,
+        stream: &str,
+        consumer: &str,
+    ) -> BoxedFuture<'a, Result<i64>> {
+        if let Err(e) = validate_shared_name(stream) {
+            return Box::pin(async move { Err(e) });
+        }
+        if let Err(e) = validate_local_name(consumer) {
+            return Box::pin(async move { Err(e) });
+        }
+        let stream = stream.to_string();
+        let consumer = consumer.to_string();
+        self.with_conn(move |conn| {
+            stream_get_offset(conn, &consumer, &stream).map_err(sqlite_error)
+        })
+    }
+
+    fn read_since_tx<'a>(
+        &'a mut self,
+        stream: &str,
+        offset: i64,
+        limit: i64,
+    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
+        if let Err(e) = validate_shared_name(stream) {
+            return Box::pin(async move { Err(e) });
+        }
+        let stream = stream.to_string();
+        self.with_conn(move |conn| {
+            if limit <= 0 {
+                return Ok(Vec::new());
+            }
+            let page = stream_read_since(conn, &stream, offset, limit).map_err(sqlite_error)?;
+            stream_events_from_json_page(&page, stream)
+        })
+    }
+
+    fn read_from_consumer_tx<'a>(
+        &'a mut self,
+        stream: &str,
+        consumer: &str,
+        limit: i64,
+    ) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
+        if let Err(e) = validate_shared_name(stream) {
+            return Box::pin(async move { Err(e) });
+        }
+        if let Err(e) = validate_local_name(consumer) {
+            return Box::pin(async move { Err(e) });
+        }
+        let stream = stream.to_string();
+        let consumer = consumer.to_string();
+        self.with_conn(move |conn| {
+            if limit <= 0 {
+                return Ok(Vec::new());
+            }
+            let from = stream_get_offset(conn, &consumer, &stream).map_err(sqlite_error)?;
+            let page = stream_read_since(conn, &stream, from, limit).map_err(sqlite_error)?;
+            stream_events_from_json_page(&page, stream)
+        })
+    }
+
+    fn outbox_enqueue_tx<'a>(
+        &'a mut self,
+        outbox: &str,
+        opts: EnqueueOpts,
+        payload: Value,
+    ) -> BoxedFuture<'a, Result<i64>> {
+        if let Err(e) = validate_shared_name(outbox) {
+            return Box::pin(async move { Err(e) });
+        }
+        let outbox = outbox.to_string();
+        self.with_conn(move |conn| {
+            let bytes = alkstore::encode_payload(&payload)?;
+            let resolved = resolve_enqueue_opts(conn, &opts)?;
+            let stamps = stamps_with_override(outbox_default_stamps(), &opts);
+            enqueue(
+                conn,
+                &outbox_backing_queue_name(&outbox),
+                std::str::from_utf8(&bytes)
+                    .map_err(|e| Error::Codec(format!("row payload must be utf-8: {e}")))?,
+                resolved.run_at,
+                opts.priority,
+                resolved.expires_at,
+                stamps,
+            )
+            .map_err(sqlite_error)
+        })
+    }
+
+    fn commit(mut self: Box<Self>) -> BoxedFuture<'static, Result<()>> {
+        let Some(conn) = self.conn.take() else {
+            return Box::pin(async { Err(Error::Closed) });
+        };
+        let writer = self.writer.clone();
+        Box::pin(async move {
+            blocking(move || {
+                conn.execute_batch("COMMIT").map_err(sqlite_error)?;
+                writer.release(conn);
+                Ok(())
+            })
+            .await
+        })
+    }
+}
+
+impl Drop for SqliteTxHandle {
+    fn drop(&mut self) {
+        if let Some(conn) = self.conn.take() {
+            match conn.execute_batch("ROLLBACK") {
+                Ok(()) => self.writer.release(conn),
+                Err(rollback) => {
+                    eprintln!(
+                        "alkstore: error: rollback on a dropped transaction handle \
+                         left the connection in an unknown state: {rollback}"
+                    );
+                    drop(conn);
+                    if let Ok(fresh) = (self.reopen)() {
+                        self.writer.release(fresh);
+                    }
+                }
+            }
+        }
+    }
+}
+
+/// Decode the substrate's job-row JSON into the contract value
+/// (`Job::from_row` — the engine-side constructor, ADR-021 §2's
+/// `claimed_at` surfaced).
+fn job_from_json(value: &serde_json::Value) -> Result<Option<Job>> {
+    let state = match value["state"].as_str() {
+        Some("pending") => JobState::Pending,
+        Some("processing") => JobState::Processing,
+        Some("dead") => JobState::Dead,
+        other => {
+            return Err(Error::Codec(format!(
+                "job row state must be pending|processing|dead, got {other:?}"
+            )));
+        }
+    };
+    let payload: Vec<u8> = value["payload"]
+        .as_str()
+        .ok_or_else(|| Error::Codec("job row payload must be a string".to_string()))?
+        .as_bytes()
+        .to_vec();
+    let field = |name: &str| -> Result<i64> {
+        serde_json::from_value(value[name].clone())
+            .map_err(|e| Error::Codec(format!("job row {name} must be i64: {e}")))
+    };
+    let opt_field = |name: &str| -> Result<Option<i64>> {
+        if value[name].is_null() {
+            Ok(None)
+        } else {
+            serde_json::from_value(value[name].clone())
+                .map_err(|e| Error::Codec(format!("job row {name} must be i64: {e}")))
+        }
+    };
+    let opt_str = |name: &str| -> Result<Option<String>> {
+        if value[name].is_null() {
+            Ok(None)
+        } else {
+            serde_json::from_value(value[name].clone())
+                .map_err(|e| Error::Codec(format!("job row {name} must be a string: {e}")))
+        }
+    };
+    Ok(Some(Job::from_row(
+        field("id")?,
+        value["queue"]
+            .as_str()
+            .ok_or_else(|| Error::Codec("job row queue must be a string".to_string()))?
+            .to_string(),
+        state,
+        payload,
+        field("priority")?,
+        field("run_at")?,
+        field("attempts")?,
+        field("max_attempts")?,
+        opt_str("worker_id")?,
+        opt_field("claimed_at")?,
+        opt_field("claim_expires_at")?,
+        field("created_at")?,
+        opt_field("expires_at")?,
+        field("visibility_timeout_s")?,
+        field("backoff_base_s")?,
+        opt_field("dead_letter_retention_s")?,
+        opt_str("last_error")?,
+        opt_field("died_at")?,
+    )))
+}
+
+/// Decode the substrate's stream-page JSON into the contract's
+/// [`StreamEvent`] values (`from_row` — `topic` carries the contract's
+/// `stream`).
+fn stream_events_from_json_page(page: &str, stream: String) -> Result<Vec<StreamEvent>> {
+    let array: serde_json::Value = serde_json::from_str(page)
+        .map_err(|e| Error::Codec(format!("stream page must be json: {e}")))?;
+    let rows = array
+        .as_array()
+        .ok_or_else(|| Error::Codec("stream page must be a json array".to_string()))?;
+    let mut out = Vec::with_capacity(rows.len());
+    for row in rows {
+        out.push(StreamEvent::from_row(
+            serde_json::from_value(row["offset"].clone())
+                .map_err(|e| Error::Codec(format!("stream row offset must be i64: {e}")))?,
+            stream.clone(),
+            if row["key"].is_null() {
+                None
+            } else {
+                serde_json::from_value(row["key"].clone())
+                    .map_err(|e| Error::Codec(format!("stream row key must be a string: {e}")))?
+            },
+            row["payload"]
+                .as_str()
+                .ok_or_else(|| Error::Codec("stream row payload must be a string".to_string()))?
+                .as_bytes()
+                .to_vec(),
+            serde_json::from_value(row["created_at"].clone())
+                .map_err(|e| Error::Codec(format!("stream row created_at must be i64: {e}")))?,
+        ));
+    }
+    Ok(out)
+}