tests/_agent_home_guard.py

@@ -0,0 +1,125 @@
+"""Issue #55 - helpers for the session guard over the user's real agent homes.
+
+An agent pane inherits CLAUDE_PROJECT_DIR=<real ~/.claude/projects> from
+lib.sh (every MAM pane does), and DiscoveryContext prefers that variable over
+the home_dir a test passes, so a suite run by an agent wrote fake conversation
+folders into the user's real claude directory. The per-test fixture
+(conftest._isolate_agent_homes) is the fix; this guard is the alarm: it
+snapshots the direct entries of every real conversation directory before the
+suite and reports the new ones after it. Read-only - it never hashes, touches
+or removes anything there. Pure functions, unit-tested in
+test_issue55_agent_homes.py; the name does not start with `test_`, so pytest
+never collects it.
+"""
+from __future__ import annotations
+
+import os
+import tempfile
+
+# The pane-inherited variables conftest._isolate_agent_homes drops for every
+# test, so HOME (which the tests set) governs, as it does in a plain shell.
+DROPPED_VARS = ("HOME_DIR", "CLAUDE_PROJECT_DIR", "CODEX_HOME", "LOCAL_BIN")
+# Read by the code under test but left alone, on purpose: HOME is where the
+# real herdr tests and the shim bootstrap find the server socket, and it is
+# what the suite's own tests set for an isolated home; XDG_CONFIG_HOME and
+# HERDR_CONFIG_PATH are herdr's own config locations (mam_sandbox unsets
+# them, Issue #45).
+LEFT_ALONE = ("HOME", "XDG_CONFIG_HOME", "HERDR_CONFIG_PATH")
+
+
+def real_home(ambient):
+    return ambient.get("HOME_DIR") or ambient.get("HOME") or os.path.expanduser("~")
+
+
+def real_claude_dir(ambient):
+    return ambient.get("CLAUDE_PROJECT_DIR") or os.path.join(real_home(ambient), ".claude", "projects")
+
+
+def watch_dirs(ambient):
+    """[(agent, directory)]: where each adapter reads and writes conversations,
+    as the ambient environment (recorded before any fixture ran) names them."""
+    home = real_home(ambient)
+    codex = ambient.get("CODEX_HOME") or os.path.join(home, ".codex")
+    dirs = [("claude", real_claude_dir(ambient))]
+    dirs += [("agy", os.path.join(home, ".gemini", "antigravity-cli", d))
+             for d in ("conversations", "cache", "brain", "annotations", "presence")]
+    dirs += [("codex", os.path.join(codex, "sessions")),
+             ("grok", os.path.join(home, ".grok", "sessions")),
+             ("opencode", os.path.join(home, ".local", "share", "opencode")),
+             ("hermes", os.path.join(home, ".hermes"))]
+    return dirs
+
+
+def _listdir(path):
+    try:
+        return set(os.listdir(path))
+    except OSError:
+        return set()
+
+
+def snapshot(dirs):
+    """Direct entry names per watched directory; a missing one counts as empty."""
+    return {d: _listdir(d) for _agent, d in dirs}
+
+
+def decode_ws_key(name, root="/"):
+    """The existing directory whose workspace key (lib_py.verify_session:
+    realpath with `/` and `_` turned into `-`) is `name`, or None. A key is
+    ambiguous on its own (a `-` may have been a `/`, a `_` or a `-`), so it is
+    read against the file system: from `root`, each component is an existing
+    entry whose own key is the next piece of the name."""
+    if not name.startswith("-"):
+        return None
+    stack = [(root, name[1:])]
+    while stack:
+        cur, rest = stack.pop()
+        if rest == "":
+            return cur
+        try:
+            entries = os.listdir(cur)
+        except OSError:
+            continue
+        for entry in entries:
+            key = entry.replace("_", "-")
+            if rest == key:
+                full = os.path.join(cur, entry)
+                if os.path.isdir(full):
+                    return full
+            elif rest.startswith(key + "-"):
+                stack.append((os.path.join(cur, entry), rest[len(key) + 1:]))
+    return None
+
+
+def _under(path, roots):
+    real = os.path.realpath(path)
+    return any(real == r or real.startswith(r + os.sep) for r in roots)
+
+
+def leaks(dirs, before, claude_dir, tmp_roots=()):
+    """New entries since `before`, as [{"path", "agent"}] sorted by path.
+
+    One exception, for the claude directory only: an entry whose name decodes
+    to a directory that exists on this host and is not under a temp root
+    (tempfile.gettempdir() and the given `tmp_roots`, e.g. pytest's basetemp)
+    is a conversation an operating agent started meanwhile, not a test's.
+    Every other directory has no exception."""
+    roots = [os.path.realpath(tempfile.gettempdir())] + [os.path.realpath(r) for r in tmp_roots]
+    claude_real = os.path.realpath(claude_dir)
+    now = snapshot(dirs)
+    found = []
+    for agent, d in dirs:
+        for name in sorted(now[d] - before.get(d, set())):
+            if os.path.realpath(d) == claude_real:
+                ws = decode_ws_key(name)
+                if ws is not None and not _under(ws, roots):
+                    continue
+            found.append({"path": os.path.join(d, name), "agent": agent})
+    return sorted(found, key=lambda f: f["path"])
+
+
+def describe(found):
+    """Failure message for the session guard."""
+    lines = ["Issue #55: {} entr{} made during this session in a real agent home "
+             "(left in place; check and remove by hand):".format(len(found), "y" if len(found) == 1 else "ies")]
+    lines += ["  {path}  [{agent}]".format(**f) for f in found]
+    return "\n".join(lines)

