//! The queue mechanism's acceptance tests (`pg-engine-queues`): the
//! enqueue→claim→ack lifecycle with stamps visible via `get_job`
//! matching the resolution rules, the extent guard (ADR-023 §2), the
//! exactly-once handout under concurrent claims (two workers, one row;
//! 6×40 drain, no double handouts), the backoff curve's range pin
//! (statistical, tolerance-bounded) with the 1-hour cap and the
//! `Some(d)` override, the uniform validity predicate refusing lapsed
//! ops with `Ok(false)` and a reclaim consuming an attempt, the
//! dead-letter defaults and dead-row `get_job` visibility, the
//! pre-claim exhausted-reclaimable sweep, claim ordering, cancel +
//! `ack_batch`'s per-id predicate without a worker filter, `sweep_
//! expired`'s both-states move + retention (incl. forever-none), the
//! wake-driven claim loop pinned end-to-end (enqueue → notify → wake →
//! claim, cross-connection; the LISTEN-driven posture with the re-poll
//! safety net — the contract claim ops stay pull-shaped), queue-name
//! scoping, the validation posture (empty/reserved queue names, empty
//! `worker_id`), and the closed-store fail-closed shape on all queue
//! ops (ADR-010; ADR-019 §1–§3; ADR-020; ADR-023 §2).
//!
//! Harness convention (the schema/open tasks'): connection settings
//! ride the environment (`ALKSTORE_PG_HOST/PORT/USER/PASSWORD/DB`),
//! never hardcoded; tests without a reachable server skip cleanly so
//! the workspace gates stay green server-less. Isolation is a fresh
//! unique schema per test.

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::time::Duration;

use alkstore::{EnqueueOpts, JobHandle, JobState, QueueOpts, Store};
use serde_json::{Value, json};

use crate::queue::{EXHAUSTED_ERROR, EXPIRED_ERROR, FAILED_ERROR, backoff_delay_s};
use crate::store::{PgStore, open_store};

const ENV_HOST: &str = "ALKSTORE_PG_HOST";
const ENV_PORT: &str = "ALKSTORE_PG_PORT";
const ENV_USER: &str = "ALKSTORE_PG_USER";
const ENV_PASSWORD: &str = "ALKSTORE_PG_PASSWORD";
const ENV_DB: &str = "ALKSTORE_PG_DB";

fn harness_dsn() -> Option<String> {
    let host = std::env::var(ENV_HOST).ok()?;
    let port: u16 = std::env::var(ENV_PORT).ok()?.parse().ok()?;
    let user = std::env::var(ENV_USER).ok()?;
    let password = std::env::var(ENV_PASSWORD).ok()?;
    let db = std::env::var(ENV_DB).unwrap_or_else(|_| "postgres".to_string());
    Some(format!(
        "host={host} port={port} user={user} password={password} dbname={db}"
    ))
}

fn instance_namer(tag: &str) -> impl Fn() -> String + use<'_> {
    let counter = AtomicU64::new(0);
    move || {
        format!(
            "{tag}_{}_{}_{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .as_nanos(),
            counter.fetch_add(1, Ordering::SeqCst),
        )
    }
}

/// A raw admin client for cross-connection probes (NOTIFY senders,
/// schema teardown, row backdating) — separate from the store's pool.
async fn harness_client() -> Option<tokio_postgres::Client> {
    let (client, connection) = tokio_postgres::connect(&harness_dsn()?, tokio_postgres::NoTls)
        .await
        .ok()?;
    tokio::spawn(async move {
        let _ = connection.await;
    });
    Some(client)
}

async fn drop_schema(client: &tokio_postgres::Client, schema: &str) {
    if schema == crate::schema::DEFAULT_SCHEMA {
        return;
    }
    let sql = format!(
        "DROP SCHEMA IF EXISTS {} CASCADE",
        crate::schema::quote_identifier(schema)
    );
    let _ = client.batch_execute(&sql).await;
}

fn test_opts(schema: &str) -> crate::opts::PgOpts {
    crate::opts::PgOpts {
        schema: schema.to_string(),
        ..crate::opts::PgOpts::default()
    }
}

/// Open a store on the harness server with a fresh unique schema —
/// `None` = no harness server (the skip posture). The returned store
/// closes on `Fixture` drop (the schema teardown included).
struct Fixture {
    store: PgStore,
    admin: tokio_postgres::Client,
    schema: String,
}

impl Fixture {
    async fn open(namer: &impl Fn() -> String) -> Option<Fixture> {
        let dsn = harness_dsn()?;
        let schema = namer();
        let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
        let admin = harness_client().await?;
        Some(Fixture {
            store,
            admin,
            schema,
        })
    }
}

impl Drop for Fixture {
    fn drop(&mut self) {
        self.store.close();
    }
}

fn opts() -> QueueOpts {
    QueueOpts {
        visibility_timeout_s: 300,
        max_attempts: 3,
        backoff_base_s: 5,
        dead_letter_retention_s: None,
    }
}

fn now_unix() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_secs() as i64
}

/// Backdate a live row's `expires_at` past the wall clock (the honest
/// cross-connection probe for the sweep's both-states move — a fresh
/// enqueue would carry a future expiry, so the admin UPDATE backdates
/// it: schema-qualified, typed-binded).
async fn backdate_expires(admin: &tokio_postgres::Client, schema: &str, id: i64) {
    admin
        .execute(
            &format!(
                "UPDATE {} SET expires_at = $2 WHERE id = $1",
                crate::schema::QualifiedTable {
                    schema,
                    name: crate::schema::tables::JOB,
                }
            ),
            &[&id, &(now_unix() - 1)],
        )
        .await
        .unwrap();
}

