fix(facts): a doc re-judged minimal left its previous numbers in the store

pipeline/clean/src/clean/main.py

@@ -216,6 +216,8 @@ async def worker(doc):
                     vinfo += " · self-corrected"
                 if fx:
                     vinfo += f" · facts {fx.get('kept', 0)}"
+                    if fx.get("cleared"):
+                        vinfo += f" (cleared {fx['cleared']} stale)"
                 log(f"OK {res.get('path') or 'skipped'} ({res.get('method')}/{rep}){vinfo}{tinfo}")
             except Exception as ex:  # noqa: BLE001
                 if is_rate_limit(ex):

pipeline/clean/src/clean/worker.py

@@ -156,6 +156,27 @@ def _verify(o):
         facts_counts = {"kept": len(fact_rows), "rejected": len(rejected)}
         if rejected:
             facts_counts["rejected_reasons"] = [list(r) for r in rejected[:8]]
+    elif facts_dir and out.representation == "minimal":
+        # A `minimal` page is a POINTER — title, type, source link (agents.py: "raw data without
+        # narrative ... -> pointer"). Note what that does and does not mean: the SOURCE may still be
+        # full of figures; it is the PAGE that stopped carrying them. Every row in the store cites a
+        # page via page_path, so rows left from an earlier pass would cite a page that cannot support
+        # them, and the answer layer would serve a figure its own citation does not contain.
+        #
+        # So the test is what the page can support, NOT whether the extractor happens to be
+        # configured: a pointer page cannot support a figure either way. That also keeps this
+        # consistent with the `skipped` path above, which deletes unconditionally because a skipped
+        # doc has no page at all — the same reasoning, one step further.
+        #
+        # A full/digest page whose extractor is switched off is the opposite case and deliberately
+        # untouched: that page still carries its narrative and supports its rows; we are simply not
+        # refreshing them. They go stale until the document's bytes change (classify_pending keys on
+        # the content hash, so a config flip alone never re-runs the document).
+        n = factstore.delete_facts(facts_dir, file_id)
+        if n:
+            # reported rather than logged here: main owns the pass log, and importing it would be a
+            # cycle. A destructive step must show up in the audit log like its two sibling deletes.
+            facts_counts = {"kept": 0, "rejected": 0, "cleared": n}
 
     result = {
         "fileId": file_id,

pipeline/clean/tests/test_facts.py

@@ -329,6 +329,105 @@ def test_process_one_skipped_doc_clears_stale_facts(tmp_path):
     assert factstore.query_facts(fdir) == []                   # stale numbers gone
 
 
+def test_process_one_minimal_representation_clears_stale_facts(tmp_path):
+    """factstore's contract is replace-per-document: "a reprocess REPLACEs the document's
+    observations atomically". The `skipped` path enforces it — "a doc downgraded to noise must not
+    leave stale numbers behind" — but that is one of several ways a pass can produce no facts.
+
+    A prose doc re-judged `minimal` takes the same route: it yields no rows, so nothing replaced and
+    nothing deleted, and the previous pass's numbers stayed in the store with a page_path pointing at
+    a page that no longer carries them. The answer layer then serves a figure its own citation does
+    not contain."""
+    from tests.test_worker import FakeProcessor, _output
+
+    from clean.fake_llm import FakeProseFactsProcessor
+    from clean.worker import process_one
+
+    src = tmp_path / "Quarterly Report Q1 2026.md"
+    src.write_text(QUARTERLY)
+    doc = {"fileId": "FMIN", "path": str(src),
+           "entry": {"name": src.name, "drivePath": "/X/Clients/1. Globex/q.md",
+                     "orgUnit": "Clients", "sourceUri": "local://q"}}
+    fdir = str(tmp_path / "facts")
+    brain = str(tmp_path / "brain")
+
+    # pass 1: a full prose doc, so the prose facts agent runs and stores an observation
+    res = asyncio.run(process_one(doc, FakeProcessor(_output()), str(tmp_path), brain,
+                                  prose_facts_processor=FakeProseFactsProcessor(), facts_dir=fdir))
+    assert res["facts"]["kept"] >= 1
+    assert factstore.query_facts(fdir)
+
+    # pass 2: the same doc now judged `minimal` -> no rows this pass, so none may survive from before
+    res2 = asyncio.run(process_one(doc, FakeProcessor(_output(representation="minimal")),
+                                   str(tmp_path), brain,
+                                   prose_facts_processor=FakeProseFactsProcessor(), facts_dir=fdir))
+    assert res2["skipped"] is False
+    assert factstore.query_facts(fdir) == []          # today: the pass-1 rows are still here
+
+
+def test_switching_the_extractor_off_does_not_wipe_the_store(tmp_path):
+    """The other direction, and the reason the clear is conditional: producing no rows because the
+    extractor is SWITCHED OFF is a config change, not a statement about the document. Turning
+    CLEAN_FACTS_PROSE off to save cost must not silently destroy the facts store — those rows go
+    stale, which turning it back on repairs, where deletion would not."""
+    from tests.test_worker import FakeProcessor, _output
+
+    from clean.fake_llm import FakeProseFactsProcessor
+    from clean.worker import process_one
+
+    src = tmp_path / "Quarterly Report Q1 2026.md"
+    src.write_text(QUARTERLY)
+    doc = {"fileId": "FOFF", "path": str(src),
+           "entry": {"name": src.name, "drivePath": "/X/Clients/1. Globex/q.md",
+                     "orgUnit": "Clients", "sourceUri": "local://q"}}
+    fdir = str(tmp_path / "facts")
+    brain = str(tmp_path / "brain")
+
+    asyncio.run(process_one(doc, FakeProcessor(_output()), str(tmp_path), brain,
+                            prose_facts_processor=FakeProseFactsProcessor(), facts_dir=fdir))
+    assert factstore.query_facts(fdir)
+
+    # same document, same content — only the extractor is gone. The page is still full/digest, so it
+    # still carries the narrative those rows cite: they go stale rather than disappearing.
+    asyncio.run(process_one(doc, FakeProcessor(_output()), str(tmp_path), brain, facts_dir=fdir))
+    assert factstore.query_facts(fdir), "a config change must not delete the document's facts"
+
+    # ...but a POINTER page cannot support a figure whatever the config says, so `minimal` clears
+    # even with the extractor off. That is the same reasoning as the `skipped` path, which deletes
+    # unconditionally because a skipped doc has no page at all — the two must not disagree.
+    asyncio.run(process_one(doc, FakeProcessor(_output(representation="minimal")),
+                            str(tmp_path), brain, facts_dir=fdir))
+    assert factstore.query_facts(fdir) == []
+
+
+def test_a_sheet_pass_with_zero_kept_rows_still_clears(tmp_path):
+    """`fact_rows is not None` matters: an extraction that ran and kept NOTHING yields an empty list,
+    which must still replace (i.e. clear) the document's rows. `if fact_rows:` would skip that and
+    silently leave the previous pass's numbers behind — the same bug one branch over."""
+    from tests.test_worker import FakeProcessor, _output
+
+    from clean.worker import process_one
+
+    fdir = str(tmp_path / "facts")
+    csv = tmp_path / "data.csv"
+    csv.write_text("label,value\nheadcount,312\n")
+    doc = {"fileId": "FSHEET", "path": str(csv), "entry": {"name": "data.csv", "sourceUri": "u"}}
+    factstore.replace_facts(fdir, "FSHEET", _rows("FSHEET", _obs()), page_path="stale.md",
+                            entity=None, org_unit=None, extracted_at="t")
+    assert factstore.query_facts(fdir)
+
+    class NoFacts:
+        async def run(self, prompt, *, deps=None, usage_limits=None):
+            from clean.fake_llm import fake_result
+            from clean.schemas import FactsOutput
+            return fake_result(FactsOutput(observations=[], reason="no numeric cells"))
+
+    res = asyncio.run(process_one(doc, FakeProcessor(_output()), str(tmp_path),
+                                  str(tmp_path / "brain"), facts_processor=NoFacts(), facts_dir=fdir))
+    assert res["facts"]["kept"] == 0
+    assert factstore.query_facts(fdir) == []
+
+
 def test_process_one_without_facts_processor_is_unchanged(tmp_path):
     from tests.test_worker import FakeProcessor, _output