"""Issue #72 - the multi-agent-mux-update skill.

Every test updates a copy of one workspace installed from this checkout
(MAM_REPO_URL, MAM_SKIP_VENV=1). herdr is a fake that answers `session list
--json` and `--session S agent list` from a JSON spec, so the pre-check's view
of the live agents is set per test; the monitor daemon is a stub reconcile.sh
that holds .mam/monitor.lock like the real one. Nothing here touches a real
herdr server or the control root's monitor.
"""
from __future__ import annotations

import fcntl
import hashlib
import json
import os
import shutil
import signal
import sqlite3
import subprocess
import sys
import time
from pathlib import Path

import pytest
import yaml

REPO_ROOT = Path(__file__).resolve().parent.parent
UPDATE_REL = ".agents/skills/multi-agent-mux-update/scripts/update.sh"
MONITOR_REL = ".agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh"

FAKE_HERDR = r'''#!/usr/bin/env python3
import json, os, sys
spec = json.load(open(os.environ["FAKE_HERDR_SPEC"]))
a = sys.argv[1:]
server = "default"
if a[:1] == ["--session"]:
    server, a = a[1], a[2:]
with open(os.environ["FAKE_HERDR_SPEC"] + ".calls", "a") as f:
    f.write(json.dumps([server] + a) + "\n")
if a[:2] == ["session", "list"]:
    if spec.get("fail") == "session-list":
        sys.exit(1)
    print(json.dumps({"sessions": [{"name": n, "running": r} for n, r in spec.get("sessions", {"default": True}).items()]}))
elif a[:2] == ["agent", "list"]:
    if spec.get("fail") == "agent-list:" + server:
        sys.exit(1)
    print(json.dumps({"result": {"agents": spec.get("agents", {}).get(server, [])}}))
elif a[:2] == ["agent", "get"]:
    hit = [x for x in spec.get("agents", {}).get(server, []) if a[2] in (x.get("name"), x.get("pane_id"))]
    if not hit:
        sys.exit(1)
    print(json.dumps({"result": {"agent": hit[0]}}))
elif a[:2] == ["pane", "close"]:
    spec.setdefault("agents", {})[server] = [x for x in spec.get("agents", {}).get(server, [])
                                             if x.get("pane_id") != a[2]]
    json.dump(spec, open(os.environ["FAKE_HERDR_SPEC"], "w"))
'''

STUB_MONITOR = r'''#!/usr/bin/env bash
# Issue #72 test stub of reconcile.sh: a wrapper whose python child holds monitor.lock.
d="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../../.." && pwd -P)"
echo "start $$ $*" >> "$d/.mam/stub-monitor.log"
python3 -c '
import fcntl, os, signal, sys, time
if os.environ.get("STUB_MONITOR_IGNORE_TERM") == "1":
    signal.signal(signal.SIGTERM, signal.SIG_IGN)
f = open(sys.argv[1], "w")
fcntl.flock(f, fcntl.LOCK_EX)
time.sleep(600)
' "$d/.mam/monitor.lock" &
child=$!
if [ "${STUB_MONITOR_IGNORE_TERM:-0}" = "1" ]; then
  trap '' TERM
else
  trap 'kill $child 2>/dev/null; exit 143' TERM
fi
while kill -0 "$child" 2>/dev/null; do wait "$child"; done
'''


def _clean_env():
    env = {k: v for k, v in os.environ.items()
           if not k.startswith(("HERDR_", "XDG_", "MAM_", "STUB_"))
           and k not in ("WORKSPACE_ROOT", "AGENT_SESSIONS_YAML", "DELEGATE_JOB_LOGS_DIR", "PYTHONPATH")}
    env.update(MAM_REPO_URL=str(REPO_ROOT), MAM_SKIP_VENV="1",
               MAM_INSTALLER_URL="file://" + str(REPO_ROOT / "deploy" / "install.sh"),
               MAM_ENV_FILE="/nonexistent/.mam.env", PYTHONDONTWRITEBYTECODE="1")
    return env


@pytest.fixture(scope="module")
def template(tmp_path_factory):
    ws = tmp_path_factory.mktemp("tpl72") / "ws"
    ws.mkdir()
    res = subprocess.run(["bash", str(REPO_ROOT / "deploy" / "install.sh"), str(ws)], env=_clean_env(),
                         capture_output=True, text=True, timeout=600)
    assert res.returncode == 0, res.stdout + res.stderr
    return ws


class Box:
    def __init__(self, root: Path, ws: Path):
        self.root, self.ws = root, ws
        self.mam = ws / ".mam"
        self.bin = root / "bin"
        self.spec = root / "herdr.json"
        self.env = _clean_env()
        self.env["PATH"] = "%s:%s" % (self.bin, os.environ["PATH"])
        self.env["FAKE_HERDR_SPEC"] = str(self.spec)
        self.helpers = []
        self.herdr()

    def herdr(self, sessions=None, agents=None, fail=None):
        self.spec.write_text(json.dumps({"sessions": sessions or {"default": True},
                                         "agents": agents or {}, "fail": fail}))

    def update(self, *args, env=None, timeout=300):
        e = dict(self.env, **(env or {}))
        return subprocess.run(["bash", str(self.ws / UPDATE_REL), "--yes", *args], cwd=str(self.ws), env=e,
                              stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=timeout)

    def lib(self, script, env=None, timeout=120):
        e = dict(self.env, WORKSPACE_ROOT=str(self.ws), **(env or {}))
        return subprocess.run(["bash", "-c", 'source "$1/.agents/skills/lib.sh"\n' + script, "_", str(self.ws)],
                              env=e, capture_output=True, text=True, timeout=timeout)

    def installer(self, body):
        p = self.root / "installer.sh"
        p.write_text("#!/usr/bin/env bash\nws=\"$1\"\n" + body)
        self.env["MAM_INSTALLER_URL"] = "file://" + str(p)

    def helper(self, *argv, **kw):
        p = subprocess.Popen(list(argv), start_new_session=True, **kw)
        self.helpers.append(p)
        return p

    def stub_monitor(self):
        p = self.ws / MONITOR_REL
        p.write_text(STUB_MONITOR)

    def start_monitor(self, env=None):
        """The way create and resume start it: lib.sh start_watchdog (detached, so
        not a child of this test that would linger as a zombie)."""
        res = self.lib('start_watchdog test "$WORKSPACE_ROOT"', env=dict(env or {}))
        assert res.returncode == 0 and res.stdout.strip(), res.stdout + res.stderr
        _until(lambda: _lock_held(self.mam / "monitor.lock"))

    def monitor_pids(self):
        return _exact_monitor_pids(self.ws / MONITOR_REL)

    def decoy(self):
        """A process that only names this workspace's monitor in an argument."""
        return self.helper("/bin/bash", "-c", "sleep 120 & wait $!",
                           "bash %s --subscribe" % (self.ws / MONITOR_REL))

    def monitor_starts(self):
        p = self.mam / "stub-monitor.log"
        return p.read_text().count("start ") if p.exists() else 0


