diff --git a/loop/__main__.py b/loop/__main__.py index ca5774f..b635b6c 100644 --- a/loop/__main__.py +++ b/loop/__main__.py @@ -1,11 +1,67 @@ from __future__ import annotations import json +import re import sys from pathlib import Path from .contract import doctor_report +_PROG = "python3 -m loop" + +_COMMANDS = ("scaffold", "doctor", "validate", "verify", "inspect") + +# Read commands operate on an EXISTING contract dir; scaffold CREATES one, so it +# is exempt from the "target must exist" guard. +_READ_COMMANDS = ("doctor", "validate", "verify", "inspect") + +_USAGE = f"usage: {_PROG} " + +_HELP = f"""{_PROG} — validate and inspect a portable repo-OS loop contract. + +{_USAGE} + +commands: + scaffold Write a fresh, doctor-clean loop contract into . + doctor Validate the contract objects (manifest, state, tasks, terminal). + validate Alias for doctor. + verify Alias for doctor — check the contract's state. + inspect Score an existing loop against the prime-directive checklist + (emits a weak/strong verdict and a gap report). + +arguments: + A workspace root or its .loop/ directory. + +options: + -h, --help Show this help and exit. + --version Show the version and exit. +""" + + +def _version() -> str: + """Return the package version. Single source of truth is pyproject.toml. + + Prefer installed metadata (which is itself generated from pyproject); fall + back to reading pyproject.toml at the repo root so `--version` still works + from an uninstalled/editable checkout. + """ + try: + from importlib.metadata import PackageNotFoundError, version + + try: + return version("loop-engineer") + except PackageNotFoundError: + pass + except Exception: # pragma: no cover - importlib.metadata ships on 3.10+ + pass + pyproject = Path(__file__).resolve().parent.parent / "pyproject.toml" + try: + text = pyproject.read_text(encoding="utf-8") + except OSError: # pragma: no cover - repo layout guarantees this file + return "unknown" + match = re.search(r'(?m)^version\s*=\s*"([^"]+)"', text) + return match.group(1) if match else "unknown" + def _print_json(report: dict) -> int: print(json.dumps(report, indent=2)) @@ -14,16 +70,38 @@ def _print_json(report: dict) -> int: def main(argv: list[str] | None = None) -> int: argv = list(sys.argv[1:] if argv is None else argv) - if not argv or argv[0] in {"-h", "--help"}: + + if not argv: + print(_USAGE, file=sys.stderr) + return 2 + if argv[0] in {"-h", "--help"}: + print(_HELP) + return 0 + if argv[0] == "--version": + print(_version()) + return 0 + + command = argv.pop(0) + if command not in _COMMANDS: + print(f"unknown loop command: {command}", file=sys.stderr) + print(_USAGE, file=sys.stderr) + return 2 + + if not argv: + print(f"{command}: missing target argument", file=sys.stderr) + print(_USAGE, file=sys.stderr) + return 2 + target = Path(argv[0]) + + if command in _READ_COMMANDS and not target.exists(): print( - "usage: python -m loop ", + f"{command}: target path does not exist: {target}\n" + f" pass an existing workspace root or its .loop/ directory " + f"(run `{_PROG} scaffold {target}` to create a new contract).", file=sys.stderr, ) return 2 - command = argv.pop(0) - target = Path(argv[0]) if argv else Path.cwd() - if command == "scaffold": from .scaffold import scaffold @@ -38,19 +116,16 @@ def main(argv: list[str] | None = None) -> int: if command in {"doctor", "validate", "verify"}: return _print_json(doctor_report(target)) - if command == "inspect": - # Keep the historical inspector script as the scoring UI over the same - # contract artifacts; import lazily to avoid making scripts/ a package. - scripts_dir = Path(__file__).resolve().parent.parent / "scripts" - sys.path.insert(0, str(scripts_dir)) - import inspect_loop # type: ignore + # command == "inspect": keep the historical inspector script as the scoring + # UI over the same contract artifacts; import lazily to avoid making + # scripts/ a package. + scripts_dir = Path(__file__).resolve().parent.parent / "scripts" + sys.path.insert(0, str(scripts_dir)) + import inspect_loop # type: ignore - report = inspect_loop.inspect_loop(str(target)) - print(json.dumps(report, indent=2)) - return 0 if report.get("verdict") != "weak" else 1 - - print(f"unknown loop command: {command}", file=sys.stderr) - return 2 + report = inspect_loop.inspect_loop(str(target)) + print(json.dumps(report, indent=2)) + return 0 if report.get("verdict") != "weak" else 1 if __name__ == "__main__": diff --git a/scripts/anticheat_scan.py b/scripts/anticheat_scan.py index fd6a821..d6705ac 100644 --- a/scripts/anticheat_scan.py +++ b/scripts/anticheat_scan.py @@ -24,8 +24,16 @@ python3 anticheat_scan.py --diff changes.diff --trajectory trace.json ``--trajectory`` is a JSON array of strings (tool calls / paths the loop touched). -Prints findings as JSON. Exit 0 = clean, 1 = downgrade (FailedUnverifiable), -2 = gate tampering (FailedSafety). +Prints findings as JSON. The exit code splits the scan's *outcome* from an +*operational* failure so a broken invocation is never read as a clean scan or as +findings: + + * 0 — clean: the scan ran and found nothing. + * 1 — findings: a downgrade to FailedUnverifiable, or review-only findings. + * 2 — gate tampering: a critical finding -> FailedSafety. + * 3 — operational error: the scan could not run at all (unreadable ``--diff`` / + ``--trajectory`` file, malformed ``--trajectory`` JSON, failed + ``--self-check`` git call, or a bad argument). """ from __future__ import annotations @@ -34,8 +42,17 @@ import json import pathlib import re +import subprocess import sys +# Exit codes. Findings (1, 2) and the scan's severity live apart from an +# operational failure (3), so a broken invocation never masquerades as a clean +# scan (0) or as a downgrade finding (1). +EXIT_CLEAN = 0 +EXIT_FINDINGS = 1 +EXIT_GATE_TAMPERING = 2 +EXIT_OPERATIONAL_ERROR = 3 + # Path markers for a gate file. A bare name (e.g. "self_eval.py") matches by # BASENAME / path-segment, never by raw substring — so a test file whose name # merely CONTAINS a gate name (test_anticheat_scan.py) is not mis-upgraded. @@ -411,8 +428,19 @@ def scan( return {"findings": findings, "clean": not findings, "downgrade_to": downgrade} +class _ArgParser(argparse.ArgumentParser): + """Argparse variant that exits with the operational-error code on a bad + argument, instead of argparse's default 2 (which would collide with the + gate-tampering finding code).""" + + def error(self, message: str): # noqa: D401 - argparse override + self.print_usage(sys.stderr) + print(f"{self.prog}: error: {message}", file=sys.stderr) + raise SystemExit(EXIT_OPERATIONAL_ERROR) + + def main(argv: list[str]) -> int: - ap = argparse.ArgumentParser(description="Anti-cheat trajectory scan.") + ap = _ArgParser(description="Anti-cheat trajectory scan.") ap.add_argument("--diff", help="path to a unified diff file (else read stdin)") ap.add_argument("--files", help="comma-separated changed files (overrides diff parse)") ap.add_argument("--trajectory", help="path to a JSON array of trajectory strings") @@ -423,34 +451,39 @@ def main(argv: list[str]) -> int: ) args = ap.parse_args(argv) - if args.self_check: - import subprocess - - here = pathlib.Path(__file__).resolve().parent - diff_text = subprocess.run( - ["git", "diff", "--", "scripts/anticheat_scan.py", - "scripts/test_anticheat_scan.py"], - cwd=here.parent, capture_output=True, text=True, check=True, - ).stdout - elif args.diff: - with open(args.diff, encoding="utf-8") as fh: - diff_text = fh.read() - elif not sys.stdin.isatty(): - diff_text = sys.stdin.read() - else: - diff_text = "" - - files = args.files.split(",") if args.files else None - trajectory = None - if args.trajectory: - with open(args.trajectory, encoding="utf-8") as fh: - trajectory = json.load(fh) + # Gathering inputs is where a *broken invocation* surfaces (unreadable file, + # malformed JSON, failed git). Fail with the distinct operational code so it + # is never confused with a clean scan or a findings downgrade. + try: + if args.self_check: + here = pathlib.Path(__file__).resolve().parent + diff_text = subprocess.run( + ["git", "diff", "--", "scripts/anticheat_scan.py", + "scripts/test_anticheat_scan.py"], + cwd=here.parent, capture_output=True, text=True, check=True, + ).stdout + elif args.diff: + with open(args.diff, encoding="utf-8") as fh: + diff_text = fh.read() + elif not sys.stdin.isatty(): + diff_text = sys.stdin.read() + else: + diff_text = "" + + files = args.files.split(",") if args.files else None + trajectory = None + if args.trajectory: + with open(args.trajectory, encoding="utf-8") as fh: + trajectory = json.load(fh) + except (OSError, json.JSONDecodeError, subprocess.CalledProcessError) as exc: + print(f"{ap.prog}: operational error: {exc}", file=sys.stderr) + return EXIT_OPERATIONAL_ERROR result = scan(diff_text=diff_text, changed_files=files, trajectory=trajectory) print(json.dumps(result, indent=2)) if result["downgrade_to"] == "FailedSafety": - return 2 - return 0 if result["clean"] else 1 + return EXIT_GATE_TAMPERING + return EXIT_CLEAN if result["clean"] else EXIT_FINDINGS if __name__ == "__main__": diff --git a/scripts/rollout_ledger.py b/scripts/rollout_ledger.py index 82c505a..a4e4295 100644 --- a/scripts/rollout_ledger.py +++ b/scripts/rollout_ledger.py @@ -36,6 +36,10 @@ import sys from pathlib import Path + +def _warn(message: str) -> None: + print(f"rollout_ledger: {message}", file=sys.stderr) + RECORD_FIELDS = ( "id", "parent", @@ -64,21 +68,49 @@ def append(record: dict, path: str | Path) -> dict: return written -def read(path: str | Path) -> list[dict]: - """Return every record in the ledger, in append order. Empty if absent.""" +def _read_with_stats(path: str | Path) -> tuple[list[dict], int]: + """Read the ledger, tolerating corruption. Returns ``(records, malformed)``. + + A ledger is append-only JSONL that may have been truncated mid-write or hand- + edited, so a single bad line must not lose the whole lineage. A line that is + not a JSON object (unparseable, or valid JSON that is not a dict) is skipped + with a stderr warning and counted, never raised. + """ p = Path(path) if not p.exists(): - return [] - records = [] - for line in p.read_text(encoding="utf-8").splitlines(): - if line.strip(): - records.append(json.loads(line)) + return [], 0 + records: list[dict] = [] + malformed = 0 + for lineno, line in enumerate(p.read_text(encoding="utf-8").splitlines(), start=1): + if not line.strip(): + continue + try: + record = json.loads(line) + except json.JSONDecodeError as exc: + malformed += 1 + _warn(f"skipping malformed ledger line {lineno}: {exc}") + continue + if not isinstance(record, dict): + malformed += 1 + _warn(f"skipping malformed ledger line {lineno}: not a JSON object") + continue + records.append(record) + return records, malformed + + +def read(path: str | Path) -> list[dict]: + """Return every valid record in the ledger, in append order. Empty if absent. + + Malformed lines are skipped (with a stderr warning), never fatal. + """ + records, _ = _read_with_stats(path) return records def summarize(path: str | Path) -> dict: - """Compute repair-productivity (productive fraction) and the candidate count.""" - records = read(path) + """Compute repair-productivity (productive fraction), the candidate count, and + the number of malformed lines skipped.""" + records, malformed = _read_with_stats(path) count = len(records) productive = sum(1 for r in records if r.get("productive")) repair_productivity = productive / count if count else 0.0 @@ -86,6 +118,7 @@ def summarize(path: str | Path) -> dict: "count": count, "productive": productive, "repair_productivity": repair_productivity, + "malformed": malformed, } diff --git a/scripts/runtime_monitor.py b/scripts/runtime_monitor.py index a2315fb..80e5a44 100644 --- a/scripts/runtime_monitor.py +++ b/scripts/runtime_monitor.py @@ -287,6 +287,22 @@ def health_report(loop_dir) -> dict: } +def _exit_code(report: dict) -> int: + """Map a health report to an explicit exit code so a script/CI can react + without re-parsing the JSON: + + * 0 — healthy or already terminal: recommendation ``continue``/``done``. + * 1 — intervention recommended (``replan``/``revert``/``approval``) or a + ``degraded`` RUNLOG whose detectors are inert. + * 2 — precondition/operational error (missing or invalid loop state). + """ + if report.get("status") == "error": + return 2 + if report.get("recommendation") in {"continue", "done"}: + return 0 + return 1 + + def main(argv: list[str] | None = None) -> int: argv = list(sys.argv[1:] if argv is None else argv) if not argv: @@ -294,7 +310,7 @@ def main(argv: list[str] | None = None) -> int: return 2 report = health_report(argv[0]) print(json.dumps(report, indent=2)) - return 0 + return _exit_code(report) if __name__ == "__main__": diff --git a/scripts/test_anticheat_scan.py b/scripts/test_anticheat_scan.py index aa74df6..fe63e97 100644 --- a/scripts/test_anticheat_scan.py +++ b/scripts/test_anticheat_scan.py @@ -361,3 +361,57 @@ def test_self_neuter_of_gate_matcher_is_detected(): ) prefixed_sigs = {f["signature"] for f in acs.scan(diff_text=prefixed)["findings"]} assert "scanner_self_edit" in prefixed_sigs + + +# --- M4-CLI item 7: exit-code split (clean / findings / operational error) --- + +import io # noqa: E402 +import json # noqa: E402 + + +def _main(monkeypatch, args, stdin_text=""): + """Run acs.main with a controlled stdin (pytest replaces sys.stdin with an + object whose read() raises, so the scanner's stdin fallback would blow up).""" + monkeypatch.setattr("sys.stdin", io.StringIO(stdin_text)) + return acs.main(args) + + +def test_exit_zero_on_clean_scan(monkeypatch): + rc = _main(monkeypatch, ["--files", "src/widget.py"]) + assert rc == 0 + + +def test_exit_one_on_review_findings(monkeypatch): + # A test-file mutation is a review-flag finding (not clean, not tampering). + rc = _main(monkeypatch, ["--files", "tests/test_foo.py"]) + assert rc == 1 + + +def test_exit_two_on_gate_tampering(monkeypatch): + rc = _main(monkeypatch, ["--files", "scripts/self_eval.py"]) + assert rc == 2 + + +def test_operational_error_on_missing_trajectory_file_is_distinct(monkeypatch): + # A broken invocation must NOT be read as a clean scan (0) or as findings (1). + rc = _main(monkeypatch, ["--trajectory", "/no/such/trajectory.json"]) + assert rc not in (0, 1, 2) + assert rc == 3 + + +def test_operational_error_on_missing_diff_file(monkeypatch): + rc = _main(monkeypatch, ["--diff", "/no/such/diff.txt"]) + assert rc == 3 + + +def test_operational_error_on_malformed_trajectory_json(monkeypatch, tmp_path): + bad = tmp_path / "traj.json" + bad.write_text("{not valid json", encoding="utf-8") + rc = _main(monkeypatch, ["--trajectory", str(bad)]) + assert rc == 3 + + +def test_module_docstring_documents_operational_error_exit_code(): + doc = (acs.__doc__ or "").lower() + assert "operational error" in doc + assert "clean" in doc and "findings" in doc diff --git a/scripts/test_loop_cli.py b/scripts/test_loop_cli.py new file mode 100644 index 0000000..bacaeee --- /dev/null +++ b/scripts/test_loop_cli.py @@ -0,0 +1,149 @@ +"""CLI/UX contract tests for the `python3 -m loop` entry point (M4-CLI / S6). + +Each test pins one behavior from the launch-criterion CLI polish batch. They run +the real entry point as a subprocess so STDOUT vs STDERR and the process exit +code are exercised exactly as a user sees them. +""" + +import json +import re +import subprocess +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent + + +def _run(*args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [sys.executable, "-m", "loop", *args], + cwd=ROOT, + text=True, + capture_output=True, + ) + + +def _pyproject_version() -> str: + text = (ROOT / "pyproject.toml").read_text(encoding="utf-8") + match = re.search(r'(?m)^version\s*=\s*"([^"]+)"', text) + assert match, "pyproject.toml has no version field" + return match.group(1) + + +# --- item 1: --help / -h to STDOUT, exit 0, per-command descriptions --------- + + +def test_help_flag_exits_zero_and_prints_usage_to_stdout(): + result = _run("--help") + assert result.returncode == 0, result.stderr + assert result.stdout.strip(), "help must print to stdout" + assert result.stderr == "", "help must not go to stderr" + assert "usage" in result.stdout.lower() + + +def test_short_help_flag_matches_long_help(): + long = _run("--help") + short = _run("-h") + assert short.returncode == 0 + assert short.stdout == long.stdout + + +def test_help_lists_every_command_with_a_description(): + out = _run("--help").stdout + for command in ("scaffold", "doctor", "validate", "verify", "inspect"): + assert command in out, f"help omits command {command!r}" + # Per-command descriptions, not a bare command list. + assert "Validate" in out or "validate the contract" in out.lower() + assert "Score" in out or "score" in out.lower() + + +# --- item 2: --version prints the package version (single source: pyproject) -- + + +def test_version_flag_prints_package_version_and_exits_zero(): + result = _run("--version") + assert result.returncode == 0, result.stderr + assert result.stdout.strip() == _pyproject_version() + assert result.stderr == "" + + +# --- item 3: usage/help text says `python3 -m loop`, never `python -m loop` --- + + +def test_usage_text_uses_python3_invocation(): + help_out = _run("--help").stdout + assert "python3 -m loop" in help_out + assert "python -m loop" not in help_out + # The bare usage (no args -> stderr) must also use python3. + no_args = _run() + assert "python3 -m loop" in no_args.stderr + assert "python -m loop" not in no_args.stderr + + +# --- item 5: missing target argument -> usage + nonzero (no traceback) -------- + + +def test_missing_target_argument_prints_usage_and_exits_nonzero(): + result = _run("doctor") + assert result.returncode != 0 + assert "usage" in result.stderr.lower() + assert "Traceback" not in result.stderr + + +def test_missing_target_for_inspect_does_not_default_to_cwd(): + result = _run("inspect") + assert result.returncode != 0 + assert "Traceback" not in result.stderr + # Must NOT silently inspect the current directory and emit a JSON report. + assert not result.stdout.strip().startswith("{") + + +# --- item 4: nonexistent target -> distinct actionable error ------------------ + + +def test_nonexistent_target_gives_distinct_actionable_error(tmp_path): + missing = tmp_path / "does-not-exist" + result = _run("doctor", str(missing)) + assert result.returncode != 0 + assert "Traceback" not in result.stderr + # Distinct from a malformed/empty contract: names the missing path on stderr, + # not a JSON `missing_file` issues report on stdout. + assert "does not exist" in result.stderr.lower() + assert str(missing) in result.stderr + assert result.stdout.strip() == "" + + +def test_nonexistent_target_differs_from_malformed_contract(tmp_path): + # An existing-but-empty dir is a malformed contract: JSON issues on stdout, + # exit 1. A nonexistent path is a different failure: message on stderr, and + # the two must not produce the same output. + empty = tmp_path / "empty" + empty.mkdir() + missing = tmp_path / "missing" + + malformed = _run("doctor", str(empty)) + nonexistent = _run("doctor", str(missing)) + + assert malformed.stdout.strip().startswith("{") # JSON issues report + assert nonexistent.stdout.strip() == "" # no JSON report + assert malformed.stdout != nonexistent.stdout + assert malformed.stderr != nonexistent.stderr + + +# --- item 6: `verify` alias is wired and reconciled with docs ----------------- + + +def test_verify_is_a_wired_doctor_alias(tmp_path): + empty = tmp_path / "ws" + empty.mkdir() + doctor = _run("doctor", str(empty)) + verify = _run("verify", str(empty)) + assert verify.returncode == doctor.returncode + assert verify.stdout == doctor.stdout + # And a valid JSON report either way. + json.loads(verify.stdout) + + +def test_help_describes_verify_and_validate_as_aliases(): + out = _run("--help").stdout.lower() + assert "alias" in out, "help must reconcile verify/validate as doctor aliases" diff --git a/scripts/test_rollout_ledger.py b/scripts/test_rollout_ledger.py index 482c34a..4943abf 100644 --- a/scripts/test_rollout_ledger.py +++ b/scripts/test_rollout_ledger.py @@ -101,3 +101,59 @@ def test_summarize_of_empty_ledger_is_zero(tmp_path): # Assert assert summary["count"] == 0 assert summary["repair_productivity"] == 0.0 + + +# --- M4-CLI item 9: malformed ledger lines are tolerated, not fatal ---------- + + +def test_read_skips_malformed_line_without_crashing(tmp_path): + ledger = tmp_path / "rollout.jsonl" + rollout_ledger.append(_candidate(id="cand-1", productive=True), ledger) + with ledger.open("a", encoding="utf-8") as fh: + fh.write("this is not json\n") + rollout_ledger.append(_candidate(id="cand-2", productive=False), ledger) + + # Must not raise on the corrupt middle line. + records = rollout_ledger.read(ledger) + + ids = [r["id"] for r in records] + assert ids == ["cand-1", "cand-2"] + + +def test_read_skips_valid_json_that_is_not_an_object(tmp_path): + ledger = tmp_path / "rollout.jsonl" + with ledger.open("a", encoding="utf-8") as fh: + fh.write("[1, 2, 3]\n") # valid JSON, but not a ledger record object + + records = rollout_ledger.read(ledger) + + assert records == [] + + +def test_read_warns_to_stderr_on_malformed_line(tmp_path, capsys): + ledger = tmp_path / "rollout.jsonl" + rollout_ledger.append(_candidate(id="cand-1"), ledger) + with ledger.open("a", encoding="utf-8") as fh: + fh.write("{broken\n") + + rollout_ledger.read(ledger) + + err = capsys.readouterr().err + assert err.strip(), "a malformed line must warn to stderr" + assert "malformed" in err.lower() or "skip" in err.lower() + + +def test_summarize_counts_malformed_lines(tmp_path): + ledger = tmp_path / "rollout.jsonl" + rollout_ledger.append(_candidate(id="cand-1", productive=True), ledger) + with ledger.open("a", encoding="utf-8") as fh: + fh.write("garbage line\n") + fh.write("also not json {,,}\n") + rollout_ledger.append(_candidate(id="cand-2", productive=True), ledger) + + summary = rollout_ledger.summarize(ledger) + + # Valid candidates are summarized; malformed lines are counted separately. + assert summary["count"] == 2 + assert summary["malformed"] == 2 + assert summary["repair_productivity"] == 1.0 diff --git a/scripts/test_runtime_monitor.py b/scripts/test_runtime_monitor.py index 65216e7..605163d 100644 --- a/scripts/test_runtime_monitor.py +++ b/scripts/test_runtime_monitor.py @@ -125,7 +125,8 @@ def test_cli_emits_json(tmp_path, capsys): # Assert parsed = json.loads(out) assert parsed["stalled"] is True - assert rc == 0 + # A stalled loop recommends intervention (replan) -> nonzero exit (item 8). + assert rc == 1 def _line(it, task, score_text): @@ -318,8 +319,100 @@ def test_documented_cli_by_path_resolves_dotloop_runlog(tmp_path): env=env, ) - assert proc.returncode == 0, proc.stderr + # The loop is stalled, so the CLI exits 1 (intervention) rather than crashing + # (a path-resolution failure would give a traceback + non-JSON stdout). + assert proc.returncode == 1, proc.stderr report = json.loads(proc.stdout) assert report["status"] == "ok", report assert report["iterations_observed"] == 4, report assert report["stalled"] is True, report + + +# --- M4-CLI item 8: explicit CLI exit codes per outcome --------------------- + + +def test_cli_exit_zero_when_healthy(tmp_path): + state = { + "active_task": "M5", + "best_score": 0.95, + "iteration_id": "4", + "budget_remaining": {"iterations": 6, "cost": 100}, + } + runlog = _runlog([(1, "M1", 0.4), (2, "M2", 0.6), (3, "M3", 0.8), (4, "M5", 0.95)]) + loop_dir = _write_loop(tmp_path, state, runlog) + + assert runtime_monitor.main([str(loop_dir)]) == 0 + + +def test_cli_exit_one_when_stalled(tmp_path): + state = {"active_task": "M2", "best_score": 0.5, "iteration_id": "4"} + runlog = _runlog([(1, "M2", 0.5), (2, "M2", 0.5), (3, "M2", 0.5), (4, "M2", 0.5)]) + loop_dir = _write_loop(tmp_path, state, runlog) + + # A loop that needs intervention (replan) must not exit 0. + assert runtime_monitor.main([str(loop_dir)]) == 1 + + +def test_cli_exit_one_when_repair_churn(tmp_path): + state = {"active_task": "M3", "best_score": 0.6, "iteration_id": "5"} + runlog = "\n".join( + [ + "# RUNLOG", + "", + "- iter 1: active_task=M3 verify=FAIL best_score=0.6 repair attempt=1 productive=false", + "- iter 2: active_task=M3 verify=FAIL best_score=0.6 repair attempt=2 productive=false", + "- iter 3: active_task=M3 verify=FAIL best_score=0.6 repair attempt=3 productive=false", + ] + ) + "\n" + loop_dir = _write_loop(tmp_path, state, runlog) + + assert runtime_monitor.main([str(loop_dir)]) == 1 + + +def test_cli_exit_one_when_budget_overrun(tmp_path): + state = { + "active_task": "M4", + "best_score": 0.7, + "iteration_id": "9", + "budget_remaining": {"iterations": 0, "cost": 0}, + } + runlog = _runlog([(i, "M4", 0.7) for i in range(1, 10)]) + loop_dir = _write_loop(tmp_path, state, runlog) + + assert runtime_monitor.main([str(loop_dir)]) == 1 + + +def test_cli_exit_two_when_state_missing(tmp_path): + loop_dir = tmp_path / ".loop" + loop_dir.mkdir() + + # A precondition/operational error (no state.json) is distinct from an + # intervention recommendation. + assert runtime_monitor.main([str(loop_dir)]) == 2 + + +def test_cli_exit_one_when_runlog_unparseable(tmp_path): + state = {"active_task": "T2", "state": "execute", "iteration_id": 3, "best_score": 0.8} + runlog = ( + "# RUNLOG\n\n" + "## Iteration 1\n" + "- **active_task:** `T1`\n" + "- **score:** 0.61\n" + ) + loop_dir = _write_loop(tmp_path, state, runlog) + + assert runtime_monitor.main([str(loop_dir)]) == 1 + + +def test_cli_exit_zero_when_terminal(tmp_path): + state = { + "active_task": None, + "state": "terminal", + "terminal_state": "Succeeded", + "iteration_id": 7, + } + runlog = _runlog([(1, "T1", 0.5)]) + loop_dir = _write_loop(tmp_path, state, runlog) + + # A finished loop is a clean exit, not an intervention. + assert runtime_monitor.main([str(loop_dir)]) == 0