//! Real reviews: a review request for the classifier's forge account
//! ([`LOGIN`]) runs the repo's rules against the PR and posts the result as
//! that account's review, on the commit the run read.
//!
//! Only a review request starts a run. Opening or pushing to a PR makes none itself: the
//! notification router's `head-classifier` flow requests the review. A fresh
//! request retires the account's earlier review: the forge clears the
//! `official` flag on a reviewer's previous reviews when they are requested
//! again, and only official reviews count toward approvals or block a merge.
//!
//! A request for a head whose newest recorded review approved or requested
//! changes runs nothing: the account posts that verdict again, saying so. A
//! head whose newest review had no clean verdict, or that has none recorded,
//! gets a fresh run.
//!
//! A run with no clean verdict never approves: see [`no_clean_verdict_event`].
//!
//! Each accepted request waits in [`REQUEST_STREAM`], a work queue on the
//! swarm queue with one subject per PR that keeps one message per subject, so
//! a newer request for a PR replaces one still waiting. The controller takes
//! a request only once fewer than `max_concurrent_reviews` runs go, as the
//! swarm's [settings](super::settings) say, and acks it once its record is
//! written. A request a controller took and never acked, because it stopped
//! first, is delivered again. See [`consume`] for the rest.

use std::collections::HashSet;
use std::fmt::Write as _;
use std::future::Future;
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use std::time::{Duration, Instant};

use anyhow::Context as _;
use async_nats::jetstream::{self, AckKind, consumer::PullConsumer};
use forgejo_api::ForgejoError;
use futures_util::StreamExt as _;
use serde::{Deserialize, Serialize};
use swarm_wire::review_classifier::{
    Overall, REQUEST_CONSUMER, REQUEST_STREAM, REQUEST_SUBJECT_PREFIX, Reply, Row, TryOut, Verdict,
};
use tokio::sync::Notify;
use tokio::task::JoinSet;

use super::metrics::{Choice, Metrics, Request};
use super::runs::{self, RunOutcome, RunRecord, Runs, Trigger};
use super::settings::{Settings, SettingsSource};
use super::{FileSource, PullSource, RunSink, split_repo};
use crate::agent_renewal::unix_now;
use crate::forge::pull_diff::CONTEXT_LINES;
use crate::forge::{self, PullForReview, classifier_account::LOGIN, classifier_account::Review};
use crate::routes::AppState;

/// The waits before each retry of a post that failed with a [`transient`]
/// error or got no answer: six tries spread over about five minutes, sized to
/// outlast a swarm-bao restart (about 3-4 minutes observed on 2026-10-14).
/// Each try has its own [`forge::FORGE_DEADLINE`].
const POST_RETRY_WAITS: [Duration; 5] = [
    Duration::from_secs(5),
    Duration::from_secs(15),
    Duration::from_secs(45),
    Duration::from_secs(90),
    Duration::from_secs(150),
];

/// The queue's timing in production.
const TIMING: Timing = Timing {
    ack_wait: Duration::from_mins(1),
    pull_wait: Duration::from_secs(30),
    busy_retry: Duration::from_secs(30),
};

/// How the controller paces [`REQUEST_STREAM`].
#[derive(Debug, Clone, Copy)]
struct Timing {
    /// The consumer's ack wait: how long a taken request may go without an
    /// ack or a progress ack before the queue delivers it again. A request in
    /// hand gets a progress ack every third of it, so a run of any length
    /// keeps its request.
    ack_wait: Duration,
    /// How long one pull waits for a request; also the wait before opening
    /// the consumer again after that failed.
    pull_wait: Duration,
    /// How long a request for a PR whose review is going waits before the
    /// queue delivers it again.
    busy_retry: Duration,
}

/// The forge's review events.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Event {
    Approve,
    RequestChanges,
    Comment,
}

impl Event {
    fn forge(self) -> &'static str {
        match self {
            Self::Approve => "APPROVED",
            Self::RequestChanges => "REQUEST_CHANGES",
            Self::Comment => "COMMENT",
        }
    }
}

/// The metrics label of the review `event` a run chose.
const fn verdict_label(event: Event) -> Choice {
    match event {
        Event::Approve => Choice::Approve,
        Event::RequestChanges => Choice::RequestChanges,
        Event::Comment => Choice::NoVerdict,
    }
}

/// What a run posts when it has no clean verdict: the rules didn't load, the
/// PR couldn't be read, the run failed, or any classification has no answer.
/// A comment neither approves nor blocks the merge.
fn no_clean_verdict_event() -> Event {
    Event::Comment
}

/// A review request for [`LOGIN`], as it waits in [`REQUEST_STREAM`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct Requested {
    owner: String,
    name: String,
    pr: i64,
    /// The PR's head when the request was made, as the delivery names it;
    /// empty when it names none.
    head: String,
    /// Who asked for the review.
    requester: String,
}

#[derive(Deserialize)]
struct Payload {
    action: String,
    number: i64,
    pull_request: Option<PullRequest>,
    requested_reviewer: Option<User>,
    repository: Repository,
    sender: Option<User>,
}

#[derive(Deserialize)]
struct PullRequest {
    head: Head,
}

#[derive(Deserialize)]
struct Head {
    sha: String,
}

#[derive(Deserialize)]
struct User {
    login: String,
}

#[derive(Deserialize)]
struct Repository {
    full_name: String,
}

impl Requested {
    /// The request in a `pull_request_review_request` delivery, when it
    /// asks for [`LOGIN`]'s review. A removed request, a request for anyone
    /// else and an unparsable body are `None`.
    fn parse(body: &[u8]) -> Option<Self> {
        let payload: Payload = match serde_json::from_slice(body) {
            Ok(p) => p,
            Err(e) => {
                tracing::warn!(error = %e, "webhook: review-request delivery did not parse");
                return None;
            }
        };
        let reviewer = payload.requested_reviewer?;
        if payload.action != "review_requested" || !reviewer.login.eq_ignore_ascii_case(LOGIN) {
            return None;
        }
        let (owner, name) = split_repo(&payload.repository.full_name).ok()?;
        Some(Self {
            owner: owner.to_owned(),
            name: name.to_owned(),
            pr: Some(payload.number).filter(|n| *n > 0)?,
            head: payload.pull_request.map(|p| p.head.sha).unwrap_or_default(),
            requester: payload.sender.map(|s| s.login).unwrap_or_default(),
        })
    }

    fn pr_key(&self) -> PrKey {
        (self.owner.clone(), self.name.clone(), self.pr)
    }

    /// This PR's subject in [`REQUEST_STREAM`].
    fn subject(&self) -> String {
        format!(
            "{REQUEST_SUBJECT_PREFIX}.{}.{}.{}",
            token(&self.owner),
            token(&self.name),
            self.pr
        )
    }
}

/// `text` as one subject token: ASCII letters, digits and `-` as they are,
/// every other byte as `_` and its two hex digits. So a `.` in a repo name
/// can't split it, and two names never share a token.
fn token(text: &str) -> String {
    let mut out = String::with_capacity(text.len());
    for b in text.bytes() {
        if b.is_ascii_alphanumeric() || b == b'-' {
            out.push(char::from(b));
        } else {
            let _ = write!(out, "_{b:02x}");
        }
    }
    out
}

/// Repo owner, repo name and PR number.
type PrKey = (String, String, i64);

/// The run places, and the PRs with a review going.
///
/// A request for a PR whose review is going waits in the queue until that
/// review ends, so one PR never has two runs at once.
pub(crate) struct Reviews {
    slots: Mutex<Slots>,
    /// Woken when a run ends or the limit changes.
    changed: Notify,
    going: Mutex<HashSet<PrKey>>,
}

/// How many runs may go at once, and how many do.
struct Slots {
    limit: usize,
    running: usize,
}

impl Default for Reviews {
    fn default() -> Self {
        Self {
            slots: Mutex::new(Slots {
                limit: Settings::default().max_concurrent_reviews,
                running: 0,
            }),
            changed: Notify::new(),
            going: Mutex::default(),
        }
    }
}

impl Reviews {
    fn slots(&self) -> MutexGuard<'_, Slots> {
        self.slots.lock().unwrap_or_else(PoisonError::into_inner)
    }

    /// Let `limit` runs go at once from now on. Runs already going when it
    /// shrinks finish; no new one starts until fewer than `limit` run.
    pub(crate) fn resize(&self, limit: usize) {
        self.slots().limit = limit;
        self.changed.notify_waiters();
    }

    #[cfg(test)]
    pub(crate) fn limit(&self) -> usize {
        self.slots().limit
    }

    /// `req`'s PR, marked as having a review going, or `None` when one is
    /// going already.
    fn claim(self: &Arc<Self>, req: &Requested) -> Option<Claim> {
        let pr = req.pr_key();
        let mut going = self.going.lock().unwrap_or_else(PoisonError::into_inner);
        going.insert(pr.clone()).then(|| Claim {
            reviews: Arc::clone(self),
            pr,
        })
    }

    /// A place to run in, once fewer runs go than the limit allows.
    async fn slot(self: &Arc<Self>) -> Slot {
        loop {
            let changed = self.changed.notified();
            tokio::pin!(changed);
            changed.as_mut().enable();
            {
                let mut slots = self.slots();
                if slots.running < slots.limit {
                    slots.running += 1;
                    return Slot(Arc::clone(self));
                }
            }
            changed.await;
        }
    }
}

/// One PR's review going; dropping it lets the PR's next request run.
struct Claim {
    reviews: Arc<Reviews>,
    pr: PrKey,
}

impl Drop for Claim {
    fn drop(&mut self) {
        self.reviews
            .going
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .remove(&self.pr);
    }
}

/// One run holding a place; dropping it lets the next one start.
struct Slot(Arc<Reviews>);

impl Drop for Slot {
    fn drop(&mut self) {
        self.0.slots().running -= 1;
        self.0.changed.notify_waiters();
    }
}

/// Queue the review request in a verified `review-request` delivery that
/// asks for [`LOGIN`], and return at once.
pub(crate) fn on_delivery(state: &AppState, body: &[u8]) {
    let Some(request) = Requested::parse(body) else {
        return;
    };
    let Some(forge) = state.forge.get().map(Arc::clone) else {
        tracing::warn!(
            repo = %format!("{}/{}", request.owner, request.name),
            pr = request.pr,
            "review requested, but this host has no forge access yet; nothing is posted"
        );
        return;
    };
    let client = state.status.as_ref().map(|s| s.queue_client());
    let runs = state.runs.clone();
    let metrics = Metrics::global();
    tokio::spawn(async move { accept(client.as_ref(), &*forge, &runs, &metrics, &request).await });
}

/// Put `request` in [`REQUEST_STREAM`]. When that fails, the account posts a
/// no-verdict comment saying so, recorded in `runs`.
async fn accept(
    client: Option<&async_nats::Client>,
    poster: &impl ReviewPoster,
    runs: &Runs,
    metrics: &Metrics,
    request: &Requested,
) {
    let repo = format!("{}/{}", request.owner, request.name);
    let queued = match client {
        Some(client) => enqueue(client, request).await,
        None => Err(anyhow::anyhow!("no swarm queue is configured on this host")),
    };
    let error = match queued {
        Ok(()) => {
            metrics.request(Request::Queued);
            tracing::info!(
                %repo, pr = request.pr, head_sha = request.head.as_str(),
                "review request queued"
            );
            return;
        }
        Err(e) => format!("{e:#}"),
    };
    metrics.request(Request::QueueFailed);
    metrics.verdict(Choice::NoVerdict);
    tracing::warn!(
        %repo, pr = request.pr, head_sha = request.head.as_str(), %error,
        "queueing a review request failed; posting a no-verdict comment"
    );
    let started_unix = unix_now();
    let ruling = Ruling::no_verdict(
        format!(
            "the request could not be queued: {error}; remove the review request and request \
             it again"
        ),
        &[],
    );
    let reason = ruling.reason.clone();
    let sent = post(poster, request, None, ruling.review()).await;
    let ended = ended(RunOutcome::NoVerdict, reason, sent);
    runs.record(record(request, None, started_unix, ended, (None, None)))
        .await;
}

/// Open [`REQUEST_STREAM`], creating it if nothing has yet.
async fn open_stream(client: &async_nats::Client) -> anyhow::Result<jetstream::stream::Stream> {
    swarm_queue_client::ensure_connected(client)?;
    jetstream::new(client.clone())
        .get_or_create_stream(jetstream::stream::Config {
            name: REQUEST_STREAM.to_owned(),
            description: Some(
                "Review requests the review classifier has accepted and not yet finished"
                    .to_owned(),
            ),
            subjects: vec![format!("{REQUEST_SUBJECT_PREFIX}.>")],
            retention: jetstream::stream::RetentionPolicy::WorkQueue,
            max_messages_per_subject: 1,
            ..Default::default()
        })
        .await
        .with_context(|| format!("opening the {REQUEST_STREAM} stream"))
}

