"""Issue #40: a post-discovered session id (agy, opencode, codex, hermes) is
pinned once a local job's `completed` is verified, by the receiving side
(lib_py.agents.autopin), and agy never records another workspace's - or
another session's - conversation.

Completion conditions:
  1) an unpinned agy row never records the id of a conversation that only
     another workspace updated: test_agy_verify_rejects_id_cached_for_another_workspace,
     test_reconcile_does_not_pin_another_workspaces_agy_conversation,
     test_autopin_does_not_pin_another_workspaces_agy_conversation
  2) the first completed job pins the session's own id, no daemon:
     test_completed_onboarding_pins_agy_without_daemon (+ reconcile, lib.sh and
     create --onboard variants), test_completed_first_job_pins_opencode_codex_hermes
  3) one pin per job/token/attempt; ambiguity pins nothing and leaves a reason:
     test_wrapper_and_loop_recovery_pin_once, test_ambiguous_candidates_leave_reason_without_pin
  4) pre-assigned rows and rows with an id are untouched:
     test_preassigned_and_existing_ids_are_untouched

Everything runs against temporary homes, registries and a mock herdr.
"""
import json
import os
import sqlite3
import subprocess
import time
from pathlib import Path

import pytest
import yaml

REPO = Path(__file__).resolve().parents[1]
SKILLS = REPO / ".agents" / "skills"

UA = "a4000001-0000-4000-8000-000000000001"
UB = "b4000002-0000-4000-8000-000000000002"
UC = "c4000003-0000-4000-8000-000000000003"


# --------------------------------------------------------------------------- helpers

def _agy_conv(home, uuid, mtime=None):
    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_cache(home, mapping):
    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 _row(name, agent, cwd, epoch, pid=4242, **extra):
    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


def _control_plane(root, rows):
    mam = Path(root) / ".mam"
    (mam / "jobs").mkdir(parents=True, exist_ok=True)
    yaml_path = mam / "agent-sessions.yaml"
    yaml_path.write_text(yaml.safe_dump({"herdr_sessions": rows}, sort_keys=False))
    return yaml_path, mam / "jobs"


def _job(jobs_dir, job_id, session, status="completed", token="tok-1", attempt=1,
         transport="herdr", agent="agy"):
    rec = {"job_id": job_id, "status": status, "agent": agent, "agent_session": session,
           "dispatch_token": token, "attempt_id": attempt, "transport": transport}
    (Path(jobs_dir) / f"{job_id}.json").write_text(json.dumps(rec))


def _rows(yaml_path):
    return {r["name"]: r for r in yaml.safe_load(Path(yaml_path).read_text())["herdr_sessions"]}


def _handle(jobs_dir, job_id, yaml_path, home, **kw):
    from lib_py.agents import autopin
    kw.setdefault("viewport", lambda name, agent, cwd: 0)
    kw.setdefault("resume", lambda name, agent, cwd: "resume_verified")
    return autopin.handle_completed(str(jobs_dir), job_id, yaml_path=str(yaml_path), home_dir=str(home), **kw)


@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


@pytest.fixture
def fake_opencode(tmp_path, monkeypatch):
    """No real opencode is ever asked for its paths: the adapter falls back
    to <home>/.local/share/opencode/opencode.db."""
    bindir = tmp_path / "fakebin"
    bindir.mkdir(exist_ok=True)
    exe = bindir / "opencode"
    exe.write_text("#!/bin/sh\nexit 1\n")
    exe.chmod(0o755)
    monkeypatch.setenv("PATH", f"{bindir}:{os.environ.get('PATH', '')}")
    monkeypatch.delenv("XDG_DATA_HOME", raising=False)
    return exe


