multi-agent-mux/tests/test_issue21_worktree_prereqs.py

"""Issue #21 phase-1 prerequisites for the worktree operating model.

  §1 (item 1) delegate-job wrapper: absolute brief / start-command registry /
     audit-log root / Output Report Path, runnable from any cwd.
  §2 (item 2) loop runner --worktree / --base / --review-only and the
     fail-closed empty change-set guard; the control plane stays in the main
     checkout.
  §3 (item 3) control-plane binding persisted on the session row and forwarded
     to the pane; status / reconcile / stop honour MAM_JOBS_DIR.

Every test writes only under tmp_path / mam_sandbox (Issue #20).
"""

from __future__ import annotations

import json
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"
WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
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))

UNREACHABLE = {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "MQTT_TLS": "0"}
GIT_ID = {"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@example.invalid",
          "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@example.invalid"}


def _env(home: Path, extra=None, drop=()) -> dict:
    env = dict(os.environ)
    env["DELEGATE_JOB_PYTHON"] = sys.executable
    env["AGENT_PYTHON_BIN"] = sys.executable
    env["HOME"] = str(home)
    env["MAM_ENV_FILE"] = str(home / ".mam.env.absent")
    env["PYTHONPATH"] = str(SKILLS)
    for k in ("MQTT_BROKER", "MQTT_HOST", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD",
              "MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR", "WORKSPACE_ROOT"):
        env.pop(k, None)
    env.update(UNREACHABLE)
    if extra:
        env.update({k: str(v) for k, v in extra.items()})
    for k in drop:
        env.pop(k, None)
    return env


def _only_job_id(reg: Path) -> str:
    ids = [p.stem for p in reg.glob("*.json")]
    assert len(ids) == 1, ids
    return ids[0]


def _kill_timer(reg: Path, jid: str) -> None:
    try:
        pid = int((reg / f"{jid}.heartbeat.pid").read_text().strip())
        os.kill(pid, signal.SIGKILL)
    except (OSError, ValueError):
        pass


def _dry_submit(cwd: Path, env: dict, *args) -> subprocess.CompletedProcess:
    return subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--prompt", "p21",
                           "--dry-run", *args],
                          capture_output=True, text=True, env=env, cwd=str(cwd), timeout=60)


def _injected(stdout: str, name: str) -> str:
    m = re.search(name + r"='(.*?)'", stdout)
    assert m, stdout
    return m.group(1)


# ---------------------------------------------------------------------------
# §1 item 1 — the wrapper's injected paths are absolute
# ---------------------------------------------------------------------------

def test_t1_1_injected_paths_are_absolute(tmp_path):
    a = tmp_path / "A"
    a.mkdir()
    real_a = os.path.realpath(a)
    env = _env(tmp_path, {"WORKSPACE_ROOT": a, "DELEGATE_JOB_LOGS_DIR": a / "logs"})
    res = _dry_submit(a, env, "--workdir", str(a))
    assert res.returncode == 0, res.stderr
    jid = _only_job_id(a / ".mam" / "jobs")
    brief = f"{real_a}/.mam/jobs/{jid}/brief.md"
    assert res.stdout.count(brief) == 2, res.stdout
    start = _injected(res.stdout, "start")
    assert f"--registry-dir {real_a}/.mam/jobs " in start
    assert start.startswith(f"DELEGATE_JOB_LOGS_DIR={real_a}/logs "), start
    # no cwd-relative registry path is left anywhere in the injected text
    assert res.stdout.count(f".mam/jobs/{jid}") == res.stdout.count(f"{real_a}/.mam/jobs/{jid}")


