From 46e9b54cb23eeb0827f8bb9b1a64edae1b0771f0 Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sun, 23 Aug 2026 21:01:13 +0100 Subject: [PATCH 1/7] Read the never-ran side of the harvest pass record MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pass records exist so "ran and promoted nothing" is distinguishable from "never ran", but only one side had a reader: _stale_passes flags records whose rules version is old, and an item with NO harvest record was never examined — a session that skipped harvest left an item digested and cited on the strength of the shared link alone, on no surface. lint now lists "harvest these (enrichment landed, no harvest pass on record)": a live item that owes no further work, has at least one fetched page on record — a done unit that is not via media/extract-asset, the URL cap's own reading of "page" — and no harvest pass under any rules version. The bounds follow the ingest procedure's: an item still owing a unit is raw and has not reached the harvest step; a no-source capture and an item whose every unit died have no page to read the subject rule over and owe description + digest instead, which the run report already names. A pass covers its item by trailing shortid — the match exclusions already use across renames — so a renamed item's standing record never reads as a skipped judgment, and a migration reseed keeps answering to the rules-version check, not this one. The finding fires beside the digest backstop rather than deferring to it: that backstop's repair is digest → place → wiki, which never runs harvest, so deferring would carry the skipped judgment straight through the repair. Like the stale-pass row it never exits 1 — no later mechanical stage breaks on a missing pass the way the wiki layer breaks on a malformed digest, and the repair is judgment, the report's business. Lives beside digest_orphans in run.py, on the same ownership seam; torn pass lines stay lint's loud parse. Co-Authored-By: Claude Fable 5 --- design/ingestion-pipeline.md | 27 ++++++- instance/skills/dex-lint/SKILL.md | 7 ++ src/dex_engine/lint.py | 6 +- src/dex_engine/pipeline/run.py | 82 ++++++++++++++++++++++ src/dex_engine/render/surfaces.py | 8 +++ tests/test_lint.py | 112 ++++++++++++++++++++++++++++++ 6 files changed, 239 insertions(+), 3 deletions(-) diff --git a/design/ingestion-pipeline.md b/design/ingestion-pipeline.md index a417511..e675392 100644 --- a/design/ingestion-pipeline.md +++ b/design/ingestion-pipeline.md @@ -556,7 +556,7 @@ Files: | file | holds | |---|---| | `state/enrichment-ledger.jsonl` | work units (§5) | -| `state/passes.jsonl` | per-item stage records — `{stage: harvest, item, rules, date}`; "ran and promoted nothing" must be distinguishable from "never ran" | +| `state/passes.jsonl` | per-item stage records — `{stage: harvest, item, rules, date}`; "ran and promoted nothing" must be distinguishable from "never ran", and the health check reads both sides of that distinction (§10) | | `state/migrations.jsonl` | applied-migrations log (§12) | | `state/issue-reports.jsonl` | filed/commented issue fingerprints (§13) | | `state/digests/.md` | per-item fact indexes; the one markdown corner of `state/`. Claude's judgment, the engine's shape: `enrich item digest --file ` serializes it (§14). Removed with its item by `dex exclude` — the one thing that ever deletes one | @@ -1245,6 +1245,31 @@ Harvest-rule changes bump a version constant in the engine; passes are recorded in `state/passes.jsonl`; re-assessment of old items is migration-seeded (§12), not scan-inferred. +**The pass record has a reader on both sides of its distinction.** The +health check flags harvest passes recorded under an older rules version +(re-judge), and it flags harvest that **never ran**: a live item that owes +no further work, has at least one fetched page on record — a `done` unit +that is not a media download or an extracted asset, the URL cap's own +reading of "page" — and has no harvest pass under any rules version. +Without that reader, a session that skipped harvest left an item digested +and cited on the strength of the shared link alone, on no surface. The +predicate follows the ingest procedure's own bounds: an item still owing a +unit derives `raw` and has not reached the harvest step, so it is never +listed; a no-source capture, and an item whose every unit died unfetched, +has no page to read the subject rule over and owes no pass — its owed work +is description and digest, which the run report already names. A recorded +pass covers its item by trailing shortid, the same match exclusions use +across renames, so a renamed item's standing record never reads as a +skipped judgment; an item re-seeded by migration keeps its original pass +and answers to the rules-version check, not this one. The finding fires +beside the enrichment-newer-than-digest backstop rather than deferring to +it — the backstop's repair is digest → place → wiki, which never runs +harvest, so deferring would carry the skipped judgment straight through +the repair. Like the stale-pass row it is a finding, never exit 1: no +later mechanical stage breaks on a missing harvest pass the way the wiki +layer breaks on a malformed digest, and the repair — run the judgment now, +then `enrich pass --stage harvest` — is judgment, the report's business. + ## 11. Rendering: judgment decides, code renders Composition that is fully determined by data is computed in code and emitted diff --git a/instance/skills/dex-lint/SKILL.md b/instance/skills/dex-lint/SKILL.md index c3e2768..222904b 100644 --- a/instance/skills/dex-lint/SKILL.md +++ b/instance/skills/dex-lint/SKILL.md @@ -95,6 +95,13 @@ step 4. - **Possible restated facts** — read each flagged pair: same fact → merge into one sentence carrying both citations; genuinely distinct → leave them (the flag is a question, not a verdict). + - **Harvest these (no pass on record)** — the item's pages were fetched + but no harvest pass was ever recorded: "never ran" is the standing + state, not "ran and promoted nothing". Run the harvest judgment now + under the current subject rule (dex-run's `references/ingest-item.md`), + then `bin/dex enrich pass --stage harvest` — the pass is + recorded even when nothing was promoted, and recording it is what + keeps this row empty. - **Harvest passes under old rules** — re-run the harvest judgment for those items under the current subject rule (dex-run's `references/ingest-item.md`), then `bin/dex enrich pass ... --stage diff --git a/src/dex_engine/lint.py b/src/dex_engine/lint.py index 20bfd75..6f018ef 100644 --- a/src/dex_engine/lint.py +++ b/src/dex_engine/lint.py @@ -18,7 +18,8 @@ enrichment directory, where the item that owns it cannot list it — both asked of the item the corpus says owns the unit), waiting cohorts and cognitive-job - summary, harvest passes recorded under old rules, and the + summary, items whose fetched pages landed but whose harvest pass was + never recorded, harvest passes recorded under old rules, and the enrichment-newer-than-digest orphan listing (the interrupted-session backstop, shared with ``enrich status``). @@ -63,7 +64,7 @@ from .pipeline import ledger from .pipeline.ownership import corpus_owners from .pipeline.registry import DRIVERS -from .pipeline.run import CAP_BOUNDS, HARVEST_RULES_VERSION, digest_orphans +from .pipeline.run import CAP_BOUNDS, HARVEST_RULES_VERSION, digest_orphans, never_harvested from .pipeline.transcribe import read_enrichment_fields from .pipeline.types import Config, Format, Instance, LedgerEntry, Need, Status from .render import surfaces @@ -551,6 +552,7 @@ def _state_checks( payload["missing_outputs"] = integrity.missing payload["misfiled_outputs"] = integrity.misfiled payload["capped"] = _cap_fires(entries) + payload["never_harvested"] = never_harvested(instance) payload["stale_passes"] = _stale_passes(instance) threads = _incomplete_threads(instance) payload["incomplete_threads"] = threads.rows diff --git a/src/dex_engine/pipeline/run.py b/src/dex_engine/pipeline/run.py index 918ae71..6b4a65f 100644 --- a/src/dex_engine/pipeline/run.py +++ b/src/dex_engine/pipeline/run.py @@ -90,6 +90,7 @@ "head_sniffer", "is_drainable", "mark", + "never_harvested", "no_providers", "record_pass", "run", @@ -2298,6 +2299,87 @@ def _last_digested(instance: Instance) -> dict[str, datetime.date]: return newest +def never_harvested(instance: Instance) -> list[str]: + """Items whose fetched pages landed with no harvest pass ever recorded. + + A recorded pass is what makes "ran and promoted nothing" distinguishable + from "never ran", and this is the reader of the distinction: the + stale-pass check reads only the records that exist, so an item whose + session skipped harvest outright was digested and cited on the strength + of the shared link alone, on no surface at all. + + Who owes a pass is bounded the way the ingest procedure bounds the + step. An item still owing a unit derives ``raw`` and has not reached + harvest, so it is never listed, however long it stays parked — the same + rule :func:`digest_orphans` applies. An item with no fetched page at + all — a no-source capture, or one whose every unit died unfetched — has + no pages to read the subject rule over; its work is description and + digest, and it owes no pass. A fetched page is a ``done`` unit that is + not a media download or an extracted asset, the same reading the URL + cap counts (:meth:`_Drain.fetched_count`). + + A recorded pass covers an item by the id's trailing shortid, the + trailing-id match exclusions already use across renames: a rename keeps + the shortid and rewrites the slug, so a full-id match would report + every renamed item's harvest as never run while the record stands in + the file under the old name. + + An item on the digest backstop is listed here too when both hold: the + two findings are different facts with different repairs — the + backstop's repair is digest → place → wiki, which never runs harvest — + so each check answers for itself, as the referential-integrity rows do. + + Args: + instance: The instance. + + Returns: + Item ids owing a harvest pass that no record covers, sorted. + """ + entries = _ledger_or_none(instance) + if not entries: + return [] + owners = _unit_owners(instance, entries) + owing = _items_owing_work(entries, owners) + live = {path.stem for path in instance.corpus_dir.glob("*/*.md")} + fetched = { + item_id + for entry in entries.values() + if entry.status is Status.DONE and entry.via not in ("media", "extract-asset") + for item_id in owners.get(entry.hash, (entry.item,)) + } + recorded = _harvested_shortids(instance) + return sorted( + item_id + for item_id in fetched & live + if item_id not in owing and item_id.rsplit("-", 1)[-1] not in recorded + ) + + +def _harvested_shortids(instance: Instance) -> set[str]: + """The trailing shortid of every item a harvest pass was recorded for. + + Any rules version counts: a pass under old rules RAN, which is the + stale-pass finding, never this one's. + """ + path = instance.passes_path + if not path.exists(): + return set() + recorded: set[str] = set() + 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; this reader skips the line + if not isinstance(record, dict) or record.get("stage") != "harvest": + continue + item = record.get("item") + if isinstance(item, str): + recorded.add(item.rsplit("-", 1)[-1]) + return recorded + + def _writes_that_did_not_land( instance: Instance, written: Mapping[str, LedgerEntry] ) -> dict[str, LedgerEntry | None]: diff --git a/src/dex_engine/render/surfaces.py b/src/dex_engine/render/surfaces.py index ee74456..ccc6998 100644 --- a/src/dex_engine/render/surfaces.py +++ b/src/dex_engine/render/surfaces.py @@ -982,6 +982,7 @@ def _render_ingest_receipt(payload: Mapping[str, object]) -> str: "misfiled_outputs", "waiting", "cognitive", + "never_harvested", "stale_passes", "capped", "incomplete_threads", @@ -1040,6 +1041,7 @@ def _render_health_report(payload: Mapping[str, object]) -> str: "misfiled_outputs": [{"item": str, "path": str}], "waiting": {"": int}, "cognitive": [{"item": str, "url": str, "need": str}], + "never_harvested": [str], # fetched pages landed, no pass on record "stale_passes": [{"item": str, "rules": int}], # judgment drift — recorded for this surface, shown on no other "capped": [{"item": str, "url": str, "reason": str}], # re-entry cap fires @@ -1266,6 +1268,12 @@ def _health_state(surface: str, payload: Mapping[str, object]) -> list[str]: blocks.append(kernel.bullet(kernel.inline([kernel.bold(item), kernel.code(need)]), depth=1)) blocks.append(kernel.detail(url, depth=1)) blocks += _health_elided(len(cognitive)) + blocks += _health_names( + surface, + payload, + "never_harvested", + "harvest these (enrichment landed, no harvest pass on record)", + ) stale_passes = _health_rows(surface, payload, "stale_passes", ("item",), int_keys=("rules",)) blocks += _health_listing( "harvest passes under old rules (re-judge)", diff --git a/tests/test_lint.py b/tests/test_lint.py index 72afbd8..f55d4ef 100644 --- a/tests/test_lint.py +++ b/tests/test_lint.py @@ -747,6 +747,118 @@ def test_the_superseded_line_is_not_the_one_checked(self, instance): assert "done entries whose output file is gone from disk — none" in outcome.report +NEVER_HARVESTED = "harvest these (enrichment landed, no harvest pass on record)" + + +class TestNeverHarvested: + """Never-ran must be told apart from ran-and-promoted-nothing — and read.""" + + def _bare_wiki(self, instance): + write_taxonomy(instance) + write_index(instance, "") + + def _record_pass(self, instance, item_id: str = ITEM, rules: int = 1) -> None: + record = {"stage": "harvest", "item": item_id, "rules": rules, "date": "2026-08-19"} + instance.passes_path.write_text(json.dumps(record) + "\n") + + def test_a_never_run_harvest_is_a_finding_not_a_failure(self, instance): + # The repair — run the judgment now, then record the pass — is the + # report's business, so the finding never exits 1. + self._bare_wiki(instance) + write_corpus_stub(instance) + ledger.append(instance.ledger_path, done_entry("73bd784849")) + outcome = lint(instance) + assert f"{NEVER_HARVESTED} — **1**" in outcome.report + assert f"**{ITEM}**" in outcome.report + assert outcome.exit_code == 0 + + def test_ran_and_promoted_nothing_never_fires(self, instance): + # The pass record IS the distinction: one on file means harvest ran, + # however little it promoted. + self._bare_wiki(instance) + write_corpus_stub(instance) + ledger.append(instance.ledger_path, done_entry("73bd784849")) + self._record_pass(instance) + outcome = lint(instance) + assert f"{NEVER_HARVESTED} — none" in outcome.report + + def test_a_pass_under_old_rules_ran_and_is_the_other_finding(self, instance): + self._bare_wiki(instance) + write_corpus_stub(instance) + ledger.append(instance.ledger_path, done_entry("73bd784849")) + self._record_pass(instance, rules=0) + outcome = lint(instance) + assert f"{NEVER_HARVESTED} — none" in outcome.report + assert "harvest passes under old rules (re-judge) — **1**" in outcome.report + + def test_an_item_still_owing_a_unit_is_not_yet_owing_harvest(self, instance): + # One outstanding unit derives `raw`, and a raw item has not reached + # the harvest step — however long it stays parked. + self._bare_wiki(instance) + write_corpus_stub(instance) + ledger.append(instance.ledger_path, done_entry("73bd784849")) + ledger.append(instance.ledger_path, stamped(waiting_entry(Need.TRANSCRIBE, "aaaaaaaaaa"))) + outcome = lint(instance) + assert f"{NEVER_HARVESTED} — none" in outcome.report + + def test_a_no_source_item_owes_no_pass(self, instance): + # A text/image-only capture seeds no units: nothing was fetched, so + # there are no pages to read the subject rule over — its owed work + # is description and digest, and the run report lists exactly that. + self._bare_wiki(instance) + write_corpus_stub(instance) + outcome = lint(instance) + assert f"{NEVER_HARVESTED} — none" in outcome.report + + def test_a_media_only_landing_is_not_a_fetched_page(self, instance): + # Media downloads and extracted assets are bytes, not pages — the + # same reading the URL cap counts by — so they alone never put an + # item on the hook. + self._bare_wiki(instance) + write_corpus_stub(instance) + entry = LedgerEntry( + hash="73bd784849", + url="https://example.test/photo.jpg", + item=ITEM, + kind=Kind.WEB, + status=Status.DONE, + engine="0.1.0", + date=TODAY, + via="media", + parent="aaaaaaaaaa", + depth=1, + ) + ledger.append(instance.ledger_path, entry) + outcome = lint(instance) + assert f"{NEVER_HARVESTED} — none" in outcome.report + + def test_a_renamed_items_pass_still_covers_it(self, instance): + # Same shortid, new slug: the pass stands in the file under the old + # id, and a full-id match would call the judgment never-run. + self._bare_wiki(instance) + renamed = "2026-08-19-example-renamed-55ad7b" + url = "https://example.test/moved" + write_corpus_item(instance, renamed, urls=[url]) + ledger.append(instance.ledger_path, done_entry(work_identity(url, DRIVERS), url=url)) + self._record_pass(instance, item_id=ITEM) # the pre-rename id, same 55ad7b + outcome = lint(instance) + assert f"{NEVER_HARVESTED} — none" in outcome.report + + def test_fires_beside_the_digest_backstop_not_behind_it(self, instance): + # An interrupted session shows on "digest these", whose repair is + # digest → place → wiki — harvest is not in that path, so deferring + # here would carry the skipped judgment straight through the repair. + self._bare_wiki(instance) + write_corpus_stub(instance) + ledger.append(instance.ledger_path, done_entry("73bd784849")) + item_dir = instance.enrichment_dir / ITEM + item_dir.mkdir(parents=True) + (item_dir / "web-73bd78.md").write_text("landed, session died") + outcome = lint(instance) + assert f"{NEVER_HARVESTED} — **1**" in outcome.report + assert "digest these (enrichment newer than digest) — **1**" in outcome.report + + def capped_entry( unit_hash: str, *, item: str = ITEM, cap: Cap, url: str, forced: bool = False ) -> LedgerEntry: From 6753ed6d822c9e58f66ab2e1b4e649c1b44e2154 Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sun, 23 Aug 2026 21:37:39 +0100 Subject: [PATCH 2/7] Send the run backstop through the whole remaining procedure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dex-run step 8 backstop told the session to complete an interrupted item's "digest → place → wiki steps now" — a repair that skips harvest, the step that comes before digest in the per-item procedure. An interruption can predate harvest just as easily as digest, and the `enrich status` surface the step reads has no never-harvested row, so a harvest skipped by this path stayed invisible until the next health check flagged it (the lint check this branch already carries). The lint catch is the backstop's backstop; the instruction must stop causing the state it catches. The step now sends the session through the full remaining per-item procedure — harvest → digest → place → wiki, per references/ingest-item.md — and says why: the listing states that a digest is owed, never that harvest ran. The dex-lint fix-list's own "finish digest → place → wiki" line is left as it stands: the lint report lists a skipped harvest as its own "harvest these" row beside the digest row, so a lint session covers harvest through that row rather than through this one. Co-Authored-By: Claude Fable 5 --- instance/skills/dex-run/SKILL.md | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/instance/skills/dex-run/SKILL.md b/instance/skills/dex-run/SKILL.md index 138a55e..0629778 100644 --- a/instance/skills/dex-run/SKILL.md +++ b/instance/skills/dex-run/SKILL.md @@ -109,8 +109,12 @@ Then do a run now, whatever the host. act on nothing. 8. **Backstop.** `bin/dex enrich status` — any item listed under **Digest - these** is an interrupted previous session: complete its digest → - place → wiki steps now. Every item listed is digestible; one still owing + these** is an interrupted previous session: complete its remaining + per-item steps now, harvest → digest → place → wiki + (`references/ingest-item.md`). The listing says a digest is owed, not + that harvest ran — the interruption can predate either step, and no + surface here would catch a skipped harvest before the next health + check. Every item listed is digestible; one still owing a unit is `raw` and never appears there, however long it stays parked. **Needs you** and **Waiting on the engine** here are the standing view of everything parked in the instance, not just this session's: judge the From ff52994bc2d7e2f3adbf98d48707318bbf0d4574 Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sun, 23 Aug 2026 21:38:04 +0100 Subject: [PATCH 3/7] Stop promising a synthesis refresh nothing performs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit dex-query told the owner that syntheses are refreshed by the health check as new material lands. Nothing does that: lint's staleness check reads topic pages against their taxonomy members, and a synthesis has no topic, so it is never compared against anything. Nor will such a check be built — a synthesis is question-shaped and dated on purpose, and a mechanical check cannot tell one gone stale from one deliberately scoped to its date; that distinction is judgment. The skill now says what is true: refreshing happens in the query path itself, when a session reads a synthesis whose generated: date predates material that changes its answer, and rewrites it. The design doc makes no synthesis-refresh claim, so it needed no correction. Co-Authored-By: Claude Fable 5 --- instance/skills/dex-query/SKILL.md | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/instance/skills/dex-query/SKILL.md b/instance/skills/dex-query/SKILL.md index 985a285..a44119b 100644 --- a/instance/skills/dex-query/SKILL.md +++ b/instance/skills/dex-query/SKILL.md @@ -29,5 +29,8 @@ full fetched source — and summarize from that. page), file it as `wiki/syntheses/.md` — frontmatter `type: synthesis`, `question:`, `generated:` date; body cites item ids and wikilinks related topics. Add to index, log it, commit and push. Syntheses are question-shaped, dated - snapshots; the health check refreshes them as new material lands. + snapshots; no mechanical check refreshes one — a check cannot tell a + stale synthesis from a deliberately time-scoped one. Refreshing is + judgment, done here: when a query reads a synthesis whose `generated:` + date predates material that changes its answer, rewrite it. 6. Never cite wiki pages as sources — item ids only. From be6a17e8d7f57df9abba7fe6217882fbfe000d59 Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sun, 23 Aug 2026 21:44:18 +0100 Subject: [PATCH 4/7] Record the digest pass in the digest verb itself MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Writing a digest and recording its pass were two commands, and only the first had a product the session could see. A forgotten `enrich pass --stage digest` cost the staleness backstop its comparand silently: a digest with no pass record is dated by nothing, so no later enrichment ever reads as newer than it, and the item escapes "Digest these" forever. One motion, one verb: `enrich item digest` now records the pass itself, through the same record_pass path the CLI's pass verb calls — one pass-writing path, not two. Ordering is chosen by which crash residue has a reader. The record goes in after validation and before the file write: a refused payload records nothing, and a crash between the two writes leaves a pass with no digest file, which digest_orphans' no-digest branch lists loudly. The opposite order leaves a digest with no pass — the very unreadable state this change removes — so the record goes first. The confirmation line says the pass was recorded, so a session never doubts it into a duplicate `enrich pass`. The skills drop the separate digest pass step (the ingest procedure's §6 now says the verb records it; dex-run's state-writes rule and the contract say the same), while `enrich pass --stage digest` stays as the manual re-record and the harvest/wiki recorder. Design §4/§10/§14 updated to the same account. Updated assertions: test_writes_the_canonical_digest, test_a_rewrite_says_so_and_replaces_ every_field and test_item_digest_writes_the_digest now expect the "digest pass recorded" confirmation. Co-Authored-By: Claude Fable 5 --- README.md | 2 +- design/ingestion-pipeline.md | 19 +++++- instance/dex-contract.md | 5 +- instance/skills/dex-run/SKILL.md | 5 +- .../skills/dex-run/references/ingest-item.md | 3 +- .../dex-run/references/state-formats.md | 4 +- src/dex_engine/enrich.py | 6 +- src/dex_engine/pipeline/digest.py | 31 +++++++-- tests/pipeline/test_digest.py | 63 ++++++++++++++++++- tests/test_enrich.py | 6 +- 10 files changed, 124 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index e0103ba..5a4a5fc 100644 --- a/README.md +++ b/README.md @@ -276,7 +276,7 @@ cwd = instance root; the tag lives in `.dex-engine-pin`, bumped by sync): | `dex-enrich mark` | heal one ledger entry: the sanctioned correction verb | | `dex-enrich pass` | record a stage completion (harvest/digest/wiki) in `state/passes.jsonl` | | `dex-enrich item new` | create a corpus item from a capture file (id rules and provenance; code writes frontmatter) | -| `dex-enrich item digest` | write an item's digest from a JSON payload — signal, topics and facts are the judgment, the file's shape is the engine's | +| `dex-enrich item digest` | write an item's digest from a JSON payload — signal, topics and facts are the judgment, the file's shape is the engine's; the digest pass is recorded in the same call | | `dex-normalize` | raw chat exports to corpus items (DiscordChatExporter JSON) | | `dex-lint` | mechanical health check: wikilinks, citations (shortid flags included), orphans, index drift, stale pages, count drift, restated-fact warnings, ledger schema, ledger↔corpus integrity, cap fires, thread-completeness markers, digest shape, pass records (`--write` reconciles derived wiki frontmatter) | | `dex-exclude ` | permanently purge out-of-scope items — corpus file, enrichment, ledger entries — surviving re-normalization | diff --git a/design/ingestion-pipeline.md b/design/ingestion-pipeline.md index e675392..808255c 100644 --- a/design/ingestion-pipeline.md +++ b/design/ingestion-pipeline.md @@ -556,7 +556,7 @@ Files: | file | holds | |---|---| | `state/enrichment-ledger.jsonl` | work units (§5) | -| `state/passes.jsonl` | per-item stage records — `{stage: harvest, item, rules, date}`; "ran and promoted nothing" must be distinguishable from "never ran", and the health check reads both sides of that distinction (§10) | +| `state/passes.jsonl` | per-item stage records — `{stage: harvest, item, rules, date}`; "ran and promoted nothing" must be distinguishable from "never ran", and the health check reads both sides of that distinction (§10). The digest pass is recorded by `enrich item digest` itself, in the same call as the file (§10); `enrich pass` records the harvest and wiki stages, and remains the manual re-record for any stage | | `state/migrations.jsonl` | applied-migrations log (§12) | | `state/issue-reports.jsonl` | filed/commented issue fingerprints (§13) | | `state/digests/.md` | per-item fact indexes; the one markdown corner of `state/`. Claude's judgment, the engine's shape: `enrich item digest --file ` serializes it (§14). Removed with its item by `dex exclude` — the one thing that ever deletes one | @@ -1270,6 +1270,19 @@ later mechanical stage breaks on a missing harvest pass the way the wiki layer breaks on a malformed digest, and the repair — run the judgment now, then `enrich pass --stage harvest` — is judgment, the report's business. +**The digest pass is recorded by the digest verb, not by a second +command.** `enrich item digest` records the pass itself, through the same +`record_pass` path `enrich pass` uses — a separate recording command was a +step a session could forget, and a forgotten one silently cost the +staleness backstop its comparand: a digest with no pass record is dated by +nothing, so no later enrichment could ever read as newer than it. The +verb records after validation and before the file write, so both failure +shapes stay honest: a refused payload records nothing, and a crash +between the record and the write leaves a pass with no digest file — +which the backstop's no-digest branch lists loudly — never the unreadable +opposite. `enrich pass --stage digest` remains the manual re-record, and +the harvest and wiki stages still record through `enrich pass`. + ## 11. Rendering: judgment decides, code renders Composition that is fully determined by data is computed in code and emitted @@ -1863,7 +1876,9 @@ src/dex_engine/ total validation — a missing, unknown or mistyped key, a `signal` outside the vocabulary, empty `topics` or `facts`, an id naming no corpus item — and nothing is - written on a refusal. Rewriting is allowed and carries + written on a refusal, no pass record included. Records + the digest pass itself, before the file write (§10). + Rewriting is allowed and carries nothing over: every field is the payload's judgment or the corpus item's fact) normalize.py imports shared detect/types (private kind_of copy deleted) diff --git a/instance/dex-contract.md b/instance/dex-contract.md index 5ed2366..154d8aa 100644 --- a/instance/dex-contract.md +++ b/instance/dex-contract.md @@ -66,8 +66,9 @@ never state, never synced. is append-only and verb-written: ledger heals via `bin/dex enrich mark`, stage records via `bin/dex enrich pass`. Digests too: the judgment goes in as JSON and `bin/dex enrich item digest --file ` writes the - file. A hand-appended line is how state and reality diverge; the verbs - are what make a malformed record impossible. + file, recording the digest pass in the same call. A hand-appended line + is how state and reality diverge; the verbs are what make a malformed + record impossible. - `wiki/` is a build artifact: regenerable, never the only home of a fact. Pages cite corpus item ids in backticks — full ids, ALWAYS, and NEVER other wiki pages — so citations stay mechanically checkable. diff --git a/instance/skills/dex-run/SKILL.md b/instance/skills/dex-run/SKILL.md index 0629778..493586c 100644 --- a/instance/skills/dex-run/SKILL.md +++ b/instance/skills/dex-run/SKILL.md @@ -149,8 +149,9 @@ Then do a run now, whatever the host. Where a state file has a verb, the verb writes it and you never do: corpus items come from `enrich item new`, digests from `enrich item digest ---file cache/digest.json`, ledger lines from `enrich mark` and the run -itself, stage records from `enrich pass`. You supply the judgment as JSON +--file cache/digest.json` (which records the digest pass itself), ledger +lines from `enrich mark` and the run itself, harvest and wiki stage +records from `enrich pass`. You supply the judgment as JSON or arguments; the engine decides the shape, so a malformed file cannot be written. Same motion as rendering, below. `state/taxonomy.json` is the one you still write directly. diff --git a/instance/skills/dex-run/references/ingest-item.md b/instance/skills/dex-run/references/ingest-item.md index c25c636..431bc77 100644 --- a/instance/skills/dex-run/references/ingest-item.md +++ b/instance/skills/dex-run/references/ingest-item.md @@ -141,7 +141,8 @@ tweet earns two, and padding to a number invents facts. Interpretive context lives here, not in item bodies. Topics: canonical names from `state/taxonomy.json` when it exists; otherwise 2–5 kebab-case candidates. Revising a digest later is the same call with a new payload — the file is -rewritten whole. Then `bin/dex enrich pass --stage digest`. +rewritten whole. The verb records the digest pass itself — the +confirmation says so — so there is no separate pass command to run here. A parked item (waiting/blocked/manual) still exists — provenance and note were captured at ingest — but gets no digest or wiki work until its diff --git a/instance/skills/dex-run/references/state-formats.md b/instance/skills/dex-run/references/state-formats.md index 1284f3c..cf7f622 100644 --- a/instance/skills/dex-run/references/state-formats.md +++ b/instance/skills/dex-run/references/state-formats.md @@ -185,7 +185,9 @@ a malformed record impossible: - `state/passes.jsonl` — per-item stage records `{stage, item, date, rules?}` ("ran and promoted nothing" is distinguishable from "never ran"; `rules` versions the harvest rules). Written by - `bin/dex enrich pass`. + `bin/dex enrich pass`; the digest pass is recorded by `enrich item + digest` itself, in the same call as the file (`enrich pass --stage + digest` remains the manual re-record). - `state/migrations.jsonl` — applied-migrations log `{number, engine, date}`. Written by sync's migration runner. - `state/issue-reports.jsonl` — what this instance filed/commented diff --git a/src/dex_engine/enrich.py b/src/dex_engine/enrich.py index 97ca9e9..0503004 100644 --- a/src/dex_engine/enrich.py +++ b/src/dex_engine/enrich.py @@ -131,7 +131,9 @@ def build_parser() -> argparse.ArgumentParser: help="provenance display name (default: owner)", ) digest_parser = item_commands.add_parser( - "digest", help="write an item's digest from a JSON payload (judgment in, shape decided)" + "digest", + help="write an item's digest from a JSON payload (judgment in, shape decided) " + "and record the digest pass", ) digest_parser.add_argument( "--file", @@ -168,7 +170,7 @@ def _dispatch(args: argparse.Namespace, ctx: RunContext) -> str: # noqa: PLR091 case "pass": return record_pass(ctx, args.item, args.stage) case "item" if args.item_command == "digest": - return item_digest(args.file, instance=ctx.instance) + return item_digest(args.file, ctx=ctx) case "item": return item_new( args.capture, diff --git a/src/dex_engine/pipeline/digest.py b/src/dex_engine/pipeline/digest.py index 76741f9..f2b5571 100644 --- a/src/dex_engine/pipeline/digest.py +++ b/src/dex_engine/pipeline/digest.py @@ -16,6 +16,15 @@ again — and nothing carries over from the file being replaced. Every field is either the payload's judgment or the corpus item's fact, so a rewrite is a full re-derivation and there is nothing only the old file knew. + +The verb also records the item's digest pass — the comparand the +staleness backstop reads — through the same seam ``enrich pass`` uses, so +a digest cannot land unrecorded. The record goes in after validation and +before the file write: a refused payload records nothing, and a crash +between the two writes leaves a pass with no digest file, which the +backstop's no-digest branch lists loudly. The other residue — a digest +with no pass — has no reader at all, and is exactly the silently +uncomparable state this recording exists to end. """ import json @@ -25,6 +34,7 @@ from dex_engine import atomic, corpus +from .run import RunContext, record_pass from .types import Instance __all__ = ["SIGNALS", "DigestPayloadError", "item_digest"] @@ -47,13 +57,21 @@ class DigestPayloadError(ValueError): """A digest payload does not conform to the shape the verb writes.""" -def item_digest(payload_path: Path, *, instance: Instance) -> str: - """Write one item's digest from its JSON payload. +def item_digest(payload_path: Path, *, ctx: RunContext) -> str: + """Write one item's digest from its JSON payload, recording the pass. + + The digest pass goes through :func:`record_pass` — the one + pass-writing path — after the payload survives validation and before + the file lands: a refused payload records nothing, and a crash + between the record and the write leaves a pass with no digest file, + the state the staleness backstop's no-digest branch reports loudly. + Writing the file first would leave the opposite residue, a digest no + record dates, which no reader ever flags. Args: payload_path: The JSON payload (conventionally under ``cache/``): ``{"id", "signal", "topics", "facts"}``, ``entities`` optional. - instance: The instance. + ctx: The run context. Returns: A one-line confirmation naming the file written. @@ -61,10 +79,12 @@ def item_digest(payload_path: Path, *, instance: Instance) -> str: Raises: DigestPayloadError: The payload is not a JSON object, a key is missing, unknown, engine-owned or the wrong type, or the id - names no readable corpus item. Nothing is written. + names no readable corpus item. Nothing is written and no pass + is recorded. OSError: The payload file cannot be read, or the digest cannot be written. """ + instance = ctx.instance payload = _load(payload_path) item_id = _text(payload_path, payload, "id") if not _ITEM_ID_RE.fullmatch(item_id): @@ -83,13 +103,14 @@ def item_digest(payload_path: Path, *, instance: Instance) -> str: facts = _facts(payload_path, payload) item = _corpus_item(payload_path, instance, item_id) + record_pass(ctx, item_id, "digest") target = instance.digests_dir / f"{item_id}.md" rewrite = target.exists() target.parent.mkdir(parents=True, exist_ok=True) atomic.write_text( target, _serialize(item, signal=signal, topics=topics, entities=entities, facts=facts) ) - return f"{'rewrote' if rewrite else 'wrote'} state/digests/{item_id}.md" + return f"{'rewrote' if rewrite else 'wrote'} state/digests/{item_id}.md · digest pass recorded" def _corpus_item(payload_path: Path, instance: Instance, item_id: str) -> corpus.CorpusItem: diff --git a/tests/pipeline/test_digest.py b/tests/pipeline/test_digest.py index ecd5b0b..86a72de 100644 --- a/tests/pipeline/test_digest.py +++ b/tests/pipeline/test_digest.py @@ -7,6 +7,9 @@ from dex_engine import corpus from dex_engine.pipeline.digest import DigestPayloadError, item_digest +from dex_engine.pipeline.run import digest_orphans +from tests.conftest import FakeDriver +from tests.pipeline.test_run import make_ctx from tests.test_lint import lint, write_index, write_taxonomy ITEM = "2026-08-19-example-55ad7b" @@ -45,7 +48,7 @@ def digest(instance, payload=None, **overrides) -> str: path = instance.cache_dir / "digest.json" body = {**PAYLOAD, **overrides} if payload is None else payload path.write_text(json.dumps(body) if not isinstance(body, str) else body) - return item_digest(path, instance=instance) + return item_digest(path, ctx=make_ctx(instance, FakeDriver())) def digest_text(instance, item_id: str = ITEM) -> str: @@ -55,7 +58,7 @@ def digest_text(instance, item_id: str = ITEM) -> str: class TestWrite: def test_writes_the_canonical_digest(self, instance): write_item(instance) - assert digest(instance) == f"wrote state/digests/{ITEM}.md" + assert digest(instance) == f"wrote state/digests/{ITEM}.md · digest pass recorded" assert digest_text(instance) == ( f"---\nid: {ITEM}\ndate: 2026-08-19\nsignal: high\n" "topics: [agentic-engineering, claude]\n---\n" @@ -114,7 +117,7 @@ def test_a_rewrite_says_so_and_replaces_every_field(self, instance): # Nothing carries over: each field is the payload's judgment or the # corpus item's fact, so the second write is a full re-derivation. assert digest(instance, signal="low", topics=["brewing"], facts=["one fact."]) == ( - f"rewrote state/digests/{ITEM}.md" + f"rewrote state/digests/{ITEM}.md · digest pass recorded" ) assert digest_text(instance) == ( f"---\nid: {ITEM}\ndate: 2026-08-19\nsignal: low\ntopics: [brewing]\n---\n- one fact.\n" @@ -129,6 +132,60 @@ def test_a_refused_rewrite_leaves_the_standing_digest_alone(self, instance): assert digest_text(instance) == before +class TestPassRecording: + """The verb records the digest pass itself — no separate command to forget.""" + + def _pass_records(self, instance) -> list[dict]: + if not instance.passes_path.exists(): + return [] + return [ + json.loads(line) + for line in instance.passes_path.read_text().split("\n") + if line.strip() + ] + + def test_the_write_records_the_digest_pass(self, instance): + # The pass record is the staleness backstop's comparand; a digest + # that landed without one was invisible to it forever. + write_item(instance) + digest(instance) + assert self._pass_records(instance) == [ + {"stage": "digest", "item": ITEM, "date": "2026-08-20"} + ] + + def test_a_rewrite_records_a_fresh_pass(self, instance): + write_item(instance) + digest(instance) + digest(instance, signal="low") + assert [r["stage"] for r in self._pass_records(instance)] == ["digest", "digest"] + + def test_a_refused_payload_records_nothing(self, instance): + write_item(instance) + with pytest.raises(DigestPayloadError): + digest(instance, signal="urgent") + assert self._pass_records(instance) == [] + + def test_the_record_lands_before_the_file_so_a_crash_reads_loud(self, instance, monkeypatch): + # A crash between the two writes must leave the state a reader + # flags: a pass with no digest file is the backstop's no-digest + # orphan, while a digest with no pass is dated by nothing and + # silent forever — so the record goes first. + write_item(instance) + item_dir = instance.enrichment_dir / ITEM + item_dir.mkdir(parents=True) + (item_dir / "web-abc123.md").write_text("enrichment on disk\n") + + def refuse(_path, _text): + raise OSError("No space left on device") + + monkeypatch.setattr("dex_engine.atomic.write_text", refuse) + with pytest.raises(OSError, match="No space left"): + digest(instance) + assert [r["stage"] for r in self._pass_records(instance)] == ["digest"] + assert not (instance.digests_dir / f"{ITEM}.md").exists() + assert digest_orphans(instance) == [ITEM] # the residue is a loud state + + class TestPayloadValidation: def test_a_bogus_signal_is_refused(self, instance): write_item(instance) diff --git a/tests/test_enrich.py b/tests/test_enrich.py index 2f79166..59eafae 100644 --- a/tests/test_enrich.py +++ b/tests/test_enrich.py @@ -143,8 +143,12 @@ def test_item_digest_writes_the_digest(self, instance, monkeypatch, capsys): ) capsys.readouterr() main(["item", "digest", "--file", str(payload)]) - assert capsys.readouterr().out == f"wrote state/digests/{item_id}.md\n" + assert capsys.readouterr().out == ( + f"wrote state/digests/{item_id}.md · digest pass recorded\n" + ) assert (instance.digests_dir / f"{item_id}.md").read_text().startswith("---\n") + record = json.loads(instance.passes_path.read_text().split("\n")[0]) + assert (record["stage"], record["item"]) == ("digest", item_id) def test_a_refused_digest_payload_exits_with_a_stated_line(self, instance, monkeypatch): monkeypatch.chdir(instance.root) From 2849271423db9c7a91763275972150bba8ce498c Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sun, 23 Aug 2026 21:47:27 +0100 Subject: [PATCH 5/7] Read the cited-but-unplaced face of the coverage invariant MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit state-formats.md states the invariant: every corpus item appears in at least one topic's items or in uncategorized-shares. lint only ever read one face of it — orphans fires on an item that is uncited AND unledgered — so an item a page cites that no topic records was invisible on every surface: a citation is not a placement, and nothing asked the placement question about the cited. The check shares the derivations the orphan logic already computes: the cited set is the wiki scan's, the uncategorized set stays the orphan check's own, and the new placed set is the union of every topic's items (uncategorized-shares included, being a topic in the same dict), read with the same tolerance for hand-mangled shapes. It renders beside the orphan row in the wiki section, capped and elided like every listing, and like orphans it is a finding, never exit 1 — the repair is judgment. The dex-lint fix-list names it: append the id to each matching topic's items, or ledger it into uncategorized-shares. Co-Authored-By: Claude Fable 5 --- instance/skills/dex-lint/SKILL.md | 5 ++++ src/dex_engine/lint.py | 29 ++++++++++++++++++++++- src/dex_engine/render/surfaces.py | 5 ++++ tests/test_lint.py | 39 +++++++++++++++++++++++++++++++ 4 files changed, 77 insertions(+), 1 deletion(-) diff --git a/instance/skills/dex-lint/SKILL.md b/instance/skills/dex-lint/SKILL.md index 222904b..75e5cd2 100644 --- a/instance/skills/dex-lint/SKILL.md +++ b/instance/skills/dex-lint/SKILL.md @@ -87,6 +87,11 @@ step 4. digest: cite it on the best existing page, or ledger it into `uncategorized-shares` in `state/taxonomy.json` if genuinely low-signal. + - **Items a page cites but no taxonomy topic records** — the other face + of the coverage invariant: a citation is not a placement. Read the + digest and append the id to each matching topic's `items` in + `state/taxonomy.json`, or ledger it into `uncategorized-shares` if + genuinely low-signal. - **Index drift** — regenerate the affected `wiki/index.md` entries. - **Stale pages** (members newer than the page) — fold the newer items in via rewrite-not-append; if the new material supersedes old claims, diff --git a/src/dex_engine/lint.py b/src/dex_engine/lint.py index 6f018ef..a1d6322 100644 --- a/src/dex_engine/lint.py +++ b/src/dex_engine/lint.py @@ -4,7 +4,9 @@ wiki — broken wikilinks (vs reserved/unbuilt), citations of ids not in the corpus, shortid-shaped citations (backticked 6-hex is a probable malformed - citation everywhere, index included), coverage orphans, index consistency, + citation everywhere, index included), coverage orphans and their cited + complement (items pages cite that no taxonomy topic records), index + consistency, stale pages, page item-count drift (frontmatter ``items:`` vs the page's MEMBER count — its taxonomy topic's items, or its entity-members list; never its citation count), and difflib sentence similarity ("possible restated @@ -155,6 +157,12 @@ def run_lint( ledgered = _uncategorized_items(taxonomy) orphans = sorted(corpus_ids - scan.cited - ledgered) + # The coverage invariant's other face: a citation is not a placement, + # so a cited item can still be in no topic's items and off the + # uncategorized ledger — invisible to the orphan check, which only + # asks about the uncited. The cited set is the wiki scan's; the + # placed set is the taxonomy's own listing. + unplaced = sorted(scan.cited - _placed_items(taxonomy)) unindexed, ghost_index = _index_consistency(instance, pages, taxonomy) # Shortid-shaped citations flag everywhere — index included: latent # shortids in an index never tripped the old citation check because no @@ -177,6 +185,7 @@ def run_lint( "bad_citations": scan.bad_citations, "shortid_citations": scan.shortid_citations, "orphans": orphans, + "unplaced": unplaced, "unindexed": unindexed, "ghost_index": ghost_index, "stale_pages": scan.stale_pages, @@ -230,6 +239,24 @@ def _uncategorized_items(taxonomy: dict[str, object]) -> set[str]: return {item for item in items if isinstance(item, str)} +def _placed_items(taxonomy: dict[str, object]) -> set[str]: + """Every item id some topic's ``items`` records — uncategorized-shares included. + + The coverage invariant's whole placed set, read with the same + tolerance for hand-mangled shapes :func:`_uncategorized_items` shows: + a malformed topic contributes nothing rather than a crash. + """ + topics = taxonomy.get("topics", {}) + if not isinstance(topics, dict): + return set() + placed: set[str] = set() + for topic in topics.values(): + items = topic.get("items", []) if isinstance(topic, dict) else [] + if isinstance(items, list): + placed |= {item for item in items if isinstance(item, str)} + return placed + + def _pre_taxonomy_outcome(instance: Instance) -> LintOutcome | None: """The two pre-taxonomy states: fresh instance, or broken mid-ingest. diff --git a/src/dex_engine/render/surfaces.py b/src/dex_engine/render/surfaces.py index ccc6998..c0fd6c8 100644 --- a/src/dex_engine/render/surfaces.py +++ b/src/dex_engine/render/surfaces.py @@ -969,6 +969,7 @@ def _render_ingest_receipt(payload: Mapping[str, object]) -> str: "bad_citations", "shortid_citations", "orphans", + "unplaced", "unindexed", "ghost_index", "stale_pages", @@ -1025,6 +1026,7 @@ def _render_health_report(payload: Mapping[str, object]) -> str: "bad_citations": [{"page": str, "id": str}], "shortid_citations": [{"page": str, "token": str}], "orphans": [str], + "unplaced": [str], # cited, but in no topic's items and unledgered "unindexed": [str], "ghost_index": [str], "stale_pages": [{"page": str, "newer": int}], @@ -1162,6 +1164,9 @@ def _health_wiki(surface: str, payload: Mapping[str, object], pages: int) -> lis blocks += _health_names( surface, payload, "orphans", "items no page cites and the taxonomy does not record" ) + blocks += _health_names( + surface, payload, "unplaced", "items a page cites but no taxonomy topic records" + ) blocks += _health_names(surface, payload, "unindexed", "pages missing from index") blocks += _health_names(surface, payload, "ghost_index", "ghost index entries") stale = _health_rows(surface, payload, "stale_pages", ("page",), int_keys=("newer",)) diff --git a/tests/test_lint.py b/tests/test_lint.py index f55d4ef..ba5c377 100644 --- a/tests/test_lint.py +++ b/tests/test_lint.py @@ -197,6 +197,45 @@ def test_orphans_respect_the_uncategorized_ledger(self, instance): assert orphaned in outcome.report assert "items no page cites and the taxonomy does not record — **1**" in outcome.report + def test_cited_but_unplaced_items_are_flagged(self, instance): + # A citation is not a placement: the coverage invariant wants every + # item in some topic's `items` or the uncategorized ledger, and the + # orphan check only ever asks about the uncited. + write_corpus_stub(instance) + write_taxonomy(instance, topics={"brewing": {"items": []}}) + write_page(instance, "brewing", page_text(items=None, body=f"A fact `{ITEM}`.\n")) + write_index(instance, "[[brewing]]\n") + outcome = lint(instance) + assert "items a page cites but no taxonomy topic records — **1**" in outcome.report + assert f"**{ITEM}**" in outcome.report + assert outcome.exit_code == 0 # a finding to repair, like the orphan row + + def test_a_placed_and_cited_item_is_not_unplaced(self, instance): + write_corpus_stub(instance) + write_taxonomy(instance, topics={"brewing": {"items": [ITEM]}}) + write_page(instance, "brewing", page_text(body=f"A fact `{ITEM}`.\n")) + write_index(instance, "[[brewing]]\n") + outcome = lint(instance) + assert "items a page cites but no taxonomy topic records — none" in outcome.report + + def test_an_uncited_unplaced_item_is_the_orphan_finding_not_this_one(self, instance): + write_corpus_stub(instance) + write_taxonomy(instance, topics={"brewing": {"items": []}}) + write_index(instance, "") + outcome = lint(instance) + assert "items a page cites but no taxonomy topic records — none" in outcome.report + assert "items no page cites and the taxonomy does not record — **1**" in outcome.report + + def test_an_uncategorized_ledgered_citation_is_placed(self, instance): + # uncategorized-shares is a topic like any other to the placed set: + # a cited item resting on the ledger satisfies the invariant. + write_corpus_stub(instance) + write_taxonomy(instance, topics={"uncategorized-shares": {"items": [ITEM]}}) + write_page(instance, "brewing", page_text(items=None, body=f"A fact `{ITEM}`.\n")) + write_index(instance, "[[brewing]]\n") + outcome = lint(instance) + assert "items a page cites but no taxonomy topic records — none" in outcome.report + def test_index_consistency(self, instance): write_taxonomy(instance, topics={"brewing": {"items": []}}) write_page(instance, "brewing", page_text(items=None, body="text\n")) From 1de525bf85c69d9477a9a4a49a8cfd68cb3f6bfc Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sun, 23 Aug 2026 21:51:53 +0100 Subject: [PATCH 6/7] Stop calling a resting media unit the engine's to retry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A media unit parked blocked/waiting/error rests untouched while media_fetch is `none` — the drain defers every media job before spending anything — yet the standing view and run report filed it under "Waiting on the engine — N entries it retries by itself", with "attempt n of m" promising retries that are switched off. What actually unblocks it is the owner flipping the config. The classification is a rendering fact, not a ledger one, so no status changes: the row builder — _parked_row, the one place both surfaces' rows are made and the one with the config in hand, since payloads are self-contained by design — marks such a row `resting`, gives it the reason that names the unblock ("media_fetch is `none` — stays parked until it is turned back on", the drain's own note vocabulary), and drops the attempt framing. The surfaces' ownership split reads the mark: a resting row is the owner's, rendered under Needs you with no retry note. `manual` media rows are excluded — they wait on the owner either way, and turning media back on resumes nothing for them, so their stated reason stands. Design §11's split account and vocabulary table updated to carry the exception. Co-Authored-By: Claude Fable 5 --- design/ingestion-pipeline.md | 11 ++++-- src/dex_engine/pipeline/run.py | 35 +++++++++++++++++-- src/dex_engine/render/surfaces.py | 35 ++++++++++++++++--- tests/pipeline/test_run.py | 57 +++++++++++++++++++++++++++++++ tests/render/test_surfaces.py | 39 +++++++++++++++++++++ 5 files changed, 166 insertions(+), 11 deletions(-) diff --git a/design/ingestion-pipeline.md b/design/ingestion-pipeline.md index 808255c..e01921e 100644 --- a/design/ingestion-pipeline.md +++ b/design/ingestion-pipeline.md @@ -1392,7 +1392,12 @@ retry without being asked. So `manual` (the engine has given up) becomes **Needs you**, while `blocked`, `waiting` and `error` (the engine retries by itself) become **Waiting on the engine**, with the retry state visible on the entry. The same rule renames every other engine-internal label to what -the reader must do about it. +the reader must do about it. One exception rides the row, not the status: +a parked media unit under `media_fetch: none` is deferred by the drain, +never retried, so it waits on the owner flipping the config — the payload +builder, the one place the config is in hand (payloads stay +self-contained), marks the row `resting`, and it renders under **Needs +you** with a reason naming what unblocks it and no retry framing. **Open question — where `error` entries belong.** They sit under **Waiting on the engine** today, on the rule above: `error` retries by itself, once @@ -1422,8 +1427,8 @@ silently short list is a lie about scale. |---|---| | **Needs writing up** | items with new enrichment for the session to digest | | **Read these yourself** | jobs resolving to the cognitive floor (OCR, extraction Claude must do with eyes) | -| **Needs you** | entries the engine has given up on: ledger status `manual` | -| **Waiting on the engine** | entries the engine retries unasked: `blocked`, `waiting`, `error` | +| **Needs you** | entries the engine has given up on (ledger status `manual`), plus media units resting under `media_fetch: none` — the owner's config change is what moves them | +| **Waiting on the engine** | entries the engine retries unasked: `blocked`, `waiting`, `error` — minus media units resting under `media_fetch: none`, which it will not retry | | **Not finished** | items still `raw` because a unit they own has not landed | | **Digest these** | items whose enrichment is newer than their digest | | **Waiting on a capability** | the `waiting` cohort, counted by the capability it needs | diff --git a/src/dex_engine/pipeline/run.py b/src/dex_engine/pipeline/run.py index 6b4a65f..383d97e 100644 --- a/src/dex_engine/pipeline/run.py +++ b/src/dex_engine/pipeline/run.py @@ -1422,7 +1422,9 @@ def record(self, entry: LedgerEntry, *, count: bool = False) -> LedgerEntry: 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(_parked_row(stamped, stamped.item)) + self.parked.append( + _parked_row(stamped, stamped.item, media_fetch=self.ctx.config.media_fetch) + ) return stamped def record_outcome( # noqa: PLR0913 — one keyword per ledger schema slot, all optional @@ -1604,13 +1606,36 @@ def _cognitive_jobs(self) -> list[dict[str, str]]: ] -def _parked_row(entry: LedgerEntry, item_id: str) -> dict[str, object]: +def _parked_row(entry: LedgerEntry, item_id: str, *, media_fetch: MediaFetch) -> dict[str, object]: """One parked entry as both report surfaces read it. The run report says what THIS run parked and the standing view says what is parked now; they are the same shape read at two moments, so they are built here once. + + Built here, with the config in hand, is also where a resting media + unit is told apart: under ``media_fetch: none`` the drain defers every + media job, so a parked one waits on the owner flipping the config, not + on any retry the engine will make. The renderer cannot ask the config + (payloads are self-contained), so the row itself carries the + classification — marked ``resting``, reason naming what unblocks it, + and no attempt count, because "attempt n of m" is retry framing for + retries that are switched off. ``manual`` media rows keep their own + stated reason: they wait on the owner either way, and turning media + back on resumes nothing for them. """ + if ( + entry.via == "media" + and media_fetch is MediaFetch.NONE + and entry.status is not Status.MANUAL + ): + return { + "item": item_id, + "url": entry.url, + "status": entry.status.value, + "reason": "media_fetch is `none` — stays parked until it is turned back on", + "resting": True, + } row: dict[str, object] = { "item": item_id, "url": entry.url, @@ -2120,7 +2145,11 @@ def _parked_units(ctx: RunContext, entries: dict[str, LedgerEntry]) -> list[dict """ owners = _unit_owners(ctx.instance, entries, ctx.drivers) rows = [ - _parked_row(entry, owners.get(entry.hash, (entry.item,))[0]) + _parked_row( + entry, + owners.get(entry.hash, (entry.item,))[0], + media_fetch=ctx.config.media_fetch, + ) for entry in entries.values() if entry.status in _PARKED ] diff --git a/src/dex_engine/render/surfaces.py b/src/dex_engine/render/surfaces.py index c0fd6c8..33bb753 100644 --- a/src/dex_engine/render/surfaces.py +++ b/src/dex_engine/render/surfaces.py @@ -245,7 +245,7 @@ def _parked_rows( surface, entry, required=frozenset({"item", "url", "status", "reason"}), - optional=frozenset({"attempts", "attempt_cap"}), + optional=frozenset({"attempts", "attempt_cap", "resting"}), where=where, ) status = _status_at(surface, entry, "status", where) @@ -267,6 +267,10 @@ def _parked_rows( if "attempts" not in entry: _fail(surface, f"{where}attempt_cap without attempts says nothing") row["attempt_cap"] = _int_at(surface, entry, "attempt_cap", where) + if "resting" in entry: + if entry["resting"] is not True: + _fail(surface, f"{where}resting can only be true — omit the key otherwise") + row["resting"] = True rows.append(row) return rows @@ -280,9 +284,20 @@ def _parked_rows( } +def _owner_is_you(row: dict[str, object]) -> bool: + """Whether the row's next action belongs to the reader, not the engine. + + ``manual`` always does; a ``resting`` row does too, whatever its + status — the builder marked it because the engine will not touch it + (a media unit deferred under ``media_fetch: none``), so only the + owner's own change of config moves it. + """ + return row["status"] in _OWNER_IS_YOU or bool(row.get("resting")) + + def _parked_section(parked: list[dict[str, object]], *, mine: bool) -> str: """One parked section: the entries whose next action belongs to one owner.""" - rows = [row for row in parked if (row["status"] in _OWNER_IS_YOU) == mine] + rows = [row for row in parked if _owner_is_you(row) == mine] if not rows: return "" if mine: @@ -309,6 +324,10 @@ def _parked_section(parked: list[dict[str, object]], *, mine: bool) -> str: def _parked_tags(row: dict[str, object]) -> list[str]: status = Status(str(row["status"])) tags = [kernel.code(status.value)] + if row.get("resting"): + # No retry note and no attempt framing: nothing here retries, and + # the row's reason names what unblocks it. + return tags attempts = row.get("attempts") if isinstance(attempts, int): cap = row.get("attempt_cap") @@ -336,7 +355,9 @@ def _render_enrich_report(payload: Mapping[str, object]) -> str: "parked": [{"item": str, "url": str, # survives the session "status": str, "reason": str, "attempts": int, # optional: retry state - "attempt_cap": int}], # (blocked entries) + "attempt_cap": int, # (blocked entries) + "resting": true}], # optional: the engine will + # not act; the owner will "incomplete": [{"item": str, # optional: touched items "landed": int, # still owed work — the "total": int, # ledger units that have @@ -352,7 +373,9 @@ def _render_enrich_report(payload: Mapping[str, object]) -> str: ``parked`` arrives as one list and renders as two sections, split on who owns the next action: ``manual`` is the engine giving up, everything - else re-enters the queue by itself. + else re-enters the queue by itself — except a row marked ``resting``, + which the builder classified as the owner's because the engine will + not act on it. """ surface = "enrich-report" _check_keys( @@ -556,7 +579,9 @@ def _render_status(payload: Mapping[str, object]) -> str: "parked": [{"item": str, "url": str, # optional: every unit parked "status": str, "reason": str, # now, whichever run "attempts": int, # parked it - "attempt_cap": int}], + "attempt_cap": int, + "resting": true}], # optional: the engine will not + # act on it; the owner will "orphans": [str], # optional: item ids whose # enrichment is newer than their # digest (interrupted-session diff --git a/tests/pipeline/test_run.py b/tests/pipeline/test_run.py index dd67ece..9536804 100644 --- a/tests/pipeline/test_run.py +++ b/tests/pipeline/test_run.py @@ -1094,6 +1094,63 @@ def test_media_config_none_gates_the_redrain_too(self, instance): assert ledger.load(instance.ledger_path)[work_hash(self.IMG1)].status is Status.DONE assert (instance.root / f"enrichment/{ITEM}/media-0.png").read_bytes() == b"png" + def _park_blocked_media(self, instance) -> FakeDriver: + """One media unit parked ``blocked`` by a 503 under an active config.""" + write_item(instance) + outage = HttpResponse(status=503, content_type="text/html", body=b"") + driver = FakeDriver(fetch_fn=self.media_fetch([self.IMG1])) + run_mod.run(make_ctx(instance, driver, transport=FakeTransport({self.IMG1: outage}))) + assert ledger.load(instance.ledger_path)[work_hash(self.IMG1)].status is Status.BLOCKED + return driver + + def test_a_resting_media_unit_waits_on_the_owner_not_the_engine(self, instance): + # Under `none` the drain defers every media job, so the standing + # view's "Waiting on the engine — it retries by itself" was false: + # the unit waits on the owner turning media_fetch back on, and the + # row must say so. + driver = self._park_blocked_media(instance) + report = run_mod.status_report( + make_ctx(instance, driver, config=Config(media_fetch=MediaFetch.NONE)) + ) + assert "Waiting on the engine" not in report + yours = report.split("### Needs you", 1)[1] + assert self.IMG1 in yours + assert "media_fetch is `none` — stays parked until it is turned back on" in yours + assert "attempt 1 of 5" not in report # retry framing, and nothing retries + + def test_a_parked_media_unit_under_an_active_config_still_waits_on_the_engine(self, instance): + driver = self._park_blocked_media(instance) + report = run_mod.status_report(make_ctx(instance, driver)) + assert "### Waiting on the engine — 1 entry it retries by itself" in report + assert "attempt 1 of 5" in report + + def test_a_manual_media_unit_keeps_its_own_reason_under_none(self, instance): + # `manual` media rows already wait on the owner, and flipping the + # config resumes nothing for them — their stated reason stands. + write_item(instance) + ledger.append( + instance.ledger_path, + LedgerEntry( + hash=work_hash(self.IMG1), + url=self.IMG1, + item=ITEM, + kind=Kind.WEB, + status=Status.MANUAL, + engine="0.2.0", + date=datetime.date(2026, 8, 19), + via="media", + parent=work_hash(URL), + depth=1, + reason="download failed 5 times — inspect by hand", + ), + ) + report = run_mod.status_report( + make_ctx(instance, FakeDriver(), config=Config(media_fetch=MediaFetch.NONE)) + ) + yours = report.split("### Needs you", 1)[1] + assert "download failed 5 times — inspect by hand" in yours + assert "media_fetch is `none`" not in yours + def test_oversize_media_is_skipped_with_reason(self, instance): write_item(instance) huge = HttpResponse( diff --git a/tests/render/test_surfaces.py b/tests/render/test_surfaces.py index 7b2485e..8ea05a1 100644 --- a/tests/render/test_surfaces.py +++ b/tests/render/test_surfaces.py @@ -327,6 +327,45 @@ def test_no_parked_work_renders_no_section(self): assert "Needs you" not in out assert "Waiting on the engine" not in out + def test_a_resting_row_is_yours_not_the_engines(self): + # The builder marks a row resting when the engine will not act on + # it (a media unit under `media_fetch: none`); "Waiting on the + # engine — it retries by itself" would be false for it. + parked = [ + { + "item": "2026-08-19-example-55ad7b", + "url": "https://cdn.example.test/a.png", + "status": "blocked", + "reason": "media_fetch is `none` — stays parked until it is turned back on", + "resting": True, + } + ] + out = render("status", {"counts": {"blocked": 1}, "parked": parked}) + assert "### Needs you — 1 entry the engine has given up on" in out + assert "Waiting on the engine" not in out + assert " ↳ media_fetch is `none` — stays parked until it is turned back on" in out + + def test_a_resting_row_carries_no_retry_note(self): + # "retries when a provider appears" is the waiting status's note, + # and it is exactly what a resting unit will not do. + parked = [ + { + "item": "2026-08-19-example-55ad7b", + "url": "https://cdn.example.test/a.png", + "status": "waiting", + "reason": "media_fetch is `none` — stays parked until it is turned back on", + "resting": True, + } + ] + out = render("status", {"counts": {"waiting": 1}, "parked": parked}) + assert "- **2026-08-19-example-55ad7b** · `waiting`" in out + assert "retries when a provider appears" not in out + + def test_a_resting_value_other_than_true_is_loud(self): + parked = [{"item": "i", "url": "u", "status": "blocked", "reason": "r", "resting": False}] + with pytest.raises(PayloadError, match="resting"): + render("status", {"counts": {}, "parked": parked}) + def test_unparked_status_in_parked_is_loud(self): parked = [{"item": "i", "url": "u", "status": "done", "reason": "r"}] with pytest.raises(PayloadError, match="parked"): From 16ed8018f26fe6c7424bceb1468921691213eb26 Mon Sep 17 00:00:00 2001 From: Lee Overy Date: Sun, 23 Aug 2026 21:54:14 +0100 Subject: [PATCH 7/7] Converge a synced skill onto the template's shape, not just its content MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Template evolution across releases can replace a synced file with a same-named directory, or fold a directory down to one file. Sync only knew how to mirror content: mkdir(parents=True) over the standing file raised FileExistsError, read_text over the standing directory raised IsADirectoryError, and either one killed the whole sync — an instance that could not take the release that shipped the change. Inside the dex-* directories sync owns, the template is authoritative — the rule the prune step already established — so a shape conflict is now converged, not crashed on: the conflicting entry is removed, the removal reported as a machinery change ("template ships a directory here now" / "template ships a file here now"), and the copy writes the template's shape. Both directions are handled, and a symlink is unlinked rather than followed, as everywhere in sync. Non-dex trees are untouched, as before: nothing instance-owned is ever removed to make room. All three new tests crash the unfixed sync (FileExistsError on file→directory, IsADirectoryError on the other two). Design §12's mirror account gains the shape sentence. Co-Authored-By: Claude Fable 5 --- design/ingestion-pipeline.md | 7 +++- src/dex_engine/sync.py | 62 ++++++++++++++++++++++++++++++------ tests/test_sync.py | 45 ++++++++++++++++++++++++++ 3 files changed, 103 insertions(+), 11 deletions(-) diff --git a/design/ingestion-pipeline.md b/design/ingestion-pipeline.md index e01921e..b9d357f 100644 --- a/design/ingestion-pipeline.md +++ b/design/ingestion-pipeline.md @@ -1476,7 +1476,12 @@ believes it. (`dex-ingest`, split into `dex-capture` + `dex-run`, is the case this exists for.) The same holds INSIDE a live skill: a synced `dex-*` directory mirrors the template exactly, so a reference file the template dropped is removed too, reported the same way — copy-only sync left it -loading its stale procedure in every instance forever. A symlinked skill +loading its stale procedure in every instance forever. The mirror binds +shape as well as content: where a release replaces a synced file with a +same-named directory, or folds a directory down to a file, the conflicting +entry is removed — reported as a machinery change — and the template's +shape is written; copy-only sync crashed on the standing entry instead and +took the whole sync with it. A symlinked skill directory is unlinked rather than recursed into, so whatever it pointed at is left alone. Nothing outside the `dex-` prefix is ever touched — an owner's own skills are instance-owned. diff --git a/src/dex_engine/sync.py b/src/dex_engine/sync.py index 9f5bd70..d92befe 100644 --- a/src/dex_engine/sync.py +++ b/src/dex_engine/sync.py @@ -285,12 +285,49 @@ def _write_if_changed(root: Path, dest: Path, content: str, changed: list[str]) changed.append(str(dest.relative_to(root))) -def _copy_tree(root: Path, src_dir: Traversable, dest_dir: Path, changed: list[str]) -> None: +def _copy_tree( + root: Path, src_dir: Traversable, dest_dir: Path, changed: list[str], *, owned: bool = False +) -> None: for item in src_dir.iterdir(): + dest = dest_dir / item.name if item.is_dir(): - _copy_tree(root, item, dest_dir / item.name, changed) + if owned: + _converge_shape(root, dest, changed, to_dir=True) + _copy_tree(root, item, dest, changed, owned=owned) elif item.is_file(): - _write_if_changed(root, dest_dir / item.name, item.read_text(encoding="utf-8"), changed) + if owned: + _converge_shape(root, dest, changed, to_dir=False) + _write_if_changed(root, dest, item.read_text(encoding="utf-8"), changed) + + +def _converge_shape(root: Path, dest: Path, changed: list[str], *, to_dir: bool) -> None: + """Clear what stands at ``dest`` when its shape contradicts the template's. + + A release may replace a synced file with a same-named directory, or + fold a directory down to one file. Inside a ``dex-*`` skill the + template is authoritative — the rule pruning already applies — so the + conflicting entry is removed, reported as a machinery change, and the + copy that follows writes the template's shape. Without this the sync + itself died: ``mkdir`` over the standing file raised + ``FileExistsError``, and ``read_text`` over the standing directory its + own ``OSError``. A symlink is unlinked rather than followed, so + whatever it pointed at is left alone — pruning's rule here too. + """ + if to_dir: + # A symlink resolving to a real directory keeps its shape; a file, + # a symlink to one, or a dangling symlink does not. + conflict = (dest.is_symlink() or dest.exists()) and not dest.is_dir() + noun = "directory" + else: + conflict = dest.is_dir() + noun = "file" + if not conflict: + return + if dest.is_dir() and not dest.is_symlink(): + shutil.rmtree(dest) + else: + dest.unlink() + changed.append(f"removed {dest.relative_to(root)} (template ships a {noun} here now)") def sync(root: Path, template: Traversable | None = None) -> list[str]: @@ -316,7 +353,9 @@ def sync(root: Path, template: Traversable | None = None) -> list[str]: Returns: Change descriptions: paths (relative to ``root``) that were written because they differed, plus ``removed `` entries for retired - skills and for files a synced ``dex-*`` skill no longer carries. + skills, for files a synced ``dex-*`` skill no longer carries, and + for entries cleared because the template's shape changed (a file + where it now ships a directory, or the reverse). """ tpl = template if template is not None else _bundled_template() # Ensured here, not only at scaffold: a migrated pre-existing instance @@ -331,12 +370,15 @@ def sync(root: Path, template: Traversable | None = None) -> list[str]: if skill.is_dir(): template_skills.add(skill.name) dest = root / ".claude" / "skills" / skill.name - _copy_tree(root, skill, dest, changed) - if skill.name.startswith("dex-"): - # The synced skill mirrors the template exactly: a file the - # template dropped would otherwise keep loading its stale - # procedure in every session, forever. Only the dex-* - # directories sync owns are pruned. + engine_owned = skill.name.startswith("dex-") + # The synced dex-* skill mirrors the template exactly — shape + # included: a same-named file where the template now ships a + # directory (or the reverse) is converged, not crashed on. A + # file the template dropped would otherwise keep loading its + # stale procedure in every session, forever. Only the dex-* + # directories sync owns are converged and pruned. + _copy_tree(root, skill, dest, changed, owned=engine_owned) + if engine_owned: _prune_tree(root, skill, dest, changed) _remove_retired_skills(root, template_skills, changed) _write_if_changed( diff --git a/tests/test_sync.py b/tests/test_sync.py index e8f0bf3..b7a37be 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -373,6 +373,51 @@ def test_a_retired_directory_inside_a_live_skill_is_removed_whole(self, inst, te assert not retired.exists() assert "removed .claude/skills/dex-run/old-references (retired skill file)" in changed + def test_a_file_grown_into_a_directory_converges_to_the_template(self, inst, template): + # A release replaced a synced file with a same-named directory; the + # instance still holds the file. The template is authoritative + # inside dex-* — without the convergence, mkdir over the standing + # file killed the whole sync with FileExistsError. + skill = inst.root / ".claude" / "skills" / "dex-run" + skill.mkdir(parents=True) + (skill / "references").write_text("used to be one file\n") + changed = sync(inst.root, template=template) + assert (skill / "references").is_dir() + assert (skill / "references" / "ingest-item.md").read_text() == "reference\n" + assert ( + "removed .claude/skills/dex-run/references (template ships a directory here now)" + in changed + ) + + def test_a_directory_folded_into_a_file_converges_to_the_template(self, inst, template): + # The mirror case: the template ships SKILL.md as a file where the + # instance holds a directory of that name — read_text over it + # killed the sync just as dead. + skill = inst.root / ".claude" / "skills" / "dex-run" + stale = skill / "SKILL.md" + stale.mkdir(parents=True) + (stale / "part-one.md").write_text("was split up\n") + changed = sync(inst.root, template=template) + assert (skill / "SKILL.md").is_file() + assert (skill / "SKILL.md").read_text() == "run skill\n" + assert "removed .claude/skills/dex-run/SKILL.md (template ships a file here now)" in changed + + def test_a_shape_change_over_a_symlink_unlinks_without_following(self, inst, template): + # The template ships SKILL.md as a file; the instance holds a + # symlink to a directory there. The link is dropped, its target + # untouched — pruning's own symlink rule. + target = inst.root / "elsewhere" + target.mkdir() + (target / "kept.md").write_text("linked content\n") + skill = inst.root / ".claude" / "skills" / "dex-run" + skill.mkdir(parents=True) + (skill / "SKILL.md").symlink_to(target) + changed = sync(inst.root, template=template) + assert (skill / "SKILL.md").is_file() + assert not (skill / "SKILL.md").is_symlink() + assert (target / "kept.md").exists() # the link's target is untouched + assert "removed .claude/skills/dex-run/SKILL.md (template ships a file here now)" in changed + def test_non_engine_skills_are_never_touched(self, inst, template): theirs = inst.root / ".claude" / "skills" / "my-own-skill" theirs.mkdir(parents=True)