review classifier: keep stored runs this build can't decode

A run record this build couldn't decode was dropped from the list it
read, and the next recorded run wrote the list back without it, so a
rollback followed by a roll-forward lost the newer build's runs for
good. The stored list is now a list of entries, each a decoded run or
the raw JSON of one that didn't decode. The route still shows only the
decoded runs. An undecodable entry keeps its place, counts toward the
100, and is written back as it was read. A read that finds any logs one
WARN with their count and the first error.

The settings decode derived its known fields from what
`Settings::default()` serializes, so a later field skipped when unset
would have been stripped from a valid stored document and lost on the
next PUT. The known fields are now an explicit list. A test serializes
a fully populated `Settings`, checks that its keys match the list, and
checks that `decode` round-trips it.

docs/swarm/review-classifier.md

@@ -135,7 +135,9 @@ The model, the effort and how many requested reviews run at once are one
 document for the whole swarm. `swarm-controller` keeps it in its
 `review-classifier-settings` KV bucket under the key `all`, and no other
 service reads it. With no document stored, the classifier runs on `haiku` at
-claude's default effort, two reviews at a time.
+claude's default effort, two reviews at a time. The controller
+ignores a stored field it doesn't know, and a document it still can't decode
+counts as none, with a warning in its log, until the next PUT replaces it.
 
 `GET /api/review-classifier/settings` answers the settings in force.
 `PUT /api/review-classifier/settings` replaces them whole and answers what it
@@ -155,7 +157,8 @@ stored. Both take and give one shape:
 The controller reads the settings for every run and sends the model and
 effort in the run request, so a change applies from the next run with no
 restart. A try-out reads them when it's posted, a review when it gets its
-place. Lowering the limit lets the runs already going finish.
+place. Lowering the limit lets the runs already going finish. A run request
+with no model runs on `haiku`.
 
 The PUT needs the `Remote-User` header the gateway sets, as the try-out POST
 does. Every error is problem+json with the reason in `detail`:
@@ -387,7 +390,10 @@ The newest 100 runs are one JSON array under the key `recent` in the
 controller reads or writes. The controller creates the bucket on first use and
 adds each run with a compare-and-set write, so the list survives a controller
 restart or a move to another host. Recording the 101st run drops the oldest;
-there's no time-based retention. The controller logs a failed write, and it
+there's no time-based retention. The controller leaves a run it can't decode, such
+as one a newer build recorded, out of the list with a warning in its log. It
+keeps the run in the store as it was and counts it toward the 100. A document
+that isn't a list reads as no runs. The controller logs a failed write, and it
 changes nothing on the forge.
 
 `GET /api/review-classifier/runs` returns the list, newest first. Like the

swarm-controller/src/review_classifier/runs.rs

@@ -23,7 +23,7 @@ use serde::{Deserialize, Serialize};
 use swarm_queue_client::review_classifier::{Overall, Reply, Row};
 use utoipa::ToSchema;
 
-use crate::schedules::{read, write};
+use crate::schedules::{read_with, write};
 use crate::{AppState, error_problem};
 
 /// The bucket holding the runs. Agents hold no grant on it.
@@ -119,28 +119,76 @@ pub(super) fn answer(reply: &Reply) -> (Option<Vec<Row>>, Option<Overall>) {
     }
 }
 
