fix(issue-18): bind the heartbeat timer to the agent process; scope started to the attempt; ignore re-sent finished dispatches

Review round 1 (claude NOT PASS) corrections:
- B2: the timer now watches a process - `--watch-pid`, or by default the
  nearest agent-CLI ancestor of the publishing command - and stops within
  one period once it is gone, so a crashed agent stops heartbeating and is
  closed after K x N; a remote runner watches itself. Documented as process
  liveness (an alive-but-stuck agent keeps heartbeating).
- R1: `_wait_started` accepts an events.log `started` only when it carries
  the current dispatch token (loop jobs share one events.log).
- R2: the remote runner ignores a re-sent dispatch of an attempt that
  already reached its terminal.
- R3: registry.md EOF blank line (git diff --check clean).
- R4: the ack error states the effective budget when the wall clock cut it.
- P3: stop_timer verifies the pidfile pid is this job's timer before
  signalling; a retained state message keeps its own seq watermark so it
  can never make the monotonic guard drop its own event.
- MULTI_AGENT_RULES.md timeout section updated for the contract.
- tests: agent-process death stops the timer, recycled-pid safety,
  attempt-scoped started, re-sent dispatch; acceptance 3 now kills the
  agent process instead of the timer. Help captures refreshed
  (--watch-pid). The wrapper can be sourced with
  MAM_DELEGATE_JOB_NO_MAIN=1 for helper-level tests.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff

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

@@ -18,6 +18,15 @@ timer stops on its own when
 A terminal publish also stops it directly through the pidfile
 (`<registry>/<job_id>.heartbeat.pid`) so no heartbeat can trail a `completed`.
 
+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
+that is an agent CLI (claude, codex, agy, ...; the agent runs the start
+command from its shell tool, so that ancestor is the agent itself) - and stops
+within one period once that process is gone. A crashed or killed agent
+therefore stops heartbeating and is closed after K x N like any silent one;
+an agent that is alive but stuck keeps heartbeating, which is the contract's
+deliberate trade-off (liveness of the process, not of the work).
+
 Each beat is an ordinary `publish_event` call in this process: it takes the
 next seq, is signed with the job's auth_token, carries the dispatch token and
 attempt id the timer was started with, and is recorded in events.log before
@@ -46,6 +55,64 @@ HEARTBEAT_DETAIL = "heartbeat"
 MAX_IN_FLIGHT = 8
 
 
+# The agent CLIs, as lib_py.agents.base names them. Repeated here rather than
+# imported: lib_py.delegation may not import a sibling lib_py package (a
+# frozen loop binds delegation to a different tree, see __init__).
+AGENT_CLI_NAMES = ("claude", "agy", "hermes", "grok", "opencode", "codex")
+
+
+def _is_agent_cli(command: str) -> bool:
+    """Some token before the first option has an agent basename
+    (`claude ...`, `node .../claude ...`, `/bin/sh .../agy ...`)."""
+    for tok in command.split():
+        if tok.startswith("-"):
+            break
+        if os.path.basename(tok) in AGENT_CLI_NAMES:
+            return True
+    return False
+
+
+def _process_table():
+    """{pid: (ppid, command)} from ps; empty when ps is unavailable."""
+    try:
+        out = subprocess.run(["ps", "-axo", "pid=,ppid=,command="], capture_output=True, text=True).stdout
+    except OSError:
+        return {}
+    table = {}
+    for line in out.splitlines():
+        parts = line.split(None, 2)
+        if len(parts) < 2:
+            continue
+        try:
+            pid, ppid = int(parts[0]), int(parts[1])
+        except ValueError:
+            continue
+        table[pid] = (ppid, parts[2] if len(parts) > 2 else "")
+    return table
+
+
+def nearest_agent_cli_ancestor(start_pid: Optional[int] = None) -> Optional[int]:
+    """PID of the closest ancestor (start_pid included) that is an agent CLI.
+
+    The agent executes the injected start command through its shell tool, so
+    the chain is agent CLI -> shell -> python publish_event; the agent is the
+    first agent-CLI process met walking up. None when there is no such
+    ancestor (a test, a plain shell, a remote runner)."""
+    table = _process_table()
+    pid = start_pid if start_pid is not None else os.getpid()
+    seen = set()
+    while pid and pid > 1 and pid not in seen:
+        seen.add(pid)
+        entry = table.get(pid)
+        if entry is None:
+            return None
+        ppid, command = entry
+        if _is_agent_cli(command):
+            return pid
+        pid = ppid
+    return None
+
+
 def pidfile_path(job_id: str, registry_dir: str) -> str:
     return os.path.join(registry_dir, "{}.heartbeat.pid".format(job_id))
 
@@ -57,7 +124,12 @@ def _pid_alive(pid: int) -> bool:
         return False
     except PermissionError:
         return True
-    return True
+    try:
+        state = subprocess.run(["ps", "-o", "stat=", "-p", str(pid)],
+                               capture_output=True, text=True).stdout.strip()
+    except OSError:
+        return True
+    return bool(state) and not state.startswith("Z")
 
 
 def running_timer(job_id: str, registry_dir: str) -> Optional[int]:
@@ -72,11 +144,21 @@ def running_timer(job_id: str, registry_dir: str) -> Optional[int]:
     return None
 
 
+def _is_timer_process(pid: int, job_id: str) -> bool:
+    """The pid really is this job's timer (a stale pidfile after a crash may
+    name a recycled pid that now belongs to something else)."""
+    entry = _process_table().get(pid)
+    if entry is None:
+        return False
+    command = entry[1]
+    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."""
     pid = running_timer(job_id, registry_dir)
     path = pidfile_path(job_id, registry_dir)
-    if pid is None:
+    if pid is None or not _is_timer_process(pid, job_id):
         try:
             os.unlink(path)
         except OSError:
@@ -95,13 +177,17 @@ def stop_timer(job_id: str, registry_dir: str) -> bool:
 
 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) -> int:
+                python: Optional[str] = None, watch_pid: Optional[int] = None) -> int:
     """Spawn the detached timer process and return its pid.
 
     A timer that already runs for the job is stopped first: a second `started`
     for the same job (a retry of the same command) must not double the rate.
+    ``watch_pid`` binds the timer to a process (default: the nearest agent-CLI
+    ancestor of this process, if any); 0 disables the watch explicitly.
     """
     stop_timer(job_id, registry_dir)
+    if watch_pid is None:
+        watch_pid = nearest_agent_cli_ancestor()
     cmd: List[str] = [python or sys.executable, "-m", "lib_py.delegation.heartbeat",
                       "--job", job_id, "--registry-dir", registry_dir,
                       "--interval", str(int(interval))]
@@ -109,6 +195,8 @@ def start_timer(job_id: str, registry_dir: str, interval: int,
         cmd += ["--expect-dispatch-token", dispatch_token]
     if attempt_id is not None:
         cmd += ["--attempt-id", str(int(attempt_id))]
+    if watch_pid:
+        cmd += ["--watch-pid", str(int(watch_pid))]
     env = dict(os.environ)
     # The timer imports lib_py from the same tree as this process.
     pkg_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
@@ -119,11 +207,15 @@ def start_timer(job_id: str, registry_dir: str, interval: int,
                                 start_new_session=True, env=env, cwd=os.getcwd())
     with open(pidfile_path(job_id, registry_dir), "w") as fh:
         fh.write(str(proc.pid))
-    logger.info("heartbeat timer pid %d for job %s every %ss", proc.pid, job_id, interval)
+    logger.info("heartbeat timer pid %d for job %s every %ss (watching pid %s)",
+                proc.pid, job_id, interval, watch_pid or "none")
     return proc.pid
 
 
-def _should_stop(job_id: str, registry_dir: str, dispatch_token, attempt_id) -> Optional[str]:
+def _should_stop(job_id: str, registry_dir: str, dispatch_token, attempt_id,
+                 watch_pid: Optional[int] = None) -> Optional[str]:
+    if watch_pid and not _pid_alive(watch_pid):
+        return "watched agent process {} is gone".format(watch_pid)
     try:
         rec = load_job(job_id, registry_dir)
     except FileNotFoundError:
@@ -138,7 +230,8 @@ def _should_stop(job_id: str, registry_dir: str, dispatch_token, attempt_id) ->
 
 
 def run(job_id: str, registry_dir: str, interval: int, dispatch_token=None,
-        attempt_id=None, max_lifetime: Optional[float] = None) -> int:
+        attempt_id=None, max_lifetime: Optional[float] = None,
+        watch_pid: Optional[int] = None) -> int:
     """The timer loop. Runs until a stop condition holds."""
     from lib_py.delegation import publish_event  # late import: the CLI imports us
 
@@ -173,7 +266,7 @@ def run(job_id: str, registry_dir: str, interval: int, dispatch_token=None,
         next_due += interval
         if next_due < time.monotonic():
             next_due = time.monotonic() + interval  # fell far behind: resync
-        reason = _should_stop(job_id, registry_dir, dispatch_token, attempt_id)
+        reason = _should_stop(job_id, registry_dir, dispatch_token, attempt_id, watch_pid)
         if reason:
             logger.info("heartbeat timer for %s stopping: %s", job_id, reason)
             break
@@ -227,12 +320,14 @@ def main(argv=None) -> int:
     parser.add_argument("--expect-dispatch-token", default=None)
     parser.add_argument("--attempt-id", type=int, default=None)
     parser.add_argument("--max-lifetime", type=float, default=None)
+    parser.add_argument("--watch-pid", type=int, default=None,
+                        help="stop once this process (the agent) is gone")
     parser.add_argument("-v", "--verbose", action="store_true")
     args = parser.parse_args(argv)
     mqtt_common.ensure_env_loaded()
     mqtt_common.setup_logging(logging.DEBUG if args.verbose else logging.INFO)
     return run(args.job, args.registry_dir, args.interval, args.expect_dispatch_token,
-               args.attempt_id, args.max_lifetime)
+               args.attempt_id, args.max_lifetime, args.watch_pid)
 
 
 if __name__ == "__main__":

.agents/skills/lib_py/delegation/job_subscriber.py

@@ -139,6 +139,7 @@ class _Watcher:
         # gap is only meaningful after that - the initial last_seq comes from
         # the registry and may be behind the wire (or ahead of a fresh remote).
         self.accepted_any: Set[str] = set()
+        self.state_seq: Dict[str, int] = {}
 
     def _note_seq_gap(self, jid: str, expected: int, got: int, via: str, topic: str) -> None:
         """Issue #18: a seq the subscriber never saw is a detected loss."""
