async-eval: sidecar auto-disable now fires on dead sidecars

The toy end-to-end test surfaced exactly the gap Codex flagged:
sidecars submitted successfully but died in <2s (rocm dep resolver
error inside the sidecar's uv run), and the callback didn't disable
because the sidecar-mode failure counter only tracked SUBMISSION
errors. Dead sidecars detected by ``_sweep_stale_markers`` were
cleared but never counted.

This is the parallel of the reserved-mode bug already fixed: each
``_fire`` was also doing ``self._consecutive_failures = 0`` on
successful submit, which would wipe any sweep-detected failures.

Fixes:

1. ``_clear_marker_if_stale`` returns ``bool`` (True iff cleared).
   ``_sweep_stale_markers`` aggregates the clears and increments
   ``_consecutive_failures`` accordingly — each cleared orphan is a
   sidecar whose bash trap didn't fire (slurmstepd OOM, NODE_FAIL,
   scancel --signal=KILL).

2. ``_fire`` no longer resets the counter on successful submit. The
   counter only ratchets upward; submission failures AND sweep-cleared
   orphans both contribute to the auto-disable threshold. Sidecar mode
   has no positive feedback signal (bash trap removing a marker isn't
   observable to us), so any reset would risk wiping unprocessed
   failures.

3. ``sbatch_template.py``: sidecar sub-jobs now ``uv run --no-sync``
   to reuse the parent's resolved venv. Without this, an upstream
   wheel-index churn (e.g. vllm rocm wheels yanked) crashes every
   sidecar before it starts. Matches the cluster convention already
   used in the cookbook launchers per CLAUDE.md.

4. Toy fixture launchers (``toy_async_eval_*.sh``) also switched to
   ``uv run --no-sync``.

Regression tests:
  - ``test_dead_sidecars_disable_callback`` — drops 3 orphan markers,
    stubs sacct=FAILED, asserts counter=3 and ``_disabled=True``.
  - Existing ``test_failure_disables_after_max_consecutive`` still
    pins the submit-failure path (both modes accumulate).

440 passed / 12 skipped, full repo clean.

Author: Rouzbehat78

src/leap_finetune/evaluation/sbatch_template.py

