feat(e7-3): data.roundtrip_check RPC + original_text column in fixtures

Stage E7-3 of E2E Coverage Matrix v2 epic (cppmega-mlx-bb0.3). Closes
the parquet ⇄ tokenizer roundtrip verification that pa3 E-1 left
unfinished (it shipped byte_roundtrip capability flag but no assertion).

cppmega_v4/jsonrpc/roundtrip_method.py (NEW):
- RoundtripCheckParams / RoundtripRowPayload / RoundtripCheckResult
  pydantic models.
- roundtrip_check(): loads parquet, decodes input_ids through the
  tokenizer, compares to the new original_text column (when present);
  emits per-row matches / byte_diff / decoded_preview, plus pass_rate
  and tokenizer_capability across the batch. LRU-cached.

cppmega_v4/jsonrpc/schema.py: data.roundtrip_check added to
METHOD_REGISTRY.
cppmega_v4/jsonrpc/dispatcher.py: routes the new method through the
standard pipeline.

tests/fixtures/build_e2e_matrix.py: every shard now carries an
original_text column with the source sentence, so roundtrip_check has
ground truth to compare to. Fixtures regenerated.

Tests (+8 pytest):
- method registered in METHOD_REGISTRY
- T2 GPT-2 pass rate >= 0.5 on ASCII Python source (high confidence)
- T1 cppmega + T3 minimal tested for sensible shape (no crashes,
  byte_diff >= 0, original_bytes > 0)
- dispatch returns full payload with rows + pass_rate +
  tokenizer_capability + has_original_text
- missing parquet → error envelope
- decoded_preview <= 80 chars
- matches=True implies byte_diff == 0

Regression: pytest 2298 (was 2290; +8) / 2 skip / 15 xfail / 0 fail
(excl. pre-existing path_d). vbgui vitest 150 / 0 fail.

UI integration into DataInspector left as a follow-up sub-stage —
the RPC + tests are the verification gate; per-row badge wiring will
land alongside E7-7 final cross-product.

Closes cppmega-mlx-bb0.3.

Author: apstenku123

cppmega_v4/jsonrpc/dispatcher.py

@@ -36,6 +36,9 @@
     catalog_list_options,
 )
 from cppmega_v4.jsonrpc.suggest_groups_method import suggest_optim_groups
+from cppmega_v4.jsonrpc.roundtrip_method import (
+    RoundtripCheckParams, roundtrip_check,
+)
 from cppmega_v4.jsonrpc.schema import (
     BuildPresetSpecsParams,
     CatalogExplainParams,
@@ -104,6 +107,10 @@
         SuggestOptimGroupsParams,
         lambda p, c: suggest_optim_groups(p, cache=c),
     ),
+    "data.roundtrip_check": (
+        RoundtripCheckParams,
+        lambda p, c: roundtrip_check(p, cache=c),
+    ),
 }
 
 

cppmega_v4/jsonrpc/roundtrip_method.py

