Monitoring#

The read-only run snapshot and the usage sampling that fills it.

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).

class matmelt.core.monitor.StageSnapshot(name, backend_name, backend_kind, backend_healthy, ready, running, done_total, failed, speed_per_min)[source]#

Bases: object

One configured stage’s live numbers, as rendered in the pipeline panel.

name#

The stage’s canonical name.

Type:

str

backend_name#

Which configured backend runs it.

Type:

str

backend_kind#

That backend’s kind, or "?" when the caller did not supply one.

Type:

str

backend_healthy#

The caller’s cached health_check verdict. None means “not probed”, rendered neutrally and never as a failure.

Type:

bool or None

ready#

Materials sitting at this stage’s input state and not in flight. Disjoint from running by construction.

Type:

int

running#

Materials linked to a live ledger job of this stage.

Type:

int

done_total#

All-time completions of this stage, counted from the ledger, so a fan-out parent stays counted after it retires as EXPANDED.

Type:

int

failed#

Materials terminally FAILED at this stage.

Type:

int

speed_per_min#

Completions of this stage within the rolling window, per minute; 0.0 when idle.

Type:

float

class matmelt.core.monitor.RunningMaterial(material_id, stage, backend_name, job_key, elapsed_s, attempts, cpu_percent, gpu_percent)[source]#

Bases: object

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.

material_id#

The material riding the job.

Type:

str

stage#

Which stage’s job it is.

Type:

str

backend_name#

Where that job is running.

Type:

str

job_key#

The ledger’s key for the job, and the key the usage samplers report against.

Type:

str

elapsed_s#

Seconds since the job was ledgered, i.e. how long this material has been computing in its current job.

Type:

float

attempts#

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.

Type:

int

cpu_percent#

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.

Type:

float or None

gpu_percent#

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.

Type:

float or None

class matmelt.core.monitor.TrackedMaterial(material_id, found, state, stage, backend_name, in_state_s, attempts, fail_reason, job_key)[source]#

Bases: object

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.

material_id#

The id the caller asked to follow, echoed verbatim.

Type:

str

found#

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.

Type:

bool

state#

The material’s current lifecycle state.

Type:

MaterialState or None

stage#

The stage of the ledger job currently carrying this material, or None when it sits between jobs (or is terminal).

Type:

str or None

backend_name#

Where that job runs, same None rule.

Type:

str or None

in_state_s#

Seconds since the record’s updated_at, i.e. time in the current state. 0.0 when not found.

Type:

float

attempts#

Consecutive failed attempts at the current stage.

Type:

int

fail_reason#

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.

Type:

str or None

job_key#

The ledger key of the carrying job, same None rule as stage.

Type:

str or None

class matmelt.core.monitor.MonitorSnapshot(run_name, generated_at, started_at, elapsed_s, total, by_state, in_flight, stages, recent_failures, recent_successes, running, throughput_per_min, eta_s, stop_after, window_s, tracked=None)[source]#

Bases: object

Everything one refresh tick of the monitor shows, fully computed.

run_name#

The configured run name, shown in the header.

Type:

str

generated_at#

The now this snapshot was computed against.

Type:

float

started_at#

The run’s wall-clock anchor – the earliest material’s creation time. None on an empty store.

Type:

float or None

elapsed_s#

Time since started_at; 0.0 on an empty store.

Type:

float

total#

How many materials the store holds.

Type:

int

by_state#

The full per-state tally. Absent states are zero-filled by the renderer, not here.

Type:

dict

in_flight#

Materials currently linked to live ledger jobs.

Type:

int

stages#

One StageSnapshot per configured stage, in funnel order.

Type:

tuple

recent_failures#

Newest-first terminal failures. Each record carries fail_stage, fail_reason, and updated_at as the failed-at time.

Type:

tuple

recent_successes#

Newest-first DONE materials, where updated_at is the finish time and updated_at - created_at is how long the ride took.

Type:

tuple

running#

One RunningMaterial per material on a live ledger job, longest-running first.

Type:

tuple

throughput_per_min#

Run-level finish rate over the rolling window, counted from FINALIZE_STAGE ledger rows.

Type:

float

eta_s#

The naive remaining-over-throughput estimate, and None whenever throughput is zero – the renderer shows a dash rather than dividing by zero.

Type:

float or None

stop_after#

The configured stop stage, echoed for the header.

Type:

str or None

window_s#

Width of the rolling window the rates were computed over.

Type:

float

tracked#

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.

Type:

tuple or None

