GodopuandClaude Opus 5
ebe58f6be0
Round-2 review of952657e(reviewer-grok-01): R1's disk-heartbeat evidence comes from events.log, but publish_event appended to it only after with_retry(_publish_once) returned - after the broker ACK or all of its retries. With the public broker at ~5 s per publish, an agent's heartbeat reached the disk after the grace window and the live job was closed as deadline_timeout (test_wrapper_idle_timeout_grace_accepts_disk_heartbeat failed 2/2 on952657ehere, as it did for the reviewer). - publish_event: new record_local_event appends the prepared payload to events.log right after prepare_event (validation and seq allocation first) and before the wire publish; record_outcome takes local_event_recorded=True so the line is not written twice, while the audit "published" entry and the status sync stay after the publish - DelegationService.publish_event: the same local-first order - wrapper comment and VERSIONS.md v4.4.2 item 6 now say the heartbeat is on disk before the publish; new-test count 21 - tests: the heartbeat is already in events.log when the publish starts (and is recorded once), and a publish failing on every retry leaves one local record with published=False in the audit; both fail on952657e. The wrapper disk-heartbeat test keeps the real publish_event path and now passes 3/3 at the same broker latency Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HCGpJXhSZasJLU38ksDghe
512 lines
22 KiB
Python
512 lines
22 KiB
Python
#!/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 through the pane
|
||
# ---------------------------------------------------------------------------
|
||
def test_accepted_timeout_confirmed_by_busy_token(tmp_path):
|
||
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={"SKS_DELIVERY_GRACE_S": "5"})
|
||
assert res.returncode == 0, f"rc={res.returncode}\n{res.stderr}"
|
||
_assert_single_injection(res)
|
||
assert "delivery unknown" not in res.stderr
|
||
|
||
|
||
def test_accepted_timeout_confirmed_by_pane_change_after_fast_turn(tmp_path):
|
||
"""A turn that already finished before the first poll shows no busy token,
|
||
only the answer; the pane moving on from its pre-submit snapshot is enough."""
|
||
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": "5"})
|
||
assert res.returncode == 0, 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. 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))
|
||
|
||
res = _run_lib_sh(
|
||
'send_keys_safe "agt_codex" "review please" "job_14f"',
|
||
tmp_path, env={"SKS_IDLE_HANDSHAKE": "0", "SKS_DELIVERY_GRACE_S": "5"},
|
||
)
|
||
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.
|
||
ack = subprocess.run(
|
||
[sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg_dir),
|
||
"update", "--job", jid, "--status", "running"],
|
||
capture_output=True, text=True, cwd=str(tmp_path), env=env,
|
||
)
|
||
assert ack.returncode == 0, ack.stderr
|
||
|
||
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) must see a working codex through the pane
|
||
# ---------------------------------------------------------------------------
|
||
def test_wrapper_idle_timeout_grace_accepts_pane_busy_token(tmp_path, mock_herdr):
|
||
"""herdr keeps reporting `idle` for a codex-like agent that is visibly
|
||
working (busy token in the pane). When the subscriber idles out, the
|
||
wrapper must keep waiting for the on-disk result instead of closing the
|
||
job as deadline_timeout mid-turn; the agent then completes out of band."""
|
||
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": "5"})
|
||
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"
|
||
|
||
# Well past the 3 s idle timeout, the agent finishes.
|
||
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,
|
||
)
|
||
assert done.returncode == 0, done.stderr
|
||
|
||
stdout, stderr = proc.communicate(timeout=90)
|
||
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 after the working grace, got {proc.returncode}\n{stdout}\n{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", "60",
|
||
"--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"
|
||
|
||
def heartbeat(detail: str):
|
||
subprocess.run(
|
||
[sys.executable, str(PUBLISH_PY), "--registry-dir", str(reg_dir),
|
||
"--job", jid, "--event", "progress", "--detail", detail],
|
||
capture_output=True, text=True, cwd=str(tmp_path), env=env,
|
||
)
|
||
|
||
# Subscriber idles out at ~6 s (plus broker connect); heartbeats then keep
|
||
# arriving 2 s apart, 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).
|
||
time.sleep(9)
|
||
for i in range(5):
|
||
heartbeat(f"still working {i}")
|
||
time.sleep(2)
|
||
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=90)
|
||
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 not a moment past it beyond closing and publishing the error.
|
||
assert elapsed <= timeout_s + 12, f"idle grace outlived --timeout {timeout_s}s: {elapsed:.1f}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. The audit log resolves against
|
||
the cwd, so it lands under tmp_path too, and no real .mam.env is read."""
|
||
from lib_py.delegation import registry
|
||
|
||
monkeypatch.chdir(tmp_path)
|
||
monkeypatch.delenv("DELEGATE_JOB_LOGS_DIR", raising=False)
|
||
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))
|
||
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):
|
||
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):
|
||
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
|