From 916fdb35b8c0ec2eaf878cf07905303392d8ea10 Mon Sep 17 00:00:00 2001 From: atlas Date: Fri, 9 Oct 2026 23:11:58 +0200 Subject: [PATCH] swarm: per-agent inference settings (model, effort, subagent_models) in KV swarm-controller keeps each agent's inference settings in a new `agent-inference` KV bucket, one key per agent, and is its only writer (`GET`/`PUT`/`DELETE /api/agents/{name}/inference`; `swarmctl agent inference get|set|clear`). Every field is optional. Each agent reads its own key every 10 s through a direct-get grant on that key alone, and a set `model` or `effort` replaces the Bus value, so the next turn runs on it and the badge follows. The swarm value is applied without persisting it: `Bus::set_model` and `set_effort` are split into a persisting setter and a non-persisting `apply_*`, and the poller uses the latter. Otherwise the swarm value would overwrite `hyperhive-model`/`hyperhive-effort`, and clearing it, or a restart, would no longer find the agent's own value. An unset field changes nothing; a field cleared after being applied goes back to the local value (`harness_state::local_model`/`local_effort`, the same resolution `Bus::new` uses), read again at the clear. `subagent_models` is validated, stored and shown by `get`; no agent applies it yet. Refs #5034 --- docs/README.md | 2 + docs/swarm/README.md | 2 + docs/swarm/inference.md | 68 +++++ docs/swarm/schedules.md | 4 +- docs/tools/swarmctl-cli.md | 81 ++++++ docs/turn-loop/claude-invocation.md | 4 +- hive-agent/Cargo.toml | 12 +- hive-agent/src/events.rs | 48 ++-- hive-agent/src/harness_state.rs | 21 ++ hive-agent/src/main.rs | 3 + hive-agent/src/swarm_inference.rs | 316 ++++++++++++++++++++++ nix/host-modules/swarm-nats.nix | 3 + nix/module-eval/nats-authelia.nix | 3 +- swarm-controller/README.md | 4 + swarm-controller/src/inference.rs | 295 ++++++++++++++++++++ swarm-controller/src/main.rs | 59 ++++ swarm-controller/src/schedules.rs | 10 +- swarm-nats-auth/src/main.rs | 4 +- swarm-nats-auth/src/policy.rs | 33 +++ swarm-queue-client/Cargo.toml | 3 + swarm-queue-client/src/agent_inference.rs | 235 ++++++++++++++++ swarm-queue-client/src/lib.rs | 5 + swarmctl/src/inference.rs | 182 +++++++++++++ swarmctl/src/main.rs | 69 +++++ 24 files changed, 1432 insertions(+), 34 deletions(-) create mode 100644 docs/swarm/inference.md create mode 100644 hive-agent/src/swarm_inference.rs create mode 100644 swarm-controller/src/inference.rs create mode 100644 swarm-queue-client/src/agent_inference.rs create mode 100644 swarmctl/src/inference.rs diff --git a/docs/README.md b/docs/README.md index e39e8b79..a202d3b2 100644 --- a/docs/README.md +++ b/docs/README.md @@ -84,6 +84,8 @@ declarations. sealed?** → [`swarm/bao.md`](swarm/bao.md). - **How does an agent get a prompt at set times, and why doesn't it fire twice after downtime?** → [`swarm/schedules.md`](swarm/schedules.md). +- **How do I set an agent's model or effort from the swarm, and what happens + to its own pick?** → [`swarm/inference.md`](swarm/inference.md). ## Scheduler, CI, observability diff --git a/docs/swarm/README.md b/docs/swarm/README.md index 2ab9facd..bca5ebe8 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -23,6 +23,8 @@ grafana, the controller, authelia, …) → graph, agent creation. → [below](#swarm-controller) - **scheduled prompts** — named prompts for one or more agents at set times, kept by the swarm and fired in each agent's container. → [`schedules.md`](schedules.md) +- **inference settings** — the model and effort the swarm sets for an agent, + over its own. → [`inference.md`](inference.md) - **swarm UI** — the operator's day-to-day surface, on the swarm apex, `admins` only. → [`ui.md`](ui.md) - **shared services** — one forge, homeserver, SSO, queue and metrics/logs diff --git a/docs/swarm/inference.md b/docs/swarm/inference.md new file mode 100644 index 00000000..d46f2e37 --- /dev/null +++ b/docs/swarm/inference.md @@ -0,0 +1,68 @@ +# Per-agent inference settings + +The swarm can set the model and reasoning effort each agent's turns run on, +plus the models it may start subagents on. A field you set wins over the +agent's own value; a field you leave unset keeps the agent on its own value, +so an agent you set nothing for runs as it does today. + +## Setting them + +```sh +swarmctl agent inference set iris --model opus --effort high +swarmctl agent inference get iris +swarmctl agent inference clear iris +``` + +- `set` changes only the flags you pass; the other fields keep what's + stored. A hive must declare the agent. +- `--effort` is one of `low`, `medium`, `high`, `xhigh`, `max`. +- `--subagent-models` takes a comma-separated list. It's stored and shown, + but agents don't apply it yet. +- `get` shows each field, naming the unset ones. +- `clear` removes every field; the agent goes back to its own values. + +The full flag reference is [`swarmctl-cli.md`](../tools/swarmctl-cli.md). + +## When an agent picks a change up + +Each agent reads its settings every 10 seconds and switches on the next +read: its next turn runs on the new model or effort, and its web UI badge +shows it. A turn already running keeps what it started with. + +- **Its own value is the one it runs without the swarm**, from a `/model` + or `/effort` pick in its web UI or from its agent.nix. Setting a swarm + value doesn't overwrite it. +- **Clearing a field** (or the whole key) puts the agent back on its own + value on the next read. +- **A `/model` or `/effort` pick while the swarm sets that field** changes + the agent's own value, but the swarm value comes back on the next read. +- **After a restart**, the agent runs its own values until its first read + of its settings succeeds. + +
Implementation + +The settings live in the `agent-inference` KV bucket, one key per agent, +which swarm-controller alone writes: + +```json +{ "model": "opus", "effort": "high", "subagent_models": ["haiku", "sonnet"] } +``` + +Every field is optional, and `{}` reads the same as no key. + +| method | path | body | +| -------- | ------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------ | +| `GET` | `/api/agents//inference` | —; answers the stored value, `{}` when none | +| `PUT` | `/api/agents//inference` | the whole value, replacing the stored one; the controller refuses unknown fields, an unknown effort, or a model name with whitespace | +| `DELETE` | `/api/agents//inference` | —; answers 204 | + +An agent's own queue credential grants it +`$JS.API.DIRECT.GET.KV_agent-inference.$KV.agent-inference.`, the +direct read of its own key, and nothing else in this bucket — the same shape +as its schedule key ([`schedules.md`](schedules.md)). + +The harness (`hive-agent`'s `swarm_inference`) applies a set field to the +in-memory model or effort without writing `hyperhive-model` or +`hyperhive-effort`, which keep the agent's own value. + +
diff --git a/docs/swarm/schedules.md b/docs/swarm/schedules.md index 742d25f5..a60ed95b 100644 --- a/docs/swarm/schedules.md +++ b/docs/swarm/schedules.md @@ -132,8 +132,8 @@ Each prompt they answer with has `id`, `name`, `agents` (sorted), `body`, `next_fire_unix`: the next time on the prompt's grid, as of the moment the controller answers, `null` when paused or a one-shot whose time has passed. -An agent's own queue credential grants exactly one subject in the swarm's -JetStream: `$JS.API.DIRECT.GET.KV_agent-schedules.$KV.agent-schedules.`, +An agent's own queue credential grants exactly one subject in this bucket: +`$JS.API.DIRECT.GET.KV_agent-schedules.$KV.agent-schedules.`, the direct read of its own key. It can't read another agent's key, open or watch the bucket, or write to it. A KV watch would need a consumer, and a consumer create subject doesn't name a key, so granting one would let the diff --git a/docs/tools/swarmctl-cli.md b/docs/tools/swarmctl-cli.md index fe16f17a..45ac1a86 100644 --- a/docs/tools/swarmctl-cli.md +++ b/docs/tools/swarmctl-cli.md @@ -9,6 +9,10 @@ This document contains the help content for the `swarmctl` command-line program. * [`swarmctl agent create`↴](#swarmctl-agent-create) * [`swarmctl agent mint-identity`↴](#swarmctl-agent-mint-identity) * [`swarmctl agent mint-forge-token`↴](#swarmctl-agent-mint-forge-token) +* [`swarmctl agent inference`↴](#swarmctl-agent-inference) +* [`swarmctl agent inference get`↴](#swarmctl-agent-inference-get) +* [`swarmctl agent inference set`↴](#swarmctl-agent-inference-set) +* [`swarmctl agent inference clear`↴](#swarmctl-agent-inference-clear) * [`swarmctl schedule`↴](#swarmctl-schedule) * [`swarmctl schedule add`↴](#swarmctl-schedule-add) * [`swarmctl schedule list`↴](#swarmctl-schedule-list) @@ -59,6 +63,7 @@ Manage agents across the swarm * `create` — Queue creation of a new agent on a hive in this swarm * `mint-identity` — Queue a re-mint of an existing agent's identity at the swarm's secret store * `mint-forge-token` — Check one agent's forge token, and mint it if it's missing or stale +* `inference` — Show or change the model, effort and subagent models the swarm sets for an agent @@ -137,6 +142,82 @@ Queues and returns, the same way `agent create` does — watch the swarm UI's jo +## `swarmctl agent inference` + +Show or change the model, effort and subagent models the swarm sets for an agent. + +A field the swarm sets wins over the agent's own value (from its `/model` or `/effort` pick or its agent.nix) from its next turn on; the agent reads its settings every ten seconds. A field the swarm doesn't set, or clears, leaves the agent on its own value. + +**Usage:** `swarmctl agent inference ` + +###### **Subcommands:** + +* `get` — Show what the swarm sets for an agent, naming the fields it doesn't set +* `set` — Set fields for an agent. Only the flags given change; the others keep what's stored +* `clear` — Clear every field for an agent, so it goes back to its own values + + + +## `swarmctl agent inference get` + +Show what the swarm sets for an agent, naming the fields it doesn't set + +**Usage:** `swarmctl agent inference get [OPTIONS] ` + +###### **Arguments:** + +* `` — Agent name + +###### **Options:** + +* `--controller-socket ` — swarm-controller's unix socket. + + Supplied by the nix module that installs this binary, from the same `socketPath` option the daemon binds; falls back to `SWARM_CONTROLLER_SOCKET`. + + + +## `swarmctl agent inference set` + +Set fields for an agent. Only the flags given change; the others keep what's stored + +**Usage:** `swarmctl agent inference set [OPTIONS] ` + +###### **Arguments:** + +* `` — Agent name. A hive must declare it + +###### **Options:** + +* `--model ` — The model its turns run on, as claude's `--model` takes it +* `--effort ` — The reasoning effort its turns run at + + Possible values: `low`, `medium`, `high`, `xhigh`, `max` + +* `--subagent-models ` — The models it may start subagents on, comma-separated. Stored and shown; the agent doesn't apply it yet +* `--controller-socket ` — swarm-controller's unix socket. + + Supplied by the nix module that installs this binary, from the same `socketPath` option the daemon binds; falls back to `SWARM_CONTROLLER_SOCKET`. + + + +## `swarmctl agent inference clear` + +Clear every field for an agent, so it goes back to its own values + +**Usage:** `swarmctl agent inference clear [OPTIONS] ` + +###### **Arguments:** + +* `` — Agent name + +###### **Options:** + +* `--controller-socket ` — swarm-controller's unix socket. + + Supplied by the nix module that installs this binary, from the same `socketPath` option the daemon binds; falls back to `SWARM_CONTROLLER_SOCKET`. + + + ## `swarmctl schedule` Manage scheduled prompts. diff --git a/docs/turn-loop/claude-invocation.md b/docs/turn-loop/claude-invocation.md index 38d9b4f7..b6138e0a 100644 --- a/docs/turn-loop/claude-invocation.md +++ b/docs/turn-loop/claude-invocation.md @@ -54,7 +54,9 @@ back to `"haiku"` if unset). The operator can flip it at runtime with `/model ` in the web terminal — the next turn picks it up. The harness persists the choice to `/harness/hyperhive-model` so it survives restart; override path: `HYPERHIVE_MODEL_FILE` env var -for tests. +for tests. A model or effort the swarm sets for the agent replaces the +`Bus` value, and the harness doesn't persist it; clearing it goes back to +the persisted or nix value → [`swarm/inference.md`](../swarm/inference.md). `harness_state::context_window_tokens(model)` looks up context-window size per-model. Resolution order (first diff --git a/hive-agent/Cargo.toml b/hive-agent/Cargo.toml index 3f5063ed..3ef9f6da 100644 --- a/hive-agent/Cargo.toml +++ b/hive-agent/Cargo.toml @@ -37,10 +37,14 @@ rusqlite.workspace = true schemars.workspace = true serde.workspace = true serde_json.workspace = true -# `agent-schedules` alone: the schedule ticker reads its own key with one direct -# get. `kv`/`notices` name buckets and streams this harness opens neither end -# of, and the terminal publisher is a plain core-subject publish. -swarm-queue-client = { workspace = true, features = ["agent-schedules"] } +# `agent-schedules` and `agent-inference`: the schedule ticker and the +# inference poller each read their own key with one direct get. `kv`/`notices` +# name buckets and streams this harness opens neither end of, and the terminal +# publisher is a plain core-subject publish. +swarm-queue-client = { workspace = true, features = [ + "agent-schedules", + "agent-inference", +] } # This agent's own queue secret, read from the store under its own certificate. swarm-secret-client.workspace = true tokio.workspace = true diff --git a/hive-agent/src/events.rs b/hive-agent/src/events.rs index e14aeb08..135b48a7 100644 --- a/hive-agent/src/events.rs +++ b/hive-agent/src/events.rs @@ -19,9 +19,8 @@ use serde::{Deserialize, Serialize}; use tokio::sync::broadcast; use crate::harness_state::{ - DEFAULT_EFFORT, DEFAULT_MODEL, configured_effort, configured_model, context_window_tokens, - load_effort, load_model, persist_effort, persist_model, read_harness_state, - write_harness_state, + configured_effort, configured_model, context_window_tokens, local_effort, local_model, + persist_effort, persist_model, read_harness_state, write_harness_state, }; const CHANNEL_CAPACITY: usize = 256; @@ -394,21 +393,8 @@ impl Bus { } }; let (tx, _) = broadcast::channel(CHANNEL_CAPACITY); - // Priority: HIVE_DEFAULT_MODEL (from services.hyperhive.agent.model in agent.nix) > - // persisted runtime override > compiled-in DEFAULT_MODEL. - // The nix config always wins on rebuild; the persisted file is kept - // for within-session tracking only (see persist_model / set_model). - let initial_model = configured_model().map_or_else( - || load_model().unwrap_or_else(|| DEFAULT_MODEL.to_owned()), - str::to_owned, - ); - // Effort precedence is the inverse of model: a persisted operator - // pick wins over the nix `HIVE_DEFAULT_EFFORT` baseline so the - // runtime choice survives harness restart (it only resets on - // `--purge`). nix is the default for a never-touched agent. - let initial_effort = load_effort() - .or_else(|| configured_effort().map(str::to_owned)) - .unwrap_or_else(|| DEFAULT_EFFORT.to_owned()); + let initial_model = local_model(configured_model()); + let initial_effort = local_effort(configured_effort()); // Restore rate_limited (and needs_login) from the consolidated // harness state file so the dashboard shows the correct status // on cold load if the harness crashed while parked. @@ -590,10 +576,21 @@ impl Bus { /// Panics if the internal lock is poisoned. pub fn set_model(&self, name: impl Into) { let value: String = name.into(); - self.model.lock().unwrap().clone_from(&value); if let Err(e) = persist_model(&value) { tracing::warn!(error = ?e, "model: persist failed"); } + self.apply_model(value); + } + + /// Switch the model for future turns without persisting it, so + /// `hyperhive-model` keeps the agent's local value. For a value the + /// swarm sets (`crate::swarm_inference`). + /// + /// # Panics + /// + /// Panics if the internal lock is poisoned. + pub fn apply_model(&self, value: String) { + self.model.lock().unwrap().clone_from(&value); // Mirror the resolved model into hyperhive-harness.json so // hive-c0re can surface it on the dashboard badge without reading // the hyperhive-model override file directly. @@ -624,10 +621,21 @@ impl Bus { /// Panics if the internal lock is poisoned. pub fn set_effort(&self, level: impl Into) { let value: String = level.into(); - self.effort.lock().unwrap().clone_from(&value); if let Err(e) = persist_effort(&value) { tracing::warn!(error = ?e, "effort: persist failed"); } + self.apply_effort(value); + } + + /// Switch the effort level without persisting it, so `hyperhive-effort` + /// keeps the agent's local value. For a value the swarm sets + /// (`crate::swarm_inference`). + /// + /// # Panics + /// + /// Panics if the internal lock is poisoned. + pub fn apply_effort(&self, value: String) { + self.effort.lock().unwrap().clone_from(&value); self.emit(LiveEvent::EffortChanged { effort: value }); } diff --git a/hive-agent/src/harness_state.rs b/hive-agent/src/harness_state.rs index 92c87126..492680f7 100644 --- a/hive-agent/src/harness_state.rs +++ b/hive-agent/src/harness_state.rs @@ -53,6 +53,27 @@ pub(crate) fn load_effort() -> Option { } } +/// The model this agent runs when the swarm sets none. Priority: +/// `configured` (`HIVE_DEFAULT_MODEL`, from `services.hyperhive.agent.model` +/// in agent.nix) > the persisted runtime override > [`DEFAULT_MODEL`]. +pub(crate) fn local_model(configured: Option<&str>) -> String { + configured.map_or_else( + || load_model().unwrap_or_else(|| DEFAULT_MODEL.to_owned()), + str::to_owned, + ) +} + +/// The effort this agent runs at when the swarm sets none. The inverse of +/// [`local_model`]: a persisted operator pick wins over `configured` +/// (`HIVE_DEFAULT_EFFORT`), so the runtime choice survives a harness restart +/// (it only resets on `--purge`); nix is the default for a never-touched +/// agent. +pub(crate) fn local_effort(configured: Option<&str>) -> String { + load_effort() + .or_else(|| configured.map(str::to_owned)) + .unwrap_or_else(|| DEFAULT_EFFORT.to_owned()) +} + pub(crate) fn persist_effort(level: &str) -> std::io::Result<()> { let path = effort_file_path(); if let Some(parent) = path.parent() { diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index a52ff3f3..733d561c 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -32,6 +32,7 @@ mod state_entry_watch; mod stats; mod swarm_agent_icon; mod swarm_agent_state; +mod swarm_inference; mod swarm_input; mod swarm_queue; mod swarm_term; @@ -532,6 +533,8 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { swarm_agent_state::spawn(&bus); // Once per start: a new icon only arrives with a new harness process. swarm_agent_icon::spawn(); + // The model and effort the swarm sets for this agent, over its local ones. + tokio::spawn(swarm_inference::run(bus.clone())); // Set by the web UI's `/api/cancel` on a successful SIGINT, read-and- // cleared by `handle_turn` before building the next wake prompt — see // `hive_sh4re::inbox::INTERRUPTED_HINT`. Shared between the web server task and diff --git a/hive-agent/src/swarm_inference.rs b/hive-agent/src/swarm_inference.rs new file mode 100644 index 00000000..30e32c70 --- /dev/null +++ b/hive-agent/src/swarm_inference.rs @@ -0,0 +1,316 @@ +//! Applies the model and effort the swarm sets for this agent. +//! +//! swarm-controller keeps the agent's settings under its key in the +//! `agent-inference` KV bucket. This task reads that key every +//! [`POLL_INTERVAL`] with the one subject the agent's own queue credential is +//! granted there (`swarm_queue_client::agent_inference::read_own`). +//! +//! **A set field wins**: it replaces the Bus value, so the next turn runs on +//! it and the web UI's badge follows. It is not persisted: `hyperhive-model` +//! and `hyperhive-effort` keep the agent's local value. +//! +//! **An unset field changes nothing**, so an agent the swarm sets nothing for +//! runs on its local value ([`crate::harness_state::local_model`] and +//! [`crate::harness_state::local_effort`]). A field that was set and then +//! cleared goes back to that local value, read again at that moment. +//! +//! A failed read changes nothing. Until the first successful read after a +//! start, the agent runs on its local values. +//! +//! `subagent_models` is not read here. + +use std::time::Duration; + +use swarm_queue_client::agent_inference::{AgentInference, Effort, Read, read_own}; + +use crate::events::Bus; +use crate::harness_state::{configured_effort, configured_model, local_effort, local_model}; + +/// How often the key is read. Also the longest a change can take to reach +/// the agent's next turn. +const POLL_INTERVAL: Duration = Duration::from_secs(10); + +/// Where the settings are applied, and where the local values come from. +trait Target { + fn model(&self) -> String; + fn effort(&self) -> String; + fn apply_model(&self, value: String); + fn apply_effort(&self, value: String); + fn local_model(&self) -> String; + fn local_effort(&self) -> String; +} + +/// The harness's Bus, with the nix values read once at start. +struct Harness { + bus: Bus, + nix_model: Option<&'static str>, + nix_effort: Option<&'static str>, +} + +impl Target for Harness { + fn model(&self) -> String { + self.bus.model() + } + fn effort(&self) -> String { + self.bus.effort() + } + fn apply_model(&self, value: String) { + self.bus.apply_model(value); + } + fn apply_effort(&self, value: String) { + self.bus.apply_effort(value); + } + fn local_model(&self) -> String { + local_model(self.nix_model) + } + fn local_effort(&self) -> String { + local_effort(self.nix_effort) + } +} + +/// Which fields currently carry the swarm's value. +#[derive(Debug, Default)] +struct Overridden { + model: bool, + effort: bool, +} + +/// What one field should become: the swarm's value while it sets one, the +/// local value on the read after it stops, and otherwise `None` (left alone). +/// Also `None` when the value is already what it should be. +fn next( + swarm: Option<&str>, + overridden: &mut bool, + current: &str, + local: impl FnOnce() -> String, +) -> Option { + let wanted = match swarm { + Some(value) => { + *overridden = true; + value.to_owned() + } + None if std::mem::take(overridden) => local(), + None => return None, + }; + (wanted != current).then_some(wanted) +} + +impl Overridden { + /// Apply one read of the key to `target`. + fn apply(&mut self, read: &AgentInference, target: &impl Target) { + if let Some(model) = next( + read.model.as_deref(), + &mut self.model, + &target.model(), + || target.local_model(), + ) { + tracing::info!(%model, swarm = self.model, "inference: switching model"); + target.apply_model(model); + } + if let Some(effort) = next( + read.effort.map(Effort::as_str), + &mut self.effort, + &target.effort(), + || target.local_effort(), + ) { + tracing::info!(%effort, swarm = self.effort, "inference: switching effort"); + target.apply_effort(effort); + } + } +} + +/// Run the poller for the harness's lifetime, if this agent has a label to +/// name its key with. +pub async fn run(bus: Bus) { + let agent = crate::identity::label(); + if agent.is_empty() { + tracing::warn!("this agent has no label; not reading its swarm inference settings"); + return; + } + let target = Harness { + bus, + nix_model: configured_model(), + nix_effort: configured_effort(), + }; + let client = crate::swarm_queue::client().await; + tracing::info!(%agent, "reading swarm inference settings from the swarm queue"); + let mut overridden = Overridden::default(); + loop { + if let Some(read) = read(&client, &agent).await { + overridden.apply(&read, &target); + } + tokio::time::sleep(POLL_INTERVAL).await; + } +} + +/// The key's current value, or `None` when it could not be read. +async fn read(client: &async_nats::Client, agent: &str) -> Option { + // An unconnected client does not fail a JetStream request, it hangs on it. + swarm_queue_client::ensure_connected(client).ok()?; + match read_own(client, agent).await { + Ok(Read::Current(current)) => Some(current), + Ok(Read::NoBucket) => { + tracing::debug!("inference: the agent-inference bucket does not answer yet"); + None + } + Err(e) => { + tracing::warn!(error = %e, "inference: reading this agent's settings failed"); + None + } + } +} + +#[cfg(test)] +mod tests { + use std::cell::RefCell; + + use super::*; + + /// A Bus stand-in recording every apply. + struct Fake { + model: RefCell, + effort: RefCell, + local_model: RefCell, + local_effort: String, + applies: RefCell, + } + + impl Fake { + fn new() -> Self { + Self { + model: RefCell::new("sonnet".to_owned()), + effort: RefCell::new("medium".to_owned()), + local_model: RefCell::new("sonnet".to_owned()), + local_effort: "medium".to_owned(), + applies: RefCell::new(0), + } + } + fn now(&self) -> (String, String) { + (self.model.borrow().clone(), self.effort.borrow().clone()) + } + } + + impl Target for Fake { + fn model(&self) -> String { + self.model.borrow().clone() + } + fn effort(&self) -> String { + self.effort.borrow().clone() + } + fn apply_model(&self, value: String) { + *self.applies.borrow_mut() += 1; + *self.model.borrow_mut() = value; + } + fn apply_effort(&self, value: String) { + *self.applies.borrow_mut() += 1; + *self.effort.borrow_mut() = value; + } + fn local_model(&self) -> String { + self.local_model.borrow().clone() + } + fn local_effort(&self) -> String { + self.local_effort.clone() + } + } + + fn set(model: Option<&str>, effort: Option) -> AgentInference { + AgentInference { + model: model.map(str::to_owned), + effort, + subagent_models: None, + } + } + + fn pair(model: &str, effort: &str) -> (String, String) { + (model.to_owned(), effort.to_owned()) + } + + #[test] + fn a_set_field_replaces_the_bus_value() { + let fake = Fake::new(); + let mut o = Overridden::default(); + o.apply(&set(Some("opus"), Some(Effort::High)), &fake); + assert_eq!(fake.now(), pair("opus", "high")); + // Re-reading the same value applies nothing again. + o.apply(&set(Some("opus"), Some(Effort::High)), &fake); + assert_eq!(*fake.applies.borrow(), 2); + } + + /// An absent key, and a key with every field unset, leave the agent on + /// whatever it runs now — even when that differs from its local value. + #[test] + fn an_absent_key_leaves_the_bus_alone() { + let fake = Fake::new(); + *fake.model.borrow_mut() = "haiku".to_owned(); + let mut o = Overridden::default(); + for _ in 0..3 { + o.apply(&AgentInference::default(), &fake); + } + assert_eq!(fake.now(), pair("haiku", "medium")); + assert_eq!(*fake.applies.borrow(), 0); + } + + /// Only the field the swarm sets moves; the other keeps its value. + #[test] + fn an_unset_field_beside_a_set_one_is_left_alone() { + let fake = Fake::new(); + *fake.effort.borrow_mut() = "max".to_owned(); + let mut o = Overridden::default(); + o.apply(&set(Some("opus"), None), &fake); + assert_eq!(fake.now(), pair("opus", "max")); + } + + #[test] + fn a_cleared_field_goes_back_to_the_local_value() { + let fake = Fake::new(); + let mut o = Overridden::default(); + o.apply(&set(Some("opus"), Some(Effort::Xhigh)), &fake); + assert_eq!(fake.now(), pair("opus", "xhigh")); + // The effort is cleared, the model still set. + o.apply(&set(Some("opus"), None), &fake); + assert_eq!(fake.now(), pair("opus", "medium")); + // The local value is read at the clear, not at start. + "haiku".clone_into(&mut fake.local_model.borrow_mut()); + o.apply(&AgentInference::default(), &fake); + assert_eq!(fake.now(), pair("haiku", "medium")); + // Once reverted, a later unset read changes nothing. + *fake.model.borrow_mut() = "sonnet".to_owned(); + o.apply(&AgentInference::default(), &fake); + assert_eq!(fake.now(), pair("sonnet", "medium")); + } + + /// On the real Bus: the swarm's value reaches the Bus but not the + /// persisted files, so clearing it, or a restart, finds the local value. + #[test] + fn the_swarm_value_is_not_persisted_over_the_local_one() { + let dir = tempfile::tempdir().unwrap(); + let prev = std::env::var_os("HYPERHIVE_HARNESS_DIR"); + // SAFETY: no other test in this crate reads or sets this variable. + unsafe { std::env::set_var("HYPERHIVE_HARNESS_DIR", dir.path()) }; + let bus = Bus::new(); + bus.set_model("sonnet"); + bus.set_effort("low"); + let target = Harness { + bus: bus.clone(), + nix_model: None, + nix_effort: None, + }; + let mut o = Overridden::default(); + o.apply(&set(Some("opus"), Some(Effort::Max)), &target); + let swarm = (bus.model(), bus.effort()); + let files = ( + crate::harness_state::load_model(), + crate::harness_state::load_effort(), + ); + o.apply(&AgentInference::default(), &target); + let cleared = (bus.model(), bus.effort()); + match prev { + // SAFETY: as above. + Some(v) => unsafe { std::env::set_var("HYPERHIVE_HARNESS_DIR", v) }, + None => unsafe { std::env::remove_var("HYPERHIVE_HARNESS_DIR") }, + } + assert_eq!(swarm, pair("opus", "max")); + assert_eq!(files, (Some("sonnet".to_owned()), Some("low".to_owned()))); + assert_eq!(cleared, pair("sonnet", "low")); + } +} diff --git a/nix/host-modules/swarm-nats.nix b/nix/host-modules/swarm-nats.nix index 0c56e96b..9a0ff768 100644 --- a/nix/host-modules/swarm-nats.nix +++ b/nix/host-modules/swarm-nats.nix @@ -739,6 +739,9 @@ in # `CONSUMER.CREATE` (a watch's create subject carries no key, # so it would cover every agent's) and no `$KV` write. "--agent-token-publish-subject ${lib.escapeShellArg "\$\$JS.API.DIRECT.GET.KV_agent-schedules.\$\$KV.agent-schedules.{agent}"}" + # Its own key of the `agent-inference` KV bucket + # (`swarm_queue_client::agent_inference`), on the same terms. + "--agent-token-publish-subject ${lib.escapeShellArg "\$\$JS.API.DIRECT.GET.KV_agent-inference.\$\$KV.agent-inference.{agent}"}" "--store-cert-role ${lib.escapeShellArg authCertRole}" ]; # The store identity arrives by `LoadCredential` and is named diff --git a/nix/module-eval/nats-authelia.nix b/nix/module-eval/nats-authelia.nix index 1d009b75..04f9506f 100644 --- a/nix/module-eval/nats-authelia.nix +++ b/nix/module-eval/nats-authelia.nix @@ -295,7 +295,7 @@ let # The whole per-agent grant, compared as a list rather than by infix: a # wider subject added beside these (`$$JS.API.>`, another stream's name) # would pass every presence check above. - name = "a verified agent's grant is exactly its own subjects, and its one JetStream API subject reads its own schedule key"; + name = "a verified agent's grant is exactly its own subjects, and its JetStream API subjects read its own schedule and inference keys"; ok = let args = lib.splitString " " (responderOf natsOldPath).serviceConfig.ExecStart; @@ -314,6 +314,7 @@ let "'$$KV.agent-icons.{agent}'" "'$$SWARM.term.{agent}.sub.>'" "'$$JS.API.DIRECT.GET.KV_agent-schedules.$$KV.agent-schedules.{agent}'" + "'$$JS.API.DIRECT.GET.KV_agent-inference.$$KV.agent-inference.{agent}'" ]; } { diff --git a/swarm-controller/README.md b/swarm-controller/README.md index c1889227..f98852be 100644 --- a/swarm-controller/README.md +++ b/swarm-controller/README.md @@ -46,6 +46,10 @@ The swarm's control plane. The swarm UI and `swarmctl` are its clients. `agent-schedules`, both of which this daemon alone writes (`GET`/`POST /api/schedules`, `PATCH`/`DELETE /api/schedules/{id}`); each agent fires them itself → [`docs/swarm/schedules.md`](../docs/swarm/schedules.md). +- **Inference settings** — each agent's model, effort and subagent models, + one key per agent in the `agent-inference` KV bucket, which this daemon + alone writes (`GET`/`PUT`/`DELETE /api/agents/{name}/inference`); each + agent applies them itself → [`docs/swarm/inference.md`](../docs/swarm/inference.md). - **Relays** — each agent's terminal and turn-state header as SSE, agent icons, the cross-repo issue report, and the UI's quick links. diff --git a/swarm-controller/src/inference.rs b/swarm-controller/src/inference.rs new file mode 100644 index 00000000..6d92d97e --- /dev/null +++ b/swarm-controller/src/inference.rs @@ -0,0 +1,295 @@ +//! `/api/agents/{name}/inference` — the model, effort and subagent models the +//! swarm sets for one agent. +//! +//! Each agent's settings are one value under its own key of `agent-inference` +//! (`swarm_queue_client::agent_inference`), which this daemon alone writes. +//! The agent reads its key and applies each set field over its local value; +//! an unset field leaves the local value in force. + +use std::collections::BTreeSet; + +use axum::Json; +use axum::extract::{Path, State}; +use axum::http::StatusCode; +use problem_details::ProblemDetails; +use serde::{Deserialize, Serialize}; +use swarm_queue_client::agent_inference::{self, AgentInference, Effort}; +use utoipa::ToSchema; + +use crate::schedules::{live, queue, read, refused, write}; +use crate::{AppState, error_problem, wanted_error_status}; + +/// How many times a losing writer re-reads and re-applies, as in +/// `wanted::MAX_ATTEMPTS`. +const MAX_ATTEMPTS: usize = 5; + +/// One agent's inference settings. An absent field is not set at swarm level, +/// and the agent keeps its own value for it. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +#[serde(deny_unknown_fields)] +pub struct InferenceSettings { + /// The model the agent's turns run on, as claude's `--model` takes it. + #[serde(default, skip_serializing_if = "Option::is_none")] + model: Option, + /// One of `low`, `medium`, `high`, `xhigh`, `max`. + #[serde(default, skip_serializing_if = "Option::is_none")] + #[schema(value_type = Option)] + effort: Option, + /// The models the agent may start its subagents on. At least one. + #[serde(default, skip_serializing_if = "Option::is_none")] + subagent_models: Option>, +} + +impl From for InferenceSettings { + fn from(v: AgentInference) -> Self { + Self { + model: v.model, + effort: v.effort, + subagent_models: v.subagent_models, + } + } +} + +/// A model name as the agent passes it to claude: one non-empty word. +fn check_model(name: &str, what: &str) -> Result<(), String> { + if name.is_empty() { + return Err(format!("the {what} is empty")); + } + if name.chars().any(|c| c.is_whitespace() || c.is_control()) { + return Err(format!("the {what} {name:?} contains whitespace")); + } + Ok(()) +} + +impl InferenceSettings { + /// The value to store, or why it is refused. + fn into_stored(self) -> Result { + if let Some(model) = &self.model { + check_model(model, "model")?; + } + if let Some(models) = &self.subagent_models { + if models.is_empty() { + return Err( + "subagent_models is empty; leave it out to keep the agent's own list" + .to_owned(), + ); + } + let mut seen = BTreeSet::new(); + for model in models { + check_model(model, "subagent model")?; + if !seen.insert(model) { + return Err(format!("subagent model {model:?} is listed twice")); + } + } + } + Ok(AgentInference { + model: self.model, + effort: self.effort, + subagent_models: self.subagent_models, + }) + } +} + +fn agent_name(name: &str) -> Result { + hive_types::Ident::parse(name) + .map(hive_types::Ident::into_string) + .map_err(|reason| (StatusCode::BAD_REQUEST, reason.to_owned())) +} + +fn queue_problem(e: &anyhow::Error) -> ProblemDetails { + tracing::warn!(error = %format!("{e:#}"), "inference: queue failure"); + error_problem(wanted_error_status(e), &format!("{e:#}")) +} + +async fn open( + client: &async_nats::Client, +) -> Result { + agent_inference::open_or_create(client) + .await + .map_err(|e| queue_problem(&anyhow::Error::new(e))) +} + +#[utoipa::path( + get, + path = "/api/agents/{name}/inference", + params(("name" = String, Path, description = "agent name")), + responses( + (status = 200, description = "the agent's settings; `{}` when none is set", + body = InferenceSettings), + (status = 400, description = "not an identifier (problem+json)", body = String), + (status = 503, description = "no swarm queue is configured on this host, or it is \ + not connected (problem+json)", body = String), + (status = 500, description = "the settings could not be read (problem+json)", + body = String), + ), + tag = "agents" +)] +pub(crate) async fn get_agent_inference( + State(state): State, + Path(name): Path, +) -> Result, ProblemDetails> { + let agent = agent_name(&name).map_err(refused)?; + let client = queue(&state).map_err(refused)?; + let store = open(&client).await?; + let (current, _) = read::(&store, &agent) + .await + .map_err(|e| queue_problem(&e))?; + Ok(Json(current.into())) +} + +#[utoipa::path( + put, + path = "/api/agents/{name}/inference", + params(("name" = String, Path, description = "agent name")), + request_body = InferenceSettings, + responses( + (status = 200, description = "the settings as stored; they replace the previous \ + ones whole", body = InferenceSettings), + (status = 400, description = "not an identifier, not declared on any hive, or a \ + model name is empty, has whitespace, or is listed twice (problem+json)", + body = String), + (status = 422, description = "an unknown field, or an effort that is not one of \ + the five levels"), + (status = 503, description = "no swarm queue is configured on this host, or it is \ + not connected (problem+json)", body = String), + (status = 500, description = "the settings could not be written (problem+json)", + body = String), + ), + tag = "agents" +)] +pub(crate) async fn put_agent_inference( + State(state): State, + Path(name): Path, + Json(req): Json, +) -> Result, ProblemDetails> { + let agent = agent_name(&name).map_err(refused)?; + let value = req + .into_stored() + .map_err(|why| error_problem(StatusCode::BAD_REQUEST, &why))?; + let client = queue(&state).map_err(refused)?; + if !live(&state).await?.contains(&agent) { + return Err(error_problem( + StatusCode::BAD_REQUEST, + &format!("no hive's wanted state declares an agent {agent:?} that is not destroyed"), + )); + } + let store = open(&client).await?; + for _ in 0..MAX_ATTEMPTS { + let (current, revision) = read::(&store, &agent) + .await + .map_err(|e| queue_problem(&e))?; + if current == value + || write(&store, &agent, &value, revision) + .await + .map_err(|e| queue_problem(&e))? + { + tracing::info!(%agent, ?value, "set inference settings"); + return Ok(Json(value.into())); + } + } + Err(queue_problem(&anyhow::anyhow!( + "gave up writing the inference settings of {agent} after {MAX_ATTEMPTS} conflicting \ + writes" + ))) +} + +#[utoipa::path( + delete, + path = "/api/agents/{name}/inference", + params(("name" = String, Path, description = "agent name")), + responses( + (status = 204, description = "cleared; the agent goes back to its own values"), + (status = 400, description = "not an identifier (problem+json)", body = String), + (status = 503, description = "no swarm queue is configured on this host, or it is \ + not connected (problem+json)", body = String), + (status = 500, description = "the settings could not be cleared (problem+json)", + body = String), + ), + tag = "agents" +)] +pub(crate) async fn delete_agent_inference( + State(state): State, + Path(name): Path, +) -> Result { + let agent = agent_name(&name).map_err(refused)?; + let client = queue(&state).map_err(refused)?; + let store = open(&client).await?; + store + .delete(&agent) + .await + .map_err(|e| queue_problem(&anyhow::Error::new(e)))?; + tracing::info!(%agent, "cleared inference settings"); + Ok(StatusCode::NO_CONTENT) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn settings(json: &str) -> serde_json::Result { + serde_json::from_str(json) + } + + #[test] + fn a_full_put_is_stored_as_sent() { + let stored = settings(r#"{"model":"opus","effort":"high","subagent_models":["haiku"]}"#) + .unwrap() + .into_stored() + .unwrap(); + assert_eq!( + stored, + AgentInference { + model: Some("opus".to_owned()), + effort: Some(Effort::High), + subagent_models: Some(vec!["haiku".to_owned()]), + } + ); + assert_eq!( + serde_json::to_value(InferenceSettings::from(stored)).unwrap(), + serde_json::json!({"model": "opus", "effort": "high", "subagent_models": ["haiku"]}) + ); + } + + /// Each field is optional, and one left out stays out of the stored value. + #[test] + fn a_partial_put_leaves_the_other_fields_unset() { + let stored = settings(r#"{"effort":"max"}"#) + .unwrap() + .into_stored() + .unwrap(); + assert_eq!( + stored, + AgentInference { + effort: Some(Effort::Max), + ..AgentInference::default() + } + ); + } + + #[test] + fn an_unknown_field_or_effort_does_not_decode() { + for json in [ + r#"{"modle":"opus"}"#, + r#"{"available_models":["opus"]}"#, + r#"{"effort":"extreme"}"#, + r#"{"effort":"HIGH"}"#, + ] { + assert!(settings(json).is_err(), "{json}"); + } + // Control: the same shape with a known field and level decodes. + assert!(settings(r#"{"model":"opus","effort":"xhigh"}"#).is_ok()); + } + + #[test] + fn bad_model_names_are_refused() { + for json in [ + r#"{"model":""}"#, + r#"{"model":"opus 4"}"#, + r#"{"subagent_models":[]}"#, + r#"{"subagent_models":["haiku",""]}"#, + r#"{"subagent_models":["haiku","haiku"]}"#, + ] { + assert!(settings(json).unwrap().into_stored().is_err(), "{json}"); + } + } +} diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 32b79f2c..bee0b7ee 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -52,6 +52,7 @@ mod config_pr; mod forge; mod forge_account; mod github_account; +mod inference; mod issue_report; mod linked_accounts; mod matrix_account; @@ -3211,6 +3212,11 @@ fn build_app(state: AppState) -> axum::Router { .routes(routes!(get_jobq_rollup)) .routes(routes!(get_agent_config_pr)) .routes(routes!(get_agent_icon)) + .routes(routes!( + inference::get_agent_inference, + inference::put_agent_inference, + inference::delete_agent_inference + )) .routes(routes!(get_config_prs)) .routes(routes!(create_agent)) .routes(routes!(mint_agent_identity)) @@ -5479,6 +5485,59 @@ mod tests { ); } + /// Every inference verb refuses a name that is not an identifier with a + /// 400 and answers 503 without a queue; a PUT whose body is refused is a + /// 400 before the queue is asked. + #[tokio::test] + async fn the_inference_routes_check_the_name_and_body_before_the_queue() { + use super::inference::{delete_agent_inference, get_agent_inference, put_agent_inference}; + use axum::extract::{Path, State}; + use axum::http::StatusCode; + + fn status(r: Result) -> Option { + r.map(|_| ()).expect_err("must not succeed").status + } + let (state, _sched) = state_with_roster(); + let body = |json: &str| axum::Json(serde_json::from_str(json).expect("decodes")); + for name in ["iris", "Not An Ident"] { + let want = Some(if name == "iris" { + StatusCode::SERVICE_UNAVAILABLE + } else { + StatusCode::BAD_REQUEST + }); + let path = || Path(name.to_owned()); + assert_eq!( + status(get_agent_inference(State(state.clone()), path()).await), + want, + "GET {name}" + ); + assert_eq!( + status( + put_agent_inference(State(state.clone()), path(), body(r#"{"model":"opus"}"#)) + .await + ), + want, + "PUT {name}" + ); + assert_eq!( + status(delete_agent_inference(State(state.clone()), path()).await), + want, + "DELETE {name}" + ); + } + assert_eq!( + status( + put_agent_inference( + State(state), + Path("iris".to_owned()), + body(r#"{"model":"opus 4"}"#) + ) + .await + ), + Some(StatusCode::BAD_REQUEST) + ); + } + #[tokio::test] async fn listing_subagents_needs_a_queue_and_an_identifier() { let (state, _sched) = state_with_roster(); diff --git a/swarm-controller/src/schedules.rs b/swarm-controller/src/schedules.rs index 0a52ac90..ee080e5a 100644 --- a/swarm-controller/src/schedules.rs +++ b/swarm-controller/src/schedules.rs @@ -404,7 +404,7 @@ async fn open_or_create_definitions(client: &async_nats::Client) -> anyhow::Resu /// The value of `key` decoded, with the revision to write it back against. /// An absent or deleted key is the default value. -async fn read( +pub(crate) async fn read( store: &Store, key: &str, ) -> anyhow::Result<(T, Option)> { @@ -425,7 +425,7 @@ async fn read( /// Write `value` to `key` against `revision`. `Ok(false)` when another write /// landed first. -async fn write( +pub(crate) async fn write( store: &Store, key: &str, value: &T, @@ -549,7 +549,7 @@ pub(crate) fn spawn_reconcile( } /// The connected queue client, or the problem. -fn queue(state: &AppState) -> Result { +pub(crate) fn queue(state: &AppState) -> Result { let status = state.status.as_ref().ok_or_else(|| { ( StatusCode::SERVICE_UNAVAILABLE, @@ -566,12 +566,12 @@ fn queue(state: &AppState) -> Result { Ok(client) } -fn refused((status, detail): (StatusCode, String)) -> ProblemDetails { +pub(crate) fn refused((status, detail): (StatusCode, String)) -> ProblemDetails { error_problem(status, &detail) } /// Every agent some hive declares and has not destroyed. -async fn live(state: &AppState) -> Result, ProblemDetails> { +pub(crate) async fn live(state: &AppState) -> Result, ProblemDetails> { let Some(wanted) = state.wanted.as_ref() else { return Err(error_problem( StatusCode::SERVICE_UNAVAILABLE, diff --git a/swarm-nats-auth/src/main.rs b/swarm-nats-auth/src/main.rs index ebde0b60..c4f3d3ba 100644 --- a/swarm-nats-auth/src/main.rs +++ b/swarm-nats-auth/src/main.rs @@ -507,12 +507,14 @@ mod tests { "$SWARM.term.{agent}.sub.>", "--agent-token-publish-subject", "$JS.API.DIRECT.GET.KV_agent-schedules.$KV.agent-schedules.{agent}", + "--agent-token-publish-subject", + "$JS.API.DIRECT.GET.KV_agent-inference.$KV.agent-inference.{agent}", "--store-cert-role", "swarm-nats-auth", ]) .expect("the unit's own argument vector must parse"); assert_eq!(args.agent_client_suffix, "-agent"); - assert_eq!(args.agent_token_publish_subjects.len(), 5); + assert_eq!(args.agent_token_publish_subjects.len(), 6); } /// A request with no `Nats-Server-Xkey` is dropped before anything reads diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 8db0db2b..5577cf62 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -1440,6 +1440,39 @@ mod tests { assert!(covers("$KV.*.>", &key_subject("argus"))); } + /// The inference template as `swarm-nats.nix` spells it reads one key, the + /// agent's own, and nothing that writes, watches or opens the bucket. + #[test] + fn an_agent_may_read_only_its_own_inference_key() { + use swarm_queue_client::agent_inference::{STREAM, direct_get_subject, key_subject}; + + let p = policy() + .with_agent_token_subjects(vec![ + "$JS.API.DIRECT.GET.KV_agent-inference.$KV.agent-inference.{agent}".to_owned(), + ]) + .expect("a per-agent template is valid"); + let atlas = p.agent_token_permissions("atlas").expect("configured"); + assert_eq!(atlas.publish, vec![direct_get_subject("atlas")]); + let argus = p.agent_token_permissions("argus").expect("configured"); + assert_eq!(argus.publish, vec![direct_get_subject("argus")]); + + for refused in [ + direct_get_subject("argus"), + format!("$JS.API.DIRECT.GET.{STREAM}"), + format!("$JS.API.STREAM.INFO.{STREAM}"), + format!("$JS.API.CONSUMER.CREATE.{STREAM}"), + key_subject("atlas"), + swarm_queue_client::agent_schedules::direct_get_subject("atlas"), + ] { + assert!( + !atlas.publish.iter().any(|g| covers(g, &refused)), + "atlas's grant covers {refused}" + ); + } + // Control: the matcher matches the one subject granted. + assert!(covers(&atlas.publish[0], &direct_get_subject("atlas"))); + } + #[test] fn an_agent_token_subject_without_the_placeholder_is_refused() { let err = policy() diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index 1fac3713..4d81730c 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -37,6 +37,9 @@ subagent-term = ["async-nats/jetstream"] # schedule key without opening the bucket. Explicit for the same reason as # `notices`. 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"] [dependencies] # Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature diff --git a/swarm-queue-client/src/agent_inference.rs b/swarm-queue-client/src/agent_inference.rs new file mode 100644 index 00000000..3130745c --- /dev/null +++ b/swarm-queue-client/src/agent_inference.rs @@ -0,0 +1,235 @@ +//! The per-agent inference-settings KV bucket: one key per agent, holding the +//! model, effort and subagent model list the swarm sets for it. +//! +//! **swarm-controller is the only writer**, and creates the bucket on its +//! first write. Every field is optional: an absent field leaves the agent on +//! its own local value, so an agent with no key runs exactly as it would with +//! no swarm. +//! +//! **An agent reads with one `DIRECT.GET` on its own key, not a watch**, for +//! the reason [`crate::agent_schedules`] gives: the direct get names the key +//! in its subject ([`crate::agent_inference::direct_get_subject`]), which is +//! what lets the queue grant scope the read to one agent. +//! +//! Keyed by the agent alone, like [`crate::agent_schedules`]: an agent can +//! move between hives. + +/// The KV bucket inference settings are published into, keyed by agent name. +pub const BUCKET: &str = "agent-inference"; + +/// The stream backing [`BUCKET`]. `create_key_value` prefixes `KV_`. +pub const STREAM: &str = "KV_agent-inference"; + +/// The subject `agent`'s entry is stored under. +#[must_use] +pub fn key_subject(agent: &str) -> String { + format!("$KV.{BUCKET}.{agent}") +} + +/// The request subject that reads `agent`'s current entry, and so the one +/// subject the queue grants that agent in this bucket. +#[must_use] +pub fn direct_get_subject(agent: &str) -> String { + format!("$JS.API.DIRECT.GET.{STREAM}.{}", key_subject(agent)) +} + +/// One agent's inference settings: the value under `BUCKET`/``. +/// `None` is "not set at swarm level", never "use a default". +#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct AgentInference { + /// The model the agent's turns run on. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub model: Option, + /// The reasoning effort the agent's turns run at. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub effort: Option, + /// The models the agent may start its subagents on. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub subagent_models: Option>, +} + +/// A claude `--effort` level, ascending. +#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum Effort { + Low, + Medium, + High, + Xhigh, + Max, +} + +impl Effort { + /// The level as claude's `--effort` takes it, and as it is stored. + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::Low => "low", + Self::Medium => "medium", + Self::High => "high", + Self::Xhigh => "xhigh", + Self::Max => "max", + } + } +} + +/// Open the inference bucket for writing, creating it if nothing has yet. +/// +/// **Controller-side only.** `history: 1`: an agent wants its current value. +#[cfg(feature = "kv")] +pub async fn open_or_create( + client: &async_nats::Client, +) -> Result { + let js = async_nats::jetstream::new(client.clone()); + crate::kv::open_or_create( + &js, + BUCKET, + "agent-inference", + "Inference settings the swarm controller sets for each agent".to_owned(), + ) + .await +} + +/// What reading one agent's key found. +#[cfg(feature = "agent-inference")] +#[derive(Debug)] +pub enum Read { + /// The bucket answered; an absent or deleted key has every field unset. + Current(AgentInference), + /// Nothing answers for the bucket's stream: no setting has been written + /// yet, or `JetStream` is not serving it right now. Not an unset value, + /// since the second case says nothing about the agent's settings. + NoBucket, +} + +/// Read `agent`'s current settings with the one subject its grant covers. +/// +/// # Errors +/// +/// A failed request or an undecodable value. A request the queue refused for +/// permissions reaches the client as a timeout, not as a refusal. +#[cfg(feature = "agent-inference")] +pub async fn read_own( + client: &async_nats::Client, + agent: &str, +) -> Result> { + use async_nats::jetstream::stream::DirectGetErrorKind; + + let js = async_nats::jetstream::new(client.clone()); + let stream = js.get_stream_no_info(STREAM).await?; + let message = match stream.direct_get_last_for_subject(key_subject(agent)).await { + Ok(message) => message, + Err(e) if matches!(e.kind(), DirectGetErrorKind::NotFound) => { + return Ok(Read::Current(AgentInference::default())); + } + Err(e) if is_no_responders(&e) => return Ok(Read::NoBucket), + Err(e) => return Err(e.into()), + }; + // A KV delete or purge is stored as an empty marker carrying this header. + let removed = message + .headers + .get("KV-Operation") + .is_some_and(|op| matches!(op.as_str(), "DEL" | "PURGE")); + if removed { + return Ok(Read::Current(AgentInference::default())); + } + Ok(Read::Current(serde_json::from_slice(&message.payload)?)) +} + +#[cfg(feature = "agent-inference")] +fn is_no_responders(e: &async_nats::jetstream::stream::DirectGetError) -> bool { + matches!( + e.kind(), + async_nats::jetstream::stream::DirectGetErrorKind::ErrorResponse(status, _) + if status == async_nats::StatusCode::NO_RESPONDERS + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The names the grant in `swarm-nats.nix` spells with `{agent}`. + #[test] + fn names_match_the_queue_grant() { + assert_eq!(STREAM, format!("KV_{BUCKET}")); + assert_eq!(key_subject("iris"), "$KV.agent-inference.iris"); + assert_eq!( + direct_get_subject("iris"), + "$JS.API.DIRECT.GET.KV_agent-inference.$KV.agent-inference.iris" + ); + } + + #[test] + fn a_full_value_round_trips() { + let value = AgentInference { + model: Some("opus".to_owned()), + effort: Some(Effort::Xhigh), + subagent_models: Some(vec!["haiku".to_owned(), "sonnet".to_owned()]), + }; + let encoded = serde_json::to_string(&value).unwrap(); + assert_eq!( + encoded, + r#"{"model":"opus","effort":"xhigh","subagent_models":["haiku","sonnet"]}"# + ); + assert_eq!( + serde_json::from_str::(&encoded).unwrap(), + value + ); + } + + /// An unset field is left out on the wire, and `{}` is every field unset. + #[test] + fn unset_fields_are_absent() { + let value = AgentInference { + effort: Some(Effort::Low), + ..AgentInference::default() + }; + assert_eq!( + serde_json::to_string(&value).unwrap(), + r#"{"effort":"low"}"# + ); + assert_eq!( + serde_json::from_str::("{}").unwrap(), + AgentInference::default() + ); + } + + #[test] + fn every_effort_is_stored_as_its_cli_spelling() { + for effort in [ + Effort::Low, + Effort::Medium, + Effort::High, + Effort::Xhigh, + Effort::Max, + ] { + assert_eq!( + serde_json::to_string(&effort).unwrap(), + format!("\"{}\"", effort.as_str()) + ); + } + } + + #[test] + fn an_unknown_effort_is_refused() { + for doc in [ + r#"{"effort":"extreme"}"#, + r#"{"effort":"High"}"#, + r#"{"effort":""}"#, + ] { + assert!( + serde_json::from_str::(doc).is_err(), + "{doc}" + ); + } + } + + /// Fields a newer controller adds must not cost an older agent its value. + #[test] + fn decoding_tolerates_unknown_fields() { + let decoded: AgentInference = + serde_json::from_str(r#"{"model":"opus","temperature":1}"#).unwrap(); + assert_eq!(decoded.model.as_deref(), Some("opus")); + } +} diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index dc03f3a6..7605f67a 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -217,6 +217,11 @@ pub mod agent_icon; /// writing needs `kv` and reading needs `agent-schedules`. pub mod agent_schedules; +/// The bucket swarm-controller sets each agent's model, effort and subagent +/// models in — one key per agent, read by that agent alone. The names are +/// unconditional; writing needs `kv` and reading needs `agent-inference`. +pub mod agent_inference; + #[cfg(feature = "kv")] mod kv; diff --git a/swarmctl/src/inference.rs b/swarmctl/src/inference.rs new file mode 100644 index 00000000..7b814976 --- /dev/null +++ b/swarmctl/src/inference.rs @@ -0,0 +1,182 @@ +//! `swarmctl agent inference` — the model, effort and subagent models the +//! swarm sets for one agent, through swarm-controller's +//! `/api/agents/{name}/inference` routes. +//! +//! The agent picks a change up on its next read of its key, within ten +//! seconds, and its next turn runs on it. A field the swarm does not set +//! leaves the agent on its own value. +//! +//! The shape below mirrors the controller's `InferenceSettings`, for the +//! reason [`crate::agent`]'s module doc gives. + +use std::path::Path; + +use anyhow::{Context as _, Result, bail}; +use serde::{Deserialize, Serialize}; + +use crate::agent::{parse_ident, send}; + +/// One agent's settings; an absent field is not set at swarm level. +#[derive(Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +struct Settings { + #[serde(default, skip_serializing_if = "Option::is_none")] + model: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + effort: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + subagent_models: Option>, +} + +/// What `agent inference set` was asked to change. +pub(crate) struct SetFlags { + pub model: Option, + pub effort: Option, + pub subagent_models: Option>, +} + +impl SetFlags { + /// `current` with each given flag written over it. + fn merge(self, current: Settings) -> Result { + if self.model.is_none() && self.effort.is_none() && self.subagent_models.is_none() { + bail!("nothing to change: pass --model, --effort or --subagent-models"); + } + Ok(Settings { + model: self.model.or(current.model), + effort: self.effort.or(current.effort), + subagent_models: self.subagent_models.or(current.subagent_models), + }) + } +} + +fn runtime() -> Result { + tokio::runtime::Builder::new_current_thread() + .enable_io() + .build() + .context("starting a tokio runtime for the controller request") +} + +/// The agent's settings path. The name is checked first, so it cannot +/// reshape the URI. +fn member(agent: &str) -> Result { + let agent = parse_ident(agent, "agent name")?; + Ok(format!("/api/agents/{agent}/inference")) +} + +/// Run `swarmctl agent inference get`. +pub(crate) fn get(socket: &Path, agent: &str) -> Result<()> { + let uri = member(agent)?; + let current: Settings = runtime()?.block_on(send( + socket, + hyper::Method::GET, + &uri, + None::<&()>, + "inference-get", + ))?; + print!("{}", render(¤t)); + Ok(()) +} + +/// Run `swarmctl agent inference set`: read the agent's settings, write the +/// flags over them, and store the result. +pub(crate) fn set(socket: &Path, agent: &str, flags: SetFlags) -> Result<()> { + let uri = member(agent)?; + let rt = runtime()?; + let current: Settings = rt.block_on(send( + socket, + hyper::Method::GET, + &uri, + None::<&()>, + "inference-get", + ))?; + let wanted = flags.merge(current)?; + let stored: Settings = rt.block_on(send( + socket, + hyper::Method::PUT, + &uri, + Some(&wanted), + "inference-set", + ))?; + print!("{}", render(&stored)); + Ok(()) +} + +/// Run `swarmctl agent inference clear`. +pub(crate) fn clear(socket: &Path, agent: &str) -> Result<()> { + let uri = member(agent)?; + runtime()?.block_on(send::<(), ()>( + socket, + hyper::Method::DELETE, + &uri, + None, + "inference-clear", + ))?; + println!("cleared; {agent} goes back to its own model and effort"); + Ok(()) +} + +/// One line per field, naming the unset ones. +fn render(s: &Settings) -> String { + const UNSET: &str = "(not set; the agent's own)"; + format!( + "model: {}\neffort: {}\nsubagent_models: {}\n", + s.model.as_deref().unwrap_or(UNSET), + s.effort.as_deref().unwrap_or(UNSET), + s.subagent_models + .as_ref() + .map_or_else(|| UNSET.to_owned(), |m| m.join(", ")) + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn flags() -> SetFlags { + SetFlags { + model: None, + effort: None, + subagent_models: None, + } + } + + /// A flag replaces its field and every other field keeps what is stored. + #[test] + fn set_merges_into_the_stored_value() { + let current: Settings = + serde_json::from_str(r#"{"model":"sonnet","subagent_models":["haiku"]}"#).unwrap(); + let merged = SetFlags { + effort: Some("high".to_owned()), + ..flags() + } + .merge(current) + .unwrap(); + assert_eq!( + serde_json::to_string(&merged).unwrap(), + r#"{"model":"sonnet","effort":"high","subagent_models":["haiku"]}"# + ); + let merged = SetFlags { + model: Some("opus".to_owned()), + ..flags() + } + .merge(merged) + .unwrap(); + assert_eq!(merged.model.as_deref(), Some("opus")); + assert_eq!(merged.effort.as_deref(), Some("high")); + } + + #[test] + fn set_with_no_flag_is_refused() { + assert!(flags().merge(Settings::default()).is_err()); + } + + #[test] + fn an_empty_value_renders_every_field_as_unset() { + let row = render(&serde_json::from_str("{}").unwrap()); + assert_eq!(row.lines().count(), 3); + assert!( + row.lines() + .all(|l| l.ends_with("(not set; the agent's own)")), + "{row}" + ); + } +} diff --git a/swarmctl/src/main.rs b/swarmctl/src/main.rs index 1afa1339..bdcc8d90 100644 --- a/swarmctl/src/main.rs +++ b/swarmctl/src/main.rs @@ -29,6 +29,7 @@ mod agent; mod forge; +mod inference; mod schedule; mod users; @@ -210,6 +211,55 @@ enum AgentVerb { /// Queues and returns, the same way `agent create` does — watch the /// swarm UI's job view for the outcome. MintForgeToken(AgentMintForgeTokenArgs), + /// Show or change the model, effort and subagent models the swarm sets + /// for an agent. + /// + /// A field the swarm sets wins over the agent's own value (from its + /// `/model` or `/effort` pick or its agent.nix) from its next turn on; + /// the agent reads its settings every ten seconds. A field the swarm + /// doesn't set, or clears, leaves the agent on its own value. + Inference { + #[command(subcommand)] + command: InferenceVerb, + }, +} + +#[derive(Subcommand)] +enum InferenceVerb { + /// Show what the swarm sets for an agent, naming the fields it doesn't + /// set. + Get(InferenceAgentArgs), + /// Set fields for an agent. Only the flags given change; the others keep + /// what's stored. + Set(InferenceSetArgs), + /// Clear every field for an agent, so it goes back to its own values. + Clear(InferenceAgentArgs), +} + +#[derive(Args)] +struct InferenceAgentArgs { + /// Agent name. + agent: String, + #[command(flatten)] + socket: ControllerSocketArg, +} + +#[derive(Args)] +struct InferenceSetArgs { + /// Agent name. A hive must declare it. + agent: String, + /// The model its turns run on, as claude's `--model` takes it. + #[arg(long, value_name = "MODEL")] + model: Option, + /// The reasoning effort its turns run at. + #[arg(long, value_name = "LEVEL", value_parser = ["low", "medium", "high", "xhigh", "max"])] + effort: Option, + /// The models it may start subagents on, comma-separated. Stored and + /// shown; the agent doesn't apply it yet. + #[arg(long, value_name = "MODELS", value_delimiter = ',')] + subagent_models: Option>, + #[command(flatten)] + socket: ControllerSocketArg, } #[derive(Args)] @@ -496,6 +546,9 @@ fn main() -> Result<()> { let socket = path_from(args.controller_socket, "SWARM_CONTROLLER_SOCKET")?; agent::mint_forge_token(&socket, &args.name) } + Verb::Agent { + command: AgentVerb::Inference { command }, + } => inference(command), // Same socket-resolution reasoning as `Create` above. Verb::Forge { command: ForgeVerb::MakeAdmin(args), @@ -544,6 +597,22 @@ fn main() -> Result<()> { } } +fn inference(command: InferenceVerb) -> Result<()> { + match command { + InferenceVerb::Get(a) => inference::get(&a.socket.resolve()?, &a.agent), + InferenceVerb::Set(a) => inference::set( + &a.socket.resolve()?, + &a.agent, + inference::SetFlags { + model: a.model, + effort: a.effort, + subagent_models: a.subagent_models, + }, + ), + InferenceVerb::Clear(a) => inference::clear(&a.socket.resolve()?, &a.agent), + } +} + fn schedule(command: ScheduleVerb) -> Result<()> { match command { ScheduleVerb::Add(a) => schedule::add(