GodopuandClaude Fable 5.1
9e78ff3ba3
The `--status pending` reset is a redispatch and issues a new dispatch token, but the loop read `_disp_tok` only after the registry and idle waits. A busy or idle_wait close from phase two on therefore carried the previous phase's token and the engine refused it, leaving the record pending while the caller saw session_busy. The read now follows the reset directly; s1 pins reset < token read < registry wait. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
295 lines
14 KiB
Python
295 lines
14 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 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.
|
|
- 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 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"
|
|
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 _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,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 ack_timeout_sec (180)
|
|
("pending", 181, 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
|
|
("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)
|
|
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 ")
|
|
else:
|
|
assert got == (set(), set())
|
|
|
|
|
|
@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())
|
|
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")
|
|
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))
|
|
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
|
|
|
|
|
|
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_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
|
|
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}"
|
|
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(tmp_path, monkeypatch):
|
|
me = _register(tmp_path)
|
|
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"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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 and f"{b} stale pending, no dispatch signal for 180s" in res.stdout
|
|
|
|
|
|
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()
|
|
assert src.count('_registry_wait "$JOB_ID" "$AGENT_SESSION"') == 1
|
|
assert src.count('_registry_wait "$JOB_ID" "$current_session"') == 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
|