fix(issue-50): the waiter and its hold are pinned to one attempt and session; an unsignalled pending record lives by its ack budget (review 55219be1 R1, R2)

R1: wait-open-jobs takes --dispatch-token (the token the wrapper read for
this attempt) and keeps it, with the session it asked about, for its whole
life; every write (signal, admission, the hold's renewal) is a guarded
engine merge that re-reads the record under the registry lock and refuses
once the record is not that attempt's pending one in that session. A record
redispatched, moved or closed by someone else makes the waiter answer
`superseded`: the bash wrapper exits 5 without a close, the record is its
new owner's. hold-admission is given the waiter's pinned identity instead
of reading a token from disk, so a hold that starts late holds nothing.

R2: a pending record that never carried a waiter's signal (create --onboard
registers and injects directly and waits ack_timeout_sec for `started`) is
live for that budget from created_at, not for the 30 s signal window; the
window applies only once a dispatcher has signalled.

Tests: u1 rows for the two pending lifetimes, u3b asserts the guard, u3f/u3g/u3h
(cross-session redispatch, in-place attempt change, token changed before the
waiter started) with the real waiter subprocess, u4e/u4f (late hold, running
hold released by a redispatch), s1 pins the token argument and the no-close
branch; real herdr C10 (onboarding valid before and after 30 s blocks the
session), C11 (expired onboarding is stale), C12 (redispatched waiter leaves
its record alone, exit 5).

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

.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
@@ -81,6 +86,7 @@ DISPATCH_SIGNAL_WINDOW = DISPATCH_SIGNAL_EVERY * DISPATCH_SIGNAL_MISSES
 # 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:
@@ -258,15 +264,18 @@ 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
-           `dispatch_wait_at`, else created_at) within DISPATCH_SIGNAL_WINDOW.
-           A live pending record whose admission names its current dispatch
-           token is reported as `admitted`: it holds the session as a running
-           job does, whatever its place in the FIFO.
+           idle_timeout_sec; `pending` with a waiter's signal (`dispatch_wait_at`)
+           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). A live pending record whose admission names
+           its current dispatch token is reported as `admitted`: it holds the
+           session as a running job does, whatever its place in the FIFO.
     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
-           a dead record cannot block a session forever."""
+           zombie's unclosed record, an onboarding never acknowledged. They
+           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 = [], []
@@ -291,13 +300,18 @@ 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":
-            seen = _parse_ts(rec.get(DISPATCH_SIGNAL_FIELD) or rec.get("created_at"))
-            if seen is not None and now - seen <= DISPATCH_SIGNAL_WINDOW:
+            if rec.get(DISPATCH_SIGNAL_FIELD):
+                seen, window = _parse_ts(rec.get(DISPATCH_SIGNAL_FIELD)), DISPATCH_SIGNAL_WINDOW
+                why = f"no dispatch signal for {window}s"
+            else:
+                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:
                 admitted = bool(rec.get(ADMITTED_TOKEN_FIELD)) and \
                     rec.get(ADMITTED_TOKEN_FIELD) == rec.get("dispatch_token")
                 live.append((str(rec.get("created_at") or ""), other, "admitted" if admitted else status))
             else:
-                stale.append((other, f"pending, no dispatch signal for {DISPATCH_SIGNAL_WINDOW}s"))
+                stale.append((other, f"pending, {why}"))
     return live, stale
 
 
@@ -309,43 +323,61 @@ def _first_in_line(live, mine) -> bool:
     return min([(c, j) for c, j, _st in live] + [mine]) == mine
 
 
-def _admit(registry_dir: str, session: str, job_id: str) -> bool:
+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 _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 (its dispatch token: a
-    redispatch voids it); the signal is renewed with it. False when the
-    record is not pending any more, its attempt changed, or another waiter
-    was admitted first - the caller keeps waiting and sees that one."""
-    from lib_py.delegation import engine, mqtt_common
-    me = _load_record(registry_dir, job_id) or {}
-    token = me.get("dispatch_token")
-    if not token:
-        print(f"dispatch wait: job {job_id} carries no dispatch token; cannot be admitted", file=sys.stderr)
-        return False
-    mine = (str(me.get("created_at") or ""), job_id)
+    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 guard(record):
-        if record.get("status") != "pending" or record.get("dispatch_token") != token:
-            return False
+    def first(_record):
         live, _stale = open_jobs(registry_dir, session, job_id)
         return _first_in_line(live, mine)
 
     now = mqtt_common._utcnow()
