fix(vertex): synchronous CustomJob.submit; train UI GCP arch sync

- Replace CustomJob.run(sync=False) with submit() so create_custom_job
finishes before the HTTP response (avoids phantom submitted_count).
- Return job_resource_names for verification; log each created job.
- Include Vertex launch module and related training/dashboard updates.

Made-with: Cursor

Dockerfile.training

@@ -1,8 +1,13 @@
 # GPU training image for Vertex AI / local CUDA (PyTorch + PyTorch Geometric).
-# Build: docker build -f Dockerfile.training -t excel-chunker-train .
+#
+# Base tag must exist on Docker Hub (e.g. cudnn8-runtime, not cudnn9 unless published).
+# Push to Artifact Registry (avoids docker-credential-gcloud PATH issues on Mac):
+#   ./scripts/push_training_image.sh
+#
+# Local build (no push): docker build -f Dockerfile.training -t excel-chunker-train .
 # Run (CPU smoke): docker run --rm excel-chunker-train python train_compare.py --help
 
-FROM pytorch/pytorch:2.3.1-cuda12.1-cudnn9-runtime
+FROM pytorch/pytorch:2.3.1-cuda12.1-cudnn8-runtime
 
 WORKDIR /app
 

_sync_labels.py

@@ -1,24 +1,10 @@
-"""Sync latest label files from Drive to local data/labeled/."""
-import drive_client, metadata_client, config
+#!/usr/bin/env python3
+"""Backward-compatible entrypoint — use ``scripts/sync_labels_from_drive.py``."""
 
-rows = metadata_client.get_all_rows()
-labelled = [r for r in rows if r["status"] == "labelled" and r.get("labels_file_id") and r["labels_file_id"] != "local-only"]
-print(f"Labelled sheets on Drive: {len(labelled)}")
+import subprocess
+import sys
+from pathlib import Path
 
-downloaded = 0
-failed = 0
-for r in labelled:
-    sid = r["drive_file_id"]
-    sn = r["sheet_name"]
-    safe = f"{sid}_{sn}".replace("/", "_").replace(" ", "_") + ".json"
-    local = config.LABELED_DIR / safe
-    try:
-        drive_client.download_file(r["labels_file_id"], str(local))
-        downloaded += 1
-        if downloaded % 20 == 0:
-            print(f"  ...downloaded {downloaded}/{len(labelled)}")
-    except Exception as e:
-        print(f"  FAILED: {safe}: {e}")
-        failed += 1
-
-print(f"Done: {downloaded} downloaded, {failed} failed -> {config.LABELED_DIR}")
+root = Path(__file__).resolve().parent
+script = root / "scripts" / "sync_labels_from_drive.py"
+raise SystemExit(subprocess.call([sys.executable, str(script)] + sys.argv[1:]))

app.py

@@ -23,7 +23,7 @@ from typing import Any, Optional
 from fastapi import FastAPI, HTTPException, Query, Depends, Header, Request, Response
 from fastapi.staticfiles import StaticFiles
 from fastapi.responses import FileResponse, JSONResponse
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 import config
 from training_progress import SuiteProgress
@@ -123,6 +123,13 @@ def _ensure_xlsx(drive_file_id: str) -> str:
 
 app = FastAPI(title="Spreadsheet Labeler", lifespan=_app_lifespan)
 
+
+@app.get("/favicon.ico", include_in_schema=False)
+def favicon_ico():
+    """Browsers request this by default; avoid noisy 404 in devtools."""
+    return Response(status_code=204)
+
+
 _claim_lock = threading.Lock()
 _labelled_count_floor = 0
 _prediction_cache: dict[str, list[dict]] = {}
@@ -433,7 +440,7 @@ def save_and_next(payload: SavePayload, _=Depends(_require_auth)):
             "cells": [c.model_dump() for c in payload.cells],
         }
 
-    safe_name = f"{payload.spreadsheet_id}_{payload.sheet_name}".replace("/", "_").replace(" ", "_")
+    safe_name = metadata_client.label_file_stem(payload.spreadsheet_id, payload.sheet_name)
     json_bytes = json.dumps(labeled_data, indent=2, ensure_ascii=False).encode("utf-8")
 
     # Save locally
@@ -601,7 +608,7 @@ def save_review(payload: SavePayload, _=Depends(_require_auth)):
             "cells": [c.model_dump() for c in payload.cells],
         }
 
-    safe_name = f"{payload.spreadsheet_id}_{payload.sheet_name}".replace("/", "_").replace(" ", "_")
+    safe_name = metadata_client.label_file_stem(payload.spreadsheet_id, payload.sheet_name)
     json_bytes = json.dumps(labeled_data, indent=2, ensure_ascii=False).encode("utf-8")
 
     local_label_path = config.LABELED_DIR / f"{safe_name}.json"
@@ -652,7 +659,7 @@ def backfill_review_labels(_=Depends(_require_auth)):
     for r in rows:
         sid = r.get("drive_file_id", "")
         sn = r.get("sheet_name", "")
-        safe = f"{sid}_{sn}".replace("/", "_").replace(" ", "_") + ".json"
+        safe = metadata_client.label_file_stem(sid, sn) + ".json"
         row_by_key[safe] = r
 
     updated = 0
@@ -755,7 +762,7 @@ def _load_saved_labels(
     labels_file_id: str = "",
 ) -> list[dict] | None:
     """Load previously saved labels from local JSON or Drive."""
-    safe_name = f"{spreadsheet_id}_{sheet_name}".replace("/", "_").replace(" ", "_")
+    safe_name = metadata_client.label_file_stem(spreadsheet_id, sheet_name)
     local_path = config.LABELED_DIR / f"{safe_name}.json"
 
     data = None
@@ -1063,25 +1070,55 @@ _MAX_TRAIN_LOG_LINES = 200
 _GCP_SUITE_CACHE: dict[str, Any] = {
     "uri": "",
     "ts": 0.0,
+    "wall_ts": 0.0,
     "data": None,
     "err": None,
 }
 _GCP_SUITE_LOCK = threading.Lock()
-_GCP_SUITE_TTL_SEC = 5.0
+# Short TTL: dashboard polls ~2s; keeps GCS list fresh without hammering the API.
+_GCP_SUITE_TTL_SEC = 2.0
+
+
+def _invalidate_gcp_suite_cache() -> None:
+    """Force the next /api/train/status to refetch GCS (e.g. after launching new Vertex jobs)."""
+    with _GCP_SUITE_LOCK:
+        _GCP_SUITE_CACHE["ts"] = 0.0
+
+
+def _gcp_wall_ts_iso(wall_ts: float) -> str | None:
+    if not wall_ts or wall_ts <= 0:
+        return None
+    return datetime.fromtimestamp(wall_ts, tz=timezone.utc).isoformat(timespec="seconds")
+
+_docker_image_lock = threading.Lock()
+_docker_image_status: dict[str, Any] = {
+    "running": False,
+    "log": [],
+    "ok": None,
+    "error": None,
+}
 
 
-def _get_gcp_suite_cached() -> tuple[dict[str, Any] | None, str | None]:
-    """Merge remote Vertex progress from ``…/experiments/*/progress.json`` on GCS (cached)."""
-    uri = (os.environ.get("EXCEL_CHUNKER_GCS_EXPERIMENTS_URI") or "").strip()
+def _get_gcp_suite_cached() -> tuple[dict[str, Any] | None, str | None, str | None]:
+    """Merge remote Vertex progress from ``…/experiments/*/progress.json`` on GCS (cached).
+
+    Returns ``(suite_dict, error, refreshed_at_iso)`` for the train dashboard.
+    """
+    uri = config.gcp_experiments_uri_for_progress()
     if not uri:
-        return None, None
+        return None, None, None
     now = time.monotonic()
     with _GCP_SUITE_LOCK:
         if (
             _GCP_SUITE_CACHE["uri"] == uri
             and now - float(_GCP_SUITE_CACHE["ts"]) < _GCP_SUITE_TTL_SEC
         ):
-            return _GCP_SUITE_CACHE["data"], _GCP_SUITE_CACHE["err"]
+            wt = float(_GCP_SUITE_CACHE.get("wall_ts") or 0)
+            return (
+                _GCP_SUITE_CACHE["data"],
+                _GCP_SUITE_CACHE["err"],
+                _gcp_wall_ts_iso(wt),
+            )
     err: str | None = None
     data: dict[str, Any] | None = None
     try:
@@ -1091,12 +1128,18 @@ def _get_gcp_suite_cached() -> tuple[dict[str, Any] | None, str | None]:
     except Exception as e:
         err = str(e)[:800]
         logger.warning("GCS training progress fetch failed: %s", e)
+    refreshed_iso: str | None = None
     with _GCP_SUITE_LOCK:
         _GCP_SUITE_CACHE["uri"] = uri
         _GCP_SUITE_CACHE["ts"] = now
         _GCP_SUITE_CACHE["data"] = data
         _GCP_SUITE_CACHE["err"] = err
-    return data, err
+        if data is not None:
+            _GCP_SUITE_CACHE["wall_ts"] = time.time()
+            refreshed_iso = _gcp_wall_ts_iso(float(_GCP_SUITE_CACHE["wall_ts"]))
+        else:
+            refreshed_iso = _gcp_wall_ts_iso(float(_GCP_SUITE_CACHE.get("wall_ts") or 0))
+    return data, err, refreshed_iso
 
 
 @app.get("/api/train/dataset_stats")
@@ -1478,15 +1521,173 @@ def experiment_detail(run: str, _=Depends(_require_auth)):
 @app.get("/api/train/status")
 def train_status():
     """Return current training status, log, and per-unit progress (local and/or GCS)."""
