review classifier: recover from a stored record it can't decode

The controller keeps the classifier's newest runs as one JSON array and
its settings as one JSON object, each under one KV key. A document this
build could not decode, as after a rollback past a build that added an
outcome or a settings field, failed every read and every write until
someone deleted the key by hand. A classifier newer than its controller
refused every run request for the missing `model`.

The run store now decodes the array one record at a time: a record that
does not decode is dropped with a WARN naming the bucket, key, index and
error, and a document that is not an array reads as no runs. Listing
keeps working, and the next recorded run writes back only the records
that decoded.

The stored settings ignore a field this build does not know; one that
still does not decode reads as the defaults with a WARN. The PUT body
keeps `deny_unknown_fields`. A PUT now always writes, so a PUT of the
defaults also replaces an undecodable document.

A run request without `model` runs on haiku, the swarm-wide setting's
default, now one constant in swarm-queue-client.

`schedules::read` gains a sibling, `read_with`, taking the decoder.

Closes #5162

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.
@@ -133,7 +133,7 @@ pub(crate) trait RunDoc {
 
 impl RunDoc for Store {
     async fn read(&self) -> anyhow::Result<(Vec<RunRecord>, Option<u64>)> {
-        read(self, KEY).await
+        read_with(self, KEY, |bytes| Ok(decode(bytes))).await
     }
 
     async fn write(&self, runs: Vec<RunRecord>, revision: Option<u64>) -> anyhow::Result<bool> {
@@ -141,6 +141,37 @@ impl RunDoc for Store {
     }
 }
 
+/// The records in `bytes`, the stored document. A record this build cannot
+/// decode, such as one a newer build wrote, is dropped with a warning, and a
+/// document that is not a JSON array reads as no runs; either way the next
+/// write replaces the document with one holding only decodable records.
+fn decode(bytes: &[u8]) -> Vec<RunRecord> {
+    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();
+        }
+    };
+    values
+        .into_iter()
+        .enumerate()
+        .filter_map(|(index, value)| {
+            serde_json::from_value(value)
+                .inspect_err(|e| {
+                    tracing::warn!(
+                        bucket = BUCKET, key = KEY, index, error = %e,
+                        "dropping a review classifier run that is not decodable"
+                    );
+                })
+                .ok()
+        })
+        .collect()
+}
+
 /// 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<()> {
@@ -417,6 +448,42 @@ 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, and the
+    /// next write leaves a document this build decodes whole.
+    #[tokio::test]
+    async fn an_undecodable_record_or_document_is_dropped_not_fatal() {
+        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<RunRecord>>(&entry).expect("decodes whole")
+        };
+
+        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(&stored().await), ["c", "b", "a"]);
+
+        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;
+        assert_eq!(ids(&stored().await), ["d"]);
+    }
+
     #[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,40 @@ 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
+}
+
+/// `bytes`, the stored document, as settings. Unlike a `PUT` body, a field
+/// this build does not know, 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. The known fields are those
+/// [`Settings::default`] encodes, which is every field.
+fn decode(bytes: &[u8]) -> Settings {
+    let decoded = serde_json::from_slice(bytes).and_then(|mut value: serde_json::Value| {
+        let known = serde_json::to_value(Settings::default())?;
+        if let (Some(fields), Some(known)) = (value.as_object_mut(), known.as_object()) {
+            fields.retain(|name, _| known.contains_key(name));
+        }
+        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 +212,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");
@@ -314,6 +338,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