car-registry 0.31.0

//! Lifecycle-managed agent supervisor.
//!
//! [`Supervisor`] reads a declarative manifest from
//! `~/.car/agents.json`, spawns each entry as a child process, and
//! keeps it running per the configured restart policy. Stdout/stderr
//! are captured per-agent under `<log_dir>/<id>.{stdout,stderr}.log`.
//!
//! Sibling to the parent crate's [`crate::AgentRegistry`]:
//! `AgentRegistry` is observe-only ("agents have announced
//! themselves"); `Supervisor` is declarative ("agents we should
//! keep running"). Closes [Parslee-ai/car-releases#27].
//!
//! ## Example
//!
//! ```no_run
//! # async fn run() -> Result<(), Box<dyn std::error::Error>> {
//! use car_registry::supervisor::{AgentSpec, RestartPolicy, Supervisor};
//!
//! let supervisor = Supervisor::user_default()?;
//! supervisor
//!     .upsert(AgentSpec {
//!         id: "trader".into(),
//!         name: "Trader".into(),
//!         // Absolute path to an executable — `$PATH` lookup and
//!         // scratch dirs (/tmp etc.) are rejected at upsert time
//!         // since the 2026-05 audit.
//!         command: "/usr/local/bin/node".into(),
//!         args: vec!["/Users/me/git/trader/index.js".into()],
//!         cwd: Some("/Users/me/git/trader".into()),
//!         env: Default::default(),
//!         restart: RestartPolicy::OnFailure,
//!         max_restarts: 10,
//!         backoff_secs: 5,
//!         // Default is `false`. Opt in per-agent only when you
//!         // genuinely want this to come up at car-server boot.
//!         auto_start: true,
//!         // Empty string lets upsert mint a fresh token; pass a
//!         // real one only when re-importing an existing manifest.
//!         token: String::new(),
//!     })
//!     .await?;
//! supervisor.start_all().await;
//! # Ok(()) }
//! ```
//!
//! ## Design
//!
//! - **Single owner.** One [`Supervisor`] per process AND one per
//!   manifest file across processes. Two would double-spawn every
//!   declared agent against shared external state. Enforced via an
//!   OS-level exclusive lock on `<manifest_path>.lock`, held for the
//!   supervisor's lifetime; the second acquirer fails fast with
//!   [`SupervisorError::AlreadyRunning`]. Closes #44.
//! - **Manifest is the source of truth.** Mutations write through
//!   atomically; the on-disk JSON is authoritative across restarts.
//! - **Idempotent state transitions.** `start` on an already-running
//!   agent is a no-op; `stop` on a stopped one is too.
//! - **Restart policy is declarative.** The supervisor task loops
//!   until the policy says stop (max-restarts hit, or `Never`).

use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, HashMap};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use thiserror::Error;
use tokio::sync::{Mutex, RwLock};
use tokio::task::JoinHandle;

use crate::manifest::AgentManifest;

#[derive(Debug, Error)]
pub enum SupervisorError {
    #[error("invalid agent id (must be non-empty, alphanumeric + `-_.`): {0:?}")]
    InvalidId(String),
    #[error("invalid agent command: {reason} ({command:?})")]
    InvalidCommand {
        command: String,
        reason: &'static str,
    },
    #[error("agent {0} not found")]
    NotFound(String),
    /// [`Supervisor::wait_for`] hit its deadline before the agent reached any
    /// target status.
    #[error("timed out after {timeout:?} waiting for agent {id} to reach a target status (last status: {last:?})")]
    WaitTimeout {
        id: String,
        last: AgentStatus,
        timeout: std::time::Duration,
    },
    #[error("could not resolve home directory")]
    NoHomeDir,
    #[error("supervisor I/O error: {0}")]
    Io(#[from] std::io::Error),
    #[error("supervisor JSON error: {0}")]
    Json(#[from] serde_json::Error),
    /// Catch-all for manifest-validation and dispatch errors that
    /// don't fit the narrower variants. Used by the
    /// `manifest.toml`-driven path (Parslee-ai/car#182).
    #[error("{0}")]
    Other(String),
    /// Another supervisor process already holds the manifest lock.
    /// Refusing to spawn would-be-duplicate children. Operators
    /// hitting this should stop the other supervisor (or run with
    /// `--no-supervisor` once that flag lands) — see #44 for the
    /// double-spawn-against-live-state bug this guard exists to
    /// prevent.
    #[error("another supervisor already owns this manifest (lock file: {0}). Refusing to spawn duplicates.")]
    AlreadyRunning(PathBuf),
}

/// What to do when a managed agent exits.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum RestartPolicy {
    /// Don't restart. The agent runs once.
    Never,
    /// Restart only when the process exits non-zero or is killed.
    /// Clean exits stop supervision.
    #[default]
    OnFailure,
    /// Restart unconditionally (also on clean exit).
    Always,
}

/// Runtime status of a managed agent. Distinct from
/// [`car_registry::AgentStatus`] which is the *agent's* self-reported
/// liveness signal — supervisor status describes what *we* know
/// about the child process from this side.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum AgentStatus {
    /// Never started, or stopped after a clean exit / explicit stop.
    #[default]
    Stopped,
    /// Spawn requested; process not yet visible.
    Starting,
    /// Process is alive and recent.
    Running,
    /// Process exited unexpectedly; supervisor is waiting out the
    /// backoff before respawning.
    Backoff,
    /// Process kept failing past `max_restarts`. Supervisor stopped
    /// trying. Manual `start` resets this.
    Errored,
}

/// Declarative spec for a managed agent. Persisted in
/// `<manifest_dir>/agents.json`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AgentSpec {
    /// Stable identifier — also the log-file prefix and the manifest
    /// key. Restricted to filename-safe characters.
    pub id: String,
    /// Human-readable label for UI.
    pub name: String,
    /// Program to exec. **Must be an absolute path** that points at
    /// an existing executable file the launching user can run, and
    /// must not live under a world-writable directory
    /// (`/tmp`, `/private/tmp`, `/var/tmp`, `/dev/shm`). The 2026-05
    /// security audit found this was the load-bearing capability in
    /// a drive-by-RCE chain, so the validation is enforced at
    /// [`Supervisor::upsert`] time rather than left to the spawn
    /// path. `$PATH` lookup is intentionally rejected to remove the
    /// PATH-injection variant.
    pub command: String,
    /// Arguments passed to `command`. Empty by default.
    #[serde(default)]
    pub args: Vec<String>,
    /// Working directory the child runs in. Defaults to the parent's
    /// cwd when `None`.
    #[serde(default)]
    pub cwd: Option<PathBuf>,
    /// Extra environment variables. Merged on top of the parent's
    /// env — `PATH`, `HOME`, etc. inherit unless explicitly
    /// overridden here.
    #[serde(default)]
    pub env: BTreeMap<String, String>,
    /// What to do when the child exits. See [`RestartPolicy`].
    #[serde(default)]
    pub restart: RestartPolicy,
    /// Cap on consecutive restart attempts. After this many failures
    /// in a row the supervisor gives up and marks the agent
    /// `Errored`. A successful long-run resets the counter.
    #[serde(default = "default_max_restarts")]
    pub max_restarts: u32,
    /// Base backoff before a restart attempt, in seconds. Used as
    /// the floor for an exponential, capped, jittered delay that
    /// grows with consecutive failures — see `restart_backoff`.
    #[serde(default = "default_backoff")]
    pub backoff_secs: u64,
    /// When `true`, [`Supervisor::start_all`] launches this agent on
    /// car-server boot. Manual `start` ignores this field. **Defaults
    /// to `false`** since 2026-05 — the prior default-on combined
    /// with unauth WS + unvalidated `command` to land an attacker's
    /// binary at every login. Operators who want boot-time auto-start
    /// must opt in explicitly per agent.
    #[serde(default)]
    pub auto_start: bool,

    /// Per-agent auth token (#169). Minted by the supervisor on first
    /// upsert as a 43-char base64url-no-pad random string and persisted
    /// alongside the rest of the spec. Subsequent upserts that don't
    /// supply a token retain the existing one — rotation is explicit
    /// (operator passes `token: ""` or a new value at upsert time).
    /// The supervisor injects this into the child's environment as
    /// `CAR_AGENT_TOKEN` at spawn; the WS dispatcher matches it
    /// against the value the child presents in
    /// `session.auth { token, agent_id }` to bind the connection.
    #[serde(default)]
    pub token: String,
}

fn default_max_restarts() -> u32 {
    10
}
fn default_backoff() -> u64 {
    5
}

/// What [`Supervisor::list`] returns — spec + observed runtime.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ManagedAgent {
    #[serde(flatten)]
    pub spec: AgentSpec,
    pub status: AgentStatus,
    /// PID of the running child, when one exists.
    pub pid: Option<u32>,
    /// Exit code of the most recent terminated child. `None` until
    /// the first exit. `Some(-1)` for "killed by signal" — exact
    /// signal number isn't preserved.
    pub last_exit_code: Option<i32>,
    /// Number of consecutive restart attempts since the last clean
    /// state. Resets on a manual `start` or after the agent runs
    /// successfully for `restart_clear_secs`.
    pub restart_count: u32,
    /// UNIX timestamp when the *current* child was spawned. `None`
    /// when stopped.
    pub started_at: Option<i64>,
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StopSignal {
    /// SIGTERM, then SIGKILL after `grace_secs`. Default.
    Term,
    /// SIGKILL immediately. No grace.
    Kill,
}

impl Default for StopSignal {
    fn default() -> Self {
        StopSignal::Term
    }
}

/// Which captured stream(s) `read_log` should return.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum LogStream {
    /// Both streams, each independently tailed (legacy default — the
    /// returned `lines` is stdout-then-stderr).
    #[default]
    Combined,
    /// Only `<id>.stdout.log` — the agent's live activity.
    Stdout,
    /// Only `<id>.stderr.log` — errors/crash dumps.
    Stderr,
}

impl LogStream {
    /// Parse the wire string. Unknown values fall back to `Combined`
    /// rather than erroring — a viewer passing a typo still gets logs.
    pub fn from_wire(s: Option<&str>) -> LogStream {
        match s {
            Some("stdout") => LogStream::Stdout,
            Some("stderr") => LogStream::Stderr,
            _ => LogStream::Combined,
        }
    }
}

/// Result of [`Supervisor::read_log`] — per-stream tails plus the
/// metadata a real log viewer needs (file paths to reveal, total line
/// counts for a scrollbar, and whether older lines remain for paging).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LogTail {
    /// Legacy combined view: stdout lines then stderr lines, each
    /// independently budgeted. Kept so existing callers don't break.
    pub lines: Vec<String>,
    /// The stdout window (empty when `stream` excluded it).
    pub stdout: Vec<String>,
    /// The stderr window (empty when `stream` excluded it).
    pub stderr: Vec<String>,
    /// Total stdout lines in the scanned tail (before windowing), so a
    /// viewer can show "showing N of M" and size a scrollbar. Exact for
    /// any log within the tail byte ceiling (every realistic log); for a
    /// pathological multi-GB log it counts only the lines scanned within
    /// the last ceiling bytes, and `more` will be `true`.
    pub stdout_total: usize,
    /// Total stderr lines in the scanned tail (before windowing). Same
    /// ceiling caveat as `stdout_total`.
    pub stderr_total: usize,
    /// Absolute path to the stdout log file (for "reveal in Finder").
    pub stdout_path: String,
    /// Absolute path to the stderr log file.
    pub stderr_path: String,
    /// `true` when paging further back (larger `offset`) would surface
    /// older lines on at least one included stream — drives "load more".
    pub more: bool,
}

/// One stream's windowed read plus the bookkeeping `read_log` folds
/// into [`LogTail`]. Internal.
#[derive(Default)]
struct StreamWindow {
    lines: Vec<String>,
    total: usize,
    more: bool,
}

#[derive(Debug, Default, Serialize, Deserialize)]
struct Manifest {
    #[serde(default)]
    agents: Vec<AgentSpec>,
}

struct AgentSlot {
    spec: AgentSpec,
    runtime: AgentRuntime,
    /// Drop-on-stop sentinel — the supervisor task watches this for
    /// closure to know it should stop respawning. We just carry the
    /// sender; the task holds the receiver.
    stop_tx: Option<tokio::sync::watch::Sender<bool>>,
    /// Handle to the supervisor task. `await`ing it joins the
    /// supervision loop; `abort()` cancels it.
    task: Option<JoinHandle<()>>,
    /// Windows-only: Job Object owning the current child's process
    /// tree. Created at spawn time, dropped (which kills the tree on
    /// the kernel side if not already terminated) when the slot's
    /// child exits or the supervisor stops. `None` on non-Windows
    /// targets and during the brief window between spawn and Job
    /// Object setup. See [`JobObject`] and #231 §5.1.
    job: Option<Arc<JobObject>>,
}

#[derive(Debug, Clone, Default)]
struct AgentRuntime {
    status: AgentStatus,
    pid: Option<u32>,
    last_exit_code: Option<i32>,
    restart_count: u32,
    started_at: Option<i64>,
}

/// Process supervisor.
///
/// Cheap to clone; all state lives behind `Arc<RwLock<...>>`. Hold
/// one across the whole process — two would race on the manifest
/// file and double-spawn children.
#[derive(Clone)]
pub struct Supervisor {
    manifest_path: PathBuf,
    log_dir: PathBuf,
    state: Arc<RwLock<HashMap<String, AgentSlot>>>,
    /// Mutex held during manifest write. Kept separate from `state`
    /// so list/upsert reads don't block on disk I/O.
    manifest_lock: Arc<Mutex<()>>,
    /// OS-level exclusive lock on `<manifest_path>.lock`, held for
    /// the Supervisor's lifetime. Prevents two car-server processes
    /// on the same machine from both supervising the same manifest
    /// and double-spawning every agent against shared external
    /// state (broker accounts, on-disk state dirs, etc.). The lock
    /// is dropped automatically when the file handle drops, so
    /// `Drop`-ing the Supervisor releases it. Closes #44.
    _process_lock: Arc<std::fs::File>,
    /// Grace window before SIGKILL when stopping with `Term`.
    pub grace_secs: u64,
    /// Default environment exported into every spawned child *before*
    /// the per-spec `spec.env` is merged on top. Used by the daemon
    /// to pass `CAR_DAEMON_URL` / `CAR_AUTH_TOKEN` / `CAR_AGENT_ID`
    /// (#172, #169) without each lifecycle-agent SDK needing to know
    /// the platform-specific path the token lives at. Per-spec env
    /// still wins on conflict — operators can override.
    default_child_env: Arc<RwLock<BTreeMap<String, String>>>,
}

impl Supervisor {
    /// Use `~/.car/agents.json`, `~/.car/agents/`, and `~/.car/logs/`.
    /// Creates the parent directories if missing. Dual-reads the
    /// legacy JSON file and the new manifest directory per
    /// Parslee-ai/car#182 phase 1.
    pub fn user_default() -> Result<Self, SupervisorError> {
        let manifest_path = Self::user_default_manifest_path()?;
        let log_dir = manifest_path
            .parent()
            .map(|p| p.join("logs"))
            .unwrap_or_else(|| PathBuf::from("logs"));
        Self::with_paths(manifest_path, log_dir)
    }

    /// Resolve `~/.car/agents.json` without acquiring the singleton
    /// lock. Useful for read-only callers (e.g. FFI consumers) that
    /// want to enumerate declared agents while another process owns
    /// the live supervisor — pair with [`Supervisor::list_from_manifest`].
    pub fn user_default_manifest_path() -> Result<PathBuf, SupervisorError> {
        let home = std::env::var_os("HOME")
            .or_else(|| std::env::var_os("USERPROFILE"))
            .ok_or(SupervisorError::NoHomeDir)?;
        Ok(PathBuf::from(home).join(".car").join("agents.json"))
    }

    /// Read declared agents from `manifest_path` without acquiring
    /// the supervisor singleton lock. Runtime fields (`status`,
    /// `pid`, `restart_count`, etc.) are returned at their defaults —
    /// they're owned by whatever process currently supervises the
    /// manifest and aren't observable from outside that process.
    ///
    /// Use this from read-only inspection paths (FFI consumers, CLI
    /// status commands) when a live supervisor in another process
    /// holds the manifest lock. Mutations still require an instance
    /// method on a `Supervisor` that successfully called
    /// [`Supervisor::with_paths`] — silently bypassing the lock
    /// would re-introduce the double-spawn class from #44.
    ///
    /// Today this only reads the legacy `agents.json` file. The new
    /// `<dir>/agents/<id>/manifest.toml` layout is intentionally
    /// excluded from the fallback for now; entries that exist only
    /// in the new layout won't surface here until that follow-up
    /// lands. The common case during migration is that legacy +
    /// new-layout name the same agents (#182 phase 1 mirrors on
    /// every boot), so the omission is conservative rather than
    /// load-bearing.
    pub fn list_from_manifest(manifest_path: &Path) -> Result<Vec<ManagedAgent>, SupervisorError> {
        let m = load_manifest(manifest_path)?;
        let mut out: Vec<ManagedAgent> = m
            .agents
            .into_iter()
            .map(|spec| ManagedAgent {
                spec,
                status: AgentStatus::default(),
                pid: None,
                last_exit_code: None,
                restart_count: 0,
                started_at: None,
            })
            .collect();
        out.sort_by(|a, b| a.spec.id.cmp(&b.spec.id));
        Ok(out)
    }

    /// Read-only health view over `manifest_path` without acquiring
    /// the supervisor singleton lock. See
    /// [`Supervisor::list_from_manifest`] for the contract; same
    /// legacy-only caveat applies.
    pub fn health_from_manifest(manifest_path: &Path) -> Result<Vec<AgentHealth>, SupervisorError> {
        let m = load_manifest(manifest_path)?;
        let mut out: Vec<AgentHealth> = m
            .agents
            .into_iter()
            .map(|spec| {
                let command = spec.command.clone();
                match validate_command(&command) {
                    Ok(()) => AgentHealth {
                        id: spec.id,
                        command,
                        ok: true,
                        reason: None,
                    },
                    Err(e) => AgentHealth {
                        id: spec.id,
                        command,
                        ok: false,
                        reason: Some(e.to_string()),
                    },
                }
            })
            .collect();
        out.sort_by(|a, b| a.id.cmp(&b.id));
        Ok(out)
    }

