fix(issue-14): record every event on disk before the wire publish

Round-2 review of 952657e (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 on 952657e here, 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 on 952657e.
  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

.agents/skills/lib_py/delegation/publish_event.py

@@ -162,14 +162,31 @@ def prepare_event(job_id, event, detail="", data=None,
     return PreparedEvent(job_id, event, detail, seq, topic, retain, payload, body, config)
 
 
+def record_local_event(prepared: PreparedEvent, registry_dir: str) -> None:
+    """Append the event to the job's on-disk events.log BEFORE any wire publish.
+
+    The delegate wrapper reads this log as liveness evidence once its
+    subscriber has idled out (`_recent_disk_heartbeat`). Recorded only after
+    the publish - which blocks on the broker ACK and its retries - a slow
+    broker's heartbeat reached the disk too late for that grace window, and
+    live jobs were closed as deadline_timeout. Best effort, like the rest of
+    the record.
+    """
+    registry.append_event(prepared.job_id, registry_dir, prepared.payload)
+
+
 def record_outcome(prepared: PreparedEvent, registry_dir: str,
                    publish_ok: bool, publish_error: Optional[str],
                    expected_dispatch_token: Optional[str] = None,
-                   expected_attempt: Optional[int] = None) -> None:
+                   expected_attempt: Optional[int] = None,
+                   local_event_recorded: bool = False) -> None:
     """Persist the audit record, the registry event log and the status sync.
 
     All three are best effort and all three happen whether or not the publish
-    succeeded — that is the contract G-1/G-2/G-4 pin down.
+    succeeded — that is the contract G-1/G-2/G-4 pin down. A caller that
+    already wrote the registry event log before publishing
+    (``record_local_event``) passes ``local_event_recorded=True`` so the line
+    is not appended twice.
     """
     mqtt_common.append_event(prepared.job_id, {
         "event": "published",
@@ -184,6 +201,7 @@ def record_outcome(prepared: PreparedEvent, registry_dir: str,
         "publish_error": publish_error,
     })
 
+    if not local_event_recorded:
         registry.append_event(prepared.job_id, registry_dir, prepared.payload)
     new_status = EVENT_TO_STATUS.get(prepared.event)
     if new_status:
@@ -337,6 +355,10 @@ def main(argv=None) -> int:
         logger.error("%s", exc)
         return 1
 
+    # Local first: the event reaches events.log before the publish can block on
+    # the broker ACK and its retries (see record_local_event).
+    record_local_event(prepared, args.registry_dir)
+
     # Policy Note: seq consumption on failure is intentional (see prepare_event)
     # and the failure is documented explicitly in the audit record.
     publish = with_retry(
@@ -362,7 +384,8 @@ def main(argv=None) -> int:
 
     record_outcome(prepared, args.registry_dir, publish_ok, publish_error,
                    expected_dispatch_token=sync_expected_token,
-                   expected_attempt=sync_expected_attempt)
+                   expected_attempt=sync_expected_attempt,
+                   local_event_recorded=True)
 
     if not publish_ok:
         return 2

.agents/skills/lib_py/delegation/service.py

@@ -84,6 +84,9 @@ class DelegationService(object):
             data=data, registry_dir=self.registry_dir,
             dispatch_token=eff_dispatch_token, attempt_id=eff_attempt_id,
         )
+        # Local first, as in the CLI: the event reaches events.log before the
+        # publish can block on the broker (see publish_event.record_local_event).
+        publish_event.record_local_event(prepared, self.registry_dir)
         publisher = self._publisher
         if publisher is None:
             from lib_py.delegation.publisher import OneShotPublisher
@@ -93,6 +96,7 @@ class DelegationService(object):
             prepared, self.registry_dir, ok, error,
             expected_dispatch_token=eff_dispatch_token,
             expected_attempt=eff_attempt_id,
+            local_event_recorded=True,
         )
         return 0 if ok else 2
 

.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job

@@ -621,13 +621,13 @@ _within_deadline() {
 }
 
 # ISSUE-14: agent-agnostic liveness. publish_event appends every event to the
