car_server_core/

run_store.rs

//! Disk-backed run-trace store (agent run tracing, U3).
//!
//! Persists the per-run trace stream — the `RunStarted` line, each
//! `RunTurn`, then the terminal `RunEnded`/`Incomplete` — as JSONL so a
//! run survives daemon restarts and stays grouped by run independent of
//! the WS connection that produced it (R4). One file per run:
//!
//! ```text
//! ~/.car/runs/{agent_id}/{run_id}.jsonl
//! ```
//!
//! ## Source-of-truth split
//!
//! This disk store is the source of truth for **replay** (U5):
//! [`RunStore::get_run_trace`] / [`RunStore::list_runs`] read it directly
//! and work after a restart when memory is empty. The in-memory
//! `RunMeta.turns` buffer (U2) stays the source for the **live** stream
//! (U4) — U3 does not remove it; it mirrors what was recorded onto disk.
//!
//! ## Layout, perms, backup exclusion (R14)
//!
//! Prompts and CLI output may carry secrets, so the `runs/` tree is
//! created `0700` and every file `0600` (Unix). The `runs/` dir is
//! marked backup-excluded — a `.nobackup` marker file plus, on macOS,
//! the `com.apple.metadata:com_apple_backup_excludeItem` xattr — so Time
//! Machine / iCloud don't silently copy plaintext traces off the box.
//!
//! ## Index
//!
//! There is no separate index file to drift: the directory tree *is* the
//! index. `list_runs(agent_id)` scans `runs/{agent_id}/`; `run_id ->
//! agent_id` resolves by scanning `runs/*/` for the matching
//! `{run_id}.jsonl` (U5's `runs.get_trace` takes only a `run_id`). Each
//! file's first line is its `RunStarted`, so a run's `started_at` /
//! `intent` / status come from reading the file head + tail, not a
//! sidecar.
//!
//! ## Retention (R6)
//!
//! GC runs on daemon boot ([`RunStore::gc`]): per agent it keeps the **50
//! most recent completed runs** and drops anything older than **30 days**,
//! whichever is more restrictive. A still-in-progress run (no terminal
//! record) is **never** evicted. Both limits are configurable via
//! `~/.car/config.toml` (`[runs] max_per_agent` / `max_age_days`) with
//! that restrictive default.
//!
//! Records are appended at turn granularity (not per token) — the same
//! coarse boundary the in-memory buffer uses. A corrupt/partial trailing
//! JSONL line loads the prior valid records rather than failing the whole
//! run (the error-path test).

use car_proto::{RunRecord, RunTermination};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::io::{BufRead, Write};
use std::path::{Path, PathBuf};

/// Default per-agent cap: keep the 50 most recent completed runs.
pub const DEFAULT_MAX_RUNS_PER_AGENT: usize = 50;
/// Default age cap: drop completed runs older than 30 days.
pub const DEFAULT_MAX_AGE_DAYS: i64 = 30;

/// Terminal/in-progress status of a run, derived from its records.
///
/// Distinct from the run-level `OutcomeStatus` carried inside a terminal
/// `Outcome` — this is the coarse "what state is this run in?" the run
/// list renders. `Incomplete` is the orphan case (no terminal record + no
/// live harness, R5); `InProgress` is a run still being written.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RunStatus {
    /// No terminal record yet — still being written by a live harness.
    InProgress,
    /// `runs.complete` reported a terminal `AgentOutcome`. The concrete
    /// `OutcomeStatus` lives on the `RunEnded` record itself.
    Completed,
    /// The harness disconnected without reporting an outcome (R5).
    Incomplete,
}

/// One row of [`RunStore::list_runs`] — the summary U5's `runs.list`
/// returns. Cheap to build (head + tail of the JSONL file), so listing an
/// agent's runs doesn't load every turn.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RunSummary {
    pub run_id: String,
    pub agent_id: String,
    pub intent: String,
    pub started_at: DateTime<Utc>,
    /// When the terminal record was written, if any.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ended_at: Option<DateTime<Utc>>,
    pub status: RunStatus,
    /// Number of `RunTurn` records persisted for this run.
    pub turn_count: usize,
}

/// Retention policy for the run store (R6). Defaults are the restrictive
/// 50-per-agent / 30-day caps; `~/.car/config.toml [runs]` overrides.
#[derive(Debug, Clone, Copy)]
pub struct RetentionConfig {
    pub max_per_agent: usize,
    pub max_age_days: i64,
}

impl Default for RetentionConfig {
    fn default() -> Self {
        Self {
            max_per_agent: DEFAULT_MAX_RUNS_PER_AGENT,
            max_age_days: DEFAULT_MAX_AGE_DAYS,
        }
    }
}

/// `[runs]` section of `~/.car/config.toml`. Both keys optional; an
/// absent key keeps the restrictive default.
#[derive(Debug, Clone, Default, Deserialize)]
struct RunsConfigFile {
    #[serde(default)]
    runs: RunsSection,
}

#[derive(Debug, Clone, Default, Deserialize)]
struct RunsSection {
    #[serde(default)]
    max_per_agent: Option<usize>,
    #[serde(default)]
    max_age_days: Option<i64>,
}

impl RetentionConfig {
    /// Load the retention policy from `<car_dir>/config.toml`'s `[runs]`
    /// section, falling back to the restrictive default for any missing
    /// key or an unreadable/malformed file (config errors must never make
    /// the daemon refuse to start — same posture as the rest of `.car`).
    pub fn from_car_dir(car_dir: &Path) -> Self {
        let mut cfg = Self::default();
        let path = car_dir.join("config.toml");
        let Ok(text) = std::fs::read_to_string(&path) else {
            return cfg;
        };
        let Ok(parsed) = toml::from_str::<RunsConfigFile>(&text) else {
            return cfg;
        };
        if let Some(n) = parsed.runs.max_per_agent {
            cfg.max_per_agent = n;
        }
        if let Some(d) = parsed.runs.max_age_days {
            cfg.max_age_days = d;
        }
        cfg
    }
}

