alkstore/alkstore-contract-suite/src/properties.rs

//! The suite's property rows — one named, independently-runnable
//! property per backlog row, version-stamped per the
//! [convention](crate::version_stamp).
//!
//! Assertion posture: rows panic with a label naming the entry point
//! and the broken expectation — a failing row is a test failure, which
//! is the suite's purpose as a test-side artifact (relaxing the
//! family-wide no-panic rule here only).
//!
//! Determinism: no wall-clock-timing-value assertions and no two-task
//! race windows; rows keep to single-store sequential drives and the
//! factory's isolation guarantee.
//!
//! Orchestration posture: cross-connection wake orchestration stays
//! engine-test-side. **Runner-driving rows are admitted as an
//! extension** (the backlog's scheduler row demands the tick machinery
//! in the suite, and it cannot be pinned sequentially —
//! `run_schedules` is the drive): a spawned `run_schedules(stop)`
//! task with a core [`StopToken`], driven against state outcomes only
//! (jobs fired, leadership returned), with tolerances bounded well
//! above the boundary resolution — elapsed-boundary bands, never
//! timing-value assertions, never two-task race windows. The wave-5
//! scheduler rows are the extension's riders; this is a suite-side
//! posture change (ADR-017 §2 class 4 — the suite tests the contract,
//! it is not the contract).
//!
//! The N-5 panic probe is admitted by its own backlog row's mechanics
//! for the same reason (the disposition demands a real engine's
//! handle): a spawned `with_tx` task joined via
//! `JoinError::is_panic()` — `catch_unwind` around an async closure
//! cannot see the panic point across an await — with the post-panic
//! assertions read-only until convergence, single-task driving, no
//! two-task race window.
//!
//! Rows exercise a *close/dispose* arm by dropping the
//! store handle (the dispose carrier every engine implements — the
//! drop path closes receivers terminally); teardown-then-fresh-reopen
//! rows would be factory-contract tests (ADR-022), not contract rows,
//! so none exist.

use std::sync::Arc;

use serde_json::json;

use alkstore::{
    BoxedFuture, Delivery, EnqueueOpts, Error, Job, JobHandle, JobState, QueueOpts,
    RESERVED_PREFIX, Result, ScheduleOpts, StopToken, Store, StreamEvent, encode_payload,
    validate_shared_name,
};

use crate::factory::StoreFactory;

struct NoDelivery;

impl Delivery for NoDelivery {
    fn deliver<'a>(&'a mut self, _job: Box<dyn JobHandle>) -> BoxedFuture<'a, Result<()>> {
        Box::pin(std::future::ready(Ok(())))
    }
}

fn expect_invalid<T>(label: &str, out: alkstore::Result<T>) {
    match out {
        Err(Error::InvalidName { .. }) => {}
        Err(other) => {
            panic!("{label}: expected `InvalidName`, got {other:?} — entry-point validation broke")
        }
        Ok(_) => panic!("{label}: expected `InvalidName` rejection, got Ok"),
    }
}

fn expect_reserved<T>(label: &str, out: alkstore::Result<T>) {
    match out {
        Err(Error::ReservedName { .. }) => {}
        Err(other) => {
            panic!("{label}: expected `ReservedName`, got {other:?} — entry-point validation broke")
        }
        Ok(_) => panic!("{label}: expected `ReservedName` rejection, got Ok"),
    }
}

fn expect_ok<T>(label: &str, out: alkstore::Result<T>) {
    if let Err(other) = out {
        panic!("{label}: expected success (no validation error), got {other:?}");
    }
}

/// **Exemplar row** — name validation at every entry point,
/// auto-commit and tx paths alike.
///
/// Every name-bearing entry point on the [`Store`]
/// surface — and every name-bearing `*_tx` twin on the
/// [`TxHandle`](alkstore::TxHandle) — rejects, before any engine round
/// trip: the empty (or whitespace-only) name with
/// [`Error::InvalidName`], and the reserved-prefix name
/// (`__alkstore_...`) with [`Error::ReservedName`], on the
/// shared-namespace kinds — channels, streams, queues, outboxes,
/// locks, schedule names (including `unschedule`), and `schedule()`'s
/// queue argument.
/// Consumer-local identifiers take the non-empty rule only: `owner`
/// reserved-prefixed is legal, `owner` empty is not — likewise
/// `worker_id` on the claim/pull entry points (`claim_one`,
/// `claim_batch`, `run_once`) and the consumer arguments on the
/// stream handle's offset/read forms. The tx path is
/// exercised through `begin_tx` and `with_tx` (the provided wrapper):
/// validation must not be bypassable by the commit-atomic path.
///
/// Contract stamp: ADR-008 §4 (reserved namespace + entry-point
/// validation); ADR-021 §3 (the schedule queue argument); ADR-015 §2
/// (the keyed publish's non-empty-key rule).
///
/// Engine-independent, pure contract: runnable against any factory.
/// The happy-path spot-checks prove the rejections are the
/// validation's doing, not a store-wide failure — valid names
/// construct handles without a validation error, and a
/// reserved-prefixed *local* identifier is accepted.
pub async fn name_validation_rejects_empty_and_reserved(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    // Auto-commit paths — empty names.
    expect_invalid("notify(empty channel)", store.notify("", json!(null)).await);
    expect_invalid("listen(empty channel)", store.listen("").await);
    expect_invalid("stream(empty name)", store.stream("").await);
    expect_invalid(
        "queue(empty name)",
        store.queue("", alkstore::QueueOpts::default()).await,
    );
    expect_invalid("outbox(empty name)", store.outbox("").await);
    expect_invalid(
        "try_lock(empty name)",
        store.try_lock("", "worker", 60).await,
    );
    expect_invalid(
        "try_lock(empty owner)",
        store.try_lock("lock", "", 60).await,
    );
    expect_invalid(
        "schedule(empty name)",
        store
            .schedule("", "@every 1s", "work", json!(null), Default::default())
            .await,
    );
    expect_invalid(
        "schedule(empty queue argument)",
        store
            .schedule("tick", "@every 1s", "", json!(null), Default::default())
            .await,
    );
    expect_invalid("unschedule(empty name)", store.unschedule("").await);
    expect_reserved(
        "unschedule(reserved name)",
        store.unschedule(RESERVED_PREFIX).await,
    );

    // Auto-commit paths — reserved-prefix names.
    expect_reserved(
        "notify(reserved channel)",
        store.notify(RESERVED_PREFIX, json!(null)).await,
    );
    expect_reserved(
        "listen(reserved channel)",
        store.listen(RESERVED_PREFIX).await,
    );
    expect_reserved("stream(reserved name)", store.stream(RESERVED_PREFIX).await);
    expect_reserved(
        "queue(reserved name)",
        store
            .queue(RESERVED_PREFIX, alkstore::QueueOpts::default())
            .await,
    );
    expect_reserved("outbox(reserved name)", store.outbox(RESERVED_PREFIX).await);
    expect_reserved(
        "try_lock(reserved name)",
        store.try_lock(RESERVED_PREFIX, "worker", 60).await,
    );
    expect_reserved(
        "schedule(reserved name)",
        store
            .schedule(
                RESERVED_PREFIX,
                "@every 1s",
                "work",
                json!(null),
                Default::default(),
            )
            .await,
    );
    expect_reserved(
        "schedule(reserved queue argument)",
        store
            .schedule(
                "tick",
                "@every 1s",
                RESERVED_PREFIX,
                json!(null),
                Default::default(),
            )
            .await,
    );

    // Happy-path spot-checks: valid names pass validation; a
    // reserved-prefixed owner (consumer-local identifier) is legal.
    expect_ok("stream(valid name)", store.stream("work").await);
    expect_ok(
        "queue(valid name)",
        store.queue("work", alkstore::QueueOpts::default()).await,
    );
    expect_ok(
        "try_lock(reserved-prefixed owner is legal)",
        store.try_lock("lock", RESERVED_PREFIX, 60).await,
    );

    // Handle-level name-bearing entry points: the claim/pull ops'
    // `worker_id` and the stream handle's consumer arguments are
    // consumer-local identifiers (non-empty only, ADR-008 §4's
    // third-round annotation) — validated at the handle's entry
    // points identically to the store-level ones.
    let queue = store
        .queue("work", alkstore::QueueOpts::default())
        .await
        .expect("queue handle");
    expect_invalid("claim_one(empty worker_id)", queue.claim_one("").await);
    expect_invalid(
        "claim_batch(empty worker_id)",
        queue.claim_batch("", 1).await,
    );
    let stream = store.stream("work").await.expect("stream handle");
    expect_invalid(
        "save_offset(empty consumer)",
        stream.save_offset("", 0).await,
    );
    expect_invalid("get_offset(empty consumer)", stream.get_offset("").await);
    expect_invalid(
        "read_from_consumer(empty consumer)",
        stream.read_from_consumer("", 10).await,
    );
    expect_invalid(
        "get_offset(whitespace-only consumer)",
        stream.get_offset(" \t ").await,
    );
    let mut outbox = store.outbox("work").await.expect("outbox handle");
    expect_invalid(
        "run_once(empty worker_id)",
        outbox.run_once("", &mut NoDelivery).await,
    );

    // Tx paths — the twins validate identically (begin_tx route). The
    // battery lives in its own block so the handle drops before the
    // `with_tx` leg below: a real engine's `begin_tx` parks a
    // concurrent begin on the writer slot (ADR-007's honest model —
    // the mock harness never saw this), so a hold-over handle here
    // would deadlock the row, not test it.
    {
        let mut tx = store.begin_tx().await.expect("begin_tx opens");
        expect_invalid(
            "enqueue_tx(empty queue)",
            tx.enqueue_tx("", Default::default(), json!(null)).await,
        );
        expect_reserved(
            "enqueue_tx(reserved queue)",
            tx.enqueue_tx(RESERVED_PREFIX, Default::default(), json!(null))
                .await,
        );
        expect_invalid(
            "publish_tx(empty stream)",
            tx.publish_tx("", json!(null)).await,
        );
        expect_reserved(
            "publish_tx(reserved stream)",
            tx.publish_tx(RESERVED_PREFIX, json!(null)).await,
        );
        expect_invalid(
            "notify_tx(empty channel)",
            tx.notify_tx("", json!(null)).await,
        );
        expect_reserved(
            "notify_tx(reserved channel)",
            tx.notify_tx(RESERVED_PREFIX, json!(null)).await,
        );
        expect_invalid(
            "publish_with_key_tx(empty stream)",
            tx.publish_with_key_tx("", None, json!(null)).await,
        );
        expect_reserved(
            "publish_with_key_tx(reserved stream)",
            tx.publish_with_key_tx(RESERVED_PREFIX, None, json!(null))
                .await,
        );
        expect_invalid(
            "publish_with_key_tx(empty key)",
            tx.publish_with_key_tx("work", Some(String::new()), json!(null))
                .await,
        );
        expect_invalid(
            "publish_with_key_tx(whitespace key)",
            tx.publish_with_key_tx("work", Some("   ".to_string()), json!(null))
                .await,
        );
        expect_invalid(
            "save_offset_tx(empty stream)",
            tx.save_offset_tx("", "consumer", 0).await,
        );
        expect_reserved(
            "save_offset_tx(reserved stream)",
            tx.save_offset_tx(RESERVED_PREFIX, "consumer", 0).await,
        );
        expect_invalid(
            "save_offset_tx(empty consumer)",
            tx.save_offset_tx("work", "", 0).await,
        );
        expect_invalid("get_job_tx(empty queue)", tx.get_job_tx("", 1).await);
        expect_reserved(
            "get_job_tx(reserved queue)",
            tx.get_job_tx(RESERVED_PREFIX, 1).await,
        );
        expect_invalid(
            "get_offset_tx(empty stream)",
            tx.get_offset_tx("", "consumer").await,
        );
        expect_reserved(
            "get_offset_tx(reserved stream)",
            tx.get_offset_tx(RESERVED_PREFIX, "consumer").await,
        );
        expect_invalid(
            "read_since_tx(empty stream)",
            tx.read_since_tx("", 0, 10).await,
        );
        expect_reserved(
            "read_since_tx(reserved stream)",
            tx.read_since_tx(RESERVED_PREFIX, 0, 10).await,
        );
        expect_invalid(
            "read_from_consumer_tx(empty stream)",
            tx.read_from_consumer_tx("", "consumer", 10).await,
        );
        expect_reserved(
            "read_from_consumer_tx(reserved stream)",
            tx.read_from_consumer_tx(RESERVED_PREFIX, "consumer", 10)
                .await,
        );
        expect_invalid(
            "read_from_consumer_tx(empty consumer)",
            tx.read_from_consumer_tx("work", "", 10).await,
        );
        expect_invalid(
            "get_offset_tx(empty consumer)",
            tx.get_offset_tx("work", "").await,
        );
        expect_invalid(
            "outbox_enqueue_tx(empty outbox)",
            tx.outbox_enqueue_tx("", Default::default(), json!(null))
                .await,
        );
        expect_reserved(
            "outbox_enqueue_tx(reserved outbox)",
            tx.outbox_enqueue_tx(RESERVED_PREFIX, Default::default(), json!(null))
                .await,
        );
    }

    // Tx path via `with_tx` (the provided wrapper): a validation error
    // inside the closure propagates as the wrapper's `Err` (and the
    // disposition rolls back) — the commit-atomic path cannot bypass
    // validation either. The battery's handle dropped above (never
    // committed — validation never reached the storage layer), so the
    // writer slot is free for this wrapper's own begin.
    let result = store
        .with_tx(Box::new(|tx| {
            Box::pin(async move {
                validate_shared_name("")?;
                tx.notify_tx("work", json!(null)).await
            }) as BoxedFuture<'_, Result<()>>
        }))
        .await;
    expect_invalid("with_tx(inner empty-name rejection)", result);

    factory.teardown().await.expect("factory teardown");
}

/// **Extent-clamp semantics** — extents clamp to empty, boundaries are
/// total (the numeric-domain rule).
///
/// `claim_batch(n <= 0)` claims nothing: empty `Vec`, no error, no
/// dialect reinterpretation — a non-positive extent never hands out
/// more rows than a positive-`n` claim on the same queue state would
/// (the SQLite `LIMIT -1` dialect artifact is dead at the trait-impl
/// entry, before any substrate round trip; the queued rows survive
/// untouched and remain claimable after). Both stream reads clamp
/// identically: `read_since(limit <= 0)` and
/// `read_from_consumer(limit <= 0)` return the empty `Vec` at the
/// trait-impl entry. Boundaries are total without a guard:
/// `trim_to(horizon < min surviving offset)` deletes nothing
/// (idempotent, `Ok(0)`) and surviving rows' offsets are never
/// renumbered.
///
/// Contract stamp: ADR-023 §2 (numeric domains: extents clamp to
/// empty, boundaries total); ADR-010 §1 (the no-work-is-a-value
/// vocabulary the clamp composes with).
///
/// Runnable on any engine (SQLite now, pg at wave 5 via the same row).
pub async fn extent_clamp_semantics(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    let queue = store
        .queue("work", alkstore::QueueOpts::default())
        .await
        .unwrap();
    let mut ids = Vec::new();
    for i in 0..5 {
        ids.push(
            queue
                .enqueue(json!({"i": i}), EnqueueOpts::default())
                .await
                .unwrap(),
        );
    }

    for n in [0i64, -1, i64::MIN] {
        let claimed = queue.claim_batch("w1", n).await.unwrap();
        assert!(
            claimed.is_empty(),
            "claim_batch(n = {n}) must clamp to the empty Vec"
        );
    }
    // The negative extent never became "no limit": the five rows are
    // still pending and a positive-n claim on the same state hands out
    // exactly its extent.
    let claimed = queue.claim_batch("w1", 2).await.unwrap();
    assert_eq!(claimed.len(), 2, "a positive extent claims normally");
    assert_eq!(
        claimed.iter().map(|h| h.job().id).collect::<Vec<_>>(),
        ids[..2]
    );
    assert_eq!(
        queue.get_job(ids[2]).await.unwrap().unwrap().state,
        JobState::Pending,
        "the guarded calls destroyed nothing"
    );

    let stream = store.stream("events").await.unwrap();
    stream.publish(json!({"n": 1})).await.unwrap();
    stream.publish(json!({"n": 2})).await.unwrap();
    stream.save_offset("c", 5).await.unwrap();
    for limit in [0i64, -1, i64::MIN] {
        let read = stream.read_since(0, limit).await.unwrap();
        assert!(
            read.is_empty(),
            "read_since(limit = {limit}) must read nothing"
        );
        let read = stream.read_from_consumer("c", limit).await.unwrap();
        assert!(
            read.is_empty(),
            "read_from_consumer(limit = {limit}) must read nothing"
        );
    }
    let whole = stream.read_since(0, 10).await.unwrap();
    assert_eq!(
        whole.len(),
        2,
        "the clamped reads never handed the stream out in bulk"
    );

    // Boundary totality: a horizon below every surviving offset deletes
    // nothing, idempotently, and survivors keep their offsets.
    let deleted = stream.trim_to(-7).await.unwrap();
    assert_eq!(deleted, 0, "a negative trim horizon deletes nothing");
    let after = stream.read_since(0, 10).await.unwrap();
    let survivor_offsets: Vec<i64> = after.iter().map(|e| e.offset).collect();
    assert_eq!(
        survivor_offsets,
        whole.iter().map(|e| e.offset).collect::<Vec<_>>(),
        "surviving offsets are immutable, never renumbered"
    );

    factory.teardown().await.expect("factory teardown");
}

