diff --git a/vero/src/vero/harbor/cli.py b/vero/src/vero/harbor/cli.py index e68ce4c..1433295 100644 --- a/vero/src/vero/harbor/cli.py +++ b/vero/src/vero/harbor/cli.py @@ -125,3 +125,26 @@ def finalize_cmd(token_file, output): out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(reward)) click.echo(json.dumps(reward, indent=2)) + + +@harbor.command("dashboard") +@click.option("--meta", "meta_path", type=click.Path(path_type=Path), default=None, + help="JSON with experiment questions, pre-registered bars, and a scorecard.") +@click.option("--jobs", "jobs_dirs", type=click.Path(path_type=Path), multiple=True, + help="Harbor jobs dir(s) to scan for reward.json finals. Repeatable.") +@click.option("--host", default="127.0.0.1", show_default=True, + help="Bind address. Ledger rows include hidden-split scores; keep it local.") +@click.option("--port", default=8300, show_default=True) +def dashboard_cmd(meta_path, jobs_dirs, host, port): + """Operator dashboard: watch running optimization trials live (host-side). + + Discovers eval-sidecar containers via docker, tails each trial's ledger + through the admin /experiments endpoint, scans jobs dirs for finals, and + serves a self-refreshing status page. Admin-side tool: run it where the + trials run, not inside a task container. + """ + from vero.harbor.dashboard import DashboardConfig, serve_dashboard + + serve_dashboard(DashboardConfig( + meta_path=meta_path, jobs_dirs=list(jobs_dirs), host=host, port=port, + )) diff --git a/vero/src/vero/harbor/dashboard.py b/vero/src/vero/harbor/dashboard.py new file mode 100644 index 0000000..533d789 --- /dev/null +++ b/vero/src/vero/harbor/dashboard.py @@ -0,0 +1,355 @@ +"""Local dashboard for watching Harbor optimization trials. + +``vero harbor dashboard`` serves a single-page UI (stdlib HTTP server, no new +dependencies) that answers, at a glance: which experiments are running, what +each one is doing right now, and how finished ones scored against their +pre-registered bars. + +Data sources, all local to the operator's machine: + +- ``docker ps`` discovers live eval-sidecar containers (one per running trial). +- Each sidecar's admin ``/experiments`` endpoint (via ``docker exec``, using the + in-container admin token) provides the live ledger: one row per recorded + eval (commit, split, score, error rate). This is the same admin-observability + surface the ``/experiments`` endpoint was added for; the dashboard never + reaches into agent volumes. +- ``/status`` (agent-facing, unauthenticated) provides remaining budget. +- Harbor jobs directories are scanned for ``reward.json`` (the trial's final). +- An optional ``--meta`` JSON file supplies display context the containers + cannot know: the experiment's question, pre-registered bars (baseline / + win bar / healthy ceiling), notes, and a scorecard of finished experiments. + +The server binds 127.0.0.1 by default: ledger rows include hidden-split scores, +so the page is for the operator's eyes, not the network's. +""" + +from __future__ import annotations + +import json +import logging +import subprocess +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +logger = logging.getLogger(__name__) + +SIDECAR_SUFFIX = "-eval-sidecar-1" + + +# --------------------------------------------------------------------------- +# Pure helpers (unit-tested; no docker, no filesystem) +# --------------------------------------------------------------------------- + + +def experiment_key(container_name: str) -> str | None: + """``__-eval-sidecar-1`` -> ```` (None if not a sidecar). + + Harbor composes container names as ``-eval-sidecar-1`` where the + project is ``__``; the task dir name is the stable + key an operator recognizes (e.g. ``gaia-exp11-task``). + """ + if not container_name.endswith(SIDECAR_SUFFIX): + return None + project = container_name[: -len(SIDECAR_SUFFIX)] + return project.split("__")[0] if "__" in project else project + + +def derive_phase(*, has_container: bool, n_rounds: int, final: dict | None) -> str: + """One word for where a trial is: building -> running -> done (or gone). + + ``gone`` = no live container and no reward.json: either torn down abnormally + or never started; the operator should look at the launch log. + """ + if final is not None: + return "done" + if has_container: + return "running" if n_rounds > 0 else "building" + return "gone" + + +def verdict(final: dict | None, bars: dict | None) -> str | None: + """PASSED / FAILED against the pre-registered win bar, when both are known.""" + if not final or not bars or "win_bar" not in bars: + return None + scores = [v for v in final.values() if isinstance(v, (int, float))] + if not scores: + return None + return "PASSED" if scores[0] > bars["win_bar"] else "FAILED" + + +def normalize_rounds(payload: dict | None) -> list[dict]: + """/experiments payload -> display rows (newest last), tolerant of Nones.""" + rows = [] + for e in (payload or {}).get("experiments", []): + rows.append( + { + "commit": str(e.get("commit") or "")[:10], + "split": e.get("split"), + "score": e.get("mean_score"), + "error_rate": e.get("error_rate"), + "created_at": e.get("created_at"), + } + ) + return rows + + +def merge_status( + *, + meta: dict, + sidecars: dict[str, dict], + finals: dict[str, dict], +) -> dict: + """Join meta-declared experiments with live containers and finished rewards. + + Experiments the meta file does not mention still appear (key only), so a + running trial is never invisible just because the operator forgot to list + it. Meta-only entries appear too, phase ``gone``, so a crashed launch is + conspicuous rather than silently absent. + """ + keys = list( + dict.fromkeys( + [e["key"] for e in meta.get("experiments", [])] + + list(sidecars) + + list(finals) + ) + ) + by_key = {e["key"]: e for e in meta.get("experiments", [])} + out = [] + for key in keys: + m = by_key.get(key, {}) + live = sidecars.get(key) + final = finals.get(key) + rounds = normalize_rounds(live.get("experiments") if live else None) + out.append( + { + "key": key, + "title": m.get("title", key), + "question": m.get("question"), + "bars": m.get("bars"), + "note": m.get("note"), + "phase": derive_phase( + has_container=live is not None, + n_rounds=len(rounds), + final=final, + ), + "rounds": rounds, + "budget": (live or {}).get("status", {}).get("splits"), + "final": final, + "verdict": verdict(final, m.get("bars")), + } + ) + return {"experiments": out, "history": meta.get("history", [])} + + +# --------------------------------------------------------------------------- +# Collectors (thin side-effecting shims around docker + the filesystem) +# --------------------------------------------------------------------------- + + +def _docker(*args: str, timeout: int = 15) -> str: + res = subprocess.run( + ["docker", *args], capture_output=True, text=True, timeout=timeout + ) + return res.stdout if res.returncode == 0 else "" + + +def discover_sidecars() -> dict[str, str]: + """{experiment key: container name} for every live eval sidecar.""" + out = {} + for name in _docker("ps", "--format", "{{.Names}}").splitlines(): + key = experiment_key(name.strip()) + if key: + out[key] = name.strip() + return out + + +_TOKEN_SNIPPET = ( + "import json;" + "print(open(json.load(open('/opt/serve.json'))['admin_token_path']).read().strip())" +) + + +def snapshot_sidecar(container: str) -> dict: + """Live ledger + budget for one sidecar, via docker exec (admin-side).""" + snap: dict = {} + token = _docker("exec", container, "python3", "-c", _TOKEN_SNIPPET).strip() + if token: + raw = _docker( + "exec", container, "curl", "-s", + "-H", f"Authorization: Bearer {token}", + "http://localhost:8000/experiments", + ) + try: + snap["experiments"] = json.loads(raw) + except json.JSONDecodeError: + pass + raw = _docker("exec", container, "curl", "-s", "http://localhost:8000/status") + try: + snap["status"] = json.loads(raw) + except json.JSONDecodeError: + snap["status"] = {} + return snap + + +def scan_finals(jobs_dirs: list[Path]) -> dict[str, dict]: + """{experiment key: reward dict} from every reward.json under the jobs dirs. + + The experiment key is recovered from the trial directory name harbor writes + (``__``), so finals keep matching after containers are gone. + """ + finals: dict[str, dict] = {} + for jd in jobs_dirs: + for rj in Path(jd).glob("**/reward.json"): + key = None + for part in rj.parts: + if "__" in part: + key = part.split("__")[0] + if key is None: + key = Path(jd).name + try: + finals[key] = json.loads(rj.read_text()) + except (OSError, json.JSONDecodeError): + continue + return finals + + +def collect(meta_path: Path | None, jobs_dirs: list[Path]) -> dict: + meta: dict = {} + if meta_path and Path(meta_path).exists(): + try: + meta = json.loads(Path(meta_path).read_text()) + except json.JSONDecodeError: + logger.warning("meta file %s is not valid JSON; ignoring", meta_path) + sidecars = { + key: snapshot_sidecar(container) + for key, container in discover_sidecars().items() + } + return merge_status(meta=meta, sidecars=sidecars, finals=scan_finals(jobs_dirs)) + + +# --------------------------------------------------------------------------- +# HTTP server + page +# --------------------------------------------------------------------------- + +PAGE = """ +vero harbor: experiments + + +

