feat(v7-l37..45): live WS sparkline + overflow markers + dead-man-switch + reconnect + verify stream

Closes the 9-item live-WS gap audit:

L37 LossChart now drives off liveTrainEvents — sparkline grows as
   events arrive, not only on pipeline.run resolve.
L38 overflowSteps prop on LossChart renders red dashed bars at the
   exact x positions of overflow events on the live curve.
L39 LiveTrainPanel polls Date.now() per second while in-flight;
   when (now - last.ts) > stallSeconds (default 8s) it surfaces
   '⚠ stalled X.Xs' so a hung backend is visible.
L40 finish:'ok' frame raises a one-shot toast 'train done' with
   a dismiss button.
L41 useLiveTrainStream hook reconnects with capped exponential
   backoff (500ms..5s) on unexpected socket close; tracks attempts
   so the UI shows '↻ reconnect #N'. Skipped after the finish frame.
L42 stage_train per-step publish now carries grad_norms dict
   (per-brick L2 norms via _safe_per_brick_grads).
L43 mem_mb on each event from mx.metal.get_peak_memory().
L44 expert_load array on each event when an MoE module is present
   (last_router_output.load).
L45 /ws/verify/{spec_hash} WS endpoint + verify_event_bus + phase
   publishes inside verify() (start → graph_built → resolve_shapes
   → memory_estimated → done). spec_hash helper sha256s canonical
   VerifyParams JSON.

- tests/v4/test_verify_event_bus.py (4 cases): stable hash, phase
  emit, no-subscriber no-op, route presence.
- tests/v5/test_train_event_bus_payload.py: per-step payload has
  grad_norms/mem_mb/expert_load/ts fields.
- e2e 89: real pipeline.run + WS sparkline visible, finish toast,
  overflow path on fp16. page.on('console') capture annotated into
  test.info() so flaky runs leave diagnostic frames in the report.

Author: apstenku123

cppmega_v4/jsonrpc/methods.py

@@ -298,10 +298,23 @@ def verify(params: VerifyParams, *, cache: LRUCache | None = None) -> VerifyResu
     if hit is not None:
         return hit
 
+    # V7-L45: stream progress so /ws/verify/{spec_hash} subscribers see
+    # phase transitions on long verifies.
+    from cppmega_v4.runtime import verify_event_bus as _vb
+    _vh = _vb.spec_hash(params)
+
+    def _emit(phase: str, extra: dict | None = None) -> None:
+        try:
+            _vb.publish(_vh, {"phase": phase, **(extra or {})})
+        except Exception:
+            pass
+
+    _emit("start")
     t0 = time.perf_counter()
     specs = _graph_to_specs(params.graph)
     hidden = params.dim_env.get("H", 64)
     graph = from_block_specs(specs, hidden_size=hidden, instantiate=False)
+    _emit("graph_built", {"node_count": len(specs)})
 
     available = frozenset(params.available_side_channels)
     resolved = resolve_shapes(
@@ -310,12 +323,14 @@ def verify(params: VerifyParams, *, cache: LRUCache | None = None) -> VerifyResu
     )
     fusion_plan = tuple(plan_fusion_regions(graph))
 
+    _emit("resolve_shapes")
     result_one = verify_and_estimate(
         graph,
         dim_env=params.dim_env,
         training=params.training,
         available_side_channels=available,
     )
+    _emit("memory_estimated")
 
     mem = result_one.memory
     per_brick: dict[str, PerBrickMemory] = {}
@@ -462,6 +477,12 @@ def verify(params: VerifyParams, *, cache: LRUCache | None = None) -> VerifyResu
         elapsed_ms=elapsed_ms,
     )
     _cache_store(cache, key, out)
+    _emit("done", {"elapsed_ms": round(
+        (time.perf_counter() - t0) * 1000.0, 4)})
+    try:
+        _vb.publish(_vh, None)   # sentinel marks completion to subs
+    except Exception:
+        pass
     return out
 
 

cppmega_v4/jsonrpc/server.py

@@ -152,6 +152,42 @@ def _try_get(timeout: float = 0.2):
         finally:
             train_event_bus.unsubscribe(run_id, q)
 
+    @app.websocket("/ws/verify/{spec_hash}")
+    async def ws_verify(socket: WebSocket, spec_hash: str) -> None:
+        """V7-L45: live verify progress stream.
+
+        Client subscribes by spec_hash (sha256 of the canonical
+        VerifyParams JSON) *before* calling verify. The handler emits
+        {phase} frames as it walks resolve/memory/distributed checks
+        and a final {finish:'ok'} when verify returns."""
+        from cppmega_v4.runtime import verify_event_bus
+        import queue as _queue
+        await socket.accept()
+        q = verify_event_bus.subscribe(spec_hash)
+
+        def _try_get(timeout: float = 0.2):
+            try:
+                return q.get(timeout=timeout)
+            except _queue.Empty:
+                return _SENTINEL_EMPTY
+
+        try:
+            while True:
+                ev = await asyncio.to_thread(_try_get, 0.2)
+                if ev is _SENTINEL_EMPTY:
+                    await asyncio.sleep(0)
+                    continue
+                if ev is None:
+                    await socket.send_json({"finish": "ok",
+                                             "spec_hash": spec_hash})
+                    return
+                await socket.send_json({"spec_hash": spec_hash,
+                                         "event": ev})
+        except WebSocketDisconnect:
+            pass
+        finally:
+            verify_event_bus.unsubscribe(spec_hash, q)
+
     @app.websocket("/ws")
     async def ws(socket: WebSocket) -> None:
         await socket.accept()