def _seed_class_a(agent, home, ws, ws_other, sid, other_sid, started):
    """One session of `agent` in ws (sid) and one in another workspace (other_sid), both new."""
    if agent == "opencode":
        db = Path(home) / ".local" / "share" / "opencode" / "opencode.db"
        db.parent.mkdir(parents=True, exist_ok=True)
        conn = sqlite3.connect(str(db))
        conn.execute("CREATE TABLE IF NOT EXISTS session_v2 (id TEXT PRIMARY KEY, parent_id TEXT, directory TEXT, time_created INTEGER)")
        for u, w, t in ((sid, ws, started), (other_sid, ws_other, started + 5)):
            conn.execute("INSERT OR REPLACE INTO session_v2 VALUES (?, NULL, ?, ?)", (u, str(w), int(t * 1000)))
        conn.commit()
        conn.close()
    elif agent == "hermes":
        db = Path(home) / ".hermes" / "state.db"
        db.parent.mkdir(parents=True, exist_ok=True)
        conn = sqlite3.connect(str(db))
        conn.execute("CREATE TABLE IF NOT EXISTS sessions (id TEXT PRIMARY KEY, cwd TEXT, started_at REAL)")
        for u, w, t in ((sid, ws, started), (other_sid, ws_other, started + 5)):
            conn.execute("INSERT OR REPLACE INTO sessions VALUES (?, ?, ?)", (u, str(w), t))
        conn.commit()
        conn.close()
    elif agent == "codex":
        codex = Path(home) / ".codex"
        sess = codex / "sessions" / "2026" / "10" / "20"
        sess.mkdir(parents=True, exist_ok=True)
        conn = sqlite3.connect(str(codex / "state_5.sqlite"))
        conn.execute("CREATE TABLE IF NOT EXISTS threads (id TEXT PRIMARY KEY, cwd TEXT, rollout_path TEXT, created_at INTEGER, updated_at INTEGER)")
        for u, w, t in ((sid, ws, started), (other_sid, ws_other, started + 5)):
            rollout = sess / f"rollout-2026-10-20T00-00-00-{u}.jsonl"
            rollout.write_text(json.dumps({"payload": {"id": u, "cwd": str(w)}}) + "\n")
            conn.execute("INSERT OR REPLACE INTO threads VALUES (?, ?, ?, ?, ?)", (u, str(w), str(rollout), int(t), int(t)))
        conn.commit()
        conn.close()


# --------------------------------------------------------------------------- adapter contract

def test_adapter_id_lifecycle_classification():
    from lib_py.agents.base import POST_DISCOVERED, PRE_ASSIGNED
    from lib_py.agents.registry import get_adapter
    assert {a: get_adapter(a).id_lifecycle for a in ("agy", "opencode", "codex", "hermes", "claude", "grok")} == {
        "agy": POST_DISCOVERED, "opencode": POST_DISCOVERED, "codex": POST_DISCOVERED,
        "hermes": POST_DISCOVERED, "claude": PRE_ASSIGNED, "grok": PRE_ASSIGNED}


# --------------------------------------------------------------------------- 1) another workspace's agy id

def test_agy_verify_rejects_id_cached_for_another_workspace(tmp_path):
    from lib_py.verify_session import verify_session_uuid
    home, ws_a, ws_b = tmp_path / "home", tmp_path / "ws_a", tmp_path / "ws_b"
    ws_a.mkdir()
    ws_b.mkdir()
    epoch = int(time.time()) - 100
    _agy_conv(home, UB)
    _agy_cache(home, {ws_b: UB})
    row_a = _row("ws-a-creator-agy", "agy", ws_a, epoch)
    assert verify_session_uuid(str(ws_a), "agy", UB, row=row_a, home_dir=str(home), mode="discover") is False
    # the same id is fine for the workspace the cache names
    row_b = _row("ws-b-creator-agy", "agy", ws_b, epoch)
    assert verify_session_uuid(str(ws_b), "agy", UB, row=row_b, home_dir=str(home), mode="discover") is True


def _mock_agent(state_file, name, cwd, pid, epoch):
    state = json.loads(Path(state_file).read_text())
    state.setdefault("agents", {})[name] = {"pid": pid, "cwd": str(cwd), "command": "agy", "agent": "agy",
                                            "created": epoch, "buffer": f"Antigravity  {cwd}\n>"}
    Path(state_file).write_text(json.dumps(state))


def _reconcile(sandbox):
    script = sandbox / ".agents" / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh"
    res = subprocess.run(["bash", str(script), "--once", "--emit-diff"], capture_output=True, text=True,
                         cwd=str(sandbox), env={**os.environ, "WORKSPACE_ROOT": str(sandbox)}, timeout=120)
    assert res.returncode == 0, res.stderr
    return res


