multi-agent-mux/tests/test_issue18_liveness_contract.py

#!/usr/bin/env python3
"""Issue #18 — liveness and delivery are decided by the event protocol, not by
screen scraping.

Contract (registry fields, wrapper options):
  * heartbeat obligation  - the agent publishes `progress` at least every N s;
    `publish_event --event started --heartbeat N` starts an automatic timer
    that does it for the agent and stops itself at the terminal event.
  * unresponsive          - no event (disk or wire) for K x N s closes the job
    as deadline_timeout; herdr status and pane text are diagnostics only.
  * delivery              - `started` must arrive within T s of the injection,
    else the inject fails closed (rc 1, registry error, single injection).
  * wire durability       - every event is followed by a retained
    `<prefix>/state` message on the same connection; a subscriber that
    (re)connects recovers the status at once and records seq gaps.
  * remote transport      - `--agent-session remote:<name>` delivers the brief
    over a signed retained `<prefix>/dispatch`; a runner on another host with
    its own registry and broker credentials runs it, and the orchestrator's
    subscriber mirrors its events into the orchestrator's registry.

Tests marked "broker" talk to the configured broker (the public default in the
sandbox); each stays under a minute.
"""
from __future__ import annotations

import json
import os
import re
import signal
import subprocess
import sys
import threading
import time
import uuid
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parent.parent
SKILLS = REPO_ROOT / ".agents" / "skills"
LIB_SH = SKILLS / "lib.sh"
WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
SCRIPTS = SKILLS / "multi-agent-mux-delegate-job" / "scripts"
PUBLISH_PY = SCRIPTS / "publish_event.py"
REGISTRY_PY = SCRIPTS / "registry.py"
REMOTE_PY = SCRIPTS / "remote.py"

UNREACHABLE = {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "MQTT_TLS": "0"}


def _env(cwd: Path, extra=None) -> dict:
    env = dict(os.environ)
    env["DELEGATE_JOB_PYTHON"] = sys.executable
    env["AGENT_PYTHON_BIN"] = sys.executable
    env["WORKSPACE_ROOT"] = str(cwd)
    env["HOME"] = str(cwd)
    env["MAM_ENV_FILE"] = str(cwd / ".mam.env.absent")
    env["DELEGATE_JOB_LOGS_DIR"] = str(cwd / "logs")
    env["PYTHONPATH"] = str(SKILLS)
    for k in ("MQTT_BROKER", "MQTT_HOST", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD"):
        env.pop(k, None)
    if extra:
        env.update({k: str(v) for k, v in extra.items()})
    return env


def _events(reg_dir: Path, jid: str) -> list:
    p = reg_dir / f"{jid}.events.log"
    if not p.exists():
        return []
    return [json.loads(l) for l in p.read_text().splitlines() if l.strip()]


def _job(reg_dir: Path, jid: str) -> dict:
    return json.loads((reg_dir / f"{jid}.json").read_text())


def _wait_job_id(reg_dir: Path, timeout=20) -> str:
    deadline = time.time() + timeout
    while time.time() < deadline:
        files = list(reg_dir.glob("*.json"))
        if files:
            return files[0].stem
        time.sleep(0.1)
    raise AssertionError("job was not registered")


def _publish(reg_dir: Path, jid: str, event: str, env: dict, *extra) -> subprocess.CompletedProcess:
    """Publish as the agent would: with the dispatch identity the wrapper's
    injected command carries (`--expect-dispatch-token`)."""
    tok = _job(reg_dir, jid).get("dispatch_token")
    ident = ["--expect-dispatch-token", tok] if tok else []
    return subprocess.run([sys.executable, str(PUBLISH_PY), "--registry-dir", str(reg_dir), "--job", jid,
                           "--event", event, "--attempts", "2", *ident, *extra],
                          capture_output=True, text=True, env=env, cwd=str(reg_dir.parent), timeout=120)


def _timer_pid(reg_dir: Path, jid: str):
    p = reg_dir / f"{jid}.heartbeat.pid"
    try:
        return int(p.read_text().strip())
    except (OSError, ValueError):
        return None


def _alive(pid) -> bool:
    if not pid:
        return False
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    return subprocess.run(["ps", "-o", "stat=", "-p", str(pid)], capture_output=True, text=True).stdout.strip() not in ("", "Z", "Z+")


def _kill_timer(reg_dir: Path, jid: str):
    pid = _timer_pid(reg_dir, jid)
    if pid:
        try:
            os.kill(pid, signal.SIGKILL)
        except OSError:
            pass


# ---------------------------------------------------------------------------
# 1. registry: the contract is part of the record
# ---------------------------------------------------------------------------

def test_register_derives_the_idle_window_from_the_contract(tmp_path, monkeypatch):
    from lib_py.delegation import registry

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    reg = tmp_path / "jobs"
    jid = registry.register_job("contract", registry_dir=str(reg))
    rec = _job(reg, jid)
    assert (rec["heartbeat_sec"], rec["heartbeat_misses"], rec["ack_timeout_sec"]) == (30, 4, 180)
    assert rec["idle_timeout_sec"] == 120 == registry.liveness_window(30, 4)
    assert rec["transport"] == "herdr"

    jid2 = registry.register_job("contract", registry_dir=str(reg), heartbeat_sec=5, heartbeat_misses=3)
    assert _job(reg, jid2)["idle_timeout_sec"] == 15
    jid3 = registry.register_job("contract", registry_dir=str(reg), heartbeat_sec=5, heartbeat_misses=3,
                                 idle_timeout_sec=40)
    assert _job(reg, jid3)["idle_timeout_sec"] == 40, "an explicit idle timeout is kept"


def test_wrapper_registers_the_contract_and_derives_idle_timeout(tmp_path):
    """--dry-run registers the job without any herdr; K x N shows up as idle."""
    reg = tmp_path / "jobs"
    reg.mkdir()
    res = subprocess.run(
        ["bash", str(WRAPPER), "submit", "--agent", "human", "--prompt", "contract", "--dry-run",
         "--registry-dir", str(reg), "--workdir", str(tmp_path),
         "--heartbeat", "7", "--heartbeat-misses", "3", "--ack-timeout", "11"],
        capture_output=True, text=True, env=_env(tmp_path, UNREACHABLE), cwd=str(tmp_path), timeout=60)
    assert res.returncode == 0, res.stderr
    jid = _wait_job_id(reg)
    rec = _job(reg, jid)
    assert (rec["heartbeat_sec"], rec["heartbeat_misses"], rec["ack_timeout_sec"], rec["idle_timeout_sec"]) == (7, 3, 11, 21)
    assert "--event started --heartbeat 7" in res.stdout, "the injected start command must arm the timer"
    assert "within 11 s" in res.stdout


# ---------------------------------------------------------------------------
# 2. the automatic heartbeat (no broker needed: events land on disk first)
# ---------------------------------------------------------------------------

def test_started_with_heartbeat_beats_on_disk_until_the_terminal_event(tmp_path):
    reg = tmp_path / "jobs"
    env = _env(tmp_path, UNREACHABLE)
    jid = subprocess.run([sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg), "register",
                          "--prompt", "hb", "--heartbeat-sec", "1", "--heartbeat-misses", "3"],
                         capture_output=True, text=True, env=env, cwd=str(tmp_path)).stdout.strip()
    assert jid
    try:
        res = _publish(reg, jid, "started", env, "--heartbeat", "1", "--attempts", "1")
        assert res.returncode == 2, "the broker is unreachable, the event is still recorded locally"
        pid = _timer_pid(reg, jid)
        assert pid and _alive(pid), "a detached timer must be running"
        time.sleep(4.5)
        beats = [e for e in _events(reg, jid) if e["event"] == "progress"]
        assert len(beats) >= 3, [e["event"] for e in _events(reg, jid)]
        assert all(e["detail"] == "heartbeat" for e in beats)
        assert all(e["data"].get("dispatch_token") == _job(reg, jid)["dispatch_token"] for e in beats), \
            "heartbeats carry the dispatch identity"
        seqs = [e["seq"] for e in _events(reg, jid)]
        assert seqs == sorted(seqs) and len(set(seqs)) == len(seqs)

        res = _publish(reg, jid, "completed", env, "--attempts", "1")
        assert res.returncode == 2
        deadline = time.time() + 5
        while time.time() < deadline and _alive(pid):
            time.sleep(0.1)
        assert not _alive(pid), "the terminal event must stop the timer"
        assert _timer_pid(reg, jid) is None
        assert _events(reg, jid)[-1]["event"] == "completed", "no heartbeat may trail the terminal event"
        assert _job(reg, jid)["status"] == "completed"
    finally:
        _kill_timer(reg, jid)


