fix: lead-estimate heartbeat (v2.1) — effective TTL, key reuse, permanent stops

Adversarial self-review of the alternate-beat fix found confirmed
inward-drift liveness hazards: at steady state every attempt fired at
exactly ttl/2 lead so one failed extend put the retry at lead 0; the 1s
interval floor guaranteed lapse for spec-legal ttl in (1000,2000); and
sleep-after-response re-arming slipped beats by one RTT per cycle.

The heartbeat now estimates its remaining lead from the AUTHORITATIVE
expires_at_ms the server returns, compared clock-skew-free (server-frame
differences plus client-monotonic elapsed only), extending by ttl when
lead < 1.5*ttl and skipping otherwise. Failures retry next beat with the
SAME idempotency key (a lost-response extend cannot double-apply);
permanent codes (RESERVATION_EXPIRED / RESERVATION_FINALIZED /
MAX_EXTENSIONS_EXCEEDED / TENANT_CLOSED / NOT_FOUND, or status 410) stop
the heartbeat for good; the 1s interval floor is removed.

Spec-review round: tenant policy max_reservation_ttl_ms (default 1h)
silently caps granted TTLs and the create response exposes no effective
TTL — seeding from the requested ttl could schedule the first beat hours
after expiry. The client now captures the HTTP Date header
(CyclesResponse.server_date_ms) and seeds the heartbeat from
effective_ttl = clamp(expires_at_ms - Date, 1000, requested), still
skew-free (server-frame difference).

All four heartbeats (sync/async lifecycle, sync/async streaming) share
the design. Spec guidance updated in cycles-protocol#148 (02d1270).

531 tests pass at 100% coverage; ruff and mypy --strict clean.

Author: amavashev

AUDIT.md

@@ -13,15 +13,22 @@
 
 ---
 
-## 2026-07-27 — Heartbeat drift fix + actual_source marker (v0.5.1)
+## 2026-07-27 — Heartbeat lead-estimate redesign + actual_source marker (v0.5.1)
 
 The heartbeat extended by full ttl_ms every ttl/2 beat while extend_by_ms
 is relative to current expiry — drifting expiry outward +ttl/2 per beat
 (zombie budget lockup on kill; max_extensions burned 2× too fast). All
 four heartbeats now alternate-beat extend (lead stays [ttl/2, 1.5×ttl]).
+Self-review found the first fix (alternate-beat) introduced inward-drift
+hazards (single-failure lead-0, sub-2s-ttl floor decay, RTT slippage); the
+heartbeat now runs on a clock-skew-free lead estimate from the
+authoritative expires_at_ms, derives the effective TTL from the Date
+header (tenant max_reservation_ttl_ms caps grants — default 1h), reuses
+the extend idempotency key on retries, and stops permanently on
+expired/finalized/max-extensions/tenant-closed/not-found.
 Commits whose actual was defaulted from the estimate now carry
 metadata.actual_source="estimate" for audit honesty. Spec guidance:
-cycles-protocol#148. 517 tests pass at 100% coverage.
+cycles-protocol#148. 531 tests pass at 100% coverage.
 
 ## 2026-07-27 — Durable commit retries (journal + /v1/events fallback)
 

CHANGELOG.md