def _until(cond, timeout=15.0):
    end = time.monotonic() + timeout
    while time.monotonic() < end:
        if cond():
            return True
        time.sleep(0.1)
    raise AssertionError("condition not met within %ss" % timeout)


def _exact_monitor_pids(script) -> list:
    """`bash <script> --subscribe ...` processes, judged here independently of
    the code under test (a `pgrep -f` would also count a decoy)."""
    out = subprocess.run(["ps", "-A", "-ww", "-o", "pid=", "-o", "args="], capture_output=True, text=True).stdout
    want = "%s --subscribe" % script
    pids = []
    for line in out.splitlines():
        pid, _, args = line.strip().partition(" ")
        argv0, _, rest = args.partition(" ")
        if os.path.basename(argv0) == "bash" and (rest == want or rest.startswith(want + " ")):
            pids.append(pid)
    return pids


def _descendants(pid) -> list:
    kids = subprocess.run(["pgrep", "-P", str(pid)], capture_output=True, text=True).stdout.split()
    return [d for k in kids for d in [k] + _descendants(k)]


def _processes_naming(path) -> list:
    """Live processes whose command line names PATH (read only, to report leftovers)."""
    out = subprocess.run(["ps", "-A", "-ww", "-o", "pid=", "-o", "args="], capture_output=True, text=True).stdout
    return [line.strip() for line in out.splitlines()
            if str(path) in line and line.split(None, 1)[0] != str(os.getpid())]


def _lock_held(path: Path) -> bool:
    if not path.exists():
        return False
    with open(path, "a") as fh:
        try:
            fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
            fcntl.flock(fh, fcntl.LOCK_UN)
            return False
        except BlockingIOError:
            return True


@pytest.fixture
def box(template, tmp_path):
    root = Path(os.path.realpath(tmp_path))
    ws = root / "ws"
    shutil.copytree(template, ws, symlinks=True)
    (root / "bin").mkdir()
    herdr = root / "bin" / "herdr"
    herdr.write_text(FAKE_HERDR)
    herdr.chmod(0o755)
    b = Box(root, ws)
    try:
        yield b
    finally:
        # Only what this test started: its helpers' process groups, and the
        # monitors whose exact command line is this test's own workspace
        # (detached, so in a group of their own), with their children. No
        # pattern kill: the host runs other workspaces' monitors.
        for p in b.helpers:
            try:
                os.killpg(p.pid, signal.SIGKILL)
            except OSError:
                pass
            p.wait()
        for pid in _exact_monitor_pids(ws / MONITOR_REL):
            for q in [pid] + _descendants(pid):
                try:
                    os.kill(int(q), signal.SIGKILL)
                except OSError:
                    pass
        try:
            _until(lambda: not _processes_naming(ws), timeout=10)
        except AssertionError:
            raise AssertionError("processes of this test outlived it: %s" % _processes_naming(ws))


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

SKIP_SNAPSHOT = ("shim/", "update-admission.lock", "agent-sessions.db.lock", "monitor.lock", "stub-monitor.log",
                 "agent-sessions.db-wal", "agent-sessions.db-shm")


def _db_content(p: Path) -> str:
    """The registry DB by its rows: on macOS SQLite keeps -wal/-shm after close and
    a read rewrites -shm, and a checkpoint moves the same rows between -wal and
    the file, so neither file's bytes say whether the data changed. A file that
    is not a database is compared by its bytes."""
    conn = sqlite3.connect(str(p))
    try:
        return "DB:" + hashlib.sha256("\n".join(conn.iterdump()).encode()).hexdigest()
    except sqlite3.DatabaseError:
        return hashlib.sha256(p.read_bytes()).hexdigest()
    finally:
        conn.close()


def snapshot(ws: Path) -> dict:
    out = {}
    for dp, dns, fns in os.walk(ws):
        dns[:] = [d for d in dns if d != "__pycache__"]
        for f in fns:
            p = Path(dp) / f
            rel = str(p.relative_to(ws))
            if rel.startswith(".mam/") and any(s in rel for s in SKIP_SNAPSHOT):
                continue
            if rel == ".mam/agent-sessions.db":
                out[rel] = _db_content(p)
                continue
            out[rel] = "L:" + os.readlink(p) if p.is_symlink() else hashlib.sha256(p.read_bytes()).hexdigest()
    return out


def write_rows(ws: Path, rows, **top):
    d = dict(top, herdr_sessions=rows)
    (ws / ".mam" / "agent-sessions.yaml").write_text(yaml.safe_dump(d, sort_keys=False))


def row(name, status="running", server="default", cwd=None, **extra):
    r = {"name": name, "status": status, "role": "creator", "herdr_session": server, "herdr_server": server,
         "pane": {"cwd": cwd or "/elsewhere", "cmd": "claude"}}
    r.update(extra)
    return r


def write_job(d: Path, jid, status="running", session="herdr:x", **extra):
    d.mkdir(parents=True, exist_ok=True)
    rec = dict({"job_id": jid, "status": status, "agent_session": session}, **extra)
    (d / (jid + ".json")).write_text(json.dumps(rec))


