diff --git a/metainfer/tasks/evalscope_correctness/__init__.py b/metainfer/tasks/evalscope_correctness/__init__.py new file mode 100644 index 00000000..4c8632b7 --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/__init__.py @@ -0,0 +1,30 @@ +"""evalscope-correctness — evaluate an OpenAI-compatible endpoint with +EvalScope and report correctness results. + +A self-contained task plugin: run EvalScope against a user-supplied, +already-running OpenAI-compatible chat endpoint, preserve the raw EvalScope +artifacts under the workspace, and surface a normalized result (per-dataset +primary score / sample count / optional quality-gate verdict) in the WebUI. + +Design notes +------------ +* This is a **single-run** task — no iteration loop, no shared state graph, + no sub-agent pipeline. The "orchestrator" process here is really a thin + supervisor that runs EvalScope in an isolated child process per dataset. +* Execution **completeness** (did EvalScope evaluate every sample without + truncation / errors / count mismatch) is tracked separately from the + optional **model-quality gate** (minimum per-dataset accuracy/pass@1 the + user may configure). A run that completes but fails its quality gate is + still a *complete* evaluation (``final_status="success"``); only infra / + config / incomplete-result failures surface as ``stopped``. The + authoritative pass/fail for quality lives in ``state_dir/result.json``. +* The API secret is never persisted in ``requirements.json`` — only the name + of an environment variable holding the key is stored, and the key is + copied only into the child process's environment at run time. + +Importing this package registers the orchestrator ``TaskPlugin`` and the +web ``WebPlugin`` (the canonical single discovery point for new task types). +""" + +from .orchestrator import plugin as _task_plugin # noqa: F401 — registers TaskPlugin +from .server import plugin as _web_plugin # noqa: F401 — registers WebPlugin diff --git a/metainfer/tasks/evalscope_correctness/form.yaml b/metainfer/tasks/evalscope_correctness/form.yaml new file mode 100644 index 00000000..439612ee --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/form.yaml @@ -0,0 +1,136 @@ +# Question bank for the `evalscope-correctness` task type. +# +# Runs EvalScope correctness benchmarks against an already-running, +# OpenAI-compatible chat endpoint (e.g. an SGLang server). The task does +# NOT launch or modify the endpoint — it only sends requests and reports +# correctness. +# +# Notes: +# - "file" is not used here; all inputs are text/select/multiselect/number. +# - multiselect / select values are the option LABELS as shown below. +# - Threshold fields are OPTIONAL per-dataset minimums in [0,1]. A dataset +# with no threshold is reported score-only. The quality gate passes only +# when every configured threshold passes. +# - Only the NAME of an env var holding the API key is stored; the secret +# itself is read from the environment at run time and never persisted. + +- key: api_url + question: "OpenAI-compatible endpoint base URL to evaluate (e.g. http://127.0.0.1:30000/v1):" + header: "Endpoint URL" + required: true + form: text + +- key: model + question: "Model name the endpoint serves (sent as the OpenAI `model` field):" + header: "Model" + required: true + form: text + +- key: model_id + question: "Short display id used to name the EvalScope artifact dir (defaults to the model name):" + header: "Model id" + required: false + form: text + +- key: benchmarks + question: "Which built-in EvalScope correctness benchmarks to run? (select one or more)" + header: "Benchmarks" + required: true + form: multiselect + options: + - label: "gsm8k" + description: "Grade-school math — accuracy (mean)." + - label: "gpqa_diamond" + description: "Hard graduate-level QA — accuracy (mean)." + - label: "humaneval" + description: "Code synthesis — pass@1. Requires Docker sandbox execution." + +- key: custom_benchmarks + question: "Additional EvalScope dataset names supported by this install, comma-separated (e.g. arc, mmlu, race). Evaluated alongside the benchmarks above." + header: "Custom datasets" + required: false + form: text + +- key: limit + question: "Optional per-subset sample limit for a quick smoke run (blank = full dataset):" + header: "Sample limit" + required: false + form: number + default: "" + +- key: max_tokens + question: "Max generation tokens per request (raise this if any sample is reported truncated):" + header: "Max tokens" + required: false + form: number + default: 8192 + +- key: timeout_seconds + question: "Per-request timeout in seconds:" + header: "Timeout (s)" + required: false + form: number + default: 300 + +- key: eval_batch_size + question: "Concurrent request batch size (correctness-safe values only):" + header: "Batch size" + required: false + form: select + options: + - label: "1" + description: "Serial — most conservative." + - label: "2" + description: "Small concurrency (default)." + - label: "4" + description: "More concurrency." + +- key: seed + question: "Sampling seed (deterministic, temperature is forced to 0 for correctness):" + header: "Seed" + required: false + form: number + default: 42 + +- key: dataset_cache_dir + question: "Optional EvalScope dataset cache dir (blank = EvalScope default):" + header: "Dataset cache" + required: false + form: text + +- key: api_key_env_var + question: "Name of the environment variable holding the API key (blank = no auth / EvalScope EMPTY). The secret itself is never stored." + header: "API key env" + required: false + form: text + default: "EVALSCOPE_API_KEY" + +# --- Optional per-dataset minimum-score gates (all in [0,1]) ---------- +# Leave blank to report a dataset's score without gating it. + +- key: gate_gsm8k + question: "Minimum GSM8K accuracy to pass (0.0-1.0; blank = report only):" + header: "GSM8K gate" + required: false + form: number + default: "" + +- key: gate_gpqa_diamond + question: "Minimum GPQA-Diamond accuracy to pass (0.0-1.0; blank = report only):" + header: "GPQA gate" + required: false + form: number + default: "" + +- key: gate_humaneval + question: "Minimum HumanEval pass@1 to pass (0.0-1.0; blank = report only):" + header: "HumanEval gate" + required: false + form: number + default: "" + +- key: gate_custom_json + question: "Optional JSON map of per-custom-dataset minimums, e.g. {\"arc\": 0.6, \"mmlu\": 0.7}. Ignored if no custom datasets." + header: "Custom gates" + required: false + form: textarea diff --git a/metainfer/tasks/evalscope_correctness/orchestrator/__init__.py b/metainfer/tasks/evalscope_correctness/orchestrator/__init__.py new file mode 100644 index 00000000..ed421f09 --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/orchestrator/__init__.py @@ -0,0 +1,24 @@ +"""evalscope-correctness orchestrator package. + +Self-contained supervisor + EvalScope runner. The framework +(:mod:`metainfer.orchestrator`) never imports this pipeline directly — it +dispatches to the CLI module declared on :data:`plugin.PLUGIN`. + +Layout:: + + plugin.py TaskPlugin descriptor + cli.py ``run --state-dir … --workspace-dir …`` + config.py parse + validate the evaluation request (from form.yaml) + orchestrator.py supervisor lifecycle: StateStore, PID/signal handling, + per-dataset runner, atomic result.json + runner.py launch an isolated EvalScope child process per dataset + evalscope_worker.py the child: build TaskConfig, run_task, emit a + self-describing ``attempt.json`` (never the secret) + report.py normalize EvalScope reports → result.json (completeness + vs quality-gate separation), all pure + testable +""" + +from metainfer.orchestrator.tasks import register +from .plugin import PLUGIN + +register(PLUGIN) diff --git a/metainfer/tasks/evalscope_correctness/orchestrator/cli.py b/metainfer/tasks/evalscope_correctness/orchestrator/cli.py new file mode 100644 index 00000000..3608eb7d --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/orchestrator/cli.py @@ -0,0 +1,46 @@ +"""CLI entry point for the evalscope-correctness orchestrator subprocess. + +The launcher spawns:: + + python -m metainfer.tasks.evalscope_correctness.orchestrator.cli \\ + run --state-dir … --workspace-dir … + +Contract required by the framework: ``run`` subcommand + ``--state-dir`` and +``--workspace-dir`` flags. +""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="metainfer-evalscope-correctness", + description="MetaInfer EvalScope correctness orchestrator.", + ) + sub = parser.add_subparsers(dest="cmd", required=True) + + run_p = sub.add_parser("run", help="Run the EvalScope correctness evaluation") + run_p.add_argument("requirements", type=Path, help="Path to requirements.json") + run_p.add_argument("--state-dir", type=Path, default=None, + help="Metadata dir (run.json, timeline.jsonl, result.json).") + run_p.add_argument("--workspace-dir", type=Path, default=None, + help="Generated-artifacts dir (raw EvalScope outputs).") + + args = parser.parse_args(argv) + + if args.cmd == "run": + from .orchestrator import run_with_requirements + return run_with_requirements( + requirements_path=args.requirements, + state_dir=args.state_dir, + workspace_dir=args.workspace_dir, + ) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/metainfer/tasks/evalscope_correctness/orchestrator/config.py b/metainfer/tasks/evalscope_correctness/orchestrator/config.py new file mode 100644 index 00000000..2c12fce3 --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/orchestrator/config.py @@ -0,0 +1,364 @@ +"""Parse + validate an evalscope-correctness evaluation request. + +Turns the flat ``requirements.json`` (form answers spread to the top +level, per CLAUDE.md) into a validated :class:`EvalConfig`. All reads go +through :func:`metainfer.orchestrator.requirements.req_field*` so legacy +nested form/answers fixtures keep working. + +Two things this module owns that matter for correctness: + +* **Secret handling.** Only the NAME of an env var holding the API key is + accepted and stored (``api_key_env_var``). The secret itself is never a + form field and never lands in ``requirements.json``; the runner resolves + it from the environment at run time and injects it into the child + process's env only. +* **The restart fingerprint.** :meth:`EvalConfig.fingerprint` covers every + immutable field that determines *what* EvalScope computes. Changing a + quality **gate** or ``model_id`` must NOT invalidate already-complete + evaluations, so those are excluded from the fingerprint. On resume, the + supervisor reuses datasets already evaluated under the same fingerprint + and only evaluates the missing/incomplete ones. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional +from urllib.parse import urlparse + +from metainfer.orchestrator.requirements import ( + req_field, + req_field_float, + req_field_int, +) + +# Preset EvalScope dataset names offered in form.yaml. +PRESET_GSM8K = "gsm8k" +PRESET_GPQA = "gpqa_diamond" +PRESET_HUMANEVAL = "humaneval" +PRESET_IDS = (PRESET_GSM8K, PRESET_GPQA, PRESET_HUMANEVAL) + +# Datasets whose scoring requires executing generated code in a sandbox. +SANDBOX_DATASETS = frozenset({PRESET_HUMANEVAL}) + +# Defaults mirroring the plan's safe correctness posture. +DEFAULT_MAX_TOKENS = 8192 +DEFAULT_TIMEOUT_SECONDS = 300 +DEFAULT_SEED = 42 +DEFAULT_KEY_ENV_VAR = "EVALSCOPE_API_KEY" +ALLOWED_BATCH_SIZES = (1, 2, 4) +TEMPERATURE = 0.0 + +# Env-var name pattern (for api_key_env_var). We accept a NAME only. +_ENV_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") +# Custom dataset name pattern (anything EvalScope could reasonably address). +_DATASET_NAME_RE = re.compile(r"^[A-Za-z0-9_\-\.]+$") + +# Which form gate key feeds which preset. +_PRESET_GATE_KEY = { + PRESET_GSM8K: "gate_gsm8k", + PRESET_GPQA: "gate_gpqa_diamond", + PRESET_HUMANEVAL: "gate_humaneval", +} + + +class ConfigError(ValueError): + """Raised when a request cannot be turned into a valid evaluation.""" + + +@dataclass(frozen=True) +class EvalTarget: + """One dataset to evaluate. + + ``dataset`` is the exact EvalScope dataset ``name`` passed to + ``TaskConfig.datasets``. ``gate`` is an optional minimum score in + [0, 1] (None = report-only for this dataset). ``needs_sandbox`` flags + datasets whose reference scoring executes generated code. + """ + + dataset: str + gate: Optional[float] = None + needs_sandbox: bool = False + + +@dataclass +class EvalConfig: + """Validated evaluation request. + + Fields are deliberately plain so the module is pure and testable with + no dependency on EvalScope itself. + """ + + api_url: str + model: str + targets: List[EvalTarget] + model_id: str = "" + seed: int = DEFAULT_SEED + temperature: float = TEMPERATURE + eval_batch_size: int = 2 + timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS + max_tokens: int = DEFAULT_MAX_TOKENS + limit: Optional[int] = None + dataset_cache_dir: str = "" + api_key_env_var: str = DEFAULT_KEY_ENV_VAR + gate_custom_json: Dict[str, float] = field(default_factory=dict) + + # -- derived queries -------------------------------------------------- # + + @property + def dataset_ids(self) -> List[str]: + """Requested EvalScope dataset names, in request order.""" + return [t.dataset for t in self.targets] + + @property + def requires_key(self) -> bool: + """True if the request needs a real API key (env var named + set).""" + return bool(self.api_key_env_var) + + def gate_for(self, dataset: str) -> Optional[float]: + for t in self.targets: + if t.dataset == dataset: + return t.gate + return None + + def fingerprint(self) -> str: + """Stable id of the immutable eval request. + + Restart semantics: an evaluation already complete for this + fingerprint is reused on resume. Quality gates and ``model_id`` are + intentionally excluded — adjusting a minimum score must not force a + re-run of an already-complete dataset. + """ + canon = { + "api_url": self.api_url, + "model": self.model, + "datasets": sorted(self.dataset_ids), + "seed": int(self.seed), + "temperature": float(self.temperature), + "max_tokens": int(self.max_tokens), + "eval_batch_size": int(self.eval_batch_size), + "limit": int(self.limit) if self.limit is not None else None, + } + blob = json.dumps(canon, sort_keys=True, separators=(",", ":")).encode("utf-8") + return "sha256:" + hashlib.sha256(blob).hexdigest() + + def to_child_json(self) -> Dict[str, Any]: + """Non-secret config passed to the isolated EvalScope child. + + Never contains the API key — only ``api_key_env_var`` (the name). + The child reads the actual secret from its own environment. + """ + return { + "api_url": self.api_url, + "model": self.model, + "model_id": self.model_id or self.model, + "datasets": self.dataset_ids, + "temperature": self.temperature, + "seed": self.seed, + "eval_batch_size": self.eval_batch_size, + "timeout_seconds": self.timeout_seconds, + "max_tokens": self.max_tokens, + "limit": self.limit, + "dataset_cache_dir": self.dataset_cache_dir, + "api_key_env_var": self.api_key_env_var, + } + + +# --------------------------------------------------------------------------- # +# Parsing helpers +# --------------------------------------------------------------------------- # + +def _split_csv(value: Any) -> List[str]: + """Split a possibly-list / comma-separated string into trimmed pieces.""" + if value is None: + return [] + if isinstance(value, str): + parts = value.split(",") + elif isinstance(value, (list, tuple, set)): + parts = value + else: + raise ConfigError(f"expected list or comma-separated string, got {type(value).__name__}") + return [str(p).strip() for p in parts if str(p).strip()] + + +def _require_text(req: Dict[str, Any], key: str) -> str: + v = req_field(req, key) + if v is None or not str(v).strip(): + raise ConfigError(f"'{key}' is required") + return str(v).strip() + + +def _validate_gate(value: Any, label: str) -> Optional[float]: + """Validate an optional gate; returns None when blank/unset.""" + if value is None or str(value).strip() == "": + return None + try: + f = float(value) + except (TypeError, ValueError): + raise ConfigError(f"{label} must be a number in [0, 1]") + if not (0.0 <= f <= 1.0): + raise ConfigError(f"{label} must be in [0, 1], got {f}") + return f + + +def _parse_gate_custom(req: Dict[str, Any]) -> Dict[str, float]: + raw = req_field(req, "gate_custom_json") + if raw is None or str(raw).strip() == "": + return {} + if isinstance(raw, dict): + data = raw + else: + text = str(raw).strip() + try: + data = json.loads(text) + except ValueError as exc: + raise ConfigError(f"gate_custom_json is not valid JSON: {exc}") + if not isinstance(data, dict): + raise ConfigError("gate_custom_json must be a JSON object mapping dataset -> gate") + out: Dict[str, float] = {} + for k, v in data.items(): + gate = _validate_gate(v, f"custom gate for '{k}'") + out[str(k)] = gate + return out + + +def parse_requirements(req: Dict[str, Any]) -> EvalConfig: + """Validate + normalize a request into an :class:`EvalConfig`. + + Raises :class:`ConfigError` with a human-readable message on any + invalid input. The caller turns a ConfigError into a ``stopped`` + run rather than a crash. + """ + if not isinstance(req, dict): + raise ConfigError("requirements must be a JSON object") + + api_url = _require_text(req, "api_url") + _validate_url(api_url) + model = _require_text(req, "model") + + model_id = str(req_field(req, "model_id") or "").strip() + + # -- dataset selection ------------------------------------------------ # + preset_labels = _split_csv(req_field(req, "benchmarks")) + # Normalize selected preset labels to their dataset ids (hand-writers may + # use the id directly; the form emits the label, which equals the id). + presets = [p for p in preset_labels if p in PRESET_IDS] + unknown = [p for p in preset_labels if p not in PRESET_IDS] + if unknown: + raise ConfigError(f"unknown benchmark selections: {', '.join(unknown)}") + + custom_names = [] + for name in _split_csv(req_field(req, "custom_benchmarks")): + if name in PRESET_IDS: + # Already selected via the preset list; dedupe quietly. + if name not in presets: + presets.append(name) + continue + if not _DATASET_NAME_RE.match(name): + raise ConfigError( + f"custom benchmark name {name!r} has invalid characters " + "(allowed: letters, digits, _ - .)" + ) + if name not in custom_names: + custom_names.append(name) + + if not presets and not custom_names: + raise ConfigError("select at least one benchmark (or a custom dataset)") + + # -- gates ------------------------------------------------------------- # + gate_custom = _parse_gate_custom(req) + if custom_names: + # A custom gate key that references a non-requested dataset is a typo. + for name in gate_custom: + if name not in custom_names: + raise ConfigError( + f"custom gate for '{name}' references a dataset that was not requested" + ) + elif gate_custom: + raise ConfigError("gate_custom_json given but no custom benchmarks requested") + + targets: List[EvalTarget] = [] + for preset in presets: + gate = _validate_gate(req_field(req, _PRESET_GATE_KEY[preset]), f"gate for {preset}") + targets.append( + EvalTarget( + dataset=preset, + gate=gate, + needs_sandbox=preset in SANDBOX_DATASETS, + ) + ) + for name in custom_names: + targets.append( + EvalTarget( + dataset=name, + gate=gate_custom.get(name), + needs_sandbox=False, + ) + ) + + # -- numerics ---------------------------------------------------------- # + max_tokens = req_field_int(req, "max_tokens", DEFAULT_MAX_TOKENS) + if max_tokens is not None and max_tokens < 1: + raise ConfigError("max_tokens must be >= 1") + + timeout = req_field_int(req, "timeout_seconds", DEFAULT_TIMEOUT_SECONDS) + if timeout is None or timeout < 1: + raise ConfigError("timeout_seconds must be >= 1") + + batch = req_field_int(req, "eval_batch_size", 2) + if batch not in ALLOWED_BATCH_SIZES: + raise ConfigError(f"eval_batch_size must be one of {list(ALLOWED_BATCH_SIZES)}") + + seed = req_field_int(req, "seed", DEFAULT_SEED) + if seed is None: + seed = DEFAULT_SEED + + limit_raw = req_field(req, "limit") + limit: Optional[int] = None + if limit_raw is not None and str(limit_raw).strip() != "": + limit = req_field_int(req, "limit") + if limit is None or limit < 1: + raise ConfigError("limit must be a positive integer") + + cache_dir = str(req_field(req, "dataset_cache_dir") or "").strip() + + # -- auth -------------------------------------------------------------- # + key_env = str(req_field(req, "api_key_env_var") or "").strip() + if key_env == "": + key_env = "" # no auth + elif not _ENV_NAME_RE.match(key_env): + raise ConfigError( + "api_key_env_var must be the NAME of an environment variable " + "(letters/digits/_ only), not a secret value" + ) + + return EvalConfig( + api_url=api_url, + model=model, + targets=targets, + model_id=model_id, + seed=int(seed), + temperature=TEMPERATURE, + eval_batch_size=int(batch), + timeout_seconds=int(timeout), + max_tokens=int(max_tokens), + limit=limit, + dataset_cache_dir=cache_dir, + api_key_env_var=key_env, + gate_custom_json=gate_custom, + ) + + +def _validate_url(url: str) -> None: + parsed = urlparse(url) + if parsed.scheme not in ("http", "https") or not parsed.netloc: + raise ConfigError(f"api_url must be an http(s) URL, got {url!r}") + + +def safe_basename(value: str) -> str: + """Collapse a free-text model id into a filesystem-safe artifact name.""" + cleaned = re.sub(r"[^A-Za-z0-9_.\-]", "_", value).strip("._") + return cleaned or "model" diff --git a/metainfer/tasks/evalscope_correctness/orchestrator/evalscope_worker.py b/metainfer/tasks/evalscope_correctness/orchestrator/evalscope_worker.py new file mode 100644 index 00000000..28b7666e --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/orchestrator/evalscope_worker.py @@ -0,0 +1,131 @@ +"""EvalScope child worker — runs one dataset evaluation in isolation. + +The supervisor launches this module as a subprocess *per dataset*: + + python -m metainfer.tasks.evalscope_correctness.orchestrator.evalscope_worker + +Non-secret config arrives on **stdin** as one JSON line; the API key is +delivered only via the child's ``EVALSCOPE_API_KEY`` environment variable +(or ``EVALSCOPE_*_KEY`` matching ``api_key_env_var``). The worker never +reads, writes, or logs the secret. + +Keeping EvalScope in an isolated child process (rather than calling +``run_task`` inside the orchestrator) protects MetaInfer's process from +EvalScope's heavy imports / logging reconfiguration / threads, and gives +the supervisor a hard wall-clock + signal boundary around each dataset. + +Design invariants: +* Runs exactly the datasets listed in the stdin config (the supervisor + already narrowed it to one) and writes results into ``work_dir``. +* Exit code 0 == EvalScope reported success. Any exception is printed to + stderr (never the secret) and surfaced as a nonzero exit so the + supervisor records the dataset as not-produced. +""" + +from __future__ import annotations + +import json +import sys +from typing import Any, Dict, Optional + + +def _load_config() -> Dict[str, Any]: + """Read the non-secret config JSON from stdin.""" + raw = sys.stdin.read() + try: + cfg = json.loads(raw) + except ValueError as exc: + raise RuntimeError(f"invalid config on stdin: {exc}") from exc + if not isinstance(cfg, dict): + raise RuntimeError("config on stdin must be a JSON object") + return cfg + + +def _build_task_config(cfg: Dict[str, Any]) -> Dict[str, Any]: + """Construct the EvalScope ``TaskConfig`` payload (as a dict). + + We pass a plain dict to :func:`evalscope.run_task`, which coerces it + into a ``TaskConfig`` internally — so the worker needs no direct import + of pydantic ``SecretStr`` or the config class. + """ + api_key = _resolve_api_key(cfg.get("api_key_env_var")) + + task: Dict[str, Any] = { + "model": cfg["model"], + "model_id": cfg.get("model_id") or cfg["model"], + "api_url": cfg["api_url"], + "api_key": api_key, # SecretStr is derived by TaskConfig's validator + "eval_type": "openai_api", + "eval_backend": "Native", + "datasets": list(cfg["datasets"]), + # limit is per-subset; None = full dataset. + "limit": cfg.get("limit"), + # Remote (openai_api) defaults eval_batch_size to 8 unless set — + # correctness evals force an explicit, conservative value. + "eval_batch_size": int(cfg.get("eval_batch_size") or 1), + "seed": int(cfg.get("seed") or 42), + "no_timestamp": True, + "work_dir": cfg["work_dir"], + "collect_perf": False, + "generation_config": { + "temperature": float(cfg.get("temperature") or 0.0), + "max_tokens": int(cfg.get("max_tokens") or 8192), + "timeout": float(cfg.get("timeout_seconds") or 300.0), + }, + } + if cfg.get("dataset_cache_dir"): + task["dataset_dir"] = cfg["dataset_cache_dir"] + if cfg.get("needs_sandbox"): + # Reference scoring executes generated code (e.g. HumanEval). The + # supervisor preflights Docker availability; enabling the sandbox + # here keeps execution off the MetaInfer host. + task["use_sandbox"] = True + return task + + +def _resolve_api_key(env_var: Optional[str]) -> str: + """Return the key to send, or ``EMPTY`` for an unauthenticated endpoint. + + The secret (if any) is read from this child's own environment — the + supervisor copied it here. It is never part of the config that crossed + stdin. + """ + if env_var: + import os + value = os.environ.get(env_var) + if value: + return value + return "EMPTY" + + +def main(argv: Optional[list] = None) -> int: + try: + cfg = _load_config() + task = _build_task_config(cfg) + except Exception as exc: # noqa: BLE001 — surface any config error + print(f"[evalscope-worker] config error: {exc}", file=sys.stderr) + return 2 + + try: + from evalscope import run_task # lazy: heavy import stays in this child + except Exception as exc: # noqa: BLE001 + print( + f"[evalscope-worker] EvalScope is not importable here: {exc}. " + "Install with: pip install 'evalscope>=1.11,<2'", + file=sys.stderr, + ) + return 3 + + try: + run_task(task) + except Exception as exc: # noqa: BLE001 — turn any failure into exit code + # Never print the api key; the config/exception may embed task dicts, + # so only the exception type + message go to stderr. + print(f"[evalscope-worker] run failed: {type(exc).__name__}: {exc}", + file=sys.stderr) + return 4 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/metainfer/tasks/evalscope_correctness/orchestrator/orchestrator.py b/metainfer/tasks/evalscope_correctness/orchestrator/orchestrator.py new file mode 100644 index 00000000..14e5be71 --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/orchestrator/orchestrator.py @@ -0,0 +1,429 @@ +"""Supervisor + lifecycle for the evalscope-correctness orchestrator. + +This task has no iteration loop and no sub-agents. The "orchestrator" +is a thin supervisor that, for each requested dataset, runs EvalScope in +an isolated child process (see :mod:`.runner` / :mod:`.evalscope_worker`), +then normalizes the raw reports into the single authoritative +``state_dir/result.json``. + +Lifecycle mirrors the framework conventions (see +:mod:`metainfer.orchestrator._bootstrap` and ``calc_value/orchestrator``): +write the PID file, install SIGTERM/SIGINT handlers, drive the pipeline, +clear the PID file on exit. Because our child stays in the orchestrator's +process group (no ``start_new_session``), the launcher's group-kill stops +EvalScope together with this supervisor. + +Result semantics (see :mod:`.report`): +* ``result.json`` is the single authority for correctness outcome. +* ``run.json.final_status`` is ``success`` when the evaluation is *complete* + (every dataset fully evaluated, no truncation/errors/mismatch) — even if + an optional quality gate failed. Incomplete / config / infra outcomes are + ``stopped``. + +State layout:: + + / requirements.json, run.json, timeline.jsonl, + orchestrator.{pid,log}, result.json + /evalscope//attempt-/ (raw EvalScope) +""" + +from __future__ import annotations + +import json +import os +import signal +import time +from pathlib import Path +from typing import Any, Dict, List, Optional + +from metainfer.orchestrator._bootstrap import ( + clear_pid_file, + set_process_name, + write_pid_file, +) +from metainfer.orchestrator.state import StateStore + +from . import config as _config +from . import report as _report +from . import runner as _runner + +# Phase tokens the shell stores opaquely (see CLAUDE.md). +PHASE_CONFIGURING = "configuring" +PHASE_RUNNING = "running" +PHASE_FINALIZING = "finalizing" +PHASE_DONE = "complete" +PHASE_STOPPED = "stopped" + + +# --------------------------------------------------------------------------- # +# Attempt-dir helpers +# --------------------------------------------------------------------------- # + +def _read_attempt_meta(attempt_dir: Path) -> Dict[str, Any]: + path = attempt_dir / "attempt.json" + if not path.exists(): + return {} + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (ValueError, OSError): + return {} + return data if isinstance(data, dict) else {} + + +def _write_attempt_meta(attempt_dir: Path, meta: Dict[str, Any]) -> None: + path = attempt_dir / "attempt.json" + tmp = path.with_suffix(".json.tmp") + tmp.write_text(json.dumps(meta, indent=2), encoding="utf-8") + tmp.replace(path) + + +def _attempt_number(attempt_dir: Path) -> int: + try: + return int(attempt_dir.name.rsplit("-", 1)[1]) + except (IndexError, ValueError): + return 0 + + +def _next_attempt_dir(evalscope_root: Path, dataset: str) -> Path: + existing = evalscope_root.glob(f"{dataset}/attempt-*") + highest = max((_attempt_number(p) for p in existing), default=0) + return evalscope_root / dataset / f"attempt-{highest + 1}" + + +def _latest_complete_attempt( + evalscope_root: Path, + dataset: str, + fingerprint: str, + threshold: Optional[float], +) -> Optional[Path]: + """Newest attempt matching the fingerprint that normalizes complete. + + Used for restart reuse: a dataset already fully evaluated under the same + immutable request fingerprint is not re-run. Completeness is derived + fresh from the on-disk evidence (SSOT), not from a cached flag. + """ + best: Optional[Path] = None + best_n = -1 + for attempt_dir in sorted(evalscope_root.glob(f"{dataset}/attempt-*")): + meta = _read_attempt_meta(attempt_dir) + if meta.get("fingerprint") != fingerprint: + continue + if meta.get("exit_code") not in (0, None): + continue # EvalScope did not finish cleanly; re-run. + row = _report.normalize_attempt_dir( + attempt_dir, dataset, threshold=threshold + ) + n = _attempt_number(attempt_dir) + if row.get("complete") and n > best_n: + best, best_n = attempt_dir, n + return best + + +# --------------------------------------------------------------------------- # +# Request → datasets +# --------------------------------------------------------------------------- # + +def _resolve_api_key(cfg: _config.EvalConfig) -> Optional[Dict[str, str]]: + """Map the env var holding the key → (var, value) for the child env. + + Returns None (no auth) when no env var is configured, or when the + configured var is unset in this process (worker falls back to EMPTY). + """ + if not cfg.api_key_env_var: + return None + value = os.environ.get(cfg.api_key_env_var) + if value is None: + return None + return {cfg.api_key_env_var: value} + + +def run_with_requirements( + requirements_path: Path, + *, + state_dir: Optional[Path] = None, + workspace_dir: Optional[Path] = None, +) -> int: + """Per-task orchestrator entry point. Returns the process exit code.""" + if not requirements_path.exists(): + raise FileNotFoundError(f"requirements file not found: {requirements_path}") + + req: Dict[str, Any] = json.loads( + requirements_path.read_text(encoding="utf-8") + ) + task_id = req.get("task_id", "task") + + if state_dir is None or workspace_dir is None: + from metainfer.server import paths as _web_paths + if state_dir is None: + state_dir = _web_paths.task_dir(task_id) + if workspace_dir is None: + workspace_dir = _web_paths.workspace_dir(task_id) + + state_dir.mkdir(parents=True, exist_ok=True) + workspace_dir.mkdir(parents=True, exist_ok=True) + + # Copy requirements into state_dir for self-containment. + target_req = state_dir / "requirements.json" + if requirements_path.resolve() != target_req.resolve(): + target_req.write_text( + requirements_path.read_text(encoding="utf-8"), encoding="utf-8" + ) + + set_process_name("metainfer-esc-och") + write_pid_file(state_dir / "orchestrator.pid", task_id) + + # Holder so the signal handler can terminate the running child. + active: Dict[str, Any] = {"proc": None} + + def _on_signal(signum, _frame): + proc = active.get("proc") + if proc is not None and proc.poll() is None: + try: + proc.terminate() + try: + proc.wait(timeout=5) + except Exception: # noqa: BLE001 + proc.kill() + except Exception: # noqa: BLE001 — best-effort + pass + try: + clear_pid_file(state_dir / "orchestrator.pid") + except Exception: # noqa: BLE001 + pass + os._exit(143 if signum == signal.SIGTERM else 130) + + prev_term = signal.signal(signal.SIGTERM, _on_signal) + prev_int = signal.signal(signal.SIGINT, _on_signal) + + store = StateStore(state_dir) + _rs, is_resume = store.init_or_resume(task_id) + + # ---- Input validation (invalid immutable inputs → stopped) ----------- # + try: + cfg = _config.parse_requirements(req) + except _config.ConfigError as exc: + store.update_run( + current_phase=PHASE_STOPPED, + finished=True, + final_status="stopped", + last_transition_label=f"input validation: {exc}", + ) + store.append_timeline("evalscope.stop.invalid_input", {"error": str(exc)}) + clear_pid_file(state_dir / "orchestrator.pid") + _restore_signals(prev_term, prev_int) + return 2 + + store.update_run(current_phase=PHASE_CONFIGURING) + store.append_timeline( + "evalscope.start", + { + "task_id": task_id, + "resume": is_resume, + "datasets": cfg.dataset_ids, + "fingerprint": cfg.fingerprint(), + "api_url": cfg.api_url, + "model": cfg.model, + }, + ) + + # ---- Preflight (EvalScope install; Docker for sandboxed datasets) ---- # + pf_error = _runner.preflight_for(cfg) + if pf_error is not None: + store.update_run( + current_phase=PHASE_STOPPED, + finished=True, + final_status="stopped", + last_transition_label=f"preflight: {pf_error}", + ) + store.append_timeline("evalscope.stop.preflight", {"error": pf_error}) + clear_pid_file(state_dir / "orchestrator.pid") + _restore_signals(prev_term, prev_int) + return 2 + + evalscope_root = workspace_dir / "evalscope" + evalscope_root.mkdir(parents=True, exist_ok=True) + fingerprint = cfg.fingerprint() + env_extra = _resolve_api_key(cfg) + + chosen: Dict[str, Optional[Path]] = {} + + try: + for target in cfg.targets: + store.update_run( + current_phase=PHASE_RUNNING, + current_iteration=int(cfg.targets.index(target)) + 1, + last_transition_label=f"evaluating {target.dataset}", + ) + attempt_dir = _latest_complete_attempt( + evalscope_root, target.dataset, fingerprint, cfg.gate_for(target.dataset) + ) + if attempt_dir is not None: + store.append_timeline( + "evalscope.dataset.reused", + {"dataset": target.dataset, "attempt": attempt_dir.name}, + ) + chosen[target.dataset] = attempt_dir + continue + + attempt_dir = _next_attempt_dir(evalscope_root, target.dataset) + started = time.time() + store.append_timeline( + "evalscope.dataset.start", + {"dataset": target.dataset, "attempt": attempt_dir.name}, + ) + result = _runner.run_dataset( + cfg, + target, + attempt_dir=attempt_dir, + log_file=attempt_dir / "worker.log", + active=active, + env_extra=env_extra, + ) + _write_attempt_meta( + attempt_dir, + { + "dataset": target.dataset, + "fingerprint": fingerprint, + "exit_code": result.exit_code, + "pid": result.pid, + "started_at": started, + "finished_at": time.time(), + }, + ) + store.append_timeline( + "evalscope.dataset.finished", + { + "dataset": target.dataset, + "attempt": attempt_dir.name, + "exit_code": result.exit_code, + }, + ) + chosen[target.dataset] = attempt_dir + except Exception as exc: # noqa: BLE001 — infra crash mid-run + store.update_run( + current_phase=PHASE_STOPPED, + finished=True, + final_status="stopped", + last_transition_label=f"crash: {type(exc).__name__}: {exc}", + ) + store.append_timeline("evalscope.crash", {"error": str(exc)}) + clear_pid_file(state_dir / "orchestrator.pid") + _restore_signals(prev_term, prev_int) + return 1 + + # ---- Finalize: build result.json + mark run --------------------------- # + store.update_run(current_phase=PHASE_FINALIZING) + result = _finalize( + cfg=cfg, + evalscope_root=evalscope_root, + chosen=chosen, + workspace_dir=workspace_dir, + ) + _write_result_atomic(state_dir / "result.json", result) + + overall_complete = result["complete"] + if overall_complete: + label = _success_label(result) + final_status = "success" + else: + label = "incomplete result: " + "; ".join(result["complete_reasons"]) + final_status = "stopped" + + store.update_run( + current_phase=PHASE_DONE if overall_complete else PHASE_STOPPED, + finished=True, + final_status=final_status, + last_transition_label=label, + ) + store.append_timeline( + "evalscope.finish", + { + "complete": overall_complete, + "final_status": final_status, + "quality_passed": result["quality"]["passed"], + "quality_configured": result["quality"]["configured"], + }, + ) + clear_pid_file(state_dir / "orchestrator.pid") + _restore_signals(prev_term, prev_int) + return 0 if overall_complete else 2 + + +def _finalize( + *, + cfg: _config.EvalConfig, + evalscope_root: Path, + chosen: Dict[str, Optional[Path]], + workspace_dir: Path, +) -> Dict[str, Any]: + """Normalize every target's chosen attempt into the authoritative result.""" + rows: List[Dict[str, Any]] = [] + for target in cfg.targets: + attempt_dir = chosen.get(target.dataset) + if attempt_dir is None: + rows.append( + _report.missing_dataset_row( + target.dataset, threshold=cfg.gate_for(target.dataset) + ) + ) + continue + try: + rel = attempt_dir.relative_to(workspace_dir) + except ValueError: + rel = Path(attempt_dir.name) + rows.append( + _report.normalize_attempt_dir( + attempt_dir, + target.dataset, + threshold=cfg.gate_for(target.dataset), + raw_relative=str(rel), + ) + ) + + complete = all(r["complete"] for r in rows) + complete_reasons: List[str] = [] + for r in rows: + if not r["complete"]: + complete_reasons.extend(r["reasons"]) + + gated = [r for r in rows if r.get("threshold") is not None] + quality_configured = bool(gated) + if gated: + quality_passed = all( + r.get("complete") and r.get("threshold_met") is True for r in gated + ) + else: + quality_passed = None + + return { + "schema_version": 1, + "task_type": "evalscope-correctness", + "fingerprint": cfg.fingerprint(), + "complete": complete, + "complete_reasons": complete_reasons, + "quality": { + "configured": quality_configured, + "passed": quality_passed, + }, + "datasets": rows, + } + + +def _success_label(result: Dict[str, Any]) -> str: + q = result["quality"] + if not q["configured"]: + return "evaluation complete (no quality gates configured)" + if q["passed"]: + return "evaluation complete; all quality gates passed" + return "evaluation complete; quality gate NOT met (score below configured minimum)" + + +def _write_result_atomic(path: Path, result: Dict[str, Any]) -> None: + tmp = path.with_suffix(".tmp") + tmp.write_text(json.dumps(result, indent=2), encoding="utf-8") + tmp.replace(path) + + +def _restore_signals(prev_term, prev_int) -> None: + signal.signal(signal.SIGTERM, prev_term) + signal.signal(signal.SIGINT, prev_int) diff --git a/metainfer/tasks/evalscope_correctness/orchestrator/plugin.py b/metainfer/tasks/evalscope_correctness/orchestrator/plugin.py new file mode 100644 index 00000000..198c0c27 --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/orchestrator/plugin.py @@ -0,0 +1,16 @@ +"""TaskPlugin descriptor for evalscope-correctness. + +This task has no iteration loop and no shared state graph, so +``phases_module`` is empty (allowed — see :class:`TaskPlugin`) and +``diagnostic_globs`` is empty (nothing is copied forward between +iterations that never exist). +""" + +from metainfer.orchestrator.tasks.base import TaskPlugin + +PLUGIN = TaskPlugin( + task_type="evalscope-correctness", + cli_module="metainfer.tasks.evalscope_correctness.orchestrator.cli", + phases_module="", + diagnostic_globs=(), +) diff --git a/metainfer/tasks/evalscope_correctness/orchestrator/report.py b/metainfer/tasks/evalscope_correctness/orchestrator/report.py new file mode 100644 index 00000000..d6ce823c --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/orchestrator/report.py @@ -0,0 +1,427 @@ +"""Normalize EvalScope outputs into ``result.json``. + +The single authority for a run's correctness outcome is +``state_dir/result.json`` (written atomically by the supervisor). Raw +EvalScope files under the workspace are immutable evidence only — this +module reads them and derives the normalized result, but never mutates +them. + +Two independent judgments live in the result: + +* **Completeness** (``complete``) — did EvalScope evaluate every sample + cleanly? Fails on a malformed prediction row, a per-row model error, a + missing/unschema'd report, a sample-count mismatch, or any + ``choices[].stop_reason == "length"`` truncation. A truncated sample must + be re-run with a higher ``max_tokens``; it must never silently contribute + to a final correctness number. +* **Quality gate** (``quality``) — optional per-dataset minimums the user + configured. A complete run may still FAIL its quality gate; that is a + *legitimate result*, not a run failure. Only infra/config/incomplete + outcomes make ``run.json.final_status == "stopped"``. + +Primary-metric selection is by **full identity** (name + aggregation + +dimensions), never by ``metrics[0]`` order. For the preset datasets the rule +is fixed; for custom datasets we honor the report's own +``primary_metric_identity`` when it resolves to a real metric, else the +first metric. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple + +# Human-readable metric id → EvalScope metric identity. +ACC_MEAN = {"name": "accuracy", "aggregation": "mean", "dimensions": {}} +PASS_AT_1 = {"name": "accuracy", "aggregation": "pass_at_k", "dimensions": {"k": 1}} + +# Fixed primary-metric rules for the built-in presets. +PRESET_PRIMARY_IDENTITY = { + "gsm8k": ACC_MEAN, + "gpqa_diamond": ACC_MEAN, + "humaneval": PASS_AT_1, +} + +# Report schema versions this normalizer understands. +SUPPORTED_SCHEMA_VERSION = 2 + + +class ReportError(ValueError): + """A report/prediction set could not be normalized into a dataset row.""" + + +# --------------------------------------------------------------------------- # +# Identity helpers +# --------------------------------------------------------------------------- # + +def identity_key(identity: Optional[Dict[str, Any]]) -> Tuple: + """Stable, comparable key for a metric identity dict.""" + if not isinstance(identity, dict): + return () + dims = identity.get("dimensions") or {} + return ( + identity.get("name"), + identity.get("aggregation"), + tuple(sorted((str(k), str(v)) for k, v in dims.items())), + ) + + +def find_metric( + metrics: Sequence[Dict[str, Any]], identity: Dict[str, Any] +) -> Optional[Dict[str, Any]]: + """Return the metric whose identity matches ``identity`` exactly.""" + want = identity_key(identity) + for m in metrics: + if identity_key(m.get("identity")) == want: + return m + return None + + +def metric_id(identity: Dict[str, Any]) -> str: + """Short human label, e.g. ``accuracy/mean`` or ``accuracy/pass_at_k(k=1)``.""" + agg = identity.get("aggregation", "?") + dims = identity.get("dimensions") or {} + if dims: + extra = ", ".join(f"{k}={v}" for k, v in sorted(dims.items())) + return f"{identity.get('name', '?')}/{agg}({extra})" + return f"{identity.get('name', '?')}/{agg}" + + +def primary_identity_for( + report: Dict[str, Any], dataset: str +) -> Optional[Dict[str, Any]]: + """Resolve the primary metric identity for a report. + + Order: + 1. The report's own ``primary_metric_identity``, if it resolves to a + real metric in ``report['metrics']``. + 2. The fixed preset rule (gsm8k / gpqa_diamond / humaneval). + 3. ``metrics[0]`` identity (custom datasets without a primary id). + """ + metrics = report.get("metrics") or [] + declared = report.get("primary_metric_identity") + if isinstance(declared, dict) and find_metric(metrics, declared) is not None: + return declared + preset = PRESET_PRIMARY_IDENTITY.get(dataset) + if preset is not None: + return preset + if metrics: + ident = metrics[0].get("identity") + if isinstance(ident, dict): + return ident + return None + + +def extract_primary( + report: Dict[str, Any], dataset: str +) -> Tuple[Optional[Dict[str, Any]], Optional[float], Optional[int]]: + """Return ``(identity, score, num)`` for the primary metric. + + ``score``/``num`` are None if the primary metric cannot be located or + has no score — the caller treats that as an incomplete dataset. + """ + identity = primary_identity_for(report, dataset) + if identity is None: + return None, None, None + metric = find_metric(report.get("metrics") or [], identity) + if metric is None: + return identity, None, None + score = metric.get("score") + if not isinstance(score, (int, float)): + score = None + return identity, score, metric.get("num") + + +# --------------------------------------------------------------------------- # +# Prediction-row analysis +# --------------------------------------------------------------------------- # + +def classify_row(row: Any) -> str: + """Classify one parsed prediction row: ``ok`` | ``error`` | ``truncated``. + + An explicit model error beats truncation; otherwise a stop_reason of + ``"length"`` marks the row truncated. A non-dict row is ``malformed`` + (a defensive case — the loader already filters to JSON objects). + """ + if not isinstance(row, dict): + return "malformed" + if row.get("error"): + return "error" + model_output = row.get("model_output") + if isinstance(model_output, dict) and model_output.get("error"): + return "error" + choices = model_output.get("choices") if isinstance(model_output, dict) else None + if isinstance(choices, list) and choices: + first = choices[0] + if isinstance(first, dict) and first.get("stop_reason") == "length": + return "truncated" + return "ok" + + +def analyze_rows(rows: Sequence[Any], parse_failures: int = 0) -> Dict[str, int]: + """Tally row classifications + unparseable JSONL lines.""" + counts = { + "total": len(rows), + "ok": 0, + "error": 0, + "truncated": 0, + "malformed": parse_failures, + } + for row in rows: + cls = classify_row(row) + if cls not in counts: + cls = "malformed" + counts[cls] += 1 + return counts + + +def _trim_perf(report: Dict[str, Any]) -> Optional[Dict[str, Any]]: + """Bounded, rounded performance summary from the report (may be None).""" + summary = (report.get("perf_metrics") or {}).get("summary") + if not isinstance(summary, dict): + return None + + def _rounded(value: Any) -> Any: + if isinstance(value, dict): + return {k: _rounded(v) for k, v in value.items()} + if isinstance(value, (int, float)) and not isinstance(value, bool): + return round(float(value), 4) + return value + + out = {k: _rounded(v) for k, v in summary.items()} + latency = out.get("latency") + if isinstance(latency, dict): + out["latency"] = { + k: latency[k] for k in ("mean", "p99") if k in latency + } + return out + + +# --------------------------------------------------------------------------- # +# Per-dataset normalization +# --------------------------------------------------------------------------- # + +def normalize_dataset( + report: Dict[str, Any], + rows: Sequence[Any], + *, + dataset: str, + threshold: Optional[float] = None, + raw_relative: Optional[str] = None, + parse_failures: int = 0, +) -> Dict[str, Any]: + """Turn a parsed report + prediction rows into one dataset result dict. + + Pure function — no I/O, easy to unit test with fixtures. Raises + :class:`ReportError` only if the report is not an EvalScope-shaped JSON + object (no ``metrics`` list). + """ + if not isinstance(report, dict): + raise ReportError("report is not a JSON object") + if not isinstance(report.get("metrics"), list): + raise ReportError("report has no 'metrics' list (not an EvalScope report)") + + reasons: List[str] = [] + schema_version = report.get("schema_version") + if not isinstance(schema_version, int) or schema_version < SUPPORTED_SCHEMA_VERSION: + reasons.append( + f"report schema_version {schema_version!r} < {SUPPORTED_SCHEMA_VERSION}" + ) + + identity, score, num = extract_primary(report, dataset) + if identity is None: + reasons.append("could not determine primary metric") + elif score is None: + reasons.append("primary metric has no score") + + counts = analyze_rows(rows, parse_failures=parse_failures) + if counts["malformed"]: + reasons.append(f"{counts['malformed']} malformed prediction line(s)") + if counts["error"]: + reasons.append(f"{counts['error']} prediction(s) had model errors") + if counts["truncated"]: + reasons.append( + f"{counts['truncated']} sample(s) truncated (stop_reason='length'); " + "re-run with a higher max_tokens" + ) + + execution = report.get("execution_summary") or {} + requested = execution.get("requested") + if requested is not None and requested != counts["total"]: + reasons.append( + f"sample-count mismatch: report requested {requested}, " + f"found {counts['total']} prediction line(s)" + ) + + complete = not reasons + threshold_met: Optional[bool] = None + if threshold is not None: + # A truncated/incomplete run must never silently "pass" its gate: the + # quality verdict only counts against a fully-executed evaluation. + threshold_met = ( + complete and score is not None and float(score) >= float(threshold) + ) + + return { + "dataset": dataset, + "status": "complete" if complete else "incomplete", + "complete": complete, + "has_report": True, + "primary_metric": metric_id(identity) if identity else None, + "primary_metric_identity": identity, + "score": score, + "num": num, + "requested": requested if requested is not None else counts["total"], + "predicted": counts["total"], + "errored": counts["error"], + "truncated": counts["truncated"], + "malformed": counts["malformed"], + "threshold": threshold, + "threshold_met": threshold_met, + "performance": _trim_perf(report), + "reasons": reasons, + "raw_relative": raw_relative, + } + + +def missing_dataset_row( + dataset: str, + *, + threshold: Optional[float] = None, + reason: str = "no EvalScope report/predictions produced", +) -> Dict[str, Any]: + """Dataset result for a target EvalScope never produced artifacts for.""" + return { + "dataset": dataset, + "status": "incomplete", + "complete": False, + "has_report": False, + "primary_metric": None, + "primary_metric_identity": None, + "score": None, + "num": None, + "requested": None, + "predicted": 0, + "errored": 0, + "truncated": 0, + "malformed": 0, + "threshold": threshold, + "threshold_met": None, + "performance": None, + "reasons": [reason], + "raw_relative": None, + } + + +# --------------------------------------------------------------------------- # +# I/O helpers (thin; kept out of the pure core above) +# --------------------------------------------------------------------------- # + +def load_predictions(file_paths: Iterable[Path]) -> Tuple[List[Any], int]: + """Parse prediction JSONL rows, returning ``(rows, parse_failures)``.""" + rows: List[Any] = [] + failures = 0 + for path in file_paths: + try: + with path.open(encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if not line: + continue + try: + rows.append(json.loads(line)) + except ValueError: + failures += 1 + except OSError: + failures += 1 + return rows, failures + + +def locate_report(report_root: Path, dataset: str) -> Optional[Path]: + """Find the EvalScope report file whose ``dataset_name`` matches. + + Matches by content (``dataset_name``), not filename — the file lives + under a ``/`` subdir that varies with the request. + """ + if not report_root.is_dir(): + return None + for path in sorted(report_root.rglob("*.json")): + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (ValueError, OSError): + continue + if isinstance(data, dict) and data.get("dataset_name") == dataset: + return path + return None + + +def load_report_file(path: Path) -> Dict[str, Any]: + return json.loads(path.read_text(encoding="utf-8")) + + +def normalize_attempt_dir( + attempt_dir: Path, + dataset: str, + *, + threshold: Optional[float] = None, + raw_relative: Optional[str] = None, +) -> Dict[str, Any]: + """Normalize one EvalScope attempt dir into a dataset result row. + + Thin glue: locate + parse the report and predictions EvalScope wrote + into ``attempt_dir``, then run the pure :func:`normalize_dataset`. + Returns a :func:`missing_dataset_row` when EvalScope produced no report. + """ + report_path = locate_report(attempt_dir / "reports", dataset) + if report_path is None: + return missing_dataset_row( + dataset, threshold=threshold, + reason="no EvalScope report produced for this dataset", + ) + report = load_report_file(report_path) + pred_files = locate_predictions(attempt_dir / "predictions", dataset) + rows, failures = load_predictions(pred_files) + return normalize_dataset( + report, + rows, + dataset=dataset, + threshold=threshold, + raw_relative=raw_relative, + parse_failures=failures, + ) + + +def locate_predictions(pred_root: Path, dataset: str) -> List[Path]: + """Return the prediction JSONL files for a dataset. + + File layout varies (``/_.jsonl``), so we match + by filename prefix first, then fall back to the first row's + ``metadata.task_id`` ("/"). + """ + if not pred_root.is_dir(): + return [] + matched = [] + for path in sorted(pred_root.rglob("*.jsonl")): + if _prediction_matches_dataset(path, dataset): + matched.append(path) + return matched + + +def _prediction_matches_dataset(path: Path, dataset: str) -> bool: + if path.name.startswith(dataset + "_") or path.name == dataset + ".jsonl": + return True + try: + with path.open(encoding="utf-8") as fh: + for line in fh: + if not line.strip(): + continue + try: + first = json.loads(line) + except ValueError: + return False + task_id = (first.get("metadata") or {}).get("task_id") or "" + return task_id.startswith(dataset + "/") + except OSError: + return False + return False diff --git a/metainfer/tasks/evalscope_correctness/orchestrator/runner.py b/metainfer/tasks/evalscope_correctness/orchestrator/runner.py new file mode 100644 index 00000000..40e7b25b --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/orchestrator/runner.py @@ -0,0 +1,175 @@ +"""Launch an isolated EvalScope child process for one dataset. + +Responsibilities: +* **Preflight** — a lazy, non-crashing check that EvalScope (``>=1.11,<2``) + is installed and that Docker is available when a sandboxed dataset + (HumanEval) is requested. MetaInfer's own dependency set stays unchanged; + EvalScope is only imported inside the child. +* **Child launch** — copy the non-secret config to stdin, copy the API key + only into the child's environment, run the worker module, and wait. + The child is NOT given a new session/process group, so it stays inside + the orchestrator's process group: MetaInfer's existing launcher/PID + lifecycle kills it together with the task. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Dict, Optional + +from . import config as _config + +# Module path of the isolated worker (launched via ``python -m ...``). +_WORKER_MODULE = ( + "metainfer.tasks.evalscope_correctness.orchestrator.evalscope_worker" +) + +# EvalScope version window we validate against. +EVALSCOPE_MIN = (1, 11) +EVALSCOPE_MAX = (2, 0) + + +@dataclass +class Preflight: + """Result of checking the environment can run a target.""" + + ok: bool + error: str = "" + + def __bool__(self) -> bool: # pragma: no cover - trivial + return self.ok + + +def check_evalscope_install() -> Preflight: + """Verify a compatible EvalScope is importable (lazily, no heavy import). + + Uses importlib metadata rather than importing evalscope so we don't + trigger its heavy import chain inside the orchestrator. + """ + import importlib.metadata + try: + version = importlib.metadata.version("evalscope") + except importlib.metadata.PackageNotFoundError: + return Preflight( + False, + "EvalScope is not installed. Install with: " + "pip install 'evalscope>=1.11,<2'", + ) + parts = [] + for seg in version.split(".")[:2]: + try: + parts.append(int(seg)) + except ValueError: + parts.append(0) + vtuple = tuple(parts[:2]) + if not (EVALSCOPE_MIN <= vtuple < EVALSCOPE_MAX): + return Preflight( + False, + f"EvalScope {version} is not in the supported range " + f"[{EVALSCOPE_MIN[0]}.{EVALSCOPE_MIN[1]}, {EVALSCOPE_MAX[0]}.0); " + "install 'evalscope>=1.11,<2'", + ) + return Preflight(True) + + +def check_sandbox_available() -> Preflight: + """Verify Docker (the sandbox engine) is available on PATH.""" + import shutil + if shutil.which("docker") is None: + return Preflight( + False, + "Docker is required to sandbox HumanEval code execution, but " + "'docker' was not found on PATH. Start Docker or omit HumanEval.", + ) + return Preflight(True) + + +def preflight_for(cfg: _config.EvalConfig) -> Optional[str]: + """Return an error string if the environment can't run the request. + + Checks EvalScope installation always; checks Docker only when a + sandboxed dataset is requested. A return of ``None`` means ready. + """ + ev = check_evalscope_install() + if not ev.ok: + return ev.error + if any(t.needs_sandbox for t in cfg.targets): + sand = check_sandbox_available() + if not sand.ok: + return sand.error + return None + + +@dataclass +class ChildResult: + exit_code: int + pid: Optional[int] = None + + +def run_dataset( + cfg: _config.EvalConfig, + target, + *, + attempt_dir: Path, + log_file: Path, + active: Dict[str, Any], + env_extra: Optional[Dict[str, str]] = None, +) -> ChildResult: + """Evaluate one dataset in a fresh attempt directory. + + ``active`` is a mutable holder (``{'proc': None}``) the caller uses to + terminate the running child on SIGTERM/SIGINT; we register the Popen + there for the duration of the wait. ``env_extra`` lets the caller inject + the API key (and any other vars) into the child environment without + them ever touching argv or logs. + """ + attempt_dir.mkdir(parents=True, exist_ok=True) + + # Build the non-secret child config (never the key). + child_cfg: Dict[str, Any] = cfg.to_child_json() + child_cfg["work_dir"] = str(attempt_dir) + child_cfg["datasets"] = [target.dataset] + child_cfg["needs_sandbox"] = bool(target.needs_sandbox) + + # Preserve the exact config we sent as immutable evidence. + (attempt_dir / "child_config.json").write_text( + json.dumps(child_cfg, indent=2), encoding="utf-8" + ) + + env = dict(os.environ) + if env_extra: + env.update(env_extra) + + log_file.parent.mkdir(parents=True, exist_ok=True) + stdout_fh = log_file.open("ab", buffering=0) if str(log_file) != "-" else None + + payload = json.dumps(child_cfg) + "\n" + proc = subprocess.Popen( + [sys.executable, "-m", _WORKER_MODULE], + stdin=subprocess.PIPE, + stdout=stdout_fh if stdout_fh else None, + stderr=subprocess.STDOUT, + env=env, + # No start_new_session: keep the child inside the orchestrator's + # process group so a group kill reaches it too. + start_new_session=False, + text=True, + ) + + active["proc"] = proc + try: + assert proc.stdin is not None + proc.stdin.write(payload) + proc.stdin.close() + exit_code = proc.wait() + finally: + active["proc"] = None + if stdout_fh is not None: + stdout_fh.close() + + return ChildResult(exit_code=exit_code, pid=proc.pid) diff --git a/metainfer/tasks/evalscope_correctness/server/__init__.py b/metainfer/tasks/evalscope_correctness/server/__init__.py new file mode 100644 index 00000000..6b668a0d --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/server/__init__.py @@ -0,0 +1 @@ +"""evalscope-correctness web plugin package.""" diff --git a/metainfer/tasks/evalscope_correctness/server/_state_readers.py b/metainfer/tasks/evalscope_correctness/server/_state_readers.py new file mode 100644 index 00000000..e5135aeb --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/server/_state_readers.py @@ -0,0 +1,34 @@ +"""State-dir readers for evalscope-correctness. + +Reads only the authoritative normalized result from +``/result.json``. Raw EvalScope artifacts live under the +workspace and are intentionally NOT served through an unbounded API +response — they stay on disk as immutable evidence. We read one fixed +file, so this module stays trivially small and safe. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Dict, Optional + + +def read_result(state_dir: Path) -> Optional[Dict[str, Any]]: + """Return the parsed ``result.json`` or ``None`` if not yet written / + malformed. + + ``None`` is indistinguishable to the caller from "not ready", which is + exactly what we want for a task whose result appears once at the end of + a run. + """ + path = state_dir / "result.json" + if not path.exists(): + return None + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (ValueError, OSError): + return None + if not isinstance(data, dict): + return None + return data diff --git a/metainfer/tasks/evalscope_correctness/server/plugin.py b/metainfer/tasks/evalscope_correctness/server/plugin.py new file mode 100644 index 00000000..58ae1382 --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/server/plugin.py @@ -0,0 +1,46 @@ +"""WebPlugin for evalscope-correctness — registers routes + detail view.""" + +from __future__ import annotations + +from pathlib import Path +from typing import List + +from metainfer.server._helpers import state_dir_for +from metainfer.server.registry import WebPlugin, register + +from .routes import build_router + +PLUGIN_TYPE = "evalscope-correctness" +_FRONTEND_DIR = Path(__file__).resolve().parent.parent / "static" + + +def _extra_watch_paths(entry) -> List[Path]: + """Tell the SSE watcher when ``result.json`` changes so the detail + view refetches the moment a run completes. + + ``result.json`` lives under ``state_dir`` (already watched via + run.json/timeline.jsonl), but the shell's watcher only monitors a fixed + relpath set by default; pointing it at the authoritative result file + keeps the UI fresh without an arbitrary polling interval in the browser. + """ + return [state_dir_for(entry) / "result.json"] + + +plugin = WebPlugin( + type=PLUGIN_TYPE, + label="EvalScope Correctness", + description=( + "Run EvalScope correctness benchmarks (GSM8K, GPQA-Diamond, HumanEval, " + "or custom datasets) against an OpenAI-compatible endpoint and report " + "normalized per-dataset scores with optional minimum-score gates." + ), + build_router=build_router, + detail_view_module="app/evalscope-detail", + detail_view_export="default", + frontend_dir=_FRONTEND_DIR, + importmap_entries={}, + extra_stylesheets=["evalscope.css"], + extra_watch_paths=_extra_watch_paths, +) + +register(plugin) diff --git a/metainfer/tasks/evalscope_correctness/server/routes.py b/metainfer/tasks/evalscope_correctness/server/routes.py new file mode 100644 index 00000000..5ab300d2 --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/server/routes.py @@ -0,0 +1,38 @@ +"""FastAPI router for evalscope-correctness. + +Routes mounted under ``/api/evalscope-correctness/{task_id}``: + + GET /result → normalized result.json (204 when not yet available) + +We deliberately expose only this one fixed endpoint. There is no route that +takes a free-form filename or streams raw prediction content — the raw +EvalScope artifacts stay on disk in the workspace. +""" + +from __future__ import annotations + +from fastapi import APIRouter, HTTPException + +from metainfer.server._helpers import ( + require_task_type, + state_dir_for, + task_or_404, +) +from . import _state_readers + +PLUGIN_TYPE = "evalscope-correctness" + + +def build_router(plugin) -> APIRouter: + router = APIRouter() + + @router.get("/result") + def get_result(task_id: str): + entry = task_or_404(task_id) + require_task_type(entry, PLUGIN_TYPE) + data = _state_readers.read_result(state_dir_for(entry)) + if data is None: + raise HTTPException(404, "result not yet available") + return data + + return router diff --git a/metainfer/tasks/evalscope_correctness/static/evalscope-detail.js b/metainfer/tasks/evalscope_correctness/static/evalscope-detail.js new file mode 100644 index 00000000..c63c32c6 --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/static/evalscope-detail.js @@ -0,0 +1,178 @@ +/** evalscope-correctness — task detail view. + * + * Renders the authoritative result (``state_dir/result.json``) served at + * ``/api/evalscope-correctness//result``: + * - a headline completeness verdict (did EvalScope evaluate everything + * without truncation / errors / count mismatch?) + * - an optional quality-gate verdict (all configured minimums met?) + * - a per-dataset table: primary score, sample count, metric id, + * configured threshold, and completeness counts + * - concise failure guidance when a run is incomplete. + * + * The shell passes ``{ taskId, run, status, data }``. We ignore most shell + * data and own our polling so the view is self-contained. + */ +import { html } from "htm/preact"; +import { useCallback, useEffect, useState } from "preact/hooks"; + +const PLUGIN_TYPE = "evalscope-correctness"; + +function getResult(taskId) { + return fetch(`/api/${PLUGIN_TYPE}/${encodeURIComponent(taskId)}/result`, { + cache: "no-store", + }).then(async (r) => { + if (!r.ok) throw new Error(`result: ${r.status}`); + return r.json(); + }); +} + +function pct(v) { + if (v == null) return "—"; + return `${(v * 100).toFixed(1)}%`; +} + +function VerdictChip({ ok }) { + if (ok == null) return html`pending`; + return ok + ? html`passed` + : html`not met`; +} + +function StatusChip({ complete }) { + return complete + ? html`complete` + : html`incomplete`; +} + +export default function EvalScopeDetail({ taskId, run }) { + const [result, setResult] = useState(null); + const [err, setErr] = useState(null); + const [ready, setReady] = useState(false); + + const refresh = useCallback(async () => { + if (!taskId) return; + try { + const r = await getResult(taskId); + setResult(r); + setErr(null); + } catch (e) { + // 404 = not ready yet (still configuring / running / errored preflight). + setResult((prev) => prev); // keep last known result if we had one + setErr(e.message); + } finally { + setReady(true); + } + }, [taskId]); + + useEffect(() => { refresh(); }, [refresh]); + + useEffect(() => { + if (!taskId) return; + const id = setInterval(refresh, 5000); + return () => clearInterval(id); + }, [taskId, refresh]); + + // Refetch immediately when the run transitions (started → running → done). + useEffect(() => { refresh(); }, [run?.current_phase, run?.finished]); + + const phase = run?.current_phase || "idle"; + if (!ready) { + return html`
Loading evaluation…
`; + } + + return html` +
+
+ phase: ${phase} + ${result && result.complete + ? html`evaluation complete` + : err && !result + ? html`result not ready yet` + : html`incomplete`} +
+ + ${result + ? html`<${ResultBody} result=${result} />` + : html`
+ No result yet. The evaluation is ${phase === "idle" ? "pending" : phase}.${ + err ? ` (${err})` : "" + } +
`} +
+ `; +} + +function ResultBody({ result }) { + const datasets = result.datasets || []; + const q = result.quality || {}; + const reasons = result.complete_reasons || []; + + return html` +
+
+

Execution completeness

+

${result.complete + ? "Every requested dataset was fully evaluated with no truncation or errors." + : "The evaluation is incomplete — see reasons below. Truncated samples are never counted toward a final score."}

+

${datasets.length} dataset(s) · fingerprint ${(result.fingerprint || "").slice(0, 16)}…

+
+
+

Quality gate

+

${q.configured + ? q.passed + ? "All configured minimum scores were met." + : "One or more datasets scored below its configured minimum." + : "No minimum-score gates configured — results are reported only."}

+
+
+ + ${reasons.length > 0 && html` +
+

Why the run is incomplete

+
    ${reasons.map((r, i) => html`
  • ${r}
    • `)}
+

Fix the underlying issue and resume the task. Datasets already fully evaluated under the same + request fingerprint are reused automatically.

+
+ `} + +
+

Per-dataset results

+ + + + + + + + + + + + + ${datasets.map((d) => html` + + + + + + + + + ${!d.complete && (d.reasons || []).length > 0 && html` + + + + `} + `)} + +
DatasetStatusMetricScoreSamplesGate
${d.dataset}<${StatusChip} complete=${d.complete} />${d.primary_metric || "—"}${pct(d.score)}${d.score != null ? ` of ${d.num ?? "?"} samples` : ""}${d.num ?? "—"} (pred ${d.predicted ?? "—"}) + ${d.threshold == null + ? html`—` + : html`${pct(d.threshold)} <${VerdictChip} ok=${d.threshold_met} />`} +
+
    ${d.reasons.map((r, i) => html`
  • ${d.dataset}: ${r}
    • `)}
