Watch
0
0
Fork
You've already forked hyperhive
0

swarm-controller: post the review classifier's verdict on a review request

Requesting the forge account `review-classifier` as a reviewer now runs the
repo's rules against the PR and posts the result as that account's review,
on the head commit the run read. A push starts no run; only a fresh review
request does.

- An instance-wide `pull_request_review_request` system hook delivers to
  `/webhook/forge/review-request`. A request for any other reviewer, or a
  removed request, does nothing.
- The controller reads the PR, loads the rules with the same reader as
  `GET /api/review-classifier/rules`, sends one request on the try-out
  subject, and posts APPROVED or REQUEST_CHANGES with the table as the body.
- With no clean verdict (PR or rules don't load, the run fails, or any
  classification has no answer) it posts a COMMENT saying why, never an
  approval. The event is one function, `no_clean_verdict_event`.
- Table cells are flattened, markup-escaped, and every `@` gets a
  zero-width space, so a model's reason can't break the table or mention
  anyone.
- At most two requested reviews run at once; the rest wait in memory.
- The account is minted the way hive accounts are: a five-minute pass
  queues a `MintClassifierForgeToken` node when the token at
  `swarm/services/review-classifier/forge/token` is missing or stale. The
  token is `write:repository`. The controller's bao policy gains that leaf
  (create/read/update), pinned in module-eval bao-grants. The classifier
  never sees the token.

The account gets no repo access automatically; a repo admin adds it by
hand. A re-request retires the account's earlier review: Forgejo clears
`official` on a reviewer's earlier reviews when it is requested again
(models/issues/review.go AddReviewRequest, v16.0.5), and approvals and
rejection blocks count official reviews only.

Refs #5137
This commit is contained in:
atlas 2026-10-12 14:28:42 +02:00
commit a669cd96b6
17 changed files with 1156 additions and 50 deletions

View file

@ -40,6 +40,7 @@ use crate::webhook::DeliveryKind;
use swarm_queue_client::review_classifier::Commit;
pub mod agent_token;
pub mod classifier_account;
pub mod docs_grant;
pub mod hive_token;
pub mod legacy_tokens;
@ -1289,6 +1290,9 @@ impl Client {
// in an org created after this hook was registered, with no
// per-org registration to keep in sync as orgs come and go.
DeliveryKind::VcsActivity => ("push", HookScope::Instance),
// Instance-wide too: any repo whose admin gave the
// classifier's account access can request its review.
DeliveryKind::ReviewRequest => ("pull_request_review_request", HookScope::Instance),
};
self.ensure_hook(&scope, &target_url, event, secret)
.await

View file

