fix: retry asset downloads on 5xx and recover them across runs

An asset download is two hops — mint a signed URL, then fetch the bytes — and only the first went through _make_request. The second was a bare session.request with no 5xx handling, which lost two images in the 2026-10-20 09:00 run: one to a 500 the retried first hop had already recovered from, one to a 503. Both re-downloaded cleanly an hour later (106 KB and 1.3 MB), so the bytes were there all along. A third asset timed out and aborted its whole conversation. curl_cffi's RequestException subclasses OSError, so the un-wrapped timeout slipped past resolve_media's except ProviderError into the export loop's except OSError and was reported as "File write failed for conversation 6a945155" — a conversation that had already downloaded ~20 images. Nothing was wrong with any file write. The retry loop is factored out of _make_request into _send_with_retries, which returns the raw response so a bytes caller gets the same handling as a JSON one. Hop 2 goes through it: 5xx retried, Retry-After honoured, timeouts surfacing as ProviderError for resolve_media to placeholder. The hop is now logged, which it never was. Assets get their own ladder — an immediate retry, then ~2/4/8/16s with jitter kept, since a storage incident fails many assets at once — bounded by a per-asset wall-clock budget, because the timeout case costs 40s a go. Beyond the retry: - Transient failures (5xx, timeout, network) are recorded as pending_media so a later run re-fetches just that asset; capped at 3 runs. 404/403 stay permanent — settled 2026-08-17 for 19 assets. - A circuit breaker stops downloads after 5 consecutive transient failures, recording the rest as retryable. Permanent failures never trip it; any success resets the streak. - Media failures now appear in the set ids. The run that lost two assets pushed a clean "OK". Keeps FUTURE.md's per-conversation cache reset correctly closed. The README split (FUTURE.md §2) stays otial enough to justify breaking every anchor link. 37 new tests (414 passing). Verified live: both lost assets recovered, doctor green, a read-only chatgpt dns. The cross-run retry and breaker are unit-tested only. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/cwa7qnGp Claude-Session: https://claude.ai/code/session_012gCWSknsWvhLnMFwa7qnGp
2026-10-20 10:32:20 -04:00
parent 805689c649
commit 8c3e00a520
12 changed files with 990 additions and 63 deletions
@@ -254,3 +254,97 @@ class TestReexportPreservesJoplinLink:
        pending = tmp_cache.get_joplin_pending("chatgpt")
        assert len(pending) == 1
        assert pending[0][1]["joplin_note_id"] == "note-123"
