multi-agent-mux/tests/test_issue55_agent_homes.py

"""Issue #55: tests never write into the user's real agent homes.

- L: the variable list conftest._isolate_agent_homes drops is pinned against
  the code - every `*_DIR` / `*_HOME` / `HOME` / `LOCAL_BIN` variable the
  adapters, lib_py/paths.py and lib.sh read from the environment is either
  dropped by the fixture, left alone on purpose, or a control-plane path
  isolated by _isolate_mam_state_dirs. A new path variable in an adapter
  fails here.
- F: what the fixture does, and that mam_sandbox still wins.
- G: the session guard's helpers (tests/_agent_home_guard.py).
- R: the issue's reproduction - with CLAUDE_PROJECT_DIR set like an agent
  pane's, test_adapter_discover passes and writes nothing there.
"""
from __future__ import annotations

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

import pytest

import _agent_home_guard as guard

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))

# Control-plane paths lib.sh reads the same way; every test gets its own from
# _isolate_mam_state_dirs (Issue #20), and SKS_REGISTRY_DIR is what the
# delegate-job wrapper exports for the job it is dispatching.
CONTROL_PLANE = {"MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR", "SKS_REGISTRY_DIR"}
_PATH_LIKE = re.compile(r"(?:_DIR|_HOME|^HOME|LOCAL_BIN)$")
_PY_READ = re.compile(r"""os\.(?:environ(?:\.get)?\s*[\[(]|getenv\()\s*['"]([A-Z_][A-Z0-9_]*)['"]""")
_SH_OVERRIDE = re.compile(r"\$\{([A-Z_][A-Z0-9_]*)(?::-|-|:\+|:=)")
_SH_HOME = re.compile(r"\$\{?HOME\}?(?![A-Z0-9_])")


def _path_vars_read_by_the_code():
    found = set()
    files = list((SKILLS / "lib_py" / "agents").rglob("*.py")) + [SKILLS / "lib_py" / "paths.py"]
    for f in files:
        found |= set(_PY_READ.findall(f.read_text(encoding="utf-8")))
    lib = (SKILLS / "lib.sh").read_text(encoding="utf-8")
    found |= set(_SH_OVERRIDE.findall(lib))
    if _SH_HOME.search(lib):
        found.add("HOME")
    return {v for v in found if _PATH_LIKE.search(v)}


# ---------------------------------------------------------------------------
# L - the fixture's list is the code's list
# ---------------------------------------------------------------------------

def test_l1_every_path_variable_the_code_reads_is_accounted_for():
    read = _path_vars_read_by_the_code()
    assert {"CLAUDE_PROJECT_DIR", "HOME_DIR", "CODEX_HOME", "HOME"} <= read, read  # the scan sees the known ones
    accounted = set(guard.DROPPED_VARS) | set(guard.LEFT_ALONE) | CONTROL_PLANE
    assert read - accounted == set(), (
        "path variables the adapters/lib.sh read but conftest._isolate_agent_homes does not drop: "
        f"{sorted(read - accounted)}")
    assert set(guard.DROPPED_VARS) <= read, f"dropped but no longer read: {sorted(set(guard.DROPPED_VARS) - read)}"


def test_l2_the_scan_sees_each_read_idiom(tmp_path, monkeypatch):
    py = tmp_path / "a.py"
    py.write_text('x = os.environ.get("A_DIR") or os.environ["B_HOME"] or os.getenv(\'C_DIR\')\n'
                  'y = os.environ.get("NOT_A_PATH")\n')
    sh = tmp_path / "lib.sh"
    sh.write_text('D_DIR="${D_DIR:-$HOME/x}"\nE="${E_HOME-}"\necho $SKILL_DIR ${F_DIR}\n')
    monkeypatch.setattr(sys.modules[__name__], "SKILLS", tmp_path)
    (tmp_path / "lib_py" / "agents").mkdir(parents=True)
    py.rename(tmp_path / "lib_py" / "agents" / "a.py")
    (tmp_path / "lib_py" / "paths.py").write_text("")
    assert _path_vars_read_by_the_code() == {"A_DIR", "B_HOME", "C_DIR", "D_DIR", "E_HOME", "HOME"}


# ---------------------------------------------------------------------------
# F - the fixture
# ---------------------------------------------------------------------------

def test_f1_every_test_runs_without_the_panes_agent_home_variables(tmp_path):
    for var in guard.DROPPED_VARS:
        assert var not in os.environ, (var, os.environ[var])
    assert not os.environ["HOME"].startswith(str(tmp_path)), "HOME is left alone (herdr sockets live under it)"


def test_f2_discovery_context_follows_the_home_dir_it_is_given(tmp_path):
    """The issue's mechanism: DiscoveryContext prefers CLAUDE_PROJECT_DIR over
    the home_dir it is given. With the pane's value dropped, home_dir wins."""
    from lib_py.agents.base import DiscoveryContext
    ctx = DiscoveryContext(workspace=str(tmp_path / "ws"), agent_name="claude", home_dir=str(tmp_path / "home"))
    assert ctx.claude_dir == str(tmp_path / "home" / ".claude" / "projects")


def test_f3_mam_sandbox_still_wins(mam_sandbox):
    assert os.environ["CLAUDE_PROJECT_DIR"] == str(mam_sandbox / ".claude" / "projects")
    assert os.environ["HOME_DIR"] == str(mam_sandbox)
    assert os.environ["HOME"] == str(mam_sandbox)


# ---------------------------------------------------------------------------
# G - the guard's helpers
# ---------------------------------------------------------------------------

