test(issue-50): c10 makes the same-second tie instead of timing it (reviews 98b31bb2 / 352066ec / 244b8072 F-B)

The live onboarding test asserted that the first job's wrapper registered in
the same wall-clock second as the onboarding it had just placed at a second
boundary. Process start-up has no sub-second contract, so the setup assertion
failed now and then before any dispatch behaviour was checked. The test now
stamps the onboarding record, which it owns and which nothing else depends on
yet, with the job's created_at once that record exists; the stamp is written
by atomic replace so a scanning wrapper never skips a half-written record.
The lifecycle assertions and the fixed-id tie coverage in test_u2h are unchanged.

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

.agents/skills/lib.sh

@@ -1486,6 +1486,56 @@ _herdr_scope_is_pane_inherited() {
   [ -n "${HERDR_WORKSPACE_ID:-}" ] && [ -n "$pane" ] && [ "${pane%%:*}" = "$HERDR_WORKSPACE_ID" ]
 }
 
+# Issue #50 follow-up: one pane claim at a time per herdr server.
+#
+# The shim's new-session chooses a pane - an idle one in a workspace on the
+# same cwd (BUG-4), a split of one, or the root pane of a fresh workspace - and
+# then claims it with `agent start`. Two shims on one server (several creates
+# at once into one folder) could choose the same idle pane in between, and the
+# later `agent start` failed with agent_pane_busy. mam_herdr runs the whole
+# new-session call holding one lock per server: a flock on fd 9, opened for
+# the block below and taken by lib_py.pane_lock (which also bounds the wait,
+# MAM_PANE_CLAIM_WAIT_S). It ends with the block or with this process, however
+# that ends, so a dead holder leaves nothing behind; the file is the target
+# server's, so other servers' creates are not held. The lock lives here, not
+# in the shim heredoc, which stays byte-identical (test_issue23_24 S11).
+#
+# _mam_pane_claim_target <herdr argv...>: prints the server a new-session call
+# names ("" for the ambient target, which lib_py.pane_lock resolves to the
+# caller pane's server - the one herdr answers on - or the default server);
+# rc 1 for any other call.
+_mam_pane_claim_target() {
+  local sess="${HERDR_SESSION_NAME:-}"
+  while [ "${1:-}" = "-L" ]; do sess="${2:-}"; shift 2 || return 1; done
+  [ "${1:-}" = "new-session" ] || return 1
+  [ "$sess" = "default" ] && sess=""
+  printf '%s\n' "$sess"
+}
+
+# _mam_with_pane_claim <server> <cmd...>: run <cmd> holding <server>'s pane lock.
+_mam_with_pane_claim() {
+  local sess="$1" f=""
+  shift
+  f=$(PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" python3 -m lib_py.pane_lock path --session "$sess") || f=""
+  if [ -z "$f" ]; then
+    echo "ERROR: no pane lock path for herdr session ${sess:-default} (lib_py.pane_lock); not placing this session" >&2
+    return 1
+  fi
+  mkdir -p "$(dirname "$f")"
+  {
+    case "$(PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" python3 -m lib_py.pane_lock acquire --fd 9)" in
+      locked) ;;
+      timeout)
+        echo "ERROR: another pane claim on herdr session ${sess:-default} did not end within MAM_PANE_CLAIM_WAIT_S; not placing this session" >&2
+        return 1 ;;
+      *)
+        echo "ERROR: the pane lock of herdr session ${sess:-default} could not be decided (lib_py.pane_lock); not placing this session" >&2
+        return 1 ;;
+    esac
+    "$@"
+  } 9>>"$f"
+}
+
 mam_herdr() {
   _init_herdr_isolation
   # Built as an argv array and decided inside `if`: this runs under
@@ -1503,6 +1553,11 @@ mam_herdr() {
       echo "mam_herdr: '$1' looks up by name; the caller pane's own workspace ${HERDR_WORKSPACE_ID} is not a scope" >&2
     fi
   fi
+  local _claim_sess=""
+  if _claim_sess=$(_mam_pane_claim_target "$@"); then
+    _mam_with_pane_claim "$_claim_sess" ${_scope_strip[@]+"${_scope_strip[@]}"} "$WORKSPACE_ROOT/.mam/shim/herdr" "$@"
+    return
+  fi
   ${_scope_strip[@]+"${_scope_strip[@]}"} "$WORKSPACE_ROOT/.mam/shim/herdr" "$@"
 }
 

.agents/skills/lib_py/atomic_yaml.py

@@ -115,6 +115,8 @@ def atomic_dump_yaml_main():
             d['herdr_sessions'] = []
 
         old_sessions = get_all_sessions_status(d)
+        # a mutation that replaces a row the monitor registered for a pane it
+        # created itself (create_session.sh) drops that name from old_roles
         old_roles = {s.get('name'): s.get('role') for s in db_sessions if s.get('role')}
 
         # --- caller mutation (module scope: sees d, yaml, os, glob, subprocess) ---

.agents/skills/lib_py/delegation/engine.py

@@ -284,11 +284,16 @@ def _notification(job_id, kind, **fields) -> Dict[str, Any]:
 def transition(job_id, request="merge", *, registry_dir=None, expected_status=None,
                expected_attempt=None, fields=None, source="unknown",
                notify=None, expected_notification=_UNSET,
-               expected_dispatch_token=None) -> TransitionResult:
+               expected_dispatch_token=None, guard=None) -> TransitionResult:
     """Decide and commit one state transition. The only writer of job status.
 
     Returns a result whose `notifications` have ALREADY been flushed; the field
     is kept so callers and tests can see what the transition emitted.
+
+    `guard(record)`, when given, runs under the registry lock before the
+    decision and refuses the commit by returning false: a precondition that
+    spans other records (Issue #50 dispatch admission), which the per-record
+    expectations cannot express. A refusal by the guard is not a conflict.
     """
     registry_dir = registry_dir or mqtt_common.DEFAULT_REGISTRY_DIR
     fields = dict(fields or {})
@@ -297,10 +302,13 @@ def transition(job_id, request="merge", *, registry_dir=None, expected_status=No
     with mqtt_common.registry_lock(registry_dir):
         record = mqtt_common.load_job(job_id, registry_dir)
         from_status = record.get("status")
-        decision = _decide(record, request, expected_status, expected_attempt,
+        if guard is not None and not guard(record):
-                           source=source, fields=fields,
+            decision = _Decision(False, from_status, "guard refused")
-                           expected_notification=expected_notification,
+        else:
-                           expected_dispatch_token=expected_dispatch_token)
+            decision = _decide(record, request, expected_status, expected_attempt,
+                               source=source, fields=fields,
+                               expected_notification=expected_notification,
+                               expected_dispatch_token=expected_dispatch_token)
 
         notifications: List[Dict[str, Any]] = []
         if decision.commit:

.agents/skills/lib_py/delegation/wrapper.py

@@ -27,7 +27,12 @@ free, and create --onboard waits for the onboarding to end):
   open-jobs --job J --agent-session S --registry-dir R
                                              diagnostic: the live open jobs of S
   wait-open-jobs ... --wait-s W --start-time T0 --timeout T [--watch-pid P]
