swarm-controller: harden the review classifier's real reviews

Follow-ups from the review of #5142.

A `Done` answer with fewer or more rows than the classifications sent, or
none, now posts the no-verdict comment instead of the classifier's overall
verdict, so a short answer can't approve.

Requests are accepted per repo, PR and head into a set bounded at the two
running plus eight waiting. A request for a key already running or waiting
starts nothing and logs at info; a request with every place taken starts no
run and posts a no-verdict comment saying so. Before, every delivery spawned
a task waiting on the semaphore, so toggling the review request queued one
paid run per toggle.

Table cells get a zero-width space after `#`, `!`, `://` and `www.`, so a
model's reason can't render an issue or PR reference or a bare URL as a link.

The secret publisher's policy denies the classifier's forge token leaf, the
way #5139 denied Grafana's admin password, with a module-eval-bao-grants arm
pinning it. The controller posts reviews with whatever token is there, so a
publisher write picked the account reviews come from.

Closes #5143, part of #5096.

docs/swarm/credentials.md

@@ -74,7 +74,7 @@ of the cell says how.
 | `swarm/services/<clientId>/oidc/client`                     | authelia                                                                                                                                                            | that service's own reader, under its own certificate — the queue responder's under `swarm-nats-auth`'s, the homeserver's under `swarm-matrix-ctl`'s, the forge's under `swarm-forge-publisher`'s, the review classifier's under `swarm-review-classifier`'s; the bao UI's by the store's granter | must be stated                                                                                                                                                                                                                                                                                                                                                       | must be stated                                                                                                                                                                                                                                                                            |
 | `swarm/services/ci-runner/registration-token`               | `swarm-controller`, from the forge's admin API, at start and in a five-minute pass; its grant on the path is write-only                                             | `swarm-bao-ci-runner` on the runner's host, under the runner's own certificate, into a file only `swarm-ci-runner-register` in the runner's container loads; no other principal's policy reads it                                                                                                | ❌ the forge hands out one token until an admin resets it; the controller writes a reset token within five minutes                                                                                                                                                                                                                                                   | ❌ `swarm-bao-ci-runner` reads it when it starts, at boot or deploy, and restarts the runner when the value changed                                                                                                                                                                       |
 | `swarm/services/review-classifier/anthropic-api-key`        | the operator, by hand in the store's UI, field `api_key`                                                                                                            | `swarm-review-classifier` in its container, under its own certificate, at the start of every run; held for that run and handed to `claude`'s environment only; no other principal's policy reads it                                                                                              | ❌ an operator's key; replaced by writing a new one                                                                                                                                                                                                                                                                                                                  | ✅ read on every run, so a written or removed key takes effect on the next run                                                                                                                                                                                                            |
-| `swarm/services/review-classifier/forge/token`              | `swarm-controller`, with the forge's admin API, for its `review-classifier` account, at start and in a five-minute pass                                             | `swarm-controller` under its own certificate, before it decides whether to mint, and on every review it posts as the account; the classifier service's policy doesn't read it                                                                                                                    | ✅ the controller's pass re-mints when the stored token is missing or no longer matches the forge (last eight characters and scopes)                                                                                                                                                                                                                                 | ✅ the controller reads the store on every review it posts                                                                                                                                                                                                                                |
+| `swarm/services/review-classifier/forge/token`              | `swarm-controller`, with the forge's admin API, for its `review-classifier` account, at start and in a five-minute pass; the secret publisher's policy denies it    | `swarm-controller` under its own certificate, before it decides whether to mint, and on every review it posts as the account; the classifier service's policy doesn't read it                                                                                                                    | ✅ the controller's pass re-mints when the stored token is missing or no longer matches the forge (last eight characters and scopes)                                                                                                                                                                                                                                 | ✅ the controller reads the store on every review it posts                                                                                                                                                                                                                                |
 | _(not in the store)_ a hive's mTLS leaf                     | the store's own PKI, or an operator placing it by hand                                                                                                              | its own client, off disk — the exception above, because it's what makes every other row's pull possible                                                                                                                                                                                          | must be stated                                                                                                                                                                                                                                                                                                                                                       | must be stated                                                                                                                                                                                                                                                                            |
 
 <!-- vale write-good.Passive = YES -->

docs/swarm/review-classifier.md

@@ -96,8 +96,11 @@ the earlier one. A repo whose branch protection also dismisses stale approvals (
 [PR review gate](../process/pr-review-gate.md)) stops counting an approval
 on an older commit even without a re-request.
 
