Providers: wrong exception class escaping the lifecycle

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
+`<hash>.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.
 

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]

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"

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:

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()