"""Issue #50, against a real herdr 0.9.3 server: the registry decides whether a
session is free (C), and three `create --onboard` at once do not contend (A).

Each test gets a one-off herdr server (a session named t50-<uuid>) under a
short mkdtemp HOME - herdr's sockets, logs and server config live there, the
user's herdr config is never touched, and a scratch HOME under tmp_path would
exceed the unix socket path limit. The pane runs a fake claude (a claude-like
welcome, then it echoes its input), which herdr 0.9.3 reports as `idle`; the
fake never shows `working`, so `agent prompt --wait` times out and the
injection is judged by the job's own `started` (Issue #18), which these tests
publish as the agent would, through the real publish_event.py and the suite's
local broker. The box is torn down after every test.

Measured on 2026-10-10 (job 51ccd800, herdr 0.9.3, a real claude): herdr
reported `working` only while a job ran and `idle` ~10 s after `completed`,
background subagent (`← 1 agent`) or not, and a second job was dispatched;
so the herdr idle wait stays and the registry wait runs before it.
"""
from __future__ import annotations

import calendar
import json
import os
import re
import shutil
import signal
import subprocess
import sys
import tempfile
import threading
import time
import uuid
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parent.parent
SKILLS = REPO_ROOT / ".agents" / "skills"
WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
PUBLISH_PY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py"
REGISTRY_PY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "registry.py"
CREATE_SH = SKILLS / "multi-agent-mux-create" / "scripts" / "create_session.sh"
if str(SKILLS) not in sys.path:
    sys.path.insert(0, str(SKILLS))

from lib_py.delegation import registry  # noqa: E402

HERDR = shutil.which("herdr", path="/home/linuxbrew/.linuxbrew/bin:/usr/local/bin:" + os.environ.get("PATH", ""))
pytestmark = pytest.mark.skipif(not HERDR, reason="herdr is not installed")

SESSION = "t50-creator-claude"
FAKE_CLAUDE = r'''#!/usr/bin/env bash
# fake claude for the real-herdr tests: a claude-like welcome, then it echoes its input
if [ "${1:-}" = "auth" ]; then echo '{"loggedIn": true}'; exit 0; fi
printf '\n ▐▛███▛█   Claude Code v2.1.0 (fake)\n▝▜██████▀  Sonnet · Claude Max\n\n'
printf '───────────────────────────────────────────────\n❯ Try "how does <filepath> work?"\n───────────────────────────────────────────────\n  ⏵⏵ bypass permissions on (shift+tab to cycle)\n'
while IFS= read -r line; do printf '\n● got: %s\n❯ \n' "$line"; done
'''