+
+
+# ---------------------------------------------------------------------------
+# Cross-run media retry.
+#
+# Before this, a transient asset failure was permanent: resolve_media kept a
+# placeholder, mark_exported ran anyway, and the conversation was never looked
+# at again. Two assets were lost that way on 2026-10-20 to a 500 and a 503 —
+# both recoverable. The manifest now remembers which refs are still missing.
+# ---------------------------------------------------------------------------
+
+
+class TestPendingMediaRetry:
+    UPDATED = "2026-10-20T09:00:00Z"
+
+    def test_pending_media_forces_a_re_export(self, tmp_cache):
+        tmp_cache.mark_exported(
+            "chatgpt", "conv-1",
+            {"updated_at": self.UPDATED, "pending_media": ["sediment://file_x"]},
+        )
+        assert tmp_cache.is_cached("chatgpt", "conv-1", self.UPDATED) is False
+
+    def test_no_pending_media_is_an_ordinary_hit(self, tmp_cache):
+        tmp_cache.mark_exported("chatgpt", "conv-1", {"updated_at": self.UPDATED})
+        assert tmp_cache.is_cached("chatgpt", "conv-1", self.UPDATED) is True
+
+    def test_a_resolved_asset_clears_the_flag(self, tmp_cache):
+        """The run that recovers the asset must stop the daily re-export."""
+        tmp_cache.mark_exported(
+            "chatgpt", "conv-1",
+            {"updated_at": self.UPDATED, "pending_media": ["sediment://file_x"]},
+        )
+        assert tmp_cache.is_cached("chatgpt", "conv-1", self.UPDATED) is False
+
+        tmp_cache.mark_exported("chatgpt", "conv-1", {"updated_at": self.UPDATED})
+        assert tmp_cache.is_cached("chatgpt", "conv-1", self.UPDATED) is True
+        entry = tmp_cache.get_all_entries("chatgpt")["conv-1"]
+        assert "pending_media" not in entry
+
+    def test_retries_are_capped_so_it_does_not_re_export_forever(self, tmp_cache):
+        """An asset that 5xxs every morning must stop costing a re-export."""
+        from src.cache import MAX_MEDIA_RETRY_RUNS
+
+        for run in range(MAX_MEDIA_RETRY_RUNS):
+            tmp_cache.mark_exported(
+                "chatgpt", "conv-1",
+                {"updated_at": self.UPDATED, "pending_media": ["sediment://file_x"]},
+            )
+            assert tmp_cache.is_cached("chatgpt", "conv-1", self.UPDATED) is False, (
+                f"run {run + 1} should still retry"
+            )
+
+        tmp_cache.mark_exported(
+            "chatgpt", "conv-1",
+            {"updated_at": self.UPDATED, "pending_media": ["sediment://file_x"]},
+        )
+        assert tmp_cache.is_cached("chatgpt", "conv-1", self.UPDATED) is True
+
+    def test_the_lost_refs_are_kept_as_a_record(self, tmp_cache):
+        from src.cache import MAX_MEDIA_RETRY_RUNS
+
+        for _ in range(MAX_MEDIA_RETRY_RUNS + 1):
+            tmp_cache.mark_exported(
+                "chatgpt", "conv-1",
+                {"updated_at": self.UPDATED, "pending_media": ["sediment://file_x"]},
+            )
+        entry = tmp_cache.get_all_entries("chatgpt")["conv-1"]
+        assert entry["pending_media"] == ["sediment://file_x"]
+
+    def test_joplin_link_survives_a_media_re_export(self, tmp_cache):
+        """Re-exporting to recover an asset must not orphan the Joplin note."""
+        tmp_cache.mark_exported("chatgpt", "conv-1", {"updated_at": self.UPDATED})
+        tmp_cache.mark_joplin_synced("chatgpt", "conv-1", "note-123")
+        tmp_cache.set_joplin_resources("chatgpt", "conv-1", {"media/a.png": "res-1"})
+
+        tmp_cache.mark_exported(
+            "chatgpt", "conv-1",
+            {"updated_at": self.UPDATED, "pending_media": ["sediment://file_x"]},
+        )
+        entry = tmp_cache.get_all_entries("chatgpt")["conv-1"]
+        assert entry["joplin_note_id"] == "note-123"
+        # Resource ids are the idempotency record: without them the re-sync
+        # would re-upload every image in the note, not just the new one.
+        assert entry["joplin_resources"] == {"media/a.png": "res-1"}
+
+    def test_pending_media_survives_a_manifest_round_trip(self, tmp_path):
+        from src.cache import Cache
+
+        c = Cache(tmp_path)
+        c.mark_exported(
+            "chatgpt", "conv-1",
+            {"updated_at": self.UPDATED, "pending_media": ["sediment://file_x"]},
+        )
+        assert Cache(tmp_path).is_cached("chatgpt", "conv-1", self.UPDATED) is False
@@ -661,3 +661,66 @@ class TestNoRealNotificationsDuringTests:
        monkeypatch.setenv("NTFY_TOPIC", "some-topic")
        assert "127.0.0.1" in os.getenv("NTFY_SERVER", "")
        assert notify_mod.send("t", "m") is False
