Files
multi-agent-mux/tests/conftest.py
T
GodopuandClaude Fable 5.1 4da4a50682 test(issue-19): run the sandbox reaper through an import-time Popen
The reaper executes in fixture teardown, before monkeypatch restores a
test's patches; G5 forbids subprocess.run/Popen for the test's duration and
subprocess.run resolves Popen through the module at call time, so the
teardown tripped the spy. Bind the Popen class at import instead.

Full suite at 38d8ccb: 870 passed, 1 error (this teardown), 0 leaked daemons.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff
2026-09-13 16:09:28 +09:00

1413 lines
51 KiB
Python

import os
import sys
import shutil
import json
import uuid
import sqlite3
import subprocess as _subprocess
from pathlib import Path
import pytest
# Issue #19: the sandbox reaper runs in fixture teardown, before `monkeypatch`
# undoes a test's patches. Bind the real Popen class once at import so a test
# that spies on or forbids subprocess.run/Popen (G5) cannot break the reaper.
# (subprocess.run resolves `Popen` through the module at call time, so binding
# `run` alone is not enough.)
_REAL_POPEN = _subprocess.Popen
def _run_real(cmd):
"""stdout of `cmd` via the import-time Popen; '' on any OS error."""
try:
proc = _REAL_POPEN(cmd, stdout=_subprocess.PIPE, stderr=_subprocess.DEVNULL, text=True)
out, _ = proc.communicate(timeout=10)
return out
except (OSError, _subprocess.SubprocessError):
return ""
SKILLS_DIR = str(Path(__file__).resolve().parent.parent / ".agents" / "skills")
if SKILLS_DIR not in sys.path:
sys.path.insert(0, SKILLS_DIR)
def _subscriber_daemon_pids(root):
"""PIDs of `reconcile.sh --subscribe` daemons whose command line lives under
`root`, plus their children (the python subscriber the bash wrapper waits on).
Issue #19: `create_session.sh` starts one such daemon per workspace,
detached into its own session with `--idle-timeout 0`. A sandbox that ran
it must reap it, or every suite run leaves ~30 daemons holding wildcard
subscriptions on the default public broker behind."""
root = str(root)
found = []
for pid in _run_real(["pgrep", "-f", "reconcile.sh --subscribe"]).split():
if root in _run_real(["ps", "-o", "args=", "-p", pid]):
found.append(int(pid))
kids = []
for pid in found:
kids += [int(k) for k in _run_real(["pgrep", "-P", str(pid)]).split() if k.isdigit()]
return found + kids
def _reap_subscriber_daemons(root):
"""Terminate (then kill) every subscriber daemon anchored under `root`.
Returns the PIDs that were still alive before the sweep."""
import signal as _sig
import time as _t
pids = _subscriber_daemon_pids(root)
for sig in (_sig.SIGTERM, _sig.SIGKILL):
for pid in pids:
try:
os.kill(pid, sig)
except OSError:
pass
deadline = _t.time() + 3
while _t.time() < deadline and any(_pid_alive(p) for p in pids):
_t.sleep(0.1)
if not any(_pid_alive(p) for p in pids):
break
return pids
def _pid_alive(pid):
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
return True
# a zombie still answers kill(0); ps state Z means it is gone for our purposes
state = _run_real(["ps", "-o", "stat=", "-p", str(pid)]).strip()
return bool(state) and not state.startswith("Z")
@pytest.fixture
def mam_sandbox(tmp_path, monkeypatch, request):
"""
Manages a temporary sandboxed directory for .mam/ configuration files
and copies the scripts from .agents/skills to prevent contaminating the real codebase.
Also overrides environments (HOME, PATH, AGENT_SESSIONS_YAML, etc.) for isolation.
"""
# Issue #19: reap any subscriber daemon a test spawned under this sandbox,
# whatever path it came from (create_session.sh, start_watchdog, a direct
# `reconcile.sh --subscribe`). Registered first so it runs after every
# other finalizer of the test.
request.addfinalizer(lambda: _reap_subscriber_daemons(tmp_path))
# 1. Copy the skill scripts to sandboxed tmp_path/skills and tmp_path/.agents/skills
src_skills = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".agents", "skills"))
shutil.copytree(src_skills, tmp_path / "skills")
shutil.copytree(src_skills, tmp_path / ".agents" / "skills")
# 2. Create sandboxed directory structure
mam_dir = tmp_path / ".mam"
mam_dir.mkdir(parents=True, exist_ok=True)
yaml_path = mam_dir / "agent-sessions.yaml"
# Create empty registry structure
yaml_path.write_text("herdr_sessions: []\ndelegation_jobs: []\n")
# 3. Setup mock home directory contents
(tmp_path / ".claude" / "projects").mkdir(parents=True, exist_ok=True)
(tmp_path / ".local" / "bin").mkdir(parents=True, exist_ok=True)
# 4. Monkeypatch environment variables to point inside the sandbox
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("HOME_DIR", str(tmp_path))
monkeypatch.setenv("CLAUDE_PROJECT_DIR", str(tmp_path / ".claude" / "projects"))
monkeypatch.setenv("LOCAL_BIN", str(tmp_path / ".local" / "bin"))
monkeypatch.setenv("AGENT_SESSIONS_YAML", str(yaml_path))
monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
# Issue #15: the job registry and its audit log must land in the sandbox
# too, whatever the cwd of the script under test - not in the repo's own
# .mam/jobs. Explicit here even though the defaults now follow
# WORKSPACE_ROOT, so a test that unsets WORKSPACE_ROOT still stays inside.
monkeypatch.setenv("MAM_JOBS_DIR", str(mam_dir / "jobs"))
monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(mam_dir / "delegate_job_logs"))
# D6 (issue #5 phase 1): point the delegation env loader at a file that does not
# exist, so importing or running delegation code inside a sandbox never injects the
# real .mam.env into this process's os.environ.
monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
# Issue #19: never let code under test detach a real subscriber daemon
# (`reconcile.sh --subscribe --idle-timeout 0`) from a sandbox. Tests that
# exercise the daemon itself override this with "0" or delenv.
monkeypatch.setenv("MAM_NO_SUBSCRIBER_DAEMON", "1")
monkeypatch.delenv("HERDR_SESSION_NAME", raising=False)
monkeypatch.delenv("HERDR_SERVER_NAME", raising=False)
monkeypatch.delenv("HERDR_WORKSPACE_ID", raising=False)
import sys
monkeypatch.setenv("AGENT_PYTHON_BIN", sys.executable)
return tmp_path
@pytest.fixture
def mock_herdr(mam_sandbox, monkeypatch):
"""
Generates a mock 'herdr' executable and prepends it to PATH.
Allows tests to control and verify state via mock_herdr_state.json.
Automatically generates mock session database files upon 'agent start'.
"""
tmp_path = mam_sandbox
state_file = tmp_path / "mock_herdr_state.json"
# Initial mock state
initial_state = {
"workspaces": [
{
"workspace_id": "w1",
"label": "default",
"cwd": str(tmp_path)
}
],
"agents": {},
"calls": []
}
state_file.write_text(json.dumps(initial_state, indent=2))
bin_dir = tmp_path / "bin"
bin_dir.mkdir(parents=True, exist_ok=True)
herdr_bin = bin_dir / "herdr"
# Write Python implementation of mock herdr (using string replacement to avoid f-string escaping issues)
# Double escape backslashes for newline (\\\\n) and write correctly
code = """#!/usr/bin/env python3
import sys
import os
import json
import uuid
import sqlite3
import fcntl
import time
state_file = "STATE_FILE_PLACEHOLDER"
if not os.path.exists(state_file):
sys.exit(0)
# Lock the state file exclusively to prevent concurrent race conditions
lock_f = open(state_file + ".lock", "a")
fcntl.flock(lock_f, fcntl.LOCK_EX)
state = {"workspaces": [], "agents": {}, "calls": []}
if os.path.exists(state_file):
for _retry in range(50):
try:
with open(state_file, 'r') as f:
content = f.read().strip()
if content:
state = json.loads(content)
break
except Exception:
pass
time.sleep(0.02)
# Record the command call
state.setdefault("calls", []).append(sys.argv[1:])
try:
with open(state_file + ".trace", "a") as tf:
tf.write(f"PID {os.getpid()} ARGS: {sys.argv[1:]}\\n")
tf.write(f" AGENTS: {list(state.get('agents', {}).keys())}\\n")
except Exception:
pass
def save_state():
# LOCK INVARIANT (W17):
# lock_f (state_file + .lock) acquired at entry is deliberately NOT closed inside save_state().
# This exclusive global process lock is the LOAD-BEARING guarantee of consistency for
# overwriting disk_state["agents"] and disk_state["workspaces"].
# If lock scope is ever narrowed or made fine-grained, state merging MUST switch to
# explicit key-by-key dictionary diff merging first before removing the global lock.
disk_state = {"workspaces": [], "agents": {}, "calls": [], "panes": []}
if os.path.exists(state_file):
for _retry in range(50):
try:
with open(state_file, 'r') as f:
content = f.read().strip()
if content:
disk_state = json.loads(content)
break
except Exception:
pass
time.sleep(0.02)
disk_state["agents"] = state.get("agents", {})
state["agents"] = disk_state["agents"]
if "workspaces" in state:
disk_ws = disk_state.setdefault("workspaces", [])
for w in state["workspaces"]:
if not any(dw.get("workspace_id") == w.get("workspace_id") for dw in disk_ws):
disk_ws.append(w)
if "panes" in state:
disk_panes = disk_state.setdefault("panes", [])
for p in state["panes"]:
matched = False
for i, dp in enumerate(disk_panes):
if dp.get("pane_id") == p.get("pane_id"):
disk_panes[i] = p
matched = True
break
if not matched:
disk_panes.append(p)
disk_calls = disk_state.setdefault("calls", [])
if sys.argv[1:] and (not disk_calls or disk_calls[-1] != sys.argv[1:]):
disk_calls.append(sys.argv[1:])
tmp_state = state_file + f".tmp.{os.getpid()}"
with open(tmp_state, 'w') as f:
json.dump(disk_state, f, indent=2)
f.flush()
os.fsync(f.fileno())
os.replace(tmp_state, state_file)
# Save calls immediately so they persist even if we exit early or error out
save_state()
def _sanitize_agent_name(name):
if not name:
return "agent"
s = str(name).lower()
import re, hashlib
s = re.sub(r'[^a-z0-9_-]', '-', s)
if not s or not s[0].isalpha():
s = "x-" + s
if len(s) > 32:
h = hashlib.sha1(s.encode('utf-8')).hexdigest()[:8]
s = f"{s[:23]}-{h}"
return s
def _match_agent(k, target):
if not k or not target:
return False
return k == target or _sanitize_agent_name(k) == target or _sanitize_agent_name(target) == k or _sanitize_agent_name(k) == _sanitize_agent_name(target)
args = sys.argv[1:]
while args and args[0] in ("-L", "--server", "-s", "--session"):
if len(args) > 1:
args = args[2:]
else:
args = args[1:]
if not args:
sys.exit(0)
cmd1 = args[0]
if cmd1 == "workspace":
if len(args) < 2:
sys.exit(0)
cmd2 = args[1]
if cmd2 == "list":
# W10: WorkspaceInfo has NO cwd key
wss = []
for w in state.get("workspaces", []):
wss.append({"workspace_id": w["workspace_id"], "label": w.get("label", "default")})
res = {"workspaces": wss}
print(json.dumps({"result": res}))
sys.exit(0)
elif cmd2 == "create":
label = "default"
cwd = "."
i = 2
while i < len(args):
if args[i] == "--label":
label = args[i+1]
i += 2
elif args[i] == "--cwd":
cwd = args[i+1]
i += 2
elif args[i] == "--env":
env_val = args[i+1]
if "=" in env_val:
k, v = env_val.split("=", 1)
os.environ[k] = v
i += 2
else:
i += 1
workspaces = state.get("workspaces", [])
ws_id = f"w{len(workspaces) + 1}"
workspaces.append({"workspace_id": ws_id, "label": label, "cwd": cwd})
state["workspaces"] = workspaces
# W10: workspace create returns full workspace_created object
root_pane_id = f"{ws_id}:p1"
panes = state.get("panes", [])
panes.append({"pane_id": root_pane_id, "workspace_id": ws_id, "cwd": cwd, "tab_id": f"{ws_id}:t1"})
state["panes"] = panes
save_state()
print(json.dumps({
"result": {
"type": "workspace_created",
"workspace": {"workspace_id": ws_id, "label": label},
"tab": {"tab_id": f"{ws_id}:t1"},
"root_pane": {"pane_id": root_pane_id}
}
}))
sys.exit(0)
elif cmd1 == "pane":
if len(args) < 2:
sys.exit(0)
cmd2 = args[1]
if cmd2 == "list":
target_ws = ""
i = 2
while i < len(args):
if args[i] == "--workspace":
target_ws = args[i+1]
i += 2
else:
i += 1
panes_list = []
seen = set()
# Merge agent-derived panes with state["panes"] (dedupe by pane_id).
# Label-only panes live in state["panes"] and must remain visible
# even when other agents are registered.
for name, data in state.get("agents", {}).items():
ws_id = data.get("workspace_id", "w1")
if target_ws and ws_id != target_ws:
continue
pid = data.get("pane_id", f"{ws_id}:p1")
entry = {
"pane_id": pid,
"workspace_id": ws_id,
"cwd": data.get("cwd", "."),
"tab_id": data.get("tab_id", f"{ws_id}:t1"),
"agent": data.get("agent", "claude"),
"name": name,
}
if data.get("label"):
entry["label"] = data["label"]
panes_list.append(entry)
seen.add(pid)
for p in state.get("panes", []):
if target_ws and p.get("workspace_id") != target_ws:
continue
pid = p.get("pane_id")
if pid in seen:
for existing in panes_list:
if existing.get("pane_id") == pid:
for k in ("label", "name"):
if p.get(k) and not existing.get(k):
existing[k] = p[k]
break
continue
panes_list.append(p)
if pid:
seen.add(pid)
print(json.dumps({"result": {"panes": panes_list}}))
sys.exit(0)
elif cmd2 == "split":
# W11: pane split handler with --env support
i = 2
while i < len(args):
if args[i] == "--env":
env_val = args[i+1]
if "=" in env_val:
k, v = env_val.split("=", 1)
os.environ[k] = v
i += 2
else:
i += 1
print(json.dumps({"result": {"type": "pane_split", "pane": {"pane_id": "w1:p_split"}}}))
sys.exit(0)
elif cmd2 == "layout":
# W11: pane layout handler returning area, panes, focused_pane_id
focused = "w1:p1"
panes_list = []
agents = state.get("agents", {})
if state.get("layout_geometry") == "2x2":
# Issue #7 tests: real per-workspace geometry. Only the panes of
# the sampled pane's workspace are reported, laid out as the 2x2
# grid herdr builds, so lib_py.layout judges a full workspace as
# overflow while a sibling workspace with room is offered a split.
sample = ""
i = 2
while i < len(args):
if args[i] == "--pane" and i + 1 < len(args):
sample = args[i+1]
i += 2
else:
i += 1
ws_of = None
for data in agents.values():
if data.get("pane_id") == sample:
ws_of = data.get("workspace_id")
break
if ws_of is None and ":" in sample:
ws_of = sample.split(":")[0]
members = [d for d in agents.values() if ws_of is None or d.get("workspace_id", "w1") == ws_of]
n = len(members)
for idx, data in enumerate(members):
if n == 1:
rect = {"x": 0, "y": 0, "width": 184, "height": 78}
elif n == 2:
rect = {"x": 92 * idx, "y": 0, "width": 92, "height": 78}
elif n == 3:
rect = [{"x": 0, "y": 0, "width": 92, "height": 39},
{"x": 0, "y": 39, "width": 92, "height": 39},
{"x": 92, "y": 0, "width": 92, "height": 78}][idx]
else:
rect = {"x": 92 * (idx % 2), "y": 39 * (idx // 2), "width": 92, "height": 39}
panes_list.append({"pane_id": data.get("pane_id", f"w1:p{idx+1}"), "rect": rect})
else:
count = max(len(agents), 1)
w_per_pane = 184 // count
idx = 0
for name, data in agents.items():
panes_list.append({
"pane_id": data.get("pane_id", f"w1:p{idx+1}"),
"rect": {"width": w_per_pane, "height": 78}
})
idx += 1
if not panes_list:
panes_list = [{"pane_id": "w1:p1", "rect": {"width": 184, "height": 78}}]
print(json.dumps({
"result": {
"area": {"width": 184, "height": 78},
"focused_pane_id": panes_list[0]["pane_id"],
"panes": panes_list
}
}))
sys.exit(0)
elif cmd2 == "send-keys":
if len(args) < 4:
sys.exit(1)
name = args[2]
key = args[3]
agents = state.get("agents", {})
actual_name = name
for a_name, data in agents.items():
if data.get("pane_id") == name or a_name == name:
actual_name = a_name
break
if actual_name in agents:
agents[actual_name]["sent_keys"] = agents[actual_name].get("sent_keys", []) + [key]
if key in ("Enter", "C-m"):
agents[actual_name]["buffer"] = agents[actual_name].get("buffer", "") + "\\\\n\\\\nesc to interrupt"
state["agents"] = agents
save_state()
sys.exit(0)
for p in state.get("panes", []):
if p.get("pane_id") == name:
p["sent_keys"] = p.get("sent_keys", []) + [key]
if key in ("Enter", "C-m"):
p["buffer"] = p.get("buffer", "") + "\\n\\nesc to interrupt"
save_state()
sys.exit(0)
sys.exit(1)
elif cmd2 == "send-text":
if len(args) < 4:
sys.exit(1)
pane_id = args[2]
text = args[3]
found = False
for a_name, data in state.get("agents", {}).items():
if data.get("pane_id") == pane_id or a_name == pane_id:
data["sent_text"] = data.get("sent_text", "") + text
data["buffer"] = data.get("buffer", "") + "\\n" + text
found = True
break
if not found:
for p in state.get("panes", []):
if p.get("pane_id") == pane_id:
p["sent_text"] = p.get("sent_text", "") + text
p["buffer"] = p.get("buffer", "") + "\\n" + text
found = True
break
if found:
save_state()
sys.exit(0)
sys.exit(1)
elif cmd2 == "read":
if len(args) < 3:
sys.exit(1)
pane_id = args[2]
for a_name, data in state.get("agents", {}).items():
if data.get("pane_id") == pane_id or a_name == pane_id:
print(data.get("buffer", "Ready"))
sys.exit(0)
for p in state.get("panes", []):
if p.get("pane_id") == pane_id:
print(p.get("buffer", ""))
sys.exit(0)
sys.stderr.write("Pane " + pane_id + " not found\\n")
sys.exit(1)
elif cmd2 == "rename":
if len(args) < 4:
sys.exit(1)
pane_id = args[2]
label = args[3]
panes = state.setdefault("panes", [])
renamed = False
for p in panes:
if p.get("pane_id") == pane_id:
p["label"] = label
renamed = True
break
if not renamed:
panes.append({"pane_id": pane_id, "label": label})
for a_name, data in state.get("agents", {}).items():
if data.get("pane_id") == pane_id:
data["label"] = label
save_state()
sys.exit(0)
elif cmd2 == "process-info":
pane_id = ""
if "--pane" in args:
pane_id = args[args.index("--pane") + 1]
pid = 9999
for name, data in state.get("agents", {}).items():
if data.get("pane_id") == pane_id or pane_id == "w1:p1":
pid = data.get("pid", 9999)
break
res = {
"process_info": {
"foreground_processes": [{"pid": pid}],
"shell_pid": pid
}
}
print(json.dumps({"result": res}))
sys.exit(0)
elif cmd2 == "close":
pane_id = args[2]
agents = state.get("agents", {})
to_delete = []
for name, data in agents.items():
if data.get("pane_id") == pane_id or pane_id == "w1:p1" or name == pane_id:
to_delete.append(name)
for name in to_delete:
del agents[name]
state["agents"] = agents
save_state()
sys.exit(0)
elif cmd1 == "agent":
if len(args) < 2:
sys.exit(0)
cmd2 = args[1]
if cmd2 == "list":
agents_list = []
for name, data in state.get("agents", {}).items():
agents_list.append({
"name": name,
"agent": data.get("agent", "claude"),
"agent_status": data.get("agent_status", "idle"),
"status": data.get("status", "running"),
"cwd": data.get("cwd", ""),
"pane_id": data.get("pane_id", "w1:p1"),
"workspace_id": data.get("workspace_id", "w1")
})
print(json.dumps({"result": {"agents": agents_list}}))
sys.exit(0)
elif cmd2 == "start":
if len(args) < 3:
sys.exit(1)
name = args[2]
ws = ""
cwd = ""
pane = ""
agent_type = ""
# Find where -- is
try:
double_dash_idx = args.index("--")
agent_cmd = args[double_dash_idx+1:]
opts = args[3:double_dash_idx]
except ValueError:
agent_cmd = []
opts = args[3:]
# Whitelist herdr 0.8.0 allowed flags for agent start
whitelist = {"--kind", "--pane", "--timeout"}
i = 0
unknown_flags = []
while i < len(opts):
opt = opts[i]
if opt not in whitelist:
unknown_flags.append(opt)
i += 1
continue
if opt == "--kind":
agent_type = opts[i+1]
i += 2
elif opt == "--pane":
pane = opts[i+1]
i += 2
elif opt == "--timeout":
i += 2
else:
i += 1
if unknown_flags:
sys.stderr.write("unknown option: " + str(unknown_flags[0]) + "\\n")
sys.exit(1)
if not agent_type or not pane:
missing_arg = "--pane" if not pane else "--kind"
sys.stderr.write("missing required " + missing_arg + "\\n")
sys.exit(1)
# Name validation: ^[a-z][a-z0-9_-]{0,31}$
import re
if not re.match(r'^[a-z][a-z0-9_-]{0,31}$', name):
import json
err_payload = {
"id": "cli:agent:start",
"error": {
"code": "invalid_agent_name",
"message": "agent name must start with a lowercase letter and contain only lowercase letters, digits, '-' or '_' (1-32 characters)"
}
}
sys.stderr.write(json.dumps(err_payload) + "\\n")
sys.exit(1)
# Look up cwd from target pane in state if not explicitly passed
if not cwd and pane:
for p in state.get("panes", []):
if p.get("pane_id") == pane:
cwd = p.get("cwd", "")
ws = p.get("workspace_id", ws)
break
if not ws and pane and ":" in pane:
ws = pane.split(":")[0]
# TUI Welcome Tokens definition to prevent TUI readiness check timeout
resolved_cwd = cwd or "TMP_PATH_PLACEHOLDER"
buffer_content = {
"claude": f"Anthropic Claude Ready in {resolved_cwd}",
"agy": f"Antigravity Ready in {resolved_cwd}",
"hermes": f"Hermes Ready in {resolved_cwd}",
"grok": f"Grok Ready in {resolved_cwd}",
"opencode": f"OpenCode Ready in {resolved_cwd}",
"codex": f"Codex Ready in {resolved_cwd}"
}.get(agent_type, f"Ready in {resolved_cwd}")
agents = state.get("agents", {})
agents[name] = {
"agent": agent_type,
"status": "running",
"cwd": cwd or "TMP_PATH_PLACEHOLDER",
"workspace_id": ws or "w1",
"pid": 9999,
"pane_id": pane or f"w1:p_{name}",
"command": " ".join(agent_cmd),
"buffer": buffer_content
}
# Generate session/conversation UUID and files to simulate agent startup
session_uuid = ""
cmd_str = " ".join(agent_cmd)
if "--session-id" in cmd_str:
parts = cmd_str.split()
if "--session-id" in parts:
_i = parts.index("--session-id")
if _i + 1 < len(parts):
session_uuid = parts[_i + 1]
if not session_uuid:
session_uuid = str(uuid.uuid4())
own_key_map = {
"claude": "claude_session_id_own",
"agy": "agy_conversation_id_own",
"hermes": "hermes_conversation_id_own",
"grok": "grok_session_id_own",
"opencode": "opencode_session_id_own",
"codex": "codex_session_id_own"
}
own_key = own_key_map.get(agent_type)
if own_key:
agents[name][own_key] = session_uuid
# Find isolation directory (e.g. if HOME has agent_homes/<uuid>)
home_dir = os.environ.get("HOME", "TMP_PATH_PLACEHOLDER")
ws_abs = os.path.realpath(cwd or "TMP_PATH_PLACEHOLDER")
ws_key = ws_abs.replace('/', '-').replace('_', '-')
if agent_type == "claude":
ccd = os.environ.get("CLAUDE_CONFIG_DIR")
if ccd:
cp_dir = os.path.join(ccd, "projects")
else:
cp_dir = os.environ.get("CLAUDE_PROJECT_DIR", f"{home_dir}/.claude/projects")
proj_dir = os.path.join(cp_dir, ws_key)
os.makedirs(proj_dir, exist_ok=True)
jsonl_file = os.path.join(proj_dir, f"{session_uuid}.jsonl")
with open(jsonl_file, 'w') as jf:
jf.write(json.dumps({"sessionId": session_uuid}) + "\\n")
elif agent_type == "agy":
db_dir = os.path.join(home_dir, ".gemini", "antigravity-cli", "conversations")
os.makedirs(db_dir, exist_ok=True)
db_file = os.path.join(db_dir, f"{session_uuid}.db")
with open(db_file, 'w') as df:
df.write("")
lc_dir = os.path.join(home_dir, ".gemini", "antigravity-cli", "cache")
os.makedirs(lc_dir, exist_ok=True)
lc_file = os.path.join(lc_dir, "last_conversations.json")
lc_data = {}
if os.path.exists(lc_file):
try:
with open(lc_file, 'r') as lcf:
lc_data = json.load(lcf)
except Exception:
pass
lc_data[ws_abs] = session_uuid
with open(lc_file, 'w') as lcf:
json.dump(lc_data, lcf)
elif agent_type == "hermes":
hermes_dir = os.path.join(home_dir, ".hermes")
os.makedirs(hermes_dir, exist_ok=True)
hdb = os.path.join(hermes_dir, "state.db")
conn = sqlite3.connect(hdb)
conn.execute("CREATE TABLE IF NOT EXISTS sessions (id TEXT PRIMARY KEY, cwd TEXT, started_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)")
conn.execute("INSERT OR REPLACE INTO sessions (id, cwd) VALUES (?, ?)", (session_uuid, ws_abs))
conn.commit()
conn.close()
elif agent_type == "opencode":
opencode_dir = os.path.join(home_dir, ".local", "share", "opencode")
os.makedirs(opencode_dir, exist_ok=True)
odb = os.path.join(opencode_dir, "opencode.db")
conn = sqlite3.connect(odb)
conn.execute("CREATE TABLE IF NOT EXISTS session (id TEXT PRIMARY KEY, directory TEXT, time_created INTEGER, cwd TEXT, created_at REAL)")
now_ms = int(time.time() * 1000)
conn.execute("INSERT OR REPLACE INTO session (id, directory, time_created, cwd, created_at) VALUES (?, ?, ?, ?, ?)", (session_uuid, ws_abs, now_ms, ws_abs, time.time()))
conn.commit()
conn.close()
elif agent_type == "codex":
codex_home = os.environ.get("CODEX_HOME", os.path.join(home_dir, ".codex"))
os.makedirs(codex_home, exist_ok=True)
db_path = os.path.join(codex_home, "state_5.sqlite")
conn = sqlite3.connect(db_path)
conn.execute("CREATE TABLE IF NOT EXISTS threads (id TEXT PRIMARY KEY, cwd TEXT, rollout_path TEXT, created_at INTEGER, updated_at INTEGER)")
sessions_dir = os.path.join(codex_home, "sessions", "2026", "09", "05")
os.makedirs(sessions_dir, exist_ok=True)
rollout_file = os.path.join(sessions_dir, f"rollout-001-{session_uuid}.jsonl")
now_ts = int(time.time())
with open(rollout_file, "w") as rf:
rf.write(json.dumps({"payload": {"cwd": ws_abs, "id": session_uuid}}) + "\\n")
conn.execute("INSERT OR REPLACE INTO threads (id, cwd, rollout_path, created_at, updated_at) VALUES (?, ?, ?, ?, ?)", (session_uuid, ws_abs, rollout_file, now_ts, now_ts))
conn.commit()
conn.close()
state["agents"] = agents
save_state()
print(json.dumps({
"result": {
"type": "agent_started",
"agent": {
"name": name,
"workspace_id": ws or "w1",
"pane_id": f"w1:p_{name}",
"cwd": cwd or "TMP_PATH_PLACEHOLDER"
},
"argv": agent_cmd
}
}))
sys.exit(0)
elif cmd2 == "get":
if len(args) < 3:
sys.exit(1)
name = args[2]
agents = state.get("agents", {})
matched_k = None
for k in agents:
if _match_agent(k, name):
matched_k = k
break
if matched_k:
agent_data = agents[matched_k]
pane_info = {
"pid": agent_data.get("pid", 9999),
"cwd": agent_data.get("cwd", ""),
"command": agent_data.get("command", ""),
"argv": agent_data.get("command", "")
}
agent_st = agent_data.get("agent_status")
if agent_st is None:
seq = agent_data.get("status_sequence")
if seq is None:
seq = state.get("status_sequence")
if seq and isinstance(seq, list) and len(seq) > 0:
agent_st = seq[0]
else:
agent_st = "idle"
print(json.dumps({
"pane": pane_info,
"result": {
"agent": {
"agent": agent_data.get("agent", "claude"),
"status": agent_data.get("status", "running"),
"agent_status": agent_st,
"interactive_ready": agent_data.get("interactive_ready", True),
"cwd": agent_data.get("cwd", ""),
"pane_id": agent_data.get("pane_id", "w1:p1"),
"workspace_id": agent_data.get("workspace_id", "w1")
}
}
}))
sys.exit(0)
else:
sys.stderr.write("Agent " + name + " not found\\n")
sys.exit(1)
elif cmd2 == "wait":
if len(args) < 3:
sys.exit(1)
name = args[2]
until_list = []
timeout_ms = None
i = 3
while i < len(args):
if args[i] == "--until" and i + 1 < len(args):
until_list.append(args[i+1])
i += 2
elif args[i] == "--timeout" and i + 1 < len(args):
try:
timeout_ms = int(args[i+1])
except Exception:
pass
i += 2
else:
i += 1
agents = state.get("agents", {})
matched_k = None
for k in agents:
if _match_agent(k, name):
matched_k = k
break
if not matched_k:
sys.stderr.write(json.dumps({
"error": {
"code": "agent_not_found",
"message": "agent " + name + " not found"
}
}) + "\\n")
sys.exit(1)
agent_data = agents[matched_k]
seq = agent_data.get("status_sequence")
if seq is None:
seq = state.get("status_sequence")
if seq is not None and isinstance(seq, list) and len(seq) > 0:
current_status = seq.pop(0)
agent_data["agent_status"] = current_status
state["agents"] = agents
save_state()
else:
current_status = agent_data.get("agent_status", "idle")
targets = until_list if until_list else ["idle", "done", "blocked"]
if current_status in targets:
print(json.dumps({
"result": {
"agent": {
"agent": agent_data.get("agent", "claude"),
"agent_status": current_status,
"interactive_ready": agent_data.get("interactive_ready", True),
"cwd": agent_data.get("cwd", ""),
"pane_id": agent_data.get("pane_id", "w1:p1"),
"workspace_id": agent_data.get("workspace_id", "w1")
}
}
}))
sys.exit(0)
else:
sys.stderr.write(json.dumps({
"error": {
"code": "timeout",
"message": "timed out waiting for agent " + name
}
}) + "\\n")
sys.exit(1)
elif cmd2 == "read":
if len(args) < 3:
sys.exit(1)
name = args[2]
agents = state.get("agents", {})
matched_k = None
for k in agents:
if _match_agent(k, name):
matched_k = k
break
if matched_k:
buffer_content = agents[matched_k].get("buffer", "Ready")
print(buffer_content)
sys.exit(0)
else:
sys.stderr.write("Agent " + name + " not found\\n")
sys.exit(1)
elif cmd2 == "prompt":
wait_flag = False
until_val = None
timeout_val = None
pos_args = []
i = 2
while i < len(args):
if args[i] == "--wait":
wait_flag = True
i += 1
elif args[i] == "--until" and i + 1 < len(args):
until_val = args[i+1]
i += 2
elif args[i] == "--timeout" and i + 1 < len(args):
timeout_val = args[i+1]
i += 2
else:
pos_args.append(args[i])
i += 1
if len(pos_args) < 2:
sys.exit(1)
name, text = pos_args[0], pos_args[1]
agents = state.get("agents", {})
matched_k = None
for k in agents:
if _match_agent(k, name):
matched_k = k
break
if not matched_k:
sys.stderr.write(json.dumps({
"error": {
"code": "agent_not_found",
"message": "Agent " + name + " not found"
}
}) + "\\n")
sys.exit(1)
target_agent = agents[matched_k]
target_agent["sent_text"] = target_agent.get("sent_text", "") + text
target_agent["buffer"] = target_agent.get("buffer", "") + "\\n" + text + "\\n\\nesc to interrupt"
if wait_flag:
p_res = state.get("prompt_wait_result")
if p_res is None:
p_res = target_agent.get("prompt_wait_result", "ok")
if p_res == "ok":
if until_val == "working":
target_agent["agent_status"] = "working"
state["agents"] = agents
save_state()
print(json.dumps({"id": "cli:agent:prompt", "result": {"type": "ok"}}))
sys.exit(0)
elif p_res == "empty_output":
state["agents"] = agents
save_state()
sys.exit(1)
else:
state["agents"] = agents
save_state()
sys.stderr.write(json.dumps({
"error": {
"code": p_res,
"message": "agent prompt result: " + str(p_res)
}
}) + "\\n")
sys.exit(1)
else:
p_res = state.get("prompt_wait_result")
if p_res is None:
p_res = target_agent.get("prompt_wait_result", "ok")
if p_res == "empty_output":
sys.exit(1)
if p_res != "ok":
sys.stderr.write(json.dumps({
"error": {
"code": p_res,
"message": "agent prompt result: " + str(p_res)
}
}) + "\\n")
sys.exit(1)
print(json.dumps({"id": "cli:agent:prompt", "result": {"type": "ok"}}))
sys.exit(0)
else:
# Real herdr has no `agent send`. Unknown subcommands must fail
# (exit 1) rather than silently succeeding.
sys.stderr.write("error: unrecognized subcommand '" + cmd2 + "'\\n")
sys.exit(1)
elif cmd1 == "session":
if len(args) < 2:
sys.exit(0)
cmd2 = args[1]
if cmd2 == "list":
session_names = ["custom_server", "default", "multi-agent-mux"]
session_names.extend(list(state.get("agents", {}).keys()))
if os.environ.get("MAM_SESS"):
session_names.append(os.environ["MAM_SESS"])
res = {"sessions": [{"name": s, "running": True} for s in set(session_names)]}
print(json.dumps(res))
sys.exit(0)
if len(args) < 3:
sys.exit(1)
name = args[2]
agents = state.get("agents", {})
if cmd2 == "stop":
if name in agents:
agents[name]["status"] = "stopped"
state["agents"] = agents
save_state()
sys.exit(0)
else:
sys.exit(1)
elif cmd2 == "delete":
if name in agents:
del agents[name]
state["agents"] = agents
save_state()
sys.exit(0)
else:
sys.exit(1)
elif cmd1 == "send-keys":
target = ""
key = ""
i = 1
while i < len(args):
if args[i] in ("-t", "--target") and i + 1 < len(args):
target = args[i+1]
i += 2
else:
key = args[i]
i += 1
agents = state.get("agents", {})
actual_name = target
for a_name, data in agents.items():
if data.get("pane_id") == target or a_name == target:
actual_name = a_name
break
if actual_name in agents:
agents[actual_name]["sent_keys"] = agents[actual_name].get("sent_keys", []) + [key]
if key in ("Enter", "C-m"):
agents[actual_name]["buffer"] = agents[actual_name].get("buffer", "") + "\\\\n\\\\nesc to interrupt"
state["agents"] = agents
save_state()
sys.exit(0)
else:
sys.exit(0)
elif cmd1 == "capture-pane":
target = ""
i = 1
while i < len(args):
if args[i] in ("-t", "--target") and i + 1 < len(args):
target = args[i+1]
i += 2
else:
i += 1
agents = state.get("agents", {})
actual_name = target
for a_name, data in agents.items():
if data.get("pane_id") == target or a_name == target:
actual_name = a_name
break
if actual_name in agents:
print(agents[actual_name].get("buffer", "Ready"))
else:
print("Ready")
sys.exit(0)
elif cmd1 == "list-panes":
session_target = ""
if "-t" in args:
session_target = args[args.index("-t") + 1]
agents = state.get("agents", {})
matched_k = None
for k in agents:
if _match_agent(k, session_target):
matched_k = k
break
if matched_k:
data = agents[matched_k]
pid = data.get("pid", 9999)
cwd = data.get("cwd", "TMP_PATH_PLACEHOLDER")
cmd = data.get("command", "claude")
if "-F" in args:
f_idx = args.index("-F")
fmt = args[f_idx + 1] if f_idx + 1 < len(args) else ""
if "pane_current_path" in fmt or "|" in fmt:
print(f"{pid}|{cwd}|{cmd}")
else:
print(f"{pid}")
else:
print(f"{pid}|{cwd}|{cmd}")
sys.exit(0)
else:
sys.exit(1)
elif cmd1 == "has-session":
sess_target = ""
if "-t" in args:
sess_target = args[args.index("-t") + 1]
elif len(args) > 1:
sess_target = args[1]
agents = state.get("agents", {})
if any(_match_agent(k, sess_target) for k in agents):
sys.exit(0)
else:
sys.exit(1)
elif cmd1 == "kill-session":
sess_target = ""
if "-t" in args:
sess_target = args[args.index("-t") + 1]
elif len(args) > 1:
sess_target = args[1]
agents = state.get("agents", {})
to_del = [k for k in agents if _match_agent(k, sess_target)]
for k in to_del:
del agents[k]
state["agents"] = agents
save_state()
sys.exit(0)
elif cmd1 == "ls":
if "-F" in args:
# Real POSIX seconds, not a sentinel: a below-floor value silently
# disables the stale-transcript guard, so a mock that emits one lets
# B-4-class regressions pass unnoticed. Tests may pin a specific
# creation time per agent via the "created" key in the state file.
now = int(time.time())
for name, data in state.get("agents", {}).items():
print(name + "|" + str(data.get("created") or now))
sys.exit(0)
agents_list = []
for name, data in state.get("agents", {}).items():
agents_list.append({
"name": name,
"agent": data.get("agent", "claude"),
"agent_status": data.get("status", "running")
})
print(json.dumps({"result": {"agents": agents_list}}))
sys.exit(0)
sys.exit(0)
""".replace("STATE_FILE_PLACEHOLDER", str(state_file)).replace("TMP_PATH_PLACEHOLDER", str(tmp_path))
herdr_bin.write_text(code)
herdr_bin.chmod(0o755)
# Update PATH using monkeypatch.
# Prepend bin_dir, and append standard Homebrew/system paths so lib.sh won't prepend them
old_path = os.environ.get("PATH", "")
extra_dirs = [
"/home/linuxbrew/.linuxbrew/bin",
"/home/linuxbrew/.linuxbrew/sbin",
f"{tmp_path}/.local/bin",
f"{tmp_path}/.npm-global/bin"
]
new_path = old_path
for d in extra_dirs:
if d not in new_path:
new_path = f"{new_path}:{d}"
monkeypatch.setenv("PATH", f"{bin_dir}:{new_path}")
return state_file
@pytest.fixture
def mock_agents(mam_sandbox, monkeypatch):
"""
Generates mock agent binaries (claude, agy, hermes)
and places them in the sandboxed bin folder to satisfy preflight checks.
"""
tmp_path = mam_sandbox
bin_dir = tmp_path / "bin"
bin_dir.mkdir(parents=True, exist_ok=True)
# 1. claude mock
claude_bin = bin_dir / "claude"
claude_bin.write_text("""#!/usr/bin/env python3
import sys
import json
args = sys.argv[1:]
if len(args) >= 2 and args[0] == "auth" and args[1] == "status":
print(json.dumps({"loggedIn": True}))
sys.exit(0)
sys.exit(0)
""")
claude_bin.chmod(0o755)
# 2. agy mock
agy_bin = bin_dir / "agy"
agy_bin.write_text("""#!/usr/bin/env python3
import sys
args = sys.argv[1:]
if len(args) >= 1 and args[0] == "models":
print("gemini-1.5-pro\\ngemini-1.5-flash")
sys.exit(0)
sys.exit(0)
""")
agy_bin.chmod(0o755)
# 3. hermes mock
hermes_bin = bin_dir / "hermes"
hermes_bin.write_text("""#!/usr/bin/env python3
import sys
args = sys.argv[1:]
if len(args) >= 1 and args[0] == "status":
print("Hermes functional")
sys.exit(0)
sys.exit(0)
""")
hermes_bin.chmod(0o755)
# 4. grok mock
grok_bin = bin_dir / "grok"
grok_bin.write_text("""#!/usr/bin/env python3
import sys
sys.exit(0)
""")
grok_bin.chmod(0o755)
# 5. opencode mock
opencode_bin = bin_dir / "opencode"
opencode_bin.write_text("""#!/usr/bin/env python3
import sys, os
args = sys.argv[1:]
if len(args) >= 2 and args[0] == "db" and args[1] == "path":
home = os.environ.get("HOME", "")
db_path = f"{home}/.local/share/opencode/opencode.db"
os.makedirs(os.path.dirname(db_path), exist_ok=True)
if not os.path.exists(db_path):
import sqlite3
conn = sqlite3.connect(db_path)
conn.execute("CREATE TABLE IF NOT EXISTS session (id TEXT PRIMARY KEY, directory TEXT, time_created INTEGER)")
conn.close()
print(db_path)
sys.exit(0)
sys.exit(0)
""")
opencode_bin.chmod(0o755)
# 6. codex mock
codex_bin = bin_dir / "codex"
codex_bin.write_text("""#!/usr/bin/env python3
import sys, os
args = sys.argv[1:]
if len(args) >= 1 and args[0] == "--version":
print("codex 0.1.0")
sys.exit(0)
if len(args) >= 1 and args[0] == "delete":
sys.exit(0)
sys.exit(0)
""")
codex_bin.chmod(0o755)
# 7. uuidgen mock to guarantee isolated creation UUIDs
uuidgen_bin = bin_dir / "uuidgen"
uuidgen_bin.write_text("""#!/usr/bin/env python3
import uuid
print(str(uuid.uuid4()))
""")
uuidgen_bin.chmod(0o755)
# Prepend bin directory to PATH
old_path = os.environ.get("PATH", "")
extra_dirs = [
"/home/linuxbrew/.linuxbrew/bin",
"/home/linuxbrew/.linuxbrew/sbin",
f"{tmp_path}/.local/bin",
f"{tmp_path}/.npm-global/bin"
]
new_path = old_path
for d in extra_dirs:
if d not in new_path:
new_path = f"{new_path}:{d}"
monkeypatch.setenv("PATH", f"{bin_dir}:{new_path}")
return bin_dir
# ==============================================================================
# Issue #5 Phase 1 — tree provenance guards (PV-2 / PV-3)
#
# The delegation shims bind the `lib_py.delegation` package to the tree next to
# themselves. Two invariants must hold for the whole session, and both are
# checked once at teardown so a violation anywhere in the suite is caught:
#
# PV-2 every imported `lib_py.*` module OTHER than the delegation package
# resolves to the real skills tree (a sandbox copy must never shadow
# lib_py.agents / layout / state / ...).
# PV-3 no sys.path entry under pytest's basetemp contributes a `lib_py`
# package directory. (Sandbox *scripts* dirs are fine: they contain no
# lib_py, which is why test_tier1_unit.py's inserts stay legal.)
# ==============================================================================
# ==============================================================================
# Ambient MAM_VERSION isolation.
#
# lib.sh exports MAM_VERSION, so any shell that has sourced it — an agent
# session, a developer's terminal — hands that value down to pytest and on to
# every subprocess the suite spawns. lib_py.loop.eager.eager_load() compares it
# against FRAMEWORK_VERSION and refuses to start on a mismatch, so an inherited
# value from a *pre-upgrade* shell makes the loop tests fail against a tree that
# is perfectly consistent with itself.
#
# The suite must measure the tree under test, not the version its launcher
# happened to source, so the ambient value is dropped for the whole session.
# Tests that exercise the gate set MAM_VERSION explicitly and are unaffected.
# ==============================================================================
@pytest.fixture(scope="session", autouse=True)
def _drop_ambient_mam_version():
previous = os.environ.pop("MAM_VERSION", None)
try:
yield
finally:
if previous is not None:
os.environ["MAM_VERSION"] = previous
@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
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
basetemp = str(Path(tmp_path_factory.getbasetemp()).resolve())
leaked = _subscriber_daemon_pids(basetemp)
if leaked:
_reap_subscriber_daemons(basetemp)
assert not leaked, (
"Issue #19: {} subscriber daemon process(es) anchored under {} survived the "
"suite (reaped now): {}".format(len(leaked), basetemp, leaked)
)
@pytest.fixture(scope="session", autouse=True)
def delegation_tree_provenance_guard(tmp_path_factory):
yield
real_skills = Path(SKILLS_DIR).resolve()
basetemp = Path(tmp_path_factory.getbasetemp()).resolve()
def _under(path, root):
try:
Path(path).resolve().relative_to(root)
return True
except (ValueError, OSError):
return False
# --- PV-2 -----------------------------------------------------------------
offenders = []
for name, mod in sorted(sys.modules.items()):
if name != "lib_py" and not name.startswith("lib_py."):
continue
if name == "lib_py.delegation" or name.startswith("lib_py.delegation."):
continue # delegation may legitimately bind to a sandbox copy (D2/PV-1)
origin = getattr(mod, "__file__", None)
if not origin:
continue
if not _under(origin, real_skills):
offenders.append("{} -> {}".format(name, origin))
assert not offenders, (
"PV-2: non-delegation lib_py modules were loaded from outside the real skills "
"tree ({}):\n {}".format(real_skills, "\n ".join(offenders))
)
# --- PV-3 -----------------------------------------------------------------
polluted = [
entry for entry in sys.path
if entry and _under(entry, basetemp) and (Path(entry) / "lib_py").is_dir()
]
assert not polluted, (
"PV-3: sys.path carries temp-tree entries that provide a lib_py package "
"(these would shadow the real tree for any later first-time import):\n "
+ "\n ".join(polluted)
)