def test_heartbeat_timer_stops_on_its_own_when_the_job_is_redispatched(tmp_path):
    reg = tmp_path / "jobs"
    env = _env(tmp_path, UNREACHABLE)
    jid = subprocess.run([sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg), "register",
                          "--prompt", "hb"], capture_output=True, text=True, env=env, cwd=str(tmp_path)).stdout.strip()
    try:
        _publish(reg, jid, "started", env, "--heartbeat", "1", "--attempts", "1")
        pid = _timer_pid(reg, jid)
        assert _alive(pid)
        # An operator moves the job to the next attempt: the old attempt's timer must not keep beating.
        _publish(reg, jid, "completed", env, "--attempts", "1")
        subprocess.run([sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg), "update", "--job", jid,
                        "--status", "pending", "--attempt-id", "2"], capture_output=True, text=True, env=env, cwd=str(tmp_path))
        deadline = time.time() + 5
        while time.time() < deadline and _alive(pid):
            time.sleep(0.1)
        assert not _alive(pid)
    finally:
        _kill_timer(reg, jid)


def test_heartbeat_stops_when_the_watched_agent_process_dies(tmp_path):
    """Review B2: the timer follows the agent's life. The 'agent' here is a
    shell that runs the start command with --watch-pid $$ and then idles;
    killing that shell must stop the heartbeats within about one period."""
    reg = tmp_path / "jobs"
    env = _env(tmp_path, UNREACHABLE)
    jid = subprocess.run([sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg), "register",
                          "--prompt", "hb"], capture_output=True, text=True, env=env, cwd=str(tmp_path)).stdout.strip()
    agent = subprocess.Popen(
        ["bash", "-c", f'"{sys.executable}" "{PUBLISH_PY}" --registry-dir "{reg}" --job {jid} --event started '
                       f'--heartbeat 1 --attempts 1 --watch-pid $$ >/dev/null 2>&1; sleep 120'],
        cwd=str(tmp_path), env=env)
    try:
        deadline = time.time() + 15
        while time.time() < deadline and not _timer_pid(reg, jid):
            time.sleep(0.1)
        pid = _timer_pid(reg, jid)
        assert pid and _alive(pid)
        time.sleep(2.5)
        before = len([e for e in _events(reg, jid) if e["event"] == "progress"])
        assert before >= 2
        agent.kill()  # the agent crashes
        agent.wait(timeout=10)
        deadline = time.time() + 6
        while time.time() < deadline and _alive(pid):
            time.sleep(0.1)
        assert not _alive(pid), "timer must stop once the watched agent is gone"
        time.sleep(1.5)
        after = len([e for e in _events(reg, jid) if e["event"] == "progress"])
        assert after <= before + 2, f"heartbeats kept coming after the agent died: {before} -> {after}"
        assert _job(reg, jid)["status"] == "running", "the timer stops; the orchestrator's K x N closes the job"
    finally:
        if agent.poll() is None:
            agent.kill()
        _kill_timer(reg, jid)


def test_concurrent_start_timer_calls_leave_exactly_one_tracked_owner(tmp_path, monkeypatch):
    """Review round 4 (codex R3): two `start_timer` calls for the same
    job/attempt at once - both see no timer, both spawn, one pidfile - left an
    untracked timer that could publish after the terminal event. The lifecycle
    lock serialises observe/drain/spawn/publish-pid, so exactly one timer
    survives, the pidfile names it, and the terminal stop drains it."""
    from lib_py.delegation import heartbeat, registry

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    monkeypatch.setenv("MQTT_BROKER", "127.0.0.1")
    monkeypatch.setenv("MQTT_PORT", "9")
    reg = tmp_path / "jobs"
    jid = registry.register_job("race", registry_dir=str(reg))
    tok = _job(reg, jid)["dispatch_token"]
    pids = []
    errors = []

    def start():
        try:
            pids.append(heartbeat.start_timer(jid, str(reg), 1, dispatch_token=tok, attempt_id=1, watch_pid=0))
        except Exception as exc:  # pragma: no cover - surfaced below
            errors.append(exc)

    threads = [threading.Thread(target=start) for _ in range(2)]
    for t in threads:
        t.start()
    for t in threads:
        t.join(timeout=60)
    try:
        assert not errors, errors
        assert len(pids) == 2 and all(pids)
        time.sleep(1.0)
        alive = [p for p in pids if _alive(p)]
        assert len(alive) == 1, f"exactly one timer may survive two concurrent starts, alive={alive} pids={pids}"
        assert _timer_pid(reg, jid) == alive[0], "the pidfile names the surviving owner"
        # the terminal path drains the tracked owner and nothing else is left
        assert heartbeat.stop_timer(jid, str(reg), drain=5.0) is True
        assert not any(_alive(p) for p in pids)
        assert _timer_pid(reg, jid) is None
    finally:
        for p in pids:
            try:
                os.kill(p, signal.SIGKILL)
            except OSError:
                pass


def test_start_timer_after_the_terminal_barrier_does_not_arm(tmp_path, monkeypatch):
    """A `started --heartbeat` that loses the race against the terminal
    publish must not leave a timer behind."""
    from lib_py.delegation import heartbeat, registry

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    reg = tmp_path / "jobs"
    jid = registry.register_job("late-start", registry_dir=str(reg))
    tok = _job(reg, jid)["dispatch_token"]
    heartbeat.raise_barrier(jid, str(reg), dispatch_token=tok, attempt_id=1)
    assert heartbeat.start_timer(jid, str(reg), 1, dispatch_token=tok, attempt_id=1, watch_pid=0) == 0
    assert _timer_pid(reg, jid) is None
    assert not (reg / f"{jid}.heartbeat.pid").exists()


def test_stop_timer_never_signals_a_recycled_pid(tmp_path, monkeypatch):
    """A stale pidfile (timer crashed) may name a pid that now belongs to
    something else; stop_timer must only signal a heartbeat timer of this job."""
    from lib_py.delegation import heartbeat

    reg = tmp_path / "jobs"
    reg.mkdir()
    bystander = subprocess.Popen(["sleep", "30"])
    try:
        (reg / "job1.heartbeat.pid").write_text(str(bystander.pid))
        assert heartbeat.running_timer("job1", str(reg)) == bystander.pid
        assert heartbeat.stop_timer("job1", str(reg)) is False
        assert bystander.poll() is None, "the bystander must not be killed"
        assert not (reg / "job1.heartbeat.pid").exists(), "the stale pidfile is removed"
    finally:
        bystander.kill()


def test_heartbeat_is_refused_on_a_terminal_event(tmp_path, monkeypatch):
    from lib_py.delegation import publish_event as pe, registry

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    reg = tmp_path / "jobs"
    jid = registry.register_job("hb", registry_dir=str(reg))
    assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "completed", "--heartbeat", "5"]) == 1
    assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "progress", "--heartbeat", "0"]) == 1
    assert _events(reg, jid) == [], "a refused call consumes no seq and records nothing"


# ---------------------------------------------------------------------------
# 3. wire durability: retained state next to every event, seq gaps detected
# ---------------------------------------------------------------------------

def test_every_event_carries_a_signed_retained_state_message(tmp_path, monkeypatch):
    from lib_py.delegation import mqtt_common, publish_event as pe, registry

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    reg = tmp_path / "jobs"
    jid = registry.register_job("state", registry_dir=str(reg), auth_token="tok-" + "a" * 20)
    sent = []

    def fake_publish(config, topic, body, retain, extra=()):
        sent.append((topic, json.loads(body), retain, [(t, json.loads(b), r) for t, b, r in extra]))

    monkeypatch.setattr(pe, "_publish_once", fake_publish)
    tok = _job(reg, jid)["dispatch_token"]
    assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "started", "--expect-dispatch-token", tok]) == 0
    assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "completed", "--detail", "PASS",
                    "--expect-dispatch-token", tok]) == 0
    prefix = _job(reg, jid)["topic_prefix"]
    assert [s[0] for s in sent] == [f"{prefix}/events"] * 2
    for (topic, ev, retain, extra), status in zip(sent, ("running", "completed")):
        assert len(extra) == 1
        st_topic, st, st_retain = extra[0]
        assert st_topic == f"{prefix}/state" and st_retain is True
        assert st["event"] == "state" and st["status"] == status and st["last_event"] == ev["event"]
        assert st["seq"] == ev["seq"] and st["job_id"] == jid
        assert mqtt_common.verify_hmac(st, "tok-" + "a" * 20), "state must be signed with the job token"
        assert st["data"]["dispatch_token"] == ev["data"]["dispatch_token"]
    assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "progress", "--no-state",
                    "--expect-dispatch-token", tok]) == 0
    assert len(sent) == 3 and sent[-1][3] == [], "--no-state publishes the event alone"


class _Msg:
    def __init__(self, topic, payload):
        self.topic = topic
        self.payload = json.dumps(payload).encode()


def _signed(job_id, seq, event, token, **fields):
    from lib_py.delegation import publish_event as pe
    p = pe.build_payload(job_id, seq, event, "", {"dispatch_token": "d1", "attempt_id": 1}, token)
    if fields:
        # re-sign after adding state fields, as build_state_payload does
        import hashlib, hmac
        p.update(fields)
        p["data"].pop("hmac_sig", None)
        sp = {k: v for k, v in p.items() if k != "data"}
        sp["data"] = dict(p["data"])
        p["data"]["hmac_sig"] = hmac.new(token.encode(), json.dumps(sp, sort_keys=True, separators=(",", ":")).encode(),
                                         hashlib.sha256).hexdigest()
    return p


def test_subscriber_records_seq_gaps_and_recovers_terminal_from_state(tmp_path, monkeypatch):
    from lib_py.delegation import job_subscriber as js, mqtt_common

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    token = "t" * 24
    w = js._Watcher({"j1"}, {"j1": token}, {"j1": 0}, {"j1": "d1"}, {"j1": 1})
    w.on_message(None, None, _Msg("x/events", _signed("j1", 1, "started", token)))
    w.on_message(None, None, _Msg("x/events", _signed("j1", 2, "progress", token)))
    w.on_message(None, None, _Msg("x/events", _signed("j1", 5, "progress", token)))  # 3 and 4 were lost
    audit = [e for e in mqtt_common.iter_logged_events("j1", mqtt_common.get_logs_dir()) if e["event"] == "seq_gap"]
    assert len(audit) == 1 and (audit[0]["expected_seq"], audit[0]["seq"], audit[0]["via"]) == (3, 5, "events")
    assert w.events.qsize() == 3

    # a retained state after a reconnect: newer seq than anything seen, terminal status
    st = _signed("j1", 8, "state", token, status="completed", last_event="completed")
    w.on_message(None, None, _Msg("x/state", st))
    assert w.events.qsize() == 4
    topic, payload = list(w.events.queue)[-1]
    assert topic == "x/state" and payload["event"] == "state" and payload["status"] == "completed"
    gaps = [e for e in mqtt_common.iter_logged_events("j1", mqtt_common.get_logs_dir()) if e["event"] == "seq_gap"]
    assert len(gaps) == 2 and gaps[-1]["via"] == "state"
    # a forged state (wrong token) is dropped like a forged event
    bad = _signed("j1", 9, "state", "wrong", status="error", last_event="error")
    w.on_message(None, None, _Msg("x/state", bad))
    assert w.events.qsize() == 4
    # a state for a stale attempt (other dispatch token) is dropped
    stale = _signed("j1", 10, "state", token, status="error", last_event="error")
    stale["data"]["dispatch_token"] = "d0"
    w.on_message(None, None, _Msg("x/state", stale))
    assert w.events.qsize() == 4


def test_subscriber_subscribes_to_the_state_topic(tmp_path):
    src = (SKILLS / "lib_py" / "delegation" / "job_subscriber.py").read_text()
    assert 'state_topics.append(f"{prefix}/state")' in src
    assert "for topic in subscribed_topics + state_topics:" in src


# ---------------------------------------------------------------------------
# 4. delivery: `started` decides, screen signals do not
# ---------------------------------------------------------------------------

def _codex_like_mock(pane_before: str, pane_after: str, code: str = "timeout") -> str:
    return f'''
_pane_quiescent() {{ return 0; }}
_pane_dialog_open() {{ return 1; }}
FLAG="$WORKSPACE_ROOT/.prompt-submitted"
rm -f "$FLAG"
_sks_herdr() {{
  case "${{1:-}}" in
    agent)
      case "${{2:-}}" in
        prompt) touch "$FLAG"; echo '{{"error":{{"code":"{code}","message":"x"}}}}' >&2; return 1 ;;
        get)    echo '{{"result":{{"agent":{{"agent_status":"idle"}}}}}}'; return 0 ;;
      esac ;;
    capture-pane)
      if [ -e "$FLAG" ]; then printf '%s\\n' "{pane_after}"; else printf '%s\\n' "{pane_before}"; fi
      return 0 ;;
    paste-buffer|set-buffer|delete-buffer|send-keys)
      echo "PASTE_PATH_USED: $*" >&2; return 0 ;;
  esac
  return 0
}}
'''


def _run_lib_sh(snippet: str, cwd: Path, env=None, timeout=60):
    return subprocess.run(["bash", "-c", f'source "{LIB_SH}" >/dev/null 2>&1; {snippet}'],
                          capture_output=True, text=True, cwd=str(cwd), env=_env(cwd, env), timeout=timeout)


def test_send_keys_safe_is_confirmed_by_the_job_started_event(tmp_path, monkeypatch):
    """herdr says `timeout`, agent_status stays idle, the pane never changes:
    the only thing that confirms delivery is the job's own `started`."""
    from lib_py.delegation import registry

    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    reg = tmp_path / "jobs"
    jid = registry.register_job("ack", registry_dir=str(reg))
    env = _env(tmp_path, {**UNREACHABLE, "SKS_DELIVERY_GRACE_S": "14", "SKS_REGISTRY_DIR": str(reg)})

    def agent_starts_later():
        time.sleep(2)
        subprocess.run([sys.executable, str(PUBLISH_PY), "--registry-dir", str(reg), "--job", jid,
                        "--event", "started", "--attempts", "1"], capture_output=True, env=env, cwd=str(tmp_path))

    t = threading.Thread(target=agent_starts_later, daemon=True)
    t.start()
    started = time.time()
    res = subprocess.run(["bash", "-c", f'source "{LIB_SH}" >/dev/null 2>&1; '
                          + _codex_like_mock("idle", "idle") + f'send_keys_safe "reviewer-codex-01" "prompt" "{jid}"'],
                         capture_output=True, text=True, cwd=str(tmp_path), env=env, timeout=60)
    t.join()
    assert res.returncode == 0, f"rc={res.returncode}\n{res.stderr}"
    assert time.time() - started < 11, "confirmed once started landed, not at the end of the 14 s grace"
    assert "PASTE_PATH_USED" not in res.stderr