class Box:
    def __init__(self, home: Path, sess: str):
        self.home, self.sess = home, sess
        self.ctl, self.ws, self.bin = home / "ctl", home / "ws", home / "bin"
        self.jobs = self.ctl / ".mam" / "jobs"
        self.yaml = self.ctl / ".mam" / "agent-sessions.yaml"

    # -- environment -------------------------------------------------------
    def env(self, **over) -> dict:
        env = {k: v for k, v in os.environ.items()
               if not k.startswith("HERDR_") and k not in ("MAM_JOB_ID", "MAM_VERSION", "CLAUDE_PROJECT_DIR",
                                                           "HOME_DIR", "XDG_CONFIG_HOME")}
        env.update(HOME=str(self.home), PATH=f"{self.bin}:{env.get('PATH', '')}", HERDR_SESSION_NAME=self.sess,
                   WORKSPACE_ROOT=str(self.ctl), AGENT_SESSIONS_YAML=str(self.yaml), MAM_JOBS_DIR=str(self.jobs),
                   DELEGATE_JOB_LOGS_DIR=str(self.ctl / ".mam" / "delegate_job_logs"),
                   MAM_ENV_FILE=str(self.ctl / ".mam.env.absent"), MAM_NO_SUBSCRIBER_DAEMON="1",
                   DELEGATE_JOB_PYTHON=sys.executable, AGENT_PYTHON_BIN=sys.executable,
                   MAM_ONBOARD_ACK_S="60")
        env.update({k: str(v) for k, v in over.items()})
        return env

    def sh(self, script: str, timeout=120, **over) -> subprocess.CompletedProcess:
        """A bash snippet after `source lib.sh` (so `herdr` is the shim)."""
        return subprocess.run(["bash", "-c", f'source "{SKILLS}/lib.sh"\n{script}'], capture_output=True, text=True,
                              cwd=str(self.ctl), env=self.env(**over), timeout=timeout)

    def herdr(self, *args, timeout=30) -> subprocess.CompletedProcess:
        """The real binary against this box's server (no shim)."""
        env = {"HOME": str(self.home), "PATH": os.environ.get("PATH", "")}
        return subprocess.run([HERDR, "--session", self.sess, *args], capture_output=True, text=True, env=env,
                              timeout=timeout)

    def agent_status(self, name=SESSION) -> str:
        res = self.herdr("agent", "get", name)
        try:
            return json.loads(res.stdout)["result"]["agent"].get("agent_status", "")
        except (ValueError, KeyError):
            return ""

    def pane_text(self, name=SESSION) -> str:
        res = self.herdr("agent", "read", name, "--source", "recent")
        try:
            d = json.loads(res.stdout)["result"]
            return (d.get("read") or d).get("text", "")
        except (ValueError, KeyError, AttributeError):
            return res.stdout

    def server_pids(self) -> list:
        out = subprocess.run(["pgrep", "-f", f"herdr --session {self.sess} server"], capture_output=True, text=True).stdout
        return [int(p) for p in out.split()]

    # -- registry --------------------------------------------------------------
    def register(self, session=f"herdr:{SESSION}", **kw) -> str:
        kw.setdefault("ack_timeout_sec", 60)
        return registry.register_job(prompt="prior work", agent="claude-code", agent_session=session,
                                     registry_dir=str(self.jobs), **kw)

    def record(self, jid: str) -> dict:
        return json.loads((self.jobs / f"{jid}.json").read_text())

    def events(self, jid: str) -> list:
        path = self.jobs / f"{jid}.events.log"
        if not path.exists():
            return []
        return [json.loads(l) for l in path.read_text().splitlines() if l.strip()]

    def publish(self, jid: str, event: str, detail: str = "") -> None:
        rec = self.record(jid)
        cmd = [sys.executable, str(PUBLISH_PY), "--registry-dir", str(self.jobs), "--job", jid, "--event", event,
               "--expect-dispatch-token", rec["dispatch_token"], "--attempt-id", str(rec.get("attempt_id", 1))]
        if detail:
            cmd += ["--detail", detail]
        res = subprocess.run(cmd, capture_output=True, text=True, env=self.env(), timeout=120)
        assert res.returncode == 0, (event, res.stdout, res.stderr)

    def job_ids(self, session=f"herdr:{SESSION}") -> set:
        out = set()
        for p in self.jobs.glob("*.json"):
            try:
                if json.loads(p.read_text()).get("agent_session") == session:
                    out.add(p.stem)
            except ValueError:
                pass
        return out

    def wait_new_job(self, known: set, timeout=60) -> str:
        deadline = time.time() + timeout
        while time.time() < deadline:
            new = self.job_ids() - known
            if new:
                assert len(new) == 1, new
                return new.pop()
            time.sleep(0.2)
        raise AssertionError("no new job was registered")

    def wait_field(self, jid: str, key: str, timeout=60) -> str:
        deadline = time.time() + timeout
        while time.time() < deadline:
            if self.record(jid).get(key):
                return self.record(jid)[key]
            time.sleep(0.2)
        raise AssertionError(f"{jid} never got {key}")

    # -- the wrapper ---------------------------------------------------------------
    def submit(self, prompt="work", session=f"herdr:{SESSION}", timeout="300", ack="60", **over) -> subprocess.Popen:
        out = tempfile.TemporaryFile("w+", dir=str(self.home))
        err = tempfile.TemporaryFile("w+", dir=str(self.home))
        proc = subprocess.Popen(["bash", str(WRAPPER), "submit", "--agent", "claude-code", "--agent-session", session,
                                 "--role", "Worker", "--timeout", timeout, "--ack-timeout", ack,
                                 "--registry-dir", str(self.jobs), "--prompt", prompt],
                                stdout=out, stderr=err, text=True, cwd=str(self.ctl), env=self.env(**over))
        proc._out, proc._err = out, err  # type: ignore[attr-defined]
        return proc

    @staticmethod
    def finish(proc: subprocess.Popen, timeout=240):
        """The wrapper's (rc, stdout, stderr), after every timer-driven publish
        has returned: the wrapper ends as soon as the subscriber sees
        `completed` on the wire, while publish_event's registry write can
        land a moment later."""
        rc = proc.wait(timeout=timeout)
        for t in list(_TIMERS):
            t.join(timeout=60)
        proc._out.seek(0); proc._err.seek(0)  # type: ignore[attr-defined]
        return rc, proc._out.read(), proc._err.read()  # type: ignore[attr-defined]


def _row(sess: str, name=SESSION, ws="") -> str:
    return (f"- name: {name}\n  status: running\n  role: creator\n  herdr_session: {sess}\n  herdr_server: {sess}\n"
            f"  herdr_workspace: t50\n  delegate_job_id: null\n  pane: {{index: 0, pid: 0, cmd: claude, cwd: {ws}}}\n")


def _make_box(sess: str) -> Box:
    home = Path(tempfile.mkdtemp(prefix="mam50-"))  # short: a unix socket path lives under it
    box = Box(home, sess)
    for d in (box.bin, box.ws, box.ctl / ".agents", box.jobs, box.ctl / ".mam" / "delegate_job_logs"):
        d.mkdir(parents=True)
    (box.ctl / ".agents" / "skills").symlink_to(SKILLS)
    # a fresh HOME without ~/.zshrc puts zsh's first-run menu in every new
    # pane: the first `agent start` would type into it and time out (30 s)
    (home / ".zshrc").touch()
    (box.bin / "claude").write_text(FAKE_CLAUDE)
    (box.bin / "claude").chmod(0o755)
    box.yaml.write_text("herdr_sessions:\n" + _row(sess, ws=str(box.ws)))
    return box


