feat(vbgui-fa.2): cppmega-run CLI + pipeline.yaml + pipeline.run JSON-RPC

Stage F-A.2 of the Visual Builder GUI epic (cppmega-mlx-o0k). Lands
the generic runner described in VisualBuilderPlan.md §5.3.

- runner/stages.py: 12 built-in stage callables (parse,
  verify_build_spec, apply_rewrites, resolve_shapes, estimate_memory,
  check_gotchas, build_model, dry_forward, input_parity_check,
  loss_smoke, optimizer_smoke, train). Each is a pure function over
  StageContext; failures fold into StageResult(status="fail").
- runner/pipeline.py: Pipeline manifest (declarative stages +
  options + continue_on_failure), Pipeline.from_yaml loader,
  PipelineReport, run_pipeline orchestrator. Stops on first failure
  by default, marks remaining stages as "skipped".
- runner/cli.py: cppmega-run argparse entry. Smoke pipeline by
  default, --stages all / smoke / csv aliases, --pipeline yaml,
  --json output, --continue-on-failure. Exit codes 0/1/2.
- jsonrpc/dispatcher.py: pipeline.run method now routes through the
  runner. Lazy import breaks the schema↔runner cycle.
- pyproject.toml: console-script cppmega-run + cppmega_v4* added to
  setuptools packages.find include list.

Tests (46 new): pipeline registry sanity, every built-in stage in
isolation, pipeline failure cascade + continue_on_failure, yaml
round-trip, JSON dumpability, CLI exit codes + JSON output +
--stages smoke/all/csv, pipeline.run JSON-RPC dispatch, subprocess
system test against `python -m cppmega_v4.runner.cli`.

Full v4 regression: 2054 passed / 5 skipped / 15 xfailed / 0 failed.

Closes cppmega-mlx-o0k.6.

Author: apstenku123

cppmega_v4/jsonrpc/dispatcher.py

@@ -28,6 +28,8 @@
     JsonRpcError,
     JsonRpcRequest,
     JsonRpcResponse,
+    PipelineRunParams,
+    PipelineRunResult,
     ProbeRunParams,
     SuggestAdaptersParams,
     SuggestShardingParams,
@@ -61,9 +63,26 @@
         ProbeRunParams,
         lambda p, c: probe_run(p, cache=c),
     ),
+    "pipeline.run": (
+        PipelineRunParams,
+        lambda p, c: _pipeline_run(p),
+    ),
 }
 
 