def test_send_keys_safe_ignores_pane_change_and_busy_token(tmp_path):
    """The screen moving on, or showing a busy token, is not delivery any more:
    with no `started` the accepted submission still fails closed (rc 4)."""
    for after in ("Working (0s • esc to interrupt)\\n› Ask Codex", "• OK\\n› Ask Codex"):
        res = _run_lib_sh(_codex_like_mock("idle", after) + 'send_keys_safe "reviewer-codex-01" "prompt" "adhoc"',
                          tmp_path, env={**UNREACHABLE, "SKS_DELIVERY_GRACE_S": "2"})
        assert res.returncode == 4, f"rc={res.returncode}\n{res.stderr}"
        assert "delivery unknown/failed" in res.stderr
        assert "PASTE_PATH_USED" not in res.stderr


def test_wait_started_only_accepts_this_attempts_started(tmp_path):
    """Review R1: a loop job keeps one events.log across iterations. A
    `started` left by the previous attempt (other dispatch token) must not
    acknowledge the new injection; one with the current token must."""
    reg = tmp_path / "jobs"
    reg.mkdir()
    jid = "abcd1234"
    (reg / f"{jid}.json").write_text(json.dumps({"job_id": jid, "status": "pending", "dispatch_token": "new-token"}))
    log = reg / f"{jid}.events.log"
    log.write_text(json.dumps({"event": "started", "seq": 1, "data": {"dispatch_token": "old-token"}}) + "\n")
    env = _env(tmp_path, {"MAM_DELEGATE_JOB_NO_MAIN": "1"})

    def wait_started(extra: str) -> subprocess.CompletedProcess:
        # the wrapper sets -e: capture the helper's rc instead of dying on it
        script = (f'source "{WRAPPER}"; set +e; REGISTRY_DIR="{reg}"; PY="{sys.executable}"; TIMEOUT=60; START_TIME=$(date +%s); '
                  f'_disp_tok="new-token"; {extra} _wait_started {jid} 3; echo "rc=$?"')
        return subprocess.run(["bash", "-c", script], capture_output=True, text=True, env=env, cwd=str(tmp_path), timeout=60)

    res = wait_started("")
    assert "rc=1" in res.stdout, f"old attempt's started must not count\n{res.stdout}\n{res.stderr}"
    with open(log, "a") as fh:
        fh.write(json.dumps({"event": "started", "seq": 2, "data": {"dispatch_token": "new-token"}}) + "\n")
    res = wait_started("")
    assert "rc=0" in res.stdout and "recorded 'started' in events.log" in res.stderr, res.stderr


def test_remote_runner_ignores_a_resent_dispatch_of_a_finished_attempt(tmp_path, monkeypatch):
    """Review R2: a retained dispatch delivered again (runner restart, broker
    replay) must not re-run an attempt that already reached its terminal."""
    from lib_py.delegation import remote, registry

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    reg = tmp_path / "jobs"
    d = {"job_id": "feed1234", "brief": "b", "dispatch_token": "tok-1", "attempt_id": 1, "auth_token": None,
         "timeout_sec": 30, "heartbeat_sec": 5, "heartbeat_misses": 2, "ack_timeout_sec": 5}
    assert remote.mirror_job(d, str(reg), "s") == "feed1234"
    assert remote.mirror_job(d, str(reg), "s") == "feed1234", "same attempt, still open: idempotent"
    from lib_py.delegation import mqtt_common
    mqtt_common.update_job_status("feed1234", str(reg), _source="registry_cli", status="completed")
    assert remote.mirror_job(d, str(reg), "s") is None, "finished attempt: the re-sent dispatch is ignored"
    assert remote.handle_dispatch(d, str(reg), "s", "echo should-not-run") == 0
    d2 = dict(d, dispatch_token="tok-2", attempt_id=2)
    assert remote.mirror_job(d2, str(reg), "s") == "feed1234", "a new attempt is mirrored afresh"
    assert _job(reg, "feed1234")["dispatch_token"] == "tok-2" and _job(reg, "feed1234")["status"] == "pending"


def test_remote_runner_never_rolls_a_newer_attempt_back(tmp_path, monkeypatch):
    """Review round 2 (codex R1): a validly signed dispatch for attempt 1 that
    arrives after attempt 2 is stale. It must neither replace the record nor
    touch its events.log, and it is never run; the same attempt under another
    token is refused as a conflict."""
    from lib_py.delegation import remote

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    reg = tmp_path / "jobs"
    jid = uuid.uuid4().hex[:8]
    a1 = {"job_id": jid, "brief": "attempt one", "dispatch_token": "tok-" + uuid.uuid4().hex, "attempt_id": 1,
          "auth_token": None, "timeout_sec": 30, "heartbeat_sec": 5, "heartbeat_misses": 2, "ack_timeout_sec": 5}
    b2 = dict(a1, brief="attempt two", dispatch_token="tok-" + uuid.uuid4().hex, attempt_id=2)
    assert remote.mirror_job(a1, str(reg), "s") == jid
    assert remote.mirror_job(b2, str(reg), "s") == jid, "a newer attempt is admitted"
    log = reg / f"{jid}.events.log"
    log.write_text(json.dumps({"event": "started", "seq": 1, "data": {"dispatch_token": b2["dispatch_token"]}}) + "\n")
    record_before, log_before = (reg / f"{jid}.json").read_text(), log.read_text()

    assert remote.mirror_job(a1, str(reg), "s") is None, "the replayed attempt-1 dispatch is stale"
    assert remote.mirror_job(dict(b2, dispatch_token="tok-" + uuid.uuid4().hex), str(reg), "s") is None, \
        "the same attempt under another token is a conflict"
    assert (reg / f"{jid}.json").read_text() == record_before, "no record rollback"
    assert log.read_text() == log_before, "no events.log rollback"
    ran = tmp_path / "ran"
    assert remote.handle_dispatch(a1, str(reg), "s", f"touch {ran}") == 0
    assert not ran.exists(), "a stale dispatch is never executed"
    rec = _job(reg, jid)
    assert (rec["attempt_id"], rec["dispatch_token"]) == (2, b2["dispatch_token"])
    assert (reg / jid / "brief.md").read_text() == "attempt two"


def test_remote_command_timeout_ends_the_whole_process_group(tmp_path, monkeypatch):
    """Review round 2 (codex R2): at timeout_sec the runner stops the whole
    command - not only its shell - before it reports the error, so no
    descendant keeps working past the deadline."""
    from lib_py.delegation import remote

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    published = []
    monkeypatch.setattr(remote, "_publish", lambda jid, reg, event, detail, d, heartbeat=None:
                        published.append((event, detail)) or 0)
    reg = tmp_path / "jobs"
    pid_file, marker = tmp_path / "child.pid", tmp_path / "late.marker"
    child = tmp_path / "child.py"
    child.write_text("import os, time\n"
                     f"open({str(pid_file)!r}, 'w').write(str(os.getpid()))\n"
                     "time.sleep(2)\n"
                     f"open({str(marker)!r}, 'w').write('too late')\n")
    d = {"job_id": uuid.uuid4().hex[:8], "brief": "b", "dispatch_token": "tok-" + uuid.uuid4().hex, "attempt_id": 1,
         "auth_token": None, "timeout_sec": 1, "heartbeat_sec": 5, "heartbeat_misses": 2, "ack_timeout_sec": 5}
    pid = None
    try:
        # `; :` keeps the shell alive as the python child's parent (no exec)
        rc = remote.handle_dispatch(d, str(reg), "s", f'"{sys.executable}" "{child}"; :')
        assert rc == 124
        assert published[-1][0] == "error" and "timeout_sec 1" in published[-1][1], published
        pid = int(pid_file.read_text())
        time.sleep(2.5)
        assert not marker.exists(), "a descendant kept working past the deadline"
        assert not _alive(pid)
    finally:
        if pid and _alive(pid):
            os.kill(pid, signal.SIGKILL)