/// Publish `request` on its PR's subject, replacing a request for the PR
/// that's still waiting, once the queue has stored it.
async fn enqueue(client: &async_nats::Client, request: &Requested) -> anyhow::Result<()> {
    open_stream(client).await?;
    let payload = serde_json::to_vec(request).context("encoding the review request")?;
    jetstream::new(client.clone())
        .publish(request.subject(), payload.into())
        .await
        .context("publishing the review request")?
        .await
        .context("the queue did not store the review request")?;
    Ok(())
}

/// Open [`REQUEST_CONSUMER`] on [`REQUEST_STREAM`], creating either, with
/// `ack_wait`.
async fn open_consumer(
    client: &async_nats::Client,
    ack_wait: Duration,
) -> anyhow::Result<PullConsumer> {
    open_stream(client)
        .await?
        .create_consumer(jetstream::consumer::pull::Config {
            durable_name: Some(REQUEST_CONSUMER.to_owned()),
            description: Some("swarm-controller's requested review runs".to_owned()),
            ack_policy: jetstream::consumer::AckPolicy::Explicit,
            ack_wait,
            ..Default::default()
        })
        .await
        .with_context(|| format!("opening the {REQUEST_CONSUMER} consumer on {REQUEST_STREAM}"))
}

/// Take the queued review requests and review each, for as long as the
/// controller runs, once this host has forge access.
pub(crate) fn spawn_queue(state: &AppState) {
    let Some(client) = state.status.as_ref().map(|s| s.queue_client()) else {
        tracing::warn!("no swarm queue is configured on this host; requested reviews don't run");
        return;
    };
    let forge = Arc::clone(&state.forge);
    let reviews = Arc::clone(&state.reviews);
    let runs = state.runs.clone();
    tokio::spawn(async move {
        let forge = loop {
            if let Some(forge) = forge.get() {
                break Arc::clone(forge);
            }
            tokio::time::sleep(TIMING.pull_wait).await;
        };
        let queue = Queue(Some(client.clone()));
        let handles = Handles {
            forge: Arc::clone(&forge),
            queue,
            poster: forge,
            runs,
            metrics: Metrics::global(),
        };
        consume(&client, TIMING, &reviews, handles).await;
    });
}

/// What a review reaches outside the queue: the forge it reads the PR and
/// rules from, the classifier it runs on, the poster for its review, and the
/// store that records it.
struct Handles<S, Q, P> {
    forge: Arc<S>,
    queue: Q,
    poster: Arc<P>,
    runs: Runs,
    metrics: Metrics,
}

impl<S, Q: Clone, P> Clone for Handles<S, Q, P> {
    fn clone(&self) -> Self {
        Self {
            forge: Arc::clone(&self.forge),
            queue: self.queue.clone(),
            poster: Arc::clone(&self.poster),
            runs: self.runs.clone(),
            metrics: self.metrics.clone(),
        }
    }
}

/// Take requests from [`REQUEST_STREAM`] one at a time, each once a run
/// place is free, and review each in a task of its own. Never returns;
/// dropping it stops every review it started, and the queue delivers their
/// requests again.
///
/// The limit is read from the settings before each wait for a place. A
/// request that doesn't parse is dropped. One for a PR whose review is going
/// waits for [`Timing::busy_retry`] in the queue, without holding a place.
async fn consume<S, Q, P>(
    client: &async_nats::Client,
    timing: Timing,
    reviews: &Arc<Reviews>,
    handles: Handles<S, Q, P>,
) where
    S: FileSource + PullSource + Send + Sync + 'static,
    Q: RunSink + SettingsSource,
    P: ReviewPoster + Send + Sync + 'static,
{
    let consumer = loop {
        match open_consumer(client, timing.ack_wait).await {
            Ok(consumer) => break consumer,
            Err(e) => tracing::warn!(
                error = %format!("{e:#}"),
                "opening the review request queue failed; trying again"
            ),
        }
        tokio::time::sleep(timing.pull_wait).await;
    };
    let mut tasks = JoinSet::new();
    loop {
        while let Some(ended) = tasks.try_join_next() {
            if let Err(e) = ended {
                tracing::error!(error = %e, "a requested review's task ended early");
            }
        }
        match handles.queue.settings().await {
            Ok(settings) => reviews.resize(settings.max_concurrent_reviews),
            Err(e) => tracing::warn!(
                error = %format!("{e:#}"),
                "reading the review classifier's settings failed; the next run waits under \
                 the last limit read"
            ),
        }
        let slot = reviews.slot().await;
        let message = match next(&consumer, timing.pull_wait).await {
            Ok(Some(message)) => message,
            Ok(None) => continue,
            Err(e) => {
                tracing::warn!(
                    error = %format!("{e:#}"),
                    "taking a review request from the queue failed; trying again"
                );
                drop(slot);
                tokio::time::sleep(timing.pull_wait).await;
                continue;
            }
        };
        tasks.spawn(take(
            message,
            slot,
            timing,
            Arc::clone(reviews),
            handles.clone(),
        ));
    }
}

/// The next request in the queue, or `None` when none came within `wait`.
async fn next(
    consumer: &PullConsumer,
    wait: Duration,
) -> anyhow::Result<Option<jetstream::Message>> {
    let mut batch = consumer
        .batch()
        .max_messages(1)
        .expires(wait)
        .messages()
        .await
        .context("asking the queue for a review request")?;
    match batch.next().await {
        Some(Ok(message)) => Ok(Some(message)),
        Some(Err(e)) => Err(anyhow::anyhow!(e).context("reading a review request")),
        None => Ok(None),
    }
}

/// Review the request in `message`, holding `slot`, and ack it once its
/// record is written.
async fn take<S, Q, P>(
    message: jetstream::Message,
    slot: Slot,
    timing: Timing,
    reviews: Arc<Reviews>,
    handles: Handles<S, Q, P>,
) where
    S: FileSource + PullSource + Send + Sync + 'static,
    Q: RunSink + SettingsSource,
    P: ReviewPoster + Send + Sync + 'static,
{
    let request: Requested = match serde_json::from_slice(&message.payload) {
        Ok(request) => request,
        Err(e) => {
            tracing::warn!(
                subject = %message.subject, error = %e,
                "a queued review request did not parse; dropping it"
            );
            handles.metrics.request(Request::Unparsable);
            if let Err(e) = message.ack_with(AckKind::Term).await {
                tracing::warn!(error = %e, "dropping an unparsable review request failed");
            }
            return;
        }
    };
    let Some(claim) = reviews.claim(&request) else {
        drop(slot);
        handles.metrics.request(Request::Deferred);
        tracing::info!(
            repo = %format!("{}/{}", request.owner, request.name), pr = request.pr,
            head_sha = request.head.as_str(),
            "review requested while this PR's review is going; it waits in the queue"
        );
        if let Err(e) = message
            .ack_with(AckKind::Nak(Some(timing.busy_retry)))
            .await
        {
            tracing::warn!(error = %e, "putting a review request back in the queue failed");
        }
        return;
    };
    let Handles {
        forge,
        queue,
        poster,
        runs,
        metrics,
    } = &handles;
    let work = review(&**forge, queue, &**poster, runs, metrics, &request, slot);
    in_progress(&message, timing.ack_wait / 3, work).await;
    drop(claim);
    if let Err(e) = message.ack().await {
        tracing::warn!(
            repo = %format!("{}/{}", request.owner, request.name), pr = request.pr,
            error = %e,
            "acking a finished review request failed; the queue delivers it again"
        );
    }
}

/// Run `work`, telling the queue every `every` that `message` is still in
/// hand.
async fn in_progress(
    message: &jetstream::Message,
    every: Duration,
    work: impl Future<Output = ()>,
) {
    tokio::pin!(work);
    let mut ticks = tokio::time::interval_at(tokio::time::Instant::now() + every, every);
    loop {
        tokio::select! {
            () = &mut work => return,
            _ = ticks.tick() => {
                if let Err(e) = message.ack_with(AckKind::Progress).await {
                    tracing::warn!(error = %e, "telling the queue a review is still going failed");
                }
            }
        }
    }
}

/// Review `req`, holding `slot`, and record the result in `runs`.
///
/// A request whose head is still the PR's, and whose head's newest review
/// [`reposted`] covers, frees `slot` and posts that verdict again. Any other
/// runs on the PR's current head, reading the settings for the model and
/// effort: a request for a head the PR has moved on from gets a review of the
/// new head, never the old head's verdict.
async fn review(
    forge: &(impl FileSource + PullSource),
    sink: &(impl RunSink + SettingsSource),
    poster: &impl ReviewPoster,
    runs: &Runs,
    metrics: &Metrics,
    req: &Requested,
    slot: Slot,
) {
    let pull = read_pull(forge, req).await;
    let moved = matches!(&pull, Ok(pull) if !req.head.is_empty() && pull.head_sha != req.head);
    if moved {
        tracing::info!(
            repo = %format!("{}/{}", req.owner, req.name), pr = req.pr,
            head_sha = req.head.as_str(),
            "review requested on a head the PR has moved on from; reviewing its current head"
        );
    }
    if !moved
        && let Some(earlier) = newest_review(runs, req).await
        && let Some(ruling) = reposted(&earlier)
    {
        drop(slot);
        metrics.request(Request::Reposted);
        metrics.verdict(verdict_label(ruling.event));
        runs.record(repost(poster, req, &earlier, ruling).await)
            .await;
        return;
    }
    metrics.request(Request::Started);
    let run = Box::pin(run(forge, pull, sink, poster, metrics, req)).await;
    drop(slot);
    runs.record(run).await;
}

/// The PR `req` names, as the forge answers within [`forge::FORGE_DEADLINE`],
/// or why it couldn't be read.
async fn read_pull(pulls: &impl PullSource, req: &Requested) -> Result<PullForReview, String> {
    match forge::within_forge_deadline(
        "review classifier PR read",
        pulls.pull(&req.owner, &req.name, req.pr),
    )
    .await
    {
        Some(Ok(pull)) => Ok(pull),
        Some(Err(e)) => Err(format!("reading the PR failed: {e:#}")),
        None => Err("the forge did not answer within the deadline".to_owned()),
    }
}

/// The newest recorded review of the head `req` names. `None` when it names
/// none, nothing is recorded for it, or the store can't be read; each of
/// those leads to a fresh run.
async fn newest_review(runs: &Runs, req: &Requested) -> Option<RunRecord> {
    if req.head.is_empty() {
        return None;
    }
    let repo = format!("{}/{}", req.owner, req.name);
    match runs
        .newest_review(&repo, req.pr.unsigned_abs(), &req.head)
        .await
    {
        Ok(found) => found,
        Err(e) => {
            tracing::warn!(
                %repo, pr = req.pr, head_sha = req.head.as_str(), error = %format!("{e:#}"),
                "reading the review classifier's earlier runs failed; running the review"
            );
            None
        }
    }
}

/// The ruling that posts `earlier`'s verdict again, when it approved or
/// requested changes. `None` for any other outcome, such as no verdict, which
/// a fresh run may improve on.
fn reposted(earlier: &RunRecord) -> Option<Ruling> {
    let event = match earlier.outcome {
        RunOutcome::Approved => Event::Approve,
        RunOutcome::ChangesRequested => Event::RequestChanges,
        _ => return None,
    };
    let (Some(rows), Some(overall)) = (&earlier.rows, earlier.overall) else {
        return None;
    };
    let head = earlier.head_sha.as_deref().unwrap_or_default();
    Some(Ruling {
        event,
        body: format!(
            "Already reviewed `{head}`, verdict unchanged.\n\n{}\n{}",
            table(rows),
            overall_line(overall)
        ),
        reason: Some(format!("a re-post of run {}'s verdict", earlier.id)),
    })
}

/// Post `ruling`, `earlier`'s verdict, on `req`'s PR at `earlier`'s head. Its
/// record, carrying `earlier`'s answer.
async fn repost(
    poster: &impl ReviewPoster,
    req: &Requested,
    earlier: &RunRecord,
    ruling: Ruling,
) -> RunRecord {
    let started_unix = unix_now();
    tracing::info!(
        repo = %earlier.repo, pr = req.pr, head_sha = req.head.as_str(), earlier = %earlier.id,
        "review requested on a head already reviewed; posting its verdict again"
    );
    let reason = ruling.reason.clone();
    let sent = post(poster, req, earlier.head_sha.as_deref(), ruling.review()).await;
    record(
        req,
        earlier.head_sha.clone(),
        started_unix,
        ended(earlier.outcome, reason, sent),
        (earlier.rows.clone(), earlier.overall),
    )
}

/// Where a review is posted: the forge in production, a fake in the tests.
pub(crate) trait ReviewPoster {
    /// The posted review's URL, when the forge names one.
    fn post(
        &self,
        owner: &str,
        repo: &str,
        number: i64,
        commit: Option<&str>,
        review: Review,
    ) -> impl Future<Output = anyhow::Result<Option<String>>> + Send;
}

impl ReviewPoster for forge::Client {
    async fn post(
        &self,
        owner: &str,
        repo: &str,
        number: i64,
        commit: Option<&str>,
        review: Review,
    ) -> anyhow::Result<Option<String>> {
        self.post_classifier_review(owner, repo, number, commit, review)
            .await
    }
}

/// The swarm queue, when this host has one.
#[derive(Clone)]
struct Queue(Option<async_nats::Client>);

