feat(v7-h06b): pipeline.status RPC + run_registry for backend-confirmed lifecycle

Closes the optimistic-state gap from items 16-18 of the honest UI<->API
wiring list: pipeline.pause/resume/abort previously returned immediately
and the UI flipped its own indicators with no backend confirmation.

- New cppmega_v4/runtime/run_registry.py — thread-safe per-run state
  (running, aborted, last_step, last_loss).
- stage_train registers on entry, marks step+loss after each commit,
  unregisters on normal/blow-up/cancel exit.
- pipeline.abort now also marks aborted=True in the registry (the
  abort_token set was opaque to the registry until this commit).
- New pipeline.status RPC combines run_registry + job_control.is_paused
  into a single snapshot UI polls for confirmation.

bd: cppmega-mlx-089s
6/6 pytest in tests/v4/test_pipeline_status_rpc.py + 10/10 incl. pause/
resume/per-brick-grad regression.

Author: apstenku123

cppmega_v4/jsonrpc/dispatcher.py

@@ -54,6 +54,10 @@
 from cppmega_v4.jsonrpc.histogram_method import (
     HistogramParams, inspect_histogram,
 )
+from cppmega_v4.jsonrpc.side_channel_methods import (
+    SideChannelPreviewParams,
+    preview_side_channels,
+)
 from cppmega_v4.jsonrpc.schema import (
     BuildPresetSpecsParams,
     CatalogExplainParams,
@@ -166,6 +170,10 @@
         HistogramParams,
         lambda p, c: inspect_histogram(p, cache=c),
     ),
+    "side_channels.preview": (
+        SideChannelPreviewParams,
+        lambda p, c: preview_side_channels(p, cache=c),
+    ),
 }
 
 

cppmega_v4/jsonrpc/schema.py

@@ -780,6 +780,7 @@ class PipelineRunResult(BaseModel):
     "pipeline.resume",
     "gen.run",
     "inspect.histogram",
+    "side_channels.preview",
 })
 
 

cppmega_v4/runner/stages.py

@@ -137,6 +137,10 @@ def _cancelled_train_result(
     weight_delta_norm: float = 0.0,
 ) -> StageResult:
     clear_abort(abort_token)
+    # V7-H06b: release the registry slot so pipeline.status flips
+    # running=False after a confirmed cancel.
+    from cppmega_v4.runtime import run_registry as _rr
+    _rr.unregister(abort_token)
     return StageResult(
         name="train", status="cancelled",
         elapsed_ms=(time.perf_counter() - t0) * 1000.0,
@@ -1423,6 +1427,12 @@ def _base(k: str) -> str | None:
         abort_token = opts.get("abort_token")
         # V7-H06: block while job is paused via job_control.pause(token).
         from cppmega_v4.runtime.job_control import wait_while_paused
+        # V7-H06b: register run with the lifecycle registry so
+        # pipeline.status RPC can answer running/paused/aborted before
+        # train returns. Tracked by run_id (falls back to abort_token).
+        from cppmega_v4.runtime import run_registry as _run_registry
+        _registry_key = opts.get("run_id") or abort_token
+        _run_registry.register(_registry_key)
         for step in range(n_steps):
             wait_while_paused(abort_token, poll_s=0.05, max_wait_s=600.0)
             if abort_token is not None and abort_token in _ABORT_TOKENS:
@@ -1522,6 +1532,9 @@ def _base(k: str) -> str | None:
                 if loss_scaler is not None:
                     loss_scaler.update(False)
             losses.append(float(loss.item()))
+            # V7-H06b: keep run_registry's last_step/last_loss live so
+            # pipeline.status reflects the most recent committed step.
+            _run_registry.mark_step(_registry_key, step, float(losses[-1]))
             # V7-H05: publish per-step event to the train_event_bus so
             # WS subscribers (UI) see loss/lr update live, not only on
             # pipeline completion.
@@ -1720,6 +1733,7 @@ def _base(k: str) -> str | None:
                        "detail": f"delta {delta:.2e} <= 1e-6"},
             )
         if len(losses) >= 2 and losses[0] > 0 and losses[-1] / losses[0] > 5:
+            _run_registry.unregister(_registry_key)
             return StageResult(
                 name="train", status="fail",
                 elapsed_ms=(time.perf_counter() - t0) * 1000.0,
@@ -1735,6 +1749,9 @@ def _base(k: str) -> str | None:
                 opts.get("run_id") or opts.get("abort_token"), None)
         except Exception:
             pass