-At most two requested reviews run at once. More wait in the controller's
-memory, so a controller restart drops them, and nothing gets posted for them.
+At most two requested reviews run at once, and up to eight more wait in the
+controller's memory, so a controller restart drops them, and nothing gets
+posted for them. A request for a PR and head that's already running or
+waiting starts nothing. A request with all ten places taken starts no run:
+the account posts a `**No verdict:**` comment saying so.
 
 ## Components
 
@@ -241,16 +244,18 @@ Repos that rely on it typically raise their required approval count to
 account for it.
 
 A run with no clean verdict never approves. When the PR or the rules don't
-load, the run fails, or **any** classification has no answer, the account
-posts a **comment** instead, starting with `**No verdict:**` and the reason,
+load, the run fails, the classifier answers a different number of
+classifications than the controller sent it, or **any** classification has no answer,
+the account posts a **comment** instead, starting with `**No verdict:**` and the reason,
 followed by the table when there is one. A comment neither approves nor
 blocks the merge: nine timeouts and one pass post a comment, not an
 approval.
 
 The controller flattens every cell to one line, with `|`, backticks, brackets, `<`, `>`,
 `&`, `#` and other markup characters escaped, and a zero-width space after
-each `@`, so a model's `reason` can't break the table, embed markup, or
-mention anyone.
+each `@`, `#`, `!`, `://` and `www.`, so a model's `reason` can't break the
+table, embed markup, mention anyone, reference an issue or PR, or carry a
+link.
 
 ## Runtimes
 

nix/host-modules/swarm-bao.nix

@@ -489,6 +489,9 @@ let
   # Grafana's admin password is denied the same way: a write to it sets the
   # password `grafana.service` applies to the server-admin account at its next
   # start.
+  # The review classifier's forge token is denied too: the controller posts
+  # the classifier's reviews with whatever token is there, so a write to it
+  # picks the account those reviews come from.
   secretPublisherPolicyText = ''
     path "${credentialMountPath}/data/swarm/hives/*" {
       capabilities = ["create", "update"]
@@ -514,6 +517,10 @@ let
       capabilities = ["deny"]
     }
 
+    path "${credentialMountPath}/data/${classifierForgeTokenLeaf}" {
+      capabilities = ["deny"]
+    }
+
     path "${credentialMountPath}/data/${controllerClientLeaf}" {
       capabilities = ["create", "update"]
     }

nix/module-eval/bao-grants.nix

@@ -479,6 +479,23 @@ let
             "update"
           ];
     }
+    {
+      # The review classifier's forge token sits under the publisher's
+      # `services/*` write too, and the controller posts the classifier's
+      # reviews with whatever token is there. The classifier's OIDC leaf is the
+      # control: still the prefix's.
+      name = "the publisher may not write the review classifier's forge token";
+      ok =
+        let
+          grants = grantsIn baoGrantHere.systemd.services.swarm-bao-secret-publisher-policy.script;
+        in
+        (grantFor grants "secret/data/swarm/services/review-classifier/forge/token").caps == [ "deny" ]
+        &&
+          (grantFor grants "secret/data/swarm/services/swarm-review-classifier/oidc/client").caps == [
+            "create"
+            "update"
+          ];
+    }
     {
       # The ordering is load-bearing and invisible at runtime: the controller's
       # unit creates the KV and cert-auth mounts this one writes into, so

swarm-controller/src/forge_account.rs

@@ -165,7 +165,7 @@ mod tests {
             create_gate: std::sync::Arc::default(),
             accounts: Accounts::Fake(Arc::clone(store)),
             try_outs: super::super::review_classifier::TryOuts::default(),
-            review_slots: super::super::review_slots(),
+            reviews: std::sync::Arc::default(),
         }
     }
 

swarm-controller/src/github_account.rs

@@ -135,7 +135,7 @@ mod tests {
             create_gate: std::sync::Arc::default(),
             accounts: Accounts::Fake(Arc::clone(store)),
             try_outs: super::super::review_classifier::TryOuts::default(),
-            review_slots: super::super::review_slots(),
+            reviews: std::sync::Arc::default(),
         }
     }
 

swarm-controller/src/linked_accounts.rs

@@ -970,7 +970,7 @@ pub(crate) mod tests {
             create_gate: std::sync::Arc::default(),
             accounts: super::Accounts::Fake(Arc::clone(store)),
             try_outs: super::super::review_classifier::TryOuts::default(),
-            review_slots: super::super::review_slots(),
+            reviews: std::sync::Arc::default(),
         }
     }
 

