multi-agent-mux/tests/test_issue14_delivery_ack.py

#!/usr/bin/env python3
"""Issue #14 — a prompt that herdr accepted must not be reported as undelivered.

`herdr agent prompt --wait --until working --timeout 5000` answers `timeout` (or
`agent_prompt_stalled`) when the input was sent but no `working` state was
observed afterwards. Agents whose agent_status never leaves `idle` mid-turn
(codex under herdr 0.9.0) therefore had every delivered prompt closed as
`agent_blocked`, and their own `started` / `completed` events, published seconds
later, were refused as late. send_keys_safe now confirms an accepted submission
through the pane, and the delegate wrapper additionally honours the registry's
own acknowledgement before finalising a job as an inject failure.
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import time
from pathlib import Path

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"
REGISTRY_PY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "registry.py"

TIMEOUT_JSON = '{"error":{"code":"timeout","message":"timed out waiting for agent status"},"id":"cli:agent:prompt"}'
IDLE_JSON = '{"result":{"agent":{"agent_status":"idle"}}}'


def _base_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)
    if extra:
        env.update({k: str(v) for k, v in extra.items()})
    return env


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


def _codex_like_mock(pre_pane: str, post_pane: str, prompt_code: str = "timeout") -> str:
    """A bash `_sks_herdr` stand-in replaying what herdr 0.9.0 shows for codex:
    the RPC returns `timeout`, agent_status stays `idle`, and only the pane
    reveals what happened. Captures switch from PRE to POST once the prompt
    has been submitted. Any paste-buffer use is a contract violation."""
    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":"{prompt_code}","message":"x"}}}}' >&2; return 1 ;;
        get)    echo '{IDLE_JSON}'; return 0 ;;
      esac ;;
    capture-pane)
      if [ -e "$FLAG" ]; then printf '%s\\n' "{post_pane}"; else printf '%s\\n' "{pre_pane}"; fi
      return 0 ;;
    paste-buffer|set-buffer|delete-buffer|send-keys)
      echo "PASTE_PATH_USED: $*" >&2; return 0 ;;
  esac
  return 0
}}
'''


CODEX_IDLE = "identified.\\n› Ask Codex to do anything"
CODEX_WORKING = "Working (0s • esc to interrupt)\\n› Ask Codex to do anything"
CODEX_ANSWERED = "• OK\\n› Ask Codex to do anything"


def _assert_single_injection(res: subprocess.CompletedProcess):
    assert "PASTE_PATH_USED" not in res.stderr, f"must not fall back to paste-buffer: {res.stderr}"


# ---------------------------------------------------------------------------
# lib.sh: accepted submission confirmed by the job's own `started` (Issue #18)
# ---------------------------------------------------------------------------
def _job_that_starts(tmp_path, delay_s: float):
    """A registered job whose `started` lands `delay_s` after the call - the
    protocol signal that replaced the pane (Issue #18). Returns (env, job_id)."""
    import threading
    from lib_py.delegation import registry

    reg = tmp_path / "jobs"
    env = {"MAM_ENV_FILE": str(tmp_path / ".mam.env.absent"), "DELEGATE_JOB_LOGS_DIR": str(tmp_path / "logs"),
           "MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "SKS_REGISTRY_DIR": str(reg)}
    saved = {k: os.environ.get(k) for k in env}
    os.environ.update(env)
    try:
        jid = registry.register_job("issue 14 ack", registry_dir=str(reg))
    finally:
        for k, v in saved.items():
            if v is None:
                os.environ.pop(k, None)
            else:
                os.environ[k] = v

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

    threading.Thread(target=start, daemon=True).start()
    return env, jid


def test_accepted_timeout_confirmed_by_started_event_busy_token_alone_is_not_enough(tmp_path):
    """The pane shows the busy token, but that is not delivery any more: the
    job's `started` (published 2 s in) is what confirms it, and the busy token
    on its own (second run) still fails closed."""
    env, jid = _job_that_starts(tmp_path, 2.0)
    script = _codex_like_mock(CODEX_IDLE, CODEX_WORKING) + \
        f'send_keys_safe "reviewer-codex-01" "review the triage report" "{jid}"'
    res = _run_lib_sh(script, tmp_path, env={**env, "SKS_DELIVERY_GRACE_S": "8"})
    assert res.returncode == 0, f"rc={res.returncode}\n{res.stderr}"
    _assert_single_injection(res)
    assert "delivery unknown" not in res.stderr

    script = _codex_like_mock(CODEX_IDLE, CODEX_WORKING) + \
        'send_keys_safe "reviewer-codex-01" "review the triage report" "job_14a"'
    res = _run_lib_sh(script, tmp_path, env={**env, "SKS_DELIVERY_GRACE_S": "2"})
    assert res.returncode == 4, f"busy token must not confirm delivery: rc={res.returncode}\n{res.stderr}"
    _assert_single_injection(res)


def test_accepted_timeout_pane_change_after_fast_turn_is_not_delivery(tmp_path):
    """A turn that already finished shows only the answer. The pane moving on
    used to count; under the protocol contract only `started` does."""
    script = _codex_like_mock(CODEX_IDLE, CODEX_ANSWERED) + \
        'send_keys_safe "reviewer-codex-01" "Reply with OK" "job_14b"'
    res = _run_lib_sh(script, tmp_path, env={"SKS_DELIVERY_GRACE_S": "2"})
    assert res.returncode == 4, f"rc={res.returncode}\n{res.stderr}"
    _assert_single_injection(res)


def test_accepted_timeout_with_no_signal_still_fails_closed(tmp_path):
    """Nothing moves, nothing works: the grace window expires into rc=4 exactly
    as before, without ever re-submitting."""
    script = _codex_like_mock(CODEX_IDLE, CODEX_IDLE) + \
        'send_keys_safe "reviewer-codex-01" "silent prompt" "job_14c"'
    started = time.time()
    res = _run_lib_sh(script, tmp_path, env={"SKS_DELIVERY_GRACE_S": "2"})
    elapsed = time.time() - started
    assert res.returncode == 4, f"rc={res.returncode}\n{res.stderr}"
    assert "delivery unknown/failed" in res.stderr
    _assert_single_injection(res)
    assert 2 <= elapsed < 20, f"grace window not honoured: {elapsed:.1f}s"


def test_pre_existing_busy_token_is_not_taken_as_delivery(tmp_path):
    """A busy token that was already on screen before the prompt is a leftover
    of the previous turn, not evidence for this one."""
    script = _codex_like_mock(CODEX_WORKING, CODEX_WORKING) + \
        'send_keys_safe "reviewer-codex-01" "prompt into a busy pane" "job_14d"'
    res = _run_lib_sh(script, tmp_path, env={"SKS_DELIVERY_GRACE_S": "2"})
    assert res.returncode == 4, f"rc={res.returncode}\n{res.stderr}"
    _assert_single_injection(res)


def test_transport_error_is_not_treated_as_accepted(tmp_path):
    """Only `timeout` / `agent_prompt_stalled` mean the input was sent. A
    transport-level failure keeps the single 2 s probe and rc=4, even if the
    pane happens to change meanwhile (T2-13 contract, pinned with pane noise)."""
    script = _codex_like_mock(CODEX_IDLE, CODEX_WORKING, prompt_code="transport_error") + \
        'send_keys_safe "reviewer-codex-01" "prompt" "job_14e"'
    started = time.time()
    res = _run_lib_sh(script, tmp_path, env={"SKS_DELIVERY_GRACE_S": "30"})
    elapsed = time.time() - started
    assert res.returncode == 4, f"rc={res.returncode}\n{res.stderr}"
    _assert_single_injection(res)
    assert elapsed < 15, f"no grace poll may run for a non-accepted code: {elapsed:.1f}s"


def test_mock_herdr_timeout_with_idle_status_is_confirmed(tmp_path, mock_herdr):
    """Same scenario through the shared mock herdr: `timeout` from agent prompt,
    agent_status pinned to idle, the pane gaining the busy token after
    submission - and the job's `started` landing 2 s in, which is what
    confirms it (Issue #18). Exactly one prompt call, no paste-buffer fallback."""
    state_file = Path(mock_herdr)
    state = json.loads(state_file.read_text())
    state["agents"]["agt_codex"] = {
        "agent": "codex",
        "agent_status": "idle",
        "prompt_wait_result": "timeout",
        "buffer": "Ready",
    }
    state_file.write_text(json.dumps(state, indent=2))

    env, jid = _job_that_starts(tmp_path, 2.0)
    res = _run_lib_sh(
        f'send_keys_safe "agt_codex" "review please" "{jid}"',
        tmp_path, env={**env, "SKS_IDLE_HANDSHAKE": "0", "SKS_DELIVERY_GRACE_S": "8"},
    )
    assert res.returncode == 0, f"rc={res.returncode}\n{res.stderr}"

    final_state = json.loads(state_file.read_text())
    calls = final_state.get("calls", [])
    prompt_calls = [c for c in calls if len(c) >= 2 and c[0] == "agent" and c[1] == "prompt"]
    paste_calls = [c for c in calls if len(c) >= 2 and (c[0] in ("paste-buffer", "set-buffer") or c[1] in ("paste-buffer", "set-buffer"))]
    assert len(prompt_calls) == 1, f"prompt must be submitted exactly once, got {len(prompt_calls)}"
    assert not paste_calls, f"no paste-buffer fallback allowed: {paste_calls}"


# ---------------------------------------------------------------------------
# wrapper: the registry's own acknowledgement beats an unconfirmed inject
# ---------------------------------------------------------------------------
def test_wrapper_honours_registry_ack_over_unconfirmed_inject(tmp_path, mock_herdr):
    """The RPC fails outright, but the agent has already published `started`
    (status running) by the time the wrapper decides: the job must not be
    closed as agent_blocked. With the broker unreachable the wrapper then
    follows its normal broker_unavailable path (rc=3), never the inject path."""
    reg_dir = tmp_path / "jobs"
    reg_dir.mkdir(parents=True, exist_ok=True)

    state_file = Path(mock_herdr)
    state = json.loads(state_file.read_text())
    state["agents"]["ack-agent"] = {
        "agent": "codex",
        "agent_status": "idle",
        "prompt_wait_result": "transport_error",
    }
    state["prompt_wait_result"] = "transport_error"
    state_file.write_text(json.dumps(state, indent=2))

    env = _base_env(tmp_path, {
        "MQTT_HOST": "127.0.0.1", "MQTT_PORT": "65534",
        "SKS_IDLE_TIMEOUT_MS": "500", "MAM_INJECT_ACK_GRACE_S": "8",
    })
    proc = subprocess.Popen(
        [
            "bash", str(WRAPPER), "submit",
            "--agent", "codex",
            "--agent-session", "herdr:ack-agent",
            "--prompt", "issue 14 ack",
            "--registry-dir", str(reg_dir),
            "--workdir", str(tmp_path),
            "--timeout", "6",
            "--idle-timeout", "2",
        ],
        cwd=str(tmp_path), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
    )

    jid = ""
    for _ in range(100):
        job_files = list(reg_dir.glob("*.json"))
        if job_files:
            jid = job_files[0].stem
            break
        time.sleep(0.1)
    assert jid, "job was not registered"

    # The agent acknowledges the job out of band, as codex does via
    # publish_event - the real producer, with the dispatch token the wrapper
    # handed it. (Issue #23 review R1: a bare registry status flip is not an
    # acknowledgement any more; only the agent's own `started` naming this
    # dispatch is. publish_event writes that line locally before its wire
    # publish, so an unreachable broker does not hide it.)
    tok = json.loads((reg_dir / f"{jid}.json").read_text()).get("dispatch_token") or ""
    subprocess.run(
        [sys.executable, str(PUBLISH_PY), "--registry-dir", str(reg_dir), "--job", jid,
         "--event", "started", "--expect-dispatch-token", tok, "--attempt-id", "1"],
        capture_output=True, text=True, cwd=str(tmp_path), env=env, timeout=60,
    )
    started_lines = [l for l in (reg_dir / f"{jid}.events.log").read_text().splitlines()
                     if '"started"' in l]
    assert started_lines and tok in started_lines[-1], started_lines

    stdout, stderr = proc.communicate(timeout=90)
    assert "treating the prompt as delivered and continuing" in stderr, f"ack guard did not fire:\n{stderr}\n{stdout}"
    assert "프롬프트 주입 실패" not in stderr
    assert proc.returncode == 3, f"expected the broker_unavailable path (rc=3), got {proc.returncode}\n{stderr}"

    events_log = reg_dir / f"{jid}.events.log"
    if not events_log.exists():
        events_log = reg_dir / jid / "events.log"
    if events_log.exists():
        assert "agent_blocked" not in events_log.read_text(), events_log.read_text()
    job_data = json.loads((reg_dir / f"{jid}.json").read_text())
    assert job_data.get("status") != "pending"


# ---------------------------------------------------------------------------
# wrapper: idle-timeout grace (R1) is decided by the contract, not the pane
# ---------------------------------------------------------------------------
def test_wrapper_idle_timeout_grace_ignores_pane_busy_token(tmp_path, mock_herdr):
    """Issue #18 flips the Issue #14 expectation: a codex-like agent that is
    visibly working (busy token in the pane) but publishes nothing after its
    `started` is unresponsive once K x N (here --idle-timeout 3 s) has passed.
    The wrapper closes it as deadline_timeout and a `completed` written 14 s
    in is refused by the CAS table - the pane never extended the window."""
    reg_dir = tmp_path / "jobs"
    reg_dir.mkdir(parents=True, exist_ok=True)

    state_file = Path(mock_herdr)
    state = json.loads(state_file.read_text())
    # `timeout` from agent prompt keeps agent_status idle; the mock appends the
    # busy token to the pane on submission, exactly like a codex turn.
    state["agents"]["slow-codex"] = {
        "agent": "codex",
        "agent_status": "idle",
        "prompt_wait_result": "timeout",
        "buffer": "Ready",
    }
    state_file.write_text(json.dumps(state, indent=2))

    env = _base_env(tmp_path, {"SKS_IDLE_TIMEOUT_MS": "500", "SKS_DELIVERY_GRACE_S": "2", "MAM_INJECT_ACK_GRACE_S": "20"})
    proc = subprocess.Popen(
        [
            "bash", str(WRAPPER), "submit",
            "--agent", "codex",
            "--agent-session", "herdr:slow-codex",
            "--prompt", "issue 14 idle grace",
            "--registry-dir", str(reg_dir),
            "--workdir", str(tmp_path),
            "--timeout", "40",
            "--idle-timeout", "3",
        ],
        cwd=str(tmp_path), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
    )

    jid = ""
    for _ in range(100):
        job_files = list(reg_dir.glob("*.json"))
        if job_files:
            jid = job_files[0].stem
            break
        time.sleep(0.1)
    assert jid, "job was not registered"

    # The agent acknowledges (started) without arming a heartbeat timer ...
    time.sleep(1)
    tok = json.loads((reg_dir / f"{jid}.json").read_text())["dispatch_token"]
    ack = subprocess.run(
        [sys.executable, str(PUBLISH_PY), "--registry-dir", str(reg_dir), "--job", jid, "--event", "started",
         "--expect-dispatch-token", tok],
        capture_output=True, text=True, cwd=str(tmp_path), env=env,
    )
    assert ack.returncode == 0, ack.stderr
    # ... and goes silent while its pane keeps showing the busy token.
    time.sleep(14)
    done = subprocess.run(
        [sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg_dir),
         "update", "--job", jid, "--status", "completed"],
        capture_output=True, text=True, cwd=str(tmp_path), env=env,
    )

    stdout, stderr = proc.communicate(timeout=90)
    job_data = json.loads((reg_dir / f"{jid}.json").read_text())
    assert proc.returncode == 2, f"expected deadline_timeout (rc=2): the pane is not liveness, got {proc.returncode}\n{stdout}\n{stderr}"
    assert job_data.get("status") == "error", f"status={job_data.get('status')}\n{stdout}\n{stderr}"
    assert done.returncode != 0 or "refused" in done.stderr, "the late completed must be refused"
    assert "diagnostic only" in stderr


# ---------------------------------------------------------------------------
# wrapper: idle-timeout grace (R1) must honour on-disk progress heartbeats
# ---------------------------------------------------------------------------
PUBLISH_PY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py"


def test_wrapper_idle_timeout_grace_accepts_disk_heartbeat(tmp_path):
    """No herdr evidence at all (a human agent, or an agent running its tests
    in a background shell): once the subscriber has idled out, the wrapper
    must keep the job open as long as fresh `progress` heartbeats keep landing
    in events.log, and pick up the eventual completion. Each heartbeat here
    arrives after the previous idle window would have expired."""
    reg_dir = tmp_path / "jobs"
    reg_dir.mkdir(parents=True, exist_ok=True)
    env = _base_env(tmp_path)
    proc = subprocess.Popen(
        [
            "bash", str(WRAPPER), "submit",
            "--agent", "human",
            "--prompt", "issue 14 disk heartbeat",
            "--registry-dir", str(reg_dir),
            "--workdir", str(tmp_path),
            "--timeout", "90",
            "--idle-timeout", "6",
        ],
        cwd=str(tmp_path), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
    )

    jid = ""
    for _ in range(100):
        job_files = list(reg_dir.glob("*.json"))
        if job_files:
            jid = job_files[0].stem
            break
        time.sleep(0.1)
    assert jid, "job was not registered"

    # Heartbeats are fired without waiting for the wire publish: publish_event
    # records the event in events.log before it touches the broker (ebe58f6),
    # and that on-disk timestamp is what the wrapper's grace reads. Blocking on
    # each publish (2-6 s against the public broker, with retries) spaced the
    # disk records further apart than the 6 s idle window and the job was
    # closed before the next heartbeat landed - a broker-latency flake, not a
    # wrapper defect. An agent's own heartbeat timer behaves the same way.
    beats = []

    def heartbeat(detail: str):
        beats.append(subprocess.Popen(
            [sys.executable, str(PUBLISH_PY), "--registry-dir", str(reg_dir),
             "--job", jid, "--event", "progress", "--detail", detail],
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, cwd=str(tmp_path), env=env,
        ))

    # Subscriber idles out at ~6 s plus the broker connect (the public broker
    # has taken 5-10 s); heartbeats then keep landing 2 s apart for 20 s, well
    # inside the 6 s window each one re-opens (the wrapper's poll iteration
    # costs ~2 s of herdr RPCs, so the ratio matters) and long enough to
    # outlast a slow connect.
    time.sleep(9)
    for i in range(10):
        heartbeat(f"still working {i}")
        time.sleep(2)
    for b in beats:
        b.wait(timeout=60)
    done = subprocess.run(
        [sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg_dir),
         "update", "--job", jid, "--status", "completed"],
        capture_output=True, text=True, cwd=str(tmp_path), env=env,
    )
    assert done.returncode == 0, done.stderr

    stdout, stderr = proc.communicate(timeout=150)
    job_data = json.loads((reg_dir / f"{jid}.json").read_text())
    assert job_data.get("status") == "completed", f"status={job_data.get('status')}\n{stdout}\n{stderr}"
    assert proc.returncode == 0, f"expected rc=0 while heartbeats kept landing, got {proc.returncode}\n{stdout}\n{stderr}"


