hive-c0re: bound the read-only container listing and systemctl status calls

priv_client::list_containers now waits at most 30s (LIST_WAIT) for
hive-priv's `nixos-container list`, via the existing `within` helper. Its
periodic callers (crash_watch, agent_sockets, swarm_agent_status, wanted,
mcp_sockets, swarm_status via container_view) already treat an Err as a
skipped cycle, so a hung listing no longer stops their loops for good.

lifecycle::is_running / unit_state run `systemctl is-active` with
kill_on_drop and a 10s timeout (IS_ACTIVE_TIMEOUT); a timeout warns and
reads as not-running / UnitState::Other, the same as a systemctl that
could not be spawned. The new lifecycle::running_state returns None for
that case, and crash_watch keeps the agent's previous observation on None
so a hung systemctl is neither reported as a crash nor as a recovery.

`call` itself and every mutating op (start/stop/destroy/send-snapshot/
daemon-reload) stay unbounded.

Closes #5061

hive-c0re/src/lifecycle/mod.rs

@@ -616,34 +616,70 @@ impl UnitState {
 pub async fn unit_state(name: &str) -> UnitState {
     let container = container_name(name);
     let unit = format!("container@{container}.service");
-    match Command::new("systemctl")
-        .args(["is-active", &unit])
-        .output()
+    // Not "the unit is fine" and not "the unit failed" — we simply do not
+    // know, and saying `Failed` here would report a gave-up agent every time
+    // `systemctl` itself was unavailable or hung.
+    systemctl_is_active(&[&unit])
         .await
-    {
-        Ok(out) => UnitState::from_is_active(&String::from_utf8_lossy(&out.stdout)),
-        // Not "the unit is fine" and not "the unit failed" — we simply do
-        // not know, and saying `Failed` here would report a gave-up agent
-        // every time `systemctl` itself was unavailable.
-        Err(_) => UnitState::Other,
-    }
+        .map_or(UnitState::Other, |out| {
+            UnitState::from_is_active(&String::from_utf8_lossy(&out.stdout))
+        })
 }
 
 /// True when the container's systemd unit is active. Used by the dashboard
-/// to gate stop/restart buttons.
+/// to gate stop/restart buttons. A state that could not be read (see
+/// [`running_state`]) is `false`.
 ///
 /// Kept boolean on purpose: nearly every caller is reconcile/power logic
 /// asking "is it up? if not, start it", and that question has two answers.
 /// A caller that needs to tell *failed* from *stopped* wants
 /// [`unit_state`] instead.
 pub async fn is_running(name: &str) -> bool {
+    running_state(name).await == Some(true)
+}
+
+/// Whether the container's systemd unit is active; `None` when `systemctl`
+/// could not be run or did not answer within [`IS_ACTIVE_TIMEOUT`].
+pub async fn running_state(name: &str) -> Option<bool> {
     let container = container_name(name);
     let unit = format!("container@{container}.service");
-    Command::new("systemctl")
-        .args(["is-active", "--quiet", &unit])
-        .status()
+    systemctl_is_active(&["--quiet", &unit])
         .await
-        .is_ok_and(|s| s.success())
+        .map(|out| out.status.success())
+}
+
+/// How long one `systemctl is-active` may run before the unit's state counts
+/// as unreadable. Periodic workers make this call per agent inline.
+const IS_ACTIVE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+
+/// Run `systemctl is-active <args>`, killed after [`IS_ACTIVE_TIMEOUT`].
+async fn systemctl_is_active(args: &[&str]) -> Option<std::process::Output> {
+    let output = Command::new("systemctl")
+        .arg("is-active")
+        .args(args)
+        .kill_on_drop(true)
+        .output();
+    output_within(IS_ACTIVE_TIMEOUT, output, args).await
+}
+
+/// Await a child's output for at most `limit`. `None` when it failed to run
+/// or timed out; the timeout drops `fut`, which kills a child spawned with
+/// `kill_on_drop(true)`.
+async fn output_within<T>(
+    limit: std::time::Duration,
+    fut: impl std::future::Future<Output = std::io::Result<T>>,
+    args: &[&str],
+) -> Option<T> {
+    if let Ok(result) = tokio::time::timeout(limit, fut).await {
+        result.ok()
+    } else {
+        tracing::warn!(
+            ?args,
+            timeout_s = limit.as_secs(),
+            "systemctl is-active timed out"
+        );
+        None
+    }
 }
 
 /// Fully tear down a sub-agent's container: stop + remove via `nixos-container

hive-c0re/src/lifecycle/tests.rs

@@ -449,3 +449,27 @@ fn is_actionable_nix_line_keeps_warnings_and_errors_demotes_chatter() {
         assert!(!is_actionable_nix_line(line), "{line}");
     }
 }
