agentplane/store/

postgres_cases.rs

//! The case layer on `PostgreSQL`.
//!
//! Every store here exists to settle one race, and the races are the reason this
//! file is not a mechanical translation of the `SQLite` one:
//!
//! | Store | Race | How Postgres settles it |
//! |---|---|---|
//! | cases | two messages, one new matter | partial unique index on open keys |
//! | events | one message, two waiters | `UPDATE … RETURNING` claims in one statement |
//! | timers | one wake-up, two sweeps | same |
//! | tasks | one decision, two reviewers | same |
//! | batches | one item, two reservations | `ON CONFLICT DO NOTHING`, then read back |
//!
//! `UPDATE … RETURNING` is the reason several of these are *simpler* here than
//! in `SQLite` rather than merely different. The read-then-write that `SQLite` has
//! to wrap in a transaction becomes a single statement whose result tells you
//! whether you won — there is no window to reason about because there is no
//! second statement.
//!
//! Whether that reasoning is right is not left to the reader:
//! `testkit::conformance_case` runs the same battery against this and against
//! `SQLite`.

use async_trait::async_trait;
use deadpool_postgres::Pool;
use tokio_postgres::error::SqlState;

use crate::batch::{BatchCensus, BatchStore, ItemOutcome, ItemRecord};
use crate::case::{BufferedEvent, ClaimError, TargetedDelivery};
use crate::case::{CaseCensus, CaseStore, Correlation, EventStore, TaskStore, TimerStore};
use crate::core::{
    BatchId, BreachNote, Case, CaseId, CaseStatus, CaseVersion, CorrelationKey, DeadLetter,
    Deadline, DeadlineState, Digest, EffectKey, InboundEvent, LegalHold, OnExpiry, Priority, RunId,
    Spend, StoreError, Subscription, Task, TaskId, TaskState, Timer, Timestamp,
};

use super::postgres::{PostgresStore, amount_of, be, pool_err, sql_amount};

/// A hold's stored attribution, or corruption.
///
/// Fails closed for the reason every decoder in this crate does: a hold this
/// build cannot read is a preservation order a retention pass would walk past,
/// and from the outside that is indistinguishable from one that was released.
fn hold_row(reason: String, actor: &str, basis: &str) -> Result<super::HoldRow, StoreError> {
    Ok(super::HoldRow {
        reason,
        by: super::decode_operator(actor, basis, "case_legal_holds")?,
    })
}

pub(super) const CASE_SCHEMA: &str = "
-- Every table here leads with the tenant, for the reason the journal schema
-- gives: a key component turns a forgotten predicate into an empty result
-- instead of another tenant's row. The foreign keys carry it too, so a child
-- row cannot reference a parent in a different tenant.
CREATE TABLE IF NOT EXISTS cases (
    tenant    TEXT   NOT NULL,
    case_id   TEXT   NOT NULL,
    kind      TEXT   NOT NULL,
    status    TEXT   NOT NULL,
    state     TEXT   NOT NULL,
    -- Bumped by every state write, which must name the version it read. This
    -- backend is the one that exists for several plane instances sharing a
    -- store, so the overlapping read-modify-write is not a corner case here —
    -- it is the normal operating condition.
    version   BIGINT NOT NULL DEFAULT 0 CHECK (version >= 0),
    opened_at BIGINT NOT NULL,
    PRIMARY KEY (tenant, case_id)
);

-- The worklist read: cases by status, newest first.
CREATE INDEX IF NOT EXISTS cases_by_status
    ON cases (tenant, status, opened_at DESC);

CREATE TABLE IF NOT EXISTS case_correlation (
    tenant    TEXT    NOT NULL,
    case_id   TEXT    NOT NULL,
    namespace TEXT    NOT NULL,
    value     TEXT    NOT NULL,
    open      BOOLEAN NOT NULL DEFAULT TRUE,
    PRIMARY KEY (tenant, case_id, namespace, value),
    FOREIGN KEY (tenant, case_id) REFERENCES cases (tenant, case_id) ON DELETE CASCADE
);

-- One open case per business key. This is the arbiter, not a hint: two inbound
-- messages racing to open the same matter both attempt the insert, and exactly
-- one succeeds. The loser re-reads and attaches.
-- Scoped to the tenant, because a correlation key is a *business* value:
-- `document`/`DOC-1` means something different to every tenant, and two of them
-- using it is ordinary rather than a collision. Globally unique, one tenant's
-- run would attach to another's case and the two would share a history, a
-- deadline set and an erasure unit.
CREATE UNIQUE INDEX IF NOT EXISTS case_correlation_open
    ON case_correlation (tenant, namespace, value) WHERE open;

CREATE TABLE IF NOT EXISTS case_runs (
    tenant  TEXT   NOT NULL,
    case_id TEXT   NOT NULL,
    run_id  TEXT   NOT NULL,
    seq     BIGINT NOT NULL CHECK (seq >= 0),
    PRIMARY KEY (tenant, case_id, run_id),
    FOREIGN KEY (tenant, case_id) REFERENCES cases (tenant, case_id) ON DELETE CASCADE
);

-- One run per position. `attach_run` allocates seq as MAX+1, and two instances
-- attaching different runs concurrently would both compute the same next
-- position; the constraint makes the collision an error the store retries
-- rather than two runs silently sharing a place in the case's order.
CREATE UNIQUE INDEX IF NOT EXISTS case_runs_order
    ON case_runs (tenant, case_id, seq);

-- The blobs a case produced. The case is what an erasure request names, and a
-- digest cannot be reversed to find its case, so the link has to be recorded
-- when the bytes are written or it cannot be recovered at all.
CREATE TABLE IF NOT EXISTS case_blobs (
    tenant     TEXT   NOT NULL,
    case_id    TEXT   NOT NULL,
    digest     BYTEA  NOT NULL,
    written_at BIGINT NOT NULL,
    PRIMARY KEY (tenant, case_id, digest),
    FOREIGN KEY (tenant, case_id) REFERENCES cases (tenant, case_id) ON DELETE CASCADE
);

CREATE INDEX IF NOT EXISTS case_blobs_time
    ON case_blobs (tenant, case_id, written_at);

-- `by_actor` and `by_basis` are on the same footing as the reason: a
-- preservation order the runtime cannot check is worth exactly the name beside
-- it, and the basis says whether an authenticator produced that name or
-- somebody holding the connection string typed it.
CREATE TABLE IF NOT EXISTS case_legal_holds (
    tenant    TEXT   NOT NULL,
    case_id   TEXT   NOT NULL,
    placed_at BIGINT NOT NULL,
    reason    TEXT   NOT NULL,
    by_actor  TEXT   NOT NULL,
    by_basis  TEXT   NOT NULL,
    PRIMARY KEY (tenant, case_id),
    FOREIGN KEY (tenant, case_id) REFERENCES cases (tenant, case_id) ON DELETE CASCADE
);

-- The listing, oldest hold first. Ascending unlike every other backlog index
-- here, because `release_hold` empties this one and the longest-standing
-- unlifted hold is the entry somebody has to ask about.
CREATE INDEX IF NOT EXISTS case_legal_holds_by_time
    ON case_legal_holds (tenant, placed_at, case_id);

-- The last recovery rehearsal, one row per tenant that the latest write
-- replaces. A history of drills is a different artifact and a larger promise
-- than the question an audit asks: *when did you last rehearse, and did it
-- pass*. The columns are counts, an instant and the checkpoint the plane was
-- serving — no case is referenced, so nothing cascades and a drill's verdict
-- outlives the matters it walked, which is the point.
CREATE TABLE IF NOT EXISTS case_last_drill (
    tenant      TEXT    NOT NULL,
    ran_at      BIGINT  NOT NULL,
    sound       BOOLEAN NOT NULL,
    cases       BIGINT  NOT NULL,
    findings    BIGINT  NOT NULL,
    -- Kept apart from `sound`: a pass over nothing is not a pass.
    not_checked BIGINT  NOT NULL,
    origin      TEXT    NOT NULL,
    log_size    BIGINT  NOT NULL,
    PRIMARY KEY (tenant)
);

CREATE TABLE IF NOT EXISTS case_deadlines (
    tenant          TEXT   NOT NULL,
    case_id         TEXT   NOT NULL,
    name            TEXT   NOT NULL,
    resolved_at     BIGINT NOT NULL,
    calendar_digest BYTEA  NOT NULL,
    warn_at         BIGINT,
    state           TEXT   NOT NULL,
    -- Who accounted for the breach, once somebody has. Absent on every other
    -- state; `acknowledged_at` is what says the other two mean anything.
    acknowledged_at    BIGINT,
    acknowledged_by    TEXT,
    acknowledged_basis TEXT,
    acknowledged_note  TEXT,
    PRIMARY KEY (tenant, case_id, name),
    FOREIGN KEY (tenant, case_id) REFERENCES cases (tenant, case_id) ON DELETE CASCADE
);
-- A breach applied whose account the sweep has not yet written: set by the
-- breach in its own statement, cleared once the sweep's notes land.
ALTER TABLE case_deadlines ADD COLUMN IF NOT EXISTS breach_unnoted BOOLEAN NOT NULL DEFAULT FALSE;
CREATE INDEX IF NOT EXISTS case_deadlines_unnoted
    ON case_deadlines (tenant, resolved_at)
    WHERE breach_unnoted;

-- The sweep's read: outstanding obligations by due instant.
CREATE INDEX IF NOT EXISTS case_deadlines_due
    ON case_deadlines (tenant, state, resolved_at);

-- The obligation listing and its gauge: breaches nobody has accounted for,
-- longest-overdue first. Partial, because an acknowledged breach is never
-- selected and a deployment's whole history of them should not be paged
-- through to answer what is still open.
CREATE INDEX IF NOT EXISTS case_deadlines_unaccounted
    ON case_deadlines (tenant, resolved_at)
    WHERE state = 'breached' AND acknowledged_at IS NULL;

CREATE TABLE IF NOT EXISTS inbound_events (
    -- The dedup identity is (source, id), CloudEvents' uniqueness pair. Keying
    -- on id alone deduplicates two producers into each other: id is unique
    -- within a producer, and the collision is silent because the second message
    -- looks exactly like a retry of the first.
    tenant      TEXT    NOT NULL,
    event_id    TEXT    NOT NULL,
    source      TEXT    NOT NULL,
    -- The producer's own id, stored rather than split back out of the key: a
    -- reconstructed event must be the event that arrived, and parsing the key
    -- apart would be a second place that has to agree about the separator.
    bare_id     TEXT    NOT NULL,
    kind        TEXT    NOT NULL,
    payload     TEXT    NOT NULL,
    -- The operator this plane minted the message for, where it minted one.
    -- Two columns rather than a rendered string, for the reason every other
    -- operator column here is two: a name and what established it are two
    -- facts, and `alice (asserted)` is a sentence a reader has to parse back.
    -- Null for everything that arrived over a wire.
    by_actor    TEXT,
    by_basis    TEXT,
    received_at BIGINT  NOT NULL,
    claimed_by  TEXT,
    claimed_at  BIGINT,
    -- The wait a standing claim was made for, until that wait's unsubscribe
    -- consumes it. A wait recovers only its own standing claim, and retiring a
    -- wait sheds only what it consumed.
    claimed_for TEXT,
    -- Delivered to one run by name: that run's alone, dead-lettered rather
    -- than offered to another run when its addressee concludes unconsumed.
    targeted    BOOLEAN NOT NULL DEFAULT FALSE,
    dead        BOOLEAN NOT NULL DEFAULT FALSE,
    dead_reason TEXT,
    PRIMARY KEY (tenant, event_id)
);
-- Set by `erase_payload` and never cleared: an erased row is never handed to
-- a run as a value, whatever its claim says.
ALTER TABLE inbound_events ADD COLUMN IF NOT EXISTS erased BOOLEAN NOT NULL DEFAULT FALSE;

-- The sweep's read: live unclaimed events by age. Partial, so the index is
-- exactly the sweep's candidate set rather than the whole buffer.
CREATE INDEX IF NOT EXISTS inbound_events_live
    ON inbound_events (tenant, received_at)
    WHERE claimed_by IS NULL AND NOT dead;

-- The dead-letter view: retired events, newest first.
CREATE INDEX IF NOT EXISTS inbound_events_dead
    ON inbound_events (tenant, received_at DESC)
    WHERE dead;

-- The unsubscribe strip's read: the rows a run holds claimed, whose delivered
-- payloads are shed once the run's wait is satisfied and journaled.
CREATE INDEX IF NOT EXISTS inbound_events_claimed
    ON inbound_events (tenant, claimed_by)
    WHERE claimed_by IS NOT NULL;

CREATE TABLE IF NOT EXISTS inbound_correlation (
    tenant    TEXT NOT NULL,
    event_id  TEXT NOT NULL,
    namespace TEXT NOT NULL,
    value     TEXT NOT NULL,
    PRIMARY KEY (tenant, event_id, namespace, value),
    FOREIGN KEY (tenant, event_id)
        REFERENCES inbound_events (tenant, event_id) ON DELETE CASCADE
);

-- The claim path's join: which buffered events carry this business key.
CREATE INDEX IF NOT EXISTS inbound_correlation_by_key
    ON inbound_correlation (tenant, namespace, value);

-- The match path: an arriving event finds the runs waiting for it here. The
-- tenant leads for the same reason it leads `case_correlation_open`, and this
-- is the worse of the two failures — one tenant's message resuming another
-- tenant's run hands it a payload it was never sent.
CREATE TABLE IF NOT EXISTS subscriptions (
    tenant     TEXT   NOT NULL,
    run_id     TEXT   NOT NULL,
    effect_key TEXT   NOT NULL,
    case_id    TEXT,
    step       BIGINT NOT NULL,
    phase      TEXT   NOT NULL,
    event_kind TEXT   NOT NULL,
    namespace  TEXT   NOT NULL,
    value      TEXT   NOT NULL,
    created_at BIGINT NOT NULL,
    -- Set while the wait holds a claimed event nothing has delivered yet: the
    -- redelivery pass reads these rows rather than every registered wait.
    parked_at  BIGINT,
    PRIMARY KEY (tenant, run_id, effect_key, namespace, value)
);
-- The one event source this wait accepts; NULL accepts any.
ALTER TABLE subscriptions ADD COLUMN IF NOT EXISTS from_source TEXT;

-- The match path's read: who is waiting on this kind and key. The primary key
-- leads with the run for targeted delivery; this serves the arriving event,
-- which knows only what it carries.
CREATE INDEX IF NOT EXISTS subscriptions_by_key
    ON subscriptions (tenant, event_kind, namespace, value);

-- The redelivery sweep's read: the oldest waits, bounded. Registration order is
-- the index's own order, so a tick walks `limit` rows instead of sorting every
-- subscription the tenant has — which is the shape the embedded backend already
-- had in `SUBS_BY_TIME` and this one did not, so the two agreed on the answer
-- and disagreed on the cost, on a path that runs every tick against a
-- population a plane is *expected* to accumulate.
CREATE INDEX IF NOT EXISTS subscriptions_waiting
    ON subscriptions (tenant, created_at, run_id, effect_key);

-- The redelivery pass's read: waits parked with a claimed event. Partial, so a
-- plane's long legitimate waits never stand in front of the pairs it is for.
CREATE INDEX IF NOT EXISTS subscriptions_parked
    ON subscriptions (tenant, parked_at, run_id, effect_key)
    WHERE parked_at IS NOT NULL;

CREATE TABLE IF NOT EXISTS timers (
    tenant     TEXT   NOT NULL,
    run_id     TEXT   NOT NULL,
    effect_key TEXT   NOT NULL,
    case_id    TEXT,
    step       BIGINT NOT NULL,
    phase      TEXT   NOT NULL,
    fire_at    BIGINT NOT NULL CHECK (fire_at >= 0),
    claimed_at BIGINT CHECK (claimed_at >= 0),
    PRIMARY KEY (tenant, run_id, effect_key)
);
-- The due sweep's read, soonest first: a tick walks the due rows rather than
-- every timer the tenant holds.
CREATE INDEX IF NOT EXISTS timers_due ON timers (tenant, fire_at);