def live_marker(path: Path, box: Box):
    """A marker owned by a live process (pid + lstart as loop_lock.sh writes them)."""
    p = box.helper("sleep", "120")
    ls = subprocess.run(["ps", "-p", str(p.pid), "-o", "lstart="], capture_output=True, text=True).stdout
    ls = " ".join(ls.split())
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("pid=%d\nlstart=%s\nstarted=x\n" % (p.pid, ls))
    return p


def dead_marker(path: Path):
    p = subprocess.Popen(["true"])
    p.wait()
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("pid=%d\nlstart=Thu Jan  1 00:00:00 1970\nstarted=x\n" % p.pid)


def assert_refused_unchanged(box: Box, before: dict, res, *needles):
    assert res.returncode == 1, res.stdout + res.stderr
    assert "update refused; nothing was changed" in res.stderr, res.stderr
    for n in needles:
        assert n in res.stderr, (n, res.stderr)
    assert snapshot(box.ws) == before
    assert not (box.mam / "update-in-progress").exists(), "the refusing update releases its marker"
    assert not (box.mam / "update-backup").exists()


# --- success and the #8 state rules ------------------------------------------------

def test_update_in_place_keeps_mam_and_merges_install_state(box):
    mam = box.mam
    (mam / "version.txt").write_text("source=old\ncommit=stale0000\nfetched_at=then\nmethod=git\n")
    (mam / "herdr_workspace_id").write_text("w9\n")
    for d in ("briefs", "diagnostics", "resumelogs", "stoplogs", "jobs", "delegate_job_logs/j1"):
        (mam / d).mkdir(parents=True, exist_ok=True)
        (mam / d / "keep.txt").write_text(d)
    write_job(mam / "jobs", "done1", status="completed")
    (mam / "multi-agent-mux-monitor.log").write_text("log\n")
    (box.ws / ".mam.env").write_text("MQTT_BROKER=example\n")
    with open(mam / "install_state", "a") as fh:
        fh.write("custom_marker=kept\n")
    write_rows(box.ws, [row("a", status="stopped")], orchestrator_uuids=["u1"])
    keep = {k: v for k, v in snapshot(box.ws).items()
            if k.startswith((".mam/herdr_workspace_id", ".mam/briefs", ".mam/diagnostics", ".mam/resumelogs",
                             ".mam/stoplogs", ".mam/jobs", ".mam/delegate_job_logs", ".mam/multi-agent-mux-monitor.log"))
            or k == ".mam.env"}

    res = box.update()
    assert res.returncode == 0, res.stdout + res.stderr
    after = snapshot(box.ws)
    assert {k: after.get(k) for k in keep} == keep, "everything that is not the installer's is kept"
    assert "stale0000" not in (mam / "version.txt").read_text()
    state = (mam / "install_state").read_text().splitlines()
    assert "custom_marker=kept" in state and state.count("gitignore_created=1") == 1, state
    assert not (box.ws / ".mam_deploy" / "update.sh").exists()
    assert not (mam / "update-backup").exists() and not (mam / "update-in-progress").exists()
    assert "Update complete." in res.stdout


def test_old_format_registry_is_converted_by_the_new_code(box):
    old = row("old-one", status="stopped", start_command="HERDR_SESSION_NAME=x herdr new-session",
              kill_command="HERDR_SESSION_NAME=x herdr kill-session", attach_command="HERDR_SESSION_NAME=x herdr agent attach old-one")
    write_rows(box.ws, [old])
    res = box.update()
    assert res.returncode == 0, res.stdout + res.stderr
    assert "registry-format: 1 -> 2, rows changed: 1" in res.stdout
    y = yaml.safe_load((box.mam / "agent-sessions.yaml").read_text())
    assert y["registry_format"] == 2
    conn = sqlite3.connect(str(box.mam / "agent-sessions.db"))
    try:
        db = {r["name"]: r for r in (json.loads(x[0]) for x in conn.execute("SELECT data FROM sessions"))}
        st = json.loads(conn.execute("SELECT data FROM state WHERE id=1").fetchone()[0])
    finally:
        conn.close()
    assert {r["name"]: r for r in y["herdr_sessions"]} == db and st["registry_format"] == 2
    assert "kill_command" not in db["old-one"] and "start_command" not in db["old-one"]


def test_no_registry_is_nothing_to_convert(box):
    res = box.update()
    assert res.returncode == 0, res.stdout + res.stderr
    assert "registry: nothing to convert (no registry)" in res.stdout
    assert not (box.mam / "agent-sessions.yaml").exists()


def test_a_failed_conversion_keeps_the_install_and_exits_nonzero(box, tmp_path):
    src = tmp_path / "src"
    shutil.copytree(REPO_ROOT, src, symlinks=True,
                    ignore=shutil.ignore_patterns(".git", ".venv", ".mam", "__pycache__", "tests", "nats-docker"))
    fmt = src / ".agents" / "skills" / "lib_py" / "registry_format.py"
    fmt.write_text("import sys\nif __name__ == '__main__':\n    sys.exit(3)\n" + fmt.read_text())
    box.env["MAM_REPO_URL"] = str(src)
    box.stub_monitor()
    box.start_monitor()
    write_rows(box.ws, [row("a", status="stopped")])
    res = box.update()
    assert res.returncode == 3, res.stdout + res.stderr
    assert "the registry was not converted" in res.stderr
    assert (box.ws / ".agents/skills/lib_py/registry_format.py").read_text() == fmt.read_text(), "install kept"
    assert box.monitor_pids() and box.monitor_starts() == 2, "the monitor is started again anyway"
    assert not (box.mam / "update-backup").exists() and not (box.mam / "update-in-progress").exists()