    /// Construct with explicit paths. The manifest directory is
    /// derived as `<manifest_path>/../agents/` so tests + the
    /// default share one resolution rule.
    ///
    /// **Dual-read migration** (Parslee-ai/car#182 phase 1):
    /// loads agents from BOTH the legacy `agents.json` AND the
    /// new `<dir>/agents/<id>/manifest.toml` layout. Legacy
    /// entries that don't yet exist in the new layout are mirrored
    /// at boot. On the next boot, both sources name the same
    /// agents; the migration is idempotent. The legacy file
    /// remains the read-source-of-truth for one more minor release
    /// before removal — `tracing::warn!` fires when it carries
    /// entries so operators see the deprecation.
    pub fn with_paths(manifest_path: PathBuf, log_dir: PathBuf) -> Result<Self, SupervisorError> {
        if let Some(parent) = manifest_path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        std::fs::create_dir_all(&log_dir)?;

        // Acquire the cross-process singleton lock BEFORE any other
        // state mutation. Closes #44: two car-server processes
        // supervising the same manifest would each spawn every
        // declared agent against shared external state (broker
        // accounts, on-disk state dirs, even named OS resources),
        // and the second process's children would clobber the
        // first's child-tracking maps as they self-registered.
        //
        // The lock file lives at `<manifest_path>.lock`. We never
        // write to it — the handle's existence + an exclusive
        // advisory lock are the entire protocol. The handle is held
        // in `_process_lock` for the supervisor's lifetime; the OS
        // releases the lock when the last `Arc` clone drops (i.e.
        // when the supervisor itself drops). Lock files are
        // intentionally not cleaned up on drop: a unlink-on-drop
        // races against a new acquirer creating the file before our
        // Arc actually goes away.
        let lock_path = {
            let mut s = manifest_path.as_os_str().to_owned();
            s.push(".lock");
            PathBuf::from(s)
        };
        let lock_file = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&lock_path)?;
        match lock_file.try_lock() {
            Ok(()) => {}
            Err(std::fs::TryLockError::WouldBlock) => {
                return Err(SupervisorError::AlreadyRunning(lock_path));
            }
            Err(std::fs::TryLockError::Error(e)) => return Err(SupervisorError::Io(e)),
        }

        let agents_dir = manifest_path
            .parent()
            .map(|p| p.join("agents"))
            .unwrap_or_else(|| PathBuf::from("agents"));
        std::fs::create_dir_all(&agents_dir)?;

        let legacy = load_manifest(&manifest_path)?;
        let mut by_id: HashMap<String, AgentSpec> = HashMap::new();

        // Legacy entries first. Fire a single deprecation warning
        // when the file contains anything; per-entry warnings
        // would spam the daemon log every boot.
        if !legacy.agents.is_empty() {
            tracing::warn!(
                count = legacy.agents.len(),
                path = %manifest_path.display(),
                "loading agents from legacy agents.json. This file is \
                 deprecated; entries are mirrored to agents/<id>/manifest.toml \
                 and the legacy file will stop being read in a future release."
            );
        }
        for spec in legacy.agents {
            by_id.insert(spec.id.clone(), spec);
        }

        // New layout overrides legacy on conflict — once a manifest
        // file exists for an id, it's the source of truth.
        let manifests = crate::manifest::load_manifest_dir(&agents_dir)?;
        for m in &manifests {
            // Skip pure_data + health_url-only entries: the
            // supervisor only spawns command-shaped externals in
            // phase 1. They stay registered in the directory but
            // don't become AgentSlots.
            if m.is_pure_data() || m.is_remote_service() {
                continue;
            }
            match crate::manifest::to_agent_spec(m) {
                Ok(spec) => {
                    by_id.insert(spec.id.clone(), spec);
                }
                Err(e) => {
                    tracing::warn!(
                        manifest_id = %m.agent.id,
                        error = %e,
                        "manifest.toml could not project to an AgentSpec; \
                         agent will not be supervised this boot"
                    );
                }
            }
        }

        // Mirror legacy-only entries into the new directory layout.
        // Idempotent — entries that already have a manifest.toml
        // skip the write. This is the migration: subsequent boots
        // find them via both sources, and the legacy file becomes
        // a read-only deprecation surface.
        let existing_manifest_ids: std::collections::HashSet<&str> =
            manifests.iter().map(|m| m.agent.id.as_str()).collect();
        for spec in by_id.values() {
            if existing_manifest_ids.contains(spec.id.as_str()) {
                continue;
            }
            let m = crate::manifest::from_legacy_spec(spec);
            if let Err(e) = crate::manifest::write_manifest(&agents_dir, &m) {
                tracing::warn!(
                    id = %spec.id,
                    error = %e,
                    "failed to mirror legacy AgentSpec to manifest.toml; \
                     entry remains in agents.json only"
                );
            }
        }

