From eabd8288217e597b76805856f72caa3eb3411a65 Mon Sep 17 00:00:00 2001 From: Stanley Phoong Date: Wed, 26 Aug 2026 13:11:07 -0700 Subject: [PATCH 1/7] fix(swebench-service): make a non-launching Pyxis step machine-readable Builds on "give Pyxis container creation its own deadline", which attached srun's own output to these failures and made them *readable*. They are still not *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. 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` -- not `started` -- with srun's output empty. The step script never ran its first line. The command provably did not execute, so re-running it cannot double-apply an edit, a removal or a test run. That is the entire safety argument, and it is only available if the status bytes are captured rather than compared and thrown away. Two changes: * `StepNotLaunched(RunnerError)` records `srun_rc`, the observed `status`, and `provable_non_execution` (`status == "pending"` and no sentinel). It subclasses `RunnerError`, so every existing `except RunnerError` is unaffected, and the status bytes now appear in the message too. * An in-band sentinel, `__MLPERF_STEP_RC__ `, becomes the primary result channel. It travels on srun's stdout, so a step reports its outcome without depending on a readable shared filesystem -- a real failure mode of its own on a distributed one -- and it is what makes "no sentinel" half of the provability test. The nonce makes it unforgeable by the command's own output; it is stripped before the output is returned. This deliberately stops at reporting. Nothing here retries. Tests: `test_step_failure_reports_whether_non_execution_is_provable` covers the three decision points (pending / started / finished), `test_step_reports_its_return_code_in_band`, `test_step_sentinel_cannot_be_forged_by_command_output`, `test_read_step_sentinel_ignores_unrelated_output` and `test_step_not_launched_is_a_runner_error`. Against the parent commit the raised error has no `provable_non_execution`, no `srun_rc`, no `status`, and does not quote the status bytes. --- .../evaluation/swebench_service/README.md | 16 +++ .../swebench_service/pyxis_environment.py | 101 +++++++++++++-- .../swebench_service/test_runner.py | 116 ++++++++++++++++++ 3 files changed, 221 insertions(+), 12 deletions(-) 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..2b5388584 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} @@ -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: diff --git a/tests/unit/evaluation/swebench_service/test_runner.py b/tests/unit/evaluation/swebench_service/test_runner.py index 594e34bf5..52ffd729a 100644 --- a/tests/unit/evaluation/swebench_service/test_runner.py +++ b/tests/unit/evaluation/swebench_service/test_runner.py @@ -22,8 +22,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, ) @@ -1226,6 +1228,120 @@ 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) + + +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") From 69ea14dbb6416db815a908da02df4569e024902b Mon Sep 17 00:00:00 2001 From: Stanley Phoong Date: Sat, 8 Aug 2026 17:40:25 -0700 Subject: [PATCH 2/7] feat(swe-bench): conservative claim reaper + PID-only memory guard reaper.py releases a stale claim only when it has no result, its heartbeat is past stale_after, AND its owner is provably gone. Liveness is a pluggable protocol: LocalProcessLiveness pairs pid with boot id so a recycled pid on a rebooted host is not read as a live owner, and SlurmStepLiveness treats a step missing from scontrol inside a live job as dead, because the job-level rule alone deadlocks the queue forever. An indeterminate probe releases NOTHING - a false reap creates two owners, duplicate results and a wrong denominator. guards.py kills a runaway graded test only under a full conjunction (RSS over threshold AND cwd inside the testbed AND a container-supervisor ancestor). Kills are by PID and refuse self and any ancestor of self; there is no pattern-kill path in the module at all, and a test greps the source to keep it that way. Each term reports its evidence count, and HealthVerdict.combine returns INDETERMINATE rather than UNHEALTHY when a term has zero evidence, so a conjunctive guard cannot collapse into its weakest clause. --- .../swe_bench_distributed/__init__.py | 11 + .../swe_bench_distributed/guards.py | 314 ++++++++++++++++++ .../swe_bench_distributed/reaper.py | 242 ++++++++++++++ .../test_guards_and_reaper.py | 295 ++++++++++++++++ .../swe_bench_distributed/test_liveness.py | 163 +++++++++ 5 files changed, 1025 insertions(+) create mode 100644 src/inference_endpoint/evaluation/swe_bench_distributed/guards.py create mode 100644 src/inference_endpoint/evaluation/swe_bench_distributed/reaper.py create mode 100644 tests/unit/evaluation/swe_bench_distributed/test_guards_and_reaper.py create mode 100644 tests/unit/evaluation/swe_bench_distributed/test_liveness.py diff --git a/src/inference_endpoint/evaluation/swe_bench_distributed/__init__.py b/src/inference_endpoint/evaluation/swe_bench_distributed/__init__.py index c97b98f55..a7b6f488c 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 ( CompletenessReport, MergeRefusal, @@ -25,20 +26,30 @@ UnitResult, WorkQueue, ) +from .reaper import LocalProcessLiveness, OwnerLiveness, SlurmStepLiveness, reap from .units import Unit, UnitPlan, plan_units __all__ = [ "ClaimError", "CompletenessReport", + "HealthTerm", + "HealthVerdict", + "LocalProcessLiveness", + "MemoryGuard", "MergeRefusal", "MergeResult", + "OwnerLiveness", + "SlurmStepLiveness", "Unit", "UnitOutcome", "UnitPlan", "UnitResult", "WorkQueue", "assess_run", + "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 From 639da398f131029bee698534e8e4027d1c62cded Mon Sep 17 00:00:00 2001 From: Stanley Phoong Date: Wed, 26 Aug 2026 13:23:02 -0700 Subject: [PATCH 3/7] feat(swe-bench): retry infrastructure faults only where non-execution is proved The reaper returns a claim whose owner is provably gone. This is the same argument one level down: an operation that provably never ran can be run again, and one that may have run cannot. 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 step's status file still read `pending` -- not `started` -- with no in-band sentinel. The step script never executed its first line. Re-running those commands cannot double-apply an edit, a removal or a test run. That is the entire safety argument, and it is why the gate is `provable_non_execution` rather than "an error happened". A failure that does not make that claim is re-raised immediately and does not consume the budget, which also makes every exception type this module has never heard of safe by default. `infra_retry` provides three things: * `retry_on_provable_non_execution(...)` -- bounded attempts, and the gate. The evidence is read as an attribute rather than an isinstance check, so the producer of the evidence and this consumer stay decoupled. * `InfraRetryLedger` -- every attempt and its outcome, appended as JSONL so a run that dies still leaves its retry history behind, and held in memory so the counters survive a ledger that cannot be written. Accounting must never be able to take a run down. * `summary()` -- `infra_retries_total`, `instances_saved_by_retry`, `infra_retries_exhausted`, the succeeded-on-attempt distribution, and `run_quality: CLEAN | OK_WITH_RETRIES | DEGRADED`. The counting is the point. The banked campaign retried environment faults without limit and without counting them (`wq_worker.sh:41` WQ_MAX_ATTEMPTS=5, `:256` "ENVIRONMENT FAULTS DO NOT CONSUME THE UNIT'S ATTEMPT BUDGET"), which is exactly 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. Measured effect of adding this loop: RunnerError 59 -> 7 and resolve 47.0% -> 70.0% against a banked 70.67% on the identical 200 instances -- a rescue on that scale is not a clean run, and `run_quality` says so even at 200/200. This is the fleet-side half: the decision rule, the accounting and the quality verdict. The next commit applies the same rule inside the SWE-bench service, where the Pyxis step that produces `provable_non_execution` runs. Tests: `TestTheSafetyGate` covers both sides of the decision boundary (`pending` retried, `started` never retried, unfamiliar exception never retried) plus the bound; `TestAccounting` covers recovery, exhaustion, not-retryable, ledger durability and a ledger that cannot be written; `TestRunQuality` covers all three verdicts including DEGRADED on volume alone, where every operation eventually succeeded. --- .../swe_bench_distributed/__init__.py | 14 + .../swe_bench_distributed/infra_retry.py | 261 ++++++++++++++++++ .../swe_bench_distributed/test_infra_retry.py | 243 ++++++++++++++++ 3 files changed, 518 insertions(+) create mode 100644 src/inference_endpoint/evaluation/swe_bench_distributed/infra_retry.py create mode 100644 tests/unit/evaluation/swe_bench_distributed/test_infra_retry.py diff --git a/src/inference_endpoint/evaluation/swe_bench_distributed/__init__.py b/src/inference_endpoint/evaluation/swe_bench_distributed/__init__.py index a7b6f488c..7269ab520 100644 --- a/src/inference_endpoint/evaluation/swe_bench_distributed/__init__.py +++ b/src/inference_endpoint/evaluation/swe_bench_distributed/__init__.py @@ -12,6 +12,14 @@ """ 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, @@ -34,11 +42,15 @@ "CompletenessReport", "HealthTerm", "HealthVerdict", + "InfraRetryLedger", "LocalProcessLiveness", "MemoryGuard", "MergeRefusal", "MergeResult", "OwnerLiveness", + "RetryOutcome", + "RetryRecord", + "RunQuality", "SlurmStepLiveness", "Unit", "UnitOutcome", @@ -47,9 +59,11 @@ "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/infra_retry.py b/src/inference_endpoint/evaluation/swe_bench_distributed/infra_retry.py new file mode 100644 index 000000000..20d51e19d --- /dev/null +++ b/src/inference_endpoint/evaluation/swe_bench_distributed/infra_retry.py @@ -0,0 +1,261 @@ +# 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 + +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 + + +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 + ) + + 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[T]( + 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/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..9b803fe35 --- /dev/null +++ b/tests/unit/evaluation/swe_bench_distributed/test_infra_retry.py @@ -0,0 +1,243 @@ +# 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 From 402b433a12ad9a4107b1b8807147baacc9dd4fd2 Mon Sep 17 00:00:00 2001 From: Stanley Phoong Date: Wed, 26 Aug 2026 13:55:36 -0700 Subject: [PATCH 4/7] feat(swebench-service): re-attempt a Pyxis step that provably never launched Wires the retry decision into the place the failure actually happens. `run_srun_step()` now re-attempts a step only when `StepNotLaunched` reports `provable_non_execution` -- the status file still `pending` and no in-band sentinel, so the step script did not run even its first line and the command definitely did not execute. A `StepNotLaunched` that reached `started`, and every other failure, is raised immediately: re-running work that may already have run can apply an edit twice, delete twice, or double a test run, and none of those announce themselves. 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`. Measured effect of retrying exactly those: `RunnerError` 59 -> 7 and resolve 47.0% -> 70.0% against a banked 70.67% on the identical 200 instances. Bounded by `SWEBENCH_PYXIS_STEP_RETRIES` (default 3, set 1 to disable), and every attempt and outcome is appended to `SWEBENCH_PYXIS_INFRA_RETRY_LOG` when set. The record shape is deliberately identical to `swe_bench_distributed.infra_retry.RetryRecord`, which gains `InfraRetryLedger.from_jsonl()` to read it back and publish `infra_retries_total`, `instances_saved_by_retry`, `infra_retries_exhausted` and `run_quality`. The service is an isolated subproject and must not import the benchmark client, so the two halves share a file format rather than a module -- and a test on each side pins that agreement, because if it breaks the retries stop reaching the run-level quality flag and a rescued run looks clean. A retry loop that quietly absorbs the defect it compensates for turns a broken cluster into an invisible one. That is why the accounting is not optional and why `run_quality` reports DEGRADED on volume alone. Tests: `TestStepRetry` covers both sides of the boundary (`pending` retried, `started` never retried), the bound, the recorded outcomes for recovery and exhaustion, and a log path that cannot be written. `TestReadingBackAWritten Ledger` covers the cross-process format, including a truncated final line from a run that died. An autouse fixture pins the existing single-shot tests to one attempt so they keep asserting single-shot behaviour. --- .../swe_bench_distributed/infra_retry.py | 41 ++++++ .../swebench_service/pyxis_environment.py | 107 +++++++++++++- .../swe_bench_distributed/test_infra_retry.py | 49 +++++++ .../swebench_service/test_runner.py | 137 ++++++++++++++++++ 4 files changed, 333 insertions(+), 1 deletion(-) diff --git a/src/inference_endpoint/evaluation/swe_bench_distributed/infra_retry.py b/src/inference_endpoint/evaluation/swe_bench_distributed/infra_retry.py index 20d51e19d..013a6f7bb 100644 --- a/src/inference_endpoint/evaluation/swe_bench_distributed/infra_retry.py +++ b/src/inference_endpoint/evaluation/swe_bench_distributed/infra_retry.py @@ -152,6 +152,47 @@ def record( "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 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 2b5388584..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 @@ -173,7 +173,7 @@ def build_srun_command( return command -def run_srun_step( +def _run_srun_step_once( *, argv: list[str], status_path: Path, @@ -318,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_infra_retry.py b/tests/unit/evaluation/swe_bench_distributed/test_infra_retry.py index 9b803fe35..a34471ca0 100644 --- a/tests/unit/evaluation/swe_bench_distributed/test_infra_retry.py +++ b/tests/unit/evaluation/swe_bench_distributed/test_infra_retry.py @@ -241,3 +241,52 @@ def test_an_exhaustion_is_degraded_regardless_of_volume(self): ) 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/swebench_service/test_runner.py b/tests/unit/evaluation/swebench_service/test_runner.py index 52ffd729a..6497a20cf 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 @@ -14,6 +15,7 @@ import msgspec.json import pytest import yaml + from inference_endpoint.evaluation.swebench_service.swebench_service import ( pyxis_worker as worker_mod, ) @@ -44,6 +46,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 @@ -1284,6 +1297,130 @@ def fake_run(command, **kwargs): 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) From 39adfdf92a19a1e78530b0260c47b9c6b7922eac Mon Sep 17 00:00:00 2001 From: Stanley Phoong Date: Thu, 27 Aug 2026 08:28:52 -0700 Subject: [PATCH 5/7] fix(swe-bench): make liveness protocol stub explicit --- .../evaluation/swe_bench_distributed/reaper.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/inference_endpoint/evaluation/swe_bench_distributed/reaper.py b/src/inference_endpoint/evaluation/swe_bench_distributed/reaper.py index 0905cee42..3224193db 100644 --- a/src/inference_endpoint/evaluation/swe_bench_distributed/reaper.py +++ b/src/inference_endpoint/evaluation/swe_bench_distributed/reaper.py @@ -49,7 +49,8 @@ class LivenessVerdict: class OwnerLiveness(Protocol): """Decides whether the process that claimed a unit still exists.""" - def probe(self, owner: OwnerRecord) -> LivenessVerdict: ... + def probe(self, owner: OwnerRecord) -> LivenessVerdict: + pass class LocalProcessLiveness: From a250cb0cb26f8b26d9d2c7afcbbb4375f295d320 Mon Sep 17 00:00:00 2001 From: Stanley Phoong Date: Thu, 27 Aug 2026 08:39:04 -0700 Subject: [PATCH 6/7] fix(swe-bench): keep retry typing mypy compatible --- .../evaluation/swe_bench_distributed/infra_retry.py | 10 ++++++---- .../evaluation/swe_bench_distributed/test_liveness.py | 3 ++- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/src/inference_endpoint/evaluation/swe_bench_distributed/infra_retry.py b/src/inference_endpoint/evaluation/swe_bench_distributed/infra_retry.py index 013a6f7bb..c9a8b0cb8 100644 --- a/src/inference_endpoint/evaluation/swe_bench_distributed/infra_retry.py +++ b/src/inference_endpoint/evaluation/swe_bench_distributed/infra_retry.py @@ -37,7 +37,7 @@ from dataclasses import dataclass from enum import StrEnum from pathlib import Path -from typing import Any +from typing import Any, TypeVar logger = logging.getLogger(__name__) @@ -47,6 +47,8 @@ #: 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. @@ -235,8 +237,8 @@ def run_quality(self) -> RunQuality: return RunQuality.OK_WITH_RETRIES -def retry_on_provable_non_execution[T]( - operation: Callable[[], T], +def retry_on_provable_non_execution( + operation: Callable[[], _T], *, target: str, ledger: InfraRetryLedger | None = None, @@ -244,7 +246,7 @@ def retry_on_provable_non_execution[T]( backoff_s: float = 2.0, max_backoff_s: float = 30.0, sleep: Callable[[float], None] = time.sleep, -) -> T: +) -> _T: """Call ``operation``, retrying only failures that prove it never ran. Raises the last failure when the budget is exhausted, and re-raises diff --git a/tests/unit/evaluation/swe_bench_distributed/test_liveness.py b/tests/unit/evaluation/swe_bench_distributed/test_liveness.py index e65b32b9b..054adb66f 100644 --- a/tests/unit/evaluation/swe_bench_distributed/test_liveness.py +++ b/tests/unit/evaluation/swe_bench_distributed/test_liveness.py @@ -8,6 +8,7 @@ import os import socket import subprocess +from typing import Any import pytest @@ -22,7 +23,7 @@ def owner(**overrides) -> OwnerRecord: - payload = { + payload: dict[str, Any] = { "unit_id": "run-a.s00", "host": socket.gethostname(), "pid": os.getpid(), From c1ebd2d3f6e86e002be0b4c4faec993b8e221796 Mon Sep 17 00:00:00 2001 From: Stanley Phoong Date: Thu, 27 Aug 2026 08:49:00 -0700 Subject: [PATCH 7/7] style(swe-bench): normalize distributed test imports --- .../evaluation/swe_bench_distributed/test_guards_and_reaper.py | 1 - tests/unit/evaluation/swe_bench_distributed/test_infra_retry.py | 1 - tests/unit/evaluation/swe_bench_distributed/test_liveness.py | 1 - tests/unit/evaluation/swebench_service/test_runner.py | 1 - 4 files changed, 4 deletions(-) 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 index aebdec290..dbc4ab753 100644 --- a/tests/unit/evaluation/swe_bench_distributed/test_guards_and_reaper.py +++ b/tests/unit/evaluation/swe_bench_distributed/test_guards_and_reaper.py @@ -9,7 +9,6 @@ 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, diff --git a/tests/unit/evaluation/swe_bench_distributed/test_infra_retry.py b/tests/unit/evaluation/swe_bench_distributed/test_infra_retry.py index a34471ca0..618484c16 100644 --- a/tests/unit/evaluation/swe_bench_distributed/test_infra_retry.py +++ b/tests/unit/evaluation/swe_bench_distributed/test_infra_retry.py @@ -8,7 +8,6 @@ import json import pytest - from inference_endpoint.evaluation.swe_bench_distributed.infra_retry import ( InfraRetryLedger, RetryOutcome, diff --git a/tests/unit/evaluation/swe_bench_distributed/test_liveness.py b/tests/unit/evaluation/swe_bench_distributed/test_liveness.py index 054adb66f..f043348b4 100644 --- a/tests/unit/evaluation/swe_bench_distributed/test_liveness.py +++ b/tests/unit/evaluation/swe_bench_distributed/test_liveness.py @@ -11,7 +11,6 @@ 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, diff --git a/tests/unit/evaluation/swebench_service/test_runner.py b/tests/unit/evaluation/swebench_service/test_runner.py index 6497a20cf..7d3f8cac4 100644 --- a/tests/unit/evaluation/swebench_service/test_runner.py +++ b/tests/unit/evaluation/swebench_service/test_runner.py @@ -15,7 +15,6 @@ import msgspec.json import pytest import yaml - from inference_endpoint.evaluation.swebench_service.swebench_service import ( pyxis_worker as worker_mod, )