From e3d24f6070cb72f9a6fd1d74ab1eec4b8a04984f Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sat, 22 Aug 2026 09:38:24 +0100 Subject: [PATCH 1/8] Normalize http.client protocol failures at the transport seam MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A server closing the socket mid-body raises http.client.IncompleteRead out of response.read() — the routine failure mode for a 100MB podcast enclosure. IncompleteRead is an HTTPException, not an OSError, so every caller's `except OSError` connection guard missed it and a truncated download landed as `error` with a filed issue, frozen until the next engine release, where blocked-with-retries is the honest outcome. Normalizing the whole http.client family into ConnectionError at the two urllib seams (the driver transport and whisper-api's multipart POST) fixes all eight call sites at once, including the media stage and the enclosure download, and routes them through classify_connection. Co-Authored-By: Claude Fable 5 --- design/ingestion-pipeline.md | 7 ++ .../capabilities/transcribe/whisper_api.py | 15 ++-- src/dex_engine/drivers/transport.py | 76 ++++++++++++++----- tests/drivers/conftest.py | 43 ++++++++++- tests/drivers/test_transport.py | 52 +++++++++++++ tests/pipeline/test_transcribe.py | 30 +++++++- 6 files changed, 195 insertions(+), 28 deletions(-) create mode 100644 tests/drivers/test_transport.py diff --git a/design/ingestion-pipeline.md b/design/ingestion-pipeline.md index 3fff3ca..4ed4733 100644 --- a/design/ingestion-pipeline.md +++ b/design/ingestion-pipeline.md @@ -412,6 +412,13 @@ independently is seven chances to reintroduce it): the classifier is total; no HTTP outcome is unclassifiable. Routed through by every driver's HTTP path. The §15 regression pin tests the classifier once and holds for all drivers. +- **The transport seam normalizes `http.client`'s protocol failures into + `OSError`** before any caller sees them, so the connection classifier + covers them like any other. `IncompleteRead` — a server closing the + socket mid-body, the routine failure mode for a 100MB enclosure — is an + `HTTPException`, not an `OSError`, and would otherwise slip past every + `except OSError` guard and land as `error` + a filed issue. A truncated + read is a connection failure: **`blocked`, retried**, never an engine bug. - **Media URLs are hashed un-canonicalized** — signed query params ARE the resource. Side effect, accepted: an expiring signed URL re-mints a fresh entry per parent rerun, and the stale one retires through normal blocked diff --git a/src/dex_engine/capabilities/transcribe/whisper_api.py b/src/dex_engine/capabilities/transcribe/whisper_api.py index 6bf7378..887a548 100644 --- a/src/dex_engine/capabilities/transcribe/whisper_api.py +++ b/src/dex_engine/capabilities/transcribe/whisper_api.py @@ -29,6 +29,7 @@ from pathlib import Path from typing import Protocol +from dex_engine.drivers.transport import normalize_httplib_errors from dex_engine.pipeline.classify import ProviderInputError, ProviderUnavailableError, scrub from dex_engine.pipeline.types import Availability @@ -96,7 +97,8 @@ def urllib_multipart_post( ``(status, body)`` — HTTP failures return, never raise. Raises: - OSError: Connection-level failure (DNS, refused, reset, timeout). + OSError: Connection-level failure (DNS, refused, reset, timeout, a + body truncated mid-read — normalized by the transport seam). """ boundary = uuid.uuid4().hex parts: list[bytes] = [] @@ -116,11 +118,12 @@ def urllib_multipart_post( if api_key: headers["Authorization"] = f"Bearer {api_key}" request = urllib.request.Request(url, data=body, headers=headers, method="POST") # noqa: S310 — https endpoint from config - try: - with urllib.request.urlopen(request, timeout=_TIMEOUT_SECONDS) as response: # noqa: S310 - return response.status, response.read() - except urllib.error.HTTPError as e: - return e.code, e.read() + with normalize_httplib_errors(): + try: + with urllib.request.urlopen(request, timeout=_TIMEOUT_SECONDS) as response: # noqa: S310 + return response.status, response.read() + except urllib.error.HTTPError as e: + return e.code, e.read() def run_ffmpeg(args: list[str]) -> None: diff --git a/src/dex_engine/drivers/transport.py b/src/dex_engine/drivers/transport.py index 05a33b1..59c07fd 100644 --- a/src/dex_engine/drivers/transport.py +++ b/src/dex_engine/drivers/transport.py @@ -5,7 +5,9 @@ returns an :class:`HttpResponse` for *any* HTTP-level response — 4xx/5xx included, so callers can route status codes through the central classifier — and lets connection-level failures (DNS, refused, timeout) propagate as -``OSError`` for ``classify_connection``. +``OSError`` for ``classify_connection``. ``http.client``'s own protocol +failures are normalized into that same ``OSError`` shape here, once, rather +than at each of the eight call sites (:func:`normalize_httplib_errors`). The browser UA is deliberate: the motivating incident was Cloudflare challenging trafilatura's own fetch client; urllib with a browser UA avoids @@ -13,8 +15,10 @@ """ import contextlib +import http.client import urllib.error import urllib.request +from collections.abc import Iterator from dataclasses import dataclass from typing import Protocol @@ -23,6 +27,7 @@ "DEFAULT_TIMEOUT", "HttpResponse", "Transport", + "normalize_httplib_errors", "urllib_transport", ] @@ -60,6 +65,33 @@ def __call__(self, url: str, *, method: str = "GET") -> HttpResponse: ... +def _httplib_reason(exc: http.client.HTTPException) -> str: + if isinstance(exc, http.client.IncompleteRead): + expected = f", {exc.expected} more expected" if exc.expected else "" + return f"truncated response body ({len(exc.partial)} bytes read{expected})" + return f"{type(exc).__name__}: {exc}" + + +@contextlib.contextmanager +def normalize_httplib_errors() -> Iterator[None]: + """Re-raise ``http.client`` protocol failures as ``ConnectionError``. + + ``http.client.HTTPException`` is NOT an ``OSError``: a server closing + the socket mid-body — the routine failure mode for a 100MB podcast + enclosure — raises ``IncompleteRead`` from ``response.read()``, which + every caller's ``except OSError`` connection guard misses, so a + truncated download lands as an engine bug with an issue filed instead + of a retryable ``blocked``. Normalizing at the seam classifies it as + the connection failure it is, for every transport caller at once. + (``RemoteDisconnected`` already inherits ``ConnectionResetError`` and + needs no help; it passes through this guard unchanged in meaning.) + """ + try: + yield + except http.client.HTTPException as e: + raise ConnectionError(_httplib_reason(e)) from e + + def _media_type(content_type: str | None) -> str: return (content_type or "").split(";")[0].strip().lower() @@ -82,30 +114,36 @@ def urllib_transport(url: str, *, method: str = "GET") -> HttpResponse: Raises: ValueError: ``url`` is not http(s). - OSError: Connection-level failure (DNS, refused, reset, timeout); - ``urllib.error.URLError`` is an ``OSError`` subclass. + OSError: Connection-level failure (DNS, refused, reset, timeout, a + truncated body); ``urllib.error.URLError`` is an ``OSError`` + subclass and ``http.client``'s family is normalized into one. """ if not url.startswith(("http://", "https://")): raise ValueError(f"transport fetches http(s) URLs only, got {url!r}") request = urllib.request.Request( # noqa: S310 — scheme checked above url, headers={"User-Agent": BROWSER_UA}, method=method ) - try: - with urllib.request.urlopen(request, timeout=DEFAULT_TIMEOUT) as response: # noqa: S310 - body = b"" if method == "HEAD" else response.read() + with normalize_httplib_errors(): + try: + with urllib.request.urlopen(request, timeout=DEFAULT_TIMEOUT) as response: # noqa: S310 + body = b"" if method == "HEAD" else response.read() + return HttpResponse( + status=response.status, + content_type=_media_type(response.headers.get("Content-Type")), + body=body, + content_length=_content_length(response.headers.get("Content-Length")), + ) + except urllib.error.HTTPError as e: + body = b"" + # A truncated error page still classifies by its status: the + # partial body is only detail, so the read failure is dropped. + with contextlib.suppress(OSError, ValueError, http.client.HTTPException): + body = e.read() return HttpResponse( - status=response.status, - content_type=_media_type(response.headers.get("Content-Type")), + status=e.code, + content_type=_media_type(e.headers.get("Content-Type") if e.headers else None), body=body, - content_length=_content_length(response.headers.get("Content-Length")), + content_length=_content_length( + e.headers.get("Content-Length") if e.headers else None + ), ) - except urllib.error.HTTPError as e: - body = b"" - with contextlib.suppress(OSError, ValueError): - body = e.read() - return HttpResponse( - status=e.code, - content_type=_media_type(e.headers.get("Content-Type") if e.headers else None), - body=body, - content_length=_content_length(e.headers.get("Content-Length") if e.headers else None), - ) diff --git a/tests/drivers/conftest.py b/tests/drivers/conftest.py index e13ce12..1a896a6 100644 --- a/tests/drivers/conftest.py +++ b/tests/drivers/conftest.py @@ -1,7 +1,10 @@ """Shared driver-test plumbing: fixture loading, fake transports, work units.""" +import contextlib import json -from collections.abc import Mapping +import socket +import threading +from collections.abc import Iterator, Mapping from pathlib import Path import pytest @@ -75,3 +78,41 @@ def __call__(self, url: str, *, method: str = "GET") -> HttpResponse: @pytest.fixture def fake_transport(): return FakeTransport + + +@contextlib.contextmanager +def truncating_server( + *, body: bytes = b"ID3\x04\x00\x00\x00partial audio", declared: int = 5_000_000 +) -> Iterator[str]: + """Serve one 200 whose Content-Length far exceeds the bytes sent, then hang up. + + A real socket, deliberately: the truncated-body failure lives inside + ``http.client``'s read path, and only a genuine short read raises the + ``IncompleteRead`` the transport has to normalize. Yields the URL. + """ + listener = socket.socket() + listener.bind(("127.0.0.1", 0)) + listener.listen(8) + host, port = listener.getsockname() + + def serve() -> None: + while True: + try: + conn, _ = listener.accept() + except OSError: + return # the listener closed: the context manager is done + with conn, contextlib.suppress(OSError): + conn.recv(65536) # drain the request so the close sends FIN, not RST + conn.sendall( + b"HTTP/1.1 200 OK\r\nContent-Type: audio/mpeg\r\n" + b"Content-Length: %d\r\nConnection: close\r\n\r\n%s" % (declared, body) + ) + conn.shutdown(socket.SHUT_WR) + + thread = threading.Thread(target=serve, daemon=True) + thread.start() + try: + yield f"http://{host}:{port}/ep42.mp3" + finally: + listener.close() + thread.join(timeout=5) diff --git a/tests/drivers/test_transport.py b/tests/drivers/test_transport.py new file mode 100644 index 0000000..99aea80 --- /dev/null +++ b/tests/drivers/test_transport.py @@ -0,0 +1,52 @@ +"""Tests for the HTTP seam: what it returns, and what shape it raises. + +The truncated-body cases run against a real localhost socket — the failure +lives inside ``http.client``'s read path and no in-process double reproduces +it faithfully. +""" + +import http.client + +import pytest + +from dex_engine.capabilities import Capabilities +from dex_engine.drivers.file import FileDriver +from dex_engine.drivers.transport import normalize_httplib_errors, urllib_transport +from dex_engine.pipeline.classify import classify_connection +from dex_engine.pipeline.types import Kind, Status +from tests.drivers.conftest import make_unit, truncating_server + + +class TestHttplibNormalization: + def test_incomplete_read_becomes_a_classified_connection_failure(self): + # A server that closes mid-body — the routine failure mode for a + # 100MB enclosure — must not escape as an engine bug: IncompleteRead + # is an HTTPException, which no caller's `except OSError` catches. + with truncating_server() as url, pytest.raises(OSError) as caught: # noqa: PT011 — the OSError shape IS the assertion + urllib_transport(url) + assert "truncated response body" in str(caught.value) + assert classify_connection(caught.value).status is Status.BLOCKED + + def test_the_family_normalizes_not_just_incomplete_read(self): + with pytest.raises(OSError, match="BadStatusLine"), normalize_httplib_errors(): + raise http.client.BadStatusLine("garbage") + + def test_an_oserror_passes_through_untouched(self): + # RemoteDisconnected already IS a ConnectionResetError; the guard + # must not re-wrap what classify_connection already reads. + original = ConnectionResetError("reset by peer") + with pytest.raises(ConnectionResetError) as caught, normalize_httplib_errors(): + raise original + assert caught.value is original + + +class TestThroughADriverFetch: + def test_truncated_download_is_blocked_never_an_engine_error(self): + driver = FileDriver( + capabilities=Capabilities(transcribers=(), extractors=()), + transport=urllib_transport, + ) + with truncating_server() as url: + result = driver.fetch(make_unit(url, Kind.FILE)) + assert result.status is Status.BLOCKED + assert "truncated response body" in (result.reason or "") diff --git a/tests/pipeline/test_transcribe.py b/tests/pipeline/test_transcribe.py index bc791f1..f0c3e87 100644 --- a/tests/pipeline/test_transcribe.py +++ b/tests/pipeline/test_transcribe.py @@ -10,7 +10,7 @@ from dex_engine import corpus from dex_engine.capabilities import Capabilities from dex_engine.drivers.podcast import PodcastDriver -from dex_engine.drivers.transport import HttpResponse +from dex_engine.drivers.transport import HttpResponse, urllib_transport from dex_engine.drivers.youtube import ProbeError, _video_meta from dex_engine.pipeline import ledger from dex_engine.pipeline import run as run_mod @@ -39,7 +39,12 @@ from dex_engine.pipeline.urls import work_hash from tests.capabilities.conftest import FakeTranscriber, fixture_bytes from tests.conftest import FakeDriver -from tests.drivers.conftest import FakeTransport, fixture_text, html_response +from tests.drivers.conftest import ( + FakeTransport, + fixture_text, + html_response, + truncating_server, +) from tests.pipeline.test_run import ITEM, TODAY, entry_for, make_ctx, write_item VIDEO_URL = "https://youtube.com/watch?v=abc123" @@ -527,6 +532,27 @@ def test_enclosure_fetch_failure_takes_the_blocked_lifecycle(self, instance): assert entry.needs is Need.TRANSCRIBE # the typed routing signal assert (entry.reason or "").startswith("audio acquisition failed") + def test_truncated_enclosure_is_blocked_never_an_engine_error(self, instance): + # A server hanging up mid-body is the routine failure for a 100MB + # enclosure; the IncompleteRead it raises is not an OSError, so + # unnormalized it landed `error` + a filed issue, frozen until the + # next engine release. + self.park_via_driver(instance) + entry = self.entry(instance) + record = instance.enrichment_dir / ITEM / f"podcast-{entry.hash[:6]}.md" + with truncating_server() as url: + record.write_text( + record.read_text(encoding="utf-8").replace(self.ENCLOSURE, url), encoding="utf-8" + ) + ctx = transcribe_ctx(instance, transport=urllib_transport) + run_mod.run_transcribe(ctx) + drained = ledger.load(instance.ledger_path)[entry.hash] + assert drained.status is Status.BLOCKED + assert drained.attempts == 1 + assert "truncated response body" in (drained.reason or "") + # The issue filer fires on `error` outcomes only — none here. + assert drained.error is None + def test_gone_enclosure_is_manual_with_the_reresolve_route(self, instance): # A 404ing enclosure is often an expired signed URL — the episode is # NOT confirmed gone; manual, with the requeue route stated. From 3187f030d60750110b5cc8a09077e1bb7e080ad3 Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sat, 22 Aug 2026 09:42:49 +0100 Subject: [PATCH 2/8] Keep whisper-local's probe and decode failures inside the contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two escapes from the same provider. HuggingFace raises HFValidationError from the cache lookup for repo-id shapes like `Systran/faster-whisper/ large-v3` that the name check waves through on its slash — and that probe runs on every CLI verb, so a typo in transcribe_model crashed status, run and transcribe outright. The cache check is now guarded: a name HF rejects reports one unavailable provider, an unreadable cache says so. And faster-whisper indexes a container's audio streams unguarded, so a video-only file arrives as a bare IndexError — an engine bug with a filed issue, where whisper-api already calls the same file manual. It maps to ProviderInputError like every other bad input. Co-Authored-By: Claude Fable 5 --- design/ingestion-pipeline.md | 6 ++- .../capabilities/transcribe/whisper_local.py | 48 ++++++++++++++---- tests/capabilities/test_whisper_local.py | 49 +++++++++++++++++++ tests/pipeline/test_transcribe.py | 44 +++++++++++++++++ 4 files changed, 137 insertions(+), 10 deletions(-) diff --git a/design/ingestion-pipeline.md b/design/ingestion-pipeline.md index 4ed4733..756eb9f 100644 --- a/design/ingestion-pipeline.md +++ b/design/ingestion-pipeline.md @@ -492,6 +492,9 @@ mechanism, not a hack. drop to `small` for long/backlogged queues, stay up for dense technical audio). First run downloads the model (~HF cache, once per machine) — the report surfaces it so slow-first-run is explained. No file-length limits. + **The availability probe never raises**: every CLI verb runs it, so a + model name HuggingFace rejects (`a/b/c`) costs one unavailable provider + with the reason stated — never a crashed `status`/`run`/`transcribe`. - `whisper-api` — one provider class, OpenAI-compatible, `base_url` + key from config/env. Pointed at Groq et al: GPU-fast, ~pennies/hour (roadmap item tracks the provider investigation). **ffmpeg chunking** (~20-min @@ -532,7 +535,8 @@ mechanism, not a hack. Provider contract: `available()` failures (model missing, broken install) park jobs as `waiting` with the reason — the wait list is normally empty for transcription, not absent. A provider raises `ProviderInputError` for -bad-input cases (corrupt audio, malformed file) → the run layer maps it +bad-input cases (corrupt audio, a container carrying no audio stream at +all, malformed file) → the run layer maps it `manual`; **`ProviderUnavailableError`** for call-time availability failures (API 5xx/429, missing binary, model-download failure) → the job **re-parks `waiting`** with the reason, never burning blocked attempts; an diff --git a/src/dex_engine/capabilities/transcribe/whisper_local.py b/src/dex_engine/capabilities/transcribe/whisper_local.py index eab12ad..05ecd8a 100644 --- a/src/dex_engine/capabilities/transcribe/whisper_local.py +++ b/src/dex_engine/capabilities/transcribe/whisper_local.py @@ -57,6 +57,11 @@ def model_is_cached(model: str) -> bool: Returns: True when the converted model's weights are on disk. + + Raises: + ValueError: HuggingFace rejects ``model`` as a repo id (``a/b/c``). + ``available()`` turns this into an unavailable, never a crash. + OSError: The HF cache is unreadable. """ from faster_whisper.utils import _MODELS # noqa: PLC0415 — lazy from huggingface_hub import try_to_load_from_cache # noqa: PLC0415 — lazy @@ -73,6 +78,16 @@ def _known_model(model: str) -> bool: return "/" in model or model in _MODELS +def _unknown_model(model: str) -> Availability: + return Availability( + ok=False, + reason=( + f"unknown whisper model {model!r} — use a faster-whisper size name " + "or an HF repo id" + ), + ) + + def _load_failure_types() -> tuple[type[Exception], ...]: """The load-time failure classes that mean "not available right now". @@ -120,7 +135,7 @@ def available(self) -> Availability: An uncached model is still available — the first transcription downloads it — but the fact is noted so the report can surface the - slow first run. + slow first run. Nothing here may raise: every CLI verb probes. """ try: import faster_whisper # noqa: F401, PLC0415 — lazy availability probe @@ -129,14 +144,22 @@ def available(self) -> Availability: # probe itself must never crash the run. return Availability(ok=False, reason=f"faster-whisper not importable: {scrub(str(e))}") if not _known_model(self.model): + return _unknown_model(self.model) + try: + cached = self._cached(self.model) + except ValueError: + # `a/b/c` has a slash, so _known_model waves it through, and + # HuggingFace then rejects it (HFValidationError, a ValueError) + # from the cache lookup itself. This probe runs on EVERY CLI + # verb: a typo in transcribe_model would otherwise crash + # status/run/transcribe outright instead of reporting one + # unavailable provider. + return _unknown_model(self.model) + except OSError as e: return Availability( - ok=False, - reason=( - f"unknown whisper model {self.model!r} — use a faster-whisper size " - "name or an HF repo id" - ), + ok=False, reason=f"whisper model cache is unreadable: {scrub(str(e))}" ) - if not self._cached(self.model): + if not cached: return Availability( ok=True, reason=( @@ -159,8 +182,8 @@ def transcribe(self, audio: Path, initial_prompt: str) -> str: Raises: ProviderUnavailableError: The model could not be loaded (a failed download, a broken cache) — the job stays waiting. - ProviderInputError: The audio could not be decoded or yielded no - speech — the manual path. + ProviderInputError: The audio could not be decoded, holds no + audio stream, or yielded no speech — the manual path. """ try: model = self._load(self.model) @@ -173,6 +196,13 @@ def transcribe(self, audio: Path, initial_prompt: str) -> str: # faster-whisper decodes lazily — errors for corrupt audio # surface during iteration, so the join stays inside the try. text = "".join(segment.text for segment in segments).strip() + except IndexError as e: + # A container with no audio stream at all (a video-only mp4): + # faster-whisper indexes the stream list unguarded, so the + # failure arrives as a bare IndexError rather than a decode + # error. Bad input like any other — whisper-api already calls + # the same file manual. + raise ProviderInputError(f"no audio stream in {audio.name}") from e except (OSError, ValueError, RuntimeError, EOFError) as e: # PyAV maps FFmpeg decode failures onto OSError/ValueError # subclasses: bad input, by contract — never an engine bug. diff --git a/tests/capabilities/test_whisper_local.py b/tests/capabilities/test_whisper_local.py index 189f044..c57d0db 100644 --- a/tests/capabilities/test_whisper_local.py +++ b/tests/capabilities/test_whisper_local.py @@ -1,7 +1,9 @@ """Tests for whisper-local: fake model seams; the real model is live-only.""" import math +import shutil import struct +import subprocess import wave from pathlib import Path @@ -54,6 +56,26 @@ def test_hf_repo_ids_are_accepted(self): availability = local(FakeModel([]), model="Systran/faster-whisper-large-v3").available() assert availability.ok is True + def test_a_repo_id_huggingface_rejects_is_unavailable_not_a_crash(self): + # `Systran/faster-whisper/large-v3` (an easy typo for the real + # `Systran/faster-whisper-large-v3`) has a slash, so the name check + # waves it through and HF's cache lookup raises HFValidationError. + # The real cache seam runs here — the point is that it never + # escapes the probe every CLI verb makes. + availability = WhisperLocal(model="Systran/faster-whisper/large-v3").available() + assert availability.ok is False + assert "unknown whisper model" in availability.reason + + def test_an_unreadable_cache_is_unavailable_with_its_own_reason(self): + def cached(_model: str) -> bool: + raise PermissionError(13, "Permission denied") + + model = FakeModel([]) + provider = WhisperLocal(model="medium", load=lambda _n: model, cached=cached) + availability = provider.available() + assert availability.ok is False + assert "cache is unreadable" in availability.reason + def test_import_probe_never_crashes(self): # On this machine faster-whisper imports; the probe path for a # broken install is the (ImportError, OSError) net — asserted by @@ -81,6 +103,33 @@ def test_decode_failure_is_bad_input(self): with pytest.raises(ProviderInputError, match="could not decode"): local(model).transcribe(Path("/audio/corrupt.mp3"), "") + def test_a_container_with_no_audio_stream_is_bad_input(self): + model = FakeModel([], raise_=IndexError("tuple index out of range")) + with pytest.raises(ProviderInputError, match=r"no audio stream in videoonly\.mp4"): + local(model).transcribe(Path("/audio/videoonly.mp4"), "") + + def test_real_video_only_container_raises_the_indexerror_we_map(self, tmp_path): + # Pins the world half of the mapping above: faster-whisper's own + # decode indexes the stream list unguarded, so a video-only file + # surfaces as IndexError and not as any decode error. Decoding + # needs no model, so this stays off the live marker. + if shutil.which("ffmpeg") is None: + pytest.skip("ffmpeg is not on PATH") + from faster_whisper.audio import decode_audio # noqa: PLC0415 — heavy dep, one test + + video = tmp_path / "videoonly.mp4" + subprocess.run( # noqa: S603 — fixed args, no shell + [ + shutil.which("ffmpeg") or "ffmpeg", "-nostdin", "-loglevel", "error", "-y", + "-f", "lavfi", "-i", "color=c=black:s=64x64:d=1", + "-c:v", "libx264", "-pix_fmt", "yuv420p", str(video), + ], + check=True, + capture_output=True, + ) # fmt: skip + with pytest.raises(IndexError): + decode_audio(str(video)) + def test_silent_audio_is_bad_input(self): with pytest.raises(ProviderInputError, match="no speech"): local(FakeModel([])).transcribe(Path("/audio/silence.wav"), "") diff --git a/tests/pipeline/test_transcribe.py b/tests/pipeline/test_transcribe.py index f0c3e87..c550c71 100644 --- a/tests/pipeline/test_transcribe.py +++ b/tests/pipeline/test_transcribe.py @@ -576,6 +576,50 @@ def test_missing_enrichment_record_is_manual(self, instance): assert "no enrichment record" in (entry.reason or "") +class TestMalformedModelName: + """A model name HuggingFace rejects must cost one provider, not the verb.""" + + MODEL = "Systran/faster-whisper/large-v3" # the real repo id has no third slash + + @pytest.fixture(autouse=True) + def _no_api_key(self, monkeypatch): + # Both transcribers must be unavailable for the drain to park — + # a developer's exported key would otherwise reach a real endpoint. + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + monkeypatch.delenv("OPENAI_BASE_URL", raising=False) + + def ctx(self, instance, *, from_config: bool): + # The config shape and the `--model` shape reach the same provider. + caps = ( + Capabilities.build(Config(transcribe_model=self.MODEL)) + if from_config + else Capabilities.build(Config(), model=self.MODEL) + ) + return make_ctx( + instance, FakeDriver(), capabilities=caps, provider_available=caps.available + ) + + @pytest.mark.parametrize("from_config", [True, False]) + def test_status_reports_the_provider_unavailable(self, instance, from_config): + ctx = self.ctx(instance, from_config=from_config) + report = " ".join(run_mod.status_report(ctx).split()) # the surface wraps + assert "unknown whisper model" in report + + @pytest.mark.parametrize("from_config", [True, False]) + def test_run_and_transcribe_park_the_job_instead_of_crashing(self, instance, from_config): + write_item(instance, urls=[VIDEO_URL]) + seed_waiting(instance) + ctx = self.ctx(instance, from_config=from_config) + run_mod.run(ctx) + assert entry_for(ctx, VIDEO_URL).status is Status.WAITING + report = " ".join(run_mod.run_transcribe(ctx).split()) # the surface wraps + assert "no transcription provider available" in report + assert "unknown whisper model" in report + entry = entry_for(ctx, VIDEO_URL) + assert entry.status is Status.WAITING # no clock — it waits for a provider + assert entry.needs is Need.TRANSCRIBE + + class TestRunAutoDrain: def test_enrich_run_drains_waiting_transcribe_mechanically(self, instance): # Waiting means no mechanical provider — the moment one exists, From cf7fc2bf427ba2701fbe80da13d0139a062feaaf Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sat, 22 Aug 2026 09:45:09 +0100 Subject: [PATCH 3/8] Raise csv-builtin's field bound and state a reader refusal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit csv.Error escapes from the reader's iteration, not its construction, so the comprehension building the rows let it through: a CSV field over the stdlib's 128KB default — routine for an embedded JSON blob — landed as an engine bug with an issue filed. A big field is not a malformed file, so the bound is raised to 16MB rather than the failure merely being caught. It stays finite: one unterminated quote in a file that only looked like CSV makes the whole file a single field, and a bound turns that into a stated manual instead of eating the machine's memory. Anything the reader still refuses is ProviderInputError. The limit is process-global, so it is restored. Co-Authored-By: Claude Fable 5 --- design/ingestion-pipeline.md | 7 ++++- .../capabilities/extract/csv_builtin.py | 30 +++++++++++++++++-- tests/capabilities/test_extract.py | 26 ++++++++++++++++ 3 files changed, 60 insertions(+), 3 deletions(-) diff --git a/design/ingestion-pipeline.md b/design/ingestion-pipeline.md index 756eb9f..2ed9583 100644 --- a/design/ingestion-pipeline.md +++ b/design/ingestion-pipeline.md @@ -522,7 +522,12 @@ mechanism, not a hack. form) or returns none — graceful text-only degradation. Images merely *linked* from a document stay links in the markdown, like web body links — harvest judgment promotes them if they matter. -- `csv-builtin` — stdlib, zero deps. +- `csv-builtin` — stdlib, zero deps. The per-field size bound is raised far + past the stdlib's 128KB default — an embedded JSON blob is a big cell in a + fine file — but stays finite, since one unterminated quote in a file that + merely looked like CSV makes the whole file a single field. A reader + refusal (past the bound, a stray newline in an unquoted field) is stated + bad input → `manual`, never an engine bug. - `cognitive` — floor: parks `needs: extract` for the ingest session. - The **Format is the contract, not the tool**: providers register per format; if anydoc dies, each format falls back independently (or parks) and diff --git a/src/dex_engine/capabilities/extract/csv_builtin.py b/src/dex_engine/capabilities/extract/csv_builtin.py index 4124aeb..d1d0262 100644 --- a/src/dex_engine/capabilities/extract/csv_builtin.py +++ b/src/dex_engine/capabilities/extract/csv_builtin.py @@ -15,6 +15,13 @@ _SNIFF_SAMPLE_CHARS = 4096 _FALLBACK_DELIMITERS = ",;\t|" +# The stdlib's 128KB-per-field default trips on an embedded JSON blob, +# which is a big cell in a fine file, not a broken one — so the limit is +# raised. Not removed: one unterminated quote in a file that only looked +# like CSV makes the whole file a single field, and a bound turns that +# into a stated `manual` instead of eating the machine's memory. +_MAX_FIELD_CHARS = 16 * 1024 * 1024 + class CsvBuiltinExtractor: """Render a CSV as a markdown table; the first row is the header.""" @@ -41,12 +48,14 @@ def extract(self, data: bytes, fmt: Format) -> Extraction: The extraction; CSVs embed nothing, so assets are always empty. Raises: - ProviderInputError: Not CSV work, or the file has no rows. + ProviderInputError: Not CSV work, the file has no rows, or the + reader refused it (a field past the size bound, a stray + newline in an unquoted field). """ if not self.supports(fmt): raise ProviderInputError(f"csv-builtin extracts CSV only, got {fmt.value!r}") text = data.decode("utf-8-sig", errors="replace") - rows = [row for row in csv.reader(io.StringIO(text), dialect=_dialect(text)) if row] + rows = _rows(text) if not rows: raise ProviderInputError("CSV has no rows") width = max(len(row) for row in rows) @@ -56,6 +65,23 @@ def extract(self, data: bytes, fmt: Format) -> Extraction: return Extraction(markdown="\n".join(lines) + "\n") +def _rows(text: str) -> list[list[str]]: + """Every non-empty row, or a stated bad input — csv.Error never escapes. + + The reader raises during ITERATION, not construction, so the whole + comprehension sits inside the guard. + """ + previous = csv.field_size_limit(_MAX_FIELD_CHARS) + try: + return [row for row in csv.reader(io.StringIO(text), dialect=_dialect(text)) if row] + except csv.Error as e: + raise ProviderInputError(f"CSV could not be read: {e}") from e + finally: + # The limit is process-global state; leaving it raised would + # change how every later reader in this run behaves. + csv.field_size_limit(previous) + + def _dialect(text: str) -> type[csv.Dialect] | csv.Dialect: """Sniff the delimiter from a leading sample; excel (comma) is the fallback.""" try: diff --git a/tests/capabilities/test_extract.py b/tests/capabilities/test_extract.py index c183e58..a9f7605 100644 --- a/tests/capabilities/test_extract.py +++ b/tests/capabilities/test_extract.py @@ -1,7 +1,10 @@ """Tests for the extract providers: real anydoc over fixtures, csv-builtin, floors.""" +import csv + import pytest +from dex_engine.capabilities.extract import csv_builtin from dex_engine.capabilities.extract.anydoc import AnydocExtractor from dex_engine.capabilities.extract.cognitive import CognitiveExtractor from dex_engine.capabilities.extract.csv_builtin import CsvBuiltinExtractor @@ -85,6 +88,29 @@ def test_non_csv_work_is_refused(self): with pytest.raises(ProviderInputError, match="csv-builtin"): CsvBuiltinExtractor().extract(b"%PDF-1.4", Format.PDF) + def test_a_field_over_the_stdlib_default_extracts(self): + # An embedded JSON blob past the 128KB default is a big cell in a + # fine file — extraction, not a stated failure and not a crash. + blob = "x" * (256 * 1024) + data = f'a,b\n1,"{blob}"\n'.encode() + markdown = CsvBuiltinExtractor().extract(data, Format.CSV).markdown + assert blob in markdown + + def test_a_field_past_the_bound_is_bad_input(self, monkeypatch): + # csv.Error escapes from the reader's ITERATION; unhandled it was + # an engine bug with an issue filed over a routine wide file. + monkeypatch.setattr(csv_builtin, "_MAX_FIELD_CHARS", 64) + data = b'a,b\n1,"' + b"x" * 512 + b'"\n' + with pytest.raises(ProviderInputError, match="could not be read"): + CsvBuiltinExtractor().extract(data, Format.CSV) + + def test_the_process_wide_field_limit_is_restored(self, monkeypatch): + monkeypatch.setattr(csv_builtin, "_MAX_FIELD_CHARS", 64) + before = csv.field_size_limit() + with pytest.raises(ProviderInputError): + CsvBuiltinExtractor().extract(b'a\n"' + b"x" * 512 + b'"\n', Format.CSV) + assert csv.field_size_limit() == before + class TestCognitiveFloors: @pytest.mark.parametrize("floor", [CognitiveExtractor(), CognitiveOcr()]) From d14c70f2943b72c51d808479c4bbf9e59baa82a5 Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sat, 22 Aug 2026 09:52:01 +0100 Subject: [PATCH 4/8] Budget the whisper prompt head-first and guard the enclosure body MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Whisper keeps a prompt's LAST ~224 tokens and discards the front, so an 800-char priming string (~270 tokens of jargon-dense text) threw away the title and show it exists to carry. The composed prompt is now budgeted to ~200 tokens: the head survives whole, the vocabulary is trimmed from its own tail, and whisper-api's continuity tail shares that budget instead of being appended past it. Second, a CDN error page cached as .mp3 decoded as garbage and parked the episode manual forever. An empty body, a body short of its declared Content-Length, or an HTML page where audio was expected is now blocked and never written under the audio name — nothing about the episode was learned. Bytes that are merely bad audio still reach the provider and its manual stands. The file driver's HTML-lead sniff moves to pipeline/detect, where the byte sniffing lives, so both callers read the same bytes. Co-Authored-By: Claude Fable 5 --- design/ingestion-pipeline.md | 17 ++- .../capabilities/transcribe/whisper_api.py | 22 ++- src/dex_engine/drivers/file.py | 11 +- src/dex_engine/pipeline/detect.py | 19 +++ src/dex_engine/pipeline/transcribe.py | 77 ++++++++-- tests/capabilities/test_whisper_api.py | 14 +- tests/pipeline/test_transcribe.py | 132 +++++++++++++++++- 7 files changed, 266 insertions(+), 26 deletions(-) diff --git a/design/ingestion-pipeline.md b/design/ingestion-pipeline.md index 2ed9583..210ad89 100644 --- a/design/ingestion-pipeline.md +++ b/design/ingestion-pipeline.md @@ -507,7 +507,14 @@ mechanism, not a hack. `needs: transcribe` with the pointer. - Accuracy: **`initial_prompt` priming** with the item's known vocabulary (video title + description, episode title + show notes) — mechanical, free, - large win on names/jargon. Transcripts are stored **raw**, stamped + large win on names/jargon. **Whisper keeps a prompt's LAST ~224 tokens and + discards its front**, so the composed prompt is budgeted to ~200 tokens + (600 chars, at the 3 chars/token this jargon-dense text measures): the + title/show head always survives and the VOCABULARY is trimmed from its + tail. whisper-api's chunk-continuity tail shares that one budget rather + than adding to it — otherwise the continuity a later chunk gains would + cost it the two names priming exists to carry. Transcripts are stored + **raw**, stamped `via`/`model` in frontmatter; corrections live downstream in digest/wiki where judgment already operates — enrichment stays the mechanical record. @@ -551,7 +558,13 @@ retry on new engine). The availability seam is **per-format** — provider. **Acquisition failures are not provider failures**: a failed audio download (yt-dlp breakage, blocked enclosure GET) classifies through the normal §5 lifecycle — `blocked` with attempts, escalating to `manual` -at 5 — never as no-clock waiting. The blocked line keeps +at 5 — never as no-clock waiting. **A 200 that cannot be the audio is an +acquisition failure too**: an empty body, a body short of its declared +`Content-Length`, or an HTML page where audio was expected (a CDN's cached +error page) is `blocked` and never written under the audio name — cached as +`.mp3` it decodes as garbage and parks the episode `manual` forever +over a fault that has nothing to do with the episode. Bytes that are merely +bad audio still reach the provider, and its `manual` stands. The blocked line keeps `needs: transcribe`, so the retry routes back through the transcribe drain rather than the driver (§3: a typed field, never the reason's wording). Config keys: `transcribe_base_url`, diff --git a/src/dex_engine/capabilities/transcribe/whisper_api.py b/src/dex_engine/capabilities/transcribe/whisper_api.py index 887a548..213c753 100644 --- a/src/dex_engine/capabilities/transcribe/whisper_api.py +++ b/src/dex_engine/capabilities/transcribe/whisper_api.py @@ -31,6 +31,7 @@ from dex_engine.drivers.transport import normalize_httplib_errors from dex_engine.pipeline.classify import ProviderInputError, ProviderUnavailableError, scrub +from dex_engine.pipeline.transcribe import PROMPT_MAX_CHARS from dex_engine.pipeline.types import Availability __all__ = [ @@ -50,9 +51,10 @@ # cap at any sane audio bitrate. CHUNK_SECONDS = 1200 -# Whisper's prompt window is ~224 tokens; the continuity tail plus the -# vocabulary priming must fit, so the tail is bounded in characters. -_CONTINUITY_TAIL_CHARS = 400 +# The continuity tail's share of PROMPT_MAX_CHARS: ~65 tokens of the +# running transcript is ample for carrying word boundaries and style into +# the next chunk, and it leaves two thirds of the window to the priming. +_CONTINUITY_TAIL_CHARS = 200 _HTTP_OK_FLOOR, _HTTP_OK_CEILING = 200, 300 _HTTP_BAD_REQUEST = 400 @@ -310,8 +312,18 @@ def _transcribe_chunk(self, chunk: Path, prompt: str) -> str: def _chunk_prompt(initial_prompt: str, previous: list[str]) -> str: - """Vocabulary priming plus the running transcript's tail (chunk continuity).""" + """Vocabulary priming plus the running transcript's tail (chunk continuity). + + Both halves live inside one ``PROMPT_MAX_CHARS`` budget, because + whisper reads a prompt from its END: appending an unbudgeted tail + would push the priming's head — the title and show — out of the + window it exists to occupy. The tail comes off the transcript's end, + the priming off its own end, so the two names always survive. + """ if not previous: return initial_prompt tail = " ".join(previous)[-_CONTINUITY_TAIL_CHARS:].strip() - return f"{initial_prompt}\n{tail}".strip() + if not tail: + return initial_prompt + head = initial_prompt[: max(PROMPT_MAX_CHARS - len(tail) - 1, 0)].rstrip() + return f"{head}\n{tail}".strip() diff --git a/src/dex_engine/drivers/file.py b/src/dex_engine/drivers/file.py index 6e44c62..b67051c 100644 --- a/src/dex_engine/drivers/file.py +++ b/src/dex_engine/drivers/file.py @@ -30,7 +30,7 @@ classify_connection, classify_http, ) -from dex_engine.pipeline.detect import sniff_format +from dex_engine.pipeline.detect import looks_like_html, sniff_format from dex_engine.pipeline.types import ( Format, Kind, @@ -48,13 +48,6 @@ _LFS_POINTER_PREFIX = b"version https://git-lfs" -_HTML_LEADS = (b" bool: - lead = data.removeprefix(b"\xef\xbb\xbf").lstrip()[:64].lower() - return lead.startswith(_HTML_LEADS) - class FileDriver: """Extract captured or URL-served binaries via the extract registry.""" @@ -115,7 +108,7 @@ def fetch(self, unit: WorkUnit) -> Result: if ( not unit.url.startswith("file:") and sniff_format(data) is None - and _looks_like_html(data) + and looks_like_html(data) ): return Result(status=Status.QUEUED, meta={}, redetect=Redetection(kind=Kind.WEB)) fmt = sniff_format(data, name=name) or unit.format diff --git a/src/dex_engine/pipeline/detect.py b/src/dex_engine/pipeline/detect.py index 3107f9a..80f1ec2 100644 --- a/src/dex_engine/pipeline/detect.py +++ b/src/dex_engine/pipeline/detect.py @@ -31,9 +31,28 @@ "canonical_url", "detect", "detect_kind", + "looks_like_html", "sniff_format", ] +# HTML leads, BOM- and whitespace-tolerant. Bytes decide what a body IS, +# never the content type a server claimed: shared by the file driver's +# re-route to web work and the enclosure download's error-page guard. +_HTML_LEADS = (b" bool: + """Whether the leading bytes read as an HTML/XML document. + + Args: + data: The leading bytes (the whole body is fine). + + Returns: + True for a markup lead. + """ + lead = data.removeprefix(b"\xef\xbb\xbf").lstrip()[:64].lower() + return lead.startswith(_HTML_LEADS) + # url -> media type ("application/pdf"), or None when the HEAD itself failed. # A failed sniff is INCONCLUSIVE, never classified: the GET that follows # will surface the real status through the classifier. diff --git a/src/dex_engine/pipeline/transcribe.py b/src/dex_engine/pipeline/transcribe.py index 083e48e..e7e0fbb 100644 --- a/src/dex_engine/pipeline/transcribe.py +++ b/src/dex_engine/pipeline/transcribe.py @@ -21,13 +21,15 @@ from urllib.parse import urlsplit from dex_engine import atomic -from dex_engine.drivers.transport import Transport +from dex_engine.drivers.transport import HttpResponse, Transport from dex_engine.drivers.youtube import ProbeError, classify_probe_failure from .classify import Classification, classify_connection, classify_http +from .detect import looks_like_html from .types import LedgerEntry, Status __all__ = [ + "PROMPT_MAX_CHARS", "TRANSCRIBE_RUN_CAP", "Acquired", "DownloadAudio", @@ -44,10 +46,15 @@ # never monopolize a machine. `enrich transcribe --limit` overrides. TRANSCRIBE_RUN_CAP = 10 -# Whisper's prompt window is ~224 tokens; priming beyond it is discarded -# anyway, so the vocabulary text is bounded in characters. -_PROMPT_MAX_CHARS = 800 +# Whisper's prompt window is ~224 tokens and it keeps the LAST of them: +# an overlong prompt loses its FRONT, which is exactly the title/show the +# priming exists to carry. So the whole composed prompt — vocabulary here +# plus whisper-api's continuity tail — is budgeted to ~200 tokens. Jargon +# tokenizes badly: 800 chars of this text measured ~270 tokens (≈2.96 +# chars/token), so 3 chars/token is the conservative rate and 600 the cap. +PROMPT_MAX_CHARS = 600 +_SEPARATOR = " — " _AUDIO_EXT_DEFAULT = "mp3" _TRANSCRIPT_HEADING = "## Transcript" @@ -260,6 +267,12 @@ def _download_enclosure( return classify_connection(e) if not response.ok: return classify_http(response.status) + unusable = _not_audio(response) + if unusable is not None: + # Never cached under .: a stored error page is + # indistinguishable from audio on the retry, and the provider + # would park the episode manual forever over the CDN's mistake. + return Classification(status=Status.BLOCKED, reason=unusable) cache_dir.mkdir(parents=True, exist_ok=True) path = cache_dir / f"{stem}.{_audio_ext(url)}" # Atomic: a crash mid-write must never leave a truncated file under the @@ -268,6 +281,28 @@ def _download_enclosure( return path +def _not_audio(response: HttpResponse) -> str | None: + """Why this 200 body cannot be the episode's audio, or None. + + A CDN serving an error page — or half a body — under a 200 says + nothing about the EPISODE, so it is ``blocked`` and retried, not the + ``manual`` a provider would produce after failing to decode the + garbage. Only shapes that are certainly not audio are caught here; + real audio that turns out to be broken still reaches the provider and + still parks manual with what the decoder said. + """ + if not response.body: + return "enclosure returned an empty response body" + if response.content_length is not None and len(response.body) < response.content_length: + return ( + f"enclosure body is truncated ({len(response.body)} of " + f"{response.content_length} declared bytes)" + ) + if looks_like_html(response.body): + return f"enclosure served an HTML page ({response.content_type or 'no content type'})" + return None + + def _audio_ext(url: str) -> str: tail = urlsplit(url).path.rsplit("/", 1)[-1] ext = tail.rsplit(".", 1)[-1].lower() if "." in tail else "" @@ -276,10 +311,36 @@ def _audio_ext(url: str) -> str: return ext -def _prompt(*parts: str | None) -> str: - """Known-vocabulary priming, single-line, bounded.""" - text = " — ".join(" ".join(part.split()) for part in parts if part) - return text[:_PROMPT_MAX_CHARS] +def _prompt(title: str | None, show: str | None, vocabulary: str) -> str: + """Known-vocabulary priming, single-line, budgeted head-first. + + The head (title, then channel/show) always survives: whisper reads a + prompt from its END, so the VOCABULARY is what gets truncated — from + its own tail — to fit :data:`PROMPT_MAX_CHARS`. Truncating the + composed string instead would keep 600 characters of show notes and + throw away the two names the priming was built for. + + Args: + title: The episode/video title. + show: The show or channel name. + vocabulary: Show notes or description — the trimmable part. + + Returns: + The priming text, at most :data:`PROMPT_MAX_CHARS` characters. + """ + head = _SEPARATOR.join(_flat(part) for part in (title, show) if part and part.strip()) + vocab = _flat(vocabulary) + if not head: + return vocab[:PROMPT_MAX_CHARS] + head = head[:PROMPT_MAX_CHARS] + room = PROMPT_MAX_CHARS - len(head) - len(_SEPARATOR) + if not vocab or room <= 0: + return head + return f"{head}{_SEPARATOR}{vocab[:room]}" + + +def _flat(text: str) -> str: + return " ".join(text.split()) # --------------------------------------------------------------------------- diff --git a/tests/capabilities/test_whisper_api.py b/tests/capabilities/test_whisper_api.py index 82877b6..9f8091b 100644 --- a/tests/capabilities/test_whisper_api.py +++ b/tests/capabilities/test_whisper_api.py @@ -12,6 +12,7 @@ _chunk_prompt, ) from dex_engine.pipeline.classify import ProviderInputError, ProviderUnavailableError +from dex_engine.pipeline.transcribe import PROMPT_MAX_CHARS, _prompt class FakePost: @@ -230,4 +231,15 @@ def test_first_chunk_is_the_initial_prompt_alone(self): def test_tail_is_bounded(self): prompt = _chunk_prompt("v", ["word " * 500]) - assert len(prompt) <= 402 # initial + newline + bounded tail + assert len(prompt) <= 202 # initial + newline + bounded tail + + def test_the_continuity_tail_never_pushes_the_title_out_of_the_window(self): + # Whisper keeps the LAST ~224 tokens: an unbudgeted tail appended + # to a full priming string would discard the head the priming + # exists to carry. Both halves share one budget. + priming = _prompt("Ledgers as Work Queues", "Engineering Distilled", "jargon " * 400) + assert len(priming) == PROMPT_MAX_CHARS # the worst case + prompt = _chunk_prompt(priming, ["transcribed words " * 200]) + assert prompt.startswith("Ledgers as Work Queues — Engineering Distilled") + assert prompt.endswith("transcribed words") + assert len(prompt) <= PROMPT_MAX_CHARS diff --git a/tests/pipeline/test_transcribe.py b/tests/pipeline/test_transcribe.py index c550c71..627b52f 100644 --- a/tests/pipeline/test_transcribe.py +++ b/tests/pipeline/test_transcribe.py @@ -14,15 +14,21 @@ from dex_engine.drivers.youtube import ProbeError, _video_meta from dex_engine.pipeline import ledger from dex_engine.pipeline import run as run_mod -from dex_engine.pipeline.classify import ProviderInputError, ProviderUnavailableError +from dex_engine.pipeline.classify import ( + Classification, + ProviderInputError, + ProviderUnavailableError, +) from dex_engine.pipeline.registry import build_drivers from dex_engine.pipeline.run import _Drain from dex_engine.pipeline.transcribe import ( + PROMPT_MAX_CHARS, TRANSCRIBE_RUN_CAP, Acquired, YoutubeAudio, _cached_audio, _download_enclosure, + _prompt, acquire_youtube_audio, read_enrichment, ) @@ -553,6 +559,33 @@ def test_truncated_enclosure_is_blocked_never_an_engine_error(self, instance): # The issue filer fires on `error` outcomes only — none here. assert drained.error is None + def test_cdn_error_page_takes_the_blocked_lifecycle_not_manual(self, instance): + # A 200 carrying an error page cached as .mp3 decoded as + # garbage and parked the episode manual forever; nothing about the + # EPISODE was learned, so it is blocked and retried. + self.park_via_driver(instance) + page = html_response("\nAccess denied") + ctx = transcribe_ctx(instance, transport=FakeTransport({self.ENCLOSURE: page})) + run_mod.run_transcribe(ctx) + entry = ledger.load(instance.ledger_path)[self.entry(instance).hash] + assert entry.status is Status.BLOCKED + assert entry.attempts == 1 + assert entry.needs is Need.TRANSCRIBE + assert "HTML page" in (entry.reason or "") + assert audio_files(instance) == [] # the page never became cached "audio" + + def test_real_audio_that_will_not_decode_still_parks_manual(self, instance): + # The guards must not swallow the genuine bad-audio case: bytes + # that are not markup reach the provider, and its verdict stands. + self.park_via_driver(instance) + angry = FakeTranscriber(raise_=ProviderInputError("could not decode the audio")) + transport = FakeTransport({self.ENCLOSURE: html_response("AUDIO-BYTES")}) + ctx = transcribe_ctx(instance, transcriber=angry, transport=transport) + run_mod.run_transcribe(ctx) + entry = ledger.load(instance.ledger_path)[self.entry(instance).hash] + assert entry.status is Status.MANUAL + assert "could not decode" in (entry.reason or "") + def test_gone_enclosure_is_manual_with_the_reresolve_route(self, instance): # A 404ing enclosure is often an expired signed URL — the episode is # NOT confirmed gone; manual, with the requeue route stated. @@ -802,6 +835,53 @@ def test_completed_download_lands_under_the_final_name(self, tmp_path): assert [p.name for p in tmp_path.iterdir()] == ["abc.mp3"] +class TestEnclosureBodyGuards: + """A 200 that isn't audio is the CDN's failure, not the episode's.""" + + ENCLOSURE = "https://cdn.pods.test/ed/ep42.mp3?sig=abc123" + + def download(self, tmp_path, response) -> object: + transport = FakeTransport({self.ENCLOSURE: response}) + return _download_enclosure(self.ENCLOSURE, tmp_path, "abc", transport) + + def audio(self, body: bytes, **kwargs) -> HttpResponse: + return HttpResponse(status=200, content_type="audio/mpeg", body=body, **kwargs) + + def test_an_empty_body_is_blocked(self, tmp_path): + outcome = self.download(tmp_path, self.audio(b"")) + assert isinstance(outcome, Classification) + assert outcome.status is Status.BLOCKED + assert "empty response body" in outcome.reason + assert list(tmp_path.iterdir()) == [] # nothing cached to poison the retry + + def test_a_body_short_of_its_declared_length_is_blocked(self, tmp_path): + outcome = self.download(tmp_path, self.audio(b"ID3" + b"x" * 17, content_length=5_000_000)) + assert isinstance(outcome, Classification) + assert outcome.status is Status.BLOCKED + assert "truncated" in outcome.reason + assert list(tmp_path.iterdir()) == [] + + def test_an_html_error_page_is_blocked_not_manual(self, tmp_path): + # Cached as .mp3 this decodes as garbage and parks the + # episode manual forever; blocked is the honest reading. + page = HttpResponse( + status=200, + content_type="text/html", + body=b"\nAccess denied", + ) + outcome = self.download(tmp_path, page) + assert isinstance(outcome, Classification) + assert outcome.status is Status.BLOCKED + assert "HTML page" in outcome.reason + assert list(tmp_path.iterdir()) == [] + + def test_a_complete_body_still_downloads(self, tmp_path): + body = b"ID3\x04\x00\x00audio bytes" + outcome = self.download(tmp_path, self.audio(body, content_length=len(body))) + assert isinstance(outcome, Path) + assert outcome.read_bytes() == body + + class TestCachedAudio: def test_hard_crash_atomic_temp_is_a_partial_never_audio(self, tmp_path): # A kill mid-atomic-write can orphan the temp file itself; it must @@ -832,6 +912,56 @@ def test_missing_cache_dir_is_simply_empty(self, tmp_path): assert _cached_audio(tmp_path / "audio", "abc") is None +class TestPromptBudget: + """Whisper reads a prompt from its END — the head must never be the casualty.""" + + TITLE = "Ledgers as Work Queues" + SHOW = "Engineering Distilled" + HEAD = f"{TITLE} — {SHOW}" + + def test_the_head_survives_a_vocabulary_far_past_the_budget(self): + prompt = _prompt(self.TITLE, self.SHOW, "jargon " * 400) + assert prompt.startswith(self.HEAD) + assert len(prompt) == PROMPT_MAX_CHARS + + def test_the_vocabulary_is_trimmed_from_its_tail(self): + prompt = _prompt(self.TITLE, self.SHOW, "anydoc CTranslate2 " + "x" * 2000) + assert prompt.startswith(f"{self.HEAD} — anydoc CTranslate2 ") + assert not prompt.endswith("x" * 2000) + + def test_a_short_prompt_is_untouched(self): + assert _prompt(self.TITLE, self.SHOW, "notes") == f"{self.HEAD} — notes" + assert _prompt(self.TITLE, None, "") == self.TITLE + + def test_vocabulary_alone_is_still_bounded(self): + assert len(_prompt(None, None, "y " * 1000)) == PROMPT_MAX_CHARS + + def test_an_overlong_head_keeps_its_front_and_drops_the_vocabulary(self): + prompt = _prompt("T" * 900, self.SHOW, "notes") + assert prompt == "T" * PROMPT_MAX_CHARS + + def test_the_acquisition_path_composes_within_the_budget(self, instance, tmp_path): + # The real composition site, not just the helper: a talkative + # video description must not cost the title and channel. + entry = seed_waiting(instance) + + def download(_url, cache_dir, stem) -> YoutubeAudio: + cache_dir.mkdir(parents=True, exist_ok=True) + path = cache_dir / f"{stem}.m4a" + path.write_bytes(b"fake-audio") + return YoutubeAudio( + path=path, + title="Ledgers at Scale", + channel="Engineering Distilled", + description="sponsors and links " * 200, + ) + + acquired = acquire_youtube_audio(entry, tmp_path, download) + assert isinstance(acquired, Acquired) + assert acquired.prompt.startswith("Ledgers at Scale — Engineering Distilled") + assert len(acquired.prompt) <= PROMPT_MAX_CHARS + + class TestReadEnrichment: def test_round_trips_quoted_values(self, tmp_path): record = tmp_path / "podcast-abc123.md" From d3eb76d839470929fc19ef96ca738c11a3e894bb Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sat, 22 Aug 2026 09:56:36 +0100 Subject: [PATCH 5/8] Close three provider-boundary gaps in the registry and extract path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The capability report dropped an ok-with-caveat note for available and active providers, so "model not cached — the first transcription downloads it" never reached the surface built to explain it; the note now rides any state, and the renderer prints it on the active row too. Duplicate provider names in config passed through, probing the provider twice and printing it twice on the report — refused now, as loudly as an unknown name. The extract dispatch handled ProviderInputError and ScannedDocumentError but not ProviderUnavailableError, so a contract-honoring extract provider that failed at call time got error plus a filed issue where the transcribe path re-parks waiting. It re-parks too. Co-Authored-By: Claude Fable 5 --- design/ingestion-pipeline.md | 9 ++++++++- src/dex_engine/capabilities/__init__.py | 25 +++++++++++++++++-------- src/dex_engine/drivers/file.py | 11 ++++++++++- src/dex_engine/pipeline/types.py | 6 ++++-- src/dex_engine/render/surfaces.py | 9 +++++---- tests/capabilities/test_registry.py | 23 +++++++++++++++++++++++ tests/drivers/test_file.py | 16 +++++++++++++++- 7 files changed, 82 insertions(+), 17 deletions(-) diff --git a/design/ingestion-pipeline.md b/design/ingestion-pipeline.md index 210ad89..fb21e4f 100644 --- a/design/ingestion-pipeline.md +++ b/design/ingestion-pipeline.md @@ -551,7 +551,9 @@ bad-input cases (corrupt audio, a container carrying no audio stream at all, malformed file) → the run layer maps it `manual`; **`ProviderUnavailableError`** for call-time availability failures (API 5xx/429, missing binary, model-download failure) → the job -**re-parks `waiting`** with the reason, never burning blocked attempts; an +**re-parks `waiting`** with the reason, never burning blocked attempts — +for **every** capability, extract included (`waiting` + `needs: extract`), +not transcribe alone; an uncaught crash is an engine bug and takes the `error` path (issue filed, retry on new engine). The availability seam is **per-format** — `available(need, format)` — so a PDF wait never wakes for a CSV-only @@ -582,6 +584,11 @@ round-trip depends on the frontmatter pointer existing, show notes or not. **Capability report** (a render surface): each capability, active provider, dormant upgrades and what they'd need — `transcribe: whisper-local (active) · whisper-api available — set OPENAI_API_KEY`. +A note rides **any** state, active included: an ok-with-caveat availability +("model not cached — the first transcription downloads it") is exactly what +this surface exists to explain. Provider order is named once per capability +— a repeated name is refused as loudly as an unknown one, since it would +probe the provider twice and print it twice here. Discoverable, never nagging. This is how a free-floor instance learns what a key would buy. diff --git a/src/dex_engine/capabilities/__init__.py b/src/dex_engine/capabilities/__init__.py index 61ad9fa..55472d6 100644 --- a/src/dex_engine/capabilities/__init__.py +++ b/src/dex_engine/capabilities/__init__.py @@ -206,6 +206,14 @@ def _ordered(need: Need, providers: dict[str, _P], config: Config) -> tuple[_P, if unknown: known = ", ".join(sorted(providers)) or "none" raise ValueError(f"providers.{need.value}: unknown provider(s) {unknown} — known: {known}") + repeated = sorted({name for name in configured if configured.count(name) > 1}) + if repeated: + # Silently tolerated, a repeat probes the provider twice and prints + # it twice on the report — config typos fail loudly, like unknown + # names, rather than producing a quietly odd registry. + raise ValueError( + f"providers.{need.value}: duplicate provider(s) {repeated} — name each one once" + ) defaults = [name for name in DEFAULT_PROVIDER_ORDER[need] if name not in configured] return tuple(providers[name] for name in (*configured, *defaults)) @@ -231,15 +239,16 @@ def _capability_row( active_seen = False for provider in mechanical: availability = provider.available() - if availability.ok and not active_seen: - providers.append({"name": provider.name, "state": "active"}) + row = {"name": provider.name, "state": "unavailable"} + if availability.ok: + row["state"] = "active" if not active_seen else "available" active_seen = True - elif availability.ok: - providers.append({"name": provider.name, "state": "available"}) - else: - providers.append( - {"name": provider.name, "state": "unavailable", "note": availability.reason} - ) + # An ok-with-caveat reason ("model not cached — the first + # transcription downloads it") is exactly what the report exists to + # explain; the surface takes a note on any state. + if availability.reason: + row["note"] = availability.reason + providers.append(row) if floor is not None: # The floor is active only when nothing mechanical outranks it. state = "available" if active_seen else "active" diff --git a/src/dex_engine/drivers/file.py b/src/dex_engine/drivers/file.py index b67051c..c7efe54 100644 --- a/src/dex_engine/drivers/file.py +++ b/src/dex_engine/drivers/file.py @@ -8,7 +8,8 @@ the first available mechanical extractor for the format. No provider for the format → ``waiting`` + ``needs: extract`` with the -registry's stated reason. A scanned/image-only document → ``waiting`` + +registry's stated reason — and so does a provider that reported available +and then failed at call time. A scanned/image-only document → ``waiting`` + ``needs: ocr``. Embedded assets ride the Result for the run layer to write under the media caps, ledgered ``via: extract-asset`` — this driver, like every driver, never touches the ledger or the disk outputs. @@ -26,6 +27,7 @@ from dex_engine.capabilities import Capabilities from dex_engine.pipeline.classify import ( + ProviderUnavailableError, ScannedDocumentError, classify_connection, classify_http, @@ -178,6 +180,13 @@ def _extract(self, data: bytes, fmt: Format, name: str | None) -> Result: extraction = extractor.extract(data, fmt) except ScannedDocumentError as e: return Result(status=Status.WAITING, meta={}, needs=Need.OCR, reason=str(e)) + except ProviderUnavailableError as e: + # A provider that reported available() and then failed at call + # time: the capability is, in truth, not available. The same + # re-park the transcribe drain gives it — waiting has no + # escalation clock — never error + a filed issue about the + # world's weather. + return Result(status=Status.WAITING, meta={}, needs=Need.EXTRACT, reason=str(e)) meta: dict[str, str | int | None] = { "title": name, "format": fmt.value, diff --git a/src/dex_engine/pipeline/types.py b/src/dex_engine/pipeline/types.py index 3d2055f..548a949 100644 --- a/src/dex_engine/pipeline/types.py +++ b/src/dex_engine/pipeline/types.py @@ -417,8 +417,10 @@ def extract(self, data: bytes, fmt: Format) -> Extraction: """Extract markdown (and embedded assets, as bytes) from a document. Raises ``ProviderInputError`` for documents it cannot parse - (→ manual) and ``ScannedDocumentError`` for image-only documents - (→ the OCR path: ``waiting`` + ``needs: ocr``). + (→ manual), ``ScannedDocumentError`` for image-only documents + (→ the OCR path: ``waiting`` + ``needs: ocr``), and + ``ProviderUnavailableError`` for capability-level failures + discovered at call time (→ the job re-parks ``waiting``). """ ... diff --git a/src/dex_engine/render/surfaces.py b/src/dex_engine/render/surfaces.py index 1a94cee..906850a 100644 --- a/src/dex_engine/render/surfaces.py +++ b/src/dex_engine/render/surfaces.py @@ -459,9 +459,10 @@ def _render_capability_report(payload: Mapping[str, object]) -> str: "providers": [ {"name": str, "state": "active" | "available" | "unavailable", - "note": str}]} # note optional; what a dormant - ] # provider would need - } + "note": str}]} # note optional, on ANY state: + ] # what a dormant provider + } # would need, or an active + # one's caveat """ surface = "capability-report" _check_keys(surface, payload, required=frozenset({"capabilities"})) @@ -494,7 +495,7 @@ def _render_capability_report(payload: Mapping[str, object]) -> str: _fail(surface, f"{pwhere}state must be one of {options}, got {state!r}") note = _str_at(surface, provider, "note", pwhere) if "note" in provider else "" if state == "active": - parts.append(f"{pname} (active)") + parts.append(f"{pname} (active) — {note}" if note else f"{pname} (active)") elif note: parts.append(f"{pname} {state} — {note}") else: diff --git a/tests/capabilities/test_registry.py b/tests/capabilities/test_registry.py index 23e05cf..ed33df2 100644 --- a/tests/capabilities/test_registry.py +++ b/tests/capabilities/test_registry.py @@ -39,6 +39,12 @@ def test_unknown_provider_name_is_loud(self): with pytest.raises(ValueError, match="unknown provider"): Capabilities.build(Config(providers={"extract": ["textract"]})) + def test_duplicate_provider_names_are_loud(self): + # A repeat probes the provider twice and prints it twice on the + # report; a config typo fails like any other, not quietly. + with pytest.raises(ValueError, match="duplicate provider"): + Capabilities.build(Config(providers={"transcribe": ["whisper-api", "whisper-api"]})) + def test_cognitive_is_not_configurable(self): # The floor is ALWAYS last in resolution order — config cannot # move it, so naming it is refused rather than silently reordered. @@ -156,6 +162,23 @@ def test_active_dormant_unavailable_and_floor_states(self): } ] + def test_an_ok_with_caveat_note_survives_to_the_report(self): + # "model not cached — the first transcription downloads it" is + # precisely what the report exists to explain; being available is + # no reason to drop it. + caveat = "model 'medium' is not cached yet — the first transcription downloads it" + c = caps( + transcribers=( + FakeTranscriber("whisper-local", reason=caveat), + FakeTranscriber("whisper-api", reason="keyless local server"), + ) + ) + rows = provider_rows(c.report_payload())["transcribe"] + assert rows[0] == {"name": "whisper-local", "state": "active", "note": caveat} + assert rows[1]["note"] == "keyless local server" + flat = " ".join(surfaces.render("capability-report", c.report_payload()).split()) + assert f"whisper-local (active) — {caveat}" in flat + def test_payload_renders_on_the_surface(self): # The designed example line, verbatim shape: active first, the # unavailable provider with its unmet requirement after the dash. diff --git a/tests/drivers/test_file.py b/tests/drivers/test_file.py index f6e3bda..6b86c33 100644 --- a/tests/drivers/test_file.py +++ b/tests/drivers/test_file.py @@ -5,7 +5,7 @@ from dex_engine.capabilities import Capabilities from dex_engine.drivers.file import FileDriver from dex_engine.drivers.transport import HttpResponse -from dex_engine.pipeline.classify import ProviderInputError +from dex_engine.pipeline.classify import ProviderInputError, ProviderUnavailableError from dex_engine.pipeline.types import Config, Format, Kind, Need, Status from tests.capabilities.conftest import FakeExtractor, fixture_bytes from tests.drivers.conftest import FakeTransport, make_unit, reason_of @@ -267,6 +267,20 @@ def test_scanned_document_takes_the_ocr_path(self, tmp_path): assert result.needs is Need.OCR assert "OCR" in reason_of(result) + def test_a_call_time_availability_failure_re_parks_waiting(self, tmp_path): + # available() said yes and the call said otherwise (a rate-limited + # hosted extractor, a model download that died): the capability is + # in truth unavailable, so the job waits — never error + an issue + # filed about the world's weather, which is the transcribe path's + # treatment of the same exception. + (tmp_path / "doc.pdf").write_bytes(fixture_bytes("paper.pdf")) + flaky = FakeExtractor(raise_=ProviderUnavailableError("extract API returned HTTP 429")) + d = FileDriver(capabilities=caps(flaky), root=tmp_path) + result = d.fetch(make_unit("file:doc.pdf", Kind.FILE)) + assert result.status is Status.WAITING + assert result.needs is Need.EXTRACT + assert "HTTP 429" in reason_of(result) + def test_provider_input_errors_propagate_for_the_run_loop(self, tmp_path): # The driver never swallows bad-input raises: the run loop owns the # ProviderInputError → manual mapping. From 9eefb9d5768c3186fd3093d97721937e456ab90d Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sat, 22 Aug 2026 09:57:28 +0100 Subject: [PATCH 6/8] Pin whisper-api's POST seam against a truncated response MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The multipart POST is the second urllib seam wrapped by the transport normalization; nothing asserted it, so a future edit could drop the guard and send truncated uploads back to the engine-bug path. The socket double now reads the whole request — headers and body — before answering: replying mid-upload closes the socket on unread bytes and the client sees a reset rather than the truncated read under test. Co-Authored-By: Claude Fable 5 --- design/ingestion-pipeline.md | 20 ++++++++++---------- tests/drivers/conftest.py | 26 +++++++++++++++++++++++++- tests/drivers/test_transport.py | 11 +++++++++++ 3 files changed, 46 insertions(+), 11 deletions(-) diff --git a/design/ingestion-pipeline.md b/design/ingestion-pipeline.md index fb21e4f..438120c 100644 --- a/design/ingestion-pipeline.md +++ b/design/ingestion-pipeline.md @@ -547,15 +547,14 @@ mechanism, not a hack. Provider contract: `available()` failures (model missing, broken install) park jobs as `waiting` with the reason — the wait list is normally empty for transcription, not absent. A provider raises `ProviderInputError` for -bad-input cases (corrupt audio, a container carrying no audio stream at -all, malformed file) → the run layer maps it -`manual`; **`ProviderUnavailableError`** for call-time availability -failures (API 5xx/429, missing binary, model-download failure) → the job -**re-parks `waiting`** with the reason, never burning blocked attempts — -for **every** capability, extract included (`waiting` + `needs: extract`), -not transcribe alone; an -uncaught crash is an engine bug and takes the `error` path (issue filed, -retry on new engine). The availability seam is **per-format** — +bad-input cases (corrupt audio, a container carrying no audio stream at all, +malformed file) → the run layer maps it `manual`; +**`ProviderUnavailableError`** for call-time availability failures (API +5xx/429, missing binary, model-download failure) → the job **re-parks +`waiting`** with the reason, never burning blocked attempts — for **every** +capability, extract included (`waiting` + `needs: extract`), not transcribe +alone; an uncaught crash is an engine bug and takes the `error` path (issue +filed, retry on new engine). The availability seam is **per-format** — `available(need, format)` — so a PDF wait never wakes for a CSV-only provider. **Acquisition failures are not provider failures**: a failed audio download (yt-dlp breakage, blocked enclosure GET) classifies through @@ -566,7 +565,8 @@ acquisition failure too**: an empty body, a body short of its declared error page) is `blocked` and never written under the audio name — cached as `.mp3` it decodes as garbage and parks the episode `manual` forever over a fault that has nothing to do with the episode. Bytes that are merely -bad audio still reach the provider, and its `manual` stands. The blocked line keeps +bad audio still reach the provider, and its `manual` stands. The blocked +line keeps `needs: transcribe`, so the retry routes back through the transcribe drain rather than the driver (§3: a typed field, never the reason's wording). Config keys: `transcribe_base_url`, diff --git a/tests/drivers/conftest.py b/tests/drivers/conftest.py index 1a896a6..e6c7ff4 100644 --- a/tests/drivers/conftest.py +++ b/tests/drivers/conftest.py @@ -2,6 +2,7 @@ import contextlib import json +import re import socket import threading from collections.abc import Iterator, Mapping @@ -80,6 +81,29 @@ def fake_transport(): return FakeTransport +def _drain_request(conn: socket.socket) -> None: + """Read the whole request — headers AND body — before answering. + + Answering a POST while the client is still uploading closes the socket + on unread bytes, and the client sees a reset instead of the truncated + read the test is about. + """ + data = b"" + while b"\r\n\r\n" not in data: + chunk = conn.recv(65536) + if not chunk: + return + data += chunk + head, _, body = data.partition(b"\r\n\r\n") + declared = re.search(rb"(?i)content-length:\s*(\d+)", head) + outstanding = int(declared.group(1)) - len(body) if declared else 0 + while outstanding > 0: + chunk = conn.recv(min(outstanding, 65536)) + if not chunk: + return + outstanding -= len(chunk) + + @contextlib.contextmanager def truncating_server( *, body: bytes = b"ID3\x04\x00\x00\x00partial audio", declared: int = 5_000_000 @@ -102,7 +126,7 @@ def serve() -> None: except OSError: return # the listener closed: the context manager is done with conn, contextlib.suppress(OSError): - conn.recv(65536) # drain the request so the close sends FIN, not RST + _drain_request(conn) conn.sendall( b"HTTP/1.1 200 OK\r\nContent-Type: audio/mpeg\r\n" b"Content-Length: %d\r\nConnection: close\r\n\r\n%s" % (declared, body) diff --git a/tests/drivers/test_transport.py b/tests/drivers/test_transport.py index 99aea80..6a6ab54 100644 --- a/tests/drivers/test_transport.py +++ b/tests/drivers/test_transport.py @@ -10,6 +10,7 @@ import pytest from dex_engine.capabilities import Capabilities +from dex_engine.capabilities.transcribe.whisper_api import urllib_multipart_post from dex_engine.drivers.file import FileDriver from dex_engine.drivers.transport import normalize_httplib_errors, urllib_transport from dex_engine.pipeline.classify import classify_connection @@ -40,6 +41,16 @@ def test_an_oserror_passes_through_untouched(self): assert caught.value is original +class TestThroughTheWhisperApiPost: + def test_a_truncated_response_reaches_the_provider_as_an_oserror(self): + # The second urllib seam: whisper-api's multipart POST guards on + # `except OSError` too, and maps it to ProviderUnavailableError — + # the job waits. Only if the httplib failure arrives as an OSError. + with truncating_server() as url, pytest.raises(OSError) as caught: # noqa: PT011 — the OSError shape IS the assertion + urllib_multipart_post(url, api_key=None, fields={}, filename="a.mp3", file_bytes=b"x") + assert "truncated response body" in str(caught.value) + + class TestThroughADriverFetch: def test_truncated_download_is_blocked_never_an_engine_error(self): driver = FileDriver( From ad5b4d3598954c9e95d64773f260f59998396a05 Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sat, 22 Aug 2026 11:40:18 +0100 Subject: [PATCH 7/8] Budget the whisper prompt in tokens and put the names last MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The head-first character budget was plain truncation wearing a docstring: the head already sat at the FRONT of the composed string, so `text[:600]` kept it either way, and the claim that truncating instead "would keep 600 characters of show notes and throw away the two names" was false. Worse, whisper counts its window in TOKENS. 600 characters is ~180 tokens of Cyrillic and ~690 of Chinese, and past 223 tokens whisper keeps only `previous_tokens[-(448 // 2 - 1):]` — it discards the FRONT. For every CJK or Cyrillic item the title was therefore the first thing lost, by the code written to save it, and whisper-api's chunk continuity compounded it. Two changes. The title/show now goes at the END of the prompt, behind the vocabulary: whatever overflows the window is show notes, by construction, in any script. And the budget is spent through a per-character estimate of what whisper's BPE charges, measured against openai/whisper-tiny's tokenizer and rounded up per script family — so a prompt does not overflow in the first place. The tokenizer itself is not loaded: that means a HuggingFace fetch inside a step that primes remote providers needing no local model, to buy an exactness the head's placement already made unnecessary. A mis-estimate now costs vocabulary, not names. A live-marked test checks the estimate against the real tokenizer in nine scripts. Co-Authored-By: Claude Fable 5 --- design/ingestion-pipeline.md | 24 ++- pyproject.toml | 1 + .../capabilities/transcribe/whisper_api.py | 32 ++-- src/dex_engine/pipeline/transcribe.py | 139 +++++++++++++++--- tests/capabilities/test_whisper_api.py | 37 +++-- tests/pipeline/test_transcribe.py | 112 ++++++++++++-- 6 files changed, 283 insertions(+), 62 deletions(-) diff --git a/design/ingestion-pipeline.md b/design/ingestion-pipeline.md index 438120c..9096d70 100644 --- a/design/ingestion-pipeline.md +++ b/design/ingestion-pipeline.md @@ -507,14 +507,22 @@ mechanism, not a hack. `needs: transcribe` with the pointer. - Accuracy: **`initial_prompt` priming** with the item's known vocabulary (video title + description, episode title + show notes) — mechanical, free, - large win on names/jargon. **Whisper keeps a prompt's LAST ~224 tokens and - discards its front**, so the composed prompt is budgeted to ~200 tokens - (600 chars, at the 3 chars/token this jargon-dense text measures): the - title/show head always survives and the VOCABULARY is trimmed from its - tail. whisper-api's chunk-continuity tail shares that one budget rather - than adding to it — otherwise the continuity a later chunk gains would - cost it the two names priming exists to carry. Transcripts are stored - **raw**, stamped + large win on names/jargon. **Whisper keeps a prompt's LAST 223 tokens and + discards its front** (`previous_tokens[-(448 // 2 - 1):]`), so the + title/show is written at the END of the composed prompt and the trimmable + vocabulary in front of it: whatever overflows the window is show notes, + never the two names the priming exists to carry. The window is counted in + TOKENS, which is not a character count in any script but English — 600 + characters is ~180 tokens of Cyrillic and ~690 of Chinese — so the budget + (~200 tokens) is spent through a per-script estimate of what whisper's BPE + will charge, measured against its tokenizer and rounded up per script + family. An estimate, not the tokenizer itself: loading it would put a + HuggingFace fetch in a step that primes remote providers needing no local + model, and placing the head last is what makes the names safe, so the + estimate only decides how much vocabulary rides along. whisper-api's + chunk-continuity tail shares that one budget rather than adding to it, and + trims the priming from its front for the same reason. Transcripts are + stored **raw**, stamped `via`/`model` in frontmatter; corrections live downstream in digest/wiki where judgment already operates — enrichment stays the mechanical record. diff --git a/pyproject.toml b/pyproject.toml index 11cc95c..d9829ad 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -75,6 +75,7 @@ convention = "google" "ANN", # type annotations not required on tests "PLR2004", # magic values are expected in assertions "INP001", # pytest test dirs are not packages + "RUF001", # non-Latin test data — what the prompt budget is about — is not a typo ] [tool.pytest.ini_options] diff --git a/src/dex_engine/capabilities/transcribe/whisper_api.py b/src/dex_engine/capabilities/transcribe/whisper_api.py index 213c753..36f6536 100644 --- a/src/dex_engine/capabilities/transcribe/whisper_api.py +++ b/src/dex_engine/capabilities/transcribe/whisper_api.py @@ -31,7 +31,11 @@ from dex_engine.drivers.transport import normalize_httplib_errors from dex_engine.pipeline.classify import ProviderInputError, ProviderUnavailableError, scrub -from dex_engine.pipeline.transcribe import PROMPT_MAX_CHARS +from dex_engine.pipeline.transcribe import ( + PROMPT_MAX_TOKENS, + estimated_tokens, + keep_last_tokens, +) from dex_engine.pipeline.types import Availability __all__ = [ @@ -51,10 +55,13 @@ # cap at any sane audio bitrate. CHUNK_SECONDS = 1200 -# The continuity tail's share of PROMPT_MAX_CHARS: ~65 tokens of the +# The continuity tail's share of PROMPT_MAX_TOKENS: 50 tokens of the # running transcript is ample for carrying word boundaries and style into -# the next chunk, and it leaves two thirds of the window to the priming. -_CONTINUITY_TAIL_CHARS = 200 +# the next chunk, and it leaves three quarters of the window to the priming. +_CONTINUITY_TAIL_TOKENS = 50 + +# The "\n" joining the two halves. +_NEWLINE_TOKENS = 1 _HTTP_OK_FLOOR, _HTTP_OK_CEILING = 200, 300 _HTTP_BAD_REQUEST = 400 @@ -314,16 +321,19 @@ def _transcribe_chunk(self, chunk: Path, prompt: str) -> str: def _chunk_prompt(initial_prompt: str, previous: list[str]) -> str: """Vocabulary priming plus the running transcript's tail (chunk continuity). - Both halves live inside one ``PROMPT_MAX_CHARS`` budget, because + Both halves live inside one ``PROMPT_MAX_TOKENS`` budget, because whisper reads a prompt from its END: appending an unbudgeted tail - would push the priming's head — the title and show — out of the - window it exists to occupy. The tail comes off the transcript's end, - the priming off its own end, so the two names always survive. + would push the priming out of the window it exists to occupy. The + continuity tail comes off the transcript's end, and the priming is + trimmed from its FRONT — the title and show are the last thing in it + (:func:`_prompt`), so trimming the front spends show notes to keep + them. """ if not previous: return initial_prompt - tail = " ".join(previous)[-_CONTINUITY_TAIL_CHARS:].strip() + tail = keep_last_tokens(" ".join(previous), _CONTINUITY_TAIL_TOKENS).strip() if not tail: return initial_prompt - head = initial_prompt[: max(PROMPT_MAX_CHARS - len(tail) - 1, 0)].rstrip() - return f"{head}\n{tail}".strip() + room = PROMPT_MAX_TOKENS - estimated_tokens(tail) - _NEWLINE_TOKENS + priming = keep_last_tokens(initial_prompt, max(room, 0)).lstrip() + return f"{priming}\n{tail}".strip() diff --git a/src/dex_engine/pipeline/transcribe.py b/src/dex_engine/pipeline/transcribe.py index e7e0fbb..2d4f4ad 100644 --- a/src/dex_engine/pipeline/transcribe.py +++ b/src/dex_engine/pipeline/transcribe.py @@ -15,6 +15,7 @@ import contextlib import json +import unicodedata from collections.abc import Callable from dataclasses import dataclass from pathlib import Path @@ -29,13 +30,17 @@ from .types import LedgerEntry, Status __all__ = [ - "PROMPT_MAX_CHARS", + "HEAD_MAX_TOKENS", + "PROMPT_MAX_TOKENS", "TRANSCRIBE_RUN_CAP", "Acquired", "DownloadAudio", "YoutubeAudio", "acquire_podcast_audio", "acquire_youtube_audio", + "estimated_tokens", + "keep_first_tokens", + "keep_last_tokens", "podcast_body", "read_enrichment", "youtube_body", @@ -46,13 +51,19 @@ # never monopolize a machine. `enrich transcribe --limit` overrides. TRANSCRIBE_RUN_CAP = 10 -# Whisper's prompt window is ~224 tokens and it keeps the LAST of them: -# an overlong prompt loses its FRONT, which is exactly the title/show the -# priming exists to carry. So the whole composed prompt — vocabulary here -# plus whisper-api's continuity tail — is budgeted to ~200 tokens. Jargon -# tokenizes badly: 800 chars of this text measured ~270 tokens (≈2.96 -# chars/token), so 3 chars/token is the conservative rate and 600 the cap. -PROMPT_MAX_CHARS = 600 +# Whisper's prompt window is 223 tokens (`previous_tokens[-(448 // 2 - 1):]`) +# and it keeps the LAST of them, so anything overlong loses its FRONT. The +# composed prompt therefore ends with the title/show and begins with the +# trimmable vocabulary: whatever whisper drops, it drops from the show notes. +# The budgets below decide how much vocabulary survives, never whether the +# names do — which is why an estimated token count is enough and no +# tokenizer is loaded here (see _token_cost). 200 leaves the window room +# for whisper's own framing tokens. +PROMPT_MAX_TOKENS = 200 + +# The title/show anchor's own ceiling. A real one costs ~20 tokens; the cap +# exists so a pathological title cannot eat the whole prompt. +HEAD_MAX_TOKENS = 60 _SEPARATOR = " — " _AUDIO_EXT_DEFAULT = "mp3" @@ -312,13 +323,16 @@ def _audio_ext(url: str) -> str: def _prompt(title: str | None, show: str | None, vocabulary: str) -> str: - """Known-vocabulary priming, single-line, budgeted head-first. + """Known-vocabulary priming, single-line, with the names LAST. - The head (title, then channel/show) always survives: whisper reads a - prompt from its END, so the VOCABULARY is what gets truncated — from - its own tail — to fit :data:`PROMPT_MAX_CHARS`. Truncating the - composed string instead would keep 600 characters of show notes and - throw away the two names the priming was built for. + Whisper reads a prompt from its END and discards the front of anything + that overflows its 223-token window, so the head — title, then + channel/show — is written at the END, behind the vocabulary. Two things + then have to go wrong before the names are lost rather than one: the + prompt must overflow the window AND the overflow must reach past the + show notes. The vocabulary is trimmed from its own tail (show notes + open with the description and close with sponsor links), the head from + its own tail, both against a token estimate. Args: title: The episode/video title. @@ -326,23 +340,106 @@ def _prompt(title: str | None, show: str | None, vocabulary: str) -> str: vocabulary: Show notes or description — the trimmable part. Returns: - The priming text, at most :data:`PROMPT_MAX_CHARS` characters. + The priming text, an estimated :data:`PROMPT_MAX_TOKENS` at most. """ - head = _SEPARATOR.join(_flat(part) for part in (title, show) if part and part.strip()) + parts = (_flat(part) for part in (title, show) if part and part.strip()) + head = keep_first_tokens(_SEPARATOR.join(parts), HEAD_MAX_TOKENS).strip() vocab = _flat(vocabulary) if not head: - return vocab[:PROMPT_MAX_CHARS] - head = head[:PROMPT_MAX_CHARS] - room = PROMPT_MAX_CHARS - len(head) - len(_SEPARATOR) - if not vocab or room <= 0: + return keep_first_tokens(vocab, PROMPT_MAX_TOKENS).strip() + room = PROMPT_MAX_TOKENS - estimated_tokens(head) - estimated_tokens(_SEPARATOR) + vocab = keep_first_tokens(vocab, room).strip() if room > 0 else "" + if not vocab: return head - return f"{head}{_SEPARATOR}{vocab[:room]}" + return f"{vocab}{_SEPARATOR}{head}" def _flat(text: str) -> str: return " ".join(text.split()) +# --------------------------------------------------------------------------- +# The token budget. Whisper counts its prompt window in TOKENS, so a +# character cap is a different quantity in every script: 600 characters is +# ~180 tokens of Cyrillic but ~690 of Chinese, and the overflow is taken off +# the front. Costs below are whisper-BPE tokens per character, measured +# against openai/whisper-tiny's tokenizer over natural-language samples +# (English prose 0.22, English jargon 0.37, Cyrillic 0.29, Greek 0.41, +# Hebrew 0.56, Arabic 0.57, Hangul 0.76, Thai 0.96, kana/Han 0.80-1.15, +# Devanagari 1.12, emoji and mathematical symbols 1.7-2.2) and rounded up +# per family. +# +# An estimate, not that tokenizer: loading it means a HuggingFace fetch of +# whisper-tiny's tokenizer.json on first use, in a step that composes +# prompts for remote providers that need no local model and may not be +# running whisper's tokenizer at all. It buys exactness the placement of the +# head has already made unnecessary — a mis-estimate here costs vocabulary. +# --------------------------------------------------------------------------- + + +def _token_cost(char: str) -> float: + if char < "Ā": # ASCII and Latin-1 + return 0.4 + if unicodedata.category(char)[0] == "S": # emoji, mathematical, other symbols + return 2.5 + if char < "֐": # Latin extended, IPA, Greek, Cyrillic, Armenian + return 0.6 + if char < "ࠀ": # Hebrew, Arabic, Syriac, Thaana + return 0.8 + return 1.5 # Indic, Thai, Hangul, kana, Han, and every other script + + +def estimated_tokens(text: str) -> float: + """Estimate what whisper's tokenizer will make of ``text``. + + Args: + text: Any prompt fragment. + + Returns: + The estimated token count — deliberately generous per character. + """ + return sum(_token_cost(char) for char in text) + + +def keep_first_tokens(text: str, budget: float) -> str: + """The longest PREFIX of ``text`` estimated to fit ``budget`` tokens. + + Args: + text: The text to trim. + budget: The token allowance. + + Returns: + ``text`` itself when it already fits, else its trimmed front. + """ + spent = 0.0 + for index, char in enumerate(text): + spent += _token_cost(char) + if spent > budget: + return text[:index] + return text + + +def keep_last_tokens(text: str, budget: float) -> str: + """The longest SUFFIX of ``text`` estimated to fit ``budget`` tokens. + + The suffix is what survives whisper's own truncation, so this is the + trim that composes with it rather than against it. + + Args: + text: The text to trim. + budget: The token allowance. + + Returns: + ``text`` itself when it already fits, else its trimmed tail. + """ + spent = 0.0 + for index, char in enumerate(reversed(text)): + spent += _token_cost(char) + if spent > budget: + return text[len(text) - index :] + return text + + # --------------------------------------------------------------------------- # Transcript bodies. Raw transcripts, stamped via/model by the caller; # corrections live downstream in digest/wiki where judgment operates. diff --git a/tests/capabilities/test_whisper_api.py b/tests/capabilities/test_whisper_api.py index 9f8091b..47c1c46 100644 --- a/tests/capabilities/test_whisper_api.py +++ b/tests/capabilities/test_whisper_api.py @@ -12,7 +12,13 @@ _chunk_prompt, ) from dex_engine.pipeline.classify import ProviderInputError, ProviderUnavailableError -from dex_engine.pipeline.transcribe import PROMPT_MAX_CHARS, _prompt +from dex_engine.pipeline.transcribe import _prompt, estimated_tokens + +# Whisper's real prompt window — `previous_tokens[-(448 // 2 - 1):]`. A +# literal on purpose: the budget is only honest checked against the window. +WHISPER_WINDOW_TOKENS = 223 + +CJK_NOTES = "这是一个关于软件工程的播客节目，今天我们讨论账本与工作队列的设计。" * 40 class FakePost: @@ -226,20 +232,33 @@ def test_empty_transcripts_are_bad_input(self, tmp_path): class TestChunkPrompt: + TITLE = "Ledgers as Work Queues" + SHOW = "Engineering Distilled" + HEAD = f"{TITLE} — {SHOW}" + def test_first_chunk_is_the_initial_prompt_alone(self): assert _chunk_prompt("vocab", []) == "vocab" def test_tail_is_bounded(self): prompt = _chunk_prompt("v", ["word " * 500]) - assert len(prompt) <= 202 # initial + newline + bounded tail + assert estimated_tokens(prompt) <= WHISPER_WINDOW_TOKENS def test_the_continuity_tail_never_pushes_the_title_out_of_the_window(self): - # Whisper keeps the LAST ~224 tokens: an unbudgeted tail appended - # to a full priming string would discard the head the priming - # exists to carry. Both halves share one budget. - priming = _prompt("Ledgers as Work Queues", "Engineering Distilled", "jargon " * 400) - assert len(priming) == PROMPT_MAX_CHARS # the worst case + # Whisper keeps the LAST 223 tokens: an unbudgeted tail appended to + # a full priming string would discard the head the priming exists to + # carry. Both halves share one budget, and the priming is trimmed + # from its front — where its expendable vocabulary lives. + priming = _prompt(self.TITLE, self.SHOW, "jargon " * 400) prompt = _chunk_prompt(priming, ["transcribed words " * 200]) - assert prompt.startswith("Ledgers as Work Queues — Engineering Distilled") + assert self.HEAD in prompt assert prompt.endswith("transcribed words") - assert len(prompt) <= PROMPT_MAX_CHARS + assert estimated_tokens(prompt) <= WHISPER_WINDOW_TOKENS + + def test_a_non_latin_chunk_prompt_keeps_the_title_inside_the_window(self): + # The case a character budget got backwards: 200 characters of + # Chinese continuity is already 230 tokens — a whole window — so an + # unbudgeted tail would leave nothing of the priming at all. + priming = _prompt(self.TITLE, self.SHOW, CJK_NOTES) + prompt = _chunk_prompt(priming, ["这是转录的文本内容。" * 100]) + assert self.HEAD in prompt + assert estimated_tokens(prompt) <= WHISPER_WINDOW_TOKENS diff --git a/tests/pipeline/test_transcribe.py b/tests/pipeline/test_transcribe.py index 627b52f..9e4ed12 100644 --- a/tests/pipeline/test_transcribe.py +++ b/tests/pipeline/test_transcribe.py @@ -22,7 +22,7 @@ from dex_engine.pipeline.registry import build_drivers from dex_engine.pipeline.run import _Drain from dex_engine.pipeline.transcribe import ( - PROMPT_MAX_CHARS, + HEAD_MAX_TOKENS, TRANSCRIBE_RUN_CAP, Acquired, YoutubeAudio, @@ -30,6 +30,8 @@ _download_enclosure, _prompt, acquire_youtube_audio, + estimated_tokens, + keep_last_tokens, read_enrichment, ) from dex_engine.pipeline.types import ( @@ -912,33 +914,80 @@ def test_missing_cache_dir_is_simply_empty(self, tmp_path): assert _cached_audio(tmp_path / "audio", "abc") is None +# Whisper's real prompt window: `previous_tokens[-(448 // 2 - 1):]` in +# faster_whisper/transcribe.py, and the same model server-side. A literal, +# deliberately — the budget is only honest if it is checked against the +# window rather than against itself. +WHISPER_WINDOW_TOKENS = 223 + +# Enough of each script to blow any budget, so the trim is always exercised. +CJK_NOTES = "这是一个关于软件工程的播客节目，今天我们讨论账本与工作队列的设计。" * 40 +CYRILLIC_NOTES = "Это подкаст о разработке программного обеспечения и практиках. " * 40 + + class TestPromptBudget: - """Whisper reads a prompt from its END — the head must never be the casualty.""" + """Whisper reads a prompt from its END, and counts the window in TOKENS.""" TITLE = "Ledgers as Work Queues" SHOW = "Engineering Distilled" HEAD = f"{TITLE} — {SHOW}" - def test_the_head_survives_a_vocabulary_far_past_the_budget(self): + def test_the_names_are_the_last_thing_in_the_prompt(self): + # Not merely present: LAST. Whatever whisper drops for being over + # the window, it drops from the front, so the front is where the + # expendable vocabulary belongs. prompt = _prompt(self.TITLE, self.SHOW, "jargon " * 400) - assert prompt.startswith(self.HEAD) - assert len(prompt) == PROMPT_MAX_CHARS + assert prompt.endswith(self.HEAD) + assert prompt.startswith("jargon") def test_the_vocabulary_is_trimmed_from_its_tail(self): prompt = _prompt(self.TITLE, self.SHOW, "anydoc CTranslate2 " + "x" * 2000) - assert prompt.startswith(f"{self.HEAD} — anydoc CTranslate2 ") - assert not prompt.endswith("x" * 2000) + assert prompt.startswith("anydoc CTranslate2 ") + assert prompt.endswith(self.HEAD) + assert "x" * 2000 not in prompt def test_a_short_prompt_is_untouched(self): - assert _prompt(self.TITLE, self.SHOW, "notes") == f"{self.HEAD} — notes" + assert _prompt(self.TITLE, self.SHOW, "notes") == f"notes — {self.HEAD}" assert _prompt(self.TITLE, None, "") == self.TITLE + def test_a_latin_prompt_fits_whispers_window(self): + prompt = _prompt(self.TITLE, self.SHOW, "jargon " * 400) + assert estimated_tokens(prompt) <= WHISPER_WINDOW_TOKENS + # And uses most of it: a budget that fits by being tiny is no budget. + assert estimated_tokens(prompt) > WHISPER_WINDOW_TOKENS * 0.7 + + def test_a_non_latin_prompt_fits_whispers_window_too(self): + # The character cap this replaced was ~180 tokens of Cyrillic but + # ~690 of Chinese — three windows' worth, and whisper takes the + # overflow off the FRONT, where the title used to sit. + for notes in (CJK_NOTES, CYRILLIC_NOTES): + prompt = _prompt(self.TITLE, self.SHOW, notes) + assert estimated_tokens(prompt) <= WHISPER_WINDOW_TOKENS + assert prompt.endswith(self.HEAD) + + def test_the_budget_is_tokens_not_characters(self): + latin = _prompt(self.TITLE, self.SHOW, "jargon " * 400).removesuffix(self.HEAD) + cjk = _prompt(self.TITLE, self.SHOW, CJK_NOTES).removesuffix(self.HEAD) + # Same token allowance spends very different numbers of characters — + # a character cap would make these two lengths equal. + assert len(latin) > 3 * len(cjk) + assert abs(estimated_tokens(latin) - estimated_tokens(cjk)) < 5 + + def test_emoji_dense_notes_are_charged_for_what_they_cost(self): + # A show-notes header of emoji tokenizes at ~2 tokens per character. + prompt = _prompt(self.TITLE, self.SHOW, "🚀🎧📚✨🔥" * 200) + assert estimated_tokens(prompt) <= WHISPER_WINDOW_TOKENS + assert prompt.endswith(self.HEAD) + def test_vocabulary_alone_is_still_bounded(self): - assert len(_prompt(None, None, "y " * 1000)) == PROMPT_MAX_CHARS + assert estimated_tokens(_prompt(None, None, "y " * 1000)) <= WHISPER_WINDOW_TOKENS + assert estimated_tokens(_prompt(None, None, CJK_NOTES)) <= WHISPER_WINDOW_TOKENS - def test_an_overlong_head_keeps_its_front_and_drops_the_vocabulary(self): + def test_an_overlong_head_is_capped_rather_than_eating_the_prompt(self): prompt = _prompt("T" * 900, self.SHOW, "notes") - assert prompt == "T" * PROMPT_MAX_CHARS + assert estimated_tokens(prompt) <= WHISPER_WINDOW_TOKENS + assert prompt.startswith("notes — ") + assert estimated_tokens(prompt.removeprefix("notes — ")) <= HEAD_MAX_TOKENS def test_the_acquisition_path_composes_within_the_budget(self, instance, tmp_path): # The real composition site, not just the helper: a talkative @@ -958,8 +1007,45 @@ def download(_url, cache_dir, stem) -> YoutubeAudio: acquired = acquire_youtube_audio(entry, tmp_path, download) assert isinstance(acquired, Acquired) - assert acquired.prompt.startswith("Ledgers at Scale — Engineering Distilled") - assert len(acquired.prompt) <= PROMPT_MAX_CHARS + assert acquired.prompt.endswith("Ledgers at Scale — Engineering Distilled") + assert estimated_tokens(acquired.prompt) <= WHISPER_WINDOW_TOKENS + + +class TestTokenEstimate: + def test_keeping_the_tail_keeps_the_end(self): + assert keep_last_tokens("abcdefghij", 1) == "ij" + assert keep_last_tokens("abc", 100) == "abc" + assert keep_last_tokens("", 10) == "" + + def test_non_latin_characters_cost_more_than_ascii(self): + assert estimated_tokens("这是一个") > estimated_tokens("abcd") + assert estimated_tokens("абвг") > estimated_tokens("abcd") + + @pytest.mark.live + def test_the_estimate_covers_what_whispers_tokenizer_actually_does(self): + # Opt-in: needs openai/whisper-tiny's tokenizer.json from HuggingFace. + # Drift check on the cost table — a prompt this estimator passes must + # really fit whisper's window, in every script, or the head it puts + # last is the only thing that survives the overflow. + from huggingface_hub import hf_hub_download # noqa: PLC0415 — live-only dep + from tokenizers import Tokenizer # noqa: PLC0415 — live-only dep + + tokenizer = Tokenizer.from_file(hf_hub_download("openai/whisper-tiny", "tokenizer.json")) + samples = { + "english": "anydoc CTranslate2 ledger idempotent frontmatter yt-dlp " * 40, + "chinese": CJK_NOTES, + "cyrillic": CYRILLIC_NOTES, + "japanese": "これはソフトウェアエンジニアリングに関するポッドキャストです。" * 40, + "korean": "이것은 소프트웨어 엔지니어링에 관한 팟캐스트입니다 그리고 " * 40, + "hindi": "यह सॉफ्टवेयर इंजीनियरिंग के बारे में एक पॉडकास्ट है। " * 40, + "thai": "นี่คือพอดแคสต์เกี่ยวกับวิศวกรรมซอฟต์แวร์และแนวปฏิบัติ " * 40, + "arabic": "هذه حلقة بودكاست عن هندسة البرمجيات وممارسات الهندسة اليومية. " * 40, + "emoji": "🚀🎧📚✨🔥" * 200, + } + for script, notes in samples.items(): + prompt = _prompt("Ledgers as Work Queues", "Engineering Distilled", notes) + real = len(tokenizer.encode(prompt, add_special_tokens=False).ids) + assert real <= WHISPER_WINDOW_TOKENS, f"{script}: {real} tokens" class TestReadEnrichment: From 3da62d1cb2b136445f4aa96457a05cb18691b204 Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sat, 22 Aug 2026 11:44:13 +0100 Subject: [PATCH 8/8] Hold the status when an HTTP error body reads short MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three sites of one defect. The transport's error-body read already suppressed `http.client.HTTPException` and nothing pinned it: with the suppress narrowed to OSError the whole suite stayed green, while a real socket serving a truncated 404 turned the item's `dead` into `blocked` — the read failure escaping through `normalize_httplib_errors` as a connection error. That case is now a socket test. whisper-api's multipart POST, twenty lines from its sibling, had no such guard at all: a 400 whose body stopped short of its Content-Length raised out of `e.read()`, and `_transcribe_chunk`'s `except OSError` filed it as ProviderUnavailableError. The episode then waited forever on an endpoint that was answering fine, instead of parking manual under the escalation clock the bad audio had earned. The two seams are symmetric now, both pinned against a socket that hangs up mid-body. dex-inbox's two urllib sites had neither the guard nor a caller that could report what escaped: `main` states OSError, ValueError and RuntimeError, and an IncompleteRead is none of them, so a GitHub asset download cut mid-body reached the operator as a traceback. Both sites go through the seam's guard. Co-Authored-By: Claude Fable 5 --- design/ingestion-pipeline.md | 10 +++++ .../capabilities/transcribe/whisper_api.py | 16 ++++++- src/dex_engine/drivers/transport.py | 4 ++ src/dex_engine/inbox.py | 44 ++++++++++++------- tests/drivers/conftest.py | 32 +++++++++++--- tests/drivers/test_transport.py | 33 +++++++++++++- tests/test_inbox.py | 32 ++++++++++++++ 7 files changed, 145 insertions(+), 26 deletions(-) diff --git a/design/ingestion-pipeline.md b/design/ingestion-pipeline.md index 9096d70..09b05d2 100644 --- a/design/ingestion-pipeline.md +++ b/design/ingestion-pipeline.md @@ -419,6 +419,16 @@ independently is seven chances to reintroduce it): `HTTPException`, not an `OSError`, and would otherwise slip past every `except OSError` guard and land as `error` + a filed issue. A truncated read is a connection failure: **`blocked`, retried**, never an engine bug. + The same guard wraps `dex-inbox`'s two urllib sites — an asset download is + the same large-read shape — so a truncated read there is a stated + `dex-inbox:` failure, not a traceback. +- **A truncated ERROR body never costs the status code.** The body of a + 4xx/5xx is only detail; the status is the finding. Both HTTP seams (the + transport, whisper-api's multipart POST) drop a failed error-body read and + return the status, because losing it inverts the verdict: a `404`'s `dead` + would arrive as `blocked`, and whisper-api's `400` — the audio is bad, + `manual` under the escalation clock — would arrive as an unreachable + endpoint, `waiting` with no clock at all. - **Media URLs are hashed un-canonicalized** — signed query params ARE the resource. Side effect, accepted: an expiring signed URL re-mints a fresh entry per parent rerun, and the stale one retires through normal blocked diff --git a/src/dex_engine/capabilities/transcribe/whisper_api.py b/src/dex_engine/capabilities/transcribe/whisper_api.py index 36f6536..3aa4be3 100644 --- a/src/dex_engine/capabilities/transcribe/whisper_api.py +++ b/src/dex_engine/capabilities/transcribe/whisper_api.py @@ -17,6 +17,8 @@ one uniform path. """ +import contextlib +import http.client import json import os import shutil @@ -132,7 +134,19 @@ def urllib_multipart_post( with urllib.request.urlopen(request, timeout=_TIMEOUT_SECONDS) as response: # noqa: S310 return response.status, response.read() except urllib.error.HTTPError as e: - return e.code, e.read() + detail = b"" + # As at the transport seam: the status is the whole finding and + # the body is only detail, so a body that stops short of its + # Content-Length must not cost the code. Losing it here inverts + # the provider's verdict — 400 says the AUDIO is bad (manual, + # with an escalation clock), while the read failure escaping as + # a connection error says the ENDPOINT is down (waiting, with + # none). ``HTTPException`` is named because it is not an + # ``OSError``: an ``IncompleteRead`` would otherwise leave here + # through ``normalize_httplib_errors`` as exactly that. + with contextlib.suppress(OSError, ValueError, http.client.HTTPException): + detail = e.read() + return e.code, detail def run_ffmpeg(args: list[str]) -> None: diff --git a/src/dex_engine/drivers/transport.py b/src/dex_engine/drivers/transport.py index 59c07fd..558d73c 100644 --- a/src/dex_engine/drivers/transport.py +++ b/src/dex_engine/drivers/transport.py @@ -137,6 +137,10 @@ def urllib_transport(url: str, *, method: str = "GET") -> HttpResponse: body = b"" # A truncated error page still classifies by its status: the # partial body is only detail, so the read failure is dropped. + # ``HTTPException`` is named because it is not an ``OSError`` — + # an ``IncompleteRead`` here would otherwise leave through + # ``normalize_httplib_errors`` as a connection failure, turning + # a 404's `dead` into `blocked`. with contextlib.suppress(OSError, ValueError, http.client.HTTPException): body = e.read() return HttpResponse( diff --git a/src/dex_engine/inbox.py b/src/dex_engine/inbox.py index f305576..811a049 100644 --- a/src/dex_engine/inbox.py +++ b/src/dex_engine/inbox.py @@ -52,6 +52,7 @@ from email.message import Message as HTTPMessage from . import atomic +from .drivers.transport import normalize_httplib_errors from .pipeline.capture import parse_capture from .pipeline.detect import sniff_format from .pipeline.types import Instance @@ -151,12 +152,17 @@ def _api_request( if body: headers["Content-Type"] = "application/json" request = urllib.request.Request(url, data=body, method=method, headers=headers) # noqa: S310 — https-only API root - try: - with urllib.request.urlopen(request, timeout=60) as response: # noqa: S310 - raw = response.read() - return response.status, (json.loads(raw) if raw else None) - except urllib.error.HTTPError as e: - return e.code, None + # The same seam guard the drivers' transport uses: `http.client`'s + # protocol failures are not OSErrors, and main()'s wrapper catches + # OSError — an unguarded IncompleteRead reaches the operator as a + # traceback instead of a `dex-inbox: ...` line. + with normalize_httplib_errors(): + try: + with urllib.request.urlopen(request, timeout=60) as response: # noqa: S310 + raw = response.read() + return response.status, (json.loads(raw) if raw else None) + except urllib.error.HTTPError as e: + return e.code, None class _NoRedirect(urllib.request.HTTPRedirectHandler): @@ -187,18 +193,22 @@ def _download_asset(asset_url: str, token: str) -> bytes: }, ) opener = urllib.request.build_opener(_NoRedirect) - try: - with opener.open(request, timeout=120) as response: - return response.read() - except urllib.error.HTTPError as e: - target = e.headers.get("Location") - if e.code in (301, 302, 303, 307, 308) and target: - location = urllib.request.Request( # noqa: S310 — the API's signed redirect target - target, headers={"User-Agent": UA} - ) - with urllib.request.urlopen(location, timeout=600) as response: # noqa: S310 + # An asset read is the large-download shape: the server hanging up + # mid-body raises IncompleteRead, which is not an OSError and so would + # escape main()'s wrapper as a traceback rather than a stated failure. + with normalize_httplib_errors(): + try: + with opener.open(request, timeout=120) as response: return response.read() - raise + except urllib.error.HTTPError as e: + target = e.headers.get("Location") + if e.code in (301, 302, 303, 307, 308) and target: + location = urllib.request.Request( # noqa: S310 — the API's signed redirect target + target, headers={"User-Agent": UA} + ) + with urllib.request.urlopen(location, timeout=600) as response: # noqa: S310 + return response.read() + raise def default_seams() -> GithubSeams: diff --git a/tests/drivers/conftest.py b/tests/drivers/conftest.py index e6c7ff4..d0b6165 100644 --- a/tests/drivers/conftest.py +++ b/tests/drivers/conftest.py @@ -106,31 +106,51 @@ def _drain_request(conn: socket.socket) -> None: @contextlib.contextmanager def truncating_server( - *, body: bytes = b"ID3\x04\x00\x00\x00partial audio", declared: int = 5_000_000 + *, + body: bytes = b"ID3\x04\x00\x00\x00partial audio", + declared: int = 5_000_000, + status: bytes = b"200 OK", + redirect_first: bool = False, ) -> Iterator[str]: - """Serve one 200 whose Content-Length far exceeds the bytes sent, then hang up. + """Serve one response whose Content-Length far exceeds the bytes sent, then hang up. A real socket, deliberately: the truncated-body failure lives inside ``http.client``'s read path, and only a genuine short read raises the ``IncompleteRead`` the transport has to normalize. Yields the URL. + + ``status`` truncates an ERROR body instead — the case where the read + failure must not cost the status code. ``redirect_first`` answers the + first connection with a 302 to the same server, for the download paths + that follow one by hand. """ listener = socket.socket() listener.bind(("127.0.0.1", 0)) listener.listen(8) host, port = listener.getsockname() + served = 0 def serve() -> None: + nonlocal served while True: try: conn, _ = listener.accept() except OSError: return # the listener closed: the context manager is done + served += 1 with conn, contextlib.suppress(OSError): _drain_request(conn) - conn.sendall( - b"HTTP/1.1 200 OK\r\nContent-Type: audio/mpeg\r\n" - b"Content-Length: %d\r\nConnection: close\r\n\r\n%s" % (declared, body) - ) + if redirect_first and served == 1: + conn.sendall( + b"HTTP/1.1 302 Found\r\nLocation: http://%s:%d/signed\r\n" + b"Content-Length: 0\r\nConnection: close\r\n\r\n" + % (host.encode(), port) + ) + else: + conn.sendall( + b"HTTP/1.1 %s\r\nContent-Type: audio/mpeg\r\n" + b"Content-Length: %d\r\nConnection: close\r\n\r\n%s" + % (status, declared, body) + ) conn.shutdown(socket.SHUT_WR) thread = threading.Thread(target=serve, daemon=True) diff --git a/tests/drivers/test_transport.py b/tests/drivers/test_transport.py index 6a6ab54..ec154f5 100644 --- a/tests/drivers/test_transport.py +++ b/tests/drivers/test_transport.py @@ -10,10 +10,10 @@ import pytest from dex_engine.capabilities import Capabilities -from dex_engine.capabilities.transcribe.whisper_api import urllib_multipart_post +from dex_engine.capabilities.transcribe.whisper_api import WhisperApi, urllib_multipart_post from dex_engine.drivers.file import FileDriver from dex_engine.drivers.transport import normalize_httplib_errors, urllib_transport -from dex_engine.pipeline.classify import classify_connection +from dex_engine.pipeline.classify import ProviderInputError, classify_connection, classify_http from dex_engine.pipeline.types import Kind, Status from tests.drivers.conftest import make_unit, truncating_server @@ -32,6 +32,16 @@ def test_the_family_normalizes_not_just_incomplete_read(self): with pytest.raises(OSError, match="BadStatusLine"), normalize_httplib_errors(): raise http.client.BadStatusLine("garbage") + def test_a_truncated_error_body_still_classifies_by_its_status(self): + # The other half: the body of a 4xx is only detail, and losing the + # read must not lose the STATUS. Without the guard the truncated + # read escapes as a connection failure and a 404's `dead` — the + # item is gone — arrives as `blocked`, retried forever. + with truncating_server(status=b"404 Not Found") as url: + response = urllib_transport(url) + assert response.status == 404 + assert classify_http(response.status).status is Status.DEAD + def test_an_oserror_passes_through_untouched(self): # RemoteDisconnected already IS a ConnectionResetError; the guard # must not re-wrap what classify_connection already reads. @@ -50,6 +60,25 @@ def test_a_truncated_response_reaches_the_provider_as_an_oserror(self): urllib_multipart_post(url, api_key=None, fields={}, filename="a.mp3", file_bytes=b"x") assert "truncated response body" in str(caught.value) + def test_a_truncated_error_body_keeps_its_status(self): + with truncating_server(status=b"400 Bad Request", body=b"unsupported forma") as url: + status, body = urllib_multipart_post( + url, api_key=None, fields={}, filename="a.mp3", file_bytes=b"x" + ) + assert status == 400 + assert body == b"" # the detail goes, as at the transport seam; the status stays + + def test_a_truncated_400_still_parks_the_episode_manual(self, tmp_path): + # The verdict the status carries: 400 is the AUDIO's fault — manual, + # under the escalation clock. A read failure escaping instead reads + # as the endpoint being down: waiting, retried forever, no clock. + chunk = tmp_path / "chunk-000.mp3" + chunk.write_bytes(b"audio") + with truncating_server(status=b"400 Bad Request", body=b"unsupported forma") as url: + provider = WhisperApi(base_url=url, api_key="k", post=urllib_multipart_post) + with pytest.raises(ProviderInputError, match="rejected the audio"): + provider._transcribe_chunk(chunk, "") # noqa: SLF001 — the seam under test + class TestThroughADriverFetch: def test_truncated_download_is_blocked_never_an_engine_error(self): diff --git a/tests/test_inbox.py b/tests/test_inbox.py index 9c5c4cf..da26546 100644 --- a/tests/test_inbox.py +++ b/tests/test_inbox.py @@ -8,6 +8,8 @@ from dex_engine.inbox import ( GithubSeams, + _api_request, + _download_asset, _rewrite_pointer, _run_git, build_parser, @@ -17,6 +19,7 @@ reconcile, ) from tests.capabilities.conftest import fixture_bytes +from tests.drivers.conftest import truncating_server REPO = "owner/instance" ASSET_URL = "https://api.github.com/repos/owner/instance/releases/assets/123" @@ -374,6 +377,35 @@ def test_binary_blob_stdout_decodes_tolerantly(self, tmp_path, monkeypatch): assert not staged.startswith("version https://git-lfs") +class TestRealHttpSeams: + """The two live urllib sites, against a real socket that hangs up mid-body. + + ``main`` reports ``OSError``/``ValueError``/``RuntimeError`` as a stated + failure; ``http.client``'s protocol errors are none of those, so an + unnormalized truncated read reaches the operator as a traceback. + """ + + def test_a_truncated_api_read_arrives_as_a_reportable_error(self): + with truncating_server() as url, pytest.raises(OSError) as caught: # noqa: PT011 — the OSError shape IS the assertion + _api_request(url, "token") + assert "truncated response body" in str(caught.value) + + def test_a_truncated_asset_download_arrives_as_a_reportable_error(self): + with truncating_server() as url, pytest.raises(OSError) as caught: # noqa: PT011 — the OSError shape IS the assertion + _download_asset(url, "token") + assert "truncated response body" in str(caught.value) + + def test_a_truncated_read_after_the_redirect_arrives_the_same_way(self): + # The asset path's own 302 follow — a second urlopen, and the one + # that carries the large download. + with ( + truncating_server(redirect_first=True) as url, + pytest.raises(OSError) as caught, # noqa: PT011 — the OSError shape IS the assertion + ): + _download_asset(url, "token") + assert "truncated response body" in str(caught.value) + + class TestParser: def test_bare_and_ensure(self): assert build_parser().parse_args([]).command is None