def test_refuses_outside_an_install_and_without_yes(box):
    (box.ws / ".mam_deploy" / "remove.sh").unlink()
    res = box.update()
    assert res.returncode == 1 and "not an installed MAM workspace" in res.stderr
    shutil.copy(REPO_ROOT / "deploy" / "remove.sh", box.ws / ".mam_deploy" / "remove.sh")
    res = subprocess.run(["bash", str(box.ws / UPDATE_REL)], env=box.env, stdin=subprocess.DEVNULL,
                         capture_output=True, text=True, timeout=60)
    assert res.returncode == 1 and "-y/--yes" in res.stderr
    res = box.update("--force")
    assert res.returncode == 2, "--force is not an alias of --yes"


# --- refusals (decision 1) ---------------------------------------------------------

def test_refuses_a_running_row(box):
    write_rows(box.ws, [row("busy-creator", cwd="/somewhere/else")])
    before = snapshot(box.ws)
    res = box.update()
    assert_refused_unchanged(box, before, res, "session busy-creator is running", "stop_session.sh --session busy-creator")


def test_refuses_an_unreadable_registry(box):
    (box.mam / "agent-sessions.db").write_text("not a database")
    before = snapshot(box.ws)
    assert_refused_unchanged(box, before, box.update(), "cannot be read")


def test_refuses_a_rowless_agent_on_any_running_server(box):
    box.herdr(sessions={"default": True, "picked-by-hand": True, "down": False},
              agents={"picked-by-hand": [{"name": "stray", "pane_id": "w3:p1", "cwd": str(box.ws / "sub")}],
                      "down": [{"name": "ghost", "pane_id": "w4:p1", "cwd": str(box.ws)}]})
    before = snapshot(box.ws)
    res = box.update()
    assert_refused_unchanged(box, before, res, "herdr agent stray (pane w3:p1, server picked-by-hand)")
    assert "ghost" not in res.stderr, "a server that is not running is not asked"
    calls = (box.root / "herdr.json.calls").read_text()
    assert '"down"' not in calls


def test_the_rowless_agent_hint_stops_it_on_the_server_it_was_found_on(box):
    """R3: the printed stop command names the server; without it stop looks on
    the workspace's server, where a same-named agent runs outside the workspace."""
    write_rows(box.ws, [])
    box.herdr(sessions={"default": True, "unrecorded-server": True},
              agents={"unrecorded-server": [{"name": "stray", "pane_id": "w3:p1", "cwd": str(box.ws / "sub")}],
                      "default": [{"name": "stray", "pane_id": "w1:p1", "cwd": str(box.root / "elsewhere")}]})
    res = box.update()
    assert res.returncode == 1, res.stdout + res.stderr
    line = next(ln for ln in res.stderr.splitlines() if "herdr agent stray" in ln)
    cmd = line.split("stop it: ", 1)[1]
    assert "--herdr-session unrecorded-server" in cmd, cmd
    stop = subprocess.run(["bash", "-c", cmd], cwd=str(box.ws), env=box.env, stdin=subprocess.DEVNULL,
                          capture_output=True, text=True, timeout=120)
    assert stop.returncode == 0, stop.stdout + stop.stderr
    agents = json.loads(box.spec.read_text())["agents"]
    assert agents["unrecorded-server"] == []
    assert [x["pane_id"] for x in agents["default"]] == ["w1:p1"], "the same name elsewhere is left alone"
    res = box.update()
    assert res.returncode == 0, res.stdout + res.stderr


def test_an_agent_outside_the_workspace_does_not_refuse(box):
    box.herdr(agents={"default": [{"name": "other", "pane_id": "w1:p2", "cwd": str(box.root / "elsewhere")}]})
    res = box.update()
    assert res.returncode == 0, res.stdout + res.stderr


@pytest.mark.parametrize("fail", ["session-list", "agent-list:default"])
def test_refuses_when_herdr_cannot_answer(box, fail):
    box.herdr(fail=fail)
    before = snapshot(box.ws)
    assert_refused_unchanged(box, before, box.update(), "cannot list the live agents")


def test_open_jobs_refuse_and_terminal_or_dry_run_do_not(box, tmp_path):
    jobs = box.mam / "jobs"
    write_job(jobs, "done1", status="completed")
    write_job(jobs, "err1", status="error")
    write_job(jobs, "can1", status="cancelled")
    write_job(jobs, "dry1", status="pending", dry_run=True)
    assert box.update().returncode == 0

    write_job(jobs, "open1", status="pending", session="remote:x")
    before = snapshot(box.ws)
    res = box.update()
    assert_refused_unchanged(box, before, res, "job open1", "close --job open1 --request cancelled")
    (jobs / "open1.json").unlink()

    (jobs / "broken.json").write_text("{not json")
    before = snapshot(box.ws)
    assert_refused_unchanged(box, before, box.update(), "job record %s cannot be read" % (jobs / "broken.json"))
    (jobs / "broken.json").unlink()

    extra = tmp_path / "mam-jobs-dir"
    write_job(extra, "env1", status="running")
    box.env["MAM_JOBS_DIR"] = str(extra)
    before = snapshot(box.ws)
    assert_refused_unchanged(box, before, box.update(), "job env1")
    del box.env["MAM_JOBS_DIR"]


@pytest.mark.skipif(os.geteuid() == 0, reason="root reads a mode-000 directory")
def test_a_jobs_directory_that_cannot_be_listed_refuses(box, tmp_path):
    """R2: an unreadable .mam/jobs (or MAM_JOBS_DIR under an unreadable parent)
    is not an empty registry."""
    jobs = box.mam / "jobs"
    write_job(jobs, "hidden1", status="running")
    before = snapshot(box.ws)
    jobs.chmod(0)
    try:
        res = box.update()
    finally:
        jobs.chmod(0o755)
    assert_refused_unchanged(box, before, res, "jobs directory %s cannot be listed" % jobs)
    (jobs / "hidden1.json").unlink()

    parent = tmp_path / "locked"
    write_job(parent / "jobs", "env1", status="running")
    box.env["MAM_JOBS_DIR"] = str(parent / "jobs")
    before = snapshot(box.ws)
    parent.chmod(0)
    try:
        res = box.update()
    finally:
        parent.chmod(0o755)
    assert_refused_unchanged(box, before, res, "jobs directory %s cannot be listed" % (parent / "jobs"))
    del box.env["MAM_JOBS_DIR"]


