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


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