/// JSONL run-trace store rooted at `<car_dir>/runs/`.
///
/// Stateless across calls — each append opens, writes, and closes the
/// run's file. Flush points are sparse (turn granularity), so there is no
/// long-lived file handle to manage, and a concurrently-restarting daemon
/// always sees a consistent on-disk tail.
#[derive(Debug, Clone)]
pub struct RunStore {
    /// `<car_dir>/runs` — the tree root, created `0700`.
    root: PathBuf,
    retention: RetentionConfig,
}

impl RunStore {
    /// Construct a store rooted at `runs_root` (the `runs/` dir itself).
    /// Use [`RunStore::from_journal_dir`] from the daemon, which derives
    /// the root from the configured journal dir; this constructor is the
    /// test/embedder seam.
    pub fn new(runs_root: PathBuf, retention: RetentionConfig) -> Self {
        Self {
            root: runs_root,
            retention,
        }
    }

    /// Derive the store from the daemon's journal dir. The journal lives
    /// at `~/.car/journals`, so the run store is its sibling
    /// `~/.car/runs`; retention is read from `~/.car/config.toml`. When
    /// the journal dir has no parent (a bare relative path), the store
    /// falls back to `journal_dir/../runs` resolved lexically.
    pub fn from_journal_dir(journal_dir: &Path) -> Self {
        let car_dir = journal_dir
            .parent()
            .map(Path::to_path_buf)
            .unwrap_or_else(|| PathBuf::from("."));
        let root = car_dir.join("runs");
        let retention = RetentionConfig::from_car_dir(&car_dir);
        Self::new(root, retention)
    }

    /// The `runs/` tree root.
    pub fn root(&self) -> &Path {
        &self.root
    }

    /// Path to a run's JSONL file: `runs/{agent_id}/{run_id}.jsonl`.
    fn run_path(&self, agent_id: &str, run_id: &str) -> PathBuf {
        self.root
            .join(sanitize(agent_id))
            .join(format!("{}.jsonl", sanitize(run_id)))
    }

    /// Ensure the `runs/` root exists with `0700` perms and is marked
    /// backup-excluded. Idempotent. Called lazily on the first append so
    /// constructing a `RunStore` is free (no disk touch until a run is
    /// actually recorded).
    fn ensure_root(&self) -> std::io::Result<()> {
        let created = !self.root.exists();
        std::fs::create_dir_all(&self.root)?;
        set_dir_perms(&self.root)?;
        if created {
            mark_backup_excluded(&self.root);
        }
        Ok(())
    }

    /// Ensure an agent's run dir exists `0700`.
    fn ensure_agent_dir(&self, agent_id: &str) -> std::io::Result<PathBuf> {
        self.ensure_root()?;
        let dir = self.root.join(sanitize(agent_id));
        std::fs::create_dir_all(&dir)?;
        set_dir_perms(&dir)?;
        Ok(dir)
    }

