feat(v8-r09): HF Hub quickstart — stream → tokenize → parquet

Closes cppmega-mlx-yz6n.9. New module scripts/data/hf_quickstart.py
streams a HF Hub dataset via datasets.load_dataset(streaming=True),
tokenizes each document with cppmega tokenizer, emits a parquet shard
with the canonical 4-column schema (token_ids/doc_ids/byte_offsets/
byte_lengths). Pure-Python in-memory shim hf_quickstart_from_iterable
keeps tests offline.

New RPC: data.hf_quickstart(dataset_id, split, tokenizer, n_tokens,
job_id, ...) → {parquet_path, n_tokens_written, n_docs_seen,
elapsed_ms}. Network can be disabled via VBGUI_DISABLE_NETWORK=1 env.

New WS: /ws/data/{job_id} streams start/progress/done frames from the
new data_event_bus (same publish/subscribe pattern as train/gen/verify).

UI: HFQuickStartModal — opens from DataInspector header "HF quickstart"
button, runs the RPC, surfaces the result parquet path back to the
inspector's path field.

Tests: 6 new pytest (parquet schema, event bus, RPC env-block,
dispatch envelope), 3 new vitest (modal mount, Run+result, error
banner). Regression: 950 pytest + 466 vitest green.

Author: apstenku123

cppmega_v4/jsonrpc/dispatcher.py

@@ -83,6 +83,10 @@
     AutoFitParams,
     auto_fit,
 )
+from cppmega_v4.jsonrpc.hf_quickstart_method import (
+    HfQuickstartParams,
+    hf_quickstart_method,
+)
 from cppmega_v4.jsonrpc.tokenizer_roundtrip_text_method import (
     TokenizerRoundtripTextParams,
     roundtrip_text,
@@ -145,6 +149,10 @@
         AutoFitParams,
         lambda p, c: auto_fit(p, cache=c),
     ),
+    "data.hf_quickstart": (
+        HfQuickstartParams,
+        lambda p, c: hf_quickstart_method(p, cache=c),
+    ),
     "probe.run": (
         ProbeRunParams,
         lambda p, c: probe_run(p, cache=c),

cppmega_v4/jsonrpc/hf_quickstart_method.py

@@ -0,0 +1,80 @@
+"""V8-R09: ``data.hf_quickstart`` RPC handler.
+
+Wraps :func:`scripts.data.hf_quickstart.hf_quickstart` for the GUI.
+The handler runs synchronously — progress arrives on the
+``/ws/data/{job_id}`` WebSocket from the underlying data_event_bus.
+"""
+
+from __future__ import annotations
+
+import os
+from pydantic import BaseModel, ConfigDict
+
+from cppmega_v4.jsonrpc.cache import LRUCache
+from cppmega_v4.jsonrpc.methods import _cache_lookup, _cache_store
+
+
+__all__ = [
+    "HfQuickstartParams",
+    "HfQuickstartResult",
+    "hf_quickstart_method",
+]
+
+
+class HfQuickstartParams(BaseModel):
+    """Input — HF dataset + tokenizer + token budget."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    dataset_id: str
+    split: str = "train"
+    tokenizer: str = "cppmega_v3"
+    n_tokens: int = 100_000
+    job_id: str | None = None
+    out_dir: str | None = None
+    text_field: str = "text"
+
+
+class HfQuickstartResult(BaseModel):
+    """Output — parquet path + token / doc counters + elapsed wall."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    parquet_path: str
+    n_tokens_written: int
+    n_docs_seen: int
+    elapsed_ms: float
+
+
+def hf_quickstart_method(
+    params: HfQuickstartParams, *, cache: LRUCache | None = None,
+) -> HfQuickstartResult:
+    """Run the HF quickstart and return the resulting parquet shard."""
+    # Cache only on full param-key equality; HF jobs aren't typically
+    # repeated with identical params so the hit rate is near-zero.
+    key, hit = _cache_lookup(cache, "data.hf_quickstart", params)
+    if hit is not None:
+        return hit
+
+    if os.environ.get("VBGUI_DISABLE_NETWORK") == "1":
+        raise RuntimeError(
+            "HF Hub network access disabled via VBGUI_DISABLE_NETWORK")
+
+    from scripts.data.hf_quickstart import hf_quickstart as _hf
+    r = _hf(
+        dataset_id=params.dataset_id,
+        split=params.split,
+        tokenizer=params.tokenizer,
+        n_tokens=params.n_tokens,
+        job_id=params.job_id,
+        out_dir=params.out_dir,
+        text_field=params.text_field,
+    )
+    out = HfQuickstartResult(
+        parquet_path=r.parquet_path,
+        n_tokens_written=r.n_tokens_written,
+        n_docs_seen=r.n_docs_seen,
+        elapsed_ms=r.elapsed_ms,
+    )
+    _cache_store(cache, key, out)
+    return out

cppmega_v4/jsonrpc/schema.py

@@ -810,6 +810,7 @@ class PipelineRunResult(BaseModel):
     "architectures.scale_down",
     "memory.matrix",
     "architectures.auto_fit",
+    "data.hf_quickstart",
 })
 
 

