"""Tests for task add-blast-radius-scheduler. Covers R1-R6 from tasks/add-blast-radius-scheduler/SPEC.md. Unit tests stub subprocess.run; integration tests use a real git repo on tmp_path. """ import json import subprocess import sys import importlib.util from pathlib import Path from typing import Optional import pytest _RUNNER_PATH = Path.home() / ".automaton" / "scripts" / "loop-runner.py" _spec = importlib.util.spec_from_file_location("loop_runner_br", _RUNNER_PATH) lr = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(lr) def _state(loop_path: Path) -> dict: return json.loads((loop_path / ".state.loop").read_text()) def _write_state(loop_path: Path, state: dict) -> None: (loop_path / ".state.loop").write_text(json.dumps(state, indent=2, sort_keys=True) + "\n") def _make_loop(project: Path, name: str = "br-loop", cfg_overrides: Optional[dict] = None, state_overrides: Optional[dict] = None) -> Path: lp = project / ".automaton" / "loops" / name lp.mkdir(parents=True, exist_ok=True) cfg = { "name": name, "description": "test loop", "schedule": {"interval_seconds": 3600}, "brakes": {"max_iterations": 25, "max_budget_usd": None, "score_plateau_window": 5}, "blast_radius": {"file_scope": [], "use_worktree": True}, "work_source": {"kind": "single"}, "roles": { "implement": {"prompt": "test-impl.md"}, "verify": {"prompt": "test-verify.md"}, "orchestrate": {"prompt": "test-orch.md"}, }, } if cfg_overrides: cfg.update(cfg_overrides) (lp / "loop.json").write_text(json.dumps(cfg, indent=2) + "\n") state = { "schema_version": 1, "name": name, "status": "running", "halt_reason": None, "iteration_count": 0, "resumed_count": 0, "last_tick_at": None, "last_verdict": None, "score_history": [], "current_task": None, "worktree_branch": None, "worktree_path": None, } if state_overrides: state.update(state_overrides) (lp / ".state.loop").write_text(json.dumps(state, indent=2, sort_keys=True) + "\n") (lp / ".state.log").write_text("") roles = cfg.get("roles") or {} if isinstance(roles, dict): for role_cfg in roles.values(): if isinstance(role_cfg, dict) and role_cfg.get("prompt"): prompt_ref = role_cfg["prompt"] if not (Path.home() / ".automaton" / "prompts" / prompt_ref).exists(): try: (lp / prompt_ref).write_text(f"prompt: {prompt_ref}\n") except OSError: pass return lp def _make_task(project: Path, name: str) -> Path: tp = project / ".automaton" / "tasks" / name tp.mkdir(parents=True, exist_ok=True) (tp / ".state").write_text("implement\n") return tp class _FakeSubprocess: def __init__(self): self.rules: list = [] self.invocations: list = [] def add(self, needle, handler): self.rules.append((needle, handler)) def add_simple(self, needle, stdout="", rc=0): def handler(argv): class R: pass r = R() r.stdout = stdout r.stderr = "" r.returncode = rc return r self.rules.append((needle, handler)) def run(self, argv, *args, **kwargs): self.invocations.append(list(argv)) for needle, handler in self.rules: if any(needle in str(a) for a in argv): return handler(argv) class R: pass r = R() r.stdout = "" r.stderr = "" r.returncode = 0 return r @pytest.fixture def tmp_project(tmp_path): (tmp_path / ".automaton" / "tasks").mkdir(parents=True) return tmp_path @pytest.fixture def fake_run(monkeypatch): fake = _FakeSubprocess() monkeypatch.setattr(subprocess, "run", fake.run) return fake def _gate_ok(fake): fake.add_simple("--check-gate", json.dumps({"ok": True})) def _ctx_ok(fake): fake.add_simple("--loop-mode", json.dumps({"loop_mode_eligible": True})) def _verdict_stdout(p=True, score=0.9, hint=""): body = {"pass": p, "score": score} if hint: body["next_hint"] = hint return json.dumps(body) def _tick_args(loop_name, project): class A: pass a = A() a.mode = "tick" a.loop = loop_name a.project = str(project) a.json_output = False return a def _run_tick(loop_name, project): return lr.cmd_tick(_tick_args(loop_name, project)) # --------------------------------------------------------------------------- # R1 -- _ensure_worktree basic behavior # --------------------------------------------------------------------------- class TestEnsureWorktree: def test_ensure_worktree_creates_worktree(self, tmp_project, fake_run): _gate_ok(fake_run) _ctx_ok(fake_run) _make_task(tmp_project, "wt-task") lp = _make_loop(tmp_project, state_overrides={"current_task": "wt-task"}) fake_run.add_simple("rev-parse", "true") fake_run.add_simple("worktree add", "", rc=0) fake_run.add_simple("test-impl", _verdict_stdout()) fake_run.add_simple("test-verify", _verdict_stdout()) fake_run.add_simple("test-orch", "") _run_tick("br-loop", tmp_project) st = _state(lp) assert st["worktree_path"] is not None assert st["worktree_path"].endswith("worktree") assert st["worktree_branch"] == "loop/br-loop" def test_ensure_worktree_reuses_existing(self, tmp_project, fake_run): _gate_ok(fake_run) _ctx_ok(fake_run) _make_task(tmp_project, "wt-reuse") existing_wt = tmp_project / ".automaton" / "loops" / "br-loop" / "worktree" existing_wt.mkdir(parents=True) lp = _make_loop(tmp_project, state_overrides={ "current_task": "wt-reuse", "worktree_path": str(existing_wt), "worktree_branch": "loop/br-loop", }) fake_run.add_simple("test-impl", _verdict_stdout()) fake_run.add_simple("test-verify", _verdict_stdout()) fake_run.add_simple("test-orch", "") _run_tick("br-loop", tmp_project) git_calls = [inv for inv in fake_run.invocations if "git" in inv] assert len(git_calls) == 0 def test_ensure_worktree_use_worktree_false_returns_project_root(self, tmp_project, fake_run): _gate_ok(fake_run) _ctx_ok(fake_run) _make_task(tmp_project, "wt-off") lp = _make_loop(tmp_project, cfg_overrides={"blast_radius": {"file_scope": [], "use_worktree": False}}, state_overrides={"current_task": "wt-off"}) fake_run.add_simple("test-impl", _verdict_stdout()) fake_run.add_simple("test-verify", _verdict_stdout()) fake_run.add_simple("test-orch", "") _run_tick("br-loop", tmp_project) st = _state(lp) assert st["worktree_path"] is None git_calls = [inv for inv in fake_run.invocations if "git" in inv] assert len(git_calls) == 0 def test_ensure_worktree_missing_field_defaults_true(self, tmp_project, fake_run): _gate_ok(fake_run) _ctx_ok(fake_run) _make_task(tmp_project, "wt-default") cfg = { "name": "br-loop", "description": "test", "schedule": {"interval_seconds": 3600}, "brakes": {"max_iterations": 25, "max_budget_usd": None, "score_plateau_window": 5}, "blast_radius": {"file_scope": []}, "work_source": {"kind": "single"}, "roles": { "implement": {"prompt": "test-impl.md"}, "verify": {"prompt": "test-verify.md"}, "orchestrate": {"prompt": "test-orch.md"}, }, } lp = tmp_project / ".automaton" / "loops" / "br-loop" lp.mkdir(parents=True, exist_ok=True) (lp / "loop.json").write_text(json.dumps(cfg, indent=2) + "\n") (lp / ".state.loop").write_text(json.dumps({ "schema_version": 1, "name": "br-loop", "status": "running", "halt_reason": None, "iteration_count": 0, "resumed_count": 0, "last_tick_at": None, "last_verdict": None, "score_history": [], "current_task": "wt-default", "worktree_branch": None, "worktree_path": None, }, indent=2, sort_keys=True) + "\n") (lp / ".state.log").write_text("") fake_run.add_simple("rev-parse", "true") fake_run.add_simple("worktree add", "", rc=0) fake_run.add_simple("test-impl", _verdict_stdout()) fake_run.add_simple("test-verify", _verdict_stdout()) fake_run.add_simple("test-orch", "") _run_tick("br-loop", tmp_project) st = _state(lp) assert st["worktree_path"] is not None # --------------------------------------------------------------------------- # R2 -- graceful degradation # --------------------------------------------------------------------------- class TestGracefulDegradation: def test_falls_back_when_not_git_repo(self, tmp_project, fake_run): _gate_ok(fake_run) _ctx_ok(fake_run) _make_task(tmp_project, "no-git") lp = _make_loop(tmp_project, state_overrides={"current_task": "no-git"}) fake_run.add_simple("rev-parse", "", rc=128) fake_run.add_simple("test-impl", _verdict_stdout()) fake_run.add_simple("test-verify", _verdict_stdout()) fake_run.add_simple("test-orch", "") _run_tick("br-loop", tmp_project) st = _state(lp) assert st["worktree_path"] is None log = (lp / ".state.log").read_text() assert "not a git repo" in log def test_falls_back_when_git_missing(self, tmp_project, fake_run): _gate_ok(fake_run) _ctx_ok(fake_run) _make_task(tmp_project, "no-git-bin") def git_not_found(argv, *a, **kw): class R: pass r = R() r.stdout = "" r.stderr = "" r.returncode = 0 if "git" in argv: raise FileNotFoundError("git not found") return r monkeypatch_fn = fake_run.run original_run = subprocess.run class CombinedFake: def run(self, argv, *a, **kw): fake_run.invocations.append(list(argv)) if "git" in argv: raise FileNotFoundError("git not found") for needle, handler in fake_run.rules: if any(needle in str(x) for x in argv): return handler(argv) class R: pass r = R() r.stdout = "" r.stderr = "" r.returncode = 0 return r subprocess.run = CombinedFake().run lp = _make_loop(tmp_project, state_overrides={"current_task": "no-git-bin"}) try: fake_run.add_simple("test-impl", _verdict_stdout()) fake_run.add_simple("test-verify", _verdict_stdout()) fake_run.add_simple("test-orch", "") _run_tick("br-loop", tmp_project) finally: subprocess.run = original_run st = _state(lp) assert st["worktree_path"] is None def test_falls_back_when_worktree_add_fails(self, tmp_project, fake_run): _gate_ok(fake_run) _ctx_ok(fake_run) _make_task(tmp_project, "wt-fail") lp = _make_loop(tmp_project, state_overrides={"current_task": "wt-fail"}) fake_run.add_simple("rev-parse", "true") def fail_add(argv): class R: pass r = R() r.stdout = "" r.stderr = "worktree add failed" r.returncode = 1 return r fake_run.add("worktree", fail_add) fake_run.add_simple("test-impl", _verdict_stdout()) fake_run.add_simple("test-verify", _verdict_stdout()) fake_run.add_simple("test-orch", "") _run_tick("br-loop", tmp_project) st = _state(lp) assert st["worktree_path"] is None log = (lp / ".state.log").read_text() assert "worktree add failed" in log or "worktree" in log # --------------------------------------------------------------------------- # R4 -- branch already exists # --------------------------------------------------------------------------- class TestBranchExists: def test_reuses_existing_branch(self, tmp_project, fake_run): _gate_ok(fake_run) _ctx_ok(fake_run) _make_task(tmp_project, "wt-branch") lp = _make_loop(tmp_project, state_overrides={"current_task": "wt-branch"}) fake_run.add_simple("rev-parse", "true") call_count = {"n": 0} def add_handler(argv): call_count["n"] += 1 class R: pass r = R() if "-b" in argv: r.stdout = "" r.stderr = "fatal: a branch named 'loop/br-loop' already exists" r.returncode = 128 else: r.stdout = "" r.stderr = "" r.returncode = 0 return r fake_run.add_simple("test-impl", _verdict_stdout()) fake_run.add_simple("test-verify", _verdict_stdout()) fake_run.add_simple("test-orch", "") fake_run.add("worktree", add_handler) _run_tick("br-loop", tmp_project) st = _state(lp) assert st["worktree_path"] is not None assert call_count["n"] == 2 # --------------------------------------------------------------------------- # R5 -- state consistency # --------------------------------------------------------------------------- class TestStateConsistency: def test_clears_stale_worktree_path(self, tmp_project, fake_run): _gate_ok(fake_run) _ctx_ok(fake_run) _make_task(tmp_project, "wt-stale") lp = _make_loop(tmp_project, state_overrides={ "current_task": "wt-stale", "worktree_path": "/nonexistent/path/worktree", "worktree_branch": "loop/br-loop", }) fake_run.add_simple("rev-parse", "true") fake_run.add_simple("worktree add", "", rc=0) fake_run.add_simple("test-impl", _verdict_stdout()) fake_run.add_simple("test-verify", _verdict_stdout()) fake_run.add_simple("test-orch", "") _run_tick("br-loop", tmp_project) st = _state(lp) assert st["worktree_path"] != "/nonexistent/path/worktree" assert st["worktree_path"] is not None def test_recreates_after_deletion(self, tmp_project, fake_run): _gate_ok(fake_run) _ctx_ok(fake_run) _make_task(tmp_project, "wt-recreate") lp = _make_loop(tmp_project, state_overrides={ "current_task": "wt-recreate", "worktree_path": str(tmp_project / ".automaton" / "loops" / "br-loop" / "old-wt"), "worktree_branch": "loop/br-loop", }) fake_run.add_simple("rev-parse", "true") fake_run.add_simple("worktree add", "", rc=0) fake_run.add_simple("test-impl", _verdict_stdout()) fake_run.add_simple("test-verify", _verdict_stdout()) fake_run.add_simple("test-orch", "") _run_tick("br-loop", tmp_project) st = _state(lp) assert st["worktree_path"].endswith("worktree") # --------------------------------------------------------------------------- # R3 -- tick integration # --------------------------------------------------------------------------- class TestTickIntegration: def test_tick_creates_worktree_on_first_tick(self, tmp_project, fake_run): _gate_ok(fake_run) _ctx_ok(fake_run) _make_task(tmp_project, "wt-first") lp = _make_loop(tmp_project, state_overrides={"current_task": "wt-first"}) fake_run.add_simple("rev-parse", "true") fake_run.add_simple("worktree add", "", rc=0) fake_run.add_simple("test-impl", _verdict_stdout()) fake_run.add_simple("test-verify", _verdict_stdout()) fake_run.add_simple("test-orch", "") summary = _run_tick("br-loop", tmp_project) assert summary["skipped"] is False assert summary["iter"] == 1 st = _state(lp) assert st["worktree_path"] is not None def test_tick_reuses_worktree_on_second_tick(self, tmp_project, fake_run): _gate_ok(fake_run) _ctx_ok(fake_run) _make_task(tmp_project, "wt-second") existing_wt = tmp_project / ".automaton" / "loops" / "br-loop" / "worktree" existing_wt.mkdir(parents=True) lp = _make_loop(tmp_project, state_overrides={ "current_task": "wt-second", "worktree_path": str(existing_wt), "worktree_branch": "loop/br-loop", }) fake_run.add_simple("test-impl", _verdict_stdout()) fake_run.add_simple("test-verify", _verdict_stdout()) fake_run.add_simple("test-orch", "") _run_tick("br-loop", tmp_project) git_calls = [inv for inv in fake_run.invocations if "git" in inv] assert len(git_calls) == 0 def test_tick_falls_back_to_project_root_when_no_git(self, tmp_project, fake_run): _gate_ok(fake_run) _ctx_ok(fake_run) _make_task(tmp_project, "wt-nogit") lp = _make_loop(tmp_project, state_overrides={"current_task": "wt-nogit"}) fake_run.add_simple("rev-parse", "", rc=128) fake_run.add_simple("test-impl", _verdict_stdout()) fake_run.add_simple("test-verify", _verdict_stdout()) fake_run.add_simple("test-orch", "") summary = _run_tick("br-loop", tmp_project) assert summary["skipped"] is False st = _state(lp) assert st["worktree_path"] is None # --------------------------------------------------------------------------- # R6 -- platform path handling # --------------------------------------------------------------------------- class TestPlatformPaths: def test_worktree_path_uses_pathlib(self, tmp_project, fake_run): _gate_ok(fake_run) _ctx_ok(fake_run) _make_task(tmp_project, "wt-path") lp = _make_loop(tmp_project, state_overrides={"current_task": "wt-path"}) captured = {} def capture_git(argv, *a, **kw): if "worktree" in argv and "add" in argv: captured["worktree_argv"] = list(argv) class R: pass r = R() if "rev-parse" in argv: r.stdout = "true" else: r.stdout = "" r.stderr = "" r.returncode = 0 return r fake_run.add("rev-parse", capture_git) fake_run.add("worktree", capture_git) fake_run.add_simple("test-impl", _verdict_stdout()) fake_run.add_simple("test-verify", _verdict_stdout()) fake_run.add_simple("test-orch", "") _run_tick("br-loop", tmp_project) assert "worktree_argv" in captured wt_path_arg = next(a for a in captured["worktree_argv"] if "worktree" in a and a != "worktree") assert "/" in wt_path_arg or "\\" in wt_path_arg # --------------------------------------------------------------------------- # Regression -- existing loop with worktree_path ticks unchanged # --------------------------------------------------------------------------- class TestRegression: def test_existing_loop_with_worktree_path_ticks_unchanged(self, tmp_project, fake_run): _gate_ok(fake_run) _ctx_ok(fake_run) _make_task(tmp_project, "wt-reg") existing_wt = tmp_project / ".automaton" / "loops" / "br-loop" / "worktree" existing_wt.mkdir(parents=True) lp = _make_loop(tmp_project, state_overrides={ "current_task": "wt-reg", "worktree_path": str(existing_wt), "worktree_branch": "loop/br-loop", }) fake_run.add_simple("test-impl", _verdict_stdout(True, 0.9, "n")) fake_run.add_simple("test-verify", _verdict_stdout(True, 0.9, "n")) fake_run.add_simple("test-orch", "") summary = _run_tick("br-loop", tmp_project) assert summary["skipped"] is False assert summary["iter"] == 1 assert summary["verdict"]["pass"] is True st = _state(lp) assert st["worktree_path"] == str(existing_wt)