car-registry 0.55.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(),
//!         // Optional least-privilege daemon scope; absent keeps legacy access.
//!         method_allowlist: None,
//!         // Host-facing capabilities (e.g. ["chat".into()]); empty by default.
//!         capabilities: Vec::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. On Unix,
//!   both token-bearing formats are `0600`, with the aggregate under an
//!   owner-only (`0700`) parent and each per-agent manifest under its own `0700`
//!   directory; reads repair permissions left by older releases before parsing.
//! - **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::io::Write;
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(transparent)]
    InvalidMethodAllowlist(#[from] crate::method_allowlist::InvalidMethodAllowlist),
    #[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, schemars::JsonSchema,
)]
#[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
/// [`AgentStatus`](crate::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, schemars::JsonSchema,
)]
#[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,

    /// Optional daemon-method scope bound to this agent's token at
    /// `session.auth`. `None` preserves the historical unrestricted token;
    /// `Some([])` denies every application method. Entries must be
    /// agent-usable method names from the generated capability manifest.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub method_allowlist: Option<Vec<String>>,

    /// Capabilities this agent advertises to hosts (e.g. `["chat"]` so CarHost
    /// shows the Chat tab and routes `agents.chat` to it). Surfaced verbatim in
    /// `agents.list`. Empty by default; additive, so existing manifests parse
    /// unchanged. See `docs/proposals/agent-chat-surface.md`.
    #[serde(default)]
    pub capabilities: Vec<String>,
}

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

/// Authorization state atomically resolved from a successful per-agent token
/// check. The daemon binds this scope to the authenticated WebSocket session.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AgentTokenBinding {
    pub method_allowlist: Option<Vec<String>>,
}

/// 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>,
    /// When set, a live process this supervisor does not own is already
    /// running as this agent, and that is why the slot is in `Backoff`
    /// rather than `Running` (car#732).
    ///
    /// The failure this exists for looked, from `car ls`, like a broken
    /// agent: `backoff`, `pid: null`, and a perfectly healthy process
    /// serving on the agent's port. Everything appeared broken except the
    /// thing that actually worked, and the only way to see the cause was
    /// `lsof`. Naming the pid here turns a mystery into an instruction.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub blocked_by_pid: Option<i32>,
}

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

/// 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>,
    /// See [`ManagedAgent::blocked_by_pid`].
    blocked_by_pid: Option<i32>,
}