        let mut state: HashMap<String, AgentSlot> = HashMap::new();
        for (id, spec) in by_id {
            state.insert(
                id,
                AgentSlot {
                    spec,
                    runtime: AgentRuntime::default(),
                    stop_tx: None,
                    task: None,
                    job: None,
                },
            );
        }
        Ok(Self {
            manifest_path,
            log_dir,
            state: Arc::new(RwLock::new(state)),
            manifest_lock: Arc::new(Mutex::new(())),
            _process_lock: Arc::new(lock_file),
            grace_secs: 10,
            default_child_env: Arc::new(RwLock::new(BTreeMap::new())),
        })
    }

    /// Replace the default-child-env table. Subsequent spawns inherit
    /// these env vars; per-spec `spec.env` is still merged on top and
    /// wins on conflict. Called once by car-server at boot to inject
    /// `CAR_DAEMON_URL` + `CAR_AUTH_TOKEN` (and, once #169 lands,
    /// `CAR_AGENT_ID` / `CAR_AGENT_TOKEN`).
    pub async fn set_default_child_env<I, K, V>(&self, entries: I)
    where
        I: IntoIterator<Item = (K, V)>,
        K: Into<String>,
        V: Into<String>,
    {
        let mut g = self.default_child_env.write().await;
        g.clear();
        for (k, v) in entries {
            g.insert(k.into(), v.into());
        }
    }

    /// Read-only snapshot of the default-child-env table. Used by
    /// `spawn_child` and exposed for tests.
    pub async fn default_child_env(&self) -> BTreeMap<String, String> {
        self.default_child_env.read().await.clone()
    }

    /// Path to the on-disk manifest.
    pub fn manifest_path(&self) -> &Path {
        &self.manifest_path
    }

    /// Directory log files are written under.
    pub fn log_dir(&self) -> &Path {
        &self.log_dir
    }

    /// Snapshot every managed agent. Sorted by id for deterministic
    /// UI ordering.
    pub async fn list(&self) -> Vec<ManagedAgent> {
        let state = self.state.read().await;
        let mut out: Vec<ManagedAgent> = state
            .values()
            .map(|slot| self.snapshot_locked(slot))
            .collect();
        out.sort_by(|a, b| a.spec.id.cmp(&b.spec.id));
        out
    }

    /// Snapshot one agent's spec + live runtime, or `None` if absent.
    pub async fn get(&self, id: &str) -> Option<ManagedAgent> {
        self.state
            .read()
            .await
            .get(id)
            .map(|slot| self.snapshot_locked(slot))
    }

    /// Block until agent `id`'s status is one of `targets`, or `timeout`
    /// elapses. Polls the live in-memory runtime every `poll_interval` (clamped
    /// to at least 10ms and to the time remaining). Returns the matching
    /// snapshot, [`SupervisorError::WaitTimeout`] if the deadline passes first,
    /// or [`SupervisorError::NotFound`] if the agent leaves the manifest.
    ///
    /// This is the persistent-agent analogue of a "wait until ready / wait until
    /// done" primitive: pass `[Running]` to wait for an agent to come up, or
    /// `[Stopped, Errored]` to wait for a one-shot child to finish. The
    /// supervisor's own restart loop drives the status transitions; this only
    /// observes them, so it never itself starts or stops the agent.
    ///
    /// Because it polls, a status the agent passes *through* between two polls
    /// (e.g. a child that races `Starting → Running → Stopped` faster than
    /// `poll_interval`) can be missed. That's fine for the intended uses —
    /// waiting for a long-lived agent's `Running`, or a one-shot's terminal
    /// `Stopped`/`Errored`, both of which are stable once reached.
    pub async fn wait_for(
        &self,
        id: &str,
        targets: &[AgentStatus],
        timeout: std::time::Duration,
        poll_interval: std::time::Duration,
    ) -> Result<ManagedAgent, SupervisorError> {
        let deadline = tokio::time::Instant::now() + timeout;
        let poll = poll_interval.max(std::time::Duration::from_millis(10));
        loop {
            let snap = self
                .get(id)
                .await
                .ok_or_else(|| SupervisorError::NotFound(id.to_string()))?;
            if targets.contains(&snap.status) {
                return Ok(snap);
            }
            let now = tokio::time::Instant::now();
            if now >= deadline {
                return Err(SupervisorError::WaitTimeout {
                    id: id.to_string(),
                    last: snap.status,
                    timeout,
                });
            }
            tokio::time::sleep(poll.min(deadline - now)).await;
        }
    }

    /// Re-validate the `command` of every managed agent. Useful after
    /// a system upgrade (Node moved between minor versions, Homebrew
    /// pruned a symlink) to surface broken specs before the next
    /// `start` does. Returns one [`AgentHealth`] per agent, sorted by
    /// id for stable UI ordering.
    pub async fn health(&self) -> Vec<AgentHealth> {
        let state = self.state.read().await;
        let mut out: Vec<AgentHealth> = state
            .values()
            .map(|slot| {
                let command = slot.spec.command.clone();
                match validate_command(&command) {
                    Ok(()) => AgentHealth {
                        id: slot.spec.id.clone(),
                        command,
                        ok: true,
                        reason: None,
                    },
                    Err(e) => AgentHealth {
                        id: slot.spec.id.clone(),
                        command,
                        ok: false,
                        reason: Some(e.to_string()),
                    },
                }
            })
            .collect();
        out.sort_by(|a, b| a.id.cmp(&b.id));
        out
    }

    /// Add or replace an agent's spec. Persists the manifest. The
    /// agent is NOT auto-started by this method — call
    /// [`Supervisor::start`] (or [`Supervisor::start_all`] on next
    /// boot).
    ///
    /// Both the id and the command are validated up front. The
    /// command must be an absolute path to an existing executable
    /// outside world-writable scratch directories — see
    /// [`validate_command`] for the full rule set. A spec that
    /// fails validation is rejected without touching disk or
    /// in-memory state.
    pub async fn upsert(&self, mut spec: AgentSpec) -> Result<ManagedAgent, SupervisorError> {
        validate_id(&spec.id)?;
        validate_command(&spec.command)?;
        {
            let mut state = self.state.write().await;
            // Token policy (#169): mint on first upsert, retain on
            // re-upsert unless the caller passed a non-empty value
            // (explicit rotation). An incoming empty token on an
            // existing entry keeps the prior token — saves the
            // operator from having to refetch + replay it.
            if spec.token.is_empty() {
                if let Some(existing) = state.get(&spec.id) {
                    if !existing.spec.token.is_empty() {
                        spec.token = existing.spec.token.clone();
                    }
                }
                if spec.token.is_empty() {
                    spec.token = mint_agent_token();
                }
            }
            if let Some(existing) = state.get_mut(&spec.id) {
                existing.spec = spec.clone();
            } else {
                state.insert(
                    spec.id.clone(),
                    AgentSlot {
                        spec: spec.clone(),
                        runtime: AgentRuntime::default(),
                        stop_tx: None,
                        task: None,
                        job: None,
                    },
                );
            }
        }
        self.persist().await?;
        Ok(ManagedAgent {
            spec,
            status: AgentStatus::Stopped,
            pid: None,
            last_exit_code: None,
            restart_count: 0,
            started_at: None,
        })
    }

    /// Install a contributed-agent manifest (Parslee-ai/car#182
    /// phase 3). Runs the install-time validator
    /// (`car_min_version`, capability negotiation, optional-cap
    /// reporting), then projects the manifest into an `AgentSpec`
    /// and adopts it via `upsert`. Pure-data + health_url-only
    /// manifests are tracked on disk but not adopted into the
    /// spawnable set — phase 1's projection rules still apply.
    ///
    /// Returns the install report on success so callers can warn
    /// users about missing optional capabilities. Returns an
    /// error on any blocker (version mismatch, required
    /// capability missing, signature failure when phase 3
    /// strictness lands).
    pub async fn install_manifest(
        &self,
        manifest: AgentManifest,
        host: &crate::install::HostCapabilities,
    ) -> Result<(crate::install::InstallCheckReport, Option<ManagedAgent>), SupervisorError> {
        let report = crate::install::install_check(&manifest, host)?;
        if manifest.is_pure_data() || manifest.is_remote_service() {
            // Track on disk but don't adopt. Operators can still
            // see + manage the manifest via the registry surface;
            // the supervisor just doesn't spawn it.
            let agents_dir = self
                .manifest_path
                .parent()
                .map(|p| p.join("agents"))
                .unwrap_or_else(|| PathBuf::from("agents"));
            std::fs::create_dir_all(&agents_dir)?;
            crate::manifest::write_manifest(&agents_dir, &manifest)?;
            return Ok((report, None));
        }
        let mut spec = crate::manifest::to_agent_spec(&manifest)?;
        // Preserve the manifest's identity bits that AgentSpec
        // doesn't currently carry — version-aware addressing
        // resolves via the on-disk manifest tree, not the
        // in-memory AgentSpec. We still mint a fresh token if the
        // manifest didn't carry one (legacy install paths).
        if spec.token.is_empty() {
            spec.token = mint_agent_token();
        }
        let managed = self.upsert(spec).await?;
        if managed.spec.auto_start {
            let started = self.start(&managed.spec.id).await?;
            Ok((report, Some(started)))
        } else {
            Ok((report, Some(managed)))
        }
    }

    /// Return the per-agent token for `id`, or `None` if no such
    /// agent is supervised or the token field is empty. Used by the
    /// daemon's `session.auth` handler to validate
    /// `agent_id` + `token` pairs (#169).
    pub async fn agent_token(&self, id: &str) -> Option<String> {
        let state = self.state.read().await;
        let slot = state.get(id)?;
        if slot.spec.token.is_empty() {
            None
        } else {
            Some(slot.spec.token.clone())
        }
    }

    /// Constant-time check that `token` matches the stored token for
    /// `id`. Returns `false` when `id` is unknown.
    pub async fn validate_agent_token(&self, id: &str, token: &str) -> bool {
        let Some(stored) = self.agent_token(id).await else {
            return false;
        };
        constant_time_eq(stored.as_bytes(), token.as_bytes())
    }

    /// Remove an agent's spec. Stops the running child first if it's
    /// up. Idempotent — `Ok(false)` when nothing matched.
    pub async fn remove(&self, id: &str) -> Result<bool, SupervisorError> {
        validate_id(id)?;
        // Stop first so the supervisor task isn't left dangling.
        let _ = self.stop(id, StopSignal::Term).await;
        let removed = {
            let mut state = self.state.write().await;
            state.remove(id).is_some()
        };
        if removed {
            self.persist().await?;
        }
        Ok(removed)
    }

    /// Spawn the agent's child if it isn't already running. No-op
    /// when the agent is currently `Running` or `Starting`. Resets
    /// `restart_count` on every manual start.
    pub async fn start(&self, id: &str) -> Result<ManagedAgent, SupervisorError> {
        validate_id(id)?;
        let spec = {
            let state = self.state.read().await;
            let slot = state
                .get(id)
                .ok_or_else(|| SupervisorError::NotFound(id.to_string()))?;
            if matches!(
                slot.runtime.status,
                AgentStatus::Running | AgentStatus::Starting
            ) {
                return Ok(self.snapshot_locked(slot));
            }
            slot.spec.clone()
        };
        self.spawn_supervision(spec).await;
        // Brief pause so the spawned task has a chance to flip
        // status to Starting before we report. Not load-bearing —
        // callers re-poll via `list`.
        tokio::task::yield_now().await;
        Ok(self.get(id).await.unwrap_or_else(|| ManagedAgent {
            spec: AgentSpec {
                id: id.to_string(),
                name: id.to_string(),
                command: String::new(),
                args: vec![],
                cwd: None,
                env: BTreeMap::new(),
                restart: RestartPolicy::default(),
                max_restarts: default_max_restarts(),
                backoff_secs: default_backoff(),
                auto_start: false,
                token: String::new(),
            },
            status: AgentStatus::Starting,
            pid: None,
            last_exit_code: None,
            restart_count: 0,
            started_at: None,
        }))
    }

    /// Stop the agent and prevent the supervisor from respawning it.
    /// `Term` sends SIGTERM and waits up to `grace_secs` before
    /// escalating to SIGKILL; `Kill` skips the grace.
    pub async fn stop(
        &self,
        id: &str,
        signal: StopSignal,
    ) -> Result<ManagedAgent, SupervisorError> {
        // Preserve NotFound semantics: a stop against an unknown id
        // is an error, not a silent no-op.
        {
            let state = self.state.read().await;
            if !state.contains_key(id) {
                return Err(SupervisorError::NotFound(id.to_string()));
            }
        }
        // Signal the loop to exit, kill the child (cascade-kill the
        // entire process tree on Windows via the Job Object — see
        // #231 §5.1), abort the task.
        self.teardown_running(id, signal).await;
        {
            let mut state = self.state.write().await;
            if let Some(slot) = state.get_mut(id) {
                slot.runtime.status = AgentStatus::Stopped;
                slot.runtime.pid = None;
                slot.runtime.started_at = None;
            }
        }
        self.get(id)
            .await
            .ok_or_else(|| SupervisorError::NotFound(id.to_string()))
    }

    /// Stop then start. Resets `restart_count` via the [`Supervisor::start`]
    /// path.
    pub async fn restart(&self, id: &str) -> Result<ManagedAgent, SupervisorError> {
        let _ = self.stop(id, StopSignal::Term).await;
        self.start(id).await
    }

    /// Spawn every manifest agent whose `auto_start` is true. Used
    /// by car-server's main on boot. Returns the ids spawned.
    ///
    /// Skips agents whose pid file at `~/.car/run/<id>.pid` references
    /// a live process — that signals an instance is already running
    /// outside this supervisor (orphaned from a previous car-server,
    /// running under a different supervisor, etc.). Auto-starting a
    /// second instance in that case has caused real production
    /// damage: two trader processes trading the same account when a
    /// car-server restart left the prior trader as a launchd-orphan
    /// and the new car-server unaware of it.
    ///
    /// The pid-file convention is opt-in per agent: agents that want
    /// double-spawn protection write `~/.car/run/<id>.pid` themselves
    /// at startup (e.g. trader does this in `src/daemon.js`). Agents
    /// that don't write the file fall through to the existing spawn
    /// behavior unchanged.
    pub async fn start_all(&self) -> Vec<String> {
        let candidates: Vec<AgentSpec> = {
            let state = self.state.read().await;
            state
                .values()
                .filter(|slot| {
                    slot.spec.auto_start
                        && !matches!(
                            slot.runtime.status,
                            AgentStatus::Running | AgentStatus::Starting
                        )
                })
                .map(|slot| slot.spec.clone())
                .collect()
        };
        let mut started = Vec::with_capacity(candidates.len());
        for spec in candidates {
            if let Some(ext_pid) = external_agent_pid(&spec.id) {
                tracing::warn!(
                    agent = %spec.id,
                    pid = ext_pid,
                    "agent already running externally (pid file at ~/.car/run/{}.pid). Skipping auto_start — call agents.start once the external instance exits to take over supervision.",
                    spec.id
                );
                continue;
            }
            let id = spec.id.clone();
            self.spawn_supervision(spec).await;
            started.push(id);
        }
        started
    }

    /// Read the last `n` lines from the agent's combined log.
    ///
    /// Backward-compatible thin wrapper over [`Supervisor::read_log`]:
    /// returns stdout then stderr, but — unlike the old naive version
    /// — each stream is tailed to its own `n`-line budget rather than
    /// concatenating the *whole* of one file in front of the other.
    /// That fix matters: a long stale stderr can no longer bury the
    /// live stdout of a healthy agent (Parslee-ai/car#273). Returns an
    /// empty `Vec` when neither log exists yet.
    pub async fn tail_log(&self, id: &str, n: usize) -> Result<Vec<String>, SupervisorError> {
        let tail = self.read_log(id, LogStream::Combined, n, 0).await?;
        Ok(tail.lines)
    }

    /// Read a window of an agent's logs with stream selection and
    /// paging.
    ///
    /// The capture format is raw child output — there are no per-line
    /// timestamps, so true cross-stream timestamp interleaving isn't
    /// possible without changing how we write the files. Instead each
    /// stream is tailed independently to its own `n`-line budget, and
    /// the result exposes per-stream tails plus file paths and total
    /// line counts so a viewer can show stdout and stderr in separate
    /// panes, page back through history (`offset` lines from the end),
    /// "load more", or reveal the underlying file. This is what makes
    /// a long stderr stop hiding a healthy agent's live stdout
    /// (Parslee-ai/car#273).
    ///
    /// - `stream` selects which stream(s) to read.
    /// - `n` caps lines *per included stream* (`0` ⇒ no per-line cap —
    ///   the whole file, still bounded by the tail byte ceiling; see
    ///   [`read_stream_window`]).
    /// - `offset` skips that many lines from the end of each stream
    ///   before taking the window, so `(offset=n)` pages back one
    ///   screen. Applied per-stream.
    ///
    /// Each stream is read via a bounded backward seek — at most
    /// [`LOG_TAIL_BYTE_CEILING`] bytes from the end — not a whole-file
    /// slurp. These logs are append-only and never rotated, so a
    /// crash-looping agent can produce a multi-GB file; reading only the
    /// needed window keeps the Follow poll's cost bounded.
    pub async fn read_log(
        &self,
        id: &str,
        stream: LogStream,
        n: usize,
        offset: usize,
    ) -> Result<LogTail, SupervisorError> {
        validate_id(id)?;
        let stdout_path = self.log_dir.join(format!("{id}.stdout.log"));
        let stderr_path = self.log_dir.join(format!("{id}.stderr.log"));

        let want_stdout = matches!(stream, LogStream::Stdout | LogStream::Combined);
        let want_stderr = matches!(stream, LogStream::Stderr | LogStream::Combined);

        let stdout = if want_stdout {
            read_stream_window(&stdout_path, n, offset).await?
        } else {
            StreamWindow::default()
        };
        let stderr = if want_stderr {
            read_stream_window(&stderr_path, n, offset).await?
        } else {
            StreamWindow::default()
        };

        // Combined `lines` keeps the legacy ordering contract (stdout
        // then stderr) but each side is now independently budgeted, so
        // neither buries the other.
        let mut lines = Vec::with_capacity(stdout.lines.len() + stderr.lines.len());
        lines.extend(stdout.lines.iter().cloned());
        lines.extend(stderr.lines.iter().cloned());

        // "more" is true when paging further back would surface older
        // lines on any included stream.
        let more = stdout.more || stderr.more;

        Ok(LogTail {
            lines,
            stdout: stdout.lines,
            stderr: stderr.lines,
            stdout_total: stdout.total,
            stderr_total: stderr.total,
            stdout_path: stdout_path.to_string_lossy().into_owned(),
            stderr_path: stderr_path.to_string_lossy().into_owned(),
            more,
        })
    }

    fn snapshot_locked(&self, slot: &AgentSlot) -> ManagedAgent {
        ManagedAgent {
            spec: slot.spec.clone(),
            status: slot.runtime.status,
            pid: slot.runtime.pid,
            last_exit_code: slot.runtime.last_exit_code,
            restart_count: slot.runtime.restart_count,
            started_at: slot.runtime.started_at,
        }
    }

    async fn persist(&self) -> Result<(), SupervisorError> {
        let _g = self.manifest_lock.lock().await;
        let (manifest, current_ids): (Manifest, std::collections::HashSet<String>) = {
            let state = self.state.read().await;
            let mut agents: Vec<AgentSpec> = state.values().map(|slot| slot.spec.clone()).collect();
            agents.sort_by(|a, b| a.id.cmp(&b.id));
            let ids: std::collections::HashSet<String> =
                agents.iter().map(|s| s.id.clone()).collect();
            (
                Manifest {
                    agents: agents.clone(),
                },
                ids,
            )
        };
        // Dual-write during the migration window (Parslee-ai/car#182
        // phase 1): legacy JSON stays the canonical read source for
        // one more minor release, but every persist also mirrors to
        // `agents/<id>/manifest.toml` so the new layout never
        // drifts behind the legacy file. Phase N+2 deletes this
        // legacy write.
        write_json_atomic(&self.manifest_path, &manifest)?;
        let agents_dir = self
            .manifest_path
            .parent()
            .map(|p| p.join("agents"))
            .unwrap_or_else(|| PathBuf::from("agents"));
        if let Err(e) = std::fs::create_dir_all(&agents_dir) {
            tracing::warn!(
                dir = %agents_dir.display(),
                error = %e,
                "could not create agents/ dir for manifest mirror"
            );
            return Ok(());
        }
        // Write a manifest.toml per current AgentSpec.
        for spec in &manifest.agents {
            let m = crate::manifest::from_legacy_spec(spec);
            if let Err(e) = crate::manifest::write_manifest(&agents_dir, &m) {
                tracing::warn!(
                    id = %spec.id,
                    error = %e,
                    "mirroring AgentSpec to manifest.toml failed; legacy \
                     agents.json was still updated"
                );
            }
        }
        // Reap manifest dirs whose ids are no longer in state. Only
        // remove dirs we know we own — skip anything that doesn't
        // look like a supervised-agent layout (i.e., must contain a
        // manifest.toml).
        if let Ok(entries) = std::fs::read_dir(&agents_dir) {
            for entry in entries.flatten() {
                let p = entry.path();
                if !p.is_dir() {
                    continue;
                }
                let Some(name) = p.file_name().and_then(|s| s.to_str()) else {
                    continue;
                };
                if current_ids.contains(name) {
                    continue;
                }
                if p.join("manifest.toml").is_file() {
                    if let Err(e) = std::fs::remove_dir_all(&p) {
                        tracing::warn!(
                            dir = %p.display(),
                            error = %e,
                            "reaping stale manifest dir failed"
                        );
                    }
                }
            }
        }
        Ok(())
    }

    async fn spawn_supervision(&self, spec: AgentSpec) {
        // Teardown-first. If a loop is already running for this id —
        // even one mid-backoff that hasn't respawned yet, or a
        // healthy child whose `start` raced past the status guard —
        // signal it to stop, kill its child, and abort the task
        // before we install the new loop. Skipping this let a second
        // loop start alongside the first; the two raced on shared
        // resources (a TCP port), one won and ran on as an untracked
        // orphan while the other stormed to `Errored`. Idempotent and
        // a near-instant no-op when nothing is running (no stop_tx /
        // pid / task to reap).
        self.teardown_running(&spec.id, StopSignal::Term).await;
        let (tx, rx) = tokio::sync::watch::channel(false);
        let task = tokio::spawn(supervisor_loop(self.clone(), spec.clone(), rx));
        let mut state = self.state.write().await;
        if let Some(slot) = state.get_mut(&spec.id) {
            slot.runtime.status = AgentStatus::Starting;
            slot.runtime.restart_count = 0;
            slot.stop_tx = Some(tx);
            slot.task = Some(task);
        }
    }

    /// Tear down any live supervisor loop + child for `id` without
    /// touching the spec or the slot itself. Takes the stop sender,
    /// task handle, and pid out of the slot, signals the loop to
    /// stop, kills the child (graceful for `Term`), then aborts the
    /// task. Shared by [`Supervisor::stop`] and
    /// [`Supervisor::spawn_supervision`] so a (re)start can never run
    /// the old loop alongside the new one. A no-op when the slot is
    /// absent or already idle.
    async fn teardown_running(&self, id: &str, signal: StopSignal) {
        // Collect everything we need under the lock in one go:
        // - stop_tx / task: the supervision loop's exit signal + handle
        // - pid: for the kill call
        // - job: Windows Job Object handle (None elsewhere; None on
        //   Windows if spawn-time Job Object setup failed). Taken
        //   here so it doesn't outlive the kill — when the Arc drops
        //   at the end of this function, KILL_ON_JOB_CLOSE acts as
        //   the safety net even if TerminateJobObject failed. See
        //   #231 §5.1 and the JobObject wrapper.
        // - started_at: spawn timestamp, used by the Windows fallback
        //   path (job=None) to verify the pid hasn't been recycled
        //   before calling TerminateProcess. Without this, a
        //   supervisor whose Job Object setup failed could terminate
        //   an unrelated process that happens to hold the recycled
        //   pid.
        let (stop_tx, task, pid, job, started_at) = {
            let mut state = self.state.write().await;
            match state.get_mut(id) {
                Some(slot) => (
                    slot.stop_tx.take(),
                    slot.task.take(),
                    slot.runtime.pid,
                    slot.job.take(),
                    slot.runtime.started_at,
                ),
                None => return,
            }
        };
        if let Some(tx) = stop_tx {
            let _ = tx.send(true);
        }
        if let Some(pid) = pid {
            kill_process(pid, signal, self.grace_secs, job, started_at).await;
        }
        if let Some(handle) = task {
            // Don't await — the task may be mid-backoff sleep.
            // Aborting is cleaner than blocking the caller; the
            // `kill_on_drop` backstop on the child covers the case
            // where the abort drops a `Child` we hadn't recorded a
            // pid for yet.
            handle.abort();
        }
    }

    async fn set_status(
        &self,
        id: &str,
        status: AgentStatus,
        pid: Option<u32>,
        started_at: Option<i64>,
    ) {
        let mut state = self.state.write().await;
        if let Some(slot) = state.get_mut(id) {
            // Clear the prior failure code on a successful (re)start.
            // Without this, `agents.list` advertises `last_exit_code: 1`
            // forever after one crash even though the agent is currently
            // up, which confuses operators triaging "is this thing
            // healthy right now?" The exit code is still recorded in
            // tracing when the actual exit happens.
            if matches!(status, AgentStatus::Running) {
                slot.runtime.last_exit_code = None;
            }
            slot.runtime.status = status;
            slot.runtime.pid = pid;
            slot.runtime.started_at = started_at;
        }
    }

    /// Store the spawn-time Windows Job Object handle in the agent's
    /// slot. Called immediately before `set_status(Running, ...)` so
    /// the slot has a complete picture of the running child before
    /// any concurrent `stop()` call could observe it. No-op when
    /// `job` is `None` (off Windows, or Windows where Job Object
    /// setup failed). See #231 §5.1.
    async fn store_job(&self, id: &str, job: Option<Arc<JobObject>>) {
        let mut state = self.state.write().await;
        if let Some(slot) = state.get_mut(id) {
            slot.job = job;
        }
    }

    async fn record_exit(&self, id: &str, exit_code: i32) {
        let mut state = self.state.write().await;
        if let Some(slot) = state.get_mut(id) {
            slot.runtime.last_exit_code = Some(exit_code);
            slot.runtime.pid = None;
            slot.runtime.started_at = None;
            // NOTE: we intentionally do NOT clear `slot.job` here.
            //
            // §5.1 is about `car stop` cascading to grandchildren —
            // that's what TerminateJobObject in `kill_process` handles.
            // It does NOT specify behavior on *natural* process exit.
            // Unix's pre-existing behavior is to let SIGCHLD'd
            // grandchildren keep running (this is a legitimate
            // pattern — `parent.exe → daemon.exe → detaches and
            // self-supervises` is how many Windows installers work).
            // Clearing slot.job here would drop the JobObject Arc,
            // which on Windows triggers KILL_ON_JOB_CLOSE and kills
            // those detached grandchildren — a semantic divergence
            // from Unix and from the §5.1 scope.
            //
            // The next `store_job` (on respawn via the supervision
            // loop) replaces the Arc and drops the old one, which
            // *does* cascade-kill any zombies from the prior
            // lifecycle. That's the right time to clean up — we're
            // about to spawn a fresh process tree and zombies from
            // the dead generation are unambiguously stale.
            //
            // `Supervisor::stop` also takes `slot.job`, so a manual
            // stop while the agent is in a restart-backoff window
            // correctly cascade-kills any in-flight zombies.
            //
            // On non-Windows builds slot.job is always None and this
            // entire concern is moot.
        }
    }

    async fn bump_restart(&self, id: &str) -> u32 {
        let mut state = self.state.write().await;
        if let Some(slot) = state.get_mut(id) {
            slot.runtime.restart_count = slot.runtime.restart_count.saturating_add(1);
            slot.runtime.status = AgentStatus::Backoff;
            slot.runtime.restart_count
        } else {
            0
        }
    }

    async fn mark_errored(&self, id: &str) {
        let mut state = self.state.write().await;
        if let Some(slot) = state.get_mut(id) {
            slot.runtime.status = AgentStatus::Errored;
        }
    }
}

async fn supervisor_loop(
    supervisor: Supervisor,
    spec: AgentSpec,
    mut stop_rx: tokio::sync::watch::Receiver<bool>,
) {
    let id = spec.id.clone();
    loop {
        if *stop_rx.borrow() {
            return;
        }
        // Pre-spawn pid-file check. start_all() already filters at
        // boot, but agents.start() also routes through this loop
        // (via spawn_supervision), and after a backoff sleep the
        // external situation may have changed. Re-check each
        // iteration so the supervisor can take over cleanly the
        // moment the external instance exits.
        if let Some(ext_pid) = external_agent_pid(&spec.id) {
            tracing::warn!(
                agent = %id,
                pid = ext_pid,
                "external agent instance still alive (pid file). Supervisor refusing to double-spawn; sleeping {}s then re-checking.",
                spec.backoff_secs.max(5)
            );
            supervisor
                .set_status(&id, AgentStatus::Backoff, None, None)
                .await;
            let backoff = std::time::Duration::from_secs(spec.backoff_secs.max(5));
            tokio::select! {
                _ = stop_rx.changed() => return,
                _ = tokio::time::sleep(backoff) => continue,
            }
        }
        let default_env = supervisor.default_child_env().await;
        match spawn_child(&supervisor.log_dir, &spec, &default_env).await {
            Ok(SpawnedChild {
                mut child,
                pid,
                job,
            }) => {
                let started_at = chrono::Utc::now().timestamp();
                // Store the Windows Job Object handle in the slot so
                // `Supervisor::stop` can find it. No-op on non-Windows
                // (job is always `None` there). Done before the status
                // update so the kill path can't observe a Running pid
                // without its job handle in the rare interleaving
                // where stop() reads state right after set_status.
                supervisor.store_job(&id, job).await;
                supervisor
                    .set_status(&id, AgentStatus::Running, Some(pid), Some(started_at))
                    .await;
                tokio::select! {
                    biased;
                    _ = stop_rx.changed() => {
                        // Outer stop won; the explicit kill happens
                        // in `Supervisor::stop`. Try waitpid briefly
                        // so we don't leave a zombie if the kill
                        // already landed.
                        let _ = child.wait().await;
                        return;
                    }
                    res = child.wait() => {
                        let code = match res {
                            Ok(status) => status.code().unwrap_or(-1),
                            Err(_) => -1,
                        };
                        supervisor.record_exit(&id, code).await;
                        let should_restart = match spec.restart {
                            RestartPolicy::Never => false,
                            RestartPolicy::OnFailure => code != 0,
                            RestartPolicy::Always => true,
                        };
                        if !should_restart {
                            supervisor.set_status(&id, AgentStatus::Stopped, None, None).await;
                            return;
                        }
                        let count = supervisor.bump_restart(&id).await;
                        if count > spec.max_restarts {
                            tracing::warn!(agent = %id, count, max = spec.max_restarts,
                                "agent exceeded max_restarts; marking errored");
                            supervisor.mark_errored(&id).await;
                            return;
                        }
                        let backoff = restart_backoff(spec.backoff_secs, count);
                        tokio::select! {
                            _ = stop_rx.changed() => return,
                            _ = tokio::time::sleep(backoff) => {}
                        }
                    }
                }
            }
            Err(e) => {
                tracing::error!(agent = %id, error = %e, "spawn failed");
                supervisor.record_exit(&id, -1).await;
                let count = supervisor.bump_restart(&id).await;
                if count > spec.max_restarts {
                    supervisor.mark_errored(&id).await;
                    return;
                }
                let backoff = restart_backoff(spec.backoff_secs, count);
                tokio::select! {
                    _ = stop_rx.changed() => return,
                    _ = tokio::time::sleep(backoff) => {}
                }
            }
        }
    }
}