def test_reconcile_does_not_pin_another_workspaces_agy_conversation(mam_sandbox, mock_herdr):
    """The issue's example: proj-A's agy session is unpinned, only proj-B's
    conversation changed after A started. reconcile used to glob every *.db
    and record B's id on A."""
    ws_a, ws_b = mam_sandbox / "proj-a", mam_sandbox / "proj-b"
    ws_a.mkdir()
    ws_b.mkdir()
    epoch = int(time.time()) - 100
    _agy_conv(mam_sandbox, UB)
    _agy_cache(mam_sandbox, {ws_b: UB})
    name = "proj-a-creator-agy"
    yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
    yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [_row(name, "agy", ws_a, epoch, pid=os.getpid())]}))
    _mock_agent(mock_herdr, name, ws_a, os.getpid(), epoch)

    _reconcile(mam_sandbox)
    row = _rows(yaml_path)[name]
    assert row["status"] == "running"
    assert row["agy_conversation_id_own"] is None


def test_autopin_does_not_pin_another_workspaces_agy_conversation(tmp_path, owners):
    home, ws_a, ws_b = tmp_path / "home", tmp_path / "ws_a", tmp_path / "ws_b"
    ws_a.mkdir()
    ws_b.mkdir()
    epoch = int(time.time()) - 100
    _agy_conv(home, UB)
    _agy_cache(home, {ws_b: UB})
    held, _ = owners
    held[UB] = [4242]  # even a holder that looks like A's pane does not make B's id A's
    yaml_path, jobs = _control_plane(tmp_path / "ctl", [_row("a-agy", "agy", ws_a, epoch, pid=4242)])
    _job(jobs, "j1", "herdr:a-agy")
    res = _handle(jobs, "j1", yaml_path, home)
    assert res["status"] == "no-candidate"
    row = _rows(yaml_path)["a-agy"]
    assert row["agy_conversation_id_own"] is None
    assert "autopin no-candidate (job j1)" in row["last_visible_status"]


# --------------------------------------------------------------------------- 2) first completed job pins

def test_completed_onboarding_pins_agy_without_daemon(tmp_path, owners, monkeypatch):
    """A's onboarding completed; B's conversation was updated meanwhile (and
    later than A's). A gets A's id, owned by A's pane."""
    monkeypatch.setenv("MAM_NO_SUBSCRIBER_DAEMON", "1")
    home, ws_a, ws_b = tmp_path / "home", tmp_path / "ws_a", tmp_path / "ws_b"
    ws_a.mkdir()
    ws_b.mkdir()
    now = time.time()
    epoch = int(now) - 100
    _agy_conv(home, UA, mtime=now - 10)
    _agy_conv(home, UB, mtime=now)
    _agy_cache(home, {ws_a: UA, ws_b: UB})
    held, _ = owners
    held[UA], held[UB] = [5001], [6001]
    yaml_path, jobs = _control_plane(tmp_path / "ctl", [_row("a-agy", "agy", ws_a, epoch, pid=5001)])
    _job(jobs, "onb1", "herdr:a-agy")

    res = _handle(jobs, "onb1", yaml_path, home)
    assert res["status"] == "pinned", res
    row = _rows(yaml_path)["a-agy"]
    assert row["agy_conversation_id_own"] == UA
    assert row["session_id_source"] == "discovered"
    assert row["session_id_verified"] is True
    assert row["last_visible_status"] == "resume_verified"


@pytest.mark.parametrize("agent,sid,other", [("opencode", "ses_own_a", "ses_other_b"),
                                            ("codex", UA, UB), ("hermes", "h-own-a", "h-other-b")])
def test_completed_first_job_pins_opencode_codex_hermes(tmp_path, monkeypatch, fake_opencode, agent, sid, other):
    home, ws, ws_other = tmp_path / "home", tmp_path / "ws", tmp_path / "ws_other"
    ws.mkdir()
    ws_other.mkdir()
    monkeypatch.setenv("HOME", str(home))
    epoch = int(time.time()) - 100
    _seed_class_a(agent, home, ws, ws_other, sid, other, started=epoch + 10)
    yaml_path, jobs = _control_plane(tmp_path / "ctl", [_row(f"w-creator-{agent}", agent, ws, epoch)])
    _job(jobs, "first", f"herdr:w-creator-{agent}", agent=agent)

    res = _handle(jobs, "first", yaml_path, home)
    assert res["status"] == "pinned", res
    row = _rows(yaml_path)[f"w-creator-{agent}"]
    from lib_py.agents.registry import own_key
    assert row[own_key(agent)] == sid
    assert (row["session_id_source"], row["session_id_verified"]) == ("discovered", True)


