review classifier: queued review requests survive a restart and are never refused

docs/swarm/review-classifier.md

@@ -118,14 +118,24 @@ the earlier one. A repo whose branch protection also dismisses stale approvals (
 on an older commit even without a re-request.
 
 How many requested reviews run at once is a swarm-wide setting, two by
-default (see [Swarm-level settings](#swarm-level-settings)). Up to eight more
-wait in the controller's memory, a bound that isn't configurable, so a
-controller restart drops them, and nothing gets posted for them. Waiting
-reviews start in no guaranteed order. A request for a PR and head that's
-already running, waiting or re-posting starts nothing, and the head stays
-taken until the controller writes its record. A request with every running place
-and all eight waiting places taken starts no run: the account posts a
-`**No verdict:**` comment saying so.
+default (see [Swarm-level settings](#swarm-level-settings)). The controller
+turns no request away. It puts each one in the `review-requests` work-queue
+stream on the swarm queue, one subject per PR, and the stream keeps one
+message per subject: a newer request for a PR replaces one still waiting.
+The controller takes a request once a run place is free and acks it after
+writing the run's record. While a run goes, the controller tells the queue
+every 20 seconds that the request is still in hand. A request it took and
+never acked, because the controller stopped first, comes back within a
+minute, and the controller that's running then reviews it. A restart
+therefore repeats a run that was going. When it can't queue a request, the
+account posts a `**No verdict:**` comment saying so.
+
+When the controller takes a request, it reads the PR first. It drops a
+request whose head is no longer the PR's head, and posts and records nothing
+for it.
+A request for a PR whose review is still going waits in the queue until that
+review ends, so a PR never has two runs at once. A request that re-posts an
+earlier verdict waits for a run place too, and frees it before it posts.
 
 ## Components
 
@@ -206,7 +216,7 @@ replaces it.
   runtime's own default. On the ACP runtime it applies only where the agent
   offers it
 - `max_concurrent_reviews`: 1 to 8, how many requested reviews run at once.
-  Try-outs don't count toward it. Up to eight more reviews wait for a place
+  Try-outs don't count toward it. The rest wait in the queue for a place
 - `runtime`: `api-key` or `acp` (see [Runtimes](#runtimes))
 - `acp`: the provider the ACP runtime runs on, or `null`. It stays stored
   while the runtime is `api-key`, so switching back and forth needs no
@@ -574,13 +584,12 @@ The route answers **201** with `branch`, `path` (the file written) and
 
 `swarm-controller` records every run when it ends: a try-out when the
 classifier's reply arrives, a requested review once the controller has posted
-its review or the post has failed, and a request turned away because the
-queue was full once its comment is on the forge or the post has failed. The
-controller records a re-posted verdict the same way, with the outcome and answer of
-the run it repeats and that run's id in its `detail`. A review request for a
-head that's already running, waiting or re-posting starts nothing and isn't
-recorded. A run still going isn't
-in the list, and a controller restart drops a run in progress.
+its review or the post has failed, and a request the controller couldn't queue
+once its comment is on the forge or the post has failed. The controller records a
+re-posted verdict the same way, with the outcome and answer of the run it
+repeats and that run's id in its `detail`. A dropped request, whose head the
+PR has moved on from, isn't recorded. A run still going isn't in the list; a
+controller restart stops it, and the queue hands its request back.
 
 The newest 100 runs are one JSON array under the key `recent` in the
 `review-classifier-runs` KV bucket on the swarm queue, which only the
@@ -602,11 +611,12 @@ header. Each run has:
 - `repo` (`owner/name`), `pr`, and `requester`: the swarm UI user who started
   the try-out, or the forge user who requested the review
 - `head_sha`: the commit the run read, absent when the PR wasn't read, as for
-  an unreadable PR or a full queue
+  an unreadable PR or a request the controller couldn't queue
 - `started_unix` and `finished_unix`: a try-out starts at its POST, a review
-  once it has a run place
-- `outcome`: `approved`, `changes-requested`, `no-verdict`, `queue-full` or
-  `post-failed` for a review; `done` or `failed` for a try-out
+  once it has a run place and has read the PR
+- `outcome`: `approved`, `changes-requested`, `no-verdict` or `post-failed`
+  for a review, or `queue-full` on a record written before the request queue;
+  `done` or `failed` for a try-out
 - `detail`: the reason, for every outcome but `approved`,
   `changes-requested` and `done`. For `post-failed` it's the run's own reason,
   when it has one, then why the post failed.

swarm-controller/README.md

@@ -69,8 +69,9 @@ The swarm's control plane. The swarm UI and `swarmctl` are its clients.
   the harness to write when no file exists →
   [`docs/swarm/README.md`](../docs/swarm/README.md#an-agents-agentsmd).
 - **Review classifier** — try-outs from the swarm UI and reviews on review
-  requests, and the newest 100 finished runs, one document in the
-  `review-classifier-runs` KV bucket, which this daemon alone writes
+  requests, the requested reviews not yet finished, in the `review-requests`
+  work-queue stream, and the newest 100 finished runs, one document in the
+  `review-classifier-runs` KV bucket, both of which this daemon alone writes
   (`POST /api/review-classifier/try-out`, `GET /api/review-classifier/try-out/{id}`,
   `GET /api/review-classifier/rules`, `GET /api/review-classifier/runs`), and
   the try-out page's branch for a PR, pushed as the classifier's forge account

swarm-controller/src/main.rs

@@ -235,7 +235,12 @@ async fn main() -> Result<()> {
         reviews: Arc::default(),
         alerts: alerts::from_env(),
     };
+    serve(listener, state).await
+}
 
+/// Start taking the queued review requests, then serve `state` on `listener`.
+async fn serve(listener: tokio::net::UnixListener, state: AppState) -> Result<()> {
+    review_classifier::review::spawn_queue(&state);
     axum::serve(listener, build_app(state))
         .await
         .context("serving swarm-controller")

swarm-controller/src/review_classifier/runs.rs

@@ -60,12 +60,14 @@ pub(crate) enum RunOutcome {
     Approved,
     /// A review requested changes.
     ChangesRequested,
-    /// A comment was posted: the rules did not load, the PR was unreadable,
-    /// the run failed, the classifier sent a different number of answers
-    /// than it was sent classifications, or any classification got none.
+    /// A comment was posted: the request could not be queued, the rules did
+    /// not load, the PR was unreadable, the run failed, the classifier sent a
+    /// different number of answers than it was sent classifications, or any
+    /// classification got none.
     NoVerdict,
     /// No run started, as every run and wait place was taken; a comment
-    /// saying so was posted.
+    /// saying so was posted. This build records none; it is here so that a
+    /// stored record of it still reads.
     QueueFull,
     /// The review could not be posted.
     PostFailed,

swarm-nats-auth/src/policy.rs

@@ -664,6 +664,25 @@ impl Policy {
         subjects
     }
 
+    /// The review requests the controller queues and takes
+    /// (`swarm_wire::review_classifier::REQUEST_STREAM`): publishing one on
+    /// its PR's subject, creating the controller's durable consumer, pulling
+    /// from it and acking what it pulled. `STREAM.INFO.*` and
+    /// `STREAM.CREATE.*` open the stream. The controller is the stream's one
+    /// publisher, so no other grant may name its subjects. A refused publish
+    /// reaches the client as a timeout.
+    fn review_request_subjects() -> [String; 4] {
+        use swarm_wire::review_classifier::{
+            REQUEST_CONSUMER, REQUEST_STREAM, REQUEST_SUBJECT_PREFIX,
+        };
+        [
+            format!("{REQUEST_SUBJECT_PREFIX}.>"),
+            format!("$JS.API.CONSUMER.CREATE.{REQUEST_STREAM}.{REQUEST_CONSUMER}"),
+            format!("$JS.API.CONSUMER.MSG.NEXT.{REQUEST_STREAM}.{REQUEST_CONSUMER}"),
+            format!("$JS.ACK.{REQUEST_STREAM}.{REQUEST_CONSUMER}.>"),
+        ]
+    }
+
     fn reader_subjects(&self) -> Vec<String> {
         let stream = self.stream();
         let mut subjects = Self::jetstream_minimum().to_vec();
@@ -774,6 +793,7 @@ impl Policy {
             swarm_wire::subagent_term::consumer_create_grant(),
             "$KV.*.>".to_owned(),
         ]);
+        subjects.extend(Self::review_request_subjects());
         // The per-agent status bucket, read-side only. Same five subjects as
         // the hive-status block above and for the same measured reasons —
         // including BOTH `CONSUMER.CREATE` forms, since `keys()` builds an
@@ -1594,6 +1614,56 @@ mod tests {
         }
     }
 
+    /// Queueing a review request, taking it through the controller's durable
+    /// consumer and acking it are the reader's, and nobody else's.
+    #[test]
+    fn only_the_reader_may_queue_and_take_review_requests() {
+        use swarm_wire::review_classifier::{
+            REQUEST_CONSUMER, REQUEST_STREAM, REQUEST_SUBJECT_PREFIX,
+        };
+
+        let subjects = [
+            format!("{REQUEST_SUBJECT_PREFIX}.hyperhive.hyperhive.5219"),
+            format!("$JS.API.STREAM.INFO.{REQUEST_STREAM}"),
+            format!("$JS.API.STREAM.CREATE.{REQUEST_STREAM}"),
+            format!("$JS.API.CONSUMER.CREATE.{REQUEST_STREAM}.{REQUEST_CONSUMER}"),
+            format!("$JS.API.CONSUMER.MSG.NEXT.{REQUEST_STREAM}.{REQUEST_CONSUMER}"),
+            format!("$JS.ACK.{REQUEST_STREAM}.{REQUEST_CONSUMER}.1.4.4.1761000000000000000.0"),
+        ];
+        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");
+        for subject in &subjects {
+            assert!(
+                reader.publish.iter().any(|g| covers(g, subject)),
+                "the reader may not publish {subject}"
+            );
+        }
+        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"),
+            ),
+        ] {
+            for subject in &subjects {
+                assert!(
+                    !grant.publish.iter().any(|g| covers(g, subject)),
+                    "{who} may publish {subject}: {:?}",
+                    grant.publish
+                );
+            }
+        }
+        // The consumer grants name the one consumer.
+        let other = format!("$JS.API.CONSUMER.MSG.NEXT.{REQUEST_STREAM}.another");
+        assert!(!reader.publish.iter().any(|g| covers(g, &other)));
+    }
+
     const BUILDER: &str = "swarm-builder";
 
     #[test]

swarm-wire/src/review_classifier.rs

@@ -1,6 +1,7 @@
 //! The review classifier's runs: the subject swarm-controller sends a run
 //! on, and what travels each way. A try-out and a requested review are the
-//! same request.
+//! same request. Also the stream the controller queues requested reviews in,
+//! which the auth-callout responder grants to the controller alone.
 //!
 //! Three crates name these: the controller sends the request, the classifier
 //! answers it, and the auth-callout responder grants the one subject to each.
@@ -20,6 +21,19 @@ pub const SUBJECT: &str = "$SWARM.review-classifier.try-out";
 /// the requests instead of running each one twice.
 pub const QUEUE_GROUP: &str = "review-classifier";
 
+/// The work-queue stream holding the review requests the controller has
+/// accepted and not yet finished. The controller is its one publisher and its
+/// one consumer.
+pub const REQUEST_STREAM: &str = "review-requests";
+
+/// The subjects of [`REQUEST_STREAM`] start with this; each PR's request is
+/// one subject below it.
+pub const REQUEST_SUBJECT_PREFIX: &str = "$SWARM.review-request";
+
+/// The durable consumer on [`REQUEST_STREAM`] the controller takes requests
+/// from.
+pub const REQUEST_CONSUMER: &str = "swarm-controller";
+
 /// The most classifications one run carries.
 pub const MAX_CLASSIFICATIONS: usize = 10;