vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

Compare commits

base: vinzenz:daaf920e06a0a3c9c326b4677c6bb0b2234a00aa
Branches Tags
vinzenz:main
..
compare: vinzenz:30d9e1d8e6810c6541648e8832838bd935d031e7
Branches Tags
vinzenz:main
5 changed files with 30 additions and 307 deletions
Download patch file Download diff file Expand all files Collapse all files

12
docs/swarm/review-classifier.md
View file

@ -135,9 +135,7 @@ 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. 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.
claude's default effort, two reviews at a time.
`GET /api/review-classifier/settings` answers the settings in force.
`PUT /api/review-classifier/settings` replaces them whole and answers what it
@ -157,8 +155,7 @@ 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. A run request
with no model runs on `haiku`.
place. Lowering the limit lets the runs already going finish.
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`:
@ -390,10 +387,7 @@ 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 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
there's no time-based retention. 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

163
swarm-controller/src/review_classifier/runs.rs
View file

@ -23,7 +23,7 @@ use serde::{Deserialize, Serialize};
use swarm_queue_client::review_classifier::{Overall, Reply, Row};
use utoipa::ToSchema;
use crate::schedules::{read_with, write};
use crate::schedules::{read, write};
use crate::{AppState, error_problem};
/// The bucket holding the runs. Agents hold no grant on it.
@ -119,76 +119,28 @@ 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 entries, with the revision to write them back against.
fn read(&self) -> impl Future<Output = anyhow::Result<(Vec<Stored>, Option<u64>)>> + Send;
/// The runs, with the revision to write them back against.
fn read(&self) -> impl Future<Output = anyhow::Result<(Vec<RunRecord>, Option<u64>)>> + Send;
/// `Ok(false)` when another write landed since `revision`.
fn write(
&self,
runs: Vec<Stored>,
runs: Vec<RunRecord>,
revision: Option<u64>,
) -> impl Future<Output = anyhow::Result<bool>> + Send;
}
impl RunDoc for Store {
async fn read(&self) -> anyhow::Result<(Vec<Stored>, Option<u64>)> {
read_with(self, KEY, |bytes| Ok(decode(bytes))).await
async fn read(&self) -> anyhow::Result<(Vec<RunRecord>, Option<u64>)> {
read(self, KEY).await
}
async fn write(&self, runs: Vec<Stored>, revision: Option<u64>) -> anyhow::Result<bool> {
async fn write(&self, runs: Vec<RunRecord>, 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<()> {
@ -202,9 +154,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`] entries.
fn newest(run: &RunRecord, mut runs: Vec<Stored>) -> Vec<Stored> {
runs.insert(0, Stored::Run(run.clone()));
/// `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.truncate(MAX_RUNS);
runs
}
@ -259,15 +211,7 @@ impl Runs {
#[cfg(test)]
Self::Fake(doc) => doc.read().await,
};
read.map(|(runs, _)| {
runs.into_iter()
.filter_map(|entry| match entry {
Stored::Run(run) => Some(run),
Stored::Unreadable(_) => None,
})
.collect()
})
.map_err(|e| {
read.map(|(runs, _)| runs).map_err(|e| {
let detail = format!("{e:#}");
tracing::warn!(error = %detail, "reading the review classifier runs failed");
error_problem(StatusCode::INTERNAL_SERVER_ERROR, &detail)
@ -366,7 +310,7 @@ pub(crate) mod tests {
/// every read.
#[derive(Default)]
pub(crate) struct FakeDoc {
state: Mutex<(Vec<Stored>, u64)>,
state: Mutex<(Vec<RunRecord>, u64)>,
conflicts: Mutex<usize>,
fail: bool,
}
@ -386,37 +330,26 @@ 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<Stored>, Option<u64>)> {
async fn read(&self) -> anyhow::Result<(Vec<RunRecord>, 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<Stored>, revision: Option<u64>) -> anyhow::Result<bool> {
async fn write(&self, runs: Vec<RunRecord>, 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, Stored::Run(other));
state.0.insert(0, other);
state.1 += 1;
}
if revision.unwrap_or(0) != state.1 {
@ -484,72 +417,6 @@ 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());

112
swarm-controller/src/review_classifier/settings.rs
View file

@ -22,7 +22,7 @@ use serde::{Deserialize, Serialize};
use swarm_queue_client::agent_inference::Effort;
use utoipa::ToSchema;
use crate::schedules::{queue, read_with, refused, write};
use crate::schedules::{queue, read, 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: swarm_queue_client::review_classifier::DEFAULT_MODEL.to_owned(),
model: "haiku".to_owned(),
effort: None,
max_concurrent_reviews: 2,
}
@ -90,42 +90,13 @@ pub(crate) trait SettingsSource {
impl SettingsSource for async_nats::Client {
async fn settings(&self) -> anyhow::Result<Settings> {
let (settings, _) = stored(&open(self).await?)
let (settings, _) = read::<Settings>(&open(self).await?, KEY)
.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());
@ -214,10 +185,13 @@ 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 (_, revision) = stored(&store).await.map_err(|e| queue_problem(&e))?;
if write(&store, KEY, &value, revision)
let (current, revision) = read::<Settings>(&store, KEY)
.await
.map_err(|e| queue_problem(&e))?
.map_err(|e| queue_problem(&e))?;
if current == value
|| write(&store, KEY, &value, revision)
.await
.map_err(|e| queue_problem(&e))?
{
state.reviews.resize(value.max_concurrent_reviews);
tracing::info!(%operator, ?value, "set the review classifier's settings");
@ -251,24 +225,6 @@ 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 [
@ -358,56 +314,6 @@ 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();

16
swarm-controller/src/schedules.rs
View file

@ -407,18 +407,6 @@ 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)
@ -430,7 +418,9 @@ pub(crate) async fn read_with<T: Default>(
if entry.operation != Operation::Put {
return Ok((T::default(), Some(entry.revision)));
}
Ok((decode(&entry.value)?, Some(entry.revision)))
let value =
serde_json::from_slice(&entry.value).with_context(|| format!("{key} is not decodable"))?;
Ok((value, Some(entry.revision)))
}
/// Write `value` to `key` against `revision`. `Ok(false)` when another write

34
swarm-queue-client/src/review_classifier.rs
View file

@ -33,14 +33,6 @@ 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)]
@ -59,9 +51,6 @@ 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>,
@ -187,27 +176,4 @@ 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))
);
}
}