    /// Append one or more `RunRecord`s to a run's file, creating it `0600`
    /// on first write. Records are written one JSONL line each, in order.
    ///
    /// This is the single low-level flush primitive the wiring calls at
    /// each boundary: `RunStarted` on `runs.start`, `RunTurn`s as the
    /// recorder produces them, and the terminal `RunEnded`/`Incomplete` on
    /// `runs.complete`/disconnect.
    pub fn append_records(
        &self,
        agent_id: &str,
        run_id: &str,
        records: &[RunRecord],
    ) -> std::io::Result<()> {
        if records.is_empty() {
            return Ok(());
        }
        self.ensure_agent_dir(agent_id)?;
        let path = self.run_path(agent_id, run_id);
        let existed = path.exists();
        let mut file = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&path)?;
        // FIX 6/7: build the WHOLE batch into one buffer and write it with a
        // single `write_all`. With `O_APPEND` a single write is positioned
        // and appended atomically relative to other appenders, so two
        // concurrent batches for the same file can't interleave mid-record
        // (FIX 6, defense-in-depth). A leading '\n' repairs a torn tail
        // (FIX 7) as part of the same atomic write.
        let mut buf: Vec<u8> = Vec::new();
        if !existed {
            // New file: tighten to 0600 before any secret-bearing line
            // lands. Done after create so the mode isn't masked by umask.
            set_file_perms(&path)?;
        } else if last_byte_is_not_newline(&path)? {
            // Torn tail (FIX 7): a prior append was cut short (ENOSPC/EINTR
            // mid-line, or a crash) leaving a record with no trailing '\n'.
            // Appending here would concatenate our valid record onto the
            // torn one, so `load_records` would drop OUR good record too,
            // not just the trailing garbage. Prepend a '\n' so the torn
            // fragment becomes its own (skippable) line and our record lands
            // intact on the next line.
            buf.push(b'\n');
        }
        for rec in records {
            let line = serde_json::to_string(rec)
                .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
            buf.extend_from_slice(line.as_bytes());
            buf.push(b'\n');
        }
        file.write_all(&buf)?;
        Ok(())
    }

    /// Append the `RunStarted` line + create the run file (`runs.start`).
    pub fn write_started(&self, started: &car_proto::RunStarted) -> std::io::Result<()> {
        let rec = RunRecord::Started(started.clone());
        self.append_records(&started.agent_id, &started.run_id, &[rec])
    }

    /// Append `RunTurn` records (the recorder's per-proposal output).
    pub fn append_turns(
        &self,
        agent_id: &str,
        run_id: &str,
        turns: &[RunRecord],
    ) -> std::io::Result<()> {
        self.append_records(agent_id, run_id, turns)
    }

    /// Append the terminal `RunEnded` line (`runs.complete` or the
    /// disconnect-`Incomplete` path).
    pub fn write_ended(&self, ended: &car_proto::RunEnded) -> std::io::Result<()> {
        let rec = RunRecord::Ended(ended.clone());
        self.append_records(&ended.agent_id, &ended.run_id, &[rec])
    }

    /// Load a run's full ordered trace from disk by `run_id` — the U5
    /// `runs.get_trace` read path. Works after a restart when memory is
    /// empty. Resolves `run_id -> agent_id` by scanning the tree, then
    /// reads the JSONL. A corrupt/partial trailing line is skipped, so the
    /// prior valid records still load (never fails the whole run).
    /// Returns `None` when no file exists for the `run_id`.
    pub fn get_run_trace(&self, run_id: &str) -> Option<Vec<RunRecord>> {
        let path = self.resolve_run_path(run_id)?;
        Some(load_records(&path))
    }

    /// Load a run's trace given both keys (cheaper — no tree scan). Used
    /// by `list_runs` internally and available to callers that already
    /// know the owning agent.
    pub fn get_run_trace_for(&self, agent_id: &str, run_id: &str) -> Option<Vec<RunRecord>> {
        let path = self.run_path(agent_id, run_id);
        if path.exists() {
            Some(load_records(&path))
        } else {
            None
        }
    }

    /// List an agent's runs newest-first — the U5 `runs.list` read path.
    /// Each summary is built from the run file's records (head for
    /// `RunStarted`, tail for the terminal record, count of `Turn`s).
    /// Returns an empty Vec for an agent with no runs (the empty-state).
    pub fn list_runs(&self, agent_id: &str) -> Vec<RunSummary> {
        let dir = self.root.join(sanitize(agent_id));
        let mut out = Vec::new();
        let Ok(entries) = std::fs::read_dir(&dir) else {
            return out;
        };
        for entry in entries.flatten() {
            let path = entry.path();
            if path.extension().and_then(|e| e.to_str()) != Some("jsonl") {
                continue;
            }
            if let Some(summary) = summarize_file(&path) {
                out.push(summary);
            }
        }
        // Newest first by start time.
        out.sort_by(|a, b| b.started_at.cmp(&a.started_at));
        out
    }

    /// Resolve `run_id -> path` by scanning `runs/*/{run_id}.jsonl`. U5's
    /// `runs.get_trace` takes only a `run_id`, so the owning agent must be
    /// discovered. Returns the first match (run ids are uuids — unique).
    fn resolve_run_path(&self, run_id: &str) -> Option<PathBuf> {
        let file_name = format!("{}.jsonl", sanitize(run_id));
        let agent_dirs = std::fs::read_dir(&self.root).ok()?;
        for agent in agent_dirs.flatten() {
            let candidate = agent.path().join(&file_name);
            if candidate.is_file() {
                return Some(candidate);
            }
        }
        None
    }

    /// Resolve the owning `agent_id` for a `run_id` from disk. Mirrors
    /// [`Self::resolve_run_path`] but returns the agent dir name — U5's
    /// authorization check (KTD10) needs `run_id -> agent_id` to verify
    /// ownership before serving a trace.
    pub fn agent_for_run(&self, run_id: &str) -> Option<String> {
        let path = self.resolve_run_path(run_id)?;
        path.parent()
            .and_then(Path::file_name)
            .and_then(|s| s.to_str())
            .map(str::to_string)
    }

    /// Retention GC (R6) — call on daemon boot. Per agent: keep the most
    /// recent `max_per_agent` **completed** runs and drop any completed
    /// run older than `max_age_days`, whichever is more restrictive. An
    /// in-progress run (no terminal record) is NEVER evicted. Returns the
    /// number of run files removed.
    pub fn gc(&self) -> usize {
        let mut removed = 0;
        let Ok(agent_dirs) = std::fs::read_dir(&self.root) else {
            return 0;
        };
        let cutoff = Utc::now() - chrono::Duration::days(self.retention.max_age_days);
        for agent in agent_dirs.flatten() {
            let agent_path = agent.path();
            if !agent_path.is_dir() {
                continue;
            }
            removed += self.gc_agent_dir(&agent_path, cutoff);
        }
        removed
    }

    /// Adopt crash-orphaned runs at boot (FIX 4). A daemon crash mid-run
    /// leaves an on-disk run with `RunStarted` (+ `Turn`s) but no terminal
    /// `RunEnded`, so it reads `InProgress` forever and the GC — which never
    /// evicts an in-progress run — can never reclaim it. The file leaks
    /// across every crash.
    ///
    /// This runs at store construction/boot, BEFORE `gc()`. At that moment
    /// the in-memory `runs` map is always empty, so any run that is
    /// `InProgress` on disk cannot have a live harness writing to it — it is
    /// necessarily a crashed prior process. We append an `Incomplete`
    /// terminal `RunEnded` marker to adopt it, making it terminal and thus
    /// age-GC-eligible (so a later `gc()` in the same boot can reclaim it).
    ///
    /// Returns the number of runs adopted. Best-effort: an unwritable file is
    /// skipped rather than failing startup.
    pub fn adopt_orphans(&self) -> usize {
        let mut adopted = 0;
        let Ok(agent_dirs) = std::fs::read_dir(&self.root) else {
            return 0;
        };
        let now = Utc::now();
        for agent in agent_dirs.flatten() {
            let agent_path = agent.path();
            if !agent_path.is_dir() {
                continue;
            }
            let Ok(entries) = std::fs::read_dir(&agent_path) else {
                continue;
            };
            for entry in entries.flatten() {
                let path = entry.path();
                if path.extension().and_then(|e| e.to_str()) != Some("jsonl") {
                    continue;
                }
                let Some(summary) = summarize_file(&path) else {
                    continue;
                };
                if summary.status != RunStatus::InProgress {
                    continue;
                }
                // No terminal record + memory empty at boot => crash orphan.
                let incomplete = RunRecord::Ended(car_proto::RunEnded {
                    run_id: summary.run_id.clone(),
                    agent_id: summary.agent_id.clone(),
                    termination: RunTermination::Incomplete,
                    ended_at: now,
                });
                if self
                    .append_records(&summary.agent_id, &summary.run_id, &[incomplete])
                    .is_ok()
                {
                    adopted += 1;
                }
            }
        }
        adopted
    }

    /// GC one agent's run dir against the retention caps.
    fn gc_agent_dir(&self, agent_path: &Path, age_cutoff: DateTime<Utc>) -> usize {
        // Collect (path, summary) for every run file, ignoring unreadable
        // ones (a malformed file with no RunStarted can't be summarized;
        // leave it rather than risk evicting something we can't classify).
        let mut runs: Vec<(PathBuf, RunSummary)> = Vec::new();
        let Ok(entries) = std::fs::read_dir(agent_path) else {
            return 0;
        };
        for entry in entries.flatten() {
            let path = entry.path();
            if path.extension().and_then(|e| e.to_str()) != Some("jsonl") {
                continue;
            }
            if let Some(s) = summarize_file(&path) {
                runs.push((path, s));
            }
        }
        // Newest-first so the most recent completed runs rank first.
        runs.sort_by(|a, b| b.1.started_at.cmp(&a.1.started_at));

        let mut removed = 0;
        // Rank among COMPLETED/Incomplete runs only — an in-progress run
        // must not consume a keeper slot, so the count cap is measured by
        // completed-run rank, not the combined sorted index (R6).
        let mut completed_rank = 0usize;
        for (path, summary) in runs.iter() {
            // NEVER evict an in-progress run — it has no terminal record
            // and a live harness may still be writing to it (R6).
            if summary.status == RunStatus::InProgress {
                continue;
            }
            let over_count = completed_rank >= self.retention.max_per_agent;
            completed_rank += 1;
            // Age the run by its TERMINAL time, not its start. A long run
            // that started >max_age_days ago but completed recently is still
            // a recent result and must not be evicted (FIX 2). Completed and
            // Incomplete runs always have an `ended_at`; fall back to
            // `started_at` only if a terminal record somehow lacks one.
            let term_time = summary.ended_at.unwrap_or(summary.started_at);
            let too_old = term_time < age_cutoff;
            if over_count || too_old {
                if std::fs::remove_file(path).is_ok() {
                    removed += 1;
                }
            }
        }
        removed
    }
}

