hyperhive/swarm-controller/src/review_classifier/review.rs

//! 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; a push does not. 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 run with no clean verdict never approves: see [`no_clean_verdict_event`].
//! At most `max_concurrent_reviews` runs go at once, as the swarm's
//! [settings](super::settings) say when a run asks for its place, and at most
//! [`MAX_WAITING_REVIEWS`] more wait here, in memory, so a restart drops them.
//! See [`Reviews`] for what happens to a request beyond that.

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

use serde::Deserialize;
use swarm_queue_client::review_classifier::{Overall, Reply, Row, TryOut, Verdict};
use tokio::sync::Notify;
use tokio::task::JoinHandle;

use super::runs::{self, RunOutcome, RunRecord, Runs, Trigger};
use super::settings::{Settings, SettingsSource};
use super::{FileSource, PullSource, RunSink, split_repo};
use crate::AppState;
use crate::agent_renewal::unix_now;
use crate::forge::{self, classifier_account::LOGIN, classifier_account::Review};

/// The most review runs waiting for a slot.
const MAX_WAITING_REVIEWS: usize = 8;

/// 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",
        }
    }
}

/// 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`].
#[derive(Debug, Clone, PartialEq, Eq)]
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 key(&self) -> Key {
        (
            self.owner.clone(),
            self.name.clone(),
            self.pr,
            self.head.clone(),
        )
    }
}

/// Repo owner, repo name, PR number and the head the request named.
type Key = (String, String, i64, String);

/// The review runs this controller has accepted and not yet finished.
///
/// A request whose [`Key`] is already running or waiting starts nothing. A
/// request beyond the run limit + [`MAX_WAITING_REVIEWS`] starts no run
/// either; the account posts a no-verdict comment saying so.
pub(crate) struct Reviews {
    slots: Mutex<Slots>,
    /// Woken when a run ends or the limit changes.
    changed: Notify,
    accepted: Mutex<HashSet<Key>>,
}

/// 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(),
            accepted: Mutex::default(),
        }
    }
}

/// Why [`Reviews::accept`] turned a request away.
#[derive(Debug, PartialEq, Eq)]
enum Refused {
    /// A run for the same key is running or waiting.
    Duplicate,
    /// Every run and wait place is taken, by this many runs.
    Full(usize),
}

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
    }

    fn accept(self: &Arc<Self>, key: Key) -> Result<Accepted, Refused> {
        let mut accepted = self.accepted.lock().unwrap_or_else(PoisonError::into_inner);
        if accepted.contains(&key) {
            return Err(Refused::Duplicate);
        }
        if accepted.len() >= self.slots().limit + MAX_WAITING_REVIEWS {
            return Err(Refused::Full(accepted.len()));
        }
        accepted.insert(key.clone());
        Ok(Accepted {
            reviews: Arc::clone(self),
            key,
        })
    }

    /// 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 accepted run; dropping it frees its key and its place.
struct Accepted {
    reviews: Arc<Reviews>,
    key: Key,
}

impl Drop for Accepted {
    fn drop(&mut self) {
        self.reviews
            .accepted
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .remove(&self.key);
    }
}

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

/// Start a review run for 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 queue = Queue(state.status.as_ref().map(|s| s.queue_client()));
    let runs = state.runs.clone();
    start(
        &state.reviews,
        Arc::clone(&forge),
        queue,
        forge,
        runs,
        request,
    );
}

/// Accept `request` into `reviews` and spawn its run, or the no-verdict
/// comment when there's no place for it, and record it in `runs` when it
/// ends. The task, if one was spawned.
///
/// The run reads the settings for the limit before it waits for a place, and
/// again for the model and effort once it has one.
fn start<S, Q, P>(
    reviews: &Arc<Reviews>,
    forge: Arc<S>,
    queue: Q,
    poster: Arc<P>,
    runs: Runs,
    request: Requested,
) -> Option<JoinHandle<()>>
where
    S: FileSource + PullSource + Send + Sync + 'static,
    Q: RunSink + SettingsSource,
    P: ReviewPoster + Send + Sync + 'static,
{
    let repo = format!("{}/{}", request.owner, request.name);
    match reviews.accept(request.key()) {
        Ok(accepted) => {
            let reviews = Arc::clone(reviews);
            Some(tokio::spawn(async move {
                match 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 run waits under \
                         the last limit read"
                    ),
                }
                let slot = reviews.slot().await;
                let run = Box::pin(run(&*forge, &*forge, &queue, &*poster, &request)).await;
                drop(slot);
                drop(accepted);
                runs.record(run).await;
            }))
        }
        Err(Refused::Duplicate) => {
            tracing::info!(
                %repo, pr = request.pr, head_sha = request.head.as_str(),
                "review requested while a run for this head is running or waiting; nothing more starts"
            );
            None
        }
        Err(Refused::Full(taken)) => {
            tracing::warn!(
                %repo, pr = request.pr, head_sha = request.head.as_str(),
                "review requested, but every run and wait place is taken; posting a no-verdict comment"
            );
            Some(tokio::spawn(async move {
                let started_unix = unix_now();
                let ruling = queue_full_ruling(taken);
                let reason = ruling.reason.clone();
                let sent = post(&*poster, &request, None, ruling.review()).await;
                let ended = ended(RunOutcome::QueueFull, reason, sent);
                runs.record(record(&request, None, started_unix, ended, (None, None)))
                    .await;
            }))
        }
    }
}