def test_other_control_planes_count_only_jobs_addressed_to_this_registry(box, tmp_path):
    other = tmp_path / "other-plane" / "jobs"
    write_job(other, "theirs", status="running", session="herdr:not-ours")
    write_rows(box.ws, [row("wt-creator", status="stopped", control_plane={"mam_jobs_dir": str(other)})])
    assert box.update().returncode == 0, "a job of another registry's session is not ours"
    write_job(other, "ours", status="running", session="herdr:wt-creator")
    before = snapshot(box.ws)
    assert_refused_unchanged(box, before, box.update(), "job ours")


def test_loop_marker_live_refuses_stale_does_not(box):
    live_marker(box.mam / "loop-guard-active", box)
    before = snapshot(box.ws)
    assert_refused_unchanged(box, before, box.update(), "a review loop is running")
    dead_marker(box.mam / "loop-guard-active")
    res = box.update()
    assert res.returncode == 0, res.stdout + res.stderr
    assert "stale loop marker" in res.stdout


# --- the calling agent (decision 7, N2) ----------------------------------------------

def _me_env(server="mam-x", pane="w5:p1"):
    return {"HERDR_SESSION": server, "HERDR_PANE_ID": pane}


def test_the_calling_agent_its_row_and_its_running_job_are_excluded(box):
    me = {"name": "me-creator", "pane_id": "w5:p1", "cwd": str(box.ws)}
    box.herdr(sessions={"default": True, "mam-x": True}, agents={"mam-x": [me]})
    write_rows(box.ws, [row("me-creator", server="mam-x", cwd=str(box.ws))])
    write_job(box.mam / "jobs", "myjob", status="running", session="herdr:me-creator")
    res = box.update(env=_me_env())
    assert res.returncode == 0, res.stdout + res.stderr

    write_job(box.mam / "jobs", "myjob2", status="pending", session="herdr:me-creator")
    before = snapshot(box.ws)
    assert_refused_unchanged(box, before, box.update(env=_me_env()), "job myjob2")


def test_same_pane_id_on_another_server_is_not_excluded(box):
    box.herdr(sessions={"default": True, "mam-x": True, "mam-y": True},
              agents={"mam-x": [{"name": "me", "pane_id": "w5:p1", "cwd": str(box.ws)}],
                      "mam-y": [{"name": "twin", "pane_id": "w5:p1", "cwd": str(box.ws)}]})
    before = snapshot(box.ws)
    res = box.update(env=_me_env())
    assert_refused_unchanged(box, before, res, "herdr agent twin (pane w5:p1, server mam-y)")
    assert "agent me " not in res.stderr


def test_a_nameless_caller_pane_is_excluded_by_pane_only(box):
    box.herdr(sessions={"default": True, "mam-x": True},
              agents={"mam-x": [{"name": None, "pane_id": "w5:p1", "cwd": str(box.ws)}]})
    assert box.update(env=_me_env()).returncode == 0
    write_job(box.mam / "jobs", "nameless", status="running", session="herdr:")
    write_job(box.mam / "jobs", "nonename", status="running", session="herdr:None")
    before = snapshot(box.ws)
    assert_refused_unchanged(box, before, box.update(env=_me_env()), "job nameless", "job nonename")


@pytest.mark.parametrize("env", [{}, {"HERDR_PANE_ID": "w5:p1"}, {"HERDR_SESSION": "mam-x"}])
def test_nothing_is_excluded_when_the_caller_is_not_identified(box, env):
    box.herdr(sessions={"default": True, "mam-x": True},
              agents={"mam-x": [{"name": "me", "pane_id": "w5:p1", "cwd": str(box.ws)}]})
    before = snapshot(box.ws)
    res = box.update(env=env)
    assert_refused_unchanged(box, before, res, "herdr agent me", "nothing was excluded as the calling agent")


# --- the monitor (decision 2, U3) ----------------------------------------------------

def test_a_running_monitor_is_stopped_and_started_again(box):
    box.stub_monitor()
    box.start_monitor()
    first = set(box.monitor_pids())
    res = box.update()
    assert res.returncode == 0, res.stdout + res.stderr
    assert "monitor: stopping the daemon" in res.stdout and "monitor: restarted" in res.stdout
    now = set(box.monitor_pids())
    assert now and not (now & first) and box.monitor_starts() == 2


def test_no_monitor_before_means_none_after(box):
    box.stub_monitor()
    res = box.update()
    assert res.returncode == 0, res.stdout + res.stderr
    assert not box.monitor_pids() and box.monitor_starts() == 0


def test_no_subscriber_daemon_flag_wins(box):
    box.stub_monitor()
    box.start_monitor()
    res = box.update(env={"MAM_NO_SUBSCRIBER_DAEMON": "1"})
    assert res.returncode == 0, res.stdout + res.stderr
    assert "monitor: not restarted (MAM_NO_SUBSCRIBER_DAEMON=1)" in res.stdout
    assert not box.monitor_pids()


def test_a_monitor_that_ignores_term_is_killed_restarted_and_nothing_installed(box):
    box.stub_monitor()
    box.start_monitor(env={"STUB_MONITOR_IGNORE_TERM": "1"})
    box.installer("touch \"$ws/INSTALLER-RAN\"\n")
    before = snapshot(box.ws)
    res = box.update(env={"MAM_UPDATE_MONITOR_STOP_S": "1"})
    assert res.returncode == 1, res.stdout + res.stderr
    assert "it was killed and is started again" in res.stderr
    assert not (box.ws / "INSTALLER-RAN").exists()
    assert snapshot(box.ws) == before
    assert box.monitor_pids() and box.monitor_starts() == 2
    assert not (box.mam / "update-in-progress").exists()


def test_a_monitor_lock_held_by_an_unknown_process_refuses(box):
    lock = box.mam / "monitor.lock"
    with open(lock, "w") as fh:
        fcntl.flock(fh, fcntl.LOCK_EX)
        before = snapshot(box.ws)
        res = box.update()
    assert_refused_unchanged(box, before, res, "monitor.lock is held by a monitor the skills did not start")


