feat(issue-50): the delegate-job wrapper waits by the registry before it injects (C)

Reproduction first (herdr 0.9.3, a real claude, one-off server, job
51ccd800): after a job that left a background subagent running
(`← 1 agent` on screen for 5 min), herdr reported `working` only while
the job ran and `idle` ~10 s after `completed`; a second submit with a
20 s idle wait was injected, not cancelled. The 2026-10-04 observation
(0.9.1, 10+ min of `working`) does not reproduce on 0.9.3, so the herdr
idle wait stays and the registry wait runs before it (plan §4.3 (나)).

- lib_py/delegation/wrapper.py: `open-jobs` (diagnostic) and
  `wait-open-jobs`. An open job of the session is live when the #18
  contract says so: `running` with an event in events.log within
  idle_timeout_sec; `pending` with a dispatch signal (dispatch_wait_at,
  else created_at) within ack_timeout_sec. A dead record is reported
  (`stale open job <id> ignored (...)`) and never waited for. The waiter
  signals every 10 s by merging dispatch_wait_at into its own pending
  record through the registry's transition boundary (status untouched, no
  event), so later waiters see it alive; a waiter whose wrapper is gone
  stops and its signal goes stale (Creator's challenge, job bf8863e1).
  Release is FIFO by (created_at, job_id): `clear` when no live running
  job remains and this job is the earliest live pending one; `busy <ids>`
  at the bound min(MAM_DISPATCH_WAIT_S, remaining wall clock).
- the wrapper: _registry_wait after the preflight in both dispatch paths;
  `busy` closes the job as cancelled/session_busy (phase registry_wait),
  exit 5 - the existing contract.
- tests/test_issue50_dispatch.py: the open-job table (status x age x
  signal x spelling), FIFO and the tie, the signal as a registry merge,
  the bound, a dead wrapper, finalize-end's new phase, both call sites.
- tests/test_issue50_dispatch_herdr.py: a real herdr 0.9.3 server per
  test (short mkdtemp HOME, fake claude the server reports idle, the real
  publish path over the suite's broker): C1 wait-then-run, C2 the bound,
  C3 a dead running record, C4 the loop runner's first job behind an
  onboarding, C5 a session without a row, C7 two waiters in FIFO order
  with no cancellation, C8 a killed waiter skipped once stale; and three
  concurrent `create --onboard` / `worktree_prepare.sh --onboard` in one
  server (one server process survives the shims' concurrent bootstrap,
  so no bootstrap lock is needed).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

tests/test_issue50_dispatch.py

@@ -0,0 +1,273 @@
+"""Issue #50 C: the delegate-job wrapper waits by the registry before it injects.
+
+- U: the decisions in lib_py.delegation.wrapper - the open-job table (status x
+  signal age x session spelling), FIFO release among live pending waiters,
+  the waiter's own signal written through the registry, the bound, a dead
+  wrapper, and finalize-end's `registry_wait` phase closing as session_busy.
+- H: against a real herdr 0.9.3 server (one-off session, fake claude in the
+  pane, the real publish path through publish_event.py and the suite's local
+  broker): see test_issue50_dispatch_herdr.py.
+"""
+from __future__ import annotations
+
+import argparse
+import calendar
+import json
+import os
+import subprocess
+import sys
+import time
+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"
+if str(SKILLS) not in sys.path:
+    sys.path.insert(0, str(SKILLS))
+
+from lib_py.delegation import mqtt_common, registry, wrapper as decide  # noqa: E402
+
+
+def _iso(epoch):
+    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(epoch))
+
+
+def _register(reg, session="herdr:s", ack=180, idle=120, **over):
+    jid = registry.register_job(prompt="p", agent="claude-code", agent_session=session, registry_dir=str(reg),
+                                ack_timeout_sec=ack, idle_timeout_sec=idle)
+    if over:
+        _edit(reg, jid, **over)
+    return jid
+
+
+def _edit(reg, jid, **fields):
+    path = reg / f"{jid}.json"
+    rec = json.loads(path.read_text())
+    rec.update(fields)
+    path.write_text(json.dumps(rec))
+
+
+def _event(reg, jid, epoch, event="progress"):
+    with open(reg / f"{jid}.events.log", "a") as fh:
+        fh.write(json.dumps({"job_id": jid, "event": event, "timestamp": _iso(epoch)}) + "\n")
+
+
+def _ns(reg, job, session="herdr:s", wait_s=0, start_time=None, timeout=3600, watch_pid=None):
+    return argparse.Namespace(job=job, agent_session=session, registry_dir=str(reg), wait_s=wait_s,
+                              start_time=start_time, timeout=timeout, watch_pid=watch_pid)
+
+
+# ---------------------------------------------------------------------------
+# U1 - the open-job table
+# ---------------------------------------------------------------------------
+
+@pytest.mark.parametrize("status,age,signal,want", [
+    ("running", 10, None, "live"),          # an event idle_timeout_sec (120) ago at most
+    ("running", 121, None, "stale"),
+    ("running", None, None, "stale"),       # running with no events.log at all
+    ("pending", 10, None, "live"),          # created within ack_timeout_sec (180)
+    ("pending", 181, None, "stale"),
+    ("pending", 181, 5, "live"),            # old record, fresh dispatch_wait_at
+    ("pending", 181, 181, "stale"),
+    ("pending", 10, 181, "stale"),          # the signal, once present, is what counts
+    ("completed", 10, None, None),
+    ("error", 10, None, None),
+    ("cancelled", 10, None, None),
+])
+def test_u1_open_jobs_table(tmp_path, status, age, signal, want):
+    NOW = int(time.time())   # the running check reads the real clock (recent_disk_heartbeat)
+    other = _register(tmp_path, status=status)
+    if status == "running" and age is not None:
+        _event(tmp_path, other, NOW - age)
+    if status == "pending":
+        _edit(tmp_path, other, created_at=_iso(NOW - age))
+        if signal is not None:
+            _edit(tmp_path, other, dispatch_wait_at=_iso(NOW - signal))
+    me = _register(tmp_path)
+    live, stale = decide.open_jobs(str(tmp_path), "herdr:s", me, now=NOW)
+    got = {j for _c, j, _s in live}, {j for j, _r in stale}
+    if want == "live":
+        assert got == ({other}, set())
+    elif want == "stale":
+        assert got == (set(), {other}), got
+        assert stale[0][1].startswith(status + ", no ")
+    else:
+        assert got == (set(), set())
+
+
+@pytest.mark.parametrize("record_session,asked", [("herdr:s", "s"), ("s", "herdr:s"), ("herdr:s", "herdr:s")])
+def test_u1b_the_session_spelling_is_normalised(tmp_path, record_session, asked):
+    other = _register(tmp_path, session=record_session)
+    me = _register(tmp_path, session=asked)
+    live, _ = decide.open_jobs(str(tmp_path), asked, me)
+    assert [j for _c, j, _s in live] == [other]
+    live, _ = decide.open_jobs(str(tmp_path), "herdr:another", me)
+    assert live == []
+
+
+def test_u1c_the_job_itself_and_unreadable_records_are_skipped(tmp_path):
+    me = _register(tmp_path)
+    (tmp_path / "broken1.json").write_text("{not json")
+    assert decide.open_jobs(str(tmp_path), "herdr:s", me) == ([], [])
+
+
+# ---------------------------------------------------------------------------
+# U2 - FIFO release and the bound (wait_s=0: an immediate decision)
+# ---------------------------------------------------------------------------
+
+def test_u2_clear_only_when_no_live_running_and_this_job_is_the_earliest_pending(tmp_path, capsys):
+    a = _register(tmp_path, status="running")
+    _event(tmp_path, a, time.time())
+    me = _register(tmp_path)
+    assert decide.wait_open_jobs(_ns(tmp_path, me)) == f"busy {a}"
+    # A ends: clear
+    _edit(tmp_path, a, status="completed")
+    assert decide.wait_open_jobs(_ns(tmp_path, me)) == "clear"
+    # an earlier live pending waiter B: this job waits behind it (busy at the bound)
+    b = _register(tmp_path, created_at=_iso(time.time() - 30))
+    assert decide.wait_open_jobs(_ns(tmp_path, me)) == f"busy {b}"
+    # a later one does not hold this job up
+    _edit(tmp_path, b, created_at=_iso(time.time() + 30))
+    assert decide.wait_open_jobs(_ns(tmp_path, me)) == "clear"
+
+
+def test_u2b_a_tie_on_created_at_is_broken_by_job_id(tmp_path):
+    same = _iso(time.time())
+    first = _register(tmp_path, created_at=same, job_id="aaaa0001")
+    second = _register(tmp_path, created_at=same, job_id="aaaa0002")
+    # records carry their job_id inside; the file name is what open_jobs keys on
+    for jid, real in ((first, "aaaa0001"), (second, "aaaa0002")):
+        os.rename(tmp_path / f"{jid}.json", tmp_path / f"{real}.json")
+    assert decide.wait_open_jobs(_ns(tmp_path, "aaaa0001")) == "clear"
+    assert decide.wait_open_jobs(_ns(tmp_path, "aaaa0002")) == "busy aaaa0001"
+
+
+def test_u2c_a_dead_waiter_is_ignored_and_reported(tmp_path, capsys):
+    b = _register(tmp_path, ack=5, created_at=_iso(time.time() - 60), dispatch_wait_at=_iso(time.time() - 6))
+    me = _register(tmp_path)
+    assert decide.wait_open_jobs(_ns(tmp_path, me)) == "clear"
+    assert f"stale open job {b} ignored (pending, no dispatch signal for 5s)" in capsys.readouterr().err
+
+
+def test_u2d_the_bound_is_the_remaining_wall_clock(tmp_path):
+    a = _register(tmp_path, status="running")
+    _event(tmp_path, a, time.time())
+    me = _register(tmp_path)
+    t0 = time.time()
+    # 600 s asked for, 0 s of this job's budget left: decided at once
+    assert decide.wait_open_jobs(_ns(tmp_path, me, wait_s=600, start_time=int(time.time()) - 100, timeout=50)) == f"busy {a}"
+    assert time.time() - t0 < 5
+
+
+# ---------------------------------------------------------------------------
+# U3 - the waiter's signal goes through the registry, status untouched
+# ---------------------------------------------------------------------------
+
+def test_u3_the_waiter_signals_on_its_own_record(tmp_path):
+    a = _register(tmp_path, status="running")
+    _event(tmp_path, a, time.time())
+    me = _register(tmp_path)
+    before = json.loads((tmp_path / f"{me}.json").read_text())
+    assert "dispatch_wait_at" not in before
+    t0 = time.time()
+    assert decide.wait_open_jobs(_ns(tmp_path, me, wait_s=2)) == f"busy {a}"
+    assert 1.5 <= time.time() - t0 < 10
+    after = json.loads((tmp_path / f"{me}.json").read_text())
+    assert after["status"] == "pending"
+    assert calendar.timegm(time.strptime(after["dispatch_wait_at"], "%Y-%m-%dT%H:%M:%SZ")) >= int(t0) - 1
+    assert after["transition_seq"] == before.get("transition_seq", 0) + 1, "one merge per DISPATCH_SIGNAL_EVERY"
+    assert not (tmp_path / f"{me}.events.log").exists(), "events.log is the agent's"
+    # the signal is what a later waiter judges this record by
+    later = _register(tmp_path)
+    live, stale = decide.open_jobs(str(tmp_path), "herdr:s", later)
+    assert me in {j for _c, j, _s in live}
+
+
+def test_u3b_the_signal_is_a_registry_merge(tmp_path, monkeypatch):
+    me = _register(tmp_path)
+    seen = {}
+    monkeypatch.setattr(mqtt_common, "update_job_status",
+                        lambda job_id, registry_dir, _source="unknown", **fields: seen.update(
+                            job=job_id, reg=registry_dir, source=_source, fields=fields))
+    decide._signal_dispatch_wait(str(tmp_path), me)
+    assert seen["job"] == me and seen["reg"] == str(tmp_path) and seen["source"] == "dispatch_wait"
+    assert set(seen["fields"]) == {"dispatch_wait_at"}
+
+
+# ---------------------------------------------------------------------------
+# U4 - the CLI: open-jobs (diagnostic) and a wrapper that is gone
+# ---------------------------------------------------------------------------
+
+def _cli(reg, *args):
+    return subprocess.run([sys.executable, "-m", "lib_py.delegation.wrapper", *args], capture_output=True,
+                          text=True, cwd=str(SKILLS), env=dict(os.environ, PYTHONPATH=str(SKILLS)), timeout=60)
+
+
+def test_u4_open_jobs_prints_live_and_stale(tmp_path):
+    a = _register(tmp_path, status="running")
+    _event(tmp_path, a, time.time())
+    b = _register(tmp_path, created_at=_iso(time.time() - 1000))
+    me = _register(tmp_path)
+    res = _cli(tmp_path, "open-jobs", "--job", me, "--agent-session", "herdr:s", "--registry-dir", str(tmp_path))
+    assert res.returncode == 0, res.stderr
+    assert f"{a}  running" in res.stdout and f"{b}  stale     pending, no dispatch signal for 180s" in res.stdout
+
+
+def test_u4b_wait_open_jobs_stops_with_its_wrapper(tmp_path):
+    a = _register(tmp_path, status="running")
+    _event(tmp_path, a, time.time())
+    me = _register(tmp_path)
+    gone = subprocess.Popen(["true"])
+    gone.wait()
+    res = _cli(tmp_path, "wait-open-jobs", "--job", me, "--agent-session", "herdr:s", "--registry-dir", str(tmp_path),
+               "--wait-s", "30", "--timeout", "3600", "--watch-pid", str(gone.pid))
+    assert res.stdout == "" and "is gone" in res.stderr, res.stdout + res.stderr
+
+
+# ---------------------------------------------------------------------------
+# U5 - finalize-end: phase registry_wait closes as cancelled/session_busy
+# ---------------------------------------------------------------------------
+
+def test_u5_finalize_end_registry_wait_is_a_session_busy_cancel(tmp_path):
+    reg = tmp_path / "reg"
+    reg.mkdir()
+    me = _register(reg)
+    scripts = tmp_path / "scripts"
+    scripts.mkdir()
+    log = tmp_path / "calls.log"
+    for name in ("registry.py", "publish_event.py"):
+        (scripts / name).write_text("import sys\n"
+                                    f"open({str(log)!r}, 'a').write(sys.argv[0].split('/')[-1] + ' ' + ' '.join(sys.argv[1:]) + '\\n')\n")
+    res = subprocess.run([sys.executable, "-m", "lib_py.delegation.wrapper", "finalize-end", "--job", me,
+                          "--registry-dir", str(reg), "--status", "error", "--timeout", "60", "--idle-timeout", "10",
+                          "--sub-rc", "1", "--phase", "registry_wait", "--type", "direct", "--final", "error",
+                          "--python", sys.executable, "--scripts-dir", str(scripts)],
+                         capture_output=True, text=True, cwd=str(SKILLS), env=dict(os.environ, PYTHONPATH=str(SKILLS)),
+                         timeout=60)
+    assert res.returncode == 0, res.stderr
+    calls = log.read_text().splitlines()
+    assert any(c.startswith("registry.py") and "close --job " + me + " --request cancelled --detail session_busy" in c
+               for c in calls), calls
+    assert any(c.startswith("publish_event.py") and "--event cancelled --detail session_busy" in c for c in calls), calls
+    assert "closed by" not in res.stderr
+
+
+# ---------------------------------------------------------------------------
+# S - the bash wrapper calls the decision in both dispatch paths
+# ---------------------------------------------------------------------------
+
+def test_s1_both_dispatch_paths_wait_by_the_registry_before_the_idle_wait():
+    src = WRAPPER.read_text()
+    assert src.count('_registry_wait "$JOB_ID" "$AGENT_SESSION"') == 1
+    assert src.count('_registry_wait "$JOB_ID" "$current_session"') == 1
+    for block in src.split("wait_for_agent_idle ")[1:]:
+        pass
+    direct = src.index('_registry_wait "$JOB_ID" "$AGENT_SESSION"')
+    assert direct < src.index('wait_for_agent_idle "$AGENT_SESSION"')
+    loop = src.index('_registry_wait "$JOB_ID" "$current_session"')
+    assert loop < src.index('wait_for_agent_idle "$current_session"')
+    body = src.split("_registry_wait() {")[1].split("\n}\n")[0]
+    assert "wait-open-jobs" in body and 'finalize_unfinished "$job_id" "error" "$SUB_RC_ERROR" "registry_wait"' in body
+    assert "exit 5" in body and "mam_int_knob MAM_DISPATCH_WAIT_S 600 0" in body

tests/test_issue50_dispatch_herdr.py

@@ -0,0 +1,572 @@
+"""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 _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)):
+        subprocess.run([HERDR, *sub], capture_output=True, text=True, env={"HOME": str(box.home), "PATH": os.environ["PATH"]},
+                       timeout=30)
+        time.sleep(0.5)
+    for pid in box.server_pids():
+        try:
+            os.kill(pid, signal.SIGKILL)
+        except OSError:
+            pass
+    shutil.rmtree(box.home, ignore_errors=True)
+
+
+@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)
+    proc_c = box.submit(prompt="work C", MAM_DISPATCH_WAIT_S="60")
+    c = box.wait_new_job(known | {b})
+    _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)