fix(issue-18): serialize the heartbeat timer lifecycle per job

Review round 4 (codex R3): two concurrent `start_timer` calls for the same
attempt both passed the "no timer" check, both spawned, and the pidfile
tracked only one - the untracked timer could still publish a retained
`running` after `completed`. A per-job flock (`<job>.heartbeat.lock`,
separate from the registry/seq lock so in-flight beats can still take
their seq) now makes observe/drain/spawn/publish-pid one ownership
operation; the terminal stop takes the same lock; a start that finds the
attempt ended (terminal record or raised barrier) does not arm; the timer
removes its own pidfile only under the lock so a replacement's pid is never
deleted. Regressions: concurrent starts leave exactly one tracked owner that
the terminal stop drains; a start after the barrier arms nothing.

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

@@ -41,6 +41,8 @@ checks read.
 from __future__ import annotations
 
 import argparse
+import contextlib
+import fcntl
 import json
 import logging
 import os
@@ -177,6 +179,32 @@ def drain_seconds() -> float:
         return DEFAULT_DRAIN_S
 
 
+def lifecycle_lock_path(job_id: str, registry_dir: str) -> str:
+    return os.path.join(registry_dir, "{}.heartbeat.lock".format(job_id))
+
+
+@contextlib.contextmanager
+def lifecycle_lock(job_id: str, registry_dir: str):
+    """The per-job timer ownership lock (review round 4, codex R3).
+
+    Observing/draining the previous owner, spawning the replacement and
+    publishing its pid through the pidfile is one ownership operation, and the
+    terminal stop takes part in it, so two concurrent `started --heartbeat`
+    calls cannot both arm a timer with only one of them tracked, and a start
+    that loses the race against a terminal publish finds the barrier and does
+    not arm at all. This is a separate flock from the registry/seq lock on
+    purpose: an in-flight beat needs that lock to take its seq, and the drain
+    waits for exactly those beats, so holding the seq lock across the drain
+    would deadlock."""
+    os.makedirs(registry_dir, exist_ok=True)
+    with open(lifecycle_lock_path(job_id, registry_dir), "a+") as fh:
+        fcntl.flock(fh.fileno(), fcntl.LOCK_EX)
+        try:
+            yield
+        finally:
+            fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
+
+
 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.
 
@@ -184,7 +212,13 @@ def stop_timer(job_id: str, registry_dir: str, drain: Optional[float] = None) ->
     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"."""
+    stopped". Serialised with `start_timer` through the lifecycle lock."""
+    with lifecycle_lock(job_id, registry_dir):
+        return _stop_timer_locked(job_id, registry_dir, drain)
+
+
+def _stop_timer_locked(job_id: str, registry_dir: str, drain: Optional[float] = None) -> bool:
+    """`stop_timer` body; the caller holds the lifecycle lock."""
     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):