-# job's on-disk events.log right after its wire publish attempt, whether that
+# job's on-disk events.log BEFORE its wire publish (record_local_event), so a
-# publish succeeded or not, so a `progress` heartbeat the subscriber never saw
+# `progress` heartbeat the subscriber never saw (it had already idled out, or
-# (it had already idled out, or the broker dropped it) is still visible here -
+# the broker dropped it) is visible here the moment it is sent - a slow or
-# late by at most its publish retries while the broker is failing. An agent
+# failing broker no longer delays it by its publish retries. An agent running
-# running its tests in a background shell looks idle to herdr and shows no
+# its tests in a background shell looks idle to herdr and shows no busy
-# busy token, yet keeps heartbeating - this is the only evidence that covers
+# token, yet keeps heartbeating - this is the only evidence that covers that
-# that case.
+# case.
 _recent_disk_heartbeat() {
   local job_id="$1" window="$2" f="$REGISTRY_DIR/$job_id.events.log"
   [ -f "$f" ] || return 1

VERSIONS.md

@@ -47,11 +47,11 @@
 3. **쉼 `ls` 의 "no server running" 번역 (Issue #9)**: 서버가 내려가 있으면 쉼의 `ls` 가 `no server running (herdr session <name>)` 으로 보고하고, `reconcile.sh` 는 이를 빈 서버로 판정해 `herdr_confirmed` 를 유지한다(파싱 실패로 모든 서버의 drift 감지가 멈추는 회귀 방지). 번역은 `agent list` 가 **실패**하고 그 오류 봉투(stderr·stdout 어느 쪽이든)의 `error.code` 가 정확히 `server_not_running` 일 때만 적용한다. 성공한 목록에 같은 문자열이 들어 있으면(예: 에이전트 이름 `server_not_running-codex`) 번역하지 않으므로 살아 있는 서버가 빈 서버로 오판되어 drift A 로 전 세션이 종료 처리되지 않는다. 그 밖의 실패는 출력이 목록 파서에 닿기 전에 비-0 으로 끝나(`… state unknown`, reconcile 의 빈 서버 문구와 겹치지 않음) unknown(`herdr_confirmed=False`)으로 남는다 — stdout 으로 온 오류 문서도 더 이상 빈 목록으로 읽히지 않는다.
 4. **accepted 제출의 전달 확인 `SKS_DELIVERY_GRACE_S` (Issue #14, `send_keys_safe`)**: `agent prompt --wait` 결과가 `timeout`/`agent_prompt_stalled` 이면 herdr 가 입력을 이미 수락한 것으로 보고, `SKS_DELIVERY_GRACE_S`(기본 15초) 동안 `agent_status` working, 제출 전 화면에 없던 busy 토큰(`esc to interrupt`), 제출 전 스냅샷 대비 pane 변화 중 하나를 전달 확인 신호로 인정한다. 신호가 없으면 종전대로 rc 4 로 fail-closed 하고, 그 밖의 오류(`transport_error` 등)는 accepted 로 취급하지 않는다. 단일 주입 계약(재전송 없음)은 유지된다.
 5. **래퍼 레지스트리 ack 유예 `MAM_INJECT_ACK_GRACE_S` (Issue #14, `multi-agent-mux-delegate-job`)**: `send_keys_safe` 가 전달을 확인하지 못해도 잡을 `agent_blocked` 로 닫기 전에 `MAM_INJECT_ACK_GRACE_S`(기본 10초) 동안 레지스트리를 관측하고, `pending` 이 아닌 상태(에이전트의 `started` 등)가 보이면 전달된 것으로 보고 진행한다(`_inject_acknowledged`).
-6. **R1 유예의 pane busy 토큰·디스크 heartbeat 인정 (Issue #14, `_agent_still_working`, `_recent_disk_heartbeat`)**: 구독자 idle 타임아웃 후 R1 유예가 herdr `working` 외에 pane 최근 8줄의 `esc to interrupt` 토큰, 그리고 잡 `events.log` 의 최근 이벤트(`publish_event` 가 와이어 발행 시도 **뒤**에 성공 여부와 무관하게 로컬에 기록하는 `progress` heartbeat — 브로커 장애 시에는 발행 재시도 시간만큼 늦게 기록된다)도 작업 중 근거로 받아들인다. 유예에 들어가면 `IDLE_TIMEOUT` 창 하나가 열리고, 근거가 관측될 때마다 그 시점부터 창이 다시 열린다. 따라서 **아무 근거도 없는 무응답 에이전트는 마지막 와이어 이벤트 후 약 2×`IDLE_TIMEOUT` 에 닫히고**(디스크 heartbeat 를 인정하기 전에는 `IDLE_TIMEOUT`; Issue #6 수용 기준 "≤ `IDLE_TIMEOUT`+5s 내 `error`"는 "≤ 2×`IDLE_TIMEOUT` + 여유"로 바뀐다), **heartbeat 간격은 약 2×`IDLE_TIMEOUT` 까지 허용된다**(heartbeat 하나가 `IDLE_TIMEOUT` 만큼 오래될 때까지 근거로 인정되므로). 근거는 idle 창만 늘릴 뿐 벽시계는 늘리지 않는다: 루프는 매 반복 시계로 다시 읽은 `START_TIME + TIMEOUT` 에서 끝나고, herdr 관측 호출도 그 시각에 끊는다(`_within_deadline`). `broker_unavailable` 디스크 재관측 루프도 같은 시계 기준이다. 이로써 idle 타임아웃보다 긴 codex 리뷰, heartbeat 간격이 한 번 idle 창을 넘긴 리뷰, 백그라운드 셸로 테스트를 돌려 herdr 에 idle 로 보이는 creator 가 턴 도중 `deadline_timeout` 으로 닫히고 이후 `completed` 가 거부되던 문제를 막는다.
+6. **R1 유예의 pane busy 토큰·디스크 heartbeat 인정 (Issue #14, `_agent_still_working`, `_recent_disk_heartbeat`)**: 구독자 idle 타임아웃 후 R1 유예가 herdr `working` 외에 pane 최근 8줄의 `esc to interrupt` 토큰, 그리고 잡 `events.log` 의 최근 이벤트(`publish_event` 가 **와이어 발행 전에** 로컬에 기록하는 `progress` heartbeat — 브로커가 느리거나 죽어 있어도 발행·재시도를 기다리지 않고 즉시 디스크에 남는다. CLI `publish_event.py` 와 `DelegationService.publish_event` 모두 `record_local_event` 로 먼저 기록하고, 감사 로그의 발행 결과와 상태 동기화만 발행 뒤에 남긴다)도 작업 중 근거로 받아들인다. 유예에 들어가면 `IDLE_TIMEOUT` 창 하나가 열리고, 근거가 관측될 때마다 그 시점부터 창이 다시 열린다. 따라서 **아무 근거도 없는 무응답 에이전트는 마지막 와이어 이벤트 후 약 2×`IDLE_TIMEOUT` 에 닫히고**(디스크 heartbeat 를 인정하기 전에는 `IDLE_TIMEOUT`; Issue #6 수용 기준 "≤ `IDLE_TIMEOUT`+5s 내 `error`"는 "≤ 2×`IDLE_TIMEOUT` + 여유"로 바뀐다), **heartbeat 간격은 약 2×`IDLE_TIMEOUT` 까지 허용된다**(heartbeat 하나가 `IDLE_TIMEOUT` 만큼 오래될 때까지 근거로 인정되므로). 근거는 idle 창만 늘릴 뿐 벽시계는 늘리지 않는다: 루프는 매 반복 시계로 다시 읽은 `START_TIME + TIMEOUT` 에서 끝나고, herdr 관측 호출도 그 시각에 끊는다(`_within_deadline`). `broker_unavailable` 디스크 재관측 루프도 같은 시계 기준이다. 이로써 idle 타임아웃보다 긴 codex 리뷰, heartbeat 간격이 한 번 idle 창을 넘긴 리뷰, 백그라운드 셸로 테스트를 돌려 herdr 에 idle 로 보이는 creator 가 턴 도중 `deadline_timeout` 으로 닫히고 이후 `completed` 가 거부되던 문제를 막는다.
 
 #### ✅ 품질 검증
 - **T1-7 트리거 변경** (`tests/test_issue6_fail_closed.py`): 주입 실패 유발값을 `prompt_wait_result=timeout` → `transport_error` 로 변경(Issue #14 이후 `timeout` 은 accepted 제출이므로 더 이상 확정 실패가 아님)하고 `MAM_INJECT_ACK_GRACE_S=1` 로 대기를 단축.
-- **신규 테스트 19개**: `tests/test_issue9_server_env.py` 9개(마커 제거, `MAM_NO_AUTOSTART` 부트스트랩 생략·호출 전달, status 조회 무기동, 정적 opt-out, `ls` 번역(최소·herdr 0.9 실제 오류 봉투·stdout 봉투), 성공 목록 속 `server_not_running` 문자열 비오판, 그 밖의 실패 unknown 유지(stderr·stdout), reconcile drift A 비발생, 다른 실패 시 reconcile `herdr_confirmed=False`), `tests/test_issue14_delivery_ack.py` 10개(busy 토큰·pane 변화 확인, 무신호 fail-closed, 선행 busy 토큰 배제, `transport_error` 비수용, mock herdr timeout 확인, 래퍼 레지스트리 ack, idle 타임아웃 pane busy 유예, idle 타임아웃 디스크 heartbeat 유예, heartbeat 가 이어져도 벽시계 `TIMEOUT` 안에서 종료).
+- **신규 테스트 21개**: `tests/test_issue9_server_env.py` 9개(마커 제거, `MAM_NO_AUTOSTART` 부트스트랩 생략·호출 전달, status 조회 무기동, 정적 opt-out, `ls` 번역(최소·herdr 0.9 실제 오류 봉투·stdout 봉투), 성공 목록 속 `server_not_running` 문자열 비오판, 그 밖의 실패 unknown 유지(stderr·stdout), reconcile drift A 비발생, 다른 실패 시 reconcile `herdr_confirmed=False`), `tests/test_issue14_delivery_ack.py` 12개(busy 토큰·pane 변화 확인, 무신호 fail-closed, 선행 busy 토큰 배제, `transport_error` 비수용, mock herdr timeout 확인, 래퍼 레지스트리 ack, idle 타임아웃 pane busy 유예, idle 타임아웃 디스크 heartbeat 유예(`publish_event` 실제 경로), heartbeat 가 이어져도 벽시계 `TIMEOUT` 안에서 종료, 와이어 발행 전 로컬 기록, 발행 실패·재시도에도 로컬 기록 1회).
 - **테스트 기대값 정정**: `tests/test_a4_adapter_contract.py` 가 `4d444ce` 의 agy `exit_key=/exit` 변경을 반영하도록 수정(그동안 전체 스위트에서 1건 실패하던 문제 해소).
 - **실측 (커밋 `497633f` 기록)**: 새로 부트스트랩한 서버에 마커가 없고 claude pane 의 transcript 저장이 복구됨, codex 잡 `d04ebcb5` 의 `started`/`completed` 가 `agent_blocked` 없이 수락됨.
 

tests/test_issue14_delivery_ack.py

@@ -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