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<u32>`. 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.
This commit is contained in:
atlas
mara
parent
6d7fc0dd99
commit
9a71d1e094
10 changed files with 393 additions and 94 deletions
|
|
@ -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<String> {
|
|||
/// 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<crate::swarm_term::Publisher>,
|
||||
/// 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<crate::swarm_models::Reader>,
|
||||
/// 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<crate::swarm_models::Reader>) -> 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<Option<u32>> {
|
||||
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<u32>) -> 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<crate::swarm_models::Read>, local: Option<String>) -> 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<crate::swarm_models::Read>, local: Option<String>) ->
|
|||
);
|
||||
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<u32> {
|
||||
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<State>, req: StartRequest) -> anyhow::Result<String> {
|
||||
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<String>,
|
||||
}
|
||||
|
||||
/// 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<String> {
|
||||
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<i32>) -> 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<String> {
|
|||
))
|
||||
}
|
||||
|
||||
/// `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.
|
||||
|
|
|
|||
|
|
@ -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<String>),
|
||||
/// 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<async_nats::Client>) -> 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<Target>, client: Arc<OnceLock<async_nats::
|
|||
loop {
|
||||
match connect(&target).await {
|
||||
Ok(connected) => {
|
||||
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<Target>, client: Arc<OnceLock<async_nats::
|
|||
tracing::warn!(
|
||||
error = format!("{e:#}"),
|
||||
retry_in = ?RETRY_AFTER,
|
||||
"subagent models: connecting with this agent's own queue credential failed; \
|
||||
HIVE_AVAILABLE_MODELS applies until a connect succeeds"
|
||||
"subagent settings: connecting with this agent's own queue credential failed; \
|
||||
HIVE_AVAILABLE_MODELS and no subagent cap apply until a connect succeeds"
|
||||
);
|
||||
tokio::time::sleep(RETRY_AFTER).await;
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue