From 9611f8515909a7bc0b08b3d3a5b16f23cb8a2838 Mon Sep 17 00:00:00 2001
From: "jiangfangqin.jfq" <jiangfangqin.jfq@digital-engine.com>
Date: Wed, 29 Jul 2026 15:10:45 +0800
Subject: [PATCH 1/8] refactor(metrics): extract shared TensorBoard read path
 to api/metric_reader

---
 areno/api/metric_reader.py | 230 +++++++++++++++++++++++++++++++++++++
 areno/dashboard/server.py  |  64 +++--------
 2 files changed, 249 insertions(+), 45 deletions(-)
 create mode 100644 areno/api/metric_reader.py

diff --git a/areno/api/metric_reader.py b/areno/api/metric_reader.py
new file mode 100644
index 00000000..9d6eae64
--- /dev/null
+++ b/areno/api/metric_reader.py
@@ -0,0 +1,230 @@
+"""Read-side helpers for querying metric history from local run artifacts.
+
+Issue #254 extracts the dashboard's TensorBoard scalar reading into a set of
+pure, side-effect-free functions so the ``areno metrics`` CLI and the dashboard
+share one fact source. Reading is CPU-only; the heavy ``tensorboard`` import is
+deferred to inside :func:`read_scalar_points`.
+
+Scope of the first version (issue #254 plan):
+  - First TLS data source only: ``events.out.tfevents.*`` scalars.
+  - ``--pid`` selects a run by filename pid suffix.
+  - jsonl fallback and ``--run <id>`` are follow-ups (tracked separately).
+
+Reading semantics mirror ``areno/dashboard/server.py``'s
+``_load_tensorboard_scalars`` byte-for-byte so a dashboard switch is
+behavior-preserving:
+  - ``EventAccumulator(size_guidance={"scalars": 10000})``
+  - ``accumulator.Scalars(tag)[-500:]``
+  - NaN values skipped
+  - ``(name, step, value)`` de-duplication
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+import math
+from pathlib import Path
+from typing import Any
+
+
+def now() -> str:
+    """UTC timestamp string, matching ``areno/dashboard/server.py``'s ``now``."""
+    return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat()
+
+
+def number_like(value: Any) -> bool:
+    """True if ``value`` is a usable finite number (mirrors server's check)."""
+    try:
+        return not math.isnan(float(value))
+    except (TypeError, ValueError):
+        return False
+
+
+def tensorboard_event_sources(path: Path, pid: int | None) -> list[Path]:
+    """Locate ``events.out.tfevents.*`` files under ``path``.
+
+    ``pid`` filters by the filename pid suffix when given; ``None`` merges every
+    event file in the directory. Falls back to ``[path]`` when no event files
+    exist so a fresh directory degrades to an empty read rather than a crash.
+    """
+    event_files = sorted(path.rglob("events.out.tfevents.*"), key=lambda item: item.stat().st_mtime)
+    if not event_files:
+        return [path]
+    if pid is None:
+        return event_files
+    pid_marker = f".{pid}."
+    return [file for file in event_files if pid_marker in file.name or file.parent.name == f"pid-{pid}"]
+
+
+def locate_event_files(metrics_dir: str | Path, pid: int | None = None) -> list[Path]:
+    """Public locator: resolve ``metrics_dir`` and return event file paths."""
+    return tensorboard_event_sources(Path(metrics_dir), pid)
+
+
+def read_scalar_points(metrics_dir: str | Path, pid: int | None = None) -> list[dict[str, Any]]:
+    """Read TensorBoard scalars into de-duplicated ``[{name, value, step, time}]``.
+
+    Iteration order, truncation (:code:`[-500:]`), NaN skip, and
+    ``(name, step, value)`` dedup match the dashboard read path exactly; callers
+    that feed these points back through the dashboard's ``_add_metric`` get
+    byte-identical ``job.metrics`` ordering.
+    """
+    path = Path(metrics_dir)
+    try:
+        from tensorboard.backend.event_processing.event_accumulator import EventAccumulator
+    except Exception:
+        return []
+
+    points: list[dict[str, Any]] = []
+    seen: set[tuple[str, int, float]] = set()
+    for accumulator_path in tensorboard_event_sources(path, pid):
+        try:
+            accumulator = EventAccumulator(str(accumulator_path), size_guidance={"scalars": 10000})
+            accumulator.Reload()
+            tags = accumulator.Tags().get("scalars", [])
+        except Exception:
+            continue
+        for tag in tags:
+            try:
+                events = accumulator.Scalars(tag)[-500:]
+            except Exception:
+                continue
+            for event in events:
+                step = int(event.step)
+                value = float(event.value)
+                if math.isnan(value):
+                    continue
+                key = (tag, step, value)
+                if key in seen:
+                    continue
+                seen.add(key)
+                points.append({"name": tag, "value": value, "step": step, "time": now()})
+    return points
+
+
+def list_available_tags(points: list[dict[str, Any]]) -> list[str]:
+    """Return the distinct metric names found in ``points``, sorted for display."""
+    names: set[str] = set()
+    for point in points:
+        name = str(point.get("name") or "")
+        if name:
+            names.add(name)
+    return sorted(names)
+
+
+def summarize_metric(
+    points: list[dict[str, Any]], name: str, *, recent_n: int = 20
+) -> dict[str, Any]:
+    """Aggregate one metric into ``{name, count, last, min, max, recent, trend}``.
+
+    - ``count``: number of finite points for ``name`` (NaN-free, since the reader
+      already skips NaN).
+    - ``last``: value at the highest step.
+    - ``min``/``max``: streaming single-pass over all points (O(1) memory,
+      independent of the ``[-500:]`` truncation).
+    - ``recent``: the last ``recent_n`` values in step order.
+    - ``trend``: the full bounded series normalized to ``[0, 1]``; ``render_table``
+      maps it to a UTF-8 sparkline, ``render_json`` returns it verbatim.
+    """
+    series = sorted(
+        (point for point in points if point.get("name") == name and number_like(point.get("value"))),
+        key=lambda point: int(point.get("step") or 0),
+    )
+    values = [float(point.get("value")) for point in series]
+    count = len(values)
+    if count == 0:
+        last = min_v = max_v = None
+        recent: list[float] = []
+        trend: list[float] = []
+    else:
+        last = values[-1]
+        min_v = max_v = values[0]
+        for value in values:
+            if value < min_v:
+                min_v = value
+            if value > max_v:
+                max_v = value
+        last_step = max(int(point.get("step") or 0) for point in series)
+        for point in series:
+            if int(point.get("step") or 0) == last_step:
+                last = float(point.get("value"))
+        recent = values[-recent_n:] if recent_n > 0 else []
+        trend = _normalize(values)
+
+    return {
+        "name": name,
+        "count": count,
+        "last": last,
+        "min": min_v,
+        "max": max_v,
+        "recent": recent,
+        "trend": trend,
+    }
+
+
+def _normalize(values: list[float]) -> list[float]:
+    """Scale ``values`` to ``[0, 1]``; a flat series maps to ``0.5`` everywhere."""
+    if not values:
+        return []
+    low = min(values)
+    high = max(values)
+    if high == low:
+        return [0.5 for _ in values]
+    span = high - low
+    return [(value - low) / span for value in values]
+
+
+_SPARKS = "▁▂▃▄▅▆▇█"
+
+
+def render_table(summary: dict[str, Any]) -> str:
+    """Render a summary as text: header row + values + a hand-written sparkline.
+
+    No external dependency (``rich``/``sparklines``) -- the trend is mapped to the
+    8-glyph ``▁▂▃▄▅▆▇█`` rung by normalized value.
+    """
+    name = summary.get("name", "")
+    count = summary.get("count", 0)
+    trend = summary.get("trend") or []
+    sparkline = _sparkline(trend)
+    lines = [
+        f"metric   {name}",
+        f"count    {count}",
+        f"last     {_fmt(summary.get('last'))}",
+        f"min      {_fmt(summary.get('min'))}",
+        f"max      {_fmt(summary.get('max'))}",
+        f"trend    {sparkline}",
+        f"recent   {', '.join(_fmt(value) for value in (summary.get('recent') or []))}",
+    ]
+    return "\n".join(lines)
+
+
+def _sparkline(trend: list[float]) -> str:
+    if not trend:
+        return ""
+    glyphs = []
+    for value in trend:
+        index = min(len(_SPARKS) - 1, max(0, int(value * len(_SPARKS))))
+        glyphs.append(_SPARKS[index])
+    return "".join(glyphs)
+
+
+def _fmt(value: Any) -> str:
+    if value is None:
+        return "-"
+    if isinstance(value, float):
+        return f"{value:.6g}"
+    return str(value)
+
+
+def render_json(summary: dict[str, Any]) -> dict[str, Any]:
+    """Return the summary as a JSON-serializable dict (trend = normalized array)."""
+    return {
+        "name": summary.get("name"),
+        "count": summary.get("count"),
+        "last": summary.get("last"),
+        "min": summary.get("min"),
+        "max": summary.get("max"),
+        "recent": summary.get("recent"),
+        "trend": summary.get("trend"),
+    }
\ No newline at end of file
diff --git a/areno/dashboard/server.py b/areno/dashboard/server.py
index 9bb8b4d4..4c6b26f1 100644
--- a/areno/dashboard/server.py
+++ b/areno/dashboard/server.py
@@ -23,6 +23,7 @@
 from typing import Any
 from uuid import uuid4
 