+
+
+# ---------------------------------------------------------------------------
+# Media failures in the push notification.
+#
+# On 2026-10-20 a scheduled run lost two assets and pushed a clean "OK" with
+# no hint of it — the counts were in the log file only, which is a place you
+# have to think to go and look. Counts only, never asset ids: a topic on
+# public ntfy.sh is readable by anyone who knows its name.
+# ---------------------------------------------------------------------------
+
+
+class TestMediaFailuresInNotification:
+    def test_media_failures_appear_in_the_message(self):
+        from src.notify import format_summary
+
+        _, message, _, _ = format_summary(
+            {"chatgpt": {"exported": 24, "skipped": 826, "failed": 0, "media_failed": 2}},
+            None,
+            None,
+        )
+        assert "2 media failed" in message
+
+    def test_absent_when_there_are_none(self):
+        from src.notify import format_summary
+
+        _, message, _, _ = format_summary(
+            {"chatgpt": {"exported": 5, "skipped": 1, "failed": 0, "media_failed": 0}},
+            None,
+            None,
+        )
+        assert "media failed" not in message
+
+    def test_media_failures_alone_do_not_fail_the_run(self):
+        """The transcript was archived and a transient asset retries later."""
+        from src.notify import format_summary
+
+        title, _, tags, priority = format_summary(
+            {"chatgpt": {"exported": 1, "skipped": 0, "failed": 0, "media_failed": 3}},
+            None,
+            None,
+        )
+        assert title.startswith("AI archive OK")
+        assert priority == "default"
+        assert tags == "white_check_mark"
+
+    def test_no_asset_ids_leak_into_the_payload(self):
+        from src.notify import format_summary
+
+        _, message, _, _ = format_summary(
+            {"chatgpt": {"exported": 1, "skipped": 0, "failed": 0, "media_failed": 1}},
+            None,
+            None,
+        )
+        assert "sediment" not in message and "file_" not in message
+
+    def test_older_summaries_without_the_key_still_render(self):
+        from src.notify import format_summary
+
+        _, message, _, _ = format_summary(
+            {"codex": {"exported": 0, "skipped": 29, "failed": 0}}, None, None
+        )
+        assert "codex: 0 exported, 29 up to date" in message
@@ -100,7 +100,7 @@ class TestResolveMedia:

        n = resolve_media(conv, provider, tmp_path, "provider/project/year", "images", report)

-        assert n == 1
+        assert n.downloaded == 1
        assert report.media_downloaded == 1
        assert block["local_path"] == "media/file_img1.png"
        rendered = render_blocks_to_markdown([block])
@@ -117,7 +117,7 @@ class TestResolveMedia:
        report = LossReport()

        n = resolve_media(conv, provider, tmp_path, "provider/project/year", "images", report)
-        assert n == 0
+        assert n.downloaded == 0
        assert "local_path" not in block
        assert provider.calls == []

@@ -129,7 +129,7 @@ class TestResolveMedia:
        report = LossReport()

        n = resolve_media(conv, provider, tmp_path, "provider/project/year", "all", report)
-        assert n == 1
+        assert n.downloaded == 1
        assert block["local_path"] == "media/file_audio1.wav"
        rendered = render_blocks_to_markdown([block])
        assert "media/file_audio1.wav" in rendered and rendered.startswith("> 📎")
@@ -139,7 +139,9 @@ class TestResolveMedia:
        block = make_image_placeholder(ref="sediment://file_x", source="user_upload")
        conv = _conv_with([block])
        report = LossReport()
-        assert resolve_media(conv, provider, tmp_path, "provider/project/year", "off", report) == 0
+        assert resolve_media(
+            conv, provider, tmp_path, "provider/project/year", "off", report
+        ).downloaded == 0
        assert provider.calls == []

    def test_idempotent_uses_disk(self, tmp_path):
@@ -164,7 +166,7 @@ class TestResolveMedia:
        report = LossReport()

        n = resolve_media(conv, provider, tmp_path, "provider/project/year", "images", report)
-        assert n == 0
+        assert n.downloaded == 0
        assert "local_path" not in block
        assert report.media_failed["expired-or-missing"] == 1
        # Still renders as a placeholder, not a broken image link
@@ -192,7 +194,9 @@ class TestResolveMedia:
            pass
        block = make_image_placeholder(ref="sediment://file_x", source="user_upload")
        conv = _conv_with([block])
-        assert resolve_media(conv, NoDownload(), tmp_path, "provider/project/year", "images", LossReport()) == 0
+        assert resolve_media(
+            conv, NoDownload(), tmp_path, "provider/project/year", "images", LossReport()
+        ).downloaded == 0


 # ---------------------------------------------------------------------------
@@ -262,3 +266,156 @@ class TestUploadMediaAndRewrite:
        new_body, res_map = upload_media_and_rewrite(body, tmp_path, client, {})
        assert new_body == body
        assert res_map == {}