-    try:
-        return engine.transition(job_id, "merge", registry_dir=registry_dir, source="dispatch_admit", guard=guard,
-                                 fields={DISPATCH_SIGNAL_FIELD: now, ADMITTED_AT_FIELD: now,
-                                         ADMITTED_TOKEN_FIELD: token}).committed
-    except Exception as exc:  # the record is gone or unwritable: not admitted, keep waiting
-        print(f"dispatch wait: could not admit {job_id}: {exc}", file=sys.stderr)
-        return False
+    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) -> None:
+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). Its output is not the answer."""
+    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)]
@@ -362,49 +394,55 @@ def _pythonpath() -> str:
 
 
 def hold_admission(a: argparse.Namespace) -> str:
-    """`released` once the record is not this attempt's admitted pending one
-    any more, `timeout` at the wall clock; the signal is renewed every
-    DISPATCH_SIGNAL_EVERY s in between. OwnerGone (the wrapper) ends it."""
-    me = _load_record(a.registry_dir, a.job) or {}
-    token = me.get(ADMITTED_TOKEN_FIELD)
+    """`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 token is None or rec is None or rec.get("status") != "pending" \
-                or rec.get("dispatch_token") != token or rec.get(ADMITTED_TOKEN_FIELD) != token:
+        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)
+            _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) -> None:
+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:
-        mqtt_common.update_job_status(job_id, registry_dir, _source="dispatch_wait",
-                                      **{DISPATCH_SIGNAL_FIELD: mqtt_common._utcnow()})
-    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)
+    guard = (lambda record: record.get(ADMITTED_TOKEN_FIELD) == token) if admitted else None
+    return _merge_own(registry_dir, job_id, token, session, "dispatch_wait",
+                      {DISPATCH_SIGNAL_FIELD: mqtt_common._utcnow()}, guard)
 
 
 def wait_open_jobs(a: argparse.Namespace) -> str:
     """`clear` when no live running or admitted job of the session remains,
     this job is the earliest live pending one (FIFO by (created_at, job_id))
     and its admission is committed (_admit); `busy <ids>` when the bound runs
-    out first. The bound is min(--wait-s, this job's remaining wall clock).
-    Every DISPATCH_SIGNAL_EVERY s the waiter signals on its own record so
-    later 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)."""
+    out first; `superseded` once the record is not this attempt's pending one
+    in this session any more (redispatched, moved, closed by someone else) -
+    nothing is written or closed then, the record is its owner's. 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. Every DISPATCH_SIGNAL_EVERY s
+    the waiter signals on its own record so later 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))
@@ -412,19 +450,21 @@ def wait_open_jobs(a: argparse.Namespace) -> str:
     last_signal = None
     while True:
         now = _now()
+        if not _own(_load_record(a.registry_dir, a.job), token, a.agent_session):
+            return "superseded"
         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 _first_in_line(live, mine) and _admit(a.registry_dir, a.agent_session, a.job):
+        if _first_in_line(live, mine) and _admit(a.registry_dir, a.agent_session, a.job, token, mine):
             if getattr(a, "hold", False):
-                _start_hold(a)
+                _start_hold(a, token)
             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)
+            _signal_dispatch_wait(a.registry_dir, a.job, token, a.agent_session)
             last_signal = now
         _sleep(a, 1)
 
@@ -583,9 +623,12 @@ 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)

.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, or the session's open jobs in the registry did not end within `MAM_DISPATCH_WAIT_S` (Issue #50) (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"
@@ -540,7 +540,7 @@ EOF
       local cur_agent="$AGENT"
       [ "$_phase" = "reviewer" ] && cur_agent="$REVIEWER"
       if [ "$cur_agent" != "human" ] && ! _is_remote "$current_session"; then
-        _registry_wait "$JOB_ID" "$current_session"
+        _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"
@@ -844,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
-# by this job's wall clock) closes this job as cancelled/session_busy and
-# exits 5, like the preflight and the idle wait.
+# ignored). The waiter is pinned to this attempt's token and session (review
+# 55219be1 R1): a record redispatched, moved or closed by someone else while
+# 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 "$$" --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/test_issue50_dispatch.py

@@ -1,11 +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,
-  the admission persisted on the released record (review aa1bff92 F1) and
-  held alive after the answer (F2), 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.
+  signal age x ack budget x session spelling), FIFO release among live
+  pending waiters, the admission persisted on the released record (review
+  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.
@@ -83,24 +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", [
-    ("running", 10, None, "live"),          # an event idle_timeout_sec (120) ago at most
-    ("running", 121, None, "stale"),
-    ("running", None, None, "stale"),       # running with no events.log at all
-    ("pending", 10, None, "live"),          # created within the signal window (30)
-    ("pending", 30, None, "live"),
-    ("pending", 31, None, "stale"),
-    ("pending", 181, 5, "live"),            # old record, fresh dispatch_wait_at
-    ("pending", 181, 31, "stale"),
-    ("pending", 10, 31, "stale"),           # the signal, once present, is what counts
-    ("pending", 1, 2, "live"),              # ack_timeout_sec (1 here) is not the window
-    ("completed", 10, None, None),
-    ("error", 10, None, None),
-    ("cancelled", 10, None, None),
+@pytest.mark.parametrize("status,age,signal,ack,want", [
+    ("running", 10, None, 1, "live"),       # an event idle_timeout_sec (120) ago at most
+    ("running", 121, None, 1, "stale"),
+    ("running", None, None, 1, "stale"),    # running with no events.log at all
+    ("pending", 10, None, 180, "live"),     # never signalled: created within its ack budget
+    ("pending", 31, None, 180, "live"),     # ... past the signal window too (review 55219be1 R2:
+    ("pending", 180, None, 180, "live"),    #     create --onboard waits that long for `started`)
+    ("pending", 181, None, 180, "stale"),
+    ("pending", 10, None, 1, "stale"),      # never signalled, past its ack budget: a dead producer
+    ("pending", 181, 5, 1, "live"),         # old record, fresh dispatch_wait_at
+    ("pending", 181, 31, 180, "stale"),     # signalled: the window (30), not the ack budget
+    ("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, ack=1)
+    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":
@@ -115,8 +118,11 @@ def test_u1_open_jobs_table(tmp_path, status, age, signal, want):
     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")])
@@ -304,13 +310,13 @@ def test_u3e_the_admission_is_rechecked_under_the_registry_lock(tmp_path, monkey
 
 
 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):
-    # the engine refuses a merge into a terminal record, so the waiter would
-    # stay invisible to every other waiter's FIFO
+    # 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)) == f"busy {a}"
+    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)
@@ -318,15 +324,114 @@ def test_u3c_a_terminal_self_record_cannot_carry_the_signal(tmp_path):
     assert me not in {j for _c, j, _s in live}
 
 
-def test_u3b_the_signal_is_a_registry_merge(tmp_path, monkeypatch):
+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(
-                            job=job_id, reg=registry_dir, source=_source, fields=fields))
-    decide._signal_dispatch_wait(str(tmp_path), me)
-    assert seen["job"] == me and seen["reg"] == str(tmp_path) and seen["source"] == "dispatch_wait"
-    assert set(seen["fields"]) == {"dispatch_wait_at"}
+
+    def transition(job_id, request, registry_dir=None, source="unknown", guard=None, fields=None, **kw):
+        seen.update(job=job_id, request=request, reg=registry_dir, source=source, guard=guard, fields=fields)
+        return argparse.Namespace(committed=True)
+
+    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_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
 
 
 # ---------------------------------------------------------------------------
@@ -345,7 +450,8 @@ def test_u4_open_jobs_prints_live_and_stale(tmp_path):
     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 {WINDOW}s" 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
 
 
 def _hold_pids(jid):
@@ -378,19 +484,68 @@ def test_u4c_the_wait_holds_its_admission_until_the_record_leaves_pending(tmp_pa
             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)
-    res = _cli(tmp_path, "hold-admission", "--job", me, "--registry-dir", str(tmp_path), "--timeout", "3600")
+    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 = _cli(tmp_path, "hold-admission", "--job", me, "--registry-dir", str(tmp_path), "--timeout", "3600",
-               "--watch-pid", str(gone.pid))
+    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):
     a = _register(tmp_path, status="running")
     _event(tmp_path, a, time.time())
@@ -436,8 +591,9 @@ 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
-    assert src.count('_registry_wait "$JOB_ID" "$current_session"') == 1
+    # ... each with the token it read for this attempt, pinned (review 55219be1 R1)
+    assert src.count('_registry_wait "$JOB_ID" "$AGENT_SESSION" "$_disp_tok"') == 1
+    assert src.count('_registry_wait "$JOB_ID" "$current_session" "$_disp_tok"') == 1
     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"')
@@ -454,3 +610,6 @@ def test_s1_both_dispatch_paths_wait_by_the_registry_before_the_idle_wait():
     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,7 @@ so the herdr idle wait stays and the registry wait runs before it.
 """
 from __future__ import annotations
 
+import calendar
 import json
 import os
 import re
@@ -37,6 +38,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))
@@ -503,6 +505,83 @@ def test_c9_live_waiters_with_short_ack_windows_are_never_stale_nor_co_admitted(
     assert pane.index(f'Your job_id is "{b}"') < pane.index(f'Your job_id is "{c}"'), "injected in FIFO order"
 
 
+# ---------------------------------------------------------------------------
+# C10 - a pending onboarding (registered and injected with no waiter, as
+# create --onboard does) holds the session for its ack budget, not for the
+# 30 s signal window (review 55219be1 R2); C11 is its expired control
+# ---------------------------------------------------------------------------
+
+def _onboarding(box: Box, ack: int) -> str:
+    a = box.register(ack_timeout_sec=ack)
+    prompt = box.herdr("agent", "prompt", SESSION, "Onboarding " + a)
+    assert prompt.returncode == 0, prompt.stdout + prompt.stderr
+    return a
+
+
+def test_c10_a_pending_onboarding_within_its_ack_budget_blocks_the_session_past_the_signal_window(box):
+    t0 = time.time()
+    a = _onboarding(box, 180)
+    known = box.job_ids()
+    for label in ("before", "after"):
+        if label == "after":
+            time.sleep(max(0.0, t0 + 31.5 - time.time()))       # A is now older than the signal window
+        proc = box.submit(prompt="work " + label, MAM_DISPATCH_WAIT_S="3")
+        b = box.wait_new_job(known)
+        known.add(b)
+        rc, out, err = box.finish(proc)
+        assert rc == 5, (label, out + err)
+        rec = box.record(b)
+        assert rec["status"] == "cancelled" and rec.get("detail") == "session_busy", (label, rec)
+        assert a in err and "still has open jobs in the registry" in err and "stale open job" not in err, (label, err)
+        assert b not in box.pane_text(), label
+        assert box.record(a)["status"] == "pending" and "dispatch_wait_at" not in box.record(a), label
+    created = calendar.timegm(time.strptime(box.record(a)["created_at"], "%Y-%m-%dT%H:%M:%SZ"))
+    assert 31 <= time.time() - created < 180, "the second answer came past the window, within the budget"
+
+
+def test_c11_an_onboarding_never_acknowledged_within_its_ack_budget_is_stale(box):
+    a = _onboarding(box, 2)
+    time.sleep(3.2)
+    known = box.job_ids()
+    proc = box.submit(MAM_DISPATCH_WAIT_S="30")
+    b = box.wait_new_job(known)
+    _after(2, box.publish, b, "started")
+    _after(4, box.publish, b, "completed", "b done")
+    rc, out, err = box.finish(proc)
+    assert rc == 0, out + err
+    assert f"stale open job {a} ignored (pending, no dispatch signal and no started within 2s)" in err, err
+    assert box.record(b)["status"] == "completed" and box.record(a)["status"] == "pending"
+    assert b in box.pane_text()
+
+
+# ---------------------------------------------------------------------------
+# C12 - a waiting job redispatched by someone else (new attempt) is left to
+# that attempt: the old wrapper exits 5 without a close (review 55219be1 R1)
+# ---------------------------------------------------------------------------
+
+def test_c12_a_waiting_job_redispatched_by_someone_else_is_left_to_its_new_attempt(box):
+    a = box.register()
+    box.publish(a, "started")
+    known = box.job_ids()
+    proc = box.submit(MAM_DISPATCH_WAIT_S="60")
+    b = box.wait_new_job(known)
+    box.wait_field(b, "dispatch_wait_at")
+    old = box.record(b)["dispatch_token"]
+    res = subprocess.run([sys.executable, str(REGISTRY_PY), "--registry-dir", str(box.jobs), "update", "--job", b,
+                          "--status", "pending", "--attempt-id", "2"], capture_output=True, text=True, env=box.env(),
+                         timeout=60)
+    assert res.returncode == 0, res.stdout + res.stderr
+    moved = box.record(b)
+    assert moved["dispatch_token"] != old and moved["attempt_id"] == 2
+    box.publish(a, "completed", "prior done")
+    rc, out, err = box.finish(proc)
+    assert rc == 5, out + err
+    assert f"job {b} was redispatched, moved or closed by someone else" in err, err
+    assert box.record(b) == moved, "not closed, not admitted, not renewed: attempt 2's dispatcher owns it"
+    assert box.events(b) == [] and b not in box.pane_text()
+    assert out.strip().splitlines()[-1].endswith(f"/delegate_job_logs/{b}"), out
+
+
 # ---------------------------------------------------------------------------
 # A - three `create --onboard` at once in one server
 # ---------------------------------------------------------------------------