swarm-controller/src/main.rs

@@ -893,16 +893,8 @@ struct AppState {
     /// The review classifier try-outs started here, until their results
     /// expire.
     try_outs: review_classifier::TryOuts,
-    /// Held by each requested review's run, so at most
-    /// `review_classifier::review::MAX_CONCURRENT_REVIEWS` go at once.
-    review_slots: Arc<tokio::sync::Semaphore>,
-}
-
-/// A fresh [`AppState::review_slots`].
-fn review_slots() -> Arc<tokio::sync::Semaphore> {
-    Arc::new(tokio::sync::Semaphore::new(
-        review_classifier::review::MAX_CONCURRENT_REVIEWS,
-    ))
+    /// The requested reviews running or waiting.
+    reviews: Arc<review_classifier::review::Reviews>,
 }
 
 /// Env var the controller's NixOS module sets from
@@ -3275,7 +3267,7 @@ async fn main() -> Result<()> {
         create_gate: Arc::default(),
         accounts: linked_accounts::Accounts::Live,
         try_outs: review_classifier::TryOuts::default(),
-        review_slots: review_slots(),
+        reviews: Arc::default(),
     };
 
     let app = build_app(state);
@@ -3464,7 +3456,7 @@ mod tests {
             create_gate: std::sync::Arc::default(),
             accounts: super::linked_accounts::Accounts::Fake(std::sync::Arc::default()),
             try_outs: super::review_classifier::TryOuts::default(),
-            review_slots: super::review_slots(),
+            reviews: std::sync::Arc::default(),
         };
         (state, sched)
     }

swarm-controller/src/matrix_account.rs

@@ -579,7 +579,7 @@ mod tests {
             create_gate: std::sync::Arc::default(),
             accounts: Accounts::Fake(Arc::clone(store)),
             try_outs: super::super::review_classifier::TryOuts::default(),
-            review_slots: super::super::review_slots(),
+            reviews: std::sync::Arc::default(),
         }
     }
 

swarm-controller/src/review_classifier/review.rs

@@ -8,22 +8,29 @@
 //! 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; the rest wait here, in
-//! memory, so a restart drops them.
+//! At most [`MAX_CONCURRENT_REVIEWS`] runs go at once 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;
+use std::sync::{Arc, Mutex, PoisonError};
 
 use serde::Deserialize;
 use swarm_queue_client::review_classifier::{Overall, Reply, Row, TryOut, Verdict};
+use tokio::sync::Semaphore;
+use tokio::task::JoinHandle;
 
 use super::{FileSource, PullSource, RunSink, split_repo};
 use crate::AppState;
 use crate::forge::{self, classifier_account::LOGIN, classifier_account::Review};
 
 /// The most review runs in flight at once.
-pub(crate) const MAX_CONCURRENT_REVIEWS: usize = 2;
+const MAX_CONCURRENT_REVIEWS: usize = 2;
+
+/// 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)]
@@ -56,6 +63,9 @@ 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,
 }