+from areno.api.metric_reader import number_like
 from areno.cli.dashboard_registry import GLOBAL_REGISTRY_FILE
 from areno.cli.diagnostics import collect_env, run_checks
 from areno.dashboard.agent_context import agent_system_prompt
@@ -317,36 +318,27 @@ def _load_tensorboard_scalars(self, job: Job, path: Path) -> None:
             from tensorboard.backend.event_processing.event_accumulator import EventAccumulator
         except Exception:
             return
+        # Lazy import: read_scalar_points imports TensorBoard's EventAccumulator
+        # internally, so keep it out of module import time.
+        from areno.api.metric_reader import read_scalar_points
+
         job.timeperf = [row for row in job.timeperf if row.get("source") != "metrics"]
         job._timeperf_keys = {int(row.get("step", -1)) for row in job.timeperf}
         by_step: dict[int, dict[str, float]] = {}
-        for accumulator_path in tensorboard_event_sources(path, job_pid(job)):
-            try:
-                accumulator = EventAccumulator(str(accumulator_path), size_guidance={"scalars": 10000})
-                accumulator.Reload()
-                tags = accumulator.Tags().get("scalars", [])
-            except Exception:
-                continue
-            for tag in tags:
-                try:
-                    events = accumulator.Scalars(tag)[-500:]
-                except Exception:
-                    continue
-                for event in events:
-                    step = int(event.step)
-                    value = float(event.value)
-                    if math.isnan(value):
-                        continue
-                    self._add_metric(job, tag, value, step)
-                    time_name = tensorboard_time_segment_name(tag)
-                    if time_name:
-                        by_step.setdefault(step, {})[time_name] = value
-                    if tag in {"train/step_e2e_time_s", "time/total", "time/e2e"}:
-                        by_step.setdefault(step, {})["total"] = value
-                    elif tag == "train/step_rollout_time_s":
-                        by_step.setdefault(step, {})["rollout"] = value
-                    elif tag in {"train/step_train_time_s", "train/policy_train_wall_time_s"}:
-                        by_step.setdefault(step, {})["train"] = value
+        for point in read_scalar_points(path, job_pid(job)):
+            name = str(point.get("name") or "")
+            step = int(point.get("step") or 0)
+            value = float(point.get("value"))
+            self._add_metric(job, name, value, step)
+            time_name = tensorboard_time_segment_name(name)
+            if time_name:
+                by_step.setdefault(step, {})[time_name] = value
+            if name in {"train/step_e2e_time_s", "time/total", "time/e2e"}:
+                by_step.setdefault(step, {})["total"] = value
+            elif name == "train/step_rollout_time_s":
+                by_step.setdefault(step, {})["rollout"] = value
+            elif name in {"train/step_train_time_s", "train/policy_train_wall_time_s"}:
+                by_step.setdefault(step, {})["train"] = value
         for step, values in sorted(by_step.items()):
             total = values.pop("total", None)
             if total is None:
@@ -796,24 +788,6 @@ def job_pid(job: Job) -> int | None:
     return job.pid
 
 
-def number_like(value: Any) -> bool:
-    try:
-        return not math.isnan(float(value))
-    except (TypeError, ValueError):
-        return False
-
-
-def tensorboard_event_sources(path: Path, pid: int | None) -> list[Path]:
-    event_files = sorted(path.rglob("events.out.tfevents.*"), key=lambda item: item.stat().st_mtime)
-    if not event_files:
-        return [path]
-    if pid is None:
-        return event_files
-    pid_marker = f".{pid}."
-    matched = [file for file in event_files if pid_marker in file.name or file.parent.name == f"pid-{pid}"]
-    return matched
-
-
 def rollout_sample_sources(path: Path, pid: int | None) -> list[Path]:
     if pid is not None:
         candidates = [

From 875564c27295679b3c7e186838fda807c26201df Mon Sep 17 00:00:00 2001
From: "jiangfangqin.jfq" <jiangfangqin.jfq@digital-engine.com>
Date: Wed, 29 Jul 2026 15:11:50 +0800
Subject: [PATCH 2/8] feat(cli): add 'areno metrics'

---
 areno/cli/main.py      |   1 +
 areno/cli/metrics.py   | 126 +++++++++++++++++++++++++++++++++++++++++
 docs/cli/metrics.rst   | 115 +++++++++++++++++++++++++++++++++++++
 docs/reference/cli.rst |   1 +
 4 files changed, 243 insertions(+)
 create mode 100644 areno/cli/metrics.py
 create mode 100644 docs/cli/metrics.rst

diff --git a/areno/cli/main.py b/areno/cli/main.py
index 8a6dc0bd..775a869b 100644
--- a/areno/cli/main.py
+++ b/areno/cli/main.py
@@ -21,6 +21,7 @@ class ArenoCli(click.Group):
         "dashboard": ("areno.cli.dashboard", "dashboard_command", "Start or stop the AReno React dashboard."),
         "train": ("areno.cli.train", "train_command", "Run SFT, DPO, GSPO, GRPO, or PPO training."),
         "serve": ("areno.cli.serve", "serve_command", "Serve an OpenAI-compatible chat API."),
+        "metrics": ("areno.cli.metrics", "metrics_command", "Query metric history from local run artifacts."),
     }
 
     def list_commands(self, ctx: click.Context) -> list[str]:
diff --git a/areno/cli/metrics.py b/areno/cli/metrics.py
new file mode 100644
index 00000000..5f80ecc3
--- /dev/null
+++ b/areno/cli/metrics.py
@@ -0,0 +1,126 @@
+"""``areno metrics`` -- query metric history from local run artifacts.
+
+Issue #254 adds a read-only CLI that summarizes one metric (last/min/max/
+recent/trend) from the ``events.out.tfevents.*`` artifacts a run writes. The
+heavywork lives in the light, pure :mod:`areno.api.metric_reader` module; this
+file only wires Click options, renders, and turns not-found cases into a clear
+error plus the list of available tags.
+
+The command never writes or starts anything -- it only reads artifacts.
+"""
+
+from __future__ import annotations
+
+import json as _json
+from pathlib import Path
+from typing import Any
+
+import click
+
+from areno.api import metric_reader
+
+
+@click.command(name="metrics", context_settings={"help_option_names": ["-h", "--help"]})
+@click.option(
+    "--metrics-dir",
+    "metrics_dir",
+    default=None,
+    help=(
+        "Directory holding the run's events.out.tfevents.* artifacts. "
+        "Omit to use areno's default metrics log dir."
+    ),
+)
+@click.option(
+    "--pid",
+    type=int,
+    default=None,
+    help="Filter event files by the pid suffix in the filename; merge every run when omitted.",
+)
+@click.option(
+    "--name",
+    "name",
+    default=None,
+    help="Metric tag to summarize (e.g. rollout/rewards_mean). Omit to list available tags.",
+)
+@click.option(
+    "--limit",
+    default=20,
+    show_default=True,
+    help="Number of recent values to include in recent/trend.",
+)
+@click.option("--json", "as_json", is_flag=True, help="Emit a machine-readable JSON object.")
+def metrics_command(
+    metrics_dir: str | None,
+    pid: int | None,
+    name: str | None,
+    limit: int,
+    as_json: bool,
+) -> None:
+    """Query metric history (last/min/max/recent/trend) from local run artifacts."""
+    if metrics_dir is None:
+        # Lazy import keeps the CLI module light; the default lives with the
+        # dashboard server as the single source (no duplicated absolute path).
+        from areno.dashboard.server import DEFAULT_METRICS_LOG_DIR
+
+        metrics_dir = DEFAULT_METRICS_LOG_DIR
+
+    # Surface a missing directory early with a clear, located error -- do not let
+    # the reader silently degrade an empty path into an empty result.
+    directory = Path(metrics_dir)
+    if not directory.exists():
+        raise click.ClickException(f"metrics dir not found: {metrics_dir} (pid={pid})")
+
+    points = metric_reader.read_scalar_points(metrics_dir, pid)
+    tags = metric_reader.list_available_tags(points)
+
+    # No tag requested: list what's available so the user can pick one.
+    if not name:
+        _emit_available(tags, metrics_dir, pid, as_json=as_json)
+        return
+
+    if name not in tags:
+        _emit_available(tags, metrics_dir, pid, as_json=as_json, missing=name)
+        raise click.ClickException(f"metric name not found: {name} (in {metrics_dir}, pid={pid})")
+
+    summary = metric_reader.summarize_metric(points, name, recent_n=limit)
+    if as_json:
+        click.echo(_json.dumps(metric_reader.render_json(summary), indent=2, sort_keys=True))
+    else:
+        click.echo(metric_reader.render_table(summary))
+
+
+def _emit_available(
+    tags: list[str],
+    metrics_dir: str,
+    pid: int | None,
+    *,
+    as_json: bool,
+    missing: str | None = None,
+) -> None:
+    """Print the available metric tags for the chosen run.
+
+    ``missing`` is set when emitting the list as part of a not-found error so
+    the user sees both the unknown name they asked for and what they can use.
+    """
+    if as_json:
+        payload: dict[str, Any] = {
+            "metrics_dir": metrics_dir,
+            "pid": pid,
+            "available_tags": tags,
+        }
+        if missing is not None:
+            payload["missing"] = missing
+        click.echo(_json.dumps(payload, indent=2, sort_keys=True))
+        return
+
+    label = (
+        f"metric '{missing}' not found in {metrics_dir} (pid={pid}); available tags"
+        if missing is not None
+        else f"available metric tags in {metrics_dir} (pid={pid})"
+    )
+    if not tags:
+        click.echo(f"{label}: (none found)")
+        return
+    click.echo(label)
+    for tag in tags:
+        click.echo(f"  {tag}")
\ No newline at end of file
diff --git a/docs/cli/metrics.rst b/docs/cli/metrics.rst
new file mode 100644
index 00000000..66b86381
--- /dev/null
+++ b/docs/cli/metrics.rst
@@ -0,0 +1,115 @@
+:orphan:
+
+Metrics CLI reference
+=====================
+
+``areno metrics``
+
+Query metric history from local run artifacts -- read the ``events.out.tfevents.*``
+files a training run writes and summarize one metric into ``last`` / ``min`` /
+``max`` / ``recent`` values / a compact ``trend``. The command is read-only: it
+never writes, starts, or contacts training/serving processes. It is the
+command-line counterpart of the dashboard's metric view and shares the same
+read-side code, so the two never drift.
+
+.. code-block:: bash
+
+   areno train --ckpt Qwen/Qwen3-0.6B --dataset-path gsm8k:main \
+     --reward-fn-path examples/math/math_verify_reward.py --algo gspo \
+     --metrics-log-dir /tmp/areno/tfevent
+
+   areno metrics --metrics-dir /tmp/areno/tfevent --name rollout/rewards_mean
+
+Output (text table with a hand-written UTF-8 sparkline trend):
+
+.. code-block:: text
+
+   metric   rollout/rewards_mean
+   count    5
+   last     1
+   min      0.5
+   max      1
+   trend    ▁▃▅▇█
+   recent   0.5, 0.625, 0.75, 0.875, 1
+
+For structured output (piped into ``jq`` or other tools):
+
+.. code-block:: bash
+
+   areno metrics --metrics-dir /tmp/areno/tfevent \
+     --name rollout/rewards_mean --json
+
+.. code-block:: json
+
+   {
+     "count": 5,
+     "last": 1.0,
+     "max": 1.0,
+     "min": 0.5,
+     "name": "rollout/rewards_mean",
+     "recent": [0.5, 0.625, 0.75, 0.875, 1.0],
+     "trend": [0.0, 0.25, 0.5, 0.75, 1.0]
+   }
+
+areno metrics
+-------------
+
+Summarize one metric from local run artifacts.
+
+Options:
+
+``--metrics-dir TEXT``
+   Directory holding the run's ``events.out.tfevents.*`` artifacts. Omit to use
+   areno's default metrics log dir (the same ``--metrics-log-dir`` training
+   writes to by default).
+
+``--pid INTEGER``
+   Filter event files by the ``pid`` suffix embedded in the filename
+   (``events.out.tfevents.<host>.<pid>.*``). Omit to merge every run writing
+   into the same directory.
+
+``--name TEXT``
+   Metric tag to summarize, for example ``rollout/rewards_mean``. Omit to list
+   the available tags found in the directory. When the tag is not found, the
+   command exits non-zero and prints the available tags so you can pick one.
+
+``--limit INTEGER``
+   Number of recent values to include in ``recent`` and ``trend``. Defaults to
+   ``20``.
+
+``--json``
+   Emit a machine-readable JSON object instead of the text table. The ``trend``
+   field is the normalized ``[0, 1]`` series; the text table renders the same
+   series as a UTF-8 sparkline.
+
+Input contract
+--------------
+
+* Source: TensorBoard scalar events (``events.out.tfevents.*``). Each tag is
+  truncated to its last ``500`` points (``EventAccumulator`` with
+  ``size_guidance={"scalars": 10000}``), ``NaN`` values are skipped, and
+  ``(name, step, value)`` triples are de-duplicated.
+* ``min`` / ``max`` are computed in a single streaming pass over the bounded
+  series, so memory stays bounded regardless of run length.
+* ``rollout_samples.*.jsonl`` files are **not** read as metrics; only
+  ``events.out.tfevents.*`` is.
+
+Limitations (first version)
+---------------------------
+
+* Only TensorBoard scalars are read. A jsonl fallback source and a friendly
+  ``--run <id>`` selector backed by the dashboard jobs registry are planned as
+  follow-ups; use ``--pid`` today to disambiguate runs sharing one directory.
+* The command is read-only and CPU-only; it does not require CUDA.
+
+Examples
+--------
+
+List the metrics available from the default log dir::
+
+   areno metrics
+
+Query one metric from a specific run identified by pid::
+
+   areno metrics --metrics-dir /tmp/areno/tfevent --pid 12345 \
+     --name rollout/rewards_mean --limit 50
\ No newline at end of file
diff --git a/docs/reference/cli.rst b/docs/reference/cli.rst
index 7e7b7e88..4e5a9200 100644
--- a/docs/reference/cli.rst
+++ b/docs/reference/cli.rst
@@ -11,4 +11,5 @@ Command pages:
 * :doc:`/cli/agent`
 * :doc:`/cli/dataset_loaders`
 * :doc:`/cli/observability`
+* :doc:`/cli/metrics`
 * :doc:`/cli/diagnostics`

From 9c3b521665cde27ca955af4bd13dd01502b62eac Mon Sep 17 00:00:00 2001
From: "jiangfangqin.jfq" <jiangfangqin.jfq@digital-engine.com>
Date: Wed, 29 Jul 2026 17:26:52 +0800
Subject: [PATCH 3/8] fix(metrics): bound trend/sparkline by --limit

---
 areno/api/metric_reader.py | 10 ++++++----
 1 file changed, 6 insertions(+), 4 deletions(-)

