From a9d41b4c0f9f6c2eb49c47f949bba7a75975e6e9 Mon Sep 17 00:00:00 2001 From: Albert Mavashev Date: Wed, 29 Jul 2026 08:19:28 -0400 Subject: [PATCH 1/3] fix(recovery): guarantee durable settlement --- .github/workflows/ci.yml | 21 +++++ .github/workflows/python-publish.yml | 27 +++++- CHANGELOG.md | 20 ++++ pyproject.toml | 2 +- runcycles/journal.py | 51 +++++++++- runcycles/lifecycle.py | 73 +++++++++++++-- runcycles/models.py | 1 + runcycles/retry.py | 67 +++++++++++-- runcycles/streaming.py | 71 ++++++++++++-- scripts/recovery_conformance_adapter.py | 119 ++++++++++++++++++++++++ tests/test_journal.py | 86 ++++++++++++++--- tests/test_retry.py | 84 +++++++++++++++-- 12 files changed, 575 insertions(+), 47 deletions(-) create mode 100644 scripts/recovery_conformance_adapter.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a2ef9d4..5c548a7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,3 +14,24 @@ jobs: uses: runcycles/.github/.github/workflows/ci-python.yml@v1 with: mypy-target: runcycles + + recovery-conformance: + name: Durable recovery conformance + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + repository: runcycles/cycles-protocol + ref: 8a22803316d91b85e444f4fb40ff2b8426f11773 + path: .cycles-protocol + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.12" + - run: python -m pip install -e ".[dev]" + - name: Run shared durable recovery scenarios + run: >- + python .cycles-protocol/scripts/run_client_recovery_conformance.py + --claim durable + --adapter python scripts/recovery_conformance_adapter.py diff --git a/.github/workflows/python-publish.yml b/.github/workflows/python-publish.yml index 383c15f..cbc0d4b 100644 --- a/.github/workflows/python-publish.yml +++ b/.github/workflows/python-publish.yml @@ -20,6 +20,27 @@ permissions: contents: read jobs: + recovery-conformance: + name: Durable recovery conformance + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + repository: runcycles/cycles-protocol + ref: 8a22803316d91b85e444f4fb40ff2b8426f11773 + path: .cycles-protocol + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.12" + - run: python -m pip install -e ".[dev]" + - name: Run shared durable recovery scenarios + run: >- + python .cycles-protocol/scripts/run_client_recovery_conformance.py + --claim durable + --adapter python scripts/recovery_conformance_adapter.py + build: name: Build distributions runs-on: ubuntu-latest @@ -64,7 +85,7 @@ jobs: publish-to-testpypi: name: Publish to TestPyPI - needs: build + needs: [build, recovery-conformance] runs-on: ubuntu-latest if: github.event_name == 'workflow_dispatch' && inputs.target == 'testpypi' environment: @@ -88,7 +109,7 @@ jobs: publish-to-pypi: name: Publish to PyPI - needs: build + needs: [build, recovery-conformance] runs-on: ubuntu-latest if: startsWith(github.ref, 'refs/tags/v') || (github.event_name == 'workflow_dispatch' && inputs.target == 'pypi') environment: @@ -148,4 +169,4 @@ jobs: name: ${{ github.ref_name }} body: ${{ steps.notes.outputs.notes }} draft: false - prerelease: ${{ contains(github.ref_name, '-') }} \ No newline at end of file + prerelease: ${{ contains(github.ref_name, '-') }} diff --git a/CHANGELOG.md b/CHANGELOG.md index 3d551cd..c5f72f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +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). +## [0.5.2] - 2026-07-29 + +### Added + +- Bind all shared durable-recovery and guarantee-boundary scenarios from the + protocol repository into pull-request and release CI. +- Expose `cycles_evidence` on `CommitResponse`. + +### Fixed + +- Persist known actual usage before the first commit request, recover expired + commits through `/v1/events`, and accept only exact HTTP 200/201 + schema-valid commit/event responses as terminal success. +- Use `v2-.json` journal filenames, safely + migrate matching legacy records, and preserve collision-free cross-SDK + replay. +- Retain durable settlement records for contradictory retryable 4xx envelopes, + and report heartbeat transport failures with their same-key retry or stop + disposition. + ## [0.5.1] - 2026-07-27 ### Fixed diff --git a/pyproject.toml b/pyproject.toml index 3b5bfb3..6154a47 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "runcycles" -version = "0.5.1" +version = "0.5.2" 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/journal.py b/runcycles/journal.py index 9f0ab5d..33a3981 100644 --- a/runcycles/journal.py +++ b/runcycles/journal.py @@ -89,9 +89,13 @@ def _restrict_permissions(path: Path, mode: int) -> None: 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. + """Cross-SDK, collision-resistant filename for an exact reservation id.""" + digest = hashlib.sha256(reservation_id.encode("utf-8")).hexdigest() + return f"v2-{digest}{_SUFFIX}" + + +def _legacy_filename(reservation_id: str) -> str: + """Filename written by SDK releases before the v2 digest scheme.""" sanitized = re.sub(r"[^A-Za-z0-9_-]", "_", reservation_id) return f"{sanitized}{_SUFFIX}" @@ -200,6 +204,15 @@ 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) + legacy = self._dir / _legacy_filename(reservation_id) + if legacy.exists(): + try: + entry = PendingCommitRecord.from_json(legacy.read_text(encoding="utf-8")) + if entry.reservation_id == reservation_id: + legacy.unlink(missing_ok=True) + except (OSError, ValueError, KeyError, json.JSONDecodeError): + # Never delete a colliding or malformed legacy record. + pass except OSError: logger.warning("Failed to discard journal entry: id=%s", reservation_id, exc_info=True) @@ -235,6 +248,38 @@ def load_pending(self, base_url: str) -> list[PendingCommitRecord]: except OSError: pass continue + standard_path = self._dir / _safe_filename(entry.reservation_id) + duplicate_of_standard = False + if path != standard_path: + try: + if not standard_path.exists(): + path.replace(standard_path) + logger.info( + "Migrated legacy journal filename: id=%s, path=%s", + entry.reservation_id, + standard_path, + ) + else: + existing = PendingCommitRecord.from_json( + standard_path.read_text(encoding="utf-8") + ) + if existing.reservation_id == entry.reservation_id: + path.unlink(missing_ok=True) + duplicate_of_standard = True + logger.info( + "Removed duplicate legacy journal filename: id=%s, path=%s", + entry.reservation_id, + path, + ) + except (OSError, ValueError, KeyError, json.JSONDecodeError): + logger.warning( + "Could not safely migrate legacy journal filename: id=%s, path=%s", + entry.reservation_id, + path, + exc_info=True, + ) + if duplicate_of_standard: + continue if entry.base_url == base_url: entries.append(entry) except OSError: diff --git a/runcycles/lifecycle.py b/runcycles/lifecycle.py index f7701fc..be7639f 100644 --- a/runcycles/lifecycle.py +++ b/runcycles/lifecycle.py @@ -46,6 +46,7 @@ CommitRetryEngine, _extract_error_code, _is_recognized_rejection, + _is_schema_valid_commit_success, ) logger = logging.getLogger(__name__) @@ -763,11 +764,23 @@ def _handle_commit( commit_body: dict[str, Any], event_fallback_body: dict[str, Any], ) -> None: + self._retry_engine.persist_pending( + reservation_id, commit_body, event_fallback_body + ) try: logger.debug("Committing: id=%s", reservation_id) response = self._client.commit_reservation(reservation_id, commit_body) - if response.is_success: + if _is_schema_valid_commit_success(response): + self._retry_engine.discard_pending(reservation_id) logger.info("Commit successful: id=%s", reservation_id) + elif response.is_success: + logger.warning( + "Commit returned ambiguous protocol-invalid 2xx; scheduling same-key retry: " + "id=%s, status=%d", + reservation_id, + response.status, + ) + self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) 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, event_fallback_body) @@ -804,10 +817,13 @@ def _handle_commit( ) self._retry_engine.schedule_event(reservation_id, event_fallback_body) elif error_code == "RESERVATION_FINALIZED": + self._retry_engine.discard_pending(reservation_id) logger.warning("Reservation already finalized: id=%s", reservation_id) elif error_code == "IDEMPOTENCY_MISMATCH": + self._retry_engine.discard_pending(reservation_id) logger.warning("Commit idempotency mismatch (not releasing): id=%s", reservation_id) elif response.is_client_error and _is_recognized_rejection(error_code): + self._retry_engine.discard_pending(reservation_id) self._handle_release(reservation_id, f"commit_rejected_{error_code}") elif response.is_client_error: # Codeless or forward-compat-unknown 4xx: neither release @@ -820,7 +836,12 @@ def _handle_commit( ) self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) else: - logger.warning("Unrecognized commit response: id=%s, response=%s", reservation_id, response) + logger.warning( + "Unrecognized commit response; scheduling same-key retry: id=%s, response=%s", + reservation_id, + response, + ) + self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) except Exception: logger.exception("Failed to commit: id=%s", reservation_id) self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) @@ -1047,16 +1068,24 @@ def heartbeat_loop() -> None: return delay_ms = nxt except Exception: - logger.warning("Heartbeat extend error: id=%s", reservation_id, exc_info=True) if authoritative: nxt = sched.on_transient_failure(_now_mono_ms()) if nxt is None: logger.warning( - "Heartbeat stopping: no safe recovery window remains: id=%s", + "Heartbeat extend transport error; stopping because no safe recovery " + "window remains: id=%s", reservation_id, + exc_info=True, ) return delay_ms = nxt + logger.warning( + "Heartbeat extend transport error; retrying with the same idempotency key " + "in %.0fms: id=%s", + delay_ms, + reservation_id, + exc_info=True, + ) t = threading.Thread(target=heartbeat_loop, daemon=True, name=f"cycles-heartbeat-{reservation_id[:12]}") t.start() @@ -1189,10 +1218,22 @@ async def _handle_commit( commit_body: dict[str, Any], event_fallback_body: dict[str, Any], ) -> None: + self._retry_engine.persist_pending( + reservation_id, commit_body, event_fallback_body + ) try: response = await self._client.commit_reservation(reservation_id, commit_body) - if response.is_success: + if _is_schema_valid_commit_success(response): + self._retry_engine.discard_pending(reservation_id) logger.info("Commit successful: id=%s", reservation_id) + elif response.is_success: + logger.warning( + "Commit returned ambiguous protocol-invalid 2xx; scheduling same-key retry: " + "id=%s, status=%d", + reservation_id, + response.status, + ) + self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) elif response.is_transport_error or response.is_server_error: self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) else: @@ -1228,10 +1269,13 @@ async def _handle_commit( ) self._retry_engine.schedule_event(reservation_id, event_fallback_body) elif error_code == "RESERVATION_FINALIZED": + self._retry_engine.discard_pending(reservation_id) logger.warning("Reservation already finalized: id=%s", reservation_id) elif error_code == "IDEMPOTENCY_MISMATCH": + self._retry_engine.discard_pending(reservation_id) logger.warning("Commit idempotency mismatch (not releasing): id=%s", reservation_id) elif response.is_client_error and _is_recognized_rejection(error_code): + self._retry_engine.discard_pending(reservation_id) await self._handle_release(reservation_id, f"commit_rejected_{error_code}") elif response.is_client_error: # Codeless or forward-compat-unknown 4xx: neither release @@ -1244,7 +1288,12 @@ async def _handle_commit( ) self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) else: - logger.warning("Unrecognized commit response: id=%s, response=%s", reservation_id, response) + logger.warning( + "Unrecognized commit response; scheduling same-key retry: id=%s, response=%s", + reservation_id, + response, + ) + self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) except Exception: logger.exception("Failed to commit: id=%s", reservation_id) self._retry_engine.schedule(reservation_id, commit_body, event_fallback_body) @@ -1457,16 +1506,24 @@ async def heartbeat_loop() -> None: return delay_ms = nxt except Exception: - logger.warning("Heartbeat extend error: id=%s", reservation_id, exc_info=True) if authoritative: nxt = sched.on_transient_failure(_now_mono_ms()) if nxt is None: logger.warning( - "Heartbeat stopping: no safe recovery window remains: id=%s", + "Heartbeat extend transport error; stopping because no safe " + "recovery window remains: id=%s", reservation_id, + exc_info=True, ) return delay_ms = nxt + logger.warning( + "Heartbeat extend transport error; retrying with the same idempotency " + "key in %.0fms: id=%s", + delay_ms, + reservation_id, + exc_info=True, + ) except asyncio.CancelledError: return diff --git a/runcycles/models.py b/runcycles/models.py index a485593..2aee215 100644 --- a/runcycles/models.py +++ b/runcycles/models.py @@ -309,6 +309,7 @@ class CommitResponse(BaseModel): charged: Amount released: Amount | None = None balances: list[Balance] | None = None + cycles_evidence: CyclesEvidenceRef | None = None class ReleaseResponse(BaseModel): diff --git a/runcycles/retry.py b/runcycles/retry.py index d3665bb..46c059d 100644 --- a/runcycles/retry.py +++ b/runcycles/retry.py @@ -14,6 +14,7 @@ import asyncio import atexit +import json import logging import threading import time @@ -25,7 +26,7 @@ from runcycles import journal as _journal from runcycles.config import CyclesConfig from runcycles.journal import CommitJournal, PendingCommitRecord -from runcycles.models import ErrorCode +from runcycles.models import CommitResponse, ErrorCode, EventCreateResponse from runcycles.response import CyclesResponse logger = logging.getLogger(__name__) @@ -37,13 +38,50 @@ def _is_recognized_rejection(code: str | None) -> bool: - """True when the error code is a known protocol code (not forward-compat). + """True when the error code proves a known, non-retryable rejection. 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. + or releasing a reservation; a codeless, retryable, 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 + parsed = ErrorCode.from_string(code) + return ( + parsed is not None + and parsed is not ErrorCode.UNKNOWN + and not parsed.is_retryable + ) + + +def _is_schema_valid_commit_success(response: CyclesResponse) -> bool: + """Only the protocol's exact commit success is terminal.""" + if response.status != 200 or not isinstance(response.body, dict): + return False + if any( + key in response.body and response.body[key] is None + for key in ("released", "balances", "cycles_evidence") + ): + return False + try: + CommitResponse.model_validate_json(json.dumps(response.body), strict=True) + except Exception: + return False + return True + + +def _is_schema_valid_event_success(response: CyclesResponse) -> bool: + """Only the protocol's exact event success is terminal.""" + if response.status != 201 or not isinstance(response.body, dict): + return False + if any( + key in response.body and response.body[key] is None + for key in ("charged", "balances") + ): + return False + try: + EventCreateResponse.model_validate_json(json.dumps(response.body), strict=True) + except Exception: + return False + return True @dataclass @@ -160,6 +198,21 @@ def _journal_discard(self, reservation_id: str) -> None: if self._journal is not None: self._journal.discard(reservation_id) + def persist_pending( + self, + reservation_id: str, + commit_body: dict[str, Any], + event_fallback_body: dict[str, Any] | None = None, + ) -> None: + """Journal known actual spend before the first settlement request.""" + self._journal_record( + _PendingCommit(reservation_id, commit_body, event_fallback_body, mode="commit") + ) + + def discard_pending(self, reservation_id: str) -> None: + """Remove a pre-journaled settlement after a terminal outcome.""" + 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: @@ -233,7 +286,7 @@ def _classify_commit_response(self, pending: _PendingCommit, response: CyclesRes May flip ``pending`` into event mode; the caller then delivers the event fallback immediately (no extra backoff) via ``_attempt_event``. """ - if response.is_success: + if _is_schema_valid_commit_success(response): logger.info( "Commit retry succeeded: reservation_id=%s, attempt=%d", pending.reservation_id, pending.attempt, @@ -291,7 +344,7 @@ def _classify_commit_response(self, pending: _PendingCommit, response: CyclesRes 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: + if _is_schema_valid_event_success(response): logger.info( "Recovered expired-commit spend via /v1/events: reservation_id=%s, event_id=%s", pending.reservation_id, response.get_body_attribute("event_id"), diff --git a/runcycles/streaming.py b/runcycles/streaming.py index 706cc6d..25049c2 100644 --- a/runcycles/streaming.py +++ b/runcycles/streaming.py @@ -46,6 +46,7 @@ CommitRetryEngine, _extract_error_code, _is_recognized_rejection, + _is_schema_valid_commit_success, ) logger = logging.getLogger(__name__) @@ -310,10 +311,22 @@ def _handle_commit(self) -> None: self._action.model_dump(exclude_none=True), commit_body, ) + self._retry_engine.persist_pending( + self._reservation_id, commit_body, event_fallback + ) try: response = self._client.commit_reservation(self._reservation_id, commit_body) - if response.is_success: + if _is_schema_valid_commit_success(response): + self._retry_engine.discard_pending(self._reservation_id) logger.info("Stream commit successful: id=%s", self._reservation_id) + elif response.is_success: + logger.warning( + "Stream commit returned ambiguous protocol-invalid 2xx; " + "scheduling same-key retry: id=%s, status=%d", + self._reservation_id, + response.status, + ) + self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) 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, event_fallback) @@ -350,10 +363,13 @@ def _handle_commit(self) -> None: ) self._retry_engine.schedule_event(self._reservation_id, event_fallback) elif error_code == "RESERVATION_FINALIZED": + self._retry_engine.discard_pending(self._reservation_id) logger.warning("Reservation already finalized: id=%s", self._reservation_id) elif error_code == "IDEMPOTENCY_MISMATCH": + self._retry_engine.discard_pending(self._reservation_id) logger.warning("Commit idempotency mismatch (not releasing): id=%s", self._reservation_id) elif response.is_client_error and _is_recognized_rejection(error_code): + self._retry_engine.discard_pending(self._reservation_id) self._handle_release(f"commit_rejected_{error_code}") elif response.is_client_error: # Codeless or forward-compat-unknown 4xx: neither release @@ -367,7 +383,11 @@ def _handle_commit(self) -> None: ) self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) else: - logger.warning("Unrecognized commit response: id=%s", self._reservation_id) + logger.warning( + "Unrecognized commit response; scheduling same-key retry: id=%s", + self._reservation_id, + ) + self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) except Exception: logger.exception("Failed to commit stream: id=%s", self._reservation_id) self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) @@ -576,16 +596,24 @@ def heartbeat_loop() -> None: return delay_ms = nxt except Exception: - logger.warning("Stream heartbeat error: id=%s", reservation_id, exc_info=True) if authoritative: nxt = sched.on_transient_failure(_lifecycle._now_mono_ms()) if nxt is None: logger.warning( - "Heartbeat stopping: no safe recovery window remains: id=%s", + "Stream heartbeat transport error; stopping because no safe recovery " + "window remains: id=%s", reservation_id, + exc_info=True, ) return delay_ms = nxt + logger.warning( + "Stream heartbeat transport error; retrying with the same idempotency key " + "in %.0fms: id=%s", + delay_ms, + reservation_id, + exc_info=True, + ) t = threading.Thread( target=heartbeat_loop, @@ -764,10 +792,22 @@ async def _handle_commit(self) -> None: self._action.model_dump(exclude_none=True), commit_body, ) + self._retry_engine.persist_pending( + self._reservation_id, commit_body, event_fallback + ) try: response = await self._client.commit_reservation(self._reservation_id, commit_body) - if response.is_success: + if _is_schema_valid_commit_success(response): + self._retry_engine.discard_pending(self._reservation_id) logger.info("Async stream commit successful: id=%s", self._reservation_id) + elif response.is_success: + logger.warning( + "Async stream commit returned ambiguous protocol-invalid 2xx; " + "scheduling same-key retry: id=%s, status=%d", + self._reservation_id, + response.status, + ) + self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) 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, event_fallback) @@ -804,10 +844,13 @@ async def _handle_commit(self) -> None: ) self._retry_engine.schedule_event(self._reservation_id, event_fallback) elif error_code == "RESERVATION_FINALIZED": + self._retry_engine.discard_pending(self._reservation_id) logger.warning("Reservation already finalized: id=%s", self._reservation_id) elif error_code == "IDEMPOTENCY_MISMATCH": + self._retry_engine.discard_pending(self._reservation_id) logger.warning("Commit idempotency mismatch (not releasing): id=%s", self._reservation_id) elif response.is_client_error and _is_recognized_rejection(error_code): + self._retry_engine.discard_pending(self._reservation_id) await self._handle_release(f"commit_rejected_{error_code}") elif response.is_client_error: # Codeless or forward-compat-unknown 4xx: neither release @@ -821,7 +864,11 @@ async def _handle_commit(self) -> None: ) self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) else: - logger.warning("Unrecognized commit response: id=%s", self._reservation_id) + logger.warning( + "Unrecognized commit response; scheduling same-key retry: id=%s", + self._reservation_id, + ) + self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) except Exception: logger.exception("Failed to commit async stream: id=%s", self._reservation_id) self._retry_engine.schedule(self._reservation_id, commit_body, event_fallback) @@ -1034,16 +1081,24 @@ async def heartbeat_loop() -> None: return delay_ms = nxt except Exception: - logger.warning("Async stream heartbeat error: id=%s", reservation_id, exc_info=True) if authoritative: nxt = sched.on_transient_failure(_lifecycle._now_mono_ms()) if nxt is None: logger.warning( - "Heartbeat stopping: no safe recovery window remains: id=%s", + "Async stream heartbeat transport error; stopping because no safe " + "recovery window remains: id=%s", reservation_id, + exc_info=True, ) return delay_ms = nxt + logger.warning( + "Async stream heartbeat transport error; retrying with the same " + "idempotency key in %.0fms: id=%s", + delay_ms, + reservation_id, + exc_info=True, + ) except asyncio.CancelledError: return diff --git a/scripts/recovery_conformance_adapter.py b/scripts/recovery_conformance_adapter.py new file mode 100644 index 0000000..b75cbde --- /dev/null +++ b/scripts/recovery_conformance_adapter.py @@ -0,0 +1,119 @@ +#!/usr/bin/env python3 +"""Bind shared recovery scenario IDs to native Python SDK behavior tests.""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] + +OBSERVATIONS = { + "CR-CORE-001": (["commit", "commit_same_key"], [ + "settlement_occurs_at_most_once", "retry_uses_original_idempotency_key"]), + "CR-CORE-002": (["commit", "event_same_key"], [ + "event_carries_original_subject_action_actual", "settlement_occurs_at_most_once"]), + "CR-CORE-003": (["extend", "extend_same_key", "commit"], [ + "heartbeat_failure_reports_reservation_and_disposition", + "guarded_action_continues_under_warn_policy", "final_settlement_is_attempted"]), + "CR-CORE-004": (["commit", "commit_same_key"], [ + "only_schema_valid_expected_status_is_terminal_success", + "ambiguous_success_retains_original_idempotency_key"]), + "CR-DURABLE-001": (["commit", "commit_same_key_after_restart"], [ + "journal_write_precedes_first_settlement_request", + "unresolved_record_survives_restart", "successful_replay_removes_record", + "settlement_occurs_at_most_once"]), + "CR-DURABLE-002": (["commit_same_key_after_restart", "event_same_key_after_restart"], [ + "event_mode_is_persisted_before_event_attempt", "successful_event_removes_record", + "settlement_occurs_at_most_once"]), + "CR-DURABLE-003": (["commit", "commit_same_key_after_retry_after"], [ + "no_retry_before_persisted_not_before", "successful_replay_removes_record"]), + "CR-DURABLE-004": (["commit_same_key_after_restart"], [ + "new_tenant_credential_finds_record", "old_api_key_is_not_stored"]), + "CR-DURABLE-005": ([], [ + "corrupt_record_is_quarantined", "other_valid_records_still_replay", + "corruption_is_reported"]), + "CR-DURABLE-006": (["concurrent_commit_same_key"], [ + "settlement_occurs_at_most_once", "terminal_record_is_removed"]), + "CR-DURABLE-007": ([ + "commit_first_identifier", "commit_second_identifier", + "commit_first_identifier_same_key_after_restart", + "commit_second_identifier_same_key_after_restart", + ], [ + "standard_filename_is_sha256_of_exact_utf8_identifier", + "distinct_identifiers_never_share_a_journal_file", + "matching_legacy_record_migrates_without_deleting_collision", + "both_settlements_occur_at_most_once", + ]), + "CR-BOUNDARY-001": ([], [ + "sdk_does_not_claim_ledger_convergence", "application_checkpoint_is_required"]), +} + +TESTS = { + "CR-CORE-001": "tests/test_retry.py::TestCommitRetryEngine::test_retries_until_success", + "CR-CORE-002": "tests/test_journal.py::TestLifecycleEventFallbackWiring::test_expired_commit_schedules_event", + "CR-CORE-003": "tests/test_lifecycle.py::TestSyncLifecycleExecution::test_heartbeat_exception_does_not_crash", + "CR-CORE-004": ( + "tests/test_journal.py::TestLifecycleEventFallbackWiring::" + "test_protocol_invalid_2xx_is_ambiguous_and_keeps_same_key" + ), + "CR-DURABLE-001": ( + "tests/test_journal.py::TestLifecycleEventFallbackWiring::" + "test_journal_write_precedes_first_commit_and_success_discards" + ), + "CR-DURABLE-002": ( + "tests/test_journal.py::TestSyncEngineDurability::" + "test_expired_then_event_transient_continues_in_event_mode" + ), + "CR-DURABLE-003": "tests/test_journal.py::TestRateLimitedRetry::test_replay_restores_future_retry_after_floor", + "CR-DURABLE-004": ( + "tests/test_journal.py::TestAuthFailureRetention::test_replay_survives_api_key_rotation_with_tenant" + ), + "CR-DURABLE-005": "tests/test_journal.py::TestCommitJournal::test_corrupt_file_renamed_and_skipped", + "CR-DURABLE-006": "tests/test_journal.py::TestSyncReplay::test_replay_happens_once_per_directory", + "CR-DURABLE-007": ( + "tests/test_journal.py::TestCommitJournal::test_colliding_legacy_ids_are_distinct_and_migrate_safely" + ), + "CR-BOUNDARY-001": "tests/test_lifecycle.py::TestEvaluateActual::test_no_fallback_raises", +} + + +def main() -> int: + if len(sys.argv) != 2: + print("expected one scenario ID", file=sys.stderr) + return 2 + scenario = json.load(sys.stdin) + scenario_id = sys.argv[1] + if scenario.get("id") != scenario_id or scenario_id not in TESTS: + print("unknown or mismatched scenario ID", file=sys.stderr) + return 2 + if "expected_requests" in scenario or "assertions" in scenario: + print("runner disclosed conformance oracle", file=sys.stderr) + return 2 + + completed = subprocess.run( + [sys.executable, "-m", "pytest", "-q", TESTS[scenario_id]], + cwd=ROOT, + text=True, + capture_output=True, + check=False, + ) + if completed.stdout: + print(completed.stdout, file=sys.stderr, end="") + if completed.stderr: + print(completed.stderr, file=sys.stderr, end="") + requests, assertions = OBSERVATIONS[scenario_id] + json.dump({ + "scenario_id": scenario_id, + "passed": completed.returncode == 0, + "observed_requests": requests, + "assertions": assertions, + "diagnostic": f"native pytest exit code {completed.returncode}", + }, sys.stdout) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_journal.py b/tests/test_journal.py index d30ee44..f37cdd4 100644 --- a/tests/test_journal.py +++ b/tests/test_journal.py @@ -88,7 +88,13 @@ def _event_success() -> CyclesResponse: def _commit_success() -> CyclesResponse: - return CyclesResponse.success(200, {"status": "COMMITTED"}) + return CyclesResponse.success( + 200, + { + "status": "COMMITTED", + "charged": {"unit": "USD_MICROCENTS", "amount": 100}, + }, + ) def _record(reservation_id: str = "rsv_1", **overrides: Any) -> PendingCommitRecord: @@ -210,15 +216,37 @@ def test_corrupt_rename_failure_is_swallowed(self, tmp_path: Path, monkeypatch: 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_hashes_exact_utf8(self) -> None: + assert _safe_filename("rsv_abc-123") == ( + "v2-fe159e4dab8f7a05b609a0810be4800a22f3b08fc55a2762d2cded54d1484ec9.json" + ) + assert _safe_filename("rsv/../etc") == ( + "v2-08bf4457fdfa1fb11d02fb7f030f05c745a38058462cd1d158c40e644310d1de.json" + ) + + def test_safe_filename_unicode_vector_is_cross_sdk_stable(self) -> None: + assert _safe_filename("r🚀") == ( + "v2-34c5b33347a139e63c81ea72943cc15dd4c2087dc1eaa756a78f3c49974e0b87.json" + ) + + def test_colliding_legacy_ids_are_distinct_and_migrate_safely( + self, tmp_path: Path + ) -> None: + directory = tmp_path / "j" + journal = CommitJournal(directory) + journal.record(_record("rsv/a")) + journal.record(_record("rsv_a")) + assert _safe_filename("rsv/a") != _safe_filename("rsv_a") + assert len(list(directory.glob("*.json"))) == 2 - 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" + legacy = directory / "rsv_a.json" + legacy.write_text(_record("rsv/a").to_json(), encoding="utf-8") + journal.discard("rsv_a") + assert legacy.exists() + loaded = journal.load_pending(BASE_URL) + assert not legacy.exists() + assert (directory / _safe_filename("rsv/a")).exists() + assert [entry.reservation_id for entry in loaded] == ["rsv/a"] def test_stale_temp_files_are_reaped_on_load(self, tmp_path: Path) -> None: import os @@ -297,7 +325,7 @@ def test_journal_files_are_private(self, tmp_path: Path) -> None: if os.name == "posix": assert directory.stat().st_mode & 0o777 == 0o700 - assert (directory / "rsv_a.json").stat().st_mode & 0o777 == 0o600 + assert (directory / _safe_filename("rsv_a")).stat().st_mode & 0o777 == 0o600 def test_permission_tightening_failure_is_swallowed( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -910,7 +938,9 @@ def test_replay_isolated_by_api_key(self, tmp_path: Path) -> None: 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"] + assert _journal_files(tmp_path) == [ + _identity_dir(tmp_path, api_key="other-key") / _safe_filename("rsv_b") + ] engine_b = CommitRetryEngine(_config(tmp_path, api_key="other-key")) client_b = MagicMock() @@ -1185,6 +1215,40 @@ def test_expired_commit_schedules_event(self, tmp_path: Path) -> None: assert event_body["metadata"]["recovered_reservation_id"] == "rsv_test" mock_client.release_reservation.assert_not_called() + def test_journal_write_precedes_first_commit_and_success_discards( + 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 = _commit_success() + calls = MagicMock() + calls.attach_mock(engine.persist_pending, "persist") + calls.attach_mock(mock_client.commit_reservation, "commit") + + lifecycle.execute(lambda: "result", (), {}, _make_cfg()) + + names = [call[0] for call in calls.mock_calls] + assert names.index("persist") < names.index("commit") + engine.discard_pending.assert_called_once_with("rsv_test") + + def test_protocol_invalid_2xx_is_ambiguous_and_keeps_same_key( + 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.success( + 200, {"status": "COMMITTED"} + ) + + lifecycle.execute(lambda: "result", (), {}, _make_cfg()) + + engine.schedule.assert_called_once() + assert ( + engine.schedule.call_args.args[1]["idempotency_key"] + == engine.persist_pending.call_args.args[1]["idempotency_key"] + ) + engine.discard_pending.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() diff --git a/tests/test_retry.py b/tests/test_retry.py index 80f7691..cb13b01 100644 --- a/tests/test_retry.py +++ b/tests/test_retry.py @@ -9,7 +9,64 @@ from runcycles.config import CyclesConfig from runcycles.response import CyclesResponse -from runcycles.retry import AsyncCommitRetryEngine, CommitRetryEngine +from runcycles.retry import ( + AsyncCommitRetryEngine, + CommitRetryEngine, + _is_recognized_rejection, + _is_schema_valid_commit_success, + _is_schema_valid_event_success, +) + + +class TestSettlementSuccessValidation: + def test_only_known_non_retryable_codes_are_terminal_rejections(self) -> None: + assert _is_recognized_rejection("INVALID_REQUEST") + assert not _is_recognized_rejection("INTERNAL_ERROR") + assert not _is_recognized_rejection("LIMIT_EXCEEDED") + assert not _is_recognized_rejection("FUTURE_CODE") + + def test_commit_accepts_evidence_and_rejects_schema_invalid_optional_values(self) -> None: + body = { + "status": "COMMITTED", + "charged": {"unit": "USD_MICROCENTS", "amount": 1}, + "cycles_evidence": { + "evidence_id": "a" * 64, + "cycles_evidence_url": "https://cycles.example/v1/evidence/id", + }, + } + assert _is_schema_valid_commit_success(CyclesResponse.success(200, body)) + assert not _is_schema_valid_commit_success( + CyclesResponse.success(200, {**body, "balances": None}) + ) + assert not _is_schema_valid_commit_success( + CyclesResponse.success( + 200, + { + **body, + "charged": {"unit": "FUTURE_UNIT", "amount": 1}, + }, + ) + ) + assert not _is_schema_valid_commit_success( + CyclesResponse.success( + 200, + { + **body, + "charged": {"unit": "USD_MICROCENTS", "amount": "1"}, + }, + ) + ) + + def test_event_follows_exact_wire_schema(self) -> None: + assert _is_schema_valid_event_success( + CyclesResponse.success(201, {"status": "APPLIED", "event_id": ""}) + ) + assert not _is_schema_valid_event_success( + CyclesResponse.success( + 201, + {"status": "APPLIED", "event_id": "event-1", "charged": None}, + ) + ) @pytest.fixture @@ -48,7 +105,10 @@ def test_retries_until_success(self, config: CyclesConfig) -> None: # First call fails with 500, second succeeds mock_client.commit_reservation.side_effect = [ CyclesResponse.http_error(500, "Server error"), - CyclesResponse.success(200, {"status": "COMMITTED"}), + CyclesResponse.success(200, { + "status": "COMMITTED", + "charged": {"unit": "USD_MICROCENTS", "amount": 1}, + }), ] engine.set_client(mock_client) @@ -91,7 +151,10 @@ def test_handles_exception_during_retry(self, config: CyclesConfig) -> None: # First call throws, second succeeds mock_client.commit_reservation.side_effect = [ ConnectionError("network down"), - CyclesResponse.success(200, {"status": "COMMITTED"}), + CyclesResponse.success(200, { + "status": "COMMITTED", + "charged": {"unit": "USD_MICROCENTS", "amount": 1}, + }), ] engine.set_client(mock_client) @@ -123,7 +186,10 @@ async def test_retries_until_success(self, config: CyclesConfig) -> None: mock_client = AsyncMock() mock_client.commit_reservation.side_effect = [ CyclesResponse.http_error(500, "Server error"), - CyclesResponse.success(200, {"status": "COMMITTED"}), + CyclesResponse.success(200, { + "status": "COMMITTED", + "charged": {"unit": "USD_MICROCENTS", "amount": 1}, + }), ] engine.set_client(mock_client) @@ -162,7 +228,10 @@ async def test_handles_exception_during_retry(self, config: CyclesConfig) -> Non mock_client = AsyncMock() mock_client.commit_reservation.side_effect = [ ConnectionError("network down"), - CyclesResponse.success(200, {"status": "COMMITTED"}), + CyclesResponse.success(200, { + "status": "COMMITTED", + "charged": {"unit": "USD_MICROCENTS", "amount": 1}, + }), ] engine.set_client(mock_client) @@ -195,7 +264,10 @@ def test_schedule_creates_thread(self, config: CyclesConfig) -> None: engine = CommitRetryEngine(config) mock_client = MagicMock() # Return success immediately so the thread finishes quickly - mock_client.commit_reservation.return_value = CyclesResponse.success(200, {"status": "COMMITTED"}) + mock_client.commit_reservation.return_value = CyclesResponse.success(200, { + "status": "COMMITTED", + "charged": {"unit": "USD_MICROCENTS", "amount": 1}, + }) engine.set_client(mock_client) engine.schedule("rsv_1", {"idempotency_key": "k1"}) From 2378a526b74764f00c306fa7af0817445e229b8a Mon Sep 17 00:00:00 2001 From: Albert Mavashev Date: Wed, 29 Jul 2026 09:45:35 -0400 Subject: [PATCH 2/3] fix(recovery): harden journal conformance evidence --- CHANGELOG.md | 3 + runcycles/journal.py | 31 ++- scripts/recovery_conformance_adapter.py | 99 ++++----- tests/test_journal.py | 261 ++++++++++++++++------- tests/test_lifecycle.py | 265 ++++++++++++++++-------- tests/test_retry.py | 81 +++++--- 6 files changed, 472 insertions(+), 268 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c5f72f4..8ae2567 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Retain durable settlement records for contradictory retryable 4xx envelopes, and report heartbeat transport failures with their same-key retry or stop disposition. +- Quarantine unsupported or structurally invalid journal records without + aborting replay, keep serialization failures best-effort, and report exact + native test evidence to the shared conformance runner. ## [0.5.1] - 2026-07-27 diff --git a/runcycles/journal.py b/runcycles/journal.py index 33a3981..8de33d4 100644 --- a/runcycles/journal.py +++ b/runcycles/journal.py @@ -131,7 +131,12 @@ def to_json(self) -> str: @classmethod def from_json(cls, raw: str) -> PendingCommitRecord: data = json.loads(raw) - reservation_id = data["reservation_id"] + if not isinstance(data, dict): + raise ValueError("journal record must be a JSON object") + version = data.get("version") + if not isinstance(version, int) or isinstance(version, bool) or version != _RECORD_VERSION: + raise ValueError(f"unsupported journal version: {version!r}") + reservation_id = data.get("reservation_id") mode = data.get("mode", "commit") if not isinstance(reservation_id, str) or not reservation_id: raise ValueError("journal record missing reservation_id") @@ -141,15 +146,27 @@ def from_json(cls, raw: str) -> PendingCommitRecord: 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") + for body_key in ("commit_body", "event_fallback_body"): + if data.get(body_key) is not None and not isinstance(data[body_key], dict): + raise ValueError(f"journal record has invalid {body_key}") + if "base_url" in data and not isinstance(data["base_url"], str): + raise ValueError("journal record has invalid base_url") + recorded_at_raw = data.get("recorded_at_ms", 0) + if not isinstance(recorded_at_raw, int) or isinstance(recorded_at_raw, bool) or recorded_at_raw < 0: + raise ValueError("journal record has invalid recorded_at_ms") not_before_raw = data.get("not_before_ms") + if not_before_raw is not None and ( + not isinstance(not_before_raw, int) or isinstance(not_before_raw, bool) or not_before_raw < 0 + ): + raise ValueError("journal record has invalid 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, + recorded_at_ms=recorded_at_raw, + not_before_ms=not_before_raw, ) @@ -186,14 +203,14 @@ def record(self, entry: PendingCommitRecord) -> None: tmp.write_text(entry.to_json(), encoding="utf-8") _restrict_permissions(tmp, 0o600) tmp.replace(target) - except OSError: + except Exception: 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: + except Exception: logger.warning( "Failed to journal pending commit (continuing without durability): id=%s", entry.reservation_id, @@ -260,9 +277,7 @@ def load_pending(self, base_url: str) -> list[PendingCommitRecord]: standard_path, ) else: - existing = PendingCommitRecord.from_json( - standard_path.read_text(encoding="utf-8") - ) + existing = PendingCommitRecord.from_json(standard_path.read_text(encoding="utf-8")) if existing.reservation_id == entry.reservation_id: path.unlink(missing_ok=True) duplicate_of_standard = True diff --git a/scripts/recovery_conformance_adapter.py b/scripts/recovery_conformance_adapter.py index b75cbde..5ed88ae 100644 --- a/scripts/recovery_conformance_adapter.py +++ b/scripts/recovery_conformance_adapter.py @@ -10,73 +10,45 @@ ROOT = Path(__file__).resolve().parents[1] -OBSERVATIONS = { - "CR-CORE-001": (["commit", "commit_same_key"], [ - "settlement_occurs_at_most_once", "retry_uses_original_idempotency_key"]), - "CR-CORE-002": (["commit", "event_same_key"], [ - "event_carries_original_subject_action_actual", "settlement_occurs_at_most_once"]), - "CR-CORE-003": (["extend", "extend_same_key", "commit"], [ - "heartbeat_failure_reports_reservation_and_disposition", - "guarded_action_continues_under_warn_policy", "final_settlement_is_attempted"]), - "CR-CORE-004": (["commit", "commit_same_key"], [ - "only_schema_valid_expected_status_is_terminal_success", - "ambiguous_success_retains_original_idempotency_key"]), - "CR-DURABLE-001": (["commit", "commit_same_key_after_restart"], [ - "journal_write_precedes_first_settlement_request", - "unresolved_record_survives_restart", "successful_replay_removes_record", - "settlement_occurs_at_most_once"]), - "CR-DURABLE-002": (["commit_same_key_after_restart", "event_same_key_after_restart"], [ - "event_mode_is_persisted_before_event_attempt", "successful_event_removes_record", - "settlement_occurs_at_most_once"]), - "CR-DURABLE-003": (["commit", "commit_same_key_after_retry_after"], [ - "no_retry_before_persisted_not_before", "successful_replay_removes_record"]), - "CR-DURABLE-004": (["commit_same_key_after_restart"], [ - "new_tenant_credential_finds_record", "old_api_key_is_not_stored"]), - "CR-DURABLE-005": ([], [ - "corrupt_record_is_quarantined", "other_valid_records_still_replay", - "corruption_is_reported"]), - "CR-DURABLE-006": (["concurrent_commit_same_key"], [ - "settlement_occurs_at_most_once", "terminal_record_is_removed"]), - "CR-DURABLE-007": ([ - "commit_first_identifier", "commit_second_identifier", - "commit_first_identifier_same_key_after_restart", - "commit_second_identifier_same_key_after_restart", - ], [ - "standard_filename_is_sha256_of_exact_utf8_identifier", - "distinct_identifiers_never_share_a_journal_file", - "matching_legacy_record_migrates_without_deleting_collision", - "both_settlements_occur_at_most_once", - ]), - "CR-BOUNDARY-001": ([], [ - "sdk_does_not_claim_ledger_convergence", "application_checkpoint_is_required"]), -} - TESTS = { - "CR-CORE-001": "tests/test_retry.py::TestCommitRetryEngine::test_retries_until_success", - "CR-CORE-002": "tests/test_journal.py::TestLifecycleEventFallbackWiring::test_expired_commit_schedules_event", - "CR-CORE-003": "tests/test_lifecycle.py::TestSyncLifecycleExecution::test_heartbeat_exception_does_not_crash", + "CR-CORE-001": ("tests/test_retry.py::TestCommitRetryEngine::test_retries_until_success",), + "CR-CORE-002": ("tests/test_journal.py::TestLifecycleEventFallbackWiring::test_expired_commit_schedules_event",), + "CR-CORE-003": ("tests/test_lifecycle.py::TestSyncLifecycleExecution::test_heartbeat_exception_does_not_crash",), "CR-CORE-004": ( "tests/test_journal.py::TestLifecycleEventFallbackWiring::" - "test_protocol_invalid_2xx_is_ambiguous_and_keeps_same_key" + "test_protocol_invalid_2xx_is_ambiguous_and_keeps_same_key", ), "CR-DURABLE-001": ( "tests/test_journal.py::TestLifecycleEventFallbackWiring::" - "test_journal_write_precedes_first_commit_and_success_discards" + "test_journal_write_precedes_first_commit_and_success_discards", + "tests/test_journal.py::TestSyncReplay::test_replays_pending_commit_on_set_client", + "tests/test_retry.py::TestCommitRetryEngine::test_retries_until_success", ), "CR-DURABLE-002": ( - "tests/test_journal.py::TestSyncEngineDurability::" - "test_expired_then_event_transient_continues_in_event_mode" + "tests/test_journal.py::TestSyncEngineDurability::test_expired_then_event_transient_continues_in_event_mode", + "tests/test_journal.py::TestSyncReplay::test_replays_event_mode_entry", + ), + "CR-DURABLE-003": ( + "tests/test_journal.py::TestRateLimitedRetry::test_429_commit_is_transient_and_honors_retry_after", + "tests/test_journal.py::TestRateLimitedRetry::test_replay_restores_future_retry_after_floor", + "tests/test_journal.py::TestRateLimitedRetry::test_429_then_success_discards_journal", ), - "CR-DURABLE-003": "tests/test_journal.py::TestRateLimitedRetry::test_replay_restores_future_retry_after_floor", "CR-DURABLE-004": ( - "tests/test_journal.py::TestAuthFailureRetention::test_replay_survives_api_key_rotation_with_tenant" + "tests/test_journal.py::TestAuthFailureRetention::test_replay_survives_api_key_rotation_with_tenant", + ), + "CR-DURABLE-005": ( + "tests/test_journal.py::TestCommitJournal::" + "test_corrupt_and_unsupported_records_are_quarantined_without_blocking_valid", + ), + "CR-DURABLE-006": ( + "tests/test_journal.py::TestSyncReplay::test_concurrent_replay_workers_reuse_one_key_and_remove_record", ), - "CR-DURABLE-005": "tests/test_journal.py::TestCommitJournal::test_corrupt_file_renamed_and_skipped", - "CR-DURABLE-006": "tests/test_journal.py::TestSyncReplay::test_replay_happens_once_per_directory", "CR-DURABLE-007": ( - "tests/test_journal.py::TestCommitJournal::test_colliding_legacy_ids_are_distinct_and_migrate_safely" + "tests/test_journal.py::TestCommitJournal::test_colliding_legacy_ids_are_distinct_and_migrate_safely", + ), + "CR-BOUNDARY-001": ( + "tests/test_lifecycle.py::TestSyncLifecycleExecution::test_missing_actual_surfaces_without_settlement", ), - "CR-BOUNDARY-001": "tests/test_lifecycle.py::TestEvaluateActual::test_no_fallback_raises", } @@ -94,7 +66,7 @@ def main() -> int: return 2 completed = subprocess.run( - [sys.executable, "-m", "pytest", "-q", TESTS[scenario_id]], + [sys.executable, "-m", "pytest", "-q", *TESTS[scenario_id]], cwd=ROOT, text=True, capture_output=True, @@ -104,14 +76,15 @@ def main() -> int: print(completed.stdout, file=sys.stderr, end="") if completed.stderr: print(completed.stderr, file=sys.stderr, end="") - requests, assertions = OBSERVATIONS[scenario_id] - json.dump({ - "scenario_id": scenario_id, - "passed": completed.returncode == 0, - "observed_requests": requests, - "assertions": assertions, - "diagnostic": f"native pytest exit code {completed.returncode}", - }, sys.stdout) + json.dump( + { + "scenario_id": scenario_id, + "passed": completed.returncode == 0, + "native_tests": list(TESTS[scenario_id]), + "diagnostic": f"native pytest exit code {completed.returncode}", + }, + sys.stdout, + ) return 0 diff --git a/tests/test_journal.py b/tests/test_journal.py index f37cdd4..65f01a0 100644 --- a/tests/test_journal.py +++ b/tests/test_journal.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import threading import time from pathlib import Path from typing import Any @@ -71,14 +72,16 @@ def _event_body() -> dict[str, Any]: def _expired_response() -> CyclesResponse: return CyclesResponse.http_error( - 410, "Expired", + 410, + "Expired", body={"error": "RESERVATION_EXPIRED", "message": "Expired", "request_id": "r1"}, ) def _finalized_response() -> CyclesResponse: return CyclesResponse.http_error( - 409, "Finalized", + 409, + "Finalized", body={"error": "RESERVATION_FINALIZED", "message": "Finalized", "request_id": "r2"}, ) @@ -110,7 +113,10 @@ def _record(reservation_id: str = "rsv_1", **overrides: Any) -> PendingCommitRec def _identity_dir( - tmp_path: Path, api_key: str = "test-key", base_url: str = BASE_URL, tenant: str | None = None, + 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) @@ -162,25 +168,39 @@ 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: + def test_corrupt_and_unsupported_records_are_quarantined_without_blocking_valid( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: directory = tmp_path / "j" journal = CommitJournal(directory) journal.record(_record("rsv_good")) (directory / "rsv_bad.json").write_text("{not json", encoding="utf-8") + (directory / "rsv_future.json").write_text( + _record("rsv_future").to_json().replace('"version": 1', '"version": 2'), + encoding="utf-8", + ) + (directory / "rsv_array.json").write_text("[]", 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() + for stem in ("rsv_bad", "rsv_future", "rsv_array"): + assert (directory / f"{stem}.corrupt").exists() + assert not (directory / f"{stem}.json").exists() + messages = [record.getMessage() for record in caplog.records] + assert sum("Skipping corrupt journal entry" in message for message in messages) == 3 + assert any("rsv_future.json" in message for message in messages) 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": {}}', + "no_rid.json": ('{"version": 1, "reservation_id": "", "mode": "commit", "commit_body": {}}'), + "bad_mode.json": ('{"version": 1, "reservation_id": "r1", "mode": "sideways", "commit_body": {}}'), + "commit_no_body.json": ('{"version": 1, "reservation_id": "r2", "mode": "commit"}'), + "event_no_body.json": ('{"version": 1, "reservation_id": "r3", "mode": "event", "commit_body": {}}'), + "bad_timestamp.json": ( + '{"version": 1, "reservation_id": "r4", "mode": "commit", "commit_body": {}, "recorded_at_ms": "now"}' + ), } for name, content in cases.items(): (directory / name).write_text(content, encoding="utf-8") @@ -195,6 +215,16 @@ def test_record_swallows_os_errors(self, tmp_path: Path, monkeypatch: pytest.Mon journal.record(_record("rsv_a")) # must not raise assert journal.load_pending(BASE_URL) == [] + def test_record_swallows_serialization_errors(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + journal = CommitJournal(tmp_path / "j") + monkeypatch.setattr( + PendingCommitRecord, + "to_json", + MagicMock(side_effect=TypeError("not JSON serializable")), + ) + journal.record(_record("rsv_a")) # must not raise or block settlement + 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")) @@ -225,13 +255,9 @@ def test_safe_filename_hashes_exact_utf8(self) -> None: ) def test_safe_filename_unicode_vector_is_cross_sdk_stable(self) -> None: - assert _safe_filename("r🚀") == ( - "v2-34c5b33347a139e63c81ea72943cc15dd4c2087dc1eaa756a78f3c49974e0b87.json" - ) + assert _safe_filename("r🚀") == ("v2-34c5b33347a139e63c81ea72943cc15dd4c2087dc1eaa756a78f3c49974e0b87.json") - def test_colliding_legacy_ids_are_distinct_and_migrate_safely( - self, tmp_path: Path - ) -> None: + def test_colliding_legacy_ids_are_distinct_and_migrate_safely(self, tmp_path: Path) -> None: directory = tmp_path / "j" journal = CommitJournal(directory) journal.record(_record("rsv/a")) @@ -268,7 +294,9 @@ def test_stale_temp_files_are_reaped_on_load(self, tmp_path: Path) -> None: 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, + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, ) -> None: import os @@ -328,7 +356,9 @@ def test_journal_files_are_private(self, tmp_path: Path) -> None: assert (directory / _safe_filename("rsv_a")).stat().st_mode & 0o777 == 0o600 def test_permission_tightening_failure_is_swallowed( - self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr(Path, "chmod", MagicMock(side_effect=OSError("not supported"))) journal = CommitJournal(tmp_path / "j") @@ -336,7 +366,9 @@ def test_permission_tightening_failure_is_swallowed( 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, + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, ) -> None: journal = CommitJournal(tmp_path / "j") monkeypatch.setattr(Path, "replace", MagicMock(side_effect=OSError("locked"))) @@ -345,7 +377,9 @@ def test_failed_publish_cleans_up_temp_file( assert list((tmp_path / "j").glob("*.json")) == [] def test_failed_publish_and_cleanup_never_raise( - self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, ) -> None: journal = CommitJournal(tmp_path / "j") monkeypatch.setattr(Path, "replace", MagicMock(side_effect=OSError("locked"))) @@ -469,7 +503,9 @@ 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"}, + 409, + "Mismatch", + body={"error": "IDEMPOTENCY_MISMATCH", "message": "m", "request_id": "r"}, ) engine.set_client(mock_client) @@ -498,10 +534,19 @@ def test_expired_then_event_transient_continues_in_event_mode(self, tmp_path: Pa 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(), - ] + + def event_response(_: dict[str, Any]) -> CyclesResponse: + persisted = CommitJournal(_identity_dir(tmp_path)).load_pending(BASE_URL) + assert len(persisted) == 1 + assert persisted[0].mode == "event" + assert persisted[0].event_fallback_body == _event_body() + return ( + CyclesResponse.http_error(503, "unavailable") + if mock_client.create_event.call_count == 1 + else _event_success() + ) + + mock_client.create_event.side_effect = event_response engine.set_client(mock_client) pending = _PendingCommit("rsv_1", _commit_body(), _event_body()) @@ -510,6 +555,7 @@ def test_expired_then_event_transient_continues_in_event_mode(self, tmp_path: Pa # 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 + assert _journal_files(tmp_path) == [] def test_disabled_with_journal_persists_entry(self, tmp_path: Path) -> None: engine = CommitRetryEngine(_config(tmp_path, retry_enabled=False)) @@ -620,7 +666,8 @@ def test_429_commit_is_transient_and_honors_retry_after(self, tmp_path: Path) -> # keep retrying, waiting at least the server's Retry-After. engine = CommitRetryEngine(_config(tmp_path)) response = CyclesResponse.http_error( - 429, "Rate limited", + 429, + "Rate limited", body={"error": "LIMIT_EXCEEDED", "message": "slow down", "request_id": "r9"}, headers={"retry-after": "2"}, ) @@ -668,7 +715,9 @@ 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"}, + 429, + "busy", + body={"error": "LIMIT_EXCEEDED", "message": "m", "request_id": "r"}, ) engine.set_client(mock_client) @@ -701,7 +750,9 @@ def test_retry_after_floor_is_persisted(self, tmp_path: Path) -> 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, + 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)) @@ -715,7 +766,9 @@ def test_replay_restores_future_retry_after_floor( 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, + 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)) @@ -776,7 +829,9 @@ def test_unclassifiable_commit_4xx_retains_journal(self, tmp_path: Path) -> None mock_client2 = MagicMock() mock_client2.commit_reservation.return_value = CyclesResponse.http_error( - 422, "future", body={"error": "FUTURE_REJECTION_CODE", "message": "m", "request_id": "r"}, + 422, + "future", + body={"error": "FUTURE_REJECTION_CODE", "message": "m", "request_id": "r"}, ) engine2 = CommitRetryEngine(_config(tmp_path)) engine2.set_client(mock_client2) @@ -817,7 +872,8 @@ def test_bodyless_410_triggers_event_fallback(self, tmp_path: Path) -> None: 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", + 429, + "busy", body={"error": "LIMIT_EXCEEDED", "message": "m", "request_id": "r"}, headers={"retry-after": "7200"}, ) @@ -833,7 +889,9 @@ def test_honored_retry_after_is_clamped(self, tmp_path: Path, monkeypatch: pytes assert captured[-1].retry_after_s == 3600.0 def test_restored_replay_floor_is_clamped( - self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, + 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)) @@ -848,7 +906,11 @@ def test_restored_replay_floor_is_clamped( 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")) + identity_dir = _identity_dir(tmp_path, api_key="old-key", tenant="acme") + CommitJournal(identity_dir).record(_record("rsv_old")) + persisted = next(identity_dir.glob("*.json")) + assert "old-key" not in str(persisted) + assert "old-key" not in persisted.read_text(encoding="utf-8") engine = CommitRetryEngine(_config(tmp_path, api_key="rotated-key", tenant="acme")) mock_client = MagicMock() @@ -879,9 +941,7 @@ def test_replays_pending_commit_on_set_client(self, tmp_path: Path) -> None: 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) - ) + CommitJournal(_identity_dir(tmp_path)).record(_record("rsv_old", mode="event", commit_body=None)) engine = CommitRetryEngine(_config(tmp_path)) mock_client = MagicMock() @@ -892,23 +952,57 @@ def test_replays_event_mode_entry(self, tmp_path: Path) -> None: 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")) + def test_concurrent_replay_workers_reuse_one_key_and_remove_record(self, tmp_path: Path) -> None: 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) + # Claim the initially empty directory before seeding the record so both + # independently constructed workers can be driven concurrently below. + first.set_client(MagicMock()) + second.set_client(MagicMock()) + journal = CommitJournal(_identity_dir(tmp_path)) + journal.record(_record("rsv_old")) + loaded_a = journal.load_pending(BASE_URL)[0] + loaded_b = CommitJournal(_identity_dir(tmp_path)).load_pending(BASE_URL)[0] + + barrier = threading.Barrier(2) + seen_keys: list[str] = [] + seen_lock = threading.Lock() + + def settle(_: str, body: dict[str, Any]) -> CyclesResponse: + barrier.wait(timeout=2) + with seen_lock: + seen_keys.append(str(body["idempotency_key"])) + return _commit_success() - assert client1.commit_reservation.call_count == 1 - client2.commit_reservation.assert_not_called() + client = MagicMock() + client.commit_reservation.side_effect = settle + first._client = client + second._client = client + pending_a = _PendingCommit( + loaded_a.reservation_id, + loaded_a.commit_body, + loaded_a.event_fallback_body, + ) + pending_b = _PendingCommit( + loaded_b.reservation_id, + loaded_b.commit_body, + loaded_b.event_fallback_body, + ) + + threads = [ + threading.Thread(target=first._retry_loop, args=(pending_a,)), + threading.Thread(target=second._retry_loop, args=(pending_b,)), + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=5) + + assert all(not thread.is_alive() for thread in threads) + assert seen_keys == ["ck-1", "ck-1"] + assert len(set(seen_keys)) == 1 + assert _journal_files(tmp_path) == [] def test_replay_skips_other_server_entries(self, tmp_path: Path) -> None: # Defense-in-depth: a mismatched-base_url record inside the identity @@ -938,9 +1032,7 @@ def test_replay_isolated_by_api_key(self, tmp_path: Path) -> None: 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") / _safe_filename("rsv_b") - ] + assert _journal_files(tmp_path) == [_identity_dir(tmp_path, api_key="other-key") / _safe_filename("rsv_b")] engine_b = CommitRetryEngine(_config(tmp_path, api_key="other-key")) client_b = MagicMock() @@ -955,9 +1047,7 @@ 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) - ) + 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() @@ -1090,7 +1180,9 @@ async def test_flush_zero_timeout_returns_immediately(self, tmp_path: Path) -> N await engine.flush() # must not raise or block async def test_schedule_seeds_retry_after( - self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, ) -> None: engine = AsyncCommitRetryEngine(_config(tmp_path)) engine.set_client(AsyncMock()) @@ -1152,7 +1244,10 @@ def test_builds_spec_shape_reusing_commit_idempotency_key(self) -> None: "metadata": {"run": "abc"}, } body = _build_event_fallback_body( - "rsv_9", {"tenant": "acme"}, {"kind": "llm.completion", "name": "gpt"}, commit_body, + "rsv_9", + {"tenant": "acme"}, + {"kind": "llm.completion", "name": "gpt"}, + commit_body, ) assert body["idempotency_key"] == "ck-9" @@ -1167,7 +1262,10 @@ def test_builds_spec_shape_reusing_commit_idempotency_key(self) -> None: def test_without_metrics_or_metadata(self) -> None: body = _build_event_fallback_body( - "rsv_9", {"tenant": "acme"}, {"kind": "k", "name": "n"}, _commit_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"} @@ -1179,14 +1277,17 @@ def test_without_metrics_or_metadata(self) -> None: 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}, - }) + 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: @@ -1212,12 +1313,14 @@ def test_expired_commit_schedules_event(self, tmp_path: Path) -> None: rid, event_body = engine.schedule_event.call_args.args assert rid == "rsv_test" assert event_body["subject"] == {"tenant": "acme"} + assert event_body["action"] == {"kind": "unknown", "name": "unknown"} + assert event_body["actual"] == {"unit": "USD_MICROCENTS", "amount": 1000} assert event_body["metadata"]["recovered_reservation_id"] == "rsv_test" + persisted_body = engine.persist_pending.call_args.args[1] + assert event_body["idempotency_key"] == persisted_body["idempotency_key"] mock_client.release_reservation.assert_not_called() - def test_journal_write_precedes_first_commit_and_success_discards( - self, tmp_path: Path - ) -> None: + def test_journal_write_precedes_first_commit_and_success_discards(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 = _commit_success() @@ -1231,14 +1334,10 @@ def test_journal_write_precedes_first_commit_and_success_discards( assert names.index("persist") < names.index("commit") engine.discard_pending.assert_called_once_with("rsv_test") - def test_protocol_invalid_2xx_is_ambiguous_and_keeps_same_key( - self, tmp_path: Path - ) -> None: + def test_protocol_invalid_2xx_is_ambiguous_and_keeps_same_key(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.success( - 200, {"status": "COMMITTED"} - ) + mock_client.commit_reservation.return_value = CyclesResponse.success(200, {"status": "COMMITTED"}) lifecycle.execute(lambda: "result", (), {}, _make_cfg()) @@ -1297,7 +1396,8 @@ def test_rate_limited_first_commit_schedules_retry_not_release(self, tmp_path: P 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", + 429, + "busy", body={"error": "LIMIT_EXCEEDED", "message": "slow down", "request_id": "r9"}, headers={"retry-after": "3"}, ) @@ -1364,7 +1464,8 @@ async def test_rate_limited_first_commit_schedules_retry_not_release(self, tmp_p 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", + 429, + "busy", body={"error": "LIMIT_EXCEEDED", "message": "slow down", "request_id": "r9"}, headers={"retry-after": "3"}, ) @@ -1460,7 +1561,8 @@ def test_rate_limited_first_commit_schedules_retry_not_release(self, tmp_path: P 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", + 429, + "busy", body={"error": "LIMIT_EXCEEDED", "message": "slow down", "request_id": "r9"}, headers={"retry-after": "3"}, ) @@ -1545,7 +1647,8 @@ async def test_rate_limited_first_commit_schedules_retry_not_release(self, tmp_p 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", + 429, + "busy", body={"error": "LIMIT_EXCEEDED", "message": "slow down", "request_id": "r9"}, headers={"retry-after": "3"}, ) diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py index 3e446a0..ed6cb14 100644 --- a/tests/test_lifecycle.py +++ b/tests/test_lifecycle.py @@ -332,7 +332,8 @@ def test_basic(self) -> None: class TestBuildProtocolExceptionEdgeCases: def test_maps_overdraft_limit_exceeded(self) -> None: response = CyclesResponse.http_error( - 409, "Over limit", + 409, + "Over limit", body={"error": "OVERDRAFT_LIMIT_EXCEEDED", "message": "Over limit", "request_id": "r1"}, ) exc = _build_protocol_exception("Failed", response) @@ -340,7 +341,8 @@ def test_maps_overdraft_limit_exceeded(self) -> None: def test_maps_debt_outstanding(self) -> None: response = CyclesResponse.http_error( - 409, "Debt", + 409, + "Debt", body={"error": "DEBT_OUTSTANDING", "message": "Debt", "request_id": "r2"}, ) exc = _build_protocol_exception("Failed", response) @@ -351,7 +353,8 @@ def test_retry_after_header_fallback(self) -> None: # the HTTP Retry-After header (seconds); surface it as retry_after_ms # when the body carries no retry_after_ms field. response = CyclesResponse.http_error( - 429, "Rate limited", + 429, + "Rate limited", body={"error": "LIMIT_EXCEEDED", "message": "Rate limited", "request_id": "r6"}, headers={"retry-after": "3"}, ) @@ -362,7 +365,8 @@ def test_retry_after_header_fallback(self) -> None: def test_retry_after_body_wins_over_header(self) -> None: response = CyclesResponse.http_error( - 429, "Rate limited", + 429, + "Rate limited", body={ "error": "LIMIT_EXCEEDED", "message": "Rate limited", @@ -376,7 +380,8 @@ def test_retry_after_body_wins_over_header(self) -> None: def test_maps_tenant_closed(self) -> None: response = CyclesResponse.http_error( - 409, "Tenant closed", + 409, + "Tenant closed", body={"error": "TENANT_CLOSED", "message": "Tenant closed", "request_id": "r5"}, ) exc = _build_protocol_exception("Failed", response) @@ -387,7 +392,8 @@ def test_maps_tenant_closed(self) -> None: def test_maps_reservation_expired(self) -> None: response = CyclesResponse.http_error( - 410, "Expired", + 410, + "Expired", body={"error": "RESERVATION_EXPIRED", "message": "Expired", "request_id": "r3"}, ) exc = _build_protocol_exception("Failed", response) @@ -395,7 +401,8 @@ def test_maps_reservation_expired(self) -> None: def test_maps_reservation_finalized(self) -> None: response = CyclesResponse.http_error( - 409, "Finalized", + 409, + "Finalized", body={"error": "RESERVATION_FINALIZED", "message": "Finalized", "request_id": "r4"}, ) exc = _build_protocol_exception("Failed", response) @@ -404,7 +411,8 @@ def test_maps_reservation_finalized(self) -> None: def test_fallback_when_body_not_error_response(self) -> None: """When body has an error field but doesn't parse as ErrorResponse.""" response = CyclesResponse.http_error( - 500, "Something broke", + 500, + "Something broke", body={"error": "INTERNAL_ERROR"}, # missing required 'message' and 'request_id' ) exc = _build_protocol_exception("Call failed", response) @@ -420,6 +428,7 @@ def test_fallback_raw_error_no_error_message(self) -> None: # --- Helper to build a mock sync client --- + def _make_config() -> CyclesConfig: return CyclesConfig( base_url="http://localhost:7878", @@ -432,53 +441,71 @@ def _make_config() -> CyclesConfig: 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}, - }) + 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 _deny_response() -> CyclesResponse: - return CyclesResponse.success(200, { - "decision": "DENY", - "affected_scopes": ["tenant:acme"], - "reason_code": "BUDGET_EXCEEDED", - }) + return CyclesResponse.success( + 200, + { + "decision": "DENY", + "affected_scopes": ["tenant:acme"], + "reason_code": "BUDGET_EXCEEDED", + }, + ) def _dry_run_allow_response() -> CyclesResponse: - return CyclesResponse.success(200, { - "decision": "ALLOW", - "affected_scopes": ["tenant:acme"], - "scope_path": "tenant:acme", - "reserved": {"unit": "USD_MICROCENTS", "amount": 1000}, - }) + return CyclesResponse.success( + 200, + { + "decision": "ALLOW", + "affected_scopes": ["tenant:acme"], + "scope_path": "tenant:acme", + "reserved": {"unit": "USD_MICROCENTS", "amount": 1000}, + }, + ) def _dry_run_deny_response() -> CyclesResponse: - return CyclesResponse.success(200, { - "decision": "DENY", - "affected_scopes": ["tenant:acme"], - "reason_code": "BUDGET_EXCEEDED", - }) + return CyclesResponse.success( + 200, + { + "decision": "DENY", + "affected_scopes": ["tenant:acme"], + "reason_code": "BUDGET_EXCEEDED", + }, + ) def _commit_success() -> CyclesResponse: - return CyclesResponse.success(200, { - "status": "COMMITTED", - "charged": {"unit": "USD_MICROCENTS", "amount": 1000}, - }) + return CyclesResponse.success( + 200, + { + "status": "COMMITTED", + "charged": {"unit": "USD_MICROCENTS", "amount": 1000}, + }, + ) def _release_success() -> CyclesResponse: - return CyclesResponse.success(200, { - "status": "RELEASED", - "released": {"unit": "USD_MICROCENTS", "amount": 1000}, - }) + return CyclesResponse.success( + 200, + { + "status": "RELEASED", + "released": {"unit": "USD_MICROCENTS", "amount": 1000}, + }, + ) def _make_cfg(**kwargs: object) -> DecoratorConfig: @@ -514,10 +541,26 @@ def test_dry_run_allow_returns_result(self) -> None: result = lifecycle.execute(lambda: "should not run", (), {}, cfg) from runcycles.models import DryRunResult + assert isinstance(result, DryRunResult) assert result.is_allowed() mock_client.commit_reservation.assert_not_called() + def test_missing_actual_surfaces_without_settlement(self, monkeypatch: pytest.MonkeyPatch) -> None: + lifecycle, mock_client = self._make_lifecycle() + mock_client.create_reservation.return_value = _allow_response() + mock_client.release_reservation.return_value = _release_success() + schedule_event = MagicMock() + monkeypatch.setattr(lifecycle._retry_engine, "schedule_event", schedule_event) + + cfg = _make_cfg(use_estimate_if_actual_not_provided=False) + with pytest.raises(ValueError, match="actual expression is required"): + lifecycle.execute(lambda: "result", (), {}, cfg) + + mock_client.commit_reservation.assert_not_called() + schedule_event.assert_not_called() + mock_client.release_reservation.assert_called_once() + def test_deny_raises(self) -> None: lifecycle, mock_client = self._make_lifecycle() mock_client.create_reservation.return_value = _deny_response() @@ -529,11 +572,14 @@ def test_deny_raises(self) -> None: def test_missing_reservation_id_raises(self) -> None: lifecycle, mock_client = self._make_lifecycle() - mock_client.create_reservation.return_value = CyclesResponse.success(200, { - "decision": "ALLOW", - "affected_scopes": ["tenant:acme"], - # reservation_id intentionally missing - }) + mock_client.create_reservation.return_value = CyclesResponse.success( + 200, + { + "decision": "ALLOW", + "affected_scopes": ["tenant:acme"], + # reservation_id intentionally missing + }, + ) cfg = _make_cfg() @@ -544,7 +590,8 @@ def test_commit_finalized_does_not_release(self) -> None: lifecycle, mock_client = self._make_lifecycle() mock_client.create_reservation.return_value = _allow_response() mock_client.commit_reservation.return_value = CyclesResponse.http_error( - 409, "Finalized", + 409, + "Finalized", body={"error": "RESERVATION_FINALIZED", "message": "Already committed", "request_id": "r1"}, ) @@ -557,7 +604,8 @@ def test_commit_expired_does_not_release(self) -> None: lifecycle, mock_client = self._make_lifecycle() mock_client.create_reservation.return_value = _allow_response() mock_client.commit_reservation.return_value = CyclesResponse.http_error( - 410, "Expired", + 410, + "Expired", body={"error": "RESERVATION_EXPIRED", "message": "Expired", "request_id": "r1"}, ) @@ -570,7 +618,8 @@ def test_commit_idempotency_mismatch_does_not_release(self) -> None: lifecycle, mock_client = self._make_lifecycle() mock_client.create_reservation.return_value = _allow_response() mock_client.commit_reservation.return_value = CyclesResponse.http_error( - 409, "Mismatch", + 409, + "Mismatch", body={"error": "IDEMPOTENCY_MISMATCH", "message": "Mismatch", "request_id": "r1"}, ) @@ -583,7 +632,8 @@ def test_commit_other_client_error_triggers_release(self) -> None: lifecycle, mock_client = self._make_lifecycle() mock_client.create_reservation.return_value = _allow_response() mock_client.commit_reservation.return_value = CyclesResponse.http_error( - 400, "Bad request", + 400, + "Bad request", body={"error": "UNIT_MISMATCH", "message": "Unit mismatch", "request_id": "r1"}, ) mock_client.release_reservation.return_value = _release_success() @@ -603,9 +653,7 @@ def test_commit_transport_error_schedules_retry(self) -> None: lifecycle = CyclesLifecycle(mock_client, retry_engine, {"tenant": "acme"}) mock_client.create_reservation.return_value = _allow_response() - mock_client.commit_reservation.return_value = CyclesResponse.transport_error( - ConnectionError("network down") - ) + mock_client.commit_reservation.return_value = CyclesResponse.transport_error(ConnectionError("network down")) cfg = _make_cfg() lifecycle.execute(lambda: "result", (), {}, cfg) @@ -671,9 +719,13 @@ def test_heartbeat_extends_reservation(self) -> None: lifecycle, mock_client = self._make_lifecycle() mock_client.create_reservation.return_value = _allow_response() mock_client.commit_reservation.return_value = _commit_success() - mock_client.extend_reservation.return_value = CyclesResponse.success(200, { - "status": "ACTIVE", "expires_at_ms": 9999999999, - }) + mock_client.extend_reservation.return_value = CyclesResponse.success( + 200, + { + "status": "ACTIVE", + "expires_at_ms": 9999999999, + }, + ) # Use a very short TTL so heartbeat fires during execution cfg = _make_cfg(ttl_ms=2000) @@ -701,11 +753,14 @@ def slow_fn() -> str: result = lifecycle.execute(slow_fn, (), {}, cfg) assert result == "done" - def test_heartbeat_exception_does_not_crash(self) -> None: + def test_heartbeat_exception_does_not_crash(self, caplog: pytest.LogCaptureFixture) -> None: lifecycle, mock_client = self._make_lifecycle() mock_client.create_reservation.return_value = _allow_response() mock_client.commit_reservation.return_value = _commit_success() - mock_client.extend_reservation.side_effect = ConnectionError("network down") + mock_client.extend_reservation.side_effect = [ + ConnectionError("network down"), + CyclesResponse.success(200, {"status": "ACTIVE", "expires_at_ms": 9999999999}), + ] cfg = _make_cfg(ttl_ms=2000) @@ -715,6 +770,16 @@ def slow_fn() -> str: result = lifecycle.execute(slow_fn, (), {}, cfg) assert result == "done" + assert mock_client.extend_reservation.call_count >= 2 + sent = [call.args[1]["idempotency_key"] for call in mock_client.extend_reservation.call_args_list[:2]] + assert sent[0] == sent[1] + mock_client.commit_reservation.assert_called_once() + heartbeat_messages = [ + record.getMessage() + for record in caplog.records + if "Heartbeat extend transport error" in record.getMessage() + ] + assert any("rsv_test" in message and "same idempotency key" in message for message in heartbeat_messages) def test_commit_unrecognized_response_logged(self) -> None: lifecycle, mock_client = self._make_lifecycle() @@ -727,7 +792,8 @@ def test_commit_unrecognized_response_logged(self) -> None: def test_reservation_creation_failure_raises(self) -> None: lifecycle, mock_client = self._make_lifecycle() mock_client.create_reservation.return_value = CyclesResponse.http_error( - 500, "Internal error", + 500, + "Internal error", body={"error": "INTERNAL_ERROR", "message": "Server down", "request_id": "r1"}, ) @@ -744,6 +810,7 @@ def test_heartbeat_skipped_when_ttl_zero(self) -> None: # We can't easily set ttl_ms=0 since validation rejects it, but we can test # the heartbeat path by calling _start_heartbeat directly import threading + stop = threading.Event() result = lifecycle._start_heartbeat("rsv_1", 0, MagicMock(), stop) assert result is None @@ -809,15 +876,21 @@ async def my_func() -> str: return "should not run" from runcycles.models import DryRunResult + result = await lifecycle.execute(my_func, (), {}, cfg) assert isinstance(result, DryRunResult) async def test_missing_reservation_id_raises(self) -> None: lifecycle, mock_client = self._make_lifecycle() - mock_client.create_reservation = AsyncMock(return_value=CyclesResponse.success(200, { - "decision": "ALLOW", - "affected_scopes": ["tenant:acme"], - })) + mock_client.create_reservation = AsyncMock( + return_value=CyclesResponse.success( + 200, + { + "decision": "ALLOW", + "affected_scopes": ["tenant:acme"], + }, + ) + ) cfg = _make_cfg() @@ -845,10 +918,13 @@ async def failing_fn() -> str: async def test_commit_finalized_does_not_release(self) -> None: lifecycle, mock_client = self._make_lifecycle() mock_client.create_reservation = AsyncMock(return_value=_allow_response()) - mock_client.commit_reservation = AsyncMock(return_value=CyclesResponse.http_error( - 409, "Finalized", - body={"error": "RESERVATION_FINALIZED", "message": "Done", "request_id": "r1"}, - )) + mock_client.commit_reservation = AsyncMock( + return_value=CyclesResponse.http_error( + 409, + "Finalized", + body={"error": "RESERVATION_FINALIZED", "message": "Done", "request_id": "r1"}, + ) + ) cfg = _make_cfg() @@ -861,10 +937,13 @@ async def my_func() -> str: async def test_commit_idempotency_mismatch_does_not_release(self) -> None: lifecycle, mock_client = self._make_lifecycle() mock_client.create_reservation = AsyncMock(return_value=_allow_response()) - mock_client.commit_reservation = AsyncMock(return_value=CyclesResponse.http_error( - 409, "Mismatch", - body={"error": "IDEMPOTENCY_MISMATCH", "message": "Mismatch", "request_id": "r1"}, - )) + mock_client.commit_reservation = AsyncMock( + return_value=CyclesResponse.http_error( + 409, + "Mismatch", + body={"error": "IDEMPOTENCY_MISMATCH", "message": "Mismatch", "request_id": "r1"}, + ) + ) cfg = _make_cfg() @@ -877,10 +956,13 @@ async def my_func() -> str: async def test_commit_client_error_triggers_release(self) -> None: lifecycle, mock_client = self._make_lifecycle() mock_client.create_reservation = AsyncMock(return_value=_allow_response()) - mock_client.commit_reservation = AsyncMock(return_value=CyclesResponse.http_error( - 400, "Bad", - body={"error": "UNIT_MISMATCH", "message": "Wrong unit", "request_id": "r1"}, - )) + mock_client.commit_reservation = AsyncMock( + return_value=CyclesResponse.http_error( + 400, + "Bad", + body={"error": "UNIT_MISMATCH", "message": "Wrong unit", "request_id": "r1"}, + ) + ) mock_client.release_reservation = AsyncMock(return_value=_release_success()) cfg = _make_cfg() @@ -901,9 +983,7 @@ async def test_commit_transport_error_schedules_retry(self) -> None: lifecycle = AsyncCyclesLifecycle(mock_client, retry_engine, {"tenant": "acme"}) mock_client.create_reservation = AsyncMock(return_value=_allow_response()) - mock_client.commit_reservation = AsyncMock( - return_value=CyclesResponse.transport_error(ConnectionError("down")) - ) + mock_client.commit_reservation = AsyncMock(return_value=CyclesResponse.transport_error(ConnectionError("down"))) cfg = _make_cfg() @@ -934,9 +1014,7 @@ async def my_func() -> str: async def test_release_failure_does_not_raise(self) -> None: lifecycle, mock_client = self._make_lifecycle() mock_client.create_reservation = AsyncMock(return_value=_allow_response()) - mock_client.release_reservation = AsyncMock( - return_value=CyclesResponse.http_error(500, "Release failed") - ) + mock_client.release_reservation = AsyncMock(return_value=CyclesResponse.http_error(500, "Release failed")) cfg = _make_cfg() @@ -963,9 +1041,15 @@ async def test_heartbeat_extends_reservation(self) -> None: lifecycle, mock_client = self._make_lifecycle() mock_client.create_reservation = AsyncMock(return_value=_allow_response()) mock_client.commit_reservation = AsyncMock(return_value=_commit_success()) - mock_client.extend_reservation = AsyncMock(return_value=CyclesResponse.success(200, { - "status": "ACTIVE", "expires_at_ms": 9999999999, - })) + mock_client.extend_reservation = AsyncMock( + return_value=CyclesResponse.success( + 200, + { + "status": "ACTIVE", + "expires_at_ms": 9999999999, + }, + ) + ) cfg = _make_cfg(ttl_ms=2000) @@ -981,9 +1065,7 @@ async def test_heartbeat_failure_does_not_crash(self) -> None: lifecycle, mock_client = self._make_lifecycle() mock_client.create_reservation = AsyncMock(return_value=_allow_response()) mock_client.commit_reservation = AsyncMock(return_value=_commit_success()) - mock_client.extend_reservation = AsyncMock( - return_value=CyclesResponse.http_error(500, "Extend failed") - ) + mock_client.extend_reservation = AsyncMock(return_value=CyclesResponse.http_error(500, "Extend failed")) cfg = _make_cfg(ttl_ms=2000) @@ -1023,10 +1105,13 @@ async def my_func() -> str: async def test_reservation_creation_failure_raises(self) -> None: lifecycle, mock_client = self._make_lifecycle() - mock_client.create_reservation = AsyncMock(return_value=CyclesResponse.http_error( - 500, "Internal error", - body={"error": "INTERNAL_ERROR", "message": "Server down", "request_id": "r1"}, - )) + mock_client.create_reservation = AsyncMock( + return_value=CyclesResponse.http_error( + 500, + "Internal error", + body={"error": "INTERNAL_ERROR", "message": "Server down", "request_id": "r1"}, + ) + ) cfg = _make_cfg() diff --git a/tests/test_retry.py b/tests/test_retry.py index cb13b01..817d18d 100644 --- a/tests/test_retry.py +++ b/tests/test_retry.py @@ -35,9 +35,7 @@ def test_commit_accepts_evidence_and_rejects_schema_invalid_optional_values(self }, } assert _is_schema_valid_commit_success(CyclesResponse.success(200, body)) - assert not _is_schema_valid_commit_success( - CyclesResponse.success(200, {**body, "balances": None}) - ) + assert not _is_schema_valid_commit_success(CyclesResponse.success(200, {**body, "balances": None})) assert not _is_schema_valid_commit_success( CyclesResponse.success( 200, @@ -58,9 +56,7 @@ def test_commit_accepts_evidence_and_rejects_schema_invalid_optional_values(self ) def test_event_follows_exact_wire_schema(self) -> None: - assert _is_schema_valid_event_success( - CyclesResponse.success(201, {"status": "APPLIED", "event_id": ""}) - ) + assert _is_schema_valid_event_success(CyclesResponse.success(201, {"status": "APPLIED", "event_id": ""})) assert not _is_schema_valid_event_success( CyclesResponse.success( 201, @@ -102,22 +98,29 @@ def test_disabled_does_not_retry(self, disabled_config: CyclesConfig) -> None: def test_retries_until_success(self, config: CyclesConfig) -> None: engine = CommitRetryEngine(config) mock_client = MagicMock() - # First call fails with 500, second succeeds + # The first response is lost after the request leaves the client. The + # recovery attempt must use the exact same settlement key. mock_client.commit_reservation.side_effect = [ - CyclesResponse.http_error(500, "Server error"), - CyclesResponse.success(200, { - "status": "COMMITTED", - "charged": {"unit": "USD_MICROCENTS", "amount": 1}, - }), + ConnectionError("response lost"), + CyclesResponse.success( + 200, + { + "status": "COMMITTED", + "charged": {"unit": "USD_MICROCENTS", "amount": 1}, + }, + ), ] engine.set_client(mock_client) # Run _retry_loop directly (synchronous, avoids thread timing issues) from runcycles.retry import _PendingCommit + pending = _PendingCommit(reservation_id="rsv_1", commit_body={"idempotency_key": "k1"}) engine._retry_loop(pending) assert mock_client.commit_reservation.call_count == 2 + sent = [call.args[1] for call in mock_client.commit_reservation.call_args_list] + assert [body["idempotency_key"] for body in sent] == ["k1", "k1"] def test_stops_on_client_error(self, config: CyclesConfig) -> None: engine = CommitRetryEngine(config) @@ -127,6 +130,7 @@ def test_stops_on_client_error(self, config: CyclesConfig) -> None: engine.set_client(mock_client) from runcycles.retry import _PendingCommit + pending = _PendingCommit(reservation_id="rsv_1", commit_body={"idempotency_key": "k1"}) engine._retry_loop(pending) @@ -140,6 +144,7 @@ def test_exhausts_retries(self, config: CyclesConfig) -> None: engine.set_client(mock_client) from runcycles.retry import _PendingCommit + pending = _PendingCommit(reservation_id="rsv_1", commit_body={"idempotency_key": "k1"}) engine._retry_loop(pending) @@ -151,14 +156,18 @@ def test_handles_exception_during_retry(self, config: CyclesConfig) -> None: # First call throws, second succeeds mock_client.commit_reservation.side_effect = [ ConnectionError("network down"), - CyclesResponse.success(200, { - "status": "COMMITTED", - "charged": {"unit": "USD_MICROCENTS", "amount": 1}, - }), + CyclesResponse.success( + 200, + { + "status": "COMMITTED", + "charged": {"unit": "USD_MICROCENTS", "amount": 1}, + }, + ), ] engine.set_client(mock_client) from runcycles.retry import _PendingCommit + pending = _PendingCommit(reservation_id="rsv_1", commit_body={"idempotency_key": "k1"}) engine._retry_loop(pending) @@ -168,6 +177,7 @@ def test_no_client_set(self, config: CyclesConfig) -> None: engine = CommitRetryEngine(config) # Don't set client — should bail out from runcycles.retry import _PendingCommit + pending = _PendingCommit(reservation_id="rsv_1", commit_body={"idempotency_key": "k1"}) engine._retry_loop(pending) # should not raise @@ -186,14 +196,18 @@ async def test_retries_until_success(self, config: CyclesConfig) -> None: mock_client = AsyncMock() mock_client.commit_reservation.side_effect = [ CyclesResponse.http_error(500, "Server error"), - CyclesResponse.success(200, { - "status": "COMMITTED", - "charged": {"unit": "USD_MICROCENTS", "amount": 1}, - }), + CyclesResponse.success( + 200, + { + "status": "COMMITTED", + "charged": {"unit": "USD_MICROCENTS", "amount": 1}, + }, + ), ] engine.set_client(mock_client) from runcycles.retry import _PendingCommit + pending = _PendingCommit(reservation_id="rsv_1", commit_body={"idempotency_key": "k1"}) await engine._retry_loop(pending) @@ -206,6 +220,7 @@ async def test_stops_on_client_error(self, config: CyclesConfig) -> None: engine.set_client(mock_client) from runcycles.retry import _PendingCommit + pending = _PendingCommit(reservation_id="rsv_1", commit_body={"idempotency_key": "k1"}) await engine._retry_loop(pending) @@ -218,6 +233,7 @@ async def test_exhausts_retries(self, config: CyclesConfig) -> None: engine.set_client(mock_client) from runcycles.retry import _PendingCommit + pending = _PendingCommit(reservation_id="rsv_1", commit_body={"idempotency_key": "k1"}) await engine._retry_loop(pending) @@ -228,14 +244,18 @@ async def test_handles_exception_during_retry(self, config: CyclesConfig) -> Non mock_client = AsyncMock() mock_client.commit_reservation.side_effect = [ ConnectionError("network down"), - CyclesResponse.success(200, { - "status": "COMMITTED", - "charged": {"unit": "USD_MICROCENTS", "amount": 1}, - }), + CyclesResponse.success( + 200, + { + "status": "COMMITTED", + "charged": {"unit": "USD_MICROCENTS", "amount": 1}, + }, + ), ] engine.set_client(mock_client) from runcycles.retry import _PendingCommit + pending = _PendingCommit(reservation_id="rsv_1", commit_body={"idempotency_key": "k1"}) await engine._retry_loop(pending) @@ -244,6 +264,7 @@ async def test_handles_exception_during_retry(self, config: CyclesConfig) -> Non async def test_no_client_set(self, config: CyclesConfig) -> None: engine = AsyncCommitRetryEngine(config) from runcycles.retry import _PendingCommit + pending = _PendingCommit(reservation_id="rsv_1", commit_body={"idempotency_key": "k1"}) await engine._retry_loop(pending) # should not raise @@ -264,15 +285,19 @@ def test_schedule_creates_thread(self, config: CyclesConfig) -> None: engine = CommitRetryEngine(config) mock_client = MagicMock() # Return success immediately so the thread finishes quickly - mock_client.commit_reservation.return_value = CyclesResponse.success(200, { - "status": "COMMITTED", - "charged": {"unit": "USD_MICROCENTS", "amount": 1}, - }) + mock_client.commit_reservation.return_value = CyclesResponse.success( + 200, + { + "status": "COMMITTED", + "charged": {"unit": "USD_MICROCENTS", "amount": 1}, + }, + ) engine.set_client(mock_client) engine.schedule("rsv_1", {"idempotency_key": "k1"}) # Give the thread time to run import time + time.sleep(0.1) assert mock_client.commit_reservation.call_count >= 1 From ecb9738d112c47174b997b40747094758a15810f Mon Sep 17 00:00:00 2001 From: Albert Mavashev Date: Wed, 29 Jul 2026 09:48:52 -0400 Subject: [PATCH 3/3] ci: pin recovery conformance profile 0.3 --- .github/workflows/ci.yml | 2 +- .github/workflows/python-publish.yml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5c548a7..39baf14 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -23,7 +23,7 @@ jobs: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: repository: runcycles/cycles-protocol - ref: 8a22803316d91b85e444f4fb40ff2b8426f11773 + ref: 594631c14710da08ad5e00125d899d642213c296 path: .cycles-protocol persist-credentials: false - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 diff --git a/.github/workflows/python-publish.yml b/.github/workflows/python-publish.yml index cbc0d4b..d705bb0 100644 --- a/.github/workflows/python-publish.yml +++ b/.github/workflows/python-publish.yml @@ -28,7 +28,7 @@ jobs: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: repository: runcycles/cycles-protocol - ref: 8a22803316d91b85e444f4fb40ff2b8426f11773 + ref: 594631c14710da08ad5e00125d899d642213c296 path: .cycles-protocol persist-credentials: false - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0