def _start_pane(box: Box, name=SESSION, cwd=None) -> None:
    res = box.sh(f'herdr new-session -d -s {name} -c "{cwd or box.ws}" "claude --dangerously-skip-permissions"')
    assert res.returncode == 0, res.stdout + res.stderr
    deadline = time.time() + 30
    while time.time() < deadline and box.agent_status(name) != "idle":
        time.sleep(0.5)
    assert box.agent_status(name) == "idle", box.herdr("agent", "list").stdout


def _kill_servers(box: Box) -> None:
    for pid in box.server_pids():
        try:
            os.kill(pid, signal.SIGKILL)
        except OSError:
            pass


def _teardown(box: Box) -> None:
    if os.environ.get("MAM_T50_KEEP_BOX"):   # debugging aid: leave the box and the server in place
        print("KEEP_BOX", box.home, box.sess)
        return
    for sub in (("session", "stop", box.sess), ("session", "delete", box.sess)):
        try:
            subprocess.run([HERDR, *sub], capture_output=True, text=True,
                           env={"HOME": str(box.home), "PATH": os.environ["PATH"]}, timeout=30)
        except subprocess.TimeoutExpired:
            pass                             # the kill below is what must run (review F2)
        time.sleep(0.5)
    _kill_servers(box)
    shutil.rmtree(box.home, ignore_errors=True)
    # herdr writes its agent-detection state under $HOME/.local/state a moment
    # after the server is gone, and a late client call can bootstrap a new
    # server: sweep and kill once more, then prove the server is gone
    time.sleep(1)
    _kill_servers(box)
    shutil.rmtree(box.home, ignore_errors=True)
    deadline = time.time() + 5
    while box.server_pids() and time.time() < deadline:
        time.sleep(0.2)
    assert not box.server_pids(), f"a server of {box.sess} survived teardown"
    assert not box.home.exists(), f"{box.home} survived teardown"


@pytest.fixture
def box(local_mqtt_broker):
    _TIMERS.clear()
    b = _make_box(f"t50-{uuid.uuid4().hex[:8]}")
    try:
        _start_pane(b)
        yield b
    finally:
        _teardown(b)


_TIMERS: list = []


def _after(delay: float, fn, *args):
    t = threading.Timer(delay, fn, args)
    t.daemon = True
    t.start()
    _TIMERS.append(t)
    return t


def _ts(ev: dict) -> str:
    return ev["timestamp"]


# ---------------------------------------------------------------------------
# C1 - an open prior job ends, then the new one is injected
# ---------------------------------------------------------------------------

def test_c1_a_job_waits_for_the_sessions_open_job_and_then_runs(box):
    a = box.register()
    box.publish(a, "started")                       # running, one event on disk
    known = box.job_ids()
    proc = box.submit(MAM_DISPATCH_WAIT_S="20")
    b = box.wait_new_job(known)
    _after(3, box.publish, a, "completed", "prior done")
    _after(5, box.publish, b, "started")
    _after(8, box.publish, b, "completed", "b done")
    rc, out, err = box.finish(proc)
    assert rc == 0, out + err
    assert box.record(b)["status"] == "completed" and box.record(b).get("detail") is None
    assert "cancelled" not in out and "stale open job" not in err
    assert box.record(b).get("dispatch_wait_at"), "B waited (and signalled) while A was open"
    assert _ts(box.events(b)[0]) >= _ts(box.events(a)[-1]), "B started only after A ended"
    assert "got: Your job_id is \"" + b in box.pane_text().replace("\n", "") or b in box.pane_text()


# ---------------------------------------------------------------------------
# C2 - the bound: an open job that never ends cancels the new one as session_busy
# ---------------------------------------------------------------------------

def test_c2_the_wait_bound_closes_the_job_as_session_busy_naming_the_open_one(box):
    a = box.register()
    box.publish(a, "started")
    known = box.job_ids()
    t0 = time.time()
    proc = box.submit(MAM_DISPATCH_WAIT_S="3")
    b = box.wait_new_job(known)
    rc, out, err = box.finish(proc)
    assert rc == 5, out + err
    assert time.time() - t0 < 60
    rec = box.record(b)
    assert rec["status"] == "cancelled" and rec.get("detail") == "session_busy", rec
    assert a in err and "still has open jobs in the registry" in err, err
    assert [e["event"] for e in box.events(b)] == ["cancelled"]
    assert box.record(a)["status"] == "running", "the open job is left alone"


# ---------------------------------------------------------------------------
# C3 - a running record nothing has touched for idle_timeout_sec is stale
# ---------------------------------------------------------------------------

def test_c3_a_dead_running_record_does_not_block_the_session(box):
    a = box.register(idle_timeout_sec=2)
    box.publish(a, "started")
    time.sleep(3)                                   # its only event is now older than idle_timeout_sec
    known = box.job_ids()
    proc = box.submit(MAM_DISPATCH_WAIT_S="30")
    b = box.wait_new_job(known)
    _after(2, box.publish, b, "started")
    _after(4, box.publish, b, "completed", "b done")
    rc, out, err = box.finish(proc)
    assert rc == 0, out + err
    assert f"stale open job {a} ignored (running, no event for 2s)" in err, err
    assert box.record(b)["status"] == "completed"
    rec = box.record(b)   # nothing to wait for: the only signal is the admission's own
    assert rec["dispatch_admitted_token"] == rec["dispatch_token"] and rec["dispatch_wait_at"] == rec["dispatch_admitted_at"]


