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 assertions and no two-task race
//! windows — orchestration that would need one (runner spawning,
//! cross-connection wakes) stays engine-test-side; rows keep to
//! single-store sequential drives and the factory's isolation
//! guarantee. 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 serde_json::json;

use alkstore::{
    BoxedFuture, Delivery, EnqueueOpts, Error, JobHandle, JobState, RESERVED_PREFIX, Result,
    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`](alkstore::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");
}

/// **`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");
}

/// **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.)
///
/// Contract stamp: ADR-021 §1 (tx-side read methods,
/// read-your-own-writes); ADR-007 (the commit-atomic seam the reads
/// ride).
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");
    tx.commit().await.unwrap();

    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.
///
/// The `run_at`-alone-literal and neither-field-resolves-to-now legs
/// are wall-clock-adjacent; they pin with wave 5's equivalence row
/// (this row's precedence/resolution legs are the engine-agnostic
/// claim).
///
/// Contract stamp: ADR-020 §1 (delay wins over `run_at`); §2
/// (relative `expires` → absolute row value); §3/§3a (the derived
/// default stamps — plain queue 300/3/5/none, per-job override);
/// 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"
    );

    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).
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"
    );

    // 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"
    );
}

/// **`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.
fn past_stamp_sleep(millis: u64) {
    std::thread::sleep(std::time::Duration::from_millis(millis));
}

/// **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).
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"
    );

    // 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);
    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);
    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);

    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);
    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);
    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);

    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);

    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)
}