cppmega_v4/jsonrpc/server.py

@@ -171,6 +171,43 @@ def _try_get(timeout: float = 0.2):
         finally:
             train_event_bus.unsubscribe(run_id, q)
 
+    @app.websocket("/ws/data/{job_id}")
+    async def ws_data(socket: WebSocket, job_id: str) -> None:
+        """V8-R09: live data-job progress stream.
+
+        The UI opens this WS *before* invoking data.hf_quickstart or
+        data.github_corpus with the same ``job_id``. Producer publishes
+        {phase: "start"|"progress"|"done", ...} dicts onto
+        ``data_event_bus``; the None sentinel ends the stream with a
+        final {finish:"ok"} frame."""
+        from cppmega_v4.runtime import data_event_bus
+        import queue as _queue
+        await socket.accept()
+        q = data_event_bus.subscribe(job_id)
+
+        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",
+                                             "job_id": job_id})
+                    return
+                await socket.send_json({"job_id": job_id,
+                                         "event": ev})
+        except WebSocketDisconnect:
+            pass
+        finally:
+            data_event_bus.unsubscribe(job_id, q)
+
     @app.websocket("/ws/verify/{spec_hash}")
     async def ws_verify(socket: WebSocket, spec_hash: str) -> None:
         """V7-L45: live verify progress stream.

cppmega_v4/runtime/data_event_bus.py

@@ -0,0 +1,58 @@
+"""V8-R09: per-job pub/sub bus for HF / GitHub data quickstart jobs.
+
+Pattern identical to ``train_event_bus`` — see that module's docstring
+for the protocol. Subscribers drain until they see ``None`` (the
+completion sentinel emitted by the producer when the job finishes).
+"""
+
+from __future__ import annotations
+
+import queue
+import threading
+from typing import Any
+
+
+_LOCK = threading.Lock()
+_QUEUES: dict[str, list[queue.Queue]] = {}
+
+
+def publish(job_id: str | None, event: dict[str, Any] | None) -> None:
+    if not job_id:
+        return
+    with _LOCK:
+        subs = list(_QUEUES.get(job_id, []))
+    for q in subs:
+        try:
+            q.put_nowait(event)
+        except Exception:
+            pass
+
+
+def subscribe(job_id: str) -> queue.Queue:
+    q: queue.Queue = queue.Queue(maxsize=1024)
+    with _LOCK:
+        _QUEUES.setdefault(job_id, []).append(q)
+    return q
+
+
+def unsubscribe(job_id: str, q: queue.Queue) -> None:
+    with _LOCK:
+        subs = _QUEUES.get(job_id, [])
+        if q in subs:
+            subs.remove(q)
+        if not subs:
+            _QUEUES.pop(job_id, None)
+
+
+def reset() -> None:
+    with _LOCK:
+        _QUEUES.clear()
+
+
+def subscriber_count(job_id: str) -> int:
+    with _LOCK:
+        return len(_QUEUES.get(job_id, []))
+
+
+__all__ = ["publish", "subscribe", "unsubscribe", "reset",
+            "subscriber_count"]