# ---------------------------------------------------------------------------
# C4 - the loop's first job to a creator whose onboarding is still running
# ---------------------------------------------------------------------------

def test_c4_the_loops_first_job_waits_for_the_onboarding_to_end(box, monkeypatch):
    onboarding = box.register(ack_timeout_sec=180)  # create registers it exactly so: agent_session=herdr:<session>
    box.publish(onboarding, "started")
    for k, v in box.env(MAM_DISPATCH_WAIT_S="30").items():
        monkeypatch.setenv(k, v)
    monkeypatch.delenv("MAM_REAL_ROOT", raising=False)
    from lib_py.loop import runner as lr
    r = lr.LoopRunner(["--creator", SESSION, "--task", "first task"])
    known = box.job_ids()
    result = {}

    def drive():
        first = box.wait_new_job(known)
        result["first"] = first
        time.sleep(5)
        box.publish(onboarding, "completed", "Onboarding complete; aligned with role creator")
        time.sleep(2)
        box.publish(first, "started")
        time.sleep(2)
        box.publish(first, "completed", "first done")

    t = threading.Thread(target=drive, daemon=True)
    t.start()
    jid = r.delegate_job(SESSION, "claude-code", "Worker", "the loop's first job")  # the runner adds herdr:
    t.join(timeout=60)
    assert jid == result.get("first"), (jid, result)
    assert jid not in r.failed_submit_jobs
    rec = box.record(jid)
    assert rec["status"] == "completed" and rec.get("detail") is None, rec
    assert _ts(box.events(jid)[0]) >= _ts(box.events(onboarding)[-1])


# ---------------------------------------------------------------------------
# C5 - a session create never registered: the record's agent_session alone decides
# ---------------------------------------------------------------------------

def test_c5_a_session_without_a_row_is_judged_by_the_records_alone(local_mqtt_broker):
    # resolve_herdr_session falls back to the workspace slug when the yaml has no
    # row, so the server is named exactly that: mam-<parent>-<work> of the control root
    home = Path(tempfile.mkdtemp(prefix="mam50-"))
    sess = f"mam-{home.name}-ctl".lower().replace("_", "-")
    shutil.rmtree(home)
    _TIMERS.clear()
    b = Box(home, sess)
    for d in (b.bin, b.ws, b.ctl / ".agents", b.jobs, b.ctl / ".mam" / "delegate_job_logs"):
        d.mkdir(parents=True)
    (b.ctl / ".agents" / "skills").symlink_to(SKILLS)
    (b.bin / "claude").write_text(FAKE_CLAUDE)
    (b.bin / "claude").chmod(0o755)
    b.yaml.write_text("herdr_sessions: []\n")
    try:
        _start_pane(b)
        a = b.register()
        b.publish(a, "started")
        known = b.job_ids()
        proc = b.submit(MAM_DISPATCH_WAIT_S="20")
        c = b.wait_new_job(known)
        _after(3, b.publish, a, "completed", "prior done")
        _after(5, b.publish, c, "started")
        _after(7, b.publish, c, "completed", "c done")
        rc, out, err = b.finish(proc)
        assert rc == 0, out + err
        assert b.record(c)["status"] == "completed" and b.record(c).get("dispatch_wait_at")
        assert _ts(b.events(c)[0]) >= _ts(b.events(a)[-1])
    finally:
        _teardown(b)


# ---------------------------------------------------------------------------
# C7 - two waiters behind one session: FIFO, one at a time, no cancellation
# ---------------------------------------------------------------------------

def test_c7_two_waiters_run_one_after_the_other_and_neither_is_cancelled(box):
    a = box.register()
    box.publish(a, "started")
    known = box.job_ids()
    proc_b = box.submit(prompt="work B", MAM_DISPATCH_WAIT_S="40")
    b = box.wait_new_job(known)
    time.sleep(1.2)
    proc_c = box.submit(prompt="work C", MAM_DISPATCH_WAIT_S="40")
    c = box.wait_new_job(known | {b})
    assert box.record(b)["created_at"] <= box.record(c)["created_at"]
    box.wait_field(c, "dispatch_wait_at")           # both are waiting (C signals too)
    box.wait_field(b, "dispatch_wait_at")
    _after(2, box.publish, a, "completed", "prior done")
    _after(4, box.publish, b, "started")
    _after(7, box.publish, b, "completed", "b done")
    # C is published started only once B has ended - as the real agent, which
    # cannot answer a prompt it has not been given, would
    _after(9, box.publish, c, "started")
    _after(11, box.publish, c, "completed", "c done")
    rc_b, out_b, err_b = box.finish(proc_b)
    rc_c, out_c, err_c = box.finish(proc_c)
    assert (rc_b, rc_c) == (0, 0), out_b + err_b + out_c + err_c
    assert box.record(b)["status"] == box.record(c)["status"] == "completed"
    assert "cancelled" not in out_b + out_c and "stale open job" not in err_b + err_c
    assert _ts(box.events(b)[0]) >= _ts(box.events(a)[-1])
    assert _ts(box.events(c)[0]) >= _ts(box.events(b)[-1]), "C started only after B ended"
    # the admission (review aa1bff92 F1): B's before its own `started`, C's only once B ended
    assert box.record(b)["dispatch_admitted_at"] <= _ts(box.events(b)[0])
    assert box.record(c)["dispatch_admitted_at"] >= _ts(box.events(b)[-1])
    pane = box.pane_text()
    assert pane.index(f'Your job_id is "{b}"') < pane.index(f'Your job_id is "{c}"'), "injected in FIFO order"


