From b86271936c52387ee7ec96d8ebeb8599ce36de32 Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sat, 22 Aug 2026 10:28:32 +0100 Subject: [PATCH 01/21] Derive item completeness and digest staleness from the ledger MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An item is one unit of knowledge: the post, the thread parents above it, the links harvest promoted, the media, the video awaiting its transcript. Nothing about it advances to digest or wiki until every part has landed, so `status: enriched` now means the whole item landed — no unit queued, waiting, blocked, error, or manual — rather than "some markdown exists in the enrichment directory". A dead or skipped unit owes nothing and holds nothing hostage. The derivation reads the ledger; the run passes the map it already holds, and the quiet verbs read it themselves without gaining a way to die after their own write has landed. The run report then states the shape of what is missing per item — "3 of 4 units landed — 1 waiting on transcription" — so a session never infers completeness from a list of units. Digest staleness moves off file mtimes for the same reason it was undetectable on a second machine: git stamps every file at checkout, so after a clone the whole tree shares one mtime. The comparison is now the item's newest `done` ledger date against its recorded digest pass date — both committed state, day granularity, same-day enrich-then-digest not stale. (The digest file's own `date:` is the item's SHARE date per state-formats.md, so it cannot serve as the digest's date.) Co-Authored-By: Claude Fable 5 --- design/ingestion-pipeline.md | 34 ++- .../skills/dex-run/references/ingest-item.md | 11 +- src/dex_engine/pipeline/run.py | 172 +++++++++++++-- src/dex_engine/render/surfaces.py | 81 ++++++- tests/pipeline/test_run.py | 204 +++++++++++++++++- tests/render/test_surfaces.py | 74 +++++++ 6 files changed, 550 insertions(+), 26 deletions(-) diff --git a/design/ingestion-pipeline.md b/design/ingestion-pipeline.md index 09b05d2..90bc2b7 100644 --- a/design/ingestion-pipeline.md +++ b/design/ingestion-pipeline.md @@ -139,6 +139,12 @@ and image-only captures — no URLs, no work units): they surface as cognitive work ("awaiting description + digest") so a capture with nothing to fetch is never invisible to the session. +**An incomplete item says so on the report.** Every item the run touched +that still owes work gets a line stating the shape of it — "3 of 4 units +landed — 1 waiting on transcription" — so completeness is read, never +inferred from a list of units. Items the run did not touch are the standing +view's job (`enrich status`), not the run report's. + ## 2. Interfaces `typing.Protocol` throughout (structural interfaces; engine floor is 3.11). @@ -282,10 +288,16 @@ vocabulary only and never become work units: **Need** — `transcribe, extract, ocr`. Needs are mechanical and resource-keyed only. Cognitive obligations on *items* (re-judge under new harvest rules, refresh a stale digest) are never queued — they are **derived -state**, computed on demand from files already on disk (`passes.jsonl` rules -version vs the current constant; digest date vs the ledger's last `done` for -the item). Mechanical obligations are queued because they're work to be -done; staleness is derived because it's a fact that shows. +state**, computed on demand from state already committed (`passes.jsonl` +rules version vs the current constant; the item's digest pass date in +`passes.jsonl` vs the ledger's last `done` for the item). Mechanical +obligations are queued because they're work to be done; staleness is +derived because it's a fact that shows. **Both comparands are dates in +committed state, never file mtimes**: git stamps every file at checkout, so +on a second machine the whole tree shares one mtime and staleness becomes +undetectable. Day granularity is the intent — enriching and digesting in +one session is not stale. The digest file's own `date:` is the item's +*share* date, so it can never serve here. **`via`** (provenance) stays a documented string, not an enum — `harvest, thread, media, sniff, extract-asset, migration-` — because @@ -1167,10 +1179,16 @@ Skill changes shipping with this: interpretive context (what a linked video is, how a thread relates) belongs in the digest, not the item body; thread context itself lives in the enrichment via walk-up. After creation, exactly two frontmatter - fields ever change (`status`, `enrichment:` listing), both derived from - disk, both written by corpus.py. The derivation rule: `status: - enriched` iff `enrichment//` holds markdown files, and the listing - is those filenames, sorted. Every run reconciles **every** item against + fields ever change (`status`, `enrichment:` listing), both derived, both + written by corpus.py. The derivation rule: the listing is the markdown + filenames in `enrichment//`, sorted; `status: enriched` iff the item + holds enrichment **and no unit it owns is still outstanding** — queued, + waiting, blocked, error, or manual. An item is one unit of knowledge (the + post, the thread parents above it, the links harvest promoted, the media, + the video awaiting its transcript) and nothing about it advances to + digest or wiki until every part has landed, so one pending transcript + holds the whole item at `raw`; a `dead` or `skipped` unit owes nothing and + holds nothing hostage. Every run reconciles **every** item against disk — not just the units it drained — writing only on change, so enrichment written outside the drain (media descriptions, cognitive heals) converges at the next run; `enrich mark` and `enrich pass` diff --git a/instance/skills/dex-run/references/ingest-item.md b/instance/skills/dex-run/references/ingest-item.md index 905965f..916ef4a 100644 --- a/instance/skills/dex-run/references/ingest-item.md +++ b/instance/skills/dex-run/references/ingest-item.md @@ -36,8 +36,10 @@ page or video title, once known). The body is the owner's note **verbatim and stays that way** — your interpretive context (what a linked video is, how a thread relates) belongs in the digest, never the item body; thread context lives in the enrichment via walk-up. After creation, exactly two -frontmatter fields ever change (`status`, `enrichment:`), both derived -from disk, both written by the engine. +frontmatter fields ever change (`status`, `enrichment:`), both derived by +the engine: the listing from the enrichment directory, the status from the +ledger — `enriched` only once every unit the item owns has landed (or is +confirmed gone or deliberately skipped), `raw` while any is still owed. Then delete the capture file — the capture is preserved in git history and its content lives on in the corpus. @@ -120,7 +122,10 @@ otherwise 2–5 kebab-case candidates. Then A parked item (waiting/blocked/manual) still exists — provenance and note were captured at ingest — but gets no digest or wiki work until its -sources land. Don't force it; the report tracks it. +sources land. Don't force it; the run report's **incomplete** section names +each such item and what it is still owed ("3 of 4 units landed — 1 waiting +on transcription"), and the item stays `status: raw` until the last unit +lands. ## 7. Place (judgment) diff --git a/src/dex_engine/pipeline/run.py b/src/dex_engine/pipeline/run.py index 3862507..15e3049 100644 --- a/src/dex_engine/pipeline/run.py +++ b/src/dex_engine/pipeline/run.py @@ -113,6 +113,10 @@ HARVEST_RULES_VERSION = 1 _PARKED = frozenset({Status.WAITING, Status.BLOCKED, Status.ERROR, Status.MANUAL}) +# Work still owed on an item: queued and every parked status. Its complement +# — done, dead, skipped — is a unit that has landed, is confirmed gone, or +# was deliberately closed out; none of the three is owed anything further. +_OUTSTANDING = _PARKED | {Status.QUEUED} _PASS_STAGES = frozenset({"harvest", "digest", "wiki"}) # Whitespace and control characters anywhere in a URL make the HTTP client @@ -291,6 +295,10 @@ class _Drain: item_paths: dict[str, Path] = field(default_factory=dict) item_status: dict[str, str] = field(default_factory=dict) no_source_items: list[str] = field(default_factory=list) + # Items this run wrote an outcome for — the report's incompleteness + # section covers exactly these (an item nothing happened to this run has + # nothing new to say; `enrich status` holds the standing view). + touched: set[str] = field(default_factory=set) # Rerun-cohort pacing bookkeeping (full drains only). The counted set # exists because a redetected unit re-enters the live queue and pops # twice — one logical unit must spend one slot of the cohort, not two. @@ -1156,6 +1164,7 @@ def record(self, entry: LedgerEntry, *, count: bool = False) -> None: self.entries[stamped.hash] = stamped if count: self.counts[stamped.status] = self.counts.get(stamped.status, 0) + 1 + self.touched.add(stamped.item) if count and stamped.status in _PARKED: self.parked.append( { @@ -1215,7 +1224,9 @@ def _refresh_items(self) -> None: write-only-on-change rule keeps untouched items untouched. """ for item_id, path in self.item_paths.items(): - detail = _refresh_item_frontmatter(self.ctx.instance, item_id, path) + detail = _refresh_item_frontmatter( + self.ctx.instance, item_id, path, entries=self.entries + ) if detail is not None: self.notes.append( f"item {item_id}: an enrichment file cannot be listed in " @@ -1258,6 +1269,9 @@ def report_payload(self) -> dict[str, object]: "items": items, "parked": self.parked, } + incomplete = self._incomplete_items() + if incomplete: + payload["incomplete"] = incomplete cognitive = self._cognitive_jobs() if cognitive: payload["cognitive"] = cognitive @@ -1265,6 +1279,39 @@ def report_payload(self) -> dict[str, object]: payload["notes"] = list(self.notes) return payload + def _incomplete_items(self) -> list[dict[str, object]]: + """Why each touched item is still raw — the shape of what is missing. + + An item is one unit of knowledge, and none of it advances to digest + or wiki until every part has landed. The report states that shape + per item so a session never has to infer completeness by reading a + list of units. + """ + rows: list[dict[str, object]] = [] + for item_id in sorted(self.touched): + units = [entry for entry in self.entries.values() if entry.item == item_id] + outstanding = [entry for entry in units if entry.status in _OUTSTANDING] + if not outstanding: + continue + groups: dict[tuple[str, str | None], int] = {} + for entry in outstanding: + key = (entry.status.value, entry.needs.value if entry.needs else None) + groups[key] = groups.get(key, 0) + 1 + rows.append( + { + "item": item_id, + "landed": len(units) - len(outstanding), + "total": len(units), + "outstanding": [ + {"status": status, "count": count} + if needs is None + else {"status": status, "needs": needs, "count": count} + for (status, needs), count in groups.items() + ], + } + ) + return rows + def _cognitive_jobs(self) -> list[dict[str, str]]: """Waiting jobs that resolve to the cognitive floor. @@ -1316,9 +1363,18 @@ def _drop_superseded_outputs(instance: Instance, entry: LedgerEntry, path: str) def _refresh_item_frontmatter( - instance: Instance, item_id: str, path: Path | None = None + instance: Instance, + item_id: str, + path: Path | None = None, + *, + entries: dict[str, LedgerEntry] | None = None, ) -> str | None: - """Derive one item's ``status``/``enrichment:`` from disk; write only on change. + """Derive one item's ``status``/``enrichment:``; write only on change. + + The listing is the enrichment directory's markdown files; the status is + the ledger's answer to "has the whole item landed" — ``enriched`` only + when no unit the item owns is still outstanding, ``raw`` while one is. + Callers holding the ledger pass ``entries``; the rest read it here. Silent when the corpus file is gone or unreadable: a heal may outlive its item (excluded after capture), an unreadable item is seeding's to @@ -1339,9 +1395,10 @@ def _refresh_item_frontmatter( # had landed, and the retry duplicated the record. return None files = sorted(p.name for p in (instance.enrichment_dir / item_id).glob("*.md")) + units = _item_units(instance, item_id, entries) try: updated = dataclasses.replace( - item, status="enriched" if files else "raw", enrichment=files + item, status=_derived_status(item, files, units), enrichment=files ) if updated != item: corpus.write_item(path, updated) @@ -1350,6 +1407,41 @@ def _refresh_item_frontmatter( return None +def _item_units( + instance: Instance, item_id: str, entries: dict[str, LedgerEntry] | None +) -> list[LedgerEntry] | None: + """The item's ledger units — from the caller's map, or read here. + + None means the ledger could not be read at all: the quiet callers write + first and refresh after, so a nonconforming line must leave the status + alone rather than raise past a write that already landed. + """ + if entries is None: + try: + entries = ledger.load(instance.ledger_path) + except (OSError, UnicodeDecodeError, ledger.LedgerSchemaError): + return None + return [entry for entry in entries.values() if entry.item == item_id] + + +def _derived_status( + item: corpus.CorpusItem, files: list[str], units: list[LedgerEntry] | None +) -> str: + """``enriched`` iff the item holds enrichment and owes no further work. + + An item is one unit of knowledge — the post, the thread parents above + it, the links harvest promoted, the media, the video awaiting its + transcript — so one outstanding unit keeps the whole item ``raw`` and + out of digest/wiki. A dead link or a deliberately skipped unit owes + nothing and holds nothing hostage. + """ + if not files: + return "raw" + if units is None: + return item.status + return "raw" if any(unit.status in _OUTSTANDING for unit in units) else "enriched" + + # --------------------------------------------------------------------------- # Verbs (called by the CLI; zero business logic lives there). # --------------------------------------------------------------------------- @@ -1658,32 +1750,84 @@ def digest_orphans(instance: Instance) -> list[str]: Shared by ``enrich status`` and lint's health check: both list the same interrupted-session orphans, computed in one place. + Staleness is read from committed dates, never file mtimes: git stamps + every file at checkout, so on a second machine the whole tree carries + one mtime and staleness would be undetectable. The two dates are the + item's newest ``done`` ledger line and its digest pass record — both + travel in git, and day granularity is the intent (enriching and + digesting in one session is not stale). + Args: instance: The instance. Returns: - Item ids whose enrichment outputs are newer than their digest (or - that have outputs and no digest at all). + Item ids whose enrichment landed after their digest pass (or that + have enrichment and no digest at all). """ orphans = [] if not instance.enrichment_dir.is_dir(): return orphans + enriched_on = _last_enriched(instance) + digested_on = _last_digested(instance) for item_dir in sorted(instance.enrichment_dir.iterdir()): if not item_dir.is_dir(): continue - outputs = list(item_dir.glob("*.md")) - if not outputs: + if not any(item_dir.glob("*.md")): continue - digest = instance.digests_dir / f"{item_dir.name}.md" - if not digest.exists(): + if not (instance.digests_dir / f"{item_dir.name}.md").exists(): orphans.append(item_dir.name) continue - newest = max(path.stat().st_mtime for path in outputs) - if newest > digest.stat().st_mtime: + landed = enriched_on.get(item_dir.name) + digested = digested_on.get(item_dir.name) + # No dates on either side is no claim: enrichment written outside + # the ledger (a media description) and digests older than the pass + # record are facts about history, not staleness. + if landed is not None and digested is not None and digested < landed: orphans.append(item_dir.name) return orphans +def _last_enriched(instance: Instance) -> dict[str, datetime.date]: + """Each item's newest ``done`` ledger date.""" + try: + entries = ledger.load(instance.ledger_path) + except (OSError, UnicodeDecodeError, ledger.LedgerSchemaError): + # A nonconforming ledger is lint's own loud finding; the staleness + # backstop simply has nothing to compare and says nothing. + return {} + newest: dict[str, datetime.date] = {} + for entry in entries.values(): + if entry.status is Status.DONE: + newest[entry.item] = max(newest.get(entry.item, entry.date), entry.date) + return newest + + +def _last_digested(instance: Instance) -> dict[str, datetime.date]: + """Each item's newest recorded digest pass date.""" + path = instance.passes_path + if not path.exists(): + return {} + newest: dict[str, datetime.date] = {} + for line in path.read_text(encoding="utf-8").split("\n"): + if not line.strip(): + continue + try: + record = json.loads(line) + except json.JSONDecodeError: + continue # lint parses this file loudly; the backstop skips the line + if not isinstance(record, dict) or record.get("stage") != "digest": + continue + item, raw_date = record.get("item"), record.get("date") + if not isinstance(item, str) or not isinstance(raw_date, str): + continue + try: + date = datetime.date.fromisoformat(raw_date) + except ValueError: + continue + newest[item] = max(newest.get(item, date), date) + return newest + + def mark( # noqa: PLR0913 — the verb mirrors its CLI flags ctx: RunContext, url: str, @@ -1775,7 +1919,9 @@ def mark( # noqa: PLR0913 — the verb mirrors its CLI flags # drop the item keeps two files for one unit and serves the stale # pre-correction view to the digest and query layers forever. _drop_superseded_outputs(ctx.instance, healed, effective_path) - _refresh_item_frontmatter(ctx.instance, prior.item) + # The heal's own line included: the drain's map is the ledger as of this + # write, so a heal that completes an item flips its status in the same call. + _refresh_item_frontmatter(ctx.instance, prior.item, entries=drain.entries) return f"marked {prior.url} ({prior.hash}) {status.value}" diff --git a/src/dex_engine/render/surfaces.py b/src/dex_engine/render/surfaces.py index 906850a..d9c7e1a 100644 --- a/src/dex_engine/render/surfaces.py +++ b/src/dex_engine/render/surfaces.py @@ -205,6 +205,13 @@ def _render_enrich_report(payload: Mapping[str, object]) -> str: # fresh / rerun / waiting-drained "parked": [{"item": str, "url": str, "status": str, "reason": str}], # survives the session + "incomplete": [{"item": str, # optional: touched items + "landed": int, # still owed work — the + "total": int, # ledger units that have + "outstanding": [ # landed, of all it owns + {"status": str, # an outstanding status + "needs": str, # optional + "count": int}]}], "cognitive": [{"item": str, "url": str, "need": str}], # optional: jobs for the session "issues_filed": int, # optional, default 0 @@ -216,10 +223,11 @@ def _render_enrich_report(payload: Mapping[str, object]) -> str: surface, payload, required=frozenset({"counts", "items", "parked"}), - optional=frozenset({"cognitive", "issues_filed", "notes"}), + optional=frozenset({"incomplete", "cognitive", "issues_filed", "notes"}), ) counts = _counts_at(surface, payload, "counts") item_rows = _enrich_item_rows(surface, payload) + incomplete_rows = _enrich_incomplete_rows(surface, payload) parked_rows = _enrich_parked_rows(surface, payload) cognitive_rows = _enrich_cognitive_rows(surface, payload) issues_filed = _int_at(surface, payload, "issues_filed", default=0) @@ -237,6 +245,13 @@ def _render_enrich_report(payload: Mapping[str, object]) -> str: lines.append(kernel.table(item_rows, indent=2).rstrip("\n")) else: lines.append("cognitive work — none (nothing new or changed)") + if incomplete_rows: + lines.append("") + lines.append( + f"incomplete — {_plural(len(incomplete_rows), 'item')} still raw " + "until every unit lands:" + ) + lines.append(kernel.table(incomplete_rows, indent=2).rstrip("\n")) lines.append("") if parked_rows: verb = "survives" if len(parked_rows) == 1 else "survive" @@ -272,6 +287,70 @@ def _enrich_item_rows(surface: str, payload: Mapping[str, object]) -> list[list[ return rows +# What an outstanding unit is waiting for, as prose. Statuses that are not +# outstanding never reach this surface — the payload validation refuses them. +_OUTSTANDING_STATUSES = frozenset(_PARKED_STATUSES | {Status.QUEUED}) +_NEED_NOUNS = {Need.TRANSCRIBE: "transcription", Need.EXTRACT: "extraction", Need.OCR: "OCR"} +_STATUS_PHRASES = { + Status.QUEUED: "queued", + Status.WAITING: "waiting", + Status.BLOCKED: "blocked", + Status.ERROR: "in error", + Status.MANUAL: "needing a decision", +} + + +def _enrich_incomplete_rows(surface: str, payload: Mapping[str, object]) -> list[list[str]]: + rows = [] + for i, entry in enumerate(_obj_list_at(surface, payload, "incomplete", required=False)): + where = f"incomplete[{i}]." + _check_keys( + surface, + entry, + required=frozenset({"item", "landed", "total", "outstanding"}), + where=where, + ) + landed = _int_at(surface, entry, "landed", where) + total = _int_at(surface, entry, "total", where) + if landed > total: + _fail(surface, f"{where}landed ({landed}) exceeds total ({total})") + shape = f"{landed} of {_plural(total, 'unit')} landed" + outstanding = _enrich_outstanding(surface, entry, where) + rows.append([_str_at(surface, entry, "item", where), f"{shape} — {outstanding}"]) + return rows + + +def _enrich_outstanding(surface: str, entry: Mapping[str, object], where: str) -> str: + parts = [] + for j, group in enumerate(_obj_list_at(surface, entry, "outstanding", required=True)): + gwhere = f"{where}outstanding[{j}]." + _check_keys( + surface, + group, + required=frozenset({"status", "count"}), + optional=frozenset({"needs"}), + where=gwhere, + ) + status = _status_at(surface, group, "status", gwhere) + if status not in _OUTSTANDING_STATUSES: + allowed = ", ".join(sorted(s.value for s in _OUTSTANDING_STATUSES)) + _fail( + surface, + f"{gwhere}status must be an outstanding status ({allowed}), got {status.value!r}", + ) + phrase = _STATUS_PHRASES[status] + if "needs" in group: + need = _str_at(surface, group, "needs", gwhere) + if need not in {n.value for n in Need}: + options = ", ".join(n.value for n in Need) + _fail(surface, f"{gwhere}needs must be one of {options}, got {need!r}") + phrase += f" on {_NEED_NOUNS[Need(need)]}" + parts.append(f"{_int_at(surface, group, 'count', gwhere)} {phrase}") + if not parts: + _fail(surface, f"{where}outstanding must name at least one outstanding unit") + return ", ".join(parts) + + def _enrich_parked_rows(surface: str, payload: Mapping[str, object]) -> list[list[str]]: rows = [] for i, entry in enumerate(_obj_list_at(surface, payload, "parked", required=True)): diff --git a/tests/pipeline/test_run.py b/tests/pipeline/test_run.py index 7db45f5..c52526b 100644 --- a/tests/pipeline/test_run.py +++ b/tests/pipeline/test_run.py @@ -375,6 +375,72 @@ def test_mark_still_heals_when_the_listing_cannot_refresh(self, instance): assert confirmation.endswith("done") # the heal landed, quietly assert entry_for(ctx).status is Status.DONE + VIDEO_URL = "https://example.test/video" + + def _mixed_item(self, instance, second: Result): + """An item whose blog unit lands and whose video unit gets ``second``.""" + path = write_item(instance, urls=[URL, self.VIDEO_URL]) + + def fetch(unit): + if unit.url == self.VIDEO_URL: + return second + return Result(status=Status.DONE, meta={"title": "t"}, body="substantial body " * 30) + + return path, make_ctx(instance, FakeDriver(fetch_fn=fetch)) + + def _waiting_video(self) -> Result: + return Result( + status=Status.WAITING, + meta={"title": "v"}, + needs=Need.TRANSCRIBE, + reason="no captions available", + body="## Description\n\nwhat the video covers", + ) + + def test_an_outstanding_unit_keeps_the_whole_item_raw(self, instance): + path, ctx = self._mixed_item(instance, self._waiting_video()) + run_mod.run(ctx) + item = corpus.read_item(path) + assert item.status == "raw" # the transcript is still owed + assert len(item.enrichment) == 2 # both units wrote what they had + + def test_the_item_flips_to_enriched_when_the_last_unit_lands(self, instance): + path, ctx = self._mixed_item(instance, self._waiting_video()) + run_mod.run(ctx) + assert corpus.read_item(path).status == "raw" + video_out = f"enrichment/{ITEM}/web-{work_hash(self.VIDEO_URL)[:6]}.md" + run_mod.mark(ctx, self.VIDEO_URL, Status.DONE, path=video_out) + assert corpus.read_item(path).status == "enriched" + + def test_a_dead_unit_never_holds_the_item_hostage(self, instance): + dead = Result(status=Status.DEAD, meta={}, reason="404") + path, ctx = self._mixed_item(instance, dead) + run_mod.run(ctx) + assert corpus.read_item(path).status == "enriched" # confirmed gone owes nothing + + def test_skipped_and_manual_units_differ(self, instance): + skipped = Result(status=Status.SKIPPED, meta={}, reason="deliberately not fetched") + path, ctx = self._mixed_item(instance, skipped) + run_mod.run(ctx) + assert corpus.read_item(path).status == "enriched" + run_mod.mark(ctx, self.VIDEO_URL, Status.MANUAL, reason="reopened by hand") + assert corpus.read_item(path).status == "raw" + + def test_mark_and_pass_work_on_an_item_with_outstanding_units(self, instance): + path, ctx = self._mixed_item(instance, self._waiting_video()) + run_mod.run(ctx) + assert run_mod.record_pass(ctx, ITEM, "harvest").endswith(ITEM) + confirmation = run_mod.mark(ctx, URL, Status.SKIPPED, reason="superseded by hand") + assert confirmation.endswith("skipped") + assert corpus.read_item(path).status == "raw" # the video is still waiting + + def test_an_unreadable_ledger_leaves_the_derived_status_alone(self, instance): + path, ctx = self._mixed_item(instance, self._waiting_video()) + run_mod.run(ctx) + instance.ledger_path.write_text("{not json at all\n", encoding="utf-8") + assert run_mod.record_pass(ctx, ITEM, "digest").endswith(ITEM) # the write stands + assert corpus.read_item(path).status == "raw" + def test_mark_for_an_excluded_items_unit_still_heals(self, instance): path = write_item(instance) ctx = make_ctx(instance, FakeDriver()) @@ -385,6 +451,79 @@ def test_mark_for_an_excluded_items_unit_still_heals(self, instance): assert entry_for(ctx).status is Status.SKIPPED +class TestIncompleteItemsOnTheReport: + """The run says WHY an item is still raw — never left to inference.""" + + VIDEO_URL = "https://example.test/video" + + def _ctx(self, instance, second: Result): + write_item(instance, urls=[URL, self.VIDEO_URL]) + + def fetch(unit): + if unit.url == self.VIDEO_URL: + return second + return Result(status=Status.DONE, meta={"title": "t"}, body="substantial body " * 30) + + return make_ctx(instance, FakeDriver(fetch_fn=fetch)) + + def test_the_report_names_the_shape_of_what_is_missing(self, instance): + waiting = Result( + status=Status.WAITING, + meta={"title": "v"}, + needs=Need.TRANSCRIBE, + reason="no captions available", + ) + report = run_mod.run(self._ctx(instance, waiting)) + flat = " ".join(report.split()) + assert "incomplete — 1 item still raw until every unit lands" in flat + assert f"{ITEM} 1 of 2 units landed — 1 waiting on transcription" in flat + + def test_a_complete_item_gets_no_line(self, instance): + dead = Result(status=Status.DEAD, meta={}, reason="404") + report = run_mod.run(self._ctx(instance, dead)) + assert "incomplete" not in report + + def test_the_payload_carries_the_counts(self, instance, monkeypatch): + waiting = Result( + status=Status.WAITING, meta={}, needs=Need.TRANSCRIBE, reason="no captions available" + ) + captured: dict[str, dict] = {} + rendered = run_mod.surfaces.render + + def spy(surface, payload): + captured[surface] = dict(payload) + return rendered(surface, payload) + + monkeypatch.setattr(run_mod.surfaces, "render", spy) + run_mod.run(self._ctx(instance, waiting)) + assert captured["enrich-report"]["incomplete"] == [ + { + "item": ITEM, + "landed": 1, + "total": 2, + "outstanding": [{"status": "waiting", "needs": "transcribe", "count": 1}], + } + ] + + def test_an_untouched_item_is_not_listed(self, instance): + # The standing view is `enrich status`; the run reports on what it + # touched, so a long-parked item does not repeat every run. + waiting = Result( + status=Status.WAITING, meta={}, needs=Need.TRANSCRIBE, reason="no captions available" + ) + ctx = self._ctx(instance, waiting) + run_mod.run(ctx) + assert "incomplete" not in run_mod.run(ctx) # nothing drainable the second time + + def test_the_other_sections_still_render(self, instance): + waiting = Result( + status=Status.WAITING, meta={}, needs=Need.TRANSCRIBE, reason="no captions available" + ) + report = run_mod.run(self._ctx(instance, waiting)) + assert "cognitive work — 1 item with new or changed content:" in report + assert "parked — 1 entry survives this session" in report + + class TestChildren: def test_children_re_enter_with_provenance_and_drain_same_run(self, instance): write_item(instance) @@ -1550,6 +1689,68 @@ def test_digested_items_are_not_orphans(self, instance): report = run_mod.status_report(ctx) assert ITEM not in report + def _enriched_and_digested(self, instance, *, enriched, digested) -> None: + """One item with a done ledger line and a recorded digest pass.""" + item_dir = instance.enrichment_dir / ITEM + item_dir.mkdir(parents=True, exist_ok=True) + (item_dir / "web-abc123.md").write_text("enriched", encoding="utf-8") + instance.digests_dir.mkdir(parents=True, exist_ok=True) + (instance.digests_dir / f"{ITEM}.md").write_text("digested", encoding="utf-8") + ledger.append( + instance.ledger_path, + LedgerEntry( + hash=work_hash(URL), + url=URL, + item=ITEM, + kind=Kind.WEB, + status=Status.DONE, + engine="0.2.0", + date=enriched, + path=f"enrichment/{ITEM}/web-abc123.md", + ), + ) + instance.passes_path.write_text( + json.dumps({"stage": "digest", "item": ITEM, "date": digested.isoformat()}) + "\n", + encoding="utf-8", + ) + + def test_a_clone_still_sees_stale_enrichment(self, instance): + # Every file carries the checkout's mtime after a clone; the dates + # in the ledger and the pass record are what survive git. + self._enriched_and_digested( + instance, enriched=datetime.date(2026, 8, 20), digested=datetime.date(2026, 8, 18) + ) + stamp = (instance.digests_dir / f"{ITEM}.md").stat().st_mtime + for path in instance.root.rglob("*"): + if path.is_file(): + os.utime(path, (stamp, stamp)) + assert run_mod.digest_orphans(instance) == [ITEM] + + def test_same_day_enrich_then_digest_is_not_stale(self, instance): + day = datetime.date(2026, 8, 20) + self._enriched_and_digested(instance, enriched=day, digested=day) + assert run_mod.digest_orphans(instance) == [] + + def test_a_digest_pass_after_the_enrichment_is_not_stale(self, instance): + self._enriched_and_digested( + instance, enriched=datetime.date(2026, 8, 18), digested=datetime.date(2026, 8, 20) + ) + assert run_mod.digest_orphans(instance) == [] + + def test_enrichment_with_no_digest_is_still_an_orphan(self, instance): + self._enriched_and_digested( + instance, enriched=datetime.date(2026, 8, 20), digested=datetime.date(2026, 8, 20) + ) + (instance.digests_dir / f"{ITEM}.md").unlink() + assert run_mod.digest_orphans(instance) == [ITEM] + + def test_an_unreadable_ledger_makes_no_staleness_claim(self, instance): + self._enriched_and_digested( + instance, enriched=datetime.date(2026, 8, 20), digested=datetime.date(2026, 8, 18) + ) + instance.ledger_path.write_text("{not json\n", encoding="utf-8") + assert run_mod.digest_orphans(instance) == [] + class TestIssueFiling: """The filer wiring: error outcomes reach it; the report says so.""" @@ -1934,7 +2135,7 @@ def file_fetch(_unit): assert parked.status is Status.MANUAL assert web_out.exists() # the enrichment the item already had stands assert corpus.read_item(item_path).enrichment == [web_out.name] - assert corpus.read_item(item_path).status == "enriched" + assert corpus.read_item(item_path).status == "raw" # the parked unit is still owed # The success that replaces it is what drops it. mode["extract"] = True @@ -1944,6 +2145,7 @@ def file_fetch(_unit): assert done.status is Status.DONE assert not web_out.exists() assert corpus.read_item(item_path).enrichment == [f"file-{done.hash[:6]}.md"] + assert corpus.read_item(item_path).status == "enriched" # These two URLs share a sha1[:6] (6f0d01) and detect as different # kinds, so under one item they want output names that differ only by diff --git a/tests/render/test_surfaces.py b/tests/render/test_surfaces.py index 06a15d4..48dff89 100644 --- a/tests/render/test_surfaces.py +++ b/tests/render/test_surfaces.py @@ -108,6 +108,80 @@ def test_multiline_string_field_is_a_payload_error_not_a_kernel_error(self): with pytest.raises(PayloadError, match="single-line"): render("enrich-report", {"counts": {}, "items": items, "parked": []}) + def test_incomplete_items_state_the_shape_of_what_is_missing(self): + payload = { + **ENRICH_PAYLOAD, + "incomplete": [ + { + "item": "2026-08-19-example-55ad7b", + "landed": 3, + "total": 4, + "outstanding": [{"status": "waiting", "needs": "transcribe", "count": 1}], + } + ], + } + out = render("enrich-report", payload) + assert "incomplete — 1 item still raw until every unit lands" in out + assert "3 of 4 units landed — 1 waiting on transcription" in out + assert_no_trailing_whitespace(out) + + def test_several_outstanding_shapes_read_as_one_line(self): + payload = { + "counts": {}, + "items": [], + "parked": [], + "incomplete": [ + { + "item": "2026-08-19-example-55ad7b", + "landed": 0, + "total": 3, + "outstanding": [ + {"status": "blocked", "count": 2}, + {"status": "manual", "count": 1}, + ], + } + ], + } + out = render("enrich-report", payload) + assert "0 of 3 units landed — 2 blocked, 1 needing a decision" in out + + def test_absent_incomplete_says_nothing(self): + assert "incomplete" not in render("enrich-report", {**ENRICH_PAYLOAD, "incomplete": []}) + + def test_a_landed_status_in_outstanding_is_loud(self): + payload = { + "counts": {}, + "items": [], + "parked": [], + "incomplete": [ + { + "item": "i", + "landed": 0, + "total": 1, + "outstanding": [{"status": "done", "count": 1}], + } + ], + } + with pytest.raises(PayloadError, match="outstanding status"): + render("enrich-report", payload) + + def test_landed_beyond_total_is_loud(self): + payload = { + "counts": {}, + "items": [], + "parked": [], + "incomplete": [ + { + "item": "i", + "landed": 5, + "total": 1, + "outstanding": [{"status": "queued", "count": 1}], + } + ], + } + with pytest.raises(PayloadError, match="exceeds total"): + render("enrich-report", payload) + def test_multiline_note_is_a_payload_error(self): notes = ["a note\nwith a newline"] with pytest.raises(PayloadError, match=r"notes\[0\]"): From 5a68b96e4bfeac4e14d8ae107d60a799bddc1dde Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sat, 22 Aug 2026 10:35:00 +0100 Subject: [PATCH 02/21] Park youtube videos with the description already fetched MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A video with no usable captions returned `waiting` with no body, so the description yt-dlp had just handed over was thrown away — and a video deleted or made private during a transcription backlog took it for good. The podcast driver in the same position writes its show notes first; the two kinds now behave alike. The park writes the description as its `## Description` section and the transcribe drain appends the transcript to that same file instead of recomposing from a fresh probe, so a re-drain supersedes the transcript without duplicating or losing anything around it. That requires one body shape per kind: the transcript is now always its own `## Transcript` section, description or not, because the drain splits the stored body on that heading. Co-Authored-By: Claude Fable 5 --- design/ingestion-pipeline.md | 9 +++++ src/dex_engine/drivers/youtube.py | 41 +++++++++++++++++---- src/dex_engine/pipeline/run.py | 11 ++++-- src/dex_engine/pipeline/transcribe.py | 40 +++++++++++++++++---- tests/drivers/test_youtube.py | 17 +++++++++ tests/pipeline/test_transcribe.py | 51 +++++++++++++++++++++++++-- 6 files changed, 151 insertions(+), 18 deletions(-) diff --git a/design/ingestion-pipeline.md b/design/ingestion-pipeline.md index 90bc2b7..34aaf10 100644 --- a/design/ingestion-pipeline.md +++ b/design/ingestion-pipeline.md @@ -611,6 +611,15 @@ waiting-transcribe park that carries an enclosure pointer **always writes its park file** — §9's round-trip depends on the frontmatter pointer existing, show notes or not. +**A park never discards content it already fetched.** Both transcribable +kinds write what they have at park time — a podcast's show notes, a +video's description — and the drain **appends** the transcript to that +file rather than replacing it: a source that goes private during a +transcription backlog would otherwise take the fetched content with it. +One body shape per kind, whichever route produced it — the transcript is +always its own `## Transcript` section, so the drain can split a stored +body on that heading and compose onto what is already there. + **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`. diff --git a/src/dex_engine/drivers/youtube.py b/src/dex_engine/drivers/youtube.py index d0f883b..15fc4a9 100644 --- a/src/dex_engine/drivers/youtube.py +++ b/src/dex_engine/drivers/youtube.py @@ -20,7 +20,10 @@ The driver NEVER downloads audio — audio acquisition belongs to the transcribe drain. No usable captions means ``Result(waiting, needs=transcribe)``: the work is parked for the -capability, not given up on. +capability, not given up on — and the park carries the description it +already fetched, so the run layer writes that content now rather than +holding it hostage to a transcription backlog the video may not survive. +The drain appends the transcript to that same file. Probe failures are mapped honestly: an HTTP code buried in yt-dlp's message routes through the central classifier; private/sign-in walls are ``manual``; @@ -126,6 +129,13 @@ # defaults to `blocked`: the motivating-incident class, again. _GONE_MARKERS = ("video unavailable", "removed", "terminated") +# The youtube body's two sections. The transcribe drain splits a stored +# body on these same headings to append a transcript to a park's +# description (pipeline/transcribe.py); drivers cannot import that module +# (it imports this one), so the pairing is pinned by test instead. +_DESCRIPTION_HEADING = "## Description" +_TRANSCRIPT_HEADING = "## Transcript" + _VTT_NOISE_PREFIXES = ("WEBVTT", "Kind:", "Language:", "NOTE", "align:") _VTT_TAG_RE = re.compile(r"<[^>]+>") @@ -219,6 +229,7 @@ def fetch(self, unit: WorkUnit) -> Result: return Result( status=Status.WAITING, meta=meta, + body=_description_section(info) or None, needs=Need.TRANSCRIBE, reason="no captions available", ) @@ -248,6 +259,7 @@ def _fetch_transcript( return Result( status=Status.WAITING, meta=meta, + body=_description_section(info) or None, needs=Need.TRANSCRIBE, reason="captions track too thin to be a transcript", ) @@ -379,12 +391,29 @@ def _caption_track_url(info: dict) -> str | None: return None -def _body(info: dict, transcript: str) -> str: - """Description + transcript sections, as today; transcript alone otherwise.""" +def _description_section(info: dict) -> str: + """The description as its own body section, or "" when there is none. + + A park writes this on its own: the description is content already + fetched, and a video that goes private during a transcription backlog + would otherwise take it with it. The transcript is appended to this + same section later, never written over it. + """ description = (info.get("description") or "").strip() - if description: - return f"## Description\n\n{description}\n\n## Transcript\n\n{transcript}" - return transcript + return f"{_DESCRIPTION_HEADING}\n\n{description}" if description else "" + + +def _body(info: dict, transcript: str) -> str: + """Description section + transcript section — one body shape per kind. + + The transcript is always its own labelled section, description or not: + the drain splits a stored body on that heading to append a transcript + to what the park already wrote (``pipeline/transcribe.py``). + """ + section = _description_section(info) + if section: + return f"{section}\n\n{_TRANSCRIPT_HEADING}\n\n{transcript}" + return f"{_TRANSCRIPT_HEADING}\n\n{transcript}" def clean_vtt(vtt: str) -> str: diff --git a/src/dex_engine/pipeline/run.py b/src/dex_engine/pipeline/run.py index 15e3049..3d9d354 100644 --- a/src/dex_engine/pipeline/run.py +++ b/src/dex_engine/pipeline/run.py @@ -661,12 +661,17 @@ def _note_first_run(self, transcriber: Transcriber) -> None: def _acquire_audio(self, entry: LedgerEntry) -> Acquired | Classification: """Audio acquisition belongs to the drain, per kind.""" audio_dir = self.ctx.instance.cache_dir / "audio" + # Both kinds park with content already in hand (a description, show + # notes) and both compose the transcript onto what that park wrote, + # so both acquisitions read the unit's own output file. + name = f"{entry.kind.value}-{entry.hash[:6]}.md" + enrichment = self.ctx.instance.enrichment_dir / entry.item / name match entry.kind: case Kind.YOUTUBE: - return acquire_youtube_audio(entry, audio_dir, self.ctx.download_audio) + return acquire_youtube_audio( + entry, enrichment, audio_dir, self.ctx.download_audio + ) case Kind.PODCAST: - name = f"{entry.kind.value}-{entry.hash[:6]}.md" - enrichment = self.ctx.instance.enrichment_dir / entry.item / name return acquire_podcast_audio(entry, enrichment, audio_dir, self.ctx.transport) case _: return Classification( diff --git a/src/dex_engine/pipeline/transcribe.py b/src/dex_engine/pipeline/transcribe.py index 2d4f4ad..645ef61 100644 --- a/src/dex_engine/pipeline/transcribe.py +++ b/src/dex_engine/pipeline/transcribe.py @@ -67,7 +67,12 @@ _SEPARATOR = " — " _AUDIO_EXT_DEFAULT = "mp3" +# The body sections both transcribable kinds compose around. The youtube +# driver writes the same two headings on its parks (drivers/youtube.py); +# it cannot import them from here (this module imports that one), so the +# pairing is pinned by test. _TRANSCRIPT_HEADING = "## Transcript" +_DESCRIPTION_HEADING = "## Description" @dataclass(frozen=True, slots=True, kw_only=True) @@ -148,12 +153,18 @@ class Acquired: def acquire_youtube_audio( - entry: LedgerEntry, cache_dir: Path, download: DownloadAudio + entry: LedgerEntry, enrichment_path: Path, cache_dir: Path, download: DownloadAudio ) -> Acquired | Classification: """Acquire a YouTube unit's audio via the yt-dlp seam. + The description the park already wrote is the body the transcript + joins — the same composition the podcast path uses — so a re-drain + appends to what is on disk instead of writing over it. A park that had + no description to write falls back to the probe's own. + Args: entry: The waiting ledger entry. + enrichment_path: ``enrichment//youtube-.md``. cache_dir: ``cache/audio/``. download: The yt-dlp seam. @@ -173,8 +184,20 @@ def acquire_youtube_audio( "duration_min": audio.duration_min, "upload_date": audio.upload_date, } - prompt = _prompt(audio.title, audio.channel, audio.description) - return Acquired(audio=audio.path, meta=meta, prompt=prompt, prefix=audio.description) + description = _stored_description(enrichment_path) or audio.description + prompt = _prompt(audio.title, audio.channel, description) + return Acquired(audio=audio.path, meta=meta, prompt=prompt, prefix=description) + + +def _stored_description(path: Path) -> str: + """The description a youtube park already wrote, or "" when there is none.""" + if not path.exists(): + return "" + _fields, body = read_enrichment(path) + head = _pre_transcript(body) + if head.startswith(_DESCRIPTION_HEADING): + return head[len(_DESCRIPTION_HEADING) :].strip() + return head def acquire_podcast_audio( @@ -447,10 +470,15 @@ def keep_last_tokens(text: str, budget: float) -> str: def youtube_body(description: str, transcript: str) -> str: - """Description + transcript sections — the youtube driver's own pattern.""" + """Description + transcript sections — the youtube driver's own pattern. + + The transcript is always its own labelled section: a re-drain splits + the stored body on that heading, and a bare transcript would come back + as "description" and be duplicated under itself. + """ if description: - return f"## Description\n\n{description}\n\n{_TRANSCRIPT_HEADING}\n\n{transcript}" - return transcript + return f"{_DESCRIPTION_HEADING}\n\n{description}\n\n{_TRANSCRIPT_HEADING}\n\n{transcript}" + return f"{_TRANSCRIPT_HEADING}\n\n{transcript}" def podcast_body(show_notes: str, transcript: str) -> str: diff --git a/tests/drivers/test_youtube.py b/tests/drivers/test_youtube.py index 3a7fc94..b9cf764 100644 --- a/tests/drivers/test_youtube.py +++ b/tests/drivers/test_youtube.py @@ -357,12 +357,29 @@ def test_no_captions_parks_waiting_transcribe_and_never_downloads_audio(self): assert reason_of(result) == "no captions available" assert result.meta["title"] == "Unindexed Conference Talk" + def test_the_park_carries_the_description_it_already_fetched(self): + # A video that goes private during a transcription backlog must not + # take its description with it: the run layer writes this now. + driver = driver_for(INFO_WITHOUT) + body = body_of(driver.fetch(make_unit(URL, Kind.YOUTUBE))) + assert body.startswith("## Description") + assert "Recorded on a phone" in body + assert "## Transcript" not in body # the drain appends that later + + def test_a_description_less_video_still_parks_cleanly(self): + info = {k: v for k, v in INFO_WITHOUT.items() if k != "description"} + result = driver_for(info).fetch(make_unit(URL, Kind.YOUTUBE)) + assert result.status is Status.WAITING + assert result.needs is Need.TRANSCRIBE + assert result.body is None # nothing to write, nothing invented + def test_thin_captions_track_parks_waiting_transcribe(self): thin = "WEBVTT\n\n1\n00:00:00.000 --> 00:00:01.000\nhi\n" driver = driver_for(INFO_WITH, {TRACK_URL: vtt_response(thin)}) result = driver.fetch(make_unit(URL, Kind.YOUTUBE)) assert result.status is Status.WAITING assert result.needs is Need.TRANSCRIBE + assert "## Description" in body_of(result) # the description still lands class TestProbeClassification: diff --git a/tests/pipeline/test_transcribe.py b/tests/pipeline/test_transcribe.py index 9e4ed12..ab215a8 100644 --- a/tests/pipeline/test_transcribe.py +++ b/tests/pipeline/test_transcribe.py @@ -1,6 +1,7 @@ """Transcribe-drain tests: acquisition, priming, lifecycle, caps.""" import dataclasses +import json import os import re from pathlib import Path @@ -11,7 +12,7 @@ from dex_engine.capabilities import Capabilities from dex_engine.drivers.podcast import PodcastDriver from dex_engine.drivers.transport import HttpResponse, urllib_transport -from dex_engine.drivers.youtube import ProbeError, _video_meta +from dex_engine.drivers.youtube import ProbeError, YouTubeDriver, _video_meta from dex_engine.pipeline import ledger from dex_engine.pipeline import run as run_mod from dex_engine.pipeline.classify import ( @@ -153,12 +154,56 @@ def test_end_to_end_acquire_prime_write_delete(self, instance): assert audio_files(instance) == [] assert ITEM in report # the item lands on the cognitive work list + def _park_through_the_driver(self, instance) -> Path: + """Park the video the way the real driver does, and return its file.""" + write_item(instance, urls=[VIDEO_URL]) + info = json.loads(fixture_text("youtube", "info-without-captions.json")) + driver = YouTubeDriver(probe=lambda _url: info, transport=FakeTransport({})) + run_mod.run(make_ctx(instance, FakeDriver(), drivers=[driver])) + return instance.enrichment_dir / ITEM / f"youtube-{work_hash(VIDEO_URL)[:6]}.md" + + def test_the_transcript_joins_the_parks_description_in_one_file(self, instance): + # The driver's park wrote the description; the drain must append to + # it, never write over it — and the two modules must agree on the + # section headings they compose around. + park = self._park_through_the_driver(instance) + assert "Recorded on a phone" in park.read_text() + transcriber = FakeTranscriber("whisper-local", text="The transcript text.", model="medium") + run_mod.run_transcribe(transcribe_ctx(instance, transcriber=transcriber)) + content = park.read_text() + assert "Recorded on a phone" in content # the park's own description stands + assert "anydoc" not in content # not the re-probe's, which differs + assert "## Transcript\n\nThe transcript text." in content + + def test_a_re_drain_never_duplicates_the_transcript(self, instance): + park = self._park_through_the_driver(instance) + first = FakeTranscriber("whisper-local", text="First pass words.") + run_mod.run_transcribe(transcribe_ctx(instance, transcriber=first)) + ledger.append( + instance.ledger_path, + dataclasses.replace( + ledger.load(instance.ledger_path)[work_hash(VIDEO_URL)], + status=Status.WAITING, + needs=Need.TRANSCRIBE, + path=None, + title=None, + ), + ) + second = FakeTranscriber("whisper-local", text="Second pass words.") + run_mod.run_transcribe(transcribe_ctx(instance, transcriber=second)) + content = park.read_text() + assert content.count("## Transcript") == 1 + assert content.count("## Description") == 1 + assert "Recorded on a phone" in content + assert "Second pass words." in content + assert "First pass words." not in content # superseded, not stacked + def test_drained_meta_shape_matches_the_captions_path(self, instance, tmp_path): # One frontmatter shape per kind, whichever route produced the # transcript: the drain's meta keys are exactly # the captions path's. entry = seed_waiting(instance) - acquired = acquire_youtube_audio(entry, tmp_path, FakeDownload()) + acquired = acquire_youtube_audio(entry, tmp_path / "park.md", tmp_path, FakeDownload()) assert isinstance(acquired, Acquired) assert set(acquired.meta) == set(_video_meta({})) @@ -1005,7 +1050,7 @@ def download(_url, cache_dir, stem) -> YoutubeAudio: description="sponsors and links " * 200, ) - acquired = acquire_youtube_audio(entry, tmp_path, download) + acquired = acquire_youtube_audio(entry, tmp_path / "park.md", tmp_path, download) assert isinstance(acquired, Acquired) assert acquired.prompt.endswith("Ledgers at Scale — Engineering Distilled") assert estimated_tokens(acquired.prompt) <= WHISPER_WINDOW_TOKENS From 6b5be746760cfde9b4c7c2516eb7ef28ffff91d2 Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sat, 22 Aug 2026 10:42:44 +0100 Subject: [PATCH 03/21] Route indie podcast episode pages to the podcast driver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit §9 states three routes; only two existed. The third could not be built out of URL patterns — bare `/feed` and `/rss` suffixes and an RSS `` are blog vocabulary, so `matches()` guessing at them would steal ordinary posts from the web driver. Registry order supplies the answer instead. Nothing claims an indie episode page, so the catch-all fetches it; a page advertising its own audio — an `og:audio` pointer or an `