feat(FS-1988): add min_attempts and absolute_max_elapsed_time_ms to BackoffStrategy

Closes a short-circuit in retry_with_backoff{,_async} where a single slow
first attempt can exhaust max_elapsed_time before any retry fires. With
the partitioner config (max_elapsed_time=5min, CLIENT_TIMEOUT_MS=30min),
any chunk whose first attempt exceeds 5 minutes blew the retry budget
on attempt 1 -- zero retries on subsequent transient errors. Documented
in two customer-visible regressions in the platform partition path.

New fields on BackoffStrategy:

* min_attempts (default 0) -- minimum number of retry attempts that must
  fire before max_elapsed_time is honored. Counts retries (not the
  initial attempt). Default 0 preserves existing behavior.
* absolute_max_elapsed_time_ms (default None) -- cap on when a new retry
  can START. Does NOT interrupt an in-flight func() call. Worst-case
  wall-clock under this cap is absolute_max_elapsed_time_ms +
  per_attempt_timeout. Default None preserves existing behavior.

Loop changes in both retry_with_backoff and retry_with_backoff_async:

* Post-attempt cap check honors min_attempts as a floor on the soft cap
  and treats the hard cap as unconditional.
* Pre-sleep hard-cap check refuses to sleep into a doomed retry whose
  projected start would exceed the hard cap.
* Post-sleep verification (belt-and-suspenders against late wakeups and
  rounding drift in the projection).
* Helper extraction (_cap_hit_after_attempt, _raise_or_return_after_cap)
  dedupes logic between sync and async paths.

Validation rejects min_attempts < 0, absolute_max_elapsed_time_ms <= 0,
and absolute_max_elapsed_time_ms below max_elapsed_time.

.genignore: ignore src/unstructured_client/utils/retries.py to preserve
these fields across Speakeasy regens. Documented merge procedure for
future template updates. Pushing these fields upstream to Speakeasy
templates is filed as a follow-up.

Tests:

* T1-T14 in unit/test_retries.py with a fake-clock harness that
  monkeypatches time.time / time.sleep / asyncio.sleep / random.uniform.
  Covers the v1 reproducer (slow first attempt + min_attempts floor),
  floor-is-not-a-ceiling semantics, hard cap overrides floor, sleep
  truncation, TemporaryError early-return through both caps, and
  PermanentError short-circuit immunity.
* Validation tests for the new __init__ guards.
* New integration test
  test_split_pdf_cache_tmp_data_chunk_request_stream_is_replay_safe
  pins the body-replay invariant: chunks built from open file objects
  (cache_tmp_data=True) must produce replay-safe httpx requests so SDK
  retries actually deliver the original multipart payload. Iterates
  request.stream twice directly to bypass request.read() caching.

Author: aballman

.genignore

@@ -19,3 +19,16 @@ src/unstructured_client/users.py
 #  - Adjust the custom url snippets in the file
 #  - Bring back the ignore line and commit
 src/unstructured_client/general.py
+
+# Ignore retries.py so we can carry the `min_attempts` and
+# `absolute_max_elapsed_time_ms` fields on BackoffStrategy plus the
+# soft/hard cap logic in retry_with_backoff{,_async}. Filed in FS-1988.
+# Speakeasy regen would otherwise wipe these fields. To pick up upstream
+# fixes to the generated retry runtime:
+#  - Comment out this ignore line
+#  - Generate locally
+#  - Manually re-merge the FS-1988 additions
+#  - Re-add the ignore line and commit
+# Long-term: push these fields upstream to Speakeasy templates so this
+# ignore can be removed.
+src/unstructured_client/utils/retries.py

CHANGELOG.md

@@ -3,6 +3,9 @@
 ### Breaking changes
 * Removed deprecated connector config models from the SDK (e.g. `S3SourceConnectorConfig`, `AzureDestinationConnectorConfig`). Pass connector configs as plain dicts with arbitrary fields. The SDK is no longer coupled to backend connector schemas — new fields work without an SDK upgrade.
 