def test_g1_watch_dirs_follow_the_ambient_environment(tmp_path):
    home = tmp_path / "h"
    dirs = dict(guard.watch_dirs({"HOME": str(home)}))
    assert dirs["claude"] == str(home / ".claude" / "projects")
    assert dirs["codex"] == str(home / ".codex" / "sessions")
    assert dirs["hermes"] == str(home / ".hermes")
    # explicit variables win over HOME, HOME_DIR over HOME
    named = dict(guard.watch_dirs({"HOME": "/nowhere", "HOME_DIR": str(home),
                                   "CLAUDE_PROJECT_DIR": str(tmp_path / "cp"), "CODEX_HOME": str(tmp_path / "cx")}))
    assert named["claude"] == str(tmp_path / "cp")
    assert named["codex"] == str(tmp_path / "cx" / "sessions")
    assert named["grok"] == str(home / ".grok" / "sessions")
    assert [a for a, _ in guard.watch_dirs({})].count("agy") == 5


def test_g2_leaks_are_the_new_entries_and_a_missing_dir_is_empty(tmp_path):
    claude = tmp_path / "claude"
    grok = tmp_path / "grok"
    claude.mkdir()
    (claude / "old").mkdir()
    dirs = [("claude", str(claude)), ("grok", str(grok))]
    before = guard.snapshot(dirs)
    assert before == {str(claude): {"old"}, str(grok): set()}
    assert guard.leaks(dirs, before, str(claude)) == []
    (claude / "-tmp-made-by-a-test-ws").mkdir()
    grok.mkdir()
    (grok / "s1.json").write_text("{}")
    found = guard.leaks(dirs, before, str(claude), tmp_roots=[str(tmp_path)])
    assert found == [{"path": str(claude / "-tmp-made-by-a-test-ws"), "agent": "claude"},
                     {"path": str(grok / "s1.json"), "agent": "grok"}]
    msg = guard.describe(found)
    assert "Issue #55: 2 entries" in msg and str(grok / "s1.json") in msg and "[claude]" in msg


def test_g3_decode_ws_key_reads_the_key_against_the_file_system(tmp_path):
    from lib_py.verify_session import workspace_key
    ws = tmp_path / "my_proj-x" / "sub_dir"
    ws.mkdir(parents=True)
    key = workspace_key(str(ws))
    assert guard.decode_ws_key(key) == os.path.realpath(ws)
    assert guard.decode_ws_key(key + "-gone") is None
    assert guard.decode_ws_key("not-a-key") is None


def test_g4_the_claude_exception_passes_an_operating_agents_folder_only(tmp_path, monkeypatch):
    from lib_py.verify_session import workspace_key
    claude = tmp_path / "claude"
    claude.mkdir()
    dirs = [("claude", str(claude))]
    before = guard.snapshot(dirs)
    # a folder for a real directory outside every temp root: an operating agent's
    live = tmp_path / "live-ws"
    live.mkdir()
    monkeypatch.setattr(guard.tempfile, "gettempdir", lambda: str(tmp_path / "systmp"))
    (claude / workspace_key(str(live))).mkdir()
    assert guard.leaks(dirs, before, str(claude)) == []
    # the same name under a temp root (pytest's basetemp here) is a test's
    assert [f["path"] for f in guard.leaks(dirs, before, str(claude), tmp_roots=[str(tmp_path)])] == \
        [str(claude / workspace_key(str(live)))]
    # a folder whose name decodes to no directory on this host is a test's too
    (claude / "-nowhere-on-this-host-ws").mkdir()
    assert [f["path"] for f in guard.leaks(dirs, before, str(claude))] == [str(claude / "-nowhere-on-this-host-ws")]
    # the exception never applies to another agent's directory
    grok = tmp_path / "grok"
    grok.mkdir()
    dirs2 = [("grok", str(grok))]
    before2 = guard.snapshot(dirs2)
    (grok / workspace_key(str(live))).mkdir()
    assert len(guard.leaks(dirs2, before2, str(claude))) == 1


def test_g5_the_session_guard_is_wired_to_the_recorded_ambient_environment():
    src = (REPO_ROOT / "tests" / "conftest.py").read_text(encoding="utf-8")
    assert "def agent_home_leak_guard(_ambient_agent_homes, tmp_path_factory)" in src
    assert "_RECORDED_AGENT_HOMES.update" in src.split("def pytest_configure")[1].split("def pytest_unconfigure")[0]


# ---------------------------------------------------------------------------
# R - the reproduction from the issue, now a regression
# ---------------------------------------------------------------------------

def test_r1_adapter_discover_writes_nothing_into_a_panes_claude_project_dir(tmp_path):
    """Issue #55 as reported: CLAUDE_PROJECT_DIR set like an agent pane's. On
    base 42e5173 this run left `<real-like>/-…-test-adapter-discover0-ws/u-c1.jsonl`."""
    real_like = tmp_path / "real-like"
    real_like.mkdir()
    env = dict(os.environ, CLAUDE_PROJECT_DIR=str(real_like))
    for k in ("HERDR_WORKSPACE_ID", "HERDR_PANE_ID", "HERDR_TAB_ID", "HERDR_SOCKET_PATH"):
        env.pop(k, None)
    res = subprocess.run([sys.executable, "-m", "pytest", "-p", "no:cacheprovider", "-q",
                          "--basetemp", str(tmp_path / "bt"),
                          str(REPO_ROOT / "tests" / "test_a4_adapter_contract.py"), "-k", "adapter_discover"],
                         capture_output=True, text=True, cwd=str(REPO_ROOT), env=env, timeout=600)
    assert res.returncode == 0, res.stdout[-3000:] + res.stderr[-2000:]
    assert "1 passed" in res.stdout, res.stdout[-1000:]
    assert sorted(real_like.rglob("*")) == [], [str(p) for p in real_like.rglob("*")]