def _fake_lsof(bindir, uuid, pid):
    exe = Path(bindir) / "lsof"
    exe.write_text(f'#!/bin/sh\nfor a in "$@"; do case "$a" in *{uuid}*) echo {pid}; exit 0;; esac; done\nexit 1\n')
    exe.chmod(0o755)


def test_reconcile_pins_agy_owned_by_the_pane(mam_sandbox, mock_herdr):
    """drift C uses the same capture: A's cached id, held by A's pane, is pinned;
    B's newer conversation in another workspace is not considered."""
    ws_a, ws_b = mam_sandbox / "proj-a", mam_sandbox / "proj-b"
    ws_a.mkdir()
    ws_b.mkdir()
    now = time.time()
    epoch = int(now) - 100
    _agy_conv(mam_sandbox, UA, mtime=now - 10)
    _agy_conv(mam_sandbox, UB, mtime=now)
    _agy_cache(mam_sandbox, {ws_a: UA, ws_b: UB})
    _fake_lsof(mam_sandbox / "bin", UA, os.getpid())
    name = "proj-a-creator-agy"
    yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
    yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [_row(name, "agy", ws_a, epoch, pid=os.getpid())]}))
    _mock_agent(mock_herdr, name, ws_a, os.getpid(), epoch)

    _reconcile(mam_sandbox)
    row = _rows(yaml_path)[name]
    assert row["agy_conversation_id_own"] == UA
    assert (row["session_id_source"], row["session_id_verified"]) == ("discovered", True)


def test_lib_sh_helper_pins_through_the_cli(mam_sandbox, mock_herdr, fake_opencode):
    """The wrapper's and create's entry point: lib.sh delegate_autopin_completed
    runs `python -m lib_py.agents.autopin` against the sandbox control plane."""
    ws, ws_other = mam_sandbox / "proj", mam_sandbox / "other"
    ws.mkdir()
    ws_other.mkdir()
    epoch = int(time.time()) - 100
    _seed_class_a("opencode", mam_sandbox, ws, ws_other, "ses_own", "ses_other", started=epoch + 10)
    yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
    name = "proj-creator-opencode"
    yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [_row(name, "opencode", ws, epoch)]}))
    jobs = mam_sandbox / ".mam" / "jobs"
    jobs.mkdir(parents=True, exist_ok=True)
    _job(jobs, "jcli", f"herdr:{name}", agent="opencode")

    lib_sh = mam_sandbox / ".agents" / "skills" / "lib.sh"
    res = subprocess.run(["bash", "-c", f'source "{lib_sh}" && delegate_autopin_completed jcli tok-1 "{jobs}"'],
                         capture_output=True, text=True, env=dict(os.environ), timeout=120)
    assert res.returncode == 0, res.stderr
    row = _rows(yaml_path)[name]
    assert row["opencode_session_id_own"] == "ses_own", res.stderr
    marker = json.loads((jobs / "jcli" / "autopin.json").read_text())
    assert marker["key"] == "jcli:tok-1:1" and marker["result"]["status"] == "pinned"
    # a stale dispatch token is not this attempt's completion
    _job(jobs, "jstale", f"herdr:{name}", agent="opencode", token="tok-new")
    subprocess.run(["bash", "-c", f'source "{lib_sh}" && delegate_autopin_completed jstale tok-old "{jobs}"'],
                   capture_output=True, text=True, env=dict(os.environ), timeout=120)
    assert not (jobs / "jstale" / "autopin.json").exists()


def test_create_onboard_pins_without_daemon(mam_sandbox, mock_herdr, mock_agents):
    """create --onboard with MAM_NO_SUBSCRIBER_DAEMON=1 (mam_sandbox): the
    mock agent completes the onboarding and create runs the handler itself."""
    create = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
    res = subprocess.run(["bash", str(create), "--workspace", str(mam_sandbox), "--agent", "opencode",
                          "--role", "creator", "--onboard"],
                         capture_output=True, text=True, cwd=str(mam_sandbox), timeout=300)
    assert res.returncode == 0, res.stderr
    yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
    (row,) = _rows(yaml_path).values()
    job = row["delegate_job_id"]
    marker = json.loads((mam_sandbox / ".mam" / "jobs" / job / "autopin.json").read_text())
    # create's async reconcile may get there first; either way the handler ran
    assert marker["result"]["status"] in ("pinned", "already-set"), marker
    assert row["opencode_session_id_own"]
    assert (row["session_id_source"], row["session_id_verified"]) == ("discovered", True)