CREATE TABLE IF NOT EXISTS tasks (
    tenant          TEXT   NOT NULL,
    task_id         TEXT   NOT NULL,
    run_id          TEXT   NOT NULL,
    case_id         TEXT,
    kind            TEXT   NOT NULL,
    justification   TEXT   NOT NULL,
    -- Arrays, never a delimited string. Role and actor names are the four-eyes
    -- control's operands and this store does not get to constrain their
    -- alphabet: an exclusion list stored as 'a,b'-joined text reads an actor
    -- named 'a,b' back as two actors named neither, and the person barred from
    -- deciding is barred no longer.
    candidate_roles TEXT[] NOT NULL,
    escalate_to     TEXT[] NOT NULL,
    excluded_actors TEXT[] NOT NULL,
    assignee        TEXT,
    priority        TEXT   NOT NULL,
    state           TEXT   NOT NULL,
    on_expiry       TEXT   NOT NULL,
    -- `Priority::rank`, stored rather than expressed in SQL. Queue order is one
    -- rule; a `CASE priority WHEN …` here would be a second copy of the
    -- priority vocabulary, in a dialect the compiler cannot check, whose
    -- fallback arm sorts an unranked priority last — in the one index whose
    -- whole job is order.
    priority_rank   SMALLINT NOT NULL,
    created_at      BIGINT NOT NULL,
    due_at          BIGINT,
    -- `Withheld::as_str` when the proposal is sealed at rest; the decorator
    -- that sealed it writes it, so no argument value can spell it.
    withheld        TEXT,
    PRIMARY KEY (tenant, task_id)
);

-- The queue's read: pending work, most urgent first and oldest within a rank.
CREATE INDEX IF NOT EXISTS tasks_queue_rank
    ON tasks (tenant, state, priority_rank, created_at);

-- The overdue sweep's read. Partial: most tasks have no window to close.
CREATE INDEX IF NOT EXISTS tasks_due
    ON tasks (tenant, due_at)
    WHERE due_at IS NOT NULL;

CREATE TABLE IF NOT EXISTS batches (
    tenant      TEXT    NOT NULL,
    batch_id    TEXT    NOT NULL,
    plan_digest TEXT    NOT NULL,
    exhausted   BOOLEAN NOT NULL DEFAULT FALSE,
    PRIMARY KEY (tenant, batch_id)
);

CREATE TABLE IF NOT EXISTS batch_items (
    tenant   TEXT   NOT NULL,
    batch_id TEXT   NOT NULL,
    -- Byte order, as the embedded backend compares keys: the resume cursor is
    -- a `<` over these, and a locale collation would order a source's pages
    -- differently from the source.
    item_key TEXT   COLLATE \"C\" NOT NULL,
    run_id   TEXT   NOT NULL,
    outcome  TEXT,
    detail   TEXT,
    -- Spend cannot be negative; the docs say so, and a constraint is the only
    -- form of cannot a shared store honours.
    tokens   BIGINT NOT NULL DEFAULT 0 CHECK (tokens >= 0),
    minor    BIGINT NOT NULL DEFAULT 0 CHECK (minor >= 0),
    PRIMARY KEY (tenant, batch_id, item_key),
    FOREIGN KEY (tenant, batch_id) REFERENCES batches (tenant, batch_id) ON DELETE CASCADE
);

-- The runs a tenant currently has executing.
--
-- A set rather than a counter, and the difference is recovery: a counter that is
-- incremented on admission leaks a slot every time a process dies before the
-- decrement, and nothing can say which increments were real. A set names its
-- members, so a stranded slot is attributable to a run an operator can look up,
-- and releasing it is idempotent by construction.
CREATE TABLE IF NOT EXISTS quota_running (
    tenant      TEXT   NOT NULL,
    run_id      TEXT   NOT NULL,
    admitted_at BIGINT NOT NULL,
    PRIMARY KEY (tenant, run_id)
);

-- The emergency stop: one row per standing halt, none for the rest.
--
-- In the database rather than in a process, because a switch that stops only
-- the instance it was thrown on is not a switch — it is the in-process-counter
-- failure arriving during an incident.
--
-- `scope` is part of the key, not a column beside it: a halt on one agent and a
-- halt on the whole tenant are two rows rather than one flag the last writer
-- wins. An incident that widens and then partly resolves is the ordinary shape,
-- and a single overwritable flag gets it wrong in the direction that lets work
-- through. The scope spellings are the ones `HaltScope::parse` accepts, and
-- they are not restated here: a grammar written twice is a grammar that loses a
-- form the day one is added.
--
-- `by_actor` and `by_basis` are the whole evidentiary weight of the row. The
-- runtime cannot check an emergency stop — there is no verdict to re-derive —
-- so who asked, and what established that name, is the record. The basis is
-- kept beside the name rather than inferred from the surface, because the same
-- act arrives from an authenticated API caller and from somebody holding the
-- database URL, and a reader years later cannot tell those apart from a name.
CREATE TABLE IF NOT EXISTS quota_halted (
    tenant      TEXT        NOT NULL,
    scope       TEXT        NOT NULL,
    reason      TEXT        NOT NULL,
    by_actor    TEXT        NOT NULL,
    by_basis    TEXT        NOT NULL,
    thrown_at   BIGINT      NOT NULL,
    PRIMARY KEY (tenant, scope)
);

-- The manifest registry: one row per published name and version.
--
-- The primary key is the immutability rule as a database constraint rather
-- than as application logic, for the reason exactly-once is a unique index
-- here: application logic can be bypassed by the next caller and a constraint
-- cannot. What the transaction around it adds is the *decision* — a version
-- holding different content is a refusal, not an overwrite.
--
-- `key_id` and `signature` are empty for a publish that happened before
-- anybody signed. That is an operational fact rather than a defect, and it is
-- not a sentinel overlapping a real value: a signature cannot exist without a
-- key id.
CREATE TABLE IF NOT EXISTS registry_manifests (
    tenant      TEXT   NOT NULL,
    name        TEXT   NOT NULL,
    version     TEXT   NOT NULL,
    digest      TEXT   NOT NULL,
    yaml        TEXT   NOT NULL,
    key_id      TEXT   NOT NULL DEFAULT '',
    signature   TEXT   NOT NULL DEFAULT '',
    PRIMARY KEY (tenant, name, version)
);

CREATE TABLE IF NOT EXISTS quota_spent (
    tenant      TEXT   NOT NULL,
    period      TEXT   NOT NULL,
    tokens      BIGINT NOT NULL DEFAULT 0 CHECK (tokens >= 0),
    minor_units BIGINT NOT NULL DEFAULT 0 CHECK (minor_units >= 0),
    PRIMARY KEY (tenant, period)
);

-- One exact receipt per live execution pass. The payload is retained so a
-- retry with changed accounting is damage, not a second interpretation of the
-- same key.
CREATE TABLE IF NOT EXISTS quota_settled (
    tenant       TEXT    NOT NULL,
    run_id       TEXT    NOT NULL,
    epoch        BIGINT  NOT NULL CHECK (epoch >= 0),
    period       TEXT,
    tokens       BIGINT  NOT NULL CHECK (tokens >= 0),
    minor_units  BIGINT  NOT NULL CHECK (minor_units >= 0),
    release_slot BOOLEAN NOT NULL,
    concludes    BOOLEAN NOT NULL,
    PRIMARY KEY (tenant, run_id, epoch)
);

-- What an open run still holds against a period: written beside the slot at
-- admission, reduced by every pass settlement and removed by the concluding
-- one, each in the transaction that writes the receipt. One row per run, so a
-- resume in a later period moves it rather than adding a second.
CREATE TABLE IF NOT EXISTS quota_reserved (
    tenant      TEXT   NOT NULL,
    run_id      TEXT   NOT NULL,
    period      TEXT   NOT NULL,
    tokens      BIGINT NOT NULL CHECK (tokens >= 0),
    minor_units BIGINT NOT NULL CHECK (minor_units >= 0),
    PRIMARY KEY (tenant, run_id)
);

-- One row per dispatch of a rate-ceilinged tool. The key is the idempotency:
-- a retry and a recovered re-dispatch carry the same run and first-attempt
-- key and find their own row. The count is the rows for (tenant, grant_ref)
-- inside the window, decided under a per-grant advisory lock. Rows age out
-- with the widest window and are never refunded early.
CREATE TABLE IF NOT EXISTS quota_rate (
    tenant      TEXT   NOT NULL,
    grant_ref   TEXT   NOT NULL,
    run_id      TEXT   NOT NULL,
    dispatch    TEXT   NOT NULL,
    reserved_at BIGINT NOT NULL,
    PRIMARY KEY (tenant, grant_ref, run_id, dispatch)
);
";

fn corrupt(what: &str, e: impl std::fmt::Display) -> StoreError {
    StoreError::Corrupt {
        seq: 0,
        detail: format!("{what}: {e}"),
    }
}

/// Read a stored column through the type's own vocabulary.
///
/// Every decoder below goes through the core type's `parse` rather than a match
/// of its own — see [`CaseStatus::parse`] for why nothing else may spell a
/// stored vocabulary, and why a reader that defaults is worse than one that
/// refuses. The whole point of `phase` is telling a step's forward pass from
/// its compensating one, so a silent `Forward` hands the unwind logic a
/// compensating record wearing the wrong half of the saga; `Deny` and `Normal`
/// are safe values, and that is precisely what makes them the wrong answer — a
/// fail-closed default is still a fact this store invented about a row it could
/// not read. Every call site already returns `StoreError`, so refusing costs
/// nothing but the `?`.
fn decoded<T>(what: &str, raw: &str, parsed: Option<T>) -> Result<T, StoreError> {
    parsed.ok_or_else(|| corrupt(&format!("unknown {what}"), raw))
}

fn status_from(s: &str) -> Result<CaseStatus, StoreError> {
    decoded("case status", s, CaseStatus::parse(s))
}

fn deadline_state_from(s: &str) -> Result<DeadlineState, StoreError> {
    decoded("deadline state", s, DeadlineState::parse(s))
}

fn task_state_from(s: &str) -> Result<TaskState, StoreError> {
    decoded("task state", s, TaskState::parse(s))
}

fn priority_from(s: &str) -> Result<Priority, StoreError> {
    decoded("task priority", s, Priority::parse(s))
}

fn expiry_from(s: &str) -> Result<OnExpiry, StoreError> {
    decoded("expiry policy", s, OnExpiry::parse(s))
}

fn phase_from(s: &str) -> Result<crate::core::Phase, StoreError> {
    decoded("step phase", s, crate::core::Phase::parse(s))
}

/// The stored spellings of every state a typed predicate admits.
///
/// A membership test in SQL takes its list from the rule rather than repeating
/// it. A literal `IN ('open','claimed','escalated')` is the vocabulary spelled
/// again in a dialect no compiler checks, and it keeps working when a variant
/// is added — by leaving the new state out of the worklist, the backlog count
/// and the expiry sweep at once, which is a task that exists and is in no index.
fn task_states(admits: fn(TaskState) -> bool) -> Vec<&'static str> {
    TaskState::ALL
        .iter()
        .copied()
        .filter(|s| admits(*s))
        .map(TaskState::as_str)
        .collect()
}

/// The obligation half of [`task_states`].
fn deadline_states(admits: fn(DeadlineState) -> bool) -> Vec<&'static str> {
    DeadlineState::ALL
        .iter()
        .copied()
        .filter(|s| admits(*s))
        .map(DeadlineState::as_str)
        .collect()
}

impl PostgresStore {
    fn pool(&self) -> &Pool {
        self.pool_ref()
    }
}

#[async_trait]
impl CaseStore for PostgresStore {
    fn tenant(&self) -> &str {
        self.tenant_str()
    }

