feat(e-audit-01): DataInspector parquet file picker + /upload/parquet

Closes bd cppmega-mlx-zcbs (P0 BLOCKING). Pre-fix DataInspector path
field was text-only (DataInspector.tsx:194-198) — users could only
reference fixture paths inside the repo, not bring their own parquet.

Backend:
  - cppmega_v4/jsonrpc/uploads.py: save_upload() writes body to
    /tmp/vbgui_uploads/<uuid>.parquet, cleanup_stale() drops files
    older than 24 h, _enforce_cap() caps at 100 files total. Cleanup
    + cap run on every save so /tmp can't grow unbounded.
  - POST /upload/parquet endpoint in server.py: 400 on non-.parquet
    extension or empty body; returns {path, bytes, filename}.

UI:
  - <input type=file accept=.parquet> with testid
    data-inspector-file-upload alongside the existing text path
    field. On change posts FormData to /upload/parquet, writes the
    returned absolute path back into the path input.
  - data-inspector-file-upload-error span surfaces network/HTTP errors.

Tests (8 total):
  - 5 pytest: persistence under /tmp + uuid name, non-parquet 400,
    empty-body 400, TTL cleanup drops aged file, save_upload runs
    cleanup per call.
  - 3 vitest: input testid + accept attribute, fetch POST +
    auto-populate path, HTTP error surfaces into error span.
  - 1 Playwright regression at vbgui/e2e/scenarios/
    91_data_inspector_file_upload.spec.ts: setInputFiles against a
    real fixture parquet, assert path field regex-matches
    /tmp/vbgui_uploads/<uuid>.parquet.

Author: apstenku123

cppmega_v4/jsonrpc/server.py

@@ -16,7 +16,9 @@
 import json
 from contextlib import asynccontextmanager
 
-from fastapi import FastAPI, WebSocket, WebSocketDisconnect
+from fastapi import (
+    FastAPI, File, HTTPException, UploadFile, WebSocket, WebSocketDisconnect,
+)
 from fastapi.middleware.cors import CORSMiddleware
 
 from cppmega_v4.jsonrpc.cache import LRUCache
@@ -74,6 +76,23 @@ def cache_clear() -> dict[str, str]:
         cache.clear()
         return {"status": "cleared"}
 
+    @app.post("/upload/parquet")
+    async def upload_parquet(file: UploadFile = File(...)) -> dict[str, str | int]:
+        """E-AUDIT-01: persist an uploaded parquet to /tmp/vbgui_uploads
+        and return its absolute path so DataInspector can populate the
+        path field. 24 h TTL + 100-file cap applied on every upload."""
+        from cppmega_v4.jsonrpc.uploads import save_upload
+        if not file.filename or not file.filename.endswith(".parquet"):
+            raise HTTPException(
+                status_code=400, detail="only .parquet uploads accepted")
+        body = await file.read()
+        if not body:
+            raise HTTPException(
+                status_code=400, detail="empty upload body")
+        path = save_upload(body)
+        return {"path": path, "bytes": len(body),
+                "filename": file.filename}
+
     @app.post("/rpc")
     async def rpc(payload: dict) -> dict:
         response = await _dispatch(payload, cache)

cppmega_v4/jsonrpc/uploads.py

