diff --git a/AUDIT.md b/AUDIT.md index 423d652..82674ec 100644 --- a/AUDIT.md +++ b/AUDIT.md @@ -13,6 +13,58 @@ --- +## 2026-07-28 — Server-authoritative heartbeat via remaining_ttl_ms (v0.5.1, same PR) + +Spec review round 5 proved the fallback heuristic's regime detection +undecidable (sticky band window [0.75×min(ttl/2,30s), 0.9×ttl)) and +immediate priming schedule-dependent under maximum-lead clamping, so the +protocol adopted remaining_ttl_ms on create+extend responses (spec +v0.1.25.16; cycles-server v0.1.25.59 emits it, recomputed fresh on +idempotent replays and excluded from evidence). The SDK implements the +spec's hardened NORMATIVE algorithm when the field is present: per-attempt +rtt, lead_floor = max(0, remaining − rtt), retry_reserve = +2×max(request_timeout_budget, 1s, 2×rtt_max) + max(1s, 2×rtt_max) (the +enforced httpx timeouts define the budget), next beat at lead_floor − +retry_reserve from response receipt, recomputed per schema-valid 200 +(non-200 2xx is ambiguous → same-key recovery); recovery repeats while +retry_window = lead_estimate − attempt_budget − safety_margin stays +positive with a no-progress guard, 429 Retry-After honored only inside the +window, other 4xx stop without key rotation; a lease that cannot hold the +reserve gets one immediate fresh attempt then stop-and-surface; lead_min +skip bypassed; no primed extension when the create carries the field. +Bookkeeping keeps running so the v2.3+band heuristic (now explicitly +best-effort fallback) resumes seamlessly if the field disappears. Final +self-review also made the response contract uniform across both scheduling +modes: only a complete schema-valid HTTP 200 create/extend response succeeds; +ambiguous 2xx keeps the same key. The enforced timeout covers the whole +attempt, first-delay setup time is deducted, and reliable pre-field RTT +samples remain in the safety budget. 597 tests pass at 99.07% coverage. + +## 2026-07-27 — Heartbeat conservative-lead 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). Four +adversarial review rounds refined the replacement; final (v2.3) design: +conservative lead LOWER BOUND lead_min = Σ measured grants − monotonic +elapsed (grants from successive returned expires_at_ms — same server +frame only, no cross-clock arithmetic); FIRST extension fires immediately +(any bounded first delay can outlive a tenant-policy-capped lease); +cadence splits by regime — a grant tracking the lease drives +clamp(grant/2, 500ms, ttl/2), while a grant merely mirroring elapsed time +(maximum-lead clamping: grant ≤ 0, or grant < 0.9×ttl inside a +[0.75, 1.25]×elapsed band — the lower edge keeps a post-skip small grant +from sticking in the hold) carries no wire cadence signal, so the loop +holds min(ttl/2, 30s) and warns once instead of burning max_extensions at +the floor; a transient failure on the primed beat backs off to the held +cadence (no hot loop); skip at lead_min ≥ 1.5×last_grant; extend +idempotency key reused on retries; permanent stop on expired/finalized/ +max-extensions/tenant-closed/not-found (and raw 404/410). The HTTP Date +header plays no heartbeat role (RFC 9110 §6.6.1; Redis TIME vs container +clock). Commits whose actual was defaulted from the estimate now carry +metadata.actual_source="estimate" for audit honesty. Spec guidance: +cycles-protocol#148. 533 tests pass at 100% coverage. + ## 2026-07-27 — Durable commit retries (journal + /v1/events fallback) Pending commits no longer exist only in memory: the retry engines journal diff --git a/CHANGELOG.md b/CHANGELOG.md index 010ddd7..3d551cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,20 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog 1.1.0](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.5.1] - 2026-07-27 + +### Fixed + +- **Final lease-response and timing conformance (supersedes the older fallback wording below).** Both server-authoritative and fieldless fallback scheduling count only a complete, schema-valid HTTP 200 create/extend response as success; malformed or non-200 2xx responses remain ambiguous and are recovered with the same idempotency key. Create performs one same-key recovery attempt, the enforced timeout covers the whole attempt, post-receipt setup time is deducted from the first delay, and reliable RTT samples gathered before a rolling-upgrade server starts sending `remaining_ttl_ms` remain part of the safety budget. +- **Heartbeat redesign (conservative lead lower bound, immediate prime, regime-aware cadence)**: four adversarial review rounds refined the design. The heartbeat maintains `lead_min = Σ measured grants − monotonic elapsed` (grants = differences of successive returned `expires_at_ms` — the same server frame; no cross-clock arithmetic anywhere), starting at 0, so the FIRST extension fires immediately: any bounded first delay could outlive a tenant-policy-capped lease (a 24h request silently capped to seconds would expire before a delayed first beat), and priming costs one extension without changing total protected runtime. Cadence then splits by regime, detected from the measured grant: when the grant tracks the requested lease, cadence = `clamp(grant/2, 500ms, ttl/2)` — per-extend policy clamps automatically tighten the beat; when the grant merely mirrors elapsed time (maximum-lead clamping: `grant ≤ 0`, or `grant < 0.9×ttl` with `grant` inside `[0.75, 1.25]×elapsed-since-last-success`), no cadence signal exists on the wire — the loop holds `min(ttl/2, 30s)` and warns once that the extension budget will deplete, instead of collapsing to the floor and burning `max_extensions` in seconds. The band's lower edge makes misclassification non-sticky: a real per-extend grant seen across a skip-doubled gap (where grant ≈ elapsed exactly) lands in the hold once, but at the held cadence its grant/elapsed ratio falls below 0.75 and cadence re-tightens — with an upper bound alone the hold would stick and a clamped-grant lease would decay to a lapse. A transient failure on the primed (delay-0) beat backs off to the held cadence — never a hot loop. Skip when `lead_min ≥ 1.5×last_grant`. The HTTP `Date` header plays no role in the heartbeat (RFC 9110 §6.6.1: best-effort, whole-second, and possibly a different clock than the one stamping `expires_at_ms` — Redis TIME in the reference server); `CyclesResponse.server_date_ms` remains as a general accessor. Failed extends retry with the same idempotency key; permanent codes (`RESERVATION_EXPIRED`/`RESERVATION_FINALIZED`/`MAX_EXTENSIONS_EXCEEDED`/`TENANT_CLOSED`/`NOT_FOUND`, raw 404/410) stop the heartbeat; no interval floor. Spec guidance: cycles-protocol#148. +- **Heartbeat (superseded intermediate designs, kept for history)**: 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 + +- **Server-authoritative heartbeat scheduling via `remaining_ttl_ms`** (spec v0.1.25.16, cycles-protocol#148): when a create or extend response carries `remaining_ttl_ms` — the server's own statement of the live lease, measured on the clock that stamps `expires_at_ms` — the heartbeat runs the spec's NORMATIVE algorithm: per-attempt monotonic rtt (max tracked), `lead_floor = max(0, remaining − rtt)`, `attempt_budget = max(request_timeout_budget, 1s, 2×max rtt)` (the SDK's enforced httpx timeouts: connect + read + write), `safety_margin = max(1s, 2×max rtt)`, `retry_reserve = 2×attempt_budget + safety_margin`, next beat at `lead_floor − retry_reserve` after response receipt, recomputed on every field-bearing schema-valid HTTP 200 — never accumulated from expiry differences; the heuristic `lead_min` skip is bypassed and no primed first extension is spent (the create response's field drives the first delay). Only a schema-valid 200 counts as an observed success: an ambiguous non-200 2xx is recovered with the SAME idempotency key. Transient failures (timeout/connection/5xx/429/ambiguous 2xx) recover while the freshly recomputed `retry_window = lead_estimate − attempt_budget − safety_margin` stays positive (progress-guarded against zero-time loops; window < 0 stops and surfaces); a 429's `Retry-After` (delta-seconds) is honored only inside the window, never re-invented earlier; other 4xx stop without key rotation. A success whose lease cannot hold the retry reserve permits ONE immediate fresh attempt, then the heartbeat stops and surfaces that the lease is shorter than the retry-safety budget. Grants/lead bookkeeping keeps running in both modes, so the heuristic below takes over seamlessly if the field disappears; fieldless servers see unchanged fallback behavior. The heuristic (conservative lead lower bound + regime band) is now explicitly a **best-effort fallback**: spec review round 5 proved regime detection from `(grant, elapsed)` observables undecidable in general — the band's sticky window `grant ∈ [0.75×min(ttl/2, 30s), 0.9×ttl)` misclassifies permanently — which is precisely why the wire field exists. +- `metadata.actual_source: "estimate"` is stamped on commits whose actual was structurally defaulted from the estimate (`@cycles` without an `actual` expression; streams with no recorded cost or a raised `cost_fn`), so audit evidence distinguishes measured spend from assumed spend. Defaults are unchanged; the marker flows into `/v1/events` recovery bodies via the shared metadata. + ## [0.5.0] - 2026-07-27 Durable commit retries. Previously a commit that failed transiently lived only in an in-memory daemon thread (or an unreferenced asyncio task): a process exit — even a clean one — dropped it, and once the reservation's grace period elapsed the server's expiry sweep returned the reserved budget to the pool, permanently under-counting spend that had already happened. Pending commits are now journaled to disk before retry, replayed on the next run, flushed (bounded) at interpreter exit, and — when the reservation has already expired — recovered via `POST /v1/events`, the spec's post-hoc direct-debit endpoint. diff --git a/pyproject.toml b/pyproject.toml index 1c04204..3b5bfb3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "runcycles" -version = "0.5.0" +version = "0.5.1" description = "Python AI agent budget control — enforce LLM cost limits, tool permissions, and multi-tenant policies before agent actions execute." readme = "README.md" license = "Apache-2.0" diff --git a/runcycles/__init__.py b/runcycles/__init__.py index beca392..44d8ec0 100644 --- a/runcycles/__init__.py +++ b/runcycles/__init__.py @@ -25,6 +25,7 @@ CommitRequest, CommitResponse, CommitStatus, + CyclesEvidenceRef, CyclesMetrics, Decision, DecisionRequest, @@ -92,6 +93,7 @@ "Caps", "Decision", "CyclesMetrics", + "CyclesEvidenceRef", "Balance", "CommitOveragePolicy", "ErrorCode", diff --git a/runcycles/client.py b/runcycles/client.py index 4b8581b..75518a0 100644 --- a/runcycles/client.py +++ b/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"} diff --git a/runcycles/lifecycle.py b/runcycles/lifecycle.py index e4afabc..f7701fc 100644 --- a/runcycles/lifecycle.py +++ b/runcycles/lifecycle.py @@ -3,11 +3,14 @@ from __future__ import annotations import asyncio +import json import logging +import math +import queue import threading import time import uuid -from collections.abc import Callable +from collections.abc import Awaitable, Callable from dataclasses import dataclass from typing import Any @@ -34,12 +37,14 @@ Decision, DryRunResult, ReservationCreateResponse, + ReservationExtendResponse, Subject, ) from runcycles.response import CyclesResponse from runcycles.retry import ( AsyncCommitRetryEngine, CommitRetryEngine, + _extract_error_code, _is_recognized_rejection, ) @@ -154,7 +159,10 @@ def _build_reservation_body( def _build_commit_body( - actual: int, unit: str, metrics: CyclesMetrics | None, metadata: dict[str, Any] | None, + actual: int, + unit: str, + metrics: CyclesMetrics | None, + metadata: dict[str, Any] | None, ) -> dict[str, Any]: body: dict[str, Any] = { "idempotency_key": str(uuid.uuid4()), @@ -168,7 +176,10 @@ def _build_commit_body( def _build_event_fallback_body( - reservation_id: str, subject: dict[str, Any], action: dict[str, Any], commit_body: dict[str, Any], + reservation_id: str, + subject: dict[str, Any], + action: dict[str, Any], + commit_body: dict[str, Any], ) -> dict[str, Any]: """Build a POST /v1/events body that records the spend of a commit whose reservation expired before the commit landed (the server has already @@ -198,6 +209,343 @@ 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 + } +) +# Skip an extension while lead_min is at least this multiple of the last +# measured grant; below it, extend. Attempts then carry enough margin to +# tolerate failed beats on the success path. +_LEAD_TARGET_FACTOR = 1.5 + + +def _timeout_budget_ms(config: Any) -> float: + """Outer deadline enforced around one complete create/extend attempt. + + HTTPX's connect/read/write/pool settings are phase or inactivity + timeouts, not a whole-request deadline. The lifecycle therefore wraps + each lease-bearing attempt in this total deadline. Pool acquisition and + response-body parsing are inside the same bound. + """ + parts = (float(config.connect_timeout), float(config.read_timeout), 5.0) + if any(not math.isfinite(part) or part <= 0 for part in parts): + return math.inf + return sum(parts) * 1000.0 + + +def _remaining_at_schedule_start( + remaining_ms: int, + received_ms: float | None, + now_ms: float, +) -> int: + """Deduct local setup time elapsed after the create response arrived.""" + if received_ms is None: + return remaining_ms + elapsed_ms = now_ms - received_ms + if not math.isfinite(elapsed_ms) or elapsed_ms < 0: + return 0 + return max(0, math.floor(remaining_ms - elapsed_ms)) + + +def _run_sync_attempt(call: Callable[[], CyclesResponse], timeout_budget_ms: float) -> CyclesResponse: + """Run one synchronous HTTP attempt under a real whole-attempt deadline. + + HTTPX cannot impose a total deadline over all pool/connect/write/read + phases. A daemon worker lets the heartbeat regain control at the + configured deadline. A timed-out request may still finish in the + background, which is why every recovery reuses the same idempotency key. + """ + if not math.isfinite(timeout_budget_ms): + return call() + + result: queue.Queue[tuple[bool, CyclesResponse | Exception]] = queue.Queue(maxsize=1) + + def invoke() -> None: + try: + result.put((True, call())) + except Exception as exc: # propagate the original client failure + result.put((False, exc)) + + worker = threading.Thread(target=invoke, daemon=True, name="cycles-http-attempt") + worker.start() + worker.join(timeout_budget_ms / 1000.0) + if worker.is_alive(): + raise TimeoutError(f"Cycles HTTP attempt exceeded {timeout_budget_ms:g}ms") + ok, value = result.get_nowait() + if ok: + return value # type: ignore[return-value] + raise value # type: ignore[misc] + + +async def _run_async_attempt( + call: Callable[[], Awaitable[CyclesResponse]], + timeout_budget_ms: float, +) -> CyclesResponse: + """Run one asynchronous HTTP attempt under a whole-attempt deadline.""" + if not math.isfinite(timeout_budget_ms): + return await call() + return await asyncio.wait_for(call(), timeout=timeout_budget_ms / 1000.0) + + +_CREATE_RESPONSE_FIELDS = frozenset( + { + "decision", + "reservation_id", + "affected_scopes", + "expires_at_ms", + "remaining_ttl_ms", + "scope_path", + "reserved", + "caps", + "reason_code", + "retry_after_ms", + "balances", + "cycles_evidence", + } +) + + +def _contains_json_null(value: Any) -> bool: + """Whether a lease response contains an OpenAPI-non-nullable JSON null.""" + if value is None: + return True + if isinstance(value, dict): + return any(_contains_json_null(item) for item in value.values()) + if isinstance(value, list): + return any(_contains_json_null(item) for item in value) + return False + + +def _schema_valid_create(response: CyclesResponse) -> ReservationCreateResponse | None: + """Return a fully validated create body only for exact HTTP 200.""" + if response.status != 200 or not isinstance(response.body, dict): + return None + body = response.body + if set(body) - _CREATE_RESPONSE_FIELDS: + return None + if _contains_json_null(body): + return None + remaining = body.get("remaining_ttl_ms") + if remaining is not None and (not isinstance(remaining, int) or isinstance(remaining, bool) or remaining < 0): + return None + expires = body.get("expires_at_ms") + if expires is not None and (not isinstance(expires, int) or isinstance(expires, bool) or expires < 0): + return None + affected = body.get("affected_scopes") + if not isinstance(affected, list) or any(not isinstance(scope, str) for scope in affected): + return None + try: + return ReservationCreateResponse.model_validate_json(json.dumps(body), strict=True) + except Exception: + return None + + +def _schema_valid_extend(response: CyclesResponse) -> ReservationExtendResponse | None: + """Return a fully validated extend body only for exact HTTP 200.""" + if response.status != 200 or not isinstance(response.body, dict): + return None + body = response.body + if set(body) - {"status", "expires_at_ms", "remaining_ttl_ms", "balances"}: + return None + if _contains_json_null(body): + return None + for name in ("expires_at_ms", "remaining_ttl_ms"): + value = body.get(name) + if value is not None and (not isinstance(value, int) or isinstance(value, bool) or value < 0): + return None + try: + return ReservationExtendResponse.model_validate_json(json.dumps(body), strict=True) + except Exception: + return None + + +def _create_is_recoverable(response: CyclesResponse) -> bool: + return response.status < 0 or response.status >= 500 or 200 <= response.status < 300 + + +def _extend_error_is_recoverable(response: CyclesResponse) -> bool: + """Whether a non-2xx extend outcome is recoverable in field mode.""" + return response.is_transport_error or response.is_server_error or response.status == 429 + + +def _ambiguous_create_error(response: CyclesResponse) -> CyclesProtocolError: + return CyclesProtocolError( + "Create reservation did not produce a schema-valid HTTP 200 response", + status=response.status, + ) + + +def _create_reservation_with_recovery( + client: CyclesClient, + body: dict[str, Any], +) -> tuple[CyclesResponse, ReservationCreateResponse, float, float]: + """Create with at most one immediate same-key ambiguity recovery.""" + timeout_budget_ms = _timeout_budget_ms(client._config) + last_exception: Exception | None = None + for attempt in range(2): + sent_ms = _now_mono_ms() + try: + response = _run_sync_attempt( + lambda: client.create_reservation(body), + timeout_budget_ms, + ) + except Exception as exc: + last_exception = exc + if attempt == 0: + continue + raise CyclesProtocolError( + f"Create reservation remained ambiguous after same-key retry: {exc}", + ) from exc + parsed = _schema_valid_create(response) + if parsed is not None: + received_ms = _now_mono_ms() + return response, parsed, received_ms - sent_ms, received_ms + if attempt == 0 and _create_is_recoverable(response): + continue + if response.status < 200 or response.status >= 300: + raise _build_protocol_exception("Failed to create reservation", response) + raise _ambiguous_create_error(response) + raise AssertionError(last_exception) # pragma: no cover + + +async def _create_reservation_with_recovery_async( + client: AsyncCyclesClient, + body: dict[str, Any], +) -> tuple[CyclesResponse, ReservationCreateResponse, float, float]: + """Async create with at most one immediate same-key ambiguity recovery.""" + timeout_budget_ms = _timeout_budget_ms(client._config) + last_exception: Exception | None = None + for attempt in range(2): + sent_ms = _now_mono_ms() + try: + response = await _run_async_attempt( + lambda: client.create_reservation(body), + timeout_budget_ms, + ) + except Exception as exc: + last_exception = exc + if attempt == 0: + continue + raise CyclesProtocolError( + f"Create reservation remained ambiguous after same-key retry: {exc}", + ) from exc + parsed = _schema_valid_create(response) + if parsed is not None: + received_ms = _now_mono_ms() + return response, parsed, received_ms - sent_ms, received_ms + if attempt == 0 and _create_is_recoverable(response): + continue + if response.status < 200 or response.status >= 300: + raise _build_protocol_exception("Failed to create reservation", response) + raise _ambiguous_create_error(response) + raise AssertionError(last_exception) # pragma: no cover + + +class _AuthoritativeScheduler: + """Field-mode heartbeat scheduling per the NORMATIVE algorithm in the + spec's HEARTBEAT GUIDANCE (v0.1.25.16). All methods return the next + delay in ms, or ``None`` when the spec requires the heartbeat to stop + and surface. State: max observed rtt, the lead floor established by the + last schema-valid response, a zero-delay streak (a success that cannot + hold the retry reserve permits ONE immediate fresh attempt, then stop), + and the last failure's retry window (recovery may repeat with the same + idempotency key only while the freshly recomputed window shrinks).""" + + def __init__(self, timeout_budget_ms: float) -> None: + self._timeout_budget_ms = timeout_budget_ms + self._rtt_max_ms = 0.0 + self._lead_floor_ms: float | None = None + self._lead_anchor_ms: float | None = None + self._zero_streak = 0 + self._prev_fail_window: float | None = None + + def _attempt_budget_ms(self) -> float: + return max(self._timeout_budget_ms, 1000.0, 2.0 * self._rtt_max_ms) + + def _safety_margin_ms(self) -> float: + return max(1000.0, 2.0 * self._rtt_max_ms) + + def observe_rtt(self, rtt_ms: float) -> None: + """Retain every reliable schema-valid create/extend RTT sample.""" + if math.isfinite(rtt_ms) and rtt_ms >= 0: + self._rtt_max_ms = max(self._rtt_max_ms, rtt_ms) + + def on_valid_success( + self, + remaining_ms: int, + rtt_ms: float, + now_ms: float, + ) -> float | None: + """Schema-valid HTTP 200 carrying remaining_ttl_ms. retry_reserve = + 2×attempt_budget + safety_margin covers one failed attempt, one + same-key retry, and margin.""" + if not math.isfinite(rtt_ms) or rtt_ms < 0: + # Unknown/unreliable timing cannot be treated as zero elapsed. + self._rtt_max_ms = math.inf + lead_floor_ms = 0.0 + else: + self.observe_rtt(rtt_ms) + lead_floor_ms = max(0.0, float(remaining_ms) - rtt_ms) + self._prev_fail_window = None + self._lead_floor_ms = lead_floor_ms + self._lead_anchor_ms = now_ms + reserve = 2.0 * self._attempt_budget_ms() + self._safety_margin_ms() + delay = self._lead_floor_ms - reserve + if delay > 0: + self._zero_streak = 0 + return delay + # The lease cannot hold the retry reserve: one immediate fresh + # attempt (new key) is permitted — an additive-delta server may + # establish positive lead — then the client must stop rather than + # burn a maximum-lead server's extension budget in a tight loop. + self._zero_streak += 1 + if self._zero_streak >= 2: + return None + return 0.0 + + def lead_estimate_ms(self, now_ms: float) -> float: + if self._lead_floor_ms is None or self._lead_anchor_ms is None: + return 0.0 + elapsed_ms = now_ms - self._lead_anchor_ms + if not math.isfinite(elapsed_ms) or elapsed_ms < 0: + return 0.0 + return max(0.0, self._lead_floor_ms - elapsed_ms) + + def on_transient_failure( + self, + now_ms: float, + rate_limited: bool = False, + retry_after_ms: int | None = None, + ) -> float | None: + """Timeout / connection error / 5xx / 429 / ambiguous 2xx. The + retry_window is recomputed from the same last schema-valid response; + recovery repeats only while it shrinks (progress guard), and a 429 + may only be honored inside the window — never re-invented earlier.""" + lead_est = self.lead_estimate_ms(now_ms) + window = lead_est - self._attempt_budget_ms() - self._safety_margin_ms() + if window < 0: + return None + if self._prev_fail_window is not None and window >= self._prev_fail_window: + return None # no progress between consecutive failures + self._prev_fail_window = window + if rate_limited: + if retry_after_ms is None or retry_after_ms < 0 or retry_after_ms > window: + return None + return float(retry_after_ms) + return min(30_000.0, lead_est / 4.0, window) + + 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} @@ -271,7 +619,10 @@ class CyclesLifecycle: """Synchronous lifecycle orchestrator: reserve → execute → commit/release.""" def __init__( - self, client: CyclesClient, retry_engine: CommitRetryEngine, default_subject: dict[str, str | None], + self, + client: CyclesClient, + retry_engine: CommitRetryEngine, + default_subject: dict[str, str | None], ) -> None: self._client = client self._retry_engine = retry_engine @@ -294,13 +645,10 @@ def execute( logger.debug("Creating reservation: body=%s", create_body) res_t1 = time.monotonic() - res_response = self._client.create_reservation(create_body) - - if not res_response.is_success: - logger.error("Reservation failed: response=%s", res_response) - raise _build_protocol_exception("Failed to create reservation", res_response) - - res_result = ReservationCreateResponse.model_validate(res_response.body) + res_response, res_result, _create_rtt_ms, _create_received_ms = _create_reservation_with_recovery( + self._client, + create_body, + ) res_t2 = time.monotonic() decision = res_result.decision @@ -335,7 +683,9 @@ def execute( logger.info( "Reservation created: id=%s, decision=%s, elapsed=%dms", - reservation_id, decision, int((res_t2 - res_t1) * 1000), + reservation_id, + decision, + int((res_t2 - res_t1) * 1000), ) # Set context @@ -354,7 +704,15 @@ def execute( # Start heartbeat heartbeat_stop = threading.Event() - heartbeat_thread = self._start_heartbeat(reservation_id, cfg.ttl_ms, ctx, heartbeat_stop) + heartbeat_thread = self._start_heartbeat( + reservation_id, + cfg.ttl_ms, + ctx, + heartbeat_stop, + res_result.remaining_ttl_ms, + _create_rtt_ms, + _create_received_ms, + ) try: result = fn(*args, **kwargs) @@ -371,9 +729,19 @@ def execute( if metrics.latency_ms is None: metrics.latency_ms = method_elapsed - commit_body = _build_commit_body(actual_amount, cfg.unit, metrics, ctx.commit_metadata) + commit_metadata = ctx.commit_metadata + if cfg.actual is None: + # The estimate is being recorded as the actual (documented + # fallback). Mark the evidence so auditors can distinguish + # measured spend from assumed spend. + logger.debug("No actual expression; committing estimate as actual: id=%s", reservation_id) + commit_metadata = {**(commit_metadata or {}), "actual_source": "estimate"} + commit_body = _build_commit_body(actual_amount, cfg.unit, metrics, commit_metadata) event_fallback = _build_event_fallback_body( - reservation_id, create_body["subject"], create_body["action"], commit_body, + reservation_id, + create_body["subject"], + create_body["action"], + commit_body, ) self._handle_commit(reservation_id, commit_body, event_fallback) @@ -390,7 +758,10 @@ def execute( _clear_context() def _handle_commit( - self, reservation_id: str, commit_body: dict[str, Any], event_fallback_body: dict[str, Any], + self, + reservation_id: str, + commit_body: dict[str, Any], + event_fallback_body: dict[str, Any], ) -> None: try: logger.debug("Committing: id=%s", reservation_id) @@ -411,7 +782,9 @@ def _handle_commit( # honoring the server's Retry-After. logger.warning("Commit rate-limited; scheduling retry: id=%s", reservation_id) self._retry_engine.schedule( - reservation_id, commit_body, event_fallback_body, + reservation_id, + commit_body, + event_fallback_body, retry_after_ms=response.retry_after_ms_header, ) elif response.status in (401, 403): @@ -420,7 +793,8 @@ def _handle_commit( # that would return budget for real spend. logger.error( "Commit got authentication failure (status=%d); journaling for replay: id=%s", - response.status, reservation_id, + response.status, + reservation_id, ) self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) elif response.status == 410 or error_code == "RESERVATION_EXPIRED": @@ -439,9 +813,10 @@ def _handle_commit( # Codeless or forward-compat-unknown 4xx: neither release # nor drop — retain the spend record. logger.error( - "Commit got unclassifiable client error (status=%d, error=%s); " - "journaling for replay: id=%s", - response.status, error_code, reservation_id, + "Commit got unclassifiable client error (status=%d, error=%s); journaling for replay: id=%s", + response.status, + error_code, + reservation_id, ) self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) else: @@ -463,26 +838,225 @@ def _handle_release(self, reservation_id: str, reason: str) -> None: logger.exception("Failed to release: id=%s", reservation_id) def _start_heartbeat( - self, reservation_id: str, ttl_ms: int, ctx: CyclesContext, stop_event: threading.Event, + self, + reservation_id: str, + ttl_ms: int, + ctx: CyclesContext, + stop_event: threading.Event, + initial_remaining_ms: int | None = None, + initial_rtt_ms: float | None = None, + initial_received_ms: float | None = None, ) -> threading.Thread | None: if ttl_ms <= 0: return None - interval_s = max(ttl_ms / 2, 1000) / 1000.0 def heartbeat_loop() -> None: - while not stop_event.wait(timeout=interval_s): + # Conservative-lead heartbeat (v2.3): the only rigorous, + # cross-clock-free quantity a client can maintain is a LOWER + # BOUND on its remaining lead: + # lead_min = sum(measured grants) - monotonic elapsed + # where each grant is the difference of successive returned + # expires_at_ms values (same server frame). lead_min starts at + # 0, so the FIRST extension fires immediately — establishing + # real measured margin and revealing the actual per-extend + # grant. Cadence then splits by regime: a grant that tracks the + # lease (grant ≫ elapsed) drives cadence at grant/2; a grant + # that merely mirrors elapsed time (maximum-lead clamping) + # carries no cadence signal, so the loop holds a bounded + # cadence instead of tightening. Skip when lead_min >= + # 1.5*last_grant. Failed extends retry with the SAME body (same + # idempotency key); permanent rejections stop the heartbeat. + prev_expiry = ctx.expires_at_ms + anchor_ms = _now_mono_ms() + grants_sum = 0.0 + last_grant: float | None = None + pending_body: dict[str, Any] | None = None + last_success_ms = anchor_ms + held_delay_ms = min(ttl_ms / 2, 30_000.0) + clamp_warned = False + # Authoritative scheduling (spec v0.1.25.16 PRIMARY ALGORITHM): + # a schema-valid 200 carrying remaining_ttl_ms drives scheduling + # exactly; the measured-grant heuristic below is the + # NON-NORMATIVE fallback for servers that omit the field. + timeout_budget_ms = _timeout_budget_ms(self._client._config) + sched = _AuthoritativeScheduler(timeout_budget_ms) + if initial_rtt_ms is not None: + sched.observe_rtt(initial_rtt_ms) + authoritative = initial_remaining_ms is not None + if initial_remaining_ms is not None: + remaining_at_start = _remaining_at_schedule_start( + initial_remaining_ms, + initial_received_ms, + anchor_ms, + ) + first_delay = sched.on_valid_success( + remaining_at_start, + initial_rtt_ms or 0.0, + anchor_ms, + ) + # Unreachable on the first scheduler call (the zero-delay + # streak cannot be exhausted yet); kept as a typed guard. + if first_delay is None: # pragma: no cover + logger.warning( + "Heartbeat not started: lease shorter than the retry-safety budget: id=%s", + reservation_id, + ) + return + delay_ms = first_delay + else: + # Immediate first extension (fallback): with lead_min starting + # at 0 and no lease signal on the wire, any bounded first + # delay can outlive a policy-capped lease. Priming costs one + # extension; total protected runtime is unchanged. + delay_ms = 0.0 + while not stop_event.wait(timeout=delay_ms / 1000.0): + if not authoritative: + # After the primed (delay-0) first beat, the baseline + # cadence is the held delay — a transient failure must + # not hot-loop. (Authoritative zero delays are meaningful: + # the one-immediate-attempt guards bound them.) + delay_ms = delay_ms or held_delay_ms + if not authoritative: + lead_min = grants_sum - (_now_mono_ms() - anchor_ms) + if last_grant is not None and lead_min >= _LEAD_TARGET_FACTOR * last_grant: + continue try: - body = _build_extend_body(ttl_ms) - response = self._client.extend_reservation(reservation_id, body) - if response.is_success: - new_expires = response.get_body_attribute("expires_at_ms") - if new_expires is not None: - ctx.update_expires_at_ms(int(new_expires)) + body = pending_body if pending_body is not None else _build_extend_body(ttl_ms) + pending_body = body + sent_ms = _now_mono_ms() + response = _run_sync_attempt( + lambda: self._client.extend_reservation(reservation_id, body), + timeout_budget_ms, + ) + parsed_extend = _schema_valid_extend(response) + recv_ms = _now_mono_ms() + if response.is_success and parsed_extend is None: + # Any non-200 or schema-invalid 2xx is ambiguous: it is + # never an observed success and the key stays pending. + logger.warning( + "Heartbeat ambiguous response (status=%d): id=%s", + response.status, + reservation_id, + ) + if authoritative: + nxt = sched.on_transient_failure(_now_mono_ms()) + if nxt is None: + logger.warning( + "Heartbeat stopping: no safe recovery window remains: id=%s", + reservation_id, + ) + return + delay_ms = nxt + else: + delay_ms = held_delay_ms + elif parsed_extend is not None: + pending_body = None + new_expires = parsed_extend.expires_at_ms + ctx.update_expires_at_ms(new_expires) + grant = float(new_expires - prev_expiry) if prev_expiry is not None else float(ttl_ms) + prev_expiry = new_expires + grant = max(grant, 0.0) + now_ms = _now_mono_ms() + elapsed_since_success = now_ms - last_success_ms + last_success_ms = now_ms + grants_sum += grant + last_grant = grant + rtt_ms = recv_ms - sent_ms + sched.observe_rtt(rtt_ms) + remaining = parsed_extend.remaining_ttl_ms + if remaining is not None: + # Server-authoritative lease (spec v0.1.25.16 + # PRIMARY ALGORITHM): schedule from this response + # alone; the heuristic arms below only serve + # servers that omit the field. + authoritative = True + nxt = sched.on_valid_success(remaining, rtt_ms, recv_ms) + if nxt is None: + logger.warning( + "Heartbeat stopping: lease shorter than the retry-safety budget: id=%s", + reservation_id, + ) + return + delay_ms = nxt + elif grant <= 0 or ( + grant < 0.9 * ttl_ms + and 0.75 * elapsed_since_success <= grant <= 1.25 * elapsed_since_success + ): + # Lead-clamping server: the grant mirrors elapsed + # time, not lease size — no cadence signal exists + # on the wire. Hold a bounded cadence; never + # tighten toward the floor (that would burn the + # max_extensions budget in seconds). The lower + # band keeps this non-sticky: a real small grant + # seen across a skip-doubled gap lands here once, + # but at the held cadence its grant/elapsed ratio + # falls below 0.75 and cadence re-tightens. + authoritative = False + delay_ms = held_delay_ms + if not clamp_warned: + clamp_warned = True + logger.warning( + "Server appears to clamp lease lead; extension budget will deplete: id=%s", + reservation_id, + ) + else: + authoritative = False + delay_ms = min(max(grant / 2, 500.0), ttl_ms / 2) logger.debug("Heartbeat extend ok: id=%s", reservation_id) else: + code = _extract_error_code(response) + if response.status in (404, 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) + if authoritative: + rate_limited = response.status == 429 + if not rate_limited and 400 <= response.status < 500: + # Unrecoverable request/auth failure (spec): never + # rotate the key on an unchanged request. + logger.warning( + "Heartbeat stopping on client error (%s, status=%d): id=%s", + code, + response.status, + reservation_id, + ) + return + if not _extend_error_is_recoverable(response): + logger.warning( + "Heartbeat stopping on unexpected HTTP status %d: id=%s", + response.status, + reservation_id, + ) + return + retry_after = response.retry_after_ms_header if rate_limited else None + nxt = sched.on_transient_failure( + _now_mono_ms(), + rate_limited=rate_limited, + retry_after_ms=retry_after, + ) + if nxt is None: + logger.warning( + "Heartbeat stopping: no safe recovery window remains: id=%s", + reservation_id, + ) + return + delay_ms = nxt except Exception: logger.warning("Heartbeat extend error: id=%s", reservation_id, exc_info=True) + if authoritative: + nxt = sched.on_transient_failure(_now_mono_ms()) + if nxt is None: + logger.warning( + "Heartbeat stopping: no safe recovery window remains: id=%s", + reservation_id, + ) + return + delay_ms = nxt t = threading.Thread(target=heartbeat_loop, daemon=True, name=f"cycles-heartbeat-{reservation_id[:12]}") t.start() @@ -493,7 +1067,10 @@ class AsyncCyclesLifecycle: """Asynchronous lifecycle orchestrator: reserve → execute → commit/release.""" def __init__( - self, client: AsyncCyclesClient, retry_engine: AsyncCommitRetryEngine, default_subject: dict[str, str | None], + self, + client: AsyncCyclesClient, + retry_engine: AsyncCommitRetryEngine, + default_subject: dict[str, str | None], ) -> None: self._client = client self._retry_engine = retry_engine @@ -511,12 +1088,9 @@ async def execute( logger.debug("Estimated usage: estimate=%d", estimate) create_body = _build_reservation_body(cfg, estimate, self._default_subject, args, kwargs) - res_response = await self._client.create_reservation(create_body) - - if not res_response.is_success: - raise _build_protocol_exception("Failed to create reservation", res_response) - - res_result = ReservationCreateResponse.model_validate(res_response.body) + res_response, res_result, _create_rtt_ms, _create_received_ms = await _create_reservation_with_recovery_async( + self._client, create_body + ) res_t2 = time.monotonic() decision = res_result.decision @@ -558,7 +1132,14 @@ async def execute( ) _set_context(ctx) - heartbeat_task = self._start_heartbeat(reservation_id, cfg.ttl_ms, ctx) + heartbeat_task = self._start_heartbeat( + reservation_id, + cfg.ttl_ms, + ctx, + res_result.remaining_ttl_ms, + _create_rtt_ms, + _create_received_ms, + ) try: result = await fn(*args, **kwargs) @@ -572,9 +1153,18 @@ async def execute( if metrics.latency_ms is None: metrics.latency_ms = method_elapsed - commit_body = _build_commit_body(actual_amount, cfg.unit, metrics, ctx.commit_metadata) + commit_metadata = ctx.commit_metadata + if cfg.actual is None: + # See the sync lifecycle: estimate recorded as actual is + # marked so the evidence stays honest. + logger.debug("No actual expression; committing estimate as actual: id=%s", reservation_id) + commit_metadata = {**(commit_metadata or {}), "actual_source": "estimate"} + commit_body = _build_commit_body(actual_amount, cfg.unit, metrics, commit_metadata) event_fallback = _build_event_fallback_body( - reservation_id, create_body["subject"], create_body["action"], commit_body, + reservation_id, + create_body["subject"], + create_body["action"], + commit_body, ) await self._handle_commit(reservation_id, commit_body, event_fallback) @@ -594,7 +1184,10 @@ async def execute( _clear_context() async def _handle_commit( - self, reservation_id: str, commit_body: dict[str, Any], event_fallback_body: dict[str, Any], + self, + reservation_id: str, + commit_body: dict[str, Any], + event_fallback_body: dict[str, Any], ) -> None: try: response = await self._client.commit_reservation(reservation_id, commit_body) @@ -613,7 +1206,9 @@ async def _handle_commit( # honoring the server's Retry-After. logger.warning("Commit rate-limited; scheduling retry: id=%s", reservation_id) self._retry_engine.schedule( - reservation_id, commit_body, event_fallback_body, + reservation_id, + commit_body, + event_fallback_body, retry_after_ms=response.retry_after_ms_header, ) elif response.status in (401, 403): @@ -622,7 +1217,8 @@ async def _handle_commit( # that would return budget for real spend. logger.error( "Commit got authentication failure (status=%d); journaling for replay: id=%s", - response.status, reservation_id, + response.status, + reservation_id, ) self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) elif response.status == 410 or error_code == "RESERVATION_EXPIRED": @@ -641,9 +1237,10 @@ async def _handle_commit( # Codeless or forward-compat-unknown 4xx: neither release # nor drop — retain the spend record. logger.error( - "Commit got unclassifiable client error (status=%d, error=%s); " - "journaling for replay: id=%s", - response.status, error_code, reservation_id, + "Commit got unclassifiable client error (status=%d, error=%s); journaling for replay: id=%s", + response.status, + error_code, + reservation_id, ) self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) else: @@ -663,26 +1260,213 @@ async def _handle_release(self, reservation_id: str, reason: str) -> None: except Exception: logger.exception("Failed to release: id=%s", reservation_id) - def _start_heartbeat(self, reservation_id: str, ttl_ms: int, ctx: CyclesContext) -> asyncio.Task[None] | None: + def _start_heartbeat( + self, + reservation_id: str, + ttl_ms: int, + ctx: CyclesContext, + initial_remaining_ms: int | None = None, + initial_rtt_ms: float | None = None, + initial_received_ms: float | None = None, + ) -> asyncio.Task[None] | None: if ttl_ms <= 0: return None - interval_s = max(ttl_ms / 2, 1000) / 1000.0 async def heartbeat_loop() -> None: + # Conservative-lead heartbeat (v2.3) — see the sync heartbeat. + prev_expiry = ctx.expires_at_ms + anchor_ms = _now_mono_ms() + grants_sum = 0.0 + last_grant: float | None = None + pending_body: dict[str, Any] | None = None + last_success_ms = anchor_ms + held_delay_ms = min(ttl_ms / 2, 30_000.0) + clamp_warned = False + # Authoritative scheduling (spec v0.1.25.16 PRIMARY ALGORITHM): + # a schema-valid 200 carrying remaining_ttl_ms drives scheduling + # exactly; the measured-grant heuristic below is the + # NON-NORMATIVE fallback for servers that omit the field. + timeout_budget_ms = _timeout_budget_ms(self._client._config) + sched = _AuthoritativeScheduler(timeout_budget_ms) + if initial_rtt_ms is not None: + sched.observe_rtt(initial_rtt_ms) + authoritative = initial_remaining_ms is not None + if initial_remaining_ms is not None: + remaining_at_start = _remaining_at_schedule_start( + initial_remaining_ms, + initial_received_ms, + anchor_ms, + ) + first_delay = sched.on_valid_success( + remaining_at_start, + initial_rtt_ms or 0.0, + anchor_ms, + ) + # Unreachable on the first scheduler call (the zero-delay + # streak cannot be exhausted yet); kept as a typed guard. + if first_delay is None: # pragma: no cover + logger.warning( + "Heartbeat not started: lease shorter than the retry-safety budget: id=%s", + reservation_id, + ) + return + delay_ms = first_delay + else: + # Immediate first extension (fallback): with lead_min starting + # at 0 and no lease signal on the wire, any bounded first + # delay can outlive a policy-capped lease. Priming costs one + # extension; total protected runtime is unchanged. + delay_ms = 0.0 try: while True: - await asyncio.sleep(interval_s) + await asyncio.sleep(delay_ms / 1000.0) + if not authoritative: + # After the primed (delay-0) first beat, the baseline + # cadence is the held delay — a transient failure must + # not hot-loop. (Authoritative zero delays are + # meaningful: the one-immediate-attempt guards bound + # them.) + delay_ms = delay_ms or held_delay_ms + if not authoritative: + lead_min = grants_sum - (_now_mono_ms() - anchor_ms) + if last_grant is not None and lead_min >= _LEAD_TARGET_FACTOR * last_grant: + continue try: - body = _build_extend_body(ttl_ms) - response = await self._client.extend_reservation(reservation_id, body) - if response.is_success: - new_expires = response.get_body_attribute("expires_at_ms") - if new_expires is not None: - ctx.update_expires_at_ms(int(new_expires)) + body = pending_body if pending_body is not None else _build_extend_body(ttl_ms) + pending_body = body + sent_ms = _now_mono_ms() + response = await _run_async_attempt( + lambda: self._client.extend_reservation(reservation_id, body), + timeout_budget_ms, + ) + parsed_extend = _schema_valid_extend(response) + recv_ms = _now_mono_ms() + if response.is_success and parsed_extend is None: + # Any non-200 or schema-invalid 2xx is ambiguous: + # never rotate the pending idempotency key. + logger.warning( + "Heartbeat ambiguous response (status=%d): id=%s", + response.status, + reservation_id, + ) + if authoritative: + nxt = sched.on_transient_failure(_now_mono_ms()) + if nxt is None: + logger.warning( + "Heartbeat stopping: no safe recovery window remains: id=%s", + reservation_id, + ) + return + delay_ms = nxt + else: + delay_ms = held_delay_ms + elif parsed_extend is not None: + pending_body = None + new_expires = parsed_extend.expires_at_ms + ctx.update_expires_at_ms(new_expires) + grant = float(new_expires - prev_expiry) if prev_expiry is not None else float(ttl_ms) + prev_expiry = new_expires + grant = max(grant, 0.0) + now_ms = _now_mono_ms() + elapsed_since_success = now_ms - last_success_ms + last_success_ms = now_ms + grants_sum += grant + last_grant = grant + rtt_ms = recv_ms - sent_ms + sched.observe_rtt(rtt_ms) + remaining = parsed_extend.remaining_ttl_ms + if remaining is not None: + # Server-authoritative lease (spec v0.1.25.16 + # PRIMARY ALGORITHM): schedule from this response + # alone; the heuristic arms below only serve + # servers that omit the field. + authoritative = True + nxt = sched.on_valid_success(remaining, rtt_ms, recv_ms) + if nxt is None: + logger.warning( + "Heartbeat stopping: lease shorter than the retry-safety budget: id=%s", + reservation_id, + ) + return + delay_ms = nxt + elif grant <= 0 or ( + grant < 0.9 * ttl_ms + and 0.75 * elapsed_since_success <= grant <= 1.25 * elapsed_since_success + ): + # Lead-clamping server: the grant mirrors elapsed + # time, not lease size — no cadence signal exists + # on the wire. Hold a bounded cadence; never + # tighten toward the floor (that would burn the + # max_extensions budget in seconds). The lower + # band keeps this non-sticky: a real small grant + # seen across a skip-doubled gap lands here once, + # but at the held cadence its grant/elapsed ratio + # falls below 0.75 and cadence re-tightens. + authoritative = False + delay_ms = held_delay_ms + if not clamp_warned: + clamp_warned = True + logger.warning( + "Server appears to clamp lease lead; extension budget will deplete: id=%s", + reservation_id, + ) + else: + authoritative = False + delay_ms = min(max(grant / 2, 500.0), ttl_ms / 2) else: + code = _extract_error_code(response) + if response.status in (404, 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) + if authoritative: + rate_limited = response.status == 429 + if not rate_limited and 400 <= response.status < 500: + # Unrecoverable request/auth failure (spec): never + # rotate the key on an unchanged request. + logger.warning( + "Heartbeat stopping on client error (%s, status=%d): id=%s", + code, + response.status, + reservation_id, + ) + return + if not _extend_error_is_recoverable(response): + logger.warning( + "Heartbeat stopping on unexpected HTTP status %d: id=%s", + response.status, + reservation_id, + ) + return + retry_after = response.retry_after_ms_header if rate_limited else None + nxt = sched.on_transient_failure( + _now_mono_ms(), + rate_limited=rate_limited, + retry_after_ms=retry_after, + ) + if nxt is None: + logger.warning( + "Heartbeat stopping: no safe recovery window remains: id=%s", + reservation_id, + ) + return + delay_ms = nxt except Exception: logger.warning("Heartbeat extend error: id=%s", reservation_id, exc_info=True) + if authoritative: + nxt = sched.on_transient_failure(_now_mono_ms()) + if nxt is None: + logger.warning( + "Heartbeat stopping: no safe recovery window remains: id=%s", + reservation_id, + ) + return + delay_ms = nxt except asyncio.CancelledError: return diff --git a/runcycles/models.py b/runcycles/models.py index 90b6dfe..a485593 100644 --- a/runcycles/models.py +++ b/runcycles/models.py @@ -4,8 +4,9 @@ from enum import Enum from typing import Annotated, Any +from urllib.parse import urlsplit -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, field_validator class Unit(str, Enum): @@ -97,21 +98,21 @@ def from_string(cls, value: str | None) -> ErrorCode | None: # --- Core value objects --- -_SNAKE_CASE_CONFIG = ConfigDict(populate_by_name=True) +_SNAKE_CASE_CONFIG = ConfigDict(populate_by_name=True, extra="forbid") class Amount(BaseModel): model_config = _SNAKE_CASE_CONFIG unit: Unit - amount: int = Field(ge=0) + amount: int = Field(ge=0, le=9_223_372_036_854_775_807) class SignedAmount(BaseModel): model_config = _SNAKE_CASE_CONFIG unit: Unit - amount: int # can be negative + amount: int = Field(ge=-9_223_372_036_854_775_808, le=9_223_372_036_854_775_807) class Subject(BaseModel): @@ -192,6 +193,22 @@ class Balance(BaseModel): is_over_limit: bool | None = None +class CyclesEvidenceRef(BaseModel): + """Transport reference to a server-issued CyclesEvidence envelope.""" + + model_config = _SNAKE_CASE_CONFIG + + evidence_id: str = Field(pattern=r"^[0-9a-f]{64}$") + cycles_evidence_url: str = Field(min_length=1) + + @field_validator("cycles_evidence_url") + @classmethod + def require_absolute_uri(cls, value: str) -> str: + if not urlsplit(value).scheme: + raise ValueError("cycles_evidence_url must be an absolute URI") + return value + + # --- Request models --- @@ -265,13 +282,18 @@ class ReservationCreateResponse(BaseModel): decision: Decision reservation_id: str | None = None affected_scopes: list[str] - expires_at_ms: int | None = None + expires_at_ms: Annotated[int, Field(ge=0, le=9_223_372_036_854_775_807)] | None = None + # Server-authoritative remaining lease (ms) at response evaluation + # (spec v0.1.25.16). Optional: older servers omit it; when present the + # heartbeat schedules from it directly. + remaining_ttl_ms: Annotated[int, Field(ge=0, le=9_223_372_036_854_775_807)] | None = None scope_path: str | None = None reserved: Amount | None = None caps: Caps | None = None - reason_code: str | None = None - retry_after_ms: int | None = None + reason_code: Annotated[str, Field(max_length=128)] | None = None + retry_after_ms: Annotated[int, Field(ge=0)] | None = None balances: list[Balance] | None = None + cycles_evidence: CyclesEvidenceRef | None = None def is_allowed(self) -> bool: return self.decision in (Decision.ALLOW, Decision.ALLOW_WITH_CAPS) @@ -301,7 +323,8 @@ class ReservationExtendResponse(BaseModel): model_config = _SNAKE_CASE_CONFIG status: ExtendStatus - expires_at_ms: int + expires_at_ms: Annotated[int, Field(ge=0, le=9_223_372_036_854_775_807)] + remaining_ttl_ms: Annotated[int, Field(ge=0, le=9_223_372_036_854_775_807)] | None = None balances: list[Balance] | None = None diff --git a/runcycles/response.py b/runcycles/response.py index c65eed3..c3c86df 100644 --- a/runcycles/response.py +++ b/runcycles/response.py @@ -2,7 +2,9 @@ from __future__ import annotations +import re from dataclasses import dataclass, field +from email.utils import parsedate_to_datetime from typing import Any from runcycles.models import ErrorResponse @@ -25,7 +27,11 @@ def success(cls, status: int, body: dict[str, Any], headers: dict[str, str] | No @classmethod def http_error( - cls, status: int, error_message: str, body: dict[str, Any] | None = None, headers: dict[str, str] | None = None, + cls, + status: int, + error_message: str, + body: dict[str, Any] | None = None, + headers: dict[str, str] | None = None, ) -> CyclesResponse: return cls(status=status, body=body, error_message=error_message, headers=headers or {}) @@ -66,11 +72,31 @@ def retry_after_ms_header(self) -> int | None: and is ignored gracefully). """ val = self.headers.get("retry-after") + if val is None: + return None + stripped = val.strip() + if re.fullmatch(r"[0-9]+", stripped) is None: + return None + seconds = int(stripped) + if seconds > 9_223_372_036_854_775_807 // 1000: + return None + return seconds * 1000 + + @property + def server_date_ms(self) -> int | None: + """HTTP ``Date`` header as epoch milliseconds. + + ``Date`` is best-effort HTTP metadata and may come from a different + clock than body timestamps or be replaced by an intermediary. It is + not used for heartbeat scheduling. Returns ``None`` when absent or + unparseable. + """ + val = self.headers.get("date") if val is None: return None try: - return int(val) * 1000 - except ValueError: + return int(parsedate_to_datetime(val).timestamp() * 1000) + except (ValueError, TypeError): return None @property diff --git a/runcycles/streaming.py b/runcycles/streaming.py index 511da03..706cc6d 100644 --- a/runcycles/streaming.py +++ b/runcycles/streaming.py @@ -11,16 +11,27 @@ from dataclasses import dataclass, field from typing import Any +from runcycles import lifecycle as _lifecycle from runcycles._validation import validate_grace_period_ms, validate_subject, validate_ttl_ms from runcycles.client import AsyncCyclesClient, CyclesClient from runcycles.context import CyclesContext, _clear_context, _set_context from runcycles.exceptions import CyclesProtocolError from runcycles.lifecycle import ( + _LEAD_TARGET_FACTOR, + _PERMANENT_EXTEND_CODES, + _AuthoritativeScheduler, _build_commit_body, _build_event_fallback_body, _build_extend_body, _build_protocol_exception, _build_release_body, + _create_reservation_with_recovery, + _create_reservation_with_recovery_async, + _remaining_at_schedule_start, + _run_async_attempt, + _run_sync_attempt, + _schema_valid_extend, + _timeout_budget_ms, ) from runcycles.models import ( Action, @@ -28,12 +39,12 @@ Caps, CyclesMetrics, Decision, - ReservationCreateResponse, Subject, ) from runcycles.retry import ( AsyncCommitRetryEngine, CommitRetryEngine, + _extract_error_code, _is_recognized_rejection, ) @@ -92,17 +103,22 @@ def _resolve_actual_cost( usage: StreamUsage, cost_fn: Callable[[StreamUsage], int] | None, estimate_amount: int, -) -> int: - """Resolve the actual cost: explicit > cost_fn > estimate fallback.""" +) -> tuple[int, bool]: + """Resolve the actual cost: explicit > cost_fn > estimate fallback. + + Returns ``(amount, from_estimate)`` — the flag is True when the estimate + was substituted for an unmeasured actual, so the commit can carry an + ``actual_source`` marker for audit honesty. + """ if usage.actual_cost is not None: - return usage.actual_cost + return usage.actual_cost, False if cost_fn is not None: try: - return cost_fn(usage) + return cost_fn(usage), False except Exception: logger.warning("cost_fn raised, falling back to estimate", exc_info=True) - return estimate_amount - return estimate_amount + return estimate_amount, True + return estimate_amount, True def _build_stream_metrics( @@ -168,6 +184,9 @@ def __init__( self._usage = StreamUsage() self._reservation_id: str | None = None + self._initial_remaining: int | None = None + self._create_rtt: float | None = None + self._create_received_ms: float | None = None self._caps: Caps | None = None self._decision: Decision = Decision.ALLOW self._ctx: CyclesContext | None = None @@ -207,12 +226,10 @@ def __enter__(self) -> StreamReservation: self._grace_period_ms, ) - response = self._client.create_reservation(body) - - if not response.is_success: - raise _build_protocol_exception("Failed to create reservation", response) - - result = ReservationCreateResponse.model_validate(response.body) + response, result, _create_rtt_ms, _create_received_ms = _create_reservation_with_recovery( + self._client, + body, + ) if result.decision == Decision.DENY: raise _build_protocol_exception("Reservation denied", response) @@ -224,6 +241,9 @@ def __enter__(self) -> StreamReservation: ) self._reservation_id = result.reservation_id + self._initial_remaining = result.remaining_ttl_ms + self._create_rtt = _create_rtt_ms + self._create_received_ms = _create_received_ms self._decision = result.decision self._caps = result.caps @@ -273,11 +293,15 @@ def __exit__( def _handle_commit(self) -> None: elapsed_ms = int((time.monotonic() - self._start_time) * 1000) - actual = _resolve_actual_cost(self._usage, self._cost_fn, self._estimate.amount) + actual, actual_from_estimate = _resolve_actual_cost(self._usage, self._cost_fn, self._estimate.amount) ctx_metrics = self._ctx.metrics if self._ctx else None metrics = _build_stream_metrics(self._usage, elapsed_ms, ctx_metrics) unit = self._estimate.unit if isinstance(self._estimate.unit, str) else self._estimate.unit.value - commit_body = _build_commit_body(actual, unit, metrics, self._metadata) + commit_metadata = self._metadata + if actual_from_estimate: + # Estimate recorded as actual — mark the evidence (audit honesty). + commit_metadata = {**(commit_metadata or {}), "actual_source": "estimate"} + commit_body = _build_commit_body(actual, unit, metrics, commit_metadata) assert self._reservation_id is not None event_fallback = _build_event_fallback_body( @@ -304,7 +328,9 @@ def _handle_commit(self) -> None: # honoring the server's Retry-After. logger.warning("Stream commit rate-limited; scheduling retry: id=%s", self._reservation_id) self._retry_engine.schedule( - self._reservation_id, commit_body, event_fallback, + self._reservation_id, + commit_body, + event_fallback, retry_after_ms=response.retry_after_ms_header, ) elif response.status in (401, 403): @@ -313,7 +339,8 @@ def _handle_commit(self) -> None: # that would return budget for real spend. logger.error( "Stream commit got authentication failure (status=%d); journaling for replay: id=%s", - response.status, self._reservation_id, + response.status, + self._reservation_id, ) self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) elif response.status == 410 or error_code == "RESERVATION_EXPIRED": @@ -334,7 +361,9 @@ def _handle_commit(self) -> None: logger.error( "Stream commit got unclassifiable client error (status=%d, error=%s); " "journaling for replay: id=%s", - response.status, error_code, self._reservation_id, + response.status, + error_code, + self._reservation_id, ) self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) else: @@ -358,24 +387,205 @@ def _handle_release(self, reason: str) -> None: def _start_heartbeat(self) -> threading.Thread | None: if self._ttl_ms <= 0: return None - interval_s = max(self._ttl_ms / 2, 1000) / 1000.0 assert self._reservation_id is not None reservation_id: str = self._reservation_id ctx = self._ctx def heartbeat_loop() -> None: - while not self._heartbeat_stop.wait(timeout=interval_s): + # Conservative-lead heartbeat (v2.3) — see CyclesLifecycle heartbeat. + ttl_ms = self._ttl_ms + prev_expiry = ctx.expires_at_ms if ctx is not None else None + anchor_ms = _lifecycle._now_mono_ms() + grants_sum = 0.0 + last_grant: float | None = None + pending_body: dict[str, Any] | None = None + initial_remaining_ms = self._initial_remaining + initial_rtt_ms = self._create_rtt + initial_received_ms = self._create_received_ms + last_success_ms = anchor_ms + held_delay_ms = min(ttl_ms / 2, 30_000.0) + clamp_warned = False + # Authoritative scheduling (spec v0.1.25.16 PRIMARY ALGORITHM): + # a schema-valid 200 carrying remaining_ttl_ms drives scheduling + # exactly; the measured-grant heuristic below is the + # NON-NORMATIVE fallback for servers that omit the field. + timeout_budget_ms = _timeout_budget_ms(self._client._config) + sched = _AuthoritativeScheduler(timeout_budget_ms) + if initial_rtt_ms is not None: + sched.observe_rtt(initial_rtt_ms) + authoritative = initial_remaining_ms is not None + if initial_remaining_ms is not None: + remaining_at_start = _remaining_at_schedule_start( + initial_remaining_ms, + initial_received_ms, + anchor_ms, + ) + first_delay = sched.on_valid_success( + remaining_at_start, + initial_rtt_ms or 0.0, + anchor_ms, + ) + # Unreachable on the first scheduler call (the zero-delay + # streak cannot be exhausted yet); kept as a typed guard. + if first_delay is None: # pragma: no cover + logger.warning( + "Heartbeat not started: lease shorter than the retry-safety budget: id=%s", + reservation_id, + ) + return + delay_ms = first_delay + else: + # Immediate first extension (fallback): with lead_min starting + # at 0 and no lease signal on the wire, any bounded first + # delay can outlive a policy-capped lease. Priming costs one + # extension; total protected runtime is unchanged. + delay_ms = 0.0 + while not self._heartbeat_stop.wait(timeout=delay_ms / 1000.0): + if not authoritative: + # After the primed (delay-0) first beat, the baseline + # cadence is the held delay — a transient failure must + # not hot-loop. (Authoritative zero delays are meaningful: + # the one-immediate-attempt guards bound them.) + delay_ms = delay_ms or held_delay_ms + if not authoritative: + lead_min = grants_sum - (_lifecycle._now_mono_ms() - anchor_ms) + if last_grant is not None and lead_min >= _LEAD_TARGET_FACTOR * last_grant: + continue try: - body = _build_extend_body(self._ttl_ms) - response = self._client.extend_reservation(reservation_id, body) - if response.is_success: - new_expires = response.get_body_attribute("expires_at_ms") - if new_expires is not None and ctx is not None: - ctx.update_expires_at_ms(int(new_expires)) + body = pending_body if pending_body is not None else _build_extend_body(ttl_ms) + pending_body = body + sent_ms = _lifecycle._now_mono_ms() + response = _run_sync_attempt( + lambda: self._client.extend_reservation(reservation_id, body), + timeout_budget_ms, + ) + parsed_extend = _schema_valid_extend(response) + recv_ms = _lifecycle._now_mono_ms() + if response.is_success and parsed_extend is None: + logger.warning( + "Heartbeat ambiguous response (status=%d): id=%s", + response.status, + reservation_id, + ) + if authoritative: + nxt = sched.on_transient_failure(_lifecycle._now_mono_ms()) + if nxt is None: + logger.warning( + "Heartbeat stopping: no safe recovery window remains: id=%s", + reservation_id, + ) + return + delay_ms = nxt + else: + delay_ms = held_delay_ms + elif parsed_extend is not None: + pending_body = None + new_expires = parsed_extend.expires_at_ms + if ctx is not None: + ctx.update_expires_at_ms(new_expires) + grant = float(new_expires - prev_expiry) if prev_expiry is not None else float(ttl_ms) + prev_expiry = new_expires + grant = max(grant, 0.0) + now_ms = _lifecycle._now_mono_ms() + elapsed_since_success = now_ms - last_success_ms + last_success_ms = now_ms + grants_sum += grant + last_grant = grant + rtt_ms = recv_ms - sent_ms + sched.observe_rtt(rtt_ms) + remaining = parsed_extend.remaining_ttl_ms + if remaining is not None: + # Server-authoritative lease (spec v0.1.25.16 + # PRIMARY ALGORITHM): schedule from this response + # alone; the heuristic arms below only serve + # servers that omit the field. + authoritative = True + nxt = sched.on_valid_success(remaining, rtt_ms, recv_ms) + if nxt is None: + logger.warning( + "Heartbeat stopping: lease shorter than the retry-safety budget: id=%s", + reservation_id, + ) + return + delay_ms = nxt + elif grant <= 0 or ( + grant < 0.9 * ttl_ms + and 0.75 * elapsed_since_success <= grant <= 1.25 * elapsed_since_success + ): + # Lead-clamping server: the grant mirrors elapsed + # time, not lease size — no cadence signal exists + # on the wire. Hold a bounded cadence; never + # tighten toward the floor (that would burn the + # max_extensions budget in seconds). The lower + # band keeps this non-sticky: a real small grant + # seen across a skip-doubled gap lands here once, + # but at the held cadence its grant/elapsed ratio + # falls below 0.75 and cadence re-tightens. + authoritative = False + delay_ms = held_delay_ms + if not clamp_warned: + clamp_warned = True + logger.warning( + "Server appears to clamp lease lead; extension budget will deplete: id=%s", + reservation_id, + ) + else: + authoritative = False + delay_ms = min(max(grant / 2, 500.0), ttl_ms / 2) else: + code = _extract_error_code(response) + if response.status in (404, 410) or code in _PERMANENT_EXTEND_CODES: + logger.warning( + "Stream heartbeat stopping permanently (%s, status=%d): id=%s", + code, + response.status, + reservation_id, + ) + return logger.warning("Stream heartbeat failed: id=%s", reservation_id) + if authoritative: + rate_limited = response.status == 429 + if not rate_limited and 400 <= response.status < 500: + # Unrecoverable request/auth failure (spec): never + # rotate the key on an unchanged request. + logger.warning( + "Heartbeat stopping on client error (%s, status=%d): id=%s", + code, + response.status, + reservation_id, + ) + return + if not _lifecycle._extend_error_is_recoverable(response): + logger.warning( + "Heartbeat stopping on unexpected HTTP status %d: id=%s", + response.status, + reservation_id, + ) + return + retry_after = response.retry_after_ms_header if rate_limited else None + nxt = sched.on_transient_failure( + _lifecycle._now_mono_ms(), + rate_limited=rate_limited, + retry_after_ms=retry_after, + ) + if nxt is None: + logger.warning( + "Heartbeat stopping: no safe recovery window remains: id=%s", + reservation_id, + ) + return + delay_ms = nxt except Exception: logger.warning("Stream heartbeat error: id=%s", reservation_id, exc_info=True) + if authoritative: + nxt = sched.on_transient_failure(_lifecycle._now_mono_ms()) + if nxt is None: + logger.warning( + "Heartbeat stopping: no safe recovery window remains: id=%s", + reservation_id, + ) + return + delay_ms = nxt t = threading.Thread( target=heartbeat_loop, @@ -427,6 +637,9 @@ def __init__( self._usage = StreamUsage() self._reservation_id: str | None = None + self._initial_remaining: int | None = None + self._create_rtt: float | None = None + self._create_received_ms: float | None = None self._caps: Caps | None = None self._decision: Decision = Decision.ALLOW self._ctx: CyclesContext | None = None @@ -465,12 +678,9 @@ async def __aenter__(self) -> AsyncStreamReservation: self._grace_period_ms, ) - response = await self._client.create_reservation(body) - - if not response.is_success: - raise _build_protocol_exception("Failed to create reservation", response) - - result = ReservationCreateResponse.model_validate(response.body) + response, result, _create_rtt_ms, _create_received_ms = await _create_reservation_with_recovery_async( + self._client, body + ) if result.decision == Decision.DENY: raise _build_protocol_exception("Reservation denied", response) @@ -482,6 +692,9 @@ async def __aenter__(self) -> AsyncStreamReservation: ) self._reservation_id = result.reservation_id + self._initial_remaining = result.remaining_ttl_ms + self._create_rtt = _create_rtt_ms + self._create_received_ms = _create_received_ms self._decision = result.decision self._caps = result.caps @@ -534,11 +747,15 @@ async def __aexit__( async def _handle_commit(self) -> None: elapsed_ms = int((time.monotonic() - self._start_time) * 1000) - actual = _resolve_actual_cost(self._usage, self._cost_fn, self._estimate.amount) + actual, actual_from_estimate = _resolve_actual_cost(self._usage, self._cost_fn, self._estimate.amount) ctx_metrics = self._ctx.metrics if self._ctx else None metrics = _build_stream_metrics(self._usage, elapsed_ms, ctx_metrics) unit = self._estimate.unit if isinstance(self._estimate.unit, str) else self._estimate.unit.value - commit_body = _build_commit_body(actual, unit, metrics, self._metadata) + commit_metadata = self._metadata + if actual_from_estimate: + # Estimate recorded as actual — mark the evidence (audit honesty). + commit_metadata = {**(commit_metadata or {}), "actual_source": "estimate"} + commit_body = _build_commit_body(actual, unit, metrics, commit_metadata) assert self._reservation_id is not None event_fallback = _build_event_fallback_body( @@ -563,11 +780,11 @@ async def _handle_commit(self) -> None: # Rate-limited, not rejected: releasing here would return # budget for spend that already happened. Retry instead, # honoring the server's Retry-After. - logger.warning( - "Async stream commit rate-limited; scheduling retry: id=%s", self._reservation_id - ) + logger.warning("Async stream commit rate-limited; scheduling retry: id=%s", self._reservation_id) self._retry_engine.schedule( - self._reservation_id, commit_body, event_fallback, + self._reservation_id, + commit_body, + event_fallback, retry_after_ms=response.retry_after_ms_header, ) elif response.status in (401, 403): @@ -576,7 +793,8 @@ async def _handle_commit(self) -> None: # that would return budget for real spend. logger.error( "Stream commit got authentication failure (status=%d); journaling for replay: id=%s", - response.status, self._reservation_id, + response.status, + self._reservation_id, ) self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) elif response.status == 410 or error_code == "RESERVATION_EXPIRED": @@ -597,7 +815,9 @@ async def _handle_commit(self) -> None: logger.error( "Async stream commit got unclassifiable client error (status=%d, error=%s); " "journaling for replay: id=%s", - response.status, error_code, self._reservation_id, + response.status, + error_code, + self._reservation_id, ) self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) else: @@ -621,7 +841,6 @@ async def _handle_release(self, reason: str) -> None: def _start_heartbeat(self) -> asyncio.Task[None] | None: if self._ttl_ms <= 0: return None - interval_s = max(self._ttl_ms / 2, 1000) / 1000.0 assert self._reservation_id is not None reservation_id: str = self._reservation_id ctx = self._ctx @@ -629,20 +848,202 @@ def _start_heartbeat(self) -> asyncio.Task[None] | None: ttl_ms = self._ttl_ms async def heartbeat_loop() -> None: + # Conservative-lead heartbeat (v2.3) — see CyclesLifecycle heartbeat. + prev_expiry = ctx.expires_at_ms if ctx is not None else None + anchor_ms = _lifecycle._now_mono_ms() + grants_sum = 0.0 + last_grant: float | None = None + pending_body: dict[str, Any] | None = None + initial_remaining_ms = self._initial_remaining + initial_rtt_ms = self._create_rtt + initial_received_ms = self._create_received_ms + last_success_ms = anchor_ms + held_delay_ms = min(ttl_ms / 2, 30_000.0) + clamp_warned = False + # Authoritative scheduling (spec v0.1.25.16 PRIMARY ALGORITHM): + # a schema-valid 200 carrying remaining_ttl_ms drives scheduling + # exactly; the measured-grant heuristic below is the + # NON-NORMATIVE fallback for servers that omit the field. + timeout_budget_ms = _timeout_budget_ms(self._client._config) + sched = _AuthoritativeScheduler(timeout_budget_ms) + if initial_rtt_ms is not None: + sched.observe_rtt(initial_rtt_ms) + authoritative = initial_remaining_ms is not None + if initial_remaining_ms is not None: + remaining_at_start = _remaining_at_schedule_start( + initial_remaining_ms, + initial_received_ms, + anchor_ms, + ) + first_delay = sched.on_valid_success( + remaining_at_start, + initial_rtt_ms or 0.0, + anchor_ms, + ) + # Unreachable on the first scheduler call (the zero-delay + # streak cannot be exhausted yet); kept as a typed guard. + if first_delay is None: # pragma: no cover + logger.warning( + "Heartbeat not started: lease shorter than the retry-safety budget: id=%s", + reservation_id, + ) + return + delay_ms = first_delay + else: + # Immediate first extension (fallback): with lead_min starting + # at 0 and no lease signal on the wire, any bounded first + # delay can outlive a policy-capped lease. Priming costs one + # extension; total protected runtime is unchanged. + delay_ms = 0.0 try: while True: - await asyncio.sleep(interval_s) + await asyncio.sleep(delay_ms / 1000.0) + if not authoritative: + # After the primed (delay-0) first beat, the baseline + # cadence is the held delay — a transient failure must + # not hot-loop. (Authoritative zero delays are + # meaningful: the one-immediate-attempt guards bound + # them.) + delay_ms = delay_ms or held_delay_ms + if not authoritative: + lead_min = grants_sum - (_lifecycle._now_mono_ms() - anchor_ms) + if last_grant is not None and lead_min >= _LEAD_TARGET_FACTOR * last_grant: + continue try: - body = _build_extend_body(ttl_ms) - response = await client.extend_reservation(reservation_id, body) - if response.is_success: - new_expires = response.get_body_attribute("expires_at_ms") - if new_expires is not None and ctx is not None: - ctx.update_expires_at_ms(int(new_expires)) + body = pending_body if pending_body is not None else _build_extend_body(ttl_ms) + pending_body = body + sent_ms = _lifecycle._now_mono_ms() + response = await _run_async_attempt( + lambda: client.extend_reservation(reservation_id, body), + timeout_budget_ms, + ) + parsed_extend = _schema_valid_extend(response) + recv_ms = _lifecycle._now_mono_ms() + if response.is_success and parsed_extend is None: + logger.warning( + "Heartbeat ambiguous response (status=%d): id=%s", + response.status, + reservation_id, + ) + if authoritative: + nxt = sched.on_transient_failure(_lifecycle._now_mono_ms()) + if nxt is None: + logger.warning( + "Heartbeat stopping: no safe recovery window remains: id=%s", + reservation_id, + ) + return + delay_ms = nxt + else: + delay_ms = held_delay_ms + elif parsed_extend is not None: + pending_body = None + new_expires = parsed_extend.expires_at_ms + if ctx is not None: + ctx.update_expires_at_ms(new_expires) + grant = float(new_expires - prev_expiry) if prev_expiry is not None else float(ttl_ms) + prev_expiry = new_expires + grant = max(grant, 0.0) + now_ms = _lifecycle._now_mono_ms() + elapsed_since_success = now_ms - last_success_ms + last_success_ms = now_ms + grants_sum += grant + last_grant = grant + rtt_ms = recv_ms - sent_ms + sched.observe_rtt(rtt_ms) + remaining = parsed_extend.remaining_ttl_ms + if remaining is not None: + # Server-authoritative lease (spec v0.1.25.16 + # PRIMARY ALGORITHM): schedule from this response + # alone; the heuristic arms below only serve + # servers that omit the field. + authoritative = True + nxt = sched.on_valid_success(remaining, rtt_ms, recv_ms) + if nxt is None: + logger.warning( + "Heartbeat stopping: lease shorter than the retry-safety budget: id=%s", + reservation_id, + ) + return + delay_ms = nxt + elif grant <= 0 or ( + grant < 0.9 * ttl_ms + and 0.75 * elapsed_since_success <= grant <= 1.25 * elapsed_since_success + ): + # Lead-clamping server: the grant mirrors elapsed + # time, not lease size — no cadence signal exists + # on the wire. Hold a bounded cadence; never + # tighten toward the floor (that would burn the + # max_extensions budget in seconds). The lower + # band keeps this non-sticky: a real small grant + # seen across a skip-doubled gap lands here once, + # but at the held cadence its grant/elapsed ratio + # falls below 0.75 and cadence re-tightens. + authoritative = False + delay_ms = held_delay_ms + if not clamp_warned: + clamp_warned = True + logger.warning( + "Server appears to clamp lease lead; extension budget will deplete: id=%s", + reservation_id, + ) + else: + authoritative = False + delay_ms = min(max(grant / 2, 500.0), ttl_ms / 2) else: + code = _extract_error_code(response) + if response.status in (404, 410) or code in _PERMANENT_EXTEND_CODES: + logger.warning( + "Async stream heartbeat stopping permanently (%s, status=%d): id=%s", + code, + response.status, + reservation_id, + ) + return logger.warning("Async stream heartbeat failed: id=%s", reservation_id) + if authoritative: + rate_limited = response.status == 429 + if not rate_limited and 400 <= response.status < 500: + # Unrecoverable request/auth failure (spec): never + # rotate the key on an unchanged request. + logger.warning( + "Heartbeat stopping on client error (%s, status=%d): id=%s", + code, + response.status, + reservation_id, + ) + return + if not _lifecycle._extend_error_is_recoverable(response): + logger.warning( + "Heartbeat stopping on unexpected HTTP status %d: id=%s", + response.status, + reservation_id, + ) + return + retry_after = response.retry_after_ms_header if rate_limited else None + nxt = sched.on_transient_failure( + _lifecycle._now_mono_ms(), + rate_limited=rate_limited, + retry_after_ms=retry_after, + ) + if nxt is None: + logger.warning( + "Heartbeat stopping: no safe recovery window remains: id=%s", + reservation_id, + ) + return + delay_ms = nxt except Exception: logger.warning("Async stream heartbeat error: id=%s", reservation_id, exc_info=True) + if authoritative: + nxt = sched.on_transient_failure(_lifecycle._now_mono_ms()) + if nxt is None: + logger.warning( + "Heartbeat stopping: no safe recovery window remains: id=%s", + reservation_id, + ) + return + delay_ms = nxt except asyncio.CancelledError: return diff --git a/tests/test_decorator.py b/tests/test_decorator.py index 5e48e75..f0d5aaa 100644 --- a/tests/test_decorator.py +++ b/tests/test_decorator.py @@ -1,5 +1,7 @@ """Tests for the @cycles decorator.""" +import re + import pytest from runcycles.client import AsyncCyclesClient, CyclesClient @@ -14,6 +16,20 @@ def config() -> CyclesConfig: return CyclesConfig(base_url="http://localhost:7878", api_key="test-key", tenant="acme") +@pytest.fixture(autouse=True) +def _allow_heartbeat_extends(httpx_mock) -> None: # type: ignore[no-untyped-def] + # The v2.3 heartbeat primes an IMMEDIATE extend on entry; these tests + # run the real decorator flow, so accept (and ignore) heartbeat extends. + httpx_mock.add_response( + method="POST", + url=re.compile(r"http://localhost:7878/v1/reservations/[^/]+/extend$"), + json={"status": "ACTIVE"}, + status_code=200, + is_optional=True, + is_reusable=True, + ) + + class TestCyclesDecoratorSync: def test_basic_lifecycle(self, config: CyclesConfig, httpx_mock) -> None: # type: ignore[no-untyped-def] # Mock reservation creation diff --git a/tests/test_heartbeat.py b/tests/test_heartbeat.py new file mode 100644 index 0000000..3bb6158 --- /dev/null +++ b/tests/test_heartbeat.py @@ -0,0 +1,1666 @@ +"""Deterministic tests for the conservative-lead (v2.3) heartbeat and the +``actual_source: estimate`` audit marker.""" + +from __future__ import annotations + +import asyncio +import logging +import threading +import time +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from runcycles.config import CyclesConfig +from runcycles.exceptions import CyclesProtocolError +from runcycles.lifecycle import ( + AsyncCyclesLifecycle, + CyclesLifecycle, + DecoratorConfig, + _AuthoritativeScheduler, + _create_reservation_with_recovery, + _create_reservation_with_recovery_async, + _remaining_at_schedule_start, + _run_async_attempt, + _run_sync_attempt, + _schema_valid_create, + _schema_valid_extend, +) +from runcycles.models import Action, Amount, Subject, Unit +from runcycles.response import CyclesResponse +from runcycles.retry import AsyncCommitRetryEngine, CommitRetryEngine +from runcycles.streaming import AsyncStreamReservation, StreamReservation + +TTL = 60_000 +INITIAL_EXPIRY = 60_000 # server frame; arbitrary origin + + +class FakeClock: + def __init__(self) -> None: + self.t = 0.0 + + def now(self) -> float: + return self.t + + +def _config() -> CyclesConfig: + return CyclesConfig( + base_url="http://localhost:7878", + api_key="test-key", + tenant="acme", + retry_enabled=False, + ) + + +def _extend_ok( + expires_at_ms: int | None, + remaining_ttl_ms: int | None = None, +) -> CyclesResponse: + body: dict[str, Any] = {"status": "ACTIVE"} + if expires_at_ms is not None: + body["expires_at_ms"] = expires_at_ms + if remaining_ttl_ms is not None: + body["remaining_ttl_ms"] = remaining_ttl_ms + return CyclesResponse.success(200, body) + + +def _allow_response() -> CyclesResponse: + return CyclesResponse.success( + 200, + { + "decision": "ALLOW", + "reservation_id": "rsv_test", + "expires_at_ms": int(time.time() * 1000) + 600_000, + "affected_scopes": ["tenant:acme"], + "scope_path": "tenant:acme", + "reserved": {"unit": "USD_MICROCENTS", "amount": 1000}, + }, + ) + + +def _commit_success() -> CyclesResponse: + return CyclesResponse.success(200, {"status": "COMMITTED"}) + + +def _make_sync() -> tuple[CyclesLifecycle, MagicMock]: + client = MagicMock() + client._config = _config() + engine = MagicMock(spec=CommitRetryEngine) + return CyclesLifecycle(client, engine, {"tenant": "acme"}), client + + +def _make_async() -> tuple[AsyncCyclesLifecycle, AsyncMock]: + client = AsyncMock() + client._config = _config() + engine = MagicMock(spec=AsyncCommitRetryEngine) + return AsyncCyclesLifecycle(client, engine, {"tenant": "acme"}), client + + +def _ctx(expires_at_ms: int | None = INITIAL_EXPIRY) -> MagicMock: + ctx = MagicMock() + ctx.expires_at_ms = expires_at_ms + return ctx + + +def _run_sync_beats( + lifecycle: CyclesLifecycle, + clock: FakeClock, + monkeypatch: pytest.MonkeyPatch, + beats: int, + ttl: int = TTL, + ctx: MagicMock | None = None, + initial_remaining_ms: int | None = None, +) -> list[float]: + """Drive the sync heartbeat for `beats` iterations, advancing the fake + clock by the beat interval on every wait. Returns the wait timeouts.""" + monkeypatch.setattr("runcycles.lifecycle._now_mono_ms", clock.now) + timeouts: list[float] = [] + calls = {"n": 0} + + def wait(timeout: float | None = None) -> bool: + timeouts.append(timeout or 0.0) + calls["n"] += 1 + if calls["n"] > beats: + return True + clock.t += (timeout or 0.0) * 1000.0 + return False + + stop = threading.Event() + stop.wait = wait # type: ignore[method-assign] + thread = lifecycle._start_heartbeat( + "rsv_1", + ttl, + ctx or _ctx(), + stop, + initial_remaining_ms, + ) + assert thread is not None + thread.join(timeout=5) + assert not thread.is_alive() + return timeouts + + +class TestStrictLeaseAttemptContract: + def test_field_mode_retains_rtt_observed_before_field_appears(self) -> None: + scheduler = _AuthoritativeScheduler(7_000) + scheduler.observe_rtt(6_000) + + assert scheduler.on_valid_success(60_000, 100, 0) == 23_900 + + @pytest.mark.parametrize( + ("received_ms", "now_ms", "expected"), + [(1000.0, 3000.0, 58_000.0), (3000.0, 1000.0, 0.0), (None, 3000.0, 60_000.0)], + ) + def test_create_lead_deducts_post_receipt_setup_time( + self, + received_ms: float | None, + now_ms: float, + expected: float, + ) -> None: + assert _remaining_at_schedule_start(60_000, received_ms, now_ms) == expected + + def test_sync_create_recovers_once_with_same_body(self) -> None: + client = MagicMock() + client._config = _config() + body = {"idempotency_key": "same-key"} + client.create_reservation.side_effect = [ + CyclesResponse.success(202, _allow_response().body), + _allow_response(), + ] + + response, parsed, rtt_ms, received_ms = _create_reservation_with_recovery(client, body) + + assert response.status == 200 + assert parsed.reservation_id == "rsv_test" + assert rtt_ms >= 0 + assert received_ms >= rtt_ms + assert client.create_reservation.call_count == 2 + assert all(call.args[0] is body for call in client.create_reservation.call_args_list) + + def test_sync_create_rejects_malformed_200_after_one_recovery(self) -> None: + client = MagicMock() + client._config = _config() + malformed = CyclesResponse.success( + 200, + { + "decision": "ALLOW", + "reservation_id": "rsv_test", + "affected_scopes": ["tenant:acme"], + "unexpected": True, + }, + ) + client.create_reservation.return_value = malformed + + with pytest.raises(CyclesProtocolError, match="schema-valid HTTP 200"): + _create_reservation_with_recovery(client, {"idempotency_key": "same-key"}) + + assert client.create_reservation.call_count == 2 + + @pytest.mark.asyncio + async def test_async_create_recovers_transport_failure_with_same_body(self) -> None: + client = AsyncMock() + client._config = _config() + body = {"idempotency_key": "same-key"} + client.create_reservation.side_effect = [ConnectionError("lost"), _allow_response()] + + response, parsed, rtt_ms, received_ms = await _create_reservation_with_recovery_async(client, body) + + assert response.status == 200 + assert parsed.reservation_id == "rsv_test" + assert rtt_ms >= 0 + assert received_ms >= rtt_ms + assert client.create_reservation.await_count == 2 + assert all(call.args[0] is body for call in client.create_reservation.call_args_list) + + @pytest.mark.parametrize( + "response", + [ + CyclesResponse.success( + 202, + {"status": "ACTIVE", "expires_at_ms": 1, "remaining_ttl_ms": 0}, + ), + CyclesResponse.success(200, {"status": "ACTIVE"}), + CyclesResponse.success(200, {"status": "ACTIVE", "expires_at_ms": -1}), + CyclesResponse.success(200, {"status": "UNKNOWN", "expires_at_ms": 1}), + CyclesResponse.success(200, {"status": "ACTIVE", "expires_at_ms": 1, "balances": {}}), + CyclesResponse.success(200, {"status": "ACTIVE", "expires_at_ms": 1, "extra": True}), + ], + ) + def test_extend_rejects_non_200_or_non_schema_response( + self, + response: CyclesResponse, + ) -> None: + assert _schema_valid_extend(response) is None + + def test_strict_create_and_extend_accept_schema_valid_200(self) -> None: + assert _schema_valid_create(_allow_response()) is not None + parsed = _schema_valid_extend( + CyclesResponse.success( + 200, + {"status": "ACTIVE", "expires_at_ms": 1, "remaining_ttl_ms": 0, "balances": []}, + ), + ) + assert parsed is not None + assert parsed.remaining_ttl_ms == 0 + + @pytest.mark.parametrize( + "body", + [ + {"decision": "ALLOW", "affected_scopes": [], "caps": None}, + { + "decision": "ALLOW", + "affected_scopes": [], + "reserved": {"unit": "TOKENS", "amount": "1"}, + }, + { + "decision": "ALLOW", + "affected_scopes": [], + "balances": [ + { + "scope": "tenant:acme", + "scope_path": "tenant:acme", + "remaining": {"unit": "TOKENS", "amount": 1}, + "reserved": None, + } + ], + }, + { + "decision": "ALLOW", + "affected_scopes": [], + "cycles_evidence": { + "evidence_id": "a" * 64, + "cycles_evidence_url": "relative", + }, + }, + { + "decision": "ALLOW", + "affected_scopes": [], + "remaining_ttl_ms": 9_223_372_036_854_775_808, + }, + { + "decision": "ALLOW", + "affected_scopes": [], + "expires_at_ms": -1, + }, + ], + ) + def test_create_rejects_non_schema_optional_fields(self, body: dict[str, Any]) -> None: + assert _schema_valid_create(CyclesResponse.success(200, body)) is None + + def test_extend_rejects_nested_null_and_int64_overflow(self) -> None: + for body in ( + { + "status": "ACTIVE", + "expires_at_ms": 1, + "balances": [ + { + "scope": "tenant:acme", + "scope_path": "tenant:acme", + "remaining": {"unit": "TOKENS", "amount": 1}, + "debt": None, + } + ], + }, + { + "status": "ACTIVE", + "expires_at_ms": 9_223_372_036_854_775_808, + }, + ): + assert _schema_valid_extend(CyclesResponse.success(200, body)) is None + + def test_sync_outer_deadline_covers_complete_attempt(self) -> None: + with pytest.raises(TimeoutError, match="exceeded"): + _run_sync_attempt( + lambda: (time.sleep(0.05), _allow_response())[1], + timeout_budget_ms=1, + ) + + @pytest.mark.asyncio + async def test_async_outer_deadline_covers_complete_attempt(self) -> None: + async def slow() -> CyclesResponse: + await asyncio.sleep(0.05) + return _allow_response() + + with pytest.raises(asyncio.TimeoutError): + await _run_async_attempt(slow, timeout_budget_ms=1) + + +class TestSyncHeartbeatLeadEstimate: + def test_extends_only_when_lead_below_threshold( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # lead_min starts at 0 and the first beat fires IMMEDIATELY: beats + # 1-3 extend (grants measured at +ttl each), beat 4 skips once + # lead_min reaches 1.5*grant (180k-90k=90k), beat 5 extends again. + lifecycle, client = _make_sync() + client.extend_reservation.side_effect = [_extend_ok(INITIAL_EXPIRY + (n + 1) * TTL) for n in range(4)] + + timeouts = _run_sync_beats(lifecycle, FakeClock(), monkeypatch, beats=5) + + assert client.extend_reservation.call_count == 4 + assert timeouts[0] == 0.0 + assert timeouts[1] == min(TTL / 2, 30_000) / 1000.0 + + def test_interval_has_no_floor_for_small_ttl( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # ttl=1200 → after the immediate first beat, cadence must be 600ms + # (the old 1s floor guaranteed lapse in this spec-legal range). + lifecycle, client = _make_sync() + client.extend_reservation.return_value = _extend_ok(None) + + ctx = _ctx(1200) + timeouts = _run_sync_beats( + lifecycle, + FakeClock(), + monkeypatch, + beats=2, + ttl=1200, + ctx=ctx, + ) + + assert timeouts[0] == 0.0 + assert timeouts[1] == 0.6 + + def test_failed_extend_retries_with_same_idempotency_key( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # A lost/failed extend may have been applied server-side: the retry + # must reuse the same key so it cannot double-extend. After a + # success, a fresh key is used. + lifecycle, client = _make_sync() + client.extend_reservation.side_effect = [ + CyclesResponse.http_error(500, "boom"), + _extend_ok(INITIAL_EXPIRY + TTL), + _extend_ok(INITIAL_EXPIRY + 2 * TTL), + ] + + _run_sync_beats(lifecycle, FakeClock(), monkeypatch, beats=3) + + bodies = [c.args[1] for c in client.extend_reservation.call_args_list] + assert len(bodies) == 3 + assert bodies[0]["idempotency_key"] == bodies[1]["idempotency_key"] + assert bodies[2]["idempotency_key"] != bodies[0]["idempotency_key"] + + def test_permanent_code_stops_heartbeat( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + lifecycle, client = _make_sync() + client.extend_reservation.return_value = CyclesResponse.http_error( + 409, + "capped", + body={"error": "MAX_EXTENSIONS_EXCEEDED", "message": "m", "request_id": "r"}, + ) + + _run_sync_beats(lifecycle, FakeClock(), monkeypatch, beats=4) + + # One doomed call, then the loop self-terminates — no retry spam. + assert client.extend_reservation.call_count == 1 + + def test_clamped_grants_extend_every_beat( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # Server clamps each grant to ttl/4: the lead estimate sees the + # small grants (authoritative expires_at) and keeps extending. + lifecycle, client = _make_sync() + client.extend_reservation.side_effect = [_extend_ok(INITIAL_EXPIRY + (n + 1) * (TTL // 4)) for n in range(3)] + + timeouts = _run_sync_beats(lifecycle, FakeClock(), monkeypatch, beats=3) + + assert client.extend_reservation.call_count == 3 + # Cadence re-derived from the MEASURED grant (ttl/4 → beat at ttl/8). + assert timeouts[1] == (TTL / 4 / 2) / 1000.0 + + def test_grant_clamp_misclassification_after_skip_is_transient( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # After a skip, the next measured grant arrives across a doubled + # gap, so grant ≈ elapsed and the beat lands in the lead-clamp arm + # once. The classifier's lower band (0.75×elapsed) must make that + # non-sticky: at the held cadence the ratio falls to ~0.5, the + # regime reads normal again, and cadence re-tightens — without the + # band the hold sticks and a ttl/4-grant lease decays to a lapse. + lifecycle, client = _make_sync() + grant = TTL // 4 # 15000 → cadence ttl/8 = 7500ms + client.extend_reservation.side_effect = [_extend_ok(INITIAL_EXPIRY + (n + 1) * grant) for n in range(6)] + + timeouts = _run_sync_beats(lifecycle, FakeClock(), monkeypatch, beats=7) + + # b1-b3 extend @7.5s, b4 skips (lead 22.5k ≥ 1.5×15k), b5 extends + # across the doubled gap (misclassified → one 30s hold), b6 + # re-tightens to 7.5s, b7 extends on cadence. + assert client.extend_reservation.call_count == 6 + assert timeouts == [0.0, 7.5, 7.5, 7.5, 7.5, 30.0, 7.5, 7.5] + + def test_missing_expires_in_response_falls_back_to_plus_ttl( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + lifecycle, client = _make_sync() + client.extend_reservation.return_value = _extend_ok(None) + ctx = _ctx() + + _run_sync_beats(lifecycle, FakeClock(), monkeypatch, beats=3, ctx=ctx) + + # Fallback grant = requested ttl; lead builds from 0 so all three + # beats extend — and ctx is never updated without an expires value. + assert client.extend_reservation.call_count == 3 + ctx.update_expires_at_ms.assert_not_called() + + def test_unknown_initial_expiry_anchors_on_first_success( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + lifecycle, client = _make_sync() + client.extend_reservation.side_effect = [ + _extend_ok(500_000), # beat 1: fallback grant, sets frame + _extend_ok(500_000 + 3 * TTL), # beat 2: big measured grant + ] + + _run_sync_beats(lifecycle, FakeClock(), monkeypatch, beats=3, ctx=_ctx(None)) + + # Beat 1 (immediate) extends with the fallback grant (no prior + # frame); beat 2 measures a 3*ttl grant. Beat 3: lead_min = + # (ttl + 3*ttl) - 60s = 180s < 1.5*last_grant = 270s → NOT a skip, + # so a 3rd call happens; its StopIteration is swallowed as a + # transient error and the count stays meaningful at 3. + assert client.extend_reservation.call_count == 3 + + def test_tenant_closed_stops_heartbeat( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + lifecycle, client = _make_sync() + client.extend_reservation.return_value = CyclesResponse.http_error( + 409, + "closed", + body={"error": "TENANT_CLOSED", "message": "m", "request_id": "r"}, + ) + + _run_sync_beats(lifecycle, FakeClock(), monkeypatch, beats=4) + + assert client.extend_reservation.call_count == 1 + + def test_transient_failure_then_recovery( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # Non-permanent failures warn and keep the loop alive. + lifecycle, client = _make_sync() + client.extend_reservation.side_effect = [ + CyclesResponse.http_error(503, "unavailable"), + ConnectionError("network down"), + _extend_ok(INITIAL_EXPIRY + 3 * TTL), + ] + + _run_sync_beats(lifecycle, FakeClock(), monkeypatch, beats=3) + + assert client.extend_reservation.call_count == 3 + + +@pytest.mark.asyncio +class TestAsyncHeartbeatLeadEstimate: + async def _run( + self, + lifecycle: AsyncCyclesLifecycle, + beats: int, + monkeypatch: pytest.MonkeyPatch, + ctx: MagicMock | None = None, + ) -> None: + clock = FakeClock() + monkeypatch.setattr("runcycles.lifecycle._now_mono_ms", clock.now) + count = 0 + + async def fake_sleep(s: float) -> None: + nonlocal count + count += 1 + if count > beats: + raise asyncio.CancelledError + clock.t += s * 1000.0 + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + task = lifecycle._start_heartbeat("rsv_1", TTL, ctx or _ctx()) + assert task is not None + assert await task is None + + async def test_extends_only_when_lead_below_threshold( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + lifecycle, client = _make_async() + client.extend_reservation.side_effect = [_extend_ok(INITIAL_EXPIRY + (n + 1) * TTL) for n in range(4)] + + await self._run(lifecycle, 5, monkeypatch) + + # Immediate first beat, then lead_min builds from 0: beats 1-3 + # extend, beat 4 skips at lead_min >= 1.5*grant, beat 5 extends. + assert client.extend_reservation.await_count == 4 + + async def test_permanent_code_stops_heartbeat( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + lifecycle, client = _make_async() + client.extend_reservation.return_value = CyclesResponse.http_error(410, "gone") + + await self._run(lifecycle, 4, monkeypatch) + + assert client.extend_reservation.await_count == 1 + + async def test_transient_failure_missing_expires_and_late_anchor( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # Covers the async warn-continue, +=ttl fallback, and late-anchor branches. + lifecycle, client = _make_async() + client.extend_reservation.side_effect = [ + CyclesResponse.http_error(500, "boom"), + _extend_ok(None), # frame not yet anchored + _extend_ok(700_000), # late anchor + _extend_ok(None), # anchored: += ttl fallback + ] + + await self._run(lifecycle, 4, monkeypatch, ctx=_ctx(None)) + + assert client.extend_reservation.await_count == 4 + + +class TestStreamingHeartbeatLeadEstimate: + def test_sync_stream_lead_estimate_pattern( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + clock = FakeClock() + monkeypatch.setattr("runcycles.lifecycle._now_mono_ms", clock.now) + client = MagicMock() + client._config = _config() + client.extend_reservation.side_effect = [_extend_ok(INITIAL_EXPIRY + (n + 1) * TTL) for n in range(4)] + stream = StreamReservation( + client, + subject=Subject(tenant="acme"), + action=Action(kind="k", name="n"), + estimate=Amount(unit=Unit.USD_MICROCENTS, amount=1000), + ttl_ms=TTL, + ) + stream._reservation_id = "rsv_1" + stream._ctx = _ctx() + calls = {"n": 0} + + def wait(timeout: float | None = None) -> bool: + calls["n"] += 1 + if calls["n"] > 5: + return True + clock.t += (timeout or 0.0) * 1000.0 + return False + + stream._heartbeat_stop.wait = wait # type: ignore[method-assign] + thread = stream._start_heartbeat() + assert thread is not None + thread.join(timeout=5) + + assert client.extend_reservation.call_count == 4 + + def test_sync_stream_permanent_and_fallback_branches( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # Covers the sync-stream +=ttl fallback, late-anchor, transient-warn, + # and permanent-stop branches in one deterministic run. + clock = FakeClock() + monkeypatch.setattr("runcycles.lifecycle._now_mono_ms", clock.now) + client = MagicMock() + client._config = _config() + client.extend_reservation.side_effect = [ + CyclesResponse.http_error(500, "boom"), # transient: warn, retry + _extend_ok(None), # no anchor yet: no-op + _extend_ok(900_000), # late anchor + _extend_ok(None), # anchored: += ttl fallback + CyclesResponse.http_error(410, "gone"), # permanent: stop + ] + stream = StreamReservation( + client, + subject=Subject(tenant="acme"), + action=Action(kind="k", name="n"), + estimate=Amount(unit=Unit.USD_MICROCENTS, amount=1000), + ttl_ms=TTL, + ) + stream._reservation_id = "rsv_1" + stream._ctx = _ctx(None) + calls = {"n": 0} + + def wait(timeout: float | None = None) -> bool: + calls["n"] += 1 + if calls["n"] > 8: + return True + clock.t += (timeout or 0.0) * 1000.0 + return False + + stream._heartbeat_stop.wait = wait # type: ignore[method-assign] + thread = stream._start_heartbeat() + assert thread is not None + thread.join(timeout=5) + + # Stops at the 410 — remaining beats never call extend. + assert client.extend_reservation.call_count == 5 + + def test_sync_stream_field_mode_cycle( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # Authoritative mode in the sync stream heartbeat: create field + # drives the first delay; 503 and exception retries stay bounded. + clock = FakeClock() + monkeypatch.setattr("runcycles.lifecycle._now_mono_ms", clock.now) + client = MagicMock() + client._config = _config() + client.extend_reservation.side_effect = [ + _extend_ok(INITIAL_EXPIRY + TTL, remaining_ttl_ms=60_000), + CyclesResponse.http_error(503, "unavailable"), + ConnectionError("down"), + _extend_ok(INITIAL_EXPIRY + 2 * TTL, remaining_ttl_ms=60_000), + ] + stream = StreamReservation( + client, + subject=Subject(tenant="acme"), + action=Action(kind="k", name="n"), + estimate=Amount(unit=Unit.USD_MICROCENTS, amount=1000), + ttl_ms=TTL, + ) + stream._reservation_id = "rsv_1" + stream._ctx = _ctx() + stream._initial_remaining = 60_000 + calls = {"n": 0} + timeouts: list[float] = [] + + def wait(timeout: float | None = None) -> bool: + timeouts.append(timeout or 0.0) + calls["n"] += 1 + if calls["n"] > 4: + return True + clock.t += (timeout or 0.0) * 1000.0 + return False + + stream._heartbeat_stop.wait = wait # type: ignore[method-assign] + thread = stream._start_heartbeat() + assert thread is not None + thread.join(timeout=5) + + assert client.extend_reservation.call_count == 4 + assert timeouts[0] == 35.0 + assert timeouts[2] == 6.25 + assert timeouts[3] == 4.6875 + + @pytest.mark.asyncio + async def test_async_stream_field_mode_cycle( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + clock = FakeClock() + monkeypatch.setattr("runcycles.lifecycle._now_mono_ms", clock.now) + client = AsyncMock() + client._config = _config() + client.extend_reservation.side_effect = [ + _extend_ok(INITIAL_EXPIRY + TTL, remaining_ttl_ms=60_000), + CyclesResponse.http_error(503, "unavailable"), + ConnectionError("down"), + _extend_ok(INITIAL_EXPIRY + 2 * TTL, remaining_ttl_ms=60_000), + ] + stream = AsyncStreamReservation( + client, + subject=Subject(tenant="acme"), + action=Action(kind="k", name="n"), + estimate=Amount(unit=Unit.USD_MICROCENTS, amount=1000), + ttl_ms=TTL, + ) + stream._reservation_id = "rsv_1" + stream._ctx = _ctx() + stream._initial_remaining = 60_000 + count = 0 + sleeps: list[float] = [] + + async def fake_sleep(s: float) -> None: + nonlocal count + sleeps.append(s) + count += 1 + if count > 4: + raise asyncio.CancelledError + clock.t += s * 1000.0 + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + task = stream._start_heartbeat() + assert task is not None + assert await task is None + + assert client.extend_reservation.await_count == 4 + assert sleeps[0] == 35.0 + assert sleeps[2] == 6.25 + assert sleeps[3] == 4.6875 + + @pytest.mark.asyncio + async def test_async_stream_lead_estimate_pattern( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + clock = FakeClock() + monkeypatch.setattr("runcycles.lifecycle._now_mono_ms", clock.now) + client = AsyncMock() + client._config = _config() + client.extend_reservation.side_effect = [_extend_ok(INITIAL_EXPIRY + (n + 1) * TTL) for n in range(4)] + stream = AsyncStreamReservation( + client, + subject=Subject(tenant="acme"), + action=Action(kind="k", name="n"), + estimate=Amount(unit=Unit.USD_MICROCENTS, amount=1000), + ttl_ms=TTL, + ) + stream._reservation_id = "rsv_1" + stream._ctx = _ctx() + count = 0 + + async def fake_sleep(s: float) -> None: + nonlocal count + count += 1 + if count > 5: + raise asyncio.CancelledError + clock.t += s * 1000.0 + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + task = stream._start_heartbeat() + assert task is not None + assert await task is None + + assert client.extend_reservation.await_count == 4 + + @pytest.mark.asyncio + async def test_async_stream_permanent_and_fallback_branches( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + clock = FakeClock() + monkeypatch.setattr("runcycles.lifecycle._now_mono_ms", clock.now) + client = AsyncMock() + client._config = _config() + client.extend_reservation.side_effect = [ + CyclesResponse.http_error(500, "boom"), + _extend_ok(None), # no anchor yet: no-op + _extend_ok(900_000), # late anchor + _extend_ok(None), # anchored: += ttl fallback + CyclesResponse.http_error(410, "gone"), + ] + stream = AsyncStreamReservation( + client, + subject=Subject(tenant="acme"), + action=Action(kind="k", name="n"), + estimate=Amount(unit=Unit.USD_MICROCENTS, amount=1000), + ttl_ms=TTL, + ) + stream._reservation_id = "rsv_1" + stream._ctx = _ctx(None) + count = 0 + + async def fake_sleep(s: float) -> None: + nonlocal count + count += 1 + if count > 8: + raise asyncio.CancelledError + clock.t += s * 1000.0 + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + task = stream._start_heartbeat() + assert task is not None + assert await task is None + + assert client.extend_reservation.await_count == 5 + + +# --------------------------------------------------------------------------- +# Server-authoritative scheduling (remaining_ttl_ms, spec v0.1.25.16) +# --------------------------------------------------------------------------- + + +class TestAuthoritativeScheduling: + # With the SDK's enforced httpx timeouts (connect 2s + read 5s + write + # 5s), request_timeout_budget = 12000ms; at rtt 0: attempt_budget = + # 12000, safety_margin = 1000, retry_reserve = 2×12000 + 1000 = 25000. + RESERVE = 25_000.0 + + def test_create_remaining_drives_first_beat_and_steady_cadence( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # remaining=60000 → next_delay = 60000 − 25000 = 35000ms. Every + # extend echoes the field, so the cadence holds at 35s and the + # heuristic lead_min skip NEVER fires even though accumulated + # fallback grants would trip it. + lifecycle, client = _make_sync() + client.extend_reservation.return_value = _extend_ok( + INITIAL_EXPIRY + TTL, + remaining_ttl_ms=60_000, + ) + + timeouts = _run_sync_beats( + lifecycle, + FakeClock(), + monkeypatch, + beats=4, + initial_remaining_ms=60_000, + ) + + assert client.extend_reservation.call_count == 4 + assert timeouts == [35.0, 35.0, 35.0, 35.0, 35.0] + + def test_lease_below_reserve_one_immediate_attempt_then_stop( + self, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + ) -> None: + # 24h request capped to a 1s lease: the lease cannot hold the + # 25s retry reserve → next_delay 0 → ONE immediate fresh attempt is + # permitted; when its success also yields 0, the client MUST stop + # and surface (spec zero-delay guard) rather than tight-loop a + # maximum-lead server's extension budget away. + lifecycle, client = _make_sync() + client.extend_reservation.return_value = _extend_ok( + INITIAL_EXPIRY + TTL, + remaining_ttl_ms=1_000, + ) + + with caplog.at_level(logging.WARNING, logger="runcycles.lifecycle"): + timeouts = _run_sync_beats( + lifecycle, + FakeClock(), + monkeypatch, + beats=4, + ttl=86_400_000, + initial_remaining_ms=1_000, + ) + + # Immediate first beat (streak 1), its success hits streak 2 → stop. + assert timeouts[0] == 0.0 + assert client.extend_reservation.call_count == 1 + assert [r for r in caplog.records if "retry-safety budget" in r.message] + + def test_lead_clamp_server_with_field_no_warn_no_collapse( + self, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + ) -> None: + # A maximum-lead-clamping server that DOES emit remaining_ttl_ms: + # expiry echoes elapsed (grant ≈ elapsed, the heuristic's worst + # case) but the field carries the true 60s lead → authoritative arm + # schedules cleanly and the clamp warning never fires. + clock = FakeClock() + lifecycle, client = _make_sync() + + def clamped_extend(rid: str, body: dict[str, Any]) -> CyclesResponse: + return _extend_ok(INITIAL_EXPIRY + int(clock.t), remaining_ttl_ms=60_000) + + client.extend_reservation.side_effect = clamped_extend + + with caplog.at_level(logging.WARNING, logger="runcycles.lifecycle"): + timeouts = _run_sync_beats( + lifecycle, + clock, + monkeypatch, + beats=3, + initial_remaining_ms=60_000, + ) + + assert client.extend_reservation.call_count == 3 + assert timeouts == [35.0, 35.0, 35.0, 35.0] + assert not [r for r in caplog.records if "clamp lease lead" in r.message] + + def test_field_disappearing_mid_flight_resumes_heuristic( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # Beat 1's response lacks the field → the v2.3+band heuristic takes + # over seamlessly from its maintained bookkeeping. + lifecycle, client = _make_sync() + client.extend_reservation.side_effect = [ + _extend_ok(INITIAL_EXPIRY + TTL), # no field: fallback + _extend_ok(INITIAL_EXPIRY + 2 * TTL), + ] + + timeouts = _run_sync_beats( + lifecycle, + FakeClock(), + monkeypatch, + beats=2, + initial_remaining_ms=60_000, + ) + + assert client.extend_reservation.call_count == 2 + # First delay authoritative (35s); after the fieldless response the + # normal-regime cadence (ttl/2 = 30s) applies. + assert timeouts[0] == 35.0 + assert timeouts[1] == 30.0 + + def test_transient_failure_recovery_window_shrinks_then_same_key( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # 503 at the scheduled beat (t=35s): lead_est = 60000 − 35000 = + # 25000; retry_window = 25000 − 12000 − 1000 = 12000 → retry after + # min(30000, 25000/4, 12000) = 6250ms, with the SAME key. + lifecycle, client = _make_sync() + client.extend_reservation.side_effect = [ + CyclesResponse.http_error(503, "unavailable"), + _extend_ok(INITIAL_EXPIRY + TTL, remaining_ttl_ms=60_000), + ] + + timeouts = _run_sync_beats( + lifecycle, + FakeClock(), + monkeypatch, + beats=2, + initial_remaining_ms=60_000, + ) + + assert client.extend_reservation.call_count == 2 + assert timeouts[0] == 35.0 + assert timeouts[1] == 6.25 + bodies = [c.args[1] for c in client.extend_reservation.call_args_list] + assert bodies[0]["idempotency_key"] == bodies[1]["idempotency_key"] + + def test_repeated_failures_stop_when_no_window_progress( + self, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + ) -> None: + # Recovery repeats with the same key while the freshly recomputed + # retry_window shrinks: 6250 → 4687.5 → 1062.5 → 0 (one immediate + # retry) → no progress at 0 → MUST stop. Five attempts total. + lifecycle, client = _make_sync() + client.extend_reservation.return_value = CyclesResponse.http_error(503, "down") + + with caplog.at_level(logging.WARNING, logger="runcycles.lifecycle"): + timeouts = _run_sync_beats( + lifecycle, + FakeClock(), + monkeypatch, + beats=10, + initial_remaining_ms=60_000, + ) + + assert client.extend_reservation.call_count == 5 + assert timeouts == [35.0, 6.25, 4.6875, 1.0625, 0.0] + assert [r for r in caplog.records if "no safe recovery window" in r.message] + keys = {c.args[1]["idempotency_key"] for c in client.extend_reservation.call_args_list} + assert len(keys) == 1 # every recovery reused the same key + + def test_ambiguous_2xx_is_not_applied_same_key_recovery( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # A non-200 2xx in authoritative mode is ambiguous (spec): never + # scheduled from — recovered with the SAME idempotency key. + lifecycle, client = _make_sync() + client.extend_reservation.side_effect = [ + CyclesResponse.success(202, {"status": "ACTIVE", "remaining_ttl_ms": 60_000}), + _extend_ok(INITIAL_EXPIRY + 2 * TTL, remaining_ttl_ms=60_000), + ] + + timeouts = _run_sync_beats( + lifecycle, + FakeClock(), + monkeypatch, + beats=2, + initial_remaining_ms=60_000, + ) + + assert client.extend_reservation.call_count == 2 + assert timeouts[1] == 6.25 + bodies = [c.args[1] for c in client.extend_reservation.call_args_list] + assert bodies[0]["idempotency_key"] == bodies[1]["idempotency_key"] + + def test_429_honored_only_within_window( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # Retry-After 3s ≤ window 12000 → retried after exactly 3s, same key. + lifecycle, client = _make_sync() + client.extend_reservation.side_effect = [ + CyclesResponse.http_error(429, "limited", headers={"retry-after": "3"}), + _extend_ok(INITIAL_EXPIRY + TTL, remaining_ttl_ms=60_000), + ] + + timeouts = _run_sync_beats( + lifecycle, + FakeClock(), + monkeypatch, + beats=2, + initial_remaining_ms=60_000, + ) + + assert client.extend_reservation.call_count == 2 + assert timeouts[1] == 3.0 + bodies = [c.args[1] for c in client.extend_reservation.call_args_list] + assert bodies[0]["idempotency_key"] == bodies[1]["idempotency_key"] + + def test_429_missing_or_oversized_retry_after_stops( + self, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + ) -> None: + # Retry-After exceeding the retry window (20s > 12000ms) must not be + # re-invented earlier: stop and surface. + lifecycle, client = _make_sync() + client.extend_reservation.return_value = CyclesResponse.http_error( + 429, + "limited", + headers={"retry-after": "20"}, + ) + + with caplog.at_level(logging.WARNING, logger="runcycles.lifecycle"): + _run_sync_beats( + lifecycle, + FakeClock(), + monkeypatch, + beats=4, + initial_remaining_ms=60_000, + ) + + assert client.extend_reservation.call_count == 1 + assert [r for r in caplog.records if "no safe recovery window" in r.message] + + def test_other_4xx_stops_without_key_rotation( + self, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + ) -> None: + lifecycle, client = _make_sync() + client.extend_reservation.return_value = CyclesResponse.http_error( + 400, + "bad", + body={"error": "INVALID_REQUEST", "message": "m", "request_id": "r"}, + ) + + with caplog.at_level(logging.WARNING, logger="runcycles.lifecycle"): + _run_sync_beats( + lifecycle, + FakeClock(), + monkeypatch, + beats=4, + initial_remaining_ms=60_000, + ) + + assert client.extend_reservation.call_count == 1 + assert [r for r in caplog.records if "client error" in r.message] + + def test_unexpected_3xx_stops_without_retry( + self, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + ) -> None: + lifecycle, client = _make_sync() + client.extend_reservation.return_value = CyclesResponse.http_error(302, "redirect") + + with caplog.at_level(logging.WARNING, logger="runcycles.lifecycle"): + _run_sync_beats( + lifecycle, + FakeClock(), + monkeypatch, + beats=4, + initial_remaining_ms=60_000, + ) + + assert client.extend_reservation.call_count == 1 + assert [r for r in caplog.records if "unexpected HTTP status" in r.message] + + @pytest.mark.asyncio + async def test_async_field_mode_full_cycle( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # Covers the async authoritative arms: initial field delay, 503 + # recovery, exception recovery with a shrinking window, success. + clock = FakeClock() + monkeypatch.setattr("runcycles.lifecycle._now_mono_ms", clock.now) + lifecycle, client = _make_async() + client.extend_reservation.side_effect = [ + _extend_ok(INITIAL_EXPIRY + TTL, remaining_ttl_ms=60_000), + CyclesResponse.http_error(503, "unavailable"), + ConnectionError("down"), + _extend_ok(INITIAL_EXPIRY + 2 * TTL, remaining_ttl_ms=60_000), + ] + count = 0 + sleeps: list[float] = [] + + async def fake_sleep(s: float) -> None: + nonlocal count + sleeps.append(s) + count += 1 + if count > 4: + raise asyncio.CancelledError + clock.t += s * 1000.0 + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + task = lifecycle._start_heartbeat("rsv_1", TTL, _ctx(), 60_000) + assert task is not None + assert await task is None + + assert client.extend_reservation.await_count == 4 + assert sleeps[0] == 35.0 # from the create field + assert sleeps[2] == 6.25 # 503: window 12000, lead_est/4 = 6250 + assert sleeps[3] == 4.6875 # exception: window shrank to 5750 + + def test_scheduler_edge_cases_direct(self) -> None: + from runcycles.lifecycle import _AuthoritativeScheduler + + sched = _AuthoritativeScheduler(12_000.0) + # No schema-valid response yet → lead estimate is 0 and any failure + # has no recovery window. + assert sched.lead_estimate_ms(1_000.0) == 0.0 + assert sched.on_transient_failure(1_000.0) is None + # Establish a lead, then: missing and negative Retry-After stop. + assert sched.on_valid_success(60_000, 0.0, 0.0) == 35_000.0 + assert sched.on_transient_failure(35_000.0, rate_limited=True, retry_after_ms=None) is None + sched2 = _AuthoritativeScheduler(12_000.0) + sched2.on_valid_success(60_000, 0.0, 0.0) + assert sched2.on_transient_failure(35_000.0, rate_limited=True, retry_after_ms=-1) is None + # Unknown/backward timing never fabricates lease lead. + sched3 = _AuthoritativeScheduler(12_000.0) + assert sched3.on_valid_success(60_000, -1.0, 0.0) == 0.0 + assert sched3.lead_estimate_ms(-1.0) == 0.0 + + def test_sync_repeated_ambiguous_2xx_stops( + self, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + ) -> None: + lifecycle, client = _make_sync() + client.extend_reservation.return_value = CyclesResponse.success(202, {"status": "ACTIVE"}) + + with caplog.at_level(logging.WARNING, logger="runcycles.lifecycle"): + _run_sync_beats( + lifecycle, + FakeClock(), + monkeypatch, + beats=10, + initial_remaining_ms=60_000, + ) + + assert client.extend_reservation.call_count == 5 + assert [r for r in caplog.records if "no safe recovery window" in r.message] + + def test_sync_repeated_exceptions_stop( + self, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + ) -> None: + lifecycle, client = _make_sync() + client.extend_reservation.side_effect = ConnectionError("down") + + with caplog.at_level(logging.WARNING, logger="runcycles.lifecycle"): + _run_sync_beats( + lifecycle, + FakeClock(), + monkeypatch, + beats=10, + initial_remaining_ms=60_000, + ) + + assert client.extend_reservation.call_count == 5 + assert [r for r in caplog.records if "no safe recovery window" in r.message] + + def test_create_response_model_parses_remaining(self) -> None: + from runcycles.models import ReservationCreateResponse + + body = { + "decision": "ALLOW", + "reservation_id": "rsv_1", + "affected_scopes": ["tenant:acme"], + "expires_at_ms": 1_000_000, + "remaining_ttl_ms": 10_000, + } + parsed = ReservationCreateResponse.model_validate(body) + assert parsed.remaining_ttl_ms == 10_000 + body.pop("remaining_ttl_ms") + assert ReservationCreateResponse.model_validate(body).remaining_ttl_ms is None + + +# Scenario matrix for the authoritative stop/recovery arms, exercised +# against every loop variant (async lifecycle, sync stream, async stream). +# Each entry: (responses factory, initial_remaining, expected extend calls). +# - repeated 503 / ambiguous 202 / exceptions: recovery windows shrink +# 6250 → 4687.5 → 1062.5 → 0 → no-progress stop = 5 attempts; +# - 400: unrecoverable client error, stop after 1; +# - tiny lease: zero-delay guard, 1 immediate attempt then stop; +# - 429 with Retry-After 3s inside the window: honored, then success. +_STOP_MATRIX = [ + ("s503", lambda: CyclesResponse.http_error(503, "down"), 60_000, 5), + ("s202", lambda: CyclesResponse.success(202, {"status": "ACTIVE"}), 60_000, 5), + ( + "s400", + lambda: CyclesResponse.http_error( + 400, + "bad", + body={"error": "INVALID_REQUEST", "message": "m", "request_id": "r"}, + ), + 60_000, + 1, + ), + ("s302", lambda: CyclesResponse.http_error(302, "redirect"), 60_000, 1), + ("szero", lambda: _extend_ok(INITIAL_EXPIRY + TTL, remaining_ttl_ms=1_000), 1_000, 1), + ("sexc", lambda: ConnectionError("down"), 60_000, 5), +] + + +def _matrix_side_effect(factory: Any) -> Any: + def _effect(*args: Any, **kwargs: Any) -> CyclesResponse: + result = factory() + if isinstance(result, Exception): + raise result + return result + + return _effect + + +class TestAuthoritativeStopMatrix: + @pytest.mark.parametrize( + ("name", "factory", "initial", "expected"), + _STOP_MATRIX, + ids=[m[0] for m in _STOP_MATRIX], + ) + @pytest.mark.asyncio + async def test_async_lifecycle_arms( + self, + name: str, + factory: Any, + initial: int, + expected: int, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + clock = FakeClock() + monkeypatch.setattr("runcycles.lifecycle._now_mono_ms", clock.now) + lifecycle, client = _make_async() + client.extend_reservation.side_effect = _matrix_side_effect(factory) + count = 0 + + async def fake_sleep(s: float) -> None: + nonlocal count + count += 1 + if count > 10: + raise asyncio.CancelledError + clock.t += s * 1000.0 + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + task = lifecycle._start_heartbeat("rsv_1", TTL, _ctx(), initial) + assert task is not None + assert await task is None + + assert client.extend_reservation.await_count == expected + + @pytest.mark.asyncio + async def test_async_lifecycle_429_honored( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + clock = FakeClock() + monkeypatch.setattr("runcycles.lifecycle._now_mono_ms", clock.now) + lifecycle, client = _make_async() + client.extend_reservation.side_effect = [ + CyclesResponse.http_error(429, "limited", headers={"retry-after": "3"}), + _extend_ok(INITIAL_EXPIRY + TTL, remaining_ttl_ms=60_000), + ] + count = 0 + sleeps: list[float] = [] + + async def fake_sleep(s: float) -> None: + nonlocal count + sleeps.append(s) + count += 1 + if count > 2: + raise asyncio.CancelledError + clock.t += s * 1000.0 + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + task = lifecycle._start_heartbeat("rsv_1", TTL, _ctx(), 60_000) + assert task is not None + assert await task is None + + assert client.extend_reservation.await_count == 2 + assert sleeps[1] == 3.0 + + def _make_stream(self) -> tuple[StreamReservation, MagicMock]: + client = MagicMock() + client._config = _config() + stream = StreamReservation( + client, + subject=Subject(tenant="acme"), + action=Action(kind="k", name="n"), + estimate=Amount(unit=Unit.USD_MICROCENTS, amount=1000), + ttl_ms=TTL, + ) + stream._reservation_id = "rsv_1" + stream._ctx = _ctx() + return stream, client + + @pytest.mark.parametrize( + ("name", "factory", "initial", "expected"), + _STOP_MATRIX, + ids=[m[0] for m in _STOP_MATRIX], + ) + def test_sync_stream_arms( + self, + name: str, + factory: Any, + initial: int, + expected: int, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + clock = FakeClock() + monkeypatch.setattr("runcycles.lifecycle._now_mono_ms", clock.now) + stream, client = self._make_stream() + client.extend_reservation.side_effect = _matrix_side_effect(factory) + stream._initial_remaining = initial + calls = {"n": 0} + + def wait(timeout: float | None = None) -> bool: + calls["n"] += 1 + if calls["n"] > 10: + return True + clock.t += (timeout or 0.0) * 1000.0 + return False + + stream._heartbeat_stop.wait = wait # type: ignore[method-assign] + thread = stream._start_heartbeat() + assert thread is not None + thread.join(timeout=5) + + assert client.extend_reservation.call_count == expected + + def test_sync_stream_429_honored(self, monkeypatch: pytest.MonkeyPatch) -> None: + clock = FakeClock() + monkeypatch.setattr("runcycles.lifecycle._now_mono_ms", clock.now) + stream, client = self._make_stream() + client.extend_reservation.side_effect = [ + CyclesResponse.http_error(429, "limited", headers={"retry-after": "3"}), + _extend_ok(INITIAL_EXPIRY + TTL, remaining_ttl_ms=60_000), + ] + stream._initial_remaining = 60_000 + timeouts: list[float] = [] + calls = {"n": 0} + + def wait(timeout: float | None = None) -> bool: + timeouts.append(timeout or 0.0) + calls["n"] += 1 + if calls["n"] > 2: + return True + clock.t += (timeout or 0.0) * 1000.0 + return False + + stream._heartbeat_stop.wait = wait # type: ignore[method-assign] + thread = stream._start_heartbeat() + assert thread is not None + thread.join(timeout=5) + + assert client.extend_reservation.call_count == 2 + assert timeouts[1] == 3.0 + + def _make_async_stream(self) -> tuple[AsyncStreamReservation, AsyncMock]: + client = AsyncMock() + client._config = _config() + stream = AsyncStreamReservation( + client, + subject=Subject(tenant="acme"), + action=Action(kind="k", name="n"), + estimate=Amount(unit=Unit.USD_MICROCENTS, amount=1000), + ttl_ms=TTL, + ) + stream._reservation_id = "rsv_1" + stream._ctx = _ctx() + return stream, client + + @pytest.mark.parametrize( + ("name", "factory", "initial", "expected"), + _STOP_MATRIX, + ids=[m[0] for m in _STOP_MATRIX], + ) + @pytest.mark.asyncio + async def test_async_stream_arms( + self, + name: str, + factory: Any, + initial: int, + expected: int, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + clock = FakeClock() + monkeypatch.setattr("runcycles.lifecycle._now_mono_ms", clock.now) + stream, client = self._make_async_stream() + client.extend_reservation.side_effect = _matrix_side_effect(factory) + stream._initial_remaining = initial + count = 0 + + async def fake_sleep(s: float) -> None: + nonlocal count + count += 1 + if count > 10: + raise asyncio.CancelledError + clock.t += s * 1000.0 + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + task = stream._start_heartbeat() + assert task is not None + assert await task is None + + assert client.extend_reservation.await_count == expected + + @pytest.mark.asyncio + async def test_async_stream_429_honored( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + clock = FakeClock() + monkeypatch.setattr("runcycles.lifecycle._now_mono_ms", clock.now) + stream, client = self._make_async_stream() + client.extend_reservation.side_effect = [ + CyclesResponse.http_error(429, "limited", headers={"retry-after": "3"}), + _extend_ok(INITIAL_EXPIRY + TTL, remaining_ttl_ms=60_000), + ] + stream._initial_remaining = 60_000 + count = 0 + sleeps: list[float] = [] + + async def fake_sleep(s: float) -> None: + nonlocal count + sleeps.append(s) + count += 1 + if count > 2: + raise asyncio.CancelledError + clock.t += s * 1000.0 + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + task = stream._start_heartbeat() + assert task is not None + assert await task is None + + assert client.extend_reservation.await_count == 2 + assert sleeps[1] == 3.0 + + +# --------------------------------------------------------------------------- +# Date header accessor (kept as a general response accessor; the heartbeat +# no longer consumes it — RFC 9110 §6.6.1 makes it a different clock). +# --------------------------------------------------------------------------- + + +class TestServerDateAccessor: + def test_server_date_ms_parses_http_date(self) -> None: + response = CyclesResponse.success( + 200, + {}, + headers={"date": "Mon, 27 Jul 2026 12:00:00 GMT"}, + ) + assert response.server_date_ms == 1785153600000 + + def test_server_date_ms_absent_or_garbage(self) -> None: + assert CyclesResponse.success(200, {}).server_date_ms is None + assert CyclesResponse.success(200, {}, headers={"date": "not a date"}).server_date_ms is None + + +# --------------------------------------------------------------------------- +# actual_source marker +# --------------------------------------------------------------------------- + + +def _cfg(**kwargs: Any) -> DecoratorConfig: + defaults: dict[str, Any] = {"estimate": 1000, "tenant": "acme", "ttl_ms": 60_000} + defaults.update(kwargs) + return DecoratorConfig(**defaults) + + +class TestFirstBeatAndRegimes: + def test_first_beat_is_immediate_even_for_huge_ttl( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # A 24h request silently capped to a small lease by tenant policy + # must still survive: only a zero first delay guarantees the first + # extension lands before ANY possible capped lease expires. + lifecycle, client = _make_sync() + client.extend_reservation.return_value = _extend_ok(None) + timeouts = _run_sync_beats( + lifecycle, + FakeClock(), + monkeypatch, + beats=1, + ttl=86_400_000, + ) + assert timeouts[0] == 0.0 + + def test_first_beat_failure_does_not_hot_loop( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # A transient failure on the primed (delay-0) beat must back off to + # the held cadence, not spin at 0ms against a down server. + lifecycle, client = _make_sync() + client.extend_reservation.return_value = CyclesResponse.http_error(503, "down") + timeouts = _run_sync_beats(lifecycle, FakeClock(), monkeypatch, beats=2) + assert timeouts[0] == 0.0 + assert timeouts[1] == 30.0 + assert client.extend_reservation.call_count == 2 + + def test_lead_clamp_regime_holds_cadence_and_warns_once( + self, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + ) -> None: + # Maximum-lead clamping: every grant merely mirrors elapsed time + # (expires_at stays ~now + cap), so grant/2 cadence would collapse + # to the floor and burn max_extensions in seconds. The loop must + # hold the bounded cadence and warn exactly once. + clock = FakeClock() + lifecycle, client = _make_sync() + + def clamped_extend(rid: str, body: dict[str, Any]) -> CyclesResponse: + return _extend_ok(INITIAL_EXPIRY + int(clock.t)) + + client.extend_reservation.side_effect = clamped_extend + + with caplog.at_level(logging.WARNING, logger="runcycles.lifecycle"): + timeouts = _run_sync_beats(lifecycle, clock, monkeypatch, beats=3) + + # Beat 1 measures grant 0 (prime), beats 2-3 measure grant == + # elapsed: all extend, cadence never tightens below the held delay. + assert client.extend_reservation.call_count == 3 + assert timeouts == [0.0, 30.0, 30.0, 30.0] + clamp_warnings = [r for r in caplog.records if "clamp lease lead" in r.message] + assert len(clamp_warnings) == 1 + + +class TestActualSourceMarker: + def test_fallback_commit_carries_marker(self) -> None: + lifecycle, client = _make_sync() + client.create_reservation.return_value = _allow_response() + client.commit_reservation.return_value = _commit_success() + + lifecycle.execute(lambda: "result", (), {}, _cfg()) # no actual expression + + body = client.commit_reservation.call_args.args[1] + assert body["metadata"]["actual_source"] == "estimate" + + def test_measured_commit_has_no_marker(self) -> None: + lifecycle, client = _make_sync() + client.create_reservation.return_value = _allow_response() + client.commit_reservation.return_value = _commit_success() + + lifecycle.execute(lambda: "result", (), {}, _cfg(actual=lambda _r: 900)) + + body = client.commit_reservation.call_args.args[1] + assert "metadata" not in body or "actual_source" not in body.get("metadata", {}) + + @pytest.mark.asyncio + async def test_async_fallback_commit_carries_marker(self) -> None: + lifecycle, client = _make_async() + client.create_reservation.return_value = _allow_response() + client.commit_reservation.return_value = _commit_success() + + async def fn() -> str: + return "result" + + await lifecycle.execute(fn, (), {}, _cfg()) + + body = client.commit_reservation.call_args.args[1] + assert body["metadata"]["actual_source"] == "estimate" + + def test_stream_fallback_carries_marker_and_measured_does_not(self) -> None: + client = MagicMock() + client._config = _config() + client.create_reservation.return_value = _allow_response() + client.commit_reservation.return_value = _commit_success() + stream = StreamReservation( + client, + subject=Subject(tenant="acme"), + action=Action(kind="k", name="n"), + estimate=Amount(unit=Unit.USD_MICROCENTS, amount=1000), + ttl_ms=60_000, + ) + with stream: + pass # no actual cost recorded → estimate fallback + body = client.commit_reservation.call_args.args[1] + assert body["metadata"]["actual_source"] == "estimate" + + client2 = MagicMock() + client2._config = _config() + client2.create_reservation.return_value = _allow_response() + client2.commit_reservation.return_value = _commit_success() + stream2 = StreamReservation( + client2, + subject=Subject(tenant="acme"), + action=Action(kind="k", name="n"), + estimate=Amount(unit=Unit.USD_MICROCENTS, amount=1000), + ttl_ms=60_000, + ) + with stream2: + stream2.usage.set_actual_cost(700) + body2 = client2.commit_reservation.call_args.args[1] + assert "metadata" not in body2 or "actual_source" not in body2.get("metadata", {}) + + @pytest.mark.asyncio + async def test_async_stream_fallback_carries_marker(self) -> None: + client = AsyncMock() + client._config = _config() + client.create_reservation.return_value = _allow_response() + client.commit_reservation.return_value = _commit_success() + stream = AsyncStreamReservation( + client, + subject=Subject(tenant="acme"), + action=Action(kind="k", name="n"), + estimate=Amount(unit=Unit.USD_MICROCENTS, amount=1000), + ttl_ms=60_000, + ) + async with stream: + pass + body = client.commit_reservation.call_args.args[1] + assert body["metadata"]["actual_source"] == "estimate" diff --git a/tests/test_response.py b/tests/test_response.py index 3f55a3d..f6851c8 100644 --- a/tests/test_response.py +++ b/tests/test_response.py @@ -76,10 +76,43 @@ def test_retry_after_header_seconds_to_ms(self) -> None: assert resp.retry_after_ms_header == 3000 def test_retry_after_header_non_numeric_ignored(self) -> None: + for value in ( + "Wed, 21 Oct 2026 07:28:00 GMT", + "-1", + "+1", + "1e2", + "1.5", + "9223372036854775808", + ): + resp = CyclesResponse.http_error( + 429, + "Rate limited", + headers={"retry-after": value}, + ) + assert resp.retry_after_ms_header is None + + def test_retry_after_header_allows_ows(self) -> None: resp = CyclesResponse.http_error( - 429, "Rate limited", headers={"retry-after": "Wed, 21 Oct 2026 07:28:00 GMT"}, + 429, + "Rate limited", + headers={"retry-after": " 0 "}, ) - assert resp.retry_after_ms_header is None + assert resp.retry_after_ms_header == 0 + + def test_retry_after_header_checks_millisecond_overflow(self) -> None: + max_seconds = 9_223_372_036_854_775_807 // 1000 + accepted = CyclesResponse.http_error( + 429, + "Rate limited", + headers={"retry-after": str(max_seconds)}, + ) + overflow = CyclesResponse.http_error( + 429, + "Rate limited", + headers={"retry-after": str(max_seconds + 1)}, + ) + assert accepted.retry_after_ms_header == max_seconds * 1000 + assert overflow.retry_after_ms_header is None def test_missing_headers_return_none(self) -> None: resp = CyclesResponse.success(200, {}) @@ -95,7 +128,5 @@ def test_transport_error_no_headers(self) -> None: assert resp.headers == {} def test_cycles_tenant_header(self) -> None: - resp = CyclesResponse.success( - 200, {}, headers={"x-cycles-tenant": "acme"} - ) + resp = CyclesResponse.success(200, {}, headers={"x-cycles-tenant": "acme"}) assert resp.cycles_tenant == "acme" diff --git a/tests/test_streaming.py b/tests/test_streaming.py index b1ad873..aede0cd 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -229,7 +229,7 @@ def test_subject_with_no_standard_fields_raises(self) -> None: class TestResolveActualCost: def test_explicit_actual_cost(self) -> None: u = StreamUsage(actual_cost=777) - assert _resolve_actual_cost(u, lambda _: 999, 1000) == 777 + assert _resolve_actual_cost(u, lambda _: 999, 1000) == (777, False) def test_cost_fn(self) -> None: u = StreamUsage(tokens_input=100, tokens_output=50) @@ -237,7 +237,7 @@ def test_cost_fn(self) -> None: def cost_fn(usage: StreamUsage) -> int: return usage.tokens_input * 2 + usage.tokens_output * 3 - assert _resolve_actual_cost(u, cost_fn, 1000) == 350 + assert _resolve_actual_cost(u, cost_fn, 1000) == (350, False) def test_cost_fn_error_falls_back_to_estimate(self) -> None: u = StreamUsage() @@ -245,11 +245,11 @@ def test_cost_fn_error_falls_back_to_estimate(self) -> None: def bad_fn(_: StreamUsage) -> int: raise ValueError("oops") - assert _resolve_actual_cost(u, bad_fn, 1000) == 1000 + assert _resolve_actual_cost(u, bad_fn, 1000) == (1000, True) def test_fallback_to_estimate(self) -> None: u = StreamUsage() - assert _resolve_actual_cost(u, None, 500) == 500 + assert _resolve_actual_cost(u, None, 500) == (500, True) # --------------------------------------------------------------------------- @@ -620,7 +620,8 @@ def test_metadata_passed_to_commit(self) -> None: pass commit_body = mock.commit_reservation.call_args[0][1] - assert commit_body["metadata"] == {"source": "test"} + # actual_source marker added because no actual cost was recorded + assert commit_body["metadata"] == {"source": "test", "actual_source": "estimate"} def test_metrics_include_tokens(self) -> None: mock = _make_mock_client() @@ -1310,7 +1311,8 @@ async def test_metadata_passed_to_commit(self) -> None: pass commit_body = mock.commit_reservation.call_args[0][1] - assert commit_body["metadata"] == {"key": "val"} + # actual_source marker added because no actual cost was recorded + assert commit_body["metadata"] == {"key": "val", "actual_source": "estimate"} @pytest.mark.asyncio async def test_caps_propagated(self) -> None: