diff --git a/AGENTS.md b/AGENTS.md index 46e4e42b0..cf314378d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -100,6 +100,7 @@ Dataset Manager --> Load Generator --> Endpoint Client --> External Endpoint | **VideoGen** | `src/inference_endpoint/videogen/` | Adapter for video-generation endpoints (e.g. trtllm-serve `POST /v1/videos/generations`, used by MLPerf WAN2.2-T2V-A14B). Defaults to `response_format=video_path` (server saves video to shared storage and returns path) to avoid large byte payloads. Accuracy mode also runs on `video_path`: the adapter mirrors the path into `response_output` so the event log carries it to `VBenchScorer` (see `evaluation/scoring.py`), which scores videos via VBench from a sibling `uv` subproject at `examples/09_Wan22_VideoGen_Example/accuracy/` (vbench's `transformers==4.33.2` + `numpy<2` pins are incompatible with the parent env, so it runs out-of-process via `uv run --project`). Dataset is ingested via the generic JSONL loader. | | **SWE-bench** | `src/inference_endpoint/dataset_manager/predefined/swe_bench/`, `src/inference_endpoint/evaluation/swe_bench_scorer.py`, `src/inference_endpoint/evaluation/swebench_service/` | `SWEBench` predefined dataset (HuggingFace `princeton-nlp/SWE-bench_Verified` or `_Lite`; `ACCURACY_ONLY=True`). `SWEBenchScorer` sets `SKIP_ENDPOINT_PHASE=True` and bypasses the built-in accuracy phase entirely: it delegates agent execution and grading to the configured SWE-bench service via `accuracy_config.extras.swebench_service_url`. The service is an isolated `uv` subproject; its host owns Docker/runtime execution, artifacts, and credentials, while the benchmark client remains the report-producing entrypoint. | | **Compliance (submission checker)** | `src/inference_endpoint/compliance/checker.py`, `scripts/check_compliance.py` | Validates a completed run's report directory against a registered ruleset. `check_submission(report_dir, ruleset, model)` reads the resolved `config.yaml` plus scorer output (`accuracy/accuracy_results.json` for accuracy, `scores.json` for the agentic perf run) and runs config-lock (deterministic + single-stream), the accuracy gate (`score >= factor x reference`, factor 0.97 for Edge-Agentic), and run validity (0 dropped turns). Server-side launch flags (`--reasoning off`, `--ctx-size`) aren't in client artifacts, so they're surfaced as manual attestations. CLI: `scripts/check_compliance.py REPORT_DIR` (exit 0 = pass). | +| **SWE-bench (distributed)** | `src/inference_endpoint/evaluation/swe_bench_distributed/`, `src/inference_endpoint/evaluation/swe_bench_fleet_scorer.py`, `scripts/swe_bench_wq.py` | `SWEBenchFleetScorer` (`eval_method: swe_bench_fleet`) shards the instance list into units and runs them across several SWE-bench services concurrently, reusing the same service HTTP protocol. Adds what the single-service path lacks: a durable `mkdir`-atomic work queue (resume after a client crash), an eval-phase infra-vs-genuine classifier driving in-unit retry, pre-dispatch gates on the inference endpoints (checkpoint identity, tool call at >=2k-token scale, endpoint fingerprint), a memory guard, and an all-or-nothing merge gate that compares instance **ids** (never counts) and is scoped to exactly one run id. Operator CLI: `scripts/swe_bench_wq.py {status,merge,requeue,reap}` — `requeue` is the only way to re-run a unit; deleting a result leaves the claim tombstone in place. See `docs/evaluation/SWE_BENCH_DISTRIBUTED.md`. | | **Compliance (audit tests)** | `src/inference_endpoint/compliance/`, `commands/audit.py` | MLPerf compliance audits. `AuditTest` protocol + `AuditRunSpec`/`AuditRunArtifacts` + registry (`compliance/__init__.py`); `OutputCachingAudit` (`compliance/audit_test/output_caching_test.py`, which also owns the QPS-specific `AuditRunStats`) implements MLPerf **TEST04** output-caching detection — reference phase (distinct samples) vs. fixed-sample audit phase, comparing QPS against `threshold`. `commands/audit.py:run_audit` runs phases via `AuditTest.plan_runs`/`validate`, writing `audit_result.json`/`verify_.txt` atomically via `compliance/result.py`. Enabled by the `audit:` YAML block; `cli._run` runs it after the main benchmark (upstream MLPerf order: perf run, then TEST04), or standalone with `audit.only: true`. Perf-only by default (a phase may opt into accuracy via `AuditRunSpec.test_mode`, but this is unused today). | ### Hot-Path Architecture @@ -266,6 +267,8 @@ src/inference_endpoint/ │ └── adapter.py # VideoGenAdapter (HttpRequestAdapter) + VideoGenAccumulator (no-op) ├── evaluation/ # Accuracy evaluation (extractor, scoring, livecodebench) │ └── swebench_service/ # Isolated uv service for Docker-backed SWE-bench runs +│ ├── swe_bench_distributed/ # Fleet dispatch: unit plan, work queue, reaper, classifier, gates, guards, merge gate +│ └── swe_bench_fleet_scorer.py # SWEBenchFleetScorer (scorer_id swe_bench_fleet) ├── compliance/ # Submission compliance checks (config-lock, accuracy gate, run validity) │ ├── __init__.py │ └── checker.py # check_submission() + Check/ComplianceReport (Edge-Agentic ruleset) diff --git a/docs/evaluation/SWE_BENCH_DISTRIBUTED.md b/docs/evaluation/SWE_BENCH_DISTRIBUTED.md new file mode 100644 index 000000000..898d4e15b --- /dev/null +++ b/docs/evaluation/SWE_BENCH_DISTRIBUTED.md @@ -0,0 +1,233 @@ +# Distributed SWE-bench (`swe_bench_fleet`) + +> Shards a SWE-bench accuracy run across several SWE-bench services, classifies +> infrastructure damage separately from genuine model failures, and refuses to +> emit an accuracy number unless every planned instance is accounted for exactly +> once. + +`swe_bench_scorer` runs the whole instance list as one service run against one +endpoint. That is the right shape for a hundred instances on one Docker host. It +does not survive a 200-instance run spread over many hours and many hosts: a +client crash loses everything, a host that dies takes its instances with it, and +an evaluation container that wedges is booked as an ordinary `error` — accounted +for, never retried, and silently subtracted from the score. + +`swe_bench_fleet` addresses those six gaps and nothing else. It reuses the +service HTTP protocol, the Docker/Pyxis runtimes, the exact instance-id binding, +the artifact allow-list and the secret redaction unchanged. + +## Configuration + +```yaml +datasets: + - name: swe_bench + accuracy_config: + eval_method: swe_bench_fleet + extras: + swebench_service_urls: + - http://swe-host-1:18080 + - http://swe-host-2:18080 + swebench_service_auth_token: ${SWEBENCH_TOKEN} + num_instances: 200 + shard_size: 10 # instances per unit; 200 / 10 = 20 units + max_attempts: 3 + expected_model: Org/Model-FP8 # optional; gates checkpoint identity + min_prompt_tokens: 2000 # tool-call gate scale floor + stall_timeout_s: 10800 +``` + +| Extra | Default | Meaning | +| --- | --- | --- | +| `swebench_service_urls` | required | One URL per service host. Duplicates are refused: two entries for one host is not extra capacity, it is two runs contending for the same container runtime. | +| `shard_size` | 10 | Instances per unit. | +| `max_attempts` | 3 | Counted attempts before a unit is abandoned. Environment faults are not counted. | +| `expected_model` | none | When set, every endpoint must serve exactly this checkpoint. | +| `min_prompt_tokens` | 2000 | Floor the tool-call gate must prove it reaches. `0` waives the proof. | +| `stall_timeout_s` | 10800 | A service completing no unit in this long is quarantined even if healthy. | + +## What runs, in order + +1. **Preflight gates** (`preflight()`, before the benchmark starts) — every + service's `/health`, plus the endpoint gates below. Any failure raises + `SetupError` before a single instance is dispatched, and every gate runs so + one preflight reports every problem. +2. **Plan** — the instance list is split into units, and the plan is written + once to `units.json` with a digest over the run id and the ordered ids. +3. **Dispatch** — one in-flight service run per service; each service loop + claims a unit, submits it, polls, downloads artifacts, classifies, and + publishes or retries. +4. **Merge gate** — `merge_run(queue, run_id)`. All-or-nothing. + +## The gates + +| Gate | Refuses | +| --- | --- | +| `CheckpointIdentityGate` | Any endpoint not serving *exactly* `expected_model`. `/get_model_info` is preferred over `/v1/models`, because the latter echoes `--served-model-name`, which is routinely identical across checkpoints. Comparison is `==`: `Org/Model` is a strict prefix of `Org/Model-FP8`, so any `startswith`/`in` test accepts FP8 as BF16. Unidentifiable means fail. | +| `ToolCallGate` | Any endpoint that does not return a well-formed tool call — right name, `arguments` parse as JSON, non-empty `command`. | +| `EndpointFingerprintGate` | Any endpoint whose identity cannot be read at all. Records a fingerprint compared again at publish time. | + +**The scale rule.** Every gate implements `assert_scale()`, it runs before the +gate's own check, and a scale failure is a *gate failure*, never a skip. +`ToolCallGate` measures its prompt with the server's own `/tokenize` and fails if +the prompt is below `min_prompt_tokens`. This exists because a tool-call gate +that exercised exactly the right operation with a 278-token prompt passed +cleanly while every prompt above 2000 tokens silently returned an empty +completion. SWE-bench prompts are all far above 2000 tokens; the gate was green +and the run scored zero. **A gate that cannot prove its scale is not a gate.** + +## The work queue + +Under `/swe_bench_wq/`: + +``` +units.json immutable plan + digest +claims//owner host, pid, boot id, plan digest, SLURM job/step +claims//hb heartbeat (mtime only) +results/.json terminal record (succeeded OR abandoned) +failed/..json one per counted attempt +failed/env/.*.json environment faults (not counted) +failed/artifacts/.attemptN/ evidence snapshot taken before a retry +``` + +A unit is available when it is in the plan and has **neither** a claim **nor** a +result. Claiming is `os.mkdir` and nothing else — `makedirs(exist_ok=True)` would +hand the unit to every caller. + +### Re-running a unit + +```bash +python scripts/swe_bench_wq.py requeue REPORT_DIR run-a.s07 +``` + +`requeue` is the **only** supported way. Deleting the result file does not +requeue anything: the claim tombstone still hides the unit. `requeue` removes the +result, the claim and the counted attempt records together, and prints exactly +what it removed. + +### Reaping abandoned claims + +```bash +python scripts/swe_bench_wq.py reap REPORT_DIR # dry run +python scripts/swe_bench_wq.py reap REPORT_DIR --apply --slurm +``` + +A claim is released only when it has no result, its heartbeat is stale, **and** +its owner is provably gone. Uncertainty never escalates: if the liveness probe +fails, times out, or returns an implausible answer, nothing is released. A false +reap gives one unit two owners, duplicate results, and a wrong denominator, with +no error anywhere. + +`SlurmStepLiveness` treats an owner as dead when its job is absent from `squeue` +**or** its step is absent from `scontrol show step` while the job lives — a step +can die inside a live job, and the job-level rule alone then blocks those units +for the whole allocation. Step liveness never uses `squeue -s`, which reports +only `.extern` on the clusters this targets and would mark every live step dead. + +## Classification and retry + +Every unit is classified after the service reports success. The rule list is +ordered and first-match-wins; the order is load-bearing. + +| Kind | Class | Why | +| --- | --- | --- | +| `container_fork_eagain`, `container_exec_refused`, `runtime_read_timeout`, `image_build_timeout`, `image_build_error`, `step_infrastructure_failure`, `endpoint_changed` | infra → **retry** | Defects in infrastructure we provided. | +| `test_timeout` | genuine | A patch that makes the suite loop is a failing patch. | +| `test_memory_exceeded` | genuine | A patch that makes a graded test allocate without bound is a failing patch. The alternative to killing it was never "the test passes", it was "the host OOMs and the instance still never completes". | +| `patch_apply_failed` | genuine | The model emitted a diff that does not apply. SWE-bench books it as `error`, but it is model behaviour. | +| `unknown` | genuine | **The bias rule.** | + +**The bias rule is deliberately asymmetric.** An error that cannot be classified +confidently is genuine, never infrastructure. A false bad-run costs one redo; a +false retry biases the measurement toward optimism, and an optimistic accuracy +number is worse than no number. + +`endpoint_changed` deserves its own note: an engine restarted under a live client +yields a run that scores near zero and exits successfully, and nothing in the +result distinguishes it from a genuinely bad model. The endpoint fingerprint +recorded at claim time is re-read at publish time; a change requeues the unit. + +### Attempt accounting + +* **Environment fault** (service unreachable, submit failed) — recorded under + `failed/env/`, does **not** consume the attempt budget, and counts toward + quarantining that service. A broken host is a property of the host, not of the + unit. +* **Infra / failed** — counted. After `max_attempts` the unit is published as + `abandoned` and its claim released, so it stops burning capacity and shows up + loudly in the merge gate instead of spinning forever. + +Before every retry, the small files that explain the failure are snapshotted to +`failed/artifacts/.attemptN/`, because the unit's run directory is reused +and a unit that fails then succeeds would otherwise leave only the success's +artifacts behind. + +## The merge gate + +`merge_run(queue, run_id)` produces a number only when **all** hold: + +1. every planned unit has a terminal result; +2. no result is abandoned; +3. every unit's accounted instance **ids** equal its planned ids exactly — a set + comparison, never a count, because a shard with one duplicate and one missing + id has the right count and the wrong content; +4. the union across units equals the plan, with no id claimed twice; +5. every result carries the plan's digest; +6. no unit lost instances to infrastructure. + +Otherwise it raises `MergeRefusal` listing every reason. There is no force flag, +no partial-credit path, and **no `merge_all`**: `run_id` is required, and merging +"everything that looks finished" once combined hundreds of banked results from +unrelated configurations into one number. + +`verify_inventory()` cross-checks three independently produced views — the plan, +the claim directory and the result directory — and treats disagreement as an +error. Checking one view against itself is how a verification pass agrees with a +broken system. + +## Resource guards + +`MemoryGuard` kills a graded test only when its resident memory is at or above +`kill_bytes` (default 150 GiB) **and** it has a container-supervisor ancestor. +There is deliberately no working-directory term: an earlier version required a +cwd inside the testbed and skipped a runaway that had grown to 667 GiB because +its cwd was `/tmp`. Every extra conjunct is another way for the guard to miss +what it exists to catch. + +Two rules are enforced by construction and by test: + +* **Kill by pid, never by pattern.** There is no `pkill`/`pgrep` path in the + module and it never shells out; `kill_by_pid` refuses this process and its + ancestors. A pattern can match the guard's own command line, and a long-lived + daemon can carry a dead process's argv for days. +* **A conjunctive guard must not degenerate.** `combine_terms()` returns + `INDETERMINATE`, never `UNHEALTHY`, when any term has zero evidence. An + AND-guard whose honest term loses its data source collapses into its weaker + clauses and starts firing on healthy targets. + +The kill marker is written *before* the signal, and its phase is load-bearing: +only `eval.*` markers make an instance's error a genuine failure. An `agent` +kill merely makes one tool call return an error observation, so it is recorded +for audit and must not influence classification. An unresolvable container name +fails closed to `unknown`. + +## Operational notes + +* `--kill-on-bad-exit=0` does **not** prevent a scheduler force-terminating a + whole step when one node OOMs; OOM escalation is separate from task exit codes. + The defence is small, independently retryable units plus `MemoryGuard` acting + before the host dies — not the flag. +* A service that answers `/health` while completing nothing is the silent + failure. Verify the effect, never the status: the dispatcher quarantines a + service that has completed no unit within `stall_timeout_s` and requeues its + work. +* In shell tooling around this queue, remember `grep -c` exits 1 on zero + matches; a pipeline under `set -e` will abort on an empty, correct answer. + +## What is intentionally not here + +Cluster-lifecycle machinery — allocation rotation, holder chaining, image-store +construction and distribution, node suspect lists, multi-configuration campaign +bookkeeping — is out of scope for a benchmark client. The Pyxis runtime pulls +per-instance images from a registry and never builds a store, so that whole +class of image-corruption failure is architecturally absent rather than worked +around. diff --git a/scripts/swe_bench_wq.py b/scripts/swe_bench_wq.py new file mode 100644 index 000000000..b82648df5 --- /dev/null +++ b/scripts/swe_bench_wq.py @@ -0,0 +1,174 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Operator tool for a distributed SWE-bench work queue. + + swe_bench_wq.py status REPORT_DIR + swe_bench_wq.py merge REPORT_DIR --run-id RUN + swe_bench_wq.py requeue REPORT_DIR UNIT_ID [UNIT_ID ...] + swe_bench_wq.py reap REPORT_DIR [--apply] + +Two deliberate omissions: + +* There is no ``merge --all``. A merge is always scoped to one run id; merging + "everything that looks finished" once combined hundreds of banked results + from unrelated configurations into a single number. +* There is no way to re-run a unit other than ``requeue``. Deleting a result + file does not requeue anything, because the claim tombstone still hides the + unit; ``requeue`` removes the result, the claim and the attempt records + together and prints exactly what it removed. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from inference_endpoint.evaluation.swe_bench_distributed.fleet import ( # noqa: E402 + QUEUE_DIRNAME, +) +from inference_endpoint.evaluation.swe_bench_distributed.merge import ( # noqa: E402 + MergeRefusal, + merge_run, + verify_inventory, +) +from inference_endpoint.evaluation.swe_bench_distributed.queue import ( # noqa: E402 + WorkQueue, +) +from inference_endpoint.evaluation.swe_bench_distributed.reaper import ( # noqa: E402 + LocalProcessLiveness, + SlurmStepLiveness, + reap, +) + + +def _open(report_dir: Path) -> WorkQueue: + root = report_dir / QUEUE_DIRNAME + if not root.exists(): + root = report_dir + return WorkQueue.open(root) + + +def cmd_status(args: argparse.Namespace) -> int: + queue = _open(args.report_dir) + results = queue.results() + claimed = queue.claimed_unit_ids() + inventory = verify_inventory(queue) + print(f"run: {queue.plan.run_id}") + print(f"plan digest: {queue.plan.digest[:16]}") + print(f"units: {len(queue.plan.units)}") + print(f" with result: {len(results)}") + print(f" claimed: {len(claimed)}") + print(f" available: {len(queue.available_unit_ids())}") + abandoned = [uid for uid, result in results.items() if result.abandoned] + if abandoned: + print(f" ABANDONED: {len(abandoned)} -> {', '.join(sorted(abandoned)[:8])}") + infra = [uid for uid, result in results.items() if result.infra_error_count] + if infra: + print(f" infra-damaged:{len(infra)} -> {', '.join(sorted(infra)[:8])}") + if not inventory.consistent: + print("\nINVENTORY DISAGREEMENT (claims, results and ids do not agree):") + for label, values in ( + ("missing results", inventory.missing_units), + ("results outside the plan", inventory.foreign_units), + ("unreadable results", inventory.unreadable_units), + ("ownerless claims", inventory.ownerless_claims), + ): + if values: + print(f" {label}: {len(values)} -> {', '.join(values[:8])}") + return 0 + + +def cmd_merge(args: argparse.Namespace) -> int: + queue = _open(args.report_dir) + try: + result = merge_run(queue, args.run_id) + except MergeRefusal as exc: + print(f"REFUSED to score run {exc.run_id}:") + for reason in exc.reasons: + print(f" - {reason}") + return 1 + print(json.dumps(result.to_dict(), indent=2)) + return 0 + + +def cmd_requeue(args: argparse.Namespace) -> int: + queue = _open(args.report_dir) + for unit_id in args.unit_ids: + removed = queue.requeue(unit_id) + total = sum(len(paths) for paths in removed.values()) + print(f"{unit_id}: removed {total} record(s)") + for kind, paths in removed.items(): + for path in paths: + print(f" {kind}: {path}") + if total == 0: + print(" (nothing to remove; the unit was already runnable)") + return 0 + + +def cmd_reap(args: argparse.Namespace) -> int: + queue = _open(args.report_dir) + liveness = SlurmStepLiveness() if args.slurm else LocalProcessLiveness() + report = reap( + queue, + liveness, + stale_after_s=args.stale, + step_stale_after_s=args.step_stale, + apply=args.apply, + ) + verb = "released" if args.apply else "would release" + print(f"{verb} {len(report.released)} claim(s)") + for unit_id in report.released: + print(f" {unit_id}") + if args.verbose: + for unit_id, reason in sorted(report.kept.items()): + print(f" kept {unit_id}: {reason}") + if not args.apply and report.released: + print("\nthis was a dry run; pass --apply to release") + return 0 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + sub = parser.add_subparsers(dest="command", required=True) + + status = sub.add_parser("status", help="summarise the queue") + status.add_argument("report_dir", type=Path) + status.set_defaults(func=cmd_status) + + merge = sub.add_parser("merge", help="score exactly one run") + merge.add_argument("report_dir", type=Path) + merge.add_argument( + "--run-id", + required=True, + help="required; a merge is always scoped to one run", + ) + merge.set_defaults(func=cmd_merge) + + requeue = sub.add_parser( + "requeue", help="make units runnable again (result + claim + attempts)" + ) + requeue.add_argument("report_dir", type=Path) + requeue.add_argument("unit_ids", nargs="+") + requeue.set_defaults(func=cmd_requeue) + + reap_parser = sub.add_parser("reap", help="release claims whose owner is gone") + reap_parser.add_argument("report_dir", type=Path) + reap_parser.add_argument("--apply", action="store_true", help="actually release") + reap_parser.add_argument("--slurm", action="store_true", help="use SLURM liveness") + reap_parser.add_argument("--stale", type=float, default=3600.0) + reap_parser.add_argument("--step-stale", type=float, default=900.0) + reap_parser.add_argument("--verbose", action="store_true") + reap_parser.set_defaults(func=cmd_reap) + + args = parser.parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/inference_endpoint/config/schema.py b/src/inference_endpoint/config/schema.py index 4b72244bd..c871f3124 100644 --- a/src/inference_endpoint/config/schema.py +++ b/src/inference_endpoint/config/schema.py @@ -143,6 +143,7 @@ class ScorerMethod(str, Enum): BFCL_V4 = "bfcl_v4" LEGACY_MLPERF_DEEPSEEK_R1 = "legacy_mlperf_deepseek_r1" SWE_BENCH = "swe_bench_scorer" + SWE_BENCH_FLEET = "swe_bench_fleet" # Audit configuration; runnable test registry lives in compliance/. @@ -1126,6 +1127,9 @@ def _resolve_and_validate(self) -> Self: if not self.model_params.name: raise ValueError("Required: --model-params.name [--model]") + # Only the single-service scorer is limited to one endpoint. The fleet + # scorer runs many service runs, each against one endpoint, so it has + # no such restriction. uses_swe_bench = any( dataset.accuracy_config is not None and dataset.accuracy_config.eval_method == ScorerMethod.SWE_BENCH @@ -1271,7 +1275,8 @@ def _resolve_and_validate(self) -> Self: acc = ds.accuracy_config if ( acc is not None - and acc.eval_method == ScorerMethod.SWE_BENCH + and acc.eval_method + in (ScorerMethod.SWE_BENCH, ScorerMethod.SWE_BENCH_FLEET) and (acc.extras is None or acc.extras.get("workers") is None) ): new_extras = {**(acc.extras or {}), "workers": concurrency} diff --git a/src/inference_endpoint/evaluation/scoring.py b/src/inference_endpoint/evaluation/scoring.py index 790a17bcb..d626bf9d5 100644 --- a/src/inference_endpoint/evaluation/scoring.py +++ b/src/inference_endpoint/evaluation/scoring.py @@ -2143,5 +2143,8 @@ def score_breakdown(self) -> dict[str, Any] | None: return self._breakdown -# Late import registers the extracted scorer without introducing a cycle. +# Late imports register the extracted scorers without introducing a cycle. +from .swe_bench_fleet_scorer import ( # noqa: E402 + SWEBenchFleetScorer as SWEBenchFleetScorer, +) from .swe_bench_scorer import SWEBenchScorer as SWEBenchScorer # noqa: E402 diff --git a/src/inference_endpoint/evaluation/swe_bench_distributed/fleet.py b/src/inference_endpoint/evaluation/swe_bench_distributed/fleet.py new file mode 100644 index 000000000..44723dabc --- /dev/null +++ b/src/inference_endpoint/evaluation/swe_bench_distributed/fleet.py @@ -0,0 +1,434 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Fan a SWE-bench accuracy run out across a fleet of SWE-bench services.""" + +from __future__ import annotations + +import logging +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any +from urllib.parse import urljoin + +import msgspec +import yaml + +from ...exceptions import SetupError +from .classify import classify_unit +from .gates import ( + CheckpointIdentityGate, + EndpointFingerprintGate, + Gate, + GateFailure, + ToolCallGate, + run_gates, +) +from .merge import MergeRefusal, merge_run +from .queue import UnitOutcome, UnitResult, WorkQueue +from .reaper import LocalProcessLiveness, reap +from .units import Unit, plan_units + +logger = logging.getLogger(__name__) + +QUEUE_DIRNAME = "swe_bench_wq" +UNITS_DIRNAME = "units" + +#: Buckets a SWE-bench run report uses for instances that reached an outcome. +#: ``incomplete_ids`` is deliberately absent: an incomplete instance is exactly +#: what "not accounted for" means, and it must fail the merge gate. +ACCOUNTED_ID_KEYS = ( + "resolved_ids", + "unresolved_ids", + "empty_patch_ids", + "error_ids", +) + + +class ServiceQuarantined(RuntimeError): + """A service was withdrawn from the fleet.""" + + +@dataclass(slots=True) +class ServiceState: + """Per-service bookkeeping for the dispatcher.""" + + url: str + completed_units: int = 0 + consecutive_env_faults: int = 0 + last_progress_at: float = field(default_factory=time.monotonic) + quarantined_reason: str | None = None + + @property + def available(self) -> bool: + return self.quarantined_reason is None + + +@dataclass(slots=True) +class DispatchOutcome: + """What one attempt at one unit produced.""" + + result: UnitResult + terminal: bool + + +class FleetDispatcher: + """Claim units, run them on services, classify, retry, and merge. + + Concurrency is one in-flight service run per service. The service itself + parallelises within a run (``workers`` / ``max_eval_workers``), so a second + concurrent run per service would only contend for the same host. + """ + + def __init__( + self, + *, + queue: WorkQueue, + service_urls: list[str], + submit: Any, + poll: Any, + collect: Any, + fingerprint: Any = None, + max_attempts: int = 3, + stall_timeout_s: float = 3 * 60 * 60, + max_consecutive_env_faults: int = 3, + idle_poll_s: float = 1.0, + unit_root: Path | None = None, + killed_dir: Path | None = None, + ) -> None: + if not service_urls: + raise SetupError("the SWE-bench fleet needs at least one service URL") + self.queue = queue + self.services = {url: ServiceState(url=url) for url in service_urls} + self.submit = submit + self.poll = poll + self.collect = collect + self.fingerprint = fingerprint + self.max_attempts = max_attempts + self.stall_timeout_s = stall_timeout_s + self.max_consecutive_env_faults = max_consecutive_env_faults + self.idle_poll_s = idle_poll_s + self.unit_root = unit_root + self.killed_dir = killed_dir + self._lock = threading.Lock() + self._in_flight = 0 + + # ------------------------------------------------------------ dispatch -- + + def run(self) -> None: + """Drive every planned unit to a terminal result.""" + with ThreadPoolExecutor(max_workers=len(self.services)) as pool: + futures = [ + pool.submit(self._service_loop, url) for url in list(self.services) + ] + for future in futures: + future.result() + + def _service_loop(self, url: str) -> None: + state = self.services[url] + while state.available: + fingerprint = self._fingerprint(url) + unit = self._take_unit(fingerprint) + if unit is None: + # An empty queue does not mean the run is finished. A unit in + # flight on another service can be released back at any moment + # -- a failed attempt, a quarantined peer -- and a worker that + # exits on the first empty poll leaves that unit for nobody. + # Only "nothing available AND nothing in flight" ends the run. + if self._idle_is_terminal(): + return + time.sleep(self.idle_poll_s) + continue + try: + outcome = self._attempt(unit, state, fingerprint) + self._settle(unit, state, outcome) + finally: + with self._lock: + self._in_flight -= 1 + self._check_stall(state) + + def _take_unit(self, fingerprint: str | None) -> Unit | None: + """Claim the next available unit and mark it in flight, atomically. + + Claiming and counting must happen under one lock. With the increment + after the claim there is a window in which a peer sees "nothing + available" (this unit is claimed) and "nothing in flight" (not yet + counted), concludes the run is over, and exits -- leaving the unit with + nobody to retry it if this attempt fails. + """ + with self._lock: + for unit_id in self.queue.available_unit_ids(): + if self.queue.claim(unit_id, endpoint_fingerprint=fingerprint) is None: + continue # another process won the filesystem claim + self._in_flight += 1 + return self.queue.plan.unit(unit_id) + return None + + def _idle_is_terminal(self) -> bool: + with self._lock: + return self._in_flight == 0 and not self.queue.available_unit_ids() + + def _fingerprint(self, url: str) -> str | None: + if self.fingerprint is None: + return None + try: + return self.fingerprint() + except Exception: # noqa: BLE001 - a fingerprint is advisory at claim time + logger.debug("could not fingerprint endpoints for %s", url, exc_info=True) + return None + + def _attempt( + self, unit: Unit, state: ServiceState, claim_fingerprint: str | None + ) -> DispatchOutcome: + started = time.monotonic() + base = UnitResult( + unit_id=unit.unit_id, + run_id=unit.run_id, + plan_digest=self.queue.plan.digest, + outcome=UnitOutcome.FAILED, + service_url=state.url, + endpoint_fingerprint=claim_fingerprint, + ) + try: + service_run_id = self.submit(state.url, unit) + except Exception as exc: # noqa: BLE001 - any submit failure is the host's + base.outcome = UnitOutcome.ENV_FAULT + base.detail = f"submit failed: {type(exc).__name__}: {exc}" + base.duration_s = time.monotonic() - started + return DispatchOutcome(result=base, terminal=False) + + base.service_run_id = service_run_id + try: + status = self.poll(state.url, service_run_id) + except Exception as exc: # noqa: BLE001 + base.outcome = UnitOutcome.ENV_FAULT + base.detail = f"poll failed: {type(exc).__name__}: {exc}" + base.duration_s = time.monotonic() - started + return DispatchOutcome(result=base, terminal=False) + + base.duration_s = time.monotonic() - started + if status.get("status") != "succeeded": + base.outcome = UnitOutcome.FAILED + base.detail = ( + f"service run ended {status.get('status')}: {status.get('error')}" + ) + return DispatchOutcome(result=base, terminal=False) + + report, output_dir = self.collect(state.url, service_run_id, unit, status) + # An engine restarted mid-unit yields a plausible run that scores near + # zero and exits successfully. Comparing the fingerprint is the only + # thing that distinguishes it from a genuinely bad model. + publish_fingerprint = self._fingerprint(state.url) + endpoint_changed = ( + claim_fingerprint is not None + and publish_fingerprint is not None + and claim_fingerprint != publish_fingerprint + ) + + accounted, resolved = accounted_and_resolved(report) + classification = classify_unit( + output_dir, + report.get("error_ids"), + killed_dir=self.killed_dir, + infrastructure_failure=bool(report.get("infrastructure_failure")), + endpoint_changed=endpoint_changed, + ) + base.accounted_instance_ids = accounted + base.resolved_instance_ids = resolved + base.infra_error_count = classification.infra_count + base.genuine_error_count = classification.genuine_count + base.error_kinds = classification.as_counts() + + if classification.should_retry: + # The agent phase succeeded and the service said so, but instances + # were lost to infrastructure. Publishing this as a success is how a + # run silently becomes unable to ever reach a full result. + base.outcome = UnitOutcome.INFRA + base.detail = ( + f"{classification.infra_count} instance(s) lost to infrastructure: " + + ", ".join(f"{k}={v}" for k, v in base.error_kinds.items()) + ) + return DispatchOutcome(result=base, terminal=False) + + missing = set(unit.instance_ids) - set(accounted) + if missing: + base.outcome = UnitOutcome.FAILED + base.detail = f"{len(missing)} instance(s) unaccounted for" + return DispatchOutcome(result=base, terminal=False) + + base.outcome = UnitOutcome.SUCCEEDED + return DispatchOutcome(result=base, terminal=True) + + def _settle( + self, unit: Unit, state: ServiceState, outcome: DispatchOutcome + ) -> None: + result = outcome.result + if outcome.terminal: + self.queue.publish(result) + state.completed_units += 1 + state.consecutive_env_faults = 0 + state.last_progress_at = time.monotonic() + return + + if self.unit_root is not None: + attempt = self.queue.attempts(unit.unit_id) + 1 + self.queue.snapshot_evidence( + unit.unit_id, self.unit_root / unit.unit_id, attempt + ) + + attempts = self.queue.record_attempt(result) + + if result.outcome is UnitOutcome.ENV_FAULT: + # The unit is fine; the host is not. Do not charge the unit, and + # withdraw the service if it keeps doing this. + state.consecutive_env_faults += 1 + if state.consecutive_env_faults >= self.max_consecutive_env_faults: + state.quarantined_reason = ( + f"{state.consecutive_env_faults} consecutive environment faults" + ) + logger.error( + "quarantining SWE-bench service %s: %s", + state.url, + state.quarantined_reason, + ) + self.queue.release(unit.unit_id) + return + + state.consecutive_env_faults = 0 + if attempts >= self.max_attempts: + # Stop burning slots. An abandoned unit is a loud, terminal record + # that the merge gate refuses, not a unit that spins forever. + self.queue.abandon(result) + state.last_progress_at = time.monotonic() + return + self.queue.release(unit.unit_id) + + def _check_stall(self, state: ServiceState) -> None: + """Withdraw a service that is healthy but not producing. + + Health is not progress. A service that answers ``/health`` while + completing nothing is the silent failure mode: verify the effect, never + the status. + """ + if not state.available: + return + idle = time.monotonic() - state.last_progress_at + if idle > self.stall_timeout_s: + state.quarantined_reason = ( + f"no unit completed in {idle:.0f}s despite a healthy service" + ) + logger.error( + "quarantining SWE-bench service %s: %s", + state.url, + state.quarantined_reason, + ) + + @property + def quarantined(self) -> dict[str, str]: + return { + url: state.quarantined_reason + for url, state in self.services.items() + if state.quarantined_reason is not None + } + + +def accounted_and_resolved( + report: dict[str, Any], +) -> tuple[tuple[str, ...], tuple[str, ...]]: + """Extract the accounted and resolved instance ids from a SWE-bench report. + + Ids, not counts. A shard with one duplicate and one missing id has the right + count and the wrong content, and only an id comparison catches it. + """ + accounted: list[str] = [] + seen: set[str] = set() + for key in ACCOUNTED_ID_KEYS: + for instance_id in report.get(key) or (): + text = str(instance_id) + if text in seen: + # Preserve the duplicate so the merge gate can refuse it rather + # than silently deduplicating a real accounting bug. + accounted.append(text) + continue + seen.add(text) + accounted.append(text) + resolved = tuple(str(x) for x in report.get("resolved_ids") or ()) + return tuple(accounted), resolved + + +def build_gates( + *, + expected_model: str | None, + tool_call_model: str | None, + min_prompt_tokens: int, + api_key: str | None = None, +) -> tuple[list[Gate], EndpointFingerprintGate]: + """Assemble the pre-dispatch gates. + + ``expected_model`` is optional only because not every deployment pins a + checkpoint path; when it is set the identity gate is mandatory. + """ + fingerprint_gate = EndpointFingerprintGate(api_key=api_key) + gates: list[Gate] = [] + if expected_model: + gates.append(CheckpointIdentityGate(expected_model, api_key=api_key)) + if tool_call_model: + gates.append( + ToolCallGate( + tool_call_model, + min_prompt_tokens=min_prompt_tokens, + api_key=api_key, + ) + ) + gates.append(fingerprint_gate) + return gates, fingerprint_gate + + +def load_benchmark_config(report_dir: Path) -> dict[str, Any]: + config_path = report_dir / "config.yaml" + if not config_path.exists(): + raise FileNotFoundError( + f"config.yaml not found at {config_path}. The fleet scorer must run " + "inside a benchmark that has already written its config." + ) + with config_path.open() as handle: + config = yaml.safe_load(handle) + if not isinstance(config, dict): + raise ValueError(f"benchmark config at {config_path} must be a YAML mapping") + return config + + +def write_merge_artifacts( + report_dir: Path, payload: dict[str, Any], name: str = "swe_bench_merge.json" +) -> Path: + path = report_dir / name + tmp = path.with_name(f".{path.name}.tmp") + tmp.write_bytes(msgspec.json.encode(payload)) + tmp.replace(path) + return path + + +__all__ = [ + "ACCOUNTED_ID_KEYS", + "QUEUE_DIRNAME", + "UNITS_DIRNAME", + "DispatchOutcome", + "FleetDispatcher", + "GateFailure", + "LocalProcessLiveness", + "MergeRefusal", + "ServiceQuarantined", + "ServiceState", + "accounted_and_resolved", + "build_gates", + "load_benchmark_config", + "merge_run", + "plan_units", + "reap", + "run_gates", + "urljoin", + "write_merge_artifacts", +] diff --git a/src/inference_endpoint/evaluation/swe_bench_fleet_scorer.py b/src/inference_endpoint/evaluation/swe_bench_fleet_scorer.py new file mode 100644 index 000000000..2848d9213 --- /dev/null +++ b/src/inference_endpoint/evaluation/swe_bench_fleet_scorer.py @@ -0,0 +1,392 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""SWE-bench accuracy scorer that fans out across a fleet of SWE-bench services. + +:class:`~inference_endpoint.evaluation.swe_bench_scorer.SWEBenchScorer` runs the +whole instance list as one service run. This scorer shards it, runs the shards +concurrently on several services, refuses to score a run whose instances are not +all accounted for, and retries the shards that lost instances to infrastructure +rather than the model. + +Configured entirely through ``accuracy_config.extras``:: + + accuracy_config: + eval_method: swe_bench_fleet + extras: + swebench_service_urls: + - http://swe-host-1:18080 + - http://swe-host-2:18080 + shard_size: 10 + max_attempts: 3 + expected_model: Org/Model-FP8 # optional; gates the checkpoint + min_prompt_tokens: 2000 +""" + +from __future__ import annotations + +import logging +import time +from pathlib import Path +from typing import Any, ClassVar +from urllib.parse import urljoin + +import msgspec + +from ..dataset_manager.dataset import Dataset +from ..exceptions import SetupError +from .extractor import Extractor +from .scoring import Scorer +from .swe_bench_distributed.fleet import ( + QUEUE_DIRNAME, + UNITS_DIRNAME, + FleetDispatcher, + accounted_and_resolved, + build_gates, + load_benchmark_config, + write_merge_artifacts, +) +from .swe_bench_distributed.gates import GateFailure, run_gates +from .swe_bench_distributed.merge import MergeRefusal, merge_run +from .swe_bench_distributed.queue import WorkQueue +from .swe_bench_distributed.units import Unit, plan_units +from .swe_bench_scorer import SWEBenchScorer + +logger = logging.getLogger(__name__) + + +class SWEBenchFleetScorer(Scorer, scorer_id="swe_bench_fleet"): + """Distributed SWE-bench scoring across N services.""" + + REQUIRES_EXTRACTOR: ClassVar[bool] = False + SKIP_ENDPOINT_PHASE: ClassVar[bool] = True + DEFAULT_SHARD_SIZE: ClassVar[int] = 10 + DEFAULT_MAX_ATTEMPTS: ClassVar[int] = 3 + DEFAULT_MIN_PROMPT_TOKENS: ClassVar[int] = 2000 + DEFAULT_STALL_TIMEOUT_S: ClassVar[int] = 3 * 60 * 60 + DEFAULT_SERVICE_TIMEOUT_S: ClassVar[int] = 24 * 60 * 60 + DEFAULT_POLL_INTERVAL_S: ClassVar[float] = 5.0 + + def __init__( + self, + dataset_name: str, + dataset: Dataset, + report_dir: Any, + extractor: type[Extractor] | None = None, + ground_truth_column: str | None = "instance_id", + **extras: Any, + ) -> None: + super().__init__( + dataset_name=dataset_name, + dataset=dataset, + report_dir=report_dir, + extractor=extractor, + ground_truth_column=ground_truth_column or "instance_id", + ) + self.report_dir = self.report_dir.resolve() + self.options = self._resolve_options(extras) + + # --------------------------------------------------------------- config -- + + @classmethod + def _service_urls(cls, extras: dict[str, Any]) -> list[str]: + raw = extras.get("swebench_service_urls") + if raw is None: + single = extras.get("swebench_service_url") + raw = [single] if single else [] + if isinstance(raw, str): + raw = [part.strip() for part in raw.split(",") if part.strip()] + urls = [SWEBenchScorer._normalize_service_url(url) for url in raw or []] + if not urls: + raise SetupError( + "accuracy_config.extras.swebench_service_urls is required for " + "swe_bench_fleet; list one URL per SWE-bench service host." + ) + duplicates = sorted({url for url in urls if urls.count(url) > 1}) + if duplicates: + # Two entries for one host is not extra capacity; it is two + # concurrent runs contending for the same Docker/Pyxis runtime. + raise SetupError( + "duplicate SWE-bench service URLs: " + ", ".join(duplicates) + ) + return urls + + @classmethod + def _resolve_options(cls, extras: dict[str, Any]) -> dict[str, Any]: + options = dict(SWEBenchScorer._resolve_dataset_options(extras)) + options["service_urls"] = cls._service_urls(extras) + options["auth_token"] = extras.get("swebench_service_auth_token") or None + options["num_instances"] = SWEBenchScorer._get_extra_int( + extras, + "num_instances", + default=SWEBenchScorer.DEFAULT_NUM_INSTANCES, + min_value=1, + ) + options["shard_size"] = SWEBenchScorer._get_extra_int( + extras, "shard_size", default=cls.DEFAULT_SHARD_SIZE, min_value=1 + ) + options["workers"] = SWEBenchScorer._get_extra_int( + extras, "workers", default=SWEBenchScorer.DEFAULT_WORKERS, min_value=1 + ) + options["max_eval_workers"] = SWEBenchScorer._get_extra_int( + extras, + "max_eval_workers", + default=SWEBenchScorer.DEFAULT_MAX_EVAL_WORKERS, + min_value=1, + ) + options["max_attempts"] = SWEBenchScorer._get_extra_int( + extras, "max_attempts", default=cls.DEFAULT_MAX_ATTEMPTS, min_value=1 + ) + options["min_prompt_tokens"] = SWEBenchScorer._get_extra_int( + extras, + "min_prompt_tokens", + default=cls.DEFAULT_MIN_PROMPT_TOKENS, + min_value=0, + ) + options["stall_timeout_s"] = SWEBenchScorer._get_extra_int( + extras, + "stall_timeout_s", + default=cls.DEFAULT_STALL_TIMEOUT_S, + min_value=1, + ) + options["service_timeout_s"] = SWEBenchScorer._get_extra_int( + extras, + "service_timeout_s", + default=cls.DEFAULT_SERVICE_TIMEOUT_S, + min_value=1, + ) + options["poll_interval_s"] = SWEBenchScorer._get_extra_float( + extras, + "poll_interval_s", + default=cls.DEFAULT_POLL_INTERVAL_S, + min_value=0, + ) + options["swebench_template"] = SWEBenchScorer._resolve_service_template(extras) + options["expected_model"] = extras.get("expected_model") or None + options["run_id"] = str(extras.get("run_id") or "swe_bench") + return options + + @classmethod + def dataset_loader_kwargs(cls, extras: dict[str, Any]) -> dict[str, Any]: + return SWEBenchScorer._resolve_dataset_options(extras) + + @classmethod + def external_sample_count(cls, extras: dict[str, Any]) -> int | None: + return SWEBenchScorer.external_sample_count(extras) + + # ------------------------------------------------------------ preflight -- + + @classmethod + def preflight( + cls, extras: dict[str, Any], *, loaded_sample_count: int | None = None + ) -> None: + """Health-check every service and run the pre-dispatch gates. + + Every problem is reported from one preflight. A run that starts against + a mis-served checkpoint, or against an endpoint that cannot emit a tool + call at SWE-bench prompt scale, produces a plausible-looking low score + hours later and costs the whole run. + """ + options = cls._resolve_options(extras) + for url in options["service_urls"]: + SWEBenchScorer._check_health(url, options["auth_token"]) + + endpoints = extras.get("endpoint_urls") or [] + if not endpoints: + logger.info( + "swe_bench_fleet: no endpoint URLs available at preflight; " + "checkpoint and tool-call gates run at dispatch instead" + ) + return + gates, _ = build_gates( + expected_model=options["expected_model"], + tool_call_model=extras.get("model_name"), + min_prompt_tokens=options["min_prompt_tokens"], + api_key=extras.get("endpoint_api_key"), + ) + try: + run_gates(gates, list(endpoints)) + except GateFailure as exc: + raise SetupError(str(exc)) from exc + + def score_single_sample(self, value: str, ground_truth: str) -> float: + raise RuntimeError( + "SWEBenchFleetScorer scores whole units through services; call score()." + ) + + # ---------------------------------------------------------------- score -- + + def score(self) -> tuple[float | None, int]: + self.complete = True + config = load_benchmark_config(self.report_dir) + model_params = config.get("model_params") or {} + model_name = model_params.get("name") + if not model_name: + raise ValueError("model_params.name is required in the benchmark config") + endpoint_config = config.get("endpoint_config") or {} + endpoint_urls = list(endpoint_config.get("endpoints") or []) + if not endpoint_urls: + raise SetupError("the benchmark config lists no endpoint URLs") + + instance_ids = self._instance_ids() + if not instance_ids: + logger.warning("swe_bench_fleet: no instances selected") + self.complete = False + return None, 1 + + gates, fingerprint_gate = build_gates( + expected_model=self.options["expected_model"], + tool_call_model=model_name, + min_prompt_tokens=self.options["min_prompt_tokens"], + api_key=endpoint_config.get("api_key"), + ) + try: + run_gates(gates, endpoint_urls) + except GateFailure as exc: + raise SetupError(str(exc)) from exc + + plan = plan_units( + self.options["run_id"], instance_ids, shard_size=self.options["shard_size"] + ) + queue = WorkQueue(self.report_dir / QUEUE_DIRNAME, plan) + unit_root = self.report_dir / UNITS_DIRNAME + unit_root.mkdir(parents=True, exist_ok=True) + + self._model_name = model_name + self._endpoint_urls = endpoint_urls + self._endpoint_api_key = endpoint_config.get("api_key") + self._generation_params = SWEBenchScorer._generation_params(model_params) + self._unit_root = unit_root + + def fingerprint() -> str | None: + values = [fingerprint_gate.fingerprint(url) for url in endpoint_urls] + if any(value is None for value in values): + return None + return "|".join(v for v in values if v is not None) + + dispatcher = FleetDispatcher( + queue=queue, + service_urls=self.options["service_urls"], + submit=self._submit_unit, + poll=self._poll_unit, + collect=self._collect_unit, + fingerprint=fingerprint, + max_attempts=self.options["max_attempts"], + stall_timeout_s=self.options["stall_timeout_s"], + unit_root=unit_root, + ) + dispatcher.run() + + payload: dict[str, Any] = { + "run_id": plan.run_id, + "plan_digest": plan.digest, + "services": self.options["service_urls"], + "quarantined": dispatcher.quarantined, + } + try: + merged = merge_run(queue, plan.run_id) + except MergeRefusal as exc: + payload["refused"] = exc.reasons + write_merge_artifacts(self.report_dir, payload) + logger.error("swe_bench_fleet: %s", exc) + self.complete = False + return None, 1 + + payload["merge"] = merged.to_dict() + write_merge_artifacts(self.report_dir, payload) + logger.info( + "swe_bench_fleet: resolved %d / %d (%.1f%%) across %d units", + merged.resolved_instances, + merged.total_instances, + merged.resolved_rate * 100, + merged.unit_count, + ) + return merged.resolved_rate, 1 + + # --------------------------------------------------------- service glue -- + + def _instance_ids(self) -> list[str]: + if self.dataset.dataframe is None: + raise RuntimeError( + "SWEBench dataset must be loaded before scoring; call dataset.load()." + ) + frame = self.dataset.dataframe + total = min(self.options["num_instances"], len(frame)) + return [ + str(instance_id) + for instance_id in frame.iloc[:total][self.ground_truth_column].tolist() + ] + + def _submit_unit(self, service_url: str, unit: Unit) -> str: + payload = { + "model_name": self._model_name, + # The service accepts exactly one endpoint URL per run; the fleet's + # parallelism comes from running many units, not many endpoints. + "endpoint_urls": self._endpoint_urls[:1], + "endpoint_api_key": self._endpoint_api_key, + "generation_params": self._generation_params, + "subset": self.options["subset"], + "split": self.options["split"], + "num_instances": len(unit.instance_ids), + "workers": self.options["workers"], + "max_eval_workers": self.options["max_eval_workers"], + "evaluated_instance_ids": list(unit.instance_ids), + "template": self.options["swebench_template"], + } + submitted = SWEBenchScorer._http_json( + urljoin(service_url, "v1/runs"), + method="POST", + payload=payload, + timeout_s=30.0, + auth_token=self.options["auth_token"], + ) + run_id = str(submitted.get("run_id") or "") + if not run_id: + raise SetupError(f"{service_url} did not return a run_id") + return run_id + + def _poll_unit(self, service_url: str, service_run_id: str) -> dict[str, Any]: + deadline = time.monotonic() + self.options["service_timeout_s"] + status: dict[str, Any] = {"status": "queued"} + while status.get("status") not in {"succeeded", "failed", "cancelled"}: + if time.monotonic() >= deadline: + SWEBenchScorer._cancel_service_run( + service_url, service_run_id, self.options["auth_token"] + ) + raise SetupError( + f"timed out waiting for {service_url} run {service_run_id}" + ) + time.sleep(self.options["poll_interval_s"]) + status = SWEBenchScorer._http_json( + urljoin(service_url, f"v1/runs/{service_run_id}"), + timeout_s=30.0, + auth_token=self.options["auth_token"], + ) + return status + + def _collect_unit( + self, + service_url: str, + service_run_id: str, + unit: Unit, + status: dict[str, Any], + ) -> tuple[dict[str, Any], Path]: + target = self._unit_root / unit.unit_id + target.mkdir(parents=True, exist_ok=True) + SWEBenchScorer._download_artifacts( + service_url, status, target, self.options["auth_token"] + ) + report = status.get("result") + if not isinstance(report, dict): + results_path = target / "swe_bench_results.json" + if results_path.exists(): + try: + report = msgspec.json.decode(results_path.read_bytes(), type=dict) + except msgspec.DecodeError: + report = {} + else: + report = {} + return report, target + + +__all__ = ["SWEBenchFleetScorer", "accounted_and_resolved"] diff --git a/tests/unit/evaluation/swe_bench_distributed/test_fleet.py b/tests/unit/evaluation/swe_bench_distributed/test_fleet.py new file mode 100644 index 000000000..02ee81d3b --- /dev/null +++ b/tests/unit/evaluation/swe_bench_distributed/test_fleet.py @@ -0,0 +1,293 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Fleet dispatch: fan-out, classification-driven retry, quarantine, merge.""" + +from __future__ import annotations + +import itertools + +import pytest + +from inference_endpoint.evaluation.swe_bench_distributed.fleet import ( + FleetDispatcher, + accounted_and_resolved, +) +from inference_endpoint.evaluation.swe_bench_distributed.merge import ( + MergeRefusal, + merge_run, +) +from inference_endpoint.evaluation.swe_bench_distributed.queue import ( + UnitOutcome, + WorkQueue, +) +from inference_endpoint.evaluation.swe_bench_distributed.units import plan_units + +pytestmark = pytest.mark.unit + +IDS = [f"repo__proj-{i:02d}" for i in range(30)] +SERVICES = ["http://svc-a:18080", "http://svc-b:18080"] + + +@pytest.fixture +def queue(tmp_path): + return WorkQueue(tmp_path / "wq", plan_units("run-a", IDS, shard_size=10)) + + +class FakeFleet: + """A scripted stand-in for the SWE-bench service HTTP protocol.""" + + def __init__(self, queue: WorkQueue, tmp_path, *, resolved_per_unit: int = 4): + self.queue = queue + self.tmp_path = tmp_path + self.resolved_per_unit = resolved_per_unit + self.counter = itertools.count() + self.submitted: list[tuple[str, str]] = [] + self.submit_errors: dict[str, Exception] = {} + self.status_for_unit: dict[str, str] = {} + self.error_ids_for_unit: dict[str, list[str]] = {} + self.fingerprints: list[str] = ["fp-1"] + + def submit(self, service_url, unit): + error = self.submit_errors.get(service_url) + if error is not None: + raise error + self.submitted.append((service_url, unit.unit_id)) + return f"svc-run-{next(self.counter)}" + + def poll(self, service_url, service_run_id): + unit_id = self.submitted[-1][1] + return {"status": self.status_for_unit.get(unit_id, "succeeded")} + + def collect(self, service_url, service_run_id, unit, status): + error_ids = self.error_ids_for_unit.get(unit.unit_id, []) + remaining = [i for i in unit.instance_ids if i not in error_ids] + resolved = remaining[: self.resolved_per_unit] + unresolved = remaining[self.resolved_per_unit :] + output_dir = self.tmp_path / "units" / unit.unit_id + output_dir.mkdir(parents=True, exist_ok=True) + report = { + "resolved_ids": resolved, + "unresolved_ids": unresolved, + "error_ids": error_ids, + "empty_patch_ids": [], + } + return report, output_dir + + def fingerprint(self): + return self.fingerprints[0] + + def write_eval_log(self, unit_id: str, instance_id: str, text: str) -> None: + log_dir = ( + self.tmp_path + / "units" + / unit_id + / "logs" + / "run_evaluation" + / "r" + / "m" + / instance_id + ) + log_dir.mkdir(parents=True, exist_ok=True) + (log_dir / "run_instance.log").write_text(text) + + +def dispatcher_for(queue, fleet, **overrides): + kwargs = { + "queue": queue, + "service_urls": SERVICES, + "submit": fleet.submit, + "poll": fleet.poll, + "collect": fleet.collect, + "fingerprint": fleet.fingerprint, + "max_attempts": 3, + } + kwargs.update(overrides) + return FleetDispatcher(**kwargs) + + +class TestAccounting: + def test_every_outcome_bucket_counts_as_accounted(self): + report = { + "resolved_ids": ["a"], + "unresolved_ids": ["b"], + "empty_patch_ids": ["c"], + "error_ids": ["d"], + } + accounted, resolved = accounted_and_resolved(report) + assert set(accounted) == {"a", "b", "c", "d"} + assert resolved == ("a",) + + def test_incomplete_instances_are_not_accounted(self): + # "Incomplete" is exactly what unaccounted means; counting it would let + # a partial shard through the merge gate. + accounted, _ = accounted_and_resolved( + {"resolved_ids": ["a"], "incomplete_ids": ["b"]} + ) + assert accounted == ("a",) + + def test_duplicates_are_preserved_for_the_gate_to_refuse(self): + accounted, _ = accounted_and_resolved( + {"resolved_ids": ["a"], "unresolved_ids": ["a"]} + ) + assert accounted == ("a", "a") + + +class TestHappyPath: + def test_all_units_complete_and_merge(self, queue, tmp_path): + fleet = FakeFleet(queue, tmp_path) + dispatcher_for(queue, fleet).run() + + assert len(queue.completed_unit_ids()) == 3 + result = merge_run(queue, "run-a") + assert result.total_instances == 30 + assert result.resolved_instances == 12 + + def test_work_is_spread_across_services(self, queue, tmp_path): + fleet = FakeFleet(queue, tmp_path) + dispatcher_for(queue, fleet).run() + assert len({service for service, _ in fleet.submitted}) >= 1 + assert len(fleet.submitted) == 3 + + def test_every_unit_runs_exactly_once(self, queue, tmp_path): + fleet = FakeFleet(queue, tmp_path) + dispatcher_for(queue, fleet).run() + dispatched = [unit_id for _, unit_id in fleet.submitted] + assert sorted(dispatched) == sorted(queue.plan.unit_ids) + + +class TestInfraRetry: + def test_an_eval_infra_error_requeues_a_succeeded_run(self, queue, tmp_path): + fleet = FakeFleet(queue, tmp_path) + fleet.error_ids_for_unit["run-a.s00"] = [IDS[0]] + fleet.write_eval_log("run-a.s00", IDS[0], "container state improper") + + dispatcher_for(queue, fleet, max_attempts=1).run() + + # The service said "succeeded" and every instance was accounted for, so + # nothing but classification distinguishes this from a real result. + stored = queue.result("run-a.s00") + assert stored is not None + assert stored.abandoned + assert stored.outcome is UnitOutcome.INFRA + assert stored.infra_error_count == 1 + with pytest.raises(MergeRefusal): + merge_run(queue, "run-a") + + def test_a_genuine_error_is_scored_not_retried(self, queue, tmp_path): + fleet = FakeFleet(queue, tmp_path) + fleet.error_ids_for_unit["run-a.s00"] = [IDS[0]] + fleet.write_eval_log("run-a.s00", IDS[0], "Test timed out after 1800s") + + dispatcher_for(queue, fleet).run() + + stored = queue.result("run-a.s00") + assert stored is not None + assert stored.outcome is UnitOutcome.SUCCEEDED + assert stored.genuine_error_count == 1 + assert merge_run(queue, "run-a").total_instances == 30 + + def test_a_transient_infra_error_succeeds_on_retry(self, queue, tmp_path): + fleet = FakeFleet(queue, tmp_path) + fleet.error_ids_for_unit["run-a.s00"] = [IDS[0]] + fleet.write_eval_log("run-a.s00", IDS[0], "container state improper") + + original_collect = fleet.collect + + def collect_once(service_url, service_run_id, unit, status): + result = original_collect(service_url, service_run_id, unit, status) + fleet.error_ids_for_unit.pop(unit.unit_id, None) + return result + + dispatcher_for(queue, fleet, collect=collect_once).run() + + stored = queue.result("run-a.s00") + assert stored is not None and stored.outcome is UnitOutcome.SUCCEEDED + assert queue.attempts("run-a.s00") == 1 + assert merge_run(queue, "run-a").total_instances == 30 + + def test_a_unit_is_abandoned_after_max_attempts(self, queue, tmp_path): + fleet = FakeFleet(queue, tmp_path) + fleet.status_for_unit["run-a.s00"] = "failed" + + dispatcher_for(queue, fleet, max_attempts=2).run() + + stored = queue.result("run-a.s00") + assert stored is not None and stored.abandoned + assert queue.attempts("run-a.s00") == 2 + # An abandoned unit must be loud, not silent: it stops burning slots and + # the gate refuses the run. + assert queue.claimed_unit_ids() == set() + with pytest.raises(MergeRefusal, match="abandoned"): + merge_run(queue, "run-a") + + +class TestEndpointFingerprint: + def test_a_restarted_engine_requeues_the_unit(self, queue, tmp_path): + fleet = FakeFleet(queue, tmp_path) + original_collect = fleet.collect + + def collect_then_restart(service_url, service_run_id, unit, status): + result = original_collect(service_url, service_run_id, unit, status) + if unit.unit_id == "run-a.s00" and fleet.fingerprints[0] == "fp-1": + fleet.fingerprints[0] = "fp-2" + return result + + dispatcher_for(queue, fleet, collect=collect_then_restart, max_attempts=1).run() + + stored = queue.result("run-a.s00") + assert stored is not None + # The run "succeeded" and every instance was accounted for; only the + # fingerprint says it was scored against a different engine. + assert stored.outcome is UnitOutcome.INFRA + assert "endpoint_changed" in stored.error_kinds + + +class TestEnvironmentFaults: + def test_a_submit_failure_does_not_charge_the_unit(self, queue, tmp_path): + fleet = FakeFleet(queue, tmp_path) + fleet.submit_errors[SERVICES[0]] = OSError("service host is broken") + + dispatcher_for(queue, fleet).run() + + # A broken host is a property of the host. The units still complete, on + # the other service, with a clean attempt ledger. + assert len(queue.completed_unit_ids()) == 3 + assert all(queue.attempts(unit_id) == 0 for unit_id in queue.plan.unit_ids) + assert merge_run(queue, "run-a").total_instances == 30 + + def test_a_persistently_broken_service_is_quarantined(self, queue, tmp_path): + fleet = FakeFleet(queue, tmp_path) + fleet.submit_errors[SERVICES[0]] = OSError("service host is broken") + + dispatcher = dispatcher_for(queue, fleet, max_consecutive_env_faults=2) + dispatcher.run() + + assert SERVICES[0] in dispatcher.quarantined + assert SERVICES[1] not in dispatcher.quarantined + + +class TestStallQuarantine: + def test_a_healthy_but_unproductive_service_is_withdrawn(self, queue, tmp_path): + fleet = FakeFleet(queue, tmp_path) + dispatcher = dispatcher_for(queue, fleet, stall_timeout_s=-1) + dispatcher.run() + # Health is not progress. A service answering /health while completing + # nothing is the silent failure: verify the effect, never the status. + assert dispatcher.quarantined + assert all( + "no unit completed" in reason for reason in dispatcher.quarantined.values() + ) + + +class TestResume: + def test_a_restarted_client_does_not_redo_completed_units(self, queue, tmp_path): + fleet = FakeFleet(queue, tmp_path) + dispatcher_for(queue, fleet, service_urls=SERVICES[:1]).run() + first_pass = len(fleet.submitted) + + reopened = WorkQueue.open(queue.root) + dispatcher_for(reopened, fleet, service_urls=SERVICES[:1]).run() + + assert len(fleet.submitted) == first_pass + assert merge_run(reopened, "run-a").total_instances == 30 diff --git a/tests/unit/evaluation/swe_bench_distributed/test_fleet_scorer.py b/tests/unit/evaluation/swe_bench_distributed/test_fleet_scorer.py new file mode 100644 index 000000000..b96026227 --- /dev/null +++ b/tests/unit/evaluation/swe_bench_distributed/test_fleet_scorer.py @@ -0,0 +1,78 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Configuration and registration of the fleet scorer.""" + +from __future__ import annotations + +import pytest + +from inference_endpoint.config.schema import ScorerMethod +from inference_endpoint.evaluation.scoring import Scorer +from inference_endpoint.evaluation.swe_bench_fleet_scorer import SWEBenchFleetScorer +from inference_endpoint.exceptions import SetupError + +pytestmark = pytest.mark.unit + +URLS = ["http://svc-a:18080", "http://svc-b:18080"] + + +class TestRegistration: + def test_the_scorer_is_registered(self): + assert Scorer.get("swe_bench_fleet") is SWEBenchFleetScorer + + def test_the_scorer_method_enum_is_in_sync(self): + assert ScorerMethod.SWE_BENCH_FLEET.value in Scorer.available_scorers() + + def test_it_skips_the_endpoint_phase(self): + # Like the single-service scorer, this one drives the run itself rather + # than consuming responses collected by the load generator. + assert SWEBenchFleetScorer.SKIP_ENDPOINT_PHASE + assert not SWEBenchFleetScorer.REQUIRES_EXTRACTOR + + +class TestOptions: + def test_service_urls_are_normalised(self): + options = SWEBenchFleetScorer._resolve_options( + {"swebench_service_urls": ["http://svc-a:18080/"]} + ) + assert options["service_urls"] == ["http://svc-a:18080/"] + + def test_a_comma_separated_string_is_accepted(self): + options = SWEBenchFleetScorer._resolve_options( + {"swebench_service_urls": "http://svc-a:18080, http://svc-b:18080"} + ) + assert len(options["service_urls"]) == 2 + + def test_the_single_service_key_still_works(self): + options = SWEBenchFleetScorer._resolve_options( + {"swebench_service_url": "http://svc-a:18080"} + ) + assert options["service_urls"] == ["http://svc-a:18080/"] + + def test_no_service_urls_is_a_setup_error(self): + with pytest.raises(SetupError, match="swebench_service_urls is required"): + SWEBenchFleetScorer._resolve_options({}) + + def test_duplicate_service_urls_are_refused(self): + # Two entries for one host is not extra capacity; it is two concurrent + # runs contending for the same container runtime. + with pytest.raises(SetupError, match="duplicate"): + SWEBenchFleetScorer._resolve_options( + {"swebench_service_urls": ["http://svc-a:18080", "http://svc-a:18080/"]} + ) + + def test_defaults_are_sane(self): + options = SWEBenchFleetScorer._resolve_options( + {"swebench_service_urls": URLS} + ) + assert options["shard_size"] == 10 + assert options["max_attempts"] == 3 + # The tool-call gate's floor must stay at SWE-bench prompt scale. + assert options["min_prompt_tokens"] == 2000 + + def test_a_bad_shard_size_is_rejected(self): + with pytest.raises(SetupError, match="shard_size"): + SWEBenchFleetScorer._resolve_options( + {"swebench_service_urls": URLS, "shard_size": 0} + )