@@ -9,7 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
-- **Heartbeat extend drift**: `extend_by_ms` is relative to the reservation's *current* `expires_at_ms` (spec), but the heartbeat extended by the full `ttl_ms` on every `ttl/2` beat — drifting expiry outward by `ttl/2` per beat. A killed process left its reserved budget locked until the drifted expiry (a zombie-reservation window scaling with runtime, bounded only by `max_extensions`), and long runs burned the `max_extensions` budget twice as fast as needed, losing heartbeat protection mid-flight. All four heartbeats (both lifecycles, both streaming context managers) now use alternate-beat extension: extend on the first beat and every second beat after a success, retrying immediately after a failure. Expiry lead stays within `[ttl/2, 1.5×ttl]`; extension consumption is halved. Fleet-wide fix (TS/Java/Rust ship the same change); spec guidance added in cycles-protocol#148.
+- **Heartbeat redesign (lead-estimate)**: the initial alternate-beat fix traded outward drift for inward-drift liveness hazards (single-failure lead-0 retry, guaranteed lapse for ttl < 2s under the old 1s interval floor, RTT slippage) found by adversarial self-review. The heartbeat now estimates its remaining lead from the authoritative `expires_at_ms` the server returns — compared clock-skew-free (server-frame differences + client-monotonic elapsed only) — and extends when lead < 1.5×ttl. It also: derives the **effective TTL** from `expires_at_ms − Date` header (tenant policy `max_reservation_ttl_ms`, default 1h, silently caps grants — a 24h request would otherwise heartbeat 12h late); reuses the same idempotency key when retrying a failed extend (a lost response cannot double-extend); stops permanently on `RESERVATION_EXPIRED`/`RESERVATION_FINALIZED`/`MAX_EXTENSIONS_EXCEEDED`/`TENANT_CLOSED`/`NOT_FOUND`; and drops the 1s interval floor. Spec guidance: cycles-protocol#148.
+- **Heartbeat extend drift** (superseded by the redesign above, kept for history): `extend_by_ms` is relative to the reservation's *current* `expires_at_ms` (spec), but the heartbeat extended by the full `ttl_ms` on every `ttl/2` beat — drifting expiry outward by `ttl/2` per beat. A killed process left its reserved budget locked until the drifted expiry (a zombie-reservation window scaling with runtime, bounded only by `max_extensions`), and long runs burned the `max_extensions` budget twice as fast as needed, losing heartbeat protection mid-flight. All four heartbeats (both lifecycles, both streaming context managers) now use alternate-beat extension: extend on the first beat and every second beat after a success, retrying immediately after a failure. Expiry lead stays within `[ttl/2, 1.5×ttl]`; extension consumption is halved. Fleet-wide fix (TS/Java/Rust ship the same change); spec guidance added in cycles-protocol#148.
 
 ### Added
 

runcycles/client.py

@@ -41,7 +41,14 @@ def _extract_idempotency_key(body: dict[str, Any]) -> str | None:
     return body.get("idempotency_key")
 
 
-_RESPONSE_HEADERS = ("x-request-id", "x-ratelimit-remaining", "x-ratelimit-reset", "x-cycles-tenant", "retry-after")
+_RESPONSE_HEADERS = (
+    "x-request-id",
+    "x-ratelimit-remaining",
+    "x-ratelimit-reset",
+    "x-cycles-tenant",
+    "retry-after",
+    "date",
+)
 
 _BALANCE_FILTER_PARAMS = {"tenant", "workspace", "app", "workflow", "agent", "toolset"}
 

runcycles/lifecycle.py

@@ -40,6 +40,7 @@
 from runcycles.retry import (
     AsyncCommitRetryEngine,
     CommitRetryEngine,
+    _extract_error_code,
     _is_recognized_rejection,
 )
 
@@ -198,6 +199,49 @@ def _build_release_body(reason: str) -> dict[str, Any]:
     return {"idempotency_key": str(uuid.uuid4()), "reason": reason}
 
 
+def _now_mono_ms() -> float:
+    """Monotonic milliseconds — the heartbeat's only clock (test seam)."""
+    return time.monotonic() * 1000.0
+
+
+# Extend failures that can never succeed again — the heartbeat stops on them.
+_PERMANENT_EXTEND_CODES = frozenset(
+    {
+        "RESERVATION_EXPIRED",
+        "RESERVATION_FINALIZED",
+        "MAX_EXTENSIONS_EXCEEDED",
+        "TENANT_CLOSED",  # closure is irreversible per cascade semantics
+        "NOT_FOUND",  # a purged reservation never comes back
+    }
+)
+
+
+def _effective_ttl_ms(
+    requested_ttl_ms: int,
+    expires_at_ms: int | None,
+    server_date_ms: int | None,
+) -> int:
+    """The TTL the server actually granted, best-effort.
+
+    Tenant policy ``max_reservation_ttl_ms`` (default 1h) silently caps the
+    granted TTL, and the create response has no effective-TTL field —
+    scheduling the heartbeat from the REQUESTED ttl can put the first beat
+    long after expiry. Derive the grant from two server-frame values —
+    ``expires_at_ms`` minus the HTTP ``Date`` header — which stays
+    clock-skew-free (the header's ~1s resolution is negligible against
+    multi-second TTLs). Falls back to the requested ttl when either value
+    is unavailable.
+    """
+    if expires_at_ms is None or server_date_ms is None:
+        return requested_ttl_ms
+    derived = expires_at_ms - server_date_ms
+    return max(1000, min(derived, requested_ttl_ms))
+# Lead threshold: extend when the estimated remaining lifetime drops below
+# this multiple of ttl. Attempts then happen with ~ttl of margin, tolerating
+# failed beats; the success-path lead stays within ~[ttl, 2*ttl].
+_LEAD_TARGET_FACTOR = 1.5
+
+
 def _build_extend_body(ttl_ms: int) -> dict[str, Any]:
     validate_extend_by_ms(ttl_ms)
     return {"idempotency_key": str(uuid.uuid4()), "extend_by_ms": ttl_ms}
