1417 lines
55 KiB
Python
1417 lines
55 KiB
Python
"""Issue #78: multi-agent-mux-reset — unit tests for lib_py.reset and
|
|
end-to-end tests for reset.sh.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import json
|
|
import os
|
|
import re
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
SKILLS = REPO / ".agents" / "skills"
|
|
if str(SKILLS) not in sys.path:
|
|
sys.path.insert(0, str(SKILLS))
|
|
|
|
UA = "a7800001-0000-4000-8000-000000000001"
|
|
UB = "b7800002-0000-4000-8000-000000000002"
|
|
UC = "c7800003-0000-4000-8000-000000000003"
|
|
UD = "d7800004-0000-4000-8000-000000000004"
|
|
UORC = "e7800005-0000-4000-8000-000000000005"
|
|
|
|
pytestmark = pytest.mark.usefixtures("own_workspace_root")
|
|
|
|
|
|
# --------------------------------------------------------------------------- Helpers
|
|
|
|
def _row(
|
|
name: str,
|
|
agent: str,
|
|
cwd: Path,
|
|
role: str = "reviewer",
|
|
status: str = "running",
|
|
created_at: str = "2026-10-01T00:00:00Z",
|
|
server: str = "default",
|
|
label: str = "ws",
|
|
own_id: Optional[str] = None,
|
|
pid: int = 4242,
|
|
epoch: int = 1700000000,
|
|
) -> dict:
|
|
from lib_py.agents.registry import own_key
|
|
|
|
key = own_key(agent) if agent in ("claude", "agy", "codex", "grok", "hermes", "opencode") else "unknown_id"
|
|
return {
|
|
"name": name,
|
|
"status": status,
|
|
"role": role,
|
|
"agent": agent,
|
|
"herdr_session": server,
|
|
"herdr_server": server,
|
|
"herdr_workspace": label,
|
|
"herdr_session_created_at": created_at,
|
|
"herdr_session_epoch": epoch,
|
|
key: own_id,
|
|
"session_id_source": "discovered" if own_id else "pending-discovery",
|
|
"session_id_verified": bool(own_id),
|
|
"pane": {"cwd": str(cwd), "pid": pid, "cmd": agent},
|
|
}
|
|
|
|
|
|
def _run_reset_py(
|
|
args: List[str],
|
|
state: Optional[dict] = None,
|
|
stdin_text: Optional[str] = None,
|
|
env_extra: Optional[dict] = None,
|
|
):
|
|
env = dict(os.environ)
|
|
env["PYTHONPATH"] = str(SKILLS)
|
|
env["PYTHONDONTWRITEBYTECODE"] = "1"
|
|
if state is not None:
|
|
env["MAM_STATE_JSON"] = json.dumps(state)
|
|
if env_extra:
|
|
env.update(env_extra)
|
|
return subprocess.run(
|
|
[sys.executable, "-m", "lib_py.reset", *args],
|
|
env=env,
|
|
input=stdin_text,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=15,
|
|
)
|
|
|
|
|
|
def _write_rows(sandbox: Path, rows: list, orchestrator_uuids: Optional[List[str]] = None) -> Path:
|
|
lib = sandbox / ".agents" / "skills" / "lib.sh"
|
|
yaml_path = sandbox / ".mam" / "agent-sessions.yaml"
|
|
mutation = f"d['herdr_sessions'] = {repr(rows)}\n"
|
|
if orchestrator_uuids is not None:
|
|
mutation += f"d['orchestrator_uuids'] = {repr(orchestrator_uuids)}\n"
|
|
res = subprocess.run(
|
|
["bash", "-c", f'source "{lib}" && atomic_dump_yaml "{yaml_path}"'],
|
|
input=mutation,
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(sandbox),
|
|
timeout=60,
|
|
)
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
return yaml_path
|
|
|
|
|
|
def _read_rows(sandbox: Path) -> Dict[str, dict]:
|
|
yaml_path = sandbox / ".mam" / "agent-sessions.yaml"
|
|
data = yaml.safe_load(yaml_path.read_text(encoding="utf-8")) or {}
|
|
return {r["name"]: r for r in data.get("herdr_sessions") or []}
|
|
|
|
|
|
def _db_session_names(sandbox: Path) -> Dict[str, str]:
|
|
import sqlite3
|
|
|
|
db_path = sandbox / ".mam" / "agent-sessions.db"
|
|
if not db_path.exists():
|
|
return {}
|
|
conn = sqlite3.connect(str(db_path))
|
|
try:
|
|
return dict(conn.execute("SELECT name, status FROM sessions").fetchall())
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _claude_transcript(sandbox: Path, ws: Path, uuid: str, mtime: Optional[float] = None) -> Path:
|
|
from lib_py.verify_session import workspace_key
|
|
|
|
pdir = sandbox / ".claude" / "projects" / workspace_key(str(ws))
|
|
pdir.mkdir(parents=True, exist_ok=True)
|
|
f = pdir / f"{uuid}.jsonl"
|
|
f.write_text(json.dumps({"sessionId": uuid, "cwd": str(ws), "type": "user"}) + "\n", encoding="utf-8")
|
|
if mtime is not None:
|
|
os.utime(f, (mtime, mtime))
|
|
return f
|
|
|
|
|
|
def _agy_full_artifacts(home: Path, uuid: str, mtime: Optional[float] = None) -> List[Path]:
|
|
import sqlite3
|
|
|
|
base = Path(home) / ".gemini" / "antigravity-cli"
|
|
db = base / "conversations" / f"{uuid}.db"
|
|
db.parent.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(str(db))
|
|
conn.execute("CREATE TABLE steps (id INTEGER)")
|
|
conn.execute("INSERT INTO steps VALUES (1)")
|
|
conn.commit()
|
|
conn.close()
|
|
if mtime is not None:
|
|
os.utime(db, (mtime, mtime))
|
|
wal = Path(str(db) + "-wal")
|
|
shm = Path(str(db) + "-shm")
|
|
wal.write_text("wal", encoding="utf-8")
|
|
shm.write_text("shm", encoding="utf-8")
|
|
brain = base / "brain" / uuid
|
|
brain.mkdir(parents=True, exist_ok=True)
|
|
(brain / "note.md").write_text("note", encoding="utf-8")
|
|
ann = base / "annotations" / f"{uuid}.pbtxt"
|
|
ann.parent.mkdir(parents=True, exist_ok=True)
|
|
ann.write_text("pbtxt", encoding="utf-8")
|
|
lock = base / "presence" / f"{uuid}.lock"
|
|
lock.parent.mkdir(parents=True, exist_ok=True)
|
|
lock.write_text("", encoding="utf-8")
|
|
return [db, wal, shm, brain, ann, lock]
|
|
|
|
|
|
def _agy_cache(home: Path, mapping: dict) -> None:
|
|
lc = Path(home) / ".gemini" / "antigravity-cli" / "cache" / "last_conversations.json"
|
|
lc.parent.mkdir(parents=True, exist_ok=True)
|
|
lc.write_text(
|
|
json.dumps({os.path.realpath(str(k)): v for k, v in mapping.items()}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def _fake_lsof_multi(bindir: Path, mapping: dict) -> Path:
|
|
cases = "\n".join(f" *{k}*) echo {v}; found=1 ;;" for k, v in mapping.items())
|
|
exe = Path(bindir) / "lsof"
|
|
exe.write_text(
|
|
"#!/bin/sh\n"
|
|
"found=0\n"
|
|
'for a in "$@"; do\n'
|
|
' case "$a" in\n'
|
|
f"{cases}\n"
|
|
" esac\n"
|
|
"done\n"
|
|
'[ "$found" = "1" ] && exit 0\n'
|
|
"exit 1\n",
|
|
encoding="utf-8",
|
|
)
|
|
exe.chmod(0o755)
|
|
return exe
|
|
|
|
|
|
def _mock_herdr_agent(
|
|
state_file: Path,
|
|
name: str,
|
|
cwd: Path,
|
|
pid: int = 4242,
|
|
epoch: int = 1700000000,
|
|
agent: str = "claude",
|
|
pane_id: str = "",
|
|
) -> None:
|
|
state = json.loads(Path(state_file).read_text(encoding="utf-8"))
|
|
pid_str = pane_id or f"w1:p_{name}"
|
|
state.setdefault("agents", {})[name] = {
|
|
"pid": pid,
|
|
"cwd": str(cwd),
|
|
"command": agent,
|
|
"agent": agent,
|
|
"pane_id": pid_str,
|
|
"workspace_id": "w1",
|
|
"created": epoch,
|
|
"buffer": f"Ready in {cwd}\n>",
|
|
}
|
|
Path(state_file).write_text(json.dumps(state), encoding="utf-8")
|
|
|
|
|
|
def _wait_reconcile_idle(sandbox: Path, timeout: float = 30.0) -> None:
|
|
"""Wait until any background `reconcile.sh` spawned under `sandbox` finishes.
|
|
Uses `ps -eo pid=,args=` (never `pgrep -f`) so it never matches itself.
|
|
"""
|
|
sb = os.path.realpath(str(sandbox))
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
res = subprocess.run(
|
|
["ps", "-eo", "pid=,args="],
|
|
stdin=subprocess.DEVNULL,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5,
|
|
)
|
|
active = []
|
|
for line in (res.stdout or "").splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
parts = line.split(None, 1)
|
|
if len(parts) < 2 or not parts[0].isdigit():
|
|
continue
|
|
pid = int(parts[0])
|
|
if pid == os.getpid():
|
|
continue
|
|
args = parts[1]
|
|
if "reconcile.sh" in args and (sb in args or str(sandbox) in args):
|
|
active.append((pid, args))
|
|
if not active:
|
|
return
|
|
time.sleep(0.2)
|
|
pytest.fail(f"reconcile.sh still running under {sandbox} after {timeout}s: {active}")
|
|
|
|
|
|
def _install_herdr_spy(sandbox: Path) -> Path:
|
|
"""Wrap sandbox/bin/herdr to log verb sequence, spec-at-first-kill, and optional
|
|
run_loop.sh probe at first agent-start.
|
|
"""
|
|
bin_dir = sandbox / "bin"
|
|
herdr_bin = bin_dir / "herdr"
|
|
herdr_real = bin_dir / "herdr.real"
|
|
if not herdr_real.exists():
|
|
herdr_bin.rename(herdr_real)
|
|
spy_log = sandbox / "herdr_spy.jsonl"
|
|
probe_file = sandbox / "herdr_spy_loop_probe.json"
|
|
|
|
wrapper = f"""#!{sys.executable}
|
|
import json, os, subprocess, sys
|
|
|
|
real = {repr(str(herdr_real))}
|
|
spy_log = {repr(str(spy_log))}
|
|
probe_file = {repr(str(probe_file))}
|
|
sandbox = {repr(str(sandbox))}
|
|
|
|
raw_args = sys.argv[1:]
|
|
args = list(raw_args)
|
|
while args and args[0] in ("-L", "--server", "-s", "--session"):
|
|
args = args[2:] if len(args) > 1 else args[1:]
|
|
|
|
if args and args[0] in ("kill-session", "has-session", "list-panes", "capture-pane", "send-keys", "ls"):
|
|
verb = args[0]
|
|
else:
|
|
verb = " ".join(args[:2]) if len(args) >= 2 else (args[0] if args else "")
|
|
rec = {{"verb": verb, "args": args}}
|
|
|
|
if verb in ("kill-session", "pane close"):
|
|
spec_p = os.path.join(sandbox, ".mam", "reset-spec.json")
|
|
rec["spec_exists"] = os.path.isfile(spec_p)
|
|
if rec["spec_exists"]:
|
|
try:
|
|
with open(spec_p, encoding="utf-8") as fh:
|
|
rec["spec"] = json.load(fh)
|
|
except Exception as exc:
|
|
rec["spec_error"] = str(exc)
|
|
|
|
if verb == "agent start" and os.environ.get("SPY_PROBE_RUN_LOOP") == "1" and not os.path.exists(probe_file):
|
|
env = dict(os.environ, MAM_LOOP_NO_FREEZE="1")
|
|
loop_sh = os.path.join(sandbox, ".agents", "skills", "multi-agent-mux-loop", "scripts", "run_loop.sh")
|
|
res = subprocess.run(
|
|
["bash", loop_sh, "--creator", "x", "--task", "t"],
|
|
cwd=sandbox,
|
|
env=env,
|
|
stdin=subprocess.DEVNULL,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
)
|
|
with open(probe_file, "w", encoding="utf-8") as pf:
|
|
json.dump({{"rc": res.returncode, "stdout": res.stdout, "stderr": res.stderr}}, pf)
|
|
|
|
with open(spy_log, "a", encoding="utf-8") as lf:
|
|
lf.write(json.dumps(rec) + "\\n")
|
|
|
|
os.execv(real, [real, *raw_args])
|
|
"""
|
|
herdr_bin.write_text(wrapper, encoding="utf-8")
|
|
herdr_bin.chmod(0o755)
|
|
return spy_log
|
|
|
|
|
|
def _read_spy_log(spy_log: Path) -> List[dict]:
|
|
if not spy_log.exists():
|
|
return []
|
|
return [json.loads(line) for line in spy_log.read_text(encoding="utf-8").splitlines() if line.strip()]
|
|
|
|
|
|
def _run_reset(
|
|
sandbox: Path,
|
|
*args: str,
|
|
stdin_text: Optional[str] = None,
|
|
env_extra: Optional[dict] = None,
|
|
) -> subprocess.CompletedProcess:
|
|
script = sandbox / ".agents" / "skills" / "multi-agent-mux-reset" / "scripts" / "reset.sh"
|
|
env = dict(
|
|
os.environ,
|
|
SKS_QUIESCENT_TRIES="0",
|
|
SKS_DIALOG_TIMEOUT="0",
|
|
SKS_BUSY_TIMEOUT="0",
|
|
MAM_ONBOARD_ACK_S="3",
|
|
MAM_ONBOARD_WAIT_S="3",
|
|
MAM_NO_SUBSCRIBER_DAEMON="1",
|
|
PYTHONDONTWRITEBYTECODE="1",
|
|
)
|
|
if env_extra:
|
|
env.update(env_extra)
|
|
return subprocess.run(
|
|
["bash", str(script), *args],
|
|
cwd=str(sandbox),
|
|
env=env,
|
|
input=stdin_text,
|
|
stdin=subprocess.DEVNULL if stdin_text is None else None,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=180,
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- Unit tests: lib_py.reset
|
|
|
|
def test_static_checks_and_skill_version():
|
|
"""§5.2-12:
|
|
- bash -n reset.sh
|
|
- ast.parse(feature_version=(3, 9)) on lib_py/reset.py
|
|
- no `kill` command in reset.sh
|
|
- multi-agent-mux-reset/SKILL.md frontmatter version == lib.sh MAM_VERSION
|
|
"""
|
|
reset_sh = SKILLS / "multi-agent-mux-reset" / "scripts" / "reset.sh"
|
|
res = subprocess.run(["bash", "-n", str(reset_sh)], capture_output=True, text=True, timeout=10)
|
|
assert res.returncode == 0, res.stderr
|
|
|
|
src_py = (SKILLS / "lib_py" / "reset.py").read_text(encoding="utf-8")
|
|
ast.parse(src_py, filename="reset.py", feature_version=(3, 9))
|
|
|
|
src_sh = reset_sh.read_text(encoding="utf-8")
|
|
for lineno, line in enumerate(src_sh.splitlines(), 1):
|
|
stripped = line.strip()
|
|
if stripped.startswith("#"):
|
|
continue
|
|
assert not re.search(r"\bkill\b", stripped), f"reset.sh:{lineno} contains forbidden `kill`: {line}"
|
|
|
|
# Self-contained version check (no import of test_version_consistency)
|
|
lib_text = (SKILLS / "lib.sh").read_text(encoding="utf-8")
|
|
m_lib = re.search(r'^MAM_VERSION="([^"]+)"', lib_text, re.M)
|
|
assert m_lib, "MAM_VERSION not found in lib.sh"
|
|
mam_version = m_lib.group(1)
|
|
|
|
skill_text = (SKILLS / "multi-agent-mux-reset" / "SKILL.md").read_text(encoding="utf-8")
|
|
frontmatter = skill_text.split("---", 2)[1]
|
|
m_skill = re.search(r"^version:\s*(\S+)\s*$", frontmatter, re.M)
|
|
assert m_skill, "version not found in multi-agent-mux-reset/SKILL.md frontmatter"
|
|
assert m_skill.group(1) == mam_version
|
|
|
|
|
|
def test_select_targets_and_plan_cli(tmp_path: Path):
|
|
"""§5.2-1: select_targets chooses only status==running rows at realpath(WS),
|
|
resolving symlinks, listing skipped rows with reasons and orchestrator_uuids,
|
|
and refusing running control-root rows with missing agent or empty role.
|
|
"""
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
wt = ws / "wt" / "x"
|
|
wt.mkdir(parents=True)
|
|
sym_ws = tmp_path / "sym-ws"
|
|
sym_ws.symlink_to(ws)
|
|
|
|
state = {
|
|
"orchestrator_uuids": [UORC],
|
|
"herdr_sessions": [
|
|
_row("reviewer-claude-01", "claude", ws, role="planner,reviewer", own_id=UA, server="srv1", label="lbl1"),
|
|
_row("reviewer-grok-01", "grok", sym_ws, role="reviewer", own_id=UB, server="default", label="lbl2"),
|
|
_row("creator-claude-wt", "claude", wt, role="creator", own_id=UC),
|
|
_row("stopped-codex-01", "codex", ws, status="stopped", own_id=UD),
|
|
_row("dead-hermes-01", "hermes", ws, status="terminated", own_id=None),
|
|
],
|
|
}
|
|
|
|
res = _run_reset_py(["plan", "--workspace", str(ws)], state=state)
|
|
assert res.returncode == 0, res.stderr
|
|
plan = json.loads(res.stdout)
|
|
assert [t["name"] for t in plan["targets"]] == ["reviewer-claude-01", "reviewer-grok-01"]
|
|
assert plan["targets"][0] == {
|
|
"name": "reviewer-claude-01",
|
|
"agent": "claude",
|
|
"role": "planner,reviewer",
|
|
"herdr_server": "srv1",
|
|
"herdr_workspace": "lbl1",
|
|
"old_id": UA,
|
|
"old_created_at": "2026-10-01T00:00:00Z",
|
|
}
|
|
skipped = {s["name"]: s["reason"] for s in plan["skipped"]}
|
|
assert "not the control root" in skipped["creator-claude-wt"]
|
|
assert skipped["stopped-codex-01"] == "status stopped"
|
|
assert skipped["dead-hermes-01"] == "status terminated"
|
|
assert plan["orchestrators"] == [UORC]
|
|
assert plan["errors"] == []
|
|
|
|
# Empty role or unknown agent on a running control-root row causes plan & precheck to return 1
|
|
bad_state = {
|
|
"herdr_sessions": [
|
|
_row("bad-role-claude", "claude", ws, role="", own_id=UA),
|
|
{
|
|
"name": "mystery-row",
|
|
"status": "running",
|
|
"role": "reviewer",
|
|
"agent": "not-an-agent",
|
|
"pane": {"cwd": str(ws), "pid": 123, "cmd": "bash"},
|
|
},
|
|
]
|
|
}
|
|
res_bad = _run_reset_py(["plan", "--workspace", str(ws)], state=bad_state)
|
|
assert res_bad.returncode == 1
|
|
bad_plan = json.loads(res_bad.stdout)
|
|
assert len(bad_plan["errors"]) == 2
|
|
|
|
res_pre = _run_reset_py(["precheck", "--workspace", str(ws)], state=bad_state)
|
|
assert res_pre.returncode == 1
|
|
assert "bad-role-claude" in res_pre.stdout
|
|
assert "mystery-row" in res_pre.stdout
|
|
|
|
|
|
def test_spec_file_write_read_and_validation(tmp_path: Path):
|
|
"""D5: write-spec atomically writes .mam/reset-spec.json; corrupt, wrong-version,
|
|
or wrong-workspace spec files are rejected with 'check it, then remove it by hand'.
|
|
"""
|
|
from lib_py.reset import SpecError, read_spec
|
|
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
targets = [
|
|
{
|
|
"name": "reviewer-claude-01",
|
|
"agent": "claude",
|
|
"role": "reviewer",
|
|
"herdr_server": "default",
|
|
"herdr_workspace": "ws",
|
|
"old_id": UA,
|
|
"old_created_at": "2026-10-01T00:00:00Z",
|
|
}
|
|
]
|
|
res = _run_reset_py(
|
|
["write-spec", "--workspace", str(ws)],
|
|
stdin_text=json.dumps({"targets": targets}),
|
|
)
|
|
assert res.returncode == 0, res.stderr
|
|
spath = ws / ".mam" / "reset-spec.json"
|
|
assert spath.is_file()
|
|
doc = read_spec(str(ws))
|
|
assert doc["version"] == 1
|
|
assert doc["workspace"] == os.path.realpath(str(ws))
|
|
assert doc["agents"] == targets
|
|
|
|
# Wrong workspace
|
|
other_ws = tmp_path / "other"
|
|
other_ws.mkdir()
|
|
with pytest.raises(SpecError, match="check it, then remove it by hand"):
|
|
read_spec(str(other_ws))
|
|
|
|
# Wrong version
|
|
spath.write_text(json.dumps({"version": 99, "workspace": str(ws), "agents": []}), encoding="utf-8")
|
|
with pytest.raises(SpecError, match="check it, then remove it by hand"):
|
|
read_spec(str(ws))
|
|
|
|
# Unknown or non-canonical agent in saved spec (review P2)
|
|
for bad_agent in ("not-an-agent", "CLAUDE"):
|
|
spath.write_text(
|
|
json.dumps(
|
|
{
|
|
"version": 1,
|
|
"workspace": str(ws),
|
|
"agents": [
|
|
{
|
|
"name": "reviewer-bad-01",
|
|
"agent": bad_agent,
|
|
"role": "reviewer",
|
|
"herdr_server": "default",
|
|
"herdr_workspace": "ws",
|
|
"old_id": UA,
|
|
"old_created_at": "2026-10-01T00:00:00Z",
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
with pytest.raises(SpecError, match="unknown agent"):
|
|
read_spec(str(ws))
|
|
|
|
# Corrupt JSON
|
|
spath.write_text("{corrupt", encoding="utf-8")
|
|
with pytest.raises(SpecError, match="check it, then remove it by hand"):
|
|
read_spec(str(ws))
|
|
res_entries = _run_reset_py(["entries", "--workspace", str(ws)], state={})
|
|
assert res_entries.returncode == 1
|
|
assert "check it, then remove it by hand" in res_entries.stderr
|
|
|
|
|
|
def test_entries_5_states_and_resumed_row_9b(tmp_path: Path):
|
|
"""D6 1-5 & §5.2-9b:
|
|
#1 row absent -> purged
|
|
#2 same created_at (even if normal stop recorded a new id onto old_id=None) -> pending
|
|
#3 new created_at and id == old_id -> pending (resumed original conversation, 9b)
|
|
#4 new created_at and id empty -> created
|
|
#5 new created_at and id != old_id -> done
|
|
"""
|
|
from lib_py.reset import write_spec
|
|
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
spec_agents = [
|
|
{
|
|
"name": "s1-purged",
|
|
"agent": "claude",
|
|
"role": "reviewer",
|
|
"herdr_server": "default",
|
|
"herdr_workspace": "ws",
|
|
"old_id": UA,
|
|
"old_created_at": "2026-10-01T00:00:00Z",
|
|
},
|
|
{
|
|
"name": "s2-same-inst-newly-recorded",
|
|
"agent": "agy",
|
|
"role": "reviewer",
|
|
"herdr_server": "default",
|
|
"herdr_workspace": "ws",
|
|
"old_id": None,
|
|
"old_created_at": "2026-10-01T00:00:00Z",
|
|
},
|
|
{
|
|
"name": "s3-resumed-same-id",
|
|
"agent": "claude",
|
|
"role": "reviewer",
|
|
"herdr_server": "default",
|
|
"herdr_workspace": "ws",
|
|
"old_id": UB,
|
|
"old_created_at": "2026-10-01T00:00:00Z",
|
|
},
|
|
{
|
|
"name": "s4-created-no-id",
|
|
"agent": "hermes",
|
|
"role": "reviewer",
|
|
"herdr_server": "default",
|
|
"herdr_workspace": "ws",
|
|
"old_id": UC,
|
|
"old_created_at": "2026-10-01T00:00:00Z",
|
|
},
|
|
{
|
|
"name": "s5-done",
|
|
"agent": "codex",
|
|
"role": "reviewer",
|
|
"herdr_server": "default",
|
|
"herdr_workspace": "ws",
|
|
"old_id": None,
|
|
"old_created_at": "2026-10-01T00:00:00Z",
|
|
},
|
|
]
|
|
write_spec(str(ws), spec_agents)
|
|
|
|
state = {
|
|
"herdr_sessions": [
|
|
_row(
|
|
"s2-same-inst-newly-recorded",
|
|
"agy",
|
|
ws,
|
|
status="stopped",
|
|
created_at="2026-10-01T00:00:00Z",
|
|
own_id=UA,
|
|
),
|
|
_row(
|
|
"s3-resumed-same-id",
|
|
"claude",
|
|
ws,
|
|
status="running",
|
|
created_at="2026-10-02T00:00:00Z",
|
|
own_id=UB,
|
|
),
|
|
_row(
|
|
"s4-created-no-id",
|
|
"hermes",
|
|
ws,
|
|
status="running",
|
|
created_at="2026-10-02T00:00:00Z",
|
|
own_id=None,
|
|
),
|
|
_row(
|
|
"s5-done",
|
|
"codex",
|
|
ws,
|
|
status="running",
|
|
created_at="2026-10-02T00:00:00Z",
|
|
own_id=UD,
|
|
),
|
|
]
|
|
}
|
|
|
|
res = _run_reset_py(["entries", "--workspace", str(ws)], state=state)
|
|
assert res.returncode == 0, res.stderr
|
|
parsed = {}
|
|
for line in res.stdout.strip().splitlines():
|
|
kv = dict(tok.split("=", 1) for tok in shlex.split(line))
|
|
parsed[kv["R_NAME"]] = kv
|
|
|
|
assert parsed["s1-purged"]["R_STATE"] == "purged"
|
|
assert parsed["s2-same-inst-newly-recorded"]["R_STATE"] == "pending"
|
|
assert parsed["s2-same-inst-newly-recorded"]["R_ROW_ID"] == UA
|
|
assert parsed["s3-resumed-same-id"]["R_STATE"] == "pending"
|
|
assert parsed["s4-created-no-id"]["R_STATE"] == "created"
|
|
assert parsed["s5-done"]["R_STATE"] == "done"
|
|
|
|
res_tbl = _run_reset_py(
|
|
[
|
|
"table",
|
|
"--workspace",
|
|
str(ws),
|
|
"--skipped",
|
|
"s2-same-inst-newly-recorded:purge_refused",
|
|
],
|
|
state=state,
|
|
)
|
|
assert res_tbl.returncode == 4
|
|
rows_by_name = {}
|
|
for line in res_tbl.stdout.strip().splitlines()[1:]:
|
|
cols = line.split(None, 6)
|
|
rows_by_name[cols[0]] = cols
|
|
|
|
assert rows_by_name["s2-same-inst-newly-recorded"][1:6] == ["agy", "-", "-", "no", "-"]
|
|
assert rows_by_name["s2-same-inst-newly-recorded"][6] == "skipped: purge refused"
|
|
assert rows_by_name["s3-resumed-same-id"][1:6] == ["claude", UB, UB, "yes", "no"]
|
|
assert rows_by_name["s4-created-no-id"][1:6] == ["hermes", UC, "-", "no", "-"]
|
|
assert rows_by_name["s4-created-no-id"][6] == "created; id not recorded yet"
|
|
assert rows_by_name["s5-done"][1:6] == ["codex", "-", UD, "yes", "yes"]
|
|
assert rows_by_name["s5-done"][6] == "done"
|
|
|
|
# Review F1: (a) old_id=None + same instance + id recorded on row -> record-id persists old_id=UA
|
|
from lib_py.reset import read_spec
|
|
|
|
res_rec = _run_reset_py(
|
|
["record-id", "--workspace", str(ws), "--session", "s2-same-inst-newly-recorded"],
|
|
state=state,
|
|
)
|
|
assert res_rec.returncode == 0, res_rec.stderr
|
|
spec_after = read_spec(str(ws))
|
|
s2_entry = next(e for e in spec_after["agents"] if e["name"] == "s2-same-inst-newly-recorded")
|
|
assert s2_entry["old_id"] == UA
|
|
|
|
# Review F1: (b) after resuming that skipped row (new created_at, id still UA), it is classified
|
|
# as 'pending' (NOT falsely 'done') and table exits non-zero (4) with NEW!=OLD == 'no'
|
|
state["herdr_sessions"][0] = _row(
|
|
"s2-same-inst-newly-recorded",
|
|
"agy",
|
|
ws,
|
|
status="running",
|
|
created_at="2026-10-02T00:00:00Z",
|
|
own_id=UA,
|
|
)
|
|
res_f1_entries = _run_reset_py(
|
|
["entries", "--workspace", str(ws), "--session", "s2-same-inst-newly-recorded"],
|
|
state=state,
|
|
)
|
|
assert res_f1_entries.returncode == 0
|
|
kv_f1 = dict(tok.split("=", 1) for tok in shlex.split(res_f1_entries.stdout.strip()))
|
|
assert kv_f1["R_STATE"] == "pending"
|
|
assert kv_f1["R_OLD_ID"] == UA
|
|
|
|
res_f1_tbl = _run_reset_py(["table", "--workspace", str(ws)], state=state)
|
|
assert res_f1_tbl.returncode == 4
|
|
rows_f1 = {
|
|
line.split(None, 6)[0]: line.split(None, 6)
|
|
for line in res_f1_tbl.stdout.strip().splitlines()[1:]
|
|
}
|
|
assert rows_f1["s2-same-inst-newly-recorded"][1:7] == ["agy", UA, UA, "yes", "no", "pending"]
|
|
|
|
|
|
# --------------------------------------------------------------------------- Integration tests: reset.sh
|
|
|
|
def test_no_yes_non_tty_exits_3_without_touching_locks(mam_sandbox: Path, mock_herdr: Path, mock_agents):
|
|
"""§5.2-3: Without --yes and with non-TTY stdin, reset.sh exits 3 before touching
|
|
any lock (.mam/loop-guard-active or .mam/update-admission.lock) or registry.
|
|
"""
|
|
_write_rows(mam_sandbox, [_row("reviewer-claude-01", "claude", mam_sandbox, own_id=UA)])
|
|
yaml_before = (mam_sandbox / ".mam" / "agent-sessions.yaml").read_bytes()
|
|
|
|
res = _run_reset(mam_sandbox)
|
|
assert res.returncode == 3
|
|
assert "not a terminal" in res.stderr
|
|
assert (mam_sandbox / ".mam" / "agent-sessions.yaml").read_bytes() == yaml_before
|
|
assert not (mam_sandbox / ".mam" / "loop-guard-active").exists()
|
|
assert not (mam_sandbox / ".mam" / "update-admission.lock").exists()
|
|
assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()
|
|
|
|
|
|
def test_dry_run_changes_nothing_in_mam_tree(mam_sandbox: Path, mock_herdr: Path, mock_agents):
|
|
"""§5.2-1 & §5.2-4: --dry-run prints targets (with all spec fields), skipped rows,
|
|
and orchestrator_uuids, exits 0, and leaves the entire .mam tree (names, sizes,
|
|
mtimes) and mock_herdr state untouched.
|
|
"""
|
|
wt = mam_sandbox / "wt" / "issue-1"
|
|
wt.mkdir(parents=True)
|
|
_write_rows(
|
|
mam_sandbox,
|
|
[
|
|
_row("reviewer-claude-01", "claude", mam_sandbox, role="planner,reviewer", server="srv-main", label="lbl-main", own_id=UA),
|
|
_row("reviewer-grok-01", "grok", mam_sandbox, role="reviewer", server="default", label="lbl-main", own_id=UB),
|
|
_row("creator-claude-wt", "claude", wt, role="creator", own_id=UC),
|
|
_row("stopped-codex-01", "codex", mam_sandbox, status="stopped", own_id=UD),
|
|
],
|
|
orchestrator_uuids=[UORC],
|
|
)
|
|
mam_dir = mam_sandbox / ".mam"
|
|
|
|
def _snapshot_mam():
|
|
return {
|
|
str(p.relative_to(mam_dir)): (p.stat().st_size, p.stat().st_mtime_ns)
|
|
for p in sorted(mam_dir.rglob("*"))
|
|
}
|
|
|
|
snap_before = _snapshot_mam()
|
|
yaml_before = (mam_dir / "agent-sessions.yaml").read_bytes()
|
|
|
|
res = _run_reset(mam_sandbox, "--dry-run")
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert _snapshot_mam() == snap_before
|
|
assert (mam_dir / "agent-sessions.yaml").read_bytes() == yaml_before
|
|
|
|
out = res.stdout
|
|
assert "target: reviewer-claude-01" in out
|
|
assert "agent=claude" in out
|
|
assert "role=planner,reviewer" in out
|
|
assert "herdr_server=srv-main" in out
|
|
assert "herdr_workspace=lbl-main" in out
|
|
assert f"old_id={UA}" in out
|
|
assert "old_created_at=2026-10-01T00:00:00Z" in out
|
|
assert "target: reviewer-grok-01" in out
|
|
assert "skip: creator-claude-wt" in out and "not the control root" in out
|
|
assert "skip: stopped-codex-01 (status stopped)" in out
|
|
assert f"orchestrator {UORC} (not a row; never reset)" in out
|
|
|
|
herdr_state = json.loads(mock_herdr.read_text(encoding="utf-8"))
|
|
assert herdr_state.get("calls") == []
|
|
|
|
|
|
def test_precheck_refusals_change_nothing(mam_sandbox: Path, mock_herdr: Path, mock_agents):
|
|
"""§5.2-2: Precheck refusals (open job, live loop marker, live update marker,
|
|
unrecreatable target, caller self-target) exit 1 and change nothing; stale
|
|
markers do not refuse.
|
|
"""
|
|
tf = _claude_transcript(mam_sandbox, mam_sandbox, UA)
|
|
_mock_herdr_agent(mock_herdr, "reviewer-claude-01", mam_sandbox, agent="claude")
|
|
_write_rows(mam_sandbox, [_row("reviewer-claude-01", "claude", mam_sandbox, own_id=UA)])
|
|
spy_log = _install_herdr_spy(mam_sandbox)
|
|
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
yaml_before = yaml_path.read_bytes()
|
|
db_before = _db_session_names(mam_sandbox)
|
|
tf_before = tf.read_bytes()
|
|
|
|
def _assert_nothing_changed(expect_loop_marker: bool = False):
|
|
assert yaml_path.read_bytes() == yaml_before
|
|
assert _db_session_names(mam_sandbox) == db_before
|
|
assert tf.read_bytes() == tf_before
|
|
assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()
|
|
calls = _read_spy_log(spy_log)
|
|
assert not any(c["verb"] in ("kill-session", "pane close", "agent start") for c in calls)
|
|
if not expect_loop_marker:
|
|
assert not (mam_sandbox / ".mam" / "loop-guard-active").exists()
|
|
|
|
# (a) Open job
|
|
jobs_dir = mam_sandbox / ".mam" / "jobs"
|
|
jobs_dir.mkdir(parents=True, exist_ok=True)
|
|
job_file = jobs_dir / "job78a.json"
|
|
job_file.write_text(
|
|
json.dumps({"job_id": "job78a", "agent_session": "herdr:reviewer-claude-01", "status": "running"}),
|
|
encoding="utf-8",
|
|
)
|
|
res_job = _run_reset(mam_sandbox, "--yes")
|
|
assert res_job.returncode == 1
|
|
assert "job78a" in res_job.stderr
|
|
assert "close --job job78a --request cancelled" in res_job.stderr
|
|
_assert_nothing_changed()
|
|
job_file.unlink()
|
|
|
|
# Helper to get lstart for a live process
|
|
holder = subprocess.Popen(["sleep", "300"])
|
|
try:
|
|
lstart = subprocess.run(
|
|
["ps", "-p", str(holder.pid), "-o", "lstart="],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
).stdout.strip()
|
|
lstart = " ".join(lstart.split())
|
|
|
|
# (b) Live loop marker
|
|
loop_marker = mam_sandbox / ".mam" / "loop-guard-active"
|
|
loop_marker.write_text(f"pid={holder.pid}\nlstart={lstart}\nstarted=2026-10-29T00:00:00Z\n", encoding="utf-8")
|
|
res_loop = _run_reset(mam_sandbox, "--yes")
|
|
assert res_loop.returncode == 1
|
|
assert "loop" in res_loop.stderr
|
|
assert str(holder.pid) in res_loop.stderr
|
|
_assert_nothing_changed(expect_loop_marker=True)
|
|
loop_marker.unlink()
|
|
|
|
# (c) Live update marker
|
|
upd_marker = mam_sandbox / ".mam" / "update-in-progress"
|
|
upd_marker.write_text(f"pid={holder.pid}\nlstart={lstart}\nstarted=2026-10-29T00:00:00Z\n", encoding="utf-8")
|
|
res_upd = _run_reset(mam_sandbox, "--yes")
|
|
assert res_upd.returncode == 1
|
|
assert "update" in res_upd.stderr
|
|
_assert_nothing_changed()
|
|
upd_marker.unlink()
|
|
finally:
|
|
holder.kill()
|
|
holder.wait()
|
|
|
|
# Stale control: dead PID in loop and update markers does not refuse --dry-run
|
|
(mam_sandbox / ".mam" / "loop-guard-active").write_text("pid=999999\nlstart=Thu Jan 1 00:00:00 1970\n", encoding="utf-8")
|
|
(mam_sandbox / ".mam" / "update-in-progress").write_text("pid=999999\nlstart=Thu Jan 1 00:00:00 1970\n", encoding="utf-8")
|
|
res_stale = _run_reset(mam_sandbox, "--dry-run")
|
|
assert res_stale.returncode == 0, res_stale.stdout + res_stale.stderr
|
|
(mam_sandbox / ".mam" / "loop-guard-active").unlink()
|
|
(mam_sandbox / ".mam" / "update-in-progress").unlink()
|
|
|
|
# (d) Caller self-target (D4-4)
|
|
res_self = _run_reset(mam_sandbox, "--yes", env_extra={"CLAUDE_CODE_SESSION_ID": UA})
|
|
assert res_self.returncode == 1
|
|
assert "you are reviewer-claude-01, one of the targets" in res_self.stderr
|
|
_assert_nothing_changed()
|
|
|
|
# (e) Saved spec with valid pending target first and unknown agent second (review P2)
|
|
spec_file = mam_sandbox / ".mam" / "reset-spec.json"
|
|
bad_spec_text = json.dumps(
|
|
{
|
|
"version": 1,
|
|
"workspace": str(mam_sandbox),
|
|
"written_at": "2026-10-29T00:00:00Z",
|
|
"agents": [
|
|
{
|
|
"name": "reviewer-claude-01",
|
|
"agent": "claude",
|
|
"role": "reviewer",
|
|
"herdr_server": "default",
|
|
"herdr_workspace": "ws",
|
|
"old_id": UA,
|
|
"old_created_at": "2026-10-01T00:00:00Z",
|
|
},
|
|
{
|
|
"name": "reviewer-bad-01",
|
|
"agent": "not-an-agent",
|
|
"role": "reviewer",
|
|
"herdr_server": "default",
|
|
"herdr_workspace": "ws",
|
|
"old_id": UB,
|
|
"old_created_at": "2026-10-01T00:00:00Z",
|
|
},
|
|
],
|
|
}
|
|
)
|
|
spec_file.write_text(bad_spec_text, encoding="utf-8")
|
|
res_bad_spec = _run_reset(mam_sandbox, "--yes")
|
|
assert res_bad_spec.returncode == 1
|
|
assert "unknown agent" in res_bad_spec.stderr
|
|
assert "check it, then remove it by hand" in res_bad_spec.stderr
|
|
assert spec_file.read_text(encoding="utf-8") == bad_spec_text
|
|
spec_file.unlink()
|
|
_assert_nothing_changed()
|
|
|
|
|
|
def test_full_reset_success_spec_ordering_table_and_loop_block(
|
|
mam_sandbox: Path, mock_herdr: Path, mock_agents
|
|
):
|
|
"""§5.2-5, §5.2-6, §5.2-10:
|
|
- .mam/reset-spec.json is written atomically with both targets before the first kill call
|
|
- Both claude sessions are stopped (normal then purge, no UUID on CLI) and recreated sequentially
|
|
- Old transcripts are removed, new IDs recorded != old IDs, result table shows RECORDED=yes, NEW!=OLD=yes, done
|
|
- Spec file and loop marker are cleaned up on exit 0
|
|
- During the reset, run_loop.sh is blocked by .mam/loop-guard-active ("Another multi-agent-mux-loop is already running")
|
|
"""
|
|
name_a = "reviewer-claude-01"
|
|
name_b = "reviewer-claude-02"
|
|
tf_a = _claude_transcript(mam_sandbox, mam_sandbox, UA)
|
|
tf_b = _claude_transcript(mam_sandbox, mam_sandbox, UB)
|
|
_mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a")
|
|
_mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="claude", pane_id="w1:p_b")
|
|
_write_rows(
|
|
mam_sandbox,
|
|
[
|
|
_row(name_a, "claude", mam_sandbox, role="planner,reviewer", server="default", label="lbl-a", own_id=UA),
|
|
_row(name_b, "claude", mam_sandbox, role="reviewer", server="default", label="lbl-b", own_id=UB),
|
|
],
|
|
)
|
|
spy_log = _install_herdr_spy(mam_sandbox)
|
|
|
|
res = _run_reset(mam_sandbox, "--yes", env_extra={"SPY_PROBE_RUN_LOOP": "1"})
|
|
_wait_reconcile_idle(mam_sandbox)
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
|
|
# Spec file & loop marker removed after completion
|
|
assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()
|
|
assert not (mam_sandbox / ".mam" / "loop-guard-active").exists()
|
|
|
|
# Old transcripts purged
|
|
assert not tf_a.exists()
|
|
assert not tf_b.exists()
|
|
|
|
# Recreated rows have same name, role, herdr_server, herdr_workspace, and new IDs != old IDs
|
|
rows = _read_rows(mam_sandbox)
|
|
assert set(rows.keys()) == {name_a, name_b}
|
|
assert rows[name_a]["role"] == "planner,reviewer"
|
|
assert rows[name_a]["herdr_server"] == "default"
|
|
assert rows[name_a]["herdr_workspace"] == "lbl-a"
|
|
new_a = rows[name_a]["claude_session_id_own"]
|
|
assert new_a and new_a != UA
|
|
|
|
assert rows[name_b]["role"] == "reviewer"
|
|
assert rows[name_b]["herdr_server"] == "default"
|
|
assert rows[name_b]["herdr_workspace"] == "lbl-b"
|
|
new_b = rows[name_b]["claude_session_id_own"]
|
|
assert new_b and new_b != UB
|
|
|
|
# Result table in stdout
|
|
out_lines = [l for l in res.stdout.splitlines() if l.startswith(name_a) or l.startswith(name_b)]
|
|
assert len(out_lines) == 2
|
|
assert out_lines[0].split() == [name_a, "claude", UA, new_a, "yes", "yes", "done"]
|
|
assert out_lines[1].split() == [name_b, "claude", UB, new_b, "yes", "yes", "done"]
|
|
|
|
# Stop invoked twice per agent (normal, then --purge-conversation --yes), no UUID on CLI
|
|
stop_traces = [l for l in res.stdout.splitlines() if l.startswith("reset [") and "stop_session.sh" in l]
|
|
assert len(stop_traces) == 4
|
|
for st in stop_traces:
|
|
assert UA not in st and UB not in st
|
|
|
|
# §5.2-5 & §5.2-6: Spy log verifies spec file existed at first kill and sequential order
|
|
calls = _read_spy_log(spy_log)
|
|
first_kill = next(c for c in calls if c["verb"] in ("pane close", "kill-session"))
|
|
assert first_kill["spec_exists"] is True
|
|
assert first_kill["spec"]["version"] == 1
|
|
assert [a["name"] for a in first_kill["spec"]["agents"]] == [name_a, name_b]
|
|
|
|
# Sequential order: kill A -> start A -> kill B -> start B
|
|
events = []
|
|
for c in calls:
|
|
if c["verb"] == "kill-session":
|
|
tgt = c["args"][2] if len(c["args"]) > 2 and c["args"][1] == "-t" else c["args"][1]
|
|
if not events or events[-1] != ("kill", tgt):
|
|
events.append(("kill", tgt))
|
|
elif c["verb"] == "agent start":
|
|
events.append(("start", c["args"][2]))
|
|
assert events == [
|
|
("kill", name_a),
|
|
("start", name_a),
|
|
("kill", name_b),
|
|
("start", name_b),
|
|
]
|
|
|
|
# §5.2-10: During the reset, run_loop.sh was refused by loop-guard-active
|
|
probe_data = json.loads((mam_sandbox / "herdr_spy_loop_probe.json").read_text(encoding="utf-8"))
|
|
assert probe_data["rc"] == 1
|
|
combined_probe = probe_data["stdout"] + probe_data["stderr"]
|
|
assert "Another multi-agent-mux-loop is already running" in combined_probe
|
|
|
|
# Control after reset finishes: run_loop.sh passes the loop lock and fails at "is not registered"
|
|
loop_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh"
|
|
post_loop = subprocess.run(
|
|
["bash", str(loop_sh), "--creator", "x", "--task", "t"],
|
|
cwd=str(mam_sandbox),
|
|
env=dict(os.environ, MAM_LOOP_NO_FREEZE="1"),
|
|
stdin=subprocess.DEVNULL,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
)
|
|
assert "is not registered" in (post_loop.stdout + post_loop.stderr)
|
|
|
|
|
|
def test_mid_run_failure_and_rerun_resumes_from_spec(
|
|
mam_sandbox: Path, mock_herdr: Path, mock_agents
|
|
):
|
|
"""§5.2-7: A (claude) succeeds, B (grok) fails at create (grok preflight fails).
|
|
Run 1 exits 1 with 'failed at create (rc 1)' and leaves .mam/reset-spec.json.
|
|
Restoring grok binary and rerunning skips A (0 stop calls for A) and only creates B.
|
|
"""
|
|
name_a = "reviewer-claude-01"
|
|
name_b = "reviewer-grok-01"
|
|
_claude_transcript(mam_sandbox, mam_sandbox, UA)
|
|
_mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a")
|
|
_mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="grok", pane_id="w1:p_b")
|
|
_write_rows(
|
|
mam_sandbox,
|
|
[
|
|
_row(name_a, "claude", mam_sandbox, own_id=UA),
|
|
_row(name_b, "grok", mam_sandbox, own_id=UB),
|
|
],
|
|
)
|
|
spy_log = _install_herdr_spy(mam_sandbox)
|
|
|
|
# Make bin/grok exit 1 (and XAI_API_KEY="") so B's create_session.sh preflight fails with exit 1
|
|
grok_bin = mam_sandbox / "bin" / "grok"
|
|
grok_bin.write_text("#!/bin/sh\nexit 1\n", encoding="utf-8")
|
|
grok_bin.chmod(0o755)
|
|
|
|
res1 = _run_reset(mam_sandbox, "--yes", env_extra={"XAI_API_KEY": ""})
|
|
_wait_reconcile_idle(mam_sandbox)
|
|
assert res1.returncode == 1
|
|
assert (mam_sandbox / ".mam" / "reset-spec.json").is_file()
|
|
assert "failed at create (rc 1)" in res1.stdout
|
|
|
|
rows1 = _read_rows(mam_sandbox)
|
|
assert name_a in rows1 and rows1[name_a]["claude_session_id_own"] != UA
|
|
assert name_b not in rows1 # B was purged before its create failed
|
|
|
|
# Restore grok binary, clear spy log, and rerun
|
|
grok_bin.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
grok_bin.chmod(0o755)
|
|
spy_log.unlink()
|
|
|
|
res2 = _run_reset(mam_sandbox, "--yes", env_extra={"XAI_API_KEY": ""})
|
|
_wait_reconcile_idle(mam_sandbox)
|
|
assert res2.returncode == 0, res2.stdout + res2.stderr
|
|
assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()
|
|
|
|
# Spy confirms A had 0 stop/kill/start calls on rerun, and B was only created
|
|
calls2 = _read_spy_log(spy_log)
|
|
assert not any(c["verb"] in ("pane close", "kill-session") for c in calls2)
|
|
starts2 = [c["args"][2] for c in calls2 if c["verb"] == "agent start"]
|
|
assert starts2 == [name_b]
|
|
assert f"reset [{name_a}]:" not in res2.stdout
|
|
|
|
|
|
def test_purge_refused_exit_4_preserves_stopped_row_and_rerun_completes(
|
|
mam_sandbox: Path, mock_herdr: Path, mock_agents
|
|
):
|
|
"""§5.2-8: A's transcript is held open by a real sleep 300 process (via fake lsof).
|
|
Run 1: A is stopped in step 1, refused at purge in step 3 (exit 4 -> skipped: purge refused),
|
|
prints holder PID and 'sleep 300', keeps A's row (status=stopped, own_id=UA) and transcript,
|
|
leaves sleep alive, completes B, and exits 4 with spec file intact.
|
|
Run 2 (after holder ends): A is purged and recreated, B is untouched, exits 0.
|
|
"""
|
|
name_a = "reviewer-claude-01"
|
|
name_b = "reviewer-claude-02"
|
|
tf_a = _claude_transcript(mam_sandbox, mam_sandbox, UA)
|
|
tf_b = _claude_transcript(mam_sandbox, mam_sandbox, UB)
|
|
_mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a")
|
|
_mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="claude", pane_id="w1:p_b")
|
|
_write_rows(
|
|
mam_sandbox,
|
|
[
|
|
_row(name_a, "claude", mam_sandbox, own_id=UA),
|
|
_row(name_b, "claude", mam_sandbox, own_id=UB),
|
|
],
|
|
)
|
|
spy_log = _install_herdr_spy(mam_sandbox)
|
|
|
|
holder = subprocess.Popen(["sleep", "300"])
|
|
try:
|
|
lsof_exe = _fake_lsof_multi(mam_sandbox / "bin", {UA: holder.pid})
|
|
res1 = _run_reset(mam_sandbox, "--yes")
|
|
_wait_reconcile_idle(mam_sandbox)
|
|
assert res1.returncode == 4, res1.stdout + res1.stderr
|
|
combined1 = res1.stdout + res1.stderr
|
|
assert f"pid {holder.pid}" in combined1
|
|
assert "sleep 300" in combined1
|
|
assert "skipped: purge refused" in res1.stdout
|
|
assert (mam_sandbox / ".mam" / "reset-spec.json").is_file()
|
|
|
|
# Holder process is still alive (reset never kills external processes)
|
|
assert holder.poll() is None
|
|
|
|
# A's row is stopped with UA preserved and transcript intact; B is done
|
|
rows1 = _read_rows(mam_sandbox)
|
|
assert rows1[name_a]["status"] == "stopped"
|
|
assert rows1[name_a]["claude_session_id_own"] == UA
|
|
assert tf_a.is_file()
|
|
assert not tf_b.exists()
|
|
assert rows1[name_b]["claude_session_id_own"] != UB
|
|
|
|
# Release holder and rerun
|
|
lsof_exe.unlink()
|
|
finally:
|
|
holder.kill()
|
|
holder.wait()
|
|
|
|
spy_log.unlink()
|
|
res2 = _run_reset(mam_sandbox, "--yes")
|
|
_wait_reconcile_idle(mam_sandbox)
|
|
assert res2.returncode == 0, res2.stdout + res2.stderr
|
|
assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()
|
|
assert not tf_a.exists()
|
|
|
|
rows2 = _read_rows(mam_sandbox)
|
|
assert rows2[name_a]["claude_session_id_own"] != UA
|
|
assert f"reset [{name_b}]:" not in res2.stdout
|
|
|
|
|
|
def test_post_discovered_without_recorded_id_8b_and_exit_5(
|
|
mam_sandbox: Path, mock_herdr: Path, mock_agents
|
|
):
|
|
"""§5.2-8b (i) & (ii) and §5.2-9:
|
|
- 8b(i): agy row has no recorded own_id while herdr is live and cache points to UA.
|
|
Run 1 with UA held by fake lsof: step 1 captures UA onto the row (status=stopped),
|
|
step 3 refuses purge (exit 4).
|
|
Run 2 after releasing lsof: step 3 purges UA (all UA artifacts deleted!) and recreates
|
|
the agy session. Because mock agy does not record a new ID, exits 5 (created).
|
|
Run 3 (while still without ID): 0 stop calls (does not re-purge the new session!), exits 5.
|
|
Run 4 (after recording new ID UB): exits 0, removes spec file, table shows OLD_ID='-' and NEW!=OLD='yes'.
|
|
- 8b(ii): agy row has no recorded own_id and no conversation on disk -> step 2 skips with
|
|
'skipped: no conversation id recorded' (exit 4), keeping the stopped row without calling purge.
|
|
"""
|
|
name_a = "reviewer-agy-01"
|
|
now = time.time()
|
|
epoch = int(now) - 100
|
|
a_files = _agy_full_artifacts(mam_sandbox, UA, mtime=now)
|
|
_agy_cache(mam_sandbox, {mam_sandbox: UA})
|
|
|
|
holder = subprocess.Popen(["sleep", "300"])
|
|
try:
|
|
lsof_exe = _fake_lsof_multi(mam_sandbox / "bin", {UA: holder.pid})
|
|
_mock_herdr_agent(mock_herdr, name_a, mam_sandbox, pid=holder.pid, epoch=epoch, agent="agy", pane_id="w1:p_a")
|
|
_write_rows(
|
|
mam_sandbox,
|
|
[_row(name_a, "agy", mam_sandbox, own_id=None, pid=holder.pid, epoch=epoch)],
|
|
)
|
|
spy_log = _install_herdr_spy(mam_sandbox)
|
|
|
|
# Run 1: step 1 captures UA onto the row and record-id persists UA into reset-spec.json;
|
|
# step 3 refuses purge because lsof holds UA
|
|
res1 = _run_reset(mam_sandbox, "--yes")
|
|
_wait_reconcile_idle(mam_sandbox)
|
|
assert res1.returncode == 4, res1.stdout + res1.stderr
|
|
rows1 = _read_rows(mam_sandbox)
|
|
assert rows1[name_a]["status"] == "stopped"
|
|
assert rows1[name_a]["agy_conversation_id_own"] == UA
|
|
assert a_files[0].exists()
|
|
spec1 = json.loads((mam_sandbox / ".mam" / "reset-spec.json").read_text(encoding="utf-8"))
|
|
assert spec1["agents"][0]["old_id"] == UA
|
|
|
|
# Simulate resuming the skipped session (new created_at, own_id still UA) before Run 2 (review F1):
|
|
# because record-id persisted old_id=UA, Run 2 still treats it as 'pending' and purges UA
|
|
rows1[name_a]["status"] = "running"
|
|
rows1[name_a]["herdr_session_created_at"] = "2026-10-02T00:00:00Z"
|
|
_write_rows(mam_sandbox, [rows1[name_a]])
|
|
_mock_herdr_agent(mock_herdr, name_a, mam_sandbox, pid=holder.pid, epoch=epoch + 10, agent="agy", pane_id="w1:p_a")
|
|
|
|
# Release lsof and Run 2: step 3 purges UA (deletes all UA artifacts!) and creates new agy
|
|
lsof_exe.unlink()
|
|
finally:
|
|
holder.kill()
|
|
holder.wait()
|
|
|
|
res2 = _run_reset(mam_sandbox, "--yes")
|
|
_wait_reconcile_idle(mam_sandbox)
|
|
assert res2.returncode == 5, res2.stdout + res2.stderr
|
|
assert "created; id not recorded yet" in res2.stdout
|
|
for p in a_files:
|
|
assert not p.exists(), f"UA artifact should have been purged on rerun: {p}"
|
|
assert (mam_sandbox / ".mam" / "reset-spec.json").is_file()
|
|
|
|
# Run 3 (§5.2-9): rerunning while in `created` state performs 0 stop/create calls and exits 5
|
|
spy_log.unlink()
|
|
res3 = _run_reset(mam_sandbox, "--yes")
|
|
_wait_reconcile_idle(mam_sandbox)
|
|
assert res3.returncode == 5
|
|
assert _read_spy_log(spy_log) == []
|
|
assert f"reset [{name_a}]:" not in res3.stdout
|
|
|
|
# Record new ID UB onto the recreated row and Run 4 -> exits 0, OLD_ID=UA, NEW!=OLD='yes'
|
|
rows3 = _read_rows(mam_sandbox)
|
|
rows3[name_a]["agy_conversation_id_own"] = UB
|
|
_write_rows(mam_sandbox, [rows3[name_a]])
|
|
|
|
res4 = _run_reset(mam_sandbox, "--yes")
|
|
_wait_reconcile_idle(mam_sandbox)
|
|
assert res4.returncode == 0, res4.stdout + res4.stderr
|
|
assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()
|
|
tbl_line = [l for l in res4.stdout.splitlines() if l.startswith(name_a)][0]
|
|
assert tbl_line.split() == [name_a, "agy", UA, UB, "yes", "yes", "done"]
|
|
|
|
|
|
def test_post_discovered_uncapturable_id_8b_ii(
|
|
mam_sandbox: Path, mock_herdr: Path, mock_agents
|
|
):
|
|
"""§5.2-8b(ii): agy row has no recorded own_id and no conversation on disk -> step 2 skips
|
|
with 'skipped: no conversation id recorded' (exit 4), keeping the stopped row without
|
|
calling purge; manual purge followed by rerun recreates via D6 #1 (purged).
|
|
"""
|
|
name_b = "reviewer-agy-02"
|
|
epoch = int(time.time()) - 100
|
|
_mock_herdr_agent(mock_herdr, name_b, mam_sandbox, epoch=epoch, agent="agy", pane_id="w1:p_b")
|
|
_write_rows(mam_sandbox, [_row(name_b, "agy", mam_sandbox, own_id=None, epoch=epoch)])
|
|
spy_log = _install_herdr_spy(mam_sandbox)
|
|
|
|
res_ii = _run_reset(mam_sandbox, "--yes")
|
|
_wait_reconcile_idle(mam_sandbox)
|
|
assert res_ii.returncode == 4, res_ii.stdout + res_ii.stderr
|
|
assert "skipped: no conversation id recorded" in res_ii.stdout
|
|
assert not any(
|
|
l.startswith("reset [") and "--purge-conversation" in l for l in res_ii.stdout.splitlines()
|
|
)
|
|
rows_ii = _read_rows(mam_sandbox)
|
|
assert name_b in rows_ii and rows_ii[name_b]["status"] == "stopped"
|
|
|
|
# Manual purge drops the row; rerunning reset then sees D6 #1 (purged) and runs create only
|
|
stop_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh"
|
|
res_manual = subprocess.run(
|
|
["bash", str(stop_sh), "--session", name_b, "--agent", "agy", "--purge-conversation", "--yes"],
|
|
cwd=str(mam_sandbox),
|
|
env=dict(os.environ, SKS_QUIESCENT_TRIES="0", SKS_DIALOG_TIMEOUT="0", SKS_BUSY_TIMEOUT="0"),
|
|
stdin=subprocess.DEVNULL,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=60,
|
|
)
|
|
assert res_manual.returncode == 0, res_manual.stdout + res_manual.stderr
|
|
assert name_b not in _read_rows(mam_sandbox)
|
|
|
|
spy_log.unlink()
|
|
res_ii_rerun = _run_reset(mam_sandbox, "--yes")
|
|
_wait_reconcile_idle(mam_sandbox)
|
|
assert res_ii_rerun.returncode == 5, res_ii_rerun.stdout + res_ii_rerun.stderr
|
|
calls_ii = _read_spy_log(spy_log)
|
|
assert not any(c["verb"] in ("pane close", "kill-session") for c in calls_ii)
|
|
assert [c["args"][2] for c in calls_ii if c["verb"] == "agent start"] == [name_b]
|
|
assert name_b in _read_rows(mam_sandbox)
|
|
|
|
# Recording new ID UB when old_id remained None produces OLD_ID='-' and NEW!=OLD='yes'
|
|
rows_ii_after = _read_rows(mam_sandbox)
|
|
rows_ii_after[name_b]["agy_conversation_id_own"] = UB
|
|
_write_rows(mam_sandbox, [rows_ii_after[name_b]])
|
|
res_ii_done = _run_reset(mam_sandbox, "--yes")
|
|
_wait_reconcile_idle(mam_sandbox)
|
|
assert res_ii_done.returncode == 0, res_ii_done.stdout + res_ii_done.stderr
|
|
assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()
|
|
tbl_ii = [l for l in res_ii_done.stdout.splitlines() if l.startswith(name_b)][0]
|
|
assert tbl_ii.split() == [name_b, "agy", "-", UB, "yes", "yes", "done"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"sig_name,expected_rc",
|
|
[("SIGTERM", 143), ("SIGINT", 130), ("SIGHUP", 129)],
|
|
)
|
|
def test_signal_interrupts_reset_preserves_spec_and_holds_lock_until_child_exits(
|
|
mam_sandbox: Path, mock_herdr: Path, mock_agents, sig_name: str, expected_rc: int
|
|
):
|
|
"""Review P1 / N1:
|
|
When SIGTERM, SIGINT, or SIGHUP arrives while an in-flight stop/create child is running:
|
|
- reset.sh keeps holding .mam/loop-guard-active (so a competing run_loop.sh is still refused)
|
|
until the in-flight child finishes;
|
|
- once the child finishes, reset.sh immediately exits with the signal's nonzero code (143/130/129),
|
|
releasing .mam/loop-guard-active on exit, preserving .mam/reset-spec.json, and NOT continuing
|
|
to purge/create the current or subsequent targets.
|
|
"""
|
|
import signal
|
|
|
|
sig = getattr(signal, sig_name)
|
|
name_a = "reviewer-claude-01"
|
|
name_b = "reviewer-claude-02"
|
|
tf_a = _claude_transcript(mam_sandbox, mam_sandbox, UA)
|
|
tf_b = _claude_transcript(mam_sandbox, mam_sandbox, UB)
|
|
_mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a")
|
|
_mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="claude", pane_id="w1:p_b")
|
|
_write_rows(
|
|
mam_sandbox,
|
|
[
|
|
_row(name_a, "claude", mam_sandbox, own_id=UA),
|
|
_row(name_b, "claude", mam_sandbox, own_id=UB),
|
|
],
|
|
)
|
|
|
|
# Wrap bin/herdr so the first kill-session blocks until release_gate is created
|
|
bin_dir = mam_sandbox / "bin"
|
|
herdr_bin = bin_dir / "herdr"
|
|
herdr_real = bin_dir / "herdr.real"
|
|
if not herdr_real.exists():
|
|
herdr_bin.rename(herdr_real)
|
|
entered_gate = mam_sandbox / "child_entered"
|
|
release_gate = mam_sandbox / "child_release"
|
|
spy_log = mam_sandbox / "herdr_spy.jsonl"
|
|
|
|
herdr_bin.write_text(
|
|
f"""#!{sys.executable}
|
|
import json, os, sys, time
|
|
|
|
real = {repr(str(herdr_real))}
|
|
entered = {repr(str(entered_gate))}
|
|
release = {repr(str(release_gate))}
|
|
spy_log = {repr(str(spy_log))}
|
|
|
|
raw_args = sys.argv[1:]
|
|
args = list(raw_args)
|
|
while args and args[0] in ("-L", "--server", "-s", "--session"):
|
|
args = args[2:] if len(args) > 1 else args[1:]
|
|
verb = args[0] if (args and args[0] in ("kill-session", "has-session", "list-panes", "capture-pane", "send-keys", "ls")) else (" ".join(args[:2]) if len(args) >= 2 else "")
|
|
with open(spy_log, "a", encoding="utf-8") as lf:
|
|
lf.write(json.dumps({{"verb": verb, "args": args}}) + "\\n")
|
|
|
|
if verb == "kill-session" and not os.path.exists(entered):
|
|
with open(entered, "w", encoding="utf-8") as fh:
|
|
fh.write("1")
|
|
deadline = time.time() + 15.0
|
|
while not os.path.exists(release) and time.time() < deadline:
|
|
time.sleep(0.05)
|
|
|
|
os.execv(real, [real, *raw_args])
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
herdr_bin.chmod(0o755)
|
|
|
|
script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-reset" / "scripts" / "reset.sh"
|
|
env = dict(
|
|
os.environ,
|
|
SKS_QUIESCENT_TRIES="0",
|
|
SKS_DIALOG_TIMEOUT="0",
|
|
SKS_BUSY_TIMEOUT="0",
|
|
MAM_ONBOARD_ACK_S="3",
|
|
MAM_ONBOARD_WAIT_S="3",
|
|
MAM_NO_SUBSCRIBER_DAEMON="1",
|
|
PYTHONDONTWRITEBYTECODE="1",
|
|
)
|
|
proc = subprocess.Popen(
|
|
["bash", str(script), "--yes"],
|
|
cwd=str(mam_sandbox),
|
|
env=env,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
try:
|
|
deadline = time.time() + 15.0
|
|
while not entered_gate.exists() and time.time() < deadline:
|
|
assert proc.poll() is None, "reset.sh exited before reaching kill-session"
|
|
time.sleep(0.05)
|
|
assert entered_gate.exists(), "timed out waiting for kill-session child"
|
|
|
|
# Signal the reset parent while the foreground child is still blocked
|
|
proc.send_signal(sig)
|
|
time.sleep(0.2)
|
|
|
|
# While the child is still running, reset.sh has not exited and still holds the loop lock
|
|
assert proc.poll() is None
|
|
assert (mam_sandbox / ".mam" / "loop-guard-active").is_file()
|
|
if sig_name == "SIGTERM":
|
|
loop_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh"
|
|
competing = subprocess.run(
|
|
["bash", str(loop_sh), "--creator", "x", "--task", "t"],
|
|
cwd=str(mam_sandbox),
|
|
env=dict(os.environ, MAM_LOOP_NO_FREEZE="1"),
|
|
stdin=subprocess.DEVNULL,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=20,
|
|
)
|
|
assert competing.returncode == 1
|
|
assert "Another multi-agent-mux-loop is already running" in (competing.stdout + competing.stderr)
|
|
|
|
# Release the child and wait for reset.sh to terminate
|
|
release_gate.write_text("1", encoding="utf-8")
|
|
stdout, stderr = proc.communicate(timeout=30)
|
|
finally:
|
|
if proc.poll() is None:
|
|
proc.kill()
|
|
proc.wait()
|
|
_wait_reconcile_idle(mam_sandbox)
|
|
|
|
assert proc.returncode == expected_rc, f"rc={proc.returncode}, stdout={stdout}, stderr={stderr}"
|
|
assert not (mam_sandbox / ".mam" / "loop-guard-active").exists()
|
|
assert (mam_sandbox / ".mam" / "reset-spec.json").is_file()
|
|
# Neither transcript was purged, and no create or second-target operation ran
|
|
assert tf_a.is_file()
|
|
assert tf_b.is_file()
|
|
calls = _read_spy_log(spy_log)
|
|
assert not any(c["verb"] == "agent start" for c in calls)
|
|
assert f"reset [{name_b}]:" not in stdout
|
|
assert not any(
|
|
l.startswith("reset [") and "--purge-conversation" in l for l in stdout.splitlines()
|
|
)
|