diff --git a/Cargo.lock b/Cargo.lock index 6d7f6459..e7d1b608 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4928,6 +4928,7 @@ dependencies = [ "thiserror 2.0.18", "tokio", "tracing", + "utoipa", ] [[package]] diff --git a/swarm-nats-auth/src/main.rs b/swarm-nats-auth/src/main.rs index c4f3d3ba..31d7d2ce 100644 --- a/swarm-nats-auth/src/main.rs +++ b/swarm-nats-auth/src/main.rs @@ -157,6 +157,11 @@ struct Args { #[arg(long = "agent-token-publish-subject")] agent_token_publish_subjects: Vec, + /// Client ids that answer the review classifier's try-out requests. + /// Repeatable, empty by default, which denies such a client. + #[arg(long = "review-classifier-client")] + review_classifier_clients: Vec, + /// Role on the secret store's `cert` auth mount this responder logs in /// as to read the callout keys, its own OIDC client secret and agent /// credentials. The store is found @@ -220,7 +225,8 @@ async fn serve(args: &Args) -> anyhow::Result<()> { args.reader_clients.clone(), args.hive_publish_subjects.clone(), )? - .with_agent_token_subjects(args.agent_token_publish_subjects.clone())?; + .with_agent_token_subjects(args.agent_token_publish_subjects.clone())? + .with_review_classifiers(args.review_classifier_clients.clone()); let http = reqwest::Client::new(); let settings = swarm_secret_client::client::Settings::from_env() .context("reading the secret store's address and this responder's identity")?; diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 5615bdb3..b9d6db69 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -30,6 +30,10 @@ pub struct Permissions { /// subscribes to plus its own reply inbox. Never `Some(vec![])`: NATS /// reads an empty allow-list as no restriction at all. pub subscribe: Option>, + /// How long the client may answer a request it received, on that + /// request's reply subject, or `None` for no such allowance. NATS' + /// `allow_responses`: the reply subject need not be in `publish`. + pub responses: Option, } /// Which client ids get which permissions. @@ -45,6 +49,7 @@ pub struct Policy { readers: Vec, extra_hive_subjects: Vec, agent_token_subjects: Vec, + review_classifiers: Vec, } /// Placeholder replaced with the hive's own name in `extra_hive_subjects`. @@ -114,6 +119,7 @@ impl Policy { readers, extra_hive_subjects, agent_token_subjects: Vec::new(), + review_classifiers: Vec::new(), }) } @@ -135,6 +141,13 @@ impl Policy { Ok(self) } + /// Set the client ids that answer the review classifier's try-outs. + #[must_use] + pub fn with_review_classifiers(mut self, clients: Vec) -> Self { + self.review_classifiers = clients; + self + } + /// The permissions for an agent whose own credential was verified, or /// `None` when none are configured. /// @@ -153,6 +166,7 @@ impl Policy { format!("{}.>", swarm_queue_client::agent_inbox_prefix(agent)), swarm_queue_client::agent_input_subject(agent), ]), + responses: None, }) } @@ -182,17 +196,39 @@ impl Policy { return Some(Permissions { publish: self.hive_subjects(hive), subscribe: Some(Self::hive_subscribe(client_id, hive)), + responses: None, }); } if self.readers.iter().any(|r| r == client_id) { return Some(Permissions { publish: self.reader_subjects(), subscribe: Some(Self::reader_subscribe(client_id)), + responses: None, }); } + if self.review_classifiers.iter().any(|c| c == client_id) { + return Some(Self::review_classifier_permissions(client_id)); + } None } + /// What the review classifier may do: hear try-out requests and answer + /// each on its reply subject, for as long as a run may take. + /// + /// Its publish list is its own inbox, which reaches no other client: + /// NATS reads an empty allow-list as every subject, so the list carries + /// one entry even though the answers go out under `responses`. + fn review_classifier_permissions(client_id: &str) -> Permissions { + Permissions { + publish: vec![Self::inbox(client_id)], + subscribe: Some(vec![ + swarm_queue_client::review_classifier::SUBJECT.to_owned(), + Self::inbox(client_id), + ]), + responses: Some(swarm_queue_client::review_classifier::RUN_DEADLINE), + } + } + /// What a hive may subscribe to: the two events the controller sends it, /// and its own reply inbox. /// @@ -512,6 +548,9 @@ impl Policy { // `from` as given, which holds only while this client is the // family's sole publisher: no hive or agent grant may name it. swarm_queue_client::AGENT_INPUT_SUBJECT_WILDCARD.to_owned(), + // A review classifier try-out, sent as a request: the reply comes + // back on this client's inbox, which it already subscribes to. + swarm_queue_client::review_classifier::SUBJECT.to_owned(), // The wanted-state buckets — one per hive, all created and written // by this single client. // @@ -1091,6 +1130,81 @@ mod tests { } } + const CLASSIFIER: &str = "swarm-review-classifier"; + + fn with_classifier() -> Policy { + policy().with_review_classifiers(vec![CLASSIFIER.to_owned()]) + } + + #[test] + fn the_review_classifier_hears_its_subject_and_inbox_only_and_answers_by_reply() { + use swarm_queue_client::review_classifier::{RUN_DEADLINE, SUBJECT}; + use swarm_queue_client::{agent_input_subject, client_inbox_prefix}; + + let p = with_classifier() + .permissions(CLASSIFIER) + .expect("a configured classifier is admitted"); + let inbox = client_inbox_prefix(CLASSIFIER); + assert!(may_subscribe(&p, SUBJECT)); + assert!(may_subscribe(&p, &reply(&inbox))); + for refused in [ + ">".to_owned(), + "$SWARM.>".to_owned(), + "$SWARM.review-classifier.other".to_owned(), + "$SWARM.term.atlas".to_owned(), + agent_input_subject("atlas"), + reply(&client_inbox_prefix("swarm-controller")), + "_INBOX.nuid123.token456".to_owned(), + ] { + assert!( + !may_subscribe(&p, &refused), + "the classifier may subscribe to {refused}" + ); + } + // It publishes nothing but replies: its own inbox keeps the list + // non-empty, and reaches no other client. + assert_eq!(p.publish, vec![format!("{inbox}.>")]); + assert_eq!(p.responses, Some(RUN_DEADLINE)); + } + + #[test] + fn an_unknown_client_is_denied_with_a_review_classifier_configured() { + let p = with_classifier(); + assert_eq!(p.permissions("some-other-service"), None); + assert_eq!(p.permissions("swarm-review-classifier-2"), None); + // The classifier's id is nothing until it is configured. + assert_eq!(policy().permissions(CLASSIFIER), None); + } + + #[test] + fn only_the_reader_may_send_a_try_out() { + use swarm_queue_client::review_classifier::SUBJECT; + + let p = with_classifier() + .with_agent_token_subjects(vec!["$SWARM.term.{agent}".to_owned()]) + .expect("a per-agent template is valid"); + let reader = p.permissions("swarm-controller").expect("the reader"); + assert!(reader.publish.iter().any(|g| covers(g, SUBJECT))); + assert_eq!(reader.responses, None); + for (who, grant) in [ + ("a hive", p.permissions("hive-alpha").expect("a hive")), + ( + "the classifier", + p.permissions(CLASSIFIER).expect("admitted"), + ), + ( + "an agent", + p.agent_token_permissions("atlas").expect("configured"), + ), + ] { + assert!( + !grant.publish.iter().any(|g| covers(g, SUBJECT)), + "{who} may send a try-out: {:?}", + grant.publish + ); + } + } + #[test] fn the_prefix_alone_names_no_hive() { // `hive-` would otherwise produce `$KV.hive-status.`, a subject nobody diff --git a/swarm-nats-auth/src/respond.rs b/swarm-nats-auth/src/respond.rs index 8ae52c61..4e938349 100644 --- a/swarm-nats-auth/src/respond.rs +++ b/swarm-nats-auth/src/respond.rs @@ -160,6 +160,9 @@ struct UserNats { /// Subjects this client may subscribe to; absent means every subject. #[serde(rename = "sub", skip_serializing_if = "Option::is_none")] subscribe: Option, + /// The reply allowance; absent means none. + #[serde(rename = "resp", skip_serializing_if = "Option::is_none")] + responses: Option, subs: i64, data: i64, payload: i64, @@ -178,6 +181,25 @@ struct Permission { allow: Vec, } +/// NATS' `allow_responses`: the client may publish `max` messages to the +/// reply subject of each request it receives, within `ttl` nanoseconds (Go's +/// `time.Duration` in `jwt/v2`'s `ResponsePermission`). +#[derive(Serialize)] +struct ResponsePermission { + max: i64, + ttl: i64, +} + +impl ResponsePermission { + /// One answer per request, within `window`. + fn one_within(window: std::time::Duration) -> Self { + Self { + max: 1, + ttl: i64::try_from(window.as_nanos()).unwrap_or(i64::MAX), + } + } +} + /// Mint the user JWT an admitted client presents. /// /// `account` is the account **name** from the server's `accounts` block (the @@ -188,8 +210,7 @@ fn user_jwt( issuer: &KeyPair, account: &str, user_nkey: &str, - publish: Vec, - subscribe: Option>, + permissions: &crate::policy::Permissions, ) -> String { sign( Claims { @@ -201,8 +222,14 @@ fn user_jwt( aud: Some(account.to_owned()), nats: UserNats { kind: "user", - publish: Permission { allow: publish }, - subscribe: subscribe.map(|allow| Permission { allow }), + publish: Permission { + allow: permissions.publish.clone(), + }, + subscribe: permissions + .subscribe + .clone() + .map(|allow| Permission { allow }), + responses: permissions.responses.map(ResponsePermission::one_within), subs: -1, data: -1, payload: -1, @@ -228,14 +255,7 @@ pub fn grant( permissions: &crate::policy::Permissions, ) -> String { let now = now_secs(); - let jwt = user_jwt( - now, - issuer, - account, - user_nkey, - permissions.publish.clone(), - permissions.subscribe.clone(), - ); + let jwt = user_jwt(now, issuer, account, user_nkey, permissions); response(now, issuer, server_id, user_nkey, Ok(jwt)) } @@ -359,6 +379,7 @@ mod tests { crate::policy::Permissions { publish: vec!["$KV.hive-status.alpha".to_owned()], subscribe: None, + responses: None, } } @@ -401,6 +422,27 @@ mod tests { assert!(user["nats"]["sub"].is_null(), "{}", user["nats"]); } + #[test] + fn the_issued_user_jwt_carries_a_reply_allowance_only_when_given_one() { + let account = KeyPair::new_account(); + let answering = crate::policy::Permissions { + responses: Some(std::time::Duration::from_mins(31)), + ..perms() + }; + let wrapper = decode_claims(&grant(&account, "APP", "NSERVER", "UCLIENT", &answering)); + let user = decode_claims(wrapper["nats"]["jwt"].as_str().expect("a user jwt")); + // The names and units nats-server's `jwt/v2` reads: `max` messages, + // `ttl` as a Go duration in nanoseconds. + assert_eq!( + user["nats"]["resp"], + serde_json::json!({ "max": 1, "ttl": 31 * 60 * 1_000_000_000_i64 }) + ); + + let wrapper = decode_claims(&grant(&account, "APP", "NSERVER", "UCLIENT", &perms())); + let user = decode_claims(wrapper["nats"]["jwt"].as_str().expect("a user jwt")); + assert!(user["nats"]["resp"].is_null(), "{}", user["nats"]); + } + #[test] fn a_grant_carries_a_user_jwt_and_no_error() { let account = KeyPair::new_account(); diff --git a/swarm-nats-auth/src/xkey.rs b/swarm-nats-auth/src/xkey.rs index d7052136..b68bbcb4 100644 --- a/swarm-nats-auth/src/xkey.rs +++ b/swarm-nats-auth/src/xkey.rs @@ -246,6 +246,7 @@ pub(crate) mod tests { let permissions = crate::policy::Permissions { publish: vec!["$KV.hive-status.alpha".to_owned()], subscribe: None, + responses: None, }; let token = crate::respond::grant( &issuer, diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index 8a549768..1123f9a3 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -44,6 +44,9 @@ agent-schedules = ["async-nats/jetstream"] # `agent_inference::read_own`, for the agent harness, which reads its own # inference key the same way. Explicit for the same reason as `notices`. agent-inference = ["async-nats/jetstream"] +# `ToSchema` on the review classifier's shared types, for swarm-controller, +# whose HTTP API serves the same shapes the queue carries. +schema = ["dep:utoipa"] [dependencies] # Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature @@ -68,6 +71,7 @@ strum.workspace = true thiserror.workspace = true tokio.workspace = true tracing.workspace = true +utoipa = { workspace = true, optional = true } [lints] workspace = true diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 87c14990..fa5b991b 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -392,6 +392,11 @@ pub mod agent_term; /// feature; the names are unconditional, like [`agent_icon`]'s. pub mod subagent_term; +/// The review classifier's try-out subject and the request and reply it +/// carries. The `schema` feature adds `ToSchema` for an HTTP API serving the +/// same shapes. +pub mod review_classifier; + /// Only the fields this needs; authelia returns several. #[derive(serde::Deserialize)] struct TokenResponse { diff --git a/swarm-queue-client/src/review_classifier.rs b/swarm-queue-client/src/review_classifier.rs new file mode 100644 index 00000000..bdc216a7 --- /dev/null +++ b/swarm-queue-client/src/review_classifier.rs @@ -0,0 +1,168 @@ +//! The review classifier's try-out: the subject swarm-controller sends a run +//! on, and what travels each way. +//! +//! Three crates name these: the controller sends the request, the classifier +//! answers it, and the auth-callout responder grants the one subject to each. + +use std::time::Duration; + +use serde::{Deserialize, Serialize}; + +/// The subject a try-out request goes to. The controller is its one +/// publisher and the classifier its one subscriber; the classifier answers +/// on the request's reply subject. +pub const SUBJECT: &str = "$SWARM.review-classifier.try-out"; + +/// The queue group the classifier subscribes in, so a second instance shares +/// the requests instead of running each one twice. +pub const QUEUE_GROUP: &str = "review-classifier"; + +/// The most classifications one run carries. +pub const MAX_CLASSIFICATIONS: usize = 10; + +/// The longest one classification's model run may take before it is stopped +/// and its row carries a timeout. +pub const CLASSIFICATION_DEADLINE: Duration = Duration::from_mins(3); + +/// The longest a whole run may take: every classification at its deadline, +/// run one after another, plus a minute for the rest. The controller waits +/// this long for the reply, and the queue lets the classifier answer for +/// this long. +pub const RUN_DEADLINE: Duration = Duration::from_mins(31); + +/// One run: the PR as the controller read it, and the classifications to +/// answer about it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct TryOut { + /// The swarm UI user who started the run, as the gateway named them. + pub operator: String, + /// `owner/name`. + pub repo: String, + pub pr: u64, + /// The commit the commits and the diff were read at. + pub head_sha: String, + /// Oldest first. + pub commits: Vec, + /// The PR's unified diff, as the forge serves it. + pub diff: String, + pub classifications: Vec, +} + +/// One commit of the PR. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Commit { + pub sha: String, + /// The full message, subject and body. + pub message: String, +} + +/// One yes/no rule to check the PR against. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +pub struct Classification { + pub name: String, + /// The yes/no question the model answers about the PR. + pub description: String, + pub polarity: Polarity, + /// The certainty, 0 to 100, a verdict needs to count. + pub threshold: u8, +} + +/// Which verdict is the wanted one. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +#[serde(rename_all = "lowercase")] +pub enum Polarity { + /// A match is the wanted outcome. + Positive, + /// A match is a violation. + Negative, +} + +/// The classifier's answer to one [`TryOut`]. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "state", rename_all = "kebab-case")] +pub enum Reply { + Done { + rows: Vec, + overall: Overall, + }, + /// Nothing ran to a verdict: no API key, or every classification failed. + Failed { + error: String, + }, +} + +/// One classification's result, as the result table shows it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +pub struct Row { + pub name: String, + pub polarity: Polarity, + pub verdict: Verdict, + pub certainty: u8, + pub threshold: u8, + /// Whether the row counts toward the overall verdict: applicable, and + /// its certainty reached the threshold. + pub counted: bool, + /// For a counted row, whether the verdict is on the wanted side of the + /// polarity. `false` for every row that is not counted. + pub passed: bool, + pub reason: String, + /// Why this classification has no answer. Such a row is never counted. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +#[serde(rename_all = "kebab-case")] +pub enum Verdict { + Match, + NoMatch, + NotApplicable, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +#[serde(rename_all = "kebab-case")] +pub enum Overall { + Approve, + RequestChanges, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_wire_spells_the_enums_as_the_api_does() { + let row = Row { + name: "datetime".to_owned(), + polarity: Polarity::Positive, + verdict: Verdict::NotApplicable, + certainty: 95, + threshold: 70, + counted: false, + passed: false, + reason: "The PR adds no timestamp.".to_owned(), + error: None, + }; + let json = serde_json::to_value(Reply::Done { + rows: vec![row], + overall: Overall::RequestChanges, + }) + .expect("encodes"); + assert_eq!(json["state"], "done"); + assert_eq!(json["overall"], "request-changes"); + assert_eq!(json["rows"][0]["polarity"], "positive"); + assert_eq!(json["rows"][0]["verdict"], "not-applicable"); + assert!(json["rows"][0].get("error").is_none()); + } + + #[test] + fn the_run_deadline_covers_every_classification_at_its_deadline() { + let all = CLASSIFICATION_DEADLINE * u32::try_from(MAX_CLASSIFICATIONS).expect("small"); + assert!(RUN_DEADLINE > all); + } +}