matmelt.core.monitor.build_snapshot(store, *, run_name, stage_names, stage_backends, backend_kinds, backend_health=None, stop_after=None, now=None, window_s=300.0, failure_limit=10, success_limit=10, cpu_by_job=None, gpu_by_job=None, tracked_ids=None)[source]#

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 (StateStore) – The state store to read. Every query the tick needs happens here, once.

  • run_name (str) – The configured run name, passed through to the snapshot.

  • stage_names (Sequence[str]) – The configured funnel in order. Canonical names only.

  • stage_backends (Mapping[str, str]) – Each stage’s backend name – plain strings, so this module never touches config or backend objects.

  • backend_kinds (Mapping[str, str]) – Each backend name’s kind. A name absent here renders as "?".

  • backend_health (Mapping[str, bool] | None, default: None) – The caller’s cached probe results. A missing entry means “unknown” and renders neutrally.

  • stop_after (str | None, default: None) – The configured stop stage, echoed into the snapshot.

  • now (float | None, default: None) – The timestamp to compute against. Injectable for tests; defaults to time.time().

  • window_s (float, default: 300.0) – Width of the rolling window for the per-stage speeds and the run throughput.

  • failure_limit (int, default: 10) – How many recent failures to carry.

  • success_limit (int, default: 10) – How many recent successes to carry.

  • cpu_by_job (Mapping[str, float] | None, default: None) – 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.

  • gpu_by_job (Mapping[str, float] | None, default: None) – 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 (Sequence[str] | None, default: None) – 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:

Everything one refresh tick renders, with no further queries needed.

Return type:

MonitorSnapshot

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.

Per-job CPU/GPU usage sampling for the monitor: best-effort, never fatal.

The dashboard’s usage columns are cosmetic telemetry, and this module treats them that way everywhere: every function returns None (or an empty mapping) instead of raising, a dead PID or unreachable host is a dash on screen, and nothing here is ever allowed to take the monitor down.

The seam mirrors backend health probing: backends that can measure their jobs expose the duck-typed hooks

sample_cpu_percents(handles) -> list[float | None] sample_gpu_percents(handles) -> list[float | None]

taking the ledger’s serialized handle dicts (asdict(Handle)) in order and answering one reading (or None) per handle, never raising. For CPU the local backend answers from /proc deltas (true instantaneous usage of the job’s whole process subtree, via LocalCpuTracker below) and ssh_exec from one batched remote ps (lifetime-average pcpu, ps semantics – honest enough for long-running science jobs). For GPU both answer from nvidia-smi pmon (per-process SM utilisation, summed over the job’s subtree); a machine without nvidia-smi – or a backend without the hook – simply contributes no readings, so a CPU-only rig shows a dash, never an error. UsageSampler groups the in-flight ledger by backend, applies a per-backend TTL (local sampling is free and can run every tick; a remote sample is an SSH round-trip and is cached, the _HealthCache discipline), and folds everything into the {job_key: percent} mappings build_snapshot takes as cpu_by_job and gpu_by_job.

The /proc/pmon parsing and subtree arithmetic live here as pure, injectable functions so both the backends and the tests use one implementation.

matmelt.core.cpusample.REMOTE_CPU_TTL_S = 10.0#

How long one remote CPU sample stays fresh (seconds). Remote sampling is an SSH round-trip; local sampling passes 0.0 (every tick).

matmelt.core.cpusample.clock_ticks_per_s()[source]#

The kernel’s USER_HZ, the unit /proc CPU times are counted in.

Returns:

Clock ticks per second, or _DEFAULT_HZ (100) where sysconf cannot answer – a wrong-but-sane divisor beats no reading at all for cosmetic telemetry.

Return type:

float

matmelt.core.cpusample.parse_proc_stat(text)[source]#

Parse one /proc/<pid>/stat line.

The comm field is parenthesised and may itself contain spaces or parentheses ((Web Content)), so fields are taken after the LAST ) – the only parse that is correct for every process name.

Parameters:

text (str) – The full contents of a /proc/<pid>/stat file.

Returns:

(pid, ppid, cpu_ticks), where cpu_ticks is utime + stime (user + system jiffies of this process alone, excluding children). Anything malformed is None, never an exception.

Return type:

tuple or None

Examples

>>> parse_proc_stat("42 (my proc) S 7 42 42 0 -1 0 0 0 0 0 30 12 0 0")
(42, 7, 42.0)
>>> parse_proc_stat("not a stat line") is None
True
matmelt.core.cpusample.scan_proc(proc_root=PosixPath('/proc'))[source]#

Read every process’s parent and CPU time in one pass.

Parameters:

proc_root (Path, default: PosixPath('/proc')) – Where to scan. Injectable so tests can build a fake /proc in tmp_path; the default is the real one.

Returns:

{pid: (ppid, cpu_ticks)}. Unreadable or vanished entries (processes exit mid-scan constantly) are skipped silently, and an absent proc_root yields an empty table.

Return type:

dict

matmelt.core.cpusample.subtree_pids(table, root)[source]#

Collect root and every process descended from it.

A job is a process tree – the wrapper shell plus the runner plus whatever multiprocessing spawned – so per-job accounting must cover the whole subtree, not just the pid the backend recorded.

Parameters:

  • table (Mapping[int, tuple[int, float]]) – {pid: (ppid, value)}, as returned by scan_proc.

  • root (int) – The pid to start from.

Returns:

root plus all descendants, or None when root is not in the table (the process is gone). A visited set guards traversal against a corrupt table; real ppid chains cannot cycle.

Return type:

set of int or None

Examples