+ ${d.raw_relative ? html`raw artifacts: ${d.raw_relative}` : ""} +
+
+ `; +} diff --git a/metainfer/tasks/evalscope_correctness/static/evalscope.css b/metainfer/tasks/evalscope_correctness/static/evalscope.css new file mode 100644 index 00000000..06a8dbf5 --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/static/evalscope.css @@ -0,0 +1,98 @@ +/* evalscope-correctness detail-view styles. Task-scoped under .esc-* to + avoid leaking into the shell. */ + +.esc-detail { + font-size: 13px; + line-height: 1.5; +} +.esc-muted { color: var(--text-muted, #8a8f98); } +.esc-small { font-size: 11px; color: var(--text-muted, #8a8f98); } + +.esc-header { + display: flex; + align-items: center; + gap: 10px; + margin-bottom: 14px; +} +.esc-phase { + font-family: var(--mono, monospace); + font-size: 11px; + padding: 2px 8px; + border-radius: 10px; + background: var(--bg-elevated, #2b2f36); + color: var(--text-muted, #8a8f98); +} + +.esc-chip { + font-size: 11px; + font-weight: 600; + padding: 2px 9px; + border-radius: 10px; + text-transform: uppercase; + letter-spacing: 0.3px; +} +.esc-chip-ok { background: rgba(39,174,96,.16); color: #27ae60; } +.esc-chip-fail { background: rgba(231,76,60,.16); color: #e74c3c; } +.esc-chip-pending { background: rgba(241,196,15,.16); color: #f1c40f; } + +.esc-verdicts { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(260px, 1fr)); + gap: 12px; + margin-bottom: 14px; +} +.esc-verdict { + border: 1px solid var(--border, #3a3f47); + border-radius: 8px; + padding: 10px 12px; + background: var(--bg-elevated, #24272c); +} +.esc-verdict h3 { margin: 0 0 6px; font-size: 12px; text-transform: uppercase; letter-spacing: .4px; } +.esc-verdict p { margin: 0; } +.esc-verdict-ok { border-left: 3px solid #27ae60; } +.esc-verdict-fail { border-left: 3px solid #e74c3c; } +.esc-verdict-pending { border-left: 3px solid #f1c40f; } + +.esc-reasons { + border: 1px solid rgba(231,76,60,.4); + border-radius: 8px; + padding: 10px 12px; + margin-bottom: 14px; + background: rgba(231,76,60,.06); +} +.esc-reasons h3 { margin: 0 0 6px; font-size: 12px; color: #e74c3c; } +.esc-reasons ul { margin: 0; padding-left: 18px; } +.esc-reasons li { margin: 2px 0; } + +.esc-panel { + border: 1px solid var(--border, #3a3f47); + border-radius: 8px; + padding: 12px; + background: var(--bg-elevated, #24272c); +} +.esc-panel h3 { margin: 0 0 10px; font-size: 13px; } + +.esc-table { + width: 100%; + border-collapse: collapse; + font-size: 12.5px; +} +.esc-table th, .esc-table td { + text-align: left; + padding: 7px 9px; + border-bottom: 1px solid var(--border, #3a3f47); +} +.esc-table th { + color: var(--text-muted, #8a8f98); + font-weight: 600; + font-size: 11px; + text-transform: uppercase; + letter-spacing: .4px; +} +.esc-row-incomplete td { background: rgba(231,76,60,.04); } +.esc-reason-row td { + background: rgba(231,76,60,.05); + padding-left: 22px; +} +.esc-reason-row ul { margin: 2px 0; padding-left: 16px; } +.esc-reason-row li { color: #e74c3c; font-size: 12px; } diff --git a/metainfer/tasks/evalscope_correctness/tests/__init__.py b/metainfer/tasks/evalscope_correctness/tests/__init__.py new file mode 100644 index 00000000..3ae0120d --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for the evalscope-correctness task plugin.""" diff --git a/metainfer/tasks/evalscope_correctness/tests/_fixtures.py b/metainfer/tasks/evalscope_correctness/tests/_fixtures.py new file mode 100644 index 00000000..3f887559 --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/tests/_fixtures.py @@ -0,0 +1,107 @@ +"""Compact EvalScope report/prediction fixtures for tests. + +The shapes here mirror the schema EvalScope 1.11 actually writes (verified +against real report JSON): schema_version 2, a ``metrics`` list keyed by +full identity, ``primary_metric_identity``, and ``execution_summary``. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Dict, List, Optional + + +def _acc_mean(num: int, score: float) -> Dict[str, Any]: + return { + "identity": {"name": "accuracy", "aggregation": "mean", "dimensions": {}}, + "score": score, + "num": num, + } + + +def _pass_at_k(num: int, score: float, k: int = 1) -> Dict[str, Any]: + return { + "identity": { + "name": "accuracy", "aggregation": "pass_at_k", "dimensions": {"k": k}, + }, + "score": score, + "num": num, + } + + +def gsm8k_report(num: int = 8, score: float = 0.75) -> Dict[str, Any]: + return { + "schema_version": 2, + "name": "gsm8k", + "dataset_name": "gsm8k", + "metrics": [_acc_mean(num, score)], + "primary_metric_identity": {"name": "accuracy", "aggregation": "mean", + "dimensions": {}}, + "execution_summary": { + "requested": num, "succeeded": num, "errored": 0, "incomplete": False, + "subsets": {"main": {"requested": num, "succeeded": num, "errored": 0}}, + }, + } + + +def humaneval_report(num: int = 8, score: float = 0.5) -> Dict[str, Any]: + """HumanEval: mean + pass@1 present; primary is pass@1 (index 1).""" + return { + "schema_version": 2, + "name": "humaneval", + "dataset_name": "humaneval", + "metrics": [_acc_mean(num, score), _pass_at_k(num, score, k=1)], + "primary_metric_identity": { + "name": "accuracy", "aggregation": "pass_at_k", "dimensions": {"k": 1}, + }, + "execution_summary": { + "requested": num, "succeeded": num, "errored": 0, "incomplete": False, + "subsets": {"openai_humaneval": {"requested": num, "succeeded": num, + "errored": 0}}, + }, + } + + +def custom_report(dataset: str, num: int = 8, score: float = 0.6) -> Dict[str, Any]: + """A dataset with no declared primary metric → normalizer uses metrics[0].""" + return { + "schema_version": 2, + "name": dataset, + "dataset_name": dataset, + "metrics": [_acc_mean(num, score)], + "primary_metric_identity": None, + "execution_summary": { + "requested": num, "succeeded": num, "errored": 0, "incomplete": False, + }, + } + + +def pred_row(idx: int, dataset: str, stop: str = "stop", + error: Optional[str] = None) -> Dict[str, Any]: + row: Dict[str, Any] = { + "index": idx, + "model_output": { + "choices": [{"stop_reason": stop}], + "error": None, + }, + "error": None, + } + if error is not None: + row["error"] = error + return row + + +def write_attempt(attempt_dir: Path, dataset: str, report: Dict[str, Any], + rows: List[Dict[str, Any]], *, filename_prefix: str, + model_id: str = "Q") -> Path: + """Write an EvalScope-shaped attempt dir; returns the attempt dir.""" + reports = attempt_dir / "reports" / model_id + preds = attempt_dir / "predictions" / model_id + reports.mkdir(parents=True, exist_ok=True) + preds.mkdir(parents=True, exist_ok=True) + (reports / f"{dataset}.json").write_text(json.dumps(report), encoding="utf-8") + with (preds / f"{filename_prefix}_{dataset}.jsonl").open("w") as fh: + for row in rows: + fh.write(json.dumps(row) + "\n") + return attempt_dir diff --git a/metainfer/tasks/evalscope_correctness/tests/test_config.py b/metainfer/tasks/evalscope_correctness/tests/test_config.py new file mode 100644 index 00000000..bae7ea27 --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/tests/test_config.py @@ -0,0 +1,138 @@ +"""Config parsing/validation + fingerprint stability.""" + +from __future__ import annotations + +import pytest + +from metainfer.tasks.evalscope_correctness.orchestrator import config as _config +from metainfer.tasks.evalscope_correctness.orchestrator.config import ( + ConfigError, + parse_requirements, +) + + +def _req(**over): + base = { + "task_type": "evalscope-correctness", + "api_url": "http://127.0.0.1:30000/v1", + "model": "Qwen", + "benchmarks": ["gsm8k", "humaneval"], + } + base.update(over) + return base + + +def test_parses_presets_and_gates(): + cfg = parse_requirements(_req(gate_gsm8k="0.7", gate_humaneval="0.5")) + assert cfg.dataset_ids == ["gsm8k", "humaneval"] + assert cfg.gate_for("gsm8k") == pytest.approx(0.7) + assert cfg.gate_for("humaneval") == pytest.approx(0.5) + humaneval = [t for t in cfg.targets if t.dataset == "humaneval"][0] + assert humaneval.needs_sandbox is True + gsm = [t for t in cfg.targets if t.dataset == "gsm8k"][0] + assert gsm.needs_sandbox is False + + +def test_custom_dataset_and_json_gate(): + cfg = parse_requirements(_req( + benchmarks=["gsm8k"], + custom_benchmarks="arc, mmlu", + gate_custom_json='{"arc": 0.6}', + )) + assert cfg.dataset_ids == ["gsm8k", "arc", "mmlu"] + assert cfg.gate_for("arc") == pytest.approx(0.6) + assert cfg.gate_for("mmlu") is None + + +def test_custom_gate_referencing_unrequested_dataset_is_error(): + with pytest.raises(ConfigError, match="was not requested"): + parse_requirements(_req( + benchmarks=["gsm8k"], + custom_benchmarks="arc", + gate_custom_json='{"mmlu": 0.6}', + )) + + +def test_gate_custom_without_custom_dataset_is_error(): + with pytest.raises(ConfigError, match="no custom benchmarks requested"): + parse_requirements(_req( + benchmarks=["gsm8k"], + gate_custom_json='{"arc": 0.6}', + )) + + +@pytest.mark.parametrize("field,value", [ + ("api_url", ""), + ("model", " "), +]) +def test_required_fields(field, value): + with pytest.raises(ConfigError, match="required"): + parse_requirements(_req(**{field: value})) + + +def test_no_benchmarks_is_error(): + with pytest.raises(ConfigError, match="select at least one benchmark"): + parse_requirements(_req(benchmarks=[])) + + +def test_bad_url_is_error(): + with pytest.raises(ConfigError, match="http"): + parse_requirements(_req(api_url="not a url")) + + +def test_gate_out_of_range_is_error(): + for val in ("1.2", "-0.1"): + with pytest.raises(ConfigError, match=r"\[0, 1\]"): + parse_requirements(_req(gate_gsm8k=val)) + + +def test_batch_size_validated(): + with pytest.raises(ConfigError, match="eval_batch_size"): + parse_requirements(_req(eval_batch_size="16")) + assert parse_requirements(_req(eval_batch_size="1")).eval_batch_size == 1 + + +def test_temperature_forced_to_zero(): + cfg = parse_requirements(_req(seed="7", max_tokens="4096")) + assert cfg.temperature == 0.0 + assert cfg.seed == 7 + assert cfg.max_tokens == 4096 + + +def test_env_var_name_rejects_secret_shaped_value(): + # A bare secret (with a hyphen) must not be accepted as an env-var name. + with pytest.raises(ConfigError, match="environment variable"): + parse_requirements(_req(api_key_env_var="sk-abc123secret")) + # A valid NAME is fine. + assert parse_requirements(_req(api_key_env_var="MY_LLM_KEY")).api_key_env_var == "MY_LLM_KEY" + + +def test_empty_env_var_means_no_auth(): + assert parse_requirements(_req(api_key_env_var="")).api_key_env_var == "" + + +def test_fingerprint_ignores_gate_and_model_id(): + a = parse_requirements(_req(gate_gsm8k="0.5", model_id="alias")) + b = parse_requirements(_req(gate_gsm8k="0.9", model_id="other")) + assert a.fingerprint() == b.fingerprint() + + +def test_fingerprint_changes_with_immutable_inputs(): + base = parse_requirements(_req()) + # max_tokens change → different request → new fingerprint. + other = parse_requirements(_req(max_tokens="4096")) + assert base.fingerprint() != other.fingerprint() + + +def test_child_json_never_contains_secret(): + cfg = parse_requirements(_req(api_key_env_var="EVALSCOPE_API_KEY")) + child = cfg.to_child_json() + # Only the NAME is present; no key value, no key field. + assert child["api_key_env_var"] == "EVALSCOPE_API_KEY" + assert "api_key" not in child + assert "secret" not in json_dumps_lower(child) + + +def json_dumps_lower(obj) -> str: + import json + return json.dumps(obj).lower() diff --git a/metainfer/tasks/evalscope_correctness/tests/test_orchestrator.py b/metainfer/tasks/evalscope_correctness/tests/test_orchestrator.py new file mode 100644 index 00000000..9bd2c094 --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/tests/test_orchestrator.py @@ -0,0 +1,149 @@ +"""Orchestrator lifecycle: invalid-input, preflight, resume reuse, secret.""" + +from __future__ import annotations + +import json + +import pytest + +from metainfer.tasks.evalscope_correctness.orchestrator import ( + orchestrator as _orch, +) +from metainfer.tasks.evalscope_correctness.orchestrator import config as _config +from metainfer.tasks.evalscope_correctness.orchestrator import report as _report + +from ._fixtures import gsm8k_report, humaneval_report, pred_row, write_attempt + + +def _write_requirements(state_dir, data): + state_dir.mkdir(parents=True, exist_ok=True) + p = state_dir / "req.json" + p.write_text(json.dumps(data), encoding="utf-8") + return p + + +def _good_req(): + return { + "task_id": "esc-t", + "task_type": "evalscope-correctness", + "label": "t", + "api_url": "http://127.0.0.1:30000/v1", + "model": "Qwen", + "benchmarks": ["gsm8k"], + "gate_gsm8k": "0.7", + } + + +def test_invalid_input_stops_without_running(tmp_path): + req = _write_requirements(tmp_path, {**_good_req(), "api_url": "nope"}) + state_dir, work_dir = tmp_path / "state", tmp_path / "work" + code = _orch.run_with_requirements( + req, state_dir=state_dir, workspace_dir=work_dir + ) + assert code == 2 + run = json.loads((state_dir / "run.json").read_text()) + assert run["final_status"] == "stopped" + assert "input validation" in run["last_transition_label"] + # pid file must be cleared (exited). + pid = json.loads((state_dir / "orchestrator.pid").read_text()) + assert pid["pid"] is None + + +def test_preflight_failure_stops(tmp_path, monkeypatch): + from metainfer.tasks.evalscope_correctness.orchestrator import runner as _runner + monkeypatch.setattr(_runner, "preflight_for", lambda cfg: "EvalScope not installed") + req = _write_requirements(tmp_path, _good_req()) + code = _orch.run_with_requirements( + req, state_dir=tmp_path / "state", workspace_dir=tmp_path / "work" + ) + assert code == 2 + run = json.loads((tmp_path / "state" / "run.json").read_text()) + assert run["final_status"] == "stopped" + assert "preflight" in run["last_transition_label"] + + +def test_supervisor_reuses_complete_attempt_without_running(tmp_path, monkeypatch): + """A dataset already complete under the fingerprint must not re-run.""" + req = _write_requirements(tmp_path, _good_req()) + cfg = _config.parse_requirements(json.loads(req.read_text())) + fp = cfg.fingerprint() + evalscope_root = tmp_path / "work" / "evalscope" + + # Seed a complete attempt for gsm8k under the current fingerprint. + a1 = evalscope_root / "gsm8k" / "attempt-1" + write_attempt(a1, "gsm8k", gsm8k_report(num=4, score=0.8), + [pred_row(i, "gsm8k") for i in range(4)], + filename_prefix="gsm8k", model_id="Qwen") + (a1 / "attempt.json").write_text(json.dumps({ + "dataset": "gsm8k", "fingerprint": fp, "exit_code": 0, + }), encoding="utf-8") + + def _should_not_run(*_a, **_kw): + raise AssertionError("a complete dataset must not be re-run") + + monkeypatch.setattr(_orch._runner, "run_dataset", _should_not_run) + + code = _orch.run_with_requirements( + req, state_dir=tmp_path / "state", workspace_dir=tmp_path / "work" + ) + assert code == 0 # reuse is transparent → success + result = json.loads((tmp_path / "state" / "result.json").read_text()) + assert result["complete"] is True + assert result["datasets"][0]["dataset"] == "gsm8k" + assert result["datasets"][0]["raw_relative"] == "evalscope/gsm8k/attempt-1" + + +def test_attempt_with_different_fingerprint_is_not_reused(tmp_path): + req = _write_requirements(tmp_path, _good_req()) + cfg = _config.parse_requirements(json.loads(req.read_text())) + fp = cfg.fingerprint() + evalscope_root = tmp_path / "work" / "evalscope" + a1 = evalscope_root / "gsm8k" / "attempt-1" + write_attempt(a1, "gsm8k", gsm8k_report(num=4, score=0.8), + [pred_row(i, "gsm8k") for i in range(4)], filename_prefix="gsm8k") + (a1 / "attempt.json").write_text(json.dumps({ + "dataset": "gsm8k", "fingerprint": "sha256:different", "exit_code": 0, + }), encoding="utf-8") + assert _orch._latest_complete_attempt( + evalscope_root, "gsm8k", fp, None + ) is None + + +def test_attempt_that_did_not_finish_cleanly_is_not_reused(tmp_path): + req = _write_requirements(tmp_path, _good_req()) + cfg = _config.parse_requirements(json.loads(req.read_text())) + fp = cfg.fingerprint() + evalscope_root = tmp_path / "work" / "evalscope" + a1 = evalscope_root / "gsm8k" / "attempt-1" + write_attempt(a1, "gsm8k", gsm8k_report(num=4, score=0.8), + [pred_row(i, "gsm8k") for i in range(4)], filename_prefix="gsm8k") + (a1 / "attempt.json").write_text(json.dumps({ + "dataset": "gsm8k", "fingerprint": fp, "exit_code": 4, # worker failed + }), encoding="utf-8") + assert _orch._latest_complete_attempt( + evalscope_root, "gsm8k", fp, None + ) is None + + +def test_finalize_completeness_and_gate(tmp_path): + """A complete-but-below-gate evaluation is still reported as complete; + the quality gate is the independent pass/fail.""" + evalscope_root = tmp_path / "work" / "evalscope" + work = tmp_path / "work" + a1 = evalscope_root / "gsm8k" / "attempt-1" + write_attempt(a1, "gsm8k", gsm8k_report(num=4, score=0.6), # below 0.7 gate + [pred_row(i, "gsm8k") for i in range(4)], filename_prefix="gsm8k") + cfg = _config.parse_requirements(_good_req()) # gate_gsm8k=0.7 + result = _orch._finalize( + cfg=cfg, + evalscope_root=evalscope_root, + chosen={"gsm8k": a1}, + workspace_dir=work, + ) + assert result["complete"] is True # evaluation fully executed + assert result["quality"]["configured"] is True + assert result["quality"]["passed"] is False # 0.6 < 0.7 + d0 = result["datasets"][0] + assert d0["score"] == pytest.approx(0.6) + assert d0["threshold"] == 0.7 + assert d0["threshold_met"] is False diff --git a/metainfer/tasks/evalscope_correctness/tests/test_plugin.py b/metainfer/tasks/evalscope_correctness/tests/test_plugin.py new file mode 100644 index 00000000..58ea8a09 --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/tests/test_plugin.py @@ -0,0 +1,41 @@ +"""Auto-registration + form schema for the evalscope-correctness plugin.""" + +from __future__ import annotations + +import pytest + +import metainfer.tasks # noqa: F401 — triggers plugin registration + + +def test_task_plugin_registered(): + from metainfer.orchestrator.tasks import get_task + plugin = get_task("evalscope-correctness") + assert plugin.task_type == "evalscope-correctness" + assert plugin.cli_module.endswith("orchestrator.cli") + # Single-run task: no state graph, no copy-forward globs. + assert plugin.phases_module == "" + assert plugin.diagnostic_globs == () + + +def test_web_plugin_registered(): + from metainfer.server.registry import get + plugin = get("evalscope-correctness") + assert plugin is not None + assert plugin.type == "evalscope-correctness" + assert plugin.detail_view_module == "app/evalscope-detail" + assert "evalscope.css" in plugin.extra_stylesheets + assert plugin.frontend_dir is not None + assert plugin.extra_watch_paths is not None + + +def test_form_schema_shape(): + from metainfer.server.forms import load_form_schema + schema = load_form_schema("evalscope-correctness") + assert schema is not None + keys = {f["key"] for f in schema["fields"]} + for required in ("api_url", "model", "benchmarks"): + assert required in keys + by_key = {f["key"]: f for f in schema["fields"]} + assert by_key["benchmarks"]["type"] == "multiselect" + opts = {o["label"] for o in by_key["benchmarks"]["options"]} + assert {"gsm8k", "gpqa_diamond", "humaneval"} <= opts diff --git a/metainfer/tasks/evalscope_correctness/tests/test_report.py b/metainfer/tasks/evalscope_correctness/tests/test_report.py new file mode 100644 index 00000000..b1a0fb13 --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/tests/test_report.py @@ -0,0 +1,166 @@ +"""Report normalization — completeness vs quality-gate semantics.""" + +from __future__ import annotations + +import pytest + +from metainfer.tasks.evalscope_correctness.orchestrator import report as _report + +from ._fixtures import ( + custom_report, + gsm8k_report, + humaneval_report, + pred_row, +) + + +def _rows(n, stop="stop", errored_at=None): + return [pred_row(i, "gsm8k", stop=stop, error=errored_at if i == errored_at else None) + for i in range(n)] + + +def test_gsm8k_accuracy_mean_selected(): + report = gsm8k_report(num=8, score=0.75) + rows = _rows(8) + out = _report.normalize_dataset(report, rows, dataset="gsm8k") + assert out["complete"] is True + assert out["primary_metric"] == "accuracy/mean" + assert out["score"] == pytest.approx(0.75) + assert out["num"] == 8 + assert out["requested"] == 8 and out["predicted"] == 8 + + +def test_humaneval_picks_pass_at_k_not_metrics0(): + """metrics[0] is accuracy/mean, but the primary is pass@1 (index 1).""" + report = humaneval_report(num=8, score=0.5) + rows = _rows(8) + out = _report.normalize_dataset(report, rows, dataset="humaneval") + assert out["complete"] is True + assert out["primary_metric"] == "accuracy/pass_at_k(k=1)" + assert out["score"] == pytest.approx(0.5) + + +def test_truncation_fails_completeness(): + report = gsm8k_report(num=8, score=0.5) + rows = _rows(8) + rows[3] = pred_row(3, "gsm8k", stop="length") # stop_reason length + out = _report.normalize_dataset(report, rows, dataset="gsm8k") + assert out["complete"] is False + assert out["truncated"] == 1 + assert any("truncated" in r for r in out["reasons"]) + + +def test_model_error_fails_completeness(): + report = gsm8k_report(num=8, score=0.5) + rows = _rows(8, errored_at=2) + rows[2]["model_output"]["error"] = "rate limited" + out = _report.normalize_dataset(report, rows, dataset="gsm8k") + assert out["complete"] is False + assert out["errored"] == 1 + + +def test_parse_failures_count_as_malformed(): + report = gsm8k_report(num=8, score=0.5) + rows = _rows(8) + out = _report.normalize_dataset(report, rows, dataset="gsm8k", parse_failures=2) + assert out["complete"] is False + assert out["malformed"] == 2 + + +def test_sample_count_mismatch_fails_completeness(): + report = gsm8k_report(num=10, score=0.5) # requested 10 + rows = _rows(8) # only 8 predictions + out = _report.normalize_dataset(report, rows, dataset="gsm8k") + assert out["complete"] is False + assert any("sample-count mismatch" in r for r in out["reasons"]) + + +def test_missing_metric_score_fails_completeness(): + report = gsm8k_report(num=8, score=None) + rows = _rows(8) + out = _report.normalize_dataset(report, rows, dataset="gsm8k") + assert out["complete"] is False + assert out["score"] is None + assert any("no score" in r for r in out["reasons"]) + + +def test_non_evalscope_report_raises(): + with pytest.raises(_report.ReportError): + _report.normalize_dataset({"foo": 1}, [], dataset="gsm8k") + + +def test_schema_version_below_supported_fails(): + report = gsm8k_report() + report["schema_version"] = 1 + out = _report.normalize_dataset(report, _rows(8), dataset="gsm8k") + assert out["complete"] is False + assert any("schema_version" in r for r in out["reasons"]) + + +def test_custom_dataset_uses_metrics0_when_no_primary(): + report = custom_report("arc", num=8, score=0.6) + rows = [pred_row(i, "arc") for i in range(8)] + out = _report.normalize_dataset(report, rows, dataset="arc") + assert out["complete"] is True + assert out["primary_metric"] == "accuracy/mean" + assert out["score"] == pytest.approx(0.6) + + +# --------------------------------------------------------------------------- # +# Threshold / quality-gate semantics +# --------------------------------------------------------------------------- # + +def test_threshold_met_when_at_or_above(): + report = gsm8k_report(num=8, score=0.7) + out = _report.normalize_dataset(report, _rows(8), dataset="gsm8k", + threshold=0.7) + assert out["threshold_met"] is True + assert out["threshold"] == 0.7 + + +def test_threshold_not_met_when_below(): + report = gsm8k_report(num=8, score=0.6) + out = _report.normalize_dataset(report, _rows(8), dataset="gsm8k", + threshold=0.7) + assert out["threshold_met"] is False + + +def test_no_threshold_reports_only(): + report = gsm8k_report(num=8, score=0.6) + out = _report.normalize_dataset(report, _rows(8), dataset="gsm8k") + assert out["threshold"] is None + assert out["threshold_met"] is None + + +def test_incomplete_dataset_with_threshold_is_not_met(): + report = gsm8k_report(num=8, score=0.6) + rows = _rows(8) + rows[0] = pred_row(0, "gsm8k", stop="length") + out = _report.normalize_dataset(report, rows, dataset="gsm8k", threshold=0.5) + # Truncated → not complete → gate must not silently "pass". + assert out["complete"] is False + assert out["threshold_met"] is False + + +def test_missing_dataset_row(): + out = _report.missing_dataset_row("gsm8k", threshold=0.7, reason="boom") + assert out["complete"] is False + assert out["has_report"] is False + assert out["reasons"] == ["boom"] + assert out["threshold"] == 0.7 + + +def test_perf_summary_bounded(): + report = gsm8k_report() + report["perf_metrics"] = {"summary": { + "n_samples": 8, + "latency": {"mean": 12.3456789, "p99": 99.999999, "min": 1.0, "max": 500.0}, + "throughput": {"avg_output_tps": 61.62}, + "usage": {"input_tokens": {"mean": 178.0}}, + }} + out = _report.normalize_dataset(report, _rows(8), dataset="gsm8k") + perf = out["performance"] + # Only a bounded latency subset survives (mean/p99), rounded. + assert set(perf["latency"].keys()) == {"mean", "p99"} + assert perf["latency"]["mean"] == pytest.approx(12.3457) + assert "min" not in perf["latency"] diff --git a/metainfer/tasks/evalscope_correctness/tests/test_runner.py b/metainfer/tasks/evalscope_correctness/tests/test_runner.py new file mode 100644 index 00000000..047ef732 --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/tests/test_runner.py @@ -0,0 +1,175 @@ +"""Runner: EvalScope install checks, Docker preflight, child env isolation.""" + +from __future__ import annotations + +import json + +import pytest + +from metainfer.tasks.evalscope_correctness.orchestrator import runner as _runner +from metainfer.tasks.evalscope_correctness.orchestrator import config as _config +from metainfer.tasks.evalscope_correctness.orchestrator import evalscope_worker as _worker + + +def _version(v): + # importlib.metadata.version("evalscope") is called with a package arg. + return lambda _pkg: v + + +def test_evalscope_ok_within_range(monkeypatch): + monkeypatch.setattr("importlib.metadata.version", _version("1.11.0")) + assert _runner.check_evalscope_install().ok is True + + +def test_evalscope_missing(monkeypatch): + def _missing(pkg): + from importlib.metadata import PackageNotFoundError + raise PackageNotFoundError(pkg) + monkeypatch.setattr("importlib.metadata.version", _missing) + pre = _runner.check_evalscope_install() + assert pre.ok is False + assert "install" in pre.error and "1.11" in pre.error + + +@pytest.mark.parametrize("v", ["1.9.3", "2.0.1"]) +def test_evalscope_out_of_range(monkeypatch, v): + monkeypatch.setattr("importlib.metadata.version", _version(v)) + pre = _runner.check_evalscope_install() + assert pre.ok is False + assert "range" in pre.error + + +def test_sandbox_requires_docker(monkeypatch): + monkeypatch.setattr("shutil.which", lambda name: None) + pre = _runner.check_sandbox_available() + assert pre.ok is False + assert "Docker" in pre.error + + +def test_sandbox_ok_when_docker_present(monkeypatch): + monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/docker") + assert _runner.check_sandbox_available().ok is True + + +def test_preflight_gates_humaneval_on_docker(monkeypatch): + monkeypatch.setattr("importlib.metadata.version", _version("1.11.0")) + monkeypatch.setattr("shutil.which", lambda name: None) + cfg = _config.parse_requirements({ + "api_url": "http://x:1/v1", "model": "Q", + "benchmarks": ["humaneval"], + }) + err = _runner.preflight_for(cfg) + assert err is not None and "Docker" in err + + +def test_preflight_no_docker_needed_for_gsm8k(monkeypatch): + monkeypatch.setattr("importlib.metadata.version", _version("1.11.0")) + cfg = _config.parse_requirements({ + "api_url": "http://x:1/v1", "model": "Q", "benchmarks": ["gsm8k"], + }) + assert _runner.preflight_for(cfg) is None + + +# --------------------------------------------------------------------------- # +# Worker task-config construction (secret stays in env, not in argv/stdin) +# --------------------------------------------------------------------------- # + +def test_worker_builds_task_config(monkeypatch): + monkeypatch.setenv("EVALSCOPE_API_KEY", "sk-secret-value") + cfg = { + "api_url": "http://x:1/v1", "model": "Q", "model_id": "", + "datasets": ["gsm8k"], "temperature": 0.0, "seed": 42, + "eval_batch_size": 2, "timeout_seconds": 300, "max_tokens": 8192, + "limit": None, "dataset_cache_dir": "", "api_key_env_var": "EVALSCOPE_API_KEY", + "work_dir": "/tmp/attempt-1", "needs_sandbox": False, + } + task = _worker._build_task_config(cfg) + assert task["model"] == "Q" + assert task["datasets"] == ["gsm8k"] + assert task["eval_type"] == "openai_api" + assert task["eval_batch_size"] == 2 # remote default (8) overridden + assert task["no_timestamp"] is True + assert task["generation_config"]["temperature"] == 0.0 + assert task["generation_config"]["max_tokens"] == 8192 + assert task["api_key"] == "sk-secret-value" # injected from env + assert "use_sandbox" not in task + + +def test_worker_enables_sandbox_for_humaneval(monkeypatch): + monkeypatch.setenv("EVALSCOPE_API_KEY", "sk-secret-value") + cfg = { + "api_url": "http://x:1/v1", "model": "Q", "model_id": "", + "datasets": ["humaneval"], "temperature": 0.0, "seed": 42, + "eval_batch_size": 2, "timeout_seconds": 300, "max_tokens": 8192, + "limit": None, "dataset_cache_dir": "", + "api_key_env_var": "EVALSCOPE_API_KEY", "work_dir": "/tmp/a", + "needs_sandbox": True, + } + assert _worker._build_task_config(cfg)["use_sandbox"] is True + + +def test_worker_uses_EMPTY_when_no_key(monkeypatch): + monkeypatch.delenv("EVALSCOPE_API_KEY", raising=False) + cfg = { + "api_url": "http://x:1/v1", "model": "Q", "model_id": "", + "datasets": ["gsm8k"], "temperature": 0.0, "seed": 42, + "eval_batch_size": 1, "timeout_seconds": 300, "max_tokens": 8192, + "limit": None, "dataset_cache_dir": "", + "api_key_env_var": "EVALSCOPE_API_KEY", "work_dir": "/tmp/a", + "needs_sandbox": False, + } + assert _worker._build_task_config(cfg)["api_key"] == "EMPTY" + + +# --------------------------------------------------------------------------- # +# run_dataset: secret in child env only, never argv / on-disk config +# --------------------------------------------------------------------------- # + +class _FakePopen: + """Captures argv + env; never actually runs EvalScope.""" + + def __init__(self, argv, stdin=None, stdout=None, stderr=None, env=None, + start_new_session=False, text=False): + import io + self.argv = argv + self.env = env + self.stdin = io.StringIO() # writable so run_dataset can send config + self._start_new_session = start_new_session + self.pid = 12345 + + def wait(self, timeout=None): + return 0 + + +def test_run_dataset_puts_key_in_env_only(monkeypatch, tmp_path): + cfg = _config.parse_requirements({ + "api_url": "http://x:1/v1", "model": "Q", "benchmarks": ["gsm8k"], + "api_key_env_var": "EVALSCOPE_API_KEY", + }) + monkeypatch.setenv("EVALSCOPE_API_KEY", "sk-topsecret") + attempt_dir = tmp_path / "gsm8k" / "attempt-1" + captured = {} + + def _fake_popen(*a, **kw): + captured["argv"] = a[0] + captured["env"] = dict(kw.get("env", {})) + return _FakePopen(*a, **kw) + + monkeypatch.setattr(_runner.subprocess, "Popen", _fake_popen) + + result = _runner.run_dataset( + cfg, cfg.targets[0], attempt_dir=attempt_dir, + log_file=attempt_dir / "worker.log", + active={}, + env_extra={"EVALSCOPE_API_KEY": "sk-topsecret"}, + ) + assert result.exit_code == 0 + # The secret appears in the child's environment… + assert captured["env"]["EVALSCOPE_API_KEY"] == "sk-topsecret" + # …but never on argv… + assert "sk-topsecret" not in json.dumps(captured["argv"]) + # …and never in the on-disk child config we wrote as evidence. + disk_cfg = json.loads((attempt_dir / "child_config.json").read_text()) + assert "sk-topsecret" not in json.dumps(disk_cfg) + assert "api_key" not in disk_cfg + assert disk_cfg["api_key_env_var"] == "EVALSCOPE_API_KEY" diff --git a/metainfer/tasks/evalscope_correctness/tests/test_server.py b/metainfer/tasks/evalscope_correctness/tests/test_server.py new file mode 100644 index 00000000..c8107646 --- /dev/null +++ b/metainfer/tasks/evalscope_correctness/tests/test_server.py @@ -0,0 +1,96 @@ +"""Server readers + ``GET /result`` route for evalscope-correctness.""" + +from __future__ import annotations + +import json + +import pytest +from fastapi.testclient import TestClient + +from metainfer.testing import isolated_env # noqa: F401 + +from metainfer.server import app as _app +from metainfer.server import tasks as _tasks +from metainfer.server.tasks import TaskEntry + +from metainfer.tasks.evalscope_correctness.server import _state_readers as readers + + +@pytest.fixture +def client(isolated_env): + return TestClient(_app.create_app()) + + +def _register_task(state_dir, task_id="esc-1", type_="evalscope-correctness", + workspace_dir=None): + state_dir.mkdir(parents=True, exist_ok=True) + if workspace_dir is None: + workspace_dir = state_dir + workspace_dir.mkdir(parents=True, exist_ok=True) + entry = TaskEntry( + id=task_id, type=type_, label="esc", state_dir=str(state_dir), + workspace_dir=str(workspace_dir), created_at=0.0, + ) + _tasks.add_task(entry) + return entry + + +def _seed_result(state_dir, payload): + (state_dir / "result.json").write_text(json.dumps(payload), encoding="utf-8") + + +def _good_result(): + return { + "schema_version": 1, + "task_type": "evalscope-correctness", + "complete": True, + "quality": {"configured": True, "passed": True}, + "datasets": [{"dataset": "gsm8k", "complete": True, "score": 0.8}], + } + + +# --------------------------------------------------------------------------- # +# Readers +# --------------------------------------------------------------------------- # + +def test_read_result_none_when_missing(tmp_path): + assert readers.read_result(tmp_path) is None + + +def test_read_result_returns_dict(tmp_path): + _seed_result(tmp_path, {"complete": True}) + assert readers.read_result(tmp_path) == {"complete": True} + + +def test_read_result_none_when_malformed(tmp_path): + (tmp_path / "result.json").write_text("{not json", encoding="utf-8") + assert readers.read_result(tmp_path) is None + + +# --------------------------------------------------------------------------- # +# GET /result +# --------------------------------------------------------------------------- # + +def test_result_404_when_not_ready(client, isolated_env): + _register_task(isolated_env["home"] / "tasks" / "esc-1") + assert client.get("/api/evalscope-correctness/esc-1/result").status_code == 404 + + +def test_result_200_when_ready(client, isolated_env): + state_dir = isolated_env["home"] / "tasks" / "esc-1" + _register_task(state_dir) + _seed_result(state_dir, _good_result()) + resp = client.get("/api/evalscope-correctness/esc-1/result") + assert resp.status_code == 200 + assert resp.json()["complete"] is True + assert resp.json()["datasets"][0]["dataset"] == "gsm8k" + + +def test_result_404_for_unknown_task(client, isolated_env): + assert client.get("/api/evalscope-correctness/nope/result").status_code == 404 + + +def test_result_409_for_wrong_task_type(client, isolated_env): + _register_task(isolated_env["home"] / "tasks" / "gf-1", task_id="gf-1", + type_="gen-infer-framework") + assert client.get("/api/evalscope-correctness/gf-1/result").status_code == 409