cppmega_v4/runner/stages.py

@@ -206,10 +206,20 @@ def stage_verify_build_spec(ctx: StageContext) -> StageResult:
     _tok = _stage_abort_token(ctx, "verify_build_spec")
     if _tok and _tok in _ABORT_TOKENS:
         clear_abort(_tok)
+        # V7-H06b/H10: mirror the train-side contract so the UI sees
+        # extras.aborted=True on the cancelled stage; mark the run as
+        # aborted in the registry so pipeline.status reflects the
+        # cancel even when train never ran.
+        try:
+            from cppmega_v4.runtime import run_registry as _rr
+            _rr.mark_aborted(_tok)
+        except Exception:
+            pass
         return StageResult(
             name="verify_build_spec", status="cancelled",
             elapsed_ms=(time.perf_counter() - t0) * 1000.0,
             error={"type": "Aborted", "abort_token": _tok},
+            extras={"aborted": True, "abort_token": _tok},
         )
     try:
         if ctx.build_spec is None:
@@ -362,10 +372,16 @@ def stage_dry_forward(ctx: StageContext) -> StageResult:
     _tok = _stage_abort_token(ctx, "dry_forward")
     if _tok and _tok in _ABORT_TOKENS:
         clear_abort(_tok)
+        try:
+            from cppmega_v4.runtime import run_registry as _rr
+            _rr.mark_aborted(_tok)
+        except Exception:
+            pass
         return StageResult(
             name="dry_forward", status="cancelled",
             elapsed_ms=(time.perf_counter() - t0) * 1000.0,
             error={"type": "Aborted", "abort_token": _tok},
+            extras={"aborted": True, "abort_token": _tok},
         )
     try:
         opts = ctx.opts("dry_forward")
@@ -1573,9 +1589,51 @@ def _base(k: str) -> str | None:
             _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.
+            # pipeline completion. V7-L37..44 expansion: payload now also
+            # carries per-brick grad-norms, current memory peak, and
+            # per-expert load (when MoE is present) so the UI can render
+            # live timelines instead of waiting for completion.
             try:
                 from cppmega_v4.runtime import train_event_bus
+                # Per-brick grad-norms (cheap: norm over already-flat
+                # grad tree — see _safe_per_brick_grads).
+                try:
+                    step_grad_norms = _safe_per_brick_grads(
+                        all_modules, grads)
+                except Exception:
+                    step_grad_norms = {}
+                # Per-step memory peak (best-effort; mx.metal may be
+                # unavailable on non-Apple hosts).
+                step_mem_mb: float | None = None
+                try:
+                    if hasattr(mx, "metal"):
+                        step_mem_mb = round(
+                            int(mx.metal.get_peak_memory()) / (1024 * 1024),
+                            3)
+                except Exception:
+                    step_mem_mb = None
+                # Per-step expert load when an MoE brick is in the model.
+                step_expert_load: list[float] | None = None
+                try:
+                    if 'moe_module' in dir() and moe_module is not None:
+                        last_load = getattr(
+                            getattr(moe_module, 'last_router_output',
+                                    None), 'load', None)
+                        if last_load is None:
+                            # Fallback: recompute the router load over the
+                            # cached output if the module exposes it.
+                            ro = getattr(moe_module, 'last_router_output',
+                                          None)
+                            if ro is not None:
+                                last_load = ro.load
+                        if last_load is not None:
+                            step_expert_load = [
+                                round(float(v), 6)
+                                for v in last_load.tolist()
+                            ]
+                except Exception:
+                    step_expert_load = None
+
                 train_event_bus.publish(
                     opts.get("run_id") or opts.get("abort_token"),
                     {"step": int(step),
@@ -1584,7 +1642,11 @@ def _base(k: str) -> str | None:
                              if lr_trajectory else None,
                      "overflow":
                        bool(_scaler_overflow_this_step
-                            if loss_scaler is not None else False)},
+                            if loss_scaler is not None else False),
+                     "grad_norms": step_grad_norms,
+                     "mem_mb": step_mem_mb,
+                     "expert_load": step_expert_load,
+                     "ts": float(time.time())},
                 )
             except Exception:
                 pass

cppmega_v4/runtime/verify_event_bus.py

