swarm-nats-auth: a review classifier hears try-outs and answers by reply only

swarm-queue-client gains review_classifier: the try-out subject, the
request and reply that travel on it, and the deadlines both ends share.
The controller's reader grant may publish the subject; a client named
with --review-classifier-client may subscribe to it and its inbox, and
answers under NATS allow_responses (resp in the user JWT) within the
run deadline. Every other client id is still denied.

Cargo.lock

@@ -4928,6 +4928,7 @@ dependencies = [
  "thiserror 2.0.18",
  "tokio",
  "tracing",
+ "utoipa",
 ]
 
 [[package]]

swarm-nats-auth/src/main.rs

@@ -157,6 +157,11 @@ struct Args {
     #[arg(long = "agent-token-publish-subject")]
     agent_token_publish_subjects: Vec<String>,
 
+    /// 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<String>,
+
     /// 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")?;

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<Vec<String>>,
+    /// 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<std::time::Duration>,
 }
 
 /// Which client ids get which permissions.
@@ -45,6 +49,7 @@ pub struct Policy {
     readers: Vec<String>,
     extra_hive_subjects: Vec<String>,
     agent_token_subjects: Vec<String>,
+    review_classifiers: Vec<String>,
 }
 
 /// 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<String>) -> 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

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<Permission>,
+    /// The reply allowance; absent means none.
+    #[serde(rename = "resp", skip_serializing_if = "Option::is_none")]
+    responses: Option<ResponsePermission>,
     subs: i64,
     data: i64,
     payload: i64,
@@ -178,6 +181,25 @@ struct Permission {
     allow: Vec<String>,
 }
 
+/// 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<String>,
-    subscribe: Option<Vec<String>>,
+    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();

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,

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

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 {

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<Commit>,
+    /// The PR's unified diff, as the forge serves it.
+    pub diff: String,
+    pub classifications: Vec<Classification>,
+}
+
+/// 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<Row>,
+        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<String>,
+}
+
+#[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);
+    }
+}