From 0927f2dd9c23d34afbc48f772ef695c0fff3ba4d Mon Sep 17 00:00:00 2001 From: Johannes du Plessis Date: Tue, 16 Jun 2026 19:38:36 -0700 Subject: [PATCH] feat: Run reviewer eval in a GitHub Action; dashboard becomes read-only (#1556) * Run reviewer eval in a GitHub Action; make dashboard a read-only progress view The dashboard launched the eval as a subprocess inside the serving deployment worker, so a container recycle killed long runs and discarded results that had already completed server-side. Move the harness to a workflow_dispatch Action (run on prod). run_eval now publishes status/progress/log-tail to the LangGraph store record the dashboard reads, so /admin/evals stays a live view; a killed Action surfaces as failed via the stale-heartbeat reconcile. * reviewer_eval workflow: pass inputs via env, no shell interpolation Addresses the reviewer finding: workflow_dispatch string inputs were interpolated into the run: block (limit unquoted), allowing shell injection in a job holding LANGSMITH/ANTHROPIC keys. Pass inputs through env and reference quoted "$VARS"; validate limit is numeric and build its flag in bash. --------- Co-authored-by: open-swe[bot] --- .github/workflows/reviewer_eval.yml | 127 +++++++ agent/dashboard/eval_jobs.py | 289 ++-------------- agent/dashboard/routes.py | 128 +------ agent/reviewer_eval_store.py | 24 ++ evals/reviewer/README.md | 27 +- evals/reviewer/run_eval.py | 122 ++++++- evals/reviewer/store_reporter.py | 121 +++++++ evals/reviewer/target.py | 19 ++ tests/test_eval_jobs.py | 148 ++------ tests/test_eval_store_reporter.py | 95 ++++++ tests/test_reviewer_eval_target.py | 7 + ui/src/lib/api.ts | 15 +- ui/src/routes/admin_.evals.tsx | 511 +++------------------------- 13 files changed, 624 insertions(+), 1009 deletions(-) create mode 100644 .github/workflows/reviewer_eval.yml create mode 100644 agent/reviewer_eval_store.py create mode 100644 evals/reviewer/store_reporter.py create mode 100644 tests/test_eval_store_reporter.py diff --git a/.github/workflows/reviewer_eval.yml b/.github/workflows/reviewer_eval.yml new file mode 100644 index 00000000..66f62356 --- /dev/null +++ b/.github/workflows/reviewer_eval.yml @@ -0,0 +1,127 @@ +# Runs the reviewer benchmark (evals/reviewer/run_eval.py) on a durable runner instead of as a +# subprocess inside the serving deployment. Trigger it from the Actions UI / `gh workflow run`, +# selecting the `prod` branch so the harness + judge match the deployed reviewer it scores. +# +# Progress is streamed to the LangGraph store record the dashboard reads, so the run shows up live +# at /admin/evals. Required repository config: +# secrets: LANGSMITH_API_KEY, ANTHROPIC_API_KEY (judge runs in-process; reviewer model keys +# are NOT needed — the reviewer runs in the deployment) +# secret or var: LANGGRAPH_URL (the deployment URL the eval drives + reports to) +name: Reviewer eval + +on: + workflow_dispatch: + inputs: + model_id: + description: Reviewer model id + type: string + default: google_genai:gemini-3.5-flash + reasoning_effort: + description: Reasoning effort + type: string + default: medium + dataset_name: + description: LangSmith dataset + type: string + default: openswe-reviewer-v1 + experiment_prefix: + description: Run name (LangSmith experiment prefix) + type: string + default: openswe-review-confidence + max_concurrency: + description: Max concurrent PRs + type: string + default: "5" + score_mode: + description: all_findings | surfaced_findings + type: choice + default: all_findings + options: + - all_findings + - surfaced_findings + severity_threshold: + description: Severity threshold (surfaced_findings only) + type: choice + default: medium + options: + - low + - medium + - high + - critical + cap: + description: Max surfaced findings per PR (surfaced_findings only) + type: string + default: "4" + limit: + description: Run only the first N examples (blank = full dataset) + type: string + default: "" + langsmith_project: + description: LangSmith tracing project for eval traces + type: string + default: open-swe-evals + assistant_id: + description: Reviewer assistant id + type: string + default: reviewer + +concurrency: + group: reviewer-eval + cancel-in-progress: false + +permissions: + contents: read + +jobs: + reviewer-eval: + name: Reviewer eval + runs-on: ubuntu-latest + timeout-minutes: 360 + steps: + - uses: actions/checkout@v6 + - uses: astral-sh/setup-uv@08807647e7069bb48b6ef5acd8ec9567f424441b # v8.1.0 + - name: Install dependencies + run: uv sync --locked + - name: Run reviewer eval + # Inputs are passed via env and referenced as quoted "$VARS" — never + # interpolated into the script — so dispatcher-supplied text is treated + # as data, not shell syntax. + env: + LANGSMITH_API_KEY: ${{ secrets.LANGSMITH_API_KEY }} + LANGCHAIN_API_KEY: ${{ secrets.LANGSMITH_API_KEY }} + ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }} + LANGGRAPH_URL: ${{ secrets.LANGGRAPH_URL || vars.LANGGRAPH_URL }} + REVIEWER_EVAL_REPORT_STORE: "1" + INPUT_MODEL_ID: ${{ inputs.model_id }} + INPUT_REASONING_EFFORT: ${{ inputs.reasoning_effort }} + INPUT_DATASET_NAME: ${{ inputs.dataset_name }} + INPUT_EXPERIMENT_PREFIX: ${{ inputs.experiment_prefix }} + INPUT_MAX_CONCURRENCY: ${{ inputs.max_concurrency }} + INPUT_SCORE_MODE: ${{ inputs.score_mode }} + INPUT_SEVERITY_THRESHOLD: ${{ inputs.severity_threshold }} + INPUT_CAP: ${{ inputs.cap }} + INPUT_LIMIT: ${{ inputs.limit }} + INPUT_LANGSMITH_PROJECT: ${{ inputs.langsmith_project }} + INPUT_ASSISTANT_ID: ${{ inputs.assistant_id }} + run: | + set -euo pipefail + limit_args=() + if [ -n "${INPUT_LIMIT}" ]; then + if ! [[ "${INPUT_LIMIT}" =~ ^[0-9]+$ ]]; then + echo "limit must be a positive integer, got: ${INPUT_LIMIT}" >&2 + exit 1 + fi + limit_args=(--limit "${INPUT_LIMIT}") + fi + uv run python -m evals.reviewer.run_eval \ + --model-id "${INPUT_MODEL_ID}" \ + --reasoning-effort "${INPUT_REASONING_EFFORT}" \ + --dataset-name "${INPUT_DATASET_NAME}" \ + --experiment-prefix "${INPUT_EXPERIMENT_PREFIX}" \ + --max-concurrency "${INPUT_MAX_CONCURRENCY}" \ + --score-mode "${INPUT_SCORE_MODE}" \ + --severity-threshold "${INPUT_SEVERITY_THRESHOLD}" \ + --cap "${INPUT_CAP}" \ + --langsmith-project "${INPUT_LANGSMITH_PROJECT}" \ + --assistant-id "${INPUT_ASSISTANT_ID}" \ + "${limit_args[@]}" diff --git a/agent/dashboard/eval_jobs.py b/agent/dashboard/eval_jobs.py index 6a09d299..7b36ccd8 100644 --- a/agent/dashboard/eval_jobs.py +++ b/agent/dashboard/eval_jobs.py @@ -1,39 +1,30 @@ -"""Kick off and track the reviewer eval from the admin dashboard. +"""Track the reviewer eval for the admin dashboard. -Runs ``evals.reviewer.run_eval`` as a subprocess so its LangSmith tracing -project (``open-swe-evals``) stays isolated from the deployment's production -project, and so a long eval does not block the server event loop. Status is -persisted in the LangGraph store so it survives across dashboard requests. +The eval itself runs in the ``Reviewer eval`` GitHub Action (durable runner, +isolated from the serving deployment). The Action's harness reports progress +into a LangGraph store record (namespace ``["evals"]``, key ``"reviewer"``) via +``evals.reviewer.store_reporter``; this module reads that record for the +dashboard and reconciles a run whose heartbeat has gone stale (e.g. the Action +was killed) to ``failed``. """ from __future__ import annotations -import asyncio -import contextlib import logging import os -import re -import sys -import uuid from datetime import UTC, datetime -from pathlib import Path from typing import Any, Literal, TypedDict from langgraph_sdk import get_client -logger = logging.getLogger(__name__) +from agent.reviewer_eval_store import ( + _HEARTBEAT_STALE_SECONDS, + DEFAULT_EVAL_PROJECT, + EVALS_NAMESPACE, + REVIEWER_EVAL_KEY, +) -EVALS_NAMESPACE: list[str] = ["evals"] -REVIEWER_EVAL_KEY = "reviewer" -DEFAULT_EVAL_PROJECT = "open-swe-evals" -_MODULE = "evals.reviewer.run_eval" -_LOG_TAIL_CHARS = 12000 -_EXPERIMENT_URL_RE = re.compile(r"https://\S*smith\.langchain\.com/\S+") -# The owning worker refreshes the heartbeat this often while the subprocess -# runs; a record is only reconciled as failed once its heartbeat is older than -# the stale threshold, so polls on other workers don't kill a live run. -_HEARTBEAT_INTERVAL_SECONDS = 10 -_HEARTBEAT_STALE_SECONDS = 60 +logger = logging.getLogger(__name__) EvalStatus = Literal["idle", "running", "completed", "failed"] ScoreMode = Literal["all_findings", "surfaced_findings"] @@ -68,18 +59,6 @@ DEFAULT_REVIEWER_EVAL_CONFIG: ReviewerEvalConfig = { "cap": 4, } -# Identifies this process so heartbeat ownership can be reasoned about across -# workers that share the persisted record. -_WORKER_ID = uuid.uuid4().hex - -# Live subprocess handles keyed by eval name, owned by the worker that launched -# them. The store record is the source of truth across workers/requests. -_PROCS: dict[str, asyncio.subprocess.Process] = {} - -# Rolling tail of subprocess output, kept by the owning worker so the heartbeat -# loop can persist a live log tail to the store while the eval runs. -_LOG_BUFFERS: dict[str, str] = {} - def _client(): return get_client() @@ -89,10 +68,6 @@ def _now_iso() -> str: return datetime.now(UTC).isoformat() -def _repo_root() -> Path: - return Path(__file__).resolve().parents[2] - - def _resolve_langgraph_url() -> str | None: return os.environ.get("LANGGRAPH_URL") or os.environ.get("LANGGRAPH_URL_PROD") @@ -112,27 +87,6 @@ def _resolve_eval_config(config: ReviewerEvalConfig | None = None) -> ReviewerEv return resolved -def _config_cli_args(config: ReviewerEvalConfig) -> list[str]: - values: list[tuple[str, object]] = [ - ("--dataset-name", config["dataset_name"]), - ("--experiment-prefix", config["experiment_prefix"]), - ("--max-concurrency", config["max_concurrency"]), - ("--langsmith-project", config["langsmith_project"]), - ("--assistant-id", config["assistant_id"]), - ("--model-id", config["model_id"]), - ("--reasoning-effort", config["reasoning_effort"]), - ("--score-mode", config["score_mode"]), - ("--severity-threshold", config["severity_threshold"]), - ("--cap", config["cap"]), - ] - if config["langgraph_url"]: - values.append(("--langgraph-url", config["langgraph_url"])) - args: list[str] = [] - for flag, value in values: - args.extend([flag, str(value)]) - return args - - def _idle_record() -> dict[str, Any]: config = _resolve_eval_config() return { @@ -152,6 +106,9 @@ def _idle_record() -> dict[str, Any]: "log_tail": None, "worker_id": None, "heartbeat": None, + "progress": None, + "github_run_url": None, + "trigger": None, "updated_at": _now_iso(), } @@ -177,11 +134,6 @@ async def _put_record(record: dict[str, Any]) -> dict[str, Any]: return record -def _is_locally_running() -> bool: - proc = _PROCS.get(REVIEWER_EVAL_KEY) - return proc is not None and proc.returncode is None - - def _heartbeat_age_seconds(record: dict[str, Any]) -> float | None: """Seconds since the record's heartbeat, or ``None`` if absent/unparseable.""" hb = record.get("heartbeat") @@ -202,218 +154,25 @@ def _is_heartbeat_fresh(record: dict[str, Any]) -> bool: async def get_reviewer_eval_status() -> dict[str, Any]: - """Return the latest reviewer-eval status, reconciling stale ``running``. + """Return the latest reviewer-eval status, reconciling a stale ``running``. - The owning worker refreshes the record's heartbeat while its subprocess - runs. A poll from any worker only marks the run failed once the heartbeat - is stale, so a status check on a worker that doesn't own the process (no - local handle but a fresh heartbeat) leaves a live run untouched. + The GitHub Action refreshes the record's heartbeat while it runs. A poll + only marks the run failed once the heartbeat is stale, so a healthy run is + left untouched and a killed Action surfaces as ``failed`` within the stale + threshold. """ record = await _get_record() if record is None: return _idle_record() if record.get("status") != "running": return record - if _is_locally_running() or _is_heartbeat_fresh(record): + if _is_heartbeat_fresh(record): return record return await _put_record( { **record, "status": "failed", "finished_at": record.get("finished_at") or _now_iso(), - "error": "Eval process is no longer tracked (server restarted?).", - } - ) - - -async def start_reviewer_eval( - *, - limit: int | None, - config: ReviewerEvalConfig | None = None, - created_by: str, -) -> dict[str, Any]: - """Launch the reviewer eval subprocess and persist a ``running`` record. - - Raises ``RuntimeError`` if an eval is already running on this or another - worker (detected via a fresh heartbeat on the shared record). - """ - if _is_locally_running(): - raise RuntimeError("a reviewer eval is already running") - existing = await _get_record() - if existing and existing.get("status") == "running" and _is_heartbeat_fresh(existing): - raise RuntimeError("a reviewer eval is already running") - - config_snapshot = _resolve_eval_config(config) - project = config_snapshot["langsmith_project"] - cmd = [sys.executable, "-m", _MODULE] - if limit is not None and limit > 0: - cmd += ["--limit", str(limit)] - cmd += _config_cli_args(config_snapshot) - - env = { - **os.environ, - "LANGSMITH_PROJECT": project, - "LANGCHAIN_PROJECT": project, - "LANGSMITH_TRACING": "true", - } - langgraph_url = _resolve_langgraph_url() - if langgraph_url: - env["LANGGRAPH_URL"] = langgraph_url - - try: - proc = await asyncio.create_subprocess_exec( - *cmd, - cwd=str(_repo_root()), - env=env, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.STDOUT, - ) - except Exception as exc: - logger.exception("Failed to launch reviewer eval subprocess") - return await _put_record( - { - **_idle_record(), - "status": "failed", - "run_name": config_snapshot["experiment_prefix"], - "langsmith_project": project, - "limit": limit, - "config_snapshot": config_snapshot, - "finished_at": _now_iso(), - "created_by": created_by, - "error": f"Failed to launch eval: {exc}", - } - ) - - _PROCS[REVIEWER_EVAL_KEY] = proc - record = await _put_record( - { - **_idle_record(), - "status": "running", - "run_name": config_snapshot["experiment_prefix"], - "langsmith_project": project, - "limit": limit, - "config_snapshot": config_snapshot, - "started_at": _now_iso(), - "finished_at": None, - "created_by": created_by, - "pid": proc.pid, - "worker_id": _WORKER_ID, - "heartbeat": _now_iso(), - } - ) - asyncio.create_task( - _monitor( - proc, - created_by=created_by, - limit=limit, - config_snapshot=config_snapshot, - ) - ) - return record - - -async def cancel_reviewer_eval() -> dict[str, Any]: - """Terminate a locally-running reviewer eval, if any.""" - proc = _PROCS.get(REVIEWER_EVAL_KEY) - if proc is not None and proc.returncode is None: - try: - proc.terminate() - except ProcessLookupError: - pass - record = await _get_record() or _idle_record() - return await _put_record( - { - **record, - "status": "failed", - "finished_at": _now_iso(), - "error": "Eval cancelled by an admin.", - } - ) - - -async def _heartbeat_loop(proc: asyncio.subprocess.Process) -> None: - """Refresh heartbeat and live log tail while the owned subprocess is alive.""" - while proc.returncode is None: - await asyncio.sleep(_HEARTBEAT_INTERVAL_SECONDS) - if proc.returncode is not None: - return - record = await _get_record() - if not record or record.get("status") != "running": - return - await _put_record( - { - **record, - "heartbeat": _now_iso(), - "log_tail": _LOG_BUFFERS.get(REVIEWER_EVAL_KEY) or record.get("log_tail"), - } - ) - - -async def _stream_output(proc: asyncio.subprocess.Process) -> tuple[str, str | None]: - """Read stdout to EOF, keeping a rolling tail and the last experiment URL. - - The tail is published to ``_LOG_BUFFERS`` as it grows so the heartbeat loop - can persist it mid-run. Reading in fixed chunks avoids the line-length cap - that ``StreamReader.readline`` would impose on long log lines. - """ - tail = "" - experiment_url: str | None = None - if proc.stdout is None: - return tail, experiment_url - while True: - chunk = await proc.stdout.read(4096) - if not chunk: - break - tail = (tail + chunk.decode("utf-8", errors="replace"))[-_LOG_TAIL_CHARS:] - urls = _EXPERIMENT_URL_RE.findall(tail) - if urls: - experiment_url = urls[-1] - _LOG_BUFFERS[REVIEWER_EVAL_KEY] = tail - return tail, experiment_url - - -async def _monitor( - proc: asyncio.subprocess.Process, - *, - created_by: str, - limit: int | None, - config_snapshot: ReviewerEvalConfig, -) -> None: - heartbeat = asyncio.create_task(_heartbeat_loop(proc)) - tail = "" - experiment_url: str | None = None - try: - tail, experiment_url = await _stream_output(proc) - await proc.wait() - except Exception: - logger.exception("Error while monitoring reviewer eval subprocess") - finally: - heartbeat.cancel() - with contextlib.suppress(asyncio.CancelledError): - await heartbeat - _PROCS.pop(REVIEWER_EVAL_KEY, None) - _LOG_BUFFERS.pop(REVIEWER_EVAL_KEY, None) - - log_tail = tail or None - exit_code = proc.returncode - status: EvalStatus = "completed" if exit_code == 0 else "failed" - error = None if status == "completed" else f"Eval exited with code {exit_code}." - - record = await _get_record() or _idle_record() - await _put_record( - { - **record, - "status": status, - "run_name": config_snapshot["experiment_prefix"], - "langsmith_project": config_snapshot["langsmith_project"], - "limit": limit, - "config_snapshot": config_snapshot, - "created_by": created_by, - "finished_at": _now_iso(), - "pid": proc.pid, - "exit_code": exit_code, - "experiment_url": experiment_url, - "error": error, - "log_tail": log_tail, + "error": "Eval process is no longer tracked (GitHub Action stopped?).", } ) diff --git a/agent/dashboard/routes.py b/agent/dashboard/routes.py index 34d3136a..ba4f9f19 100644 --- a/agent/dashboard/routes.py +++ b/agent/dashboard/routes.py @@ -5,12 +5,12 @@ from __future__ import annotations import hmac import logging import os -from typing import Any, Literal +from typing import Any import httpx from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request from fastapi.responses import RedirectResponse, Response, StreamingResponse -from pydantic import BaseModel, field_validator, model_validator +from pydantic import BaseModel from .admin import is_admin from .agent_instructions import ( @@ -33,11 +33,7 @@ from .enabled_repos import ( set_review_repo_enabled, ) from .eval_jobs import ( - DEFAULT_REVIEWER_EVAL_CONFIG, - ReviewerEvalConfig, - cancel_reviewer_eval, get_reviewer_eval_status, - start_reviewer_eval, ) from .oauth import ( COOKIE_NAME, @@ -56,7 +52,7 @@ from .oauth import ( require_session, sanitize_redirect_to, ) -from .options import SUPPORTED_MODEL_IDS, SUPPORTED_MODELS, model_supports_effort +from .options import SUPPORTED_MODELS from .profiles import ( ProfileUpdate, get_profile, @@ -569,130 +565,14 @@ async def admin_delete_user_mapping( return {"deleted": deleted} -ScoreMode = Literal["all_findings", "surfaced_findings"] -Severity = Literal["low", "medium", "high", "critical"] - - -class ReviewerEvalStartBody(BaseModel): - limit: int | None = None - dataset_name: str = DEFAULT_REVIEWER_EVAL_CONFIG["dataset_name"] - experiment_prefix: str = DEFAULT_REVIEWER_EVAL_CONFIG["experiment_prefix"] - max_concurrency: int = DEFAULT_REVIEWER_EVAL_CONFIG["max_concurrency"] - langsmith_project: str = DEFAULT_REVIEWER_EVAL_CONFIG["langsmith_project"] - langgraph_url: str = DEFAULT_REVIEWER_EVAL_CONFIG["langgraph_url"] - assistant_id: str = DEFAULT_REVIEWER_EVAL_CONFIG["assistant_id"] - model_id: str = DEFAULT_REVIEWER_EVAL_CONFIG["model_id"] - reasoning_effort: str = DEFAULT_REVIEWER_EVAL_CONFIG["reasoning_effort"] - score_mode: ScoreMode = DEFAULT_REVIEWER_EVAL_CONFIG["score_mode"] - severity_threshold: Severity = DEFAULT_REVIEWER_EVAL_CONFIG["severity_threshold"] - cap: int = DEFAULT_REVIEWER_EVAL_CONFIG["cap"] - - @field_validator( - "dataset_name", - "experiment_prefix", - "langsmith_project", - "assistant_id", - "model_id", - "reasoning_effort", - mode="before", - ) - @classmethod - def _normalize_required_string(cls, value: object) -> str: - if not isinstance(value, str): - raise ValueError("must be a string") - text = value.strip() - if not text: - raise ValueError("must not be blank") - return text - - @field_validator("langgraph_url", mode="before") - @classmethod - def _normalize_optional_string(cls, value: object) -> str: - if value is None: - return "" - if not isinstance(value, str): - raise ValueError("must be a string") - return value.strip() - - @field_validator("limit") - @classmethod - def _validate_limit(cls, value: int | None) -> int | None: - if value is not None and value <= 0: - raise ValueError("limit must be positive") - return value - - @field_validator("max_concurrency") - @classmethod - def _validate_max_concurrency(cls, value: int) -> int: - if value <= 0: - raise ValueError("max_concurrency must be positive") - return value - - @field_validator("cap") - @classmethod - def _validate_cap(cls, value: int) -> int: - if value < 0: - raise ValueError("cap must be non-negative") - return value - - @model_validator(mode="after") - def _validate_model_effort(self) -> ReviewerEvalStartBody: - if self.model_id not in SUPPORTED_MODEL_IDS: - raise ValueError(f"unsupported reviewer eval model: {self.model_id}") - if not model_supports_effort(self.model_id, self.reasoning_effort): - raise ValueError( - f"effort {self.reasoning_effort!r} not supported by model {self.model_id!r}" - ) - return self - - def eval_config(self) -> ReviewerEvalConfig: - return { - "dataset_name": self.dataset_name, - "experiment_prefix": self.experiment_prefix, - "max_concurrency": self.max_concurrency, - "langsmith_project": self.langsmith_project, - "langgraph_url": self.langgraph_url, - "assistant_id": self.assistant_id, - "model_id": self.model_id, - "reasoning_effort": self.reasoning_effort, - "score_mode": self.score_mode, - "severity_threshold": self.severity_threshold, - "cap": self.cap, - } - - @router.get("/admin/evals/reviewer") async def admin_get_reviewer_eval( _admin: dict[str, Any] = _ADMIN_DEP, ) -> dict[str, Any]: + """Read-only status for the reviewer eval (triggered from the GitHub Action).""" return await get_reviewer_eval_status() -@router.post("/admin/evals/reviewer") -async def admin_start_reviewer_eval( - body: ReviewerEvalStartBody, - session: dict[str, Any] = _ADMIN_DEP, -) -> dict[str, Any]: - status = await get_reviewer_eval_status() - if status.get("status") == "running": - raise HTTPException(409, "a reviewer eval is already running") - try: - return await start_reviewer_eval( - limit=body.limit, - config=body.eval_config(), - created_by=session["sub"], - ) - except RuntimeError as exc: - raise HTTPException(409, str(exc)) from exc - - -@router.delete("/admin/evals/reviewer") -async def admin_cancel_reviewer_eval( - _admin: dict[str, Any] = _ADMIN_DEP, -) -> dict[str, Any]: - return await cancel_reviewer_eval() - - def _next_link_url(link_header: str | None) -> str | None: if not link_header: return None diff --git a/agent/reviewer_eval_store.py b/agent/reviewer_eval_store.py new file mode 100644 index 00000000..505284bd --- /dev/null +++ b/agent/reviewer_eval_store.py @@ -0,0 +1,24 @@ +"""Shared constants for the reviewer-eval status record. + +Kept deliberately light (no dashboard/server imports) so the eval harness in the +GitHub Action can publish progress to the store without importing the FastAPI +dashboard. Both ``agent.dashboard.eval_jobs`` (reader) and +``evals.reviewer.store_reporter`` (writer) import from here. +""" + +from __future__ import annotations + +import re + +EVALS_NAMESPACE: list[str] = ["evals"] +REVIEWER_EVAL_KEY = "reviewer" +DEFAULT_EVAL_PROJECT = "open-swe-evals" + +_LOG_TAIL_CHARS = 12000 +_EXPERIMENT_URL_RE = re.compile(r"https://\S*smith\.langchain\.com/\S+") + +# The running Action refreshes the heartbeat this often; a record is only +# reconciled as failed once its heartbeat is older than the stale threshold, so +# a brief dashboard/Action lag doesn't kill a live run. +_HEARTBEAT_INTERVAL_SECONDS = 10 +_HEARTBEAT_STALE_SECONDS = 60 diff --git a/evals/reviewer/README.md b/evals/reviewer/README.md index dacb8ea7..2e5767d1 100644 --- a/evals/reviewer/README.md +++ b/evals/reviewer/README.md @@ -12,6 +12,7 @@ evals/reviewer/ ├── config.toml # default benchmark run config ├── judge.py # claude-opus-4-5 pairwise match evaluator + aggregate ├── target.py # invokes the reviewer graph over langgraph_sdk +├── store_reporter.py # publishes live progress to the dashboard store record └── run_eval.py # client.aevaluate entrypoint ``` @@ -54,14 +55,26 @@ Smoke-test with 3 PRs first: uv run python -m evals.reviewer.run_eval --limit 3 ``` -### From the admin dashboard +### From the GitHub Action (recommended for full runs) -Admins can also kick off the eval from the **Reviewer eval** section on the -dashboard Admin page (no local shell needed). It launches the same -`run_eval` runner as a subprocess against the running deployment, with an -optional limit for a smoke test, and surfaces live status plus the LangSmith -experiment link. The deployment must have `LANGSMITH_API_KEY` / `ANTHROPIC_API_KEY` -in its environment. +Trigger the **Reviewer eval** workflow (`.github/workflows/reviewer_eval.yml`) +from the Actions UI or `gh workflow run reviewer_eval.yml --ref prod -f limit=3`. +Run it on the **prod** branch so the harness/judge match the deployed reviewer it +scores. Running it on a durable runner (instead of inside the serving deployment) +means a deploy or container recycle can't kill a long run. + +The Action sets `REVIEWER_EVAL_REPORT_STORE=1`, so `run_eval` publishes live +status/progress/logs to the LangGraph store record the dashboard reads — watch it +at **Admin → Reviewer eval** (`/admin/evals`), which is now a read-only progress +view (status, `completed / total`, log tail, LangSmith experiment link, and a link +back to the GitHub run). If the Action is cancelled/killed, the heartbeat goes +stale and the dashboard flips the run to `failed` within ~60s. + +Required repository config: + +- secrets: `LANGSMITH_API_KEY`, `ANTHROPIC_API_KEY` (the judge runs in-process; + reviewer-model keys are **not** needed — the reviewer runs in the deployment). +- secret or var: `LANGGRAPH_URL` — the deployment URL the eval drives and reports to. ### Tracing project diff --git a/evals/reviewer/run_eval.py b/evals/reviewer/run_eval.py index c068d024..69e5f21d 100644 --- a/evals/reviewer/run_eval.py +++ b/evals/reviewer/run_eval.py @@ -7,8 +7,12 @@ Usage: from __future__ import annotations import argparse +import asyncio +import contextlib import logging import os +import sys +import threading import tomllib from collections.abc import Iterable, Mapping from pathlib import Path @@ -19,8 +23,15 @@ from langgraph_sdk import get_client from langsmith import Client, aevaluate from langsmith.schemas import Example +from agent.reviewer_eval_store import _EXPERIMENT_URL_RE, _LOG_TAIL_CHARS from evals.reviewer.judge import aggregate_pr, judge_match -from evals.reviewer.target import drain_thread_ids, get_langgraph_url, review_pr +from evals.reviewer.store_reporter import StoreReporter, is_enabled +from evals.reviewer.target import ( + drain_thread_ids, + get_completed_count, + get_langgraph_url, + review_pr, +) logger = logging.getLogger(__name__) @@ -195,6 +206,74 @@ def _apply_langsmith_project(project: str | None) -> None: os.environ.setdefault("LANGSMITH_TRACING", "true") +class _TailCapture: + """Thread-safe rolling tail of eval output + the last LangSmith experiment URL.""" + + def __init__(self) -> None: + self._buf = "" + self._url: str | None = None + self._lock = threading.Lock() + + def append(self, text: str) -> None: + if not text: + return + with self._lock: + self._buf = (self._buf + text)[-_LOG_TAIL_CHARS:] + found = _EXPERIMENT_URL_RE.findall(self._buf) + if found: + self._url = found[-1] + + def tail(self) -> str | None: + with self._lock: + return self._buf or None + + def url(self) -> str | None: + with self._lock: + return self._url + + +class _BufferingHandler(logging.Handler): + """Mirror log records into a ``_TailCapture`` so the reporter can publish them.""" + + def __init__(self, capture: _TailCapture) -> None: + super().__init__() + self._capture = capture + + def emit(self, record: logging.LogRecord) -> None: + try: + self._capture.append(self.format(record) + "\n") + except Exception: + pass + + +class _TeeStream: + """Write to the original stream and mirror into the capture (for ``print``ed output).""" + + def __init__(self, original: Any, capture: _TailCapture) -> None: + self._original = original + self._capture = capture + + def write(self, text: str) -> int: + self._capture.append(text) + return self._original.write(text) + + def flush(self) -> None: + self._original.flush() + + def __getattr__(self, name: str) -> Any: + return getattr(self._original, name) + + +def _resolve_total(dataset_name: str, data: str | list[Example]) -> int | None: + if isinstance(data, list): + return len(data) + try: + return Client().read_dataset(dataset_name=dataset_name).example_count + except Exception: + logger.warning("Could not resolve dataset example count for progress", exc_info=True) + return None + + async def _cleanup_threads(thread_ids: Iterable[str]) -> None: """Delete LangGraph threads created during the eval. @@ -270,6 +349,31 @@ async def main() -> None: else: data = dataset_name + reporter: StoreReporter | None = None + heartbeat: asyncio.Task[None] | None = None + log_handler: logging.Handler | None = None + original_stdout = sys.stdout + if is_enabled(): + capture = _TailCapture() + log_handler = _BufferingHandler(capture) + log_handler.setFormatter( + logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s") + ) + logging.getLogger().addHandler(log_handler) + sys.stdout = _TeeStream(original_stdout, capture) + reporter = StoreReporter( + config=dict(config), + limit=args.limit, + total=_resolve_total(dataset_name, data), + created_by=None, + completed_getter=get_completed_count, + tail_getter=capture.tail, + experiment_url_getter=capture.url, + ) + await reporter.start() + heartbeat = reporter.run_heartbeat() + + eval_error: BaseException | None = None try: await aevaluate( review_pr, @@ -280,7 +384,21 @@ async def main() -> None: max_concurrency=max_concurrency, num_repetitions=1, ) + except BaseException as exc: + eval_error = exc + raise finally: + if reporter is not None: + if heartbeat is not None: + heartbeat.cancel() + with contextlib.suppress(asyncio.CancelledError): + await heartbeat + status = "failed" if eval_error is not None else "completed" + error = None if eval_error is None else f"{type(eval_error).__name__}: {eval_error}" + await reporter.finish(status=status, error=error) + if log_handler is not None: + logging.getLogger().removeHandler(log_handler) + sys.stdout = original_stdout if not args.no_cleanup: thread_ids = drain_thread_ids() if thread_ids: @@ -289,6 +407,4 @@ async def main() -> None: if __name__ == "__main__": - import asyncio - asyncio.run(main()) diff --git a/evals/reviewer/store_reporter.py b/evals/reviewer/store_reporter.py new file mode 100644 index 00000000..595754ef --- /dev/null +++ b/evals/reviewer/store_reporter.py @@ -0,0 +1,121 @@ +"""Publish reviewer-eval progress to the LangGraph store for the dashboard. + +When the eval runs in the ``Reviewer eval`` GitHub Action it writes the same +store record the dashboard reads (namespace ``["evals"]``, key ``"reviewer"``), +so ``/admin/evals`` shows the run live. The dashboard reconciles a run whose +heartbeat goes stale to ``failed`` (see ``agent.dashboard.eval_jobs``), so the +reporter must keep heartbeating while the eval runs. +""" + +from __future__ import annotations + +import asyncio +import logging +import os +from collections.abc import Callable +from datetime import UTC, datetime +from typing import Any + +from langgraph_sdk import get_client + +from agent.reviewer_eval_store import ( + _HEARTBEAT_INTERVAL_SECONDS, + EVALS_NAMESPACE, + REVIEWER_EVAL_KEY, +) + +logger = logging.getLogger(__name__) + + +def is_enabled() -> bool: + """True when the eval should publish progress to the store (set by the Action).""" + return bool(os.environ.get("REVIEWER_EVAL_REPORT_STORE")) and bool( + os.environ.get("LANGGRAPH_URL") + ) + + +def _now_iso() -> str: + return datetime.now(UTC).isoformat() + + +def github_run_url() -> str | None: + server = os.environ.get("GITHUB_SERVER_URL") + repo = os.environ.get("GITHUB_REPOSITORY") + run_id = os.environ.get("GITHUB_RUN_ID") + if server and repo and run_id: + return f"{server}/{repo}/actions/runs/{run_id}" + return None + + +class StoreReporter: + """Writes the reviewer-eval status record + heartbeats to the deployment store.""" + + def __init__( + self, + *, + config: dict[str, Any], + limit: int | None, + total: int | None, + created_by: str | None, + completed_getter: Callable[[], int], + tail_getter: Callable[[], str | None], + experiment_url_getter: Callable[[], str | None], + ) -> None: + self._config = config + self._limit = limit + self._total = total + self._created_by = created_by or os.environ.get("GITHUB_ACTOR") + self._completed_getter = completed_getter + self._tail_getter = tail_getter + self._experiment_url_getter = experiment_url_getter + self._github_run_url = github_run_url() + self._worker_id = os.environ.get("GITHUB_RUN_ID") + self._started_at = _now_iso() + # get_client auto-loads the api key from LANGGRAPH/LANGSMITH/LANGCHAIN env. + self._client = get_client(url=os.environ["LANGGRAPH_URL"]) + + def _record(self, *, status: str, **overrides: Any) -> dict[str, Any]: + record: dict[str, Any] = { + "name": REVIEWER_EVAL_KEY, + "status": status, + "run_name": self._config.get("experiment_prefix"), + "langsmith_project": self._config.get("langsmith_project"), + "limit": self._limit, + "config_snapshot": self._config, + "started_at": self._started_at, + "finished_at": None, + "created_by": self._created_by, + "pid": None, + "exit_code": None, + "experiment_url": self._experiment_url_getter(), + "error": None, + "log_tail": self._tail_getter(), + "worker_id": self._worker_id, + "heartbeat": _now_iso(), + "progress": {"completed": self._completed_getter(), "total": self._total}, + "github_run_url": self._github_run_url, + "trigger": "github_action", + "updated_at": _now_iso(), + } + record.update(overrides) + return record + + async def _put(self, record: dict[str, Any]) -> None: + try: + await self._client.store.put_item(EVALS_NAMESPACE, REVIEWER_EVAL_KEY, record) + except Exception: + logger.warning("Failed to publish reviewer eval status to store", exc_info=True) + + async def start(self) -> None: + await self._put(self._record(status="running")) + + async def _heartbeat_loop(self) -> None: + while True: + await asyncio.sleep(_HEARTBEAT_INTERVAL_SECONDS) + await self._put(self._record(status="running")) + + def run_heartbeat(self) -> asyncio.Task[None]: + return asyncio.create_task(self._heartbeat_loop()) + + async def finish(self, *, status: str, error: str | None = None) -> None: + await self._put(self._record(status=status, finished_at=_now_iso(), error=error)) diff --git a/evals/reviewer/target.py b/evals/reviewer/target.py index 4b3a0bc8..f22261c4 100644 --- a/evals/reviewer/target.py +++ b/evals/reviewer/target.py @@ -29,6 +29,24 @@ _VALID_SEVERITIES: set[Severity] = {"low", "medium", "high", "critical"} _THREAD_IDS: set[str] = set() _THREAD_IDS_LOCK = threading.Lock() +_COMPLETED = 0 +_COMPLETED_LOCK = threading.Lock() + + +def _record_completed() -> None: + global _COMPLETED + with _COMPLETED_LOCK: + _COMPLETED += 1 + + +def get_completed_count() -> int: + """Number of examples that have finished so far in this process. + + Read by ``store_reporter`` to publish progress to the dashboard. + """ + with _COMPLETED_LOCK: + return _COMPLETED + def _record_thread_id(thread_id: str) -> None: with _THREAD_IDS_LOCK: @@ -144,6 +162,7 @@ async def review_pr(inputs: dict[str, Any]) -> dict[str, Any]: len(comments), thread_id, ) + _record_completed() return {"comments": comments} except Exception: logger.exception("Reviewer eval example failed: repo=%s pr=%s", repo, pr_number) diff --git a/tests/test_eval_jobs.py b/tests/test_eval_jobs.py index b748f45a..2f969cb5 100644 --- a/tests/test_eval_jobs.py +++ b/tests/test_eval_jobs.py @@ -1,20 +1,13 @@ from __future__ import annotations from datetime import UTC, datetime, timedelta -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock, patch import pytest from agent.dashboard import eval_jobs -@pytest.fixture(autouse=True) -def _clear_procs(): - eval_jobs._PROCS.clear() - yield - eval_jobs._PROCS.clear() - - @pytest.mark.asyncio async def test_get_status_returns_idle_when_no_record() -> None: with patch.object(eval_jobs, "_get_record", new=AsyncMock(return_value=None)): @@ -23,6 +16,18 @@ async def test_get_status_returns_idle_when_no_record() -> None: assert status["name"] == eval_jobs.REVIEWER_EVAL_KEY +@pytest.mark.asyncio +async def test_get_status_returns_terminal_record_unchanged() -> None: + record = {"name": "reviewer", "status": "completed", "experiment_url": "https://x"} + with ( + patch.object(eval_jobs, "_get_record", new=AsyncMock(return_value=record)), + patch.object(eval_jobs, "_put_record", new=AsyncMock(side_effect=lambda r: r)) as put, + ): + status = await eval_jobs.get_reviewer_eval_status() + assert status is record + put.assert_not_awaited() + + @pytest.mark.asyncio async def test_get_status_reconciles_stale_running() -> None: stale = (datetime.now(UTC) - timedelta(seconds=300)).isoformat() @@ -39,7 +44,7 @@ async def test_get_status_reconciles_stale_running() -> None: @pytest.mark.asyncio async def test_get_status_keeps_running_with_fresh_heartbeat() -> None: - """A poll on a worker without the local handle must not kill a live run.""" + """A poll must not kill a run whose Action is still heartbeating.""" fresh = datetime.now(UTC).isoformat() record = {"name": "reviewer", "status": "running", "heartbeat": fresh} with ( @@ -52,125 +57,12 @@ async def test_get_status_keeps_running_with_fresh_heartbeat() -> None: @pytest.mark.asyncio -async def test_start_reviewer_eval_launches_subprocess() -> None: - proc = MagicMock() - proc.pid = 4321 - proc.returncode = None - create = AsyncMock(return_value=proc) - - def _consume(coro): - coro.close() - return MagicMock() - +async def test_get_status_fails_running_without_heartbeat() -> None: + """A running record with no heartbeat at all is treated as stale.""" + record = {"name": "reviewer", "status": "running"} with ( - patch.object(eval_jobs.asyncio, "create_subprocess_exec", new=create), - patch.object(eval_jobs.asyncio, "create_task", new=_consume), - patch.object(eval_jobs, "_get_record", new=AsyncMock(return_value=None)), - patch.object(eval_jobs, "_put_record", new=AsyncMock(side_effect=lambda r: r)), - patch.object(eval_jobs, "_resolve_langgraph_url", return_value="https://lg.test"), - ): - record = await eval_jobs.start_reviewer_eval(limit=3, created_by="octo") - - assert record["status"] == "running" - assert record["limit"] == 3 - assert record["run_name"] == "openswe-review-confidence" - assert record["config_snapshot"]["model_id"] == "google_genai:gemini-3.5-flash" - assert record["pid"] == 4321 - assert record["heartbeat"] is not None - assert record["worker_id"] == eval_jobs._WORKER_ID - assert eval_jobs._PROCS[eval_jobs.REVIEWER_EVAL_KEY] is proc - - args, kwargs = create.call_args - assert "--limit" in args and "3" in args - assert "--experiment-prefix" in args and "openswe-review-confidence" in args - assert "--model-id" in args and "google_genai:gemini-3.5-flash" in args - assert "--reasoning-effort" in args and "medium" in args - assert kwargs["env"]["LANGSMITH_PROJECT"] == eval_jobs.DEFAULT_EVAL_PROJECT - assert kwargs["env"]["LANGGRAPH_URL"] == "https://lg.test" - - -@pytest.mark.asyncio -async def test_start_reviewer_eval_uses_config_snapshot() -> None: - proc = MagicMock() - proc.pid = 4321 - proc.returncode = None - create = AsyncMock(return_value=proc) - config: eval_jobs.ReviewerEvalConfig = { - **eval_jobs.DEFAULT_REVIEWER_EVAL_CONFIG, - "experiment_prefix": "custom-run", - "langsmith_project": "custom-project", - "model_id": "anthropic:claude-opus-4-8", - "reasoning_effort": "high", - "score_mode": "surfaced_findings", - "severity_threshold": "high", - "cap": 2, - } - - def _consume(coro): - coro.close() - return MagicMock() - - with ( - patch.object(eval_jobs.asyncio, "create_subprocess_exec", new=create), - patch.object(eval_jobs.asyncio, "create_task", new=_consume), - patch.object(eval_jobs, "_get_record", new=AsyncMock(return_value=None)), + patch.object(eval_jobs, "_get_record", new=AsyncMock(return_value=record)), patch.object(eval_jobs, "_put_record", new=AsyncMock(side_effect=lambda r: r)), ): - record = await eval_jobs.start_reviewer_eval( - limit=None, - config=config, - created_by="octo", - ) - - assert record["run_name"] == "custom-run" - assert record["langsmith_project"] == "custom-project" - assert record["config_snapshot"] == config - - args, kwargs = create.call_args - assert "--experiment-prefix" in args and "custom-run" in args - assert "--langsmith-project" in args and "custom-project" in args - assert "--model-id" in args and "anthropic:claude-opus-4-8" in args - assert "--score-mode" in args and "surfaced_findings" in args - assert "--severity-threshold" in args and "high" in args - assert "--cap" in args and "2" in args - assert kwargs["env"]["LANGSMITH_PROJECT"] == "custom-project" - - -@pytest.mark.asyncio -async def test_start_reviewer_eval_rejects_when_running() -> None: - running = MagicMock() - running.returncode = None - eval_jobs._PROCS[eval_jobs.REVIEWER_EVAL_KEY] = running - with pytest.raises(RuntimeError): - await eval_jobs.start_reviewer_eval(limit=None, created_by="octo") - - -@pytest.mark.asyncio -async def test_stream_output_keeps_rolling_tail_and_experiment_url() -> None: - url = "https://smith.langchain.com/o/x/experiments/abc" - chunks = [ - f"starting eval {url}\n".encode(), - *[f"row {i} done\n".encode() for i in range(2000)], - b"", - ] - stdout = MagicMock() - stdout.read = AsyncMock(side_effect=chunks) - proc = MagicMock() - proc.stdout = stdout - - tail, experiment_url = await eval_jobs._stream_output(proc) - - assert experiment_url == url - assert len(tail) <= eval_jobs._LOG_TAIL_CHARS - assert tail.endswith("row 1999 done\n") - assert url not in tail # scrolled out of the window but still captured - assert eval_jobs.REVIEWER_EVAL_KEY not in tail - - -@pytest.mark.asyncio -async def test_start_reviewer_eval_rejects_fresh_run_on_other_worker() -> None: - fresh = datetime.now(UTC).isoformat() - record = {"name": "reviewer", "status": "running", "heartbeat": fresh} - with patch.object(eval_jobs, "_get_record", new=AsyncMock(return_value=record)): - with pytest.raises(RuntimeError): - await eval_jobs.start_reviewer_eval(limit=None, created_by="octo") + status = await eval_jobs.get_reviewer_eval_status() + assert status["status"] == "failed" diff --git a/tests/test_eval_store_reporter.py b/tests/test_eval_store_reporter.py new file mode 100644 index 00000000..09ed6021 --- /dev/null +++ b/tests/test_eval_store_reporter.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from evals.reviewer import store_reporter +from evals.reviewer.store_reporter import StoreReporter, github_run_url, is_enabled + +_CONFIG = { + "experiment_prefix": "openswe-review-confidence", + "langsmith_project": "open-swe-evals", + "model_id": "google_genai:gemini-3.5-flash", +} + + +def _make_reporter( + monkeypatch: pytest.MonkeyPatch, completed: int = 0 +) -> tuple[StoreReporter, MagicMock]: + monkeypatch.setenv("LANGGRAPH_URL", "https://lg.test") + monkeypatch.setenv("GITHUB_SERVER_URL", "https://github.com") + monkeypatch.setenv("GITHUB_REPOSITORY", "langchain-ai/open-swe") + monkeypatch.setenv("GITHUB_RUN_ID", "12345") + monkeypatch.setenv("GITHUB_ACTOR", "octocat") + client = MagicMock() + client.store.put_item = AsyncMock() + with patch.object(store_reporter, "get_client", return_value=client): + reporter = StoreReporter( + config=dict(_CONFIG), + limit=3, + total=10, + created_by=None, + completed_getter=lambda: completed, + tail_getter=lambda: "tail", + experiment_url_getter=lambda: "https://smith.langchain.com/exp", + ) + return reporter, client + + +def test_is_enabled(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("REVIEWER_EVAL_REPORT_STORE", raising=False) + monkeypatch.delenv("LANGGRAPH_URL", raising=False) + assert is_enabled() is False + monkeypatch.setenv("REVIEWER_EVAL_REPORT_STORE", "1") + assert is_enabled() is False # still needs LANGGRAPH_URL + monkeypatch.setenv("LANGGRAPH_URL", "https://lg.test") + assert is_enabled() is True + + +def test_github_run_url(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GITHUB_SERVER_URL", "https://github.com") + monkeypatch.setenv("GITHUB_REPOSITORY", "langchain-ai/open-swe") + monkeypatch.setenv("GITHUB_RUN_ID", "999") + assert github_run_url() == "https://github.com/langchain-ai/open-swe/actions/runs/999" + monkeypatch.delenv("GITHUB_RUN_ID") + assert github_run_url() is None + + +@pytest.mark.asyncio +async def test_start_writes_running_record(monkeypatch: pytest.MonkeyPatch) -> None: + reporter, client = _make_reporter(monkeypatch, completed=2) + await reporter.start() + + client.store.put_item.assert_awaited_once() + namespace, key, record = client.store.put_item.await_args.args + assert namespace == ["evals"] + assert key == "reviewer" + assert record["status"] == "running" + assert record["trigger"] == "github_action" + assert record["progress"] == {"completed": 2, "total": 10} + assert record["github_run_url"] == "https://github.com/langchain-ai/open-swe/actions/runs/12345" + assert record["created_by"] == "octocat" # falls back to GITHUB_ACTOR + assert record["worker_id"] == "12345" + assert record["run_name"] == "openswe-review-confidence" + assert record["limit"] == 3 + assert record["heartbeat"] + + +@pytest.mark.asyncio +async def test_finish_writes_terminal_record(monkeypatch: pytest.MonkeyPatch) -> None: + reporter, client = _make_reporter(monkeypatch) + await reporter.finish(status="failed", error="boom") + + _, _, record = client.store.put_item.await_args.args + assert record["status"] == "failed" + assert record["error"] == "boom" + assert record["finished_at"] + + +@pytest.mark.asyncio +async def test_put_swallows_store_errors(monkeypatch: pytest.MonkeyPatch) -> None: + reporter, client = _make_reporter(monkeypatch) + client.store.put_item = AsyncMock(side_effect=RuntimeError("store down")) + # Should not raise — store failures must not crash the eval. + await reporter.start() diff --git a/tests/test_reviewer_eval_target.py b/tests/test_reviewer_eval_target.py index 78d8b968..30fe863b 100644 --- a/tests/test_reviewer_eval_target.py +++ b/tests/test_reviewer_eval_target.py @@ -123,3 +123,10 @@ async def test_extract_surfaced_comments_uses_publish_filter( "severity": "high", } ] + + +def test_completed_counter_increments() -> None: + start = target.get_completed_count() + target._record_completed() + target._record_completed() + assert target.get_completed_count() == start + 2 diff --git a/ui/src/lib/api.ts b/ui/src/lib/api.ts index da19019c..05aa3d59 100644 --- a/ui/src/lib/api.ts +++ b/ui/src/lib/api.ts @@ -448,8 +448,9 @@ export interface ReviewerEvalConfig { cap: number } -export interface ReviewerEvalStartRequest extends ReviewerEvalConfig { - limit: number | null +export interface ReviewerEvalProgress { + completed: number + total: number | null } export interface ReviewerEvalStatus { @@ -467,6 +468,9 @@ export interface ReviewerEvalStatus { experiment_url: string | null error: string | null log_tail: string | null + progress?: ReviewerEvalProgress | null + github_run_url?: string | null + trigger?: string | null updated_at: string } @@ -616,13 +620,6 @@ export const api = { { method: "POST" } ), getReviewerEval: () => request("/admin/evals/reviewer"), - startReviewerEval: (body: ReviewerEvalStartRequest) => - request("/admin/evals/reviewer", { - method: "POST", - body: JSON.stringify(body), - }), - cancelReviewerEval: () => - request("/admin/evals/reviewer", { method: "DELETE" }), logout: () => request("/auth/logout", { method: "POST" }), } diff --git a/ui/src/routes/admin_.evals.tsx b/ui/src/routes/admin_.evals.tsx index acbf9d5b..6c888f49 100644 --- a/ui/src/routes/admin_.evals.tsx +++ b/ui/src/routes/admin_.evals.tsx @@ -1,30 +1,13 @@ import { Navigate, createFileRoute } from "@tanstack/react-router" -import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query" +import { useQuery } from "@tanstack/react-query" import { useEffect, useRef, useState } from "react" -import type { ReactNode } from "react" -import type { - ModelOption, - ReviewerEvalConfig, - ReviewerEvalScoreMode, - ReviewerEvalSeverity, - ReviewerEvalStartRequest, - ReviewerEvalStatus, -} from "@/lib/api" +import type { ReviewerEvalStatus } from "@/lib/api" import { AppShell, SettingsSection } from "@/components/AppShell" import { Button } from "@/components/ui/button" -import { Input } from "@/components/ui/input" -import { - Select, - SelectContent, - SelectItem, - SelectTrigger, - SelectValue, -} from "@/components/ui/select" import { Skeleton } from "@/components/ui/skeleton" import { api } from "@/lib/api" import { useSession } from "@/lib/session" -import { cn } from "@/lib/utils" export const Route = createFileRoute("/admin_/evals")({ component: ReviewerEvalPage }) @@ -45,467 +28,42 @@ function ReviewerEvalPage() { - + ) } -const DEFAULT_REVIEWER_EVAL_CONFIG: ReviewerEvalConfig = { - dataset_name: "openswe-reviewer-v1", - experiment_prefix: "openswe-review-confidence", - max_concurrency: 5, - langsmith_project: "open-swe-evals", - langgraph_url: "", - assistant_id: "reviewer", - model_id: "google_genai:gemini-3.5-flash", - reasoning_effort: "medium", - score_mode: "all_findings", - severity_threshold: "medium", - cap: 4, -} - -interface ReviewerEvalFormState { - dataset_name: string - experiment_prefix: string - max_concurrency: string - langsmith_project: string - langgraph_url: string - assistant_id: string - model_id: string - reasoning_effort: string - score_mode: ReviewerEvalScoreMode - severity_threshold: ReviewerEvalSeverity - cap: string - limit: string -} - -function formFromConfig( - config: ReviewerEvalConfig, - limit: number | null = null -): ReviewerEvalFormState { - return { - dataset_name: config.dataset_name, - experiment_prefix: config.experiment_prefix, - max_concurrency: String(config.max_concurrency), - langsmith_project: config.langsmith_project, - langgraph_url: config.langgraph_url, - assistant_id: config.assistant_id, - model_id: config.model_id, - reasoning_effort: config.reasoning_effort, - score_mode: config.score_mode, - severity_threshold: config.severity_threshold, - cap: String(config.cap), - limit: limit ? String(limit) : "", - } -} - -function parsePositiveInt(label: string, value: string): number { - const n = Number(value.trim()) - if (!Number.isInteger(n) || n <= 0) { - throw new Error(`${label} must be a positive whole number`) - } - return n -} - -function parseOptionalPositiveInt(label: string, value: string): number | null { - return value.trim() ? parsePositiveInt(label, value) : null -} - -function parseNonNegativeInt(label: string, value: string): number { - const n = Number(value.trim()) - if (!Number.isInteger(n) || n < 0) { - throw new Error(`${label} must be a non-negative whole number`) - } - return n -} - -function requireText(label: string, value: string): string { - const text = value.trim() - if (!text) throw new Error(`${label} is required`) - return text -} - -function FieldGroup({ - label, - description, - children, -}: { - label: string - description?: string - children: ReactNode -}) { - return ( -
-
- {label} - {description && ( - {description} - )} -
- {children} -
- ) -} - -function Field({ - label, - className, - children, -}: { - label: string - className?: string - children: ReactNode -}) { - return ( -
- - {label} - - {children} -
- ) -} - -function ReviewerEvalRunner() { - const qc = useQueryClient() - const [draft, setDraft] = useState(() => - formFromConfig(DEFAULT_REVIEWER_EVAL_CONFIG) - ) - const [error, setError] = useState(null) - const initialized = useRef(false) - - const status = useQuery({ +function useReviewerEvalStatus() { + return useQuery({ queryKey: ["reviewerEval"], queryFn: api.getReviewerEval, refetchInterval: (query) => query.state.data?.status === "running" ? 5000 : false, }) - const options = useQuery({ queryKey: ["options"], queryFn: api.options }) - - const data = status.data - const running = data?.status === "running" - const currentModel: ModelOption | undefined = - options.data?.models.find((m) => m.id === draft.model_id) ?? - options.data?.models[0] - - useEffect(() => { - if (initialized.current || !data?.config_snapshot) return - initialized.current = true - setDraft(formFromConfig(data.config_snapshot, data.limit)) - }, [data?.config_snapshot, data?.limit]) - - useEffect(() => { - if (!currentModel) return - if (currentModel.id !== draft.model_id) { - setDraft((current) => ({ - ...current, - model_id: currentModel.id, - reasoning_effort: currentModel.default_effort, - })) - return - } - if (!currentModel.efforts.includes(draft.reasoning_effort)) { - setDraft((current) => ({ - ...current, - reasoning_effort: currentModel.default_effort, - })) - } - }, [currentModel, draft.model_id, draft.reasoning_effort]) - - const setField = ( - key: TKey, - value: ReviewerEvalFormState[TKey] - ) => { - setDraft((current) => ({ ...current, [key]: value })) - } - - const buildRequest = (): ReviewerEvalStartRequest => { - return { - dataset_name: requireText("Dataset", draft.dataset_name), - experiment_prefix: requireText("Run name", draft.experiment_prefix), - max_concurrency: parsePositiveInt("Max concurrency", draft.max_concurrency), - langsmith_project: requireText("LangSmith project", draft.langsmith_project), - langgraph_url: draft.langgraph_url.trim(), - assistant_id: requireText("Assistant ID", draft.assistant_id), - model_id: requireText("Model", draft.model_id), - reasoning_effort: requireText("Effort", draft.reasoning_effort), - score_mode: draft.score_mode, - severity_threshold: draft.severity_threshold, - cap: parseNonNegativeInt("Cap", draft.cap), - limit: parseOptionalPositiveInt("Limit", draft.limit), - } - } - - const onSuccess = (next: ReviewerEvalStatus) => { - qc.setQueryData(["reviewerEval"], next) - setError(null) - } - const onError = (e: Error) => setError(e.message) - - const start = useMutation({ - mutationFn: () => { - return api.startReviewerEval(buildRequest()) - }, - onSuccess, - onError, - }) - const cancel = useMutation({ - mutationFn: () => api.cancelReviewerEval(), - onSuccess, - onError, - }) +} +function ReviewerEvalStatusSection() { + const status = useReviewerEvalStatus() return ( - <> - -
- -
- - setField("experiment_prefix", e.target.value)} - /> - - - setField("dataset_name", e.target.value)} - /> - - - setField("limit", e.target.value)} - /> - - - setField("max_concurrency", e.target.value)} - /> - -
- - -
- - - - - - -
- - -
- - - - - - - - setField("cap", e.target.value)} - /> - -
- - -
- - setField("langsmith_project", e.target.value)} - /> - - - setField("langgraph_url", e.target.value)} - /> - - - setField("assistant_id", e.target.value)} - /> - -
- -
-
- Start - - Only one reviewer eval can run at a time. - -
-
- - {running && ( - - )} -
-
-
- {error &&

{error}

} - - - - - - + + + ) } +function progressLabel(data: ReviewerEvalStatus): string | null { + if (!data.progress) return null + const { completed, total } = data.progress + return `${completed} / ${total ?? "?"}` +} + function ReviewerEvalStatusView({ data }: { data: ReviewerEvalStatus | null }) { if (!data) { return ( @@ -519,6 +77,7 @@ function ReviewerEvalStatusView({ data }: { data: ReviewerEvalStatus | null }) { return (
+ @@ -528,8 +87,7 @@ function ReviewerEvalStatusView({ data }: { data: ReviewerEvalStatus | null }) { - - + {data.started_at && ( )} + {data.github_run_url && ( + + View GitHub run + + )} {data.error && {data.error}}
) @@ -577,12 +145,7 @@ function StatusLine({ } function ReviewerEvalLogs() { - const status = useQuery({ - queryKey: ["reviewerEval"], - queryFn: api.getReviewerEval, - refetchInterval: (query) => - query.state.data?.status === "running" ? 5000 : false, - }) + const status = useReviewerEvalStatus() const logTail = status.data?.log_tail ?? null const running = status.data?.status === "running" @@ -606,7 +169,7 @@ function ReviewerEvalLogs() { return (