fix(bib): hoard IOM revisions instead of discarding them

attach-iom compared each download with an item's *first* attachment
(a dead Zotero-relative storage: row on 29 manuals), then called
attach_file with the same title — which returns the existing attachment
without copying. Every revised chapter since 2026-04-15 was downloaded
nightly and thrown away; bib and Zotero both still hold the first copy.

Revisions are now kept, never replaced: compared against the newest
copy on disk and attached as <stem>.<YYYY-MM-DD><ext> titled
'<title> (<date>)'. Zotero sync likewise never overwrites: a changed
same-name file is added beside the old one under a dated name, and a
missing Zotero file is restored. The corpus index reads only the newest
member of each version family, so hoarded revisions are not indexed
twice.

src/bib/iom.py

@@ -32,6 +32,7 @@ import json
 import logging
 import re
 from dataclasses import dataclass
+from datetime import date
 from pathlib import Path
 from typing import TYPE_CHECKING, Iterable
 
@@ -326,18 +327,20 @@ def _sha256(path: Path) -> str:
 
 
 def _existing_attachment_hash(store: Store, item_key: str) -> str | None:
-    """SHA-256 of an item's first already-attached file, or None."""
+    """SHA-256 of an item's newest attached file that exists on disk, or
+    None. Newest, not first: revisions are hoarded as later attachments,
+    and 29 manuals carry a dead Zotero-relative ``storage:`` row first."""
     con = store._con()  # noqa: SLF001
