multi-agent-mux/tests/test_issue5_phase3_loop.py

"""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: <rev>], [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

# Issue #17: shells started here inherit a WORKSPACE_ROOT of this test's own
# (lib.sh would otherwise rewrite the checkout's .mam/shim/herdr).
pytestmark = pytest.mark.usefixtures("own_workspace_root")


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 <agent-session-name> --task <goal-text>" 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"