diff --git a/src/inference_endpoint/evaluation/swe_bench_distributed/__init__.py b/src/inference_endpoint/evaluation/swe_bench_distributed/__init__.py index 3507289bb..73fd2cbab 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 .guards import HealthTerm, HealthVerdict, MemoryGuard, combine_terms, kill_by_pid from .merge import MergeRefusal, MergeResult, merge_run, verify_inventory from .queue import ( ClaimError, @@ -18,18 +19,28 @@ UnitResult, WorkQueue, ) +from .reaper import LocalProcessLiveness, OwnerLiveness, SlurmStepLiveness, reap from .units import Unit, UnitPlan, plan_units __all__ = [ "ClaimError", + "HealthTerm", + "HealthVerdict", + "LocalProcessLiveness", + "MemoryGuard", "MergeRefusal", "MergeResult", + "OwnerLiveness", + "SlurmStepLiveness", "Unit", "UnitOutcome", "UnitPlan", "UnitResult", "WorkQueue", + "combine_terms", + "kill_by_pid", "merge_run", "plan_units", + "reap", "verify_inventory", ] diff --git a/src/inference_endpoint/evaluation/swe_bench_distributed/guards.py b/src/inference_endpoint/evaluation/swe_bench_distributed/guards.py new file mode 100644 index 000000000..92801fe28 --- /dev/null +++ b/src/inference_endpoint/evaluation/swe_bench_distributed/guards.py @@ -0,0 +1,314 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Resource guards for graded SWE-bench evaluation. + +A graded test runs inside an evaluation container with no memory limit. A model +patch that makes a test allocate without bound will take the host down, and when +a whole client fleet shares one scheduler step, one host's OOM destroys every +peer's work along with it -- ``--kill-on-bad-exit=0`` does **not** prevent that, +because the scheduler escalates OOM separately from task exit codes. + +Killing such a process is correct, not a distortion. A patch that makes a graded +test allocate without bound is a failing patch, exactly as a patch that makes it +loop forever is; the alternative to killing was never "the test passes", it was +"the host dies and the instance still never completes". The kill is recorded as +a marker file and the classifier books it as a genuine failure. + +TWO RULES THAT ARE ENFORCED BY CONSTRUCTION HERE: + +1. **Kill by pid, never by pattern.** A pattern such as ``runtests.py`` can + appear in the guard's own command line, and a long-lived daemon can carry a + dead process's argv for days. This module contains no ``pkill``/``pgrep`` + path at all, and :func:`kill_by_pid` refuses self and its own ancestors. +2. **A conjunctive guard must not degenerate.** When one honest term of an + AND-guard permanently loses its data source, the conjunction collapses into + its remaining, weaker clauses and starts firing on healthy targets -- that is + how an idle watchdog killed a live bring-up. :func:`combine_terms` therefore + returns ``INDETERMINATE``, never ``UNHEALTHY``, if any term has no evidence. +""" + +from __future__ import annotations + +import json +import logging +import os +import signal +import time +from dataclasses import dataclass, field +from enum import StrEnum +from pathlib import Path + +logger = logging.getLogger(__name__) + +DEFAULT_KILL_BYTES = 150 * 1024**3 +DEFAULT_WARN_BYTES = 100 * 1024**3 +#: Ancestors that prove a process is inside a container supervisor. +CONTAINER_SUPERVISORS = ("conmon", "containerd-shim", "runc", "enroot", "crun") +_ANCESTOR_DEPTH = 6 + + +class HealthVerdict(StrEnum): + HEALTHY = "healthy" + UNHEALTHY = "unhealthy" + INDETERMINATE = "indeterminate" + + +@dataclass(slots=True) +class HealthTerm: + """One clause of a conjunctive guard, with its evidence count. + + ``evidence`` is the number of observations the term actually made. A term + that made none cannot vote, and must not be silently read as ``HEALTHY`` + (which would let the conjunction fire on the strength of the other clauses + alone) nor as ``UNHEALTHY``. + """ + + name: str + verdict: HealthVerdict + evidence: int + detail: str = "" + + +def combine_terms(terms: list[HealthTerm]) -> tuple[HealthVerdict, str]: + """AND the terms, refusing to act on an unevidenced conjunction.""" + if not terms: + return HealthVerdict.INDETERMINATE, "no terms" + blind = [term.name for term in terms if term.evidence <= 0] + if blind: + return ( + HealthVerdict.INDETERMINATE, + "no evidence for term(s): " + + ", ".join(blind) + + " -- a conjunction with a blind term cannot be trusted to be true", + ) + indeterminate = [ + term.name for term in terms if term.verdict is HealthVerdict.INDETERMINATE + ] + if indeterminate: + return ( + HealthVerdict.INDETERMINATE, + "indeterminate term(s): " + ", ".join(indeterminate), + ) + healthy = [term.name for term in terms if term.verdict is HealthVerdict.HEALTHY] + if healthy: + return HealthVerdict.HEALTHY, "healthy term(s): " + ", ".join(healthy) + return HealthVerdict.UNHEALTHY, "; ".join( + f"{term.name}: {term.detail}" for term in terms + ) + + +class SelfKillRefused(RuntimeError): + """Refused to signal this process or one of its ancestors.""" + + +def ancestors( + pid: int, *, depth: int = _ANCESTOR_DEPTH, proc: Path | None = None +) -> list[int]: + """Parent pids of ``pid``, nearest first.""" + root = proc if proc is not None else Path("/proc") + found: list[int] = [] + current = pid + for _ in range(depth): + try: + stat = (root / str(current) / "status").read_text() + except OSError: + break + parent = None + for line in stat.splitlines(): + if line.startswith("PPid:"): + try: + parent = int(line.split()[1]) + except (IndexError, ValueError): + parent = None + break + if parent is None or parent <= 0 or parent in found: + break + found.append(parent) + current = parent + return found + + +def kill_by_pid( + pid: int, *, sig: int = signal.SIGKILL, proc: Path | None = None +) -> bool: + """Signal exactly one pid. + + Refuses this process and any of its ancestors. There is deliberately no + pattern-matching variant of this function: matching by command line is how a + guard kills itself, or kills whatever inherited a stale argv. + """ + if pid <= 0: + raise SelfKillRefused(f"refusing to signal pid {pid}") + if pid == os.getpid(): + raise SelfKillRefused("refusing to signal self") + if pid in ancestors(os.getpid(), proc=proc): + raise SelfKillRefused(f"refusing to signal ancestor pid {pid}") + try: + os.kill(pid, sig) + except ProcessLookupError: + return False + except OSError: + logger.warning("could not signal pid %d", pid, exc_info=True) + return False + return True + + +@dataclass(slots=True) +class ProcessSample: + pid: int + rss_bytes: int + #: Name of the container this process belongs to, if it could be resolved. + #: This is what determines the phase, so an unresolvable name is not "not a + #: test" -- see :meth:`MemoryGuard.phase_for`. + container_name: str | None = None + ancestor_names: tuple[str, ...] = () + #: Advisory only. Deliberately NOT a predicate: see MemoryGuard's docstring. + cwd: str = "" + + +@dataclass(slots=True) +class GuardAction: + pid: int + rss_bytes: int + verdict: HealthVerdict + reason: str + killed: bool = False + terms: list[HealthTerm] = field(default_factory=list) + + +class MemoryGuard: + """Kill a runaway graded test, and only a runaway graded test. + + A process is a candidate only when **both** terms hold: + + * resident memory at or above ``kill_bytes`` (default 150 GiB; a healthy + graded test uses single-digit GiB, so the headroom is roughly thirty-fold) + * it has a container-supervisor ancestor -- it is inside a container + + THERE IS DELIBERATELY NO WORKING-DIRECTORY TERM. An earlier version required + the process's cwd to be inside the testbed, on the reasoning that a graded + test runs there. It does not always: a runaway that had grown to 667 GiB was + skipped for 105 minutes because its cwd was ``/tmp``. Every additional + conjunct is another way for the guard to miss what it exists to catch, so + the predicate set is the smallest one that cannot match a benchmark client, + an engine, a login shell or the guard itself -- all of which fail the + container term. ``cwd`` is still sampled, as advisory detail only. + """ + + def __init__( + self, + *, + kill_bytes: int = DEFAULT_KILL_BYTES, + warn_bytes: int = DEFAULT_WARN_BYTES, + killed_dir: Path | None = None, + supervisors: tuple[str, ...] = CONTAINER_SUPERVISORS, + eval_container_prefixes: tuple[str, ...] = ("sweb.eval",), + agent_container_prefixes: tuple[str, ...] = ("minisweagent",), + ) -> None: + self.kill_bytes = kill_bytes + self.warn_bytes = warn_bytes + self.killed_dir = killed_dir + self.supervisors = supervisors + self.eval_container_prefixes = eval_container_prefixes + self.agent_container_prefixes = agent_container_prefixes + + def phase_for(self, sample: ProcessSample) -> str: + """Which phase a runaway belongs to, from its container name. + + Fails closed to ``"unknown"``. An unresolvable container name must not + stop the kill -- the process is still a confirmed runaway inside a + container -- but it must also not be booked as an eval kill, because + only an eval kill turns an instance's error into a genuine failure. + """ + name = sample.container_name or "" + if any(name.startswith(prefix) for prefix in self.eval_container_prefixes): + return "eval" + if any(name.startswith(prefix) for prefix in self.agent_container_prefixes): + return "agent" + return "unknown" + + def evaluate(self, sample: ProcessSample) -> GuardAction: + terms = [ + HealthTerm( + name="rss", + verdict=( + HealthVerdict.UNHEALTHY + if sample.rss_bytes >= self.kill_bytes + else HealthVerdict.HEALTHY + ), + evidence=1 if sample.rss_bytes >= 0 else 0, + detail=f"{sample.rss_bytes / 1024**3:.1f} GiB", + ), + HealthTerm( + name="in_container", + verdict=( + HealthVerdict.UNHEALTHY + if any(name in self.supervisors for name in sample.ancestor_names) + else HealthVerdict.HEALTHY + ), + evidence=len(sample.ancestor_names), + detail=f"ancestors={list(sample.ancestor_names)}", + ), + ] + verdict, reason = combine_terms(terms) + return GuardAction( + pid=sample.pid, + rss_bytes=sample.rss_bytes, + verdict=verdict, + reason=reason, + terms=terms, + ) + + def act( + self, + sample: ProcessSample, + *, + instance_id: str | None = None, + phase: str | None = None, + apply: bool = False, + ) -> GuardAction: + """Evaluate and, when ``apply``, kill by pid and record a marker. + + The marker is written *before* the kill: a SIGKILLed test leaves an + ambiguous log, so the record of having killed it is the only reliable + evidence, and it has to exist even if the process dies first. + """ + action = self.evaluate(sample) + if action.verdict is not HealthVerdict.UNHEALTHY or not apply: + return action + resolved_phase = phase if phase is not None else self.phase_for(sample) + if self.killed_dir is not None and instance_id: + self.record_kill(instance_id, sample, phase=resolved_phase) + action.killed = kill_by_pid(sample.pid) + return action + + def record_kill( + self, instance_id: str, sample: ProcessSample, *, phase: str = "eval" + ) -> Path: + """Write the ``....json`` marker. + + 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 and the agent carries on, so it must never influence + classification. + """ + import socket + + assert self.killed_dir is not None + self.killed_dir.mkdir(parents=True, exist_ok=True) + host = socket.gethostname() + path = self.killed_dir / f"{phase}.{instance_id}.{host}.{sample.pid}.json" + path.write_text( + json.dumps( + { + "phase": phase, + "instance_id": instance_id, + "host": host, + "pid": sample.pid, + "rss_bytes": sample.rss_bytes, + "killed_at": time.time(), + } + ) + ) + return path diff --git a/src/inference_endpoint/evaluation/swe_bench_distributed/reaper.py b/src/inference_endpoint/evaluation/swe_bench_distributed/reaper.py new file mode 100644 index 000000000..0905cee42 --- /dev/null +++ b/src/inference_endpoint/evaluation/swe_bench_distributed/reaper.py @@ -0,0 +1,242 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Return orphaned claims to the queue. Nothing else. + +A false reap is the worst thing this system can do. Releasing a claim whose +owner is still running puts the unit back in the queue while it is executing, a +second worker takes it, both write results, and the run has duplicate work, a +wrong denominator, and no error anywhere -- the exact silent corruption the +atomic claim exists to prevent, reintroduced by the janitor. + +Therefore the reaper is conservative in one specific direction: **uncertainty +never escalates.** If liveness cannot be determined, nothing is released. +""" + +from __future__ import annotations + +import logging +import os +import subprocess +from dataclasses import dataclass, field +from enum import StrEnum +from typing import Protocol + +from .queue import OwnerRecord, WorkQueue + +logger = logging.getLogger(__name__) + +_PROBE_TIMEOUT_S = 60 + + +class Liveness(StrEnum): + ALIVE = "alive" + DEAD = "dead" + #: Could not tell. Treated as ALIVE for the purpose of reaping. + INDETERMINATE = "indeterminate" + + +@dataclass(frozen=True, slots=True) +class LivenessVerdict: + state: Liveness + #: Which layer decided. ``"step"`` gets a shorter staleness threshold: a + #: step that died inside a live job took its tasks with it immediately, so + #: there is no reason to wait an hour to believe it. + scope: str = "process" + detail: str = "" + + +class OwnerLiveness(Protocol): + """Decides whether the process that claimed a unit still exists.""" + + def probe(self, owner: OwnerRecord) -> LivenessVerdict: ... + + +class LocalProcessLiveness: + """Liveness by pid, scoped to one host and one boot. + + A pid on its own is not evidence: after a reboot the same number can belong + to something unrelated, so an owner from a different boot of this host is + dead, and an owner from a different host is indeterminate (we cannot see it). + """ + + def __init__(self, *, host: str | None = None, boot: str | None = None) -> None: + import socket + + from .queue import boot_id + + self.host = host if host is not None else socket.gethostname() + self.boot = boot if boot is not None else boot_id() + + def probe(self, owner: OwnerRecord) -> LivenessVerdict: + if owner.host != self.host: + return LivenessVerdict( + Liveness.INDETERMINATE, "process", f"owner is on {owner.host}" + ) + if owner.boot_id and owner.boot_id != self.boot: + return LivenessVerdict( + Liveness.DEAD, "process", "host rebooted since claim" + ) + if owner.pid <= 0: + return LivenessVerdict(Liveness.INDETERMINATE, "process", "no pid recorded") + try: + os.kill(owner.pid, 0) + except ProcessLookupError: + return LivenessVerdict(Liveness.DEAD, "process", "pid gone") + except PermissionError: + # Exists, owned by someone else. + return LivenessVerdict(Liveness.ALIVE, "process", "pid exists") + except OSError: + return LivenessVerdict(Liveness.INDETERMINATE, "process", "kill(0) failed") + return LivenessVerdict(Liveness.ALIVE, "process", "pid exists") + + +class SlurmStepLiveness: + """Liveness by SLURM job *and step*. + + An owner is dead when its job is absent from ``squeue``, **or** when the job + is alive but its step is gone. The second clause is not optional: a step can + die inside a live job (a killed srun, an OOM-terminated step) and SLURM + kills that step's tasks, but the job never leaves ``squeue``, so a + job-level-only rule blocks those units for the entire life of the + allocation. + + Step liveness comes from ``scontrol show step``, never ``squeue -s``: on the + clusters this was built for ``squeue -s`` reports only ``.extern`` and never + the worker step, so using it would mark every live step dead and falsely + reap every claim. + + Every failure to read SLURM yields ``INDETERMINATE``. An unavailable + ``squeue`` must never be read as "no jobs are running". + """ + + def __init__(self, *, timeout_s: int = _PROBE_TIMEOUT_S) -> None: + self.timeout_s = timeout_s + + def _run(self, argv: list[str]) -> str | None: + try: + completed = subprocess.run( + argv, capture_output=True, text=True, timeout=self.timeout_s + ) + except (OSError, subprocess.SubprocessError): + logger.warning("reaper: %s unavailable; releasing nothing", argv[0]) + return None + if completed.returncode != 0: + return None + return completed.stdout + + def live_job_ids(self) -> set[str] | None: + out = self._run(["squeue", "-h", "-o", "%i"]) + if out is None: + return None + ids: set[str] = set() + for token in out.split(): + token = token.strip() + if not token: + continue + ids.add(token) + ids.add(token.split("_")[0].split(".")[0]) + if not ids and os.environ.get("SLURM_JOB_ID"): + # An empty queue is legitimate in general, but not while we are + # ourselves inside a job. That is what a broken squeue looks like. + logger.warning( + "reaper: squeue returned empty while inside a job; releasing nothing" + ) + return None + return ids + + def live_step_ids(self, job_id: str) -> set[str] | None: + out = self._run(["scontrol", "show", "step", str(job_id)]) + if out is None: + return None + steps = { + token.split("=", 1)[1].split(".", 1)[1] + for token in out.split() + if token.startswith("StepId=") and "." in token.split("=", 1)[1] + } + # A successful scontrol listing no step at all is implausible while the + # job exists (there is always .extern): indeterminate, not empty. + return steps or None + + def probe(self, owner: OwnerRecord) -> LivenessVerdict: + if not owner.slurm_job_id: + return LivenessVerdict(Liveness.INDETERMINATE, "job", "no job id recorded") + jobs = self.live_job_ids() + if jobs is None: + return LivenessVerdict(Liveness.INDETERMINATE, "job", "squeue unreadable") + if owner.slurm_job_id not in jobs: + return LivenessVerdict(Liveness.DEAD, "job", "job absent from squeue") + if not owner.slurm_step_id: + return LivenessVerdict( + Liveness.ALIVE, "job", "job present, no step recorded" + ) + steps = self.live_step_ids(owner.slurm_job_id) + if steps is None: + # Indeterminate step liveness must not become MORE aggressive than + # the job-level answer, which is "alive". + return LivenessVerdict(Liveness.ALIVE, "job", "step list unreadable") + if owner.slurm_step_id in steps: + return LivenessVerdict(Liveness.ALIVE, "step", "step present") + return LivenessVerdict(Liveness.DEAD, "step", "step gone inside a live job") + + +@dataclass(slots=True) +class ReapReport: + released: list[str] = field(default_factory=list) + kept: dict[str, str] = field(default_factory=dict) + dry_run: bool = True + + def __bool__(self) -> bool: # pragma: no cover - convenience + return bool(self.released) + + +def reap( + queue: WorkQueue, + liveness: OwnerLiveness, + *, + stale_after_s: float = 3600.0, + step_stale_after_s: float = 900.0, + apply: bool = False, + now: float | None = None, +) -> ReapReport: + """Release claims whose owner is provably gone and which produced no result. + + All three conditions must hold: no result, a stale-enough heartbeat, and a + ``DEAD`` liveness verdict. A verdict scoped to ``"step"`` uses the shorter + ``step_stale_after_s``: when a step dies inside a job that stays in the + queue, the job-level rule alone never fires and those units stay blocked for + the entire life of the allocation. + """ + report = ReapReport(dry_run=not apply) + completed = queue.completed_unit_ids() + for unit_id in sorted(queue.claimed_unit_ids()): + if unit_id in completed: + # Claims for completed units are harmless bookkeeping. + report.kept[unit_id] = "has result" + continue + age = queue.heartbeat_age(unit_id, now=now) + if age is None: + report.kept[unit_id] = "no heartbeat to age" + continue + owner = queue.owner(unit_id) + if owner is None: + # We cannot prove anything about an unreadable owner, so age alone + # decides. A claim holding anything but pure bookkeeping is not ours + # to reason about at all. + if not queue.is_pure_bookkeeping(unit_id): + report.kept[unit_id] = "claim holds non-bookkeeping contents" + continue + verdict = LivenessVerdict(Liveness.DEAD, "process", "owner unreadable") + else: + verdict = liveness.probe(owner) + threshold = step_stale_after_s if verdict.scope == "step" else stale_after_s + if age < threshold: + report.kept[unit_id] = f"heartbeat {age:.0f}s < {threshold:.0f}s" + continue + if verdict.state is not Liveness.DEAD: + report.kept[unit_id] = f"owner {verdict.state.value}: {verdict.detail}" + continue + report.released.append(unit_id) + if apply: + queue.release(unit_id) + return report diff --git a/tests/unit/evaluation/swe_bench_distributed/test_guards_and_reaper.py b/tests/unit/evaluation/swe_bench_distributed/test_guards_and_reaper.py new file mode 100644 index 000000000..aebdec290 --- /dev/null +++ b/tests/unit/evaluation/swe_bench_distributed/test_guards_and_reaper.py @@ -0,0 +1,295 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Resource guards and the claim reaper.""" + +from __future__ import annotations + +import os +import time + +import pytest + +from inference_endpoint.evaluation.swe_bench_distributed import guards as guards_mod +from inference_endpoint.evaluation.swe_bench_distributed.guards import ( + DEFAULT_KILL_BYTES, + HealthTerm, + HealthVerdict, + MemoryGuard, + ProcessSample, + SelfKillRefused, + combine_terms, + kill_by_pid, +) +from inference_endpoint.evaluation.swe_bench_distributed.queue import ( + UnitOutcome, + UnitResult, + WorkQueue, +) +from inference_endpoint.evaluation.swe_bench_distributed.reaper import ( + Liveness, + LivenessVerdict, + reap, +) +from inference_endpoint.evaluation.swe_bench_distributed.units import plan_units + +pytestmark = pytest.mark.unit + +GIB = 1024**3 + + +def executable_source(module) -> str: + """Module source with comments and string literals removed.""" + import tokenize + + kept = [] + with open(module.__file__, "rb") as handle: + for token in tokenize.tokenize(handle.readline): + if token.type in {tokenize.COMMENT, tokenize.STRING}: + continue + kept.append(token.string) + return " ".join(kept) + + +class FakeLiveness: + def __init__(self, verdict: LivenessVerdict) -> None: + self.verdict = verdict + + def probe(self, owner): + return self.verdict + + +@pytest.fixture +def queue(tmp_path): + plan = plan_units("run-a", [f"i-{i}" for i in range(20)], shard_size=10) + return WorkQueue(tmp_path / "wq", plan) + + +class TestConjunction: + def test_all_unhealthy_terms_fire(self): + terms = [ + HealthTerm("a", HealthVerdict.UNHEALTHY, evidence=1), + HealthTerm("b", HealthVerdict.UNHEALTHY, evidence=1), + ] + assert combine_terms(terms)[0] is HealthVerdict.UNHEALTHY + + def test_a_blind_term_makes_the_conjunction_indeterminate(self): + # When an honest term permanently loses its data source, an AND-guard + # collapses into its remaining, weaker clauses and starts firing on + # healthy targets. That is how an idle watchdog killed a live bring-up. + terms = [ + HealthTerm("loud", HealthVerdict.UNHEALTHY, evidence=1), + HealthTerm("blind", HealthVerdict.UNHEALTHY, evidence=0), + ] + verdict, reason = combine_terms(terms) + assert verdict is HealthVerdict.INDETERMINATE + assert "blind" in reason + + def test_a_blind_term_never_yields_unhealthy(self): + terms = [HealthTerm("blind", HealthVerdict.UNHEALTHY, evidence=0)] + assert combine_terms(terms)[0] is not HealthVerdict.UNHEALTHY + + def test_one_healthy_term_spares_the_target(self): + terms = [ + HealthTerm("a", HealthVerdict.UNHEALTHY, evidence=1), + HealthTerm("b", HealthVerdict.HEALTHY, evidence=1), + ] + assert combine_terms(terms)[0] is HealthVerdict.HEALTHY + + def test_no_terms_is_indeterminate(self): + assert combine_terms([])[0] is HealthVerdict.INDETERMINATE + + +class TestKillDiscipline: + def test_there_is_no_pattern_kill_path_in_the_module(self): + # A pattern such as "runtests.py" can appear in the guard's own command + # line, and a long-lived daemon can carry a dead process's argv for days. + # Executable code is inspected with comments and strings removed, so the + # docstring explaining the rule cannot satisfy the test for it. + code = executable_source(guards_mod) + assert "pkill" not in code + assert "pgrep" not in code + + def test_the_guard_never_shells_out(self): + # There is no command line to match against in the first place. + code = executable_source(guards_mod) + assert "subprocess" not in code + assert "os.system" not in code + + def test_killing_self_is_refused(self): + with pytest.raises(SelfKillRefused, match="self"): + kill_by_pid(os.getpid()) + + def test_killing_an_ancestor_is_refused(self): + with pytest.raises(SelfKillRefused, match="ancestor"): + kill_by_pid(os.getppid()) + + def test_a_nonsense_pid_is_refused(self): + with pytest.raises(SelfKillRefused): + kill_by_pid(0) + + +class TestMemoryGuard: + def runaway(self, **overrides): + payload = { + "pid": 4242, + "rss_bytes": 200 * GIB, + "container_name": "sweb.eval.arm64.repo__proj-1", + "ancestor_names": ("bash", "conmon"), + } + payload.update(overrides) + return ProcessSample(**payload) + + def test_a_runaway_graded_test_is_unhealthy(self): + action = MemoryGuard().evaluate(self.runaway()) + assert action.verdict is HealthVerdict.UNHEALTHY + + def test_a_runaway_outside_the_testbed_is_still_caught(self): + # An earlier version required cwd inside /testbed. A runaway that had + # grown to 667 GiB was skipped for 105 minutes because its cwd was /tmp, + # so cwd is advisory detail and never a predicate. + action = MemoryGuard().evaluate(self.runaway(cwd="/tmp")) + assert action.verdict is HealthVerdict.UNHEALTHY + assert {term.name for term in action.terms} == {"rss", "in_container"} + + def test_a_large_process_outside_a_container_is_spared(self): + action = MemoryGuard().evaluate(self.runaway(ancestor_names=("bash", "sshd"))) + assert action.verdict is HealthVerdict.HEALTHY + + def test_a_normal_test_is_spared(self): + action = MemoryGuard().evaluate(self.runaway(rss_bytes=3 * GIB)) + assert action.verdict is HealthVerdict.HEALTHY + + def test_unreadable_ancestry_is_indeterminate_not_a_kill(self): + action = MemoryGuard().evaluate(self.runaway(ancestor_names=())) + assert action.verdict is HealthVerdict.INDETERMINATE + + def test_the_default_threshold_leaves_wide_headroom(self): + assert DEFAULT_KILL_BYTES >= 100 * GIB + + @pytest.mark.parametrize( + ("container_name", "phase"), + [ + ("sweb.eval.arm64.repo__proj-1", "eval"), + ("minisweagent-abc123", "agent"), + ("something-else", "unknown"), + (None, "unknown"), + ], + ) + def test_phase_comes_from_the_container_name_and_fails_closed( + self, container_name, phase + ): + # Only an eval kill turns an instance's error into a genuine failure, so + # an unresolvable name must not be booked as one. + guard = MemoryGuard() + assert guard.phase_for(self.runaway(container_name=container_name)) == phase + + def test_the_marker_is_written_before_the_kill(self, tmp_path, monkeypatch): + killed_dir = tmp_path / "killed" + order: list[str] = [] + monkeypatch.setattr( + guards_mod, + "kill_by_pid", + lambda pid, **kwargs: order.append("kill") or True, + ) + guard = MemoryGuard(killed_dir=killed_dir) + original = guard.record_kill + + def traced(*args, **kwargs): + order.append("marker") + return original(*args, **kwargs) + + monkeypatch.setattr(guard, "record_kill", traced) + guard.act(self.runaway(), instance_id="repo__proj-1", apply=True) + + # A SIGKILLed test leaves an ambiguous log, so the record of having + # killed it must survive even if the process dies first. + assert order == ["marker", "kill"] + assert list(killed_dir.glob("eval.repo__proj-1.*.json")) + + def test_dry_evaluation_does_not_kill(self, tmp_path, monkeypatch): + monkeypatch.setattr( + guards_mod, "kill_by_pid", lambda *a, **k: pytest.fail("killed") + ) + action = MemoryGuard(killed_dir=tmp_path).act( + self.runaway(), instance_id="x", apply=False + ) + assert not action.killed + + +class TestReaper: + def stale_claim(self, queue, unit_id="run-a.s00", age=7200.0): + queue.claim(unit_id) + heartbeat = queue.claims_dir / unit_id / "hb" + past = time.time() - age + os.utime(heartbeat, (past, past)) + + def test_a_dead_owner_with_no_result_is_released(self, queue): + self.stale_claim(queue) + report = reap(queue, FakeLiveness(LivenessVerdict(Liveness.DEAD)), apply=True) + assert report.released == ["run-a.s00"] + assert "run-a.s00" in queue.available_unit_ids() + + def test_a_live_owner_is_never_released(self, queue): + self.stale_claim(queue) + report = reap(queue, FakeLiveness(LivenessVerdict(Liveness.ALIVE)), apply=True) + # A false reap gives one unit two owners, duplicate results and a wrong + # denominator, with no error anywhere. + assert report.released == [] + + def test_an_indeterminate_probe_releases_nothing(self, queue): + self.stale_claim(queue) + report = reap( + queue, + FakeLiveness(LivenessVerdict(Liveness.INDETERMINATE)), + apply=True, + ) + assert report.released == [] + assert "indeterminate" in report.kept["run-a.s00"] + + def test_a_fresh_heartbeat_is_never_released(self, queue): + queue.claim("run-a.s00") + report = reap(queue, FakeLiveness(LivenessVerdict(Liveness.DEAD)), apply=True) + assert report.released == [] + + def test_a_claim_with_a_result_is_never_released(self, queue): + self.stale_claim(queue) + unit = queue.plan.unit("run-a.s00") + queue.results_dir.joinpath("run-a.s00.json").write_text( + UnitResult( + unit_id="run-a.s00", + run_id="run-a", + plan_digest=queue.plan.digest, + outcome=UnitOutcome.SUCCEEDED, + accounted_instance_ids=unit.instance_ids, + ).to_dict() + and '{"unit_id": "run-a.s00"}' + ) + report = reap(queue, FakeLiveness(LivenessVerdict(Liveness.DEAD)), apply=True) + assert report.released == [] + + def test_a_dead_step_uses_the_shorter_threshold(self, queue): + # A step that dies inside a live job takes its tasks with it at once, so + # waiting an hour would block those units for the whole allocation. + self.stale_claim(queue, age=1200.0) + report = reap( + queue, + FakeLiveness(LivenessVerdict(Liveness.DEAD, scope="step")), + stale_after_s=3600.0, + step_stale_after_s=900.0, + apply=True, + ) + assert report.released == ["run-a.s00"] + + def test_dry_run_reports_without_releasing(self, queue): + self.stale_claim(queue) + report = reap(queue, FakeLiveness(LivenessVerdict(Liveness.DEAD))) + assert report.released == ["run-a.s00"] + assert queue.claimed_unit_ids() == {"run-a.s00"} + + def test_a_claim_with_unexpected_contents_is_left_alone(self, queue): + self.stale_claim(queue) + (queue.claims_dir / "run-a.s00" / "surprise").write_text("x") + (queue.claims_dir / "run-a.s00" / "owner").unlink() + report = reap(queue, FakeLiveness(LivenessVerdict(Liveness.DEAD)), apply=True) + assert report.released == [] diff --git a/tests/unit/evaluation/swe_bench_distributed/test_liveness.py b/tests/unit/evaluation/swe_bench_distributed/test_liveness.py new file mode 100644 index 000000000..e65b32b9b --- /dev/null +++ b/tests/unit/evaluation/swe_bench_distributed/test_liveness.py @@ -0,0 +1,163 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Owner-liveness probes. Uncertainty must never escalate to DEAD.""" + +from __future__ import annotations + +import os +import socket +import subprocess + +import pytest + +from inference_endpoint.evaluation.swe_bench_distributed.queue import OwnerRecord +from inference_endpoint.evaluation.swe_bench_distributed.reaper import ( + Liveness, + LocalProcessLiveness, + SlurmStepLiveness, +) + +pytestmark = pytest.mark.unit + + +def owner(**overrides) -> OwnerRecord: + payload = { + "unit_id": "run-a.s00", + "host": socket.gethostname(), + "pid": os.getpid(), + "boot_id": "boot-1", + "plan_digest": "d" * 64, + "claimed_at": 0.0, + } + payload.update(overrides) + return OwnerRecord(**payload) + + +class TestLocalProcessLiveness: + def test_a_live_pid_on_this_boot_is_alive(self): + probe = LocalProcessLiveness(boot="boot-1") + assert probe.probe(owner()).state is Liveness.ALIVE + + def test_a_missing_pid_is_dead(self): + probe = LocalProcessLiveness(boot="boot-1") + # 2**22 is above the default pid_max on Linux, so it cannot exist. + assert probe.probe(owner(pid=2**22)).state is Liveness.DEAD + + def test_a_different_boot_is_dead(self): + # After a reboot the same pid number can belong to something unrelated, + # so a live-looking pid is not evidence that the owner survived. + probe = LocalProcessLiveness(boot="boot-2") + assert probe.probe(owner()).state is Liveness.DEAD + + def test_another_host_is_indeterminate_not_dead(self): + probe = LocalProcessLiveness(boot="boot-1") + assert probe.probe(owner(host="elsewhere")).state is Liveness.INDETERMINATE + + def test_a_missing_pid_record_is_indeterminate(self): + probe = LocalProcessLiveness(boot="boot-1") + assert probe.probe(owner(pid=0)).state is Liveness.INDETERMINATE + + +class FakeSlurm(SlurmStepLiveness): + def __init__(self, responses): + super().__init__() + self.responses = responses + self.calls: list[list[str]] = [] + + def _run(self, argv): + self.calls.append(argv) + response = self.responses.get(argv[0]) + if isinstance(response, Exception): + raise response + return response + + +class TestSlurmStepLiveness: + def slurm_owner(self, **overrides): + return owner(slurm_job_id="1000", slurm_step_id="3", **overrides) + + def test_a_job_absent_from_squeue_is_dead(self): + probe = FakeSlurm({"squeue": "2000\n"}) + verdict = probe.probe(self.slurm_owner()) + assert verdict.state is Liveness.DEAD + assert verdict.scope == "job" + + def test_a_live_job_and_step_is_alive(self): + probe = FakeSlurm( + { + "squeue": "1000\n", + "scontrol": "StepId=1000.3 State=RUNNING StepId=1000.extern", + } + ) + assert probe.probe(self.slurm_owner()).state is Liveness.ALIVE + + def test_a_dead_step_inside_a_live_job_is_dead(self): + # A step can die inside a live job -- a killed srun, an OOM-terminated + # step -- and the job never leaves the queue, so a job-level-only rule + # blocks those units for the entire allocation. + probe = FakeSlurm( + {"squeue": "1000\n", "scontrol": "StepId=1000.extern State=RUNNING"} + ) + verdict = probe.probe(self.slurm_owner()) + assert verdict.state is Liveness.DEAD + assert verdict.scope == "step" + + def test_step_liveness_uses_scontrol_not_squeue_s(self): + probe = FakeSlurm( + {"squeue": "1000\n", "scontrol": "StepId=1000.3 State=RUNNING"} + ) + probe.probe(self.slurm_owner()) + # `squeue -s` reports only `.extern` on the clusters this targets, so it + # would mark every live step dead and falsely reap every claim. + assert ["scontrol", "show", "step", "1000"] in probe.calls + assert not any("-s" in argv for argv in probe.calls if argv[0] == "squeue") + + def test_an_unreadable_squeue_is_indeterminate(self): + probe = FakeSlurm({"squeue": None}) + assert probe.probe(self.slurm_owner()).state is Liveness.INDETERMINATE + + def test_an_unreadable_step_list_falls_back_to_the_job_answer(self): + # Indeterminate step liveness must never be more aggressive than the + # job-level answer, which is "alive". + probe = FakeSlurm({"squeue": "1000\n", "scontrol": None}) + assert probe.probe(self.slurm_owner()).state is Liveness.ALIVE + + def test_an_empty_scontrol_listing_is_treated_as_unreadable(self): + probe = FakeSlurm({"squeue": "1000\n", "scontrol": "no steps here"}) + assert probe.probe(self.slurm_owner()).state is Liveness.ALIVE + + def test_an_owner_without_a_job_id_is_indeterminate(self): + probe = FakeSlurm({"squeue": "1000\n"}) + assert probe.probe(owner()).state is Liveness.INDETERMINATE + + def test_an_empty_queue_inside_a_job_is_implausible(self, monkeypatch): + # An empty successful squeue is what a broken squeue looks like. It must + # never be read as "no jobs are running" while we are inside a job. + monkeypatch.setenv("SLURM_JOB_ID", "1000") + probe = FakeSlurm({"squeue": ""}) + assert probe.live_job_ids() is None + + def test_an_empty_queue_outside_a_job_is_trusted(self, monkeypatch): + monkeypatch.delenv("SLURM_JOB_ID", raising=False) + probe = FakeSlurm({"squeue": ""}) + assert probe.live_job_ids() == set() + + def test_array_job_ids_are_matched_by_base_id(self): + probe = FakeSlurm( + {"squeue": "1000_4\n", "scontrol": "StepId=1000.3 State=RUNNING"} + ) + assert probe.probe(self.slurm_owner()).state is Liveness.ALIVE + + def test_a_failing_command_is_indeterminate_not_dead(self): + probe = SlurmStepLiveness(timeout_s=1) + + def boom(argv, **kwargs): + raise subprocess.SubprocessError("no slurm here") + + original = subprocess.run + subprocess.run = boom + try: + assert probe.live_job_ids() is None + finally: + subprocess.run = original