# --------------------------------------------------------------------------- agy: the cwd cache proves nothing

def test_agy_same_cwd_sibling_never_cross_pins(tmp_path, owners):
    """Two agy sessions in one cwd; the cache names U, whose presence lock A's
    pane holds. Only A may get U - before and after A is pinned."""
    from lib_py.verify_session import verify_session_uuid
    home, ws = tmp_path / "home", tmp_path / "ws"
    ws.mkdir()
    epoch = int(time.time()) - 100
    _agy_conv(home, UA)
    _agy_cache(home, {ws: UA})
    held, parents = owners
    held[UA] = [5001]
    parents.update({5001: 1, 6001: 1})
    yaml_path, jobs = _control_plane(tmp_path / "ctl", [_row("a-agy", "agy", ws, epoch, pid=5001),
                                                        _row("b-agy", "agy", ws, epoch, pid=6001)])
    _job(jobs, "jb1", "herdr:b-agy")
    assert _handle(jobs, "jb1", yaml_path, home)["status"] == "owner-mismatch"
    _job(jobs, "ja1", "herdr:a-agy")
    assert _handle(jobs, "ja1", yaml_path, home)["status"] == "pinned"
    _job(jobs, "jb2", "herdr:b-agy")
    assert _handle(jobs, "jb2", yaml_path, home)["status"] == "sibling-claimed"
    rows = _rows(yaml_path)
    assert rows["a-agy"]["agy_conversation_id_own"] == UA
    assert rows["b-agy"]["agy_conversation_id_own"] is None
    # a cache hit no longer bypasses the sibling exclusion
    row_b = dict(rows["b-agy"], _sibling_claimed_uuids=[UA])
    assert verify_session_uuid(str(ws), "agy", UA, row=row_b, home_dir=str(home), mode="discover") is False


@pytest.mark.parametrize("holders,parents,expected", [
    (None, {}, "unverifiable-owner"),               # lsof missing or failed
    ([], {}, "unverifiable-owner"),                 # nobody holds the lock
    ([7001], {7001: 1, 5001: 1}, "owner-mismatch"),  # held outside the pane's tree
    ([7001], {7001: 7000, 7000: 5001, 5001: 1}, "captured"),  # a descendant of the pane
])
def test_agy_capture_requires_presence_owned_by_pane(tmp_path, owners, holders, parents, expected):
    from lib_py.agents.autopin import capture
    home, ws = tmp_path / "home", tmp_path / "ws"
    ws.mkdir()
    epoch = int(time.time()) - 100
    _agy_conv(home, UA)
    _agy_cache(home, {ws: UA})
    held, ppid = owners
    if holders is None:
        held["__lsof__"] = None
    else:
        held[UA] = holders
    ppid.update(parents)
    row = _row("a-agy", "agy", ws, epoch, pid=5001)
    res = capture(row, "agy", {"herdr_sessions": [row]}, home_dir=str(home))
    assert res["status"] == expected, res
    assert (res["session_id"] == UA) == (expected == "captured")


def test_agy_holder_pids_reads_lsof(tmp_path, monkeypatch):
    """The real helper: lsof's pids, [] when it finds no holder, None without lsof."""
    from lib_py.agents.adapters import agy
    f = tmp_path / "x.lock"
    f.write_text("")
    bindir = tmp_path / "bin"
    bindir.mkdir()
    monkeypatch.setenv("PATH", str(bindir))
    assert agy._holder_pids([str(f)]) is None
    _fake_lsof(bindir, "x.lock", 4321)
    assert agy._holder_pids([str(f)]) == [4321]
    assert agy._holder_pids([str(tmp_path / "missing.lock")]) == []
    (bindir / "lsof").write_text("#!/bin/sh\nexit 1\n")
    assert agy._holder_pids([str(f)]) == []


# --------------------------------------------------------------------------- 3) once per job, ambiguity

def _count_captures(monkeypatch):
    from lib_py.agents import autopin
    calls = []
    real = autopin.capture

    def counting(*a, **kw):
        calls.append(a[0].get("name"))
        return real(*a, **kw)

    monkeypatch.setattr(autopin, "capture", counting)
    return calls