@ -0,0 +1,328 @@
//! The review classifier's forge account, [`LOGIN`]: created here with the
//! admin API and given one token, stored at
//! `swarm_secret_client::review_classifier::forge_token_path`. Requesting
//! this account as a reviewer starts a classifier run, and this daemon posts
//! the result as the account's review (`crate::review_classifier::review`).
//! The classifier itself never holds the token.
//!
//! The account gets no repo access here. A repo admin adds it to a repo by
//! hand: read to request it as a reviewer, write (or a place on the branch
//! protection's approvals whitelist) for its approval to count.
//!
//! The token decision is [`super::agent_token`]'s, under this module's name
//! and scopes, so a stored token the forge still lists with the same last
//! eight characters and scopes is kept, and a pass whose reads fail queues
//! nothing.
use std::sync::Arc;
use anyhow::{Context, Result, bail};
use forgejo_api::structs::{CreatePullReviewOptions, RepoSearchQuery};
use forgejo_api::{Auth, Forgejo};
use swarm_secret_client::forge;
use swarm_secret_client::review_classifier::forge_token_path;
use super::Client;
use super::agent_token::{Decision, Observed, classify_named};
/// The classifier's forge login: the reviewer someone requests.
pub const LOGIN: &str = "review-classifier";
/// The forge's name for the account's one token.
pub const TOKEN_NAME: &str = "swarm-review-classifier";
/// What the token may do: post reviews, which the forge files under the
/// repository scope. Which repos is the account's access, granted by hand.
pub const TOKEN_SCOPES: &str = "write:repository";
/// How often [`spawn`] re-checks the account and its token.
const RECONCILE_INTERVAL: std::time::Duration = std::time::Duration::from_mins(5);
/// The email this module creates the account with, and the one it requires
/// before acting on an account named [`LOGIN`]. Under its own domain, so it
/// equals no agent's or hive's account email.
const EMAIL: &str = "review-classifier@services.hyperhive.local";
/// Whether a pass that found `seen` queues a `MintClassifierForgeToken` job.
/// [`Observed::Unknown`] never does, so a forge or store outage can't rotate
/// the token, and neither does a [`Decision::Conflict`], which the job would
/// refuse.
pub fn needs_job(seen: Observed) -> bool {
matches!(
seen,
Observed::NoForgeUser | Observed::Decided(Decision::Mint(_))
)
}
/// One review as the classifier's account posts it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Review {
/// `APPROVED`, `REQUEST_CHANGES` or `COMMENT`.
pub event: &'static str,
pub body: String,
}
impl Client {
/// Make sure [`LOGIN`] exists and holds a live [`TOKEN_NAME`] token in
/// the store, minting one when it does not. The whole job of the
/// `MintClassifierForgeToken` node.
///
/// # Errors
/// When the store or the forge refuses a step, when an account named
/// [`LOGIN`] exists with another email, or when the read-back does not
/// show the new token.
pub async fn ensure_classifier_forge_token(&self) -> Result<()> {
let path = forge_token_path()?;
self.ensure_classifier_user().await?;
let store = crate::store::connect()
.await
.context("logging in to the swarm secret store")?;
let (stored, listed) = self.read_classifier_token(&store, &path).await?;
let reason = match classify_named(TOKEN_NAME, TOKEN_SCOPES, stored.as_ref(), &listed) {
Decision::Keep => {
tracing::debug!("review classifier forge token is current; left as it is");
return Ok(());
}
Decision::Conflict => bail!(
"the forge lists more than one {TOKEN_NAME:?} token for {LOGIN:?}; delete them \
by id and re-run"
),
Decision::Mint(reason) => reason,
};
let credential = self
.replace_token(&store, LOGIN, TOKEN_NAME, TOKEN_SCOPES, &listed, &path)
.await?;
Forgejo::new(Auth::Token(&credential.value), self.url.clone())
.context("building a forge client with the new token")?
.repo_search(RepoSearchQuery::default())
.await
.with_context(|| format!("using {LOGIN}'s new forge token"))?;
let (stored, listed) = self.read_classifier_token(&store, &path).await?;
let now = classify_named(TOKEN_NAME, TOKEN_SCOPES, stored.as_ref(), &listed);
if now != Decision::Keep {
bail!("{LOGIN}'s new forge token does not read back as current: {now:?}");
}
tracing::info!(?reason, %path, "review classifier forge token minted and stored");
Ok(())
}
/// Post `review` on PR `number` of `owner/repo` as [`LOGIN`], on
/// `commit` when one is given. Reads the token from the store on every
/// call.
///
/// # Errors
/// When the store holds no token yet, or the forge refuses the review.
pub async fn post_classifier_review(
&self,
owner: &str,
repo: &str,
number: i64,
commit: Option<&str>,
review: Review,
) -> Result<()> {
let path = forge_token_path()?;
let store = crate::store::connect()
.await
.context("logging in to the swarm secret store")?;
let Some(credential) = store
.read_optional::<forge::Credential>(&path)
.await
.with_context(|| format!("reading {path}"))?
else {
bail!("nothing at {path}: the classifier's forge token is not minted yet");
};
Forgejo::new(Auth::Token(&credential.value), self.url.clone())
.context("building a forge client with the classifier's token")?
.repo_create_pull_review(
owner,
repo,
number,
CreatePullReviewOptions {
body: Some(review.body),
comments: None,
commit_id: commit.map(str::to_owned),
event: Some(review.event.to_owned()),
},
)
.await
.with_context(|| {
format!("posting {LOGIN}'s review on PR {number} of {owner}/{repo}")
})?;
Ok(())
}
/// The stored token at `path` and every token the forge lists for
/// [`LOGIN`], which must exist.
async fn read_classifier_token(
&self,
store: &swarm_secret_client::SecretStore,
path: &str,
) -> Result<(
Option<forge::Credential>,
Vec<forgejo_api::structs::AccessToken>,
)> {
let stored = store
.read_optional(path)
.await
.with_context(|| format!("reading {path}"))?;
let Some(listed) = self.list_agent_tokens(LOGIN).await? else {
bail!("the forge has no user {LOGIN:?} after this job created it");
};
Ok((stored, listed))
}
/// Create [`LOGIN`] if there is none, refuse one with another email, and
/// lock it out of creating repos.
async fn ensure_classifier_user(&self) -> Result<()> {
self.create_user(LOGIN, EMAIL).await?;
let user = self
.api
.user_get(LOGIN)
.await
.with_context(|| format!("read forge user {LOGIN}"))?;
if user.email.as_deref() != Some(EMAIL) {
bail!(
"forge account `{LOGIN}` exists but is not the review classifier's (email {:?}, \
expected `{EMAIL}`); refusing to mint a token for it",
user.email
);
}
self.api
.admin_edit_user(LOGIN, super::repo_creation_lockdown(LOGIN))
.await
.with_context(|| format!("disable repo creation for forge user {LOGIN}"))?;
Ok(())
}
/// What the store and the forge say about the account's token.
async fn observe_classifier(&self) -> Result<Observed> {
let path = forge_token_path()?;
let store = crate::store::connect()
.await
.context("logging in to the swarm secret store")?;
let stored = match store.read_optional::<forge::Credential>(&path).await {
Ok(stored) => stored,
Err(e) => {
tracing::warn!(error = %e, "review classifier forge token: store read failed");
return Ok(Observed::Unknown);
}
};
Ok(match self.list_agent_tokens(LOGIN).await {
Ok(Some(listed)) => Observed::Decided(classify_named(
TOKEN_NAME,
TOKEN_SCOPES,
stored.as_ref(),
&listed,
)),
Ok(None) => Observed::NoForgeUser,
Err(e) => {
tracing::warn!(error = %format!("{e:#}"), "review classifier forge token: forge read failed");
Observed::Unknown
}
})
}
}
/// Check the account now and every [`RECONCILE_INTERVAL`] after, and call
/// `enqueue`, which inserts a `MintClassifierForgeToken` node, when it needs
/// a job. A pass that fails is logged and retried on the next tick.
pub fn spawn(client: Arc<Client>, enqueue: impl Fn() + Send + 'static) {
tokio::spawn(async move {
let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
loop {
ticker.tick().await;
let Some(outcome) = super::within_forge_deadline(
"review classifier forge token",
client.observe_classifier(),
)
.await
else {
continue;
};
match outcome {
Ok(seen) if needs_job(seen) => {
tracing::info!(?seen, "review classifier forge token: queueing a job");
enqueue();
}
Ok(seen) => tracing::debug!(?seen, "review classifier forge token: no job"),
Err(e) => tracing::warn!(
error = %format!("{e:#}"),
retry_in_s = RECONCILE_INTERVAL.as_secs(),
"review classifier forge token: pass failed; retrying next tick"
),
}
}
});
}
#[cfg(test)]
mod tests {
use super::super::agent_token::MintReason;
use super::*;
#[test]
fn the_token_writes_repositories_and_nothing_else() {
assert_eq!(TOKEN_SCOPES, "write:repository");
assert_ne!(TOKEN_NAME, super::super::agent_token::AGENT_TOKEN_NAME);
assert_ne!(TOKEN_NAME, super::super::hive_token::HIVE_TOKEN_NAME);
}
#[test]
fn a_mint_or_a_missing_account_queues_and_nothing_else_does() {
assert!(needs_job(Observed::NoForgeUser));
assert!(needs_job(Observed::Decided(Decision::Mint(
MintReason::NotStored
))));
assert!(!needs_job(Observed::Decided(Decision::Keep)));
assert!(!needs_job(Observed::Decided(Decision::Conflict)));
assert!(!needs_job(Observed::Unknown));
}
#[test]
fn the_accounts_email_is_no_agents() {
assert_ne!(EMAIL, super::super::agent_email(LOGIN));
assert_ne!(EMAIL, super::super::legacy_agent_email(LOGIN));
}
/// An existing account named like the classifier with an agent's email
/// is refused before the lockdown, so no token is minted for it.
#[tokio::test]
async fn the_job_refuses_an_account_that_is_not_the_classifiers() {
use super::super::tests::{lockdown_patch, stub_forge};
let (client, seen) = stub_forge(
(409, r#"{"message":"user already exists"}"#),
(200, r#"{"email":"review-classifier@hyperhive.local"}"#),
(200, "{}"),
)
.await;
let err = client
.ensure_classifier_user()
.await
.expect_err("an agent's account is not the classifier's");
assert!(format!("{err:#}").contains("not the review classifier's"));
assert!(lockdown_patch(&seen, LOGIN).is_none());
}
#[tokio::test]
async fn the_job_locks_down_the_classifiers_own_account() {
use super::super::tests::{lockdown_patch, stub_forge};
let (client, seen) = stub_forge(
(409, r#"{"message":"user already exists"}"#),
(
200,
r#"{"email":"review-classifier@services.hyperhive.local"}"#,
),
(200, "{}"),
)
.await;
client
.ensure_classifier_user()
.await
.expect("the classifier's own account");
let body = lockdown_patch(&seen, LOGIN).expect("no lockdown PATCH sent");
assert_eq!(body["max_repo_creation"], 0);
}
}

View file

@ -165,6 +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(),
}
}

View file

@ -135,6 +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(),
}
}