vero harbor: experiments

+
live trials, pre-registered bars, and the campaign scorecard. Auto-refreshes.
+
+
+
+ + +""" + + +@dataclass +class DashboardConfig: + meta_path: Path | None = None + jobs_dirs: list[Path] = field(default_factory=list) + host: str = "127.0.0.1" + port: int = 8300 + + +def make_handler(config: DashboardConfig): + class Handler(BaseHTTPRequestHandler): + def do_GET(self): # noqa: N802 (BaseHTTPRequestHandler API) + if self.path.startswith("/api/status"): + body = json.dumps( + collect(config.meta_path, config.jobs_dirs) + ).encode() + ctype = "application/json" + elif self.path == "/" or self.path.startswith("/index"): + body = PAGE.encode() + ctype = "text/html; charset=utf-8" + else: + self.send_error(404) + return + self.send_response(200) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): # quiet: one poll every 20s per client + pass + + return Handler + + +def serve_dashboard(config: DashboardConfig) -> None: + server = ThreadingHTTPServer( + (config.host, config.port), make_handler(config) + ) + logger.info("dashboard on http://%s:%d", config.host, config.port) + print(f"vero harbor dashboard: http://{config.host}:{config.port}") + server.serve_forever() diff --git a/vero/tests/test_harbor_dashboard.py b/vero/tests/test_harbor_dashboard.py new file mode 100644 index 0000000..dbfbd2b --- /dev/null +++ b/vero/tests/test_harbor_dashboard.py @@ -0,0 +1,128 @@ +"""Tests for vero.harbor.dashboard: the pure status-assembly layer. + +The docker/HTTP collectors are thin shims; everything that decides what the +operator sees (key derivation, phase, verdicts, the meta/live/final join) is +pure and covered here without docker. +""" + +import json + +from vero.harbor.dashboard import ( + derive_phase, + experiment_key, + merge_status, + normalize_rounds, + scan_finals, + verdict, +) + + +class TestExperimentKey: + def test_sidecar_name_maps_to_task_key(self): + assert experiment_key("gaia-exp11-task__nxkmtjb-eval-sidecar-1") == "gaia-exp11-task" + + def test_non_sidecar_containers_ignored(self): + assert experiment_key("gaia-exp11-task__nxkmtjb-main-1") is None + assert experiment_key("k8s_coredns_whatever") is None + + def test_sidecar_without_run_suffix_still_keys(self): + assert experiment_key("plain-task-eval-sidecar-1") == "plain-task" + + +class TestPhase: + def test_container_without_rounds_is_building(self): + assert derive_phase(has_container=True, n_rounds=0, final=None) == "building" + + def test_container_with_rounds_is_running(self): + assert derive_phase(has_container=True, n_rounds=3, final=None) == "running" + + def test_reward_json_wins_over_everything(self): + assert derive_phase(has_container=False, n_rounds=0, final={"accuracy": 0.4}) == "done" + assert derive_phase(has_container=True, n_rounds=9, final={"accuracy": 0.4}) == "done" + + def test_no_container_no_final_is_gone(self): + # A crashed launch must be conspicuous, not silently absent. + assert derive_phase(has_container=False, n_rounds=0, final=None) == "gone" + + +class TestVerdict: + def test_passed_when_above_bar(self): + assert verdict({"accuracy": 0.6}, {"win_bar": 0.41}) == "PASSED" + + def test_failed_when_at_or_below_bar(self): + assert verdict({"accuracy": 0.40}, {"win_bar": 0.41}) == "FAILED" + assert verdict({"accuracy": 0.41}, {"win_bar": 0.41}) == "FAILED" # bar is strict + + def test_no_bars_no_verdict(self): + assert verdict({"accuracy": 0.6}, None) is None + assert verdict(None, {"win_bar": 0.41}) is None + + +class TestNormalizeRounds: + def test_rows_truncate_commit_and_carry_scores(self): + payload = {"experiments": [ + {"commit": "d5a5facb994912345678", "split": "train", + "mean_score": 0.5555, "error_rate": 0.0, "created_at": "t1"}, + {"commit": None, "split": None, "mean_score": None, "error_rate": None}, + ]} + rows = normalize_rounds(payload) + assert rows[0]["commit"] == "d5a5facb99" + assert rows[0]["score"] == 0.5555 + assert rows[1]["commit"] == "" # tolerant of null ledger rows + assert normalize_rounds(None) == [] + + +class TestMergeStatus: + META = { + "experiments": [ + {"key": "gaia-exp11-task", "title": "Exp11: floor dogfood", + "question": "Does the floor revert harm?", + "bars": {"baseline": 0.317, "win_bar": 0.454}}, + {"key": "gaia-exp12-task", "title": "Exp12: queued"}, + ], + "history": [{"exp": "7", "question": "positive control", "final": "0.6", "verdict": "WIN"}], + } + + def test_meta_live_and_final_join_by_key(self): + sidecars = {"gaia-exp11-task": { + "experiments": {"experiments": [ + {"commit": "abc1234500", "split": "train", "mean_score": 0.25, "error_rate": 0.0}]}, + "status": {"splits": [{"split": "train", "remaining_run_budget": 5}]}, + }} + out = merge_status(meta=self.META, sidecars=sidecars, finals={}) + e11 = next(e for e in out["experiments"] if e["key"] == "gaia-exp11-task") + assert e11["phase"] == "running" + assert e11["rounds"][0]["score"] == 0.25 + assert e11["budget"] == [{"split": "train", "remaining_run_budget": 5}] + # meta-only experiment shows as gone (conspicuous, not hidden) + e12 = next(e for e in out["experiments"] if e["key"] == "gaia-exp12-task") + assert e12["phase"] == "gone" + assert out["history"] == self.META["history"] + + def test_unlisted_live_trial_still_appears(self): + # A running trial the meta forgot must never be invisible. + sidecars = {"gaia-expX-task": {"experiments": {"experiments": []}, "status": {}}} + out = merge_status(meta=self.META, sidecars=sidecars, finals={}) + ex = next(e for e in out["experiments"] if e["key"] == "gaia-expX-task") + assert ex["phase"] == "building" + assert ex["title"] == "gaia-expX-task" # falls back to the key + + def test_final_produces_done_and_verdict(self): + finals = {"gaia-exp11-task": {"accuracy": 0.5}} + out = merge_status(meta=self.META, sidecars={}, finals=finals) + e11 = next(e for e in out["experiments"] if e["key"] == "gaia-exp11-task") + assert e11["phase"] == "done" + assert e11["final"] == {"accuracy": 0.5} + assert e11["verdict"] == "PASSED" # 0.5 > 0.454 + + +class TestScanFinals: + def test_reward_json_keyed_by_trial_dir_task_name(self, tmp_path): + trial = tmp_path / "jobs" / "2026-07-04__x" / "gaia-exp11-task__AbCdEf" / "verifier" + trial.mkdir(parents=True) + (trial / "reward.json").write_text(json.dumps({"accuracy": 0.4})) + finals = scan_finals([tmp_path / "jobs"]) + assert finals == {"gaia-exp11-task": {"accuracy": 0.4}} + + def test_missing_dir_is_empty(self, tmp_path): + assert scan_finals([tmp_path / "nope"]) == {}