From e9efabf74bcae0e1f44cc4366d4bbb61b5d98690 Mon Sep 17 00:00:00 2001 From: Godopu Date: Sat, 10 Oct 2026 15:39:21 +0000 Subject: [PATCH] feat(issue-50): the delegate-job wrapper waits by the registry before it injects (C) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reproduction first (herdr 0.9.3, a real claude, one-off server, job 51ccd800): after a job that left a background subagent running (`← 1 agent` on screen for 5 min), herdr reported `working` only while the job ran and `idle` ~10 s after `completed`; a second submit with a 20 s idle wait was injected, not cancelled. The 2026-10-04 observation (0.9.1, 10+ min of `working`) does not reproduce on 0.9.3, so the herdr idle wait stays and the registry wait runs before it (plan §4.3 (나)). - lib_py/delegation/wrapper.py: `open-jobs` (diagnostic) and `wait-open-jobs`. An open job of the session is live when the #18 contract says so: `running` with an event in events.log within idle_timeout_sec; `pending` with a dispatch signal (dispatch_wait_at, else created_at) within ack_timeout_sec. A dead record is reported (`stale open job ignored (...)`) and never waited for. The waiter signals every 10 s by merging dispatch_wait_at into its own pending record through the registry's transition boundary (status untouched, no event), so later waiters see it alive; a waiter whose wrapper is gone stops and its signal goes stale (Creator's challenge, job bf8863e1). Release is FIFO by (created_at, job_id): `clear` when no live running job remains and this job is the earliest live pending one; `busy ` at the bound min(MAM_DISPATCH_WAIT_S, remaining wall clock). - the wrapper: _registry_wait after the preflight in both dispatch paths; `busy` closes the job as cancelled/session_busy (phase registry_wait), exit 5 - the existing contract. - tests/test_issue50_dispatch.py: the open-job table (status x age x signal x spelling), FIFO and the tie, the signal as a registry merge, the bound, a dead wrapper, finalize-end's new phase, both call sites. - tests/test_issue50_dispatch_herdr.py: a real herdr 0.9.3 server per test (short mkdtemp HOME, fake claude the server reports idle, the real publish path over the suite's broker): C1 wait-then-run, C2 the bound, C3 a dead running record, C4 the loop runner's first job behind an onboarding, C5 a session without a row, C7 two waiters in FIFO order with no cancellation, C8 a killed waiter skipped once stale; and three concurrent `create --onboard` / `worktree_prepare.sh --onboard` in one server (one server process survives the shims' concurrent bootstrap, so no bootstrap lock is needed). Co-Authored-By: Claude Fable 5.1 --- .agents/skills/lib_py/delegation/wrapper.py | 148 ++++- .../multi-agent-mux-delegate-job | 32 +- tests/test_issue50_dispatch.py | 273 +++++++++ tests/test_issue50_dispatch_herdr.py | 572 ++++++++++++++++++ 4 files changed, 1021 insertions(+), 4 deletions(-) create mode 100644 tests/test_issue50_dispatch.py create mode 100644 tests/test_issue50_dispatch_herdr.py diff --git a/.agents/skills/lib_py/delegation/wrapper.py b/.agents/skills/lib_py/delegation/wrapper.py index 3e2043f..eb55837 100644 --- a/.agents/skills/lib_py/delegation/wrapper.py +++ b/.agents/skills/lib_py/delegation/wrapper.py @@ -18,11 +18,16 @@ no answer) cannot pass for "no". The wrapper uses --print. finalize-begin / finalize-end ... finalize_unfinished, split around the one herdr diagnostic probe bash still runs -Issue #50 (create --onboard waits for the onboarding to end): +Issue #50 (the registry, not herdr's agent_status, says whether a session is +free, and create --onboard waits for the onboarding to end): wait-terminal --job J --registry-dir R --timeout T [--watch-pid P] prints the job's terminal status, or `timeout` / `missing` onboard-exit create's exit code for it: 0, 44, 45 + 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 ` 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 @@ -58,6 +63,9 @@ 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 +DISPATCH_SIGNAL_FIELD = "dispatch_wait_at" +DISPATCH_SIGNAL_EVERY = 10 def _now() -> int: @@ -203,6 +211,121 @@ def wait_terminal(a: argparse.Namespace) -> str: _sleep(a, 1) +# --------------------------------------------------------------------------- +# Issue #50 C: the registry says whether a session is free +# --------------------------------------------------------------------------- + +def _parse_ts(text: str) -> Optional[int]: + try: + return calendar.timegm(time.strptime(str(text), "%Y-%m-%dT%H:%M:%SZ")) + except (TypeError, ValueError): + return None + + +def _load_record(registry_dir: str, job_id: str) -> Optional[dict]: + try: + with open(os.path.join(registry_dir, f"{job_id}.json")) as fh: + rec = json.load(fh) + except Exception: + return None + return rec if isinstance(rec, dict) else None + + +def _int_field(rec: dict, key: str, default: int) -> int: + try: + return max(1, int(rec.get(key) or default)) + except (TypeError, ValueError): + return default + + +def open_jobs(registry_dir: str, session: str, job_id: str, now: Optional[int] = None): + """(live, stale) among the other non-terminal records of `session`. + + 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. + 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.""" + now = _now() if now is None else now + sess = _bare(session) + live, stale = [], [] + try: + names = sorted(os.listdir(registry_dir)) + except OSError: + return live, stale + for name in names: + if not name.endswith(".json"): + continue + other = name[:-len(".json")] + if other == job_id: + continue + rec = _load_record(registry_dir, other) + if rec is None or _bare(str(rec.get("agent_session") or "")) != sess: + continue + status = rec.get("status") + if status == "running": + idle = _int_field(rec, "idle_timeout_sec", DEFAULT_IDLE_TIMEOUT_SEC) + if recent_disk_heartbeat(registry_dir, other, idle): + live.append((str(rec.get("created_at") or ""), other, status)) + 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)) + else: + stale.append((other, f"pending, no dispatch signal for {ack}s")) + return live, stale + + +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, + no event in events.log - that log is the agent's).""" + 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) + + +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.""" + 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() + deadline = _now() + max(0, min(a.wait_s, remaining)) + reported = set() + last_signal = None + while True: + now = _now() + 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 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 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) + last_signal = now + _sleep(a, 1) + + def finalize_begin(a: argparse.Namespace) -> str: """`keep` (FINAL_JOB_STATUS untouched), `done `, or `go|probe `.""" @@ -289,7 +412,7 @@ def finalize_end(a: argparse.Namespace) -> str: clock = "wall" close_req = "wall_clock_timeout" if clock == "wall" and a.phase != "inject" else "deadline_timeout" detail = "" - if a.phase in ("preflight", "idle_wait"): + if a.phase in ("preflight", "idle_wait", "registry_wait"): close_req, detail = "cancelled", "session_busy" # Issue #34: a target the preflight could not see was never busy if a.phase == "preflight" and a.preflight_reason: @@ -347,6 +470,16 @@ def _parser() -> argparse.ArgumentParser: s.add_argument("--timeout", type=int, required=True) s.add_argument("--watch-pid", type=int, default=None, help="the caller's pid; gone -> stop waiting") sub.add_parser("onboard-exit").add_argument("status") + for name in ("open-jobs", "wait-open-jobs"): + s = sub.add_parser(name) + s.add_argument("--job", required=True) + s.add_argument("--agent-session", required=True) + s.add_argument("--registry-dir", required=True) + if name == "wait-open-jobs": + s.add_argument("--wait-s", type=int, required=True, help="MAM_DISPATCH_WAIT_S") + 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") for name in ("finalize-begin", "finalize-end"): s = sub.add_parser(name) s.add_argument("--job", required=True) @@ -400,6 +533,17 @@ def main(argv: Optional[List[str]] = None) -> int: print(f"wait-terminal: caller pid {a.watch_pid} is gone; stopped waiting on job {a.job}", file=sys.stderr) elif a.cmd == "onboard-exit": print(onboard_exit(a.status)) + elif a.cmd == "open-jobs": + live, stale = open_jobs(a.registry_dir, a.agent_session, a.job) + for created, job, status in live: + print(f"{job} {status:8s} created {created}") + for job, reason in stale: + print(f"{job} stale {reason}") + elif a.cmd == "wait-open-jobs": + try: + 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) 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 d697e1f..444f705 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 @@ -354,8 +354,11 @@ EOF exit 5 fi - # 2) TUI idle handshake + # 2) The registry, then the TUI idle handshake (Issue #50: the session's + # 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" local idle_rc=0 wait_for_agent_idle "$AGENT_SESSION" || idle_rc=$? export _SKS_IDLE_RC="$idle_rc" @@ -515,10 +518,11 @@ EOF exit 5 fi - # Idle wait + # Registry wait (Issue #50), then idle wait local cur_agent="$AGENT" [ "$_phase" = "reviewer" ] && cur_agent="$REVIEWER" if [ "$cur_agent" != "human" ] && ! _is_remote "$current_session"; then + _registry_wait "$JOB_ID" "$current_session" local idle_rc=0 wait_for_agent_idle "$current_session" || idle_rc=$? export _SKS_IDLE_RC="$idle_rc" @@ -835,6 +839,30 @@ _within_deadline() { wait "$pid" } +# _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. +_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 "$$" + case "$res" in + clear) return 0 ;; + 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" + local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}" + echo "$logs_root/$job_id" + exit 5 ;; + esac + _decision_failed wait-open-jobs "answer '$res'" +} + # finalize_unfinished [phase] [session] # Closes a job that did not complete (Issue #33 labels, ISSUE-14/18 disk grace, # Issue #34 cancel reasons) and sets FINAL_JOB_STATUS. The decisions, the disk diff --git a/tests/test_issue50_dispatch.py b/tests/test_issue50_dispatch.py new file mode 100644 index 0000000..03b5ca7 --- /dev/null +++ b/tests/test_issue50_dispatch.py @@ -0,0 +1,273 @@ +"""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_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 + for block in src.split("wait_for_agent_idle ")[1:]: + pass + 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"') + 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 diff --git a/tests/test_issue50_dispatch_herdr.py b/tests/test_issue50_dispatch_herdr.py new file mode 100644 index 0000000..358955b --- /dev/null +++ b/tests/test_issue50_dispatch_herdr.py @@ -0,0 +1,572 @@ +"""Issue #50, against a real herdr 0.9.3 server: the registry decides whether a +session is free (C), and three `create --onboard` at once do not contend (A). + +Each test gets a one-off herdr server (a session named t50-) under a +short mkdtemp HOME - herdr's sockets, logs and server config live there, the +user's herdr config is never touched, and a scratch HOME under tmp_path would +exceed the unix socket path limit. The pane runs a fake claude (a claude-like +welcome, then it echoes its input), which herdr 0.9.3 reports as `idle`; the +fake never shows `working`, so `agent prompt --wait` times out and the +injection is judged by the job's own `started` (Issue #18), which these tests +publish as the agent would, through the real publish_event.py and the suite's +local broker. The box is torn down after every test. + +Measured on 2026-10-10 (job 51ccd800, herdr 0.9.3, a real claude): herdr +reported `working` only while a job ran and `idle` ~10 s after `completed`, +background subagent (`← 1 agent`) or not, and a second job was dispatched; +so the herdr idle wait stays and the registry wait runs before it. +""" +from __future__ import annotations + +import json +import os +import re +import shutil +import signal +import subprocess +import sys +import tempfile +import threading +import time +import uuid +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" +PUBLISH_PY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.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)) + +from lib_py.delegation import registry # noqa: E402 + +HERDR = shutil.which("herdr", path="/home/linuxbrew/.linuxbrew/bin:/usr/local/bin:" + os.environ.get("PATH", "")) +pytestmark = pytest.mark.skipif(not HERDR, reason="herdr is not installed") + +SESSION = "t50-creator-claude" +FAKE_CLAUDE = r'''#!/usr/bin/env bash +# fake claude for the real-herdr tests: a claude-like welcome, then it echoes its input +if [ "${1:-}" = "auth" ]; then echo '{"loggedIn": true}'; exit 0; fi +printf '\n ▐▛███▛█ Claude Code v2.1.0 (fake)\n▝▜██████▀ Sonnet · Claude Max\n\n' +printf '───────────────────────────────────────────────\n❯ Try "how does work?"\n───────────────────────────────────────────────\n ⏵⏵ bypass permissions on (shift+tab to cycle)\n' +while IFS= read -r line; do printf '\n● got: %s\n❯ \n' "$line"; done +''' + + +class Box: + def __init__(self, home: Path, sess: str): + self.home, self.sess = home, sess + self.ctl, self.ws, self.bin = home / "ctl", home / "ws", home / "bin" + self.jobs = self.ctl / ".mam" / "jobs" + self.yaml = self.ctl / ".mam" / "agent-sessions.yaml" + + # -- environment ------------------------------------------------------- + def env(self, **over) -> dict: + env = {k: v for k, v in os.environ.items() + if not k.startswith("HERDR_") and k not in ("MAM_JOB_ID", "MAM_VERSION", "CLAUDE_PROJECT_DIR", + "HOME_DIR", "XDG_CONFIG_HOME")} + env.update(HOME=str(self.home), PATH=f"{self.bin}:{env.get('PATH', '')}", HERDR_SESSION_NAME=self.sess, + WORKSPACE_ROOT=str(self.ctl), AGENT_SESSIONS_YAML=str(self.yaml), MAM_JOBS_DIR=str(self.jobs), + DELEGATE_JOB_LOGS_DIR=str(self.ctl / ".mam" / "delegate_job_logs"), + MAM_ENV_FILE=str(self.ctl / ".mam.env.absent"), MAM_NO_SUBSCRIBER_DAEMON="1", + DELEGATE_JOB_PYTHON=sys.executable, AGENT_PYTHON_BIN=sys.executable, + MAM_ONBOARD_ACK_S="60") + env.update({k: str(v) for k, v in over.items()}) + return env + + def sh(self, script: str, timeout=120, **over) -> subprocess.CompletedProcess: + """A bash snippet after `source lib.sh` (so `herdr` is the shim).""" + return subprocess.run(["bash", "-c", f'source "{SKILLS}/lib.sh"\n{script}'], capture_output=True, text=True, + cwd=str(self.ctl), env=self.env(**over), timeout=timeout) + + def herdr(self, *args, timeout=30) -> subprocess.CompletedProcess: + """The real binary against this box's server (no shim).""" + env = {"HOME": str(self.home), "PATH": os.environ.get("PATH", "")} + return subprocess.run([HERDR, "--session", self.sess, *args], capture_output=True, text=True, env=env, + timeout=timeout) + + def agent_status(self, name=SESSION) -> str: + res = self.herdr("agent", "get", name) + try: + return json.loads(res.stdout)["result"]["agent"].get("agent_status", "") + except (ValueError, KeyError): + return "" + + def pane_text(self, name=SESSION) -> str: + res = self.herdr("agent", "read", name, "--source", "recent") + try: + d = json.loads(res.stdout)["result"] + return (d.get("read") or d).get("text", "") + except (ValueError, KeyError, AttributeError): + return res.stdout + + def server_pids(self) -> list: + out = subprocess.run(["pgrep", "-f", f"herdr --session {self.sess} server"], capture_output=True, text=True).stdout + return [int(p) for p in out.split()] + + # -- registry -------------------------------------------------------------- + def register(self, session=f"herdr:{SESSION}", **kw) -> str: + kw.setdefault("ack_timeout_sec", 60) + return registry.register_job(prompt="prior work", agent="claude-code", agent_session=session, + registry_dir=str(self.jobs), **kw) + + def record(self, jid: str) -> dict: + return json.loads((self.jobs / f"{jid}.json").read_text()) + + def events(self, jid: str) -> list: + path = self.jobs / f"{jid}.events.log" + if not path.exists(): + return [] + return [json.loads(l) for l in path.read_text().splitlines() if l.strip()] + + def publish(self, jid: str, event: str, detail: str = "") -> None: + rec = self.record(jid) + cmd = [sys.executable, str(PUBLISH_PY), "--registry-dir", str(self.jobs), "--job", jid, "--event", event, + "--expect-dispatch-token", rec["dispatch_token"], "--attempt-id", str(rec.get("attempt_id", 1))] + if detail: + cmd += ["--detail", detail] + res = subprocess.run(cmd, capture_output=True, text=True, env=self.env(), timeout=120) + assert res.returncode == 0, (event, res.stdout, res.stderr) + + def job_ids(self, session=f"herdr:{SESSION}") -> set: + out = set() + for p in self.jobs.glob("*.json"): + try: + if json.loads(p.read_text()).get("agent_session") == session: + out.add(p.stem) + except ValueError: + pass + return out + + def wait_new_job(self, known: set, timeout=60) -> str: + deadline = time.time() + timeout + while time.time() < deadline: + new = self.job_ids() - known + if new: + assert len(new) == 1, new + return new.pop() + time.sleep(0.2) + raise AssertionError("no new job was registered") + + def wait_field(self, jid: str, key: str, timeout=60) -> str: + deadline = time.time() + timeout + while time.time() < deadline: + if self.record(jid).get(key): + return self.record(jid)[key] + time.sleep(0.2) + raise AssertionError(f"{jid} never got {key}") + + # -- the wrapper --------------------------------------------------------------- + def submit(self, prompt="work", session=f"herdr:{SESSION}", timeout="300", ack="60", **over) -> subprocess.Popen: + out = tempfile.TemporaryFile("w+", dir=str(self.home)) + err = tempfile.TemporaryFile("w+", dir=str(self.home)) + proc = subprocess.Popen(["bash", str(WRAPPER), "submit", "--agent", "claude-code", "--agent-session", session, + "--role", "Worker", "--timeout", timeout, "--ack-timeout", ack, + "--registry-dir", str(self.jobs), "--prompt", prompt], + stdout=out, stderr=err, text=True, cwd=str(self.ctl), env=self.env(**over)) + proc._out, proc._err = out, err # type: ignore[attr-defined] + return proc + + @staticmethod + def finish(proc: subprocess.Popen, timeout=240): + """The wrapper's (rc, stdout, stderr), after every timer-driven publish + has returned: the wrapper ends as soon as the subscriber sees + `completed` on the wire, while publish_event's registry write can + land a moment later.""" + rc = proc.wait(timeout=timeout) + for t in list(_TIMERS): + t.join(timeout=60) + proc._out.seek(0); proc._err.seek(0) # type: ignore[attr-defined] + return rc, proc._out.read(), proc._err.read() # type: ignore[attr-defined] + + +def _row(sess: str, name=SESSION, ws="") -> str: + return (f"- name: {name}\n status: running\n role: creator\n herdr_session: {sess}\n herdr_server: {sess}\n" + f" herdr_workspace: t50\n delegate_job_id: null\n pane: {{index: 0, pid: 0, cmd: claude, cwd: {ws}}}\n") + + +def _make_box(sess: str) -> Box: + home = Path(tempfile.mkdtemp(prefix="mam50-")) # short: a unix socket path lives under it + box = Box(home, sess) + for d in (box.bin, box.ws, box.ctl / ".agents", box.jobs, box.ctl / ".mam" / "delegate_job_logs"): + d.mkdir(parents=True) + (box.ctl / ".agents" / "skills").symlink_to(SKILLS) + (box.bin / "claude").write_text(FAKE_CLAUDE) + (box.bin / "claude").chmod(0o755) + box.yaml.write_text("herdr_sessions:\n" + _row(sess, ws=str(box.ws))) + return box + + +def _start_pane(box: Box, name=SESSION, cwd=None) -> None: + res = box.sh(f'herdr new-session -d -s {name} -c "{cwd or box.ws}" "claude --dangerously-skip-permissions"') + assert res.returncode == 0, res.stdout + res.stderr + deadline = time.time() + 30 + while time.time() < deadline and box.agent_status(name) != "idle": + time.sleep(0.5) + assert box.agent_status(name) == "idle", box.herdr("agent", "list").stdout + + +def _teardown(box: Box) -> None: + if os.environ.get("MAM_T50_KEEP_BOX"): # debugging aid: leave the box and the server in place + print("KEEP_BOX", box.home, box.sess) + return + for sub in (("session", "stop", box.sess), ("session", "delete", box.sess)): + subprocess.run([HERDR, *sub], capture_output=True, text=True, env={"HOME": str(box.home), "PATH": os.environ["PATH"]}, + timeout=30) + time.sleep(0.5) + for pid in box.server_pids(): + try: + os.kill(pid, signal.SIGKILL) + except OSError: + pass + shutil.rmtree(box.home, ignore_errors=True) + + +@pytest.fixture +def box(local_mqtt_broker): + _TIMERS.clear() + b = _make_box(f"t50-{uuid.uuid4().hex[:8]}") + try: + _start_pane(b) + yield b + finally: + _teardown(b) + + +_TIMERS: list = [] + + +def _after(delay: float, fn, *args): + t = threading.Timer(delay, fn, args) + t.daemon = True + t.start() + _TIMERS.append(t) + return t + + +def _ts(ev: dict) -> str: + return ev["timestamp"] + + +# --------------------------------------------------------------------------- +# C1 - an open prior job ends, then the new one is injected +# --------------------------------------------------------------------------- + +def test_c1_a_job_waits_for_the_sessions_open_job_and_then_runs(box): + a = box.register() + box.publish(a, "started") # running, one event on disk + known = box.job_ids() + proc = box.submit(MAM_DISPATCH_WAIT_S="20") + b = box.wait_new_job(known) + _after(3, box.publish, a, "completed", "prior done") + _after(5, box.publish, b, "started") + _after(8, box.publish, b, "completed", "b done") + rc, out, err = box.finish(proc) + assert rc == 0, out + err + assert box.record(b)["status"] == "completed" and box.record(b).get("detail") is None + assert "cancelled" not in out and "stale open job" not in err + assert box.record(b).get("dispatch_wait_at"), "B waited (and signalled) while A was open" + assert _ts(box.events(b)[0]) >= _ts(box.events(a)[-1]), "B started only after A ended" + assert "got: Your job_id is \"" + b in box.pane_text().replace("\n", "") or b in box.pane_text() + + +# --------------------------------------------------------------------------- +# C2 - the bound: an open job that never ends cancels the new one as session_busy +# --------------------------------------------------------------------------- + +def test_c2_the_wait_bound_closes_the_job_as_session_busy_naming_the_open_one(box): + a = box.register() + box.publish(a, "started") + known = box.job_ids() + t0 = time.time() + proc = box.submit(MAM_DISPATCH_WAIT_S="3") + b = box.wait_new_job(known) + rc, out, err = box.finish(proc) + assert rc == 5, out + err + assert time.time() - t0 < 60 + rec = box.record(b) + assert rec["status"] == "cancelled" and rec.get("detail") == "session_busy", rec + assert a in err and "still has open jobs in the registry" in err, err + assert [e["event"] for e in box.events(b)] == ["cancelled"] + assert box.record(a)["status"] == "running", "the open job is left alone" + + +# --------------------------------------------------------------------------- +# C3 - a running record nothing has touched for idle_timeout_sec is stale +# --------------------------------------------------------------------------- + +def test_c3_a_dead_running_record_does_not_block_the_session(box): + a = box.register(idle_timeout_sec=2) + box.publish(a, "started") + time.sleep(3) # its only event is now older than idle_timeout_sec + 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 (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" + + +# --------------------------------------------------------------------------- +# C4 - the loop's first job to a creator whose onboarding is still running +# --------------------------------------------------------------------------- + +def test_c4_the_loops_first_job_waits_for_the_onboarding_to_end(box, monkeypatch): + onboarding = box.register(ack_timeout_sec=180) # create registers it exactly so: agent_session=herdr: + box.publish(onboarding, "started") + for k, v in box.env(MAM_DISPATCH_WAIT_S="30").items(): + monkeypatch.setenv(k, v) + monkeypatch.delenv("MAM_REAL_ROOT", raising=False) + from lib_py.loop import runner as lr + r = lr.LoopRunner(["--creator", SESSION, "--task", "first task"]) + known = box.job_ids() + result = {} + + def drive(): + first = box.wait_new_job(known) + result["first"] = first + time.sleep(5) + box.publish(onboarding, "completed", "Onboarding complete; aligned with role creator") + time.sleep(2) + box.publish(first, "started") + time.sleep(2) + box.publish(first, "completed", "first done") + + t = threading.Thread(target=drive, daemon=True) + t.start() + jid = r.delegate_job(SESSION, "claude-code", "Worker", "the loop's first job") # the runner adds herdr: + t.join(timeout=60) + assert jid == result.get("first"), (jid, result) + assert jid not in r.failed_submit_jobs + rec = box.record(jid) + assert rec["status"] == "completed" and rec.get("detail") is None, rec + assert _ts(box.events(jid)[0]) >= _ts(box.events(onboarding)[-1]) + + +# --------------------------------------------------------------------------- +# C5 - a session create never registered: the record's agent_session alone decides +# --------------------------------------------------------------------------- + +def test_c5_a_session_without_a_row_is_judged_by_the_records_alone(local_mqtt_broker): + # resolve_herdr_session falls back to the workspace slug when the yaml has no + # row, so the server is named exactly that: mam-- of the control root + home = Path(tempfile.mkdtemp(prefix="mam50-")) + sess = f"mam-{home.name}-ctl".lower().replace("_", "-") + shutil.rmtree(home) + _TIMERS.clear() + b = Box(home, sess) + for d in (b.bin, b.ws, b.ctl / ".agents", b.jobs, b.ctl / ".mam" / "delegate_job_logs"): + d.mkdir(parents=True) + (b.ctl / ".agents" / "skills").symlink_to(SKILLS) + (b.bin / "claude").write_text(FAKE_CLAUDE) + (b.bin / "claude").chmod(0o755) + b.yaml.write_text("herdr_sessions: []\n") + try: + _start_pane(b) + a = b.register() + b.publish(a, "started") + known = b.job_ids() + proc = b.submit(MAM_DISPATCH_WAIT_S="20") + c = b.wait_new_job(known) + _after(3, b.publish, a, "completed", "prior done") + _after(5, b.publish, c, "started") + _after(7, b.publish, c, "completed", "c done") + rc, out, err = b.finish(proc) + assert rc == 0, out + err + assert b.record(c)["status"] == "completed" and b.record(c).get("dispatch_wait_at") + assert _ts(b.events(c)[0]) >= _ts(b.events(a)[-1]) + finally: + _teardown(b) + + +# --------------------------------------------------------------------------- +# C7 - two waiters behind one session: FIFO, one at a time, no cancellation +# --------------------------------------------------------------------------- + +def test_c7_two_waiters_run_one_after_the_other_and_neither_is_cancelled(box): + a = box.register() + box.publish(a, "started") + known = box.job_ids() + proc_b = box.submit(prompt="work B", MAM_DISPATCH_WAIT_S="40") + b = box.wait_new_job(known) + time.sleep(1.2) + proc_c = box.submit(prompt="work C", MAM_DISPATCH_WAIT_S="40") + c = box.wait_new_job(known | {b}) + assert box.record(b)["created_at"] <= box.record(c)["created_at"] + box.wait_field(c, "dispatch_wait_at") # both are waiting (C signals too) + box.wait_field(b, "dispatch_wait_at") + _after(2, box.publish, a, "completed", "prior done") + _after(4, box.publish, b, "started") + _after(7, box.publish, b, "completed", "b done") + # C is published started only once B has ended - as the real agent, which + # cannot answer a prompt it has not been given, would + _after(9, box.publish, c, "started") + _after(11, 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 box.record(b)["status"] == box.record(c)["status"] == "completed" + 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" + 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" + + +# --------------------------------------------------------------------------- +# C8 - a waiter whose wrapper died: its signal goes stale and is skipped +# --------------------------------------------------------------------------- + +def test_c8_a_dead_waiter_is_skipped_once_its_signal_is_stale(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="120") + b = box.wait_new_job(known) + box.wait_field(b, "dispatch_wait_at") + os.kill(proc_b.pid, signal.SIGKILL) # the wrapper is gone; its waiter sees that and stops + proc_b.wait(timeout=30) + proc_c = box.submit(prompt="work C", MAM_DISPATCH_WAIT_S="60") + c = box.wait_new_job(known | {b}) + _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") + 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 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]) + + +# --------------------------------------------------------------------------- +# A - three `create --onboard` at once in one server +# --------------------------------------------------------------------------- + +def _drive_onboardings(box: Box, names: list, delays: list, result: dict): + seen = {} + deadline = time.time() + 120 + while len(seen) < len(names) and time.time() < deadline: + for p in box.jobs.glob("*.json"): + try: + rec = json.loads(p.read_text()) + except ValueError: + continue + sess = rec.get("agent_session", "") + if sess.startswith("herdr:") and sess[6:] in names and sess[6:] not in seen: + seen[sess[6:]] = p.stem + box.publish(p.stem, "started") + result.setdefault("started_at", {})[sess[6:]] = time.time() + time.sleep(0.3) + result["jobs"] = dict(seen) + for name, delay in zip(names, delays): + jid = seen.get(name) + if jid: + _after(delay, box.publish, jid, "completed", "Onboarding complete; aligned with role creator") + + +def test_a_three_concurrent_creates_with_onboarding(local_mqtt_broker): + b = _make_box(f"t50-{uuid.uuid4().hex[:8]}") + b.yaml.write_text("herdr_sessions: []\n") + names = [f"t50-{i}-creator-claude" for i in (1, 2, 3)] + procs = [] + try: + for i, name in enumerate(names, 1): + ws = b.home / f"ws{i}" + ws.mkdir() + out = tempfile.TemporaryFile("w+", dir=str(b.home)) + err = tempfile.TemporaryFile("w+", dir=str(b.home)) + procs.append((name, subprocess.Popen( + ["bash", str(CREATE_SH), "--workspace", str(ws), "--agent", "claude", "--role", "creator", + "--session", name, "--onboard", "--herdr-session", b.sess], + stdout=out, stderr=err, text=True, cwd=str(b.ctl), env=b.env(MAM_ONBOARD_WAIT_S="120")), out, err)) + t0 = time.time() + result = {} + drv = threading.Thread(target=_drive_onboardings, args=(b, names, [10, 15, 20], result), daemon=True) + drv.start() + outcome = {} + for name, proc, out, err in procs: + rc = proc.wait(timeout=240) + out.seek(0); err.seek(0) + outcome[name] = (rc, out.read(), err.read()) + elapsed = time.time() - t0 + drv.join(timeout=5) + assert all(o[0] == 0 for o in outcome.values()), {n: (o[0], o[2][-1500:]) for n, o in outcome.items()} + # The three start-ups overlap, and the onboarding waits do too: once the + # last prompt was delivered (its `started`), the whole run ends with the + # slowest onboarding (20 s) plus create's wrap-up, not with their sum (45 s). + last_started = max(result["started_at"].values()) - t0 + assert elapsed - last_started <= 20 + 15, (elapsed, last_started, result["started_at"]) + assert max(result["started_at"].values()) - min(result["started_at"].values()) <= 20, \ + ("the three creates did not run side by side", result["started_at"]) + for name in names: + jid = result["jobs"][name] + assert f"onboarding completed: job {jid} (" in outcome[name][1], outcome[name][1] + assert b.record(jid)["status"] == "completed" + rows = re.findall(r"^- name: (\S+)", b.yaml.read_text(), re.M) + assert sorted(rows) == sorted(names), rows + assert len(set(result["jobs"].values())) == 3 and all((b.jobs / f"{j}.json").exists() for j in result["jobs"].values()) + assert len(b.server_pids()) == 1, "three shims bootstrapped the server at once; one must remain" + assert len(json.loads(b.herdr("agent", "list").stdout)["result"]["agents"]) == 3 + finally: + _teardown(b) + + +def _git(repo: Path, *args): + subprocess.run(["git", "-c", "user.email=t@example.com", "-c", "user.name=t", *args], cwd=str(repo), check=True, + capture_output=True, text=True) + + +def test_a2_three_concurrent_worktree_prepares_with_onboarding(local_mqtt_broker): + """worktree_prepare.sh --onboard x3 (different --path, one server): the git + worktree step, the registry and the server's first start do not contend.""" + prep = SKILLS / "multi-agent-mux-create" / "scripts" / "worktree_prepare.sh" + b = _make_box(f"t50-{uuid.uuid4().hex[:8]}") + b.yaml.write_text("herdr_sessions: []\n") + _git(b.ctl, "init", "-q") + (b.ctl / "README").write_text("ctl\n") + _git(b.ctl, "add", "README") + _git(b.ctl, "commit", "-q", "-m", "init") + branches = [f"b{i}" for i in (1, 2, 3)] + names = [f"creator-claude-{br}" for br in branches] + procs = [] + try: + for br in branches: + out = tempfile.TemporaryFile("w+", dir=str(b.home)) + err = tempfile.TemporaryFile("w+", dir=str(b.home)) + procs.append((br, subprocess.Popen( + ["bash", str(prep), "--path", str(b.home / f"wt-{br}"), "--branch", br, "--creator-agent", "claude", + "--onboard", "--herdr-session", b.sess], + stdout=out, stderr=err, text=True, cwd=str(b.ctl), env=b.env(MAM_ONBOARD_WAIT_S="120")), out, err)) + t0 = time.time() + result = {} + drv = threading.Thread(target=_drive_onboardings, args=(b, names, [10, 15, 20], result), daemon=True) + drv.start() + outcome = {} + for br, proc, out, err in procs: + rc = proc.wait(timeout=300) + out.seek(0); err.seek(0) + outcome[br] = (rc, out.read(), err.read()) + elapsed = time.time() - t0 + drv.join(timeout=5) + assert all(o[0] == 0 for o in outcome.values()), {n: (o[0], o[2][-1500:]) for n, o in outcome.items()} + last_started = max(result["started_at"].values()) - t0 + assert elapsed - last_started <= 20 + 15, (elapsed, last_started, result["started_at"]) + for br in branches: + assert f"WORKTREE={b.home / f'wt-{br}'}" in outcome[br][1] and f"SESSION=creator-claude-{br}" in outcome[br][1] + assert (b.home / f"wt-{br}" / "README").exists() + assert b.record(result["jobs"][f"creator-claude-{br}"])["status"] == "completed" + listed = subprocess.run(["git", "worktree", "list"], cwd=str(b.ctl), capture_output=True, text=True).stdout + assert all(f"[{br}]" in listed for br in branches), listed + rows = re.findall(r"^- name: (\S+)", b.yaml.read_text(), re.M) + assert sorted(rows) == sorted(names), rows + assert len(b.server_pids()) == 1 + finally: + _teardown(b)