/// The ruling for a request [`Reviews`] had no place for, with `taken` runs
/// running or waiting.
fn queue_full_ruling(taken: usize) -> Ruling {
    let reason = format!(
        "{taken} reviews are running or waiting, so this request starts no run; \
         request the review again later"
    );
    Ruling {
        event: no_clean_verdict_event(),
        body: no_verdict(&reason, &[]),
        reason: Some(reason),
    }
}

/// 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 {
    /// Nothing reached the classifier, for this reason.
    NotRun(String),
    /// The classifier answered a run of `sent` classifications.
    Ran { sent: usize, reply: Reply },
}

/// Read the PR and the rules, run them, and post the review. The finished
/// run.
async fn run(
    files: &impl FileSource,
    pulls: &impl PullSource,
    sink: &(impl RunSink + SettingsSource),
    poster: &impl ReviewPoster,
    req: &Requested,
) -> RunRecord {
    let started_unix = unix_now();
    let (head, outcome) = classify(files, pulls, sink, req).await;
    let ruling = review_for(&outcome);
    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(_) => (None, None),
    };
    record(
        req,
        head,
        started_unix,
        ended(posted_as, reason, sent),
        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 sent = forge::within_forge_deadline(
        "review classifier review",
        poster.post(&req.owner, &req.name, req.pr, head, review),
    )
    .await;
    match sent {
        Some(Ok(url)) => {
            tracing::info!(
                %repo, pr = req.pr, head_sha = head.unwrap_or_default(), event,
                "review classifier review posted"
            );
            Ok(url)
        }
        Some(Err(e)) => {
            let error = format!("{e:#}");
            tracing::warn!(
                %repo, pr = req.pr, %error,
                "posting the review classifier's review failed"
            );
            Err(format!("posting the review failed: {error}"))
        }
        None => Err("the forge did not answer the review's post within the deadline".to_owned()),
    }
}

