zc2 0.0.29

//! Process supervision for `zc agent` (spec §6.5): spawn the broker and the
//! workers in their own process groups, observe them, drain and stop them,
//! and re-adopt them after an agent restart.

use crate::agent::config::AgentConfig;
use crate::agent::files::{ensure_dir, ChildRecord, Children, WORKER_LOG_DIR};
use crate::agent::merge::LocalView;
use crate::agent::plan::{self, Action, BrokerObs, Desired, Observed, WorkerObs};
use crate::agent::summary::{Problem, Requests, WorkerCounts};
use std::collections::{BTreeMap, HashSet};
use std::io::{Read as _, Seek, SeekFrom};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

pub const WORKER_LOG_MAX_BYTES: u64 = 5 * 1024 * 1024;
/// How long a `drain_timeout` stays on the problem list after it happened.
const DRAIN_TIMEOUT_SHOWN_FOR: Duration = Duration::from_secs(300);
/// A slot or the broker is considered "stable" (its crash-backoff counter
/// resets to zero) once it has run continuously for this long without
/// crashing. Same five-minute window `plan::is_crashloop` uses.
const CRASH_STABLE_AFTER: Duration = Duration::from_secs(300);
/// `plan::is_crashloop` only looks at failures within this window; failure
/// timestamps older than this are pruned so the lists don't grow forever.
const CRASHLOOP_WINDOW: Duration = Duration::from_secs(300);

/// The timeout of every HTTP call to the local broker (`/health`, `GET
/// /workers`, drain, delete). A reconcile tick already running when the agent
/// shuts down can still spend two of these on its broker checks before it
/// next looks at the stop flag.
pub(crate) const BROKER_CHECK_TIMEOUT: Duration = Duration::from_secs(2);

/// One entry of the broker's `GET /workers`, the fields the agent uses.
#[derive(Debug, Clone, PartialEq, serde::Deserialize)]
pub struct BrokerWorker {
    pub id: String,
    pub name: String,
    pub status: String,
    #[serde(default)]
    pub active_requests: u32,
    #[serde(default)]
    pub requests_5h: u64,
    #[serde(default)]
    pub requests_1w: u64,
}

/// The slice of the local broker the supervisor talks to.
pub trait BrokerApi {
    /// `/health` answers and identifies as `zakuro-broker`.
    fn is_broker(&self) -> bool;
    fn workers(&self) -> Result<Vec<BrokerWorker>, String>;
    fn drain(&self, id: &str) -> Result<(), String>;
    fn delete(&self, id: &str) -> Result<(), String>;

    /// `/health` at an arbitrary port answers and identifies as
    /// `zakuro-broker`. `is_broker()` is really `is_broker_at(self.port)`;
    /// `observe` also calls this directly at the *preferred* broker port
    /// (always 9000, regardless of any fallback already in use), to decide
    /// `Unmanaged` before ever choosing a fallback. `HttpBrokerApi` overrides
    /// this with a real `/health` GET, so the conservative "no" default below
    /// is never used in production; `FakeBroker` overrides it too, to drive
    /// `a_foreign_broker_on_the_preferred_port_is_unmanaged`.
    fn is_broker_at(&self, _port: u16) -> bool {
        false
    }
    /// True when the broker could actually bind `port` right now. No default:
    /// this does real socket I/O, so a fake that forgets to override it must
    /// fail to compile rather than silently binding real ports from the test
    /// binary. See `HttpBrokerApi::port_free` for the real implementation and
    /// why a loopback-only check is not enough on macOS.
    fn port_free(&self, port: u16) -> bool;
    /// Bind an OS-assigned port, read it back, and drop the socket
    /// immediately so the caller can bind it again itself. No default, for
    /// the same reason as `port_free`.
    fn os_assigned_port(&self) -> Option<u16>;
}

/// The broker falls back to this range (9001..=9010, the same range zc
/// clients scan for a local broker -- see `src/autobroker.rs`) when the
/// preferred port (9000) is held by something that isn't our broker.
pub const BROKER_FALLBACK_START: u16 = 9001;
pub const BROKER_FALLBACK_END: u16 = 9010;

/// Pure port choice: prefer `preferred`; else the first free port in
/// `fallback_start..=fallback_end`; else whatever `os_pick` returns; else
/// `None` when nothing at all is available.
pub fn choose_broker_port(
    preferred: u16,
    fallback_start: u16,
    fallback_end: u16,
    mut is_free: impl FnMut(u16) -> bool,
    os_pick: impl FnOnce() -> Option<u16>,
) -> Option<u16> {
    if is_free(preferred) {
        return Some(preferred);
    }
    for p in fallback_start..=fallback_end {
        if is_free(p) {
            return Some(p);
        }
    }
    os_pick()
}

pub struct HttpBrokerApi {
    pub port: u16,
    pub worker_key: Option<String>,
}

impl HttpBrokerApi {
    fn base(&self) -> String {
        format!("http://127.0.0.1:{}", self.port)
    }
    fn http() -> ureq::Agent {
        ureq::Agent::new_with_config(
            ureq::Agent::config_builder()
                .timeout_global(Some(BROKER_CHECK_TIMEOUT))
                .http_status_as_error(false)
                .build(),
        )
    }
}

/// A 2xx status counts as success; anything else — including no response at
/// all — is an error the caller must not read as "zero workers" or "idle".
fn is_success_status(status: u16) -> bool {
    (200..300).contains(&status)
}

impl BrokerApi for HttpBrokerApi {
    fn is_broker(&self) -> bool {
        self.is_broker_at(self.port)
    }
    fn is_broker_at(&self, port: u16) -> bool {
        Self::http()
            .get(&format!("http://127.0.0.1:{port}/health"))
            .call()
            .ok()
            .filter(|r| r.status().as_u16() == 200)
            .and_then(|r| r.into_body().read_to_string().ok())
            .is_some_and(|b| b.contains("zakuro-broker"))
    }
    /// A specific-address bind (`127.0.0.1:port`) succeeds on macOS/BSD even
    /// when another socket already listens on the wildcard `*:port`, because
    /// Rust std sets `SO_REUSEADDR` on Unix. The broker binds the wildcard
    /// address (`host: "0.0.0.0"`), so that blind spot would let the agent
    /// pick a port its own broker then fails to bind, crash-looping forever
    /// instead of moving on to the next candidate. A port is free only when
    /// BOTH hold: nothing accepts a loopback connection (catches a
    /// loopback-only holder, e.g. a Docker publish of `127.0.0.1:9000`) AND a
    /// wildcard bind of our own would actually succeed (catches a `*:port`
    /// holder, which the loopback check alone would miss).
    fn port_free(&self, port: u16) -> bool {
        !crate::up::is_port_open(port) && std::net::TcpListener::bind(("0.0.0.0", port)).is_ok()
    }
    fn os_assigned_port(&self) -> Option<u16> {
        std::net::TcpListener::bind(("127.0.0.1", 0))
            .ok()
            .and_then(|l| l.local_addr().ok())
            .map(|a| a.port())
    }
    fn workers(&self) -> Result<Vec<BrokerWorker>, String> {
        #[derive(serde::Deserialize)]
        struct List {
            workers: Vec<BrokerWorker>,
        }
        let resp = Self::http()
            .get(&format!("{}/workers", self.base()))
            .call()
            .map_err(|e| e.to_string())?;
        let status = resp.status().as_u16();
        if !is_success_status(status) {
            return Err(format!("workers: HTTP {status}"));
        }
        let body = resp
            .into_body()
            .read_to_string()
            .map_err(|e| e.to_string())?;
        serde_json::from_str::<List>(&body)
            .map(|l| l.workers)
            .map_err(|e| e.to_string())
    }
    fn drain(&self, id: &str) -> Result<(), String> {
        let resp = Self::http()
            .post(&format!("{}/workers/{id}/drain", self.base()))
            .send_empty()
            .map_err(|e| e.to_string())?;
        match resp.status().as_u16() {
            200 => Ok(()),
            s => Err(format!("drain {id}: HTTP {s}")),
        }
    }
    fn delete(&self, id: &str) -> Result<(), String> {
        let mut req = Self::http().delete(&format!("{}/workers/{id}", self.base()));
        if let Some(k) = &self.worker_key {
            req = req.header("X-Worker-Key", k.as_str());
        }
        let resp = req.call().map_err(|e| e.to_string())?;
        match resp.status().as_u16() {
            200 | 404 => Ok(()),
            s => Err(format!("delete {id}: HTTP {s}")),
        }
    }
}

/// Replace `{port}` and `{name}` in every argument.
pub fn render_argv(template: &[String], port: u16, name: &str) -> Vec<String> {
    template
        .iter()
        .map(|a| {
            a.replace("{port}", &port.to_string())
                .replace("{name}", name)
        })
        .collect()
}

/// Spawn `argv` in its own process group (so a signal to the group reaches
/// python under `uv`), appending stdout+stderr to `log`.
pub fn spawn_in_group(
    argv: &[String],
    cwd: Option<&Path>,
    env: &[(String, String)],
    log: &Path,
) -> std::io::Result<Child> {
    use std::os::unix::fs::OpenOptionsExt;
    use std::os::unix::process::CommandExt;
    let (prog, rest) = argv
        .split_first()
        .ok_or_else(|| std::io::Error::other("empty command"))?;
    if let Some(parent) = log.parent() {
        ensure_dir(parent)?;
    }
    let out = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .mode(0o644)
        .open(log)?;
    let err = out.try_clone()?;
    let mut cmd = Command::new(prog);
    cmd.args(rest)
        .stdin(Stdio::null())
        .stdout(Stdio::from(out))
        .stderr(Stdio::from(err))
        .process_group(0);
    if let Some(d) = cwd {
        cmd.current_dir(d);
    }
    for (k, v) in env {
        cmd.env(k, v);
    }
    cmd.spawn()
}

/// Signal the process group led by `pid` (our children lead their own
/// group). Refuses `pid <= 1`: we must never signal the whole session or PID 1.
pub fn signal_group(pid: u32, sig: libc::c_int) -> bool {
    if pid <= 1 {
        return false;
    }
    unsafe { libc::killpg(pid as libc::pid_t, sig) == 0 }
}

/// True while any process remains in the group led by `pid` (a `kill(pid, 0)`
/// -style liveness probe, but for the whole process group via `killpg`).
fn group_alive(pid: u32) -> bool {
    if pid <= 1 {
        return false;
    }
    unsafe { libc::killpg(pid as libc::pid_t, 0) == 0 }
}

/// The process exists and is not a zombie. Zombies count as dead: in a CI
/// container PID 1 may never reap orphaned grandchildren.
pub fn is_alive(pid: u32) -> bool {
    if unsafe { libc::kill(pid as libc::pid_t, 0) } != 0 {
        return false;
    }
    #[cfg(target_os = "linux")]
    {
        if let Ok(stat) = std::fs::read_to_string(format!("/proc/{pid}/stat")) {
            let state = stat
                .rfind(')')
                .and_then(|i| stat[i + 1..].split_whitespace().next());
            if state == Some("Z") {
                return false;
            }
        }
    }
    true
}

#[cfg(target_os = "linux")]
fn linux_start_ticks(pid: u32) -> Option<String> {
    let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
    // Field 22 (starttime); fields after the ")" that ends comm start at field 3.
    let after = &stat[stat.rfind(')')? + 1..];
    after
        .split_whitespace()
        .nth(19)
        .map(|t| format!("ticks:{t}"))
}

/// A process's start time: with its pid, identifies it across agent restarts
/// (pids are reused). `/proc` on Linux, `ps -o lstart=` elsewhere.
pub fn proc_start_time(pid: u32) -> Option<String> {
    #[cfg(target_os = "linux")]
    {
        if let Some(t) = linux_start_ticks(pid) {
            return Some(t);
        }
    }
    let out = Command::new("ps")
        .args(["-o", "lstart=", "-p", &pid.to_string()])
        .output()
        .ok()?;
    let s = String::from_utf8_lossy(&out.stdout).trim().to_string();
    (out.status.success() && !s.is_empty()).then_some(s)
}

/// Truncate `path` to empty once it passes `max` bytes. Children write with
/// O_APPEND, so they keep writing at the new end.
pub fn truncate_if_large(path: &Path, max: u64) -> std::io::Result<bool> {
    match std::fs::metadata(path) {
        Ok(m) if m.len() > max => {
            std::fs::OpenOptions::new()
                .write(true)
                .truncate(true)
                .open(path)?;
            Ok(true)
        }
        _ => Ok(false),
    }
}

/// The last `n` lines of `path`. Seeks to the last 64 KiB before reading, so
/// this never pulls a large log file fully into memory.
pub fn tail_lines(path: &Path, n: usize) -> String {
    const CAP: u64 = 64 * 1024;
    let mut f = match std::fs::File::open(path) {
        Ok(f) => f,
        Err(_) => return String::new(),
    };
    let len = f.metadata().map(|m| m.len()).unwrap_or(0);
    let start = len.saturating_sub(CAP);
    if start > 0 && f.seek(SeekFrom::Start(start)).is_err() {
        return String::new();
    }
    let mut buf = Vec::new();
    let _ = f.read_to_end(&mut buf);
    let text = String::from_utf8_lossy(&buf).to_string();
    let lines: Vec<&str> = text.lines().collect();
    lines[lines.len().saturating_sub(n)..].join("\n")
}

