hive-c0re: let a swarm cap replace the hive's answer, not join its chain

mara's ruling: "swarm level overrides replace the hive ui limits. otherwise
dont change the logic."

  swarm value present  ->  it is the value; nothing hive-side is consulted
  swarm value absent   ->  exactly today's chain: per-agent field, else hive

Per field, so a swarm memory ceiling leaves CPU on the hive's chain.

The fallback chain itself is not touched: effective_from_with_swarm calls
effective_from as it was and lets the swarm replace its result, and
effective_with_swarm delegates to effective, so there stays one place the
chain lives. `an_empty_swarm_map_leaves_the_existing_chain_bit_for_bit`
asserts the wrapper is a no-op across every shape the chain distinguishes.

The swarm's caps live in their own swarm-resource-limits.json, not in the
operator's resource-limits.json: different writers, different lifetimes,
and a swarm write into the operator's file would make "reset to hive
defaults" destroy something the operator never set. It is read and written
through the same read_map / write_map, so an unreadable file is an error,
and write_dropins resolves both files through limits_to_apply, whose
keep-the-last-applied-drop-in policy covers the swarm file too.

stage_limits hands limits_to_apply the swarm caps file too, so the
agent's memory-max-bytes copy, which it sizes its heap and subagents
against, is the MemoryMax= systemd enforces: the swarm cap when there is
one.

Nothing writes swarm-resource-limits.json yet; the reader comes with the
swarm's agent-limits bucket later in this series.

hive-c0re/src/agent_config/resource_limits.rs

@@ -187,6 +187,86 @@ fn set_limits_at(path: &Path, name: &str, limits: &AgentLimits) -> std::io::Resu
     super::write_map(path, &current)
 }
 
+/// Caps the swarm controller declared for this hive's agents.
+///
+/// A separate file from the operator's overrides above, not a merge into it:
+/// the two have different writers and different lifetimes, and a swarm
+/// declaration landing in the file an operator edits would make "reset to
+/// hive defaults" destroy something the operator never set.
+const SWARM_LIMITS_FILE: &str = "swarm-resource-limits.json";
+
+#[must_use]
+pub fn swarm_limits_path() -> PathBuf {
+    crate::paths::meta_root().join(SWARM_LIMITS_FILE)
+}
+
+/// Read the swarm-declared caps. An absent file is the empty map — no swarm
+/// opinion, so the hive's own chain decides. A file that exists but can't be
+/// read or parsed is an error, for the reason [`read`] gives: reading it as
+/// empty would silently drop a cap.
+pub fn read_swarm() -> std::io::Result<BTreeMap<String, AgentLimits>> {
+    super::read_map(&swarm_limits_path())
+}
+
+/// Persist the swarm-declared caps atomically.
+///
+/// # Errors
+/// The meta dir cannot be created, or the file cannot be written.
+pub fn write_swarm(map: &BTreeMap<String, AgentLimits>) -> std::io::Result<()> {
+    super::write_map(&swarm_limits_path(), map)
+}
+
+/// [`effective_from`] with the swarm's declaration on top.
+///
+/// A swarm cap **replaces** the hive's answer for that field rather than
+/// joining the fallback chain as another tier: a tier a hive can out-set is
+/// not a cap. Per field, so a swarm memory ceiling leaves CPU on the hive's
+/// chain.
+///
+/// The chain itself is untouched — [`effective_from`] is called exactly as
+/// before and its result is what a missing swarm value falls back to.
+#[must_use]
+pub fn effective_from_with_swarm(
+    swarm: &BTreeMap<String, AgentLimits>,
+    limits: &BTreeMap<String, AgentLimits>,
+    name: &str,
+    hive_cpu_quota: &str,
+    hive_memory_max: &str,
+) -> (String, String) {
+    let (cpu, mem) = effective_from(limits, name, hive_cpu_quota, hive_memory_max);
+    match swarm.get(name) {
+        Some(s) => (
+            s.cpu_quota.clone().unwrap_or(cpu),
+            s.memory_max.clone().unwrap_or(mem),
+        ),
+        None => (cpu, mem),
+    }
+}
+
+/// [`effective_from_with_swarm`] for a caller resolving a single agent, from
+/// the operator overrides at `limits_path` and the swarm caps at `swarm_path`.
+///
+/// Delegates to [`effective`] rather than re-deriving the hive's answer, so
+/// there stays exactly one place the fallback chain lives. Errors as [`read`]
+/// does, for either file.
+pub fn effective_with_swarm(
+    limits_path: &Path,
+    swarm_path: &Path,
+    name: &str,
+    hive_cpu_quota: &str,
+    hive_memory_max: &str,
+) -> std::io::Result<(String, String)> {
+    let (cpu, mem) = effective(limits_path, name, hive_cpu_quota, hive_memory_max)?;
+    let swarm: BTreeMap<String, AgentLimits> = super::read_map(swarm_path)?;
+    Ok(match swarm.get(name) {
+        Some(s) => (
+            s.cpu_quota.clone().unwrap_or(cpu),
+            s.memory_max.clone().unwrap_or(mem),
+        ),
+        None => (cpu, mem),
+    })
+}
+
 /// Validate a systemd `CPUQuota=` value. Percentages only, and values
 /// above 100% are legitimate (one full core is 100%).
 ///