_PAUSING_TIMER = r'''
import json, os, sys, time
from lib_py.delegation import heartbeat as hb, publish_event as pe

capture, paused, go = sys.argv[1], sys.argv[2], sys.argv[3]


def fake_publish(config, topic, body, retain, extra=()):
    with open(capture, "a") as fh:
        fh.write(json.dumps({"who": "timer", "event": json.loads(body)["event"],
                             "state": [json.loads(b).get("status") for _t, b, _r in extra]}) + "\n")


pe._publish_once = fake_publish
real_prepare = pe.prepare_event


def paused_prepare(*a, **k):
    open(paused, "w").close()
    deadline = time.time() + 20
    while not os.path.exists(go) and time.time() < deadline:
        time.sleep(0.05)
    return real_prepare(*a, **k)


pe.prepare_event = paused_prepare
argv = sys.argv[5:]  # argv[4] names the module so stop_timer recognises this process as the job's timer
job, reg = argv[argv.index("--job") + 1], argv[argv.index("--registry-dir") + 1]
with open(hb.pidfile_path(job, reg), "w") as fh:
    fh.write(str(os.getpid()))
sys.exit(hb.main(argv))
'''


def test_no_heartbeat_follows_the_terminal_event_even_when_one_is_in_flight(tmp_path, monkeypatch):
    """Review round 2 (codex R3): a beat already running when the agent
    publishes `completed` must not publish after it. The timer's first beat is
    held just before it asks for a seq; the terminal publish raises the
    attempt's barrier, and the beat - released only then - is refused under the
    seq lock. Nothing follows `completed` on disk or on the (stubbed) wire, and
    the last retained state stays `completed`."""
    from lib_py.delegation import publish_event as pe, registry

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    monkeypatch.setenv("MAM_HEARTBEAT_DRAIN_S", "8")
    reg = tmp_path / "jobs"
    jid = registry.register_job("barrier", registry_dir=str(reg), auth_token="tok-" + uuid.uuid4().hex)
    tok = _job(reg, jid)["dispatch_token"]
    capture, paused, go = tmp_path / "wire.ndjson", tmp_path / "beat.paused", tmp_path / "beat.go"
    script = tmp_path / "pausing_timer.py"
    script.write_text(_PAUSING_TIMER)
    env = dict(os.environ, PYTHONPATH=str(SKILLS), MAM_ENV_FILE=str(tmp_path / ".mam.env.absent"),
               DELEGATE_JOB_LOGS_DIR=str(tmp_path / "logs"))
    timer = subprocess.Popen([sys.executable, str(script), str(capture), str(paused), str(go),
                              "lib_py.delegation.heartbeat", "--job", jid, "--registry-dir", str(reg),
                              "--interval", "1", "--expect-dispatch-token", tok],
                             cwd=str(tmp_path), env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        deadline = time.time() + 15
        while time.time() < deadline and not paused.exists():
            time.sleep(0.05)
        assert paused.exists(), "the timer's first beat never reached its publish"

        def fake_publish(config, topic, body, retain, extra=()):
            with open(capture, "a") as fh:
                fh.write(json.dumps({"who": "terminal", "event": json.loads(body)["event"],
                                     "state": [json.loads(b).get("status") for _t, b, _r in extra]}) + "\n")

        monkeypatch.setattr(pe, "_publish_once", fake_publish)
        result = {}
        terminal = threading.Thread(target=lambda: result.update(rc=pe.main(
            ["--registry-dir", str(reg), "--job", jid, "--event", "completed", "--detail", "PASS",
             "--expect-dispatch-token", tok, "--attempts", "1"])))
        terminal.start()
        barrier = reg / f"{jid}.heartbeat.barrier"
        deadline = time.time() + 10
        while time.time() < deadline and terminal.is_alive() and not barrier.exists():
            time.sleep(0.02)
        go.touch()  # release the in-flight beat
        terminal.join(timeout=30)
        assert result.get("rc") == 0
        assert timer.wait(timeout=30) is not None
        wire = [json.loads(l) for l in capture.read_text().splitlines() if l.strip()]
        assert wire and wire[-1]["event"] == "completed", f"something was published after completed: {wire}"
        assert wire[-1]["state"] == ["completed"], wire
        local = [e["event"] for e in _events(reg, jid)]
        assert local and local[-1] == "completed", local
        assert _job(reg, jid)["status"] == "completed"
    finally:
        if timer.poll() is None:
            timer.kill()


_WIRE_HELD_TIMER = r'''
import json, os, sys, time
from lib_py.delegation import heartbeat as hb, publish_event as pe

capture, paused, go = sys.argv[1], sys.argv[2], sys.argv[3]


def held_publish(config, topic, body, retain, extra=()):
    # the beat was admitted and recorded on disk already: hold it at the wire
    open(paused, "w").close()
    deadline = time.time() + 20
    while not os.path.exists(go) and time.time() < deadline:
        time.sleep(0.05)
    with open(capture, "a") as fh:
        fh.write(json.dumps({"who": "old-timer", "event": json.loads(body)["event"],
                             "state": [json.loads(b).get("status") for _t, b, _r in extra]}) + "\n")


pe._publish_once = held_publish
argv = sys.argv[5:]  # argv[4] names the module so stop_timer recognises this process as the job's timer
job, reg = argv[argv.index("--job") + 1], argv[argv.index("--registry-dir") + 1]
with open(hb.pidfile_path(job, reg), "w") as fh:
    fh.write(str(os.getpid()))
sys.exit(hb.main(argv))
'''


def test_replacing_the_timer_drains_the_old_one_before_the_terminal(tmp_path, monkeypatch):
    """Review round 3 (codex R3): a timer replaced by a second
    `started --heartbeat` must be gone before it loses the pidfile. Its beat -
    already admitted and on disk, held at the wire boundary - may otherwise be
    published after `completed`, because the terminal publish can only find
    and drain the new timer. After the fix the old timer is dead by the time
    start_timer returns and nothing it held follows the terminal."""
    from lib_py.delegation import heartbeat, publish_event as pe, registry

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    reg = tmp_path / "jobs"
    jid = registry.register_job("replace", registry_dir=str(reg), auth_token="tok-" + uuid.uuid4().hex)
    tok = _job(reg, jid)["dispatch_token"]
    capture, paused, go = tmp_path / "wire.ndjson", tmp_path / "beat.paused", tmp_path / "beat.go"

    def fake_publish(config, topic, body, retain, extra=()):
        with open(capture, "a") as fh:
            fh.write(json.dumps({"who": "parent", "event": json.loads(body)["event"],
                                 "state": [json.loads(b).get("status") for _t, b, _r in extra]}) + "\n")

    monkeypatch.setattr(pe, "_publish_once", fake_publish)
    assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "started",
                    "--expect-dispatch-token", tok, "--attempts", "1"]) == 0
    script = tmp_path / "wire_held_timer.py"
    script.write_text(_WIRE_HELD_TIMER)
    env = dict(os.environ, PYTHONPATH=str(SKILLS))
    old = subprocess.Popen([sys.executable, str(script), str(capture), str(paused), str(go),
                            "lib_py.delegation.heartbeat", "--job", jid, "--registry-dir", str(reg),
                            "--interval", "1", "--expect-dispatch-token", tok],
                           cwd=str(tmp_path), env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    new_pid = None
    try:
        deadline = time.time() + 15
        while time.time() < deadline and not paused.exists():
            time.sleep(0.05)
        assert paused.exists(), "the old timer's beat never reached the wire boundary"

        new_pid = heartbeat.start_timer(jid, str(reg), 60, dispatch_token=tok, python=sys.executable, watch_pid=0)
        assert old.poll() is not None, "the replaced timer must be gone before its pidfile is handed over"
        assert _timer_pid(reg, jid) == new_pid

        assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "completed", "--detail", "PASS",
                        "--expect-dispatch-token", tok, "--attempts", "1"]) == 0
        go.touch()  # would release the held beat if its timer were still alive
        time.sleep(0.5)
        wire = [json.loads(l) for l in capture.read_text().splitlines() if l.strip()]
        assert all(w["who"] != "old-timer" for w in wire), f"the replaced timer published: {wire}"
        assert wire[-1]["event"] == "completed" and wire[-1]["state"] == ["completed"], wire
        assert [e["event"] for e in _events(reg, jid)][-1] == "completed"
        deadline = time.time() + 5
        while time.time() < deadline and _alive(new_pid):
            time.sleep(0.1)
        assert not _alive(new_pid), "the terminal publish stops the replacement timer"
    finally:
        if old.poll() is None:
            old.kill()
        if new_pid and _alive(new_pid):
            os.kill(new_pid, signal.SIGKILL)