-                                             prints `clear` or `busy <ids>`
+                 [--dispatch-token K] [--hold]   prints `clear`, `busy <ids>`, or
+                                             `superseded` (the record is not this
+                                             attempt's pending one any more)
+  hold-admission --job J --agent-session S --dispatch-token K --registry-dir R
+                 --timeout T [--start-time T0] [--watch-pid P]
+                                             the detached hold `clear` starts with --hold
 
 finalize-end answers on fd 3 (stdout when fd 3 is closed): its stdout carries
 registry.py's close output to the wrapper's stdout, as before. It runs as a
@@ -63,9 +68,25 @@ EXIT_CODES = {"completed": 0, "error": 1, "timeout": 2, "broker_unavailable": 3,
 # 44 = the agent ended it with error (or it was cancelled), 45 = it did not end
 # within MAM_ONBOARD_WAIT_S (or its record is gone). The session is kept either way.
 ONBOARD_EXIT_CODES = {"completed": 0, "error": 44, "cancelled": 44, "timeout": 45, "missing": 45}
-DEFAULT_ACK_TIMEOUT_SEC, DEFAULT_IDLE_TIMEOUT_SEC = 180, 120  # registry.py's defaults, as a last resort
+DEFAULT_IDLE_TIMEOUT_SEC = 120  # registry.py's default, as a last resort
+# Issue #50: a pending record's liveness is its dispatcher's signal - written
+# every DISPATCH_SIGNAL_EVERY s while it waits and, once admitted, while it
+# holds the session - and it stays live for EVERY x MISSES s after the last
+# one (Issue #18's heartbeat x misses, on the dispatcher's side). The window
+# is not ack_timeout_sec: that is the agent's budget to answer a prompt and
+# goes down to 1 s, below any cadence a second-resolution signal can keep
+# (review aa1bff92 F2).
 DISPATCH_SIGNAL_FIELD = "dispatch_wait_at"
 DISPATCH_SIGNAL_EVERY = 10
+DISPATCH_SIGNAL_MISSES = 3
+DISPATCH_SIGNAL_WINDOW = DISPATCH_SIGNAL_EVERY * DISPATCH_SIGNAL_MISSES
+# The admission (review aa1bff92 F1): `clear` is persisted on the admitted
+# record, scoped to its attempt by the dispatch token, so a later candidate -
+# a lower id in the same second, a redispatched older loop job - waits for it
+# until it ends or its signal goes stale.
+ADMITTED_AT_FIELD = "dispatch_admitted_at"
+ADMITTED_TOKEN_FIELD = "dispatch_admitted_token"
+DEFAULT_ACK_TIMEOUT_SEC = 180   # registry.DEFAULT_ACK_TIMEOUT_SEC (a test keeps them equal)
 
 
 def _now() -> int:
@@ -243,12 +264,23 @@ def open_jobs(registry_dir: str, session: str, job_id: str, now: Optional[int] =
 
     live:  [(created_at, job_id, status)] - open jobs the Issue #18 contract
            still calls alive: `running` with an event in events.log within
-           idle_timeout_sec; `pending` with a dispatch signal (the waiter's
+           idle_timeout_sec; `pending` with a waiter's signal (`dispatch_wait_at`)
-           `dispatch_wait_at`, else created_at) within ack_timeout_sec.
+           within DISPATCH_SIGNAL_WINDOW, or - never signalled - created within
+           its own ack_timeout_sec: a producer that registers and injects
+           without a waiter (create --onboard) waits that long for `started`
+           (review 55219be1 R2). Such a never-signalled record is reported as
+           `injected` and a live pending record whose admission names its
+           current dispatch token as `admitted`: both hold the session as a
+           running job does, whatever their place in the FIFO - the prompt is
+           already in the pane (review 5a09a13f F-A: in the FIFO an `injected`
+           record of the same second lost to a lower job id). A waiter's own
+           record is unsignalled only until its first scan (wait_open_jobs
+           signals first).
     stale: [(job_id, reason)] - non-terminal records nothing has touched for
            longer than that: a wrapper killed before or while dispatching, a
-           zombie's unclosed record. They are reported, never waited for, so
+           zombie's unclosed record, an onboarding never acknowledged. They
-           a dead record cannot block a session forever."""
+           are reported, never waited for, so a dead record cannot block a
+           session forever."""
     now = _now() if now is None else now
     sess = _bare(session)
     live, stale = [], []
@@ -273,35 +305,168 @@ def open_jobs(registry_dir: str, session: str, job_id: str, now: Optional[int] =
             else:
                 stale.append((other, f"running, no event for {idle}s"))
         elif status == "pending":
-            ack = _int_field(rec, "ack_timeout_sec", DEFAULT_ACK_TIMEOUT_SEC)
+            if rec.get(DISPATCH_SIGNAL_FIELD):
-            seen = _parse_ts(rec.get(DISPATCH_SIGNAL_FIELD) or rec.get("created_at"))
+                seen, window = _parse_ts(rec.get(DISPATCH_SIGNAL_FIELD)), DISPATCH_SIGNAL_WINDOW
-            if seen is not None and now - seen <= ack:
+                why = f"no dispatch signal for {window}s"
-                live.append((str(rec.get("created_at") or ""), other, status))
             else:
-                stale.append((other, f"pending, no dispatch signal for {ack}s"))
+                seen, window = _parse_ts(rec.get("created_at")), _int_field(rec, "ack_timeout_sec", DEFAULT_ACK_TIMEOUT_SEC)
+                why = f"no dispatch signal and no started within {window}s"
+            if seen is not None and now - seen <= window:
+                if not rec.get(DISPATCH_SIGNAL_FIELD):
+                    kind = "injected"
+                elif rec.get(ADMITTED_TOKEN_FIELD) and rec.get(ADMITTED_TOKEN_FIELD) == rec.get("dispatch_token"):
+                    kind = "admitted"
+                else:
+                    kind = status
+                live.append((str(rec.get("created_at") or ""), other, kind))
+            else:
+                stale.append((other, f"pending, {why}"))
     return live, stale
 
 
-def _signal_dispatch_wait(registry_dir: str, job_id: str) -> None:
+def _first_in_line(live, mine) -> bool:
+    """No live running, admitted or injected job, and `mine` earliest among
+    the live pending ones (FIFO by (created_at, job_id))."""
+    if any(st != "pending" for _c, _j, st in live):
+        return False
+    return min([(c, j) for c, j, _st in live] + [mine]) == mine
+
+
+def _own(rec: Optional[dict], token: Optional[str], session: str) -> bool:
+    """The record is this attempt's pending one, still bound to the session the
+    waiter scans - the identity the waiter and its hold are pinned to for their
+    whole life (review 55219be1 R1): the token the wrapper read for this
+    attempt and the session it asked about, never re-read from the record. A
+    redispatch (new token), a move to another session or a close by someone
+    else ends them without a write."""
+    return bool(rec) and bool(token) and rec.get("status") == "pending" and rec.get("dispatch_token") == token \
+        and _bare(str(rec.get("agent_session") or "")) == _bare(session)
+
+
+def _started_own(rec: Optional[dict], token: Optional[str], session: str) -> bool:
+    """This attempt's record, in this session, already `running`: its own
+    `started` landed before the prompt went out (the registry ack of Issue
+    #14 wins over an unreachable broker; a test that acknowledges at
+    registration). Nothing was redispatched, moved or closed by someone else
+    - there is just nothing left to wait for."""
+    return bool(rec) and bool(token) and rec.get("status") == "running" and rec.get("dispatch_token") == token \
+        and _bare(str(rec.get("agent_session") or "")) == _bare(session)
+
+
+def _merge_own(registry_dir: str, job_id: str, token: Optional[str], session: str, source: str, fields: dict,
+               guard=None) -> bool:
+    """One merge through the engine, committed only while the record re-read
+    under the registry lock is still this attempt's (_own) and `guard`, if
+    any, agrees. False when it is not, or the record is gone or unwritable."""
+    from lib_py.delegation import engine
+
+    def ok(record):
+        return _own(record, token, session) and (guard is None or guard(record))
+
+    try:
+        return engine.transition(job_id, "merge", registry_dir=registry_dir, source=source, guard=ok,
+                                 fields=fields).committed
+    except Exception as exc:  # the record is gone or unwritable: nothing written, say so
+        print(f"dispatch wait: could not write {source} on {job_id}: {exc}", file=sys.stderr)
+        return False
+
+
+def _admit(registry_dir: str, session: str, job_id: str, token: Optional[str], mine) -> bool:
+    """`clear`, persisted: one merge through the engine whose guard re-reads
+    the session's open jobs under the registry lock, so two waiters cannot
+    both be admitted. Scoped to this attempt and session (_own: a redispatch
+    voids it); the signal is renewed with it. False when the record is not
+    this attempt's pending one any more or another waiter was admitted first
+    - the caller keeps waiting and sees that one."""
+    from lib_py.delegation import mqtt_common
+
+    def first(_record):
+        live, _stale = open_jobs(registry_dir, session, job_id)
+        return _first_in_line(live, mine)
+
+    now = mqtt_common._utcnow()
+    return _merge_own(registry_dir, job_id, token, session, "dispatch_admit",
+                      {DISPATCH_SIGNAL_FIELD: now, ADMITTED_AT_FIELD: now, ADMITTED_TOKEN_FIELD: token}, first)
+
+
+def _start_hold(a: argparse.Namespace, token: str) -> None:
+    """The waiter answers `clear` and ends; a detached hold keeps the admitted
+    record's signal fresh until the record leaves pending (the agent's
+    `started`, a close), its attempt changes, the wrapper is gone, or the
+    wall clock ends (review aa1bff92 F2). It is given the waiter's pinned
+    identity, not the record's. Its output is not the answer."""
+    cmd = [sys.executable, "-m", "lib_py.delegation.wrapper", "hold-admission", "--job", a.job,
+           "--agent-session", a.agent_session, "--dispatch-token", token,
+           "--registry-dir", a.registry_dir, "--timeout", str(a.timeout)]
+    if a.start_time is not None:
+        cmd += ["--start-time", str(a.start_time)]
+    if a.watch_pid:
+        cmd += ["--watch-pid", str(a.watch_pid)]
+    subprocess.Popen(cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+                     start_new_session=True, env=dict(os.environ, PYTHONPATH=_pythonpath()))
+
+
+def _pythonpath() -> str:
+    skills = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+    extra = os.environ.get("PYTHONPATH", "")
+    return skills + (os.pathsep + extra if extra else "")
+
+
+def hold_admission(a: argparse.Namespace) -> str:
+    """`released` once the record is not the admitted pending one of the
+    attempt and session the hold was given (a token that changed before it
+    started counts), `timeout` at the wall clock; the signal is renewed every
+    DISPATCH_SIGNAL_EVERY s in between, under the same identity (_merge_own).
+    OwnerGone (the wrapper) ends it."""
+    token = a.dispatch_token
+    deadline = (a.start_time if a.start_time is not None else _now()) + a.timeout
+    last_signal = _now()
+    while True:
+        _sleep(a, 1)
+        rec = _load_record(a.registry_dir, a.job)
+        if not _own(rec, token, a.agent_session) or rec.get(ADMITTED_TOKEN_FIELD) != token:
+            return "released"
+        now = _now()
+        if now >= deadline:
+            return "timeout"
+        if now - last_signal >= DISPATCH_SIGNAL_EVERY:
+            _signal_dispatch_wait(a.registry_dir, a.job, token, a.agent_session, admitted=True)
+            last_signal = now
+
+
+def _signal_dispatch_wait(registry_dir: str, job_id: str, token: Optional[str], session: str,
+                          admitted: bool = False) -> bool:
     """The waiter's liveness: a merge of dispatch_wait_at into its own pending
     record through the registry's one transition boundary (status untouched,
-    no event in events.log - that log is the agent's)."""
+    no event in events.log - that log is the agent's), only while the record
+    is still this attempt's (_own); for the hold, only while its admission
+    stands. False when nothing was written."""
     from lib_py.delegation import mqtt_common  # loads paho, as the package import did already
-    try:
+    guard = (lambda record: record.get(ADMITTED_TOKEN_FIELD) == token) if admitted else None
-        mqtt_common.update_job_status(job_id, registry_dir, _source="dispatch_wait",
+    return _merge_own(registry_dir, job_id, token, session, "dispatch_wait",
-                                      **{DISPATCH_SIGNAL_FIELD: mqtt_common._utcnow()})
+                      {DISPATCH_SIGNAL_FIELD: mqtt_common._utcnow()}, guard)
-    except Exception as exc:  # the record is gone or unwritable: keep waiting, say so once per tick
-        print(f"dispatch wait: could not signal on {job_id}: {exc}", file=sys.stderr)
 
 
 def wait_open_jobs(a: argparse.Namespace) -> str:
-    """`clear` when no live open job of the session remains and this job is
+    """`clear` when no live running or admitted job of the session remains,
-    the earliest live pending one (FIFO by (created_at, job_id)); `busy <ids>`
+    this job is the earliest live pending one (FIFO by (created_at, job_id))
-    when the bound runs out first. The bound is min(--wait-s, this job's
+    and its admission is committed (_admit); `busy <ids>` when the bound runs
-    remaining wall clock). Every DISPATCH_SIGNAL_EVERY s the waiter signals on
+    out first; `superseded` once the record is not this attempt's pending one
-    its own record so later waiters see it alive; a waiter whose wrapper is
+    in this session any more (redispatched, moved, closed by someone else) -
-    gone stops (OwnerGone) and its signal goes stale."""
+    nothing is written or closed then, the record is its owner's; `clear`
+    at once, with no admission, when it is this attempt's and already
+    running (_started_own). The bound
+    is min(--wait-s, this job's remaining wall clock). The identity is pinned
+    at the start: --dispatch-token (the token the wrapper read for this
+    attempt), else the record's token read once. The waiter signals on its own
+    record before its first scan - an unsignalled live record is one a
+    producer injected without a waiter (open_jobs: `injected`), and this one
+    is not - and every DISPATCH_SIGNAL_EVERY s after, so other waiters see it
+    alive; a waiter whose wrapper is gone stops (OwnerGone) and its signal
+    goes stale.
+    With --hold the admission is kept alive after the answer (_start_hold)."""
     me = _load_record(a.registry_dir, a.job) or {}
+    token = getattr(a, "dispatch_token", None) or me.get("dispatch_token")
     mine = (str(me.get("created_at") or ""), a.job)
     remaining = (a.start_time if a.start_time is not None else _now()) + a.timeout - _now()
     deadline = _now() + max(0, min(a.wait_s, remaining))
@@ -309,20 +474,23 @@ def wait_open_jobs(a: argparse.Namespace) -> str:
     last_signal = None
     while True:
         now = _now()
+        rec = _load_record(a.registry_dir, a.job)
+        if not _own(rec, token, a.agent_session):
+            return "clear" if _started_own(rec, token, a.agent_session) else "superseded"
+        if last_signal is None or now - last_signal >= DISPATCH_SIGNAL_EVERY:
+            _signal_dispatch_wait(a.registry_dir, a.job, token, a.agent_session)
+            last_signal = now
         live, stale = open_jobs(a.registry_dir, a.agent_session, a.job, now)
         for other, reason in stale:
             if other not in reported:
                 reported.add(other)
                 print(f"stale open job {other} ignored ({reason})", file=sys.stderr)
-        if not any(st == "running" for _c, _j, st in live):
+        if _first_in_line(live, mine) and _admit(a.registry_dir, a.agent_session, a.job, token, mine):
-            queue = [(c, j) for c, j, st in live if st == "pending"] + [mine]
+            if getattr(a, "hold", False):
-            if min(queue) == mine:
+                _start_hold(a, token)
-                return "clear"
+            return "clear"
         if now >= deadline:
             return "busy " + " ".join(j for _c, j, _st in live)
-        if last_signal is None or now - last_signal >= DISPATCH_SIGNAL_EVERY:
-            _signal_dispatch_wait(a.registry_dir, a.job)
-            last_signal = now
         _sleep(a, 1)
 
 
@@ -480,6 +648,16 @@ def _parser() -> argparse.ArgumentParser:
             s.add_argument("--start-time", type=int, default=None)
             s.add_argument("--timeout", type=int, required=True, help="the job's wall-clock budget")
             s.add_argument("--watch-pid", type=int, default=None, help="the wrapper's pid; gone -> stop waiting")
+            s.add_argument("--dispatch-token", default=None, help="this attempt's token (pinned; else read once)")
+            s.add_argument("--hold", action="store_true", help="keep the admission's signal fresh after `clear`")
+    s = sub.add_parser("hold-admission")
+    s.add_argument("--job", required=True)
+    s.add_argument("--agent-session", required=True)
+    s.add_argument("--dispatch-token", required=True, help="the waiter's pinned token")
+    s.add_argument("--registry-dir", required=True)
+    s.add_argument("--start-time", type=int, default=None)
+    s.add_argument("--timeout", type=int, required=True)
+    s.add_argument("--watch-pid", type=int, default=None, help="the wrapper's pid; gone -> stop holding")
     for name in ("finalize-begin", "finalize-end"):
         s = sub.add_parser(name)
         s.add_argument("--job", required=True)
@@ -544,6 +722,11 @@ def main(argv: Optional[List[str]] = None) -> int:
             print(wait_open_jobs(a))
         except OwnerGone:
             print(f"wait-open-jobs: wrapper pid {a.watch_pid} is gone; stopped waiting on job {a.job}", file=sys.stderr)
+    elif a.cmd == "hold-admission":
+        try:
+            print(hold_admission(a))
+        except OwnerGone:
+            print(f"hold-admission: wrapper pid {a.watch_pid} is gone; stopped holding job {a.job}", file=sys.stderr)
     else:
         try:
             _answer(finalize_end(a))

.agents/skills/lib_py/pane_lock.py

@@ -0,0 +1,98 @@
+"""Issue #50 follow-up: one pane claim at a time per herdr server.
+
+The herdr shim's `new-session` chooses a pane - an idle one in a workspace on
+the same cwd (BUG-4), a split of one, or the root pane of a fresh workspace -
+and then claims it with `agent start`. Two shims on one server (several
+`create` at once into one folder) could choose the same idle pane in between,
+and the later `agent start` failed with agent_pane_busy. The shim now holds
+one lock per server from the choice to the claim.
+
+  pane_lock.py path --session S   the lock file of herdr session S, under the
+                                  HOME whose socket the session uses. S empty
+                                  (or `default`) is the shim's ambient target:
+                                  herdr then answers on the server of the pane
+                                  this runs in (HERDR_SESSION / HERDR_SOCKET_PATH,
+                                  as lib.sh _herdr_caller_server reads them), or
+                                  the default server outside any pane - the
+                                  lock is that server's (review e8e471ff R1)
+  pane_lock.py acquire --fd N     flock(LOCK_EX) on the inherited fd N; prints
+                                  `locked`, or `timeout` once MAM_PANE_CLAIM_WAIT_S
+                                  (default 120, minimum 1) has passed
+
+The lock is a BSD flock, which belongs to the open file the shim holds on fd 9:
+it outlives this helper and ends with the shim's process, however that ends
+(exit, kill -9), so a dead holder leaves nothing to steal. flock(1) is not on
+macOS; fcntl.flock is. Stdlib only, Python 3.9.
+"""
+from __future__ import annotations
+
+import argparse
+import fcntl
+import os
+import sys
+import time
+from typing import List, Mapping, Optional
+
+from lib_py import ints
+
+DEFAULT_WAIT_S = 120
+
+
+def caller_server(env: Mapping[str, str]) -> str:
+    """The server whose pane this process runs in, by the pane's exports:
+    HERDR_SESSION when present, else HERDR_SOCKET_PATH's layout
+    ($HOME/.config/herdr/herdr.sock -> default, .../sessions/<name>/herdr.sock
+    -> name); "" outside a pane or for an unrecognised layout."""
+    if env.get("HERDR_SESSION"):
+        return env["HERDR_SESSION"]
+    sock = env.get("HERDR_SOCKET_PATH") or ""
+    if sock.endswith("/.config/herdr/herdr.sock"):
+        return "default"
+    if sock.endswith("/herdr.sock") and "/.config/herdr/sessions/" in sock:
+        return sock[:-len("/herdr.sock")].rsplit("/", 1)[-1]
+    return ""
+
+
+def lock_path(session: Optional[str], env: Mapping[str, str]) -> str:
+    home = env.get("HOME") or env.get("HOME_DIR") or os.path.expanduser("~")
+    if session in (None, "", "default"):
+        session = caller_server(env) or "default"
+    return os.path.join(home, ".local", "state", "mam", "pane-lock", session + ".lock")
+
+
+def wait_s(env: Mapping[str, str]) -> int:
+    return ints.positive_int(env.get("MAM_PANE_CLAIM_WAIT_S"), 1) or DEFAULT_WAIT_S
+
+
+def acquire(fd: int, wait: float) -> str:
+    """`locked`, or `timeout` when the lock was not free within `wait` s. The
+    deadline is on the monotonic clock; a step of the wall clock does not
+    stretch or cut it."""
+    deadline = time.monotonic() + wait
+    while True:
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            return "locked"
+        except OSError:
+            if time.monotonic() >= deadline:
+                return "timeout"
+            time.sleep(0.2)
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    p = argparse.ArgumentParser(prog="pane_lock.py")
+    sub = p.add_subparsers(dest="cmd", required=True)
+    s = sub.add_parser("path")
+    s.add_argument("--session", default="")
+    s = sub.add_parser("acquire")
+    s.add_argument("--fd", type=int, required=True)
+    a = p.parse_args(argv)
+    if a.cmd == "path":
+        print(lock_path(a.session, os.environ))
+    else:
+        print(acquire(a.fd, wait_s(os.environ)))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

.agents/skills/multi-agent-mux-create/SKILL.md

@@ -178,13 +178,15 @@ bash .agents/skills/multi-agent-mux-create/scripts/create_session.sh --workspace
 
 `MAM_ONBOARD_WAIT_S` (default 600): how long `--onboard` waits for the onboarding job to end after the `started` handshake.
 
+`MAM_PANE_CLAIM_WAIT_S` (default 120): creates (and resumes) on one herdr server choose and claim their pane one at a time; how long one waits for another's claim to end before giving up (exit 1, nothing spawned).
+
 | Exit | Meaning |
 |---|---|
 | 0 | Created; or `--dry-run` |
 | 1 | Pre-flight failure, nothing was spawned: workspace not a directory, herdr not installed, agent CLI not in `PATH`, agent not authenticated. After spawn: the task brief could not be written or the instruction injection failed (the session is rolled back) |
 | 2 | Bad arguments; control-plane binding refused for a worktree; whitespace in a control-plane path that must be forwarded |
 | 3 | A herdr session of that name already exists (`multi-agent-mux-resume`, or `multi-agent-mux-stop` first) |
-| 4 | The YAML already has a `running` row of that name. The YAML write raises it after spawn, and the herdr session is rolled back |
+| 4 | The YAML already has a `running` row of that name. The YAML write raises it after spawn, and the herdr session is rolled back. A row the monitor registered for this very pane while it was being created is replaced instead |
 | 42 | The TUI never became ready and the process is gone, or it is alive and `MAM_KILL_ON_READY_TIMEOUT=1` (rolled back) |
 | 43 | The rollback could not kill the herdr session; it is still running |
 | 44 | The onboarding job ended with `error` (or was cancelled); the session is kept (Issue #50) |

.agents/skills/multi-agent-mux-create/scripts/create_session.sh

@@ -527,6 +527,20 @@ sessions = d.setdefault('herdr_sessions', [])
 # P0-D: 같은 이름 엔트리가 status=running 이면만 거부. terminated/archived 는
 # 재사용 가능 — 낡은 엔트리를 제거하고 새로 append (create -> delete -> create).
 running_same = [s for s in sessions if s.get('name') == name and s.get('status') == 'running']
+# Issue #50 follow-up: a reconcile pass (another create's end-of-run `--once`,
+# the subscriber's cycle) can adopt the pane this create started moments ago,
+# before this row lands. A running row of this name that the monitor
+# registered, with no job, for a pane created after this create began, is
+# this create's own pane - herdr keeps live names unique per server and the
+# preflight refused an older session of this name - so it is replaced, not
+# refused as a second live session.
+def _adopted_own(s):
+    return (s.get('last_visible_note') == 'auto-registered by monitor' and not s.get('delegate_job_id')
+            and epoch.isdigit() and int(s.get('herdr_session_epoch') or 0) >= int(epoch) - 1)
+if running_same and all(_adopted_own(s) for s in running_same):
+    print(f"NOTE: the monitor registered {name} while it was being created; that row is replaced by this one", flush=True)
+    running_same = []
+    old_roles.pop(name, None)   # its role was read off the name; --role is the fact (atomic_yaml's role guard)
 if running_same:
     print(f"ERROR: {name} already running in agent-sessions.yaml", flush=True)
     raise SystemExit(4)

.agents/skills/multi-agent-mux-delegate-job/SKILL.md

@@ -104,7 +104,7 @@ Under fail-closed operation (`MAM_DELEGATE_FAIL_CLOSED=1`, default), the CLI ret
 | `2` | Timeout | `error` (`deadline_timeout`) | Subscriber idle timeout (K x N without an event; the total budget is the wall-clock row below). **R1 liveness grace (Issue #18)**: after the subscriber idles out the wrapper keeps re-observing the disk while the job's events.log shows an event at most K x N (`--idle-timeout`) old - the automatic heartbeat counts - up to the wall-clock `--timeout`. herdr status and the pane are logged as a diagnostic only. |
 | `2` | Wall clock | `error` (`wall_clock_timeout`) | The budget (`--timeout`) ran out, events or not (Issue #33; was `deadline_timeout` before v4.8.0). |
 | `3` | Broker Error | `error` (`wall_clock_timeout` once the budget is spent, else `deadline_timeout`) | Broker unavailable (`MQTT_BROKER_UNAVAILABLE`) and terminal status not found on disk during deadline. |
-| `5` | Session Busy | `cancelled` (`session_busy`) | Precondition failed / `session_busy`: Preflight check (`preflight_agent`) or TUI idle handshake wait failed (prompt never delivered, retryable). |
+| `5` | Session Busy | `cancelled` (`session_busy`) | Precondition failed / `session_busy`: Preflight check (`preflight_agent`) or TUI idle handshake wait failed, or the session's open jobs in the registry did not end within `MAM_DISPATCH_WAIT_S` (Issue #50) (prompt never delivered, retryable); or the record was redispatched or closed by someone else while it waited (then this wrapper leaves it untouched: no close, no `cancelled`). |
 
 In all execution paths, the final line of standard output prints the persistent audit log directory path (`<logs_root>/<job_id>`), fulfilling the parsing contract for orchestrator runners.
 

.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job

@@ -358,7 +358,7 @@ EOF
     #    open jobs - an onboarding still running, a job sent by hand - end
     #    first, in order; herdr's agent_status is the screen, read after)
     if [ "$AGENT" != "human" ] && ! _is_remote "$AGENT_SESSION"; then
-      _registry_wait "$JOB_ID" "$AGENT_SESSION"
+      _registry_wait "$JOB_ID" "$AGENT_SESSION" "$_disp_tok"
       local idle_rc=0
       wait_for_agent_idle "$AGENT_SESSION" || idle_rc=$?
       export _SKS_IDLE_RC="$idle_rc"
@@ -518,22 +518,6 @@ EOF
         exit 5
       fi
 
-      # Registry wait (Issue #50), then idle wait
-      local cur_agent="$AGENT"
-      [ "$_phase" = "reviewer" ] && cur_agent="$REVIEWER"
-      if [ "$cur_agent" != "human" ] && ! _is_remote "$current_session"; then
-        _registry_wait "$JOB_ID" "$current_session"
-        local idle_rc=0
-        wait_for_agent_idle "$current_session" || idle_rc=$?
-        export _SKS_IDLE_RC="$idle_rc"
-        if [ $idle_rc -eq 1 ] || [ $idle_rc -eq 2 ]; then
-          finalize_unfinished "$JOB_ID" "error" "$idle_rc" "idle_wait"
-          local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
-          echo "$logs_root/$JOB_ID"
-          exit 5
-        fi
-      fi
-
       # Update job details in registry
       "$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" update \
           --job "$JOB_ID" \
@@ -544,9 +528,30 @@ EOF
           --role "$display_role" \
           --status "pending"
 
+      # The reset above is a redispatch: it issued this attempt's token. Read it
+      # now, so a close from the waits below (busy, idle_wait) names this
+      # attempt, not the previous phase's (review reviewer-grok-01)
       local _disp_tok=""
       _disp_tok="$("$PY" -c "import json, os; p=os.path.join('$REGISTRY_DIR', '$JOB_ID.json'); print(json.load(open(p)).get('dispatch_token','') or '') if os.path.exists(p) else print('')" 2>/dev/null || true)"
 
+      # Registry wait (Issue #50), then idle wait - on the record just reset to
+      # pending for this phase, so the waiter's signal is written and other
+      # waiters see this job in the FIFO (review F1)
+      local cur_agent="$AGENT"
+      [ "$_phase" = "reviewer" ] && cur_agent="$REVIEWER"
+      if [ "$cur_agent" != "human" ] && ! _is_remote "$current_session"; then
+        _registry_wait "$JOB_ID" "$current_session" "$_disp_tok"
+        local idle_rc=0
+        wait_for_agent_idle "$current_session" || idle_rc=$?
+        export _SKS_IDLE_RC="$idle_rc"
+        if [ $idle_rc -eq 1 ] || [ $idle_rc -eq 2 ]; then
+          finalize_unfinished "$JOB_ID" "error" "$idle_rc" "idle_wait"
+          local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
+          echo "$logs_root/$JOB_ID"
+          exit 5
+        fi
+      fi
+
       # Start subscriber
       local logf="$REGISTRY_DIR/${JOB_ID}.iter_${iteration}_${display_role}.subscriber.out"
       local loop_sub_args=(--attempt-id "$attempt" --ack-timeout "$TIMEOUT")
@@ -839,20 +844,29 @@ _within_deadline() {
   wait "$pid"
 }
 
-# _registry_wait <job_id> <session>  (Issue #50)
+# _registry_wait <job_id> <session> <dispatch_token>  (Issue #50)
 # Waits, by the registry alone, until no live open job of <session> remains
 # and this job is the earliest pending one (lib_py.delegation.wrapper
 # wait-open-jobs: FIFO, the waiter signals on its own record, a dead record is
-# ignored). A wait that runs out (MAM_DISPATCH_WAIT_S, default 600 s, bounded
+# ignored). The waiter is pinned to this attempt's token and session (review
-# by this job's wall clock) closes this job as cancelled/session_busy and
+# 55219be1 R1): a record redispatched, moved or closed by someone else while
-# exits 5, like the preflight and the idle wait.
+# it waited is `superseded` - this wrapper exits 5 without touching it, the
+# record is its new owner's. A wait that runs out (MAM_DISPATCH_WAIT_S,
+# default 600 s, bounded by this job's wall clock) closes this job as
+# cancelled/session_busy and exits 5, like the preflight and the idle wait.
 _registry_wait() {
-  local job_id="$1" sess="$2" res
+  local job_id="$1" sess="$2" tok="$3" res
   _decided res wait-open-jobs --job "$job_id" --agent-session "$sess" --registry-dir "$REGISTRY_DIR" \
+      ${tok:+--dispatch-token "$tok"} \
       --wait-s "$(mam_int_knob MAM_DISPATCH_WAIT_S 600 0)" ${START_TIME:+--start-time "$START_TIME"} \
-      --timeout "$TIMEOUT" --watch-pid "$$"
+      --timeout "$TIMEOUT" --watch-pid "$$" --hold
   case "$res" in
     clear) return 0 ;;
+    superseded)
+      echo "ERROR: job $job_id was redispatched, moved or closed by someone else while it waited for session '$sess'; this attempt is not dispatched and the record is left as it is" >&2
+      local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
+      echo "$logs_root/$job_id"
+      exit 5 ;;
     busy*)
       echo "ERROR: session '$sess' still has open jobs in the registry (${res#busy }); job $job_id is not dispatched" >&2
       finalize_unfinished "$job_id" "error" "$SUB_RC_ERROR" "registry_wait" "$sess"

tests/data/skill_code_blocks.json

@@ -88,8 +88,8 @@
   },
   {
    "skill": "multi-agent-mux-create",
-   "start": 202,
+   "start": 204,
-   "end": 206,
+   "end": 208,
    "lang": "bash",
    "sha": "38836d31bf38",
    "class": "c",
@@ -103,8 +103,8 @@
   },
   {
    "skill": "multi-agent-mux-create",
-   "start": 228,
+   "start": 230,
-   "end": 250,
+   "end": 252,
    "lang": "bash",
    "sha": "d90cfd2555fd",
    "class": "b",
@@ -617,7 +617,7 @@
     "other"
    ],
    "passthrough": {
-    "4": "multi-agent-mux-create/scripts/create_session.sh:532"
+    "4": "multi-agent-mux-create/scripts/create_session.sh:546"
    }
   }
  },

tests/test_herdr_shim_contract.py

@@ -604,3 +604,37 @@ kill "$_mam_server_pid" 2>/dev/null || true
         f"daemon child stayed in the caller's process group (not detached): {fields}"
     )
 
+
+
+def test_h26_new_session_claims_its_pane_under_the_servers_lock(mam_sandbox, mock_herdr, mock_agents):
+    """Issue #50 follow-up: from the pane choice to `agent start` the shim holds
+    one lock per herdr server (lib_py.pane_lock, a flock on fd 9). While
+    another shim holds it, this one waits MAM_PANE_CLAIM_WAIT_S and then gives
+    up without creating, splitting or starting anything; once it is free the
+    same call goes through as before."""
+    import glob
+    import os
+    tmp_path = mam_sandbox
+    res = _run_lib(tmp_path, f'herdr new-session -d -s "test-creator-claude" -c "{tmp_path}" "claude --dangerously-skip-permissions"')
+    assert res.returncode == 0, res.stderr
+    locks = glob.glob(str(tmp_path / ".local" / "state" / "mam" / "pane-lock" / "*.lock"))
+    assert len(locks) == 1, locks                     # one file per server, under the sandbox HOME
+    n_calls = len(_load_mock_state(mock_herdr).get("calls", []))
+    holder = subprocess.Popen(["bash", "-c", 'exec 9>>"$1"; python3 -m lib_py.pane_lock acquire --fd 9; sleep 600',
+                               "holder", locks[0]], stdout=subprocess.PIPE, text=True,
+                              env=dict(os.environ, PYTHONPATH=str(tmp_path / ".agents" / "skills")))
+    try:
+        assert holder.stdout.readline().strip() == "locked"
+        os.environ["MAM_PANE_CLAIM_WAIT_S"] = "1"
+        res = _run_lib(tmp_path, f'herdr new-session -d -s "test-reviewer-claude" -c "{tmp_path}" "claude --dangerously-skip-permissions"')
+        assert res.returncode == 1 and "did not end within MAM_PANE_CLAIM_WAIT_S" in res.stderr, (res.returncode, res.stderr)
+        new_calls = [c[2:] if c[:1] == ["--session"] else c for c in _load_mock_state(mock_herdr).get("calls", [])[n_calls:]]
+        assert not [c for c in new_calls if c[:2] in (["workspace", "create"], ["pane", "split"], ["agent", "start"])], new_calls
+        assert not [n for n in _load_mock_state(mock_herdr).get("agents", {}) if "test-reviewer-claude" in n]
+    finally:
+        os.environ.pop("MAM_PANE_CLAIM_WAIT_S", None)
+        holder.kill()
+        holder.wait()
+    res = _run_lib(tmp_path, f'herdr new-session -d -s "test-reviewer-claude" -c "{tmp_path}" "claude --dangerously-skip-permissions"')
+    assert res.returncode == 0, res.stderr
+    assert [n for n in _load_mock_state(mock_herdr).get("agents", {}) if "test-reviewer-claude" in n]

tests/test_issue26_ints.py

@@ -55,6 +55,10 @@ INT_KNOBS = {
     "MAM_DIFF_MAX_BYTES": (1, "200000"),    # 0 would truncate every diff
     "MAM_DIFF_MAX_LINES": (1, "4000"),
     "RECONCILE_POLL_INTERVAL": (1, "15"),   # sleep; 0 would poll in a busy loop
+    # Issue #50: create's wait for the onboarding job to end (0 would exit 45 at
+    # once) and the wrapper's registry wait (0: decide at once - clear or busy)
+    "MAM_ONBOARD_WAIT_S": (1, "600"),
+    "MAM_DISPATCH_WAIT_S": (0, "600"),
     # delegate_wait_started's ack_s (MAM_ONBOARD_ACK_S / DELEGATE_JOB_ACK_S via create_session.sh)
     "delegate_wait_started:ack_s": (0, "180"),
     # wrapper flags
@@ -84,7 +88,8 @@ INPUTS = ["30", "0700", "0900", " 30 ", "0", "-5", "abc", "", "1e3", "9999999999
 # DELEGATE_JOB_/RECONCILE_…:-<digit>…}` that is neither here nor normalised
 # through INT_KNOBS, so a new knob has to be classified to land.
 # Issue #17 item 4: MAM_DELEGATE_FAIL_CLOSED and MAM_STALL_WATCH are read by
-# lib_py/delegation/wrapper.py now, not by shell code.
+# lib_py/delegation/wrapper.py now, not by shell code; MAM_PANE_CLAIM_WAIT_S
+# (minimum 1, default 120) by lib_py/pane_lock.py (test_issue50_pane_lock P5).
 NOT_ARITHMETIC = {
     "MAM_ALLOW_UNSIGNED_DISPATCH": "string compare", "MAM_DELEGATE_JOB_NO_MAIN": "string compare",
     "MAM_NO_AUTOSTART": "string compare", "MAM_SHIM_DEBUG": "string compare",

tests/test_issue50_dispatch.py

@@ -1,9 +1,12 @@
 """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,
+  signal age x ack budget x session spelling), FIFO release among live
-  the waiter's own signal written through the registry, the bound, a dead
+  pending waiters, the admission persisted on the released record (review
-  wrapper, and finalize-end's `registry_wait` phase closing as session_busy.
+  aa1bff92 F1) and held alive after the answer (F2), the waiter and its hold
+  pinned to one attempt and session (review 55219be1 R1), 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.
@@ -16,6 +19,7 @@ import json
 import os
 import subprocess
 import sys
+import threading
 import time
 from pathlib import Path
 
@@ -24,6 +28,8 @@ 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"
+REGISTRY_PY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "registry.py"
+WINDOW = 30   # DISPATCH_SIGNAL_EVERY (10) x DISPATCH_SIGNAL_MISSES (3), pinned here
 if str(SKILLS) not in sys.path:
     sys.path.insert(0, str(SKILLS))
 
@@ -49,6 +55,21 @@ def _edit(reg, jid, **fields):
     path.write_text(json.dumps(rec))
 
 
+def _rec(reg, jid):
+    return json.loads((reg / f"{jid}.json").read_text())
+
+
+def _at(text):
+    return calendar.timegm(time.strptime(text, "%Y-%m-%dT%H:%M:%SZ"))
+
+
+def _register_as(reg, jid, **over):
+    """A record under a chosen id (the file name is what open_jobs keys on)."""
+    real = _register(reg, job_id=jid, **over)
+    os.rename(reg / f"{real}.json", reg / f"{jid}.json")
+    return jid
+
+
 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")
@@ -63,22 +84,26 @@ def _ns(reg, job, session="herdr:s", wait_s=0, start_time=None, timeout=3600, wa
 # U1 - the open-job table
 # ---------------------------------------------------------------------------
 
-@pytest.mark.parametrize("status,age,signal,want", [
+@pytest.mark.parametrize("status,age,signal,ack,want", [
-    ("running", 10, None, "live"),          # an event idle_timeout_sec (120) ago at most
+    ("running", 10, None, 1, "live"),       # an event idle_timeout_sec (120) ago at most
-    ("running", 121, None, "stale"),
+    ("running", 121, None, 1, "stale"),
-    ("running", None, None, "stale"),       # running with no events.log at all
+    ("running", None, None, 1, "stale"),    # running with no events.log at all
-    ("pending", 10, None, "live"),          # created within ack_timeout_sec (180)
+    ("pending", 10, None, 180, "live"),     # never signalled: created within its ack budget
-    ("pending", 181, None, "stale"),
+    ("pending", 31, None, 180, "live"),     # ... past the signal window too (review 55219be1 R2:
-    ("pending", 181, 5, "live"),            # old record, fresh dispatch_wait_at
+    ("pending", 180, None, 180, "live"),    #     create --onboard waits that long for `started`)
-    ("pending", 181, 181, "stale"),
+    ("pending", 181, None, 180, "stale"),
-    ("pending", 10, 181, "stale"),          # the signal, once present, is what counts
+    ("pending", 10, None, 1, "stale"),      # never signalled, past its ack budget: a dead producer
-    ("completed", 10, None, None),
+    ("pending", 181, 5, 1, "live"),         # old record, fresh dispatch_wait_at
-    ("error", 10, None, None),
+    ("pending", 181, 31, 180, "stale"),     # signalled: the window (30), not the ack budget
-    ("cancelled", 10, None, None),
+    ("pending", 10, 31, 180, "stale"),      # the signal, once present, is what counts
+    ("pending", 1, 2, 1, "live"),           # ack_timeout_sec (1 here) is not the signalled window
+    ("completed", 10, None, 1, None),
+    ("error", 10, None, 1, None),
+    ("cancelled", 10, None, 1, None),
 ])
-def test_u1_open_jobs_table(tmp_path, status, age, signal, want):
+def test_u1_open_jobs_table(tmp_path, status, age, signal, ack, want):
     NOW = int(time.time())   # the running check reads the real clock (recent_disk_heartbeat)
-    other = _register(tmp_path, status=status)
+    other = _register(tmp_path, status=status, ack=ack)
     if status == "running" and age is not None:
         _event(tmp_path, other, NOW - age)
     if status == "pending":
@@ -90,11 +115,32 @@ def test_u1_open_jobs_table(tmp_path, status, age, signal, want):
     got = {j for _c, j, _s in live}, {j for j, _r in stale}
     if want == "live":
         assert got == ({other}, set())
+        if status == "pending":   # never signalled: injected by its producer, a holder; signalled: a waiter in the FIFO
+            assert live[0][2] == ("injected" if signal is None else "pending")
     elif want == "stale":
         assert got == (set(), {other}), got
         assert stale[0][1].startswith(status + ", no ")
+        if status == "pending":
+            assert stale[0][1].endswith(f"for {WINDOW}s" if signal else f"no started within {ack}s")
     else:
         assert got == (set(), set())
+    assert decide.DEFAULT_ACK_TIMEOUT_SEC == registry.DEFAULT_ACK_TIMEOUT_SEC
+
+
+@pytest.mark.parametrize("token_match,signal,want", [(True, 5, "admitted"), (False, 5, "pending"), (True, 31, "stale")])
+def test_u1d_an_admission_is_scoped_to_the_attempt_and_lives_by_the_signal(tmp_path, token_match, signal, want):
+    NOW = int(time.time())
+    other = _register(tmp_path)
+    token = _rec(tmp_path, other)["dispatch_token"] if token_match else "a-previous-attempt"
+    _edit(tmp_path, other, dispatch_wait_at=_iso(NOW - signal), dispatch_admitted_at=_iso(NOW - signal),
+          dispatch_admitted_token=token)
+    me = _register(tmp_path)
+    live, stale = decide.open_jobs(str(tmp_path), "herdr:s", me, now=NOW)
+    if want == "stale":
+        assert (live, [j for j, _r in stale]) == ([], [other])
+    else:
+        assert [(j, st) for _c, j, st in live] == [(other, want)]
+    assert decide.DISPATCH_SIGNAL_WINDOW == WINDOW
 
 
 @pytest.mark.parametrize("record_session,asked", [("herdr:s", "s"), ("s", "herdr:s"), ("herdr:s", "herdr:s")])
@@ -125,8 +171,8 @@ def test_u2_clear_only_when_no_live_running_and_this_job_is_the_earliest_pending
     # 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)
+    # an earlier live pending waiter B (signalled: a waiter is on it): this job waits behind it (busy at the bound)
-    b = _register(tmp_path, created_at=_iso(time.time() - 30))
+    b = _register(tmp_path, created_at=_iso(time.time() - 30), dispatch_wait_at=_iso(time.time()))
     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))
@@ -135,20 +181,82 @@ def test_u2_clear_only_when_no_live_running_and_this_job_is_the_earliest_pending
 
 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")
+    _register_as(tmp_path, "aaaa0001", created_at=same)
-    second = _register(tmp_path, created_at=same, job_id="aaaa0002")
+    _register_as(tmp_path, "aaaa0002", created_at=same, dispatch_wait_at=same)   # a live waiter
-    # 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"
 
 
+# review 5a09a13f F-A: a live pending record nobody signals on is one its
+# producer injected without a waiter (create --onboard, a job sent by hand);
+# the prompt is in the pane, so it holds the session - never a FIFO peer that
+# a lower job id in the same second, or an earlier second, could pass
+
+@pytest.mark.parametrize("onboarding,waiter", [("ffffffff", "00000001"), ("00000001", "ffffffff")])
+def test_u2h_an_injected_record_holds_the_session_whatever_its_place_in_the_fifo(tmp_path, onboarding, waiter):
+    same = _iso(time.time())
+    _register_as(tmp_path, onboarding, created_at=same)                 # unsignalled, ack 180: live
+    _register_as(tmp_path, waiter, created_at=same)
+    assert decide.wait_open_jobs(_ns(tmp_path, waiter)) == f"busy {onboarding}"
+    assert "dispatch_admitted_token" not in _rec(tmp_path, waiter) and "dispatch_wait_at" in _rec(tmp_path, waiter)
+    assert "dispatch_wait_at" not in _rec(tmp_path, onboarding)
+    # registered after the waiter (which was still in its preflight): the same
+    _edit(tmp_path, onboarding, created_at=_iso(_at(same) + 1))
+    assert decide.wait_open_jobs(_ns(tmp_path, waiter)) == f"busy {onboarding}"
+    # control: the same record with a waiter's signal on it is a FIFO peer
+    _edit(tmp_path, onboarding, created_at=same, dispatch_wait_at=same)
+    assert decide.wait_open_jobs(_ns(tmp_path, waiter)) == ("clear" if waiter < onboarding else f"busy {onboarding}")
+
+
 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))
+    b = _register(tmp_path, ack=5, created_at=_iso(time.time() - 60), dispatch_wait_at=_iso(time.time() - WINDOW - 1))
     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
+    assert f"stale open job {b} ignored (pending, no dispatch signal for {WINDOW}s)" in capsys.readouterr().err
+
+
+# review aa1bff92 F1: `clear` is an admission the record carries, so a candidate
+# that would sort first - a lower id in the same second, a redispatched older
+# loop job - waits for the admitted job until it ends
+
+def test_u2e_a_lower_id_arriving_after_clear_waits_for_the_admitted_job(tmp_path):
+    same = _iso(time.time())
+    _register_as(tmp_path, "ffff0001", created_at=same)
+    assert decide.wait_open_jobs(_ns(tmp_path, "ffff0001")) == "clear"
+    _register_as(tmp_path, "0000000a", created_at=same)
+    assert decide.wait_open_jobs(_ns(tmp_path, "0000000a")) == "busy ffff0001"
+    assert decide.wait_open_jobs(_ns(tmp_path, "0000000a", wait_s=2)) == "busy ffff0001", "still, at the bound"
+    # the admitted job's `started` keeps the hold (running, with an event)...
+    _edit(tmp_path, "ffff0001", status="running")
+    _event(tmp_path, "ffff0001", time.time(), "started")
+    assert decide.wait_open_jobs(_ns(tmp_path, "0000000a")) == "busy ffff0001"
+    # ... and its end releases the next one
+    _edit(tmp_path, "ffff0001", status="completed")
+    assert decide.wait_open_jobs(_ns(tmp_path, "0000000a")) == "clear"
+
+
+def test_u2f_an_older_loop_attempt_redispatched_after_another_admission_waits(tmp_path):
+    old = _register(tmp_path, status="completed")
+    time.sleep(1.1)
+    cur = _register(tmp_path)
+    assert decide.wait_open_jobs(_ns(tmp_path, cur)) == "clear"
+    res = subprocess.run([sys.executable, str(REGISTRY_PY), "--registry-dir", str(tmp_path), "update", "--job", old,
+                          "--status", "pending", "--attempt-id", "2"], capture_output=True, text=True, timeout=60)
+    assert res.returncode == 0 and _rec(tmp_path, old)["status"] == "pending", res.stdout + res.stderr
+    assert _rec(tmp_path, old)["created_at"] < _rec(tmp_path, cur)["created_at"], "it would sort first"
+    assert decide.wait_open_jobs(_ns(tmp_path, old)) == f"busy {cur}"
+    _edit(tmp_path, cur, status="completed")
+    assert decide.wait_open_jobs(_ns(tmp_path, old)) == "clear"
+
+
+def test_u2g_an_admission_whose_owner_stopped_signalling_expires(tmp_path, capsys):
+    a = _register(tmp_path)
+    assert decide.wait_open_jobs(_ns(tmp_path, a)) == "clear"
+    b = _register(tmp_path)
+    assert decide.wait_open_jobs(_ns(tmp_path, b)) == f"busy {a}"
+    _edit(tmp_path, a, dispatch_wait_at=_iso(time.time() - WINDOW - 1))     # its hold is gone
+    assert decide.wait_open_jobs(_ns(tmp_path, b)) == "clear"
+    assert f"stale open job {a} ignored (pending, no dispatch signal for {WINDOW}s)" in capsys.readouterr().err
 
 
 def test_u2d_the_bound_is_the_remaining_wall_clock(tmp_path):
@@ -185,15 +293,194 @@ def test_u3_the_waiter_signals_on_its_own_record(tmp_path):
     assert me in {j for _c, j, _s in live}
 
 
-def test_u3b_the_signal_is_a_registry_merge(tmp_path, monkeypatch):
+def test_u3d_clear_is_persisted_on_the_record_scoped_to_its_attempt(tmp_path):
     me = _register(tmp_path)
+    t0 = int(time.time())
+    assert decide.wait_open_jobs(_ns(tmp_path, me)) == "clear"
+    rec = _rec(tmp_path, me)
+    assert rec["status"] == "pending" and rec["dispatch_admitted_token"] == rec["dispatch_token"]
+    assert _at(rec["dispatch_admitted_at"]) >= t0 and _at(rec["dispatch_wait_at"]) >= t0, "the signal is renewed with it"
+    assert not (tmp_path / f"{me}.events.log").exists(), "an admission is not the agent's `started`"
+    later = _register(tmp_path)
+    assert decide.open_jobs(str(tmp_path), "herdr:s", later)[0] == [(rec["created_at"], me, "admitted")]
+    # a redispatch (new attempt, new token) voids it
+    res = subprocess.run([sys.executable, str(REGISTRY_PY), "--registry-dir", str(tmp_path), "update", "--job", me,
+                          "--status", "pending", "--attempt-id", "2"], capture_output=True, text=True, timeout=60)
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert _rec(tmp_path, me)["dispatch_token"] != rec["dispatch_token"]
+    assert decide.open_jobs(str(tmp_path), "herdr:s", later)[0] == [(rec["created_at"], me, "pending")]
+
+
+def test_u3e_the_admission_is_rechecked_under_the_registry_lock(tmp_path, monkeypatch):
+    # the waiter's read said "first in line"; by the time its merge holds the
+    # lock another waiter was admitted: the guard refuses, the wait goes on
+    me = _register(tmp_path)
+    other = _register(tmp_path)
+    real_open_jobs = decide.open_jobs
+    calls = []
+
+    def open_jobs(registry_dir, session, job_id, now=None):
+        calls.append(job_id)
+        live, stale = real_open_jobs(registry_dir, session, job_id, now)
+        if len(calls) >= 2:   # the guard's read (and every poll after it)
+            live = [(c, j, "admitted") for c, j, _st in live]
+        return live, stale
+
+    monkeypatch.setattr(decide, "open_jobs", open_jobs)
+    assert decide.wait_open_jobs(_ns(tmp_path, me, wait_s=1)) == f"busy {other}"
+    assert "dispatch_admitted_token" not in _rec(tmp_path, me)
+    assert len(calls) >= 2
+
+
+def test_u3c_a_terminal_self_record_cannot_carry_the_signal(tmp_path):
+    # why the loop path resets its record to pending before it waits (s1): a
+    # terminal record is not this attempt's pending one, the waiter ends at
+    # once without a write (superseded) and is in no other waiter's FIFO
+    a = _register(tmp_path, status="running")
+    _event(tmp_path, a, time.time())
+    me = _register(tmp_path, status="completed")
+    assert decide.wait_open_jobs(_ns(tmp_path, me, wait_s=1)) == "superseded"
+    after = json.loads((tmp_path / f"{me}.json").read_text())
+    assert after["status"] == "completed" and "dispatch_wait_at" not in after
+    later = _register(tmp_path)
+    live, _stale = decide.open_jobs(str(tmp_path), "herdr:s", later)
+    assert me not in {j for _c, j, _s in live}
+
+
+def test_u3b_the_signal_is_a_registry_merge_guarded_by_the_pinned_identity(tmp_path, monkeypatch):
+    from lib_py.delegation import engine
+    me = _register(tmp_path)
+    tok = _rec(tmp_path, me)["dispatch_token"]
     seen = {}
-    monkeypatch.setattr(mqtt_common, "update_job_status",
+
-                        lambda job_id, registry_dir, _source="unknown", **fields: seen.update(
+    def transition(job_id, request, registry_dir=None, source="unknown", guard=None, fields=None, **kw):
-                            job=job_id, reg=registry_dir, source=_source, fields=fields))
+        seen.update(job=job_id, request=request, reg=registry_dir, source=source, guard=guard, fields=fields)
-    decide._signal_dispatch_wait(str(tmp_path), me)
+        return argparse.Namespace(committed=True)
-    assert seen["job"] == me and seen["reg"] == str(tmp_path) and seen["source"] == "dispatch_wait"
+
-    assert set(seen["fields"]) == {"dispatch_wait_at"}
+    monkeypatch.setattr(engine, "transition", transition)
+    assert decide._signal_dispatch_wait(str(tmp_path), me, tok, "herdr:s") is True
+    assert seen["job"] == me and seen["request"] == "merge" and seen["reg"] == str(tmp_path)
+    assert seen["source"] == "dispatch_wait" and set(seen["fields"]) == {"dispatch_wait_at"}
+    # the guard, run under the registry lock, is the waiter's identity (review 55219be1 R1)
+    mine = {"status": "pending", "dispatch_token": tok, "agent_session": "herdr:s"}
+    assert seen["guard"](mine) and seen["guard"](dict(mine, agent_session="s"))
+    assert not seen["guard"](dict(mine, dispatch_token="another-attempt"))
+    assert not seen["guard"](dict(mine, agent_session="herdr:t"))
+    assert not seen["guard"](dict(mine, status="running"))
+    # the hold's renewal also requires its admission to stand
+    decide._signal_dispatch_wait(str(tmp_path), me, tok, "herdr:s", admitted=True)
+    assert not seen["guard"](mine) and seen["guard"](dict(mine, dispatch_admitted_token=tok))
+    assert not seen["guard"](dict(mine, dispatch_admitted_token="another-attempt"))
+
+
+def _waiter(reg, job, session="herdr:s", wait_s="20", token=None):
+    cmd = [sys.executable, "-m", "lib_py.delegation.wrapper", "wait-open-jobs", "--job", job, "--agent-session",
+           session, "--registry-dir", str(reg), "--wait-s", wait_s, "--timeout", "3600", "--watch-pid", str(os.getpid())]
+    if token:
+        cmd += ["--dispatch-token", token]
+    return subprocess.Popen(cmd, cwd=str(SKILLS), env=dict(os.environ, PYTHONPATH=str(SKILLS)), stdout=subprocess.PIPE,
+                            stderr=subprocess.PIPE, text=True)
+
+
+def _redispatch(reg, jid, *extra):
+    res = subprocess.run([sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg), "update", "--job", jid,
+                          "--status", "pending", "--attempt-id", "2", *extra], capture_output=True, text=True, timeout=60)
+    assert res.returncode == 0, res.stdout + res.stderr
+
+
+def test_u3f_a_waiter_redispatched_to_another_session_is_superseded_and_admits_nothing(tmp_path):
+    # review 55219be1 R1: A runs in S, C runs in T, B waits behind A in S; B is
+    # redispatched (attempt 2) to T; A ends. The attempt-1 waiter must not
+    # admit attempt 2 against S's occupancy: it ends, writes nothing
+    a = _register(tmp_path, status="running")
+    _event(tmp_path, a, time.time())
+    c = _register(tmp_path, session="herdr:t", status="running")
+    _event(tmp_path, c, time.time())
+    b = _register(tmp_path)
+    old = _rec(tmp_path, b)["dispatch_token"]
+    proc = _waiter(tmp_path, b, token=old)
+    try:
+        deadline = time.time() + 5
+        while time.time() < deadline and not _rec(tmp_path, b).get("dispatch_wait_at"):
+            time.sleep(0.05)
+        _redispatch(tmp_path, b, "--agent-session", "herdr:t")
+        moved = _rec(tmp_path, b)
+        assert moved["dispatch_token"] != old and moved["agent_session"] == "herdr:t"
+        _edit(tmp_path, a, status="completed")
+        out, err = proc.communicate(timeout=10)
+        assert out.strip() == "superseded", (out, err)
+        rec = _rec(tmp_path, b)
+        assert "dispatch_admitted_token" not in rec and rec["status"] == "pending" and rec["attempt_id"] == 2
+        assert rec["dispatch_wait_at"] == moved["dispatch_wait_at"] and rec["transition_seq"] == moved["transition_seq"], \
+            "nothing was renewed after the redispatch"
+        assert _rec(tmp_path, c)["status"] == "running"
+        assert decide.wait_open_jobs(_ns(tmp_path, b, "herdr:t")) == f"busy {c}", "T is what attempt 2 is judged by"
+    finally:
+        if proc.poll() is None:
+            proc.kill()
+            proc.communicate()
+
+
+def test_u3g_a_waiter_whose_attempt_changed_in_place_leaves_the_admission_to_the_new_dispatcher(tmp_path):
+    a = _register(tmp_path, status="running")
+    _event(tmp_path, a, time.time())
+    b = _register(tmp_path)
+    old = _rec(tmp_path, b)["dispatch_token"]
+    proc = _waiter(tmp_path, b, token=old)
+    try:
+        deadline = time.time() + 5
+        while time.time() < deadline and not _rec(tmp_path, b).get("dispatch_wait_at"):
+            time.sleep(0.05)
+        _redispatch(tmp_path, b)
+        _edit(tmp_path, a, status="completed")
+        out, err = proc.communicate(timeout=10)
+        assert out.strip() == "superseded", (out, err)
+        assert "dispatch_admitted_token" not in _rec(tmp_path, b)
+        new = _rec(tmp_path, b)["dispatch_token"]
+        assert decide.wait_open_jobs(_ns(tmp_path, b)) == "clear", "attempt 2's own dispatcher admits it"
+        assert _rec(tmp_path, b)["dispatch_admitted_token"] == new != old
+    finally:
+        if proc.poll() is None:
+            proc.kill()
+            proc.communicate()
+
+
+def test_u3i_an_attempt_already_running_under_its_own_token_is_clear_not_superseded(tmp_path):
+    # the Issue #14 registry ack: the agent's `started` for this attempt lands
+    # before the prompt goes out (test_issue30_stall_watch_stop). Nothing to
+    # wait for, nothing foreign: `clear` at once, no admission written
+    a = _register(tmp_path, status="running")
+    _event(tmp_path, a, time.time())
+    me = _register(tmp_path)
+    tok = _rec(tmp_path, me)["dispatch_token"]
+    _edit(tmp_path, me, status="running")
+    before = _rec(tmp_path, me)
+    t0 = time.time()
+    assert decide.wait_open_jobs(_ns(tmp_path, me, wait_s=30)) == "clear"
+    assert time.time() - t0 < 5 and _rec(tmp_path, me) == before, "no wait, no write"
+    res = _cli(tmp_path, "wait-open-jobs", "--job", me, "--agent-session", "herdr:s", "--registry-dir", str(tmp_path),
+               "--wait-s", "30", "--timeout", "3600", "--dispatch-token", tok, "--hold")
+    assert res.stdout.strip() == "clear" and not _hold_pids(me), res.stderr
+    # running under another attempt's token, or in another session, is still someone else's
+    assert decide.wait_open_jobs(_ns(tmp_path, me, session="herdr:t")) == "superseded"
+    _edit(tmp_path, me, dispatch_token="another-attempt")
+    res = _cli(tmp_path, "wait-open-jobs", "--job", me, "--agent-session", "herdr:s", "--registry-dir", str(tmp_path),
+               "--wait-s", "30", "--timeout", "3600", "--dispatch-token", tok)
+    assert res.stdout.strip() == "superseded", res.stderr
+    _edit(tmp_path, me, dispatch_token=tok, status="completed")
+    assert decide.wait_open_jobs(_ns(tmp_path, me)) == "superseded"
+
+
+def test_u3h_the_pinned_token_is_the_wrappers_not_the_records(tmp_path):
+    # the record's token changed before the waiter started (a redispatch raced
+    # the wrapper): the waiter given the old token is superseded at once
+    me = _register(tmp_path)
+    old = _rec(tmp_path, me)["dispatch_token"]
+    _redispatch(tmp_path, me)
+    proc = _waiter(tmp_path, me, token=old)
+    out, err = proc.communicate(timeout=10)
+    assert out.strip() == "superseded" and "dispatch_admitted_token" not in _rec(tmp_path, me), (out, err)
+    assert decide.wait_open_jobs(_ns(tmp_path, me)) == "clear"   # the current token, read once at the start
 
 
 # ---------------------------------------------------------------------------
@@ -209,10 +496,105 @@ 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))
+    c = _register(tmp_path)                                  # within its ack budget, no waiter on it
     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
