multi-agent-mux/tests/test_issue5_phase1_delegation.py

#!/usr/bin/env python3
"""Issue #5 Phase 1 — delegation extraction characterization & contract tests.

Written BEFORE the `lib_py/delegation/` extraction (P1-00) so the refactor has a
real regression oracle: the three shell helpers in `lib.sh` had zero direct test
coverage. Every test here must pass identically against the legacy inline bodies
and against the Python-backed ones, which is why the shell tests are
parametrized over `MAM_DELEGATION_PY` in {"0", "1"}.

Sections
  1. C-01..C-09  shell helper golden behaviour (non-fatal contract, stdout shape,
                 daemon PID/dedup/logging).
  2. CLI contract snapshots — argparse option names/choices/defaults, asserted
     against the parser rather than --help text so cosmetic wording is free.
  3. Auto-completion invariant guards — the three surfaces that decide job status
     must not shift by a single line during Phase 1.
"""
from __future__ import annotations

import os
import re
import signal
import subprocess
import sys
import time
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parent.parent
SKILLS = REPO_ROOT / ".agents" / "skills"
FIXTURES = REPO_ROOT / "tests" / "fixtures" / "issue_5" / "phase1"

DELEGATION_MODES = ("0", "1")  # MAM_DELEGATION_PY: legacy inline body / python body


# ==============================================================================
# Helpers
# ==============================================================================

def _sandbox_skills(mam_sandbox) -> Path:
    return Path(mam_sandbox) / "skills"


def _run_helper(mam_sandbox, snippet: str, env=None, timeout=30):
    """Source the sandbox lib.sh and run a shell snippet against it."""
    lib_sh = _sandbox_skills(mam_sandbox) / "lib.sh"
    run_env = dict(os.environ)
    run_env.pop("HERDR_SESSION_NAME", None)
    run_env.pop("HERDR_SERVER_NAME", None)
    if env:
        run_env.update({k: str(v) for k, v in env.items()})
    return subprocess.run(
        ["bash", "-c", 'source "{}" >/dev/null 2>&1; {}'.format(lib_sh, snippet)],
        capture_output=True, text=True, env=run_env, timeout=timeout,
    )


def _stub_env(tmp_path, *, exit_code=0, stdout=""):
    """Point _delegate_py_bin at the recording stub via AGENT_PYTHON_BIN."""
    log = tmp_path / "stub_argv.log"
    return {
        "AGENT_PYTHON_BIN": str(FIXTURES / "fake_python.sh"),
        "STUB_ARGV_LOG": str(log),
        "STUB_EXIT": str(exit_code),
        "STUB_STDOUT": stdout,
    }, log