def test_an_idle_timer_stops_gracefully_well_before_the_drain_deadline(tmp_path, monkeypatch, caplog):
    """Review round 5 (codex F1): the stopper holds the lifecycle lock while it
    waits for the timer to exit, so the exiting timer must not wait for that
    lock. An idle timer (nothing in flight) ends on TERM at once - well before
    a generous drain deadline and without the SIGKILL fallback."""
    import logging
    from lib_py.delegation import heartbeat, registry

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    reg = tmp_path / "jobs"
    jid = registry.register_job("idle-stop", registry_dir=str(reg))
    tok = _job(reg, jid)["dispatch_token"]
    pid = heartbeat.start_timer(jid, str(reg), 60, dispatch_token=tok, python=sys.executable, watch_pid=0)
    try:
        assert pid and _alive(pid)
        time.sleep(2.0)  # past interpreter start-up: TERM handler installed, nothing in flight
        caplog.set_level(logging.WARNING, logger="delegate_job.heartbeat")
        t0 = time.monotonic()
        assert heartbeat.stop_timer(jid, str(reg), drain=10) is True
        elapsed = time.monotonic() - t0
        assert elapsed < 3.0, f"an idle timer took {elapsed:.1f}s to stop (drain deadline 10 s)"
        assert "did not drain" not in caplog.text, caplog.text
        assert not _alive(pid)
        assert not (reg / f"{jid}.heartbeat.pid").exists()
    finally:
        if _alive(pid):
            os.kill(pid, signal.SIGKILL)


def test_a_timer_whose_pid_cannot_be_recorded_is_killed_at_once(tmp_path, monkeypatch):
    """Review round 5 (codex F2): if publishing the new pid fails after the
    timer was spawned, that exact process is killed and reaped before the
    lifecycle lock is released - no untracked owner survives, no partial
    pidfile is left, and the lock is free for the next owner."""
    import builtins
    import errno
    import fcntl
    from lib_py.delegation import heartbeat, registry

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    reg = tmp_path / "jobs"
    jid = registry.register_job("pidfile-fault", registry_dir=str(reg))
    pidfile = str(reg / f"{jid}.heartbeat.pid")
    spawned = []
    real_popen, real_open = heartbeat.subprocess.Popen, builtins.open

    def recording_popen(*a, **k):
        proc = real_popen(*a, **k)
        args = a[0] if a else k.get("args", [])
        if any("lib_py.delegation.heartbeat" in str(x) for x in args):
            spawned.append(proc)
        return proc

    def failing_open(file, mode="r", *a, **k):
        if str(file) == pidfile and "w" in mode:
            raise OSError(errno.ENOSPC, "No space left on device", pidfile)
        return real_open(file, mode, *a, **k)

    monkeypatch.setattr(heartbeat.subprocess, "Popen", recording_popen)
    monkeypatch.setattr(builtins, "open", failing_open)
    try:
        with pytest.raises(OSError) as exc:
            heartbeat.start_timer(jid, str(reg), 60, python=sys.executable, watch_pid=0)
    finally:
        monkeypatch.setattr(builtins, "open", real_open)
    try:
        assert exc.value.errno == errno.ENOSPC
        assert len(spawned) == 1, spawned
        proc = spawned[0]
        assert proc.poll() is not None, "the spawned timer must not survive untracked"
        assert proc.returncode == -signal.SIGKILL
        assert not os.path.exists(pidfile)
        with open(reg / f"{jid}.heartbeat.lock", "a+") as fh:  # the lifecycle lock is free again
            fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
            fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
    finally:
        for proc in spawned:
            if proc.poll() is None:
                proc.kill()
                proc.wait(timeout=5)


def test_a_timer_that_cannot_be_spawned_leaves_nothing_behind(tmp_path, monkeypatch):
    """Review round 5 (codex F2), the other half: when the spawn itself fails
    the error propagates, no pidfile is written, and the lifecycle lock is
    released and serves the next owner at once."""
    import fcntl
    from lib_py.delegation import heartbeat, registry

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    reg = tmp_path / "jobs"
    jid = registry.register_job("spawn-fault", registry_dir=str(reg))
    with pytest.raises(OSError):
        heartbeat.start_timer(jid, str(reg), 60, python=str(tmp_path / "no-such-python"), watch_pid=0)
    assert not (reg / f"{jid}.heartbeat.pid").exists()
    with open(reg / f"{jid}.heartbeat.lock", "a+") as fh:
        fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
    pid = heartbeat.start_timer(jid, str(reg), 60, python=sys.executable, watch_pid=0)
    try:
        assert pid and _timer_pid(reg, jid) == pid and _alive(pid)
    finally:
        heartbeat.stop_timer(jid, str(reg), drain=5)
        if _alive(pid):
            os.kill(pid, signal.SIGKILL)


def _signed_state(job_id, seq, token, dispatch_token, status="running", attempt_id=1):
    import hashlib, hmac
    from lib_py.delegation import publish_event as pe
    p = pe.build_payload(job_id, seq, "state", "progress", {"dispatch_token": dispatch_token, "attempt_id": attempt_id},
                         token)
    p.update(status=status, last_event="progress")
    p["data"].pop("hmac_sig", None)
    sp = {k: v for k, v in p.items() if k != "data"}
    sp["data"] = dict(p["data"])
    p["data"]["hmac_sig"] = hmac.new(token.encode(), json.dumps(sp, sort_keys=True, separators=(",", ":")).encode(),
                                     hashlib.sha256).hexdigest()
    return p


def test_subscriber_state_renews_liveness_only_when_it_advances(tmp_path, monkeypatch):
    """Review round 2 (codex R4): the same retained snapshot re-delivered (as on
    every reconnect) or an older one is still queued - a terminal status must
    be recoverable - but does not renew the idle clock; only a state newer than
    every event and state already accepted does."""
    from lib_py.delegation import job_subscriber as js

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    token = "t" * 24
    w = js._Watcher({"j1"}, {"j1": token}, {"j1": 0}, {"j1": "d1"}, {"j1": 1})
    w.on_message(None, None, _Msg("x/events", _signed("j1", 1, "started", token)))
    w.on_message(None, None, _Msg("x/events", _signed("j1", 10, "progress", token)))
    for _ in range(3):  # duplicate delivery of an old snapshot
        w.on_message(None, None, _Msg("x/state", _signed("j1", 1, "state", token, status="running", last_event="started")))
    w.on_message(None, None, _Msg("x/state", _signed("j1", 10, "state", token, status="running", last_event="progress")))
    w.on_message(None, None, _Msg("x/state", _signed("j1", 11, "state", token, status="running", last_event="progress")))
    states = [p for _t, p in list(w.events.queue) if p["event"] == "state"]
    assert [p["_renews_liveness"] for p in states] == [False, False, False, False, True]


