Harden CLIVE-I Pagination and Retries

Add lazy pagination, rate limiting, retries, and typed failures to CLIVE-I.

Introduction

30 Second Summary

A list can look complete while quietly stopping after its first page. A burst of traffic can turn the next request into a failure.

In this project, you will harden CLIVE-I's synchronous Simpro client so callers can iterate through validated records across every page. You will add a deterministic resilience policy without changing its synchronous architecture.

What You'll Build

At handoff, you will watch the offline gate pass against your review-ready client without Docker or a live service.

By the end of this project, you'll have:

  • Stream every page by calling iter_all() to receive validated Job models one at a time. Malformed pagination metadata surfaces as SimproProtocolError.
  • Recover predictably from 429 responses using bounded delays. Every request attempt passes through one shared token bucket under a stable correlation ID.
  • Prove compatibility offline with focused RESPX tests plus the complete quality gate. Confirm the synchronous HTTPX call path remains compatible. Confirm the raw JSON methods remain available.
  • Secret Mission: Trace the mixed recovery path from 401 through 429 to success as one state machine. Match each request attempt to its limiter acquisition. Explain the token refresh under one unchanged correlation header.

Are there any prerequisites?

You need access to the clean CLIVE-I checkout on its Linux server. Expected familiarity covers Python classes plus generators plus pytest.

Before We Start

This checkpoint fixes the compatibility boundary before you change the existing CLIVE-I Simpro client. Its synchronous HTTPX architecture and raw JSON methods stay in place.

Verify the Baseline and Freeze the Design

A clean public Git branch cannot prove that the Linux checkout in front of you is unchanged. Local edits or regressions would make every later test result ambiguous.

This checkpoint captures the local starting point before any source file changes. It also freezes the compatibility boundaries for the existing synchronous HTTPX client.

In this step, get ready to:
  • Confirm the checkout is clean at a recorded starting commit.
  • Synchronize the pinned project environment.
  • Prove the baseline passes within the approved design boundaries.
Confirm the clean checkout

Your local working tree is the real starting point for this project. Capturing its status protects the later diff from unrelated changes.

  • Use the terminal panel in VS Code on the Linux server for this checkpoint.
  • Select the terminal attached to the existing CLIVE-I checkout.
  • Check the checkout for local changes by running this command:
git status --short

What does this check prove?

This check exposes staged changes, working-tree changes, or untracked files that would contaminate the baseline.

An empty result confirms that the checkout has no local changes.

Status output is not empty?

  • Stop this checkpoint if any status line appears.
  • Protect intentional work through your team's Git workflow before continuing.

Ask for help before changing the checkout if you do not recognize a listed file: Help me interpret my Git status

  • Identify the exact starting commit by running this command:
git rev-parse --verify HEAD

Why record the starting commit?

The commit ID anchors every later source change to one exact version of CLIVE-I. Reviewers can use it to reproduce your starting state.

You should see a commit ID printed in the terminal.

  • Record the printed value as your starting commit ID.

That gives you an audit anchor for every change that follows.

Synchronize the pinned environment

The project environment must match the repository before its baseline results mean anything. Synchronization aligns installed dependencies with the existing lockfile.

  • Synchronize the project dependencies by running this command:
uv sync

What does synchronization do?

This command aligns the project environment with its dependency declarations and lockfile. The development dependency group is included by default.

You should see the command return to the terminal prompt without an error.

Dependency synchronization failed?

  • Leave the dependency declarations unchanged.
  • Leave the lockfile unchanged.
  • Read the first dependency or lock mismatch reported in the terminal.

Use the reported message when asking for help: Help me diagnose the uv sync failure

Freeze the approved design

A passing baseline protects current behavior. The design boundaries protect compatibility while pagination and resilience are added later.

  • Confirm that the new public pagination entry point is ResourceEndpoint.iter_all().
  • Preserve fetch_page().
  • Preserve the raw JSON client methods.
  • Keep one synchronous SimproClient as the shared transport boundary.
  • Exclude Docker from the implementation.
  • Exclude live API calls from the test strategy.

Why freeze these boundaries?

The existing public methods remain available to callers who depend on raw responses. The new behavior extends the client without replacing its synchronous transport model.

Before you run the gate, consider whether the untouched checkout will confirm one consistent baseline.

  • Run the complete checkpoint gate by running these commands:
git status --short
git rev-parse --verify HEAD
uv run --locked pytest tests/test_auth.py tests/test_client.py tests/test_endpoints.py

What does the gate verify?

  • The first Git check confirms that dependency synchronization left the checkout clean.
  • The second Git check prints the exact commit behind the baseline.
  • The pytest run checks authentication, client behavior, and endpoint compatibility without updating the lockfile.

You should see no status lines before the commit ID. That printed ID should match your starting commit ID.

The test summary should report no failures across the three selected test files.

The baseline gate did not pass?

  • Stop if the final status check lists a file.
  • Read the first failing test before changing any source code.
  • Confirm that the failing command ran inside the existing CLIVE-I checkout.

Share the first failure when asking for help: Help me diagnose my CLIVE-I baseline failure

Your checkout now has a clean, reproducible baseline with frozen compatibility boundaries. Next, you will add lazy typed pagination without disturbing the existing page and raw response methods.

Add Lazy Multi-Page Iteration

Your clean baseline proves that CLIVE-I works before this change. The next gap appears when a caller needs more than one API page.

The existing fetch_page() method returns one page at a time. A naive collector could load every result into memory or accept contradictory pagination metadata.

You will add a lazy generator that defers its first request until iteration begins. It will validate each page before yielding typed models in order.

In this step, get ready to:
  • Add a typed protocol exception for invalid pagination metadata.
  • Add lazy multi-page iteration to every resource endpoint.
  • Prove deferred requests and protocol validation with offline pagination tests.
Define typed pagination failures

Pagination headers form a contract between the API and the client. SimproProtocolError gives callers a typed failure when that contract is missing or contradictory.

  • In src/simpro_client/exceptions.py, add this class below SimproNotFoundError:
class SimproProtocolError(SimproError):
    def __init__(
        self,
        message: str,
        *,
        method: str | None = None,
        url: str | None = None,
        correlation_id: str | None = None,
    ) -> None:
        self.method = method
        self.url = url
        self.correlation_id = correlation_id
        super().__init__(message)

What does this exception preserve?

  • The SimproError base class keeps the existing exception interface compatible.
  • The method field records the request method involved in the protocol failure.
  • The url field identifies the affected resource path.
  • The correlation_id field connects the failure to the surrounding request trace.
  • Save src/simpro_client/exceptions.py.
  • Confirm that your editor shows no syntax diagnostics for SimproProtocolError.

✔️ Awesome, I've got everything!

Your exception hierarchy now includes the protocol failure used by pagination validation.

ⓧ I'd like to double check the full code

Compare src/simpro_client/exceptions.py with this complete checkpoint version.

"""Typed exception hierarchy for Simpro API errors."""


class SimproError(Exception):
    def __init__(self, message: str) -> None:
        self.message = message
        super().__init__(message)


class SimproAuthError(SimproError):
    pass


class SimproAPIError(SimproError):
    def __init__(
        self,
        message: str,
        status_code: int,
        response_body: str = "",
        *,
        method: str | None = None,
        url: str | None = None,
        correlation_id: str | None = None,
        retry_count: int = 0,
    ) -> None:
        self.status_code = status_code
        self.response_body = response_body
        self.method = method
        self.url = url
        self.correlation_id = correlation_id
        self.retry_count = retry_count
        super().__init__(message)


class SimproNotFoundError(SimproAPIError):
    def __init__(self, message: str = "Resource not found", **context) -> None:
        super().__init__(message, 404, **context)