tests/conftest.py

@@ -130,15 +130,24 @@ _TESTS_DIR = str(Path(__file__).resolve().parent)
 if _TESTS_DIR not in sys.path:
     sys.path.insert(0, _TESTS_DIR)
 import _mam_state_guard  # noqa: E402
+import _agent_home_guard  # noqa: E402
 
 _BINDING_VARS = ("WORKSPACE_ROOT", "AGENT_SESSIONS_YAML", "MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR")
 _RECORDED_BINDING = {}
+# Issue #55: the agent-home variables as the pane had them, recorded before
+# any fixture points them into a tmp_path, so the session guard knows which
+# real directories to watch. They are not dropped: every test gets its own
+# values from _isolate_agent_homes.
+_AGENT_HOME_VARS = ("HOME",) + _agent_home_guard.DROPPED_VARS
+_RECORDED_AGENT_HOMES = {}
 _DROPPED_BINDING = pytest.StashKey[dict]()
 
 
 def pytest_configure(config):
     _RECORDED_BINDING.clear()
     _RECORDED_BINDING.update({k: os.environ[k] for k in _BINDING_VARS if os.environ.get(k)})
+    _RECORDED_AGENT_HOMES.clear()
+    _RECORDED_AGENT_HOMES.update({k: os.environ[k] for k in _AGENT_HOME_VARS if os.environ.get(k)})
     dropped = {k: os.environ.pop(k) for k in ("AGENT_SESSIONS_YAML", "MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR")
                if k in os.environ}
     ws = os.environ.get("WORKSPACE_ROOT")
@@ -177,6 +186,30 @@ def _isolate_mam_state_dirs(_ambient_control_plane, tmp_path, monkeypatch):
     monkeypatch.setenv("MAM_ENV_FILE", str(state / ".mam.env.absent"))
 
 
+@pytest.fixture(scope="session")
+def _ambient_agent_homes():
+    """The agent-home variables pytest_configure recorded, as a copy."""
+    return dict(_RECORDED_AGENT_HOMES)
+
+
+@pytest.fixture(autouse=True)
+def _isolate_agent_homes(_isolate_mam_state_dirs, monkeypatch):
+    """Issue #55: the agent-home variables an agent pane carries (lib.sh
+    exports HOME_DIR, CLAUDE_PROJECT_DIR and LOCAL_BIN into every pane, and
+    the adapters read CODEX_HOME) are dropped for every test. DiscoveryContext
+    prefers CLAUDE_PROJECT_DIR over the home_dir a test passes, and every
+    adapter path reads HOME_DIR before HOME, so with the pane's values a test
+    that isolates itself through `home_dir` or HOME still read and wrote the
+    user's real directories. Dropped rather than pointed into tmp_path: the
+    suite's tests isolate a home by setting HOME (or HOME_DIR themselves),
+    and a fixture-set HOME_DIR would shadow every one of them the same way.
+    HOME stays: the real herdr tests and the shim bootstrap find the server
+    socket under it. mam_sandbox and a test's own setenv run later and win.
+    test_issue55_agent_homes pins the variable list against the code."""
+    for var in _agent_home_guard.DROPPED_VARS:
+        monkeypatch.delenv(var, raising=False)
+
+
 @pytest.fixture
 def own_workspace_root(tmp_path, monkeypatch):
     """Issue #17: every `source lib.sh` rewrites $WORKSPACE_ROOT/.mam/shim/herdr,
@@ -1512,6 +1545,26 @@ def mam_state_leak_guard(_ambient_control_plane):
     assert not found, _mam_state_guard.describe(found)
 
 
+@pytest.fixture(scope="session", autouse=True)
+def agent_home_leak_guard(_ambient_agent_homes, tmp_path_factory):
+    """Issue #55: no test may leave an entry in the user's real agent homes -
+    the claude projects directory and the agy, codex, grok, opencode and
+    hermes conversation directories the ambient environment names. The
+    per-test fixture is the fix, this is the regression alarm; it reads the
+    directories' entry names and never hashes or deletes. A claude folder an
+    operating agent started meanwhile (its name decodes to a real directory
+    outside the temp roots) is not counted; a new conversation file an
+    operating agent writes into one of the other directories would be a
+    false positive, which is rare (an existing conversation writes to its
+    existing file)."""
+    dirs = _agent_home_guard.watch_dirs(_ambient_agent_homes)
+    before = _agent_home_guard.snapshot(dirs)
+    yield
+    found = _agent_home_guard.leaks(dirs, before, _agent_home_guard.real_claude_dir(_ambient_agent_homes),
+                                    tmp_roots=[str(tmp_path_factory.getbasetemp())])
+    assert not found, _agent_home_guard.describe(found)
+
+
 @pytest.fixture(scope="session", autouse=True)
 def subscriber_daemon_leak_guard(tmp_path_factory):
     """Issue #19: no `reconcile.sh --subscribe` daemon anchored in this session's

tests/test_issue55_agent_homes.py

@@ -0,0 +1,203 @@
+"""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("*")]