fix(issue-18): drain the previous heartbeat timer before handing over its pidfile

Review round 3 (planner-reviewer-codex-01, NOT PASS R3 at 7e1d3f0):
start_timer stopped a running timer with a bare TERM and overwrote the only
pidfile. A beat the old timer had already admitted - seq taken, recorded on
disk, held at the wire - stayed alive but undiscoverable, so a later
terminal publish drained only the replacement and the old beat could still
put a retained `running` on the broker after `completed`.

- heartbeat.start_timer: stop_timer(drain=MAM_HEARTBEAT_DRAIN_S). The old
  timer is drained or killed, and confirmed gone, before the pidfile names
  the new one. A terminal publish racing the replacement either drains the
  same old pid or meets a new timer whose beats the barrier refuses.
- test: an old timer held at the wire after admission is replaced, then
  the job completes; the old timer is gone before the handover and nothing
  it held follows the terminal (fails on 7e1d3f0).
- docs: MESSAGING 4.3 no longer says any state resets the idle clock;
  MESSAGING 2.5 and VERSIONS state the replacement rule; test count 26.

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

tests/test_issue18_liveness_contract.py

@@ -671,6 +671,95 @@ def test_no_heartbeat_follows_the_terminal_event_even_when_one_is_in_flight(tmp_
             timer.kill()
 
 
+_WIRE_HELD_TIMER = r'''
+import json, os, sys, time
+from lib_py.delegation import heartbeat as hb, publish_event as pe
+
+capture, paused, go = sys.argv[1], sys.argv[2], sys.argv[3]
+
+
+def held_publish(config, topic, body, retain, extra=()):
+    # the beat was admitted and recorded on disk already: hold it at the wire
+    open(paused, "w").close()
+    deadline = time.time() + 20
+    while not os.path.exists(go) and time.time() < deadline:
+        time.sleep(0.05)
+    with open(capture, "a") as fh:
+        fh.write(json.dumps({"who": "old-timer", "event": json.loads(body)["event"],
+                             "state": [json.loads(b).get("status") for _t, b, _r in extra]}) + "\n")
+
+
+pe._publish_once = held_publish
+argv = sys.argv[5:]  # argv[4] names the module so stop_timer recognises this process as the job's timer
+job, reg = argv[argv.index("--job") + 1], argv[argv.index("--registry-dir") + 1]
+with open(hb.pidfile_path(job, reg), "w") as fh:
+    fh.write(str(os.getpid()))
+sys.exit(hb.main(argv))
+'''
+
+
+def test_replacing_the_timer_drains_the_old_one_before_the_terminal(tmp_path, monkeypatch):
+    """Review round 3 (codex R3): a timer replaced by a second
+    `started --heartbeat` must be gone before it loses the pidfile. Its beat -
+    already admitted and on disk, held at the wire boundary - may otherwise be
+    published after `completed`, because the terminal publish can only find
+    and drain the new timer. After the fix the old timer is dead by the time
+    start_timer returns and nothing it held follows the terminal."""
+    from lib_py.delegation import heartbeat, publish_event as pe, 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("replace", registry_dir=str(reg), auth_token="tok-" + uuid.uuid4().hex)
+    tok = _job(reg, jid)["dispatch_token"]
+    capture, paused, go = tmp_path / "wire.ndjson", tmp_path / "beat.paused", tmp_path / "beat.go"
+
+    def fake_publish(config, topic, body, retain, extra=()):
+        with open(capture, "a") as fh:
+            fh.write(json.dumps({"who": "parent", "event": json.loads(body)["event"],
+                                 "state": [json.loads(b).get("status") for _t, b, _r in extra]}) + "\n")
+
+    monkeypatch.setattr(pe, "_publish_once", fake_publish)
+    assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "started",
+                    "--expect-dispatch-token", tok, "--attempts", "1"]) == 0
+    script = tmp_path / "wire_held_timer.py"
+    script.write_text(_WIRE_HELD_TIMER)
+    env = dict(os.environ, PYTHONPATH=str(SKILLS))
+    old = subprocess.Popen([sys.executable, str(script), str(capture), str(paused), str(go),
+                            "lib_py.delegation.heartbeat", "--job", jid, "--registry-dir", str(reg),
+                            "--interval", "1", "--expect-dispatch-token", tok],
+                           cwd=str(tmp_path), env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    new_pid = None
+    try:
+        deadline = time.time() + 15
+        while time.time() < deadline and not paused.exists():
+            time.sleep(0.05)
+        assert paused.exists(), "the old timer's beat never reached the wire boundary"
+
+        new_pid = heartbeat.start_timer(jid, str(reg), 60, dispatch_token=tok, python=sys.executable, watch_pid=0)
+        assert old.poll() is not None, "the replaced timer must be gone before its pidfile is handed over"
+        assert _timer_pid(reg, jid) == new_pid
+
+        assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "completed", "--detail", "PASS",
+                        "--expect-dispatch-token", tok, "--attempts", "1"]) == 0
+        go.touch()  # would release the held beat if its timer were still alive
+        time.sleep(0.5)
+        wire = [json.loads(l) for l in capture.read_text().splitlines() if l.strip()]
+        assert all(w["who"] != "old-timer" for w in wire), f"the replaced timer published: {wire}"
+        assert wire[-1]["event"] == "completed" and wire[-1]["state"] == ["completed"], wire
+        assert [e["event"] for e in _events(reg, jid)][-1] == "completed"
+        deadline = time.time() + 5
+        while time.time() < deadline and _alive(new_pid):
+            time.sleep(0.1)
+        assert not _alive(new_pid), "the terminal publish stops the replacement timer"
+    finally:
+        if old.poll() is None:
+            old.kill()
+        if new_pid and _alive(new_pid):
+            os.kill(new_pid, signal.SIGKILL)
+
+
 def _signed_state(job_id, seq, token, dispatch_token, status="running", attempt_id=1):
     import hashlib, hmac
     from lib_py.delegation import publish_event as pe