class SimproProtocolError(SimproError):
    def __init__(
        self,
        message: str,
        *,
        method: str | None = None,
        url: str | None = None,
        correlation_id: str | None = None,
    ) -> None:
        self.method = method
        self.url = url
        self.correlation_id = correlation_id
        super().__init__(message)

What should match?

The existing exception constructors remain intact. The new protocol exception adds optional request context without requiring a response status.

Build the lazy endpoint iterator

A generator pauses at each yielded model. That pause lets a caller consume the current page before the client requests another one.

Why place iteration on ResourceEndpoint?

Every typed resource endpoint already inherits from ResourceEndpoint. Adding iter_all() there gives all endpoints the same pagination behavior.

The method continues using the synchronous SimproClient response path. Existing raw JSON methods and fetch_page() remain available.

  • In src/simpro_client/endpoints/base.py, replace the import and page-definition section with this code:
"""Reusable read-only endpoint behavior."""

from collections.abc import Iterator, Mapping
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Generic, TypeVar

from pydantic import BaseModel

from simpro_client.exceptions import SimproProtocolError
from simpro_client.logging import get_correlation_id

if TYPE_CHECKING:
    from httpx import Headers
    from simpro_client.client import SimproClient

ModelT = TypeVar("ModelT", bound=BaseModel)
FilterValue = str | int | float | bool


@dataclass(frozen=True)
class Page(Generic[ModelT]):
    items: list[ModelT]
    page: int
    page_size: int
    total: int
    count: int
    pages: int

What does this section prepare?

  • The Iterator return type describes a stream of validated models.
  • The Headers type documents the input accepted by pagination header validation.
  • The generic Page type preserves the model type returned by fetch_page().
  • The get_correlation_id() helper provides trace context for protocol failures.
  • Save src/simpro_client/endpoints/base.py.
  • Confirm that your editor recognizes Iterator and SimproProtocolError without unresolved-import diagnostics.
  • In ResourceEndpoint, add iter_all() immediately below fetch_page():
    def iter_all(self, *, company_id: int | None = None, page_size: int = 30, filters: Mapping[str, FilterValue] | None = None, **scope_ids: int) -> Iterator[ModelT]:
        values = self._route_values(company_id, scope_ids)
        path = self._render(self.collection_path, values)
        page = 1
        while True:
            params: dict[str, Any] = dict(filters or {})
            params.update(page=page, pageSize=page_size)
            response = self._client._get_response(path, params=params)
            items = [self.model.model_validate(item) for item in response.json()]
            pages = self._header_int(response.headers, "Result-Pages", path)
            if pages < 1:
                raise self._protocol_error(f"Invalid Result-Pages={pages}", path)
            if page > pages:
                raise self._protocol_error(f"Page {page} exceeds Result-Pages={pages}", path)
            if not items and page < pages:
                raise self._protocol_error(f"Empty page {page} before Result-Pages={pages}", path)
            yield from items
            if page == pages:
                return
            page += 1

How does lazy paging work?

  • The generator body begins running only when the caller requests its first value.
  • A fresh parameter dictionary preserves the caller's filters for every page.
  • Pydantic validates each response item through model_validate() before the item reaches the caller.
  • The yield from items statement pauses iteration after the current page has been consumed.
  • The metadata checks reject impossible page counts and premature empty pages.
  • Save src/simpro_client/endpoints/base.py.
  • Confirm that your editor shows no syntax diagnostics inside iter_all().
  • In the same file, replace the existing pagination-header helper with these two helpers:
    @classmethod
    def _header_int(cls, headers: "Headers", name: str, path: str) -> int:
        value = headers.get(name)
        if value is None:
            raise cls._protocol_error(f"Missing pagination header: {name}", path)
        try:
            return int(value)
        except ValueError as exc:
            raise cls._protocol_error(f"Invalid pagination header: {name}", path) from exc

    @staticmethod
    def _protocol_error(message: str, path: str) -> SimproProtocolError:
        return SimproProtocolError(message, method="GET", url=path, correlation_id=get_correlation_id())

What do the helpers validate?

The _header_int() helper converts pagination metadata into an integer. Missing or non-integer values become SimproProtocolError instances.

The _protocol_error() helper adds the request method and path. It also captures the current correlation ID.

  • Save src/simpro_client/endpoints/base.py.
  • Confirm that your editor shows no syntax diagnostics for the completed endpoint module.

✔️ Awesome, I've got everything!

Your endpoint base now retains fetch_page() while exposing lazy typed iteration through iter_all().

ⓧ I'd like to double check the full code

Compare src/simpro_client/endpoints/base.py with this complete checkpoint version.

"""Reusable read-only endpoint behavior."""

from collections.abc import Iterator, Mapping
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Generic, TypeVar

from pydantic import BaseModel

from simpro_client.exceptions import SimproProtocolError
from simpro_client.logging import get_correlation_id

if TYPE_CHECKING:
    from httpx import Headers
    from simpro_client.client import SimproClient

ModelT = TypeVar("ModelT", bound=BaseModel)
FilterValue = str | int | float | bool


@dataclass(frozen=True)
class Page(Generic[ModelT]):
    items: list[ModelT]
    page: int
    page_size: int
    total: int
    count: int
    pages: int


class ResourceEndpoint(Generic[ModelT]):
    model: type[ModelT]
    collection_path: str
    detail_path: str
    item_key: str

    def __init__(self, client: "SimproClient") -> None:
        self._client = client

    def get(self, item_id: int, *, company_id: int | None = None, **scope_ids: int) -> ModelT:
        values = self._route_values(company_id, scope_ids)
        values[self.item_key] = item_id
        payload = self._client.get(self._render(self.detail_path, values))
        return self.model.model_validate(payload)

    def fetch_page(self, *, company_id: int | None = None, page: int = 1, page_size: int = 30, filters: Mapping[str, FilterValue] | None = None, **scope_ids: int) -> Page[ModelT]:
        values = self._route_values(company_id, scope_ids)
        path = self._render(self.collection_path, values)
        params: dict[str, Any] = dict(filters or {})
        params.update(page=page, pageSize=page_size)
        response = self._client._get_response(path, params=params)
        items = [self.model.model_validate(item) for item in response.json()]
        return Page(items=items, page=page, page_size=page_size, total=self._header_int(response.headers, "Result-Total", path), count=self._header_int(response.headers, "Result-Count", path), pages=self._header_int(response.headers, "Result-Pages", path))

    def iter_all(self, *, company_id: int | None = None, page_size: int = 30, filters: Mapping[str, FilterValue] | None = None, **scope_ids: int) -> Iterator[ModelT]:
        values = self._route_values(company_id, scope_ids)
        path = self._render(self.collection_path, values)
        page = 1
        while True:
            params: dict[str, Any] = dict(filters or {})
            params.update(page=page, pageSize=page_size)
            response = self._client._get_response(path, params=params)
            items = [self.model.model_validate(item) for item in response.json()]
            pages = self._header_int(response.headers, "Result-Pages", path)
            if pages < 1:
                raise self._protocol_error(f"Invalid Result-Pages={pages}", path)
            if page > pages:
                raise self._protocol_error(f"Page {page} exceeds Result-Pages={pages}", path)
            if not items and page < pages:
                raise self._protocol_error(f"Empty page {page} before Result-Pages={pages}", path)
            yield from items
            if page == pages:
                return
            page += 1

    @staticmethod
    def _route_values(company_id, scope_ids):
        values = dict(scope_ids)
        if company_id is not None:
            values["company_id"] = company_id
        return values

    @staticmethod
    def _render(template, values):
        try:
            return template.format(**values)
        except KeyError as exc:
            raise ValueError(f"Missing route parameter: {exc.args[0]}") from exc

    @classmethod
    def _header_int(cls, headers: "Headers", name: str, path: str) -> int:
        value = headers.get(name)
        if value is None:
            raise cls._protocol_error(f"Missing pagination header: {name}", path)
        try:
            return int(value)
        except ValueError as exc:
            raise cls._protocol_error(f"Invalid pagination header: {name}", path) from exc

    @staticmethod
    def _protocol_error(message: str, path: str) -> SimproProtocolError:
        return SimproProtocolError(message, method="GET", url=path, correlation_id=get_correlation_id())

