diff --git a/docs/agent-lifecycle/persistence.md b/docs/agent-lifecycle/persistence.md index 3ca1043c..851f7e73 100644 --- a/docs/agent-lifecycle/persistence.md +++ b/docs/agent-lifecycle/persistence.md @@ -411,6 +411,13 @@ Contents: default for CPU. The `CPUWeight=` / `IOWeight=` shares in the same drop-in have **no** per-agent override — they're hive-wide only and come straight off `HiveEnv`, so this file has no field for them. +- `swarm-resource-limits.json` — the caps the swarm sets for this + hive's agents, same shape as `resource-limits.json`. Written only by + `workers::swarm_limits`, which copies each agent's key of the swarm's + `agent-limits` bucket here every 30 seconds. A field set here replaces + that field's answer from `resource-limits.json` and the hive-wide + default. Every value passes the same validators as the operator's + overrides first, and one bad value makes the worker change nothing here. The root agent has the meta dir RO-mounted at `/meta/`. diff --git a/docs/scheduler/coordinator.md b/docs/scheduler/coordinator.md index 20d4dab3..d6cf23ed 100644 --- a/docs/scheduler/coordinator.md +++ b/docs/scheduler/coordinator.md @@ -452,14 +452,24 @@ regardless. rebuild, so changes take effect on the next lifecycle op without requiring a host rebuild. +The answer per agent and per field is, first match wins: + +1. the swarm's cap for the agent (see below); +2. the operator's per-agent override in `meta/resource-limits.json`; +3. `agentCpuQuota` / `agentMemoryMax`. + +A swarm cap replaces the hive's answer, so a hive can't raise an agent +above it. Like the rest of the drop-in, a changed swarm cap reaches a +running agent at its next spawn or rebuild. + The container can't read its own cap, so `write_dropins` also writes the effective `MemoryMax=` into the agent's runtime dir (`/run/hive/memory-max-bytes` inside) before every start: a byte count, or `infinity` for no cap. A percentage becomes bytes the way systemd resolves it, as that share of the host's physical memory rounded down to a page. A value systemd ignores (zero, over 100%) is `infinity`. When hive-c0re keeps -the last drop-in because it can't read the overrides file, the file carries -that drop-in's `MemoryMax=`. +the last drop-in because it can't read the overrides or the swarm caps file, +the file carries that drop-in's `MemoryMax=`. At boot the container derives two values from a byte count: `BUN_JSC_forceRAMSize` (75%, claude-code's heap ceiling) and `MemoryHigh=` on @@ -469,6 +479,34 @@ changed cap reaches them at the next container start without a rebuild. `hyperhive-memory-cap-check.service` logs a warning naming the path in the container's journal. The container's `MemoryMax=` still applies. +
Implementation: swarm caps + +swarm-controller keeps the caps in the `agent-limits` KV bucket, one key per +agent and no hive in it, so an agent that moves to another hive keeps its +caps. swarm-controller alone writes it: + +| method | path | body | +| -------- | ---------------------------- | --------------------------------------------------------------------------------------------- | +| `GET` | `/api/agents//limits` | —; answers the stored caps, `{}` when there are none | +| `PUT` | `/api/agents//limits` | `{"cpu_quota": "400%", "memory_max": "8G"}`, every field optional, replacing the stored value | +| `DELETE` | `/api/agents//limits` | —; answers 204 | + +`PUT` refuses an agent no hive's wanted state declares, unknown fields, zero +and anything systemd doesn't take, and stores memory the way systemd spells +it (`8gb` as `8G`). + +Every 30 seconds hive-c0re (`workers::swarm_limits`) reads the key of each +agent it has and stores the result in `meta/swarm-resource-limits.json`. Each +value passes the same validators as the operator's overrides first. One bad +value or a failed read leaves the file unchanged, and repeated failures raise a +dashboard banner. + +A hive's queue grant reads every key of the bucket +(`$JS.API.DIRECT.GET.KV_agent-limits.>`) and writes none. Scoping it to the +hive's own agents would need a roster in the auth responder. + +
+ The same drop-in carries `CPUWeight=` / `IOWeight=` from `agentCpuWeight` / `agentIoWeight`. Those are a different kind of setting: the quota and the memory max are **hard caps** that throttle diff --git a/docs/swarm/README.md b/docs/swarm/README.md index 8c89734c..0535bd71 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -25,6 +25,9 @@ grafana, the controller, authelia, …) → 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) +- **resource caps** — the CPU and memory cap the swarm sets for an agent, + over its hive's. → + [`coordinator.md`](../scheduler/coordinator.md#container-resource-limits) - **review classifier** — planned, not implemented yet: checks a PR against its repo's own yes/no rules and posts the result as a review. → [`review-classifier.md`](review-classifier.md) diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index ac74773f..9e8d0f0c 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -53,7 +53,11 @@ strum.workspace = true # Offering this hive's status to the swarm (`swarm_status`). The same crate # the swarm controller reads it with, and `kv` for the same reason: the # bucket's name and creation config belong to neither end of it alone. -swarm-queue-client = { workspace = true, features = ["kv", "notices"] } +swarm-queue-client = { workspace = true, features = [ + "kv", + "notices", + "agent-limits", +] } swarm-secret-client.workspace = true tokio.workspace = true tokio-stream.workspace = true diff --git a/hive-c0re/src/swarm_status.rs b/hive-c0re/src/swarm_status.rs index 13a1d455..c4cfc54b 100644 --- a/hive-c0re/src/swarm_status.rs +++ b/hive-c0re/src/swarm_status.rs @@ -103,6 +103,12 @@ pub fn spawn( shutdown.clone(), )); + // The swarm's resource caps, on the same connection again. + tokio::spawn(crate::workers::swarm_limits::follow( + client.clone(), + shutdown.clone(), + )); + let mut health = SweepHealth::new("swarm_status_publish", "warn", FAILURES_BEFORE_BANNER); loop { match publish(&client, &hive).await { diff --git a/hive-c0re/src/workers/mod.rs b/hive-c0re/src/workers/mod.rs index c3772e92..6f2ffa4c 100644 --- a/hive-c0re/src/workers/mod.rs +++ b/hive-c0re/src/workers/mod.rs @@ -1,6 +1,7 @@ //! Background tasks and periodic sweeps: crash/login watcher, boot-time //! auto-update reconcile, the agent-sockets.json writer loop, the MCP socket listener -//! reconcile loop, knowledge-repo sync, and the swarm wanted-state pull. +//! reconcile loop, knowledge-repo sync, the swarm wanted-state pull, and the +//! swarm resource-cap read. //! Each submodule is re-exported //! at the crate root, so `crate::crash_watch::…` etc. keep working unchanged. @@ -9,4 +10,5 @@ pub mod auto_update; pub mod crash_watch; pub mod knowledge; pub mod mcp_sockets; +pub mod swarm_limits; pub mod wanted; diff --git a/hive-c0re/src/workers/swarm_limits.rs b/hive-c0re/src/workers/swarm_limits.rs new file mode 100644 index 00000000..ec3b0979 --- /dev/null +++ b/hive-c0re/src/workers/swarm_limits.rs @@ -0,0 +1,207 @@ +//! Reading the caps the swarm sets for this hive's agents. +//! +//! swarm-controller keeps one key per agent in the `agent-limits` bucket +//! ([`swarm_queue_client::agent_limits`]). Every [`READ_INTERVAL`] this reads +//! the key of each agent this hive has and stores what it finds in +//! `swarm-resource-limits.json`, where the drop-in and the dashboard read it. +//! A running agent gets a changed cap at its first spawn or rebuild after +//! that read. +//! +//! A failed read, or a bucket that does not answer, leaves the stored caps as +//! they are: dropping them would read exactly like the swarm clearing them. + +use std::collections::BTreeMap; +use std::time::Duration; + +use anyhow::{Context, Result}; +use swarm_queue_client::agent_limits::{self, AgentCaps, Read}; + +use crate::resource_limits::AgentLimits; +use crate::stats::sweep_health::{self, SweepHealth}; + +/// How often the caps are read. +pub const READ_INTERVAL: Duration = Duration::from_secs(30); + +/// At [`READ_INTERVAL`] this is ~90 seconds of failing reads. +const FAILURES_BEFORE_BANNER: u32 = 3; + +/// Read the caps now and every [`READ_INTERVAL`] until shutdown. +pub async fn follow(client: async_nats::Client, mut shutdown: tokio::sync::watch::Receiver) { + let mut health = SweepHealth::new("swarm_limits_read", "warn", FAILURES_BEFORE_BANNER); + loop { + match sync(&client).await { + Ok(()) => health.record_ok(), + Err(e) => { + tracing::warn!(error = ?e, "swarm limits: reading the caps failed"); + let err = format!("{e:#}"); + health.record_err(|ctx| { + let age = ctx.since_last_ok.map_or_else( + || "no success this session".to_owned(), + |d| format!("last ok {} ago", sweep_health::fmt_age(d)), + ); + format!( + "reading the swarm's resource caps is failing ({} consecutive, {age}) \ + — agents keep the caps last read: {err}", + ctx.consecutive + ) + }); + } + } + tokio::select! { + () = tokio::time::sleep(READ_INTERVAL) => {} + _ = shutdown.changed() => { + tracing::info!("swarm limits: shutdown signal received"); + return; + } + } + } +} + +/// Read every agent's key and store the result, unless a read fails or a +/// value is refused. +async fn sync(client: &async_nats::Client) -> Result<()> { + // An unconnected client does not fail a JetStream request, it hangs on it. + swarm_queue_client::ensure_connected(client)?; + let agents = crate::lifecycle::agents_for_meta_listing() + .await + .context("enumerating this hive's agents")?; + let mut declared = BTreeMap::new(); + for spec in agents { + let read = agent_limits::read(client, &spec.name) + .await + .map_err(|e| anyhow::anyhow!(e)) + .with_context(|| format!("reading the swarm's caps for {}", spec.name))?; + match read { + Read::Current(caps) => { + declared.insert(spec.name, caps); + } + Read::NoBucket => { + tracing::debug!("swarm limits: the agent-limits bucket does not answer"); + return Ok(()); + } + } + } + let caps = validated(&declared)?; + if caps != crate::resource_limits::read_swarm().context("reading the stored swarm caps")? { + crate::resource_limits::write_swarm(&caps).context("storing the swarm caps")?; + tracing::info!(?caps, "swarm limits: stored the swarm's resource caps"); + } + Ok(()) +} + +/// The caps to store for `declared`, by agent. An agent with no field set +/// gets no entry. +/// +/// Every value goes through the validators the operator's own input path +/// uses, because it is rendered verbatim into the container's systemd +/// drop-in. One bad value refuses the whole set. +fn validated(declared: &BTreeMap) -> Result> { + let mut out = BTreeMap::new(); + for (agent, caps) in declared { + if caps.is_empty() { + continue; + } + let limits = validated_caps(caps) + .map_err(|why| anyhow::anyhow!("the swarm set an invalid cap for {agent}: {why}"))?; + out.insert(agent.clone(), limits); + } + Ok(out) +} + +fn validated_caps(caps: &AgentCaps) -> Result { + if let Some(cpu) = &caps.cpu_quota { + crate::resource_limits::validate_cpu_quota(cpu)?; + } + Ok(AgentLimits { + cpu_quota: caps.cpu_quota.clone(), + memory_max: caps + .memory_max + .as_deref() + .map(crate::resource_limits::validate_memory_max) + .transpose()?, + }) +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use super::validated; + use crate::resource_limits::AgentLimits; + use swarm_queue_client::agent_limits::AgentCaps; + + /// `atlas`'s answer with `atlas_key` as the value read for it, over a hive + /// chain that caps it at 800% / 16G. + fn atlas_with(atlas_key: &str) -> (String, String) { + let declared = BTreeMap::from([( + "atlas".to_owned(), + serde_json::from_str::(atlas_key).expect("decodes"), + )]); + let stored = validated(&declared).expect("valid caps"); + let hive_side = BTreeMap::from([( + "atlas".to_owned(), + AgentLimits { + cpu_quota: Some("800%".to_owned()), + memory_max: Some("16G".to_owned()), + }, + )]); + crate::resource_limits::effective_from_with_swarm( + &stored, &hive_side, "atlas", "200%", "4G", + ) + } + + /// The other half of swarm-controller's + /// `agent_limits::the_route_writes_the_caps_a_hive_reads_and_delete_clears_them`: + /// the first value is the bytes that route stores. The cap replaces the + /// hive's chain; a deleted key reads as `{}` and gives the chain back. + #[test] + fn the_caps_the_route_stores_replace_the_hive_chain_and_a_cleared_key_restores_it() { + assert_eq!( + atlas_with(r#"{"cpu_quota":"400%","memory_max":"8G"}"#), + ("400%".to_owned(), "8G".to_owned()) + ); + assert_eq!(atlas_with("{}"), ("800%".to_owned(), "16G".to_owned())); + } + + /// Per field: a swarm memory cap leaves CPU on the hive's chain. + #[test] + fn a_field_the_swarm_leaves_unset_stays_with_the_hive() { + assert_eq!( + atlas_with(r#"{"memory_max":"2G"}"#), + ("800%".to_owned(), "2G".to_owned()) + ); + } + + #[test] + fn an_agent_with_no_caps_gets_no_entry() { + let declared = BTreeMap::from([("atlas".to_owned(), AgentCaps::default())]); + assert!(validated(&declared).expect("valid").is_empty()); + } + + /// The values are rendered verbatim into a systemd drop-in, so a value + /// the hive's own validators refuse, a newline smuggling in a directive + /// among them, refuses the whole set, the valid cap beside it included. + #[test] + fn an_invalid_cap_is_refused_and_nothing_is_stored() { + for bad in [ + r#"{"cpu_quota":"400%\nExecStartPre=/bin/sh -c id #%"}"#, + r#"{"memory_max":"8G\nDevicePolicy=auto"}"#, + r#"{"cpu_quota":"lots"}"#, + r#"{"memory_max":"8X"}"#, + ] { + let declared = BTreeMap::from([ + ("atlas".to_owned(), AgentCaps::default()), + ( + "iris".to_owned(), + serde_json::from_str(r#"{"memory_max":"2G"}"#).expect("decodes"), + ), + ( + "sock".to_owned(), + serde_json::from_str(bad).expect("decodes"), + ), + ]); + let err = validated(&declared).expect_err(bad); + assert!(err.to_string().contains("sock"), "{err:#}"); + } + } +} diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index b9d6db69..86a4d703 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -490,6 +490,17 @@ impl Policy { format!("$JS.API.STREAM.CREATE.{agent_status}"), format!("$KV.{}.{hive}.*", swarm_queue_client::agent_status::BUCKET), ]); + // Reading the caps the swarm sets for the agents this hive runs + // (`swarm_queue_client::agent_limits`): a direct get, so read-only, and + // no `STREAM.INFO` because the reader skips the lookup. + // + // Every agent's key, not just this hive's: the bucket is keyed by agent + // alone, and naming one hive's agents here needs the roster + // `hive_name`'s doc argues against. + subjects.push(format!( + "$JS.API.DIRECT.GET.{}.>", + swarm_queue_client::agent_limits::STREAM + )); subjects.extend( self.extra_hive_subjects .iter() @@ -712,6 +723,44 @@ mod tests { } } + /// A hive reads any agent's caps and can change none of them, and the + /// grant reaches no other bucket. + #[test] + fn a_hive_may_read_every_agents_caps_and_write_none() { + use swarm_queue_client::agent_limits::{STREAM, direct_get_subject, key_subject}; + + let p = policy() + .permissions("hive-alpha") + .expect("a hive is admitted"); + let may = |subject: &str| p.publish.iter().any(|g| covers(g, subject)); + + // Its own agents and agents running elsewhere alike. + for agent in ["atlas", "iris"] { + assert!(may(&direct_get_subject(agent)), "cannot read {agent}"); + } + // A KV put or delete is a publish to the key's subject; the rest + // reshape or empty the stream. + for forbidden in [ + key_subject("atlas"), + format!("$JS.API.STREAM.CREATE.{STREAM}"), + format!("$JS.API.STREAM.UPDATE.{STREAM}"), + format!("$JS.API.STREAM.DELETE.{STREAM}"), + format!("$JS.API.STREAM.PURGE.{STREAM}"), + format!("$JS.API.STREAM.MSG.DELETE.{STREAM}"), + format!("$JS.API.CONSUMER.CREATE.{STREAM}"), + ] { + assert!(!may(&forbidden), "hive alpha may publish {forbidden}"); + } + // The read reaches only this bucket. + for other in [ + swarm_queue_client::agent_inference::direct_get_subject("atlas"), + swarm_queue_client::agent_schedules::direct_get_subject("atlas"), + "$JS.API.DIRECT.GET.KV_hive-wanted-beta.$KV.hive-wanted-beta.beta".to_owned(), + ] { + assert!(!may(&other), "hive alpha may publish {other}"); + } + } + #[test] fn an_extra_subject_without_the_placeholder_is_refused() { // 🩸 The option exists to put a second stream inside ONE hive's