test(retry): cover the Retry-After wiring, and scope the transport claim

Three review points from #75.

Nothing exercised the path that carries `Retry-After` off a classified
error and into the sleep. `_retry_delay` was tested directly, and the
retry-loop tests stub sleep with a lambda that discards its argument, so
a regression passing `None` there — or swapping the two positional args
— would have left every test green. Refuse a load the way the API
refuses a contended table, record what sleep actually received, and
assert the waits floor on the header. That also covers classify →
transient → retry end to end, which was only covered per-piece, and the
no-header case, so the floor is visibly the header's contribution rather
than a hard-coded one.

Also pin that a `CONFLICT` surfaces on the first attempt, which is the
half of the 409 split that had no loop-level test.

The README and `test_retry_policy`'s module docstring both stated "a
load is not idempotent" without qualification. That remains true of the
transport, which sees a method and a status and cannot know what it
would be replaying — but unscoped it reads as repo-wide and contradicts
the call-layer retry. Both now say which layer they mean.

Author: anoop-narang

CHANGELOG.md

@@ -43,6 +43,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
   This lengthens a 20-attempt budget from 285s to roughly 316-405s.
 
+- docs: scope the "a load is not idempotent" claim in the README and in
+  `test_retry_policy` to the transport layer, which is where it is still true
+  and where those two were always talking about. Left unscoped they read as
+  repo-wide and contradict the call-layer retry above.
+
 ### Added
 
 - `HotdataError` carries `status_code`, `code` and `retry_after_seconds`. The

README.md

@@ -10,7 +10,7 @@ Runtime boundary and guarantees are defined in `CONTRACT.md`.
 
 - **Environment-driven client setup** — create clients from `HOTDATA_API_KEY`, optional `HOTDATA_API_URL`, and `HOTDATA_WORKSPACE`.
 - **Workspace resolution** — choose an explicit workspace from env, otherwise discover workspaces and select the active workspace or first available workspace.
-- **HTTP resilience** — retry SQL execution on stale pooled sockets. Transport-level retries are the SDK's own default, which this package leaves in place so a non-idempotent request is never replayed on a response status.
+- **HTTP resilience** — retry SQL execution on stale pooled sockets. Transport-level retries are the SDK's own default, which this package leaves in place so a request is never blindly replayed on a response status. That is a claim about the transport, which cannot know what it would be replaying. `ManagedDatabaseClient` retries at the call layer, which can: a managed load is safe to re-send because it carries the same `upload_id` and the API replays its receipt for that id rather than applying the load twice.
 - **SQL execution helper** — run SQL through `POST /v1/query`, poll async query runs when needed, and return a `QueryResult`.
 - **Result utilities** — convert query results to records, pandas DataFrames, or metadata dictionaries for adapter display layers.
 - **History helpers** — list recent results and query run history with normalized dataclasses.

tests/test_managed_client.py

@@ -8,8 +8,10 @@
 import pyarrow as pa
 import pytest
 from hotdata.models.query_response import QueryResponse
+from hotdata.rest import ApiException
 
 import hotdata_framework.managed_client as mc
+from hotdata_framework.errors import HotdataTerminalError
 
 
 def _query_response(result_id: str) -> QueryResponse:
@@ -158,9 +160,7 @@ def get_result_arrow(self, result_id: str, *, x_database_id: str) -> pa.Table:
     assert arrow_scopes == ["db1"]
 
 
