fix(container): companion probes are passive — no builds inside reconcile checks

Root cause of the .198 load spiral (2026-07-28): needs_repair() called
ensure_image_present() every 30s tick to render the expected unit, so
under IO pressure the image-existence check timed out, read as "image
missing", and a 900s podman build ran inside the PROBE while the
companion was up — each build pegging the disk that made probes fail.

- needs_repair() is now build/pull-free: unit file present → service
  is-active (10s cap; a hung systemctl reads as "assume active", never
  as dead) → unit matches one of the three image refs install_one could
  have written → context-newer-than-image staleness only when the unit
  uses the auto-built :latest.
- Per-companion 10-min repair cooldown after a failed install_one, so a
  failing build retries at most every REPAIR_COOLDOWN instead of every
  reconcile tick.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

core/archipelago/src/container/companion.rs

@@ -22,8 +22,10 @@
 //! single declarative call.
 
 use anyhow::{Context, Result};
+use std::collections::HashMap;
 use std::path::PathBuf;
-use std::time::Duration;
+use std::sync::{LazyLock, Mutex};
+use std::time::{Duration, Instant};
 use tokio::fs;
 use tokio::process::Command;
 use tracing::{info, warn};
@@ -35,6 +37,15 @@ const COMPANION_REGISTRY: &str = "146.59.87.168:3000/lfg2025";
 const COMPANION_IMAGE_CHECK_TIMEOUT: Duration = Duration::from_secs(15);
 const COMPANION_BUILD_TIMEOUT: Duration = Duration::from_secs(900);
 const COMPANION_PULL_TIMEOUT: Duration = Duration::from_secs(300);
+/// After a failed repair (image build/pull included), leave the companion
+/// alone for this long. Without it, a node under IO pressure retried a 900s
+/// image build every 30s reconcile tick — each build pegging the disk that
+/// made the probes fail in the first place (live-diagnosed on zaza-optiplex
+/// 2026-07-28: load 50, podman scans starved, apps page stuck).
+const REPAIR_COOLDOWN: Duration = Duration::from_secs(600);
+
+static REPAIR_FAILED_AT: LazyLock<Mutex<HashMap<&'static str, Instant>>> =
+    LazyLock::new(|| Mutex::new(HashMap::new()));
 
 /// Static description of one companion. The full list per backend
 /// app_id lives in `companions_for`.
@@ -464,12 +475,28 @@ pub async fn reconcile(installed_apps: &[String]) -> Vec<(String, anyhow::Error)
             match needs_repair(spec).await {
                 Ok(false) => {}
                 Ok(true) => {
+                    if let Some(failed_at) =
+                        REPAIR_FAILED_AT.lock().unwrap().get(spec.name).copied()
+                    {
+                        if failed_at.elapsed() < REPAIR_COOLDOWN {
+                            continue;
+                        }
+                    }
                     info!(
                         companion = spec.name,
                         "reconcile: companion not active, repairing"
                     );
-                    if let Err(e) = install_one(spec).await {
-                        failures.push((spec.name.to_string(), e));
+                    match install_one(spec).await {
+                        Ok(()) => {
+                            REPAIR_FAILED_AT.lock().unwrap().remove(spec.name);
+                        }
+                        Err(e) => {
+                            REPAIR_FAILED_AT
+                                .lock()
+                                .unwrap()
+                                .insert(spec.name, Instant::now());
+                            failures.push((spec.name.to_string(), e));
+                        }
                     }
                 }
                 Err(e) => {
@@ -484,19 +511,58 @@ pub async fn reconcile(installed_apps: &[String]) -> Vec<(String, anyhow::Error)
 
 /// Does this companion need install_one to be re-run? Returns true if
 /// the unit file is missing, stale, or the service is not active.
+///
+/// This probe runs every reconcile tick for every companion, so it must be
+/// PASSIVE: no image builds, no pulls. It used to call ensure_image_present
+/// to render the expected unit — under IO pressure the image-existence check
+/// inside timed out, read as "image missing", and a 900s `podman build` ran
+/// inside the probe even though the companion was up (the .198 load spiral).
 async fn needs_repair(spec: &CompanionSpec) -> Result<bool> {
     let dir = quadlet::unit_dir().await?;
     let unit_path = dir.join(format!("{}.container", spec.name));
     if !fs::try_exists(&unit_path).await.unwrap_or(false) {
         return Ok(true);
     }
-    let expected_image = ensure_image_present(spec).await?;
-    let expected_unit = build_unit(spec, &expected_image);
-    if expected_unit.render() != fs::read_to_string(&unit_path).await.unwrap_or_default() {
+    let svc = format!("{}.service", spec.name);
+    // A hung `systemctl is-active` under IO pressure must not read as
+    // "companion dead" — that's a repair (and possibly an image build) fired
+    // off exactly when the node can least afford one.
+    match tokio::time::timeout(Duration::from_secs(10), quadlet::is_active(&svc)).await {
+        Ok(active) => {
+            if !active {
+                return Ok(true);
+            }
+        }
+        Err(_) => {
+            warn!(
+                companion = spec.name,
+                "is-active probe timed out; assuming active"
+            );
+        }
+    }
+    // Service is running. Flag it stale only on definitive, cheap signals:
+    // the on-disk unit matching none of the image refs install_one could
+    // have written, or a local build context newer than the built image.
+    let on_disk = fs::read_to_string(&unit_path).await.unwrap_or_default();
+    let local_image = format!("localhost/{}:latest", spec.image_base);
+    let local_image_compat = format!("localhost/{}:local", spec.image_base);
+    let registry_image = format!("{}/{}:latest", COMPANION_REGISTRY, spec.image_base);
+    let matches_known_shape = [&local_image, &local_image_compat, &registry_image]
+        .iter()
+        .any(|img| build_unit(spec, img).render() == on_disk);
+    if !matches_known_shape {
         return Ok(true);
     }
-    let svc = format!("{}.service", spec.name);
-    Ok(!quadlet::is_active(&svc).await)
+    if on_disk.contains(&local_image) && !on_disk.contains(&local_image_compat) {
+        for dir in spec.build_dir_candidates {
+            let dockerfile = PathBuf::from(dir).join("Dockerfile");
+            if fs::try_exists(&dockerfile).await.unwrap_or(false) {
+                // Conservative on any timeout/error inside: reuse the cache.
+                return Ok(context_is_newer_than_image(dir, &local_image).await);
+            }
+        }
+    }
+    Ok(false)
 }
 
 #[cfg(test)]