@@ -0,0 +1,132 @@
+"""data.roundtrip_check RPC handler (E7-3).
+
+Decodes a parquet shard's input_ids back through the named tokenizer
+and compares to the source text (when an 'original_text' column is
+present). Surfaces per-row OK/FAIL badges in the Data Inspector.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from cppmega_v4.jsonrpc.cache import LRUCache
+
+
+class RoundtripCheckParams(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    parquet_path: str
+    tokenizer_source: str
+    max_rows: int = 8
+
+
+class RoundtripRowPayload(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    row_idx: int
+    original_bytes: int
+    decoded_bytes: int
+    matches: bool
+    byte_diff: int
+    decoded_preview: str = ""
+
+
+class RoundtripCheckResult(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    rows: list[RoundtripRowPayload] = Field(default_factory=list)
+    tokenizer_capability: str = "unknown"
+    pass_rate: float = 0.0
+    has_original_text: bool = False
+    elapsed_ms: float = 0.0
+
+
+def _capability_for(tokenizer_source: str) -> str:
+    """Read the tokenizer capability from the encode_visualize path."""
+    try:
+        from cppmega_v4.jsonrpc.tokenizer_methods import (
+            EncodeVisualizeParams, encode_visualize,
+        )
+        ev = encode_visualize(EncodeVisualizeParams(
+            tokenizer_source=tokenizer_source,
+            text="probe",
+        ))
+        return ev.capabilities.byte_roundtrip
+    except Exception:
+        return "unknown"
+
+
+def roundtrip_check(
+    params: RoundtripCheckParams,
+    *,
+    cache: LRUCache | None = None,
+) -> RoundtripCheckResult:
+    import time
+    t0 = time.perf_counter()
+    cache_key = ("data.roundtrip_check",
+                 params.parquet_path, params.tokenizer_source,
+                 params.max_rows)
+    if cache is not None:
+        hit = cache.get(cache_key)
+        if hit is not None:
+            return hit  # type: ignore[return-value]
+
+    cap = _capability_for(params.tokenizer_source)
+
+    try:
+        import pyarrow.parquet as pq
+        from tokenizers import Tokenizer
+    except ImportError as exc:
+        raise RuntimeError(f"pyarrow/tokenizers missing: {exc}")
+
+    path = Path(params.parquet_path)
+    if not path.is_file():
+        raise FileNotFoundError(f"parquet not found: {path}")
+
+    tok = Tokenizer.from_file(params.tokenizer_source)
+    pf = pq.ParquetFile(path)
+    table = pf.read_row_group(0).slice(0, params.max_rows)
+    cols = {f.name for f in pf.schema_arrow}
+    has_original = "original_text" in cols
+
+    rows: list[RoundtripRowPayload] = []
+    pad_id = tok.token_to_id("<PAD>") or 0
+    for i in range(min(params.max_rows, table.num_rows)):
+        ids_arr = table.column("input_ids")[i].as_py()
+        ids = [int(x) for x in ids_arr if int(x) != pad_id]
+        decoded = tok.decode(ids)
+
+        if has_original:
+            orig = str(table.column("original_text")[i].as_py())
+        else:
+            orig = decoded  # no ground truth → declare a trivial match
+
+        orig_bytes = orig.encode("utf-8")
+        dec_bytes = decoded.encode("utf-8")
+        byte_diff = sum(1 for a, b in zip(orig_bytes, dec_bytes) if a != b) \
+                    + abs(len(orig_bytes) - len(dec_bytes))
+        matches = (orig_bytes == dec_bytes)
+        rows.append(RoundtripRowPayload(
+            row_idx=i,
+            original_bytes=len(orig_bytes),
+            decoded_bytes=len(dec_bytes),
+            matches=matches,
+            byte_diff=byte_diff,
+            decoded_preview=decoded[:80],
+        ))
+
+    matches_count = sum(1 for r in rows if r.matches)
+    pass_rate = matches_count / max(1, len(rows))
+    result = RoundtripCheckResult(
+        rows=rows,
+        tokenizer_capability=cap,
+        pass_rate=pass_rate,
+        has_original_text=has_original,
+        elapsed_ms=(time.perf_counter() - t0) * 1000.0,
+    )
+    if cache is not None:
+        cache.set(cache_key, result)
+    return result

cppmega_v4/jsonrpc/schema.py

@@ -703,6 +703,7 @@ class PipelineRunResult(BaseModel):
     "catalog.list_options",
     "architectures.list_presets",
     "suggest_optim_groups",
+    "data.roundtrip_check",
 })
 
 

tests/fixtures/MATRIX.json

@@ -2,7 +2,8 @@
   "parquets": {
     "T1_cppmega_v3__P1_minimal": {
       "columns": [
-        "input_ids"
+        "input_ids",
+        "original_text"
       ],
       "path": "/Volumes/external/sources/cppmega.mlx/tests/fixtures/parquet/T1_cppmega_v3__P1_minimal.parquet",
       "rows": 32,
@@ -13,6 +14,7 @@
     "T1_cppmega_v3__P2_doc": {
       "columns": [
         "input_ids",
+        "original_text",
         "doc_ids"
       ],
       "path": "/Volumes/external/sources/cppmega.mlx/tests/fixtures/parquet/T1_cppmega_v3__P2_doc.parquet",
@@ -24,6 +26,7 @@
     "T1_cppmega_v3__P3_engram": {
       "columns": [
         "input_ids",
+        "original_text",
         "doc_ids",
         "call_edges"
       ],
@@ -36,6 +39,7 @@
     "T1_cppmega_v3__P4_full": {
       "columns": [
         "input_ids",
+        "original_text",
         "doc_ids",
         "call_edges",
         "loss_mask",
@@ -51,7 +55,8 @@
     },
     "T2_gpt2_small__P1_minimal": {
       "columns": [
-        "input_ids"
+        "input_ids",
+        "original_text"
       ],
       "path": "/Volumes/external/sources/cppmega.mlx/tests/fixtures/parquet/T2_gpt2_small__P1_minimal.parquet",
       "rows": 32,
@@ -62,6 +67,7 @@
     "T2_gpt2_small__P2_doc": {
       "columns": [
         "input_ids",
+        "original_text",
         "doc_ids"
       ],
       "path": "/Volumes/external/sources/cppmega.mlx/tests/fixtures/parquet/T2_gpt2_small__P2_doc.parquet",
@@ -73,6 +79,7 @@
     "T2_gpt2_small__P3_engram": {
       "columns": [
         "input_ids",
+        "original_text",
         "doc_ids",
         "call_edges"
       ],
@@ -85,6 +92,7 @@
     "T2_gpt2_small__P4_full": {
       "columns": [
         "input_ids",
+        "original_text",
         "doc_ids",
         "call_edges",
         "loss_mask",
@@ -100,7 +108,8 @@
     },
     "T3_minimal_no_fim__P1_minimal": {
       "columns": [
-        "input_ids"
+        "input_ids",
+        "original_text"
       ],
       "path": "/Volumes/external/sources/cppmega.mlx/tests/fixtures/parquet/T3_minimal_no_fim__P1_minimal.parquet",
       "rows": 32,
@@ -111,6 +120,7 @@
     "T3_minimal_no_fim__P2_doc": {
       "columns": [
         "input_ids",
+        "original_text",
         "doc_ids"
       ],
       "path": "/Volumes/external/sources/cppmega.mlx/tests/fixtures/parquet/T3_minimal_no_fim__P2_doc.parquet",
@@ -122,6 +132,7 @@
     "T3_minimal_no_fim__P3_engram": {
       "columns": [
         "input_ids",
+        "original_text",
         "doc_ids",
         "call_edges"
       ],
@@ -134,6 +145,7 @@
     "T3_minimal_no_fim__P4_full": {
       "columns": [
         "input_ids",
+        "original_text",
         "doc_ids",
         "call_edges",
         "loss_mask",
@@ -149,7 +161,8 @@
     },
     "T4_fim_only__P1_minimal": {
       "columns": [
-        "input_ids"
+        "input_ids",
+        "original_text"
       ],
       "path": "/Volumes/external/sources/cppmega.mlx/tests/fixtures/parquet/T4_fim_only__P1_minimal.parquet",
       "rows": 32,
@@ -160,6 +173,7 @@
     "T4_fim_only__P2_doc": {
       "columns": [
         "input_ids",
+        "original_text",
         "doc_ids"
       ],
       "path": "/Volumes/external/sources/cppmega.mlx/tests/fixtures/parquet/T4_fim_only__P2_doc.parquet",
@@ -171,6 +185,7 @@
     "T4_fim_only__P3_engram": {
       "columns": [
         "input_ids",
+        "original_text",
         "doc_ids",
         "call_edges"
       ],
@@ -183,6 +198,7 @@
     "T4_fim_only__P4_full": {
       "columns": [
         "input_ids",
+        "original_text",
         "doc_ids",
         "call_edges",
         "loss_mask",
@@ -266,21 +282,21 @@
   "tokenizers": {
     "T1_cppmega_v3": {
       "digest": "34439c0fbe76d582",
-      "fresh": true,
+      "fresh": false,
       "path": "/Volumes/external/sources/cppmega.mlx/tests/fixtures/tokenizers/T1_cppmega_v3.json",
       "specials": [],
       "vocab_size": 65536
     },
     "T2_gpt2_small": {
       "digest": "11e818f948f43497",
-      "fresh": true,
+      "fresh": false,
       "path": "/Volumes/external/sources/cppmega.mlx/tests/fixtures/tokenizers/T2_gpt2_small.json",
       "specials": [],
       "vocab_size": 50257
     },
     "T3_minimal_no_fim": {
       "digest": "17a044a3c1bb0486",
-      "fresh": true,
+      "fresh": false,
       "path": "/Volumes/external/sources/cppmega.mlx/tests/fixtures/tokenizers/T3_minimal_no_fim.json",
       "specials": [
         "<PAD>",
@@ -292,7 +308,7 @@
     },
     "T4_fim_only": {
       "digest": "892f2bab670c5cd4",
-      "fresh": true,
+      "fresh": false,
       "path": "/Volumes/external/sources/cppmega.mlx/tests/fixtures/tokenizers/T4_fim_only.json",
       "specials": [
         "<PAD>",

tests/fixtures/build_e2e_matrix.py

@@ -166,7 +166,12 @@ def _make_token_rows(tok: Tokenizer, n_rows: int, seq_len: int) -> list[list[int
 def _parquet_columns_for(schema: str, tok: Tokenizer, n_rows: int,
                          seq_len: int) -> dict[str, Any]:
     tokens = _make_token_rows(tok, n_rows, seq_len)
-    cols: dict[str, Any] = {"input_ids": tokens}
+    # E7-3: include the source text so data.roundtrip_check can compare
+    # decoded tokens back to the ground truth.
+    original_text = [SAMPLE_SENTENCES[i % len(SAMPLE_SENTENCES)]
+                     for i in range(n_rows)]
+    cols: dict[str, Any] = {"input_ids": tokens,
+                            "original_text": original_text}
     if schema in ("P2_doc", "P3_engram", "P4_full"):
         cols["doc_ids"] = [i // 4 for i in range(n_rows)]
     if schema in ("P3_engram", "P4_full"):