car-registry 0.17.0

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

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

use crate::manifest::AgentManifest;

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

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

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

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

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

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

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

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

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

#[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<()>>,
}

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

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

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

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

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

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

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

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

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

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

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

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

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

        let mut state: HashMap<String, AgentSlot> = HashMap::new();
        for (id, spec) in by_id {
            state.insert(
                id,
                AgentSlot {
                    spec,
                    runtime: AgentRuntime::default(),
                    stop_tx: None,
                    task: None,
                },
            );
        }
        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| 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,
            })
            .collect();
        out.sort_by(|a, b| a.spec.id.cmp(&b.spec.id));
        out
    }

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

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

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

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

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

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

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

    /// Stop the agent and prevent the supervisor from respawning it.
    /// `Term` sends SIGTERM and waits up to `grace_secs` before
    /// escalating to SIGKILL; `Kill` skips the grace.
    pub async fn stop(
        &self,
        id: &str,
        signal: StopSignal,
    ) -> Result<ManagedAgent, SupervisorError> {
        validate_id(id)?;
        // Tell the supervisor task to exit and abort it. Then kill
        // any running child. Read the pid + senders inside the lock,
        // do the actual work outside.
        let (stop_tx, task, pid) = {
            let mut state = self.state.write().await;
            let slot = state
                .get_mut(id)
                .ok_or_else(|| SupervisorError::NotFound(id.to_string()))?;
            (slot.stop_tx.take(), slot.task.take(), slot.runtime.pid)
        };
        if let Some(tx) = stop_tx {
            let _ = tx.send(true);
        }
        if let Some(pid) = pid {
            kill_process(pid, signal, self.grace_secs).await;
        }
        if let Some(handle) = task {
            // Don't await — the task may be in the middle of a
            // backoff sleep. Aborting is cleaner than blocking the
            // caller for backoff_secs.
            handle.abort();
        }
        {
            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.snapshot(id)
            .await
            .ok_or_else(|| SupervisorError::NotFound(id.to_string()))
    }

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

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

    /// Read the last `n` lines from the agent's combined log
    /// (interleaved stdout + stderr by file order — stderr
    /// ordering relative to stdout isn't preserved across files).
    /// Returns an empty `Vec` when the log doesn't exist yet.
    pub async fn tail_log(&self, id: &str, n: usize) -> Result<Vec<String>, 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 mut combined = Vec::new();
        for path in [&stdout_path, &stderr_path] {
            if !path.exists() {
                continue;
            }
            let contents = tokio::fs::read_to_string(path).await?;
            for line in contents.lines() {
                combined.push(line.to_string());
            }
        }
        if combined.len() > n {
            let drop = combined.len() - n;
            combined.drain(0..drop);
        }
        Ok(combined)
    }

    async fn snapshot(&self, id: &str) -> Option<ManagedAgent> {
        let state = self.state.read().await;
        state.get(id).map(|slot| self.snapshot_locked(slot))
    }

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

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

    async fn spawn_supervision(&self, spec: AgentSpec) {
        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);
        }
    }

    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) {
            slot.runtime.status = status;
            slot.runtime.pid = pid;
            slot.runtime.started_at = started_at;
        }
    }

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

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

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

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

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

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

    let mut cmd = Command::new(&spec.command);
    cmd.args(&spec.args);
    if let Some(cwd) = &spec.cwd {
        cmd.current_dir(cwd);
    }
    // Default env (CAR_DAEMON_URL / CAR_AUTH_TOKEN — #172) goes in
    // first so the per-spec env below wins on conflict. Operators
    // who want to override the daemon-supplied URL (e.g. point a
    // child at a different car-server during development) can still
    // do so via `spec.env`.
    for (k, v) in default_env {
        cmd.env(k, v);
    }
    // Per-agent identity (#169). Always set — even when the daemon
    // is `--no-auth` and `spec.token` was minted but is unused. The
    // child can still call `session.auth { agent_id }` to bind its
    // connection to its supervised identity.
    cmd.env("CAR_AGENT_ID", &spec.id);
    if !spec.token.is_empty() {
        cmd.env("CAR_AGENT_TOKEN", &spec.token);
    }
    for (k, v) in &spec.env {
        cmd.env(k, v);
    }
    cmd.stdin(Stdio::null());
    cmd.stdout(Stdio::from(stdout));
    cmd.stderr(Stdio::from(stderr));
    // Detach from the parent's controlling terminal so SIGINT to
    // the supervisor doesn't propagate to children automatically;
    // we control kills via SIGTERM/SIGKILL.
    #[cfg(unix)]
    unsafe {
        // tokio::process::Command exposes pre_exec directly on
        // unix; setsid puts the child in its own process group so
        // signals to the supervisor don't propagate to it through
        // the controlling terminal.
        cmd.pre_exec(|| {
            if libc_setsid() == -1 {
                return Err(std::io::Error::last_os_error());
            }
            Ok(())
        });
    }
    let child = cmd.spawn()?;
    let pid = child.id().ok_or_else(|| {
        std::io::Error::new(std::io::ErrorKind::Other, "child spawned without pid")
    })?;
    Ok((child, pid))
}

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

async fn kill_process(pid: u32, signal: StopSignal, grace_secs: u64) {
    #[cfg(unix)]
    {
        let pid_i = pid as i32;
        match signal {
            StopSignal::Term => {
                send_signal(pid_i, libc_sigterm());
                let deadline = std::time::Duration::from_secs(grace_secs.max(1));
                let mut waited = std::time::Duration::ZERO;
                let step = std::time::Duration::from_millis(200);
                while waited < deadline {
                    if !pid_alive(pid_i) {
                        return;
                    }
                    tokio::time::sleep(step).await;
                    waited += step;
                }
                send_signal(pid_i, libc_sigkill());
            }
            StopSignal::Kill => {
                send_signal(pid_i, libc_sigkill());
            }
        }
    }
    #[cfg(not(unix))]
    {
        // Best-effort on non-unix: TerminateProcess equivalent isn't
        // wired here yet. Hosts that need it on Windows can extend
        // this branch.
        let _ = (pid, signal, grace_secs);
    }
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

        // Second boot — token already preserved in mirror, mirror
        // not rewritten (the migration only writes when the mirror
        // didn't already exist).
        let _s2 = Supervisor::with_paths(legacy, tmp.path().join("logs")).unwrap();
        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.
        drop(s1);
        let _s3 = Supervisor::with_paths(manifest, logs).unwrap();
    }

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

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

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

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

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

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

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

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

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

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

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

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

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