car-registry 0.47.0

//! File-based agent registry for cross-process discovery.
//!
//! Each running CAR-hosted agent writes a JSON file to
//! `~/.car/registry/<name>.json` so the CAR menubar / system tray
//! (and any other UI surface) can list running agents and link to
//! their bespoke dashboards. Closes [#111] —
//! deliberately tiny: ~50 LOC of API surface, no dashboard
//! framework, no opinionated UI, no auth, no process management
//! beyond bookkeeping.
//!
//! ## Design
//!
//! - **One file per agent.** `register()` writes
//!   `<dir>/<sanitized-name>.json` atomically (temp + rename).
//!   Concurrent registrations from different agents don't collide
//!   because each owns its filename.
//! - **Heartbeat-based liveness.** Each entry carries
//!   `last_heartbeat_at` (UNIX seconds). [`AgentRegistry::reap_stale`]
//!   deletes any entry whose heartbeat is older than the supplied
//!   max-age. Agents heartbeat every 20s; default reap threshold
//!   is 60s so two missed heartbeats trigger removal.
//! - **No daemon.** Every operation is a synchronous file-system
//!   call. Callers (UI surfaces, agent processes) own the polling
//!   cadence.
//!
//! ## Example
//!
//! ```ignore
//! use car_registry::{AgentEntry, AgentRegistry, AgentStatus};
//!
//! let registry = AgentRegistry::user_default()?;
//! registry.register(&AgentEntry::new("trader-paper", "http://127.0.0.1:8731"))?;
//!
//! // Later, on the menubar side:
//! for entry in registry.list()? {
//!     println!("{} → {}", entry.name, entry.dashboard_url);
//! }
//!
//! // Periodic heartbeat from the agent:
//! registry.heartbeat("trader-paper")?;
//!
//! // Periodic stale sweep from the menubar:
//! registry.reap_stale(60)?;
//! ```
//!
//! [#111]: https://github.com/Parslee-ai/car/issues/111

use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};

/// Lifecycle supervisor — declarative manifest at
/// `~/.car/agents.json` driving spawn/restart/stop of long-lived
/// child processes. Sibling concern to the observe-only
/// [`AgentRegistry`] above: registry tracks "what agents have
/// announced themselves," supervisor manages "what agents we should
/// keep running." Closes Parslee-ai/car-releases#27.
pub mod proc;
pub mod supervisor;

/// Declarative, in-daemon agents (no external process) — a parallel registry
/// to [`supervisor`]. See [`declarative::DeclarativeAgentSpec`].
pub mod declarative;

/// Routing learning state — per-agent success stats + directed agent→agent
/// edge weights that let capability-similarity routing learn from outcomes
/// (AgentNet-style self-organization). See [`routing::RoutingStore`].
pub mod routing;

/// `manifest.toml` on-disk format for contributed agents
/// (Parslee-ai/car#182, phase 1). Sits alongside legacy
/// `~/.car/agents.json` during the dual-read migration window;
/// see `docs/proposals/contributed-agents.md`.
pub mod manifest;

/// Install-time validation for contributed-agent manifests
/// (Parslee-ai/car#182 phase 3). `car_min_version` enforcement,
/// capability negotiation against a host advertisement, and
/// per-version side-by-side install resolution. Called by
/// `Supervisor::install_manifest` before adoption.
pub mod install;

