fix(issue-14): record every event on disk before the wire publish
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
This commit is contained in:
1 parent
952657efb4
commit
ebe58f6be0
5 files changed
+109
-13
No files matched your search
@@ -440,3 +440,72 @@ def test_wrapper_idle_grace_never_outlives_the_wall_clock(tmp_path):
|
||||
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
|
||||
Reference in new issue
Block a user