>>> table = {1: (0, 0.0), 2: (1, 0.0), 3: (2, 0.0), 9: (0, 0.0)}
>>> sorted(subtree_pids(table, 1))
[1, 2, 3]
>>> subtree_pids(table, 404) is None
True
matmelt.core.cpusample.subtree_total(table, root)[source]#

Sum the per-process values over root and all its descendants.

Parameters:

  • table (Mapping[int, tuple[int, float]]) – {pid: (ppid, value)} – CPU ticks locally, ps pcpu remotely.

  • root (int) – The pid whose subtree is summed.

Returns:

The subtree total, or None when root is not in the table (the process is gone). See subtree_pids for the traversal contract.

Return type:

float or None

matmelt.core.cpusample.parse_pmon(text)[source]#

Parse nvidia-smi pmon -c 1 -s u output into per-process SM usage.

Parameters:

text (str) – The command’s stdout. Its table is gpu pid type sm mem enc dec command, with # header lines and - for idle or unavailable cells.

Returns:

{pid: sm_percent}. A - cell reads as 0.0 – a listed process doing no GPU work is at 0% – and a pid appearing on several GPUs is summed. Garbage lines are skipped; a machine without the tool never reaches this parser, because the caller answers None instead.

Return type:

dict

Examples

>>> parse_pmon("# gpu pid type sm mem enc dec command\n0 1234 C 45 12 - - python3")
{1234: 45.0}
matmelt.core.cpusample.gpu_percent_for_roots(process_table, pmon, roots)[source]#

Per-root GPU percentages: pmon SM readings summed over each subtree.

Parameters:

  • process_table (Mapping[int, tuple[int, float]]) – {pid: (ppid, value)} for the machine the jobs run on, used only for its parent links.

  • pmon (Mapping[int, float]) – {pid: sm_percent} from parse_pmon.

  • roots (Sequence[int | None]) – One root pid per job, in the caller’s order. None marks a job whose pid is not known.

Returns:

One reading per root, in the same order. None roots and roots absent from the process table (gone) stay None; a live subtree with no pmon entries is an honest 0.0 – running, but not on the GPU.

Return type:

list

class matmelt.core.cpusample.LocalCpuTracker(*, clock=<built-in function monotonic>, proc_root=PosixPath('/proc'), ticks_per_s=None)[source]#

Bases: object

Delta-based CPU usage for local process subtrees, one /proc scan per call.

Readings are a percentage of one core, so a 4-worker job legitimately reads ~400%. State for vanished pids is dropped, so a recycled PID can never inherit a stale baseline.

Parameters:

  • clock (Callable[[], float], default: <built-in function monotonic>) – Monotonic time source, injectable for tests.

  • proc_root (Path, default: PosixPath('/proc')) – Where to scan, injectable for tests.

  • ticks_per_s (float | None, default: None) – Override for the kernel’s USER_HZ. None asks clock_ticks_per_s.

sample(root_pids)[source]#

Take one reading per requested pid, against the previous call.

Parameters:

root_pids (Sequence[int]) – The job root pids to measure.

Returns:

{pid: percent or None}. The percentage is the subtree’s tick delta since this tracker last saw that pid, over the wall-clock delta. The first sighting of a pid has no delta and answers None – an honest “measuring”, never a fabricated 0.0 – as does a pid that has exited.

Return type:

dict

class matmelt.core.cpusample.UsageSampler(backends, *, hook_name='sample_cpu_percents', ttl_by_backend=None, default_ttl_s=10.0)[source]#

Bases: object

The monitor’s per-tick usage sweep over the in-flight ledger.

Groups jobs by backend, asks each backend exposing the sampling hook for readings, and folds the answers into {job_key: percent}. Jobs without a reading are simply absent, so the renderer’s .get shows a dash. A hook that raises despite its contract is treated as “no readings” for that sweep – telemetry must never take the dashboard down.

Parameters:

  • backends (Mapping[str, object]) – {backend_name: backend}. A backend without the hook contributes nothing, which is how a CPU-only rig shows dashes rather than errors.

  • hook_name (str, default: 'sample_cpu_percents') – Which duck-typed hook to call: sample_cpu_percents by default. Construct a second sampler with sample_gpu_percents for the GPU column.

  • ttl_by_backend (Mapping[str, float] | None, default: None) – Per-backend cache lifetime in seconds. The caller passes 0.0 for backends whose sampling is free (local, every tick).

  • default_ttl_s (float, default: 10.0) – Lifetime for backends not named in ttl_by_backend, so an SSH round-trip happens at most every REMOTE_CPU_TTL_S regardless of the dashboard’s refresh interval.

sample(jobs, now)[source]#

Read usage for the given in-flight jobs, throttled per backend.

Parameters:

  • jobs (Sequence[InFlightJob]) – The in-flight ledger rows to measure, in any order.

  • now (float) – The current monotonic time, against which each backend’s cache TTL is judged. The caller owns the clock so a tick measures once.

Returns:

{job_key: percent}, merged across backends. A backend still inside its TTL contributes its cached readings unchanged.

Return type:

dict