fix(issue-18): admit remote attempts atomically, end timed-out commands with their group, fence heartbeats off the terminal, renew liveness only on news

Review round 2 (planner-reviewer-codex-01, NOT PASS at d09c162):

- R1 remote.mirror_job: admission under the registry lock. The same
  dispatch_token is idempotent (a finished attempt never runs again), a
  lower attempt or the same attempt under another token is refused before
  any write, and only a strictly newer attempt replaces the record
  (register_job(supersede_older_attempt=True); events.log kept, seqs stay
  monotonic).
- R2 remote.handle_dispatch: the --exec command runs in its own session;
  at timeout_sec, or when the runner is interrupted (SIGTERM now unwinds
  it), the whole process group is terminated (TERM, then KILL) before the
  error is published.
- R3 terminal barrier: a terminal publish raises an attempt-scoped barrier
  under the lock that allocates seqs, stops the heartbeat timer, gives it
  MAM_HEARTBEAT_DRAIN_S (default 1 s) to exit and then kills it, and only
  then takes its own seq. Beats are admitted under that lock
  (next_seq(admit=...)); a beat still on the wire is cancelled with the
  timer rather than holding the terminal back; and a late non-terminal
  event never publishes a running state over a terminal record.
- R4 job_subscriber: a retained state renews the idle clock only when its
  seq is newer than everything already accepted; re-delivered or older
  snapshots still recover a terminal status.
- tests: five regressions, all failing on 77f2829; T1-14's wall clock
  15 -> 30 s against public-broker latency.
- docs: MESSAGING 2.5 (barrier, state renewal, detection vs final close),
  heartbeat docstring, loop SKILL.md (remote rows need the Python runner),
  VERSIONS v4.5.0 items 1/4/5 and the test count (25).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGpJXhSZasJLU38ksDghe

.agents/skills/lib_py/delegation/heartbeat.py

@@ -15,8 +15,13 @@ timer stops on its own when
   * the job record disappears, or
   * its lifetime exceeds the job's wall-clock timeout plus one period.
 
-A terminal publish also stops it directly through the pidfile
-(`<registry>/<job_id>.heartbeat.pid`) so no heartbeat can trail a `completed`.
+A terminal publish also stops it directly, so no heartbeat can trail a
+`completed` (review round 2, codex R3): it raises an attempt-scoped barrier
+under the registry lock that allocates seqs (a beat either took its seq before
+it or is refused when it asks for one), stops the timer through the pidfile
+(`<registry>/<job_id>.heartbeat.pid`), and waits for the timer to drain its
+in-flight beats - killing it after `MAM_HEARTBEAT_DRAIN_S` - before it takes
+its own seq.
 
 The timer is bound to the agent's life (review B2): it watches one process -
 `--watch-pid`, or by default the nearest ancestor of the publishing process
@@ -36,6 +41,7 @@ checks read.
 from __future__ import annotations
 
 import argparse
+import json
 import logging
 import os
 import signal
@@ -154,8 +160,31 @@ def _is_timer_process(pid: int, job_id: str) -> bool:
     return "lib_py.delegation.heartbeat" in command and job_id in command
 
 
-def stop_timer(job_id: str, registry_dir: str) -> bool:
-    """Terminate the job's timer if one runs. Returns True when one was signalled."""
+DRAIN_ENV = "MAM_HEARTBEAT_DRAIN_S"
+# Short on purpose: the barrier already keeps any beat from taking a seq after
+# the terminal, so the drain only lets an idle timer exit cleanly. A beat still
+# on the wire is cancelled with the timer (its seq is older than the terminal's)
+# rather than holding back the terminal event - which may be the orchestrator
+# closing the job - for seconds of broker round trips.
+DEFAULT_DRAIN_S = 1.0
+
+
+def drain_seconds() -> float:
+    """How long a terminal publish waits for the timer to drain (env override)."""
+    try:
+        return max(0.0, float(os.environ.get(DRAIN_ENV, DEFAULT_DRAIN_S)))
+    except ValueError:
+        return DEFAULT_DRAIN_S
+
+
+def stop_timer(job_id: str, registry_dir: str, drain: Optional[float] = None) -> bool:
+    """Terminate the job's timer if one runs. Returns True when one was signalled.
+
+    ``drain`` (seconds): also wait for the timer to exit - on TERM it stops
+    scheduling and joins its in-flight beats - and SIGKILL it if it has not
+    gone by then, so none of its beats can still publish afterwards. A terminal
+    publish uses this (review round 2, codex R3): "TERM sent" is not "timer
+    stopped"."""
     pid = running_timer(job_id, registry_dir)
     path = pidfile_path(job_id, registry_dir)
     if pid is None or not _is_timer_process(pid, job_id):