+/// One entry of the stored list.
+#[derive(Debug, Clone, PartialEq, Serialize)]
+#[serde(untagged)]
+pub(crate) enum Stored {
+    Run(RunRecord),
+    /// A record this build cannot decode, such as one a newer build wrote.
+    /// Left out of the list but kept, counted toward [`MAX_RUNS`] and written
+    /// back as it was read, so a newer build reads it again after a
+    /// roll-forward.
+    Unreadable(serde_json::Value),
+}
+
 /// Where the runs are kept: [`BUCKET`] in production, a fake in the tests.
 pub(crate) trait RunDoc {
-    /// The runs, with the revision to write them back against.
-    fn read(&self) -> impl Future<Output = anyhow::Result<(Vec<RunRecord>, Option<u64>)>> + Send;
+    /// The entries, with the revision to write them back against.
+    fn read(&self) -> impl Future<Output = anyhow::Result<(Vec<Stored>, Option<u64>)>> + Send;
     /// `Ok(false)` when another write landed since `revision`.
     fn write(
         &self,
-        runs: Vec<RunRecord>,
+        runs: Vec<Stored>,
         revision: Option<u64>,
     ) -> impl Future<Output = anyhow::Result<bool>> + Send;
 }
 
 impl RunDoc for Store {
-    async fn read(&self) -> anyhow::Result<(Vec<RunRecord>, Option<u64>)> {
-        read(self, KEY).await
+    async fn read(&self) -> anyhow::Result<(Vec<Stored>, Option<u64>)> {
+        read_with(self, KEY, |bytes| Ok(decode(bytes))).await
     }
 
-    async fn write(&self, runs: Vec<RunRecord>, revision: Option<u64>) -> anyhow::Result<bool> {
+    async fn write(&self, runs: Vec<Stored>, revision: Option<u64>) -> anyhow::Result<bool> {
         write(self, KEY, &runs, revision).await
     }
 }
 
+/// The entries in `bytes`, the stored document, with a warning when any is
+/// [`Stored::Unreadable`]. A document that is not a JSON array reads as no
+/// entries, with a warning, and the next write replaces it.
+fn decode(bytes: &[u8]) -> Vec<Stored> {
+    let values: Vec<serde_json::Value> = match serde_json::from_slice(bytes) {
+        Ok(values) => values,
+        Err(e) => {
+            tracing::warn!(
+                bucket = BUCKET, key = KEY, error = %e,
+                "the review classifier runs are not decodable; reading them as none"
+            );
+            return Vec::new();
+        }
+    };
+    let mut unreadable = 0_usize;
+    let mut first_error = None;
+    let entries = values
+        .into_iter()
+        .map(|value| match RunRecord::deserialize(&value) {
+            Ok(run) => Stored::Run(run),
+            Err(e) => {
+                unreadable += 1;
+                first_error.get_or_insert(e);
+                Stored::Unreadable(value)
+            }
+        })
+        .collect();
+    if let Some(error) = first_error {
+        tracing::warn!(
+            bucket = BUCKET, key = KEY, unreadable, %error,
+            "review classifier runs this build cannot decode are left out of the list and kept"
+        );
+    }
+    entries
+}
+
 /// Put `run` first in `doc`, keeping the newest [`MAX_RUNS`], re-reading and
 /// re-applying when another write landed in between.
 async fn push(doc: &impl RunDoc, run: &RunRecord) -> anyhow::Result<()> {
@@ -154,9 +202,9 @@ async fn push(doc: &impl RunDoc, run: &RunRecord) -> anyhow::Result<()> {
     anyhow::bail!("gave up recording the run after {MAX_ATTEMPTS} conflicting writes")
 }
 
-/// `runs` with `run` in front, cut to [`MAX_RUNS`].
-fn newest(run: &RunRecord, mut runs: Vec<RunRecord>) -> Vec<RunRecord> {
-    runs.insert(0, run.clone());
+/// `runs` with `run` in front, cut to [`MAX_RUNS`] entries.
+fn newest(run: &RunRecord, mut runs: Vec<Stored>) -> Vec<Stored> {
+    runs.insert(0, Stored::Run(run.clone()));
     runs.truncate(MAX_RUNS);
     runs
 }
@@ -211,7 +259,15 @@ impl Runs {
             #[cfg(test)]
             Self::Fake(doc) => doc.read().await,
         };
-        read.map(|(runs, _)| runs).map_err(|e| {
+        read.map(|(runs, _)| {
+            runs.into_iter()
+                .filter_map(|entry| match entry {
+                    Stored::Run(run) => Some(run),
+                    Stored::Unreadable(_) => None,
+                })
+                .collect()
+        })
+        .map_err(|e| {
             let detail = format!("{e:#}");
             tracing::warn!(error = %detail, "reading the review classifier runs failed");
             error_problem(StatusCode::INTERNAL_SERVER_ERROR, &detail)
@@ -310,7 +366,7 @@ pub(crate) mod tests {
     /// every read.
     #[derive(Default)]
     pub(crate) struct FakeDoc {
-        state: Mutex<(Vec<RunRecord>, u64)>,
+        state: Mutex<(Vec<Stored>, u64)>,
         conflicts: Mutex<usize>,
         fail: bool,
     }
@@ -330,26 +386,37 @@ pub(crate) mod tests {
             }
         }
 
+        /// The decodable runs, as the list shows them.
         pub(crate) fn runs(&self) -> Vec<RunRecord> {
+            self.entries()
+                .into_iter()
+                .filter_map(|entry| match entry {
+                    Stored::Run(run) => Some(run),
+                    Stored::Unreadable(_) => None,
+                })
+                .collect()
+        }
+
+        fn entries(&self) -> Vec<Stored> {
             self.state.lock().expect("unpoisoned").0.clone()
         }
     }
 
     impl RunDoc for FakeDoc {
-        async fn read(&self) -> anyhow::Result<(Vec<RunRecord>, Option<u64>)> {
+        async fn read(&self) -> anyhow::Result<(Vec<Stored>, Option<u64>)> {
             anyhow::ensure!(!self.fail, "the store is down");
             let (runs, revision) = self.state.lock().expect("unpoisoned").clone();
             Ok((runs, Some(revision).filter(|r| *r > 0)))
         }
 
-        async fn write(&self, runs: Vec<RunRecord>, revision: Option<u64>) -> anyhow::Result<bool> {
+        async fn write(&self, runs: Vec<Stored>, revision: Option<u64>) -> anyhow::Result<bool> {
             let mut state = self.state.lock().expect("unpoisoned");
             let mut conflicts = self.conflicts.lock().expect("unpoisoned");
             if *conflicts > 0 {
                 *conflicts -= 1;
                 let mut other = record(&format!("other-{conflicts}"));
                 other.trigger = Trigger::Review;
-                state.0.insert(0, other);
+                state.0.insert(0, Stored::Run(other));
                 state.1 += 1;
             }
             if revision.unwrap_or(0) != state.1 {
@@ -417,6 +484,72 @@ pub(crate) mod tests {
         assert_eq!(ids(&listed), ["b", "a"]);
     }
 
+    /// A record from a newer build among readable ones, and a document that
+    /// is no list at all: listing and recording both keep working. The newer
+    /// record is left out of the list and written back as it was; the
+    /// non-list document is replaced by one this build decodes.
+    #[tokio::test]
+    async fn an_undecodable_record_is_kept_and_a_non_list_replaced() {
+        let nats = crate::test_nats::TestNats::start();
+        let client = nats.client().await;
+        let Ok(store) = open(Some(&client)).await else {
+            panic!("the bucket did not open");
+        };
+        let runs = Runs::Live(Some(client));
+        let stored = || async {
+            let entry = store.get(KEY).await.expect("readable").expect("a document");
+            serde_json::from_slice::<Vec<serde_json::Value>>(&entry).expect("a JSON array")
+        };
+
+        let mut newer = serde_json::to_value(record("newer")).expect("encodes");
+        newer["outcome"] = "a-later-outcome".into();
+        let doc = serde_json::json!([record("b"), newer, record("a")]);
+        store
+            .put(KEY, doc.to_string().into())
+            .await
+            .expect("written");
+        assert_eq!(ids(&runs.list().await.expect("listed")), ["b", "a"]);
+        runs.record(record("c")).await;
+        assert_eq!(ids(&runs.list().await.expect("listed")), ["c", "b", "a"]);
+        let after = stored().await;
+        assert_eq!(after.len(), 4, "{after:?}");
+        assert_eq!(after[2], newer, "kept untouched, in its place");
+
+        store
+            .put(KEY, br#"{"not":"a list"}"#.to_vec().into())
+            .await
+            .expect("written");
+        assert!(runs.list().await.expect("listed").is_empty());
+        runs.record(record("d")).await;
+        let after = stored().await;
+        let decoded: Vec<RunRecord> = serde_json::from_value(after.into()).expect("decodes whole");
+        assert_eq!(ids(&decoded), ["d"]);
+    }
+
+    #[tokio::test]
+    async fn an_unreadable_entry_counts_toward_the_hundred() {
+        let doc = FakeDoc::default();
+        {
+            let mut state = doc.state.lock().expect("unpoisoned");
+            state.0 = (0..MAX_RUNS - 1)
+                .map(|i| Stored::Run(record(&format!("r{i}"))))
+                .collect();
+            state
+                .0
+                .insert(50, Stored::Unreadable(serde_json::json!({"from": "later"})));
+            state.1 = 1;
+        }
+        push(&doc, &record("new")).await.expect("stored");
+        let entries = doc.entries();
+        assert_eq!(entries.len(), MAX_RUNS);
+        assert!(entries.iter().any(|e| matches!(e, Stored::Unreadable(_))));
+        assert_eq!(
+            doc.runs().len(),
+            MAX_RUNS - 1,
+            "the unreadable entry took a run's place"
+        );
+    }
+
     #[tokio::test]
     async fn the_route_lists_the_runs_newest_first_and_a_down_store_is_an_error() {
         let doc = Arc::new(FakeDoc::default());

swarm-controller/src/review_classifier/settings.rs

@@ -22,7 +22,7 @@ use serde::{Deserialize, Serialize};
 use swarm_queue_client::agent_inference::Effort;
 use utoipa::ToSchema;
 
-use crate::schedules::{queue, read, refused, write};
+use crate::schedules::{queue, read_with, refused, write};
 use crate::{AppState, error_problem, wanted_error_status};
 
 /// The bucket holding the settings. No other service holds a grant on it.
@@ -59,7 +59,7 @@ pub(crate) struct Settings {
 impl Default for Settings {
     fn default() -> Self {
         Self {
-            model: "haiku".to_owned(),
+            model: swarm_queue_client::review_classifier::DEFAULT_MODEL.to_owned(),
             effort: None,
             max_concurrent_reviews: 2,
         }
@@ -90,13 +90,42 @@ pub(crate) trait SettingsSource {
 
 impl SettingsSource for async_nats::Client {
     async fn settings(&self) -> anyhow::Result<Settings> {
-        let (settings, _) = read::<Settings>(&open(self).await?, KEY)
+        let (settings, _) = stored(&open(self).await?)
             .await
             .context("reading the review classifier's settings")?;
         Ok(settings)
     }
 }
 
+/// The stored settings, with the revision to write them back against.
+async fn stored(store: &Store) -> anyhow::Result<(Settings, Option<u64>)> {
+    read_with(store, KEY, |bytes| Ok(decode(bytes))).await
+}
+
+/// Every field of [`Settings`]. `a_full_document_round_trips` fails when one
+/// is missing.
+const FIELDS: &[&str] = &["model", "effort", "max_concurrent_reviews"];
+
+/// `bytes`, the stored document, as settings. Unlike a `PUT` body, a field
+/// not in [`FIELDS`], such as one a newer build added, is ignored; a document
+/// that still does not decode is [`Settings::default`], with a warning,
+/// until the next `PUT` replaces it.
+fn decode(bytes: &[u8]) -> Settings {
+    let decoded = serde_json::from_slice(bytes).and_then(|mut value: serde_json::Value| {
+        if let Some(fields) = value.as_object_mut() {
+            fields.retain(|name, _| FIELDS.contains(&name.as_str()));
+        }
+        serde_json::from_value(value)
+    });
+    decoded.unwrap_or_else(|e| {
+        tracing::warn!(
+            bucket = BUCKET, key = KEY, error = %e,
+            "the review classifier settings are not decodable; using the defaults"
+        );
+        Settings::default()
+    })
+}
+
 /// Open [`BUCKET`], creating it if nothing has yet.
 async fn open(client: &async_nats::Client) -> anyhow::Result<Store> {
     let js = async_nats::jetstream::new(client.clone());
@@ -185,13 +214,10 @@ pub(crate) async fn put_settings(
     let client = queue(&state).map_err(refused)?;
     let store = open(&client).await.map_err(|e| queue_problem(&e))?;
     for _ in 0..MAX_ATTEMPTS {
-        let (current, revision) = read::<Settings>(&store, KEY)
+        let (_, revision) = stored(&store).await.map_err(|e| queue_problem(&e))?;
+        if write(&store, KEY, &value, revision)
             .await
-            .map_err(|e| queue_problem(&e))?;
-        if current == value
-            || write(&store, KEY, &value, revision)
-                .await
-                .map_err(|e| queue_problem(&e))?
+            .map_err(|e| queue_problem(&e))?
         {
             state.reviews.resize(value.max_concurrent_reviews);
             tracing::info!(%operator, ?value, "set the review classifier's settings");
@@ -225,6 +251,24 @@ mod tests {
         );
     }
 
+    /// Every field set, every `Option` `Some`: a stored document keeps all of
+    /// them, so a field missing from [`FIELDS`] fails here.
+    #[test]
+    fn a_full_document_round_trips() {
+        let full = Settings {
+            model: "sonnet".to_owned(),
+            effort: Some(Effort::High),
+            max_concurrent_reviews: 5,
+        };
+        let encoded = serde_json::to_value(&full).unwrap();
+        let mut keys: Vec<_> = encoded.as_object().unwrap().keys().collect();
+        let mut fields = FIELDS.to_vec();
+        keys.sort_unstable();
+        fields.sort_unstable();
+        assert_eq!(keys, fields);
+        assert_eq!(decode(&serde_json::to_vec(&full).unwrap()), full);
+    }
+
     #[test]
     fn bad_settings_are_refused() {
         for json in [
@@ -314,6 +358,56 @@ mod tests {
             .expect("a request")
     }
 
+    /// A stored field this build does not know is ignored, and a document
+    /// that still does not decode reads as the defaults; either way the GET
+    /// and a run's read keep working, and a PUT, even of the defaults,
+    /// replaces it with one that decodes.
+    #[tokio::test]
+    async fn an_undecodable_document_reads_as_the_defaults_until_a_put() {
+        let nats = TestNats::start();
+        let (app, _) = app(&nats).await;
+        let client = nats.client().await;
+        let store = open(&client).await.expect("the bucket");
+        let defaults = serde_json::to_value(Settings::default()).unwrap();
+
+        let newer = serde_json::json!({
+            "model": "sonnet", "effort": "high", "max_concurrent_reviews": 3, "later": true,
+        });
+        store
+            .put(KEY, newer.to_string().into())
+            .await
+            .expect("written");
+        let (status, read, _) = send(&app, get()).await;
+        assert_eq!(status, StatusCode::OK, "{read}");
+        assert_eq!(
+            read,
+            serde_json::json!({"model": "sonnet", "effort": "high", "max_concurrent_reviews": 3})
+        );
+
+        store
+            .put(
+                KEY,
+                br#"{"model":"sonnet","effort":"later"}"#.to_vec().into(),
+            )
+            .await
+            .expect("written");
+        let (status, read, _) = send(&app, get()).await;
+        assert_eq!(status, StatusCode::OK, "{read}");
+        assert_eq!(read, defaults);
+        assert_eq!(
+            client.settings().await.expect("readable"),
+            Settings::default()
+        );
+
+        let (status, stored, _) = send(&app, put(&defaults.to_string())).await;
+        assert_eq!(status, StatusCode::OK, "{stored}");
+        let entry = store.get(KEY).await.expect("readable").expect("a document");
+        assert_eq!(
+            serde_json::from_slice::<Settings>(&entry).expect("decodes strictly"),
+            Settings::default()
+        );
+    }
+
     #[tokio::test]
     async fn a_put_is_what_the_get_and_the_next_run_read() {
         let nats = TestNats::start();

swarm-controller/src/schedules.rs

@@ -407,6 +407,18 @@ async fn open_or_create_definitions(client: &async_nats::Client) -> anyhow::Resu
 pub(crate) async fn read<T: Default + serde::de::DeserializeOwned>(
     store: &Store,
     key: &str,
+) -> anyhow::Result<(T, Option<u64>)> {
+    read_with(store, key, |bytes| {
+        serde_json::from_slice(bytes).with_context(|| format!("{key} is not decodable"))
+    })
+    .await
+}
+
+/// As [`read`], with the stored bytes decoded by `decode`.
+pub(crate) async fn read_with<T: Default>(
+    store: &Store,
+    key: &str,
+    decode: impl FnOnce(&[u8]) -> anyhow::Result<T>,
 ) -> anyhow::Result<(T, Option<u64>)> {
     let Some(entry) = store
         .entry(key)
@@ -418,9 +430,7 @@ pub(crate) async fn read<T: Default + serde::de::DeserializeOwned>(
     if entry.operation != Operation::Put {
         return Ok((T::default(), Some(entry.revision)));
     }
-    let value =
-        serde_json::from_slice(&entry.value).with_context(|| format!("{key} is not decodable"))?;
-    Ok((value, Some(entry.revision)))
+    Ok((decode(&entry.value)?, Some(entry.revision)))
 }
 
 /// Write `value` to `key` against `revision`. `Ok(false)` when another write

swarm-queue-client/src/review_classifier.rs

@@ -33,6 +33,14 @@ pub const CLASSIFICATION_DEADLINE: Duration = Duration::from_mins(3);
 /// this long.
 pub const RUN_DEADLINE: Duration = Duration::from_mins(31);
 
+/// The model a run uses when its request names none, and the swarm-wide
+/// setting's default.
+pub const DEFAULT_MODEL: &str = "haiku";
+
+fn default_model() -> String {
+    DEFAULT_MODEL.to_owned()
+}
+
 /// One run: the PR as the controller read it, and the classifications to
 /// answer about it.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -51,6 +59,9 @@ pub struct TryOut {
     pub diff: String,
     pub classifications: Vec<Classification>,
     /// The model every classification runs on, as claude's `--model` takes it.
+    /// Absent is [`DEFAULT_MODEL`], so a controller that sends none still gets
+    /// its run answered.
+    #[serde(default = "default_model")]
     pub model: String,
     /// The effort every classification runs at; `None` passes no `--effort`.
     pub effort: Option<Effort>,
@@ -176,4 +187,27 @@ mod tests {
         let all = CLASSIFICATION_DEADLINE * u32::try_from(MAX_CLASSIFICATIONS).expect("small");
         assert!(RUN_DEADLINE > all);
     }
+
+    #[test]
+    fn a_request_without_a_model_runs_on_the_default() {
+        let mut request = serde_json::json!({
+            "operator": "alice",
+            "repo": "o/r",
+            "pr": 7,
+            "head_sha": "abc123",
+            "commits": [],
+            "diff": "",
+            "classifications": [],
+        });
+        let run: TryOut = serde_json::from_value(request.clone()).expect("decodes");
+        assert_eq!((run.model.as_str(), run.effort), (DEFAULT_MODEL, None));
+        // Control: a named model and effort are the ones used.
+        request["model"] = "sonnet".into();
+        request["effort"] = "high".into();
+        let run: TryOut = serde_json::from_value(request).expect("decodes");
+        assert_eq!(
+            (run.model.as_str(), run.effort),
+            ("sonnet", Some(Effort::High))
+        );
+    }
 }