From 9a71d1e094dfc7628c9b5a66ad1d51479afae6e3 Mon Sep 17 00:00:00 2001 From: atlas Date: Tue, 13 Oct 2026 19:24:13 +0200 Subject: [PATCH] swarm-ui: per-agent cap on concurrently running subagents Nothing capped how many subagents an agent runs at once; the operator wants to set that per agent from swarm-ui (#5234, shape (b)). AgentInference gains `max_subagents: Option`. An absent field decodes as None, so every stored value still parses and means "no cap", which is today's behaviour. swarm-controller accepts 1..=32 on PUT. 0 would refuse every start; 32 is far more concurrent claude sessions than one agent container has memory for, so anything above it is a typo, not a cap. The subagent daemon already read the agent's agent-inference key for subagent_models; it now reads it on every start and continue (not only those naming a model) and refuses a new run once the cap's worth of names is tracked, counted under the same lock as the name reservation. Every tracked slot counts: running, starting, a goal run between turns, and an interrupted one still exiting. A goal re-prompt never goes through the reservation, so it is never refused. A failed read applies no cap, as it falls back for models. swarmctl's `inference set` does GET, merge, PUT; its mirror type now carries the field so a set from the CLI keeps a cap set in swarm-ui instead of dropping it, and `get` prints it. --- docs/swarm/inference.md | 21 +- docs/swarm/ui.md | 9 +- docs/tools/subagent.md | 20 ++ .../src/agent/detail/AgentInference.tsx | 56 +++- hive-agent/src/swarm_inference.rs | 1 + hive-subagent-mcp/src/session.rs | 244 ++++++++++++++---- hive-subagent-mcp/src/swarm_models.rs | 42 +-- swarm-controller/src/inference.rs | 47 +++- swarm-queue-client/src/agent_inference.rs | 19 +- swarmctl/src/inference.rs | 28 +- 10 files changed, 393 insertions(+), 94 deletions(-) diff --git a/docs/swarm/inference.md b/docs/swarm/inference.md index 50115aeb..c4fd4505 100644 --- a/docs/swarm/inference.md +++ b/docs/swarm/inference.md @@ -1,7 +1,8 @@ -# Per-agent inference settings +or a model name with whitespace or a leading `-`, or a `max_subagents` outside 1 to 32 |# 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 +plus the models it may start subagents on and how many subagents it may run +at once. 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. @@ -20,6 +21,10 @@ swarmctl agent inference clear iris agent's `availableModels` as the models its subagent `start` and `continue` calls may name, read on each call ([`subagent.md`](../tools/subagent.md#which-models-a-subagent-may-run-on)). +- You set the subagent cap, `max_subagents`, in the swarm UI's inference + dialog; `set` has no flag for it and keeps the stored one. It's a whole + number from 1 to 32; unset means no cap + ([`subagent.md`](../tools/subagent.md#how-many-subagents-may-run-at-once)). - `get` shows each field, naming the unset ones (or says the swarm sets nothing for the agent), and then the model and effort the agent reports running on, each with where it comes from. @@ -39,6 +44,7 @@ command. model: opus effort: (not set; the agent's own) subagent_models: (not set; the agent's own) +max_subagents: (not set; no cap) running model: opus (swarm) running effort: medium (local_file) ``` @@ -83,7 +89,12 @@ 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"] } +{ + "model": "opus", + "effort": "high", + "subagent_models": ["haiku", "sonnet"], + "max_subagents": 3 +} ``` Every field is optional. The agent runs the same with `{}` as with no key; @@ -108,8 +119,8 @@ in-memory model or effort without writing `hyperhive-model` or `hyperhive-effort`, which keep the agent's own value. The subagent daemon (`hive-subagent-mcp`'s `swarm_models`) reads -`subagent_models` with the same direct get, under the same credential, on -each `start` or `continue` that names a model. +`subagent_models` and `max_subagents` with the same direct get, under the +same credential, on each `start` or `continue`. `effective` comes from the agent's turn-state header on `$SWARM.agent-state.` ([`README.md`](README.md)), which carries diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index 06cd4ad4..f20a9e2b 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -221,11 +221,12 @@ live rows only. The agent's detail panel shows its model and effort under **inference**, each with where it comes from: the swarm, its agent.nix, its own pick, or the -default. Below them is the subagent model list the swarm sets, or "its own" -when it sets none. **edit** opens a dialog with the three fields, filled with -what the swarm stores; saving does a `PUT /api/agents/{agent}/inference` of +default. Below them are the subagent model list the swarm sets, or "its own" +when it sets none, and the subagent cap, which reads "no cap" when unset. +**edit** opens a dialog with the four fields, filled with what the swarm +stores; saving does a `PUT /api/agents/{agent}/inference` of the whole value. A field you leave empty stays out of it, so the agent keeps -its own value for that field. **clear the swarm's setting** sends the `DELETE`. +its own value for that field; an empty max subagents is no cap. **clear the swarm's setting** sends the `DELETE`. The panel reads `GET /api/agents/{agent}/inference`; before the agent's first report reaches the controller, it shows the swarm's stored values only. [inference](inference.md) describes the settings themselves. diff --git a/docs/tools/subagent.md b/docs/tools/subagent.md index f75aeace..9c1ad469 100644 --- a/docs/tools/subagent.md +++ b/docs/tools/subagent.md @@ -337,6 +337,26 @@ happens until the daemon's first connection to the swarm queue succeeds. The match ignores case. On an ACP agent the names are the agent's own, such as `/` on `opencode`. +## How many subagents may run at once + +When the swarm sets `max_subagents` for the agent +([`inference.md`](../swarm/inference.md)), the daemon refuses a `start`, or a +`continue` of a session that isn't running, once that many subagents are +running, before anything spawns: + +```text +3 subagents running, cap is 3 (set in swarm-ui β†’ this agent β†’ inference) β€” wait for one to finish, or `interrupt` one +``` + +Every subagent the daemon tracks counts, whatever its role: one running a +turn, one starting, a goal run between turns, and an interrupted one whose +process hasn't exited yet. A goal run's next turn isn't a new start, and the +cap never stops it. Lowering the cap stops nothing already running. + +The daemon reads the cap with the model list, on every `start` and +`continue`, so a change applies to the next call. Without a swarm value, or +when the read fails, no cap applies. + ## Runtime A subagent runs on its parent agent's runtime diff --git a/frontend/packages/swarm-ui/src/agent/detail/AgentInference.tsx b/frontend/packages/swarm-ui/src/agent/detail/AgentInference.tsx index f23936d3..92c5fd74 100644 --- a/frontend/packages/swarm-ui/src/agent/detail/AgentInference.tsx +++ b/frontend/packages/swarm-ui/src/agent/detail/AgentInference.tsx @@ -29,6 +29,7 @@ interface Settings { model?: string; effort?: string; subagent_models?: string[]; + max_subagents?: number; } // Mirrors `EffectiveInference`. @@ -66,6 +67,18 @@ function modelNameProblem(name: string): string | null { return null; } +// The highest `max_subagents` the swarm accepts. +const MAX_SUBAGENTS = 32; + +// `max_subagents` as typed: empty is no cap, else a whole number in range. +function maxSubagentsValue(text: string): number | null | undefined { + const t = text.trim(); + if (t === "") return undefined; + if (!/^\d+$/.test(t)) return null; + const n = Number(t); + return n >= 1 && n <= MAX_SUBAGENTS ? n : null; +} + // Subagent models are typed as one list, split on commas and whitespace. function parseModels(text: string): string[] { return text.split(/[\s,]+/).filter((m) => m !== ""); @@ -159,11 +172,17 @@ export function AgentInference({ agent }: { agent: string }) { value={swarmModels.length > 0 ? swarmModels.join(", ") : "its own"} /> +
  • + +
    • setEditing(true)} /> (null); const subagentList = parseModels(subagents); - const modelProblem = + const formProblem = modelNameProblem(model.trim()) ?? subagentList.map(modelNameProblem).find((p) => p !== null) ?? - null; + (maxSubagentsValue(maxSubagents) === null + ? `max subagents is a whole number from 1 to ${MAX_SUBAGENTS}, or empty for no cap` + : null); const empty = - model.trim() === "" && effort === "" && subagentList.length === 0; + model.trim() === "" && + effort === "" && + subagentList.length === 0 && + maxSubagents.trim() === ""; async function send(init: RequestInit) { setBusy(true); @@ -238,6 +265,8 @@ function InferenceForm({ if (model.trim() !== "") body.model = model.trim(); if (effort !== "") body.effort = effort; if (subagentList.length > 0) body.subagent_models = subagentList; + const max = maxSubagentsValue(maxSubagents); + if (typeof max === "number") body.max_subagents = max; void send({ method: "PUT", headers: { "content-type": "application/json" }, @@ -249,8 +278,10 @@ function InferenceForm({

    Sets what the swarm makes {agent} run on. A field left empty stays the - agent's own value. The agent picks a change up within seconds; a turn - already running finishes on what it started with. + agent's own value; an empty max subagents is no cap. The agent picks a + change up within seconds; a turn already running finishes on what it + started with, and subagents already running past a lowered cap keep + running.

    - {modelProblem ? ( -

    {modelProblem}

    + + {formProblem ? ( +

    {formProblem}

    ) : null} {problem ? ( @@ -284,7 +322,7 @@ function InferenceForm({ diff --git a/hive-agent/src/swarm_inference.rs b/hive-agent/src/swarm_inference.rs index 57d96420..5d4dd6fa 100644 --- a/hive-agent/src/swarm_inference.rs +++ b/hive-agent/src/swarm_inference.rs @@ -227,6 +227,7 @@ mod tests { model: model.map(str::to_owned), effort, subagent_models: None, + max_subagents: None, } } diff --git a/hive-subagent-mcp/src/session.rs b/hive-subagent-mcp/src/session.rs index 43ca24b6..f6377784 100644 --- a/hive-subagent-mcp/src/session.rs +++ b/hive-subagent-mcp/src/session.rs @@ -99,6 +99,7 @@ use hive_runtime::{ AcpCommand, AcpRuntime, PermissionAsk, PermissionPolicy, Runtime as _, RuntimeSpec, TurnStop, }; use hive_sh4re::permissions::ToolGroup; +use swarm_queue_client::agent_inference::AgentInference; use tokio::sync::oneshot; use crate::swarm_term::Output; @@ -250,6 +251,15 @@ enum Occupancy { Stopping, } +/// Why [`State::reserve_under`] claimed nothing. +#[derive(Debug, PartialEq, Eq)] +enum Refusal { + /// The name is already tracked. + Taken, + /// `running` names are tracked, and the swarm caps this agent at `cap`. + AtCap { running: usize, cap: u32 }, +} + /// What settled a resumed turn's bounded wait β€” the one thing `continue` /// blocks on before it answers. enum ResumeVerdict { @@ -354,7 +364,7 @@ fn searched_location(config: &Config) -> Option { /// means it was interrupted and its process is still winding down. Each /// claims the name. `Starting` closes a TOCTOU window: checking "is `name` /// free" and committing to it are two different lock acquisitions unless -/// the check *is* the reservation β€” see `reserve`. +/// the check *is* the reservation β€” see `reserve_under`. /// /// `killed` is the other half of that map's story: `running` says what is in /// flight *now*, `killed` remembers the names whose last turn ended on a @@ -401,8 +411,9 @@ pub struct State { /// Where every subagent's output lines go up to the swarm, when this /// agent can publish there ([`State::publishing_to`]). term: Option, - /// Where the swarm's `subagent_models` for this agent is read from, when - /// this agent can read it ([`State::reading_models_from`]). + /// Where the swarm's `subagent_models` and `max_subagents` for this agent + /// are read from, when this agent can read them + /// ([`State::reading_models_from`]). models: Option, /// The agent's own model list, checked when the swarm sets none: /// [`available_models`] outside tests. @@ -483,8 +494,9 @@ impl State { self } - /// Check every named `model` against the swarm's `subagent_models` as - /// `models` reads it, where the swarm sets one. + /// Check every named `model` against the swarm's `subagent_models`, and + /// every new run against its `max_subagents`, as `models` reads them, + /// where the swarm sets them. #[must_use] pub fn reading_models_from(mut self, models: Option) -> Self { self.models = models; @@ -540,17 +552,19 @@ impl State { command } - /// Refuse a `model` outside the list in effect for this agent (see - /// `check_model`). The swarm key is read only for a named model. - async fn allow_model(&self, model: Option<&str>) -> anyhow::Result<()> { - if model.is_none() { - return Ok(()); - } + /// Read the swarm's key once, refuse a `model` outside the list in effect + /// for this agent (see `check_model`), and return the subagent cap in + /// effect (see `subagent_cap`). + async fn swarm_limits(&self, model: Option<&str>) -> anyhow::Result> { let read = match &self.models { Some(reader) => Some(reader.read().await), None => None, }; - check_model(model, &model_list(read, (self.local_models)())) + let cap = subagent_cap(read.as_ref()); + if model.is_some() { + check_model(model, &model_list(read, (self.local_models)()))?; + } + Ok(cap) } /// Run every subagent on `runtime` β€” the daemon passes the parent @@ -640,19 +654,39 @@ impl State { /// owning it" in the *same* lock acquisition, so two calls racing the /// same name can't both pass a check before either commits (the exact /// same-name concurrent-run hazard this module's doc warns about). - /// Returns `false` (reserving nothing) if `name` is already reserved - /// or running. - fn reserve(&self, name: &str) -> bool { + /// Refuses (reserving nothing) a `name` already tracked, or any name + /// once `cap` names are tracked. The count is checked under the same + /// lock, so two racing calls can't both take the last place. + /// + /// Every tracked name counts toward `cap`, a goal run between turns + /// and an interrupted one still winding down included: each holds a + /// process or is about to. A goal's next turn never comes through here, + /// so a run already counted is never refused its re-prompt. + fn reserve_under(&self, name: &str, cap: Option) -> Result<(), Refusal> { let mut running = self.running.lock().unwrap_or_else(PoisonError::into_inner); if running.contains_key(name) { - return false; + return Err(Refusal::Taken); + } + if let Some(cap) = cap + && running.len() >= usize::try_from(cap).unwrap_or(usize::MAX) + { + return Err(Refusal::AtCap { + running: running.len(), + cap, + }); } running.insert(name.to_owned(), Slot::Starting); - true + Ok(()) + } + + /// [`State::reserve_under`] with no cap: `false` if `name` is taken. + #[cfg(test)] + fn reserve(&self, name: &str) -> bool { + self.reserve_under(name, None).is_ok() } /// Release a reservation that never made it to a real spawn (an error - /// on the slow path between `reserve` and `Claude::spawn` succeeding). + /// on the slow path between `reserve_under` and `Claude::spawn` succeeding). /// A no-op if the entry was already upgraded to `Running` β€” this only /// ever clears a still-`Starting` placeholder, never a live process. /// @@ -782,7 +816,7 @@ impl State { /// could. /// /// ⚠️ **Callers must only call this once the call is known to proceed** - /// β€” after `reserve` in `start`, after the session lookup in a + /// β€” after `reserve_under` in `start`, after the session lookup in a /// `continue` (`begin_continue`) β€” so a refused call never changes what /// a later bare `continue` resolves to. `peek_dir` is the non-committing /// variant for a call that might still be refused, or that shouldn't @@ -1037,7 +1071,10 @@ enum ModelList { fn model_list(read: Option, local: Option) -> ModelList { use crate::swarm_models::Read; match read { - Some(Read::Set(models)) => ModelList::Swarm(models), + Some(Read::Current(AgentInference { + subagent_models: Some(models), + .. + })) => ModelList::Swarm(models), Some(Read::Failed(error)) => { tracing::warn!( %error, @@ -1046,7 +1083,25 @@ fn model_list(read: Option, local: Option) -> ); ModelList::Local(local) } - Some(Read::Unset) | None => ModelList::Local(local), + Some(Read::Current(_)) | None => ModelList::Local(local), + } +} + +/// The swarm's `max_subagents` for this agent, or `None`, no cap: when it +/// sets none, when `read` is `None` because this daemon has no swarm queue, +/// or when the read failed, which is logged. +fn subagent_cap(read: Option<&crate::swarm_models::Read>) -> Option { + use crate::swarm_models::Read; + match read? { + Read::Current(settings) => settings.max_subagents, + Read::Failed(error) => { + tracing::warn!( + %error, + "reading this agent's subagent settings from the swarm failed; applying no \ + subagent cap" + ); + None + } } } @@ -1347,7 +1402,8 @@ pub struct StartRequest { /// # Errors /// /// A name already running, an invalid name, a `model` the operator hasn't -/// made available (see `check_model`), an archive failure, or the +/// made available (see `check_model`), this agent at its subagent cap (see +/// `subagent_cap`), an archive failure, or the /// underlying `Claude::spawn` failing (binary missing, etc.) β€” the last /// case is the only one that can happen *after* commit-to-run, and it's /// exactly why nothing is registered in `running` until spawn actually @@ -1355,15 +1411,20 @@ pub struct StartRequest { pub async fn start(state: &Arc, req: StartRequest) -> anyhow::Result { let name = req.name.as_str(); validate_name(name)?; - state.allow_model(req.model.as_deref()).await?; - // Before `reserve`, so a role this agent does not have costs the caller + let cap = state.swarm_limits(req.model.as_deref()).await?; + // Before `reserve_under`, so a role this agent does not have costs the caller // an error and nothing else: no name claimed, no session archived, no // process spawned. let role = req.role.as_deref().map(crate::role::load).transpose()?; - if !state.reserve(name) { - return Err(name_taken(state, name, "use `continue` or `interrupt`")); + if let Err(refusal) = state.reserve_under(name, cap) { + return Err(refused( + state, + name, + &refusal, + "use `continue` or `interrupt`", + )); } - // Only commit the remembered `dir` now that `reserve` has actually + // Only commit the remembered `dir` now that `reserve_under` has actually // claimed `name` β€” see `resolve_dir`'s doc for why the order matters, // and the same reasoning governs the three records below it. let dir = state.resolve_dir(name, req.dir.as_deref()); @@ -1412,7 +1473,7 @@ struct SpawnSpec { role: Option, } -/// The slow, fallible part of `start`, run only after `reserve` has +/// The slow, fallible part of `start`, run only after `reserve_under` has /// already closed the TOCTOU window β€” split out so `start` can release the /// reservation on any error path here without duplicating that logic per /// failure site. @@ -1526,7 +1587,8 @@ fn start_on_runtime( /// # Errors /// /// An invalid name, one already running, a `model` the operator hasn't made -/// available (see `check_model`), no session titled `name` in the project +/// available (see `check_model`), this agent at its subagent cap (see +/// `subagent_cap`), no session titled `name` in the project /// dir searched, `Claude::spawn` failing, or the resumed turn failing within /// `RESUME_GRACE`. pub async fn continue_( @@ -1538,11 +1600,12 @@ pub async fn continue_( dir: Option<&str>, ) -> anyhow::Result { validate_name(name)?; - state.allow_model(model.as_deref()).await?; - if !state.reserve(name) { - return Err(name_taken( + let cap = state.swarm_limits(model.as_deref()).await?; + if let Err(refusal) = state.reserve_under(name, cap) { + return Err(refused( state, name, + &refusal, "use `interrupt` first if you meant to redirect it", )); } @@ -1617,7 +1680,7 @@ fn note_resumed_after_kill(msg: &str, killed: Option) -> String { } } -/// The slow, fallible part of `continue_`, run only after `reserve` has +/// The slow, fallible part of `continue_`, run only after `reserve_under` has /// already closed the TOCTOU window β€” same split rationale as /// `start_reserved`. `dir` is the caller's own, not yet resolved against the /// remembered one. @@ -2509,7 +2572,24 @@ pub fn interrupt(state: &State, name: &str) -> anyhow::Result { )) } -/// `start`'s and `continue`'s refusal for a name `reserve` found claimed: +/// `start`'s and `continue`'s refusal for a name `reserve_under` would not +/// claim. +fn refused(state: &State, name: &str, refusal: &Refusal, hint: &str) -> anyhow::Error { + match refusal { + Refusal::Taken => name_taken(state, name, hint), + Refusal::AtCap { running, cap } => anyhow::anyhow!( + "{running} {} running, cap is {cap} (set in swarm-ui β†’ this agent β†’ \ + inference) β€” wait for one to finish, or `interrupt` one", + if *running == 1 { + "subagent" + } else { + "subagents" + } + ), + } +} + +/// `start`'s and `continue`'s refusal for a name `reserve_under` found claimed: /// `hint` says what to do about a live session, and an interrupted one that /// hasn't exited yet gets its own answer. fn name_taken(state: &State, name: &str, hint: &str) -> anyhow::Error { @@ -3013,7 +3093,7 @@ mod tests { // The exact scenario argus caught in review: a rejected concurrent // `start` used to commit its `dir` before the reservation check // refused it, corrupting what a later bare `continue` would resolve - // to. Reordering `reserve` before `resolve_dir` in `start` closes it + // to. Reordering `reserve_under` before `resolve_dir` in `start` closes it // β€” a name that's already reserved must never reach `resolve_dir` at // all, so `dirs` stays exactly as a caller left it. let state = Arc::new(State::new(PathBuf::from("/dev/null"), signal_url())); @@ -4563,7 +4643,83 @@ mod tests { } fn swarm(models: &[&str]) -> crate::swarm_models::Read { - crate::swarm_models::Read::Set(models.iter().map(|m| (*m).to_owned()).collect()) + crate::swarm_models::Read::Current(AgentInference { + subagent_models: Some(models.iter().map(|m| (*m).to_owned()).collect()), + ..AgentInference::default() + }) + } + + fn unset() -> crate::swarm_models::Read { + crate::swarm_models::Read::Current(AgentInference::default()) + } + + fn capped(max: u32) -> crate::swarm_models::Read { + crate::swarm_models::Read::Current(AgentInference { + max_subagents: Some(max), + ..AgentInference::default() + }) + } + + #[test] + fn reserve_under_refuses_a_new_name_at_the_cap_and_takes_one_below_it() { + let state = State::new(PathBuf::from("/dev/null"), signal_url()); + assert_eq!(state.reserve_under("a", Some(2)), Ok(())); + assert_eq!(state.reserve_under("b", Some(2)), Ok(())); + assert_eq!( + state.reserve_under("c", Some(2)), + Err(Refusal::AtCap { running: 2, cap: 2 }) + ); + assert_eq!(state.reserve_under("c", None), Ok(()), "no cap, no limit"); + assert_eq!( + state.reserve_under("a", Some(3)), + Err(Refusal::Taken), + "a taken name at the cap is refused as taken, not as over the cap" + ); + } + + /// A run between turns and one winding down both still count. + #[test] + fn every_tracked_slot_counts_toward_the_cap() { + let state = State::new(PathBuf::from("/dev/null"), signal_url()); + assert!(state.reserve("a")); + state.between_turns("a"); + assert!(state.reserve("b")); + state + .running + .lock() + .unwrap() + .insert("b".to_owned(), Slot::Stopping); + assert_eq!( + state.reserve_under("c", Some(2)), + Err(Refusal::AtCap { running: 2, cap: 2 }) + ); + } + + #[test] + fn the_cap_is_the_swarm_value_and_none_otherwise() { + assert_eq!(subagent_cap(Some(&capped(3))), Some(3)); + assert_eq!(subagent_cap(Some(&unset())), None); + assert_eq!(subagent_cap(None), None); + let failed = crate::swarm_models::Read::Failed("no answer".to_owned()); + assert_eq!(subagent_cap(Some(&failed)), None); + } + + /// `start` and `continue` read the cap live: the swarm's cap of 1 refuses + /// a second name, naming the count and the cap. + #[tokio::test] + async fn start_and_continue_refuse_a_new_run_at_the_swarm_cap() { + let (state, reads) = state_reading(capped(1), "busy"); + let started = start(&state, start_on("other", None)).await; + let continued = continue_on(&state, "other", None).await; + for result in [started, continued] { + let err = result.expect_err("one running, cap is 1").to_string(); + assert!( + err.contains("1 subagent running, cap is 1") && err.contains("swarm-ui"), + "{err}" + ); + } + assert_eq!(reads.load(std::sync::atomic::Ordering::Relaxed), 2); + assert_eq!(state.occupancy("other"), None, "nothing was reserved"); } /// The swarm's list replaces the local one, so it can widen it as well @@ -4580,13 +4736,10 @@ mod tests { #[test] fn without_a_swarm_list_the_local_one_applies() { let env = || Some("haiku,sonnet".to_owned()); - for read in [None, Some(crate::swarm_models::Read::Unset)] { + for read in [None, Some(unset())] { assert_eq!(model_list(read, env()), local(Some("haiku,sonnet"))); } - assert_eq!( - model_list(Some(crate::swarm_models::Read::Unset), None), - local(None) - ); + assert_eq!(model_list(Some(unset()), None), local(None)); } #[test] @@ -4630,7 +4783,7 @@ mod tests { /// A state whose swarm read answers `read` and whose own list is `haiku`, /// with `name` already reserved: a call that gets past the model check - /// stops at `reserve` and spawns nothing. Also the reader's read count. + /// stops at `reserve_under` and spawns nothing. Also the reader's read count. fn state_reading( read: crate::swarm_models::Read, name: &str, @@ -4666,7 +4819,7 @@ mod tests { .await } - /// The refusal comes from the model check, ahead of `reserve`'s. + /// The refusal comes from the model check, ahead of `reserve_under`'s. #[tokio::test] async fn start_and_continue_refuse_a_model_outside_the_swarm_list_before_reserving() { let (state, reads) = state_reading(swarm(&["sonnet"]), "n"); @@ -4692,8 +4845,9 @@ mod tests { } } + /// The key is read without a model too, for the cap; no model is checked. #[tokio::test] - async fn start_and_continue_without_a_model_do_not_read_the_swarm() { + async fn start_and_continue_without_a_model_skip_the_model_check() { let (state, reads) = state_reading(swarm(&["sonnet"]), "n"); let started = start(&state, start_on("n", None)).await; let continued = continue_on(&state, "n", None).await; @@ -4701,7 +4855,7 @@ mod tests { let err = result.expect_err("the name is reserved").to_string(); assert!(err.contains("already running"), "{err}"); } - assert_eq!(reads.load(std::sync::atomic::Ordering::Relaxed), 0); + assert_eq!(reads.load(std::sync::atomic::Ordering::Relaxed), 2); } /// A `tracing` writer whose output a test can read back. diff --git a/hive-subagent-mcp/src/swarm_models.rs b/hive-subagent-mcp/src/swarm_models.rs index 2c4595c0..9cf482a8 100644 --- a/hive-subagent-mcp/src/swarm_models.rs +++ b/hive-subagent-mcp/src/swarm_models.rs @@ -1,11 +1,13 @@ -//! Reads the subagent model list the swarm sets for this agent. +//! Reads the subagent settings the swarm sets for this agent. //! //! swarm-controller keeps the agent's settings under its key in the //! `agent-inference` KV bucket. Its `subagent_models`, when set, is the list //! a `start` or `continue` naming a `model` is checked against, in place of -//! `HIVE_AVAILABLE_MODELS` (`crate::session::check_model`). +//! `HIVE_AVAILABLE_MODELS` (`crate::session::check_model`). Its +//! `max_subagents`, when set, caps how many subagents run at once +//! (`crate::session::State::reserve_under`). //! -//! **Read on every such call**, never cached, with the one subject the +//! **Read on every `start` and `continue`**, never cached, with the one subject the //! agent's own queue credential is granted in that bucket //! (`swarm_queue_client::agent_inference::read_own`). The connection is made //! the way [`crate::swarm_term`] makes its own: as the agent, never as the @@ -13,14 +15,15 @@ //! //! **A subagent start never waits on the queue.** Before the first connect //! succeeds, and when a read fails or outlasts `READ_TIMEOUT`, the read is -//! [`Read::Failed`] and the caller checks against `HIVE_AVAILABLE_MODELS`. +//! [`Read::Failed`]: the caller checks against `HIVE_AVAILABLE_MODELS` and +//! applies no cap. #[cfg(test)] use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, OnceLock}; use std::time::Duration; -use swarm_queue_client::agent_inference::{self, read_own}; +use swarm_queue_client::agent_inference::{self, AgentInference, read_own}; use crate::swarm_term::{Target, connect}; @@ -32,13 +35,12 @@ const RETRY_AFTER: Duration = Duration::from_mins(1); /// misgranted agent's `start` waits. const READ_TIMEOUT: Duration = Duration::from_secs(2); -/// What one read of this agent's key said about `subagent_models`. +/// What one read of this agent's key found. #[derive(Debug, Clone, PartialEq, Eq)] pub enum Read { - /// The swarm sets the list. - Set(Vec), - /// The key, the field, or the bucket is absent. - Unset, + /// The settings the swarm holds; an absent key or bucket has every field + /// unset. + Current(AgentInference), /// Nothing could be read, and why. Failed(String), } @@ -70,7 +72,7 @@ impl Reader { Ok(None) => { tracing::info!( "no swarm queue address or no secret store for this agent; subagent models \ - are checked against HIVE_AVAILABLE_MODELS only" + are checked against HIVE_AVAILABLE_MODELS only, and no subagent cap applies" ); return None; } @@ -78,7 +80,7 @@ impl Reader { tracing::warn!( error = format!("{e:#}"), "this agent's secret store coordinates are incomplete; subagent models are \ - checked against HIVE_AVAILABLE_MODELS only" + checked against HIVE_AVAILABLE_MODELS only, and no subagent cap applies" ); return None; } @@ -103,7 +105,7 @@ impl Reader { (Self { source }, reads) } - /// This agent's `subagent_models` as the swarm holds it now. + /// This agent's settings as the swarm holds them now. pub async fn read(&self) -> Read { match &self.source { Source::Queue { agent, client } => read_queue(agent, client).await, @@ -128,12 +130,10 @@ async fn read_queue(agent: &str, client: &OnceLock) -> Read Err(_) => Read::Failed(format!("no answer within {READ_TIMEOUT:?}")), Ok(Err(e)) => Read::Failed(e.to_string()), Ok(Ok(agent_inference::Read::NoBucket)) => { - tracing::debug!("subagent models: the agent-inference bucket does not answer"); - Read::Unset - } - Ok(Ok(agent_inference::Read::Current(current))) => { - current.subagent_models.map_or(Read::Unset, Read::Set) + tracing::debug!("subagent settings: the agent-inference bucket does not answer"); + Read::Current(AgentInference::default()) } + Ok(Ok(agent_inference::Read::Current(current))) => Read::Current(current), } } @@ -142,7 +142,7 @@ async fn connect_until_up(target: Arc, client: Arc { - tracing::info!(url = %target.url, "subagent models: connected as this agent"); + tracing::info!(url = %target.url, "subagent settings: connected as this agent"); let _already_set = client.set(connected); return; } @@ -150,8 +150,8 @@ async fn connect_until_up(target: Arc, client: Arc>, + /// How many subagents the agent may have running at once, from 1 to + /// [`MAX_SUBAGENTS`]. Absent is no cap. + #[serde(default, skip_serializing_if = "Option::is_none")] + max_subagents: Option, } impl From for InferenceSettings { @@ -55,6 +65,7 @@ impl From for InferenceSettings { model: v.model, effort: v.effort, subagent_models: v.subagent_models, + max_subagents: v.max_subagents, } } } @@ -95,10 +106,18 @@ impl InferenceSettings { } } } + if let Some(max) = self.max_subagents + && !(1..=MAX_SUBAGENTS).contains(&max) + { + return Err(format!( + "max_subagents is {max}; it is from 1 to {MAX_SUBAGENTS}, or left out for no cap" + )); + } Ok(AgentInference { model: self.model, effort: self.effort, subagent_models: self.subagent_models, + max_subagents: self.max_subagents, }) } } @@ -266,8 +285,8 @@ pub(crate) async fn get_agent_inference( (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, starts with `-`, or is listed twice \ - (problem+json)", + model name is empty, has whitespace, starts with `-`, or is listed \ + twice, or `max_subagents` is outside 1 to 32 (problem+json)", body = String), (status = 422, description = "an unknown field, or an effort that is not one of \ the five levels"), @@ -363,6 +382,7 @@ mod tests { model: Some("opus".to_owned()), effort: Some(Effort::High), subagent_models: Some(vec!["haiku".to_owned()]), + max_subagents: None, } ); assert_eq!( @@ -489,6 +509,25 @@ mod tests { } } + #[test] + fn max_subagents_outside_its_range_is_refused() { + for json in [r#"{"max_subagents":0}"#, r#"{"max_subagents":33}"#] { + assert!(settings(json).unwrap().into_stored().is_err(), "{json}"); + } + assert!(settings(r#"{"max_subagents":-1}"#).is_err()); + } + + #[test] + fn max_subagents_in_range_is_stored() { + for max in [1, 3, MAX_SUBAGENTS] { + let stored = settings(&format!(r#"{{"max_subagents":{max}}}"#)) + .unwrap() + .into_stored() + .unwrap(); + assert_eq!(stored.max_subagents, Some(max)); + } + } + /// Control for the leading-dash refusal: full model ids have dashes inside. #[test] fn a_dash_inside_a_model_name_is_accepted() { diff --git a/swarm-queue-client/src/agent_inference.rs b/swarm-queue-client/src/agent_inference.rs index 35e26686..ad69f0f8 100644 --- a/swarm-queue-client/src/agent_inference.rs +++ b/swarm-queue-client/src/agent_inference.rs @@ -1,5 +1,5 @@ //! The per-agent inference-settings KV bucket: one key per agent, holding the -//! model, effort and subagent model list the swarm sets for it. +//! model, effort, subagent model list and subagent cap 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 @@ -46,6 +46,10 @@ pub struct AgentInference { /// The models the agent may start its subagents on. #[serde(default, skip_serializing_if = "Option::is_none")] pub subagent_models: Option>, + /// How many subagents the agent may have running at once. `None` is no + /// cap. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_subagents: Option, } /// A claude `--effort` level, ascending. @@ -196,11 +200,12 @@ mod tests { model: Some("opus".to_owned()), effort: Some(Effort::Xhigh), subagent_models: Some(vec!["haiku".to_owned(), "sonnet".to_owned()]), + max_subagents: Some(3), }; let encoded = serde_json::to_string(&value).unwrap(); assert_eq!( encoded, - r#"{"model":"opus","effort":"xhigh","subagent_models":["haiku","sonnet"]}"# + r#"{"model":"opus","effort":"xhigh","subagent_models":["haiku","sonnet"],"max_subagents":3}"# ); assert_eq!( serde_json::from_str::(&encoded).unwrap(), @@ -225,6 +230,16 @@ mod tests { ); } + /// A value written before `max_subagents` existed decodes with no cap. + #[test] + fn a_value_without_max_subagents_has_no_cap() { + let decoded: AgentInference = + serde_json::from_str(r#"{"model":"opus","effort":"high","subagent_models":["haiku"]}"#) + .unwrap(); + assert_eq!(decoded.max_subagents, None); + assert_eq!(decoded.model.as_deref(), Some("opus")); + } + #[test] fn every_effort_is_stored_as_its_cli_spelling() { for effort in [ diff --git a/swarmctl/src/inference.rs b/swarmctl/src/inference.rs index 840e1508..6fc6f252 100644 --- a/swarmctl/src/inference.rs +++ b/swarmctl/src/inference.rs @@ -24,6 +24,8 @@ struct Settings { effort: Option, #[serde(default, skip_serializing_if = "Option::is_none")] subagent_models: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + max_subagents: Option, } /// What `GET` answers: the settings, and what the agent reported running on. @@ -59,6 +61,7 @@ impl SetFlags { model: self.model.or(current.model), effort: self.effort.or(current.effort), subagent_models: self.subagent_models.or(current.subagent_models), + max_subagents: current.max_subagents, }) } } @@ -145,12 +148,14 @@ fn render_configured(s: Option<&Settings>) -> String { fn render(s: &Settings) -> String { const UNSET: &str = "(not set; the agent's own)"; format!( - "model: {}\neffort: {}\nsubagent_models: {}\n", + "model: {}\neffort: {}\nsubagent_models: {}\nmax_subagents: {}\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(", ")) + .map_or_else(|| UNSET.to_owned(), |m| m.join(", ")), + s.max_subagents + .map_or_else(|| "(not set; no cap)".to_owned(), |n| n.to_string()) ) } @@ -208,6 +213,19 @@ mod tests { assert_eq!(merged.effort.as_deref(), Some("high")); } + /// `set` has no flag for the cap, and a stored one survives it. + #[test] + fn set_keeps_the_stored_max_subagents() { + let current: Settings = serde_json::from_str(r#"{"max_subagents":3}"#).unwrap(); + let merged = SetFlags { + model: Some("opus".to_owned()), + ..flags() + } + .merge(current) + .unwrap(); + assert_eq!(merged.max_subagents, Some(3)); + } + #[test] fn set_with_no_flag_is_refused() { assert!(flags().merge(Settings::default()).is_err()); @@ -236,16 +254,18 @@ mod tests { let empty: View = serde_json::from_str(r#"{"configured":{},"effective":null}"#).unwrap(); assert_eq!( render_configured(empty.configured.as_ref()).lines().count(), - 3 + 4 ); } #[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_eq!(row.lines().count(), 4); + assert!(row.ends_with("max_subagents: (not set; no cap)\n"), "{row}"); assert!( row.lines() + .take(3) .all(|l| l.ends_with("(not set; the agent's own)")), "{row}" );