overbrainer 0.9.0

//! The Training view's model: the runs of `runs/`, their metrics, and the
//! training tasks the TUI follows.

use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::path::Path;
use std::time::{Duration, SystemTime};

use super::tasks::TaskId;
use crate::events::Event;
use crate::exec::JobStatus;
use crate::runpod::{PodRecord, PodState, PodStatus};
use crate::runs::{RunRecord, RunState, Runs};
use crate::system::SystemSample;
use crate::train::{METRICS_FILE, MetricLine, Pace, Phase, Phases, TrainMetric, parse_line};

/// Samples of a run's machine kept for the system panel: ten minutes at one
/// sample every 10 seconds.
pub(super) const SYSTEM_SAMPLES: usize = 60;

/// A run of `runs/`, with its pod record when it has one.
#[derive(Debug, Clone, PartialEq)]
pub(super) struct RunRow {
    /// `run.json`.
    pub(super) record: RunRecord,
    /// `pod.json`, for a Runpod run.
    pub(super) pod: Option<PodRecord>,
}

/// What the pod of `record` spent by `now` at `rate` USD per hour, while it
/// exists; `None` when the rate or its creation time is unknown.
fn live_spend(record: &PodRecord, rate: Option<f64>, now: SystemTime) -> Option<f64> {
    Some(rate? * record.uptime(now)?.as_secs_f64() / 3600.0)
}

/// The rate of the pod of `record`, USD per hour: as recorded, else as its
/// `latest` status gave it.
pub(super) fn pod_rate(record: &PodRecord, latest: Option<&PodStatus>) -> Option<f64> {
    record.cost_per_hour.or(match latest {
        Some(PodStatus::Created { cost_per_hour, .. }) => *cost_per_hour,
        _ => None,
    })
}

/// What the pod of `record` spent by `now`: as its `latest` status or its
/// record gave it once deleted, else its rate (see [`pod_rate`]) times its
/// uptime. `None` when unknown.
pub(super) fn pod_spend(
    record: &PodRecord,
    latest: Option<&PodStatus>,
    now: SystemTime,
) -> Option<f64> {
    match latest {
        Some(PodStatus::Deleted {
            estimated_spend, ..
        }) => estimated_spend.or(record.estimated_spend),
        _ if record.state == PodState::Deleted => record.estimated_spend,
        _ => live_spend(record, pod_rate(record, latest), now),
    }
}

/// What a training task does.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) enum Job {
    /// Starts a run, then follows it.
    Start {
        /// Whether its target is a Runpod one.
        runpod: bool,
    },
    /// Follows a run again.
    Attach,
    /// Cancels a run's job.
    Cancel,
    /// Stops a run's job with a snapshot, then follows it until it ends.
    Stop,
}

/// Whether a training task is detached: its token cancelled, so its raced watch
/// ends and the flow reports that the run keeps running.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum Detach {
    /// Not asked to.
    No,
    /// Asked to before its job's first status: its token is cancelled only then,
    /// so a start is never cut.
    OnStart,
    /// Its token is cancelled.
    Done,
}

/// A training task the TUI runs.
#[derive(Debug, Clone, PartialEq)]
pub(super) struct Follow {
    /// What it does.
    pub(super) job: Job,
    /// Its run; empty until a new run is created.
    pub(super) run_id: String,
    /// Whether its job's status was seen: its watch began.
    pub(super) watching: bool,
    /// The latest status of its job: once it exited, the run's results are
    /// being retrieved.
    pub(super) status: Option<JobStatus>,
    /// The latest status of its pod.
    pub(super) pod: Option<PodStatus>,
    /// The lines the command line would print.
    pub(super) lines: Vec<String>,
    /// Metrics its forwarder skipped.
    pub(super) skipped: u64,
    /// Whether the run is cancelled once this task, detached, ends.
    pub(super) cancel_after: bool,
    /// Whether the run is stopped with a snapshot once this task, detached,
    /// ends.
    pub(super) stop_after: bool,
    /// The step of the snapshot its run ended stopped with, once it did.
    pub(super) stopped_at: Option<u64>,
    /// Whether it is detached, or to be once its job's first status arrives.
    pub(super) detach: Detach,
}