# ---------------------------------------------------------------------------
# C8 - a waiter whose wrapper died: its signal goes stale and is skipped
# ---------------------------------------------------------------------------

def test_c8_a_dead_waiter_is_skipped_once_its_signal_is_stale(box):
    a = box.register()
    box.publish(a, "started")
    known = box.job_ids()
    proc_b = box.submit(prompt="work B", ack="5", MAM_DISPATCH_WAIT_S="120")
    b = box.wait_new_job(known)
    box.wait_field(b, "dispatch_wait_at")
    os.kill(proc_b.pid, signal.SIGKILL)             # the wrapper is gone; its waiter sees that and stops
    proc_b.wait(timeout=30)
    time.sleep(1.2)                                 # C's created_at is a later second: B stays ahead in the FIFO
    proc_c = box.submit(prompt="work C", MAM_DISPATCH_WAIT_S="60")
    c = box.wait_new_job(known | {b})
    assert box.record(b)["created_at"] < box.record(c)["created_at"]
    _after(2, box.publish, a, "completed", "prior done")
    _after(40, box.publish, c, "started")           # after B's signal window (30 s) has passed
    _after(42, box.publish, c, "completed", "c done")
    rc, out, err = box.finish(proc_c)
    assert rc == 0, out + err
    assert f"stale open job {b} ignored (pending, no dispatch signal for 30s)" in err, err
    assert box.record(c)["status"] == "completed", (box.record(c), box.events(c), out, err)
    assert box.record(b)["status"] == "pending", "the dead waiter's record is left as it is"
    assert _ts(box.events(c)[0]) >= _ts(box.events(a)[-1])


# ---------------------------------------------------------------------------
# C9 - C8's live counterpart (review aa1bff92 F2): waiters with short ack
# windows, held across many of them, are never stale and never co-admitted
# ---------------------------------------------------------------------------

def test_c9_live_waiters_with_short_ack_windows_are_never_stale_nor_co_admitted(box):
    a = box.register()
    box.publish(a, "started")
    known = box.job_ids()
    proc_b = box.submit(prompt="work B", ack="5", MAM_DISPATCH_WAIT_S="60")
    b = box.wait_new_job(known)
    time.sleep(1.2)
    proc_c = box.submit(prompt="work C", ack="5", MAM_DISPATCH_WAIT_S="60")
    c = box.wait_new_job(known | {b})
    box.wait_field(c, "dispatch_wait_at")
    box.wait_field(b, "dispatch_wait_at")
    _after(12, box.publish, a, "completed", "prior done")   # both waited through two ack windows
    _after(16, box.publish, b, "started")                   # B is injected at ~13 s
    _after(19, box.publish, b, "completed", "b done")
    _after(23, box.publish, c, "started")                   # C at ~20 s
    _after(25, box.publish, c, "completed", "c done")
    rc_b, out_b, err_b = box.finish(proc_b)
    rc_c, out_c, err_c = box.finish(proc_c)
    assert (rc_b, rc_c) == (0, 0), out_b + err_b + out_c + err_c
    assert "stale open job" not in err_b + err_c, err_b + err_c
    assert "cancelled" not in out_b + out_c
    assert box.record(b)["status"] == box.record(c)["status"] == "completed"
    assert box.record(c)["dispatch_admitted_at"] >= _ts(box.events(b)[-1]), "C admitted only once B ended"
    assert _ts(box.events(c)[0]) >= _ts(box.events(b)[-1])
    pane = box.pane_text()
    assert pane.index(f'Your job_id is "{b}"') < pane.index(f'Your job_id is "{c}"'), "injected in FIFO order"


# ---------------------------------------------------------------------------
# C10 - a pending onboarding (registered and injected with no waiter, as
# create --onboard does) holds the session for its ack budget, not for the
# 30 s signal window (review 55219be1 R2); C11 is its expired control
# ---------------------------------------------------------------------------

def _onboarding(box: Box, ack: int) -> str:
    a = box.register(ack_timeout_sec=ack)
    prompt = box.herdr("agent", "prompt", SESSION, "Onboarding " + a)
    assert prompt.returncode == 0, prompt.stdout + prompt.stderr
    return a


