tools: analyze whether media loss is age-based

"Do I have to export within N days?" is answerable from the exports
already on disk — the renderer records every image's outcome inline
(![source](media/…) when saved, a placeholder when not) and the
conversation date is in the filename. Group by month and source and the
hypotheses separate: a clean old/new cutoff means expiry, user_upload
dying at an age model_generated survives means the source matters, and
losses scattered through months that otherwise downloaded fine means
neither.

Offline, no token, no API calls.
This commit is contained in:
JesseMarkowitz
2026-08-17 08:45:21 -04:00
parent f40b25001a
commit 04191eed8c
6 changed files with 326 additions and 215 deletions
+117
View File
@@ -1182,3 +1182,120 @@ class TestErrorBodyDiagnostics:
described = _describe_error_body(resp)
assert "sk-abc" not in described
assert "[REDACTED]" in described
# ---------------------------------------------------------------------------
# Deleted assets: ChatGPT's download endpoint answers a missing upload with
# 403 Forbidden, not 404. Measured 2026-08-17 over 18 such assets — every one
# returned 404 "File not found" on /files/{id}, while assets that downloaded
# fine returned 200 on both in the same session.
# ---------------------------------------------------------------------------
class TestDeletedAssetReporting:
class _Resp:
def __init__(self, status, payload=None):
self.status_code = status
self.ok = 200 <= status < 400
self.reason = ""
self.headers: dict = {}
self.text = ""
self._payload = payload or {}
def json(self):
return self._payload
def _provider(self, responses):
"""responses: dict of url-substring → _Resp, consumed by substring match."""
from src.providers.chatgpt import ChatGPTProvider
p = ChatGPTProvider.__new__(ChatGPTProvider)
p._request_delay = 0
p._last_request_at = None
calls: list[str] = []
def request(method, url, **kwargs):
calls.append(url)
for fragment, resp in responses.items():
if url.endswith(fragment):
return resp
raise AssertionError(f"unexpected URL: {url}")
p._session = type("S", (), {"request": staticmethod(request)})()
p._calls = calls
return p
def test_403_confirmed_missing_is_reported_as_gone(self):
from src.providers.base import ProviderError
p = self._provider({
"/download": self._Resp(403, {"detail": "Forbidden"}),
"/files/file_x": self._Resp(404, {"detail": "File not found"}),
})
with pytest.raises(ProviderError) as exc:
p.download_asset("sediment://file_x")
message = str(exc.value.original)
assert "no longer exists" in message
assert "not recoverable" in message
assert any(u.endswith("/files/file_x") for u in p._calls), "probe not sent"
def test_403_on_an_asset_that_still_exists_stays_a_403(self):
"""Don't call a live asset deleted — that would hide a real problem."""
from src.providers.base import ProviderError
p = self._provider({
"/download": self._Resp(403, {"detail": "Forbidden"}),
"/files/file_x": self._Resp(200, {"id": "file_x"}),
})
with pytest.raises(ProviderError) as exc:
p.download_asset("sediment://file_x")
message = str(exc.value.original)
assert "403" in message
assert "no longer exists" not in message
def test_probe_failure_does_not_mask_the_original_403(self):
"""A probe that errors must leave the 403 intact, not swallow it."""
from src.providers.base import ProviderError
p = self._provider({"/download": self._Resp(403, {"detail": "Forbidden"})})
# No entry for /files/file_x — the fake session raises, standing in
# for a network error on the probe.
with pytest.raises(ProviderError) as exc:
p.download_asset("sediment://file_x")
assert "403" in str(exc.value.original)
assert "no longer exists" not in str(exc.value.original)
def test_no_probe_on_the_happy_path(self):
"""The probe costs a request — it must not fire on a good download."""
class _Bytes:
status_code = 200
content = b"data"
headers = {"content-type": "image/png"}
p = self._provider({
"/download": self._Resp(200, {"download_url": "https://cdn/x", "file_name": "a.png"}),
"https://cdn/x": _Bytes(),
})
content, mime, name = p.download_asset("sediment://file_x")
assert (content, mime, name) == (b"data", "image/png", "a.png")
assert not any(u.endswith("/files/file_x") for u in p._calls), "probed needlessly"
def test_classified_as_expired_not_forbidden(self):
"""The run summary must not call a deleted upload a permissions error."""
from src.media import _classify_failure
from src.providers.base import ProviderError
err = ProviderError(
"chatgpt",
"download_asset(file_x)",
RuntimeError(
"Asset no longer exists — HTTP 404 'File not found' on /files/{id}. "
"The upload was deleted or expired server-side; it is not "
"recoverable from ChatGPT."
),
)
assert _classify_failure(err) == "expired-or-missing"