What should remain compatible?

The get() and fetch_page() methods remain public. The new iterator reuses the same synchronous response path and model validation behavior.

Prove deferred paging offline

Pagination needs behavioral tests because a correct final list can still hide eager requests. RESPX can record each mocked request without contacting a live API.

What must the pagination suite prove?

  • Creating an iter_all() iterator sends no resource request.
  • Requesting the first value fetches only page 1.
  • Consuming the current page triggers the next page request.
  • Empty and single-page responses finish cleanly.
  • Multi-page responses preserve model types and item order.
  • The Status filter remains unchanged on every page request.
  • Missing or non-integer Result-Pages metadata raises SimproProtocolError.
  • Zero or negative page counts raise SimproProtocolError.
  • Contradictory page counts or premature empty pages raise SimproProtocolError.
  • Create tests/test_pagination.py inside the existing tests directory using your editor's file tree.

You should see test_pagination.py listed beside the existing endpoint tests.

  • Write offline pagination tests that cover the contract above.
  • Save tests/test_pagination.py.

Before you run the gate, do you expect creating the iterator to send a request or waiting for the first consumed value to send it?

  • Run the focused pagination gate from the VS Code terminal by running this command:
uv run --locked pytest tests/test_pagination.py

What does this gate prove?

The gate passes when deferred execution, typed ordering, preserved filters, and protocol failures all match the pagination contract.

The request-count assertions also expose the intended shortfall. Consuming one page can trigger the next HTTP request immediately because this step adds no request metering.

Pagination tests failing?

Check that iter_all() contains a yield inside its generator body. Building a list before returning would send requests eagerly.

Check that every page rebuilds parameters from the original filters. Mutating the caller's filter mapping can change later requests.

Check that _header_int() converts missing and malformed pagination values into SimproProtocolError.

Help me debug my pagination tests.

Your resource endpoints now stream validated models across every reported page. Next, you will control the rapid request cadence exposed by this lazy iterator.

Enforce a Shared Token Bucket

Your lazy iterator now streams typed resources one page at a time. Consecutive pages can still send requests faster than the approved client policy permits.

A shared token bucket controls that pace at the central HTTP boundary. Every endpoint draws from the same bucket because every endpoint belongs to one synchronous client.

In this step, get ready to:
  • Build a thread-safe token bucket with deterministic time controls.
  • Configure a burst capacity of 8 with a refill rate of 8.0 tokens per second.
  • Connect one shared limiter to the client request boundary.
Build the token bucket

The bucket begins full so callers can use the approved burst immediately. Injected clock and sleeper functions make every refill calculation deterministic in tests.

  • Create src/simpro_client/rate_limiter.py from the VS Code Explorer sidebar.
  • Add the bucket state by copying this code into src/simpro_client/rate_limiter.py:
"""Thread-safe token-bucket rate limiting."""

import time
from collections.abc import Callable
from threading import Lock


class TokenBucket:
    def __init__(
        self,
        refill_rate: float = 8.0,
        capacity: int = 8,
        *,
        clock: Callable[[], float] = time.monotonic,
        sleeper: Callable[[float], None] = time.sleep,
    ) -> None:
        if refill_rate <= 0:
            raise ValueError("refill_rate must be greater than zero")
        if capacity <= 0:
            raise ValueError("capacity must be greater than zero")
        self._refill_rate = refill_rate
        self._capacity = float(capacity)
        self._tokens = float(capacity)
        self._clock = clock
        self._sleeper = sleeper
        self._updated_at = clock()
        self._lock = Lock()

How is the bucket initialized?

  • The constructor rejects non-positive settings before they can produce invalid wait calculations.
  • The _tokens balance starts at the configured capacity. This makes the initial burst available immediately.
  • The injected clock measures refill time without depending on wall-clock changes.
  • The injected sleeper lets tests record waits without pausing the test suite.
  • Save src/simpro_client/rate_limiter.py.
  • Confirm rate_limiter.py appears beside client.py in the VS Code Explorer sidebar.

Can't find the limiter module?

Check that the file is inside src/simpro_client.

Confirm the filename contains an underscore between rate and limiter.

Help me check the limiter module location.

Each acquisition refills the balance before consuming one token. A depleted bucket calculates its wait under the lock but performs the sleep after releasing that lock.

  • Add acquire() below __init__() in the TokenBucket class by copying this code:
    def acquire(self) -> None:
        while True:
            with self._lock:
                now = self._clock()
                elapsed = max(0.0, now - self._updated_at)
                self._tokens = min(
                    self._capacity,
                    self._tokens + elapsed * self._refill_rate,
                )
                self._updated_at = now
                if self._tokens >= 1.0:
                    self._tokens -= 1.0
                    return
                wait_seconds = (1.0 - self._tokens) / self._refill_rate
            self._sleeper(wait_seconds)

How does acquisition work?

  • Elapsed monotonic time becomes new tokens according to _refill_rate.
  • The min() calculation prevents the balance from exceeding its capacity.
  • The method consumes one token when the balance reaches at least 1.0.
  • The call to _sleeper sits outside the lock context. Another thread can inspect the bucket while one caller waits.
  • Save src/simpro_client/rate_limiter.py.
  • Check the VS Code Problems panel for syntax errors in rate_limiter.py.

You should see no syntax errors for the new module.

Seeing an indentation problem?

Match the indentation of def acquire with def __init__.

Keep self._sleeper(wait_seconds) outside the with self._lock block.

Help me inspect the bucket indentation.

A fake clock turns waiting into recorded arithmetic. This lets pytest verify exact delays without using real sleep.

  • Create tests/test_rate_limiter.py from the VS Code Explorer sidebar.
  • Add the controlled clock by copying this code into tests/test_rate_limiter.py:
from simpro_client.rate_limiter import TokenBucket


class FakeTime:
    def __init__(self):
        self.now = 0.0
        self.sleeps = []

    def clock(self):
        return self.now

    def sleep(self, seconds):
        self.sleeps.append(seconds)
        self.now += seconds

What does FakeTime control?

  • The clock() method returns the controlled time.
  • The sleep() method records each requested delay.
  • Each fake sleep advances now. The next acquisition sees enough elapsed time to refill.
  • Save tests/test_rate_limiter.py.
  • Check the VS Code Problems panel for syntax errors in test_rate_limiter.py.

You should see no syntax errors in the controlled clock.

Is the test module invalid?

Confirm the import points to simpro_client.rate_limiter.

Check that FakeTime is defined at the left edge of the file.

Help me fix the FakeTime test module.

  • Add the burst test below FakeTime by copying this code:
def test_burst_refill_and_wait():
    fake = FakeTime()
    bucket = TokenBucket(2.0, 2, clock=fake.clock, sleeper=fake.sleep)
    bucket.acquire()
    bucket.acquire()
    assert fake.sleeps == []
    bucket.acquire()
    assert fake.sleeps == [0.5]

What does the burst test prove?

  • The first two acquisitions consume the full capacity without sleeping.
  • The third acquisition needs one token at a refill rate of 2.0 tokens per second.
  • The recorded wait of 0.5 seconds proves the refill calculation.
  • Save tests/test_rate_limiter.py.

Before you run the test, how long should the third acquisition wait after two tokens have been consumed?

  • Run the focused bucket test from the existing VS Code terminal with this command:
uv run --locked pytest tests/test_rate_limiter.py

What should you see?

The focused test completes without failures. This proves the bucket permits two immediate acquisitions before recording a half-second wait.

Is the burst test failing?

Check that _tokens starts from capacity.

Confirm the wait calculation divides the missing token amount by _refill_rate.

Help me debug the burst calculation.

The second test makes the locking rule observable. Its sleeper tries to acquire the bucket lock before advancing fake time.

  • Add the lock-release test below test_burst_refill_and_wait() by copying this code:
def test_lock_is_released_before_sleep():
    fake = FakeTime()
    bucket = TokenBucket(1.0, 1, clock=fake.clock, sleeper=fake.sleep)
    bucket.acquire()

    def sleep(seconds):
        assert bucket._lock.acquire(blocking=False)
        bucket._lock.release()
        fake.sleep(seconds)

    bucket._sleeper = sleep
    bucket.acquire()
    assert fake.sleeps == [1.0]

How does the test inspect the lock?

  • The first acquisition empties a bucket with capacity 1.
  • The replacement sleeper tries to acquire bucket._lock without blocking.
  • A successful acquisition proves the original lock context has ended before sleeping.
  • The recorded one-second wait matches a refill rate of one token per second.
  • Save tests/test_rate_limiter.py.

Before you run both tests, do you expect the replacement sleeper to acquire the bucket lock successfully?

  • Run both bucket tests from the existing VS Code terminal with this command:
uv run --locked pytest tests/test_rate_limiter.py

What should you see now?

Both rate-limiter tests complete without failures. The bucket now has deterministic evidence for burst behavior.

The lock test proves that sleeping occurs outside the protected section.

Does the lock assertion fail?

Check that self._sleeper(wait_seconds) has less indentation than the code inside with self._lock.

Keep wait_seconds inside the loop so each retry uses the current balance.

Help me fix the sleep-under-lock failure.

✔️ Awesome, I've got everything!

Your bucket implementation has passed its two focused tests. Save both files before continuing.

ⓧ I'd like to double check the full code

Compare src/simpro_client/rate_limiter.py with this complete file.

"""Thread-safe token-bucket rate limiting."""

import time
from collections.abc import Callable
from threading import Lock


class TokenBucket:
    def __init__(
        self,
        refill_rate: float = 8.0,
        capacity: int = 8,
        *,
        clock: Callable[[], float] = time.monotonic,
        sleeper: Callable[[float], None] = time.sleep,
    ) -> None:
        if refill_rate <= 0:
            raise ValueError("refill_rate must be greater than zero")
        if capacity <= 0:
            raise ValueError("capacity must be greater than zero")
        self._refill_rate = refill_rate
        self._capacity = float(capacity)
        self._tokens = float(capacity)
        self._clock = clock
        self._sleeper = sleeper
        self._updated_at = clock()
        self._lock = Lock()

    def acquire(self) -> None:
        while True:
            with self._lock:
                now = self._clock()
                elapsed = max(0.0, now - self._updated_at)
                self._tokens = min(
                    self._capacity,
                    self._tokens + elapsed * self._refill_rate,
                )
                self._updated_at = now
                if self._tokens >= 1.0:
                    self._tokens -= 1.0
                    return
                wait_seconds = (1.0 - self._tokens) / self._refill_rate
            self._sleeper(wait_seconds)

What should match?

The lock protects refill arithmetic. The sleeper runs after the lock context ends.

Compare tests/test_rate_limiter.py with this complete file.

from simpro_client.rate_limiter import TokenBucket


class FakeTime:
    def __init__(self):
        self.now = 0.0
        self.sleeps = []

    def clock(self):
        return self.now

    def sleep(self, seconds):
        self.sleeps.append(seconds)
        self.now += seconds


def test_burst_refill_and_wait():
    fake = FakeTime()
    bucket = TokenBucket(2.0, 2, clock=fake.clock, sleeper=fake.sleep)
    bucket.acquire()
    bucket.acquire()
    assert fake.sleeps == []
    bucket.acquire()
    assert fake.sleeps == [0.5]


def test_lock_is_released_before_sleep():
    fake = FakeTime()
    bucket = TokenBucket(1.0, 1, clock=fake.clock, sleeper=fake.sleep)
    bucket.acquire()

    def sleep(seconds):
        assert bucket._lock.acquire(blocking=False)
        bucket._lock.release()
        fake.sleep(seconds)

    bucket._sleeper = sleep
    bucket.acquire()
    assert fake.sleeps == [1.0]

What should match?

The file contains one controlled clock plus two focused tests. No test uses real sleep.

That proves the tricky part. Your bucket can wait without blocking every other thread that shares it.

Configure the limiter policy

The client needs explicit defaults for its shared bucket. Environment settings preserve deployment overrides while the Python fields provide the approved fallback values.

  • Return to src/simpro_client/config.py in the VS Code editor.
  • Add the limiter fields below max_retries by copying these lines:
    limiter_capacity: int = Field(default=8)
    limiter_refill_rate: float = Field(default=8.0)

What do these settings control?

  • The limiter_capacity field permits a burst of 8 requests.
  • The limiter_refill_rate field replenishes 8.0 tokens per second.
  • Save src/simpro_client/config.py.
  • Check the VS Code Problems panel for configuration syntax errors.

You should see no syntax errors in config.py.

Are the settings outside SimproSettings?

Match the indentation of limiter_capacity with max_retries.

Keep both limiter fields above get_settings().

Help me place the limiter settings.

  • Return to .env.example in the VS Code editor.
  • Add SIMPRO_LIMITER_CAPACITY below SIMPRO_MAX_RETRIES with the value 8.
  • Add SIMPRO_LIMITER_REFILL_RATE on the next line with the value 8.0.
  • Save .env.example.
  • Confirm the final two lines show the capacity first.

You should see SIMPRO_LIMITER_CAPACITY=8 followed by SIMPRO_LIMITER_REFILL_RATE=8.0.

Do the environment values differ?

Keep the capacity value as the integer 8.

Keep the refill-rate value as the decimal 8.0.

Help me compare the limiter environment settings.

✔️ Awesome, I've got everything!

Both configuration surfaces now use the approved limiter defaults. Save them before wiring the client.

ⓧ I'd like to double check the full code

Compare .env.example with this complete file.

# Simpro API Configuration
# Copy this file to .env and fill in your values
SIMPRO_BASE_URL=http://simpro-mock:8000/api/v1.0
SIMPRO_TOKEN_URL=http://simpro-mock:8000/oauth2/token
SIMPRO_CLIENT_ID=your-client-id-here
SIMPRO_CLIENT_SECRET=your-client-secret-here
SIMPRO_AUTH_MODE=client_credentials
# SIMPRO_API_KEY=your-api-key-here
SIMPRO_COMPANY_ID_SERVICE=1
SIMPRO_COMPANY_ID_PROJECTS=2
SIMPRO_TIMEOUT=30.0
SIMPRO_MAX_RETRIES=3
SIMPRO_LIMITER_CAPACITY=8
SIMPRO_LIMITER_REFILL_RATE=8.0

What should match?

The final two lines define the approved capacity plus the approved refill rate.

Compare src/simpro_client/config.py with this complete file.

"""Configuration management using pydantic-settings."""

from functools import lru_cache

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict


class SimproSettings(BaseSettings):
    """Simpro API connection settings loaded from environment variables."""

    model_config = SettingsConfigDict(
        env_file=".env",
        env_prefix="SIMPRO_",
        case_sensitive=False,
        extra="ignore",
    )

    base_url: str = Field(description="Base URL of the Simpro API")
    token_url: str = Field(description="OAuth2 token endpoint URL")
    client_id: str = Field(description="OAuth2 Client ID")
    client_secret: str = Field(description="OAuth2 Client Secret")
    api_key: str | None = Field(default=None, description="Optional static API key")
    auth_mode: str = Field(default="client_credentials")
    company_id_service: int = Field(default=1)
    company_id_projects: int = Field(default=2)
    timeout: float = Field(default=30.0)
    max_retries: int = Field(default=3)
    limiter_capacity: int = Field(default=8)
    limiter_refill_rate: float = Field(default=8.0)


@lru_cache
def get_settings() -> SimproSettings:
    """Get cached application settings."""
    return SimproSettings()

What should match?

The existing connection fields remain unchanged. The two limiter fields appear immediately after max_retries.

Share one limiter across the client

One bucket belongs to each SimproClient instance. Every endpoint created by that client therefore consumes tokens from one shared balance.

  • Return to src/simpro_client/client.py in the VS Code editor.
  • Find this existing logging import:
from simpro_client.logging import RequestTimer, configure_logging, get_correlation_id

Why use this import as the anchor?

The new limiter module belongs beside the existing project imports. This anchor keeps the addition in the same import group.

  • Add the limiter import immediately below the logging import so the section looks like this:
from simpro_client.logging import RequestTimer, configure_logging, get_correlation_id
from simpro_client.rate_limiter import TokenBucket

What does this import provide?

The client can now construct TokenBucket from the dependency-free limiter module.

  • Find the existing httpx.Client construction inside SimproClient.__init__().
  • Add the shared bucket directly below that construction by copying this code:
        self._limiter = TokenBucket(
            self._settings.limiter_refill_rate, self._settings.limiter_capacity
        )

Why construct one bucket here?

The constructor creates exactly one _limiter for the client. Requests from all twelve endpoint instances use that same object.

  • Find the resource request path that calls self._http.request(...).
  • Add this acquisition immediately before each resource request attempt:
            self._limiter.acquire()

What does this boundary cover?

Every resource attempt acquires one token before the synchronous request begins. Authentication token requests retain their dedicated client path.

  • Save src/simpro_client/client.py.
  • Check that self._limiter is assigned once inside the client constructor.
  • Confirm self._limiter.acquire() runs before the resource request call.

You should see one bucket construction for the client plus one acquisition at the shared request boundary.

Is TokenBucket undefined?

Check that the import uses simpro_client.rate_limiter.

Confirm rate_limiter.py is saved inside src/simpro_client.

Help me fix the TokenBucket integration.

A recording test double makes limiter acquisitions countable without introducing timing behavior. Retry tests can compare that count with the number of mocked resource attempts.

  • Return to tests/test_retries.py in the VS Code editor.
  • Add the recording limiter below the imports by copying this class:
class RecordingLimiter:
    def __init__(self):
        self.call_count = 0

    def acquire(self):
        self.call_count += 1

What does the recording limiter measure?

  • Each call to acquire() increments call_count.
  • A retry test can replace the real bucket with this object. The resulting count reveals whether each resource attempt crossed the shared boundary.
  • Save tests/test_retries.py.
  • Confirm RecordingLimiter defines one counter plus one acquisition method.

Does the recording limiter fail to load?

Place RecordingLimiter at module level below the imports.

Check that call_count starts at 0.

Help me fix the recording limiter.

Before you run the focused gate, do you expect every mocked resource attempt to cross the same limiter instance?

  • Run the focused limiter and retry tests from the existing VS Code terminal with this command:
uv run --locked pytest tests/test_rate_limiter.py tests/test_retries.py

What should you see?

Both focused test files complete without failures. The bucket permits its configured burst before refilling from controlled time.

The lock test proves sleeping occurs outside the protected section. The retry coverage still passes through the shared resource request boundary.

Are the focused tests failing?

Start with the first failing test name. A burst failure points to token arithmetic.

A lock failure points to the indentation of _sleeper. A client failure points to the limiter import or its request-boundary placement.

Help me debug the focused limiter gate.

That is the shared boundary proved. Every endpoint now draws from one client-wide bucket before its resource request attempt.

Your synchronous client now controls request pace across every resource endpoint. Next up, you will add bounded rate-limit recovery while preserving the independent authentication retry.

Add Retry Recovery and Typed Failures

The shared token bucket now meters every resource request. Your lazy iterator can no longer outrun the client-wide policy.

A 429 response still ends the request immediately. This step adds bounded recovery while preserving the synchronous HTTPX architecture.

In this step, get ready to:
  • Complete the typed exception hierarchy with transport context.
  • Replace recursive response handling with one synchronous attempt loop.
  • Prove retry timing and independent recovery budgets with offline tests.
Complete the typed failure hierarchy

Callers need stable exception types for client failures and server failures. Each exception also needs enough request context for diagnosis.

What context belongs on an error?

  • The method and URL identify the failed request.
  • The status code and response body preserve the server result.
  • The correlation ID links every attempt from one logical request.
  • Retry counts show how much recovery work happened before the failure surfaced.
  • Switch back to src/simpro_client/exceptions.py in the VS Code tab from earlier.
  • Update the exception hierarchy to match the complete reference below.
  • Save src/simpro_client/exceptions.py.

✔️ Awesome, I've got everything!

Your hierarchy now separates client failures from server failures. Rate-limit exhaustion also carries retry timing and attempt context.

ⓧ I'd like to double check the full code

"""Typed exception hierarchy for Simpro API errors."""


class SimproError(Exception):
    def __init__(self, message: str) -> None:
        self.message = message
        super().__init__(message)


class SimproAuthError(SimproError):
    pass


class SimproAPIError(SimproError):
    def __init__(
        self,
        message: str,
        status_code: int,
        response_body: str = "",
        *,
        method: str | None = None,
        url: str | None = None,
        correlation_id: str | None = None,
        retry_count: int = 0,
    ) -> None:
        self.status_code = status_code
        self.response_body = response_body
        self.method = method
        self.url = url
        self.correlation_id = correlation_id
        self.retry_count = retry_count
        super().__init__(message)


class SimproClientError(SimproAPIError):
    pass


class SimproRateLimitError(SimproClientError):
    def __init__(
        self,
        message: str = "Rate limit exceeded",
        retry_after: float | None = None,
        *,
        method: str | None = None,
        url: str | None = None,
        response_body: str = "",
        correlation_id: str | None = None,
        retry_count: int = 0,
        attempt_count: int = 1,
    ) -> None:
        self.retry_after = retry_after
        self.attempt_count = attempt_count
        super().__init__(
            message,
            429,
            response_body,
            method=method,
            url=url,
            correlation_id=correlation_id,
            retry_count=retry_count,
        )


class SimproNotFoundError(SimproClientError):
    def __init__(self, message: str = "Resource not found", **context) -> None:
        super().__init__(message, 404, **context)


class SimproServerError(SimproAPIError):
    pass


class SimproProtocolError(SimproError):
    def __init__(
        self,
        message: str,
        *,
        method: str | None = None,
        url: str | None = None,
        correlation_id: str | None = None,
    ) -> None:
        self.method = method
        self.url = url
        self.correlation_id = correlation_id
        super().__init__(message)

How does the hierarchy fit together?

  • SimproClientError represents non-rate-limit 4xx failures.
  • SimproRateLimitError fixes the status at 429. It also stores the final retry delay and total attempt count.
  • SimproServerError represents 5xx failures.
  • SimproProtocolError remains available for malformed pagination metadata.
  • Confirm the existing client behavior still passes by running this focused test:
uv run --locked pytest tests/test_client.py

What does this command check?

The locked environment runs the client tests without changing the dependency lockfile. A passing result confirms that the expanded hierarchy preserves the existing synchronous client behavior.

You should see the focused client tests finish without failures.

Seeing an exception test fail?

Compare the constructor parameters in src/simpro_client/exceptions.py with the full reference. Check that SimproNotFoundError still passes status 404 to its parent.

Check that every keyword-only context value reaches SimproAPIError.

Help me diagnose the exception hierarchy failure.

Build the synchronous recovery loop

