From c5844a8e12e4e8d59507a1ec26f7984a25f8866a Mon Sep 17 00:00:00 2001 From: Godopu Date: Tue, 15 Sep 2026 02:02:52 +0000 Subject: [PATCH] test(issue-20): session guard for new entries in the recorded control planes' .mam/jobs and .mam/delegate_job_logs Watches this checkout's .mam and the ones the recorded binding named. A new entry is a leak when it is an audit log without a registry record, or when its agent_session is missing or not a registered session; live delegations write the record first and target a registered session. Reports, never deletes. The #19 guard's docstring pointed at a #15 session guard that never existed. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Sdga4A8i1iSc8DaaW9iKU2 --- tests/_mam_state_guard.py | 157 ++++++++++++++++++++++++++ tests/conftest.py | 23 +++- tests/test_issue20_state_isolation.py | 103 +++++++++++++++++ 3 files changed, 282 insertions(+), 1 deletion(-) create mode 100644 tests/_mam_state_guard.py create mode 100644 tests/test_issue20_state_isolation.py diff --git a/tests/_mam_state_guard.py b/tests/_mam_state_guard.py new file mode 100644 index 0000000..e90d23b --- /dev/null +++ b/tests/_mam_state_guard.py @@ -0,0 +1,157 @@ +"""Issue #20 - helpers for the session guard over real control planes. + +A pytest process can reach two kinds of `.mam`: this checkout's own, and the +one the agent pane's control-plane binding named before conftest's +pytest_configure dropped it (a worktree creator's pane carries +WORKSPACE_ROOT, AGENT_SESSIONS_YAML, MAM_JOBS_DIR and DELEGATE_JOB_LOGS_DIR, +all at the live main checkout). The session guard snapshots their job +registries and audit-log roots and, at the end, reports the new entries no +live delegation explains. Pure functions, unit-tested in +test_issue20_state_isolation.py (S4, S5); the name does not start with +`test_`, so pytest never collects it. +""" +from __future__ import annotations + +import json +import os +import re + +SESSION_PREFIXES = ("herdr:", "remote:") +# Top-level items of `herdr_sessions:` as atomic_dump_yaml writes them +# (`- name: ` at column 0); indented `- name:` lines belong to nested +# lists such as mcp_attachments. A regex keeps the guard free of PyYAML. +_SESSION_LINE = re.compile(r"""^- name:\s*['"]?([^'"\s]+)""") + + +def _listdir(path): + try: + return os.listdir(path) + except OSError: + return [] + + +def _load(path): + try: + with open(path, encoding="utf-8") as fh: + data = json.load(fh) + except (OSError, ValueError): + return {} + return data if isinstance(data, dict) else {} + + +def watch_roots(repo_root, recorded): + """`.mam` directories to watch, realpath-deduplicated, in order: this + checkout's own (always, so a leak that creates it is still seen), then + those the recorded binding names - `$WORKSPACE_ROOT/.mam` and the parents + of MAM_JOBS_DIR / DELEGATE_JOB_LOGS_DIR - when they exist.""" + roots = [os.path.realpath(os.path.join(str(repo_root), ".mam"))] + candidates = [] + if recorded.get("WORKSPACE_ROOT"): + candidates.append(os.path.join(recorded["WORKSPACE_ROOT"], ".mam")) + for var in ("MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR"): + if recorded.get(var): + candidates.append(os.path.dirname(os.path.abspath(recorded[var]))) + for path in candidates: + real = os.path.realpath(path) + if os.path.isdir(real) and real not in roots: + roots.append(real) + return roots + + +def state_dirs(roots, recorded): + """(registry dirs, audit-log dirs): `/jobs` and + `/delegate_job_logs` of every root, plus a recorded MAM_JOBS_DIR / + DELEGATE_JOB_LOGS_DIR that is not one of those.""" + jobs = [os.path.join(r, "jobs") for r in roots] + logs = [os.path.join(r, "delegate_job_logs") for r in roots] + for var, dirs in (("MAM_JOBS_DIR", jobs), ("DELEGATE_JOB_LOGS_DIR", logs)): + if recorded.get(var): + real = os.path.realpath(recorded[var]) + if os.path.isdir(real) and real not in dirs: + dirs.append(real) + return jobs, logs + + +def snapshot(roots, recorded): + """Registry records (`*.json`) and audit-log entries per watched dir; a + missing dir counts as empty.""" + jobs, logs = state_dirs(roots, recorded) + return { + "jobs": {d: {n for n in _listdir(d) if n.endswith(".json")} for d in jobs}, + "logs": {d: set(_listdir(d)) for d in logs}, + } + + +def session_names(yaml_path): + """Session names registered in an agent-sessions.yaml; empty if unreadable.""" + try: + with open(yaml_path, encoding="utf-8", errors="replace") as fh: + return {m.group(1) for m in map(_SESSION_LINE.match, fh) if m} + except OSError: + return set() + + +def _bare(session): + for prefix in SESSION_PREFIXES: + if session.startswith(prefix): + return session[len(prefix):] + return session + + +def leaks(roots, before, recorded): + """New entries since `before` that no live delegation explains. + + A real delegation writes its registry record before its audit log and + targets a registered session (registry.register_job), so a new entry is a + leak when + (a) it is an audit log whose job has no record in any watched registry, or + (b) its record's agent_session is missing or names no session in the + watched agent-sessions.yaml files. + Returns [{"path", "reason", "agent_session", "prompt"}] sorted by path.""" + now = snapshot(roots, recorded) + yamls = [os.path.join(r, "agent-sessions.yaml") for r in roots] + if recorded.get("AGENT_SESSIONS_YAML"): + yamls.append(recorded["AGENT_SESSIONS_YAML"]) + known = set() + for path in yamls: + known |= session_names(path) + records = {} + for d, names in now["jobs"].items(): + for name in names: + records.setdefault(name[:-len(".json")], os.path.join(d, name)) + + found = [] + + def judge(path, meta, reason=None): + session = meta.get("agent_session") + if reason is None: + if not session: + reason = "no agent_session" + elif _bare(session) not in known: + reason = "agent_session names no registered session" + else: + return + found.append({"path": path, "reason": reason, "agent_session": session, + "prompt": str(meta.get("prompt", ""))[:60]}) + + for d, names in now["jobs"].items(): + for name in names - before["jobs"].get(d, set()): + path = os.path.join(d, name) + judge(path, _load(path)) + for d, names in now["logs"].items(): + for name in names - before["logs"].get(d, set()): + path = os.path.join(d, name) + if name in records: + judge(path, _load(records[name])) + else: + judge(path, _load(os.path.join(path, "meta.json")), "audit log without a registry record") + return sorted(found, key=lambda f: f["path"]) + + +def describe(found): + """Failure message for the session guard.""" + lines = ["Issue #20: {} entr{} made during this session in a real control plane " + "(left in place; check and remove by hand):".format(len(found), "y" if len(found) == 1 else "ies")] + lines += [" {path} [{reason}] agent_session={agent_session!r} prompt={prompt!r}".format(**f) + for f in found] + return "\n".join(lines) diff --git a/tests/conftest.py b/tests/conftest.py index 25e6531..511f2a5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -126,6 +126,10 @@ def _pid_alive(pid): # one) reports new entries in the recorded control planes. # ============================================================================== _REPO_ROOT = Path(__file__).resolve().parent.parent +_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 _BINDING_VARS = ("WORKSPACE_ROOT", "AGENT_SESSIONS_YAML", "MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR") _RECORDED_BINDING = {} @@ -1440,10 +1444,27 @@ def _drop_ambient_mam_version(): os.environ["MAM_VERSION"] = previous +@pytest.fixture(scope="session", autouse=True) +def mam_state_leak_guard(_ambient_control_plane): + """Issue #20: no test may leave a record in a real `.mam/jobs` or an audit + log in a real `.mam/delegate_job_logs` - this checkout's, or the control + plane the ambient binding named before pytest_configure dropped it. + Delegations the live system runs meanwhile are told apart by + _mam_state_guard.leaks (record first, registered session). The per-test + defaults are the fix, this is the regression alarm; it reports and never + deletes - these are real control planes. The session registry yaml/db is + not hashed: reconcile and create write it during a run.""" + roots = _mam_state_guard.watch_roots(_REPO_ROOT, _ambient_control_plane) + before = _mam_state_guard.snapshot(roots, _ambient_control_plane) + yield + found = _mam_state_guard.leaks(roots, before, _ambient_control_plane) + assert not found, _mam_state_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 - pytest basetemp may survive the suite. Mirrors the #15 registry guard: the + pytest basetemp may survive the suite. Mirrors the #20 state guard: the per-sandbox finalizer is the fix, this is the regression alarm. Leftovers are reaped after being reported so a failing run still cleans up.""" yield diff --git a/tests/test_issue20_state_isolation.py b/tests/test_issue20_state_isolation.py new file mode 100644 index 0000000..b2fbef2 --- /dev/null +++ b/tests/test_issue20_state_isolation.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +"""Issue #20 - test state stays out of every real control plane. + +conftest records the control-plane binding an agent pane carries and drops it +before collection (pytest_configure), gives every test a registry and an +audit-log root under its own tmp_path (_isolate_mam_state_dirs), and a +session guard reports new entries in a real `.mam/jobs` or +`.mam/delegate_job_logs` (mam_state_leak_guard, helpers in _mam_state_guard). +""" +from __future__ import annotations + +import hashlib +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +import _mam_state_guard as guard + +REPO_ROOT = Path(__file__).resolve().parent.parent +SKILLS = REPO_ROOT / ".agents" / "skills" + + +def _fake_control_plane(root: Path, sessions=()) -> Path: + """A checkout-shaped tree with a `.mam` registry, audit-log root and session yaml.""" + mam = root / ".mam" + (mam / "jobs").mkdir(parents=True) + (mam / "delegate_job_logs").mkdir() + rows = "".join(f"- name: {s}\n status: running\n mcp_attachments:\n - name: stitch\n" for s in sessions) + (mam / "agent-sessions.yaml").write_text("herdr_sessions:\n" + rows if rows else "herdr_sessions: []\n") + return root + + +def _record(mam: Path, jid: str, session) -> None: + rec = {"job_id": jid, "prompt": f"prompt of {jid}"} + if session: + rec["agent_session"] = session + (mam / "jobs" / f"{jid}.json").write_text(json.dumps(rec)) + + +def _log(mam: Path, jid: str, session=None) -> None: + d = mam / "delegate_job_logs" / jid + d.mkdir() + if session: + (d / "meta.json").write_text(json.dumps({"job_id": jid, "agent_session": session})) + + +# --------------------------------------------------------------------------- +# the session guard (S4, S5) +# --------------------------------------------------------------------------- +def test_s4_guard_flags_only_entries_no_live_delegation_explains(tmp_path): + main = _fake_control_plane(tmp_path / "main", sessions=("creator-real",)) + mam = main / ".mam" + _record(mam, "old00001", "herdr:nobody") # before the snapshot: not judged + _log(mam, "old00001") + recorded = {"WORKSPACE_ROOT": str(main)} + roots = guard.watch_roots(tmp_path / "repo", recorded) + before = guard.snapshot(roots, recorded) + + _record(mam, "live0001", "herdr:creator-real") # a live delegation: record first, registered session + _log(mam, "live0001", "herdr:creator-real") + _record(mam, "live0002", "remote:creator-real") + _log(mam, "orphan01", "herdr:creator-real") # (a) audit log with no record, even for a known session + _record(mam, "stray001", "herdr:nobody") # (b) unknown session + _record(mam, "stitch01", "herdr:stitch") # (b) `stitch` is an mcp attachment, not a session + _record(mam, "anon0001", None) # (b) no session at all + + found = {Path(f["path"]).name: f for f in guard.leaks(roots, before, recorded)} + assert sorted(found) == ["anon0001.json", "orphan01", "stitch01.json", "stray001.json"], found + assert found["orphan01"]["reason"] == "audit log without a registry record" + assert found["anon0001.json"]["reason"] == "no agent_session" + assert found["stray001.json"]["reason"] == "agent_session names no registered session" + assert found["stray001.json"]["prompt"] == "prompt of stray001" + assert "orphan01" in guard.describe(list(found.values())) + + +def test_s5_watch_roots_follow_the_recorded_binding(tmp_path): + repo = _fake_control_plane(tmp_path / "repo") + main = _fake_control_plane(tmp_path / "main") + (tmp_path / "main-link").symlink_to(main) + own, other = str((repo / ".mam").resolve()), str((main / ".mam").resolve()) + + assert guard.watch_roots(repo, {}) == [own] + # WORKSPACE_ROOT naming this checkout is counted once + assert guard.watch_roots(repo, {"WORKSPACE_ROOT": str(repo)}) == [own] + # a worktree creator's pane: all four at another checkout, reached through a symlink too + pane = {"WORKSPACE_ROOT": str(tmp_path / "main-link"), + "AGENT_SESSIONS_YAML": str(main / ".mam" / "agent-sessions.yaml"), + "MAM_JOBS_DIR": str(main / ".mam" / "jobs"), + "DELEGATE_JOB_LOGS_DIR": str(main / ".mam" / "delegate_job_logs")} + assert guard.watch_roots(repo, pane) == [own, other] + # a recorded root that does not exist is skipped; the checkout's own .mam never is + bare = tmp_path / "bare" + assert guard.watch_roots(bare, {"WORKSPACE_ROOT": str(tmp_path / "gone")}) == [str((bare / ".mam").resolve())] + # a registry outside any /jobs is watched directly + odd = tmp_path / "elsewhere" / "registry" + odd.mkdir(parents=True) + rec = {"MAM_JOBS_DIR": str(odd)} + jobs, logs = guard.state_dirs(guard.watch_roots(repo, rec), rec) + assert str(odd.resolve()) in jobs and str(odd.resolve()) not in logs