@pytest.mark.parametrize("finder", ["python", "shell"])
def test_only_this_workspaces_exact_monitor_command_line_is_its_monitor(tmp_path, finder):
    """The update signals what `lib_py.update monitor` returns, and the legacy
    start_watchdog (MAM_DELEGATION_PY=0) reuses what lib.sh _monitor_pids finds:
    `bash <this workspace's reconcile.sh> --subscribe ...` only - not another
    workspace's daemon the path matched as a regex (ws.a, wsXa), nor a process
    that names the path in an argument. A path with a space is still found."""
    root = Path(os.path.realpath(tmp_path))
    procs = []

    def daemon_in(ws):
        script = ws / MONITOR_REL
        script.parent.mkdir(parents=True)
        script.write_text("sleep 120 & wait $!\n")
        procs.append(subprocess.Popen(["bash", str(script), "--subscribe", "--idle-timeout", "0"],
                                      start_new_session=True))
        return str(procs[-1].pid)

    def monitor(ws):
        if finder == "shell":
            res = subprocess.run(["bash", "-c", 'source "$1" && _monitor_pids "$2"', "_",
                                  str(REPO_ROOT / ".agents" / "skills" / "lib.sh"), str(ws / MONITOR_REL)],
                                 env=dict(_clean_env(), WORKSPACE_ROOT=str(root)), capture_output=True, text=True)
            assert res.returncode == 0, res.stderr
            return res.stdout.split()
        res = subprocess.run([sys.executable, "-m", "lib_py.update", "monitor", "--workspace", str(ws)],
                             cwd=str(REPO_ROOT / ".agents" / "skills"), capture_output=True, text=True)
        assert res.returncode == 0, res.stderr
        return json.loads(res.stdout)["pids"]

    try:
        a, other, spaced = root / "ws.a", root / "wsXa", root / "w s"
        a.mkdir()
        daemon_in(other)
        procs.append(subprocess.Popen(["/bin/bash", "-c", "sleep 120 & wait $!",
                                       "bash %s --subscribe" % (a / MONITOR_REL)], start_new_session=True))
        pid = daemon_in(spaced)
        _until(lambda: monitor(spaced) == [pid])
        assert monitor(a) == []
    finally:
        for p in procs:
            try:
                os.killpg(p.pid, signal.SIGKILL)
            except OSError:
                pass
            p.wait()


@pytest.mark.parametrize("case", ["installed", "killed", "rolled-back"])
@pytest.mark.parametrize("mode", ["0", "1"])
def test_every_restart_starts_a_real_monitor_in_both_delegation_modes(box, mode, case):
    """MAM_DELEGATION_PY=0 keeps the legacy start_watchdog, the switch the
    delegate skill documents. Its restart, after an install, after a stop
    timeout or after a rollback, must start this workspace's monitor again, out
    of the caller's process group (decision 2), and not report a process that
    only names the monitor path in an argument."""
    env = {"MAM_DELEGATION_PY": mode}
    box.stub_monitor()
    box.start_monitor(env=dict(env, STUB_MONITOR_IGNORE_TERM="1") if case == "killed" else env)
    first = set(box.monitor_pids())
    decoy = box.decoy()
    _until(lambda: str(decoy.pid) in subprocess.run(["ps", "-A", "-ww", "-o", "pid=", "-o", "args="],
                                                     capture_output=True, text=True).stdout)
    if case == "killed":
        env["MAM_UPDATE_MONITOR_STOP_S"] = "1"
    if case == "rolled-back":
        box.installer("exit 7\n")
    res = box.update(env=env)
    assert res.returncode == (0 if case == "installed" else 1), res.stdout + res.stderr
    assert {"installed": "monitor: restarted", "killed": "it was killed and is started again",
            "rolled-back": "rolled back"}[case] in res.stdout + res.stderr
    assert "PID %s)" % decoy.pid not in res.stdout, "the decoy is reported as the monitor"
    _until(lambda: bool(box.monitor_pids()))
    assert not (set(box.monitor_pids()) & first) and box.monitor_starts() == 2
    # box.update runs update.sh in this test's process group: the caller's.
    assert all(os.getpgid(int(p)) != os.getpgrp() for p in box.monitor_pids()), "left in the caller's group"
    assert decoy.poll() is None, "the decoy was not signalled"


# --- rollback (decisions 3, 10; U4) --------------------------------------------------

FAILING_INSTALL = """
echo broken >> "$ws/.agents/skills/lib.sh"
rm -f "$ws/.mam_deploy/remove.sh"
echo new > "$ws/MESSAGING.md"
mkdir -p "$ws/.venv/lib" && echo junk > "$ws/.venv/lib/junk"
echo junk > "$ws/.mam/version.txt"
echo junk > "$ws/.mam/asset_hashes.txt"
mkdir -p "$ws/.agents/skills/half-installed" && echo x > "$ws/.agents/skills/half-installed/f"
python3 -c 'import yaml,sys; p=sys.argv[1]; d=yaml.safe_load(open(p)); d["herdr_sessions"]=[]; yaml.safe_dump(d, open(p, "w"))' "$ws/.mam/agent-sessions.yaml"
exit 7
"""