#[derive(Debug, thiserror::Error)]
pub enum RegistryError {
    #[error("invalid agent name (must be non-empty, alphanumeric + `-_.`): {0:?}")]
    InvalidName(String),
    #[error("could not resolve home directory")]
    NoHomeDir,
    #[error("registry I/O error: {0}")]
    Io(#[from] std::io::Error),
    #[error("registry JSON error: {0}")]
    Json(#[from] serde_json::Error),
}

/// Lifecycle status an agent advertises. Free-form strings would be
/// fine here, but a small enum keeps the menubar's display logic
/// (badge colour, tooltip) consistent across agents.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum AgentStatus {
    /// Currently doing work.
    Running,
    /// Idle but accepting requests.
    #[default]
    Idle,
    /// Hit an error; dashboard URL may still load but agent isn't
    /// processing.
    Errored,
    /// Shutting down. Menubar may grey it out before the entry is
    /// removed.
    Stopping,
}

/// One agent's entry in the registry. Mirrors the contract from
/// [#111].
///
/// Fields are intentionally minimal — agents that want to expose
/// richer state put it on their own dashboard. The registry is just
/// "what's running and where do I find it."
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AgentEntry {
    /// Stable identifier. Used as the filename
    /// (`<name>.json`) so duplicate names overwrite — by design,
    /// since two processes claiming the same name is a config bug.
    pub name: String,
    /// URL where the agent's UI lives. Menubar renders a clickable
    /// link.
    pub dashboard_url: String,
    /// Lifecycle status. See [`AgentStatus`].
    #[serde(default)]
    pub status: AgentStatus,
    /// Optional human-friendly label distinct from `name`. e.g.
    /// `name = "trader-paper"`, `display_name = "Trader (paper)"`.
    #[serde(default)]
    pub display_name: Option<String>,
    /// Optional natural-language description of what this service does
    /// ("checks whether a flight trip is feasible for the fleet"). Used by
    /// `discovery.resolve` to rank the service against a natural-language need
    /// — without it the service still resolves but ranks on its label alone.
    /// Distinct from `display_name`, which is a short UI label.
    #[serde(default)]
    pub capability: Option<String>,
    /// TCP port the dashboard listens on, when applicable.
    #[serde(default)]
    pub port: Option<u16>,
    /// PID for the menubar's "is this still alive" liveness check
    /// when callers want a stronger signal than heartbeat alone.
    #[serde(default)]
    pub pid: Option<u32>,
    /// UNIX seconds when the entry was first registered.
    #[serde(default)]
    pub registered_at: u64,
    /// UNIX seconds of the last heartbeat. The reap sweep deletes
    /// entries older than the configured threshold.
    #[serde(default)]
    pub last_heartbeat_at: u64,
}

impl AgentEntry {
    pub fn new(name: impl Into<String>, dashboard_url: impl Into<String>) -> Self {
        let now = now_secs();
        Self {
            name: name.into(),
            dashboard_url: dashboard_url.into(),
            status: AgentStatus::default(),
            display_name: None,
            capability: None,
            port: None,
            pid: None,
            registered_at: now,
            last_heartbeat_at: now,
        }
    }

    pub fn with_display_name(mut self, label: impl Into<String>) -> Self {
        self.display_name = Some(label.into());
        self
    }

    pub fn with_capability(mut self, capability: impl Into<String>) -> Self {
        self.capability = Some(capability.into());
        self
    }

    pub fn with_port(mut self, port: u16) -> Self {
        self.port = Some(port);
        self
    }

    pub fn with_pid(mut self, pid: u32) -> Self {
        self.pid = Some(pid);
        self
    }

    pub fn with_status(mut self, status: AgentStatus) -> Self {
        self.status = status;
        self
    }
}

/// File-system handle to the registry directory.
///
/// Cheap to clone; the only state is the directory path.
#[derive(Debug, Clone)]
pub struct AgentRegistry {
    dir: PathBuf,
}

impl AgentRegistry {
    /// Use `~/.car/registry/`. Creates the directory if missing.
    /// Errors if `$HOME` (or `%USERPROFILE%`) can't be resolved.
    pub fn user_default() -> Result<Self, RegistryError> {
        let home = std::env::var_os("HOME")
            .or_else(|| std::env::var_os("USERPROFILE"))
            .ok_or(RegistryError::NoHomeDir)?;
        let dir = PathBuf::from(home).join(".car").join("registry");
        Self::open(dir)
    }

    /// Use a specific directory. Tests pass a `tempfile::TempDir`
    /// here; production callers either use [`Self::user_default`]
    /// or honour an embedder-supplied path.
    pub fn open(dir: impl Into<PathBuf>) -> Result<Self, RegistryError> {
        let dir = dir.into();
        std::fs::create_dir_all(&dir)?;
        Ok(Self { dir })
    }

    /// Directory the registry writes to.
    pub fn dir(&self) -> &Path {
        &self.dir
    }