def test_wrapper_idle_grace_never_outlives_the_wall_clock(tmp_path):
    """Heartbeats may extend the idle window, never the overall --timeout.
    Fresh heartbeats land in events.log every second for the whole run, and
    every python probe the wrapper makes is slowed down (herdr-RPC-like
    latency). Counting sleeps instead of reading the clock let the grace loop
    run about twice as long as TIMEOUT; the job must instead close as
    deadline_timeout within TIMEOUT plus a small close/publish margin."""
    import threading

    timeout_s, idle_s = 20, 4
    reg_dir = tmp_path / "jobs"
    reg_dir.mkdir(parents=True, exist_ok=True)
    slow_py = tmp_path / "slow-python"
    slow_py.write_text(f'#!/bin/sh\nsleep 0.7\nexec "{sys.executable}" "$@"\n')
    slow_py.chmod(0o755)
    env = _base_env(tmp_path, {"DELEGATE_JOB_PYTHON": slow_py})

    started = time.time()
    proc = subprocess.Popen(
        [
            "bash", str(WRAPPER), "submit",
            "--agent", "human",
            "--prompt", "issue 14 wall clock",
            "--registry-dir", str(reg_dir),
            "--workdir", str(tmp_path),
            "--timeout", str(timeout_s),
            "--idle-timeout", str(idle_s),
        ],
        cwd=str(tmp_path), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
    )

    jid = ""
    for _ in range(200):
        job_files = list(reg_dir.glob("*.json"))
        if job_files:
            jid = job_files[0].stem
            break
        time.sleep(0.1)
    assert jid, "job was not registered"
    events_log = reg_dir / f"{jid}.events.log"

    stop = threading.Event()

    def keep_heartbeating():
        # The same line shape publish_event appends; only its timestamp matters
        # to _recent_disk_heartbeat.
        seq = 10_000
        while not stop.is_set():
            seq += 1
            line = {"schema_version": 1, "seq": seq, "job_id": jid, "event": "progress",
                    "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
                    "detail": "still working", "data": {}}
            with open(events_log, "a") as fh:
                fh.write(json.dumps(line) + "\n")
            stop.wait(1)

    beat = threading.Thread(target=keep_heartbeating, daemon=True)
    beat.start()
    try:
        stdout, stderr = proc.communicate(timeout=120)
    finally:
        stop.set()
    elapsed = time.time() - started

    job_data = json.loads((reg_dir / f"{jid}.json").read_text())
    assert proc.returncode == 2, f"expected the deadline_timeout path (rc=2), got {proc.returncode}\n{stderr}"
    assert job_data.get("status") == "error", f"status={job_data.get('status')}\n{stderr}"
    # The heartbeats were honoured: the job was not closed after the first idle
    # window, it stayed open up to the wall clock ...
    assert elapsed >= timeout_s - 1, f"closed before the wall clock: {elapsed:.1f}s\n{stderr}"
    # ... and the close itself landed within the wall clock plus a small margin.
    # Measured on the record's own close time: the error publish that follows
    # the close can take 10-20 s against the public broker and is not what
    # this test is about.
    import calendar
    closed_at = calendar.timegm(time.strptime(job_data["updated_at"], "%Y-%m-%dT%H:%M:%SZ"))
    close_delay = closed_at - started
    # Under full-suite load the wrapper's final poll, close and registry write
    # have landed 8-9 s past the deadline (three full runs); the defect this
    # pins was a close at ~2 x TIMEOUT, so a 15 s margin still separates them.
    assert close_delay <= timeout_s + 15, f"idle grace outlived --timeout {timeout_s}s: closed after {close_delay:.0f}s\n{stderr}"