/// Return `true` when the file's last byte is not `\n` — i.e. the previous
/// append was torn (cut short mid-line by ENOSPC/EINTR/crash) and a fresh
/// append must insert a separating newline first (FIX 7). An empty or
/// nonexistent file returns `false` (nothing to repair). Reads only the
/// final byte via a seek, so it's cheap regardless of file size.
fn last_byte_is_not_newline(path: &Path) -> std::io::Result<bool> {
    use std::io::{Read, Seek, SeekFrom};
    let mut f = std::fs::File::open(path)?;
    let len = f.seek(SeekFrom::End(0))?;
    if len == 0 {
        return Ok(false);
    }
    f.seek(SeekFrom::End(-1))?;
    let mut buf = [0u8; 1];
    f.read_exact(&mut buf)?;
    Ok(buf[0] != b'\n')
}

/// Load every parseable `RunRecord` from a JSONL file, in file order.
/// Blank lines and unparseable lines (a corrupt/partial trailing line) are
/// skipped — the prior valid records still load (error-path).
fn load_records(path: &Path) -> Vec<RunRecord> {
    let Ok(file) = std::fs::File::open(path) else {
        return Vec::new();
    };
    let reader = std::io::BufReader::new(file);
    let mut out = Vec::new();
    for line in reader.lines() {
        let Ok(line) = line else { break };
        if line.trim().is_empty() {
            continue;
        }
        if let Ok(rec) = serde_json::from_str::<RunRecord>(&line) {
            out.push(rec);
        }
        // else: skip the bad line, keep going (handles a partial trailing
        // write left by a crash mid-append).
    }
    out
}

/// Build a [`RunSummary`] from a run's JSONL file: `RunStarted` from the
/// first valid record, terminal status from the last, and `turn_count`
/// from the `Turn`s in between. Returns `None` when the file has no
/// `RunStarted` (it can't be keyed/summarized).
fn summarize_file(path: &Path) -> Option<RunSummary> {
    let records = load_records(path);
    let mut started: Option<car_proto::RunStarted> = None;
    let mut ended: Option<car_proto::RunEnded> = None;
    let mut turn_count = 0usize;
    for rec in &records {
        match rec {
            RunRecord::Started(s) => started = Some(s.clone()),
            RunRecord::Ended(e) => ended = Some(e.clone()),
            RunRecord::Turn(_) => turn_count += 1,
        }
    }
    let started = started?;
    let (status, ended_at) = match &ended {
        Some(e) => {
            let status = match &e.termination {
                RunTermination::Outcome { .. } => RunStatus::Completed,
                RunTermination::Incomplete => RunStatus::Incomplete,
            };
            (status, Some(e.ended_at))
        }
        None => (RunStatus::InProgress, None),
    };
    Some(RunSummary {
        run_id: started.run_id,
        agent_id: started.agent_id,
        intent: started.intent,
        started_at: started.started_at,
        ended_at,
        status,
        turn_count,
    })
}