def _install_fake_monitor(mam_sandbox) -> Path:
    """Replace the sandbox reconcile.sh with a sleeping stub and return its path."""
    target = (Path(mam_sandbox) / ".agents" / "skills"
              / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh")
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text((FIXTURES / "fake_reconcile.sh").read_text(encoding="utf-8"),
                      encoding="utf-8")
    target.chmod(0o755)
    return target


def _kill(pids):
    for pid in pids:
        try:
            os.kill(int(pid), signal.SIGKILL)
        except (OSError, ValueError):
            pass


def _alive(pid) -> bool:
    try:
        os.kill(int(pid), 0)
        return True
    except (OSError, ValueError):
        return False


# ==============================================================================
# 1. Shell helper golden behaviour (C-01 .. C-09)
# ==============================================================================

@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c01_publish_event_empty_job_is_noop(mam_sandbox, mode):
    """C-01: an empty job id returns 0 without ever starting an interpreter."""
    env, log = _stub_env(mam_sandbox, exit_code=7)
    env["MAM_DELEGATION_PY"] = mode
    res = _run_helper(mam_sandbox, 'delegate_publish_event "" started', env=env)
    assert res.returncode == 0, res.stderr
    assert not log.exists(), "python must not be invoked for an empty job id"


@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c02_publish_event_missing_script_is_non_fatal(mam_sandbox, mode):
    """C-02: a missing publish_event.py must not abort the caller."""
    script = (_sandbox_skills(mam_sandbox) / "multi-agent-mux-delegate-job"
              / "scripts" / "publish_event.py")
    script.unlink()
    env, _ = _stub_env(mam_sandbox)
    env["MAM_DELEGATION_PY"] = mode
    res = _run_helper(mam_sandbox, 'delegate_publish_event j1 started', env=env)
    assert res.returncode == 0, res.stderr


@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c03_publish_event_broker_failure_is_non_fatal(mam_sandbox, mode):
    """C-03: a non-zero interpreter exit (broker down) still returns 0."""
    env, log = _stub_env(mam_sandbox, exit_code=2)
    env["MAM_DELEGATION_PY"] = mode
    res = _run_helper(
        mam_sandbox,
        'delegate_publish_event j1 completed "done"; echo "rc=$?"',
        env=env,
    )
    assert res.returncode == 0, res.stderr
    assert "rc=0" in res.stdout, res.stdout
    assert log.exists(), "the interpreter should have been invoked"


@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c04_publish_event_forwards_job_event_detail(mam_sandbox, mode):
    """C-04: the job id, event and detail reach the interpreter argv intact."""
    env, log = _stub_env(mam_sandbox)
    env["MAM_DELEGATION_PY"] = mode
    res = _run_helper(
        mam_sandbox,
        'delegate_publish_event abc123 progress "half way"',
        env=env,
    )
    assert res.returncode == 0, res.stderr
    argv = log.read_text(encoding="utf-8")
    assert "--job abc123" in argv, argv
    assert "--event progress" in argv, argv
    assert "--detail half way" in argv, argv


@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c05_submit_job_stdout_is_exactly_the_job_id(mam_sandbox, mode):
    """C-05: create_session.sh captures this with $( ) — stdout must be the id alone."""
    env, _ = _stub_env(mam_sandbox, stdout="deadbeef")
    env["MAM_DELEGATION_PY"] = mode
    res = _run_helper(
        mam_sandbox,
        'delegate_submit_job "a prompt" claude-code herdr:claude',
        env=env,
    )
    assert res.returncode == 0, res.stderr
    assert res.stdout == "deadbeef\n", repr(res.stdout)


@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c06_watchdog_missing_monitor_script_errors(mam_sandbox, mode):
    """C-06: exact stderr wording and rc=1 when reconcile.sh is absent."""
    workdir = Path(mam_sandbox) / "empty_ws"
    workdir.mkdir()
    expected = (workdir / ".agents" / "skills" / "multi-agent-mux-monitor"
                / "scripts" / "reconcile.sh")
    res = _run_helper(
        mam_sandbox,
        'start_watchdog job1 "{}"'.format(workdir),
        env={"MAM_DELEGATION_PY": mode},
    )
    assert res.returncode == 1, res.stdout
    assert res.stderr.strip() == "ERROR: monitor script not found: {}".format(expected)


@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c07_watchdog_spawns_daemon_with_workdir_and_log(mam_sandbox, mode):
    """C-07: prints a live numeric PID, anchors cwd at workdir, appends to the log."""
    _install_fake_monitor(mam_sandbox)
    workdir = Path(mam_sandbox)
    (workdir / ".mam").mkdir(exist_ok=True)
    spawned = []
    try:
        res = _run_helper(
            mam_sandbox,
            'start_watchdog job1 "{}"'.format(workdir),
            env={"MAM_DELEGATION_PY": mode},
        )
        assert res.returncode == 0, res.stderr
        pid = res.stdout.strip()
        assert re.fullmatch(r"\d+", pid), repr(res.stdout)
        spawned.append(pid)
        deadline = time.time() + 5
        log_file = workdir / ".mam" / "multi-agent-mux-monitor.log"
        while time.time() < deadline and not log_file.exists():
            time.sleep(0.1)
        assert log_file.exists(), "monitor log was not created"
        assert _alive(pid), "spawned daemon is not running"
    finally:
        _kill(spawned)


@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c08_watchdog_second_call_reuses_running_daemon(mam_sandbox, mode):
    """C-08: the pgrep pre-check must return the existing PID, not spawn a second."""
    monitor = _install_fake_monitor(mam_sandbox)
    workdir = Path(mam_sandbox)
    (workdir / ".mam").mkdir(exist_ok=True)
    spawned = []
    try:
        first = _run_helper(mam_sandbox, 'start_watchdog job1 "{}"'.format(workdir),
                            env={"MAM_DELEGATION_PY": mode})
        pid1 = first.stdout.strip()
        spawned.append(pid1)
        assert _alive(pid1)
        second = _run_helper(mam_sandbox, 'start_watchdog job1 "{}"'.format(workdir),
                             env={"MAM_DELEGATION_PY": mode})
        assert second.returncode == 0, second.stderr
        assert second.stdout.strip() == pid1, (first.stdout, second.stdout)
        matches = subprocess.run(
            ["pgrep", "-f", "bash {} --subscribe".format(monitor)],
            capture_output=True, text=True,
        ).stdout.split()
        spawned.extend(matches)
        assert len(matches) == 1, "expected exactly one daemon, got {}".format(matches)
    finally:
        _kill(spawned)


@pytest.mark.parametrize("mode", DELEGATION_MODES)
def test_c09_start_watchdog_and_ensure_subscriber_daemon_agree(mam_sandbox, mode):
    """C-09: `start_watchdog` survives as an alias once the python body lands.

    Passes on the legacy tree (only start_watchdog defined) and on the refactored
    tree (both defined and behaviourally identical)."""
    _install_fake_monitor(mam_sandbox)
    workdir = Path(mam_sandbox)
    (workdir / ".mam").mkdir(exist_ok=True)
    probe = _run_helper(
        mam_sandbox,
        'declare -F start_watchdog >/dev/null && echo HAS_SW; '
        'declare -F ensure_subscriber_daemon >/dev/null && echo HAS_ESD',
        env={"MAM_DELEGATION_PY": mode},
    )
    assert "HAS_SW" in probe.stdout, "start_watchdog must always remain defined"
    if "HAS_ESD" not in probe.stdout:
        return  # legacy tree: alias not introduced yet (P1-02)

    spawned = []
    try:
        a = _run_helper(mam_sandbox, 'start_watchdog job1 "{}"'.format(workdir),
                        env={"MAM_DELEGATION_PY": mode})
        b = _run_helper(mam_sandbox, 'ensure_subscriber_daemon job1 "{}"'.format(workdir),
                        env={"MAM_DELEGATION_PY": mode})
        spawned += [a.stdout.strip(), b.stdout.strip()]
        assert a.returncode == b.returncode == 0, (a.stderr, b.stderr)
        assert a.stdout.strip() == b.stdout.strip(), (a.stdout, b.stdout)
    finally:
        _kill(spawned)


# ==============================================================================
# 2. CLI contract snapshots
# ==============================================================================

def _parser_snapshot(parser):
    """{option string: (nargs-ish kind, choices, default)} for one parser level."""
    snap = {}
    for action in parser._actions:
        if not action.option_strings:
            snap["<positional:{}>".format(action.dest)] = (
                action.type.__name__ if action.type else None,
                tuple(action.choices) if action.choices else None,
                action.default,
            )
            continue
        key = ",".join(sorted(action.option_strings))
        snap[key] = (
            action.type.__name__ if action.type else None,
            tuple(action.choices) if action.choices else None,
            action.default,
        )
    return snap


def _import_delegation(mam_sandbox, name):
    """Import a delegation module the way production callers do (top-level name)."""
    scripts = str(_sandbox_skills(mam_sandbox) / "multi-agent-mux-delegate-job" / "scripts")
    if scripts not in sys.path:
        sys.path.insert(0, scripts)
    return __import__(name)


def test_publish_event_cli_contract_snapshot(mam_sandbox):
    publish_event = _import_delegation(mam_sandbox, "publish_event")
    assert publish_event.VALID_EVENTS == (
        "started", "permission_required", "progress", "completed", "error", "cancelled",
    )
    assert publish_event.TERMINAL_EVENTS == ("completed", "error", "cancelled")
    # main() builds its parser inline, so --help is the only way to exercise it.
    with pytest.raises(SystemExit) as exc:
        publish_event.main(["--help"])
    assert exc.value.code == 0
    # Defaults that shell and wrapper callers depend on.
    src = Path(publish_event.__file__).read_text(encoding="utf-8")
    assert '"--attempts", type=int, default=3' in src
    assert '"--registry-dir", default=DEFAULT_REGISTRY_DIR' in src
    assert '"--event", default="progress", choices=VALID_EVENTS' in src


def test_registry_cli_contract_snapshot(mam_sandbox):
    registry = _import_delegation(mam_sandbox, "registry")
    parser = registry._build_parser()
    top = _parser_snapshot(parser)
    assert "--registry-dir" in top
    subparsers = [a for a in parser._actions if hasattr(a, "choices") and isinstance(a.choices, dict)]
    assert subparsers, "registry must expose subcommands"
    commands = set(subparsers[0].choices)
    assert commands == {
        "register", "list", "get", "status", "update", "get-feedback", "pick", "logs",
    }, commands
    reg = _parser_snapshot(subparsers[0].choices["register"])
    assert reg["--prompt"][2] is None
    assert reg["--agent"][2] == "claude-code"
    assert reg["--agent-session"][2] == "herdr:claude"
    assert reg["--role"][2] == "Worker"
    assert reg["--timeout"][2] == 3600
    assert reg["--idle-timeout"][2] == 120
    assert reg["--bits"][2] == 32
    assert reg["--job-type"][1] == ("direct", "loop", "discuss")
    assert reg["--max-iterations"][2] == 5
    assert registry.VALID_STATUSES == (
        "pending", "running", "completed", "error", "cancelled",
    )


def test_job_subscriber_cli_contract_snapshot(mam_sandbox):
    job_subscriber = _import_delegation(mam_sandbox, "job_subscriber")
    assert job_subscriber.TERMINAL_EVENTS == ("completed", "error")
    src = Path(job_subscriber.__file__).read_text(encoding="utf-8")
    for doc in ("  0  all watched jobs reached", "  1  any watched job reached",
                "  2  timed out"):
        assert doc in src, doc
    assert 'return 3' in src, "fatal-error exit code 3 must survive"


# ==============================================================================
# 3. Auto-completion invariant guards (Phase 1 must not move these)
# ==============================================================================

def test_event_to_status_mapping_is_frozen(mam_sandbox):
    publish_event = _import_delegation(mam_sandbox, "publish_event")
    assert publish_event.EVENT_TO_STATUS == {
        "started": "running",
        "completed": "completed",
        "error": "error",
        "cancelled": "cancelled",
    }


def test_wrapper_exit_code_mapping_literals_are_frozen():
    wrapper = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
    text = wrapper.read_text(encoding="utf-8")
    for literal in (
        'if [[ $sub_rc -eq 0 ]]; then',
        'elif [[ $sub_rc -eq 1 ]]; then',
        'elif [[ $sub_rc -eq 3 ]]; then',
        'job_status="completed"',
        'job_status="error"',
    ):
        assert literal in text, literal


def test_reconcile_terminal_mutation_literals_are_frozen():
    reconcile = SKILLS / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh"
    text = reconcile.read_text(encoding="utf-8")
    for literal in (
        "s['delegate_job_id'] = None",
        "s['status'] = 'terminated'",
        "kill-session",
        "if _event in ('completed', 'cancelled')",
    ):
        assert literal in text, literal