impl Follow {
    /// A task doing `job` on run `run_id`.
    pub(super) fn new(job: Job, run_id: &str) -> Self {
        Self {
            job,
            run_id: run_id.to_string(),
            watching: false,
            status: None,
            pod: None,
            lines: Vec::new(),
            skipped: 0,
            cancel_after: false,
            stop_after: false,
            stopped_at: None,
            detach: Detach::No,
        }
    }

    /// Whether it is a start whose job has not started yet: it must not be
    /// interrupted, and it holds the data lock.
    pub(super) fn starting(&self) -> bool {
        matches!(self.job, Job::Start { .. }) && !self.watching
    }

    /// Its run, or `a new run` before it is created.
    pub(super) fn run(&self) -> String {
        if self.run_id.is_empty() {
            "a new run".to_string()
        } else {
            format!("run {}", self.run_id)
        }
    }
}

/// What a training task does to its run, as the Training view shows it:
/// computed in one place, [`RunActivity::of`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum RunActivity {
    /// No task on the run.
    None,
    /// A task follows the run: its job started, or it attached to it.
    Followed,
    /// A start whose job has not started yet; `c` abandons a Runpod one.
    Starting {
        /// Whether its target is a Runpod one.
        runpod: bool,
    },
    /// A start whose job has not started yet, being abandoned: its token is
    /// cancelled.
    Abandoning {
        /// Whether its target is a Runpod one.
        runpod: bool,
    },
    /// A cancel task, or a task whose run is cancelled once it ended.
    Cancelling,
    /// A stop task, or a task whose run is stopped once it ended.
    Stopping,
}

impl RunActivity {
    /// What `follow`, the task on a run if any, does to it.
    pub(super) fn of(follow: Option<&Follow>) -> Self {
        let Some(follow) = follow else {
            return Self::None;
        };
        if follow.job == Job::Cancel || follow.cancel_after {
            return Self::Cancelling;
        }
        if follow.job == Job::Stop || follow.stop_after {
            return Self::Stopping;
        }
        let runpod = follow.job == (Job::Start { runpod: true });
        match (follow.starting(), follow.detach) {
            (true, Detach::Done) => Self::Abandoning { runpod },
            (true, _) => Self::Starting { runpod },
            (false, _) => Self::Followed,
        }
    }

    /// The activity in one word, as the runs table shows it; empty for none.
    /// A start being abandoned still reads `starting`: it is, until its
    /// flow ends.
    pub(super) fn label(self) -> &'static str {
        match self {
            Self::None => "",
            Self::Followed => "followed",
            Self::Starting { .. } | Self::Abandoning { .. } => "starting",
            Self::Cancelling => "cancelling",
            Self::Stopping => "stopping",
        }
    }

    /// Whether `c` abandons the run rather than cancel it: a Runpod start
    /// still starting, which has no job to cancel yet.
    pub(super) fn abandons(self) -> bool {
        matches!(
            self,
            Self::Starting { runpod: true } | Self::Abandoning { runpod: true }
        )
    }
}

/// How the last task of a run ended.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub(super) struct Ended {
    /// The lines it reported.
    pub(super) lines: Vec<String>,
    /// Its error, if it failed or was detached.
    pub(super) error: Option<String>,
    /// Metrics its forwarder skipped.
    pub(super) skipped: u64,
    /// Whether the run's metrics were read again from its local file since: they
    /// then hold every metric, late or skipped.
    pub(super) healed: bool,
}

/// What a read of `runs/` found: the runs newest first, or why they cannot be
/// listed, and the warnings of the records it skipped.
#[derive(Debug, Clone, PartialEq)]
pub(super) struct Listing {
    /// The runs, newest first.
    pub(super) runs: Result<Vec<RunRow>, String>,
    /// Why a run record or a pod record was skipped, one line each.
    pub(super) skipped: Vec<String>,
}

/// Reads the runs of the project in `dir`, with their pod records: local files
/// only. Blocking.
pub(super) fn list_runs(dir: &Path) -> Listing {
    let runs = Runs::new(dir);
    let mut skipped = Vec::new();
    let listed = runs.list_with(|id, error| {
        skipped.push(format!("skipping unreadable run record for {id}: {error}"));
    });
    let runs = match listed {
        Ok(records) => Ok(records
            .into_iter()
            .rev()
            .map(|record| {
                let pod = match PodRecord::load(&runs, &record.id) {
                    Ok(pod) => pod,
                    Err(error) => {
                        skipped.push(format!(
                            "cannot read the pod record of run {}: {error}",
                            record.id
                        ));
                        None
                    },
                };
                RunRow { record, pod }
            })
            .collect()),
        Err(error) => Err(format!("cannot list the runs: {error}")),
    };
    Listing { runs, skipped }
}