/// 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 `agents.json`, `agents/`, and `logs/` under the CAR state
    /// root (`~/.car` by default, or `$CAR_HOME` when set).
    /// 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 `agents.json` under the CAR state root 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`].
    ///
    /// The root follows `car_home` — `$CAR_HOME` when set, otherwise
    /// `~/.car` — so a relocated daemon supervises its own manifest
    /// instead of the primary's. Errors when neither `$CAR_HOME` nor a
    /// home directory resolves.
    pub fn user_default_manifest_path() -> Result<PathBuf, SupervisorError> {
        let root = car_home::root().ok_or(SupervisorError::NoHomeDir)?;
        Ok(root.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,
                blocked_by_pid: 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() {
            ensure_private_directory(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)?;
        crate::method_allowlist::normalize_method_allowlist(&mut spec.method_allowlist)?;
        {
            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();
                        // The scope belongs to the retained token. An older
                        // upsert client that knows nothing about
                        // `method_allowlist` must not silently turn a scoped
                        // token back into an unrestricted one. To remove a
                        // scope, rotate/provide the token explicitly.
                        if spec.method_allowlist.is_none() {
                            spec.method_allowlist = existing.spec.method_allowlist.clone();
                        }
                    }
                }
                if spec.token.is_empty() {
                    spec.token = mint_agent_token();
                }
            }
            crate::method_allowlist::normalize_method_allowlist(&mut spec.method_allowlist)?;
            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_upsert(&spec.id).await?;
        Ok(ManagedAgent {
            spec,
            status: AgentStatus::Stopped,
            pid: None,
            last_exit_code: None,
            restart_count: 0,
            started_at: None,
            blocked_by_pid: 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())
        }
    }

    /// Validate a per-agent token and return the method scope bound to it.
    ///
    /// Token comparison and scope lookup happen under one state read so an
    /// upsert cannot pair the old token with a newly-written scope. A manually
    /// edited legacy manifest with an invalid scope fails closed at auth even
    /// though normal upserts reject it earlier.
    pub async fn authenticate_agent_token(
        &self,
        id: &str,
        token: &str,
    ) -> Result<Option<AgentTokenBinding>, SupervisorError> {
        let state = self.state.read().await;
        let Some(slot) = state.get(id) else {
            return Ok(None);
        };
        if slot.spec.token.is_empty()
            || !constant_time_eq(slot.spec.token.as_bytes(), token.as_bytes())
        {
            return Ok(None);
        }
        let mut method_allowlist = slot.spec.method_allowlist.clone();
        crate::method_allowlist::normalize_method_allowlist(&mut method_allowlist)?;
        Ok(Some(AgentTokenBinding { method_allowlist }))
    }

    /// Constant-time check that `token` matches the stored token for
    /// `id`. Returns `false` when `id` is unknown or its method scope is
    /// invalid.
    pub async fn validate_agent_token(&self, id: &str, token: &str) -> bool {
        self.authenticate_agent_token(id, token)
            .await
            .ok()
            .flatten()
            .is_some()
    }

    /// 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(),
                method_allowlist: None,
                capabilities: Vec::new(),
            },
            status: AgentStatus::Starting,
            pid: None,
            last_exit_code: None,
            restart_count: 0,
            started_at: None,
            blocked_by_pid: 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 every managed agent, concurrently, under one overall deadline.
    ///
    /// **Any agent lifecycle CAR manages, CAR shuts down.** Leaving a
    /// supervised agent to be reparented and killed by something else is not a
    /// tidiness problem: it denies that agent the catchable SIGTERM its own
    /// shutdown handler needs, so *its* children survive it. car#1106 is what
    /// that costs — a CAR upgrade orphaned a supervised trading daemon's
    /// worker, and two workers ran against one live brokerage account for
    /// ~7 minutes. The singleton lock could not help: the lock is held by the
    /// daemon, the side effects happen in the worker one level below it.
    ///
    /// Concurrent rather than serial, under a **total** deadline rather than
    /// per-agent, because the caller is usually racing something else's
    /// patience. On an upgrade the host will replace this process whether or
    /// not shutdown has finished; N agents each taking their own `grace_secs`
    /// in turn can exceed that window and reintroduce the orphan by a slower
    /// route. One bounded wait for everything is the shape that actually
    /// completes.
    ///
    /// Returns the ids that were still alive when the deadline expired. Empty
    /// is the healthy answer; a non-empty result means those agents were
    /// escalated to SIGKILL without completing their own shutdown, and is worth
    /// logging loudly at the call site.
    pub async fn stop_all(&self, budget: std::time::Duration) -> Vec<String> {
        let ids: Vec<String> = {
            let state = self.state.read().await;
            state.keys().cloned().collect()
        };
        if ids.is_empty() {
            return Vec::new();
        }
        let mut set = tokio::task::JoinSet::new();
        for id in ids.clone() {
            let this = self.clone();
            set.spawn(async move {
                // `Term` so each agent runs its own shutdown handler and can
                // stop the children it is responsible for. That handler is the
                // entire point — `Kill` here would reproduce car#1106.
                let _ = this.stop(&id, StopSignal::Term).await;
            });
        }
        let drain = async { while set.join_next().await.is_some() {} };
        match tokio::time::timeout(budget, drain).await {
            Ok(_) => Vec::new(),
            Err(_) => {
                // Report what is still running rather than what we asked to
                // stop: an agent that exited inside the budget is not a
                // straggler even if the join as a whole timed out.
                let state = self.state.read().await;
                ids.into_iter()
                    .filter(|id| {
                        state
                            .get(id)
                            .and_then(|slot| slot.runtime.pid)
                            .is_some_and(|pid| pid_alive(pid as i32))
                    })
                    .collect()
            }
        }
    }

    /// 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
    }

    /// Remove empty log files left by processes that are gone and unregistered
    /// (Parslee-ai/car#937). Returns how many were removed.
    ///
    /// [`reap_empty_logs`] stops the directory growing from here on, but only
    /// for exits this supervisor observes. It does nothing about what has
    /// already piled up — two months of it on the reported machine — nor about
    /// a process whose supervisor was SIGKILLed before it could reap. This is
    /// the sweep that clears both.
    ///
    /// Three conditions, all required, because this deletes files:
    /// 1. **Zero bytes.** A log with content is evidence and is never removed.
    /// 2. **Not in the registry.** A registered agent owns its log whether or
    ///    not it happens to be running right now — `agents.list` links to it.
    /// 3. **No live pid.** An unregistered process can still be *running* with
    ///    an empty log; unlinking underneath it would send everything it later
    ///    writes to an inode nobody can open. `external_agent_pid` already
    ///    encapsulates "is there a live process for this id".
    ///
    /// Each is a distinct way to be wrong, and the intersection is narrow on
    /// purpose: an inert file for a process that no longer exists and that
    /// nothing on disk claims.
    pub async fn sweep_orphan_logs(&self) -> usize {
        let registered: std::collections::HashSet<String> =
            { self.state.read().await.keys().cloned().collect() };

        let Ok(entries) = std::fs::read_dir(&self.log_dir) else {
            return 0;
        };
        let mut removed = 0usize;
        for entry in entries.flatten() {
            let path = entry.path();
            let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
                continue;
            };
            let Some(id) = name
                .strip_suffix(".stdout.log")
                .or_else(|| name.strip_suffix(".stderr.log"))
            else {
                continue;
            };
            if registered.contains(id) {
                continue;
            }
            match entry.metadata() {
                Ok(meta) if meta.is_file() && meta.len() == 0 => {}
                _ => continue,
            }
            if external_agent_pid(id).is_some() {
                continue;
            }
            if std::fs::remove_file(&path).is_ok() {
                removed += 1;
            }
        }
        if removed > 0 {
            tracing::info!(
                removed,
                dir = %self.log_dir.display(),
                "swept empty log files belonging to no registered agent and no live process"
            );
        }
        removed
    }

    /// Spawn every manifest agent whose `auto_start` is true. Used
    /// by car-server's main on boot. Returns the ids spawned.
    ///
    /// Skips an agent when [`external_agent_pid`] finds a live process
    /// already answering for its id — an instance running outside this
    /// supervisor (orphaned from a previous car-server, started by hand,
    /// running under a different supervisor). 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.
    ///
    /// Two records can name that process, and both are consulted:
    /// [`supervisor_pid_file`], which CAR writes for every child it spawns
    /// (car#732 — so the guard is no longer opt-in, which it was when only
    /// an agent writing its own file counted, and 1 of 5 agents did), and
    /// [`agent_owned_pid_file`], which an agent may write for itself and
    /// which is the only thing that can catch an instance CAR never
    /// spawned. An agent that writes neither falls through to the spawn
    /// behavior unchanged.
    pub async fn start_all(&self) -> Vec<String> {
        // Clear orphaned empties before starting anything, so a fleet view
        // taken right after boot describes the agents that exist rather than
        // every process that ever ran (#937).
        self.sweep_orphan_logs().await;
        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) = external_agent_pid(&spec.id) {
                // Report the resolved path rather than a hardcoded `~/.car`
                // one: under a CAR_HOME state root the literal would send an
                // operator to a file that isn't there. It also says which of
                // the two records blocked — CAR's or the agent's (car#931).
                tracing::warn!(
                    agent = %spec.id,
                    pid = ext.pid,
                    pid_file = %ext.path.display(),
                    "agent already running externally. Skipping auto_start — call agents.start once the external instance exits to take over supervision.",
                );
                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,
            blocked_by_pid: slot.runtime.blocked_by_pid,
        }
    }

    /// Persist an upsert without rewriting any other agent's manifest.
    ///
    /// `manifest.toml` is operator-editable and can contain comments that the
    /// in-memory [`AgentSpec`] does not retain. Re-serializing every slot here
    /// would silently replace unrelated on-disk edits with stale memory just
    /// because another agent changed.
    async fn persist_upsert(&self, id: &str) -> Result<(), SupervisorError> {
        self.persist_with_manifest_scope(Some(id)).await
    }

    async fn persist(&self) -> Result<(), SupervisorError> {
        self.persist_with_manifest_scope(None).await
    }

    /// Persist the legacy aggregate plus either one changed manifest (`Some`)
    /// or the complete manifest set (`None`, used after removal so stale agent
    /// directories can be reaped).
    async fn persist_with_manifest_scope(
        &self,
        manifest_id: Option<&str>,
    ) -> 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 the
        // changed AgentSpec to `agents/<id>/manifest.toml`. 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(());
        }
        // An upsert owns only its agent's file. A removal uses the full scope
        // so the remaining mirrors are synchronized before stale dirs are
        // reaped.
        for spec in manifest.agents.iter().filter(|spec| match manifest_id {
            Some(id) => spec.id == id,
            None => true,
        }) {
            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"
                );
            }
        }
        // Upsert cannot make an id stale, and therefore must not touch any
        // unrelated manifest directory. Removal uses the full scope and reaps
        // manifest dirs whose ids are no longer in state.
        if manifest_id.is_none() {
            crate::manifest::reap_stale(&agents_dir, &current_ids);
        }
        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();
        }
    }

    /// Record (or clear) the live untracked pid that is preventing this
    /// agent from starting. Separate from [`Self::set_status`] because the
    /// blocker must survive the very transition that recorded it.
    async fn set_blocked_by(&self, id: &str, pid: Option<i32>) {
        let mut state = self.state.write().await;
        if let Some(slot) = state.get_mut(id) {
            slot.runtime.blocked_by_pid = pid;
        }
    }

    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;
            // Any ordinary transition clears a stale blocker: the reason
            // recorded earlier no longer explains the current state.
            // `set_blocked_by` re-sets it after this call when it still holds.
            slot.runtime.blocked_by_pid = None;
        }
    }

    /// 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) = external_agent_pid(&spec.id) {
            tracing::warn!(
                agent = %id,
                pid = ext.pid,
                pid_file = %ext.path.display(),
                "external agent instance still alive (pid file). Supervisor refusing to double-spawn; sleeping {}s then re-checking. Leave it and supervision resumes automatically once it exits. If you want it gone, check that pid {} really is this agent first (`ps -p <pid> -o command=`) — a pid file left by a killed agent can be recycled by an unrelated process, in which case remove the pid file rather than killing anything — then `car restart {}`.",
                spec.backoff_secs.max(5),
                ext.pid,
                id
            );
            supervisor
                .set_status(&id, AgentStatus::Backoff, None, None)
                .await;
            // After set_status, which clears it — this is the reason for the
            // Backoff and has to outlive the transition that caused it.
            supervisor.set_blocked_by(&id, Some(ext.pid)).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;
                // Written before the status flips to Running: if the host is
                // killed in the window between spawn and this line, the pid
                // file is what tells the next supervisor an orphan exists.
                write_supervisor_pid_file(&id, pid);
                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;
                        clear_supervisor_pid_file(&id, pid);
                        reap_empty_logs(supervisor.log_dir(), &id);
                        return;
                    }
                    res = child.wait() => {
                        let code = match res {
                            Ok(status) => status.code().unwrap_or(-1),
                            Err(_) => -1,
                        };
                        // The child is gone, so the file no longer describes
                        // anything live. Leaving it would make the next spawn
                        // refuse on a ghost.
                        clear_supervisor_pid_file(&id, pid);
                        // The process is gone; if it never wrote anything, the
                        // file it left behind describes nothing (#937).
                        reap_empty_logs(supervisor.log_dir(), &id);
                        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>>,
}

