diff --git a/AUDIT.md b/AUDIT.md index 63cb0e1..423d652 100644 --- a/AUDIT.md +++ b/AUDIT.md @@ -1,6 +1,7 @@ # Cycles Protocol v0.1.25 — Client (Python) Audit -**Date:** 2026-07-10 (unreleased — `TENANT_CLOSED` + `LIMIT_EXCEEDED` error-code support. `TENANT_CLOSED` per runtime spec v0.1.25.13 (`cycles-protocol-v0.yaml`, runcycles/cycles-protocol#125): `ErrorCode.TENANT_CLOSED` enum member, `TenantClosedError` subclass wired into the lifecycle error-code→exception mapping (reservation-creation surfaces), `CyclesProtocolError.is_tenant_closed()` helper. `LIMIT_EXCEEDED` per runtime spec v0.1.25.12 (revision 2026-07-04, HTTP 429 rate limiting): enum-only member matching the `BUDGET_FROZEN`/`BUDGET_CLOSED` pattern, classified retryable at both the enum and exception layers (429 is transient; previously it fell through to `UNKNOWN`, which happened to be retryable, so semantics are unchanged — now typed). Enum reordered to mirror spec declaration order. Both purely additive; previously both codes fell through the `ErrorCode.from_string` forward-compat path to `UNKNOWN`. See the dated entries at the end of this file. 398 tests pass at 100% coverage.), +**Date:** 2026-07-27 (v0.5.0 — durable commit retries: on-disk pending-commit journal with next-run replay, bounded atexit flush, and `POST /v1/events` recovery for commits that land after reservation expiry; async retry-task GC fix; `retry_enabled=False` now journals instead of silently dropping. Review hardening: per-identity journal partitioning (tenant-keyed when configured — rotation-safe), 429 transient with `Retry-After` incl. first-attempt commits (no more release-on-429), 401/403 retained, `0700`/`0600` journal permissions, process-wide flush deadline; round 3: first-attempt 401/403 journaled (not released), Retry-After persisted across restarts, PBKDF2 30k rounds, unique journal temp files. See the dated entry below. 491 tests pass at 100% coverage.), +2026-07-10 (v0.5.0 — `TENANT_CLOSED` + `LIMIT_EXCEEDED` error-code support. `TENANT_CLOSED` per runtime spec v0.1.25.13 (`cycles-protocol-v0.yaml`, runcycles/cycles-protocol#125): `ErrorCode.TENANT_CLOSED` enum member, `TenantClosedError` subclass wired into the lifecycle error-code→exception mapping (reservation-creation surfaces), `CyclesProtocolError.is_tenant_closed()` helper. `LIMIT_EXCEEDED` per runtime spec v0.1.25.12 (revision 2026-07-04, HTTP 429 rate limiting): enum-only member matching the `BUDGET_FROZEN`/`BUDGET_CLOSED` pattern, classified retryable at both the enum and exception layers (429 is transient; previously it fell through to `UNKNOWN`, which happened to be retryable, so semantics are unchanged — now typed). Enum reordered to mirror spec declaration order. Both purely additive; previously both codes fell through the `ErrorCode.from_string` forward-compat path to `UNKNOWN`. See the dated entries at the end of this file. 398 tests pass at 100% coverage.), 2026-07-09 (README + docstring transport-error documentation fix, no version bump — see the dated entry at the end of this file. `CyclesTransportError` is exported but never raised by the SDK; README and its docstring now describe the actual `status == -1` surfacing.), 2026-07-03 (integration-test-only, no version bump — `test_health_check` now probes the public `/actuator/health/readiness` endpoint instead of aggregate `/actuator/health`, which requires `X-Admin-API-Key` since cycles-server v0.1.25.45 and fails closed with 500 when the server has no admin key configured. The old assertion had failed the org nightly Full-Stack Integration every night since 2026-06-28. No library code change.), 2026-05-22 (v0.4.3 — `expires_from`/`expires_to` and `finalized_from`/`finalized_to` ISO-8601 window-filter passthrough on `list_reservations` per `cycles-protocol-v0.yaml` revision 2026-05-22; closes the Python-client side of runcycles/cycles-server#162. No code change — `**query_params` already forwards arbitrary kwargs. Added sync + async regression tests; unlike `from`/`to` the new param names are plain kwargs (no Python-reserved-word workaround needed). 393 tests pass at 100% coverage.), @@ -12,6 +13,43 @@ --- +## 2026-07-27 — Durable commit retries (journal + /v1/events fallback) + +Pending commits no longer exist only in memory: the retry engines journal +each one to disk (`~/.runcycles/commit-journal`, config/env overridable) +before retrying, replay survivors on the next run, and flush bounded at +interpreter exit. A commit answered `RESERVATION_EXPIRED` — where the server +has already returned the reserved budget to the pool — is recovered via +`POST /v1/events` (spec-conformant `EventCreateRequest`, commit idempotency +key reused, recovery markers in `metadata`). Also fixes the async engine's +unreferenced-task GC hazard and the silent drop under `retry_enabled=False`. +Post-review (PR #89) hardening, round 1: journal records are partitioned +into per-identity subdirectories so co-located clients with different +credentials never replay or 401-discard each other's records and replay +claims cannot cross identities; HTTP 429 / `LIMIT_EXCEEDED` on retried +attempts is transient (record retained, `Retry-After` honored) instead of +a terminal discard; the atexit flush enforces one process-wide +`retry_flush_timeout` deadline instead of per-engine. Round 2: a +rate-limited *first* commit attempt now schedules a retry in all four +lifecycle variants instead of releasing the reservation (a release +returned budget for spend that already happened); the identity fingerprint +uses the configured tenant when set, so API-key rotation no longer orphans +pending records, and 401/403 retains the journal entry instead of +discarding it; journal directories/files are created `0700`/`0600` where +supported. Round 3: first-attempt 401/403 also journals instead of +releasing (same class as the 429 gap, all four variants); the `Retry-After` +floor is persisted as an absolute `not_before_ms` and restored on replay; +PBKDF2 rounds reduced 600k → 30k (~20 ms cold, cache 256 — input is a +high-entropy machine credential, rounds only defend the weak-key +fallback); journal temp files get unique per-writer names so concurrent +processes cannot publish each other's partial writes. Fleet self-review +round: ASCII-explicit sanitizer + pinned cross-SDK fingerprint vectors +(interop with TS/Java identity dirs), blank-tenant normalization, 1-hour +clamp on honored Retry-After and restored floors, status-410 expired +trigger, unclassifiable-4xx retention (never release/discard on codeless +or unknown-code responses), base-dir permissions, stale-temp reaping. +506 tests pass at 100% coverage. + ## 2026-07-26 — Python publishing workflow maintenance Dependabot PRs #82–#86 update the SHA-pinned PyPI trusted-publishing action to diff --git a/CHANGELOG.md b/CHANGELOG.md index 1de47f5..010ddd7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,26 @@ 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). -## [Unreleased] +## [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. + +### Added + +- `runcycles.journal`: file-per-commit `CommitJournal` (atomic write, idempotent replay). Config: `journal_enabled` (default `True`), `journal_dir` (default `~/.runcycles/commit-journal`), `retry_flush_timeout` (default 10 s); env `CYCLES_JOURNAL_ENABLED`, `CYCLES_JOURNAL_DIR`, `CYCLES_RETRY_FLUSH_TIMEOUT`. Records are partitioned into per-identity subdirectories (directories `0700`, files `0600` where supported) keyed by a non-secret PBKDF2-HMAC-SHA256 fingerprint of the server plus principal — the configured `tenant` when set (rotation-safe: any same-tenant credential can settle the records), else the API key — so clients with different servers or principals sharing a journal directory never replay each other's records, and one identity's replay claim cannot starve another's. The first engine created per identity replays surviving entries; corrupt files are renamed `*.corrupt` for operator triage. +- Event fallback: when a commit (first attempt or retry) returns `RESERVATION_EXPIRED`, the SDK posts the spend to `/v1/events` reusing the commit's idempotency key, with `metadata.recovered_reservation_id` / `metadata.recovery_reason` markers and no `overage_policy` (spec default `ALLOW_IF_AVAILABLE` never rejects). Applies to the `@cycles` lifecycles and both streaming context managers. `RESERVATION_FINALIZED` is still treated as settled. +- `flush()` on both retry engines; a process-wide `atexit` hook flushes sync engines under one shared `retry_flush_timeout` deadline (not per engine) so daemon retry threads aren't killed mid-backoff on clean exit and shutdown time stays bounded regardless of engine count. +- Rate-limit awareness end to end: HTTP 429 / `LIMIT_EXCEEDED` on the *first* commit attempt schedules a retry instead of releasing the reservation (a release would return budget for spend that already happened) in all four lifecycle variants, passing the server's `Retry-After` into the engine; on retried commit/event attempts the journal entry is retained and the next attempt waits at least `Retry-After` (consistent with `ErrorCode.is_retryable`). The `Retry-After` floor is persisted in the journal record as an absolute `not_before_ms`, so a restart during a long server-mandated wait does not replay into the window early. +- Authentication failures (401/403) on any commit attempt — first or retried — and on event fallbacks journal the spend instead of releasing or discarding it, so spend recorded during a key misconfiguration or rotation window replays once credentials are fixed. + +### Fixed + +- Self-review hardening (fleet-wide adversarial review): filename sanitization is ASCII-explicit, matching the TS/Java SDKs, so sibling SDKs sharing a tenant identity directory can always discard records this SDK wrote (and vice versa); the two cross-SDK PBKDF2 fingerprint vectors are now pinned in this suite; a whitespace-only `tenant` falls back to the key principal (matching Java); honored `Retry-After` values and restored journal floors are clamped to 1 hour; HTTP 410 triggers the expired/event-fallback path even when the response body was mangled in transit; a 4xx with no recognizable protocol error code (proxy error pages, forward-compat future codes) is no longer treated as a genuine rejection — the journal entry is retained and the reservation is never released; the base journal directory is also permission-tightened and stale temp files from crashed writers are reaped after 1 hour. +- With `retry_enabled=False`, failed commits were dropped with only a warning; they are now journaled for replay (the old drop behavior remains only when the journal is also disabled). +- `AsyncCommitRetryEngine` created retry tasks without holding a reference, so a pending retry could be garbage-collected mid-flight; task references are now held until completion. +- Commit retries exhausting, or landing after expiry, no longer lose the spend record silently: the journal entry is retained (transient exhaustion) or the event fallback records it (expiry). + +--- `TENANT_CLOSED` + `LIMIT_EXCEEDED` error-code support. `TENANT_CLOSED` implements the runtime spec v0.1.25.13 revision of `cycles-protocol-v0.yaml` ([runcycles/cycles-protocol#125](https://github.com/runcycles/cycles-protocol/pull/125)): servers return HTTP 409 `error=TENANT_CLOSED` on reservation create/commit/release/extend when the owning tenant is CLOSED (mirrors governance spec Rule 2). `LIMIT_EXCEEDED` closes the same class of gap for the runtime spec v0.1.25.12 revision (2026-07-04): HTTP 429 rate-limit responses carry `error=LIMIT_EXCEEDED` plus `Retry-After` / `X-RateLimit-Reset` headers. diff --git a/README.md b/README.md index 0285718..fefd339 100644 --- a/README.md +++ b/README.md @@ -227,9 +227,52 @@ CyclesConfig( retry_initial_delay=0.5, retry_multiplier=2.0, retry_max_delay=30.0, + retry_flush_timeout=10.0, + journal_enabled=True, + journal_dir=None, # None → ~/.runcycles/commit-journal ) ``` +### Commit durability + +A commit records spend that has already happened, so the SDK never lets one +exist only in memory. Every commit scheduled for background retry is first +journaled to disk and removed only on a terminal outcome. Records live under +`journal_dir` (default `~/.runcycles/commit-journal`) in a per-identity +subdirectory (directories `0700`, files `0600` where supported) keyed by a +non-secret fingerprint of the server plus principal — the configured +`tenant` when set (stable across API-key rotation; any same-tenant +credential can settle the records), otherwise the API key. Clients using +different servers or principals on the same machine never replay each +other's records. Without a tenant configured, rotating the API key orphans +pending records under the old fingerprint directory; records are plain +JSON, so moving them into the new identity directory is safe — replay is +idempotent: + +- **Process exit**: an `atexit` hook waits up to `retry_flush_timeout` seconds + (one process-wide budget shared across all engines) for in-flight retries; + anything unfinished stays journaled and is replayed automatically the next + time the process creates a client lifecycle. +- **Rate limiting**: HTTP 429 / `LIMIT_EXCEEDED` responses are transient + everywhere — a rate-limited *first* commit attempt is scheduled for retry + (never released, which would return budget for spend that already + happened), the journal entry is kept, and the next attempt waits at least + the server's `Retry-After`. The floor is persisted as an absolute + timestamp, so a restart mid-wait still honors it. +- **Authentication failures**: 401/403 on any commit attempt — first or + retried — journals the spend (never releases it) and stops the current + run's attempts, so spend recorded during a key misconfiguration or + rotation window replays once credentials are fixed. +- **Reservation expired before the commit landed**: the server has already + returned the reserved budget to the pool, so the SDK re-records the spend + via `POST /v1/events` (the protocol's post-hoc direct-debit endpoint), + tagging the event metadata with `recovered_reservation_id` for + reconciliation. Commit and event requests both carry idempotency keys, so + replays across restarts (or from multiple processes sharing a journal + directory) are exactly-once. +- Set `journal_enabled=False` (or `CYCLES_JOURNAL_ENABLED=false`) to opt out + and restore fire-and-forget behavior. + ### Default client / config Instead of passing `client=` to every `@cycles` decorator, set a module-level default: diff --git a/pyproject.toml b/pyproject.toml index 73107d1..1c04204 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "runcycles" -version = "0.4.3" +version = "0.5.0" 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/config.py b/runcycles/config.py index baf8bd0..b653e1a 100644 --- a/runcycles/config.py +++ b/runcycles/config.py @@ -31,6 +31,13 @@ class CyclesConfig: retry_initial_delay: float = 0.5 retry_multiplier: float = 2.0 retry_max_delay: float = 30.0 + # Bounded wait (seconds) at interpreter exit for in-flight commit retries. + # 0 disables the wait; journaled entries replay on the next run either way. + retry_flush_timeout: float = 10.0 + + # Durable journal for pending commits (survives process restarts) + journal_enabled: bool = True + journal_dir: str | None = None # None → ~/.runcycles/commit-journal @classmethod def from_env(cls, prefix: str = "CYCLES_") -> CyclesConfig: @@ -40,7 +47,8 @@ def from_env(cls, prefix: str = "CYCLES_") -> CyclesConfig: CYCLES_APP, CYCLES_WORKFLOW, CYCLES_AGENT, CYCLES_TOOLSET, CYCLES_CONNECT_TIMEOUT, CYCLES_READ_TIMEOUT, CYCLES_RETRY_ENABLED, CYCLES_RETRY_MAX_ATTEMPTS, CYCLES_RETRY_INITIAL_DELAY, - CYCLES_RETRY_MULTIPLIER, CYCLES_RETRY_MAX_DELAY. + CYCLES_RETRY_MULTIPLIER, CYCLES_RETRY_MAX_DELAY, + CYCLES_RETRY_FLUSH_TIMEOUT, CYCLES_JOURNAL_ENABLED, CYCLES_JOURNAL_DIR. """ base_url = os.environ.get(f"{prefix}BASE_URL", "") api_key = os.environ.get(f"{prefix}API_KEY", "") @@ -66,4 +74,7 @@ def from_env(cls, prefix: str = "CYCLES_") -> CyclesConfig: retry_initial_delay=float(os.environ.get(f"{prefix}RETRY_INITIAL_DELAY", "0.5")), retry_multiplier=float(os.environ.get(f"{prefix}RETRY_MULTIPLIER", "2.0")), retry_max_delay=float(os.environ.get(f"{prefix}RETRY_MAX_DELAY", "30.0")), + retry_flush_timeout=float(os.environ.get(f"{prefix}RETRY_FLUSH_TIMEOUT", "10.0")), + journal_enabled=os.environ.get(f"{prefix}JOURNAL_ENABLED", "true").lower() == "true", + journal_dir=os.environ.get(f"{prefix}JOURNAL_DIR"), ) diff --git a/runcycles/journal.py b/runcycles/journal.py new file mode 100644 index 0000000..9f0ab5d --- /dev/null +++ b/runcycles/journal.py @@ -0,0 +1,242 @@ +"""Durable on-disk journal for pending commits awaiting retry. + +Committed spend must survive process restarts: a commit that fails +transiently and only lives in an in-memory retry thread vanishes if the +process exits, and once the reservation's grace period elapses the server +returns the reserved budget to the pool — the ledger under-counts real +spend. The journal records every pending commit before the background +retry starts, and removes it only on a terminal outcome. On the next +process start the SDK replays surviving entries: commit first (idempotent), +falling back to ``POST /v1/events`` when the reservation has expired. + +Journal I/O is strictly best-effort — a failure to persist must never break +the commit path itself. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import re +import time +import uuid +from dataclasses import dataclass, field +from functools import lru_cache +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + +_RECORD_VERSION = 1 +_SUFFIX = ".json" + + +def default_journal_dir() -> Path: + """Default location for the pending-commit journal.""" + return Path.home() / ".runcycles" / "commit-journal" + + +@lru_cache(maxsize=256) +def _principal_digest(base_url: str, principal: str) -> str: + # PBKDF2 rather than a bare hash: the principal may embed a credential, + # and a KDF makes offline recovery from a leaked directory name harder + # for a weak user-chosen key. The round count is deliberately modest + # (~tens of ms cold, cached per identity per process): the principal is + # normally a high-entropy machine credential, so rounds only defend the + # weak-key fallback, and this runs synchronously at engine setup — + # password-storage round counts would stall async callers. Parameters + # are fixed constants — the digest must be deterministic across + # processes and releases. + digest = hashlib.pbkdf2_hmac( + "sha256", + principal.encode(), + f"runcycles-commit-journal\n{base_url}".encode(), + 30_000, + ) + return digest.hex()[:16] + + +def auth_fingerprint(base_url: str, api_key: str, tenant: str | None = None) -> str: + """Non-secret identity for one (server, principal) pair. + + Journal records are stored under a per-identity subdirectory so that + clients sharing a journal directory but using different servers or + principals never replay each other's records. When a tenant is + configured it is the principal — stable across API-key rotation, and + any same-tenant credential may settle the records. Without a tenant + the API key itself is the principal; rotating it then orphans pending + records under the old fingerprint (records are plain JSON, so an + operator can move them into the new identity directory — replay is + idempotent). The truncated PBKDF2-HMAC-SHA256 digest is not reversible, + so the fingerprint is safe to use as a directory name. + """ + principal = f"tenant\n{tenant}" if tenant and tenant.strip() else f"key\n{api_key}" + return _principal_digest(base_url, principal) + + +def _restrict_permissions(path: Path, mode: int) -> None: + """Best-effort permission tightening — records carry spend metadata. + + No-op semantics on platforms without POSIX modes; failure never blocks + the write itself. + """ + try: + path.chmod(mode) + except OSError: + pass + + +def _safe_filename(reservation_id: str) -> str: + # ASCII-only, matching the TS/Java SDKs exactly: same-tenant clients in + # other languages settle records from this directory, and their discard() + # must compute the identical filename or the record replays forever. + sanitized = re.sub(r"[^A-Za-z0-9_-]", "_", reservation_id) + return f"{sanitized}{_SUFFIX}" + + +@dataclass +class PendingCommitRecord: + """One journaled pending commit (or its event fallback).""" + + reservation_id: str + base_url: str + mode: str = "commit" # "commit" | "event" + commit_body: dict[str, Any] | None = None + event_fallback_body: dict[str, Any] | None = None + recorded_at_ms: int = field(default_factory=lambda: int(time.time() * 1000)) + # Absolute wall-clock floor (ms) for the next attempt, set from a 429's + # Retry-After. Absolute so it survives a process restart mid-wait. + not_before_ms: int | None = None + + def to_json(self) -> str: + return json.dumps( + { + "version": _RECORD_VERSION, + "reservation_id": self.reservation_id, + "base_url": self.base_url, + "mode": self.mode, + "commit_body": self.commit_body, + "event_fallback_body": self.event_fallback_body, + "recorded_at_ms": self.recorded_at_ms, + "not_before_ms": self.not_before_ms, + } + ) + + @classmethod + def from_json(cls, raw: str) -> PendingCommitRecord: + data = json.loads(raw) + reservation_id = data["reservation_id"] + mode = data.get("mode", "commit") + if not isinstance(reservation_id, str) or not reservation_id: + raise ValueError("journal record missing reservation_id") + if mode not in ("commit", "event"): + raise ValueError(f"journal record has unknown mode: {mode}") + if mode == "commit" and not isinstance(data.get("commit_body"), dict): + raise ValueError("commit-mode journal record missing commit_body") + if mode == "event" and not isinstance(data.get("event_fallback_body"), dict): + raise ValueError("event-mode journal record missing event_fallback_body") + not_before_raw = data.get("not_before_ms") + return cls( + reservation_id=reservation_id, + base_url=data.get("base_url", ""), + mode=mode, + commit_body=data.get("commit_body"), + event_fallback_body=data.get("event_fallback_body"), + recorded_at_ms=int(data.get("recorded_at_ms", 0)), + not_before_ms=int(not_before_raw) if not_before_raw is not None else None, + ) + + +class CommitJournal: + """File-per-pending-commit journal. + + Each record is one JSON file named after its reservation id, written + atomically (temp file + rename) and deleted on a terminal outcome. + One file per commit avoids cross-process file locking; concurrent + replay by multiple processes is safe because commit and event requests + both carry idempotency keys. + """ + + def __init__(self, directory: Path) -> None: + self._dir = directory + + @property + def directory(self) -> Path: + return self._dir + + def record(self, entry: PendingCommitRecord) -> None: + """Persist a pending commit. Never raises.""" + try: + self._dir.mkdir(parents=True, exist_ok=True) + _restrict_permissions(self._dir.parent, 0o700) + _restrict_permissions(self._dir, 0o700) + target = self._dir / _safe_filename(entry.reservation_id) + # Unique temp name per writer: concurrent processes may settle + # the same reservation (replay is idempotent), and a shared + # temp filename would let one truncate the other mid-write and + # atomically publish partial JSON. + tmp = self._dir / f"{target.name}.{os.getpid()}.{uuid.uuid4().hex[:8]}.tmp" + try: + tmp.write_text(entry.to_json(), encoding="utf-8") + _restrict_permissions(tmp, 0o600) + tmp.replace(target) + except OSError: + try: + tmp.unlink(missing_ok=True) + except OSError: + pass + raise + logger.debug("Journaled pending commit: id=%s, path=%s", entry.reservation_id, target) + except OSError: + logger.warning( + "Failed to journal pending commit (continuing without durability): id=%s", + entry.reservation_id, + exc_info=True, + ) + + def discard(self, reservation_id: str) -> None: + """Remove a journal entry after a terminal outcome. Never raises.""" + try: + (self._dir / _safe_filename(reservation_id)).unlink(missing_ok=True) + except OSError: + logger.warning("Failed to discard journal entry: id=%s", reservation_id, exc_info=True) + + def load_pending(self, base_url: str) -> list[PendingCommitRecord]: + """Load surviving entries for the given server. Never raises. + + Only entries recorded against the same ``base_url`` are returned — + a journal directory may be shared by processes talking to different + servers, and replaying against the wrong one would fail auth. + Unparseable files are renamed to ``*.corrupt`` so they surface to + operators instead of being retried forever. + """ + entries: list[PendingCommitRecord] = [] + try: + if not self._dir.is_dir(): + return entries + cutoff = time.time() - 3600 + for tmp in self._dir.glob("*.tmp"): + # Crashed writers leave unique temp files behind; reap the + # stale ones so they don't accumulate forever. + try: + if tmp.stat().st_mtime < cutoff: + tmp.unlink(missing_ok=True) + except OSError: + pass + for path in sorted(self._dir.glob(f"*{_SUFFIX}")): + try: + entry = PendingCommitRecord.from_json(path.read_text(encoding="utf-8")) + except (OSError, ValueError, KeyError, json.JSONDecodeError): + logger.warning("Skipping corrupt journal entry: %s", path, exc_info=True) + try: + path.replace(path.with_suffix(".corrupt")) + except OSError: + pass + continue + if entry.base_url == base_url: + entries.append(entry) + except OSError: + logger.warning("Failed to scan commit journal: %s", self._dir, exc_info=True) + return entries diff --git a/runcycles/lifecycle.py b/runcycles/lifecycle.py index 409d3ec..e4afabc 100644 --- a/runcycles/lifecycle.py +++ b/runcycles/lifecycle.py @@ -37,7 +37,11 @@ Subject, ) from runcycles.response import CyclesResponse -from runcycles.retry import AsyncCommitRetryEngine, CommitRetryEngine +from runcycles.retry import ( + AsyncCommitRetryEngine, + CommitRetryEngine, + _is_recognized_rejection, +) logger = logging.getLogger(__name__) @@ -163,6 +167,33 @@ def _build_commit_body( return body +def _build_event_fallback_body( + 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 + returned the reserved budget to the pool at that point). + + Reuses the commit's idempotency key — the event idempotency namespace is + separate, so replays across process restarts stay exactly-once. Omits + overage_policy: the spec default ALLOW_IF_AVAILABLE never rejects, which + is the right bias when the spend has already happened. + """ + metadata = dict(commit_body.get("metadata") or {}) + metadata["recovered_reservation_id"] = reservation_id + metadata["recovery_reason"] = "commit_after_reservation_expired" + body: dict[str, Any] = { + "idempotency_key": commit_body["idempotency_key"], + "subject": subject, + "action": action, + "actual": commit_body["actual"], + "metadata": metadata, + } + if "metrics" in commit_body: + body["metrics"] = commit_body["metrics"] + return body + + def _build_release_body(reason: str) -> dict[str, Any]: return {"idempotency_key": str(uuid.uuid4()), "reason": reason} @@ -341,7 +372,10 @@ def execute( metrics.latency_ms = method_elapsed commit_body = _build_commit_body(actual_amount, cfg.unit, metrics, ctx.commit_metadata) - self._handle_commit(reservation_id, commit_body) + event_fallback = _build_event_fallback_body( + reservation_id, create_body["subject"], create_body["action"], commit_body, + ) + self._handle_commit(reservation_id, commit_body, event_fallback) return result @@ -355,7 +389,9 @@ def execute( heartbeat_thread.join(timeout=1.0) _clear_context() - def _handle_commit(self, reservation_id: str, commit_body: dict[str, Any]) -> None: + def _handle_commit( + self, reservation_id: str, commit_body: dict[str, Any], event_fallback_body: dict[str, Any], + ) -> None: try: logger.debug("Committing: id=%s", reservation_id) response = self._client.commit_reservation(reservation_id, commit_body) @@ -363,23 +399,56 @@ def _handle_commit(self, reservation_id: str, commit_body: dict[str, Any]) -> No logger.info("Commit successful: id=%s", reservation_id) elif response.is_transport_error or response.is_server_error: logger.warning("Commit failed (retryable): id=%s, status=%d", reservation_id, response.status) - self._retry_engine.schedule(reservation_id, commit_body) + self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) else: error_code = None error_resp = response.get_error_response() if error_resp and error_resp.error_code: error_code = error_resp.error_code.value - if error_code in ("RESERVATION_FINALIZED", "RESERVATION_EXPIRED"): - logger.warning("Reservation already finalized/expired: id=%s", reservation_id) + if response.status == 429 or error_code == "LIMIT_EXCEEDED": + # Rate-limited, not rejected: releasing here would return + # budget for spend that already happened. Retry instead, + # 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, + retry_after_ms=response.retry_after_ms_header, + ) + elif response.status in (401, 403): + # Credentials failed after the spend happened: journal the + # commit for replay once they're fixed. Never release — + # that would return budget for real spend. + logger.error( + "Commit got authentication failure (status=%d); journaling for replay: id=%s", + response.status, reservation_id, + ) + self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) + elif response.status == 410 or error_code == "RESERVATION_EXPIRED": + logger.warning( + "Reservation expired before commit; recovering spend via POST /v1/events: id=%s", + reservation_id, + ) + self._retry_engine.schedule_event(reservation_id, event_fallback_body) + elif error_code == "RESERVATION_FINALIZED": + logger.warning("Reservation already finalized: id=%s", reservation_id) elif error_code == "IDEMPOTENCY_MISMATCH": logger.warning("Commit idempotency mismatch (not releasing): id=%s", reservation_id) - elif response.is_client_error: + elif response.is_client_error and _is_recognized_rejection(error_code): self._handle_release(reservation_id, f"commit_rejected_{error_code}") + elif response.is_client_error: + # 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, + ) + self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) else: logger.warning("Unrecognized commit response: id=%s, response=%s", reservation_id, response) except Exception: logger.exception("Failed to commit: id=%s", reservation_id) - self._retry_engine.schedule(reservation_id, commit_body) + self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) def _handle_release(self, reservation_id: str, reason: str) -> None: try: @@ -504,7 +573,10 @@ async def execute( metrics.latency_ms = method_elapsed commit_body = _build_commit_body(actual_amount, cfg.unit, metrics, ctx.commit_metadata) - await self._handle_commit(reservation_id, commit_body) + event_fallback = _build_event_fallback_body( + reservation_id, create_body["subject"], create_body["action"], commit_body, + ) + await self._handle_commit(reservation_id, commit_body, event_fallback) return result @@ -521,29 +593,64 @@ async def execute( pass _clear_context() - async def _handle_commit(self, reservation_id: str, commit_body: dict[str, Any]) -> None: + async def _handle_commit( + 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) if response.is_success: logger.info("Commit successful: id=%s", reservation_id) elif response.is_transport_error or response.is_server_error: - self._retry_engine.schedule(reservation_id, commit_body) + self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) else: error_code = None error_resp = response.get_error_response() if error_resp and error_resp.error_code: error_code = error_resp.error_code.value - if error_code in ("RESERVATION_FINALIZED", "RESERVATION_EXPIRED"): - logger.warning("Reservation already finalized/expired: id=%s", reservation_id) + if response.status == 429 or error_code == "LIMIT_EXCEEDED": + # Rate-limited, not rejected: releasing here would return + # budget for spend that already happened. Retry instead, + # 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, + retry_after_ms=response.retry_after_ms_header, + ) + elif response.status in (401, 403): + # Credentials failed after the spend happened: journal the + # commit for replay once they're fixed. Never release — + # that would return budget for real spend. + logger.error( + "Commit got authentication failure (status=%d); journaling for replay: id=%s", + response.status, reservation_id, + ) + self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) + elif response.status == 410 or error_code == "RESERVATION_EXPIRED": + logger.warning( + "Reservation expired before commit; recovering spend via POST /v1/events: id=%s", + reservation_id, + ) + self._retry_engine.schedule_event(reservation_id, event_fallback_body) + elif error_code == "RESERVATION_FINALIZED": + logger.warning("Reservation already finalized: id=%s", reservation_id) elif error_code == "IDEMPOTENCY_MISMATCH": logger.warning("Commit idempotency mismatch (not releasing): id=%s", reservation_id) - elif response.is_client_error: + elif response.is_client_error and _is_recognized_rejection(error_code): await self._handle_release(reservation_id, f"commit_rejected_{error_code}") + elif response.is_client_error: + # 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, + ) + self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) else: logger.warning("Unrecognized commit response: id=%s, response=%s", reservation_id, response) except Exception: logger.exception("Failed to commit: id=%s", reservation_id) - self._retry_engine.schedule(reservation_id, commit_body) + self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) async def _handle_release(self, reservation_id: str, reason: str) -> None: try: diff --git a/runcycles/retry.py b/runcycles/retry.py index 621e112..d3665bb 100644 --- a/runcycles/retry.py +++ b/runcycles/retry.py @@ -1,32 +1,124 @@ -"""Background commit retry engine with exponential backoff.""" +"""Background commit retry engine with exponential backoff. + +Durability model: every scheduled retry is journaled to disk first (see +:mod:`runcycles.journal`) and the entry is removed only on a terminal +outcome, so pending commits survive process restarts and are replayed the +next time an engine is created for the same journal directory. When a +retried commit lands after the reservation's grace period (the server +answers ``RESERVATION_EXPIRED`` and has already returned the reserved +budget to the pool), the engine falls back to ``POST /v1/events`` — the +spec's post-hoc direct-debit endpoint — so the spend is still recorded. +""" from __future__ import annotations import asyncio +import atexit import logging import threading import time +import weakref from dataclasses import dataclass +from pathlib import Path from typing import Any +from runcycles import journal as _journal from runcycles.config import CyclesConfig +from runcycles.journal import CommitJournal, PendingCommitRecord +from runcycles.models import ErrorCode +from runcycles.response import CyclesResponse logger = logging.getLogger(__name__) +# Upper bound on any honored server-requested or restored delay (1 hour). +# A mangled Retry-After or a corrupted journal floor must not park a spend +# record for days. +_MAX_HONORED_DELAY_S = 3600.0 + + +def _is_recognized_rejection(code: str | None) -> bool: + """True when the error code is a known protocol code (not forward-compat). + + Only a recognized rejection justifies destroying a durable spend record + or releasing a reservation; a codeless or unknown-future-code 4xx (a + proxy error page, a newer server) is retained instead. + """ + return code is not None and ErrorCode.from_string(code) is not ErrorCode.UNKNOWN + @dataclass class _PendingCommit: reservation_id: str - commit_body: dict[str, Any] + commit_body: dict[str, Any] | None = None + event_fallback_body: dict[str, Any] | None = None + mode: str = "commit" # "commit" | "event" attempt: int = 0 + # Server-requested minimum delay (seconds) before the next attempt, + # set from a 429's Retry-After and consumed by the retry loop. + retry_after_s: float | None = None -class CommitRetryEngine: - """Retries failed commits in background threads with exponential backoff. +def _extract_error_code(response: CyclesResponse) -> str | None: + error_resp = response.get_error_response() + if error_resp and error_resp.error_code: + return error_resp.error_code.value + raw = response.get_body_attribute("error") + return raw if isinstance(raw, str) else None - Used by the sync lifecycle. Commits that fail transiently are scheduled - for retry. The engine stops retrying after ``max_attempts``. - """ + +# Journal replay must happen at most once per identity directory per process: +# the first engine created for a (server, credential) identity claims its +# subdirectory and replays surviving entries; later engines (and the +# claimer's own in-flight work) are excluded. Because the claim is scoped to +# the identity subdirectory — not the shared parent — an engine for one +# server/key can never block another identity's entries from replaying. +_replay_lock = threading.Lock() +_replayed_dirs: set[Path] = set() + + +def _claim_replay(directory: Path) -> bool: + with _replay_lock: + if directory in _replayed_dirs: + return False + _replayed_dirs.add(directory) + return True + + +# Retry threads are daemons so they never wedge interpreter exit, but a +# clean exit would otherwise kill them mid-backoff. A single atexit hook +# gives in-flight retries a bounded window to finish; whatever remains +# stays journaled and replays on the next run. +_flush_lock = threading.Lock() +_live_engines: weakref.WeakSet[CommitRetryEngine] = weakref.WeakSet() +_atexit_registered = False + + +def _flush_all_engines() -> None: + # One process-wide deadline: retry_flush_timeout bounds the whole exit + # wait, not each engine. With several engines (decorators, streams) the + # remaining budget shrinks as each is flushed. + engines = list(_live_engines) + if not engines: + return + deadline = time.monotonic() + max(engine._flush_timeout for engine in engines) + for engine in engines: + remaining = deadline - time.monotonic() + if remaining <= 0: + return + engine.flush(remaining) + + +def _register_engine_for_flush(engine: CommitRetryEngine) -> None: + global _atexit_registered + with _flush_lock: + _live_engines.add(engine) + if not _atexit_registered: + atexit.register(_flush_all_engines) + _atexit_registered = True + + +class _RetryEngineBase: + """Configuration, journal plumbing, and outcome classification shared by both engines.""" def __init__(self, config: CyclesConfig) -> None: self._enabled = config.retry_enabled @@ -34,132 +126,451 @@ def __init__(self, config: CyclesConfig) -> None: self._initial_delay = config.retry_initial_delay self._multiplier = config.retry_multiplier self._max_delay = config.retry_max_delay + self._flush_timeout = config.retry_flush_timeout + self._base_url = config.base_url self._client: Any = None # set by lifecycle to avoid circular import + self._journal: CommitJournal | None = None + if config.journal_enabled: + base = Path(config.journal_dir) if config.journal_dir else _journal.default_journal_dir() + # Per-identity subdirectory: clients sharing a journal directory + # but using different servers or API keys must never replay each + # other's records (a foreign record would 401/403 and be + # discarded as terminal — permanent spend loss). + self._journal = CommitJournal( + base / _journal.auth_fingerprint(config.base_url, config.api_key, config.tenant) + ) + + def _journal_record(self, pending: _PendingCommit) -> None: + if self._journal is not None: + not_before_ms = None + if pending.retry_after_s is not None: + not_before_ms = int(time.time() * 1000 + pending.retry_after_s * 1000) + self._journal.record( + PendingCommitRecord( + reservation_id=pending.reservation_id, + base_url=self._base_url, + mode=pending.mode, + commit_body=pending.commit_body, + event_fallback_body=pending.event_fallback_body, + not_before_ms=not_before_ms, + ) + ) + + def _journal_discard(self, reservation_id: str) -> None: + if self._journal is not None: + self._journal.discard(reservation_id) + + def _load_replay_entries(self) -> list[_PendingCommit]: + """Claim and load journaled entries for this engine's server, if eligible.""" + if not self._enabled or self._journal is None or self._client is None: + return [] + if not _claim_replay(self._journal.directory): + return [] + entries = self._journal.load_pending(self._base_url) + if entries: + logger.info( + "Replaying %d journaled pending commit(s) from %s", len(entries), self._journal.directory + ) + now_ms = time.time() * 1000 + pendings = [] + for e in entries: + # Restore a persisted Retry-After floor as a relative delay; a + # floor already in the past falls back to normal backoff. + retry_after_s: float | None = None + if e.not_before_ms is not None and e.not_before_ms > now_ms: + retry_after_s = min((e.not_before_ms - now_ms) / 1000.0, _MAX_HONORED_DELAY_S) + pendings.append( + _PendingCommit( + reservation_id=e.reservation_id, + commit_body=e.commit_body, + event_fallback_body=e.event_fallback_body, + mode=e.mode, + retry_after_s=retry_after_s, + ) + ) + return pendings + + def _log_disabled_drop(self, pending: _PendingCommit) -> None: + if self._journal is not None: + logger.warning( + "Retry disabled; pending %s journaled for replay on next run: reservation_id=%s", + pending.mode, pending.reservation_id, + ) + else: + logger.warning( + "Retry and journal disabled, dropping failed %s: reservation_id=%s", + pending.mode, pending.reservation_id, + ) + + def _delay_for(self, pending: _PendingCommit) -> float: + delay = min(self._initial_delay * (self._multiplier ** pending.attempt), self._max_delay) + if pending.retry_after_s is not None: + # A 429 asked us to wait: honor the server's Retry-After when it + # exceeds our own backoff, then clear it — it applies once. + delay = max(delay, pending.retry_after_s) + pending.retry_after_s = None + return delay + + def _is_rate_limited(self, pending: _PendingCommit, response: CyclesResponse) -> bool: + """Detect 429 / LIMIT_EXCEEDED and stash its Retry-After. Transient, never terminal.""" + if response.status != 429 and _extract_error_code(response) != "LIMIT_EXCEEDED": + return False + retry_after_ms = response.retry_after_ms_header + if retry_after_ms is not None: + pending.retry_after_s = min(retry_after_ms / 1000.0, _MAX_HONORED_DELAY_S) + # Persist the floor: a restart during a long Retry-After wait + # must not replay into the window the server told us to avoid. + self._journal_record(pending) + logger.warning( + "%s retry rate-limited: reservation_id=%s, attempt=%d, retry_after_ms=%s", + pending.mode, pending.reservation_id, pending.attempt, retry_after_ms, + ) + return True + + def _classify_commit_response(self, pending: _PendingCommit, response: CyclesResponse) -> bool: + """Handle a commit attempt's response. Returns True when terminal. + + May flip ``pending`` into event mode; the caller then delivers the + event fallback immediately (no extra backoff) via ``_attempt_event``. + """ + if response.is_success: + logger.info( + "Commit retry succeeded: reservation_id=%s, attempt=%d", + pending.reservation_id, pending.attempt, + ) + self._journal_discard(pending.reservation_id) + return True + if self._is_rate_limited(pending, response): + return False + if response.status in (401, 403): + logger.error( + "Commit retry got authentication failure (status=%d); journal entry retained — " + "fix credentials and restart to replay: reservation_id=%s", + response.status, pending.reservation_id, + ) + return True + if response.is_client_error: + code = _extract_error_code(response) + # Status 410 catches expired responses whose body was mangled in + # transit — the 429 path is status-hardened, this one must be too. + if response.status == 410 or code == "RESERVATION_EXPIRED": + if pending.event_fallback_body: + logger.warning( + "Reservation expired before commit landed; falling back to POST /v1/events: " + "reservation_id=%s", + pending.reservation_id, + ) + pending.mode = "event" + pending.attempt = 0 + self._journal_record(pending) + return False + logger.error( + "Reservation expired with no event fallback; spend is unrecorded " + "(journal entry retained): reservation_id=%s", + pending.reservation_id, + ) + return True + if not _is_recognized_rejection(code): + logger.error( + "Commit retry got unclassifiable client error (status=%d, error=%s); " + "journal entry retained: reservation_id=%s", + response.status, code, pending.reservation_id, + ) + return True + logger.warning( + "Commit retry got non-retryable error: reservation_id=%s, status=%d, error=%s", + pending.reservation_id, response.status, code, + ) + self._journal_discard(pending.reservation_id) + return True + logger.warning( + "Commit retry failed: reservation_id=%s, attempt=%d, status=%d", + pending.reservation_id, pending.attempt, response.status, + ) + return False + + def _classify_event_response(self, pending: _PendingCommit, response: CyclesResponse) -> bool: + """Handle an event-fallback attempt's response. Returns True when terminal.""" + if response.is_success: + logger.info( + "Recovered expired-commit spend via /v1/events: reservation_id=%s, event_id=%s", + pending.reservation_id, response.get_body_attribute("event_id"), + ) + self._journal_discard(pending.reservation_id) + return True + if self._is_rate_limited(pending, response): + return False + if response.status in (401, 403): + logger.error( + "Event fallback got authentication failure (status=%d); journal entry retained — " + "fix credentials and restart to replay: reservation_id=%s", + response.status, pending.reservation_id, + ) + return True + if response.is_client_error: + code = _extract_error_code(response) + if not _is_recognized_rejection(code): + logger.error( + "Event fallback got unclassifiable client error (status=%d, error=%s); " + "journal entry retained: reservation_id=%s", + response.status, code, pending.reservation_id, + ) + return True + logger.error( + "Event fallback rejected (%s); spend recovery failed: reservation_id=%s, status=%d", + code, pending.reservation_id, response.status, + ) + self._journal_discard(pending.reservation_id) + return True + logger.warning( + "Event fallback failed: reservation_id=%s, attempt=%d, status=%d", + pending.reservation_id, pending.attempt, response.status, + ) + return False + + def _log_exhausted(self, pending: _PendingCommit) -> None: + logger.error( + "%s retry exhausted: reservation_id=%s, attempts=%d%s", + pending.mode, pending.reservation_id, self._max_attempts, + " (journal entry retained for replay on next run)" if self._journal is not None else "", + ) + + +class CommitRetryEngine(_RetryEngineBase): + """Retries failed commits in background threads with exponential backoff. + + Used by the sync lifecycle. Commits that fail transiently are scheduled + for retry. The engine stops retrying after ``max_attempts``; entries + remain journaled for replay on the next run. + """ + + def __init__(self, config: CyclesConfig) -> None: + super().__init__(config) + self._threads: set[threading.Thread] = set() + self._threads_lock = threading.Lock() def set_client(self, client: Any) -> None: self._client = client + for pending in self._load_replay_entries(): + self._spawn(pending) - def schedule(self, reservation_id: str, commit_body: dict[str, Any]) -> None: + def schedule( + self, + reservation_id: str, + commit_body: dict[str, Any], + event_fallback_body: dict[str, Any] | None = None, + retry_after_ms: int | None = None, + ) -> None: + pending = _PendingCommit(reservation_id, commit_body, event_fallback_body, mode="commit") + if retry_after_ms is not None: + # A rate-limited first attempt passes its Retry-After along so + # the first background retry honors the server's delay. + pending.retry_after_s = min(retry_after_ms / 1000.0, _MAX_HONORED_DELAY_S) + self._submit(pending) + + def schedule_event(self, reservation_id: str, event_body: dict[str, Any]) -> None: + """Deliver spend via POST /v1/events for a reservation that already expired.""" + self._submit(_PendingCommit(reservation_id, None, event_body, mode="event")) + + def _submit(self, pending: _PendingCommit) -> None: + self._journal_record(pending) if not self._enabled: - logger.warning("Retry disabled, dropping failed commit: reservation_id=%s", reservation_id) + self._log_disabled_drop(pending) return + self._spawn(pending) - pending = _PendingCommit(reservation_id=reservation_id, commit_body=commit_body) - thread = threading.Thread(target=self._retry_loop, args=(pending,), daemon=True) + def _spawn(self, pending: _PendingCommit) -> None: + thread = threading.Thread( + target=self._run, + args=(pending,), + daemon=True, + name=f"cycles-commit-retry-{pending.reservation_id[:12]}", + ) + with self._threads_lock: + self._threads.add(thread) + _register_engine_for_flush(self) thread.start() + def _run(self, pending: _PendingCommit) -> None: + try: + self._retry_loop(pending) + finally: + with self._threads_lock: + self._threads.discard(threading.current_thread()) + + def flush(self, timeout: float | None = None) -> None: + """Wait (bounded) for in-flight retry threads to finish. + + Called automatically at interpreter exit. Anything still pending + when the timeout elapses stays journaled and replays on the next run. + """ + if timeout is None: + timeout = self._flush_timeout + if timeout <= 0: + return + deadline = time.monotonic() + timeout + with self._threads_lock: + threads = list(self._threads) + for thread in threads: + remaining = deadline - time.monotonic() + if remaining <= 0: + return + if thread is not threading.current_thread() and thread.is_alive(): + thread.join(timeout=remaining) + def _retry_loop(self, pending: _PendingCommit) -> None: while pending.attempt < self._max_attempts: - delay = min(self._initial_delay * (self._multiplier ** pending.attempt), self._max_delay) + delay = self._delay_for(pending) pending.attempt += 1 logger.info( - "Scheduling commit retry: reservation_id=%s, attempt=%d/%d, delay=%.1fs", - pending.reservation_id, pending.attempt, self._max_attempts, delay, + "Scheduling %s retry: reservation_id=%s, attempt=%d/%d, delay=%.1fs", + pending.mode, pending.reservation_id, pending.attempt, self._max_attempts, delay, ) time.sleep(delay) try: if self._client is None: - logger.error("No client set on retry engine, cannot retry commit") - return - response = self._client.commit_reservation(pending.reservation_id, pending.commit_body) - if response.is_success: - logger.info( - "Commit retry succeeded: reservation_id=%s, attempt=%d", - pending.reservation_id, pending.attempt, - ) + logger.error("No client set on retry engine, cannot retry %s", pending.mode) return - elif response.is_client_error: - logger.warning( - "Commit retry got non-retryable error: reservation_id=%s, status=%d", - pending.reservation_id, response.status, - ) + if self._attempt_once(pending): return - else: - logger.warning( - "Commit retry failed: reservation_id=%s, attempt=%d, status=%d", - pending.reservation_id, pending.attempt, response.status, - ) except Exception: logger.exception( - "Commit retry error: reservation_id=%s, attempt=%d", - pending.reservation_id, pending.attempt, + "%s retry error: reservation_id=%s, attempt=%d", + pending.mode, pending.reservation_id, pending.attempt, ) - logger.error( - "Commit retry exhausted: reservation_id=%s, attempts=%d", - pending.reservation_id, self._max_attempts, - ) + self._log_exhausted(pending) + + def _attempt_once(self, pending: _PendingCommit) -> bool: + if pending.mode == "commit": + response = self._client.commit_reservation(pending.reservation_id, pending.commit_body) + terminal = self._classify_commit_response(pending, response) + if not terminal and pending.mode == "event": + # Expired → deliver the event fallback immediately, no extra backoff. + return self._attempt_once(pending) + return terminal + response = self._client.create_event(pending.event_fallback_body) + return self._classify_event_response(pending, response) -class AsyncCommitRetryEngine: +class AsyncCommitRetryEngine(_RetryEngineBase): """Retries failed commits as async tasks with exponential backoff. - Used by the async lifecycle. + Used by the async lifecycle. Task references are held until completion + so pending retries cannot be garbage-collected mid-flight. """ def __init__(self, config: CyclesConfig) -> None: - self._enabled = config.retry_enabled - self._max_attempts = config.retry_max_attempts - self._initial_delay = config.retry_initial_delay - self._multiplier = config.retry_multiplier - self._max_delay = config.retry_max_delay - self._client: Any = None + super().__init__(config) + self._tasks: set[asyncio.Task[None]] = set() + self._replay_deferred = False def set_client(self, client: Any) -> None: self._client = client + self._maybe_replay() + + def schedule( + self, + reservation_id: str, + commit_body: dict[str, Any], + event_fallback_body: dict[str, Any] | None = None, + retry_after_ms: int | None = None, + ) -> None: + pending = _PendingCommit(reservation_id, commit_body, event_fallback_body, mode="commit") + if retry_after_ms is not None: + # A rate-limited first attempt passes its Retry-After along so + # the first background retry honors the server's delay. + pending.retry_after_s = min(retry_after_ms / 1000.0, _MAX_HONORED_DELAY_S) + self._submit(pending) + + def schedule_event(self, reservation_id: str, event_body: dict[str, Any]) -> None: + """Deliver spend via POST /v1/events for a reservation that already expired.""" + self._submit(_PendingCommit(reservation_id, None, event_body, mode="event")) - def schedule(self, reservation_id: str, commit_body: dict[str, Any]) -> None: + def _submit(self, pending: _PendingCommit) -> None: + self._journal_record(pending) if not self._enabled: - logger.warning("Retry disabled, dropping failed commit: reservation_id=%s", reservation_id) + self._log_disabled_drop(pending) return + try: + loop = asyncio.get_running_loop() + except RuntimeError: + if self._journal is not None: + logger.error( + "No running event loop; pending %s journaled for replay on next run: reservation_id=%s", + pending.mode, pending.reservation_id, + ) + else: + logger.error( + "No running event loop, cannot schedule async %s retry: reservation_id=%s", + pending.mode, pending.reservation_id, + ) + return + if self._replay_deferred: + self._replay_deferred = False + self._maybe_replay() + self._spawn(loop, pending) + + def _spawn(self, loop: asyncio.AbstractEventLoop, pending: _PendingCommit) -> None: + task = loop.create_task(self._retry_loop(pending)) + self._tasks.add(task) + task.add_done_callback(self._tasks.discard) - pending = _PendingCommit(reservation_id=reservation_id, commit_body=commit_body) + def _maybe_replay(self) -> None: + if not self._enabled or self._journal is None or self._client is None: + return try: loop = asyncio.get_running_loop() - loop.create_task(self._retry_loop(pending)) except RuntimeError: - logger.error("No running event loop, cannot schedule async commit retry: reservation_id=%s", reservation_id) + # No loop yet (e.g. engine built during sync setup) — replay on + # the first schedule() call, which requires a running loop. + self._replay_deferred = True + return + for pending in self._load_replay_entries(): + self._spawn(loop, pending) + + async def flush(self, timeout: float | None = None) -> None: + """Wait (bounded) for in-flight retry tasks to finish.""" + if timeout is None: + timeout = self._flush_timeout + if timeout <= 0: + return + tasks = [t for t in self._tasks if not t.done()] + if tasks: + await asyncio.wait(tasks, timeout=timeout) async def _retry_loop(self, pending: _PendingCommit) -> None: while pending.attempt < self._max_attempts: - delay = min(self._initial_delay * (self._multiplier ** pending.attempt), self._max_delay) + delay = self._delay_for(pending) pending.attempt += 1 logger.info( - "Scheduling async commit retry: reservation_id=%s, attempt=%d/%d, delay=%.1fs", - pending.reservation_id, pending.attempt, self._max_attempts, delay, + "Scheduling async %s retry: reservation_id=%s, attempt=%d/%d, delay=%.1fs", + pending.mode, pending.reservation_id, pending.attempt, self._max_attempts, delay, ) await asyncio.sleep(delay) try: if self._client is None: - logger.error("No client set on async retry engine, cannot retry commit") + logger.error("No client set on async retry engine, cannot retry %s", pending.mode) return - response = await self._client.commit_reservation(pending.reservation_id, pending.commit_body) - if response.is_success: - logger.info( - "Async commit retry succeeded: reservation_id=%s, attempt=%d", - pending.reservation_id, pending.attempt, - ) + if await self._attempt_once(pending): return - elif response.is_client_error: - logger.warning( - "Async commit retry got non-retryable error: reservation_id=%s, status=%d", - pending.reservation_id, response.status, - ) - return - else: - logger.warning( - "Async commit retry failed: reservation_id=%s, attempt=%d, status=%d", - pending.reservation_id, pending.attempt, response.status, - ) except Exception: logger.exception( - "Async commit retry error: reservation_id=%s, attempt=%d", - pending.reservation_id, pending.attempt, + "Async %s retry error: reservation_id=%s, attempt=%d", + pending.mode, pending.reservation_id, pending.attempt, ) - logger.error( - "Async commit retry exhausted: reservation_id=%s, attempts=%d", - pending.reservation_id, self._max_attempts, - ) + self._log_exhausted(pending) + + async def _attempt_once(self, pending: _PendingCommit) -> bool: + if pending.mode == "commit": + response = await self._client.commit_reservation(pending.reservation_id, pending.commit_body) + terminal = self._classify_commit_response(pending, response) + if not terminal and pending.mode == "event": + # Expired → deliver the event fallback immediately, no extra backoff. + return await self._attempt_once(pending) + return terminal + response = await self._client.create_event(pending.event_fallback_body) + return self._classify_event_response(pending, response) diff --git a/runcycles/streaming.py b/runcycles/streaming.py index 38adb7b..511da03 100644 --- a/runcycles/streaming.py +++ b/runcycles/streaming.py @@ -17,6 +17,7 @@ from runcycles.exceptions import CyclesProtocolError from runcycles.lifecycle import ( _build_commit_body, + _build_event_fallback_body, _build_extend_body, _build_protocol_exception, _build_release_body, @@ -30,7 +31,11 @@ ReservationCreateResponse, Subject, ) -from runcycles.retry import AsyncCommitRetryEngine, CommitRetryEngine +from runcycles.retry import ( + AsyncCommitRetryEngine, + CommitRetryEngine, + _is_recognized_rejection, +) logger = logging.getLogger(__name__) @@ -275,29 +280,68 @@ def _handle_commit(self) -> None: commit_body = _build_commit_body(actual, unit, metrics, self._metadata) assert self._reservation_id is not None + event_fallback = _build_event_fallback_body( + self._reservation_id, + self._subject.model_dump(exclude_none=True), + self._action.model_dump(exclude_none=True), + commit_body, + ) try: response = self._client.commit_reservation(self._reservation_id, commit_body) if response.is_success: logger.info("Stream commit successful: id=%s", self._reservation_id) elif response.is_transport_error or response.is_server_error: logger.warning("Stream commit failed (retryable): id=%s", self._reservation_id) - self._retry_engine.schedule(self._reservation_id, commit_body) + self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) else: error_code = None error_resp = response.get_error_response() if error_resp and error_resp.error_code: error_code = error_resp.error_code.value - if error_code in ("RESERVATION_FINALIZED", "RESERVATION_EXPIRED"): - logger.warning("Reservation already finalized/expired: id=%s", self._reservation_id) + if response.status == 429 or error_code == "LIMIT_EXCEEDED": + # Rate-limited, not rejected: releasing here would return + # budget for spend that already happened. Retry instead, + # 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, + retry_after_ms=response.retry_after_ms_header, + ) + elif response.status in (401, 403): + # Credentials failed after the spend happened: journal the + # commit for replay once they're fixed. Never release — + # 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, + ) + self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) + elif response.status == 410 or error_code == "RESERVATION_EXPIRED": + logger.warning( + "Reservation expired before commit; recovering spend via POST /v1/events: id=%s", + self._reservation_id, + ) + self._retry_engine.schedule_event(self._reservation_id, event_fallback) + elif error_code == "RESERVATION_FINALIZED": + logger.warning("Reservation already finalized: id=%s", self._reservation_id) elif error_code == "IDEMPOTENCY_MISMATCH": logger.warning("Commit idempotency mismatch (not releasing): id=%s", self._reservation_id) - elif response.is_client_error: + elif response.is_client_error and _is_recognized_rejection(error_code): self._handle_release(f"commit_rejected_{error_code}") + elif response.is_client_error: + # Codeless or forward-compat-unknown 4xx: neither release + # nor drop — retain the spend record. + logger.error( + "Stream commit got unclassifiable client error (status=%d, error=%s); " + "journaling for replay: id=%s", + response.status, error_code, self._reservation_id, + ) + self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) else: logger.warning("Unrecognized commit response: id=%s", self._reservation_id) except Exception: logger.exception("Failed to commit stream: id=%s", self._reservation_id) - self._retry_engine.schedule(self._reservation_id, commit_body) + self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) def _handle_release(self, reason: str) -> None: assert self._reservation_id is not None @@ -497,29 +541,70 @@ async def _handle_commit(self) -> None: commit_body = _build_commit_body(actual, unit, metrics, self._metadata) assert self._reservation_id is not None + event_fallback = _build_event_fallback_body( + self._reservation_id, + self._subject.model_dump(exclude_none=True), + self._action.model_dump(exclude_none=True), + commit_body, + ) try: response = await self._client.commit_reservation(self._reservation_id, commit_body) if response.is_success: logger.info("Async stream commit successful: id=%s", self._reservation_id) elif response.is_transport_error or response.is_server_error: logger.warning("Async stream commit failed (retryable): id=%s", self._reservation_id) - self._retry_engine.schedule(self._reservation_id, commit_body) + self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) else: error_code = None error_resp = response.get_error_response() if error_resp and error_resp.error_code: error_code = error_resp.error_code.value - if error_code in ("RESERVATION_FINALIZED", "RESERVATION_EXPIRED"): - logger.warning("Reservation already finalized/expired: id=%s", self._reservation_id) + if response.status == 429 or error_code == "LIMIT_EXCEEDED": + # 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 + ) + self._retry_engine.schedule( + self._reservation_id, commit_body, event_fallback, + retry_after_ms=response.retry_after_ms_header, + ) + elif response.status in (401, 403): + # Credentials failed after the spend happened: journal the + # commit for replay once they're fixed. Never release — + # 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, + ) + self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) + elif response.status == 410 or error_code == "RESERVATION_EXPIRED": + logger.warning( + "Reservation expired before commit; recovering spend via POST /v1/events: id=%s", + self._reservation_id, + ) + self._retry_engine.schedule_event(self._reservation_id, event_fallback) + elif error_code == "RESERVATION_FINALIZED": + logger.warning("Reservation already finalized: id=%s", self._reservation_id) elif error_code == "IDEMPOTENCY_MISMATCH": logger.warning("Commit idempotency mismatch (not releasing): id=%s", self._reservation_id) - elif response.is_client_error: + elif response.is_client_error and _is_recognized_rejection(error_code): await self._handle_release(f"commit_rejected_{error_code}") + elif response.is_client_error: + # Codeless or forward-compat-unknown 4xx: neither release + # nor drop — retain the spend record. + 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, + ) + self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) else: logger.warning("Unrecognized commit response: id=%s", self._reservation_id) except Exception: logger.exception("Failed to commit async stream: id=%s", self._reservation_id) - self._retry_engine.schedule(self._reservation_id, commit_body) + self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) async def _handle_release(self, reason: str) -> None: assert self._reservation_id is not None diff --git a/tests/conftest.py b/tests/conftest.py index 40fd0ad..b194297 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,7 @@ """Shared test fixtures.""" +from pathlib import Path + import pytest @@ -9,3 +11,18 @@ def _reset_default_client() -> None: import runcycles.decorator as dec dec._default_client = None dec._default_config = None + + +@pytest.fixture(autouse=True) +def _isolate_commit_journal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Point the default commit journal at a per-test temp dir and reset replay state. + + Without this, any engine built from a default CyclesConfig would write + journal files into the real ``~/.runcycles`` during tests. + """ + import runcycles.journal as journal_mod + import runcycles.retry as retry_mod + + monkeypatch.setattr(journal_mod, "default_journal_dir", lambda: tmp_path / "commit-journal") + with retry_mod._replay_lock: + retry_mod._replayed_dirs.clear() diff --git a/tests/test_journal.py b/tests/test_journal.py new file mode 100644 index 0000000..d30ee44 --- /dev/null +++ b/tests/test_journal.py @@ -0,0 +1,1517 @@ +"""Tests for the durable commit journal, retry-engine durability, and event fallback.""" + +from __future__ import annotations + +import asyncio +import time +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from runcycles.config import CyclesConfig +from runcycles.journal import ( + CommitJournal, + PendingCommitRecord, + _safe_filename, + auth_fingerprint, + default_journal_dir, +) +from runcycles.lifecycle import ( + AsyncCyclesLifecycle, + CyclesLifecycle, + DecoratorConfig, + _build_event_fallback_body, +) +from runcycles.response import CyclesResponse +from runcycles.retry import ( + AsyncCommitRetryEngine, + CommitRetryEngine, + _extract_error_code, + _PendingCommit, +) + +BASE_URL = "http://localhost:7878" + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _config(tmp_path: Path, **overrides: Any) -> CyclesConfig: + defaults: dict[str, Any] = dict( + base_url=BASE_URL, + api_key="test-key", + retry_enabled=True, + retry_max_attempts=3, + retry_initial_delay=0.001, + retry_multiplier=1.0, + retry_max_delay=0.005, + retry_flush_timeout=5.0, + journal_enabled=True, + journal_dir=str(tmp_path / "journal"), + ) + defaults.update(overrides) + return CyclesConfig(**defaults) + + +def _commit_body() -> dict[str, Any]: + return {"idempotency_key": "ck-1", "actual": {"unit": "USD_MICROCENTS", "amount": 100}} + + +def _event_body() -> dict[str, Any]: + return { + "idempotency_key": "ck-1", + "subject": {"tenant": "acme"}, + "action": {"kind": "llm.completion", "name": "gpt"}, + "actual": {"unit": "USD_MICROCENTS", "amount": 100}, + } + + +def _expired_response() -> CyclesResponse: + return CyclesResponse.http_error( + 410, "Expired", + body={"error": "RESERVATION_EXPIRED", "message": "Expired", "request_id": "r1"}, + ) + + +def _finalized_response() -> CyclesResponse: + return CyclesResponse.http_error( + 409, "Finalized", + body={"error": "RESERVATION_FINALIZED", "message": "Finalized", "request_id": "r2"}, + ) + + +def _event_success() -> CyclesResponse: + return CyclesResponse.success(201, {"status": "APPLIED", "event_id": "evt_1"}) + + +def _commit_success() -> CyclesResponse: + return CyclesResponse.success(200, {"status": "COMMITTED"}) + + +def _record(reservation_id: str = "rsv_1", **overrides: Any) -> PendingCommitRecord: + defaults: dict[str, Any] = dict( + reservation_id=reservation_id, + base_url=BASE_URL, + mode="commit", + commit_body=_commit_body(), + event_fallback_body=_event_body(), + ) + defaults.update(overrides) + return PendingCommitRecord(**defaults) + + +def _identity_dir( + tmp_path: Path, api_key: str = "test-key", base_url: str = BASE_URL, tenant: str | None = None, +) -> Path: + """The per-identity subdirectory an engine with these credentials uses.""" + return tmp_path / "journal" / auth_fingerprint(base_url, api_key, tenant) + + +def _journal_files(tmp_path: Path) -> list[Path]: + d = tmp_path / "journal" + return sorted(d.rglob("*.json")) if d.is_dir() else [] + + +# --------------------------------------------------------------------------- +# CommitJournal +# --------------------------------------------------------------------------- + + +class TestCommitJournal: + def test_record_load_discard_roundtrip(self, tmp_path: Path) -> None: + journal = CommitJournal(tmp_path / "j") + journal.record(_record("rsv_a")) + + loaded = journal.load_pending(BASE_URL) + assert len(loaded) == 1 + entry = loaded[0] + assert entry.reservation_id == "rsv_a" + assert entry.mode == "commit" + assert entry.commit_body == _commit_body() + assert entry.event_fallback_body == _event_body() + assert entry.recorded_at_ms > 0 + + journal.discard("rsv_a") + assert journal.load_pending(BASE_URL) == [] + + def test_record_overwrites_same_reservation(self, tmp_path: Path) -> None: + journal = CommitJournal(tmp_path / "j") + journal.record(_record("rsv_a")) + journal.record(_record("rsv_a", mode="event")) + loaded = journal.load_pending(BASE_URL) + assert len(loaded) == 1 + assert loaded[0].mode == "event" + + def test_load_filters_by_base_url(self, tmp_path: Path) -> None: + journal = CommitJournal(tmp_path / "j") + journal.record(_record("rsv_a")) + journal.record(_record("rsv_b", base_url="http://other:9999")) + loaded = journal.load_pending(BASE_URL) + assert [e.reservation_id for e in loaded] == ["rsv_a"] + + def test_load_missing_dir_returns_empty(self, tmp_path: Path) -> None: + journal = CommitJournal(tmp_path / "does-not-exist") + assert journal.load_pending(BASE_URL) == [] + + def test_corrupt_file_renamed_and_skipped(self, tmp_path: Path) -> None: + directory = tmp_path / "j" + journal = CommitJournal(directory) + journal.record(_record("rsv_good")) + (directory / "rsv_bad.json").write_text("{not json", encoding="utf-8") + + loaded = journal.load_pending(BASE_URL) + assert [e.reservation_id for e in loaded] == ["rsv_good"] + assert (directory / "rsv_bad.corrupt").exists() + assert not (directory / "rsv_bad.json").exists() + + def test_semantically_invalid_records_are_corrupt(self, tmp_path: Path) -> None: + directory = tmp_path / "j" + directory.mkdir(parents=True) + cases = { + "no_rid.json": '{"reservation_id": "", "mode": "commit", "commit_body": {}}', + "bad_mode.json": '{"reservation_id": "r1", "mode": "sideways", "commit_body": {}}', + "commit_no_body.json": '{"reservation_id": "r2", "mode": "commit"}', + "event_no_body.json": '{"reservation_id": "r3", "mode": "event", "commit_body": {}}', + } + for name, content in cases.items(): + (directory / name).write_text(content, encoding="utf-8") + + journal = CommitJournal(directory) + assert journal.load_pending(BASE_URL) == [] + assert len(list(directory.glob("*.corrupt"))) == len(cases) + + def test_record_swallows_os_errors(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + journal = CommitJournal(tmp_path / "j") + monkeypatch.setattr(Path, "mkdir", MagicMock(side_effect=OSError("disk full"))) + journal.record(_record("rsv_a")) # must not raise + assert journal.load_pending(BASE_URL) == [] + + def test_discard_swallows_os_errors(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + journal = CommitJournal(tmp_path / "j") + journal.record(_record("rsv_a")) + monkeypatch.setattr(Path, "unlink", MagicMock(side_effect=OSError("locked"))) + journal.discard("rsv_a") # must not raise + + def test_load_swallows_scan_errors(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + journal = CommitJournal(tmp_path / "j") + journal.record(_record("rsv_a")) + monkeypatch.setattr(Path, "glob", MagicMock(side_effect=OSError("io error"))) + assert journal.load_pending(BASE_URL) == [] + + def test_corrupt_rename_failure_is_swallowed(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + directory = tmp_path / "j" + directory.mkdir(parents=True) + (directory / "rsv_bad.json").write_text("{not json", encoding="utf-8") + monkeypatch.setattr(Path, "replace", MagicMock(side_effect=OSError("locked"))) + + journal = CommitJournal(directory) + assert journal.load_pending(BASE_URL) == [] # skipped, no raise + + def test_safe_filename_sanitizes(self) -> None: + assert _safe_filename("rsv_abc-123") == "rsv_abc-123.json" + assert _safe_filename("rsv/../etc") == "rsv____etc.json" + + def test_safe_filename_is_ascii_only(self) -> None: + # Cross-SDK invariant: TS/Java sanitize with [^A-Za-z0-9_-]; a + # Unicode-alphanumeric-preserving Python name would never be + # discardable by a sibling SDK sharing the identity directory. + assert _safe_filename("rsvé٣x") == "rsv__x.json" + + def test_stale_temp_files_are_reaped_on_load(self, tmp_path: Path) -> None: + import os + + directory = tmp_path / "j" + journal = CommitJournal(directory) + journal.record(_record("rsv_a")) + + stale = directory / "rsv_x.json.999.deadbeef.tmp" + stale.write_text("{partial", encoding="utf-8") + two_hours_ago = time.time() - 7200 + os.utime(stale, (two_hours_ago, two_hours_ago)) + fresh = directory / "rsv_y.json.999.cafecafe.tmp" + fresh.write_text("{partial", encoding="utf-8") + + journal.load_pending(BASE_URL) + + assert not stale.exists() # reaped + assert fresh.exists() # a live writer's temp is left alone + + def test_temp_reap_failure_is_swallowed( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, + ) -> None: + import os + + directory = tmp_path / "j" + journal = CommitJournal(directory) + journal.record(_record("rsv_a")) + stale = directory / "rsv_x.json.999.deadbeef.tmp" + stale.write_text("{partial", encoding="utf-8") + two_hours_ago = time.time() - 7200 + os.utime(stale, (two_hours_ago, two_hours_ago)) + monkeypatch.setattr(Path, "unlink", MagicMock(side_effect=OSError("locked"))) + + assert [e.reservation_id for e in journal.load_pending(BASE_URL)] == ["rsv_a"] + + def test_blank_tenant_treated_as_absent(self) -> None: + # Matches Java's isBlank presence check — whitespace-only tenant + # falls back to the key principal so all SDKs share one identity. + assert auth_fingerprint(BASE_URL, "test-key", " ") == auth_fingerprint(BASE_URL, "test-key") + + def test_fingerprint_pins_cross_sdk_vectors(self) -> None: + # These exact values are asserted in the TS and Java SDK suites; a + # drift here breaks journal interop for every sibling SDK. + assert auth_fingerprint("http://localhost", "test-key") == "68c905017df7dbfc" + assert auth_fingerprint("http://localhost", "any-key", "acme") == "8baba538fb970da4" + + def test_default_journal_dir_under_home(self) -> None: + # Note: conftest patches the module attribute; this exercises the real function. + path = default_journal_dir() + assert path == Path.home() / ".runcycles" / "commit-journal" + + def test_auth_fingerprint_is_stable_and_identity_scoped(self) -> None: + fp = auth_fingerprint(BASE_URL, "test-key") + assert fp == auth_fingerprint(BASE_URL, "test-key") + assert len(fp) == 16 + assert all(c in "0123456789abcdef" for c in fp) + assert fp != auth_fingerprint(BASE_URL, "other-key") + assert fp != auth_fingerprint("http://other:9999", "test-key") + + def test_auth_fingerprint_tenant_is_rotation_safe(self) -> None: + # With a tenant, the principal is the tenant: rotating the API key + # keeps the same identity directory. + fp = auth_fingerprint(BASE_URL, "test-key", tenant="acme") + assert fp == auth_fingerprint(BASE_URL, "rotated-key", tenant="acme") + assert fp != auth_fingerprint(BASE_URL, "test-key", tenant="globex") + assert fp != auth_fingerprint(BASE_URL, "test-key") # key-based differs + assert fp != auth_fingerprint("http://other:9999", "test-key", tenant="acme") + + def test_journal_files_are_private(self, tmp_path: Path) -> None: + import os + + directory = tmp_path / "j" + journal = CommitJournal(directory) + journal.record(_record("rsv_a")) + + if os.name == "posix": + assert directory.stat().st_mode & 0o777 == 0o700 + assert (directory / "rsv_a.json").stat().st_mode & 0o777 == 0o600 + + def test_permission_tightening_failure_is_swallowed( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, + ) -> None: + monkeypatch.setattr(Path, "chmod", MagicMock(side_effect=OSError("not supported"))) + journal = CommitJournal(tmp_path / "j") + journal.record(_record("rsv_a")) # write must still succeed + assert [e.reservation_id for e in journal.load_pending(BASE_URL)] == ["rsv_a"] + + def test_failed_publish_cleans_up_temp_file( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, + ) -> None: + journal = CommitJournal(tmp_path / "j") + monkeypatch.setattr(Path, "replace", MagicMock(side_effect=OSError("locked"))) + journal.record(_record("rsv_a")) # must not raise + assert list((tmp_path / "j").glob("*.tmp")) == [] + assert list((tmp_path / "j").glob("*.json")) == [] + + def test_failed_publish_and_cleanup_never_raise( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, + ) -> None: + journal = CommitJournal(tmp_path / "j") + monkeypatch.setattr(Path, "replace", MagicMock(side_effect=OSError("locked"))) + monkeypatch.setattr(Path, "unlink", MagicMock(side_effect=OSError("also locked"))) + journal.record(_record("rsv_a")) # must not raise + + def test_stale_temp_files_are_ignored(self, tmp_path: Path) -> None: + directory = tmp_path / "j" + directory.mkdir(parents=True) + # A crashed writer's leftover: partial JSON under a unique temp name. + (directory / "rsv_x.json.999.deadbeef.tmp").write_text("{partial", encoding="utf-8") + + journal = CommitJournal(directory) + journal.record(_record("rsv_a")) + assert [e.reservation_id for e in journal.load_pending(BASE_URL)] == ["rsv_a"] + + +# --------------------------------------------------------------------------- +# _extract_error_code +# --------------------------------------------------------------------------- + + +class TestExtractErrorCode: + def test_from_error_response(self) -> None: + assert _extract_error_code(_expired_response()) == "RESERVATION_EXPIRED" + + def test_from_raw_body(self) -> None: + response = CyclesResponse.http_error(400, "Bad", body={"error": "SOMETHING_ODD"}) + assert _extract_error_code(response) == "SOMETHING_ODD" + + def test_none_when_absent(self) -> None: + assert _extract_error_code(CyclesResponse.http_error(400, "Bad")) is None + + +# --------------------------------------------------------------------------- +# CommitRetryEngine durability +# --------------------------------------------------------------------------- + + +class TestSyncEngineDurability: + def test_schedule_journals_then_success_discards(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.commit_reservation.return_value = _commit_success() + engine.set_client(mock_client) + + engine.schedule("rsv_1", _commit_body(), _event_body()) + engine.flush(timeout=5.0) + + assert mock_client.commit_reservation.call_count == 1 + assert _journal_files(tmp_path) == [] + + def test_expired_falls_back_to_event(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.commit_reservation.return_value = _expired_response() + mock_client.create_event.return_value = _event_success() + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", _commit_body(), _event_body()) + engine._journal_record(pending) + engine._retry_loop(pending) + + assert mock_client.commit_reservation.call_count == 1 + mock_client.create_event.assert_called_once_with(_event_body()) + assert _journal_files(tmp_path) == [] + + def test_expired_without_fallback_retains_journal(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.commit_reservation.return_value = _expired_response() + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", _commit_body()) + engine._journal_record(pending) + engine._retry_loop(pending) + + assert mock_client.commit_reservation.call_count == 1 + mock_client.create_event.assert_not_called() + assert len(_journal_files(tmp_path)) == 1 + + def test_finalized_discards_journal(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.commit_reservation.return_value = _finalized_response() + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", _commit_body(), _event_body()) + engine._journal_record(pending) + engine._retry_loop(pending) + + assert _journal_files(tmp_path) == [] + + def test_exhausted_retains_journal(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(500, "boom") + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", _commit_body(), _event_body()) + engine._journal_record(pending) + engine._retry_loop(pending) + + assert mock_client.commit_reservation.call_count == 3 + assert len(_journal_files(tmp_path)) == 1 + + def test_schedule_event_posts_event(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.create_event.return_value = _event_success() + engine.set_client(mock_client) + + engine.schedule_event("rsv_1", _event_body()) + engine.flush(timeout=5.0) + + mock_client.create_event.assert_called_once_with(_event_body()) + mock_client.commit_reservation.assert_not_called() + assert _journal_files(tmp_path) == [] + + def test_event_client_error_discards_journal(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.create_event.return_value = CyclesResponse.http_error( + 409, "Mismatch", body={"error": "IDEMPOTENCY_MISMATCH", "message": "m", "request_id": "r"}, + ) + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", None, _event_body(), mode="event") + engine._journal_record(pending) + engine._retry_loop(pending) + + assert mock_client.create_event.call_count == 1 + assert _journal_files(tmp_path) == [] + + def test_event_transient_then_success(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.create_event.side_effect = [ + CyclesResponse.http_error(500, "boom"), + _event_success(), + ] + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", None, _event_body(), mode="event") + engine._retry_loop(pending) + + assert mock_client.create_event.call_count == 2 + + def test_expired_then_event_transient_continues_in_event_mode(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.commit_reservation.return_value = _expired_response() + mock_client.create_event.side_effect = [ + CyclesResponse.http_error(503, "unavailable"), + _event_success(), + ] + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", _commit_body(), _event_body()) + engine._retry_loop(pending) + + # One commit attempt, then immediate event attempt, then one retried event attempt. + assert mock_client.commit_reservation.call_count == 1 + assert mock_client.create_event.call_count == 2 + + def test_disabled_with_journal_persists_entry(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path, retry_enabled=False)) + mock_client = MagicMock() + engine.set_client(mock_client) + + engine.schedule("rsv_1", _commit_body(), _event_body()) + + mock_client.commit_reservation.assert_not_called() + assert len(_journal_files(tmp_path)) == 1 + + def test_disabled_without_journal_drops(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path, retry_enabled=False, journal_enabled=False)) + mock_client = MagicMock() + engine.set_client(mock_client) + + engine.schedule("rsv_1", _commit_body()) + + mock_client.commit_reservation.assert_not_called() + assert _journal_files(tmp_path) == [] + + def test_flush_zero_timeout_returns_immediately(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path, retry_flush_timeout=0.0)) + engine.flush() # must not raise or block + + def test_flush_gives_up_at_deadline(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + + def _slow_commit(*args: Any, **kwargs: Any) -> CyclesResponse: + time.sleep(0.2) + return _commit_success() + + mock_client.commit_reservation.side_effect = _slow_commit + engine.set_client(mock_client) + + engine.schedule("rsv_1", _commit_body()) + engine.schedule("rsv_2", _commit_body()) + # First join consumes the whole budget; the second iteration hits the deadline. + engine.flush(timeout=0.05) + engine.flush(timeout=5.0) # clean up before the test ends + + def test_atexit_hook_flushes_registered_engines(self, tmp_path: Path) -> None: + import runcycles.retry as retry_mod + + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.commit_reservation.return_value = _commit_success() + engine.set_client(mock_client) + + engine.schedule("rsv_1", _commit_body()) + retry_mod._flush_all_engines() # what atexit runs at interpreter exit + + assert mock_client.commit_reservation.call_count == 1 + assert _journal_files(tmp_path) == [] + + def test_flush_all_engines_with_no_engines(self) -> None: + import weakref + + import runcycles.retry as retry_mod + + original = retry_mod._live_engines + retry_mod._live_engines = weakref.WeakSet() + try: + retry_mod._flush_all_engines() # must not raise + finally: + retry_mod._live_engines = original + + def test_flush_all_engines_shares_one_deadline(self, tmp_path: Path) -> None: + # Finding 4: exit flush is bounded by retry_flush_timeout for the + # whole process, not per engine. + import weakref + + import runcycles.retry as retry_mod + + config = _config(tmp_path, retry_flush_timeout=0.5) + + def _slow_commit(*args: Any, **kwargs: Any) -> CyclesResponse: + time.sleep(2.0) + return _commit_success() + + engines = [] + for i in range(2): + engine = CommitRetryEngine(config) + mock_client = MagicMock() + mock_client.commit_reservation.side_effect = _slow_commit + engine.set_client(mock_client) + engine.schedule(f"rsv_{i}", _commit_body()) + engines.append(engine) + + isolated: weakref.WeakSet[CommitRetryEngine] = weakref.WeakSet(engines) + original = retry_mod._live_engines + retry_mod._live_engines = isolated + try: + start = time.monotonic() + retry_mod._flush_all_engines() + elapsed = time.monotonic() - start + finally: + retry_mod._live_engines = original + + # One shared 0.5s budget — sequential per-engine flushes would take ~1.0s+. + assert elapsed < 0.9 + + +class TestRateLimitedRetry: + def test_429_commit_is_transient_and_honors_retry_after(self, tmp_path: Path) -> None: + # Finding 2: a rate-limited commit must keep its journal entry and + # keep retrying, waiting at least the server's Retry-After. + engine = CommitRetryEngine(_config(tmp_path)) + response = CyclesResponse.http_error( + 429, "Rate limited", + body={"error": "LIMIT_EXCEEDED", "message": "slow down", "request_id": "r9"}, + headers={"retry-after": "2"}, + ) + pending = _PendingCommit("rsv_1", _commit_body(), _event_body()) + engine._journal_record(pending) + + assert engine._classify_commit_response(pending, response) is False + assert pending.retry_after_s == 2.0 + assert len(_journal_files(tmp_path)) == 1 # retained, not discarded + + # The Retry-After floor is re-journaled as an absolute timestamp so + # a restart mid-wait does not replay into the server's window. + entry = CommitJournal(_identity_dir(tmp_path)).load_pending(BASE_URL)[0] + now_ms = time.time() * 1000 + assert entry.not_before_ms is not None + assert now_ms < entry.not_before_ms <= now_ms + 2500 + + delay = engine._delay_for(pending) + assert delay >= 2.0 # server's Retry-After wins over backoff + assert pending.retry_after_s is None # consumed — applies once + + def test_429_without_body_detected_by_status(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + pending = _PendingCommit("rsv_1", _commit_body()) + assert engine._classify_commit_response(pending, CyclesResponse.http_error(429, "busy")) is False + assert pending.retry_after_s is None # no header → plain backoff + + def test_429_then_success_discards_journal(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.commit_reservation.side_effect = [ + CyclesResponse.http_error(429, "busy", headers={"retry-after": "0"}), + _commit_success(), + ] + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", _commit_body()) + engine._journal_record(pending) + engine._retry_loop(pending) + + assert mock_client.commit_reservation.call_count == 2 + assert _journal_files(tmp_path) == [] + + def test_429_event_fallback_is_transient(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.create_event.return_value = CyclesResponse.http_error( + 429, "busy", body={"error": "LIMIT_EXCEEDED", "message": "m", "request_id": "r"}, + ) + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", None, _event_body(), mode="event") + engine._journal_record(pending) + engine._retry_loop(pending) + + # Exhausts attempts without ever discarding the durable record. + assert mock_client.create_event.call_count == 3 + assert len(_journal_files(tmp_path)) == 1 + + def test_schedule_seeds_retry_after(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + engine.set_client(MagicMock()) + captured: list[_PendingCommit] = [] + monkeypatch.setattr(engine, "_spawn", captured.append) + + engine.schedule("rsv_1", _commit_body(), _event_body(), retry_after_ms=1500) + + assert captured[0].retry_after_s == 1.5 + + def test_retry_after_floor_is_persisted(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + pending = _PendingCommit("rsv_1", _commit_body(), retry_after_s=60.0) + engine._journal_record(pending) + + entry = CommitJournal(_identity_dir(tmp_path)).load_pending(BASE_URL)[0] + now_ms = time.time() * 1000 + assert entry.not_before_ms is not None + assert now_ms + 55_000 < entry.not_before_ms <= now_ms + 60_500 + + def test_replay_restores_future_retry_after_floor( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, + ) -> None: + future_ms = int(time.time() * 1000 + 5000) + CommitJournal(_identity_dir(tmp_path)).record(_record("rsv_old", not_before_ms=future_ms)) + + engine = CommitRetryEngine(_config(tmp_path)) + captured: list[_PendingCommit] = [] + monkeypatch.setattr(engine, "_spawn", captured.append) + engine.set_client(MagicMock()) + + assert captured[0].retry_after_s is not None + assert 3.0 < captured[0].retry_after_s <= 5.0 + + def test_replay_ignores_past_retry_after_floor( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, + ) -> None: + past_ms = int(time.time() * 1000 - 1000) + CommitJournal(_identity_dir(tmp_path)).record(_record("rsv_old", not_before_ms=past_ms)) + + engine = CommitRetryEngine(_config(tmp_path)) + captured: list[_PendingCommit] = [] + monkeypatch.setattr(engine, "_spawn", captured.append) + engine.set_client(MagicMock()) + + assert captured[0].retry_after_s is None # falls back to normal backoff + + +class TestAuthFailureRetention: + def test_401_commit_retains_journal(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(401, "Unauthorized") + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", _commit_body(), _event_body()) + engine._journal_record(pending) + engine._retry_loop(pending) + + # Terminal for this run, but the durable record survives for replay + # once credentials are fixed. + assert mock_client.commit_reservation.call_count == 1 + assert len(_journal_files(tmp_path)) == 1 + + def test_403_event_fallback_retains_journal(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.create_event.return_value = CyclesResponse.http_error(403, "Forbidden") + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", None, _event_body(), mode="event") + engine._journal_record(pending) + engine._retry_loop(pending) + + assert mock_client.create_event.call_count == 1 + assert len(_journal_files(tmp_path)) == 1 + + def test_unclassifiable_commit_4xx_retains_journal(self, tmp_path: Path) -> None: + # A codeless 4xx (proxy junk) or a forward-compat unknown code must + # neither release nor destroy the durable record. + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.commit_reservation.side_effect = [ + CyclesResponse.http_error(400, "proxy junk"), + ] + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", _commit_body(), _event_body()) + engine._journal_record(pending) + engine._retry_loop(pending) + + assert mock_client.commit_reservation.call_count == 1 + assert len(_journal_files(tmp_path)) == 1 + + mock_client2 = MagicMock() + mock_client2.commit_reservation.return_value = CyclesResponse.http_error( + 422, "future", body={"error": "FUTURE_REJECTION_CODE", "message": "m", "request_id": "r"}, + ) + engine2 = CommitRetryEngine(_config(tmp_path)) + engine2.set_client(mock_client2) + pending2 = _PendingCommit("rsv_2", _commit_body(), _event_body()) + engine2._journal_record(pending2) + engine2._retry_loop(pending2) + assert len(_journal_files(tmp_path)) == 2 # both retained + + def test_unclassifiable_event_4xx_retains_journal(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.create_event.return_value = CyclesResponse.http_error(400, "proxy junk") + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", None, _event_body(), mode="event") + engine._journal_record(pending) + engine._retry_loop(pending) + + assert mock_client.create_event.call_count == 1 + assert len(_journal_files(tmp_path)) == 1 + + def test_bodyless_410_triggers_event_fallback(self, tmp_path: Path) -> None: + # The 429 path is status-hardened; a mangled 410 body must still + # reach the event fallback rather than the terminal branches. + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(410, "gone") + mock_client.create_event.return_value = _event_success() + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", _commit_body(), _event_body()) + engine._journal_record(pending) + engine._retry_loop(pending) + + mock_client.create_event.assert_called_once_with(_event_body()) + assert _journal_files(tmp_path) == [] + + def test_honored_retry_after_is_clamped(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + engine = CommitRetryEngine(_config(tmp_path)) + response = CyclesResponse.http_error( + 429, "busy", + body={"error": "LIMIT_EXCEEDED", "message": "m", "request_id": "r"}, + headers={"retry-after": "7200"}, + ) + pending = _PendingCommit("rsv_1", _commit_body()) + assert engine._classify_commit_response(pending, response) is False + assert pending.retry_after_s == 3600.0 # 2h clamped to the 1h ceiling + + captured: list[_PendingCommit] = [] + monkeypatch.setattr(engine, "_spawn", captured.append) + engine.set_client(MagicMock()) + engine.schedule("rsv_2", _commit_body(), retry_after_ms=7_200_000) + # captured[-1]: set_client's replay may capture the journaled rsv_1 first + assert captured[-1].retry_after_s == 3600.0 + + def test_restored_replay_floor_is_clamped( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, + ) -> None: + far_future = int(time.time() * 1000 + 7_200_000) + CommitJournal(_identity_dir(tmp_path)).record(_record("rsv_old", not_before_ms=far_future)) + + engine = CommitRetryEngine(_config(tmp_path)) + captured: list[_PendingCommit] = [] + monkeypatch.setattr(engine, "_spawn", captured.append) + engine.set_client(MagicMock()) + + assert captured[0].retry_after_s == 3600.0 + + def test_replay_survives_api_key_rotation_with_tenant(self, tmp_path: Path) -> None: + # Records written under the old key are found by the rotated key + # because the identity is the tenant, not the credential. + CommitJournal(_identity_dir(tmp_path, api_key="old-key", tenant="acme")).record(_record("rsv_old")) + + engine = CommitRetryEngine(_config(tmp_path, api_key="rotated-key", tenant="acme")) + mock_client = MagicMock() + mock_client.commit_reservation.return_value = _commit_success() + engine.set_client(mock_client) + engine.flush(timeout=5.0) + + mock_client.commit_reservation.assert_called_once_with("rsv_old", _commit_body()) + assert _journal_files(tmp_path) == [] + + +# --------------------------------------------------------------------------- +# Journal replay +# --------------------------------------------------------------------------- + + +class TestSyncReplay: + def test_replays_pending_commit_on_set_client(self, tmp_path: Path) -> None: + CommitJournal(_identity_dir(tmp_path)).record(_record("rsv_old")) + + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.commit_reservation.return_value = _commit_success() + engine.set_client(mock_client) + engine.flush(timeout=5.0) + + mock_client.commit_reservation.assert_called_once_with("rsv_old", _commit_body()) + assert _journal_files(tmp_path) == [] + + def test_replays_event_mode_entry(self, tmp_path: Path) -> None: + CommitJournal(_identity_dir(tmp_path)).record( + _record("rsv_old", mode="event", commit_body=None) + ) + + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + mock_client.create_event.return_value = _event_success() + engine.set_client(mock_client) + engine.flush(timeout=5.0) + + mock_client.create_event.assert_called_once_with(_event_body()) + assert _journal_files(tmp_path) == [] + + def test_replay_happens_once_per_directory(self, tmp_path: Path) -> None: + CommitJournal(_identity_dir(tmp_path)).record(_record("rsv_old")) + config = _config(tmp_path) + + first = CommitRetryEngine(config) + client1 = MagicMock() + client1.commit_reservation.return_value = _commit_success() + first.set_client(client1) + first.flush(timeout=5.0) + + second = CommitRetryEngine(config) + client2 = MagicMock() + second.set_client(client2) + second.flush(timeout=5.0) + + assert client1.commit_reservation.call_count == 1 + client2.commit_reservation.assert_not_called() + + def test_replay_skips_other_server_entries(self, tmp_path: Path) -> None: + # Defense-in-depth: a mismatched-base_url record inside the identity + # dir (should not happen — the fingerprint includes base_url) is + # still filtered out rather than replayed against the wrong server. + journal = CommitJournal(_identity_dir(tmp_path)) + journal.record(_record("rsv_other", base_url="http://other:9999")) + + engine = CommitRetryEngine(_config(tmp_path)) + mock_client = MagicMock() + engine.set_client(mock_client) + engine.flush(timeout=5.0) + + mock_client.commit_reservation.assert_not_called() + assert len(_journal_files(tmp_path)) == 1 # left in place, never discarded + + def test_replay_isolated_by_api_key(self, tmp_path: Path) -> None: + # Finding 1: same server, different credentials → separate identity + # dirs. Client A must never replay (and 401-discard) client B's spend. + CommitJournal(_identity_dir(tmp_path, api_key="test-key")).record(_record("rsv_a")) + CommitJournal(_identity_dir(tmp_path, api_key="other-key")).record(_record("rsv_b")) + + engine_a = CommitRetryEngine(_config(tmp_path)) + client_a = MagicMock() + client_a.commit_reservation.return_value = _commit_success() + engine_a.set_client(client_a) + engine_a.flush(timeout=5.0) + + client_a.commit_reservation.assert_called_once_with("rsv_a", _commit_body()) + assert _journal_files(tmp_path) == [_identity_dir(tmp_path, api_key="other-key") / "rsv_b.json"] + + engine_b = CommitRetryEngine(_config(tmp_path, api_key="other-key")) + client_b = MagicMock() + client_b.commit_reservation.return_value = _commit_success() + engine_b.set_client(client_b) + engine_b.flush(timeout=5.0) + + client_b.commit_reservation.assert_called_once_with("rsv_b", _commit_body()) + assert _journal_files(tmp_path) == [] + + def test_one_server_claim_does_not_block_another(self, tmp_path: Path) -> None: + # Finding 3: the replay claim is scoped to the identity subdirectory, + # so server A's engine starting first cannot starve server B's entries. + other_url = "http://other:9999" + CommitJournal(_identity_dir(tmp_path, base_url=other_url)).record( + _record("rsv_b", base_url=other_url) + ) + + engine_a = CommitRetryEngine(_config(tmp_path)) + client_a = MagicMock() + engine_a.set_client(client_a) # claims A's (empty) identity dir first + engine_a.flush(timeout=5.0) + client_a.commit_reservation.assert_not_called() + + engine_b = CommitRetryEngine(_config(tmp_path, base_url=other_url)) + client_b = MagicMock() + client_b.commit_reservation.return_value = _commit_success() + engine_b.set_client(client_b) + engine_b.flush(timeout=5.0) + + client_b.commit_reservation.assert_called_once_with("rsv_b", _commit_body()) + assert _journal_files(tmp_path) == [] + + def test_no_replay_when_retry_disabled(self, tmp_path: Path) -> None: + CommitJournal(_identity_dir(tmp_path)).record(_record("rsv_old")) + + engine = CommitRetryEngine(_config(tmp_path, retry_enabled=False)) + mock_client = MagicMock() + engine.set_client(mock_client) + + mock_client.commit_reservation.assert_not_called() + assert len(_journal_files(tmp_path)) == 1 + + def test_no_replay_when_journal_disabled(self, tmp_path: Path) -> None: + engine = CommitRetryEngine(_config(tmp_path, journal_enabled=False)) + mock_client = MagicMock() + engine.set_client(mock_client) + mock_client.commit_reservation.assert_not_called() + + +# --------------------------------------------------------------------------- +# AsyncCommitRetryEngine durability +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +class TestAsyncEngineDurability: + async def test_schedule_holds_task_reference_and_discards_on_success(self, tmp_path: Path) -> None: + engine = AsyncCommitRetryEngine(_config(tmp_path)) + mock_client = AsyncMock() + mock_client.commit_reservation.return_value = _commit_success() + engine.set_client(mock_client) + + engine.schedule("rsv_1", _commit_body(), _event_body()) + assert len(engine._tasks) == 1 # reference held → cannot be garbage-collected + await engine.flush(timeout=5.0) + + assert mock_client.commit_reservation.call_count == 1 + assert _journal_files(tmp_path) == [] + assert engine._tasks == set() + + async def test_expired_falls_back_to_event(self, tmp_path: Path) -> None: + engine = AsyncCommitRetryEngine(_config(tmp_path)) + mock_client = AsyncMock() + mock_client.commit_reservation.return_value = _expired_response() + mock_client.create_event.return_value = _event_success() + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", _commit_body(), _event_body()) + engine._journal_record(pending) + await engine._retry_loop(pending) + + mock_client.create_event.assert_awaited_once_with(_event_body()) + assert _journal_files(tmp_path) == [] + + async def test_expired_without_fallback_retains_journal(self, tmp_path: Path) -> None: + engine = AsyncCommitRetryEngine(_config(tmp_path)) + mock_client = AsyncMock() + mock_client.commit_reservation.return_value = _expired_response() + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", _commit_body()) + engine._journal_record(pending) + await engine._retry_loop(pending) + + assert len(_journal_files(tmp_path)) == 1 + + async def test_schedule_event_posts_event(self, tmp_path: Path) -> None: + engine = AsyncCommitRetryEngine(_config(tmp_path)) + mock_client = AsyncMock() + mock_client.create_event.return_value = _event_success() + engine.set_client(mock_client) + + engine.schedule_event("rsv_1", _event_body()) + await engine.flush(timeout=5.0) + + mock_client.create_event.assert_awaited_once_with(_event_body()) + assert _journal_files(tmp_path) == [] + + async def test_exhausted_retains_journal(self, tmp_path: Path) -> None: + engine = AsyncCommitRetryEngine(_config(tmp_path)) + mock_client = AsyncMock() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(500, "boom") + engine.set_client(mock_client) + + pending = _PendingCommit("rsv_1", _commit_body(), _event_body()) + engine._journal_record(pending) + await engine._retry_loop(pending) + + assert mock_client.commit_reservation.await_count == 3 + assert len(_journal_files(tmp_path)) == 1 + + async def test_disabled_with_journal_persists_entry(self, tmp_path: Path) -> None: + engine = AsyncCommitRetryEngine(_config(tmp_path, retry_enabled=False)) + mock_client = AsyncMock() + engine.set_client(mock_client) + + engine.schedule("rsv_1", _commit_body()) + + mock_client.commit_reservation.assert_not_called() + assert len(_journal_files(tmp_path)) == 1 + + async def test_replay_on_set_client_with_running_loop(self, tmp_path: Path) -> None: + CommitJournal(_identity_dir(tmp_path)).record(_record("rsv_old")) + + engine = AsyncCommitRetryEngine(_config(tmp_path)) + mock_client = AsyncMock() + mock_client.commit_reservation.return_value = _commit_success() + engine.set_client(mock_client) + await engine.flush(timeout=5.0) + + mock_client.commit_reservation.assert_awaited_once_with("rsv_old", _commit_body()) + assert _journal_files(tmp_path) == [] + + async def test_flush_zero_timeout_returns_immediately(self, tmp_path: Path) -> None: + engine = AsyncCommitRetryEngine(_config(tmp_path, retry_flush_timeout=0.0)) + await engine.flush() # must not raise or block + + async def test_schedule_seeds_retry_after( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, + ) -> None: + engine = AsyncCommitRetryEngine(_config(tmp_path)) + engine.set_client(AsyncMock()) + captured: list[_PendingCommit] = [] + monkeypatch.setattr(engine, "_spawn", lambda loop, pending: captured.append(pending)) + + engine.schedule("rsv_1", _commit_body(), _event_body(), retry_after_ms=1500) + + assert captured[0].retry_after_s == 1.5 + + +class TestAsyncEngineNoLoop: + def test_schedule_without_loop_keeps_journal_entry(self, tmp_path: Path) -> None: + engine = AsyncCommitRetryEngine(_config(tmp_path)) + engine.set_client(AsyncMock()) + + engine.schedule("rsv_1", _commit_body(), _event_body()) + + # Could not spawn a task, but the entry survives for the next run. + assert len(_journal_files(tmp_path)) == 1 + + def test_schedule_without_loop_and_journal_drops(self, tmp_path: Path) -> None: + engine = AsyncCommitRetryEngine(_config(tmp_path, journal_enabled=False)) + engine.set_client(AsyncMock()) + engine.schedule("rsv_1", _commit_body()) # must not raise + assert _journal_files(tmp_path) == [] + + def test_deferred_replay_runs_at_first_schedule(self, tmp_path: Path) -> None: + CommitJournal(_identity_dir(tmp_path)).record(_record("rsv_old")) + + engine = AsyncCommitRetryEngine(_config(tmp_path)) + mock_client = AsyncMock() + mock_client.commit_reservation.return_value = _commit_success() + engine.set_client(mock_client) # no loop yet → replay deferred + assert engine._replay_deferred is True + + async def _run() -> None: + engine.schedule("rsv_new", _commit_body(), _event_body()) + await engine.flush(timeout=5.0) + + asyncio.run(_run()) + + committed_ids = {call.args[0] for call in mock_client.commit_reservation.await_args_list} + assert committed_ids == {"rsv_old", "rsv_new"} + assert _journal_files(tmp_path) == [] + + +# --------------------------------------------------------------------------- +# Event fallback body construction +# --------------------------------------------------------------------------- + + +class TestBuildEventFallbackBody: + def test_builds_spec_shape_reusing_commit_idempotency_key(self) -> None: + commit_body = { + "idempotency_key": "ck-9", + "actual": {"unit": "USD_MICROCENTS", "amount": 250}, + "metrics": {"latency_ms": 12}, + "metadata": {"run": "abc"}, + } + body = _build_event_fallback_body( + "rsv_9", {"tenant": "acme"}, {"kind": "llm.completion", "name": "gpt"}, commit_body, + ) + + assert body["idempotency_key"] == "ck-9" + assert body["subject"] == {"tenant": "acme"} + assert body["action"] == {"kind": "llm.completion", "name": "gpt"} + assert body["actual"] == {"unit": "USD_MICROCENTS", "amount": 250} + assert body["metrics"] == {"latency_ms": 12} + assert body["metadata"]["run"] == "abc" + assert body["metadata"]["recovered_reservation_id"] == "rsv_9" + assert body["metadata"]["recovery_reason"] == "commit_after_reservation_expired" + assert "overage_policy" not in body # server default ALLOW_IF_AVAILABLE never rejects + + def test_without_metrics_or_metadata(self) -> None: + body = _build_event_fallback_body( + "rsv_9", {"tenant": "acme"}, {"kind": "k", "name": "n"}, _commit_body(), + ) + assert "metrics" not in body + assert set(body["metadata"]) == {"recovered_reservation_id", "recovery_reason"} + + +# --------------------------------------------------------------------------- +# Lifecycle wiring: expired commit → schedule_event, transient → fallback passed +# --------------------------------------------------------------------------- + + +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 _make_cfg() -> DecoratorConfig: + return DecoratorConfig(estimate=1000, tenant="acme", ttl_ms=60000) + + +class TestLifecycleEventFallbackWiring: + def _make(self, tmp_path: Path) -> tuple[CyclesLifecycle, MagicMock, MagicMock]: + mock_client = MagicMock() + mock_client._config = _config(tmp_path) + engine = MagicMock(spec=CommitRetryEngine) + lifecycle = CyclesLifecycle(mock_client, engine, {"tenant": "acme"}) + return lifecycle, mock_client, engine + + def test_expired_commit_schedules_event(self, tmp_path: Path) -> None: + lifecycle, mock_client, engine = self._make(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = _expired_response() + + lifecycle.execute(lambda: "result", (), {}, _make_cfg()) + + engine.schedule_event.assert_called_once() + rid, event_body = engine.schedule_event.call_args.args + assert rid == "rsv_test" + assert event_body["subject"] == {"tenant": "acme"} + assert event_body["metadata"]["recovered_reservation_id"] == "rsv_test" + mock_client.release_reservation.assert_not_called() + + def test_transient_commit_passes_event_fallback(self, tmp_path: Path) -> None: + lifecycle, mock_client, engine = self._make(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(500, "boom") + + lifecycle.execute(lambda: "result", (), {}, _make_cfg()) + + engine.schedule.assert_called_once() + args = engine.schedule.call_args.args + assert args[0] == "rsv_test" + assert args[2]["metadata"]["recovered_reservation_id"] == "rsv_test" + + def test_finalized_commit_does_not_schedule_event(self, tmp_path: Path) -> None: + lifecycle, mock_client, engine = self._make(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = _finalized_response() + + lifecycle.execute(lambda: "result", (), {}, _make_cfg()) + + engine.schedule_event.assert_not_called() + engine.schedule.assert_not_called() + + def test_unclassifiable_4xx_schedules_not_release(self, tmp_path: Path) -> None: + lifecycle, mock_client, engine = self._make(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(400, "proxy junk") + + lifecycle.execute(lambda: "result", (), {}, _make_cfg()) + + engine.schedule.assert_called_once() + mock_client.release_reservation.assert_not_called() + + def test_bodyless_410_schedules_event(self, tmp_path: Path) -> None: + lifecycle, mock_client, engine = self._make(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(410, "gone") + + lifecycle.execute(lambda: "result", (), {}, _make_cfg()) + + engine.schedule_event.assert_called_once() + mock_client.release_reservation.assert_not_called() + + def test_rate_limited_first_commit_schedules_retry_not_release(self, tmp_path: Path) -> None: + # A 429 on the initial commit must never release the reservation — + # that would return budget for spend that already happened. + lifecycle, mock_client, engine = self._make(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error( + 429, "busy", + body={"error": "LIMIT_EXCEEDED", "message": "slow down", "request_id": "r9"}, + headers={"retry-after": "3"}, + ) + + lifecycle.execute(lambda: "result", (), {}, _make_cfg()) + + engine.schedule.assert_called_once() + assert engine.schedule.call_args.kwargs["retry_after_ms"] == 3000 + mock_client.release_reservation.assert_not_called() + + def test_auth_failure_first_commit_journals_not_release(self, tmp_path: Path) -> None: + # Credentials expiring between reserve and commit must journal the + # spend, never release it. + lifecycle, mock_client, engine = self._make(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(401, "Unauthorized") + + lifecycle.execute(lambda: "result", (), {}, _make_cfg()) + + engine.schedule.assert_called_once() + assert engine.schedule.call_args.args[0] == "rsv_test" + mock_client.release_reservation.assert_not_called() + + +@pytest.mark.asyncio +class TestAsyncLifecycleEventFallbackWiring: + def _make(self, tmp_path: Path) -> tuple[AsyncCyclesLifecycle, AsyncMock, MagicMock]: + mock_client = AsyncMock() + mock_client._config = _config(tmp_path) + engine = MagicMock(spec=AsyncCommitRetryEngine) + lifecycle = AsyncCyclesLifecycle(mock_client, engine, {"tenant": "acme"}) + return lifecycle, mock_client, engine + + async def test_expired_commit_schedules_event(self, tmp_path: Path) -> None: + lifecycle, mock_client, engine = self._make(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = _expired_response() + + async def fn() -> str: + return "result" + + await lifecycle.execute(fn, (), {}, _make_cfg()) + + engine.schedule_event.assert_called_once() + rid, event_body = engine.schedule_event.call_args.args + assert rid == "rsv_test" + assert event_body["metadata"]["recovery_reason"] == "commit_after_reservation_expired" + + async def test_transient_commit_passes_event_fallback(self, tmp_path: Path) -> None: + lifecycle, mock_client, engine = self._make(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(500, "boom") + + async def fn() -> str: + return "result" + + await lifecycle.execute(fn, (), {}, _make_cfg()) + + engine.schedule.assert_called_once() + args = engine.schedule.call_args.args + assert args[2]["metadata"]["recovered_reservation_id"] == "rsv_test" + + async def test_rate_limited_first_commit_schedules_retry_not_release(self, tmp_path: Path) -> None: + lifecycle, mock_client, engine = self._make(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error( + 429, "busy", + body={"error": "LIMIT_EXCEEDED", "message": "slow down", "request_id": "r9"}, + headers={"retry-after": "3"}, + ) + + async def fn() -> str: + return "result" + + await lifecycle.execute(fn, (), {}, _make_cfg()) + + engine.schedule.assert_called_once() + assert engine.schedule.call_args.kwargs["retry_after_ms"] == 3000 + mock_client.release_reservation.assert_not_called() + + async def test_auth_failure_first_commit_journals_not_release(self, tmp_path: Path) -> None: + lifecycle, mock_client, engine = self._make(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(403, "Forbidden") + + async def fn() -> str: + return "result" + + await lifecycle.execute(fn, (), {}, _make_cfg()) + + engine.schedule.assert_called_once() + assert engine.schedule.call_args.args[0] == "rsv_test" + mock_client.release_reservation.assert_not_called() + + async def test_unclassifiable_4xx_schedules_not_release(self, tmp_path: Path) -> None: + lifecycle, mock_client, engine = self._make(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(400, "proxy junk") + + async def fn() -> str: + return "result" + + await lifecycle.execute(fn, (), {}, _make_cfg()) + + engine.schedule.assert_called_once() + mock_client.release_reservation.assert_not_called() + + +# --------------------------------------------------------------------------- +# Streaming wiring: expired commit → schedule_event +# --------------------------------------------------------------------------- + + +class TestStreamingEventFallbackWiring: + def _make_stream(self, tmp_path: Path) -> tuple[Any, MagicMock, MagicMock]: + from runcycles.client import CyclesClient + from runcycles.models import Action, Amount, Subject, Unit + from runcycles.streaming import StreamReservation + + mock_client = MagicMock(spec=CyclesClient) + mock_client._config = _config(tmp_path) + stream = StreamReservation( + mock_client, + subject=Subject(tenant="acme"), + action=Action(kind="llm.completion", name="gpt"), + estimate=Amount(unit=Unit.USD_MICROCENTS, amount=1000), + ttl_ms=60_000, + ) + engine = MagicMock(spec=CommitRetryEngine) + stream._retry_engine = engine + return stream, mock_client, engine + + def test_expired_commit_schedules_event(self, tmp_path: Path) -> None: + stream, mock_client, engine = self._make_stream(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = _expired_response() + + with stream: + pass + + engine.schedule_event.assert_called_once() + rid, event_body = engine.schedule_event.call_args.args + assert rid == "rsv_test" + assert event_body["subject"] == {"tenant": "acme"} + assert event_body["metadata"]["recovered_reservation_id"] == "rsv_test" + + def test_transient_commit_passes_event_fallback(self, tmp_path: Path) -> None: + stream, mock_client, engine = self._make_stream(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(500, "boom") + + with stream: + pass + + engine.schedule.assert_called_once() + args = engine.schedule.call_args.args + assert args[2]["metadata"]["recovered_reservation_id"] == "rsv_test" + + def test_rate_limited_first_commit_schedules_retry_not_release(self, tmp_path: Path) -> None: + stream, mock_client, engine = self._make_stream(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error( + 429, "busy", + body={"error": "LIMIT_EXCEEDED", "message": "slow down", "request_id": "r9"}, + headers={"retry-after": "3"}, + ) + + with stream: + pass + + engine.schedule.assert_called_once() + assert engine.schedule.call_args.kwargs["retry_after_ms"] == 3000 + mock_client.release_reservation.assert_not_called() + + def test_auth_failure_first_commit_journals_not_release(self, tmp_path: Path) -> None: + stream, mock_client, engine = self._make_stream(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(401, "Unauthorized") + + with stream: + pass + + engine.schedule.assert_called_once() + assert engine.schedule.call_args.args[0] == "rsv_test" + mock_client.release_reservation.assert_not_called() + + def test_unclassifiable_4xx_schedules_not_release(self, tmp_path: Path) -> None: + stream, mock_client, engine = self._make_stream(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(400, "proxy junk") + + with stream: + pass + + engine.schedule.assert_called_once() + mock_client.release_reservation.assert_not_called() + + +@pytest.mark.asyncio +class TestAsyncStreamingEventFallbackWiring: + async def _make_stream(self, tmp_path: Path) -> tuple[Any, AsyncMock, MagicMock]: + from runcycles.client import AsyncCyclesClient + from runcycles.models import Action, Amount, Subject, Unit + from runcycles.streaming import AsyncStreamReservation + + mock_client = AsyncMock(spec=AsyncCyclesClient) + mock_client._config = _config(tmp_path) + stream = AsyncStreamReservation( + mock_client, + subject=Subject(tenant="acme"), + action=Action(kind="llm.completion", name="gpt"), + estimate=Amount(unit=Unit.USD_MICROCENTS, amount=1000), + ttl_ms=60_000, + ) + engine = MagicMock(spec=AsyncCommitRetryEngine) + stream._retry_engine = engine + return stream, mock_client, engine + + async def test_expired_commit_schedules_event(self, tmp_path: Path) -> None: + stream, mock_client, engine = await self._make_stream(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = _expired_response() + + async with stream: + pass + + engine.schedule_event.assert_called_once() + rid, event_body = engine.schedule_event.call_args.args + assert rid == "rsv_test" + assert event_body["metadata"]["recovery_reason"] == "commit_after_reservation_expired" + + async def test_transient_commit_passes_event_fallback(self, tmp_path: Path) -> None: + stream, mock_client, engine = await self._make_stream(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(500, "boom") + + async with stream: + pass + + engine.schedule.assert_called_once() + args = engine.schedule.call_args.args + assert args[2]["metadata"]["recovered_reservation_id"] == "rsv_test" + + async def test_rate_limited_first_commit_schedules_retry_not_release(self, tmp_path: Path) -> None: + stream, mock_client, engine = await self._make_stream(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error( + 429, "busy", + body={"error": "LIMIT_EXCEEDED", "message": "slow down", "request_id": "r9"}, + headers={"retry-after": "3"}, + ) + + async with stream: + pass + + engine.schedule.assert_called_once() + assert engine.schedule.call_args.kwargs["retry_after_ms"] == 3000 + mock_client.release_reservation.assert_not_called() + + async def test_auth_failure_first_commit_journals_not_release(self, tmp_path: Path) -> None: + stream, mock_client, engine = await self._make_stream(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(403, "Forbidden") + + async with stream: + pass + + engine.schedule.assert_called_once() + assert engine.schedule.call_args.args[0] == "rsv_test" + mock_client.release_reservation.assert_not_called() + + async def test_unclassifiable_4xx_schedules_not_release(self, tmp_path: Path) -> None: + stream, mock_client, engine = await self._make_stream(tmp_path) + mock_client.create_reservation.return_value = _allow_response() + mock_client.commit_reservation.return_value = CyclesResponse.http_error(400, "proxy junk") + + async with stream: + pass + + engine.schedule.assert_called_once() + mock_client.release_reservation.assert_not_called() diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 88d7578..b1ad873 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -568,7 +568,7 @@ def test_commit_client_error_triggers_release(self) -> None: mock.commit_reservation.return_value = CyclesResponse.http_error( 400, "Bad request", - body={"error": "VALIDATION_ERROR", "message": "Bad", "request_id": "r1"}, + body={"error": "INVALID_REQUEST", "message": "Bad", "request_id": "r1"}, ) mock.release_reservation.return_value = _release_success() @@ -876,7 +876,7 @@ async def test_commit_client_error_triggers_release(self) -> None: mock.commit_reservation.return_value = CyclesResponse.http_error( 400, "Bad request", - body={"error": "VALIDATION_ERROR", "message": "Bad", "request_id": "r1"}, + body={"error": "INVALID_REQUEST", "message": "Bad", "request_id": "r1"}, ) mock.release_reservation.return_value = _release_success()