-    row = con.execute(
+    for row in con.execute(
         """SELECT a.storage_path FROM attachments a
            JOIN items i ON a.item_id = i.id
-           WHERE i.key = ? ORDER BY a.id LIMIT 1""",
+           WHERE i.key = ? ORDER BY a.id DESC""",
         (item_key,),
-    ).fetchone()
-    if not row or not row["storage_path"]:
-        return None
-    p = Path(row["storage_path"])
-    return _sha256(p) if p.is_file() else None
+    ).fetchall():
+        p = Path(row["storage_path"] or "")
+        if row["storage_path"] and p.is_file():
+            return _sha256(p)
+    return None
 
 
 def download_attachments(
@@ -347,11 +350,17 @@ def download_attachments(
     pubs: Iterable[str] | None = None,
     force: bool = False,
     tmp_dir: Path | None = None,
+    today: date | None = None,
 ) -> dict[str, int]:
     """Download every Manual's PDF and attach it via Store.attach_file.
 
-    Idempotent: the remote file's SHA-256 is compared against the
-    currently-attached copy, and the download is skipped if unchanged.
+    Idempotent: the remote file's SHA-256 is compared against the newest
+    attached copy, and the download is skipped if unchanged. A revision
+    is **hoarded**, never replaced: it is attached as
+    ``<stem>.<YYYY-MM-DD><ext>`` titled ``"<title> (<date>)"`` beside
+    every earlier copy, so chapters can be compared over time (the
+    corpus index reads only the newest; ``llm.source``). ``force``
+    re-downloads but still stores nothing that matches the newest copy.
     Returns ``{pub: count_attached}``.
     """
     wanted = set(pubs) if pubs else None
@@ -379,11 +388,25 @@ def download_attachments(
             continue
 
         remote_hash = _sha256(tmp_path)
-        if not force and _existing_attachment_hash(store, item.key) == remote_hash:
+        newest = _existing_attachment_hash(store, item.key)
+        if newest == remote_hash:
             tmp_path.unlink(missing_ok=True)
             continue
 
-        store.attach_file(item.key, tmp_path, title=item.title)
+        title = item.title
+        if newest is not None:
+            stamp = (today or date.today()).isoformat()
+            con = store._con()  # noqa: SLF001
+            if con.execute(
+                "SELECT 1 FROM attachments a JOIN items i ON i.id = a.item_id "
+                "WHERE i.key = ? AND a.filename = ?",
+                (item.key, f"{item.title} ({stamp})"),
+            ).fetchone():
+                stamp = f"{stamp}-{remote_hash[:8]}"  # second revision that day
+            dated = tmp_path.with_name(f"{tmp_path.stem}.{stamp}{tmp_path.suffix}")
+            tmp_path.replace(dated)
+            tmp_path, title = dated, f"{item.title} ({stamp})"
+        store.attach_file(item.key, tmp_path, title=title)
         tmp_path.unlink(missing_ok=True)
         results[pub] = results.get(pub, 0) + 1
         log.info("attached: %s (%s)", filename, pub or "?")

src/bib/sync.py

@@ -20,6 +20,7 @@ from __future__ import annotations
 
 import json
 import shutil
+from datetime import date
 from pathlib import Path
 from typing import TYPE_CHECKING
 
@@ -516,11 +517,15 @@ def _sync_attachments(
     zot_storage: Path,
 ) -> int:
     """Copy every bib attachment for ``bib_item`` into Zotero's storage
-    and create the matching ``itemAttachments`` row. Idempotent: an
-    existing same-filename child attachment is left alone."""
+    and create the matching ``itemAttachments`` row. Idempotent, and it
+    never overwrites: when bib's file differs from the same-named one
+    already in Zotero, bib's copy is added beside it as
+    ``<stem>.<YYYY-MM-DD><ext>`` (the bib attachment's date), so every
+    version is kept for comparison. A Zotero file missing from disk is
+    restored. Returns the number of files added or restored."""
     con = store._con()  # noqa: SLF001
     rows = con.execute(
-        """SELECT a.filename, a.content_type, a.storage_path
+        """SELECT a.filename, a.content_type, a.storage_path, a.created_at
            FROM attachments a
            JOIN items i ON a.item_id = i.id
            WHERE i.key = ?""",
@@ -530,11 +535,13 @@ def _sync_attachments(
     if not rows:
         return 0
 
-    # Existing child attachments keyed by path so we don't duplicate.
+    # Existing child attachments by path → their item (id, key), so a
+    # same-name file is neither duplicated nor left stale.
     existing = {
-        r[0]
+        r[0]: (r[1], r[2])
         for r in db.con.execute(
-            "SELECT ia.path FROM itemAttachments ia WHERE ia.parentItemID = ?",
+            "SELECT ia.path, i.itemID, i.key FROM itemAttachments ia "
+            "JOIN items i ON i.itemID = ia.itemID WHERE ia.parentItemID = ?",
             (zot_parent_id,),
         ).fetchall()
         if r[0]
@@ -550,22 +557,60 @@ def _sync_attachments(
         disk_name = src.name
         zot_path = f"storage:{disk_name}"
         if zot_path in existing:
-            continue
+            _att_id, att_key = existing[zot_path]
+            zfile = zot_storage / att_key / disk_name
+            if not zfile.is_file():
+                zfile.parent.mkdir(parents=True, exist_ok=True)
+                shutil.copy2(src, zfile)  # restore, nothing to overwrite
+                count += 1
+                continue
+            if _same_file(src, zfile):
+                continue
+            created = row["created_at"] if "created_at" in row.keys() else ""
+            stamp = (created or "")[:10] or date.today().isoformat()
+            disk_name = f"{src.stem}.{stamp}{src.suffix}"
+            zot_path = f"storage:{disk_name}"
+            if zot_path in existing:
+                _id2, key2 = existing[zot_path]
+                prior = zot_storage / key2 / disk_name
+                if prior.is_file() and _same_file(src, prior):
+                    continue
+                disk_name = f"{src.stem}.{stamp}-{_sha8(src)}{src.suffix}"
+                zot_path = f"storage:{disk_name}"
+                if zot_path in existing:
+                    continue
         att_key = generate_key()
         dest_dir = zot_storage / att_key
         dest_dir.mkdir(parents=True, exist_ok=True)
         shutil.copy2(src, dest_dir / disk_name)
-        db.add_attachment(
+        new_id = db.add_attachment(
             zot_parent_id,
             link_mode=1,  # Zotero.Attachments.LINK_MODE_IMPORTED_FILE
             content_type=row["content_type"] or "application/pdf",
             path=zot_path,
             key=att_key,
         )
+        existing[zot_path] = (new_id, att_key)
         count += 1
     return count
 
 
+def _same_file(a: Path, b: Path) -> bool:
+    """Byte-identical? Size + mtime short-circuit (copy2 keeps mtime)."""
+    import filecmp
+
+    sa, sb = a.stat(), b.stat()
+    if sa.st_size != sb.st_size:
+        return False
+    return sa.st_mtime_ns == sb.st_mtime_ns or filecmp.cmp(a, b, shallow=False)
+
+
+def _sha8(p: Path) -> str:
+    import hashlib
+
+    return hashlib.sha256(p.read_bytes()).hexdigest()[:8]
+
+
 def _sync_fr_links(
     db: Db,
     store: Store,

src/llm/source.py

@@ -24,6 +24,7 @@ from __future__ import annotations
 
 import logging
 import os
+import re
 import shutil
 import sqlite3
 from dataclasses import dataclass
@@ -74,17 +75,40 @@ def _year_of(store: Store, item_key: str) -> str:
     return row[0].split(":", 1)[1] if row else ""
 
 
+# A hoarded revision: <stem>.<YYYY-MM-DD>[-<sha8>]<ext> beside <stem><ext>.
+_VERSION_RE = re.compile(r"\.\d{4}-\d{2}-\d{2}(?:-[0-9a-f]{8})?(?=\.[^.]+$)")
+
+
+def _newest_versions(paths: list[Path]) -> list[Path]:
+    """One path per version family (``_VERSION_RE``), in the family's
+    first-attached position: its newest member on disk, else its newest.
+    Older revisions stay attached (hoarded) but are not indexed twice."""
+    order: list[str] = []
+    members: dict[str, list[Path]] = {}
+    for p in paths:  # attachment id order
+        fam = _VERSION_RE.sub("", p.name)
+        if fam not in members:
+            order.append(fam)
+            members[fam] = []
+        members[fam].append(p)
+    out = []
+    for fam in order:
+        on_disk = [p for p in members[fam] if p.exists()]
+        out.append((on_disk or members[fam])[-1])
+    return out
+
+
 def _attachment_paths(store: Store, item_key: str) -> list[Path]:
     rows = (
         store._con()
         .execute(
             "SELECT a.storage_path FROM attachments a "
-            "JOIN items i ON i.id = a.item_id WHERE i.key = ?",
+            "JOIN items i ON i.id = a.item_id WHERE i.key = ? ORDER BY a.id",
             (item_key,),
         )
         .fetchall()
     )
-    return [Path(r[0]) for r in rows]
+    return _newest_versions([Path(r[0]) for r in rows])
 
 
 def _attachment_paths_by_key(store: Store) -> dict[str, list[Path]]:
@@ -97,14 +121,15 @@ def _attachment_paths_by_key(store: Store) -> dict[str, list[Path]]:
     rows = (
         store._con()
         .execute(
-            "SELECT i.key, a.storage_path FROM attachments a JOIN items i ON i.id = a.item_id"
+            "SELECT i.key, a.storage_path FROM attachments a "
+            "JOIN items i ON i.id = a.item_id ORDER BY a.id"
         )
         .fetchall()
     )
     by_key: dict[str, list[Path]] = {}
     for key, path in rows:
         by_key.setdefault(key, []).append(Path(path))
-    return by_key
+    return {k: _newest_versions(v) for k, v in by_key.items()}
 
 
 # ── comments ──

tests/bib/test_iom_exercise.py

@@ -231,7 +231,11 @@ class TestExistingAttachmentHash:
         store = MagicMock()
         con = MagicMock()
         store._con.return_value = con
-        con.execute.return_value.fetchone.return_value = {"storage_path": str(f)}
+        # newest first; a dead Zotero-relative row is skipped
+        con.execute.return_value.fetchall.return_value = [
+            {"storage_path": "storage:gone.pdf"},
+            {"storage_path": str(f)},
+        ]
         result = _existing_attachment_hash(store, "K1")
         assert result == _sha256(f)
 

tests/bib/test_iom_full.py

@@ -269,3 +269,104 @@ class TestDownloadAttachments:
         # Only 100-04 should be processed
         assert "100-04" in results or results == {}
         store.close()
+
+
+class TestRevisionHoard:
+    """A revised chapter is kept as a new dated version; nothing is replaced.
+
+    attach-iom used to call attach_file with the same title, which
+    returns the existing attachment — every revision since April was
+    downloaded and discarded."""
+
+    def _setup(self, tmp_path):
+        from bib.item import Manual
+        from bib.store import Store
+
+        store = Store(tmp_path / "bib.sqlite", storage_dir=tmp_path / "storage")
+        item = Manual(
+            title="IOM Ch1",
+            url="https://cms.gov/manuals/downloads/clm104c01.pdf",
+            pub_number="100-04",
+        )
+        key = store.create(item)
+        client = MagicMock()
+        return store, key, client
+
+    def _files(self, store, key):
+        from pathlib import Path
+
+        rows = (
+            store._con()
+            .execute(  # noqa: SLF001
+                "SELECT a.filename, a.storage_path FROM attachments a "
+                "JOIN items i ON i.id = a.item_id WHERE i.key = ? ORDER BY a.id",
+                (key,),
+            )
+            .fetchall()
+        )
+        return [(r[0], Path(r[1]).name, Path(r[1]).read_bytes()) for r in rows]
+
+    def test_revision_is_added_as_dated_version_and_old_kept(self, tmp_path):
+        from datetime import date
+
+        from bib.iom import download_attachments
+
+        store, key, client = self._setup(tmp_path)
+        client.get.return_value = MagicMock(content=b"PDF v1")
+        download_attachments(
+            store, client, tmp_dir=tmp_path / "t", today=date(2026, 4, 15)
+        )
+        client.get.return_value = MagicMock(content=b"PDF v2 revised")
+        out = download_attachments(
+            store, client, tmp_dir=tmp_path / "t", today=date(2026, 11, 11)
+        )
+        assert out == {"100-04": 1}
+        assert self._files(store, key) == [
+            ("IOM Ch1", "clm104c01.pdf", b"PDF v1"),
+            ("IOM Ch1 (2026-11-11)", "clm104c01.2026-11-11.pdf", b"PDF v2 revised"),
+        ]
+
+    def test_unchanged_against_the_newest_version_is_skipped(self, tmp_path):
+        from datetime import date
+
+        from bib.iom import download_attachments
+
+        store, key, client = self._setup(tmp_path)
+        client.get.return_value = MagicMock(content=b"PDF v1")
+        download_attachments(
+            store, client, tmp_dir=tmp_path / "t", today=date(2026, 4, 15)
+        )
+        client.get.return_value = MagicMock(content=b"PDF v2")
+        download_attachments(
+            store, client, tmp_dir=tmp_path / "t", today=date(2026, 11, 11)
+        )
+        out = download_attachments(
+            store, client, tmp_dir=tmp_path / "t", today=date(2026, 11, 12)
+        )
+        assert out == {}
+        assert len(self._files(store, key)) == 2
+
+    def test_dead_first_attachment_does_not_force_reattach(self, tmp_path):
+        # 29 manuals carry a Zotero-relative 'storage:x.pdf' row first;
+        # the newest real file is what a download is compared with.
+        from datetime import date
+
+        from bib.iom import download_attachments
+
+        store, key, client = self._setup(tmp_path)
+        con = store._con()  # noqa: SLF001
+        item_id = con.execute("SELECT id FROM items WHERE key=?", (key,)).fetchone()[0]
+        con.execute(
+            "INSERT INTO attachments (item_id, key, filename, content_type, storage_path) "
+            "VALUES (?, 'ZZZZZZZZ', 'clm104c01.pdf', 'application/pdf', 'storage:clm104c01.pdf')",
+            (item_id,),
+        )
+        con.commit()
+        client.get.return_value = MagicMock(content=b"PDF v1")
+        download_attachments(
+            store, client, tmp_dir=tmp_path / "t", today=date(2026, 4, 15)
+        )
+        out = download_attachments(
+            store, client, tmp_dir=tmp_path / "t", today=date(2026, 11, 11)
+        )
+        assert out == {}

tests/bib/test_sync_deeper.py

@@ -89,6 +89,45 @@ class TestSyncAttachments:
         assert n1 == 1
         assert n2 == 0
 
+    def test_changed_same_name_file_is_hoarded_beside_the_old_copy(self, tmp_path):
+        # A bib file that changed under the same name is added to Zotero as
+        # a dated copy; the existing Zotero file is never overwritten.
+        path = _setup(tmp_path)
+        storage = tmp_path / "storage"
+        storage.mkdir()
+        src_file = tmp_path / "source" / "chapter.pdf"
+        src_file.parent.mkdir()
+        src_file.write_bytes(b"%PDF old chapter")
+        row = {
+            "filename": "chapter.pdf",
+            "content_type": "application/pdf",
+            "storage_path": str(src_file),
+            "created_at": "2026-04-15T03:23:12Z",
+        }
+        store = MagicMock()
+        store._con.return_value.execute.return_value.fetchall.return_value = [row]
+        bib_item = MagicMock(key="K1")
+        with Db(path) as db:
+            parent_id = db.create_item(TYPE_MAP["document"])
+            db.commit()
+            assert _sync_attachments(db, store, bib_item, parent_id, storage) == 1
+            db.commit()
+            src_file.write_bytes(b"%PDF new chapter, revised")
+            row["created_at"] = "2026-11-11T18:00:00Z"
+            assert _sync_attachments(db, store, bib_item, parent_id, storage) == 1
+            db.commit()
+            assert _sync_attachments(db, store, bib_item, parent_id, storage) == 0
+            paths = sorted(
+                r[0]
+                for r in db.con.execute(
+                    "SELECT path FROM itemAttachments WHERE parentItemID = ?",
+                    (parent_id,),
+                )
+            )
+        assert paths == ["storage:chapter.2026-11-11.pdf", "storage:chapter.pdf"]
+        blobs = sorted(p.read_bytes() for p in storage.rglob("*.pdf"))
+        assert blobs == [b"%PDF new chapter, revised", b"%PDF old chapter"]
+
     def test_missing_file(self, tmp_path):
         path = _setup(tmp_path)
         store = MagicMock()

tests/llm/test_source.py

@@ -391,3 +391,26 @@ class TestDefaultRoot:
         from conf import ROOT
 
         assert _default_root() == ROOT / ".state" / "comments"
+
+
+def test_dated_versions_index_only_the_newest(tmp_path):
+    """IOM revisions are hoarded as <stem>.<YYYY-MM-DD><ext> beside the
+    original; the corpus reads one file per version family — the newest
+    — so a chapter is not indexed once per revision."""
+    from bib.item import Manual
+    from bib.store import Store
+    from llm.source import _attachment_paths, _attachment_paths_by_key
+
+    store = Store(tmp_path / "b.sqlite", storage_dir=tmp_path / "st")
+    key = store.create(Manual(title="Ch1", url="https://x/clm104c01.pdf"))
+    for name, body in (
+        ("clm104c01.pdf", b"v1"),
+        ("clm104c01.2026-11-11.pdf", b"v2"),
+        ("appendix.pdf", b"a"),
+    ):
+        f = tmp_path / name
+        f.write_bytes(body)
+        store.attach_file(key, f, title=name)
+    names = [p.name for p in _attachment_paths(store, key)]
+    assert names == ["clm104c01.2026-11-11.pdf", "appendix.pdf"]
+    assert [p.name for p in _attachment_paths_by_key(store)[key]] == names