diff --git a/design/ingestion-pipeline.md b/design/ingestion-pipeline.md index 3fff3ca..09b05d2 100644 --- a/design/ingestion-pipeline.md +++ b/design/ingestion-pipeline.md @@ -412,6 +412,23 @@ 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. + 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 @@ -485,6 +502,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 @@ -497,7 +517,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. 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. @@ -512,7 +547,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 @@ -525,17 +565,26 @@ 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 -`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 -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 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`, @@ -553,6 +602,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/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/__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/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/src/dex_engine/capabilities/transcribe/whisper_api.py b/src/dex_engine/capabilities/transcribe/whisper_api.py index 6bf7378..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 @@ -29,7 +31,13 @@ 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.transcribe import ( + PROMPT_MAX_TOKENS, + estimated_tokens, + keep_last_tokens, +) from dex_engine.pipeline.types import Availability __all__ = [ @@ -49,9 +57,13 @@ # 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_TOKENS: 50 tokens of the +# running transcript is ample for carrying word boundaries and style into +# 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 @@ -96,7 +108,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 +129,24 @@ 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: + 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: @@ -307,8 +333,21 @@ 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_TOKENS`` budget, because + whisper reads a prompt from its END: appending an unbudgeted tail + 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() - return f"{initial_prompt}\n{tail}".strip() + tail = keep_last_tokens(" ".join(previous), _CONTINUITY_TAIL_TOKENS).strip() + if not tail: + return initial_prompt + 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/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/src/dex_engine/drivers/file.py b/src/dex_engine/drivers/file.py index 6e44c62..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,11 +27,12 @@ from dex_engine.capabilities import Capabilities from dex_engine.pipeline.classify import ( + ProviderUnavailableError, ScannedDocumentError, 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 +50,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 +110,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 @@ -185,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/drivers/transport.py b/src/dex_engine/drivers/transport.py index 05a33b1..558d73c 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,40 @@ 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. + # ``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( - 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/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/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..2d4f4ad 100644 --- a/src/dex_engine/pipeline/transcribe.py +++ b/src/dex_engine/pipeline/transcribe.py @@ -15,25 +15,32 @@ import contextlib import json +import unicodedata from collections.abc import Callable from dataclasses import dataclass from pathlib import Path 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__ = [ + "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", @@ -44,10 +51,21 @@ # 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 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" _TRANSCRIPT_HEADING = "## Transcript" @@ -260,6 +278,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 +292,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 +322,122 @@ 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, with the names LAST. + + 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. + show: The show or channel name. + vocabulary: Show notes or description — the trimmable part. + + Returns: + The priming text, an estimated :data:`PROMPT_MAX_TOKENS` at most. + """ + 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 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"{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 # --------------------------------------------------------------------------- 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_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()]) 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/capabilities/test_whisper_api.py b/tests/capabilities/test_whisper_api.py index 82877b6..47c1c46 100644 --- a/tests/capabilities/test_whisper_api.py +++ b/tests/capabilities/test_whisper_api.py @@ -12,6 +12,13 @@ _chunk_prompt, ) from dex_engine.pipeline.classify import ProviderInputError, ProviderUnavailableError +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: @@ -225,9 +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) <= 402 # 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 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 self.HEAD in prompt + assert prompt.endswith("transcribed words") + 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/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/drivers/conftest.py b/tests/drivers/conftest.py index e13ce12..d0b6165 100644 --- a/tests/drivers/conftest.py +++ b/tests/drivers/conftest.py @@ -1,7 +1,11 @@ """Shared driver-test plumbing: fixture loading, fake transports, work units.""" +import contextlib import json -from collections.abc import Mapping +import re +import socket +import threading +from collections.abc import Iterator, Mapping from pathlib import Path import pytest @@ -75,3 +79,84 @@ def __call__(self, url: str, *, method: str = "GET") -> HttpResponse: @pytest.fixture 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, + status: bytes = b"200 OK", + redirect_first: bool = False, +) -> Iterator[str]: + """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) + 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) + thread.start() + try: + yield f"http://{host}:{port}/ep42.mp3" + finally: + listener.close() + thread.join(timeout=5) 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. diff --git a/tests/drivers/test_transport.py b/tests/drivers/test_transport.py new file mode 100644 index 0000000..ec154f5 --- /dev/null +++ b/tests/drivers/test_transport.py @@ -0,0 +1,92 @@ +"""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.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 ProviderInputError, classify_connection, classify_http +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_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. + original = ConnectionResetError("reset by peer") + with pytest.raises(ConnectionResetError) as caught, normalize_httplib_errors(): + raise original + 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) + + 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): + 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..9e4ed12 100644 --- a/tests/pipeline/test_transcribe.py +++ b/tests/pipeline/test_transcribe.py @@ -10,20 +10,28 @@ 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 -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 ( + HEAD_MAX_TOKENS, TRANSCRIBE_RUN_CAP, Acquired, YoutubeAudio, _cached_audio, _download_enclosure, + _prompt, acquire_youtube_audio, + estimated_tokens, + keep_last_tokens, read_enrichment, ) from dex_engine.pipeline.types import ( @@ -39,7 +47,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 +540,54 @@ 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_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. @@ -550,6 +611,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, @@ -732,6 +837,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 @@ -762,6 +914,140 @@ 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, and counts the window in TOKENS.""" + + TITLE = "Ledgers as Work Queues" + SHOW = "Engineering Distilled" + HEAD = f"{TITLE} — {SHOW}" + + 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.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("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"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 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_is_capped_rather_than_eating_the_prompt(self): + prompt = _prompt("T" * 900, self.SHOW, "notes") + 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 + # 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.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: def test_round_trips_quoted_values(self, tmp_path): record = tmp_path / "podcast-abc123.md" 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