cy0307/awesome-loop-engineering

1#!/usr/bin/env python32"""Smoke-test the dependency-light runnable loop examples."""3 4from __future__ import annotations5 6import json7import os8import shlex9import subprocess10import sys11import tempfile12from pathlib import Path13 14 15ROOT = Path(__file__).resolve().parents[1]16RUNNABLE = ROOT / "examples" / "runnable"17DOCS_DRIFT = RUNNABLE / "docs-drift" / "docs-drift-loop.py"18 19 20def run(21    argv: list[str],22    *,23    cwd: Path = ROOT,24    env: dict[str, str] | None = None,25    expected: tuple[int, ...] = (0,),26) -> subprocess.CompletedProcess[str]:27    result = subprocess.run(argv, cwd=cwd, env=env, text=True, capture_output=True, check=False, timeout=30)28    if result.returncode not in expected:29        raise RuntimeError(30            f"command failed ({result.returncode}): {shlex.join(argv)}\nstdout:\n{result.stdout}\nstderr:\n{result.stderr}"31        )32    return result33 34 35def write(path: Path, text: str) -> None:36    path.write_text(text, encoding="utf-8")37 38 39def main() -> int:40    run(["bash", "-n", str(RUNNABLE / "test-repair-loop.sh"), str(RUNNABLE / "threshold-monitor-loop.sh")])41    run([sys.executable, "-m", "py_compile", str(RUNNABLE / "queue-worker-loop.py"), str(DOCS_DRIFT)])42 43    with tempfile.TemporaryDirectory(prefix="ale-runnable-") as temporary:44        temp = Path(temporary)45        queue = temp / "queue.jsonl"46        write(queue, '{"id":"docs-1","objective":"Check the docs"}\n')47        run([sys.executable, str(RUNNABLE / "queue-worker-loop.py"), "--queue", str(queue), "--dry-run"])48 49        monitor_state = temp / "monitor.md"50        monitor_env = {51            **os.environ,52            "PROBE_CMD": "printf '42\\n'",53            "THRESHOLD": "100",54            "MAX_SAMPLES": "1",55            "PROGRESS_FILE": str(monitor_state),56        }57        run(["bash", str(RUNNABLE / "threshold-monitor-loop.sh")], env=monitor_env)58 59        detector_clean = temp / "detector-clean.py"60        write(detector_clean, "raise SystemExit(0)\n")61        state = temp / "docs-drift.jsonl"62        run(63            [64                sys.executable,65                str(DOCS_DRIFT),66                "--discover-command",67                shlex.join([sys.executable, str(detector_clean)]),68                "--state",69                str(state),70            ]71        )72        receipt = json.loads(state.read_text(encoding="utf-8").splitlines()[-1])73        if receipt["status"] != "no_drift":74            raise RuntimeError("docs-drift no-work exit did not persist a no_drift receipt")75 76        detector_drift = temp / "detector-drift.py"77        write(detector_drift, "print('README output differs from current CLI help')\nraise SystemExit(1)\n")78        dry_run = run(79            [80                sys.executable,81                str(DOCS_DRIFT),82                "--discover-command",83                shlex.join([sys.executable, str(detector_drift)]),84                "--dry-run",85            ]86        )87        if json.loads(dry_run.stdout)["status"] != "drift_detected":88            raise RuntimeError("docs-drift dry run did not expose confirmed evidence")89 90        report_state = temp / "report-state.jsonl"91        run(92            [93                sys.executable,94                str(DOCS_DRIFT),95                "--discover-command",96                shlex.join([sys.executable, str(detector_drift)]),97                "--report-only",98                "--state",99                str(report_state),100            ],101            expected=(2,),102        )103        report_receipt = json.loads(report_state.read_text(encoding="utf-8").splitlines()[-1])104        if report_receipt["status"] != "reported":105            raise RuntimeError("docs-drift report-only mode did not persist an escalation receipt")106 107        repository = temp / "repo"108        repository.mkdir()109        write(repository / "README.md", "old command\n")110        run(["git", "init", "-q"], cwd=repository)111        run(["git", "add", "README.md"], cwd=repository)112        run(113            [114                "git",115                "-c",116                "user.name=Loop Smoke Test",117                "-c",118                "user.email=loop-smoke@example.com",119                "commit",120                "-qm",121                "fixture",122            ],123            cwd=repository,124        )125 126        agent = temp / "agent.py"127        verifier = temp / "verifier.py"128        write(agent, "from pathlib import Path\nPath('README.md').write_text('current command\\n', encoding='utf-8')\n")129        write(130            verifier,131            "from pathlib import Path\nraise SystemExit(0 if Path('README.md').read_text(encoding='utf-8') == 'current command\\n' else 1)\n",132        )133        patch_state = temp / "patch-state.jsonl"134        run(135            [136                sys.executable,137                str(DOCS_DRIFT),138                "--discover-command",139                shlex.join([sys.executable, str(detector_drift)]),140                "--agent-command",141                shlex.join([sys.executable, str(agent)]),142                "--verify-command",143                shlex.join([sys.executable, str(verifier)]),144                "--allowed-path",145                "README.md",146                "--workdir",147                str(repository),148                "--state",149                str(patch_state),150            ]151        )152        patch_receipt = json.loads(patch_state.read_text(encoding="utf-8").splitlines()[-1])153        if patch_receipt["status"] != "verified" or patch_receipt["changed_paths"] != ["README.md"]:154            raise RuntimeError("docs-drift patch mode did not persist a verified scoped change")155 156        run(["git", "restore", "README.md"], cwd=repository)157        unsafe_agent = temp / "unsafe-agent.py"158        write(unsafe_agent, "from pathlib import Path\nPath('outside.txt').write_text('unsafe\\n')\nraise SystemExit(1)\n")159        unsafe_state = temp / "unsafe-state.jsonl"160        run(161            [162                sys.executable,163                str(DOCS_DRIFT),164                "--discover-command",165                shlex.join([sys.executable, str(detector_drift)]),166                "--agent-command",167                shlex.join([sys.executable, str(unsafe_agent)]),168                "--verify-command",169                shlex.join([sys.executable, str(verifier)]),170                "--allowed-path",171                "README.md",172                "--workdir",173                str(repository),174                "--state",175                str(unsafe_state),176            ],177            expected=(2,),178        )179        unsafe_receipt = json.loads(unsafe_state.read_text(encoding="utf-8").splitlines()[-1])180        if unsafe_receipt["status"] != "escalated" or "out-of-scope paths" not in unsafe_receipt["reason"]:181            raise RuntimeError("docs-drift did not stop an out-of-scope edit from a failed agent command")182 183    print("Runnable loop smoke checks passed.")184    return 0185 186 187if __name__ == "__main__":188    raise SystemExit(main())189