    /// Write or replace this agent's entry. Atomic — writes to a
    /// temp file then renames so concurrent readers never see a
    /// half-written JSON.
    pub fn register(&self, entry: &AgentEntry) -> Result<(), RegistryError> {
        let path = self.entry_path(&entry.name)?;
        let mut entry = entry.clone();
        if entry.registered_at == 0 {
            entry.registered_at = now_secs();
        }
        if entry.last_heartbeat_at == 0 {
            entry.last_heartbeat_at = entry.registered_at;
        }
        write_json_atomic(&path, &entry)?;
        Ok(())
    }

    /// Bump `last_heartbeat_at` to the current time. Cheap — reads
    /// the existing entry, updates the timestamp, writes back.
    /// Returns `Ok(false)` if the named agent isn't currently
    /// registered (caller can decide whether to re-register).
    pub fn heartbeat(&self, name: &str) -> Result<bool, RegistryError> {
        let path = self.entry_path(name)?;
        if !path.exists() {
            return Ok(false);
        }
        let bytes = std::fs::read(&path)?;
        let mut entry: AgentEntry = serde_json::from_slice(&bytes)?;
        entry.last_heartbeat_at = now_secs();
        write_json_atomic(&path, &entry)?;
        Ok(true)
    }

    /// Remove the named agent's entry. No-op if it doesn't exist.
    pub fn unregister(&self, name: &str) -> Result<(), RegistryError> {
        let path = self.entry_path(name)?;
        if path.exists() {
            std::fs::remove_file(&path)?;
        }
        Ok(())
    }

    /// All currently-registered agents. Reads the whole directory
    /// each call — fine for small N (the realistic ceiling is dozens
    /// of agents per host). Files that fail to deserialise are
    /// silently skipped to avoid one corrupt entry blocking
    /// discovery of every other agent.
    pub fn list(&self) -> Result<Vec<AgentEntry>, RegistryError> {
        let mut out = Vec::new();
        for entry in std::fs::read_dir(&self.dir)? {
            let entry = entry?;
            let path = entry.path();
            if path.extension().and_then(|s| s.to_str()) != Some("json") {
                continue;
            }
            if let Ok(bytes) = std::fs::read(&path) {
                if let Ok(parsed) = serde_json::from_slice::<AgentEntry>(&bytes) {
                    out.push(parsed);
                }
            }
        }
        out.sort_by(|a, b| a.name.cmp(&b.name));
        Ok(out)
    }

    /// Delete entries whose `last_heartbeat_at` is older than
    /// `max_age_secs`. Returns the names of entries **this call actually
    /// removed**, so callers can log / notify.
    ///
    /// Recommended cadence: call from the menubar process every
    /// ~30s with `max_age_secs = 60` (catches an agent that missed
    /// two consecutive 20s heartbeats).
    ///
    /// Two properties matter once a *scheduled* caller drives this (a
    /// dormant primitive tolerates sloppiness a timer does not):
    ///
    /// - **The staleness verdict is re-checked against the entry on disk
    ///   immediately before the unlink.** `list()` is a snapshot; an agent
    ///   that re-registered between the snapshot and the delete has a fresh
    ///   heartbeat and is kept. This is load-bearing because [`heartbeat`]
    ///   returns `Ok(false)` on a missing file and does **not** recreate it,
    ///   so a wrongly-reaped entry does not self-heal on the next beat —
    ///   avoiding the wrong reap is the only reliable guard. The residual
    ///   read→unlink window is microseconds and, against a `max_age_secs`
    ///   far above any live heartbeat interval, effectively closed.
    /// - **Only files actually unlinked are reported.** A file that can't be
    ///   deleted (e.g. a permission error) is never reported as reaped — else
    ///   a scheduled caller would "reap" the same undeletable file every
    ///   cycle forever. `NotFound` (another sweeper won the race) is the
    ///   idempotent success outcome and is skipped silently.
    pub fn reap_stale(&self, max_age_secs: u64) -> Result<Vec<String>, RegistryError> {
        let cutoff = now_secs().saturating_sub(max_age_secs);
        let entries = self.list()?;
        let mut reaped = Vec::new();
        for entry in entries {
            if entry.last_heartbeat_at >= cutoff {
                continue;
            }
            let Ok(path) = self.entry_path(&entry.name) else {
                continue;
            };
            // Re-read under the delete decision — the snapshot may be stale.
            match std::fs::read(&path) {
                Ok(bytes) => match serde_json::from_slice::<AgentEntry>(&bytes) {
                    // Refreshed since the snapshot — a live agent beat again. Keep it.
                    Ok(current) if current.last_heartbeat_at >= cutoff => continue,
                    // Still stale, or now-unparseable junk — reap.
                    Ok(_) | Err(_) => {}
                },
                // Vanished or unreadable since the snapshot — nothing to reap.
                Err(_) => continue,
            }
            match std::fs::remove_file(&path) {
                Ok(()) => reaped.push(entry.name),
                // Another sweeper won the race — idempotent, not our reap.
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                // Real failure (perms, etc.): surface it, but do NOT claim we
                // removed it — a false report would recur every cycle.
                Err(e) => tracing::warn!(
                    agent = %entry.name,
                    path = %path.display(),
                    error = %e,
                    "car-registry: failed to reap stale entry"
                ),
            }
        }
        Ok(reaped)
    }