def test_c10_a_pending_onboarding_within_its_ack_budget_blocks_the_session_past_the_signal_window(box):
    t0 = time.time()
    a = _onboarding(box, 180)
    known = box.job_ids()
    for label in ("before", "after"):
        if label == "after":
            time.sleep(max(0.0, t0 + 31.5 - time.time()))       # A is now older than the signal window
        proc = box.submit(prompt="work " + label, MAM_DISPATCH_WAIT_S="3")
        b = box.wait_new_job(known)
        known.add(b)
        rc, out, err = box.finish(proc)
        assert rc == 5, (label, out + err)
        rec = box.record(b)
        assert rec["status"] == "cancelled" and rec.get("detail") == "session_busy", (label, rec)
        assert a in err and "still has open jobs in the registry" in err and "stale open job" not in err, (label, err)
        assert b not in box.pane_text(), label
        assert box.record(a)["status"] == "pending" and "dispatch_wait_at" not in box.record(a), label
    created = calendar.timegm(time.strptime(box.record(a)["created_at"], "%Y-%m-%dT%H:%M:%SZ"))
    assert 31 <= time.time() - created < 180, "the second answer came past the window, within the budget"


def test_c11_an_onboarding_never_acknowledged_within_its_ack_budget_is_stale(box):
    a = _onboarding(box, 2)
    time.sleep(3.2)
    known = box.job_ids()
    proc = box.submit(MAM_DISPATCH_WAIT_S="30")
    b = box.wait_new_job(known)
    _after(2, box.publish, b, "started")
    _after(4, box.publish, b, "completed", "b done")
    rc, out, err = box.finish(proc)
    assert rc == 0, out + err
    assert f"stale open job {a} ignored (pending, no dispatch signal and no started within 2s)" in err, err
    assert box.record(b)["status"] == "completed" and box.record(a)["status"] == "pending"
    assert b in box.pane_text()


# ---------------------------------------------------------------------------
# C12 - a waiting job redispatched by someone else (new attempt) is left to
# that attempt: the old wrapper exits 5 without a close (review 55219be1 R1)
# ---------------------------------------------------------------------------

def test_c12_a_waiting_job_redispatched_by_someone_else_is_left_to_its_new_attempt(box):
    a = box.register()
    box.publish(a, "started")
    known = box.job_ids()
    proc = box.submit(MAM_DISPATCH_WAIT_S="60")
    b = box.wait_new_job(known)
    box.wait_field(b, "dispatch_wait_at")
    old = box.record(b)["dispatch_token"]
    res = subprocess.run([sys.executable, str(REGISTRY_PY), "--registry-dir", str(box.jobs), "update", "--job", b,
                          "--status", "pending", "--attempt-id", "2"], capture_output=True, text=True, env=box.env(),
                         timeout=60)
    assert res.returncode == 0, res.stdout + res.stderr
    moved = box.record(b)
    assert moved["dispatch_token"] != old and moved["attempt_id"] == 2
    box.publish(a, "completed", "prior done")
    rc, out, err = box.finish(proc)
    assert rc == 5, out + err
    assert f"job {b} was redispatched, moved or closed by someone else" in err, err
    assert box.record(b) == moved, "not closed, not admitted, not renewed: attempt 2's dispatcher owns it"
    assert box.events(b) == [] and b not in box.pane_text()
    assert out.strip().splitlines()[-1].endswith(f"/delegate_job_logs/{b}"), out


# ---------------------------------------------------------------------------
# A - three `create --onboard` at once in one server
# ---------------------------------------------------------------------------

def _drive_onboardings(box: Box, names: list, delays: list, result: dict):
    seen = {}
    deadline = time.time() + 120
    while len(seen) < len(names) and time.time() < deadline:
        for p in box.jobs.glob("*.json"):
            try:
                rec = json.loads(p.read_text())
            except ValueError:
                continue
            sess = rec.get("agent_session", "")
            if sess.startswith("herdr:") and sess[6:] in names and sess[6:] not in seen:
                seen[sess[6:]] = p.stem
                box.publish(p.stem, "started")
                result.setdefault("started_at", {})[sess[6:]] = time.time()
        time.sleep(0.3)
    result["jobs"] = dict(seen)
    for name, delay in zip(names, delays):
        jid = seen.get(name)
        if jid:
            _after(delay, box.publish, jid, "completed", "Onboarding complete; aligned with role creator")


def _hold_pane_creation(b: Box, seconds: int) -> None:
    """The real herdr on the box's PATH, with the reply of a pane-creating call
    (workspace create, pane split) held back `seconds`. The shim's window -
    a pane exists idle, `agent start` has not claimed it yet - is the one the
    orchestrator hit; this makes it long enough for a shim started a moment
    later to walk into it on every run, not one run in three."""
    (b.bin / "herdr").write_text(f'''#!/usr/bin/env bash
case " $* " in
  *" workspace create "*|*" pane split "*)
    out=$("{HERDR}" "$@"); rc=$?
    sleep {seconds}
    printf '%s\\n' "$out"; exit $rc ;;
esac
exec "{HERDR}" "$@"
''')
    (b.bin / "herdr").chmod(0o755)