/// Hard ceiling on a single restart backoff. Exponential growth
/// stops doubling here so a long-lived crash loop settles into a
/// steady, calm retry cadence rather than hour-long sleeps.
const BACKOFF_CAP_SECS: u64 = 60;

/// Backoff before the next restart attempt. Exponential in the
/// consecutive-restart `attempt` (1-based) on top of the spec's
/// `backoff_secs` floor, capped at [`BACKOFF_CAP_SECS`], with a
/// small additive jitter.
///
/// This replaced a flat `backoff_secs` sleep. The flat delay let a
/// crash-looping agent — e.g. one whose listen port was already
/// held by an orphan — burn through `max_restarts` in well under a
/// minute (5s × 10 ≈ 50s) and slam straight into `Errored`.
/// Doubling (5s, 10s, 20s, 40s, 60s, 60s, …) turns the same loop
/// into a widening retry that gives a transient conflict time to
/// clear. Jitter (≤ ~12% of the delay) desynchronises multiple
/// agents that fail together so they don't reconverge into a
/// thundering restart herd. Derived from the wall clock to avoid
/// pulling in an RNG dependency; it's purely additive and never
/// shortens the floor.
fn restart_backoff(base_secs: u64, attempt: u32) -> std::time::Duration {
    let base = base_secs.max(1);
    // Clamp the shift well below 64 so `1 << shift` can't overflow
    // or panic; the cap below makes anything past a handful of
    // attempts moot anyway.
    let shift = attempt.saturating_sub(1).min(16);
    let grown = base.saturating_mul(1u64 << shift).min(BACKOFF_CAP_SECS);
    let jitter_ceiling_ms = grown.saturating_mul(1000) / 8;
    let jitter_ms = if jitter_ceiling_ms == 0 {
        0
    } else {
        (jitter_nanos() % u128::from(jitter_ceiling_ms)) as u64
    };
    std::time::Duration::from_secs(grown) + std::time::Duration::from_millis(jitter_ms)
}

/// Cheap, dependency-free entropy for backoff jitter: the
/// sub-second nanosecond component of the wall clock. Not
/// cryptographic and not meant to be — it only needs to vary
/// enough between two agents' restart timings to break lockstep.
fn jitter_nanos() -> u128 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| u128::from(d.subsec_nanos()))
        .unwrap_or(0)
}

/// Result of [`spawn_child`]. The `job` field is `Some` only on
/// Windows builds where the Job Object setup succeeded; it's the
/// cascade-kill handle used by [`kill_process`] to atomically
/// terminate the entire supervised tree on stop. See #231 §5.1.
struct SpawnedChild {
    child: tokio::process::Child,
    pid: u32,
    job: Option<Arc<JobObject>>,
}

async fn spawn_child(
    log_dir: &Path,
    spec: &AgentSpec,
    default_env: &BTreeMap<String, String>,
) -> std::io::Result<SpawnedChild> {
    use std::process::Stdio;
    use tokio::process::Command;

    let stdout_path = log_dir.join(format!("{}.stdout.log", spec.id));
    let stderr_path = log_dir.join(format!("{}.stderr.log", spec.id));
    let stdout = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(&stdout_path)?;
    let stderr = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(&stderr_path)?;

    let mut cmd = Command::new(&spec.command);
    cmd.args(&spec.args);
    if let Some(cwd) = &spec.cwd {
        cmd.current_dir(cwd);
    }
    // Default env (CAR_DAEMON_URL / CAR_AUTH_TOKEN — #172) goes in
    // first so the per-spec env below wins on conflict. Operators
    // who want to override the daemon-supplied URL (e.g. point a
    // child at a different car-server during development) can still
    // do so via `spec.env`.
    for (k, v) in default_env {
        cmd.env(k, v);
    }
    // Per-agent identity (#169). Always set — even when the daemon
    // is `--no-auth` and `spec.token` was minted but is unused. The
    // child can still call `session.auth { agent_id }` to bind its
    // connection to its supervised identity.
    cmd.env("CAR_AGENT_ID", &spec.id);
    if !spec.token.is_empty() {
        cmd.env("CAR_AGENT_TOKEN", &spec.token);
    }
    for (k, v) in &spec.env {
        cmd.env(k, v);
    }
    cmd.stdin(Stdio::null());
    cmd.stdout(Stdio::from(stdout));
    cmd.stderr(Stdio::from(stderr));
    // Hard backstop against orphaned children: if the supervisor
    // loop future is ever dropped (task aborted by `stop`, or the
    // slot torn down by a re-`start`) while this child is still
    // alive, tokio SIGKILLs it on `Child` drop. Without this, an
    // abort dropped the `Child` without signalling the process, so
    // a `start` issued mid-backoff could leave the prior instance
    // running untracked — the bug that left a live agent bound to
    // its port while the slot reported `Errored`. The graceful
    // SIGTERM path in `stop`/`teardown_running` still runs first;
    // this only catches the drop that would otherwise leak.
    cmd.kill_on_drop(true);
    // Detach from the parent's controlling terminal so SIGINT to
    // the supervisor doesn't propagate to children automatically;
    // we control kills via SIGTERM/SIGKILL.
    #[cfg(unix)]
    unsafe {
        // tokio::process::Command exposes pre_exec directly on
        // unix; setsid puts the child in its own process group so
        // signals to the supervisor don't propagate to it through
        // the controlling terminal.
        cmd.pre_exec(|| {
            if libc_setsid() == -1 {
                return Err(std::io::Error::last_os_error());
            }
            Ok(())
        });
    }
    let child = cmd.spawn()?;
    let pid = child.id().ok_or_else(|| {
        std::io::Error::new(std::io::ErrorKind::Other, "child spawned without pid")
    })?;

    // Windows-only: assign the new child to a Job Object so that
    // `car stop` can later kill the entire spawned tree atomically
    // via TerminateJobObject. Failures here degrade to the
    // pre-§5.1 behavior (single-pid TerminateProcess, no cascade);
    // they're logged but don't fail the spawn — the agent should
    // still get to run, even if shutdown will leak grandchildren.
    #[cfg(target_os = "windows")]
    let job = match JobObject::new() {
        Ok(j) => match j.assign(pid) {
            Ok(()) => Some(Arc::new(j)),
            Err(e) => {
                tracing::warn!(
                    agent = %spec.id,
                    pid,
                    error = ?e,
                    "Job Object created but process assignment failed; \
                     cascade-kill on stop will be disabled for this child"
                );
                None
            }
        },
        Err(e) => {
            tracing::warn!(
                agent = %spec.id,
                pid,
                error = ?e,
                "CreateJobObjectW failed; cascade-kill on stop will be \
                 disabled for this child"
            );
            None
        }
    };
    #[cfg(not(target_os = "windows"))]
    let job: Option<Arc<JobObject>> = None;

    Ok(SpawnedChild { child, pid, job })
}

#[cfg(unix)]
fn libc_setsid() -> i32 {
    extern "C" {
        fn setsid() -> i32;
    }
    unsafe { setsid() }
}

/// Kill a supervised process tree.
///
/// On Unix this sends `SIGTERM` and falls back to `SIGKILL` after a
/// grace period. On Windows this uses the Job Object handle minted at
/// spawn time (`spawn_child`) to atomically cascade-terminate every
/// process in the supervised tree — required to avoid the §5.1
/// zombie-leak class where `cmd.exe → ping.exe` would lose the parent
/// but keep the grandchild alive. The `job` parameter is unused on
/// Unix and is `None` on Windows only if the spawn-time Job Object
/// allocation or assignment failed (in which case we degrade to a
/// best-effort `TerminateProcess` on the parent pid alone — losing
/// the cascade — and emit a warning so the operator can investigate).
async fn kill_process(
    pid: u32,
    signal: StopSignal,
    grace_secs: u64,
    #[cfg_attr(unix, allow(unused_variables))] job: Option<Arc<JobObject>>,
    #[cfg_attr(unix, allow(unused_variables))] started_at_unix: Option<i64>,
) {
    #[cfg(unix)]
    {
        let _ = job;
        let pid_i = pid as i32;
        match signal {
            StopSignal::Term => {
                send_signal(pid_i, libc_sigterm());
                let deadline = std::time::Duration::from_secs(grace_secs.max(1));
                let mut waited = std::time::Duration::ZERO;
                let step = std::time::Duration::from_millis(200);
                while waited < deadline {
                    if !pid_alive(pid_i) {
                        return;
                    }
                    tokio::time::sleep(step).await;
                    waited += step;
                }
                send_signal(pid_i, libc_sigkill());
            }
            StopSignal::Kill => {
                send_signal(pid_i, libc_sigkill());
            }
        }
    }
    #[cfg(target_os = "windows")]
    {
        let _ = (signal, grace_secs);
        match job {
            Some(j) => {
                // Atomic tree-kill via TerminateJobObject — cascades
                // to every process assigned to the job. The grace
                // semantics from the Unix branch don't translate
                // directly (Windows has no SIGTERM equivalent for a
                // job); we go straight to TerminateJobObject with
                // exit code 1. JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
                // (set at spawn time) means the kill happens anyway
                // when the Arc drops — TerminateJobObject just makes
                // it immediate and signals "we killed this", not "it
                // exited."
                if let Err(e) = j.terminate(1) {
                    tracing::warn!(
                        pid,
                        error = ?e,
                        "TerminateJobObject failed; supervised child tree may be incomplete-killed"
                    );
                }
            }
            None => {
                // Spawn-time Job Object setup failed. Fall through to
                // single-process TerminateProcess so we at least kill
                // the parent — the §5.1 grandchild-leak class returns,
                // but the operator sees the parent stop. Emit a
                // tracing event so the operator knows the cascade
                // didn't happen.
                //
                // Before terminating, verify the pid hasn't been
                // recycled — Windows reuses pids aggressively, and
                // a supervisor that captured pid N for "agent A"
                // could find pid N now belongs to an unrelated
                // process (notepad.exe, antivirus, anything). Use
                // GetProcessTimes to compare the process's creation
                // time against `started_at_unix` from the slot. If
                // they don't match within a tolerance, bail.
                tracing::warn!(
                    pid,
                    "Windows supervised process has no Job Object — \
                     using TerminateProcess fallback (no cascade kill; \
                     grandchildren may leak). See Parslee-ai/car#231 §5.1."
                );
                terminate_process_by_pid_verified(pid, started_at_unix);
            }
        }
    }
    // No fallback for unix targets that aren't macOS/Linux — none are
    // currently supported, but if a future target lands without a kill
    // path defined, the build will fail loudly at the cfg matrix.
}

/// Best-effort Windows TerminateProcess on a single PID, with a
/// creation-time identity check to defeat pid reuse.
///
/// Used only when the supervised process's Job Object wasn't
/// successfully created at spawn time — the cascade behavior is
/// lost but we still stop the parent. Pre-§5.1 behavior, with one
/// safety improvement over the original Unix code: Windows recycles
/// pids aggressively, and a supervisor that captured pid N at spawn
/// could find pid N now belongs to an unrelated process by the time
/// `stop` runs. We compare the live process's creation time
/// (`GetProcessTimes`) against `expected_started_at_unix` from the
/// slot. A mismatch beyond a small tolerance means the pid has been
/// reused — bail with a warning instead of terminating someone
/// else's process.
///
/// `expected_started_at_unix == None` means the supervisor never
/// captured a spawn timestamp (shouldn't happen in practice once a
/// process is Running, but defensive): in that case we skip the
/// terminate entirely rather than risk killing the wrong process.
#[cfg(target_os = "windows")]
fn terminate_process_by_pid_verified(pid: u32, expected_started_at_unix: Option<i64>) {
    use windows::Win32::Foundation::{CloseHandle, FALSE, FILETIME};
    use windows::Win32::System::Threading::{
        GetProcessTimes, OpenProcess, TerminateProcess, PROCESS_QUERY_LIMITED_INFORMATION,
        PROCESS_TERMINATE,
    };

    let expected = match expected_started_at_unix {
        Some(t) => t,
        None => {
            tracing::warn!(
                pid,
                "fallback terminate skipped: no spawn timestamp in slot — \
                 cannot verify pid identity against possible reuse"
            );
            return;
        }
    };

    unsafe {
        // QUERY_LIMITED_INFORMATION is enough for GetProcessTimes;
        // PROCESS_TERMINATE is needed for the actual kill. Request
        // both up front so we don't have to open twice.
        let access = PROCESS_TERMINATE | PROCESS_QUERY_LIMITED_INFORMATION;
        let handle = match OpenProcess(access, FALSE, pid) {
            Ok(h) => h,
            Err(e) => {
                tracing::warn!(pid, ?e, "OpenProcess for fallback terminate failed");
                return;
            }
        };

        let mut creation = FILETIME::default();
        let mut exit_ft = FILETIME::default();
        let mut kernel = FILETIME::default();
        let mut user = FILETIME::default();
        let times_result =
            GetProcessTimes(handle, &mut creation, &mut exit_ft, &mut kernel, &mut user);
        if let Err(e) = times_result {
            tracing::warn!(
                pid,
                ?e,
                "GetProcessTimes failed during pid-reuse verification — \
                 skipping terminate to avoid potentially killing the wrong process"
            );
            let _ = CloseHandle(handle);
            return;
        }

        // Convert FILETIME (100ns ticks since 1601-01-01 UTC) to
        // Unix seconds (since 1970-01-01 UTC). The constant is the
        // number of 100ns ticks between those epochs.
        const TICKS_BETWEEN_EPOCHS: u64 = 116_444_736_000_000_000;
        const TICKS_PER_SECOND: u64 = 10_000_000;
        let creation_ticks =
            ((creation.dwHighDateTime as u64) << 32) | (creation.dwLowDateTime as u64);
        let actual_unix = if creation_ticks >= TICKS_BETWEEN_EPOCHS {
            ((creation_ticks - TICKS_BETWEEN_EPOCHS) / TICKS_PER_SECOND) as i64
        } else {
            // Pre-1970 creation time means we're reading garbage
            // (shouldn't happen on a real Windows system); refuse
            // to terminate.
            tracing::warn!(
                pid,
                creation_ticks,
                "process creation time is pre-1970 — refusing to terminate"
            );
            let _ = CloseHandle(handle);
            return;
        };

        // Allow a 2-second tolerance for clock skew between the
        // supervisor's chrono::Utc::now() at spawn and the kernel's
        // FILETIME. In practice these are within milliseconds, but
        // a 2s window prevents a fast retry from accidentally
        // failing the check.
        let drift = (actual_unix - expected).abs();
        if drift > 2 {
            tracing::warn!(
                pid,
                expected,
                actual_unix,
                drift,
                "pid reuse detected: process creation time differs from supervised spawn timestamp — \
                 refusing to terminate (the process now holding this pid is not the one we supervised)"
            );
            let _ = CloseHandle(handle);
            return;
        }

        // Identity verified; safe to terminate.
        if let Err(e) = TerminateProcess(handle, 1) {
            tracing::warn!(
                pid,
                ?e,
                "TerminateProcess failed (process may be protected by anti-malware \
                 or already exited)"
            );
        }
        let _ = CloseHandle(handle);
    }
}

/// RAII wrapper around a Windows Job Object handle. The job is
/// created with `JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE` so that, even if
/// the supervisor crashes before `kill_process` runs, the OS will
/// kill the entire process tree when the last handle to the job
/// drops. Cheap to clone via `Arc`; the underlying handle is only
/// closed when the final clone is dropped.
///
/// Holds a raw `HANDLE`. `Send + Sync` are safe to assert because
/// the only mutation we ever perform on the handle is via Windows
/// APIs that are themselves thread-safe (`TerminateJobObject`,
/// `CloseHandle`, `AssignProcessToJobObject`).
#[cfg(target_os = "windows")]
pub(crate) struct JobObject {
    handle: windows::Win32::Foundation::HANDLE,
}

#[cfg(target_os = "windows")]
unsafe impl Send for JobObject {}
#[cfg(target_os = "windows")]
unsafe impl Sync for JobObject {}

#[cfg(target_os = "windows")]
impl JobObject {
    /// Create a new Job Object with `KILL_ON_JOB_CLOSE` set. The
    /// returned wrapper owns the handle via RAII: even if no process
    /// is ever assigned via [`Self::assign`], dropping the wrapper
    /// closes the handle and the kernel reclaims the (now-empty)
    /// Job Object — no leak, just a wasted system call.
    ///
    /// IMPORTANT (Win32 RAII pattern): the wrapper is constructed
    /// *before* the `SetInformationJobObject` call so that if that
    /// fallible setup step fails, `Drop` runs and closes the handle.
    /// The previous (now-fixed) ordering would propagate the error
    /// via `?` *before* `Self` was constructed, leaking the kernel
    /// handle on every failed setup attempt.
    pub(crate) fn new() -> windows::core::Result<Self> {
        use windows::Win32::System::JobObjects::{
            CreateJobObjectW, JobObjectExtendedLimitInformation, SetInformationJobObject,
            JOBOBJECT_EXTENDED_LIMIT_INFORMATION, JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE,
        };
        unsafe {
            let handle = CreateJobObjectW(None, windows::core::PCWSTR::null())?;
            // Construct the wrapper FIRST so Drop owns the handle if
            // anything below fails. Standard Rust FFI-RAII pattern.
            let job = Self { handle };

            // Populate the extended-limit struct: only the
            // KILL_ON_JOB_CLOSE flag matters; everything else stays
            // at zero (no CPU/memory/active-process limits — we just
            // want the cascade-kill on close). The struct is copied
            // synchronously by SetInformationJobObject into kernel
            // memory before returning, so the stack lifetime here is
            // sufficient.
            let mut info = JOBOBJECT_EXTENDED_LIMIT_INFORMATION::default();
            info.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
            let info_ptr = &info as *const _ as *const std::ffi::c_void;
            let info_size = std::mem::size_of::<JOBOBJECT_EXTENDED_LIMIT_INFORMATION>() as u32;
            SetInformationJobObject(
                job.handle,
                JobObjectExtendedLimitInformation,
                info_ptr,
                info_size,
            )?; // If this fails, `job` drops here, Drop closes the handle.
            Ok(job)
        }
    }