/// Sanitize an id for use as a path segment — strip any path separators
/// and `..` so a hostile `agent_id`/`run_id` can't escape the `runs/`
/// tree. Ids are uuids / slugs in practice; this is defense in depth.
fn sanitize(id: &str) -> String {
    let cleaned: String = id
        .chars()
        .map(|c| match c {
            '/' | '\\' | '\0' => '_',
            c => c,
        })
        .collect();
    let trimmed = cleaned.trim_matches('.');
    if trimmed.is_empty() {
        "_".to_string()
    } else {
        trimmed.to_string()
    }
}

/// Set a directory to `0700` (owner-only). No-op on non-Unix.
#[cfg(unix)]
fn set_dir_perms(path: &Path) -> std::io::Result<()> {
    use std::os::unix::fs::PermissionsExt;
    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700))
}

#[cfg(not(unix))]
fn set_dir_perms(_path: &Path) -> std::io::Result<()> {
    Ok(())
}

/// Set a file to `0600` (owner read/write only). No-op on non-Unix.
#[cfg(unix)]
fn set_file_perms(path: &Path) -> std::io::Result<()> {
    use std::os::unix::fs::PermissionsExt;
    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))
}

#[cfg(not(unix))]
fn set_file_perms(_path: &Path) -> std::io::Result<()> {
    Ok(())
}

/// Mark a directory backup-excluded so Time Machine / iCloud don't copy
/// plaintext traces off the machine (R14). Best-effort: writes a
/// `.nobackup` marker file and, on macOS, sets the
/// `com.apple.metadata:com_apple_backup_excludeItem` xattr. Failures are
/// swallowed — exclusion is a hardening measure, not a correctness gate.
fn mark_backup_excluded(dir: &Path) {
    // Portable marker — picked up by some backup tools and a clear human
    // signal regardless of platform.
    let _ = std::fs::write(
        dir.join(".nobackup"),
        b"car run traces - excluded from backup\n",
    );
    #[cfg(target_os = "macos")]
    set_macos_backup_excluded(dir);
}

/// Set the macOS Time Machine exclusion xattr on `dir`. Uses the `xattr`
/// CLI (always present on macOS) rather than linking a native crate. Time
/// Machine treats any non-empty value on this attr as "exclude". Best-
/// effort — a missing `xattr` or a failure is ignored.
#[cfg(target_os = "macos")]
fn set_macos_backup_excluded(dir: &Path) {
    // `xattr -w` writes a string value; Time Machine only checks for the
    // attr's presence/non-emptiness, not its exact bytes.
    let _ = std::process::Command::new("xattr")
        .args(["-w", "com.apple.metadata:com_apple_backup_excludeItem", "1"])
        .arg(dir)
        .output();
}

#[cfg(test)]
mod tests {
    use super::*;
    use car_ir::{AgentOutcome, OutcomeMetrics, OutcomeStatus};
    use car_proto::{RunEnded, RunStarted, RunTurn, VerifierVerdict};
    use serde_json::json;

    fn store(root: PathBuf) -> RunStore {
        RunStore::new(root, RetentionConfig::default())
    }

    fn started(run_id: &str, agent_id: &str, when: DateTime<Utc>) -> RunStarted {
        RunStarted {
            run_id: run_id.to_string(),
            agent_id: agent_id.to_string(),
            intent: "do the thing".to_string(),
            outcome_description: None,
            started_at: when,
        }
    }

    fn turn(index: usize, prompt: &str) -> RunRecord {
        RunRecord::Turn(RunTurn {
            index,
            prompt: Some(prompt.to_string()),
            tool: Some("drive_cli".to_string()),
            parameters: json!({ "prompt": prompt }),
            output: Some(json!({ "exit_code": 0 })),
            cli_outcome: None,
            verifier_verdict: VerifierVerdict::NotRun,
            policy_rejected: None,
        })
    }

    fn ended(run_id: &str, agent_id: &str, status: OutcomeStatus) -> RunRecord {
        let outcome = AgentOutcome {
            status,
            summary: "done".to_string(),
            evidence: vec![],
            metrics: OutcomeMetrics::default(),
            timestamp: Utc::now(),
        };
        RunRecord::Ended(RunEnded {
            run_id: run_id.to_string(),
            agent_id: agent_id.to_string(),
            termination: RunTermination::Outcome { status, outcome },
            ended_at: Utc::now(),
        })
    }

    /// A completed run is readable from a brand-new store instance — the
    /// "simulated daemon restart" (new store, empty memory): the trace
    /// must come back from disk (R4).
    #[test]
    fn completed_run_readable_after_restart() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let s1 = store(root.clone());
        s1.write_started(&started("run-1", "agent-a", Utc::now()))
            .unwrap();
        s1.append_turns("agent-a", "run-1", &[turn(0, "first")])
            .unwrap();
        s1.append_records(
            "agent-a",
            "run-1",
            &[ended("run-1", "agent-a", OutcomeStatus::Success)],
        )
        .unwrap();

