Skip to main content

leviath_cli/
runstate.rs

1//! On-disk run state for background agent executions.
2//!
3//! Each run lives under `~/.leviath/runs/<run-id>/` with:
4//! - `meta.json`    - run metadata, updated atomically (tmp + rename)
5//! - `output.log`  - append-only combined worker stdout (legacy/fallback)
6//! - `stages.json` - index of per-stage records
7//! - `stages/<idx>/output.log` - readable agent output for that stage
8//! - `stages/<idx>/logs.log`   - operational events + tool activity
9//! - `stages/<idx>/context.json` - context snapshot for that stage
10//!
11//! The dashboard's activity log is persisted separately at:
12//! - `~/.leviath/dashboard.log` - never cleared, appended across sessions
13//!
14//! # Who writes, and which copy is authoritative
15//!
16//! There are two answers to "what runs exist", and that is deliberate. The ECS
17//! world is the live one: it knows wait reasons and tick-fresh progress for the
18//! runs the daemon is holding right now, and `host.rs`'s `list()` reads it.
19//! The runs directory is the durable one: it survives a crash or a daemon that
20//! is not running, and `list_runs` below reads it. Disk lags the world by at
21//! most one persistence tick, so the two disagreeing is expected rather than a
22//! bug, and every reconciliation of that gap goes through `looks_abandoned`.
23//!
24//! The runtime's `persistence_bridge` is the only thing that writes a live
25//! run's state. The writers in this module are `#[cfg(test)]` so that stays
26//! true by compilation rather than by convention: a test can lay down a run
27//! directory to read back, and production has no second path to the same files.
28
29use std::path::{Path, PathBuf};
30use std::sync::Arc;
31use std::time::{SystemTime, UNIX_EPOCH};
32
33// The plain run-state data types (RunMeta, RunStatus, the snapshot structs, and
34// the per-stage records) live in `leviath_core::run_meta`. Re-exported here so
35// `crate::runstate::RunMeta` / `runstate::RunMeta` call sites across the cli
36// resolve. All on-disk IO for these types remains in this module.
37pub use leviath_core::run_meta::{
38    ContextSnapshot, RegionEntrySnapshot, RegionSnapshot, RunMeta, RunStatus, StageRecord,
39    StageRunStatus,
40};
41
42/// Atomically write a context snapshot for the run.
43///
44/// Test-only. Production writes go through the runtime's `persistence_bridge`,
45/// which is the sole writer of a live run's on-disk state; this exists so a
46/// test can lay down a run directory to read back. See the module doc.
47#[cfg(test)]
48pub fn write_context_snapshot(run_id: &str, snap: &ContextSnapshot) -> anyhow::Result<()> {
49    write_context_snapshot_to(&run_dir(run_id), snap)
50}
51
52/// Atomically write pre-serialized `json` to `path` (via a `.json.tmp`
53/// sibling + rename).
54///
55/// Non-generic (takes an already-serialized string) so it has a single
56/// monomorphization and every region - including the `std::fs` error `?`
57/// arms - is exercised by real tests. Serialization is performed by the
58/// callers, whose concrete production types
59/// (`ContextSnapshot`/`RunMeta`/`&[StageRecord]`) are provably infallible to
60/// serialize (see the `.expect` sites).
61/// Write `body` to `path` atomically, readable only by this user.
62///
63/// Not JSON-specific despite where it started: the final-output sidecar is raw
64/// content, and wants the same private-then-rename treatment for the same
65/// reason.
66fn write_private_atomic(path: &std::path::Path, body: &str) -> anyhow::Result<()> {
67    let tmp = path.with_extension("tmp");
68    // `write_private`: these files carry the run's full task prompt,
69    // conversation and tool output - and `meta.json` carries the webhook
70    // signing secret. They were written with a plain `fs::write` at the umask
71    // default (typically 0644), protected only by the 0700 on the enclosing run
72    // directory. That is one `chmod` away from being readable, and defence in
73    // depth is the whole point of a mode on the file itself.
74    leviath_sys::write_private(&tmp, body.as_bytes())?;
75    std::fs::rename(&tmp, path)?;
76    Ok(())
77}
78
79#[cfg(test)]
80fn write_context_snapshot_to(dir: &std::path::Path, snap: &ContextSnapshot) -> anyhow::Result<()> {
81    let json = serde_json::to_string_pretty(snap)
82        .expect("infallible: ContextSnapshot always serializes to JSON");
83    write_private_atomic(&dir.join("context.json"), &json)
84}
85
86/// Read the context snapshot for a run, if present.
87pub fn read_context_snapshot(run_id: &str) -> Option<ContextSnapshot> {
88    let path = run_dir(run_id).join("context.json");
89    let json = std::fs::read_to_string(&path).ok()?;
90    serde_json::from_str(&json).ok()
91}
92
93/// A parse cache keyed by a file's `(mtime, len)`: the file is re-read and
94/// re-parsed only when its stat changes.
95///
96/// For pollers reading run state on a tick. The dashboard synced at 10Hz by
97/// re-parsing every run's `meta.json`, `stages.json`, and whole
98/// `context.json`; with 50 runs on disk that was on the order of 100 MB/s of
99/// allocate-and-parse-and-free for files that change at most once per persist
100/// tick. A `stat` costs microseconds; this turns the steady-state tick into
101/// stats plus clones of shared `Arc`s.
102///
103/// `(mtime, len)` rather than mtime alone: the persistence lane's atomic
104/// rename gives every update a fresh temp inode and mtime, but coarse mtime
105/// granularity on some filesystems can miss two updates in the same instant -
106/// the length check catches most of those, and a same-length same-instant
107/// rewrite is indistinguishable anyway one tick later.
108pub struct StatCache<T> {
109    entries: std::collections::HashMap<PathBuf, (std::time::SystemTime, u64, Option<Arc<T>>)>,
110}
111
112impl<T> Default for StatCache<T> {
113    fn default() -> Self {
114        Self {
115            entries: std::collections::HashMap::new(),
116        }
117    }
118}
119
120impl<T> StatCache<T> {
121    /// The value parsed from `path`, re-reading only when the file's stat
122    /// changed since the last call. `None` when the file is missing,
123    /// unreadable, or `parse` rejects it - negative results are cached too, so
124    /// a persistently-bad file costs one stat per tick, not one parse.
125    pub fn get_with(
126        &mut self,
127        path: &Path,
128        parse: impl FnOnce(&str) -> Option<T>,
129    ) -> Option<Arc<T>> {
130        let Ok(meta) = std::fs::metadata(path) else {
131            self.entries.remove(path);
132            return None;
133        };
134        // A filesystem with no mtimes degrades to epoch (so length changes
135        // still refresh) rather than growing an unreachable error arm.
136        let stamp = (meta.modified().unwrap_or(std::time::UNIX_EPOCH), meta.len());
137        if let Some((mtime, len, value)) = self.entries.get(path)
138            && (*mtime, *len) == stamp
139        {
140            return value.clone();
141        }
142        let value = std::fs::read_to_string(path)
143            .ok()
144            .and_then(|text| parse(&text))
145            .map(Arc::new);
146        self.entries
147            .insert(path.to_path_buf(), (stamp.0, stamp.1, value.clone()));
148        value
149    }
150
151    /// Drop entries for files under runs that no longer exist, so a
152    /// long-lived poller's cache stays bounded by the live run set.
153    pub fn retain_under(&mut self, keep: &std::collections::HashSet<PathBuf>) {
154        self.entries.retain(|path, _| {
155            path.parent()
156                .is_some_and(|dir| keep.contains(&dir.to_path_buf()))
157        });
158    }
159}
160
161/// Read + parse a run's portable archive (`<run_dir>/run.lvr`), returning its
162/// records, or `None` if the archive is missing or unreadable.
163///
164/// Materializes the whole journal. For anything that only walks the timeline
165/// (the history API, journal search highlights), prefer [`visit_run_archive`]:
166/// a mature run's journal is tens of MB, and parsing it whole per request was
167/// the API's single largest transient allocation.
168pub fn read_run_archive(run_id: &str) -> Option<Vec<leviath_core::run_archive::RunRecord>> {
169    let path = run_dir(run_id).join("run.lvr");
170    let bytes = std::fs::read(&path).ok()?;
171    leviath_core::run_archive::read_archive(&mut bytes.as_slice())
172        .ok()
173        .map(|(_version, records)| records)
174}
175
176/// Stream a run's raw journal records through `visit`, one at a time, without
177/// materializing the archive. Same lenient tail handling as
178/// [`visit_run_archive`]. For consumers that inspect records rather than
179/// replayed points (journal search).
180pub fn visit_run_records(
181    run_id: &str,
182    visit: &mut dyn FnMut(&leviath_core::run_archive::RunRecord) -> std::ops::ControlFlow<()>,
183) -> Option<()> {
184    let path = run_dir(run_id).join("run.lvr");
185    let file = std::fs::File::open(&path).ok()?;
186    let mut reader = std::io::BufReader::with_capacity(64 * 1024, file);
187    leviath_core::run_archive::read_archive_start(&mut reader).ok()?;
188    while let Ok(Some(record)) = leviath_core::run_archive::read_record(&mut reader) {
189        if visit(&record).is_break() {
190            break;
191        }
192    }
193    Some(())
194}
195
196/// Stream a run's archive through a [`visit_points`] visitor without ever
197/// materializing the journal: one buffered pass over `run.lvr`, one record and
198/// one running window in memory. Returns `None` if the archive is missing or
199/// its preamble is invalid; a torn tail (a live run mid-append) just ends the
200/// walk with the points already visited.
201///
202/// [`visit_points`]: leviath_core::run_archive::visit_points
203pub fn visit_run_archive(
204    run_id: &str,
205    visit: &mut dyn FnMut(leviath_core::run_archive::PointRef<'_>) -> std::ops::ControlFlow<()>,
206) -> Option<()> {
207    let path = run_dir(run_id).join("run.lvr");
208    let file = std::fs::File::open(&path).ok()?;
209    let mut reader = std::io::BufReader::with_capacity(64 * 1024, file);
210    leviath_core::run_archive::visit_archive_points(&mut reader, visit).ok()
211}
212
213/// A run's context-window history: the full window (+ metadata) at each recorded
214/// point over time, oldest first. Empty when there's no readable archive.
215///
216/// Every point's `meta` is [`RunMeta::redacted`]. The journal stores `RunMeta`
217/// whole - including `callback_secret`, which the daemon needs to keep signing
218/// webhooks for a run it reloads - so a replayed point carries the secret unless
219/// it is stripped here. `GET /api/agents/{id}/context/history` serialized these
220/// points directly, which handed the webhook signing key to any holder of the
221/// API token: the same disclosure `redacted()` was introduced for on
222/// `/api/agents`, re-opened through the archive.
223///
224/// Redacted in this shared reader rather than in that one handler so the next
225/// consumer of a run's history inherits the fix instead of having to remember
226/// it. No caller needs the secret: the CLI printer, the dashboard, and the API
227/// all only display these points.
228pub fn context_history(run_id: &str) -> Vec<leviath_core::run_archive::RunPoint> {
229    read_run_archive(run_id)
230        .map(|records| leviath_core::run_archive::replay_points(&records))
231        .unwrap_or_default()
232        .into_iter()
233        .map(|point| leviath_core::run_archive::RunPoint {
234            meta: point.meta.redacted(),
235            ..point
236        })
237        .collect()
238}
239
240fn now_secs() -> i64 {
241    SystemTime::now()
242        .duration_since(UNIX_EPOCH)
243        .map(|d| d.as_secs() as i64)
244        .unwrap_or(0)
245}
246
247/// Inner implementation of `runs_dir`, parameterised so it can be tested
248/// without touching the process-global env. All callers go through `runs_dir`.
249///
250/// The fallback resolves through [`crate::config::leviath_home_dir`], not
251/// `dirs::home_dir` directly, so `LEVIATH_HOME` redirects the runs dir like it
252/// redirects the config, the control socket and the agents dir. With the raw
253/// OS home instead, a test that sets `LEVIATH_HOME` would be isolated
254/// everywhere *except* here and still write runs into the developer's real
255/// `~/.leviath/runs`. `LEVIATH_RUNS_DIR` wins over both.
256fn runs_dir_from(env_override: Option<&str>) -> PathBuf {
257    if let Some(dir) = env_override {
258        return PathBuf::from(dir);
259    }
260    leviath_core::paths::data_dir()
261        .unwrap_or_default()
262        .join("runs")
263}
264
265/// Directory where all run state is stored.
266pub fn runs_dir() -> PathBuf {
267    runs_dir_from(std::env::var("LEVIATH_RUNS_DIR").ok().as_deref())
268}
269
270/// Directory for a specific run.
271///
272/// A `run_id` that is not a single safe path component resolves to
273/// `<runs_dir>/<invalid>`, a name that cannot exist - so a caller that passes an
274/// attacker-supplied id gets a miss rather than a traversal. `run_id` reaches
275/// this from URL segments on `GET /api/agents/{id}/logs` and friends, where
276/// `Path::join` would otherwise happily accept `../../` or an absolute path.
277///
278/// Returning a definitely-missing path rather than an `Option` keeps every
279/// caller's "no such run" branch as the single failure path, instead of adding a
280/// second one that all of them would have to handle identically.
281pub fn run_dir(run_id: &str) -> PathBuf {
282    if !leviath_core::is_safe_path_component(run_id) {
283        tracing::warn!(run_id = %run_id, "rejected an unsafe run id");
284        return runs_dir().join("<invalid>");
285    }
286    runs_dir().join(run_id)
287}
288
289/// Inner implementation of `dashboard_log_path`, parameterised so it can be
290/// tested without touching the process-global env. All callers go through
291/// `dashboard_log_path`.
292fn dashboard_log_path_from(env_override: Option<&str>) -> PathBuf {
293    if let Some(path) = env_override {
294        return PathBuf::from(path);
295    }
296    leviath_core::paths::data_dir()
297        .unwrap_or_default()
298        .join("dashboard.log")
299}
300
301/// Path to the persistent dashboard activity log (~/.leviath/dashboard.log).
302///
303/// Honours the `LEVIATH_DASHBOARD_LOG_PATH` override when set (tests use it via
304/// `isolate_runs_dir_for_test`); otherwise resolves the real home-relative
305/// path. This function only *computes* a `PathBuf` - it never writes - so both
306/// arms are safe to exercise directly in tests. The write side
307/// ([`append_dashboard_log`] and `Dashboard::add_log`) is what must stay off
308/// the user's real log in tests: `append_dashboard_log`'s own tests set the
309/// override, and `Dashboard` carries an injected log path (a temp dir under
310/// `make_test_dashboard`) so no dashboard-input test ever appends to the real
311/// `~/.leviath/dashboard.log`.
312pub fn dashboard_log_path() -> PathBuf {
313    match std::env::var("LEVIATH_DASHBOARD_LOG_PATH") {
314        Ok(path) => dashboard_log_path_from(Some(&path)),
315        Err(_) => dashboard_log_path_from(None),
316    }
317}
318
319/// Append a timestamped line to the persistent dashboard activity log at the
320/// default [`dashboard_log_path`]. Silently ignores I/O errors - best-effort.
321pub fn append_dashboard_log(msg: &str) {
322    append_dashboard_log_to(&dashboard_log_path(), msg);
323}
324
325/// Append a timestamped line to the dashboard activity log at an explicit
326/// `path`. Silently ignores I/O errors - the dashboard log is best-effort.
327///
328/// The path is a parameter so `Dashboard` can inject a test-isolated log
329/// location, guaranteeing no dashboard-input test appends to the user's real
330/// `~/.leviath/dashboard.log` (see [`dashboard_log_path`]).
331pub fn append_dashboard_log_to(path: &Path, msg: &str) {
332    append_dashboard_log_capped(path, msg, DASHBOARD_LOG_MAX_BYTES);
333}
334
335/// The dashboard log is capped at this size; once the live file reaches it, the
336/// file is rolled (see [`roll_log_if_over_cap`]) so it can't grow without bound
337/// across a long-lived daemon's lifetime.
338const DASHBOARD_LOG_MAX_BYTES: u64 = 5 * 1024 * 1024;
339
340/// Append with an explicit cap (the public entry points use
341/// [`DASHBOARD_LOG_MAX_BYTES`]; tests pass a small cap to exercise rolling).
342fn append_dashboard_log_capped(path: &Path, msg: &str, max_bytes: u64) {
343    use std::io::Write;
344    // Ensure the parent directory exists (first-run case).
345    if let Some(parent) = path.parent() {
346        let _ = std::fs::create_dir_all(parent);
347    }
348    roll_log_if_over_cap(path, max_bytes);
349    if let Ok(mut file) = leviath_sys::open_private_append(path) {
350        let timestamp = chrono::Local::now().format("%Y-%m-%d %H:%M:%S");
351        let _ = writeln!(file, "{} {}", timestamp, msg);
352    }
353}
354
355/// The path the rolled (previous-generation) log is moved to: `<name>.1`.
356fn rolled_log_path(path: &Path) -> PathBuf {
357    let mut name = path.as_os_str().to_owned();
358    name.push(".1");
359    PathBuf::from(name)
360}
361
362/// Roll the live log to `<name>.1` once it reaches `max_bytes`, replacing any
363/// existing rolled file, so the live file restarts empty and at most one
364/// previous generation is retained (bounded ~2×cap on disk). Best-effort - a
365/// failed rename just leaves the log to keep growing rather than erroring.
366fn roll_log_if_over_cap(path: &Path, max_bytes: u64) {
367    let over = std::fs::metadata(path)
368        .map(|m| m.len() >= max_bytes)
369        .unwrap_or(false);
370    if over {
371        let _ = std::fs::rename(path, rolled_log_path(path));
372    }
373}
374
375/// How many random bits go in a run ID's suffix, rendered as 12 hex digits.
376/// Collisions only matter within one wall-clock second for one agent name, so 48
377/// bits is many orders of magnitude more than needed while staying short enough
378/// to read in `lev ps` and the dashboard.
379const RUN_ID_ENTROPY_BITS: u32 = 48;
380
381/// Generate a unique run ID: `<agent_name>-<timestamp>-<random>`.
382///
383/// The suffix is **random**, not derived. A derived suffix like
384/// `(now ^ (now >> 16) ^ counter)` over a process-local counter defends a
385/// `lev run --count N` batch inside one process but degenerates to a pure
386/// function of the current second across separate processes: three concurrent
387/// `lev run` invocations all mint `fetcher-1785127214-8b48` and silently share
388/// one run directory. Nothing downstream detects that - `create_dir_all` is a
389/// no-op on an existing directory and the persistence worker then
390/// last-writer-wins over `meta.json` / `context.json` / `run.lvr`, interleaving
391/// two runs' state irrecoverably.
392///
393/// The `<name>-<secs>-<hex>` shape is preserved: the timestamp keeps IDs sorting
394/// and reading chronologically, and the dashboard's short-ID display
395/// (`split('-').next_back()`) still lands on the unique component.
396///
397/// The name is folded to **ASCII** alphanumerics, which is stricter than it
398/// looks necessary. The id becomes a directory name, and [`run_dir`] resolves an
399/// id that is not a safe path component to `<invalid>`. A Unicode fold let an
400/// agent named `café` mint `café-...`: the daemon created that directory
401/// happily, and then every CLI read of the run looked in `<invalid>` and found
402/// nothing. The minter has to satisfy the rule the readers enforce.
403pub fn new_run_id(agent_name: &str) -> String {
404    use rand::RngExt as _;
405    let entropy: u64 = rand::rng().random::<u64>() >> (u64::BITS - RUN_ID_ENTROPY_BITS);
406    let safe_name = agent_name.replace(|c: char| !c.is_ascii_alphanumeric() && c != '-', "-");
407    format!("{}-{}-{:012x}", safe_name, now_secs(), entropy)
408}
409
410/// Create the run directory and write initial metadata.
411pub fn create_run(meta: &RunMeta) -> anyhow::Result<()> {
412    create_run_in(&run_dir(&meta.run_id), meta)
413}
414
415/// Create an explicit run directory and write initial metadata into it.
416///
417/// Callers that already know the directory should prefer this over
418/// [`create_run`], which resolves it from the home directory - the daemon's
419/// spawner stakes out the run dir under its own configured `runs_dir`.
420pub(crate) fn create_run_in(dir: &std::path::Path, meta: &RunMeta) -> anyhow::Result<()> {
421    std::fs::create_dir_all(dir)?;
422
423    // Restrict the run directory to owner-only (no-op on non-Unix).
424    let _ = leviath_sys::secure_dir_perms(dir);
425
426    write_meta_to(dir, meta)
427}
428
429/// Atomically write run metadata (write to tmp, then rename).
430pub fn write_meta(meta: &RunMeta) -> anyhow::Result<()> {
431    write_meta_to(&run_dir(&meta.run_id), meta)
432}
433
434/// Atomically write `meta.json` into an explicit run directory.
435///
436/// Callers that already know the directory should prefer this over
437/// [`write_meta`], which resolves it from the home directory - the daemon's
438/// recovery pass works from its configured `runs_dir` instead.
439pub(crate) fn write_meta_to(dir: &std::path::Path, meta: &RunMeta) -> anyhow::Result<()> {
440    let json =
441        serde_json::to_string_pretty(meta).expect("infallible: RunMeta always serializes to JSON");
442    write_private_atomic(&dir.join("meta.json"), &json)
443}
444
445/// Read run metadata for a given run ID.
446pub fn read_meta(run_id: &str) -> anyhow::Result<RunMeta> {
447    read_meta_from(&run_dir(run_id))
448}
449
450/// Read a run's final output, content included.
451///
452/// The descriptor in `meta.json` says whether there is one and how big it is;
453/// this fetches the bytes from the sidecar beside it. Returns `None` when the
454/// run produced no answer, or when the sidecar is missing (a run written by a
455/// build that stored the answer inline, or one whose directory was pruned).
456pub fn read_final_output(run_id: &str) -> Option<leviath_core::FinalOutput> {
457    let meta = read_meta(run_id).ok()?;
458    let descriptor = meta.final_output?;
459    let content = std::fs::read_to_string(final_output_path(&run_dir(run_id))).ok()?;
460    Some(leviath_core::FinalOutput {
461        content,
462        format: descriptor.format,
463        stage: descriptor.stage,
464        submitted_at: descriptor.submitted_at,
465        truncated: descriptor.truncated,
466        artifacts: descriptor.artifacts,
467    })
468}
469
470/// Where a run's answer lives, beside its `meta.json`.
471pub fn final_output_path(dir: &std::path::Path) -> PathBuf {
472    dir.join(leviath_core::FINAL_OUTPUT_FILE)
473}
474
475/// Write a run's answer to its sidecar, atomically.
476///
477/// Raw content with no wrapper: serving it is a read, and `lev result --raw` is
478/// a copy. The descriptor in `meta.json` is what says it exists.
479///
480/// Test-only; see [`write_context_snapshot`].
481#[cfg(test)]
482pub fn write_final_output(dir: &std::path::Path, content: &str) -> anyhow::Result<()> {
483    write_private_atomic(&final_output_path(dir), content)
484}
485
486/// Whether an on-disk run status means the run has finished and should be left
487/// alone. `Starting`/`Running`/`WaitingInput` are all "still going" as far as
488/// anything reading the runs dir is concerned.
489pub fn is_terminal_status(status: &RunStatus) -> bool {
490    matches!(
491        status,
492        RunStatus::Complete
493            | RunStatus::CompleteInteractive
494            | RunStatus::Error
495            | RunStatus::Cancelled
496    )
497}
498
499/// How long a run may claim to be live on disk, while the daemon is not holding
500/// it, before anything treats it as abandoned.
501///
502/// Comfortably longer than the persistence heartbeat, so a live-but-slow run (a
503/// long inference writes nothing else) is never mistaken for a dead one.
504pub const STALE_AFTER_SECS: i64 = 300;
505
506/// Whether a run that claims to be live on disk has nothing driving it: the
507/// daemon is not holding it *and* it has not moved in [`STALE_AFTER_SECS`].
508///
509/// `live` is the set of run ids the daemon reports hosting, or `None` when it
510/// gave no answer this poll. Both halves are needed and each is wrong on its
511/// own. An unreachable daemon reports an empty set, so the id check alone would
512/// condemn every healthy run the moment the daemon restarted. And a run parked
513/// on a long inference legitimately does not move for minutes, so the clock
514/// alone would condemn a run that is working. `None` therefore answers `false`
515/// for everything: no answer is not evidence.
516///
517/// Ages against `last_progress_at`, falling back to `updated_at` for runs
518/// written before that field existed. The fallback preserves the older, weaker
519/// behavior for old runs rather than declaring them all stale at once.
520///
521/// One definition, shared by the dashboard's STALE badge and by `lev ps --all`,
522/// so what an operator sees and what a harness reconciles against cannot drift.
523pub fn looks_abandoned(
524    meta: &RunMeta,
525    live: Option<&std::collections::HashSet<String>>,
526    now: i64,
527) -> bool {
528    let Some(live) = live else {
529        return false; // no answer from the daemon; assume nothing
530    };
531    if is_terminal_status(&meta.status) || live.contains(&meta.run_id) {
532        return false;
533    }
534    let moved_at = meta.last_progress_at.unwrap_or(meta.updated_at);
535    now.saturating_sub(moved_at) > STALE_AFTER_SECS
536}
537
538/// The outcome of forcing a run to a terminal state on disk.
539#[derive(Debug, Clone, Copy, PartialEq, Eq)]
540pub enum ForceCancelOutcome {
541    /// The run was live on disk and is now recorded terminal.
542    Terminated,
543    /// The run was already finished; nothing was written.
544    AlreadyTerminal,
545    /// No run directory with that id exists.
546    NoSuchRun,
547    /// The directory exists but its metadata could not be rewritten.
548    WriteFailed,
549}
550
551impl ForceCancelOutcome {
552    /// Whether the id named a run at all - i.e. whether the cancel had a target,
553    /// regardless of whether it needed to write anything.
554    pub fn found_run(&self) -> bool {
555        !matches!(self, Self::NoSuchRun)
556    }
557}
558
559/// Force a run's on-disk metadata to `Cancelled`, in the runs dir resolved from
560/// the environment. See [`force_cancel_in`].
561pub fn force_cancel(run_id: &str) -> ForceCancelOutcome {
562    force_cancel_in(&run_dir(run_id), now_secs())
563}
564
565/// Force the run in `run_dir` to `Cancelled`, stamping `updated_at` with `now`.
566///
567/// This is the floor under every kill path: it needs nothing but the filesystem,
568/// so it works for a run the daemon can't rebuild (blueprint deleted, metadata
569/// corrupt, died mid-spawn) and for a run whose daemon is gone entirely. Both
570/// the daemon's force-terminator seam and `lev cancel --force` route here so
571/// there is one definition of "terminated on disk".
572///
573/// A directory whose `meta.json` is missing or unparseable still gets a minimal
574/// `Cancelled` record written: such a run is otherwise skipped by `list_runs`,
575/// which makes it invisible *and* permanent.
576pub fn force_cancel_in(run_dir: &Path, now: i64) -> ForceCancelOutcome {
577    force_terminal_in(run_dir, RunStatus::Cancelled, None, now)
578}
579
580/// Force the run in `run_dir` to `Error` with `message`, stamping `updated_at`.
581///
582/// For the spawn that never became a run. The spawner stakes out the run
583/// directory and writes a `Starting` placeholder *before* building the agent, so
584/// a spawn that fails leaves something to diagnose - but `Starting` is not
585/// terminal, so that placeholder went on claiming the run was alive for ever,
586/// showing up in `lev ps` and the dashboard with nothing behind it (issue #190).
587/// Recording the failure where the placeholder is turns it into an answer.
588pub fn force_error_in(run_dir: &Path, message: &str, now: i64) -> ForceCancelOutcome {
589    force_terminal_in(run_dir, RunStatus::Error, Some(message.to_string()), now)
590}
591
592/// Rewrite the run in `run_dir` to a terminal `status`, attaching `error` when
593/// there is something to say. Shared by [`force_cancel_in`] and
594/// [`force_error_in`] so "terminated on disk" has one implementation.
595fn force_terminal_in(
596    run_dir: &Path,
597    status: RunStatus,
598    error: Option<String>,
599    now: i64,
600) -> ForceCancelOutcome {
601    if !run_dir.is_dir() {
602        return ForceCancelOutcome::NoSuchRun;
603    }
604    let run_id = run_dir
605        .file_name()
606        .map(|n| n.to_string_lossy().into_owned())
607        .unwrap_or_default();
608    let terminated = match read_meta_from(run_dir) {
609        Ok(meta) if is_terminal_status(&meta.status) => return ForceCancelOutcome::AlreadyTerminal,
610        Ok(meta) => RunMeta {
611            status,
612            updated_at: now,
613            // Keep whatever the run had already recorded when there is nothing
614            // new to say (the cancel path).
615            error: error.clone().or(meta.error),
616            ..meta
617        },
618        // Unreadable metadata: synthesize just enough to record the outcome. The
619        // run id is the directory name, which is the one field always recoverable.
620        Err(_) => RunMeta {
621            status,
622            updated_at: now,
623            error: Some(
624                error
625                    .clone()
626                    .unwrap_or_else(|| "run metadata was unreadable; cancelled".to_string()),
627            ),
628            ..RunMeta::new(
629                run_id.clone(),
630                run_id,
631                String::new(),
632                String::new(),
633                None,
634                String::new(),
635                0,
636            )
637        },
638    };
639    match write_meta_to(run_dir, &terminated) {
640        Ok(()) => ForceCancelOutcome::Terminated,
641        Err(e) => {
642            // Formatted outside the macro: a method call inside a `%field` is
643            // only evaluated when a subscriber visits the value, so it would go
644            // unexercised under the tests' no-op subscriber.
645            let path = run_dir.display().to_string();
646            tracing::warn!(
647                run_dir = %path,
648                error = %e,
649                "could not force a run to a terminal state on disk"
650            );
651            ForceCancelOutcome::WriteFailed
652        }
653    }
654}
655
656/// Read run metadata out of an explicit run directory (the daemon works from its
657/// own configured `runs_dir` rather than the home-resolved one).
658pub(crate) fn read_meta_from(dir: &std::path::Path) -> anyhow::Result<RunMeta> {
659    let path = dir.join("meta.json");
660    let json = std::fs::read_to_string(&path)?;
661    Ok(serde_json::from_str(&json)?)
662}
663
664/// Inner implementation of `list_runs`, parameterised so the early-return
665/// branch can be exercised in tests without deleting real on-disk state.
666fn list_runs_in_dir(dir: PathBuf) -> Vec<RunMeta> {
667    if !dir.exists() {
668        return Vec::new();
669    }
670
671    let mut runs = Vec::new();
672
673    if let Ok(entries) = std::fs::read_dir(&dir) {
674        for entry in entries.filter_map(|e| e.ok()) {
675            let meta_path = entry.path().join("meta.json");
676            if let Ok(json) = std::fs::read_to_string(&meta_path)
677                && let Ok(meta) = serde_json::from_str::<RunMeta>(&json)
678            {
679                runs.push(meta);
680            }
681        }
682    }
683
684    runs.sort_by_key(|r| std::cmp::Reverse(r.started_at));
685    runs
686}
687
688/// List all runs, sorted by started_at descending (most recent first).
689/// Silently skips any runs whose metadata cannot be read.
690pub fn list_runs() -> Vec<RunMeta> {
691    list_runs_in_dir(runs_dir())
692}
693
694/// [`list_runs`] through a [`StatCache`], for pollers: each `meta.json` is
695/// re-parsed only when its stat changes, and cache entries for deleted runs
696/// are dropped. Same ordering and skip-unreadable behavior as `list_runs`.
697pub fn list_runs_cached(cache: &mut StatCache<RunMeta>) -> Vec<Arc<RunMeta>> {
698    let dir = runs_dir();
699    let mut runs = Vec::new();
700    let mut live_dirs = std::collections::HashSet::new();
701    if let Ok(entries) = std::fs::read_dir(&dir) {
702        for entry in entries.filter_map(|e| e.ok()) {
703            live_dirs.insert(entry.path());
704            let meta_path = entry.path().join("meta.json");
705            if let Some(meta) = cache.get_with(&meta_path, |json| {
706                serde_json::from_str::<RunMeta>(json).ok()
707            }) {
708                runs.push(meta);
709            }
710        }
711    }
712    cache.retain_under(&live_dirs);
713    runs.sort_by_key(|r| std::cmp::Reverse(r.started_at));
714    runs
715}
716
717/// [`read_stages_index`] through a [`StatCache`], for pollers.
718pub fn read_stages_index_cached(
719    run_id: &str,
720    cache: &mut StatCache<Vec<StageRecord>>,
721) -> Vec<StageRecord> {
722    let path = run_dir(run_id).join("stages.json");
723    cache
724        .get_with(&path, |json| serde_json::from_str(json).ok())
725        .map(|records| records.as_ref().clone())
726        .unwrap_or_default()
727}
728
729/// [`read_context_snapshot`] through a [`StatCache`], for pollers. The
730/// snapshot is shared, not cloned: a context window is the largest thing in a
731/// run dir, and handing out copies per tick is the churn this cache removes.
732pub fn read_context_snapshot_cached(
733    run_id: &str,
734    cache: &mut StatCache<ContextSnapshot>,
735) -> Option<Arc<ContextSnapshot>> {
736    let path = run_dir(run_id).join("context.json");
737    cache.get_with(&path, |json| serde_json::from_str(json).ok())
738}
739
740/// Read the last `max_bytes` of any file on disk, returning UTF-8 text.
741/// If the file is smaller than `max_bytes` the whole file is returned.
742/// Partial UTF-8 at the truncation boundary is handled by skipping to the
743/// first newline.  Returns an empty string on any I/O error.
744pub fn tail_file(path: &std::path::Path, max_bytes: u64) -> String {
745    use std::io::{Read, Seek, SeekFrom};
746
747    let mut file = match std::fs::File::open(path) {
748        Ok(f) => f,
749        Err(_) => return String::new(),
750    };
751
752    // Use fstat on the open fd rather than a separate stat() call - avoids the
753    // TOCTOU window between existence check and metadata read. Falls back to 0
754    // (read everything) if fstat somehow fails on an already-open fd.
755    let file_size = file.metadata().map(|m| m.len()).unwrap_or(0);
756
757    if file_size <= max_bytes {
758        let mut buf = Vec::new();
759        let _ = file.read_to_end(&mut buf);
760        return String::from_utf8_lossy(&buf).to_string();
761    }
762
763    let offset = file_size - max_bytes;
764    let _ = file.seek(SeekFrom::Start(offset));
765
766    let mut buf = Vec::new();
767    let _ = file.read_to_end(&mut buf);
768
769    // Skip to the first newline so we don't emit a partial line at the start.
770    if let Some(nl) = buf.iter().position(|&b| b == b'\n') {
771        String::from_utf8_lossy(&buf[nl + 1..]).to_string()
772    } else {
773        String::from_utf8_lossy(&buf).to_string()
774    }
775}
776
777// ─── Per-stage persistence ────────────────────────────────────────────────────
778
779/// Directory for per-stage files within a run.
780pub fn stage_dir(run_id: &str, stage_idx: usize) -> PathBuf {
781    run_dir(run_id).join("stages").join(stage_idx.to_string())
782}
783
784/// Atomically write the stages index for a run.
785///
786/// Test-only; see [`write_context_snapshot`].
787#[cfg(test)]
788pub fn write_stages_index(run_id: &str, stages: &[StageRecord]) -> anyhow::Result<()> {
789    write_stages_index_to(&run_dir(run_id), stages)
790}
791
792#[cfg(test)]
793fn write_stages_index_to(dir: &std::path::Path, stages: &[StageRecord]) -> anyhow::Result<()> {
794    let json = serde_json::to_string_pretty(&stages)
795        .expect("infallible: StageRecord slice always serializes to JSON");
796    write_private_atomic(&dir.join("stages.json"), &json)
797}
798
799/// Read the stages index for a run, or return an empty vec on any error.
800pub fn read_stages_index(run_id: &str) -> Vec<StageRecord> {
801    read_stages_index_from(&run_dir(run_id))
802}
803
804/// [`read_stages_index`] for a run directory the caller already holds.
805///
806/// Restart recovery works from its configured runs directory rather than the
807/// home one, so it cannot resolve the path itself.
808pub fn read_stages_index_from(dir: &std::path::Path) -> Vec<StageRecord> {
809    let json = match std::fs::read_to_string(dir.join("stages.json")) {
810        Ok(j) => j,
811        Err(_) => return Vec::new(),
812    };
813    serde_json::from_str(&json).unwrap_or_default()
814}
815
816/// Ensure the per-stage directory exists (called before first write).
817#[cfg(test)]
818fn ensure_stage_dir(run_id: &str, stage_idx: usize) {
819    let dir = stage_dir(run_id, stage_idx);
820    let _ = leviath_sys::create_private_dir_all(&dir);
821}
822
823/// Append a line of readable agent output to the per-stage output log.
824///
825/// Test-only; see [`write_context_snapshot`].
826#[cfg(test)]
827pub fn append_stage_output(run_id: &str, stage_idx: usize, text: &str) {
828    use std::io::Write;
829    ensure_stage_dir(run_id, stage_idx);
830    let path = stage_dir(run_id, stage_idx).join("output.log");
831    if let Ok(mut file) = leviath_sys::open_private_append(&path) {
832        let _ = writeln!(file, "{}", text);
833    }
834}
835
836/// Append a line of operational/tool-activity log to the per-stage logs file.
837///
838/// Test-only; see [`write_context_snapshot`].
839#[cfg(test)]
840pub fn append_stage_log(run_id: &str, stage_idx: usize, text: &str) {
841    use std::io::Write;
842    ensure_stage_dir(run_id, stage_idx);
843    let path = stage_dir(run_id, stage_idx).join("logs.log");
844    if let Ok(mut file) = leviath_sys::open_private_append(&path) {
845        let _ = writeln!(file, "{}", text);
846    }
847}
848
849/// Atomically write a context snapshot for a specific stage.
850///
851/// Test-only; see [`write_context_snapshot`].
852#[cfg(test)]
853pub fn write_stage_context(
854    run_id: &str,
855    stage_idx: usize,
856    snap: &ContextSnapshot,
857) -> anyhow::Result<()> {
858    ensure_stage_dir(run_id, stage_idx);
859    write_context_snapshot_to(&stage_dir(run_id, stage_idx), snap)
860}
861
862/// Read the context snapshot for a specific stage, if present.
863pub fn read_stage_context(run_id: &str, stage_idx: usize) -> Option<ContextSnapshot> {
864    let path = stage_dir(run_id, stage_idx).join("context.json");
865    let json = std::fs::read_to_string(&path).ok()?;
866    serde_json::from_str(&json).ok()
867}
868
869/// Read the last `max_bytes` of the readable output log for a specific stage.
870pub fn tail_stage_output(run_id: &str, stage_idx: usize, max_bytes: u64) -> String {
871    tail_file(&stage_dir(run_id, stage_idx).join("output.log"), max_bytes)
872}
873
874/// Read the last `max_bytes` of the operational log for a specific stage.
875pub fn tail_stage_log(run_id: &str, stage_idx: usize, max_bytes: u64) -> String {
876    tail_file(&stage_dir(run_id, stage_idx).join("logs.log"), max_bytes)
877}
878
879/// Which stage's logs to read.
880#[derive(Debug, Clone, Copy, PartialEq, Eq)]
881pub enum StageSelector {
882    /// The stage the run is on now - the last entry in `stages.json`. What a
883    /// caller tailing a live run wants, and what `agent_result` already picked.
884    Current,
885    /// One specific stage by index.
886    Index(usize),
887    /// Every stage, oldest first, with a separator between them.
888    All,
889}
890
891/// Which of a stage's two logs to read.
892#[derive(Debug, Clone, Copy, PartialEq, Eq)]
893pub enum LogStream {
894    /// `output.log` - the assistant's readable output.
895    Output,
896    /// `logs.log` - operational lines: `[tool] …`, `[Tokens: …]`, `[error] …`.
897    Operational,
898}
899
900/// Read a run's logs, choosing the stage and the stream.
901///
902/// Exists because there were two answers in the codebase to "where is a run's
903/// output", and one of them was wrong: `GET /api/agents/{id}/logs` read
904/// `<run_dir>/output.log`, which nothing has ever written, so it returned an
905/// empty string for every run there has ever been. The real logs are per-stage,
906/// under `stages/<idx>/`. Routing both that handler and `agent_result` through
907/// here leaves one answer.
908///
909/// Stages come from `stages.json` rather than a `read_dir` of `stages/`, because
910/// that index is the record of which stages exist and in what order - the
911/// directory is just where their bytes landed.
912///
913/// `max_bytes` applies to what is returned, so for [`StageSelector::All`] it
914/// bounds the joined text rather than each stage separately: "the last N bytes
915/// of what you asked for" holds whatever the selector was.
916pub fn tail_run_logs(
917    run_id: &str,
918    selector: StageSelector,
919    stream: LogStream,
920    max_bytes: u64,
921) -> String {
922    let read = |idx: usize| match stream {
923        LogStream::Output => tail_stage_output(run_id, idx, max_bytes),
924        LogStream::Operational => tail_stage_log(run_id, idx, max_bytes),
925    };
926    let stages = read_stages_index(run_id);
927    match selector {
928        StageSelector::Index(idx) => read(idx),
929        StageSelector::Current => match stages.len().checked_sub(1) {
930            Some(last) => read(last),
931            // No stages recorded yet. Fall back to the legacy run-level file:
932            // nothing writes it today, but a run whose stage dirs were pruned
933            // still reads honestly instead of claiming it produced nothing.
934            None => tail_file(&run_dir(run_id).join("output.log"), max_bytes),
935        },
936        StageSelector::All => {
937            let joined = stages
938                .iter()
939                .map(|stage| {
940                    format!(
941                        "===== stage {}: {} =====\n{}",
942                        stage.index,
943                        stage.name,
944                        read(stage.index)
945                    )
946                })
947                .collect::<Vec<_>>()
948                .join("\n");
949            // Re-bound the join: each part was capped individually, so their
950            // concatenation can exceed the cap the caller asked for.
951            let start = leviath_core::text::floor_char_boundary(
952                &joined,
953                joined.len().saturating_sub(max_bytes as usize),
954            );
955            joined.split_at(start).1.to_string()
956        }
957    }
958}
959
960/// Build the isolated base directory for a run-state test and create its
961/// `runs/` subdir. Returned so the caller's closure can plant fixtures under it.
962///
963/// Rooted under `~/.leviath-test/rs-<hash>` rather than `std::env::temp_dir()`:
964/// some dashboard render tests display a real on-disk path inside a fixed-width
965/// terminal area and assert on a substring near its *end*, and macOS's real
966/// temp dir (`/var/folders/xy/.../T/`) is long enough to push realistic paths
967/// past the render width and truncate the asserted suffix. `unique` is hashed
968/// short for the same reason (test names run 60+ chars). `.leviath-test` is a
969/// sibling of `.leviath`, never read by `lev dash`/`lev serve`, so even if a
970/// killed test process skips cleanup it can't leak into the real dashboard.
971#[cfg(test)]
972fn make_runs_base_dir(unique: &str) -> std::path::PathBuf {
973    use std::hash::{Hash, Hasher};
974    let mut hasher = std::collections::hash_map::DefaultHasher::new();
975    unique.hash(&mut hasher);
976    let short = format!("{:x}", hasher.finish() & 0xffff_ffff);
977    let base_dir = dirs::home_dir()
978        .unwrap_or_default()
979        .join(".leviath-test")
980        .join(format!("rs-{short}"));
981    let _ = std::fs::create_dir_all(base_dir.join("runs"));
982    base_dir
983}
984
985/// The env overrides that point run-state I/O at `base_dir` instead of the
986/// real `~/.leviath/`. Handed to `temp_env` for scoped set-and-restore.
987#[cfg(test)]
988fn runs_dir_isolation_vars(
989    base_dir: &std::path::Path,
990) -> [(&'static str, Option<std::ffi::OsString>); 2] {
991    [
992        (
993            "LEVIATH_RUNS_DIR",
994            Some(base_dir.join("runs").into_os_string()),
995        ),
996        (
997            "LEVIATH_DASHBOARD_LOG_PATH",
998            Some(base_dir.join("dashboard.log").into_os_string()),
999        ),
1000    ]
1001}
1002
1003/// Runs `f` with `LEVIATH_RUNS_DIR`/`LEVIATH_DASHBOARD_LOG_PATH` pointed at a
1004/// fresh isolated temp directory (passed to `f`), restoring them afterwards.
1005/// Closure-scoped (not an RAII guard) because edition 2024 makes `set_var`
1006/// `unsafe`, which the crate forbids; `temp_env` serializes it process-wide.
1007#[cfg(test)]
1008pub(crate) fn with_isolated_runs_dir<R>(unique: &str, f: impl FnOnce(&std::path::Path) -> R) -> R {
1009    let base_dir = make_runs_base_dir(unique);
1010    let result = temp_env::with_vars(runs_dir_isolation_vars(&base_dir), || f(&base_dir));
1011    let _ = std::fs::remove_dir_all(&base_dir);
1012    result
1013}
1014
1015/// Async counterpart of [`with_isolated_runs_dir`] for `#[tokio::test]`s.
1016#[cfg(test)]
1017pub(crate) async fn with_isolated_runs_dir_async<R, Fut>(
1018    unique: &str,
1019    f: impl FnOnce(std::path::PathBuf) -> Fut,
1020) -> R
1021where
1022    Fut: std::future::Future<Output = R>,
1023{
1024    let base_dir = make_runs_base_dir(unique);
1025    let result =
1026        temp_env::async_with_vars(runs_dir_isolation_vars(&base_dir), f(base_dir.clone())).await;
1027    let _ = std::fs::remove_dir_all(&base_dir);
1028    result
1029}
1030
1031#[cfg(test)]
1032mod tests {
1033    use super::*;
1034
1035    /// `run_id` arrives from URL segments on `GET /api/agents/{id}/logs` and
1036    /// friends. `Path::join` neither normalizes `..` nor resists an absolute
1037    /// path, so an unvalidated id read files anywhere. An unsafe one resolves to
1038    /// a name that cannot exist, giving the caller a plain miss.
1039    #[test]
1040    fn run_dir_refuses_an_unsafe_run_id() {
1041        crate::test_support::with_tracing(|| {
1042            for bad in ["../../etc", "/etc/passwd", "..", "a/b"] {
1043                let dir = run_dir(bad);
1044                let shown = dir.display().to_string();
1045                assert!(dir.ends_with("<invalid>"), "{bad} resolved to {shown}");
1046                assert!(!dir.exists(), "{bad} must not resolve to a real path");
1047            }
1048            // An ordinary id is untouched.
1049            assert!(run_dir("run-abc123").ends_with("run-abc123"));
1050        });
1051    }
1052
1053    // ─── looks_abandoned ────────────────────────────────────────────────────
1054
1055    /// A run claiming to be live on disk, last moved at 1000.
1056    fn live_on_disk(run_id: &str) -> RunMeta {
1057        let mut meta = RunMeta::new(
1058            run_id.to_string(),
1059            "coder".to_string(),
1060            "/agents/coder".to_string(),
1061            "t".to_string(),
1062            None,
1063            "/w".to_string(),
1064            1,
1065        );
1066        meta.status = RunStatus::Running;
1067        meta.updated_at = 1_000;
1068        meta.last_progress_at = Some(1_000);
1069        meta
1070    }
1071
1072    fn held(ids: &[&str]) -> std::collections::HashSet<String> {
1073        ids.iter().map(|s| (*s).to_string()).collect()
1074    }
1075
1076    /// The shape issue #202 reported: disk says running, the daemon is not
1077    /// hosting it, and it has not moved in a long time.
1078    #[test]
1079    fn a_run_nothing_is_driving_looks_abandoned() {
1080        let meta = live_on_disk("r1");
1081        assert!(looks_abandoned(
1082            &meta,
1083            Some(&held(&["other"])),
1084            1_000 + STALE_AFTER_SECS + 1
1085        ));
1086    }
1087
1088    /// The arm that decides whether a reconciler is safe to run at all. A daemon
1089    /// that is restarting gives no answer, which looks exactly like every run
1090    /// dying at once; anything that acted on it would cancel a whole factory.
1091    #[test]
1092    fn no_answer_from_the_daemon_condemns_nothing() {
1093        let meta = live_on_disk("r1");
1094        assert!(!looks_abandoned(
1095            &meta,
1096            None,
1097            1_000 + STALE_AFTER_SECS * 100
1098        ));
1099    }
1100
1101    #[test]
1102    fn a_run_the_daemon_is_hosting_is_never_abandoned() {
1103        let meta = live_on_disk("r1");
1104        assert!(!looks_abandoned(
1105            &meta,
1106            Some(&held(&["r1"])),
1107            1_000 + STALE_AFTER_SECS * 100
1108        ));
1109    }
1110
1111    /// A run parked on a long inference has not moved and is still working, so
1112    /// the window has to be wider than the persistence heartbeat.
1113    #[test]
1114    fn a_slow_run_inside_the_window_is_left_alone() {
1115        let meta = live_on_disk("r1");
1116        assert!(!looks_abandoned(
1117            &meta,
1118            Some(&held(&[])),
1119            1_000 + STALE_AFTER_SECS - 1
1120        ));
1121    }
1122
1123    /// A finished run is not abandoned, it is done. The daemon unloads it within
1124    /// seconds of it going terminal, so it is absent from the live set for the
1125    /// rest of time and would otherwise trip every other check here.
1126    #[test]
1127    fn a_finished_run_is_not_abandoned() {
1128        for status in [
1129            RunStatus::Complete,
1130            RunStatus::CompleteInteractive,
1131            RunStatus::Error,
1132            RunStatus::Cancelled,
1133        ] {
1134            let mut meta = live_on_disk("r1");
1135            meta.status = status.clone();
1136            assert!(
1137                !looks_abandoned(&meta, Some(&held(&[])), 1_000 + STALE_AFTER_SECS * 100),
1138                "{status} is finished, not abandoned"
1139            );
1140        }
1141    }
1142
1143    /// The progress stamp wins over the heartbeat. A wedged run keeps rewriting
1144    /// `updated_at` every 30 seconds, so judging on it would never age anything
1145    /// out, which is the reason issue #202 could not be fixed from meta.json
1146    /// before the stamp existed.
1147    #[test]
1148    fn a_fresh_heartbeat_does_not_rescue_a_run_that_stopped_moving() {
1149        let mut meta = live_on_disk("r1");
1150        let now = 1_000 + STALE_AFTER_SECS * 10;
1151        meta.updated_at = now; // the heartbeat, still beating
1152        meta.last_progress_at = Some(1_000); // but nothing has moved since 1000
1153        assert!(looks_abandoned(&meta, Some(&held(&[])), now));
1154    }
1155
1156    /// A run written before the stamp existed falls back to `updated_at`, so old
1157    /// runs keep the older, weaker behavior instead of all reading as stale.
1158    #[test]
1159    fn a_run_without_the_stamp_falls_back_to_updated_at() {
1160        let mut meta = live_on_disk("r1");
1161        meta.last_progress_at = None;
1162        meta.updated_at = 1_000;
1163        assert!(looks_abandoned(
1164            &meta,
1165            Some(&held(&[])),
1166            1_000 + STALE_AFTER_SECS + 1
1167        ));
1168        meta.updated_at = 1_000 + STALE_AFTER_SECS;
1169        assert!(!looks_abandoned(
1170            &meta,
1171            Some(&held(&[])),
1172            1_000 + STALE_AFTER_SECS + 1
1173        ));
1174    }
1175
1176    #[test]
1177    fn write_json_atomic_fs_write_failure() {
1178        // Drive the `std::fs::write(&tmp, json)?` error arm: writing the
1179        // `.json.tmp` sibling into a directory that does not exist fails.
1180        let path = std::path::Path::new("/nonexistent/leviath/runstate-cov/out.json");
1181        let result = write_private_atomic(path, "{}");
1182        assert!(result.is_err());
1183        assert!(!path.exists());
1184    }
1185
1186    // ─── RunStatus ──────────────────────────────────────────────────────────
1187
1188    #[test]
1189    fn run_status_serde_roundtrip() {
1190        for status in [
1191            RunStatus::Starting,
1192            RunStatus::Running,
1193            RunStatus::WaitingInput,
1194            RunStatus::Complete,
1195            RunStatus::CompleteInteractive,
1196            RunStatus::Paused,
1197            RunStatus::Error,
1198            RunStatus::Cancelled,
1199        ] {
1200            let json = serde_json::to_string(&status).unwrap();
1201            let back: RunStatus = serde_json::from_str(&json).unwrap();
1202            assert_eq!(status, back);
1203        }
1204    }
1205
1206    #[test]
1207    fn run_status_display() {
1208        assert_eq!(RunStatus::Starting.to_string(), "Starting");
1209        assert_eq!(RunStatus::Running.to_string(), "Running");
1210        assert_eq!(RunStatus::WaitingInput.to_string(), "WaitingInput");
1211        assert_eq!(RunStatus::Complete.to_string(), "Complete");
1212        assert_eq!(
1213            RunStatus::CompleteInteractive.to_string(),
1214            "CompleteInteractive"
1215        );
1216        assert_eq!(RunStatus::Paused.to_string(), "Paused");
1217        assert_eq!(RunStatus::Error.to_string(), "Error");
1218        assert_eq!(RunStatus::Cancelled.to_string(), "Cancelled");
1219    }
1220
1221    #[test]
1222    fn run_status_snake_case_serialization() {
1223        let json = serde_json::to_string(&RunStatus::WaitingInput).unwrap();
1224        assert_eq!(json, "\"waiting_input\"");
1225        let json = serde_json::to_string(&RunStatus::CompleteInteractive).unwrap();
1226        assert_eq!(json, "\"complete_interactive\"");
1227    }
1228
1229    // ─── StageRunStatus ─────────────────────────────────────────────────────
1230
1231    #[test]
1232    fn stage_run_status_serde_roundtrip() {
1233        for status in [
1234            StageRunStatus::Pending,
1235            StageRunStatus::Active,
1236            StageRunStatus::WaitingInput,
1237            StageRunStatus::Complete,
1238            StageRunStatus::Error,
1239        ] {
1240            let json = serde_json::to_string(&status).unwrap();
1241            let back: StageRunStatus = serde_json::from_str(&json).unwrap();
1242            assert_eq!(status, back);
1243        }
1244    }
1245
1246    #[test]
1247    fn stage_run_status_display() {
1248        assert_eq!(StageRunStatus::Pending.to_string(), "Pending");
1249        assert_eq!(StageRunStatus::Active.to_string(), "Active");
1250        assert_eq!(StageRunStatus::WaitingInput.to_string(), "WaitingInput");
1251        assert_eq!(StageRunStatus::Complete.to_string(), "Complete");
1252        assert_eq!(StageRunStatus::Error.to_string(), "Error");
1253    }
1254
1255    // ─── RunMeta ────────────────────────────────────────────────────────────
1256
1257    #[test]
1258    fn run_meta_new_defaults() {
1259        let meta = RunMeta::new(
1260            "run-1".into(),
1261            "agent".into(),
1262            "/path".into(),
1263            "do stuff".into(),
1264            Some("gpt-4".into()),
1265            "/work".into(),
1266            3,
1267        );
1268        assert_eq!(meta.run_id, "run-1");
1269        assert_eq!(meta.agent_name, "agent");
1270        assert_eq!(meta.task, "do stuff");
1271        assert_eq!(meta.model.as_deref(), Some("gpt-4"));
1272        assert_eq!(meta.num_stages, 3);
1273        assert_eq!(meta.status, RunStatus::Starting);
1274        assert_eq!(meta.pid, 0);
1275        assert_eq!(meta.stage_index, 0);
1276        assert!(meta.error.is_none());
1277        assert!(meta.title.is_none());
1278        assert!(meta.metadata.is_empty());
1279        assert!(meta.callback_url.is_none());
1280        assert!(meta.parent_run_id.is_none());
1281    }
1282
1283    #[test]
1284    fn run_meta_serde_roundtrip() {
1285        let meta = RunMeta::new(
1286            "test-run".into(),
1287            "test-agent".into(),
1288            "/agents/test".into(),
1289            "run tests".into(),
1290            None,
1291            "/tmp".into(),
1292            2,
1293        );
1294        let json = serde_json::to_string_pretty(&meta).unwrap();
1295        let back: RunMeta = serde_json::from_str(&json).unwrap();
1296        assert_eq!(back.run_id, "test-run");
1297        assert_eq!(back.agent_name, "test-agent");
1298        assert_eq!(back.num_stages, 2);
1299        assert!(back.model.is_none());
1300    }
1301
1302    #[test]
1303    fn run_meta_touch_updates_timestamp() {
1304        let mut meta = RunMeta::new(
1305            "r".into(),
1306            "a".into(),
1307            "/p".into(),
1308            "t".into(),
1309            None,
1310            "/w".into(),
1311            1,
1312        );
1313        let before = meta.updated_at;
1314        // Touch should update (or at least not decrease) updated_at
1315        meta.touch();
1316        assert!(meta.updated_at >= before);
1317    }
1318
1319    #[test]
1320    fn run_meta_optional_fields_deserialize() {
1321        // Simulate a meta.json without optional fields (e.g., from older version)
1322        let json = serde_json::json!({
1323            "run_id": "r1",
1324            "agent_name": "a",
1325            "agent_path": "/p",
1326            "task": "t",
1327            "model": null,
1328            "pid": 123,
1329            "status": "running",
1330            "current_stage": "init",
1331            "stage_index": 0,
1332            "num_stages": 1,
1333            "iteration": 0,
1334            "prompt_tokens": 0,
1335            "completion_tokens": 0,
1336            "workdir": "/w",
1337            "started_at": 1000,
1338            "updated_at": 1000,
1339            "error": null
1340        });
1341        let meta: RunMeta = serde_json::from_value(json).unwrap();
1342        assert_eq!(meta.cached_tokens, 0);
1343        assert!(meta.title.is_none());
1344        assert!(meta.metadata.is_empty());
1345        assert!(meta.callback_url.is_none());
1346        assert!(meta.parent_run_id.is_none());
1347        // A run written before the progress stamp existed has no answer, which is
1348        // why the field is an Option: `Some(0)` would read as "last moved in 1970"
1349        // and invite a reconciler to declare it abandoned.
1350        assert!(meta.last_progress_at.is_none());
1351    }
1352
1353    /// `pid` is written by every daemon there has ever been, and is always 0 in
1354    /// the shared world. A file that omits it entirely must still load, so the
1355    /// field can be dropped in a future major without stranding old runs.
1356    #[test]
1357    fn run_meta_without_a_pid_still_loads() {
1358        let json = serde_json::json!({
1359            "run_id": "r1",
1360            "agent_name": "a",
1361            "agent_path": "/p",
1362            "task": "t",
1363            "model": null,
1364            "status": "running",
1365            "current_stage": "init",
1366            "stage_index": 0,
1367            "num_stages": 1,
1368            "iteration": 0,
1369            "prompt_tokens": 0,
1370            "completion_tokens": 0,
1371            "workdir": "/w",
1372            "started_at": 1000,
1373            "updated_at": 1000,
1374            "error": null
1375        });
1376        let meta: RunMeta = serde_json::from_value(json).unwrap();
1377        assert_eq!(meta.pid, 0);
1378    }
1379
1380    // ─── StageRecord ────────────────────────────────────────────────────────
1381
1382    #[test]
1383    fn stage_record_new_defaults() {
1384        let rec = StageRecord::new("analyze".into(), 2);
1385        assert_eq!(rec.name, "analyze");
1386        assert_eq!(rec.index, 2);
1387        assert_eq!(rec.status, StageRunStatus::Pending);
1388        assert_eq!(rec.prompt_tokens, 0);
1389        assert_eq!(rec.completion_tokens, 0);
1390        assert_eq!(rec.cached_tokens, 0);
1391        assert!(rec.started_at.is_none());
1392        assert!(rec.ended_at.is_none());
1393    }
1394
1395    #[test]
1396    fn stage_record_serde_roundtrip() {
1397        let mut rec = StageRecord::new("build".into(), 0);
1398        rec.status = StageRunStatus::Complete;
1399        rec.prompt_tokens = 100;
1400        rec.started_at = Some(1000);
1401        rec.ended_at = Some(2000);
1402
1403        let json = serde_json::to_string(&rec).unwrap();
1404        let back: StageRecord = serde_json::from_str(&json).unwrap();
1405        assert_eq!(back.name, "build");
1406        assert_eq!(back.status, StageRunStatus::Complete);
1407        assert_eq!(back.prompt_tokens, 100);
1408        assert_eq!(back.started_at, Some(1000));
1409    }
1410
1411    // ─── RegionSnapshot / ContextSnapshot ───────────────────────────────────
1412
1413    #[test]
1414    fn region_snapshot_serde_roundtrip() {
1415        let snap = RegionSnapshot {
1416            name: "system".into(),
1417            kind: "pinned".into(),
1418            current_tokens: 100,
1419            max_tokens: 500,
1420            entries: vec![RegionEntrySnapshot {
1421                content: "You are helpful".into(),
1422                tokens: 3,
1423                kind: Default::default(),
1424                metadata: None,
1425                key: None,
1426                taint: Default::default(),
1427            }],
1428        };
1429        let json = serde_json::to_string(&snap).unwrap();
1430        let back: RegionSnapshot = serde_json::from_str(&json).unwrap();
1431        assert_eq!(back.name, "system");
1432        assert_eq!(back.entries.len(), 1);
1433        assert_eq!(back.entries[0].content, "You are helpful");
1434    }
1435
1436    #[test]
1437    fn region_snapshot_empty_entries_omitted() {
1438        let snap = RegionSnapshot {
1439            name: "empty".into(),
1440            kind: "temporary".into(),
1441            current_tokens: 0,
1442            max_tokens: 100,
1443            entries: vec![],
1444        };
1445        let json = serde_json::to_value(&snap).unwrap();
1446        assert!(json.get("entries").is_none());
1447    }
1448
1449    #[test]
1450    fn context_snapshot_serde_roundtrip() {
1451        let snap = ContextSnapshot {
1452            stage_name: "analyze".into(),
1453            total_tokens: 500,
1454            max_tokens: 8192,
1455            regions: vec![RegionSnapshot {
1456                name: "history".into(),
1457                kind: "sliding".into(),
1458                current_tokens: 300,
1459                max_tokens: 2000,
1460                entries: vec![],
1461            }],
1462        };
1463        let json = serde_json::to_string(&snap).unwrap();
1464        let back: ContextSnapshot = serde_json::from_str(&json).unwrap();
1465        assert_eq!(back.stage_name, "analyze");
1466        assert_eq!(back.total_tokens, 500);
1467        assert_eq!(back.regions.len(), 1);
1468    }
1469
1470    // ─── tail_file ──────────────────────────────────────────────────────────
1471
1472    #[test]
1473    fn tail_file_nonexistent_returns_empty() {
1474        let path = std::path::Path::new("/tmp/nonexistent-leviath-test-file.txt");
1475        assert_eq!(tail_file(path, 1024), "");
1476    }
1477
1478    #[test]
1479    fn tail_file_small_file_returns_all() {
1480        let dir = tempfile::tempdir().unwrap();
1481        let path = dir.path().join("small.txt");
1482        std::fs::write(&path, "line1\nline2\nline3\n").unwrap();
1483        let result = tail_file(&path, 1024);
1484        assert_eq!(result, "line1\nline2\nline3\n");
1485    }
1486
1487    #[test]
1488    fn tail_file_large_file_returns_tail() {
1489        let dir = tempfile::tempdir().unwrap();
1490        let path = dir.path().join("large.txt");
1491        let content = "abcdefghij\n".repeat(100); // 1100 bytes
1492        std::fs::write(&path, &content).unwrap();
1493        let result = tail_file(&path, 50);
1494        // Should be less than 50 bytes, starting from a line boundary
1495        assert!(result.len() <= 50);
1496        assert!(result.ends_with('\n'));
1497    }
1498
1499    // ─── read_final_output ──────────────────────────────────────────────────
1500
1501    /// The descriptor in `meta.json` and the sidecar beside it have to agree.
1502    /// Each way they can disagree reads as "no answer", which is the only safe
1503    /// reading: half an answer is worse than none.
1504    #[test]
1505    fn read_final_output_needs_both_the_descriptor_and_the_sidecar() {
1506        with_isolated_runs_dir("read-final-output", |_| {
1507            // No run at all.
1508            assert!(read_final_output("no-such-run").is_none());
1509
1510            // A run with no answer recorded.
1511            let meta = RunMeta::new(
1512                "run-silent".to_string(),
1513                "a".to_string(),
1514                "/p".to_string(),
1515                "t".to_string(),
1516                None,
1517                "/w".to_string(),
1518                1,
1519            );
1520            create_run(&meta).expect("run dir");
1521            assert!(read_final_output("run-silent").is_none());
1522
1523            // A descriptor saying there is one, with the sidecar missing: a run
1524            // written by a build that stored the answer inline, or one whose
1525            // directory was pruned.
1526            let answer = leviath_core::output::FinalOutput::new(
1527                "the answer",
1528                Some("markdown".to_string()),
1529                "present".to_string(),
1530                42,
1531            );
1532            let mut claimed = RunMeta::new(
1533                "run-claimed".to_string(),
1534                "a".to_string(),
1535                "/p".to_string(),
1536                "t".to_string(),
1537                None,
1538                "/w".to_string(),
1539                1,
1540            );
1541            claimed.final_output = Some(answer.descriptor());
1542            create_run(&claimed).expect("run dir");
1543            assert!(read_final_output("run-claimed").is_none());
1544
1545            // And both together: the answer comes back whole.
1546            write_final_output(&run_dir("run-claimed"), &answer.content).expect("sidecar");
1547            let read = read_final_output("run-claimed").expect("both halves are there");
1548            assert_eq!(read.content, "the answer");
1549            assert_eq!(read.format.as_deref(), Some("markdown"));
1550            assert_eq!(read.stage, "present");
1551        });
1552    }
1553
1554    // ─── new_run_id ─────────────────────────────────────────────────────────
1555
1556    #[test]
1557    fn new_run_id_contains_agent_name() {
1558        let id = new_run_id("my-agent");
1559        assert!(id.starts_with("my-agent-"));
1560    }
1561
1562    #[test]
1563    fn new_run_id_sanitizes_special_chars() {
1564        let id = new_run_id("agent with spaces!");
1565        assert!(!id.contains(' '));
1566        assert!(!id.contains('!'));
1567    }
1568
1569    /// The id becomes a directory name, and every reader resolves it through
1570    /// `is_safe_path_component`. A minted id that fails that check spawns a run
1571    /// the CLI can never read back, so the two rules have to agree whatever the
1572    /// blueprint calls itself.
1573    #[test]
1574    fn every_minted_run_id_is_a_safe_path_component() {
1575        for name in [
1576            "café",
1577            "日本語",
1578            "agent with spaces!",
1579            "../escape",
1580            "a/b",
1581            "..",
1582            "",
1583            "emoji-🚀-agent",
1584            "Ünïcödé",
1585        ] {
1586            let id = new_run_id(name);
1587            assert!(
1588                leviath_core::is_safe_path_component(&id),
1589                "agent {name:?} minted {id:?}, which run_dir resolves to <invalid>"
1590            );
1591        }
1592    }
1593
1594    #[test]
1595    fn new_run_id_is_unique_across_rapid_calls_in_same_second() {
1596        // `--count N` calls `new_run_id` N times in a tight loop, all within the
1597        // same wall-clock second.
1598        let ids: std::collections::HashSet<String> =
1599            (0..100).map(|_| new_run_id("same-agent")).collect();
1600        assert_eq!(ids.len(), 100);
1601    }
1602
1603    /// Split `<name>-<secs>-<hex>` from the right - the agent name itself may
1604    /// contain dashes.
1605    fn split_run_id(id: &str) -> (&str, &str) {
1606        let mut parts = id.rsplitn(3, '-');
1607        let suffix = parts.next().expect("run id has a suffix");
1608        let secs = parts.next().expect("run id has a timestamp");
1609        (secs, suffix)
1610    }
1611
1612    #[test]
1613    fn new_run_id_suffix_is_random_not_derived_from_the_clock() {
1614        // The collision this guards against is *across processes*: a suffix
1615        // derived as `(now ^ (now >> 16) ^ counter)` over a process-local
1616        // counter that every new process starts at 0 degenerates to a pure
1617        // function of the current second. Three concurrent `lev run`
1618        // invocations all mint `fetcher-1785127214-8b48` and silently share
1619        // one run directory. A fresh process has no state to vary, so the
1620        // property that has to hold is: IDs that share a timestamp still differ.
1621        let ids: Vec<String> = (0..200).map(|_| new_run_id("same-agent")).collect();
1622        let mut by_second: std::collections::HashMap<&str, Vec<&str>> =
1623            std::collections::HashMap::new();
1624        for id in &ids {
1625            let (secs, suffix) = split_run_id(id);
1626            by_second.entry(secs).or_default().push(suffix);
1627        }
1628        let mut largest = 0;
1629        for (secs, suffixes) in &by_second {
1630            let distinct: std::collections::HashSet<&&str> = suffixes.iter().collect();
1631            assert_eq!(
1632                distinct.len(),
1633                suffixes.len(),
1634                "two runs in second {secs} share a suffix: {suffixes:?}"
1635            );
1636            largest = largest.max(suffixes.len());
1637        }
1638        // 200 calls take microseconds, so they cannot all land in distinct
1639        // seconds - without this the assertion above would be vacuous.
1640        assert!(
1641            largest > 1,
1642            "expected IDs sharing a second, got {by_second:?}"
1643        );
1644    }
1645
1646    // ─── write_meta / read_meta roundtrip ───────────────────────────────────
1647
1648    #[test]
1649    fn write_and_read_meta_roundtrip() {
1650        // Isolated via `isolate_runs_dir_for_test` so write_meta/read_meta
1651        // never touch the real ~/.leviath/runs/ - the temp dir is removed
1652        // automatically when `_guard` drops, so no manual cleanup needed.
1653        with_isolated_runs_dir("write-and-read-meta-roundtrip", |_d| {
1654            let meta = RunMeta::new(
1655                "test-roundtrip-unit".into(),
1656                "test-agent".into(),
1657                "/agents/test".into(),
1658                "unit test".into(),
1659                Some("model-x".into()),
1660                "/tmp".into(),
1661                2,
1662            );
1663
1664            create_run(&meta).unwrap();
1665            let back = read_meta(&meta.run_id).unwrap();
1666            assert_eq!(back.run_id, "test-roundtrip-unit");
1667            assert_eq!(back.agent_name, "test-agent");
1668            assert_eq!(back.task, "unit test");
1669            assert_eq!(back.model.as_deref(), Some("model-x"));
1670        });
1671    }
1672
1673    #[test]
1674    fn read_meta_returns_err_on_corrupted_json() {
1675        // Exercises `read_meta_from`'s `serde_json::from_str(&json)?` Err
1676        // arm: a `meta.json` that exists but doesn't parse as a `RunMeta`.
1677        with_isolated_runs_dir("read-meta-returns-err-on-corrupted-json", |_d| {
1678            let run_id = "corrupted-meta-run";
1679            let dir = run_dir(run_id);
1680            std::fs::create_dir_all(&dir).unwrap();
1681            std::fs::write(dir.join("meta.json"), "not valid json").unwrap();
1682
1683            let result = read_meta(run_id);
1684            assert!(result.is_err());
1685        });
1686    }
1687
1688    // ─── write_stages_index / read_stages_index roundtrip ───────────────────
1689
1690    #[test]
1691    fn write_and_read_stages_index_roundtrip() {
1692        with_isolated_runs_dir("write-and-read-stages-index-roundtrip", |_d| {
1693            let run_id = "test-stages-idx-unit";
1694            let dir = run_dir(run_id);
1695            std::fs::create_dir_all(&dir).unwrap();
1696
1697            let stages = vec![
1698                StageRecord::new("init".into(), 0),
1699                StageRecord::new("process".into(), 1),
1700            ];
1701            write_stages_index(run_id, &stages).unwrap();
1702            let back = read_stages_index(run_id);
1703            assert_eq!(back.len(), 2);
1704            assert_eq!(back[0].name, "init");
1705            assert_eq!(back[1].name, "process");
1706        });
1707    }
1708
1709    #[test]
1710    fn read_stages_index_missing_returns_empty() {
1711        let back = read_stages_index("nonexistent-run-12345");
1712        assert!(back.is_empty());
1713    }
1714
1715    // ─── write/read context snapshot ────────────────────────────────────────
1716
1717    #[test]
1718    fn write_and_read_context_snapshot_roundtrip() {
1719        with_isolated_runs_dir("write-and-read-context-snapshot-roundtrip", |_d| {
1720            let run_id = "test-ctx-snap-unit";
1721            let dir = run_dir(run_id);
1722            std::fs::create_dir_all(&dir).unwrap();
1723
1724            let snap = ContextSnapshot {
1725                stage_name: "test".into(),
1726                total_tokens: 42,
1727                max_tokens: 8192,
1728                regions: vec![],
1729            };
1730            write_context_snapshot(run_id, &snap).unwrap();
1731            let back = read_context_snapshot(run_id).unwrap();
1732            assert_eq!(back.stage_name, "test");
1733            assert_eq!(back.total_tokens, 42);
1734        });
1735    }
1736
1737    #[test]
1738    fn read_context_snapshot_missing_returns_none() {
1739        assert!(read_context_snapshot("nonexistent-ctx-run").is_none());
1740    }
1741
1742    #[test]
1743    fn read_run_archive_roundtrips_and_context_history_replays() {
1744        with_isolated_runs_dir("read-run-archive-roundtrip", |_d| {
1745            use leviath_core::run_archive::{self, RunIdentity, RunRecord};
1746            let run_id = "archive-unit";
1747            std::fs::create_dir_all(run_dir(run_id)).unwrap();
1748            let mut buf = Vec::new();
1749            run_archive::write_archive_start(&mut buf, run_archive::RUN_ARCHIVE_VERSION).unwrap();
1750            let meta = RunMeta::new(
1751                run_id.to_string(),
1752                "a".to_string(),
1753                "/p".to_string(),
1754                "t".to_string(),
1755                None,
1756                "/w".to_string(),
1757                1,
1758            );
1759            run_archive::write_record(
1760                &mut buf,
1761                &RunRecord::Header {
1762                    identity: RunIdentity {
1763                        run_id: run_id.to_string(),
1764                        machine_id: "m".to_string(),
1765                        world_id: "w".to_string(),
1766                        created_at: 0,
1767                    },
1768                    meta: Box::new(meta),
1769                },
1770            )
1771            .unwrap();
1772            run_archive::write_record(
1773                &mut buf,
1774                &RunRecord::ContextCheckpoint {
1775                    snapshot: ContextSnapshot {
1776                        stage_name: "plan".to_string(),
1777                        total_tokens: 3,
1778                        max_tokens: 100,
1779                        regions: vec![],
1780                    },
1781                    at: 1,
1782                },
1783            )
1784            .unwrap();
1785            std::fs::write(run_dir(run_id).join("run.lvr"), &buf).unwrap();
1786
1787            let records = read_run_archive(run_id).expect("archive read");
1788            assert_eq!(records.len(), 2);
1789            let history = context_history(run_id);
1790            assert_eq!(history.len(), 1);
1791            assert_eq!(history[0].context.stage_name, "plan");
1792
1793            // The streaming visitors see the same journal without ever
1794            // materializing it.
1795            let mut streamed_points = Vec::new();
1796            visit_run_archive(run_id, &mut |p| {
1797                streamed_points.push((p.index, p.context.stage_name.to_string()));
1798                std::ops::ControlFlow::Continue(())
1799            })
1800            .expect("streamed replay");
1801            assert_eq!(streamed_points, vec![(0, "plan".to_string())]);
1802
1803            let mut streamed_records = 0usize;
1804            visit_run_records(run_id, &mut |_| {
1805                streamed_records += 1;
1806                std::ops::ControlFlow::Continue(())
1807            })
1808            .expect("streamed records");
1809            assert_eq!(streamed_records, 2);
1810
1811            // And a visitor can stop early.
1812            let mut first_only = 0usize;
1813            visit_run_records(run_id, &mut |_| {
1814                first_only += 1;
1815                std::ops::ControlFlow::Break(())
1816            })
1817            .expect("streamed records with break");
1818            assert_eq!(first_only, 1);
1819        });
1820    }
1821
1822    /// The stat cache's contract: parse once, serve from cache while the stat
1823    /// is unchanged, re-parse on change, cache negative results, and forget
1824    /// files that disappear.
1825    #[test]
1826    fn stat_cache_parses_once_per_stat_change() {
1827        let dir = tempfile::tempdir().unwrap();
1828        let path = dir.path().join("value.json");
1829        std::fs::write(&path, "41").unwrap();
1830        let mut cache: StatCache<i64> = StatCache::default();
1831        let mut parses = 0;
1832        let get = |cache: &mut StatCache<i64>, path: &std::path::Path, parses: &mut usize| {
1833            cache
1834                .get_with(path, |text| {
1835                    *parses += 1;
1836                    text.trim().parse().ok()
1837                })
1838                .map(|v| *v)
1839        };
1840
1841        assert_eq!(get(&mut cache, &path, &mut parses), Some(41));
1842        assert_eq!(get(&mut cache, &path, &mut parses), Some(41));
1843        assert_eq!(parses, 1, "the second read came from the cache");
1844
1845        // A same-length rewrite with a fresh mtime re-parses (the atomic-rename
1846        // writer always produces a new inode+mtime; simulate with a bumped
1847        // mtime via a rewrite of different content and length).
1848        std::fs::write(&path, "1234").unwrap();
1849        assert_eq!(get(&mut cache, &path, &mut parses), Some(1234));
1850        assert_eq!(parses, 2);
1851
1852        // Unparseable content is cached as a miss - one parse attempt, then
1853        // stat-only until the file changes again.
1854        std::fs::write(&path, "not a number").unwrap();
1855        assert_eq!(get(&mut cache, &path, &mut parses), None);
1856        assert_eq!(get(&mut cache, &path, &mut parses), None);
1857        assert_eq!(parses, 3, "the bad file was parsed once, not per tick");
1858
1859        // A deleted file is a miss and its entry is dropped.
1860        std::fs::remove_file(&path).unwrap();
1861        assert_eq!(get(&mut cache, &path, &mut parses), None);
1862        assert_eq!(parses, 3);
1863    }
1864
1865    #[test]
1866    fn stat_cache_retain_under_drops_dead_runs() {
1867        let dir = tempfile::tempdir().unwrap();
1868        let live = dir.path().join("live");
1869        let dead = dir.path().join("dead");
1870        std::fs::create_dir_all(&live).unwrap();
1871        std::fs::create_dir_all(&dead).unwrap();
1872        std::fs::write(live.join("meta.json"), "1").unwrap();
1873        std::fs::write(dead.join("meta.json"), "2").unwrap();
1874        let mut cache: StatCache<i64> = StatCache::default();
1875        cache.get_with(&live.join("meta.json"), |t| t.trim().parse().ok());
1876        cache.get_with(&dead.join("meta.json"), |t| t.trim().parse().ok());
1877        assert_eq!(cache.entries.len(), 2);
1878
1879        let keep: std::collections::HashSet<PathBuf> = [live.clone()].into_iter().collect();
1880        cache.retain_under(&keep);
1881        assert_eq!(cache.entries.len(), 1);
1882        assert!(cache.entries.contains_key(&live.join("meta.json")));
1883    }
1884
1885    /// The cached listing and per-run readers agree with their uncached
1886    /// counterparts, and serve repeat calls without re-parsing.
1887    #[test]
1888    fn cached_run_readers_match_the_uncached_ones() {
1889        with_isolated_runs_dir("cached-run-readers", |_d| {
1890            let meta = RunMeta::new(
1891                "cached-run".to_string(),
1892                "agent".to_string(),
1893                "/p".to_string(),
1894                "t".to_string(),
1895                None,
1896                "/w".to_string(),
1897                2,
1898            );
1899            create_run(&meta).unwrap();
1900            write_stages_index(
1901                "cached-run",
1902                &[leviath_core::run_meta::StageRecord::new(
1903                    "plan".to_string(),
1904                    0,
1905                )],
1906            )
1907            .unwrap();
1908            write_context_snapshot(
1909                "cached-run",
1910                &ContextSnapshot {
1911                    stage_name: "plan".to_string(),
1912                    total_tokens: 3,
1913                    max_tokens: 100,
1914                    regions: vec![],
1915                },
1916            )
1917            .unwrap();
1918
1919            let mut metas = StatCache::default();
1920            let mut stages = StatCache::default();
1921            let mut contexts = StatCache::default();
1922
1923            let listed = list_runs_cached(&mut metas);
1924            assert_eq!(listed.len(), 1);
1925            assert_eq!(listed[0].run_id, list_runs()[0].run_id);
1926
1927            let cached_stages = read_stages_index_cached("cached-run", &mut stages);
1928            let plain_stages = read_stages_index("cached-run");
1929            assert_eq!(cached_stages.len(), plain_stages.len());
1930            assert_eq!(cached_stages[0].name, plain_stages[0].name);
1931            let cached_ctx =
1932                read_context_snapshot_cached("cached-run", &mut contexts).expect("snapshot cached");
1933            assert_eq!(
1934                *cached_ctx,
1935                read_context_snapshot("cached-run").expect("snapshot read")
1936            );
1937            // A repeat serves the SAME Arc - the whole point of the cache.
1938            let again = read_context_snapshot_cached("cached-run", &mut contexts).unwrap();
1939            assert!(Arc::ptr_eq(&cached_ctx, &again));
1940
1941            // A second run makes the listing's ordering real: newest first,
1942            // same as the uncached listing.
1943            let mut second = RunMeta::new(
1944                "cached-run-2".to_string(),
1945                "agent".to_string(),
1946                "/p".to_string(),
1947                "t".to_string(),
1948                None,
1949                "/w".to_string(),
1950                1,
1951            );
1952            second.started_at += 100;
1953            create_run(&second).unwrap();
1954            let listed = list_runs_cached(&mut metas);
1955            assert_eq!(listed.len(), 2);
1956            assert_eq!(listed[0].run_id, "cached-run-2", "newest first");
1957
1958            // A run dir with a garbled meta.json is skipped, not fatal - and
1959            // skipped cheaply on every later tick (the negative result is
1960            // cached until the file changes).
1961            std::fs::create_dir_all(run_dir("garbled-run")).unwrap();
1962            std::fs::write(run_dir("garbled-run").join("meta.json"), "not json {{").unwrap();
1963            assert_eq!(list_runs_cached(&mut metas).len(), 2);
1964
1965            // A run whose dir disappears falls out of the cached listing.
1966            std::fs::remove_dir_all(run_dir("garbled-run")).unwrap();
1967            std::fs::remove_dir_all(run_dir("cached-run")).unwrap();
1968            std::fs::remove_dir_all(run_dir("cached-run-2")).unwrap();
1969            assert!(list_runs_cached(&mut metas).is_empty());
1970            assert!(read_stages_index_cached("cached-run", &mut stages).is_empty());
1971            assert!(read_context_snapshot_cached("cached-run", &mut contexts).is_none());
1972
1973            // And a missing runs DIRECTORY altogether lists nothing (the
1974            // read_dir-failed arm).
1975            std::fs::remove_dir_all(runs_dir()).unwrap();
1976            assert!(list_runs_cached(&mut metas).is_empty());
1977        });
1978    }
1979
1980    #[test]
1981    fn streaming_visitors_return_none_when_the_archive_is_missing() {
1982        with_isolated_runs_dir("streaming-visitors-missing", |_d| {
1983            // One visitor closure of each kind, shared across every call in
1984            // this test - the last pair of calls (on a real archive) executes
1985            // them, so a missing/invalid archive is proven by the counters
1986            // staying put, not by never-run closures.
1987            let points_seen = std::cell::Cell::new(0usize);
1988            let mut on_point = |_: leviath_core::run_archive::PointRef<'_>| {
1989                points_seen.set(points_seen.get() + 1);
1990                std::ops::ControlFlow::Continue(())
1991            };
1992            let records_seen = std::cell::Cell::new(0usize);
1993            let mut on_record = |_: &leviath_core::run_archive::RunRecord| {
1994                records_seen.set(records_seen.get() + 1);
1995                std::ops::ControlFlow::Continue(())
1996            };
1997
1998            assert!(visit_run_archive("no-such-run", &mut on_point).is_none());
1999            assert!(visit_run_records("no-such-run", &mut on_record).is_none());
2000            // A file that is not an archive fails the preamble check.
2001            let run_id = "bad-preamble";
2002            std::fs::create_dir_all(run_dir(run_id)).unwrap();
2003            std::fs::write(run_dir(run_id).join("run.lvr"), b"junk").unwrap();
2004            assert!(visit_run_archive(run_id, &mut on_point).is_none());
2005            assert!(visit_run_records(run_id, &mut on_record).is_none());
2006            assert_eq!((points_seen.get(), records_seen.get()), (0, 0));
2007
2008            // The same closures over a real archive do run.
2009            let real = "streaming-visitors-real";
2010            std::fs::create_dir_all(run_dir(real)).unwrap();
2011            write_minimal_archive(real);
2012            assert!(visit_run_archive(real, &mut on_point).is_some());
2013            assert!(visit_run_records(real, &mut on_record).is_some());
2014            assert_eq!(points_seen.get(), 1);
2015            assert_eq!(records_seen.get(), 2);
2016        });
2017    }
2018
2019    /// Write a two-record archive (Header + one ContextCheckpoint) for `run_id`.
2020    fn write_minimal_archive(run_id: &str) {
2021        use leviath_core::run_archive::{self, RunIdentity, RunRecord};
2022        let mut buf = Vec::new();
2023        run_archive::write_archive_start(&mut buf, run_archive::RUN_ARCHIVE_VERSION).unwrap();
2024        let meta = RunMeta::new(
2025            run_id.to_string(),
2026            "a".to_string(),
2027            "/p".to_string(),
2028            "t".to_string(),
2029            None,
2030            "/w".to_string(),
2031            1,
2032        );
2033        run_archive::write_record(
2034            &mut buf,
2035            &RunRecord::Header {
2036                identity: RunIdentity {
2037                    run_id: run_id.to_string(),
2038                    machine_id: "m".to_string(),
2039                    world_id: "w".to_string(),
2040                    created_at: 0,
2041                },
2042                meta: Box::new(meta),
2043            },
2044        )
2045        .unwrap();
2046        run_archive::write_record(
2047            &mut buf,
2048            &RunRecord::ContextCheckpoint {
2049                snapshot: ContextSnapshot {
2050                    stage_name: "plan".to_string(),
2051                    total_tokens: 3,
2052                    max_tokens: 100,
2053                    regions: vec![],
2054                },
2055                at: 1,
2056            },
2057        )
2058        .unwrap();
2059        std::fs::write(run_dir(run_id).join("run.lvr"), &buf).unwrap();
2060    }
2061
2062    /// The journal keeps `callback_secret` (the daemon re-signs webhooks for a
2063    /// run it reloads), so a replayed point carries it unless the reader strips
2064    /// it. `GET /api/agents/{id}/context/history` serves these points straight
2065    /// out, which handed the webhook signing key to any API token holder.
2066    ///
2067    /// Asserts against the *archive* as well as the history, so the test still
2068    /// means something if the journal ever stops storing the secret: were that
2069    /// to happen, the first assertion fails rather than the second silently
2070    /// passing on a field that is no longer there to leak.
2071    #[test]
2072    fn context_history_redacts_the_webhook_secret_the_journal_keeps() {
2073        with_isolated_runs_dir("context-history-redacts-secret", |_d| {
2074            use leviath_core::run_archive::{self, RunIdentity, RunRecord};
2075            let run_id = "archive-secret-unit";
2076            std::fs::create_dir_all(run_dir(run_id)).unwrap();
2077            let mut buf = Vec::new();
2078            run_archive::write_archive_start(&mut buf, run_archive::RUN_ARCHIVE_VERSION).unwrap();
2079            let mut meta = RunMeta::new(
2080                run_id.to_string(),
2081                "a".to_string(),
2082                "/p".to_string(),
2083                "t".to_string(),
2084                None,
2085                "/w".to_string(),
2086                1,
2087            );
2088            meta.callback_url = Some("https://example.invalid/hook".to_string());
2089            meta.callback_secret = Some("super-secret-signing-key".to_string());
2090            run_archive::write_record(
2091                &mut buf,
2092                &RunRecord::Header {
2093                    identity: RunIdentity {
2094                        run_id: run_id.to_string(),
2095                        machine_id: "m".to_string(),
2096                        world_id: "w".to_string(),
2097                        created_at: 0,
2098                    },
2099                    meta: Box::new(meta),
2100                },
2101            )
2102            .unwrap();
2103            run_archive::write_record(
2104                &mut buf,
2105                &RunRecord::ContextCheckpoint {
2106                    snapshot: ContextSnapshot {
2107                        stage_name: "plan".to_string(),
2108                        total_tokens: 3,
2109                        max_tokens: 100,
2110                        regions: vec![],
2111                    },
2112                    at: 1,
2113                },
2114            )
2115            .unwrap();
2116            std::fs::write(run_dir(run_id).join("run.lvr"), &buf).unwrap();
2117
2118            // The secret really is on disk, so redaction has work to do. Read
2119            // the raw bytes rather than matching over parsed records: a match
2120            // that stops at the Header leaves its other arm unreachable, and
2121            // this says the thing that actually matters anyway.
2122            let raw = std::fs::read(run_dir(run_id).join("run.lvr")).unwrap();
2123            assert!(String::from_utf8_lossy(&raw).contains("super-secret-signing-key"));
2124
2125            // What the reader hands out has it stripped, and keeps the rest.
2126            let history = context_history(run_id);
2127            assert_eq!(history.len(), 1);
2128            assert_eq!(history[0].meta.callback_secret, None);
2129            assert_eq!(
2130                history[0].meta.callback_url.as_deref(),
2131                Some("https://example.invalid/hook")
2132            );
2133            assert_eq!(history[0].context.stage_name, "plan");
2134        });
2135    }
2136
2137    #[test]
2138    fn read_run_archive_missing_or_corrupt_returns_none() {
2139        with_isolated_runs_dir("read-run-archive-corrupt", |_d| {
2140            // Missing archive.
2141            assert!(read_run_archive("no-such-archive-run").is_none());
2142            assert!(context_history("no-such-archive-run").is_empty());
2143            // Corrupt archive (bad magic) → None, not a panic.
2144            let run_id = "corrupt-archive-unit";
2145            std::fs::create_dir_all(run_dir(run_id)).unwrap();
2146            std::fs::write(run_dir(run_id).join("run.lvr"), b"not an archive").unwrap();
2147            assert!(read_run_archive(run_id).is_none());
2148            assert!(context_history(run_id).is_empty());
2149        });
2150    }
2151
2152    // ─── stage_dir / append_stage_output / append_stage_log ─────────────────
2153
2154    #[test]
2155    fn stage_dir_path_structure() {
2156        let path = stage_dir("run-abc", 2);
2157        assert!(path.ends_with("stages/2"));
2158        assert!(path.to_str().unwrap().contains("run-abc"));
2159    }
2160
2161    #[test]
2162    fn append_and_tail_stage_output() {
2163        with_isolated_runs_dir("append-and-tail-stage-output", |_d| {
2164            let run_id = "test-stage-output-unit";
2165            append_stage_output(run_id, 0, "line 1");
2166            append_stage_output(run_id, 0, "line 2");
2167            let output = tail_stage_output(run_id, 0, 4096);
2168            assert!(output.contains("line 1"));
2169            assert!(output.contains("line 2"));
2170        });
2171    }
2172
2173    #[test]
2174    fn append_and_tail_stage_log() {
2175        with_isolated_runs_dir("append-and-tail-stage-log", |_d| {
2176            let run_id = "test-stage-log-unit";
2177            append_stage_log(run_id, 0, "event A");
2178            append_stage_log(run_id, 0, "event B");
2179            let log = tail_stage_log(run_id, 0, 4096);
2180            assert!(log.contains("event A"));
2181            assert!(log.contains("event B"));
2182        });
2183    }
2184
2185    // ─── write/read stage context ───────────────────────────────────────────
2186
2187    #[test]
2188    fn write_and_read_stage_context_roundtrip() {
2189        with_isolated_runs_dir("write-and-read-stage-context-roundtrip", |_d| {
2190            let run_id = "test-stage-ctx-unit";
2191            let snap = ContextSnapshot {
2192                stage_name: "stage-0".into(),
2193                total_tokens: 100,
2194                max_tokens: 4096,
2195                regions: vec![],
2196            };
2197            write_stage_context(run_id, 0, &snap).unwrap();
2198            let back = read_stage_context(run_id, 0).unwrap();
2199            assert_eq!(back.stage_name, "stage-0");
2200        });
2201    }
2202
2203    #[test]
2204    fn read_stage_context_missing_returns_none() {
2205        assert!(read_stage_context("nonexistent-run", 99).is_none());
2206    }
2207
2208    // ─── append_dashboard_log ─────────────────────────────────────────────
2209
2210    #[test]
2211    fn append_dashboard_log_creates_log_file() {
2212        with_isolated_runs_dir("append-dashboard-log-creates-log-file", |_d| {
2213            append_dashboard_log("coverage-test-message");
2214            assert!(dashboard_log_path().exists());
2215        });
2216    }
2217
2218    #[test]
2219    fn append_dashboard_log_open_failure_is_silently_ignored() {
2220        // Covers the `if let Ok(mut file) = ... .open(&path)` pattern *not*
2221        // matching: pre-create the resolved log path as a directory, so
2222        // opening it for append fails with `IsADirectory` - the function
2223        // must swallow this silently (best-effort logging) rather than
2224        // panic.
2225        with_isolated_runs_dir("append-dashboard-log-open-failure", |_d| {
2226            let path = dashboard_log_path();
2227            std::fs::create_dir_all(&path).unwrap();
2228            append_dashboard_log("this should not panic");
2229            assert!(path.is_dir());
2230        });
2231    }
2232
2233    #[test]
2234    fn append_dashboard_log_path_with_no_parent_skips_create_dir_all() {
2235        // Every other test resolves `dashboard_log_path()` to a path with a
2236        // real parent component, leaving the `if let Some(parent) = ...`
2237        // pattern's `None` arm (root paths like "/" have no parent) never
2238        // exercised. `temp_env::with_var` points the override at "/" for the
2239        // closure's duration (serialized process-wide, then restored).
2240        temp_env::with_var("LEVIATH_DASHBOARD_LOG_PATH", Some("/"), || {
2241            assert!(dashboard_log_path().parent().is_none());
2242            append_dashboard_log("this should not panic even with no parent");
2243        });
2244    }
2245
2246    #[test]
2247    fn dashboard_log_rolls_once_over_cap() {
2248        // A tiny cap so a couple of lines trips the roll. The over-cap live file
2249        // is moved to `<name>.1` and a fresh live file is started.
2250        let dir = tempfile::tempdir().unwrap();
2251        let path = dir.path().join("dashboard.log");
2252        append_dashboard_log_capped(&path, "first line well over the tiny cap", 8);
2253        // First write created the file; it now exceeds the 8-byte cap.
2254        assert!(path.exists());
2255        assert!(!rolled_log_path(&path).exists());
2256        // Second write sees the file over cap → rolls it and restarts.
2257        append_dashboard_log_capped(&path, "second", 8);
2258        let rolled = rolled_log_path(&path);
2259        assert!(rolled.exists(), "previous generation rolled to <name>.1");
2260        assert!(
2261            std::fs::read_to_string(&rolled)
2262                .unwrap()
2263                .contains("first line")
2264        );
2265        // The live file was restarted with only the newest line.
2266        let live = std::fs::read_to_string(&path).unwrap();
2267        assert!(live.contains("second"));
2268        assert!(!live.contains("first line"));
2269    }
2270
2271    #[test]
2272    fn dashboard_log_does_not_roll_under_cap() {
2273        let dir = tempfile::tempdir().unwrap();
2274        let path = dir.path().join("dashboard.log");
2275        append_dashboard_log_capped(&path, "a", 1_000_000);
2276        append_dashboard_log_capped(&path, "b", 1_000_000);
2277        // Both lines are in the single live file; nothing was rolled.
2278        assert!(!rolled_log_path(&path).exists());
2279        let live = std::fs::read_to_string(&path).unwrap();
2280        assert!(live.contains("a") && live.contains("b"));
2281    }
2282
2283    // ─── dashboard_log_path ────────────────────────────────────────────────
2284
2285    #[test]
2286    fn dashboard_log_path_structure() {
2287        // Exercises the real (env-reading) `dashboard_log_path()` on its
2288        // fallback branch, so - like `runs_dir_structure` below - it forces
2289        // `LEVIATH_DASHBOARD_LOG_PATH` unset via `temp_env::with_var_unset`,
2290        // which also serializes against every other temp-env test so a
2291        // concurrently-isolated test can't race this assertion.
2292        temp_env::with_var_unset("LEVIATH_DASHBOARD_LOG_PATH", || {
2293            let path = dashboard_log_path();
2294            assert!(path.to_str().unwrap().contains(".leviath"));
2295            assert!(path.to_str().unwrap().ends_with("dashboard.log"));
2296        });
2297    }
2298
2299    /// With no `LEVIATH_DASHBOARD_LOG_PATH`, the dashboard log must follow
2300    /// `LEVIATH_HOME` like every other data path. Resolving through the raw
2301    /// OS home would leave a fully isolated test session still appending to
2302    /// the developer's real `~/.leviath/dashboard.log`.
2303    #[test]
2304    fn dashboard_log_path_honors_leviath_home() {
2305        temp_env::with_vars(
2306            [
2307                ("LEVIATH_DASHBOARD_LOG_PATH", None),
2308                ("LEVIATH_HOME", Some("/custom/home")),
2309            ],
2310            || {
2311                assert_eq!(
2312                    dashboard_log_path(),
2313                    PathBuf::from("/custom/home/.leviath/dashboard.log")
2314                );
2315            },
2316        );
2317    }
2318
2319    // ─── runs_dir / run_dir ────────────────────────────────────────────────
2320
2321    #[test]
2322    fn runs_dir_structure() {
2323        // See the comment on `dashboard_log_path_structure` above - same
2324        // race, same fix, for `LEVIATH_RUNS_DIR`.
2325        temp_env::with_var_unset("LEVIATH_RUNS_DIR", || {
2326            let path = runs_dir();
2327            assert!(path.to_str().unwrap().contains(".leviath"));
2328            assert!(path.to_str().unwrap().ends_with("runs"));
2329        });
2330    }
2331
2332    #[test]
2333    fn runs_dir_from_uses_override_when_provided() {
2334        let path = runs_dir_from(Some("/custom/leviath/runs"));
2335        assert_eq!(path, PathBuf::from("/custom/leviath/runs"));
2336    }
2337
2338    #[test]
2339    fn runs_dir_from_falls_back_to_home_when_none() {
2340        let path = runs_dir_from(None);
2341        #[cfg(unix)]
2342        assert!(path.ends_with(".leviath/runs"));
2343        #[cfg(windows)]
2344        assert!(path.ends_with(".leviath\\runs"));
2345    }
2346
2347    /// With no `LEVIATH_RUNS_DIR`, the runs dir must follow `LEVIATH_HOME` - the
2348    /// same home every other leviath path resolves through. Without this, setting
2349    /// `LEVIATH_HOME` isolates a test's config/socket/agents dir while its runs
2350    /// still land in the real `~/.leviath/runs`.
2351    #[test]
2352    fn runs_dir_follows_leviath_home() {
2353        temp_env::with_vars(
2354            [
2355                ("LEVIATH_RUNS_DIR", None::<&str>),
2356                ("LEVIATH_HOME", Some("/tmp/leviath-home-runs-test")),
2357            ],
2358            || {
2359                assert_eq!(
2360                    runs_dir(),
2361                    PathBuf::from("/tmp/leviath-home-runs-test")
2362                        .join(".leviath")
2363                        .join("runs")
2364                );
2365            },
2366        );
2367    }
2368
2369    #[test]
2370    fn dashboard_log_path_from_uses_override_when_provided() {
2371        let path = dashboard_log_path_from(Some("/custom/leviath/dashboard.log"));
2372        assert_eq!(path, PathBuf::from("/custom/leviath/dashboard.log"));
2373    }
2374
2375    #[test]
2376    fn dashboard_log_path_from_falls_back_to_home_when_none() {
2377        let path = dashboard_log_path_from(None);
2378        #[cfg(unix)]
2379        assert!(path.ends_with(".leviath/dashboard.log"));
2380        #[cfg(windows)]
2381        assert!(path.ends_with(".leviath\\dashboard.log"));
2382    }
2383
2384    #[test]
2385    fn run_dir_contains_run_id() {
2386        let path = run_dir("my-run-123");
2387        assert!(path.to_str().unwrap().contains("my-run-123"));
2388    }
2389
2390    // ─── with_isolated_runs_dir ─────────────────────────────────────────────
2391
2392    #[test]
2393    fn with_isolated_runs_dir_points_at_temp_dir_and_cleans_up_after() {
2394        // Deliberately avoids a racy before/after ambient comparison (a
2395        // concurrently-isolated test could own `LEVIATH_RUNS_DIR` just before
2396        // or after this closure's temp-env window): instead assert the helper's
2397        // own hash-derived path is live *inside* the closure and removed
2398        // afterward - a property no other test can perturb, since none
2399        // produces this exact path.
2400        let inside = with_isolated_runs_dir("helper-self-test", |base_dir| {
2401            let expected = base_dir.join("runs");
2402            assert_eq!(runs_dir(), expected);
2403            assert!(runs_dir().exists());
2404            assert_eq!(dashboard_log_path(), base_dir.join("dashboard.log"));
2405            expected
2406        });
2407        // Closure returned: the temp dir the helper created is gone.
2408        assert!(!inside.exists());
2409    }
2410
2411    // ─── tail_file edge cases ──────────────────────────────────────────────
2412
2413    #[test]
2414    fn tail_file_exact_size() {
2415        let dir = tempfile::tempdir().unwrap();
2416        let path = dir.path().join("exact.txt");
2417        std::fs::write(&path, "exactly").unwrap();
2418        // max_bytes == file size
2419        let result = tail_file(&path, 7);
2420        assert_eq!(result, "exactly");
2421    }
2422
2423    #[test]
2424    fn tail_file_tail_without_newline_returns_whole_window() {
2425        // When the last `max_bytes` window of a larger file contains no '\n'
2426        // at all (a single long line with no line breaks), `tail_file` cannot
2427        // skip to a newline boundary, so it falls through to the `else` arm and
2428        // returns the whole (newline-free) tail window verbatim. Bytes are
2429        // written raw (never via `writeln!`, which would append '\n') so that
2430        // on *every* OS the tail slice is guaranteed newline-free - on Windows
2431        // ordinary text output is `\r\n`-terminated, which would otherwise keep
2432        // a '\n' in the window and take the `if` arm instead.
2433        let dir = tempfile::tempdir().unwrap();
2434        let path = dir.path().join("no_newline.txt");
2435        // 100 raw bytes, no newline anywhere.
2436        let content = "a".repeat(100);
2437        std::fs::write(&path, content.as_bytes()).unwrap();
2438        // A 10-byte window is smaller than the file (100) and contains no '\n'.
2439        let result = tail_file(&path, 10);
2440        assert_eq!(result, "aaaaaaaaaa");
2441    }
2442
2443    // ─── RunMeta metadata and callback_url ─────────────────────────────────
2444
2445    #[test]
2446    fn run_meta_with_metadata() {
2447        let mut meta = RunMeta::new(
2448            "meta-run".into(),
2449            "agent".into(),
2450            "/p".into(),
2451            "task".into(),
2452            None,
2453            "/w".into(),
2454            1,
2455        );
2456        meta.metadata
2457            .insert("key1".to_string(), "value1".to_string());
2458        meta.callback_url = Some("https://example.com/hook".to_string());
2459        meta.parent_run_id = Some("parent-123".to_string());
2460
2461        let json = serde_json::to_string(&meta).unwrap();
2462        let back: RunMeta = serde_json::from_str(&json).unwrap();
2463        assert_eq!(back.metadata.get("key1").unwrap(), "value1");
2464        assert_eq!(
2465            back.callback_url.as_deref(),
2466            Some("https://example.com/hook")
2467        );
2468        assert_eq!(back.parent_run_id.as_deref(), Some("parent-123"));
2469    }
2470
2471    // ─── StageRecord modifications ─────────────────────────────────────────
2472
2473    #[test]
2474    fn stage_record_mutation() {
2475        let mut rec = StageRecord::new("test".into(), 0);
2476        rec.status = StageRunStatus::Active;
2477        rec.started_at = Some(1000);
2478        rec.prompt_tokens = 500;
2479        rec.completion_tokens = 200;
2480        rec.cached_tokens = 50;
2481
2482        assert_eq!(rec.status, StageRunStatus::Active);
2483        assert_eq!(rec.started_at, Some(1000));
2484        assert_eq!(rec.prompt_tokens, 500);
2485        assert_eq!(rec.completion_tokens, 200);
2486        assert_eq!(rec.cached_tokens, 50);
2487
2488        rec.status = StageRunStatus::Complete;
2489        rec.ended_at = Some(2000);
2490        assert_eq!(rec.status, StageRunStatus::Complete);
2491        assert_eq!(rec.ended_at, Some(2000));
2492    }
2493
2494    // ─── ContextSnapshot with entries ──────────────────────────────────────
2495
2496    #[test]
2497    fn context_snapshot_with_entries() {
2498        let snap = ContextSnapshot {
2499            stage_name: "main".into(),
2500            total_tokens: 1000,
2501            max_tokens: 8192,
2502            regions: vec![
2503                RegionSnapshot {
2504                    name: "system".into(),
2505                    kind: "pinned".into(),
2506                    current_tokens: 100,
2507                    max_tokens: 2000,
2508                    entries: vec![
2509                        RegionEntrySnapshot {
2510                            content: "You are helpful".into(),
2511                            tokens: 3,
2512                            kind: Default::default(),
2513                            metadata: None,
2514                            key: None,
2515                            taint: Default::default(),
2516                        },
2517                        RegionEntrySnapshot {
2518                            content: "Additional instruction".into(),
2519                            tokens: 5,
2520                            kind: Default::default(),
2521                            metadata: Some(serde_json::json!({"source": "user"})),
2522                            key: None,
2523                            taint: Default::default(),
2524                        },
2525                    ],
2526                },
2527                RegionSnapshot {
2528                    name: "conversation".into(),
2529                    kind: "sliding".into(),
2530                    current_tokens: 900,
2531                    max_tokens: 6000,
2532                    entries: vec![],
2533                },
2534            ],
2535        };
2536
2537        let json = serde_json::to_string_pretty(&snap).unwrap();
2538        let back: ContextSnapshot = serde_json::from_str(&json).unwrap();
2539        assert_eq!(back.regions.len(), 2);
2540        assert_eq!(back.regions[0].entries.len(), 2);
2541        assert_eq!(back.regions[0].entries[1].tokens, 5);
2542        assert!(back.regions[0].entries[1].metadata.is_some());
2543    }
2544
2545    // ─── RegionEntrySnapshot metadata ──────────────────────────────────────
2546
2547    #[test]
2548    fn region_entry_snapshot_metadata_omitted_when_none() {
2549        let entry = RegionEntrySnapshot {
2550            content: "test".into(),
2551            tokens: 1,
2552            kind: Default::default(),
2553            metadata: None,
2554            key: None,
2555            taint: Default::default(),
2556        };
2557        let json = serde_json::to_value(&entry).unwrap();
2558        assert!(json.get("metadata").is_none());
2559    }
2560
2561    // ─── Multiple stage output appends ─────────────────────────────────────
2562
2563    #[test]
2564    fn append_stage_output_multiple_stages() {
2565        with_isolated_runs_dir("append-stage-output-multiple-stages", |_d| {
2566            let run_id = "test-multi-stage-out";
2567            append_stage_output(run_id, 0, "stage 0 output");
2568            append_stage_output(run_id, 1, "stage 1 output");
2569            append_stage_output(run_id, 2, "stage 2 output");
2570
2571            let out0 = tail_stage_output(run_id, 0, 4096);
2572            let out1 = tail_stage_output(run_id, 1, 4096);
2573            let out2 = tail_stage_output(run_id, 2, 4096);
2574
2575            assert!(out0.contains("stage 0 output"));
2576            assert!(out1.contains("stage 1 output"));
2577            assert!(out2.contains("stage 2 output"));
2578            // Verify no cross-contamination
2579            assert!(!out0.contains("stage 1 output"));
2580        });
2581    }
2582
2583    // ─── list_runs ─────────────────────────────────────────────────────────
2584
2585    #[test]
2586    fn list_runs_returns_sorted() {
2587        with_isolated_runs_dir("list-runs-returns-sorted", |_d| {
2588            let meta1 = RunMeta::new(
2589                "test-list-run-a".into(),
2590                "agent".into(),
2591                "/p".into(),
2592                "task a".into(),
2593                None,
2594                "/w".into(),
2595                1,
2596            );
2597            let meta2 = RunMeta::new(
2598                "test-list-run-b".into(),
2599                "agent".into(),
2600                "/p".into(),
2601                "task b".into(),
2602                None,
2603                "/w".into(),
2604                1,
2605            );
2606
2607            let _ = create_run(&meta1);
2608            // Small delay to ensure different timestamps
2609            let _ = create_run(&meta2);
2610
2611            let runs = list_runs();
2612            // Both should appear in the list
2613            let ids: Vec<&str> = runs.iter().map(|r| r.run_id.as_str()).collect();
2614            assert!(ids.contains(&"test-list-run-a"));
2615            assert!(ids.contains(&"test-list-run-b"));
2616        });
2617    }
2618
2619    // ─── tail_stage_log / tail_stage_output empty ──────────────────────────
2620
2621    #[test]
2622    fn tail_stage_output_nonexistent_returns_empty() {
2623        assert_eq!(tail_stage_output("no-such-run-xyz", 0, 4096), "");
2624    }
2625
2626    #[test]
2627    fn tail_stage_log_nonexistent_returns_empty() {
2628        assert_eq!(tail_stage_log("no-such-run-xyz", 0, 4096), "");
2629    }
2630
2631    // ─── list_runs_in_dir ───────────────────────────────────────────────────
2632
2633    #[test]
2634    fn list_runs_in_dir_nonexistent_returns_empty() {
2635        let result = list_runs_in_dir(PathBuf::from("/nonexistent/leviath/runs/coverage-test"));
2636        assert!(result.is_empty());
2637    }
2638
2639    #[test]
2640    fn list_runs_in_dir_empty_dir_returns_empty() {
2641        let dir = tempfile::tempdir().unwrap();
2642        let result = list_runs_in_dir(dir.path().to_path_buf());
2643        assert!(result.is_empty());
2644    }
2645
2646    #[test]
2647    fn list_runs_in_dir_unreadable_dir_returns_empty() {
2648        // Covers the `if let Ok(entries) = std::fs::read_dir(&dir)` pattern
2649        // *not* matching: `dir.exists()` is true (so the earlier early-return
2650        // is skipped) but `read_dir` fails, so the whole block is silently
2651        // skipped. Pointing at a *file* makes `read_dir` fail on every platform.
2652        let dir = tempfile::tempdir().unwrap();
2653        let not_a_dir = dir.path().join("runs-is-a-file");
2654        std::fs::write(&not_a_dir, "not a dir").unwrap();
2655        let result = list_runs_in_dir(not_a_dir);
2656        assert!(result.is_empty());
2657    }
2658
2659    #[test]
2660    fn append_stage_output_open_failure_is_silently_skipped() {
2661        // When `output.log` already exists as a *directory*, `OpenOptions::open`
2662        // fails and the write is silently skipped (the `if let Ok(file)` false
2663        // path). Making the target a directory fails the open on every platform.
2664        crate::runstate::with_isolated_runs_dir("append_stage_output_open_failure", |_d| {
2665            let run_id = "append-out-openfail";
2666            ensure_stage_dir(run_id, 0);
2667            std::fs::create_dir_all(stage_dir(run_id, 0).join("output.log")).unwrap();
2668            append_stage_output(run_id, 0, "ignored"); // must not panic
2669        });
2670    }
2671
2672    #[test]
2673    fn append_stage_log_open_failure_is_silently_skipped() {
2674        // Same as above for `logs.log` in `append_stage_log`.
2675        crate::runstate::with_isolated_runs_dir("append_stage_log_open_failure", |_d| {
2676            let run_id = "append-log-openfail";
2677            ensure_stage_dir(run_id, 0);
2678            std::fs::create_dir_all(stage_dir(run_id, 0).join("logs.log")).unwrap();
2679            append_stage_log(run_id, 0, "ignored"); // must not panic
2680        });
2681    }
2682
2683    // ─── runs_dir / list_runs edge cases ────────────────────────────────────
2684
2685    #[test]
2686    fn runs_dir_with_override_set_returns_override() {
2687        let tmpdir = tempfile::tempdir().unwrap();
2688        temp_env::with_var("LEVIATH_RUNS_DIR", Some(tmpdir.path()), || {
2689            assert_eq!(runs_dir(), tmpdir.path());
2690        });
2691    }
2692
2693    #[test]
2694    fn runs_dir_without_override_falls_back_to_home() {
2695        temp_env::with_var_unset("LEVIATH_RUNS_DIR", || {
2696            let dir = runs_dir();
2697            #[cfg(unix)]
2698            assert!(dir.ends_with(".leviath/runs"));
2699            #[cfg(windows)]
2700            assert!(dir.ends_with(".leviath\\runs"));
2701        });
2702    }
2703
2704    #[test]
2705    fn list_runs_empty_when_runs_dir_missing_or_empty() {
2706        // Isolated via `isolate_runs_dir_for_test`, so this is a genuinely
2707        // empty runs dir (not "the real dir, which we hope has no entry with
2708        // this exact bogus id") - can assert real emptiness instead of just
2709        // absence of one specific id.
2710        with_isolated_runs_dir("list-runs-empty-when-runs-dir-missing-or-empty", |_d| {
2711            let runs = list_runs();
2712            assert!(runs.is_empty());
2713        });
2714    }
2715
2716    #[test]
2717    fn tail_file_nonexistent_path_returns_empty() {
2718        let path = std::path::Path::new("/nonexistent/path/to/a/file.log");
2719        assert_eq!(tail_file(path, 1024), "");
2720    }
2721
2722    #[test]
2723    fn tail_file_small_file_returns_whole_contents() {
2724        let dir = tempfile::tempdir().unwrap();
2725        let path = dir.path().join("small.log");
2726        std::fs::write(&path, "hello world").unwrap();
2727        assert_eq!(tail_file(&path, 1024), "hello world");
2728    }
2729
2730    #[test]
2731    fn tail_file_large_file_truncates_from_offset() {
2732        let dir = tempfile::tempdir().unwrap();
2733        let path = dir.path().join("big.log");
2734        let content = "a".repeat(100) + "\nTAIL_MARKER\n";
2735        std::fs::write(&path, &content).unwrap();
2736        let tailed = tail_file(&path, 20);
2737        assert!(tailed.contains("TAIL_MARKER"));
2738        assert!(tailed.len() < content.len());
2739    }
2740
2741    #[test]
2742    fn tail_file_directory_path_returns_empty() {
2743        // metadata() and File::open() both succeed on a directory (confirmed
2744        // empirically on macOS/Linux); it's read_to_end() that fails with
2745        // "Is a directory" - and that error is deliberately discarded (`let
2746        // _ = file.read_to_end(&mut buf);`), so this exercises the
2747        // graceful-empty-buffer fallback at the bottom of the function, not
2748        // either of the two `Err(_) => return String::new()` early returns.
2749        let dir = tempfile::tempdir().unwrap();
2750        assert_eq!(tail_file(dir.path(), 4), "");
2751    }
2752
2753    #[cfg(unix)]
2754    #[test]
2755    fn tail_file_open_permission_denied_returns_empty() {
2756        // A file with no permissions at all: `Path::exists()`/`fs::metadata()`
2757        // only need search (execute) permission on the *parent* directories
2758        // to stat a path, not read permission on the file itself - so both
2759        // succeed here. `std::fs::File::open()` in read mode, however,
2760        // genuinely fails with `PermissionDenied`. Unlike the metadata-error
2761        // arm (only reachable via a delete-between-calls race), this is a
2762        // deterministic way to exercise the `File::open` `Err(_)` arm.
2763        use std::os::unix::fs::PermissionsExt;
2764
2765        let dir = tempfile::tempdir().unwrap();
2766        let path = dir.path().join("no-permissions.log");
2767        // Content must exceed max_bytes so the "whole file" fast path
2768        // (`file_size <= max_bytes`) doesn't short-circuit before reaching
2769        // the `File::open` call under test.
2770        std::fs::write(&path, "x".repeat(100)).unwrap();
2771        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o000)).unwrap();
2772
2773        assert_eq!(tail_file(&path, 4), "");
2774
2775        // Restore permissions so the tempdir can clean itself up on drop.
2776        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();
2777    }
2778
2779    // ─── hermetic write/read coverage tests (use _to/_from/_in helpers) ───────
2780
2781    #[test]
2782    fn write_context_snapshot_to_hermetic() {
2783        let dir = tempfile::tempdir().unwrap();
2784        let snap = ContextSnapshot {
2785            stage_name: "cov-stage".into(),
2786            total_tokens: 42,
2787            max_tokens: 8192,
2788            regions: vec![],
2789        };
2790        write_context_snapshot_to(dir.path(), &snap).unwrap();
2791        let json = std::fs::read_to_string(dir.path().join("context.json")).unwrap();
2792        let back: ContextSnapshot = serde_json::from_str(&json).unwrap();
2793        assert_eq!(back.total_tokens, 42);
2794    }
2795
2796    #[test]
2797    fn write_context_snapshot_to_fails_without_dir() {
2798        let snap = ContextSnapshot {
2799            stage_name: "s".into(),
2800            total_tokens: 1,
2801            max_tokens: 100,
2802            regions: vec![],
2803        };
2804        let nonexistent = std::path::Path::new("/nonexistent-cov-dir-xyzzy-abc");
2805        let result = write_context_snapshot_to(nonexistent, &snap);
2806        assert!(result.is_err());
2807    }
2808
2809    #[test]
2810    fn write_context_snapshot_to_fails_when_rename_target_is_a_dir() {
2811        // Covers the `std::fs::rename(&tmp, &path)?` `Err` arm: the tmp file
2812        // write succeeds (its directory is writable), but the final rename
2813        // fails because `context.json` already exists as a *directory* --
2814        // `rename(2)` on POSIX refuses to replace a directory with a
2815        // regular file, unlike a plain overwrite of an existing file.
2816        let dir = tempfile::tempdir().unwrap();
2817        std::fs::create_dir(dir.path().join("context.json")).unwrap();
2818        let snap = ContextSnapshot {
2819            stage_name: "s".into(),
2820            total_tokens: 1,
2821            max_tokens: 100,
2822            regions: vec![],
2823        };
2824        let result = write_context_snapshot_to(dir.path(), &snap);
2825        assert!(result.is_err());
2826    }
2827
2828    #[test]
2829    fn create_run_in_hermetic() {
2830        let tmpdir = tempfile::tempdir().unwrap();
2831        let run_dir = tmpdir.path().join("cov-run");
2832        let meta = RunMeta::new(
2833            "cov-run".into(),
2834            "cov-agent".into(),
2835            "/agents/cov".into(),
2836            "cov task".into(),
2837            None,
2838            "/tmp".into(),
2839            1,
2840        );
2841        create_run_in(&run_dir, &meta).unwrap();
2842        let back = read_meta_from(&run_dir).unwrap();
2843        assert_eq!(back.run_id, "cov-run");
2844    }
2845
2846    #[test]
2847    fn create_run_in_fails_on_bad_parent() {
2848        // A hardcoded "/nonexistent-.../run" path isn't reliably bad across
2849        // platforms: on Windows CI runners (which typically have write
2850        // access to create directories at the drive root), that path
2851        // resolves under the current drive's root and create_dir_all
2852        // actually succeeds there, while on Unix it fails because writing
2853        // to the real filesystem root needs privileges the CI user lacks --
2854        // this passed locally but failed on Windows CI. Use a path with a
2855        // regular file as a parent component instead: create_dir_all can
2856        // never succeed under a file, on any platform or set of permissions.
2857        let dir = tempfile::tempdir().unwrap();
2858        let not_a_dir = dir.path().join("not-a-directory");
2859        std::fs::write(&not_a_dir, "x").unwrap();
2860        let bad = not_a_dir.join("run");
2861        let meta = RunMeta::new(
2862            "run".into(),
2863            "a".into(),
2864            "/".into(),
2865            "t".into(),
2866            None,
2867            "/tmp".into(),
2868            1,
2869        );
2870        let result = create_run_in(&bad, &meta);
2871        assert!(result.is_err());
2872    }
2873
2874    #[test]
2875    fn write_meta_to_hermetic() {
2876        let tmpdir = tempfile::tempdir().unwrap();
2877        let meta = RunMeta::new(
2878            "cov-write-meta".into(),
2879            "a".into(),
2880            "/".into(),
2881            "t".into(),
2882            None,
2883            "/tmp".into(),
2884            1,
2885        );
2886        write_meta_to(tmpdir.path(), &meta).unwrap();
2887        let back = read_meta_from(tmpdir.path()).unwrap();
2888        assert_eq!(back.run_id, "cov-write-meta");
2889    }
2890
2891    #[test]
2892    fn write_meta_to_fails_without_dir() {
2893        let meta = RunMeta::new(
2894            "cov-no-dir".into(),
2895            "a".into(),
2896            "/".into(),
2897            "t".into(),
2898            None,
2899            "/tmp".into(),
2900            1,
2901        );
2902        let bad = std::path::Path::new("/nonexistent-cov-write-meta-xyzzy");
2903        let result = write_meta_to(bad, &meta);
2904        assert!(result.is_err());
2905    }
2906
2907    #[test]
2908    fn write_meta_to_fails_when_rename_target_is_a_dir() {
2909        // See `write_context_snapshot_to_fails_when_rename_target_is_a_dir`:
2910        // same `std::fs::rename(&tmp_path, &final_path)?` `Err` arm, forced
2911        // by pre-creating `meta.json` as a directory.
2912        let dir = tempfile::tempdir().unwrap();
2913        std::fs::create_dir(dir.path().join("meta.json")).unwrap();
2914        let meta = RunMeta::new(
2915            "cov-rename-fail".into(),
2916            "a".into(),
2917            "/".into(),
2918            "t".into(),
2919            None,
2920            "/tmp".into(),
2921            1,
2922        );
2923        let result = write_meta_to(dir.path(), &meta);
2924        assert!(result.is_err());
2925    }
2926
2927    #[test]
2928    fn read_meta_from_fails_on_missing_file() {
2929        let tmpdir = tempfile::tempdir().unwrap();
2930        let result = read_meta_from(tmpdir.path());
2931        assert!(result.is_err());
2932    }
2933
2934    #[test]
2935    fn write_stages_index_to_hermetic() {
2936        let tmpdir = tempfile::tempdir().unwrap();
2937        let stages = vec![StageRecord::new("cov-stage".into(), 0)];
2938        write_stages_index_to(tmpdir.path(), &stages).unwrap();
2939        let json = std::fs::read_to_string(tmpdir.path().join("stages.json")).unwrap();
2940        let back: Vec<StageRecord> = serde_json::from_str(&json).unwrap();
2941        assert_eq!(back.len(), 1);
2942        assert_eq!(back[0].name, "cov-stage");
2943    }
2944
2945    #[test]
2946    fn write_stages_index_to_fails_without_dir() {
2947        let stages = vec![StageRecord::new("s".into(), 0)];
2948        let bad = std::path::Path::new("/nonexistent-cov-stages-xyzzy");
2949        let result = write_stages_index_to(bad, &stages);
2950        assert!(result.is_err());
2951    }
2952
2953    #[test]
2954    fn write_stages_index_to_fails_when_rename_target_is_a_dir() {
2955        // See `write_context_snapshot_to_fails_when_rename_target_is_a_dir`:
2956        // same `std::fs::rename(&tmp, &path)?` `Err` arm, forced by
2957        // pre-creating `stages.json` as a directory.
2958        let dir = tempfile::tempdir().unwrap();
2959        std::fs::create_dir(dir.path().join("stages.json")).unwrap();
2960        let stages = vec![StageRecord::new("s".into(), 0)];
2961        let result = write_stages_index_to(dir.path(), &stages);
2962        assert!(result.is_err());
2963    }
2964
2965    #[test]
2966    fn list_runs_in_dir_includes_valid_run() {
2967        let tmpdir = tempfile::tempdir().unwrap();
2968        let run_id = "cov-listed-run";
2969        let run_subdir = tmpdir.path().join(run_id);
2970        std::fs::create_dir_all(&run_subdir).unwrap();
2971        let meta = RunMeta::new(
2972            run_id.into(),
2973            "list-agent".into(),
2974            "/agents/list".into(),
2975            "list task".into(),
2976            None,
2977            "/tmp".into(),
2978            1,
2979        );
2980        let json = serde_json::to_string_pretty(&meta).unwrap();
2981        std::fs::write(run_subdir.join("meta.json"), &json).unwrap();
2982
2983        // list_runs_in_dir now reads meta.json directly from the dir, no env var needed
2984        let runs = list_runs_in_dir(tmpdir.path().to_path_buf());
2985        assert!(runs.iter().any(|r| r.run_id == run_id));
2986    }
2987
2988    #[test]
2989    fn list_runs_in_dir_skips_entry_with_corrupted_meta_json() {
2990        // Exercises the `if let Ok(meta) = serde_json::from_str::<RunMeta>(...)`
2991        // else arm: a subdirectory whose meta.json exists and is readable as
2992        // a string, but doesn't parse as a `RunMeta`, is silently skipped
2993        // rather than propagating an error.
2994        let tmpdir = tempfile::tempdir().unwrap();
2995        let good_run_id = "cov-listed-good-run";
2996        let bad_run_id = "cov-listed-corrupted-run";
2997
2998        let good_subdir = tmpdir.path().join(good_run_id);
2999        std::fs::create_dir_all(&good_subdir).unwrap();
3000        let meta = RunMeta::new(
3001            good_run_id.into(),
3002            "list-agent".into(),
3003            "/agents/list".into(),
3004            "list task".into(),
3005            None,
3006            "/tmp".into(),
3007            1,
3008        );
3009        let json = serde_json::to_string_pretty(&meta).unwrap();
3010        std::fs::write(good_subdir.join("meta.json"), &json).unwrap();
3011
3012        let bad_subdir = tmpdir.path().join(bad_run_id);
3013        std::fs::create_dir_all(&bad_subdir).unwrap();
3014        std::fs::write(bad_subdir.join("meta.json"), "not valid json").unwrap();
3015
3016        // A subdirectory with NO meta.json exercises the *other* skip branch:
3017        // the `if let Ok(json) = read_to_string(&meta_path)` else arm (the file
3018        // can't be read), distinct from the parse-fails arm above. Covering
3019        // both here keeps list_runs_in_dir at 100% on every OS deterministically.
3020        let no_meta_run_id = "cov-listed-no-meta-run";
3021        std::fs::create_dir_all(tmpdir.path().join(no_meta_run_id)).unwrap();
3022
3023        let runs = list_runs_in_dir(tmpdir.path().to_path_buf());
3024        assert!(runs.iter().any(|r| r.run_id == good_run_id));
3025        assert!(!runs.iter().any(|r| r.run_id == bad_run_id));
3026        assert!(!runs.iter().any(|r| r.run_id == no_meta_run_id));
3027    }
3028
3029    // ─── force_cancel_in: the floor under every kill path ───
3030
3031    /// Write a run dir with `status` and return its path.
3032    fn run_dir_with(base: &std::path::Path, run_id: &str, status: RunStatus) -> PathBuf {
3033        let dir = base.join(run_id);
3034        let meta = RunMeta {
3035            status,
3036            ..RunMeta::new(
3037                run_id.into(),
3038                "a".into(),
3039                "/p".into(),
3040                "t".into(),
3041                None,
3042                "/w".into(),
3043                1,
3044            )
3045        };
3046        create_run_in(&dir, &meta).unwrap();
3047        dir
3048    }
3049
3050    #[test]
3051    fn force_cancel_terminates_every_non_terminal_status() {
3052        let base = tempfile::tempdir().unwrap();
3053        for status in [
3054            RunStatus::Starting,
3055            RunStatus::Running,
3056            RunStatus::WaitingInput,
3057        ] {
3058            let dir = run_dir_with(base.path(), &format!("live-{status}"), status.clone());
3059            assert_eq!(force_cancel_in(&dir, 99), ForceCancelOutcome::Terminated);
3060            let meta = read_meta_from(&dir).unwrap();
3061            assert_eq!(meta.status, RunStatus::Cancelled, "{status} is killable");
3062            assert_eq!(meta.updated_at, 99, "the cancel is stamped");
3063        }
3064    }
3065
3066    #[test]
3067    fn force_cancel_leaves_a_finished_run_alone() {
3068        let base = tempfile::tempdir().unwrap();
3069        for status in [
3070            RunStatus::Complete,
3071            RunStatus::CompleteInteractive,
3072            RunStatus::Error,
3073            RunStatus::Cancelled,
3074        ] {
3075            let dir = run_dir_with(base.path(), &format!("done-{status}"), status.clone());
3076            assert_eq!(
3077                force_cancel_in(&dir, 99),
3078                ForceCancelOutcome::AlreadyTerminal,
3079                "{status} is already finished"
3080            );
3081            assert_eq!(read_meta_from(&dir).unwrap().status, status);
3082        }
3083    }
3084
3085    #[test]
3086    fn force_cancel_reports_no_such_run_for_a_missing_directory() {
3087        let base = tempfile::tempdir().unwrap();
3088        let outcome = force_cancel_in(&base.path().join("ghost"), 99);
3089        assert_eq!(outcome, ForceCancelOutcome::NoSuchRun);
3090        assert!(!outcome.found_run(), "nothing to cancel");
3091    }
3092
3093    /// A run dir whose metadata can't be parsed still gets terminated. Such a run
3094    /// is skipped by `list_runs`, so leaving it alone makes it both invisible and
3095    /// permanent - the one state from which there is no way back.
3096    #[test]
3097    fn force_cancel_writes_a_record_over_unreadable_metadata() {
3098        let base = tempfile::tempdir().unwrap();
3099        let dir = base.path().join("corrupt-run");
3100        std::fs::create_dir_all(&dir).unwrap();
3101        std::fs::write(dir.join("meta.json"), "{ not json").unwrap();
3102
3103        assert_eq!(force_cancel_in(&dir, 99), ForceCancelOutcome::Terminated);
3104        let meta = read_meta_from(&dir).expect("now parses");
3105        assert_eq!(meta.status, RunStatus::Cancelled);
3106        assert_eq!(meta.run_id, "corrupt-run", "recovered from the dir name");
3107        assert!(meta.error.is_some(), "records why it was synthesized");
3108    }
3109
3110    /// A directory that exists but can't be written still counts as "found" - the
3111    /// caller must not report "no such run" for a run that plainly exists.
3112    #[test]
3113    fn force_cancel_reports_a_write_failure_but_still_found_the_run() {
3114        crate::test_support::with_tracing(|| {
3115            let base = tempfile::tempdir().unwrap();
3116            let dir = base.path().join("blocked-run");
3117            std::fs::create_dir_all(&dir).unwrap();
3118            // A directory where `meta.json` must go: the rename can't succeed.
3119            std::fs::create_dir_all(dir.join("meta.json")).unwrap();
3120
3121            let outcome = force_cancel_in(&dir, 99);
3122            assert_eq!(outcome, ForceCancelOutcome::WriteFailed);
3123            assert!(outcome.found_run());
3124        });
3125    }
3126
3127    /// The spawn that never became a run: the placeholder is `Starting`, which
3128    /// is not terminal, so it has to be rewritten or it claims to be alive for
3129    /// ever (issue #190).
3130    #[test]
3131    fn force_error_records_the_failure_over_a_starting_placeholder() {
3132        let base = tempfile::tempdir().unwrap();
3133        let dir = base.path().join("stillborn-run");
3134        let meta = RunMeta::new(
3135            "stillborn-run".to_string(),
3136            "agent".to_string(),
3137            "/no/such/agent.leviath".to_string(),
3138            "t".to_string(),
3139            None,
3140            "/tmp".to_string(),
3141            0,
3142        );
3143        create_run_in(&dir, &meta).unwrap();
3144        assert_eq!(read_meta_from(&dir).unwrap().status, RunStatus::Starting);
3145
3146        assert_eq!(
3147            force_error_in(&dir, "blueprint not found", 99),
3148            ForceCancelOutcome::Terminated
3149        );
3150
3151        let written = read_meta_from(&dir).unwrap();
3152        assert_eq!(written.status, RunStatus::Error);
3153        assert_eq!(written.error.as_deref(), Some("blueprint not found"));
3154        assert_eq!(written.updated_at, 99);
3155        // The rest of the placeholder survives, so the run still explains itself.
3156        assert_eq!(written.task, "t");
3157    }
3158
3159    #[test]
3160    fn force_error_leaves_a_run_that_already_finished_alone() {
3161        let base = tempfile::tempdir().unwrap();
3162        let dir = base.path().join("done-run");
3163        let mut meta = RunMeta::new(
3164            "done-run".to_string(),
3165            "agent".to_string(),
3166            String::new(),
3167            "t".to_string(),
3168            None,
3169            "/tmp".to_string(),
3170            0,
3171        );
3172        meta.status = RunStatus::Complete;
3173        create_run_in(&dir, &meta).unwrap();
3174
3175        assert_eq!(
3176            force_error_in(&dir, "too late", 99),
3177            ForceCancelOutcome::AlreadyTerminal
3178        );
3179        assert_eq!(read_meta_from(&dir).unwrap().status, RunStatus::Complete);
3180    }
3181
3182    #[test]
3183    fn force_cancel_keeps_an_error_the_run_had_already_recorded() {
3184        // Cancelling passes no message of its own, so whatever the run managed
3185        // to say about itself before it was killed must survive.
3186        let base = tempfile::tempdir().unwrap();
3187        let dir = base.path().join("noisy-run");
3188        let mut meta = RunMeta::new(
3189            "noisy-run".to_string(),
3190            "agent".to_string(),
3191            String::new(),
3192            "t".to_string(),
3193            None,
3194            "/tmp".to_string(),
3195            0,
3196        );
3197        meta.error = Some("a provider hiccup".to_string());
3198        create_run_in(&dir, &meta).unwrap();
3199
3200        assert_eq!(force_cancel_in(&dir, 99), ForceCancelOutcome::Terminated);
3201        let written = read_meta_from(&dir).unwrap();
3202        assert_eq!(written.status, RunStatus::Cancelled);
3203        assert_eq!(written.error.as_deref(), Some("a provider hiccup"));
3204    }
3205
3206    #[test]
3207    fn force_error_writes_its_message_over_unreadable_metadata() {
3208        let base = tempfile::tempdir().unwrap();
3209        let dir = base.path().join("corrupt-stillborn");
3210        std::fs::create_dir_all(&dir).unwrap();
3211        std::fs::write(dir.join("meta.json"), "{ not json").unwrap();
3212
3213        assert_eq!(
3214            force_error_in(&dir, "blueprint not found", 99),
3215            ForceCancelOutcome::Terminated
3216        );
3217        let written = read_meta_from(&dir).expect("now parses");
3218        assert_eq!(written.status, RunStatus::Error);
3219        assert_eq!(written.error.as_deref(), Some("blueprint not found"));
3220    }
3221
3222    #[test]
3223    fn append_dashboard_log_writes_message() {
3224        // Exercises the create_dir_all branch and writeln! branch via a unique marker.
3225        with_isolated_runs_dir("append-dashboard-log-writes-message", |_d| {
3226            let unique = format!("cov-dashboard-log-{}", std::process::id());
3227            append_dashboard_log(&unique);
3228            let content = std::fs::read_to_string(dashboard_log_path()).unwrap_or_default();
3229            assert!(content.contains(&unique));
3230        });
3231    }
3232}