multi-agent-mux/tests/test_o1_rebuttal.py

#!/usr/bin/env python3
"""
tests/test_o1_rebuttal.py — O-1 Rebuttal & Re-adjudication Protocol regression suite (V-1..V-10).
"""

import os
import subprocess
import pytest

REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
RUN_LOOP_SH = os.path.join(REPO_ROOT, ".agents", "skills", "multi-agent-mux-loop", "scripts", "run_loop.sh")
RULES_MD = os.path.join(REPO_ROOT, ".agents", "MULTI_AGENT_RULES.md")
RULES_KO_MD = os.path.join(REPO_ROOT, ".agents", "MULTI_AGENT_RULES.ko.md")
SKILL_MD = os.path.join(REPO_ROOT, ".agents", "skills", "multi-agent-mux-loop", "SKILL.md")
IMPROVEMENTS_MD = os.path.join(REPO_ROOT, "IMPROVEMENTS.md")


def _get_run_loop_helpers():
    """Extract helper functions from run_loop.sh to run standalone in bash tests."""
    with open(RUN_LOOP_SH, "r", encoding="utf-8") as f:
        content = f.read()

    # Extract has_tag_line and extract_tag_values definitions
    has_tag_line_code = content.split("has_tag_line() {")[1].split("extract_tag_values() {")[0]
    extract_tag_values_code = content.split("extract_tag_values() {")[1].split("find_report() {")[0]

    return f"""
has_tag_line() {{
{has_tag_line_code}
extract_tag_values() {{
{extract_tag_values_code}
"""


def test_v1_has_tag_line_single_and_first_line(tmp_path):
    """V-1: has_tag_line matches tag on line 1 of report file."""
    report_file = tmp_path / "report-final.md"
    report_file.write_text("[ESCALATE: PLANNER]\nDetailed feedback follows...")

    helpers = _get_run_loop_helpers()
    cmd = f"""
{helpers}
has_tag_line "{report_file}" "ESCALATE: PLANNER"
"""
    res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True)
    assert res.returncode == 0


def test_v2_extract_tag_values_multiple_rebuts(tmp_path):
    """V-2: extract_tag_values extracts all [REBUT: target] tags."""
    report_file = tmp_path / "report-final.md"
    report_file.write_text("[REBUT: reviewer_a]\n[REBUT: reviewer_b]\nContent...")

    helpers = _get_run_loop_helpers()
    cmd = f"""
{helpers}
extract_tag_values "{report_file}" "REBUT"
"""
    res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True)
    assert res.returncode == 0
    lines = [line.strip() for line in res.stdout.splitlines() if line.strip()]
    assert lines == ["reviewer_a", "reviewer_b"]


def test_v3_rebut_option_parsing_and_budget():
    """V-3: --max-rebut N option parsing and error handling."""
    # Check help
    res_help = subprocess.run(["bash", RUN_LOOP_SH, "--help"], capture_output=True, text=True)
    assert "--max-rebut N" in res_help.stdout or "--max-rebut N" in res_help.stderr

    # Check invalid value
    res_invalid = subprocess.run(["bash", RUN_LOOP_SH, "--max-rebut", "abc"], capture_output=True, text=True)
    assert res_invalid.returncode != 0
    assert "ERROR: --max-rebut requires a non-negative integer." in res_invalid.stdout or "ERROR: --max-rebut requires a non-negative integer." in res_invalid.stderr


def test_v4_rebuttal_deduplication():
    """V-4: REBUT_TARGETS deduplication logic."""
    cmd = """
    FAIL_REVS=("rev1" "rev2")
    REBUT_TARGETS=()
    INPUT_TAGS=("rev1" "rev1" "rev2" "rev3")

    for _t in "${INPUT_TAGS[@]}"; do
        for _fv in "${FAIL_REVS[@]}"; do
            [ "$_t" = "$_fv" ] || continue
            _dup=0
            for _e in ${REBUT_TARGETS[@]+"${REBUT_TARGETS[@]}"}; do
                if [ "$_e" = "$_t" ]; then _dup=1; break; fi
            done
            [ "$_dup" -eq 0 ] && REBUT_TARGETS+=("$_t")
            break
        done
    done
    echo "${REBUT_TARGETS[*]}"
    """
    res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True)
    assert res.returncode == 0
    assert res.stdout.strip() == "rev1 rev2"