/// 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, and how far the run got.
async fn classify(
    files: &impl FileSource,
    pulls: &impl PullSource,
    sink: &(impl RunSink + SettingsSource),
    req: &Requested,
) -> (Option<String>, Outcome) {
    let repo = format!("{}/{}", req.owner, req.name);
    let pull = match forge::within_forge_deadline(
        "review classifier PR read",
        pulls.pull(&req.owner, &req.name, req.pr),
    )
    .await
    {
        Some(Ok(pull)) => pull,
        Some(Err(e)) => {
            return (
                None,
                Outcome::NotRun(format!("reading the PR failed: {e:#}")),
            );
        }
        None => {
            return (
                None,
                Outcome::NotRun("the forge did not answer within the deadline".to_owned()),
            );
        }
    };
    let head = Some(pull.head_sha.clone());
    let rules = match super::rules(files, &req.owner, &req.name).await {
        Ok(axum::Json(rules)) => rules,
        Err(problem) => {
            let detail = problem.detail.unwrap_or_default();
            return (
                head,
                Outcome::NotRun(format!("the rules did not load: {detail}")),
            );
        }
    };
    let settings = match sink.settings().await {
        Ok(settings) => settings,
        Err(e) => {
            return (
                head,
                Outcome::NotRun(format!("reading the classifier's settings failed: {e:#}")),
            );
        }
    };
    let runtime = match settings.runtime() {
        Ok(runtime) => runtime,
        Err(e) => {
            return (
                head,
                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, 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::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, said) = match overall {
                    Overall::Approve => (Event::Approve, "approve"),
                    Overall::RequestChanges => (Event::RequestChanges, "request changes"),
                };
                Ruling {
                    event,
                    body: format!("{}\n**Overall: {said}.**\n", table(rows)),
                    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(
        "| classification | polarity | verdict | certainty | threshold | counted | reason |\n\
         | --- | --- | --- | --- | --- | --- | --- |\n",
    );
    for r in rows {
        let polarity = match r.polarity {
            swarm_queue_client::review_classifier::Polarity::Positive => "positive",
            swarm_queue_client::review_classifier::Polarity::Negative => "negative",
        };
        let (verdict, certainty, counted, reason) = if let Some(error) = &r.error {
            ("no answer", String::new(), "no", error.as_str())
        } else {
            let verdict = match r.verdict {
                Verdict::Match => "match",
                Verdict::NoMatch => "no match",
                Verdict::NotApplicable => "not applicable",
            };
            let counted = match (r.counted, r.passed) {
                (false, _) => "no",
                (true, true) => "yes ✅",
                (true, false) => "yes ❌",
            };
            (verdict, r.certainty.to_string(), counted, r.reason.as_str())
        };
        let _ = writeln!(
            out,
            "| {} | {polarity} | {verdict} | {certainty} | {} | {counted} | {} |",
            cell(&r.name),
            r.threshold,
            cell(reason)
        );
    }
    out
}

/// `text` as one table cell's inline text: on one line, with every character
/// that could end the cell, start markup, math (`$`) or raw HTML escaped, 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.
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(' ');
        }
        for c in word.chars() {
            match c {
                '\\' | '|' | '`' | '*' | '_' | '~' | '[' | ']' | '(' | ')' | '<' | '>' | '&'
                | '$' => {
                    out.push('\\');
                    out.push(c);
                }
                '#' | '!' => {
                    out.push('\\');
                    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 swarm_queue_client::review_classifier::Polarity;

    use super::*;
    use crate::review_classifier::runs::tests::FakeDoc;
    use crate::review_classifier::tests::{Fake, rules_json};

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

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

    /// 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(&fake, &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())
        }
    }

    /// Let every spawned task run until it waits on something.
    async fn settle() {
        for _ in 0..200 {
            tokio::task::yield_now().await;
        }
    }

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

    #[tokio::test]
    async fn a_request_for_a_head_already_running_or_waiting_starts_nothing() {
        let reviews = Arc::new(Reviews::default());
        let fake = Arc::new(Fake::new(
            Reply::Failed {
                error: String::new(),
            },
            vec![rules_json()],
        ));
        let gate = Gate::closed();
        let poster = Arc::new(Posted::default());
        let doc = Arc::new(FakeDoc::default());
        let store = Runs::Fake(Arc::clone(&doc));
        let start_at = |head: &str| {
            start(
                &reviews,
                Arc::clone(&fake),
                gate.clone(),
                Arc::clone(&poster),
                store.clone(),
                requested_at(head),
            )
        };

        // The first two run, the third waits; a repeat of any starts nothing.
        let mut tasks: Vec<_> = ["a", "b", "c"]
            .map(|h| start_at(h).expect("accepted"))
            .into();
        for head in ["a", "b", "c"] {
            assert!(start_at(head).is_none(), "{head}");
        }
        gate.open(1);
        for task in tasks.drain(..) {
            task.await.expect("ran");
        }
        assert_eq!(poster.0.lock().expect("unpoisoned").len(), 3);
        assert_eq!(
            doc.runs().len(),
            3,
            "a repeat that started nothing is no run"
        );

        // A finished run frees its head.
        start_at("a").expect("accepted again").await.expect("ran");
        assert_eq!(poster.0.lock().expect("unpoisoned").len(), 4);
        assert_eq!(doc.runs().len(), 4);
    }

    #[tokio::test]
    async fn a_request_beyond_the_wait_places_posts_a_no_verdict_comment() {
        let reviews = Arc::new(Reviews::default());
        let fake = Arc::new(Fake::new(
            Reply::Failed {
                error: String::new(),
            },
            vec![rules_json()],
        ));
        let gate = Gate::closed();
        let poster = Arc::new(Posted::default());
        let doc = Arc::new(FakeDoc::default());
        let store = Runs::Fake(Arc::clone(&doc));
        let start_at = |head: String| {
            start(
                &reviews,
                Arc::clone(&fake),
                gate.clone(),
                Arc::clone(&poster),
                store.clone(),
                requested_at(&head),
            )
            .expect("a run or a comment")
        };

        let places = Settings::default().max_concurrent_reviews + MAX_WAITING_REVIEWS;
        let runs: Vec<_> = (0..places).map(|i| start_at(format!("h{i}"))).collect();
        tokio::time::timeout(Duration::from_secs(5), start_at("over".to_owned()))
            .await
            .expect("refused without waiting for a place")
            .expect("posted");
        {
            let all = poster.0.lock().expect("unpoisoned");
            assert_eq!(all.len(), 1, "only the refusal is posted yet");
            let (target, commit, review) = &all[0];
            assert_eq!(target, "o/r#7");
            assert_eq!(*commit, None);
            assert_eq!(review.event, "COMMENT");
            assert!(
                review.body.starts_with(
                    "**No verdict:** 10 reviews are running or waiting, so this request"
                ),
                "{}",
                review.body
            );
        }
        let refused = doc.runs();
        assert_eq!(refused.len(), 1);
        assert_eq!(refused[0].outcome, RunOutcome::QueueFull);
        assert_eq!(refused[0].head_sha, None);
        assert_eq!(
            refused[0].review_url.as_deref(),
            Some("https://forge/o/r/pulls/7#review")
        );
        assert!(
            refused[0]
                .detail
                .as_deref()
                .is_some_and(|d| d.starts_with("10 reviews are running or waiting")),
            "{:?}",
            refused[0].detail
        );

        gate.open(1);
        for run in runs {
            run.await.expect("ran");
        }
        let all = poster.0.lock().expect("unpoisoned");
        assert_eq!(all.len(), places + 1);
        assert!(all[1..].iter().all(|(_, _, r)| r.event == "APPROVED"));
        let recorded = doc.runs();
        assert_eq!(recorded.len(), places + 1);
        assert!(
            recorded[..places]
                .iter()
                .all(|r| r.outcome == RunOutcome::Approved)
        );
    }

    /// 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(&fake, &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 fake = Fake::new(failed, Vec::new());
        let run = run(&fake, &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")
        );
    }

    #[tokio::test]
    async fn a_failed_record_write_still_posts_the_review() {
        let fake = Arc::new(Fake::new(
            Reply::Done {
                rows: vec![row(None, true, true)],
                overall: Overall::Approve,
            },
            vec![rules_json()],
        ));
        let poster = Arc::new(Posted::default());
        let store = Runs::Fake(Arc::new(FakeDoc::failing()));
        start(
            &Arc::new(Reviews::default()),
            Arc::clone(&fake),
            (*fake).clone(),
            Arc::clone(&poster),
            store,
            request("r"),
        )
        .expect("accepted")
        .await
        .expect("ran to the end");
        let all = poster.0.lock().expect("unpoisoned");
        assert_eq!(all.len(), 1);
        assert_eq!(all[0].2.event, "APPROVED");
    }

    #[tokio::test]
    async fn a_new_limit_applies_to_the_next_run_without_a_restart() {
        let reviews = Arc::new(Reviews::default());
        let fake = Arc::new(Fake::new(
            Reply::Failed {
                error: String::new(),
            },
            vec![rules_json()],
        ));
        let gate = Gate::closed();
        let poster = Arc::new(Posted::default());
        let start_at = |head: &str| {
            start(
                &reviews,
                Arc::clone(&fake),
                gate.clone(),
                Arc::clone(&poster),
                Runs::Fake(Arc::default()),
                requested_at(head),
            )
            .expect("accepted")
        };

        // The stored limit is read before a run waits, not the default 2.
        gate.store(1);
        let mut tasks = vec![start_at("a"), start_at("b")];
        settle().await;
        assert_eq!(gate.reached(), 1);

        // Raised, as a PUT does: the waiting run starts at once.
        gate.store(3);
        reviews.resize(3);
        settle().await;
        assert_eq!(gate.reached(), 2);
        tasks.push(start_at("c"));
        settle().await;
        assert_eq!(gate.reached(), 3);

        // Lowered: the three keep running and the next one waits.
        gate.store(1);
        tasks.push(start_at("d"));
        settle().await;
        assert_eq!(gate.reached(), 3);

        gate.open(1);
        for task in tasks {
            tokio::time::timeout(Duration::from_secs(5), task)
                .await
                .expect("a finished run frees its place")
                .expect("ran");
        }
        assert_eq!(gate.reached(), 4);
        assert_eq!(reviews.slots().running, 0, "every place is freed");
        assert_eq!(poster.0.lock().expect("unpoisoned").len(), 4);
    }

    #[tokio::test]
    async fn a_review_runs_on_the_stored_runtime_model_and_effort() {
        use swarm_queue_client::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_queue_client::review_classifier::Acp {
            base_url: "https://llm.example/v1".to_owned(),
            model: "qwen".to_owned(),
            prices: swarm_queue_client::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(&fake, &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_queue_client::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(&fake, &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("| yes ❌ |"), "{}", 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 fake = Fake::new(approve(), Vec::new());
        let poster = Posted::default();
        run(&fake, &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("neither"),
            "{}",
            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}");
        // Seven cells: eight unescaped pipes.
        let unescaped = line
            .char_indices()
            .filter(|(i, c)| *c == '|' && (*i == 0 || line.as_bytes()[i - 1] != b'\\'))
            .count();
        assert_eq!(unescaped, 8, "{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, 8, "{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 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}");
        }
    }
}