One loop can coordinate authentication recovery and rate-limit recovery without mixing their budgets. Capturing the correlation ID before that loop keeps every attempt on the same trace.

How do the recovery budgets stay independent?

The auth budget is one Boolean that permits one token invalidation. The rate budget is a separate counter controlled by max_retries.

A 401 consumes only the auth budget. A 429 consumes only the rate-limit budget.

  • Switch back to src/simpro_client/client.py in VS Code.
  • Update src/simpro_client/client.py to match the complete attempt-loop reference below.
  • Save src/simpro_client/client.py.

✔️ Awesome, I've got everything!

Your client now acquires the shared limiter before every resource attempt. Its auth retry and rate-limit retries advance independently.

ⓧ I'd like to double check the full code

"""Base HTTP client for the Simpro REST API."""

import logging
import random
import time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from typing import Any

import httpx

from simpro_client.auth import AuthManager
from simpro_client.config import SimproSettings, get_settings
from simpro_client.endpoints import (
    AssetsEndpoint, AttachmentsEndpoint, CompaniesEndpoint, ContactsEndpoint,
    CustomersEndpoint, EmployeesEndpoint, JobNotesEndpoint, JobsEndpoint,
    ProjectsEndpoint, QuotesEndpoint, SitesEndpoint, StatusesEndpoint,
)
from simpro_client.exceptions import (
    SimproAPIError, SimproClientError, SimproNotFoundError,
    SimproRateLimitError, SimproServerError,
)
from simpro_client.logging import RequestTimer, configure_logging, get_correlation_id
from simpro_client.rate_limiter import TokenBucket

_MAX_RETRY_DELAY_SECONDS = 60.0


def _utc_now() -> datetime:
    return datetime.now(timezone.utc)


class SimproClient:
    def __init__(self, settings: SimproSettings | None = None) -> None:
        self._settings = settings or get_settings()
        self._auth = AuthManager(self._settings)
        self._logger = configure_logging()
        self._http = httpx.Client(
            base_url=self._settings.base_url,
            timeout=self._settings.timeout,
            headers={"Content-Type": "application/json", "Accept": "application/json"},
        )
        self._limiter = TokenBucket(
            self._settings.limiter_refill_rate, self._settings.limiter_capacity
        )
        self._sleep = time.sleep
        self._random = random.random
        self._now = _utc_now
        self.companies = CompaniesEndpoint(self)
        self.customers = CustomersEndpoint(self)
        self.jobs = JobsEndpoint(self)
        self.quotes = QuotesEndpoint(self)
        self.contacts = ContactsEndpoint(self)
        self.sites = SitesEndpoint(self)
        self.assets = AssetsEndpoint(self)
        self.employees = EmployeesEndpoint(self)
        self.projects = ProjectsEndpoint(self)
        self.job_notes = JobNotesEndpoint(self)
        self.attachments = AttachmentsEndpoint(self)
        self.statuses = StatusesEndpoint(self)

    def get(self, path, params=None):
        return self._request("GET", path, params=params)

    def post(self, path, json=None):
        return self._request("POST", path, json=json)

    def patch(self, path, json=None):
        return self._request("PATCH", path, json=json)

    def delete(self, path):
        return self._request("DELETE", path)

    def _get_response(self, path, params=None):
        return self._request_response("GET", path, params=params)

    def _request(self, method, path, params=None, json=None, _retry_on_401=True):
        response = self._request_response(
            method, path, params=params, json=json, _retry_on_401=_retry_on_401
        )
        return None if response.status_code == 204 else response.json()

    def _request_response(
        self, method, path, params=None, json=None, _retry_on_401=True
    ):
        correlation_id = get_correlation_id()
        auth_retry_available = _retry_on_401
        rate_retry_count = 0
        attempt_count = 0
        while True:
            token = self._auth.get_token()
            headers = {
                "Authorization": f"Bearer {token}",
                "X-Correlation-ID": correlation_id,
            }
            self._limiter.acquire()
            attempt_count += 1
            with RequestTimer() as timer:
                try:
                    response = self._http.request(
                        method=method, url=path, params=params, json=json, headers=headers
                    )
                except httpx.HTTPError as exc:
                    raise SimproAPIError(
                        f"Request failed: {exc}", 0, method=method, url=path,
                        correlation_id=correlation_id, retry_count=rate_retry_count,
                    ) from exc
            self._log_request(method, path, response.status_code, timer.duration_ms)
            if response.status_code == 401 and auth_retry_available:
                self._auth.invalidate()
                auth_retry_available = False
                continue
            if response.status_code == 429:
                retry_after = self._parse_retry_after(response.headers.get("Retry-After"))
                if rate_retry_count >= self._settings.max_retries:
                    raise SimproRateLimitError(
                        retry_after=retry_after, method=method, url=path,
                        response_body=response.text, correlation_id=correlation_id,
                        retry_count=rate_retry_count, attempt_count=attempt_count,
                    )
                delay = retry_after if retry_after is not None else self._backoff_delay(rate_retry_count)
                rate_retry_count += 1
                self._sleep(delay)
                continue
            if response.status_code == 404:
                raise SimproNotFoundError(
                    f"Not found: {method} {path}", response_body=response.text,
                    method=method, url=path, correlation_id=correlation_id,
                )
            if 400 <= response.status_code < 500:
                raise SimproClientError(
                    f"Client error: {response.status_code} on {method} {path}",
                    response.status_code, response.text, method=method, url=path,
                    correlation_id=correlation_id, retry_count=rate_retry_count,
                )
            if response.status_code >= 500:
                raise SimproServerError(
                    f"Server error: {response.status_code} on {method} {path}",
                    response.status_code, response.text, method=method, url=path,
                    correlation_id=correlation_id, retry_count=rate_retry_count,
                )
            return response

    def _parse_retry_after(self, value):
        if value is None:
            return None
        candidate = value.strip()
        if candidate.isascii() and candidate.isdigit():
            return float(int(candidate))
        try:
            retry_at = parsedate_to_datetime(candidate)
        except (TypeError, ValueError, OverflowError):
            return None
        if retry_at.tzinfo is None:
            return None
        delay = (retry_at.astimezone(timezone.utc) - self._now()).total_seconds()
        return delay if delay >= 0 else None

    def _backoff_delay(self, retry_count):
        return min(
            _MAX_RETRY_DELAY_SECONDS,
            (2**retry_count) * (0.5 + self._random()),
        )

    def _log_request(self, method, url, status_code, duration_ms):
        record = logging.LogRecord(
            "simpro_client", logging.INFO, "", 0,
            f"{method} {url} -> {status_code} ({duration_ms:.1f}ms)",
            None, None,
        )
        record.method = method
        record.url = url
        record.status_code = status_code
        record.duration_ms = round(duration_ms, 1)
        self._logger.handle(record)

    def close(self):
        self._http.close()
        self._auth.close()

    def __enter__(self):
        return self

    def __exit__(self, *args: Any):
        self.close()

What does the recovery loop do?

  • The loop obtains an auth token before each resource attempt. This lets the second attempt use a refreshed token after a 401.
  • The shared limiter is acquired immediately before each HTTP request.
  • The parser accepts digit-only delay seconds. It also accepts aware HTTP-date values.
  • Malformed retry metadata falls back to bounded jitter generated through _random.
  • Confirm the revised client remains compatible by running the focused client tests again:
uv run --locked pytest tests/test_client.py

What does this rerun prove?

The focused suite exercises the public synchronous methods after the internal control flow changes. Passing tests confirm that raw JSON returns and context-manager behavior remain intact.

You should see every focused client test pass with the new attempt loop in place.

Client tests failing after the loop change?

Check that _request() still returns JSON for normal responses. Check that status 204 still returns None.

Confirm that the limiter acquisition appears before self._http.request() inside the loop.

Help me debug the synchronous attempt loop.

