Keep now on the metrics axis while a job is reporting (#4132)

* Keep `now` on the metrics axis while a job is reporting

The axis labelled its right edge `now` only when the newest sample was
under ten seconds old, borrowing `pretty_date`'s threshold. Collection
runs every ten seconds, so samples routinely arrive older than that:
polling a live job, three renders in ten showed an absolute time instead,
and under `--watch` the edge alternated between the two.

Thresholds on three collection intervals instead. A job whose instance
goes unreachable still shows its last timestamp -- that gap is minutes,
not seconds.

Draws `now` bold, in the grey `no data` already uses, so a live run is
distinguishable at a glance from one that stopped reporting.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* Drop docstrings that summarise the code they sit on

Five of the six in this module restated what the function below already
said, or argued for the option taken over one that was not. The remaining
one records that the server sends samples newest-first, which is a fact
from outside this file and draws every chart backwards if missed.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* Name the rendering helpers after what they do

`_axis` and `_stamp` said nothing about building a timeline row or
formatting a timestamp for it, and `_span` sat next to `_window` meaning
two different windows. Names that carry their meaning at the call site
beat a docstring that only carries it at the definition:

  _axis        -> _time_axis        builds the timeline row
  _stamp       -> _time_label       formats one timestamp for it
  _span        -> _shared_window    the window all jobs are drawn against
  _window      -> _job_window       one job's own first and last sample
  _lead        -> _blank_cells      cells to blank before a job started
  _drawn       -> _cells_drawn      cells a job actually fills
  _cell        -> _chart_cell       a sparkline joined to its number
  _level_cell  -> _capacity_cell    memory as a fraction of capacity
  _latest      -> _latest_value

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* Say what each rendering helper does

Renaming alone did not carry it: `_time_axis` still did not say it builds
a row, `_time_label` hid that it returns `now` for a live job, and
`_samples_num` did not say it takes the longest series. Each now has one
line stating what it does.

`_chart_cell` is gone -- it wrapped a single `Text.assemble` at three
call sites and only added a name to look up.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Andrey Cheptsov <andrey.cheptsov@github.com>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: 3 people

src/dstack/_internal/cli/utils/metrics.py

@@ -1,4 +1,4 @@
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Any, List, Optional, Sequence
 
 from rich.console import RenderableType
@@ -10,7 +10,6 @@
 from dstack._internal.core.models.instances import Resources
 from dstack._internal.core.models.metrics import JobMetrics
 from dstack._internal.core.models.runs import Job
-from dstack._internal.utils.common import pretty_date
 
 MAX_SAMPLES = 1000
 """A sample count, not a window: outruns the hour a running job retains, so a young run is
@@ -26,6 +25,9 @@
 """What the server keeps for a running job, and so the widest window there can be."""
 
 AXIS_RULE = "┄"
+NOW = "now"
+
+LIVE_THRESHOLD = timedelta(seconds=3 * WATCH_INTERVAL_SECONDS)
 _FIXED_COLUMNS = 30
 """Everything but the sparklines and the job label: the `gpu=N` column, both numbers, and
 the table's padding. Hand-measured against a `589GB/1480GB`-sized number; a wider one
@@ -47,7 +49,7 @@ def get_metrics_table(
     labels = job_labels(jobs)
     label_width = max((len(label) for label in labels), default=0)
     width = _spark_width(console_width or console.width, label_width)
-    span = _span(metrics)
+    window = _shared_window(metrics)
 
     table = Table(box=None)
     # no headers: the cells read `replica=0` and `gpu=1`, which need no naming
@@ -59,24 +61,19 @@ def get_metrics_table(
     for index, (job, job_metrics) in enumerate(zip(jobs, metrics)):
         if index:
             table.add_row("", "", "", "")
-        _add_job(table, job, job_metrics, width, labels[index], span)
+        _add_job(table, job, job_metrics, width, labels[index], window)
 
-    if span is not None:
+    if window is not None:
         table.add_row("", "", "", "")
         # the axis spans the widest chart drawn: a job with fewer samples than cells draws
         # one cell per sample and cannot fill its share
-        axis = _axis(max(_drawn(m, span, width) for m in metrics), *span)
+        axis = _time_axis(max(_cells_drawn(m, window, width) for m in metrics), *window)
         table.add_row("", "", axis, axis)
     return table
 
 
 def job_labels(jobs: Sequence[Job]) -> List[str]:
-    """`replica=`/`group=` only where they distinguish something, as `dstack ps` does --
-    one replica across four nodes is `job=0..3`, not `replica=0 job=0..3`.
-
-    Unlike `ps`, `job=` is always printed. This table is keyed by job, so every row names
-    one; `replica=` joins it only where there is more than one replica to tell apart.
-    """
+    """A label per job, naming only what tells them apart, as `dstack ps` does."""
     groups = {job.job_spec.replica_group for job in jobs}
     show_group = len(groups) > 1
     show_replica = len({job.job_spec.replica_num for job in jobs}) > 1
@@ -102,59 +99,60 @@ def _add_job(
     metrics: JobMetrics,
     width: int,
     label: str,
-    span: Optional[tuple[datetime, datetime]],
+    window: Optional[tuple[datetime, datetime]],
 ) -> None:
     resources = _get_resources(job)
-    lead = _lead(metrics, span, width)
-    cells = width - lead
+    blanks = _blank_cells(metrics, window, width)
+    cells = width - blanks
     table.add_row(
         label,
         "cpu",
-        _pad(_cpu_cell(metrics, resources, cells), lead),
-        _pad(_memory_cell(metrics, resources, cells), lead),
+        _pad(_cpu_cell(metrics, resources, cells), blanks),
+        _pad(_memory_cell(metrics, resources, cells), blanks),
     )
     for index in range(_gpus_num(metrics, resources)):
         table.add_row(
             "",
             f"gpu={index}",
-            _pad(_gpu_util_cell(metrics, index, cells), lead),
-            _pad(_gpu_memory_cell(metrics, resources, index, cells), lead),
+            _pad(_gpu_util_cell(metrics, index, cells), blanks),
+            _pad(_gpu_memory_cell(metrics, resources, index, cells), blanks),
         )
 
 
-def _span(metrics: Sequence[JobMetrics]) -> Optional[tuple[datetime, datetime]]:
-    """The window every job is drawn against: always the full retention hour.
-
-    Fixed rather than fitted to the data, so a row means the same thing in every
-    invocation and across every job. A job younger than the hour fills only its share of
-    the row and the rest is blank -- which is the fact worth seeing about a replica that
-    started two minutes ago.
-    """
-    windows = [w for w in (_window(m) for m in metrics) if w is not None]
+def _shared_window(metrics: Sequence[JobMetrics]) -> Optional[tuple[datetime, datetime]]:
+    """The window every job is charted against: the newest sample back one retention hour."""
+    windows = [w for w in (_job_window(m) for m in metrics) if w is not None]
     if not windows:
         return None
     latest, earliest = max(w[1] for w in windows), min(w[0] for w in windows)
     return min(earliest, latest - RETENTION), latest
 
 
-def _lead(metrics: JobMetrics, span: Optional[tuple[datetime, datetime]], width: int) -> int:
-    """Cells before this job's first sample -- time it was not running for."""
-    window = _window(metrics)
-    if window is None or span is None:
+def _blank_cells(
+    metrics: JobMetrics, window: Optional[tuple[datetime, datetime]], width: int
+) -> int:
+    """How many cells to leave empty before a job's chart, so it starts where it started."""
+    job_window = _job_window(metrics)
+    if job_window is None or window is None:
         return 0
-    total = (span[1] - span[0]).total_seconds()
+    total = (window[1] - window[0]).total_seconds()
     if total <= 0:
         return 0
-    return min(width - 1, max(0, round((window[0] - span[0]).total_seconds() / total * width)))
+    started = (job_window[0] - window[0]).total_seconds()
+    return min(width - 1, max(0, round(started / total * width)))
 
 
-def _drawn(metrics: JobMetrics, span: Optional[tuple[datetime, datetime]], width: int) -> int:
-    lead = _lead(metrics, span, width)
-    return lead + min(width - lead, _samples_num(metrics))
+def _cells_drawn(
+    metrics: JobMetrics, window: Optional[tuple[datetime, datetime]], width: int
+) -> int:
+    """How many cells a job's chart occupies: its empty lead plus one per sample."""
+    blanks = _blank_cells(metrics, window, width)
+    return blanks + min(width - blanks, _samples_num(metrics))
 
 
-def _pad(cell: Text, lead: int) -> Text:
-    return cell if lead <= 0 else Text.assemble(Text(" " * lead), cell)
+def _pad(cell: Text, blanks: int) -> Text:
+    """Prefix a chart cell with empty cells, for time before the job started."""
+    return cell if blanks <= 0 else Text.assemble(Text(" " * blanks), cell)
 
 
 def _cpu_cell(job_metrics: JobMetrics, resources: Optional[Resources], width: int) -> Text:
@@ -166,15 +164,15 @@ def _cpu_cell(job_metrics: JobMetrics, resources: Optional[Resources], width: in
         values = [v / cpus for v in values]
     # no core count: the value is already normalised to it, and unlike memory there is no
     # total to give the number meaning
-    return _cell(sparkline(values, width, HOST_RAMP), f"{values[-1]:.0f}%")
+    return Text.assemble(sparkline(values, width, HOST_RAMP), " ", f"{values[-1]:.0f}%")
 
 
 def _memory_cell(job_metrics: JobMetrics, resources: Optional[Resources], width: int) -> Text:
     values = _metric_values(job_metrics, "memory_working_set_bytes")
     if not values:
         return no_data()
     total = resources.memory_mib * 1024 * 1024 if resources else None
-    return _level_cell(values, total, width, HOST_RAMP)
+    return _capacity_cell(values, total, width, HOST_RAMP)
 
 
 def _gpu_memory_cell(
@@ -189,65 +187,54 @@ def _gpu_memory_cell(
     total = None
     if resources and index < len(resources.gpus):
         total = resources.gpus[index].memory_mib * 1024 * 1024
-    return _level_cell(values, total, width, GPU_RAMP)
+    return _capacity_cell(values, total, width, GPU_RAMP)
 
 
 def _gpu_util_cell(job_metrics: JobMetrics, index: int, width: int) -> Text:
     values = _metric_values(job_metrics, f"gpu_util_percent_gpu{index}")
     if not values:
         return no_data()
-    return _cell(sparkline(values, width, GPU_RAMP), f"{values[-1]:.0f}%")
+    return Text.assemble(sparkline(values, width, GPU_RAMP), " ", f"{values[-1]:.0f}%")
 
 
-def _level_cell(values: List[float], total: Optional[float], width: int, ramp: Ramp) -> Text:
+def _capacity_cell(values: List[float], total: Optional[float], width: int, ramp: Ramp) -> Text:
+    """A memory chart drawn against capacity, labelled `used/total`."""
     percents = [v / total * 100 for v in values] if total else values
     label = format_memory(values[-1], 0)
     if total:
         label += f"/{format_memory(total, 0)}"
-    return _cell(sparkline(percents, width, ramp), label)
-
-
-def _cell(spark: Text, label: str) -> Text:
-    return Text.assemble(spark, " ", label)
-
-
-def _axis(width: int, first: datetime, last: datetime) -> Text:
-    """`<oldest> ┄┄┄ <newest>`, never wider than the sparkline above it.
+    return Text.assemble(sparkline(percents, width, ramp), " ", label)
 
-    The rule is what pairs the two stamps. UTILIZATION and MEMORY each print one, so the
-    row ends up holding four times, and with the rule left blank the only cue is spacing --
-    which points the wrong way above 88 columns: at 200 there are 66 blanks between a
-    column's own two stamps but only 13 between the columns, so each column's newest time
-    reads as belonging to the next column's oldest.
 
-    A run draws one cell per sample, so for its first few minutes there are fewer cells
-    than two dates need. Dropping the date keeps the axis inside its cell; overflowing
-    instead widens the column and pulls MEMORY out of line with the charts.
-    """
-    left, right = _stamp(first), _stamp(last)
+def _time_axis(width: int, first: datetime, last: datetime) -> Text:
+    """The timeline row printed under the charts, exactly `width` columns wide."""
+    left, right = _time_label(first), _time_label(last)
     if len(left) + len(right) + 3 > width:
-        left, right = _stamp(first, clock_only=True), _stamp(last, clock_only=True)
+        left, right = _time_label(first, clock_only=True), _time_label(last, clock_only=True)
     if len(left) + len(right) + 2 > width:
         return Text("")
     fill = width - len(left) - len(right) - 2
-    return Text(f"{left} " + AXIS_RULE * fill + f" {right}", style="grey42")
+    axis = Text(f"{left} " + AXIS_RULE * fill + " ", style="grey42")
+    axis.append(right, style="bold grey58" if right == NOW else "grey42")
+    return axis
 
 
-def _stamp(moment: datetime, clock_only: bool = False) -> str:
-    if pretty_date(moment) == "now":
-        return "now"
+def _time_label(moment: datetime, clock_only: bool = False) -> str:
+    """One timestamp for the axis: `now` while a job is still reporting, a date otherwise."""
+    if datetime.now(timezone.utc) - moment < LIVE_THRESHOLD:
+        return NOW
     local = moment.astimezone()
     return f"{local:%H:%M}" if clock_only else f"{local.day} {local:%b %H:%M}"
 
 
-def _window(job_metrics: JobMetrics) -> Optional[tuple[datetime, datetime]]:
+def _job_window(job_metrics: JobMetrics) -> Optional[tuple[datetime, datetime]]:
+    """The oldest and newest sample timestamps of one job, or None if it has none."""
     stamps = [t for metric in job_metrics.metrics for t in metric.timestamps]
     return (min(stamps), max(stamps)) if stamps else None
 
 
 def _samples_num(job_metrics: JobMetrics) -> int:
-    """`slices` never draws more cells than it has samples, so the axis must stop there
-    too -- else it claims a span nothing was measured over, and Rich widens the column."""
+    """How many samples the longest series holds."""
     return max((len(metric.timestamps) for metric in job_metrics.metrics), default=0)
 
 
@@ -260,15 +247,15 @@ def _metric_values(job_metrics: JobMetrics, name: str) -> List[Any]:
     return []
 
 
-def _latest(job_metrics: JobMetrics, name: str) -> Optional[Any]:
+def _latest_value(job_metrics: JobMetrics, name: str) -> Optional[Any]:
     values = _metric_values(job_metrics, name)
     return values[-1] if values else None
 
 
 def _gpus_num(job_metrics: JobMetrics, resources: Optional[Resources]) -> int:
     if resources is not None and resources.gpus:
         return len(resources.gpus)
-    detected = _latest(job_metrics, "gpus_detected_num")
+    detected = _latest_value(job_metrics, "gpus_detected_num")
     return int(detected) if detected else 0
 
 

src/tests/_internal/cli/utils/test_metrics.py

@@ -8,6 +8,9 @@
 from rich.theme import Theme
 
 from dstack._internal.cli.utils.metrics import (
+    _job_window,
+    _time_axis,
+    _time_label,
     format_memory,
     get_metrics_table,
     job_labels,
@@ -186,6 +189,18 @@ def test_a_finished_run_cannot_look_live(self, state: str, live: bool):
         if not live:
             assert ":" in axis  # a real clock time, not an age
 
+    @pytest.mark.parametrize("age_seconds,live", [(13, True), (45, False)])
+    def test_a_sample_may_lag_a_few_intervals_and_still_read_as_live(self, age_seconds, live):
+        moment = datetime.now(timezone.utc) - timedelta(seconds=age_seconds)
+        assert (_time_label(moment) == "now") == live
+
+    @pytest.mark.parametrize("state,emphasised", [("running", True), ("terminated", False)])
+    def test_only_a_live_edge_is_emphasised(self, state: str, emphasised: bool):
+        job, metrics = make_run("saturated", state=state)
+        axis = _time_axis(60, *_job_window(metrics))
+        styles = {str(span.style) for span in axis.spans}
+        assert ("bold grey58" in styles) == emphasised
+
 
 class TestJobs:
     def test_every_job_is_shown_and_keyed(self):