View file

@ -970,6 +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(),
}
}

View file

@ -147,6 +147,10 @@ enum SwarmNodeKind {
/// the `hives` team, and holds a live token in the swarm store, where
/// that hive's hive-c0re reads it. See `forge::hive_token`.
MintHiveForgeToken { hive: String },
/// Make sure the review classifier's forge account exists and holds a
/// live token in the swarm store, where this daemon reads it to post the
/// classifier's reviews. See `forge::classifier_account`.
MintClassifierForgeToken,
/// Re-mint `agent`'s queue secret if it is still stored and old enough.
/// See `agent_renewal`.
///
@ -187,6 +191,7 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
SwarmNodeKind::MintAgentMatrixAccount { .. } => "mint_agent_matrix_account".to_owned(),
SwarmNodeKind::MintHiveSenderToken { .. } => "mint_hive_sender_token".to_owned(),
SwarmNodeKind::MintHiveForgeToken { .. } => "mint_hive_forge_token".to_owned(),
SwarmNodeKind::MintClassifierForgeToken => "mint_classifier_forge_token".to_owned(),
SwarmNodeKind::RenewAgentQueueCredential { .. } => {
"renew_agent_queue_credential".to_owned()
}
@ -217,6 +222,7 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
}
SwarmNodeKind::MintHiveSenderToken { hive }
| SwarmNodeKind::MintHiveForgeToken { hive } => serde_json::json!({ "hive": hive }),
SwarmNodeKind::MintClassifierForgeToken => serde_json::json!({}),
SwarmNodeKind::TriggerDeploy { hive, agent, rev } => {
serde_json::json!({ "agent": agent, "hive": hive, "rev": rev })
}
@ -373,6 +379,7 @@ async fn run_swarm_node(
mint_hive_sender(deps.matrix_homeserver.as_deref(), &hive).await
}
SwarmNodeKind::MintHiveForgeToken { hive } => mint_hive_forge_token(forge, &hive).await,
SwarmNodeKind::MintClassifierForgeToken => mint_classifier_forge_token(forge).await,
SwarmNodeKind::RenewAgentQueueCredential { agent } => renew_queue_credential(&agent).await,
SwarmNodeKind::SetAgentWanted { hive, agent } => match deps.wanted {
None => Outcome::Failed(
@ -527,6 +534,30 @@ async fn mint_hive_forge_token(
}
}
/// The `MintClassifierForgeToken` arm, lifted out for the same reason.
async fn mint_classifier_forge_token(
forge: Option<Arc<forge::Client>>,
) -> hive_jobq::scheduler::Outcome {
use hive_jobq::scheduler::Outcome;
let Some(client) = forge else {
return Outcome::Failed(
"no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \
not in the swarm secret store), so no review classifier forge token can be minted"
.to_owned(),
);
};
match forge_node(
"mint_classifier_forge_token",
client.ensure_classifier_forge_token(),
)
.await
{
Ok(()) => Outcome::Done,
Err(e) => Outcome::Failed(format!("{e:#}")),
}
}
/// The `MintAgentMatrixAccount` arm, lifted out for the same reason.
async fn mint_matrix_account(
homeserver: Option<&str>,
@ -862,6 +893,16 @@ 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,
))
}
/// Env var the controller's NixOS module sets from
@ -2488,6 +2529,20 @@ fn queue_hive_forge_token_mints(
Ok(ids)
}
/// Insert one `MintClassifierForgeToken` job and return its node's id.
/// `forge::classifier_account::spawn`'s periodic pass is the one caller.
fn queue_classifier_forge_token_mint(
sched: &Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>,
) -> Result<Vec<hive_jobq::NodeId>> {
sched
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert_job(None, |b| {
vec![b.node(SwarmNodeKind::MintClassifierForgeToken).guid()]
})
.map_err(|e| anyhow::anyhow!("{e}"))
}
/// The homeserver agents' own matrix accounts are minted on. Unset is this
/// controller creating agents with no matrix account, which the mint node
/// reports by name rather than `main` refusing to start.
@ -2548,6 +2603,12 @@ fn spawn_forge_workers(
}
});
let sched = Arc::clone(jobq);
forge::classifier_account::spawn(Arc::clone(&client), move || {
if let Err(e) = queue_classifier_forge_token_mint(&sched) {
tracing::warn!(error = %format!("{e:#}"), "review classifier forge token: queueing failed");
}
});
let sched = Arc::clone(jobq);
forge::docs_grant::spawn(Arc::clone(&client), move |agents| {
if let Err(e) = queue_docs_grants(&sched, agents) {
tracing::warn!(error = %format!("{e:#}"), "docs grants: queueing failed");
@ -3214,6 +3275,7 @@ async fn main() -> Result<()> {
create_gate: Arc::default(),
accounts: linked_accounts::Accounts::Live,
try_outs: review_classifier::TryOuts::default(),
review_slots: review_slots(),
};
let app = build_app(state);
@ -3402,6 +3464,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(),
};
(state, sched)
}
@ -4316,6 +4379,10 @@ mod tests {
hive: "pr1ma".to_owned(),
},
),
(
"mint_classifier_forge_token",
SwarmNodeKind::MintClassifierForgeToken,
),
];
for (label, kind) in nodes {
let mut sched = hive_jobq::scheduler::Scheduler::new(

View file

@ -579,6 +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(),
}
}

