From 30c25f193c67e2ceaa5762ebf76544c84cdb1205 Mon Sep 17 00:00:00 2001 From: Godopu Date: Sun, 11 Oct 2026 02:58:16 +0000 Subject: [PATCH] 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 --- .agents/skills/lib_py/delegation/wrapper.py | 159 +++++++----- .../multi-agent-mux-delegate-job/SKILL.md | 2 +- .../multi-agent-mux-delegate-job | 23 +- tests/test_issue50_dispatch.py | 235 +++++++++++++++--- tests/test_issue50_dispatch_herdr.py | 79 ++++++ 5 files changed, 394 insertions(+), 104 deletions(-) diff --git a/.agents/skills/lib_py/delegation/wrapper.py b/.agents/skills/lib_py/delegation/wrapper.py index 02fd66f..0f57a72 100644 --- a/.agents/skills/lib_py/delegation/wrapper.py +++ b/.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 ` + [--dispatch-token K] [--hold] prints `clear`, `busy `, 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 ` 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) diff --git a/.agents/skills/multi-agent-mux-delegate-job/SKILL.md b/.agents/skills/multi-agent-mux-delegate-job/SKILL.md index 2cd3edc..e163688 100644 --- a/.agents/skills/multi-agent-mux-delegate-job/SKILL.md +++ b/.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 (`/`), fulfilling the parsing contract for orchestrator runners. diff --git a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job index 175b014..4370f1d 100755 --- a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job +++ b/.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 (Issue #50) +# _registry_wait (Issue #50) # Waits, by the registry alone, until no live open job of 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" diff --git a/tests/test_issue50_dispatch.py b/tests/test_issue50_dispatch.py index b6fc452..83180c8 100644 --- a/tests/test_issue50_dispatch.py +++ b/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" diff --git a/tests/test_issue50_dispatch_herdr.py b/tests/test_issue50_dispatch_herdr.py index 7a9899d..3934d66 100644 --- a/tests/test_issue50_dispatch_herdr.py +++ b/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 # ---------------------------------------------------------------------------