feat(vbgui-fh): training data inspector — parquet preview + ribbons

apstenku123 · apstenku123 · commit 22a5a03c67c6 · 2026-05-21T03:17:38.000+02:00
Stage F-H of the Visual Builder GUI epic (cppmega-mlx-o0k) — the
last open child ticket. Lands the backend handler + frontend panel
researchers use to inspect what actually enters model.forward().

Backend (cppmega_v4.jsonrpc.data_methods):
- data.preview_parquet — opens a parquet shard via pyarrow, picks
  the canonical token column (input_ids / token_ids / tokens),
  surfaces every other column as a side-channel "ribbon" with per-
  row payload. Pagination via offset+limit; channel filter via
  optional channels=[...]. Computes bytes/token avg + p95 + max
  from the encoded id stream (heuristic; precise stats pair with
  the F-G tokenizer panel). LRU cache hits on identical paginations.
- Dispatcher route + METHOD_REGISTRY entry.

Frontend (vbgui/src/components/DataInspector.tsx):
- Path input + Load button, per-page metrics line, channel toggle
  chips (color-coded), one row card per packed sequence with the
  token strip plus a Ribbon per enabled channel (array channels
  paint per-token cells, scalar channels paint a single badge).
- Pagination footer (← / page x / y / →) drives the offset.
- Error envelope from RpcClient surfaces inline.

Tests:
- Python (11): first-page fetch, per-row channel payload, offset
  pagination, channel filter, bytes/token stats, cache hit, past-
  end empty, invalid limit, missing token column raises,
  dispatcher round-trip, METHOD_REGISTRY lock.
- TypeScript (6): Load populates rows + channels + metrics, channel
  toggle hides ribbon, array channel renders as per-token strip,
  pagination Next/Prev advances offset in the body, error envelope
  surfaces inline.

Full v4 regression: 2090 passed / 5 skipped / 15 xfailed / 0 failed.
Vbgui: 77 tests passed (71 + 6 new).