/// **Duration-refusal** — durations are the other half of the
/// numeric-domain rule: `try_lock(ttl <= 0)` and `renew(ttl <= 0)` are
/// opaque rejections (`Error::Database`), never a granted lock, never
/// a silent no-op grant, and not misreported as name errors. The lock
/// state is untouched by a guarded call: a guarded `try_lock` over an
/// untouched name creates no row and a later valid try succeeds — and
/// a valid holder's subsequent guarded renews leave its window
/// unshrunken.
///
/// The opaque shape is deliberate (no `InvalidArgument` variant was
/// minted — the caller's remedy is the same regardless: fix the
/// constant, ADR-008 §5's act-differently rule); the row pins the shape
/// engines must agree on, not a message text.
///
/// Contract stamp: ADR-023 §2 (durations reject); ADR-008 §5 (the
/// opaque fallback carries it, source chain in the detail); ADR-008 §7
/// (the exclusion guarantee the refusal protects).
pub async fn duration_refusal_on_non_positive_ttl(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    for ttl in [0i64, -1] {
        let out = store.try_lock("guarded", "a", ttl).await;
        match out {
            Ok(Some(_)) => panic!("try_lock(ttl = {ttl}) must not grant"),
            Ok(None) => panic!("try_lock(ttl = {ttl}) must be a rejection, not a silent miss"),
            Err(e) => assert!(
                matches!(e, Error::Database(_)),
                "opaque `Database` rejection shape, got: {e:?}"
            ),
        }
    }
    let lock = store
        .try_lock("guarded", "a", 300)
        .await
        .unwrap()
        .expect("the guarded calls fired no round trip: the name is free");
    for ttl in [0i64, -1] {
        match lock.renew(ttl).await {
            Ok(landed) => panic!("renew(ttl = {ttl}) must not land: {landed}"),
            Err(e) => assert!(
                matches!(e, Error::Database(_)),
                "opaque `Database` rejection shape, got: {e:?}"
            ),
        }
    }
    assert!(lock.renew(600).await.unwrap(), "a positive ttl renews");
    let _ = lock.release().await.unwrap();

    factory.teardown().await.expect("factory teardown");
}

/// **Lock TTL/expiry re-acquisition** — the ADR-008 §7 guarantee row:
/// a held lock excludes a second acquirer (`None` — the loser value,
/// never an error); after the TTL elapses exclusion lapses *silently*
/// — no revocation event, no error anywhere — and a second owner's
/// acquire succeeds; the original holder's post-expiry `renew` is
/// refused (`false` — the deadline-lapse refusal of the handle's
/// uniform predicate shape); and the second owner's `release` frees
/// the name for a third acquirer.
///
/// Two-leg shape, each leg owning the posture its assertions can
/// bear. The **exclusion leg** holds a long TTL (300 s — a span the
/// row's own duration cannot cross): exclusion-while-held is a
/// *live-TTL* state, and a short-TTL holder would hand that assertion
/// a wall-clock-collapsible window — second-precision engine clocks
/// stamp a 1 s expiry that lapses after as little as `1 - frac` of a
/// second, so any scheduling/server stall between the two acquisitions
/// (>~1 s suffices) spuriously fails the exclusion assert itself (the
/// 2026-10-11 flake investigation's natural repro: two of thirty
/// loaded solo runs failed exactly so, probe-confirmed on the leg). The **lapse leg** carries the Short-TTL (1 s) stamp with the
/// tolerance-sleep posture: its assertions are *post-sleep states* a
/// slow clock can only delay, never un-lapse (state outcomes
/// asserted, never durations; sequential single-store drive).
///
/// Cross-reference: the duration-guard legs (`ttl <= 0` rejecting on
/// `try_lock`/`renew`) pin in `duration_refusal_on_non_positive_ttl`
/// (ADR-023 §2) — this row owns the TTL *lifecycle* (lapse,
/// silent-exclusion loss, re-acquisition, post-expiry refusal, and
/// name-freedom after release).
///
/// Contract stamp: ADR-008 §7 (the locks guarantee row — mutual
/// exclusion bounded by TTL + renewal; expiry is a silent loss of
/// exclusivity, not an error; exclusivity resumes with
/// re-acquisition); ADR-019 §1 (the `Lock` handle surface the
/// refusals ride — repeatable `renew` with the lost-it `false`, the
/// consuming `release`).
pub async fn lock_ttl_expiry_and_reacquisition(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    // The exclusion leg — long-TTL holder deliberately (see the row
    // doc's two-leg posture): the held-lock exclusion assert must not
    // race the holder's own expiry.
    let holder = store
        .try_lock("lapse", "a", 300)
        .await
        .expect("try_lock on a free name must not error")
        .expect("the first acquire holds the free name");
    let contender = store
        .try_lock("lapse", "b", 300)
        .await
        .expect("a contender's try_lock must not error");
    assert!(
        contender.is_none(),
        "a held lock excludes a second acquirer"
    );
    assert!(
        holder
            .release()
            .await
            .expect("the exclusion leg's holder release errs only on storage failure"),
        "the exclusion leg's holder releases its own row"
    );

    // The lapse leg — a fresh short-TTL holder: TTL expiry (a 1 s
    // stamp; the sleep is bounded well past the second-precision
    // stamp's resolution) lapses exclusion silently — the second
    // owner's acquire is a clean Ok(Some) with no error preceding it
    // and no revocation event to observe.
    let lapsing_holder = store
        .try_lock("lapse", "a", 1)
        .await
        .expect("try_lock on the freed name must not error")
        .expect("the lapse leg's acquire holds the freed name");
    past_stamp_sleep(2_500).await;
    let second = store
        .try_lock("lapse", "b", 300)
        .await
        .expect("the post-expiry acquire must not error (silent lapse, no revocation event)")
        .expect("after TTL expiry exclusion has silently lapsed: a second owner acquires");

    assert!(
        !lapsing_holder
            .renew(300)
            .await
            .expect("renew errs only on storage failure, never on lapse"),
        "the original holder's post-expiry renew is refused: the lost-it arm"
    );

    assert!(
        second
            .release()
            .await
            .expect("the second owner's release errs only on storage failure"),
        "the second owner's release deletes its own row"
    );
    let third = store
        .try_lock("lapse", "c", 300)
        .await
        .expect("try_lock on the freed name must not error")
        .expect("the released name is acquirable by a third owner");
    assert!(
        third
            .release()
            .await
            .expect("the third release errs only on storage failure"),
        "the third owner releases its own row (no residue for teardown)"
    );

    factory.teardown().await.expect("factory teardown");
}

/// **Concurrent `try_lock` loser is a value** — the contention
/// posture: with a lock held, contenders' `try_lock` calls return the
/// clean no-work value (`None`) on both engines — never a `Database`
/// error, never a busy-throw surfacing (the SQLite substrate's
/// busy-path is the risk site this row pins; the pg side is the
/// reference behavior the suite drives first). Contenders are driven
/// sequentially against one held lock (several distinct owners,
/// including a repeat contender), and a release-then-contend cycle
/// proves the loser path leaves no state behind that blocks a later
/// acquire — the former loser acquires cleanly on an empty field.
///
/// Cross-reference: the TTL-lapse arms of the loser value pin in
/// `lock_ttl_expiry_and_reacquisition` — this row owns the
/// *held-lock* loser path (the substrate's busy-path behavior is the
/// thing under test, the backlog row's open question).
///
/// Contract stamp: ADR-008 §5 (the value-not-error rule — `try_lock`
/// returning `Option<Lock>` is explicitly not a taxonomy item; a
/// loser `Database` error would be a contract violation); ADR-008 §7
/// (the exclusion row the contention composes with); ADR-019 §1 (the
/// handle surface the winner's consuming `release` rides).
pub async fn concurrent_try_lock_loser_is_a_value(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    let holder = store
        .try_lock("contended", "holder", 300)
        .await
        .expect("try_lock on a free name must not error")
        .expect("the first acquire holds the free name");

    // Sequential contenders against the held lock: every loser sees
    // the clean `None` value — a `Database` error on any of them is
    // the busy-path finding this row exists to catch.
    for attempt in 0..4u32 {
        let owner = if attempt.is_multiple_of(2) {
            "contender-1"
        } else {
            "contender-2"
        };
        let out = store
            .try_lock("contended", owner, 300)
            .await
            .expect("a contended try_lock must not error (no busy-throw surfacing)");
        assert!(
            out.is_none(),
            "a held lock must lose contenders the None value, not a granted lock"
        );
    }

    // Release-then-contend: the loser path left no state — an
    // immediate re-contend after release grants, to a former loser's
    // owner, and consumes cleanly.
    assert!(
        holder
            .release()
            .await
            .expect("the winner's release errs only on storage failure"),
        "the winner's release deletes its own row"
    );
    let winner = store
        .try_lock("contended", "contender-1", 300)
        .await
        .expect("the post-release contend must not error")
        .expect("the loser path left no state blocking a later acquire");
    assert!(
        winner
            .release()
            .await
            .expect("release errs only on storage failure"),
        "the new winner releases cleanly (no residue for teardown)"
    );

    factory.teardown().await.expect("factory teardown");
}

/// **`encode_payload` typed-failure round-trip** — enqueue and publish
/// store the exact serde_json serialization of the input `Value` and
/// decode back to `Value` equality; the helper's failure arm is typed
/// (`Error::Codec`), so no silent empty-byte store exists (which would
/// decode to `Null` — the corruption ADR-023 §1 removed).
///
/// Row-driven probe: the round-trip property pins the byte-level pin
/// through the real store paths (queue enqueue, keyed publish), both
/// auto-commit; the helper's typed shape is pinned directly by
/// exercising it against the same values (serialization of a `Value`
/// is infallible in practice — the assert is that what *is* stored is
/// exactly the helper's output bytes, and that the failure arm's type
/// exists for the seam to carry). Byte equality composes with the
/// cross-engine pin (the same `Value` serializes identically on both
/// engines).
///
/// Contract stamp: ADR-023 §1 (the fallible helper — `Result<Vec<u8>,
/// Error::Codec>`, no silent empty-Vec fallback); ADR-020 §4 (store
/// exactly the serialized bytes; decode back to `Value` equality).
pub async fn payload_round_trip_stores_exact_encoding(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    let value = json!({"task": "resize", "deep": [1, 2, {"n": null, "ok": true}], "f": -0.5});
    let exact = encode_payload(&value).expect("the serialization encodes");

    let queue = store
        .queue("work", alkstore::QueueOpts::default())
        .await
        .unwrap();
    let id = queue
        .enqueue(value.clone(), EnqueueOpts::default())
        .await
        .unwrap();
    let job = queue.get_job(id).await.unwrap().expect("the row exists");
    assert_eq!(
        job.payload, exact,
        "the stored job bytes are the helper's exact serialization"
    );
    let decoded: serde_json::Value =
        alkstore::payload_as(&job.payload).expect("the stored bytes decode");
    assert_eq!(decoded, value, "the decode round-trips to value equality");

    let stream = store.stream("events").await.unwrap();
    let offset = stream
        .publish_with_key(Some("k".to_string()), value.clone())
        .await
        .unwrap();
    assert!(offset > 0, "the publish got an offset");
    let page = stream.read_since(0, 10).await.unwrap();
    assert_eq!(page.len(), 1, "exactly one event landed");
    assert_eq!(
        page[0].payload, exact,
        "the stored event bytes are the helper's exact serialization"
    );
    assert_eq!(page[0].key.as_deref(), Some("k"), "the key round-trips");
    assert_eq!(
        alkstore::payload_as::<serde_json::Value>(&page[0].payload).unwrap(),
        value,
        "the event decode round-trips to value equality"
    );

    factory.teardown().await.expect("factory teardown");
}

/// **`PayloadTooLarge` occurrence asymmetry** — the SQLite engine
/// never produces the variant at any payload size (no notification
/// limit); engine-agnostic code writes the same match on both engines
/// and the non-occurring arm simply never fires. The row drives an
/// oversized payload through the store's notify path and asserts
/// success.
///
/// This is the SQLite column's arm of the asymmetry row (the pg-side
/// rejection arm pins with wave 4's adoption — the same row text, the
/// engine-differing leg inside).
///
/// Contract stamp: ADR-016 §5 (the occurrence asymmetry — SQLite never
/// produces `PayloadTooLarge`); ADR-008 §5 (universal variant, no
/// engine-specific growth).
pub async fn payload_too_large_never_produced_on_sqlite(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    let big = "x".repeat(2_000_000);
    store
        .notify("big", json!({"blob": big}))
        .await
        .expect("SQLite notify has no payload limit at any size");

    factory.teardown().await.expect("factory teardown");
}

/// **`PayloadTooLarge` occurrence asymmetry — the mem arm** — the mem
/// engine never produces the variant at any payload size (no
/// notification limit; the wake delivers the channel alone — the
/// notify transport stores nothing), the same non-occurrence posture
/// as SQLite's arm, born pinned as mem's column of the row
/// ([ADR-024](../../docs/architecture/decisions/024-mem-engine.md)
/// §3's classification pins mem wiring this arm). The row drives an
/// oversized payload — just past the 2 MiB mark the tx seam's mem
/// unit pins — through the store's notify path and asserts success.
///
/// Contract stamp: ADR-016 §5 (the occurrence asymmetry — the
/// non-occurrence arm wiring on mem); ADR-024 §3 (mem's posture in
/// the classification); ADR-008 §5 (universal variant, no
/// engine-specific growth).
pub async fn payload_too_large_never_produced_on_mem(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    let big = "x".repeat(2 * 1024 * 1024 + 1);
    store
        .notify("big", json!({"blob": big}))
        .await
        .expect("mem notify has no payload limit at any size");

    factory.teardown().await.expect("factory teardown");
}