    fn entry_path(&self, name: &str) -> Result<PathBuf, RegistryError> {
        validate_name(name)?;
        Ok(self.dir.join(format!("{}.json", name)))
    }
}

fn validate_name(name: &str) -> Result<(), RegistryError> {
    if name.is_empty() {
        return Err(RegistryError::InvalidName(name.to_string()));
    }
    // Filename safety: allow alphanumeric, dash, underscore, dot.
    // Rejects path traversal (`..`, `/`, `\`) and weirdness that
    // would break on Windows.
    if !name
        .chars()
        .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.')
    {
        return Err(RegistryError::InvalidName(name.to_string()));
    }
    if name == "." || name == ".." {
        return Err(RegistryError::InvalidName(name.to_string()));
    }
    Ok(())
}

fn now_secs() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// Atomically replace `path` with the JSON-serialised value. Writes
/// to a sibling temp file first, then `rename`s — POSIX guarantees
/// rename is atomic on the same filesystem, so concurrent readers
/// never see a half-written entry.
fn write_json_atomic<T: Serialize>(path: &Path, value: &T) -> Result<(), RegistryError> {
    let parent = path.parent().ok_or_else(|| {
        std::io::Error::new(
            std::io::ErrorKind::InvalidInput,
            "registry path has no parent",
        )
    })?;
    let tmp = parent.join(format!(
        ".{}.tmp",
        path.file_name()
            .and_then(|s| s.to_str())
            .unwrap_or("registry-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_registry() -> (tempfile::TempDir, AgentRegistry) {
        let tmp = tempfile::TempDir::new().unwrap();
        let reg = AgentRegistry::open(tmp.path()).unwrap();
        (tmp, reg)
    }

    #[test]
    fn register_then_list_round_trips() {
        let (_tmp, reg) = temp_registry();
        reg.register(
            &AgentEntry::new("trader-paper", "http://127.0.0.1:8731")
                .with_display_name("Trader (paper)")
                .with_port(8731)
                .with_status(AgentStatus::Running),
        )
        .unwrap();
        let listed = reg.list().unwrap();
        assert_eq!(listed.len(), 1);
        assert_eq!(listed[0].name, "trader-paper");
        assert_eq!(listed[0].port, Some(8731));
        assert_eq!(listed[0].status, AgentStatus::Running);
        assert!(listed[0].registered_at > 0);
    }

    #[test]
    fn capability_round_trips() {
        let (_tmp, reg) = temp_registry();
        reg.register(
            &AgentEntry::new("fms-feasibility", "http://127.0.0.1:8132")
                .with_capability("checks whether a flight trip is feasible for the fleet"),
        )
        .unwrap();
        let listed = reg.list().unwrap();
        assert_eq!(listed.len(), 1);
        assert_eq!(
            listed[0].capability.as_deref(),
            Some("checks whether a flight trip is feasible for the fleet")
        );
    }

    #[test]
    fn legacy_entry_without_capability_deserializes() {
        // Entries written before the `capability` field must still load — the
        // `~/.car/registry/*.json` files on disk predate it.
        let (tmp, reg) = temp_registry();
        std::fs::write(
            tmp.path().join("legacy.json"),
            br#"{"name":"legacy","dashboard_url":"http://x","status":"running"}"#,
        )
        .unwrap();
        let listed = reg.list().unwrap();
        assert_eq!(listed.len(), 1);
        assert_eq!(listed[0].name, "legacy");
        assert_eq!(listed[0].capability, None);
    }

    #[test]
    fn heartbeat_bumps_timestamp() {
        let (_tmp, reg) = temp_registry();
        reg.register(&AgentEntry::new("a", "http://x")).unwrap();
        let before = reg.list().unwrap()[0].last_heartbeat_at;
        // Sleep one second to guarantee the second-resolution bump
        // is observable in the assertion.
        std::thread::sleep(std::time::Duration::from_secs(1));
        let touched = reg.heartbeat("a").unwrap();
        assert!(touched);
        let after = reg.list().unwrap()[0].last_heartbeat_at;
        assert!(after >= before);
    }

    #[test]
    fn heartbeat_unknown_returns_false() {
        let (_tmp, reg) = temp_registry();
        assert!(!reg.heartbeat("nobody-home").unwrap());
    }

    #[test]
    fn unregister_idempotent() {
        let (_tmp, reg) = temp_registry();
        reg.register(&AgentEntry::new("x", "http://x")).unwrap();
        reg.unregister("x").unwrap();
        assert!(reg.list().unwrap().is_empty());
        // Second unregister on an already-removed entry is fine.
        reg.unregister("x").unwrap();
    }

    #[test]
    fn reap_stale_removes_old_entries() {
        let (_tmp, reg) = temp_registry();
        // Register with an ancient heartbeat directly (bypasses
        // `register`'s zero-normalisation, since `register` treats
        // `0` as "use current time"). Use 1-second-past-epoch to
        // simulate a long-dead agent.
        let mut old = AgentEntry::new("zombie", "http://z");
        old.last_heartbeat_at = 1; // ancient, but non-zero
        reg.register(&old).unwrap();
        let fresh = AgentEntry::new("alive", "http://a");
        reg.register(&fresh).unwrap();

        let reaped = reg.reap_stale(60).unwrap();
        assert_eq!(reaped, vec!["zombie"]);
        let remaining = reg.list().unwrap();
        assert_eq!(remaining.len(), 1);
        assert_eq!(remaining[0].name, "alive");
    }

    #[test]
    fn reap_stale_reports_only_actually_removed_and_is_idempotent() {
        // Guards the honest-reporting fix: the returned vec must be exactly the
        // set of entries this call unlinked — never a name it failed to (or
        // didn't) remove. A scheduled caller relies on this so it can't "reap"
        // the same file every cycle forever.
        let (_tmp, reg) = temp_registry();
        for n in ["z1", "z2"] {
            let mut e = AgentEntry::new(n, "http://z");
            e.last_heartbeat_at = 1; // ancient
            reg.register(&e).unwrap();
        }
        reg.register(&AgentEntry::new("alive", "http://a")).unwrap();

        let mut reaped = reg.reap_stale(60).unwrap();
        reaped.sort();
        assert_eq!(reaped, vec!["z1".to_string(), "z2".to_string()]);

        // Every reported name is truly gone; the fresh one that was never
        // reported survives.
        let remaining: Vec<String> = reg.list().unwrap().into_iter().map(|e| e.name).collect();
        for name in &reaped {
            assert!(
                !remaining.contains(name),
                "reported reaped but still present: {name}"
            );
        }
        assert_eq!(remaining, vec!["alive".to_string()]);

        // A second sweep removes nothing and — critically — reports nothing,
        // rather than re-claiming the already-gone entries.
        assert!(reg.reap_stale(60).unwrap().is_empty());
    }

    #[test]
    fn invalid_names_rejected() {
        let (_tmp, reg) = temp_registry();
        assert!(reg.register(&AgentEntry::new("..", "http://x")).is_err());
        assert!(reg
            .register(&AgentEntry::new("evil/name", "http://x"))
            .is_err());
        assert!(reg
            .register(&AgentEntry::new("space name", "http://x"))
            .is_err());
        assert!(reg.register(&AgentEntry::new("", "http://x")).is_err());
    }

    #[test]
    fn corrupt_entries_skipped_in_list() {
        let (tmp, reg) = temp_registry();
        reg.register(&AgentEntry::new("ok", "http://o")).unwrap();
        std::fs::write(tmp.path().join("garbage.json"), b"not json").unwrap();
        let entries = reg.list().unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].name, "ok");
    }
}