@@ -525,4 +605,52 @@ mod tests {
             );
         }
     }
+
+    #[test]
+    fn a_swarm_cap_replaces_the_hive_answer_rather_than_joining_the_chain() {
+        // The whole point of the ruling: a tier a hive can out-set is not a
+        // cap. The agent override below is the strongest hive-side value
+        // there is, and the swarm still wins.
+        let hive_side = BTreeMap::from([("sock".to_owned(), limits(Some("800%"), Some("16G")))]);
+        let swarm = BTreeMap::from([("sock".to_owned(), limits(Some("150%"), Some("2G")))]);
+        assert_eq!(
+            effective_from_with_swarm(&swarm, &hive_side, "sock", HIVE_CPU, HIVE_MEM),
+            ("150%".to_owned(), "2G".to_owned())
+        );
+    }
+
+    #[test]
+    fn a_swarm_cap_is_per_field_like_every_other_tier() {
+        // A swarm that caps memory and says nothing about CPU leaves CPU on
+        // the hive's chain — here the per-agent override, not the hive default.
+        let hive_side = BTreeMap::from([("sock".to_owned(), limits(Some("800%"), None))]);
+        let swarm = BTreeMap::from([("sock".to_owned(), limits(None, Some("2G")))]);
+        assert_eq!(
+            effective_from_with_swarm(&swarm, &hive_side, "sock", HIVE_CPU, HIVE_MEM),
+            ("800%".to_owned(), "2G".to_owned())
+        );
+    }
+
+    #[test]
+    fn an_empty_swarm_map_leaves_the_existing_chain_bit_for_bit() {
+        // The control that makes the two tests above mean something: the
+        // wrapper must be a no-op when the swarm has said nothing, for every
+        // shape the old chain distinguishes. If this drifts, the swarm tier
+        // changed behaviour it was explicitly told not to change.
+        let none = BTreeMap::new();
+        let cases = [
+            BTreeMap::new(),
+            BTreeMap::from([("sock".to_owned(), limits(Some("400%"), Some("8G")))]),
+            BTreeMap::from([("sock".to_owned(), limits(Some("400%"), None))]),
+            BTreeMap::from([("sock".to_owned(), limits(None, Some("8G")))]),
+            BTreeMap::from([("other".to_owned(), limits(Some("100%"), Some("1G")))]),
+        ];
+        for hive_side in &cases {
+            assert_eq!(
+                effective_from_with_swarm(&none, hive_side, "sock", HIVE_CPU, HIVE_MEM),
+                effective_from(hive_side, "sock", HIVE_CPU, HIVE_MEM),
+                "the wrapper changed the answer with no swarm value present"
+            );
+        }
+    }
 }

hive-c0re/src/container_view.rs

@@ -171,6 +171,14 @@ pub async fn build_all(hive: &crate::coordinator::HiveEnv) -> anyhow::Result<Vec
         );
         std::collections::BTreeMap::new()
     });