Closes cppmega-mlx-o0k.8 and the cppmega-mlx-o0k epic.
diff --git a/cppmega_v4/jsonrpc/data_methods.py b/cppmega_v4/jsonrpc/data_methods.py
@@ -0,0 +1,172 @@
+"""Backend data preview — F-H ``data.preview_parquet`` handler.
+
+Loads a parquet shard, samples N rows, surfaces per-row token stream
+plus every other column as a side-channel "ribbon". The GUI uses the
+result to paint coloured strips under the token row so the researcher
+can see exactly what enters ``model.forward()``.
+
+Pyodide-friendly: only uses pyarrow + stdlib. The frontend can swap
+to in-browser hyparquet for the same shape later (F-E follow-up).
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Any
+
+import pyarrow.parquet as pq
+from pydantic import BaseModel, ConfigDict, Field
+
+from cppmega_v4.jsonrpc.cache import LRUCache
+from cppmega_v4.probe import introspect_parquet
+
+
+_PRIMARY_TOKEN_COLS: tuple[str, ...] = ("input_ids", "token_ids", "tokens")
+
+
+# ---------------------------------------------------------------------------
+# Schemas
+# ---------------------------------------------------------------------------
+
+
+class PreviewParquetParams(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    path: str
+    offset: int = 0
+    limit: int = 32
+    channels: list[str] | None = None  # None → all detected side-channels
+
+
+class PreviewRow(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    row_index: int
+    tokens: list[int]
+    channels: dict[str, Any] = Field(default_factory=dict)
+
+
+class PreviewParquetResult(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    rows: list[PreviewRow]
+    token_column: str
+    available_channels: list[str]
+    bytes_per_token_avg: float
+    bytes_per_token_p95: float
+    bytes_per_token_max: int
+    total_rows: int
+    elapsed_ms: float
+
+
+# ---------------------------------------------------------------------------
+# Handler
+# ---------------------------------------------------------------------------
+
+
+def preview_parquet(
+    params: PreviewParquetParams,
+    *,
+    cache: LRUCache | None = None,
+) -> PreviewParquetResult:
+    """Return ``params.limit`` rows starting at ``params.offset``."""
+    if params.limit < 1:
+        raise ValueError(f"limit must be ≥ 1, got {params.limit}")
+    if params.offset < 0:
+        raise ValueError(f"offset must be ≥ 0, got {params.offset}")
+
+    cache_key = f"preview::{params.path}::{params.offset}::{params.limit}::" \
+                f"{','.join(sorted(params.channels or []))}"
+    if cache is not None:
+        hit = cache.get(cache_key)
+        if hit is not None:
+            return hit
+
+    t0 = time.perf_counter()
+    pf = pq.ParquetFile(params.path)
+    schema_names = [f.name for f in pf.schema_arrow]
+    token_col = _pick_token_column(schema_names)
+    if token_col is None:
+        raise ValueError(
+            f"parquet shard {params.path!r} has no token column "
+            f"(expected one of {_PRIMARY_TOKEN_COLS})"
+        )
+
+    # Side-channel pool: every column that isn't the token stream or
+    # bookkeeping. Caller may further restrict via ``params.channels``.
+    caps = introspect_parquet(params.path, sample_rows=min(params.limit, 64))
+    available = sorted(c for c in caps.side_channels if c != token_col)
+    selected = [c for c in available if not params.channels
+                or c in params.channels]
+    cols_to_read = [token_col, *selected]
+
+    table = pf.read(columns=cols_to_read)
+    total_rows = table.num_rows
+    if params.offset >= total_rows:
+        sliced = table.slice(0, 0)
+    else:
+        n = min(params.limit, total_rows - params.offset)
+        sliced = table.slice(params.offset, n)
+
+    rows: list[PreviewRow] = []
+    token_lists: list[list[int]] = []
+    for i in range(sliced.num_rows):
+        token_val = sliced.column(token_col)[i].as_py() or []
+        tokens = [int(t) for t in token_val]
+        token_lists.append(tokens)
+        channel_payload: dict[str, Any] = {}
+        for ch in selected:
+            channel_payload[ch] = sliced.column(ch)[i].as_py()
+        rows.append(PreviewRow(
+            row_index=params.offset + i,
+            tokens=tokens,
+            channels=channel_payload,
+        ))
+
+    bpt_avg, bpt_p95, bpt_max = _bytes_per_token_stats(token_lists)
+    elapsed = (time.perf_counter() - t0) * 1000.0
+    out = PreviewParquetResult(
+        rows=rows,
+        token_column=token_col,
+        available_channels=available,
+        bytes_per_token_avg=round(bpt_avg, 3),
+        bytes_per_token_p95=round(bpt_p95, 3),
+        bytes_per_token_max=int(bpt_max),
+        total_rows=total_rows,
+        elapsed_ms=elapsed,
+    )
+    if cache is not None:
+        cache.set(cache_key, out)
+    return out
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _pick_token_column(schema_names: list[str]) -> str | None:
+    for name in _PRIMARY_TOKEN_COLS:
+        if name in schema_names:
+            return name
+    return None
+
+
+def _bytes_per_token_stats(
+    token_lists: list[list[int]],
+) -> tuple[float, float, int]:
+    """Compute per-token byte stats from the encoded id stream."""
+    if not token_lists:
+        return 0.0, 0.0, 0
+    # Each token id encodes to its big-endian byte length — proxy for
+    # the on-disk bytes/token cost (real bytes depend on vocab + BPE
+    # merge stats; the GUI uses this as a heuristic, full stats need
+    # the corresponding tokenizer).
+    lengths: list[int] = []
+    for row in token_lists:
+        for tok in row:
+            lengths.append(max(1, (tok.bit_length() + 7) // 8))
+    if not lengths:
+        return 0.0, 0.0, 0
+    avg = sum(lengths) / len(lengths)
+    sorted_lengths = sorted(lengths)
+    p95_idx = max(0, int(len(sorted_lengths) * 0.95) - 1)
+    p95 = float(sorted_lengths[p95_idx])
+    return avg, p95, max(lengths)
diff --git a/cppmega_v4/jsonrpc/dispatcher.py b/cppmega_v4/jsonrpc/dispatcher.py
@@ -27,6 +27,10 @@
     encode_visualize,
     list_presets as tokenizer_list_presets,
 )
+from cppmega_v4.jsonrpc.data_methods import (
+    PreviewParquetParams,
+    preview_parquet,
+)
 from cppmega_v4.jsonrpc.schema import (
     BuildPresetSpecsParams,
     ErrorCode,
@@ -76,6 +80,10 @@
         EncodeVisualizeParams,
         lambda p, c: encode_visualize(p, cache=c),
     ),
+    "data.preview_parquet": (
+        PreviewParquetParams,
+        lambda p, c: preview_parquet(p, cache=c),
+    ),
 }
 
 
diff --git a/cppmega_v4/jsonrpc/schema.py b/cppmega_v4/jsonrpc/schema.py
@@ -534,4 +534,5 @@ class PipelineRunResult(BaseModel):
     "backend.status",
     "tokenizer.encode_visualize",
     "tokenizer.list_presets",
+    "data.preview_parquet",
 })
diff --git a/tests/v4/test_data_inspector.py b/tests/v4/test_data_inspector.py
@@ -0,0 +1,131 @@
+"""F-H data inspector tests — preview_parquet end-to-end."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+
+from cppmega_v4.jsonrpc import LRUCache, dispatch
+from cppmega_v4.jsonrpc.data_methods import (
+    PreviewParquetParams,
+    preview_parquet,
+)
+
+
+def _write_full_parquet(p: Path, n_rows: int = 32):
+    pq.write_table(pa.table({
+        "input_ids":   [list(range(8 + i % 4)) for i in range(n_rows)],
+        "doc_ids":     [i // 4 for i in range(n_rows)],
+        "loss_mask":   [[1] * (8 + i % 4) for i in range(n_rows)],
+        "call_edges":  [[(0, 1)] for _ in range(n_rows)],
+        "type_edges":  [[(0, 2)] for _ in range(n_rows)],
+    }), p)
+
+
+# ---------------------------------------------------------------------------
+# preview_parquet
+# ---------------------------------------------------------------------------
+
+
+def test_preview_returns_first_page(tmp_path: Path):
+    p = tmp_path / "shard.parquet"
+    _write_full_parquet(p, n_rows=16)
+    r = preview_parquet(PreviewParquetParams(path=str(p), offset=0, limit=4))
+    assert r.total_rows == 16
+    assert len(r.rows) == 4
+    assert r.token_column == "input_ids"
+    assert "doc_ids" in r.available_channels
+    assert "call_edges" in r.available_channels
+
+
+def test_preview_carries_channel_payload_per_row(tmp_path: Path):
+    p = tmp_path / "shard.parquet"
+    _write_full_parquet(p, n_rows=8)
+    r = preview_parquet(PreviewParquetParams(path=str(p), offset=0, limit=2))
+    first = r.rows[0]
+    assert first.row_index == 0
+    assert len(first.tokens) > 0
+    assert "doc_ids" in first.channels
+    assert "loss_mask" in first.channels
+
+
+def test_preview_pagination_offset_works(tmp_path: Path):
+    p = tmp_path / "shard.parquet"
+    _write_full_parquet(p, n_rows=10)
+    r = preview_parquet(PreviewParquetParams(path=str(p), offset=5, limit=4))
+    assert [row.row_index for row in r.rows] == [5, 6, 7, 8]
+
+
+def test_preview_channel_filter_restricts_payload(tmp_path: Path):
+    p = tmp_path / "shard.parquet"
+    _write_full_parquet(p, n_rows=4)
+    r = preview_parquet(PreviewParquetParams(
+        path=str(p), offset=0, limit=2, channels=["doc_ids"],
+    ))
+    for row in r.rows:
+        assert set(row.channels.keys()) == {"doc_ids"}
+
+
+def test_preview_bytes_per_token_stats_present(tmp_path: Path):
+    p = tmp_path / "shard.parquet"
+    _write_full_parquet(p, n_rows=4)
+    r = preview_parquet(PreviewParquetParams(path=str(p), offset=0, limit=4))
+    assert r.bytes_per_token_avg > 0
+    assert r.bytes_per_token_max >= r.bytes_per_token_avg
+
+
+def test_preview_caches_repeat_calls(tmp_path: Path):
+    p = tmp_path / "shard.parquet"
+    _write_full_parquet(p, n_rows=4)
+    cache = LRUCache()
+    params = PreviewParquetParams(path=str(p), offset=0, limit=2)
+    preview_parquet(params, cache=cache)
+    preview_parquet(params, cache=cache)
+    assert cache.stats()["hits"] >= 1
+
+
+def test_preview_offset_past_end_returns_empty(tmp_path: Path):
+    p = tmp_path / "shard.parquet"
+    _write_full_parquet(p, n_rows=4)
+    r = preview_parquet(PreviewParquetParams(path=str(p), offset=999, limit=8))
+    assert r.rows == []
+    assert r.total_rows == 4
+
+
+def test_preview_rejects_invalid_limit(tmp_path: Path):
+    p = tmp_path / "shard.parquet"
+    _write_full_parquet(p, n_rows=4)
+    with pytest.raises(ValueError, match="limit"):
+        preview_parquet(PreviewParquetParams(path=str(p), limit=0))
+
+
+def test_preview_raises_when_token_column_missing(tmp_path: Path):
+    p = tmp_path / "no_tokens.parquet"
+    pq.write_table(pa.table({"some_col": [1, 2, 3]}), p)
+    with pytest.raises(ValueError, match="no token column"):
+        preview_parquet(PreviewParquetParams(path=str(p), limit=2))
+
+
+# ---------------------------------------------------------------------------
+# Dispatcher integration
+# ---------------------------------------------------------------------------
+
+
+def test_dispatch_data_preview_parquet_round_trip(tmp_path: Path):
+    p = tmp_path / "shard.parquet"
+    _write_full_parquet(p, n_rows=4)
+    resp = dispatch({
+        "jsonrpc": "2.0", "id": "d1", "method": "data.preview_parquet",
+        "params": {"path": str(p), "offset": 0, "limit": 2},
+    })
+    assert resp.error is None
+    assert len(resp.result["rows"]) == 2
+    assert resp.result["token_column"] == "input_ids"
+
+
+def test_method_registry_includes_data_preview():
+    from cppmega_v4.jsonrpc import METHOD_REGISTRY
+    assert "data.preview_parquet" in METHOD_REGISTRY
diff --git a/vbgui/src/components/DataInspector.tsx b/vbgui/src/components/DataInspector.tsx
diff --git a/vbgui/tests/DataInspector.test.tsx b/vbgui/tests/DataInspector.test.tsx