@@ -64,11 +74,22 @@ pub(crate) struct Requested {
 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,
@@ -100,9 +121,83 @@ impl Requested {
             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 [`MAX_CONCURRENT_REVIEWS`] + [`MAX_WAITING_REVIEWS`] starts
+/// no run either; the account posts a no-verdict comment saying so.
+pub(crate) struct Reviews {
+    slots: Arc<Semaphore>,
+    accepted: Mutex<HashSet<Key>>,
+}
+
+impl Default for Reviews {
+    fn default() -> Self {
+        Self {
+            slots: Arc::new(Semaphore::new(MAX_CONCURRENT_REVIEWS)),
+            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.
+    Full,
+}
+
+impl Reviews {
+    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() >= MAX_CONCURRENT_REVIEWS + MAX_WAITING_REVIEWS {
+            return Err(Refused::Full);
+        }
+        accepted.insert(key.clone());
+        Ok(Accepted {
+            reviews: Arc::clone(self),
+            key,
+        })
+    }
+}
+
+/// 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);
+    }
 }
 
 /// Start a review run for a verified `review-request` delivery that asks for
@@ -120,13 +215,66 @@ pub(crate) fn on_delivery(state: &AppState, body: &[u8]) {
         return;
     };
     let queue = Queue(state.status.as_ref().map(|s| s.queue_client()));
-    let slots = Arc::clone(&state.review_slots);
-    tokio::spawn(async move {
-        let Ok(_slot) = slots.acquire_owned().await else {
-            return;
-        };
-        Box::pin(run(&*forge, &*forge, &queue, &*forge, &request)).await;
-    });
+    start(&state.reviews, Arc::clone(&forge), queue, forge, request);
+}
+
+/// Accept `request` into `reviews` and spawn its run, or the no-verdict
+/// comment when there's no place for it. The task, if one was spawned.
+fn start<S, Q, P>(
+    reviews: &Arc<Reviews>,
+    forge: Arc<S>,
+    queue: Q,
+    poster: Arc<P>,
+    request: Requested,
+) -> Option<JoinHandle<()>>
+where
+    S: FileSource + PullSource + Send + Sync + 'static,
+    Q: RunSink,
+    P: ReviewPoster + Send + Sync + 'static,
+{
+    let repo = format!("{}/{}", request.owner, request.name);
+    match reviews.accept(request.key()) {
+        Ok(accepted) => {
+            let slots = Arc::clone(&reviews.slots);
+            Some(tokio::spawn(async move {
+                let Ok(_slot) = slots.acquire_owned().await else {
+                    return;
+                };
+                Box::pin(run(&*forge, &*forge, &queue, &*poster, &request)).await;
+                drop(accepted);
+            }))
+        }
+        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) => {
+            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 {
+                post(&*poster, &request, None, queue_full_review()).await;
+            }))
+        }
+    }
+}
+
+/// The review for a request [`Reviews`] had no place for.
+fn queue_full_review() -> Review {
+    Review {
+        event: no_clean_verdict_event().forge(),
+        body: no_verdict(
+            &format!(
+                "{MAX_CONCURRENT_REVIEWS} reviews are running and {MAX_WAITING_REVIEWS} are \
+                 waiting, so this request starts no run; request the review again later"
+            ),
+            &[],
+        ),
+    }
 }
 
 /// Where a review is posted: the forge in production, a fake in the tests.