impl RunSink for Queue {
    async fn run(&self, request: Vec<u8>) -> Reply {
        let Some(client) = &self.0 else {
            return Reply::Failed {
                error: "no swarm queue is configured on this host".to_owned(),
            };
        };
        if let Err(e) = swarm_queue_client::ensure_connected(client) {
            return Reply::Failed {
                error: swarm_queue_client::chain(&e),
            };
        }
        client.run(request).await
    }
}

impl SettingsSource for Queue {
    async fn settings(&self) -> anyhow::Result<Settings> {
        let Some(client) = &self.0 else {
            anyhow::bail!("no swarm queue is configured on this host");
        };
        swarm_queue_client::ensure_connected(client)?;
        client.settings().await
    }
}

/// How far a run got.
#[derive(Debug, Clone, PartialEq, Eq)]
enum Outcome {
    /// The repo has no rules file: nothing was asked of the classifier.
    NoRules { repo: String },
    /// Nothing reached the classifier, for this reason.
    NotRun(String),
    /// The classifier answered a run of `sent` classifications.
    Ran { sent: usize, reply: Reply },
}

/// Read the rules, run them on `pull`, the PR as it was read, and post the
/// review. The finished run.
async fn run(
    files: &impl FileSource,
    pull: Result<PullForReview, String>,
    sink: &(impl RunSink + SettingsSource),
    poster: &impl ReviewPoster,
    metrics: &Metrics,
    req: &Requested,
) -> RunRecord {
    let started_unix = unix_now();
    let timer = Instant::now();
    let (head, narrowed, outcome) = classify(files, pull, sink, req).await;
    let mut ruling = review_for(&outcome);
    metrics.verdict(verdict_label(ruling.event));
    if let Outcome::Ran {
        reply: Reply::Done { rows, .. },
        ..
    } = &outcome
    {
        metrics.checks(rows);
    }
    if let (Some(reason), Outcome::Ran { .. }) = (narrowed, &outcome) {
        let _ = write!(
            ruling.body,
            "\nThe classifier read the forge's diff, with 3 lines of context, not \
             {CONTEXT_LINES}: {reason}.\n"
        );
    }
    let posted_as = match ruling.event {
        Event::Approve => RunOutcome::Approved,
        Event::RequestChanges => RunOutcome::ChangesRequested,
        Event::Comment => RunOutcome::NoVerdict,
    };
    let reason = ruling.reason.clone();
    let sent = post(poster, req, head.as_deref(), ruling.review()).await;
    let answer = match &outcome {
        Outcome::Ran { reply, .. } => runs::answer(reply),
        Outcome::NotRun(_) | Outcome::NoRules { .. } => (None, None),
    };
    let ended = ended(posted_as, reason, sent);
    metrics.run_duration(ended.outcome, timer.elapsed().as_secs_f64());
    record(req, head, started_unix, ended, answer)
}

/// Post `review` on `req`'s PR, naming `head` when there is one. The posted
/// review's URL when the forge names one, or why the post failed.
async fn post(
    poster: &impl ReviewPoster,
    req: &Requested,
    head: Option<&str>,
    review: Review,
) -> Result<Option<String>, String> {
    let repo = format!("{}/{}", req.owner, req.name);
    let event = review.event;
    let mut tries = 0;
    let sent = post_retrying(poster, req, head, &review, &mut tries).await;
    match sent {
        Some(Ok(url)) => {
            tracing::info!(
                %repo, pr = req.pr, head_sha = head.unwrap_or_default(), event, tries,
                "review classifier review posted"
            );
            Ok(url)
        }
        Some(Err(e)) => {
            let error = format!("{e:#}");
            tracing::warn!(
                %repo, pr = req.pr, %error, tries,
                "posting the review classifier's review failed"
            );
            Err(format!("posting the review failed: {error}"))
        }
        None => {
            tracing::warn!(
                %repo, pr = req.pr, tries,
                "posting the review classifier's review failed: no answer within the deadline"
            );
            Err("the forge did not answer the review's post within the deadline".to_owned())
        }
    }
}

/// Post `review`, each try within [`forge::FORGE_DEADLINE`], and post it
/// again after each of [`POST_RETRY_WAITS`] while a try fails with a
/// [`transient`] error or gets no answer by the deadline. The last try's
/// result, `None` when it got no answer; `tries` counts the tries made.
async fn post_retrying(
    poster: &impl ReviewPoster,
    req: &Requested,
    head: Option<&str>,
    review: &Review,
    tries: &mut usize,
) -> Option<anyhow::Result<Option<String>>> {
    let mut waits = POST_RETRY_WAITS.iter();
    loop {
        *tries += 1;
        let sent = forge::within_forge_deadline(
            "review classifier review",
            poster.post(&req.owner, &req.name, req.pr, head, review.clone()),
        )
        .await;
        let error = match &sent {
            Some(Err(e)) if transient(e) => format!("{e:#}"),
            None => "no answer within the deadline".to_owned(),
            Some(_) => return sent,
        };
        let Some(wait) = waits.next() else {
            return sent;
        };
        tracing::info!(
            repo = %format!("{}/{}", req.owner, req.name), pr = req.pr,
            %error, tries, wait_s = wait.as_secs(),
            "posting the review classifier's review failed; trying again"
        );
        tokio::time::sleep(*wait).await;
    }
}

/// Whether a post that failed with `e` may succeed when tried again: the
/// connection to the secret store or the forge failed or timed out. An answer
/// from either, such as a 4xx, is not transient.
///
/// The store's client puts its `reqwest::Error` in the chain itself; the
/// forge's wraps it in a transparent [`ForgejoError::ReqwestError`], whose
/// chain skips it.
fn transient(e: &anyhow::Error) -> bool {
    e.chain().any(|cause| {
        let http = match cause.downcast_ref::<ForgejoError>() {
            Some(ForgejoError::ReqwestError(e)) => Some(e),
            _ => cause.downcast_ref::<reqwest::Error>(),
        };
        http.is_some_and(|e| e.is_connect() || e.is_timeout())
    })
}

/// How a run ended, as its record holds it.
struct Ended {
    outcome: RunOutcome,
    detail: Option<String>,
    review_url: Option<String>,
}

/// `posted_as` when the post went through, `post-failed` otherwise. `reason`
/// is why the run has no clean verdict, if it has none.
fn ended(
    posted_as: RunOutcome,
    reason: Option<String>,
    posted: Result<Option<String>, String>,
) -> Ended {
    match posted {
        Ok(review_url) => Ended {
            outcome: posted_as,
            detail: reason,
            review_url,
        },
        Err(error) => Ended {
            outcome: RunOutcome::PostFailed,
            detail: Some(match reason {
                Some(reason) => format!("{reason}; {error}"),
                None => error,
            }),
            review_url: None,
        },
    }
}

/// The record of a run on `req` that read `head`.
fn record(
    req: &Requested,
    head: Option<String>,
    started_unix: i64,
    ended: Ended,
    (rows, overall): (Option<Vec<Row>>, Option<Overall>),
) -> RunRecord {
    let repo = format!("{}/{}", req.owner, req.name);
    RunRecord {
        id: super::mint_id().unwrap_or_else(|_| format!("{repo}#{}@{started_unix}", req.pr)),
        trigger: Trigger::Review,
        repo,
        pr: req.pr.unsigned_abs(),
        requester: req.requester.clone(),
        head_sha: head,
        started_unix,
        finished_unix: unix_now(),
        outcome: ended.outcome,
        detail: ended.detail,
        rows,
        overall,
        review_url: ended.review_url,
    }
}

/// The PR's head when it was read, why its diff is the forge's 3-line one
/// when it is, and how far the run got on `pull`, the PR as it was read.
async fn classify(
    files: &impl FileSource,
    pull: Result<PullForReview, String>,
    sink: &(impl RunSink + SettingsSource),
    req: &Requested,
) -> (Option<String>, Option<String>, Outcome) {
    let repo = format!("{}/{}", req.owner, req.name);
    let pull = match pull {
        Ok(pull) => pull,
        Err(reason) => return (None, None, Outcome::NotRun(reason)),
    };
    let head = Some(pull.head_sha.clone());
    let narrowed = pull.narrowed;
    let rules = match super::load_rules(files, &req.owner, &req.name).await {
        Ok(super::Loaded::Rules(rules)) => rules,
        Ok(super::Loaded::Absent { .. }) => return (head, narrowed, Outcome::NoRules { repo }),
        Err(problem) => {
            let detail = problem.detail.unwrap_or_default();
            return (
                head,
                narrowed,
                Outcome::NotRun(format!("the rules did not load: {detail}")),
            );
        }
    };
    let settings = match sink.settings().await {
        Ok(settings) => settings,
        Err(e) => {
            return (
                head,
                narrowed,
                Outcome::NotRun(format!("reading the classifier's settings failed: {e:#}")),
            );
        }
    };
    let runtime = match settings.runtime() {
        Ok(runtime) => runtime,
        Err(e) => {
            return (
                head,
                narrowed,
                Outcome::NotRun(format!("the classifier's settings: {e}")),
            );
        }
    };
    let request = TryOut {
        operator: req.requester.clone(),
        repo,
        pr: req.pr.unsigned_abs(),
        head_sha: pull.head_sha,
        commits: pull.commits,
        diff: pull.diff,
        classifications: rules.classifications,
        model: settings.model,
        effort: settings.effort,
        runtime,
    };
    let sent = request.classifications.len();
    let outcome = match serde_json::to_vec(&request) {
        Ok(payload) => Outcome::Ran {
            sent,
            reply: sink.run(payload).await,
        },
        Err(e) => Outcome::NotRun(format!("encoding the run failed: {e}")),
    };
    (head, narrowed, outcome)
}

/// What a run posts: the event, the review body, and why it has no clean
/// verdict when it has none.
struct Ruling {
    event: Event,
    body: String,
    reason: Option<String>,
}

impl Ruling {
    fn review(self) -> Review {
        Review {
            event: self.event.forge(),
            body: self.body,
        }
    }

    fn no_verdict(reason: String, rows: &[Row]) -> Self {
        Self {
            event: no_clean_verdict_event(),
            body: no_verdict(&reason, rows),
            reason: Some(reason),
        }
    }
}

/// The ruling on a run's outcome.
fn review_for(outcome: &Outcome) -> Ruling {
    match outcome {
        Outcome::NoRules { repo } => Ruling {
            event: Event::Approve,
            body: format!("no review rules are configured for {repo}; approved without checks\n"),
            reason: None,
        },
        Outcome::NotRun(reason) => Ruling::no_verdict(reason.clone(), &[]),
        Outcome::Ran {
            reply: Reply::Failed { error },
            ..
        } => Ruling::no_verdict(format!("the run failed: {error}"), &[]),
        Outcome::Ran {
            sent,
            reply: Reply::Done { rows, overall },
        } => {
            let unanswered = rows.iter().filter(|r| r.error.is_some()).count();
            if rows.is_empty() || rows.len() != *sent {
                let reason = format!(
                    "the classifier answered {} of {sent} classifications",
                    rows.len()
                );
                Ruling::no_verdict(reason, rows)
            } else if unanswered > 0 {
                let reason = format!(
                    "{unanswered} of {} classifications got no answer",
                    rows.len()
                );
                Ruling::no_verdict(reason, rows)
            } else {
                let event = match overall {
                    Overall::Approve => Event::Approve,
                    Overall::RequestChanges => Event::RequestChanges,
                };
                Ruling {
                    event,
                    body: format!("{}\n{}", table(rows), overall_line(*overall)),
                    reason: None,
                }
            }
        }
    }
}

/// A review body for a run without a clean verdict.
fn no_verdict(reason: &str, rows: &[Row]) -> String {
    let mut body = format!(
        "**No verdict:** {}. This review doesn't approve or request changes.\n",
        cell(reason)
    );
    if !rows.is_empty() {
        body.push('\n');
        body.push_str(&table(rows));
    }
    body
}

/// The result table, one row per classification.
fn table(rows: &[Row]) -> String {
    let mut out = String::from(
        "| result | check | certainty | reason |\n\
         | --- | --- | --- | --- |\n",
    );
    for r in rows {
        let (certainty, reason) = match &r.error {
            Some(error) => ("—".to_owned(), error.as_str()),
            None => (r.certainty.to_string(), r.reason.as_str()),
        };
        let _ = writeln!(
            out,
            "| {} | {} | {certainty} / {} | {} |",
            result(r),
            cell(&r.name),
            r.threshold,
            cell(reason)
        );
    }
    out
}

/// A row's result cell: the symbol for what the row did to the review, then
/// the model's verdict. ✅ is a counted row that passed its polarity, ❌ a
/// counted row that failed it, ❔ a row that doesn't count: below its
/// threshold, not applicable, or without an answer. So `match` is ✅ under a
/// positive rule and ❌ under a negative one.
fn result(r: &Row) -> &'static str {
    match (r.error.is_some(), r.counted, r.passed, r.verdict) {
        (true, ..) => "❔ no answer",
        (_, true, true, Verdict::Match) => "✅ match",
        (_, true, true, Verdict::NoMatch) => "✅ no match",
        (_, true, false, Verdict::Match) => "❌ match",
        (_, true, false, Verdict::NoMatch) => "❌ no match",
        (_, _, _, Verdict::Match) => "❔ match",
        (_, _, _, Verdict::NoMatch) => "❔ no match",
        (_, _, _, Verdict::NotApplicable) => "❔ unsure",
    }
}