/// How to launch the broker and the workers.
#[derive(Debug, Clone)]
pub struct SpawnSpec {
    /// The broker command line, with `{port}` still a placeholder: it is
    /// rendered at spawn time with whatever port was actually chosen for
    /// this start (see `Supervisor::broker_port`), not a fixed one baked in
    /// up front.
    pub broker_argv: Vec<String>,
    /// Broker environment, minus `ZAKURO_AGENT_CHILD_PORT`: that is added at
    /// spawn time too, for the same reason.
    pub broker_env: Vec<(String, String)>,
    pub worker_template: Vec<String>,
    pub worker_env: Vec<(String, String)>,
    pub worker_cwd: Option<PathBuf>,
    pub base_worker_port: u16,
    /// The broker port to prefer (spec default 9000, or `ZAKURO_AGENT_PORT`'s
    /// broker counterpart). Tried first on every start; the supervisor falls
    /// back from here when it is held by something else.
    pub preferred_broker_port: u16,
    pub fp8: String,
    /// The agent dir: `broker.log` and `workers/w<i>.log` live here.
    pub log_dir: PathBuf,
    pub kill_grace: Duration,
    pub broker_stop_wait: Duration,
    pub drain_timeout: Duration,
}

impl SpawnSpec {
    /// The broker env matches `zc up -d` (`up.rs:213-236`); `child_env` comes
    /// last so tests can override any of it. The agent itself never calls
    /// `crate::broker::apply_user_broker_defaults()` — that would mutate the
    /// agent's own process environment. The spawned `<zc> broker <port>`
    /// child applies it on its own (`start_broker`, src/main.rs).
    pub fn from_config(cfg: &AgentConfig) -> Self {
        let scan_end = cfg.base_worker_port as u32 + cfg.max_workers + 19;
        let mut broker_env = vec![
            (
                "ZAKURO_SCAN_RANGE".to_string(),
                format!("{}-{scan_end}", cfg.base_worker_port),
            ),
            ("ZAKURO_SCAN_INTERVAL".to_string(), "3".to_string()),
            (
                "ZAKURO_P2P".to_string(),
                crate::broker::p2p_default().to_string(),
            ),
        ];
        broker_env.extend(cfg.child_env.iter().cloned());
        let mut worker_env = vec![
            ("ZAKURO_WORKER_TYPE".to_string(), "zakuro".to_string()),
            ("UV_NO_PROGRESS".to_string(), "1".to_string()),
        ];
        worker_env.extend(cfg.child_env.iter().cloned());
        // Workers inherit the agent's env with `child_env` on top.
        let effective = |k: &str| {
            cfg.child_env
                .iter()
                .rev()
                .find(|(ek, _)| ek == k)
                .map(|(_, v)| v.clone())
                .or_else(|| std::env::var(k).ok())
        };
        if let Some((k, v)) = crate::up::worker_bind_env(&effective) {
            worker_env.push((k.to_string(), v.to_string()));
        }
        Self {
            broker_argv: cfg.broker_argv.clone(),
            broker_env,
            worker_template: cfg.worker_template.clone(),
            worker_env,
            worker_cwd: None,
            base_worker_port: cfg.base_worker_port,
            preferred_broker_port: cfg.broker_port,
            fp8: cfg.node_fp8(),
            log_dir: cfg.dir.clone(),
            kill_grace: cfg.timings.kill_grace,
            broker_stop_wait: cfg.timings.broker_stop_wait,
            drain_timeout: cfg.timings.drain_timeout,
        }
    }
}

/// A child process; `child` is None for one adopted after an agent restart.
struct Proc {
    pid: u32,
    started_at: String,
    child: Option<Child>,
}

impl Proc {
    fn spawned(child: Child) -> Self {
        let pid = child.id();
        Self {
            pid,
            started_at: proc_start_time(pid).unwrap_or_default(),
            child: Some(child),
        }
    }
    fn alive(&mut self) -> bool {
        match &mut self.child {
            Some(c) => matches!(c.try_wait(), Ok(None)),
            None => is_alive(self.pid),
        }
    }
    /// `port` is `Some` only for the broker's own record: workers' ports are
    /// already deterministic (`base_worker_port + index`) so they carry `None`.
    fn record(&self, port: Option<u16>) -> ChildRecord {
        ChildRecord {
            pid: self.pid,
            started_at: self.started_at.clone(),
            port,
        }
    }
}

