diff --git a/.agents/skills/lib_py/delegation/engine.py b/.agents/skills/lib_py/delegation/engine.py index 46e541f..a0a97d3 100644 --- a/.agents/skills/lib_py/delegation/engine.py +++ b/.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: diff --git a/.agents/skills/lib_py/delegation/wrapper.py b/.agents/skills/lib_py/delegation/wrapper.py index eb55837..02fd66f 100644 --- a/.agents/skills/lib_py/delegation/wrapper.py +++ b/.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 ` - 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 ` 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)) 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 5fdb64b..175b014 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 @@ -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*) diff --git a/tests/test_issue50_dispatch.py b/tests/test_issue50_dispatch.py index ace0301..b6fc452 100644 --- a/tests/test_issue50_dispatch.py +++ b/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)" diff --git a/tests/test_issue50_dispatch_herdr.py b/tests/test_issue50_dispatch_herdr.py index f5d5e14..7a9899d 100644 --- a/tests/test_issue50_dispatch_herdr.py +++ b/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 # ---------------------------------------------------------------------------