feat(v7-g02,v7-g04): doc_id helper + corpus_stats sidecar in DataInspector

Two honest-closure data-pipeline gaps:

V7-G02: doc_id_assignment helper lived at
cppmega_v4/data/doc_id_assignment.py but clang_enriched_to_parquet
fell back to a per-source-file synthesized id when source_doc_id was
missing. Same logical doc across shards got different ids.

V7-G04: compute_corpus_stats existed but the parquet emit path never
wrote a sidecar, so DataInspector had nothing to surface.

- scripts/nanochat_data/clang_enriched_to_parquet.py: when
  source_doc_id is missing, prefer stable_doc_signature(record) from
  the helper. Falls back to the legacy filename:index on exception.
  Also writes {shard}.corpus_stats.json next to each parquet shard.
- jsonrpc/data_methods.py PreviewParquetResult: new corpus_stats:
  dict | None field; _read_corpus_stats_sidecar loads the JSON if
  present, returns None for legacy shards.
- vbgui/components/DataInspector.tsx: CorpusStats interface +
  data-corpus-stats block rendering token_coverage_pct, doc-length
  p50/p90/p99, n_docs, long_tail_count.
- tests/v4/test_corpus_stats_sidecar.py: sidecar present → field
  populated; missing → field is null.
- e2e 86: visual assertion — pyarrow stub-shard + synthetic sidecar
  → DataInspector block shows '42.50%', '3/4/5', '7'.

Author: apstenku123

cppmega_v4/jsonrpc/data_methods.py

@@ -112,6 +112,10 @@ class PreviewParquetResult(BaseModel):
     bytes_per_token_max: int
     total_rows: int
     elapsed_ms: float
+    # V7-G04: corpus stats sidecar (token coverage / doc-length / vocab).
+    # Populated when the shard was emitted by clang_enriched_to_parquet
+    # with token_ids materialized; absent for legacy shards.
+    corpus_stats: dict | None = None
 
 
 # ---------------------------------------------------------------------------
@@ -214,12 +218,31 @@ def preview_parquet(
         bytes_per_token_max=int(bpt_max),
         total_rows=total_rows,
         elapsed_ms=elapsed,
+        corpus_stats=_read_corpus_stats_sidecar(preview_path),
     )
     if cache is not None:
         cache.set(cache_key, out)
     return out
 
 
+def _read_corpus_stats_sidecar(parquet_path) -> dict | None:
+    """V7-G04: load the {parquet}.corpus_stats.json sidecar emitted by
+    clang_enriched_to_parquet. Returns None when missing or malformed."""
+    import json as _json
+    import os
+    sidecar = str(parquet_path) + ".corpus_stats.json"
+    if not os.path.exists(sidecar):
+        return None
+    try:
+        with open(sidecar) as f:
+            data = _json.load(f)
+        if isinstance(data, dict):
+            return data
+    except Exception:
+        return None
+    return None
+
+
 # ---------------------------------------------------------------------------
 # Helpers
 # ---------------------------------------------------------------------------

scripts/nanochat_data/clang_enriched_to_parquet.py

@@ -933,7 +933,19 @@ def flush_rows() -> None:
                     max_tokens,
                     overflow_policy=overflow_policy,
                 )