        // Brand-new store over the same root = simulated restart with empty memory.
        let s2 = store(root);
        let trace = s2
            .get_run_trace("run-1")
            .expect("trace readable after restart");
        assert!(matches!(trace.first(), Some(RunRecord::Started(_))));
        assert!(matches!(trace.last(), Some(RunRecord::Ended(_))));
        let turns = trace
            .iter()
            .filter(|r| matches!(r, RunRecord::Turn(_)))
            .count();
        assert_eq!(turns, 1);
    }

    /// Each run persists to its own `(agent_id, run_id)` file — no
    /// cross-contamination between runs or agents (R1).
    #[test]
    fn runs_isolated_per_agent_and_run() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = store(tmp.path().join("runs"));
        s.write_started(&started("run-1", "agent-a", Utc::now()))
            .unwrap();
        s.append_turns("agent-a", "run-1", &[turn(0, "a-first")])
            .unwrap();
        s.write_started(&started("run-2", "agent-a", Utc::now()))
            .unwrap();
        s.append_turns("agent-a", "run-2", &[turn(0, "a-second")])
            .unwrap();
        s.write_started(&started("run-3", "agent-b", Utc::now()))
            .unwrap();
        s.append_turns("agent-b", "run-3", &[turn(0, "b-first")])
            .unwrap();

        // Distinct files; each holds only its own turn.
        let t1 = s.get_run_trace("run-1").unwrap();
        let t2 = s.get_run_trace("run-2").unwrap();
        let t3 = s.get_run_trace("run-3").unwrap();
        assert_eq!(turn_prompt(&t1), "a-first");
        assert_eq!(turn_prompt(&t2), "a-second");
        assert_eq!(turn_prompt(&t3), "b-first");
        // run_id -> agent_id resolution works for replay authz.
        assert_eq!(s.agent_for_run("run-1").as_deref(), Some("agent-a"));
        assert_eq!(s.agent_for_run("run-3").as_deref(), Some("agent-b"));
        // agent-a lists 2 runs, agent-b lists 1.
        assert_eq!(s.list_runs("agent-a").len(), 2);
        assert_eq!(s.list_runs("agent-b").len(), 1);
    }

    fn turn_prompt(trace: &[RunRecord]) -> String {
        trace
            .iter()
            .find_map(|r| match r {
                RunRecord::Turn(t) => t.prompt.clone(),
                _ => None,
            })
            .unwrap_or_default()
    }

    /// Files are `0600` and dirs are `0700` (R14 — assert the modes).
    #[cfg(unix)]
    #[test]
    fn perms_are_0600_files_0700_dirs() {
        use std::os::unix::fs::PermissionsExt;
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let s = store(root.clone());
        s.write_started(&started("run-1", "agent-a", Utc::now()))
            .unwrap();

        let file = root.join("agent-a").join("run-1.jsonl");
        let fmode = std::fs::metadata(&file).unwrap().permissions().mode() & 0o777;
        assert_eq!(fmode, 0o600, "run file must be 0600, got {:o}", fmode);

        let root_mode = std::fs::metadata(&root).unwrap().permissions().mode() & 0o777;
        assert_eq!(
            root_mode, 0o700,
            "runs/ dir must be 0700, got {:o}",
            root_mode
        );
        let agent_mode = std::fs::metadata(root.join("agent-a"))
            .unwrap()
            .permissions()
            .mode()
            & 0o777;
        assert_eq!(
            agent_mode, 0o700,
            "agent dir must be 0700, got {:o}",
            agent_mode
        );

        // The backup-exclusion marker is present.
        assert!(root.join(".nobackup").exists(), ".nobackup marker written");
    }

    /// A run with no `RunEnded` reads `InProgress`; once the disconnect
    /// path writes the `Incomplete` terminal it reads `Incomplete` — never
    /// silently `Completed`. This is the R5 distinguishability the
    /// dashboard renders.
    #[test]
    fn orphan_run_status_distinguishes_inprogress_from_incomplete() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = store(tmp.path().join("runs"));
        // Stale start time (no harness writing — an orphan).
        let stale = Utc::now() - chrono::Duration::hours(6);
        s.write_started(&started("run-1", "agent-a", stale)).unwrap();
        s.append_turns("agent-a", "run-1", &[turn(0, "first")])
            .unwrap();

        // Still open: no terminal record yet.
        let open = &s.list_runs("agent-a")[0];
        assert_eq!(open.status, RunStatus::InProgress);

        // Disconnect path writes the Incomplete terminal.
        let incomplete = RunRecord::Ended(RunEnded {
            run_id: "run-1".to_string(),
            agent_id: "agent-a".to_string(),
            termination: RunTermination::Incomplete,
            ended_at: Utc::now(),
        });
        s.append_records("agent-a", "run-1", &[incomplete]).unwrap();
        let closed = &s.list_runs("agent-a")[0];
        assert_eq!(closed.status, RunStatus::Incomplete);
    }

    /// Retention evicts beyond the per-agent cap, never an in-progress
    /// run (R6).
    #[test]
    fn gc_evicts_beyond_per_agent_cap_but_never_in_progress() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let s = RunStore::new(
            root,
            RetentionConfig {
                max_per_agent: 3,
                max_age_days: 30,
            },
        );
        // 5 completed runs with increasing start times.
        let base = Utc::now() - chrono::Duration::days(1);
        for i in 0..5 {
            let id = format!("c{i}");
            let when = base + chrono::Duration::minutes(i);
            s.write_started(&started(&id, "agent-a", when)).unwrap();
            s.append_records(
                "agent-a",
                &id,
                &[ended(&id, "agent-a", OutcomeStatus::Success)],
            )
            .unwrap();
        }
        // 1 in-progress run (no terminal) — must survive GC.
        s.write_started(&started("live", "agent-a", Utc::now()))
            .unwrap();

        let removed = s.gc();
        // Keep 3 most-recent completed; evict the 2 oldest completed. The
        // in-progress run is never counted/evicted.
        assert_eq!(removed, 2, "should evict the 2 oldest completed runs");
        let remaining = s.list_runs("agent-a");
        // 3 kept completed + the live one = 4.
        assert_eq!(remaining.len(), 4);
        assert!(
            remaining.iter().any(|r| r.run_id == "live"),
            "in-progress run must never be evicted"
        );
        // The two oldest (c0, c1) are gone.
        assert!(!remaining.iter().any(|r| r.run_id == "c0"));
        assert!(!remaining.iter().any(|r| r.run_id == "c1"));
    }

    /// Retention evicts completed runs older than the age cap (R6).
    #[test]
    fn gc_evicts_runs_older_than_age_cap() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = RunStore::new(
            tmp.path().join("runs"),
            RetentionConfig {
                max_per_agent: 50,
                max_age_days: 30,
            },
        );
        // One old completed run — started AND ended 40 days ago (its
        // terminal time is what the age cap measures, FIX 2) — and one
        // fresh run.
        let old = Utc::now() - chrono::Duration::days(40);
        s.write_started(&started("old", "agent-a", old)).unwrap();
        s.append_records(
            "agent-a",
            "old",
            &[ended_at("old", "agent-a", OutcomeStatus::Success, old)],
        )
        .unwrap();
        s.write_started(&started("fresh", "agent-a", Utc::now()))
            .unwrap();
        s.append_records(
            "agent-a",
            "fresh",
            &[ended("fresh", "agent-a", OutcomeStatus::Success)],
        )
        .unwrap();

        let removed = s.gc();
        assert_eq!(removed, 1, "the 40-day-old run should be evicted");
        let remaining = s.list_runs("agent-a");
        assert_eq!(remaining.len(), 1);
        assert_eq!(remaining[0].run_id, "fresh");
    }

    /// An old run that is still in progress is NOT evicted by the age cap
    /// (R6 — never evict an open run, even a stale one).
    #[test]
    fn gc_never_evicts_stale_in_progress_run() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = RunStore::new(
            tmp.path().join("runs"),
            RetentionConfig {
                max_per_agent: 1,
                max_age_days: 1,
            },
        );
        let old = Utc::now() - chrono::Duration::days(40);
        // Old + in-progress (no terminal).
        s.write_started(&started("stale-live", "agent-a", old))
            .unwrap();
        let removed = s.gc();
        assert_eq!(removed, 0);
        assert!(s
            .list_runs("agent-a")
            .iter()
            .any(|r| r.run_id == "stale-live"));
    }

    /// A corrupt/partial trailing JSONL line loads the prior valid
    /// records rather than failing the whole run (error path).
    #[test]
    fn corrupt_trailing_line_loads_prior_records() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let s = store(root.clone());
        s.write_started(&started("run-1", "agent-a", Utc::now()))
            .unwrap();
        s.append_turns("agent-a", "run-1", &[turn(0, "first"), turn(1, "second")])
            .unwrap();

        // Append a partial/garbage line directly (simulating a crash
        // mid-append).
        let path = root.join("agent-a").join("run-1.jsonl");
        let mut f = std::fs::OpenOptions::new().append(true).open(&path).unwrap();
        writeln!(f, "{{\"record\":\"turn\",\"index\":2,\"prom").unwrap();

        let trace = s.get_run_trace("run-1").expect("trace still loads");
        // RunStarted + 2 valid turns; the garbage trailing line is dropped.
        let turns = trace
            .iter()
            .filter(|r| matches!(r, RunRecord::Turn(_)))
            .count();
        assert_eq!(turns, 2, "prior valid turns load; corrupt line skipped");
        assert!(matches!(trace.first(), Some(RunRecord::Started(_))));
    }

    /// `list_runs` for an agent with no runs returns empty (empty-state).
    #[test]
    fn list_runs_empty_for_unknown_agent() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = store(tmp.path().join("runs"));
        assert!(s.list_runs("nobody").is_empty());
        assert!(s.get_run_trace("nope").is_none());
        assert!(s.agent_for_run("nope").is_none());
    }

    /// `from_journal_dir` roots the store at `<car_dir>/runs` as a sibling
    /// of the journal dir.
    #[test]
    fn from_journal_dir_roots_at_car_runs() {
        let s = RunStore::from_journal_dir(Path::new("/home/u/.car/journals"));
        assert_eq!(s.root(), Path::new("/home/u/.car/runs"));
    }

    /// Retention config loads `[runs]` overrides from config.toml; missing
    /// keys keep the restrictive default.
    #[test]
    fn retention_config_reads_overrides() {
        let tmp = tempfile::TempDir::new().unwrap();
        std::fs::write(
            tmp.path().join("config.toml"),
            "[runs]\nmax_per_agent = 10\n",
        )
        .unwrap();
        let cfg = RetentionConfig::from_car_dir(tmp.path());
        assert_eq!(cfg.max_per_agent, 10);
        // Missing key keeps the default.
        assert_eq!(cfg.max_age_days, DEFAULT_MAX_AGE_DAYS);
    }

    /// A missing/malformed config.toml falls back to the restrictive
    /// default — never refuses.
    #[test]
    fn retention_config_defaults_on_missing_file() {
        let tmp = tempfile::TempDir::new().unwrap();
        let cfg = RetentionConfig::from_car_dir(tmp.path());
        assert_eq!(cfg.max_per_agent, DEFAULT_MAX_RUNS_PER_AGENT);
        assert_eq!(cfg.max_age_days, DEFAULT_MAX_AGE_DAYS);
    }

    /// Append the terminal `RunEnded` with an explicit `ended_at` (the
    /// disconnect/complete path uses `Utc::now()`, but GC ages by terminal
    /// time, so tests need to control it).
    fn ended_at(
        run_id: &str,
        agent_id: &str,
        status: OutcomeStatus,
        when: DateTime<Utc>,
    ) -> RunRecord {
        let outcome = AgentOutcome {
            status,
            summary: "done".to_string(),
            evidence: vec![],
            metrics: OutcomeMetrics::default(),
            timestamp: when,
        };
        RunRecord::Ended(RunEnded {
            run_id: run_id.to_string(),
            agent_id: agent_id.to_string(),
            termination: RunTermination::Outcome { status, outcome },
            ended_at: when,
        })
    }

    /// FIX 2: a run that STARTED >max_age_days ago but COMPLETED recently is
    /// a fresh result and must NOT be evicted by the age cap. GC must age by
    /// the terminal time, not the start time.
    #[test]
    fn gc_age_cap_uses_terminal_time_not_start() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = RunStore::new(
            tmp.path().join("runs"),
            RetentionConfig {
                max_per_agent: 50,
                max_age_days: 30,
            },
        );
        // Long-running run: started 40 days ago, completed 1 day ago.
        let started_40d = Utc::now() - chrono::Duration::days(40);
        let ended_1d = Utc::now() - chrono::Duration::days(1);
        s.write_started(&started("long", "agent-a", started_40d))
            .unwrap();
        s.append_records(
            "agent-a",
            "long",
            &[ended_at("long", "agent-a", OutcomeStatus::Success, ended_1d)],
        )
        .unwrap();

        let removed = s.gc();
        assert_eq!(
            removed, 0,
            "a run completed 1 day ago must survive the 30-day age cap, \
             even if it started 40 days ago"
        );
        let remaining = s.list_runs("agent-a");
        assert_eq!(remaining.len(), 1);
        assert_eq!(remaining[0].run_id, "long");
    }

    /// FIX 4: a crash-orphaned run (RunStarted + Turn, no RunEnded) on disk
    /// reads `InProgress`. A fresh store's `adopt_orphans()` at boot — when
    /// the in-memory map is empty — appends an `Incomplete` terminal so the
    /// run becomes terminal (and thus age-GC-eligible), no longer leaking.
    #[test]
    fn adopt_orphans_marks_crashed_inprogress_runs_incomplete() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        // Prior process wrote a start + a turn, then crashed (no terminal).
        let s1 = store(root.clone());
        s1.write_started(&started("orphan", "agent-a", Utc::now()))
            .unwrap();
        s1.append_turns("agent-a", "orphan", &[turn(0, "first")])
            .unwrap();
        assert_eq!(
            s1.list_runs("agent-a")[0].status,
            RunStatus::InProgress,
            "precondition: orphan reads InProgress before adoption"
        );

        // Fresh store = new process boot, empty in-memory map.
        let s2 = store(root);
        let adopted = s2.adopt_orphans();
        assert_eq!(adopted, 1, "the crash orphan should be adopted");

        let after = &s2.list_runs("agent-a")[0];
        assert_eq!(
            after.status,
            RunStatus::Incomplete,
            "adopted orphan now reads Incomplete (terminal)"
        );
        assert!(after.ended_at.is_some(), "terminal record has an ended_at");

        // Idempotent: a second boot finds no orphans (already terminal).
        assert_eq!(s2.adopt_orphans(), 0);
    }

    /// FIX 4 (corollary): a completed run is NOT adopted (it already has a
    /// terminal record).
    #[test]
    fn adopt_orphans_leaves_completed_runs_alone() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let s = store(root);
        s.write_started(&started("done", "agent-a", Utc::now()))
            .unwrap();
        s.append_records(
            "agent-a",
            "done",
            &[ended("done", "agent-a", OutcomeStatus::Success)],
        )
        .unwrap();
        assert_eq!(s.adopt_orphans(), 0);
        assert_eq!(s.list_runs("agent-a")[0].status, RunStatus::Completed);
    }

    /// FIX 7: a torn tail (a partial line with no trailing '\n') must not
    /// cause the NEXT valid append to be dropped. `append_records` inserts a
    /// separating '\n' so the torn fragment is its own skippable line and
    /// the new record loads intact.
    #[test]
    fn torn_tail_does_not_drop_following_valid_record() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let s = store(root.clone());
        s.write_started(&started("run-1", "agent-a", Utc::now()))
            .unwrap();
        s.append_turns("agent-a", "run-1", &[turn(0, "first")])
            .unwrap();

        // Simulate a torn append: a partial line with NO trailing newline.
        let path = root.join("agent-a").join("run-1.jsonl");
        {
            let mut f = std::fs::OpenOptions::new().append(true).open(&path).unwrap();
            // No writeln! — deliberately leaves the file's last byte != '\n'.
            f.write_all(b"{\"record\":\"turn\",\"index\":1,\"prom").unwrap();
        }
        assert!(
            last_byte_is_not_newline(&path).unwrap(),
            "precondition: tail is torn (no trailing newline)"
        );

        // Now append a FULLY VALID record via the store. Without the FIX-7
        // leading newline this would concatenate onto the torn fragment and
        // be dropped by load_records.
        s.append_turns("agent-a", "run-1", &[turn(2, "third")])
            .unwrap();

        let trace = s.get_run_trace("run-1").expect("trace loads");
        // Started + turn 0 ("first") + the new valid turn ("third"). The
        // torn middle fragment is skipped; the valid record after it loads.
        let turn_prompts: Vec<String> = trace
            .iter()
            .filter_map(|r| match r {
                RunRecord::Turn(t) => t.prompt.clone(),
                _ => None,
            })
            .collect();
        assert!(
            turn_prompts.contains(&"third".to_string()),
            "the valid record appended after a torn tail must survive, got {:?}",
            turn_prompts
        );
        assert!(matches!(trace.first(), Some(RunRecord::Started(_))));
    }
}