+    assert f"{a}  running" in res.stdout
+    assert f"{b}  stale     pending, no dispatch signal and no started within 180s" in res.stdout
+    assert f"{c}  injected" in res.stdout
+
+
+def _hold_pids(jid):
+    out = subprocess.run(["pgrep", "-f", f"hold-admission --job {jid}"], capture_output=True, text=True).stdout
+    return [int(p) for p in out.split()]
+
+
+def test_u4c_the_wait_holds_its_admission_until_the_record_leaves_pending(tmp_path):
+    # review aa1bff92 F2: after `clear` the waiter ends with its answer; the
+    # hold renews the admitted record's signal until the agent's `started`
+    me = _register(tmp_path)
+    try:
+        res = _cli(tmp_path, "wait-open-jobs", "--job", me, "--agent-session", "herdr:s", "--registry-dir",
+                   str(tmp_path), "--wait-s", "0", "--timeout", "3600", "--watch-pid", str(os.getpid()), "--hold")
+        assert res.stdout.strip() == "clear", res.stderr
+        first = _rec(tmp_path, me)["dispatch_wait_at"]
+        assert _hold_pids(me), "a detached hold is running"
+        deadline = time.time() + decide.DISPATCH_SIGNAL_EVERY + 5
+        while time.time() < deadline and _rec(tmp_path, me)["dispatch_wait_at"] == first:
+            time.sleep(0.5)
+        assert _at(_rec(tmp_path, me)["dispatch_wait_at"]) >= _at(first) + decide.DISPATCH_SIGNAL_EVERY - 1
+        assert _rec(tmp_path, me)["status"] == "pending" and not (tmp_path / f"{me}.events.log").exists()
+        _edit(tmp_path, me, status="running")   # the agent's `started`
+        deadline = time.time() + 5
+        while time.time() < deadline and _hold_pids(me):
+            time.sleep(0.2)
+        assert not _hold_pids(me), "released with the record"
+    finally:
+        for pid in _hold_pids(me):
+            os.kill(pid, 9)
+
+
+def _hold(reg, jid, token, *extra, session="herdr:s"):
+    return _cli(reg, "hold-admission", "--job", jid, "--agent-session", session, "--dispatch-token", token,
+                "--registry-dir", str(reg), "--timeout", "3600", *extra)
+
+
+def test_u4d_the_hold_ends_with_its_wrapper_and_without_an_admission(tmp_path):
+    me = _register(tmp_path)
+    tok = _rec(tmp_path, me)["dispatch_token"]
+    res = _hold(tmp_path, me, tok)
+    assert res.stdout.strip() == "released", res.stderr    # never admitted: nothing to hold
+    assert decide.wait_open_jobs(_ns(tmp_path, me)) == "clear"
+    gone = subprocess.Popen(["true"])
+    gone.wait()
+    t0 = time.time()
+    res = _hold(tmp_path, me, tok, "--watch-pid", str(gone.pid))
+    assert res.stdout == "" and "is gone" in res.stderr and time.time() - t0 < 10, res.stdout + res.stderr
+
+
+def test_u4e_a_hold_that_starts_late_holds_only_the_attempt_it_was_given(tmp_path):
+    # review 55219be1 R1: the token changed (or the record moved) between the
+    # admission and the hold's start - the hold is released without a write
+    me = _register(tmp_path)
+    assert decide.wait_open_jobs(_ns(tmp_path, me)) == "clear"
+    admitted = _rec(tmp_path, me)
+    tok = admitted["dispatch_token"]
+    res = _hold(tmp_path, me, tok, session="herdr:t")                      # not the session it admitted in
+    assert res.stdout.strip() == "released" and _rec(tmp_path, me) == admitted, res.stderr
+    _redispatch(tmp_path, me)
+    moved = _rec(tmp_path, me)
+    assert moved["dispatch_token"] != tok
+    res = _hold(tmp_path, me, tok)
+    assert res.stdout.strip() == "released" and _rec(tmp_path, me) == moved, res.stderr
+    assert "dispatch_admitted_token" in moved and moved["dispatch_admitted_token"] == tok, \
+        "attempt 1's admission is left as it is - and names no current token (u3d)"
+
+
+def test_u4f_a_running_hold_is_released_by_a_redispatch_and_renews_nothing_after_it(tmp_path):
+    me = _register(tmp_path)
+    assert decide.wait_open_jobs(_ns(tmp_path, me)) == "clear"
+    tok = _rec(tmp_path, me)["dispatch_token"]
+    first = _rec(tmp_path, me)["dispatch_wait_at"]
+    proc = subprocess.Popen([sys.executable, "-m", "lib_py.delegation.wrapper", "hold-admission", "--job", me,
+                             "--agent-session", "herdr:s", "--dispatch-token", tok, "--registry-dir", str(tmp_path),
+                             "--timeout", "3600", "--watch-pid", str(os.getpid())],
+                            cwd=str(SKILLS), env=dict(os.environ, PYTHONPATH=str(SKILLS)), stdout=subprocess.PIPE,
+                            stderr=subprocess.PIPE, text=True)
+    try:
+        deadline = time.time() + decide.DISPATCH_SIGNAL_EVERY + 5
+        while time.time() < deadline and _rec(tmp_path, me)["dispatch_wait_at"] == first:
+            time.sleep(0.5)
+        assert _rec(tmp_path, me)["dispatch_wait_at"] != first, "it renewed while the admission stood"
+        _redispatch(tmp_path, me)
+        moved = _rec(tmp_path, me)
+        out, err = proc.communicate(timeout=10)
+        assert out.strip() == "released", (out, err)
+        assert _rec(tmp_path, me) == moved, "nothing written after the redispatch"
+    finally:
+        if proc.poll() is None:
+            proc.kill()
+            proc.communicate()
 
 
 def test_u4b_wait_open_jobs_stops_with_its_wrapper(tmp_path):