def test_a_failed_install_rolls_everything_back(box):
    write_rows(box.ws, [row("a", status="stopped")])
    db = box.mam / "agent-sessions.db"
    conn = sqlite3.connect(str(db))
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA wal_autocheckpoint=0")
    conn.execute("CREATE TABLE state (id INTEGER PRIMARY KEY, data TEXT)")
    conn.execute("CREATE TABLE sessions (name TEXT PRIMARY KEY, status TEXT, pane_cwd TEXT, data JSON)")
    conn.execute("INSERT INTO state VALUES (1, ?)", (json.dumps({"orchestrator_uuids": ["in-wal"]}),))
    conn.execute("INSERT INTO sessions VALUES ('a', 'stopped', '/x', ?)", (json.dumps(row("a", status="stopped")),))
    conn.commit()
    assert (box.mam / "agent-sessions.db-wal").stat().st_size > 0, "the commit is still in the WAL"
    box.stub_monitor()
    box.start_monitor()
    box.installer(FAILING_INSTALL)
    before = snapshot(box.ws)
    try:
        res = box.update()
    finally:
        conn.close()
    assert res.returncode == 1, res.stdout + res.stderr
    assert "the install failed (rc 7); rolling back" in res.stderr and "rolled back" in res.stdout
    after = snapshot(box.ws)
    after.pop(".mam/agent-sessions.db", None)
    before.pop(".mam/agent-sessions.db", None)
    before.pop(".mam/agent-sessions.db-wal", None)
    before.pop(".mam/agent-sessions.db-shm", None)
    assert after == before, sorted(set(after.items()) ^ set(before.items()))[:10]
    assert not (box.ws / "MESSAGING.md").exists() and not (box.ws / ".venv").exists()
    check = sqlite3.connect(str(db))
    try:
        st = json.loads(check.execute("SELECT data FROM state WHERE id=1").fetchone()[0])
    finally:
        check.close()
    assert st == {"orchestrator_uuids": ["in-wal"]}, "the backup carried the WAL's commit"
    assert not (box.mam / "update-backup").exists() and not (box.mam / "update-in-progress").exists()
    assert box.monitor_pids() and box.monitor_starts() == 2, "the restored tree starts the monitor again"

    box.env["MAM_INSTALLER_URL"] = "file://" + str(REPO_ROOT / "deploy" / "install.sh")
    res = box.update()
    assert res.returncode == 0, "the skill runs again after a rollback: " + res.stdout + res.stderr


# --- leftovers (U4) ------------------------------------------------------------------

def _leftover(box: Box, state):
    bk = box.mam / "update-backup"
    (bk / "files" / ".agents" / "skills").mkdir(parents=True)
    shutil.copy2(box.ws / ".agents/skills/lib.sh", bk / "files/.agents/skills/lib.sh")
    (bk / "items").write_text(".agents/skills/lib.sh\nMESSAGING.md\n")
    (bk / "absent").write_text("MESSAGING.md\n")
    (bk / "registry").mkdir()
    (bk / "registry" / "agent-sessions.yaml").write_text("herdr_sessions: []\n")
    if state is not None:
        (bk / "state").write_text(state + "\n")
    return bk


@pytest.mark.parametrize("state", [None, "backing-up", "installed"])
def test_leftover_that_changed_nothing_or_finished_is_removed(box, state):
    _leftover(box, state)
    (box.ws / "MESSAGING.md").write_text("must survive: nothing is restored\n")
    res = box.update()
    assert res.returncode == 0, res.stdout + res.stderr
    assert "removing the backup a previous update left" in res.stdout
    assert (box.ws / "MESSAGING.md").exists()
    assert not (box.mam / "update-backup").exists()


def test_leftover_installing_restores_files_but_not_the_registry(box):
    _leftover(box, "installing")
    pristine = (box.ws / ".agents/skills/lib.sh").read_bytes()
    with open(box.ws / ".agents/skills/lib.sh", "a") as fh:
        fh.write("# half installed\n")
    (box.ws / "MESSAGING.md").write_text("made by the dead install\n")
    write_rows(box.ws, [row("written-after-crash", status="stopped")])
    box.installer("grep -q 'half installed' \"$ws/.agents/skills/lib.sh\" && exit 9\n"
                  "test -e \"$ws/MESSAGING.md\" && exit 8\n"
                  "cp \"$ws/.mam/agent-sessions.yaml\" \"$ws/../seen-registry.yaml\"\nexit 0\n")
    res = box.update()
    assert res.returncode == 0, res.stdout + res.stderr
    assert "the previous update was rolled back" in res.stdout
    assert (box.ws / ".agents/skills/lib.sh").read_bytes() == pristine
    assert "written-after-crash" in (box.root / "seen-registry.yaml").read_text(), "registry left as it was"
    assert not (box.mam / "update-backup").exists()


def test_leftover_whose_restore_fails_is_kept_and_nothing_more_changes(box):
    bk = _leftover(box, "installing")
    (bk / "items").write_text(".agents/skills/lib.sh\n.agents/skills/missing-from-backup\n")
    box.stub_monitor()
    box.start_monitor()
    box.installer("touch \"$ws/INSTALLER-RAN\"\n")
    res = box.update()
    assert res.returncode == 1, res.stdout + res.stderr
    assert "restoring the previous update's backup failed" in res.stderr
    assert bk.exists() and (bk / "state").read_text().strip() == "installing"
    assert not (box.ws / "INSTALLER-RAN").exists()
    assert not box.monitor_pids(), "the monitor stays stopped"
    assert not (box.mam / "update-in-progress").exists()


def test_a_state_write_failure_before_the_install_changes_nothing_and_restarts_the_monitor(box):
    real_mv = shutil.which("mv")
    fake = box.bin / "mv"
    fake.write_text("#!/usr/bin/env bash\n"
                    "src=\"${@: -2:1}\"\n"
                    "case \"$src\" in */update-backup/state.tmp) grep -qx installing \"$src\" && exit 1 ;; esac\n"
                    "exec %s \"$@\"\n" % real_mv)
    fake.chmod(0o755)
    box.stub_monitor()
    box.start_monitor()
    box.installer("touch \"$ws/INSTALLER-RAN\"\n")
    res = box.update()
    assert res.returncode == 1, res.stdout + res.stderr
    assert not (box.ws / "INSTALLER-RAN").exists()
    assert box.monitor_starts() == 2, "the monitor it stopped is started again"
    _until(lambda: bool(box.monitor_pids()))
    assert not (box.mam / "update-backup").exists() and not (box.mam / "update-in-progress").exists()
    assert "nothing was changed" in res.stderr