/// The closing line of a review: the overall verdict, with its symbol.
fn overall_line(overall: Overall) -> &'static str {
    match overall {
        Overall::Approve => "**Overall: ✅ approve**\n",
        Overall::RequestChanges => "**Overall: ❌ request changes**\n",
    }
}

/// `text` as one table cell's inline text: on one line, with a backslash before
/// every character that could end the cell, start markup, math (`$`) or raw
/// HTML, and a zero-width space after every `@`, `#`, `!`, `://` and `www.`, so
/// it carries no live mention, issue or PR reference, or link. The forge links
/// issue and PR numbers and bare URLs itself, so the escapes alone don't stop
/// those; `#` and `!` get only the zero-width space.
///
/// `_` is escaped except between two alphanumerics, where it can't open
/// emphasis (`ci_metrics.rs`). Other punctuation, such as `(`, `)`, `.` and
/// `:`, passes through: it starts no markup inside a table cell, and the `[`
/// that would make `(` a link target is escaped.
fn cell(text: &str) -> String {
    let mut out = String::with_capacity(text.len());
    for (i, word) in text.split_whitespace().enumerate() {
        if i > 0 {
            out.push(' ');
        }
        let chars: Vec<char> = word.chars().collect();
        for (j, &c) in chars.iter().enumerate() {
            match c {
                '_' if j > 0
                    && chars[j - 1].is_alphanumeric()
                    && chars.get(j + 1).is_some_and(|n| n.is_alphanumeric()) =>
                {
                    out.push(c);
                }
                '\\' | '|' | '`' | '*' | '_' | '~' | '[' | ']' | '<' | '>' | '&' | '$' => {
                    out.push('\\');
                    out.push(c);
                }
                '#' | '!' => {
                    out.push(c);
                    out.push('\u{200b}');
                }
                '@' => out.push_str("@\u{200b}"),
                c => out.push(c),
            }
            let tail = out.get(out.len().saturating_sub(4)..).unwrap_or_default();
            if out.ends_with("://") || tail.eq_ignore_ascii_case("www.") {
                out.push('\u{200b}');
            }
        }
    }
    out
}

#[cfg(test)]
mod tests {
    use std::sync::Mutex;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::time::Duration;

    use axum::http::StatusCode;
    use swarm_wire::review_classifier::Polarity;

    use super::*;
    use crate::review_classifier::metrics::testing::Recorder;
    use crate::review_classifier::runs::tests::FakeDoc;
    use crate::review_classifier::tests::{Fake, rules_json};
    use crate::test_nats::TestNats;
    use tokio::task::JoinHandle;

    /// Every review [`run`] posted, with the commit it named.
    #[derive(Default)]
    struct Posted(Mutex<Vec<(String, Option<String>, Review)>>);

    impl ReviewPoster for Posted {
        async fn post(
            &self,
            owner: &str,
            repo: &str,
            number: i64,
            commit: Option<&str>,
            review: Review,
        ) -> anyhow::Result<Option<String>> {
            self.0.lock().expect("unpoisoned").push((
                format!("{owner}/{repo}#{number}"),
                commit.map(str::to_owned),
                review,
            ));
            Ok(Some(format!(
                "https://forge/{owner}/{repo}/pulls/{number}#review"
            )))
        }
    }

    fn row(error: Option<&str>, counted: bool, passed: bool) -> Row {
        Row {
            name: "datetime".to_owned(),
            polarity: Polarity::Positive,
            verdict: if error.is_some() {
                Verdict::NotApplicable
            } else {
                Verdict::Match
            },
            certainty: if error.is_some() { 0 } else { 90 },
            threshold: 80,
            counted,
            passed,
            reason: String::new(),
            error: error.map(str::to_owned),
        }
    }

    /// A row as the aggregation produces it for `polarity`, `verdict`,
    /// `certainty` against `threshold`.
    fn answered(polarity: Polarity, verdict: Verdict, certainty: u8, threshold: u8) -> Row {
        let counted = verdict != Verdict::NotApplicable && certainty >= threshold;
        let wanted = match polarity {
            Polarity::Positive => Verdict::Match,
            Polarity::Negative => Verdict::NoMatch,
        };
        Row {
            name: "no-ads".to_owned(),
            polarity,
            verdict,
            certainty,
            threshold,
            counted,
            passed: counted && verdict == wanted,
            reason: "Because.".to_owned(),
            error: None,
        }
    }

    #[test]
    fn a_row_shows_whether_it_passed_not_the_raw_verdict() {
        use Polarity::{Negative, Positive};
        use Verdict::{Match, NoMatch, NotApplicable};
        for (case, r, expected) in [
            (
                "positive match passes",
                answered(Positive, Match, 85, 65),
                "| ✅ match | no-ads | 85 / 65 | Because. |",
            ),
            (
                "negative match fails",
                answered(Negative, Match, 85, 65),
                "| ❌ match | no-ads | 85 / 65 | Because. |",
            ),
            (
                "negative no match passes",
                answered(Negative, NoMatch, 97, 80),
                "| ✅ no match | no-ads | 97 / 80 | Because. |",
            ),
            (
                "positive no match fails",
                answered(Positive, NoMatch, 97, 80),
                "| ❌ no match | no-ads | 97 / 80 | Because. |",
            ),
            (
                "not applicable is unsure",
                answered(Negative, NotApplicable, 98, 70),
                "| ❔ unsure | no-ads | 98 / 70 | Because. |",
            ),
            (
                "below the threshold does not count",
                answered(Positive, Match, 41, 60),
                "| ❔ match | no-ads | 41 / 60 | Because. |",
            ),
            (
                "no answer shows its error",
                Row {
                    error: Some("timed out".to_owned()),
                    verdict: NotApplicable,
                    certainty: 0,
                    counted: false,
                    passed: false,
                    reason: String::new(),
                    ..answered(Positive, Match, 0, 65)
                },
                "| ❔ no answer | no-ads | — / 65 | timed out |",
            ),
        ] {
            let body = table(&[r]);
            assert_eq!(
                body,
                format!(
                    "| result | check | certainty | reason |\n\
                     | --- | --- | --- | --- |\n{expected}\n"
                ),
                "{case}"
            );
        }
    }

    #[test]
    fn the_overall_line_carries_its_emoji() {
        assert_eq!(overall_line(Overall::Approve), "**Overall: ✅ approve**\n");
        assert_eq!(
            overall_line(Overall::RequestChanges),
            "**Overall: ❌ request changes**\n"
        );
    }

    fn request(name: &str) -> Requested {
        Requested {
            owner: "o".to_owned(),
            name: name.to_owned(),
            pr: 7,
            head: "abc123".to_owned(),
            requester: "alice".to_owned(),
        }
    }

    /// [`run`] on `req`'s PR as `forge` reads it.
    async fn run_read(
        forge: &(impl FileSource + PullSource),
        sink: &(impl RunSink + SettingsSource),
        poster: &impl ReviewPoster,
        req: &Requested,
    ) -> RunRecord {
        run(
            forge,
            read_pull(forge, req).await,
            sink,
            poster,
            &Metrics::off(),
            req,
        )
        .await
    }

    /// A JSON rules file holding `n` classifications.
    fn rules_of(n: usize) -> (&'static str, String) {
        let classifications: Vec<_> = (0..n)
            .map(|i| {
                serde_json::json!({
                    "name": format!("c{i}"),
                    "description": "All datetimes carry time zone information.",
                    "polarity": "positive",
                    "threshold": 80,
                })
            })
            .collect();
        let (path, _) = rules_json();
        (
            path,
            serde_json::json!({ "classifications": classifications }).to_string(),
        )
    }

    /// What one run against the fake forge posts.
    async fn posted(reply: Reply, repo: &str) -> (String, Option<String>, Review) {
        posted_for(reply, repo, rules_json()).await
    }