@@ -161,11 +162,14 @@ class _Watcher:
         if not isinstance(seq, int):
             logger.warning("drop state for job %s: missing or invalid seq", jid)
             return
-        last = self.last_seq.get(jid, 0)
+        # A state message never advances the event watermark: MQTT orders
+        # messages per connection, not across topics, so a state that overtakes
+        # its own event must not make the monotonic guard drop that event.
+        last = max(self.last_seq.get(jid, 0), self.state_seq.get(jid, 0))
         if jid in self.accepted_any and seq > last + 1:
             self._note_seq_gap(jid, last + 1, seq, "state", msg.topic)
-        if seq > last:
-            self.last_seq[jid] = seq
+        if seq > self.state_seq.get(jid, 0):
+            self.state_seq[jid] = seq
         mqtt_common.append_event(jid, {
             "event": "received",
             "source_event": STATE_EVENT,

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

@@ -297,6 +297,9 @@ def main(argv=None) -> int:
                              "(Issue #18; not with a terminal event)")
     parser.add_argument("--no-state", action="store_true",
                         help="do not publish the retained <topic_prefix>/state message")
+    parser.add_argument("--watch-pid", type=int, default=None,
+                        help="bind the heartbeat timer to this process (default: the nearest "
+                             "agent-CLI ancestor; 0 = no watch)")
     parser.add_argument("-v", "--verbose", action="store_true")
     args = parser.parse_args(argv)
 
@@ -456,7 +459,8 @@ def main(argv=None) -> int:
             hb_attempt = hb_attempt if hb_attempt is not None else rec.get("attempt_id", rec.get("iteration", 1))
         try:
             pid = heartbeat_timer.start_timer(job_id, args.registry_dir, args.heartbeat,
-                                              dispatch_token=hb_token, attempt_id=hb_attempt)
+                                              dispatch_token=hb_token, attempt_id=hb_attempt,
+                                              watch_pid=args.watch_pid)
             logger.info("heartbeat timer started (pid %d, every %ss)", pid, args.heartbeat)
         except OSError as exc:
             logger.error("could not start the heartbeat timer: %s", exc)

.agents/skills/lib_py/delegation/remote.py

@@ -153,7 +153,7 @@ def clear_dispatch(config, topic: str) -> None:
 # remote side
 # ---------------------------------------------------------------------------
 
-def mirror_job(d: Dict[str, Any], registry_dir: str, session: str) -> str:
+def mirror_job(d: Dict[str, Any], registry_dir: str, session: str) -> Optional[str]:
     """Create the local job record a remote agent publishes against.
 
     The record carries the orchestrator's job_id, auth_token, dispatch_token
@@ -165,8 +165,9 @@ def mirror_job(d: Dict[str, Any], registry_dir: str, session: str) -> str:
     path = mqtt_common._job_path(jid, registry_dir)
     if path.exists():
         old = load_job(jid, registry_dir)
-        if (old.get("dispatch_token") == d.get("dispatch_token")
-                and old.get("status") not in ("completed", "error", "cancelled")):
+        if old.get("dispatch_token") == d.get("dispatch_token"):
+            if old.get("status") in ("completed", "error", "cancelled"):
+                return None  # this attempt already ran to its end: a re-sent dispatch is not run again
             return jid  # same attempt, already mirrored (duplicate delivery)
         with mqtt_common.registry_lock(registry_dir):
             path.unlink()
@@ -206,7 +207,8 @@ def _publish(jid: str, registry_dir: str, event: str, detail: str, d: Dict[str,
     if d.get("attempt_id") is not None:
         argv += ["--attempt-id", str(d["attempt_id"])]
     if heartbeat:
-        argv += ["--heartbeat", str(heartbeat)]
+        # The runner publishes in-process, so the timer must die with the runner.
+        argv += ["--heartbeat", str(heartbeat), "--watch-pid", str(os.getpid())]
     return publish_event.main(argv)
 
 
@@ -214,6 +216,10 @@ def handle_dispatch(d: Dict[str, Any], registry_dir: str, session: str,
                     exec_cmd: Optional[str], python: Optional[str] = None) -> int:
     """Run one dispatched job to its terminal event. Returns the agent rc."""
     jid = mirror_job(d, registry_dir, session)
+    if jid is None:
+        logger.info("dispatch for %s attempt %s already finished here; ignoring the re-sent copy",
+                    d.get("job_id"), d.get("attempt_id"))
+        return 0
     heartbeat = int(d.get("heartbeat_sec") or registry.DEFAULT_HEARTBEAT_SEC)
     rc_started = _publish(jid, registry_dir, "started", "remote runner {} claimed the job".format(session),
                           d, heartbeat=heartbeat)