+
+
+# ---------------------------------------------------------------------------
+# Transient vs permanent failures, and the cross-run retry they feed.
+#
+# Measured 2026-10-20 in one scheduled run: a signed URL returned 500 on one
+# asset and 503 on another. Both are recoverable, yet the conversation was
+# marked exported, so the assets were lost for good. Permanent failures must
+# stay out of the retry set — 19 assets in this archive 403/404 because their
+# bytes were lost in a 2026-07-14 Library migration, and retrying those every
+# morning would be noise that also trips the circuit breaker.
+# ---------------------------------------------------------------------------
+
+
+class TestTransientFailureRetry:
+    def _run(self, failure, tmp_path, circuit=None):
+        from src.media import resolve_media
+
+        ref = "sediment://file_x"
+        provider = _FakeProvider(fail_refs={ref: RuntimeError(failure)})
+        conv = _conv_with([make_image_placeholder(ref, "user_upload")])
+        report = LossReport()
+        outcome = resolve_media(
+            conv, provider, tmp_path, "provider/project/year", "images", report,
+            circuit=circuit,
+        )
+        return outcome, report, conv
+
+    def test_server_error_is_retryable(self, tmp_path):
+        outcome, report, conv = self._run("Signed URL: HTTP 500 after 5 retries", tmp_path)
+        assert outcome.failed == 1
+        assert outcome.retryable_refs == ["sediment://file_x"]
+        assert report.media_failed["download-error"] == 1
+        # Placeholder survives, so the ref is still in the Markdown.
+        assert "local_path" not in conv["messages"][0]["blocks"][0]
+
+    def test_service_unavailable_is_retryable(self, tmp_path):
+        outcome, _, _ = self._run("Signed URL: HTTP 503 after 5 retries", tmp_path)
+        assert outcome.retryable_refs == ["sediment://file_x"]
+
+    def test_timeout_is_retryable(self, tmp_path):
+        outcome, _, _ = self._run(
+            "Network error after 5 retries: Operation timed out", tmp_path
+        )
+        assert outcome.retryable_refs == ["sediment://file_x"]
+
+    def test_deleted_asset_is_not_retryable(self, tmp_path):
+        outcome, report, _ = self._run(
+            "Asset no longer exists — HTTP 404 'File not found'", tmp_path
+        )
+        assert outcome.failed == 1
+        assert outcome.retryable_refs == []
+        assert report.media_failed["expired-or-missing"] == 1
+
+    def test_forbidden_asset_is_not_retryable(self, tmp_path):
+        outcome, report, _ = self._run("HTTP 403 — detail: Forbidden", tmp_path)
+        assert outcome.retryable_refs == []
+        assert report.media_failed["forbidden"] == 1
+
+    def test_success_reports_nothing_retryable(self, tmp_path):
+        from src.media import resolve_media
+
+        ref = "sediment://file_x"
+        provider = _FakeProvider(assets={ref: (b"png", "image/png", None)})
+        conv = _conv_with([make_image_placeholder(ref, "user_upload")])
+        outcome = resolve_media(
+            conv, provider, tmp_path, "provider/project/year", "images", LossReport()
+        )
+        assert (outcome.downloaded, outcome.failed, outcome.retryable_refs) == (1, 0, [])
+
+
+class TestMediaCircuit:
+    def test_trips_after_consecutive_transient_failures(self):
+        from src.media import MediaCircuit
+
+        c = MediaCircuit(threshold=3)
+        for _ in range(2):
+            c.record_failure("download-error")
+        assert not c.tripped
+        c.record_failure("download-error")
+        assert c.tripped
+
+    def test_success_resets_the_streak(self):
+        from src.media import MediaCircuit
+
+        c = MediaCircuit(threshold=3)
+        c.record_failure("download-error")
+        c.record_failure("download-error")
+        c.record_success()
+        c.record_failure("download-error")
+        assert not c.tripped, "scattered failures must not trip the breaker"
+
+    def test_permanent_failures_never_trip_it(self):
+        """A deleted upload says nothing about the provider's health."""
+        from src.media import MediaCircuit
+
+        c = MediaCircuit(threshold=2)
+        for _ in range(10):
+            c.record_failure("expired-or-missing")
+            c.record_failure("forbidden")
+        assert not c.tripped
+
+    def test_open_circuit_skips_downloads_but_keeps_refs(self, tmp_path):
+        from src.media import CIRCUIT_OPEN_BUCKET, MediaCircuit, resolve_media
+
+        ref = "sediment://file_y"
+        provider = _FakeProvider(assets={ref: (b"png", "image/png", None)})
+        conv = _conv_with([make_image_placeholder(ref, "user_upload")])
+        circuit = MediaCircuit(threshold=1)
+        circuit.record_failure("download-error")
+        assert circuit.tripped
+
+        report = LossReport()
+        outcome = resolve_media(
+            conv, provider, tmp_path, "provider/project/year", "images", report,
+            circuit=circuit,
+        )
+        assert provider.calls == [], "must not touch the network while open"
+        assert outcome.downloaded == 0
+        assert outcome.retryable_refs == [ref], "skipped assets must not be abandoned"
+        assert report.media_failed[CIRCUIT_OPEN_BUCKET] == 1
+        assert circuit.skipped == 1
+
+    def test_already_downloaded_asset_served_while_open(self, tmp_path):
+        """An open circuit must not stop on-disk media being linked."""
+        from src.media import MediaCircuit, resolve_media
+
+        ref = "sediment://file_z"
+        conv = _conv_with([make_image_placeholder(ref, "user_upload")])
+        # Derive the directory the way the exporter does rather than
+        # hardcoding it — the layout is "<provider>/<project>.<year>/".
+        from src.utils import build_export_path, generate_filename
+
+        media_dir = build_export_path(
+            tmp_path,
+            "chatgpt",
+            None,
+            "2026-05-20T00:00:00+00:00",
+            generate_filename("Has Media", "conv-1", "2026-05-20T00:00:00+00:00"),
+            "provider/project/year",
+        ).parent / "media"
+        media_dir.mkdir(parents=True)
+        (media_dir / "file_z.png").write_bytes(b"already here")
+
+        circuit = MediaCircuit(threshold=1)
+        circuit.record_failure("download-error")
+        provider = _FakeProvider()
+        resolve_media(
+            conv, provider, tmp_path, "provider/project/year", "images", LossReport(),
+            circuit=circuit,
+        )
+        assert provider.calls == []
+        assert conv["messages"][0]["blocks"][0]["local_path"] == "media/file_z.png"
@@ -1399,6 +1399,8 @@ class TestDeletedAssetReporting:
        """The probe costs a request — it must not fire on a good download."""
        class _Bytes:
            status_code = 200