diff --git a/areno/api/metric_reader.py b/areno/api/metric_reader.py
index 9d6eae64..3cabceb7 100644
--- a/areno/api/metric_reader.py
+++ b/areno/api/metric_reader.py
@@ -123,8 +123,9 @@ def summarize_metric(
     - ``min``/``max``: streaming single-pass over all points (O(1) memory,
       independent of the ``[-500:]`` truncation).
     - ``recent``: the last ``recent_n`` values in step order.
-    - ``trend``: the full bounded series normalized to ``[0, 1]``; ``render_table``
-      maps it to a UTF-8 sparkline, ``render_json`` returns it verbatim.
+    - ``trend``: the last ``recent_n`` values normalized to ``[0, 1]`` -- the same
+      window as ``recent`` -- so ``--limit`` bounds the sparkline length;
+      ``render_table`` maps it to a UTF-8 sparkline, ``render_json`` returns it verbatim.
     """
     series = sorted(
         (point for point in points if point.get("name") == name and number_like(point.get("value"))),
@@ -148,8 +149,9 @@ def summarize_metric(
         for point in series:
             if int(point.get("step") or 0) == last_step:
                 last = float(point.get("value"))
-        recent = values[-recent_n:] if recent_n > 0 else []
-        trend = _normalize(values)
+        window = values[-recent_n:] if recent_n > 0 else []
+        recent = window
+        trend = _normalize(window)
 
     return {
         "name": name,

From 0a050f41cee1d80f1c3dff4a98befdd2fed8d4e0 Mon Sep 17 00:00:00 2001
From: "jiangfangqin.jfq" <jiangfangqin.jfq@digital-engine.com>
Date: Wed, 29 Jul 2026 18:08:59 +0800
Subject: [PATCH 4/8] feat(metrics): expose step + mean in areno metrics
 summary

The metrics summary now carries training progress alongside values, so the
log reads at a glance:
- new summary fields: last_step, min_step, max_step, mean, recent_steps
- table: a 'steps A -> B' line, a mean line, 'last V (step N)', and recent
  paired with absolute steps; the hand-written sparkline trend is unchanged
- JSON: the five new keys, additive only (recent/trend shape preserved)

Aggregation stays a single streaming pass (O(1) memory); min_step/max_step
span the retained [-500:] window, not full training history. The dashboard
read-side and its equivalence guard are untouched -- the new fields live in
the CLI-only summarize/render layer.

Co-Authored-By: Claude <noreply@anthropic.com>
---
 areno/api/metric_reader.py | 94 ++++++++++++++++++++++++++++++--------
 docs/cli/metrics.rst       | 49 +++++++++++++++-----
 2 files changed, 114 insertions(+), 29 deletions(-)

diff --git a/areno/api/metric_reader.py b/areno/api/metric_reader.py
index 3cabceb7..ab521592 100644
--- a/areno/api/metric_reader.py
+++ b/areno/api/metric_reader.py
@@ -115,51 +115,75 @@ def list_available_tags(points: list[dict[str, Any]]) -> list[str]:
 def summarize_metric(
     points: list[dict[str, Any]], name: str, *, recent_n: int = 20
 ) -> dict[str, Any]:
-    """Aggregate one metric into ``{name, count, last, min, max, recent, trend}``.
+    """Aggregate one metric into a summary dict.
+
+    Existing contract (unchanged, additive fields below):
+    ``{name, count, last, min, max, recent, trend}`` plus the new step-aware
+    ``last_step``/``min_step``/``max_step``/``mean``/``recent_steps``.
 
     - ``count``: number of finite points for ``name`` (NaN-free, since the reader
       already skips NaN).
-    - ``last``: value at the highest step.
-    - ``min``/``max``: streaming single-pass over all points (O(1) memory,
-      independent of the ``[-500:]`` truncation).
-    - ``recent``: the last ``recent_n`` values in step order.
-    - ``trend``: the last ``recent_n`` values normalized to ``[0, 1]`` -- the same
-      window as ``recent`` -- so ``--limit`` bounds the sparkline length;
-      ``render_table`` maps it to a UTF-8 sparkline, ``render_json`` returns it verbatim.
+    - ``last`` / ``last_step``: value and step at the highest step.
+    - ``min``/``max``/``mean``: streaming single-pass over all retained points
+      (O(1) memory, independent of the ``[-500:]`` truncation).
+    - ``min_step``/``max_step``: the step range of the **retained window**
+      (the reader keeps ``accumulator.Scalars(tag)[-500:]``). For runs longer
+      than 500 steps ``min_step`` is the earliest step still in that tail, **not**
+      the start of training; ``max_step`` is always the latest available step.
+    - ``recent`` / ``recent_steps``: the last ``recent_n`` values and the matching
+      steps, in step order (parallel arrays of equal length).
+    - ``trend``: ``recent`` normalized to ``[0, 1]`` -- the same window as
+      ``recent`` -- so ``--limit`` bounds the sparkline length; ``render_table``
+      maps it to a UTF-8 sparkline, ``render_json`` returns it verbatim.
     """
     series = sorted(
         (point for point in points if point.get("name") == name and number_like(point.get("value"))),
         key=lambda point: int(point.get("step") or 0),
     )
+    steps = [int(point.get("step") or 0) for point in series]
     values = [float(point.get("value")) for point in series]
     count = len(values)
     if count == 0:
-        last = min_v = max_v = None
+        last = min_v = max_v = mean = None
+        last_step = min_step = max_step = None
         recent: list[float] = []
+        recent_steps: list[int] = []
         trend: list[float] = []
     else:
         last = values[-1]
         min_v = max_v = values[0]
+        total = 0.0
         for value in values:
             if value < min_v:
                 min_v = value
             if value > max_v:
                 max_v = value
-        last_step = max(int(point.get("step") or 0) for point in series)
-        for point in series:
-            if int(point.get("step") or 0) == last_step:
-                last = float(point.get("value"))
+            total += value
+        mean = total / count
+        last_step = max(steps)
+        min_step = min(steps)
+        max_step = last_step
+        for idx, step in enumerate(steps):
+            if step == last_step:
+                last = values[idx]
         window = values[-recent_n:] if recent_n > 0 else []
+        window_steps = steps[-recent_n:] if recent_n > 0 else []
         recent = window
+        recent_steps = window_steps
         trend = _normalize(window)
 
     return {
         "name": name,
         "count": count,
         "last": last,
+        "last_step": last_step,
         "min": min_v,
         "max": max_v,
+        "mean": mean,
+        "min_step": min_step,
+        "max_step": max_step,
         "recent": recent,
+        "recent_steps": recent_steps,
         "trend": trend,
     }
 
@@ -180,27 +204,50 @@ def _normalize(values: list[float]) -> list[float]:
 
 
 def render_table(summary: dict[str, Any]) -> str:
-    """Render a summary as text: header row + values + a hand-written sparkline.
+    """Render a summary as text: header rows + a hand-written sparkline.
 
     No external dependency (``rich``/``sparklines``) -- the trend is mapped to the
-    8-glyph ``▁▂▃▄▅▆▇█`` rung by normalized value.
+    8-glyph ``▁▂▃▄▅▆▇█`` rung by normalized value. Step-aware fields are shown so a
+    human can read training progress (``steps``/``last``/``recent``) at a glance.
     """
     name = summary.get("name", "")
     count = summary.get("count", 0)
     trend = summary.get("trend") or []
     sparkline = _sparkline(trend)
+    last = _fmt(summary.get("last"))
+    last_step = summary.get("last_step")
+    if last_step is not None and summary.get("last") is not None:
+        last = f"{last} (step {last_step})"
+    steps = _fmt_range(summary.get("min_step"), summary.get("max_step"))
+    recent = summary.get("recent") or []
+    recent_steps = summary.get("recent_steps") or []
+    if recent_steps and len(recent_steps) == len(recent):
+        recent_str = ", ".join(f"step {s}: {_fmt(v)}" for s, v in zip(recent_steps, recent))
+    else:
+        recent_str = ", ".join(_fmt(value) for value in recent)
     lines = [
         f"metric   {name}",
         f"count    {count}",
-        f"last     {_fmt(summary.get('last'))}",
+        f"steps    {steps}",
+        f"last     {last}",
         f"min      {_fmt(summary.get('min'))}",
         f"max      {_fmt(summary.get('max'))}",
+        f"mean     {_fmt(summary.get('mean'))}",
         f"trend    {sparkline}",
-        f"recent   {', '.join(_fmt(value) for value in (summary.get('recent') or []))}",
+        f"recent   {recent_str}",
     ]
     return "\n".join(lines)
 
 
+def _fmt_range(low: Any, high: Any) -> str:
+    """Format a step range as ``low -> high`` (``-`` when either end is missing)."""
+    if low is None or high is None:
+        return "-"
+    if low == high:
+        return str(low)
+    return f"{low} -> {high}"
+
+
 def _sparkline(trend: list[float]) -> str:
     if not trend:
         return ""
@@ -220,13 +267,24 @@ def _fmt(value: Any) -> str:
 
 
 def render_json(summary: dict[str, Any]) -> dict[str, Any]:
-    """Return the summary as a JSON-serializable dict (trend = normalized array)."""
+    """Return the summary as a JSON-serializable dict (trend = normalized array).
+
+    The step-aware fields (``last_step``/``min_step``/``max_step``/``mean``/
+    ``recent_steps``) are exposed so machine consumers can read training progress
+    and pair recent values with their steps (``jq '.recent_steps, .recent'``).
+    Existing keys keep their shape; the additions are purely append-only.
+    """
     return {
         "name": summary.get("name"),
         "count": summary.get("count"),
         "last": summary.get("last"),
+        "last_step": summary.get("last_step"),
         "min": summary.get("min"),
         "max": summary.get("max"),
+        "mean": summary.get("mean"),
+        "min_step": summary.get("min_step"),
+        "max_step": summary.get("max_step"),
         "recent": summary.get("recent"),
+        "recent_steps": summary.get("recent_steps"),
         "trend": summary.get("trend"),
     }
\ No newline at end of file
diff --git a/docs/cli/metrics.rst b/docs/cli/metrics.rst
index 66b86381..d5c99c05 100644
--- a/docs/cli/metrics.rst
+++ b/docs/cli/metrics.rst
@@ -7,10 +7,13 @@ Metrics CLI reference
 
 Query metric history from local run artifacts -- read the ``events.out.tfevents.*``
 files a training run writes and summarize one metric into ``last`` / ``min`` /
-``max`` / ``recent`` values / a compact ``trend``. The command is read-only: it
-never writes, starts, or contacts training/serving processes. It is the
-command-line counterpart of the dashboard's metric view and shares the same
-read-side code, so the two never drift.
+``max`` / ``mean`` / a ``steps`` range / ``recent`` values / a compact ``trend``.
+Step-aware fields (``last_step``, ``min_step``, ``max_step``, ``recent_steps``)
+let you read training progress at a glance and pair each recent value with its
+absolute step. The command is read-only: it never writes, starts, or contacts
+training/serving processes. It is the command-line counterpart of the
+dashboard's metric view and shares the same read-side code, so the two never
+drift.
 
 .. code-block:: bash
 
@@ -26,11 +29,13 @@ Output (text table with a hand-written UTF-8 sparkline trend):
 
    metric   rollout/rewards_mean
    count    5
-   last     1
+   steps    0 -> 4
+   last     1 (step 4)
    min      0.5
    max      1
+   mean     0.75
    trend    ▁▃▅▇█
-   recent   0.5, 0.625, 0.75, 0.875, 1
+   recent   step 0: 0.5, step 1: 0.625, step 2: 0.75, step 3: 0.875, step 4: 1
 
 For structured output (piped into ``jq`` or other tools):
 
@@ -44,10 +49,15 @@ For structured output (piped into ``jq`` or other tools):
    {
      "count": 5,
      "last": 1.0,
+     "last_step": 4,
      "max": 1.0,
+     "max_step": 4,
+     "mean": 0.75,
      "min": 0.5,
+     "min_step": 0,
      "name": "rollout/rewards_mean",
      "recent": [0.5, 0.625, 0.75, 0.875, 1.0],
+     "recent_steps": [0, 1, 2, 3, 4],
      "trend": [0.0, 0.25, 0.5, 0.75, 1.0]
    }
 
@@ -74,13 +84,16 @@ Options:
    command exits non-zero and prints the available tags so you can pick one.
 
 ``--limit INTEGER``
-   Number of recent values to include in ``recent`` and ``trend``. Defaults to
-   ``20``.
+   Number of recent values to include in ``recent`` / ``recent_steps`` and
+   ``trend``. Defaults to ``20``.
 
 ``--json``
    Emit a machine-readable JSON object instead of the text table. The ``trend``
    field is the normalized ``[0, 1]`` series; the text table renders the same
-   series as a UTF-8 sparkline.
+   series as a UTF-8 sparkline. JSON also exposes ``last_step`` / ``min_step`` /
+   ``max_step`` / ``mean`` / ``recent_steps`` so machine consumers can read
+   training progress and pair each recent value with its step
+   (``jq '.max_step, (.recent_steps, .recent)'``).
 
 Input contract
 --------------
@@ -89,11 +102,25 @@ Input contract
   truncated to its last ``500`` points (``EventAccumulator`` with
   ``size_guidance={"scalars": 10000}``), ``NaN`` values are skipped, and
   ``(name, step, value)`` triples are de-duplicated.
-* ``min`` / ``max`` are computed in a single streaming pass over the bounded
-  series, so memory stays bounded regardless of run length.
+* ``min`` / ``max`` / ``mean`` are computed in a single streaming pass over the
+  bounded series, so memory stays bounded regardless of run length.
 * ``rollout_samples.*.jsonl`` files are **not** read as metrics; only
   ``events.out.tfevents.*`` is.
 
+Step range and training progress
+--------------------------------
+
+* ``max_step`` (and the high end of the text ``steps`` line) is the latest step
+  available in the artifacts -- i.e. how far training has progressed.
+* ``min_step`` / ``max_step`` describe the **retained window**, not necessarily
+  the full training history. Because each tag is truncated to its last ``500``
+  points, a run longer than ``500`` steps reports ``min_step`` as the earliest
+  step still in that tail, **not** the start of training. ``max_step`` is always
+  the current progress.
+* ``recent_steps`` is parallel to ``recent`` (equal length, step order), so each
+  recent value maps back to its absolute step -- useful when correlating with
+  training logs.
+
 Limitations (first version)
 ---------------------------
 

From 9adeb3008d71b4d3dd9b822920ed953100251506 Mon Sep 17 00:00:00 2001
From: "jiangfangqin.jfq" <jiangfangqin.jfq@digital-engine.com>
Date: Wed, 29 Jul 2026 20:55:26 +0800
Subject: [PATCH 5/8] refactor(metrics): single-source DEFAULT_METRICS_LOG_DIR,
 drop dead imports

---
 areno/cli/metrics.py      | 5 +----
 areno/dashboard/server.py | 3 +--
 2 files changed, 2 insertions(+), 6 deletions(-)

diff --git a/areno/cli/metrics.py b/areno/cli/metrics.py
index 5f80ecc3..b8232deb 100644
--- a/areno/cli/metrics.py
+++ b/areno/cli/metrics.py
@@ -18,6 +18,7 @@
 import click
 
 from areno.api import metric_reader
+from areno.api.defaults import DEFAULT_METRICS_LOG_DIR
 
 
 @click.command(name="metrics", context_settings={"help_option_names": ["-h", "--help"]})
@@ -58,10 +59,6 @@ def metrics_command(
 ) -> None:
     """Query metric history (last/min/max/recent/trend) from local run artifacts."""
     if metrics_dir is None:
-        # Lazy import keeps the CLI module light; the default lives with the
-        # dashboard server as the single source (no duplicated absolute path).
-        from areno.dashboard.server import DEFAULT_METRICS_LOG_DIR
-
         metrics_dir = DEFAULT_METRICS_LOG_DIR
 
     # Surface a missing directory early with a clear, located error -- do not let
diff --git a/areno/dashboard/server.py b/areno/dashboard/server.py
index 4c6b26f1..1ea45597 100644
--- a/areno/dashboard/server.py
+++ b/areno/dashboard/server.py
@@ -6,7 +6,6 @@
 import argparse
 import datetime as dt
 import json
-import math
 import os
 import re
 import shlex
@@ -23,6 +22,7 @@
 from typing import Any
 from uuid import uuid4
 
+from areno.api.defaults import DEFAULT_METRICS_LOG_DIR
 from areno.api.metric_reader import number_like
 from areno.cli.dashboard_registry import GLOBAL_REGISTRY_FILE
 from areno.cli.diagnostics import collect_env, run_checks
@@ -32,7 +32,6 @@
 ROOT = Path(os.environ.get("ARENO_DASHBOARD_ROOT", Path.cwd())).resolve()
 STATIC_DIR = Path(__file__).resolve().parent / "dist"
 STATE_FILE = ROOT / ".areno-dashboard-state.json"
-DEFAULT_METRICS_LOG_DIR = "/tmp/areno/tfevent"
 TIME_SEGMENT_ORDER = [
     "rollout",
     "make_sample",

From 6fc39f6565a68386ca0810b3c41531f1ca89299b Mon Sep 17 00:00:00 2001
From: "jiangfangqin.jfq" <jiangfangqin.jfq@digital-engine.com>
Date: Wed, 29 Jul 2026 21:13:59 +0800
Subject: [PATCH 6/8] refactor(metrics): trim dead code, surface
 missing-tensorboard, fix EOF newlines

---
 areno/api/metric_reader.py | 10 +---------
 areno/cli/metrics.py       | 14 +++++++++++++-
 docs/cli/metrics.rst       |  2 +-
 3 files changed, 15 insertions(+), 11 deletions(-)

diff --git a/areno/api/metric_reader.py b/areno/api/metric_reader.py
index ab521592..915890e4 100644
--- a/areno/api/metric_reader.py
+++ b/areno/api/metric_reader.py
@@ -56,11 +56,6 @@ def tensorboard_event_sources(path: Path, pid: int | None) -> list[Path]:
     return [file for file in event_files if pid_marker in file.name or file.parent.name == f"pid-{pid}"]
 
 
-def locate_event_files(metrics_dir: str | Path, pid: int | None = None) -> list[Path]:
-    """Public locator: resolve ``metrics_dir`` and return event file paths."""
-    return tensorboard_event_sources(Path(metrics_dir), pid)
-
-
 def read_scalar_points(metrics_dir: str | Path, pid: int | None = None) -> list[dict[str, Any]]:
     """Read TensorBoard scalars into de-duplicated ``[{name, value, step, time}]``.
 
@@ -163,9 +158,6 @@ def summarize_metric(
         last_step = max(steps)
         min_step = min(steps)
         max_step = last_step
-        for idx, step in enumerate(steps):
-            if step == last_step:
-                last = values[idx]
         window = values[-recent_n:] if recent_n > 0 else []
         window_steps = steps[-recent_n:] if recent_n > 0 else []
         recent = window
@@ -287,4 +279,4 @@ def render_json(summary: dict[str, Any]) -> dict[str, Any]:
         "recent": summary.get("recent"),
         "recent_steps": summary.get("recent_steps"),
         "trend": summary.get("trend"),
-    }
\ No newline at end of file
+    }
diff --git a/areno/cli/metrics.py b/areno/cli/metrics.py
index b8232deb..9dd32d8d 100644
--- a/areno/cli/metrics.py
+++ b/areno/cli/metrics.py
@@ -67,6 +67,18 @@ def metrics_command(
     if not directory.exists():
         raise click.ClickException(f"metrics dir not found: {metrics_dir} (pid={pid})")
 
+    # Probe tensorboard up front: read_scalar_points silently returns [] when the
+    # import fails, which would otherwise look identical to "no metrics found".
+    # Raise a clear, located error so a missing dependency is distinguishable from
+    # an empty directory. (The dashboard keeps read_scalar_points' graceful
+    # degrade-to-empty; this probe is CLI-only.)
+    try:
+        import tensorboard  # noqa: F401
+    except ImportError as exc:
+        raise click.ClickException(
+            "tensorboard is not installed; run `pip install tensorboard` to read metrics."
+        ) from exc
+
     points = metric_reader.read_scalar_points(metrics_dir, pid)
     tags = metric_reader.list_available_tags(points)
 
@@ -120,4 +132,4 @@ def _emit_available(
         return
     click.echo(label)
     for tag in tags:
-        click.echo(f"  {tag}")
\ No newline at end of file
+        click.echo(f"  {tag}")
diff --git a/docs/cli/metrics.rst b/docs/cli/metrics.rst
index d5c99c05..25e711ad 100644
--- a/docs/cli/metrics.rst
+++ b/docs/cli/metrics.rst
@@ -139,4 +139,4 @@ List the metrics available from the default log dir::
 Query one metric from a specific run identified by pid::
 
    areno metrics --metrics-dir /tmp/areno/tfevent --pid 12345 \
-     --name rollout/rewards_mean --limit 50
\ No newline at end of file
+     --name rollout/rewards_mean --limit 50

From 38ae5ee3e0fe1838cf48395472de0990114d54e7 Mon Sep 17 00:00:00 2001
From: "jiangfangqin.jfq" <jiangfangqin.jfq@digital-engine.com>
Date: Thu, 30 Jul 2026 08:02:05 +0800
Subject: [PATCH 7/8] test(metrics): add CPU tests for metric_reader,
 equivalence guard, and metrics CLI

---
 tests/test_cli_metrics_cpu.py               | 223 +++++++++++++++++++
 tests/test_metric_reader_cpu.py             | 230 ++++++++++++++++++++
 tests/test_metric_reader_equivalence_cpu.py | 148 +++++++++++++
 3 files changed, 601 insertions(+)
 create mode 100644 tests/test_cli_metrics_cpu.py
 create mode 100644 tests/test_metric_reader_cpu.py
 create mode 100644 tests/test_metric_reader_equivalence_cpu.py

diff --git a/tests/test_cli_metrics_cpu.py b/tests/test_cli_metrics_cpu.py
new file mode 100644
index 00000000..655df945
--- /dev/null
+++ b/tests/test_cli_metrics_cpu.py
@@ -0,0 +1,223 @@
+"""CPU tests for the ``areno metrics`` CLI (issue #254).
+
+Drives the Click command end-to-end with ``CliRunner`` against real
+``events.out.tfevents.*`` fixtures so the matrix runs through the same read
+path users hit (option wiring, exit codes, available-tag listing, JSON shape).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from click.testing import CliRunner
+from torch.utils.tensorboard import SummaryWriter
+
+from areno.cli.metrics import metrics_command
+
+
+def _writer(metrics_dir: Path) -> SummaryWriter:
+    metrics_dir.mkdir(parents=True, exist_ok=True)
+    return SummaryWriter(log_dir=str(metrics_dir))
+
+
+class MetricsCliTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.runner = CliRunner()
+
+    def _invoke(self, args: list[str]):
+        return self.runner.invoke(metrics_command, args)
+
+    def test_success_table_and_json(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            d = Path(tmp)
+            writer = _writer(d)
+            try:
+                for step in range(5):
+                    writer.add_scalar("rollout/rewards_mean", 0.5 + 0.125 * step, step)
+                writer.add_scalar("train/loss", 2.0, 0)
+            finally:
+                writer.close()
+
+            table = self._invoke(["--metrics-dir", str(d), "--name", "rollout/rewards_mean"])
+            self.assertEqual(table.exit_code, 0)
+            self.assertIn("metric   rollout/rewards_mean", table.output)
+            self.assertIn("count    5", table.output)
+            self.assertIn("steps    0 -> 4", table.output)
+            self.assertIn("mean", table.output)
+            self.assertIn("(step 4)", table.output)
+            self.assertIn("step 0: 0.5", table.output)
+            sparkline = table.output.split("trend    ", 1)[1].splitlines()[0]
+            self.assertEqual(len(sparkline), 5)
+
+            jres = self._invoke(["--metrics-dir", str(d), "--name", "rollout/rewards_mean", "--json"])
+            self.assertEqual(jres.exit_code, 0)
+            payload = json.loads(jres.output)
+            self.assertEqual(payload["name"], "rollout/rewards_mean")
+            self.assertEqual(payload["count"], 5)
+            self.assertEqual(payload["min"], 0.5)
+            self.assertEqual(payload["max"], 1.0)
+            self.assertEqual(payload["max_step"], 4)
+            self.assertEqual(payload["last_step"], 4)
+            self.assertEqual(payload["mean"], 0.75)
+            self.assertEqual(payload["recent_steps"], [0, 1, 2, 3, 4])
+            self.assertEqual(len(payload["trend"]), 5)
+            self.assertEqual(len(payload["recent"]), 5)
+
+    def test_no_name_lists_available_tags_exit_0(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            d = Path(tmp)
+            writer = _writer(d)
+            try:
+                writer.add_scalar("rollout/a", 1.0, 0)
+                writer.add_scalar("train/b", 2.0, 0)
+            finally:
+                writer.close()
+
+            res = self._invoke(["--metrics-dir", str(d)])
+            self.assertEqual(res.exit_code, 0)
+            self.assertIn("available metric tags", res.output)
+            self.assertIn("rollout/a", res.output)
+            self.assertIn("train/b", res.output)
+
+    def test_unknown_name_lists_available_tags_exit_1(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            d = Path(tmp)
+            writer = _writer(d)
+            try:
+                writer.add_scalar("rollout/a", 1.0, 0)
+            finally:
+                writer.close()
+
+            res = self._invoke(["--metrics-dir", str(d), "--name", "does/not/exist"])
+            self.assertEqual(res.exit_code, 1)
+            self.assertIn("metric 'does/not/exist' not found", res.output)
+            self.assertIn("rollout/a", res.output)
+            self.assertIn("Error:", res.output)
+
+    def test_unknown_name_json_emits_missing_tag(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            d = Path(tmp)
+            writer = _writer(d)
+            try:
+                writer.add_scalar("rollout/a", 1.0, 0)
+            finally:
+                writer.close()
+
+            res = self._invoke(["--metrics-dir", str(d), "--name", "x", "--json"])
+            # The JSON listing is emitted before the ClickException, which follows
+            # as an "Error:" line. Assert the payload unconditionally so a dropped
+            # listing fails the test instead of passing silently.
+            payload = json.loads(res.output.split("\nError:")[0])
+            self.assertEqual(payload["available_tags"], ["rollout/a"])
+            self.assertEqual(payload["missing"], "x")
+            self.assertEqual(res.exit_code, 1)
+
+    def test_empty_directory_reports_none_found(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            res = self._invoke(["--metrics-dir", tmp, "--name", "anything"])
+            self.assertEqual(res.exit_code, 1)
+            self.assertIn("available tags: (none found)", res.output)
+
+    def test_directory_with_only_rollout_samples_no_event(self):
+        # rollout_samples.jsonl must NOT be read as metrics; with no event files
+        # the run yields no metric points.
+        with tempfile.TemporaryDirectory() as tmp:
+            d = Path(tmp)
+            (d / "rollout_samples.123.jsonl").write_text('{"name":"x","value":1}\n', encoding="utf-8")
+            res = self._invoke(["--metrics-dir", str(d), "--name", "x"])
+            self.assertEqual(res.exit_code, 1)
+            self.assertIn("available tags: (none found)", res.output)
+
+    def test_missing_directory_exits_1(self):
+        res = self._invoke(["--metrics-dir", "/definitely/missing/areno/zzz", "--name", "x"])
+        self.assertEqual(res.exit_code, 1)
+        self.assertIn("metrics dir not found", res.output)
+
+    def test_missing_tensorboard_reports_clear_error(self):
+        # Simulate a missing tensorboard install by mapping the module to None,
+        # which makes the CLI's `import tensorboard` probe raise ImportError.
+        with tempfile.TemporaryDirectory() as tmp:
+            with patch.dict(sys.modules, {"tensorboard": None}):
+                res = self._invoke(["--metrics-dir", tmp, "--name", "train/loss"])
+            self.assertEqual(res.exit_code, 1)
+            self.assertIn("tensorboard is not installed", res.output)
+
+    def test_single_point_series(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            d = Path(tmp)
+            writer = _writer(d)
+            try:
+                writer.add_scalar("rollout/one", 9.0, 7)
+            finally:
+                writer.close()
+
+            res = self._invoke(["--metrics-dir", str(d), "--name", "rollout/one"])
+            self.assertEqual(res.exit_code, 0)
+            self.assertIn("count    1", res.output)
+            self.assertIn("last     9", res.output)
+            sparkline = res.output.split("trend    ", 1)[1].splitlines()[0]
+            self.assertEqual(len(sparkline), 1)
+
+    def test_all_nan_series_is_empty(self):
+        # A tag whose every point is NaN must not appear as available at all.
+        with tempfile.TemporaryDirectory() as tmp:
+            d = Path(tmp)
+            writer = _writer(d)
+            try:
+                writer.add_scalar("rollout/only_nan", float("nan"), 0)
+                writer.add_scalar("rollout/only_nan", float("nan"), 1)
+                writer.add_scalar("rollout/real", 1.0, 0)
+            finally:
+                writer.close()
+
+            listing = self._invoke(["--metrics-dir", str(d)])
+            self.assertNotIn("only_nan", listing.output)
+            self.assertIn("rollout/real", listing.output)
+
+            res = self._invoke(["--metrics-dir", str(d), "--name", "rollout/only_nan"])
+            self.assertEqual(res.exit_code, 1)
+            self.assertIn("rollout/only_nan", res.output)
+
+    def test_truncation_caps_at_500_points(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            d = Path(tmp)
+            writer = _writer(d)
+            try:
+                for step in range(600):
+                    writer.add_scalar("rollout/big", float(step), step)
+            finally:
+                writer.close()
+
+            res = self._invoke(["--metrics-dir", str(d), "--name", "rollout/big"])
+            self.assertEqual(res.exit_code, 0)
+            self.assertIn("count    500", res.output)
+            # --limit (default 20) bounds the sparkline to the recent window;
+            # count is the full 500, but trend/recent stay bounded.
+            sparkline = res.output.split("trend    ", 1)[1].splitlines()[0]
+            self.assertLessEqual(len(sparkline), 20)
+
+    def test_pid_filter_excludes_other_run(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            d = Path(tmp)
+            writer = _writer(d)
+            try:
+                writer.add_scalar("rollout/a", 1.0, 0)
+            finally:
+                writer.close()
+
+            match = self._invoke(["--metrics-dir", str(d), "--pid", str(os.getpid()), "--name", "rollout/a"])
+            self.assertEqual(match.exit_code, 0)
+
+            miss = self._invoke(["--metrics-dir", str(d), "--pid", "99999999", "--name", "rollout/a"])
+            self.assertEqual(miss.exit_code, 1)
+            self.assertIn("available tags: (none found)", miss.output)
+
+
+if __name__ == "__main__":
+    unittest.main()
\ No newline at end of file
diff --git a/tests/test_metric_reader_cpu.py b/tests/test_metric_reader_cpu.py
new file mode 100644
index 00000000..f057d990
--- /dev/null
+++ b/tests/test_metric_reader_cpu.py
@@ -0,0 +1,230 @@
+"""CPU tests for the pure ``areno.api.metric_reader`` helpers (issue #254)."""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+from torch.utils.tensorboard import SummaryWriter
+
+from areno.api import metric_reader as mr
+
+
+def _write_fixture(metrics_dir: Path) -> int:
+    """Write real tfevents covering normal / NaN / single-point series.
+
+    Returns the writer process pid so pid-filter tests have a known match.
+    Values use powers-of-two fractions to survive TensorBoard's float32
+    round-trip, so assertions can compare for exact equality.
+    """
+    metrics_dir.mkdir(parents=True, exist_ok=True)
+    writer = SummaryWriter(log_dir=str(metrics_dir))
+    try:
+        for step in range(5):
+            writer.add_scalar("rollout/rewards_mean", 0.5 + 0.125 * step, step)
+        writer.add_scalar("rollout/loss_with_nan", float("nan"), 0)
+        writer.add_scalar("rollout/loss_with_nan", 1.5, 1)
+        writer.add_scalar("rollout/single_point", 9.0, 7)
+    finally:
+        writer.close()
+    return os.getpid()
+
+
+class ReadScalarPointsTest(unittest.TestCase):
+    def test_reads_points_skipping_nan_and_preserving_order(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            _write_fixture(Path(tmp))
+            points = mr.read_scalar_points(tmp)
+
+            names_steps = [(p["name"], p["step"], p["value"]) for p in points]
+            self.assertEqual(
+                names_steps,
+                [
+                    ("rollout/rewards_mean", 0, 0.5),
+                    ("rollout/rewards_mean", 1, 0.625),
+                    ("rollout/rewards_mean", 2, 0.75),
+                    ("rollout/rewards_mean", 3, 0.875),
+                    ("rollout/rewards_mean", 4, 1.0),
+                    ("rollout/loss_with_nan", 1, 1.5),
+                    ("rollout/single_point", 7, 9.0),
+                ],
+            )
+            self.assertTrue(all("time" in p and p["time"] for p in points))
+
+    def test_empty_directory_returns_empty(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            self.assertEqual(mr.read_scalar_points(tmp), [])
+            self.assertEqual(mr.tensorboard_event_sources(Path(tmp), None), [Path(tmp)])
+
+    def test_pid_filter_matches_filename_suffix(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            pid = _write_fixture(Path(tmp))
+            self.assertEqual(len(mr.read_scalar_points(tmp, pid=pid)), 7)
+            self.assertEqual(mr.read_scalar_points(tmp, pid=99999999), [])
+
+    def test_truncation_keeps_last_500(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp)
+            path.mkdir(parents=True, exist_ok=True)
+            writer = SummaryWriter(log_dir=str(path))
+            try:
+                for step in range(600):
+                    writer.add_scalar("rollout/big", float(step), step)
+            finally:
+                writer.close()
+            points = mr.read_scalar_points(tmp, pid=None)
+            self.assertEqual(len(points), 500)
+            self.assertEqual(points[0]["step"], 100)
+            self.assertEqual(points[-1]["step"], 599)
+
+
+class SummarizeMetricTest(unittest.TestCase):
+    @staticmethod
+    def _points(name: str, values: list[float], *, start_step: int = 0) -> list[dict]:
+        return [
+            {"name": name, "value": value, "step": start_step + i, "time": "t"}
+            for i, value in enumerate(values)
+        ]
+
+    def test_normal_series_streaming_min_max_and_trend(self):
+        points = self._points("m", [1.0, 3.0, 2.0, 6.0, 4.0])
+        summary = mr.summarize_metric(points, "m", recent_n=3)
+
+        self.assertEqual(summary["count"], 5)
+        self.assertEqual(summary["last"], 4.0)
+        self.assertEqual(summary["last_step"], 4)
+        self.assertEqual(summary["min_step"], 0)
+        self.assertEqual(summary["max_step"], 4)
+        self.assertEqual(summary["mean"], 3.2)
+        self.assertEqual(summary["min"], 1.0)
+        self.assertEqual(summary["max"], 6.0)
+        self.assertEqual(summary["recent"], [2.0, 6.0, 4.0])
+        self.assertEqual(summary["recent_steps"], [2, 3, 4])
+        # trend shares the recent window, so --limit bounds the sparkline length
+        self.assertEqual(len(summary["trend"]), 3)
+        self.assertEqual(summary["trend"], [0.0, 1.0, 0.5])
+
+    def test_trend_bounded_by_recent_n(self):
+        points = self._points("m", [float(i) for i in range(10)])
+        summary = mr.summarize_metric(points, "m", recent_n=5)
+
+        self.assertEqual(len(summary["recent"]), 5)
+        self.assertEqual(len(summary["trend"]), 5)
+        self.assertEqual(summary["trend"], [0.0, 0.25, 0.5, 0.75, 1.0])
+        self.assertEqual(summary["recent_steps"], [5, 6, 7, 8, 9])
+        self.assertEqual(summary["max_step"], 9)
+        self.assertEqual(summary["mean"], 4.5)
+
+    def test_flat_series_trend_is_half(self):
+        points = self._points("m", [2.0, 2.0, 2.0])
+        summary = mr.summarize_metric(points, "m")
+        self.assertEqual(summary["trend"], [0.5, 0.5, 0.5])
+        self.assertEqual(summary["min"], 2.0)
+        self.assertEqual(summary["max"], 2.0)
+        self.assertEqual(summary["mean"], 2.0)
+        self.assertEqual(summary["recent_steps"], [0, 1, 2])
+
+    def test_single_point_series(self):
+        points = self._points("m", [9.0])
+        summary = mr.summarize_metric(points, "m")
+        self.assertEqual(summary["count"], 1)
+        self.assertEqual(summary["last"], 9.0)
+        self.assertEqual(summary["last_step"], 0)
+        self.assertEqual(summary["min_step"], 0)
+        self.assertEqual(summary["max_step"], 0)
+        self.assertEqual(summary["mean"], 9.0)
+        self.assertEqual(summary["min"], 9.0)
+        self.assertEqual(summary["max"], 9.0)
+        self.assertEqual(summary["recent"], [9.0])
+        self.assertEqual(summary["recent_steps"], [0])
+        self.assertEqual(summary["trend"], [0.5])
+
+    def test_unknown_name_yields_empty_summary(self):
+        points = self._points("m", [1.0, 2.0])
+        summary = mr.summarize_metric(points, "does/not/exist")
+        self.assertEqual(summary["count"], 0)
+        self.assertIsNone(summary["last"])
+        self.assertIsNone(summary["last_step"])
+        self.assertIsNone(summary["min_step"])
+        self.assertIsNone(summary["max_step"])
+        self.assertIsNone(summary["mean"])
+        self.assertEqual(summary["recent"], [])
+        self.assertEqual(summary["recent_steps"], [])
+        self.assertEqual(summary["trend"], [])
+
+    def test_nan_values_filtered_by_number_like(self):
+        points = [
+            {"name": "m", "value": 1.0, "step": 0, "time": "t"},
+            {"name": "m", "value": float("nan"), "step": 1, "time": "t"},
+            {"name": "m", "value": 3.0, "step": 2, "time": "t"},
+        ]
+        summary = mr.summarize_metric(points, "m")
+        self.assertEqual(summary["count"], 2)
+        self.assertEqual(summary["min"], 1.0)
+        self.assertEqual(summary["max"], 3.0)
+        self.assertEqual(summary["mean"], 2.0)
+        self.assertEqual(summary["min_step"], 0)
+        self.assertEqual(summary["max_step"], 2)
+
+
+class ListAndRenderTest(unittest.TestCase):
+    def test_list_available_tags_sorted(self):
+        points = [
+            {"name": "c", "value": 1.0, "step": 0, "time": "t"},
+            {"name": "a", "value": 1.0, "step": 0, "time": "t"},
+            {"name": "b", "value": 1.0, "step": 0, "time": "t"},
+            {"name": "", "value": 1.0, "step": 0, "time": "t"},
+        ]
+        self.assertEqual(mr.list_available_tags(points), ["a", "b", "c"])
+
+    def test_render_table_contains_sparkline_and_fields(self):
+        points = [
+            {"name": "m", "value": 1.0, "step": 0, "time": "t"},
+            {"name": "m", "value": 5.0, "step": 1, "time": "t"},
+            {"name": "m", "value": 3.0, "step": 2, "time": "t"},
+        ]
+        summary = mr.summarize_metric(points, "m")
+        rendered = mr.render_table(summary)
+
+        self.assertIn("metric   m", rendered)
+        self.assertIn("count    3", rendered)
+        self.assertIn("steps    0 -> 2", rendered)
+        self.assertIn("last     3", rendered)
+        self.assertIn("(step 2)", rendered)
+        self.assertIn("mean", rendered)
+        self.assertIn("step 0: 1", rendered)
+        self.assertIn("step 2: 3", rendered)
+        sparkline = rendered.split("trend    ", 1)[1].splitlines()[0]
+        self.assertEqual(len(sparkline), 3)
+        for glyph in sparkline:
+            self.assertIn(glyph, "▁▂▃▄▅▆▇█")
+
+    def test_render_table_handles_empty_trend(self):
+        rendered = mr.render_table(mr.summarize_metric([], "missing"))
+        self.assertIn("count    0", rendered)
+        self.assertIn("last     -", rendered)
+
+    def test_render_json_round_trips_and_keeps_trend_array(self):
+        points = [
+            {"name": "m", "value": 1.0, "step": 0, "time": "t"},
+            {"name": "m", "value": 3.0, "step": 1, "time": "t"},
+        ]
+        summary = mr.summarize_metric(points, "m")
+        decoded = json.loads(json.dumps(mr.render_json(summary)))
+        self.assertEqual(decoded["name"], "m")
+        self.assertEqual(decoded["count"], 2)
+        self.assertEqual(decoded["last"], 3.0)
+        self.assertEqual(decoded["last_step"], 1)
+        self.assertEqual(decoded["min_step"], 0)
+        self.assertEqual(decoded["max_step"], 1)
+        self.assertEqual(decoded["mean"], 2.0)
+        self.assertEqual(decoded["recent_steps"], [0, 1])
+        self.assertEqual(decoded["trend"], [0.0, 1.0])
+        self.assertEqual(decoded["recent"], [1.0, 3.0])
+
+
+if __name__ == "__main__":
+    unittest.main()
\ No newline at end of file
diff --git a/tests/test_metric_reader_equivalence_cpu.py b/tests/test_metric_reader_equivalence_cpu.py
new file mode 100644
index 00000000..dd7a0d3b
--- /dev/null
+++ b/tests/test_metric_reader_equivalence_cpu.py
@@ -0,0 +1,148 @@
+"""Equivalence guard: dashboard read-side output before vs after the metric_reader extract (issue #254).
+
+Locks the current output of ``DashboardState._load_tensorboard_scalars``
+(``job.metrics`` + ``job.timeperf``) against a real
+``events.out.tfevents.*`` fixture written via ``SummaryWriter``. After the
+extract switches ``_load_tensorboard_scalars`` to call
+``areno.api.metric_reader``, this test must keep passing unchanged -- it is
+the regression guard that proves the refactor is behavior-preserving.
+"""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from areno.dashboard.server import DashboardState, Job, ROOT
+
+
+def _write_tfevents_fixture(metrics_dir: Path) -> None:
+    """Write a small set of real TensorBoard event files.
+
+    Tags cover every branch of ``_load_tensorboard_scalars``: normal scalars,
+    NaN-skip, single-point series, and the timeperf ``time/`` +
+    ``train/step_*_time_s`` tags that drive the by_step aggregation. Values
+    use powers-of-two fractions to survive TensorBoard's float32 round-trip.
+    """
+    from torch.utils.tensorboard import SummaryWriter
+
+    metrics_dir.mkdir(parents=True, exist_ok=True)
+    writer = SummaryWriter(log_dir=str(metrics_dir))
+    try:
+        for step in range(5):
+            writer.add_scalar("rollout/rewards_mean", 0.5 + 0.125 * step, step)
+        writer.add_scalar("rollout/loss_with_nan", float("nan"), 0)
+        writer.add_scalar("rollout/loss_with_nan", 1.5, 1)
+        writer.add_scalar("rollout/loss_with_nan", float("nan"), 2)
+        writer.add_scalar("rollout/loss_with_nan", 2.5, 3)
+        writer.add_scalar("rollout/single_point", 9.0, 7)
+        for step in range(3):
+            writer.add_scalar("time/rollout", 2.0 + step, step)
+            writer.add_scalar("time/train", 1.0 + step, step)
+            writer.add_scalar("train/step_e2e_time_s", 3.0 + step, step)
+    finally:
+        writer.close()
+
+
+def _load_job_metrics(metrics_dir: str) -> Job:
+    """Drive the dashboard read path exactly as the server does."""
+    state = DashboardState()
+    job = Job(kind="train", name="probe", command=[], config={}, metrics_dir=metrics_dir)
+    state.jobs[job.id] = job
+    path = (ROOT / metrics_dir).resolve()
+    state._load_tensorboard_scalars(job, path)
+    return job
+
+
+def _strip_nondeterministic(points: list[dict]) -> list[dict]:
+    """Drop the ``time`` field, which is stamped with wall-clock at read time."""
+    return [
+        {"name": p.get("name"), "value": p.get("value"), "step": int(p.get("step") or 0)}
+        for p in points
+    ]
+
+
+def _strip_timeperf_nondeterministic(rows: list[dict]) -> list[dict]:
+    """Drop ``time`` from timeperf rows; keep the deterministic aggregation."""
+    return [{k: v for k, v in dict(row).items() if k != "time"} for row in rows]
+
+
+class MetricReaderEquivalenceTest(unittest.TestCase):
+    def test_locks_metric_points_and_timeperf(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            metrics_dir = Path(tmp)
+            _write_tfevents_fixture(metrics_dir)
+            job = _load_job_metrics(str(metrics_dir))
+
+            metrics = _strip_nondeterministic(job.metrics)
+            timeperf = _strip_timeperf_nondeterministic(job.timeperf)
+
+            self.assertEqual(
+                metrics,
+                [
+                    {"name": "rollout/rewards_mean", "value": 0.5, "step": 0},
+                    {"name": "rollout/rewards_mean", "value": 0.625, "step": 1},
+                    {"name": "rollout/rewards_mean", "value": 0.75, "step": 2},
+                    {"name": "rollout/rewards_mean", "value": 0.875, "step": 3},
+                    {"name": "rollout/rewards_mean", "value": 1.0, "step": 4},
+                    {"name": "rollout/loss_with_nan", "value": 1.5, "step": 1},
+                    {"name": "rollout/loss_with_nan", "value": 2.5, "step": 3},
+                    {"name": "rollout/single_point", "value": 9.0, "step": 7},
+                    {"name": "time/rollout", "value": 2.0, "step": 0},
+                    {"name": "time/rollout", "value": 3.0, "step": 1},
+                    {"name": "time/rollout", "value": 4.0, "step": 2},
+                    {"name": "time/train", "value": 1.0, "step": 0},
+                    {"name": "time/train", "value": 2.0, "step": 1},
+                    {"name": "time/train", "value": 3.0, "step": 2},
+                    {"name": "train/step_e2e_time_s", "value": 3.0, "step": 0},
+                    {"name": "train/step_e2e_time_s", "value": 4.0, "step": 1},
+                    {"name": "train/step_e2e_time_s", "value": 5.0, "step": 2},
+                ],
+            )
+
+            self.assertEqual(
+                timeperf,
+                [
+                    {
+                        "step": 0,
+                        "segments": [
+                            {"name": "rollout", "seconds": 2.0},
+                            {"name": "train", "seconds": 1.0},
+                        ],
+                        "rollout_s": 2.0,
+                        "train_s": 1.0,
+                        "other_s": 0.0,
+                        "total_s": 3.0,
+                        "source": "metrics",
+                    },
+                    {
+                        "step": 1,
+                        "segments": [
+                            {"name": "rollout", "seconds": 3.0},
+                            {"name": "train", "seconds": 2.0},
+                        ],
+                        "rollout_s": 3.0,
+                        "train_s": 2.0,
+                        "other_s": 0.0,
+                        "total_s": 4.0,
+                        "source": "metrics",
+                    },
+                    {
+                        "step": 2,
+                        "segments": [
+                            {"name": "rollout", "seconds": 4.0},
+                            {"name": "train", "seconds": 3.0},
+                        ],
+                        "rollout_s": 4.0,
+                        "train_s": 3.0,
+                        "other_s": 0.0,
+                        "total_s": 5.0,
+                        "source": "metrics",
+                    },
+                ],
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()
\ No newline at end of file

From 73a7c0216688528fa959bc7491d8ae75bbb0e8ad Mon Sep 17 00:00:00 2001
From: "jiangfangqin.jfq" <jiangfangqin.jfq@digital-engine.com>
Date: Thu, 30 Jul 2026 17:09:41 +0800
Subject: [PATCH 8/8] docs(metrics): fix docstring typo in areno metrics CLI

---
 areno/cli/metrics.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/areno/cli/metrics.py b/areno/cli/metrics.py
index 9dd32d8d..e9c4201b 100644
--- a/areno/cli/metrics.py
+++ b/areno/cli/metrics.py
@@ -2,7 +2,7 @@
 
 Issue #254 adds a read-only CLI that summarizes one metric (last/min/max/
 recent/trend) from the ``events.out.tfevents.*`` artifacts a run writes. The
-heavywork lives in the light, pure :mod:`areno.api.metric_reader` module; this
+heavy work lives in the light, pure :mod:`areno.api.metric_reader` module; this
 file only wires Click options, renders, and turns not-found cases into a clear
 error plus the list of available tags.