/// The Training view's state.
#[derive(Debug, Clone, Default, PartialEq)]
pub(super) struct TrainingView {
    /// The runs, newest first.
    pub(super) runs: Vec<RunRow>,
    /// The selected run, by position in `runs`.
    pub(super) selected: usize,
    /// Why the runs could not be listed at the last read; the runs shown are
    /// then those of an earlier read.
    pub(super) error: Option<String>,
    /// Metrics of each run shown, by run ID.
    pub(super) series: BTreeMap<String, Vec<TrainMetric>>,
    /// What the metric lines of each run shown say of its phase, by run ID;
    /// a run with a series but none reads its phase from the series alone.
    pub(super) phases: BTreeMap<String, Phases>,
    /// The training tasks running.
    pub(super) tasks: BTreeMap<TaskId, Follow>,
    /// How the last task of each run ended.
    pub(super) ended: BTreeMap<String, Ended>,
    /// The training tasks that ended, with their run, until another task
    /// starts on it: their messages handled after their end still count.
    pub(super) last: BTreeMap<TaskId, String>,
    /// The read of `runs/` running, if any.
    pub(super) listing: Option<TaskId>,
    /// Whether another read of `runs/` was asked for while one ran.
    pub(super) list_again: bool,
    /// The run to select once it is listed (a run just created).
    pub(super) wanted: Option<String>,
    /// The latest read of each run's local metrics, by run ID; an earlier one's
    /// result is ignored.
    pub(super) reading: BTreeMap<String, TaskId>,
    /// The listing warnings already logged: each is logged once.
    pub(super) warned: BTreeSet<String>,
    /// The runs left out of the list until the TUI restarts, by run ID: the
    /// failed runs cleared with `x`. Their files stay.
    pub(super) hidden: BTreeSet<String>,
    /// The runs whose pod is no longer shown, by run ID: dismissed with `p`.
    pub(super) dismissed: BTreeSet<String>,
    /// The last [`SYSTEM_SAMPLES`] samples of each run's machine, oldest first,
    /// by run ID. They stay once the task ends, until the TUI restarts.
    pub(super) system: BTreeMap<String, VecDeque<SystemSample>>,
    /// The preparation of a push running (`h`), if any.
    pub(super) push_prepare: Option<TaskId>,
    /// The push to the Hugging Face Hub running, if any: one at a time.
    pub(super) pushing: Option<Pushing>,
}

/// A push of a run to the Hugging Face Hub, started with `h`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct Pushing {
    /// Its task.
    pub(super) task: TaskId,
    /// The run pushed.
    pub(super) run: String,
    /// Bytes uploaded and bytes to upload, from its latest event.
    pub(super) progress: Option<(u64, u64)>,
}

/// The progress a series of metrics shows.
#[derive(Debug, Clone, Copy, PartialEq)]
pub(super) struct Progress {
    /// Steps done.
    pub(super) step: u64,
    /// Steps of the whole run.
    pub(super) max_steps: Option<u64>,
    /// Fractional epoch.
    pub(super) epoch: Option<f64>,
    /// Time left at the rate seen so far.
    pub(super) eta: Option<Duration>,
}

/// The progress of `series`: from its latest metric, with an ETA at its
/// [`Pace`].
pub(super) fn progress(series: &[TrainMetric]) -> Option<Progress> {
    let latest = series.last()?;
    let mut pace = Pace::default();
    for metric in series {
        pace.add(metric);
    }
    Some(Progress {
        step: latest.step,
        max_steps: pace.max_steps(),
        epoch: series.iter().rev().find_map(|m| m.epoch),
        eta: pace.eta(),
    })
}

/// `count` as a float, saturating at `u32::MAX`.
pub(super) fn float(count: u64) -> f64 {
    f64::from(u32::try_from(count).unwrap_or(u32::MAX))
}

