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 open(own_pidfile) as fh:
-            if fh.read().strip() == str(os.getpid()):
+        with lifecycle_lock(job_id, registry_dir):
+            with open(own_pidfile) as fh:
+                mine = fh.read().strip() == str(os.getpid())
+            if mine:
                 os.unlink(own_pidfile)
     except (OSError, ValueError):
         pass