def _loop_runner(mam_dir, workspace_root=REPO):
    from lib_py.loop.runner import LoopRunner
    r = LoopRunner.__new__(LoopRunner)
    r.workspace_root = Path(workspace_root)
    r.mam_dir, r.jobs_dir = Path(mam_dir), Path(mam_dir) / "jobs"
    r.failed_submit_jobs, r._stall_seen = set(), {}
    r.verbose, r.interrupted = False, False
    return r


def test_wrapper_and_loop_recovery_pin_once(tmp_path, monkeypatch, fake_opencode):
    home, ws, ws_other = tmp_path / "home", tmp_path / "ws", tmp_path / "ws_other"
    ws.mkdir()
    ws_other.mkdir()
    epoch = int(time.time()) - 100
    _seed_class_a("opencode", home, ws, ws_other, "ses_own", "ses_other", started=epoch + 10)
    yaml_path, jobs = _control_plane(tmp_path / "ctl", [_row("w-opencode", "opencode", ws, epoch)])
    monkeypatch.setenv("AGENT_SESSIONS_YAML", str(yaml_path))
    monkeypatch.setenv("HOME_DIR", str(home))
    _job(jobs, "dup", "herdr:w-opencode", agent="opencode")
    calls = _count_captures(monkeypatch)

    first = _handle(jobs, "dup", yaml_path, home)  # the wrapper's subscriber terminal
    assert first["status"] == "pinned"
    written = yaml_path.stat().st_mtime_ns
    marker = (jobs / "dup" / "autopin.json").read_text()
    # loop recovery (a subprocess of the same handler): deduplicated, nothing rewritten
    assert _loop_runner(tmp_path / "ctl" / ".mam").wait_for_job("dup", max_wait=5) is True
    assert (jobs / "dup" / "autopin.json").read_text() == marker
    again = _handle(jobs, "dup", yaml_path, home)
    assert again.get("deduplicated") is True and again["status"] == "pinned"
    assert calls == ["w-opencode"]
    assert yaml_path.stat().st_mtime_ns == written
    assert _rows(yaml_path)["w-opencode"]["opencode_session_id_own"] == "ses_own"


def test_loop_recovery_pins_when_the_wrapper_did_not(mam_sandbox, mock_herdr, fake_opencode):
    """wait_for_job's recovery call alone pins the id (the wrapper's call was missed)."""
    ws, ws_other = mam_sandbox / "proj", mam_sandbox / "other"
    ws.mkdir()
    ws_other.mkdir()
    epoch = int(time.time()) - 100
    _seed_class_a("opencode", mam_sandbox, ws, ws_other, "ses_own", "ses_other", started=epoch + 10)
    yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
    name = "proj-creator-opencode"
    yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [_row(name, "opencode", ws, epoch)]}))
    jobs = mam_sandbox / ".mam" / "jobs"
    jobs.mkdir(parents=True, exist_ok=True)
    _job(jobs, "jloop", f"herdr:{name}", agent="opencode")

    assert _loop_runner(mam_sandbox / ".mam", workspace_root=mam_sandbox).wait_for_job("jloop", max_wait=5) is True
    row = _rows(yaml_path)[name]
    assert row["opencode_session_id_own"] == "ses_own"
    assert json.loads((jobs / "jloop" / "autopin.json").read_text())["result"]["status"] == "pinned"


def test_ambiguous_candidates_leave_reason_without_pin(tmp_path, monkeypatch, fake_opencode):
    home, ws = tmp_path / "home", tmp_path / "ws"
    ws.mkdir()
    epoch = int(time.time()) - 100
    _seed_class_a("opencode", home, ws, ws, "ses_one", "ses_two", started=epoch + 10)  # both in ws
    yaml_path, jobs = _control_plane(tmp_path / "ctl", [_row("w-opencode", "opencode", ws, epoch)])
    _job(jobs, "amb", "herdr:w-opencode", agent="opencode")
    calls = _count_captures(monkeypatch)

    res = _handle(jobs, "amb", yaml_path, home)
    assert res["status"] == "ambiguous" and sorted(res["candidates"]) == ["ses_one", "ses_two"]
    row = _rows(yaml_path)["w-opencode"]
    assert row["opencode_session_id_own"] is None
    assert row["last_visible_status"].startswith("autopin ambiguous (job amb): 2 candidates")
    assert _handle(jobs, "amb", yaml_path, home).get("deduplicated") is True
    assert calls == ["w-opencode"]