+    let swarm_limits = crate::resource_limits::read_swarm().unwrap_or_else(|e| {
+        tracing::error!(
+            path = %crate::resource_limits::swarm_limits_path().display(),
+            error = ?e,
+            "swarm resource caps unreadable — container rows ignore them"
+        );
+        std::collections::BTreeMap::new()
+    });
     let mut out = Vec::new();
     for c in &raw {
         let Some(logical) = c.strip_prefix(AGENT_PREFIX) else {
@@ -235,7 +243,8 @@ pub async fn build_all(hive: &crate::coordinator::HiveEnv) -> anyhow::Result<Vec
             (None, None)
         };
         let paused = Coordinator::is_paused(&logical);
-        let (cpu_quota, memory_max) = crate::resource_limits::effective_from(
+        let (cpu_quota, memory_max) = crate::resource_limits::effective_from_with_swarm(
+            &swarm_limits,
             &limits,
             logical.as_str(),
             &hive.agent_cpu_quota,

hive-c0re/src/lifecycle/host_config.rs

@@ -28,6 +28,7 @@ pub async fn write_dropins(name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Re
     if let Some((cpu_quota, memory_max)) = stage_limits(
         name,
         &crate::resource_limits::resource_limits_path(),
+        &crate::resource_limits::swarm_limits_path(),
         &limits_dropin_path(&container),
         &paths.agent,
         hive,
@@ -97,12 +98,13 @@ impl HostMemory {
 fn stage_limits(
     agent_name: &str,
     limits_path: &Path,
+    swarm_path: &Path,
     dropin_path: &Path,
     runtime_dir: &Path,
     hive: &HiveEnv,
     host: Option<HostMemory>,
 ) -> Result<Option<(String, String)>> {
-    let limits = limits_to_apply(agent_name, limits_path, dropin_path, hive);
+    let limits = limits_to_apply(agent_name, limits_path, swarm_path, dropin_path, hive);
     let memory_max = match &limits {
         Some((_, memory_max)) => Some(memory_max.clone()),
         None => dropin_memory_max(agent_name, dropin_path),
@@ -224,20 +226,28 @@ fn write_memory_cap(runtime_dir: &Path, cap: Option<&str>) -> Result<()> {
 /// The `(CPUQuota, MemoryMax)` to write into `agent_name`'s limits
 /// drop-in, or `None` to leave the drop-in at `dropin_path` as it is.
 ///
-/// When the overrides file at `limits_path` can't be read, the error is
+/// When the overrides file at `limits_path` or the swarm caps at
-/// logged and the agent still starts. The last-applied drop-in is kept,
+/// `swarm_path` can't be read, the error is logged and the agent still starts.
-/// because the hive defaults can be looser than the agent's own override.
+/// The last-applied drop-in is kept, because the hive defaults can be looser
+/// than the agent's own override.
 /// With no drop-in yet, the agent gets the hive defaults: without one it
 /// would run uncapped.
 fn limits_to_apply(
     agent_name: &str,
     limits_path: &Path,
+    swarm_path: &Path,
     dropin_path: &Path,
     hive: &HiveEnv,
 ) -> Option<(String, String)> {
     let hive_cpu = &hive.agent_cpu_quota;
     let hive_mem = &hive.agent_memory_max;
-    let e = match crate::resource_limits::effective(limits_path, agent_name, hive_cpu, hive_mem) {
+    let e = match crate::resource_limits::effective_with_swarm(
+        limits_path,
+        swarm_path,
+        agent_name,
+        hive_cpu,
+        hive_mem,
+    ) {
         Ok(limits) => return Some(limits),
         Err(e) => e,
     };
@@ -838,6 +848,10 @@ mod tests {
 
     const TRUNCATED_LIMITS: &str = "{\n  \"sock\": { \"cpu_quota\": \"50%\", \"mem";
 
+    fn absent_swarm(dir: &tempfile::TempDir) -> std::path::PathBuf {
+        dir.path().join("absent-swarm-resource-limits.json")
+    }
+
     #[test]
     fn corrupt_limits_keep_an_existing_dropin() {
         let dir = tempfile::tempdir().expect("tempdir");
@@ -846,7 +860,13 @@ mod tests {
         std::fs::write(&limits, TRUNCATED_LIMITS).expect("seed limits");
         std::fs::write(&dropin, "MemoryMax=1G\n").expect("seed drop-in");
         assert_eq!(
-            limits_to_apply("sock", &limits, &dropin, &HiveEnv::default()),
+            limits_to_apply(
+                "sock",
+                &limits,
+                &absent_swarm(&dir),
+                &dropin,
+                &HiveEnv::default()
+            ),
             None
         );
         assert_eq!(
@@ -863,12 +883,47 @@ mod tests {
         let applied = limits_to_apply(
             "sock",
             &limits,
+            &absent_swarm(&dir),
             &dir.path().join("absent.conf"),
             &HiveEnv::default(),
         );
         assert_eq!(applied, Some(("200%".to_owned(), "4G".to_owned())));
     }
 
+    #[test]
+    fn corrupt_swarm_caps_keep_an_existing_dropin() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let swarm = dir.path().join("swarm-resource-limits.json");
+        let dropin = dir.path().join("hyperhive-limits.conf");
+        std::fs::write(&swarm, TRUNCATED_LIMITS).expect("seed swarm caps");
+        std::fs::write(&dropin, "MemoryMax=1G\n").expect("seed drop-in");
+        let applied = limits_to_apply(
+            "sock",
+            &dir.path().join("resource-limits.json"),
+            &swarm,
+            &dropin,
+            &HiveEnv::default(),
+        );
+        assert_eq!(applied, None);
+    }
+
+    #[test]
+    fn a_swarm_cap_is_applied_over_the_operator_override() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let limits = dir.path().join("resource-limits.json");
+        let swarm = dir.path().join("swarm-resource-limits.json");
+        std::fs::write(&limits, r#"{"sock": {"memory_max": "16G"}}"#).expect("seed limits");
+        std::fs::write(&swarm, r#"{"sock": {"memory_max": "2G"}}"#).expect("seed swarm caps");
+        let applied = limits_to_apply(
+            "sock",
+            &limits,
+            &swarm,
+            &dir.path().join("absent.conf"),
+            &HiveEnv::default(),
+        );
+        assert_eq!(applied, Some(("200%".to_owned(), "2G".to_owned())));
+    }
+
     #[test]
     fn readable_limits_are_applied_over_an_existing_dropin() {
         let dir = tempfile::tempdir().expect("tempdir");
@@ -876,7 +931,13 @@ mod tests {
         let dropin = dir.path().join("hyperhive-limits.conf");
         std::fs::write(&limits, r#"{"sock": {"memory_max": "1G"}}"#).expect("seed limits");
         std::fs::write(&dropin, "MemoryMax=8G\n").expect("seed drop-in");
-        let applied = limits_to_apply("sock", &limits, &dropin, &HiveEnv::default());
+        let applied = limits_to_apply(
+            "sock",
+            &limits,
+            &absent_swarm(&dir),
+            &dropin,
+            &HiveEnv::default(),
+        );
         assert_eq!(applied, Some(("200%".to_owned(), "1G".to_owned())));
     }
 
@@ -893,6 +954,7 @@ mod tests {
         let applied = stage_limits(
             "sock",
             &dir.path().join("resource-limits.json"),
+            &dir.path().join("swarm-resource-limits.json"),
             &dir.path().join("hyperhive-limits.conf"),
             &runtime,
             &HiveEnv::default(),
@@ -903,6 +965,28 @@ mod tests {
         (applied.map(|(_, mem)| mem), copy)
     }
 
+    /// The agent's copy follows the swarm's cap, the one systemd enforces, over
+    /// the operator's override: a heap sized against the override would
+    /// outgrow the cgroup.
+    #[test]
+    fn the_agent_reads_the_swarm_cap_over_the_operator_override() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        std::fs::write(
+            dir.path().join("resource-limits.json"),
+            r#"{"sock": {"memory_max": "16G"}}"#,
+        )
+        .expect("seed limits");
+        std::fs::write(
+            dir.path().join("swarm-resource-limits.json"),
+            r#"{"sock": {"memory_max": "2G"}}"#,
+        )
+        .expect("seed swarm caps");
+        assert_eq!(
+            stage(&dir),
+            (Some("2G".to_owned()), Some("2147483648\n".to_owned()))
+        );
+    }
+
     #[test]
     fn the_agent_reads_the_cap_its_dropin_carries() {
         let dir = tempfile::tempdir().expect("tempdir");
@@ -1035,6 +1119,7 @@ mod tests {
         let applied = stage_limits(
             "sock",
             &limits,
+            &dir.path().join("swarm-resource-limits.json"),
             &dropin,
             &runtime,
             &HiveEnv::default(),

hive-c0re/src/server.rs

@@ -426,8 +426,9 @@ async fn handle_set_resource_limits(
     let paths = Coordinator::agent_paths(name.as_str(), agent_dir)?;
     crate::lifecycle::write_dropins(name.as_str(), &hive, &paths).await?;
 
-    let (cpu, mem) = crate::resource_limits::effective(
+    let (cpu, mem) = crate::resource_limits::effective_with_swarm(
         &crate::resource_limits::resource_limits_path(),
+        &crate::resource_limits::swarm_limits_path(),
         name.as_str(),
         &hive.agent_cpu_quota,
         &hive.agent_memory_max,