"""Tests for Issue #5 Phase 3: run_loop.sh Python port and thin wrapper. Covers: - Strict parser: code fence stripping, last non-blank line verdict detection, fail-closed - Tag extraction: [ESCALATE: PLANNER], [REBUT: ], [ADJUDICATION], [ARBITRATION] - Rebuttal budget logic (iteration budget vs run-wide budget) - Promotion atomic writes - Checkpoint save / load / remove - Thin wrapper CLI compatibility, delegation and rollback (MAM_LOOP_PY=0/1) """ from __future__ import annotations import json import os import re import signal import subprocess import sys from pathlib import Path import pytest REPO_ROOT = Path(__file__).resolve().parent.parent SKILLS = REPO_ROOT / ".agents" / "skills" RUN_LOOP_SH = SKILLS / "multi-agent-mux-loop" / "scripts" / "run_loop.sh" if str(SKILLS) not in sys.path: sys.path.insert(0, str(SKILLS)) from lib_py.loop import parser, checkpoint, runner def test_p3_01_strict_verdict_ignores_fenced_code(): """P3-01: A verdict token inside markdown code fences must NOT be matched.""" content = """# Review Report The implementation has issues: ```markdown Here is how to write a verdict: [VERDICT: PASS] ``` Please fix the above lint errors. [VERDICT: NOT PASS] """ assert parser.parse_verdict(content) == "NOT PASS" def test_p3_01_strict_verdict_ignores_trailing_fenced_pass(): """P3-01: If the report ends with a code fence containing [VERDICT: PASS], it must be ignored.""" content = """# Review Report Here is a demonstration snippet: ``` [VERDICT: PASS] ``` """ assert parser.parse_verdict(content) is None def test_p3_01_strict_verdict_last_non_blank_line(): """P3-01: Verdict must occupy the last non-blank line outside code fences.""" content = """# Review Report Everything is verified cleanly. [VERDICT: PASS] """ assert parser.parse_verdict(content) == "PASS" def test_p3_01_strict_verdict_mid_text_ignored(): """P3-01: A verdict token quoted mid-text is not matched if followed by commentary.""" content = """# Review Report I initially thought [VERDICT: PASS] was warranted. However, further tests revealed a deadlock. """ assert parser.parse_verdict(content) is None def test_p3_01_has_tag_line_ignores_fenced_tags(): """P3-01: [ESCALATE: PLANNER] inside code fences must not trigger escalation.""" content = """# Review Report We discussed whether to escalate: ``` [ESCALATE: PLANNER] ``` We decided simple bugfix is sufficient. [VERDICT: NOT PASS] """ assert not parser.has_tag_line(content, "ESCALATE: PLANNER") assert parser.has_tag_line(content + "\n[ESCALATE: PLANNER]\n", "ESCALATE: PLANNER") def test_p3_01_extract_tag_values_deduplicates_and_ignores_fences(): """P3-01: extract_tag_values returns values outside fences, trimmed and clean.""" content = """# Correction Report ``` [REBUT: fake-reviewer] ``` I accept reviewer-01's first finding, but object to the second. [REBUT: reviewer-01] And also to reviewer-02: [REBUT: reviewer-02 ] """ values = parser.extract_tag_values(content, "REBUT") assert values == ["reviewer-01", "reviewer-02"] def test_p3_01_adjudication_and_arbitration_tags(): """P3-01: Adjudication and arbitration tags parsed strictly.""" s_sustained = "Re-adjudication:\n\n[ADJUDICATION: SUSTAINED]\n" s_overruled = "Re-adjudication:\n\n[ADJUDICATION: OVERRULED]\n" assert parser.has_tag_line(s_sustained, "ADJUDICATION: SUSTAINED") assert not parser.has_tag_line(s_sustained, "ADJUDICATION: OVERRULED") assert parser.has_tag_line(s_overruled, "ADJUDICATION: OVERRULED") arb_creator = "Arbitration report:\n\n[ARBITRATION: CREATOR]\n" arb_rev = "Arbitration report:\n\n[ARBITRATION: REVIEWER]\n" assert parser.has_tag_line(arb_creator, "ARBITRATION: CREATOR") assert parser.has_tag_line(arb_rev, "ARBITRATION: REVIEWER") def test_p3_01_checkpoint_atomic_roundtrip(tmp_path): """P3-01: Checkpoint writes atomically and loads cleanly.""" cp_file = tmp_path / "loop-checkpoint.json" data = { "iteration": 2, "created_jobs": ["job-1", "job-2"], "current_plan": "Step 1: Done\nStep 2: Testing", "review_status": {"reviewer-1": "NOT PASS"}, "base_commit": "abc1234", } checkpoint.save_checkpoint(cp_file, data) assert cp_file.exists() loaded = checkpoint.load_checkpoint(cp_file) assert loaded == data checkpoint.remove_checkpoint(cp_file) assert not cp_file.exists() assert checkpoint.load_checkpoint(cp_file) is None def test_p3_02_run_loop_wrapper_help_lists_same_as_bash(): """P3-02: run_loop.sh --help produces identical usage instructions and exits 1.""" res = subprocess.run(["bash", str(RUN_LOOP_SH), "--help"], capture_output=True, text=True) assert res.returncode == 1 assert "Usage:" in res.stdout and "[options] --creator --task " in res.stdout assert "--all-reviewer" in res.stdout assert "--max-rebut" in res.stdout def test_p3_02_run_loop_wrapper_delegates_to_python_default(tmp_path): """P3-02: Under MAM_LOOP_PY=1, run_loop.sh executes python lib_py.loop and reaches Target Agent check.""" marker = tmp_path / "loop-marker" env = dict(os.environ) env["MAM_LOOP_NO_FREEZE"] = "1" env["MAM_LOOP_MARKER"] = str(marker) env["MAM_LOOP_PY"] = "1" res = subprocess.run( ["bash", str(RUN_LOOP_SH), "--creator", "nonexistent-agent", "--task", "test-goal"], capture_output=True, text=True, env=env, cwd=str(REPO_ROOT), ) assert res.returncode == 1 assert "Target agent session 'nonexistent-agent' is not registered" in (res.stdout + res.stderr) def test_p3_02_run_loop_wrapper_rollback_mode(tmp_path): """P3-02: Under MAM_LOOP_PY=0, run_loop.sh executes legacy bash loop without launching python.""" marker = tmp_path / "loop-marker" env = dict(os.environ) env["MAM_LOOP_NO_FREEZE"] = "1" env["MAM_LOOP_MARKER"] = str(marker) env["MAM_LOOP_PY"] = "0" res = subprocess.run( ["bash", str(RUN_LOOP_SH), "--creator", "nonexistent-agent", "--task", "test-goal"], capture_output=True, text=True, env=env, cwd=str(REPO_ROOT), ) assert res.returncode == 1 assert "Target agent session 'nonexistent-agent' is not registered" in (res.stdout + res.stderr) def test_p3_03_runner_agent_type_resolution(): """P3-03: LoopRunner resolves agent types accurately.""" r = runner.LoopRunner(["--creator", "test-c", "--task", "test-t"]) state = { "herdr_sessions": [ {"name": "agent-1", "agent": "grok"}, {"name": "agent-2", "pane": {"cmd": "opencode"}}, {"name": "agent-codex-01"}, {"name": "agent-hermes-01"}, {"name": "agent-agy-01"}, {"name": "agent-other-01"}, ] } assert r.resolve_agent_type("agent-1", state) == "grok" assert r.resolve_agent_type("agent-2", state) == "opencode" assert r.resolve_agent_type("agent-codex-01", state) == "codex" assert r.resolve_agent_type("agent-hermes-01", state) == "hermes" assert r.resolve_agent_type("agent-agy-01", state) == "agy" assert r.resolve_agent_type("agent-other-01", state) == "claude"