diff --git a/src/inference_endpoint/evaluation/swe_bench_distributed/__init__.py b/src/inference_endpoint/evaluation/swe_bench_distributed/__init__.py index 2042c3d55..3507289bb 100644 --- a/src/inference_endpoint/evaluation/swe_bench_distributed/__init__.py +++ b/src/inference_endpoint/evaluation/swe_bench_distributed/__init__.py @@ -11,6 +11,7 @@ exactly once. """ +from .merge import MergeRefusal, MergeResult, merge_run, verify_inventory from .queue import ( ClaimError, UnitOutcome, @@ -21,10 +22,14 @@ __all__ = [ "ClaimError", + "MergeRefusal", + "MergeResult", "Unit", "UnitOutcome", "UnitPlan", "UnitResult", "WorkQueue", + "merge_run", "plan_units", + "verify_inventory", ] diff --git a/src/inference_endpoint/evaluation/swe_bench_distributed/merge.py b/src/inference_endpoint/evaluation/swe_bench_distributed/merge.py new file mode 100644 index 000000000..a8dbee7cf --- /dev/null +++ b/src/inference_endpoint/evaluation/swe_bench_distributed/merge.py @@ -0,0 +1,230 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""The merge gate: refuse to emit an accuracy unless every id is accounted for. + +The single most important property of a sharded accuracy run is that it never +divides the results of 190 instances by 200. The gate is all-or-nothing by +design: there is no force flag and no partial-credit path, because a partial +number is indistinguishable from a real one once it leaves this module. + +The gate is also scoped to exactly one run. There is no ``merge_all``. Merging +"every run that looks finished" once re-merged hundreds of banked results +belonging to unrelated configurations into one number. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from .queue import UnitOutcome, UnitResult, WorkQueue +from .units import UnitPlan + + +class MergeRefusal(RuntimeError): + """The gate refused to produce an accuracy number. + + ``reasons`` lists every independent failure, so one merge attempt reports + everything wrong rather than the first thing wrong. + """ + + def __init__(self, run_id: str, reasons: list[str]) -> None: + self.run_id = run_id + self.reasons = reasons + super().__init__( + f"refusing to score run {run_id!r}: " + + "; ".join(reasons[:10]) + + (f" (+{len(reasons) - 10} more)" if len(reasons) > 10 else "") + ) + + +@dataclass(slots=True) +class MergeResult: + run_id: str + plan_digest: str + total_instances: int + resolved_instances: int + unit_count: int + + @property + def resolved_rate(self) -> float: + return self.resolved_instances / self.total_instances + + def to_dict(self) -> dict[str, Any]: + return { + "run_id": self.run_id, + "plan_digest": self.plan_digest, + "total_instances": self.total_instances, + "resolved_instances": self.resolved_instances, + "resolved_rate": self.resolved_rate, + "unit_count": self.unit_count, + } + + +@dataclass(slots=True) +class InventoryReport: + """Cross-check of three independently produced views of the same run. + + The units the plan asked for, the units the queue recorded results for, and + the instance ids those results claim to have covered are produced by + different code paths. Checking one against itself is how a verification pass + can agree with a broken system: the instrument shares the blind spot. These + must agree with each other. + """ + + missing_units: list[str] = field(default_factory=list) + foreign_units: list[str] = field(default_factory=list) + unreadable_units: list[str] = field(default_factory=list) + ownerless_claims: list[str] = field(default_factory=list) + claims_without_results: list[str] = field(default_factory=list) + + @property + def consistent(self) -> bool: + return not ( + self.missing_units + or self.foreign_units + or self.unreadable_units + or self.ownerless_claims + ) + + +def verify_inventory(queue: WorkQueue) -> InventoryReport: + """Compare the plan, the claim directory and the result directory.""" + report = InventoryReport() + plan_units = set(queue.plan.unit_ids) + + result_files = {path.stem for path in queue.results_dir.glob("*.json")} + report.foreign_units = sorted(result_files - plan_units) + report.missing_units = sorted(plan_units - result_files) + for unit_id in sorted(result_files & plan_units): + if queue.result(unit_id) is None: + report.unreadable_units.append(unit_id) + + for unit_id in sorted(queue.claimed_unit_ids()): + if queue.owner(unit_id) is None: + report.ownerless_claims.append(unit_id) + if unit_id not in result_files: + report.claims_without_results.append(unit_id) + return report + + +def merge_run(queue: WorkQueue, run_id: str) -> MergeResult: + """Score one run, or refuse. + + ``run_id`` is required and must match the queue's plan. Passing another + run's id is an error, not a filter. + """ + plan: UnitPlan = queue.plan + if run_id != plan.run_id: + raise MergeRefusal( + run_id, + [ + f"queue at {queue.root} holds run {plan.run_id!r}, not {run_id!r}; " + "a merge is always scoped to exactly one run" + ], + ) + + reasons: list[str] = [] + inventory = verify_inventory(queue) + if inventory.foreign_units: + reasons.append( + "results present for units outside the plan: " + + ", ".join(inventory.foreign_units[:5]) + ) + if inventory.unreadable_units: + reasons.append( + "unreadable result records: " + ", ".join(inventory.unreadable_units[:5]) + ) + if inventory.ownerless_claims: + reasons.append( + "claims with no readable owner: " + + ", ".join(inventory.ownerless_claims[:5]) + ) + if inventory.missing_units: + reasons.append( + f"{len(inventory.missing_units)} of {len(plan.units)} units have no " + "result: " + ", ".join(inventory.missing_units[:5]) + ) + + results: dict[str, UnitResult] = queue.results() + seen_ids: dict[str, str] = {} + resolved: set[str] = set() + + for unit in plan.units: + result = results.get(unit.unit_id) + if result is None: + continue + if result.plan_digest != plan.digest: + reasons.append( + f"{unit.unit_id}: result belongs to plan {result.plan_digest[:12]}, " + f"not {plan.digest[:12]}" + ) + continue + if result.abandoned: + reasons.append(f"{unit.unit_id}: abandoned after {result.attempt} attempts") + continue + if result.outcome is not UnitOutcome.SUCCEEDED: + reasons.append(f"{unit.unit_id}: outcome {result.outcome.value}") + continue + if result.infra_error_count > 0: + reasons.append( + f"{unit.unit_id}: {result.infra_error_count} instance(s) lost to " + "infrastructure" + ) + continue + + expected = set(unit.instance_ids) + accounted = set(result.accounted_instance_ids) + if len(result.accounted_instance_ids) != len(accounted): + reasons.append(f"{unit.unit_id}: duplicate instance ids in its own result") + continue + # Compare ids, never counts. A shard with one duplicate and one missing + # id has the right count and the wrong content. + if accounted != expected: + missing = sorted(expected - accounted) + extra = sorted(accounted - expected) + detail = [] + if missing: + detail.append(f"missing {', '.join(missing[:5])}") + if extra: + detail.append(f"unplanned {', '.join(extra[:5])}") + reasons.append(f"{unit.unit_id}: " + "; ".join(detail)) + continue + + for instance_id in result.accounted_instance_ids: + previous = seen_ids.get(instance_id) + if previous is not None: + reasons.append( + f"instance {instance_id} accounted for by both {previous} and " + f"{unit.unit_id}" + ) + continue + seen_ids[instance_id] = unit.unit_id + unplanned_resolved = set(result.resolved_instance_ids) - expected + if unplanned_resolved: + reasons.append( + f"{unit.unit_id}: resolved ids outside its shard: " + + ", ".join(sorted(unplanned_resolved)[:5]) + ) + continue + resolved.update(result.resolved_instance_ids) + + planned_ids = set(plan.instance_ids) + if not reasons and set(seen_ids) != planned_ids: + unaccounted = sorted(planned_ids - set(seen_ids)) + reasons.append( + f"{len(unaccounted)} planned instance(s) unaccounted for: " + + ", ".join(unaccounted[:5]) + ) + + if reasons: + raise MergeRefusal(run_id, reasons) + + return MergeResult( + run_id=run_id, + plan_digest=plan.digest, + total_instances=len(planned_ids), + resolved_instances=len(resolved), + unit_count=len(plan.units), + ) diff --git a/tests/unit/evaluation/swe_bench_distributed/test_merge.py b/tests/unit/evaluation/swe_bench_distributed/test_merge.py new file mode 100644 index 000000000..528ab9335 --- /dev/null +++ b/tests/unit/evaluation/swe_bench_distributed/test_merge.py @@ -0,0 +1,189 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""The merge gate: all-or-nothing, id-based, scoped to one run.""" + +from __future__ import annotations + +import inspect + +import pytest + +from inference_endpoint.evaluation.swe_bench_distributed.merge import ( + MergeRefusal, + merge_run, + verify_inventory, +) +from inference_endpoint.evaluation.swe_bench_distributed.queue import ( + UnitOutcome, + UnitResult, + 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(20)] + + +@pytest.fixture +def queue(tmp_path): + return WorkQueue(tmp_path / "wq", plan_units("run-a", IDS, shard_size=10)) + + +def publish(queue: WorkQueue, unit_id: str, **overrides) -> None: + unit = queue.plan.unit(unit_id) + payload = { + "unit_id": unit_id, + "run_id": unit.run_id, + "plan_digest": queue.plan.digest, + "outcome": UnitOutcome.SUCCEEDED, + "accounted_instance_ids": unit.instance_ids, + "resolved_instance_ids": unit.instance_ids[:3], + } + payload.update(overrides) + queue.publish(UnitResult(**payload)) + + +def publish_all(queue: WorkQueue) -> None: + for unit_id in queue.plan.unit_ids: + publish(queue, unit_id) + + +class TestHappyPath: + def test_full_accounting_scores(self, queue): + publish_all(queue) + result = merge_run(queue, "run-a") + assert result.total_instances == 20 + assert result.resolved_instances == 6 + assert result.resolved_rate == pytest.approx(0.3) + assert result.unit_count == 2 + + +class TestRefusals: + def test_a_missing_unit_refuses(self, queue): + publish(queue, "run-a.s00") + # 10 results must never be divided by 20. + with pytest.raises(MergeRefusal, match="have no result"): + merge_run(queue, "run-a") + + def test_an_abandoned_unit_refuses(self, queue): + publish(queue, "run-a.s00") + publish(queue, "run-a.s01", abandoned=True, attempt=3) + with pytest.raises(MergeRefusal, match="abandoned"): + merge_run(queue, "run-a") + + def test_a_non_success_outcome_refuses(self, queue): + publish(queue, "run-a.s00") + publish(queue, "run-a.s01", outcome=UnitOutcome.FAILED) + with pytest.raises(MergeRefusal, match="outcome failed"): + merge_run(queue, "run-a") + + def test_infrastructure_damage_refuses(self, queue): + publish(queue, "run-a.s00") + publish(queue, "run-a.s01", infra_error_count=2) + with pytest.raises(MergeRefusal, match="lost to infrastructure"): + merge_run(queue, "run-a") + + def test_a_missing_id_refuses_even_though_the_count_is_wrong_by_one(self, queue): + publish(queue, "run-a.s00") + unit = queue.plan.unit("run-a.s01") + publish(queue, "run-a.s01", accounted_instance_ids=unit.instance_ids[:-1]) + with pytest.raises(MergeRefusal, match="missing"): + merge_run(queue, "run-a") + + def test_a_swapped_id_refuses_although_the_count_matches(self, queue): + # The whole point of comparing ids rather than counts: this shard has + # exactly ten entries and the wrong content. + publish(queue, "run-a.s00") + unit = queue.plan.unit("run-a.s01") + swapped = unit.instance_ids[:-1] + ("some__other-99",) + publish(queue, "run-a.s01", accounted_instance_ids=swapped) + with pytest.raises(MergeRefusal, match="unplanned"): + merge_run(queue, "run-a") + + def test_a_duplicated_id_within_one_unit_refuses(self, queue): + publish(queue, "run-a.s00") + unit = queue.plan.unit("run-a.s01") + duped = unit.instance_ids[:-1] + (unit.instance_ids[0],) + publish(queue, "run-a.s01", accounted_instance_ids=duped) + with pytest.raises(MergeRefusal, match="duplicate"): + merge_run(queue, "run-a") + + def test_resolved_ids_outside_the_shard_refuse(self, queue): + publish(queue, "run-a.s00") + unit = queue.plan.unit("run-a.s01") + publish( + queue, + "run-a.s01", + resolved_instance_ids=(*unit.instance_ids[:2], IDS[0]), + ) + with pytest.raises(MergeRefusal, match="outside its shard"): + merge_run(queue, "run-a") + + def test_a_foreign_plan_digest_refuses(self, queue): + publish(queue, "run-a.s00") + publish(queue, "run-a.s01") + path = queue.results_dir / "run-a.s01.json" + path.write_text(path.read_text().replace(queue.plan.digest, "f" * 64)) + with pytest.raises(MergeRefusal, match="belongs to plan"): + merge_run(queue, "run-a") + + def test_a_result_outside_the_plan_refuses(self, queue): + publish_all(queue) + (queue.results_dir / "other-run.s00.json").write_text("{}") + with pytest.raises(MergeRefusal, match="outside the plan"): + merge_run(queue, "run-a") + + def test_an_unreadable_result_refuses(self, queue): + publish_all(queue) + (queue.results_dir / "run-a.s00.json").write_text("not json") + with pytest.raises(MergeRefusal, match="unreadable"): + merge_run(queue, "run-a") + + def test_every_reason_is_reported_at_once(self, queue): + publish(queue, "run-a.s00", infra_error_count=1) + with pytest.raises(MergeRefusal) as excinfo: + merge_run(queue, "run-a") + assert len(excinfo.value.reasons) >= 2 + + +class TestScoping: + def test_a_merge_is_always_scoped_to_one_run(self, queue): + publish_all(queue) + with pytest.raises(MergeRefusal, match="scoped to exactly one run"): + merge_run(queue, "some-other-run") + + def test_there_is_no_merge_all(self): + # "Merge everything that looks finished" once combined hundreds of + # banked results from unrelated configurations into one number. + signature = inspect.signature(merge_run) + assert "run_id" in signature.parameters + assert signature.parameters["run_id"].default is inspect.Parameter.empty + assert not hasattr( + __import__( + "inference_endpoint.evaluation.swe_bench_distributed.merge", + fromlist=["merge"], + ), + "merge_all", + ) + + +class TestInventory: + def test_a_complete_run_is_consistent(self, queue): + publish_all(queue) + assert verify_inventory(queue).consistent + + def test_an_ownerless_claim_is_an_inventory_error(self, queue): + publish_all(queue) + claim_dir = queue.claims_dir / "run-a.s00" + claim_dir.mkdir(parents=True) + # Checking `owner` files with one tool and claim directories with + # another is how a verification pass agrees with a broken system. + report = verify_inventory(queue) + assert report.ownerless_claims == ["run-a.s00"] + assert not report.consistent + + def test_claims_without_results_are_reported(self, queue): + queue.claim("run-a.s00") + assert verify_inventory(queue).claims_without_results == ["run-a.s00"]