+def _pipeline_run(params: PipelineRunParams) -> PipelineRunResult:
+    # Lazy import — runner package depends on jsonrpc.schema for VerifyParams;
+    # binding the symbols here breaks the import cycle.
+    from cppmega_v4.runner import Pipeline, run_pipeline
+    pipeline = Pipeline.from_dict({
+        "stages": params.pipeline.stages,
+        "stage_options": params.pipeline.stage_options,
+        "continue_on_failure": params.pipeline.continue_on_failure,
+    })
+    report = run_pipeline(params.spec, pipeline)
+    return PipelineRunResult.model_validate(report.to_dict())
+
+
 def dispatch(
     payload: Mapping[str, Any] | JsonRpcRequest,
     *,

cppmega_v4/runner/__init__.py

@@ -0,0 +1,35 @@
+"""``cppmega-run`` pipeline runner.
+
+See ``VisualBuilderPlan.md`` §5.3 for the design.
+
+Stage F-A.2 surface:
+  - stages: STAGE_REGISTRY (12 built-in stages, SMOKE_STAGES, FULL_STAGES)
+  - pipeline: Pipeline / PipelineReport / run_pipeline
+  - cli: ``cppmega-run`` entry point exposed via console-script
+"""
+
+from __future__ import annotations
+
+from cppmega_v4.runner.pipeline import (
+    Pipeline,
+    PipelineReport,
+    run_pipeline,
+)
+from cppmega_v4.runner.stages import (
+    FULL_STAGES,
+    SMOKE_STAGES,
+    STAGE_REGISTRY,
+    StageContext,
+    StageResult,
+)
+
+__all__ = [
+    "FULL_STAGES",
+    "Pipeline",
+    "PipelineReport",
+    "SMOKE_STAGES",
+    "STAGE_REGISTRY",
+    "StageContext",
+    "StageResult",
+    "run_pipeline",
+]

cppmega_v4/runner/cli.py

@@ -0,0 +1,138 @@
+"""``cppmega-run`` — single CLI entry exposing the pipeline runner.
+
+Usage:
+    cppmega-run spec.json                          # smoke pipeline, human
+    cppmega-run spec.json --json                   # smoke pipeline, machine
+    cppmega-run spec.json --stages all             # full 11-stage pipeline
+    cppmega-run spec.json --stages parse,verify_build_spec,resolve_shapes
+    cppmega-run spec.json --pipeline pipeline.yaml
+    cppmega-run spec.json --pipeline pipeline.yaml --continue-on-failure
+
+Exit code 0 iff all stages either ``ok`` or ``skipped``. Exit 1 on any
+``fail``. Exit 2 on invalid input (spec parse / pipeline parse / argument
+errors).
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import Sequence
+
+from pydantic import ValidationError
+
+from cppmega_v4.jsonrpc.schema import VerifyParams
+from cppmega_v4.runner.pipeline import (
+    Pipeline,
+    PipelineReport,
+    run_pipeline,
+)
+from cppmega_v4.runner.stages import FULL_STAGES, SMOKE_STAGES, STAGE_REGISTRY
+
+
+def _build_parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(
+        prog="cppmega-run",
+        description="Run the cppmega Visual Builder pipeline on a model_spec.json",
+    )
+    p.add_argument("spec", help="Path to model_spec.json (VerifyParams shape)")
+    p.add_argument("--pipeline", help="Path to pipeline.yaml manifest")
+    p.add_argument(
+        "--stages",
+        help="Comma-separated stage list OR 'smoke' / 'all'. "
+             "Overrides --pipeline stage list.",
+    )
+    p.add_argument(
+        "--continue-on-failure", action="store_true",
+        help="Do not stop on first failed stage.",
+    )
+    p.add_argument("--json", action="store_true",
+                   help="Emit JSON report to stdout.")
+    return p
+
+
+def _resolve_pipeline(args: argparse.Namespace) -> Pipeline:
+    if args.pipeline:
+        pipe = Pipeline.from_yaml(args.pipeline)
+    else:
+        pipe = Pipeline.smoke()
+    if args.stages:
+        stages = _expand_stage_list(args.stages)
+        pipe = Pipeline(
+            stages=stages,
+            stage_options=pipe.stage_options,
+            continue_on_failure=pipe.continue_on_failure,
+        )
+    if args.continue_on_failure:
+        pipe = Pipeline(
+            stages=pipe.stages,
+            stage_options=pipe.stage_options,
+            continue_on_failure=True,
+        )
+    return pipe
+
+
+def _expand_stage_list(arg: str) -> tuple[str, ...]:
+    if arg == "smoke":
+        return SMOKE_STAGES
+    if arg == "all":
+        return FULL_STAGES
+    stages = tuple(s.strip() for s in arg.split(",") if s.strip())
+    unknown = [s for s in stages if s not in STAGE_REGISTRY]
+    if unknown:
+        raise SystemExit(
+            f"cppmega-run: unknown stage(s) {unknown!r}; "
+            f"available: {sorted(STAGE_REGISTRY)}"
+        )
+    return stages
+
+
+def _load_spec(path: str) -> VerifyParams:
+    raw = json.loads(Path(path).read_text())
+    return VerifyParams.model_validate(raw)
+
+
+def _print_human(report: PipelineReport) -> None:
+    for s in report.stages:
+        marker = {"ok": "PASS", "fail": "FAIL", "skipped": "SKIP"}[s.status]
+        line = f"  [{marker}] {s.name:<22} {s.elapsed_ms:8.2f} ms"
+        if s.error:
+            line += f"  error={s.error.get('type','?')}: {s.error.get('detail','?')[:80]}"
+        print(line)
+    print(f"  overall: {report.overall_status} "
+          f"total {report.total_elapsed_ms:.2f} ms")
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    parser = _build_parser()
+    args = parser.parse_args(argv)
+    try:
+        spec = _load_spec(args.spec)
+    except (FileNotFoundError, json.JSONDecodeError) as exc:
+        print(f"cppmega-run: failed to read spec: {exc}", file=sys.stderr)
+        return 2
+    except ValidationError as exc:
+        print(f"cppmega-run: invalid spec: {exc}", file=sys.stderr)
+        return 2
+
+    try:
+        pipeline = _resolve_pipeline(args)
+    except (FileNotFoundError, ValueError) as exc:
+        print(f"cppmega-run: pipeline error: {exc}", file=sys.stderr)
+        return 2
+
+    report = run_pipeline(spec, pipeline)
+
+    if args.json:
+        json.dump(report.to_dict(), sys.stdout, indent=2)
+        print()
+    else:
+        _print_human(report)
+
+    return 0 if report.overall_status == "ok" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

cppmega_v4/runner/pipeline.py

@@ -0,0 +1,106 @@
+"""Pipeline orchestrator — walk a list of stages, collect StageResults.
+
+Loads pipeline.yaml manifests, validates stage names, then dispatches
+to :data:`cppmega_v4.runner.stages.STAGE_REGISTRY`.
+
+Stops on the first failed stage unless ``continue_on_failure=True`` is
+set in the manifest. Skipped stages don't count as failures.
+"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Mapping
+
+import yaml
+
+from cppmega_v4.jsonrpc.schema import VerifyParams
+from cppmega_v4.runner.stages import (
+    SMOKE_STAGES,
+    STAGE_REGISTRY,
+    StageContext,
+    StageResult,
+)
+
+
+@dataclass
+class Pipeline:
+    """Declarative manifest: which stages, in order, with options."""
+
+    stages: tuple[str, ...]
+    stage_options: dict[str, dict[str, Any]] = field(default_factory=dict)
+    continue_on_failure: bool = False
+
+    def __post_init__(self) -> None:
+        unknown = [s for s in self.stages if s not in STAGE_REGISTRY]
+        if unknown:
+            raise ValueError(
+                f"unknown stage(s) {unknown!r}; "
+                f"available: {sorted(STAGE_REGISTRY)}"
+            )
+
+    @classmethod
+    def smoke(cls) -> Pipeline:
+        return cls(stages=SMOKE_STAGES)
+
+    @classmethod
+    def from_yaml(cls, path: str | Path) -> Pipeline:
+        data = yaml.safe_load(Path(path).read_text())
+        return cls.from_dict(data)
+
+    @classmethod
+    def from_dict(cls, data: Mapping[str, Any]) -> Pipeline:
+        stages = data.get("stages")
+        if not stages:
+            raise ValueError("pipeline manifest missing 'stages'")
+        return cls(
+            stages=tuple(stages),
+            stage_options=dict(data.get("stage_options", {})),
+            continue_on_failure=bool(data.get("continue_on_failure", False)),
+        )
+
+
+@dataclass
+class PipelineReport:
+    """Final pipeline run rollup."""
+
+    stages: list[StageResult]
+    overall_status: str
+    total_elapsed_ms: float
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "stages": [s.to_dict() for s in self.stages],
+            "overall_status": self.overall_status,
+            "total_elapsed_ms": round(self.total_elapsed_ms, 3),
+        }
+
+
+def run_pipeline(spec: VerifyParams, pipeline: Pipeline) -> PipelineReport:
+    """Run ``pipeline`` over ``spec`` and return the report.
+
+    Stops on the first failed stage unless ``continue_on_failure`` is set.
+    """
+    t0 = time.perf_counter()
+    ctx = StageContext(spec=spec, options=pipeline.stage_options)
+    results: list[StageResult] = []
+    for name in pipeline.stages:
+        stage = STAGE_REGISTRY[name]
+        result = stage(ctx)
+        results.append(result)
+        if result.status == "fail" and not pipeline.continue_on_failure:
+            # Mark every subsequent stage as skipped for visibility.
+            for remaining in pipeline.stages[len(results):]:
+                results.append(StageResult(
+                    name=remaining, status="skipped", elapsed_ms=0.0,
+                ))
+            break
+    overall = "ok"
+    if any(r.status == "fail" for r in results):
+        overall = "fail"
+    elapsed = (time.perf_counter() - t0) * 1000.0
+    return PipelineReport(
+        stages=results, overall_status=overall, total_elapsed_ms=elapsed,
+    )