From d0633a6351d0f06ac744e7c51a945a32d7a2372c Mon Sep 17 00:00:00 2001 From: pbean Date: Wed, 22 Jul 2026 11:30:02 -0700 Subject: [PATCH 1/2] refactor(cli): extract resume + sweep composition into runsetup.py (#243) PR 2 of the two-PR plan in #243 (F-2). PR 1 (#262) lifted cmd_run's composition into runsetup; this extends the identical injection seam to the two remaining logic-heavy callbacks so each becomes parse -> compose -> render: - cmd_sweep: _start_sweep's inline composition (run dir + state + pid + sweep.json + adapters + SweepEngine) -> runsetup.compose_sweep. - cmd_resume: _resume_paused_run's adapter build + engine selection (sweep/stories/plain, chosen from persisted run state) -> compose_resume. The engine classes and _make_adapters are injected from cli's own namespace so the test suite's monkeypatch.setattr(cli, ...) seams still bite. Resume keeps its policy-snapshot re-stamp + pid/pause/save bookkeeping CLI-side because that ordering is load-bearing; compose_resume receives the already-persisted state. Dropped the now-orphaned 'from .runs import RUNS_DIR' (its only use moved to runsetup). Thin render-only callbacks left untouched per the assessment's anti-goals. Behavior-preserving; argv surface unchanged; no CHANGELOG. --- src/bmad_loop/cli.py | 111 ++++++++------------------ src/bmad_loop/runsetup.py | 159 +++++++++++++++++++++++++++++++++++++- 2 files changed, 187 insertions(+), 83 deletions(-) diff --git a/src/bmad_loop/cli.py b/src/bmad_loop/cli.py index c5567dc7f..1d0eaede3 100644 --- a/src/bmad_loop/cli.py +++ b/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- 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 diff --git a/src/bmad_loop/runsetup.py b/src/bmad_loop/runsetup.py index 2fde23644..259804989 100644 --- a/src/bmad_loop/runsetup.py +++ b/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 @@ -305,11 +312,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 +393,145 @@ 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, + } + (run_dir / "sweep.json").write_text(json.dumps(options, indent=2), encoding="utf-8") + 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- 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" + opts = json.loads(opts_path.read_text(encoding="utf-8")) if opts_path.is_file() else {} + 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 + ) From 8d28322fc839850845fc9d4b17acb6d0eae3d74c Mon Sep 17 00:00:00 2001 From: pbean Date: Wed, 22 Jul 2026 12:33:03 -0700 Subject: [PATCH 2/2] fix(runsetup): harden sweep.json read + atomic write on resume recovery (#243) Review (CodeRabbit) surfaced a pre-existing latent gap carried over by the composition move: a torn/corrupt sweep.json from a crash mid-write would abort resume -- the recovery path itself -- because the read only guarded the missing-file case. - compose_sweep writes sweep.json atomically (tmp + atomic_replace), matching save_state's state.json persistence, so the corruption can't arise in the first place. - compose_resume guards the read (except (OSError, json.JSONDecodeError): {}), falling back to the same launch defaults as the missing-file arm -- mirroring tui.data's tolerant run-dir reads. - Regression test: a paused sweep run with a corrupt sweep.json resumes with default options instead of crashing. The one deliberate behavior change vs a pure move; affects only the corrupt-file edge case. No CHANGELOG. --- src/bmad_loop/runsetup.py | 17 +++++++++++++++-- tests/test_cli.py | 22 ++++++++++++++++++++++ 2 files changed, 37 insertions(+), 2 deletions(-) diff --git a/src/bmad_loop/runsetup.py b/src/bmad_loop/runsetup.py index 259804989..249c37d28 100644 --- a/src/bmad_loop/runsetup.py +++ b/src/bmad_loop/runsetup.py @@ -41,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: @@ -439,7 +440,13 @@ def compose_sweep( "max_cycles": max_cycles, "trigger": trigger, } - (run_dir / "sweep.json").write_text(json.dumps(options, indent=2), encoding="utf-8") + # 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( @@ -493,7 +500,13 @@ def compose_resume( adapters = make_adapters(project, run_dir, policy) 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 {} + 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, diff --git a/tests/test_cli.py b/tests/test_cli.py index 59cdf33ac..317e33130 100644 --- a/tests/test_cli.py +++ b/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