@@ -0,0 +1,93 @@
+"""E-AUDIT-01: parquet upload sink + 24h TTL cleanup.
+
+DataInspector previously only accepted text paths to on-disk shards
+inside the repo. Users couldn't bring their own parquet without
+copying it into the working tree first. This module backs a single
+POST /upload/parquet endpoint that:
+
+  * writes the uploaded body to ``/tmp/vbgui_uploads/<uuid>.parquet``,
+  * tracks upload timestamps,
+  * cleans entries older than 24 h on every new upload (cheap),
+  * caps total upload count at 100 (drops oldest first) so a runaway
+    client can't fill /tmp.
+
+The endpoint is intentionally simple — no auth, no chunked uploads.
+Local-dev scope only; production must front this with a real
+auth/scan layer.
+"""
+
+from __future__ import annotations
+
+import pathlib
+import time
+import uuid
+from threading import Lock
+
+UPLOAD_ROOT = pathlib.Path("/tmp/vbgui_uploads")
+TTL_SECONDS: int = 24 * 60 * 60
+MAX_FILES: int = 100
+
+_LOCK = Lock()
+
+
+def _ensure_root() -> pathlib.Path:
+    UPLOAD_ROOT.mkdir(parents=True, exist_ok=True)
+    return UPLOAD_ROOT
+
+
+def cleanup_stale(now: float | None = None) -> int:
+    """Drop files older than TTL_SECONDS; return number removed."""
+    root = _ensure_root()
+    now = now if now is not None else time.time()
+    removed = 0
+    with _LOCK:
+        for p in sorted(root.iterdir()):
+            try:
+                age = now - p.stat().st_mtime
+            except OSError:
+                continue
+            if age > TTL_SECONDS:
+                try:
+                    p.unlink()
+                    removed += 1
+                except OSError:
+                    pass
+    return removed
+
+
+def _enforce_cap() -> int:
+    """Drop oldest files past MAX_FILES; return number removed."""
+    root = _ensure_root()
+    removed = 0
+    with _LOCK:
+        entries = sorted(
+            (p for p in root.iterdir() if p.is_file()),
+            key=lambda p: p.stat().st_mtime,
+        )
+        excess = len(entries) - MAX_FILES
+        for p in entries[:max(0, excess)]:
+            try:
+                p.unlink()
+                removed += 1
+            except OSError:
+                pass
+    return removed
+
+
+def save_upload(body: bytes, *, suffix: str = ".parquet") -> str:
+    """Persist an uploaded payload; return the absolute path string.
+
+    A fresh uuid4 stem prevents collision between concurrent uploads."""
+    cleanup_stale()
+    _enforce_cap()
+    root = _ensure_root()
+    name = f"{uuid.uuid4().hex}{suffix}"
+    path = root / name
+    path.write_bytes(body)
+    return str(path)
+
+
+__all__ = [
+    "UPLOAD_ROOT", "TTL_SECONDS", "MAX_FILES",
+    "save_upload", "cleanup_stale",
+]

tests/v4/test_upload_parquet.py

@@ -0,0 +1,99 @@
+"""E-AUDIT-01: POST /upload/parquet writes to /tmp/vbgui_uploads + TTL."""
+
+from __future__ import annotations
+
+import io
+import os
+import pathlib
+import time
+
+import pytest
+
+from cppmega_v4.jsonrpc.server import create_app
+from cppmega_v4.jsonrpc.uploads import (
+    UPLOAD_ROOT, TTL_SECONDS, cleanup_stale, save_upload,
+)
+
+
+@pytest.fixture
+def app():
+    return create_app()
+
+
+@pytest.fixture(autouse=True)
+def _wipe_uploads():
+    UPLOAD_ROOT.mkdir(parents=True, exist_ok=True)
+    for p in UPLOAD_ROOT.iterdir():
+        try: p.unlink()
+        except OSError: pass
+    yield
+    for p in UPLOAD_ROOT.iterdir():
+        try: p.unlink()
+        except OSError: pass
+
+
+def test_e_audit_01_upload_persists_under_tmp_with_uuid_name(app):
+    from fastapi.testclient import TestClient
+    client = TestClient(app)
+    body = b"PAR1" + b"\x00" * 64  # synthetic parquet-ish blob
+    resp = client.post(
+        "/upload/parquet",
+        files={"file": ("shard.parquet", io.BytesIO(body),
+                         "application/octet-stream")},
+    )
+    assert resp.status_code == 200, resp.text
+    j = resp.json()
+    assert j["bytes"] == len(body)
+    assert j["filename"] == "shard.parquet"
+    assert j["path"].startswith(str(UPLOAD_ROOT))
+    assert j["path"].endswith(".parquet")
+    p = pathlib.Path(j["path"])
+    assert p.is_file()
+    assert p.read_bytes() == body
+
+
+def test_e_audit_01_rejects_non_parquet_extension(app):
+    from fastapi.testclient import TestClient
+    client = TestClient(app)
+    resp = client.post(
+        "/upload/parquet",
+        files={"file": ("malicious.exe", io.BytesIO(b"x"),
+                         "application/octet-stream")},
+    )
+    assert resp.status_code == 400
+    assert "parquet" in resp.json()["detail"].lower()
+
+
+def test_e_audit_01_rejects_empty_body(app):
+    from fastapi.testclient import TestClient
+    client = TestClient(app)
+    resp = client.post(
+        "/upload/parquet",
+        files={"file": ("empty.parquet", io.BytesIO(b""),
+                         "application/octet-stream")},
+    )
+    assert resp.status_code == 400
+
+
+def test_e_audit_01_cleanup_stale_drops_old_files():
+    """Files older than TTL_SECONDS are removed by cleanup_stale."""
+    old = save_upload(b"x" * 16)
+    young = save_upload(b"y" * 16)
+    # Backdate `old` past TTL.
+    ancient = time.time() - TTL_SECONDS - 60
+    os.utime(old, (ancient, ancient))
+
+    removed = cleanup_stale()
+    assert removed >= 1
+    assert not pathlib.Path(old).exists()
+    assert pathlib.Path(young).exists()
+
+
+def test_e_audit_01_save_upload_runs_cleanup_per_call():
+    """save_upload() implicitly invokes cleanup_stale; an aged file
+    from a prior call is gone after the next upload."""
+    old = save_upload(b"a" * 8)
+    ancient = time.time() - TTL_SECONDS - 60
+    os.utime(old, (ancient, ancient))
+    _new = save_upload(b"b" * 8)
+    assert not pathlib.Path(old).exists()

