multi-agent-mux/tests/test_issue21_phase2_loop_a.py

"""Issue #21 phase 2, loop A: worktree helpers (item 4), resume only at the
original cwd, loud control-plane binding failures (recommendation 1), the
binding persisted on resume's already-running path (recommendation 2), and the
installer's `/.venv` ignore rule (#27).

Every git repository lives under tmp_path / mam_sandbox; local submodules are
allowed through GIT_CONFIG_* environment only (never global config), and every
subprocess gets WORKSPACE_ROOT so lib.sh's .mam/shim lands in the sandbox.
"""

from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
import time
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parent.parent
SKILLS = REPO_ROOT / ".agents" / "skills"
if str(SKILLS) not in sys.path:
    sys.path.insert(0, str(SKILLS))

GIT_ID = {"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@example.invalid",
          "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@example.invalid"}
FILE_OK = {"GIT_CONFIG_COUNT": "1", "GIT_CONFIG_KEY_0": "protocol.file.allow", "GIT_CONFIG_VALUE_0": "always"}
FILE_NO = {"GIT_CONFIG_COUNT": "1", "GIT_CONFIG_KEY_0": "protocol.file.allow", "GIT_CONFIG_VALUE_0": "never"}
PREP = ("multi-agent-mux-create", "scripts", "worktree_prepare.sh")
REM = ("multi-agent-mux-stop", "scripts", "worktree_remove.sh")
CP_KEYS = ("workspace_root", "agent_sessions_yaml", "mam_jobs_dir", "delegate_job_logs_dir")


def _git(cwd, *args, env=None, check=True):
    return subprocess.run(["git", "-C", str(cwd), *args], capture_output=True, text=True, check=check,
                          env={**os.environ, **GIT_ID, **(env or {})})


def _skill(root: Path, *parts) -> str:
    return str(root / ".agents" / "skills" / Path(*parts))


def _run(root: Path, parts, *args, env=None, timeout=180):
    return subprocess.run(["bash", _skill(root, *parts), *args], capture_output=True, text=True, cwd=str(root),
                          env={**os.environ, "WORKSPACE_ROOT": str(root), **(env or {})}, timeout=timeout)


def _rows(root: Path) -> list:
    import yaml
    return (yaml.safe_load((root / ".mam" / "agent-sessions.yaml").read_text()) or {}).get("herdr_sessions") or []


def _mutate(root: Path, code: str) -> None:
    res = subprocess.run(["bash", "-c", f"source {_skill(root, 'lib.sh')} && atomic_dump_yaml {root / '.mam' / 'agent-sessions.yaml'}"],
                         input=code, capture_output=True, text=True, env={**os.environ, "WORKSPACE_ROOT": str(root)})
    assert res.returncode == 0, res.stderr


def _calls(mock_state_file, since=0) -> list:
    calls = json.loads(Path(mock_state_file).read_text()).get("calls", [])[since:]
    return [c[2:] if c[:1] == ["--session"] else c for c in calls]


def _spawn_calls(mock_state_file, since=0) -> list:
    return [c for c in _calls(mock_state_file, since)
            if c[:2] in (["pane", "split"], ["workspace", "create"], ["agent", "start"])]


def _worktrees(main: Path) -> str:
    return _git(main, "worktree", "list", "--porcelain").stdout


@pytest.fixture
def ctl(mam_sandbox, monkeypatch):
    """The sandbox is the control root: a git checkout with one commit and a .venv."""
    for k, v in GIT_ID.items():
        monkeypatch.setenv(k, v)
    main = mam_sandbox
    _git(main, "init", "-q")
    (main / "README").write_text("r\n")
    _git(main, "add", "README")
    _git(main, "commit", "-q", "-m", "init")
    (main / ".venv").mkdir(exist_ok=True)
    return main