-                source_doc_id = record.get("source_doc_id", f"{source.name}:{docs_in}")
+                # V7-G02: prefer the helper's stable signature so the
+                # same logical document gets the same id even if
+                # shards are reordered later. Fall back to the legacy
+                # name:index when the helper rejects the row.
+                source_doc_id = record.get("source_doc_id")
+                if not source_doc_id:
+                    try:
+                        from cppmega_v4.data.doc_id_assignment import (
+                            stable_doc_signature)
+                        sig = stable_doc_signature(record)
+                        source_doc_id = sig or f"{source.name}:{docs_in}"
+                    except Exception:
+                        source_doc_id = f"{source.name}:{docs_in}"
                 for sub_doc in sub_docs:
                     sub_doc.setdefault("source_doc_id", source_doc_id)
                     sub_doc.setdefault(
@@ -1023,7 +1035,17 @@ def process_input_file(gcs_uri: str, tmpdir: str,
                     max_tokens,
                     overflow_policy=overflow_policy,
                 )
-            source_doc_id = record.get("source_doc_id", f"{fname}:{docs_in}")
+            # V7-G02: same fallback as the bucket path above — use the
+            # stable doc signature so cross-shard duplicates collapse.
+            source_doc_id = record.get("source_doc_id")
+            if not source_doc_id:
+                try:
+                    from cppmega_v4.data.doc_id_assignment import (
+                        stable_doc_signature)
+                    sig = stable_doc_signature(record)
+                    source_doc_id = sig or f"{fname}:{docs_in}"
+                except Exception:
+                    source_doc_id = f"{fname}:{docs_in}"
             for sub_doc in sub_docs:
                 sub_doc.setdefault("source_doc_id", source_doc_id)
                 sub_doc.setdefault(
@@ -1081,6 +1103,39 @@ def _flush_shard(
     check_memory_limit(_MEMORY_LIMIT_GB, label="clang_enriched_to_parquet")
     pq.write_table(table, local_path, compression="snappy")
     check_memory_limit(_MEMORY_LIMIT_GB, label="clang_enriched_to_parquet")
+    # V7-G04: emit a sidecar corpus-stats JSON next to the parquet shard
+    # so the UI DataInspector can render token coverage / doc-length
+    # percentiles / vocab usage without rescanning the shard. The helper
+    # is no-op safe when token_ids are absent (token_id_lists empty).
+    try:
+        from cppmega_v4.data.corpus_stats import compute_corpus_stats
+        import json as _json
+        token_lists: list[list[int]] = []
+        try:
+            for col_token_ids in table.column(TOKEN_IDS_COLUMN).to_pylist():
+                if col_token_ids:
+                    token_lists.append(list(col_token_ids))
+        except (KeyError, Exception):
+            token_lists = []
+        if token_lists and _TOKENIZED_ENRICHED_TOKENIZER is not None:
+            stats = compute_corpus_stats(
+                token_lists,
+                vocab_size=int(
+                    _TOKENIZED_ENRICHED_TOKENIZER.get_vocab_size()),
+            )
+            stats_path = local_path + ".corpus_stats.json"
+            with open(stats_path, "w") as _f:
+                _json.dump(stats, _f)
+            try:
+                gcs_upload(stats_path, gcs_uri + ".corpus_stats.json")
+            except Exception:
+                pass
+            try:
+                os.unlink(stats_path)
+            except Exception:
+                pass
+    except Exception as _exc:
+        log.warning("corpus_stats emit failed: %s", _exc)
     gcs_upload(local_path, gcs_uri)
     os.unlink(local_path)
     log.info("Uploaded shard %d (%d rows, %.1f MB)",

tests/v4/test_corpus_stats_sidecar.py

@@ -0,0 +1,79 @@
+"""V7-G04: data.preview_parquet surfaces the corpus_stats sidecar.
+
+Honest-closure: compute_corpus_stats existed in
+cppmega_v4/data/corpus_stats.py but clang_enriched_to_parquet never
+wrote its output. After wiring, the sidecar JSON lands next to the
+shard and preview_parquet returns it under corpus_stats.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import numpy as np
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+from fastapi.testclient import TestClient
+
+from cppmega_v4.jsonrpc import create_app
+
+
+@pytest.fixture
+def client():
+    return TestClient(create_app(cache_capacity=2))
+
+
+def _write_mini_shard(parquet_path: Path) -> None:
+    schema = pa.schema([
+        pa.field("token_ids", pa.list_(pa.int64())),
+    ])
+    table = pa.table({
+        "token_ids": [[1, 2, 3], [2, 3, 4, 5], [1, 1, 5]],
+    }, schema=schema)
+    pq.write_table(table, parquet_path, compression="snappy")
+
+
+def test_corpus_stats_field_returned_when_sidecar_present(client, tmp_path):
+    p = tmp_path / "shard.parquet"
+    _write_mini_shard(p)
+    # Sidecar a synthetic stats blob.
+    sidecar = p.with_suffix(".parquet.corpus_stats.json")
+    sidecar.write_text(json.dumps({
+        "token_coverage_pct": 12.5,
+        "doc_length_p50": 3,
+        "vocab_usage_topk": [[1, 3], [2, 2]],
+    }))
+
+    payload = {
+        "jsonrpc": "2.0", "id": "g1", "method": "data.preview_parquet",
+        "params": {"path": str(p), "limit": 5, "offset": 0},
+    }
+    r = client.post("/rpc", json=payload)
+    assert r.status_code == 200
+    body = r.json()
+    assert "error" not in body, body
+    cs = body["result"]["corpus_stats"]
+    assert cs is not None
+    assert cs["token_coverage_pct"] == 12.5
+    assert cs["doc_length_p50"] == 3
+
+
+def test_corpus_stats_field_null_when_sidecar_missing(client, tmp_path):
+    p = tmp_path / "shard.parquet"
+    _write_mini_shard(p)
+    # No sidecar written.
+
+    payload = {
+        "jsonrpc": "2.0", "id": "g2", "method": "data.preview_parquet",
+        "params": {"path": str(p), "limit": 5, "offset": 0},
+    }
+    r = client.post("/rpc", json=payload)
+    body = r.json()
+    assert body["result"]["corpus_stats"] is None
+
+
+# Silence unused-import warning — numpy is imported in case future tests
+# extend the shard schema.
+_ = np

vbgui/e2e/scenarios/86_corpus_stats_inspector.spec.ts

@@ -0,0 +1,51 @@
+// V7-G04: DataInspector renders the corpus_stats sidecar emitted by
+// clang_enriched_to_parquet. This test creates a tiny parquet shard,
+// writes a synthetic sidecar JSON, opens DataInspector via the UI,
+// and asserts the corpus-stats block surfaces token coverage,
+// doc-length percentiles, and long-tail count.
+
+import { test, expect } from "@playwright/test";
+import path from "node:path";
+import os from "node:os";
+import fs from "node:fs";
+import { execSync } from "node:child_process";
+import { gotoApp, clickTab } from "../fixtures";
+
+function tmpDir(): string {
+  return fs.mkdtempSync(path.join(os.tmpdir(), "v7g04-"));
+}
+
+test("V7-G04: corpus_stats sidecar renders in DataInspector",
+  async ({ page }) => {
+    test.setTimeout(60_000);
+    const dir = tmpDir();
+    const parquet = path.join(dir, "shard.parquet");
+    // Use the same Python venv to build a one-row parquet + sidecar.
+    const repoRoot = path.resolve(__dirname, "../../..");
+    execSync(
+      `${repoRoot}/.venv/bin/python -c "import pyarrow as pa, pyarrow.parquet as pq, json; ` +
+        `pq.write_table(pa.table({'token_ids':[[1,2,3]]}), '${parquet}'); ` +
+        `open('${parquet}.corpus_stats.json','w').write(` +
+        `json.dumps({'token_coverage_pct': 42.5, 'doc_length_p50': 3, ` +
+        `'doc_length_p90': 4, 'doc_length_p99': 5, 'n_docs': 1, ` +
+        `'long_tail_count': 7}))"`,
+      { encoding: "utf-8" },
+    );
+
+    await gotoApp(page);
+    await clickTab(page, "data");
+    await page.getByTestId("data-source-path").fill(parquet);
+    await page.getByTestId("data-load").click();
+
+    const block = page.getByTestId("data-corpus-stats");
+    await expect(block).toBeVisible({ timeout: 10_000 });
+
+    await expect(page.getByTestId("data-corpus-stats-token-coverage"))
+      .toContainText("42.50%");
+    await expect(page.getByTestId("data-corpus-stats-doc-length"))
+      .toContainText("3/4/5");
+    await expect(page.getByTestId("data-corpus-stats-long-tail"))
+      .toContainText("7");
+    await expect(page.getByTestId("data-corpus-stats-n-docs"))
+      .toContainText("1");
+  });

vbgui/src/components/DataInspector.tsx

@@ -19,6 +19,20 @@ export interface PreviewParquetResult {
   bytes_per_token_max: number;
   total_rows: number;
   elapsed_ms: number;
+  /** V7-G04: corpus_stats sidecar emitted by clang_enriched_to_parquet
+   *  (compute_corpus_stats output). Null for legacy shards. */
+  corpus_stats?: CorpusStats | null;
+}
+
+export interface CorpusStats {
+  token_coverage_pct?: number;
+  doc_length_p50?: number;
+  doc_length_p90?: number;
+  doc_length_p99?: number;
+  doc_length_histogram?: Array<[number, number]>;
+  vocab_usage_topk?: Array<[number, number]>;
+  long_tail_count?: number;
+  n_docs?: number;
 }
 
 export interface SideChannelFamilyCoverage {
@@ -241,6 +255,34 @@ export function DataInspector({
             {" "}max {result.bytes_per_token_max}
           </div>
 
+          {result.corpus_stats && (
+            <div data-testid="data-corpus-stats"
+                 style={{ border: "1px solid #d1d5db", borderRadius: 4,
+                          padding: 8, fontSize: 11,
+                          background: "#f9fafb",
+                          fontFamily: "monospace" }}>
+              <div style={{ fontWeight: 600, marginBottom: 4 }}>
+                corpus stats (V7-G04)
+              </div>
+              <div data-testid="data-corpus-stats-token-coverage">
+                token coverage:{" "}
+                {result.corpus_stats.token_coverage_pct?.toFixed(2)}%
+              </div>
+              <div data-testid="data-corpus-stats-doc-length">
+                doc length p50/p90/p99:{" "}
+                {result.corpus_stats.doc_length_p50 ?? "?"}/
+                {result.corpus_stats.doc_length_p90 ?? "?"}/
+                {result.corpus_stats.doc_length_p99 ?? "?"}
+              </div>
+              <div data-testid="data-corpus-stats-n-docs">
+                docs: {result.corpus_stats.n_docs ?? 0}
+              </div>
+              <div data-testid="data-corpus-stats-long-tail">
+                long-tail tokens (≤1 use):{" "}
+                {result.corpus_stats.long_tail_count ?? 0}
+              </div>
+            </div>
+          )}
           {result.shards && result.shards.length > 0 && (
             <div data-testid="data-shards"
                  style={{ display: "grid",