View file

@ -14,6 +14,8 @@
//!
//! The operator is the `Remote-User` header the gateway sets, as for
//! `agent_input`; the classifier names them in its journal line.
//!
//! [`review`] runs the same rules on a review request and posts the result.
use std::collections::{BTreeMap, HashMap, HashSet};
use std::fmt::Write as _;
@ -35,6 +37,8 @@ use utoipa::{IntoParams, ToSchema};
use crate::forge::{self, DefaultBranchFiles, PullForReview};
use crate::{AppState, error_problem};
pub(crate) mod review;
/// How long a finished run's result stays readable.
pub(crate) const RESULT_TTL: Duration = Duration::from_hours(1);
@ -707,6 +711,12 @@ pub(crate) mod tests {
files: Vec<(&'static str, String)>,
}
impl Fake {
pub(crate) fn new(reply: Reply, files: Vec<(&'static str, String)>) -> Self {
Self { reply, files }
}
}
impl PullSource for Fake {
async fn pull(
&self,
@ -1017,6 +1027,15 @@ pub(crate) mod tests {
})
}
/// A JSON rules file holding the one `datetime` classification.
pub(crate) fn rules_json() -> (&'static str, String) {
let [_, json] = RULES_PATHS;
(
json,
rules_value(&serde_json::json!([datetime()]), &[]).to_string(),
)
}
/// `file` as both formats, each with its path.
fn both_formats(file: &serde_json::Value) -> [(&'static str, String); 2] {
let [yml, json] = RULES_PATHS;

View file

@ -0,0 +1,586 @@
//! 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; the rest wait here, in
//! memory, so a restart drops them.
use std::fmt::Write as _;
use std::future::Future;
use std::sync::Arc;
use serde::Deserialize;
use swarm_queue_client::review_classifier::{Overall, Reply, Row, TryOut, Verdict};
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;
/// 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,
/// Who asked for the review.
requester: String,
}
#[derive(Deserialize)]
struct Payload {
action: String,
number: i64,
requested_reviewer: Option<User>,
repository: Repository,
sender: Option<User>,
}
#[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)?,
requester: payload.sender.map(|s| s.login).unwrap_or_default(),
})
}
}
/// 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 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;
});
}
/// Where a review is posted: the forge in production, a fake in the tests.
pub(crate) trait ReviewPoster {
fn post(
&self,
owner: &str,
repo: &str,
number: i64,
commit: Option<&str>,
review: Review,
) -> impl Future<Output = anyhow::Result<()>> + Send;
}
impl ReviewPoster for forge::Client {
async fn post(
&self,
owner: &str,
repo: &str,
number: i64,
commit: Option<&str>,
review: Review,
) -> anyhow::Result<()> {
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
}
}
/// How far a run got.
#[derive(Debug, Clone, PartialEq, Eq)]
enum Outcome {
/// Nothing reached the classifier, for this reason.
NotRun(String),
Ran(Reply),
}
/// Read the PR and the rules, run them, and post the review.
async fn run(
files: &impl FileSource,
pulls: &impl PullSource,
sink: &impl RunSink,
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);
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),
)
.await;
match sent {
Some(Ok(())) => tracing::info!(
%repo, pr = req.pr, head_sha = head.as_deref().unwrap_or_default(), event,
"review classifier review posted"
),
Some(Err(e)) => tracing::warn!(
%repo, pr = req.pr, error = %format!("{e:#}"),
"posting the review classifier's review failed"
),
None => {}
}
}
/// 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,
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 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,
};
let outcome = match serde_json::to_vec(&request) {
Ok(payload) => Outcome::Ran(sink.run(payload).await),
Err(e) => Outcome::NotRun(format!("encoding the run failed: {e}")),
};
(head, outcome)
}
/// The review a run's outcome posts.
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 }) => (
no_clean_verdict_event(),
no_verdict(&format!("the run failed: {error}"), &[]),
),
Outcome::Ran(Reply::Done { rows, overall }) => {
let unanswered = rows.iter().filter(|r| r.error.is_some()).count();
if unanswered > 0 {
let reason = format!(
"{unanswered} of {} classifications got no answer",
rows.len()
);
(no_clean_verdict_event(), no_verdict(&reason, rows))
} else {
let (event, said) = match overall {
Overall::Approve => (Event::Approve, "approve"),
Overall::RequestChanges => (Event::RequestChanges, "request changes"),
};
(event, format!("{}\n**Overall: {said}.**\n", table(rows)))
}
}
};
Review {
event: event.forge(),
body,
}
}
/// 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 or raw HTML escaped, and every `@`
/// followed by a zero-width space, so no name after it is a live mention.
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_str("@\u{200b}"),
c => out.push(c),
}
}
}
out
}
#[cfg(test)]
mod tests {
use std::sync::Mutex;
use swarm_queue_client::review_classifier::Polarity;
use super::*;
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<()> {
self.0.lock().expect("unpoisoned").push((
format!("{owner}/{repo}#{number}"),
commit.map(str::to_owned),
review,
));
Ok(())
}
}
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,
requester: "alice".to_owned(),
}
}
/// 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()]);
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(reply, "r").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_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("\\!\\[\\]") && line.contains("\\<b\\>"),
"{line}"
);
}
#[test]
fn only_a_request_for_the_classifier_parses() {
let payload = |action: &str, reviewer: Option<&str>| {
serde_json::json!({
"action": action,
"number": 7,
"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"
);
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}");
}
}
}