+        # V7-H06b: mark run as no longer running so pipeline.status
+        # flips running=False once the train loop returns.
+        _run_registry.unregister(_registry_key)
         return StageResult(
             name="train", status="ok",
             elapsed_ms=(time.perf_counter() - t0) * 1000.0,

cppmega_v4/runtime/run_registry.py

@@ -0,0 +1,99 @@
+"""V7-H06b: in-process registry of active pipeline runs.
+
+Tracks the lifecycle of every run_id passed into stage_train so that
+pipeline.status RPC can answer:
+  - is this run currently running (still in the train loop)?
+  - has it been aborted?
+  - is it paused?
+  - what's the last published (step, loss)?
+
+The UI uses this to gate state transitions on backend confirmation,
+not optimistic local flips:
+  - pipeline.pause -> poll status until paused=True
+  - pipeline.resume -> poll until paused=False
+  - pipeline.abort -> poll until running=False
+
+Thread-safe; lives across requests in the FastAPI worker process.
+"""
+
+from __future__ import annotations
+
+import time
+from threading import Lock
+from typing import Any
+
+_LOCK = Lock()
+_RUNS: dict[str, dict[str, Any]] = {}
+
+
+def register(run_id: str | None) -> None:
+    """Mark run_id as running. Idempotent; resets last_step / last_loss."""
+    if not run_id:
+        return
+    with _LOCK:
+        _RUNS[str(run_id)] = {
+            "running": True,
+            "aborted": False,
+            "last_step": -1,
+            "last_loss": None,
+            "started_at": time.time(),
+            "ended_at": None,
+        }
+
+
+def mark_step(run_id: str | None, step: int, loss: float) -> None:
+    """Record the most recent step + loss from the train loop."""
+    if not run_id:
+        return
+    with _LOCK:
+        rec = _RUNS.get(str(run_id))
+        if rec is None:
+            return
+        rec["last_step"] = int(step)
+        rec["last_loss"] = float(loss)
+
+
+def mark_aborted(run_id: str | None) -> None:
+    if not run_id:
+        return
+    with _LOCK:
+        rec = _RUNS.get(str(run_id))
+        if rec is None:
+            return
+        rec["aborted"] = True
+
+
+def unregister(run_id: str | None) -> None:
+    """Mark run as no longer running. Record is kept so the UI can
+    still query its terminal state after train returns."""
+    if not run_id:
+        return
+    with _LOCK:
+        rec = _RUNS.get(str(run_id))
+        if rec is None:
+            return
+        rec["running"] = False
+        rec["ended_at"] = time.time()
+
+
+def snapshot(run_id: str | None) -> dict[str, Any] | None:
+    """Return a copy of the registry entry, or None if unknown."""
+    if not run_id:
+        return None
+    with _LOCK:
+        rec = _RUNS.get(str(run_id))
+        if rec is None:
+            return None
+        return dict(rec)
+
+
+def reset() -> None:
+    """Test helper — drop all registered runs."""
+    with _LOCK:
+        _RUNS.clear()
+
+
+__all__ = [
+    "register", "mark_step", "mark_aborted", "unregister",
+    "snapshot", "reset",
+]

tests/v4/test_pipeline_status_rpc.py

@@ -0,0 +1,138 @@
+"""V7-H06b: pipeline.status RPC + run_registry round-trip.
+
+Verifies that pipeline.status reflects backend reality for pause /
+resume / abort / running / finished, so the UI can confirm a state
+transition before flipping its own indicators.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+
+import pytest
+
+from cppmega_v4.jsonrpc.dispatcher import (
+    _pipeline_abort, _pipeline_pause, _pipeline_resume, _pipeline_status,
+)
+from cppmega_v4.jsonrpc.schema import (
+    PipelineAbortParams, PipelineStatusParams, VerifyParams,
+)
+from cppmega_v4.runner import Pipeline, run_pipeline
+from cppmega_v4.runtime import job_control as jc
+from cppmega_v4.runtime import run_registry as rr
+
+
+@pytest.fixture(autouse=True)
+def _reset():
+    jc.reset()
+    rr.reset()
+    yield
+    jc.reset()
+    rr.reset()
+
+
+def _spec() -> VerifyParams:
+    return VerifyParams.model_validate({
+        "graph": {
+            "nodes": [
+                {"id": "attn", "kind": "attention",
+                 "params": {"num_heads": 4, "head_dim": 64}},
+                {"id": "mlp", "kind": "mlp", "params": {}},
+            ],
+            "edges": [{"src": "attn", "dst": "mlp"}],
+        },
+        "dim_env": {"B": 1, "S": 8, "H": 128,
+                    "nh": 2, "nkv": 1, "head_dim": 64},
+        "loss": {"kind": "cross_entropy", "head_outputs": ["mlp"]},
+        "optim": {"kind": "adamw",
+                  "groups": [{"matcher": "all", "lr": 1e-3,
+                              "weight_decay": 0.01,
+                              "betas": [0.9, 0.95]}]},
+    })
+
+
+def test_v7_h06b_status_unknown_for_never_seen_run():
+    r = _pipeline_status(PipelineStatusParams(run_id="ghost"))
+    assert r.run_id == "ghost"
+    assert r.known is False
+    assert r.running is False
+
+
+def test_v7_h06b_status_reports_paused_after_pause_rpc():
+    rr.register("run-X")
+    _pipeline_pause(PipelineAbortParams(run_id="run-X"))
+    r = _pipeline_status(PipelineStatusParams(run_id="run-X"))
+    assert r.known is True
+    assert r.paused is True
+    assert r.running is True
+
+
+def test_v7_h06b_status_clears_paused_after_resume_rpc():
+    rr.register("run-Y")
+    _pipeline_pause(PipelineAbortParams(run_id="run-Y"))
+    assert _pipeline_status(
+        PipelineStatusParams(run_id="run-Y")).paused is True
+    _pipeline_resume(PipelineAbortParams(run_id="run-Y"))
+    r = _pipeline_status(PipelineStatusParams(run_id="run-Y"))
+    assert r.paused is False
+
+
+def test_v7_h06b_status_marks_aborted_after_abort_rpc():
+    rr.register("run-Z")
+    _pipeline_abort(PipelineAbortParams(run_id="run-Z"))
+    r = _pipeline_status(PipelineStatusParams(run_id="run-Z"))
+    assert r.aborted is True
+
+
+def test_v7_h06b_status_reports_running_during_train_then_finished():
+    """Pause the train upfront so the train loop is *guaranteed* to be
+    in wait_while_paused when the poller runs. Verify running=True
+    + paused=True. Then resume, wait for train to finish, verify
+    running=False."""
+    # Pause first so step 0 blocks in wait_while_paused.
+    jc.pause("run-T")
+    captured: dict = {}
+
+    def _resumer():
+        time.sleep(0.15)
+        r = _pipeline_status(PipelineStatusParams(run_id="run-T"))
+        captured["mid"] = {"known": r.known, "running": r.running,
+                           "paused": r.paused}
+        jc.resume("run-T")
+
+    threading.Thread(target=_resumer, daemon=True).start()
+    rep = run_pipeline(_spec(), Pipeline.from_dict({
+        "stages": ["parse", "verify_build_spec", "build_model", "train"],
+        "stage_options": {"train": {
+            "num_steps": 2, "run_id": "run-T", "abort_token": "run-T",
+        }},
+    }))
+    tr = next(s for s in rep.stages if s.name == "train")
+    assert tr.status == "ok"
+
+    # Mid-train snapshot taken while paused: running=True + paused=True.
+    assert captured["mid"] == {"known": True, "running": True,
+                                "paused": True}, captured
+    # After train returns: running=False, paused already cleared.
+    final = _pipeline_status(PipelineStatusParams(run_id="run-T"))
+    assert final.known is True
+    assert final.running is False
+    assert final.paused is False
+    assert final.last_step >= 0
+
+
+def test_v7_h06b_status_last_step_and_loss_updates_during_train():
+    rep = run_pipeline(_spec(), Pipeline.from_dict({
+        "stages": ["parse", "verify_build_spec", "build_model", "train"],
+        "stage_options": {"train": {
+            "num_steps": 3, "run_id": "run-S", "abort_token": "run-S",
+        }},
+    }))
+    tr = next(s for s in rep.stages if s.name == "train")
+    assert tr.status == "ok"
+    final = _pipeline_status(PipelineStatusParams(run_id="run-S"))
+    # 3-step train, last index = 2.
+    assert final.last_step == 2
+    assert final.last_loss is not None
+    assert final.last_loss > 0