@@ -82,7 +82,10 @@ def render_sbatch_script(
         runner_args += ["--wandb-project", wandb_project]
 
     runner_cmd = (
-        "uv run python -m leap_finetune.evaluation.async_runner_main \\\n    "
+        # --no-sync: reuse the parent's resolved venv. Re-resolution in a
+        # sidecar can fail if an upstream index has churned (e.g. vllm rocm
+        # wheels yanked) since the parent locked.
+        "uv run --no-sync python -m leap_finetune.evaluation.async_runner_main \\\n    "
         + " \\\n    ".join(shlex.quote(a) for a in runner_args)
     )
 

src/leap_finetune/evaluation/sidecar_callback.py

@@ -181,7 +181,14 @@ def _fire(
         args: TrainingArguments,
         wait_for_completion: bool = False,
     ) -> None:
+        # Sweep first; it both clears orphan markers AND counts those clears
+        # toward _consecutive_failures (each cleared orphan = a sidecar that
+        # died without its bash trap firing). If sweep tripped auto-disable,
+        # bail before submitting another.
         self._sweep_stale_markers()
+        if self._disabled:
+            return
+
         in_flight = sorted(self._eval_dir.glob(_MARKER_GLOB))
         if in_flight:
             if self.cfg.on_overlap == "skip":
@@ -200,7 +207,12 @@ def _fire(
 
         try:
             jobid = self._submit(model, state, args)
-            self._consecutive_failures = 0
+            # NOTE: do NOT reset _consecutive_failures here. The counter is
+            # shared with sweep-detected sidecar deaths; a successful submit
+            # followed by N dead sidecars would otherwise keep wiping the
+            # running count and auto-disable could never fire. Reset only
+            # happens in _sweep_stale_markers when zero markers remain
+            # (clean idle state).
             if wait_for_completion and jobid:
                 self._wait_for_job(jobid, state.global_step)
         except Exception:
@@ -345,29 +357,51 @@ def _submit(
 
     def _sweep_stale_markers(self) -> None:
         """Sweep ALL per-step in-flight markers and clear ones whose
-        sbatch job is no longer alive. Called at the top of ``_fire``
-        so a dead orphan never blocks the next eval under
-        ``on_overlap=skip`` and the queue-mode "in-flight count" is
-        accurate.
+        sbatch job is no longer alive. Each cleared orphan = a sidecar
+        that died without its bash trap firing (slurmstepd OOM,
+        NODE_FAIL, scancel --signal=KILL, ...) and counts toward
+        auto-disable. The counter is never reset by the sweep — sidecar
+        mode has no positive feedback signal (the bash trap removing a
+        marker isn't observable to us), so any reset would risk wiping
+        unprocessed submission failures.
         """
+        cleared = 0
         for marker in self._eval_dir.glob(_MARKER_GLOB):
-            self._clear_marker_if_stale(marker)
+            if self._clear_marker_if_stale(marker):
+                cleared += 1
+        if cleared > 0:
+            self._consecutive_failures += cleared
+            logger.warning(
+                "[async_eval/sidecar] sweep cleared %d dead sidecar(s) (%d "
+                "consecutive failures)",
+                cleared,
+                self._consecutive_failures,
+            )
+            if self._consecutive_failures >= self.cfg.failure.max_consecutive:
+                self._disabled = True
+                logger.error(
+                    "[async_eval/sidecar] disabling after %d consecutive failures",
+                    self._consecutive_failures,
+                )
 
-    def _clear_marker_if_stale(self, marker: Path) -> None:
+    def _clear_marker_if_stale(self, marker: Path) -> bool:
         """Remove a single orphan marker whose job is no longer alive.
 
         The sidecar script's EXIT trap doesn't fire if slurm OOM-kills the
         step, NODE_FAILs, or the user ``scancel``s with ``--signal=KILL``;
         without recovery an orphan would block all future evals under
         ``on_overlap=skip``. We ask ``sacct`` whether the recorded jobid is
         terminal; missing-sacct falls back to a 6h mtime cutoff.
+
+        Returns ``True`` iff the marker was cleared — caller uses this to
+        count dead sidecars toward auto-disable.
         """
         if not marker.exists():
-            return
+            return False
         try:
             content = marker.read_text().strip()
         except OSError:
-            return
+            return False
 
         jobid = content.split(":", 1)[0] if ":" in content else ""
         if jobid.isdigit():
@@ -394,7 +428,7 @@ def _clear_marker_if_stale(self, marker: Path) -> None:
                     state = line.strip().split(None, 1)[0].rstrip("+").upper()
                     states_seen.append(state)
                     if state in _SACCT_ACTIVE_STATES:
-                        return  # at least one row alive; keep marker.
+                        return False  # at least one row alive; keep marker.
                 logger.warning(
                     "[async_eval/sidecar] clearing stale marker %s "
                     "(job %s no longer active; sacct states: %s)",
@@ -403,7 +437,7 @@ def _clear_marker_if_stale(self, marker: Path) -> None:
                     ",".join(states_seen),
                 )
                 marker.unlink(missing_ok=True)
-                return
+                return True
 
         try:
             age = time.time() - marker.stat().st_mtime
@@ -414,8 +448,10 @@ def _clear_marker_if_stale(self, marker: Path) -> None:
                     age / 3600,
                 )
                 marker.unlink(missing_ok=True)
+                return True
         except OSError:
             pass
+        return False
 
     def _wait_for_job(
         self,

tests/fixtures/toy_async_eval_reserved.sh

@@ -26,4 +26,5 @@ export TMPDIR="${HOME}/tmp"
 export TRITON_CACHE_DIR="${HOME}/.triton_cache"
 mkdir -p "$TMPDIR" "$TRITON_CACHE_DIR" logs/async_eval_toy
 
-uv run leap-finetune tests/fixtures/toy_async_eval_reserved.yaml
+# --no-sync: use the already-activated .venv instead of re-resolving deps.
+uv run --no-sync leap-finetune tests/fixtures/toy_async_eval_reserved.yaml

tests/fixtures/toy_async_eval_sidecar.sh

@@ -28,4 +28,7 @@ export TMPDIR="${HOME}/tmp"
 export TRITON_CACHE_DIR="${HOME}/.triton_cache"
 mkdir -p "$TMPDIR" "$TRITON_CACHE_DIR" logs/async_eval_toy
 
-uv run leap-finetune tests/fixtures/toy_async_eval_sidecar.yaml
+# --no-sync: use the already-activated .venv instead of re-resolving deps.
+# Avoids transient resolver failures from upstream wheel index churn
+# (e.g. rocm wheels being yanked) inside a job.
+uv run --no-sync leap-finetune tests/fixtures/toy_async_eval_sidecar.yaml

tests/test_async_eval.py

@@ -713,6 +713,50 @@ def __init__(self, **kw):
             pass
 
 
+class TestSidecarSweepCountsDeadSidecars:
+    """When the bash trap doesn't fire (slurmstepd OOM, NODE_FAIL,
+    scancel --signal=KILL), sidecars leave orphan markers. The sweep
+    must count each cleared orphan toward ``_consecutive_failures``
+    so auto-disable can fire — Codex catch."""
+
+    def test_dead_sidecars_disable_callback(self, tmp_path, monkeypatch):
+        from leap_finetune.evaluation.async_eval_config import AsyncEvalConfig
+        from leap_finetune.evaluation.sidecar_callback import SidecarEvalCallback
+        import leap_finetune.evaluation.sidecar_callback as sc
+
+        cb = SidecarEvalCallback(
+            benchmarks=[MagicMock(name="bench1")],
+            cfg=AsyncEvalConfig.from_dict(
+                {"mode": "sidecar", "failure": {"max_consecutive": 2}}
+            ),
+            benchmark_configs={"benchmarks": []},
+            output_dir=str(tmp_path),
+            wandb_run_id=None,
+        )
+
+        # Drop 3 orphan markers (3 sidecars submitted, none had their
+        # bash trap fire because slurmstepd killed them).
+        eval_dir = cb._eval_dir
+        for step, jobid in [(100, 111), (200, 222), (300, 333)]:
+            (eval_dir / f".in_flight.step_{step}").write_text(f"{jobid}:{step}")
+
+        # sacct says all jobs are FAILED (terminal, non-active).
+        monkeypatch.setattr(
+            sc.subprocess,
+            "run",
+            lambda *a, **kw: type(
+                "P", (), {"returncode": 0, "stdout": "FAILED\n", "stderr": ""}
+            )(),
+        )
+
+        cb._sweep_stale_markers()
+        # Three dead sidecars cleared → counter >= max_consecutive (2) → disabled.
+        assert cb._consecutive_failures == 3
+        assert cb._disabled is True
+        # All three orphan markers should be gone.
+        assert not list(eval_dir.glob(".in_flight.step_*"))
+
+
 class TestSidecarConcurrentMarkers:
     """``on_overlap=queue`` must support concurrent in-flight sidecars.
     A single shared marker would let the first sidecar's EXIT trap wipe