/// **Drop = rollback, no-ghosts** — a transaction handle dropped
/// before commit rolls its writes back through RAII: job rows, stream
/// events, notifies, and offset saves all leave no residue the
/// transaction's own reads can no longer observe, and post-drop reads
/// (fresh tx and auto-commit handles alike) see the pre-transaction
/// state exactly.
///
/// The row drives one tx through the full no-ghosts list (every write
/// kind, ADR-021 §4's uniform list), drops without commit, then reads
/// the subjects through every surface that saw them mid-tx.
///
/// Contract stamp: ADR-021 §4 (drop = rollback; the uniform no-ghosts
/// list); ADR-007 (the caller-held-handle seam the RAII paths guard).
pub async fn drop_rollback_leaves_no_ghosts(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    let mut tx = store.begin_tx().await.unwrap();
    let ghost_id = tx
        .enqueue_tx("work", EnqueueOpts::default(), json!({"ghost": true}))
        .await
        .unwrap();
    tx.publish_tx("events", json!({"ghost": true}))
        .await
        .unwrap();
    tx.save_offset_tx("events", "c", 7).await.unwrap();

    // The ghost is real inside the tx (read-your-own-writes is what
    // makes this drop's rollback meaningful).
    assert!(
        tx.get_job_tx("work", ghost_id).await.unwrap().is_some(),
        "the tx sees its own enqueue pre-drop"
    );
    drop(tx);

    // Fresh reads: the subjects are gone through every surface.
    let mut tx = store.begin_tx().await.unwrap();
    let gone = tx.get_job_tx("work", ghost_id).await.unwrap();
    let events = tx.read_since_tx("events", 0, 100).await.unwrap();
    let checkpoint = tx.get_offset_tx("events", "c").await.unwrap();
    tx.commit().await.unwrap();
    assert_eq!(gone, None, "the dropped tx's job left no row");
    assert_eq!(events, Vec::<StreamEvent>::new(), "no ghost events");
    assert_eq!(checkpoint, 0, "the dropped tx's offset save never landed");

    let queue = store
        .queue("work", alkstore::QueueOpts::default())
        .await
        .unwrap();
    let drained = queue.claim_batch("probe", 10).await.unwrap();
    assert!(drained.is_empty(), "no ghost work became claimable");

    let stream = store.stream("events").await.unwrap();
    let events = stream.read_since(0, 100).await.unwrap();
    assert!(events.is_empty(), "no ghost events on the auto-commit read");
    assert_eq!(
        stream.get_offset("c").await.unwrap(),
        0,
        "no ghost checkpoint"
    );

    factory.teardown().await.expect("factory teardown");
}

/// **In-tx read-your-own-writes** — reads issued on the tx handle see
/// the transaction's own writes: `get_job_tx` a row the same tx
/// enqueued, `read_since_tx` an event the same tx published, and the
/// consumer-twin forms see their own saves. (Another connection
/// seeing none of it pre-commit and the post-rollback no-ghosts leg
/// pin in the drop-rollback and commit-visibility rows; this row pins
/// the visibility itself.)
///
/// The negative-seed leg (2026-10-11): a staged save seeding an
/// absent checkpoint stores raw — the boundary/offset totality of
/// ADR-023 §2 rides the tx seam too, in-tx read-back included (a
/// negative seed reads raw, never clamped against the absent-consumer
/// phantom zero, and survives the commit as the stored checkpoint).
///
/// Contract stamp: ADR-021 §1 (tx-side read methods,
/// read-your-own-writes); ADR-007 (the commit-atomic seam the reads
/// ride); ADR-023 §2 boundary/offset annotation (annotated
/// 2026-10-11 — a negative is stored as the checkpoint).
pub async fn in_tx_reads_see_own_writes(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    let mut tx = store.begin_tx().await.unwrap();
    let id = tx
        .enqueue_tx("work", EnqueueOpts::default(), json!({"n": 9}))
        .await
        .unwrap();
    let offset = tx.publish_tx("events", json!({"n": 9})).await.unwrap();

    let job = tx
        .get_job_tx("work", id)
        .await
        .unwrap()
        .expect("own write visible");
    assert_eq!(job.id, id);
    assert_eq!(job.queue, "work");
    assert_eq!(job.state, JobState::Pending);
    assert_eq!(job.payload, encode_payload(&json!({"n": 9})).unwrap());

    let page = tx.read_since_tx("events", 0, 10).await.unwrap();
    assert_eq!(page.len(), 1, "the tx sees its own publish");
    assert_eq!(page[0].offset, offset, "the assigned offset reads back");
    assert_eq!(page[0].stream, "events");

    // The twin forms see their own saves too.
    tx.save_offset_tx("events", "c", offset).await.unwrap();
    let from = tx.read_from_consumer_tx("events", "c", 10).await.unwrap();
    assert!(
        from.is_empty(),
        "everything is at or below the checkpoint just saved"
    );
    let checkpoint = tx.get_offset_tx("events", "c").await.unwrap();
    assert_eq!(checkpoint, offset, "the tx sees its own save");

    // The tx form's seed arm (ADR-023 §2, annotated 2026-10-11): a
    // staged save on a consumer with no committed checkpoint seeds
    // raw — the in-tx read-back is the raw seed, not the absent-consumer
    // phantom zero clamped over it, and the commit stores it verbatim.
    tx.save_offset_tx("events", "c2", -5).await.unwrap();
    let seeded = tx.get_offset_tx("events", "c2").await.unwrap();
    assert_eq!(
        seeded, -5,
        "a staged negative seed reads raw in-tx (no phantom-zero clamp)"
    );
    tx.commit().await.unwrap();

    let committed_seed = store
        .stream("events")
        .await
        .unwrap()
        .get_offset("c2")
        .await
        .unwrap();
    assert_eq!(
        committed_seed, -5,
        "a negative seed is stored as the checkpoint through the tx seam"
    );
    let resumed = store
        .stream("events")
        .await
        .unwrap()
        .read_from_consumer("c2", 10)
        .await
        .unwrap();
    assert_eq!(
        resumed.iter().map(|e| e.offset).collect::<Vec<_>>(),
        vec![offset],
        "reads anchored at the negative checkpoint resume forward: the log from the start"
    );

    factory.teardown().await.expect("factory teardown");
}