/// Route a Windows `.cmd`/`.bat` shim through `cmd /C`. npm installs
/// `npx`/`npm`/`yarn`/`pnpm` as batch shims, and a supervised agent whose
/// `command` is one of those fails to spawn with os error 193 ("%1 is not a
/// valid Win32 application") because `CreateProcess` can't execute a batch
/// file and `Command::new` only appends `.exe` when resolving PATH. A local
/// twin of `car_engine::spawn::program_command`: car-registry depends on
/// neither car-engine nor car-external-agents (which each carry the same
/// shim-routing logic), so the logic is duplicated rather than adding a
/// cross-crate dependency just for this.
fn program_command(program: &str) -> tokio::process::Command {
    #[cfg(windows)]
    {
        use std::path::Path;
        let has_batch_ext = |p: &Path| {
            p.extension()
                .and_then(|e| e.to_str())
                .map(|e| {
                    let e = e.to_ascii_lowercase();
                    e == "cmd" || e == "bat"
                })
                .unwrap_or(false)
        };
        let shim: Option<std::path::PathBuf> = {
            let p = Path::new(program);
            if p.extension().is_some() || p.components().count() > 1 {
                has_batch_ext(p).then(|| p.to_path_buf())
            } else if let Some(path) = std::env::var_os("PATH") {
                let pathext =
                    std::env::var_os("PATHEXT").unwrap_or_else(|| ".COM;.EXE;.BAT;.CMD".into());
                let exts: Vec<String> = pathext
                    .to_string_lossy()
                    .split(';')
                    .filter(|e| !e.is_empty())
                    .map(|e| e.to_string())
                    .collect();
                let mut found = None;
                'outer: for dir in std::env::split_paths(&path) {
                    for ext in &exts {
                        let cand = dir.join(format!("{program}{ext}"));
                        if cand.is_file() {
                            found = has_batch_ext(&cand).then_some(cand);
                            break 'outer;
                        }
                    }
                }
                found
            } else {
                None
            }
        };
        if let Some(shim) = shim {
            let mut c = tokio::process::Command::new("cmd");
            c.arg("/C").arg(shim);
            // The shim is a batch file that resolves its real interpreter
            // (usually `node`) through PATH — and cmd.exe DROPS a PATH over
            // ~8191 chars, handing the shim an empty one. A supervised agent
            // would then die at startup with "'node' is not recognized".
            // `None` (the usual case) inherits the environment unchanged.
            if let Some(path) = car_winenv::cmd_path_override() {
                c.env("PATH", path);
            }
            return c;
        }

        // The supervised command can also BE `cmd.exe` itself — a perfectly
        // legitimate spec (`command = %COMSPEC%`, `args = ["/C", "<prog> …"]`),
        // and what the cascade-kill fixtures use. That child does its own PATH
        // lookup for `<prog>`, so it hits the same over-long-PATH drop as a shim:
        // cmd receives an EMPTY PATH and the inner program is "not recognized".
        // Covering only the shim case left this open, and the workspace-wide
        // Windows gate caught it — the very same test spawns `cmd /C ping`, and
        // `ping` stopped resolving once the runner's PATH crossed the limit.
        if is_cmd_exe(program) {
            let mut c = tokio::process::Command::new(program);
            if let Some(path) = car_winenv::cmd_path_override() {
                c.env("PATH", path);
            }
            return c;
        }
    }
    tokio::process::Command::new(program)
}