@@ -260,14 +642,25 @@ def test_u5_finalize_end_registry_wait_is_a_session_busy_cancel(tmp_path):
 
 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
+    # ... each with the token it read for this attempt, pinned (review 55219be1 R1)
-    assert src.count('_registry_wait "$JOB_ID" "$current_session"') == 1
+    assert src.count('_registry_wait "$JOB_ID" "$AGENT_SESSION" "$_disp_tok"') == 1
-    for block in src.split("wait_for_agent_idle ")[1:]:
+    assert src.count('_registry_wait "$JOB_ID" "$current_session" "$_disp_tok"') == 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"')
+    # the loop path waits on the record already reset to pending for this
+    # phase (review F1): a terminal self record cannot carry the signal (u3c)
+    assert src.count('--status "pending"') == 1
+    assert src.index('--status "pending"') < loop
+    # ... and reads the token that reset issued before it waits: a busy or
+    # idle_wait close from phase two on must carry this attempt's token
+    tok = src.index('_disp_tok="$(', src.index('--status "pending"'))
+    assert src.index('--status "pending"') < tok < loop
     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
+    assert "--hold" in body, "the admission is held alive after the answer (u4c)"
+    assert '${tok:+--dispatch-token "$tok"}' in body
+    superseded = body.split("superseded)")[1].split(";;")[0]
+    assert "exit 5" in superseded and "finalize_unfinished" not in superseded, "left to its owner: no close"