View file

@ -221,6 +221,10 @@ pub(super) enum DeliveryKind {
/// `repo_create_hook`) — different scope, same event name.
#[strum(serialize = "vcs-activity")]
VcsActivity,
/// `pull_request_review_request` events on every repo in the instance,
/// for the review classifier's account (`crate::review_classifier::review`).
#[strum(serialize = "review-request")]
ReviewRequest,
}
/// The route prefix a registered `target_url` must point at.
@ -236,7 +240,12 @@ impl DeliveryKind {
/// Every kind, for the registration sweep. An array rather than a
/// hand-written list at the call site, so adding a kind cannot leave one
/// hook unregistered.
pub(super) const ALL: [Self; 3] = [Self::Knowledge, Self::ConfigPr, Self::VcsActivity];
pub(super) const ALL: [Self; 4] = [
Self::Knowledge,
Self::ConfigPr,
Self::VcsActivity,
Self::ReviewRequest,
];
/// The `target_url` to register with Forgejo for this kind, given the
/// swarm's public base URL.
@ -341,7 +350,9 @@ pub(super) fn verify(
/// [`announce_knowledge_change`]); a `ConfigPr` delivery updates
/// `crate::config_pr::ConfigPrCache` immediately (see that module's doc
/// comment for why both this and a periodic poll write the same cache), and a
/// merge queues a deploy of the merged commit on the agent's hive.
/// merge queues a deploy of the merged commit on the agent's hive. A
/// `ReviewRequest` delivery naming the review classifier's account starts a
/// review run in the background.
///
/// Returns 200 on an accepted delivery so Forgejo does not retry. A refused
/// one answers 401 (bad signature) or 503 (this daemon has no secret), and
@ -351,7 +362,8 @@ pub(super) fn verify(
path = "/webhook/forge/{kind}",
params(
("kind" = String, Path, description = "which swarm-wide hook this delivery arrived on: \
`knowledge` or `config-pr`")
`knowledge`, `config-pr`, `vcs-activity` or \
`review-request`")
),
request_body(
content = String,
@ -414,6 +426,7 @@ pub(super) async fn post_webhook_forge(
tracing::warn!("webhook: vcs-activity delivery did not parse as a push payload");
}
}
DeliveryKind::ReviewRequest => crate::review_classifier::review::on_delivery(&state, &body),
}
(StatusCode::OK, "ok").into_response()