"""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())
        if status == "pending":   # never signalled: injected by its producer, a holder; signalled: a waiter in the FIFO
            assert live[0][2] == ("injected" if signal is None else "pending")
    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) == ([], [])


def test_u1e_a_dry_run_record_is_not_an_open_job(tmp_path):
    """review 1e25cf2f F1: `submit --dry-run` registers a record it never
    dispatches; it stays pending (Issue #34 shows it as an open job), but the
    admission must not wait for it."""
    reg = tmp_path / "jobs"
    env = dict(os.environ, HOME=str(tmp_path), MAM_ENV_FILE=str(tmp_path / ".mam.env.absent"))
    res = subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--agent-session", "herdr:s",
                          "--dry-run", "--registry-dir", str(reg), "--prompt", "preview only"],
                         capture_output=True, text=True, env=env, cwd=str(tmp_path), timeout=120)
    assert res.returncode == 0, res.stdout + res.stderr
    (dry,) = [p.stem for p in reg.glob("*.json")]
    assert _rec(reg, dry)["status"] == "pending" and _rec(reg, dry)["dry_run"] is True
    me = _register(reg)
    assert decide.open_jobs(str(reg), "herdr:s", me) == ([], [])
    assert decide.wait_open_jobs(_ns(reg, me)) == "clear"
    # control: the same record without the mark is an `injected` job that holds the session
    _edit(reg, dry, dry_run=None)
    assert decide.wait_open_jobs(_ns(reg, me)) == f"busy {dry}"


# ---------------------------------------------------------------------------
# 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 (signalled: a waiter is on it): this job waits behind it (busy at the bound)
    b = _register(tmp_path, created_at=_iso(time.time() - 30), dispatch_wait_at=_iso(time.time()))
    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, dispatch_wait_at=same)   # a live waiter
    assert decide.wait_open_jobs(_ns(tmp_path, "aaaa0001")) == "clear"
    assert decide.wait_open_jobs(_ns(tmp_path, "aaaa0002")) == "busy aaaa0001"


# review 5a09a13f F-A: a live pending record nobody signals on is one its
# producer injected without a waiter (create --onboard, a job sent by hand);
# the prompt is in the pane, so it holds the session - never a FIFO peer that
# a lower job id in the same second, or an earlier second, could pass

@pytest.mark.parametrize("onboarding,waiter", [("ffffffff", "00000001"), ("00000001", "ffffffff")])
def test_u2h_an_injected_record_holds_the_session_whatever_its_place_in_the_fifo(tmp_path, onboarding, waiter):
    same = _iso(time.time())
    _register_as(tmp_path, onboarding, created_at=same)                 # unsignalled, ack 180: live
    _register_as(tmp_path, waiter, created_at=same)
    assert decide.wait_open_jobs(_ns(tmp_path, waiter)) == f"busy {onboarding}"
    assert "dispatch_admitted_token" not in _rec(tmp_path, waiter) and "dispatch_wait_at" in _rec(tmp_path, waiter)
    assert "dispatch_wait_at" not in _rec(tmp_path, onboarding)
    # registered after the waiter (which was still in its preflight): the same
    _edit(tmp_path, onboarding, created_at=_iso(_at(same) + 1))
    assert decide.wait_open_jobs(_ns(tmp_path, waiter)) == f"busy {onboarding}"
    # control: the same record with a waiter's signal on it is a FIFO peer
    _edit(tmp_path, onboarding, created_at=same, dispatch_wait_at=same)
    assert decide.wait_open_jobs(_ns(tmp_path, waiter)) == ("clear" if waiter < onboarding else f"busy {onboarding}")


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_u3i_an_attempt_already_running_under_its_own_token_is_clear_not_superseded(tmp_path):
    # the Issue #14 registry ack: the agent's `started` for this attempt lands
    # before the prompt goes out (test_issue30_stall_watch_stop). Nothing to
    # wait for, nothing foreign: `clear` at once, no admission written
    a = _register(tmp_path, status="running")
    _event(tmp_path, a, time.time())
    me = _register(tmp_path)
    tok = _rec(tmp_path, me)["dispatch_token"]
    _edit(tmp_path, me, status="running")
    before = _rec(tmp_path, me)
    t0 = time.time()
    assert decide.wait_open_jobs(_ns(tmp_path, me, wait_s=30)) == "clear"
    assert time.time() - t0 < 5 and _rec(tmp_path, me) == before, "no wait, no write"
    res = _cli(tmp_path, "wait-open-jobs", "--job", me, "--agent-session", "herdr:s", "--registry-dir", str(tmp_path),
               "--wait-s", "30", "--timeout", "3600", "--dispatch-token", tok, "--hold")
    assert res.stdout.strip() == "clear" and not _hold_pids(me), res.stderr
    # running under another attempt's token, or in another session, is still someone else's
    assert decide.wait_open_jobs(_ns(tmp_path, me, session="herdr:t")) == "superseded"
    _edit(tmp_path, me, dispatch_token="another-attempt")
    res = _cli(tmp_path, "wait-open-jobs", "--job", me, "--agent-session", "herdr:s", "--registry-dir", str(tmp_path),
               "--wait-s", "30", "--timeout", "3600", "--dispatch-token", tok)
    assert res.stdout.strip() == "superseded", res.stderr
    _edit(tmp_path, me, dispatch_token=tok, status="completed")
    assert decide.wait_open_jobs(_ns(tmp_path, me)) == "superseded"


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))
    c = _register(tmp_path)                                  # within its ack budget, no waiter on it
    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
    assert f"{c}  injected" 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"