def test_the_backup_lives_in_the_workspace_and_is_gone_after_success(box):
    box.installer("test -f \"$ws/.mam/update-backup/state\" && grep -qx installing \"$ws/.mam/update-backup/state\" "
                  "&& test -f \"$ws/.mam/update-backup/files/.agents/skills/lib.sh\" && touch \"$ws/../saw-backup\"\n")
    res = box.update()
    assert res.returncode == 0, res.stdout + res.stderr
    assert (box.root / "saw-backup").exists()
    assert not (box.mam / "update-backup").exists()


# --- the marker and the admission lock (decision 8) -----------------------------------

def test_a_live_update_marker_refuses_a_second_update_and_is_left_alone(box):
    live_marker(box.mam / "update-in-progress", box)
    text = (box.mam / "update-in-progress").read_text()
    before = snapshot(box.ws)
    res = box.update()
    assert res.returncode == 1 and "another update is in progress" in res.stderr, res.stderr
    assert (box.mam / "update-in-progress").read_text() == text
    assert snapshot(box.ws) == before


def test_a_stale_update_marker_is_taken_over(box):
    dead_marker(box.mam / "update-in-progress")
    res = box.update()
    assert res.returncode == 0, res.stdout + res.stderr
    assert "stale update marker" in res.stdout
    assert not (box.mam / "update-in-progress").exists()


def test_two_updates_at_once_one_proceeds(box):
    box.installer("touch \"$ws/../first-installing\"; sleep 4; exit 0\n")
    first = box.helper("bash", str(box.ws / UPDATE_REL), "--yes", cwd=str(box.ws), env=box.env,
                       stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
    _until(lambda: (box.root / "first-installing").exists())
    marker = (box.mam / "update-in-progress").read_text()
    second = box.update()
    assert second.returncode == 1 and "another update is in progress" in second.stderr, second.stderr
    assert (box.mam / "update-in-progress").read_text() == marker, "the running update's marker is untouched"
    out, err = first.communicate(timeout=120)
    assert first.returncode == 0, out + err


def test_entries_are_refused_while_the_copy_installs(box):
    box.installer('if bash -c \'source "$1/.agents/skills/lib.sh" && mam_update_admit\' _ "$ws" 2> "$ws/../admit.err"; '
                  'then echo admitted > "$ws/../admit"; else echo refused > "$ws/../admit"; fi\n')
    res = box.update()
    assert res.returncode == 0, res.stdout + res.stderr
    assert (box.root / "admit").read_text().strip() == "refused"
    assert "an update of %s is in progress" % box.ws in (box.root / "admit.err").read_text()


@pytest.mark.parametrize("wait,needle", [("30", "session late-row is running"), ("1", "admission lock past")])
def test_an_entry_holding_the_admission_lock_is_waited_for_or_refuses(box, wait, needle):
    script = ('mam_update_admit || exit 1\necho admitted\nsleep 3\n'
              'printf "herdr_sessions:\\n- name: late-row\\n  status: running\\n" > "$WORKSPACE_ROOT/.mam/agent-sessions.yaml"\n'
              'mam_update_admit_release\nsleep 30\n')
    entry = box.helper("bash", "-c", 'source "$1/.agents/skills/lib.sh"\n' + script, "_", str(box.ws),
                       env=dict(box.env, WORKSPACE_ROOT=str(box.ws)), stdout=subprocess.PIPE, text=True)
    assert entry.stdout.readline().strip() == "admitted"
    box.installer("touch \"$ws/INSTALLER-RAN\"\n")
    res = box.update(env={"MAM_UPDATE_ADMISSION_WAIT_S": wait})
    assert res.returncode == 1, res.stdout + res.stderr
    assert needle in res.stderr, res.stderr
    assert not (box.ws / "INSTALLER-RAN").exists()
    assert not (box.mam / "update-in-progress").exists()


# --- the first transition (decision 4, criterion 5) ------------------------------------

def test_the_old_updater_installs_the_skill_which_then_converts_the_registry(tmp_path):
    """A workspace installed by v4.11.0 still has .mam_deploy/update.sh. Run once,
    it installs the new version - the skill, and no updater left behind - and
    copies the registry unconverted; the skill run right after converts it."""
    old_src = tmp_path / "v4.11.0"
    old_src.mkdir()
    arch = subprocess.run(["git", "-C", str(REPO_ROOT), "archive", "0e1e2e8"], capture_output=True)
    if arch.returncode != 0:
        pytest.skip("no git history to rebuild v4.11.0 from")
    subprocess.run(["tar", "-x", "-C", str(old_src)], input=arch.stdout, check=True)
    root = Path(os.path.realpath(tmp_path))
    ws = root / "ws"
    ws.mkdir()
    (root / "bin").mkdir()
    (root / "bin" / "herdr").write_text(FAKE_HERDR)
    (root / "bin" / "herdr").chmod(0o755)
    b = Box(root, ws)
    res = subprocess.run(["bash", str(old_src / "deploy" / "install.sh"), str(ws)],
                         env=dict(b.env, MAM_REPO_URL=str(old_src)), capture_output=True, text=True, timeout=600)
    assert res.returncode == 0, res.stdout + res.stderr
    assert (ws / ".mam_deploy" / "update.sh").exists()
    write_rows(ws, [row("old-one", status="stopped", kill_command="x", start_command="y")])

    res = subprocess.run(["bash", str(ws / ".mam_deploy" / "update.sh"), "--force"], cwd=str(ws), env=b.env,
                         capture_output=True, text=True, timeout=600, stdin=subprocess.DEVNULL)
    assert res.returncode == 0, res.stdout + res.stderr
    assert (ws / UPDATE_REL).exists() and not (ws / ".mam_deploy" / "update.sh").exists()
    assert "registry_format" not in yaml.safe_load((ws / ".mam" / "agent-sessions.yaml").read_text())

    res = b.update()
    assert res.returncode == 0, res.stdout + res.stderr
    assert "registry-format: 1 -> 2, rows changed: 1" in res.stdout
    assert yaml.safe_load((ws / ".mam" / "agent-sessions.yaml").read_text())["registry_format"] == 2