@@ -168,6 +197,19 @@ def stop_timer(job_id: str, registry_dir: str) -> bool:
         os.kill(pid, signal.SIGTERM)
     except OSError:
         pass
+    if drain is not None:
+        deadline = time.monotonic() + drain
+        while time.monotonic() < deadline and _pid_alive(pid):
+            time.sleep(0.1)
+        if _pid_alive(pid):
+            logger.warning("heartbeat timer %d for %s did not drain within %.0fs; killing it", pid, job_id, drain)
+            try:
+                os.kill(pid, signal.SIGKILL)
+            except OSError:
+                pass
+            deadline = time.monotonic() + 2.0
+            while time.monotonic() < deadline and _pid_alive(pid):
+                time.sleep(0.05)
     try:
         os.unlink(path)
     except OSError:
@@ -175,6 +217,40 @@ def stop_timer(job_id: str, registry_dir: str) -> bool:
     return True
 
 
+def barrier_path(job_id: str, registry_dir: str) -> str:
+    return os.path.join(registry_dir, "{}.heartbeat.barrier".format(job_id))
+
+
+def raise_barrier(job_id: str, registry_dir: str, dispatch_token=None, attempt_id=None) -> None:
+    """Mark this attempt as ending (review round 2, codex R3).
+
+    Written under the registry lock - the lock `next_seq` takes - so a beat
+    either took its seq before this point (and the terminal publisher then
+    waits for the timer to drain) or meets the barrier when it asks for a seq
+    and is refused without consuming one."""
+    with mqtt_common.registry_lock(registry_dir):
+        with open(barrier_path(job_id, registry_dir), "w") as fh:
+            json.dump({"dispatch_token": dispatch_token, "attempt_id": attempt_id}, fh)
+
+
+def barrier_blocks(job_id: str, registry_dir: str, dispatch_token=None, attempt_id=None) -> bool:
+    """True when a terminal barrier was raised for this attempt (a later
+    attempt, with its own token, is not blocked). Check it under the registry
+    lock when the answer must be atomic with a seq."""
+    try:
+        with open(barrier_path(job_id, registry_dir)) as fh:
+            barrier = json.load(fh)
+    except (OSError, ValueError):
+        return False
+    if not isinstance(barrier, dict):
+        return True
+    if dispatch_token and barrier.get("dispatch_token"):
+        return barrier["dispatch_token"] == dispatch_token
+    if attempt_id is not None and barrier.get("attempt_id") is not None:
+        return int(barrier["attempt_id"]) == int(attempt_id)
+    return True
+
+
 def start_timer(job_id: str, registry_dir: str, interval: int,
                 dispatch_token: Optional[str] = None, attempt_id: Optional[int] = None,
                 python: Optional[str] = None, watch_pid: Optional[int] = None) -> int:
@@ -222,6 +298,8 @@ def _should_stop(job_id: str, registry_dir: str, dispatch_token, attempt_id,
         return "job record gone"
     if rec.get("status") in TERMINAL_STATUSES:
         return "job is {}".format(rec.get("status"))
+    if barrier_blocks(job_id, registry_dir, dispatch_token, attempt_id):
+        return "the terminal barrier for this attempt is raised"
     if dispatch_token and rec.get("dispatch_token") != dispatch_token:
         return "dispatch token changed (re-dispatched)"
     if attempt_id is not None and rec.get("attempt_id", rec.get("iteration", 1)) != attempt_id:
@@ -290,12 +368,15 @@ def run(job_id: str, registry_dir: str, interval: int, dispatch_token=None,
 
         def _beat(argv=argv):
             try:
-                rc = publish_event.main(argv)
+                # _beat=True: admitted under the seq lock, never after this
+                # attempt's terminal barrier (review round 2, codex R3)
+                rc = publish_event.main(argv, _beat=True)
             except Exception as exc:  # pragma: no cover - defensive
                 logger.warning("heartbeat publish raised: %s", exc)
                 rc = 2
             if rc == 4:
-                logger.info("heartbeat publish refused (stale attempt); timer for %s will stop", job_id)
+                logger.info("heartbeat publish refused (terminal barrier, terminal record or stale attempt); "
+                            "timer for %s will stop", job_id)
                 stop["flag"] = True
 
         t = threading.Thread(target=_beat, daemon=True)