+            ok = True
+            reason = "OK"
            content = b"data"
            headers = {"content-type": "image/png"}

@@ -1506,3 +1508,236 @@ class TestProjectAttribution:
        with caplog.at_level(logging.INFO):
            p.normalize_conversation(self._raw(gizmo_id="g-p-known"))
        assert not [r for r in caplog.records if "not in CHATGPT_PROJECT_IDS" in r.message]
+
+
+# ---------------------------------------------------------------------------
+# Asset download retries. Measured 2026-10-20 in one scheduled run:
+#
+#   09:04:05  .../files/<id>/download → 500   Retry 1/3 in 2.2s
+#   09:04:17  → 500                           Retry 2/3 in 4.3s
+#   09:04:32  → 500                           Retry 3/3 in 9.1s
+#   09:04:42  → 200                            ← metadata hop recovered
+#   09:04:47  WARNING Signed URL returned HTTP 500   ← second hop, no retry
+#
+# The metadata hop went through _make_request and won; the signed-URL hop was
+# a bare request and lost on its first try. A second asset lost the same way
+# to a 503, and a third timed out — and because curl_cffi's RequestException
+# subclasses OSError, that timeout escaped resolve_media's `except
+# ProviderError` and was reported by the export loop as "File write failed",
+# aborting a conversation that had already downloaded ~20 images.
+# ---------------------------------------------------------------------------
+
+
+class TestAssetDownloadRetries:
+    class _Resp:
+        def __init__(self, status=200, payload=None, content=b"", headers=None):
+            self.status_code = status
+            self.ok = 200 <= status < 400
+            self.reason = ""
+            self.content = content
+            self.text = ""
+            self.headers = headers or {}
+            self._payload = payload
+
+        def json(self):
+            if self._payload is None:
+                raise ValueError("not json")
+            return self._payload
+
+    class _ScriptedSession:
+        """Returns scripted responses in order; an Exception entry is raised."""
+
+        def __init__(self, script):
+            self.script = list(script)
+            self.requests = []
+            self.headers = {}
+
+        def request(self, method, url, **kwargs):
+            self.requests.append((method, url, kwargs))
+            item = self.script.pop(0)
+            if isinstance(item, BaseException):
+                raise item
+            return item
+
+    def _provider(self, script):
+        from src.providers.chatgpt import ChatGPTProvider
+
+        p = ChatGPTProvider.__new__(ChatGPTProvider)
+        p._request_delay = 0
+        p._last_request_at = None
+        p._session = self._ScriptedSession(script)
+        return p
+
+    def _meta(self):
+        """Hop 1: metadata response carrying the signed download_url."""
+        return self._Resp(payload={"download_url": "https://files.example/signed"})
+
+    def test_signed_url_5xx_is_retried_and_recovers(self, monkeypatch):
+        """The exact 2026-10-20 failure: hop 2 500s once, then succeeds."""
+        monkeypatch.setattr("src.providers.base.time.sleep", lambda s: None)
+        p = self._provider([
+            self._meta(),
+            self._Resp(500),
+            self._Resp(200, content=b"PNGDATA", headers={"content-type": "image/png"}),
+        ])
+        content, mime, _ = p.download_asset("sediment://file_x")
+        assert content == b"PNGDATA"
+        assert mime == "image/png"
+        assert len(p._session.requests) == 3, "hop 2 must be retried, not abandoned"
+
+    def test_signed_url_503_is_retried(self, monkeypatch):
+        monkeypatch.setattr("src.providers.base.time.sleep", lambda s: None)
+        p = self._provider([
+            self._meta(),
+            self._Resp(503),
+            self._Resp(200, content=b"OK", headers={"content-type": "image/png"}),
+        ])
+        assert p.download_asset("sediment://file_x")[0] == b"OK"
+
+    def test_signed_url_timeout_raises_provider_error_not_oserror(self, monkeypatch):
+        """The conversation-killer: a hop-2 timeout must be a ProviderError.
+
+        curl_cffi's RequestException is an OSError subclass, so an un-wrapped
+        timeout here bypassed resolve_media's handler and was caught by the
+        export loop's `except OSError` as "File write failed".
+        """
+        from curl_cffi.requests.exceptions import Timeout
+        from src.providers.base import ProviderError
+
+        monkeypatch.setattr("src.providers.base.time.sleep", lambda s: None)
+        assert issubclass(Timeout, OSError), "premise of this test"
+
+        p = self._provider([self._meta()] + [Timeout("timed out")] * 8)
+        with pytest.raises(ProviderError) as exc:
+            p.download_asset("sediment://file_x")
+        assert "Signed URL" in str(exc.value.original), "must name the failing hop"
+
+    def test_which_hop_failed_stays_in_the_message(self, monkeypatch):
+        """Hop 1 vs hop 2 fail for different reasons; the message must say which."""
+        from src.providers.base import ProviderError
+
+        monkeypatch.setattr("src.providers.base.time.sleep", lambda s: None)
+        p = self._provider([self._meta()] + [self._Resp(500)] * 8)
+        with pytest.raises(ProviderError) as exc:
+            p.download_asset("sediment://file_x")
+        detail = str(exc.value.original)
+        assert detail.startswith("Signed URL:")
+        assert "500" in detail
+
+    def test_hop_two_is_paced_once_per_attempt_not_twice(self, monkeypatch):
+        """Pacing twice per asset would add REQUEST_DELAY to every image."""
+        monkeypatch.setattr("src.providers.base.time.sleep", lambda s: None)
+        p = self._provider([
+            self._meta(),
+            self._Resp(200, content=b"X", headers={"content-type": "image/png"}),
+        ])
+        calls = []
+        real_pace = p._pace
+        monkeypatch.setattr(p, "_pace", lambda: calls.append(1) or real_pace())
+        p.download_asset("sediment://file_x")
+        assert len(calls) == len(p._session.requests) == 2
+
+
+class TestRetryLadder:
+    def test_immediate_first_retry_costs_nothing(self):
+        from src.providers.base import BaseProvider
+
+        assert BaseProvider._backoff_wait(None, 1, immediate_first=True) == 0.0
+
+    def test_ladder_shifts_down_so_it_reads_0_2_4_8_16(self):
+        from src.providers.base import BaseProvider
+
+        for attempt, lo in ((2, 2), (3, 4), (4, 8), (5, 16)):
+            wait = BaseProvider._backoff_wait(None, attempt, immediate_first=True)
+            assert lo <= wait <= lo * 1.3, f"attempt {attempt} → {wait}"
+
+    def test_jitter_is_never_dropped(self):
+        """A storage incident fails many assets; un-jittered retries sync up."""
+        from src.providers.base import BaseProvider
+
+        waits = {
+            BaseProvider._backoff_wait(None, 3, immediate_first=True)
+            for _ in range(25)
+        }
+        assert len(waits) > 1
+
+    def test_conversation_api_ladder_is_unchanged(self):
+        from src.providers.base import BaseProvider
+
+        assert 2 <= BaseProvider._backoff_wait(None, 1) <= 2.6
+        assert 4 <= BaseProvider._backoff_wait(None, 2) <= 5.2
+
+    def test_retry_after_wins_over_the_immediate_retry(self):
+        """A server naming a time knows better than we do; 503s often do."""
+        from src.providers.base import BaseProvider
+
+        class _R:
+            headers = {"Retry-After": "7"}
+
+        assert BaseProvider._backoff_wait(_R(), 1, immediate_first=True) == 7.0
+
+
+class TestMediaTimeBudget:
+    def test_budget_refuses_a_wait_it_cannot_afford(self, monkeypatch):
+        """Bounds the timeout case, which costs 40s a go and has no cheap end."""
+        from src.providers.base import ProviderError
+
+        monkeypatch.setattr("src.providers.base.time.sleep", lambda s: None)
+        p = TestAssetDownloadRetries()._provider([
+            TestAssetDownloadRetries._Resp(500) for _ in range(9)
+        ])
+        # First backoff is 2-2.6s, which does not fit in a 1s budget.
+        with pytest.raises(ProviderError):
+            p._send_with_retries(
+                "GET", "https://files.example/signed",
+                max_retries=5, time_budget=1.0,
+            )
+        assert len(p._session.requests) == 1
+
+    def test_budget_still_allows_the_free_immediate_retry(self, monkeypatch):
+        """The 0s retry costs nothing, so a small budget can still fund it."""
+        from src.providers.base import ProviderError
+
+        monkeypatch.setattr("src.providers.base.time.sleep", lambda s: None)
+        p = TestAssetDownloadRetries()._provider([
+            TestAssetDownloadRetries._Resp(500) for _ in range(9)
+        ])
+        with pytest.raises(ProviderError):
+            p._send_with_retries(
+                "GET", "https://files.example/signed",
+                max_retries=5, immediate_first_retry=True, time_budget=1.0,
+            )
+        # Attempt 1, free retry, then the 2s backoff exceeds what is left.
+        assert len(p._session.requests) == 2
+
+    def test_read_timeout_is_clamped_to_the_remaining_budget(self, monkeypatch):
+        """So a final attempt cannot overrun the budget by a whole timeout."""
+        from src.providers.base import ProviderError
+
+        monkeypatch.setattr("src.providers.base.time.sleep", lambda s: None)
+        p = TestAssetDownloadRetries()._provider([
+            TestAssetDownloadRetries._Resp(500) for _ in range(9)
+        ])
+        with pytest.raises(ProviderError):
+            p._send_with_retries(
+                "GET", "https://files.example/signed",
+                max_retries=5, time_budget=3.0, timeout=(10, 20),
+            )
+        _, _, kwargs = p._session.requests[0]
+        connect_to, read_to = kwargs["timeout"]
+        assert connect_to == 10
+        assert read_to <= 3.0, "read timeout must not exceed the budget"
+
+    def test_without_a_budget_the_full_ladder_runs(self, monkeypatch):
+        from src.providers.base import ProviderError
+
+        monkeypatch.setattr("src.providers.base.time.sleep", lambda s: None)
+        p = TestAssetDownloadRetries()._provider([
+            TestAssetDownloadRetries._Resp(500) for _ in range(9)
+        ])
+        with pytest.raises(ProviderError):
+            p._send_with_retries(
+                "GET", "https://files.example/signed",
+                max_retries=5, immediate_first_retry=True,
+            )
+        assert len(p._session.requests) == 6, "5 retries = 6 attempts"