"""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 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"
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)
    (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"
    assert "dispatch_wait_at" not in box.record(b), "nothing to wait for: no signal written"


# ---------------------------------------------------------------------------
# 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"
    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(14, box.publish, c, "started")           # after B's 5 s ack window has long passed
    _after(16, 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 5s)" 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])


# ---------------------------------------------------------------------------
# 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 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)]
    procs = []
    try:
        for i, name in enumerate(names, 1):
            ws = b.home / f"ws{i}"
            ws.mkdir()
            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)
        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 _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)