# ---------------------------------------------------------------------------
# publish_event: a heartbeat reaches events.log before the wire publish
# ---------------------------------------------------------------------------
def _job_for_inprocess_publish(tmp_path, monkeypatch):
    """A registered job in a private registry. Its audit log goes to the
    per-test default under tmp_path (conftest, Issue #20): deleting
    DELEGATE_JOB_LOGS_DIR here, to fall back to the cwd, fell through to an
    ambient WORKSPACE_ROOT instead - the real .mam/delegate_job_logs (PR #22
    rec 4). No real .mam.env is read."""
    from lib_py.delegation import mqtt_common, registry

    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
    reg_dir = tmp_path / "jobs"
    jid = registry.register_job("issue 14 heartbeat order", registry_dir=str(reg_dir))
    logs = Path(mqtt_common.get_logs_dir())
    assert tmp_path in logs.parents and (logs / jid).is_dir(), logs
    return reg_dir, jid


def _logged_events(path: Path) -> list:
    if not path.exists():
        return []
    return [json.loads(line) for line in path.read_text().splitlines() if line.strip()]


def test_publish_event_records_the_heartbeat_before_the_wire_publish(tmp_path, monkeypatch):
    """R1 reads events.log as liveness once the subscriber has idled out. The
    publish blocks on the broker ACK and its retries, so a heartbeat written
    only afterwards reached the disk too late whenever the broker was slow
    (~5 s per publish closed a live job in the round-2 review). The line must
    already be on disk when the wire publish starts, and be written once."""
    from lib_py.delegation import publish_event as pe

    reg_dir, jid = _job_for_inprocess_publish(tmp_path, monkeypatch)
    events_log = reg_dir / f"{jid}.events.log"
    on_disk_at_publish = []

    def slow_broker_publish(config, topic, body, retain, extra=()):
        on_disk_at_publish.append([e.get("event") for e in _logged_events(events_log)])

    monkeypatch.setattr(pe, "_publish_once", slow_broker_publish)
    rc = pe.main(["--registry-dir", str(reg_dir), "--job", jid,
                  "--event", "progress", "--detail", "still working"])
    assert rc == 0
    assert on_disk_at_publish == [["progress"]], on_disk_at_publish
    assert [e.get("event") for e in _logged_events(events_log)] == ["progress"], "recorded exactly once"


def test_publish_event_failed_publish_keeps_one_local_record(tmp_path, monkeypatch):
    """A broker that never answers must not delay the local record, retries
    must not append it again, and the audit trail still says published=False."""
    from lib_py.delegation import mqtt_common, publish_event as pe

    reg_dir, jid = _job_for_inprocess_publish(tmp_path, monkeypatch)
    events_log = reg_dir / f"{jid}.events.log"
    lines_seen = []

    def unreachable_broker(config, topic, body, retain, extra=()):
        lines_seen.append(len(_logged_events(events_log)))
        raise ConnectionError("broker unreachable")

    monkeypatch.setattr(pe, "_publish_once", unreachable_broker)
    rc = pe.main(["--registry-dir", str(reg_dir), "--job", jid, "--event", "progress",
                  "--detail", "still working", "--attempts", "2"])
    assert rc == 2
    assert lines_seen == [1, 1], lines_seen
    assert len(_logged_events(events_log)) == 1
    audit = [e for e in mqtt_common.iter_logged_events(jid, mqtt_common.get_logs_dir())
             if e.get("event") == "published"]
    assert audit and audit[-1]["published"] is False, audit