diff --git a/src/inference_endpoint/evaluation/swe_bench_distributed/__init__.py b/src/inference_endpoint/evaluation/swe_bench_distributed/__init__.py index c97b98f55..7269ab520 100644 --- a/src/inference_endpoint/evaluation/swe_bench_distributed/__init__.py +++ b/src/inference_endpoint/evaluation/swe_bench_distributed/__init__.py @@ -11,6 +11,15 @@ exactly once. """ +from .guards import HealthTerm, HealthVerdict, MemoryGuard, combine_terms, kill_by_pid +from .infra_retry import ( + InfraRetryLedger, + RetryOutcome, + RetryRecord, + RunQuality, + is_provable_non_execution, + retry_on_provable_non_execution, +) from .merge import ( CompletenessReport, MergeRefusal, @@ -25,20 +34,36 @@ UnitResult, WorkQueue, ) +from .reaper import LocalProcessLiveness, OwnerLiveness, SlurmStepLiveness, reap from .units import Unit, UnitPlan, plan_units __all__ = [ "ClaimError", "CompletenessReport", + "HealthTerm", + "HealthVerdict", + "InfraRetryLedger", + "LocalProcessLiveness", + "MemoryGuard", "MergeRefusal", "MergeResult", + "OwnerLiveness", + "RetryOutcome", + "RetryRecord", + "RunQuality", + "SlurmStepLiveness", "Unit", "UnitOutcome", "UnitPlan", "UnitResult", "WorkQueue", "assess_run", + "combine_terms", + "is_provable_non_execution", + "kill_by_pid", "merge_run", "plan_units", + "reap", + "retry_on_provable_non_execution", "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/infra_retry.py b/src/inference_endpoint/evaluation/swe_bench_distributed/infra_retry.py new file mode 100644 index 000000000..c9a8b0cb8 --- /dev/null +++ b/src/inference_endpoint/evaluation/swe_bench_distributed/infra_retry.py @@ -0,0 +1,304 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Retry infrastructure faults, but only where non-execution is *provable*. + +A retry is a correctness decision, not a convenience. Re-running a command that +may already have run can apply an edit twice, delete something twice, or double +a test run, and none of those announce themselves. So the gate here is not "an +error happened" -- it is "the work provably did not happen". + +The evidence comes from the failure itself. An exception may expose +``provable_non_execution``: for a Pyxis step that is the status file still +reading ``pending`` **and** no in-band sentinel, meaning the step script did not +run even its first line. Anything that does not make that claim is not retried, +which is the safe default for every exception type this module has never heard +of. + +Retries are bounded and, more importantly, **counted**. The banked campaign this +is ported from retried environment faults without limit and without counting +them (``wq_worker.sh:41`` ``WQ_MAX_ATTEMPTS=5``, with ``:256`` "ENVIRONMENT +FAULTS DO NOT CONSUME THE UNIT'S ATTEMPT BUDGET"), which is precisely why nobody +knew how many there had been. A retry loop that quietly absorbs the defect it +compensates for turns a broken cluster into an invisible one: the measured +effect of adding this loop was ``RunnerError`` 59 -> 7 and resolve 47.0% -> +70.0% against a banked 70.67% on the identical 200 instances, and a run that +needs that much rescuing is not a clean run even when it finishes. +""" + +from __future__ import annotations + +import json +import logging +import threading +import time +from collections import Counter +from collections.abc import Callable +from dataclasses import dataclass +from enum import StrEnum +from pathlib import Path +from typing import Any, TypeVar + +logger = logging.getLogger(__name__) + +DEFAULT_MAX_ATTEMPTS = 3 +#: Above this share of operations needing a retry, a run is DEGRADED even if +#: every unit eventually succeeded. Rescuing one operation in fifty is not a +#: healthy fleet, it is a fleet that happened to be caught. +DEGRADED_RETRY_FRACTION = 0.02 + +_T = TypeVar("_T") + + +class RetryOutcome(StrEnum): + #: Provably never executed; another attempt follows. + RETRYING = "retrying" + #: A later attempt succeeded. + RECOVERED = "recovered" + #: Provably never executed, but the attempt budget ran out. + EXHAUSTED = "exhausted" + #: The work may have executed. Retrying could double-apply it, so this is a + #: hard failure by construction. + NOT_RETRYABLE = "not_retryable" + + +class RunQuality(StrEnum): + CLEAN = "CLEAN" + OK_WITH_RETRIES = "OK_WITH_RETRIES" + DEGRADED = "DEGRADED" + + +def is_provable_non_execution(error: BaseException) -> bool: + """Whether ``error`` proves its operation never ran. + + Read as an attribute rather than an isinstance check so the producer of the + evidence (the Pyxis step runner) and this consumer stay decoupled. An + exception that does not claim the property is never retried. + """ + return getattr(error, "provable_non_execution", False) is True + + +@dataclass(frozen=True, slots=True) +class RetryRecord: + target: str + attempt: int + outcome: RetryOutcome + detail: str | None = None + at: float = 0.0 + + def to_dict(self) -> dict[str, Any]: + return { + "target": self.target, + "attempt": self.attempt, + "outcome": self.outcome.value, + "detail": self.detail, + "at": self.at, + } + + +class InfraRetryLedger: + """Every provable non-execution and what became of it. + + Appends one JSON line per event when given a path, so a run that dies still + leaves its retry history behind, and holds the same records in memory for + :meth:`summary`. Accounting must never be able to take a run down, so a + write failure is logged and swallowed -- but the in-memory counters are + updated first, so the summary is correct even then. + """ + + def __init__(self, path: Path | None = None) -> None: + self.path = Path(path) if path is not None else None + self._lock = threading.Lock() + self._records: list[RetryRecord] = [] + self._operations = 0 + + @property + def records(self) -> list[RetryRecord]: + with self._lock: + return list(self._records) + + @property + def operations(self) -> int: + """Operations submitted to the retry wrapper: the denominator.""" + with self._lock: + return self._operations + + def note_operation(self) -> None: + with self._lock: + self._operations += 1 + + def record( + self, + *, + target: str, + attempt: int, + outcome: RetryOutcome, + detail: str | None = None, + ) -> None: + entry = RetryRecord( + target=target, + attempt=attempt, + outcome=outcome, + detail=detail, + at=time.time(), + ) + with self._lock: + self._records.append(entry) + path = self.path + if path is not None: + try: + with path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(entry.to_dict()) + "\n") + except OSError: + logger.warning( + "could not append to the infra retry ledger", exc_info=True + ) + + @classmethod + def from_jsonl(cls, path: Path) -> InfraRetryLedger: + """Load a ledger written by another process. + + The SWE-bench service is an isolated subproject that must not import the + benchmark client, so its Pyxis step runner writes this same record shape + directly. Sharing a file format rather than a module is the only way the + two halves can agree, and reading it here is what turns per-step retries + into a run-level `run_quality`. + + Unparseable lines are skipped: a truncated final line from a run that + died is expected, and losing the whole history to it would be worse. + """ + ledger = cls(path) + try: + text = Path(path).read_text(encoding="utf-8") + except OSError: + return ledger + for line in text.splitlines(): + line = line.strip() + if not line: + continue + try: + raw = json.loads(line) + record = RetryRecord( + target=str(raw["target"]), + attempt=int(raw["attempt"]), + outcome=RetryOutcome(raw["outcome"]), + detail=raw.get("detail"), + at=float(raw.get("at") or 0.0), + ) + except (ValueError, KeyError, TypeError): + logger.warning("skipping unreadable infra retry record") + continue + ledger._records.append(record) + # Every first attempt that needed a retry represents one operation; a + # writer that only logs failures cannot report the clean denominator, so + # it is reported as unknown rather than guessed. + ledger._operations = sum(1 for r in ledger._records if r.attempt == 1) + return ledger + + def summary(self) -> dict[str, Any]: + """Counters a report can publish without re-deriving anything.""" + records = self.records + operations = self.operations + recovered = {r.target for r in records if r.outcome is RetryOutcome.RECOVERED} + exhausted = {r.target for r in records if r.outcome is RetryOutcome.EXHAUSTED} + return { + "infra_retries_total": len(records), + "infra_retry_operations": operations, + "infra_retry_outcomes": dict( + Counter(r.outcome.value for r in records).most_common() + ), + "infra_retry_succeeded_on_attempt": dict( + Counter( + str(r.attempt) + for r in records + if r.outcome is RetryOutcome.RECOVERED + ).most_common() + ), + # A target that recovered and later exhausted was not saved. + "instances_saved_by_retry": len(recovered - exhausted), + "infra_retries_exhausted": sum( + 1 for r in records if r.outcome is RetryOutcome.EXHAUSTED + ), + "run_quality": self.run_quality().value, + } + + def run_quality(self) -> RunQuality: + records = self.records + if any( + r.outcome in (RetryOutcome.EXHAUSTED, RetryOutcome.NOT_RETRYABLE) + for r in records + ): + return RunQuality.DEGRADED + if not records: + return RunQuality.CLEAN + operations = max(1, self.operations) + if len(records) > DEGRADED_RETRY_FRACTION * operations: + return RunQuality.DEGRADED + return RunQuality.OK_WITH_RETRIES + + +def retry_on_provable_non_execution( + operation: Callable[[], _T], + *, + target: str, + ledger: InfraRetryLedger | None = None, + max_attempts: int = DEFAULT_MAX_ATTEMPTS, + backoff_s: float = 2.0, + max_backoff_s: float = 30.0, + sleep: Callable[[float], None] = time.sleep, +) -> _T: + """Call ``operation``, retrying only failures that prove it never ran. + + Raises the last failure when the budget is exhausted, and re-raises + immediately -- without consuming the budget -- for anything that does not + prove non-execution. + """ + if max_attempts < 1: + raise ValueError("max_attempts must be at least 1") + if ledger is not None: + ledger.note_operation() + for attempt in range(1, max_attempts + 1): + try: + result = operation() + except Exception as exc: + if not is_provable_non_execution(exc): + # The work may have run. Retrying could double-apply it. + if ledger is not None: + ledger.record( + target=target, + attempt=attempt, + outcome=RetryOutcome.NOT_RETRYABLE, + detail=f"{type(exc).__name__}: {exc}", + ) + raise + if attempt == max_attempts: + if ledger is not None: + ledger.record( + target=target, + attempt=attempt, + outcome=RetryOutcome.EXHAUSTED, + detail=f"{type(exc).__name__}: {exc}", + ) + raise + if ledger is not None: + ledger.record( + target=target, + attempt=attempt, + outcome=RetryOutcome.RETRYING, + detail=f"{type(exc).__name__}: {exc}", + ) + logger.warning( + "%s provably never executed (attempt %d/%d): %s -- retrying", + target, + attempt, + max_attempts, + exc, + ) + sleep(min(max_backoff_s, backoff_s * attempt)) + continue + if attempt > 1 and ledger is not None: + ledger.record( + target=target, attempt=attempt, outcome=RetryOutcome.RECOVERED + ) + return result + raise AssertionError("unreachable") # pragma: no cover 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..3224193db --- /dev/null +++ b/src/inference_endpoint/evaluation/swe_bench_distributed/reaper.py @@ -0,0 +1,243 @@ +# 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: + pass + + +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/src/inference_endpoint/evaluation/swebench_service/README.md b/src/inference_endpoint/evaluation/swebench_service/README.md index 113f861e2..0e7a501c5 100644 --- a/src/inference_endpoint/evaluation/swebench_service/README.md +++ b/src/inference_endpoint/evaluation/swebench_service/README.md @@ -75,6 +75,22 @@ overlapping `srun` step in that container, preserving filesystem changes across turns. Tool commands run in private PID namespaces so one trajectory cannot signal processes belonging to another trajectory. +Each step reports its outcome through two channels. The primary one is in band: the +step script prints `__MLPERF_STEP_RC__ ` on `srun`'s stdout, which needs +no readable shared filesystem and is stripped from the command output before it is +returned. The fallback is the status file written into the container's `/tmp` mount. + +When a step reports through neither, `StepNotLaunched` (a `RunnerError`) is raised. +Besides `srun`'s own output it carries `srun_rc`, the observed `status` bytes, and +`provable_non_execution` -- true only when the status file is still `pending` and no +sentinel arrived, meaning the step script did not run even its first line and the +command definitely did not execute. Anything else leaves open that it did. Callers +deciding whether re-running is safe must use that flag rather than the message text. + +Cluster note: on a busy controller these failures cluster around slurmctld RPC rate +limiting (`Job credential expired`). Pacing step creation below the controller's +`rl_refill_rate` is a deployment concern rather than a property of this package. + After generation, the Pyxis worker evaluates each prediction in a fresh `srun` container step because the Docker-based SWE-bench evaluator cannot run on the compute node. It mounts the patch, SWE-bench evaluation script, and output file into diff --git a/src/inference_endpoint/evaluation/swebench_service/swebench_service/pyxis_environment.py b/src/inference_endpoint/evaluation/swebench_service/swebench_service/pyxis_environment.py index 64c381a09..4c8b78a29 100644 --- a/src/inference_endpoint/evaluation/swebench_service/swebench_service/pyxis_environment.py +++ b/src/inference_endpoint/evaluation/swebench_service/swebench_service/pyxis_environment.py @@ -51,18 +51,75 @@ "ENROOT_CONFIG_PATH", ) _STEP_STATUS = "/tmp/.mlperf_srun_status" +#: In-band marker the step script prints alongside its own return code. It is +#: the primary result channel: it travels back on srun's stdout and so needs no +#: readable shared filesystem. The status file remains the fallback. +_STEP_SENTINEL = "__MLPERF_STEP_RC__" +#: The status file contents before the step script runs its very first line. +_STEP_STATUS_PENDING = "pending" _STEP_SCRIPT = r"""set +e status_path=$1 timeout_s=$2 -shift 2 -printf 'started\n' > "$status_path" +nonce=$3 +shift 3 +printf 'started\n' > "$status_path" 2>/dev/null unshare --pid --fork --mount-proc timeout "$timeout_s" "$@" returncode=$? -printf 'finished:%s\n' "$returncode" > "$status_path" +printf 'finished:%s\n' "$returncode" > "$status_path" 2>/dev/null +printf '\n__MLPERF_STEP_RC__ %s %s\n' "$nonce" "$returncode" exit "$returncode" """ +class StepNotLaunched(RunnerError): + """An `srun` step that reported through neither result channel. + + Subclasses :class:`RunnerError` so every existing ``except RunnerError`` + keeps working, and records the facts a caller needs to reason about the + failure rather than only read about it: + + ``srun_rc`` + `srun`'s own exit status. + ``status`` + The bytes actually observed in the step status file. + ``provable_non_execution`` + True only when the status file was still ``pending`` and no in-band + sentinel arrived -- the step script did not run even its first line, so + the command definitely did not execute. Anything else leaves open that + it did, which is the distinction anyone deciding whether a re-run is + safe has to make. + """ + + def __init__( + self, + message: str, + *, + provable_non_execution: bool, + srun_rc: int | None, + status: str, + ) -> None: + super().__init__(message) + self.provable_non_execution = provable_non_execution + self.srun_rc = srun_rc + self.status = status + + +def read_step_sentinel(text: str, nonce: str) -> tuple[int | None, str]: + """Return ``(returncode, output_without_the_sentinel)`` if the step reported. + + ``(None, text)`` when the step did not report in band. The nonce makes the + marker unforgeable by the command's own output. + """ + tag = f"{_STEP_SENTINEL} {nonce} " + for line in reversed((text or "").splitlines()): + if not line.startswith(tag): + continue + value = line[len(tag) :].strip() + if value.lstrip("-").isdigit(): + return int(value), text[: text.rindex(line)].rstrip("\n") + return None, text + + def safe_srun_env() -> dict[str, str]: return {name: os.environ[name] for name in _SAFE_SRUN_ENV if name in os.environ} @@ -116,7 +173,7 @@ def build_srun_command( return command -def run_srun_step( +def _run_srun_step_once( *, argv: list[str], status_path: Path, @@ -128,7 +185,8 @@ def run_srun_step( workdir: str | None = None, stderr: int = subprocess.STDOUT, ) -> subprocess.CompletedProcess[str]: - status_path.write_text("pending\n") + nonce = uuid.uuid4().hex + status_path.write_text(f"{_STEP_STATUS_PENDING}\n") status_path.chmod(0o666) command = build_srun_command( image=image, @@ -142,6 +200,7 @@ def run_srun_step( "pyxis-step", _STEP_STATUS, str(timeout_s), + nonce, *argv, ], ) @@ -170,14 +229,32 @@ def run_srun_step( "Pyxis infrastructure failure before the command completed: " f"{type(exc).__name__}: {exc}" ) from exc - if status_path.read_text().strip() != f"finished:{result.returncode}": - if failure_path is not None: - failure_path.touch() - raise RunnerError( - "Pyxis infrastructure failure before the command completed " - f"(srun exited {result.returncode})" + _srun_evidence(result.stdout) - ) - return result + + # Primary channel: the step reported its own return code in band. + reported, cleaned = read_step_sentinel(result.stdout, nonce) + if reported is not None: + result.stdout = cleaned + result.returncode = reported + return result + + # Fallback channel: the status file the step script wrote into the mount. + try: + status = status_path.read_text().strip() + except OSError as exc: + status = f"" + if status == f"finished:{result.returncode}": + return result + + if failure_path is not None: + failure_path.touch() + raise StepNotLaunched( + "Pyxis infrastructure failure before the command completed " + f"(srun exited {result.returncode}, status={status!r})" + + _srun_evidence(result.stdout), + provable_non_execution=status == _STEP_STATUS_PENDING, + srun_rc=result.returncode, + status=status, + ) def enroot_container_name(job_id: str, container_name: str) -> str: @@ -241,6 +318,111 @@ def _srun_evidence(output: str | bytes | None, limit: int = 2000) -> str: return f"\n--- srun output ---\n{text}" +#: Bounded re-attempts for a step that provably never launched. Set to 1 to +#: disable. A retry here is only ever reached when the step script did not run +#: its first line, so it cannot double-apply work -- see run_srun_step. +_STEP_RETRIES_ENV = "SWEBENCH_PYXIS_STEP_RETRIES" +_DEFAULT_STEP_RETRIES = 3 +#: Optional JSONL sink recording every retry and its outcome. The schema matches +#: `swe_bench_distributed.infra_retry.RetryRecord`, which reads it back to +#: publish infra_retries_total / instances_saved_by_retry / run_quality. The two +#: sides cannot share code: this is an isolated subproject that must not import +#: the benchmark client, so they share a file format instead. +_STEP_RETRY_LOG_ENV = "SWEBENCH_PYXIS_INFRA_RETRY_LOG" +_RETRY_LOG_LOCK = threading.Lock() + + +def _step_retry_attempts() -> int: + raw = os.environ.get(_STEP_RETRIES_ENV, "").strip() + if not raw: + return _DEFAULT_STEP_RETRIES + try: + return max(1, int(raw)) + except ValueError: + logger.warning("ignoring non-numeric %s=%r", _STEP_RETRIES_ENV, raw) + return _DEFAULT_STEP_RETRIES + + +def _record_step_retry( + *, target: str, attempt: int, outcome: str, detail: str | None = None +) -> None: + path = os.environ.get(_STEP_RETRY_LOG_ENV) + if not path: + return + record = { + "target": target, + "attempt": attempt, + "outcome": outcome, + "detail": detail, + "at": time.time(), + } + try: + # Accounting must never be able to take a run down. + with _RETRY_LOG_LOCK, open(path, "a", encoding="utf-8") as handle: + handle.write(json.dumps(record) + "\n") + except OSError: + logger.debug("could not append to the infra retry log", exc_info=True) + + +def run_srun_step(**kwargs: Any) -> subprocess.CompletedProcess[str]: + """Run one `srun` step, re-attempting only a *provable* non-launch. + + Retrying is a correctness decision, not a convenience: re-running a command + that may already have run can apply an edit twice, delete twice, or double a + test run, and none of those announce themselves. So the only failure retried + here is :class:`StepNotLaunched` with ``provable_non_execution`` -- the status + file still ``pending`` and no in-band sentinel, meaning the step script did + not execute even its first line. Every other failure, including a + ``StepNotLaunched`` that reached ``started``, is raised immediately. + + Measured signature, from an isolated probe with no model and no GPU (20 + nodes, 200 workers, 6273 ordinary shell steps): 63 steps failed and in all 63 + the status file still read ``pending``. + + Every attempt and outcome is appended to ``SWEBENCH_PYXIS_INFRA_RETRY_LOG`` + when set. A retry loop that quietly absorbs the defect it compensates for + turns a broken cluster into an invisible one. + """ + attempts = _step_retry_attempts() + target = str(kwargs.get("name") or kwargs.get("image") or "pyxis-step") + for attempt in range(1, attempts + 1): + try: + result = _run_srun_step_once(**kwargs) + except StepNotLaunched as exc: + if not exc.provable_non_execution: + # The command may have run. Another attempt could double it. + _record_step_retry( + target=target, + attempt=attempt, + outcome="not_retryable", + detail=f"srun_rc={exc.srun_rc} status={exc.status!r}", + ) + raise + outcome = "exhausted" if attempt == attempts else "retrying" + _record_step_retry( + target=target, + attempt=attempt, + outcome=outcome, + detail=f"srun_rc={exc.srun_rc} status={exc.status!r}", + ) + if attempt == attempts: + raise + logger.warning( + "Pyxis step provably never launched (attempt %d/%d, srun rc=%s, " + "status=%r); retrying", + attempt, + attempts, + exc.srun_rc, + exc.status, + ) + time.sleep(min(30.0, 2.0 * attempt)) + continue + if attempt > 1: + _record_step_retry(target=target, attempt=attempt, outcome="recovered") + return result + raise AssertionError("unreachable") # pragma: no cover + + def resolve_image(image_registry: str, instance_id: str) -> str: if Path(instance_id).name != instance_id or instance_id in {".", ".."}: raise RunnerError(f"invalid SWE-bench instance ID: {instance_id}") 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..dbc4ab753 --- /dev/null +++ b/tests/unit/evaluation/swe_bench_distributed/test_guards_and_reaper.py @@ -0,0 +1,294 @@ +# 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_infra_retry.py b/tests/unit/evaluation/swe_bench_distributed/test_infra_retry.py new file mode 100644 index 000000000..618484c16 --- /dev/null +++ b/tests/unit/evaluation/swe_bench_distributed/test_infra_retry.py @@ -0,0 +1,291 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Retry only where non-execution is provable, and always count the retries.""" + +from __future__ import annotations + +import json + +import pytest +from inference_endpoint.evaluation.swe_bench_distributed.infra_retry import ( + InfraRetryLedger, + RetryOutcome, + RunQuality, + is_provable_non_execution, + retry_on_provable_non_execution, +) + +pytestmark = pytest.mark.unit + + +class NeverLaunched(RuntimeError): + """Stands in for a step whose status file still read ``pending``.""" + + provable_non_execution = True + + +class MayHaveRun(RuntimeError): + """Stands in for a step that reached ``started`` before it failed.""" + + provable_non_execution = False + + +def _never_sleep(_seconds: float) -> None: + return None + + +def _run(operation, **kwargs): + kwargs.setdefault("target", "run-a.s00") + kwargs.setdefault("sleep", _never_sleep) + return retry_on_provable_non_execution(operation, **kwargs) + + +class TestTheSafetyGate: + def test_a_provable_non_execution_is_retried(self): + calls = [] + + def operation(): + calls.append(1) + if len(calls) < 3: + raise NeverLaunched("status=pending") + return "ok" + + assert _run(operation, max_attempts=3) == "ok" + assert len(calls) == 3 + + def test_a_failure_that_may_have_run_is_never_retried(self): + """The whole safety argument. Re-running could double-apply the work.""" + calls = [] + + def operation(): + calls.append(1) + raise MayHaveRun("status=started") + + with pytest.raises(MayHaveRun): + _run(operation, max_attempts=5) + + assert len(calls) == 1 + + def test_an_unfamiliar_exception_is_not_retried(self): + """Absence of the claim is not evidence for it.""" + calls = [] + + def operation(): + calls.append(1) + raise ValueError("something else entirely") + + with pytest.raises(ValueError): + _run(operation, max_attempts=5) + + assert len(calls) == 1 + + def test_the_attempt_budget_is_bounded(self): + calls = [] + + def operation(): + calls.append(1) + raise NeverLaunched("status=pending") + + with pytest.raises(NeverLaunched): + _run(operation, max_attempts=4) + + assert len(calls) == 4 + + def test_a_successful_first_attempt_costs_nothing(self): + ledger = InfraRetryLedger() + + assert _run(lambda: "ok", ledger=ledger) == "ok" + + assert ledger.records == [] + assert ledger.run_quality() is RunQuality.CLEAN + + @pytest.mark.parametrize( + ("error", "provable"), + [(NeverLaunched(""), True), (MayHaveRun(""), False), (ValueError(""), False)], + ) + def test_provability_is_read_from_the_failure(self, error, provable): + assert is_provable_non_execution(error) is provable + + +class TestAccounting: + def test_a_recovered_operation_is_counted_and_attributed(self): + ledger = InfraRetryLedger() + calls = [] + + def operation(): + calls.append(1) + if len(calls) < 2: + raise NeverLaunched("status=pending") + return "ok" + + _run(operation, ledger=ledger, max_attempts=3) + summary = ledger.summary() + + assert summary["infra_retries_total"] == 2 + assert summary["instances_saved_by_retry"] == 1 + assert summary["infra_retries_exhausted"] == 0 + assert summary["infra_retry_succeeded_on_attempt"] == {"2": 1} + assert summary["infra_retry_outcomes"] == {"retrying": 1, "recovered": 1} + + def test_an_exhausted_operation_is_counted_as_exhausted(self): + ledger = InfraRetryLedger() + + with pytest.raises(NeverLaunched): + _run( + lambda: (_ for _ in ()).throw(NeverLaunched("status=pending")), + ledger=ledger, + max_attempts=2, + ) + summary = ledger.summary() + + assert summary["infra_retries_exhausted"] == 1 + assert summary["instances_saved_by_retry"] == 0 + assert summary["run_quality"] == RunQuality.DEGRADED.value + + def test_a_not_retryable_failure_is_recorded(self): + ledger = InfraRetryLedger() + + with pytest.raises(MayHaveRun): + _run( + lambda: (_ for _ in ()).throw(MayHaveRun("status=started")), + ledger=ledger, + ) + + assert ledger.records[0].outcome is RetryOutcome.NOT_RETRYABLE + + def test_the_ledger_is_durable(self, tmp_path): + """A run that dies must still leave its retry history behind.""" + path = tmp_path / "infra_retries.jsonl" + ledger = InfraRetryLedger(path) + calls = [] + + def operation(): + calls.append(1) + if len(calls) < 2: + raise NeverLaunched("status=pending") + return "ok" + + _run(operation, ledger=ledger, max_attempts=3) + + rows = [json.loads(line) for line in path.read_text().splitlines()] + assert [row["outcome"] for row in rows] == ["retrying", "recovered"] + assert all(row["target"] == "run-a.s00" for row in rows) + + def test_accounting_never_takes_the_run_down(self, tmp_path): + """A ledger that cannot be written must not fail the operation.""" + ledger = InfraRetryLedger(tmp_path / "no-such-dir" / "retries.jsonl") + calls = [] + + def operation(): + calls.append(1) + if len(calls) < 2: + raise NeverLaunched("status=pending") + return "ok" + + assert _run(operation, ledger=ledger, max_attempts=3) == "ok" + # In-memory counters are still correct. + assert ledger.summary()["instances_saved_by_retry"] == 1 + + +class TestRunQuality: + def test_no_retries_is_clean(self): + ledger = InfraRetryLedger() + for _ in range(100): + _run(lambda: "ok", ledger=ledger) + + assert ledger.summary()["run_quality"] == RunQuality.CLEAN.value + + def test_a_few_retries_is_ok_with_retries(self): + ledger = InfraRetryLedger() + for _ in range(200): + _run(lambda: "ok", ledger=ledger) + calls = [] + + def operation(): + calls.append(1) + if len(calls) < 2: + raise NeverLaunched("status=pending") + return "ok" + + _run(operation, ledger=ledger, max_attempts=3) + + assert ledger.summary()["run_quality"] == RunQuality.OK_WITH_RETRIES.value + + def test_many_retries_is_degraded_even_when_everything_succeeded(self): + """A run that leaned on the retry loop is not a clean run.""" + ledger = InfraRetryLedger() + for index in range(10): + calls = [] + + def operation(calls=calls): + calls.append(1) + if len(calls) < 2: + raise NeverLaunched("status=pending") + return "ok" + + _run(operation, ledger=ledger, target=f"unit-{index}", max_attempts=3) + + assert ledger.summary()["run_quality"] == RunQuality.DEGRADED.value + + def test_an_exhaustion_is_degraded_regardless_of_volume(self): + ledger = InfraRetryLedger() + for _ in range(1000): + _run(lambda: "ok", ledger=ledger) + with pytest.raises(NeverLaunched): + _run( + lambda: (_ for _ in ()).throw(NeverLaunched("status=pending")), + ledger=ledger, + max_attempts=2, + ) + + assert ledger.run_quality() is RunQuality.DEGRADED + + +class TestReadingBackAWrittenLedger: + """The SWE-bench service writes this shape from another process. + + It is an isolated subproject and cannot import this package, so the two + halves share a file format. If that agreement breaks, per-step retries stop + reaching the run-level `run_quality` and the run looks clean. + """ + + def test_a_written_ledger_round_trips(self, tmp_path): + path = tmp_path / "infra_retries.jsonl" + source = InfraRetryLedger(path) + source.record(target="unit-1", attempt=1, outcome=RetryOutcome.RETRYING) + source.record(target="unit-1", attempt=2, outcome=RetryOutcome.RECOVERED) + + loaded = InfraRetryLedger.from_jsonl(path) + + assert [r.outcome for r in loaded.records] == [ + RetryOutcome.RETRYING, + RetryOutcome.RECOVERED, + ] + assert loaded.summary()["instances_saved_by_retry"] == 1 + + def test_a_truncated_final_line_does_not_lose_the_history(self, tmp_path): + path = tmp_path / "infra_retries.jsonl" + path.write_text( + '{"target": "u", "attempt": 1, "outcome": "retrying", "at": 1.0}\n' + '{"target": "u", "attempt": 2, "outcome": "recov' + ) + + loaded = InfraRetryLedger.from_jsonl(path) + + assert len(loaded.records) == 1 + + def test_a_missing_ledger_is_an_empty_clean_one(self, tmp_path): + loaded = InfraRetryLedger.from_jsonl(tmp_path / "never-written.jsonl") + + assert loaded.records == [] + assert loaded.run_quality() is RunQuality.CLEAN + + def test_an_exhaustion_written_elsewhere_still_degrades_the_run(self, tmp_path): + path = tmp_path / "infra_retries.jsonl" + path.write_text( + '{"target": "u", "attempt": 1, "outcome": "retrying", "at": 1.0}\n' + '{"target": "u", "attempt": 2, "outcome": "exhausted", "at": 2.0}\n' + ) + + assert InfraRetryLedger.from_jsonl(path).run_quality() is RunQuality.DEGRADED 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..f043348b4 --- /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 +from typing import Any + +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: dict[str, Any] = { + "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 diff --git a/tests/unit/evaluation/swebench_service/test_runner.py b/tests/unit/evaluation/swebench_service/test_runner.py index 594e34bf5..7d3f8cac4 100644 --- a/tests/unit/evaluation/swebench_service/test_runner.py +++ b/tests/unit/evaluation/swebench_service/test_runner.py @@ -7,6 +7,7 @@ import subprocess import sys import threading +import time import types from pathlib import Path from typing import Literal, get_type_hints @@ -22,8 +23,10 @@ ) from inference_endpoint.evaluation.swebench_service.swebench_service.pyxis_environment import ( PyxisEnvironment, + StepNotLaunched, build_srun_command, enroot_container_name, + read_step_sentinel, resolve_image, safe_srun_env, ) @@ -42,6 +45,17 @@ pytestmark = pytest.mark.unit +@pytest.fixture(autouse=True) +def _single_step_attempt(monkeypatch): + """Most tests assert single-shot step behaviour. + + The step runner re-attempts a *provable* non-launch, so without this every + such test would run its fake three times and sleep between them. Retry + behaviour has its own tests, which opt back in explicitly. + """ + monkeypatch.setenv("SWEBENCH_PYXIS_STEP_RETRIES", "1") + + def test_pyxis_implementation_is_confined_to_environment_and_worker_modules(): package_dir = Path(runner_mod.__file__).parent @@ -1226,6 +1240,244 @@ def fake_run(command, **kwargs): assert "failed to start Pyxis container" in str(exc_info.value) +def _bare_environment(tmp_path, failure_path=None): + environment = object.__new__(PyxisEnvironment) + environment.config = types.SimpleNamespace( + cwd="/testbed", + env={}, + timeout_s=30, + interpreter=["bash", "-c"], + infrastructure_failure_path=failure_path, + ) + environment.name = "mswe_run-1_abcd1234" + environment._tmp_dir = tmp_path + return environment + + +@pytest.mark.parametrize( + ("status", "provable"), + [ + # The step script never ran its first line: the command provably did + # not execute, so re-running it cannot double-apply anything. + ("pending\n", True), + # The step script started; the command may well have executed. + ("started\n", False), + # A report for some other return code: the command ran. + ("finished:0\n", False), + ], +) +def test_step_failure_reports_whether_non_execution_is_provable( + monkeypatch, tmp_path, status, provable +): + """srun's text says *what* broke; this says whether a re-run is safe. + + Attaching srun's output made these failures readable. It does not make them + machine-actionable: nothing in the text distinguishes "the step never + launched" from "the command ran and its report was lost", and only the first + can be retried without risking double execution. + """ + environment = _bare_environment(tmp_path) + monkeypatch.setenv("SLURM_JOB_ID", "1738605") + monkeypatch.setenv("SLURMD_NODENAME", "gb-nvl-053-compute04") + + def fake_run(command, **kwargs): + (tmp_path / Path("/tmp/.mlperf_srun_status").name).write_text(status) + return subprocess.CompletedProcess(command, 7, stdout="", stderr="") + + monkeypatch.setattr(subprocess, "run", fake_run) + + with pytest.raises(StepNotLaunched) as exc_info: + environment.execute({"command": "pytest -q"}) + + failure = exc_info.value + assert failure.provable_non_execution is provable + assert failure.status == status.strip() + assert failure.srun_rc == 7 + assert repr(status.strip()) in str(failure) + + +class TestStepRetry: + """Re-attempt only a provable non-launch, and count every attempt. + + Re-running a command that may already have run can apply an edit twice, + delete twice, or double a test run. So the gate is not "an error happened". + """ + + @pytest.fixture(autouse=True) + def _fast(self, monkeypatch): + monkeypatch.setattr(time, "sleep", lambda _seconds: None) + monkeypatch.setenv("SLURM_JOB_ID", "1738605") + monkeypatch.setenv("SLURMD_NODENAME", "gb-nvl-053-compute04") + + def _environment(self, tmp_path): + return _bare_environment(tmp_path) + + def test_a_provable_non_launch_is_retried(self, monkeypatch, tmp_path): + monkeypatch.setenv("SWEBENCH_PYXIS_STEP_RETRIES", "3") + calls = [] + + def fake_run(command, **kwargs): + calls.append(command) + if len(calls) < 3: + # Status file untouched: still "pending". + return subprocess.CompletedProcess(command, 1, stdout="", stderr="") + _finish_srun_step(command, 0) + return subprocess.CompletedProcess(command, 0, stdout="ok\n", stderr="") + + monkeypatch.setattr(subprocess, "run", fake_run) + + output = self._environment(tmp_path).execute({"command": "pytest -q"}) + + assert output["returncode"] == 0 + assert len(calls) == 3 + + def test_a_step_that_started_is_never_retried(self, monkeypatch, tmp_path): + """It may have executed. Another attempt could double-apply it.""" + monkeypatch.setenv("SWEBENCH_PYXIS_STEP_RETRIES", "5") + calls = [] + + def fake_run(command, **kwargs): + calls.append(command) + (tmp_path / ".mlperf_srun_status").write_text("started\n") + return subprocess.CompletedProcess(command, 1, stdout="", stderr="") + + monkeypatch.setattr(subprocess, "run", fake_run) + + with pytest.raises(StepNotLaunched): + self._environment(tmp_path).execute({"command": "rm -rf build"}) + + assert len(calls) == 1 + + def test_the_attempt_budget_is_bounded(self, monkeypatch, tmp_path): + monkeypatch.setenv("SWEBENCH_PYXIS_STEP_RETRIES", "4") + calls = [] + + def fake_run(command, **kwargs): + calls.append(command) + return subprocess.CompletedProcess(command, 1, stdout="", stderr="") + + monkeypatch.setattr(subprocess, "run", fake_run) + + with pytest.raises(StepNotLaunched): + self._environment(tmp_path).execute({"command": "pytest -q"}) + + assert len(calls) == 4 + + def test_every_attempt_is_recorded(self, monkeypatch, tmp_path): + log = tmp_path / "infra_retries.jsonl" + monkeypatch.setenv("SWEBENCH_PYXIS_STEP_RETRIES", "3") + monkeypatch.setenv("SWEBENCH_PYXIS_INFRA_RETRY_LOG", str(log)) + calls = [] + + def fake_run(command, **kwargs): + calls.append(command) + if len(calls) < 2: + return subprocess.CompletedProcess(command, 1, stdout="", stderr="") + _finish_srun_step(command, 0) + return subprocess.CompletedProcess(command, 0, stdout="ok\n", stderr="") + + monkeypatch.setattr(subprocess, "run", fake_run) + + self._environment(tmp_path).execute({"command": "pytest -q"}) + + rows = [json.loads(line) for line in log.read_text().splitlines()] + assert [row["outcome"] for row in rows] == ["retrying", "recovered"] + + def test_an_exhausted_step_is_recorded_as_exhausted(self, monkeypatch, tmp_path): + log = tmp_path / "infra_retries.jsonl" + monkeypatch.setenv("SWEBENCH_PYXIS_STEP_RETRIES", "2") + monkeypatch.setenv("SWEBENCH_PYXIS_INFRA_RETRY_LOG", str(log)) + monkeypatch.setattr( + subprocess, + "run", + lambda command, **kwargs: subprocess.CompletedProcess( + command, 1, stdout="", stderr="" + ), + ) + + with pytest.raises(StepNotLaunched): + self._environment(tmp_path).execute({"command": "pytest -q"}) + + rows = [json.loads(line) for line in log.read_text().splitlines()] + assert [row["outcome"] for row in rows] == ["retrying", "exhausted"] + + def test_accounting_never_takes_the_step_down(self, monkeypatch, tmp_path): + monkeypatch.setenv("SWEBENCH_PYXIS_STEP_RETRIES", "3") + monkeypatch.setenv( + "SWEBENCH_PYXIS_INFRA_RETRY_LOG", str(tmp_path / "nope" / "retries.jsonl") + ) + calls = [] + + def fake_run(command, **kwargs): + calls.append(command) + if len(calls) < 2: + return subprocess.CompletedProcess(command, 1, stdout="", stderr="") + _finish_srun_step(command, 0) + return subprocess.CompletedProcess(command, 0, stdout="ok\n", stderr="") + + monkeypatch.setattr(subprocess, "run", fake_run) + + assert self._environment(tmp_path).execute({"command": "x"})["returncode"] == 0 + + +def test_step_not_launched_is_a_runner_error(): + """Existing ``except RunnerError`` handlers must keep working unchanged.""" + assert issubclass(StepNotLaunched, RunnerError) + + +def test_step_reports_its_return_code_in_band(monkeypatch, tmp_path): + """The sentinel is authoritative and is stripped from the output. + + It removes the shared-filesystem dependency from the success path: a step + can report its result even where the status file is unreadable, which on a + distributed filesystem is a real failure mode of its own. + """ + environment = _bare_environment(tmp_path) + monkeypatch.setenv("SLURM_JOB_ID", "1738605") + monkeypatch.setenv("SLURMD_NODENAME", "gb-nvl-053-compute04") + + def fake_run(command, **kwargs): + nonce = command[command.index("pyxis-step") + 3] + return subprocess.CompletedProcess( + command, + 0, + stdout=f"real output\n\n__MLPERF_STEP_RC__ {nonce} 3\n", + stderr="", + ) + + monkeypatch.setattr(subprocess, "run", fake_run) + + output = environment.execute({"command": "false"}) + + assert output["returncode"] == 3 + assert output["output"] == "real output" + + +def test_step_sentinel_cannot_be_forged_by_command_output(monkeypatch, tmp_path): + environment = _bare_environment(tmp_path) + monkeypatch.setenv("SLURM_JOB_ID", "1738605") + monkeypatch.setenv("SLURMD_NODENAME", "gb-nvl-053-compute04") + + def fake_run(command, **kwargs): + return subprocess.CompletedProcess( + command, 1, stdout="__MLPERF_STEP_RC__ deadbeef 0\n", stderr="" + ) + + monkeypatch.setattr(subprocess, "run", fake_run) + + with pytest.raises(StepNotLaunched): + environment.execute({"command": "echo spoof"}) + + +def test_read_step_sentinel_ignores_unrelated_output(): + assert read_step_sentinel("no marker here\n", "abc") == (None, "no marker here\n") + assert read_step_sentinel("out\n__MLPERF_STEP_RC__ abc x\n", "abc") == ( + None, + "out\n__MLPERF_STEP_RC__ abc x\n", + ) + assert read_step_sentinel("out\n__MLPERF_STEP_RC__ abc -1\n", "abc") == (-1, "out") + + def test_pyxis_environment_preserves_command_failure(monkeypatch, tmp_path): monkeypatch.setenv("SLURM_JOB_ID", "1738605") monkeypatch.setenv("SLURMD_NODENAME", "gb-nvl-053-compute04")