"""The monitor's data layer: one read-only snapshot of a run, fully computed.
``build_snapshot`` is everything the ``monitor`` command calls per refresh
tick: it reads the durable state through a (read-only) ``StateStore`` and
returns a ``MonitorSnapshot`` -- plain frozen dataclasses the UI layer renders
without further queries. The split keeps the CLAUDE.md arrows intact: SQL
lives in ``core/state.py``, rendering lives in ``ui/``, and this module is
pure computation between the two -- no Rich, no SQLite, no config imports
(the caller hands in plain names and mappings, exactly like the
Orchestrator's constructor).
Semantics worth naming:
* **ready** counts materials sitting at a stage's *input* state and not in
flight; **running** counts materials linked to a live ledger job of that
stage (a material's state only advances at completion, so the two never
double-count). Input states are derived from ``STAGE_OUTPUT_STATE`` by
*name* over the configured stage order -- an omitted optional stage
(``calculate``) shifts its consumer's input exactly as the orchestrator
does.
* **done_total** is the all-time completions-ledger count for the stage --
a fan-out parent that completed ``structure`` stays counted even after it
retires as ``EXPANDED``, which per-state counting would lose.
* **speed / throughput** are rolling-window rates from ledger timestamps;
overall throughput counts ``FINALIZE_STAGE`` rows (materials actually
finishing the run). ``eta_s`` is the naive remaining/throughput estimate
and ``None`` whenever throughput is zero -- the renderer shows a dash,
never a division by zero.
* **tracked** rows exist only when the caller passes ``tracked_ids`` (the
``--track`` flow): one ``TrackedMaterial`` per requested id, caller order,
unknown ids kept as honest ``found=False`` rows. ``tracked is None`` is
the renderer's signal that no tracking was requested at all.
* **running** rows are the per-material view of the in-flight ledger --
one ``RunningMaterial`` per material riding a live job, longest-running
first. ``cpu_percent`` is filled from the caller's ``cpu_by_job`` sample
(the command layer owns the probing, exactly like ``backend_health``);
a job the sampler could not see stays ``None``, rendered as a dash.
* **recent_successes** mirrors ``recent_failures``: the newest ``DONE``
materials, each record carrying ``updated_at`` (finished at) and
``created_at`` (so ``updated_at - created_at`` is how long the ride took).
"""
from __future__ import annotations
import time
from dataclasses import dataclass
from typing import TYPE_CHECKING
from matmelt.core.stages.registry import STAGE_OUTPUT_STATE
from matmelt.core.state import (
FINALIZE_STAGE,
MaterialState,
is_terminal,
)
if TYPE_CHECKING:
from collections.abc import Mapping, Sequence
from matmelt.core.state import InFlightJob, MaterialRecord, StateStore
[docs]
@dataclass(frozen=True)
class StageSnapshot:
"""One configured stage's live numbers, as rendered in the pipeline panel.
Attributes
----------
name : str
The stage's canonical name.
backend_name : str
Which configured backend runs it.
backend_kind : str
That backend's kind, or ``"?"`` when the caller did not supply one.
backend_healthy : bool or None
The caller's cached ``health_check`` verdict. ``None`` means "not
probed", rendered neutrally and never as a failure.
ready : int
Materials sitting at this stage's input state and not in flight.
Disjoint from ``running`` by construction.
running : int
Materials linked to a live ledger job of this stage.
done_total : int
All-time completions of this stage, counted from the ledger, so a
fan-out parent stays counted after it retires as ``EXPANDED``.
failed : int
Materials terminally FAILED *at* this stage.
speed_per_min : float
Completions of this stage within the rolling window, per minute;
``0.0`` when idle.
"""
name: str
backend_name: str
backend_kind: str
backend_healthy: bool | None
ready: int
running: int
done_total: int
failed: int
speed_per_min: float
[docs]
@dataclass(frozen=True)
class RunningMaterial:
"""One row of the running-now panel: a material currently riding a live job.
Built from the in-flight ledger (one row per material of every live job)
plus that material's store record -- the panel's per-material live view.
Attributes
----------
material_id : str
The material riding the job.
stage : str
Which stage's job it is.
backend_name : str
Where that job is running.
job_key : str
The ledger's key for the job, and the key the usage samplers report
against.
elapsed_s : float
Seconds since the job was ledgered, i.e. how long this material has
been computing in its current job.
attempts : int
Consecutive failed attempts at the current stage, so a requeued
material shows which try this is. ``0`` when the record is missing --
a foreign ledger row must render, never crash the dashboard.
cpu_percent : float or None
Caller-sampled CPU usage of the job's process subtree, as a percentage
of one core. ``None`` whenever the sampler had no reading, rendered as
a dash and never as ``0``.
gpu_percent : float or None
Caller-sampled GPU usage, same conventions -- ``None`` for a remote
that was not probed, a SLURM-scheduled job, a vanished PID, or a
machine without ``nvidia-smi``.
"""
material_id: str
stage: str
backend_name: str
job_key: str
elapsed_s: float
attempts: int
cpu_percent: float | None
gpu_percent: float | None
[docs]
@dataclass(frozen=True)
class TrackedMaterial:
"""One row of the tracked-detail panel: a single followed material's whereabouts.
Built only when the caller passes ``tracked_ids`` to ``build_snapshot``
(the ``--track`` flow); the renderer swaps the failures panel for a
tracked-materials table keyed on these rows.
Attributes
----------
material_id : str
The id the caller asked to follow, echoed verbatim.
found : bool
Whether the id exists in the store at all. ``False`` rows render a dim
"not found" -- a typo'd or not-yet-enqueued id must show up honestly,
never crash or silently vanish -- and every other field is then a
neutral zero or ``None`` filler.
state : MaterialState or None
The material's current lifecycle state.
stage : str or None
The stage of the ledger job currently carrying this material, or
``None`` when it sits between jobs (or is terminal).
backend_name : str or None
Where that job runs, same ``None`` rule.
in_state_s : float
Seconds since the record's ``updated_at``, i.e. time in the current
state. ``0.0`` when not found.
attempts : int
Consecutive failed attempts at the current stage.
fail_reason : str or None
The **first line only** of the record's failure reason -- reasons are
multi-line tracebacks, and the panel shows one line while the detail
stays with ``status``. ``None`` when the material has not failed.
job_key : str or None
The ledger key of the carrying job, same ``None`` rule as ``stage``.
"""
material_id: str
found: bool
state: MaterialState | None
stage: str | None
backend_name: str | None
in_state_s: float
attempts: int
fail_reason: str | None
job_key: str | None
[docs]
@dataclass(frozen=True)
class MonitorSnapshot:
"""Everything one refresh tick of the monitor shows, fully computed.
Attributes
----------
run_name : str
The configured run name, shown in the header.
generated_at : float
The ``now`` this snapshot was computed against.
started_at : float or None
The run's wall-clock anchor -- the earliest material's creation time.
``None`` on an empty store.
elapsed_s : float
Time since ``started_at``; ``0.0`` on an empty store.
total : int
How many materials the store holds.
by_state : dict
The full per-state tally. Absent states are zero-filled by the
renderer, not here.
in_flight : int
Materials currently linked to live ledger jobs.
stages : tuple
One ``StageSnapshot`` per configured stage, in funnel order.
recent_failures : tuple
Newest-first terminal failures. Each record carries ``fail_stage``,
``fail_reason``, and ``updated_at`` as the failed-at time.
recent_successes : tuple
Newest-first ``DONE`` materials, where ``updated_at`` is the finish
time and ``updated_at - created_at`` is how long the ride took.
running : tuple
One ``RunningMaterial`` per material on a live ledger job,
longest-running first.
throughput_per_min : float
Run-level finish rate over the rolling window, counted from
``FINALIZE_STAGE`` ledger rows.
eta_s : float or None
The naive remaining-over-throughput estimate, and ``None`` whenever
throughput is zero -- the renderer shows a dash rather than dividing by
zero.
stop_after : str or None
The configured stop stage, echoed for the header.
window_s : float
Width of the rolling window the rates were computed over.
tracked : tuple or None
The ``--track`` rows in the caller's requested order. ``None`` when no
ids were requested, which is exactly what the renderer switches its
third panel on: ``None`` gives the failures panel, any tuple -- even an
empty one -- gives the tracked panel.
"""
run_name: str
generated_at: float
started_at: float | None
elapsed_s: float
total: int
by_state: dict[MaterialState, int]
in_flight: int
stages: tuple[StageSnapshot, ...]
recent_failures: tuple[MaterialRecord, ...]
recent_successes: tuple[MaterialRecord, ...]
running: tuple[RunningMaterial, ...]
throughput_per_min: float
eta_s: float | None
stop_after: str | None
window_s: float
tracked: tuple[TrackedMaterial, ...] | None = None
[docs]
def build_snapshot(
store: StateStore,
*,
run_name: str,
stage_names: Sequence[str],
stage_backends: Mapping[str, str],
backend_kinds: Mapping[str, str],
backend_health: Mapping[str, bool] | None = None,
stop_after: str | None = None,
now: float | None = None,
window_s: float = 300.0,
failure_limit: int = 10,
success_limit: int = 10,
cpu_by_job: Mapping[str, float] | None = None,
gpu_by_job: Mapping[str, float] | None = None,
tracked_ids: Sequence[str] | None = None,
) -> MonitorSnapshot:
"""Read the store once and compute the full monitor snapshot.
Read-only by construction: only ``StateStore`` read methods are called, so
passing a ``read_only=True`` store is the expected usage.
Parameters
----------
store
The state store to read. Every query the tick needs happens here, once.
run_name
The configured run name, passed through to the snapshot.
stage_names
The configured funnel in order. Canonical names only.
stage_backends
Each stage's backend *name* -- plain strings, so this module never
touches config or backend objects.
backend_kinds
Each backend name's kind. A name absent here renders as ``"?"``.
backend_health
The caller's cached probe results. A missing entry means "unknown" and
renders neutrally.
stop_after
The configured stop stage, echoed into the snapshot.
now
The timestamp to compute against. Injectable for tests; defaults to
``time.time()``.
window_s
Width of the rolling window for the per-stage speeds and the run
throughput.
failure_limit
How many recent failures to carry.
success_limit
How many recent successes to carry.
cpu_by_job, gpu_by_job
The caller's latest usage samples as ``{job_key: percent}``. The
command layer owns the probing and its throttling, exactly as it does
for ``backend_health``; a job absent from a mapping renders ``None`` in
that column.
tracked_ids
Ids to follow (the ``--track`` flow): caller order is preserved,
duplicates dropped keeping the first occurrence, and unknown ids come
back as ``found=False`` rows. Each id costs one ``store.get``, which is
fine at tracked scale. Left ``None``, the snapshot is exactly what it
was before this parameter existed.
Returns
-------
MonitorSnapshot
Everything one refresh tick renders, with no further queries needed.
Raises
------
ValueError
If a stage name is not a canonical funnel stage, or has no entry in
``stage_backends`` -- fail-fast, like the Orchestrator's wiring checks.
"""
if now is None:
now = time.time()
input_states: dict[str, MaterialState] = {}
previous = MaterialState.NEW
for name in stage_names:
output = STAGE_OUTPUT_STATE.get(name)
if output is None:
raise ValueError(f"unknown stage {name!r}; known stages: {sorted(STAGE_OUTPUT_STATE)}")
if name not in stage_backends:
raise ValueError(f"stage {name!r} has no backend in stage_backends")
input_states[name] = previous
previous = output
by_state = store.count_by_state()
done_totals = store.completion_counts_since(0.0)
windowed = store.completion_counts_since(now - window_s)
failures = store.recent_failures(failure_limit)
successes = store.recent_successes(success_limit)
started_at = store.earliest_created_at()
jobs = store.list_jobs()
running_by_stage: dict[str, int] = {}
for job in jobs:
running_by_stage[job.stage] = running_by_stage.get(job.stage, 0) + len(job.material_ids)
failed_by_stage: dict[str, int] = {}
for record in store.query({"state": MaterialState.FAILED}):
stage = record.fail_stage or "?"
failed_by_stage[stage] = failed_by_stage.get(stage, 0) + 1
health = backend_health or {}
window_min = window_s / 60.0
stages = tuple(
StageSnapshot(
name=name,
backend_name=stage_backends[name],
backend_kind=backend_kinds.get(stage_backends[name], "?"),
backend_healthy=health.get(stage_backends[name]),
ready=max(0, by_state.get(input_states[name], 0) - running_by_stage.get(name, 0)),
running=running_by_stage.get(name, 0),
done_total=done_totals.get(name, 0),
failed=failed_by_stage.get(name, 0),
speed_per_min=windowed.get(name, 0) / window_min if window_min > 0 else 0.0,
)
for name in stage_names
)
throughput = windowed.get(FINALIZE_STAGE, 0) / window_min if window_min > 0 else 0.0
live = sum(count for state, count in by_state.items() if not is_terminal(state))
eta_s = (live / throughput) * 60.0 if throughput > 0 and live > 0 else None
tracked: tuple[TrackedMaterial, ...] | None = None
if tracked_ids is not None:
tracked = _build_tracked(store, tracked_ids, jobs, now)
return MonitorSnapshot(
run_name=run_name,
generated_at=now,
started_at=started_at,
elapsed_s=max(0.0, now - started_at) if started_at is not None else 0.0,
total=sum(by_state.values()),
by_state=by_state,
in_flight=sum(len(job.material_ids) for job in jobs),
stages=stages,
recent_failures=tuple(failures),
recent_successes=tuple(successes),
running=_build_running(store, jobs, cpu_by_job or {}, gpu_by_job or {}, now),
throughput_per_min=throughput,
eta_s=eta_s,
stop_after=stop_after,
window_s=window_s,
tracked=tracked,
)
def _build_running(
store: StateStore,
jobs: Sequence[InFlightJob],
cpu_by_job: Mapping[str, float],
gpu_by_job: Mapping[str, float],
now: float,
) -> tuple[RunningMaterial, ...]:
"""Build one ``RunningMaterial`` per material of every live ledger job.
Parameters
----------
store
Read for each running material's ``attempts``.
jobs
The live ledger rows.
cpu_by_job, gpu_by_job
The caller's usage samples, keyed by job key.
now
Timestamp the elapsed times are measured against.
Returns
-------
tuple
Rows sorted longest-running first, with the material id as a
deterministic tie-break. A ledger row whose material is missing from
the store -- foreign, or surgically removed -- still renders with
``attempts=0``: the dashboard must show what the ledger claims is
running, never crash on it.
"""
rows: list[RunningMaterial] = []
for job in sorted(jobs, key=lambda j: j.submitted_at):
cpu = cpu_by_job.get(job.job_key)
gpu = gpu_by_job.get(job.job_key)
for mid in job.material_ids:
record = store.get(mid)
rows.append(
RunningMaterial(
material_id=mid,
stage=job.stage,
backend_name=job.backend_name,
job_key=job.job_key,
elapsed_s=max(0.0, now - job.submitted_at),
attempts=record.attempts if record is not None else 0,
cpu_percent=cpu,
gpu_percent=gpu,
)
)
rows.sort(key=lambda r: (-r.elapsed_s, r.material_id))
return tuple(rows)
def _build_tracked(
store: StateStore,
tracked_ids: Sequence[str],
jobs: Sequence[InFlightJob],
now: float,
) -> tuple[TrackedMaterial, ...]:
"""Resolve the ``--track`` ids into ``TrackedMaterial`` rows.
Parameters
----------
store
Read once per requested id.
tracked_ids
The ids to follow, in the caller's order.
jobs
The already-fetched ledger rows, indexed here by material id so a
material riding a live job picks up its stage, backend, and key
without a second store pass. A material appearing in two jobs cannot
happen -- the schema enforces one live job per material -- but the
first is kept deterministically anyway.
now
Timestamp the in-state durations are measured against.
Returns
-------
tuple
One row per distinct id, caller order preserved, duplicates dropped
keeping the first occurrence. An unknown id becomes a ``found=False``
filler row rather than an exception: the monitor must stay up
regardless of what ids the user typed.
"""
job_by_material: dict[str, InFlightJob] = {}
for job in jobs:
for mid in job.material_ids:
job_by_material.setdefault(mid, job)
rows: list[TrackedMaterial] = []
for mid in dict.fromkeys(tracked_ids):
record = store.get(mid)
if record is None:
rows.append(
TrackedMaterial(
material_id=mid,
found=False,
state=None,
stage=None,
backend_name=None,
in_state_s=0.0,
attempts=0,
fail_reason=None,
job_key=None,
)
)
continue
live_job = job_by_material.get(mid)
rows.append(
TrackedMaterial(
material_id=mid,
found=True,
state=record.state,
stage=live_job.stage if live_job is not None else None,
backend_name=live_job.backend_name if live_job is not None else None,
in_state_s=max(0.0, now - record.updated_at),
attempts=record.attempts,
fail_reason=(record.fail_reason.splitlines()[0] if record.fail_reason else None),
job_key=live_job.job_key if live_job is not None else None,
)
)
return tuple(rows)