swarm-controller: start a review try-out on a PR and read its result

POST /api/review-classifier/try-out validates the run (owner/name repo,
a PR number, one to ten classifications with distinct names, thresholds
up to 100), reads the PR's head commit, commits and diff with the
controller's forge token, sends one request to the review classifier on
the queue and answers 202 with an id. GET .../try-out/{id} reports the
run as running, done (rows and overall verdict) or failed (the error).
Results live in memory for an hour after the reply; there is no run
store. POST then GET because a run can outlast the gateway's 60-second
proxy read timeout. Both routes are utoipa-annotated, so they show on
/api/docs/.

Cargo.lock

@@ -4850,6 +4850,7 @@ dependencies = [
  "swarm-queue-client",
  "swarm-secret-client",
  "tokio",
+ "tower",
  "tracing",
  "url",
  "utoipa",

swarm-controller/Cargo.toml

@@ -94,10 +94,13 @@ swarm-authelia-bridge-sock.workspace = true
 # config the crate spells.
 # `agent-term`: likewise every agent's terminal stream, plus the consumer it
 # replays that stream through.
+# `schema`: the review classifier try-out API serves the shapes the queue
+# carries.
 swarm-queue-client = { workspace = true, features = [
   "kv",
   "subagent-term",
   "agent-term",
+  "schema",
 ] }
 # `matrix_account.rs` writes the credential this daemon's route accepts. Same
 # crate the hive reads it back with, which is the point: the path, the field
@@ -132,6 +135,8 @@ x509-cert.workspace = true
 # tests wait out its five-minute bound on paused time.
 [dev-dependencies]
 tokio = { workspace = true, features = ["test-util"] }
+# `ServiceExt::oneshot`, to drive a route through the real router in tests.
+tower = { version = "0.5", features = ["util"] }
 
 [lints]
 workspace = true

swarm-controller/src/forge.rs

@@ -24,8 +24,9 @@ use forgejo_api::structs::{
     ChangeFileOperationOperation, ChangeFilesOptions, CreateBranchProtectionOption,
     CreateHookOption, CreateHookOptionConfig, CreateHookOptionType, CreateRepoOption,
     CreateUserOption, EditUserOption, IssueListIssuesQuery, IssueListIssuesQueryState,
-    IssueListIssuesQueryType, RepoGetContentsQuery, RepoListPullRequestsQuery,
+    IssueListIssuesQueryType, RepoDownloadPullDiffOrPatchQuery, RepoGetContentsQuery,
-    RepoListPullRequestsQueryState, RepoSearchQuery, StateType,
+    RepoGetPullRequestCommitsQuery, RepoListPullRequestsQuery, RepoListPullRequestsQueryState,
+    RepoSearchQuery, StateType,
 };
 use forgejo_api::{ApiErrorKind, Auth, Forgejo, ForgejoError};
 use futures_util::{StreamExt as _, TryStreamExt as _};
@@ -36,6 +37,7 @@ use std::sync::{Arc, OnceLock};
 use utoipa::ToSchema;
 
 use crate::webhook::DeliveryKind;
+use swarm_queue_client::review_classifier::Commit;
 
 pub mod agent_token;
 pub mod docs_grant;
@@ -55,6 +57,14 @@ pub struct ConfigPrStatus {
     pub html_url: Option<String>,
 }
 
+/// A PR as [`Client::pull_for_review`] reads it for the review classifier.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct PullForReview {
+    pub head_sha: String,
+    pub commits: Vec<Commit>,
+    pub diff: String,
+}
+
 /// One row of [`Client::issue_report`] — an open issue plus the two facts
 /// derived from its dependency graph that swarm-ui's issue-report page
 /// sorts/filters on. See that function's doc comment for how both are
@@ -1031,6 +1041,67 @@ impl Client {
         Ok(rows.into_iter().flatten().collect())
     }
 