vbgui/e2e/scenarios/91_data_inspector_file_upload.spec.ts

@@ -0,0 +1,40 @@
+// E-AUDIT-01: Playwright setInputFiles regression — drop a fixture
+// parquet via the new DataInspector file picker and assert the path
+// field auto-populates with /tmp/vbgui_uploads/<uuid>.parquet.
+
+import { test, expect } from "@playwright/test";
+import * as fs from "node:fs";
+import * as path from "node:path";
+import { gotoApp } from "../fixtures";
+
+test("E-AUDIT-01: file upload via DataInspector picker", async ({ page }) => {
+  test.setTimeout(30_000);
+  await gotoApp(page);
+
+  // Switch to the Data tab.
+  await page.getByTestId("app-tab-data").click();
+  const upload = page.getByTestId("data-inspector-file-upload");
+  await expect(upload).toBeVisible();
+
+  // Use any fixture parquet from tests/fixtures (the matrix builder
+  // populates these — pick the first available .parquet by glob).
+  const fixturesDir = path.resolve(
+    __dirname, "..", "..", "..", "tests", "fixtures");
+  let parquetPath: string | null = null;
+  for (const f of fs.readdirSync(fixturesDir)) {
+    if (f.endsWith(".parquet")) {
+      parquetPath = path.join(fixturesDir, f);
+      break;
+    }
+  }
+  if (!parquetPath) {
+    test.skip(true, "no .parquet fixture in tests/fixtures/");
+    return;
+  }
+  await upload.setInputFiles(parquetPath);
+
+  // Path field auto-populates with the backend-returned tmp path.
+  const pathInput = page.getByTestId("data-path");
+  await expect(pathInput).toHaveValue(
+    /\/tmp\/vbgui_uploads\/[0-9a-f]+\.parquet/, { timeout: 5000 });
+});

vbgui/src/components/DataInspector.tsx

