GodopuandClaude Opus 5.5
7cac6f613e
PR #69 review round 3 (rebuttal overruled; codex r2 applied as written). - A failed `agent list`, herdr's server_not_running included, is unknown and keeps an orphaned marker; only a successful list without the name prunes. - No probe floor: probes get only the time left until the wait deadline, so MAM_ID_ORDER_WAIT_S=0 asks herdr nothing (unknown, kept) and the refusal says so. - The timeout message reports the last definite answer of this wait, so an alive orphan whose final probe ran out of time still reads as alive. - Tests follow: server_not_running and cap 0 keep the marker and refuse; delayed-probe bound is the cap itself; probing tests use a small cap. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1296 lines
57 KiB
Python
1296 lines
57 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; a live attempt on the same session name is
|
|
not pruned; 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()
|
|
|
|
# A live attempt on the same session name is kept and blocks; a dead one is taken over
|
|
same = Path(dpath) / "same-agy.json"
|
|
id_order.write_marker(dpath, "same-agy", live.pid, str(ws), "agy")
|
|
assert 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) == 1
|
|
assert json.loads(same.read_text())["owner_pid"] == live.pid
|
|
live.kill()
|
|
live.wait()
|
|
assert 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) == 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."""
|
|
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"):
|
|
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), name_b],
|
|
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()
|
|
|
|
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()
|
|
finally:
|
|
holder.kill()
|
|
holder.wait()
|