feat(v7-h37): UI subscribes to /ws/verify/{spec_hash} progress stream

apstenku123 · apstenku123 · commit c3b95e93a0ae · 2026-05-23T03:57:49.000+02:00
Backend verify() (V7-L45) publishes phase events to verify_event_bus,
exposed via the /ws/verify/{spec_hash} WS endpoint. UI had no
consumer — long verifies on big graphs looked frozen.

- New useVerifyStream hook owns the WS lifecycle keyed by spec_hash.
- canonicalJson + computeSpecHash JS helpers match the backend's
  json.dumps(sort_keys=True, default=str) + sha256 contract bit-exact.
- App.tsx computes hash before each verify, flips verifyInFlight, hook
  opens WS, panel shows 'verify · &lt;latest phase&gt; (N phases)' badge.
- finally{} cleans up the stream regardless of verify outcome.

bd: cppmega-mlx-lex9
2/2 backend pytest (hash determinism + start/resolve_shapes phases)
+ 7/7 vitest (canonicalJson + WS lifecycle via MockSocket).
diff --git a/tests/v4/test_verify_event_bus_phase_stream.py b/tests/v4/test_verify_event_bus_phase_stream.py
@@ -0,0 +1,83 @@
+"""V7-H37: verify() publishes phase events to verify_event_bus so the
+UI subscriber on /ws/verify/{spec_hash} sees progress."""
+
+from __future__ import annotations
+
+import threading
+import time
+
+import pytest
+
+from cppmega_v4.jsonrpc.methods import verify
+from cppmega_v4.jsonrpc.schema import VerifyParams
+from cppmega_v4.runtime import verify_event_bus as vb
+
+
+@pytest.fixture(autouse=True)
+def _reset():
+    vb.reset()
+    yield
+    vb.reset()
+
+
+def _spec_params() -> 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_h37_spec_hash_is_stable_for_equivalent_payloads():
+    p1 = _spec_params()
+    p2 = _spec_params()
+    assert vb.spec_hash(p1) == vb.spec_hash(p2)
+    assert len(vb.spec_hash(p1)) == 64  # sha256 hex
+
+
+def test_v7_h37_verify_publishes_phase_events_then_finish():
+    params = _spec_params()
+    h = vb.spec_hash(params)
+
+    collected: list[dict | None] = []
+
+    def _collect():
+        q = vb.subscribe(h)
+        while True:
+            try:
+                ev = q.get(timeout=3.0)
+            except Exception:
+                break
+            collected.append(ev)
+            if ev is None:
+                break
+
+    t = threading.Thread(target=_collect, daemon=True)
+    t.start()
+    time.sleep(0.05)  # let subscribe land before publish
+
+    verify(params)
+    # The verify handler doesn't emit a sentinel — the WS endpoint does
+    # on disconnect. So we wait until at least the start + resolve
+    # frames land, then drain.
+    deadline = time.time() + 1.0
+    while time.time() < deadline and len(collected) < 2:
+        time.sleep(0.02)
+    vb.publish(h, None)  # close the subscriber loop
+    t.join(timeout=1.0)
+
+    phases = [e["phase"] for e in collected if isinstance(e, dict)]
+    assert "start" in phases, phases
+    assert "resolve_shapes" in phases, phases
diff --git a/vbgui/src/App.tsx b/vbgui/src/App.tsx
@@ -27,6 +27,7 @@ import { DataInspector } from "@/components/DataInspector";
 import { BrickContextPanel } from "@/components/BrickContextPanel";
 import { LiveTrainPanel } from "@/components/LiveTrainPanel";
 import { useLiveTrainStream } from "@/hooks/useLiveTrainStream";
+import { useVerifyStream, computeSpecHash } from "@/hooks/useVerifyStream";
 
 import { useRpc } from "@/hooks/useRpc";
 import { useVerifyAfter } from "@/hooks/useVerifyAfter";
@@ -220,13 +221,34 @@ export function App(): JSX.Element {
     setTrainSideChannels((prev) => prev.filter((name) => available.has(name)));
   }, [availableSideChannels]);
 
