refactor(cli): extract resume + sweep composition into runsetup.py (#243)

src/bmad_loop/cli.py

@@ -65,7 +65,6 @@
 from .model import RunState
 from .platform_util import MAX_SEGMENT
 from .process_host import ProcessHostError
-from .runs import RUNS_DIR
 
 # The run-composition helpers now live in runsetup.py (the library layer a non-CLI
 # frontend imports). They are re-exported under their historical private names —
@@ -789,46 +788,26 @@ def _start_sweep(
     trigger: str,
     run_id: str | None = None,
 ) -> int:
-    run_id = run_id or runs.new_run_id()
-    run_dir = project / RUNS_DIR / run_id
-    journal = Journal(run_dir)
-    state = RunState(
-        run_id=run_id,
-        project=str(project),
-        started_at=time.strftime("%Y-%m-%dT%H:%M:%S"),
-        policy_snapshot=pol.to_dict(),
-        run_type="sweep",
-    )
-    save_state(run_dir, state)
-    runs.write_pid(run_dir)
-    options = {
-        "prompting": prompting,
-        "decisions_only": decisions_only,
-        "max_bundles": max_bundles,
-        "repeat": repeat,
-        "max_cycles": max_cycles,
-        "trigger": trigger,
-    }
-    (run_dir / "sweep.json").write_text(json.dumps(options, indent=2), encoding="utf-8")
-    adapters = _make_adapters(project, run_dir, pol)
-    journal.append("run-start", run_id=run_id, run_type="sweep", trigger=trigger)
-    print(f"sweep {run_id} starting (attach: bmad-loop attach)")
-    engine = SweepEngine(
+    # The composition (run dir + state + pid + sweep.json + adapters + engine)
+    # lives in runsetup; this stays compose -> render. SweepEngine and
+    # _make_adapters are handed in from this module's namespace so the test suite's
+    # `monkeypatch.setattr(cli, "SweepEngine"/"_make_adapters", ...)` still applies.
+    composed = runsetup.compose_sweep(
+        project=project,
         paths=paths,
         policy=pol,
-        adapter=adapters["dev"],
-        review_adapter=adapters["review"],
-        triage_adapter=adapters["triage"],
-        run_dir=run_dir,
-        journal=journal,
-        state=state,
+        run_id=run_id,
         prompting=prompting,
         decisions_only=decisions_only,
         max_bundles=max_bundles,
         repeat=repeat,
         max_cycles=max_cycles,
+        trigger=trigger,
+        make_adapters=_make_adapters,
+        sweep_engine_cls=SweepEngine,
     )
-    summary = engine.run()
+    print(f"sweep {composed.run_id} starting (attach: bmad-loop attach)")
+    summary = composed.engine.run()
     print(summary.render())
     return 0
 
@@ -975,52 +954,26 @@ def _resume_paused_run(project: Path, run_dir: Path) -> int:
     # runs FIRST so no observer catches a window of "not paused + dead pid",
     # which tui.data classifies as INTERRUPTED.
     save_state(run_dir, state)
-    # drop any stale agent session so the run spins up a fresh one (a stopped or
-    # interrupted run can leave a lingering bmad-loop-<id> session behind).
-    runs.kill_session(run_dir.name)
-    adapters = _make_adapters(project, run_dir, pol)
-    if state.run_type == "sweep":
-        opts_path = run_dir / "sweep.json"
-        opts = json.loads(opts_path.read_text(encoding="utf-8")) if opts_path.is_file() else {}
-        engine: Engine = SweepEngine(
-            paths=paths,
-            policy=pol,
-            adapter=adapters["dev"],
-            review_adapter=adapters["review"],
-            triage_adapter=adapters["triage"],
-            run_dir=run_dir,
-            journal=journal,
-            state=state,
-            prompting=bool(opts.get("prompting", False)),
-            decisions_only=bool(opts.get("decisions_only", False)),
-            max_bundles=opts.get("max_bundles"),
-            repeat=opts.get("repeat"),
-            max_cycles=opts.get("max_cycles"),
-        )
-    else:
-        story_common = dict(
-            paths=paths,
-            policy=pol,
-            adapter=adapters["dev"],
-            review_adapter=adapters["review"],
-            run_dir=run_dir,
-            journal=journal,
-            state=state,
-            # restore the launching scope + cap so a resumed `--epic N` run keeps
-            # picking within N instead of silently widening to every epic.
-            epic_filter=state.epic_filter,
-            story_filter=state.story_filter,
-            max_stories=state.max_stories,
-            sweep_factory=_sweep_factory(project, paths),
-        )
-        # stories mode is pinned in run state at launch, so resume rebuilds the
-        # same picker (StoriesEngine) without any flag.
-        engine = (
-            StoriesEngine(**story_common, spec_folder=state.spec_folder)
-            if state.source == "stories"
-            else Engine(**story_common)
-        )
-    summary = engine.run()
+    # The adapter build + engine selection (sweep vs stories vs plain, from
+    # persisted state) lives in runsetup; the re-stamp/pid/save bookkeeping above
+    # stays here because its ordering is load-bearing. Engine/StoriesEngine/
+    # SweepEngine and _make_adapters are handed in from this module's namespace so
+    # the test suite's `monkeypatch.setattr(cli, "SweepEngine"/"Engine"/..., ...)`
+    # still applies.
+    composed = runsetup.compose_resume(
+        project=project,
+        paths=paths,
+        run_dir=run_dir,
+        state=state,
+        policy=pol,
+        journal=journal,
+        sweep_factory=_sweep_factory(project, paths),
+        make_adapters=_make_adapters,
+        engine_cls=Engine,
+        stories_engine_cls=StoriesEngine,
+        sweep_engine_cls=SweepEngine,
+    )
+    summary = composed.engine.run()
     print(summary.render())
     return 0
 

src/bmad_loop/runsetup.py

@@ -9,7 +9,13 @@
 * :func:`make_adapters` — the per-role adapter factory.
 * :func:`platform_preflight` — the multiplexer/process-host readiness probe
   ``cmd_validate`` reports.
-* :func:`build_run_state` / :func:`compose_run` — the RunState + Engine wiring.
+* :func:`build_run_state` / :func:`compose_run` — the RunState + Engine wiring
+  for ``cmd_run``.
+* :func:`compose_sweep` — the same wiring for a ``sweep`` run (``cmd_sweep`` and
+  the auto-triggered child-sweep factory).
+* :func:`compose_resume` — rebuilds the engine for a paused/interrupted run
+  (``cmd_resume`` and ``resolve``'s re-arm), selecting the sweep/stories/plain
+  variant from persisted run state.
 
 The engine class and the adapter factory are *injected* into :func:`compose_run`
 rather than referenced here directly: ``cli`` resolves ``Engine`` /
@@ -22,6 +28,7 @@
 
 from __future__ import annotations
 
+import json
 import sys
 import time
 from dataclasses import dataclass
@@ -34,6 +41,7 @@
 from .checks import Finding
 from .journal import Journal, save_state
 from .model import RunState
+from .platform_util import atomic_replace
 from .runs import RUNS_DIR
 
 if TYPE_CHECKING:
@@ -305,11 +313,13 @@ def build_run_state(
 
 @dataclass
 class ComposedRun:
-    """The composed-but-not-yet-run artifacts ``cmd_run`` renders from.
+    """The composed-but-not-yet-run artifacts a ``compose_*`` returns for its
+    callback to render from — shared by :func:`compose_run`, :func:`compose_sweep`,
+    and :func:`compose_resume`.
 
     ``engine`` is ready to :meth:`run`; ``run_id`` names the run for the attach
-    hint. ``run_dir`` / ``state`` / ``journal`` are the persisted context (already
-    written to disk) a caller other than ``cmd_run`` can inspect."""
+    hint. ``run_dir`` / ``state`` / ``journal`` are the persisted context a caller
+    other than the CLI can inspect."""
 
     engine: Engine
     run_id: str
@@ -384,3 +394,157 @@ def compose_run(
         else engine_cls(**common)
     )
     return ComposedRun(engine=engine, run_id=run_id, run_dir=run_dir, state=state, journal=journal)
+
+
+def compose_sweep(
+    *,
+    project: Path,
+    paths: bmadconfig.ProjectPaths,
+    policy: Policy,
+    run_id: str | None,
+    prompting: bool,
+    decisions_only: bool,
+    max_bundles: int | None,
+    repeat: bool | None,
+    max_cycles: int | None,
+    trigger: str,
+    make_adapters: Callable[[Path, Path, Policy], dict[str, CodingCLIAdapter]],
+    sweep_engine_cls: type[Engine],
+) -> ComposedRun:
+    """Stand up a sweep run: allocate the run dir, persist state + pid, record the
+    sweep options, build the adapters, and wire the ``SweepEngine`` — everything
+    ``cli._start_sweep`` did inline before ``engine.run()``.
+
+    ``sweep.json`` freezes the launch options so a resume rebuilds the same sweep
+    (see :func:`compose_resume`). ``make_adapters`` and ``sweep_engine_cls`` are
+    injected so ``cli`` supplies its own module-level names — keeping the test
+    suite's ``monkeypatch.setattr(cli, "SweepEngine"/"_make_adapters", ...)``
+    effective."""
+    run_id = run_id or runs.new_run_id()
+    run_dir = project / RUNS_DIR / run_id
+    journal = Journal(run_dir)
+    state = RunState(
+        run_id=run_id,
+        project=str(project),
+        started_at=time.strftime("%Y-%m-%dT%H:%M:%S"),
+        policy_snapshot=policy.to_dict(),
+        run_type="sweep",
+    )
+    save_state(run_dir, state)
+    runs.write_pid(run_dir)
+    options = {
+        "prompting": prompting,
+        "decisions_only": decisions_only,
+        "max_bundles": max_bundles,
+        "repeat": repeat,
+        "max_cycles": max_cycles,
+        "trigger": trigger,
+    }
+    # Persist the sweep options atomically (tmp + os.replace), the way save_state
+    # writes state.json: a resume reads this back to rebuild the SweepEngine, so a
+    # crash mid-write must not leave a torn file the recovery path then chokes on.
+    sweep_path = run_dir / "sweep.json"
+    sweep_tmp = sweep_path.with_suffix(".json.tmp")
+    sweep_tmp.write_text(json.dumps(options, indent=2), encoding="utf-8")
+    atomic_replace(sweep_tmp, sweep_path)
+    adapters = make_adapters(project, run_dir, policy)
+    journal.append("run-start", run_id=run_id, run_type="sweep", trigger=trigger)
+    engine: Engine = sweep_engine_cls(
+        paths=paths,
+        policy=policy,
+        adapter=adapters["dev"],
+        review_adapter=adapters["review"],
+        triage_adapter=adapters["triage"],
+        run_dir=run_dir,
+        journal=journal,
+        state=state,
+        prompting=prompting,
+        decisions_only=decisions_only,
+        max_bundles=max_bundles,
+        repeat=repeat,
+        max_cycles=max_cycles,
+    )
+    return ComposedRun(engine=engine, run_id=run_id, run_dir=run_dir, state=state, journal=journal)
+
+
+def compose_resume(
+    *,
+    project: Path,
+    paths: bmadconfig.ProjectPaths,
+    run_dir: Path,
+    state: RunState,
+    policy: Policy,
+    journal: Journal,
+    sweep_factory: Callable[[str], None],
+    make_adapters: Callable[[Path, Path, Policy], dict[str, CodingCLIAdapter]],
+    engine_cls: type[Engine],
+    stories_engine_cls: type[Engine],
+    sweep_engine_cls: type[Engine],
+) -> ComposedRun:
+    """Rebuild the engine for a paused/interrupted run and return it ready to
+    :meth:`run` — the adapter build + engine selection ``cli._resume_paused_run``
+    did inline.
+
+    ``state`` arrives already re-stamped and persisted by the caller: the resume
+    policy-snapshot reconciliation and the pause/pid/graceful-stop bookkeeping stay
+    CLI-side (their ordering is load-bearing — see ``_resume_paused_run``), so this
+    lifts only the composition. The variant is selected from persisted run state:
+    ``run_type == "sweep"`` rebuilds a ``SweepEngine`` from ``sweep.json``;
+    otherwise ``source`` picks ``StoriesEngine`` vs ``Engine``, restoring the
+    launching scope + cap so a resumed ``--epic N`` run keeps its filter. The engine
+    classes and ``make_adapters`` are injected so ``cli``'s ``monkeypatch.setattr``
+    seams bite."""
+    # drop any stale agent session so the run spins up a fresh one (a stopped or
+    # interrupted run can leave a lingering bmad-loop-<id> session behind).
+    runs.kill_session(run_dir.name)
+    adapters = make_adapters(project, run_dir, policy)
+    if state.run_type == "sweep":
+        opts_path = run_dir / "sweep.json"
+        try:
+            opts = json.loads(opts_path.read_text(encoding="utf-8")) if opts_path.is_file() else {}
+        except (OSError, json.JSONDecodeError):
+            # A torn/corrupt sweep.json (crash mid-write on an older run) must not
+            # abort the recovery path — fall back to the same launch defaults as
+            # the missing-file arm, mirroring tui.data's tolerant run-dir reads.
+            opts = {}
+        engine: Engine = sweep_engine_cls(
+            paths=paths,
+            policy=policy,
+            adapter=adapters["dev"],
+            review_adapter=adapters["review"],
+            triage_adapter=adapters["triage"],
+            run_dir=run_dir,
+            journal=journal,
+            state=state,
+            prompting=bool(opts.get("prompting", False)),
+            decisions_only=bool(opts.get("decisions_only", False)),
+            max_bundles=opts.get("max_bundles"),
+            repeat=opts.get("repeat"),
+            max_cycles=opts.get("max_cycles"),
+        )
+    else:
+        story_common = dict(
+            paths=paths,
+            policy=policy,
+            adapter=adapters["dev"],
+            review_adapter=adapters["review"],
+            run_dir=run_dir,
+            journal=journal,
+            state=state,
+            # restore the launching scope + cap so a resumed `--epic N` run keeps
+            # picking within N instead of silently widening to every epic.
+            epic_filter=state.epic_filter,
+            story_filter=state.story_filter,
+            max_stories=state.max_stories,
+            sweep_factory=sweep_factory,
+        )
+        # stories mode is pinned in run state at launch, so resume rebuilds the
+        # same picker (StoriesEngine) without any flag.
+        engine = (
+            stories_engine_cls(**story_common, spec_folder=state.spec_folder)
+            if state.source == "stories"
+            else engine_cls(**story_common)
+        )
+    return ComposedRun(
+        engine=engine, run_id=run_dir.name, run_dir=run_dir, state=state, journal=journal
+    )

tests/test_cli.py

@@ -2765,6 +2765,28 @@ def test_resume_restamps_policy_snapshot_for_sweep_runs(project, monkeypatch):
     assert load_state(run_dir).cache_read_weight() == 0.5
 
 
+def test_resume_tolerates_a_corrupt_sweep_json(project, monkeypatch):
+    """A torn/corrupt sweep.json (a crash mid-write on an older run) must not abort
+    resume — the recovery path. compose_resume guards the read and falls back to the
+    same launch defaults as the missing-file arm instead of letting json.loads raise."""
+    run_dir = _paused_run_for_resume(project, monkeypatch, run_type="sweep")
+    (run_dir / "sweep.json").write_text("{ not json", encoding="utf-8")
+
+    captured: dict = {}
+
+    class _CapturingSweep(_StubEngine):
+        def __init__(self, **kwargs):
+            captured.update(kwargs)
+
+    monkeypatch.setattr(cli, "SweepEngine", _CapturingSweep)
+
+    # resume does not raise, and the sweep is rebuilt with default options
+    assert cli._resume_paused_run(project.project, run_dir) == 0
+    assert captured["prompting"] is False
+    assert captured["decisions_only"] is False
+    assert captured["max_bundles"] is None
+
+
 def test_resume_stamps_a_legacy_run_with_no_snapshot(project, monkeypatch):
     """A run persisted before policy_snapshot existed carries `{}` and displays at
     the hardcoded 0.1 default. Its first resume stamps it — and must not report