+### Features
+* Add `min_attempts` and `absolute_max_elapsed_time_ms` fields to `BackoffStrategy`. `min_attempts` is the minimum number of retry attempts that must fire before `max_elapsed_time` is honored; defaults to `0` (preserves existing behavior). `absolute_max_elapsed_time_ms` caps when a new retry can start (does not interrupt in-flight requests); defaults to `None`. Together these close a short-circuit where a single slow first attempt could exhaust the retry budget before any retry fired. See FS-1988.
+
 ## 0.43.4
 
 ### Enhancements

_test_unstructured_client/integration/test_decorators.py

@@ -813,3 +813,79 @@ async def mock_send(_, request: httpx.Request, **kwargs):
     assert number_of_transport_failures == 0
     assert mock_endpoint_called
     assert res.status_code == 200
+
+
+def test_split_pdf_cache_tmp_data_chunk_request_stream_is_replay_safe(tmp_path):
+    """Regression test for FS-1988: when split_pdf_cache_tmp_data=True
+    is set, chunks are loaded as opened file objects (not BytesIO) and
+    handed to `create_pdf_chunk_request`. The resulting `httpx.Request`
+    must be replay-safe — i.e. its body stream can be iterated more
+    than once, returning identical bytes each time.
+
+    Body replay is what makes SDK-level retries on transient errors
+    (ReadTimeout, ConnectError, etc.) actually deliver the original
+    multipart payload to the server. A future Speakeasy template
+    change or refactor of `serialize_request_body` that produced a
+    single-consumption stream (e.g. an `Iterable[bytes]` over an open
+    file handle) would silently break retries: the server would see
+    an empty body on the second attempt.
+
+    The invariant is pinned by iterating `request.stream` twice
+    directly — NOT via `request.read()`, which caches into
+    `Request._content` and would paper over a non-replayable stream
+    after the first call. This is the actual transport-level path
+    httpcore uses when `AsyncClient.send` retries a request.
+    """
+    from unstructured_client._hooks.custom.request_utils import (
+        create_pdf_chunk_request,
+    )
+
+    # Drive `create_pdf_chunk_request` directly with a file-object
+    # chunk — the cache_tmp_data=True branch in `_get_pdf_chunk_files`.
+    chunk_path = tmp_path / "chunk.pdf"
+    src_bytes = Path("_sample_docs/layout-parser-paper.pdf").read_bytes()
+    chunk_path.write_bytes(src_bytes)
+
+    pdf_chunk_file = open(chunk_path, "rb")  # noqa: SIM115 -- closed manually
+    try:
+        form_data = {
+            "files": (chunk_path.name, src_bytes, "application/pdf"),
+            "strategy": "fast",
+        }
+        original_request = httpx.Request(
+            method="POST",
+            url="http://localhost:8000/general/v0/general",
+            headers={
+                "Content-Type": "multipart/form-data; boundary=test",
+                "User-Agent": "test",
+            },
+            content=b"",
+        )
+
+        chunk_request = create_pdf_chunk_request(
+            form_data=form_data,
+            pdf_chunk=(pdf_chunk_file, 1),
+            original_request=original_request,
+            filename=chunk_path.name,
+        )
+
+        # Iterate the body stream twice. If the underlying stream is
+        # not replayable (e.g. an open file handle that's exhausted
+        # after the first pass), the second iteration yields empty
+        # or partial bytes and the assert below fails.
+        first_pass = b"".join(chunk_request.stream)
+        second_pass = b"".join(chunk_request.stream)
+
+        assert len(first_pass) > 1000, (
+            f"First iteration produced too-small body ({len(first_pass)} "
+            "bytes); a real PDF chunk multipart envelope should be at "
+            "least kilobytes."
+        )
+        assert first_pass == second_pass, (
+            "Body stream not replay-safe: second iteration of "
+            "chunk_request.stream returned different bytes than the "
+            "first. SDK retries on transient errors would silently "
+            "send a truncated or empty body to the server."
+        )
+    finally:
+        pdf_chunk_file.close()