def test_t1_2_start_and_success_commands_run_from_another_cwd(tmp_path):
    """The core regression: a registry and an audit-log root with a space in
    them, the commands executed as written from a different cwd with no
    workspace env - the job advances, the audit trail stays in the control
    root, and nothing is written under the other cwd."""
    ctl = tmp_path / "ctl root"
    reg, logs, wt = ctl / "jobs", ctl / "logs", tmp_path / "wt"
    for d in (reg, logs, wt):
        d.mkdir(parents=True)
    env = _env(tmp_path, {"WORKSPACE_ROOT": ctl, "DELEGATE_JOB_LOGS_DIR": logs})
    res = _dry_submit(ctl, env, "--registry-dir", str(reg), "--workdir", str(ctl),
                      "--heartbeat", "7", "--ack-timeout", "11")
    assert res.returncode == 0, res.stderr
    jid = _only_job_id(reg)
    start = _injected(res.stdout, "start")
    success = _injected(res.stdout, "success").replace("<summary>", "done")
    agent_env = _env(tmp_path)  # no WORKSPACE_ROOT / MAM_JOBS_DIR / DELEGATE_JOB_LOGS_DIR
    try:
        r1 = subprocess.run(["bash", "-c", start + " --attempts 1"], capture_output=True, text=True,
                            env=agent_env, cwd=str(wt), timeout=120)
        # rc 2 = broker unreachable; the event is recorded locally first (test18 contract)
        assert r1.returncode == 2, r1.stderr
        assert json.loads((reg / f"{jid}.json").read_text())["status"] != "pending"
        r2 = subprocess.run(["bash", "-c", success + " --attempts 1"], capture_output=True, text=True,
                            env=agent_env, cwd=str(wt), timeout=120)
        assert r2.returncode == 2, r2.stderr
        assert json.loads((reg / f"{jid}.json").read_text())["status"] == "completed"
    finally:
        _kill_timer(reg, jid)
    audit = [json.loads(l) for l in (logs / jid / "events.ndjson").read_text().splitlines() if l.strip()]
    published = [json.dumps(e) for e in audit if e.get("event") == "published"]
    assert len(published) == 2, audit
    assert '"started"' in published[0] and '"completed"' in published[1], published
    assert not (wt / ".mam").exists(), "the agent's cwd must not grow a .mam"


def test_t1_3_mam_jobs_dir_is_the_default_registry(tmp_path):
    x, y = tmp_path / "X", tmp_path / "Y"
    env = _env(tmp_path, {"MAM_JOBS_DIR": x, "WORKSPACE_ROOT": tmp_path, "DELEGATE_JOB_LOGS_DIR": tmp_path / "logs"})
    res = _dry_submit(tmp_path, env, "--workdir", str(tmp_path))
    assert res.returncode == 0, res.stderr
    assert len(list(x.glob("*.json"))) == 1
    assert not (tmp_path / ".mam" / "jobs").exists()

    res = _dry_submit(tmp_path, env, "--workdir", str(tmp_path), "--registry-dir", str(y))
    assert res.returncode == 0, res.stderr
    assert len(list(y.glob("*.json"))) == 1, "an explicit --registry-dir wins over MAM_JOBS_DIR"
    assert len(list(x.glob("*.json"))) == 1


def test_t1_4_wrapper_source_guard():
    """The loop job type never builds its text under --dry-run (it returns
    early), so both paths are pinned to the shared helpers here."""
    src = WRAPPER.read_text()
    assert ".mam/jobs/$JOB_ID" not in src
    assert src.count('instructions="$(_instructions "$pub")"') == 1
    assert src.count('instructions="$(_instructions "$pub" "$iteration")"') == 1
    assert src.count('pub="$(_pub_cmd ') == 2
    report_lines = [l for l in src.splitlines() if "**Output Report Path**" in l]
    assert len(report_lines) == 2 and all("$REGISTRY_DIR/$JOB_ID/" in l for l in report_lines)


def _sent_texts(state) -> list:
    out = []

    def walk(o):
        if isinstance(o, dict):
            for k, v in o.items():
                if k == "sent_text" and isinstance(v, str):
                    out.append(v)
                else:
                    walk(v)
        elif isinstance(o, list):
            for v in o:
                walk(v)
    walk(state)
    return out


def test_t1_5_onboarding_publish_command_is_absolute(mam_sandbox, mock_herdr, mock_agents):
    tmp = mam_sandbox
    script = tmp / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
    res = subprocess.run(["bash", str(script), "--workspace", str(tmp), "--agent", "claude",
                          "--role", "creator", "--onboard"],
                         capture_output=True, text=True, cwd=str(tmp), timeout=180)
    assert res.returncode == 0, res.stdout + res.stderr
    texts = [t for t in _sent_texts(json.loads(Path(mock_herdr).read_text())) if "publish_event.py" in t]
    assert texts, "the onboarding instructions were not injected"
    text = "".join(texts)
    real = os.path.realpath(tmp)
    assert f" {real}/.agents/skills/multi-agent-mux-delegate-job/scripts/publish_event.py " in text, text
    assert f"--registry-dir {real}/.mam/jobs " in text, text
    assert f"DELEGATE_JOB_LOGS_DIR={real}/.mam/delegate_job_logs " in text, text
    assert " .agents/skills/" not in text