def test_a_replayed_retained_state_does_not_keep_a_silent_job_alive(tmp_path, monkeypatch):
    """Review round 2 (codex R4), end to end: the agent went silent after seq 10
    and the broker keeps re-delivering an old retained snapshot, as it does on
    every reconnect. The subscriber must still reach its idle deadline rather
    than run into the wall clock."""
    from lib_py.delegation import job_subscriber as js, registry

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
    reg = tmp_path / "jobs"
    token = "tok-" + uuid.uuid4().hex
    jid = registry.register_job("replay", registry_dir=str(reg), auth_token=token)
    rec = _job(reg, jid)
    rec["last_seq"] = 10
    (reg / f"{jid}.json").write_text(json.dumps(rec))
    state = _signed_state(jid, 1, token, rec["dispatch_token"])
    topic = rec["topic_prefix"] + "/state"

    class FakeClient:
        def __init__(self):
            self.on_message = self.on_connect = self.on_subscribe = self.on_disconnect = None
            self.stopped = threading.Event()

        def reconnect_delay_set(self, **_kw):
            pass

        def connect(self, *_a, **_kw):
            return 0

        def subscribe(self, *_a, **_kw):
            return (0, 1)

        def disconnect(self):
            pass

        def loop_stop(self):
            self.stopped.set()

        def loop_start(self):
            def run():
                self.on_connect(self, None, None, 0, None)
                self.on_subscribe(self, None, 1, [1], None)
                while not self.stopped.is_set():
                    self.on_message(self, None, _Msg(topic, state))
                    time.sleep(0.3)
            threading.Thread(target=run, daemon=True).start()

    fake = FakeClient()
    monkeypatch.setattr(js, "make_client", lambda *_a, **_kw: fake)
    started = time.monotonic()
    rc = js.main(["--job", jid, "--registry-dir", str(reg), "--idle-timeout", "2", "--ack-timeout", "0",
                  "--timeout", "20"])
    elapsed = time.monotonic() - started
    fake.stopped.set()
    assert rc == 2
    assert elapsed < 8, f"the replayed state kept the silent job alive for {elapsed:.1f}s"


# ---------------------------------------------------------------------------
# 5. wrapper acceptance criteria (broker)
# ---------------------------------------------------------------------------

def _headless_agent(state_file: Path, name: str, agent: str = "codex") -> None:
    state = json.loads(state_file.read_text())
    # agent_status pinned idle, prompt `timeout`, empty pane: nothing to scrape
    state["agents"][name] = {"agent": agent, "agent_status": "idle", "prompt_wait_result": "timeout", "buffer": ""}
    state_file.write_text(json.dumps(state, indent=2))