/// SIGTERM the group, wait up to `grace` for every member to exit, then
/// SIGKILL the group if any member is still around. `killpg(pid, 0)` checks
/// the whole group, not just the leader, so a member that outlives the
/// leader (e.g. a backgrounded grandchild after its parent shell exits) is
/// still caught. While polling we opportunistically reap our own leader
/// (non-blocking) so a leader that already exited doesn't keep reading as
/// "the group is still alive" just because it is an unreaped zombie. Once
/// the group is confirmed gone we never probe or signal that pid again — a
/// freed pgid can be reused by an unrelated process.
///
/// `stop` is checked on every poll: once it is set (the agent is shutting
/// down), this returns `Some(p)` — the group untouched, still only
/// SIGTERMed, never SIGKILLed — instead of running out the rest of `grace`.
/// That keeps a reconcile tick from holding the supervisor lock for the
/// whole grace period while the agent is trying to exit, WITHOUT turning
/// what was meant to be a graceful stop (the broker flushing its WAL, say)
/// into an instant kill: the caller must give `p` back to `self.broker` or
/// its worker slot, so `terminate_all` — which honors its own full
/// SIGTERM/wait/SIGKILL sequence — is the one that finishes it off.
///
/// Returns `None` once the group is actually done, either because every
/// member exited on its own within `grace` or because `grace` ran out and
/// the group was SIGKILLed.
fn terminate(mut p: Proc, grace: Duration, stop: &AtomicBool) -> Option<Proc> {
    signal_group(p.pid, libc::SIGTERM);
    let deadline = Instant::now() + grace;
    let mut gone = false;
    while Instant::now() < deadline {
        if stop.load(Ordering::Relaxed) {
            return Some(p);
        }
        if let Some(c) = p.child.as_mut() {
            let _ = c.try_wait();
        }
        if !group_alive(p.pid) {
            gone = true;
            break;
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    if !gone && group_alive(p.pid) {
        signal_group(p.pid, libc::SIGKILL);
    }
    if let Some(c) = p.child.as_mut() {
        let _ = c.wait();
    }
    None
}

#[derive(Default)]
struct Slot {
    proc_: Option<Proc>,
    draining_since: Option<Instant>,
    failures: Vec<Instant>,
    consecutive: u32,
    restart_at: Option<Instant>,
    /// When this slot's current process was last (re)confirmed running. Once
    /// a slot has run for `CRASH_STABLE_AFTER` without crashing, its
    /// `consecutive` failure count resets, so a later, unrelated crash starts
    /// back at the 1 s backoff instead of continuing to escalate.
    up_since: Option<Instant>,
}

pub struct Supervisor {
    spec: SpawnSpec,
    broker: Option<Proc>,
    /// The port this run's broker is on while it's up: `spec.preferred_broker_port`
    /// until an observe tick finds it held by something else and picks a
    /// fallback (see `choose_broker_port`), or a value restored by `adopt`.
    /// While idle this still holds the last-used value, but that's not a
    /// preview of the next port: a fresh start re-probes from
    /// `spec.preferred_broker_port` rather than resuming the fallback.
    broker_port: u16,
    broker_stopping: bool,
    /// The broker's own crash-backoff/crashloop bookkeeping, mirroring a
    /// worker `Slot`'s `failures`/`consecutive`/`restart_at`/`up_since`.
    broker_failures: Vec<Instant>,
    broker_consecutive: u32,
    broker_restart_at: Option<Instant>,
    broker_up_since: Option<Instant>,
    slots: BTreeMap<u32, Slot>,
    /// The broker's last-known worker list (also the source of each worker's
    /// broker-assigned id, for `DrainWorker`/`KillWorker`'s DELETE). Replaced
    /// only when the tick is fresh (see `workers_fresh`); otherwise it is
    /// left as-is, since an unknown observation — the broker still
    /// `Starting`/`Stopping`, or a failed `GET /workers` — must never read as
    /// "idle" for a draining worker or lose track of its id. It is cleared
    /// (to reflect a real fact, not an unknown) only once the broker is
    /// confirmed `Stopped`, `Unmanaged` or `PortBlocked`.
    last_workers: Vec<BrokerWorker>,
    /// True exactly when this tick's `last_workers` is a fresh observation:
    /// the broker was `Running` and `GET /workers` just succeeded, or the
    /// broker is confirmed gone (so "no workers" is a fact, not a guess).
    /// `false` while `Starting`/`Stopping` or after a failed fetch.
    workers_fresh: bool,
    /// Whether the current `GET /workers` failure streak has already been
    /// logged, so a persistent failure logs once, not every reconcile tick.
    workers_err_streak: bool,
    last_broker: BrokerObs,
    last_drain_timeout: Option<Instant>,
    /// Set from outside the mutex (see `stop_flag`) when the agent is
    /// shutting down. The `terminate` waits an in-flight reconcile tick may
    /// be blocked in (stopping the broker, killing a worker) check this and
    /// return early instead of running out their full grace period, so the
    /// tick releases the supervisor lock quickly and `shutdown_children` can
    /// take over.
    stop: Arc<AtomicBool>,
}

fn broker_label(b: BrokerObs) -> &'static str {
    match b {
        BrokerObs::Running => "running",
        BrokerObs::Starting => "starting",
        BrokerObs::Stopping => "stopping",
        BrokerObs::Stopped => "stopped",
        BrokerObs::Unmanaged => "unmanaged",
        BrokerObs::PortBlocked => "error",
    }
}

impl Supervisor {
    pub fn new(spec: SpawnSpec) -> Self {
        let broker_port = spec.preferred_broker_port;
        Self {
            spec,
            broker: None,
            broker_port,
            broker_stopping: false,
            broker_failures: vec![],
            broker_consecutive: 0,
            broker_restart_at: None,
            broker_up_since: None,
            slots: BTreeMap::new(),
            last_workers: vec![],
            workers_fresh: true,
            workers_err_streak: false,
            last_broker: BrokerObs::Stopped,
            last_drain_timeout: None,
            stop: Arc::new(AtomicBool::new(false)),
        }
    }

    /// A handle to this supervisor's stop flag, so the owner can request an
    /// early exit from any in-flight `terminate` wait without needing the
    /// supervisor mutex.
    pub fn stop_flag(&self) -> Arc<AtomicBool> {
        self.stop.clone()
    }

    /// The port this run's broker is on while it's up: `spec.preferred_broker_port`
    /// until a fallback is chosen, or a value restored by `adopt`. The caller
    /// builds its `BrokerApi` against this port so `/health`, `GET /workers`,
    /// drain and delete all reach the broker actually running, wherever it
    /// landed. While idle this still reports the last port used, not the
    /// next one: a fresh start re-picks from `spec.preferred_broker_port`
    /// rather than resuming the fallback.
    pub fn broker_port(&self) -> u16 {
        self.broker_port
    }

    /// Update the directory a newly spawned worker's process runs in. Called
    /// once per reconcile tick (a cheap file check upstream resolved it
    /// fresh), so a zakuro dir that appears after the agent started is
    /// picked up by the very next worker spawn, without a restart.
    pub fn set_worker_cwd(&mut self, dir: Option<PathBuf>) {
        self.spec.worker_cwd = dir;
    }

    fn worker_name(&self, i: u32) -> String {
        crate::up::worker_name(&self.spec.fp8, i as usize)
    }

    fn worker_log(&self, i: u32) -> PathBuf {
        self.spec
            .log_dir
            .join(WORKER_LOG_DIR)
            .join(format!("w{i}.log"))
    }

    fn broker_log(&self) -> PathBuf {
        self.spec.log_dir.join("broker.log")
    }

    /// Re-attach to recorded children whose pid is alive and whose start time
    /// still matches (spec §6.5 "Agent restart").
    pub fn adopt(&mut self, recorded: &Children, start_time: &dyn Fn(u32) -> Option<String>) {
        let ok = |r: &ChildRecord| {
            is_alive(r.pid) && start_time(r.pid).as_deref() == Some(r.started_at.as_str())
        };
        let adopted = |r: &ChildRecord| Proc {
            pid: r.pid,
            started_at: r.started_at.clone(),
            child: None,
        };
        if let Some(b) = recorded.broker.as_ref().filter(|b| ok(b)) {
            self.broker = Some(adopted(b));
            self.broker_port = b.port.unwrap_or(self.spec.preferred_broker_port);
        }
        for (k, r) in &recorded.workers {
            if let (Ok(i), true) = (k.parse::<u32>(), ok(r)) {
                self.slots.entry(i).or_default().proc_ = Some(adopted(r));
            }
        }
    }

    pub fn children(&self) -> Children {
        let broker_port = self.broker_port;
        Children {
            broker: self.broker.as_ref().map(|p| p.record(Some(broker_port))),
            workers: self
                .slots
                .iter()
                .filter_map(|(i, s)| s.proc_.as_ref().map(|p| (i.to_string(), p.record(None))))
                .collect(),
        }
    }

    pub fn broker_unmanaged(&self) -> bool {
        self.last_broker == BrokerObs::Unmanaged
    }

    fn observe(
        &mut self,
        api: &dyn BrokerApi,
        prerequisites_ok: bool,
        sharing: bool,
        now: Instant,
    ) -> Observed {
        let broker_alive = self.broker.as_mut().map(|b| b.alive());
        let broker = match broker_alive {
            Some(true) if self.broker_stopping => BrokerObs::Stopping,
            Some(true) if api.is_broker() => {
                // Reset the crash-backoff counter once the broker has run
                // stably for a while, mirroring a worker slot (below).
                match self.broker_up_since {
                    None => self.broker_up_since = Some(now),
                    Some(since) if now.saturating_duration_since(since) >= CRASH_STABLE_AFTER => {
                        self.broker_consecutive = 0;
                    }
                    _ => {}
                }
                BrokerObs::Running
            }
            Some(true) => BrokerObs::Starting,
            Some(false) | None => {
                if broker_alive == Some(false) {
                    // The leader is gone. A surviving group member (e.g. a
                    // python child that outlived a crashed `uv`) must not
                    // keep holding the broker port.
                    if let Some(b) = self.broker.as_ref() {
                        if group_alive(b.pid) {
                            signal_group(b.pid, libc::SIGKILL);
                        }
                    }
                    if !self.broker_stopping {
                        // It died without being asked to: a crash.
                        self.broker_failures.push(now);
                        self.broker_failures
                            .retain(|t| now.saturating_duration_since(*t) <= CRASHLOOP_WINDOW);
                        self.broker_consecutive += 1;
                        self.broker_restart_at =
                            Some(now + plan::restart_backoff(self.broker_consecutive));
                    }
                }
                self.broker = None;
                self.broker_stopping = false;
                self.broker_up_since = None;
                // The preferred port (always 9000, regardless of any
                // fallback already chosen) is checked directly: a zakuro
                // broker the agent doesn't own on 9000 is `Unmanaged`
                // whatever fallback port a previous run may have used.
                let preferred = self.spec.preferred_broker_port;
                // Computed here, after any crash just detected above has
                // already updated `broker_restart_at` for this tick: gating
                // the port probe on a stale, pre-crash value wouldn't cause
                // an immediate restart — the plan below recomputes
                // `broker_restart_ready` fresh from the updated
                // `broker_restart_at` (~809), so backoff is still honoured.
                // The only cost of using the stale value here would be one
                // wasted port probe this tick.
                let restart_ready = self.broker_restart_at.is_none_or(|t| now >= t);
                if api.is_broker_at(preferred) {
                    // A connect, not a listen: safe to do on every tick,
                    // whatever `sharing` is, and needed to report
                    // `unmanaged_broker` even while paused.
                    BrokerObs::Unmanaged
                } else if sharing && restart_ready {
                    // Only probed/chosen on a tick that is actually about to
                    // plan `StartBroker` (review N3): `choose_broker_port`'s
                    // `is_free` binds real sockets (`HttpBrokerApi::port_free`,
                    // `os_assigned_port`), and doing that every 2 s while
                    // idle -- sharing off, or the broker still in crash
                    // backoff -- can raise the macOS "accept incoming
                    // connections?" prompt long before sharing is ever
                    // turned on. `apply`'s `Action::StartBroker` arm is the
                    // only place that logs the fallback, so this stays
                    // "once per start" too.
                    match choose_broker_port(
                        preferred,
                        BROKER_FALLBACK_START,
                        BROKER_FALLBACK_END,
                        |p| api.port_free(p),
                        || api.os_assigned_port(),
                    ) {
                        Some(p) => {
                            self.broker_port = p;
                            BrokerObs::Stopped
                        }
                        None => BrokerObs::PortBlocked,
                    }
                } else {
                    // Idle: nothing is going to start this tick, so nothing
                    // is bound. `broker_port` is left as whatever it was
                    // (the preferred port initially, or wherever the last
                    // real start landed) -- irrelevant until the next tick
                    // that actually probes, which always tries the
                    // preferred port first regardless of this stale value.
                    BrokerObs::Stopped
                }
            }
        };
        self.last_broker = broker;
        match broker {
            BrokerObs::Running => match api.workers() {
                Ok(list) => {
                    if self.workers_err_streak {
                        eprintln!("  [AGENT] GET /workers: recovered");
                        self.workers_err_streak = false;
                    }
                    self.last_workers = list;
                    self.workers_fresh = true;
                }
                Err(e) => {
                    if !self.workers_err_streak {
                        eprintln!("  [AGENT] GET /workers: {e}");
                        self.workers_err_streak = true;
                    }
                    self.workers_fresh = false;
                    // `last_workers` is left as-is: see its doc comment.
                }
            },
            BrokerObs::Stopped | BrokerObs::Unmanaged | BrokerObs::PortBlocked => {
                // The broker is confirmed gone (or not ours): there really
                // are no workers now, which is a fact, not an unknown.
                self.last_workers = vec![];
                self.workers_fresh = true;
                self.workers_err_streak = false;
            }
            BrokerObs::Starting | BrokerObs::Stopping => {
                // The leader is there but `/health` hasn't answered yet (or
                // is flushing on the way down): keep the last good list —
                // see `last_workers`'s doc comment — but this tick is not fresh.
                self.workers_fresh = false;
            }
        }

        let broker_restart_ready = self.broker_restart_at.is_none_or(|t| now >= t);

        let fresh = self.workers_fresh;
        let Self {
            slots,
            last_workers,
            spec,
            ..
        } = self;
        let mut workers = Vec::with_capacity(slots.len());
        for (&i, slot) in slots.iter_mut() {
            let running = slot.proc_.as_mut().is_some_and(|p| p.alive());
            if !running {
                if let Some(p) = slot.proc_.take() {
                    // The leader is gone. A surviving group member must not
                    // keep holding port `base_worker_port + i`.
                    if group_alive(p.pid) {
                        signal_group(p.pid, libc::SIGKILL);
                    }
                    if slot.draining_since.is_none() {
                        // It died without being asked to: a crash.
                        slot.failures.push(now);
                        slot.failures
                            .retain(|t| now.saturating_duration_since(*t) <= CRASHLOOP_WINDOW);
                        slot.consecutive += 1;
                        slot.restart_at = Some(now + plan::restart_backoff(slot.consecutive));
                        slot.up_since = None;
                    }
                }
            } else {
                // A slot that has run stably for a while drops back to a
                // fresh backoff (see `Slot::up_since`'s doc comment).
                match slot.up_since {
                    None => slot.up_since = Some(now),
                    Some(since) if now.saturating_duration_since(since) >= CRASH_STABLE_AFTER => {
                        slot.consecutive = 0;
                    }
                    _ => {}
                }
            }
            let name = crate::up::worker_name(&spec.fp8, i as usize);
            let seen = last_workers.iter().find(|w| w.name == name);
            // A draining worker on a non-fresh tick must never look idle: it
            // reports at least 1 active request (never 0) until a fresh
            // observation says otherwise, or its drain has expired.
            let active_requests = if slot.draining_since.is_some() && !fresh {
                seen.map_or(1, |w| w.active_requests.max(1))
            } else {
                seen.map_or(0, |w| w.active_requests)
            };
            workers.push(WorkerObs {
                index: i,
                running,
                draining: slot.draining_since.is_some(),
                active_requests,
                drain_expired: slot
                    .draining_since
                    .is_some_and(|t| now.saturating_duration_since(t) >= spec.drain_timeout),
                restart_ready: slot.restart_at.is_none_or(|t| now >= t),
            });
        }
        Observed {
            broker,
            workers,
            prerequisites_ok,
            broker_restart_ready,
        }
    }

    fn spawn_worker(&mut self, i: u32, now: Instant) -> std::io::Result<()> {
        let port = self.spec.base_worker_port + i as u16;
        let name = self.worker_name(i);
        let argv = render_argv(&self.spec.worker_template, port, &name);
        let mut env = self.spec.worker_env.clone();
        env.push(("ZAKURO_AGENT_CHILD_PORT".to_string(), port.to_string()));
        env.push(("ZAKURO_AGENT_CHILD_NAME".to_string(), name));
        let child = spawn_in_group(
            &argv,
            self.spec.worker_cwd.as_deref(),
            &env,
            &self.worker_log(i),
        )?;
        let slot = self.slots.entry(i).or_default();
        slot.proc_ = Some(Proc::spawned(child));
        slot.draining_since = None;
        slot.up_since = Some(now);
        Ok(())
    }

    /// The broker's id for slot `i`, from the last *good* `GET /workers`
    /// observation (see `last_workers`'s doc comment).
    fn broker_id(&self, i: u32) -> Option<String> {
        let name = self.worker_name(i);
        self.last_workers
            .iter()
            .find(|w| w.name == name)
            .map(|w| w.id.clone())
    }

    fn apply(
        &mut self,
        actions: &[Action],
        observed: &Observed,
        api: &dyn BrokerApi,
        now: Instant,
    ) {
        for action in actions {
            match *action {
                Action::StartBroker => {
                    // Logged here, not where the port is chosen (`observe`):
                    // this is the one place that actually starts something,
                    // so the line prints exactly once per start, never on a
                    // tick where sharing is off or the broker is still in
                    // crash backoff.
                    if self.broker_port != self.spec.preferred_broker_port {
                        eprintln!(
                            "  [AGENT] broker: 127.0.0.1:{} is held by another program; starting the broker on 127.0.0.1:{}",
                            self.spec.preferred_broker_port, self.broker_port
                        );
                    }
                    let log = self.broker_log();
                    let argv = render_argv(&self.spec.broker_argv, self.broker_port, "");
                    let mut env = self.spec.broker_env.clone();
                    env.push((
                        "ZAKURO_AGENT_CHILD_PORT".to_string(),
                        self.broker_port.to_string(),
                    ));
                    match spawn_in_group(&argv, None, &env, &log) {
                        Ok(c) => {
                            self.broker = Some(Proc::spawned(c));
                            self.broker_up_since = Some(now);
                        }
                        Err(e) => eprintln!("  [AGENT] could not start the broker: {e}"),
                    }
                }
                Action::StopBroker => {
                    if let Some(b) = self.broker.take() {
                        self.broker_stopping = true;
                        // `Some(b)` back means shutdown interrupted the wait:
                        // give the broker back so `terminate_all` finishes
                        // stopping it with its own full grace period, rather
                        // than this tick cutting the stop short.
                        let leftover = terminate(b, self.spec.broker_stop_wait, &self.stop);
                        self.broker_stopping = false;
                        self.broker_up_since = None;
                        self.broker = leftover;
                    }
                }
                Action::StartWorker(i) | Action::RestartWorker(i) => {
                    if let Err(e) = self.spawn_worker(i, now) {
                        eprintln!("  [AGENT] could not start worker {i}: {e}");
                    }
                }
                Action::DrainWorker(i) => {
                    if let Some(id) = self.broker_id(i) {
                        if let Err(e) = api.drain(&id) {
                            eprintln!("  [AGENT] drain {id}: {e}");
                        }
                    }
                    self.slots.entry(i).or_default().draining_since = Some(now);
                }
                Action::KillWorker(i) => {
                    if observed
                        .workers
                        .iter()
                        .any(|w| w.index == i && w.drain_expired && w.active_requests > 0)
                    {
                        self.last_drain_timeout = Some(now);
                    }
                    if let Some(mut slot) = self.slots.remove(&i) {
                        if let Some(p) = slot.proc_.take() {
                            if let Some(p) = terminate(p, self.spec.kill_grace, &self.stop) {
                                // Shutdown interrupted the kill: give the
                                // slot back so `terminate_all` handles it
                                // with its own grace period, instead of
                                // deleting it from the broker and
                                // forgetting it below as if it were already
                                // gone.
                                slot.proc_ = Some(p);
                                self.slots.insert(i, slot);
                                continue;
                            }
                        }
                    }
                    if let Some(id) = self.broker_id(i) {
                        if let Err(e) = api.delete(&id) {
                            eprintln!("  [AGENT] delete {id}: {e}");
                        }
                    }
                    // Forget this worker's entry in the retained list too:
                    // otherwise a later StartWorker(i) during a `GET
                    // /workers` error streak would inherit the dead worker's
                    // stale stats and id by name.
                    let name = self.worker_name(i);
                    self.last_workers.retain(|w| w.name != name);
                }
            }
        }
    }

    /// A slot's first `POST /workers/{id}/drain` can miss: the worker hadn't
    /// registered yet (no id to send), the POST failed, or the broker
    /// restarted and re-registered it as healthy. While a fresh observation
    /// shows the broker not treating a draining slot as `draining`, send the
    /// drain again, so the broker stops routing jobs to a worker the agent is
    /// about to stop. Slots this tick kills are skipped, and a tick that
    /// didn't just read `GET /workers` sends nothing: its list is stale.
    fn resend_drains(&self, api: &dyn BrokerApi, actions: &[Action]) {
        if !self.workers_fresh || self.last_broker != BrokerObs::Running {
            return;
        }
        for (&i, slot) in &self.slots {
            if slot.draining_since.is_none() || actions.contains(&Action::KillWorker(i)) {
                continue;
            }
            let name = self.worker_name(i);
            let Some(w) = self.last_workers.iter().find(|w| w.name == name) else {
                continue;
            };
            if w.status != "draining" {
                if let Err(e) = api.drain(&w.id) {
                    eprintln!("  [AGENT] drain {} (again): {e}", w.id);
                }
            }
        }
    }

    /// One reconcile tick: observe, decide (`plan_actions`), act.
    pub fn reconcile(
        &mut self,
        desired: &Desired,
        api: &dyn BrokerApi,
        prerequisites_ok: bool,
        now: Instant,
    ) -> Vec<Action> {
        let observed = self.observe(api, prerequisites_ok, desired.sharing, now);
        let actions = plan::plan_actions(desired, &observed);
        self.resend_drains(api, &actions);
        self.apply(&actions, &observed, api, now);
        let logs: Vec<PathBuf> = self.slots.keys().map(|&i| self.worker_log(i)).collect();
        for log in logs {
            let _ = truncate_if_large(&log, WORKER_LOG_MAX_BYTES);
        }
        let _ = truncate_if_large(&self.broker_log(), WORKER_LOG_MAX_BYTES);
        actions
    }

    pub fn problems(&self, now: Instant) -> Vec<Problem> {
        let mut out = Vec::new();
        match self.last_broker {
            BrokerObs::Unmanaged => out.push(Problem::new("unmanaged_broker")),
            BrokerObs::PortBlocked => out.push(Problem::new("port_in_use")),
            _ => {}
        }
        if plan::is_crashloop(&self.broker_failures, now) {
            let log = self.broker_log();
            let mut p = Problem::new("broker_crashloop");
            if log.exists() {
                p = p.with_detail(tail_lines(&log, 20));
            }
            out.push(p);
        }
        for (&i, slot) in &self.slots {
            if plan::is_crashloop(&slot.failures, now) {
                out.push(
                    Problem::new("worker_crashloop")
                        .with_detail(tail_lines(&self.worker_log(i), 20)),
                );
            }
        }
        if self
            .last_drain_timeout
            .is_some_and(|t| now.saturating_duration_since(t) < DRAIN_TIMEOUT_SHOWN_FOR)
        {
            out.push(Problem::new("drain_timeout"));
        }
        out
    }

    pub fn local_view(&self, desired: &Desired, max: u32) -> LocalView {
        let names: HashSet<String> = self.slots.keys().map(|&i| self.worker_name(i)).collect();
        let mine: Vec<&BrokerWorker> = self
            .last_workers
            .iter()
            .filter(|w| names.contains(&w.name))
            .collect();
        let count = |f: &dyn Fn(&Slot) -> bool| self.slots.values().filter(|s| f(s)).count() as u32;
        LocalView {
            sharing: desired.sharing,
            broker: broker_label(self.last_broker).to_string(),
            workers: WorkerCounts {
                desired: desired.workers,
                running: count(&|s| s.proc_.is_some() && s.draining_since.is_none()),
                busy: mine.iter().filter(|w| w.active_requests > 0).count() as u32,
                draining: count(&|s| s.proc_.is_some() && s.draining_since.is_some()),
                max,
            },
            requests: Requests {
                last_5h: mine.iter().map(|w| w.requests_5h).sum(),
                last_1w: mine.iter().map(|w| w.requests_1w).sum(),
            },
        }
    }

    /// Agent SIGTERM (spec §6.5): SIGTERM every child group, wait up to `wait`,
    /// SIGKILL what is left. No drain. There is no `Drop` equivalent of this:
    /// a `Supervisor` that is merely dropped (e.g. across an agent restart)
    /// must leave its children running so they can later be `adopt`ed.
    pub fn terminate_all(&mut self, wait: Duration) {
        let mut procs: Vec<Proc> = self
            .slots
            .values_mut()
            .filter_map(|s| s.proc_.take())
            .collect();
        procs.extend(self.broker.take());
        for p in &procs {
            signal_group(p.pid, libc::SIGTERM);
        }
        // Once a group is confirmed gone we stop probing (and would never
        // signal) that pid again: a freed pgid can be reused by an unrelated
        // process while another group in this same call is still exiting.
        let mut gone = vec![false; procs.len()];
        let deadline = Instant::now() + wait;
        while Instant::now() < deadline && gone.iter().any(|g| !g) {
            for (p, g) in procs.iter_mut().zip(gone.iter_mut()) {
                if *g {
                    continue;
                }
                if let Some(c) = p.child.as_mut() {
                    let _ = c.try_wait();
                }
                if !group_alive(p.pid) {
                    *g = true;
                }
            }
            if gone.iter().any(|g| !g) {
                std::thread::sleep(Duration::from_millis(100));
            }
        }
        for (p, g) in procs.iter_mut().zip(gone.iter_mut()) {
            if !*g && group_alive(p.pid) {
                signal_group(p.pid, libc::SIGKILL);
            }
            if let Some(c) = p.child.as_mut() {
                let _ = c.wait();
            }
        }
        self.slots.clear();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::agent::files::tests::tmp;
    use std::cell::{Cell, RefCell};
    use std::ops::{Deref, DerefMut};

    #[derive(Default)]
    struct FakeBroker {
        up: Cell<bool>,
        /// Overrides `is_broker()` independently of `up`, to simulate a
        /// `/health` timeout while the broker's process stays as `up` says.
        health_override: Cell<Option<bool>>,
        workers: RefCell<Vec<BrokerWorker>>,
        /// When set, `workers()` returns `Err` instead of the list below,
        /// simulating a `GET /workers` timeout or a non-2xx response.
        workers_err: Cell<bool>,
        /// When set, `drain()` records the attempt but answers `Err`, as a
        /// broker that times out or rejects the POST would.
        drain_fails: Cell<bool>,
        drained: RefCell<Vec<String>>,
        deleted: RefCell<Vec<String>>,
        /// When set, `is_broker_at(p)` is true only for this port: simulates
        /// a foreign zakuro broker holding the preferred port.
        foreign_broker_port: Cell<Option<u16>>,
        /// Ports considered occupied (by anything) for the fallback-port
        /// search; empty by default, so `port_free` says yes to everything
        /// and the very first `StartBroker` always lands on the preferred
        /// port, matching every test that doesn't care about fallback.
        blocked_ports: RefCell<std::collections::HashSet<u16>>,
        /// `os_assigned_port()`'s canned answer.
        os_pick: Cell<Option<u16>>,
        /// How many times `port_free`/`os_assigned_port` were called: the
        /// real implementations do socket I/O (a wildcard bind), so a test
        /// asserting an idle tick never probes counts these instead of
        /// mocking a real listener (review N3).
        port_free_calls: RefCell<u32>,
        os_assigned_port_calls: RefCell<u32>,
    }
    impl BrokerApi for FakeBroker {
        fn is_broker(&self) -> bool {
            self.health_override.get().unwrap_or_else(|| self.up.get())
        }
        fn is_broker_at(&self, port: u16) -> bool {
            self.foreign_broker_port.get() == Some(port)
        }
        fn port_free(&self, port: u16) -> bool {
            *self.port_free_calls.borrow_mut() += 1;
            !self.blocked_ports.borrow().contains(&port)
        }
        fn os_assigned_port(&self) -> Option<u16> {
            *self.os_assigned_port_calls.borrow_mut() += 1;
            self.os_pick.get()
        }
        fn workers(&self) -> Result<Vec<BrokerWorker>, String> {
            if self.workers_err.get() {
                return Err("simulated GET /workers failure".into());
            }
            Ok(self.workers.borrow().clone())
        }
        fn drain(&self, id: &str) -> Result<(), String> {
            self.drained.borrow_mut().push(id.into());
            if self.drain_fails.get() {
                return Err("simulated POST /drain failure".into());
            }
            Ok(())
        }
        fn delete(&self, id: &str) -> Result<(), String> {
            self.deleted.borrow_mut().push(id.into());
            Ok(())
        }
    }

    fn sh(script: &str) -> Vec<String> {
        vec!["sh".into(), "-c".into(), script.into()]
    }

    fn spec(dir: &Path, worker: Vec<String>) -> SpawnSpec {
        SpawnSpec {
            broker_argv: sh("exec sleep 60"),
            broker_env: vec![],
            worker_template: worker,
            worker_env: vec![],
            worker_cwd: None,
            base_worker_port: 3960,
            preferred_broker_port: 9000,
            fp8: "testfp00".into(),
            log_dir: dir.to_path_buf(),
            kill_grace: Duration::from_secs(2),
            broker_stop_wait: Duration::from_secs(2),
            drain_timeout: Duration::from_secs(300),
        }
    }

    fn bw_n(id: &str, name: &str, active_requests: u32) -> BrokerWorker {
        BrokerWorker {
            id: id.into(),
            name: name.into(),
            status: "healthy".into(),
            active_requests,
            requests_5h: 1,
            requests_1w: 2,
        }
    }

    fn bw(id: &str, name: &str) -> BrokerWorker {
        bw_n(id, name, 0)
    }

    /// Polls (for up to 5 s) until `pred()` is true. Used instead of a fixed
    /// sleep so CI load can't turn a slow process exit into a flaky test.
    fn wait_until(mut pred: impl FnMut() -> bool) -> bool {
        for _ in 0..100 {
            if pred() {
                return true;
            }
            std::thread::sleep(Duration::from_millis(50));
        }
        pred()
    }

    /// Polls (up to 5 s) until the supervisor's own bookkeeping has reaped
    /// worker slot `i`'s process and sees it as no longer running. A raw
    /// `kill(pid, 0)` (as `is_alive` uses) can't tell an exited-but-unreaped
    /// zombie from a live process on macOS, but `Proc::alive`'s `try_wait`
    /// can, since it is the process's real parent.
    fn wait_until_worker_exited(sup: &mut Supervisor, i: u32) -> bool {
        wait_until(|| {
            sup.slots
                .get_mut(&i)
                .and_then(|s| s.proc_.as_mut())
                .map(|p| !p.alive())
                .unwrap_or(true)
        })
    }

    /// As `wait_until_worker_exited`, for the broker.
    fn wait_until_broker_exited(sup: &mut Supervisor) -> bool {
        wait_until(|| sup.broker.as_mut().map(|p| !p.alive()).unwrap_or(true))
    }

    /// Every test that spawns a real process group must reap or kill it, even
    /// on assertion failure, so the suite leaves no orphans (see task rules).
    /// Killing an already-terminated group is a harmless no-op.
    struct PgGuard(u32);
    impl Drop for PgGuard {
        fn drop(&mut self) {
            signal_group(self.0, libc::SIGKILL);
        }
    }

    /// Kills and reaps a raw, ungrouped-through-`Supervisor` `Child` on drop,
    /// even on panic.
    struct ChildGuard(Child);
    impl ChildGuard {
        fn id(&self) -> u32 {
            self.0.id()
        }
    }
    impl Drop for ChildGuard {
        fn drop(&mut self) {
            let pid = self.0.id();
            let leader_reaped = matches!(self.0.try_wait(), Ok(Some(_)));
            // Skip the SIGKILL only once BOTH the leader is already reaped
            // (the success path's own cleanup got there first) AND the
            // whole group is gone: a reaped leader can still have a live
            // group member (e.g. a backgrounded grandchild), and a freed
            // pgid can be reused by an unrelated process once truly empty.
            if leader_reaped && !group_alive(pid) {
                return;
            }
            signal_group(pid, libc::SIGKILL);
            let _ = self.0.wait();
        }
    }

    /// Kills and reaps a bare `Proc` on drop, even on panic. `terminate`
    /// already leaves the process dead in the success path; this is a
    /// backstop for when an assertion fails first.
    struct ProcGuard(Proc);
    impl ProcGuard {
        /// Hand the guarded `Proc` to something that will finish handling it
        /// (like `terminate`, which now takes it by value), without this
        /// guard's `Drop` racing that handoff. Anything that panics before
        /// this is called is still covered by the guard as usual.
        fn into_inner(self) -> Proc {
            let this = std::mem::ManuallyDrop::new(self);
            unsafe { std::ptr::read(&this.0) }
        }
    }
    impl Drop for ProcGuard {
        fn drop(&mut self) {
            let leader_reaped = match self.0.child.as_mut() {
                Some(c) => matches!(c.try_wait(), Ok(Some(_))),
                None => !is_alive(self.0.pid),
            };
            // See `ChildGuard::drop`: a reaped leader can still have a live
            // group member, so only skip once the whole group is gone too.
            if leader_reaped && !group_alive(self.0.pid) {
                return;
            }
            signal_group(self.0.pid, libc::SIGKILL);
            if let Some(c) = self.0.child.as_mut() {
                let _ = c.wait();
            }
        }
    }

    /// Kills and reaps a pid we no longer hold a `Child`/`Proc` handle for
    /// (e.g. after the `Supervisor` that owned it has already been dropped).
    struct PidReapGuard(u32);
    impl Drop for PidReapGuard {
        fn drop(&mut self) {
            signal_group(self.0, libc::SIGKILL);
            unsafe {
                libc::waitpid(self.0 as libc::pid_t, std::ptr::null_mut(), 0);
            }
        }
    }

    /// A `Supervisor` is not itself kill-on-drop (spec §6.5: children must
    /// survive an agent restart so they can be `adopt`ed, and the broker
    /// needs its full `broker_stop_wait` to flush). Every test uses this
    /// guard instead, so a test that spawns real children still cleans them
    /// up even when an assertion fails first.
    struct SupGuard(Supervisor);
    impl Deref for SupGuard {
        type Target = Supervisor;
        fn deref(&self) -> &Supervisor {
            &self.0
        }
    }
    impl DerefMut for SupGuard {
        fn deref_mut(&mut self) -> &mut Supervisor {
            &mut self.0
        }
    }
    impl Drop for SupGuard {
        fn drop(&mut self) {
            self.0.terminate_all(Duration::from_secs(2));
        }
    }

    #[test]
    fn dropping_a_plain_supervisor_leaves_its_child_running() {
        let dir = tmp("plain-drop");
        let pid = {
            let mut sup = Supervisor::new(spec(&dir, sh("exec sleep 60")));
            sup.spawn_worker(0, Instant::now()).unwrap();
            sup.children().workers["0"].pid
        }; // a plain (unguarded) Supervisor drops here
        let _reap = PidReapGuard(pid); // clean up regardless of the assertion below
        assert!(
            is_alive(pid),
            "a plain Supervisor must not kill its children on drop"
        );
    }

    #[test]
    fn reconcile_starts_scales_drains_and_pauses_real_processes() {
        use Action::*;
        let dir = tmp("sup");
        let api = FakeBroker::default();
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        let t = Instant::now();
        let on2 = Desired {
            sharing: true,
            workers: 2,
        };

        assert_eq!(sup.reconcile(&on2, &api, true, t), vec![StartBroker]);
        api.up.set(true); // the broker now answers /health
        assert_eq!(
            sup.reconcile(&on2, &api, true, t),
            vec![StartWorker(0), StartWorker(1)]
        );
        *api.workers.borrow_mut() = vec![bw("id0", "testfp00-w0"), bw("id1", "testfp00-w1")];
        assert_eq!(sup.reconcile(&on2, &api, true, t), vec![]);
        let view = sup.local_view(&on2, 5);
        assert_eq!(
            (
                view.workers.running,
                view.workers.desired,
                view.broker.as_str()
            ),
            (2, 2, "running")
        );
        assert_eq!(
            view.requests,
            Requests {
                last_5h: 2,
                last_1w: 4
            }
        );
        assert_eq!(sup.children().workers.len(), 2);
        assert!(sup.children().broker.is_some());

        let on1 = Desired {
            sharing: true,
            workers: 1,
        };
        assert_eq!(sup.reconcile(&on1, &api, true, t), vec![DrainWorker(1)]);
        assert_eq!(*api.drained.borrow(), vec!["id1".to_string()]);
        assert_eq!(sup.reconcile(&on1, &api, true, t), vec![KillWorker(1)]);
        assert_eq!(*api.deleted.borrow(), vec!["id1".to_string()]);
        assert_eq!(
            sup.children().workers.keys().cloned().collect::<Vec<_>>(),
            vec!["0".to_string()]
        );

        let off = Desired {
            sharing: false,
            workers: 1,
        };
        assert_eq!(sup.reconcile(&off, &api, true, t), vec![DrainWorker(0)]);
        assert_eq!(sup.reconcile(&off, &api, true, t), vec![KillWorker(0)]);
        assert_eq!(sup.reconcile(&off, &api, true, t), vec![StopBroker]);
        assert!(sup.children().broker.is_none());
        assert!(sup.children().workers.is_empty());
    }

    #[test]
    fn a_crashing_worker_backs_off_and_is_flagged_as_a_crashloop() {
        use Action::*;
        let dir = tmp("crash");
        let api = FakeBroker::default();
        api.up.set(true);
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("echo boom; exit 1"))));
        let t = Instant::now();
        let at = |ms: u64| t + Duration::from_millis(ms);
        let on1 = Desired {
            sharing: true,
            workers: 1,
        };
        // Adopt a fake running broker so the plan goes straight to workers.
        let broker_child =
            spawn_in_group(&sh("exec sleep 60"), None, &[], &dir.join("b.log")).unwrap();
        let _guard = PgGuard(broker_child.id()); // belt-and-braces: terminate_all reaps it below
        sup.broker = Some(Proc::spawned(broker_child));

        assert_eq!(sup.reconcile(&on1, &api, true, at(0)), vec![StartWorker(0)]);
        assert!(
            wait_until_worker_exited(&mut sup, 0),
            "worker slot 0 did not exit in time"
        );
        assert_eq!(
            sup.reconcile(&on1, &api, true, at(100)),
            vec![],
            "1st failure: 1 s backoff"
        );
        assert_eq!(
            sup.reconcile(&on1, &api, true, at(1200)),
            vec![RestartWorker(0)]
        );
        assert!(
            wait_until_worker_exited(&mut sup, 0),
            "worker slot 0 did not exit in time"
        );
        assert_eq!(
            sup.reconcile(&on1, &api, true, at(1300)),
            vec![],
            "2nd failure: 2 s backoff"
        );
        assert_eq!(
            sup.reconcile(&on1, &api, true, at(3400)),
            vec![RestartWorker(0)]
        );
        assert!(
            wait_until_worker_exited(&mut sup, 0),
            "worker slot 0 did not exit in time"
        );
        sup.reconcile(&on1, &api, true, at(3500));

        let problems = sup.problems(at(3500));
        let crash = problems
            .iter()
            .find(|p| p.code == "worker_crashloop")
            .expect("3 failures in 5 min");
        assert!(
            crash.detail.as_deref().unwrap_or("").contains("boom"),
            "detail carries the log tail"
        );
    }

    #[test]
    fn a_crashing_worker_that_stabilizes_resets_its_backoff() {
        use Action::*;
        let dir = tmp("worker-stable-reset");
        let api = FakeBroker::default();
        api.up.set(true);
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        let t = Instant::now();
        let at = |ms: u64| t + Duration::from_millis(ms);
        let on1 = Desired {
            sharing: true,
            workers: 1,
        };
        // Adopt a fake running broker so the plan goes straight to workers.
        let broker_child =
            spawn_in_group(&sh("exec sleep 60"), None, &[], &dir.join("b.log")).unwrap();
        let _guard = PgGuard(broker_child.id());
        sup.broker = Some(Proc::spawned(broker_child));

        // 1. Crash a child once.
        assert_eq!(sup.reconcile(&on1, &api, true, at(0)), vec![StartWorker(0)]);
        let pid0 = sup.children().workers["0"].pid;
        signal_group(pid0, libc::SIGKILL);
        assert!(
            wait_until_worker_exited(&mut sup, 0),
            "worker slot 0 did not exit in time"
        );
        assert_eq!(
            sup.reconcile(&on1, &api, true, at(100)),
            vec![],
            "1st failure: 1 s backoff"
        );

        // 2. Let it restart.
        assert_eq!(
            sup.reconcile(&on1, &api, true, at(1200)),
            vec![RestartWorker(0)]
        );

        // 3. Tick at +301 s while it's alive: this crosses CRASH_STABLE_AFTER
        // and resets `consecutive` back to 0.
        assert_eq!(sup.reconcile(&on1, &api, true, at(1200 + 301_000)), vec![]);

        // 4. SIGKILL the pid from `sup.children()`: a second, independent crash.
        let pid1 = sup.children().workers["0"].pid;
        signal_group(pid1, libc::SIGKILL);
        assert!(
            wait_until_worker_exited(&mut sup, 0),
            "worker slot 0 did not exit in time"
        );
        let t2 = at(1200 + 301_000 + 100);
        assert_eq!(
            sup.reconcile(&on1, &api, true, t2),
            vec![],
            "post-reset failure: backoff restarts at 1 s"
        );

        // 5. The next restart backoff must be 1 s, not 2 s.
        assert_eq!(
            sup.reconcile(&on1, &api, true, t2 + Duration::from_millis(900)),
            vec![],
            "not ready yet at 0.9 s"
        );
        assert_eq!(
            sup.reconcile(&on1, &api, true, t2 + Duration::from_millis(1100)),
            vec![RestartWorker(0)],
            "ready at 1.1 s: proves the backoff reset to 1 s, not 2 s"
        );
    }

    #[test]
    fn killpg_reaches_the_whole_process_group() {
        let dir = tmp("pg");
        let log = dir.join("pg.log");
        let child = spawn_in_group(&sh("sleep 30 & echo $!; wait"), None, &[], &log).unwrap();
        let p = ProcGuard(Proc::spawned(child));
        let pid = p.0.pid;
        let grandchild = (0..40)
            .find_map(|_| {
                std::thread::sleep(Duration::from_millis(50));
                std::fs::read_to_string(&log)
                    .ok()?
                    .trim()
                    .parse::<u32>()
                    .ok()
            })
            .expect("sh printed the sleep pid");
        assert!(is_alive(grandchild));
        let leftover = terminate(
            p.into_inner(),
            Duration::from_secs(2),
            &AtomicBool::new(false),
        );
        assert!(
            leftover.is_none(),
            "an uninterrupted terminate always finishes the group off"
        );
        assert!(!is_alive(pid));
        let gone = wait_until(|| !is_alive(grandchild));
        assert!(
            gone,
            "SIGTERM to the group must reach the grandchild, as python under uv"
        );
    }

    #[test]
    fn a_group_member_that_ignores_term_is_sigkilled() {
        let dir = tmp("ignore-term");
        let log = dir.join("ignore.log");
        let child = spawn_in_group(
            &sh("sh -c 'trap \"\" TERM; echo $$; exec sleep 30' & wait"),
            None,
            &[],
            &log,
        )
        .unwrap();
        let p = ProcGuard(Proc::spawned(child));
        let pid = p.0.pid;
        let grandchild = (0..40)
            .find_map(|_| {
                std::thread::sleep(Duration::from_millis(50));
                std::fs::read_to_string(&log)
                    .ok()?
                    .trim()
                    .parse::<u32>()
                    .ok()
            })
            .expect("sh printed the sleep pid");
        assert!(is_alive(grandchild));
        // A short grace forces the SIGKILL escalation path.
        let leftover = terminate(
            p.into_inner(),
            Duration::from_millis(300),
            &AtomicBool::new(false),
        );
        assert!(
            leftover.is_none(),
            "an uninterrupted terminate always finishes the group off"
        );
        assert!(!is_alive(pid));
        let gone = wait_until(|| !is_alive(grandchild));
        assert!(gone, "SIGKILL must reach a group member that ignores TERM");
    }

    /// The stop flag must never turn an in-flight graceful stop into an
    /// instant kill: when it is set mid-wait, `terminate` hands the
    /// still-alive group back untouched, and only `terminate_all` (with its
    /// own full grace period) actually finishes it off.
    #[test]
    fn a_stop_flag_set_mid_wait_hands_the_group_back_instead_of_sigkilling_it() {
        let dir = tmp("interrupt-terminate");
        let log = dir.join("ignore.log");
        let child = spawn_in_group(
            &sh("sh -c 'trap \"\" TERM; echo $$; exec sleep 30' & wait"),
            None,
            &[],
            &log,
        )
        .unwrap();
        let p = ProcGuard(Proc::spawned(child));
        let pid = p.0.pid;
        let grandchild = (0..40)
            .find_map(|_| {
                std::thread::sleep(Duration::from_millis(50));
                std::fs::read_to_string(&log)
                    .ok()?
                    .trim()
                    .parse::<u32>()
                    .ok()
            })
            .expect("sh printed the sleep pid");
        assert!(is_alive(grandchild));

        let stop = Arc::new(AtomicBool::new(false));
        let setter = stop.clone();
        std::thread::spawn(move || {
            std::thread::sleep(Duration::from_millis(50));
            setter.store(true, Ordering::Relaxed);
        });

        let began = Instant::now();
        // A grace far longer than the return-time bound just below, so a
        // slow return can only be explained by the flag being ignored.
        let leftover = terminate(p.into_inner(), Duration::from_secs(30), &stop);
        let elapsed = began.elapsed();
        // Re-guard the handed-back `Proc` immediately, before any assertion
        // below gets a chance to panic and leak the still-alive group: `Proc`
        // itself has no `Drop`, so nothing else here kills it on a failure.
        let mut leftover = leftover.map(ProcGuard);
        assert!(
            elapsed < Duration::from_secs(2),
            "terminate took {elapsed:?} to notice the stop flag \
             (still far below the 30 s grace this must interrupt)"
        );
        let leftover = leftover
            .take()
            .expect("an interrupted terminate must hand the group back, not kill it");
        assert!(
            group_alive(pid),
            "an interrupted terminate must not SIGKILL the group \
             (the leader itself may already have exited on the plain SIGTERM \
             it never trapped; the group member that traps it must survive)"
        );
        assert!(
            is_alive(grandchild),
            "an interrupted terminate must not touch a group member either"
        );

        // terminate_all is what actually finishes it off, with its own
        // grace; SupGuard's own Drop is a second backstop in case anything
        // between here and the end of the test panics first.
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exit 0"))));
        sup.broker = Some(leftover.into_inner());
        sup.terminate_all(Duration::from_secs(2));
        // SIGKILL is asynchronous: poll rather than assert instantly, exactly
        // as `terminate_all_also_reaches_a_group_member_that_ignores_term`
        // does below.
        assert!(
            wait_until(|| !group_alive(pid)),
            "terminate_all must still finish off a group terminate() handed back"
        );
        let gone = wait_until(|| !is_alive(grandchild));
        assert!(
            gone,
            "terminate_all must still finish off a group terminate() handed back"
        );
    }

    #[test]
    fn terminate_all_also_reaches_a_group_member_that_ignores_term() {
        use Action::*;
        let dir = tmp("ignore-term-all");
        let api = FakeBroker::default();
        api.up.set(true);
        let mut sup = SupGuard(Supervisor::new(spec(
            &dir,
            sh("sh -c 'trap \"\" TERM; echo $$; exec sleep 30' & wait"),
        )));
        let on1 = Desired {
            sharing: true,
            workers: 1,
        };
        // Adopt a fake running broker so the plan goes straight to workers.
        let broker_child =
            spawn_in_group(&sh("exec sleep 60"), None, &[], &dir.join("b.log")).unwrap();
        let _guard = PgGuard(broker_child.id());
        sup.broker = Some(Proc::spawned(broker_child));

        assert_eq!(
            sup.reconcile(&on1, &api, true, Instant::now()),
            vec![StartWorker(0)]
        );
        let log = sup.worker_log(0);
        let grandchild = (0..40)
            .find_map(|_| {
                std::thread::sleep(Duration::from_millis(50));
                std::fs::read_to_string(&log)
                    .ok()?
                    .trim()
                    .parse::<u32>()
                    .ok()
            })
            .expect("worker script printed the sleep pid");
        assert!(is_alive(grandchild));
        sup.terminate_all(Duration::from_millis(300));
        let gone = wait_until(|| !is_alive(grandchild));
        assert!(
            gone,
            "terminate_all's SIGKILL must reach a group member that ignores TERM"
        );
    }

    #[test]
    fn adoption_needs_a_live_pid_with_the_same_start_time() {
        let dir = tmp("adopt");
        let mut child = ChildGuard(
            spawn_in_group(&sh("exec sleep 30"), None, &[], &dir.join("a.log")).unwrap(),
        );
        let pid = child.id();
        let started = proc_start_time(pid).expect("ps/proc reports a start time");
        assert_eq!(
            proc_start_time(pid).as_deref(),
            Some(started.as_str()),
            "stable"
        );

        let mut recorded = Children::default();
        recorded.workers.insert(
            "0".into(),
            ChildRecord {
                pid,
                started_at: started.clone(),
                port: None,
            },
        );
        recorded.workers.insert(
            "1".into(),
            ChildRecord {
                pid,
                started_at: "reused pid".into(),
                port: None,
            },
        );
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        sup.adopt(&recorded, &proc_start_time);
        assert_eq!(
            sup.children().workers.keys().cloned().collect::<Vec<_>>(),
            vec!["0".to_string()]
        );

        sup.terminate_all(Duration::from_secs(2));
        // `wait` blocks until the (already SIGTERMed) process actually exits,
        // so there's no need for a fixed sleep here.
        let _ = child.0.wait();
        assert_eq!(proc_start_time(pid), None, "a dead pid has no start time");
    }

    #[test]
    fn logs_are_truncated_past_the_cap_and_tailed() {
        let dir = tmp("logs");
        std::fs::create_dir_all(&dir).unwrap();
        let log = dir.join("w0.log");
        std::fs::write(&log, "a\nb\nc\nd\n").unwrap();
        assert_eq!(tail_lines(&log, 2), "c\nd");
        assert!(!truncate_if_large(&log, 1024).unwrap());
        assert!(truncate_if_large(&log, 3).unwrap());
        assert_eq!(std::fs::metadata(&log).unwrap().len(), 0);
    }

    #[test]
    fn argv_templates_and_broker_env_follow_the_spec() {
        let t: Vec<String> = ["uv", "--port", "{port}", "--worker-name", "{name}"]
            .iter()
            .map(|s| s.to_string())
            .collect();
        assert_eq!(
            render_argv(&t, 3961, "fp-w1"),
            vec!["uv", "--port", "3961", "--worker-name", "fp-w1"]
        );

        let mut cfg = crate::agent::config::AgentConfig::for_dirs(tmp("spec"));
        cfg.max_workers = 5;
        let s = SpawnSpec::from_config(&cfg);
        assert!(s
            .broker_env
            .contains(&("ZAKURO_SCAN_RANGE".into(), "3960-3984".into())));
        assert!(s
            .broker_env
            .contains(&("ZAKURO_SCAN_INTERVAL".into(), "3".into())));
        assert!(s.broker_env.iter().any(|(k, _)| k == "ZAKURO_P2P"));
        assert!(
            !s.broker_env
                .iter()
                .any(|(k, _)| k == "ZAKURO_AGENT_CHILD_PORT"),
            "the broker's port env is added at spawn time, once the port is chosen"
        );
        // `broker_argv` is a template (`{port}` unrendered) until spawn time,
        // so a fallback port picked at start doesn't require re-deriving the
        // whole SpawnSpec.
        assert_eq!(
            &s.broker_argv[1..],
            &["broker".to_string(), "{port}".to_string()]
        );
        assert_eq!(s.preferred_broker_port, 9000);
        assert!(s
            .worker_env
            .contains(&("ZAKURO_WORKER_TYPE".into(), "zakuro".into())));
        assert!(s
            .worker_env
            .contains(&("UV_NO_PROGRESS".into(), "1".into())));
    }

    #[test]
    fn workers_bind_loopback_unless_the_config_picks_a_host_or_caller_auth() {
        use crate::agent::config::AgentConfig;
        // The value a worker ends up with: the last ZAKURO_HOST wins.
        let host = |cfg: &AgentConfig| {
            SpawnSpec::from_config(cfg)
                .worker_env
                .iter()
                .rev()
                .find(|(k, _)| k == "ZAKURO_HOST")
                .map(|(_, v)| v.clone())
        };
        let plain = AgentConfig::for_dirs(tmp("bind-plain"));
        assert_eq!(host(&plain).as_deref(), Some("127.0.0.1"));

        let mut insecure = AgentConfig::for_dirs(tmp("bind-insecure"));
        insecure
            .child_env
            .push(("ZAKURO_INSECURE_BIND".into(), "1".into()));
        assert_eq!(host(&insecure), None);

        let mut chosen = AgentConfig::for_dirs(tmp("bind-host"));
        chosen
            .child_env
            .push(("ZAKURO_HOST".into(), "0.0.0.0".into()));
        assert_eq!(host(&chosen).as_deref(), Some("0.0.0.0"));
    }

    // --- Broker crash backoff and crashloop (mirrors the worker slot machinery) ---

    #[test]
    fn a_crashing_broker_backs_off_then_restarts() {
        use Action::*;
        let dir = tmp("brk-backoff");
        let api = FakeBroker::default();
        let mut s = spec(&dir, sh("exec sleep 60"));
        s.broker_argv = sh("echo boom; exit 1");
        let mut sup = SupGuard(Supervisor::new(s));
        let t = Instant::now();
        let at = |ms: u64| t + Duration::from_millis(ms);
        let on = Desired {
            sharing: true,
            workers: 0,
        };

        assert_eq!(sup.reconcile(&on, &api, true, at(0)), vec![StartBroker]);
        assert!(
            wait_until_broker_exited(&mut sup),
            "the broker did not exit in time"
        );
        assert_eq!(
            sup.reconcile(&on, &api, true, at(100)),
            vec![],
            "1st failure: 1 s backoff, broker_restart_ready must be false"
        );
        assert_eq!(
            sup.reconcile(&on, &api, true, at(1200)),
            vec![StartBroker],
            "backoff elapsed: broker_restart_ready is true again"
        );
    }

    #[test]
    fn three_broker_crashes_within_five_minutes_raise_a_crashloop_problem() {
        use Action::*;
        let dir = tmp("brk-crashloop");
        let api = FakeBroker::default();
        let mut s = spec(&dir, sh("exec sleep 60"));
        s.broker_argv = sh("echo boom; exit 1");
        let mut sup = SupGuard(Supervisor::new(s));
        let t = Instant::now();
        let at = |ms: u64| t + Duration::from_millis(ms);
        let on = Desired {
            sharing: true,
            workers: 0,
        };

        assert_eq!(sup.reconcile(&on, &api, true, at(0)), vec![StartBroker]);
        assert!(
            wait_until_broker_exited(&mut sup),
            "the broker did not exit in time"
        );
        assert_eq!(
            sup.reconcile(&on, &api, true, at(100)),
            vec![],
            "1st failure: 1 s backoff"
        );
        assert_eq!(sup.reconcile(&on, &api, true, at(1200)), vec![StartBroker]);
        assert!(
            wait_until_broker_exited(&mut sup),
            "the broker did not exit in time"
        );
        assert_eq!(
            sup.reconcile(&on, &api, true, at(1300)),
            vec![],
            "2nd failure: 2 s backoff"
        );
        assert_eq!(sup.reconcile(&on, &api, true, at(3400)), vec![StartBroker]);
        assert!(
            wait_until_broker_exited(&mut sup),
            "the broker did not exit in time"
        );
        sup.reconcile(&on, &api, true, at(3500)); // 3rd failure

        let problems = sup.problems(at(3500));
        let crash = problems
            .iter()
            .find(|p| p.code == "broker_crashloop")
            .expect("3 failures in 5 min");
        assert!(
            crash.detail.as_deref().unwrap_or("").contains("boom"),
            "detail carries the broker log tail"
        );
    }

    #[test]
    fn a_crashing_broker_that_stabilizes_resets_its_backoff() {
        use Action::*;
        let dir = tmp("broker-stable-reset");
        let api = FakeBroker::default();
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        let t = Instant::now();
        let at = |ms: u64| t + Duration::from_millis(ms);
        let on = Desired {
            sharing: true,
            workers: 0,
        };

        // 1. Crash a child once.
        assert_eq!(sup.reconcile(&on, &api, true, at(0)), vec![StartBroker]);
        api.up.set(true);
        let pid0 = sup.children().broker.unwrap().pid;
        signal_group(pid0, libc::SIGKILL);
        assert!(
            wait_until_broker_exited(&mut sup),
            "the broker did not exit in time"
        );
        api.up.set(false);
        assert_eq!(
            sup.reconcile(&on, &api, true, at(100)),
            vec![],
            "1st failure: 1 s backoff"
        );

        // 2. Let it restart.
        assert_eq!(sup.reconcile(&on, &api, true, at(1200)), vec![StartBroker]);
        api.up.set(true);

        // 3. Tick at +301 s while it's alive: resets `broker_consecutive`.
        assert_eq!(sup.reconcile(&on, &api, true, at(1200 + 301_000)), vec![]);

        // 4. SIGKILL the pid from `sup.children()`: a second, independent crash.
        let pid1 = sup.children().broker.unwrap().pid;
        signal_group(pid1, libc::SIGKILL);
        assert!(
            wait_until_broker_exited(&mut sup),
            "the broker did not exit in time"
        );
        api.up.set(false);
        let t2 = at(1200 + 301_000 + 100);
        assert_eq!(
            sup.reconcile(&on, &api, true, t2),
            vec![],
            "post-reset failure: backoff restarts at 1 s"
        );

        // 5. The next restart backoff must be 1 s, not 2 s.
        assert_eq!(
            sup.reconcile(&on, &api, true, t2 + Duration::from_millis(900)),
            vec![],
            "not ready yet at 0.9 s"
        );
        assert_eq!(
            sup.reconcile(&on, &api, true, t2 + Duration::from_millis(1100)),
            vec![StartBroker],
            "ready at 1.1 s: proves the backoff reset to 1 s, not 2 s"
        );
    }

    #[test]
    fn a_deliberate_broker_stop_is_not_counted_as_a_crash_failure() {
        use Action::*;
        let dir = tmp("brk-stop");
        let api = FakeBroker::default();
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        let t = Instant::now();
        let on = Desired {
            sharing: true,
            workers: 0,
        };
        let off = Desired {
            sharing: false,
            workers: 0,
        };

        assert_eq!(sup.reconcile(&on, &api, true, t), vec![StartBroker]);
        api.up.set(true);
        assert_eq!(sup.reconcile(&off, &api, true, t), vec![StopBroker]);
        api.up.set(false); // the broker's /health stops answering once its process is gone
        assert!(sup.children().broker.is_none());
        let problems = sup.problems(t);
        assert!(
            !problems.iter().any(|p| p.code == "broker_crashloop"),
            "a clean stop is not a failure"
        );

        // No backoff was recorded, so starting it again is immediate.
        assert_eq!(sup.reconcile(&on, &api, true, t), vec![StartBroker]);
    }

    // --- An unknown `GET /workers` state must never read as "idle" ---

    /// A one-shot loopback HTTP server, in the idiom already used by
    /// src/broker/server.rs, src/enroll.rs and src/mesh_dir.rs: answers a
    /// single request with `status`/`body`, then the thread exits.
    fn stub_http_once(status: u16, body: &'static str) -> u16 {
        use std::io::{Read, Write};
        use std::net::TcpListener;
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        std::thread::spawn(move || {
            if let Ok((mut s, _)) = listener.accept() {
                let mut buf = [0u8; 1024];
                let _ = s.read(&mut buf);
                let _ = write!(
                    s,
                    "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                    body.len(),
                    body
                );
            }
        });
        port
    }

    #[test]
    fn workers_over_http_errs_on_a_non_2xx_response() {
        // A body that would parse just fine as a (trivially empty) worker
        // list, so this only fails if the status is actually checked.
        let port = stub_http_once(500, r#"{"workers":[]}"#);
        let api = HttpBrokerApi {
            port,
            worker_key: None,
        };
        assert!(
            api.workers().is_err(),
            "a non-2xx GET /workers must be an error, not an empty list"
        );
    }

    #[test]
    fn workers_over_http_parses_a_valid_2xx_response() {
        let body = r#"{"workers":[{"id":"id0","name":"fp-w0","status":"healthy","active_requests":1,"requests_5h":2,"requests_1w":3}]}"#;
        let port = stub_http_once(200, body);
        let api = HttpBrokerApi {
            port,
            worker_key: None,
        };
        let list = api.workers().expect("a 2xx body should parse");
        assert_eq!(list.len(), 1);
        assert_eq!(list[0].id, "id0");
        assert_eq!(list[0].active_requests, 1);
    }

    /// (a) A draining worker whose last GOOD entry already showed 0 active
    /// requests must not be read as idle just because the very next tick's
    /// `GET /workers` fails — that observation is unknown, not "confirmed
    /// idle". Only a FRESH tick (or drain-expiry) may complete the drain.
    #[test]
    fn a_workers_fetch_error_keeps_a_zero_reading_drain_from_completing() {
        use Action::*;
        let dir = tmp("stale-workers-zero");
        let api = FakeBroker::default();
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        let t = Instant::now();
        let on1 = Desired {
            sharing: true,
            workers: 1,
        };

        assert_eq!(sup.reconcile(&on1, &api, true, t), vec![StartBroker]);
        api.up.set(true);
        assert_eq!(sup.reconcile(&on1, &api, true, t), vec![StartWorker(0)]);
        *api.workers.borrow_mut() = vec![bw("id0", "testfp00-w0")]; // active_requests = 0
        assert_eq!(sup.reconcile(&on1, &api, true, t), vec![]);

        let off = Desired {
            sharing: false,
            workers: 1,
        };
        assert_eq!(sup.reconcile(&off, &api, true, t), vec![DrainWorker(0)]);
        assert_eq!(*api.drained.borrow(), vec!["id0".to_string()]);

        // The last GOOD reading already showed 0; a fetch error on the very
        // next tick must still not complete the drain.
        api.workers_err.set(true);
        assert_eq!(
            sup.reconcile(&off, &api, true, t),
            vec![],
            "an unknown observation must not read a zero-reading drain as idle"
        );
        assert_eq!(
            sup.broker_id(0).as_deref(),
            Some("id0"),
            "the retained id must survive the fetch-error tick"
        );

        // The next FRESH tick, still showing 0, completes the drain and
        // DELETEs the retained id.
        api.workers_err.set(false);
        assert_eq!(sup.reconcile(&off, &api, true, t), vec![KillWorker(0)]);
        assert_eq!(*api.deleted.borrow(), vec!["id0".to_string()]);
    }

    /// (b) Same, but the unknown tick comes from a `/health` timeout (the
    /// broker is observed `Starting`, not from a `GET /workers` error): the
    /// retained list and id must survive that too.
    #[test]
    fn a_health_check_timeout_keeps_a_busy_drain_from_completing() {
        use Action::*;
        let dir = tmp("stale-workers-health-timeout");
        let api = FakeBroker::default();
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        let t = Instant::now();
        let on1 = Desired {
            sharing: true,
            workers: 1,
        };

        assert_eq!(sup.reconcile(&on1, &api, true, t), vec![StartBroker]);
        api.up.set(true);
        assert_eq!(sup.reconcile(&on1, &api, true, t), vec![StartWorker(0)]);
        *api.workers.borrow_mut() = vec![bw_n("id0", "testfp00-w0", 2)];
        assert_eq!(sup.reconcile(&on1, &api, true, t), vec![]);

        let off = Desired {
            sharing: false,
            workers: 1,
        };
        assert_eq!(sup.reconcile(&off, &api, true, t), vec![DrainWorker(0)]);
        assert_eq!(*api.drained.borrow(), vec!["id0".to_string()]);

        // `/health` times out: the broker leader and the worker are both
        // still alive, but the tick is observed `Starting`, not `Running`.
        api.health_override.set(Some(false));
        assert_eq!(
            sup.reconcile(&off, &api, true, t),
            vec![],
            "a health-check timeout must not read the drain as idle"
        );
        assert_eq!(
            sup.broker_id(0).as_deref(),
            Some("id0"),
            "the retained id must survive the Starting tick"
        );

        // `/health` recovers; GET /workers still reports 2 active: still no kill.
        api.health_override.set(None);
        assert_eq!(sup.reconcile(&off, &api, true, t), vec![]);

        // Once a FRESH tick reports 0, the drain completes using the id
        // retained across the Starting tick.
        *api.workers.borrow_mut() = vec![bw("id0", "testfp00-w0")];
        assert_eq!(sup.reconcile(&off, &api, true, t), vec![KillWorker(0)]);
        assert_eq!(*api.deleted.borrow(), vec!["id0".to_string()]);
    }

    /// A killed worker's `last_workers` entry (its id and stats) must be
    /// forgotten, or a later `StartWorker` on the same slot would inherit
    /// the dead worker's id and stats by name during an error streak.
    #[test]
    fn killing_a_worker_during_an_error_streak_forgets_its_retained_entry() {
        use Action::*;
        let dir = tmp("kill-forgets-entry");
        let api = FakeBroker::default();
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        let t = Instant::now();
        let on1 = Desired {
            sharing: true,
            workers: 1,
        };

        assert_eq!(sup.reconcile(&on1, &api, true, t), vec![StartBroker]);
        api.up.set(true);
        assert_eq!(sup.reconcile(&on1, &api, true, t), vec![StartWorker(0)]);
        *api.workers.borrow_mut() = vec![bw_n("dead-id", "testfp00-w0", 9)];
        assert_eq!(sup.reconcile(&on1, &api, true, t), vec![]);

        // Scale down to 0 (sharing stays on, so the broker is untouched)
        // to drain the one live worker.
        let scale0 = Desired {
            sharing: true,
            workers: 0,
        };
        assert_eq!(sup.reconcile(&scale0, &api, true, t), vec![DrainWorker(0)]);
        assert_eq!(*api.drained.borrow(), vec!["dead-id".to_string()]);

        // Start a GET /workers error streak, then let the drained worker's
        // process actually exit: the drain still completes (the process is
        // simply gone), independent of freshness.
        api.workers_err.set(true);
        let pid0 = sup.children().workers["0"].pid;
        signal_group(pid0, libc::SIGKILL);
        assert!(wait_until_worker_exited(&mut sup, 0));
        assert_eq!(sup.reconcile(&scale0, &api, true, t), vec![KillWorker(0)]);
        assert_eq!(*api.deleted.borrow(), vec!["dead-id".to_string()]);

        // The dead entry must be gone now, even though the fetch is still
        // failing.
        assert_eq!(
            sup.broker_id(0),
            None,
            "the dead worker's id must not linger"
        );

        // Restart the same slot while the error streak continues: the new
        // worker must not inherit the dead worker's stats or id.
        assert_eq!(sup.reconcile(&on1, &api, true, t), vec![StartWorker(0)]);
        let view = sup.local_view(&on1, 5);
        assert_eq!(
            view.requests,
            Requests {
                last_5h: 0,
                last_1w: 0
            },
            "no stale stats for the new worker"
        );
        assert_eq!(view.workers.busy, 0, "no stale busy count");
        assert_eq!(
            sup.broker_id(0),
            None,
            "no stale id for the new worker either"
        );

        // A later drain must not target the dead id: no new entry is added
        // to the drained list (it still holds only the earlier, correct
        // "dead-id" drain from before the kill).
        assert_eq!(sup.reconcile(&scale0, &api, true, t), vec![DrainWorker(0)]);
        assert_eq!(
            *api.drained.borrow(),
            vec!["dead-id".to_string()],
            "no id to drain yet: the dead id must not be reused for the new worker"
        );
    }

    /// A drain that didn't take is sent again on every fresh tick until the
    /// broker itself reports the worker `draining`: the worker hadn't
    /// registered yet (no id to send), the POST failed, or the broker
    /// restarted and re-registered it as healthy. Otherwise the broker keeps
    /// routing jobs to a worker the agent is about to stop.
    #[test]
    fn a_drain_is_resent_until_the_broker_reports_the_worker_draining() {
        use Action::*;
        let dir = tmp("drain-resend");
        let api = FakeBroker::default();
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        let t = Instant::now();
        let on1 = Desired {
            sharing: true,
            workers: 1,
        };
        let off = Desired {
            sharing: false,
            workers: 1,
        };
        let drained = || api.drained.borrow().clone();
        let with_status = |id: &str, status: &str| BrokerWorker {
            status: status.into(),
            // Busy, so the drain can't complete while this test watches it.
            ..bw_n(id, "testfp00-w0", 1)
        };

        assert_eq!(sup.reconcile(&on1, &api, true, t), vec![StartBroker]);
        api.up.set(true);
        assert_eq!(sup.reconcile(&on1, &api, true, t), vec![StartWorker(0)]);

        // Paused before the worker registered: there is no id to send yet.
        assert_eq!(sup.reconcile(&off, &api, true, t), vec![DrainWorker(0)]);
        assert!(drained().is_empty());

        // It registers, and the drain POST fails.
        *api.workers.borrow_mut() = vec![with_status("id0", "healthy")];
        api.drain_fails.set(true);
        assert_eq!(sup.reconcile(&off, &api, true, t), vec![]);
        assert_eq!(drained(), vec!["id0".to_string()], "sent once it has an id");

        // The broker still reports it healthy: sent again.
        api.drain_fails.set(false);
        assert_eq!(sup.reconcile(&off, &api, true, t), vec![]);
        assert_eq!(drained().len(), 2, "re-sent after the failed POST");

        // The broker now reports it draining: nothing more is sent.
        *api.workers.borrow_mut() = vec![with_status("id0", "draining")];
        assert_eq!(sup.reconcile(&off, &api, true, t), vec![]);
        assert_eq!(drained().len(), 2);

        // The broker restarted and re-registered it as healthy, under a new id.
        *api.workers.borrow_mut() = vec![with_status("id0-b", "healthy")];
        assert_eq!(sup.reconcile(&off, &api, true, t), vec![]);
        assert_eq!(
            drained().last().map(String::as_str),
            Some("id0-b"),
            "re-sent after a broker restart"
        );

        // An unknown tick (GET /workers failed) sends nothing.
        let before = drained().len();
        api.workers_err.set(true);
        assert_eq!(sup.reconcile(&off, &api, true, t), vec![]);
        assert_eq!(drained().len(), before);
    }

    // --- A leader found dead has its process group killed ---

    #[test]
    fn a_worker_leader_that_exits_early_has_its_surviving_group_member_killed() {
        use Action::*;
        let dir = tmp("leader-exits-early");
        let api = FakeBroker::default();
        api.up.set(true);
        // The leader backgrounds a member, prints that member's own pid, then
        // exits, leaving it as the sole survivor in the group. We poll the
        // member's own pid with `is_alive` (not `group_alive` on the pgid):
        // on Linux, `group_alive` (`killpg(pg, 0)`) still counts an unreaped
        // zombie member as "the group is alive", where `is_alive` treats a
        // zombie as dead.
        let mut sup = SupGuard(Supervisor::new(spec(
            &dir,
            sh("sleep 30 & echo $!; exit 0"),
        )));
        let on1 = Desired {
            sharing: true,
            workers: 1,
        };
        // Adopt a fake running broker so the plan goes straight to workers.
        let broker_child =
            spawn_in_group(&sh("exec sleep 60"), None, &[], &dir.join("b.log")).unwrap();
        let _guard = PgGuard(broker_child.id());
        sup.broker = Some(Proc::spawned(broker_child));

        assert_eq!(
            sup.reconcile(&on1, &api, true, Instant::now()),
            vec![StartWorker(0)]
        );
        let pid0 = sup.children().workers["0"].pid; // the leader's pid == the group's pgid
                                                    // Guard the whole group so a failing assertion below can't orphan
                                                    // the backgrounded `sleep 30`.
        let _pg_guard = PgGuard(pid0);
        let log = sup.worker_log(0);
        let member = (0..40)
            .find_map(|_| {
                std::thread::sleep(Duration::from_millis(50));
                std::fs::read_to_string(&log)
                    .ok()?
                    .trim()
                    .parse::<u32>()
                    .ok()
            })
            .expect("the worker script printed the backgrounded sleep's pid");
        assert!(
            wait_until_worker_exited(&mut sup, 0),
            "the leader exits on its own almost immediately"
        );
        assert!(
            is_alive(member),
            "the backgrounded `sleep 30` must still be alive right after the leader exits"
        );

        // The next tick observes the dead leader and must kill the whole
        // group, not just note the crash.
        sup.reconcile(&on1, &api, true, Instant::now());
        assert!(
            wait_until(|| !is_alive(member)),
            "the surviving group member must be killed once the leader is found dead"
        );
    }

    // --- Broker port fallback (zc#agent-ports) ---

    /// A wildcard holder (`*:P`, exactly how the broker itself binds) must
    /// never read as free: a specific-address (`127.0.0.1:P`) bind succeeds
    /// right past it on macOS/BSD because Rust std sets `SO_REUSEADDR`, so a
    /// naive probe would hand the agent a port its own broker then fails to
    /// bind, crash-looping forever (review A1).
    #[test]
    fn port_free_is_false_when_something_holds_the_wildcard_address() {
        let holder = std::net::TcpListener::bind(("0.0.0.0", 0)).unwrap();
        let port = holder.local_addr().unwrap().port();
        let api = HttpBrokerApi {
            port: 0,
            worker_key: None,
        };
        assert!(
            !api.port_free(port),
            "a *:{port} holder must not read as free"
        );
        drop(holder);
    }

    mod choose_broker_port_tests {
        use super::choose_broker_port;

        #[test]
        fn preferred_free() {
            assert_eq!(
                choose_broker_port(9000, 9001, 9010, |p| p == 9000, || None),
                Some(9000)
            );
        }

        #[test]
        fn preferred_held_falls_back_to_9001() {
            assert_eq!(
                choose_broker_port(9000, 9001, 9010, |p| p == 9001, || None),
                Some(9001)
            );
        }

        #[test]
        fn preferred_and_9001_through_9003_held_falls_back_to_9004() {
            assert_eq!(
                choose_broker_port(9000, 9001, 9010, |p| p == 9004, || None),
                Some(9004)
            );
        }

        #[test]
        fn the_whole_range_held_falls_back_to_an_os_pick() {
            assert_eq!(
                choose_broker_port(9000, 9001, 9010, |_| false, || Some(54321)),
                Some(54321)
            );
        }

        #[test]
        fn nothing_available_is_none() {
            assert_eq!(
                choose_broker_port(9000, 9001, 9010, |_| false, || None),
                None
            );
        }
    }

    /// A zakuro broker the agent doesn't own on the *preferred* port (9000)
    /// is always `Unmanaged`, even though nothing has tried to start our own
    /// broker yet.
    #[test]
    fn a_foreign_broker_on_the_preferred_port_is_unmanaged() {
        let dir = tmp("foreign-preferred");
        let api = FakeBroker::default();
        api.foreign_broker_port.set(Some(9000));
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        let on = Desired {
            sharing: true,
            workers: 1,
        };
        assert_eq!(sup.reconcile(&on, &api, true, Instant::now()), vec![]);
        assert!(sup.broker_unmanaged());
        assert_eq!(sup.broker_port(), 9000, "never relocated: never touched");
    }

    /// The preferred port is held by something that isn't a zakuro broker:
    /// the supervisor relocates to the first free fallback port and starts
    /// the broker there, rendering argv and env with that port.
    #[test]
    fn a_blocked_preferred_port_falls_back_and_renders_the_chosen_port() {
        use Action::*;
        let dir = tmp("fallback-9001");
        let api = FakeBroker::default();
        api.blocked_ports.borrow_mut().insert(9000);
        let mut spawn_spec = spec(&dir, sh("exec sleep 60"));
        spawn_spec.broker_argv = vec!["sh".into(), "-c".into(), "exec sleep 60 {port}".into()];
        let mut sup = SupGuard(Supervisor::new(spawn_spec));
        let on = Desired {
            sharing: true,
            workers: 0,
        };
        assert_eq!(
            sup.reconcile(&on, &api, true, Instant::now()),
            vec![StartBroker]
        );
        assert_eq!(sup.broker_port(), 9001);
        assert_eq!(sup.children().broker.unwrap().port, Some(9001));
    }

    /// N3: an idle agent (sharing off) must never probe or bind a fallback
    /// port at all -- `port_free`/`os_assigned_port` do real socket I/O
    /// (`HttpBrokerApi`'s wildcard bind), and doing that every reconcile
    /// tick while nothing is ever going to start can raise the macOS
    /// "accept incoming connections?" prompt long before sharing is turned
    /// on. `broker_port` stays at the preferred port (unused, since nothing
    /// starts); a foreign broker on the preferred port is still detected
    /// (`is_broker_at` is a connect, not a listen, so it's fine to keep
    /// doing that every tick).
    #[test]
    fn an_idle_tick_never_probes_or_binds_a_fallback_port() {
        let dir = tmp("fallback-while-paused");
        let api = FakeBroker::default();
        api.blocked_ports.borrow_mut().insert(9000); // something (not us) holds 9000
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        let off = Desired {
            sharing: false,
            workers: 0,
        };
        assert_eq!(sup.reconcile(&off, &api, true, Instant::now()), vec![]);
        assert_eq!(
            *api.port_free_calls.borrow(),
            0,
            "an idle tick must never probe a fallback port"
        );
        assert_eq!(
            *api.os_assigned_port_calls.borrow(),
            0,
            "an idle tick must never bind an OS-assigned port either"
        );
        assert_eq!(
            sup.broker_port(),
            9000,
            "nothing was chosen, so it stays at the preferred port"
        );
    }

    /// The same idle tick still correctly detects a foreign zakuro broker on
    /// the preferred port (a connect, not a listen -- safe every tick).
    #[test]
    fn an_idle_tick_still_detects_an_unmanaged_broker_on_the_preferred_port() {
        let dir = tmp("unmanaged-while-paused");
        let api = FakeBroker::default();
        api.foreign_broker_port.set(Some(9000));
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        let off = Desired {
            sharing: false,
            workers: 0,
        };
        sup.reconcile(&off, &api, true, Instant::now());
        assert!(sup.broker_unmanaged());
    }

    /// Once sharing turns on, the very next tick DOES probe (exactly the
    /// `choose_broker_port` wiring the earlier fallback tests already cover)
    /// -- this pins that the gate is "about to start", not "sharing was ever
    /// turned on in the past".
    #[test]
    fn turning_sharing_on_probes_and_starts_on_the_very_next_tick() {
        use Action::*;
        let dir = tmp("fallback-then-share-on");
        let api = FakeBroker::default();
        api.blocked_ports.borrow_mut().insert(9000);
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        let off = Desired {
            sharing: false,
            workers: 0,
        };
        sup.reconcile(&off, &api, true, Instant::now());
        assert_eq!(*api.port_free_calls.borrow(), 0);

        let on = Desired {
            sharing: true,
            workers: 0,
        };
        assert_eq!(
            sup.reconcile(&on, &api, true, Instant::now()),
            vec![StartBroker]
        );
        assert!(*api.port_free_calls.borrow() > 0);
        assert_eq!(sup.broker_port(), 9001);
    }

    /// The preferred port and the whole 9001..=9010 fallback range are all
    /// held, and there is no OS-assigned port either: `port_in_use`, not a
    /// silent standstill.
    #[test]
    fn everything_blocked_and_no_os_pick_is_port_in_use() {
        let dir = tmp("all-blocked");
        let api = FakeBroker::default();
        for p in 9000..=9010 {
            api.blocked_ports.borrow_mut().insert(p);
        }
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        let on = Desired {
            sharing: true,
            workers: 0,
        };
        sup.reconcile(&on, &api, true, Instant::now());
        let problems = sup.problems(Instant::now());
        assert!(problems.iter().any(|p| p.code == "port_in_use"));
    }

    /// Once 9000..=9010 are all held, a random OS-assigned port is used
    /// instead of giving up.
    #[test]
    fn everything_blocked_falls_back_to_an_os_assigned_port() {
        use Action::*;
        let dir = tmp("os-pick");
        let api = FakeBroker::default();
        for p in 9000..=9010 {
            api.blocked_ports.borrow_mut().insert(p);
        }
        api.os_pick.set(Some(54321));
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        let on = Desired {
            sharing: true,
            workers: 0,
        };
        assert_eq!(
            sup.reconcile(&on, &api, true, Instant::now()),
            vec![StartBroker]
        );
        assert_eq!(sup.broker_port(), 54321);
    }

    /// `adopt` restores the broker's port from its recorded `ChildRecord`; a
    /// record without a `port` (written before this field existed) means
    /// 9000.
    #[test]
    fn adopt_restores_the_brokers_recorded_port_defaulting_to_9000() {
        let dir = tmp("adopt-port");
        let broker_child =
            spawn_in_group(&sh("exec sleep 60"), None, &[], &dir.join("b.log")).unwrap();
        let _guard = PgGuard(broker_child.id());
        let pid = broker_child.id();
        let started = proc_start_time(pid).unwrap_or_default();
        let child_guard = ChildGuard(broker_child);

        let mut recorded = Children::default();
        recorded.broker = Some(ChildRecord {
            pid,
            started_at: started.clone(),
            port: Some(9007),
        });
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        sup.adopt(&recorded, &proc_start_time);
        assert_eq!(sup.broker_port(), 9007);

        let mut recorded_no_port = Children::default();
        recorded_no_port.broker = Some(ChildRecord {
            pid,
            started_at: started,
            port: None,
        });
        let mut sup2 = SupGuard(Supervisor::new(spec(&dir, sh("exec sleep 60"))));
        sup2.adopt(&recorded_no_port, &proc_start_time);
        assert_eq!(sup2.broker_port(), 9000, "a missing port field means 9000");
        drop(child_guard);
    }

    /// `set_worker_cwd` is what lets the agent resolve the zakuro dir fresh
    /// on every reconcile tick (`Core::reconcile_once`) instead of once at
    /// startup: a newly spawned worker must run in whatever directory was
    /// set just before this tick, without restarting the agent.
    #[test]
    fn set_worker_cwd_changes_where_the_next_spawned_worker_runs() {
        use Action::*;
        let dir = tmp("worker-cwd");
        let cwd_a = dir.join("cwd-a");
        let cwd_b = dir.join("cwd-b");
        std::fs::create_dir_all(&cwd_a).unwrap();
        std::fs::create_dir_all(&cwd_b).unwrap();
        let api = FakeBroker::default();
        api.up.set(true);
        let mut sup = SupGuard(Supervisor::new(spec(&dir, sh("pwd"))));
        let broker_child =
            spawn_in_group(&sh("exec sleep 60"), None, &[], &dir.join("b.log")).unwrap();
        let _guard = PgGuard(broker_child.id());
        sup.broker = Some(Proc::spawned(broker_child));

        sup.set_worker_cwd(Some(cwd_a.clone()));
        assert_eq!(
            sup.reconcile(
                &Desired {
                    sharing: true,
                    workers: 1
                },
                &api,
                true,
                Instant::now()
            ),
            vec![StartWorker(0)]
        );

        sup.set_worker_cwd(Some(cwd_b.clone()));
        assert_eq!(
            sup.reconcile(
                &Desired {
                    sharing: true,
                    workers: 2
                },
                &api,
                true,
                Instant::now()
            ),
            vec![StartWorker(1)]
        );

        assert!(wait_until_worker_exited(&mut sup, 0));
        assert!(wait_until_worker_exited(&mut sup, 1));
        let printed_a = std::fs::read_to_string(sup.worker_log(0)).unwrap();
        let printed_b = std::fs::read_to_string(sup.worker_log(1)).unwrap();
        assert_eq!(
            printed_a.trim(),
            cwd_a.canonicalize().unwrap().to_str().unwrap()
        );
        assert_eq!(
            printed_b.trim(),
            cwd_b.canonicalize().unwrap().to_str().unwrap()
        );
    }
}