/// The metrics in `runs/<id>/metrics.jsonl` of the project in `dir`, and what
/// its lines say of the run's phase, malformed lines skipped; `None` when the
/// file cannot be read.
pub(super) fn read_series(dir: &Path, id: &str) -> Option<(Vec<TrainMetric>, Phases)> {
    let path = Runs::new(dir).run_dir(id).ok()?.join(METRICS_FILE);
    let content = std::fs::read_to_string(path).ok()?;
    let mut series = Vec::new();
    let mut phases = Phases::default();
    for line in content.lines().filter_map(|line| parse_line(line).ok()) {
        phases.line(&line);
        if let MetricLine::Log(metric) = line {
            series.push(metric);
        }
    }
    Some((series, phases))
}

impl TrainingView {
    /// Shows the runs of `listing` but the hidden ones, keeping the selected
    /// run, or selecting the run wanted once it is listed. When it failed, the
    /// earlier runs stay, with its error. Returns the warnings not logged yet.
    pub(super) fn listed(&mut self, listing: Listing) -> Vec<String> {
        let selected = self.selected_run().map(|row| row.record.id.clone());
        match listing.runs {
            Ok(runs) => {
                self.runs = runs;
                let hidden = &self.hidden;
                self.runs.retain(|row| !hidden.contains(&row.record.id));
                self.error = None;
            },
            Err(error) => self.error = Some(error),
        }
        if let Some(id) = selected {
            self.select(&id);
        }
        if let Some(id) = self.wanted.take() {
            if self.runs.iter().any(|row| row.record.id == id) {
                self.select(&id);
            } else {
                self.wanted = Some(id);
            }
        }
        self.selected = self.selected.min(self.runs.len().saturating_sub(1));
        listing
            .skipped
            .into_iter()
            .filter(|warning| self.warned.insert(warning.clone()))
            .collect()
    }

    /// The failed runs no task is on: what `x` clears.
    pub(super) fn failed_runs(&self) -> Vec<String> {
        self.runs
            .iter()
            .filter(|row| row.record.state == RunState::Failed)
            .map(|row| row.record.id.clone())
            .filter(|id| self.task_of(id).is_none())
            .collect()
    }

    /// Leaves runs `ids` out of the list until the TUI restarts, keeping the
    /// selected run when it stays.
    pub(super) fn hide(&mut self, ids: &[String]) {
        let selected = self.selected_run().map(|row| row.record.id.clone());
        self.hidden.extend(ids.iter().cloned());
        let hidden = &self.hidden;
        self.runs.retain(|row| !hidden.contains(&row.record.id));
        if let Some(id) = selected {
            self.select(&id);
        }
        self.selected = self.selected.min(self.runs.len().saturating_sub(1));
    }

    /// Selects run `id`, when listed.
    pub(super) fn select(&mut self, id: &str) {
        if let Some(index) = self.runs.iter().position(|row| row.record.id == id) {
            self.selected = index;
        }
    }

    /// The selected run.
    pub(super) fn selected_run(&self) -> Option<&RunRow> {
        self.runs.get(self.selected)
    }

    /// The share of its steps the selected run did, when its metrics say how
    /// many it has.
    pub(super) fn selected_ratio(&self) -> Option<f64> {
        let row = self.selected_run()?;
        let now = progress(self.series.get(&row.record.id)?)?;
        let max = now.max_steps.filter(|max| *max > 0)?;
        Some((float(now.step) / float(max)).min(1.0))
    }

    /// What the job of `row` does, while its run is recorded running: from its
    /// metric lines, and from the latest job status of the task following it.
    pub(super) fn phase(&self, row: &RunRow) -> Phase {
        let id = &row.record.id;
        let exited = self
            .task_of(id)
            .and_then(|(_, follow)| follow.status)
            .is_some_and(JobStatus::is_finished)
            && row.record.state == RunState::Running;
        let phases = self.phases.get(id).copied().unwrap_or_else(|| {
            let mut phases = Phases::default();
            for metric in self.series.get(id).into_iter().flatten() {
                phases.metric(metric);
            }
            phases
        });
        phases.phase(exited)
    }

    /// What the task on run `id`, if any, does to it.
    pub(super) fn activity(&self, id: &str) -> RunActivity {
        RunActivity::of(self.task_of(id).map(|(_, follow)| follow))
    }

    /// What the task on the selected run, if any, does to it.
    pub(super) fn selected_activity(&self) -> RunActivity {
        self.selected_run()
            .map_or(RunActivity::None, |row| self.activity(&row.record.id))
    }