def _submit(tmp_path: Path, reg: Path, session: str, agent: str, extra_args, env_extra=None, timeout_s=60):
    env = _env(tmp_path, {"SKS_IDLE_TIMEOUT_MS": "500", "SKS_EMPTY_GIVEUP": "1", **(env_extra or {})})
    proc = subprocess.Popen(
        ["bash", str(WRAPPER), "submit", "--agent", agent, "--agent-session", session, "--prompt", "issue 18",
         "--registry-dir", str(reg), "--workdir", str(tmp_path), "--timeout", str(timeout_s), *extra_args],
        cwd=str(tmp_path), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
    return proc, env


def _prompt_calls(state_file: Path) -> int:
    """`agent prompt` invocations; the isolated wrapper prefixes `--session <name>`."""
    n = 0
    for c in json.loads(state_file.read_text()).get("calls", []):
        c = list(c)
        if len(c) >= 2 and c[0] == "--session":
            c = c[2:]
        if len(c) >= 2 and c[0] == "agent" and c[1] == "prompt":
            n += 1
    return n


def test_acceptance1_headless_agent_survives_on_heartbeats_alone(tmp_path, mock_herdr):
    """herdr idle throughout, empty pane: `started --heartbeat 2` keeps the job
    alive past several K x N windows (2 x 2 = 4 s) and `completed` yields rc 0."""
    reg = tmp_path / "jobs"
    reg.mkdir()
    _headless_agent(Path(mock_herdr), "headless-codex")
    proc, env = _submit(tmp_path, reg, "herdr:headless-codex", "codex",
                        ["--heartbeat", "2", "--heartbeat-misses", "2", "--ack-timeout", "40"], timeout_s=90)
    jid = _wait_job_id(reg)
    try:
        time.sleep(1)
        assert _publish(reg, jid, "started", env, "--heartbeat", "2").returncode == 0
        time.sleep(16)  # four idle windows without any agent-published event
        assert _job(reg, jid)["status"] == "running", "heartbeats alone must keep the job open"
        assert _publish(reg, jid, "completed", env, "--detail", "PASS").returncode == 0
        stdout, stderr = proc.communicate(timeout=120)
    finally:
        _kill_timer(reg, jid)
        if proc.poll() is None:
            proc.kill()
    assert proc.returncode == 0, f"rc={proc.returncode}\n{stdout}\n{stderr}"
    assert _job(reg, jid)["status"] == "completed"
    assert _prompt_calls(Path(mock_herdr)) == 1
    beats = [e for e in _events(reg, jid) if e["event"] == "progress"]
    assert len(beats) >= 5, len(beats)
    brief = (reg / jid / "brief.md").read_text()
    assert "heartbeat every 2 s" in brief and "after 2 missed heartbeats" in brief and "within 40 s" in brief


def test_acceptance2_no_started_within_T_fails_closed_without_reinjection(tmp_path, mock_herdr):
    reg = tmp_path / "jobs"
    reg.mkdir()
    _headless_agent(Path(mock_herdr), "mute-codex")
    started = time.time()
    proc, env = _submit(tmp_path, reg, "herdr:mute-codex", "codex",
                        ["--ack-timeout", "4", "--heartbeat", "2", "--heartbeat-misses", "2"],
                        env_extra={"SKS_DELIVERY_GRACE_S": "1"}, timeout_s=60)
    stdout, stderr = proc.communicate(timeout=120)
    jid = _wait_job_id(reg)
    assert proc.returncode == 1, f"rc={proc.returncode}\n{stdout}\n{stderr}"
    assert "no 'started' within 4s" in stderr
    assert _job(reg, jid)["status"] == "error"
    assert _prompt_calls(Path(mock_herdr)) == 1, "single injection: never re-injected"
    assert time.time() - started < 45, "the ack wait is bounded by T, not the wall clock"
    ev = [e for e in _events(reg, jid) if e["event"] == "error"]
    assert ev and ev[-1]["detail"] == "agent_blocked"


def test_acceptance3_lost_heartbeats_close_the_job_and_a_late_completed_is_refused(tmp_path, mock_herdr):
    reg = tmp_path / "jobs"
    reg.mkdir()
    _headless_agent(Path(mock_herdr), "dying-codex")
    proc, env = _submit(tmp_path, reg, "herdr:dying-codex", "codex",
                        ["--heartbeat", "1", "--heartbeat-misses", "3", "--ack-timeout", "40"], timeout_s=60)
    jid = _wait_job_id(reg)
    t0 = None
    agent = None
    try:
        time.sleep(1)
        tok = _job(reg, jid)["dispatch_token"]
        # the agent: a shell that runs the injected start command (timer bound
        # to it via --watch-pid $$) and then keeps "working" silently
        agent = subprocess.Popen(
            ["bash", "-c", f'"{sys.executable}" "{PUBLISH_PY}" --registry-dir "{reg}" --job {jid} --event started '
                           f'--expect-dispatch-token {tok} --heartbeat 1 --watch-pid $$; sleep 300'],
            cwd=str(tmp_path), env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        deadline = time.time() + 30
        while time.time() < deadline and not _timer_pid(reg, jid):
            time.sleep(0.2)
        assert _timer_pid(reg, jid), "the start command must have armed the timer"
        time.sleep(4)
        agent.kill()  # the agent dies: its timer stops, no more heartbeats
        agent.wait(timeout=10)
        t0 = time.time()
        stdout, stderr = proc.communicate(timeout=120)
    finally:
        if agent is not None and agent.poll() is None:
            agent.kill()
        _kill_timer(reg, jid)
        if proc.poll() is None:
            proc.kill()
    assert proc.returncode == 2, f"rc={proc.returncode}\n{stdout}\n{stderr}"
    assert _job(reg, jid)["status"] == "error"
    assert "herdr/pane says busy=" in stderr, "screen state is logged as a diagnostic only"
    # K x N = 3 s idle + one grace window: closed long before the 60 s wall clock
    assert time.time() - t0 < 30, f"took {time.time() - t0:.0f}s"
    late = _publish(reg, jid, "completed", env, "--detail", "too late")
    assert late.returncode in (0, 2, 4)
    assert _job(reg, jid)["status"] == "error", "terminal immutability (CAS) refuses the late completed"


# ---------------------------------------------------------------------------
# 6. remote transport (broker): no shared disk, no pane, only the wire
# ---------------------------------------------------------------------------

def _skip_if_broker_did_not_deliver(reg: Path, runner_log: Path) -> None:
    """acceptance5 runs over a public broker. The remote runner already covers
    the idle window with `started --heartbeat` (remote.py handle_dispatch). The
    observed failures were of one kind: the runner published started and
    completed successfully, the orchestrator had subscribed to the right topic,
    and it received nothing at all for the job - the broker did not deliver.
    Only that proven non-delivery skips; a drop (HMAC, identity), a wrong
    topic, or a runner that failed to publish still fails."""
    log = runner_log.read_text(errors="replace")
    m = re.search(r"published started seq=1 job=(\w+)", log)
    if not m:
        return
    jid = m.group(1)
    rec = reg / f"{jid}.json"
    if not rec.exists():
        return
    topic = json.loads(rec.read_text()).get("topic_prefix") or ""
    out = reg / f"{jid}.subscriber.out"
    sub = out.read_text(errors="replace") if out.exists() else ""
    if (topic and f"SUBSCRIBED {topic}/events" in sub
            and f"job={jid}" not in sub and f"drop event for job {jid}" not in sub):
        pytest.skip(f"public broker did not deliver the runner's events to the orchestrator "
                    f"(runner published started for {jid}; subscriber received nothing for it)")


def test_acceptance5_remote_runner_completes_a_job_over_the_broker_only(tmp_path):
    orch = tmp_path / "orchestrator"
    remote = tmp_path / "remote-host"
    for d in (orch, remote):
        (d / "jobs").mkdir(parents=True)
    # A fresh session name and secret per run: dispatches are retained on the
    # (public) broker and the runner subscribes to every job's dispatch topic,
    # so with fixed values a stale dispatch left by an earlier run was served
    # instead of this run's job, which then timed out (observed: runner DONE
    # for one job id, the orchestrator waiting on another).
    tag = os.urandom(4).hex()
    session = f"sim-worker-{tag}"
    secret = f"issue18-shared-secret-{tag}"
    runner_env = _env(remote, {"MAM_DISPATCH_SECRET": secret, "MAM_JOBS_DIR": str(remote / "jobs")})
    runner_log = remote / "runner.log"
    with open(runner_log, "wb") as log:
        runner = subprocess.Popen(
            [sys.executable, str(REMOTE_PY), "run", "--session", session, "--registry-dir", str(remote / "jobs"),
             "--exec", 'echo "remote work done for $JOB_ID"', "--once", "--timeout", "150"],
            cwd=str(remote), env=runner_env, stdout=log, stderr=subprocess.STDOUT)
    try:
        deadline = time.time() + 45
        while time.time() < deadline and "SUBSCRIBED" not in runner_log.read_text(errors="replace"):
            time.sleep(0.2)
        if "SUBSCRIBED" not in runner_log.read_text(errors="replace"):
            pytest.skip("broker unreachable: " + runner_log.read_text(errors="replace")[-300:])

        reg = orch / "jobs"
        proc, env = _submit(orch, reg, f"remote:{session}", "remote",
                            ["--heartbeat", "2", "--heartbeat-misses", "10", "--ack-timeout", "60"],
                            env_extra={"MAM_DISPATCH_SECRET": secret}, timeout_s=120)
        stdout, stderr = proc.communicate(timeout=180)
        jid = _wait_job_id(reg)
        if proc.returncode != 0:
            if ("no CONNACK from broker" in stderr
                    and "published started" not in runner_log.read_text(errors="replace")):
                pytest.skip("broker unreachable when publishing the dispatch (no CONNACK); the runner never saw the job")
            _skip_if_broker_did_not_deliver(reg, runner_log)
        assert proc.returncode == 0, f"rc={proc.returncode}\n{stdout}\n{stderr}\nrunner:\n{runner_log.read_text()}"
        # delivery is confirmed by the protocol: the subscriber's attributed
        # audit record of the remote `started` (review R1: the registry status
        # is no longer evidence - it can leave pending for reasons that say
        # nothing about this dispatch)
        assert "recorded 'started'" in stderr, stderr
        assert _job(reg, jid)["status"] == "completed", \
            f"the subscriber mirrors the remote terminal into the registry\n{stdout}\n{stderr}\nrunner:\n{runner_log.read_text()}"
        assert _job(reg, jid)["transport"] == "remote"
        # The runner's own status sync follows its wire publish, so the
        # orchestrator can observe `completed` a moment before the remote
        # record says so: wait for the runner before reading its registry.
        runner.wait(timeout=60)
        # the remote side kept its own record with the orchestrator's identity
        mirror = _job(remote / "jobs", jid)
        assert mirror["status"] == "completed" and mirror["transport"] == "remote"
        assert mirror["dispatch_token"] == _job(reg, jid)["dispatch_token"]
        assert mirror["auth_token"] == _job(reg, jid)["auth_token"]
        assert (remote / "jobs" / jid / "brief.md").read_text().startswith("issue 18")
        # the agent command's last stdout line became the completed detail
        assert "remote work done for " + jid in stdout, stdout
        assert runner.returncode == 0
        assert "살아있는 에이전트 세션" not in stdout, "no pane injection path for a remote session"
    finally:
        if runner.poll() is None:
            runner.kill()


def test_remote_runner_refuses_unsigned_dispatches(tmp_path, monkeypatch):
    from lib_py.delegation import remote

    d = {"kind": "dispatch", "job_id": "j", "session": "s", "schema_version": 1}
    assert not remote.verify_dispatch(d, "secret")
    signed = remote.sign_dispatch(d, "secret")
    assert remote.verify_dispatch(signed, "secret")
    assert not remote.verify_dispatch(signed, "other")
    assert not remote.verify_dispatch(dict(signed, session="hijacked"), "secret")
    assert not remote.verify_dispatch(signed, None)
    monkeypatch.delenv("MAM_DISPATCH_SECRET", raising=False)
    assert remote.run_runner("s", str(tmp_path), allow_unsigned=False) == 1, "a runner without a secret refuses to run"


# ---------------------------------------------------------------------------
# 7. loop runner: contract knobs pass through; remote rows get remote:<name>
# ---------------------------------------------------------------------------

def test_loop_runner_passes_the_contract_and_remote_label(tmp_path, monkeypatch):
    from lib_py.loop import runner as loop_runner

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
    monkeypatch.setenv("MAM_STATE_JSON", json.dumps({"herdr_sessions": [
        {"name": "creator-claude-01", "agent": "claude", "role": "creator", "status": "running"},
        {"name": "reviewer-far-01", "agent": "claude", "role": "reviewer", "status": "running", "transport": "remote"},
    ]}))
    monkeypatch.setenv("MAM_HEARTBEAT_SEC", "20")
    monkeypatch.setenv("MAM_HEARTBEAT_MISSES", "3")
    monkeypatch.setenv("MAM_ACK_TIMEOUT_S", "45")
    r = loop_runner.LoopRunner(["--creator", "creator-claude-01", "--task", "t"])
    r.parse_args()
    assert r.session_label("creator-claude-01") == "herdr:creator-claude-01"
    assert r.session_label("reviewer-far-01") == "remote:reviewer-far-01"
    # Issue #33: the job budget is always passed explicitly (default 3600)
    assert r.contract_args() == ["--timeout", "3600", "--heartbeat", "20", "--heartbeat-misses", "3",
                                 "--ack-timeout", "45"]
    seen = []

    def fake_run(cmd, **_kw):
        seen.append(cmd)
        class P:  # noqa: D401 - tiny stand-in
            returncode = 0
            stdout = "registered job: deadbeef\n"
            stderr = ""
        return P()

    monkeypatch.setattr(loop_runner.subprocess, "run", fake_run)
    assert r.delegate_job("reviewer-far-01", "claude", "Reviewer", "review") == "deadbeef"
    cmd = seen[0]
    assert cmd[cmd.index("--agent-session") + 1] == "remote:reviewer-far-01"
    assert cmd[-6:] == ["--heartbeat", "20", "--heartbeat-misses", "3", "--ack-timeout", "45"]