@@ -354,7 +398,8 @@ def execute(
 
         # Start heartbeat
         heartbeat_stop = threading.Event()
-        heartbeat_thread = self._start_heartbeat(reservation_id, cfg.ttl_ms, ctx, heartbeat_stop)
+        hb_ttl = _effective_ttl_ms(cfg.ttl_ms, res_result.expires_at_ms, res_response.server_date_ms)
+        heartbeat_thread = self._start_heartbeat(reservation_id, hb_ttl, ctx, heartbeat_stop)
 
         try:
             result = fn(*args, **kwargs)
@@ -474,31 +519,61 @@ def _start_heartbeat(
     ) -> threading.Thread | None:
         if ttl_ms <= 0:
             return None
-        interval_s = max(ttl_ms / 2, 1000) / 1000.0
+        # No 1s floor: for spec-legal ttl < 2000 a floored interval cannot
+        # keep the reservation alive (each extend adds only ttl of lifetime).
+        interval_s = (ttl_ms / 2) / 1000.0
 
         def heartbeat_loop() -> None:
-            # Alternate-beat extension: extend_by_ms is relative to the
-            # CURRENT expiry (spec), so extending by ttl on every ttl/2 beat
-            # drifts expiry outward +ttl/2 per beat — a zombie-reservation
-            # window — and burns max_extensions twice as fast as needed.
-            # Extend on the first beat (only ttl/2 of lifetime remains) and
-            # every second beat after a success; retry right away after a
-            # failure. Expiry lead stays within [ttl/2, 1.5*ttl].
-            beats_since_extend = 1
+            # Lead-estimate heartbeat: extend_by_ms is relative to the
+            # CURRENT expiry (spec), so blind cadence-based extension either
+            # drifts expiry outward or leaves zero margin after a failed
+            # beat. Instead, estimate the remaining lead from the
+            # AUTHORITATIVE expires_at_ms the server returns, compared
+            # skew-free: server-frame differences plus client-monotonic
+            # elapsed only (never client wall clock vs server wall clock).
+            # Extend when lead < 1.5*ttl; skip otherwise. Failed extends are
+            # retried with the SAME body (same idempotency key) so a lost
+            # response cannot double-extend; permanent rejections stop the
+            # heartbeat for good.
+            initial_expiry = ctx.expires_at_ms
+            known_expiry = initial_expiry
+            anchor_ms = _now_mono_ms()
+            pending_body: dict[str, Any] | None = None
             while not stop_event.wait(timeout=interval_s):
-                beats_since_extend += 1
-                if beats_since_extend < 2:
+                elapsed = _now_mono_ms() - anchor_ms
+                if initial_expiry is not None and known_expiry is not None:
+                    lead = (known_expiry - initial_expiry) + ttl_ms - elapsed
+                else:
+                    lead = ttl_ms - elapsed
+                if lead >= _LEAD_TARGET_FACTOR * ttl_ms:
                     continue
                 try:
-                    body = _build_extend_body(ttl_ms)
+                    body = pending_body if pending_body is not None else _build_extend_body(ttl_ms)
+                    pending_body = body
                     response = self._client.extend_reservation(reservation_id, body)
                     if response.is_success:
-                        beats_since_extend = 0
+                        pending_body = None
                         new_expires = response.get_body_attribute("expires_at_ms")
                         if new_expires is not None:
                             ctx.update_expires_at_ms(int(new_expires))
+                            if initial_expiry is None:
+                                # Late anchor: treat this response as the frame origin.
+                                initial_expiry = int(new_expires)
+                                known_expiry = initial_expiry
+                                anchor_ms = _now_mono_ms()
+                            else:
+                                known_expiry = int(new_expires)
+                        elif known_expiry is not None:
+                            known_expiry += ttl_ms
                         logger.debug("Heartbeat extend ok: id=%s", reservation_id)
                     else:
+                        code = _extract_error_code(response)
+                        if response.status == 410 or code in _PERMANENT_EXTEND_CODES:
+                            logger.warning(
+                                "Heartbeat stopping permanently (%s, status=%d): id=%s",
+                                code, response.status, reservation_id,
+                            )
+                            return
                         logger.warning("Heartbeat extend failed: id=%s, status=%d", reservation_id, response.status)
                 except Exception:
                     logger.warning("Heartbeat extend error: id=%s", reservation_id, exc_info=True)
@@ -577,7 +652,8 @@ async def execute(
         )
         _set_context(ctx)
 
-        heartbeat_task = self._start_heartbeat(reservation_id, cfg.ttl_ms, ctx)
+        hb_ttl = _effective_ttl_ms(cfg.ttl_ms, res_result.expires_at_ms, res_response.server_date_ms)
+        heartbeat_task = self._start_heartbeat(reservation_id, hb_ttl, ctx)
 
         try:
             result = await fn(*args, **kwargs)
@@ -691,26 +767,50 @@ async def _handle_release(self, reservation_id: str, reason: str) -> None:
     def _start_heartbeat(self, reservation_id: str, ttl_ms: int, ctx: CyclesContext) -> asyncio.Task[None] | None:
         if ttl_ms <= 0:
             return None
-        interval_s = max(ttl_ms / 2, 1000) / 1000.0
+        # No 1s floor — see the sync heartbeat for rationale.
+        interval_s = (ttl_ms / 2) / 1000.0
 
         async def heartbeat_loop() -> None:
-            # Alternate-beat extension — see the sync heartbeat for rationale.
-            beats_since_extend = 1
+            # Lead-estimate heartbeat — see the sync heartbeat for rationale.
+            initial_expiry = ctx.expires_at_ms
+            known_expiry = initial_expiry
+            anchor_ms = _now_mono_ms()
+            pending_body: dict[str, Any] | None = None
             try:
                 while True:
                     await asyncio.sleep(interval_s)
-                    beats_since_extend += 1
-                    if beats_since_extend < 2:
+                    elapsed = _now_mono_ms() - anchor_ms
+                    if initial_expiry is not None and known_expiry is not None:
+                        lead = (known_expiry - initial_expiry) + ttl_ms - elapsed
+                    else:
+                        lead = ttl_ms - elapsed
+                    if lead >= _LEAD_TARGET_FACTOR * ttl_ms:
                         continue
                     try:
-                        body = _build_extend_body(ttl_ms)
+                        body = pending_body if pending_body is not None else _build_extend_body(ttl_ms)
+                        pending_body = body
                         response = await self._client.extend_reservation(reservation_id, body)
                         if response.is_success:
-                            beats_since_extend = 0
+                            pending_body = None
                             new_expires = response.get_body_attribute("expires_at_ms")
                             if new_expires is not None:
                                 ctx.update_expires_at_ms(int(new_expires))
+                                if initial_expiry is None:
+                                    initial_expiry = int(new_expires)
+                                    known_expiry = initial_expiry
+                                    anchor_ms = _now_mono_ms()
+                                else:
+                                    known_expiry = int(new_expires)
+                            elif known_expiry is not None:
+                                known_expiry += ttl_ms
                         else:
+                            code = _extract_error_code(response)
+                            if response.status == 410 or code in _PERMANENT_EXTEND_CODES:
+                                logger.warning(
+                                    "Heartbeat stopping permanently (%s, status=%d): id=%s",
+                                    code, response.status, reservation_id,
+                                )
+                                return
                             logger.warning("Heartbeat extend failed: id=%s", reservation_id)
                     except Exception:
                         logger.warning("Heartbeat extend error: id=%s", reservation_id, exc_info=True)

runcycles/response.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
+from email.utils import parsedate_to_datetime
 from typing import Any
 
 from runcycles.models import ErrorResponse
@@ -73,6 +74,23 @@ def retry_after_ms_header(self) -> int | None:
         except ValueError:
             return None
 
+    @property
+    def server_date_ms(self) -> int | None:
+        """HTTP ``Date`` header as epoch milliseconds (server wall clock).
+
+        Server-frame, so differencing it against other server-frame values
+        (like ``expires_at_ms``) is clock-skew-free to within the header's
+        one-second resolution plus transit latency. Returns ``None`` when
+        absent or unparseable.
+        """
+        val = self.headers.get("date")
+        if val is None:
+            return None
+        try:
+            return int(parsedate_to_datetime(val).timestamp() * 1000)
+        except (ValueError, TypeError):
+            return None
+
     @property
     def is_success(self) -> bool:
         return 200 <= self.status < 300