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 a417511..b9d357f 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). 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 | @@ -1245,6 +1245,44 @@ 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. + +**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 @@ -1354,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 @@ -1384,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 | @@ -1433,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. @@ -1838,7 +1886,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-lint/SKILL.md b/instance/skills/dex-lint/SKILL.md index c3e2768..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, @@ -95,6 +100,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/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. diff --git a/instance/skills/dex-run/SKILL.md b/instance/skills/dex-run/SKILL.md index 138a55e..493586c 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 @@ -145,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/lint.py b/src/dex_engine/lint.py index 20bfd75..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 @@ -18,7 +20,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 +66,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 @@ -154,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 @@ -176,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, @@ -229,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. @@ -551,6 +579,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/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/src/dex_engine/pipeline/run.py b/src/dex_engine/pipeline/run.py index 918ae71..383d97e 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", @@ -1421,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 @@ -1603,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, @@ -2119,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 ] @@ -2298,6 +2328,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..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 @@ -969,6 +994,7 @@ def _render_ingest_receipt(payload: Mapping[str, object]) -> str: "bad_citations", "shortid_citations", "orphans", + "unplaced", "unindexed", "ghost_index", "stale_pages", @@ -982,6 +1008,7 @@ def _render_ingest_receipt(payload: Mapping[str, object]) -> str: "misfiled_outputs", "waiting", "cognitive", + "never_harvested", "stale_passes", "capped", "incomplete_threads", @@ -1024,6 +1051,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}], @@ -1040,6 +1068,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 @@ -1160,6 +1189,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",)) @@ -1266,6 +1298,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/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/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/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"): 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) diff --git a/tests/test_lint.py b/tests/test_lint.py index 72afbd8..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")) @@ -747,6 +786,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: 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)