feat(cli): sieval run --dry-run reports each task's stage plan

CHANGELOG.md

@@ -26,6 +26,7 @@ Planned as 0.12.0.
 - `sieval.core.runners.task_resume_plan` / `plan_resume`: the per-stage resume plan, without running anything.
 - `TaskRunner.plan_run` / `MultiTaskRunner.plan_run` (`RunPlan`): the runner's per-sample plan of a run, `to_infer` included, read-only; the one `sieval run` launches by.
 - **`sieval run --dry-run` reports what the run would do** (closes #214). `-o json` adds `gpu`, `models` (each served model's plan as `infer_plans.yaml` would record it, each process its launch starts, and `needs_inference`, whether this run starts it), `tasks` (each task's `start`, `resume_plan` with the counts `runs[].resume` will record, `null` for a fresh start, `new`, `to_infer` and `to_grade`) and `needs_inference`, whether any engine starts. The plans come from the call `sieval run` launches by. A resume that would be refused, or planning that raises, fails the new `resume_plan` check; when the dry-run stops before planning, `tasks` and `needs_inference` are `null`. It launches no engine, writes no result directory, reads `--resume` directories in place, and runs on a node without GPUs given `SIEVAL_TARGET_GPU`. Planning loads the datasets and starts each vendored service a task declares, as the run does, so a dry-run that passes its checks costs a session setup. The existing checks are unchanged.
+- `sieval run --dry-run` reports each task's `stages` (per-stage `same`/`differs`/`unknown`, and `rerun_from`, the stage a resume reruns from) and its own `needs_inference`; `RunPlan` gains `stages` (closes #219).
 
 ## [0.11.0] - 2026-10-23
 

docs/guide/configuration.md

@@ -348,9 +348,17 @@ in `SIEVAL_TARGET_GPU` ([Resolving for declared GPUs](infer.md#resolving-for-dec
       "recorded": "outputs/run/echo",
       "samples": 3,
       "resume_plan": {"kept": 2, "rereported": 0, "regraded": 0, "repostprocessed": 0, "restarted": 0, "dropped": 0},
+      "stages": {
+        "identity_schema": 2,
+        "stages": {"data": "differs", "inference": "same", "postprocess": "same", "feedback": "same", "report": "same"},
+        "rerun_from": null,
+        "refused": null,
+        "legacy_grader": null
+      },
       "new": 1,
       "to_infer": 1,
-      "to_grade": 1
+      "to_grade": 1,
+      "needs_inference": true
     }
   },
   "needs_inference": true
@@ -368,7 +376,12 @@ in `SIEVAL_TARGET_GPU` ([Resolving for declared GPUs](infer.md#resolving-for-dec
   `resume_plan` counts the recorded samples as `runs[].resume` will, and is
   `null` for a fresh start; `new` counts samples with no record. `to_infer`
   counts the samples sent to the model under test, `to_grade` those run through
-  a grading stage.
+  a grading stage; `needs_inference` is whether `to_infer` is above zero.
+- `tasks.<task>.stages` compares each stage's recorded identity with this
+  run's: `same`, `differs` or `unknown`. `rerun_from` is the grading stage the
+  resume reruns from (`report` for a re-report), else `null`; a data change
+  shows in `resume_plan` as `restarted`. `legacy_grader` is `kept` or `regrade`
+  for a schema-1 directory. `stages` is `null` for a fresh start.
 - `needs_inference` is whether the run starts any engine. A task that infers
   through an `api_base` model counts in its `to_infer` but starts none.
 

sieval/cli/output.py

@@ -364,11 +364,13 @@ def _render_dry_run_plan(data: dict) -> None:
         outcomes = ", ".join(
             f"{n} {outcome}" for outcome, n in (entry["resume_plan"] or {}).items() if n
         )
+        rerun_from = (entry.get("stages") or {}).get("rerun_from")
         log_user(
-            "task {}: {}{}, {} to infer, {} to grade",
+            "task {}: {}{}{}, {} to infer, {} to grade",
             name,
             entry["start"],
             f" ({outcomes})" if outcomes else "",
+            f", from {rerun_from}" if rerun_from else "",
             entry["to_infer"],
             entry["to_grade"],
         )

sieval/cli/run.py

@@ -354,12 +354,15 @@ def _models_to_launch(
     return {
         task_models[task]
         for task, plan in plans.items()
-        if not isinstance(plan, Exception)
-        and plan["to_infer"]
-        and task_models.get(task) in served
+        if _infers(plan) and task_models.get(task) in served
     }
 
 
+def _infers(plan: RunPlan | Exception) -> bool:
+    """Whether a task's run sends any sample to its model under test."""
+    return not isinstance(plan, Exception) and plan["to_infer"] > 0
+
+
 async def _models_to_infer(
     config_path: Path,
     config: Mapping[str, Any],
@@ -401,7 +404,9 @@ def _task_json(model: str | None, plan: RunPlan | Exception) -> dict[str, object
     """A task's entry in the dry-run report.
 
     ``resume_plan`` is the plan's ``outcomes``, the counts ``runs[].resume``
-    records; ``None`` for a fresh start, which has no recorded samples.
+    records, and ``stages`` its stage comparison; both ``None`` for a fresh
+    start, which has no recorded samples. ``needs_inference`` is
+    :func:`_infers`, the test :func:`_models_to_launch` launches by.
     """
     if isinstance(plan, Exception):
         return {"model": model, "error": cli_error_message(plan)}
@@ -411,9 +416,11 @@ def _task_json(model: str | None, plan: RunPlan | Exception) -> dict[str, object
         "recorded": plan["recorded"],
         "samples": plan["samples"],
         "resume_plan": plan["outcomes"],
+        "stages": plan["stages"],
         "new": plan["new"],
         "to_infer": plan["to_infer"],
         "to_grade": plan["to_grade"],
+        "needs_inference": _infers(plan),
     }
 
 

sieval/core/runners/runner.py

@@ -106,6 +106,7 @@
 from .resume_option import as_resume, fold_resume_aliases, resume_source
 from .stage_resume import (
     LegacyStageMatch,
+    ResumePlan,
     RunPlan,
     StageResumeError,
     abandon_copy,
@@ -115,6 +116,7 @@
     end_copy,
     has_graded,
     iteration_dirs,
+    plan_resume,
     recorded_grader,
     recorded_identity,
     recorded_sample_digests,
@@ -2025,6 +2027,7 @@ async def _plan_run(self) -> RunPlan:
                 "recorded": None,
                 "samples": size,
                 "outcomes": None,
+                "stages": None,
                 "new": size,
                 "to_infer": size,
                 "to_grade": size,
@@ -2043,6 +2046,7 @@ async def _plan_run(self) -> RunPlan:
                 "recorded": recorded,
                 "samples": size,
                 "outcomes": outcomes,
+                "stages": self._stage_plan(),
                 "new": 0,
                 "to_infer": 0,
                 "to_grade": 0,
@@ -2062,11 +2066,30 @@ async def _plan_run(self) -> RunPlan:
             "recorded": recorded,
             "samples": size,
             "outcomes": self._resume_plan,
+            "stages": self._stage_plan(),
             "new": sum(1 for sid in range(size) if sid not in self._contexts),
             "to_infer": to_infer,
             "to_grade": to_grade,
         }
 
+    def _stage_plan(self) -> ResumePlan:
+        """:func:`plan_resume` over this runner's identities, with its decisions.
+
+        ``preprocess`` is decided per sample (``restarted``), so it is left
+        out. ``rerun_from`` is the runner's, which trusts a directory with no
+        identities; ``refused`` is ``None``, as a refusal raises before this.
+        """
+        current: dict[str, object] = {
+            "data": self._dataset_identity(),
+            "inference": self._inference_identity(),
+            **(self._grading_identities() or {}),
+        }
+        plan = plan_resume(self._recorded, current, self._legacy_grader_keys())
+        del plan["stages"]["preprocess"]
+        plan["rerun_from"] = self._recorded_grading_change()
+        plan["refused"] = None
+        return plan
+
     def _pending_work(self) -> tuple[int, int]:
         """``(to_infer, to_grade)``: the samples a run starting now would send
         to the model under test, and those it would grade.

sieval/core/runners/stage_resume.py

@@ -93,15 +93,18 @@ class RunPlan(TypedDict):
     ``start`` is ``fresh`` (nothing recorded), ``resume`` (stage by stage), or
     ``reuse`` (the recorded ``report.json`` stands, recomputed when only
     ``report`` changed). ``outcomes`` counts the recorded samples as
-    ``runs[].resume`` will, ``None`` for a fresh run. ``new`` counts samples
-    with no record. ``to_infer`` counts the samples the run sends to the model
-    under test, ``to_grade`` those it runs through a grading stage.
+    ``runs[].resume`` will, ``None`` for a fresh run. ``stages`` is the
+    recorded directory's :class:`ResumePlan` against this run, ``None`` for a
+    fresh run; its ``rerun_from`` is the runner's own decision. ``new`` counts
+    samples with no record. ``to_infer`` counts the samples the run sends to
+    the model under test, ``to_grade`` those it runs through a grading stage.
     """
 
     start: Literal["fresh", "resume", "reuse"]
     recorded: str | None
     samples: int
     outcomes: ResumeOutcomes | None
+    stages: "ResumePlan | None"
     new: int
     to_infer: int
     to_grade: int

tests/integration/test_lazy_launch.py

@@ -8,7 +8,10 @@
 AI-Generated Code - Claude Opus 5.5 (Anthropic)
 """
 
+import importlib
 import json
+import sys
+import textwrap
 from collections.abc import Iterator
 from pathlib import Path
 from typing import Any
@@ -419,10 +422,19 @@ def test_a_resume_reports_its_plan_and_reads_dirs_read_only(
             "recorded": str(tmp_path / "results" / "echo"),
             "samples": 3,
             "resume_plan": _resume_plan(kept=2),
+            "stages": data["tasks"]["echo"]["stages"],
             "new": 1,
             "to_infer": 1,
             "to_grade": 1,
+            "needs_inference": True,
         }
+        stages = data["tasks"]["echo"]["stages"]
+        assert stages["identity_schema"] == 2
+        assert stages["rerun_from"] is None
+        # A third sample changes the set's data identity; each recorded sample
+        # still matches its own digest, so both are kept.
+        assert stages["stages"]["data"] == "differs"
+        assert stages["stages"]["inference"] == "same"
         assert engine.launches == []
         assert _tree(tmp_path / "results") == recorded
 
@@ -450,6 +462,8 @@ def test_nothing_to_infer(
         assert data["models"]["subject"]["needs_inference"] is False
         echo = data["tasks"]["echo"]
         assert echo["resume_plan"] == _resume_plan(repostprocessed=2)
+        assert echo["stages"]["rerun_from"] == "postprocess"
+        assert echo["needs_inference"] is False
         assert (echo["to_infer"], echo["to_grade"]) == (0, 2)
 
     def test_resume_from_another_dir_copies_nothing(
@@ -510,9 +524,11 @@ def test_a_fresh_run_infers_everything(
             "recorded": None,
             "samples": 3,
             "resume_plan": None,
+            "stages": None,
             "new": 3,
             "to_infer": 3,
             "to_grade": 3,
+            "needs_inference": True,
         }
         assert not (tmp_path / "results").exists()
 
@@ -531,3 +547,112 @@ def test_a_finished_run_is_reused(
         assert echo["start"] == "reuse"
         assert echo["resume_plan"] == _resume_plan(kept=3)
         assert out["data"]["needs_inference"] is False
+
+
+def _patch_report(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Replace ``EchoTask.report`` from a package on disk, as a plugin patch does."""
+    patches = tmp_path / "patches"
+    (patches / "echo_report_patch").mkdir(parents=True, exist_ok=True)
+    (patches / "echo_report_patch" / "__init__.py").write_text(
+        textwrap.dedent(
+            """
+            async def report(self, finals, fails):
+                correct = sum(1 for f in finals if f.feedback_result["correct"])
+                return {"correct": correct, "total": len(finals) + len(fails),
+                        "patched": True}
+            """
+        )
+    )
+    monkeypatch.syspath_prepend(str(patches))
+    monkeypatch.delitem(sys.modules, "echo_report_patch", raising=False)
+    patch_module = importlib.import_module("echo_report_patch")
+    monkeypatch.setattr(EchoTask, "report", patch_module.report)
+
+
+class TestDryRunStagePlan:
+    """The dry-run's per-task stage plan, and whether the run then infers."""
+
+    H100_TEXT = TestDryRunReportsWhatARunWouldDo.H100_TEXT
+
+    def test_a_report_only_change_rereports_and_infers_nothing(
+        self, tmp_path: Path, engine: Engine, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        with _gpu(H100):
+            _sync_run(_write(tmp_path, QUESTIONS[:2]))
+        engine.launches.clear()
+        engine.requests.clear()
+        _patch_report(tmp_path, monkeypatch)
+        monkeypatch.setenv("SIEVAL_TARGET_GPU", self.H100_TEXT)
+        config = _write(tmp_path, QUESTIONS[:2])
+
+        with _gpu(None):
+            code, out = _dry_run(config, "--resume")
+
+        assert code == 0, out
+        data = out["data"]
+        echo = data["tasks"]["echo"]
+        assert echo["start"] == "reuse"
+        assert echo["resume_plan"] == _resume_plan(kept=2, rereported=2)
+        assert (echo["to_infer"], echo["needs_inference"]) == (0, False)
+        assert echo["stages"]["rerun_from"] == "report"
+        assert echo["stages"]["stages"]["report"] == "differs"
+        assert echo["stages"]["stages"]["feedback"] == "same"
+        assert data["needs_inference"] is False
+
+        # The run re-reports on a node with no GPU: no engine, no request.
+        with _gpu(None):
+            report = _sync_run(config, resume=True)
+        assert engine.launches == [] and engine.requests == []
+        assert report["echo"]["patched"] is True
+        assert _meta(tmp_path)["runs"][-1]["resume"] == _resume_plan(
+            kept=2, rereported=2
+        )
+
+    @pytest.mark.parametrize(
+        ("change", "rerun_from", "outcomes"),
+        [
+            ("none", None, {"kept": 2}),
+            ("report", "report", {"kept": 2, "rereported": 2}),
+            ("grader", "postprocess", {"repostprocessed": 2}),
+            ("new_sample", None, {"kept": 2}),
+            ("fewer_samples", None, {"kept": 1, "dropped": 1}),
+        ],
+    )
+    def test_needs_inference_agrees_with_what_the_run_launches(
+        self,
+        tmp_path: Path,
+        engine: Engine,
+        monkeypatch: pytest.MonkeyPatch,
+        change: str,
+        rerun_from: str | None,
+        outcomes: dict[str, int],
+    ) -> None:
+        with _gpu(H100):
+            _sync_run(_write(tmp_path, QUESTIONS[:2]))
+        engine.launches.clear()
+        engine.requests.clear()
+        questions = {"new_sample": QUESTIONS, "fewer_samples": QUESTIONS[:1]}.get(
+            change, QUESTIONS[:2]
+        )
+        if change == "report":
+            _patch_report(tmp_path, monkeypatch)
+        config = _write(tmp_path, questions, strict=change == "grader")
+
+        monkeypatch.setenv("SIEVAL_TARGET_GPU", self.H100_TEXT)
+        with _gpu(None):
+            code, out = _dry_run(config, "--resume")
+        monkeypatch.delenv("SIEVAL_TARGET_GPU")
+        assert code == 0, out
+        data = out["data"]
+        echo = data["tasks"]["echo"]
+        assert echo["resume_plan"] == _resume_plan(**outcomes)
+        assert echo["stages"]["rerun_from"] == rerun_from
+
+        with _gpu(H100):
+            _sync_run(config, resume=True)
+
+        assert data["needs_inference"] is (engine.launches == ["subject"])
+        assert data["needs_inference"] is bool(engine.launches)
+        assert echo["needs_inference"] is bool(engine.requests)
+        assert len(engine.requests) == echo["to_infer"]
+        assert _meta(tmp_path)["runs"][-1]["resume"] == echo["resume_plan"]

tests/unit/cli/test_output.py

@@ -431,6 +431,7 @@ def test_each_model_and_task_gets_a_line(self) -> None:
                 "echo": {
                     "start": "resume",
                     "resume_plan": _outcomes(kept=2, regraded=1),
+                    "stages": {"rerun_from": "feedback"},
                     "to_infer": 1,
                     "to_grade": 2,
                 },
@@ -445,7 +446,10 @@ def test_each_model_and_task_gets_a_line(self) -> None:
         )
         assert "model subject: engine starts" in lines
         assert "model judge: engine not started" in lines
-        assert "task echo: resume (2 kept, 1 regraded), 1 to infer, 2 to grade" in lines
+        assert (
+            "task echo: resume (2 kept, 1 regraded), from feedback, 1 to infer, "
+            "2 to grade"
+        ) in lines
         assert "task fresh: fresh, 3 to infer, 3 to grade" in lines
         assert "needs inference: yes" in lines
 

tests/unit/cli/test_run.py

@@ -356,6 +356,7 @@ def _plan(to_infer: int) -> dict[str, object]:
             "recorded": "r/t",
             "samples": 2,
             "outcomes": {"kept": 2 - to_infer},
+            "stages": {"rerun_from": None},
             "new": to_infer,
             "to_infer": to_infer,
             "to_grade": to_infer,
@@ -378,6 +379,10 @@ async def test_needs_inference_is_whether_an_engine_starts(
         assert result["needs_inference"] is False
         tasks = cast(dict, result["tasks"])
         assert tasks["t_api"]["to_infer"] == 2
+        # A task infers through `api` without starting an engine.
+        assert tasks["t_api"]["needs_inference"] is True
+        assert tasks["t_a"]["needs_inference"] is False
+        assert tasks["t_a"]["stages"] == {"rerun_from": None}
 
     @pytest.mark.anyio
     async def test_a_fresh_run_starts_every_served_engine(self, tmp_path: Path) -> None:

tests/unit/core/runners/test_plan_run.py

@@ -79,6 +79,7 @@ async def test_a_fresh_run_infers_every_sample(self, tmp_path: Path) -> None:
             "recorded": None,
             "samples": 3,
             "outcomes": None,
+            "stages": None,
             "new": 3,
             "to_infer": 3,
             "to_grade": 3,