GodopuandClaude Opus 5
c5844a8e12
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Sdga4A8i1iSc8DaaW9iKU2
104 lines
4.7 KiB
Python
104 lines
4.7 KiB
Python
#!/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 <root>/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
|