@@ -180,7 +328,8 @@ impl RunSink for Queue {
 enum Outcome {
     /// Nothing reached the classifier, for this reason.
     NotRun(String),
-    Ran(Reply),
+    /// 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.
@@ -191,18 +340,22 @@ async fn run(
     poster: &impl ReviewPoster,
     req: &Requested,
 ) {
-    let repo = format!("{}/{}", req.owner, req.name);
     let (head, outcome) = classify(files, pulls, sink, req).await;
-    let review = review_for(&outcome);
+    post(poster, req, head.as_deref(), review_for(&outcome)).await;
+}
+
+/// Post `review` on `req`'s PR, naming `head` when there is one.
+async fn post(poster: &impl ReviewPoster, req: &Requested, head: Option<&str>, review: Review) {
+    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.as_deref(), review),
+        poster.post(&req.owner, &req.name, req.pr, head, review),
     )
     .await;
     match sent {
         Some(Ok(())) => tracing::info!(
-            %repo, pr = req.pr, head_sha = head.as_deref().unwrap_or_default(), event,
+            %repo, pr = req.pr, head_sha = head.unwrap_or_default(), event,
             "review classifier review posted"
         ),
         Some(Err(e)) => tracing::warn!(
@@ -261,8 +414,12 @@ async fn classify(
         diff: pull.diff,
         classifications: rules.classifications,
     };
+    let sent = request.classifications.len();
     let outcome = match serde_json::to_vec(&request) {
-        Ok(payload) => Outcome::Ran(sink.run(payload).await),
+        Ok(payload) => Outcome::Ran {
+            sent,
+            reply: sink.run(payload).await,
+        },
         Err(e) => Outcome::NotRun(format!("encoding the run failed: {e}")),
     };
     (head, outcome)
@@ -272,13 +429,25 @@ async fn classify(
 fn review_for(outcome: &Outcome) -> Review {
     let (event, body) = match outcome {
         Outcome::NotRun(reason) => (no_clean_verdict_event(), no_verdict(reason, &[])),
-        Outcome::Ran(Reply::Failed { error }) => (
+        Outcome::Ran {
+            reply: Reply::Failed { error },
+            ..
+        } => (
             no_clean_verdict_event(),
             no_verdict(&format!("the run failed: {error}"), &[]),
         ),
-        Outcome::Ran(Reply::Done { rows, overall }) => {
+        Outcome::Ran {
+            sent,
+            reply: Reply::Done { rows, overall },
+        } => {
             let unanswered = rows.iter().filter(|r| r.error.is_some()).count();
-            if unanswered > 0 {
+            if rows.is_empty() || rows.len() != *sent {
+                let reason = format!(
+                    "the classifier answered {} of {sent} classifications",
+                    rows.len()
+                );
+                (no_clean_verdict_event(), no_verdict(&reason, rows))
+            } else if unanswered > 0 {
                 let reason = format!(
                     "{unanswered} of {} classifications got no answer",
                     rows.len()
@@ -350,8 +519,10 @@ fn table(rows: &[Row]) -> String {
 }
 
 /// `text` as one table cell's inline text: on one line, with every character
-/// that could end the cell, start markup or raw HTML escaped, and every `@`
-/// followed by a zero-width space, so no name after it is a live mention.
+/// that could end the cell, start markup 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() {
@@ -360,14 +531,22 @@ fn cell(text: &str) -> String {
         }
         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
@@ -376,6 +555,7 @@ fn cell(text: &str) -> String {
 #[cfg(test)]
 mod tests {
     use std::sync::Mutex;
+    use std::time::Duration;
 
     use swarm_queue_client::review_classifier::Polarity;
 
@@ -427,13 +607,42 @@ mod tests {
             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) {
-        let fake = Fake::new(reply, vec![rules_json()]);
+        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");
@@ -452,7 +661,7 @@ mod tests {
             rows,
             overall: Overall::Approve,
         };
-        let (_, commit, review) = posted(reply, "r").await;
+        let (_, commit, review) = posted_for(reply, "r", rules_of(10)).await;
         assert_eq!(review.event, "COMMENT");
         assert!(
             review
@@ -465,6 +674,155 @@ mod tests {
         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 review = review_for(&Outcome::Ran {
+            sent: 0,
+            reply: Reply::Done {
+                rows: Vec::new(),
+                overall: Overall::Approve,
+            },
+        });
+        assert_eq!(review.event, "COMMENT");
+    }
+
+    /// A classifier that answers approve once [`Gate::open`] lets it through.
+    #[derive(Clone)]
+    struct Gate(Arc<tokio::sync::Semaphore>);
+
+    impl Gate {
+        fn open(&self) {
+            self.0.add_permits(1);
+        }
+    }
+
+    impl RunSink for Gate {
+        async fn run(&self, _request: Vec<u8>) -> Reply {
+            let _permit = self.0.acquire().await.expect("never closed");
+            Reply::Done {
+                rows: vec![row(None, true, true)],
+                overall: Overall::Approve,
+            }
+        }
+    }
+
+    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(Arc::new(tokio::sync::Semaphore::new(0)));
+        let poster = Arc::new(Posted::default());
+        let start_at = |head: &str| {
+            start(
+                &reviews,
+                Arc::clone(&fake),
+                gate.clone(),
+                Arc::clone(&poster),
+                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();
+        for task in tasks.drain(..) {
+            task.await.expect("ran");
+        }
+        assert_eq!(poster.0.lock().expect("unpoisoned").len(), 3);
+
+        // A finished run frees its head.
+        start_at("a").expect("accepted again").await.expect("ran");
+        assert_eq!(poster.0.lock().expect("unpoisoned").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(Arc::new(tokio::sync::Semaphore::new(0)));
+        let poster = Arc::new(Posted::default());
+        let start_at = |head: String| {
+            start(
+                &reviews,
+                Arc::clone(&fake),
+                gate.clone(),
+                Arc::clone(&poster),
+                requested_at(&head),
+            )
+            .expect("a run or a comment")
+        };
+
+        let places = 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:** 2 reviews are running and 8 are waiting"),
+                "{}",
+                review.body
+            );
+        }
+
+        gate.open();
+        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"));
+    }
+
     #[tokio::test]
     async fn a_clean_run_posts_its_verdict_on_the_head_commit() {
         let (target, commit, review) = posted(
@@ -548,17 +906,28 @@ mod tests {
             .count();
         assert_eq!(unescaped, 8, "{line}");
         assert!(
-            line.contains("\\!\\[\\]") && line.contains("\\<b\\>"),
+            line.contains("\\!\u{200b}\\[\\]") && line.contains("\\<b\\>"),
             "{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" },
@@ -574,6 +943,13 @@ mod tests {
             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")),