GodopuandClaude Fable 5.1
30c25f193c
R1: wait-open-jobs takes --dispatch-token (the token the wrapper read for this attempt) and keeps it, with the session it asked about, for its whole life; every write (signal, admission, the hold's renewal) is a guarded engine merge that re-reads the record under the registry lock and refuses once the record is not that attempt's pending one in that session. A record redispatched, moved or closed by someone else makes the waiter answer `superseded`: the bash wrapper exits 5 without a close, the record is its new owner's. hold-admission is given the waiter's pinned identity instead of reading a token from disk, so a hold that starts late holds nothing. R2: a pending record that never carried a waiter's signal (create --onboard registers and injects directly and waits ack_timeout_sec for `started`) is live for that budget from created_at, not for the 30 s signal window; the window applies only once a dispatcher has signalled. Tests: u1 rows for the two pending lifetimes, u3b asserts the guard, u3f/u3g/u3h (cross-session redispatch, in-place attempt change, token changed before the waiter started) with the real waiter subprocess, u4e/u4f (late hold, running hold released by a redispatch), s1 pins the token argument and the no-close branch; real herdr C10 (onboarding valid before and after 30 s blocks the session), C11 (expired onboarding is stale), C12 (redispatched waiter leaves its record alone, exit 5). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
616 lines
31 KiB
Python
616 lines
31 KiB
Python
"""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 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.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import calendar
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
|
|
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))
|
|
|
|
from lib_py.delegation import mqtt_common, registry, wrapper as decide # noqa: E402
|
|
|
|
|
|
def _iso(epoch):
|
|
return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(epoch))
|
|
|
|
|
|
def _register(reg, session="herdr:s", ack=180, idle=120, **over):
|
|
jid = registry.register_job(prompt="p", agent="claude-code", agent_session=session, registry_dir=str(reg),
|
|
ack_timeout_sec=ack, idle_timeout_sec=idle)
|
|
if over:
|
|
_edit(reg, jid, **over)
|
|
return jid
|
|
|
|
|
|
def _edit(reg, jid, **fields):
|
|
path = reg / f"{jid}.json"
|
|
rec = json.loads(path.read_text())
|
|
rec.update(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")
|
|
|
|
|
|
def _ns(reg, job, session="herdr:s", wait_s=0, start_time=None, timeout=3600, watch_pid=None):
|
|
return argparse.Namespace(job=job, agent_session=session, registry_dir=str(reg), wait_s=wait_s,
|
|
start_time=start_time, timeout=timeout, watch_pid=watch_pid)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# U1 - the open-job table
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@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, ack, want):
|
|
NOW = int(time.time()) # the running check reads the real clock (recent_disk_heartbeat)
|
|
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":
|
|
_edit(tmp_path, other, created_at=_iso(NOW - age))
|
|
if signal is not None:
|
|
_edit(tmp_path, other, dispatch_wait_at=_iso(NOW - signal))
|
|
me = _register(tmp_path)
|
|
live, stale = decide.open_jobs(str(tmp_path), "herdr:s", me, now=NOW)
|
|
got = {j for _c, j, _s in live}, {j for j, _r in stale}
|
|
if want == "live":
|
|
assert got == ({other}, set())
|
|
elif want == "stale":
|
|
assert got == (set(), {other}), got
|
|
assert stale[0][1].startswith(status + ", no ")
|
|
if status == "pending":
|
|
assert stale[0][1].endswith(f"for {WINDOW}s" if signal else f"no started within {ack}s")
|
|
else:
|
|
assert got == (set(), set())
|
|
assert decide.DEFAULT_ACK_TIMEOUT_SEC == registry.DEFAULT_ACK_TIMEOUT_SEC
|
|
|
|
|
|
@pytest.mark.parametrize("token_match,signal,want", [(True, 5, "admitted"), (False, 5, "pending"), (True, 31, "stale")])
|
|
def test_u1d_an_admission_is_scoped_to_the_attempt_and_lives_by_the_signal(tmp_path, token_match, signal, want):
|
|
NOW = int(time.time())
|
|
other = _register(tmp_path)
|
|
token = _rec(tmp_path, other)["dispatch_token"] if token_match else "a-previous-attempt"
|
|
_edit(tmp_path, other, dispatch_wait_at=_iso(NOW - signal), dispatch_admitted_at=_iso(NOW - signal),
|
|
dispatch_admitted_token=token)
|
|
me = _register(tmp_path)
|
|
live, stale = decide.open_jobs(str(tmp_path), "herdr:s", me, now=NOW)
|
|
if want == "stale":
|
|
assert (live, [j for j, _r in stale]) == ([], [other])
|
|
else:
|
|
assert [(j, st) for _c, j, st in live] == [(other, want)]
|
|
assert decide.DISPATCH_SIGNAL_WINDOW == WINDOW
|
|
|
|
|
|
@pytest.mark.parametrize("record_session,asked", [("herdr:s", "s"), ("s", "herdr:s"), ("herdr:s", "herdr:s")])
|
|
def test_u1b_the_session_spelling_is_normalised(tmp_path, record_session, asked):
|
|
other = _register(tmp_path, session=record_session)
|
|
me = _register(tmp_path, session=asked)
|
|
live, _ = decide.open_jobs(str(tmp_path), asked, me)
|
|
assert [j for _c, j, _s in live] == [other]
|
|
live, _ = decide.open_jobs(str(tmp_path), "herdr:another", me)
|
|
assert live == []
|
|
|
|
|
|
def test_u1c_the_job_itself_and_unreadable_records_are_skipped(tmp_path):
|
|
me = _register(tmp_path)
|
|
(tmp_path / "broken1.json").write_text("{not json")
|
|
assert decide.open_jobs(str(tmp_path), "herdr:s", me) == ([], [])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# U2 - FIFO release and the bound (wait_s=0: an immediate decision)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_u2_clear_only_when_no_live_running_and_this_job_is_the_earliest_pending(tmp_path, capsys):
|
|
a = _register(tmp_path, status="running")
|
|
_event(tmp_path, a, time.time())
|
|
me = _register(tmp_path)
|
|
assert decide.wait_open_jobs(_ns(tmp_path, me)) == f"busy {a}"
|
|
# A ends: clear
|
|
_edit(tmp_path, a, status="completed")
|
|
assert decide.wait_open_jobs(_ns(tmp_path, me)) == "clear"
|
|
# an earlier live pending waiter B: this job waits behind it (busy at the bound)
|
|
b = _register(tmp_path, created_at=_iso(time.time() - 30))
|
|
assert decide.wait_open_jobs(_ns(tmp_path, me)) == f"busy {b}"
|
|
# a later one does not hold this job up
|
|
_edit(tmp_path, b, created_at=_iso(time.time() + 30))
|
|
assert decide.wait_open_jobs(_ns(tmp_path, me)) == "clear"
|
|
|
|
|
|
def test_u2b_a_tie_on_created_at_is_broken_by_job_id(tmp_path):
|
|
same = _iso(time.time())
|
|
_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() - 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 {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):
|
|
a = _register(tmp_path, status="running")
|
|
_event(tmp_path, a, time.time())
|
|
me = _register(tmp_path)
|
|
t0 = time.time()
|
|
# 600 s asked for, 0 s of this job's budget left: decided at once
|
|
assert decide.wait_open_jobs(_ns(tmp_path, me, wait_s=600, start_time=int(time.time()) - 100, timeout=50)) == f"busy {a}"
|
|
assert time.time() - t0 < 5
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# U3 - the waiter's signal goes through the registry, status untouched
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_u3_the_waiter_signals_on_its_own_record(tmp_path):
|
|
a = _register(tmp_path, status="running")
|
|
_event(tmp_path, a, time.time())
|
|
me = _register(tmp_path)
|
|
before = json.loads((tmp_path / f"{me}.json").read_text())
|
|
assert "dispatch_wait_at" not in before
|
|
t0 = time.time()
|
|
assert decide.wait_open_jobs(_ns(tmp_path, me, wait_s=2)) == f"busy {a}"
|
|
assert 1.5 <= time.time() - t0 < 10
|
|
after = json.loads((tmp_path / f"{me}.json").read_text())
|
|
assert after["status"] == "pending"
|
|
assert calendar.timegm(time.strptime(after["dispatch_wait_at"], "%Y-%m-%dT%H:%M:%SZ")) >= int(t0) - 1
|
|
assert after["transition_seq"] == before.get("transition_seq", 0) + 1, "one merge per DISPATCH_SIGNAL_EVERY"
|
|
assert not (tmp_path / f"{me}.events.log").exists(), "events.log is the agent's"
|
|
# the signal is what a later waiter judges this record by
|
|
later = _register(tmp_path)
|
|
live, stale = decide.open_jobs(str(tmp_path), "herdr:s", later)
|
|
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): a
|
|
# terminal record is not this attempt's pending one, the waiter ends at
|
|
# once without a write (superseded) and is in no other waiter's FIFO
|
|
a = _register(tmp_path, status="running")
|
|
_event(tmp_path, a, time.time())
|
|
me = _register(tmp_path, status="completed")
|
|
assert decide.wait_open_jobs(_ns(tmp_path, me, wait_s=1)) == "superseded"
|
|
after = json.loads((tmp_path / f"{me}.json").read_text())
|
|
assert after["status"] == "completed" and "dispatch_wait_at" not in after
|
|
later = _register(tmp_path)
|
|
live, _stale = decide.open_jobs(str(tmp_path), "herdr:s", later)
|
|
assert me not in {j for _c, j, _s in live}
|
|
|
|
|
|
def test_u3b_the_signal_is_a_registry_merge_guarded_by_the_pinned_identity(tmp_path, monkeypatch):
|
|
from lib_py.delegation import engine
|
|
me = _register(tmp_path)
|
|
tok = _rec(tmp_path, me)["dispatch_token"]
|
|
seen = {}
|
|
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# U4 - the CLI: open-jobs (diagnostic) and a wrapper that is gone
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _cli(reg, *args):
|
|
return subprocess.run([sys.executable, "-m", "lib_py.delegation.wrapper", *args], capture_output=True,
|
|
text=True, cwd=str(SKILLS), env=dict(os.environ, PYTHONPATH=str(SKILLS)), timeout=60)
|
|
|
|
|
|
def test_u4_open_jobs_prints_live_and_stale(tmp_path):
|
|
a = _register(tmp_path, status="running")
|
|
_event(tmp_path, a, time.time())
|
|
b = _register(tmp_path, created_at=_iso(time.time() - 1000))
|
|
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
|
|
assert f"{b} stale pending, no dispatch signal and no started within 180s" in res.stdout
|
|
|
|
|
|
def _hold_pids(jid):
|
|
out = subprocess.run(["pgrep", "-f", f"hold-admission --job {jid}"], capture_output=True, text=True).stdout
|
|
return [int(p) for p in out.split()]
|
|
|
|
|
|
def test_u4c_the_wait_holds_its_admission_until_the_record_leaves_pending(tmp_path):
|
|
# review aa1bff92 F2: after `clear` the waiter ends with its answer; the
|
|
# hold renews the admitted record's signal until the agent's `started`
|
|
me = _register(tmp_path)
|
|
try:
|
|
res = _cli(tmp_path, "wait-open-jobs", "--job", me, "--agent-session", "herdr:s", "--registry-dir",
|
|
str(tmp_path), "--wait-s", "0", "--timeout", "3600", "--watch-pid", str(os.getpid()), "--hold")
|
|
assert res.stdout.strip() == "clear", res.stderr
|
|
first = _rec(tmp_path, me)["dispatch_wait_at"]
|
|
assert _hold_pids(me), "a detached hold is running"
|
|
deadline = time.time() + decide.DISPATCH_SIGNAL_EVERY + 5
|
|
while time.time() < deadline and _rec(tmp_path, me)["dispatch_wait_at"] == first:
|
|
time.sleep(0.5)
|
|
assert _at(_rec(tmp_path, me)["dispatch_wait_at"]) >= _at(first) + decide.DISPATCH_SIGNAL_EVERY - 1
|
|
assert _rec(tmp_path, me)["status"] == "pending" and not (tmp_path / f"{me}.events.log").exists()
|
|
_edit(tmp_path, me, status="running") # the agent's `started`
|
|
deadline = time.time() + 5
|
|
while time.time() < deadline and _hold_pids(me):
|
|
time.sleep(0.2)
|
|
assert not _hold_pids(me), "released with the record"
|
|
finally:
|
|
for pid in _hold_pids(me):
|
|
os.kill(pid, 9)
|
|
|
|
|
|
def _hold(reg, jid, token, *extra, session="herdr:s"):
|
|
return _cli(reg, "hold-admission", "--job", jid, "--agent-session", session, "--dispatch-token", token,
|
|
"--registry-dir", str(reg), "--timeout", "3600", *extra)
|
|
|
|
|
|
def test_u4d_the_hold_ends_with_its_wrapper_and_without_an_admission(tmp_path):
|
|
me = _register(tmp_path)
|
|
tok = _rec(tmp_path, me)["dispatch_token"]
|
|
res = _hold(tmp_path, me, tok)
|
|
assert res.stdout.strip() == "released", res.stderr # never admitted: nothing to hold
|
|
assert decide.wait_open_jobs(_ns(tmp_path, me)) == "clear"
|
|
gone = subprocess.Popen(["true"])
|
|
gone.wait()
|
|
t0 = time.time()
|
|
res = _hold(tmp_path, me, tok, "--watch-pid", str(gone.pid))
|
|
assert res.stdout == "" and "is gone" in res.stderr and time.time() - t0 < 10, res.stdout + res.stderr
|
|
|
|
|
|
def test_u4e_a_hold_that_starts_late_holds_only_the_attempt_it_was_given(tmp_path):
|
|
# review 55219be1 R1: the token changed (or the record moved) between the
|
|
# admission and the hold's start - the hold is released without a write
|
|
me = _register(tmp_path)
|
|
assert decide.wait_open_jobs(_ns(tmp_path, me)) == "clear"
|
|
admitted = _rec(tmp_path, me)
|
|
tok = admitted["dispatch_token"]
|
|
res = _hold(tmp_path, me, tok, session="herdr:t") # not the session it admitted in
|
|
assert res.stdout.strip() == "released" and _rec(tmp_path, me) == admitted, res.stderr
|
|
_redispatch(tmp_path, me)
|
|
moved = _rec(tmp_path, me)
|
|
assert moved["dispatch_token"] != tok
|
|
res = _hold(tmp_path, me, tok)
|
|
assert res.stdout.strip() == "released" and _rec(tmp_path, me) == moved, res.stderr
|
|
assert "dispatch_admitted_token" in moved and moved["dispatch_admitted_token"] == tok, \
|
|
"attempt 1's admission is left as it is - and names no current token (u3d)"
|
|
|
|
|
|
def test_u4f_a_running_hold_is_released_by_a_redispatch_and_renews_nothing_after_it(tmp_path):
|
|
me = _register(tmp_path)
|
|
assert decide.wait_open_jobs(_ns(tmp_path, me)) == "clear"
|
|
tok = _rec(tmp_path, me)["dispatch_token"]
|
|
first = _rec(tmp_path, me)["dispatch_wait_at"]
|
|
proc = subprocess.Popen([sys.executable, "-m", "lib_py.delegation.wrapper", "hold-admission", "--job", me,
|
|
"--agent-session", "herdr:s", "--dispatch-token", tok, "--registry-dir", str(tmp_path),
|
|
"--timeout", "3600", "--watch-pid", str(os.getpid())],
|
|
cwd=str(SKILLS), env=dict(os.environ, PYTHONPATH=str(SKILLS)), stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE, text=True)
|
|
try:
|
|
deadline = time.time() + decide.DISPATCH_SIGNAL_EVERY + 5
|
|
while time.time() < deadline and _rec(tmp_path, me)["dispatch_wait_at"] == first:
|
|
time.sleep(0.5)
|
|
assert _rec(tmp_path, me)["dispatch_wait_at"] != first, "it renewed while the admission stood"
|
|
_redispatch(tmp_path, me)
|
|
moved = _rec(tmp_path, me)
|
|
out, err = proc.communicate(timeout=10)
|
|
assert out.strip() == "released", (out, err)
|
|
assert _rec(tmp_path, me) == moved, "nothing written after the redispatch"
|
|
finally:
|
|
if proc.poll() is None:
|
|
proc.kill()
|
|
proc.communicate()
|
|
|
|
|
|
def test_u4b_wait_open_jobs_stops_with_its_wrapper(tmp_path):
|
|
a = _register(tmp_path, status="running")
|
|
_event(tmp_path, a, time.time())
|
|
me = _register(tmp_path)
|
|
gone = subprocess.Popen(["true"])
|
|
gone.wait()
|
|
res = _cli(tmp_path, "wait-open-jobs", "--job", me, "--agent-session", "herdr:s", "--registry-dir", str(tmp_path),
|
|
"--wait-s", "30", "--timeout", "3600", "--watch-pid", str(gone.pid))
|
|
assert res.stdout == "" and "is gone" in res.stderr, res.stdout + res.stderr
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# U5 - finalize-end: phase registry_wait closes as cancelled/session_busy
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_u5_finalize_end_registry_wait_is_a_session_busy_cancel(tmp_path):
|
|
reg = tmp_path / "reg"
|
|
reg.mkdir()
|
|
me = _register(reg)
|
|
scripts = tmp_path / "scripts"
|
|
scripts.mkdir()
|
|
log = tmp_path / "calls.log"
|
|
for name in ("registry.py", "publish_event.py"):
|
|
(scripts / name).write_text("import sys\n"
|
|
f"open({str(log)!r}, 'a').write(sys.argv[0].split('/')[-1] + ' ' + ' '.join(sys.argv[1:]) + '\\n')\n")
|
|
res = subprocess.run([sys.executable, "-m", "lib_py.delegation.wrapper", "finalize-end", "--job", me,
|
|
"--registry-dir", str(reg), "--status", "error", "--timeout", "60", "--idle-timeout", "10",
|
|
"--sub-rc", "1", "--phase", "registry_wait", "--type", "direct", "--final", "error",
|
|
"--python", sys.executable, "--scripts-dir", str(scripts)],
|
|
capture_output=True, text=True, cwd=str(SKILLS), env=dict(os.environ, PYTHONPATH=str(SKILLS)),
|
|
timeout=60)
|
|
assert res.returncode == 0, res.stderr
|
|
calls = log.read_text().splitlines()
|
|
assert any(c.startswith("registry.py") and "close --job " + me + " --request cancelled --detail session_busy" in c
|
|
for c in calls), calls
|
|
assert any(c.startswith("publish_event.py") and "--event cancelled --detail session_busy" in c for c in calls), calls
|
|
assert "closed by" not in res.stderr
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# S - the bash wrapper calls the decision in both dispatch paths
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_s1_both_dispatch_paths_wait_by_the_registry_before_the_idle_wait():
|
|
src = WRAPPER.read_text()
|
|
# ... 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"')
|
|
assert loop < src.index('wait_for_agent_idle "$current_session"')
|
|
# the loop path waits on the record already reset to pending for this
|
|
# phase (review F1): a terminal self record cannot carry the signal (u3c)
|
|
assert src.count('--status "pending"') == 1
|
|
assert src.index('--status "pending"') < loop
|
|
# ... and reads the token that reset issued before it waits: a busy or
|
|
# idle_wait close from phase two on must carry this attempt's token
|
|
tok = src.index('_disp_tok="$(', src.index('--status "pending"'))
|
|
assert src.index('--status "pending"') < tok < loop
|
|
body = src.split("_registry_wait() {")[1].split("\n}\n")[0]
|
|
assert "wait-open-jobs" in body and 'finalize_unfinished "$job_id" "error" "$SUB_RC_ERROR" "registry_wait"' in body
|
|
assert "exit 5" in body and "mam_int_knob MAM_DISPATCH_WAIT_S 600 0" in body
|
|
assert "--hold" in body, "the admission is held alive after the answer (u4c)"
|
|
assert '${tok:+--dispatch-token "$tok"}' in body
|
|
superseded = body.split("superseded)")[1].split(";;")[0]
|
|
assert "exit 5" in superseded and "finalize_unfinished" not in superseded, "left to its owner: no close"
|