fix(clean): editing a canonical silently lost its duplicate's content

pipeline/clean/src/clean/main.py

@@ -46,8 +46,18 @@ def dedup_pending(pending, state, inventory, brain_md_dir=None, facts_dir=None):
     pending doc in this pass) becomes `duplicate` -> no LLM call, no page, state points at the
     canonical file id. Deterministic: existing processed entries win; within a pass, lowest id."""
     canonical = {}
+    # Do not seed the index with a hash its file no longer serves. This set is every pending file
+    # that has a hash; what matters is its overlap with the seeding candidates below, which also
+    # require `status == "processed"` — and a processed file can only be pending because
+    # `prev["rawHash"] != raw_hash`. So the files actually removed from the seed are exactly the
+    # processed ones whose recorded hash is stale. (`new` docs and duplicates are in the set too but
+    # could never have seeded anyway: they are not `processed`.) Seeding from a stale hash filed a
+    # doc carrying those old bytes as a duplicate of a page that does not have them — and since the
+    # duplicate's own bytes never change again, its content was lost with no error.
+    restale = {d["fileId"] for d in pending if d.get("rawHash")}
     for fid, f in state.get("files", {}).items():
-        if f.get("status") == "processed" and f.get("rawHash") and fid in inventory:
+        if (f.get("status") == "processed" and f.get("rawHash")
+                and fid in inventory and fid not in restale):
             canonical.setdefault(f["rawHash"], fid)
     kept, duplicates = [], 0
     now = datetime.datetime.now(datetime.UTC).isoformat()

pipeline/clean/src/clean/state.py

@@ -92,18 +92,29 @@ def classify_pending(inventory, state, raw_dir):
     """Pending = new / changed (hash) / previous error / requeued / restored-after-delete /
     orphaned-duplicate (its canonical is gone) / deleted."""
     pending, seen = [], set()
+    # Hash every present file up front: deciding whether a duplicate is still deduped needs its
+    # CANONICAL's current hash, not just its own. Same total work as hashing inside the loop.
+    paths = {}
     for file_id, entry in inventory.items():
-        seen.add(file_id)
         local = entry.get("localPath")
-        path = os.path.join(raw_dir, local) if local else None
-        if not path or not os.path.exists(path):
+        p = os.path.join(raw_dir, local) if local else None
+        if p and os.path.exists(p):
+            paths[file_id] = p
+    hashes = {fid: file_sha256(p) for fid, p in paths.items()}
+    for file_id, entry in inventory.items():
+        seen.add(file_id)
+        if file_id not in paths:
             continue
-        raw_hash = file_sha256(path)
+        path, raw_hash = paths[file_id], hashes[file_id]
         prev = state["files"].get(file_id)
         # A file whose source reappeared after a delete (its page was removed) must be regenerated
-        # even if the bytes are unchanged; a duplicate whose canonical left the inventory is now the
-        # only holder of that content and must get its own page.
-        orphaned_dup = prev and prev.get("status") == "duplicate" and prev.get("duplicateOf") not in inventory
+        # even if the bytes are unchanged; a duplicate is the only holder of its content — and so
+        # needs its own page — once its canonical stops serving that content, whether because the
+        # canonical LEFT the inventory or because its bytes CHANGED. `hashes.get(canon, raw_hash)`
+        # leaves a canonical that is in the inventory but missing on disk deduped, as before.
+        canon = prev.get("duplicateOf") if prev else None
+        orphaned_dup = (prev and prev.get("status") == "duplicate"
+                        and (canon not in inventory or hashes.get(canon, raw_hash) != raw_hash))
         if (not prev or prev.get("rawHash") != raw_hash
                 or prev.get("status") in ("error", "requeued", "deleted")
                 or orphaned_dup):

pipeline/clean/tests/test_main.py

@@ -227,6 +227,99 @@ async def ok(doc, *a, **kw):
     assert _state_of(env)["files"]["C"]["duplicateOf"] == "A"
 
 
+def test_dedup_never_points_a_duplicate_at_changed_content(env, monkeypatch):
+    """The invariant test_classify_pending_promotes_orphaned_duplicate states — "the content it
+    still holds gets its own page (otherwise it is lost forever)" — also has to hold when the
+    canonical's CONTENT changes, not only when the canonical leaves the inventory.
+
+    Pass 1 dedups B onto A. Then a.md is edited: A is pending with new bytes, and the old bytes
+    live only on b.md. B must get its own page, or that document is gone from the brain with no
+    error and no way back — B's bytes never change again, and its canonical stays in the
+    inventory forever."""
+    processed = []
+
+    async def ok(doc, *a, **kw):
+        processed.append(doc["fileId"])
+        return {"fileId": doc["fileId"], "skipped": False, "method": "text",
+                "path": f"general/{doc['fileId']}.md", "usage": {}}
+
+    monkeypatch.setattr(clean_main, "process_one", ok)
+    (env.raw / "b.md").write_text("doc a")            # identical to a.md
+    asyncio.run(run_once(env.cfg))
+    assert _state_of(env)["files"]["B"]["duplicateOf"] == "A"
+
+    processed.clear()
+    (env.raw / "a.md").write_text("doc a EDITED")     # the canonical's content moves on
+    stats = asyncio.run(run_once(env.cfg))
+
+    assert sorted(processed) == ["A", "B"], "B still holds the old bytes and must get its own page"
+    files = _state_of(env)["files"]
+    assert files["B"]["status"] == "processed"
+    assert files["B"].get("duplicateOf") is None
+    assert stats["duplicates"] == 0
+    # ...and it settles: no churn on the next pass
+    assert asyncio.run(run_once(env.cfg))["pending"] == 0
+
+
+def test_dedup_converges_when_a_canonical_of_several_duplicates_changes(env, monkeypatch):
+    """Convergence is what makes re-pending safe. Dedup exists to avoid LLM calls, so a fix that
+    re-pends duplicates it should not costs money on every pass forever. With THREE identical files,
+    editing the canonical must promote exactly ONE of the two duplicates and re-dedup the other onto
+    it — not give both pages, not leave both lost, and not oscillate. Reverting the canonical's bytes
+    must settle too."""
+    processed = []
+
+    async def ok(doc, *a, **kw):
+        processed.append(doc["fileId"])
+        return {"fileId": doc["fileId"], "skipped": False, "method": "text",
+                "path": f"general/{doc['fileId']}.md", "usage": {}}
+
+    monkeypatch.setattr(clean_main, "process_one", ok)
+    (env.raw / "b.md").write_text("doc a")
+    inv = json.loads((env.raw / "_state.json").read_text())
+    inv["files"]["C"] = {"name": "c.md", "localPath": "c.md", "drivePath": "/X/c.md"}
+    (env.raw / "_state.json").write_text(json.dumps(inv))
+    (env.raw / "c.md").write_text("doc a")            # A, B and C all identical
+    asyncio.run(run_once(env.cfg))
+    files = _state_of(env)["files"]
+    assert (files["B"]["status"], files["C"]["status"]) == ("duplicate", "duplicate")
+
+    (env.raw / "a.md").write_text("doc a EDITED")     # the canonical moves on
+    asyncio.run(run_once(env.cfg))
+    files = _state_of(env)["files"]
+    holders = [f for f in ("A", "B", "C") if files[f]["status"] == "processed"]
+    assert holders == ["A", "B"], files                # exactly one duplicate promoted
+    assert files["C"]["duplicateOf"] == "B"            # the other re-dedups onto it
+    assert asyncio.run(run_once(env.cfg))["pending"] == 0          # converged
+
+    (env.raw / "a.md").write_text("doc a")            # ...and reverting settles as well
+    asyncio.run(run_once(env.cfg))
+    assert _state_of(env)["files"]["A"]["duplicateOf"] == "B"
+    assert asyncio.run(run_once(env.cfg))["pending"] == 0
+
+
+def test_dedup_does_not_seed_the_index_with_a_changed_docs_old_hash(env, monkeypatch):
+    """The other half, one pass earlier: a doc pending BECAUSE its hash changed must not seed the
+    canonical index with the hash it no longer has. Otherwise a new doc carrying those old bytes is
+    filed as a duplicate of a document that does not serve them."""
+    async def ok(doc, *a, **kw):
+        return {"fileId": doc["fileId"], "skipped": False, "method": "text",
+                "path": f"general/{doc['fileId']}.md", "usage": {}}
+
+    monkeypatch.setattr(clean_main, "process_one", ok)
+    asyncio.run(run_once(env.cfg))                    # A and B processed, different content
+    inv = json.loads((env.raw / "_state.json").read_text())
+    inv["files"]["C"] = {"name": "c.md", "localPath": "c.md", "drivePath": "/X/c.md"}
+    (env.raw / "_state.json").write_text(json.dumps(inv))
+    (env.raw / "c.md").write_text("doc a")            # C carries A's ORIGINAL content
+    (env.raw / "a.md").write_text("doc a EDITED")     # ...which A no longer has
+
+    stats = asyncio.run(run_once(env.cfg))
+    files = _state_of(env)["files"]
+    assert files["C"]["status"] == "processed", "C holds content no other page serves"
+    assert stats["duplicates"] == 0
+
+
 def test_run_once_deletes_page_when_source_disappears(env, monkeypatch):
     async def fake_process_one(doc, processor, raw, out, catalog=None, **kw):
         rel = f"general/{doc['fileId']}.md"

pipeline/clean/tests/test_state.py

@@ -130,3 +130,22 @@ def test_classify_pending_promotes_orphaned_duplicate(tmp_path):
     inventory["A"] = {"localPath": "a.pdf"}
     assert classify_pending(inventory, state, raw) == []  # B stays a duplicate, A missing on disk
 
+
+def test_classify_pending_promotes_a_duplicate_whose_canonical_changed(tmp_path):
+    """Same invariant, the other way the canonical can stop holding the shared content: its bytes
+    change. Only its DISAPPEARANCE used to orphan the duplicate, so an edit to the canonical left
+    the duplicate's content with no page and no route back — the duplicate's own bytes never change
+    again, and its canonical is still right there in the inventory."""
+    raw = str(tmp_path)
+    _touch(raw, "a.pdf", "shared")
+    _touch(raw, "b.pdf", "shared")
+    h = file_sha256(os.path.join(raw, "b.pdf"))
+    inventory = {"A": {"localPath": "a.pdf"}, "B": {"localPath": "b.pdf"}}
+    state = {"files": {"A": {"rawHash": h, "status": "processed"},
+                       "B": {"rawHash": h, "status": "duplicate", "duplicateOf": "A"}}}
+    assert classify_pending(inventory, state, raw) == []        # nothing changed yet
+
+    _touch(raw, "a.pdf", "the canonical moved on")
+    pending = classify_pending(inventory, state, raw)
+    assert sorted(p["fileId"] for p in pending) == ["A", "B"]   # today: ["A"] only
+