@@ -196,6 +196,39 @@ export function DataInspector({
                value={path}
                onChange={(e) => setPath(e.target.value)}
                style={{ flex: 1, fontFamily: "monospace", fontSize: 11 }} />
+        {/* E-AUDIT-01: file upload picker. POSTs to /upload/parquet,
+            auto-populates the path field with the returned absolute
+            path under /tmp/vbgui_uploads/<uuid>.parquet. */}
+        <input data-testid="data-inspector-file-upload"
+               type="file" accept=".parquet"
+               onChange={async (ev) => {
+                 const f = ev.target.files?.[0];
+                 if (!f) return;
+                 const fd = new FormData();
+                 fd.append("file", f);
+                 const baseUrl = (
+                   (import.meta.env.VITE_BACKEND_URL as string | undefined)
+                   ?? "http://127.0.0.1:8765");
+                 try {
+                   const res = await fetch(
+                     `${baseUrl}/upload/parquet`,
+                     { method: "POST", body: fd });
+                   if (!res.ok) throw new Error(`HTTP ${res.status}`);
+                   const body = await res.json() as { path: string };
+                   setPath(body.path);
+                 } catch (err) {
+                   // Surface to the parent error pill — kept inline so
+                   // the upload picker has its own testable error testid.
+                   const msg = err instanceof Error
+                     ? err.message : String(err);
+                   const errEl = document.querySelector(
+                     "[data-testid='data-inspector-file-upload-error']");
+                   if (errEl) errEl.textContent = msg;
+                 }
+               }}
+               style={{ fontSize: 11 }} />
+        <span data-testid="data-inspector-file-upload-error"
+              style={{ color: "#b91c1c", fontSize: 10 }} />
         <button data-testid="data-load" onClick={() => load(0)}>
           Load
         </button>

vbgui/tests/DataInspectorFileUpload.test.tsx

@@ -0,0 +1,81 @@
+// E-AUDIT-01: DataInspector file upload picker.
+
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import { render, screen, fireEvent, waitFor } from "@testing-library/react";
+import { DataInspector } from "@/components/DataInspector";
+import type { RpcClient } from "@/lib/rpc";
+
+function makeRpc(): RpcClient {
+  return { call: vi.fn(async () => ({} as never)) } as unknown as RpcClient;
+}
+
+describe("E-AUDIT-01 DataInspector file upload", () => {
+  let origFetch: typeof fetch | undefined;
+  beforeEach(() => {
+    origFetch = globalThis.fetch;
+  });
+  afterEach(() => {
+    if (origFetch) globalThis.fetch = origFetch;
+  });
+
+  it("renders file input with the spec'd testid + accept=.parquet", () => {
+    render(<DataInspector rpc={makeRpc()} />);
+    const input = screen.getByTestId(
+      "data-inspector-file-upload") as HTMLInputElement;
+    expect(input).toBeTruthy();
+    expect(input.type).toBe("file");
+    expect(input.accept).toBe(".parquet");
+  });
+
+  it("uploads selected file via POST and auto-populates path field",
+     async () => {
+    const fetchMock = vi.fn(async () => ({
+      ok: true,
+      json: async () => ({ path: "/tmp/vbgui_uploads/abc123.parquet",
+                            bytes: 42, filename: "shard.parquet" }),
+    } as Response));
+    globalThis.fetch = fetchMock as unknown as typeof fetch;
+
+    render(<DataInspector rpc={makeRpc()} />);
+    const fileInput = screen.getByTestId(
+      "data-inspector-file-upload") as HTMLInputElement;
+    const blob = new File([new Uint8Array(42)], "shard.parquet",
+                          { type: "application/octet-stream" });
+    fireEvent.change(fileInput, { target: { files: [blob] } });
+
+    await waitFor(() => {
+      expect(fetchMock).toHaveBeenCalled();
+    });
+    const call = (fetchMock.mock.calls[0] ?? []) as
+      [string, RequestInit];
+    expect(call[0]).toMatch(/\/upload\/parquet$/);
+    expect(call[1].method).toBe("POST");
+    expect(call[1].body).toBeInstanceOf(FormData);
+
+    await waitFor(() => {
+      const pathInput = screen.getByTestId("data-path") as HTMLInputElement;
+      expect(pathInput.value).toBe("/tmp/vbgui_uploads/abc123.parquet");
+    });
+  });
+
+  it("surfaces HTTP error into the upload-error span", async () => {
+    const fetchMock = vi.fn(async () => ({
+      ok: false, status: 400,
+      json: async () => ({}),
+    } as Response));
+    globalThis.fetch = fetchMock as unknown as typeof fetch;
+
+    render(<DataInspector rpc={makeRpc()} />);
+    const fileInput = screen.getByTestId(
+      "data-inspector-file-upload") as HTMLInputElement;
+    const blob = new File([new Uint8Array(8)], "x.parquet",
+                          { type: "application/octet-stream" });
+    fireEvent.change(fileInput, { target: { files: [blob] } });
+
+    await waitFor(() => {
+      const err = screen.getByTestId(
+        "data-inspector-file-upload-error");
+      expect(err.textContent).toContain("400");
+    });
+  });
+});