-    gcp_uri = (os.environ.get("EXCEL_CHUNKER_GCS_EXPERIMENTS_URI") or "").strip()
-    gcp_suite, gcp_err = _get_gcp_suite_cached()
-    return {
+    gcp_uri = config.gcp_experiments_uri_for_progress()
+    gcp_suite, gcp_err, gcp_refreshed_at = _get_gcp_suite_cached()
+    if gcp_suite is not None and gcp_refreshed_at:
+        gcp_suite = {**gcp_suite, "refreshed_at": gcp_refreshed_at}
+    body = {
         **_train_status,
         "suite": _train_suite.to_api_dict(),
         "gcp_progress_enabled": bool(gcp_uri),
         "gcp_suite": gcp_suite,
         "gcp_suite_error": gcp_err,
     }
+    return JSONResponse(
+        content=body,
+        headers={
+            "Cache-Control": "no-store, no-cache",
+            "Pragma": "no-cache",
+        },
+    )
+
+
+@app.get("/api/train/gcp-defaults")
+def train_gcp_defaults(_=Depends(_require_auth)):
+    """Default GCP/Vertex values for the dashboard (from config / .env)."""
+    return {
+        "project_id": config.GCP_PROJECT_ID,
+        "region": config.GCP_REGION,
+        "gcs_data_uri": config.GCP_GCS_DATA_URI,
+        "gcs_experiments_uri": config.gcp_experiments_uri_for_progress(),
+        "artifact_registry_repo": config.GCP_ARTIFACT_REGISTRY_REPO,
+        "image_uri": config.gcp_training_image_uri(),
+        "training_image_name": config.GCP_TRAINING_IMAGE_NAME,
+        "training_image_tag": config.GCP_TRAINING_IMAGE_TAG,
+        "machine_type": config.GCP_VERTEX_MACHINE_TYPE,
+        "accelerator_type": config.GCP_VERTEX_ACCELERATOR_TYPE,
+        "accelerator_count": config.GCP_VERTEX_ACCELERATOR_COUNT,
+        "vertex_spot": False,
+        "docker_build_from_dashboard": config.docker_build_from_dashboard_enabled(),
+    }
+
+
+@app.post("/api/train/gcp-image-build")
+def start_gcp_image_build(_=Depends(_require_auth)):
+    """Run ``docker build`` + ``docker push`` for the training image (local machine only)."""
+    if not config.docker_build_from_dashboard_enabled():
+        raise HTTPException(
+            status_code=403,
+            detail="Włącz ENABLE_DOCKER_BUILD_FROM_DASHBOARD=1 tylko lokalnie (wymaga Dockera).",
+        )
+    with _docker_image_lock:
+        if _docker_image_status["running"]:
+            raise HTTPException(status_code=409, detail="Build już trwa.")
+        _docker_image_status["running"] = True
+        _docker_image_status["log"] = []
+        _docker_image_status["ok"] = None
+        _docker_image_status["error"] = None
+
+    def _run() -> None:
+        def append(line: str) -> None:
+            line = line.rstrip()
+            if not line:
+                return
+            with _docker_image_lock:
+                log = _docker_image_status["log"]
+                log.append(line)
+                if len(log) > 500:
+                    del log[: len(log) - 500]
+
+        try:
+            from docker_image_build import run_docker_build_and_push
+
+            ok, msg = run_docker_build_and_push(
+                image_uri=config.gcp_training_image_uri(),
+                repo_root=config.BASE_DIR,
+                log_append=append,
+            )
+            with _docker_image_lock:
+                _docker_image_status["ok"] = ok
+                _docker_image_status["error"] = None if ok else msg
+        except Exception as e:
+            logger.exception("docker build/push failed")
+            with _docker_image_lock:
+                _docker_image_status["ok"] = False
+                _docker_image_status["error"] = str(e)
+        finally:
+            with _docker_image_lock:
+                _docker_image_status["running"] = False
+
+    threading.Thread(target=_run, daemon=True).start()
+    return {"started": True}
+
+
+@app.get("/api/train/gcp-image-build/status")
+def gcp_image_build_status(_=Depends(_require_auth)):
+    with _docker_image_lock:
+        return {
+            "running": _docker_image_status["running"],
+            "log": list(_docker_image_status["log"]),
+            "ok": _docker_image_status["ok"],
+            "error": _docker_image_status["error"],
+        }
+
+
+class GcpLaunchRequest(BaseModel):
+    architectures: str = "gat"
+    seeds: str = "42"
+    folds: int = 5
+    epochs: int = 100
+    subset_percent: float = Field(100.0, ge=1.0, le=100.0)
+    project_id: Optional[str] = None
+    region: Optional[str] = None
+    image_uri: Optional[str] = None
+    gcs_data_uri: Optional[str] = None
+    machine_type: str = Field(default_factory=lambda: config.GCP_VERTEX_MACHINE_TYPE)
+    accelerator_type: str = Field(default_factory=lambda: config.GCP_VERTEX_ACCELERATOR_TYPE)
+    accelerator_count: int = Field(default_factory=lambda: config.GCP_VERTEX_ACCELERATOR_COUNT)
+    vertex_spot: bool = False
+    # Optional per-architecture accelerator (Vertex enum). Omitted archs use accelerator_type.
+    arch_accelerators: Optional[dict[str, str]] = None
+
+
+@app.post("/api/train/gcp-launch")
+def train_gcp_launch(req: GcpLaunchRequest, _=Depends(_require_auth)):
+    """Submit Vertex Custom Training jobs (same logic as ``scripts/launch_vertex_training.py``)."""
+    try:
+        from vertex_launch import submit_vertex_training_jobs
+    except ImportError as e:
+        raise HTTPException(
+            status_code=500,
+            detail="Install Vertex deps: pip install google-cloud-aiplatform",
+        ) from e
+
+    project = (req.project_id or config.GCP_PROJECT_ID).strip()
+    region = (req.region or config.GCP_REGION).strip()
+    image = (req.image_uri or config.gcp_training_image_uri()).strip()
+    gcs_uri = (req.gcs_data_uri or config.GCP_GCS_DATA_URI).strip()
+    archs = [a.strip() for a in req.architectures.split(",") if a.strip()]
+    seeds: list[int] = []
+    for s in req.seeds.split(","):
+        s = s.strip()
+        if s:
+            seeds.append(int(s))
+    if not archs or not seeds:
+        raise HTTPException(status_code=400, detail="architectures and seeds must be non-empty")
+
+    try:
+        out = submit_vertex_training_jobs(
+            project=project,
+            region=region,
+            image_uri=image,
+            gcs_data_uri=gcs_uri,
+            architectures=archs,
+            seeds=seeds,
+            folds=req.folds,
+            epochs=req.epochs,
+            subset_percent=req.subset_percent,
+            machine_type=req.machine_type,
+            accelerator_type=req.accelerator_type,
+            accelerator_count=req.accelerator_count,
+            use_spot=req.vertex_spot,
+            arch_accelerator=req.arch_accelerators,
+        )
+    except Exception as e:
+        logger.exception("Vertex launch failed")
+        raise HTTPException(status_code=500, detail=str(e)) from e
+
+    _invalidate_gcp_suite_cache()
+    return {"status": "ok", **out}
 
 
 class TrainRequest(BaseModel):
@@ -1502,6 +1703,13 @@ class TrainRequest(BaseModel):
     clear_previous: bool = False
     no_va_margin_loss: bool = False
     no_col_consistency_loss: bool = False
+    subset_percent: float = Field(
+        100.0,
+        ge=1.0,
+        le=100.0,
+        description="Random subset of labeled sheets before training (100 = all).",
+    )
+    subset_seed: int = Field(0, ge=0, description="RNG seed for subset sampling when < 100%.")
 
 
 @app.post("/api/train/start")
@@ -1521,9 +1729,10 @@ def start_training(req: TrainRequest, _=Depends(_require_auth)):
         try:
             if req.mode == "compare":
                 _train_suite = SuiteProgress()
+                sub = f"{req.subset_percent:g}% of sheets" if req.subset_percent < 100 else "100% of sheets"
                 _train_status["log"].append(
                     f"Architectures: {req.architectures}, Seeds: {req.seeds}, "
-                    f"Folds: {req.folds}, Epochs: {req.epochs}"
+                    f"Folds: {req.folds}, Epochs: {req.epochs}, Data: {sub} (seed {req.subset_seed})"
                 )
 
                 if req.clear_previous:
@@ -1563,22 +1772,42 @@ def start_training(req: TrainRequest, _=Depends(_require_auth)):
                 old_stdout = sys.stdout
                 sys.stdout = LogCapture()
                 try:
-                    result = run_experiments(
+                    sp = None if req.subset_percent >= 100.0 else req.subset_percent
+                    result, failed_units = run_experiments(
                         architectures=archs, seeds=seeds,
                         n_folds=req.folds, num_epochs=req.epochs,
                         do_ablations=req.ablations,
                         use_va_margin_loss=not req.no_va_margin_loss,
                         use_col_consistency_loss=not req.no_col_consistency_loss,
+                        subset_percent=sp,
+                        subset_seed=req.subset_seed,
                         progress=_train_suite,
                     )
                 finally:
                     sys.stdout = old_stdout
 
-                _train_status["result"] = {"status": "completed", "architectures": list(result.keys())}
-                _train_status["log"].append("Training completed successfully!")
+                if failed_units:
+                    _train_status["result"] = {
+                        "status": "failed",
+                        "failed_units": failed_units,
+                        "architectures": list(result.keys()),
+                    }
+                    _train_status["log"].append(
+                        f"Training finished with {failed_units} failed unit(s) "
+                        f"(e.g. CUDA OOM). Check log lines above."
+                    )
+                else:
+                    _train_status["result"] = {
+                        "status": "completed",
+                        "architectures": list(result.keys()),
+                    }
+                    _train_status["log"].append("Training completed successfully!")
 
             elif req.mode == "production":
-                _train_status["log"].append(f"Training production model for {req.epochs} epochs...")
+                sub = f"{req.subset_percent:g}% of sheets" if req.subset_percent < 100 else "100% of sheets"
+                _train_status["log"].append(
+                    f"Training production model for {req.epochs} epochs ({sub}, seed {req.subset_seed})..."
+                )
                 from train_gnn import train_model
 
                 class LogCapture(io.StringIO):
@@ -1590,7 +1819,13 @@ def start_training(req: TrainRequest, _=Depends(_require_auth)):
                 old_stdout = sys.stdout
                 sys.stdout = LogCapture()
                 try:
-                    train_model(num_epochs=req.epochs, skip_predict=True)
+                    sp = None if req.subset_percent >= 100.0 else req.subset_percent
+                    train_model(
+                        num_epochs=req.epochs,
+                        skip_predict=True,
+                        subset_percent=sp,
+                        subset_seed=req.subset_seed,
+                    )
                 finally:
                     sys.stdout = old_stdout
 
@@ -1625,25 +1860,21 @@ def run_evaluation(_=Depends(_require_auth)):
 
 @app.post("/api/train/embed")
 def run_embeddings(_=Depends(_require_auth)):
-    """Generate text embeddings for all labeled files that don't have them yet."""
-    count = 0
-    skipped = 0
-    errors = []
+    """Optional: embed every labeled JSON (same as ``embed_text.embed_all``).
+
+    Training paths already call ``ensure_embeddings_for_labeled`` before loading graphs.
+    """
     try:
-        from embed_text import embed_file
-        for fp in sorted(config.LABELED_DIR.glob("*.json")):
-            embed_path = fp.with_name(fp.stem + "_embeddings.npz")
-            if embed_path.exists():
-                skipped += 1
-                continue
-            try:
-                embed_file(fp)
-                count += 1
-            except Exception as e:
-                errors.append(f"{fp.name}: {e}")
+        from embed_text import ensure_embeddings_for_labeled
+
+        r = ensure_embeddings_for_labeled()
     except ImportError:
         return {"error": "embed_text module not available"}
-    return {"embedded": count, "skipped": skipped, "errors": errors}
+    return {
+        "embedded": r["generated"],
+        "skipped": r["skipped"],
+        "errors": r["errors"],
+    }
 
 
 # ── RAG Evaluation ─────────────────────────────────────────────────────────
@@ -2066,6 +2297,10 @@ def rag_eval_run(
     resume_structure: bool = Query(True),
     force_structure_metrics: bool = Query(False),
     save_structure_details: bool = Query(False),
+    timing_exclude_first_sheet: bool = Query(
+        False,
+        description="Exclude first predict_sheet from inference_timing mean/p95 (warmup)",
+    ),
     _=Depends(_require_auth),
 ):
     """Trigger a RAG evaluation run in the background.
@@ -2100,6 +2335,7 @@ def rag_eval_run(
                 resume_structure=resume_structure,
                 force_structure_metrics=force_structure_metrics,
                 save_structure_details=save_structure_details,
+                timing_exclude_first_sheet=timing_exclude_first_sheet,
             )
             _rag_eval_status["result"] = result
             _rag_eval_status["log"].append("RAG evaluation completed.")

config.py

@@ -81,6 +81,51 @@ EXPERIMENTS_DIR.mkdir(parents=True, exist_ok=True)
 RAG_EVAL_CACHE_DIR.mkdir(parents=True, exist_ok=True)
 RAG_EVAL_LABELED_DIR.mkdir(parents=True, exist_ok=True)
 
+# ── Vertex AI / GCS (defaults for dashboard; override via .env) ─────────────
+GCP_PROJECT_ID = (os.environ.get("GCP_PROJECT_ID") or "acoustic-atom-386613").strip()
+GCP_REGION = (os.environ.get("GCP_REGION") or "europe-west1").strip()
+GCP_GCS_DATA_URI = (os.environ.get("GCP_GCS_DATA_URI") or "gs://excel-chunker/data").strip()
+GCP_ARTIFACT_REGISTRY_REPO = (os.environ.get("GCP_ARTIFACT_REGISTRY_REPO") or "excel-chunker").strip()
+GCP_TRAINING_IMAGE_NAME = (os.environ.get("GCP_TRAINING_IMAGE_NAME") or "excel-chunker-train").strip()
+GCP_TRAINING_IMAGE_TAG = (os.environ.get("GCP_TRAINING_IMAGE_TAG") or "latest").strip()
+GCP_TRAINING_IMAGE_URI = (os.environ.get("GCP_TRAINING_IMAGE_URI") or "").strip()
+
+# Vertex Custom Training: VM + GPU must match (e.g. L4 → G2, not N1).
+# Override if you only have T4 quota: GCP_VERTEX_ACCELERATOR_TYPE=NVIDIA_TESLA_T4,
+# GCP_VERTEX_MACHINE_TYPE=n1-standard-8
+GCP_VERTEX_MACHINE_TYPE = (os.environ.get("GCP_VERTEX_MACHINE_TYPE") or "g2-standard-8").strip()
+GCP_VERTEX_ACCELERATOR_TYPE = (os.environ.get("GCP_VERTEX_ACCELERATOR_TYPE") or "NVIDIA_L4").strip()
+GCP_VERTEX_ACCELERATOR_COUNT = max(
+    1,
+    int((os.environ.get("GCP_VERTEX_ACCELERATOR_COUNT") or "1").strip() or "1"),
+)
+
+
+def gcp_training_image_uri() -> str:
+    """Full Artifact Registry Docker image URI for Vertex training workers."""
+    if GCP_TRAINING_IMAGE_URI:
+        return GCP_TRAINING_IMAGE_URI
+    return (
+        f"{GCP_REGION}-docker.pkg.dev/{GCP_PROJECT_ID}/"
+        f"{GCP_ARTIFACT_REGISTRY_REPO}/{GCP_TRAINING_IMAGE_NAME}:{GCP_TRAINING_IMAGE_TAG}"
+    )
+
+
+def gcp_experiments_uri_for_progress() -> str:
+    """GCS prefix for live progress in the dashboard (explicit env or ``.../data/experiments``)."""
+    explicit = (os.environ.get("EXCEL_CHUNKER_GCS_EXPERIMENTS_URI") or "").strip()
+    if explicit:
+        return explicit
+    if GCP_GCS_DATA_URI:
+        return GCP_GCS_DATA_URI.rstrip("/") + "/experiments"
+    return ""
+
+
+def docker_build_from_dashboard_enabled() -> bool:
+    """Allow POST /api/train/gcp-image-build only on trusted machines with Docker (local dev)."""
+    v = (os.environ.get("ENABLE_DOCKER_BUILD_FROM_DASHBOARD") or "").strip().lower()
+    return v in ("1", "true", "yes", "on")
+
 
 def get_google_credentials():
     """Return service account credentials (works for Sheets; read-only Drive)."""

embed_text.py

@@ -12,6 +12,7 @@ Usage:
 from __future__ import annotations
 
 import json
+from collections.abc import Callable
 from pathlib import Path
 
 import numpy as np
@@ -80,11 +81,61 @@ def embed_file(filepath: Path) -> Path:
     return out_path
 
 
+def _labeled_json_files(labeled_dir: Path) -> list[Path]:
+    files = sorted(labeled_dir.glob("*.json"))
+    return [f for f in files if not f.stem.endswith("_embeddings")]
+
+
+def needs_embedding_file(json_path: Path) -> bool:
+    """True if *_embeddings.npz is missing or older than the labeled JSON."""
+    out = json_path.with_name(json_path.stem + "_embeddings.npz")
+    if not out.exists():
+        return True
+    try:
+        return json_path.stat().st_mtime > out.stat().st_mtime
+    except OSError:
+        return True
+
+
+def ensure_embeddings_for_labeled(
+    *,
+    labeled_dir: Path | None = None,
+    log: Callable[[str], None] = print,
+) -> dict[str, int | list[str]]:
+    """Create or refresh ``*_embeddings.npz`` for every labeled JSON that needs it.
+
+    Call this before ``load_all_graphs`` / training so E5 text features exist.
+    Returns ``{"generated": n, "skipped": n, "errors": [...]}``.
+    """
+    root = labeled_dir or config.LABELED_DIR
+    json_files = _labeled_json_files(root)
+    generated = 0
+    skipped = 0
+    errors: list[str] = []
+
+    for fp in json_files:
+        if not needs_embedding_file(fp):
+            skipped += 1
+            continue
+        try:
+            embed_file(fp)
+            generated += 1
+        except Exception as e:
+            errors.append(f"{fp.name}: {e}")
+
+    if generated:
+        log(f"  [embed] Done: {generated} file(s) written, {skipped} already up to date.")
+    elif json_files:
+        log(f"  [embed] All {len(json_files)} labeled file(s) already have up-to-date embeddings.")
+    if errors:
+        log(f"  [embed] {len(errors)} error(s) (training may use zeros for those sheets).")
+    return {"generated": generated, "skipped": skipped, "errors": errors}
+
+
 def embed_all() -> None:
     """Process all labeled JSON files in the labeled directory."""
     labeled_dir = config.LABELED_DIR
-    json_files = sorted(labeled_dir.glob("*.json"))
-    json_files = [f for f in json_files if not f.stem.endswith("_embeddings")]
+    json_files = _labeled_json_files(labeled_dir)
 
     if not json_files:
         print(f"No labeled files found in {labeled_dir}")

eval_rag.py

@@ -29,6 +29,12 @@ Evaluation dimensions:
   3. LLM-as-a-judge: answer generation + correctness scoring
   4. (Learned graph methods) Structure vs human labels: levels 1–4 via ``evaluate_sheet``,
      saved to ``data/rag_eval_structure_metrics.json`` (see ``rag_structure_rag_analysis.py`` for joins).
+  5. (Learned graph methods) ``predict_sheet`` wall time per held-out sheet; aggregates
+     (mean/p95 ms per graph node, throughput) in ``inference_timing`` on the same JSON entry.
+     First sheet may include one-off CPU/GPU warmup — interpret percentiles accordingly.
+     When structure metrics are recomputed, the new entry is merged with the previous one so
+     ``inference_timing_per_sheet`` (and other extra keys) are not dropped unless this run
+     saves a fresh per-sheet list with ``--save-structure-details``.
 
 Usage:
     python eval_rag.py                          # run on rag_eval_dataset.json
@@ -41,6 +47,7 @@ from __future__ import annotations
 import json
 import logging
 import os
+import time
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Callable, Optional
@@ -81,6 +88,69 @@ def _embed_texts(model, texts: list[str], batch_size: int = 64) -> np.ndarray:
 GNN_CELL_LIMIT = 3000
 
 
+def _rag_grid_cell_counts(grid: dict) -> tuple[int, int]:
+    """Return (n_json_cells, n_graph_nodes).
+
+    ``n_graph_nodes`` matches ``predict_sheet`` nodes: cells with non-empty ``value``.
+    ``n_json_cells`` is ``len(cells)`` on the grid dict.
+    """
+    cells = grid.get("cells") or []
+    n_json = len(cells)
+    n_graph = sum(1 for c in cells if (c.get("value") or "") != "")
+    return n_json, n_graph
+
+
+def _aggregate_predict_timing(
+    per_sheet: list[dict[str, Any]],
+    *,
+    n_skipped_limit: int = 0,
+    stats_rows: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    """Aggregate wall-clock ``predict_sheet`` timings into JSON-serialisable metrics.
+
+    ``per_sheet`` is the full list of attempts (for counts). ``stats_rows``, if set,
+    is the subset used for mean / percentiles (e.g. exclude the first sheet for warmup).
+    """
+    rows_stats = stats_rows if stats_rows is not None else per_sheet
+    ok_rows = [r for r in rows_stats if r.get("success")]
+    n_failed = sum(1 for r in per_sheet if not r.get("skipped_limit") and not r.get("success"))
+
+    def _ms_per_node(r: dict) -> float:
+        n = max(int(r.get("n_graph_nodes") or 0), 1)
+        return float(r["elapsed_s"]) * 1000.0 / n
+
+    def _ms_per_json(r: dict) -> float:
+        n = max(int(r.get("n_json_cells") or 0), 1)
+        return float(r["elapsed_s"]) * 1000.0 / n
+
+    ms_nodes = [_ms_per_node(r) for r in ok_rows]
+    ms_json = [_ms_per_json(r) for r in ok_rows]
+    total_s = sum(float(r["elapsed_s"]) for r in ok_rows)
+    total_nodes = sum(int(r.get("n_graph_nodes") or 0) for r in ok_rows)
+
+    def _pctl(arr: list[float], q: float) -> float | None:
+        if not arr:
+            return None
+        return float(np.percentile(np.array(arr, dtype=np.float64), q))
+
+    out: dict[str, Any] = {
+        "n_sheets_attempted": len(per_sheet),
+        "n_sheets_used_for_percentiles": len(rows_stats),
+        "n_sheets_predict_ok": len(ok_rows),
+        "n_sheets_predict_failed": n_failed,
+        "n_sheets_skipped_gnn_cell_limit": int(n_skipped_limit),
+        "predict_total_s": round(total_s, 4) if ok_rows else None,
+        "total_graph_nodes": int(total_nodes) if ok_rows else None,
+        "throughput_nodes_per_s": round(total_nodes / total_s, 2) if ok_rows and total_s > 0 else None,
+        "mean_ms_per_graph_node": round(float(np.mean(ms_nodes)), 3) if ms_nodes else None,
+        "p50_ms_per_graph_node": (round(v, 3) if (v := _pctl(ms_nodes, 50)) is not None else None),
+        "p95_ms_per_graph_node": (round(v, 3) if (v := _pctl(ms_nodes, 95)) is not None else None),
+        "mean_ms_per_json_cell": round(float(np.mean(ms_json)), 3) if ms_json else None,
+        "p95_ms_per_json_cell": (round(v, 3) if (v := _pctl(ms_json, 95)) is not None else None),
+    }
+    return out
+
+
 def _predict_and_label(
     grid_data: dict,
     model=None,
@@ -137,26 +207,88 @@ def _fill_prediction_cache(
     sheet_data: dict[str, dict],
     model,
     log: Callable[[str], None] | None = None,
-) -> dict[str, list[dict] | None]:
-    """One ``predict_sheet`` per held-out sheet (under GNN_CELL_LIMIT). Used for chunking + structure metrics."""
+    *,
+    timing_exclude_first_sheet: bool = False,
+) -> tuple[dict[str, list[dict] | None], dict[str, Any]]:
+    """One ``predict_sheet`` per held-out sheet (under GNN_CELL_LIMIT).
+
+    Returns ``(pred_cache, timing_bundle)`` where ``timing_bundle`` has keys
+    ``aggregate`` (from :func:`_aggregate_predict_timing`) and ``per_sheet`` (list of
+    per-attempt records for optional persistence / debugging).
+    """
     from predict import predict_sheet
 
     cache: dict[str, list[dict] | None] = {}
+    per_sheet: list[dict[str, Any]] = []
+    n_skipped_limit = 0
+
     for sheet_id, data in sheet_data.items():
+        n_json, n_graph = _rag_grid_cell_counts(data)
         n_cells = len(data.get("cells", []))
         if n_cells > GNN_CELL_LIMIT:
             cache[sheet_id] = None
+            n_skipped_limit += 1
             continue
+        t0 = time.perf_counter()
         try:
             p = predict_sheet(data, model_override=model)
+            elapsed = time.perf_counter() - t0
             cache[sheet_id] = p if p else None
+            success = bool(p)
+            per_sheet.append({
+                "sheet_id": sheet_id,
+                "elapsed_s": round(elapsed, 6),
+                "n_json_cells": n_json,
+                "n_graph_nodes": n_graph,
+                "success": success,
+                "skipped_limit": False,
+                "ms_per_graph_node": round(elapsed * 1000.0 / max(n_graph, 1), 4),
+                "ms_per_json_cell": round(elapsed * 1000.0 / max(n_json, 1), 4),
+            })
         except (RuntimeError, MemoryError) as e:
+            elapsed = time.perf_counter() - t0
             if log:
                 log(f"  predict_sheet failed {sheet_id[:40]}... ({n_cells} cells): {e}")
             logger.warning("predict_sheet failed on %s: %s", sheet_id, e)
             import gc; gc.collect()
             cache[sheet_id] = None
-    return cache
+            per_sheet.append({
+                "sheet_id": sheet_id,
+                "elapsed_s": round(elapsed, 6),
+                "n_json_cells": n_json,
+                "n_graph_nodes": n_graph,
+                "success": False,
+                "skipped_limit": False,
+                "error": str(e),
+                "ms_per_graph_node": round(elapsed * 1000.0 / max(n_graph, 1), 4),
+                "ms_per_json_cell": round(elapsed * 1000.0 / max(n_json, 1), 4),
+            })
+
+    stats_rows = per_sheet
+    if timing_exclude_first_sheet and len(per_sheet) > 1:
+        stats_rows = per_sheet[1:]
+    agg = _aggregate_predict_timing(
+        per_sheet, n_skipped_limit=n_skipped_limit, stats_rows=stats_rows,
+    )
+    bundle = {"aggregate": agg, "per_sheet": per_sheet}
+    if log:
+        if per_sheet:
+            a = agg
+            log(
+                f"  predict_sheet timing: total={a.get('predict_total_s')}s "
+                f"ok={a.get('n_sheets_predict_ok')}/{a.get('n_sheets_attempted')} "
+                f"mean_ms/node={a.get('mean_ms_per_graph_node')} "
+                f"p95_ms/node={a.get('p95_ms_per_graph_node')} "
+                f"nodes/s={a.get('throughput_nodes_per_s')} "
+                f"skipped_limit={a.get('n_sheets_skipped_gnn_cell_limit')}"
+            )
+        else:
+            log(
+                f"  predict_sheet timing: no predict attempts "
+                f"(skipped GNN_CELL_LIMIT={n_skipped_limit} / {len(sheet_data)} sheets)"
+            )
+
+    return cache, bundle
 
 
 def _aggregate_structure_sheet_results(rows: list[dict]) -> dict[str, float | None]:
@@ -1009,8 +1141,15 @@ def _should_skip_rag_resume(
     resume: bool,
     existing_agg: dict[str, Any],
     log_fn: Callable[[str], None] | None,
+    *,
+    need_judge: bool = False,
 ) -> bool:
-    """Skip re-eval if this key already has a successful checkpoint (no ``error``)."""
+    """Skip re-eval if this key already has a successful checkpoint (no ``error``).
+
+    If ``need_judge`` is True (LLM-as-a-judge requested), only skip when retrieval
+    metrics *and* judge fields are present — otherwise a prior ``--no-judge`` run
+    would block adding judge scores.
+    """
     if not resume:
         return False
     val = existing_agg.get(key)
@@ -1018,7 +1157,15 @@ def _should_skip_rag_resume(
         return False
     if val.get("error"):
         return False
-    if any(k in val for k in ("recall@1", "mrr", "judge_score", "judge_binary_acc")):
+    has_retrieval = any(k in val for k in ("recall@1", "mrr"))
+    has_judge = any(k in val for k in ("judge_score", "judge_binary_acc"))
+    if need_judge:
+        if has_retrieval and has_judge:
+            if log_fn:
+                log_fn(f"  SKIP {key} (resume — already in rag_eval_results.json)")
+            return True
+        return False
+    if has_retrieval or has_judge:
         if log_fn:
             log_fn(f"  SKIP {key} (resume — already in rag_eval_results.json)")
         return True
@@ -1086,6 +1233,7 @@ def run_rag_evaluation_v2(
     resume_structure: bool = True,
     force_structure_metrics: bool = False,
     save_structure_details: bool = False,
+    timing_exclude_first_sheet: bool = False,
 ) -> dict:
     """Run full RAG evaluation using the rag_eval_dataset.json.
 
@@ -1108,6 +1256,9 @@ def run_rag_evaluation_v2(
         resume_structure: skip rewriting ``rag_eval_structure_metrics.json`` when an entry exists.
         force_structure_metrics: recompute structure-vs-gold even if a checkpoint exists.
         save_structure_details: include per-sheet ``_per_sheet`` in the structure metrics JSON.
+        timing_exclude_first_sheet: if True, mean / percentiles for ``inference_timing`` exclude
+            the first attempted ``predict_sheet`` call (reduces one-off warmup skew); counts
+            still reflect all sheets.
     """
     def log(msg: str):
         print(msg)
@@ -1297,7 +1448,9 @@ def run_rag_evaluation_v2(
 
             for strat in strategies:
                 key = f"{method_key_base}/{strat}"
-                if _should_skip_rag_resume(key, resume, existing_agg, log):
+                if _should_skip_rag_resume(
+                    key, resume, existing_agg, log, need_judge=use_llm_judge,
+                ):
                     results[key] = dict(existing_agg[key])
                     continue
                 log(f"  Evaluating {key}...")
@@ -1348,7 +1501,20 @@ def run_rag_evaluation_v2(
                 continue
 
             log("  predict_sheet cache (one inference per sheet)...")
-            pred_cache = _fill_prediction_cache(sheet_data, model, log)
+            pred_cache, timing_bundle = _fill_prediction_cache(
+                sheet_data, model, log,
+                timing_exclude_first_sheet=timing_exclude_first_sheet,
+            )
+            inf_agg = timing_bundle.get("aggregate") or {}
+            inf_per = timing_bundle.get("per_sheet") or []
+
+            def _attach_inference_timing(entry: dict[str, Any]) -> dict[str, Any]:
+                out = dict(entry)
+                out["inference_timing"] = inf_agg
+                if save_structure_details and inf_per:
+                    out["inference_timing_per_sheet"] = list(inf_per)
+                # If this run did not emit per-sheet timing, keep any existing list from ``entry``.
+                return out
 
             if not _should_skip_structure_resume(
                 method_key_base, resume_structure, structure_metrics_doc, force_structure_metrics,
@@ -1357,7 +1523,10 @@ def run_rag_evaluation_v2(
                     smeta = _compute_structure_vs_gold_for_arch(
                         sheet_data, pred_cache, save_per_sheet=save_structure_details,
                     )
+                    prev_entry = structure_metrics_doc.get(method_key_base)
+                    prev_dict = prev_entry if isinstance(prev_entry, dict) else {}
                     entry = {
+                        **prev_dict,
                         "method_base": method_key_base,
                         "chunking": gm,
                         "computed_at": datetime.now(timezone.utc).isoformat(),
@@ -1365,7 +1534,8 @@ def run_rag_evaluation_v2(
                         "held_out_sheet_count": len(sheet_data),
                         **smeta,
                     }
-                    structure_metrics_doc[method_key_base] = entry
+                    entry.pop("structure_metrics_error", None)
+                    structure_metrics_doc[method_key_base] = _attach_inference_timing(entry)
                     _write_rag_structure_metrics_doc(structure_metrics_doc)
                     log(
                         f"  Structure vs gold: n_sheets={smeta['n_sheets_with_gold']} → "
@@ -1374,8 +1544,28 @@ def run_rag_evaluation_v2(
                 except Exception as e:
                     log(f"  WARNING: structure metrics failed for {method_key_base}: {e}")
                     logger.exception("structure metrics")
+                    prev = structure_metrics_doc.get(method_key_base)
+                    if not isinstance(prev, dict):
+                        prev = {}
+                    fallback = {
+                        **prev,
+                        "method_base": method_key_base,
+                        "chunking": gm,
+                        "computed_at": datetime.now(timezone.utc).isoformat(),
+                        "structure_metrics_error": str(e),
+                    }
+                    structure_metrics_doc[method_key_base] = _attach_inference_timing(fallback)
+                    _write_rag_structure_metrics_doc(structure_metrics_doc)
             else:
                 log(f"  SKIP structure metrics for {method_key_base} (resume_structure)")
+                prev = structure_metrics_doc.get(method_key_base)
+                if not isinstance(prev, dict):
+                    prev = {}
+                merged = {**prev, "method_base": method_key_base, "chunking": gm}
+                merged["inference_timing_updated_at"] = datetime.now(timezone.utc).isoformat()
+                structure_metrics_doc[method_key_base] = _attach_inference_timing(merged)
+                _write_rag_structure_metrics_doc(structure_metrics_doc)
+                log(f"  Updated inference_timing only → {config.RAG_EVAL_STRUCTURE_METRICS.name}")
 
             all_chunks = []
             try:
@@ -1450,7 +1640,9 @@ def run_rag_evaluation_v2(
 
             for strat in strategies:
                 key = f"{method_key_base}/{strat}"
-                if _should_skip_rag_resume(key, resume, existing_agg, log):
+                if _should_skip_rag_resume(
+                    key, resume, existing_agg, log, need_judge=use_llm_judge,
+                ):
                     results[key] = dict(existing_agg[key])
                     continue
                 log(f"  Evaluating {key}...")
@@ -1545,7 +1737,9 @@ def run_rag_evaluation_v2(
 
         for strat in strategies:
             key = f"{method}/{strat}"
-            if _should_skip_rag_resume(key, resume, existing_agg, log):
+            if _should_skip_rag_resume(
+                key, resume, existing_agg, log, need_judge=use_llm_judge,
+            ):
                 results[key] = dict(existing_agg[key])
                 continue
             log(f"  Evaluating {key}...")
@@ -1782,6 +1976,11 @@ if __name__ == "__main__":
         action="store_true",
         help="Store per-sheet _per_sheet in rag_eval_structure_metrics.json",
     )
+    parser.add_argument(
+        "--timing-exclude-first-sheet",
+        action="store_true",
+        help="For inference_timing aggregates only: drop the first predict_sheet call from mean/p95 (warmup)",
+    )
     args = parser.parse_args()
 
     if args.pull_drive:
@@ -1805,4 +2004,5 @@ if __name__ == "__main__":
             resume_structure=not args.no_resume_structure,
             force_structure_metrics=args.force_structure_metrics,
             save_structure_details=args.save_structure_details,
+            timing_exclude_first_sheet=args.timing_exclude_first_sheet,
         )

gcp_progress_reader.py

@@ -77,7 +77,10 @@ def fetch_suite_from_gcs(uri: str) -> dict[str, Any]:
     def _sort_key(u: dict) -> tuple:
         return (u.get("seed", 0), u.get("fold", 0), u.get("arch", ""), u.get("run_tag", ""))
 
+    # Stable fallback order (seed / fold / arch).
     units.sort(key=_sort_key)
+    # Newest worker activity first so a fresh Vertex run is visible at the top.
+    units.sort(key=lambda u: u.get("updated_at") or "", reverse=True)
     done_like = {"done", "skipped", "error"}
     completed = sum(1 for u in units if u.get("status") in done_like)
     running = any(u.get("status") == "running" for u in units)

metadata_client.py

@@ -51,6 +51,11 @@ COL_TIMESTAMP = 7
 COL_NUM_LABELED_CELLS = 8
 
 
+def label_file_stem(drive_file_id: str, sheet_name: str) -> str:
+    """Stem for ``data/labeled/{stem}.json``; must match ``app.py`` save/load."""
+    return f"{drive_file_id}_{sheet_name}".replace("/", "_").replace(" ", "_")
+
+
 def _get_service():
     global _service
     if _service is None:

rag_structure_rag_analysis.py

@@ -41,6 +41,21 @@ RAG_KEYS = (
     "source_mrr",
 )
 
+# Flattened from ``rag_eval_structure_metrics.json`` ``inference_timing`` (eval_rag /predict_sheet).
+INFERENCE_TIMING_KEYS = (
+    "mean_ms_per_graph_node",
+    "p50_ms_per_graph_node",
+    "p95_ms_per_graph_node",
+    "mean_ms_per_json_cell",
+    "p95_ms_per_json_cell",
+    "predict_total_s",
+    "throughput_nodes_per_s",
+    "n_sheets_predict_ok",
+    "n_sheets_attempted",
+    "n_sheets_skipped_gnn_cell_limit",
+    "n_sheets_predict_failed",
+)
+
 
 def _load_json(path: Path) -> dict[str, Any]:
     if not path.exists():
@@ -79,6 +94,9 @@ def build_joined_rows(
             row[k] = agg.get(k)
         for k in RAG_KEYS:
             row[f"rag_{k}"] = rrag.get(k)
+        it = sentry.get("inference_timing") if isinstance(sentry.get("inference_timing"), dict) else {}
+        for k in INFERENCE_TIMING_KEYS:
+            row[f"inf_{k}"] = it.get(k)
         rows.append(row)
 
     meta = {

requirements-app.txt

@@ -4,5 +4,6 @@ google-api-python-client
 google-auth
 google-auth-oauthlib
 google-cloud-storage>=2.14.0
+google-cloud-aiplatform>=1.38.0
 python-dotenv
 openpyxl

requirements.txt

@@ -3,6 +3,7 @@ uvicorn[standard]
 google-api-python-client
 google-auth
 google-auth-oauthlib
+google-cloud-aiplatform>=1.38.0
 python-dotenv
 openpyxl
 datasets

run_ablation_sweep.py

@@ -81,7 +81,7 @@ def run_sweep(
             continue
 
         t0 = time.time()
-        exp_results = run_experiments(
+        exp_results, n_failed = run_experiments(
             architectures=[arch],
             seeds=seeds,
             n_folds=n_folds,
@@ -89,6 +89,8 @@ def run_sweep(
             graph_cfg=graph_cfg,
             max_nodes=max_nodes,
         )
+        if n_failed:
+            print(f"  WARNING: {n_failed} training unit(s) failed in this config")
         elapsed = time.time() - t0
 
         summary = {

scripts/launch_vertex_training.py

@@ -16,19 +16,28 @@ Example::
     --architectures gat,gcn \\
     --seeds 42 --folds 5
 
-Spot VMs: enable in the Cloud Console for the job, or use ``gcloud ai custom-jobs create`` with
-``--scheduling-strategy=SPOT``; the Python SDK flag varies by version.
+By default: **NVIDIA_L4** on **g2-standard-8** (quota e.g. ``CustomModelTrainingL4GPUsPerProjectPerRegion``).
+L4 requires a **G2** machine type; the launcher fixes N1+T4-style combos automatically.
+Use ``--accelerator-type NVIDIA_TESLA_T4 --machine-type n1-standard-8`` if you only have T4 quota.
+Pass ``--spot`` + ``--accelerator-type NVIDIA_TESLA_P100`` if you only have preemptible P100 quota.
 """
 
 from __future__ import annotations
 
 import argparse
 import sys
+from pathlib import Path
+
+_ROOT = Path(__file__).resolve().parents[1]
+if str(_ROOT) not in sys.path:
+    sys.path.insert(0, str(_ROOT))
+
+from vertex_launch import submit_vertex_training_jobs  # noqa: E402
 
 
 def main() -> None:
     try:
-        from google.cloud import aiplatform
+        import google.cloud.aiplatform  # noqa: F401
     except ImportError:
         print("Install: pip install -r requirements-gcp.txt", file=sys.stderr)
         raise SystemExit(1)
@@ -46,13 +55,40 @@ def main() -> None:
     parser.add_argument("--seeds", default="42", help="Comma-separated ints")
     parser.add_argument("--folds", type=int, default=5)
     parser.add_argument("--epochs", type=int, default=100)
+    parser.add_argument(
+        "--subset-percent",
+        type=float,
+        default=100.0,
+        help="Random subset of sheets on worker (1-100; 100 = all)",
+    )
     parser.add_argument(
         "--machine-type",
-        default="n1-standard-8",
-        help="e.g. n1-standard-8",
+        default="g2-standard-8",
+        help="G2 for L4 (default); N1 for T4/P100",
+    )
+    parser.add_argument(
+        "--accelerator-type",
+        default="NVIDIA_L4",
+        help="Vertex enum: NVIDIA_L4 (default), NVIDIA_TESLA_T4, …",
     )
-    parser.add_argument("--accelerator-type", default="NVIDIA_TESLA_T4")
     parser.add_argument("--accelerator-count", type=int, default=1)
+    parser.add_argument(
+        "--heavy-archs",
+        default="",
+        help="Comma-separated architectures that use --heavy-accelerator "
+        "(e.g. adj_transformer,spatial_edge_transformer,dual_modality_gnn). "
+        "Others use --accelerator-type.",
+    )
+    parser.add_argument(
+        "--heavy-accelerator",
+        default="NVIDIA_L4",
+        help="GPU for --heavy-archs (default: L4; pair resolved with --machine-type)",
+    )
+    parser.add_argument(
+        "--spot",
+        action="store_true",
+        help="SPOT/preemptible scheduling (use with P100 if you lack on-demand T4 quota)",
+    )
     parser.add_argument(
         "--staging-bucket",
         default=None,
@@ -60,71 +96,36 @@ def main() -> None:
     )
     args = parser.parse_args()
 
-    staging = args.staging_bucket
-    if not staging:
-        uri = args.gcs_data_uri.rstrip("/")
-        if not uri.startswith("gs://"):
-            raise SystemExit("--gcs-data-uri must start with gs://")
-        rest = uri[5:]
-        bucket = rest.split("/", 1)[0]
-        staging = f"gs://{bucket}"
-
     archs = [a.strip() for a in args.architectures.split(",") if a.strip()]
     seeds = [int(s.strip()) for s in args.seeds.split(",") if s.strip()]
 
-    aiplatform.init(project=args.project, location=args.region, staging_bucket=staging)
-
-    if args.gcs_data_uri.startswith("gs://"):
-        without = args.gcs_data_uri[5:]
-        parts = without.split("/", 1)
-        bucket = parts[0]
-        prefix = parts[1] if len(parts) > 1 else ""
-        fuse_root = f"/gcs/{bucket}" + (f"/{prefix}" if prefix else "")
-    else:
-        fuse_root = args.gcs_data_uri
-
-    jobs = 0
-    for seed in seeds:
-        for fold_idx in range(args.folds):
-            for arch in archs:
-                display = f"train-{arch}-s{seed}-f{fold_idx}"
-                train_args = [
-                    "train_compare.py",
-                    "--architectures",
-                    arch,
-                    "--seeds",
-                    str(seed),
-                    "--folds",
-                    str(args.folds),
-                    "--epochs",
-                    str(args.epochs),
-                    "--fold-index",
-                    str(fold_idx),
-                ]
-                env = [{"name": "EXCEL_CHUNKER_DATA_ROOT", "value": fuse_root}]
-                worker_pool = {
-                    "machine_spec": {
-                        "machine_type": args.machine_type,
-                        "accelerator_type": args.accelerator_type,
-                        "accelerator_count": args.accelerator_count,
-                    },
-                    "replica_count": 1,
-                    "container_spec": {
-                        "image_uri": args.image,
-                        "command": ["python"],
-                        "args": train_args,
-                        "env": env,
-                    },
-                }
-                job = aiplatform.CustomJob(
-                    display_name=display[:128],
-                    worker_pool_specs=[worker_pool],
-                )
-                job.run(sync=False)
-                print(f"Submitted: {display}")
-                jobs += 1
-
-    print(f"Total jobs submitted: {jobs}")
+    arch_accelerator = None
+    heavy_raw = (args.heavy_archs or "").strip()
+    if heavy_raw:
+        heavy_set = {a.strip() for a in heavy_raw.split(",") if a.strip()}
+        hacc = (args.heavy_accelerator or "").strip() or "NVIDIA_L4"
+        arch_accelerator = {a: hacc for a in archs if a in heavy_set}
+
+    out = submit_vertex_training_jobs(
+        project=args.project,
+        region=args.region,
+        image_uri=args.image,
+        gcs_data_uri=args.gcs_data_uri,
+        architectures=archs,
+        seeds=seeds,
+        folds=args.folds,
+        epochs=args.epochs,
+        subset_percent=args.subset_percent,
+        machine_type=args.machine_type,
+        accelerator_type=args.accelerator_type,
+        accelerator_count=args.accelerator_count,
+        staging_bucket=args.staging_bucket,
+        use_spot=args.spot,
+        arch_accelerator=arch_accelerator,
+    )
+    for d in out["jobs"]:
+        print(f"Submitted: {d}")
+    print(f"Total jobs submitted: {out['submitted_count']}")
 
 
 if __name__ == "__main__":

static/rag_dashboard.html

@@ -233,6 +233,7 @@ table.lb .best-cell{font-weight:700}
     <h3>Structure vs gold + RAG (same retrieval strategy)</h3>
     <p style="font-size:12px;color:var(--gray);margin:-6px 0 14px;line-height:1.45">
       Rows: GNN chunking runs (<code>graph_*</code> / <code>graph_row_*</code>) with human labels on held-out sheets.
+      <strong>Inference timing</strong> is wall-clock <code>predict_sheet</code> per sheet (graph build + forward + post); ms/node uses non-empty <code>value</code> cells as graph nodes (same as the GNN).
       <strong>Oracle RAG scores</strong> (human chunking) measure a ceiling that still depends on retrieval, generation, and Q&amp;A difficulty — not only perfect structure.
     </p>
     <div style="overflow-x:auto">
@@ -245,6 +246,10 @@ table.lb .best-cell{font-weight:700}
           <th>Judge</th>
           <th>R@1</th>
           <th>sR@1</th>
+          <th title="Mean wall ms per graph node (non-empty value cells); predict_sheet">mean ms/node</th>
+          <th title="95th percentile ms per graph node across sheets">p95 ms/node</th>
+          <th title="Sum of successful predict_sheet wall times (s)">predict total s</th>
+          <th title="total_graph_nodes / predict_total_s">nodes/s</th>
         </tr></thead>
         <tbody id="st-body"></tbody>
       </table>
@@ -634,12 +639,12 @@ async function loadStructureAnalysis() {
   const spDiv = document.getElementById('st-spearman');
   const pDiv = document.getElementById('st-pearson');
   if (!tbody) return;
-  tbody.innerHTML = '<tr><td colspan="7">Loading…</td></tr>';
+  tbody.innerHTML = '<tr><td colspan="11">Loading…</td></tr>';
   try {
     const r = await fetch('/api/rag_eval/structure_analysis?strategy=' + encodeURIComponent(strat)).then(x => x.json());
     structureAnalysisCache = r;
     if (!r.exists || !r.joined_rows) {
-      tbody.innerHTML = '<tr><td colspan="7" class="empty">No joined data. Run RAG eval with graph methods and ensure rag_eval_structure_metrics.json exists.</td></tr>';
+      tbody.innerHTML = '<tr><td colspan="11" class="empty">No joined data. Run RAG eval with graph methods and ensure rag_eval_structure_metrics.json exists.</td></tr>';
       if (disc) { disc.style.display = 'none'; }
       if (spDiv) spDiv.innerHTML = '';
       if (pDiv) pDiv.innerHTML = '';
@@ -660,6 +665,10 @@ async function loadStructureAnalysis() {
         <td>${fmtF(row.rag_judge_score, 2)}</td>
         <td>${fmtF(row['rag_recall@1'], 3)}</td>
         <td>${fmtF(row['rag_source_recall@1'], 3)}</td>
+        <td>${row.inf_mean_ms_per_graph_node != null ? fmtF(row.inf_mean_ms_per_graph_node, 2) : '—'}</td>
+        <td>${row.inf_p95_ms_per_graph_node != null ? fmtF(row.inf_p95_ms_per_graph_node, 2) : '—'}</td>
+        <td>${row.inf_predict_total_s != null ? fmtF(row.inf_predict_total_s, 3) : '—'}</td>
+        <td>${row.inf_throughput_nodes_per_s != null ? fmtF(row.inf_throughput_nodes_per_s, 1) : '—'}</td>
       </tr>`;
     }).join('');
     const sv = an.spearman_structure_vs_rag_judge || {};
@@ -689,7 +698,7 @@ async function loadStructureAnalysis() {
     } else if (pDiv) pDiv.innerHTML = '<span class="empty">Need ≥2 rows with all structure metrics.</span>';
     renderScatter(r.joined_rows);
   } catch (e) {
-    tbody.innerHTML = '<tr><td colspan="7">Error: ' + esc(String(e)) + '</td></tr>';
+    tbody.innerHTML = '<tr><td colspan="11">Error: ' + esc(String(e)) + '</td></tr>';
   }
 }
 

static/train.html

@@ -146,6 +146,15 @@ body{font-family:-apple-system,'Segoe UI',system-ui,sans-serif;background:var(--
 .filter-bar .filter-chip:hover{border-color:var(--text-3);color:var(--text-1)}
 .filter-bar .filter-chip.active{border-color:var(--blue);color:var(--blue);background:rgba(59,130,246,.08)}
 .compare-btn{margin-left:auto}
+.gcp-check-grid{display:flex;flex-wrap:wrap;gap:8px 14px;align-items:center;margin-top:6px;padding:10px 12px;background:var(--bg-1);border:1px solid var(--border);border-radius:var(--radius-sm)}
+.gcp-check-grid label{display:inline-flex;align-items:center;gap:6px;font-size:12px;color:var(--text-2);cursor:pointer;user-select:none;font-weight:500}
+.gcp-check-grid input[type=checkbox]{width:14px;height:14px;accent-color:var(--blue);cursor:pointer}
+.gcp-gpu-matrix{width:100%;max-width:520px;border-collapse:collapse;font-size:12px;margin-top:8px;background:var(--bg-1);border:1px solid var(--border);border-radius:var(--radius-sm);overflow:hidden}
+.gcp-gpu-matrix th,.gcp-gpu-matrix td{padding:8px 10px;border-bottom:1px solid var(--border);text-align:center}
+.gcp-gpu-matrix th{background:var(--bg-2);color:var(--text-3);font-size:10px;font-weight:700;text-transform:uppercase;letter-spacing:.4px}
+.gcp-gpu-matrix td:first-child{text-align:left;font-weight:600;color:var(--text-2);white-space:nowrap}
+.gcp-gpu-matrix tr:last-child td{border-bottom:none}
+.gcp-gpu-matrix input[type=radio]{width:15px;height:15px;accent-color:var(--blue);cursor:pointer;vertical-align:middle}
 
 /* ── Buttons ── */
 .btn{padding:8px 20px;font-size:12px;font-weight:700;border:none;border-radius:var(--radius-sm);cursor:pointer;
@@ -231,6 +240,7 @@ body{font-family:-apple-system,'Segoe UI',system-ui,sans-serif;background:var(--
 .form-group label{font-size:10px;color:var(--text-3);text-transform:uppercase;letter-spacing:.5px;font-weight:700}
 .form-group input,.form-group select{padding:8px 12px;background:var(--bg-0);border:1px solid var(--border);
   border-radius:var(--radius-sm);color:var(--text-1);font-size:12px;font-family:inherit}
+.form-group input.gcp-wide{max-width:100%}
 .form-group input:focus,.form-group select:focus{outline:none;border-color:var(--blue)}
 .log-terminal{background:var(--bg-0);border:1px solid var(--border);border-radius:var(--radius);padding:16px;
   font-family:'SF Mono','Fira Code','Cascadia Code',monospace;font-size:11px;line-height:1.7;
@@ -477,6 +487,8 @@ body{font-family:-apple-system,'Segoe UI',system-ui,sans-serif;background:var(--
       <div class="form-group" id="fg-seeds"><label>Seeds</label><input id="train-seeds" value="42"></div>
       <div class="form-group" id="fg-folds"><label>Folds</label><input id="train-folds" type="number" value="5" min="2" max="10"></div>
       <div class="form-group"><label>Epochs</label><input id="train-epochs" type="number" value="100" min="10" max="1000"></div>
+      <div class="form-group"><label>Data subset (%)</label><input id="train-subset-percent" type="number" value="100" min="1" max="100" step="1" title="Random share of labeled sheets after filters (100 = all). Smaller = faster."></div>
+      <div class="form-group"><label>Subset seed</label><input id="train-subset-seed" type="number" value="0" min="0" step="1" title="Reproducible random pick when subset &lt; 100%."></div>
     </div>
     <div style="display:flex;gap:12px;align-items:center">
       <button class="btn btn-primary" id="start-btn" onclick="startTraining()">
@@ -486,6 +498,80 @@ body{font-family:-apple-system,'Segoe UI',system-ui,sans-serif;background:var(--
     </div>
   </div>
 
+  <div class="card" id="gcp-vertex-card">
+    <div class="card-header">
+      <div class="card-title">Vertex AI (GCP)</div>
+    </div>
+    <p style="font-size:12px;color:var(--text-3);margin-bottom:14px;line-height:1.5">
+      Obraz Dockera <strong>nie</strong> siedzi „na Macu” jako usługa: budujesz go lokalnie
+      (<code style="font-size:11px">docker build -f Dockerfile.training …</code>),
+      <strong>tagujesz</strong> i <strong>wypychasz</strong> do Artifact Registry — stamtąd Vertex pobiera obraz przy starcie joba.
+      <strong>Typ GPU (T4 / L4 / …)</strong> nie wymaga przebudowy obrazu — ten sam tag w Registry; zmieniasz tylko maszynę Vertex.
+      Tu wysyłasz tylko żądania uruchomienia jobów (wymaga <code style="font-size:11px">gcloud auth application-default login</code>
+      lub konta serwisowego z uprawnieniami Vertex).
+    </p>
+    <div class="form-grid">
+      <div class="form-group"><label>Project ID</label><input id="gcp-project" value="acoustic-atom-386613" autocomplete="off"></div>
+      <div class="form-group"><label>Region</label><input id="gcp-region" value="europe-west1" autocomplete="off"></div>
+      <div class="form-group" style="grid-column:1/-1"><label>GCS data URI</label><input id="gcp-gcs-data" class="gcp-wide" value="gs://excel-chunker/data" autocomplete="off"></div>
+      <div class="form-group" style="grid-column:1/-1"><label>Training image (Artifact Registry)</label><input id="gcp-image" class="gcp-wide" value="europe-west1-docker.pkg.dev/acoustic-atom-386613/excel-chunker/excel-chunker-train:latest" autocomplete="off"></div>
+      <div class="form-group" style="grid-column:1/-1">
+        <label>Architectures</label>
+        <p style="font-size:11px;color:var(--text-4);margin:0 0 8px">Zaznacz, które modele wysłać na Vertex (osobno od pola „Launch Training” powyżej — tamtego nie nadpisujemy przy każdym wejściu na tę stronę).</p>
+        <div class="gcp-check-grid" id="gcp-arch-grid" aria-label="Architectures"></div>
+        <button type="button" class="btn btn-secondary btn-sm" style="margin-top:8px;font-size:11px" onclick="syncGcpArchFromTrainField()">Skopiuj architektury z pola Launch Training</button>
+      </div>
+      <div class="form-group" style="grid-column:1/-1">
+        <label>Seeds</label>
+        <div class="gcp-check-grid" id="gcp-seed-grid" aria-label="Seeds"></div>
+      </div>
+      <div class="form-group"><label>Folds</label><input id="gcp-folds" type="number" min="2" max="10" value="5"></div>
+      <div class="form-group"><label>Epochs</label><input id="gcp-epochs" type="number" min="2" max="1000" value="100"></div>
+      <div class="form-group"><label>Data subset (%)</label><input id="gcp-subset-percent" type="number" value="100" min="1" max="100" step="1" title="Same as local Train card — random share of labeled sheets on the worker."></div>
+      <div class="form-group" style="grid-column:1/-1">
+        <label>GPU na architekturę (Vertex)</label>
+        <p style="font-size:11px;color:var(--text-4);margin:6px 0 0">Wybierz <strong>jedną</strong> kolumnę (T4 / L4 / A100) dla każdej architektury. Dotyczy tylko zaznaczonych wyżej „Architectures” przy starcie jobów.</p>
+        <table class="gcp-gpu-matrix" aria-label="GPU per architecture">
+          <thead><tr><th>Architektura</th><th>T4</th><th>L4</th><th>A100</th></tr></thead>
+          <tbody>
+            <tr><td>GAT</td><td><input type="radio" name="gcp-gpu-row-gat" value="t4" checked></td><td><input type="radio" name="gcp-gpu-row-gat" value="l4"></td><td><input type="radio" name="gcp-gpu-row-gat" value="a100"></td></tr>
+            <tr><td>GCN</td><td><input type="radio" name="gcp-gpu-row-gcn" value="t4" checked></td><td><input type="radio" name="gcp-gpu-row-gcn" value="l4"></td><td><input type="radio" name="gcp-gpu-row-gcn" value="a100"></td></tr>
+            <tr><td>MLP</td><td><input type="radio" name="gcp-gpu-row-mlp" value="t4" checked></td><td><input type="radio" name="gcp-gpu-row-mlp" value="l4"></td><td><input type="radio" name="gcp-gpu-row-mlp" value="a100"></td></tr>
+            <tr><td>Adj. transformer</td><td><input type="radio" name="gcp-gpu-row-adj_transformer" value="t4"></td><td><input type="radio" name="gcp-gpu-row-adj_transformer" value="l4" checked></td><td><input type="radio" name="gcp-gpu-row-adj_transformer" value="a100"></td></tr>
+            <tr><td>Spatial edge</td><td><input type="radio" name="gcp-gpu-row-spatial_edge_transformer" value="t4"></td><td><input type="radio" name="gcp-gpu-row-spatial_edge_transformer" value="l4" checked></td><td><input type="radio" name="gcp-gpu-row-spatial_edge_transformer" value="a100"></td></tr>
+            <tr><td>Dual-modality</td><td><input type="radio" name="gcp-gpu-row-dual_modality_gnn" value="t4"></td><td><input type="radio" name="gcp-gpu-row-dual_modality_gnn" value="l4" checked></td><td><input type="radio" name="gcp-gpu-row-dual_modality_gnn" value="a100"></td></tr>
+          </tbody>
+        </table>
+        <p style="font-size:11px;color:var(--text-4);margin-top:8px">A100 wymaga osobnej quota (np. Custom model training A100). Maszyna dopasowywana automatycznie (np. <code style="font-size:10px">a2-highgpu-1g</code>).</p>
+      </div>
+      <div class="form-group"><label>GPU count (na job)</label><input id="gcp-accel-count" type="number" min="1" max="8" value="1" title="Zwykle 1 na job"></div>
+      <div class="form-group" style="grid-column:1/-1">
+        <div class="gcp-check-grid" style="margin-top:0">
+          <label title="SPOT = preemptible VM — tańsze, mogą przerywać job. Użyj gdy masz quota preemptible (np. P100 / L4).">
+            <input type="checkbox" id="gcp-vertex-spot"> Spot (preemptible)
+          </label>
+        </div>
+      </div>
+    </div>
+    <div style="display:flex;gap:12px;align-items:center;flex-wrap:wrap;margin-top:8px">
+      <button type="button" class="btn btn-primary" id="gcp-launch-btn" onclick="launchGcpVertex()">Uruchom joby na Vertex</button>
+      <span id="gcp-launch-msg" style="font-size:12px;color:var(--text-3)"></span>
+    </div>
+    <p style="font-size:11px;color:var(--text-4);margin-top:10px">Anulowanie: Vertex → Custom jobs → job → Cancel, albo <code style="font-size:10px">gcloud ai custom-jobs cancel JOB_ID --region=REGION --project=...</code>.</p>
+    <div id="gcp-docker-build-block" style="display:none;margin-top:16px;padding-top:16px;border-top:1px solid var(--border)">
+      <p style="font-size:12px;color:var(--text-3);margin-bottom:10px">
+        <strong>Build &amp; push obrazu</strong> — tylko gdy aplikacja działa <strong>lokalnie</strong> na maszynie z Dockerem i ustawisz
+        <code style="font-size:11px">ENABLE_DOCKER_BUILD_FROM_DASHBOARD=1</code>. Wymaga wcześniej
+        <code style="font-size:11px">gcloud auth configure-docker REGION-docker.pkg.dev</code>.
+      </p>
+      <div style="display:flex;gap:12px;align-items:center;flex-wrap:wrap">
+        <button type="button" class="btn btn-secondary" id="gcp-docker-build-btn" onclick="startGcpDockerBuild()">Zbuduj i wypchnij obraz</button>
+        <span id="gcp-docker-build-msg" style="font-size:12px;color:var(--text-3)"></span>
+      </div>
+      <pre class="log-terminal compact" id="gcp-docker-log" style="margin-top:10px;max-height:200px;display:none"></pre>
+    </div>
+  </div>
+
   <div class="card" id="train-progress-card-local" style="display:none">
     <div class="card-header">
       <div class="card-title">Run progress (this machine)</div>
@@ -508,9 +594,10 @@ body{font-family:-apple-system,'Segoe UI',system-ui,sans-serif;background:var(--
       <span id="suite-gcp-meta" style="font-size:11px;color:var(--text-4)"></span>
     </div>
     <p style="font-size:12px;color:var(--text-3);margin-bottom:12px">
-      Live view of <code style="font-size:11px">experiments/*/progress.json</code> on GCS. Workers must set
-      <code style="font-size:11px">EXCEL_CHUNKER_DATA_ROOT</code> so each job writes under your bucket; this app reads
-      <code style="font-size:11px">EXCEL_CHUNKER_GCS_EXPERIMENTS_URI</code>.
+      Live view of <code style="font-size:11px">experiments/*/progress.json</code> on GCS (page auto-refreshes ~every 2s). Workers must set
+      <code style="font-size:11px">EXCEL_CHUNKER_DATA_ROOT</code> so each job writes under your bucket; set
+      <code style="font-size:11px">EXCEL_CHUNKER_GCS_EXPERIMENTS_URI</code> (or <code style="font-size:11px">GCP_GCS_DATA_URI</code>) so this card is enabled.
+      <strong style="color:var(--amber)">Re-run same arch/seed/fold?</strong> If <code style="font-size:11px">results.json</code> already exists on GCS, workers skip training — progress may flip to <code style="font-size:11px">skipped</code> with old metrics. Remove those run folders on GCS or use different seeds/folds to force a full retrain.
     </p>
     <div id="suite-gcp-error" style="display:none;font-size:12px;color:var(--red);margin-bottom:10px"></div>
     <div class="suite-overall">
@@ -531,7 +618,9 @@ body{font-family:-apple-system,'Segoe UI',system-ui,sans-serif;background:var(--
       <div class="card-title">Embeddings</div>
     </div>
     <p style="font-size:12px;color:var(--text-3);margin-bottom:12px">
-      Generate E5 text embeddings for all labeled files. Already-embedded files are skipped.
+      Text embeddings (E5) are generated <strong>automatically before each training run</strong> (local and Vertex)
+      for any labeled JSON that is missing or newer than its <code style="font-size:11px">*_embeddings.npz</code>.
+      Use the button below only if you want to run a full pass manually (e.g. after bulk label edits).
     </p>
     <div style="display:flex;gap:12px;align-items:center">
       <button class="btn btn-secondary btn-sm" id="embed-btn" onclick="doEmbed()">Generate Embeddings</button>
@@ -553,6 +642,7 @@ let selectedRunIds = new Set();
 let currentSort = {col: 'macro_f1', dir: 'desc'};
 let archFilter = 'all';
 let pollTimer = null;
+let dockerBuildPollTimer = null;
 
 const LABEL_COLORS = {
   value:'#3b82f6', attribute:'#22c55e', aggregation:'#f59e0b', metadata:'#a855f7',
@@ -568,6 +658,79 @@ const ARCH_COLORS = {gat:'#3b82f6', gcn:'#a855f7', mlp:'#f59e0b', transformer:'#
   naive_t0:'#64748b', naive_t1:'#94a3b8', naive_t2:'#cbd5e1'};
 const METRIC_COLORS = {accuracy:'#3b82f6', macro_f1:'#34d399', chunk_f1:'#f59e0b', ari:'#a855f7'};
 
+/** Vertex launch: same arch set as local default train-arch; keep in sync with train_compare. */
+const GCP_ARCH_CHOICES = [
+  ['gat', 'GAT'],
+  ['gcn', 'GCN'],
+  ['mlp', 'MLP'],
+  ['adj_transformer', 'Adj. transformer'],
+  ['spatial_edge_transformer', 'Spatial edge'],
+  ['dual_modality_gnn', 'Dual-modality GNN'],
+];
+const GCP_SEED_CHOICES = [42, 2137, 10042010];
+
+function _ensureGcpCheckboxGrids() {
+  const ag = document.getElementById('gcp-arch-grid');
+  const sg = document.getElementById('gcp-seed-grid');
+  if (ag && !ag.dataset.built) {
+    ag.innerHTML = GCP_ARCH_CHOICES.map(([id, label]) =>
+      `<label><input type="checkbox" id="gcp-arch-${id}" data-arch="${id}"> ${esc(label)}</label>`
+    ).join('');
+    ag.dataset.built = '1';
+  }
+  if (sg && !sg.dataset.built) {
+    sg.innerHTML = GCP_SEED_CHOICES.map(s =>
+      `<label><input type="checkbox" id="gcp-seed-${s}" data-seed="${s}"> ${s}</label>`
+    ).join('');
+    sg.dataset.built = '1';
+  }
+}
+
+function gcpSelectedArchitectures() {
+  return GCP_ARCH_CHOICES.map(([id]) => id).filter(id => {
+    const el = document.getElementById(`gcp-arch-${id}`);
+    return el && el.checked;
+  });
+}
+
+function gcpSelectedSeeds() {
+  return GCP_SEED_CHOICES.filter(s => {
+    const el = document.getElementById(`gcp-seed-${s}`);
+    return el && el.checked;
+  });
+}
+
+function _applyGcpArchFromString(commaList) {
+  const want = new Set((commaList || '').split(',').map(s => s.trim()).filter(Boolean));
+  GCP_ARCH_CHOICES.forEach(([id]) => {
+    const el = document.getElementById(`gcp-arch-${id}`);
+    if (el) el.checked = want.has(id);
+  });
+  if (want.size === 0) {
+    const gat = document.getElementById('gcp-arch-gat');
+    if (gat) gat.checked = true;
+  }
+}
+
+/** Jawna synchronizacja checkboxów Vertex z polem #train-arch (nie wywołuj automatycznie przy każdym wejściu na Train). */
+window.syncGcpArchFromTrainField = function() {
+  _ensureGcpCheckboxGrids();
+  const v = document.getElementById('train-arch') && document.getElementById('train-arch').value;
+  _applyGcpArchFromString(v || '');
+};
+
+function _applyGcpSeedsFromString(commaList) {
+  const nums = (commaList || '').split(',').map(s => parseInt(s.trim(), 10)).filter(n => !isNaN(n));
+  GCP_SEED_CHOICES.forEach(s => {
+    const el = document.getElementById(`gcp-seed-${s}`);
+    if (el) el.checked = nums.includes(s);
+  });
+  if (!GCP_SEED_CHOICES.some(s => document.getElementById(`gcp-seed-${s}`)?.checked)) {
+    const el = document.getElementById('gcp-seed-42');
+    if (el) el.checked = true;
+  }
+}
+
 /* ──────────────────────────── API ──────────────────────────── */
 function headers() {
   const h = {'Content-Type':'application/json'};
@@ -596,6 +759,11 @@ window.navigate = function(page) {
   if (page === 'experiments') loadExperiments();
   if (page === 'models') loadModels();
   if (page === 'compare') renderCompare();
+  if (page === 'train') {
+    loadGcpTrainDefaults();
+    if (!pollTimer) startPolling();
+    /* Nie wywołuj tutaj _applyGcpArchFromString — nadpisywałoby checkboxy Vertex przy każdym powrocie na Train. */
+  }
 };
 document.querySelectorAll('.nav-item').forEach(n => n.addEventListener('click', () => navigate(n.dataset.page)));
 
@@ -1185,12 +1353,15 @@ document.getElementById('train-mode').addEventListener('change', e => {
 });
 
 window.startTraining = async function() {
+  const pct = parseFloat(document.getElementById('train-subset-percent').value);
   const body = {
     mode: document.getElementById('train-mode').value,
     architectures: document.getElementById('train-arch').value,
     seeds: document.getElementById('train-seeds').value,
     folds: parseInt(document.getElementById('train-folds').value),
     epochs: parseInt(document.getElementById('train-epochs').value),
+    subset_percent: Number.isFinite(pct) ? Math.min(100, Math.max(1, pct)) : 100,
+    subset_seed: parseInt(document.getElementById('train-subset-seed').value, 10) || 0,
   };
   const res = await api('/api/train/start', {method:'POST', body:JSON.stringify(body)});
   if (res && res.started) {
@@ -1201,6 +1372,162 @@ window.startTraining = async function() {
   }
 };
 
+/** Same defaults as ``config.py`` (Vertex dashboard; local use). */
+const GCP_FORM_DEFAULTS = {
+  project_id: 'acoustic-atom-386613',
+  region: 'europe-west1',
+  gcs_data_uri: 'gs://excel-chunker/data',
+  image_uri: 'europe-west1-docker.pkg.dev/acoustic-atom-386613/excel-chunker/excel-chunker-train:latest',
+  accelerator_count: 1,
+  vertex_spot: false,
+};
+
+/** Vertex accelerator enum z wiersza tabeli GPU (radio: t4 | l4 | a100). */
+function gcpGpuVertexEnumForArch(archId) {
+  const sel = document.querySelector(`input[name="gcp-gpu-row-${archId}"]:checked`);
+  const v = (sel && sel.value) || 't4';
+  if (v === 'l4') return 'NVIDIA_L4';
+  if (v === 'a100') return 'NVIDIA_TESLA_A100';
+  return 'NVIDIA_TESLA_T4';
+}
+
+async function loadGcpTrainDefaults() {
+  _ensureGcpCheckboxGrids();
+  const raw = await api('/api/train/gcp-defaults');
+  const d = raw && !raw.detail ? raw : {};
+  const set = (id, v) => {
+    const el = document.getElementById(id);
+    if (el && v != null && v !== '') el.value = v;
+  };
+  set('gcp-project', (d.project_id || GCP_FORM_DEFAULTS.project_id).trim());
+  set('gcp-region', (d.region || GCP_FORM_DEFAULTS.region).trim());
+  set('gcp-gcs-data', (d.gcs_data_uri || GCP_FORM_DEFAULTS.gcs_data_uri).trim());
+  set('gcp-image', (d.image_uri || GCP_FORM_DEFAULTS.image_uri).trim());
+  set('gcp-accel-count', String(d.accelerator_count != null ? d.accelerator_count : GCP_FORM_DEFAULTS.accelerator_count));
+  const spotEl = document.getElementById('gcp-vertex-spot');
+  if (spotEl) spotEl.checked = d.vertex_spot === true || d.vertex_spot === 1;
+  const dock = document.getElementById('gcp-docker-build-block');
+  if (dock) dock.style.display = d.docker_build_from_dashboard ? 'block' : 'none';
+  _applyGcpSeedsFromString(document.getElementById('train-seeds').value);
+  document.getElementById('gcp-folds').value = document.getElementById('train-folds').value;
+  document.getElementById('gcp-epochs').value = document.getElementById('train-epochs').value;
+  const gsp = document.getElementById('gcp-subset-percent');
+  const tsp = document.getElementById('train-subset-percent');
+  if (gsp && tsp) gsp.value = tsp.value;
+}
+
+async function pollDockerBuildStatus() {
+  const s = await api('/api/train/gcp-image-build/status');
+  const logEl = document.getElementById('gcp-docker-log');
+  const msg = document.getElementById('gcp-docker-build-msg');
+  const btn = document.getElementById('gcp-docker-build-btn');
+  if (!s || s.detail) return;
+  if (s.log && s.log.length) {
+    logEl.style.display = 'block';
+    logEl.textContent = s.log.join('\n');
+    logEl.scrollTop = logEl.scrollHeight;
+  }
+  if (!s.running && s.ok !== null && s.ok !== undefined) {
+    if (dockerBuildPollTimer) { clearInterval(dockerBuildPollTimer); dockerBuildPollTimer = null; }
+    btn.disabled = false;
+    if (s.ok) {
+      msg.textContent = 'Gotowe — obraz w Artifact Registry.';
+      msg.style.color = 'var(--green)';
+    } else {
+      msg.textContent = s.error || 'Błąd';
+      msg.style.color = 'var(--red)';
+    }
+  }
+}
+
+window.startGcpDockerBuild = async function() {
+  const msg = document.getElementById('gcp-docker-build-msg');
+  const btn = document.getElementById('gcp-docker-build-btn');
+  const logEl = document.getElementById('gcp-docker-log');
+  msg.textContent = '';
+  logEl.style.display = 'none';
+  logEl.textContent = '';
+  btn.disabled = true;
+  const res = await api('/api/train/gcp-image-build', { method: 'POST', body: '{}' });
+  if (!res) {
+    btn.disabled = false;
+    msg.textContent = 'Nie uruchomiono (403: dodaj ENABLE_DOCKER_BUILD_FROM_DASHBOARD=1 do .env i zrestartuj uvicorn; albo brak uprawnień).';
+    msg.style.color = 'var(--red)';
+    return;
+  }
+  if (res.detail) {
+    btn.disabled = false;
+    msg.textContent = _formatVertexErr(res);
+    msg.style.color = 'var(--red)';
+    return;
+  }
+  msg.textContent = 'Trwa build… (log poniżej)';
+  msg.style.color = 'var(--blue)';
+  if (dockerBuildPollTimer) clearInterval(dockerBuildPollTimer);
+  dockerBuildPollTimer = setInterval(pollDockerBuildStatus, 1500);
+  pollDockerBuildStatus();
+};
+
+function _formatVertexErr(res) {
+  if (!res) return 'Brak odpowiedzi';
+  const d = res.detail;
+  if (typeof d === 'string') return d;
+  if (Array.isArray(d)) return d.map(x => (x.msg != null ? x.msg : JSON.stringify(x))).join('; ');
+  return 'Błąd';
+}
+
+window.launchGcpVertex = async function() {
+  const msg = document.getElementById('gcp-launch-msg');
+  const btn = document.getElementById('gcp-launch-btn');
+  _ensureGcpCheckboxGrids();
+  const archs = gcpSelectedArchitectures();
+  const seeds = gcpSelectedSeeds();
+  if (!archs.length) {
+    msg.textContent = 'Zaznacz co najmniej jedną architekturę.';
+    msg.style.color = 'var(--amber)';
+    return;
+  }
+  if (!seeds.length) {
+    msg.textContent = 'Zaznacz co najmniej jeden seed (42, 2137 lub 10042010).';
+    msg.style.color = 'var(--amber)';
+    return;
+  }
+  const gPct = parseFloat(document.getElementById('gcp-subset-percent').value);
+  const archAccelerators = {};
+  archs.forEach(arch => {
+    archAccelerators[arch] = gcpGpuVertexEnumForArch(arch);
+  });
+  const firstAccel = archAccelerators[archs[0]] || 'NVIDIA_TESLA_T4';
+  const body = {
+    architectures: archs.join(','),
+    seeds: seeds.join(','),
+    folds: parseInt(document.getElementById('gcp-folds').value, 10),
+    epochs: parseInt(document.getElementById('gcp-epochs').value, 10),
+    subset_percent: Number.isFinite(gPct) ? Math.min(100, Math.max(1, gPct)) : 100,
+    project_id: document.getElementById('gcp-project').value.trim() || null,
+    region: document.getElementById('gcp-region').value.trim() || null,
+    image_uri: document.getElementById('gcp-image').value.trim() || null,
+    gcs_data_uri: document.getElementById('gcp-gcs-data').value.trim() || null,
+    machine_type: 'n1-standard-8',
+    accelerator_type: firstAccel,
+    accelerator_count: parseInt(document.getElementById('gcp-accel-count').value, 10) || 1,
+    vertex_spot: !!document.getElementById('gcp-vertex-spot')?.checked,
+    arch_accelerators: archAccelerators,
+  };
+  msg.textContent = 'Wysyłanie…';
+  msg.style.color = 'var(--blue)';
+  btn.disabled = true;
+  const res = await api('/api/train/gcp-launch', { method: 'POST', body: JSON.stringify(body) });
+  btn.disabled = false;
+  if (res && res.status === 'ok') {
+    msg.textContent = `Wysłano ${res.submitted_count} job(ów). Sprawdź Vertex → Custom jobs.`;
+    msg.style.color = 'var(--green)';
+  } else {
+    msg.textContent = _formatVertexErr(res);
+    msg.style.color = 'var(--red)';
+  }
+};
+
 function startPolling() {
   if (pollTimer) clearInterval(pollTimer);
   pollTimer = setInterval(pollStatus, 2000);
@@ -1253,9 +1580,12 @@ function renderSuiteProgress(suite, mode) {
       ? `${src ? src + ' · ' : ''}Overall: ${done} / ${total} units (${pct}%)`
       : `${src ? src + ' · ' : ''}Last: ${done} / ${total} units (${pct}%)`;
   document.getElementById(ids.bar).style.width = pct + '%';
+  const gcpRead = suite.gcs_uri && suite.refreshed_at
+    ? ` · last GCS read ${suite.refreshed_at.slice(0, 19).replace('T', ' ')} UTC`
+    : '';
   meta.textContent = suite.started_at
     ? (suite.finished_at ? `${suite.started_at.slice(11,19)} → ${suite.finished_at.slice(11,19)} UTC` : 'Running…')
-    : (suite.gcs_uri ? 'Refreshed every ~5s (server cache)' : '');
+    : (suite.gcs_uri ? `Poll ~2s · server cache ~2s${gcpRead}` : '');
 
   const units = suite.units || [];
   const host = document.getElementById(ids.units);
@@ -1316,7 +1646,7 @@ function renderGcpProgressPanel(data) {
 }
 
 async function pollStatus() {
-  const data = await api('/api/train/status');
+  const data = await api('/api/train/status', { cache: 'no-store' });
   if (!data) return;
   const badge = document.getElementById('train-badge');
   const globalBadge = document.getElementById('global-status');
@@ -1334,12 +1664,16 @@ async function pollStatus() {
     if (data.result.status === 'completed') {
       badge.className = 'status-pill done'; badge.textContent = 'Completed';
       globalBadge.className = 'status-pill done'; globalBadge.textContent = 'Done';
+    } else if (data.result.status === 'failed') {
+      const n = data.result.failed_units != null ? ` (${data.result.failed_units})` : '';
+      badge.className = 'status-pill idle'; badge.textContent = 'Failed units' + n;
+      globalBadge.className = 'status-pill idle'; globalBadge.textContent = 'Some folds failed';
     } else {
       badge.className = 'status-pill idle'; badge.textContent = 'Error';
       globalBadge.className = 'status-pill idle'; globalBadge.textContent = 'Error';
     }
     startBtn.disabled = false;
-    if (pollTimer) { clearInterval(pollTimer); pollTimer = null; }
+    /* Keep polling: Vertex/GCS progress and log tail should update without a local run. */
   } else {
     badge.className = 'status-pill idle'; badge.textContent = 'Idle';
     globalBadge.className = 'status-pill idle'; globalBadge.textContent = 'Idle';
@@ -1348,7 +1682,7 @@ async function pollStatus() {
 
   const lines = data.log || [];
   logBox.innerHTML = lines.map(l => {
-    if (l.startsWith('ERROR')) return `<span class="l-err">${esc(l)}</span>`;
+    if (l.startsWith('ERROR') || l.includes('failed unit')) return `<span class="l-err">${esc(l)}</span>`;
     if (l.includes('completed') || l.includes('saved') || l.includes('Best')) return `<span class="l-ok">${esc(l)}</span>`;
     if (l.startsWith('Training') || l.startsWith('===')) return `<span class="l-info">${esc(l)}</span>`;
     return `<span class="l-dim">${esc(l)}</span>`;
@@ -1392,14 +1726,15 @@ window.doEmbed = async function() {
 /* ──────────────────────────── Init ──────────────────────────── */
 async function init() {
   loadOverview();
+  _ensureGcpCheckboxGrids();
+  _applyGcpArchFromString(document.getElementById('train-arch').value);
+  await loadGcpTrainDefaults();
   const status = await api('/api/train/status');
   if (status && status.running) {
     navigate('train');
-    startPolling();
-  } else {
-    // Last run may already be finished; still paint log + badges (otherwise refresh shows empty).
-    pollStatus();
   }
+  /* Always poll: local training log, badges, and GCS progress (when EXCEL_CHUNKER_GCS_EXPERIMENTS_URI / data URI is set). */
+  startPolling();
 }
 
 init();

train_compare.py

@@ -19,6 +19,7 @@ from __future__ import annotations
 import argparse
 import json
 import random
+import sys
 import time
 from collections import defaultdict
 from datetime import datetime, timezone
@@ -700,13 +701,18 @@ def run_experiments(
     fold_index: int | None = None,
     sheet_list_file: Path | str | None = None,
     max_graphs: int | None = None,
+    subset_percent: float | None = None,
     subset_seed: int = 0,
     progress: Any = None,
-) -> dict:
+) -> tuple[dict, int]:
     """Run the full experiment suite.
 
     If *progress* is a ``training_progress.SuiteProgress`` instance, it is updated
     for each (architecture, seed, fold) unit and each training epoch (dashboard).
+
+    Returns ``(results_by_arch, failed_unit_count)``. Failed units are those where
+    ``_train_one`` raised (e.g. CUDA OOM); the process used to exit 0 anyway — callers
+    should check ``failed_unit_count`` or rely on CLI ``sys.exit(1)``.
     """
     device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
     print(f"Device: {device}")
@@ -738,6 +744,16 @@ def run_experiments(
         }
         print(f"  Sheet allowlist: {len(sheet_allowlist)} stems from {sheet_list_file}")
 
+    try:
+        from embed_text import ensure_embeddings_for_labeled
+
+        ensure_embeddings_for_labeled()
+    except ImportError:
+        print(
+            "  Warning: embed_text not available; training uses zero vectors where "
+            "*_embeddings.npz is missing."
+        )
+
     graphs = load_all_graphs(
         max_nodes=max_nodes,
         use_row_col_edges=use_row_col_edges,
@@ -745,6 +761,7 @@ def run_experiments(
         graph_cfg=graph_cfg,
         sheet_allowlist=sheet_allowlist,
         max_graphs=max_graphs,
+        subset_percent=subset_percent,
         subset_seed=subset_seed,
     )
     loaded_splits: dict | None = None
@@ -764,7 +781,7 @@ def run_experiments(
         if progress is not None:
             progress.reset_for_new_run([])
             progress.mark_suite_end(datetime.now(timezone.utc).isoformat())
-        return {}
+        return {}, 0
 
     in_dim = graphs[0].x.size(1)
     class_weights = compute_class_weights(graphs)
@@ -825,7 +842,7 @@ def _run_experiment_suite_inner(
     class_weights: torch.Tensor,
     loaded_splits: dict | None,
     fold_index: int | None,
-) -> dict:
+) -> tuple[dict, int]:
     """Body of ``run_experiments`` after data load (factored for ``try/finally`` on dashboard progress)."""
     device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
 
@@ -834,6 +851,7 @@ def _run_experiment_suite_inner(
             SpatialEdgeTransformerModel.precompute_spatial(g)
 
     all_results: dict[str, list] = defaultdict(list)
+    failed_units = 0
 
     for seed in seeds:
         random.seed(seed)
@@ -967,6 +985,7 @@ def _run_experiment_suite_inner(
                         progress_epoch_callback=_epoch_hook,
                     )
                 except Exception as e:
+                    failed_units += 1
                     print(f"  ERROR in {run_tag}: {e}")
                     print(f"  Skipping {run_tag} and continuing...")
                     if progress is not None:
@@ -1250,7 +1269,12 @@ def _run_experiment_suite_inner(
                / "best_model.pt")
         if src.exists():
             import shutil
-            dst = config.MODELS_DIR / "best_model.pt"
+            # Unique name: parallel Vertex jobs share GCS (gcsfuse); copying to the same
+            # ``models/best_model.pt`` from two workers causes Errno 116 (stale file handle).
+            dst = (
+                config.MODELS_DIR
+                / f"best_model_{best['arch']}_s{best['seed']}_f{best['fold']}.pt"
+            )
             shutil.copy2(src, dst)
             print(f"\nBest model ({best['arch']} seed={best['seed']} "
                   f"fold={best['fold']}) promoted to {dst}")
@@ -1265,7 +1289,15 @@ def _run_experiment_suite_inner(
             }, f, indent=2, default=str)
         print(f"Ranking saved to {summary_path}")
 
-    return dict(all_results)
+    if failed_units:
+        print(f"\n{'!'*80}")
+        print(
+            f"FAILED: {failed_units} training unit(s) raised an exception "
+            f"(e.g. CUDA OOM). This run is incomplete."
+        )
+        print(f"{'!'*80}\n")
+
+    return dict(all_results), failed_units
 
 
 # ═══════════════════════════════════════════════════════════════════════════
@@ -1307,8 +1339,14 @@ def main():
                         help="File with one labeled JSON stem per line (# comments allowed)")
     parser.add_argument("--max-graphs", type=int, default=None,
                         help="After filters, randomly keep at most this many graphs")
+    parser.add_argument(
+        "--subset-percent",
+        type=float,
+        default=100.0,
+        help="Random fraction of candidate sheets before CV (1-100; 100 = all, same as dashboard)",
+    )
     parser.add_argument("--subset-seed", type=int, default=0,
-                        help="RNG seed for --max-graphs subsampling")
+                        help="RNG seed for --subset-percent / --max-graphs subsampling")
 
     # GraphConfig options
     parser.add_argument("--graph-config", type=str, default=None,
@@ -1349,7 +1387,7 @@ def main():
         if cli_overrides:
             graph_cfg = GraphConfig(**cli_overrides)
 
-    run_experiments(
+    _, failed = run_experiments(
         architectures=archs,
         seeds=seeds,
         n_folds=args.folds,
@@ -1369,8 +1407,11 @@ def main():
         fold_index=args.fold_index,
         sheet_list_file=args.sheet_list,
         max_graphs=args.max_graphs,
+        subset_percent=None if args.subset_percent >= 100.0 else args.subset_percent,
         subset_seed=args.subset_seed,
     )
+    if failed:
+        sys.exit(1)
 
 
 if __name__ == "__main__":

train_gnn.py

@@ -1811,6 +1811,7 @@ def load_all_graphs(max_nodes: int = 5000,
                     graph_cfg: GraphConfig | None = None,
                     sheet_allowlist: set[str] | None = None,
                     max_graphs: int | None = None,
+                    subset_percent: float | None = None,
                     subset_seed: int = 0) -> list[Data]:
     """Load all labeled JSONs as PyG Data objects.
 
@@ -1825,6 +1826,9 @@ def load_all_graphs(max_nodes: int = 5000,
 
     *max_graphs*: after other filters, randomly keep at most this many graphs
     (reproducible with *subset_seed*).
+
+    *subset_percent*: if set and ``< 100``, keep a random ``pct%`` of candidates
+    (after allowlist / RAG hold-out / max_nodes), then apply *max_graphs* if set.
     """
     held_out = _load_rag_held_out_ids() if exclude_rag_holdout else set()
     if held_out:
@@ -1849,6 +1853,20 @@ def load_all_graphs(max_nodes: int = 5000,
                 continue
         candidates.append(fp)
 
+    if subset_percent is not None and subset_percent < 100:
+        n_cand = len(candidates)
+        n_keep = max(1, int(round(n_cand * (subset_percent / 100.0))))
+        if n_keep < n_cand:
+            rng = np.random.RandomState(subset_seed)
+            perm = rng.permutation(n_cand)
+            sel_idx = perm[:n_keep]
+            candidates = [candidates[i] for i in sel_idx]
+            candidates.sort(key=lambda p: p.stem)
+            print(
+                f"  Subset: {n_keep}/{n_cand} graphs ({subset_percent:g}% of candidates, "
+                f"subset_seed={subset_seed})"
+            )
+
     if max_graphs is not None and len(candidates) > max_graphs:
         rng = np.random.RandomState(subset_seed)
         perm = rng.permutation(len(candidates))
@@ -2162,13 +2180,32 @@ def _next_model_version() -> str:
     return f"{last_num + 1:03d}"
 
 
-def train_model(num_epochs: int = 100, lr: float = 1e-3, skip_predict: bool = False):
+def train_model(
+    num_epochs: int = 100,
+    lr: float = 1e-3,
+    skip_predict: bool = False,
+    *,
+    subset_percent: float | None = None,
+    subset_seed: int = 0,
+):
     from datetime import datetime, timezone
 
     device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
     print(f"Using device: {device}")
 
-    graphs = load_all_graphs()
+    try:
+        from embed_text import ensure_embeddings_for_labeled
+
+        ensure_embeddings_for_labeled()
+    except ImportError:
+        print(
+            "Warning: embed_text not available; using zero vectors where *_embeddings.npz is missing."
+        )
+
+    graphs = load_all_graphs(
+        subset_percent=subset_percent,
+        subset_seed=subset_seed,
+    )
     if len(graphs) < 2:
         print(f"Need at least 2 labeled sheets, found {len(graphs)}. Label more data first.")
         return

vertex_launch.py

@@ -0,0 +1,153 @@
+"""Submit Vertex AI Custom Training jobs (shared by CLI and dashboard)."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# Default accelerator comes from config / env (often L4 + g2 once quota exists).
+_DEFAULT_ACCEL = "NVIDIA_L4"
+
+
+def resolve_machine_type_for_accelerator(machine_type: str, accelerator_type: str) -> str:
+    """Return a Vertex-compatible machine type for the given GPU.
+
+    L4 must use **G2** (e.g. ``g2-standard-8``). T4/P100 typically use **N1**.
+    If the pair is inconsistent, we pick a safe default for the GPU family.
+    """
+    mt = (machine_type or "").strip()
+    at = (accelerator_type or "").strip().upper()
+    if "L4" in at:
+        if not mt.startswith("g2-"):
+            return "g2-standard-8"
+        return mt
+    if "A100" in at:
+        if mt.startswith("a2-") or mt.startswith("a3-"):
+            return mt
+        # Single A100 — typical Vertex pairing (region/quota dependent).
+        return "a2-highgpu-1g"
+    if "T4" in at or "P100" in at:
+        if mt.startswith("g2-"):
+            return "n1-standard-8"
+        return mt or "n1-standard-8"
+    return mt or "n1-standard-8"
+
+
+def submit_vertex_training_jobs(
+    *,
+    project: str,
+    region: str,
+    image_uri: str,
+    gcs_data_uri: str,
+    architectures: list[str],
+    seeds: list[int],
+    folds: int,
+    epochs: int,
+    machine_type: str = "g2-standard-8",
+    accelerator_type: str = _DEFAULT_ACCEL,
+    accelerator_count: int = 1,
+    staging_bucket: str | None = None,
+    subset_percent: float = 100.0,
+    use_spot: bool = False,
+    arch_accelerator: dict[str, str] | None = None,
+) -> dict[str, Any]:
+    """One Custom Job per (seed, fold, arch). Returns counts and display names.
+
+    *arch_accelerator*: optional map ``architecture -> accelerator_type`` (e.g.
+    ``{"adj_transformer": "NVIDIA_L4", "gat": "NVIDIA_TESLA_T4"}``). Omitted archs
+    use *accelerator_type*. Machine type is resolved per job via
+    ``resolve_machine_type_for_accelerator``.
+
+    use_spot: if True, passes scheduling SPOT to ``job.run()`` (preemptible VM; often
+    P100 preemptible quota). If False, on-demand GPUs (e.g. ``CustomModelTrainingT4GPUsPerProjectPerRegion``).
+    """
+    from google.cloud import aiplatform
+    from google.cloud.aiplatform.compat.types import custom_job as gca_custom_job
+
+    staging = staging_bucket
+    if not staging:
+        uri = gcs_data_uri.rstrip("/")
+        if not uri.startswith("gs://"):
+            raise ValueError("gcs_data_uri must start with gs://")
+        rest = uri[5:]
+        bucket = rest.split("/", 1)[0]
+        staging = f"gs://{bucket}"
+
+    aiplatform.init(project=project, location=region, staging_bucket=staging)
+
+    if gcs_data_uri.startswith("gs://"):
+        without = gcs_data_uri[5:]
+        parts = without.split("/", 1)
+        bucket = parts[0]
+        prefix = parts[1] if len(parts) > 1 else ""
+        fuse_root = f"/gcs/{bucket}" + (f"/{prefix}" if prefix else "")
+    else:
+        fuse_root = gcs_data_uri
+
+    # Use CustomJob.submit(), not job.run(sync=False). The latter schedules _run() in a
+    # background thread and returns immediately, so the HTTP handler could finish (and report
+    # N submitted jobs) before all create_custom_job RPCs complete — or surface errors only
+    # on the Future. submit() runs create_custom_job synchronously (does not wait for training).
+    submitted: list[str] = []
+    job_resource_names: list[str] = []
+    for seed in seeds:
+        for fold_idx in range(folds):
+            for arch in architectures:
+                accel = accelerator_type
+                if arch_accelerator and arch in arch_accelerator:
+                    accel = (arch_accelerator[arch] or "").strip() or accelerator_type
+                mach = resolve_machine_type_for_accelerator(machine_type, accel)
+                display = f"train-{arch}-s{seed}-f{fold_idx}"
+                train_args = [
+                    "train_compare.py",
+                    "--architectures",
+                    arch,
+                    "--seeds",
+                    str(seed),
+                    "--folds",
+                    str(folds),
+                    "--epochs",
+                    str(epochs),
+                    "--fold-index",
+                    str(fold_idx),
+                ]
+                if subset_percent < 100.0:
+                    train_args.extend(["--subset-percent", str(subset_percent)])
+                env = [{"name": "EXCEL_CHUNKER_DATA_ROOT", "value": fuse_root}]
+                worker_pool = {
+                    "machine_spec": {
+                        "machine_type": mach,
+                        "accelerator_type": accel,
+                        "accelerator_count": accelerator_count,
+                    },
+                    "replica_count": 1,
+                    "container_spec": {
+                        "image_uri": image_uri,
+                        "command": ["python"],
+                        "args": train_args,
+                        "env": env,
+                    },
+                }
+                job = aiplatform.CustomJob(
+                    display_name=display[:128],
+                    worker_pool_specs=[worker_pool],
+                )
+                submit_kw: dict[str, Any] = {}
+                if use_spot:
+                    submit_kw["scheduling_strategy"] = gca_custom_job.Scheduling.Strategy.SPOT
+                job.submit(**submit_kw)
+                resource = getattr(job, "resource_name", None) or getattr(
+                    job, "name", None
+                )
+                if resource:
+                    job_resource_names.append(resource)
+                    logger.info("Vertex CustomJob created: %s (%s)", display, resource)
+                submitted.append(display)
+
+    return {
+        "submitted_count": len(submitted),
+        "jobs": submitted,
+        "job_resource_names": job_resource_names,
+    }