hive-c0re: fail on an unparseable resource-limits or topology file, write both atomically

resource-limits.json and topology.json were read with parse errors
folded into an empty map, and written in place with std::fs::write. One
truncated resource-limits.json followed by a single set_limits call
rewrote the file with only that agent's entry, erasing every other
agent's CPU and memory overrides without a log line. topology.json had
the same shape: reconcile rebuilt it from the live set, losing pending
(provisioned, never spawned) names.

- agent_config::read_map / write_map are generic over the stored type.
  tool-groups and capabilities behave as before.
- resource_limits::read / effective return an error for an existing but
  unreadable file; a missing file is still the empty map. set_limits
  fails without writing on such a file, and writes atomically.
- topology: reconcile fails without writing on an unreadable file and
  writes atomically. all_agents logs the error and returns no agents,
  so a ManageRootAgent holder starts without cross-agent mounts.

Read-path behaviour on an unreadable resource-limits.json, per caller:
- write_dropins (every spawn / swap / WriteDropin): logs the error and
  keeps the limits drop-in already under /run; the agent still starts.
  With no drop-in yet (first start since boot) it writes the hive
  defaults, because no drop-in means an uncapped container.
- render_flake: propagates, so sync_agents (and spawn/rebuild/destroy
  jobs) fail. An empty map would give tighter-capped agents the hive
  memoryMaxBytes.
- container_view::build_all: logs the error each scan and renders the
  rows at the hive defaults (no ContainerView wire change).
- set_resource_limits reply: propagates.

Closes #4731

hive-c0re/src/lifecycle/host_config.rs

@@ -24,24 +24,77 @@ pub async fn write_dropins(name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Re
     validate(name)?;
     let container = container_name(name);
     set_nspawn_flags(&container, &paths.agent, &paths.claude, &paths.notes).await?;
-    let (cpu_quota, memory_max) =
-        crate::resource_limits::effective(name, &hive.agent_cpu_quota, &hive.agent_memory_max);
-    set_resource_limits(
-        &container,
-        &cpu_quota,
-        &memory_max,
-        hive.agent_cpu_weight,
-        hive.agent_io_weight,
-    )
-    .await?;
+    if let Some((cpu_quota, memory_max)) = limits_to_apply(
+        name,
+        &crate::resource_limits::resource_limits_path(),
+        &limits_dropin_path(&container),
+        hive,
+    ) {
+        set_resource_limits(
+            &container,
+            &cpu_quota,
+            &memory_max,
+            hive.agent_cpu_weight,
+            hive.agent_io_weight,
+        )
+        .await?;
+    }
     systemd_daemon_reload().await
 }
 
+/// The limits drop-in hive-priv's `write_resource_limits` writes for
+/// `container`. Must match that path: a mismatch reads as "no drop-in
+/// yet" in [`limits_to_apply`].
+fn limits_dropin_path(container: &str) -> PathBuf {
+    PathBuf::from(format!(
+        "/run/systemd/system/container@{container}.service.d/hyperhive-limits.conf"
+    ))
+}
+
+/// 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
+/// logged and the agent still starts. 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,
+    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) {
+        Ok(limits) => return Some(limits),
+        Err(e) => e,
+    };
+    if dropin_path.exists() {
+        tracing::error!(
+            agent = %agent_name,
+            path = %limits_path.display(),
+            error = ?e,
+            "resource limits unreadable — keeping the last-applied limits drop-in"
+        );
+        None
+    } else {
+        tracing::error!(
+            agent = %agent_name,
+            path = %limits_path.display(),
+            error = ?e,
+            "resource limits unreadable and no limits drop-in yet — applying hive defaults"
+        );
+        Some((hive_cpu.clone(), hive_mem.clone()))
+    }
+}
+
 /// Write a systemd drop-in for `container@<container>.service` that applies
 /// the agent's effective resource caps — its per-agent overrides from
 /// `meta/resource-limits.json` where set, the hive-wide defaults
 /// otherwise. Goes under `/run/systemd/system/...` so it's ephemeral
-/// (regenerated on every spawn / rebuild).
+/// (regenerated on every spawn / rebuild, except as [`limits_to_apply`] says).
 ///
 /// The weights are hive-wide (`services.hyperhive.agentCpuWeight` /
 /// `agentIoWeight`) — unlike the caps they have no per-agent override in
@@ -426,8 +479,9 @@ async fn set_nspawn_flags(
 mod tests {
     use super::{
         BindMount, QUEUE_CLIENT_ID_CREDENTIAL, QUEUE_SECRET_CREDENTIAL, bind_child_agent_dirs,
-        holds_manage_root_agent, queue_agent_credentials,
+        holds_manage_root_agent, limits_to_apply, queue_agent_credentials,
     };
+    use crate::coordinator::HiveEnv;
 
     fn child_binds() -> Vec<BindMount> {
         let mut binds = Vec::new();
@@ -561,4 +615,48 @@ mod tests {
         assert!(holds_manage_root_agent("ruth", &path));
         assert!(!holds_manage_root_agent("alice", &path));
     }
+
+    const TRUNCATED_LIMITS: &str = "{\n  \"sock\": { \"cpu_quota\": \"50%\", \"mem";
+
+    #[test]
+    fn corrupt_limits_keep_an_existing_dropin() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let limits = dir.path().join("resource-limits.json");
+        let dropin = dir.path().join("hyperhive-limits.conf");
+        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()),
+            None
+        );
+        assert_eq!(
+            std::fs::read(&limits).expect("read"),
+            TRUNCATED_LIMITS.as_bytes()
+        );
+    }
+
+    #[test]
+    fn corrupt_limits_without_a_dropin_apply_hive_defaults() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let limits = dir.path().join("resource-limits.json");
+        std::fs::write(&limits, TRUNCATED_LIMITS).expect("seed limits");
+        let applied = limits_to_apply(
+            "sock",
+            &limits,
+            &dir.path().join("absent.conf"),
+            &HiveEnv::default(),
+        );
+        assert_eq!(applied, Some(("200%".to_owned(), "4G".to_owned())));
+    }
+
+    #[test]
+    fn readable_limits_are_applied_over_an_existing_dropin() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let limits = dir.path().join("resource-limits.json");
+        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());
+        assert_eq!(applied, Some(("200%".to_owned(), "1G".to_owned())));
+    }
 }