+  // V7-H37: verify-progress subscriber. computeSpecHash matches the
+  // backend's verify_event_bus.spec_hash so the UI WS path lines up
+  // with the publisher inside the verify handler.
+  const [verifyInFlight, setVerifyInFlight] = useState<boolean>(false);
+  const [verifySpecHash, setVerifySpecHash] = useState<string | null>(null);
+  const verifyStream = useVerifyStream(
+    (import.meta.env.VITE_BACKEND_URL as string | undefined)
+      ?? "http://127.0.0.1:8765",
+    verifySpecHash, verifyInFlight,
+  );
+
   const runVerify = useCallback(async () => {
     const snap = wireSpecRef.current;
     if (snap.nodes.length === 0) return;
     const params = buildVerifyParams(
       snap.nodes, snap.edges, snap.spec, snap.availableSideChannels,
       snap.dimEnv,
     );
+    // V7-H37: open the verify-progress WS *before* the RPC, then close
+    // it on completion. The hook owns the lifecycle once verifyInFlight
+    // flips.
+    try {
+      const hash = await computeSpecHash(params);
+      setVerifySpecHash(hash);
+      setVerifyInFlight(true);
+    } catch {
+      // WebCrypto unavailable (jsdom non-secure); skip progress feed.
+    }
     try {
       const r = await rpc.call<{
         memory_distributed?: { worst_rank?: { total_bytes?: number };
@@ -262,6 +284,9 @@ export function App(): JSX.Element {
       }
     } catch {
       // Backend down or invalid spec; leave state and let user retry.
+    } finally {
+      // V7-H37: tear down the progress WS regardless of verify outcome.
+      setVerifyInFlight(false);
     }
   }, [rpc]);
 
@@ -736,6 +761,24 @@ export function App(): JSX.Element {
     <ReactFlowProvider>
       <div style={{ display: "flex", flexDirection: "column",
                     height: "100vh", margin: 0 }}>
+        {/* V7-H37: verify-progress badge — visible while verify is in
+            flight, shows the most recent phase emitted by the backend
+            so the user knows a long graph verify is making progress
+            (not hung). */}
+        {verifyInFlight && verifyStream.events.length > 0 && (
+          <div data-testid="verify-progress-badge"
+               style={{ position: "fixed", top: 50, right: 14,
+                        background: "#fef3c7", color: "#92400e",
+                        padding: "4px 10px", borderRadius: 4,
+                        fontFamily: "monospace", fontSize: 11,
+                        border: "1px solid #fcd34d", zIndex: 900 }}>
+            verify · {verifyStream.events[verifyStream.events.length - 1]
+                                          ?.phase}
+            <span style={{ marginLeft: 6, opacity: 0.7 }}>
+              ({verifyStream.events.length} phases)
+            </span>
+          </div>
+        )}
         <TopBar
           state={spec}
           projectName={projectName}
diff --git a/vbgui/src/hooks/useVerifyStream.ts b/vbgui/src/hooks/useVerifyStream.ts
@@ -0,0 +1,107 @@
+// V7-H37: WebSocket subscriber for /ws/verify/{spec_hash}.
+//
+// Mirrors useLiveTrainStream — opens the WS when active=true + specHash
+// is known, collects phase events ({phase, ...}) into a buffer, surfaces
+// the latest one + a count for a progress indicator. Auto-disconnects
+// on backend {finish:'ok'} frame.
+
+import { useEffect, useRef, useState } from "react";
+
+export interface VerifyPhaseEvent {
+  phase: string;
+  [k: string]: unknown;
+}
+
+export interface UseVerifyStreamState {
+  events: VerifyPhaseEvent[];
+  finished: boolean;
+  reset: () => void;
+}
+
+export function useVerifyStream(
+  baseUrl: string,
+  specHash: string | null,
+  active: boolean,
+): UseVerifyStreamState {
+  const [events, setEvents] = useState<VerifyPhaseEvent[]>([]);
+  const [finished, setFinished] = useState<boolean>(false);
+  const socketRef = useRef<WebSocket | null>(null);
+
+  const reset = () => { setEvents([]); setFinished(false); };
+
+  useEffect(() => {
+    if (!active || !specHash) {
+      socketRef.current?.close();
+      socketRef.current = null;
+      return;
+    }
+
+    let cancelled = false;
+    setEvents([]); setFinished(false);
+
+    const url = baseUrl.replace(/^http/, "ws")
+              + `/ws/verify/${specHash}`;
+    let socket: WebSocket;
+    try {
+      socket = new WebSocket(url);
+    } catch {
+      return;
+    }
+    socketRef.current = socket;
+
+    socket.onmessage = (msg) => {
+      if (cancelled) return;
+      try {
+        const frame = JSON.parse(msg.data) as
+          { event?: VerifyPhaseEvent; finish?: string };
+        if (frame.event) {
+          setEvents((prev) => [...prev, frame.event!]);
+        } else if (frame.finish) {
+          setFinished(true);
+          socket.close();
+        }
+      } catch { /* ignore malformed */ }
+    };
+
+    socket.onerror = () => { /* swallow; reconnect via effect dep */ };
+
+    return () => {
+      cancelled = true;
+      try { socket.close(); } catch { /* noop */ }
+      socketRef.current = null;
+    };
+  }, [baseUrl, specHash, active]);
+
+  return { events, finished, reset };
+}
+
+/**
+ * Canonical-JSON serializer matching backend cppmega_v4.runtime.
+ * verify_event_bus.spec_hash (json.dumps sort_keys=True default=str).
+ */
+export function canonicalJson(value: unknown): string {
+  if (value === null || typeof value !== "object") {
+    return JSON.stringify(value);
+  }
+  if (Array.isArray(value)) {
+    return "[" + value.map(canonicalJson).join(",") + "]";
+  }
+  const obj = value as Record<string, unknown>;
+  const keys = Object.keys(obj).sort();
+  const parts = keys.map(
+    (k) => JSON.stringify(k) + ":" + canonicalJson(obj[k]));
+  return "{" + parts.join(",") + "}";
+}
+
+/**
+ * Compute SHA-256 of canonical JSON, hex-encoded. Matches the
+ * backend's `spec_hash`. Uses WebCrypto so requires HTTPS or
+ * localhost (jsdom-test environments need a polyfill).
+ */
+export async function computeSpecHash(value: unknown): Promise<string> {
+  const buf = new TextEncoder().encode(canonicalJson(value));
+  const digest = await crypto.subtle.digest("SHA-256", buf);
+  return Array.from(new Uint8Array(digest))
+    .map((b) => b.toString(16).padStart(2, "0"))
+    .join("");
+}
diff --git a/vbgui/tests/useVerifyStream.test.tsx b/vbgui/tests/useVerifyStream.test.tsx
@@ -0,0 +1,97 @@
+// V7-H37: canonicalJson matches Python json.dumps(sort_keys=True);
+// useVerifyStream collects WS events + closes on finish frame.
+
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import { renderHook, act, waitFor } from "@testing-library/react";
+import { canonicalJson, useVerifyStream }
+  from "@/hooks/useVerifyStream";
+
+describe("V7-H37 canonicalJson", () => {
+  it("sorts object keys deterministically (matches Python sort_keys)", () => {
+    const out = canonicalJson({ b: 2, a: { d: 4, c: 3 } });
+    expect(out).toBe('{"a":{"c":3,"d":4},"b":2}');
+  });
+
+  it("emits arrays in order without key-sort effect", () => {
+    expect(canonicalJson([3, 1, 2])).toBe("[3,1,2]");
+  });
+
+  it("nests deeply with stable order", () => {
+    expect(canonicalJson({ z: 1, a: [{ y: 1, x: 2 }] }))
+      .toBe('{"a":[{"x":2,"y":1}],"z":1}');
+  });
+
+  it("handles primitives + null", () => {
+    expect(canonicalJson("hi")).toBe('"hi"');
+    expect(canonicalJson(42)).toBe("42");
+    expect(canonicalJson(null)).toBe("null");
+  });
+});
+
+// useVerifyStream WS lifecycle — driven by an in-test WS shim that
+// captures the constructed URL + lets the test push synthetic frames.
+
+class MockSocket {
+  static instances: MockSocket[] = [];
+  static reset(): void { MockSocket.instances = []; }
+  url: string;
+  onmessage: ((e: MessageEvent) => void) | null = null;
+  onerror: ((e: Event) => void) | null = null;
+  closed = false;
+  constructor(url: string) {
+    this.url = url;
+    MockSocket.instances.push(this);
+  }
+  close(): void { this.closed = true; }
+  push(frame: unknown): void {
+    this.onmessage?.(new MessageEvent("message",
+      { data: JSON.stringify(frame) }));
+  }
+}
+
+describe("V7-H37 useVerifyStream", () => {
+  beforeEach(() => {
+    MockSocket.reset();
+    (globalThis as unknown as { WebSocket: typeof WebSocket })
+      .WebSocket = MockSocket as unknown as typeof WebSocket;
+  });
+  afterEach(() => {
+    (globalThis as unknown as { WebSocket: typeof WebSocket | null })
+      .WebSocket = null as unknown as typeof WebSocket;
+  });
+
+  it("opens /ws/verify/{hash} URL with backend base", async () => {
+    renderHook(() => useVerifyStream(
+      "http://127.0.0.1:8765", "abc123", true));
+    await waitFor(() => {
+      expect(MockSocket.instances.length).toBe(1);
+    });
+    expect(MockSocket.instances[0]?.url)
+      .toBe("ws://127.0.0.1:8765/ws/verify/abc123");
+  });
+
+  it("doesn't open socket while inactive", () => {
+    renderHook(() => useVerifyStream("http://x", "abc", false));
+    expect(MockSocket.instances).toEqual([]);
+  });
+
+  it("collects events and flips finished on finish frame", async () => {
+    const { result } = renderHook(() => useVerifyStream(
+      "http://x", "abc", true));
+    await waitFor(() => {
+      expect(MockSocket.instances.length).toBe(1);
+    });
+    const sock = MockSocket.instances[0]!;
+    act(() => {
+      sock.push({ event: { phase: "start" }, spec_hash: "abc" });
+      sock.push({ event: { phase: "resolve_shapes" }, spec_hash: "abc" });
+      sock.push({ finish: "ok", spec_hash: "abc" });
+    });
+    await waitFor(() => {
+      expect(result.current.finished).toBe(true);
+    });
+    expect(result.current.events.map((e) => e.phase))
+      .toEqual(["start", "resolve_shapes"]);
+    expect(sock.closed).toBe(true);
+  });
+});