    async fn correlate(&self, keys: &[CorrelationKey]) -> Result<Option<CaseId>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        for k in keys {
            let row = client
                .query_opt(
                    "SELECT case_id FROM case_correlation
                      WHERE tenant = $3 AND namespace = $1 AND value = $2 AND open",
                    &[&k.namespace, &k.value, &self.tenant_name()],
                )
                .await
                .map_err(|e| be(&e))?;
            if let Some(row) = row {
                let id: String = row.get(0);
                return Ok(Some(
                    CaseId::parse(&id).map_err(|e| corrupt("bad case id", e))?,
                ));
            }
        }
        Ok(None)
    }

    async fn correlate_or_open(
        &self,
        kind: &str,
        keys: &[CorrelationKey],
        at: Timestamp,
    ) -> Result<Correlation, StoreError> {
        // Two attempts, and the second is not a retry loop for flakiness — it is
        // how the constraint arbitrates. Both messages read no open case, both
        // try to insert, and the unique index picks a winner. The loser comes
        // back here and *finds* the winner's row, which is exactly the answer it
        // should have had.
        for attempt in 0..2 {
            if let Some(existing) = self.correlate(keys).await? {
                return Ok(Correlation::Attached(existing));
            }

            let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
            let tx = client.transaction().await.map_err(|e| be(&e))?;
            let id = CaseId::generate();
            tx.execute(
                "INSERT INTO cases (case_id, kind, status, state, opened_at, tenant)
                 VALUES ($1, $2, $3, $4, $5, $6)",
                &[
                    &id.to_string(),
                    &kind.to_owned(),
                    &CaseStatus::Open.as_str(),
                    &"null".to_owned(),
                    &at.unix_timestamp(),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;

            let mut lost = false;
            for k in keys {
                let r = tx
                    .execute(
                        "INSERT INTO case_correlation (case_id, namespace, value, open, tenant)
                         VALUES ($1, $2, $3, TRUE, $4)",
                        &[&id.to_string(), &k.namespace, &k.value, &self.tenant_name()],
                    )
                    .await;
                if let Err(e) = r {
                    if e.code() == Some(&SqlState::UNIQUE_VIOLATION) {
                        lost = true;
                        break;
                    }
                    return Err(be(&e));
                }
            }
            if lost {
                // Someone else opened this matter first. Discard ours entirely —
                // a half-opened case with no keys is a case nothing can reach.
                drop(tx);
                continue;
            }
            tx.commit().await.map_err(|e| be(&e))?;
            let _ = attempt;
            return Ok(Correlation::Opened(id));
        }

        // Two losses means a third party is opening and closing this key in a
        // tight loop; say so rather than looping forever.
        self.correlate(keys)
            .await?
            .map(Correlation::Attached)
            .ok_or_else(|| {
                StoreError::Backend(
                    "could not open or attach a case: the correlation key is being opened \
                     and closed concurrently"
                        .into(),
                )
            })
    }

    async fn case(&self, id: CaseId) -> Result<Option<Case>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let Some(row) = client
            .query_opt(
                "SELECT kind, status, state, opened_at, version FROM cases
                  WHERE case_id = $1 AND tenant = $2",
                &[&id.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?
        else {
            return Ok(None);
        };
        let kind: String = row.get(0);
        let status: String = row.get(1);
        let state: String = row.get(2);
        let opened: i64 = row.get(3);
        let version: i64 = row.get(4);

        let corr = client
            .query(
                "SELECT namespace, value FROM case_correlation
                  WHERE case_id = $1 AND tenant = $2",
                &[&id.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;
        let runs = client
            .query(
                "SELECT run_id FROM case_runs
                  WHERE case_id = $1 AND tenant = $2 ORDER BY seq ASC",
                &[&id.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;

        Ok(Some(Case {
            id,
            kind,
            status: status_from(&status)?,
            correlation: corr
                .iter()
                .map(|r| CorrelationKey::new(r.get::<_, String>(0), r.get::<_, String>(1)))
                .collect(),
            state: serde_json::from_str(&state)?,
            version: CaseVersion(u64::try_from(version).unwrap_or(0)),
            opened_at: Timestamp::from_unix_timestamp(opened)
                .map_err(|e| corrupt("unrepresentable opened_at", e))?,
            runs: runs
                .iter()
                .map(|r| RunId::parse(&r.get::<_, String>(0)))
                .collect::<Result<Vec<_>, _>>()
                .map_err(|e| corrupt("bad run id", e))?,
        }))
    }

    async fn cases(&self, after: Option<CaseId>, limit: usize) -> Result<Vec<Case>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        // Ids first, rows second — through `case()`, so the assembly of a case
        // from its tables exists exactly once and the export cannot drift from
        // the ordinary reader.
        let cursor = after.map(|c| c.to_string()).unwrap_or_default();
        let rows = client
            .query(
                "SELECT case_id FROM cases
                  WHERE tenant = $1 AND case_id > $2
                  ORDER BY case_id ASC
                  LIMIT $3",
                &[
                    &self.tenant_name(),
                    &cursor,
                    &i64::try_from(limit).unwrap_or(i64::MAX),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        drop(client);
        let mut out = Vec::with_capacity(rows.len());
        for row in rows {
            let id: String = row.get(0);
            let id = CaseId::parse(&id).map_err(|e| corrupt("bad case id", e))?;
            if let Some(case) = self.case(id).await? {
                out.push(case);
            }
        }
        Ok(out)
    }

    #[allow(clippy::too_many_lines)]
    async fn import_case(
        &self,
        case: &Case,
        deadlines: &[Deadline],
        blobs: &[Digest],
    ) -> Result<(), StoreError> {
        let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let tx = client.transaction().await.map_err(|e| be(&e))?;
        let key = case.id.to_string();
        let tenant = self.tenant_name();
        let state = serde_json::to_string(&case.state)?;

        // Refused rather than upserted: a restore rebuilds a case layer, it
        // does not merge one, and an `ON CONFLICT DO UPDATE` here would let a
        // second restore silently rewrite a matter.
        let inserted = tx
            .execute(
                "INSERT INTO cases (tenant, case_id, kind, status, state, version, opened_at)
                 VALUES ($1, $2, $3, $4, $5, $6, $7)
                 ON CONFLICT DO NOTHING",
                &[
                    &tenant,
                    &key,
                    &case.kind,
                    &case.status.as_str(),
                    &state,
                    &i64::try_from(case.version.0).unwrap_or(i64::MAX),
                    &case.opened_at.unix_timestamp(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        if inserted == 0 {
            return Err(StoreError::Backend(format!(
                "case {} already exists — a restore rebuilds a case layer, it does not \
                 merge one",
                case.id
            )));
        }

        for k in &case.correlation {
            // The partial unique index over open correlation rows is the
            // arbiter, exactly as it is for `correlate_or_open`: importing an
            // open case whose key another open case claims must fail rather
            // than merge two matters.
            tx.execute(
                "INSERT INTO case_correlation (tenant, case_id, namespace, value, open)
                 VALUES ($1, $2, $3, $4, $5)",
                &[
                    &tenant,
                    &key,
                    &k.namespace,
                    &k.value,
                    &(case.status != CaseStatus::Closed),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        }

        for (at, run) in case.runs.iter().enumerate() {
            tx.execute(
                "INSERT INTO case_runs (tenant, case_id, run_id, seq)
                 VALUES ($1, $2, $3, $4)",
                &[
                    &tenant,
                    &key,
                    &run.to_string(),
                    &i64::try_from(at).unwrap_or(i64::MAX),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        }

        for deadline in deadlines {
            tx.execute(
                "INSERT INTO case_deadlines
                    (tenant, case_id, name, resolved_at, calendar_digest, warn_at, state,
                     acknowledged_at, acknowledged_by, acknowledged_note, acknowledged_basis)
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)",
                &[
                    &tenant,
                    &key,
                    &deadline.name,
                    &deadline.resolved_at.unix_timestamp(),
                    &deadline.calendar_digest.as_bytes().to_vec(),
                    &deadline.warn_at.map(time::OffsetDateTime::unix_timestamp),
                    &deadline.state.as_str(),
                    &deadline
                        .acknowledged
                        .as_ref()
                        .map(|a| a.at.unix_timestamp()),
                    &deadline
                        .acknowledged
                        .as_ref()
                        .map(|a| a.by.actor().to_owned()),
                    &deadline.acknowledged.as_ref().map(|a| a.note.clone()),
                    &deadline
                        .acknowledged
                        .as_ref()
                        .map(|a| a.by.basis().as_str()),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        }

        // Blob links, stamped from `opened_at` plus an ordinal: the export
        // carries digests without their link timestamps, the key needs
        // uniqueness and erasure needs reachability, and neither needs the
        // original instant.
        for (at, digest) in blobs.iter().enumerate() {
            tx.execute(
                "INSERT INTO case_blobs (tenant, case_id, digest, written_at)
                 VALUES ($1, $2, $3, $4)",
                &[
                    &tenant,
                    &key,
                    &digest.as_bytes().to_vec(),
                    &(case.opened_at.unix_timestamp() + i64::try_from(at).unwrap_or(i64::MAX)),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        }

        tx.commit().await.map_err(|e| be(&e))?;
        Ok(())
    }

    async fn attach_run(&self, case: CaseId, run: RunId) -> Result<(), StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        // Zero-based, matching `import_case`'s enumeration — one rule for what
        // position the first run holds, whichever door it came through.
        //
        // The MAX+1 subquery is not atomic with the insert: two instances
        // attaching different runs both compute the same next position, and
        // the `case_runs_order` unique index turns that from two runs silently
        // sharing a place in the case's history into a violation this loop
        // retries. The `ON CONFLICT` arm still absorbs only the *idempotence*
        // conflict — the same run attached twice — because it names the
        // primary key's columns; a seq collision surfaces as the error the
        // retry exists for. Bounded, because unbounded politeness under a
        // pathological writer is a hang: past the bound the collision is
        // reported rather than spun on.
        for _ in 0..8 {
            let result = client
                .execute(
                    "INSERT INTO case_runs (case_id, run_id, seq, tenant)
                     VALUES ($1, $2,
                             (SELECT COALESCE(MAX(seq) + 1, 0) FROM case_runs
                               WHERE case_id = $1 AND tenant = $3),
                             $3)
                     ON CONFLICT (tenant, case_id, run_id) DO NOTHING",
                    &[&case.to_string(), &run.to_string(), &self.tenant_name()],
                )
                .await;
            match result {
                Ok(_) => return Ok(()),
                Err(e) if e.code() == Some(&SqlState::UNIQUE_VIOLATION) => {}
                Err(e) => return Err(be(&e)),
            }
        }
        Err(StoreError::Backend(format!(
            "could not attach run {run} to case {case}: the run-order position was \
             claimed concurrently on every attempt"
        )))
    }

    async fn link_blob(
        &self,
        case: CaseId,
        digest: Digest,
        at: Timestamp,
    ) -> Result<(), StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        // The primary key makes re-linking the same bytes the same record:
        // two runs on one case writing identical content land on one digest by
        // construction, and that is one artifact.
        client
            .execute(
                "INSERT INTO case_blobs (case_id, digest, written_at, tenant)
                 VALUES ($1, $2, $3, $4)
                 ON CONFLICT (tenant, case_id, digest) DO NOTHING",
                &[
                    &case.to_string(),
                    &digest.as_bytes().as_slice(),
                    &at.unix_timestamp(),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        Ok(())
    }

    async fn blobs_of(&self, case: CaseId) -> Result<Vec<Digest>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let rows = client
            .query(
                "SELECT digest FROM case_blobs
                  WHERE case_id = $1 AND tenant = $2 ORDER BY written_at, digest",
                &[&case.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;
        let mut out = Vec::with_capacity(rows.len());
        for row in rows {
            let raw: Vec<u8> = row.get(0);
            let bytes: [u8; 32] = raw.try_into().map_err(|_| StoreError::Corrupt {
                seq: 0,
                detail: "a linked blob digest is not 32 bytes".into(),
            })?;
            out.push(Digest::from_bytes(bytes));
        }
        Ok(out)
    }

    async fn put_state(
        &self,
        case: CaseId,
        expected: CaseVersion,
        state: serde_json::Value,
    ) -> Result<CaseVersion, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let next = expected.next();
        // The row count is read, and that is the point. The previous version of
        // this method discarded it and returned `Ok(())` for a case that does
        // not exist — the same defect already found once in `release` on this
        // backend. A guard whose result nobody reads is not a guard.
        let n = client
            .execute(
                "UPDATE cases SET state = $2, version = $3
                  WHERE case_id = $1 AND version = $4 AND tenant = $5",
                &[
                    &case.to_string(),
                    &state.to_string(),
                    &i64::try_from(next.0).unwrap_or(i64::MAX),
                    &i64::try_from(expected.0).unwrap_or(i64::MAX),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        if n == 1 {
            return Ok(next);
        }
        // Tell "gone" apart from "moved on": a missing case reported as a
        // conflict sends the caller into a re-read loop against nothing.
        let current = client
            .query_opt(
                "SELECT version FROM cases WHERE case_id = $1 AND tenant = $2",
                &[&case.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;
        match current {
            Some(row) => Err(StoreError::CaseConflict {
                case: case.to_string(),
                expected: expected.0,
                current: u64::try_from(row.get::<_, i64>(0)).unwrap_or(0),
            }),
            None => Err(StoreError::NotFound(case.to_string())),
        }
    }

    async fn set_status(&self, case: CaseId, status: CaseStatus) -> Result<(), StoreError> {
        // Closing releases the correlation keys and refuses an open obligation;
        // an ordinary status write does neither. Keeping `set_status(Closed)` a
        // bare column write let the `status` column and correlation-open
        // membership disagree — a closed case stayed correlatable and a new
        // matter attached to it. Route it through `close`.
        if status == CaseStatus::Closed {
            return self.close(case).await;
        }
        let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let tx = client.transaction().await.map_err(|e| be(&e))?;
        // Read the prior status under the row lock, so *which transition this is*
        // is decided against the row this transaction is about to write.
        let was: Option<String> = tx
            .query_opt(
                "SELECT status FROM cases WHERE case_id = $1 AND tenant = $2 FOR UPDATE",
                &[&case.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?
            .map(|r| r.get(0));
        let Some(was) = was else {
            return Err(StoreError::NotFound(case.to_string()));
        };
        tx.execute(
            "UPDATE cases SET status = $2 WHERE case_id = $1 AND tenant = $3",
            &[&case.to_string(), &status.as_str(), &self.tenant_name()],
        )
        .await
        .map_err(|e| be(&e))?;

        // Leaving `Closed` takes back the correlation keys closure released, or
        // the matter comes back live-looking and unreachable. `open = FALSE`
        // rows another case has since claimed are left alone: the identifier
        // belongs to whichever matter is open for it now.
        if was == CaseStatus::Closed.as_str() {
            tx.execute(
                "UPDATE case_correlation SET open = TRUE
                  WHERE case_id = $1 AND tenant = $2
                    AND NOT EXISTS (
                          SELECT 1 FROM case_correlation other
                           WHERE other.tenant = case_correlation.tenant
                             AND other.namespace = case_correlation.namespace
                             AND other.value = case_correlation.value
                             AND other.open)",
                &[&case.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;
        }
        tx.commit().await.map_err(|e| be(&e))?;
        Ok(())
    }

    async fn close(&self, case: CaseId) -> Result<(), StoreError> {
        let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let tx = client.transaction().await.map_err(|e| be(&e))?;

        // The case row's lock, taken **before** the count and held to commit.
        // Without it this is a check and `register_deadline` is a write that
        // walks past it: at READ COMMITTED each statement reads its own
        // snapshot, so a registration in flight is invisible here and this
        // closure is invisible there, and both commit — leaving a case closed
        // and owing. `register_deadline` takes the same lock, which is what
        // makes the two decide one at a time.
        let locked = tx
            .query_opt(
                "SELECT status FROM cases WHERE case_id = $1 AND tenant = $2 FOR UPDATE",
                &[&case.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;
        if locked.is_none() {
            return Err(StoreError::NotFound(case.to_string()));
        }

        // An unmet obligation survives closure, because closure is the moment
        // people stop looking.
        let open: i64 = tx
            .query_one(
                "SELECT COUNT(*) FROM case_deadlines
                  WHERE case_id = $1 AND tenant = $2 AND state = ANY($3::text[])",
                &[
                    &case.to_string(),
                    &self.tenant_name(),
                    &deadline_states(DeadlineState::is_open),
                ],
            )
            .await
            .map_err(|e| be(&e))?
            .get(0);
        if open > 0 {
            return Err(StoreError::ObligationsOutstanding {
                case: case.to_string(),
                outstanding: usize::try_from(open).unwrap_or(usize::MAX),
            });
        }

        // The row count is read: closing a case that does not exist reported
        // success, while the redb backend answers `NotFound` — and the caller
        // of a silent close believes an audited matter was settled.
        let n = tx
            .execute(
                "UPDATE cases SET status = $2 WHERE case_id = $1 AND tenant = $3",
                &[
                    &case.to_string(),
                    &CaseStatus::Closed.as_str(),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        if n == 0 {
            return Err(StoreError::NotFound(case.to_string()));
        }
        // Releasing the keys is what lets a later message open a *new* matter
        // rather than reanimating an audited one.
        tx.execute(
            "UPDATE case_correlation SET open = FALSE WHERE case_id = $1 AND tenant = $2",
            &[&case.to_string(), &self.tenant_name()],
        )
        .await
        .map_err(|e| be(&e))?;
        tx.commit().await.map_err(|e| be(&e))?;
        Ok(())
    }

    async fn register_deadline(&self, d: &Deadline) -> Result<(), StoreError> {
        let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let tx = client.transaction().await.map_err(|e| be(&e))?;

        // The same row lock `close` takes, for the same reason: these two are
        // the pair that has to be serialized, and the anomaly they would
        // otherwise produce — both committing on each other's pre-state — is a
        // case closed with an obligation it will go on to miss.
        let status: Option<String> = tx
            .query_opt(
                "SELECT status FROM cases WHERE case_id = $1 AND tenant = $2 FOR UPDATE",
                &[&d.case.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?
            .map(|r| r.get(0));
        let Some(status) = status else {
            return Err(StoreError::NotFound(d.case.to_string()));
        };
        if status == CaseStatus::Closed.as_str() {
            return Err(StoreError::CaseClosed {
                case: d.case.to_string(),
            });
        }

        tx.execute(
            "INSERT INTO case_deadlines
                   (case_id, name, resolved_at, calendar_digest, warn_at, state, tenant,
                    acknowledged_at, acknowledged_by, acknowledged_note, acknowledged_basis)
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
                 ON CONFLICT (tenant, case_id, name) DO NOTHING",
            &[
                &d.case.to_string(),
                &d.name,
                &d.resolved_at.unix_timestamp(),
                &d.calendar_digest.as_bytes().to_vec(),
                &d.warn_at.map(Timestamp::unix_timestamp),
                &d.state.as_str(),
                &self.tenant_name(),
                &d.acknowledged.as_ref().map(|a| a.at.unix_timestamp()),
                &d.acknowledged.as_ref().map(|a| a.by.actor().to_owned()),
                &d.acknowledged.as_ref().map(|a| a.note.clone()),
                &d.acknowledged.as_ref().map(|a| a.by.basis().as_str()),
            ],
        )
        .await
        .map_err(|e| be(&e))?;
        tx.commit().await.map_err(|e| be(&e))?;
        Ok(())
    }

    async fn deadlines(&self, case: CaseId) -> Result<Vec<Deadline>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let rows = client
            .query(
                "SELECT case_id, name, resolved_at, calendar_digest, warn_at, state,
                        acknowledged_at, acknowledged_by, acknowledged_note, acknowledged_basis
                   FROM case_deadlines
                  WHERE case_id = $1 AND tenant = $2 ORDER BY resolved_at ASC",
                &[&case.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;
        rows.iter().map(deadline_from).collect()
    }

    async fn set_deadline_state(
        &self,
        case: CaseId,
        name: &str,
        state: DeadlineState,
    ) -> Result<(), StoreError> {
        let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let tx = client.transaction().await.map_err(|e| be(&e))?;

        // How an obligation ended is not editable, and the row's lock is what
        // decides that against the row this transaction writes rather than
        // against one a concurrent sweep has already moved.
        let current: Option<String> = tx
            .query_opt(
                "SELECT state FROM case_deadlines
                  WHERE case_id = $1 AND name = $2 AND tenant = $3 FOR UPDATE",
                &[&case.to_string(), &name.to_owned(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?
            .map(|r| r.get(0));
        if let Some(from) = current {
            let parsed = deadline_state_from(&from)?;
            if !parsed.may_become(state) {
                return Err(StoreError::DeadlineFinal {
                    case: case.to_string(),
                    obligation: name.to_owned(),
                    from,
                    to: state.as_str().to_owned(),
                });
            }
        }

        // The row count is read: a transition for an obligation that does not
        // exist reported success, while the redb backend answers `NotFound` —
        // and the sweep that believes it breached a deadline nobody registered
        // has written its decision into nothing.
        let n = tx
            .execute(
                "UPDATE case_deadlines SET state = $3
                  WHERE case_id = $1 AND name = $2 AND tenant = $4",
                &[
                    &case.to_string(),
                    &name.to_owned(),
                    &state.as_str(),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        if n == 0 {
            return Err(StoreError::NotFound(format!("{case}/{name}")));
        }
        tx.commit().await.map_err(|e| be(&e))?;
        Ok(())
    }

    async fn breach_deadline(
        &self,
        case: CaseId,
        name: &str,
        now: Timestamp,
    ) -> Result<bool, StoreError> {
        let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let tx = client.transaction().await.map_err(|e| be(&e))?;
        let (key, tenant) = (case.to_string(), self.tenant_name());
        // The case row's lock first, as `close` takes it: closure and this
        // breach then decide one at a time, and a closure that won has already
        // met or cancelled the obligation this checks.
        let status: Option<String> = tx
            .query_opt(
                "SELECT status FROM cases WHERE case_id = $1 AND tenant = $2 FOR UPDATE",
                &[&key, &tenant],
            )
            .await
            .map_err(|e| be(&e))?
            .map(|r| r.get(0));
        let Some(status) = status else {
            return Err(StoreError::NotFound(key));
        };
        let breached = tx
            .execute(
                "UPDATE case_deadlines SET state = $4, breach_unnoted = TRUE
                  WHERE case_id = $1 AND name = $2 AND tenant = $3
                    AND state = ANY($5::text[]) AND resolved_at <= $6",
                &[
                    &key,
                    &name.to_owned(),
                    &tenant,
                    &DeadlineState::Breached.as_str(),
                    &deadline_states(DeadlineState::is_open),
                    &now.unix_timestamp(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        if breached == 0 {
            let exists = tx
                .query_opt(
                    "SELECT 1 FROM case_deadlines
                      WHERE case_id = $1 AND name = $2 AND tenant = $3",
                    &[&key, &name.to_owned(), &tenant],
                )
                .await
                .map_err(|e| be(&e))?
                .is_some();
            tx.commit().await.map_err(|e| be(&e))?;
            return if exists {
                Ok(false)
            } else {
                Err(StoreError::NotFound(format!("{case}/{name}")))
            };
        }
        if status != CaseStatus::Closed.as_str() {
            tx.execute(
                "UPDATE cases SET status = $3 WHERE case_id = $1 AND tenant = $2",
                &[&key, &tenant, &CaseStatus::Escalated.as_str()],
            )
            .await
            .map_err(|e| be(&e))?;
        }
        tx.commit().await.map_err(|e| be(&e))?;
        Ok(true)
    }

    async fn due(&self, now: Timestamp, limit: usize) -> Result<Vec<Deadline>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let rows = client
            .query(
                "SELECT case_id, name, resolved_at, calendar_digest, warn_at, state,
                        acknowledged_at, acknowledged_by, acknowledged_note, acknowledged_basis
                   FROM case_deadlines
                  WHERE tenant = $3 AND state = ANY($4::text[])
                    AND (resolved_at <= $1 OR (warn_at IS NOT NULL AND warn_at <= $1))
                  ORDER BY resolved_at ASC LIMIT $2",
                &[
                    &now.unix_timestamp(),
                    &i64::try_from(limit).unwrap_or(i64::MAX),
                    &self.tenant_name(),
                    &deadline_states(DeadlineState::is_open),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        rows.iter().map(deadline_from).collect()
    }

    async fn breaches_to_note(&self, limit: usize) -> Result<Vec<Deadline>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let rows = client
            .query(
                "SELECT case_id, name, resolved_at, calendar_digest, warn_at, state,
                        acknowledged_at, acknowledged_by, acknowledged_note, acknowledged_basis
                   FROM case_deadlines
                  WHERE tenant = $2 AND breach_unnoted
                  ORDER BY resolved_at ASC LIMIT $1",
                &[
                    &i64::try_from(limit).unwrap_or(i64::MAX),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        rows.iter().map(deadline_from).collect()
    }

    async fn mark_breach_noted(&self, case: CaseId, name: &str) -> Result<(), StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let n = client
            .execute(
                "UPDATE case_deadlines SET breach_unnoted = FALSE
                  WHERE tenant = $1 AND case_id = $2 AND name = $3",
                &[&self.tenant_name(), &case.to_string(), &name.to_owned()],
            )
            .await
            .map_err(|e| be(&e))?;
        if n == 0 {
            return Err(StoreError::NotFound(format!("{case}/{name}")));
        }
        Ok(())
    }

    async fn breached(&self, limit: usize) -> Result<Vec<Deadline>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        // Served by `case_deadlines_due`, whose leading columns are the two this
        // filters on. No clock: a breach is a recorded state, not a comparison
        // against now.
        let rows = client
            .query(
                "SELECT case_id, name, resolved_at, calendar_digest, warn_at, state,
                        acknowledged_at, acknowledged_by, acknowledged_note, acknowledged_basis
                   FROM case_deadlines
                  WHERE tenant = $2 AND state = 'breached' AND acknowledged_at IS NULL
                  ORDER BY resolved_at ASC LIMIT $1",
                &[
                    &i64::try_from(limit).unwrap_or(i64::MAX),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        rows.iter().map(deadline_from).collect()
    }

    async fn acknowledge_breach(
        &self,
        case: CaseId,
        name: &str,
        note: &BreachNote,
    ) -> Result<bool, StoreError> {
        let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let tx = client.transaction().await.map_err(|e| be(&e))?;
        let row = tx
            .query_opt(
                "SELECT state, acknowledged_at FROM case_deadlines
                  WHERE case_id = $1 AND name = $2 AND tenant = $3 FOR UPDATE",
                &[&case.to_string(), &name.to_owned(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;
        let Some(row) = row else {
            return Err(StoreError::NotFound(format!("{case}/{name}")));
        };
        let state: String = row.get(0);
        if state != DeadlineState::Breached.as_str() {
            return Err(StoreError::NotBreached {
                case: case.to_string(),
                obligation: name.to_owned(),
                state,
            });
        }
        // The first account stands: a retry must not rewrite who looked or when.
        if row.get::<_, Option<i64>>(1).is_some() {
            return Ok(false);
        }
        tx.execute(
            "UPDATE case_deadlines
                SET acknowledged_at = $4, acknowledged_by = $5, acknowledged_note = $6,
                    acknowledged_basis = $7
              WHERE case_id = $1 AND name = $2 AND tenant = $3",
            &[
                &case.to_string(),
                &name.to_owned(),
                &self.tenant_name(),
                &note.at.unix_timestamp(),
                &note.by.actor(),
                &note.note,
                &note.by.basis().as_str(),
            ],
        )
        .await
        .map_err(|e| be(&e))?;
        tx.commit().await.map_err(|e| be(&e))?;
        Ok(true)
    }

    async fn place_hold(&self, case: CaseId, hold: &LegalHold) -> Result<bool, StoreError> {
        let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let tx = client.transaction().await.map_err(|e| be(&e))?;
        // The case row's lock, for the reason `register_deadline` takes it: two
        // snapshots each reading the other's pre-state is how a hold lands on a
        // case that an erasure in the other transaction is already destroying.
        let exists = tx
            .query_opt(
                "SELECT 1 FROM cases WHERE tenant = $1 AND case_id = $2 FOR UPDATE",
                &[&self.tenant_name(), &case.to_string()],
            )
            .await
            .map_err(|e| be(&e))?;
        if exists.is_none() {
            return Err(StoreError::NotFound(case.to_string()));
        }
        // First placement wins: `DO NOTHING` rather than `DO UPDATE`, so a retry
        // cannot move the instant or rewrite the reason.
        let placed = tx
            .execute(
                "INSERT INTO case_legal_holds
                      (tenant, case_id, placed_at, reason, by_actor, by_basis)
                      VALUES ($1, $2, $3, $4, $5, $6)
                 ON CONFLICT (tenant, case_id) DO NOTHING",
                &[
                    &self.tenant_name(),
                    &case.to_string(),
                    &hold.placed_at.unix_timestamp(),
                    &hold.reason,
                    &hold.by.actor(),
                    &hold.by.basis().as_str(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        tx.commit().await.map_err(|e| be(&e))?;
        Ok(placed == 1)
    }

    async fn release_hold(&self, case: CaseId) -> Result<bool, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let exists = client
            .query_opt(
                "SELECT 1 FROM cases WHERE tenant = $1 AND case_id = $2",
                &[&self.tenant_name(), &case.to_string()],
            )
            .await
            .map_err(|e| be(&e))?;
        if exists.is_none() {
            return Err(StoreError::NotFound(case.to_string()));
        }
        let removed = client
            .execute(
                "DELETE FROM case_legal_holds WHERE tenant = $1 AND case_id = $2",
                &[&self.tenant_name(), &case.to_string()],
            )
            .await
            .map_err(|e| be(&e))?;
        Ok(removed == 1)
    }

    async fn release_hold_if(
        &self,
        case: CaseId,
        standing: &LegalHold,
    ) -> Result<bool, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let exists = client
            .query_opt(
                "SELECT 1 FROM cases WHERE tenant = $1 AND case_id = $2",
                &[&self.tenant_name(), &case.to_string()],
            )
            .await
            .map_err(|e| be(&e))?;
        if exists.is_none() {
            return Err(StoreError::NotFound(case.to_string()));
        }
        // The compare is the statement's own predicate, so a hold placed
        // after the caller read the old one fails it and stays.
        let removed = client
            .execute(
                "DELETE FROM case_legal_holds
                  WHERE tenant = $1 AND case_id = $2 AND placed_at = $3
                    AND reason = $4 AND by_actor = $5 AND by_basis = $6",
                &[
                    &self.tenant_name(),
                    &case.to_string(),
                    &standing.placed_at.unix_timestamp(),
                    &standing.reason,
                    &standing.by.actor(),
                    &standing.by.basis().as_str(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        Ok(removed == 1)
    }

    async fn hold(&self, case: CaseId) -> Result<Option<LegalHold>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let row = client
            .query_opt(
                "SELECT placed_at, reason, by_actor, by_basis FROM case_legal_holds
                  WHERE tenant = $1 AND case_id = $2",
                &[&self.tenant_name(), &case.to_string()],
            )
            .await
            .map_err(|e| be(&e))?;
        row.map(|r| {
            Ok(super::hold_from_row(
                Timestamp::from_unix_timestamp(r.get::<_, i64>(0))
                    .map_err(|e| corrupt("unrepresentable placed_at", e))?,
                hold_row(r.get(1), &r.get::<_, String>(2), &r.get::<_, String>(3))?,
            ))
        })
        .transpose()
    }

    async fn holds(
        &self,
        after: Option<CaseId>,
        limit: usize,
    ) -> Result<Vec<(CaseId, LegalHold)>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        // `(placed_at, case_id)` as one tuple is what makes the cursor total:
        // several holds can share an instant, and comparing the stamp alone
        // would either repeat them or skip them depending on which side of the
        // boundary the page fell.
        let rows = match after {
            Some(cursor) => client
                .query(
                    "SELECT case_id, placed_at, reason, by_actor, by_basis FROM case_legal_holds
                          WHERE tenant = $1
                            AND (placed_at, case_id) > (
                                SELECT placed_at, case_id FROM case_legal_holds
                                 WHERE tenant = $1 AND case_id = $2)
                          ORDER BY placed_at, case_id
                          LIMIT $3",
                    &[
                        &self.tenant_name(),
                        &cursor.to_string(),
                        &i64::try_from(limit).unwrap_or(i64::MAX),
                    ],
                )
                .await,
            None => client
                .query(
                    "SELECT case_id, placed_at, reason, by_actor, by_basis FROM case_legal_holds
                          WHERE tenant = $1
                          ORDER BY placed_at, case_id
                          LIMIT $2",
                    &[
                        &self.tenant_name(),
                        &i64::try_from(limit).unwrap_or(i64::MAX),
                    ],
                )
                .await,
        }
        .map_err(|e| be(&e))?;

        rows.into_iter()
            .map(|r| {
                Ok((
                    CaseId::parse(&r.get::<_, String>(0)).map_err(|e| corrupt("bad case id", e))?,
                    super::hold_from_row(
                        Timestamp::from_unix_timestamp(r.get::<_, i64>(1))
                            .map_err(|e| corrupt("unrepresentable placed_at", e))?,
                        hold_row(r.get(2), &r.get::<_, String>(3), &r.get::<_, String>(4))?,
                    ),
                ))
            })
            .collect()
    }

    async fn by_status(&self, status: CaseStatus, limit: usize) -> Result<Vec<Case>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let rows = client
            .query(
                "SELECT case_id FROM cases
                  WHERE status = $1 AND tenant = $3 ORDER BY opened_at DESC LIMIT $2",
                &[
                    &status.as_str(),
                    &i64::try_from(limit).unwrap_or(i64::MAX),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        let mut out = Vec::with_capacity(rows.len());
        for row in rows {
            let id: String = row.get(0);
            let id = CaseId::parse(&id).map_err(|e| corrupt("bad case id", e))?;
            if let Some(c) = self.case(id).await? {
                out.push(c);
            }
        }
        Ok(out)
    }

    async fn record_drill(&self, record: &crate::case::DrillRecord) -> Result<(), StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        client
            .execute(
                "INSERT INTO case_last_drill
                   (tenant, ran_at, sound, cases, findings, not_checked, origin, log_size)
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
                 ON CONFLICT (tenant) DO UPDATE SET
                   ran_at = EXCLUDED.ran_at, sound = EXCLUDED.sound,
                   cases = EXCLUDED.cases, findings = EXCLUDED.findings,
                   not_checked = EXCLUDED.not_checked, origin = EXCLUDED.origin,
                   log_size = EXCLUDED.log_size",
                &[
                    &self.tenant_name(),
                    &record.at.unix_timestamp(),
                    &record.sound,
                    &i64::try_from(record.cases).unwrap_or(i64::MAX),
                    &i64::try_from(record.findings).unwrap_or(i64::MAX),
                    &i64::try_from(record.not_checked).unwrap_or(i64::MAX),
                    &record.origin,
                    &i64::try_from(record.size).unwrap_or(i64::MAX),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        Ok(())
    }

    async fn last_drill(&self) -> Result<Option<crate::case::DrillRecord>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let row = client
            .query_opt(
                "SELECT ran_at, sound, cases, findings, not_checked, origin, log_size
                   FROM case_last_drill WHERE tenant = $1",
                &[&self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;
        row.map(|r| {
            Ok(crate::case::DrillRecord {
                at: Timestamp::from_unix_timestamp(r.get::<_, i64>(0))
                    .map_err(|e| corrupt("unrepresentable drill instant", e))?,
                sound: r.get(1),
                cases: u64::try_from(r.get::<_, i64>(2)).unwrap_or(0),
                findings: u64::try_from(r.get::<_, i64>(3)).unwrap_or(0),
                not_checked: u64::try_from(r.get::<_, i64>(4)).unwrap_or(0),
                origin: r.get(5),
                size: u64::try_from(r.get::<_, i64>(6)).unwrap_or(0),
            })
        })
        .transpose()
    }

    async fn census(&self, now: Timestamp) -> Result<CaseCensus, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let row = client
            .query_one(
                "SELECT COUNT(*), MIN(opened_at) FROM cases
                  WHERE tenant = $1 AND status <> $2::text",
                &[&self.tenant_name(), &CaseStatus::Closed.as_str()],
            )
            .await
            .map_err(|e| be(&e))?;
        let open: i64 = row.get(0);
        let oldest: Option<i64> = row.get(1);
        let due: i64 = client
            .query_one(
                "SELECT COUNT(*) FROM case_deadlines
                  WHERE tenant = $2 AND state = ANY($3::text[]) AND resolved_at <= $1",
                &[
                    &now.unix_timestamp(),
                    &self.tenant_name(),
                    &deadline_states(DeadlineState::is_open),
                ],
            )
            .await
            .map_err(|e| be(&e))?
            .get(0);

        let breached: i64 = client
            .query_one(
                "SELECT COUNT(*) FROM case_deadlines
                  WHERE tenant = $1 AND state = 'breached' AND acknowledged_at IS NULL",
                &[&self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?
            .get(0);

        Ok(CaseCensus {
            open: u64::try_from(open).unwrap_or(0),
            oldest_age_secs: oldest.map(|o| {
                crate::runtime::metrics::age_secs(
                    Timestamp::from_unix_timestamp(o).unwrap_or(now),
                    now,
                )
            }),
            due: u64::try_from(due).unwrap_or(0),
            breached: u64::try_from(breached).unwrap_or(0),
        })
    }
}

fn deadline_from(row: &tokio_postgres::Row) -> Result<Deadline, StoreError> {
    let case: String = row.get(0);
    let digest: Vec<u8> = row.get(3);
    let warn: Option<i64> = row.get(4);
    let state: String = row.get(5);
    let arr: [u8; 32] = digest
        .try_into()
        .map_err(|_| corrupt("calendar digest", "not 32 bytes"))?;
    Ok(Deadline {
        case: CaseId::parse(&case).map_err(|e| corrupt("bad case id", e))?,
        name: row.get(1),
        resolved_at: Timestamp::from_unix_timestamp(row.get::<_, i64>(2))
            .map_err(|e| corrupt("unrepresentable deadline", e))?,
        calendar_digest: Digest::from_bytes(arr),
        warn_at: warn
            .map(Timestamp::from_unix_timestamp)
            .transpose()
            .map_err(|e| corrupt("unrepresentable warn_at", e))?,
        state: deadline_state_from(&state)?,
        acknowledged: acknowledged_from(row)?,
    })
}

/// The account's three columns, read as one optional fact.
///
/// Fallible rather than defaulting: a row whose `acknowledged_at` is set and
/// whose `acknowledged_by` is not is a corrupt row, and reading it as *nobody
/// has looked* would put a breach back on a listing somebody had already
/// answered — or, the other way, hide one nobody had.
fn acknowledged_from(row: &tokio_postgres::Row) -> Result<Option<BreachNote>, StoreError> {
    let at: Option<i64> = row.get(6);
    let by: Option<String> = row.get(7);
    let note: Option<String> = row.get(8);
    let basis: Option<String> = row.get(9);
    match (at, by, note, basis) {
        (None, None, None, None) => Ok(None),
        (Some(at), Some(by), Some(note), Some(basis)) => Ok(Some(BreachNote {
            by: super::decode_operator(&by, &basis, "case_deadlines")?,
            note,
            at: Timestamp::from_unix_timestamp(at)
                .map_err(|e| corrupt("unrepresentable acknowledged_at", e))?,
        })),
        _ => Err(corrupt(
            "breach acknowledgement",
            "only some of its columns are set",
        )),
    }
}

#[async_trait]
#[allow(clippy::too_many_lines)]
impl EventStore for PostgresStore {
    fn tenant(&self) -> &str {
        self.tenant_str()
    }

    async fn buffer(&self, event: &InboundEvent, at: Timestamp) -> Result<bool, StoreError> {
        let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let tx = client.transaction().await.map_err(|e| be(&e))?;
        let inserted = tx
            .execute(
                "INSERT INTO inbound_events
                   (event_id, source, bare_id, kind, payload, received_at, tenant,
                    by_actor, by_basis)
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
                 ON CONFLICT (tenant, event_id) DO NOTHING",
                &[
                    &event.dedup_key(),
                    &event.source,
                    &event.id,
                    &event.kind,
                    &event.payload.to_string(),
                    &at.unix_timestamp(),
                    &self.tenant_name(),
                    &event.by.as_ref().map(|o| o.actor().to_owned()),
                    &event.by.as_ref().map(|o| o.basis().as_str().to_owned()),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        if inserted == 0 {
            tx.commit().await.map_err(|e| be(&e))?;
            return Ok(false);
        }
        for k in &event.correlation {
            tx.execute(
                "INSERT INTO inbound_correlation (event_id, namespace, value, tenant)
                 VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING",
                // The dedup key throughout: correlation rows reference the row
                // `buffer` wrote, and that is keyed by `(source, id)`.
                &[
                    &event.dedup_key(),
                    &k.namespace,
                    &k.value,
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        }
        tx.commit().await.map_err(|e| be(&e))?;
        Ok(true)
    }

    async fn subscribe(&self, sub: &Subscription, at: Timestamp) -> Result<(), StoreError> {
        // One transaction for the whole wait. A subscription with several
        // correlation keys is one wait, and the per-statement autocommit this
        // replaces could fail halfway — a wait reachable by some of its keys
        // and not others, which matches or misses depending on which key the
        // event happens to carry. The redb backend writes them in one
        // transaction; so does this.
        let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let tx = client.transaction().await.map_err(|e| be(&e))?;
        for k in &sub.correlation {
            tx.execute(
                "INSERT INTO subscriptions
                       (run_id, effect_key, case_id, step, phase, event_kind,
                        namespace, value, created_at, tenant, from_source)
                     VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
                     ON CONFLICT DO NOTHING",
                &[
                    &sub.run.to_string(),
                    &sub.effect.to_hex(),
                    &sub.case.map(|c| c.to_string()),
                    &i64::from(sub.step.0),
                    &crate::core::Phase::as_str(sub.phase),
                    &sub.kind,
                    &k.namespace,
                    &k.value,
                    &at.unix_timestamp(),
                    &self.tenant_name(),
                    &sub.from,
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        }
        tx.commit().await.map_err(|e| be(&e))?;
        Ok(())
    }

    async fn claim_for(
        &self,
        sub: &Subscription,
        at: Timestamp,
    ) -> Result<Option<BufferedEvent>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        for k in &sub.correlation {
            // One statement. The claim predicate, the write and the wait's
            // parking are evaluated together, so two waiters cannot both come
            // away with the row and the wait takes no second message.
            //
            // Unclaimed — or claimed **for this very wait** and not yet
            // consumed. The second arm is crash recovery: `match_waiter`
            // claims durably and the run resumes in a separate step, so a
            // crash between the two leaves an event claimed for a wait that
            // never saw it. Scoped to the wait and to a standing claim, so a
            // message the run already journaled is never handed to its next
            // wait on the same key.
            let row = client
                .query_opt(
                    "WITH claimed AS (
                       UPDATE inbound_events SET claimed_by = $1, claimed_at = $2,
                                                 claimed_for = $8
                        WHERE tenant = $6 AND event_id = (
                            SELECT e.event_id FROM inbound_events e
                              JOIN inbound_correlation c
                                ON c.tenant = e.tenant AND c.event_id = e.event_id
                             WHERE e.tenant = $6 AND e.kind = $3
                               AND c.namespace = $4 AND c.value = $5
                               AND ($7::TEXT IS NULL OR e.source = $7)
                               AND ((e.claimed_by IS NULL
                                     AND NOT EXISTS (
                                         SELECT 1 FROM subscriptions p
                                          WHERE p.tenant = $6 AND p.run_id = $1
                                            AND p.effect_key = $8
                                            AND p.parked_at IS NOT NULL))
                                    OR (e.claimed_by = $1 AND e.claimed_for = $8))
                               AND NOT e.dead AND NOT e.erased
                             ORDER BY e.received_at ASC
                             FOR UPDATE SKIP LOCKED
                             LIMIT 1)
                    RETURNING bare_id, kind, payload, received_at, source, event_id,
                              by_actor, by_basis),
                     parked AS (
                       UPDATE subscriptions SET parked_at = $2
                        WHERE tenant = $6 AND run_id = $1 AND effect_key = $8
                          AND EXISTS (SELECT 1 FROM claimed))
                   SELECT * FROM claimed",
                    &[
                        &sub.run.to_string(),
                        &at.unix_timestamp(),
                        &sub.kind,
                        &k.namespace,
                        &k.value,
                        &self.tenant_name(),
                        &sub.from,
                        &sub.effect.to_hex(),
                    ],
                )
                .await
                .map_err(|e| be(&e))?;
            if let Some(row) = row {
                return Ok(Some(
                    buffered_from(&row, &client, &self.tenant_name()).await?,
                ));
            }
        }
        Ok(None)
    }

    async fn match_waiter(
        &self,
        event: &InboundEvent,
        at: Timestamp,
    ) -> Result<Option<Subscription>, StoreError> {
        let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let tx = client.transaction().await.map_err(|e| be(&e))?;

        for k in &event.correlation {
            // `FOR UPDATE` on the subscription row, for the race `deliver_to`
            // already locks against: two *different* events for one wait. The
            // claim below guards the event row, so one event never resumed two
            // runs — but nothing serialised two events electing one waiter.
            // Both selected it, both claimed their own (unclaimed) event, and
            // the loser's event was left claimed for a run whose wait the
            // winner had already satisfied: parked forever, because the sweep
            // only retires rows with no claim. Locking the row makes the loser
            // wait; the winner's retirement then deletes it, and the loser's
            // re-evaluation finds no waiter and leaves its event live.
            //
            // A parked wait is passed over: it already holds its claimed
            // event and waits only for redelivery, so a second event elected
            // for it would be claimed for a satisfied wait and never consumed.
            let Some(row) = tx
                .query_opt(
                    "SELECT run_id, effect_key, case_id, step, phase, from_source
                       FROM subscriptions
                      WHERE tenant = $4 AND event_kind = $1
                        AND namespace = $2 AND value = $3
                        AND (from_source IS NULL OR from_source = $5)
                        AND parked_at IS NULL
                        AND NOT EXISTS (
                              SELECT 1 FROM run_seal s
                               WHERE s.tenant = subscriptions.tenant
                                 AND s.run_id = subscriptions.run_id)
                      ORDER BY created_at ASC LIMIT 1
                      FOR UPDATE OF subscriptions",
                    &[
                        &event.kind,
                        &k.namespace,
                        &k.value,
                        &self.tenant_name(),
                        &event.source,
                    ],
                )
                .await
                .map_err(|e| be(&e))?
            else {
                continue;
            };
            let run: String = row.get(0);
            let effect: String = row.get(1);

            // Claiming the event in the same transaction is what stops one
            // message resuming two runs.
            let claimed = tx
                .execute(
                    "UPDATE inbound_events SET claimed_by = $2, claimed_at = $3,
                                              claimed_for = $5
                      WHERE tenant = $4 AND event_id = $1
                        AND claimed_by IS NULL AND NOT dead",
                    &[
                        &event.dedup_key(),
                        &run,
                        &at.unix_timestamp(),
                        &self.tenant_name(),
                        &effect,
                    ],
                )
                .await
                .map_err(|e| be(&e))?;
            if claimed == 0 {
                tx.commit().await.map_err(|e| be(&e))?;
                return Ok(None);
            }

            let case: Option<String> = row.get(2);
            let step: i64 = row.get(3);
            let phase: String = row.get(4);
            // The claim parks the subscription, in the same transaction: a
            // parked wait is matched no second event, and a crash before the
            // resume leaves a pair the redelivery pass finds and finishes.
            tx.execute(
                "UPDATE subscriptions SET parked_at = $4
                  WHERE tenant = $1 AND run_id = $2 AND effect_key = $3",
                &[&self.tenant_name(), &run, &effect, &at.unix_timestamp()],
            )
            .await
            .map_err(|e| be(&e))?;
            tx.commit().await.map_err(|e| be(&e))?;

            return Ok(Some(Subscription {
                run: RunId::parse(&run).map_err(|e| corrupt("bad run id", e))?,
                case: case
                    .map(|c| CaseId::parse(&c))
                    .transpose()
                    .map_err(|e| corrupt("bad case id", e))?,
                effect: EffectKey::from_hex(&effect).map_err(|e| corrupt("bad effect key", e))?,
                step: crate::core::StepId(u32::try_from(step).unwrap_or(0)),
                phase: phase_from(&phase)?,
                kind: event.kind.clone(),
                correlation: event.correlation.clone(),
                from: row.get(5),
            }));
        }
        tx.commit().await.map_err(|e| be(&e))?;
        Ok(None)
    }

    async fn deliver_to(
        &self,
        target: RunId,
        event: &InboundEvent,
        at: Timestamp,
    ) -> Result<TargetedDelivery, StoreError> {
        let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let tx = client.transaction().await.map_err(|e| be(&e))?;
        let tenant = self.tenant_name();
        let event_id = event.dedup_key();

        // A message this run already holds claimed resumes the wait it was
        // claimed for, while that claim stands — a retry after a crash between
        // claim and resume. Consumed, or claimed by another run, it is a
        // duplicate. A new message goes to a wait holding no undelivered one.
        let existing_claim: Option<(Option<String>, Option<String>)> = tx
            .query_opt(
                "SELECT claimed_by, claimed_for FROM inbound_events
                  WHERE tenant = $1 AND event_id = $2 FOR UPDATE",
                &[&tenant, &event_id],
            )
            .await
            .map_err(|e| be(&e))?
            .map(|row| (row.get(0), row.get(1)));
        let target_text = target.to_string();
        let claimed_for: Option<String> = match &existing_claim {
            Some((Some(by), Some(effect))) if *by == target_text => Some(effect.clone()),
            _ => None,
        };

        // Lock this run's candidate subscription rows. Two continuations for
        // one task then serialize before either can insert its event.
        //
        // Ordered by effect key, because that is the order the embedded
        // backend walks its subscription table in — its key is
        // `(tenant, run, effect, …)` — and when one run has several waits that
        // all match one event, *which* wait the delivery satisfies is
        // observable: the resumed step, the effect key on the wake record. Two
        // backends electing different waiters is a run that behaves
        // differently depending on which store it journals to. Lowest effect
        // key wins on both.
        let rows = tx
            .query(
                "SELECT effect_key, case_id, step, phase, namespace, value, from_source,
                        parked_at
                   FROM subscriptions
                  WHERE tenant = $1 AND run_id = $2 AND event_kind = $3
                    AND (from_source IS NULL OR from_source = $4)
                  ORDER BY effect_key ASC
                  FOR UPDATE",
                &[&tenant, &target.to_string(), &event.kind, &event.source],
            )
            .await
            .map_err(|e| be(&e))?;
        let Some(selected) = rows.iter().find(|row| {
            let effect: String = row.get(0);
            let namespace: String = row.get(4);
            let value: String = row.get(5);
            let parked: Option<i64> = row.get(7);
            let eligible = match (&existing_claim, &claimed_for) {
                (Some(_), Some(claimed)) => *claimed == effect,
                (Some(_), None) => false,
                (None, _) => parked.is_none(),
            };
            eligible
                && event
                    .correlation
                    .iter()
                    .any(|key| key.namespace == namespace && key.value == value)
        }) else {
            tx.commit().await.map_err(|e| be(&e))?;
            return Ok(if existing_claim.is_some() {
                TargetedDelivery::Duplicate
            } else {
                TargetedDelivery::NotWaiting
            });
        };

        let effect: String = selected.get(0);
        let case: Option<String> = selected.get(1);
        let step: i64 = selected.get(2);
        let phase: String = selected.get(3);
        let subscription = Subscription {
            run: target,
            case: case
                .map(|value| CaseId::parse(&value))
                .transpose()
                .map_err(|e| corrupt("bad case id", e))?,
            effect: EffectKey::from_hex(&effect).map_err(|e| corrupt("bad effect key", e))?,
            step: crate::core::StepId(u32::try_from(step).unwrap_or(0)),
            phase: phase_from(&phase)?,
            kind: event.kind.clone(),
            correlation: rows
                .iter()
                .filter(|row| row.get::<_, String>(0) == effect)
                .map(|row| CorrelationKey::new(row.get::<_, String>(4), row.get::<_, String>(5)))
                .collect(),
            from: selected.get(6),
        };

        if existing_claim.is_some() {
            // Selected only as the wait its standing claim was made for.
            tx.commit().await.map_err(|e| be(&e))?;
            return Ok(TargetedDelivery::Matched(subscription));
        }

        let inserted = tx
            .execute(
                "INSERT INTO inbound_events
                   (event_id, source, bare_id, kind, payload, received_at,
                    claimed_by, claimed_at, tenant, by_actor, by_basis, claimed_for,
                    targeted)
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $6, $8, $9, $10, $11, TRUE)
                 ON CONFLICT (tenant, event_id) DO NOTHING",
                &[
                    &event_id,
                    &event.source,
                    &event.id,
                    &event.kind,
                    &event.payload.to_string(),
                    &at.unix_timestamp(),
                    &target.to_string(),
                    &tenant,
                    &event.by.as_ref().map(|o| o.actor().to_owned()),
                    &event.by.as_ref().map(|o| o.basis().as_str().to_owned()),
                    &effect,
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        if inserted == 0 {
            // Inserted by a concurrent delivery between the read above and
            // this write: the same rule, read again.
            let row = tx
                .query_one(
                    "SELECT claimed_by, claimed_for FROM inbound_events
                      WHERE tenant = $1 AND event_id = $2",
                    &[&tenant, &event_id],
                )
                .await
                .map_err(|e| be(&e))?;
            let (by, wait): (Option<String>, Option<String>) = (row.get(0), row.get(1));
            tx.commit().await.map_err(|e| be(&e))?;
            return Ok(
                if by.as_deref() == Some(target_text.as_str()) && wait.as_deref() == Some(&effect) {
                    TargetedDelivery::Matched(subscription)
                } else {
                    TargetedDelivery::Duplicate
                },
            );
        }
        for key in &event.correlation {
            tx.execute(
                "INSERT INTO inbound_correlation (event_id, namespace, value, tenant)
                 VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING",
                &[&event_id, &key.namespace, &key.value, &tenant],
            )
            .await
            .map_err(|e| be(&e))?;
        }
        // Parked until the run's unsubscribe: a crash between this claim and
        // the resume leaves a pair the redelivery pass finds.
        tx.execute(
            "UPDATE subscriptions SET parked_at = $4
              WHERE tenant = $1 AND run_id = $2 AND effect_key = $3",
            &[&tenant, &target.to_string(), &effect, &at.unix_timestamp()],
        )
        .await
        .map_err(|e| be(&e))?;

        tx.commit().await.map_err(|e| be(&e))?;
        Ok(TargetedDelivery::Matched(subscription))
    }

    async fn unsubscribe(&self, run: RunId, effect: EffectKey) -> Result<(), StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        client
            .execute(
                "DELETE FROM subscriptions
                  WHERE tenant = $3 AND run_id = $1 AND effect_key = $2",
                &[&run.to_string(), &effect.to_hex(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;
        // A wait's unsubscribe is the store's signal that its delivery was
        // journaled, so the buffer's copy of what this wait consumed is shed
        // here — only this wait's: another wait of the run may hold a message
        // not yet journaled. The row keeps its `(source, id)` identity, claim
        // and dead-letter fields: dedup and accounting need those, and only
        // the content was ever the erasure concern. Stripping at the *claim*
        // instead would lose the payload for a run that crashed between claim
        // and resume, whose recovery re-reads it from the buffer.
        client
            .execute(
                "UPDATE inbound_events SET payload = 'null', claimed_for = NULL
                  WHERE tenant = $1 AND claimed_by = $2 AND claimed_for = $3",
                &[&self.tenant_name(), &run.to_string(), &effect.to_hex()],
            )
            .await
            .map_err(|e| be(&e))?;
        Ok(())
    }

    async fn unsubscribe_run(
        &self,
        run: RunId,
        unanswered: &[EffectKey],
    ) -> Result<crate::case::Retired, StoreError> {
        let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let tx = client.transaction().await.map_err(|e| be(&e))?;
        let retired = tx
            .query(
                "DELETE FROM subscriptions WHERE tenant = $1 AND run_id = $2
                 RETURNING effect_key",
                &[&self.tenant_name(), &run.to_string()],
            )
            .await
            .map_err(|e| be(&e))?;
        // A message claimed for a wait the run never answered reached nobody:
        // back to the buffer, unclaimed, with its payload.
        let unanswered: Vec<String> = unanswered.iter().map(|effect| effect.to_hex()).collect();
        let released = tx
            .query(
                "UPDATE inbound_events
                    SET claimed_by = NULL, claimed_at = NULL, claimed_for = NULL
                  WHERE tenant = $1 AND claimed_by = $2 AND claimed_for = ANY($3)
                    AND NOT dead AND NOT targeted
              RETURNING bare_id, kind, payload, received_at, source, event_id,
                        by_actor, by_basis",
                &[&self.tenant_name(), &run.to_string(), &unanswered],
            )
            .await
            .map_err(|e| be(&e))?;
        // One addressed to this run is its alone: dead-lettered, never offered
        // to another run waiting on the same key.
        tx.execute(
            "UPDATE inbound_events
                SET claimed_by = NULL, claimed_at = NULL, claimed_for = NULL,
                    dead = TRUE, dead_reason = $4
              WHERE tenant = $1 AND claimed_by = $2 AND claimed_for = ANY($3)
                AND NOT dead AND targeted",
            &[
                &self.tenant_name(),
                &run.to_string(),
                &unanswered,
                &crate::case::ADDRESSEE_CONCLUDED_REASON,
            ],
        )
        .await
        .map_err(|e| be(&e))?;
        // The same shedding `unsubscribe` does, for every other wait of the run.
        tx.execute(
            "UPDATE inbound_events SET payload = 'null', claimed_for = NULL
              WHERE tenant = $1 AND claimed_by = $2",
            &[&self.tenant_name(), &run.to_string()],
        )
        .await
        .map_err(|e| be(&e))?;
        tx.commit().await.map_err(|e| be(&e))?;
        let waits: std::collections::BTreeSet<String> =
            retired.iter().map(|row| row.get(0)).collect();
        let mut events = Vec::with_capacity(released.len());
        for row in &released {
            events.push(
                buffered_from(row, &client, &self.tenant_name())
                    .await?
                    .event,
            );
        }
        Ok(crate::case::Retired {
            waits: waits.len(),
            released: events,
        })
    }

    async fn park_wait(&self, sub: &Subscription, at: Timestamp) -> Result<(), StoreError> {
        let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let tx = client.transaction().await.map_err(|e| be(&e))?;
        for k in &sub.correlation {
            tx.execute(
                "INSERT INTO subscriptions
                       (run_id, effect_key, case_id, step, phase, event_kind,
                        namespace, value, created_at, tenant, parked_at, from_source)
                     VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $9, $11)
                     ON CONFLICT (tenant, run_id, effect_key, namespace, value)
                     DO UPDATE SET parked_at = EXCLUDED.parked_at",
                &[
                    &sub.run.to_string(),
                    &sub.effect.to_hex(),
                    &sub.case.map(|c| c.to_string()),
                    &i64::from(sub.step.0),
                    &crate::core::Phase::as_str(sub.phase),
                    &sub.kind,
                    &k.namespace,
                    &k.value,
                    &at.unix_timestamp(),
                    &self.tenant_name(),
                    &sub.from,
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        }
        tx.commit().await.map_err(|e| be(&e))?;
        Ok(())
    }

    async fn parked_waits(&self, limit: usize) -> Result<Vec<Subscription>, StoreError> {
        let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        // A sealed run can record no delivery. Its parked pairs are retired
        // here rather than listed, or the redelivery pass tries each, fails,
        // and finds it again every tick — and the payloads it held claimed
        // are shed as a conclusion's retirement sheds them.
        let tx = client.transaction().await.map_err(|e| be(&e))?;
        let retired = tx
            .query(
                "DELETE FROM subscriptions s
                  WHERE s.tenant = $1 AND s.parked_at IS NOT NULL
                    AND EXISTS (SELECT 1 FROM run_seal r
                                 WHERE r.tenant = s.tenant AND r.run_id = s.run_id)
              RETURNING s.run_id",
                &[&self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;
        let sealed: std::collections::BTreeSet<String> =
            retired.iter().map(|row| row.get(0)).collect();
        for run in &sealed {
            tx.execute(
                "UPDATE inbound_events SET payload = 'null', claimed_for = NULL
                  WHERE tenant = $1 AND claimed_by = $2",
                &[&self.tenant_name(), run],
            )
            .await
            .map_err(|e| be(&e))?;
        }
        tx.commit().await.map_err(|e| be(&e))?;
        // One row per wait, its keys gathered: a parked wait with two keys is
        // one pair to redeliver, not two.
        let rows = client
            .query(
                "SELECT run_id, effect_key, MIN(case_id), MIN(step), MIN(phase),
                        MIN(event_kind), array_agg(namespace), array_agg(value),
                        MIN(from_source)
                   FROM subscriptions
                  WHERE tenant = $2 AND parked_at IS NOT NULL
                  GROUP BY run_id, effect_key
                  ORDER BY MIN(parked_at) ASC, run_id, effect_key
                  LIMIT $1",
                &[
                    &i64::try_from(limit).unwrap_or(i64::MAX),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        let mut out = Vec::with_capacity(rows.len());
        for row in rows {
            let run: String = row.get(0);
            let effect: String = row.get(1);
            let case: Option<String> = row.get(2);
            let step: i64 = row.get(3);
            let phase: String = row.get(4);
            let namespaces: Vec<String> = row.get(6);
            let values: Vec<String> = row.get(7);
            out.push(Subscription {
                run: RunId::parse(&run).map_err(|e| corrupt("bad run id", e))?,
                case: case
                    .map(|c| CaseId::parse(&c))
                    .transpose()
                    .map_err(|e| corrupt("bad case id", e))?,
                effect: EffectKey::from_hex(&effect).map_err(|e| corrupt("bad effect key", e))?,
                step: crate::core::StepId(u32::try_from(step).unwrap_or(0)),
                phase: phase_from(&phase)?,
                kind: row.get(5),
                correlation: namespaces
                    .into_iter()
                    .zip(values)
                    .map(|(ns, value)| CorrelationKey::new(ns, value))
                    .collect(),
                from: row.get(8),
            });
        }
        Ok(out)
    }

    async fn erase_payload(&self, source: &str, id: &str) -> Result<bool, StoreError> {
        let mut client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let key = crate::core::origin_key(source, id);
        let tenant = self.tenant_name();
        let tx = client.transaction().await.map_err(|e| be(&e))?;
        let Some(row) = tx
            .query_opt(
                "SELECT claimed_by, claimed_for IS NOT NULL FROM inbound_events
                  WHERE tenant = $1 AND event_id = $2 FOR UPDATE",
                &[&tenant, &key],
            )
            .await
            .map_err(|e| be(&e))?
        else {
            tx.commit().await.map_err(|e| be(&e))?;
            return Ok(false);
        };
        let claimed_by: Option<String> = row.get(0);
        let standing: bool = row.get(1);
        // Undelivered: nobody claimed it, or a wait claimed it and has not
        // yet journaled it — the claim stands until that wait's unsubscribe
        // consumes it. Either way it leaves the claimable set as a dead
        // letter, in this write, so no recovery hands the run the emptied row;
        // the claim is released and the claimant's wait is unparked, so the
        // wait stays open for its deadline to bound. A delivered row keeps its
        // claim and its accounting. Every right-hand side reads the row as it
        // was before the update.
        let undelivered = claimed_by.is_none() || standing;
        tx.execute(
            "UPDATE inbound_events SET payload = 'null', erased = TRUE,
                    dead = dead OR $3,
                    dead_reason = CASE WHEN $3 AND NOT dead THEN $4 ELSE dead_reason END,
                    claimed_by = CASE WHEN $3 THEN NULL ELSE claimed_by END,
                    claimed_at = CASE WHEN $3 THEN NULL ELSE claimed_at END,
                    claimed_for = NULL
              WHERE tenant = $1 AND event_id = $2",
            &[&tenant, &key, &undelivered, &crate::case::ERASED_REASON],
        )
        .await
        .map_err(|e| be(&e))?;
        if let (true, Some(run)) = (undelivered, claimed_by) {
            tx.execute(
                "UPDATE subscriptions s SET parked_at = NULL
                  WHERE s.tenant = $1 AND s.run_id = $2 AND s.parked_at IS NOT NULL
                    AND EXISTS (SELECT 1 FROM inbound_correlation c
                                 WHERE c.tenant = s.tenant AND c.event_id = $3
                                   AND c.namespace = s.namespace AND c.value = s.value)",
                &[&tenant, &run, &key],
            )
            .await
            .map_err(|e| be(&e))?;
        }
        tx.commit().await.map_err(|e| be(&e))?;
        Ok(true)
    }

    async fn minter(
        &self,
        source: &str,
        id: &str,
    ) -> Result<Option<crate::case::Minter>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let row = client
            .query_opt(
                "SELECT by_actor, by_basis FROM inbound_events
                  WHERE tenant = $1 AND event_id = $2",
                &[&self.tenant_name(), &crate::core::origin_key(source, id)],
            )
            .await
            .map_err(|e| be(&e))?;
        let Some(row) = row else {
            return Ok(None);
        };
        match (
            row.get::<_, Option<String>>(0),
            row.get::<_, Option<String>>(1),
        ) {
            (Some(actor), Some(basis)) => Ok(Some(crate::case::Minter::Operator(
                super::decode_operator(&actor, &basis, "inbound_events")?,
            ))),
            (None, None) => Ok(Some(crate::case::Minter::Nobody)),
            _ => Err(StoreError::Corrupt {
                seq: 0,
                detail: "inbound_events holds half an operator: a minted event carries a \
                         name and what established it, or neither"
                    .to_owned(),
            }),
        }
    }

    async fn sweep_unclaimed(
        &self,
        older_than: Timestamp,
        reason: &str,
    ) -> Result<usize, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        // `<=`, not `<`: a zero grace window must retire everything already
        // buffered, and with second-granularity stamps `<` silently spares
        // anything received this second — the same boundary the redb sweep
        // states, pinned for both by the conformance battery.
        let n = client
            .execute(
                "UPDATE inbound_events SET dead = TRUE, dead_reason = $2
                  WHERE tenant = $3 AND claimed_by IS NULL AND NOT dead
                    AND received_at <= $1",
                &[
                    &older_than.unix_timestamp(),
                    &reason.to_owned(),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        Ok(usize::try_from(n).unwrap_or(0))
    }

    async fn dead_letters(&self, limit: usize) -> Result<Vec<DeadLetter>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let rows = client
            .query(
                "SELECT bare_id, kind, payload, received_at, dead_reason, source, event_id,
                        by_actor, by_basis
                   FROM inbound_events WHERE tenant = $2 AND dead
                  ORDER BY received_at DESC LIMIT $1",
                &[
                    &i64::try_from(limit).unwrap_or(i64::MAX),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        let mut out = Vec::with_capacity(rows.len());
        for row in rows {
            let payload: String = row.get(2);
            // The correlation keys are read back, as the redb backend reads
            // them: the first question about an unclaimed message is *what was
            // it correlated by*, and a reconstructed event with no keys is a
            // valid-looking message silently stripped of the field that
            // explains it.
            let event_id: String = row.get(6);
            let corr = client
                .query(
                    "SELECT namespace, value FROM inbound_correlation
                      WHERE tenant = $2 AND event_id = $1",
                    &[&event_id, &self.tenant_name()],
                )
                .await
                .map_err(|e| be(&e))?;
            out.push(DeadLetter {
                event: InboundEvent {
                    source: row.get(5),
                    id: row.get(0),
                    kind: row.get(1),
                    by: match (
                        row.get::<_, Option<String>>(7),
                        row.get::<_, Option<String>>(8),
                    ) {
                        (Some(actor), Some(basis)) => {
                            Some(super::decode_operator(&actor, &basis, "inbound_events")?)
                        }
                        _ => None,
                    },
                    correlation: corr
                        .iter()
                        .map(|r| CorrelationKey::new(r.get::<_, String>(0), r.get::<_, String>(1)))
                        .collect(),
                    payload: serde_json::from_str(&payload)?,
                },
                received_at: Timestamp::from_unix_timestamp(row.get::<_, i64>(3))
                    .map_err(|e| corrupt("unrepresentable received_at", e))?,
                // One wording across backends: a retirement that recorded no
                // reason reads "unclaimed", exactly as the redb backend
                // answers, rather than an empty string here and a word there —
                // an operator scripting on the reason field must not have to
                // know which store the plane journals to.
                reason: row
                    .get::<_, Option<String>>(4)
                    .filter(|reason| !reason.is_empty())
                    .unwrap_or_else(|| "unclaimed".to_owned()),
            });
        }
        Ok(out)
    }

    async fn waiting(&self, limit: usize) -> Result<Vec<Subscription>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let rows = client
            .query(
                "SELECT run_id, effect_key, case_id, step, phase, event_kind, namespace, value,
                        from_source
                   FROM subscriptions WHERE tenant = $2
                  ORDER BY created_at ASC LIMIT $1",
                &[
                    &i64::try_from(limit).unwrap_or(i64::MAX),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        let mut out = Vec::with_capacity(rows.len());
        for row in rows {
            let run: String = row.get(0);
            let effect: String = row.get(1);
            let case: Option<String> = row.get(2);
            let step: i64 = row.get(3);
            let phase: String = row.get(4);
            out.push(Subscription {
                run: RunId::parse(&run).map_err(|e| corrupt("bad run id", e))?,
                case: case
                    .map(|c| CaseId::parse(&c))
                    .transpose()
                    .map_err(|e| corrupt("bad case id", e))?,
                effect: EffectKey::from_hex(&effect).map_err(|e| corrupt("bad effect key", e))?,
                step: crate::core::StepId(u32::try_from(step).unwrap_or(0)),
                phase: phase_from(&phase)?,
                kind: row.get(5),
                correlation: vec![CorrelationKey::new(
                    row.get::<_, String>(6),
                    row.get::<_, String>(7),
                )],
                from: row.get(8),
            });
        }
        Ok(out)
    }
}

async fn buffered_from(
    row: &tokio_postgres::Row,
    client: &deadpool_postgres::Client,
    tenant: &str,
) -> Result<BufferedEvent, StoreError> {
    let id: String = row.get(0);
    let payload: String = row.get(2);
    // Correlation is keyed by the dedup key, which is what `buffer` wrote — not
    // by `bare_id`, the producer's own id. Reading it back under the wrong one
    // returns nothing, and a claimed event would arrive with no keys at all:
    // valid-looking, silently stripped of what it was routed on.
    let event_id: String = row.get(5);
    let corr = client
        .query(
            "SELECT namespace, value FROM inbound_correlation
              WHERE tenant = $2 AND event_id = $1",
            &[&event_id, &tenant],
        )
        .await
        .map_err(|e| be(&e))?;
    Ok(BufferedEvent {
        event: InboundEvent {
            source: row.get(4),
            id: id.clone(),
            kind: row.get(1),
            correlation: corr
                .iter()
                .map(|r| CorrelationKey::new(r.get::<_, String>(0), r.get::<_, String>(1)))
                .collect(),
            payload: serde_json::from_str(&payload)?,
            // Both columns or neither: a name with no basis is an
            // attribution this build cannot state, and inventing one would
            // claim the strong form for a row that never held it.
            by: match (
                row.get::<_, Option<String>>(6),
                row.get::<_, Option<String>>(7),
            ) {
                (Some(actor), Some(basis)) => {
                    Some(super::decode_operator(&actor, &basis, "inbound_events")?)
                }
                (None, None) => None,
                _ => {
                    return Err(StoreError::Corrupt {
                        seq: 0,
                        detail: "inbound_events holds half an operator: a minted event \
                                 carries a name and what established it, or neither"
                            .to_owned(),
                    });
                }
            },
        },
        received_at: Timestamp::from_unix_timestamp(row.get::<_, i64>(3))
            .map_err(|e| corrupt("unrepresentable received_at", e))?,
    })
}

/// How long a timer claim holds before another sweep may take it.
const CLAIM_LEASE: i64 = 60;

#[async_trait]
impl TimerStore for PostgresStore {
    fn tenant(&self) -> &str {
        crate::journal::JournalStore::tenant(self)
    }

    async fn arm(&self, timer: &Timer) -> Result<(), StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        client
            .execute(
                "INSERT INTO timers
                   (run_id, effect_key, case_id, step, phase, fire_at, tenant)
                 VALUES ($1, $2, $3, $4, $5, $6, $7)
                 ON CONFLICT (tenant, run_id, effect_key) DO NOTHING",
                &[
                    &timer.run.to_string(),
                    &timer.effect.to_hex(),
                    &timer.case.map(|c| c.to_string()),
                    &i64::from(timer.step.0),
                    &crate::core::Phase::as_str(timer.phase),
                    &timer.fire_at.unix_timestamp(),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        Ok(())
    }

    async fn claim_due(&self, now: Timestamp, limit: usize) -> Result<Vec<Timer>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        // A sealed run can record no wake. Its due timers are retired rather
        // than claimed, or each is claimed and fails once per lease period for
        // ever.
        client
            .execute(
                "DELETE FROM timers t
                  WHERE t.tenant = $2 AND t.fire_at <= $1
                    AND EXISTS (SELECT 1 FROM run_seal s
                                 WHERE s.tenant = t.tenant AND s.run_id = t.run_id)",
                &[&now.unix_timestamp(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;
        // Claimed and selected in one statement, with `SKIP LOCKED` so a second
        // sweeper takes different rows rather than blocking on the first's.
        let rows = client
            .query(
                "UPDATE timers SET claimed_at = $1
                  WHERE tenant = $4 AND (run_id, effect_key) IN (
                      SELECT run_id, effect_key FROM timers
                       WHERE tenant = $4 AND fire_at <= $1
                         AND (claimed_at IS NULL OR claimed_at <= $2)
                       ORDER BY fire_at ASC
                       FOR UPDATE SKIP LOCKED
                       LIMIT $3)
              RETURNING run_id, effect_key, case_id, step, phase, fire_at",
                &[
                    &now.unix_timestamp(),
                    &(now.unix_timestamp() - CLAIM_LEASE),
                    &i64::try_from(limit).unwrap_or(i64::MAX),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        rows.iter().map(timer_from).collect()
    }

    async fn pending_count(&self) -> Result<u64, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let n: i64 = client
            .query_one(
                "SELECT COUNT(*) FROM timers WHERE tenant = $1",
                &[&self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?
            .get(0);
        Ok(u64::try_from(n).unwrap_or(0))
    }

    async fn disarm(&self, run: RunId, effect: EffectKey) -> Result<(), StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        client
            .execute(
                "DELETE FROM timers WHERE tenant = $3 AND run_id = $1 AND effect_key = $2",
                &[&run.to_string(), &effect.to_hex(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;
        Ok(())
    }

    async fn disarm_run(&self, run: RunId) -> Result<usize, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let n = client
            .execute(
                "DELETE FROM timers WHERE tenant = $2 AND run_id = $1",
                &[&run.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;
        Ok(usize::try_from(n).unwrap_or(usize::MAX))
    }
}

fn timer_from(row: &tokio_postgres::Row) -> Result<Timer, StoreError> {
    let run: String = row.get(0);
    let effect: String = row.get(1);
    let case: Option<String> = row.get(2);
    let step: i64 = row.get(3);
    let phase: String = row.get(4);
    Ok(Timer {
        run: RunId::parse(&run).map_err(|e| corrupt("bad run id", e))?,
        case: case
            .map(|c| CaseId::parse(&c))
            .transpose()
            .map_err(|e| corrupt("bad case id", e))?,
        effect: EffectKey::from_hex(&effect).map_err(|e| corrupt("bad effect key", e))?,
        step: crate::core::StepId(u32::try_from(step).unwrap_or(0)),
        phase: phase_from(&phase)?,
        fire_at: Timestamp::from_unix_timestamp(row.get::<_, i64>(5))
            .map_err(|e| corrupt("unrepresentable fire_at", e))?,
    })
}

/// One task's row, read on a connection the caller already holds.
///
/// This exists because of a deadlock, not for tidiness. `claim` held its
/// pooled connection across a `Self::task` call, and `task` acquired a second
/// connection from the same pool — so every in-flight claim needed two
/// connections while holding one. Sixteen reviewers racing one task on a small
/// pool each held a connection and waited for another that only a waiter could
/// release: a deadlock that reproduces exactly under the concurrency the claim
/// verb exists to survive, and only where the pool is small enough to exhaust
/// — a large development machine passes over the defect a CI runner hangs on,
/// which is how it shipped. Every read made while a connection is held goes
/// through this instead of re-entering the pool.
async fn task_on(
    client: &tokio_postgres::Client,
    tenant: &str,
    id: TaskId,
) -> Result<Option<Task>, StoreError> {
    let row = client
        .query_opt(
            &format!("SELECT {TASK_COLS} FROM tasks WHERE task_id = $1 AND tenant = $2"),
            &[&id.to_hex(), &tenant],
        )
        .await
        .map_err(|e| be(&e))?;
    row.as_ref().map(task_from).transpose()
}

#[async_trait]
impl TaskStore for PostgresStore {
    fn tenant(&self) -> &str {
        self.tenant_str()
    }

    async fn open(&self, task: &Task) -> Result<Task, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        client
            .execute(
                "INSERT INTO tasks (task_id, run_id, case_id, kind, justification,
                                    candidate_roles, escalate_to, excluded_actors, assignee,
                                    priority, state, on_expiry, priority_rank, created_at,
                                    due_at, tenant, withheld)
                 VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17)
                 ON CONFLICT (tenant, task_id) DO NOTHING",
                &[
                    &task.id.to_hex(),
                    &task.run.to_string(),
                    &task.case.map(|c| c.to_string()),
                    &task.kind,
                    &serde_json::to_string(&task.justification)?,
                    &task.candidate_roles,
                    &task.escalate_to,
                    &task.excluded_actors,
                    &task.assignee,
                    &task.priority.as_str(),
                    &task.state.as_str(),
                    &OnExpiry::as_str(task.on_expiry),
                    &i16::from(task.priority.rank()),
                    &task.created_at.unix_timestamp(),
                    &task.due_at.map(Timestamp::unix_timestamp),
                    &self.tenant_name(),
                    &task.withheld.map(crate::core::Withheld::as_str),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        Ok(task_on(&client, &self.tenant_name(), task.id)
            .await?
            .unwrap_or_else(|| task.clone()))
    }

    async fn task(&self, id: TaskId) -> Result<Option<Task>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        task_on(&client, &self.tenant_name(), id).await
    }

    async fn claim(&self, id: TaskId, actor: &str, roles: &[String]) -> Result<Task, ClaimError> {
        // One connection for the whole verb — every read below goes through
        // `task_on` rather than `Self::task`, which would take a second
        // connection while this one is held. See `task_on` for the deadlock
        // that shape produced under exactly the concurrency this verb exists
        // to survive.
        let client = self
            .pool()
            .get()
            .await
            .map_err(|e| ClaimError::Store(pool_err(&e)))?;
        let tenant = self.tenant_name();
        let Some(task) = task_on(&client, &tenant, id)
            .await
            .map_err(ClaimError::Store)?
        else {
            return Err(ClaimError::NotFound(id));
        };
        // Eligibility before availability, and the order is load-bearing — see
        // `TaskStore::claim`.
        //
        // Four eyes: whoever proposed the action does not approve it.
        if task.excluded_actors.iter().any(|a| a == actor) {
            return Err(ClaimError::Excluded {
                actor: actor.to_owned(),
            });
        }
        if !task.candidate_roles.is_empty()
            && !task.candidate_roles.iter().any(|r| roles.contains(r))
        {
            return Err(ClaimError::WrongRole {
                actor: actor.to_owned(),
            });
        }
        if !task.state.is_pending() {
            return Err(ClaimError::NotPending {
                task: id,
                state: task.state,
            });
        }

        // The reservation itself is one statement, guarded on the row still
        // being unheld **and still pending**. Checking above and writing here
        // would leave a window two reviewers both pass through — and the state
        // predicate is not redundant with the eligibility read: a sweep can
        // expire the task between that read and this write, and an UPDATE
        // keyed on the assignee alone would resurrect an expired task into
        // `claimed`, un-deciding the expiry policy that already fired on it.
        // `take_over` and `release` carry the same predicate for the same
        // reason.
        //
        // The second arm is the **same-holder re-claim**, which is idempotent
        // success — the contract on `TaskStore::claim`, and the one the redb
        // backend already honoured while this one refused with
        // `AlreadyClaimed { holder: yourself }`. A claim whose acknowledgement
        // was lost is retried by an honest client, and the retry must converge
        // on "you hold it" rather than bounce off its own success. It does not
        // widen the race: a task claimed by anybody *else* still fails every
        // arm, and an expired task matches neither state predicate.
        let updated = client
            .execute(
                "UPDATE tasks SET assignee = $2, state = $5
                  WHERE task_id = $1 AND tenant = $3
                    AND ((state = ANY($4::text[])
                          AND (assignee IS NULL OR assignee = $2))
                         OR (state = $5 AND assignee = $2))",
                &[
                    &id.to_hex(),
                    &actor.to_owned(),
                    &self.tenant_name(),
                    &task_states(TaskState::is_queued),
                    &TaskState::Claimed.as_str(),
                ],
            )
            .await
            .map_err(|e| ClaimError::Store(be(&e)))?;
        if updated == 0 {
            // Which guard refused: state, or holder. Re-read and say so — a
            // barred claim reported as "held by nobody" sends the reviewer
            // back to a queue that will refuse them again.
            let current = task_on(&client, &tenant, id)
                .await
                .map_err(ClaimError::Store)?;
            return Err(match current {
                None => ClaimError::NotFound(id),
                Some(t) if !t.state.is_pending() => ClaimError::NotPending {
                    task: id,
                    state: t.state,
                },
                Some(t) => ClaimError::AlreadyClaimed {
                    task: id,
                    holder: t.assignee.unwrap_or_default(),
                },
            });
        }
        task_on(&client, &tenant, id)
            .await
            .map_err(ClaimError::Store)?
            .ok_or(ClaimError::NotFound(id))
    }

    async fn take_over(
        &self,
        id: TaskId,
        from: &str,
        actor: &str,
        roles: &[String],
    ) -> Result<Task, ClaimError> {
        // One connection for the whole verb, exactly as `claim` — see `task_on`.
        let client = self
            .pool()
            .get()
            .await
            .map_err(|e| ClaimError::Store(pool_err(&e)))?;
        let tenant = self.tenant_name();
        let Some(task) = task_on(&client, &tenant, id)
            .await
            .map_err(ClaimError::Store)?
        else {
            return Err(ClaimError::NotFound(id));
        };
        // Claim's eligibility-first order, unchanged: a take-over is a claim,
        // and four-eyes exclusion does not thin because the previous reviewer
        // left.
        if task.excluded_actors.iter().any(|a| a == actor) {
            return Err(ClaimError::Excluded {
                actor: actor.to_owned(),
            });
        }
        if !task.candidate_roles.is_empty()
            && !task.candidate_roles.iter().any(|r| roles.contains(r))
        {
            return Err(ClaimError::WrongRole {
                actor: actor.to_owned(),
            });
        }
        if !task.state.is_pending() {
            return Err(ClaimError::NotPending {
                task: id,
                state: task.state,
            });
        }

        // The displacement is one statement, guarded on the holder still being
        // the one the caller named — the compare-and-swap that keeps a
        // take-over decided from a stale view from displacing whoever holds
        // the task *now*.
        let updated = client
            .execute(
                "UPDATE tasks SET assignee = $2, state = 'claimed'
                  WHERE task_id = $1 AND tenant = $4
                    AND assignee = $3 AND state = 'claimed'",
                &[
                    &id.to_hex(),
                    &actor.to_owned(),
                    &from.to_owned(),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| ClaimError::Store(be(&e)))?;
        if updated == 0 {
            return Err(ClaimError::NotHeld {
                task: id,
                actor: from.to_owned(),
            });
        }
        task_on(&client, &tenant, id)
            .await
            .map_err(ClaimError::Store)?
            .ok_or(ClaimError::NotFound(id))
    }

    async fn release(&self, id: TaskId, actor: &str) -> Result<(), ClaimError> {
        let client = self
            .pool()
            .get()
            .await
            .map_err(|e| ClaimError::Store(pool_err(&e)))?;
        let freed = client
            .execute(
                "UPDATE tasks SET assignee = NULL, state = 'open'
                  WHERE task_id = $1 AND tenant = $3
                    AND assignee = $2 AND state = 'claimed'",
                &[&id.to_hex(), &actor.to_owned(), &self.tenant_name()],
            )
            .await
            .map_err(|e| ClaimError::Store(be(&e)))?;
        // The predicate did the work; the row count is what says whether it
        // matched. Discarding it reported success for a release that freed
        // nothing — caught by the conformance battery, not by this backend's
        // own tests.
        if freed == 0 {
            return Err(ClaimError::NotHeld {
                task: id,
                actor: actor.to_owned(),
            });
        }
        Ok(())
    }

    async fn set_state(&self, id: TaskId, state: TaskState) -> Result<bool, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        // A compare-and-set from the pending states: the settlement that lost
        // the race to a decision or an expiry must not overwrite the winner's.
        let n = client
            .execute(
                "UPDATE tasks SET state = $2
                  WHERE task_id = $1 AND tenant = $3 AND state = ANY($4::text[])",
                &[
                    &id.to_hex(),
                    &state.as_str(),
                    &self.tenant_name(),
                    &task_states(TaskState::is_pending),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        if n > 0 {
            return Ok(true);
        }
        // Nothing moved: settled already, or no such task — and the caller
        // needs to know which, as the embedded backend tells it.
        match task_on(&client, &self.tenant_name(), id).await? {
            Some(_) => Ok(false),
            None => Err(StoreError::NotFound(id.to_string())),
        }
    }

    async fn withdraw_run(&self, run: RunId, awaited: &[TaskId]) -> Result<usize, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let awaited: Vec<String> = awaited.iter().map(|id| id.to_hex()).collect();
        let n = client
            .execute(
                "UPDATE tasks SET state = $3
                  WHERE tenant = $1 AND run_id = $2 AND state = ANY($4::text[])
                    AND task_id = ANY($5::text[])",
                &[
                    &self.tenant_name(),
                    &run.to_string(),
                    &TaskState::Withdrawn.as_str(),
                    &task_states(TaskState::is_pending),
                    &awaited,
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        Ok(usize::try_from(n).unwrap_or(usize::MAX))
    }

    async fn escalate(&self, id: TaskId) -> Result<Task, StoreError> {
        // One connection for the whole verb, exactly as `claim` — see `task_on`.
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let tenant = self.tenant_name();
        let Some(task) = task_on(&client, &tenant, id).await? else {
            return Err(StoreError::NotFound(id.to_string()));
        };
        // A decided task stays decided — the sweep that escalates races the
        // reviewer it is escalating past, and the decision winning that race
        // is the outcome everybody wanted.
        if !task.state.is_pending() {
            return Ok(task);
        }
        // `Task::escalate` is the one implementation of what escalating means;
        // this backend only persists its result. The state predicate is the
        // same compare-and-swap `claim` carries: a decision landing between
        // the read above and this write must win, and an unguarded UPDATE
        // would resurrect it into `escalated`.
        let mut updated = task;
        updated.escalate();
        let n = client
            .execute(
                "UPDATE tasks SET state = $5, assignee = NULL, candidate_roles = $2
                  WHERE task_id = $1 AND tenant = $3
                    AND state = ANY($4::text[])",
                &[
                    &id.to_hex(),
                    &updated.candidate_roles,
                    &tenant,
                    &task_states(TaskState::is_pending),
                    &TaskState::Escalated.as_str(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        if n == 0 {
            // The race fired: somebody decided it. Return what stands.
            return task_on(&client, &tenant, id)
                .await?
                .ok_or_else(|| StoreError::NotFound(id.to_string()));
        }
        Ok(updated)
    }

    async fn queue(&self, roles: &[String], limit: usize) -> Result<Vec<Task>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        // Most urgent first, oldest within a rank — the trait's contract, which
        // the redb backend serves from a rank-keyed index. Ordering by age
        // alone made the limit a filter on the wrong axis: an urgent task
        // behind a page of older normal ones was simply absent from the page,
        // which for a worklist means the most important decision is the one
        // nobody is shown.
        let rows = client
            .query(
                &format!(
                    "SELECT {TASK_COLS} FROM tasks
                      WHERE tenant = $2 AND state = ANY($3::text[])
                      ORDER BY priority_rank ASC, created_at ASC
                      LIMIT $1"
                ),
                &[
                    &i64::try_from(limit).unwrap_or(i64::MAX),
                    &self.tenant_name(),
                    &task_states(TaskState::is_queued),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        let all: Result<Vec<Task>, StoreError> = rows.iter().map(task_from).collect();
        Ok(all?
            .into_iter()
            .filter(|t| {
                t.candidate_roles.is_empty() || t.candidate_roles.iter().any(|r| roles.contains(r))
            })
            .collect())
    }

    async fn for_case(&self, case: CaseId) -> Result<Vec<Task>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let rows = client
            .query(
                &format!(
                    "SELECT {TASK_COLS} FROM tasks
                      WHERE case_id = $1 AND tenant = $2 ORDER BY created_at ASC"
                ),
                &[&case.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;
        rows.iter().map(task_from).collect()
    }

    async fn open_count(&self) -> Result<u64, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let n: i64 = client
            .query_one(
                "SELECT COUNT(*) FROM tasks
                  WHERE tenant = $1 AND state = ANY($2::text[])",
                &[&self.tenant_name(), &task_states(TaskState::is_pending)],
            )
            .await
            .map_err(|e| be(&e))?
            .get(0);
        Ok(u64::try_from(n).unwrap_or(0))
    }

    async fn overdue(&self, now: Timestamp, limit: usize) -> Result<Vec<Task>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        // Not `escalated`, although escalated tasks are pending and past due:
        // this scan drives the expiry sweep, and escalation is that policy
        // having fired — see `TaskStore::overdue` for what including them
        // starves.
        let rows = client
            .query(
                &format!(
                    "SELECT {TASK_COLS} FROM tasks
                      WHERE tenant = $3 AND state = ANY($4::text[])
                        AND due_at IS NOT NULL AND due_at <= $1
                      ORDER BY due_at ASC LIMIT $2"
                ),
                &[
                    &now.unix_timestamp(),
                    &i64::try_from(limit).unwrap_or(i64::MAX),
                    &self.tenant_name(),
                    &task_states(TaskState::awaits_expiry),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        rows.iter().map(task_from).collect()
    }
}

const TASK_COLS: &str = "task_id, run_id, case_id, kind, justification, candidate_roles, \
                         escalate_to, excluded_actors, assignee, priority, state, on_expiry, \
                         created_at, due_at, withheld";

fn task_from(row: &tokio_postgres::Row) -> Result<Task, StoreError> {
    let id: String = row.get(0);
    let run: String = row.get(1);
    let case: Option<String> = row.get(2);
    let justification: String = row.get(4);
    let priority: String = row.get(9);
    let state: String = row.get(10);
    let on_expiry: String = row.get(11);
    let due: Option<i64> = row.get(13);
    let withheld: Option<String> = row.get(14);

    Ok(Task {
        id: TaskId::parse(&id).map_err(|e| corrupt("bad task id", e))?,
        run: RunId::parse(&run).map_err(|e| corrupt("bad run id", e))?,
        case: case
            .map(|c| CaseId::parse(&c))
            .transpose()
            .map_err(|e| corrupt("bad case id", e))?,
        kind: row.get(3),
        justification: serde_json::from_str(&justification)?,
        candidate_roles: row.get(5),
        escalate_to: row.get(6),
        excluded_actors: row.get(7),
        assignee: row.get(8),
        priority: priority_from(&priority)?,
        state: task_state_from(&state)?,
        on_expiry: expiry_from(&on_expiry)?,
        created_at: Timestamp::from_unix_timestamp(row.get::<_, i64>(12))
            .map_err(|e| corrupt("unrepresentable created_at", e))?,
        due_at: due
            .map(Timestamp::from_unix_timestamp)
            .transpose()
            .map_err(|e| corrupt("unrepresentable due_at", e))?,
        withheld: withheld
            .map(|w| decoded("withheld reason", &w, crate::core::Withheld::parse(&w)))
            .transpose()?,
    })
}

#[async_trait]
impl BatchStore for PostgresStore {
    fn tenant(&self) -> &str {
        crate::journal::JournalStore::tenant(self)
    }

    async fn open(&self, id: BatchId, plan_digest: &str) -> Result<(), StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        client
            .execute(
                "INSERT INTO batches (batch_id, plan_digest, tenant) VALUES ($1, $2, $3)
                 ON CONFLICT (tenant, batch_id) DO NOTHING",
                &[
                    &id.to_string(),
                    &plan_digest.to_owned(),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        // Read back what stands: whichever open won the insert, its digest is
        // the batch's, and a resume offering another one is refused rather
        // than run — one batch runs one frozen plan, and this row is the only
        // witness to which.
        let stored: String = client
            .query_one(
                "SELECT plan_digest FROM batches WHERE batch_id = $1 AND tenant = $2",
                &[&id.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?
            .get(0);
        if stored != plan_digest {
            return Err(StoreError::BatchPlanChanged {
                batch: id.to_string(),
                stored,
                offered: plan_digest.to_owned(),
            });
        }
        Ok(())
    }

    async fn plan_digest(&self, id: BatchId) -> Result<Option<String>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        Ok(client
            .query_opt(
                "SELECT plan_digest FROM batches WHERE batch_id = $1 AND tenant = $2",
                &[&id.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?
            .map(|r| r.get(0)))
    }

    async fn mark_exhausted(&self, id: BatchId) -> Result<(), StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let updated = client
            .execute(
                "UPDATE batches SET exhausted = TRUE WHERE batch_id = $1 AND tenant = $2",
                &[&id.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;
        // The row count says whether the mark landed — see `record`. This is
        // the one bit that lets a census read as *finished*, and `Ok` over
        // nothing is the quietest way to lose it.
        if updated == 0 {
            return Err(StoreError::NotFound(id.to_string()));
        }
        Ok(())
    }

    async fn is_exhausted(&self, id: BatchId) -> Result<bool, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        Ok(client
            .query_opt(
                "SELECT exhausted FROM batches WHERE batch_id = $1 AND tenant = $2",
                &[&id.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?
            .is_some_and(|r| r.get::<_, bool>(0)))
    }

    async fn reserve(
        &self,
        batch: BatchId,
        key: &str,
        run: RunId,
    ) -> Result<ItemRecord, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        // `DO NOTHING` then read back: an item already reserved must hand back
        // the *original* run id, or the journal holding its effects is orphaned
        // and they are performed again.
        client
            .execute(
                "INSERT INTO batch_items (batch_id, item_key, run_id, tenant)
                 VALUES ($1, $2, $3, $4)
                 ON CONFLICT (tenant, batch_id, item_key) DO NOTHING",
                &[
                    &batch.to_string(),
                    &key.to_owned(),
                    &run.to_string(),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        let row = client
            .query_one(
                "SELECT run_id, outcome, detail, tokens, minor FROM batch_items
                  WHERE batch_id = $1 AND item_key = $2 AND tenant = $3",
                &[&batch.to_string(), &key.to_owned(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;
        item_from(&row, key)
    }

    async fn record(
        &self,
        batch: BatchId,
        key: &str,
        outcome: &ItemOutcome,
        spend: Spend,
    ) -> Result<(), StoreError> {
        let detail = match outcome {
            ItemOutcome::Succeeded => None,
            ItemOutcome::Failed(d)
            | ItemOutcome::Quarantined(d)
            | ItemOutcome::Suspended(d)
            | ItemOutcome::Exhausted(d)
            | ItemOutcome::Withheld(d) => Some(d.clone()),
        };
        let state = outcome.as_str();
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let updated = client
            .execute(
                "UPDATE batch_items SET outcome = $3, detail = $4, tokens = $5, minor = $6
                  WHERE batch_id = $1 AND item_key = $2 AND tenant = $7",
                &[
                    &batch.to_string(),
                    &key.to_owned(),
                    &state,
                    &detail,
                    &sql_amount(spend.tokens),
                    &sql_amount(spend.minor_units),
                    &self.tenant_name(),
                ],
            )
            .await
            .map_err(|e| be(&e))?;
        // The predicate did the work; the row count says whether it matched.
        // Discarding it reported success for a record that wrote nothing —
        // the same lie a release that freed nothing tells.
        if updated == 0 {
            return Err(StoreError::NotFound(format!("{batch}/{key}")));
        }
        Ok(())
    }

    async fn cursor(&self, batch: BatchId) -> Result<Option<String>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        // The contiguous terminal prefix: an item still running or suspended
        // holds the cursor behind it, or a resume steps over work outstanding.
        let first_open: Option<String> = client
            .query_one(
                "SELECT MIN(item_key) FROM batch_items
                  WHERE batch_id = $1 AND tenant = $2
                    AND (outcome IS NULL OR NOT (outcome = ANY($3::text[])))",
                &[
                    &batch.to_string(),
                    &self.tenant_name(),
                    &ItemOutcome::terminal_tags(),
                ],
            )
            .await
            .map_err(|e| be(&e))?
            .get(0);

        let row = match first_open {
            Some(open) => client
                .query_one(
                    "SELECT MAX(item_key) FROM batch_items
                      WHERE batch_id = $1 AND item_key < $2 AND tenant = $3",
                    &[&batch.to_string(), &open, &self.tenant_name()],
                )
                .await
                .map_err(|e| be(&e))?,
            None => client
                .query_one(
                    "SELECT MAX(item_key) FROM batch_items
                      WHERE batch_id = $1 AND tenant = $2",
                    &[&batch.to_string(), &self.tenant_name()],
                )
                .await
                .map_err(|e| be(&e))?,
        };
        Ok(row.get(0))
    }

    async fn census(&self, batch: BatchId) -> Result<BatchCensus, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let rows = client
            .query(
                "SELECT outcome, COUNT(*), COALESCE(SUM(tokens),0), COALESCE(SUM(minor),0)
                   FROM batch_items WHERE batch_id = $1 AND tenant = $2
                  GROUP BY outcome",
                &[&batch.to_string(), &self.tenant_name()],
            )
            .await
            .map_err(|e| be(&e))?;

        let mut c = BatchCensus::default();
        for row in rows {
            let outcome: Option<String> = row.get(0);
            let n: i64 = row.get(1);
            let tokens: i64 = row.get(2);
            let minor: i64 = row.get(3);
            let n = amount_of(n);
            match outcome.as_deref() {
                // Bucketed off the *parsed* outcome, so an outcome added later
                // is a compiler error here rather than a row in no bucket.
                Some(tag) => {
                    match decoded("item outcome", tag, ItemOutcome::parse(tag, String::new()))? {
                        ItemOutcome::Succeeded => c.succeeded = n,
                        ItemOutcome::Failed(_) => c.failed = n,
                        ItemOutcome::Quarantined(_) => c.quarantined = n,
                        ItemOutcome::Suspended(_) => c.suspended = n,
                        ItemOutcome::Exhausted(_) => c.exhausted = n,
                        ItemOutcome::Withheld(_) => c.withheld = n,
                    }
                }
                // Reserved, no outcome recorded.
                None => c.in_flight = n,
            }
            c.spend += Spend {
                tokens: amount_of(tokens),
                minor_units: amount_of(minor),
            };
        }
        Ok(c)
    }

    async fn items(&self, batch: BatchId, limit: usize) -> Result<Vec<ItemRecord>, StoreError> {
        self.query_items(
            batch,
            limit,
            "SELECT item_key, run_id, outcome, detail, tokens, minor FROM batch_items
              WHERE batch_id = $1 AND tenant = $3 ORDER BY item_key ASC LIMIT $2",
            &[],
        )
        .await
    }

    async fn items_needing_attention(
        &self,
        batch: BatchId,
        limit: usize,
    ) -> Result<Vec<ItemRecord>, StoreError> {
        // `<> ALL` rather than `<> 'succeeded'`, and the set comes from the
        // type: an outcome added later lands in this listing without anyone
        // remembering to widen a literal here.
        //
        // `IS NULL OR` is load-bearing: a reserved item has no outcome at all,
        // and `NULL <> ALL (...)` is NULL, which `WHERE` treats as false — so
        // without it the interrupted items, the ones a crash left mid-flight,
        // would be exactly the ones this listing dropped.
        self.query_items(
            batch,
            limit,
            "SELECT item_key, run_id, outcome, detail, tokens, minor FROM batch_items
              WHERE batch_id = $1 AND tenant = $3
                AND (outcome IS NULL OR outcome <> ALL($4))
              ORDER BY item_key ASC LIMIT $2",
            &ItemOutcome::settled_tags(),
        )
        .await
    }
}

impl PostgresStore {
    /// One shape of item query, so the two listings cannot drift in how they
    /// decode a row or order it.
    async fn query_items(
        &self,
        batch: BatchId,
        limit: usize,
        sql: &str,
        settled: &[&str],
    ) -> Result<Vec<ItemRecord>, StoreError> {
        let client = self.pool().get().await.map_err(|e| pool_err(&e))?;
        let batch_key = batch.to_string();
        let capped = i64::try_from(limit).unwrap_or(i64::MAX);
        let tenant = self.tenant_name();
        let settled: Vec<String> = settled.iter().map(|s| (*s).to_owned()).collect();
        let mut params: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> =
            vec![&batch_key, &capped, &tenant];
        if !settled.is_empty() {
            params.push(&settled);
        }
        let rows = client.query(sql, &params).await.map_err(|e| be(&e))?;
        let mut out = Vec::with_capacity(rows.len());
        for row in rows {
            let key: String = row.get(0);
            out.push(ItemRecord {
                key: key.clone(),
                run: RunId::parse(&row.get::<_, String>(1))
                    .map_err(|e| corrupt("bad run id", e))?,
                outcome: outcome_from(row.get::<_, Option<String>>(2), row.get(3))?,
                spend: Spend {
                    tokens: amount_of(row.get::<_, i64>(4)),
                    minor_units: amount_of(row.get::<_, i64>(5)),
                },
            });
        }
        Ok(out)
    }
}

/// `None` means *no outcome yet* — and only that. An outcome string this
/// store cannot read is damage, not absence: decoded to `None` it would say
/// the item never ran, and the census would carry the row as in-flight
/// forever, keeping the batch `Running` over damage nobody is told about.
fn outcome_from(
    state: Option<String>,
    detail: Option<String>,
) -> Result<Option<ItemOutcome>, StoreError> {
    let d = detail.unwrap_or_default();
    let Some(state) = state else {
        return Ok(None);
    };
    decoded("item outcome", &state, ItemOutcome::parse(&state, d)).map(Some)
}

fn item_from(row: &tokio_postgres::Row, key: &str) -> Result<ItemRecord, StoreError> {
    let run: String = row.get(0);
    Ok(ItemRecord {
        key: key.to_owned(),
        run: RunId::parse(&run).map_err(|e| corrupt("bad run id", e))?,
        outcome: outcome_from(row.get::<_, Option<String>>(1), row.get(2))?,
        spend: Spend {
            tokens: amount_of(row.get::<_, i64>(3)),
            minor_units: amount_of(row.get::<_, i64>(4)),
        },
    })
}

#[cfg(test)]
mod codec_tests {
    use super::*;

    /// Every string this store writes is one it reads back as the same value.
    ///
    /// The encoder is exhaustive over the type, so the pair cannot drift while
    /// this passes: a variant added to `Phase`, `OnExpiry` or `Priority`
    /// changes the encoder's output and this fails on the decoder's refusal.
    #[test]
    fn every_written_string_decodes_to_the_value_that_wrote_it() {
        use crate::core::Phase;
        for phase in [Phase::Forward, Phase::Compensating] {
            assert_eq!(
                phase_from(crate::core::Phase::as_str(phase)).expect("round trip"),
                phase
            );
        }
        for expiry in [OnExpiry::Deny, OnExpiry::Escalate, OnExpiry::Proceed] {
            assert_eq!(
                expiry_from(OnExpiry::as_str(expiry)).expect("round trip"),
                expiry
            );
        }
        for priority in [
            Priority::Low,
            Priority::Normal,
            Priority::High,
            Priority::Urgent,
        ] {
            assert_eq!(
                priority_from(priority.as_str()).expect("round trip"),
                priority
            );
        }
    }

    /// **A row this store cannot read is not a row with a default value.**
    ///
    /// A decoder that answers with a fallback cannot report damage, so the
    /// damage arrives as a decision. `phase` is the one with teeth: it tells a
    /// step's forward pass from its compensating one, so an unreadable value
    /// answered `Forward` hands the unwind logic a compensating record wearing
    /// the wrong half of the saga. `Deny` and `Normal` are *safe* values,
    /// which is exactly what makes them the wrong answer — a fail-closed
    /// default is still a fact invented about a row nobody could read.
    #[test]
    fn an_unreadable_column_is_refused_rather_than_defaulted() {
        for (what, err) in [
            ("phase", phase_from("").err()),
            ("phase", phase_from("Compensating").err()),
            ("expiry", expiry_from("escalate_later").err()),
            ("priority", priority_from("normal ").err()),
            (
                "item outcome",
                outcome_from(Some("done".into()), None).err(),
            ),
        ] {
            assert!(
                matches!(err, Some(StoreError::Corrupt { .. })),
                "an unrecognised {what} decoded to a value instead of refusing"
            );
        }
    }
}