multi-agent-mux/tests/test_issue15_registry_isolation.py

#!/usr/bin/env python3
"""Issue #15 — the job registry and its audit log follow the workspace, not
the caller's cwd.

`DEFAULT_REGISTRY_DIR` (".mam/jobs") and the audit-log root were resolved
relative to the current directory, so `create_session.sh`'s default onboarding
job — registered through `lib_py.delegation.cli submit` without a
`--registry-dir` — landed in whatever `.mam/jobs` sat under the cwd. Run from
the repo root, the pytest suite left hundreds of `running` test records in the
real registry. The defaults now honour `MAM_JOBS_DIR` / `DELEGATE_JOB_LOGS_DIR`,
then `$WORKSPACE_ROOT/.mam/...`, and only then the cwd.
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import uuid
from pathlib import Path

import _mam_state_guard

REPO_ROOT = Path(__file__).resolve().parent.parent
SKILLS = REPO_ROOT / ".agents" / "skills"

sys.path.insert(0, str(SKILLS))
from lib_py.delegation import mqtt_common  # noqa: E402


def _listing(d: Path) -> set:
    return set(os.listdir(d)) if d.is_dir() else set()


# ---------------------------------------------------------------------------
# resolution order
# ---------------------------------------------------------------------------
def test_registry_dir_env_wins(monkeypatch, tmp_path):
    monkeypatch.setenv("MAM_JOBS_DIR", str(tmp_path / "explicit"))
    monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
    assert mqtt_common.default_registry_dir() == str(tmp_path / "explicit")


def test_registry_dir_follows_workspace_root(monkeypatch, tmp_path):
    monkeypatch.delenv("MAM_JOBS_DIR", raising=False)
    monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
    assert mqtt_common.default_registry_dir() == str(tmp_path / ".mam" / "jobs")


def test_registry_dir_falls_back_to_cwd_relative(monkeypatch, tmp_path):
    monkeypatch.delenv("MAM_JOBS_DIR", raising=False)
    monkeypatch.delenv("WORKSPACE_ROOT", raising=False)
    assert mqtt_common.default_registry_dir() == ".mam/jobs"
    # A WORKSPACE_ROOT that is not a directory is ignored, not trusted.
    monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path / "does-not-exist"))
    assert mqtt_common.default_registry_dir() == ".mam/jobs"


def test_logs_dir_follows_the_same_order(monkeypatch, tmp_path):
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
    assert mqtt_common.get_logs_dir() == str(tmp_path / "logs")
    monkeypatch.delenv("DELEGATE_JOB_LOGS_DIR")
    assert mqtt_common.get_logs_dir() == str(tmp_path / ".mam" / "delegate_job_logs")
    monkeypatch.delenv("WORKSPACE_ROOT")
    assert mqtt_common.get_logs_dir() == os.path.join(os.getcwd(), ".mam", "delegate_job_logs")


def test_cli_defaults_are_resolved_per_process(monkeypatch, tmp_path):
    """registry.py / publish_event.py build their argparse defaults at call
    time, so a process started with WORKSPACE_ROOT set registers there - and
    seeds its audit log there too."""
    monkeypatch.delenv("MAM_JOBS_DIR", raising=False)
    monkeypatch.delenv("DELEGATE_JOB_LOGS_DIR", raising=False)
    env = dict(os.environ, WORKSPACE_ROOT=str(tmp_path), PYTHONPATH=f"{SKILLS}:{os.environ.get('PYTHONPATH', '')}")
    env.pop("MAM_JOBS_DIR", None)
    res = subprocess.run(
        [sys.executable, "-m", "lib_py.delegation.cli", "submit", "--prompt", "issue 15", "--agent", "claude-code",
         "--agent-session", "herdr:issue15-probe"],
        capture_output=True, text=True, env=env, cwd=str(REPO_ROOT), timeout=60,
    )
    assert res.returncode == 0, res.stderr
    records = list((tmp_path / ".mam" / "jobs").glob("*.json"))
    assert len(records) == 1, records
    assert json.loads(records[0].read_text())["agent_session"] == "herdr:issue15-probe"
    assert (tmp_path / ".mam" / "delegate_job_logs" / records[0].stem).is_dir()


# ---------------------------------------------------------------------------
# the leak itself: create_session.sh's onboarding job from the repo root cwd
# ---------------------------------------------------------------------------
def _create_session_from_repo_root(sandbox: Path, session: str) -> subprocess.CompletedProcess:
    script = sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
    # cwd deliberately the repo root — the exact shape of the leaking tests.
    return subprocess.run(
        ["bash", str(script), "--workspace", str(sandbox), "--agent", "claude", "--role", "Creator",
         "--session", session],
        capture_output=True, text=True, cwd=str(REPO_ROOT), timeout=180,
    )


def _real_state_dirs(recorded: dict) -> list:
    """Every real registry and audit-log root this run could leak into: this
    checkout's .mam and the control plane the ambient binding named (conftest
    records it before dropping it)."""
    jobs, logs = _mam_state_guard.state_dirs(_mam_state_guard.watch_roots(REPO_ROOT, recorded), recorded)
    return jobs + logs


def _entries(dirs: list) -> set:
    return {(d, name) for d in dirs for name in _listing(Path(d))}


def _unique(prefix: str) -> str:
    """A session name no concurrent run shares (Issue #20)."""
    return f"{prefix}-{uuid.uuid4().hex[:8]}"


def _assert_registered_in_sandbox_only(sandbox: Path, session: str, real_dirs: list, before: set, res):
    """Judged on this test's own job ids and session only (Issue #20): live
    delegations and other test runs may add entries to the real .mam
    meanwhile, so comparing whole listings gave false failures."""
    assert res.returncode == 0, res.stderr + res.stdout
    records = {p.stem: json.loads(p.read_text()) for p in (sandbox / ".mam" / "jobs").glob("*.json")}
    own = [jid for jid, r in records.items() if r.get("agent_session") == f"herdr:{session}"]
    assert own, records
    for d in real_dirs:
        for jid in own:
            assert not (Path(d) / f"{jid}.json").exists(), f"onboarding job {jid} leaked into {d}"
            assert not (Path(d) / jid).exists(), f"audit log {jid} leaked into {d}"
    # A job that never reached the sandbox has an id we do not know: look for
    # the session in whatever appeared in the real roots meanwhile.
    for d, name in _entries(real_dirs) - before:
        entry = Path(d) / name
        meta = entry if name.endswith(".json") else entry / "meta.json"
        try:
            session_of = json.loads(meta.read_text()).get("agent_session")
        except (OSError, ValueError):
            continue
        assert session_of != f"herdr:{session}", f"onboarding job for {session} leaked: {entry}"


def test_onboarding_job_stays_in_sandbox_with_explicit_env(mam_sandbox, mock_herdr, mock_agents,
                                                           _ambient_control_plane):
    session = _unique("issue15-explicit-creator-claude")
    real_dirs = _real_state_dirs(_ambient_control_plane)
    before = _entries(real_dirs)
    res = _create_session_from_repo_root(mam_sandbox, session)
    _assert_registered_in_sandbox_only(mam_sandbox, session, real_dirs, before, res)


def test_onboarding_job_follows_workspace_root_alone(mam_sandbox, mock_herdr, mock_agents, monkeypatch,
                                                     _ambient_control_plane):
    """Without the explicit MAM_JOBS_DIR / DELEGATE_JOB_LOGS_DIR overrides the
    WORKSPACE_ROOT that lib.sh exports must be enough to keep the record out
    of the repo's registry."""
    monkeypatch.delenv("MAM_JOBS_DIR", raising=False)
    monkeypatch.delenv("DELEGATE_JOB_LOGS_DIR", raising=False)
    session = _unique("issue15-wsroot-creator-claude")
    real_dirs = _real_state_dirs(_ambient_control_plane)
    before = _entries(real_dirs)
    res = _create_session_from_repo_root(mam_sandbox, session)
    _assert_registered_in_sandbox_only(mam_sandbox, session, real_dirs, before, res)