    /// What one run against the fake forge, holding `rules`, posts.
    async fn posted_for(
        reply: Reply,
        repo: &str,
        rules: (&'static str, String),
    ) -> (String, Option<String>, Review) {
        let fake = Fake::new(reply, vec![rules]);
        let poster = Posted::default();
        run_read(&fake, &fake, &poster, &request(repo)).await;
        let mut all = poster.0.into_inner().expect("unpoisoned");
        assert_eq!(all.len(), 1, "one review per run");
        all.remove(0)
    }

    #[tokio::test]
    async fn nine_timeouts_and_one_pass_do_not_approve() {
        let mut rows: Vec<Row> = (0..9)
            .map(|_| row(Some("timed out after 3 minutes"), false, false))
            .collect();
        rows.push(row(None, true, true));
        // The classifier's own verdict for these rows is approve.
        let reply = Reply::Done {
            rows,
            overall: Overall::Approve,
        };
        let (_, commit, review) = posted_for(reply, "r", rules_of(10)).await;
        assert_eq!(review.event, "COMMENT");
        assert!(
            review
                .body
                .starts_with("**No verdict:** 9 of 10 classifications got no answer."),
            "{}",
            review.body
        );
        assert!(review.body.contains("timed out after 3 minutes"));
        assert_eq!(commit.as_deref(), Some("abc123"));
    }

    #[tokio::test]
    async fn a_short_or_empty_answer_does_not_approve() {
        for (sent, answered) in [(3, 2), (1, 0), (1, 2)] {
            let reply = Reply::Done {
                rows: (0..answered).map(|_| row(None, true, true)).collect(),
                overall: Overall::Approve,
            };
            let (_, _, review) = posted_for(reply, "r", rules_of(sent)).await;
            assert_eq!(review.event, "COMMENT", "{answered} of {sent}");
            assert!(
                review.body.starts_with(&format!(
                    "**No verdict:** the classifier answered {answered} of {sent} classifications."
                )),
                "{}",
                review.body
            );
        }

        // A run of no classifications never reaches the classifier; an
        // empty answer to one still has no verdict.
        let ruling = review_for(&Outcome::Ran {
            sent: 0,
            reply: Reply::Done {
                rows: Vec::new(),
                overall: Overall::Approve,
            },
        });
        assert_eq!(ruling.event, Event::Comment);
    }

    /// A classifier that counts the runs reaching it and answers approve once
    /// [`Gate::open`] lets each through, beside a store holding `settings`.
    #[derive(Clone)]
    struct Gate {
        permits: Arc<tokio::sync::Semaphore>,
        reached: Arc<AtomicUsize>,
        settings: Arc<Mutex<Settings>>,
    }

    impl Gate {
        fn closed() -> Self {
            Self {
                permits: Arc::new(tokio::sync::Semaphore::new(0)),
                reached: Arc::default(),
                settings: Arc::default(),
            }
        }

        fn open(&self, runs: usize) {
            self.permits.add_permits(runs);
        }

        fn reached(&self) -> usize {
            self.reached.load(Ordering::SeqCst)
        }

        fn store(&self, max_concurrent_reviews: usize) {
            self.settings
                .lock()
                .expect("unpoisoned")
                .max_concurrent_reviews = max_concurrent_reviews;
        }
    }

    impl RunSink for Gate {
        async fn run(&self, _request: Vec<u8>) -> Reply {
            self.reached.fetch_add(1, Ordering::SeqCst);
            let _permit = self.permits.acquire().await.expect("never closed");
            Reply::Done {
                rows: vec![row(None, true, true)],
                overall: Overall::Approve,
            }
        }
    }

    impl SettingsSource for Gate {
        async fn settings(&self) -> anyhow::Result<Settings> {
            Ok(self.settings.lock().expect("unpoisoned").clone())
        }
    }

    fn requested_at(head: &str) -> Requested {
        Requested {
            head: head.to_owned(),
            ..request("r")
        }
    }

    /// Record in `store` a review of o/r#7 at `abc123` that ended as
    /// `outcome`, with an answer as a run that got that far keeps it.
    async fn reviewed(store: &Runs, id: &str, outcome: RunOutcome) -> RunRecord {
        let earlier = RunRecord {
            trigger: Trigger::Review,
            outcome,
            rows: Some(vec![row(
                None,
                true,
                outcome != RunOutcome::ChangesRequested,
            )]),
            overall: Some(match outcome {
                RunOutcome::ChangesRequested => Overall::RequestChanges,
                _ => Overall::Approve,
            }),
            ..crate::review_classifier::runs::tests::record(id)
        };
        store.record(earlier.clone()).await;
        earlier
    }

    /// The fake forge's files, and each PR's head on o/r: `abc123` until
    /// [`Heads::move_to`] moves it.
    struct Heads {
        fake: Fake,
        heads: Mutex<std::collections::HashMap<i64, String>>,
    }

    impl Heads {
        fn new() -> Self {
            Self {
                fake: Fake::new(
                    Reply::Failed {
                        error: String::new(),
                    },
                    vec![rules_json()],
                ),
                heads: Mutex::default(),
            }
        }

        fn move_to(&self, pr: i64, head: &str) {
            self.heads
                .lock()
                .expect("unpoisoned")
                .insert(pr, head.to_owned());
        }
    }

    impl PullSource for Heads {
        async fn pull(
            &self,
            owner: &str,
            repo: &str,
            number: i64,
        ) -> anyhow::Result<PullForReview> {
            anyhow::ensure!((owner, repo) == ("o", "r"), "no such repo");
            let head = self.heads.lock().expect("unpoisoned").get(&number).cloned();
            Ok(PullForReview {
                head_sha: head.unwrap_or_else(|| "abc123".to_owned()),
                commits: Vec::new(),
                diff: String::new(),
                narrowed: None,
            })
        }
    }

    impl FileSource for Heads {
        async fn default_branch_files(
            &self,
            owner: &str,
            repo: &str,
            paths: &[&str],
        ) -> anyhow::Result<Option<forge::DefaultBranchFiles>> {
            self.fake.default_branch_files(owner, repo, paths).await
        }
    }

    /// A fake forge, a closed [`Gate`], a poster and a run store, and a
    /// starter for reviews of a head on o/r#7.
    struct Rig {
        reviews: Arc<Reviews>,
        forge: Arc<Heads>,
        gate: Gate,
        poster: Arc<Posted>,
        doc: Arc<FakeDoc>,
        store: Runs,
    }

    impl Rig {
        fn new() -> Self {
            Self::with_doc(Arc::default())
        }

        fn with_doc(doc: Arc<FakeDoc>) -> Self {
            Self {
                reviews: Arc::new(Reviews::default()),
                forge: Arc::new(Heads::new()),
                gate: Gate::closed(),
                poster: Arc::new(Posted::default()),
                store: Runs::Fake(Arc::clone(&doc)),
                doc,
            }
        }

        /// [`review`] of a request at `head`, once it has a run place.
        fn start_at(&self, head: &str) -> JoinHandle<()> {
            let reviews = Arc::clone(&self.reviews);
            let forge = Arc::clone(&self.forge);
            let gate = self.gate.clone();
            let poster = Arc::clone(&self.poster);
            let store = self.store.clone();
            let request = requested_at(head);
            tokio::spawn(async move {
                let slot = reviews.slot().await;
                review(
                    &*forge,
                    &gate,
                    &*poster,
                    &store,
                    &Metrics::off(),
                    &request,
                    slot,
                )
                .await;
            })
        }

        fn posted(&self) -> Vec<(String, Option<String>, Review)> {
            self.poster.0.lock().expect("unpoisoned").clone()
        }
    }

    /// Await `task`, a re-post. A task that ran instead waits on the closed
    /// [`Gate`], so this fails rather than hang.
    async fn reposted(task: JoinHandle<()>) {
        tokio::time::timeout(Duration::from_secs(5), task)
            .await
            .expect("re-posted without waiting for a run place")
            .expect("re-posted");
    }

    #[tokio::test]
    async fn a_request_on_a_head_with_a_clean_verdict_posts_it_again_without_a_run() {
        for (outcome, event, overall) in [
            (
                RunOutcome::Approved,
                "APPROVED",
                "**Overall: ✅ approve**\n",
            ),
            (
                RunOutcome::ChangesRequested,
                "REQUEST_CHANGES",
                "**Overall: ❌ request changes**\n",
            ),
        ] {
            let rig = Rig::new();
            let earlier = reviewed(&rig.store, "earlier", outcome).await;

            reposted(rig.start_at("abc123")).await;

            assert_eq!(
                rig.gate.reached(),
                0,
                "{event}: no run reached the classifier"
            );
            let posted = rig.posted();
            assert_eq!(posted.len(), 1, "{event}");
            let (target, commit, review) = &posted[0];
            assert_eq!(target, "o/r#7");
            assert_eq!(commit.as_deref(), Some("abc123"));
            assert_eq!(review.event, event);
            assert!(
                review
                    .body
                    .starts_with("Already reviewed `abc123`, verdict unchanged.\n\n"),
                "{}",
                review.body
            );
            assert!(
                review.body.contains("| datetime | 90 / 80 |"),
                "{}",
                review.body
            );
            assert!(review.body.ends_with(overall), "{}", review.body);

            let recorded = rig.doc.runs();
            assert_eq!(recorded.len(), 2, "{event}");
            assert_eq!(recorded[0].outcome, outcome);
            assert_eq!(recorded[0].head_sha.as_deref(), Some("abc123"));
            assert_eq!(recorded[0].rows, earlier.rows);
            assert_eq!(
                recorded[0].detail.as_deref(),
                Some("a re-post of run earlier's verdict")
            );
        }
    }

    #[tokio::test]
    async fn a_request_on_a_head_without_a_clean_verdict_runs_again() {
        for (case, seed) in [
            ("no verdict", vec![RunOutcome::NoVerdict]),
            ("post failed", vec![RunOutcome::PostFailed]),
            (
                "an approval since followed by no verdict",
                vec![RunOutcome::Approved, RunOutcome::NoVerdict],
            ),
        ] {
            let rig = Rig::new();
            for (i, outcome) in seed.into_iter().enumerate() {
                reviewed(&rig.store, &format!("earlier-{i}"), outcome).await;
            }

            let task = rig.start_at("abc123");
            rig.gate.open(1);
            task.await.expect("ran");

            assert_eq!(rig.gate.reached(), 1, "{case}: a fresh run");
            let posted = rig.posted();
            assert_eq!(posted.len(), 1, "{case}");
            assert_eq!(posted[0].2.event, "APPROVED", "{case}");
            assert!(
                !posted[0].2.body.starts_with("Already reviewed"),
                "{case}: {}",
                posted[0].2.body
            );
        }
    }

    #[tokio::test]
    async fn only_a_review_of_the_same_head_is_posted_again() {
        let approved = reviewed(&Runs::Fake(Arc::default()), "e", RunOutcome::Approved).await;
        for (case, earlier) in [
            (
                "a try-out",
                RunRecord {
                    trigger: Trigger::TryOut,
                    ..approved.clone()
                },
            ),
            (
                "another head",
                RunRecord {
                    head_sha: Some("def456".to_owned()),
                    ..approved.clone()
                },
            ),
            (
                "another PR",
                RunRecord {
                    pr: 8,
                    ..approved.clone()
                },
            ),
        ] {
            let rig = Rig::new();
            rig.store.record(earlier).await;

            let task = rig.start_at("abc123");
            rig.gate.open(1);
            task.await.expect("ran");
            assert_eq!(rig.gate.reached(), 1, "{case}: a fresh run");
        }
    }

    /// The queue's timing in the tests: short enough that a request no one
    /// acks comes back within a test.
    const TEST_TIMING: Timing = Timing {
        ack_wait: Duration::from_secs(1),
        pull_wait: Duration::from_millis(200),
        busy_retry: Duration::from_millis(200),
    };

    /// A [`Rig`] behind a real queue: requests go in through [`accept`], and
    /// [`consume`] takes them.
    struct Queued {
        _nats: TestNats,
        client: async_nats::Client,
        rig: Rig,
        recorder: Recorder,
    }

    impl Queued {
        async fn new() -> Self {
            let nats = TestNats::start();
            let client = nats.client().await;
            Self {
                _nats: nats,
                client,
                rig: Rig::new(),
                recorder: Recorder::new(),
            }
        }

        /// Queue a request for o/r#`pr` at `head`.
        async fn request(&self, pr: i64, head: &str) {
            let request = Requested {
                pr,
                ..requested_at(head)
            };
            accept(
                Some(&self.client),
                &*self.rig.poster,
                &self.rig.store,
                &self.recorder.metrics,
                &request,
            )
            .await;
            assert!(self.rig.posted().is_empty(), "queued without a comment");
        }

        /// A controller taking requests, its run places in `reviews`.
        fn consume(&self, reviews: &Arc<Reviews>) -> JoinHandle<()> {
            let client = self.client.clone();
            let reviews = Arc::clone(reviews);
            let forge = Arc::clone(&self.rig.forge);
            let gate = self.rig.gate.clone();
            let poster = Arc::clone(&self.rig.poster);
            let store = self.rig.store.clone();
            let metrics = self.recorder.metrics.clone();
            tokio::spawn(async move {
                let handles = Handles {
                    forge,
                    queue: gate,
                    poster,
                    runs: store,
                    metrics,
                };
                consume(&client, TEST_TIMING, &reviews, handles).await;
            })
        }

        /// How many requests the queue holds, taken or not.
        async fn waiting(&self) -> u64 {
            jetstream::new(self.client.clone())
                .get_stream(REQUEST_STREAM)
                .await
                .expect("the stream")
                .info()
                .await
                .expect("its info")
                .state
                .messages
        }

        async fn consumer_info(&self) -> jetstream::consumer::Info {
            let js = jetstream::new(self.client.clone());
            let stream = js.get_stream(REQUEST_STREAM).await.expect("the stream");
            let mut consumer: PullConsumer = stream
                .get_consumer(REQUEST_CONSUMER)
                .await
                .expect("the consumer");
            consumer.info().await.expect("its info").clone()
        }
    }

    /// Wait until `done` holds, for at most ten seconds.
    async fn until(what: &str, done: impl Fn() -> bool) {
        tokio::time::timeout(Duration::from_secs(10), async {
            while !done() {
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
        })
        .await
        .unwrap_or_else(|_| panic!("never: {what}"));
    }

    #[tokio::test]
    async fn a_queued_request_runs_after_a_restart() {
        let q = Queued::new().await;
        q.request(7, "abc123").await;
        let first = q.consume(&q.rig.reviews);
        until("the first controller's run started", || {
            q.rig.gate.reached() == 1
        })
        .await;
        first.abort();
        assert!(first.await.expect_err("stopped").is_cancelled());
        assert!(q.rig.posted().is_empty());

        // A restarted controller has run places of its own.
        let _second = q.consume(&Arc::new(Reviews::default()));
        until("the request ran again", || q.rig.gate.reached() == 2).await;
        q.rig.gate.open(1);
        until("its review is posted", || q.rig.posted().len() == 1).await;
        let posted = q.rig.posted();
        assert_eq!(posted[0].0, "o/r#7");
        assert_eq!(posted[0].2.event, "APPROVED");
        until("its record is written", || q.rig.doc.runs().len() == 1).await;
        tokio::time::sleep(TEST_TIMING.ack_wait * 2).await;
        assert_eq!(q.waiting().await, 0, "the finished request is acked");
        assert_eq!(q.rig.gate.reached(), 2, "no run after the ack");
    }

    #[tokio::test]
    async fn a_newer_request_for_a_pr_replaces_the_waiting_one() {
        let q = Queued::new().await;
        q.rig.forge.move_to(7, "new");
        q.request(7, "old").await;
        q.request(8, "abc123").await;
        q.request(7, "new").await;
        assert_eq!(q.waiting().await, 2, "one request per PR");

        let _consumer = q.consume(&q.rig.reviews);
        q.rig.gate.open(2);
        until("both PRs are reviewed", || q.rig.posted().len() == 2).await;
        tokio::time::sleep(TEST_TIMING.ack_wait * 2).await;
        let mut posted: Vec<_> = q
            .rig
            .posted()
            .into_iter()
            .map(|(target, commit, _)| (target, commit))
            .collect();
        posted.sort();
        assert_eq!(
            posted,
            [
                ("o/r#7".to_owned(), Some("new".to_owned())),
                ("o/r#8".to_owned(), Some("abc123".to_owned())),
            ]
        );
        assert_eq!(q.rig.gate.reached(), 2, "one run per PR");
    }

    #[tokio::test]
    async fn a_request_for_a_head_the_pr_has_moved_on_from_reviews_the_new_head() {
        let q = Queued::new().await;
        // `old` has an approval on record; the PR has since moved to `new`.
        let approved = reviewed(&q.rig.store, "earlier", RunOutcome::Approved).await;
        q.rig
            .store
            .record(RunRecord {
                id: "earlier-old".to_owned(),
                head_sha: Some("old".to_owned()),
                ..approved
            })
            .await;
        q.rig.forge.move_to(7, "new");
        q.request(7, "old").await;
        let _consumer = q.consume(&q.rig.reviews);
        until("the run reached the classifier", || {
            q.rig.gate.reached() == 1
        })
        .await;
        q.rig.gate.open(1);
        until("the new head is reviewed", || q.rig.posted().len() == 1).await;
        let posted = q.rig.posted();
        assert_eq!(posted[0].1.as_deref(), Some("new"));
        assert!(
            !posted[0].2.body.starts_with("Already reviewed"),
            "the old head's verdict was posted again: {}",
            posted[0].2.body
        );

        // Control: with the PR back on `old`, the same request re-posts its
        // recorded verdict without a run.
        q.rig.forge.move_to(7, "old");
        accept(
            Some(&q.client),
            &*q.rig.poster,
            &q.rig.store,
            &Metrics::off(),
            &requested_at("old"),
        )
        .await;
        until("the verdict is posted again", || q.rig.posted().len() == 2).await;
        let posted = q.rig.posted();
        assert_eq!(posted[1].1.as_deref(), Some("old"));
        assert!(posted[1].2.body.starts_with("Already reviewed `old`"));
        assert_eq!(q.rig.gate.reached(), 1, "the re-post ran nothing");
    }

    #[tokio::test]
    async fn nothing_is_refused_while_every_run_place_is_taken() {
        let q = Queued::new().await;
        let prs = 1..=12;
        for pr in prs.clone() {
            q.request(pr, "abc123").await;
        }
        let _consumer = q.consume(&q.rig.reviews);
        let limit = Settings::default().max_concurrent_reviews;
        until("every place is taken", || q.rig.gate.reached() == limit).await;
        tokio::time::sleep(TEST_TIMING.ack_wait * 2).await;
        assert_eq!(q.rig.gate.reached(), limit, "the rest wait");
        assert!(q.rig.posted().is_empty(), "nothing is refused");
        assert_eq!(q.waiting().await, 12);

        q.rig.gate.open(12);
        until("every PR is reviewed", || q.rig.posted().len() == 12).await;
        let mut targets: Vec<_> = q.rig.posted().into_iter().map(|(t, ..)| t).collect();
        targets.sort();
        let mut wanted: Vec<_> = prs.map(|pr| format!("o/r#{pr}")).collect();
        wanted.sort();
        assert_eq!(targets, wanted);
        assert!(q.rig.posted().iter().all(|(_, _, r)| r.event == "APPROVED"));
        assert_eq!(q.rig.gate.reached(), 12);
    }

    #[tokio::test]
    async fn a_run_longer_than_the_ack_wait_keeps_its_request() {
        let q = Queued::new().await;
        q.request(7, "abc123").await;
        let _consumer = q.consume(&q.rig.reviews);
        until("the run started", || q.rig.gate.reached() == 1).await;
        tokio::time::sleep(TEST_TIMING.ack_wait * 3).await;
        let info = q.consumer_info().await;
        assert_eq!(info.num_ack_pending, 1, "the request is in hand");
        assert_eq!(info.num_redelivered, 0, "and was never delivered again");
        q.rig.gate.open(1);
        until("its review is posted", || q.rig.posted().len() == 1).await;
        assert_eq!(q.rig.gate.reached(), 1);
    }

    #[tokio::test]
    async fn a_request_for_a_pr_whose_review_is_going_waits_for_it() {
        let q = Queued::new().await;
        q.request(7, "abc123").await;
        let _consumer = q.consume(&q.rig.reviews);
        until("the run started", || q.rig.gate.reached() == 1).await;
        q.request(7, "abc123").await;
        tokio::time::sleep(TEST_TIMING.ack_wait * 2).await;
        assert_eq!(q.rig.gate.reached(), 1, "one run on the PR at once");

        // The run approved, so the waiting request posts its verdict again.
        q.rig.gate.open(1);
        until("both are posted", || q.rig.posted().len() == 2).await;
        let posted = q.rig.posted();
        assert_eq!(posted[0].2.event, "APPROVED");
        assert!(
            posted[1].2.body.starts_with("Already reviewed `abc123`"),
            "{}",
            posted[1].2.body
        );
        let requests = q.recorder.counter(REQUESTS);
        assert_eq!(requests.get("outcome=queued"), Some(&2), "{requests:?}");
        assert_eq!(requests.get("outcome=started"), Some(&1), "{requests:?}");
        assert_eq!(requests.get("outcome=reposted"), Some(&1), "{requests:?}");
        assert!(
            requests.get("outcome=deferred").is_some_and(|n| *n >= 1),
            "the second request waited at least once: {requests:?}"
        );
        assert_eq!(q.rig.gate.reached(), 1);
    }

    #[tokio::test]
    async fn a_request_that_does_not_parse_is_dropped() {
        let q = Queued::new().await;
        q.request(8, "abc123").await;
        jetstream::new(q.client.clone())
            .publish(format!("{REQUEST_SUBJECT_PREFIX}.o.r.7"), "{}".into())
            .await
            .expect("published")
            .await
            .expect("stored");
        let _consumer = q.consume(&q.rig.reviews);
        q.rig.gate.open(2);
        until("the request that parses is reviewed", || {
            q.rig.posted().len() == 1
        })
        .await;
        tokio::time::timeout(Duration::from_secs(10), async {
            while q.waiting().await > 0 {
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
        })
        .await
        .expect("both leave the queue");
        assert_eq!(q.rig.posted()[0].0, "o/r#8");
        assert_eq!(q.rig.gate.reached(), 1);
        let requests = q.recorder.counter(REQUESTS);
        assert_eq!(requests.get("outcome=unparsable"), Some(&1), "{requests:?}");
    }

    #[tokio::test]
    async fn a_request_that_cannot_be_queued_posts_a_no_verdict_comment() {
        let rig = Rig::new();
        accept(
            None,
            &*rig.poster,
            &rig.store,
            &Metrics::off(),
            &requested_at("abc123"),
        )
        .await;
        let posted = rig.posted();
        assert_eq!(posted.len(), 1);
        let (target, commit, review) = &posted[0];
        assert_eq!(target, "o/r#7");
        assert_eq!(*commit, None);
        assert_eq!(review.event, "COMMENT");
        assert!(
            review.body.starts_with(
                "**No verdict:** the request could not be queued: no swarm queue is configured \
                 on this host; remove the review request and request it again."
            ),
            "{}",
            review.body
        );
        let recorded = rig.doc.runs();
        assert_eq!(recorded.len(), 1);
        assert_eq!(recorded[0].outcome, RunOutcome::NoVerdict);
        assert_eq!(recorded[0].head_sha, None);
    }

    #[test]
    fn each_pr_has_a_subject_of_its_own() {
        let at = |owner: &str, name: &str, pr| {
            Requested {
                owner: owner.to_owned(),
                name: name.to_owned(),
                pr,
                ..request("r")
            }
            .subject()
        };
        assert_eq!(
            at("hyperhive", "hyperhive", 5219),
            "$SWARM.review-request.hyperhive.hyperhive.5219"
        );
        assert_eq!(
            at("a.b", "c", 1),
            "$SWARM.review-request.a_2eb.c.1",
            "a dot stays inside its token"
        );
        assert_ne!(at("a.b", "c", 1), at("a", "b.c", 1));
        assert_ne!(at("a_2eb", "c", 1), at("a.b", "c", 1));
        assert_ne!(at("o", "r", 1), at("o", "r", 11));
        for subject in [at("a.b", "c d*>", 1), at("ü", "r", 1)] {
            assert_eq!(subject.split('.').count(), 5, "{subject}");
            assert!(
                !subject.contains(['*', '>', ' ']),
                "{subject} is a literal subject"
            );
        }
    }

    #[tokio::test]
    async fn a_new_limit_applies_to_the_next_run_without_a_restart() {
        let q = Queued::new().await;
        let reviews = &q.rig.reviews;
        let gate = &q.rig.gate;

        // The stored limit is read before a run waits, not the default 2.
        gate.store(1);
        q.request(1, "abc123").await;
        q.request(2, "abc123").await;
        let _consumer = q.consume(reviews);
        until("the first run started", || gate.reached() == 1).await;
        tokio::time::sleep(TEST_TIMING.ack_wait).await;
        assert_eq!(gate.reached(), 1);

        // Raised, as a PUT does: the waiting run starts at once.
        gate.store(3);
        reviews.resize(3);
        until("the waiting run started", || gate.reached() == 2).await;
        q.request(3, "abc123").await;
        until("a third run started", || gate.reached() == 3).await;

        // Lowered: the three keep running and the next one waits.
        gate.store(1);
        q.request(4, "abc123").await;
        tokio::time::sleep(TEST_TIMING.ack_wait).await;
        assert_eq!(gate.reached(), 3);

        gate.open(4);
        until("every run is posted", || q.rig.posted().len() == 4).await;
        assert_eq!(gate.reached(), 4);
        until("every place is freed", || reviews.slots().running <= 1).await;
    }

    /// The record of one run against the fake forge, holding `rules`, posted
    /// by `poster`.
    async fn recorded_for(
        reply: Reply,
        repo: &str,
        rules: (&'static str, String),
        poster: &impl ReviewPoster,
    ) -> RunRecord {
        let fake = Fake::new(reply, vec![rules]);
        run_read(&fake, &fake, poster, &request(repo)).await
    }

    #[tokio::test]
    async fn each_way_a_run_ends_has_its_outcome() {
        let answered = |rows: Vec<Row>, overall| Reply::Done { rows, overall };
        let failed = Reply::Failed {
            error: "no API key".to_owned(),
        };
        for (case, reply, repo, rules, outcome, detail) in [
            (
                "approve",
                answered(vec![row(None, true, true)], Overall::Approve),
                "r",
                rules_json(),
                RunOutcome::Approved,
                None,
            ),
            (
                "request changes",
                answered(vec![row(None, true, false)], Overall::RequestChanges),
                "r",
                rules_json(),
                RunOutcome::ChangesRequested,
                None,
            ),
            (
                "a failed run",
                failed.clone(),
                "r",
                rules_json(),
                RunOutcome::NoVerdict,
                Some("the run failed: no API key"),
            ),
            (
                "a short answer",
                answered(vec![row(None, true, true)], Overall::Approve),
                "r",
                rules_of(2),
                RunOutcome::NoVerdict,
                Some("the classifier answered 1 of 2 classifications"),
            ),
            (
                "an unanswered classification",
                answered(vec![row(Some("timed out"), false, false)], Overall::Approve),
                "r",
                rules_json(),
                RunOutcome::NoVerdict,
                Some("1 of 1 classifications got no answer"),
            ),
            (
                "an unreadable PR",
                answered(vec![row(None, true, true)], Overall::Approve),
                "elsewhere",
                rules_json(),
                RunOutcome::NoVerdict,
                Some("reading the PR failed"),
            ),
        ] {
            let run = recorded_for(reply.clone(), repo, rules, &Posted::default()).await;
            assert_eq!(run.outcome, outcome, "{case}");
            assert_eq!(run.trigger, Trigger::Review, "{case}");
            assert_eq!(run.requester, "alice", "{case}");
            match detail {
                None => assert_eq!(run.detail, None, "{case}"),
                Some(want) => assert!(
                    run.detail.as_deref().is_some_and(|d| d.starts_with(want)),
                    "{case}: {:?}",
                    run.detail
                ),
            }
            assert_eq!(
                run.review_url.as_deref(),
                Some(format!("https://forge/o/{repo}/pulls/7#review").as_str()),
                "{case}"
            );
            if repo == "elsewhere" {
                assert_eq!(run.head_sha, None, "{case}");
                assert_eq!((run.rows, run.overall), (None, None), "{case}");
            } else {
                assert_eq!(run.head_sha.as_deref(), Some("abc123"), "{case}");
                assert_eq!((run.rows, run.overall), runs::answer(&reply), "{case}");
            }
        }

        let [yml, _] = crate::review_classifier::RULES_PATHS;
        let fake = Fake::new(failed, vec![rules_json(), (yml, String::new())]);
        let run = run_read(&fake, &fake, &Posted::default(), &request("r")).await;
        assert_eq!(run.outcome, RunOutcome::NoVerdict);
        assert!(
            run.detail
                .as_deref()
                .is_some_and(|d| d.starts_with("the rules did not load")),
            "{:?}",
            run.detail
        );
    }

    /// A forge that refuses every review.
    struct Refusing;

    impl ReviewPoster for Refusing {
        async fn post(
            &self,
            _owner: &str,
            _repo: &str,
            _number: i64,
            _commit: Option<&str>,
            _review: Review,
        ) -> anyhow::Result<Option<String>> {
            anyhow::bail!("the forge answered 500")
        }
    }

    #[tokio::test]
    async fn a_failed_post_is_post_failed_with_its_reason() {
        let approve = Reply::Done {
            rows: vec![row(None, true, true)],
            overall: Overall::Approve,
        };
        let run = recorded_for(approve, "r", rules_json(), &Refusing).await;
        assert_eq!(run.outcome, RunOutcome::PostFailed);
        assert_eq!(
            run.detail.as_deref(),
            Some("posting the review failed: the forge answered 500")
        );
        assert_eq!(run.review_url, None);

        let failed = Reply::Failed {
            error: "no API key".to_owned(),
        };
        let run = recorded_for(failed, "r", rules_json(), &Refusing).await;
        assert_eq!(run.outcome, RunOutcome::PostFailed);
        assert_eq!(
            run.detail.as_deref(),
            Some("the run failed: no API key; posting the review failed: the forge answered 500")
        );
    }

    /// The forge client's error for a port nothing listens on.
    async fn refused_by_port() -> ForgejoError {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
        drop(listener);
        let forge = forgejo_api::Forgejo::new(forgejo_api::Auth::Token("stub-token"), url).unwrap();
        forge.user_get_current().await.expect_err("nothing listens")
    }

    /// A refused connection as the forge client reports it.
    async fn forge_refused() -> anyhow::Error {
        anyhow::Error::from(refused_by_port().await).context("posting the review")
    }

    /// A refused connection as the store client reports it: the
    /// `reqwest::Error` itself in the chain, under the login's context.
    async fn store_refused() -> anyhow::Error {
        let ForgejoError::ReqwestError(e) = refused_by_port().await else {
            panic!("a connect error is a reqwest error");
        };
        anyhow::Error::new(e)
            .context("An error occurred with the request")
            .context("logging in to the swarm secret store")
    }

    /// A poster that fails with each of `failures` in turn, then posts.
    struct Flaky {
        failures: Mutex<Vec<anyhow::Error>>,
        tries: AtomicUsize,
        posted: Posted,
    }

    impl Flaky {
        fn new(mut failures: Vec<anyhow::Error>) -> Self {
            failures.reverse();
            Self {
                failures: Mutex::new(failures),
                tries: AtomicUsize::new(0),
                posted: Posted::default(),
            }
        }
    }

    impl ReviewPoster for Flaky {
        async fn post(
            &self,
            owner: &str,
            repo: &str,
            number: i64,
            commit: Option<&str>,
            review: Review,
        ) -> anyhow::Result<Option<String>> {
            self.tries.fetch_add(1, Ordering::SeqCst);
            let failure = self.failures.lock().expect("unpoisoned").pop();
            match failure {
                Some(e) => Err(e),
                None => self.posted.post(owner, repo, number, commit, review).await,
            }
        }
    }

    #[tokio::test]
    async fn a_post_refused_a_connection_is_tried_again() {
        let failures = vec![forge_refused().await, store_refused().await];
        tokio::time::pause();
        let approve = Reply::Done {
            rows: vec![row(None, true, true)],
            overall: Overall::Approve,
        };
        let fake = Fake::new(approve, vec![rules_json()]);
        let poster = Flaky::new(failures);
        let started = tokio::time::Instant::now();
        let run = run_read(&fake, &fake, &poster, &request("r")).await;
        assert_eq!(run.outcome, RunOutcome::Approved, "{:?}", run.detail);
        assert_eq!(poster.tries.load(Ordering::SeqCst), 3);
        assert_eq!(poster.posted.0.lock().expect("unpoisoned").len(), 1);
        assert_eq!(fake.sent.lock().expect("unpoisoned").len(), 1, "one run");
        let waited = started.elapsed();
        assert!(
            waited >= POST_RETRY_WAITS[0] + POST_RETRY_WAITS[1] && waited < POST_RETRY_WAITS[2],
            "{waited:?}"
        );
    }

    /// A secret store that refuses every connection until `up`.
    struct Outage {
        up: tokio::time::Instant,
        refusals: Mutex<Vec<anyhow::Error>>,
        tries: AtomicUsize,
        posted: Posted,
    }

    impl ReviewPoster for Outage {
        async fn post(
            &self,
            owner: &str,
            repo: &str,
            number: i64,
            commit: Option<&str>,
            review: Review,
        ) -> anyhow::Result<Option<String>> {
            self.tries.fetch_add(1, Ordering::SeqCst);
            if tokio::time::Instant::now() < self.up {
                return Err(self
                    .refusals
                    .lock()
                    .expect("unpoisoned")
                    .pop()
                    .expect("a refusal per try"));
            }
            self.posted.post(owner, repo, number, commit, review).await
        }
    }

    #[tokio::test]
    async fn a_post_outlasts_a_secret_store_restart() {
        let mut refusals = Vec::new();
        for _ in 0..=POST_RETRY_WAITS.len() {
            refusals.push(store_refused().await);
        }
        tokio::time::pause();
        let approve = Reply::Done {
            rows: vec![row(None, true, true)],
            overall: Overall::Approve,
        };
        let fake = Fake::new(approve, vec![rules_json()]);
        let started = tokio::time::Instant::now();
        // swarm-bao was down 3m19s on 2026-10-14.
        let down = Duration::from_secs(200);
        let poster = Outage {
            up: started + down,
            refusals: Mutex::new(refusals),
            tries: AtomicUsize::new(0),
            posted: Posted::default(),
        };
        let run = run_read(&fake, &fake, &poster, &request("r")).await;
        assert_eq!(run.outcome, RunOutcome::Approved, "{:?}", run.detail);
        assert_eq!(poster.posted.0.lock().expect("unpoisoned").len(), 1);
        assert_eq!(fake.sent.lock().expect("unpoisoned").len(), 1, "one run");
        assert!(poster.tries.load(Ordering::SeqCst) > 1);
        assert!(started.elapsed() >= down, "{:?}", started.elapsed());
    }

    /// A forge whose first post never answers.
    #[derive(Default)]
    struct HangsOnce {
        tries: AtomicUsize,
        posted: Posted,
    }

    impl ReviewPoster for HangsOnce {
        async fn post(
            &self,
            owner: &str,
            repo: &str,
            number: i64,
            commit: Option<&str>,
            review: Review,
        ) -> anyhow::Result<Option<String>> {
            if self.tries.fetch_add(1, Ordering::SeqCst) == 0 {
                std::future::pending::<()>().await;
            }
            self.posted.post(owner, repo, number, commit, review).await
        }
    }

    #[tokio::test(start_paused = true)]
    async fn a_post_with_no_answer_is_abandoned_at_its_deadline_and_tried_again() {
        let approve = Reply::Done {
            rows: vec![row(None, true, true)],
            overall: Overall::Approve,
        };
        let fake = Fake::new(approve, vec![rules_json()]);
        let poster = HangsOnce::default();
        let started = tokio::time::Instant::now();
        let run = tokio::time::timeout(
            Duration::from_hours(1),
            run_read(&fake, &fake, &poster, &request("r")),
        )
        .await
        .expect("the hung try is abandoned");
        assert_eq!(run.outcome, RunOutcome::Approved, "{:?}", run.detail);
        assert_eq!(poster.tries.load(Ordering::SeqCst), 2);
        assert_eq!(poster.posted.0.lock().expect("unpoisoned").len(), 1);
        assert!(started.elapsed() >= forge::FORGE_DEADLINE + POST_RETRY_WAITS[0]);
    }

    #[tokio::test(start_paused = true)]
    async fn a_post_the_forge_refused_is_not_tried_again() {
        let approve = Reply::Done {
            rows: vec![row(None, true, true)],
            overall: Overall::Approve,
        };
        let fake = Fake::new(approve, vec![rules_json()]);
        let poster = Flaky::new(vec![
            ForgejoError::UnexpectedStatusCode(StatusCode::FORBIDDEN).into(),
        ]);
        let started = tokio::time::Instant::now();
        let run = run_read(&fake, &fake, &poster, &request("r")).await;
        assert_eq!(run.outcome, RunOutcome::PostFailed);
        assert_eq!(poster.tries.load(Ordering::SeqCst), 1);
        assert!(poster.posted.0.lock().expect("unpoisoned").is_empty());
        assert_eq!(started.elapsed(), Duration::ZERO);
    }

    #[tokio::test]
    async fn a_post_refused_every_try_is_post_failed() {
        let mut failures = Vec::new();
        for _ in 0..=POST_RETRY_WAITS.len() {
            failures.push(store_refused().await);
        }
        tokio::time::pause();
        let approve = Reply::Done {
            rows: vec![row(None, true, true)],
            overall: Overall::Approve,
        };
        let fake = Fake::new(approve, vec![rules_json()]);
        let poster = Flaky::new(failures);
        let run = run_read(&fake, &fake, &poster, &request("r")).await;
        assert_eq!(run.outcome, RunOutcome::PostFailed);
        assert_eq!(
            poster.tries.load(Ordering::SeqCst),
            POST_RETRY_WAITS.len() + 1
        );
        assert!(poster.posted.0.lock().expect("unpoisoned").is_empty());
        assert!(
            run.detail.as_deref().is_some_and(|d| d
                .starts_with("posting the review failed: logging in to the swarm secret store")),
            "{:?}",
            run.detail
        );
    }

    #[tokio::test]
    async fn a_failed_record_write_still_posts_the_review() {
        let rig = Rig::with_doc(Arc::new(FakeDoc::failing()));
        let task = rig.start_at("abc123");
        rig.gate.open(1);
        task.await.expect("ran to the end");
        let all = rig.posted();
        assert_eq!(all.len(), 1);
        assert_eq!(all[0].2.event, "APPROVED");
    }

    #[tokio::test]
    async fn a_review_runs_on_the_stored_runtime_model_and_effort() {
        use swarm_wire::agent_inference::Effort;

        let mut fake = Fake::new(
            Reply::Done {
                rows: vec![row(None, true, true)],
                overall: Overall::Approve,
            },
            vec![rules_json()],
        );
        let acp = swarm_wire::review_classifier::Acp {
            base_url: "https://llm.example/v1".to_owned(),
            model: "qwen".to_owned(),
            prices: swarm_wire::review_classifier::Prices {
                input: 1.0,
                output: 2.0,
                cache_read: None,
                cache_write: None,
            },
        };
        fake.settings = Settings {
            model: "sonnet".to_owned(),
            effort: Some(Effort::Medium),
            max_concurrent_reviews: 2,
            runtime: super::super::settings::RuntimeKind::Acp,
            acp: Some(acp.clone()),
        };
        let poster = Posted::default();
        run_read(&fake, &fake, &poster, &request("r")).await;
        let sent = fake.sent.lock().expect("unpoisoned");
        assert_eq!(sent.len(), 1);
        assert_eq!(
            (sent[0].model.as_str(), sent[0].effort),
            ("sonnet", Some(Effort::Medium))
        );
        assert_eq!(
            sent[0].runtime,
            swarm_wire::review_classifier::Runtime::Acp(acp)
        );
    }

    #[tokio::test]
    async fn unreadable_settings_run_nothing_and_post_a_comment() {
        let fake = Fake::new(
            Reply::Done {
                rows: vec![row(None, true, true)],
                overall: Overall::Approve,
            },
            vec![rules_json()],
        );
        let poster = Posted::default();
        run_read(&fake, &Queue(None), &poster, &request("r")).await;
        let (_, commit, review) = poster.0.into_inner().expect("unpoisoned").remove(0);
        assert_eq!(review.event, "COMMENT");
        assert!(
            review
                .body
                .contains("reading the classifier's settings failed: no swarm queue is configured"),
            "{}",
            review.body
        );
        assert_eq!(commit.as_deref(), Some("abc123"));
    }

    #[tokio::test]
    async fn a_clean_run_posts_its_verdict_on_the_head_commit() {
        let (target, commit, review) = posted(
            Reply::Done {
                rows: vec![row(None, true, true)],
                overall: Overall::Approve,
            },
            "r",
        )
        .await;
        assert_eq!(target, "o/r#7");
        assert_eq!(commit.as_deref(), Some("abc123"));
        assert_eq!(review.event, "APPROVED");
        assert!(
            review.body.ends_with("**Overall: ✅ approve**\n"),
            "{}",
            review.body
        );

        let (_, _, review) = posted(
            Reply::Done {
                rows: vec![row(None, true, false)],
                overall: Overall::RequestChanges,
            },
            "r",
        )
        .await;
        assert_eq!(review.event, "REQUEST_CHANGES");
        assert!(
            review.body.contains("| ❌ match | datetime | 90 / 80 |"),
            "{}",
            review.body
        );
    }

    #[tokio::test]
    async fn no_rules_a_failed_run_or_no_pr_is_a_comment() {
        let approve = || Reply::Done {
            rows: vec![row(None, true, true)],
            overall: Overall::Approve,
        };
        // `o/elsewhere` has no PR 7.
        for (reply, repo, says, commit) in [
            (
                Reply::Failed {
                    error: "no API key".to_owned(),
                },
                "r",
                "the run failed: no API key",
                Some("abc123"),
            ),
            (approve(), "elsewhere", "reading the PR failed", None),
        ] {
            let (_, posted_commit, review) = posted(reply, repo).await;
            assert_eq!(review.event, "COMMENT", "{repo}");
            assert!(review.body.contains(says), "{repo}: {}", review.body);
            assert_eq!(posted_commit.as_deref(), commit, "{repo}");
        }

        let [yml, _] = crate::review_classifier::RULES_PATHS;
        let fake = Fake::new(approve(), vec![rules_json(), (yml, String::new())]);
        let poster = Posted::default();
        run_read(&fake, &fake, &poster, &request("r")).await;
        let (_, _, review) = poster.0.into_inner().expect("unpoisoned").remove(0);
        assert_eq!(review.event, "COMMENT");
        assert!(
            review.body.contains("the rules did not load") && review.body.contains("holds both"),
            "{}",
            review.body
        );
    }

    #[tokio::test]
    async fn a_repo_without_rules_is_approved_without_a_model_call() {
        let fake = Fake::new(
            Reply::Failed {
                error: "must not run".to_owned(),
            },
            Vec::new(),
        );
        let poster = Posted::default();
        let run = run_read(&fake, &fake, &poster, &request("r")).await;
        assert_eq!(run.outcome, RunOutcome::Approved);
        assert_eq!(run.detail, None);
        assert_eq!(run.head_sha.as_deref(), Some("abc123"));
        assert!(fake.sent.lock().expect("unpoisoned").is_empty());
        let reviews = poster.0.into_inner().expect("unpoisoned");
        assert_eq!(reviews.len(), 1);
        let (pr, commit, review) = &reviews[0];
        assert_eq!(pr, "o/r#7");
        assert_eq!(commit.as_deref(), Some("abc123"));
        assert_eq!(review.event, "APPROVED");
        assert_eq!(
            review.body,
            "no review rules are configured for o/r; approved without checks\n"
        );

        // A request for the same head approves again, still without a run.
        let poster = Posted::default();
        let again = run_read(&fake, &fake, &poster, &request("r")).await;
        assert_eq!(again.outcome, RunOutcome::Approved);
        assert!(fake.sent.lock().expect("unpoisoned").is_empty());
        assert_eq!(poster.0.into_inner().expect("unpoisoned").len(), 1);

        // Control: a repo with rules still runs the classifier and posts its verdict.
        let approve = Reply::Done {
            rows: vec![row(None, true, true)],
            overall: Overall::Approve,
        };
        let with = Fake::new(approve, vec![rules_json()]);
        let poster = Posted::default();
        let run = run_read(&with, &with, &poster, &request("r")).await;
        assert_eq!(run.outcome, RunOutcome::Approved);
        assert_eq!(with.sent.lock().expect("unpoisoned").len(), 1);
        let (_, _, review) = poster.0.into_inner().expect("unpoisoned").remove(0);
        assert_eq!(review.event, "APPROVED");
        assert!(
            review.body.contains("Overall: ✅ approve"),
            "{}",
            review.body
        );
    }

    #[test]
    fn a_hostile_reason_mentions_nobody_and_stays_in_its_cell() {
        let mut r = row(None, true, true);
        r.reason = "ok | @mara please merge\n| x | ![](http://e/p.png) <b>`".to_owned();
        let body = table(&[r]);
        let line = body.lines().nth(2).expect("a data row");
        assert_eq!(body.lines().count(), 3, "{body}");
        assert!(!line.contains("@mara"), "{line}");
        assert!(line.contains("@\u{200b}mara"), "{line}");
        // Four cells: five unescaped pipes.
        let unescaped = line
            .char_indices()
            .filter(|(i, c)| *c == '|' && (*i == 0 || line.as_bytes()[i - 1] != b'\\'))
            .count();
        assert_eq!(unescaped, 5, "{line}");
        assert!(
            line.contains("!\u{200b}\\[\\]") && line.contains("\\<b\\>"),
            "{line}"
        );
    }

    #[test]
    fn a_reason_carries_no_live_math_delimiter() {
        let mut r = row(None, true, true);
        r.reason =
            "uses https://${hyperhiveCfg.swarm.forge.domain}/metrics and $$x$$ | y".to_owned();
        let body = table(&[r]);
        let line = body.lines().nth(2).expect("a data row");
        assert!(
            line.contains("\\${hyperhiveCfg") && line.contains("\\$\\$x\\$\\$"),
            "{line}"
        );
        let bare = line
            .char_indices()
            .filter(|(i, c)| *c == '$' && (*i == 0 || line.as_bytes()[i - 1] != b'\\'))
            .count();
        assert_eq!(bare, 0, "{line}");
        let unescaped = line
            .char_indices()
            .filter(|(i, c)| *c == '|' && (*i == 0 || line.as_bytes()[i - 1] != b'\\'))
            .count();
        assert_eq!(unescaped, 5, "{line}");
    }

    #[test]
    fn a_reason_carries_no_link_or_reference() {
        let text = cell("see https://evil.example/x, WWW.evil.example, #1, o/r#3 and !4");
        assert_eq!(
            text,
            "see https://\u{200b}evil.example/x, WWW.\u{200b}evil.example, #\u{200b}1, \
             o/r#\u{200b}3 and !\u{200b}4"
        );
    }

    #[test]
    fn a_plain_reason_reads_without_backslashes() {
        assert_eq!(
            cell("an added #[allow(clippy::foo)] in ci_metrics.rs (see main_loop_v2.rs)"),
            "an added #\u{200b}\\[allow(clippy::foo)\\] in ci_metrics.rs (see main_loop_v2.rs)"
        );
    }

    #[test]
    fn an_underscore_escapes_unless_it_sits_between_alphanumerics() {
        // Presence control: the edge underscores are escaped.
        assert_eq!(
            cell("_em_ __b__ a_ _b a__b snake_case"),
            "\\_em\\_ \\_\\_b\\_\\_ a\\_ \\_b a\\_\\_b snake_case"
        );
    }

    #[test]
    fn a_reason_with_markup_characters_keeps_them_inert() {
        assert_eq!(
            cell(r"a\b `c` *d* ~e~ [f] <g> &h; i|j $k"),
            r"a\\b \`c\` \*d\* \~e\~ \[f\] \<g\> \&h; i\|j \$k"
        );
    }

    #[test]
    fn only_a_request_for_the_classifier_parses() {
        let payload = |action: &str, reviewer: Option<&str>| {
            serde_json::json!({
                "action": action,
                "number": 7,
                "pull_request": { "head": { "sha": "abc123" } },
                "requested_reviewer": reviewer.map(|l| serde_json::json!({ "login": l })),
                "repository": { "full_name": "o/r" },
                "sender": { "login": "alice" },
            })
            .to_string()
        };
        assert_eq!(
            Requested::parse(payload("review_requested", Some(LOGIN)).as_bytes()),
            Some(request("r"))
        );
        assert_eq!(
            Requested::parse(payload("review_requested", Some("Review-Classifier")).as_bytes()),
            Some(request("r")),
            "the forge's logins are case-insensitive"
        );
        let mut no_head: serde_json::Value =
            serde_json::from_str(&payload("review_requested", Some(LOGIN))).expect("JSON");
        no_head["pull_request"] = serde_json::Value::Null;
        assert_eq!(
            Requested::parse(no_head.to_string().as_bytes()),
            Some(requested_at(""))
        );
        for body in [
            payload("review_request_removed", Some(LOGIN)),
            payload("review_requested", Some("argus")),
            payload("review_requested", None),
            "{}".to_owned(),
        ] {
            assert_eq!(Requested::parse(body.as_bytes()), None, "{body}");
        }
    }

    #[tokio::test]
    async fn a_run_on_the_forge_diff_says_so_in_its_review() {
        /// The body a run posts when the PR read was narrowed for `reason`,
        /// with the rules on the default branch or without them.
        async fn narrowed(reason: Option<String>, rules: bool) -> String {
            let reply = Reply::Done {
                rows: vec![row(None, true, true)],
                overall: Overall::Approve,
            };
            let files = if rules {
                vec![rules_json()]
            } else {
                Vec::new()
            };
            let mut fake = Fake::new(reply, files);
            fake.narrowed = reason;
            let poster = Posted::default();
            run_read(&fake, &fake, &poster, &request("r")).await;
            poster.0.into_inner().expect("unpoisoned").remove(0).2.body
        }
        let reason = || Some("the PR changes 101 files".to_owned());
        let body = narrowed(reason(), true).await;
        assert!(
            body.ends_with(
                "\nThe classifier read the forge's diff, with 3 lines of context, not 25: \
                 the PR changes 101 files.\n"
            ),
            "{body}"
        );
        let body = narrowed(None, true).await;
        assert!(!body.contains("forge's diff"), "{body}");
        let body = narrowed(reason(), false).await;
        assert!(body.starts_with("no review rules are configured"), "{body}");
        assert!(
            !body.contains("forge's diff"),
            "a run that never ran read no diff: {body}"
        );
    }

    /// Row `name` as the classifier answers it: `verdict`, counted and passed
    /// when it matched.
    fn named(name: &str, verdict: Verdict, error: Option<&str>) -> Row {
        Row {
            name: name.to_owned(),
            verdict,
            ..row(error, error.is_none(), error.is_none())
        }
    }

    const REQUESTS: &str = "hyperhive.review_classifier.requests";
    const VERDICTS: &str = "hyperhive.review_classifier.verdicts";
    const CHECKS: &str = "hyperhive.review_classifier.checks";
    const DURATION: &str = "hyperhive.review_classifier.run.duration";

    /// `pairs` as a count per label set.
    fn counts<const N: usize>(pairs: [(&str, u64); N]) -> std::collections::BTreeMap<String, u64> {
        pairs.into_iter().map(|(k, v)| (k.to_owned(), v)).collect()
    }

    /// One review run per reply, each on its own PR head, then a repeat of the
    /// last: the metrics name and label what each did.
    #[tokio::test]
    async fn a_review_records_its_request_verdict_checks_and_duration() {
        let recorder = Recorder::new();
        let metrics = &recorder.metrics;
        let reviews = Arc::new(Reviews::default());
        let store = Runs::Fake(Arc::default());
        let poster = Posted::default();

        // A request made at a head the PR has moved on from: reviewed at the
        // current head. Four rules, one answer of each kind: no verdict.
        let moved = Requested {
            head: "def456".to_owned(),
            ..request("r")
        };
        let rows = vec![
            named("c0", Verdict::Match, None),
            named("c1", Verdict::NoMatch, None),
            named("c2", Verdict::NotApplicable, None),
            named("c3", Verdict::NotApplicable, Some("timed out")),
        ];
        let mixed = Fake::new(
            Reply::Done {
                rows,
                overall: Overall::Approve,
            },
            vec![rules_of(4)],
        );
        review(
            &mixed,
            &mixed,
            &poster,
            &store,
            metrics,
            &moved,
            reviews.slot().await,
        )
        .await;

        // One rule that matched: the run approves, then a repeat of the
        // request posts the same verdict again without a run.
        let clean = Fake::new(
            Reply::Done {
                rows: vec![named("c0", Verdict::Match, None)],
                overall: Overall::Approve,
            },
            vec![rules_of(1)],
        );
        for _ in 0..2 {
            review(
                &clean,
                &clean,
                &poster,
                &store,
                metrics,
                &request("r"),
                reviews.slot().await,
            )
            .await;
        }

        // The queue is not configured: the request is refused with a comment.
        accept(None, &poster, &store, metrics, &request("r")).await;

        assert_eq!(
            recorder.counter(REQUESTS),
            counts([
                ("outcome=queue_failed", 1),
                ("outcome=reposted", 1),
                ("outcome=started", 2),
            ])
        );
        assert_eq!(
            recorder.counter(VERDICTS),
            counts([("outcome=approve", 2), ("outcome=no_verdict", 2)])
        );
        assert_eq!(
            recorder.counter(CHECKS),
            counts([
                ("check=c0,result=match", 2),
                ("check=c1,result=no_match", 1),
                ("check=c2,result=unsure", 1),
                ("check=c3,result=no_answer", 1),
            ])
        );
        assert_eq!(
            recorder.histogram(DURATION),
            counts([("outcome=approve", 1), ("outcome=no_verdict", 1)])
        );
    }

    /// A run whose rows ask for changes records that verdict, and one whose
    /// post the forge refuses records its duration as `post_failed`.
    #[tokio::test]
    async fn a_run_records_request_changes_and_a_failed_post() {
        let recorder = Recorder::new();
        let metrics = &recorder.metrics;
        let reply = Reply::Done {
            rows: vec![named("c0", Verdict::NoMatch, None)],
            overall: Overall::RequestChanges,
        };
        let fake = Fake::new(reply, vec![rules_of(1)]);
        run(
            &fake,
            read_pull(&fake, &request("r")).await,
            &fake,
            &Posted::default(),
            metrics,
            &request("r"),
        )
        .await;
        run(
            &fake,
            read_pull(&fake, &request("r")).await,
            &fake,
            &Refusing,
            metrics,
            &request("r"),
        )
        .await;

        assert_eq!(
            recorder.counter(VERDICTS),
            counts([("outcome=request_changes", 2)])
        );
        assert_eq!(
            recorder.histogram(DURATION),
            counts([("outcome=post_failed", 1), ("outcome=request_changes", 1)])
        );
    }

    /// A repo without rules approves with no checks: it counts as an approve
    /// verdict and an approved run, and adds no `checks` row.
    #[tokio::test]
    async fn a_repo_without_rules_records_an_approve_and_no_checks() {
        let recorder = Recorder::new();
        let metrics = &recorder.metrics;
        let none = Fake::new(
            Reply::Failed {
                error: "must not run".to_owned(),
            },
            Vec::new(),
        );
        run(
            &none,
            read_pull(&none, &request("r")).await,
            &none,
            &Posted::default(),
            metrics,
            &request("r"),
        )
        .await;

        assert_eq!(recorder.counter(VERDICTS), counts([("outcome=approve", 1)]));
        assert_eq!(
            recorder.histogram(DURATION),
            counts([("outcome=approve", 1)])
        );
        assert!(recorder.counter(CHECKS).is_empty());

        // Control: a repo with a rule records its check, so the empty
        // `checks` above is the absence of rules and not a lookup that finds
        // nothing.
        let ruled = Fake::new(
            Reply::Done {
                rows: vec![named("c0", Verdict::Match, None)],
                overall: Overall::Approve,
            },
            vec![rules_of(1)],
        );
        run(
            &ruled,
            read_pull(&ruled, &request("r")).await,
            &ruled,
            &Posted::default(),
            metrics,
            &request("r"),
        )
        .await;
        assert_eq!(
            recorder.counter(CHECKS),
            counts([("check=c0,result=match", 1)])
        );
        assert_eq!(recorder.counter(VERDICTS), counts([("outcome=approve", 2)]));
    }
}