multi-agent-mux/tests/test_loop_cli.py

#!/usr/bin/env python3
"""CLI parser and planner-resolution tests for multi-agent-mux-loop (Rev.4)."""

import os
import subprocess
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
RUN_LOOP_SH = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh"
SKILL_MD = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-loop" / "SKILL.md"


def _run_loop(args, env_extra=None, cwd=None):
    env = dict(os.environ)
    if env_extra:
        env.update(env_extra)
    return subprocess.run(
        ["bash", str(RUN_LOOP_SH), *args],
        capture_output=True,
        text=True,
        env=env,
        cwd=str(cwd or REPO_ROOT),
    )


def _combined(res):
    return (res.stdout or "") + (res.stderr or "")


def test_help_lists_creator_not_target_agent():
    res = _run_loop(["--help"])
    assert res.returncode != 0
    text = _combined(res)
    assert "--creator" in text
    assert "--planner" in text
    assert "--target-agent" not in text


def test_missing_creator_and_task_fail_fast():
    res = _run_loop([])
    assert res.returncode != 0
    assert "ERROR: --creator and --task are mandatory fields." in _combined(res)


def test_creator_parses_and_reaches_session_check(tmp_path):
    res = _run_loop(
        ["--creator", "nonexistent-agent", "--task", "t"],
        env_extra={"MAM_LOOP_NO_FREEZE": "1", "MAM_LOOP_MARKER": str(tmp_path / "loop-guard")},
    )
    assert res.returncode != 0
    text = _combined(res)
    assert "was removed" not in text
    assert "mandatory fields" not in text
    assert "not registered" in text or "already running" in text


def test_target_agent_is_rejected():
    res = _run_loop(["--target-agent", "any-session", "--task", "t"])
    assert res.returncode == 1
    text = _combined(res)
    assert "ERROR: --target-agent was removed. Use --creator <session> instead." in text
    assert "mandatory fields" not in text


def test_planner_without_plan_fail_fast():
    res = _run_loop(["--creator", "c1", "--planner", "p1", "--task", "t"])
    assert res.returncode == 1
    text = _combined(res)
    assert "ERROR: --planner was specified without --plan." in text
    assert "--plan --planner" in text


def test_skill_md_uses_creator_flag():
    content = SKILL_MD.read_text(encoding="utf-8")
    assert "--creator" in content
    assert "--planner" in content
    assert "--target-agent" not in content


def _seed_sessions(yaml_path, sessions):
    lines = ["herdr_sessions:"]
    for s in sessions:
        lines.append(f"- name: {s['name']}")
        lines.append(f"  status: {s['status']}")
        lines.append(f"  role: {s['role']}")
    yaml_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    db_path = yaml_path.with_suffix(".db")
    if db_path.exists():
        db_path.unlink()


def _loop_env(mam_sandbox, marker_path):
    return {
        "MAM_LOOP_NO_FREEZE": "1",
        "MAM_LOOP_MARKER": str(marker_path),
        "AGENT_SESSIONS_YAML": str(mam_sandbox / ".mam" / "agent-sessions.yaml"),
        "WORKSPACE_ROOT": str(mam_sandbox),
        "MAM_REAL_ROOT": str(mam_sandbox),
    }


def test_explicit_planner_unregistered(mam_sandbox, tmp_path):
    yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
    _seed_sessions(yaml_path, [{"name": "creator-1", "status": "running", "role": "creator"}])
    marker = tmp_path / "loop-guard-unreg"
    res = _run_loop(
        ["--creator", "creator-1", "--plan", "--planner", "missing-planner", "--task", "t"],
        env_extra=_loop_env(mam_sandbox, marker),
        cwd=mam_sandbox,
    )
    assert res.returncode != 0
    assert "Specified planner session 'missing-planner' is not registered in the session registry." in _combined(res)


def test_explicit_planner_not_running(mam_sandbox, tmp_path):
    yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
    _seed_sessions(
        yaml_path,
        [
            {"name": "creator-1", "status": "running", "role": "creator"},
            {"name": "planner-stopped", "status": "stopped", "role": "planner"},
        ],
    )
    marker = tmp_path / "loop-guard-stopped"
    res = _run_loop(
        ["--creator", "creator-1", "--plan", "--planner", "planner-stopped", "--task", "t"],
        env_extra=_loop_env(mam_sandbox, marker),
        cwd=mam_sandbox,
    )
    assert res.returncode != 0
    assert "Specified planner session 'planner-stopped' is not running (current status: 'stopped')." in _combined(res)


def test_explicit_planner_skips_auto_resolve(mam_sandbox, tmp_path):
    yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
    _seed_sessions(
        yaml_path,
        [
            {"name": "creator-1", "status": "running", "role": "creator"},
            {"name": "planner-first", "status": "running", "role": "planner"},
            {"name": "planner-explicit", "status": "running", "role": "planner"},
        ],
    )
    marker = tmp_path / "loop-guard-skip"
    res = _run_loop(
        ["--creator", "creator-1", "--plan", "--planner", "planner-explicit", "--task", "t"],
        env_extra=_loop_env(mam_sandbox, marker),
        cwd=mam_sandbox,
    )
    text = _combined(res)
    assert "Resolved Planner session: planner-explicit" in text
    assert "Resolved Planner session: planner-first" not in text