def test_v5_fail_closed_on_missing_sustained(tmp_path):
    """V-5: Non-SUSTAINED adjudication report fails closed (objection stands)."""
    report_overruled = tmp_path / "report_overruled.md"
    report_overruled.write_text("[ADJUDICATION: OVERRULED]\nObjection maintained.")

    report_empty = tmp_path / "report_empty.md"
    report_empty.write_text("No explicit adjudication tag present.")

    helpers = _get_run_loop_helpers()
    cmd = f"""
{helpers}
if has_tag_line "{report_overruled}" "ADJUDICATION: SUSTAINED"; then echo "SUSTAINED"; else echo "OVERRULED"; fi
if has_tag_line "{report_empty}" "ADJUDICATION: SUSTAINED"; then echo "SUSTAINED"; else echo "OVERRULED"; fi
"""
    res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True)
    assert res.returncode == 0
    lines = res.stdout.splitlines()
    assert lines[0] == "OVERRULED"
    assert lines[1] == "OVERRULED"


def test_v6_bash_3_2_empty_array_safety():
    """V-6: Empty array expansion with set -u on bash 3.2 syntax."""
    cmd = """
    set -euo pipefail
    declare -a EMPTY_ARR=()
    for item in ${EMPTY_ARR[@]+"${EMPTY_ARR[@]}"}; do
        echo "$item"
    done
    echo "SUCCESS"
    """
    res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True)
    assert res.returncode == 0
    assert "SUCCESS" in res.stdout


def test_v7_rules_docs_contain_section_3_1():
    """V-7: MULTI_AGENT_RULES.md and MULTI_AGENT_RULES.ko.md contain section 3.1."""
    with open(RULES_MD, "r", encoding="utf-8") as f:
        content_en = f.read()
    with open(RULES_KO_MD, "r", encoding="utf-8") as f:
        content_ko = f.read()

    assert "### 3.1 Rebuttal & Adjudication Protocol" in content_en
    assert "### 3.1 Rebuttal & Adjudication Protocol" in content_ko
    assert "[REBUT:" in content_en and "[ADJUDICATION: SUSTAINED]" in content_en
    assert "[REBUT:" in content_ko and "[ADJUDICATION: SUSTAINED]" in content_ko


def test_v8_skill_md_contains_max_rebut():
    """V-8: multi-agent-mux-loop/SKILL.md contains --max-rebut documentation."""
    with open(SKILL_MD, "r", encoding="utf-8") as f:
        content = f.read()

    assert "`--max-rebut`" in content
    assert "Rebuttal & Adjudication Protocol" in content


def test_v9_per_iteration_budget_reset():
    """V-9: run_loop.sh resets REBUT_BUDGET per iteration while tracking REBUT_TOTAL_BUDGET."""
    with open(RUN_LOOP_SH, "r", encoding="utf-8") as f:
        content = f.read()

    assert 'REBUT_BUDGET="$MAX_REBUT"' in content
    assert "REBUT_TOTAL_BUDGET=$((MAX_REBUT * MAX_LOOP))" in content
    assert "REBUT_TOTAL_BUDGET=$((REBUT_TOTAL_BUDGET - 1))" in content


def test_v10_improvements_md_o1_completed():
    """V-10: IMPROVEMENTS.md has O-1 moved to completed tasks section."""
    with open(IMPROVEMENTS_MD, "r", encoding="utf-8") as f:
        content = f.read()

    assert "O-1" in content
    assert "[O-1] Rebuttal & Re-adjudication Protocol" in content or "O-1: 타당하지 않은 리뷰 피드백 거부/반론 프로토콜 미지원" in content