    /// Assign a process to this job. Once assigned, the process and
    /// every child it spawns become subject to the job's
    /// KILL_ON_JOB_CLOSE limit. The process is identified by its
    /// Windows pid (matches `tokio::process::Child::id()`).
    pub(crate) fn assign(&self, pid: u32) -> windows::core::Result<()> {
        use windows::Win32::Foundation::{CloseHandle, FALSE};
        use windows::Win32::System::JobObjects::AssignProcessToJobObject;
        use windows::Win32::System::Threading::{
            OpenProcess, PROCESS_SET_QUOTA, PROCESS_TERMINATE,
        };
        unsafe {
            // PROCESS_SET_QUOTA + PROCESS_TERMINATE is the minimum
            // access mask AssignProcessToJobObject requires (per the
            // Win32 docs). FALSE for bInheritHandle — the handle
            // doesn't need to outlive this scope; the assignment
            // itself is persistent on the kernel side.
            let process_handle = OpenProcess(PROCESS_SET_QUOTA | PROCESS_TERMINATE, FALSE, pid)?;
            let result = AssignProcessToJobObject(self.handle, process_handle);
            // Always close the process handle; the job retains its
            // own reference to the process internally.
            let _ = CloseHandle(process_handle);
            result?;
            Ok(())
        }
    }

    /// Terminate every process in this job. Cascades atomically. The
    /// `exit_code` is what each terminated process reports as its
    /// exit code; convention is 1 for "killed by supervisor."
    pub(crate) fn terminate(&self, exit_code: u32) -> windows::core::Result<()> {
        use windows::Win32::System::JobObjects::TerminateJobObject;
        unsafe { TerminateJobObject(self.handle, exit_code) }
    }
}

#[cfg(target_os = "windows")]
impl Drop for JobObject {
    fn drop(&mut self) {
        // The handle's drop is the safety net: even if the supervisor
        // forgot to call `terminate` (panic, crash, abort), closing
        // the last handle while KILL_ON_JOB_CLOSE is set kills the
        // tree. Belt + suspenders.
        //
        // `is_invalid()` is forward-looking defensive code, not
        // active. Today the handle is always valid here because
        // `JobObject::new` constructs `Self` only after
        // `CreateJobObjectW` succeeds, and no other code path
        // invalidates the handle. If a future revision adds a
        // `pub fn close()` method that nulls the field, this guard
        // means Drop won't double-close. Cheap, defensive, honest.
        use windows::Win32::Foundation::CloseHandle;
        if !self.handle.is_invalid() {
            unsafe {
                let _ = CloseHandle(self.handle);
            }
        }
    }
}

/// No-op JobObject for non-Windows builds. Exists so call-site code
/// can stay platform-agnostic at the type level — `Option<Arc<JobObject>>`
/// is `None` everywhere off-Windows and the platform-specific kill
/// path ignores the field.
#[cfg(not(target_os = "windows"))]
pub(crate) struct JobObject {
    _private: (),
}

#[cfg(unix)]
fn libc_sigterm() -> i32 {
    15
}
#[cfg(unix)]
fn libc_sigkill() -> i32 {
    9
}

#[cfg(unix)]
fn send_signal(pid: i32, sig: i32) {
    extern "C" {
        fn kill(pid: i32, sig: i32) -> i32;
    }
    unsafe {
        let _ = kill(pid, sig);
    }
}

#[cfg(unix)]
fn pid_alive(pid: i32) -> bool {
    extern "C" {
        fn kill(pid: i32, sig: i32) -> i32;
    }
    // Signal 0 is the existence probe — returns 0 if the pid
    // exists and we have permission, -1 with ESRCH if not.
    unsafe { kill(pid, 0) == 0 }
}

#[cfg(not(unix))]
fn pid_alive(_pid: i32) -> bool {
    // Non-unix platforms don't currently use the pid-file double-
    // spawn guard. Treat all pids as dead so the guard is a no-op
    // there rather than blocking legitimate spawns.
    false
}

/// Path of the conventional per-agent pid file at
/// `~/.car/run/<id>.pid`. Agents that want supervisor-level
/// double-spawn protection write this file at startup with their
/// own pid (see trader's `src/daemon.js` for a reference
/// implementation). The supervisor only ever reads — never writes
/// — keeping the "agents own their own pid file" invariant clean.
///
/// Returns `None` when `$HOME` is unset (e.g. PID 1 supervisor on
/// a stripped-down container); without a home dir, the convention
/// has nowhere to live and the guard degrades to a no-op.
fn agent_pid_file(agent_id: &str) -> Option<std::path::PathBuf> {
    let home = std::env::var_os("HOME").map(std::path::PathBuf::from)?;
    Some(
        home.join(".car")
            .join("run")
            .join(format!("{agent_id}.pid")),
    )
}

/// Returns `Some(pid)` when an external (non-supervisor-managed)
/// process is currently holding the agent's pid file. Cleans up
/// stale pid files (process gone, pid unparseable) as a side
/// effect so the next caller doesn't repeat the work.
///
/// "External" here means "alive but not necessarily known to this
/// supervisor". A pid in our own `state` map could also match —
/// in which case skipping respawn is still correct (don't double-
/// spawn our own child if for some reason we re-enter).
fn external_agent_pid(agent_id: &str) -> Option<i32> {
    let path = agent_pid_file(agent_id)?;
    let content = match std::fs::read_to_string(&path) {
        Ok(s) => s,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return None,
        Err(e) => {
            tracing::warn!(
                agent = %agent_id,
                path = %path.display(),
                error = %e,
                "reading agent pid file failed; assuming no external instance"
            );
            return None;
        }
    };
    let pid: i32 = match content.trim().parse() {
        Ok(n) => n,
        Err(_) => {
            tracing::warn!(
                agent = %agent_id,
                path = %path.display(),
                content = %content.trim(),
                "agent pid file content unparseable; removing"
            );
            let _ = std::fs::remove_file(&path);
            return None;
        }
    };
    if pid_alive(pid) {
        Some(pid)
    } else {
        tracing::info!(
            agent = %agent_id,
            pid,
            path = %path.display(),
            "stale agent pid file (process gone); removing"
        );
        let _ = std::fs::remove_file(&path);
        None
    }
}

fn validate_id(id: &str) -> Result<(), SupervisorError> {
    if id.is_empty() {
        return Err(SupervisorError::InvalidId(id.to_string()));
    }
    if !id
        .chars()
        .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.')
    {
        return Err(SupervisorError::InvalidId(id.to_string()));
    }
    if id == "." || id == ".." {
        return Err(SupervisorError::InvalidId(id.to_string()));
    }
    Ok(())
}

/// Hard ceiling on bytes read from the tail of a log file, regardless
/// of how many lines the caller asks for.
///
/// Agent logs are append-only and never rotated (the spawn opens them
/// `.create(true).append(true)` and nothing in this crate caps or
/// truncates them), so a long-lived crash-looping agent can produce a
/// multi-GB file. Reading the whole thing every 2s under the viewer's
/// Follow poll would be O(file_size) I/O + allocation per tick. We cap
/// at 8 MiB: enough to hold tens of thousands of lines of normal log
/// output, small enough that the worst case is a bounded, predictable
/// read. Files at or under this size are read in full, so the exact
/// pre-#273 `total`/`more` semantics are preserved for every
/// realistically-sized log; only pathological multi-GB logs are
/// truncated, and that truncation is reported honestly (see
/// [`read_stream_window`]).
const LOG_TAIL_BYTE_CEILING: u64 = 8 * 1024 * 1024;

/// Read a windowed tail of one log file via a bounded backward seek.
///
/// Returns the last `n` lines (`0` ⇒ all, still subject to the byte
/// ceiling below) ending `offset` lines before the end, plus the
/// file's total line count and whether older lines remain past the
/// window (for "load more"). A missing file is an empty window, never
/// an error — an agent that has only ever written stdout still tails
/// cleanly.
///
/// ## Bounded read — no whole-file slurp
///
/// Agent logs are append-only and never rotated anywhere in this
/// crate, so they can grow without bound. Rather than
/// `read_to_string` the entire file (O(file_size) I/O + alloc on every
/// Follow poll), this seeks backward from EOF in chunks and reads at
/// most [`LOG_TAIL_BYTE_CEILING`] bytes — only as far as needed to
/// satisfy the requested `n + offset` line window (plus one extra line
/// so `more` can be computed honestly).
///
/// When the file fits within the ceiling, the whole file is read and
/// the `total`/`more`/window results are byte-for-byte identical to
/// the previous whole-file implementation. When the file exceeds the
/// ceiling and the requested window reaches into the truncated region,
/// the read is capped: `total` then reflects only the lines counted
/// within the last [`LOG_TAIL_BYTE_CEILING`] bytes, `more` is forced
/// `true` (older lines provably exist beyond what we scanned), and a
/// `warn!` records how many bytes were dropped. Callers that need the
/// full history are pointed at the on-disk file path (the viewer's
/// "Reveal in Finder").
async fn read_stream_window(
    path: &Path,
    n: usize,
    offset: usize,
) -> Result<StreamWindow, SupervisorError> {
    use tokio::io::{AsyncReadExt, AsyncSeekExt};

    let mut file = match tokio::fs::File::open(path).await {
        Ok(f) => f,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            return Ok(StreamWindow::default());
        }
        Err(e) => return Err(e.into()),
    };
    let file_len = file.metadata().await?.len();
    if file_len == 0 {
        return Ok(StreamWindow::default());
    }

    // Seek backward from EOF, reading CHUNK bytes at a time into a
    // front-growing buffer, until we reach the start of the file or hit
    // the byte ceiling. We deliberately scan the whole bounded tail
    // (not just enough bytes for the requested window): the ceiling is
    // already the I/O bound we care about, and scanning to it lets
    // `total`/`more` stay exact for every under-ceiling file — matching
    // the pre-#273 whole-file semantics the viewer's "showing N of M"
    // depends on.
    const CHUNK: u64 = 64 * 1024;
    let read_cap = file_len.min(LOG_TAIL_BYTE_CEILING);
    let mut buf: Vec<u8> = Vec::with_capacity(read_cap as usize);
    let mut pos = file_len; // bytes from start; we read [pos-chunk, pos)
    let mut bytes_read: u64 = 0;
    let mut hit_ceiling = false;

    loop {
        if pos == 0 {
            break; // read the whole file
        }
        if bytes_read >= read_cap {
            hit_ceiling = true;
            break;
        }
        let this_chunk = CHUNK.min(pos).min(read_cap - bytes_read);
        let chunk_start = pos - this_chunk;
        file.seek(std::io::SeekFrom::Start(chunk_start)).await?;
        let mut chunk = vec![0u8; this_chunk as usize];
        file.read_exact(&mut chunk).await?;
        // Prepend (we're walking backward).
        chunk.extend_from_slice(&buf);
        buf = chunk;
        pos = chunk_start;
        bytes_read += this_chunk;
    }

    let reached_start = pos == 0;
    // If we stopped before the start of the file, the first line in the
    // buffer is almost certainly a partial line — drop it so we never
    // surface a truncated line as if it were whole.
    let truncated = !reached_start;
    if truncated && hit_ceiling {
        tracing::warn!(
            path = %path.display(),
            file_len,
            bytes_read,
            dropped = file_len.saturating_sub(bytes_read),
            "log tail hit the {LOG_TAIL_BYTE_CEILING}-byte ceiling; older lines \
             were not scanned — use the on-disk file for full history"
        );
    }

    // Decode and split into lines. Loss-tolerant: child output is not
    // guaranteed UTF-8.
    let text = String::from_utf8_lossy(&buf);
    let mut all: Vec<&str> = text.lines().collect();
    // Drop the leading partial line when we didn't reach the file start.
    if truncated && !all.is_empty() {
        all.remove(0);
    }

    let scanned = all.len();
    // `total` is the true file line count only when we read the whole
    // file; otherwise it's the count within the bytes we scanned (the
    // viewer labels this "lines on disk", which is now "lines in the
    // tail we scanned" for truncated reads — documented).
    let total = scanned;

    // The window ends `offset` lines before the end of what we scanned.
    let end = scanned.saturating_sub(offset);
    let start = if n == 0 { 0 } else { end.saturating_sub(n) };
    let window: Vec<String> = all[start..end].iter().map(|s| s.to_string()).collect();

    // Older lines remain if the window didn't reach the top of what we
    // scanned, OR if we truncated (there's provably more beyond the
    // ceiling).
    let more = start > 0 || truncated;

    Ok(StreamWindow {
        lines: window,
        total,
        more,
    })
}

/// Validate that `command` names an executable file safe to spawn.
///
/// The rules — enforced together; failing any one rejects the spec:
///
/// 1. **Non-empty.** An empty string is meaningless.
/// 2. **Absolute path.** No `$PATH` lookup. PATH-injection (a
///    co-resident process renaming a binary on `$PATH`, or `cwd`
///    pointing at a directory the user happens to have on PATH)
///    is removed by requiring callers to spell the full path.
/// 3. **No `..` segments.** Defense against laundering a denied
///    prefix through traversal.
/// 4. **File must exist** at upsert time and be a regular file (not
///    a directory or socket). Symlinks are followed via `metadata`.
/// 5. **Executable bit set** for the launching user (POSIX).
///    Windows skips the bit check — the loader decides.
/// 6. **Not under a world-writable scratch dir** (`/tmp`,
///    `/private/tmp`, `/var/tmp`, `/dev/shm`). The 2026-05 audit
///    walked an exploit chain that staged a binary under `/tmp`
///    before calling `agents.upsert`; this denylist makes that
///    specific shape stop working without trying to enumerate every
///    legitimate prefix (which would inevitably miss a real one).
pub fn validate_command(command: &str) -> Result<(), SupervisorError> {
    if command.is_empty() {
        return Err(SupervisorError::InvalidCommand {
            command: command.to_string(),
            reason: "command is empty",
        });
    }
    let path = Path::new(command);
    if !path.is_absolute() {
        return Err(SupervisorError::InvalidCommand {
            command: command.to_string(),
            reason: "command must be an absolute path; PATH lookup is not allowed",
        });
    }
    if path
        .components()
        .any(|c| matches!(c, std::path::Component::ParentDir))
    {
        return Err(SupervisorError::InvalidCommand {
            command: command.to_string(),
            reason: "command path must not contain `..` segments",
        });
    }
    // Denylist scratch dirs before the metadata check so a missing
    // file in /tmp produces the more informative error message.
    const SCRATCH_PREFIXES: &[&str] = &["/tmp/", "/private/tmp/", "/var/tmp/", "/dev/shm/"];
    if SCRATCH_PREFIXES.iter().any(|p| command.starts_with(p)) {
        return Err(SupervisorError::InvalidCommand {
            command: command.to_string(),
            reason: "command lives under a world-writable scratch directory \
                     (/tmp, /private/tmp, /var/tmp, /dev/shm)",
        });
    }
    let meta = std::fs::metadata(path).map_err(|_| SupervisorError::InvalidCommand {
        command: command.to_string(),
        reason: "command file does not exist or is not readable",
    })?;
    if !meta.is_file() {
        return Err(SupervisorError::InvalidCommand {
            command: command.to_string(),
            reason: "command path is not a regular file",
        });
    }
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        // Any execute bit set is enough — the spawn semantics will
        // figure out which one matches the launching user.
        if meta.permissions().mode() & 0o111 == 0 {
            return Err(SupervisorError::InvalidCommand {
                command: command.to_string(),
                reason: "command file has no execute bit set",
            });
        }
    }
    Ok(())
}

/// Mint a fresh per-agent auth token (#169). 32 random bytes
/// encoded as base64url-no-pad — 43 ASCII chars, identical shape to
/// the daemon's per-launch token so audit / diff tooling treats them
/// uniformly. Mirrors `car_ffi_common::auth_token::generate` but
/// lives here to avoid pulling car-ffi-common back into car-registry
/// (car-ffi-common already depends on car-registry, so the other
/// direction would cycle).
fn mint_agent_token() -> String {
    use base64::Engine as _;
    let a = uuid::Uuid::new_v4();
    let b = uuid::Uuid::new_v4();
    let mut bytes = [0u8; 32];
    bytes[..16].copy_from_slice(a.as_bytes());
    bytes[16..].copy_from_slice(b.as_bytes());
    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(bytes)
}

/// Length-checked constant-time byte compare. Avoids leaking match
/// position via timing.
fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut diff: u8 = 0;
    for (x, y) in a.iter().zip(b.iter()) {
        diff |= x ^ y;
    }
    diff == 0
}