    /// The task following or cancelling run `id`, if any.
    pub(super) fn task_of(&self, id: &str) -> Option<(TaskId, &Follow)> {
        self.tasks
            .iter()
            .find(|(_, follow)| follow.run_id == id)
            .map(|(task, follow)| (*task, follow))
    }

    /// Whether task `id` is a training task, running or ended.
    pub(super) fn is_training(&self, id: TaskId) -> bool {
        self.tasks.contains_key(&id) || self.last.contains_key(&id)
    }

    /// The selected run, when its local metrics are to be read: not read yet,
    /// not being read, and followed by no task.
    pub(super) fn to_read(&self) -> Option<String> {
        let id = &self.selected_run()?.record.id;
        let skip = self.series.contains_key(id)
            || self.reading.contains_key(id)
            || self.task_of(id).is_some();
        (!skip).then(|| id.clone())
    }

    /// Whether run `id` is being cancelled: by a cancel task, or by one that
    /// detaches its task first.
    pub(super) fn cancelling(&self, id: &str) -> bool {
        self.activity(id) == RunActivity::Cancelling
    }

    /// Whether run `id` is being stopped with a snapshot.
    pub(super) fn stopping(&self, id: &str) -> bool {
        self.activity(id) == RunActivity::Stopping
    }

    /// Records `event` of task `id`. Returns whether it is the first job status
    /// of that task: its watch began.
    pub(super) fn event(&mut self, id: TaskId, event: Event) -> bool {
        let Some(follow) = self.tasks.get_mut(&id) else {
            return false;
        };
        match event {
            Event::Metric(metric) => {
                self.phases
                    .entry(follow.run_id.clone())
                    .or_default()
                    .metric(&metric);
                self.series
                    .entry(follow.run_id.clone())
                    .or_default()
                    .push(metric);
            },
            Event::Mark(mark) => {
                self.phases
                    .entry(follow.run_id.clone())
                    .or_default()
                    .mark(mark);
            },
            Event::JobStatus(status) => {
                follow.status = Some(status);
                if !follow.watching {
                    follow.watching = true;
                    return true;
                }
            },
            Event::PodStatus(status) => follow.pod = Some(status),
            Event::System(sample) => {
                let samples = self.system.entry(follow.run_id.clone()).or_default();
                samples.push_back(sample);
                while samples.len() > SYSTEM_SAMPLES {
                    samples.pop_front();
                }
            },
            _ => {},
        }
        false
    }
}

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

    fn log(time: f64, step: u64, loss: Option<f64>) -> TrainMetric {
        TrainMetric {
            time,
            step,
            epoch: Some(0.5),
            max_steps: Some(400),
            loss,
            eval_loss: loss.is_none().then_some(1.0),
            learning_rate: Some(2e-4),
            grad_norm: Some(0.8),
        }
    }

    #[test]
    fn the_eta_follows_the_training_logs_on_the_plugins_clock() -> Result<(), String> {
        let series = [
            log(100.0, 10, Some(2.0)),
            log(150.0, 60, None),
            log(200.0, 110, Some(1.5)),
        ];
        let now = progress(&series).ok_or("no progress")?;
        assert_eq!((now.step, now.max_steps), (110, Some(400)));
        assert_eq!(now.eta, Some(Duration::from_secs(290)));
        assert_eq!(progress(&series[..1]).and_then(|p| p.eta), None);
        assert_eq!(progress(&[]), None);
        Ok(())
    }

    #[test]
    fn one_activity_says_what_a_task_does_to_its_run() {
        let runpod = Job::Start { runpod: true };
        let task = |job: &Job, change: &dyn Fn(&mut Follow)| {
            let mut follow = Follow::new(job.clone(), "run");
            change(&mut follow);
            RunActivity::of(Some(&follow))
        };
        let keep = |_: &mut Follow| {};
        assert_eq!(RunActivity::of(None), RunActivity::None);
        assert_eq!(task(&Job::Attach, &keep), RunActivity::Followed);
        assert_eq!(task(&runpod, &|f| f.watching = true), RunActivity::Followed);
        assert_eq!(task(&runpod, &keep), RunActivity::Starting { runpod: true });
        assert_eq!(
            task(&Job::Start { runpod: false }, &keep),
            RunActivity::Starting { runpod: false }
        );
        assert_eq!(
            task(&runpod, &|f| f.detach = Detach::Done),
            RunActivity::Abandoning { runpod: true }
        );
        assert_eq!(task(&Job::Cancel, &keep), RunActivity::Cancelling);
        assert_eq!(
            task(&runpod, &|f| f.cancel_after = true),
            RunActivity::Cancelling
        );
        assert!(task(&runpod, &keep).abandons());
        assert!(task(&runpod, &|f| f.detach = Detach::Done).abandons());
        assert!(!task(&Job::Start { runpod: false }, &keep).abandons());
        assert!(!task(&Job::Attach, &keep).abandons());
        let labels: Vec<&str> = [
            RunActivity::None,
            RunActivity::Followed,
            RunActivity::Starting { runpod: true },
            RunActivity::Abandoning { runpod: true },
            RunActivity::Cancelling,
        ]
        .map(RunActivity::label)
        .to_vec();
        assert_eq!(
            labels,
            ["", "followed", "starting", "starting", "cancelling"]
        );
    }

    #[test]
    fn the_machine_of_a_run_keeps_its_last_samples() {
        let mut view = TrainingView::default();
        view.tasks
            .insert(TaskId(1), Follow::new(Job::Attach, "run-a"));
        let sample = |seconds| SystemSample {
            at: SystemTime::UNIX_EPOCH + Duration::from_secs(seconds),
            disks: Vec::new(),
            cpu: None,
            memory: None,
            gpus: Vec::new(),
        };
        for seconds in 0..70 {
            assert!(!view.event(TaskId(1), Event::System(sample(seconds))));
        }
        // An event of a task that is not running is dropped.
        view.event(TaskId(2), Event::System(sample(99)));
        let kept = view
            .system
            .get("run-a")
            .map(|s| s.iter().map(|x| x.at).collect::<Vec<_>>());
        let expected: Vec<SystemTime> = (10..70).map(sample).map(|s| s.at).collect();
        assert_eq!(kept, Some(expected));
        assert_eq!(view.system.len(), 1);
    }

    #[test]
    fn the_latest_job_status_and_marks_tell_the_phase() -> Result<(), serde_json::Error> {
        let mut view = TrainingView::default();
        let mut record: RunRecord = serde_json::from_value(serde_json::json!({
            "id": "run-a",
            "target": "box",
            "remote_dir": "/w/run-a",
            "state": "running",
            "created": "2026-09-21T14:00:00Z",
        }))?;
        view.runs.push(RunRow {
            record: record.clone(),
            pod: None,
        });
        view.tasks
            .insert(TaskId(1), Follow::new(Job::Attach, "run-a"));
        let row = |view: &TrainingView| view.runs[0].clone();
        assert!(view.event(TaskId(1), Event::JobStatus(JobStatus::Running)));
        assert!(!view.event(TaskId(1), Event::Metric(log(1.0, 400, Some(1.0)))));
        assert_eq!(view.phase(&row(&view)), Phase::Finalizing);
        view.event(
            TaskId(1),
            Event::Mark(crate::train::Mark::Stage(crate::train::JobStage::Merge)),
        );
        assert_eq!(view.phase(&row(&view)), Phase::Merging);
        assert!(!view.event(TaskId(1), Event::JobStatus(JobStatus::Exited(0))));
        assert_eq!(view.phase(&row(&view)), Phase::Retrieving);
        // Once the run is recorded ended, nothing is being retrieved.
        record.state = RunState::Succeeded;
        view.runs[0].record = record;
        assert_eq!(view.phase(&row(&view)), Phase::Merging);
        Ok(())
    }

    #[test]
    fn a_series_is_read_from_the_local_metrics_file() -> Result<(), Box<dyn std::error::Error>> {
        let dir = tempfile::tempdir()?;
        let run = dir.path().join("runs/20260921-140000-a1b2");
        std::fs::create_dir_all(&run)?;
        std::fs::write(
            run.join(METRICS_FILE),
            "{\"event\": \"begin\", \"time\": 1, \"max_steps\": 2}\nnot json\n\
             {\"event\": \"log\", \"time\": 2, \"step\": 1, \"loss\": 1.5}\n\
             {\"event\": \"end\", \"time\": 3, \"step\": 1}\n",
        )?;
        let series = read_series(dir.path(), "20260921-140000-a1b2");
        assert_eq!(
            series.map(|(s, phases)| (s.len(), phases.phase(false))),
            Some((1, Phase::Finalizing))
        );
        assert_eq!(read_series(dir.path(), "20260921-140000-ffff"), None);
        Ok(())
    }
}