Files
multi-agent-mux/tests/test_issue63_same_cwd_ids.py
T
GodopuandClaude Opus 5.5 1614af5b67 fix(id_order): a dead-owner marker on the caller's own session follows the orphan rules (Issue #63)
A marker left on the invoking session name by a killed create/resume was
pruned before herdr was asked, so a retry under the same name took it over
while the orphaned pane was still alive (and create's failure cleanup could
then kill it). It is now kept and blocks like any other orphan until its
row is registered for the attempt or herdr reports the session gone.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-23 11:47:04 +09:00

1321 lines
58 KiB
Python

"""Issue #63: same-cwd conversation-id resolution in stop/purge/resume and
same-type creation ordering for post-discovered agents.
"""
from __future__ import annotations
import json
import os
import sqlite3
import subprocess
import sys
import time
from pathlib import Path
from typing import 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 = "a6300001-0000-4000-8000-000000000001"
UB = "b6300002-0000-4000-8000-000000000002"
UC = "c6300003-0000-4000-8000-000000000003"
UX = "d6300004-0000-4000-8000-000000000004"
pytestmark = pytest.mark.usefixtures("own_workspace_root", "no_herdr_probe")
@pytest.fixture
def no_herdr_probe(monkeypatch):
"""In-process id_order.wait never asks the real herdr about a dead-owner
marker's session: it reads as gone unless a test passes pane_state."""
from lib_py.agents import id_order
monkeypatch.setattr(id_order, "herdr_session_state", lambda server, session, timeout: False)
# --------------------------------------------------------------------------- helpers
def _agy_conv(home: Path, uuid: str, mtime: Optional[float] = None) -> Path:
db = Path(home) / ".gemini" / "antigravity-cli" / "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))
return db
def _agy_full_artifacts(home: Path, uuid: str) -> List[Path]:
base = Path(home) / ".gemini" / "antigravity-cli"
db = _agy_conv(home, uuid)
wal = Path(str(db) + "-wal")
shm = Path(str(db) + "-shm")
wal.write_text("wal")
shm.write_text("shm")
brain = base / "brain" / uuid
brain.mkdir(parents=True, exist_ok=True)
(brain / "note.md").write_text("note")
ann = base / "annotations" / f"{uuid}.pbtxt"
ann.parent.mkdir(parents=True, exist_ok=True)
ann.write_text("pbtxt")
lock = base / "presence" / f"{uuid}.lock"
lock.parent.mkdir(parents=True, exist_ok=True)
lock.write_text("")
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({str(k): v for k, v in mapping.items()}))
def _claude_transcript(claude_dir: Path, ws: Path, uuid: str, mtime: Optional[float] = None) -> Path:
from lib_py.verify_session import workspace_key
pdir = Path(claude_dir) / 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")
if mtime is not None:
os.utime(f, (mtime, mtime))
return f
def _row(name: str, agent: str, cwd: Path, epoch: int, pid: int = 4242, **extra) -> dict:
own = {"agy": "agy_conversation_id_own", "opencode": "opencode_session_id_own",
"codex": "codex_session_id_own", "hermes": "hermes_conversation_id_own",
"claude": "claude_session_id_own", "grok": "grok_session_id_own"}[agent]
row = {"name": name, "status": "running", "role": "creator", "agent": agent,
"herdr_session_epoch": epoch, own: None,
"session_id_source": "pending-discovery", "session_id_verified": False,
"last_visible_status": "unverified",
"pane": {"cwd": str(cwd), "pid": pid, "cmd": agent}}
row.update(extra)
return row
# --------------------------------------------------------------------------- 2.1 workspace_uuid
def test_find_workspace_uuid_skips_exclude_ids(tmp_path):
"""rev 2 (B): find_workspace_uuid skips exclude_ids across sibling rows and discover()."""
from lib_py.workspace_uuid import find_workspace_uuid
home = tmp_path / "home"
claude_dir = home / ".claude" / "projects"
ws = tmp_path / "ws"
ws.mkdir(parents=True)
now = time.time()
_claude_transcript(claude_dir, ws, UA, mtime=now)
_claude_transcript(claude_dir, ws, UB, mtime=now - 10)
state = {
"herdr_sessions": [
_row("ws-creator-claude", "claude", ws, int(now) - 100, status="stopped",
claude_session_id_own=UA, session_id_source="discovered", session_id_verified=True),
_row("ws-reviewer-claude", "claude", ws, int(now) - 100, status="running",
claude_session_id_own=None),
]
}
assert find_workspace_uuid(str(ws), "claude", "ws-reviewer-claude", state,
str(home), str(claude_dir), exclude_ids=()) == UA
assert find_workspace_uuid(str(ws), "claude", "ws-reviewer-claude", state,
str(home), str(claude_dir), exclude_ids={UA}) == UB
def test_find_workspace_uuid_uses_explicit_home_and_claude_dir(tmp_path, monkeypatch):
"""rev 2 (B-sub): explicit home and claude_dir are forwarded to tier-1 revalidate."""
from lib_py.workspace_uuid import find_workspace_uuid
monkeypatch.setenv("HOME", str(tmp_path / "wrong-home"))
monkeypatch.delenv("HOME_DIR", raising=False)
monkeypatch.delenv("CLAUDE_PROJECT_DIR", raising=False)
home = tmp_path / "custom-home"
claude_dir = tmp_path / "custom-claude"
ws = tmp_path / "ws"
ws.mkdir(parents=True)
_agy_conv(home, UA)
_claude_transcript(claude_dir, ws, UB)
state = {
"herdr_sessions": [
_row("ws-creator-agy", "agy", ws, 100, status="stopped",
agy_conversation_id_own=UA, session_id_source="discovered", session_id_verified=True),
_row("ws-creator-claude", "claude", ws, 100, status="stopped",
claude_session_id_own=UB, session_id_source="discovered", session_id_verified=True),
]
}
assert find_workspace_uuid(str(ws), "agy", "ws-creator-agy", state,
home=str(home), claude_dir=str(claude_dir)) == UA
assert find_workspace_uuid(str(ws), "claude", "ws-creator-claude", state,
home=str(home), claude_dir=str(claude_dir)) == UB
# --------------------------------------------------------------------------- 2.2 session_ids unit tests
@pytest.fixture
def owners(monkeypatch):
"""agy presence ownership without lsof/ps: {uuid: [holder pids]} and a
pid -> ppid table the test fills in."""
from lib_py.agents.adapters import agy
held, parents = {}, {}
def holders(paths):
if held.get("__lsof__") is None and "__lsof__" in held:
return None
return sorted({p for path in paths for u, pids in held.items() if u in path for p in pids})
monkeypatch.setattr(agy, "_holder_pids", holders)
monkeypatch.setattr(agy, "_process_parents", lambda: dict(parents) or None)
return held, parents
def test_session_ids_decide_stop_and_resume_unit(tmp_path, owners):
from lib_py.agents.session_ids import decide_resume, decide_stop
home = tmp_path / "home"
ws = tmp_path / "ws"
ws.mkdir(parents=True)
now = time.time()
epoch = int(now) - 100
_agy_conv(home, UA, mtime=now - 5)
_agy_conv(home, UB, mtime=now - 2)
_agy_cache(home, {ws: UA})
held, parents = owners
held[UA] = [5001]
held[UB] = [6001]
parents.update({5001: 1, 6001: 1})
state = {
"herdr_sessions": [
_row("ws-creator-agy", "agy", ws, epoch, pid=5001),
_row("ws-reviewer-agy", "agy", ws, epoch, pid=6001),
]
}
res_a = decide_stop(state, "ws-creator-agy", "agy", str(ws), True, pane_pid=5001, purge=False, home=str(home))
assert res_a["captured"] == UA and res_a["status"] == "captured"
res_b = decide_stop(state, "ws-reviewer-agy", "agy", str(ws), True, pane_pid=6001, purge=False, home=str(home))
assert res_b["captured"] == "" and res_b["status"] == "owner-mismatch"
res_dead = decide_stop(state, "ws-reviewer-agy", "agy", str(ws), False, pane_pid=None, purge=True, home=str(home))
assert res_dead["captured"] == "" and res_dead["purge_uuid"] == ""
assert "herdr session was already gone" in res_dead["note"]
# Two same-type rows in one cwd and B has no own id -> fresh spawn with NOTE
state["herdr_sessions"][0]["status"] = "stopped"
state["herdr_sessions"][0]["agy_conversation_id_own"] = UA
state["herdr_sessions"][1]["status"] = "stopped"
r_b = decide_resume(state, "ws-reviewer-agy", "agy", str(ws), herdr_alive=False, home=str(home))
assert r_b["uuid"] == "" and r_b["error"] is None and not r_b["drop_stale_id"]
assert any("2 agy sessions share" in n for n in r_b["notes"])
def test_session_ids_own_id_priority_rules_unit(tmp_path):
"""rev 2 (A): own id never falls through to tier-2; broken registry refuses even when DB is missing."""
from lib_py.agents.session_ids import decide_resume, decide_stop
home = tmp_path / "home"
ws = tmp_path / "ws"
ws.mkdir(parents=True)
epoch = int(time.time()) - 100
# Missing UA.db on disk, but both rows claim UA
state_dup = {
"herdr_sessions": [
_row("ws-creator-agy", "agy", ws, epoch, status="stopped", agy_conversation_id_own=UA,
session_id_source="discovered", session_id_verified=True),
_row("ws-reviewer-agy", "agy", ws, epoch, status="stopped", agy_conversation_id_own=UA,
session_id_source="discovered", session_id_verified=True),
]
}
r_dup = decide_resume(state_dup, "ws-reviewer-agy", "agy", str(ws), herdr_alive=False, home=str(home))
assert r_dup["error"] and "ws-creator-agy" in r_dup["error"]
assert r_dup["notes"] == []
s_dup = decide_stop(state_dup, "ws-reviewer-agy", "agy", str(ws), False, purge=True, home=str(home))
assert s_dup["captured"] == "" and s_dup["purge_uuid"] == ""
assert "also recorded by ws-creator-agy" in s_dup["note"]
# Single row with stale own id UX (missing UX.db) while cache has UC
_agy_conv(home, UC)
_agy_cache(home, {ws: UC})
state_stale = {
"herdr_sessions": [
_row("ws-creator-agy", "agy", ws, epoch, status="stopped", agy_conversation_id_own=UX,
session_id_source="discovered", session_id_verified=True),
]
}
r_stale = decide_resume(state_stale, "ws-creator-agy", "agy", str(ws), herdr_alive=False, home=str(home))
assert r_stale["uuid"] == "" and r_stale["drop_stale_id"] is True and r_stale["error"] is None
assert any(f"recorded agy id {UX} no longer verifies" in n for n in r_stale["notes"])
s_stale = decide_stop(state_stale, "ws-creator-agy", "agy", str(ws), False, purge=False, home=str(home))
assert s_stale["captured"] == "" and "no longer verifies; kept as is" in s_stale["note"]
# --------------------------------------------------------------------------- Done Criteria 1 & 2 (stop & purge)
def _fake_lsof_multi(bindir: Path, mapping: dict) -> None:
"""Fake lsof that prints the pid from `mapping` ({uuid_substr: pid}) for any matching 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"
)
exe.chmod(0o755)
def _mock_herdr_agent(state_file: Path, name: str, cwd: Path, pid: int, epoch: int,
agent: str = "agy", pane_id: str = "") -> None:
state = json.loads(Path(state_file).read_text())
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"Antigravity {cwd}\n>",
}
Path(state_file).write_text(json.dumps(state))
def _write_rows(sandbox: Path, rows: list) -> Path:
lib = sandbox / ".agents" / "skills" / "lib.sh"
yaml_path = sandbox / ".mam" / "agent-sessions.yaml"
mutation = f"d['herdr_sessions'] = {repr(rows)}\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:
yaml_path = sandbox / ".mam" / "agent-sessions.yaml"
data = yaml.safe_load(yaml_path.read_text()) or {}
return {r["name"]: r for r in data.get("herdr_sessions") or []}
def _run_stop(sandbox: Path, session: str, agent: str = "agy", purge: bool = False) -> subprocess.CompletedProcess:
script = sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh"
cmd = ["bash", str(script), "--session", session, "--agent", agent]
if purge:
cmd += ["--purge-conversation", "--yes"]
env = dict(os.environ, SKS_QUIESCENT_TRIES="0", SKS_DIALOG_TIMEOUT="0", SKS_BUSY_TIMEOUT="0")
return subprocess.run(cmd, capture_output=True, text=True, cwd=str(sandbox), env=env, timeout=120)
@pytest.mark.parametrize("order", [("a_then_b", True), ("b_then_a", True), ("both_dead", False)])
def test_c1_stop_same_cwd_agy_sessions_records_only_verified_owner(mam_sandbox, mock_herdr, order):
"""Done Criterion 1: two agy sessions in one cwd; cache points to A's UA.
Stop A->B and B->A with herdr alive records UA on A and None on B; when
herdr is already dead for both, neither records an id and both leave a reason."""
label, alive = order
ws = mam_sandbox / "same-ws"
ws.mkdir(parents=True, exist_ok=True)
now = time.time()
epoch = int(now) - 100
_agy_full_artifacts(mam_sandbox, UA)
_agy_full_artifacts(mam_sandbox, UB)
_agy_cache(mam_sandbox, {ws: UA})
proc_a = subprocess.Popen(["sleep", "120"])
proc_b = subprocess.Popen(["sleep", "120"])
try:
_fake_lsof_multi(mam_sandbox / "bin", {UA: proc_a.pid, UB: proc_b.pid})
name_a, name_b = "same-ws-creator-agy", "same-ws-reviewer-agy"
_write_rows(mam_sandbox, [
_row(name_a, "agy", ws, epoch, pid=proc_a.pid),
_row(name_b, "agy", ws, epoch, pid=proc_b.pid),
])
if alive:
_mock_herdr_agent(mock_herdr, name_a, ws, proc_a.pid, epoch, pane_id="w1:p_a")
_mock_herdr_agent(mock_herdr, name_b, ws, proc_b.pid, epoch, pane_id="w1:p_b")
seq = [name_a, name_b] if label != "b_then_a" else [name_b, name_a]
for sname in seq:
res = _run_stop(mam_sandbox, sname, "agy")
assert res.returncode == 0, res.stdout + res.stderr
rows = _read_rows(mam_sandbox)
if alive:
assert rows[name_a]["agy_conversation_id_own"] == UA
assert rows[name_a]["resumable"] is True
assert rows[name_b]["agy_conversation_id_own"] is None
assert rows[name_b]["last_visible_status"].startswith("stop: ")
else:
assert rows[name_a]["agy_conversation_id_own"] is None
assert rows[name_b]["agy_conversation_id_own"] is None
assert "herdr session was already gone" in rows[name_a]["last_visible_status"]
assert "herdr session was already gone" in rows[name_b]["last_visible_status"]
finally:
for p in (proc_a, proc_b):
p.kill()
p.wait()
@pytest.mark.parametrize("broken_dup", [False, True])
def test_c2_purge_never_deletes_another_rows_conversation(mam_sandbox, mock_herdr, broken_dup):
"""Done Criterion 2: A is stopped with UA; B has no own id (or also has UA in a
broken registry) and herdr is dead. Purging B leaves A's conversation intact."""
ws = mam_sandbox / "same-ws"
ws.mkdir(parents=True, exist_ok=True)
epoch = int(time.time()) - 100
a_files = _agy_full_artifacts(mam_sandbox, UA)
_agy_cache(mam_sandbox, {ws: UA})
name_a, name_b = "same-ws-creator-agy", "same-ws-reviewer-agy"
row_a = _row(name_a, "agy", ws, epoch, status="stopped", agy_conversation_id_own=UA,
session_id_source="discovered", session_id_verified=True)
row_b = _row(name_b, "agy", ws, epoch, status="running",
agy_conversation_id_own=(UA if broken_dup else None))
_write_rows(mam_sandbox, [row_a, row_b])
res = _run_stop(mam_sandbox, name_b, "agy", purge=True)
assert res.returncode == 0, res.stdout + res.stderr
assert "nothing purged" in res.stdout
for p in a_files:
assert p.exists(), f"A's artifact was deleted by B's purge: {p}"
rows = _read_rows(mam_sandbox)
assert name_b not in rows
assert rows[name_a]["agy_conversation_id_own"] == UA
# --------------------------------------------------------------------------- Done Criteria 3, 4, 7 & rev 2 (resume)
def _run_resume(sandbox: Path, ws: Path, session: str, agent: str = "agy",
dry_run: bool = False) -> subprocess.CompletedProcess:
script = sandbox / ".agents" / "skills" / "multi-agent-mux-resume" / "scripts" / "resume_session.sh"
cmd = ["bash", str(script), "--workspace", str(ws), "--agent", agent, "--session", session]
if dry_run:
cmd.append("--dry-run")
return subprocess.run(cmd, capture_output=True, text=True, cwd=str(sandbox), timeout=120)
def test_c3_resume_unpinned_second_agy_row_starts_fresh_and_running_branch_keeps_epoch(
mam_sandbox, mock_herdr, mock_agents
):
"""Done Criterion 3: two stopped agy rows in one cwd (A=UA, B=None):
dry-run and real resume of B spawn without --conversation UA and emit NOTE;
when B's herdr is already alive without an id, resume does not record UA and
preserves B's herdr_session_epoch."""
ws = mam_sandbox / "same-ws"
ws.mkdir(parents=True, exist_ok=True)
epoch_a = int(time.time()) - 300
epoch_b = int(time.time()) - 200
_agy_full_artifacts(mam_sandbox, UA)
_agy_cache(mam_sandbox, {ws: UA})
name_a, name_b = "same-ws-creator-agy", "same-ws-reviewer-agy"
_write_rows(mam_sandbox, [
_row(name_a, "agy", ws, epoch_a, status="stopped", agy_conversation_id_own=UA,
session_id_source="discovered", session_id_verified=True),
_row(name_b, "agy", ws, epoch_b, status="stopped", agy_conversation_id_own=None),
])
dry = _run_resume(mam_sandbox, ws, name_b, "agy", dry_run=True)
assert dry.returncode == 0, dry.stdout + dry.stderr
assert "--conversation" not in dry.stdout
assert f"NOTE: {name_b} has no recorded agy id and 2 agy sessions share" in dry.stderr
real = _run_resume(mam_sandbox, ws, name_b, "agy", dry_run=False)
assert real.returncode == 0, real.stdout + real.stderr
rows = _read_rows(mam_sandbox)
assert rows[name_b]["status"] == "running"
assert rows[name_b]["agy_conversation_id_own"] is None
assert "--conversation" not in rows[name_b]["pane"]["cmd_full"]
# Running branch: B is already alive in herdr without an own id; epoch must stay fixed.
fixed_epoch = 1700000123
_write_rows(mam_sandbox, [
_row(name_a, "agy", ws, epoch_a, status="stopped", agy_conversation_id_own=UA,
session_id_source="discovered", session_id_verified=True),
_row(name_b, "agy", ws, fixed_epoch, status="running", agy_conversation_id_own=None),
])
_mock_herdr_agent(mock_herdr, name_b, ws, 9999, fixed_epoch, pane_id="w1:p_b")
alive_res = _run_resume(mam_sandbox, ws, name_b, "agy", dry_run=False)
assert alive_res.returncode == 0, alive_res.stdout + alive_res.stderr
rows = _read_rows(mam_sandbox)
assert rows[name_b]["agy_conversation_id_own"] is None
assert rows[name_b]["herdr_session_epoch"] == fixed_epoch
assert rows[name_b]["last_visible_status"].startswith("already running; no id recorded at ")
def test_c4_pinned_post_discovered_and_preassigned_agents_unchanged(mam_sandbox, mock_herdr, mock_agents):
"""Done Criterion 4: agy/opencode rows with own_id keep it on stop and resume with it;
claude/grok stop/resume and find_workspace_uuid behave as before."""
ws = mam_sandbox / "c4-ws"
ws.mkdir(parents=True, exist_ok=True)
epoch = int(time.time()) - 100
_agy_full_artifacts(mam_sandbox, UA)
_claude_transcript(mam_sandbox / ".claude" / "projects", ws, UB)
# Seed opencode DB for ses_c4
oc_db = mam_sandbox / ".local" / "share" / "opencode" / "opencode.db"
oc_db.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(oc_db))
conn.execute("CREATE TABLE IF NOT EXISTS session_v2 (id TEXT PRIMARY KEY, parent_id TEXT, directory TEXT, time_created INTEGER)")
conn.execute("INSERT OR REPLACE INTO session_v2 VALUES (?, NULL, ?, ?)", ("ses_c4", str(ws), epoch * 1000))
conn.commit()
conn.close()
_write_rows(mam_sandbox, [
_row("c4-creator-agy", "agy", ws, epoch, status="running", agy_conversation_id_own=UA,
session_id_source="discovered", session_id_verified=True),
_row("c4-creator-opencode", "opencode", ws, epoch, status="running", opencode_session_id_own="ses_c4",
session_id_source="discovered", session_id_verified=True),
_row("c4-creator-claude", "claude", ws, epoch, status="running", claude_session_id_own=UB,
session_id_source="assigned", session_id_verified=True),
])
for sname, agent in (("c4-creator-agy", "agy"), ("c4-creator-opencode", "opencode"), ("c4-creator-claude", "claude")):
st = _run_stop(mam_sandbox, sname, agent)
assert st.returncode == 0, st.stdout + st.stderr
rows = _read_rows(mam_sandbox)
assert rows["c4-creator-agy"]["agy_conversation_id_own"] == UA and rows["c4-creator-agy"]["resumable"] is True
assert rows["c4-creator-opencode"]["opencode_session_id_own"] == "ses_c4" and rows["c4-creator-opencode"]["resumable"] is True
assert rows["c4-creator-claude"]["claude_session_id_own"] == UB and rows["c4-creator-claude"]["resumable"] is True
r_agy = _run_resume(mam_sandbox, ws, "c4-creator-agy", "agy", dry_run=True)
assert r_agy.returncode == 0 and f"--conversation {UA}" in r_agy.stdout
r_oc = _run_resume(mam_sandbox, ws, "c4-creator-opencode", "opencode", dry_run=True)
assert r_oc.returncode == 0 and "--session ses_c4" in r_oc.stdout
r_cl = _run_resume(mam_sandbox, ws, "c4-creator-claude", "claude", dry_run=True)
assert r_cl.returncode == 0 and f"-r {UB}" in r_cl.stdout
@pytest.mark.parametrize("db_exists", [True, False])
def test_c7_resume_refuses_when_recorded_id_is_claimed_by_another_row(
mam_sandbox, mock_herdr, mock_agents, db_exists
):
"""Done Criterion 7 & rev 2 (A): when both A and B record UA (whether UA.db
exists on disk or is missing), resume B refuses with exit 1 before spawning,
names A on stderr, and emits no fresh-spawn NOTE."""
ws = mam_sandbox / "c7-ws"
ws.mkdir(parents=True, exist_ok=True)
epoch = int(time.time()) - 100
if db_exists:
_agy_full_artifacts(mam_sandbox, UA)
_agy_cache(mam_sandbox, {ws: UA})
name_a, name_b = "c7-ws-creator-agy", "c7-ws-reviewer-agy"
_write_rows(mam_sandbox, [
_row(name_a, "agy", ws, epoch, status="stopped", agy_conversation_id_own=UA,
session_id_source="discovered", session_id_verified=True),
_row(name_b, "agy", ws, epoch, status="stopped", agy_conversation_id_own=UA,
session_id_source="discovered", session_id_verified=True),
])
before = _read_rows(mam_sandbox)
res = _run_resume(mam_sandbox, ws, name_b, "agy", dry_run=False)
assert res.returncode == 1, res.stdout + res.stderr
assert name_a in res.stderr and UA in res.stderr
assert "NOTE:" not in res.stderr
state = json.loads(Path(mock_herdr).read_text())
assert name_b not in state.get("agents", {})
assert _read_rows(mam_sandbox) == before
def test_rev2_stale_own_id_single_row_never_overwrites_with_cache_and_clears_on_resume(
mam_sandbox, mock_herdr, mock_agents
):
"""rev 2 (A)-3: single agy row with stale own_id UX (UX.db missing) while cache
and disk hold UC. Stop keeps UX; resume starts a fresh conversation without UC,
emits NOTE, and clears agy_conversation_id_own to None so #40 can pin later."""
ws = mam_sandbox / "stale-ws"
ws.mkdir(parents=True, exist_ok=True)
epoch = int(time.time()) - 100
_agy_full_artifacts(mam_sandbox, UC)
_agy_cache(mam_sandbox, {ws: UC})
name = "stale-ws-creator-agy"
_write_rows(mam_sandbox, [
_row(name, "agy", ws, epoch, status="running", agy_conversation_id_own=UX,
session_id_source="discovered", session_id_verified=True),
])
st = _run_stop(mam_sandbox, name, "agy")
assert st.returncode == 0, st.stdout + st.stderr
row_after_stop = _read_rows(mam_sandbox)[name]
assert row_after_stop["agy_conversation_id_own"] == UX
assert "no longer verifies; kept as is" in row_after_stop["last_visible_status"]
res = _run_resume(mam_sandbox, ws, name, "agy", dry_run=False)
assert res.returncode == 0, res.stdout + res.stderr
assert f"NOTE: {name}'s recorded agy id {UX} no longer verifies" in res.stderr
row_after_resume = _read_rows(mam_sandbox)[name]
assert row_after_resume["agy_conversation_id_own"] is None
assert row_after_resume["session_id_source"] == "pending-discovery"
assert row_after_resume["session_id_verified"] is False
assert f"recorded id {UX} no longer verified" in row_after_resume["last_visible_status"]
assert UC not in row_after_resume["pane"]["cmd_full"]
def test_rev2_drop_stale_id_refuses_to_clear_id_claimed_by_another_row(mam_sandbox):
"""rev 2: update_yaml_resumed.sh --drop-stale-id keeps own_id if another row also claims it."""
ws = mam_sandbox / "dup-drop-ws"
ws.mkdir(parents=True, exist_ok=True)
epoch = int(time.time()) - 100
name_a, name_b = "dup-creator-agy", "dup-reviewer-agy"
_write_rows(mam_sandbox, [
_row(name_a, "agy", ws, epoch, status="stopped", agy_conversation_id_own=UA),
_row(name_b, "agy", ws, epoch, status="stopped", agy_conversation_id_own=UA),
])
upd = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-resume" / "scripts" / "update_yaml_resumed.sh"
res = subprocess.run(
["bash", str(upd), "--session", name_b, "--uuid", "", "--agent", "agy",
"--workspace", str(ws), "--drop-stale-id"],
capture_output=True, text=True, cwd=str(mam_sandbox), timeout=60,
)
assert res.returncode == 0, res.stdout + res.stderr
assert _read_rows(mam_sandbox)[name_b]["agy_conversation_id_own"] == UA
def test_rev2_claude_unpinned_sibling_row_skips_creator_claimed_id(mam_sandbox, mock_herdr, mock_agents):
"""rev 2 (B): ws-creator-claude is stopped with UA; ws-reviewer-claude has no
own_id (--wrapper row shape); disk has UA.jsonl and UB.jsonl. Resume and stop
of ws-reviewer-claude pick UB instead of creator's UA."""
ws = mam_sandbox / "claude-sib-ws"
ws.mkdir(parents=True, exist_ok=True)
now = time.time()
epoch = int(now) - 100
cdir = mam_sandbox / ".claude" / "projects"
_claude_transcript(cdir, ws, UA, mtime=now)
_claude_transcript(cdir, ws, UB, mtime=now - 10)
name_a, name_b = "claude-sib-ws-creator-claude", "claude-sib-ws-reviewer-claude"
_write_rows(mam_sandbox, [
_row(name_a, "claude", ws, epoch, status="stopped", claude_session_id_own=UA,
session_id_source="discovered", session_id_verified=True),
_row(name_b, "claude", ws, epoch, status="running", claude_session_id_own=None),
])
dry = _run_resume(mam_sandbox, ws, name_b, "claude", dry_run=True)
assert dry.returncode == 0, dry.stdout + dry.stderr
st = _run_stop(mam_sandbox, name_b, "claude")
assert st.returncode == 0, st.stdout + st.stderr
assert _read_rows(mam_sandbox)[name_b]["claude_session_id_own"] == UB
dry_after = _run_resume(mam_sandbox, ws, name_b, "claude", dry_run=True)
assert dry_after.returncode == 0, dry_after.stdout + dry_after.stderr
assert f"-r {UB}" in dry_after.stdout
# --------------------------------------------------------------------------- Done Criterion 6 (id_order & create/resume wait)
def test_c6_create_waits_or_refuses_while_same_cwd_post_discovered_row_has_no_id(
mam_sandbox, mock_herdr, mock_agents
):
"""Done Criterion 6 (a) & (b): A is running without an id and has a live pane pid.
(a) With MAM_ID_ORDER_WAIT_S=0, create B exits 1 without spawning or adding a row
and names A on stderr.
(b) With MAM_ID_ORDER_WAIT_S=30, once A records its id in a background thread,
create B proceeds (rc 0) and records herdr_session_epoch >= the pin time."""
import threading
ws = mam_sandbox / "c6-ws"
ws.mkdir(parents=True, exist_ok=True)
epoch_a = int(time.time()) - 100
name_a, name_b = "c6-ws-creator-agy", "c6-ws-reviewer-agy"
create_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
holder = subprocess.Popen(["sleep", "120"])
try:
_write_rows(mam_sandbox, [_row(name_a, "agy", ws, epoch_a, pid=holder.pid)])
_mock_herdr_agent(mock_herdr, name_a, ws, holder.pid, epoch_a, pane_id="w1:p_a")
# (a) Wait cap 0 -> immediate refusal
env0 = dict(os.environ, MAM_ID_ORDER_WAIT_S="0")
res0 = subprocess.run(
["bash", str(create_sh), "--workspace", str(ws), "--agent", "agy",
"--role", "reviewer", "--session", name_b, "--no-onboard"],
capture_output=True, text=True, cwd=str(mam_sandbox), env=env0, timeout=120,
)
assert res0.returncode == 1, res0.stdout + res0.stderr
assert name_a in res0.stderr
assert name_b not in _read_rows(mam_sandbox)
herdr_state = json.loads(Path(mock_herdr).read_text())
assert name_b not in herdr_state.get("agents", {})
# Also check resume --dry-run does not wait even with MAM_ID_ORDER_WAIT_S=0
_write_rows(mam_sandbox, [
_row(name_a, "agy", ws, epoch_a, pid=holder.pid),
_row(name_b, "agy", ws, epoch_a, status="stopped", agy_conversation_id_own=UB),
])
_agy_full_artifacts(mam_sandbox, UB)
dry_res = subprocess.run(
["bash", str(mam_sandbox / ".agents" / "skills" / "multi-agent-mux-resume" / "scripts" / "resume_session.sh"),
"--workspace", str(ws), "--agent", "agy", "--session", name_b, "--dry-run"],
capture_output=True, text=True, cwd=str(mam_sandbox), env=env0, timeout=60,
)
assert dry_res.returncode == 0, dry_res.stdout + dry_res.stderr
# Reset to only A running without id for (b)
_write_rows(mam_sandbox, [_row(name_a, "agy", ws, epoch_a, pid=holder.pid)])
pinned_at = []
def _pin_a_later():
time.sleep(1.5)
pinned_at.append(int(time.time()))
_write_rows(mam_sandbox, [
_row(name_a, "agy", ws, epoch_a, pid=holder.pid, agy_conversation_id_own=UA,
session_id_source="discovered", session_id_verified=True),
])
t = threading.Thread(target=_pin_a_later)
t.start()
try:
env30 = dict(os.environ, MAM_ID_ORDER_WAIT_S="30")
res_ok = subprocess.run(
["bash", str(create_sh), "--workspace", str(ws), "--agent", "agy",
"--role", "reviewer", "--session", name_b, "--no-onboard"],
capture_output=True, text=True, cwd=str(mam_sandbox), env=env30, timeout=120,
)
finally:
t.join(timeout=10)
assert res_ok.returncode == 0, res_ok.stdout + res_ok.stderr
assert "id-order: waiting for" in res_ok.stderr and name_a in res_ok.stderr
rows = _read_rows(mam_sandbox)
assert name_b in rows and rows[name_b]["status"] == "running"
assert pinned_at and rows[name_b]["herdr_session_epoch"] >= pinned_at[0]
finally:
holder.kill()
holder.wait()
def test_c6_id_order_markers_non_blocking_cases_and_uuid_recheck(tmp_path, monkeypatch):
"""Done Criterion 6 (c), (d), and Criterion 7 uuid recheck during wait + knob parsing."""
from lib_py.agents import id_order
home = tmp_path / "home"
ws = tmp_path / "ws"
ws_other = tmp_path / "ws-other"
ws.mkdir(parents=True)
ws_other.mkdir(parents=True)
yaml_path = tmp_path / "agent-sessions.yaml"
yaml_path.write_text(yaml.safe_dump({"herdr_sessions": []}))
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("AGENT_SESSIONS_YAML", str(yaml_path))
# Knob parsing rules
assert id_order.wait_s({}) == 600
assert id_order.wait_s({"MAM_ID_ORDER_WAIT_S": "0"}) == 0
assert id_order.wait_s({"MAM_ID_ORDER_WAIT_S": " 05 "}) == 5
for bad in ("abc", "-1", "9" * 10, ""):
assert id_order.wait_s({"MAM_ID_ORDER_WAIT_S": bad}) == 600, bad
# (c) Marker-only "being created": blocks while owner_pid is alive, passes once dead
dpath = id_order.order_dir(str(ws), "agy", home=str(home))
proc = subprocess.Popen(["sleep", "120"])
try:
id_order.write_marker(dpath, "creating-a-agy", proc.pid, str(ws), "agy")
assert id_order.wait(str(ws), "agy", "creating-b-agy", os.getpid(),
yaml_path=str(yaml_path), home=str(home), wait_s_val=0) == 1
finally:
proc.kill()
proc.wait()
# Dead owner_pid marker is pruned and wait succeeds
assert id_order.wait(str(ws), "agy", "creating-b-agy", os.getpid(),
yaml_path=str(yaml_path), home=str(home), wait_s_val=0) == 0
# Clean up creating-b-agy's own marker for next sub-checks
for f in Path(dpath).glob("*.json"):
f.unlink()
# (d) Non-blocking cases: claude, different folder, different agent, dead pane pid, pinned row
dead_proc = subprocess.Popen(["true"])
dead_proc.wait()
live_proc = subprocess.Popen(["sleep", "120"])
try:
state_nonblock = {
"herdr_sessions": [
_row("other-cwd-agy", "agy", ws_other, 100, pid=live_proc.pid),
_row("same-cwd-codex", "codex", ws, 100, pid=live_proc.pid),
_row("dead-pane-agy", "agy", ws, 100, pid=dead_proc.pid),
_row("pinned-agy", "agy", ws, 100, pid=live_proc.pid, agy_conversation_id_own=UA),
]
}
yaml_path.write_text(yaml.safe_dump(state_nonblock))
assert id_order.wait(str(ws), "claude", "new-claude", os.getpid(),
yaml_path=str(yaml_path), home=str(home), wait_s_val=0) == 0
assert id_order.wait(str(ws), "agy", "new-agy", os.getpid(),
yaml_path=str(yaml_path), home=str(home), wait_s_val=0) == 0
# Criterion 7: wait --uuid refuses when another row already claimed that uuid
assert id_order.wait(str(ws), "agy", "resuming-agy", os.getpid(), uuid=UA,
yaml_path=str(yaml_path), home=str(home), wait_s_val=0) == 1
finally:
live_proc.kill()
live_proc.wait()
def test_portability_python39_ast_and_bash_syntax():
"""Static guard: new/updated Python modules parse under Python 3.9 and shell scripts pass bash -n."""
import ast
for rel in (
"lib_py/workspace_uuid.py",
"lib_py/agents/session_ids.py",
"lib_py/agents/id_order.py",
):
src = (SKILLS / rel).read_text(encoding="utf-8")
ast.parse(src, filename=rel, feature_version=(3, 9))
for rel in (
"lib.sh",
"multi-agent-mux-stop/scripts/stop_session.sh",
"multi-agent-mux-resume/scripts/resolve_session_id.sh",
"multi-agent-mux-resume/scripts/resume_session.sh",
"multi-agent-mux-resume/scripts/update_yaml_resumed.sh",
"multi-agent-mux-create/scripts/create_session.sh",
):
res = subprocess.run(["bash", "-n", str(SKILLS / rel)], capture_output=True, text=True, timeout=30)
assert res.returncode == 0, f"{rel}: {res.stderr}"
def test_resolve_session_id_without_session_arg_returns_workspace_id(mam_sandbox):
"""resolve_session_id.sh and decide_resume with name='' return the workspace's stopped row id."""
from lib_py.agents.session_ids import decide_resume
ws = mam_sandbox / "opt-sess-ws"
ws.mkdir(parents=True, exist_ok=True)
epoch = int(time.time()) - 100
_agy_full_artifacts(mam_sandbox, UA)
_agy_cache(mam_sandbox, {ws: UA})
_claude_transcript(mam_sandbox / ".claude" / "projects", ws, UB)
rows = [
_row("opt-sess-ws-creator-agy", "agy", ws, epoch, status="stopped", agy_conversation_id_own=UA),
_row("opt-sess-ws-creator-claude", "claude", ws, epoch, status="stopped", claude_session_id_own=UB),
]
_write_rows(mam_sandbox, rows)
state = {"herdr_sessions": rows}
assert decide_resume(state, "", "agy", str(ws), home=str(mam_sandbox))["uuid"] == UA
assert decide_resume(
state, "", "claude", str(ws), home=str(mam_sandbox),
claude_dir=str(mam_sandbox / ".claude" / "projects"),
)["uuid"] == UB
script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-resume" / "scripts" / "resolve_session_id.sh"
res_agy = subprocess.run(
["bash", str(script), "--workspace", str(ws), "--agent", "agy"],
cwd=str(mam_sandbox), capture_output=True, text=True, timeout=30,
)
assert res_agy.returncode == 0
assert res_agy.stdout.strip() == UA
res_claude = subprocess.run(
["bash", str(script), "--workspace", str(ws), "--agent", "claude"],
cwd=str(mam_sandbox), capture_output=True, text=True, timeout=30,
)
assert res_claude.returncode == 0
assert res_claude.stdout.strip() == UB
def test_id_order_old_stopped_row_with_own_id_does_not_release_fresh_startup_marker(tmp_path, monkeypatch):
"""A pre-admission stopped row's own_id must not release a live startup marker
for a fresh resume (stale own_id) or a create reusing an old stopped session name."""
import threading
from lib_py.agents import id_order
from lib_py.agents.session_ids import decide_resume
home = tmp_path / "home"
ws = tmp_path / "ws"
ws.mkdir(parents=True)
yaml_path = tmp_path / "agent-sessions.yaml"
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("AGENT_SESSIONS_YAML", str(yaml_path))
for mode, old_own in (("stale_resume", UX), ("recreate_stopped", UA)):
if mode == "recreate_stopped":
_agy_conv(home, UA, time.time() - 50)
old_epoch = int(time.time()) - 100
old_stop = int(time.time()) - 10
row_a = _row(
"same-ws-creator-agy", "agy", ws, old_epoch, status="stopped",
agy_conversation_id_own=old_own, stopped_at_epoch=old_stop,
)
state = {"herdr_sessions": [row_a]}
yaml_path.write_text(yaml.safe_dump(state))
if mode == "stale_resume":
dec = decide_resume(state, "same-ws-creator-agy", "agy", str(ws), home=str(home))
assert dec["uuid"] == "" and dec["drop_stale_id"] is True
proc_a = subprocess.Popen(["sleep", "120"])
proc_b = subprocess.Popen(["sleep", "120"])
try:
# Admit A (fresh startup, uuid=''): writes live marker while old stopped row is in state
assert id_order.wait(
str(ws), "agy", "same-ws-creator-agy", proc_a.pid, uuid="",
yaml_path=str(yaml_path), home=str(home), wait_s_val=0,
) == 0
# Immediate check at cap 0: B must be blocked even before A updates its YAML row
assert id_order.wait(
str(ws), "agy", "same-ws-reviewer-agy", proc_b.pid, uuid="",
yaml_path=str(yaml_path), home=str(home), wait_s_val=0,
) == 1
# Concurrent check: B waits while A transitions old-stopped -> running-unpinned -> running-pinned
b_rc: list = []
def _run_b():
rc = id_order.wait(
str(ws), "agy", "same-ws-reviewer-agy", proc_b.pid, uuid="",
yaml_path=str(yaml_path), home=str(home), wait_s_val=10, poll=0.1,
)
b_rc.append(rc)
t = threading.Thread(target=_run_b)
t.start()
time.sleep(0.35)
assert b_rc == [], f"B unblocked prematurely while A was still old stopped row ({mode})"
# A registers its fresh running row without an own_id
running_unpinned = _row(
"same-ws-creator-agy", "agy", ws, old_epoch + 100,
pid=proc_a.pid, status="running", agy_conversation_id_own=None,
)
yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [running_unpinned]}))
time.sleep(0.35)
assert b_rc == [], f"B unblocked prematurely while A was running unpinned ({mode})"
# A records its new conversation id UB -> B unblocks
running_pinned = _row(
"same-ws-creator-agy", "agy", ws, old_epoch + 100,
pid=proc_a.pid, status="running", agy_conversation_id_own=UB,
)
yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [running_pinned]}))
t.join(timeout=5)
assert b_rc == [0], f"B did not unblock after A recorded its new id ({mode})"
finally:
proc_a.kill()
proc_b.kill()
proc_a.wait()
proc_b.wait()
def test_canonical_cwd_distinguishes_underscore_hyphen_and_nested_paths_with_symlink_control(tmp_path, monkeypatch):
"""Distinct directories whose lossy workspace_key collides (project_a vs project-a,
nested/sub vs nested-sub) do not block or merge rows; a symlink alias to the same
directory does match."""
from lib_py.agents import id_order
from lib_py.agents.session_ids import decide_resume, same_type_rows
home = tmp_path / "home"
ws_under = tmp_path / "project_a"
ws_hyphen = tmp_path / "project-a"
ws_nested = tmp_path / "nested" / "sub"
ws_flat = tmp_path / "nested-sub"
for p in (ws_under, ws_hyphen, ws_nested, ws_flat):
p.mkdir(parents=True)
ws_alias = tmp_path / "project_a_link"
ws_alias.symlink_to(ws_under)
yaml_path = tmp_path / "agent-sessions.yaml"
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("AGENT_SESSIONS_YAML", str(yaml_path))
_agy_conv(home, UA, time.time() - 10)
_agy_conv(home, UB, time.time() - 5)
_agy_cache(home, {ws_under: UA, ws_hyphen: UB})
proc = subprocess.Popen(["sleep", "120"])
try:
state_live = {
"herdr_sessions": [
_row("under-agy", "agy", ws_under, 100, pid=proc.pid, status="running"),
_row("nested-agy", "agy", ws_nested, 100, pid=proc.pid, status="running"),
]
}
yaml_path.write_text(yaml.safe_dump(state_live))
# Colliding workspace_key directories do NOT block each other
assert id_order.wait(
str(ws_hyphen), "agy", "hyphen-agy", os.getpid(),
yaml_path=str(yaml_path), home=str(home), wait_s_val=0,
) == 0
assert id_order.wait(
str(ws_flat), "agy", "flat-agy", os.getpid(),
yaml_path=str(yaml_path), home=str(home), wait_s_val=0,
) == 0
# Symlink alias to ws_under DOES see live unpinned under-agy and blocks
assert id_order.wait(
str(ws_alias), "agy", "alias-agy", os.getpid(),
yaml_path=str(yaml_path), home=str(home), wait_s_val=0,
) == 1
finally:
proc.kill()
proc.wait()
state_stopped = {
"herdr_sessions": [
_row("under-agy", "agy", ws_under, 100, status="stopped"),
_row("hyphen-agy", "agy", ws_hyphen, 100, status="stopped"),
_row("nested-agy", "agy", ws_nested, 100, status="stopped"),
_row("flat-agy", "agy", ws_flat, 100, status="stopped"),
]
}
assert [r["name"] for r in same_type_rows(state_stopped, str(ws_hyphen), "agy")] == ["hyphen-agy"]
assert [r["name"] for r in same_type_rows(state_stopped, str(ws_flat), "agy")] == ["flat-agy"]
assert [r["name"] for r in same_type_rows(state_stopped, str(ws_alias), "agy")] == ["under-agy"]
# Single stopped row in ws_hyphen uses single-row workspace discovery (UB) instead of 2-row fresh spawn
res_hyphen = decide_resume(state_stopped, "hyphen-agy", "agy", str(ws_hyphen), home=str(home))
assert res_hyphen["uuid"] == UB
assert res_hyphen["notes"] == []
def test_id_order_resolved_but_unrecorded_resume_uuid_reserves_startup_until_recorded(tmp_path, monkeypatch):
"""When a single stopped post-discovered row with no own_id resolves a uuid from disk,
id_order.wait reserves startup (writes a marker) until that uuid is recorded on the row."""
from lib_py.agents import id_order
from lib_py.agents.session_ids import decide_resume
home = tmp_path / "home"
ws = tmp_path / "ws"
ws.mkdir(parents=True)
yaml_path = tmp_path / "agent-sessions.yaml"
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("AGENT_SESSIONS_YAML", str(yaml_path))
_agy_conv(home, UA, time.time() - 10)
_agy_cache(home, {ws: UA})
old_epoch = int(time.time()) - 100
row_a = _row("same-ws-creator-agy", "agy", ws, old_epoch, status="stopped", agy_conversation_id_own=None)
state = {"herdr_sessions": [row_a]}
yaml_path.write_text(yaml.safe_dump(state))
dec = decide_resume(state, "same-ws-creator-agy", "agy", str(ws), home=str(home))
assert dec["uuid"] == UA
proc_a = subprocess.Popen(["sleep", "120"])
proc_b = subprocess.Popen(["sleep", "120"])
try:
# Admit A with resolved (but not yet recorded) uuid UA -> writes startup marker
assert id_order.wait(
str(ws), "agy", "same-ws-creator-agy", proc_a.pid, uuid=UA,
yaml_path=str(yaml_path), home=str(home), wait_s_val=0,
) == 0
# While A has not yet recorded UA on its row, B must be refused at cap 0
assert id_order.wait(
str(ws), "agy", "same-ws-reviewer-agy", proc_b.pid, uuid="",
yaml_path=str(yaml_path), home=str(home), wait_s_val=0,
) == 1
# Once A records UA on its row (via update_yaml_resumed.sh), B is released
recorded_a = _row(
"same-ws-creator-agy", "agy", ws, old_epoch + 50,
pid=proc_a.pid, status="running", agy_conversation_id_own=UA,
)
yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [recorded_a]}))
assert id_order.wait(
str(ws), "agy", "same-ws-reviewer-agy", proc_b.pid, uuid="",
yaml_path=str(yaml_path), home=str(home), wait_s_val=0,
) == 0
finally:
proc_a.kill()
proc_b.kill()
proc_a.wait()
proc_b.wait()
def test_id_order_dead_owner_marker_blocks_while_its_unregistered_herdr_session_is_alive(
tmp_path, monkeypatch, capsys
):
"""PR #69 review: a create/resume killed after spawning its pane but before
registering its row leaves a dead-owner marker. It keeps blocking while the
marker's herdr session is alive, goes once the row is registered (the row
decides) or the session is gone; on the same session name a live attempt is
not pruned and a dead one follows the orphan rules (PR #69 review r3); a
running row without a pane pid blocks (fail-closed)."""
from lib_py.agents import id_order
home = tmp_path / "home"
ws = tmp_path / "ws"
ws.mkdir(parents=True)
yaml_path = tmp_path / "agent-sessions.yaml"
yaml_path.write_text(yaml.safe_dump({"herdr_sessions": []}))
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("AGENT_SESSIONS_YAML", str(yaml_path))
dpath = id_order.order_dir(str(ws), "agy", home=str(home))
marker_a = Path(dpath) / "orphan-a-agy.json"
alive_sessions = {"orphan-a-agy"}
unknown_sessions = set()
probes = []
def pane_state(server, session, timeout):
probes.append((server, session))
if session in unknown_sessions:
return None
return session in alive_sessions
def wait_b(**kw):
return id_order.wait(str(ws), "agy", "new-b-agy", os.getpid(), yaml_path=str(yaml_path),
home=str(home), wait_s_val=0, pane_state=pane_state, **kw)
dead = subprocess.Popen(["true"])
dead.wait()
live = subprocess.Popen(["sleep", "120"])
try:
# Orphan: owner dead, no row, herdr session alive -> blocks, marker kept, kill hint printed
id_order.write_marker(dpath, "orphan-a-agy", dead.pid, str(ws), "agy", herdr_server="srv-x")
capsys.readouterr()
assert wait_b() == 1
assert marker_a.exists()
assert ("srv-x", "orphan-a-agy") in probes
err = capsys.readouterr().err
assert "HERDR_SESSION_NAME=srv-x herdr kill-session -t orphan-a-agy" in err
# No authoritative answer from herdr -> kept and blocks as well
unknown_sessions.add("orphan-a-agy")
assert wait_b() == 1
assert marker_a.exists()
assert "herdr could not tell whether its session is gone" in capsys.readouterr().err
unknown_sessions.clear()
# Row registered for the attempt -> marker pruned without a probe; the row decides
probes.clear()
epoch = int(time.time())
yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [
_row("orphan-a-agy", "agy", ws, epoch, pid=live.pid)]}))
assert wait_b() == 1
assert not marker_a.exists() and probes == []
yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [
_row("orphan-a-agy", "agy", ws, epoch, pid=live.pid, agy_conversation_id_own=UA)]}))
assert wait_b() == 0
# Session gone -> pruned, B admitted
yaml_path.write_text(yaml.safe_dump({"herdr_sessions": []}))
id_order.write_marker(dpath, "orphan-a-agy", dead.pid, str(ws), "agy", herdr_server="srv-x")
alive_sessions.clear()
assert wait_b() == 0
assert not marker_a.exists()
# Same session name: a live attempt is kept and blocks; a dead one is kept and blocks
# while its herdr session is alive or herdr cannot tell, and is taken over once gone
same = Path(dpath) / "same-agy.json"
def wait_same():
return id_order.wait(str(ws), "agy", "same-agy", os.getpid(), yaml_path=str(yaml_path),
home=str(home), wait_s_val=0, pane_state=pane_state)
id_order.write_marker(dpath, "same-agy", live.pid, str(ws), "agy")
assert wait_same() == 1
assert json.loads(same.read_text())["owner_pid"] == live.pid
live.kill()
live.wait()
alive_sessions.add("same-agy")
probes.clear()
capsys.readouterr()
assert wait_same() == 1
assert probes == [("", "same-agy")]
assert json.loads(same.read_text())["owner_pid"] == live.pid
err = capsys.readouterr().err
assert "its herdr session is alive" in err and "kill-session -t same-agy" in err
unknown_sessions.add("same-agy")
assert wait_same() == 1
assert json.loads(same.read_text())["owner_pid"] == live.pid
unknown_sessions.clear()
alive_sessions.clear()
assert wait_same() == 0
assert json.loads(same.read_text())["owner_pid"] == os.getpid()
same.unlink()
# Non-blocking review note: a running same-cwd row without a pane pid blocks
no_pid = _row("no-pid-agy", "agy", ws, epoch)
no_pid["pane"].pop("pid")
yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [no_pid]}))
assert wait_b() == 1
finally:
live.kill()
live.wait()
def test_id_order_create_waits_for_orphan_herdr_session_of_a_killed_create(
mam_sandbox, mock_herdr, mock_agents
):
"""PR #69 review reproduction: A's create spawned its pane and was killed
before appending its row (dead-owner marker, live herdr agent, no row).
create B refuses while A's herdr session is alive and proceeds once it is gone."""
from lib_py.agents import id_order
ws = mam_sandbox / "orphan-ws"
ws.mkdir(parents=True, exist_ok=True)
name_a, name_b = "orphan-ws-creator-agy", "orphan-ws-reviewer-agy"
create_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
dead = subprocess.Popen(["true"])
dead.wait()
holder = subprocess.Popen(["sleep", "120"])
try:
dpath = id_order.order_dir(str(ws), "agy", home=str(mam_sandbox))
id_order.write_marker(dpath, name_a, dead.pid, str(ws), "agy", herdr_server="orphan-ws")
_mock_herdr_agent(mock_herdr, name_a, ws, holder.pid, int(time.time()), pane_id="w1:p_a")
def create_b():
return subprocess.run(
["bash", str(create_sh), "--workspace", str(ws), "--agent", "agy",
"--role", "reviewer", "--session", name_b, "--no-onboard"],
capture_output=True, text=True, cwd=str(mam_sandbox),
env=dict(os.environ, MAM_ID_ORDER_WAIT_S="3"), timeout=120,
)
res = create_b()
assert res.returncode == 1, res.stdout + res.stderr
assert name_a in res.stderr and "kill-session -t " + name_a in res.stderr
assert "its herdr session is alive" in res.stderr
assert name_b not in _read_rows(mam_sandbox)
assert name_b not in json.loads(Path(mock_herdr).read_text()).get("agents", {})
state = json.loads(Path(mock_herdr).read_text())
state["agents"].pop(name_a)
Path(mock_herdr).write_text(json.dumps(state))
res = create_b()
assert res.returncode == 0, res.stdout + res.stderr
assert _read_rows(mam_sandbox)[name_b]["status"] == "running"
finally:
holder.kill()
holder.wait()
def test_id_order_herdr_probes_stay_within_the_wait_cap(tmp_path, monkeypatch):
"""PR #69 review r2 P2: probes of dead-owner markers get only the time left
until the wait deadline, shared by all markers, so a slow herdr cannot hold
the wait (and its lock) past the cap; cap 0 asks nothing (unknown, kept)."""
from lib_py.agents import id_order
home = tmp_path / "home"
ws = tmp_path / "ws"
ws.mkdir(parents=True)
yaml_path = tmp_path / "agent-sessions.yaml"
yaml_path.write_text(yaml.safe_dump({"herdr_sessions": []}))
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("AGENT_SESSIONS_YAML", str(yaml_path))
dpath = id_order.order_dir(str(ws), "agy", home=str(home))
dead = subprocess.Popen(["true"])
dead.wait()
for name in ("orphan-1-agy", "orphan-2-agy"):
id_order.write_marker(dpath, name, dead.pid, str(ws), "agy", herdr_server="srv")
budgets = []
def slow_probe(server, session, timeout):
budgets.append(timeout)
if timeout <= 0:
return None
time.sleep(min(timeout, 5.0))
return None
for cap in (0, 1, 3):
budgets.clear()
t0 = time.monotonic()
rc = id_order.wait(str(ws), "agy", "new-agy", os.getpid(), yaml_path=str(yaml_path),
home=str(home), wait_s_val=cap, poll=0.05, pane_state=slow_probe)
elapsed = time.monotonic() - t0
bound = cap
assert rc == 1, cap
assert elapsed < bound + 0.6, (cap, elapsed)
assert budgets and max(budgets) <= bound + 0.01, (cap, budgets)
if cap == 0:
assert elapsed < 0.3 and all(b <= 0 for b in budgets), (elapsed, budgets)
assert len(list(Path(dpath).glob("orphan-*.json"))) == 2, cap
def test_id_order_real_probe_against_mock_herdr_fails_closed(mam_sandbox, mock_herdr):
"""PR #69 review r2 P1: lib.sh mam_id_order_wait with the real probe (sandbox
copy, mock herdr). A listed orphan blocks; an unlisted one is pruned; a failed
`agent list` (server_not_running included) keeps the marker and refuses;
with MAM_ID_ORDER_WAIT_S=0 herdr is not asked and the marker is kept. A
retry on the orphan's own session name is held the same way (review r3)."""
from lib_py.agents import id_order
ws = mam_sandbox / "probe-ws"
ws.mkdir(parents=True, exist_ok=True)
name_a, name_b = "probe-ws-creator-agy", "probe-ws-reviewer-agy"
lib = mam_sandbox / ".agents" / "skills" / "lib.sh"
dpath = id_order.order_dir(str(ws), "agy", home=str(mam_sandbox))
marker_a = Path(dpath) / f"{name_a}.json"
dead = subprocess.Popen(["true"])
dead.wait()
holder = subprocess.Popen(["sleep", "120"])
def set_mock(**kw):
state = json.loads(Path(mock_herdr).read_text())
state.pop("force_error_code", None)
state["agents"].pop(name_a, None)
state.update(kw)
Path(mock_herdr).write_text(json.dumps(state))
def wait_b(cap="2", who=name_b):
for f in Path(dpath).glob(f"{name_b}.json"):
f.unlink()
id_order.write_marker(dpath, name_a, dead.pid, str(ws), "agy", herdr_server="probe-ws")
return subprocess.run(
["bash", "-c", 'source "$1" && mam_id_order_wait "$2" agy "$3"', "_", str(lib), str(ws), who],
capture_output=True, text=True, cwd=str(mam_sandbox), timeout=120,
env=dict(os.environ, MAM_ID_ORDER_WAIT_S=cap, HERDR_SESSION_NAME="probe-ws"),
)
try:
_mock_herdr_agent(mock_herdr, name_a, ws, holder.pid, int(time.time()), pane_id="w1:p_a")
res = wait_b()
assert res.returncode == 1 and "its herdr session is alive" in res.stderr, res.stderr
assert marker_a.exists()
res = wait_b(who=name_a)
assert res.returncode == 1 and "its herdr session is alive" in res.stderr, res.stderr
assert json.loads(marker_a.read_text())["owner_pid"] == dead.pid
set_mock(force_error_code="protocol_mismatch")
res = wait_b()
assert res.returncode == 1, res.stderr
assert "herdr could not tell whether its session is gone" in res.stderr
assert marker_a.exists()
set_mock(force_error_code="server_not_running")
res = wait_b()
assert res.returncode == 1, res.stderr
assert "herdr could not tell whether its session is gone" in res.stderr
assert marker_a.exists()
set_mock()
res = wait_b(cap="0")
assert res.returncode == 1, res.stderr
assert "MAM_ID_ORDER_WAIT_S=0 leaves no time to ask herdr" in res.stderr
assert marker_a.exists()
res = wait_b()
assert res.returncode == 0, res.stderr
assert not marker_a.exists()
res = wait_b(who=name_a)
assert res.returncode == 0, res.stderr
assert json.loads(marker_a.read_text())["owner_pid"] != dead.pid
finally:
holder.kill()
holder.wait()