@@ -264,10 +298,26 @@ def start_timer(job_id: str, registry_dir: str, interval: int,
     publish that can find only the new timer through the pidfile.
     ``watch_pid`` binds the timer to a process (default: the nearest agent-CLI
     ancestor of this process, if any); 0 disables the watch explicitly.
+    Returns 0 without arming when the attempt already ended (terminal record
+    or raised barrier): a start that lost the race against the terminal publish
+    must not leave a timer behind (review round 4, codex R3).
     """
-    stop_timer(job_id, registry_dir, drain=drain_seconds())
     if watch_pid is None:
         watch_pid = nearest_agent_cli_ancestor()
+    with lifecycle_lock(job_id, registry_dir):
+        _stop_timer_locked(job_id, registry_dir, drain=drain_seconds())
+        try:
+            status = load_job(job_id, registry_dir).get("status")
+        except FileNotFoundError:
+            status = None
+        if status in TERMINAL_STATUSES or barrier_blocks(job_id, registry_dir, dispatch_token, attempt_id):
+            logger.info("not arming a heartbeat timer for %s: the attempt already ended", job_id)
+            return 0
+        return _spawn_timer_locked(job_id, registry_dir, interval, dispatch_token, attempt_id, python, watch_pid)
+
+
+def _spawn_timer_locked(job_id, registry_dir, interval, dispatch_token, attempt_id, python, watch_pid) -> int:
+    """Spawn the timer and publish its pid; the caller holds the lifecycle lock."""
     cmd: List[str] = [python or sys.executable, "-m", "lib_py.delegation.heartbeat",
                       "--job", job_id, "--registry-dir", registry_dir,
                       "--interval", str(int(interval))]
@@ -388,9 +438,13 @@ def run(job_id: str, registry_dir: str, interval: int, dispatch_token=None,
         in_flight.append(t)
     for t in in_flight:
         t.join(timeout=15)
+    # Under the lifecycle lock: a replacement may be writing its own pid to
+    # this file right now, and that owner's pidfile must not be deleted.
     try:
+        with lifecycle_lock(job_id, registry_dir):
             with open(own_pidfile) as fh:
-            if fh.read().strip() == str(os.getpid()):
+                mine = fh.read().strip() == str(os.getpid())
+            if mine:
                 os.unlink(own_pidfile)
     except (OSError, ValueError):
         pass

tests/test_issue18_liveness_contract.py

@@ -258,6 +258,72 @@ def test_heartbeat_stops_when_the_watched_agent_process_dies(tmp_path):
         _kill_timer(reg, jid)
 
 
+def test_concurrent_start_timer_calls_leave_exactly_one_tracked_owner(tmp_path, monkeypatch):
+    """Review round 4 (codex R3): two `start_timer` calls for the same
+    job/attempt at once - both see no timer, both spawn, one pidfile - left an
+    untracked timer that could publish after the terminal event. The lifecycle
+    lock serialises observe/drain/spawn/publish-pid, so exactly one timer
+    survives, the pidfile names it, and the terminal stop drains it."""
+    from lib_py.delegation import heartbeat, registry
+
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
+    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
+    monkeypatch.setenv("MQTT_BROKER", "127.0.0.1")
+    monkeypatch.setenv("MQTT_PORT", "9")
+    reg = tmp_path / "jobs"
+    jid = registry.register_job("race", registry_dir=str(reg))
+    tok = _job(reg, jid)["dispatch_token"]
+    pids = []
+    errors = []
+
+    def start():
+        try:
+            pids.append(heartbeat.start_timer(jid, str(reg), 1, dispatch_token=tok, attempt_id=1, watch_pid=0))
+        except Exception as exc:  # pragma: no cover - surfaced below
+            errors.append(exc)
+
+    threads = [threading.Thread(target=start) for _ in range(2)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(timeout=60)
+    try:
+        assert not errors, errors
+        assert len(pids) == 2 and all(pids)
+        time.sleep(1.0)
+        alive = [p for p in pids if _alive(p)]
+        assert len(alive) == 1, f"exactly one timer may survive two concurrent starts, alive={alive} pids={pids}"
+        assert _timer_pid(reg, jid) == alive[0], "the pidfile names the surviving owner"
+        # the terminal path drains the tracked owner and nothing else is left
+        assert heartbeat.stop_timer(jid, str(reg), drain=5.0) is True
+        assert not any(_alive(p) for p in pids)
+        assert _timer_pid(reg, jid) is None
+    finally:
+        for p in pids:
+            try:
+                os.kill(p, signal.SIGKILL)
+            except OSError:
+                pass
+
+
+def test_start_timer_after_the_terminal_barrier_does_not_arm(tmp_path, monkeypatch):
+    """A `started --heartbeat` that loses the race against the terminal
+    publish must not leave a timer behind."""
+    from lib_py.delegation import heartbeat, registry
+
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
+    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
+    reg = tmp_path / "jobs"
+    jid = registry.register_job("late-start", registry_dir=str(reg))
+    tok = _job(reg, jid)["dispatch_token"]
+    heartbeat.raise_barrier(jid, str(reg), dispatch_token=tok, attempt_id=1)
+    assert heartbeat.start_timer(jid, str(reg), 1, dispatch_token=tok, attempt_id=1, watch_pid=0) == 0
+    assert _timer_pid(reg, jid) is None
+    assert not (reg / f"{jid}.heartbeat.pid").exists()
+
+
 def test_stop_timer_never_signals_a_recycled_pid(tmp_path, monkeypatch):
     """A stale pidfile (timer crashed) may name a pid that now belongs to
     something else; stop_timer must only signal a heartbeat timer of this job."""