-def _load_recording_runtime(
-    calls: list[str], uploads: list[str] | None = None
-) -> SimpleNamespace:
+def _load_recording_runtime(calls: list[str], uploads: list[str] | None = None) -> SimpleNamespace:
     """A runtime whose ``load_managed_table`` records each mode and always fails
     with a transient error, so retry behaviour is observable via ``calls``.
 
@@ -186,6 +186,32 @@ def load_managed_table(
     return runtime
 
 
+def _lock_refusing_runtime(retry_after: str | None) -> SimpleNamespace:
+    """A runtime whose loads are refused the way the API refuses a contended
+    table: ``409 RESOURCE_LOCKED``, optionally carrying ``Retry-After``."""
+
+    def load_managed_table(
+        database: str,
+        table: str,
+        *,
+        schema: str,
+        upload_id: str,
+        mode: str,
+        key: list[str] | None = None,
+    ) -> SimpleNamespace:
+        error = ApiException(
+            status=409,
+            reason="Conflict",
+            body='{"error":{"code":"RESOURCE_LOCKED","message":"retry shortly"}}',
+        )
+        error.headers = {"Retry-After": retry_after} if retry_after else {}
+        raise error
+
+    runtime = _fake_runtime()
+    runtime.load_managed_table = load_managed_table
+    return runtime
+
+
 def _managed_client(max_retries: int) -> Any:
     return mc.ManagedDatabaseClient(
         api_key="k",
@@ -247,6 +273,78 @@ def test_idempotent_load_retries_on_transient(monkeypatch: pytest.MonkeyPatch) -
     assert calls == ["replace", "replace", "replace"]  # retried up to max_retries
 
 
+def test_a_lock_refusal_is_retried_and_waits_the_header_out(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """End to end for the contended-table case: a `409 RESOURCE_LOCKED` is
+    classified transient, retried, and each wait honours the `Retry-After` the
+    refusal carried.
+
+    Worth having as one test rather than three: `_retry_delay` is exercised
+    directly elsewhere, but nothing else covers the wiring that carries the
+    header off the error and into the sleep. The other retry tests stub sleep
+    with a lambda that discards its argument, so a regression passing `None`
+    here — or swapping the two positional arguments — would leave them green."""
+    slept: list[float] = []
+    monkeypatch.setattr(mc.time, "sleep", lambda seconds: slept.append(seconds))
+    monkeypatch.setattr(mc.random, "random", lambda: 0.0)
+    client = _managed_client(max_retries=4)
+    client._retry_backoff_seconds = 1.5
+    client._runtime = _lock_refusing_runtime(retry_after="5")
+
+    with pytest.raises(mc.HotdataTransientError):
+        client.load_managed_table("db", "orders", schema="public", upload_id="u1", mode="append")
+
+    # Four attempts, so three waits — each floored on the header rather than
+    # taking the ramp's 1.5s / 3.0s / 4.5s.
+    assert slept == [5.0, 5.0, 5.0]
+
+
+def test_a_lock_refusal_without_a_header_falls_back_to_the_ramp(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The floor is the header's contribution, not a hard-coded one: with no
+    header the ramp decides, which keeps the refusal path working against a
+    server that does not state a wait."""
+    slept: list[float] = []
+    monkeypatch.setattr(mc.time, "sleep", lambda seconds: slept.append(seconds))
+    monkeypatch.setattr(mc.random, "random", lambda: 0.0)
+    client = _managed_client(max_retries=4)
+    client._retry_backoff_seconds = 1.5
+    client._runtime = _lock_refusing_runtime(retry_after=None)
+
+    with pytest.raises(mc.HotdataTransientError):
+        client.load_managed_table("db", "orders", schema="public", upload_id="u1", mode="append")
+
+    assert slept == [1.5, 3.0, 4.5]
+
+
+def test_a_permanent_conflict_is_not_retried(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A `CONFLICT` cannot succeed as posted, so it surfaces on the first
+    attempt instead of spending the budget to reach the same 409."""
+    monkeypatch.setattr(mc.time, "sleep", lambda _seconds: None)
+    attempts: list[int] = []
+
+    def load_managed_table(*_args: object, **_kwargs: object) -> SimpleNamespace:
+        attempts.append(1)
+        error = ApiException(
+            status=409,
+            reason="Conflict",
+            body='{"error":{"code":"CONFLICT","message":"upload already consumed"}}',
+        )
+        error.headers = {}
+        raise error
+
+    client = _managed_client(max_retries=8)
+    client._runtime = _fake_runtime()
+    client._runtime.load_managed_table = load_managed_table
+
+    with pytest.raises(HotdataTerminalError):
+        client.load_managed_table("db", "orders", schema="public", upload_id="u1", mode="append")
+
+    assert len(attempts) == 1
+
+
 def test_retry_delay_floors_on_the_servers_retry_after(monkeypatch: pytest.MonkeyPatch) -> None:
     """``Retry-After`` states how long the refused condition lasts; the ramp only
     knows how many attempts are left. Taking the larger of the two respects both,

tests/test_retry_policy.py

@@ -1,8 +1,14 @@
-"""A POST must never be replayed because of a response status.
+"""A POST must never be replayed *by the transport* because of a response status.
 
-A load is not idempotent: re-sending one that the server is still working on
-collides with the write lock the first attempt holds, and the duplicate is
-refused. The generated SDK already draws this line — ``hotdata._retry`` retries
+Re-sending a load the server is still working on collides with the write lock
+the first attempt holds, and the duplicate is refused — so a blind transport
+replay buys nothing and spends an attempt. This is a claim about the transport,
+which sees a method and a status and cannot know what it would be re-sending.
+It is not a claim that loads must never be retried: ``ManagedDatabaseClient``
+retries them at the call layer, where the same ``upload_id`` goes back out and
+the API replays its receipt for that id instead of applying the load twice.
+
+The generated SDK already draws this line — ``hotdata._retry`` retries
 a *pre-response* connection reset on any method (the stale pooled socket case,
 where the server did no work) while leaving read timeouts and status retries
 idempotent-only. This wrapper used to pass its own ``retries=`` into