"""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") # --------------------------------------------------------------------------- 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