tests/test_issue50_dispatch_herdr.py

@@ -18,6 +18,8 @@ so the herdr idle wait stays and the registry wait runs before it.
 """
 from __future__ import annotations
 
+import calendar
+import fcntl
 import json
 import os
 import re
@@ -37,6 +39,7 @@ 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))
@@ -194,6 +197,9 @@ def _make_box(sess: str) -> Box:
     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)))
@@ -209,20 +215,38 @@ def _start_pane(box: Box, name=SESSION, cwd=None) -> None:
     assert box.agent_status(name) == "idle", box.herdr("agent", "list").stdout
 
 
-def _teardown(box: Box) -> None:
+def _kill_servers(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
+
+
+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
@@ -311,7 +335,8 @@ def test_c3_a_dead_running_record_does_not_block_the_session(box):
     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"
+    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"]
 
 
 # ---------------------------------------------------------------------------
@@ -416,6 +441,9 @@ def test_c7_two_waiters_run_one_after_the_other_and_neither_is_cancelled(box):
     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"
 
@@ -433,19 +461,158 @@ def test_c8_a_dead_waiter_is_skipped_once_its_signal_is_stale(box):
     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(40, box.publish, c, "started")           # after B's signal window (30 s) has passed
-    _after(16, box.publish, c, "completed", "c done")
+    _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 5s)" in err, 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 _restamp(box: Box, jid: str, created_at: str) -> None:
+    """Give a record the test owns another created_at, atomically: a wrapper
+    scanning the registry must never find the file half-written (an unreadable
+    record is skipped, which would let it through)."""
+    rec = box.record(jid)
+    rec["created_at"] = created_at
+    tmp = box.jobs / f"{jid}.json.tmp"
+    tmp.write_text(json.dumps(rec))
+    os.replace(tmp, box.jobs / f"{jid}.json")
+
+
+def test_c10_a_pending_onboarding_within_its_ack_budget_blocks_the_session_past_the_signal_window(box):
+    # review 5a09a13f F-A: the onboarding and the first job are registered in
+    # the same second (create registers, then injects; the job's wrapper
+    # registers a moment later), where the FIFO's (created_at, job_id) once
+    # let a lower id pass the injected onboarding. The tie is made, not timed
+    # (reviews 98b31bb2 / 352066ec / 244b8072 F-B): once the first job's record
+    # exists, A takes its created_at - A has no signal and no admission, and the
+    # pane already holds its prompt, so nothing else depends on the stamp. The
+    # second job comes a different second.
+    a = box.register(ack_timeout_sec=180)
+    known = box.job_ids()
+    created = 0
+    for label in ("before", "after"):
+        if label == "after":
+            time.sleep(max(0.0, created + 31.5 - time.time()))  # A is now older than the signal window
+        proc = box.submit(prompt="work " + label, MAM_DISPATCH_WAIT_S="3")
+        if label == "before":
+            prompt = box.herdr("agent", "prompt", SESSION, "Onboarding " + a)
+            assert prompt.returncode == 0, prompt.stdout + prompt.stderr
+        b = box.wait_new_job(known)
+        known.add(b)
+        if label == "before":
+            _restamp(box, a, box.record(b)["created_at"])
+            created = calendar.timegm(time.strptime(box.record(a)["created_at"], "%Y-%m-%dT%H:%M:%SZ"))
+        print(label, "created", a, box.record(a)["created_at"], b, box.record(b)["created_at"],
+              "at +%.2fs" % (time.time() - created))
+        assert (box.record(b)["created_at"] > box.record(a)["created_at"]) == (label == "after"), label
+        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
+    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
 # ---------------------------------------------------------------------------
@@ -472,32 +639,71 @@ def _drive_onboardings(box: Box, names: list, delays: list, result: dict):
             _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)]
-    procs = []
     try:
-        for i, name in enumerate(names, 1):
+        workspaces = []
-            ws = b.home / f"ws{i}"
+        for i in (1, 2, 3):
-            ws.mkdir()
+            (b.home / f"ws{i}").mkdir()
-            out = tempfile.TemporaryFile("w+", dir=str(b.home))
+            workspaces.append(b.home / f"ws{i}")
-            err = tempfile.TemporaryFile("w+", dir=str(b.home))
+        outcome, result, elapsed = _three_creates(b, names, workspaces)
-            procs.append((name, subprocess.Popen(
+        t0 = time.time() - elapsed
-                ["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
@@ -519,6 +725,118 @@ def test_a_three_concurrent_creates_with_onboarding(local_mqtt_broker):
         _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_s2_an_ambient_and_a_named_call_share_the_servers_pane_claim():
+    """review e8e471ff R1: a call with the ambient target - no -L, HERDR_SESSION_NAME
+    empty or `default` - from a pane of server S is answered by S (the pane's
+    HERDR_SESSION / HERDR_SOCKET_PATH), so it claims under S's lock, the one a
+    call naming S takes: held, both wait; racing, they take distinct panes."""
+    b = _make_box(f"t50-{uuid.uuid4().hex[:8]}")
+    ambient = dict(HERDR_SESSION_NAME="default", HERDR_SESSION=b.sess,
+                   HERDR_SOCKET_PATH=str(b.home / ".config/herdr/sessions" / b.sess / "herdr.sock"))
+    new_session = 'herdr new-session -d -s {name} -c "$WORKSPACE_ROOT" "claude --dangerously-skip-permissions"'
+    try:
+        assert b.sh("herdr agent list").returncode == 0      # the server is up before the race
+        lock = b.home / ".local/state/mam/pane-lock" / f"{b.sess}.lock"
+        lock.parent.mkdir(parents=True, exist_ok=True)
+        with lock.open("a") as held:
+            fcntl.flock(held, fcntl.LOCK_EX)
+            for env in ({}, ambient):
+                res = b.sh(new_session.format(name="t50-held-creator-claude"), MAM_PANE_CLAIM_WAIT_S="1", **env)
+                assert res.returncode == 1 and "another pane claim" in res.stderr, (env, res.stdout, res.stderr)
+            assert json.loads(b.herdr("agent", "list").stdout)["result"]["agents"] == []
+        _hold_pane_creation(b, 8)
+        procs = []
+        for name, env in (("t50-named-creator-claude", {}), ("t50-ambient-creator-claude", ambient)):
+            if procs:
+                time.sleep(2)
+            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"\n' + new_session.format(name=name)],
+                                                 stdout=out, stderr=err, text=True, cwd=str(b.ctl), env=b.env(**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()}
+        assert "agent_pane_busy" not in "".join(o[1] + o[2] for o in outcome.values())
+        panes = _agent_panes(b)
+        assert sorted(panes) == sorted(n for n, *_ in procs), panes
+        assert len({p for p, _c in panes.values()}) == 2, ("two agents on one pane", panes)
+        assert all(Path(c).resolve() == b.ctl.resolve() for _p, c in panes.values()), panes
+    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)

tests/test_issue50_pane_lock.py

@@ -0,0 +1,187 @@
+"""Issue #50 follow-up: the herdr shim claims one pane at a time per server.
+
+- P: lib_py.pane_lock - the lock file per server, the acquire decision and its
+  bound, a lock that ends with its holder's process (no stale lock after
+  kill -9), the wait knob.
+- H: the race itself, on a real herdr server: test_issue50_dispatch_herdr.py
+  (test_s_three_concurrent_new_sessions_in_one_folder_take_distinct_panes,
+  test_a3_three_concurrent_creates_in_one_folder_take_distinct_panes).
+- the shim's use of the lock against the mock herdr: test_herdr_shim_contract.py h26.
+"""
+from __future__ import annotations
+
+import json
+import os
+import signal
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+SKILLS = REPO_ROOT / ".agents" / "skills"
+if str(SKILLS) not in sys.path:
+    sys.path.insert(0, str(SKILLS))
+
+from lib_py import pane_lock  # noqa: E402
+
+# a bash holder, as the shim is: fd 9 on the file, the lock taken by the helper, then it sleeps
+HOLDER = 'exec 9>>"$1"; python3 -m lib_py.pane_lock acquire --fd 9; echo held; sleep "${2:-600}"'
+
+
+def _holder(path: Path, seconds: str = "600", **env) -> subprocess.Popen:
+    p = subprocess.Popen(["bash", "-c", HOLDER, "holder", str(path), seconds], stdout=subprocess.PIPE, text=True,
+                         env=dict(os.environ, PYTHONPATH=str(SKILLS), **env))
+    assert p.stdout.readline().strip() == "locked"
+    assert p.stdout.readline().strip() == "held"
+    return p
+
+
+def _acquire(path: Path, **env) -> subprocess.CompletedProcess:
+    return subprocess.run(["bash", "-c", 'exec 9>>"$1"; python3 -m lib_py.pane_lock acquire --fd 9', "x", str(path)],
+                          capture_output=True, text=True, env=dict(os.environ, PYTHONPATH=str(SKILLS), **env), timeout=60)
+
+
+def test_p1_the_lock_file_is_per_server_under_the_sessions_home(tmp_path):
+    env = {"HOME": str(tmp_path)}
+    assert pane_lock.lock_path("t50-x", env) == str(tmp_path / ".local/state/mam/pane-lock/t50-x.lock")
+    assert pane_lock.lock_path("", env) == str(tmp_path / ".local/state/mam/pane-lock/default.lock")
+    assert pane_lock.lock_path("t50-x", {"HOME_DIR": str(tmp_path / "h")}).startswith(str(tmp_path / "h"))
+    assert pane_lock.lock_path("t50-x", env) != pane_lock.lock_path("t50-y", env)
+    res = subprocess.run([sys.executable, "-m", "lib_py.pane_lock", "path", "--session", "t50-x"], capture_output=True,
+                         text=True, env=dict(os.environ, PYTHONPATH=str(SKILLS), HOME=str(tmp_path)), timeout=60)
+    assert res.stdout.strip() == pane_lock.lock_path("t50-x", env), res.stderr
+
+
+# review e8e471ff R1: the ambient target (no -L, HERDR_SESSION_NAME empty or
+# `default`) is answered by the server of the pane the caller runs in - the
+# pane's HERDR_SESSION / HERDR_SOCKET_PATH say which - so that is the lock;
+# only outside a pane is it the default server's
+
+def test_p1b_the_ambient_target_locks_the_callers_server(tmp_path):
+    home = {"HOME": str(tmp_path)}
+    sock = {"HERDR_SOCKET_PATH": str(tmp_path / ".config/herdr/sessions/t50-x/herdr.sock")}
+    named = pane_lock.lock_path("t50-x", home)
+    for ambient in ("", "default", None):
+        assert pane_lock.lock_path(ambient, {**home, **sock}) == named, ambient
+        assert pane_lock.lock_path(ambient, {**home, "HERDR_SESSION": "t50-x"}) == named, ambient
+        assert pane_lock.lock_path(ambient, {**home, "HERDR_SESSION": "t50-x",          # HERDR_SESSION first
+                                             "HERDR_SOCKET_PATH": str(tmp_path / ".config/herdr/sessions/t50-y/herdr.sock")}) == named
+        assert pane_lock.lock_path(ambient, {**home, "HERDR_SOCKET_PATH": str(tmp_path / ".config/herdr/herdr.sock")}) \
+            == pane_lock.lock_path("default", home) == str(tmp_path / ".local/state/mam/pane-lock/default.lock")
+        assert pane_lock.lock_path(ambient, {**home, "HERDR_SOCKET_PATH": "/elsewhere/herdr.sock"}).endswith("/default.lock")
+        assert pane_lock.lock_path(ambient, home).endswith("/default.lock")
+    assert pane_lock.lock_path("t50-y", {**home, **sock}) != named, "a named target is that server's, wherever it is called from"
+    outside = {k: v for k, v in os.environ.items() if not k.startswith("HERDR_")}   # this test may run in a pane
+    res = subprocess.run([sys.executable, "-m", "lib_py.pane_lock", "path", "--session", ""], capture_output=True, text=True,
+                         env=dict(outside, PYTHONPATH=str(SKILLS), HOME=str(tmp_path), **sock), timeout=60)
+    assert res.stdout.strip() == named, res.stderr
+
+
+def test_p2_a_free_lock_is_taken_at_once_and_a_held_one_times_out(tmp_path):
+    lock = tmp_path / "s.lock"
+    t0 = time.time()
+    assert _acquire(lock).stdout.strip() == "locked" and time.time() - t0 < 5
+    holder = _holder(lock)
+    try:
+        t0 = time.time()
+        res = _acquire(lock, MAM_PANE_CLAIM_WAIT_S="2")
+        assert res.stdout.strip() == "timeout", res.stderr
+        assert 1.5 <= time.time() - t0 < 10
+        assert _acquire(tmp_path / "other-server.lock", MAM_PANE_CLAIM_WAIT_S="2").stdout.strip() == "locked"
+    finally:
+        holder.kill()
+        holder.wait()
+
+
+def test_p3_the_lock_ends_with_its_holders_process(tmp_path):
+    lock = tmp_path / "s.lock"
+    holder = _holder(lock)
+    os.kill(holder.pid, signal.SIGKILL)
+    holder.wait()
+    t0 = time.time()
+    assert _acquire(lock, MAM_PANE_CLAIM_WAIT_S="30").stdout.strip() == "locked"
+    assert time.time() - t0 < 5, "nothing stale to wait out"
+
+
+def test_p4_a_waiter_gets_the_lock_when_the_holder_ends(tmp_path):
+    lock = tmp_path / "s.lock"
+    holder = _holder(lock, "2")
+    t0 = time.time()
+    res = _acquire(lock, MAM_PANE_CLAIM_WAIT_S="30")
+    assert res.stdout.strip() == "locked"
+    assert 1 <= time.time() - t0 < 10
+    holder.wait()
+
+
+def test_p5_the_wait_knob_is_a_decimal_with_a_floor_of_one(monkeypatch):
+    assert pane_lock.wait_s({}) == 120
+    assert pane_lock.wait_s({"MAM_PANE_CLAIM_WAIT_S": "7"}) == 7
+    assert pane_lock.wait_s({"MAM_PANE_CLAIM_WAIT_S": "0700"}) == 700
+    for bad in ("0", "abc", "", "-1", " "):
+        assert pane_lock.wait_s({"MAM_PANE_CLAIM_WAIT_S": bad}) == 120, bad
+
+
+# ---------------------------------------------------------------------------
+# A - create keeps its own pane when the monitor registered it first. With
+# pane claims serialised, one create can reach its end-of-run reconcile while
+# another is between `agent start` and its row write; the adopted row is that
+# create's own pane, not a second live session of its name.
+# ---------------------------------------------------------------------------
+
+
+
+def _seed_row(tmp_path, name, epoch, note="auto-registered by monitor", job=None):
+    row = {"name": name, "status": "running", "role": "creator", "herdr_session_created_at": "2026-10-11T00:00:00Z",
+           "herdr_session_epoch": epoch, "herdr_session": "default", "herdr_server": "default",
+           "pane": {"index": 0, "pid": 9999, "cmd": "claude", "cmd_full": "claude", "cwd": str(tmp_path)},
+           "last_visible_status": "running"}
+    if note:
+        row["last_visible_note"] = note
+    if job:
+        row["delegate_job_id"] = job
+    yaml_path = tmp_path / ".mam" / "agent-sessions.yaml"
+    yaml_path.parent.mkdir(parents=True, exist_ok=True)
+    yaml_path.write_text("herdr_sessions:\n- " + json.dumps(row) + "\n")
+    return yaml_path
+
+
+def _create(tmp_path, name, role="creator"):
+    return subprocess.run(["bash", str(tmp_path / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"),
+                           "--workspace", str(tmp_path), "--agent", "claude", "--role", role, "--session", name,
+                           "--no-onboard"], capture_output=True, text=True, cwd=str(tmp_path), timeout=300)
+
+
+def _rows(yaml_path):
+    import yaml
+    return yaml.safe_load(yaml_path.read_text())["herdr_sessions"]
+
+
+def test_a1_create_replaces_the_monitors_registration_of_its_own_pane(mam_sandbox, mock_herdr, mock_agents):
+    name = "t50-adopted-creator-claude"
+    yaml_path = _seed_row(mam_sandbox, name, int(time.time()))      # registered after this create began
+    res = _create(mam_sandbox, name, role="Creator")               # the monitor guessed "creator" off the name
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert f"NOTE: the monitor registered {name} while it was being created" in res.stdout
+    rows = [r for r in _rows(yaml_path) if r["name"] == name]
+    assert len(rows) == 1 and rows[0]["status"] == "running" and rows[0]["role"] == "Creator"
+    assert rows[0].get("last_visible_note") != "auto-registered by monitor" and rows[0].get("start_command")
+
+
+@pytest.mark.parametrize("kind", ["real", "old-adoption", "adoption-with-job"])
+def test_a2_another_live_session_of_that_name_is_still_refused(mam_sandbox, mock_herdr, mock_agents, kind):
+    name = "t50-taken-creator-claude"
+    now = int(time.time())
+    if kind == "real":
+        yaml_path = _seed_row(mam_sandbox, name, now, note=None)
+    elif kind == "old-adoption":
+        yaml_path = _seed_row(mam_sandbox, name, now - 3600)        # a pane from before this create
+    else:
+        yaml_path = _seed_row(mam_sandbox, name, now, job="deadbeef")
+    res = _create(mam_sandbox, name)
+    assert res.returncode == 4, res.stdout + res.stderr
+    assert f"ERROR: {name} already running in agent-sessions.yaml" in res.stdout
+    rows = [r for r in _rows(yaml_path) if r["name"] == name]
+    assert len(rows) == 1 and rows[0]["status"] == "running"