@@ -0,0 +1,76 @@
+"""V7-L45: cross-thread pub/sub for live verify progress events.
+
+Mirrors train_event_bus + gen_event_bus, keyed by a stable spec_hash
+(sha256 of the canonical JSON of VerifyParams). Used so a long-running
+verify can stream progress to the UI via /ws/verify/{spec_hash}.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json as _json
+import queue
+import threading
+from typing import Any
+
+
+_LOCK = threading.Lock()
+_QUEUES: dict[str, list[queue.Queue]] = {}
+
+
+def spec_hash(spec_payload: Any) -> str:
+    """Stable sha256 over the canonical JSON of a VerifyParams-like dict.
+
+    Accepts a dict or a Pydantic-ish object via model_dump(). Used both
+    by the UI (to build the WS path) and by the verify handler (to know
+    which subscribers to notify)."""
+    if hasattr(spec_payload, "model_dump"):
+        payload = spec_payload.model_dump(mode="json")
+    else:
+        payload = spec_payload
+    return hashlib.sha256(
+        _json.dumps(payload, sort_keys=True,
+                     default=str).encode("utf-8")
+    ).hexdigest()
+
+
+def publish(key: str | None, event: dict[str, Any] | None) -> None:
+    if not key:
+        return
+    with _LOCK:
+        subs = list(_QUEUES.get(key, []))
+    for q in subs:
+        try:
+            q.put_nowait(event)
+        except Exception:
+            pass
+
+
+def subscribe(key: str) -> queue.Queue:
+    q: queue.Queue = queue.Queue(maxsize=256)
+    with _LOCK:
+        _QUEUES.setdefault(key, []).append(q)
+    return q
+
+
+def unsubscribe(key: str, q: queue.Queue) -> None:
+    with _LOCK:
+        subs = _QUEUES.get(key, [])
+        if q in subs:
+            subs.remove(q)
+        if not subs:
+            _QUEUES.pop(key, None)
+
+
+def reset() -> None:
+    with _LOCK:
+        _QUEUES.clear()
+
+
+def subscriber_count(key: str) -> int:
+    with _LOCK:
+        return len(_QUEUES.get(key, []))
+
+
+__all__ = ["spec_hash", "publish", "subscribe", "unsubscribe", "reset",
+            "subscriber_count"]

tests/v4/test_verify_event_bus.py

@@ -0,0 +1,81 @@
+"""V7-L45: verify_event_bus + verify() publish progress events."""
+
+from __future__ import annotations
+
+import queue
+
+import pytest
+from fastapi.testclient import TestClient
+
+from cppmega_v4.jsonrpc import create_app
+from cppmega_v4.jsonrpc.methods import verify
+from cppmega_v4.jsonrpc.schema import VerifyParams
+from cppmega_v4.runtime import verify_event_bus as bus
+
+
+@pytest.fixture(autouse=True)
+def _clean():
+    bus.reset()
+    yield
+    bus.reset()
+
+
+def _spec_payload() -> dict:
+    return {
+        "graph": {
+            "nodes": [
+                {"id": "attn", "kind": "attention", "params": {}},
+                {"id": "mlp", "kind": "mlp",
+                 "params": {"intermediate_size": 64, "activation": "swiglu"}},
+            ],
+            "edges": [{"src": "attn", "dst": "mlp"}],
+        },
+        "dim_env": {"B": 1, "S": 8, "H": 32, "nh": 2, "nkv": 1,
+                    "head_dim": 16},
+        "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_spec_hash_stable_across_equivalent_dicts():
+    h1 = bus.spec_hash(_spec_payload())
+    h2 = bus.spec_hash(_spec_payload())
+    assert h1 == h2
+    assert len(h1) == 64
+
+
+def test_verify_publishes_phase_events_to_subscribers():
+    params = VerifyParams.model_validate(_spec_payload())
+    h = bus.spec_hash(params)
+    q = bus.subscribe(h)
+    verify(params)
+    phases: list[str] = []
+    while True:
+        try:
+            ev = q.get_nowait()
+        except queue.Empty:
+            break
+        if ev is None:
+            phases.append("__finish__")
+            break
+        phases.append(ev["phase"])
+    # The handler emits at least start → graph_built → resolve_shapes
+    # → memory_estimated → done, plus the sentinel.
+    for required in ("start", "graph_built", "memory_estimated",
+                      "done", "__finish__"):
+        assert required in phases, phases
+
+
+def test_verify_no_subscriber_is_a_noop():
+    params = VerifyParams.model_validate(_spec_payload())
+    out = verify(params)
+    assert out.elapsed_ms >= 0
+
+
+def test_ws_verify_endpoint_appears_in_routes():
+    app = create_app(cache_capacity=2)
+    paths = {r.path for r in app.routes
+             if hasattr(r, "path") and getattr(r, "path", "")}
+    assert "/ws/verify/{spec_hash}" in paths