Prove deterministic recovery behavior

Offline tests make timing behavior repeatable by replacing sleep, clock, and random functions. RESPX supplies each response in a controlled sequence.

  • Switch back to tests/test_client.py in VS Code.
  • Update tests/test_client.py to match the focused compatibility reference below.
  • Save tests/test_client.py.

✔️ Awesome, I've got everything!

Your focused client tests now cover successful requests and one-time auth recovery. They also inspect typed 404 and 429 context.

ⓧ I'd like to double check the full code

import respx
from httpx import Response

from simpro_client.client import SimproClient
from simpro_client.exceptions import SimproNotFoundError, SimproRateLimitError


def token(settings):
    return respx.post(settings.token_url).mock(
        return_value=Response(200, json={"access_token": "tok", "expires_in": 3600})
    )


def capture(error_type, action):
    try:
        action()
    except error_type as exc:
        return exc
    raise AssertionError(f"{error_type.__name__} was not raised")


@respx.mock
def test_successful_get_and_context_manager(mock_settings):
    token(mock_settings)
    respx.get(f"{mock_settings.base_url}/companies/1/").mock(
        return_value=Response(200, json={"ID": 1, "Name": "CVC Service"})
    )
    with SimproClient(mock_settings) as client:
        assert client.get("/companies/1/")["Name"] == "CVC Service"


@respx.mock
def test_401_refreshes_once(mock_settings):
    token_route = token(mock_settings)
    respx.get(f"{mock_settings.base_url}/companies/1/jobs/").mock(
        side_effect=[Response(401), Response(200, json=[{"ID": 1}])]
    )
    with SimproClient(mock_settings) as client:
        assert client.get("/companies/1/jobs/") == [{"ID": 1}]
    assert token_route.call_count == 2


@respx.mock
def test_typed_404_and_429(mock_settings):
    token(mock_settings)
    respx.get(f"{mock_settings.base_url}/missing").mock(
        return_value=Response(404, text="missing")
    )
    with SimproClient(mock_settings) as client:
        error = capture(SimproNotFoundError, lambda: client.get("/missing"))
    assert error.response_body == "missing"

    settings = mock_settings.model_copy(update={"max_retries": 0})
    respx.get(f"{settings.base_url}/limited").mock(
        return_value=Response(429, headers={"Retry-After": "5"})
    )
    with SimproClient(settings) as client:
        error = capture(SimproRateLimitError, lambda: client.get("/limited"))
    assert error.retry_after == 5.0

What do these focused tests prove?

  • A successful synchronous GET still returns decoded JSON.
  • A 401 invalidates the token once before the successful retry.
  • A typed not-found error retains the response body.
  • An exhausted rate-limit request exposes its parsed retry delay.
  • Run the updated focused client tests:
uv run --locked pytest tests/test_client.py

What does this focused run confirm?

The command checks compatibility and the first layer of typed failure context. It also proves that a zero-retry configuration surfaces the first 429 immediately.

You should see all tests in tests/test_client.py pass.

  • Return to tests/test_retries.py from the previous step.
  • Update tests/test_retries.py to match the deterministic retry reference below.
  • Save tests/test_retries.py.

✔️ Awesome, I've got everything!

Your retry suite now controls every response and every delay. The mixed recovery test also records limiter acquisitions and correlation headers.

ⓧ I'd like to double check the full code

from datetime import datetime, timezone

import respx
from httpx import Response

from simpro_client.client import SimproClient
from simpro_client.exceptions import SimproClientError, SimproRateLimitError, SimproServerError
from simpro_client.logging import set_correlation_id


class RecordingLimiter:
    def __init__(self):
        self.call_count = 0

    def acquire(self):
        self.call_count += 1


def token(settings):
    return respx.post(settings.token_url).mock(
        return_value=Response(200, json={"access_token": "tok", "expires_in": 3600})
    )


def capture(error_type, action):
    try:
        action()
    except error_type as exc:
        return exc
    raise AssertionError(f"{error_type.__name__} was not raised")


@respx.mock
def test_retry_after_forms_and_fallback(mock_settings):
    token(mock_settings)
    route = respx.get(f"{mock_settings.base_url}/jobs")
    delays = []
    route.mock(side_effect=[Response(429, headers={"Retry-After": "2"}), Response(200, json=[])])
    with SimproClient(mock_settings) as client:
        client._sleep = delays.append
        client.get("/jobs")
    assert delays == [2.0]

    route.mock(side_effect=[Response(429, headers={"Retry-After": "Thu, 01 Jan 2026 00:00:05 GMT"}), Response(200, json=[])])
    delays.clear()
    with SimproClient(mock_settings) as client:
        client._sleep = delays.append
        client._now = lambda: datetime(2026, 1, 1, tzinfo=timezone.utc)
        client.get("/jobs")
    assert delays == [5.0]

    route.mock(side_effect=[Response(429, headers={"Retry-After": "1.5"}), Response(200, json=[])])
    delays.clear()
    with SimproClient(mock_settings) as client:
        client._sleep = delays.append
        client._random = lambda: 0.25
        client.get("/jobs")
    assert delays == [0.75]


@respx.mock
def test_401_then_429_keeps_independent_budgets(mock_settings):
    token_route = token(mock_settings)
    route = respx.get(f"{mock_settings.base_url}/jobs").mock(
        side_effect=[Response(401), Response(429, headers={"Retry-After": "1"}), Response(200, json=[])]
    )
    set_correlation_id("stable-retry-id")
    delays = []
    with SimproClient(mock_settings) as client:
        limiter = RecordingLimiter()
        client._limiter = limiter
        client._sleep = delays.append
        client.get("/jobs")
    assert token_route.call_count == 2
    assert limiter.call_count == 3
    assert delays == [1.0]
    assert [call.request.headers["X-Correlation-ID"] for call in route.calls] == ["stable-retry-id"] * 3


@respx.mock
def test_exhaustion_and_typed_errors(mock_settings):
    settings = mock_settings.model_copy(update={"max_retries": 2})
    token(settings)
    route = respx.get(f"{settings.base_url}/limited").mock(
        side_effect=[Response(429, text="slow", headers={"Retry-After": "1"})] * 3
    )
    with SimproClient(settings) as client:
        client._sleep = lambda seconds: None
        error = capture(SimproRateLimitError, lambda: client.get("/limited"))
    assert route.call_count == 3
    assert error.retry_count == 2
    assert error.attempt_count == 3

    respx.get(f"{settings.base_url}/bad").mock(return_value=Response(400, text="bad"))
    respx.get(f"{settings.base_url}/down").mock(return_value=Response(503, text="down"))
    with SimproClient(settings) as client:
        client_error = capture(SimproClientError, lambda: client.get("/bad"))
        server_error = capture(SimproServerError, lambda: client.get("/down"))
    assert client_error.status_code == 400
    assert server_error.status_code == 503

How does the suite stay deterministic?

  • Replacing _sleep records delays without waiting in real time.
  • Replacing _now fixes the clock used for an HTTP-date calculation.
  • Replacing _random makes fallback jitter predictable.
  • RecordingLimiter proves that every HTTP attempt consumes one shared limiter acquisition.

Before you run the focused gate, do you expect the mixed recovery sequence to reuse one correlation ID across all three attempts?

  • Run the complete client and retry test gate:
uv run --locked pytest tests/test_client.py tests/test_retries.py

What does the focused gate prove?

The gate checks one-time 401 recovery and bounded 429 recovery. It also checks integer delays and HTTP-date delays.

The same run verifies malformed-header fallback and typed 4xx failures. It verifies typed 5xx failures too.

You should see both test files finish without failures. The mixed sequence proves three resource attempts and three limiter acquisitions.

Retry tests failing?

If a delay assertion fails, check that _sleep receives the parsed delay or the fallback delay. Confirm that the decimal value reaches _backoff_delay().