# --------------------------------------------------------------------------- 4) untouched rows, skipped jobs

def test_preassigned_and_existing_ids_are_untouched(tmp_path, owners, fake_opencode):
    home, ws = tmp_path / "home", tmp_path / "ws"
    ws.mkdir()
    epoch = int(time.time()) - 100
    _agy_conv(home, UB)
    _agy_cache(home, {ws: UB})
    held, _ = owners
    held[UB] = [5003]
    _seed_class_a("opencode", home, ws, ws, "ses_x", "ses_y", started=epoch + 10)
    rows = [
        _row("c-claude", "claude", ws, epoch, claude_session_id_own=UA, session_id_source="assigned"),
        _row("g-grok", "grok", ws, epoch, grok_session_id_own=UC, session_id_source="assigned"),
        _row("p-agy", "agy", ws, epoch, pid=5003, agy_conversation_id_own="agy-pinned-id",
             session_id_source="discovered", session_id_verified=True, last_visible_status="resume_verified"),
        _row("o-opencode", "opencode", ws, epoch, opencode_session_id_own="ses_kept"),
    ]
    yaml_path, jobs = _control_plane(tmp_path / "ctl", rows)
    before = _rows(yaml_path)
    for i, r in enumerate(rows):
        _job(jobs, f"k{i}", f"herdr:{r['name']}", agent=r["agent"])
        assert _handle(jobs, f"k{i}", yaml_path, home)["status"] in ("pre-assigned", "already-set")
    assert _rows(yaml_path) == before


@pytest.mark.parametrize("change,reason", [
    ({"transport": "remote", "session": "remote:box"}, "not a local agent session"),
    ({"agent": "human", "session": "human"}, "not a local agent session"),
    ({"status": "error"}, "job j is error"),
    ({"status": "running"}, "job j is running"),
])
def test_ineligible_jobs_pin_nothing(tmp_path, owners, change, reason):
    home, ws = tmp_path / "home", tmp_path / "ws"
    ws.mkdir()
    epoch = int(time.time()) - 100
    _agy_conv(home, UA)
    _agy_cache(home, {ws: UA})
    owners[0][UA] = [5001]
    yaml_path, jobs = _control_plane(tmp_path / "ctl", [_row("a-agy", "agy", ws, epoch, pid=5001)])
    _job(jobs, "j", change.get("session", "herdr:a-agy"), status=change.get("status", "completed"),
         transport=change.get("transport", "herdr"), agent=change.get("agent", "agy"))
    res = _handle(jobs, "j", yaml_path, home)
    assert (res["status"], res["reason"]) == ("skipped", reason)
    assert _rows(yaml_path)["a-agy"]["agy_conversation_id_own"] is None


def test_stale_dispatch_token_pins_nothing(tmp_path, owners):
    home, ws = tmp_path / "home", tmp_path / "ws"
    ws.mkdir()
    _agy_conv(home, UA)
    _agy_cache(home, {ws: UA})
    owners[0][UA] = [5001]
    yaml_path, jobs = _control_plane(tmp_path / "ctl", [_row("a-agy", "agy", ws, int(time.time()) - 100, pid=5001)])
    _job(jobs, "j", "herdr:a-agy", token="tok-2")
    res = _handle(jobs, "j", yaml_path, home, expect_token="tok-1")
    assert (res["status"], res["reason"]) == ("skipped", "stale dispatch token")
    assert not (jobs / "j" / "autopin.json").exists()


def test_call_sites_and_publisher_boundary():
    """The wrapper's direct path and create --onboard call the handler on
    `completed`; the agent-run publisher never touches the session registry."""
    wrapper = (SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job").read_text()
    assert 'delegate_autopin_completed "$JOB_ID" "$_disp_tok" "$REGISTRY_DIR"' in wrapper
    create = (SKILLS / "multi-agent-mux-create" / "scripts" / "create_session.sh").read_text()
    assert 'delegate_autopin_completed "$DELEGATE_JOB_ID" "$DELEGATE_JOB_TOKEN"' in create
    for f in ("publish_event.py", "publisher.py"):
        src = (SKILLS / "lib_py" / "delegation" / f).read_text()
        assert "atomic_dump_yaml" not in src and "agent-sessions" not in src and "autopin" not in src