/// Acceptance: the full lifecycle — enqueue stamps the resolved values
/// (resolved `run_at` after delay-over-`run_at`, absolute `expires_at`
/// after relative-`expires`, the handle's `QueueOpts` stamps with the
/// per-job `max_attempts` override), claim returns a boxed handle
/// carrying the row, ack deletes, `get_job` sees the states (ADR-010
/// §1/§3a; ADR-020 §1–§2; ADR-019 §1).
#[tokio::test(flavor = "multi_thread")]
async fn enqueue_claim_ack_lifecycle_with_resolved_stamps() {
    let Some(fx) = Fixture::open(&instance_namer("ql")).await else {
        eprintln!("skip: no harness server");
        return;
    };

    let qopts = QueueOpts {
        visibility_timeout_s: 120,
        max_attempts: 4,
        backoff_base_s: 9,
        dead_letter_retention_s: Some(7200),
    };
    let queue = fx.store.queue("emails", qopts.clone()).await.unwrap();
    assert_eq!(queue.name(), "emails");

    let before = now_unix();
    let id = queue
        .enqueue(
            json!({"to": "a@example.com"}),
            EnqueueOpts {
                delay: Some(60),
                run_at: Some(before + 10_000),
                priority: 3,
                max_attempts: Some(7),
                expires: Some(1200),
            },
        )
        .await
        .unwrap();

    let job = queue.get_job(id).await.unwrap().expect("row exists");
    assert_eq!(job.id, id);
    assert_eq!(job.queue, "emails");
    assert_eq!(job.state, JobState::Pending);
    assert_eq!(job.priority, 3);
    assert!(
        (before + 60..=before + 62).contains(&job.run_at),
        "ready time is now + delay, delay wins over run_at: {}",
        job.run_at
    );
    assert_eq!(
        job.expires_at,
        Some(before + 1200),
        "relative expires resolves to the absolute row expiry"
    );
    assert_eq!(job.max_attempts, 7, "the per-job override stamps the row");
    assert_eq!(job.visibility_timeout_s, 120, "the handle's stamp");
    assert_eq!(job.backoff_base_s, 9, "the handle's stamp");
    assert_eq!(job.dead_letter_retention_s, Some(7200));
    assert_eq!(job.attempts, 0);
    assert_eq!(job.worker_id, None);
    assert_eq!(job.claimed_at, None);
    assert_eq!(job.claim_expires_at, None);
    assert_eq!(job.last_error, None);
    assert_eq!(job.died_at, None);
    let decoded: Value = job.payload_as().unwrap();
    assert_eq!(decoded, json!({"to": "a@example.com"}));

    // Not yet claimable — ready time is a minute out.
    assert!(queue.claim_one("w1").await.unwrap().is_none());

    // A second enqueue with run_at in the past is claimable now.
    let id2 = queue
        .enqueue(
            json!({"n": 2}),
            EnqueueOpts {
                run_at: Some(before - 10),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let job2 = queue.get_job(id2).await.unwrap().unwrap();
    assert!(
        (before - 10..=before - 9).contains(&job2.run_at),
        "run_at is used literally when delay is unset"
    );

    // Claim: the handle carries the row, the claim counts an attempt,
    // the deadline comes from the job's own visibility stamp.
    let handle = queue.claim_one("w1").await.unwrap().expect("claimable");
    let job = handle.job();
    assert_eq!(job.id, id2);
    assert_eq!(job.state, JobState::Processing);
    assert_eq!(job.attempts, 1);
    assert_eq!(job.worker_id.as_deref(), Some("w1"));
    assert!(job.claimed_at.is_some());
    assert_eq!(
        job.claim_expires_at,
        Some(job.claimed_at.unwrap() + 120),
        "the deadline comes from the job's own visibility stamp"
    );

    // get_job reflects the claim.
    let seen = queue.get_job(id2).await.unwrap().unwrap();
    assert_eq!(seen.state, JobState::Processing);
    assert_eq!(seen.worker_id.as_deref(), Some("w1"));

    // A second worker cannot claim it (exactly-once handout).
    assert!(queue.claim_one("w2").await.unwrap().is_none());

    // Ack deletes the row (ADR-010 §1).
    assert!(handle.ack().await.unwrap());
    assert!(queue.get_job(id2).await.unwrap().is_none());
    assert!(queue.claim_one("w2").await.unwrap().is_none());
    assert!(
        queue.get_job(id).await.unwrap().is_some(),
        "the delayed row survives untouched"
    );

    drop_schema(&fx.admin, &fx.schema).await;
}

/// Acceptance: the extent guard (ADR-023 §2) — `claim_batch(n <= 0)`
/// claims nothing (the empty `Vec`, no error; pg's `LIMIT` with a
/// negative would be a server error, which the guard makes
/// unreachable) and `claim_one` on an empty queue is `None` (no-work
/// is a value).
#[tokio::test(flavor = "multi_thread")]
async fn extent_guard_claims_nothing_on_non_positive_n() {
    let Some(fx) = Fixture::open(&instance_namer("qe")).await else {
        eprintln!("skip: no harness server");
        return;
    };
    let queue = fx.store.queue("work", opts()).await.unwrap();

    for i in 0..5 {
        queue
            .enqueue(json!({"i": i}), EnqueueOpts::default())
            .await
            .unwrap();
    }

    for n in [-1i64, 0] {
        let got = queue.claim_batch("w1", n).await.unwrap();
        assert!(
            got.is_empty(),
            "claim_batch({n}) must claim nothing (empty Vec, no error)"
        );
    }
    // Nothing was destroyed by the guarded call.
    let seen = queue.get_job(1).await.unwrap().unwrap();
    assert_eq!(seen.state, JobState::Pending);

    // A positive claim on the same state still works and hands out
    // exactly the requested extent.
    let got = queue.claim_batch("w1", 2).await.unwrap();
    assert_eq!(got.len(), 2);

    drop_schema(&fx.admin, &fx.schema).await;
}

/// Acceptance: exactly-once handout under concurrent claims — two
/// workers racing claim on one row (exactly one wins; the POC's 4×4
/// shape at 2×1) and a 6×40 concurrent drain with zero double-claims
/// (the POC-pinned property, engine-tested here; the drain tasks share
/// the store through `Arc<ArcSwapLike>`-free sibling stores — every
/// store instance is a stateless client of the same schema'd tables,
/// so concurrent stores over one schema are the honest multi-worker
/// shape).
#[tokio::test(flavor = "multi_thread")]
async fn concurrent_claims_hand_each_row_out_exactly_once() {
    let Some(dsn) = harness_dsn() else {
        eprintln!("skip: no harness server");
        return;
    };
    let schema = instance_namer("qx")();
    let admin = harness_client().await.unwrap();

    // Two workers, one row: exactly one claim wins.
    {
        let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
        let queue = store.queue("solo", opts()).await.unwrap();
        let id = queue
            .enqueue(json!({"only": true}), EnqueueOpts::default())
            .await
            .unwrap();
        let q1 = store.queue("solo", opts()).await.unwrap();
        let q2 = store.queue("solo", opts()).await.unwrap();
        let (a, b) = tokio::join!(q1.claim_one("w1"), q2.claim_one("w2"));
        let winners = [a.unwrap(), b.unwrap()]
            .into_iter()
            .flatten()
            .collect::<Vec<_>>();
        assert_eq!(winners.len(), 1, "exactly one worker got the row");
        assert_eq!(winners[0].job().id, id);
        store.close();
    }

    // Many concurrent stores drain a whole queue with no double
    // handout (cross-instance claims — each drain task opens its own
    // pool connection; the SKIP LOCKED handout is the exactly-once
    // property under connection diversity too).
    const JOBS: i64 = 40;
    const WORKERS: usize = 6;
    {
        let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
        let queue = store.queue("shared_arc", opts()).await.unwrap();
        for i in 0..JOBS {
            queue
                .enqueue(json!({"i": i}), EnqueueOpts::default())
                .await
                .unwrap();
        }
        store.close();

        let factory = Arc::new(TestFactory {
            dsn: dsn.clone(),
            schema: schema.clone(),
        });
        let claimed = Arc::new(AtomicUsize::new(0));
        let mut handles = Vec::new();
        for w in 0..WORKERS {
            let factory = factory.clone();
            let claimed = claimed.clone();
            handles.push(tokio::spawn(async move {
                let store = factory.open().await;
                let queue = store.queue("shared_arc", opts()).await.unwrap();
                let mut got = 0usize;
                loop {
                    let batch = queue
                        .claim_batch(&format!("w{w}"), 4)
                        .await
                        .expect("claim must not error under contention");
                    if batch.is_empty() {
                        break;
                    }
                    got += batch.len();
                    // Hold the rows (no ack) — the drain semantics are
                    // what is under test.
                }
                claimed.fetch_add(got, Ordering::SeqCst);
                got
            }));
        }
        let mut shares = Vec::new();
        for h in handles {
            shares.push(h.await.unwrap());
        }
        assert_eq!(claimed.load(Ordering::SeqCst), JOBS as usize);
        assert!(
            shares.iter().any(|s| *s > 0) && shares.iter().any(|s| *s < JOBS as usize),
            "work was shared across workers, not drained by one"
        );

        // Every job is processing now; no id claimed twice (probed via
        // a raw admin count — the bigserial ids are not the loop's
        // 1..=n range, other tables' rows interleave).
        let raw = open_store(&dsn, test_opts(&schema)).await.unwrap();
        let job_table = crate::schema::QualifiedTable {
            schema: &schema,
            name: crate::schema::tables::JOB,
        }
        .to_string();
        let conn = raw.pool().get().await.unwrap();
        let processing: i64 = conn
            .query_one(
                &format!(
                    "SELECT count(*) FROM {job_table} WHERE state = 'processing' AND queue = 'shared_arc'"
                ),
                &[],
            )
            .await
            .unwrap()
            .get(0);
        let distinct: i64 = conn
            .query_one(
                &format!(
                    "SELECT count(DISTINCT id) FROM {job_table} WHERE state = 'processing' AND queue = 'shared_arc'"
                ),
                &[],
            )
            .await
            .unwrap()
            .get(0);
        assert_eq!(processing, JOBS, "every job handed out exactly once");
        assert_eq!(distinct, JOBS, "no id claimed twice");
        store.close();
        raw.close();
    }

    drop_schema(&admin, &schema).await;
}

/// A per-test store factory for the drain tasks (a fresh store
/// instance over the same schema per call — the honest cross-instance
/// worker shape).
struct TestFactory {
    dsn: String,
    schema: String,
}

impl TestFactory {
    async fn open(&self) -> PgStore {
        open_store(&self.dsn, test_opts(&self.schema))
            .await
            .unwrap()
    }
}

/// Acceptance: the uniform validity predicate — once the claim
/// deadline lapsed, all handle ops refuse with `Ok(false)` (heartbeat,
/// ack, retry, fail — never errors), the row untouched, and a reclaim
/// consumes an attempt (ADR-010 §2). Healthy renewal is pinned first:
/// absolute reset, repeatable, `claimed_at` untouched.
#[tokio::test(flavor = "multi_thread")]
async fn lapsed_deadline_refuses_all_handle_ops_and_reclaim_consumes_an_attempt() {
    let Some(fx) = Fixture::open(&instance_namer("qv")).await else {
        eprintln!("skip: no harness server");
        return;
    };

    // Healthy phase (visibility 300): renewal lands and is repeatable.
    let queue = fx.store.queue("healthy", opts()).await.unwrap();
    let id = queue
        .enqueue(json!({"work": true}), EnqueueOpts::default())
        .await
        .unwrap();
    let handle = queue.claim_one("w1").await.unwrap().unwrap();
    let first_deadline = handle.job().claim_expires_at.unwrap();
    assert!(handle.heartbeat(5000).await.unwrap(), "renewal lands");
    assert!(handle.heartbeat(5000).await.unwrap(), "repeatable");
    let renewed = queue.get_job(id).await.unwrap().unwrap();
    assert!(
        renewed.claim_expires_at.unwrap() >= first_deadline + 4500,
        "renewal is an absolute reset from now: {} vs {first_deadline}",
        renewed.claim_expires_at.unwrap()
    );
    assert!(
        renewed.claimed_at == handle.job().claimed_at,
        "heartbeat never touches claimed_at"
    );
    assert!(handle.ack().await.unwrap(), "healthy ack lands");

    // Lapsed phase (visibility 1): every op refuses with Ok(false).
    let queue = fx
        .store
        .queue(
            "lapsed",
            QueueOpts {
                visibility_timeout_s: 1,
                max_attempts: 5,
                backoff_base_s: 5,
                dead_letter_retention_s: None,
            },
        )
        .await
        .unwrap();

    let mut ids = Vec::new();
    for _ in 0..4 {
        ids.push(
            queue
                .enqueue(json!({"work": true}), EnqueueOpts::default())
                .await
                .unwrap(),
        );
    }
    let mut claims: Vec<Box<dyn JobHandle>> = Vec::new();
    for id in &ids {
        let h = queue
            .claim_batch("w1", 1)
            .await
            .unwrap()
            .into_iter()
            .next()
            .expect("claimable");
        assert_eq!(h.job().id, *id);
        claims.push(h);
    }

    for id in &ids {
        until_deadline_lapsed(queue.as_ref(), *id).await;
    }

    assert!(
        !claims[0].heartbeat(5000).await.unwrap(),
        "late heartbeat refuses"
    );
    assert!(!claims[0].heartbeat(5000).await.unwrap(), "still refuses");
    assert!(!claims.remove(1).ack().await.unwrap(), "lapsed ack refuses");
    assert!(
        !claims
            .remove(1)
            .retry(Some("boom".to_string()), Some(10))
            .await
            .unwrap(),
        "lapsed retry refuses"
    );
    assert!(
        !claims
            .remove(1)
            .fail(Some("boom".to_string()))
            .await
            .unwrap(),
        "lapsed fail refuses"
    );

    // The rows are untouched by the refusals.
    for id in &ids {
        let job = queue.get_job(*id).await.unwrap().unwrap();
        assert_eq!(job.state, JobState::Processing);
        assert_eq!(job.attempts, 1);
    }

    // A reclaim wins and consumes an attempt.
    let reclaimed = queue.claim_one("w2").await.unwrap().expect("reclaimable");
    assert_eq!(reclaimed.job().attempts, 2, "the reclaim counts an attempt");
    assert_eq!(reclaimed.job().worker_id.as_deref(), Some("w2"));

    // Post-reclaim, the original claim's ops refuse (the reclaim won
    // atomically).
    assert!(
        !claims[0].heartbeat(5000).await.unwrap(),
        "the original claim's ops refuse post-reclaim"
    );

    drop_schema(&fx.admin, &fx.schema).await;
}

/// Wait until the claimed row's deadline has lapsed — the bounded-wait
/// posture (no tight-race assertions).
async fn until_deadline_lapsed(queue: &dyn alkstore::Queue, id: i64) {
    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
    loop {
        let lapsed = match queue.get_job(id).await {
            Ok(Some(job)) => {
                job.claim_expires_at.is_some() && job.claim_expires_at.unwrap() < now_unix()
            }
            _ => false,
        };
        if lapsed {
            return;
        }
        assert!(tokio::time::Instant::now() < deadline, "never lapsed");
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
}

/// Acceptance: the backoff curve — `retry(err, None)` computes the
/// equal-jitter exponential `delay ∈ [base·2^(a−1)/2, base·2^(a−1)]`,
/// capped at 1 hour (ADR-010 §3): statistical pin over many draws,
/// tolerance-bounded; the row's `run_at` after an end-to-end retry
/// lands in the range; `Some(d)` overrides.
#[tokio::test(flavor = "multi_thread")]
async fn backoff_curve_lands_in_the_pinned_range() {
    // Statistical pin of the pure formula (a = attempts index): base
    // 5, a=1 → [3, 5]; draws must stay in the inclusive integer range
    // and spread across most of it.
    for (base, attempts, lo, hi) in [(5, 1, 3, 5), (5, 2, 5, 10), (5, 3, 10, 20), (1, 1, 1, 1)] {
        let mut lo_seen = i64::MAX;
        let mut hi_seen = i64::MIN;
        for id in 0..400 {
            let d = backoff_delay_s(base, attempts, id);
            assert!(
                (lo..=hi).contains(&d),
                "delay {d} outside [{lo}, {hi}] for base={base} a={attempts}"
            );
            lo_seen = lo_seen.min(d);
            hi_seen = hi_seen.max(d);
        }
        if lo == hi {
            assert_eq!(lo_seen, hi_seen, "degenerate range pins to the value");
        } else {
            assert!(
                lo_seen <= lo + (hi - lo) / 2 && hi_seen >= hi - (hi - lo) / 2,
                "draws must spread over the range ([{lo_seen}, {hi_seen}] vs [{lo}, {hi}])"
            );
        }
    }

    // Cap at 1 hour: a huge doubling exponent still lands in [1800, 3600].
    for id in 0..200 {
        let d = backoff_delay_s(5, 40, id);
        assert!((1800..=3600).contains(&d), "capped delay {d} out of range");
    }

    // End-to-end: the row's run_at after retry(err, None) is now plus
    // a curve draw from the job's stamped base.
    let Some(fx) = Fixture::open(&instance_namer("qb")).await else {
        eprintln!("skip: no harness server");
        return;
    };
    let queue = fx
        .store
        .queue(
            "retrying",
            QueueOpts {
                visibility_timeout_s: 300,
                max_attempts: 5,
                backoff_base_s: 8,
                dead_letter_retention_s: None,
            },
        )
        .await
        .unwrap();
    let id = queue
        .enqueue(json!({"task": 1}), EnqueueOpts::default())
        .await
        .unwrap();
    let handle = queue.claim_one("w1").await.unwrap().unwrap();
    assert_eq!(handle.job().attempts, 1, "a = the post-claim count");
    let before = now_unix();
    assert!(handle.retry(None, None).await.unwrap());
    let job = queue.get_job(id).await.unwrap().unwrap();
    assert_eq!(job.state, JobState::Pending);
    let span_lo = before - 1 + 8 / 2;
    let span_hi = before + 1 + 8;
    assert!(
        (span_lo..=span_hi).contains(&job.run_at),
        "run_at {} must be now + a draw from [4, 8] (base 8, a=1)",
        job.run_at
    );

    // Some(d) overrides: wait out the curve draw, claim again, retry
    // with the explicit delay.
    let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
    let override_landed = loop {
        match queue.claim_one("w1").await {
            Ok(Some(h)) => {
                let ok = h
                    .retry(Some("still broken".to_string()), Some(123))
                    .await
                    .unwrap();
                assert!(ok, "the override retry lands");
                break true;
            }
            Ok(None) => {}
            Err(e) => panic!("claim errored: {e}"),
        }
        assert!(
            tokio::time::Instant::now() < deadline,
            "curve never elapsed"
        );
        tokio::time::sleep(Duration::from_millis(200)).await;
    };
    assert!(override_landed);
    let job = queue.get_job(id).await.unwrap().unwrap();
    let now = now_unix();
    assert!(
        (now + 123 - 2..=now + 123 + 2).contains(&job.run_at) && job.run_at >= before + 120,
        "the explicit delay passes through: {} (now {now})",
        job.run_at
    );

    drop_schema(&fx.admin, &fx.schema).await;
}

/// Acceptance: dead-letter — `retry` at exhausted budget moves the row
/// to dead with the pinned default string (`"max attempts exceeded"` —
/// or the caller's `Some(err)` verbatim), `fail(None)` records
/// `"failed"`, `get_job` sees dead rows with `last_error`/`died_at`
/// carrying the full stamp list (ADR-010 §1/§4; ADR-019 §3).
#[tokio::test(flavor = "multi_thread")]
async fn exhaustion_and_fail_dead_letter_with_the_pinned_strings() {
    let Some(fx) = Fixture::open(&instance_namer("qd")).await else {
        eprintln!("skip: no harness server");
        return;
    };
    let queue = fx
        .store
        .queue(
            "dying",
            QueueOpts {
                visibility_timeout_s: 300,
                max_attempts: 2,
                backoff_base_s: 5,
                dead_letter_retention_s: None,
            },
        )
        .await
        .unwrap();

    // Exhaustion via retry: claim (attempts=1), retry under budget →
    // pending, reclaim (attempts=2 == max) → the retry dead-letters.
    let id1 = queue
        .enqueue(json!({"n": 1}), EnqueueOpts::default())
        .await
        .unwrap();
    let h = queue.claim_one("w1").await.unwrap().unwrap();
    assert!(h.retry(None, Some(0)).await.unwrap(), "under budget");
    let job = queue.get_job(id1).await.unwrap().unwrap();
    assert_eq!(job.state, JobState::Pending);
    assert_eq!(job.attempts, 1);
    let h = queue.claim_one("w1").await.unwrap().unwrap();
    assert_eq!(h.job().attempts, 2, "a reclaim counts too");
    assert!(
        h.retry(Some("caller string".to_string()), None)
            .await
            .unwrap(),
        "the exhausted-budget retry lands as the dead-letter move"
    );
    let dead = queue
        .get_job(id1)
        .await
        .unwrap()
        .expect("dead rows visible");
    assert_eq!(dead.state, JobState::Dead);
    assert_eq!(
        dead.last_error.as_deref(),
        Some("caller string"),
        "retry(Some(err)) carries the caller's string on the dead row"
    );
    assert!(dead.died_at.is_some());
    assert_eq!(dead.max_attempts, 2, "the dead row carries its stamps");
    assert_eq!(dead.visibility_timeout_s, 300);
    assert_eq!(dead.backoff_base_s, 5);
    assert_eq!(dead.attempts, 2);
    assert!(
        dead.claimed_at.is_some(),
        "the dead row keeps claim history"
    );
    assert_eq!(dead.worker_id.as_deref(), Some("w1"));

    // retry(None) exhaustion pins the default string.
    let id2 = queue
        .enqueue(json!({"n": 2}), EnqueueOpts::default())
        .await
        .unwrap();
    let h = queue.claim_one("w1").await.unwrap().unwrap();
    assert_eq!(h.job().attempts, 1, "a fresh claim is one attempt");
    assert!(h.retry(None, Some(0)).await.unwrap(), "attempts 1 < 2");
    let h = queue.claim_one("w1").await.unwrap().unwrap();
    assert_eq!(h.job().attempts, 2);
    assert!(h.retry(None, None).await.unwrap(), "at-budget dead-letters");
    let dead = queue.get_job(id2).await.unwrap().unwrap();
    assert_eq!(dead.state, JobState::Dead);
    assert_eq!(
        dead.last_error.as_deref(),
        Some(EXHAUSTED_ERROR),
        "retry(None) exhaustion records the pinned default string"
    );
    assert!(dead.died_at.is_some());

    drop_schema(&fx.admin, &fx.schema).await;

    // fail(None) pins "failed"; fail(Some(e)) carries the caller's.
    let Some(fx) = Fixture::open(&instance_namer("qf")).await else {
        eprintln!("skip: no harness server");
        return;
    };
    let queue = fx.store.queue("dying", opts()).await.unwrap();
    let id = queue
        .enqueue(json!({"n": 3}), EnqueueOpts::default())
        .await
        .unwrap();
    let h = queue.claim_one("w1").await.unwrap().unwrap();
    assert!(h.fail(None).await.unwrap(), "fail lands");
    let dead = queue.get_job(id).await.unwrap().unwrap();
    assert_eq!(dead.state, JobState::Dead);
    assert_eq!(
        dead.last_error.as_deref(),
        Some(FAILED_ERROR),
        "fail(None) records the pinned default string"
    );
    assert!(dead.died_at.is_some());

    let id = queue
        .enqueue(json!({"n": 4}), EnqueueOpts::default())
        .await
        .unwrap();
    let h = queue.claim_one("w1").await.unwrap().unwrap();
    assert!(h.fail(Some("explicit reason".to_string())).await.unwrap());
    let dead = queue.get_job(id).await.unwrap().unwrap();
    assert_eq!(dead.last_error.as_deref(), Some("explicit reason"));

    // The pinning constants are the contract's defaults.
    assert_eq!(EXHAUSTED_ERROR, "max attempts exceeded");
    assert_eq!(FAILED_ERROR, "failed");
    assert_eq!(EXPIRED_ERROR, "expired");

    drop_schema(&fx.admin, &fx.schema).await;
}

/// Acceptance: the pre-claim dead-letter sweep — a reclaimable row
/// at-or-past its budget is dead-lettered by the next claim (the
/// substrate's D-12 shape: a worker dying right after its last allowed
/// claim dead-letters, not strands — ADR-010 §2/§3).
#[tokio::test(flavor = "multi_thread")]
async fn pre_claim_sweep_dead_letters_exhausted_reclaimables() {
    let Some(fx) = Fixture::open(&instance_namer("q12")).await else {
        eprintln!("skip: no harness server");
        return;
    };
    let queue = fx
        .store
        .queue(
            "d12",
            QueueOpts {
                visibility_timeout_s: 1,
                max_attempts: 1,
                backoff_base_s: 5,
                dead_letter_retention_s: None,
            },
        )
        .await
        .unwrap();
    let spent = queue
        .enqueue(json!({"k": 1}), EnqueueOpts::default())
        .await
        .unwrap();
    let fresh = queue
        .enqueue(json!({"k": 2}), EnqueueOpts::default())
        .await
        .unwrap();

    // Claim the budget-spent row (attempts 1 == max 1) and let its
    // deadline lapse: it is now a reclaimable-past-budget row.
    let h = queue.claim_one("w1").await.unwrap().unwrap();
    assert_eq!(h.job().id, spent);
    until_deadline_lapsed(queue.as_ref(), spent).await;

    // The next claim dead-letters the spent row (the pre-claim sweep)
    // and hands out the fresh one.
    let h = queue
        .claim_one("w2")
        .await
        .unwrap()
        .expect("fresh claimable");
    assert_eq!(h.job().id, fresh);
    let dead = queue.get_job(spent).await.unwrap().expect("dead visible");
    assert_eq!(dead.state, JobState::Dead);
    assert_eq!(
        dead.last_error.as_deref(),
        Some(EXHAUSTED_ERROR),
        "the exhausted reclaimable dead-letters (not strands)"
    );
    assert!(dead.died_at.is_some());

    drop_schema(&fx.admin, &fx.schema).await;
}

/// Acceptance: claim ordering — `priority DESC`, ready-time ASC,
/// enqueue order (ADR-010 §1).
#[tokio::test(flavor = "multi_thread")]
async fn claim_ordering_is_priority_desc_run_at_then_enqueue_order() {
    let Some(fx) = Fixture::open(&instance_namer("qo")).await else {
        eprintln!("skip: no harness server");
        return;
    };
    let queue = fx.store.queue("ordered", opts()).await.unwrap();
    let now = now_unix();
    let a = queue
        .enqueue(
            json!("a"),
            EnqueueOpts {
                run_at: Some(now),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let b = queue
        .enqueue(
            json!("b"),
            EnqueueOpts {
                run_at: Some(now),
                priority: 5,
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let c = queue
        .enqueue(
            json!("c"),
            EnqueueOpts {
                run_at: Some(now),
                priority: 5,
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let d = queue
        .enqueue(
            json!("d"),
            EnqueueOpts {
                run_at: Some(now - 10),
                priority: 5,
                ..Default::default()
            },
        )
        .await
        .unwrap();

    for _ in 0..3 {
        queue.claim_one("w").await.unwrap().unwrap();
    }
    // After 3 claims: d, b, c claimed (priority DESC; among equal
    // priority, run_at ASC → d's earlier ready time, then enqueue
    // order b then c); a (priority 0) is last, unclaimed.
    let ja = queue.get_job(a).await.unwrap().unwrap();
    assert_eq!(ja.state, JobState::Pending, "the lower-priority job last");
    assert_eq!(ja.worker_id, None);
    for id in [b, c, d] {
        let j = queue.get_job(id).await.unwrap().unwrap();
        assert_eq!(j.state, JobState::Processing, "claimed in order");
        assert_eq!(j.worker_id.as_deref(), Some("w"));
    }

    drop_schema(&fx.admin, &fx.schema).await;
}

/// Acceptance: `cancel` is unconditional (either state, any holder),
/// `ack_batch` applies the per-id validity predicate without a worker
/// filter (partial success ordinary), missing ids silent (ADR-010 §1;
/// ADR-019 §1).
#[tokio::test(flavor = "multi_thread")]
async fn cancel_and_ack_batch_behaviors() {
    let Some(fx) = Fixture::open(&instance_namer("qc")).await else {
        eprintln!("skip: no harness server");
        return;
    };
    let queue = fx.store.queue("ops", opts()).await.unwrap();

    let pending = queue
        .enqueue(json!({"p": 1}), EnqueueOpts::default())
        .await
        .unwrap();
    let processing = queue
        .enqueue(json!({"p": 2}), EnqueueOpts::default())
        .await
        .unwrap();
    let h = queue.claim_one("w1").await.unwrap().unwrap();
    let claimed_first = h.job().id;
    assert!(claimed_first == pending || claimed_first == processing);
    let still_live = if claimed_first == pending {
        processing
    } else {
        pending
    };
    assert!(queue.cancel(claimed_first).await.unwrap());
    assert!(queue.cancel(still_live).await.unwrap(), "unconditional");
    assert!(!queue.cancel(still_live).await.unwrap(), "gone = false");
    assert!(queue.get_job(still_live).await.unwrap().is_none());
    assert!(!h.ack().await.unwrap(), "the holder's next op refuses");

    // ack_batch: per-id predicate, no worker filter, count returned.
    let a = queue
        .enqueue(json!({"a": 1}), EnqueueOpts::default())
        .await
        .unwrap();
    let b = queue
        .enqueue(json!({"b": 1}), EnqueueOpts::default())
        .await
        .unwrap();
    let c = queue
        .enqueue(json!({"c": 1}), EnqueueOpts::default())
        .await
        .unwrap();
    let ha = queue.claim_one("w1").await.unwrap().unwrap();
    let hb = queue.claim_one("w2").await.unwrap().unwrap();
    assert_eq!(ha.job().id, a, "FIFO: the first-enqueued row claims first");
    assert_eq!(hb.job().id, b);
    assert!(queue.cancel(c).await.unwrap());

    // a is w1's claim, b is w2's — both count (no worker filter); the
    // cancelled c and the missing id do not.
    let count = queue.ack_batch(&[a, b, c, 999_999]).await.unwrap();
    assert_eq!(count, 2, "only the valid ids count; partial is ordinary");
    assert!(queue.get_job(a).await.unwrap().is_none());
    assert!(queue.get_job(b).await.unwrap().is_none());

    // An empty batch is vacuously zero.
    assert_eq!(queue.ack_batch(&[]).await.unwrap(), 0);

    drop_schema(&fx.admin, &fx.schema).await;
}

/// Acceptance: `sweep_expired` moves every past-`expires_at` row (both
/// states — the no-stranded-rows property, ADR-010 §5) to dead with
/// `"expired"`, and returns the moved count; unexpired rows are
/// untouched; no leader lock required (the bare call just runs).
#[tokio::test(flavor = "multi_thread")]
async fn sweep_expired_moves_both_states_and_applies_retention() {
    let Some(fx) = Fixture::open(&instance_namer("qs")).await else {
        eprintln!("skip: no harness server");
        return;
    };
    let queue = fx.store.queue("sweeping", opts()).await.unwrap();
    let now = now_unix();

    // Expired pending + expired processing + unexpired: only the two
    // expired rows move; the count is 2. The claim takes the head (the
    // expired-pending row, FIFO) then the honest cross-connection probe
    // back-dates both rows' expiry so the both-states move is
    // exercised deterministically.
    let expired_pending = queue
        .enqueue(
            json!({"k": 1}),
            EnqueueOpts {
                expires: Some(3600),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let expired_processing = queue
        .enqueue(
            json!({"k": 2}),
            EnqueueOpts {
                expires: Some(3600),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let unexpired = queue
        .enqueue(
            json!({"k": 3}),
            EnqueueOpts {
                expires: Some(3600),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    let h = queue.claim_one("w1").await.unwrap().unwrap();
    assert_eq!(h.job().id, expired_pending);
    backdate_expires(&fx.admin, &fx.schema, expired_pending).await;
    backdate_expires(&fx.admin, &fx.schema, expired_processing).await;

    let moved = queue.sweep_expired().await.unwrap();
    assert_eq!(moved, 2, "both expired rows moved (pending + processing)");

    for id in [expired_pending, expired_processing] {
        let dead = queue.get_job(id).await.unwrap().expect("moved to dead");
        assert_eq!(dead.state, JobState::Dead);
        assert_eq!(dead.last_error.as_deref(), Some(EXPIRED_ERROR));
        assert!(dead.died_at.is_some() && dead.died_at.unwrap() >= now);
    }
    let live = queue.get_job(unexpired).await.unwrap().unwrap();
    assert_eq!(live.state, JobState::Pending, "unexpired rows untouched");

    // A bare sweep on an empty-expiry queue is 0 and not an error.
    assert_eq!(queue.sweep_expired().await.unwrap(), 0);

    drop_schema(&fx.admin, &fx.schema).await;
}

/// Acceptance: retention — dead rows past their
/// `dead_letter_retention_s` stamp are deleted by the sweep (the only
/// sweeper dead rows ever have), rows with `None` retention live
/// forever; the deleted count reports through (ADR-010 §4/§5).
#[tokio::test(flavor = "multi_thread")]
async fn sweep_enforces_dead_letter_retention_incl_forever_none() {
    let Some(fx) = Fixture::open(&instance_namer("qr")).await else {
        eprintln!("skip: no harness server");
        return;
    };
    let keeping = fx
        .store
        .queue(
            "keeping",
            QueueOpts {
                dead_letter_retention_s: None,
                ..opts()
            },
        )
        .await
        .unwrap();
    let expiring = fx
        .store
        .queue(
            "expiring",
            QueueOpts {
                dead_letter_retention_s: Some(60),
                ..opts()
            },
        )
        .await
        .unwrap();

    let keep_id = keeping
        .enqueue(json!({"k": 1}), EnqueueOpts::default())
        .await
        .unwrap();
    let expire_id = expiring
        .enqueue(json!({"k": 2}), EnqueueOpts::default())
        .await
        .unwrap();
    let h = keeping.claim_one("w1").await.unwrap().unwrap();
    assert!(h.fail(None).await.unwrap());
    let h = expiring.claim_one("w1").await.unwrap().unwrap();
    assert!(h.fail(None).await.unwrap());

    // Move the expired-retention row's death back past its TTL (the
    // honest probe: retention reads died_at vs now).
    fx.admin
        .execute(
            &format!(
                "UPDATE {} SET died_at = $2 WHERE id = $1",
                crate::schema::QualifiedTable {
                    schema: &fx.schema,
                    name: crate::schema::tables::DEAD,
                }
            ),
            &[&expire_id, &(now_unix() - 3600)],
        )
        .await
        .unwrap();

    // The sweep on the expiring queue deletes the TTL'd row; the
    // keeping queue's row (retention None) survives.
    assert_eq!(expiring.sweep_expired().await.unwrap(), 1);
    assert!(expiring.get_job(expire_id).await.unwrap().is_none());
    assert_eq!(keeping.sweep_expired().await.unwrap(), 0);
    assert!(
        keeping.get_job(keep_id).await.unwrap().is_some(),
        "retention None = dead rows live forever"
    );

    drop_schema(&fx.admin, &fx.schema).await;
}

/// Acceptance: the wake-driven claim loop pinned end-to-end — enqueue
/// → pg_notify on the queue's wake channel → the LISTEN forwarder fan
/// → cross-connection wake → the claim (the LISTEN-driven default
/// consumption posture, ADR-004; the contract claim ops are
/// pull-shaped — a consumer's loop listens and claims on wake; the
/// re-poll safety net is the loop's fallback, pinned here only as the
/// pull-shaped surface's honest shape).
#[tokio::test(flavor = "multi_thread")]
async fn wake_driven_claim_loop_end_to_end() {
    let Some(fx) = Fixture::open(&instance_namer("qw")).await else {
        eprintln!("skip: no harness server");
        return;
    };
    let queue = fx.store.queue("wakeup", opts()).await.unwrap();
    let mut receiver = fx.store.listen("wakeup").await.unwrap();

    // Enqueue on the handle (a pooled connection, not the listener's):
    // the row lands, the wake notifies on the queue's wake channel.
    let id = queue
        .enqueue(json!({"work": true}), EnqueueOpts::default())
        .await
        .unwrap();

    // The wake arrives cross-connection (the enqueue's NOTIFY fired at
    // its implicit transaction's commit and the forwarder forwarded
    // it).
    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
    loop {
        match receiver.recv_timeout(Duration::from_millis(300)).await {
            Ok(Some(wake)) => {
                assert_eq!(wake.channel, "wakeup");
                break;
            }
            Ok(None) => {}
            Err(e) => panic!("receiver errored: {e}"),
        }
        assert!(
            tokio::time::Instant::now() < deadline,
            "the enqueue's wake never arrived"
        );
    }

    // The claim lands on the consumer's own pooled connection.
    let handle = queue.claim_one("w1").await.unwrap().expect("claimable");
    assert_eq!(handle.job().id, id);
    assert!(handle.ack().await.unwrap());
    assert!(queue.get_job(id).await.unwrap().is_none());

    // A poll-only consumer (the safety net's shape — no LISTEN wake at
    // all, just a claim loop) still lands the same work: a second
    // enqueue is claimed by an unreactive loop alone.
    let second = queue
        .enqueue(json!({"work": 2}), EnqueueOpts::default())
        .await
        .unwrap();
    let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
    let claimed = loop {
        if let Some(h) = queue.claim_one("w2").await.unwrap() {
            break h;
        }
        assert!(
            tokio::time::Instant::now() < deadline,
            "the poll-only loop never landed the row"
        );
        tokio::time::sleep(Duration::from_millis(50)).await;
    };
    assert_eq!(claimed.job().id, second);

    drop_schema(&fx.admin, &fx.schema).await;
}

/// Acceptance: entry-point validation — the constructor rejects empty
/// and reserved names (`InvalidName`/`ReservedName`), claims reject
/// empty `worker_id` (ADR-008 §4; ADR-019 §2) — nothing round-trips.
#[tokio::test(flavor = "multi_thread")]
async fn validation_rejects_bad_names() {
    let Some(fx) = Fixture::open(&instance_namer("qval")).await else {
        eprintln!("skip: no harness server");
        return;
    };

    match fx.store.queue("", opts()).await {
        Err(alkstore::Error::InvalidName { .. }) => {}
        _other => panic!("empty queue name must be InvalidName"),
    }
    match fx.store.queue(alkstore::RESERVED_PREFIX, opts()).await {
        Err(alkstore::Error::ReservedName { .. }) => {}
        _other => panic!("reserved queue name must be ReservedName"),
    }

    let queue = fx.store.queue("valid", opts()).await.unwrap();
    match queue.claim_batch("", 1).await {
        Err(alkstore::Error::InvalidName { .. }) => {}
        _other => panic!("empty worker_id must be InvalidName"),
    }
    // The non-positive extent guards to the empty value even under a
    // valid worker (ADR-023 §2).
    let got = queue.claim_batch("w1", 0).await.unwrap();
    assert!(got.is_empty());

    drop_schema(&fx.admin, &fx.schema).await;
}

/// Acceptance: closed-store fail-closed on all queue ops (the wave-3
/// shape) — the constructor, enqueue, claims, ack_batch, cancel,
/// get_job, sweep, heartbeat.
#[tokio::test(flavor = "multi_thread")]
async fn closed_store_fails_closed_on_all_queue_ops() {
    let Some(fx) = Fixture::open(&instance_namer("qcl")).await else {
        eprintln!("skip: no harness server");
        return;
    };
    let queue = fx.store.queue("valid", opts()).await.unwrap();
    let id = queue
        .enqueue(json!({}), EnqueueOpts::default())
        .await
        .unwrap();
    let handle = queue.claim_one("w1").await.unwrap().unwrap();

    fx.store.close();

    let err = fx
        .store
        .queue("valid", opts())
        .await
        .err()
        .expect("closed store refuses construction");
    assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}");
    let err = queue
        .enqueue(json!({}), EnqueueOpts::default())
        .await
        .expect_err("closed store refuses enqueue");
    assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}");
    let err = queue
        .claim_batch("w1", 1)
        .await
        .err()
        .expect("closed store refuses claim");
    assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}");
    let err = queue
        .ack_batch(&[id])
        .await
        .expect_err("closed store refuses ack_batch");
    assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}");
    let err = queue
        .cancel(id)
        .await
        .expect_err("closed store refuses cancel");
    assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}");
    let err = queue
        .get_job(id)
        .await
        .expect_err("closed store refuses get_job");
    assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}");
    let err = queue
        .sweep_expired()
        .await
        .expect_err("closed store refuses sweep");
    assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}");
    let err = handle
        .heartbeat(100)
        .await
        .expect_err("closed store refuses heartbeat");
    assert!(matches!(err, alkstore::Error::Database(_)), "got {err:?}");

    drop_schema(&fx.admin, &fx.schema).await;
}

/// Acceptance: queue scoping — two handles on two queues are isolated:
/// a claim on one queue never hands out another queue's row (the
/// queue-scoped claim predicate), `get_job` is queue-scoped, and
/// cancel/ack_batch/sweep are queue-scoped (the seam's scoping rule at
/// the auto-commit shape).
#[tokio::test(flavor = "multi_thread")]
async fn queue_handles_are_name_scoped() {
    let Some(fx) = Fixture::open(&instance_namer("qsc")).await else {
        eprintln!("skip: no harness server");
        return;
    };
    let q1 = fx.store.queue("one", opts()).await.unwrap();
    let q2 = fx.store.queue("two", opts()).await.unwrap();
    let id1 = q1
        .enqueue(json!({"q": 1}), EnqueueOpts::default())
        .await
        .unwrap();
    let id2 = q2
        .enqueue(json!({"q": 2}), EnqueueOpts::default())
        .await
        .unwrap();

    // q1's claim only ever sees q1's rows; the second claim is
    // no-work.
    let h = q1.claim_one("w1").await.unwrap().unwrap();
    assert_eq!(h.job().id, id1);
    assert!(q1.claim_one("w1").await.unwrap().is_none());
    // q2 still hands out its own.
    let h2 = q2.claim_one("w2").await.unwrap().unwrap();
    assert_eq!(h2.job().id, id2);

    // get_job is queue-scoped: q1 cannot see q2's row.
    assert!(q1.get_job(id2).await.unwrap().is_none());
    assert!(q2.get_job(id1).await.unwrap().is_none());
    assert!(q1.get_job(id1).await.unwrap().is_some());

    // Cancel/ack_batch are queue-scoped too.
    assert!(!q1.cancel(id2).await.unwrap(), "cancel is q-scoped");
    assert!(q2.cancel(id2).await.unwrap());
    assert_eq!(q1.ack_batch(&[id2]).await.unwrap(), 0);

    drop_schema(&fx.admin, &fx.schema).await;
}

/// The dead-letter string constants are the contract's pinned values
/// (the engine-default strings land identically on both engines — the
/// backlog row's wording; server-less pin, the behavioral rows above
/// exercise them through the harness).
#[test]
fn dead_letter_strings_match_the_pinned_defaults() {
    assert_eq!(EXHAUSTED_ERROR, "max attempts exceeded");
    assert_eq!(FAILED_ERROR, "failed");
    assert_eq!(EXPIRED_ERROR, "expired");
}

/// The backoff formula's unit pins (server-less): the curve's
/// attempt-index clamping and the pure formula's range behavior — the
/// statistical rows run harness-side.
#[test]
fn backoff_formula_unit_pins() {
    // attempts 1 == the first retry draws from [ceil(base/2), base].
    for base in [1, 2, 5, 100] {
        for id in 0..200 {
            let d = backoff_delay_s(base, 1, id);
            assert!(((base + 1) / 2..=base).contains(&d), "base {base} d {d}");
        }
    }
    // A zero attempts index clamps to the first-retry attempt (the
    // formula's `max(1)` floor — the draw's subsecond input varies, so
    // pin the *range* the clamp produces, not call equality).
    for id in 0..200 {
        let d = backoff_delay_s(5, 0, id);
        assert!(
            (3..=5).contains(&d),
            "clamped-attempts draw {d} out of range"
        );
    }
    // The cap saturates the doubling arithmetic (no overflow at any
    // attempt count).
    for attempts in [30, 31, 60, 10_000] {
        for id in 0..100 {
            let d = backoff_delay_s(1, attempts, id);
            assert!((1..=3600).contains(&d));
        }
    }
}

/// Acceptance: the server-side stamp-arithmetic guard (ADR-023 §2's
/// 2026-10-11 annotation) — extreme i64 stamps and per-call arguments
/// are values whose server-side sums saturate to the i64 domain, never
/// "bigint out of range" failures, and the sticky-poison property (one
/// extreme-stamped row failing every later claim that touches it) is
/// retired. Pinned per site against the live server: an
/// `visibility_timeout_s = i64::MAX` row claims to the saturated
/// `i64::MAX` deadline and neighbors claim unaffected; extreme
/// heartbeat extends renew to the saturated edge and the representable
/// `now + i64::MIN` sum rides exact (Rust `saturating_add` parity),
/// the past deadline voiding through the validity predicate on the
/// next op; the retry override reschedules `Some(i64::MAX)` to
/// `i64::MAX` (and the negative-extreme delay rides exact to
/// ready-now); the retention `i64::MIN` arm sweeps every dead row
/// (`sat(now − MIN) = i64::MAX`) while the `Some(i64::MAX)` arm
/// deletes nothing.
#[tokio::test(flavor = "multi_thread")]
async fn extreme_stamps_saturate_and_no_longer_poison_queue_ops() {
    let Some(fx) = Fixture::open(&instance_namer("qext")).await else {
        eprintln!("skip: no harness server");
        return;
    };
    let now = now_unix();

    // Claim + heartbeat over an i64::MAX-stamped row: the claim UPDATE
    // sums now + visibility_timeout_s server-side (pre-guard this
    // failed sticky per row) — it must land, deadline at the saturated
    // edge.
    let queue = fx
        .store
        .queue(
            "neverlaps",
            QueueOpts {
                visibility_timeout_s: i64::MAX,
                ..opts()
            },
        )
        .await
        .unwrap();
    let id = queue
        .enqueue(json!({"k": 1}), EnqueueOpts::default())
        .await
        .unwrap();
    let h = queue
        .claim_one("w1")
        .await
        .unwrap()
        .expect("the extreme-stamped row claims (no overflow failure)");
    assert_eq!(h.job().id, id);
    assert_eq!(
        h.job().claim_expires_at,
        Some(i64::MAX),
        "the claim deadline saturates to the i64::MAX edge"
    );

    // A neighboring normal queue's row claims unaffected (the poison
    // was per-row, not per-store; the other-rows-unaffected arm).
    let normal = fx.store.queue("normal", opts()).await.unwrap();
    let normal_id = normal
        .enqueue(json!({"k": 2}), EnqueueOpts::default())
        .await
        .unwrap();
    let nh = normal
        .claim_one("w1")
        .await
        .unwrap()
        .expect("normal rows still claim");
    assert_eq!(nh.job().id, normal_id);
    assert!(
        nh.job().claim_expires_at.unwrap() >= now + 300,
        "normal rows still claim (unaffected by the extreme row)"
    );
    assert!(nh.ack().await.unwrap(), "the normal row acks");

    // Heartbeats on the extreme row land: a normal renewal, then the
    // extreme extend renewing to the saturated edge.
    assert!(h.heartbeat(120).await.unwrap(), "heartbeat renews");
    let renewed = queue.get_job(id).await.unwrap().unwrap();
    assert!(
        renewed.claim_expires_at.unwrap() > now,
        "renewal is an absolute reset from now: {:?}",
        renewed.claim_expires_at
    );
    assert!(
        h.heartbeat(i64::MAX).await.unwrap(),
        "the i64::MAX extend renews (no overflow failure)"
    );
    let renewed = queue.get_job(id).await.unwrap().unwrap();
    assert_eq!(
        renewed.claim_expires_at,
        Some(i64::MAX),
        "sat(now + i64::MAX) = i64::MAX"
    );

    // The negative extreme: now + i64::MIN is representable, so the
    // guard rides it exact (matching Rust `saturating_add` — no
    // gratuitous clamp) and the past deadline it stamps voids the
    // claim through the validity predicate on the next op.
    let before_min = now_unix();
    assert!(
        h.heartbeat(i64::MIN).await.unwrap(),
        "the i64::MIN extend renews"
    );
    let renewed = queue.get_job(id).await.unwrap().unwrap();
    let offset = renewed.claim_expires_at.unwrap().wrapping_sub(i64::MIN);
    assert!(
        (before_min..=before_min + 2).contains(&offset),
        "sat(now + i64::MIN) rides exact: deadline {:?} vs before {before_min}",
        renewed.claim_expires_at
    );
    assert!(
        !h.heartbeat(60).await.unwrap(),
        "the past saturated deadline voids the claim on the next op"
    );

    // Retry override extremes: `Some(i64::MIN)` rides exact to
    // ready-now (the row reclaims immediately), `Some(i64::MAX)`
    // reschedules to the ready-~never saturated edge.
    let id2 = normal
        .enqueue(json!({"k": 5}), EnqueueOpts::default())
        .await
        .unwrap();
    let h = normal.claim_one("w1").await.unwrap().unwrap();
    let before = now_unix();
    assert!(
        h.retry(Some("boom".to_string()), Some(i64::MIN))
            .await
            .unwrap(),
        "the negative-extreme delay retry lands"
    );
    let job = normal.get_job(id2).await.unwrap().unwrap();
    assert_eq!(job.state, JobState::Pending);
    let offset = job.run_at.wrapping_sub(i64::MIN);
    assert!(
        (before..=before + 2).contains(&offset),
        "sat(now + i64::MIN) rides exact: run_at {} vs now {before}",
        job.run_at
    );
    let h = normal
        .claim_one("w1")
        .await
        .unwrap()
        .expect("the ready-now row reclaims");
    assert_eq!(h.job().id, id2);
    assert!(
        h.retry(Some("boom".to_string()), Some(i64::MAX))
            .await
            .unwrap(),
        "the positive-extreme delay retry lands"
    );
    let job = normal.get_job(id2).await.unwrap().unwrap();
    assert_eq!(job.run_at, i64::MAX, "sat(now + i64::MAX) = i64::MAX");

    // Retention arms: `Some(i64::MIN)` sweeps every dead row
    // (sat(now − MIN) = i64::MAX, the sweep-all edge — pre-guard this
    // sweep failed sticky with the overflow error);
    // `Some(i64::MAX)` deletes nothing (sat(now − MAX) = i64::MIN).
    let sweepall = fx
        .store
        .queue(
            "sweepall",
            QueueOpts {
                dead_letter_retention_s: Some(i64::MIN),
                ..opts()
            },
        )
        .await
        .unwrap();
    let swept_id = sweepall
        .enqueue(json!({"k": 3}), EnqueueOpts::default())
        .await
        .unwrap();
    let h = sweepall.claim_one("w1").await.unwrap().unwrap();
    assert!(h.fail(None).await.unwrap(), "the dead row exists");
    assert!(
        sweepall.sweep_expired().await.unwrap() >= 1,
        "the retention i64::MIN sweep lands (no overflow failure)"
    );
    assert!(
        sweepall.get_job(swept_id).await.unwrap().is_none(),
        "sat(now − i64::MIN) = i64::MAX sweeps the dead row"
    );

    let sweepnone = fx
        .store
        .queue(
            "sweepnone",
            QueueOpts {
                dead_letter_retention_s: Some(i64::MAX),
                ..opts()
            },
        )
        .await
        .unwrap();
    let kept_id = sweepnone
        .enqueue(json!({"k": 4}), EnqueueOpts::default())
        .await
        .unwrap();
    let h = sweepnone.claim_one("w1").await.unwrap().unwrap();
    assert!(h.fail(None).await.unwrap());
    assert_eq!(
        sweepnone.sweep_expired().await.unwrap(),
        0,
        "sat(now − i64::MAX) = i64::MIN deletes nothing"
    );
    assert!(
        sweepnone.get_job(kept_id).await.unwrap().is_some(),
        "the retention i64::MAX dead row survives"
    );

    drop_schema(&fx.admin, &fx.schema).await;
}