fix: address PR #89 review — journal identity isolation, 429 handling, flush deadline

Addresses all four review findings:

1. [P1] Replay isolated by credentials: journal records now live in a
   per-identity subdirectory keyed by a non-secret truncated SHA-256
   fingerprint of (base_url, api_key). Co-located clients with different
   servers or API keys can no longer replay each other's records — a
   foreign record previously got 401/403 and was discarded as terminal,
   permanently losing the spend.

2. [P1] HTTP 429 / LIMIT_EXCEEDED is transient, not terminal: both the
   commit and event classifiers now detect rate limiting (status 429 or
   error code), retain the journal entry, keep retrying, and make the
   next attempt wait at least the server's Retry-After (consumed once,
   max'd against the normal backoff). Consistent with
   ErrorCode.is_retryable, which already classified LIMIT_EXCEEDED
   retryable.

3. [P2] Replay claim scoped to the identity subdirectory: because the
   claim now covers exactly one (server, credential) identity, an engine
   for server A can never block server B's entries from replaying out of
   a shared journal_dir.

4. [P2] Process-wide flush deadline: _flush_all_engines() computes one
   deadline from the max engine timeout and passes each engine only the
   remaining budget, so worst-case shutdown is retry_flush_timeout, not
   engine_count x retry_flush_timeout.

No journal-layout migration needed — v0.5.0 is unreleased.

469 tests pass at 100% coverage; ruff and mypy --strict clean.
AUDIT.md, CHANGELOG.md, README.md updated.

Author: amavashev

AUDIT.md

@@ -1,6 +1,6 @@
 # Cycles Protocol v0.1.25 — Client (Python) Audit
 
