fix(issue-50): the registry wait admits one job per session and holds it (review aa1bff92 F1, F2)

F1: `clear` is now an admission persisted on the released record - a merge
through the engine whose new `guard` re-reads the session's open jobs under
the registry lock, so two waiters cannot both be admitted. It is scoped to
the attempt by the dispatch token (a redispatch voids it) and reported as
`admitted` to every other waiter, which holds the session as a running job
does, whatever its FIFO place: a lower id in the same second or a
redispatched older loop job now waits.
F2: a pending record's liveness window is the dispatcher's signal cadence
times three misses (10 s x 3 = 30 s), no longer ack_timeout_sec, which the
agent's budget takes down to 1 s. After `clear` a detached hold renews the
admitted record's signal until the record leaves pending, its attempt
changes, the wrapper is gone, or the wall clock ends.
Tests: u1 boundaries, u1d/u2e/u2f/u2g/u3d/u3e for the admission, u4c/u4d for
the hold, s1 pins --hold; real herdr: C3/C7/C8 adjusted, C9 is C8's
live-owner counterpart with short ack windows.

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

.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,
-                           source=source, fields=fields,
-                           expected_notification=expected_notification,
-                           expected_dispatch_token=expected_dispatch_token)
+        if guard is not None and not guard(record):
+            decision = _Decision(False, from_status, "guard refused")
+        else:
+            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

@@ -63,9 +63,24 @@ 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"
 
 
 def _now() -> int:
@@ -244,7 +259,10 @@ 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 ack_timeout_sec.
+           `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.
     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
@@ -273,15 +291,98 @@ 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)
             seen = _parse_ts(rec.get(DISPATCH_SIGNAL_FIELD) or rec.get("created_at"))
-            if seen is not None and now - seen <= ack:
-                live.append((str(rec.get("created_at") or ""), other, status))
+            if seen is not None and now - seen <= DISPATCH_SIGNAL_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 {ack}s"))
+                stale.append((other, f"pending, no dispatch signal for {DISPATCH_SIGNAL_WINDOW}s"))
     return live, stale
 
 
+def _first_in_line(live, mine) -> bool:
+    """No live running or admitted 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 _admit(registry_dir: str, session: str, job_id: str) -> 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)
+
+    def guard(record):
+        if record.get("status") != "pending" or record.get("dispatch_token") != token:
+            return False
+        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
+
+
+def _start_hold(a: argparse.Namespace) -> 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."""
+    cmd = [sys.executable, "-m", "lib_py.delegation.wrapper", "hold-admission", "--job", a.job,
+           "--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 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)
+    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:
+            return "released"
+        now = _now()
+        if now >= deadline:
+            return "timeout"
+        if now - last_signal >= DISPATCH_SIGNAL_EVERY:
+            _signal_dispatch_wait(a.registry_dir, a.job)
+            last_signal = now
+
+
 def _signal_dispatch_wait(registry_dir: str, job_id: str) -> None:
     """The waiter's liveness: a merge of dispatch_wait_at into its own pending
     record through the registry's one transition boundary (status untouched,
@@ -295,12 +396,14 @@ def _signal_dispatch_wait(registry_dir: str, job_id: str) -> None:
 
 
 def wait_open_jobs(a: argparse.Namespace) -> str:
-    """`clear` when no live open job of the session remains and this job is
-    the earliest live pending one (FIFO by (created_at, job_id)); `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."""
+    """`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)."""
     me = _load_record(a.registry_dir, a.job) or {}
     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()
@@ -314,10 +417,10 @@ def wait_open_jobs(a: argparse.Namespace) -> str:
             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):
-            queue = [(c, j) for c, j, st in live if st == "pending"] + [mine]
-            if min(queue) == mine:
-                return "clear"
+        if _first_in_line(live, mine) and _admit(a.registry_dir, a.agent_session, a.job):
+            if getattr(a, "hold", False):
+                _start_hold(a)
+            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:
@@ -480,6 +583,13 @@ 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("--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("--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 +654,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/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job

@@ -855,7 +855,7 @@ _registry_wait() {
   local job_id="$1" sess="$2" res
   _decided res wait-open-jobs --job "$job_id" --agent-session "$sess" --registry-dir "$REGISTRY_DIR" \
       --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 ;;
     busy*)

tests/test_issue50_dispatch.py

@@ -2,8 +2,10 @@
 
 - U: the decisions in lib_py.delegation.wrapper - the open-job table (status x
   signal age x session spelling), FIFO release among live pending waiters,
-  the waiter's own signal written through the registry, the bound, a dead
-  wrapper, and finalize-end's `registry_wait` phase closing as session_busy.
+  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.
 - 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 +18,7 @@ import json
 import os
 import subprocess
 import sys
+import threading
 import time
 from pathlib import Path
 
@@ -24,6 +27,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 +54,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")
@@ -67,18 +87,20 @@ def _ns(reg, job, session="herdr:s", wait_s=0, start_time=None, timeout=3600, wa
     ("running", 10, None, "live"),          # an event idle_timeout_sec (120) ago at most
     ("running", 121, None, "stale"),
     ("running", None, None, "stale"),       # running with no events.log at all
-    ("pending", 10, None, "live"),          # created within ack_timeout_sec (180)
-    ("pending", 181, None, "stale"),
+    ("pending", 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, 181, "stale"),
-    ("pending", 10, 181, "stale"),          # the signal, once present, is what counts
+    ("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),
 ])
 def test_u1_open_jobs_table(tmp_path, status, age, signal, want):
     NOW = int(time.time())   # the running check reads the real clock (recent_disk_heartbeat)
-    other = _register(tmp_path, status=status)
+    other = _register(tmp_path, status=status, ack=1)
     if status == "running" and age is not None:
         _event(tmp_path, other, NOW - age)
     if status == "pending":
@@ -97,6 +119,22 @@ def test_u1_open_jobs_table(tmp_path, status, age, signal, want):
         assert got == (set(), set())
 
 
+@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")])
 def test_u1b_the_session_spelling_is_normalised(tmp_path, record_session, asked):
     other = _register(tmp_path, session=record_session)
@@ -135,20 +173,61 @@ 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")
-    second = _register(tmp_path, created_at=same, job_id="aaaa0002")
-    # records carry their job_id inside; the file name is what open_jobs keys on
-    for jid, real in ((first, "aaaa0001"), (second, "aaaa0002")):
-        os.rename(tmp_path / f"{jid}.json", tmp_path / f"{real}.json")
+    _register_as(tmp_path, "aaaa0001", created_at=same)
+    _register_as(tmp_path, "aaaa0002", created_at=same)
     assert decide.wait_open_jobs(_ns(tmp_path, "aaaa0001")) == "clear"
     assert decide.wait_open_jobs(_ns(tmp_path, "aaaa0002")) == "busy aaaa0001"
 
 
 def test_u2c_a_dead_waiter_is_ignored_and_reported(tmp_path, capsys):
-    b = _register(tmp_path, ack=5, created_at=_iso(time.time() - 60), dispatch_wait_at=_iso(time.time() - 6))
+    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,6 +264,45 @@ def test_u3_the_waiter_signals_on_its_own_record(tmp_path):
     assert me in {j for _c, j, _s in live}
 
 
+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):
     # the engine refuses a merge into a terminal record, so the waiter would
@@ -227,7 +345,50 @@ 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 180s" in res.stdout
+    assert f"{a}  running" in res.stdout and f"{b}  stale     pending, no dispatch signal for {WINDOW}s" 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 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")
+    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))
+    assert res.stdout == "" and "is gone" in res.stderr and time.time() - t0 < 10, res.stdout + res.stderr
 
 
 def test_u4b_wait_open_jobs_stops_with_its_wrapper(tmp_path):
@@ -292,3 +453,4 @@ def test_s1_both_dispatch_paths_wait_by_the_registry_before_the_idle_wait():
     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)"

tests/test_issue50_dispatch_herdr.py

@@ -329,7 +329,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"]
 
 
 # ---------------------------------------------------------------------------
@@ -434,6 +435,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"
 
@@ -456,16 +460,49 @@ def test_c8_a_dead_waiter_is_skipped_once_its_signal_is_stale(box):
     c = box.wait_new_job(known | {b})
     assert box.record(b)["created_at"] < box.record(c)["created_at"]
     _after(2, box.publish, a, "completed", "prior done")
-    _after(14, box.publish, c, "started")           # after B's 5 s ack window has long passed
-    _after(16, box.publish, c, "completed", "c done")
+    _after(40, box.publish, c, "started")           # after B's signal window (30 s) has passed
+    _after(42, box.publish, c, "completed", "c done")
     rc, out, err = box.finish(proc_c)
     assert rc == 0, out + err
-    assert f"stale open job {b} ignored (pending, no dispatch signal for 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"
+
+
 # ---------------------------------------------------------------------------
 # A - three `create --onboard` at once in one server
 # ---------------------------------------------------------------------------