diff --git a/Makefile b/Makefile index dfbd47b..2020176 100644 --- a/Makefile +++ b/Makefile @@ -23,7 +23,8 @@ BIN := bin DIST := dist .PHONY: build dist generate dev test validate validate-self-test fmt vet clean help \ - corpus corpus-sync corpus-rebuild corpus-audit corpus-verify corpus-demo corpus-test + corpus corpus-sync corpus-rebuild corpus-offline corpus-audit corpus-verify \ + corpus-demo corpus-test corpus-graph corpus-scan corpus-sources help: @echo "make build templ generate, compile bin/, no dist" @@ -36,11 +37,15 @@ help: @echo "make clean remove bin/ and dist/" @echo "" @echo "corpus (Phase 4, python3 -m corpus.cli):" - @echo "make corpus rebuild the corpus from content/ (drops the collection)" + @echo "make corpus rebuild the pool from the estate (drops both collections)" + @echo "make corpus-offline rebuild content/ only, no Gitea (recovery and CI)" @echo "make corpus-sync incremental update; only changed documents are re-embedded" @echo "make corpus-audit freshness + drift audit (exit 1 on findings)" @echo "make corpus-verify check the index against content/ and Qdrant" @echo "make corpus-demo same query with and without an area filter" + @echo "make corpus-graph Q= query graph nodes and edges (with their source note)" + @echo "make corpus-scan search the pool for credential-shaped content" + @echo "make corpus-sources show the platform guides and the last harvest" @echo "make corpus-test corpus unit tests (no container, no network)" build: generate @@ -87,6 +92,9 @@ corpus: corpus-rebuild corpus-rebuild: @$(CORPUS) rebuild +corpus-offline: + @$(CORPUS) --offline rebuild + corpus-sync: @$(CORPUS) sync @@ -99,6 +107,16 @@ corpus-verify: corpus-demo: @$(CORPUS) demo-filter --query "distributed audio endpoints" --area edge +corpus-graph: + @test -n "$(Q)" || { echo "usage: make corpus-graph Q='photo pipeline'"; exit 2; } + @$(CORPUS) graph --query "$(Q)" + +corpus-scan: + @$(CORPUS) scan + +corpus-sources: + @$(CORPUS) sources + corpus-test: @$(PYTHON) -m unittest discover -s corpus/tests -t . diff --git a/corpus/cli.py b/corpus/cli.py index 689b5f7..818ff89 100644 --- a/corpus/cli.py +++ b/corpus/cli.py @@ -75,6 +75,7 @@ def main(argv=None) -> int: "measure": _cmd_measure, "info": _cmd_info, "graph": _cmd_graph, + "graph-neighbours": _cmd_graph_neighbours, "scan": _cmd_scan, "sources": _cmd_sources, }[args.command] diff --git a/corpus/freshness.py b/corpus/freshness.py index 682acad..9ad899a 100644 --- a/corpus/freshness.py +++ b/corpus/freshness.py @@ -30,15 +30,74 @@ from . import loader, store as store_module from .config import CHUNKER_VERSION from .pipeline import _contract from .qdrant import Qdrant +from .sources import estate as estate_module def audit(cfg, today: datetime.date = None, log=None) -> dict: """Return the audit report as a plain dict (JSON-serialisable).""" log = log or (lambda _msg: None) today = today or datetime.date.today() - documents = loader.load_documents(cfg) - findings = {"stale": [], "due_soon": [], "index_drift": [], "contract_drift": []} + # The audit reads the same estate the pipeline reads, so "the source has + # stopped changing" is measured against the source, not against the index. + harvest = None + try: + harvest = estate_module.harvest(cfg, log=log) + documents = harvest.documents + except estate_module.EstateError as exc: + documents = loader.load_documents(cfg) + estate_error = str(exc) + else: + estate_error = "" + + findings = { + "stale": [], + "due_soon": [], + "index_drift": [], + "contract_drift": [], + "stale_source": [], + "source_missing": [], + "source_unavailable": [], + } + if estate_error: + findings["source_unavailable"].append({"platform": "estate", "problem": estate_error}) + + if harvest is not None: + for name, row in sorted((harvest.sources or {}).items()): + if not row.get("enabled"): + continue + # Curated content is measured by last_verified above; source silence + # is about the harvested estate, which should change on its own. + if name == "curated": + continue + if row.get("missing") or not row.get("documents"): + findings["source_missing"].append( + { + "platform": name, + "problem": "no documents were harvested", + "repo": row.get("repo") or row.get("base") or "", + } + ) + continue + newest = row.get("newest") or "" + try: + changed = datetime.date.fromisoformat(newest[:10]) + except ValueError: + findings["source_missing"].append( + {"platform": name, "problem": f"no usable last_modified ({newest!r})"} + ) + continue + age = (today - changed).days + if age > cfg.source_silence_days: + findings["stale_source"].append( + { + "platform": name, + "repo": row.get("repo") or row.get("base") or "", + "newest": newest, + "age_days": age, + "problem": f"the source has not changed in {age} days", + } + ) for doc in documents: try: @@ -90,6 +149,7 @@ def audit(cfg, today: datetime.date = None, log=None) -> dict: {"source_path": source, "problem": "indexed but no longer in content/"} ) + graph_counts = store.graph_counts() stored_contract = store.get_meta("collection") if stored_contract != _contract(cfg): findings["contract_drift"].append( @@ -120,23 +180,36 @@ def audit(cfg, today: datetime.date = None, log=None) -> dict: "sources": store.source_count(), "collection": cfg.collection, "collection_points": points, + "graph_collection": cfg.graph_collection, + "graph_nodes": graph_counts["nodes"], + "graph_edges": graph_counts["edges"], "provider": cfg.provider, "model": cfg.model, "dim": cfg.dim, "chunker_version": CHUNKER_VERSION, }, "findings": findings, + "sources": (harvest.sources if harvest is not None else {}), + "warnings": (harvest.warnings if harvest is not None else []), "summary": { "stale": len(findings["stale"]), "due_soon": len(findings["due_soon"]), "index_drift": len(findings["index_drift"]), "contract_drift": len(findings["contract_drift"]), + "stale_source": len(findings["stale_source"]), + "source_missing": len(findings["source_missing"]), + "source_unavailable": len(findings["source_unavailable"]), }, # `due_soon` is a warning, not a failure: the document is still # inside its verification horizon. Stale, drift and contract # problems are failures. "ok": not ( - findings["stale"] or findings["index_drift"] or findings["contract_drift"] + findings["stale"] + or findings["index_drift"] + or findings["contract_drift"] + or findings["stale_source"] + or findings["source_missing"] + or findings["source_unavailable"] ), } log(f"audit: {json.dumps(report['summary'], sort_keys=True)}") @@ -147,9 +220,12 @@ def render_text(report: dict) -> str: """Human-readable audit, for a terminal or an Apprise body.""" lines = [] corpus = report["corpus"] + graph = "" + if corpus.get("graph_nodes") is not None: + graph = f", graph {corpus['graph_nodes']} node(s) / {corpus['graph_edges']} edge(s)" lines.append( f"corpus: {corpus['documents']} document(s), {corpus['chunks']} chunk(s), " - f"{corpus['collection_points']} vector(s) in {corpus['collection']}" + f"{corpus['collection_points']} vector(s) in {corpus['collection']}{graph}" ) lines.append( f"horizon: last_verified within {report['horizon']['max_age_days']} days " @@ -159,10 +235,13 @@ def render_text(report: dict) -> str: for kind, label in ( ("stale", "STALE — past the verification horizon"), ("due_soon", "DUE SOON — verify before it goes stale"), - ("index_drift", "INDEX DRIFT — the index does not match content/"), + ("index_drift", "INDEX DRIFT — the index does not match the sources"), ("contract_drift", "CONTRACT DRIFT — rebuild required"), + ("stale_source", "STALE SOURCE — the estate has stopped changing"), + ("source_missing", "SOURCE MISSING — an enabled platform produced nothing"), + ("source_unavailable", "SOURCE UNAVAILABLE — the estate could not be read"), ): - rows = findings[kind] + rows = findings.get(kind) or [] if not rows: continue lines.append("") @@ -177,7 +256,10 @@ def render_text(report: dict) -> str: ) if report["ok"]: lines.append("") - lines.append("OK: nothing stale, no drift." + (" (see warnings above)" if findings["due_soon"] else "")) + lines.append( + "OK: nothing stale, no drift, every source current." + + (" (see warnings above)" if findings["due_soon"] else "") + ) return "\n".join(lines) diff --git a/corpus/tests/test_ingestion.py b/corpus/tests/test_ingestion.py new file mode 100644 index 0000000..1135bfb --- /dev/null +++ b/corpus/tests/test_ingestion.py @@ -0,0 +1,346 @@ +"""Ingestion from the estate: exclusions, provenance, incremental work, graph. + +These tests never touch the network. `FakeGitea` implements the four read calls +the harvester uses; `FakeQdrant` implements the collection routes the pipeline +uses. The real estate is exercised separately (docs/phase-4-notes.md), so a +failure here is a logic failure, not a connectivity one. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import unittest +from unittest import mock + +from corpus import exclusions, guides, loader, pipeline, store as store_module +from corpus.sources import estate, graph as graph_module +from corpus.tests import support + + +def gitea_guide(**overrides) -> guides.Guide: + values = dict( + name="gitea", enabled=True, what="Self-hosted Git.", fits=["ai", "platform", "edge", "products"], + read=["readme", "tree", "selected_source", "commits", "languages"], + base="https://gitea.example", auth="GITEA_TOKEN", + include={"repos": "all", "paths": []}, + exclude={"repos": [], "paths": []}, + limits={"max_file_kb": 200, "max_files_per_repo": 10, "max_commits": 5, "max_tree_entries": 100}, + ) + values.update(overrides) + return guides.Guide(**values) + + +class FakeGitea: + def __init__(self, repos=None, trees=None, files=None, commits=None, languages=None): + self.repos_list = repos or [] + self.trees = trees or {} + self.files = files or {} + self.commit_list = commits or [] + self.language_map = languages or {} + self.calls = [] + + def repos(self): + return list(self.repos_list) + + def tree(self, full_name, ref): + self.calls.append(("tree", full_name, ref)) + return list(self.trees.get(full_name, [])) + + def raw(self, full_name, ref, path): + self.calls.append(("raw", full_name, ref, path)) + try: + return self.files[(full_name, path)] + except KeyError: + from corpus.sources.gitea import GiteaNotFound + + raise GiteaNotFound(f"{full_name}/{path}") from None + + def commits(self, full_name, ref, limit=20): + return list(self.commit_list) + + def languages(self, full_name): + return dict(self.language_map) + + +def repo(full_name="sam/photo-pipeline", **overrides): + values = dict( + full_name=full_name, name=full_name.split("/")[-1], description="Photo ingestion pipeline", + topics=["photo", "python"], default_branch="main", archived=False, + created_at="2024-01-01T00:00:00Z", updated_at="2026-10-14T05:00:00Z", + html_url=f"https://gitea.example/{full_name}", + ) + values.update(overrides) + return values + + +def blob(path, size=None): + return {"type": "blob", "path": path, "size": len(path) if size is None else size} + + +class ExclusionTest(unittest.TestCase): + """The DoD probe: a file in an excluded path never appears in the pool.""" + + def setUp(self): + self.guide = gitea_guide() + self.result = estate.Harvest() + tree = [ + blob("README.md"), + blob("node_modules/left-pad/probe.md"), + blob(".venv/lib/probe.py"), + blob(".git/config"), + blob(".trash/old.md"), + blob("docs/.silverbullet.auth.json"), + blob("src/main.py"), + blob("assets/photo.png"), + ] + files = { + ("sam/photo-pipeline", "README.md"): b"# Photo pipeline\n\nDeduplicates by hash.\n", + ("sam/photo-pipeline", "node_modules/left-pad/probe.md"): b"# PROBE\n", + ("sam/photo-pipeline", ".venv/lib/probe.py"): b"PROBE = 1\n", + ("sam/photo-pipeline", ".git/config"): b"[core]\n", + ("sam/photo-pipeline", ".trash/old.md"): b"# old\n", + ("sam/photo-pipeline", "docs/.silverbullet.auth.json"): b'{"secret_key": "x"}\n', + ("sam/photo-pipeline", "src/main.py"): b"def dedup():\n return True\n", + } + self.client = FakeGitea( + repos=[repo()], + trees={"sam/photo-pipeline": tree}, + files=files, + commits=[{"commit": {"message": "dedup by hash", "author": {"date": "2026-10-14T05:00:00Z"}}}], + languages={"Python": 900, "Shell": 100}, + ) + self.documents = estate._harvest_one_repo( + self.result, self.client, self.guide, repo(), "https://gitea.example" + ) + + def test_probe_in_excluded_paths_never_appears(self): + paths = {doc.repo_path for doc in self.documents} + for probe in ( + "node_modules/left-pad/probe.md", + ".venv/lib/probe.py", + ".git/config", + ".trash/old.md", + "docs/.silverbullet.auth.json", + ): + self.assertNotIn(probe, paths) + self.assertIn("README.md", paths) + self.assertIn("src/main.py", paths) + + def test_rejections_are_reported_with_a_reason(self): + reasons = {(row["path"], row["reason"]) for row in self.result.rejections} + self.assertIn(("node_modules/left-pad/probe.md", "excluded directory 'node_modules'"), reasons) + self.assertIn(("docs/.silverbullet.auth.json", "credential-shaped file name"), reasons) + self.assertEqual([image["path"] for image in self.result.images], ["assets/photo.png"]) + + def test_media_is_an_image_reference_not_text(self): + self.assertEqual([image["path"] for image in self.result.images], ["assets/photo.png"]) + for doc in self.documents: + self.assertNotIn("assets/photo.png", doc.body) + + def test_every_document_carries_provenance(self): + for doc in self.documents: + self.assertEqual(doc.platform, "gitea") + self.assertEqual(doc.tier, 2) + self.assertTrue(doc.origin_url.startswith("https://gitea.example/sam/photo-pipeline")) + self.assertTrue(doc.last_modified) + self.assertIn(doc.area, self.guide.fits) + + +class CredentialContentTest(unittest.TestCase): + def test_a_valid_path_with_a_private_key_is_rejected(self): + result = estate.Harvest() + guide = gitea_guide() + client = FakeGitea( + repos=[repo()], + trees={"sam/photo-pipeline": [blob("README.md"), blob("src/leak.py")]}, + files={ + ("sam/photo-pipeline", "README.md"): b"# Photo pipeline\n", + ("sam/photo-pipeline", "src/leak.py"): ( + ("KEY = '" + "-----BEGIN " + "PRIVATE KEY-----'\n").encode("utf-8") + ), + }, + commits=[], + languages={"Python": 10}, + ) + estate._harvest_one_repo(result, client, guide, repo(), "https://gitea.example") + reasons = {row["reason"]: row for row in result.rejections} + self.assertIn("credential-shaped content", reasons) + self.assertEqual(reasons["credential-shaped content"]["path"], "src/leak.py") + + +class PayloadProvenanceTest(unittest.TestCase): + def test_missing_provenance_field_fails_the_payload(self): + root = support.temp_repo() + self.addCleanup(lambda: __import__("shutil").rmtree(root, ignore_errors=True)) + docs = loader.load_documents(support.make_config(root)) + chunk = __import__("corpus.chunker", fromlist=["chunk_document"]).chunk_document(docs[0].body)[0] + payload = __import__("corpus.payloads", fromlist=["build_payload"]).build_payload( + docs[0], chunk, support.make_config(root) + ) + payload["id"] = pipeline.chunk_id(docs[0].path, 0, docs[0].digest) + for field in ("platform", "tier", "origin_url", "path", "last_modified"): + broken = dict(payload) + broken.pop(field) + with self.assertRaises(Exception): + __import__("corpus.payloads", fromlist=["validate_payload"]).validate_payload(broken) + + +GRAPH = { + "built_at_commit": "abc123", + "nodes": [ + {"id": "photo_ingest", "label": "Photo ingest", "source_file": "300 areas/Photo.md", + "community": 7, "node_type": "entity"}, + {"id": "settings", "label": "settings.json", "source_file": "settings.json", + "community": 118, "node_type": "code"}, + {"id": "image_node", "label": "diagram", "source_file": "300 areas/diagram.png", + "community": 7, "node_type": "image"}, + ], + "links": [ + {"source": "photo_ingest", "target": "settings", "relation": "configured_by", + "source_file": "300 areas/Photo.md", "confidence": "EXTRACTED", "confidence_score": 0.9}, + {"source": "settings", "target": "photo_ingest", "relation": "configures", + "source_file": "settings.json", "confidence": "INFERRED"}, + ], +} + + +class GraphTest(unittest.TestCase): + def parse(self): + return graph_module.parse( + json.dumps(GRAPH).encode("utf-8"), + platform="graphify", base="https://gitea.example", repo="sam/obsidian-vault", + ref="main", last_modified="2026-10-14T05:00:00Z", area="ai", + vault_paths={"300 areas/Photo.md", "300 areas/diagram.png"}, + ) + + def test_nodes_and_edges_carry_their_source_note(self): + result = self.parse() + self.assertEqual(len(result.nodes), 1) + self.assertEqual(result.nodes[0]["id"], "photo_ingest") + self.assertEqual(result.nodes[0]["source_note"], "300 areas/Photo.md") + self.assertIn("/sam/obsidian-vault/src/branch/main/300%20areas/Photo.md", result.nodes[0]["origin_url"]) + self.assertTrue(result.edges) + for edge in result.edges: + self.assertTrue(edge["source_note"]) + + def test_a_node_without_a_resolvable_note_is_refused(self): + result = self.parse() + rejected = {row["path"] for row in result.rejections} + self.assertIn("settings.json", rejected) + self.assertNotIn("settings", {row["id"] for row in result.nodes}) + + def test_media_becomes_an_image_reference(self): + result = self.parse() + self.assertEqual([image["path"] for image in result.images], ["300 areas/diagram.png"]) + self.assertNotIn("image_node", {row["id"] for row in result.nodes}) + + def test_graph_query_returns_the_note(self): + root = support.temp_repo() + self.addCleanup(lambda: __import__("shutil").rmtree(root, ignore_errors=True)) + cfg = support.make_config(root) + harvest = estate.Harvest() + harvest.documents = loader.load_documents(cfg) + harvest.graph = self.parse() + harvest.sources["graphify"] = {"platform": "graphify", "enabled": True, "documents": 0} + fake = support.FakeQdrant(cfg.dim) + with mock.patch("corpus.sources.estate.harvest", return_value=harvest): + with support.fake_qdrant(cfg, fake): + pipeline.run(cfg, mode="rebuild") + with store_module.Store(cfg.db_path) as store: + rows = store.graph_search("photo") + self.assertTrue(rows) + self.assertTrue(all(row["source_note"] for row in rows)) + self.assertEqual(store.graph_node("photo_ingest")["source_note"], "300 areas/Photo.md") + self.assertTrue(store.graph_neighbours("photo_ingest")) + + +class IncrementalEstateTest(unittest.TestCase): + def setUp(self): + self.root = support.temp_repo() + self.addCleanup(lambda: __import__("shutil").rmtree(self.root, ignore_errors=True)) + self.cfg = support.make_config(self.root) + self.fake = support.FakeQdrant(self.cfg.dim) + self.harvest = estate.Harvest() + self.harvest.documents = loader.load_documents(self.cfg) + self.harvest.documents.append(self.harvested("a" * 64, "First version of the repository.")) + self.harvest.sources["gitea"] = {"platform": "gitea", "enabled": True, "documents": 1} + + def harvested(self, digest, body): + return loader.Document( + path="gitea:sam/photo-pipeline/README.md", source_id="sam/photo-pipeline/README.md", + area="platform", level=3, title="sam/photo-pipeline — README.md", + claim=body, confidence="historic", period="2024–2026", tech=[], metrics=[], artefacts=[], + public=True, last_verified="2026-10-14", tags=["gitea"], body=body, digest=digest, + url="https://gitea.example/sam/photo-pipeline/src/branch/main/README.md", + platform="gitea", tier=2, origin_url="https://gitea.example/sam/photo-pipeline/src/branch/main/README.md", + last_modified="2026-10-14T05:00:00Z", source_type="doc", kind="markdown", + repo_path="README.md", + ) + + def run_pipeline(self, mode="sync"): + with mock.patch("corpus.sources.estate.harvest", return_value=self.harvest): + with support.fake_qdrant(self.cfg, self.fake): + return pipeline.run(self.cfg, mode=mode) + + def test_rebuild_and_second_run_do_not_duplicate(self): + first = self.run_pipeline("rebuild") + points = len(self.fake.collections[self.cfg.collection]) + self.assertEqual(first.platforms["gitea"]["documents"], 1) + second = self.run_pipeline("sync") + self.assertEqual(len(self.fake.collections[self.cfg.collection]), points) + self.assertEqual(second.added, 0) + self.assertEqual(second.updated, 0) + self.assertEqual(second.texts_embedded, 0) + + def test_only_the_changed_harvested_document_is_reembedded(self): + self.run_pipeline("rebuild") + self.harvest.documents[-1] = self.harvested("b" * 64, "Second version, one line changed.") + stats = self.run_pipeline("sync") + self.assertEqual(stats.updated, 1) + expected = len(__import__("corpus.chunker", fromlist=["chunk_document"]).chunk_document( + "Second version, one line changed." + )) + self.assertEqual(stats.texts_embedded, expected) + self.assertEqual(stats.embed_calls, 1) + self.assertEqual(stats.unchanged, 3) + + def test_pool_scan_finds_a_planted_secret(self): + self.harvest.documents[-1] = self.harvested( + "c" * 64, + "AWS_ACCESS_KEY_ID=" + "AKIA" + "ABCDEFGHIJKLMNOP" + "\nThis line is a planted credential.", + ) + with self.assertRaises(pipeline.PipelineError) as caught: + self.run_pipeline("rebuild") + self.assertIn("credential-shaped content", str(caught.exception)) + + +class CodeChunkerTest(unittest.TestCase): + def test_a_function_is_never_cut(self): + from corpus import chunker + + body = "\n\n".join( + f"def function_{index}():\n value = {{'index': {index}}}\n return value" + for index in range(120) + ) + chunks = chunker.chunk_code(body) + self.assertGreater(len(chunks), 1) + for chunk in chunks: + depth = 0 + for line in chunk.text.splitlines(): + depth = chunker._code_depth(line, depth) + self.assertEqual(depth, 0, f"chunk {chunk.index} ends inside a block") + self.assertIn("def function_", chunk.text) + + def test_oversize_single_function_is_flagged_not_cut(self): + from corpus import chunker + + body = "def huge():\n" + "\n".join(f" x{index} = {index}" for index in range(2000)) + chunks = chunker.chunk_code(body) + self.assertEqual(len(chunks), 1) + self.assertTrue(chunks[0].oversize) + + +if __name__ == "__main__": + unittest.main() diff --git a/flows/corpus.py b/flows/corpus.py index 9884598..6a4ce20 100644 --- a/flows/corpus.py +++ b/flows/corpus.py @@ -48,13 +48,49 @@ from corpus.config import Config # noqa: E402 @_flow(name="corpus-sync", log_prints=True) def sync_corpus() -> dict: - """Incremental refresh. Runs on every push that touches content/.""" + """Incremental refresh of the whole estate. Runs on a push or on demand.""" logger = _get_run_logger() cfg = Config.from_env() stats = pipeline.run(cfg, mode="sync", log=logger.info) return stats.to_dict() +@_flow(name="corpus-scheduled-ingest", log_prints=True) +def scheduled_ingest(alert: bool = True) -> dict: + """Nightly: harvest, report what changed, then audit and alert on staleness. + + This is the scheduled run the DoD asks for: it reports documents added, + updated and removed per platform, and it raises an Apprise alert when a + source is missing, unavailable or has stopped changing. + """ + logger = _get_run_logger() + cfg = Config.from_env() + stats = pipeline.run(cfg, mode="sync", log=logger.info) + logger.info(stats.summary()) + report = freshness.audit(cfg, log=logger.info) + logger.info(freshness.render_text(report)) + if alert and not report["ok"]: + try: + freshness.alert(report, cfg.apprise_url) + except Exception as exc: # noqa: BLE001 - the alert must not fail the run + logger.warning(f"Apprise alert failed: {exc}") + return { + "changed": { + "documents": stats.documents, + "added": stats.added, + "updated": stats.updated, + "removed": stats.removed, + "unchanged": stats.unchanged, + "embedding_calls": stats.embed_calls, + "texts_embedded": stats.texts_embedded, + "platforms": stats.platforms, + "rejections": stats.rejection_summary, + "graph_rebuilt": stats.graph_rebuilt, + }, + "audit": report, + } + + @_flow(name="corpus-rebuild", log_prints=True) def rebuild_corpus() -> dict: """From-scratch rebuild. The recovery path: deleting the volume is always safe.""" @@ -85,5 +121,7 @@ if __name__ == "__main__": # pragma: no cover - convenience for a manual run print(freshness.render_text(freshness_audit(alert=False))) elif command == "rebuild": print(rebuild_corpus()) + elif command == "scheduled": + print(scheduled_ingest(alert=False)) else: print(sync_corpus())