@pytest.fixture
def ctl_sub(ctl):
    """Control root with a submodule (bare local remote) left uninitialised in
    the main checkout, like nats-docker."""
    main = ctl
    src, bare = main.parent / f"{main.name}-subsrc", main.parent / f"{main.name}-sub.git"
    src.mkdir()
    _git(src, "init", "-q")
    (src / "s.txt").write_text("s\n")
    _git(src, "add", "s.txt")
    _git(src, "commit", "-q", "-m", "s")
    subprocess.run(["git", "clone", "-q", "--bare", str(src), str(bare)], check=True, capture_output=True)
    _git(main, "submodule", "add", "-q", str(bare), "sm", env=FILE_OK)
    _git(main, "commit", "-q", "-m", "add sm")
    _git(main, "submodule", "deinit", "-q", "-f", "sm")
    assert _git(main, "config", "--get", "submodule.sm.url", check=False).returncode != 0
    return main


def _wt(main: Path, name: str = "wt") -> Path:
    return main.parent / f"{main.name}-{name}"


def _prepare(main, wt, *extra, env=None):
    return _run(main, PREP, "--path", str(wt), "--branch", f"b-{wt.name}", *extra, env=env)


# ---------------------------------------------------------------------------
# #27: the managed .gitignore block ignores .venv as a file, directory or link
# ---------------------------------------------------------------------------

def _install_env():
    env = dict(os.environ)
    env["MAM_REPO_URL"] = str(REPO_ROOT)
    env["MAM_SKIP_VENV"] = "1"
    env["MAM_INSTALLER_URL"] = "file://" + str(REPO_ROOT / "deploy" / "install.sh")
    return env


GI_START = "# >>> MAM managed block (managed by install.sh — do not edit) >>>"
GI_END = "# <<< MAM managed block <<<"


def _block(gi: str) -> list:
    lines = gi.splitlines()
    return lines[lines.index(GI_START) + 1:lines.index(GI_END)]


@pytest.mark.parametrize("installer", ["install.sh", "install_mam.sh"])
def test_w10_managed_block_ignores_venv_in_every_form(tmp_path, installer):
    tgt = tmp_path / "tgt"
    tgt.mkdir()
    cmd = ["bash", str(REPO_ROOT / "deploy" / installer)] + ([str(tgt)] if installer == "install.sh" else ["--target", str(tgt)])
    res = subprocess.run(cmd, capture_output=True, text=True, env=_install_env(), cwd=str(tmp_path), timeout=300)
    assert res.returncode == 0, res.stdout[-2000:] + res.stderr[-2000:]
    block = _block((tgt / ".gitignore").read_text())
    assert "/.venv" in block and "/.venv/" not in block
    _git(tgt, "init", "-q")
    v = tgt / ".venv"

    def clear():
        if v.is_symlink() or v.is_file():
            v.unlink()
        elif v.exists():
            shutil.rmtree(v)

    if v.exists() or v.is_symlink():  # whatever the installer itself left there is ignored too
        assert _git(tgt, "check-ignore", "-q", ".venv", check=False).returncode == 0
    for make in (lambda p: p.symlink_to(tmp_path), lambda p: p.write_text("x"), lambda p: p.mkdir()):
        clear()
        make(v)
        assert _git(tgt, "check-ignore", "-q", ".venv", check=False).returncode == 0, make
    clear()


def test_w11_reinstall_replaces_the_old_rule_and_keeps_user_lines(tmp_path):
    tgt = tmp_path / "tgt"
    tgt.mkdir()
    # an installation made before #27: the same markers, the old directory-only rule
    (tgt / ".gitignore").write_text(f"user-one\n{GI_START}\n/.venv/\n/.mam/\n{GI_END}\nuser-two\n")
    res = subprocess.run(["bash", str(REPO_ROOT / "deploy" / "install.sh"), str(tgt)],
                         capture_output=True, text=True, env=_install_env(), cwd=str(tmp_path), timeout=300)
    assert res.returncode == 0, res.stdout[-2000:] + res.stderr[-2000:]
    gi = (tgt / ".gitignore").read_text()
    assert gi.count(GI_START) == 1 and gi.count(GI_END) == 1
    assert "/.venv" in _block(gi) and "/.venv/" not in _block(gi)
    assert "user-one" in gi.splitlines() and "user-two" in gi.splitlines()
    for ledger in list(tgt.rglob("asset_hashes.txt")) + list(tgt.rglob("install_manifest.txt")):
        assert ".gitignore" not in ledger.read_text().split(), ledger