-**Date:** 2026-07-27 (v0.5.0 — durable commit retries: on-disk pending-commit journal with next-run replay, bounded atexit flush, and `POST /v1/events` recovery for commits that land after reservation expiry; async retry-task GC fix; `retry_enabled=False` now journals instead of silently dropping. See the dated entry below. 460 tests pass at 100% coverage.),
+**Date:** 2026-07-27 (v0.5.0 — durable commit retries: on-disk pending-commit journal with next-run replay, bounded atexit flush, and `POST /v1/events` recovery for commits that land after reservation expiry; async retry-task GC fix; `retry_enabled=False` now journals instead of silently dropping. Review hardening: per-identity journal partitioning, 429 transient with `Retry-After`, process-wide flush deadline. See the dated entry below. 469 tests pass at 100% coverage.),
 2026-07-10 (v0.5.0 — `TENANT_CLOSED` + `LIMIT_EXCEEDED` error-code support. `TENANT_CLOSED` per runtime spec v0.1.25.13 (`cycles-protocol-v0.yaml`, runcycles/cycles-protocol#125): `ErrorCode.TENANT_CLOSED` enum member, `TenantClosedError` subclass wired into the lifecycle error-code→exception mapping (reservation-creation surfaces), `CyclesProtocolError.is_tenant_closed()` helper. `LIMIT_EXCEEDED` per runtime spec v0.1.25.12 (revision 2026-07-04, HTTP 429 rate limiting): enum-only member matching the `BUDGET_FROZEN`/`BUDGET_CLOSED` pattern, classified retryable at both the enum and exception layers (429 is transient; previously it fell through to `UNKNOWN`, which happened to be retryable, so semantics are unchanged — now typed). Enum reordered to mirror spec declaration order. Both purely additive; previously both codes fell through the `ErrorCode.from_string` forward-compat path to `UNKNOWN`. See the dated entries at the end of this file. 398 tests pass at 100% coverage.),
 2026-07-09 (README + docstring transport-error documentation fix, no version bump — see the dated entry at the end of this file. `CyclesTransportError` is exported but never raised by the SDK; README and its docstring now describe the actual `status == -1` surfacing.),
 2026-07-03 (integration-test-only, no version bump — `test_health_check` now probes the public `/actuator/health/readiness` endpoint instead of aggregate `/actuator/health`, which requires `X-Admin-API-Key` since cycles-server v0.1.25.45 and fails closed with 500 when the server has no admin key configured. The old assertion had failed the org nightly Full-Stack Integration every night since 2026-06-28. No library code change.),
@@ -23,7 +23,14 @@ has already returned the reserved budget to the pool — is recovered via
 `POST /v1/events` (spec-conformant `EventCreateRequest`, commit idempotency
 key reused, recovery markers in `metadata`). Also fixes the async engine's
 unreferenced-task GC hazard and the silent drop under `retry_enabled=False`.
-460 tests pass at 100% coverage.
+Post-review (PR #89) hardening: journal records are partitioned into
+per-identity subdirectories (SHA-256 fingerprint of base_url + api_key) so
+co-located clients with different credentials never replay or 401-discard
+each other's records and replay claims cannot cross identities; HTTP 429 /
+`LIMIT_EXCEEDED` is transient (record retained, `Retry-After` honored)
+instead of a terminal discard; the atexit flush enforces one process-wide
+`retry_flush_timeout` deadline instead of per-engine. 469 tests pass at
+100% coverage.
 
 ## 2026-07-26 — Python publishing workflow maintenance
 

CHANGELOG.md

@@ -11,9 +11,10 @@ Durable commit retries. Previously a commit that failed transiently lived only i
 
 ### Added
 
-- `runcycles.journal`: file-per-commit `CommitJournal` (atomic write, idempotent replay). Config: `journal_enabled` (default `True`), `journal_dir` (default `~/.runcycles/commit-journal`), `retry_flush_timeout` (default 10 s); env `CYCLES_JOURNAL_ENABLED`, `CYCLES_JOURNAL_DIR`, `CYCLES_RETRY_FLUSH_TIMEOUT`. The first engine created per journal directory replays surviving entries for its `base_url`; corrupt files are renamed `*.corrupt` for operator triage.
+- `runcycles.journal`: file-per-commit `CommitJournal` (atomic write, idempotent replay). Config: `journal_enabled` (default `True`), `journal_dir` (default `~/.runcycles/commit-journal`), `retry_flush_timeout` (default 10 s); env `CYCLES_JOURNAL_ENABLED`, `CYCLES_JOURNAL_DIR`, `CYCLES_RETRY_FLUSH_TIMEOUT`. Records are partitioned into per-identity subdirectories keyed by a non-secret SHA-256 fingerprint of `(base_url, api_key)`, so clients with different servers or credentials sharing a journal directory never replay — or 401-discard — each other's records, and one identity's replay claim cannot starve another's. The first engine created per identity replays surviving entries; corrupt files are renamed `*.corrupt` for operator triage.
 - Event fallback: when a commit (first attempt or retry) returns `RESERVATION_EXPIRED`, the SDK posts the spend to `/v1/events` reusing the commit's idempotency key, with `metadata.recovered_reservation_id` / `metadata.recovery_reason` markers and no `overage_policy` (spec default `ALLOW_IF_AVAILABLE` never rejects). Applies to the `@cycles` lifecycles and both streaming context managers. `RESERVATION_FINALIZED` is still treated as settled.
-- `flush()` on both retry engines; a process-wide `atexit` hook flushes sync engines for up to `retry_flush_timeout` seconds so daemon retry threads aren't killed mid-backoff on clean exit.
+- `flush()` on both retry engines; a process-wide `atexit` hook flushes sync engines under one shared `retry_flush_timeout` deadline (not per engine) so daemon retry threads aren't killed mid-backoff on clean exit and shutdown time stays bounded regardless of engine count.
+- Rate-limit awareness in the retry engines: HTTP 429 / `LIMIT_EXCEEDED` on a commit or event attempt is transient — the journal entry is retained and the next attempt waits at least the server's `Retry-After` (consistent with `ErrorCode.is_retryable`).
 
 ### Fixed
 

README.md

@@ -237,12 +237,19 @@ CyclesConfig(
 
 A commit records spend that has already happened, so the SDK never lets one
 exist only in memory. Every commit scheduled for background retry is first
-journaled to disk (`journal_dir`, default `~/.runcycles/commit-journal`) and
-removed only on a terminal outcome:
+journaled to disk and removed only on a terminal outcome. Records live under
+`journal_dir` (default `~/.runcycles/commit-journal`) in a per-identity
+subdirectory keyed by a non-secret fingerprint of `(base_url, api_key)`, so
+clients using different servers or credentials on the same machine never
+replay — or discard — each other's records:
 
 - **Process exit**: an `atexit` hook waits up to `retry_flush_timeout` seconds
-  for in-flight retries; anything unfinished stays journaled and is replayed
-  automatically the next time the process creates a client lifecycle.
+  (one process-wide budget shared across all engines) for in-flight retries;
+  anything unfinished stays journaled and is replayed automatically the next
+  time the process creates a client lifecycle.
+- **Rate limiting**: HTTP 429 / `LIMIT_EXCEEDED` responses are transient —
+  the journal entry is kept and the next attempt waits at least the server's
+  `Retry-After`.
 - **Reservation expired before the commit landed**: the server has already
   returned the reserved budget to the pool, so the SDK re-records the spend
   via `POST /v1/events` (the protocol's post-hoc direct-debit endpoint),

runcycles/journal.py

@@ -15,6 +15,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import json
 import logging
 import time
@@ -33,6 +34,19 @@ def default_journal_dir() -> Path:
     return Path.home() / ".runcycles" / "commit-journal"
 
 
+def auth_fingerprint(base_url: str, api_key: str) -> str:
+    """Non-secret identity for one (server, credential) pair.
+
+    Journal records are stored under a per-identity subdirectory so that
+    clients sharing a journal directory but using different servers or API
+    keys never replay (and on 401/403, discard) each other's records. A
+    truncated SHA-256 is not reversible and API keys are high-entropy, so
+    the fingerprint is safe to use as a directory name.
+    """
+    digest = hashlib.sha256(f"{base_url}\n{api_key}".encode()).hexdigest()
+    return digest[:16]
+
+
 def _safe_filename(reservation_id: str) -> str:
     sanitized = "".join(c if c.isalnum() or c in "-_" else "_" for c in reservation_id)
     return f"{sanitized}{_SUFFIX}"

runcycles/retry.py

@@ -37,6 +37,9 @@ class _PendingCommit:
     event_fallback_body: dict[str, Any] | None = None
     mode: str = "commit"  # "commit" | "event"
     attempt: int = 0
+    # Server-requested minimum delay (seconds) before the next attempt,
+    # set from a 429's Retry-After and consumed by the retry loop.
+    retry_after_s: float | None = None
 
 
 def _extract_error_code(response: CyclesResponse) -> str | None:
@@ -47,9 +50,12 @@ def _extract_error_code(response: CyclesResponse) -> str | None:
     return raw if isinstance(raw, str) else None
 
 
-# Journal replay must happen at most once per journal directory per process:
-# the first engine created for a directory claims it and replays surviving
-# entries; later engines (and the claimer's own in-flight work) are excluded.
+# Journal replay must happen at most once per identity directory per process:
+# the first engine created for a (server, credential) identity claims its
+# subdirectory and replays surviving entries; later engines (and the
+# claimer's own in-flight work) are excluded. Because the claim is scoped to
+# the identity subdirectory — not the shared parent — an engine for one
+# server/key can never block another identity's entries from replaying.
 _replay_lock = threading.Lock()
 _replayed_dirs: set[Path] = set()
 
@@ -72,8 +78,18 @@ def _claim_replay(directory: Path) -> bool:
 
 
 def _flush_all_engines() -> None:
-    for engine in list(_live_engines):
-        engine.flush()
+    # One process-wide deadline: retry_flush_timeout bounds the whole exit
+    # wait, not each engine. With several engines (decorators, streams) the
+    # remaining budget shrinks as each is flushed.
+    engines = list(_live_engines)
+    if not engines:
+        return
+    deadline = time.monotonic() + max(engine._flush_timeout for engine in engines)
+    for engine in engines:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            return
+        engine.flush(remaining)
 
 
 def _register_engine_for_flush(engine: CommitRetryEngine) -> None:
@@ -99,8 +115,12 @@ def __init__(self, config: CyclesConfig) -> None:
         self._client: Any = None  # set by lifecycle to avoid circular import
         self._journal: CommitJournal | None = None
         if config.journal_enabled:
-            directory = Path(config.journal_dir) if config.journal_dir else _journal.default_journal_dir()
-            self._journal = CommitJournal(directory)
+            base = Path(config.journal_dir) if config.journal_dir else _journal.default_journal_dir()
+            # Per-identity subdirectory: clients sharing a journal directory
+            # but using different servers or API keys must never replay each
+            # other's records (a foreign record would 401/403 and be
+            # discarded as terminal — permanent spend loss).
+            self._journal = CommitJournal(base / _journal.auth_fingerprint(config.base_url, config.api_key))
 
     def _journal_record(self, pending: _PendingCommit) -> None:
         if self._journal is not None:
@@ -151,8 +171,27 @@ def _log_disabled_drop(self, pending: _PendingCommit) -> None:
                 pending.mode, pending.reservation_id,
             )
 
-    def _delay_for(self, attempt: int) -> float:
-        return min(self._initial_delay * (self._multiplier**attempt), self._max_delay)
+    def _delay_for(self, pending: _PendingCommit) -> float:
+        delay = min(self._initial_delay * (self._multiplier ** pending.attempt), self._max_delay)
+        if pending.retry_after_s is not None:
+            # A 429 asked us to wait: honor the server's Retry-After when it
+            # exceeds our own backoff, then clear it — it applies once.
+            delay = max(delay, pending.retry_after_s)
+            pending.retry_after_s = None
+        return delay
+
+    def _is_rate_limited(self, pending: _PendingCommit, response: CyclesResponse) -> bool:
+        """Detect 429 / LIMIT_EXCEEDED and stash its Retry-After. Transient, never terminal."""
+        if response.status != 429 and _extract_error_code(response) != "LIMIT_EXCEEDED":
+            return False
+        retry_after_ms = response.retry_after_ms_header
+        if retry_after_ms is not None:
+            pending.retry_after_s = retry_after_ms / 1000.0
+        logger.warning(
+            "%s retry rate-limited: reservation_id=%s, attempt=%d, retry_after_ms=%s",
+            pending.mode, pending.reservation_id, pending.attempt, retry_after_ms,
+        )
+        return True
 
     def _classify_commit_response(self, pending: _PendingCommit, response: CyclesResponse) -> bool:
         """Handle a commit attempt's response. Returns True when terminal.
@@ -167,6 +206,8 @@ def _classify_commit_response(self, pending: _PendingCommit, response: CyclesRes
             )
             self._journal_discard(pending.reservation_id)
             return True
+        if self._is_rate_limited(pending, response):
+            return False
         if response.is_client_error:
             code = _extract_error_code(response)
             if code == "RESERVATION_EXPIRED":
@@ -207,6 +248,8 @@ def _classify_event_response(self, pending: _PendingCommit, response: CyclesResp
             )
             self._journal_discard(pending.reservation_id)
             return True
+        if self._is_rate_limited(pending, response):
+            return False
         if response.is_client_error:
             logger.error(
                 "Event fallback rejected (%s); spend recovery failed: reservation_id=%s, status=%d",
@@ -306,7 +349,7 @@ def flush(self, timeout: float | None = None) -> None:
 
     def _retry_loop(self, pending: _PendingCommit) -> None:
         while pending.attempt < self._max_attempts:
-            delay = self._delay_for(pending.attempt)
+            delay = self._delay_for(pending)
             pending.attempt += 1
             logger.info(
                 "Scheduling %s retry: reservation_id=%s, attempt=%d/%d, delay=%.1fs",
@@ -422,7 +465,7 @@ async def flush(self, timeout: float | None = None) -> None:
 
     async def _retry_loop(self, pending: _PendingCommit) -> None:
         while pending.attempt < self._max_attempts:
-            delay = self._delay_for(pending.attempt)
+            delay = self._delay_for(pending)
             pending.attempt += 1
             logger.info(
                 "Scheduling async %s retry: reservation_id=%s, attempt=%d/%d, delay=%.1fs",