def _three_creates(b: Box, names: list, workspaces: list, stagger: float = 0.0):
    """create --onboard x3 at once in one server; (outcome by name, driver result, elapsed)."""
    procs = []
    for name, ws in zip(names, workspaces):
        if procs and stagger:
            time.sleep(stagger)
        out = tempfile.TemporaryFile("w+", dir=str(b.home))
        err = tempfile.TemporaryFile("w+", dir=str(b.home))
        procs.append((name, subprocess.Popen(
            ["bash", str(CREATE_SH), "--workspace", str(ws), "--agent", "claude", "--role", "creator",
             "--session", name, "--onboard", "--herdr-session", b.sess],
            stdout=out, stderr=err, text=True, cwd=str(b.ctl), env=b.env(MAM_ONBOARD_WAIT_S="120")), out, err))
    t0 = time.time()
    result = {}
    drv = threading.Thread(target=_drive_onboardings, args=(b, names, [10, 15, 20], result), daemon=True)
    drv.start()
    outcome = {}
    for name, proc, out, err in procs:
        rc = proc.wait(timeout=240)
        out.seek(0); err.seek(0)
        outcome[name] = (rc, out.read(), err.read())
    elapsed = time.time() - t0
    drv.join(timeout=5)
    return outcome, result, elapsed


def _agent_panes(b: Box) -> dict:
    """agent name -> (pane_id, pane cwd) from the server."""
    panes = {p["pane_id"]: p for p in json.loads(b.herdr("pane", "list").stdout)["result"]["panes"]}
    out = {}
    for a in json.loads(b.herdr("agent", "list").stdout)["result"]["agents"]:
        pid = a.get("pane_id") or (a.get("pane") or {}).get("pane_id")
        out[a["name"]] = (pid, (panes.get(pid) or {}).get("cwd"))
    return out


def test_a_three_concurrent_creates_with_onboarding(local_mqtt_broker):
    b = _make_box(f"t50-{uuid.uuid4().hex[:8]}")
    b.yaml.write_text("herdr_sessions: []\n")
    names = [f"t50-{i}-creator-claude" for i in (1, 2, 3)]
    try:
        workspaces = []
        for i in (1, 2, 3):
            (b.home / f"ws{i}").mkdir()
            workspaces.append(b.home / f"ws{i}")
        outcome, result, elapsed = _three_creates(b, names, workspaces)
        t0 = time.time() - elapsed
        assert all(o[0] == 0 for o in outcome.values()), {n: (o[0], o[2][-1500:]) for n, o in outcome.items()}
        # The three start-ups overlap, and the onboarding waits do too: once the
        # last prompt was delivered (its `started`), the whole run ends with the
        # slowest onboarding (20 s) plus create's wrap-up, not with their sum (45 s).
        last_started = max(result["started_at"].values()) - t0
        assert elapsed - last_started <= 20 + 15, (elapsed, last_started, result["started_at"])
        assert max(result["started_at"].values()) - min(result["started_at"].values()) <= 20, \
            ("the three creates did not run side by side", result["started_at"])
        for name in names:
            jid = result["jobs"][name]
            assert f"onboarding completed: job {jid} (" in outcome[name][1], outcome[name][1]
            assert b.record(jid)["status"] == "completed"
        rows = re.findall(r"^- name: (\S+)", b.yaml.read_text(), re.M)
        assert sorted(rows) == sorted(names), rows
        assert len(set(result["jobs"].values())) == 3 and all((b.jobs / f"{j}.json").exists() for j in result["jobs"].values())
        assert len(b.server_pids()) == 1, "three shims bootstrapped the server at once; one must remain"
        assert len(json.loads(b.herdr("agent", "list").stdout)["result"]["agents"]) == 3
    finally:
        _teardown(b)


def test_a3_three_concurrent_creates_in_one_folder_take_distinct_panes(local_mqtt_broker):
    """The same --workspace for all three - the control root itself, where
    several reviewers are made at once: no control-plane binding is forwarded
    to the pane, so the shim reuses an idle pane when it sees one (BUG-4). The
    pane choice and its `agent start` are serialised per server, so no two
    creates claim one pane (agent_pane_busy)."""
    b = _make_box(f"t50-{uuid.uuid4().hex[:8]}")
    b.yaml.write_text("herdr_sessions: []\n")
    _hold_pane_creation(b, 8)
    names = [f"t50-{i}-creator-claude" for i in (1, 2, 3)]
    try:
        assert b.sh("herdr agent list").returncode == 0      # the server is up: the creates race on panes only
        outcome, result, elapsed = _three_creates(b, names, [b.ctl] * 3, stagger=2.0)
        assert all(o[0] == 0 for o in outcome.values()), {n: (o[0], o[2][-1500:]) for n, o in outcome.items()}
        assert "agent_pane_busy" not in "".join(o[1] + o[2] for o in outcome.values())
        for name in names:
            jid = result["jobs"][name]
            assert f"onboarding completed: job {jid} (" in outcome[name][1], outcome[name][1]
            assert b.record(jid)["status"] == "completed"
        rows = re.findall(r"^- name: (\S+)", b.yaml.read_text(), re.M)
        assert sorted(rows) == sorted(names), rows
        panes = _agent_panes(b)
        assert sorted(panes) == sorted(names), panes
        assert len({p for p, _c in panes.values()}) == 3, ("two agents on one pane", panes)
        assert all(Path(c).resolve() == b.ctl.resolve() for _p, c in panes.values()), panes
        assert len(b.server_pids()) == 1
    finally:
        _teardown(b)


def test_s_three_concurrent_new_sessions_in_one_folder_take_distinct_panes():
    """The shim alone, no create around it: three `new-session` into one cwd on
    one running server at once. Before the pane claim was serialised, two of
    them chose the same idle pane and the later `agent start` failed with
    agent_pane_busy."""
    b = _make_box(f"t50-{uuid.uuid4().hex[:8]}")
    _hold_pane_creation(b, 2)
    names = [f"t50-{i}-creator-claude" for i in (1, 2, 3)]
    try:
        assert b.sh("herdr agent list").returncode == 0      # the server is up before the race
        procs = []
        for name in names:
            if procs:
                time.sleep(0.7)
            out = tempfile.TemporaryFile("w+", dir=str(b.home))
            err = tempfile.TemporaryFile("w+", dir=str(b.home))
            procs.append((name, subprocess.Popen(
                ["bash", "-c", f'source "{SKILLS}/lib.sh"\nherdr new-session -d -s {name} -c "{b.ws}" '
                               '"claude --dangerously-skip-permissions"'],
                stdout=out, stderr=err, text=True, cwd=str(b.ctl), env=b.env()), out, err))
        outcome = {}
        for name, proc, out, err in procs:
            rc = proc.wait(timeout=120)
            out.seek(0); err.seek(0)
            outcome[name] = (rc, out.read(), err.read())
        assert all(o[0] == 0 for o in outcome.values()), {n: (o[0], o[2][-800:]) for n, o in outcome.items()}
        panes = _agent_panes(b)
        assert sorted(panes) == sorted(names), panes
        assert len({p for p, _c in panes.values()}) == 3, ("two agents on one pane", panes)
        assert all(Path(c).resolve() == b.ws.resolve() for _p, c in panes.values()), panes
        for name in names:
            deadline = time.time() + 30
            while time.time() < deadline and b.agent_status(name) != "idle":
                time.sleep(0.5)
            assert b.agent_status(name) == "idle", (name, b.herdr("agent", "list").stdout)
    finally:
        _teardown(b)


def _git(repo: Path, *args):
    subprocess.run(["git", "-c", "user.email=t@example.com", "-c", "user.name=t", *args], cwd=str(repo), check=True,
                   capture_output=True, text=True)


def test_a2_three_concurrent_worktree_prepares_with_onboarding(local_mqtt_broker):
    """worktree_prepare.sh --onboard x3 (different --path, one server): the git
    worktree step, the registry and the server's first start do not contend."""
    prep = SKILLS / "multi-agent-mux-create" / "scripts" / "worktree_prepare.sh"
    b = _make_box(f"t50-{uuid.uuid4().hex[:8]}")
    b.yaml.write_text("herdr_sessions: []\n")
    _git(b.ctl, "init", "-q")
    (b.ctl / "README").write_text("ctl\n")
    _git(b.ctl, "add", "README")
    _git(b.ctl, "commit", "-q", "-m", "init")
    branches = [f"b{i}" for i in (1, 2, 3)]
    names = [f"creator-claude-{br}" for br in branches]
    procs = []
    try:
        for br in branches:
            out = tempfile.TemporaryFile("w+", dir=str(b.home))
            err = tempfile.TemporaryFile("w+", dir=str(b.home))
            procs.append((br, subprocess.Popen(
                ["bash", str(prep), "--path", str(b.home / f"wt-{br}"), "--branch", br, "--creator-agent", "claude",
                 "--onboard", "--herdr-session", b.sess],
                stdout=out, stderr=err, text=True, cwd=str(b.ctl), env=b.env(MAM_ONBOARD_WAIT_S="120")), out, err))
        t0 = time.time()
        result = {}
        drv = threading.Thread(target=_drive_onboardings, args=(b, names, [10, 15, 20], result), daemon=True)
        drv.start()
        outcome = {}
        for br, proc, out, err in procs:
            rc = proc.wait(timeout=300)
            out.seek(0); err.seek(0)
            outcome[br] = (rc, out.read(), err.read())
        elapsed = time.time() - t0
        drv.join(timeout=5)
        assert all(o[0] == 0 for o in outcome.values()), {n: (o[0], o[2][-1500:]) for n, o in outcome.items()}
        last_started = max(result["started_at"].values()) - t0
        assert elapsed - last_started <= 20 + 15, (elapsed, last_started, result["started_at"])
        for br in branches:
            assert f"WORKTREE={b.home / f'wt-{br}'}" in outcome[br][1] and f"SESSION=creator-claude-{br}" in outcome[br][1]
            assert (b.home / f"wt-{br}" / "README").exists()
            assert b.record(result["jobs"][f"creator-claude-{br}"])["status"] == "completed"
        listed = subprocess.run(["git", "worktree", "list"], cwd=str(b.ctl), capture_output=True, text=True).stdout
        assert all(f"[{br}]" in listed for br in branches), listed
        rows = re.findall(r"^- name: (\S+)", b.yaml.read_text(), re.M)
        assert sorted(rows) == sorted(names), rows
        assert len(b.server_pids()) == 1
    finally:
        _teardown(b)