/// **Enqueue-opts resolution** — opts-to-row stamping is engine-side
/// arithmetic but contract-pinned identical across engines.
/// Delay-over-`run_at` precedence (the row shows the resolved ready
/// time only), relative `expires` resolution to the absolute row
/// expiry, the plain-queue derived stamps on the no-handle-open
/// tx shape (300/3/5/none), and the same defaults with
/// `EnqueueOpts::max_attempts` override riding over them — `get_job`
/// shows resolved values, never the raw opts.
///
/// Wave-5 completion — the two clock-adjacent legs the earlier note
/// deferred pin here: the `run_at`-alone literal leg (a future
/// `run_at` is the row's ready time *verbatim* — no arithmetic, a
/// deterministic equality; a past `run_at` is ready *now* — the
/// boundary-total rule, claimable through the `run_at <= now`
/// predicate within a bounded wait) and the neither-field leg (ready
/// now at the enqueue instant — tolerance-bounded proximity of the
/// stamp to the clock read, never a tight timing assert).
///
/// Contract stamp: ADR-020 §1 (delay wins over `run_at`; `run_at`
/// alone is literal; neither field resolves to now); §2
/// (relative `expires` → absolute row value); §3/§3a (the derived
/// default stamps — plain queue 300/3/5/none, per-job override);
/// ADR-023 §2 (the boundary-total rule — a past ready time is ready
/// now, never rejected); ADR-012 §2 (engine-side arithmetic pinned
/// equivalent).
pub async fn enqueue_opts_resolution(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    const PLAIN_STAMPS: (i64, i64, i64, Option<i64>) = (300, 3, 5, None);

    let mut tx = store.begin_tx().await.unwrap();
    let plain = tx
        .enqueue_tx("work", EnqueueOpts::default(), json!({"shape": "tx"}))
        .await
        .unwrap();
    let precedence = tx
        .enqueue_tx(
            "work",
            EnqueueOpts {
                delay: Some(60),
                run_at: Some(1_500_000_000),
                ..Default::default()
            },
            json!({"shape": "precedence"}),
        )
        .await
        .unwrap();
    let bounded = tx
        .enqueue_tx(
            "work",
            EnqueueOpts {
                expires: Some(90),
                ..Default::default()
            },
            json!({"shape": "bounded"}),
        )
        .await
        .unwrap();
    let overridden = tx
        .enqueue_tx(
            "work",
            EnqueueOpts {
                max_attempts: Some(7),
                ..Default::default()
            },
            json!({"shape": "override"}),
        )
        .await
        .unwrap();
    tx.commit().await.unwrap();

    let queue = store
        .queue("work", alkstore::QueueOpts::default())
        .await
        .unwrap();
    let plain_job = queue.get_job(plain).await.unwrap().unwrap();
    let stamps = (
        plain_job.visibility_timeout_s,
        plain_job.max_attempts,
        plain_job.backoff_base_s,
        plain_job.dead_letter_retention_s,
    );
    assert_eq!(
        stamps, PLAIN_STAMPS,
        "the no-handle-open tx shape stamps the plain-queue derived defaults"
    );

    let precedence_job = queue.get_job(precedence).await.unwrap().unwrap();
    assert_ne!(
        precedence_job.run_at, 1_500_000_000,
        "delay wins over run_at — run_at is not the row's ready time"
    );

    let bounded_job = queue.get_job(bounded).await.unwrap().unwrap();
    let expires_at = bounded_job
        .expires_at
        .expect("relative expires resolves to an absolute row expiry");
    let now = unix_now();
    assert!(
        expires_at > now + 60 && expires_at <= now + 90,
        "expires is relative from enqueue (resolved, not raw): {expires_at} vs now {now}"
    );

    let override_job = queue.get_job(overridden).await.unwrap().unwrap();
    assert_eq!(
        (
            override_job.visibility_timeout_s,
            override_job.max_attempts,
            override_job.backoff_base_s,
            override_job.dead_letter_retention_s
        ),
        (PLAIN_STAMPS.0, 7, PLAIN_STAMPS.2, PLAIN_STAMPS.3),
        "the per-job override rides over the derived defaults, others untouched"
    );

    // Wave-5 clock legs (the run_at-alone and neither-field pins).
    // Future run_at alone: the ready time is the literal — stored
    // verbatim, no arithmetic, no clock read involved (a deterministic
    // equality, not a timing assert).
    let literal = 1_800_000_000;
    let literal_id = queue
        .enqueue(
            json!({"leg": "run-at-literal"}),
            EnqueueOpts {
                run_at: Some(literal),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let literal_job = queue.get_job(literal_id).await.unwrap().unwrap();
    assert_eq!(
        literal_job.run_at, literal,
        "run_at alone is the row's ready time verbatim (delay absent)"
    );

    // Past run_at alone: ready now (the boundary-total rule) —
    // claimable within a bounded wait. The claim ordering (run_at ASC)
    // hands it out ahead of the every same-now row above, so the
    // observation is unambiguous even against the row's other legs.
    let past = unix_now() - 30;
    let past_id = queue
        .enqueue(
            json!({"leg": "run-at-past"}),
            EnqueueOpts {
                run_at: Some(past),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    assert_eq!(
        queue.get_job(past_id).await.unwrap().unwrap().run_at,
        past,
        "a past run_at stores the literal too — ready-now is the claim semantics"
    );
    let claimed_past = tokio::time::timeout(std::time::Duration::from_secs(5), async {
        loop {
            if let Some(h) = queue.claim_one("wave5-past").await.unwrap() {
                let id = h.job().id;
                if id == past_id {
                    break id;
                }
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
    })
    .await
    .expect("a past run_at resolves ready-now: claimable within the bounded wait");
    assert_eq!(
        claimed_past, past_id,
        "the past-run_at row is the ready-now claim (ordering: run_at ASC)"
    );

    // Neither field: ready now at the enqueue instant — the stamp sits
    // close to the clock read (tolerance-bounded proximity, never a
    // tight timing assert).
    let neither_id = queue
        .enqueue(json!({"leg": "neither"}), EnqueueOpts::default())
        .await
        .unwrap();
    let now = unix_now();
    assert!(
        queue
            .get_job(neither_id)
            .await
            .unwrap()
            .unwrap()
            .run_at
            .abs_diff(now)
            <= 5,
        "neither field resolves to now: the stamp is within the tolerance \
         of the clock read"
    );

    factory.teardown().await.expect("factory teardown");
}

/// The backoff row's bounded-wait claim: poll `claim_one` on `queue`
/// until `id` comes out and return that claim's handle (claim
/// ordering keeps single-job queues unambiguous; the deadline is the
/// safety bound — the assertions are on state outcomes, and the claim
/// is consumed here, not re-taken afterwards).
async fn await_claimable(
    queue: &dyn alkstore::Queue,
    id: i64,
    max_wait: std::time::Duration,
    label: &str,
) -> Box<dyn alkstore::JobHandle> {
    let deadline = std::time::Instant::now() + max_wait;
    loop {
        match queue.claim_one("backoff-w").await.unwrap() {
            Some(h) if h.job().id == id => return h,
            _ => {}
        }
        assert!(
            std::time::Instant::now() < deadline,
            "{label}: the row never became ready within the bounded wait"
        );
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }
}

/// **Backoff-curve equivalence across engines** — the equal-jitter
/// exponential curve (ADR-010 §3): `retry(err, None)` computes
/// `delay ∈ [base·2^(a−1)/2, base·2^(a−1)]` from the job's stamped
/// `backoff_base_s` (the attempt index `a` the row's post-claim
/// `attempts`), capped at 1 hour — *the range, not a jitter label, is
/// the definition*, and identical bounds across engines is the
/// equivalence pin (ADR-012 §2: engine-side arithmetic, suite-pinned
/// identical outputs). The row drives claim → `retry(err, None)`
/// cycles on a queue with `backoff_base_s = 2` (integer-second
/// resolution: the ranges \[1,2\], \[2,4\], \[4,8\] land across the first
/// three attempts), reading each computed delay back through
/// `get_job`'s resolved `run_at` minus a clock read taken before the
/// retry, asserting range membership per attempt — with a
/// one-second straddle tolerance on the upper side only (the retry's
/// internal clock read and the suite's may cross a wall second; the
/// lower side is race-free, the retry's clock read cannot precede the
/// suite's). The inter-attempt waits are tolerance-bounded and
/// sequential (the row waits for each `run_at` to pass through
/// bounded-wait claims).
///
/// The explicit-delay override legs: `retry(err, Some(d))` is honored
/// verbatim — `Some(5)` lands the row ready at the retry instant plus
/// exactly 5 s (the same straddle tolerance), and `Some(0)` resolves
/// ready-now (the boundary-total rule: a zero delay is claimable at
/// the retry's own second, within a bounded wait).
///
/// Cap-leg disposition: the 1-hour cap is not suite-pinnable (reaching
/// capped attempts requires waiting out minute-scale delays); it stays
/// pinned engine-side — both engines' unit tests pin the curve against
/// the ADR text including the cap.
///
/// Contract stamp: ADR-010 §3 (the equal-jitter exponential, the range
/// as definition, the 1-hour cap); ADR-020 §1 (the explicit-delay
/// override resolves to the row's ready time — `Some(0)` ready now);
/// ADR-012 §2 (engine-side arithmetic pinned identical); ADR-023 §2
/// (the boundary-total rule — a zero delay is ready now, never
/// rejected).
pub async fn backoff_curve_equivalence(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");
    let queue = store
        .queue(
            "curve",
            alkstore::QueueOpts {
                max_attempts: 7,
                backoff_base_s: 2,
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let id = queue
        .enqueue(json!({"leg": "curve"}), EnqueueOpts::default())
        .await
        .unwrap();

    // Three curve cycles at the first three attempts (base 2 s:
    // ranges [1,2], [2,4], [4,8] — the integer-second row resolution
    // kept honest by the 1-second straddle tolerance on the upper
    // side). The now-read before the retry makes the lower bound
    // race-free: the retry's internal clock read cannot precede it, so
    // the observed value over-reads the delay, never under-reads.
    for (attempt, lower, upper) in [(1i64, 1i64, 2i64), (2, 2, 4), (3, 4, 8)] {
        let held = await_claimable(
            queue.as_ref(),
            id,
            std::time::Duration::from_secs(upper as u64 + 10),
            &format!("attempt {attempt}"),
        )
        .await;
        assert_eq!(
            held.job().attempts,
            attempt,
            "the attempt index is the claim count"
        );
        let now = unix_now();
        assert!(
            held.retry(Some("boom".to_string()), None)
                .await
                .expect("an in-window retry lands")
        );
        let job = queue
            .get_job(id)
            .await
            .unwrap()
            .expect("the retried row lives");
        assert_eq!(
            job.state,
            JobState::Pending,
            "the retry left the row pending"
        );
        let observed = job.run_at - now;
        assert!(
            (lower..=upper + 1).contains(&observed),
            "attempt {attempt}: the computed delay must land in \
             [{lower}, {upper}] (one-second straddle tolerance), got {observed}"
        );
    }

    // Explicit-delay override: Some(5) is honored — ready at the retry
    // instant plus exactly 5 s (the same straddle tolerance), not the
    // attempt-4 curve range [8, 16].
    let held = await_claimable(
        queue.as_ref(),
        id,
        std::time::Duration::from_secs(10),
        "Some(5) leg",
    )
    .await;
    let now = unix_now();
    assert!(
        held.retry(Some("boom".to_string()), Some(5))
            .await
            .expect("an in-window retry lands")
    );
    let job = queue.get_job(id).await.unwrap().unwrap();
    let observed = job.run_at - now;
    assert!(
        (5..=6).contains(&observed),
        "retry(err, Some(5)) is honored verbatim (± one second straddle), \
         got {observed}"
    );

    // Some(0): the boundary-total rule — ready now. The retry's own
    // second already satisfies the claim predicate, so the bounded
    // wait is short and the row lands promptly.
    let held = await_claimable(
        queue.as_ref(),
        id,
        std::time::Duration::from_secs(8),
        "Some(0) leg",
    )
    .await;
    assert!(
        held.retry(Some("boom".to_string()), Some(0))
            .await
            .expect("an in-window retry lands")
    );
    await_claimable(
        queue.as_ref(),
        id,
        std::time::Duration::from_secs(5),
        "Some(0) resolves ready-now (the boundary-total rule)",
    )
    .await;

    factory.teardown().await.expect("factory teardown");
}

/// **Receiver close and save arms** — the stream subscription
/// receiver's pinned arms (ADR-021 §5): `recv()`'s error arm carries
/// `Database` only, close is terminal `None` (this SQLite column:
/// store disposal closes the receiver — the cause asymmetry is
/// documented per engine), `try_recv` distinguishes idle (`Ok(None)`)
/// from closed (`Err(Error::Closed)`), the receiver-form save is a
/// no-op before any event yields, lands the last-yielded offset
/// exactly once yielded, and regressions are silent no-ops — direct
/// and receiver forms interleave without regressing one another
/// (ADR-019 §6).
///
/// Contract stamp: ADR-021 §5 (receiver error/close arms —
/// `Database`-only `Err`, terminal `None` close, SQLite closes on
/// watcher death); ADR-019 §6 (save composition — monotone, forms
/// interleave); ADR-008 §8 (explicit saves only); ADR-023 §2
/// (annotated 2026-10-11 — the boundary/offset row's totality covers
/// `save_offset`: a negative seed is stored as the checkpoint, this
/// row's negative-leg pins).
pub async fn receiver_close_and_save_arms(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    let stream = store.stream("events").await.unwrap();
    let o1 = stream.publish(json!({"n": 1})).await.unwrap();
    let o2 = stream.publish(json!({"n": 2})).await.unwrap();

    let mut rx = stream.subscribe("c").await.unwrap();
    assert_eq!(rx.offset(), 0, "no event yielded yet");
    assert!(
        rx.save_offset().is_ok(),
        "the pre-yield save is a no-op: Ok on any engine"
    );
    // The no-op never wrote a checkpoint: a fresh consumer read-back
    // (from 0) is empty and the direct form's fresh inspection is 0.
    assert_eq!(
        store
            .stream("events")
            .await
            .unwrap()
            .get_offset("c")
            .await
            .unwrap(),
        0,
        "the pre-yield save touched no checkpoint"
    );

    let mut last = rx
        .recv()
        .await
        .expect("the drained attach read yields o1")
        .unwrap();
    assert_eq!(last.offset, o1, "the attach read yields offset ASC");
    // Drain the rest of the attach read (o2 was published pre-attach —
    // the attach read drains to the tail, so it is pending in the
    // receiver's feed too; the idle assert below reads an empty tail).
    while let Some(next) = rx.try_recv().unwrap() {
        assert!(next.offset > last.offset, "the drain stays offset ASC");
        last = next;
    }
    rx.save_offset().unwrap();
    assert_eq!(
        rx.offset(),
        last.offset,
        "the save lands the last-yielded offset exactly"
    );
    assert!(
        store
            .stream("events")
            .await
            .unwrap()
            .get_offset("c")
            .await
            .unwrap()
            > 0,
        "the checkpoint is stored (the no-op save never wrote it; this one did)"
    );

    // Idle: the feed drained to the tail — try_recv is Ok(None) (not
    // closed).
    assert_eq!(
        rx.try_recv().unwrap(),
        None,
        "an attached drained receiver idles, it is not closed"
    );

    // Direct-then-receiver composition without regression: a direct
    // save ahead moves the cursor; the receiver's subsequent save
    // cannot rewind it (monotone).
    stream.save_offset("c", o2).await.unwrap();
    let direct = store
        .stream("events")
        .await
        .unwrap()
        .get_offset("c")
        .await
        .unwrap();
    assert!(direct >= o2, "the direct save landed");
    rx.save_offset().unwrap();
    assert!(
        store
            .stream("events")
            .await
            .unwrap()
            .get_offset("c")
            .await
            .unwrap()
            >= direct,
        "the receiver form never regresses the direct form's save"
    );

    // Boundary-total save (ADR-023 §2, annotated 2026-10-11): a
    // negative offset is stored as the checkpoint — the argument is
    // total, no clamping, no rejection — and reads anchored there
    // resume forward identically (every event offset lies above the
    // negative anchor, so the anchored read yields the log from the
    // start). Fresh consumer, so this leg also pins the *seed* case —
    // the only arm where a clamping realization would diverge.
    stream.save_offset("c2", -5).await.unwrap();
    assert_eq!(
        store
            .stream("events")
            .await
            .unwrap()
            .get_offset("c2")
            .await
            .unwrap(),
        -5,
        "a negative seed is stored as the checkpoint (totality, no clamp)"
    );
    let resumed = store
        .stream("events")
        .await
        .unwrap()
        .read_from_consumer("c2", 10)
        .await
        .unwrap();
    assert_eq!(
        resumed.iter().map(|e| e.offset).collect::<Vec<_>>(),
        vec![o1, o2],
        "reads anchored at the negative checkpoint resume forward: the log from the start"
    );

    // Terminal close on disposal (this engine's documented close
    // cause: the watcher's death guard clears the feed — ADR-021 §5's
    // SQLite arm; the store handle dropping is the disposal).
    drop(store);
    let closed = rx.recv().await;
    assert!(
        closed.is_none(),
        "store disposal closes the receiver terminally: got {closed:?}"
    );
    assert!(
        matches!(rx.try_recv(), Err(Error::Closed)),
        "post-close try_recv is Err(Closed), not idle"
    );
}

/// **Wake semantics under the pinned `WakeReceiver` shapes** — the
/// wake contract's pinnable core, tolerance-bounded (wakes are
/// best-effort hints — the row asserts state outcomes, never delivery
/// counts or latencies). A pre-attached listener receives a wake after
/// a committed `notify` on its channel, with all three recv forms
/// exercised (`recv`, `try_recv`, `recv_timeout`) and every wake
/// observed carrying the listened channel — the one piece of semantic
/// content a wake carries (ADR-008 §3). A burst of notifies yields
/// *at least one* wake, never an exact per-notify count (coalescing is
/// the documented engine asymmetry — possibly coalesced on SQLite's
/// 1-slot feed, per-notify on pg; the row pins the floor, not the
/// shape). A listener attached *after* a commit never sees that
/// commit's notify (`recv_timeout` idles through a bounded absence
/// window — no replay; the notify commits before the attach, and pg's
/// no-replay hole — a commit during a forwarder gap is never
/// re-delivered — and SQLite's burst coalescing are both legal under
/// this pin). Wakes from a different channel never arrive on this
/// listener: after a foreign-channel notify, the bounded window
/// surfaces no wake carrying the foreign consumer channel (SQLite's
/// documented overtrigger may deliver a wake on the same commit, but
/// it carries the *listened* channel only — nothing about what was
/// notified; the pg fanout skips non-LISTENed channels, and the
/// reserved reconnect-wake carries its own reserved name — that
/// reserved straggler is a tolerated engine arm, not a violation), and
/// a same-channel positive control still delivers — the silence was
/// channel scoping, never a dead listener.
///
/// Cross-reference: the failure-surface close arms — SQLite's
/// watcher-death `recv() -> None` and the pg synthetic reconnect-wake
/// — are the engine-differing close arms, already pinned engine-side
/// and in `receiver_close_and_save_arms` (the stream receiver's
/// disposal leg); this row does not re-pin them — the reserved
/// reconnect-wake appears here only as a tolerated straggler.
///
/// Contract stamp: ADR-006 (the wake contract — opaque wake +
/// re-read, best-effort hints not per-commit delivery, coalescing /
/// overtrigger asymmetry, the channel-scoped fanout); ADR-008 §3 (the
/// pinned `Wake { channel }` / recv forms — the wake's channel field
/// matches the listened channel; `recv_timeout`'s expiry-without-wake
/// is the documented `Ok(None)` idle arm).
pub async fn wake_receiver_shapes(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    // Delivery + channel identity: a pre-attached listener, a
    // committed notify on its channel, every recv form once.
    let mut rx = store.listen("wake-a").await.expect("listen attaches");

    store
        .notify("wake-a", json!({"n": 1}))
        .await
        .expect("notify commits");
    let wake = tokio::time::timeout(std::time::Duration::from_secs(10), rx.recv())
        .await
        .expect("the committed notify delivers a wake — the bounded wait is enough")
        .expect("the receiver is open through a healthy listen");
    assert_eq!(
        wake.channel, "wake-a",
        "the recv wake's channel field matches the listened channel"
    );

    store
        .notify("wake-a", json!({"n": 2}))
        .await
        .expect("notify commits");
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
    let wake = loop {
        match rx.try_recv() {
            Ok(Some(w)) => break w,
            Ok(None) => {}
            Err(e) => panic!("the receiver is open through a healthy listen: {e}"),
        }
        assert!(
            std::time::Instant::now() < deadline,
            "the committed notify delivered no wake within the bounded wait"
        );
        tokio::time::sleep(std::time::Duration::from_millis(25)).await;
    };
    assert_eq!(
        wake.channel, "wake-a",
        "the try_recv wake's channel field matches the listened channel"
    );

    store
        .notify("wake-a", json!({"n": 3}))
        .await
        .expect("notify commits");
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
    let wake = loop {
        match rx
            .recv_timeout(std::time::Duration::from_millis(50))
            .await
            .expect("recv_timeout's Err is closed/storage-failure only, idle is Ok(None)")
        {
            Some(w) => break w,
            None => {
                assert!(
                    std::time::Instant::now() < deadline,
                    "the committed notify delivered no wake within the bounded wait"
                );
            }
        }
    };
    assert_eq!(
        wake.channel, "wake-a",
        "the recv_timeout wake's channel field matches the listened channel"
    );

    // The burst floor: three committed notifies, at least one wake.
    // The count is never pinned — coalescing is the documented engine
    // asymmetry (SQLite's 1-slot feed; pg per-notify), the contract
    // leaves the shape engine-owned.
    for n in 0..3 {
        store
            .notify("wake-a", json!({"burst": n}))
            .await
            .expect("notify commits");
    }
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
    let mut heard = 0;
    while heard == 0 {
        match rx
            .recv_timeout(std::time::Duration::from_millis(50))
            .await
            .expect("the receiver is open through a healthy listen")
        {
            Some(w) => {
                assert_eq!(
                    w.channel, "wake-a",
                    "the burst wake's channel field matches the listened channel"
                );
                heard += 1;
            }
            None => {
                assert!(
                    std::time::Instant::now() < deadline,
                    "a burst of notifies yields at least one wake — never zero \
                     within the bounded wait"
                );
            }
        }
    }

    // No replay: a listener attached after a commit never sees that
    // commit's notify. The pre-settle sleep closes SQLite's baseline
    // race — the watcher's tick past the settle has consumed the
    // commit's version change (a past-window spurious wake arriving
    // before the late attach is the documented overtrigger; the pin
    // starts at the attach).
    store
        .notify("wake-b", json!({"late": true}))
        .await
        .expect("notify commits");
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
    let mut late = store
        .listen("wake-b")
        .await
        .expect("the late listen attaches");
    let heard = late
        .recv_timeout(std::time::Duration::from_millis(400))
        .await
        .expect("the absence window is Ok(None) or a wake — never a hang");
    match heard {
        None => {}
        Some(w) if w.channel == alkstore::RESERVED_LISTENER_RECONNECTED => {}
        Some(w) => panic!(
            "a listener attached after a commit never sees that commit's \
             notify — got a wake carrying {}",
            w.channel
        ),
    }

    // Channel-scoped delivery: a foreign-channel notify never wakes
    // this listener. The bounded window drains no wake carrying the
    // foreign channel; what may arrive is the listened channel (SQLite's
    // legal overtrigger) or the reserved reconnect straggler (pg).
    let mut scoped = store.listen("scope-a").await.expect("listen attaches");
    store
        .notify("scope-b", json!({"foreign": true}))
        .await
        .expect("notify commits");
    let deadline = std::time::Instant::now() + std::time::Duration::from_millis(400);
    loop {
        let wake = scoped
            .recv_timeout(std::time::Duration::from_millis(100))
            .await
            .expect("the scoped listener is open");
        match wake {
            None => break,
            Some(w)
                if w.channel != "scope-a"
                    && w.channel != alkstore::RESERVED_LISTENER_RECONNECTED =>
            {
                panic!(
                    "a wake from a different channel never arrives on this \
                     listener: {}",
                    w.channel
                );
            }
            Some(_) => {}
        }
        assert!(
            std::time::Instant::now() < deadline,
            "the scoped silence window overflowed with unnamed wakes"
        );
    }

    // Positive control: a same-channel notify still delivers — the
    // idle above was channel scoping, not a dead listener.
    store
        .notify("scope-a", json!({"own": true}))
        .await
        .expect("notify commits");
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
    let wake = loop {
        match scoped
            .recv_timeout(std::time::Duration::from_millis(50))
            .await
            .expect("the scoped listener is open")
        {
            Some(w) => break w,
            None => {
                assert!(
                    std::time::Instant::now() < deadline,
                    "the same-channel positive control delivered no wake — \
                     the listener is dead, not idle, and the scoping pin \
                     above is vacuous"
                );
            }
        }
    };
    assert_eq!(
        wake.channel, "scope-a",
        "the control wake's channel field matches the listened channel"
    );

    factory.teardown().await.expect("factory teardown");
}

/// **`PayloadTooLarge` produced pg-side** — the Postgres engine
/// rejects an oversized `notify`/`notify_tx` payload client-side with
/// the typed `PayloadTooLarge` carrying the limit, **before any round
/// trip** (no listener hears the rejected payload — not delivered,
/// not truncated), at exactly the boundary (`limit` bytes = rejection;
/// under it = delivered). This is the pg column's arm of the asymmetry
/// row (the SQLite-side never-produced arm is its own row — the same
/// engine-agnostic match posture, the non-occurring arm simply never
/// fires there).
///
/// The limit read comes from the produced variant itself (the limit in
/// the variant is the runtime descriptor — ADR-016 §2's one
/// normative home), not a hardcoded number.
///
/// Contract stamp: ADR-008 §5 (universal variant, the limit in the
/// variant); ADR-016 §5 (the occurrence asymmetry — pg produces it
/// client-side at the 8000-byte wire budget); ADR-020 §4 (the
/// measured quantity is the serde_json serialization — the stored-
/// bytes form).
pub async fn payload_too_large_produced_on_pg(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    let mut listener = store.listen("big").await.expect("listen attaches");

    // Oversized: the typed arm, client-side (rejected before the
    // round trip — a truncating relay would deliver a mangled half).
    // The contract pins the number: ≤ 8000 bytes on Postgres.
    let over = json!({"blob": "x".repeat(8_000)});
    match store.notify("big", over).await {
        Err(Error::PayloadTooLarge { limit }) => assert_eq!(limit, 8000),
        other => panic!("oversized notify on pg must be PayloadTooLarge, got {other:?}"),
    }
    // The tx twin matches identically (engine-agnostic code writes one
    // match for both paths).
    let mut tx = store.begin_tx().await.expect("begin_tx opens");
    match tx
        .notify_tx("big", json!({"blob": "x".repeat(8_000)}))
        .await
    {
        Err(Error::PayloadTooLarge { .. }) => {}
        other => panic!("oversized notify_tx on pg must be PayloadTooLarge, got {other:?}"),
    }
    tx.commit()
        .await
        .expect("commit (the rejects fired before any round trip)");

    // Nothing delivered while the rejects fired: the boundary is a
    // client-side rejection, not a lossy truncation.
    let idle = listener
        .recv_timeout(std::time::Duration::from_millis(150))
        .await
        .expect("idle is Ok(None), never a timeout signal");
    assert!(
        idle.is_none(),
        "rejected notifies deliver nothing, got {idle:?}"
    );

    factory.teardown().await.expect("factory teardown");
}

/// The suite's lapse-tolerance sleep: bounded well past a 1-second
/// stamp's resolution (a stamp set on second `T` has lapsed for any
/// clock read `> T + 1`; 2.5 s guarantees it, 3.5 s the `expires: 2`
/// stamps). Timing posture: state outcomes are asserted, never
/// durations.
///
/// Harness posture: **no blocking calls inside suite rows** — this
/// helper is async (`tokio::time::sleep`) and every row awaits it;
/// the mem engine's suite column runs the current-thread flavor
/// ([ADR-024](../../docs/architecture/decisions/024-mem-engine.md)
/// §5), which is what catches a blocking stall here.
async fn past_stamp_sleep(millis: u64) {
    tokio::time::sleep(std::time::Duration::from_millis(millis)).await;
}

/// **Job-handle validity predicate** — the uniform rule (processing
/// state + unexpired claim deadline) on every handle op: an in-window
/// heartbeat lands (renewal) and an in-window ack deletes
/// (delete-on-ack); the late heartbeat is refused exactly at the
/// lapsed claim — the short-stamp row claims with a 1 s visibility
/// stamp and tolerance-sleeps past it (the sleep-then-assert-state
/// shape is sequential single-store and fits the suite's determinism
/// posture); the post-lapse ack does not land (at-least-once: the
/// worker's ack does not land after lapse), and `retry`'s refusal is
/// `false` with the row untouched (state, attempts, claimant, and
/// deadline all as the claim left them). `fail(None)` after lapse is
/// refused identically — deadline lapse refuses *all* ops, not just
/// heartbeat. `ack_batch` applies the predicate per id: a lapsed id
/// and a never-claimed id are silently not counted — the returned
/// count reflects only the live acks.
///
/// The engine-default strings land identically: `fail(None)` records
/// `"failed"`, and retry-at-exhausted-budget dead-letters with
/// `"max attempts exceeded"` (the reclaim-exhaustion arm of that
/// string pins in the queue-depth row's exhaustion leg).
///
/// Contract stamp: ADR-010 §2 (the uniform validity predicate — the
/// late-heartbeat refusal, post-lapse ack refusal); ADR-010 §1
/// (delete-on-ack; `ack_batch`'s per-id predicate, partial success
/// ordinary); ADR-010 §4 (the engine-default strings); ADR-019 §3
/// (the claim handle's op surface — one-shot consumables, repeatable
/// heartbeat); ADR-008 §5 (refusals are `false`, never errors);
/// ADR-023 §2 (annotated 2026-10-11 — the boundary/offset row's
/// totality covers `heartbeat(extend)`: a negative extension is the
/// relative-instant void this row's negative-extend leg pins).
pub async fn job_handle_validity_predicate(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    let slow = store
        .queue("handles", alkstore::QueueOpts::default())
        .await
        .unwrap();
    let fast = store
        .queue(
            "lapse",
            alkstore::QueueOpts {
                visibility_timeout_s: 1,
                ..Default::default()
            },
        )
        .await
        .unwrap();

    // In-window ops land: renewal + delete-on-ack under the live
    // claim.
    let live_id = fast
        .enqueue(json!({"leg": "in-window"}), EnqueueOpts::default())
        .await
        .unwrap();
    let held = fast
        .claim_one("w1")
        .await
        .unwrap()
        .expect("a ready row claims");
    assert_eq!(held.job().id, live_id);
    assert_eq!(held.job().state, JobState::Processing);
    assert!(
        held.heartbeat(600).await.unwrap(),
        "an in-window heartbeat lands (renewal)"
    );
    let mid = fast
        .get_job(live_id)
        .await
        .unwrap()
        .expect("the row still lives mid-claim");
    assert_eq!(mid.state, JobState::Processing);
    assert!(held.ack().await.unwrap(), "an in-window ack lands");
    assert!(
        fast.get_job(live_id).await.unwrap().is_none(),
        "delete-on-ack completed work leaves no row"
    );

    // Boundary-total heartbeat (ADR-023 §2, annotated 2026-10-11): a
    // negative extension is a relative instant — the renewed deadline
    // `now + extend` sits in the past, voiding the claim through the
    // validity predicate on the *next* op, never an error. The
    // heartbeat itself reports the successful renewal (`true`) while
    // the row was processing and unexpired at call time; the voided
    // claim then refuses all handle ops (ack here) and the row
    // survives as processing. Six-hundred-second margin: fully past,
    // deterministic without a sleep. Own queue: the voided row is
    // expired-processing (re-claimable), and this row's later legs
    // assert claim FIFO on their queues.
    let voidq = store
        .queue("voided", alkstore::QueueOpts::default())
        .await
        .unwrap();
    let voided_id = voidq
        .enqueue(json!({"leg": "negative-extend"}), EnqueueOpts::default())
        .await
        .unwrap();
    let voided = voidq
        .claim_one("w1")
        .await
        .unwrap()
        .expect("the ready row claims");
    assert_eq!(voided.job().id, voided_id);
    assert!(
        voided.heartbeat(-600).await.unwrap(),
        "a negative extension lands while the claim is unexpired (totality)"
    );
    assert!(
        !voided.ack().await.unwrap(),
        "the negative extension voided the claim: ack refuses through the validity predicate"
    );
    assert_eq!(
        voidq
            .get_job(voided_id)
            .await
            .unwrap()
            .expect("the voided row was not deleted or dead-lettered")
            .state,
        JobState::Processing,
        "the voided claim's row survives as processing (re-claimable by the queue)"
    );

    // Lapse legs: one claim per refusal op, all stamped with the 1 s
    // visibility, one tolerance sleep past every deadline at once.
    let ack_refused = fast
        .enqueue(json!({"leg": "ack-refused"}), EnqueueOpts::default())
        .await
        .unwrap();
    let retry_refused = fast
        .enqueue(json!({"leg": "retry-refused"}), EnqueueOpts::default())
        .await
        .unwrap();
    let fail_refused = fast
        .enqueue(json!({"leg": "fail-refused"}), EnqueueOpts::default())
        .await
        .unwrap();
    let lapsed_ack = fast.claim_one("w2").await.unwrap().unwrap();
    let lapsed_retry = fast.claim_one("w2").await.unwrap().unwrap();
    let lapsed_fail = fast.claim_one("w2").await.unwrap().unwrap();
    assert_eq!(
        (
            lapsed_ack.job().id,
            lapsed_retry.job().id,
            lapsed_fail.job().id
        ),
        (ack_refused, retry_refused, fail_refused),
        "claims hand out in enqueue order (equal priority)"
    );
    let retry_before = lapsed_retry.job().clone();
    past_stamp_sleep(2_500).await;
    assert!(
        !lapsed_ack.heartbeat(600).await.unwrap(),
        "late heartbeat must be refused: the deadline has lapsed"
    );
    assert!(
        !lapsed_retry
            .retry(Some("boom".to_string()), Some(5))
            .await
            .unwrap(),
        "retry's refusal under the lapsed claim is false"
    );
    assert!(
        !lapsed_fail.fail(None).await.unwrap(),
        "deadline lapse refuses fail identically — the predicate is uniform"
    );
    let retry_after = fast
        .get_job(retry_refused)
        .await
        .unwrap()
        .expect("the retried-at row survived");
    assert_eq!(
        retry_after, retry_before,
        "retry's refusal leaves the row untouched: the claim keeps pre-call state"
    );

    // ack_batch applies the predicate per id: the live id counts, the
    // lapsed id and a never-claimed id silently do not. (Own queue —
    // the lapsed legs above left reclaimable rows in `lapse`, and any
    // later claim there would reclaim FIFO-first. The live id survives
    // the 1 s stamp the same way the lapsed one lapsed: a renewal.)
    let batch = store
        .queue(
            "batch",
            alkstore::QueueOpts {
                visibility_timeout_s: 1,
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let batch_live_id = batch
        .enqueue(json!({"leg": "batch-live"}), EnqueueOpts::default())
        .await
        .unwrap();
    let batch_lapse_id = batch
        .enqueue(json!({"leg": "batch-lapsed"}), EnqueueOpts::default())
        .await
        .unwrap();
    let batch_live = batch.claim_one("w3").await.unwrap().unwrap();
    let batch_lapsed = batch.claim_one("w3").await.unwrap().unwrap();
    assert_eq!(
        (batch_live.job().id, batch_lapsed.job().id),
        (batch_live_id, batch_lapse_id),
        "claims hand out in enqueue order (equal priority)"
    );
    assert!(
        batch_live.heartbeat(600).await.unwrap(),
        "the renewal keeps this id's claim live through the batch"
    );
    past_stamp_sleep(2_500).await;
    let counted = batch
        .ack_batch(&[batch_live_id, batch_lapse_id, 987_654_321])
        .await
        .unwrap();
    assert_eq!(
        counted, 1,
        "the returned count reflects only the live acks (lapsed and nonexistent ids not counted)"
    );
    assert!(
        batch.get_job(batch_live_id).await.unwrap().is_none(),
        "the counted id acked (deleted)"
    );
    let lapsed_alive = batch
        .get_job(batch_lapse_id)
        .await
        .unwrap()
        .expect("the lapsed id's row survives the batch");
    assert_eq!(
        lapsed_alive.state,
        JobState::Processing,
        "the lapsed id was silently not counted, not deleted"
    );

    // The engine-default strings (fail(None); retry at exhausted
    // budget) on a long-visibility queue — no lapse interference.
    let failed = slow
        .enqueue(
            json!({"leg": "fail-default"}),
            EnqueueOpts {
                max_attempts: Some(1),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let held_fail = slow.claim_one("w4").await.unwrap().unwrap();
    assert_eq!(held_fail.job().id, failed);
    assert!(held_fail.fail(None).await.unwrap());
    let dead_failed = slow
        .get_job(failed)
        .await
        .unwrap()
        .expect("fail(None) dead-letters");
    assert_eq!(dead_failed.state, JobState::Dead);
    assert_eq!(dead_failed.last_error.as_deref(), Some("failed"));
    assert!(dead_failed.died_at.is_some());

    let exhausting = slow
        .enqueue(
            json!({"leg": "exhaustion"}),
            EnqueueOpts {
                max_attempts: Some(1),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let held_retry = slow.claim_one("w5").await.unwrap().unwrap();
    assert_eq!(held_retry.job().id, exhausting);
    assert!(held_retry.retry(None, Some(5)).await.unwrap());
    let dead_exhausted = slow
        .get_job(exhausting)
        .await
        .unwrap()
        .expect("retry at exhausted budget dead-letters");
    assert_eq!(dead_exhausted.state, JobState::Dead);
    assert_eq!(
        dead_exhausted.last_error.as_deref(),
        Some("max attempts exceeded"),
        "the engine-default exhaustion string lands identically"
    );
    assert!(dead_exhausted.died_at.is_some());

    factory.teardown().await.expect("factory teardown");
}

/// **Queue-depth reclaim and dead-letter** — the ADR-010 depth
/// properties over a 1 s-visibility queue: a visibility reclaim
/// consumes an attempt (the expired claim reclaims to a second
/// claimant with `attempts` incremented and `claimed_at` refreshed —
/// the original holder's ops refuse after the reclaim), exhaustion
/// dead-letters into get_job-visible dead storage (`"max attempts
/// exceeded"` + `died_at` — the pre-claim sweep shape: a row whose
/// reclaim consumed its budget dead-letters at the next claim), and
/// `cancel` is an unconditional delete of either state — not an
/// interrupt: the row is gone (live and dead alike refuse `cancel`
/// with `false`), and the holder's next op refuses with the same
/// shape as a lapse.
///
/// Contract stamp: ADR-010 §2 (a reclaim consumes an attempt; the
/// original holder refuses post-reclaim); ADR-010 §3 (exhaustion via
/// reclaim dead-letters); ADR-010 §1 (cancel — unconditional, not an
/// interrupt); ADR-010 §4 (dead-visible `get_job` with
/// `last_error`/`died_at`; the exhaustion string).
pub async fn queue_depth_reclaim_and_dead_letter(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");
    let depth = store
        .queue(
            "depth",
            alkstore::QueueOpts {
                visibility_timeout_s: 1,
                max_attempts: 2,
                ..Default::default()
            },
        )
        .await
        .unwrap();

    // Visibility reclaim consumes an attempt.
    let reclaimed = depth
        .enqueue(json!({"leg": "reclaim"}), EnqueueOpts::default())
        .await
        .unwrap();
    let first = depth.claim_one("w1").await.unwrap().unwrap();
    assert_eq!(first.job().id, reclaimed);
    assert_eq!(first.job().attempts, 1);
    let t1 = first.job().claimed_at.expect("claim stamps claimed_at");
    let e1 = first
        .job()
        .claim_expires_at
        .expect("claim stamps the deadline");
    past_stamp_sleep(2_500).await;

    let second = depth
        .claim_one("w2")
        .await
        .unwrap()
        .expect("the lapsed claim reclaims");
    assert_eq!(second.job().id, reclaimed, "the same row reclaims");
    assert_eq!(second.job().attempts, 2, "a reclaim consumes an attempt");
    assert_eq!(second.job().state, JobState::Processing);
    assert_eq!(second.job().worker_id.as_deref(), Some("w2"));
    let t2 = second
        .job()
        .claimed_at
        .expect("the reclaim re-stamps claimed_at");
    let e2 = second
        .job()
        .claim_expires_at
        .expect("the reclaim re-stamps the deadline");
    assert!(t2 > t1, "claimed_at is refreshed by the reclaim");
    assert!(e2 > e1, "the deadline is refreshed, not inherited");
    assert!(
        !first.heartbeat(600).await.unwrap(),
        "the original holder refuses after the reclaim"
    );
    assert!(second.ack().await.unwrap(), "the reclaiming holder acks");
    assert!(depth.get_job(reclaimed).await.unwrap().is_none());

    // Exhaustion: the reclaim that spends the budget, then the
    // next claim's pre-claim sweep dead-letters the row.
    let exhausted = depth
        .enqueue(json!({"leg": "exhausted"}), EnqueueOpts::default())
        .await
        .unwrap();
    depth.claim_one("w1").await.unwrap().unwrap();
    past_stamp_sleep(2_500).await;
    let at_budget = depth.claim_one("w2").await.unwrap().unwrap();
    assert_eq!(at_budget.job().id, exhausted);
    assert_eq!(
        at_budget.job().attempts,
        at_budget.job().max_attempts,
        "the reclaim spent the budget"
    );
    past_stamp_sleep(2_500).await;
    assert!(
        depth.claim_one("w3").await.unwrap().is_none(),
        "the budget-spent row dead-letters instead of reclaiming"
    );
    let dead = depth
        .get_job(exhausted)
        .await
        .unwrap()
        .expect("the dead row is get_job-visible");
    assert_eq!(
        dead.state,
        JobState::Dead,
        "exhaustion moves to dead storage"
    );
    assert_eq!(dead.last_error.as_deref(), Some("max attempts exceeded"));
    assert!(dead.died_at.is_some());

    // Cancel is an unconditional delete — either state, any holder —
    // and not an interrupt: the holder's next op refuses the same
    // shape as a lapse, and nothing re-enters the claim path.
    let cancelled_live = depth
        .enqueue(json!({"leg": "cancel-live"}), EnqueueOpts::default())
        .await
        .unwrap();
    let holder = depth.claim_one("w1").await.unwrap().unwrap();
    assert_eq!(holder.job().id, cancelled_live);
    assert!(depth.cancel(cancelled_live).await.unwrap());
    assert!(
        depth.get_job(cancelled_live).await.unwrap().is_none(),
        "cancel deletes the row whole: no re-claim, no dead row"
    );
    assert!(
        !holder.heartbeat(600).await.unwrap(),
        "not an interrupt — the holder's next op refuses identically"
    );

    let cancelled_pending = depth
        .enqueue(json!({"leg": "cancel-pending"}), EnqueueOpts::default())
        .await
        .unwrap();
    assert!(depth.cancel(cancelled_pending).await.unwrap());
    assert!(
        depth.claim_one("w1").await.unwrap().is_none(),
        "a cancelled pending row never becomes claimable"
    );

    assert!(
        !depth.cancel(987_654_321).await.unwrap(),
        "no missing live row cancels"
    );
    assert!(
        !depth.cancel(exhausted).await.unwrap(),
        "a dead row is outside cancel's live-row reach"
    );
    assert!(
        depth.get_job(exhausted).await.unwrap().is_some(),
        "cancel on a dead id deletes nothing"
    );

    factory.teardown().await.expect("factory teardown");
}

/// **Sweep: no stranded rows** — `sweep_expired()` moves *every*
/// past-expiry row (any state: the pending-expired and the
/// processing-expired alike — an enqueued job with an `expires`
/// deadline is eventually in exactly one of pending, processing, or
/// dead, never stuck unreachable), stamps the `"expired"` string, and
/// enforces dead-letter retention (rows past a set
/// `dead_letter_retention_s` delete; retention `None` = forever) —
/// the count being the moved + retention-deleted sum. Unexpired and
/// never-expiring rows stay untouched.
///
/// The M-1 follow-up (the sweep's retention half rolling back with
/// the sweep on a retention-DELETE failure) is pinned engine-side —
/// the failure is storage-level, not injectable through the contract
/// surface; see the SQLite substrate's
/// `sweep_rolls_back_the_retention_half_with_the_move` and this
/// task's Notes for the disposition.
///
/// Contract stamp: ADR-010 §5 (the no-stranded-rows property, both
/// states, and the retention enforcement); ADR-010 §4 (the
/// `"expired"` string; retention default = forever).
pub async fn sweep_no_stranded_rows(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    let sweeping = store
        .queue("sweepme", alkstore::QueueOpts::default())
        .await
        .unwrap();
    // The processing-expired row is claimed first (it is the only row
    // at claim time) with a 2 s expiry stamp; the pending-expired row
    // enqueues after the claim so FIFO never hands it out.
    let processing = sweeping
        .enqueue(
            json!({"leg": "expired-processing"}),
            EnqueueOpts {
                expires: Some(2),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let held = sweeping.claim_one("w1").await.unwrap().unwrap();
    assert_eq!(held.job().id, processing, "the expiry-stamped row claims");
    let pending = sweeping
        .enqueue(
            json!({"leg": "expired-pending"}),
            EnqueueOpts {
                expires: Some(2),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let unexpired = sweeping
        .enqueue(
            json!({"leg": "unexpired"}),
            EnqueueOpts {
                expires: Some(3600),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let never = sweeping
        .enqueue(json!({"leg": "never-expiring"}), EnqueueOpts::default())
        .await
        .unwrap();
    past_stamp_sleep(3_500).await;

    let moved = sweeping.sweep_expired().await.unwrap();
    assert_eq!(
        moved, 2,
        "every past-expiry row moves — both states, exactly the expired ones"
    );
    let dead_processing = sweeping
        .get_job(processing)
        .await
        .unwrap()
        .expect("the processing-expired row moved to dead");
    assert_eq!(dead_processing.state, JobState::Dead);
    assert_eq!(dead_processing.last_error.as_deref(), Some("expired"));
    assert!(dead_processing.died_at.is_some());
    let dead_pending = sweeping
        .get_job(pending)
        .await
        .unwrap()
        .expect("the pending-expired row moved to dead");
    assert_eq!(dead_pending.state, JobState::Dead);
    assert_eq!(dead_pending.last_error.as_deref(), Some("expired"));
    for (id, label) in [(unexpired, "unexpired"), (never, "never-expiring")] {
        let live = sweeping.get_job(id).await.unwrap().unwrap();
        assert_eq!(live.state, JobState::Pending, "{label} row untouched");
    }
    drop(held);

    // Retention enforcement: a dead row past its stamp deletes; the
    // same-shaped row on a retention-None queue lives forever.
    let retained = store
        .queue(
            "retained-1s",
            alkstore::QueueOpts {
                dead_letter_retention_s: Some(1),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let keeping = store
        .queue("keeper", alkstore::QueueOpts::default())
        .await
        .unwrap();
    let expires_1s = retained
        .enqueue(json!({"leg": "aged-out"}), EnqueueOpts::default())
        .await
        .unwrap();
    let held_dies = retained.claim_one("w1").await.unwrap().unwrap();
    assert!(held_dies.fail(None).await.unwrap());
    let expires_never = keeping
        .enqueue(json!({"leg": "kept"}), EnqueueOpts::default())
        .await
        .unwrap();
    let kept_dies = keeping.claim_one("w1").await.unwrap().unwrap();
    assert!(kept_dies.fail(None).await.unwrap());
    past_stamp_sleep(2_500).await;

    let deleted = retained.sweep_expired().await.unwrap();
    assert_eq!(
        deleted, 1,
        "the retention deletion reports through the sweep count"
    );
    assert!(
        retained.get_job(expires_1s).await.unwrap().is_none(),
        "the dead row past its retention stamp is gone"
    );
    assert_eq!(keeping.sweep_expired().await.unwrap(), 0);
    let kept = keeping.get_job(expires_never).await.unwrap().unwrap();
    assert_eq!(
        kept.state,
        JobState::Dead,
        "retention None = dead rows live forever"
    );

    factory.teardown().await.expect("factory teardown");
}

/// Shared clock helper for the suite's clock-adjacent rows: unix
/// seconds now (rows never assert tight timings — the tolerance is
/// the stamp's resolution).
fn unix_now() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}

type RunnerJoin = tokio::task::JoinHandle<Result<()>>;

/// Spawn a `run_schedules(stop)` runner driving off one shared store
/// handle (the runner-driving posture: the row holds the observing
/// Arc, the spawned task holds its clone; leadership is storage-level,
/// not handle-level). The token clone returned cancels only this
/// runner's stop.
fn spawn_runner(store: Arc<dyn Store>) -> (StopToken, RunnerJoin) {
    let stop = StopToken::new();
    let owned = store.clone();
    let token = stop.clone();
    let handle = tokio::spawn(async move { owned.run_schedules(token).await });
    (stop, handle)
}

/// Cancel and join a runner: `Ok` of the runner's own return shape
/// (clean stop `Ok(())` — ADR-019 §4 — or `LeadershipLost`), or a
/// panic if the join itself fails or the stop is not prompt.
async fn join_runner(stop: StopToken, runner: RunnerJoin) -> Result<()> {
    stop.cancel();
    match tokio::time::timeout(std::time::Duration::from_secs(10), runner).await {
        Ok(joined) => match joined {
            Ok(outcome) => outcome,
            Err(e) => panic!("the runner task join-failed: {e}"),
        },
        Err(_) => panic!("the stopped runner must terminate promptly, not hang"),
    }
}

/// Claim everything currently claimable from `queue` (one worker:
/// "suite-probe") and append the new fire ids to `fired`. Claiming is
/// the observation surface only — the fire rows are ordinary queue
/// work.
async fn collect_fires(store: &dyn Store, queue: &str, fired: &mut Vec<i64>) {
    let q = store
        .queue(queue, QueueOpts::default())
        .await
        .expect("queue handle for the fire probe");
    loop {
        let page = q
            .claim_batch("suite-probe", 50)
            .await
            .expect("the fire probe claims");
        let got = page.len() as i64;
        for handle in page {
            let id = handle.job().id;
            if !fired.contains(&id) {
                fired.push(id);
            }
        }
        if got < 50 {
            break;
        }
    }
}

/// Poll for fires until at least `min` distinct ids have landed, or
/// panic well-labelled after `deadline` (the tolerance-bounded wait:
/// the deadline is a safety bound, the assertions are on state
/// outcomes).
async fn await_fires(
    store: &dyn Store,
    queue: &str,
    fired: &mut Vec<i64>,
    min: usize,
    deadline: std::time::Duration,
    label: &str,
) {
    let end = std::time::Instant::now() + deadline;
    loop {
        collect_fires(store, queue, fired).await;
        if fired.len() >= min {
            return;
        }
        assert!(
            std::time::Instant::now() < end,
            "{label}: only {} boundary fires within the wait window",
            fired.len()
        );
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    }
}

/// The elapsed-boundary band's upper bound for a run started at
/// `started`: boundaries elapsed since start, plus slack for the
/// unix-second boundary arithmetic and the in-progress tick at
/// cancel (the engine tests' fire-window tolerance).
fn elapsed_boundary_band(started: &std::time::Instant) -> i64 {
    started.elapsed().as_secs() as i64 + 2
}

/// **Scheduler boundary fires** — the registered tick machinery, run:
/// a spawned runner fires due `@every 1s` boundaries into the named
/// queue as ordinary claimable jobs stamped with the `ScheduleOpts`
/// over the plain-queue derived defaults (`priority`/`max_attempts`
/// ride; visibility/backoff/retention take the engine defaults
/// 300/5/none), the fired rows are `get_job`-visible queue work with
/// the schedule's payload exactly, and a cancelled stop terminates the
/// runner promptly with the clean `Ok(())` shape. No double-fire per
/// boundary while one leader runs (the fire + boundary-advance commit
/// atomically — ADR-009 §4's one-firer discipline): the total fired
/// count stays inside the elapsed-boundary band above.
///
/// Contract stamp: ADR-009 §3 (boundary fires enqueue the schedule's
/// payload with `ScheduleOpts` into the named queue — ordinary queue
/// work); ADR-009 §4 (one firer per boundary, fire + advance atomic);
/// ADR-020 §3 (the stamp source: `ScheduleOpts` over the engine's
/// plain-queue derived defaults); ADR-019 §4 (the clean-stop `Ok(())`
/// return shape).
pub async fn scheduler_boundary_fires(factory: &dyn StoreFactory) {
    const LABEL: &str = "scheduler_boundary_fires";
    let store: Arc<dyn Store> = Arc::from(factory.open().await.expect("factory opens a store"));
    let started = std::time::Instant::now();

    store
        .schedule(
            "suite-tick",
            "@every 1s",
            "work",
            json!({"kind": "tick"}),
            ScheduleOpts {
                priority: 5,
                max_attempts: Some(9),
                expires: None,
            },
        )
        .await
        .expect("schedule registers");

    let (stop, runner) = spawn_runner(store.clone());
    let mut fired: Vec<i64> = Vec::new();
    await_fires(
        store.as_ref(),
        "work",
        &mut fired,
        2,
        std::time::Duration::from_secs(25),
        LABEL,
    )
    .await;

    // ADR-020 §3's stamp source: ScheduleOpts applied over the
    // plain-queue derived defaults (the fire resolves engine-side; the
    // queue was never opened before the fire).
    let q = store
        .queue("work", QueueOpts::default())
        .await
        .expect("queue handle");
    let job = q
        .get_job(fired[0])
        .await
        .expect("get_job reads the fire row")
        .expect("the fire row exists");
    assert_eq!(
        job.state,
        JobState::Processing,
        "the fire probe claimed this row"
    );
    assert_eq!(job.priority, 5, "ScheduleOpts priority rides the fire");
    assert_eq!(
        job.max_attempts, 9,
        "ScheduleOpts max_attempts overrides the default"
    );
    assert_eq!(
        job.visibility_timeout_s, 300,
        "visibility stamps the engine default"
    );
    assert_eq!(job.backoff_base_s, 5, "backoff stamps the engine default");
    assert_eq!(
        job.dead_letter_retention_s, None,
        "retention stamps the engine default"
    );
    assert_eq!(
        job.payload,
        json!({"kind": "tick"}).to_string().into_bytes(),
        "the schedule's payload is the fire's payload, exactly"
    );

    let outcome = join_runner(stop, runner).await;
    match outcome {
        Ok(()) => {}
        other => panic!("{LABEL}: a cancelled stop returns Ok(()), got {other:?}"),
    }

    // One firer per boundary: the total fired count stays inside the
    // elapsed-boundary band (a double-fire per boundary would blow
    // past it).
    collect_fires(store.as_ref(), "work", &mut fired).await;
    let band = elapsed_boundary_band(&started);
    assert!(
        fired.len() >= 2,
        "{LABEL}: two boundaries must have fired, got {}",
        fired.len()
    );
    assert!(
        (fired.len() as i64) <= band,
        "{LABEL}: no double-fire per boundary while one leader runs — \
         {} fires vs the elapsed-boundary band {band}",
        fired.len()
    );

    factory.teardown().await.expect("factory teardown");
}

/// **Scheduler bounded catch-up** — downtime replay is at-least-once
/// per elapsed boundary, bounded by the 64-cap: register, let several
/// boundaries elapse with no runner in sight (schedules never fire
/// without one — ADR-009 §1's opt-in posture), then start one: the
/// missed boundaries replay boundary-by-boundary — strictly more fires
/// land than a coalescing single-fire would leave — the count stays
/// bounded by the 64-cap and inside the elapsed-boundary band, and the
/// runner stops clean.
///
/// The beyond-cap skip-forward leg (>64 elapsed boundaries ≈ 65+ s of
/// downtime, plus a row-backdate the public surface does not express —
/// the schedule row's `next_fire_at` is engine-internal) pins
/// engine-side, where both engines' scheduler tests manipulate
/// `next_fire_at` directly
/// (`catch_up_replays_up_to_the_cap_then_skips_forward` on each); the
/// suite pins the ≤64 catch-up leg, the suite-observable one.
///
/// Contract stamp: ADR-009 §4 (at-least-once per elapsed boundary,
/// bounded catch-up — the fixed 64-per-tick contract constant, beyond
/// which missed boundaries skip forward); ADR-019 §4 (the clean-stop
/// `Ok(())` return shape).
pub async fn scheduler_bounded_catchup(factory: &dyn StoreFactory) {
    const LABEL: &str = "scheduler_bounded_catchup";
    let store: Arc<dyn Store> = Arc::from(factory.open().await.expect("factory opens a store"));
    let started = std::time::Instant::now();

    store
        .schedule(
            "suite-backlog",
            "@every 1s",
            "catchup",
            json!("w"),
            Default::default(),
        )
        .await
        .expect("schedule registers");

    // Downtime posture: no runner exists and nothing fires for it.
    let mut fired: Vec<i64> = Vec::new();
    collect_fires(store.as_ref(), "catchup", &mut fired).await;
    assert!(
        fired.is_empty(),
        "schedules never fire without a runner, got {} fires",
        fired.len()
    );
    // Several boundaries elapse (the first sits one interval past
    // registration; a 4 s runner-less pause leaves at least three
    // elapsed) — still nothing, no runner in sight.
    past_stamp_sleep(4_000).await;
    collect_fires(store.as_ref(), "catchup", &mut fired).await;
    assert!(
        fired.is_empty(),
        "elapsed boundaries pile up as schedule-row state, not fires, got {}",
        fired.len()
    );

    // Now run: the first tick replays the elapsed boundaries
    // boundary-by-boundary.
    let (stop, runner) = spawn_runner(store.clone());
    await_fires(
        store.as_ref(),
        "catchup",
        &mut fired,
        3,
        std::time::Duration::from_secs(25),
        LABEL,
    )
    .await;
    let outcome = join_runner(stop, runner).await;
    match outcome {
        Ok(()) => {}
        other => panic!("{LABEL}: a cancelled stop returns Ok(()), got {other:?}"),
    }
    collect_fires(store.as_ref(), "catchup", &mut fired).await;

    // The bounded-catch-up property: at-least-once per elapsed
    // boundary (strictly more than a coalescing fire), bounded by the
    // 64-cap (a multi-second downtime stays far below it).
    let count = fired.len() as i64;
    assert!(
        count >= 3,
        "{LABEL}: multi-boundary downtime must replay boundary-by-boundary \
         (at-least-once per elapsed boundary), got {count} fires"
    );
    assert!(
        count <= 64,
        "{LABEL}: the replay is bounded by the 64-cap, got {count}"
    );
    let band = elapsed_boundary_band(&started);
    assert!(
        count <= band,
        "{LABEL}: fires stay inside the elapsed-boundary band ({count} vs {band})"
    );

    factory.teardown().await.expect("factory teardown");
}

/// **Scheduler leadership discipline** — two concurrent runners on one
/// store: exactly one leader, the other surfaces `LeadershipLost`
/// rather than silently co-ticking (the acquire-time refusal is the
/// same return-shape outcome as a mid-run loss), the fired count stays
/// inside the elapsed-boundary band with both runners alive (fires are
/// not duplicated), and the winning leader's cancelled stop returns
/// the clean `Ok(())` shape (ADR-019 §4; the loser's return is the
/// matchable `Err(LeadershipLost)`, never an ordinary error or a
/// silent `Ok`).
///
/// Contract stamp: ADR-009 §1 (one leader at a time; leadership loss
/// ends the runner); ADR-009 §6 (`LeadershipLost` — the matchable
/// return-shape variant of the runner's stop); ADR-019 §4 (the
/// clean-stop `Ok(())` return shape under a cancelled token).
pub async fn scheduler_leadership_discipline(factory: &dyn StoreFactory) {
    const LABEL: &str = "scheduler_leadership_discipline";
    let store: Arc<dyn Store> = Arc::from(factory.open().await.expect("factory opens a store"));
    let started = std::time::Instant::now();

    store
        .schedule(
            "suite-lead",
            "@every 1s",
            "single",
            json!(1),
            Default::default(),
        )
        .await
        .expect("schedule registers");

    // Two concurrent runners on the same store: the leadership lock is
    // storage-level, so exactly one is granted and drives the tick.
    let (stop_a, runner_a) = spawn_runner(store.clone());
    let (stop_b, runner_b) = spawn_runner(store.clone());
    let mut fired: Vec<i64> = Vec::new();
    await_fires(
        store.as_ref(),
        "single",
        &mut fired,
        1,
        std::time::Duration::from_secs(25),
        LABEL,
    )
    .await;
    // A few more boundaries with both runners alive.
    tokio::time::sleep(std::time::Duration::from_secs(2)).await;

    let outcome_a = join_runner(stop_a, runner_a).await;
    let outcome_b = join_runner(stop_b, runner_b).await;

    let clean = |o: &Result<()>| matches!(o, Ok(()));
    let lost = |o: &Result<()>| matches!(o, Err(Error::LeadershipLost));
    assert!(
        (clean(&outcome_a) && lost(&outcome_b)) || (lost(&outcome_a) && clean(&outcome_b)),
        "{LABEL}: exactly one leader — one clean stop and one \
         `LeadershipLost`, got A {outcome_a:?} and B {outcome_b:?}"
    );

    // One firer per boundary while both runners were alive: the fired
    // count stays inside the elapsed-boundary band (duplicated fires
    // would blow past it).
    collect_fires(store.as_ref(), "single", &mut fired).await;
    let count = fired.len() as i64;
    let band = elapsed_boundary_band(&started);
    assert!(
        count >= 1 && count <= band,
        "{LABEL}: fires not duplicated under two runners — {count} fires \
         vs the elapsed-boundary band {band}"
    );

    factory.teardown().await.expect("factory teardown");
}

/// The delivery recorder for the tx-seam commit-atomicity rows: a
/// real `Delivery` — every claimed job is captured (its `Job` snapshot,
/// stamps included) and acked via the `Ok` disposition.
struct RecordingDelivery {
    delivered: Vec<Job>,
}

impl Delivery for RecordingDelivery {
    fn deliver<'a>(&'a mut self, job: Box<dyn JobHandle>) -> BoxedFuture<'a, Result<()>> {
        self.delivered.push(job.job().clone());
        Box::pin(std::future::ready(Ok(())))
    }
}

/// The panic probe's typed diverger: a `#[cold]` call typed `Error` so
/// the closure's tail stays a normal expression (a bare `Err(panic!())`
/// tail is unreachable-call-shaped and trips `unreachable_code`).
#[cold]
fn mid_flight_panic(message: &str) -> Error {
    panic!("{message}")
}

/// **`outbox_enqueue_tx` commit-atomicity** — the outbox send sits in
/// the caller's transaction: rollback drops the backing-queue job row
/// together with the business write (no ghost job — `run_once`
/// afterwards claims nothing), and commit makes the job claimable by
/// `run_once` exactly when the business write commits (a real delivery
/// driven through the `Delivery` closure, the job consumed after its
/// ack). The stamped opts are the outbox's derived 60/5/5 set,
/// delivered on the job handle and identical on both engines — the
/// derived backing queue is engine-internal (reserved-prefixed,
/// unreachable through the contract surface: the exemplar row pins
/// `get_job_tx`'s reserved rejection), so the delivery handle's
/// `job()` is the contract-visible inspection of the stamps.
///
/// Contract stamp: ADR-014 §1 (`outbox_enqueue_tx` — commit makes the
/// backing-queue job claimable exactly when the business write
/// commits; rollback drops both; the derived-name-only path);
/// ADR-010 §3a (the stamp point — `EnqueueOpts` resolved over the
/// outbox's derived 60/5/5 set, visible on the delivered job);
/// ADR-021 §4 (drop = rollback — the no-ghosts property the rollback
/// leg pins); ADR-007 (the caller-held seam the pair commits on).
pub async fn outbox_enqueue_tx_commit_atomicity(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    // Rollback leg: the backing-queue job row drops together with the
    // business write — nothing for `run_once` to claim afterwards.
    let mut tx = store.begin_tx().await.unwrap();
    let business = tx
        .enqueue_tx("ledger", EnqueueOpts::default(), json!({"leg": "rollback"}))
        .await
        .unwrap();
    tx.outbox_enqueue_tx("mail", EnqueueOpts::default(), json!({"send": "ghost"}))
        .await
        .unwrap();
    assert!(
        tx.get_job_tx("ledger", business).await.unwrap().is_some(),
        "read-your-own-writes makes the rollback's drop meaningful"
    );
    drop(tx);

    let mut tx = store.begin_tx().await.unwrap();
    let business_gone = tx.get_job_tx("ledger", business).await.unwrap();
    tx.commit().await.unwrap();
    assert_eq!(
        business_gone, None,
        "the rollback dropped the business write"
    );

    let mut mail = store.outbox("mail").await.unwrap();
    let claimed = mail
        .run_once(
            "suite-probe",
            &mut RecordingDelivery {
                delivered: Vec::new(),
            },
        )
        .await
        .unwrap();
    assert!(
        !claimed,
        "rollback leaves no ghost job for run_once to claim"
    );

    // Commit leg: commit lands both halves together — the business
    // write becomes `get_job`-visible and the backing-queue job becomes
    // claimable by `run_once`, which delivers through the real
    // `Delivery` closure.
    let mut tx = store.begin_tx().await.unwrap();
    let business = tx
        .enqueue_tx("ledger", EnqueueOpts::default(), json!({"leg": "commit"}))
        .await
        .unwrap();
    let queued = tx
        .outbox_enqueue_tx("mail", EnqueueOpts::default(), json!({"send": "letter"}))
        .await
        .unwrap();
    tx.commit().await.unwrap();

    let ledger = store.queue("ledger", QueueOpts::default()).await.unwrap();
    let business_row = ledger
        .get_job(business)
        .await
        .unwrap()
        .expect("the business write committed with the enqueue");
    assert_eq!(business_row.queue, "ledger");

    let mut delivery = RecordingDelivery {
        delivered: Vec::new(),
    };
    let ran = mail
        .run_once("suite-probe", &mut delivery)
        .await
        .expect("run_once claims the committed job");
    assert!(
        ran,
        "commit makes the backing-queue job claimable by run_once"
    );
    assert_eq!(
        delivery.delivered.len(),
        1,
        "the delivery closure ran exactly once"
    );
    let job = &delivery.delivered[0];
    assert_eq!(job.id, queued, "the delivered job is the tx-enqueued one");
    assert_eq!(
        job.queue,
        format!("{RESERVED_PREFIX}outbox:mail"),
        "the enqueue landed in the engine-derived backing queue"
    );
    assert_eq!(
        job.payload,
        encode_payload(&json!({"send": "letter"})).unwrap(),
        "the payload is exact"
    );
    assert_eq!(
        (
            job.visibility_timeout_s,
            job.max_attempts,
            job.backoff_base_s,
            job.dead_letter_retention_s
        ),
        (60, 5, 5, None),
        "the outbox's derived 60/5/5 stamp set rides the delivered job"
    );
    let again = mail
        .run_once("suite-probe", &mut delivery)
        .await
        .expect("run_once after the delivery");
    assert!(!again, "the delivered-then-acked job is consumed");

    factory.teardown().await.expect("factory teardown");
}

/// **`publish_with_key_tx` commit-atomicity** — the keyed publish
/// rides the tx seam: rollback drops the keyed event row with the
/// business write (no ghost event), and commit makes it visible to
/// `read_since` and to a subscriber attach, the key round-tripping on
/// the committed event. (The empty-`Some`-key `InvalidName` leg pins
/// in the exemplar row — not duplicated here.)
///
/// Contract stamp: ADR-015 §2 (`publish_with_key_tx` — the tx seam
/// carries the key; commit-atomic with the business write); ADR-015
/// §4 (the committed event's read surfaces: offset ASC, the key as
/// carried metadata); ADR-021 §4 (drop = rollback); ADR-007 (the
/// caller-held seam).
pub async fn publish_with_key_tx_commit_atomicity(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");

    // Rollback leg: the keyed event row drops together with the
    // business write — no ghost event; the key still round-tripped
    // inside the tx (read-your-own-writes makes the drop meaningful).
    let mut tx = store.begin_tx().await.unwrap();
    tx.publish_tx("ledger-events", json!({"leg": "rollback", "n": 1}))
        .await
        .unwrap();
    let keyed = tx
        .publish_with_key_tx(
            "events",
            Some("k".to_string()),
            json!({"leg": "rollback", "n": 1}),
        )
        .await
        .unwrap();
    let seen = tx.read_since_tx("events", 0, 100).await.unwrap();
    assert_eq!(seen.len(), 1, "the tx sees its own keyed publish pre-drop");
    assert_eq!(seen[0].offset, keyed);
    assert_eq!(
        seen[0].key.as_deref(),
        Some("k"),
        "the key round-trips inside the tx"
    );
    drop(tx);

    let ledger_gone = store
        .stream("ledger-events")
        .await
        .unwrap()
        .read_since(0, 100)
        .await
        .unwrap();
    let events_gone = store
        .stream("events")
        .await
        .unwrap()
        .read_since(0, 100)
        .await
        .unwrap();
    assert!(
        ledger_gone.is_empty(),
        "the rollback dropped the business write"
    );
    assert!(
        events_gone.is_empty(),
        "no ghost keyed event after the rollback"
    );

    // Commit leg: the keyed event lands with the business write —
    // visible to fresh reads with the key round-tripping, and to a
    // subscriber attaching after the commit.
    let mut tx = store.begin_tx().await.unwrap();
    let business = tx
        .publish_tx("ledger-events", json!({"leg": "commit", "n": 2}))
        .await
        .unwrap();
    let keyed = tx
        .publish_with_key_tx(
            "streamed",
            Some("k2".to_string()),
            json!({"leg": "commit", "n": 2}),
        )
        .await
        .unwrap();
    tx.commit().await.unwrap();

    let ledger = store.stream("ledger-events").await.unwrap();
    let keyed_stream = store.stream("streamed").await.unwrap();
    let ledger_page = ledger.read_since(0, 100).await.unwrap();
    assert_eq!(ledger_page.len(), 1, "the business write committed");
    assert_eq!(ledger_page[0].offset, business);
    let keyed_page = keyed_stream.read_since(0, 100).await.unwrap();
    assert_eq!(keyed_page.len(), 1, "the keyed event committed with it");
    assert_eq!(keyed_page[0].offset, keyed);
    assert_eq!(
        keyed_page[0].key.as_deref(),
        Some("k2"),
        "the key round-trips on the committed event"
    );
    assert_eq!(
        keyed_page[0].payload,
        encode_payload(&json!({"leg": "commit", "n": 2})).unwrap(),
        "the payload is exact"
    );

    let mut rx = keyed_stream.subscribe("attach").await.unwrap();
    let yielded = rx
        .recv()
        .await
        .expect("the subscriber attaches and yields")
        .expect("the attach read yields the committed event");
    assert_eq!(
        yielded.offset, keyed,
        "the subscriber attach sees the committed keyed event"
    );
    assert_eq!(
        yielded.key.as_deref(),
        Some("k2"),
        "the key round-trips on the attach"
    );

    factory.teardown().await.expect("factory teardown");
}

/// **Cross-engine stream ordering equivalence** — the ordering-guarantee
/// row: a publish sequence with keyed/unkeyed interleavings yields
/// global FIFO per stream — `offset ASC`, strictly increasing per
/// stream — and the *same publish order reads back in the same order*
/// on every read form: `read_since` (whole, cursor-paginated, and
/// mid-stream), `read_from_consumer`, and a subscriber's attach drain
/// alike. `key` round-trips exactly (`None` stays `None`, `Some` stays
/// `Some`, on every read form — the explicit-`None` keyed publish form
/// included, its event indistinguishable from a plain publish's);
/// `stream` carries the stream name; `created_at` is
/// unix-seconds-at-publish (informational — tolerance-bounded
/// proximity to now, never an ordering assertion).
///
/// Cross-engine equivalence is the row's disposition: the *same* row
/// runs against each engine's factory, and the per-stream relative
/// order it pins is safe under either engine's global counter scheme
/// (pg bigserial, SQLite AUTOINCREMENT) — absolute offset values are
/// explicitly *not* cross-pinned, cross-stream offsets never compared.
///
/// Cross-reference: the single-event key/payload byte-exactness legs
/// pin in `payload_round_trip_stores_exact_encoding` (ADR-020 §4), the
/// keyed tx-seam legs in `publish_with_key_tx_commit_atomicity` — this
/// row owns the *sequence/ordering* property.
///
/// Contract stamp: ADR-015 §4 (global FIFO by offset per stream,
/// offset-ASC direct and subscriber reads); ADR-015 §3 (the
/// `StreamEvent` shape — `offset` immutable, `stream` the name,
/// `created_at` informational); ADR-015 §1 (the key is carried
/// metadata round-tripping exactly).
pub async fn stream_ordering_equivalence(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");
    let stream = store.stream("events").await.expect("stream handle");

    let mut seq: Vec<(i64, Option<String>)> = Vec::new();
    seq.push((stream.publish(json!({"n": 0})).await.unwrap(), None));
    seq.push((
        stream
            .publish_with_key(Some("a".to_string()), json!({"n": 1}))
            .await
            .unwrap(),
        Some("a".to_string()),
    ));
    seq.push((
        stream
            .publish_with_key(None, json!({"n": 2}))
            .await
            .unwrap(),
        None,
    ));
    seq.push((stream.publish(json!({"n": 3})).await.unwrap(), None));
    seq.push((
        stream
            .publish_with_key(Some("a".to_string()), json!({"n": 4}))
            .await
            .unwrap(),
        Some("a".to_string()),
    ));
    seq.push((stream.publish(json!({"n": 5})).await.unwrap(), None));

    let offsets: Vec<i64> = seq.iter().map(|(o, _)| *o).collect();
    for pair in seq.windows(2) {
        assert!(
            pair[1].0 > pair[0].0,
            "offsets strictly increase per stream (per-stream relative order)"
        );
    }

    let whole = stream.read_since(0, 10).await.unwrap();
    assert_eq!(
        whole.iter().map(|e| e.offset).collect::<Vec<_>>(),
        offsets,
        "read_since yields the publish sequence offset ASC"
    );
    assert_eq!(
        whole.iter().map(|e| e.key.clone()).collect::<Vec<_>>(),
        seq.iter().map(|(_, k)| k.clone()).collect::<Vec<_>>(),
        "key round-trips exactly on read_since"
    );
    assert!(
        whole.iter().all(|e| e.stream == "events"),
        "`stream` carries the stream name"
    );
    let now = unix_now();
    for e in &whole {
        assert!(
            e.created_at.abs_diff(now) <= 10,
            "`created_at` is unix-seconds-at-publish, tolerance-bounded \
             near now (informational, never an ordering assertion)"
        );
    }

    // Cursor-paginated + mid-stream reads reproduce the same order.
    let mut paged: Vec<i64> = Vec::new();
    let mut cursor = 0;
    loop {
        let page = stream.read_since(cursor, 2).await.unwrap();
        if page.is_empty() {
            break;
        }
        paged.extend(page.iter().map(|e| e.offset));
        cursor = page[page.len() - 1].offset;
        if page.len() < 2 {
            break;
        }
    }
    assert_eq!(
        paged, offsets,
        "cursor-paginated reads reproduce the publish order"
    );
    let mid = stream.read_since(offsets[2], 10).await.unwrap();
    assert_eq!(
        mid.iter().map(|e| e.offset).collect::<Vec<_>>(),
        offsets[3..],
        "a mid-stream read continues in order from its cursor"
    );

    // The consumer-cursor form: an absent consumer reads from 0 (the
    // whole sequence), a mid checkpoint resumes after it.
    let fresh = stream.read_from_consumer("fresh", 10).await.unwrap();
    assert_eq!(
        fresh.iter().map(|e| e.offset).collect::<Vec<_>>(),
        offsets,
        "read_from_consumer yields the publish sequence offset ASC"
    );
    stream.save_offset("mid", offsets[2]).await.unwrap();
    let from_mid = stream.read_from_consumer("mid", 10).await.unwrap();
    assert_eq!(
        from_mid.iter().map(|e| e.offset).collect::<Vec<_>>(),
        offsets[3..],
        "read_from_consumer resumes after the stored checkpoint"
    );

    // The subscriber's attach drain: same order, keys round-trip.
    let mut rx = stream.subscribe("sub").await.unwrap();
    let first = rx
        .recv()
        .await
        .expect("the attach read yields")
        .expect("the attach read yields an event");
    let mut drained = vec![first];
    while let Some(next) = rx.try_recv().unwrap() {
        drained.push(next);
    }
    assert_eq!(
        drained.iter().map(|e| e.offset).collect::<Vec<_>>(),
        offsets,
        "the subscriber's attach drain yields the publish sequence offset ASC"
    );
    assert_eq!(
        drained.iter().map(|e| e.key.clone()).collect::<Vec<_>>(),
        seq.iter().map(|(_, k)| k.clone()).collect::<Vec<_>>(),
        "key round-trips exactly on the attach drain"
    );

    factory.teardown().await.expect("factory teardown");
}

/// **`trim_to` semantics** — the full ADR-015 §5 row: exact-boundary
/// trim (`offset <= horizon` — the horizon's own row deletes,
/// horizon+1 survives; a repeated trim at the same horizon deletes
/// nothing), surviving rows keep their offsets (gaps legal, never
/// renumbered), a read from a trimmed-away region resumes at the trim
/// horizon's first remaining row, a saved offset below the horizon
/// stays a valid position marker (`get_offset` returns it;
/// `read_from_consumer` and a fresh `subscribe`'s attach both resume
/// at the horizon, not from a renumbered past), trim emits no
/// dedicated wake and no notify (a pre-attached listener idles across
/// the trim — tolerance-bounded absence; on SQLite the `data_version`
/// watcher may still fire on the trim commit, a spurious hint that is
/// contract-legal, ADR-006 §1 — it delivers nothing), and a subscriber
/// with a saved checkpoint never loses its place across a trim (its
/// next read continues forward from the trim, not from a renumbered
/// past).
///
/// Cross-reference: the negative-horizon/immutability legs pin in
/// `extent_clamp_semantics` (ADR-023 §2's boundary-totality rule) —
/// this row owns the exact-boundary and resume legs.
///
/// Contract stamp: ADR-015 §5 (`trim_to` — exact boundary, never
/// renumbered, resume at the horizon, saved offsets stay valid, trim
/// wakes nothing, consumers never lose their place); ADR-019 §6 (the
/// saved checkpoint composes — reads anchor at the stored position,
/// replay is a read concern).
pub async fn trim_to_semantics(factory: &dyn StoreFactory) {
    let store = factory.open().await.expect("factory opens a store");
    let stream = store.stream("events").await.expect("stream handle");

    let mut offsets = Vec::new();
    for i in 0..5 {
        offsets.push(stream.publish(json!({"n": i})).await.unwrap());
    }

    // A pre-attached listener drains to the tail and checkpoints
    // explicitly; a direct-form consumer holds a checkpoint below the
    // (soon) horizon. Both are in place before any trim fires.
    let mut rx = stream.subscribe("listener").await.unwrap();
    let first = rx
        .recv()
        .await
        .expect("the attach read yields")
        .expect("the attach read yields an event");
    assert_eq!(
        first.offset, offsets[0],
        "the attach drain starts offset ASC"
    );
    let mut last = first;
    while let Some(next) = rx.try_recv().unwrap() {
        assert!(next.offset > last.offset, "the drain stays offset ASC");
        last = next;
    }
    assert_eq!(last.offset, offsets[4], "the attach drain reached the tail");
    rx.save_offset().unwrap();
    stream.save_offset("behind", offsets[1]).await.unwrap();

    // Exact boundary: the horizon's own row deletes, horizon+1
    // survives.
    let deleted = stream.trim_to(offsets[1]).await.unwrap();
    assert_eq!(
        deleted, 2,
        "trim deletes offset <= horizon exactly (the horizon's own row gone)"
    );
    let survivors = stream.read_since(0, 10).await.unwrap();
    assert_eq!(
        survivors.iter().map(|e| e.offset).collect::<Vec<_>>(),
        offsets[2..],
        "the horizon's first remaining row survives at its own offset"
    );
    assert_eq!(
        stream.trim_to(offsets[1]).await.unwrap(),
        0,
        "a repeated trim at the same horizon deletes nothing"
    );

    // Resume legs: a read from a trimmed-away region resumes at the
    // horizon's first remaining row (never an error, never a
    // renumbered row).
    let resumed = stream.read_since(offsets[0], 10).await.unwrap();
    assert_eq!(
        resumed.iter().map(|e| e.offset).collect::<Vec<_>>(),
        offsets[2..],
        "a read from a trimmed-away region resumes at the horizon's \
         first remaining row"
    );

    // The below-horizon checkpoint stays a valid position marker:
    // `get_offset` returns it, `read_from_consumer` (and a fresh
    // `subscribe`'s attach with the same consumer) resumes at the
    // horizon.
    assert_eq!(
        stream.get_offset("behind").await.unwrap(),
        offsets[1],
        "a saved offset below the horizon stays returned by get_offset"
    );
    let from = stream.read_from_consumer("behind", 10).await.unwrap();
    assert_eq!(
        from.iter().map(|e| e.offset).collect::<Vec<_>>(),
        offsets[2..],
        "read_from_consumer resumes at the horizon's first remaining row"
    );
    let mut resumed_rx = stream.subscribe("behind").await.unwrap();
    let first_remaining = resumed_rx
        .recv()
        .await
        .expect("the re-attach read yields")
        .expect("the re-attach read yields an event");
    assert_eq!(
        first_remaining.offset, offsets[2],
        "a re-attach with a below-horizon checkpoint continues at the \
         horizon's first remaining row, never a renumbered past"
    );
    let mut attach = vec![first_remaining.offset];
    while let Some(next) = resumed_rx.try_recv().unwrap() {
        attach.push(next.offset);
    }
    assert_eq!(
        attach,
        offsets[2..],
        "the re-attach drains the surviving sequence in order"
    );
    drop(resumed_rx);

    // The subscriber's place never lost: a fresh attach of the
    // pre-trim listener (checkpoint above the horizon) drains nothing
    // forward.
    let mut listener_rx = stream.subscribe("listener").await.unwrap();
    assert_eq!(
        listener_rx.try_recv().unwrap(),
        None,
        "a checkpoint above the horizon keeps its place across the trim"
    );
    drop(listener_rx);

    // Trim emits no dedicated wake and no notify: the pre-attached
    // listener idles across the trims (tolerance-bounded absence —
    // on SQLite the spurious watcher hint is contract-legal and
    // delivers nothing).
    let idle_deadline = tokio::time::Instant::now() + std::time::Duration::from_millis(400);
    while tokio::time::Instant::now() < idle_deadline {
        assert_eq!(
            rx.try_recv().unwrap(),
            None,
            "trim emits no dedicated wake — the pre-attached listener \
             idles across the trim"
        );
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }

    factory.teardown().await.expect("factory teardown");
}

/// **`with_tx` panicking closure rolls back** — a closure that panics
/// mid-flight ⇒ rollback with no residue over the uniform no-ghosts
/// list (job/event/notify/offset — ADR-021 §4's drop = rollback
/// through the panic path), and the store remains usable afterward
/// (the SQLite writer slot replenished — the wave-3 stranding fix's
/// property, now pinned at contract level; the pg pooled object
/// re-pooled). The panic crosses an await inside `with_tx`'s own
/// future — a `catch_unwind` around an async closure cannot see the
/// panic point — so the row drives a spawned task and asserts the
/// panic surfaced through the `JoinError` (`is_panic`), then the
/// post-panic state: the no-ghost reads converge (the rollback may
/// land asynchronously engine-side — the loop is the bounded wait,
/// state-outcome assertions only), the pre-panic listener hears
/// nothing on the notified channel before any post-panic commit exists
/// (the notify residue's cross-engine pin: SQLite's wake overtriggers
/// on any commit past this window, so the window precedes the
/// happy-path write; the fully-cross-engine notify-ghost pin rides
/// the engines' own rollback-ghost tests), and a fresh `with_tx`
/// commits through the same seam.
///
/// Contract stamp: ADR-007 (`with_tx` — the closure-scoped transaction
/// and its panic disposition: `Ok` ⇒ commit, `Err`/drop/panic ⇒
/// rollback); ADR-021 §4 (drop = rollback through the panic path — no
/// residue, the store usable afterward; N-5's probe against real
/// engines).
pub async fn with_tx_panicking_closure_rolls_back(factory: &dyn StoreFactory) {
    let store: Arc<dyn Store> = Arc::from(factory.open().await.expect("factory opens a store"));

    // The listener attaches before the panicking tx fires: a ghost
    // notify delivering at commit would reach it.
    let mut listener = store.listen("panic-chan").await.expect("listen attaches");

    let ghost_ids: Arc<std::sync::Mutex<Vec<i64>>> = Arc::new(std::sync::Mutex::new(Vec::new()));
    let captured = ghost_ids.clone();
    let joined = tokio::spawn({
        let store = store.clone();
        async move {
            store
                .with_tx(Box::new(move |tx| {
                    Box::pin(async move {
                        let id = tx
                            .enqueue_tx("panicked", EnqueueOpts::default(), json!({"panic": true}))
                            .await
                            .expect("the in-flight enqueue lands inside the tx");
                        captured.lock().expect("capture mutex").push(id);
                        tx.publish_tx("panicked-events", json!({"panic": true}))
                            .await
                            .expect("the in-flight publish lands inside the tx");
                        tx.notify_tx("panic-chan", json!({"panic": true}))
                            .await
                            .expect("the in-flight notify lands inside the tx");
                        tx.save_offset_tx("panicked-events", "ghostc", 7)
                            .await
                            .expect("the in-flight offset save lands inside the tx");
                        Err::<(), Error>(mid_flight_panic(
                            "mid-flight panic across with_tx's closure",
                        ))
                    }) as BoxedFuture<'_, Result<()>>
                }))
                .await
                .unwrap();
        }
    })
    .await;

    assert!(
        joined.is_err(),
        "the panic must surface through the spawned task's join, not be swallowed"
    );
    let join_error = joined.unwrap_err();
    assert!(
        join_error.is_panic(),
        "the join must carry a panic, not a cancellation: {join_error}"
    );

    // No residue: the no-ghost reads converge — the rollback may land
    // asynchronously (pg's detached rollback; SQLite's runs in the
    // unwind), so the loop is the bounded wait for a *state* outcome;
    // each iteration's begin_tx is also the seam-still-grants probe.
    let ghost_id = *ghost_ids
        .lock()
        .expect("capture mutex")
        .first()
        .expect("the enqueue's id was captured before the panic");
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15);
    loop {
        let granted = tokio::time::timeout(std::time::Duration::from_secs(5), store.begin_tx())
            .await
            .expect("begin_tx must not hang: the seam must replenish after the panic");
        let mut tx = granted.expect("begin_tx succeeds post-panic");
        let ghost_job = tx.get_job_tx("panicked", ghost_id).await.unwrap();
        let ghost_events = tx.read_since_tx("panicked-events", 0, 100).await.unwrap();
        let ghost_checkpoint = tx.get_offset_tx("panicked-events", "ghostc").await.unwrap();
        tx.commit().await.unwrap();
        if ghost_job.is_none() && ghost_events.is_empty() && ghost_checkpoint == 0 {
            break;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "the panic's rollback never converged: job {ghost_job:?}, events {}, \
             checkpoint {ghost_checkpoint}",
            ghost_events.len()
        );
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }

    // No ghost notify: the pre-panic listener hears nothing on the
    // notified channel — this window precedes any post-panic commit,
    // so an engine may not have a wake source here (SQLite's watcher
    // overtriggers on any commit; a past-window wake is not this leg's
    // business). A delivery-shaped wake is a genuine defect.
    let budget = std::time::Instant::now() + std::time::Duration::from_secs(1);
    let heard = loop {
        let wake = listener
            .recv_timeout(std::time::Duration::from_millis(150))
            .await
            .expect("the silent wait is idle, never an error");
        match wake {
            None => break None,
            Some(w) if w.channel == "panic-chan" => break Some(w),
            Some(_) => {}
        }
        assert!(
            std::time::Instant::now() < budget,
            "the silence window overflowed with unrelated wakes"
        );
    };
    assert!(
        heard.is_none(),
        "no ghost notify reached the listener through the panic path: {heard:?}"
    );

    // The store remains usable (no writer-slot stranding, no
    // pooled-object leak): the ghost never became claimable work, and
    // a fresh closure-scoped tx commits through the same seam.
    let panicked = store.queue("panicked", QueueOpts::default()).await.unwrap();
    let drained = panicked.claim_batch("probe", 10).await.unwrap();
    assert!(drained.is_empty(), "no ghost work became claimable");
    store
        .with_tx(Box::new(|tx| {
            Box::pin(async move {
                tx.enqueue_tx("after", EnqueueOpts::default(), json!({"after": true}))
                    .await?;
                Ok(())
            }) as BoxedFuture<'_, Result<()>>
        }))
        .await
        .expect("the store survives the panic: a fresh with_tx commits");
    let after = store.queue("after", QueueOpts::default()).await.unwrap();
    let claimed = after.claim_batch("probe", 10).await.unwrap();
    assert_eq!(
        claimed.len(),
        1,
        "the post-panic commit's business write is live work"
    );

    factory.teardown().await.expect("factory teardown");
}