Phase 4 ingestion: exclusion/provenance/incremental/graph tests

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 .
 

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]

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)
 
 

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

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