/// Resolve an interpreter name (`"node"`, `"python"`, `"deno"`, …)
/// to an absolute path by walking `$PATH`, then validating the
/// result with the same [`validate_command`] rules used at upsert.
///
/// Why this exists: `validate_command` rightly rejects PATH lookup
/// at upsert time (closes the PATH-injection variant the 2026-05
/// audit found). That left every lifecycle-agent installer needing
/// to know the user's interpreter path up front — which moves
/// across nvm / fnm / Homebrew / Volta upgrades. Callers can now
/// pass `interpreter: "node"` once and the supervisor resolves it
/// against the *current* PATH at upsert (#171). The resolved path
/// is then stored verbatim in the manifest, so PATH changes at
/// runtime don't silently rewire which binary the spec points to.
///
/// Rules:
/// 1. `name` must be a bare program name — no `/`, no `..`.
///    Anything path-shaped is rejected here (the caller should
///    pass it through `command` directly if they want the path).
/// 2. `$PATH` is split on the platform separator. Empty entries
///    are skipped (the POSIX "current directory" alias is *not*
///    honored — same rationale as #1).
/// 3. Each candidate `dir/name` is checked for existence + execute
///    bit (POSIX). The first match is returned.
/// 4. The resolved path is passed through [`validate_command`] so
///    e.g. an interpreter parked under `/tmp` is still rejected.
pub fn resolve_interpreter(name: &str) -> Result<PathBuf, SupervisorError> {
    if name.is_empty() {
        return Err(SupervisorError::InvalidCommand {
            command: name.to_string(),
            reason: "interpreter name is empty",
        });
    }
    if name.contains('/') || name.contains('\\') {
        return Err(SupervisorError::InvalidCommand {
            command: name.to_string(),
            reason: "interpreter name must be a bare program name, not a path; \
                     pass paths via `command`",
        });
    }
    if name == "." || name == ".." || name.contains("..") {
        return Err(SupervisorError::InvalidCommand {
            command: name.to_string(),
            reason: "interpreter name must not contain `..` segments",
        });
    }

    let path_var = std::env::var_os("PATH").ok_or(SupervisorError::InvalidCommand {
        command: name.to_string(),
        reason: "no $PATH set; cannot resolve interpreter",
    })?;
    for dir in std::env::split_paths(&path_var) {
        if dir.as_os_str().is_empty() {
            continue;
        }
        let candidate = dir.join(name);
        // metadata follows symlinks; on POSIX the executable bit
        // check inside `validate_command` is authoritative.
        if std::fs::metadata(&candidate).is_ok() {
            // Run through the same gate every `command` passes
            // through. If the resolution happens to land in /tmp
            // (unusual but possible), this rejects it.
            let abs = candidate.to_string_lossy().into_owned();
            validate_command(&abs)?;
            return Ok(candidate);
        }
    }

    Err(SupervisorError::InvalidCommand {
        command: name.to_string(),
        reason: "interpreter not found on $PATH",
    })
}

/// One entry in the [`Supervisor::health`] report.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AgentHealth {
    pub id: String,
    pub command: String,
    /// `true` when [`validate_command`] still accepts `command`. The
    /// `reason` field is only populated when `ok` is `false`.
    pub ok: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
}

fn load_manifest(path: &Path) -> Result<Manifest, SupervisorError> {
    if !path.exists() {
        return Ok(Manifest::default());
    }
    let bytes = std::fs::read(path)?;
    let manifest: Manifest = serde_json::from_slice(&bytes)?;
    Ok(manifest)
}