+
+#[tokio::test]
+async fn a_hung_systemctl_reads_as_unknown_within_the_bound() {
+    let started = tokio::time::Instant::now();
+    let got = output_within(
+        std::time::Duration::from_millis(20),
+        std::future::pending::<std::io::Result<()>>(),
+        &["is-active"],
+    )
+    .await;
+    assert!(got.is_none());
+    assert!(started.elapsed() < std::time::Duration::from_secs(5));
+}
+
+#[tokio::test]
+async fn a_systemctl_answer_inside_the_bound_passes_through() {
+    let got = output_within(
+        std::time::Duration::from_secs(5),
+        async { Ok(7_u8) },
+        &["is-active"],
+    )
+    .await;
+    assert_eq!(got, Some(7));
+}

hive-c0re/src/priv_client.rs

@@ -271,8 +271,16 @@ pub async fn destroy_container(name: &str) -> Result<()> {
     .await?)
 }
 
+/// How long hive-c0re waits for hive-priv's `nixos-container list`. Periodic
+/// workers await the listing inline, so a hung one must reach them as an
+/// error they skip a cycle on, not as a wait that stops their loop.
+const LIST_WAIT: std::time::Duration = std::time::Duration::from_secs(30);
+
+/// # Errors
+/// Returns an error if the hive-priv call fails or doesn't answer within
+/// [`LIST_WAIT`], or `nixos-container list` exits non-zero.
 pub async fn list_containers() -> Result<String> {
-    let (stdout, _) = check(call(&PrivRequest::ListContainers).await?)?;
+    let (stdout, _) = check(within(LIST_WAIT, call(&PrivRequest::ListContainers)).await?)?;
     Ok(stdout)
 }
 

hive-c0re/src/workers/crash_watch.rs

@@ -45,7 +45,8 @@ pub fn spawn(coord: Arc<Coordinator>) {
                     };
                     let logical = logical.to_owned();
                     sub_agents.push(logical.clone());
-                    if lifecycle::is_running(&logical).await {
+                    let state = lifecycle::running_state(&logical).await;
+                    if counts_as_running(state, prev_running.contains(&logical)) {
                         current_running.insert(logical.clone());
                     }
                     if hive_types::Ident::parse(&logical)
@@ -114,6 +115,12 @@ fn plan_cycle(list_result: anyhow::Result<Vec<String>>) -> CyclePlan {
     }
 }
 
+/// An unreadable unit state (`None`) keeps the agent's last observed state,
+/// so a hung or failed `systemctl` is neither a crash nor a recovery.
+fn counts_as_running(state: Option<bool>, was_running: bool) -> bool {
+    state.unwrap_or(was_running)
+}
+
 fn emit_crash_transitions(coord: &Coordinator, prev: &HashSet<String>, current: &HashSet<String>) {
     let transients = coord.transient_snapshot();
     // Operator actions whose RAII guard already cleared but only just;
@@ -234,6 +241,15 @@ mod tests {
         }
     }
 
+    #[test]
+    fn unreadable_unit_state_keeps_the_last_observation() {
+        assert!(counts_as_running(None, true), "no crash on a timeout");
+        assert!(!counts_as_running(None, false), "no recovery on a timeout");
+        // Control: a readable stop still drops out of the running set.
+        assert!(!counts_as_running(Some(false), true));
+        assert!(counts_as_running(Some(true), false));
+    }
+
     #[test]
     fn deliberate_when_either_source_says_so() {
         // Active guard, and the race repro: a lifecycle action completes and