/// True when `program` is Windows' command interpreter (by file name, so an
/// absolute `%COMSPEC%` path or a bare `cmd`/`cmd.exe` all match).
#[cfg(windows)]
fn is_cmd_exe(program: &str) -> bool {
    std::path::Path::new(program)
        .file_name()
        .and_then(|f| f.to_str())
        .map(|f| f.eq_ignore_ascii_case("cmd.exe") || f.eq_ignore_ascii_case("cmd"))
        .unwrap_or(false)
}

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

    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 = program_command(&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::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 => {
                signal_process_tree(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;
                }
                signal_process_tree(pid_i, libc_sigkill());
            }
            StopSignal::Kill => {
                signal_process_tree(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`).
///
/// NOTE: `apps/host-windows/src/jobobject.rs` carries a copy of this
/// wrapper (the tray host lives outside the workspace). A fix here
/// almost certainly applies there too — keep them in step.
///
/// Exported (`pub`) so in-workspace consumers that spawn their own bounded
/// child process trees — e.g. the coder/assistant shell tool in
/// `car-server-core` — can get the same atomic tree-kill on timeout instead of
/// orphaning grandchildren (only `kill_on_drop`, which reaps the direct child).
#[cfg(target_os = "windows")]
pub 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 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 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 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 process_group_of(pid: i32) -> i32 {
    extern "C" {
        fn getpgid(pid: i32) -> i32;
    }
    unsafe { getpgid(pid) }
}

/// Signal the child's whole process group when we can prove it is its own
/// group leader; otherwise signal just the pid (car#732).
///
/// `spawn_child` calls `setsid()`, which makes the child a session and
/// process-group leader with `pgid == pid`. Until now nothing used that
/// group as a kill unit — the Unix path signalled the direct pid only — so
/// grandchildren leaked by construction, while the Windows path next door
/// cascades atomically via `TerminateJobObject`. This is the Unix analogue.
///
/// **The `pgid == pid` check is a safety interlock, not a formality.** If
/// `setsid()` failed, the child is still in the *supervisor's* process
/// group, and `kill(-pgid, SIGKILL)` would then kill car-server and every
/// other agent it supervises. Verifying the group leader is what makes the
/// blast radius provably the child's own tree; when it does not hold we
/// fall back to the single-pid behaviour, which is exactly what shipped
/// before and leaks nothing new.
#[cfg(unix)]
fn signal_process_tree(pid: i32, sig: i32) {
    if pid > 1 && process_group_of(pid) == pid {
        // Negative pid = "the process group with this id" for kill(2).
        send_signal(-pid, sig);
    } else {
        send_signal(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 pid file the **supervisor** writes for a child it spawned:
/// `run/<id>.supervisor.pid` under the CAR state root (`~/.car` by
/// default, `$CAR_HOME` when set).
///
/// The supervisor writes this at spawn and removes it when the child exits
/// under its watch (car#732). It used to only ever read, leaving each agent
/// to write its own — which made the double-spawn guard opt-in, and in
/// practice 1 of 5 agents opted in. An orphaned agent from the other four
/// kept its port and every respawn died on bind, with nothing anywhere
/// saying why. Requiring each agent to re-report a pid the supervisor
/// already knows is what left that gap.
///
/// **The `.supervisor` infix is the whole point of this function** and is
/// reserved (car#931). Through 0.48 the supervisor wrote its record to
/// [`agent_owned_pid_file`] — `run/<id>.pid` — the same path a singleton
/// agent conventionally uses as its own exclusive lock. Two writers, two
/// incompatible meanings: CAR recorded "the child I spawned", the agent
/// recorded "the instance that owns the lock". A singleton agent therefore
/// booted, read the lock, found a live pid — **its own**, written by the
/// supervisor moments earlier — concluded another instance already held the
/// lock, and exited. Under `restart: on_failure` that became a restart storm
/// ending at `max_restarts` with nothing running, and no way to spawn the
/// agent at all: an agent that ran perfectly standalone could not be
/// supervised. CAR owns this path; `run/<id>.pid` belongs to the agent.
///
/// Returns `None` when the state root doesn't resolve — no `$CAR_HOME`
/// and neither `$HOME` nor `%USERPROFILE%` set (e.g. PID 1 supervisor
/// on a stripped-down container). Without a root, the convention has
/// nowhere to live and the guard degrades to a no-op.
fn supervisor_pid_file(agent_id: &str) -> Option<std::path::PathBuf> {
    Some(
        car_home::root()?
            .join("run")
            .join(format!("{agent_id}.supervisor.pid")),
    )
}

/// Path of the pid file an **agent** may write for itself by the
/// long-standing convention: `run/<id>.pid`.
///
/// CAR **reads** this and never writes, truncates or removes it (car#931).
/// Reading it is what the double-spawn guard was originally built on — an
/// agent started outside CAR is invisible to the supervisor's own records,
/// and honouring the agent's file is what stopped two traders running
/// against one brokerage account. Writing it is what broke singleton agents.
/// Read-only keeps the first and ends the second: a reader creates no
/// collision.
///
/// Consequently a *stale* file here is left in place rather than reaped —
/// unlinking an operator's lock file is not ours to do. A dead pid is simply
/// not a blocker, re-evaluated on every read, and the warning that names a
/// live one names this path so an operator can see whose file it is.
///
/// **The cost of not reaping is pid reuse, and it is unbounded.** An agent
/// killed without cleanup (SIGKILL, OOM) leaves this file forever. If the OS
/// later recycles that pid to an unrelated process, [`pid_alive`] — a bare
/// `kill(pid, 0)` with no identity check — reports a blocker again, and the
/// supervisor parks the agent in `Backoff` indefinitely with no CAR-side
/// remedy: the operator has to remove the file. Reaping it ourselves would
/// bound that window, at the price of unlinking a lock file whose owner may
/// still be mid-write. We take the wedge over the unlink, because the wedge
/// is loud (a `Backoff` naming the pid and the path) and the unlink is
/// silent. That is also why the diagnostic tells the operator to check the
/// pid before killing it rather than to kill it outright.
fn agent_owned_pid_file(agent_id: &str) -> Option<std::path::PathBuf> {
    Some(
        car_home::root()?
            .join("run")
            .join(format!("{agent_id}.pid")),
    )
}

/// Record a freshly spawned child's pid so a supervisor that dies without
/// running any Drop still leaves evidence of what it started (car#732).
///
/// `kill_on_drop(true)` covers an in-process teardown, but nothing runs
/// when `car-server` itself is SIGKILLed or the host app quits — which is
/// exactly when a child outlives its supervisor and keeps the port. The
/// file is the only artifact that survives that.
///
/// Best-effort: a failure here is logged and does not fail the spawn. The
/// agent running unprotected is strictly better than it not running.
fn write_supervisor_pid_file(agent_id: &str, pid: u32) {
    let Some(path) = supervisor_pid_file(agent_id) else {
        return;
    };
    if let Some(dir) = path.parent() {
        if let Err(e) = std::fs::create_dir_all(dir) {
            tracing::warn!(
                agent = %agent_id, path = %dir.display(), error = %e,
                "could not create pid-file directory; double-spawn guard degraded for this agent"
            );
            return;
        }
    }
    if let Err(e) = std::fs::write(&path, pid.to_string()) {
        tracing::warn!(
            agent = %agent_id, path = %path.display(), error = %e,
            "could not write pid file; double-spawn guard degraded for this agent"
        );
    }
}

/// Remove the pid file for a child that exited under our supervision.
///
/// Only removes when the file still names `pid`. If an agent rewrote it
/// with its own (a descendant's) pid, that process may still be alive, and
/// deleting the file would discard the very evidence the guard needs.
/// Delete this agent's log files if the process never wrote a byte to them
/// (Parslee-ai/car#937).
///
/// `spawn_child` must open both streams *before* it can hand them to the child,
/// so the files exist from the instant of spawn whether or not anything is ever
/// written. For a long-lived agent that is invisible. For a short-lived process
/// it leaves a permanent, empty artifact — and because ids for on-demand
/// backends embed a pid (`vllm-mlx-<pid>`), every single start mints a new one.
/// On the reported machine that was 61 orphans, 60 of them zero bytes,
/// accumulated over two months with nothing to remove them, against 21 real
/// agents. A consumer listing the directory to answer "what agents exist and
/// are they healthy" saw 78 entries and reported 60 of them as idle.
///
/// Reaping at exit is what makes the file's lifetime match the process's,
/// without piping the streams through the supervisor purely to delay a
/// `File::create`.
///
/// **Empty is the whole condition.** A log with any content is evidence and is
/// never touched, however dead the process or unregistered the id. So the worst
/// case is a file that should have been removed and was not — never a lost log.
fn reap_empty_logs(log_dir: &Path, agent_id: &str) {
    for stream in ["stdout", "stderr"] {
        let path = log_dir.join(format!("{agent_id}.{stream}.log"));
        match std::fs::metadata(&path) {
            Ok(meta) if meta.len() == 0 => {
                let _ = std::fs::remove_file(&path);
            }
            _ => {}
        }
    }
}

fn clear_supervisor_pid_file(agent_id: &str, pid: u32) {
    let Some(path) = supervisor_pid_file(agent_id) else {
        return;
    };
    match std::fs::read_to_string(&path) {
        Ok(content) if content.trim() == pid.to_string() => {
            let _ = std::fs::remove_file(&path);
        }
        // Someone else's pid, or no file: leave it. `external_agent_pid`
        // reaps a stale one lazily on the next read.
        _ => {}
    }
}

/// A live process that already answers for an agent id, together with the
/// pid file that named it. The path is carried so a diagnostic can say
/// *whose* record is blocking — CAR's own or the agent's — which is the
/// distinction car#931 turned on.
#[derive(Debug, Clone)]
struct ExternalInstance {
    pid: i32,
    path: std::path::PathBuf,
}

/// Read a pid file and return the pid only if that process is alive.
///
/// `reap` says whether a file that turns out to be useless (dead pid,
/// unparseable content) may be deleted. True for CAR's own record, where a
/// ghost would otherwise make the next spawn refuse forever. False for the
/// agent-owned path, which is not ours to unlink (car#931) — a dead pid
/// there is simply not a blocker, re-evaluated on every read.
fn read_live_pid(agent_id: &str, path: &Path, reap: bool) -> Option<i32> {
    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(),
                reaped = reap,
                "agent pid file content unparseable"
            );
            if reap {
                let _ = std::fs::remove_file(path);
            }
            return None;
        }
    };
    if pid_alive(pid) {
        Some(pid)
    } else {
        tracing::info!(
            agent = %agent_id,
            pid,
            path = %path.display(),
            reaped = reap,
            "stale agent pid file (process gone)"
        );
        if reap {
            let _ = std::fs::remove_file(path);
        }
        None
    }
}

/// Returns the live process currently answering for `agent_id`, if any.
///
/// Two records can name one: [`supervisor_pid_file`], written by CAR for a
/// child it spawned, and [`agent_owned_pid_file`], written by an agent that
/// declares its own instance. Both are consulted — CAR's first, since it is
/// the authoritative record for anything CAR started — because they cover
/// different cases. CAR's record catches an orphan that outlived its
/// supervisor; the agent's catches an instance started entirely outside CAR,
/// which CAR has no record of at all.
///
/// "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<ExternalInstance> {
    if let Some(path) = supervisor_pid_file(agent_id) {
        if let Some(pid) = read_live_pid(agent_id, &path, true) {
            return Some(ExternalInstance { pid, path });
        }
    }
    let path = agent_owned_pid_file(agent_id)?;
    let pid = read_live_pid(agent_id, &path, false)?;
    Some(ExternalInstance { pid, path })
}

/// An agent id must also be a legal *peer name*, because an agent is addressed
/// by its id on every agent-to-agent surface.
///
/// These were two validators with two answers. This one accepted a leading `.`
/// or `-`, an embedded `..`, and any length; `car_peers::is_valid_peer_name`
/// rejects all three. An id in the gap — `.watcher`, say — registered and
/// attached fine and was then silently unreachable: `snapshot_attached` filters
/// it out of the peer directory, so it could never receive `agents.message`,
/// and `agents.chat` refuses it with a message naming neither the cause nor a
/// fix. Deferring to the stricter rule turns "quietly unaddressable forever"
/// into one error at the moment the id is chosen, which is the only point where
/// it is cheap to change.
///
/// Migration note: an already-registered agent whose id is in that gap now
/// fails to start. It could not be messaged before either — this makes the
/// existing breakage visible rather than introducing it.
fn validate_id(id: &str) -> Result<(), SupervisorError> {
    if !car_peers::is_valid_peer_name(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).
#[cfg(not(windows))]
const SCRATCH_REASON: &str = "command lives under a world-writable scratch directory \
                             (/tmp, /private/tmp, /var/tmp, /dev/shm)";
#[cfg(windows)]
const SCRATCH_REASON: &str = "command lives under a world-writable scratch directory \
                             (%TEMP%, %SystemRoot%\\Temp, %SystemDrive%\\Users\\Public)";

/// World-writable scratch-directory prefixes a command must never live under.
/// On Windows these are resolved at call time (the per-user temp dir is not a
/// fixed literal) and returned lower-cased with backslash separators and a
/// trailing separator so [`command_under_world_writable_scratch`] can do a
/// case-insensitive directory-prefix match. See
/// `car_external_agents::detection` for the mirror of this logic.
#[cfg(windows)]
fn windows_scratch_prefixes() -> Vec<String> {
    fn norm(s: &str) -> String {
        format!("{}\\", s.replace('/', "\\").trim_end_matches('\\')).to_ascii_lowercase()
    }
    let mut v = vec![norm(&std::env::temp_dir().to_string_lossy())];
    for var in ["TEMP", "TMP"] {
        if let Some(t) = std::env::var_os(var) {
            v.push(norm(&Path::new(&t).to_string_lossy()));
        }
    }
    let sysroot = std::env::var_os("SystemRoot")
        .map(|s| Path::new(&s).to_string_lossy().into_owned())
        .unwrap_or_else(|| r"C:\Windows".to_string());
    v.push(norm(&format!("{}\\Temp", sysroot.trim_end_matches('\\'))));
    let drive = std::env::var_os("SystemDrive")
        .map(|s| Path::new(&s).to_string_lossy().into_owned())
        .unwrap_or_else(|| "C:".to_string());
    v.push(norm(&format!(
        "{}\\Users\\Public",
        drive.trim_end_matches('\\')
    )));
    v
}

/// True when `command` resolves under a world-writable scratch directory — the
/// 2026-05 audit's staging-exploit denylist. The Unix arm keeps the classic
/// `/tmp` family; the Windows arm covers `%TEMP%`/`GetTempPath`,
/// `%SystemRoot%\Temp`, and `%SystemDrive%\Users\Public`, matched
/// case-insensitively (NTFS) and separator-agnostically.
#[cfg(not(windows))]
fn command_under_world_writable_scratch(command: &str) -> bool {
    const SCRATCH_PREFIXES: &[&str] = &["/tmp/", "/private/tmp/", "/var/tmp/", "/dev/shm/"];
    SCRATCH_PREFIXES.iter().any(|p| command.starts_with(p))
}

#[cfg(windows)]
fn command_under_world_writable_scratch(command: &str) -> bool {
    let cand = command.replace('/', "\\").to_ascii_lowercase();
    windows_scratch_prefixes()
        .iter()
        .any(|p| cand.starts_with(p.as_str()))
}

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.
    if command_under_world_writable_scratch(command) {
        return Err(SupervisorError::InvalidCommand {
            command: command.to_string(),
            reason: SCRATCH_REASON,
        });
    }
    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;
        }
        // On Windows a bare interpreter name ("node") lives on disk with an
        // executable extension ("node.exe") and the shell resolves it via
        // PATHEXT — so `dir.join("node")` never matches. Try the bare name
        // first (Unix, or a name that already carries an extension), then
        // each PATHEXT extension on Windows.
        // `mut` is only exercised by the #[cfg(windows)] push below.
        #[cfg_attr(not(windows), allow(unused_mut))]
        let mut candidates = vec![dir.join(name)];
        #[cfg(windows)]
        if std::path::Path::new(name).extension().is_none() {
            let pathext =
                std::env::var("PATHEXT").unwrap_or_else(|_| ".COM;.EXE;.BAT;.CMD".to_string());
            for ext in pathext.split(';').filter(|e| !e.is_empty()) {
                candidates.push(dir.join(format!("{name}{ext}"))); // ext has the leading dot
            }
        }
        for candidate in candidates {
            // 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());
    }
    // agents.json contains every supervisor-minted per-agent token. Repair
    // installations written by older releases before every read, not only on
    // the next registry mutation.
    repair_private_file(path)?;
    let bytes = std::fs::read(path)?;
    let manifest: Manifest = serde_json::from_slice(&bytes)?;
    Ok(manifest)
}

pub(crate) fn ensure_private_directory(path: &Path) -> Result<(), std::io::Error> {
    std::fs::create_dir_all(path)?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        let metadata = std::fs::metadata(path)?;
        if metadata.permissions().mode() & 0o777 != 0o700 {
            std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700))?;
        }
    }
    Ok(())
}

pub(crate) fn repair_private_file(path: &Path) -> Result<(), std::io::Error> {
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        let metadata = match std::fs::symlink_metadata(path) {
            Ok(metadata) => metadata,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
            Err(error) => return Err(error),
        };
        if !metadata.file_type().is_file() {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!("private manifest is not a regular file: {}", path.display()),
            ));
        }
        if metadata.permissions().mode() & 0o777 != 0o600 {
            std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))?;
        }
    }
    #[cfg(not(unix))]
    let _ = path;
    Ok(())
}

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",
        )
    })?;
    ensure_private_directory(parent)?;
    let tmp = parent.join(format!(
        ".{}.{}.tmp",
        path.file_name()
            .and_then(|s| s.to_str())
            .unwrap_or("supervisor-write"),
        uuid::Uuid::new_v4().simple()
    ));
    let json = serde_json::to_vec_pretty(value)?;
    let write_result = (|| -> Result<(), std::io::Error> {
        let mut options = std::fs::OpenOptions::new();
        options.write(true).create_new(true);
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            options.mode(0o600);
        }
        let mut file = options.open(&tmp)?;
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            file.set_permissions(std::fs::Permissions::from_mode(0o600))?;
        }
        file.write_all(&json)?;
        file.sync_all()?;
        drop(file);
        std::fs::rename(&tmp, path)?;
        repair_private_file(path)?;
        Ok(())
    })();
    if let Err(error) = write_result {
        let _ = std::fs::remove_file(&tmp);
        return Err(error.into());
    }
    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)
    }

    /// car#1106: `stop_all` must take down an agent's whole process tree, and
    /// give the agent a catchable signal so it can stop its own children.
    ///
    /// The shape that broke in production: a supervised daemon that itself
    /// supervises a worker. Killing only the direct child reparents the worker
    /// to PID 1, where it keeps running — and for the reporting user that meant
    /// two processes trading one live brokerage account.
    ///
    /// Asserts on the GRANDCHILD, because the direct child dying was never the
    /// broken part.
    #[cfg(unix)]
    #[tokio::test]
    async fn stop_all_takes_down_the_whole_tree_not_just_the_direct_child() {
        let (tmp, s) = temp_supervisor();
        let marker = tmp.path().join("grandchild.pid");
        // A "daemon" that spawns a "worker" and records the worker's pid, then
        // waits. Nothing here traps SIGTERM: the tree signal has to be what
        // stops the worker.
        let script = format!(
            "sh -c 'echo $$ > {}; sleep 30' & sleep 30",
            marker.display()
        );
        let spec = AgentSpec {
            id: "tree".into(),
            name: "tree".into(),
            command: "/bin/sh".to_string(),
            args: vec!["-c".into(), script],
            ..echo_spec("tree", "x")
        };
        s.upsert(spec).await.unwrap();
        s.start("tree").await.unwrap();

        // Wait for the grandchild to exist and publish its pid.
        let mut worker_pid = None;
        for _ in 0..50 {
            if let Ok(txt) = std::fs::read_to_string(&marker) {
                if let Ok(pid) = txt.trim().parse::<i32>() {
                    worker_pid = Some(pid);
                    break;
                }
            }
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        }
        let worker_pid = worker_pid.expect("grandchild never started");
        assert!(pid_alive(worker_pid), "precondition: the worker is running");

        let straggling = s.stop_all(std::time::Duration::from_secs(20)).await;
        assert!(
            straggling.is_empty(),
            "everything should stop inside the budget, got {straggling:?}"
        );

        // The grandchild is the assertion. Allow a moment for the signal to be
        // reaped, then require it gone.
        for _ in 0..30 {
            if !pid_alive(worker_pid) {
                return;
            }
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        }
        // Do not leave a stray process behind if we are about to fail.
        signal_process_tree(worker_pid, libc_sigkill());
        panic!(
            "grandchild {worker_pid} survived stop_all — this is car#1106: the \
             worker outlives the daemon that owns it"
        );
    }

    /// `stop_all` on an empty supervisor is a no-op, not an error or a wait.
    #[tokio::test]
    async fn stop_all_with_no_agents_returns_immediately() {
        let (_tmp, s) = temp_supervisor();
        let t0 = std::time::Instant::now();
        let straggling = s.stop_all(std::time::Duration::from_secs(30)).await;
        assert!(straggling.is_empty());
        assert!(
            t0.elapsed() < std::time::Duration::from_secs(2),
            "an empty supervisor must not burn the budget"
        );
    }

    fn echo_spec(id: &str, message: &str) -> AgentSpec {
        // Portable long-running "echo then sleep" spec. `validate_command`
        // requires an absolute path, so resolve the platform shell absolutely.
        #[cfg(windows)]
        let (command, args) = (
            std::env::var("COMSPEC").unwrap_or_else(|_| r"C:\Windows\System32\cmd.exe".to_string()),
            // `& ping -n 31 localhost` ≈ `sleep 30` on Windows (no `sleep` builtin).
            vec![
                "/C".into(),
                format!("echo {message}& ping -n 31 127.0.0.1 >nul"),
            ],
        );
        #[cfg(unix)]
        let (command, args) = (
            "/bin/sh".to_string(),
            vec!["-c".into(), format!("echo {message}; sleep 30")],
        );
        AgentSpec {
            id: id.into(),
            name: id.into(),
            command,
            args,
            cwd: None,
            env: Default::default(),
            restart: RestartPolicy::Never,
            max_restarts: 1,
            backoff_secs: 1,
            auto_start: false,
            token: String::new(),
            method_allowlist: None,
            capabilities: Vec::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 upsert_preserves_other_agents_manifest_bytes() {
        let (tmp, s) = temp_supervisor();
        s.upsert(echo_spec("manifest-a", "a")).await.unwrap();

        let manifest_a = tmp.path().join("agents/manifest-a/manifest.toml");
        let generated = std::fs::read_to_string(&manifest_a).unwrap();
        let generated_name = "name = \"manifest-a\"";
        assert!(
            generated.contains(generated_name),
            "fixture must contain the generated agent name"
        );
        let edited = format!(
            "# operator comment that must survive byte-for-byte\n{}",
            generated.replace(generated_name, "name = \"edited outside CAR\"")
        );
        std::fs::write(&manifest_a, edited.as_bytes()).unwrap();

        s.upsert(echo_spec("manifest-b", "b")).await.unwrap();

        assert_eq!(
            std::fs::read(&manifest_a).unwrap(),
            edited.as_bytes(),
            "upserting B must not rewrite A from stale in-memory state"
        );
    }

    #[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.
    ///
    /// Tokio time is paused and advanced explicitly across the injected
    /// backoff, while the real `/bin/sh` child retains one bounded process
    /// integration.
    #[cfg(not(target_os = "windows"))]
    #[tokio::test(start_paused = true)]
    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(),
            method_allowlist: None,
            capabilities: Vec::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;
        for expected_count in 1..=3 {
            let wall_deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
            loop {
                let snap = s.list().await;
                peak_restart_count = peak_restart_count.max(snap[0].restart_count);
                if snap[0].restart_count >= expected_count {
                    if matches!(snap[0].status, AgentStatus::Errored) {
                        terminal = Some(snap[0].clone());
                        break;
                    }
                    if expected_count < 3 {
                        break;
                    }
                }
                assert!(
                    std::time::Instant::now() < wall_deadline,
                    "real child did not publish restart_count={expected_count} within 2s: {snap:?}"
                );
                tokio::task::yield_now().await;
            }
            if expected_count <= 2 {
                tokio::time::advance(std::time::Duration::from_secs(3)).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.
            // `validate_command` requires an absolute path (bare "cmd.exe"
            // is rejected as a PATH lookup), so resolve the real interpreter
            // via %COMSPEC% (→ C:\Windows\System32\cmd.exe).
            command: std::env::var("COMSPEC")
                .unwrap_or_else(|_| r"C:\Windows\System32\cmd.exe".to_string()),
            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(),
            method_allowlist: None,
            capabilities: Vec::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"
        );
    }

    /// Cross-platform liveness probe for tests. The product `pid_alive` is
    /// a deliberate no-op on Windows (the pid-file double-spawn guard is
    /// Unix-only), so a test that asserts a child is *alive* on Windows must
    /// probe the OS directly rather than trust `pid_alive`.
    #[cfg(unix)]
    fn test_pid_alive(pid: i32) -> bool {
        pid_alive(pid)
    }
    #[cfg(windows)]
    fn test_pid_alive(pid: i32) -> bool {
        std::process::Command::new("tasklist")
            .args(["/FI", &format!("PID eq {pid}"), "/NH", "/FO", "CSV"])
            .output()
            .map(|o| String::from_utf8_lossy(&o.stdout).contains(&format!("\"{pid}\"")))
            .unwrap_or(false)
    }

    #[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 !test_pid_alive(pid1) {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
        assert!(
            !test_pid_alive(pid1),
            "prior child must be reaped, not orphaned"
        );
        assert!(test_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. Override echo_spec's args
        // with a one-shot echo, per platform shell (cmd /C vs sh -c).
        #[cfg(unix)]
        {
            spec.args = vec!["-c".into(), "echo line-from-child".into()];
        }
        #[cfg(windows)]
        {
            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");
    }

    /// Parslee-ai/car#937 — the sweep must clear inert orphans and must not
    /// touch anything else.
    ///
    /// Every negative case here is a distinct way this could destroy something
    /// it should not, which is the whole reason the predicate is three
    /// conditions wide rather than "is it empty".
    #[tokio::test]
    async fn sweep_removes_only_empty_unregistered_logs() {
        let (_tmp, s) = temp_supervisor();
        std::fs::create_dir_all(s.log_dir()).unwrap();

        let touch = |id: &str, stream: &str, body: &str| {
            std::fs::write(s.log_dir().join(format!("{id}.{stream}.log")), body).unwrap();
        };

        // The reported case: an on-demand backend's per-pid id, long gone,
        // never wrote a byte. 60 of 61 on the operator's machine looked
        // exactly like this.
        touch("vllm-mlx-4242", "stdout", "");
        touch("vllm-mlx-4242", "stderr", "");
        // Unregistered but it LOGGED something — that is evidence, and the
        // process being dead does not make it disposable.
        touch("vllm-mlx-9999", "stdout", "panicked at ...");
        // Registered agent, currently empty. `agents.list` links to this path;
        // removing it would break a link to an agent that plainly exists.
        s.upsert(echo_spec("real-agent", "hi")).await.unwrap();
        touch("real-agent", "stdout", "");
        // Not one of ours at all.
        std::fs::write(s.log_dir().join("notes.txt"), "").unwrap();

        let removed = s.sweep_orphan_logs().await;
        assert_eq!(removed, 2, "both streams of the one inert orphan");

        let gone = |p: &str| !s.log_dir().join(p).exists();
        assert!(gone("vllm-mlx-4242.stdout.log"), "inert orphan must go");
        assert!(gone("vllm-mlx-4242.stderr.log"), "inert orphan must go");
        assert!(
            s.log_dir().join("vllm-mlx-9999.stdout.log").exists(),
            "a log with content is never removed, however dead its process"
        );
        assert!(
            s.log_dir().join("real-agent.stdout.log").exists(),
            "a registered agent owns its log even while empty"
        );
        assert!(
            s.log_dir().join("notes.txt").exists(),
            "the sweep must only ever consider *.stdout.log / *.stderr.log"
        );
    }

    /// The other half: exit-time reaping, which is what stops the directory
    /// growing in the first place. Same empty-only rule.
    #[test]
    fn reap_removes_empty_logs_and_keeps_written_ones() {
        let (_tmp, s) = temp_supervisor();
        std::fs::create_dir_all(s.log_dir()).unwrap();

        std::fs::write(s.log_dir().join("gone.stdout.log"), "").unwrap();
        std::fs::write(s.log_dir().join("gone.stderr.log"), "").unwrap();
        reap_empty_logs(s.log_dir(), "gone");
        assert!(!s.log_dir().join("gone.stdout.log").exists());
        assert!(!s.log_dir().join("gone.stderr.log").exists());

        // Wrote to stdout, nothing to stderr: keep the one that has content,
        // reap the one that does not. Mixed is the common real shape.
        std::fs::write(s.log_dir().join("mixed.stdout.log"), "started\n").unwrap();
        std::fs::write(s.log_dir().join("mixed.stderr.log"), "").unwrap();
        reap_empty_logs(s.log_dir(), "mixed");
        assert!(
            s.log_dir().join("mixed.stdout.log").exists(),
            "content is never discarded"
        );
        assert!(!s.log_dir().join("mixed.stderr.log").exists());

        // Absent files must not panic — a restart reaps an id whose logs the
        // previous exit already removed.
        reap_empty_logs(s.log_dir(), "never-existed");
    }

    /// 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);
    }

    #[cfg(windows)]
    #[tokio::test]
    async fn validate_command_rejects_windows_temp_prefix() {
        // Stage a real file under %TEMP% (a world-writable staging dir on
        // Windows) so the rejection is purely about the prefix denylist. This
        // is the exploit chain the Unix `/tmp` denylist blocks — it must be
        // blocked on Windows too.
        let bin = std::env::temp_dir().join("car-registry-validate-test.exe");
        std::fs::write(&bin, b"MZ").unwrap();

        let (_tmp, s) = temp_supervisor();
        let mut spec = echo_spec("scratch-win", "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 on Windows %TEMP%, 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");
        // A directory that exists and is absolute on the host platform.
        spec.command = if cfg!(windows) {
            r"C:\Windows".into()
        } else {
            "/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");
        // Absolute path containing a `..` segment on the host platform.
        spec.command = if cfg!(windows) {
            r"C:\Windows\..\Windows\System32\cmd.exe".into()
        } else {
            "/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();
    }

    // Unix-specific: `sh` (and the file_name == "sh" assertion) don't hold
    // on Windows, where the resolver would return `sh.exe`. Cross-platform
    // interpreter resolution is covered by the hermetic tests in manifest.rs,
    // which resolve the test binary itself rather than requiring any particular
    // interpreter to be installed (car#735).
    #[cfg(unix)]
    #[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 method_scope_is_validated_bound_and_retained_with_its_token() {
        let (_tmp, s) = temp_supervisor();
        let mut spec = echo_spec("scoped-token", "x");
        spec.token = "explicit-scoped-token".into();
        spec.method_allowlist = Some(vec![
            "mail.messages".into(),
            "infer".into(),
            "mail.messages".into(),
        ]);
        let agent = s.upsert(spec).await.unwrap();
        assert_eq!(
            agent.spec.method_allowlist,
            Some(vec!["infer".into(), "mail.messages".into()])
        );
        let binding = s
            .authenticate_agent_token("scoped-token", "explicit-scoped-token")
            .await
            .unwrap()
            .expect("matching token");
        assert_eq!(binding.method_allowlist, agent.spec.method_allowlist);

        // An old client omitting both fields keeps the credential and its
        // restriction together rather than silently widening it.
        let again = s.upsert(echo_spec("scoped-token", "y")).await.unwrap();
        assert_eq!(again.spec.token, "explicit-scoped-token");
        assert_eq!(again.spec.method_allowlist, agent.spec.method_allowlist);

        // An explicitly supplied replacement token may choose the legacy
        // unrestricted state.
        let mut rotated = echo_spec("scoped-token", "z");
        rotated.token = "replacement-unrestricted-token".into();
        let rotated = s.upsert(rotated).await.unwrap();
        assert_eq!(rotated.spec.method_allowlist, None);
    }

    #[tokio::test]
    async fn upsert_rejects_unknown_and_host_only_method_scope_entries() {
        let (_tmp, s) = temp_supervisor();
        let mut spec = echo_spec("bad-scope", "x");
        spec.method_allowlist = Some(vec!["not.a.daemon.method".into(), "agents.upsert".into()]);
        let error = s.upsert(spec).await.unwrap_err();
        assert!(matches!(error, SupervisorError::InvalidMethodAllowlist(_)));
        let message = error.to_string();
        assert!(message.contains("not.a.daemon.method"), "{message}");
        assert!(message.contains("agents.upsert"), "{message}");
        assert!(s.get("bad-scope").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}"),
            }
        }
    }

    /// agents.json carries the token that authenticates each supervised child,
    /// so both its first write and every replacement must be owner-only. Boot
    /// also repairs files and parent directories created by affected releases;
    /// waiting for another upsert would leave dormant installations exposed.
    #[cfg(unix)]
    #[tokio::test]
    async fn agents_json_is_private_on_create_rewrite_and_startup_repair() {
        use std::os::unix::fs::PermissionsExt;

        fn mode(path: &Path) -> u32 {
            std::fs::metadata(path).unwrap().permissions().mode() & 0o777
        }

        let tmp = temp_tmpdir();
        let manifest = tmp.path().join("agents.json");
        let logs = tmp.path().join("logs");
        std::fs::set_permissions(tmp.path(), std::fs::Permissions::from_mode(0o755)).unwrap();

        let supervisor = Supervisor::with_paths(manifest.clone(), logs.clone()).unwrap();
        assert_eq!(mode(tmp.path()), 0o700, "CAR_HOME must be owner-only");
        supervisor
            .upsert(echo_spec("private-agent", "first"))
            .await
            .unwrap();
        assert_eq!(mode(&manifest), 0o600, "first write exposed agents.json");
        let persisted: Manifest =
            serde_json::from_slice(&std::fs::read(&manifest).unwrap()).unwrap();
        assert!(
            !persisted.agents[0].token.is_empty(),
            "fixture must prove permissions on a token-bearing file"
        );
        drop(supervisor);

        std::fs::set_permissions(&manifest, std::fs::Permissions::from_mode(0o644)).unwrap();
        std::fs::set_permissions(tmp.path(), std::fs::Permissions::from_mode(0o755)).unwrap();
        let supervisor = reboot_with_paths(manifest.clone(), logs);
        assert_eq!(mode(tmp.path()), 0o700, "startup did not repair CAR_HOME");
        assert_eq!(mode(&manifest), 0o600, "startup did not repair agents.json");

        std::fs::set_permissions(&manifest, std::fs::Permissions::from_mode(0o644)).unwrap();
        supervisor
            .upsert(echo_spec("private-agent", "updated"))
            .await
            .unwrap();
        assert_eq!(mode(&manifest), 0o600, "rewrite exposed agents.json");
    }

    #[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(),
                method_allowlist: None,
                capabilities: Vec::new(),
            }],
        );

        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(),
            method_allowlist: None,
            capabilities: Vec::new(),
        });
        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(),
                    method_allowlist: None,
                    capabilities: Vec::new(),
                },
                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(),
                    method_allowlist: None,
                    capabilities: Vec::new(),
                },
            ],
        );

        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(),
            method_allowlist: None,
            capabilities: Vec::new(),
        });
        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(),
                method_allowlist: None,
                capabilities: Vec::new(),
            }],
        );
        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(),
            method_allowlist: None,
            capabilities: Vec::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(),
            #[cfg(unix)]
            command: "/bin/sh".into(),
            #[cfg(windows)]
            command: std::env::var("COMSPEC")
                .unwrap_or_else(|_| r"C:\Windows\System32\cmd.exe".to_string()),
            #[cfg(unix)]
            args: vec!["-c".into(), "true".into()],
            #[cfg(windows)]
            args: vec!["/C".into(), "exit 0".into()],
            cwd: None,
            env: Default::default(),
            restart: RestartPolicy::Never,
            max_restarts: 1,
            backoff_secs: 1,
            auto_start: false,
            token: String::new(),
            method_allowlist: None,
            capabilities: Vec::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(),
            #[cfg(unix)]
            command: "/bin/sh".into(),
            #[cfg(windows)]
            command: std::env::var("COMSPEC")
                .unwrap_or_else(|_| r"C:\Windows\System32\cmd.exe".to_string()),
            #[cfg(unix)]
            args: vec!["-c".into(), "echo auto-installed; sleep 30".into()],
            #[cfg(windows)]
            args: vec![
                "/C".into(),
                "echo auto-installed& ping -n 31 127.0.0.1 >nul".into(),
            ],
            cwd: None,
            env: Default::default(),
            restart: RestartPolicy::Never,
            max_restarts: 1,
            backoff_secs: 1,
            auto_start: true,
            token: String::new(),
            method_allowlist: None,
            capabilities: Vec::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!(managed.spec.auto_start);
        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.
}

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

    /// Serializes `$HOME` mutation — process-global, like `$PATH`.
    static HOME_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());

    fn with_temp_home<T>(f: impl FnOnce() -> T) -> T {
        let _guard = HOME_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = tempfile::tempdir().expect("tempdir");
        let prior = std::env::var_os("HOME");
        // CAR_HOME outranks HOME in the state root, so it has to be cleared
        // too: a developer who exports it in their shell would otherwise send
        // these pid files to their own real state root — the exact escape
        // pointing HOME at a tempdir exists to prevent.
        let prior_car_home = std::env::var_os(car_home::ENV_VAR);
        std::env::remove_var(car_home::ENV_VAR);
        std::env::set_var("HOME", tmp.path());
        let out = f();
        match prior {
            Some(p) => std::env::set_var("HOME", p),
            None => std::env::remove_var("HOME"),
        }
        if let Some(p) = prior_car_home {
            std::env::set_var(car_home::ENV_VAR, p);
        }
        out
    }

    #[test]
    fn supervisor_written_pid_file_round_trips() {
        with_temp_home(|| {
            // Our own pid is guaranteed alive, which is what the guard checks.
            let me = std::process::id();
            write_supervisor_pid_file("agent-a", me);
            let path = supervisor_pid_file("agent-a").expect("path");
            assert!(path.exists(), "supervisor must write the pid file");

            // The liveness half is Unix-only, and deliberately so: `pid_alive`
            // is `#[cfg(not(unix))] -> false`, which makes the double-spawn
            // guard a documented no-op off Unix rather than something that
            // blocks legitimate spawns. So `external_agent_pid` returning None
            // on Windows is the CONTRACT, not a defect — this assertion was
            // asserting Unix semantics unconditionally and failed on Windows
            // the moment earlier fixes let the suite reach it (car#760).
            //
            // Only the assertion is gated, not the test: writing the file and
            // resolving its path are cross-platform and stay covered on Windows.
            #[cfg(unix)]
            assert_eq!(
                external_agent_pid("agent-a").map(|e| e.pid),
                Some(me as i32),
                "a live pid in the file must be reported as a blocker"
            );
            #[cfg(not(unix))]
            assert_eq!(
                external_agent_pid("agent-a").map(|e| e.pid),
                None,
                "off Unix the guard is a deliberate no-op (pid_alive is always false)"
            );
        });
    }

    #[test]
    fn clearing_removes_only_our_own_pid() {
        with_temp_home(|| {
            let me = std::process::id();
            write_supervisor_pid_file("agent-b", me);

            // A different pid in the file means an agent rewrote it with its
            // own; that process may still be alive, so removing the file would
            // discard exactly the evidence the guard needs.
            let path = supervisor_pid_file("agent-b").expect("path");
            std::fs::write(&path, (me + 1).to_string()).unwrap();
            clear_supervisor_pid_file("agent-b", me);
            assert!(path.exists(), "must not delete another writer's pid file");

            std::fs::write(&path, me.to_string()).unwrap();
            clear_supervisor_pid_file("agent-b", me);
            assert!(!path.exists(), "must remove our own pid file on exit");
        });
    }

    #[test]
    fn a_dead_pid_is_not_a_blocker_and_the_stale_file_is_reaped() {
        with_temp_home(|| {
            let path = supervisor_pid_file("agent-c").expect("path");
            std::fs::create_dir_all(path.parent().unwrap()).unwrap();
            // Very high pid that will not be live. If it somehow is, the
            // assertion below is the thing that would flag it.
            std::fs::write(&path, "4294967000").unwrap();
            assert!(external_agent_pid("agent-c").is_none());
        });
    }

    #[test]
    fn unparseable_pid_file_is_removed_rather_than_blocking_forever() {
        with_temp_home(|| {
            let path = supervisor_pid_file("agent-d").expect("path");
            std::fs::create_dir_all(path.parent().unwrap()).unwrap();
            std::fs::write(&path, "not-a-pid").unwrap();
            assert!(external_agent_pid("agent-d").is_none());
            assert!(
                !path.exists(),
                "garbage must not wedge the agent off forever"
            );
        });
    }

    /// The regression test for car#931. The supervisor's record must never
    /// land on `run/<id>.pid`, because that is the path a singleton agent
    /// conventionally uses as its own exclusive lock. When CAR wrote it, the
    /// agent booted, read the lock, found a live pid that was *its own* — put
    /// there by the supervisor that had just spawned it — decided another
    /// instance already owned the lock, and exited. One `agents.start` then
    /// produced no running agent at all.
    #[test]
    fn supervisor_never_writes_the_agent_owned_pid_path() {
        with_temp_home(|| {
            let me = std::process::id();
            write_supervisor_pid_file("trader", me);

            let ours = supervisor_pid_file("trader").expect("path");
            let theirs = agent_owned_pid_file("trader").expect("path");
            assert_ne!(ours, theirs, "the two records must not share a path");
            assert!(
                ours.exists(),
                "the supervisor must still record the child it spawned"
            );
            assert!(
                !theirs.exists(),
                "car#931: writing the agent's own lock path makes a singleton \
                 agent race itself and refuse to start"
            );

            // …and taking the record back down must not touch it either.
            std::fs::create_dir_all(theirs.parent().unwrap()).unwrap();
            std::fs::write(&theirs, me.to_string()).unwrap();
            clear_supervisor_pid_file("trader", me);
            assert!(!ours.exists(), "our own record is removed on exit");
            assert!(
                theirs.exists(),
                "car#931: the agent's lock file is not ours to unlink"
            );
        });
    }

    /// The half of the guard that predates car#732 and still matters: an
    /// instance started entirely outside CAR leaves no supervisor record, so
    /// the agent's own pid file is the only thing that can stop a double
    /// spawn. Reading it creates no collision — only writing it did.
    #[cfg(unix)]
    #[test]
    fn an_agent_written_pid_file_still_blocks_a_double_spawn() {
        with_temp_home(|| {
            let me = std::process::id();
            let theirs = agent_owned_pid_file("agent-e").expect("path");
            std::fs::create_dir_all(theirs.parent().unwrap()).unwrap();
            std::fs::write(&theirs, me.to_string()).unwrap();

            let found = external_agent_pid("agent-e").expect("live external instance");
            assert_eq!(found.pid, me as i32);
            assert_eq!(
                found.path, theirs,
                "the diagnostic must name the agent's file, not CAR's"
            );
        });
    }

    /// A dead pid in the agent's file is not a blocker — but unlike our own
    /// record, the file stays. Unlinking an operator's lock file is not the
    /// supervisor's call (car#931), and a dead pid costs nothing: it is
    /// re-evaluated on every read.
    #[test]
    fn a_stale_agent_written_pid_file_is_ignored_but_left_in_place() {
        with_temp_home(|| {
            let theirs = agent_owned_pid_file("agent-f").expect("path");
            std::fs::create_dir_all(theirs.parent().unwrap()).unwrap();
            // Very high pid that will not be live.
            std::fs::write(&theirs, "4294967000").unwrap();
            assert!(external_agent_pid("agent-f").is_none());
            assert!(theirs.exists(), "the agent's file is not ours to reap");

            std::fs::write(&theirs, "not-a-pid").unwrap();
            assert!(external_agent_pid("agent-f").is_none());
            assert!(
                theirs.exists(),
                "even garbage in the agent's file is not ours to delete"
            );
        });
    }

    /// The supervisor's own record wins when both exist: it is authoritative
    /// for anything CAR spawned, and its stale entries are reaped so a ghost
    /// cannot wedge an agent off forever.
    #[cfg(unix)]
    #[test]
    fn our_own_record_is_consulted_before_the_agents() {
        with_temp_home(|| {
            let me = std::process::id();
            write_supervisor_pid_file("agent-g", me);
            let theirs = agent_owned_pid_file("agent-g").expect("path");
            std::fs::write(&theirs, me.to_string()).unwrap();

            let found = external_agent_pid("agent-g").expect("live instance");
            assert_eq!(
                found.path,
                supervisor_pid_file("agent-g").expect("path"),
                "CAR's own record is the authoritative one"
            );

            // A stale record of ours falls through to the agent's file and is
            // reaped on the way past.
            let ours = supervisor_pid_file("agent-g").expect("path");
            std::fs::write(&ours, "4294967000").unwrap();
            let found = external_agent_pid("agent-g").expect("live instance");
            assert_eq!(found.path, theirs);
            assert!(!ours.exists(), "our stale record is reaped");
        });
    }

    /// The safety interlock on the tree-kill. If `setsid()` ever failed, the
    /// child would still share the supervisor's process group, and signalling
    /// that group would kill car-server and every other agent it supervises.
    #[cfg(unix)]
    #[test]
    fn tree_kill_only_targets_a_process_that_leads_its_own_group() {
        // This test process is (almost certainly) NOT its own group leader —
        // cargo spawns it — so the interlock must decline to signal the group.
        let me = std::process::id() as i32;
        let leads_own_group = process_group_of(me) == me;
        if leads_own_group {
            // Nothing to prove here; the interlock would correctly allow it.
            return;
        }
        // Signal 0 is the existence probe: it performs the same permission and
        // target resolution as a real signal without delivering one, so this
        // exercises the branch safely.
        signal_process_tree(me, 0);
        assert!(
            pid_alive(me),
            "the interlock must never widen the blast radius to our own group"
        );
    }

    #[cfg(unix)]
    #[test]
    fn process_group_lookup_reports_a_plausible_group() {
        let me = std::process::id() as i32;
        assert!(
            process_group_of(me) > 0,
            "getpgid on self must succeed; the interlock depends on it"
        );
    }
    /// An agent id and a peer name are the same namespace. The ids in the gap
    /// between the two old validators registered fine and were then silently
    /// unaddressable — filtered out of the peer directory, unable to receive
    /// `agents.message`, refused by `agents.chat`.
    #[test]
    fn an_id_that_cannot_be_addressed_as_a_peer_is_refused_at_registration() {
        for bad in [".watcher", "-lead", "a..b", ""] {
            assert!(
                validate_id(bad).is_err(),
                "`{bad}` is not a legal peer name, so it must not be a legal agent id"
            );
        }
        for good in ["milo", "car-assistant", "trader.v2", "a_b-1"] {
            assert!(validate_id(good).is_ok(), "`{good}` must still register");
        }
    }

    /// The two rules must not drift apart again.
    #[test]
    fn the_agent_id_rule_is_exactly_the_peer_name_rule() {
        for candidate in [
            "milo",
            ".watcher",
            "-lead",
            "a..b",
            "",
            "ok.id",
            "UPPER",
            "with space",
            "sym!",
        ] {
            assert_eq!(
                validate_id(candidate).is_ok(),
                car_peers::is_valid_peer_name(candidate),
                "divergence on `{candidate}`"
            );
        }
    }
}