fn write_json_atomic<T: Serialize>(path: &Path, value: &T) -> Result<(), SupervisorError> {
    let parent = path.parent().ok_or_else(|| {
        std::io::Error::new(
            std::io::ErrorKind::InvalidInput,
            "manifest path has no parent",
        )
    })?;
    std::fs::create_dir_all(parent)?;
    let tmp = parent.join(format!(
        ".{}.tmp",
        path.file_name()
            .and_then(|s| s.to_str())
            .unwrap_or("supervisor-write")
    ));
    let json = serde_json::to_vec_pretty(value)?;
    std::fs::write(&tmp, json)?;
    std::fs::rename(&tmp, path)?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    fn temp_supervisor() -> (tempfile::TempDir, Supervisor) {
        // Default tempfile location is /tmp/... on Linux, which the
        // supervisor's command-sandbox denylist correctly rejects as a
        // world-writable scratch dir. Put the test tempdir under the
        // crate's target directory instead — never world-writable, always
        // present during cargo test. Canonicalize so the path has no `..`
        // segments (the sandbox also rejects those).
        let target = std::env::var_os("CARGO_TARGET_DIR")
            .map(std::path::PathBuf::from)
            .unwrap_or_else(|| {
                std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
                    .join("..")
                    .join("..")
                    .join("target")
            });
        std::fs::create_dir_all(&target).ok();
        let target = std::fs::canonicalize(&target).unwrap_or(target);
        let tmp = tempfile::TempDir::new_in(&target).unwrap();
        let s = Supervisor::with_paths(tmp.path().join("agents.json"), tmp.path().join("logs"))
            .unwrap();
        (tmp, s)
    }

    fn echo_spec(id: &str, message: &str) -> AgentSpec {
        AgentSpec {
            id: id.into(),
            name: id.into(),
            command: "/bin/sh".into(),
            args: vec!["-c".into(), format!("echo {message}; sleep 30")],
            cwd: None,
            env: Default::default(),
            restart: RestartPolicy::Never,
            max_restarts: 1,
            backoff_secs: 1,
            auto_start: false,
            token: String::new(),
        }
    }

    #[tokio::test]
    async fn upsert_persists_and_lists() {
        let (_tmp, s) = temp_supervisor();
        s.upsert(echo_spec("alpha", "hi")).await.unwrap();
        let list = s.list().await;
        assert_eq!(list.len(), 1);
        assert_eq!(list[0].spec.id, "alpha");
        assert!(s.manifest_path().exists());
    }

    #[tokio::test]
    async fn manifest_round_trips_across_supervisors() {
        // The test's intent is "the on-disk manifest is the
        // round-trip source of truth — a fresh observer sees the
        // same entries the previous owner wrote." Using
        // `list_from_manifest` (the read-only fallback the daemon
        // uses in observe-only mode) checks that property without
        // racing against OS-level lock release after `drop(s)` —
        // under high test parallelism the close/flock-release window
        // is occasionally observable and the re-acquire flaked.
        let (tmp, s) = temp_supervisor();
        s.upsert(echo_spec("a", "x")).await.unwrap();
        s.upsert(echo_spec("b", "y")).await.unwrap();

        let list = Supervisor::list_from_manifest(&tmp.path().join("agents.json")).unwrap();
        assert_eq!(list.len(), 2);
        assert_eq!(list[0].spec.id, "a");
        assert_eq!(list[1].spec.id, "b");
    }

    #[tokio::test]
    async fn start_then_stop_runs_child_and_reaps_it() {
        let (_tmp, s) = temp_supervisor();
        s.upsert(echo_spec("runme", "hello")).await.unwrap();
        s.start("runme").await.unwrap();

        // Give the child a moment to spawn and write its log line.
        for _ in 0..50 {
            let snap = s.list().await;
            if matches!(snap[0].status, AgentStatus::Running) && snap[0].pid.is_some() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }

        let snap = s.list().await;
        assert!(matches!(snap[0].status, AgentStatus::Running), "{snap:?}");
        assert!(snap[0].pid.is_some());

        let pid = snap[0].pid.unwrap() as i32;
        s.stop("runme", StopSignal::Term).await.unwrap();
        for _ in 0..50 {
            if !pid_alive(pid) {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
        assert!(!pid_alive(pid), "child must be reaped after stop");

        let after = s.list().await;
        assert!(matches!(after[0].status, AgentStatus::Stopped));
        assert!(after[0].pid.is_none());
    }

    #[tokio::test]
    async fn wait_for_blocks_until_running_and_times_out_otherwise() {
        let (_tmp, s) = temp_supervisor();
        s.upsert(echo_spec("w", "hi")).await.unwrap();

        // Not started yet → waiting for Running hits the deadline.
        let err = s
            .wait_for(
                "w",
                &[AgentStatus::Running],
                std::time::Duration::from_millis(150),
                std::time::Duration::from_millis(20),
            )
            .await
            .unwrap_err();
        assert!(
            matches!(err, SupervisorError::WaitTimeout { .. }),
            "{err:?}"
        );

        // Unknown id → NotFound, not a timeout.
        assert!(matches!(
            s.wait_for(
                "ghost",
                &[AgentStatus::Running],
                std::time::Duration::from_millis(50),
                std::time::Duration::from_millis(20),
            )
            .await,
            Err(SupervisorError::NotFound(_))
        ));

        // After start, it reaches Running well within the deadline.
        s.start("w").await.unwrap();
        let agent = s
            .wait_for(
                "w",
                &[AgentStatus::Running],
                std::time::Duration::from_secs(5),
                std::time::Duration::from_millis(25),
            )
            .await
            .unwrap();
        assert_eq!(agent.status, AgentStatus::Running);
        assert!(agent.pid.is_some());

        s.stop("w", StopSignal::Term).await.unwrap();
    }

    /// A child that always exits non-zero under `RestartPolicy::Always`
    /// must be restarted up to `max_restarts` times — bumping
    /// `restart_count` 1→2 — and then driven to the terminal `Errored`
    /// state once the count exceeds the cap, carrying the last
    /// `last_exit_code`. Exercises the crash → backoff → restart →
    /// errored sequence end to end through a real OS child.
    ///
    /// `max_restarts` is kept at 2 and `backoff_secs` at 0 (the
    /// restart_backoff floor still clamps to ~1s/attempt) so the whole
    /// sequence settles in a few seconds; the poll budget is generous to
    /// avoid flakiness under parallel test load.
    #[cfg(not(target_os = "windows"))]
    #[tokio::test]
    async fn crash_loop_restarts_then_errors_with_exit_code() {
        let (_tmp, s) = temp_supervisor();
        let spec = AgentSpec {
            id: "crasher".into(),
            name: "crasher".into(),
            command: "/bin/sh".into(),
            args: vec!["-c".into(), "exit 1".into()],
            cwd: None,
            env: Default::default(),
            restart: RestartPolicy::Always,
            max_restarts: 2,
            backoff_secs: 0,
            auto_start: false,
            token: String::new(),
        };
        s.upsert(spec).await.unwrap();
        s.start("crasher").await.unwrap();

        // The restart counter must climb to at least max_restarts (2)
        // on its way to Errored.
        let mut peak_restart_count = 0u32;
        let mut terminal = None;
        // ~30s budget: backoff sequence is ~1s + ~2s plus spawn/exit
        // overhead; generous so it never flakes.
        for _ in 0..600 {
            let snap = s.list().await;
            peak_restart_count = peak_restart_count.max(snap[0].restart_count);
            if matches!(snap[0].status, AgentStatus::Errored) {
                terminal = Some(snap[0].clone());
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }

        let terminal = terminal.expect("agent never reached terminal Errored state");
        assert!(
            matches!(terminal.status, AgentStatus::Errored),
            "expected terminal Errored, got {terminal:?}"
        );
        assert!(
            peak_restart_count >= 2,
            "restart_count should climb to at least max_restarts (2), peaked at {peak_restart_count}"
        );
        assert_eq!(
            terminal.last_exit_code,
            Some(1),
            "last_exit_code should reflect the child's `exit 1`"
        );
    }

    /// F3 / #231 §5.1: on Windows, `Supervisor::stop` must cascade-kill
    /// the entire supervised process tree — not just the parent. The
    /// canonical failure shape is `cmd.exe` (parent) spawning
    /// `ping.exe` (grandchild). Pre-PR, `car stop` killed `cmd.exe`
    /// but `ping.exe` kept running, and `list()` lied about the
    /// agent being stopped.
    ///
    /// This test spawns the parent+grandchild pair through the
    /// supervisor, captures both PIDs via WMI, calls stop, and
    /// asserts BOTH PIDs are gone. Skipped on non-Windows targets —
    /// the Unix SIGTERM/process-group path doesn't need this guard.
    #[cfg(target_os = "windows")]
    #[tokio::test]
    async fn stop_cascades_to_grandchildren_on_windows() {
        use std::process::Command as StdCommand;

        let (_tmp, s) = temp_supervisor();
        let spec = AgentSpec {
            id: "tree".into(),
            name: "tree".into(),
            // `cmd.exe /C ping -t 127.0.0.1`:
            //   - cmd.exe is the supervised parent (gets the
            //     CAR_AGENT_TOKEN env var, etc.)
            //   - ping.exe is the grandchild (lives forever — `-t`
            //     means "ping until killed")
            // Pre-§5.1 fix: stop kills cmd.exe, ping.exe leaks.
            // Post-fix: TerminateJobObject cascades both.
            command: "cmd.exe".into(),
            args: vec!["/C".into(), "ping".into(), "-t".into(), "127.0.0.1".into()],
            cwd: None,
            env: Default::default(),
            restart: RestartPolicy::Never,
            max_restarts: 1,
            backoff_secs: 1,
            auto_start: false,
            token: String::new(),
        };
        s.upsert(spec).await.unwrap();
        s.start("tree").await.unwrap();

        // Wait for the supervised cmd.exe to land in Running state.
        let mut parent_pid: Option<u32> = None;
        for _ in 0..100 {
            let snap = s.list().await;
            if matches!(snap[0].status, AgentStatus::Running) {
                if let Some(pid) = snap[0].pid {
                    parent_pid = Some(pid);
                    break;
                }
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        let parent_pid = parent_pid.expect("supervisor never reported Running pid for cmd.exe");

        // Find the ping.exe grandchild via PowerShell Get-CimInstance.
        // (We deliberately don't use `wmic` here — it was deprecated
        // in Windows 10 21H1 and removed from in-box on Windows 11
        // 24H2 / Server 2025, so a wmic-based check would fail on
        // exactly the hosts most likely to run this test in 2026,
        // looking like a supervisor bug when it's a test-infra bug.
        // Get-CimInstance is the official replacement and is
        // available on every supported Windows host.)
        // Re-query a few times because cmd.exe takes a moment to
        // spawn its child.
        let mut grandchild_pid: Option<u32> = None;
        for _ in 0..50 {
            let out = StdCommand::new("powershell")
                .args(&[
                    "-NoProfile",
                    "-Command",
                    &format!(
                        "Get-CimInstance Win32_Process \
                         -Filter 'ParentProcessId={parent_pid} AND Name=\"ping.exe\"' | \
                         Select-Object -ExpandProperty ProcessId"
                    ),
                ])
                .output();
            if let Ok(out) = out {
                let text = String::from_utf8_lossy(&out.stdout);
                // One pid per line; first non-empty line is our match.
                for line in text.lines() {
                    let trimmed = line.trim();
                    if trimmed.is_empty() {
                        continue;
                    }
                    if let Ok(pid) = trimmed.parse::<u32>() {
                        grandchild_pid = Some(pid);
                        break;
                    }
                }
                if grandchild_pid.is_some() {
                    break;
                }
            }
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        }
        let grandchild_pid = grandchild_pid
            .expect("never observed ping.exe grandchild under cmd.exe parent — supervisor or PowerShell/CIM bug");

        // Kill the tree. Pre-PR this would only kill cmd.exe.
        s.stop("tree", StopSignal::Term).await.unwrap();

        // Both PIDs should be gone within a small grace window.
        // `tasklist /FI "PID eq <n>"` returns "No tasks are running"
        // when the pid is dead.
        let pid_alive_win = |pid: u32| -> bool {
            StdCommand::new("tasklist")
                .args(&["/FI", &format!("PID eq {pid}")])
                .output()
                .map(|o| {
                    let text = String::from_utf8_lossy(&o.stdout);
                    !text.contains("No tasks are running")
                })
                .unwrap_or(false)
        };

        let mut parent_gone = false;
        let mut grandchild_gone = false;
        for _ in 0..50 {
            parent_gone = parent_gone || !pid_alive_win(parent_pid);
            grandchild_gone = grandchild_gone || !pid_alive_win(grandchild_pid);
            if parent_gone && grandchild_gone {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        }
        assert!(
            parent_gone,
            "cmd.exe (parent pid {parent_pid}) still alive after stop"
        );
        assert!(
            grandchild_gone,
            "ping.exe (grandchild pid {grandchild_pid}) still alive after stop — \
             §5.1 zombie-leak regression"
        );
    }

    #[tokio::test]
    async fn respawn_tears_down_prior_child_no_orphan() {
        // Regression for the overlapping-loop orphan bug: a second
        // supervision for the same id must kill the prior child, not
        // leave it running untracked alongside the new loop. Before
        // the teardown-first fix, `spawn_supervision` overwrote the
        // slot's stop sender/task and the old child ran on as an
        // orphan (observed in the field as a live agent bound to its
        // port while the slot reported `Errored`).
        let (_tmp, s) = temp_supervisor();
        s.upsert(echo_spec("solo", "hi")).await.unwrap();
        s.start("solo").await.unwrap();

        let mut pid1 = None;
        for _ in 0..100 {
            let snap = s.list().await;
            if matches!(snap[0].status, AgentStatus::Running) {
                if let Some(p) = snap[0].pid {
                    pid1 = Some(p as i32);
                    break;
                }
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
        let pid1 = pid1.expect("first child should reach Running");

        // Force the overlap path directly.
        let spec = s.list().await.into_iter().next().unwrap().spec;
        s.spawn_supervision(spec).await;

        let mut pid2 = None;
        for _ in 0..100 {
            let snap = s.list().await;
            if matches!(snap[0].status, AgentStatus::Running) {
                if let Some(p) = snap[0].pid {
                    if p as i32 != pid1 {
                        pid2 = Some(p as i32);
                        break;
                    }
                }
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
        let pid2 = pid2.expect("second child should reach Running with a fresh pid");
        assert_ne!(pid1, pid2, "respawn should produce a distinct child");

        for _ in 0..100 {
            if !pid_alive(pid1) {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
        assert!(!pid_alive(pid1), "prior child must be reaped, not orphaned");
        assert!(pid_alive(pid2), "new child should still be alive");

        s.stop("solo", StopSignal::Kill).await.unwrap();
    }

    #[test]
    fn restart_backoff_is_exponential_capped_and_floored() {
        // Never below the base floor.
        assert!(restart_backoff(5, 1).as_secs() >= 5);
        // Grows with consecutive attempts (attempt 3 ≈ 20s).
        assert!(restart_backoff(5, 3).as_secs() >= 20);
        // Tops out at the cap (+ at most one jitter slice = cap/8).
        let deep = restart_backoff(5, 50).as_secs();
        assert!(deep >= BACKOFF_CAP_SECS, "deep backoff {deep}s below cap");
        assert!(
            deep <= BACKOFF_CAP_SECS + BACKOFF_CAP_SECS / 8 + 1,
            "deep backoff {deep}s ignored cap"
        );
        // No overflow / panic at the integer ceiling.
        let _ = restart_backoff(5, u32::MAX);
        // A zero base is floored to 1s, not 0.
        assert!(restart_backoff(0, 1).as_secs() >= 1);
    }

    #[tokio::test]
    async fn tail_log_returns_recent_lines() {
        let (_tmp, s) = temp_supervisor();
        let mut spec = echo_spec("logs", "line-from-child");
        // Short-lived: emit one line and exit.
        spec.args = vec!["-c".into(), "echo line-from-child".into()];
        s.upsert(spec).await.unwrap();
        s.start("logs").await.unwrap();

        for _ in 0..50 {
            let lines = s.tail_log("logs", 10).await.unwrap();
            if !lines.is_empty() {
                assert!(lines.iter().any(|l| l.contains("line-from-child")));
                return;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
        panic!("tail_log never observed the child's output");
    }

    /// Write stdout/stderr log files directly so `read_log` can be
    /// exercised deterministically without spawning a child.
    fn write_logs(s: &Supervisor, id: &str, stdout: &str, stderr: &str) {
        std::fs::write(s.log_dir().join(format!("{id}.stdout.log")), stdout).unwrap();
        std::fs::write(s.log_dir().join(format!("{id}.stderr.log")), stderr).unwrap();
    }

    #[tokio::test]
    async fn long_stderr_no_longer_buries_live_stdout() {
        // The Parslee-ai/car#273 regression: a 126-line stderr crash
        // dump plus live stdout. The old code concatenated the whole
        // stderr after stdout then kept only the last 100 lines, so a
        // long stderr hid 100% of stdout. Now each stream is tailed to
        // its own budget, so live stdout always survives.
        let (_tmp, s) = temp_supervisor();
        let stdout: String = (0..10).map(|i| format!("live-stdout-{i}\n")).collect();
        let stderr: String = (0..126).map(|i| format!("old-stderr-{i}\n")).collect();
        write_logs(&s, "a", &stdout, &stderr);

        let lines = s.tail_log("a", 100).await.unwrap();
        // All 10 live stdout lines are present despite the 126-line stderr.
        for i in 0..10 {
            assert!(
                lines.iter().any(|l| l == &format!("live-stdout-{i}")),
                "live stdout line {i} was buried"
            );
        }
    }

    #[tokio::test]
    async fn read_log_stream_selection_and_paging() {
        let (_tmp, s) = temp_supervisor();
        let stdout: String = (0..50).map(|i| format!("out-{i}\n")).collect();
        let stderr: String = (0..30).map(|i| format!("err-{i}\n")).collect();
        write_logs(&s, "a", &stdout, &stderr);

        // stdout-only respects n and reports the full total + more flag.
        let t = s.read_log("a", LogStream::Stdout, 10, 0).await.unwrap();
        assert_eq!(t.stdout.len(), 10);
        assert_eq!(t.stdout.first().unwrap(), "out-40");
        assert_eq!(t.stdout.last().unwrap(), "out-49");
        assert!(t.stderr.is_empty(), "stderr excluded when stream=stdout");
        assert_eq!(t.stdout_total, 50);
        assert!(t.more, "40 older stdout lines remain");
        assert!(t.stdout_path.ends_with("a.stdout.log"));

        // Paging back one screen via offset surfaces the previous window.
        let prev = s.read_log("a", LogStream::Stdout, 10, 10).await.unwrap();
        assert_eq!(prev.stdout.first().unwrap(), "out-30");
        assert_eq!(prev.stdout.last().unwrap(), "out-39");

        // stderr-only.
        let e = s.read_log("a", LogStream::Stderr, 5, 0).await.unwrap();
        assert_eq!(e.stderr.last().unwrap(), "err-29");
        assert!(e.stdout.is_empty());
        assert_eq!(e.stderr_total, 30);

        // n == 0 means the whole stream, no cap, and no "more".
        let full = s.read_log("a", LogStream::Stdout, 0, 0).await.unwrap();
        assert_eq!(full.stdout.len(), 50);
        assert!(!full.more);
    }

    #[tokio::test]
    async fn read_log_missing_files_are_empty_not_error() {
        let (_tmp, s) = temp_supervisor();
        let t = s
            .read_log("never-ran", LogStream::Combined, 100, 0)
            .await
            .unwrap();
        assert!(t.lines.is_empty());
        assert_eq!(t.stdout_total, 0);
        assert_eq!(t.stderr_total, 0);
        assert!(!t.more);
    }

    #[tokio::test]
    async fn read_stream_window_bounded_tail_matches_whole_file_when_under_ceiling() {
        // For any normal-sized log (well under the byte ceiling) the
        // bounded backward seek must produce byte-for-byte the same
        // window/total/more as a whole-file read would. Exercise the
        // window, paging, n==0, and the "more" flag against a known file.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("x.log");
        let content: String = (0..1000).map(|i| format!("line-{i}\n")).collect();
        std::fs::write(&path, &content).unwrap();

        // Last 10 lines.
        let w = read_stream_window(&path, 10, 0).await.unwrap();
        assert_eq!(w.total, 1000);
        assert_eq!(w.lines.len(), 10);
        assert_eq!(w.lines.first().unwrap(), "line-990");
        assert_eq!(w.lines.last().unwrap(), "line-999");
        assert!(w.more, "older lines remain");

        // Page back one screen.
        let prev = read_stream_window(&path, 10, 10).await.unwrap();
        assert_eq!(prev.lines.first().unwrap(), "line-980");
        assert_eq!(prev.lines.last().unwrap(), "line-989");
        assert!(prev.more);

        // n == 0 ⇒ whole file, no "more".
        let full = read_stream_window(&path, 0, 0).await.unwrap();
        assert_eq!(full.lines.len(), 1000);
        assert_eq!(full.total, 1000);
        assert!(!full.more);

        // A window that reaches the top reports no "more".
        let top = read_stream_window(&path, 1000, 0).await.unwrap();
        assert_eq!(top.lines.len(), 1000);
        assert!(!top.more);
    }

    #[tokio::test]
    async fn read_stream_window_only_reads_a_bounded_window_for_large_files() {
        // A file far larger than any realistic tail request: the bounded
        // seek must return just the requested window (the newest lines),
        // never the whole file, and never error or hang.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("big.log");
        // ~2.4 MB: 100k lines. Small enough to stay under the 8 MiB
        // ceiling (so total stays exact) but large enough that a naive
        // full read would be wasteful per Follow poll.
        let mut content = String::with_capacity(2_400_000);
        for i in 0..100_000 {
            content.push_str(&format!("entry-{i:06}\n"));
        }
        std::fs::write(&path, &content).unwrap();

        let w = read_stream_window(&path, 5, 0).await.unwrap();
        assert_eq!(w.lines.len(), 5);
        assert_eq!(w.lines.last().unwrap(), "entry-099999");
        assert_eq!(w.lines.first().unwrap(), "entry-099995");
        assert_eq!(w.total, 100_000, "under ceiling ⇒ exact total");
        assert!(w.more);
    }

    #[tokio::test]
    async fn read_stream_window_truncates_honestly_past_the_byte_ceiling() {
        // Build a file just over the byte ceiling so the backward seek
        // stops before the start. The newest lines must still be exact,
        // the leading (partial) line must be dropped, and `more` must be
        // forced true.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("huge.log");
        // Each line is 30 bytes; produce > ceiling/30 lines.
        let per_line = 30usize;
        let line_count = (LOG_TAIL_BYTE_CEILING as usize / per_line) + 5_000;
        let mut content = String::with_capacity(line_count * per_line);
        for i in 0..line_count {
            // Zero-pad to a fixed 29-char body + newline = 30 bytes.
            content.push_str(&format!("ln{i:027}\n"));
        }
        assert!(content.len() as u64 > LOG_TAIL_BYTE_CEILING);
        std::fs::write(&path, &content).unwrap();

        let w = read_stream_window(&path, 3, 0).await.unwrap();
        // Newest lines are exact and whole.
        assert_eq!(w.lines.len(), 3);
        assert_eq!(
            w.lines.last().unwrap(),
            &format!("ln{:027}", line_count - 1)
        );
        // We scanned only a bounded tail, so total < the real line count.
        assert!(
            w.total < line_count,
            "truncated read should report fewer than all {line_count} lines, got {}",
            w.total
        );
        // And we must advertise that older lines exist beyond the window.
        assert!(w.more, "truncated tail must force more=true");
    }

    #[tokio::test]
    async fn remove_stops_running_agent() {
        let (_tmp, s) = temp_supervisor();
        s.upsert(echo_spec("ephemeral", "x")).await.unwrap();
        s.start("ephemeral").await.unwrap();
        for _ in 0..50 {
            let snap = s.list().await;
            if matches!(snap[0].status, AgentStatus::Running) {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
        let removed = s.remove("ephemeral").await.unwrap();
        assert!(removed);
        assert!(s.list().await.is_empty());
    }

    #[tokio::test]
    async fn invalid_ids_rejected() {
        let (_tmp, s) = temp_supervisor();
        let mut spec = echo_spec("ok", "x");
        spec.id = "..".into();
        assert!(s.upsert(spec).await.is_err());
    }

    #[tokio::test]
    async fn start_all_skips_auto_start_false() {
        let (_tmp, s) = temp_supervisor();
        let mut a = echo_spec("auto", "x");
        a.auto_start = true;
        let mut b = echo_spec("manual", "y");
        b.auto_start = false;
        s.upsert(a).await.unwrap();
        s.upsert(b).await.unwrap();

        let started = s.start_all().await;
        assert_eq!(started, vec!["auto".to_string()]);
    }

    #[test]
    fn auto_start_defaults_to_false_when_omitted_from_json() {
        // Pre-2026-05 default was true. Anything that round-trips a
        // partial spec (a host that omits the field, an agent
        // ingesting peer-supplied JSON) must now get false.
        let spec: AgentSpec =
            serde_json::from_str(r#"{"id":"x","name":"X","command":"/bin/sh"}"#).unwrap();
        assert!(!spec.auto_start, "default flipped 2026-05 — must be false");
    }

    #[tokio::test]
    async fn validate_command_rejects_relative_path() {
        let (_tmp, s) = temp_supervisor();
        let mut spec = echo_spec("rel", "x");
        spec.command = "sh".into();
        let err = s.upsert(spec).await.unwrap_err();
        assert!(
            matches!(err, SupervisorError::InvalidCommand { .. }),
            "expected InvalidCommand, got {err:?}"
        );
    }

    #[tokio::test]
    async fn validate_command_rejects_tmp_prefix() {
        // Stage a real, executable binary under /tmp so the rejection
        // is purely about the prefix denylist, not "file missing".
        let bin = std::env::temp_dir().join("car-registry-validate-test.sh");
        if !bin.starts_with("/tmp") && !bin.starts_with("/private/tmp") {
            // macOS resolves $TMPDIR to a per-user dir under
            // /var/folders/... — outside the denylist on purpose.
            // Skip on platforms where TMPDIR doesn't land under /tmp.
            return;
        }
        std::fs::write(&bin, "#!/bin/sh\necho hi\n").unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o755)).unwrap();
        }

        let (_tmp, s) = temp_supervisor();
        let mut spec = echo_spec("scratch", "x");
        spec.command = bin.to_string_lossy().into_owned();
        let err = s.upsert(spec).await.unwrap_err();
        assert!(
            matches!(err, SupervisorError::InvalidCommand { reason, .. }
                if reason.contains("scratch")),
            "expected scratch-dir rejection, got {err:?}"
        );

        let _ = std::fs::remove_file(&bin);
    }

    #[tokio::test]
    async fn validate_command_rejects_missing_file() {
        let (_tmp, s) = temp_supervisor();
        let mut spec = echo_spec("ghost", "x");
        spec.command = "/usr/local/bin/no-such-binary-please".into();
        let err = s.upsert(spec).await.unwrap_err();
        assert!(matches!(err, SupervisorError::InvalidCommand { .. }));
    }

    #[tokio::test]
    async fn validate_command_rejects_directory() {
        let (_tmp, s) = temp_supervisor();
        let mut spec = echo_spec("dir", "x");
        spec.command = "/usr".into();
        let err = s.upsert(spec).await.unwrap_err();
        assert!(matches!(
            err,
            SupervisorError::InvalidCommand { reason, .. } if reason.contains("regular file")
        ));
    }

    #[tokio::test]
    async fn validate_command_rejects_parent_dir_segment() {
        let (_tmp, s) = temp_supervisor();
        let mut spec = echo_spec("dotdot", "x");
        spec.command = "/usr/bin/../bin/sh".into();
        let err = s.upsert(spec).await.unwrap_err();
        assert!(matches!(
            err,
            SupervisorError::InvalidCommand { reason, .. } if reason.contains("..")
        ));
    }

    #[tokio::test]
    async fn upsert_accepts_legitimate_command() {
        let (_tmp, s) = temp_supervisor();
        // /bin/sh exists and is executable on every supported host.
        s.upsert(echo_spec("sane", "x")).await.unwrap();
    }

    #[test]
    fn resolve_interpreter_finds_sh_on_path() {
        // `sh` is present on every supported host; the prior `$PATH`
        // is preserved so the resolver walks the same directories the
        // user's shell would.
        let resolved = resolve_interpreter("sh").unwrap();
        assert!(resolved.is_absolute(), "got {:?}", resolved);
        assert_eq!(resolved.file_name().unwrap(), "sh");
    }

    #[test]
    fn resolve_interpreter_rejects_path_shaped_name() {
        let err = resolve_interpreter("/bin/sh").unwrap_err();
        assert!(matches!(
            err,
            SupervisorError::InvalidCommand { reason, .. }
                if reason.contains("bare program name")
        ));
    }

    #[test]
    fn resolve_interpreter_rejects_parent_dir_in_name() {
        let err = resolve_interpreter("..").unwrap_err();
        assert!(matches!(
            err,
            SupervisorError::InvalidCommand { reason, .. }
                if reason.contains("..")
        ));
    }

    #[test]
    fn resolve_interpreter_rejects_missing_name() {
        let err = resolve_interpreter("no-such-interpreter-please-2026").unwrap_err();
        assert!(matches!(
            err,
            SupervisorError::InvalidCommand { reason, .. }
                if reason.contains("not found on $PATH")
        ));
    }

    #[tokio::test]
    async fn upsert_mints_token_when_empty_and_retains_on_reupsert() {
        let (_tmp, s) = temp_supervisor();
        let agent = s.upsert(echo_spec("with-token", "x")).await.unwrap();
        // Minted: 43-char base64url-no-pad (32 random bytes).
        assert_eq!(agent.spec.token.len(), 43, "got {:?}", agent.spec.token);
        assert!(agent
            .spec
            .token
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_'));

        // Re-upsert WITHOUT a token retains the prior one — this is
        // the path operators take when they edit `name` / `args`
        // without intending to invalidate every connected child.
        let spec = echo_spec("with-token", "y");
        assert!(spec.token.is_empty());
        let again = s.upsert(spec).await.unwrap();
        assert_eq!(again.spec.token, agent.spec.token);

        // Re-upsert WITH an explicit token replaces — explicit
        // rotation.
        let mut spec = echo_spec("with-token", "z");
        spec.token = "rotated-explicitly-by-operator".into();
        let rotated = s.upsert(spec).await.unwrap();
        assert_eq!(rotated.spec.token, "rotated-explicitly-by-operator");

        // The lookup helper sees the rotated value.
        assert_eq!(
            s.agent_token("with-token").await.as_deref(),
            Some("rotated-explicitly-by-operator")
        );
        assert!(s.agent_token("nope").await.is_none());
    }

    #[tokio::test]
    async fn validate_agent_token_uses_constant_time_compare() {
        let (_tmp, s) = temp_supervisor();
        let agent = s.upsert(echo_spec("auth-test", "x")).await.unwrap();
        assert!(s.validate_agent_token("auth-test", &agent.spec.token).await);
        assert!(!s.validate_agent_token("auth-test", "wrong").await);
        // Wrong agent_id never matches, regardless of token.
        assert!(!s.validate_agent_token("nope", &agent.spec.token).await);
    }

    #[tokio::test]
    async fn default_child_env_is_set_and_round_trips() {
        let (_tmp, s) = temp_supervisor();
        // Empty by default — child specs see only their per-spec env.
        assert!(s.default_child_env().await.is_empty());

        s.set_default_child_env([
            ("CAR_DAEMON_URL", "ws://127.0.0.1:9100"),
            ("CAR_AUTH_TOKEN", "abc123"),
        ])
        .await;

        let got = s.default_child_env().await;
        assert_eq!(got.len(), 2);
        assert_eq!(
            got.get("CAR_DAEMON_URL").map(String::as_str),
            Some("ws://127.0.0.1:9100")
        );
        assert_eq!(
            got.get("CAR_AUTH_TOKEN").map(String::as_str),
            Some("abc123")
        );

        // A second call replaces, not merges — the daemon owns the
        // canonical set and may rotate the token across restarts.
        s.set_default_child_env([("CAR_DAEMON_URL", "ws://127.0.0.1:9200")])
            .await;
        let got = s.default_child_env().await;
        assert_eq!(got.len(), 1);
        assert_eq!(
            got.get("CAR_DAEMON_URL").map(String::as_str),
            Some("ws://127.0.0.1:9200")
        );
    }

    #[tokio::test]
    async fn health_flags_a_broken_command_after_upsert() {
        let (tmp, s) = temp_supervisor();
        // Plant a real binary, upsert against it, then delete it.
        let real = tmp.path().join("disposable.sh");
        std::fs::write(&real, "#!/bin/sh\necho hi\n").unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let mut perm = std::fs::metadata(&real).unwrap().permissions();
            perm.set_mode(0o755);
            std::fs::set_permissions(&real, perm).unwrap();
        }
        let mut spec = echo_spec("vanish", "x");
        spec.command = real.to_string_lossy().into_owned();
        s.upsert(spec).await.unwrap();

        // Healthy at first.
        let report = s.health().await;
        let me = report.iter().find(|h| h.id == "vanish").unwrap();
        assert!(me.ok, "expected fresh-upsert spec to be healthy");

        // Delete the binary out from under us — simulates an upgrade
        // that moved Node / pruned a Homebrew symlink.
        std::fs::remove_file(&real).unwrap();
        let report = s.health().await;
        let me = report.iter().find(|h| h.id == "vanish").unwrap();
        assert!(!me.ok, "expected health to flag missing command");
        assert!(
            me.reason
                .as_deref()
                .unwrap_or("")
                .contains("does not exist"),
            "got reason {:?}",
            me.reason
        );
    }

    // ---------------------------------------------------------------------
    // Phase 1 dual-read migration tests (Parslee-ai/car#182)
    // ---------------------------------------------------------------------

    /// Helper: write a legacy `agents.json` directly so we can
    /// test the migration path without going through `upsert`
    /// (which would write to both legacy + new layout).
    fn write_legacy_agents_json(path: &Path, specs: &[AgentSpec]) {
        let manifest = Manifest {
            agents: specs.to_vec(),
        };
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        std::fs::write(path, serde_json::to_vec_pretty(&manifest).unwrap()).unwrap();
    }

    fn temp_tmpdir() -> tempfile::TempDir {
        let target = std::env::var_os("CARGO_TARGET_DIR")
            .map(std::path::PathBuf::from)
            .unwrap_or_else(|| {
                std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
                    .join("..")
                    .join("..")
                    .join("target")
            });
        std::fs::create_dir_all(&target).ok();
        let target = std::fs::canonicalize(&target).unwrap_or(target);
        tempfile::TempDir::new_in(&target).unwrap()
    }

    /// Boot a supervisor over a manifest whose previous supervisor was
    /// *just dropped*, tolerating the flock/fork hand-off window: a
    /// concurrently fork/exec'ing child elsewhere in this test process
    /// (other supervisor tests spawn `/bin/sh` agents) briefly holds
    /// inherited duplicates of every open fd — including our just-closed
    /// lock fd — which keeps the advisory lock alive until its exec.
    /// So a reboot-immediately-after-drop can transiently see
    /// `AlreadyRunning`. Retry briefly; anything persistent is a real
    /// failure. Production never hits this: a daemon acquires the lock
    /// once and holds it for life, and fail-fast on a *live* holder is
    /// the intended #44 semantic.
    fn reboot_with_paths(manifest: PathBuf, logs: PathBuf) -> Supervisor {
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        loop {
            match Supervisor::with_paths(manifest.clone(), logs.clone()) {
                Ok(s) => return s,
                Err(SupervisorError::AlreadyRunning(_)) if std::time::Instant::now() < deadline => {
                    std::thread::sleep(std::time::Duration::from_millis(25));
                }
                Err(e) => panic!("reboot after drop failed: {e}"),
            }
        }
    }

    #[test]
    fn boot_with_legacy_only_mirrors_to_new_layout() {
        // Legacy agents.json carries one entry; the agents/ dir
        // doesn't exist yet. On boot, the entry should load AND a
        // matching manifest.toml should be written.
        let tmp = temp_tmpdir();
        let legacy = tmp.path().join("agents.json");
        write_legacy_agents_json(
            &legacy,
            &[AgentSpec {
                id: "legacy-ui".into(),
                name: "Legacy UI".into(),
                command: "/bin/sh".into(),
                args: vec!["-c".into(), "true".into()],
                cwd: None,
                env: Default::default(),
                restart: RestartPolicy::OnFailure,
                max_restarts: 5,
                backoff_secs: 2,
                auto_start: false,
                token: "tok-leg".into(),
            }],
        );

        let s = Supervisor::with_paths(legacy.clone(), tmp.path().join("logs")).unwrap();
        // The supervisor loads it.
        let agents = futures::executor::block_on(s.list());
        assert_eq!(agents.len(), 1);
        assert_eq!(agents[0].spec.id, "legacy-ui");

        // The mirror happened at boot.
        let mirrored = tmp.path().join("agents/legacy-ui/manifest.toml");
        assert!(
            mirrored.exists(),
            "expected mirrored manifest at {}",
            mirrored.display()
        );
        let text = std::fs::read_to_string(&mirrored).unwrap();
        let m: crate::manifest::AgentManifest = toml::from_str(&text).unwrap();
        assert_eq!(m.agent.id, "legacy-ui");
        // Migrated token round-trips.
        if let crate::manifest::TransportSpec::ExternalProcess(t) = &m.transport {
            assert_eq!(t.token, "tok-leg");
        } else {
            panic!("expected external_process transport, got {:?}", m.transport);
        }
    }

    #[test]
    fn boot_with_new_layout_only_loads_manifest_dir() {
        // No legacy file; one manifest.toml in agents/.
        let tmp = temp_tmpdir();
        let agents_dir = tmp.path().join("agents");
        std::fs::create_dir_all(&agents_dir).unwrap();
        let m = crate::manifest::from_legacy_spec(&AgentSpec {
            id: "new-only".into(),
            name: "New Only".into(),
            command: "/bin/sh".into(),
            args: vec![],
            cwd: None,
            env: Default::default(),
            restart: RestartPolicy::Never,
            max_restarts: 1,
            backoff_secs: 1,
            auto_start: false,
            token: "tok-new".into(),
        });
        crate::manifest::write_manifest(&agents_dir, &m).unwrap();

        let legacy = tmp.path().join("agents.json");
        let s = Supervisor::with_paths(legacy, tmp.path().join("logs")).unwrap();
        let agents = futures::executor::block_on(s.list());
        assert_eq!(agents.len(), 1);
        assert_eq!(agents[0].spec.id, "new-only");
        assert_eq!(agents[0].spec.token, "tok-new");
    }

    #[test]
    fn boot_with_mixed_sources_new_layout_wins_on_id_conflict() {
        // Both legacy + new contain "overlap" — the new-layout
        // entry should win. Legacy-only entries still load.
        let tmp = temp_tmpdir();
        let legacy = tmp.path().join("agents.json");
        write_legacy_agents_json(
            &legacy,
            &[
                AgentSpec {
                    id: "overlap".into(),
                    name: "Overlap (legacy)".into(),
                    command: "/bin/sh".into(),
                    args: vec!["-c".into(), "echo legacy".into()],
                    cwd: None,
                    env: Default::default(),
                    restart: RestartPolicy::Never,
                    max_restarts: 1,
                    backoff_secs: 1,
                    auto_start: false,
                    token: "legacy-token".into(),
                },
                AgentSpec {
                    id: "legacy-only".into(),
                    name: "Legacy Only".into(),
                    command: "/bin/sh".into(),
                    args: vec![],
                    cwd: None,
                    env: Default::default(),
                    restart: RestartPolicy::Never,
                    max_restarts: 1,
                    backoff_secs: 1,
                    auto_start: false,
                    token: "leg-only-tok".into(),
                },
            ],
        );

        let agents_dir = tmp.path().join("agents");
        std::fs::create_dir_all(&agents_dir).unwrap();
        let new_overlap = crate::manifest::from_legacy_spec(&AgentSpec {
            id: "overlap".into(),
            name: "Overlap (new)".into(),
            command: "/bin/sh".into(),
            args: vec!["-c".into(), "echo new".into()],
            cwd: None,
            env: Default::default(),
            restart: RestartPolicy::OnFailure,
            max_restarts: 3,
            backoff_secs: 2,
            auto_start: false,
            token: "new-token".into(),
        });
        crate::manifest::write_manifest(&agents_dir, &new_overlap).unwrap();

        let s = Supervisor::with_paths(legacy, tmp.path().join("logs")).unwrap();
        let mut agents = futures::executor::block_on(s.list());
        agents.sort_by(|a, b| a.spec.id.cmp(&b.spec.id));
        assert_eq!(agents.len(), 2);
        // New-layout value wins for the overlapping id.
        let overlap = agents.iter().find(|a| a.spec.id == "overlap").unwrap();
        assert_eq!(overlap.spec.name, "Overlap (new)");
        assert_eq!(overlap.spec.token, "new-token");
        assert_eq!(overlap.spec.args, vec!["-c", "echo new"]);
        // Legacy-only entry still loads.
        let legacy_only = agents.iter().find(|a| a.spec.id == "legacy-only").unwrap();
        assert_eq!(legacy_only.spec.token, "leg-only-tok");
    }

    #[test]
    fn migration_is_idempotent_across_reboots() {
        // Two boots over the same temp dir shouldn't change
        // anything observable beyond the migration log.
        let tmp = temp_tmpdir();
        let legacy = tmp.path().join("agents.json");
        write_legacy_agents_json(
            &legacy,
            &[AgentSpec {
                id: "iddy".into(),
                name: "Iddy".into(),
                command: "/bin/sh".into(),
                args: vec![],
                cwd: None,
                env: Default::default(),
                restart: RestartPolicy::Never,
                max_restarts: 1,
                backoff_secs: 1,
                auto_start: false,
                token: "tok-iddy".into(),
            }],
        );
        let s1 = Supervisor::with_paths(legacy.clone(), tmp.path().join("logs")).unwrap();
        let mirrored = tmp.path().join("agents/iddy/manifest.toml");
        let first_meta = std::fs::metadata(&mirrored).unwrap();
        // Release the cross-process lock so the "second boot"
        // below can acquire it — the test simulates sequential
        // reboots, not two live supervisors. (#44 guarantees the
        // latter fails fast; see same_path_supervisor_rejects_second
        // for the negative case.)
        drop(s1);

        // Second boot — token already preserved in mirror, mirror
        // not rewritten (the migration only writes when the mirror
        // didn't already exist). Reboot-after-drop, so tolerate the
        // flock/fork hand-off window (see `reboot_with_paths`).
        let _s2 = reboot_with_paths(legacy, tmp.path().join("logs"));
        let second_meta = std::fs::metadata(&mirrored).unwrap();
        // Modified time shouldn't have changed — second boot was a no-op.
        assert_eq!(
            first_meta.modified().unwrap(),
            second_meta.modified().unwrap()
        );
    }

    #[test]
    fn same_path_supervisor_rejects_second() {
        // #44: two car-server processes on the same manifest must
        // not both spawn agents. The OS-level lock on
        // `<manifest_path>.lock` enforces this; the second
        // `with_paths` returns AlreadyRunning with the lock path.
        let tmp = temp_tmpdir();
        let manifest = tmp.path().join("agents.json");
        let logs = tmp.path().join("logs");
        let s1 = Supervisor::with_paths(manifest.clone(), logs.clone()).unwrap();
        let lock_path = {
            let mut s = manifest.as_os_str().to_owned();
            s.push(".lock");
            PathBuf::from(s)
        };
        match Supervisor::with_paths(manifest.clone(), logs.clone()) {
            Err(SupervisorError::AlreadyRunning(p)) => assert_eq!(p, lock_path),
            Err(e) => panic!("expected AlreadyRunning, got error: {e}"),
            Ok(_) => panic!("expected AlreadyRunning, got Ok"),
        }
        // Dropping the first supervisor releases the lock; a third
        // boot succeeds. Validates the OS actually let go. Reboot-
        // after-drop, so tolerate the flock/fork hand-off window
        // (see `reboot_with_paths`).
        drop(s1);
        let _s3 = reboot_with_paths(manifest, logs);
    }

    #[tokio::test]
    async fn list_from_manifest_works_while_lock_is_held() {
        // Read-only fallback path: the first supervisor holds the
        // singleton lock, but `list_from_manifest` / `health_from_manifest`
        // read the legacy manifest file directly and succeed without
        // ever attempting to acquire it.
        let (tmp, s) = temp_supervisor();
        s.upsert(echo_spec("alpha", "a")).await.unwrap();
        s.upsert(echo_spec("beta", "b")).await.unwrap();
        let manifest = tmp.path().join("agents.json");

        let agents = Supervisor::list_from_manifest(&manifest).unwrap();
        assert_eq!(agents.len(), 2);
        // Sorted by id, so alpha comes first.
        assert_eq!(agents[0].spec.id, "alpha");
        assert_eq!(agents[1].spec.id, "beta");
        // Runtime fields default — the live supervisor's state is in
        // memory in this process, but the contract is "this is what an
        // external reader sees," so they're conservatively empty.
        assert_eq!(agents[0].pid, None);
        assert_eq!(agents[0].status, AgentStatus::Stopped);

        let health = Supervisor::health_from_manifest(&manifest).unwrap();
        assert_eq!(health.len(), 2);
        // /bin/echo is the command echo_spec uses; passes validate_command.
        assert!(health.iter().all(|h| h.ok), "{health:?}");
    }

    #[tokio::test]
    async fn upsert_writes_both_legacy_and_new_layout() {
        let (tmp, s) = temp_supervisor();
        s.upsert(echo_spec("dual", "x")).await.unwrap();
        // agents.json updated (legacy path).
        assert!(tmp.path().join("agents.json").exists());
        // agents/<id>/manifest.toml mirrored.
        let m_path = tmp.path().join("agents/dual/manifest.toml");
        assert!(m_path.exists(), "expected mirror at {}", m_path.display());
    }

    #[tokio::test]
    async fn install_manifest_rejects_when_host_lacks_required_capability() {
        let (_tmp, s) = temp_supervisor();
        let m = crate::manifest::from_legacy_spec(&AgentSpec {
            id: "needs-magic".into(),
            name: "Magic Agent".into(),
            command: "/bin/sh".into(),
            args: vec![],
            cwd: None,
            env: Default::default(),
            restart: RestartPolicy::Never,
            max_restarts: 1,
            backoff_secs: 1,
            auto_start: false,
            token: String::new(),
        });
        // Tack on a required capability the host can't satisfy.
        let mut m = m;
        m.capabilities = Some(crate::manifest::CapabilityDeclarations {
            required: std::collections::BTreeMap::from([(
                "inference".into(),
                vec!["text-generation".into()],
            )]),
            ..Default::default()
        });
        let host = crate::install::HostCapabilities {
            car_version: "0.8.0".into(),
            ..Default::default()
        };
        let err = s
            .install_manifest(m, &host)
            .await
            .expect_err("missing cap must fail");
        assert!(err.to_string().contains("inference.text-generation"));
        // No agent was adopted.
        assert!(s.list().await.is_empty());
    }

    #[tokio::test]
    async fn install_manifest_adopts_external_process_when_validation_passes() {
        let (_tmp, s) = temp_supervisor();
        let m = crate::manifest::from_legacy_spec(&AgentSpec {
            id: "installed-agent".into(),
            name: "Installed".into(),
            command: "/bin/sh".into(),
            args: vec!["-c".into(), "true".into()],
            cwd: None,
            env: Default::default(),
            restart: RestartPolicy::Never,
            max_restarts: 1,
            backoff_secs: 1,
            auto_start: false,
            token: String::new(),
        });
        let host = crate::install::HostCapabilities {
            car_version: "0.8.0".into(),
            ..Default::default()
        };
        let (report, managed) = s.install_manifest(m, &host).await.unwrap();
        assert!(report.missing_optional.is_empty());
        let managed = managed.expect("external_process manifest must adopt");
        assert_eq!(managed.spec.id, "installed-agent");
        assert!(!managed.spec.token.is_empty(), "token must be minted");
        assert_eq!(managed.status, AgentStatus::Stopped);
        assert_eq!(managed.pid, None);
        assert_eq!(s.list().await.len(), 1);
    }

    #[tokio::test]
    async fn install_manifest_auto_start_true_starts_external_process_immediately() {
        let (_tmp, s) = temp_supervisor();
        let m = crate::manifest::from_legacy_spec(&AgentSpec {
            id: "auto-installed-agent".into(),
            name: "Auto Installed".into(),
            command: "/bin/sh".into(),
            args: vec!["-c".into(), "echo auto-installed; sleep 30".into()],
            cwd: None,
            env: Default::default(),
            restart: RestartPolicy::Never,
            max_restarts: 1,
            backoff_secs: 1,
            auto_start: true,
            token: String::new(),
        });
        let host = crate::install::HostCapabilities {
            car_version: "0.8.0".into(),
            ..Default::default()
        };

        let (report, managed) = s.install_manifest(m, &host).await.unwrap();

        assert!(report.missing_optional.is_empty());
        let managed = managed.expect("external_process manifest must adopt");
        assert_eq!(managed.spec.id, "auto-installed-agent");
        assert_eq!(managed.spec.auto_start, true);
        assert!(
            matches!(managed.status, AgentStatus::Starting | AgentStatus::Running),
            "install should return the post-start snapshot, got {:?}",
            managed.status
        );

        for _ in 0..50 {
            let list = s.list().await;
            if matches!(list[0].status, AgentStatus::Running) {
                let _ = s.stop("auto-installed-agent", StopSignal::Term).await;
                return;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
        let latest = s.list().await;
        let _ = s.stop("auto-installed-agent", StopSignal::Term).await;
        panic!(
            "auto-started install never reached running; latest status was {:?}",
            latest[0].status
        );
    }

    #[tokio::test]
    async fn install_manifest_writes_pure_data_to_disk_without_adoption() {
        let (tmp, s) = temp_supervisor();
        let m = AgentManifest {
            agent: crate::manifest::AgentIdentity {
                id: "pure-bundle".into(),
                name: "Pure Data".into(),
                namespace: Some("parslee".into()),
                version: Some("0.1.0".into()),
                description: None,
                license: None,
                homepage: None,
            },
            publisher: None,
            runtime: None,
            lifecycle: None,
            transport: crate::manifest::TransportSpec::PureData,
            capabilities: None,
        };
        let host = crate::install::HostCapabilities {
            car_version: "0.8.0".into(),
            ..Default::default()
        };
        let (_report, managed) = s.install_manifest(m, &host).await.unwrap();
        assert!(
            managed.is_none(),
            "pure_data must NOT adopt into supervisor"
        );
        // But the manifest is on disk.
        let m_path = tmp.path().join("agents/pure-bundle/manifest.toml");
        assert!(m_path.exists());
        // The supervisor's spawnable list stays empty.
        assert!(s.list().await.is_empty());
    }

    #[tokio::test]
    async fn remove_reaps_manifest_dir() {
        let (tmp, s) = temp_supervisor();
        s.upsert(echo_spec("reap", "x")).await.unwrap();
        let agent_dir = tmp.path().join("agents/reap");
        assert!(agent_dir.exists());
        let removed = s.remove("reap").await.unwrap();
        assert!(removed);
        assert!(!agent_dir.exists(), "expected manifest dir to be reaped");
    }

    // ---- Parslee-ai/car-releases#44 — cross-process lock contention ----
    //
    // Negative case (second `with_paths` refused while first is alive):
    //   see `same_path_supervisor_rejects_second` above.
    // Read-only fallback (`list_from_manifest` / `health_from_manifest`
    // while the live supervisor holds the lock): see
    // `list_from_manifest_works_while_lock_is_held` above.
}