If the mixed recovery test fails, confirm that correlation_id is assigned before the loop. Check that auth_retry_available and rate_retry_count change independently.

Help me debug the deterministic retry suite.

That is the resilience boundary working. Your synchronous client now meters every attempt and recovers from temporary rate limits without losing request context.

Your retry policy and typed failures are ready. Next up, you will run the complete offline quality gate and prepare the review handoff.

Run the Offline Gate and Prepare Handoff

Your synchronous SimproClient now combines lazy pagination with one shared HTTP resilience boundary. Focused tests prove each new path in isolation.

Compatibility-sensitive client changes need a wider final gate. The complete offline suite checks that the new behavior still fits the existing models, endpoints, authentication, and logging.

The review handoff captures the unchanged starting commit. It also identifies every uncommitted source and test change in the Git working tree.

In this step, get ready to:
  • Run the complete offline test suite.
  • Run the complete source quality gate.
  • Prepare an uncommitted handoff from the captured review scope.
Run the complete offline suite

The focused pytest suites proved individual behaviors. The complete suite checks compatibility across every affected client boundary.

Before you run the gate, form a quick prediction about whether every offline test will pass on the first attempt.

  • Run the complete offline suite by using this command:
uv run --locked pytest tests/test_models.py tests/test_endpoints.py tests/test_pagination.py tests/test_auth.py tests/test_logging.py tests/test_client.py tests/test_rate_limiter.py tests/test_retries.py

What Does the Offline Gate Cover?

  • The model suite confirms typed validation behavior.
  • The endpoint suite confirms existing resource behavior.
  • The pagination suite confirms deferred page requests.
  • The authentication suite confirms one-time refresh compatibility.
  • The logging suite confirms correlation context behavior.
  • The client and retry suites confirm the shared resilience policy.

You should see pytest finish with no failures. The command should return control to your terminal.

That is the broad compatibility check complete. The hardened client now passes every listed offline suite.

Did the Offline Suite Fail?

Use the first failing test file to locate the affected boundary. Pagination failures point to iter_all().

Retry failures point to _request_response() or the shared limiter wiring.

Help me diagnose the offline gate failure.

Run the source quality gate

Ruff checks lint rules. Its format check confirms that the source already matches the project style.

The Git whitespace check catches conflict markers or whitespace errors before review. A clean result keeps mechanical noise out of the handoff.

Before you run these checks, predict whether they will finish without changing any files.

  • Run the complete source quality gate by using these commands:
uv run --locked ruff check src tests
uv run --locked ruff format --check src tests
git diff --check

What Does the Quality Gate Prove?

  • The lint check finds code-quality violations under src and tests.
  • The format check verifies that every inspected file already matches the formatter.
  • The whitespace check scans the pending diff for review-blocking whitespace problems.

You should see both Ruff commands finish successfully. The Git whitespace check should print nothing.

The source quality gate is clean. Your handoff now has review-ready formatting.

Did a Quality Check Fail?

Use Ruff's reported file and line to correct each lint issue. Use the formatting diff to align any file that does not match the project style.

Remove any trailing whitespace or conflict markers named by Git. Rerun the complete quality gate after the correction.

Help me fix the source quality gate.

Capture the review handoff

A useful handoff identifies the exact working-tree scope. It also preserves evidence that the implementation remains uncommitted.

Before you inspect the working tree, predict whether the current commit ID will still match the baseline from Checkpoint 1.

  • Capture the final review scope by running these commands:
git status --short
git diff --stat
git rev-parse --verify HEAD

How Do These Outputs Build the Handoff?

  • The status output identifies each uncommitted file.
  • The diff summary shows the size of the pending review.
  • The commit identifier proves whether HEAD moved after the baseline check.
  • The environment scope includes .env.example.
  • The source scope includes src/simpro_client/client.py, src/simpro_client/config.py, src/simpro_client/endpoints/base.py, src/simpro_client/exceptions.py, and src/simpro_client/rate_limiter.py.
  • The test scope includes tests/test_client.py, tests/test_pagination.py, tests/test_rate_limiter.py, and tests/test_retries.py.

The status output should list the expected uncommitted source and test files. The diff summary should describe their line changes.

The printed commit ID should match the baseline commit ID from Checkpoint 1. That unchanged value confirms that no commit was created.

Does the Review Scope Look Wrong?

An empty status means this checkout no longer contains the uncommitted implementation. A different commit ID means HEAD moved after the baseline check.

Switch back to the clean-start CLIVE-I checkout from earlier. Compare its working tree with the expected environment, source, and test files above.

Help me reconcile the Git review scope.

  • Record the changed source and test files shown in the Git status output.
  • Report the complete offline test outcome.
  • Report the lint outcome.
  • Report the format outcome.
  • Report the whitespace outcome.
  • Include the starting commit ID from Checkpoint 1.
  • Document live Simpro behavior as a remaining compatibility risk because this gate uses mocked traffic.
  • Confirm that no commit was created.

That is the hardening work proved. Your uncommitted CLIVE-I checkout is ready for review.

Secret mission

Audit the Mixed Recovery Trace

A green retry test proves the implementation works. This audit goes deeper by tracing how authentication recovery, rate-limit recovery, limiter acquisition, and correlation identity behave across one mixed request sequence.

Clean Up Your Resources

Clean Up Your Resources

The hardened CLIVE-I checkout stays entirely on your server without a running service. There are no ongoing cloud costs.

Resources you used:

  • Uncommitted implementation changes in the existing Git checkout.
  • Uncommitted offline test changes in the existing Git checkout.

Keep everything running

No cleanup is required while the implementation is awaiting review. The checkout stays ready for another offline gate.

  • Retain the uncommitted implementation in the existing CLIVE-I checkout.
  • Keep the test files beside the source changes for reviewer context.

Pause - I'll come back to this later

No process needs to be stopped because the project does not run a local service. The idle checkout uses only server disk space.

  • Leave the uncommitted files in the existing checkout.
  • Return to the same checkout when you are ready to continue the handoff.

Delete - I don't want to use this again

Discarding the project changes permanently removes the review-ready implementation. The public-main checkout itself remains in place.

  • Return to the VS Code terminal from earlier.
  • Use Git's discard workflow to restore every tracked change to HEAD.
  • Remove every untracked project file reported from the previously clean baseline.

Before you verify, do you expect the short Git status to contain any file entries?

  • Confirm the clean checkout by running this command:
git status --short

What should I see?

An empty result confirms that the tracked changes are restored. It also confirms that the project files added after the baseline are gone.

Still seeing changed files?

  • Compare each remaining path with the implementation files or offline tests before discarding it.

Get help interpreting the remaining Git entries.

Cleanup complete. Your checkout now matches the clean baseline you captured before editing.

Nice Work!

Nice Work!

You did it! CLIVE-I now streams typed pages through a synchronous Simpro client with one shared resilience policy.

You've learned how to:

  • Build a lazy generator through iter_all(). It yields validated models one at a time. It preserves filters across every reported page. Invalid pagination metadata raises SimproProtocolError.
  • Enforce client-wide rate control with one shared token bucket. Every resource request attempt acquires the same limiter. The lock is released before any wait.
  • Keep the synchronous HTTPX transport compatible. Keep the raw JSON methods intact. Use offline RESPX tests to prove one-time 401 recovery. The same suite proves bounded 429 recovery. Typed failures retain request context.
  • Secret Mission: Audit the mixed recovery state machine as 401 -> refresh -> 429 -> wait -> 200. The 401 spends the auth-refresh budget. The 429 spends one separate rate-limit retry. Each of the three resource attempts acquires the limiter once. Two token requests support the refresh. One sleep handles the rate limit. Every attempt keeps stable-retry-id in X-Correlation-ID.

Ready to quiz yourself?