+    /// What the review classifier reads of `owner/repo`'s PR `number`: its
+    /// head commit, its commits oldest first with their full messages, and
+    /// its unified diff.
+    ///
+    /// # Errors
+    /// A forge failure, with the [`ForgejoError`] underneath; a 404 on the
+    /// first call means the repo or the PR does not exist
+    /// ([`is_repo_not_found`]).
+    pub async fn pull_for_review(
+        &self,
+        owner: &str,
+        repo: &str,
+        number: i64,
+    ) -> Result<PullForReview> {
+        let pull = self
+            .api
+            .repo_get_pull_request(owner, repo, number)
+            .await
+            .with_context(|| format!("read PR {number} of {owner}/{repo}"))?;
+        let head_sha = pull
+            .head
+            .and_then(|h| h.sha)
+            .with_context(|| format!("PR {number} of {owner}/{repo} names no head commit"))?;
+        let commits = self
+            .api
+            .repo_get_pull_request_commits(
+                owner,
+                repo,
+                number,
+                RepoGetPullRequestCommitsQuery {
+                    verification: Some(false),
+                    files: Some(false),
+                },
+            )
+            .all()
+            .await
+            .with_context(|| format!("list the commits of PR {number} of {owner}/{repo}"))?
+            .into_iter()
+            .map(|c| Commit {
+                sha: c.sha.unwrap_or_default(),
+                message: c.commit.and_then(|m| m.message).unwrap_or_default(),
+            })
+            .collect();
+        let diff = self
+            .api
+            .repo_download_pull_diff_or_patch(
+                owner,
+                repo,
+                number,
+                "diff",
+                RepoDownloadPullDiffOrPatchQuery::default(),
+            )
+            .await
+            .with_context(|| format!("read the diff of PR {number} of {owner}/{repo}"))?;
+        Ok(PullForReview {
+            head_sha,
+            commits,
+            diff,
+        })
+    }
+
     /// Register the swarm-wide hooks against this controller, so a real
     /// forge event reaches [`crate::webhook`] instead of the endpoint only
     /// being reachable by hand.

swarm-controller/src/forge_account.rs

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

swarm-controller/src/github_account.rs

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

swarm-controller/src/linked_accounts.rs

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

swarm-controller/src/main.rs

@@ -59,6 +59,7 @@ mod matrix_account;
 mod otel_http_client;
 mod queue_identity;
 mod read_policy;
+mod review_classifier;
 mod schedules;
 mod status;
 mod store;
@@ -735,6 +736,7 @@ fn socket_path() -> PathBuf {
         (name = "repos", description = "forge repo browsing (swarm-ui's issue-report page)"),
         (name = "forge", description = "forge site admins (`swarmctl forge make-admin`)"),
         (name = "schedules", description = "scheduled prompts (`swarmctl schedule`)"),
+        (name = "review-classifier", description = "trying review classifications against a PR (swarm-ui's try-out page)"),
     )
 )]
 struct ApiDoc;
@@ -857,6 +859,9 @@ struct AppState {
     create_gate: Arc<tokio::sync::Mutex<()>>,
     /// The store the linked-account routes read and write.
     accounts: linked_accounts::Accounts,
+    /// The review classifier try-outs started here, until their results
+    /// expire.
+    try_outs: review_classifier::TryOuts,
 }
 
 /// Env var the controller's NixOS module sets from
@@ -3208,6 +3213,7 @@ async fn main() -> Result<()> {
         forge: forge_client,
         create_gate: Arc::default(),
         accounts: linked_accounts::Accounts::Live,
+        try_outs: review_classifier::TryOuts::default(),
     };
 
     let app = build_app(state);
@@ -3273,6 +3279,8 @@ fn build_app(state: AppState) -> axum::Router {
         .routes(routes!(issue_report::get_repos))
         .routes(routes!(issue_report::get_issue_report_all))
         .routes(routes!(issue_report::get_issue_report))
+        .routes(routes!(review_classifier::post_try_out))
+        .routes(routes!(review_classifier::get_try_out))
         .routes(routes!(webhook::post_webhook_forge))
         .split_for_parts();
     // Just the JSON, not the UI — Swagger UI itself is nginx-hosted from
@@ -3361,7 +3369,7 @@ mod tests {
         >,
     >;
 
-    fn state_with_roster() -> (super::AppState, SharedSched) {
+    pub(crate) fn state_with_roster() -> (super::AppState, SharedSched) {
         let sched =
             std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new(
                 hive_jobq::Graph::new(),
@@ -3392,6 +3400,7 @@ mod tests {
             forge: std::sync::Arc::default(),
             create_gate: std::sync::Arc::default(),
             accounts: super::linked_accounts::Accounts::Fake(std::sync::Arc::default()),
+            try_outs: super::review_classifier::TryOuts::default(),
         };
         (state, sched)
     }

swarm-controller/src/matrix_account.rs

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

swarm-controller/src/review_classifier.rs

@@ -0,0 +1,731 @@
+//! `POST /api/review-classifier/try-out` and
+//! `GET /api/review-classifier/try-out/{id}`: run classifications against a
+//! PR from the swarm UI, then read the result back.
+//!
+//! The POST reads the PR with this daemon's forge token, sends one request to
+//! the review classifier on [`SUBJECT`] and answers 202 with an id at once:
+//! one model call over a big diff can outlast the 60 seconds the gateway
+//! holds a request open. The reply is kept in memory, where the GET reads it
+//! until [`RESULT_TTL`] after it arrived. Nothing reaches the forge.
+//!
+//! The operator is the `Remote-User` header the gateway sets, as for
+//! `agent_input`; the classifier names them in its journal line.
+
+use std::collections::{HashMap, HashSet};
+use std::fmt::Write as _;
+use std::future::Future;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use axum::Json;
+use axum::extract::rejection::JsonRejection;
+use axum::extract::{Path, State};
+use axum::http::{HeaderMap, StatusCode};
+use problem_details::ProblemDetails;
+use serde::{Deserialize, Serialize};
+use swarm_queue_client::review_classifier::{
+    Classification, MAX_CLASSIFICATIONS, Overall, RUN_DEADLINE, Reply, Row, SUBJECT, TryOut,
+};
+use utoipa::ToSchema;
+
+use crate::forge::{self, PullForReview};
+use crate::{AppState, error_problem};
+
+/// How long a finished run's result stays readable.
+pub(crate) const RESULT_TTL: Duration = Duration::from_hours(1);
+
+/// The header the gateway names the session's user in.
+const OPERATOR_HEADER: &str = "remote-user";
+
+/// A run to start.
+#[derive(Debug, Deserialize, ToSchema)]
+pub(crate) struct TryOutRequest {
+    /// `owner/name`.
+    repo: String,
+    pr: u64,
+    /// One to ten, each with a distinct name.
+    classifications: Vec<Classification>,
+}
+
+/// The started run's handle.
+#[derive(Debug, Serialize, ToSchema)]
+pub(crate) struct TryOutAccepted {
+    id: String,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, ToSchema)]
+#[serde(rename_all = "lowercase")]
+pub(crate) enum RunState {
+    Running,
+    Done,
+    Failed,
+}
+
+/// A run as `GET /api/review-classifier/try-out/{id}` reports it.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, ToSchema)]
+pub(crate) struct TryOutStatus {
+    state: RunState,
+    /// The commit the PR was read at.
+    head_sha: String,
+    /// Set when `done`.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    rows: Option<Vec<Row>>,
+    /// Set when `done`.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    overall: Option<Overall>,
+    /// Set when `failed`.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    error: Option<String>,
+}
+
+impl TryOutStatus {
+    fn settled(head_sha: String, reply: Reply) -> Self {
+        match reply {
+            Reply::Done { rows, overall } => Self {
+                state: RunState::Done,
+                head_sha,
+                rows: Some(rows),
+                overall: Some(overall),
+                error: None,
+            },
+            Reply::Failed { error } => Self {
+                state: RunState::Failed,
+                head_sha,
+                rows: None,
+                overall: None,
+                error: Some(error),
+            },
+        }
+    }
+}
+
+/// The runs this daemon holds, by id.
+#[derive(Clone, Default)]
+pub(crate) struct TryOuts {
+    runs: Arc<Mutex<HashMap<String, Entry>>>,
+    /// Stands in for the forge and the queue when set.
+    #[cfg(test)]
+    fake: Option<Arc<tests::Fake>>,
+}
+
+struct Entry {
+    status: TryOutStatus,
+    started: Instant,
+    settled: Option<Instant>,
+}
+
+impl Entry {
+    /// Expired once [`RESULT_TTL`] has passed since the reply, or since the
+    /// latest a reply can come.
+    fn expired(&self, now: Instant) -> bool {
+        let since = self.settled.unwrap_or(self.started + RUN_DEADLINE);
+        now.saturating_duration_since(since) > RESULT_TTL
+    }
+}
+
+impl TryOuts {
+    fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<String, Entry>> {
+        self.runs
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+    }
+
+    fn insert(&self, id: String, head_sha: String) {
+        let now = Instant::now();
+        let mut runs = self.lock();
+        runs.retain(|_, e| !e.expired(now));
+        runs.insert(
+            id,
+            Entry {
+                status: TryOutStatus {
+                    state: RunState::Running,
+                    head_sha,
+                    rows: None,
+                    overall: None,
+                    error: None,
+                },
+                started: now,
+                settled: None,
+            },
+        );
+    }
+
+    fn settle(&self, id: &str, reply: Reply) {
+        if let Some(entry) = self.lock().get_mut(id) {
+            entry.status = TryOutStatus::settled(entry.status.head_sha.clone(), reply);
+            entry.settled = Some(Instant::now());
+        }
+    }
+
+    fn get(&self, id: &str) -> Option<TryOutStatus> {
+        let now = Instant::now();
+        self.lock()
+            .get(id)
+            .filter(|e| !e.expired(now))
+            .map(|e| e.status.clone())
+    }
+}
+
+/// Where the PR is read from: the forge in production, a fake in the tests.
+pub(crate) trait PullSource {
+    fn pull(
+        &self,
+        owner: &str,
+        repo: &str,
+        number: i64,
+    ) -> impl Future<Output = anyhow::Result<PullForReview>> + Send;
+}
+
+impl PullSource for forge::Client {
+    async fn pull(&self, owner: &str, repo: &str, number: i64) -> anyhow::Result<PullForReview> {
+        self.pull_for_review(owner, repo, number).await
+    }
+}
+
+/// Where a run is sent: the queue in production, a fake in the tests.
+pub(crate) trait RunSink: Clone + Send + Sync + 'static {
+    fn run(&self, request: Vec<u8>) -> impl Future<Output = Reply> + Send;
+}
+
+impl RunSink for async_nats::Client {
+    async fn run(&self, request: Vec<u8>) -> Reply {
+        use async_nats::RequestErrorKind;
+
+        let sent = async_nats::Request::new()
+            .payload(request.into())
+            .timeout(Some(RUN_DEADLINE));
+        let error = match self.send_request(SUBJECT, sent).await {
+            Ok(msg) => {
+                return serde_json::from_slice(&msg.payload).unwrap_or_else(|e| Reply::Failed {
+                    error: format!("the classifier's reply did not decode: {e}"),
+                });
+            }
+            Err(e) => match e.kind() {
+                RequestErrorKind::NoResponders => {
+                    "no review classifier is listening: deploy.review-classifier.enable is set on \
+                     no host, or its service is not connected to the swarm queue"
+                        .to_owned()
+                }
+                RequestErrorKind::TimedOut => format!(
+                    "the review classifier sent no answer within {} minutes",
+                    RUN_DEADLINE.as_secs() / 60
+                ),
+                _ => format!("sending the run to the review classifier failed: {e}"),
+            },
+        };
+        Reply::Failed { error }
+    }
+}
+
+/// Start a run against a PR.
+#[utoipa::path(
+    post,
+    path = "/api/review-classifier/try-out",
+    request_body = TryOutRequest,
+    responses(
+        (status = 202, description = "the run started; read it at \
+            `/api/review-classifier/try-out/{id}`", body = TryOutAccepted),
+        (status = 400, description = "the body is not a valid run: a repo that is not \
+            owner/name, no PR number, no classification or more than ten, a classification \
+            without a name or description, two with one name, or a threshold over 100 \
+            (problem+json)", body = String),
+        (status = 401, description = "the request carries no `Remote-User`, so there is no \
+            operator to name (problem+json)", body = String),
+        (status = 404, description = "the forge knows no such repo or PR (problem+json)",
+            body = String),
+        (status = 503, description = "no forge access or no swarm queue on this host, or the \
+            forge failed to answer (problem+json)", body = String),
+    ),
+    tag = "review-classifier"
+)]
+pub(crate) async fn post_try_out(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    body: Result<Json<TryOutRequest>, JsonRejection>,
+) -> Result<(StatusCode, Json<TryOutAccepted>), ProblemDetails> {
+    let operator = operator(&headers).ok_or_else(|| {
+        error_problem(
+            StatusCode::UNAUTHORIZED,
+            "no Remote-User header: the gateway names the operator, and this request did not \
+             come through it",
+        )
+    })?;
+    let Json(req) = body.map_err(|e| error_problem(StatusCode::BAD_REQUEST, &e.body_text()))?;
+    let run = Run::checked(req).map_err(|d| error_problem(StatusCode::BAD_REQUEST, &d))?;
+    #[cfg(test)]
+    if let Some(fake) = state.try_outs.fake.clone() {
+        return Box::pin(start(
+            &state.try_outs,
+            &*fake,
+            (*fake).clone(),
+            operator,
+            run,
+        ))
+        .await;
+    }
+    let Some(forge) = state.forge.get() else {
+        return Err(error_problem(
+            StatusCode::SERVICE_UNAVAILABLE,
+            "no forge access on this host yet",
+        ));
+    };
+    let Some(status) = state.status.as_ref() else {
+        return Err(error_problem(
+            StatusCode::SERVICE_UNAVAILABLE,
+            "no swarm queue is configured on this host",
+        ));
+    };
+    let queue = status.queue_client();
+    swarm_queue_client::ensure_connected(&queue).map_err(|e| {
+        error_problem(
+            StatusCode::SERVICE_UNAVAILABLE,
+            &swarm_queue_client::chain(&e),
+        )
+    })?;
+    Box::pin(start(&state.try_outs, forge.as_ref(), queue, operator, run)).await
+}
+
+/// Read a run's result.
+#[utoipa::path(
+    get,
+    path = "/api/review-classifier/try-out/{id}",
+    params(("id" = String, Path, description = "the id the POST answered with")),
+    responses(
+        (status = 200, description = "the run, running or settled", body = TryOutStatus),
+        (status = 404, description = "no run has this id, or its result expired \
+            (problem+json)", body = String),
+    ),
+    tag = "review-classifier"
+)]
+pub(crate) async fn get_try_out(
+    State(state): State<AppState>,
+    Path(id): Path<String>,
+) -> Result<Json<TryOutStatus>, ProblemDetails> {
+    state.try_outs.get(&id).map(Json).ok_or_else(|| {
+        error_problem(
+            StatusCode::NOT_FOUND,
+            "no try-out has this id: it never existed, or its result expired",
+        )
+    })
+}
+
+/// A request that passed every check, with its repo split.
+#[derive(Debug)]
+struct Run {
+    owner: String,
+    name: String,
+    pr: i64,
+    classifications: Vec<Classification>,
+}
+
+impl Run {
+    /// The request, or what is wrong with it.
+    fn checked(req: TryOutRequest) -> Result<Self, String> {
+        let (owner, name) = req
+            .repo
+            .split_once('/')
+            .filter(|(o, n)| segment(o) && segment(n))
+            .ok_or_else(|| format!("repo {:?} is not owner/name", req.repo))?;
+        let pr = i64::try_from(req.pr)
+            .ok()
+            .filter(|n| *n > 0)
+            .ok_or_else(|| format!("{} is not a PR number", req.pr))?;
+        let count = req.classifications.len();
+        if count == 0 || count > MAX_CLASSIFICATIONS {
+            return Err(format!(
+                "{count} classifications: a run carries 1 to {MAX_CLASSIFICATIONS}"
+            ));
+        }
+        let mut names = HashSet::new();
+        for c in &req.classifications {
+            let name = c.name.trim();
+            if name.is_empty() {
+                return Err("a classification has no name".to_owned());
+            }
+            if c.description.trim().is_empty() {
+                return Err(format!("classification {name} has no description"));
+            }
+            if c.threshold > 100 {
+                return Err(format!(
+                    "classification {name}'s threshold {} is over 100",
+                    c.threshold
+                ));
+            }
+            if !names.insert(name) {
+                return Err(format!("two classifications are named {name}"));
+            }
+        }
+        Ok(Self {
+            owner: owner.to_owned(),
+            name: name.to_owned(),
+            pr,
+            classifications: req.classifications,
+        })
+    }
+}
+
+/// One `owner` or `name` of a repo, as the forge allows them.
+fn segment(s: &str) -> bool {
+    !s.is_empty()
+        && s != "."
+        && s != ".."
+        && s.bytes()
+            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.'))
+}
+
+fn operator(headers: &HeaderMap) -> Option<String> {
+    headers
+        .get(OPERATOR_HEADER)
+        .and_then(|v| v.to_str().ok())
+        .map(str::trim)
+        .filter(|s| !s.is_empty())
+        .map(str::to_owned)
+}
+
+/// Read the PR, record the run and send it; the reply settles the run when
+/// it arrives.
+async fn start(
+    try_outs: &TryOuts,
+    pulls: &impl PullSource,
+    sink: impl RunSink,
+    operator: String,
+    run: Run,
+) -> Result<(StatusCode, Json<TryOutAccepted>), ProblemDetails> {
+    let repo = format!("{}/{}", run.owner, run.name);
+    let pull =
+        forge::within_forge_deadline("review try-out", pulls.pull(&run.owner, &run.name, run.pr))
+            .await
+            .ok_or_else(|| {
+                error_problem(
+                    StatusCode::SERVICE_UNAVAILABLE,
+                    "the forge did not answer within the deadline",
+                )
+            })?
+            .map_err(|e| pull_problem(&e, &repo, run.pr))?;
+    let request = TryOut {
+        operator,
+        repo,
+        pr: run.pr.unsigned_abs(),
+        head_sha: pull.head_sha.clone(),
+        commits: pull.commits,
+        diff: pull.diff,
+        classifications: run.classifications,
+    };
+    let payload = serde_json::to_vec(&request).map_err(|e| {
+        error_problem(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            &format!("encoding the run failed: {e}"),
+        )
+    })?;
+    let id = mint_id().map_err(|e| error_problem(StatusCode::INTERNAL_SERVER_ERROR, &e))?;
+    try_outs.insert(id.clone(), pull.head_sha);
+    let (runs, settle_id) = (try_outs.clone(), id.clone());
+    tokio::spawn(async move {
+        let reply = sink.run(payload).await;
+        runs.settle(&settle_id, reply);
+    });
+    tracing::info!(%id, repo = %request.repo, pr = request.pr, "review try-out started");
+    Ok((StatusCode::ACCEPTED, Json(TryOutAccepted { id })))
+}
+
+/// 404 when the forge knows no such repo or PR, 503 for anything else.
+fn pull_problem(e: &anyhow::Error, repo: &str, pr: i64) -> ProblemDetails {
+    match e.downcast_ref::<forgejo_api::ForgejoError>() {
+        Some(fe) if forge::is_repo_not_found(fe) => {
+            error_problem(StatusCode::NOT_FOUND, &format!("no PR {pr} in {repo}"))
+        }
+        _ => {
+            let detail = format!("{e:#}");
+            tracing::warn!(error = %detail, %repo, pr, "reading a PR for a review try-out failed");
+            error_problem(StatusCode::SERVICE_UNAVAILABLE, &detail)
+        }
+    }
+}
+
+/// A fresh run id: 32 hex digits from the kernel's CSPRNG.
+fn mint_id() -> Result<String, String> {
+    let mut bytes = [0u8; 16];
+    getrandom::fill(&mut bytes).map_err(|e| format!("drawing a run id: {e}"))?;
+    Ok(bytes.iter().fold(String::with_capacity(32), |mut id, b| {
+        let _ = write!(id, "{b:02x}");
+        id
+    }))
+}
+
+#[cfg(test)]
+pub(crate) mod tests {
+    use std::sync::Arc;
+
+    use axum::body::Body;
+    use axum::http::{Request, StatusCode};
+    use forgejo_api::{ApiErrorKind, ForgejoError};
+    use swarm_queue_client::review_classifier::{Overall, Polarity, Reply, Row, Verdict};
+    use tower::ServiceExt as _;
+
+    use super::*;
+
+    /// A forge holding one PR, `o/r` 7, and a classifier answering every run
+    /// with `reply`.
+    #[derive(Clone)]
+    pub(crate) struct Fake {
+        reply: Reply,
+    }
+
+    impl PullSource for Fake {
+        async fn pull(
+            &self,
+            owner: &str,
+            repo: &str,
+            number: i64,
+        ) -> anyhow::Result<PullForReview> {
+            if (owner, repo, number) == ("o", "r", 7) {
+                Ok(PullForReview {
+                    head_sha: "abc123".to_owned(),
+                    commits: Vec::new(),
+                    diff: String::new(),
+                })
+            } else {
+                Err(ForgejoError::ApiError(ApiErrorKind::NotFound { errors: None }.into()).into())
+            }
+        }
+    }
+
+    impl RunSink for Fake {
+        async fn run(&self, _request: Vec<u8>) -> Reply {
+            self.reply.clone()
+        }
+    }
+
+    fn done() -> Reply {
+        Reply::Done {
+            rows: vec![Row {
+                name: "datetime".to_owned(),
+                polarity: Polarity::Positive,
+                verdict: Verdict::Match,
+                certainty: 90,
+                threshold: 80,
+                counted: true,
+                passed: true,
+                reason: "Zones everywhere.".to_owned(),
+                error: None,
+            }],
+            overall: Overall::Approve,
+        }
+    }
+
+    fn app(reply: Reply) -> axum::Router {
+        let (mut state, _) = crate::tests::state_with_roster();
+        state.try_outs.fake = Some(Arc::new(Fake { reply }));
+        crate::build_app(state)
+    }
+
+    fn body(repo: &str, pr: u64) -> String {
+        serde_json::json!({
+            "repo": repo,
+            "pr": pr,
+            "classifications": [{
+                "name": "datetime",
+                "description": "All datetimes carry a zone.",
+                "polarity": "positive",
+                "threshold": 80,
+            }],
+        })
+        .to_string()
+    }
+
+    async fn post(app: &axum::Router, body: String) -> (StatusCode, serde_json::Value) {
+        let req = Request::post("/api/review-classifier/try-out")
+            .header("content-type", "application/json")
+            .header("remote-user", "alice")
+            .body(Body::from(body))
+            .expect("a request");
+        send(app, req).await
+    }
+
+    async fn get(app: &axum::Router, id: &str) -> (StatusCode, serde_json::Value) {
+        let req = Request::get(format!("/api/review-classifier/try-out/{id}"))
+            .body(Body::empty())
+            .expect("a request");
+        send(app, req).await
+    }
+
+    async fn send(app: &axum::Router, req: Request<Body>) -> (StatusCode, serde_json::Value) {
+        let res = app.clone().oneshot(req).await.expect("infallible");
+        let status = res.status();
+        let bytes = axum::body::to_bytes(res.into_body(), usize::MAX)
+            .await
+            .expect("a body");
+        (status, serde_json::from_slice(&bytes).unwrap_or_default())
+    }
+
+    /// The run the GET reports once the fake's reply has settled it.
+    async fn settled(app: &axum::Router, id: &str) -> serde_json::Value {
+        for _ in 0..100 {
+            let (status, run) = get(app, id).await;
+            assert_eq!(status, StatusCode::OK, "{run}");
+            if run["state"] != "running" {
+                return run;
+            }
+            tokio::task::yield_now().await;
+        }
+        panic!("run {id} never settled");
+    }
+
+    #[tokio::test]
+    async fn a_valid_run_is_202_with_an_id_the_get_reads_back() {
+        let app = app(done());
+        let (status, accepted) = post(&app, body("o/r", 7)).await;
+        assert_eq!(status, StatusCode::ACCEPTED, "{accepted}");
+        let id = accepted["id"].as_str().expect("an id");
+        let run = settled(&app, id).await;
+        assert_eq!(run["state"], "done");
+        assert_eq!(run["head_sha"], "abc123");
+        assert_eq!(run["overall"], "approve");
+        assert_eq!(run["rows"][0]["verdict"], "match");
+        assert!(run.get("error").is_none(), "{run}");
+    }
+
+    #[tokio::test]
+    async fn a_failed_run_reports_its_error_and_no_verdict() {
+        let app = app(Reply::Failed {
+            error: "no API key".to_owned(),
+        });
+        let (_, accepted) = post(&app, body("o/r", 7)).await;
+        let run = settled(&app, accepted["id"].as_str().expect("an id")).await;
+        assert_eq!(run["state"], "failed");
+        assert_eq!(run["error"], "no API key");
+        assert!(
+            run.get("overall").is_none() && run.get("rows").is_none(),
+            "{run}"
+        );
+    }
+
+    #[tokio::test]
+    async fn a_bad_body_is_400() {
+        let app = app(done());
+        let mut eleven = serde_json::from_str::<serde_json::Value>(&body("o/r", 7)).expect("json");
+        let one = eleven["classifications"][0].clone();
+        eleven["classifications"] = (0..11)
+            .map(|i| {
+                let mut c = one.clone();
+                c["name"] = format!("c{i}").into();
+                c
+            })
+            .collect();
+        let with = |field: &str, value: serde_json::Value| {
+            let mut b = serde_json::from_str::<serde_json::Value>(&body("o/r", 7)).expect("json");
+            b["classifications"][0][field] = value;
+            b.to_string()
+        };
+        let twice = {
+            let mut b = serde_json::from_str::<serde_json::Value>(&body("o/r", 7)).expect("json");
+            let c = b["classifications"][0].clone();
+            b["classifications"] = serde_json::json!([c.clone(), c]);
+            b.to_string()
+        };
+        for bad in [
+            "not json".to_owned(),
+            r#"{"repo":"o/r","pr":7}"#.to_owned(),
+            body("o", 7),
+            body("o/r/x", 7),
+            body("../r", 7),
+            body("o/r", 0),
+            r#"{"repo":"o/r","pr":7,"classifications":[]}"#.to_owned(),
+            eleven.to_string(),
+            with("polarity", "sideways".into()),
+            with("threshold", 101.into()),
+            with("name", " ".into()),
+            with("description", "".into()),
+            twice,
+        ] {
+            let (status, problem) = post(&app, bad.clone()).await;
+            assert_eq!(status, StatusCode::BAD_REQUEST, "{bad}: {problem}");
+            assert!(problem["detail"].is_string(), "{bad}: {problem}");
+        }
+    }
+
+    #[tokio::test]
+    async fn an_unknown_repo_or_pr_is_404() {
+        let app = app(done());
+        for (repo, pr) in [("o/r", 8), ("o/elsewhere", 7)] {
+            let (status, problem) = post(&app, body(repo, pr)).await;
+            assert_eq!(status, StatusCode::NOT_FOUND, "{repo} {pr}: {problem}");
+        }
+    }
+
+    #[tokio::test]
+    async fn an_unknown_id_is_404() {
+        let (status, problem) = get(&app(done()), "0123456789abcdef").await;
+        assert_eq!(status, StatusCode::NOT_FOUND, "{problem}");
+    }
+
+    #[tokio::test]
+    async fn a_run_without_an_operator_is_401() {
+        let req = Request::post("/api/review-classifier/try-out")
+            .header("content-type", "application/json")
+            .body(Body::from(body("o/r", 7)))
+            .expect("a request");
+        let (status, _) = send(&app(done()), req).await;
+        assert_eq!(status, StatusCode::UNAUTHORIZED);
+    }
+
+    /// The queue half against a real server: no classifier listening is an
+    /// error naming the option, and a listener's reply comes back decoded.
+    #[tokio::test]
+    async fn a_run_reaches_a_listening_classifier_and_fails_without_one() {
+        use futures_util::StreamExt as _;
+
+        let nats = crate::test_nats::TestNats::start();
+        let client = nats.client().await;
+        let unanswered = client.run(b"{}".to_vec()).await;
+        assert!(
+            matches!(&unanswered, Reply::Failed { error } if error.contains("deploy.review-classifier.enable")),
+            "{unanswered:?}"
+        );
+
+        let classifier = nats.client().await;
+        let mut requests = classifier
+            .queue_subscribe(
+                SUBJECT,
+                swarm_queue_client::review_classifier::QUEUE_GROUP.to_owned(),
+            )
+            .await
+            .expect("subscribe");
+        // `flush` only writes the buffer. A message the classifier sends itself
+        // and receives back proves the server processed the subscription
+        // before it, since the server reads one connection in order.
+        let mut probe = classifier
+            .subscribe("review-classifier-test.probe")
+            .await
+            .expect("subscribe");
+        classifier
+            .publish("review-classifier-test.probe", "".into())
+            .await
+            .expect("publish");
+        probe.next().await.expect("the probe comes back");
+        tokio::spawn(async move {
+            let msg = requests.next().await.expect("a request");
+            let reply = serde_json::to_vec(&done()).expect("encodes");
+            classifier
+                .publish(msg.reply.expect("a reply subject"), reply.into())
+                .await
+                .expect("publish");
+        });
+        assert_eq!(client.run(b"{}".to_vec()).await, done());
+    }
+
+    #[test]
+    fn a_result_expires_after_its_ttl() {
+        let now = Instant::now();
+        let entry = |settled: Option<Instant>| Entry {
+            status: TryOutStatus::settled("abc".to_owned(), done()),
+            started: now,
+            settled,
+        };
+        assert!(!entry(Some(now)).expired(now + RESULT_TTL));
+        assert!(entry(Some(now)).expired(now + RESULT_TTL + Duration::from_secs(1)));
+        // A run whose reply never came stays until the latest it could have.
+        assert!(!entry(None).expired(now + RUN_DEADLINE + RESULT_TTL));
+        assert!(entry(None).expired(now + RUN_DEADLINE + RESULT_TTL + Duration::from_secs(1)));
+    }
+}