fix(bib,cli): durable merge-log path, empty-body alias guard, body_sha1 backfill (refs #665)

dev/scripts/backfill_body_sha1.py

@@ -0,0 +1,103 @@
+"""One-off: backfill ``body_sha1`` onto legacy ``source:email`` rows.
+
+Ingest dedup (``bib/email_ingest.py``) matches candidate re-sends only
+on ``json_extract(extra_json,'$.body_sha1')``. Legacy ``source:email``
+rows created before Task 3's ingest-side dedup landed have no stored
+``body_sha1``, so the first post-deploy re-send of a legacy notice
+misses the match and creates a fresh item instead of aliasing onto the
+existing one.
+
+This backfills ``extra_json.body_sha1`` for every ``source:email`` item
+that's missing it, computed the same way ``email_ingest.py`` does:
+``sha1(" ".join(body.split()).encode()).hexdigest()``. The catch is
+that ``abstract`` on the stored row is ``body[:4000]`` (see
+``_ingest_message``) — for a message whose body was actually cut at
+4000 chars, hashing the truncated abstract does NOT reproduce the hash
+that would be computed from the full original body, so any such item
+would get a *wrong* body_sha1 that can never match a genuine re-send.
+Those items are skipped (``skipped_truncated``) rather than backfilled
+with a hash that would silently misbehave forever.
+
+Truncation heuristic: ``len(abstract) < 3999`` (not ``< 4000``) —
+leaves a 1-char margin since the exact boundary between "the body was
+precisely 4000 chars and happened to not get cut" and "the body was
+longer and got cut to 4000" is not distinguishable from the stored
+abstract alone; treating anything at-or-near the cap as possibly
+truncated is the safe default (a false skip just means the row keeps
+missing body_sha1 and gets caught by the next dedupe pass some other
+way; a false backfill would poison dedup with a bad hash forever).
+
+Reuses the raw-SQL extra_json read/merge/write convention from
+``dedupe_bib_legacy.py`` and ``email_ingest.py`` — ``Item.to_row()``
+only serializes the fields a subclass declares (``Source`` only knows
+``doc_type``) and would silently drop ``body_sha1`` if a merge went
+through it instead of raw SQL.
+
+``--dry-run`` is the default; pass ``--live`` to actually write.
+
+Usage: uv run python dev/scripts/backfill_body_sha1.py [--live]
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import sys
+from typing import Any
+
+TRUNCATION_THRESHOLD = 3999
+
+
+def _body_hash(text: str) -> str:
+    return hashlib.sha1(" ".join((text or "").split()).encode()).hexdigest()
+
+
+def backfill_body_sha1(store: Any, *, dry_run: bool = True) -> dict:
+    """Backfill ``body_sha1`` onto legacy ``source:email`` rows.
+
+    Returns ``{candidates, backfilled, skipped_truncated}``.
+    ``candidates`` counts every ``source:email`` row missing
+    ``body_sha1`` (whether or not it ends up backfilled);
+    ``skipped_truncated`` is the subset whose ``abstract`` looks like
+    it was cut by email_ingest's ``body[:4000]`` cap and is therefore
+    excluded from ``backfilled``.
+    """
+    con = store._con()  # noqa: SLF001
+    out = {"candidates": 0, "backfilled": 0, "skipped_truncated": 0}
+
+    rows = con.execute(
+        "SELECT i.key, i.abstract, i.extra_json FROM items i"
+        " JOIN item_tags it ON it.item_id = i.id"
+        " JOIN tags t ON t.id = it.tag_id AND t.name = 'source:email'"
+        " WHERE i.item_type = 'source'"
+        " AND json_extract(i.extra_json, '$.body_sha1') IS NULL"
+        " ORDER BY i.id"
+    ).fetchall()
+
+    for key, abstract, extra_json in rows:
+        out["candidates"] += 1
+        abstract = abstract or ""
+        if len(abstract) >= TRUNCATION_THRESHOLD:
+            out["skipped_truncated"] += 1
+            continue
+
+        if dry_run:
+            continue
+
+        ej = json.loads(extra_json or "{}")
+        ej["body_sha1"] = _body_hash(abstract)
+        store.update(key, extra_json=json.dumps(ej))
+        out["backfilled"] += 1
+
+    return out
+
+
+if __name__ == "__main__":
+    from bib import Store
+
+    live = "--live" in sys.argv
+    _store = Store()
+    _result = backfill_body_sha1(_store, dry_run=not live)
+    print(json.dumps(_result, indent=1))
+    if not live:
+        print("(dry run — pass --live to write)")

src/bib/email_ingest.py

@@ -126,8 +126,9 @@ def _ingest_message(
     subject = (msg.get("Subject") or "(no subject)").strip()
     sender_domain = _sender_domain(sender)
     body = _extract_body(msg)
+    normalized_body = " ".join((body or "").split())
     title = subject[:255]
-    body_sha1 = hashlib.sha1(" ".join((body or "").split()).encode()).hexdigest()
+    body_sha1 = hashlib.sha1(normalized_body.encode()).hexdigest()
 
     # Listservs re-send identical content under a fresh Message-ID
     # (Class B in docs/superpowers/plans/2026-08-26-zotero-dedupe.md).
@@ -140,14 +141,26 @@ def _ingest_message(
     # itself, and skip the normal upsert/tag/attachment-refresh path
     # below. json_extract is more robust than a LIKE-on-serialized-JSON
     # match since it doesn't depend on json.dumps's separator spacing.
+    #
+    # An empty/whitespace-only body always hashes to the same constant
+    # sha1("") regardless of content — attachment-only messages under a
+    # shared subject would otherwise all collapse onto one keeper, and
+    # the alias short-circuit below returns before attachments are
+    # processed, silently dropping every attachment but the first
+    # message's. Skip the lookup entirely for empty bodies so these are
+    # always treated as distinct new items.
     con = store._con()
-    dup = con.execute(
+    dup = (
-        "SELECT key, url, extra_json FROM items"
+        con.execute(
-        " WHERE item_type = 'source' AND title = ?"
+            "SELECT key, url, extra_json FROM items"
-        " AND json_extract(extra_json, '$.body_sha1') = ?"
+            " WHERE item_type = 'source' AND title = ?"
-        " AND url != ?",
+            " AND json_extract(extra_json, '$.body_sha1') = ?"
-        (title, body_sha1, url),
+            " AND url != ?",
-    ).fetchone()
+            (title, body_sha1, url),
+        ).fetchone()
+        if normalized_body
+        else None
+    )
     is_alias = dup is not None
     # An alias re-send targets the keeper's existing url so tags/collections
     # merge through Store.upsert's normal union-by-url logic below (a

src/cli/bib.py

@@ -555,9 +555,19 @@ def sync_zotero(
         if merge_dupes:
             from zot.merge import merge_url_duplicates
 
+            # Explicit log_dir, not the default ".state/zotero-dedupe": the
+            # nightly workflow runs this command via `docker exec api` with
+            # WORKDIR /app, and compose mounts only ./data:/app/data — a
+            # relative ".state/..." path lands on the container's ephemeral
+            # filesystem and is lost on the next recreate. zotero_db_path's
+            # parent (data/zotero/data/) is guaranteed to be on the mount
+            # since the DB itself lives there.
+            merge_log_dir = Path(zotero_db_path).parent / "dedupe-logs"
             stats |= {
                 f"merge_{k}": v
-                for k, v in merge_url_duplicates(zotero_db_path, dry_run=False).items()
+                for k, v in merge_url_duplicates(
+                    zotero_db_path, dry_run=False, log_dir=merge_log_dir
+                ).items()
             }
     finally:
         if hold_zotero:

tests/bib/test_backfill_body_sha1.py

@@ -0,0 +1,147 @@
+"""Tests for dev/scripts/backfill_body_sha1.py.
+
+Legacy ``source:email`` rows created before Task 3's ingest-side dedup
+have no ``body_sha1`` in ``extra_json``, so the first post-deploy
+re-send of a legacy notice creates a fresh item instead of aliasing.
+This script backfills ``body_sha1`` for legacy rows whose ``abstract``
+wasn't truncated by email_ingest's ``body[:4000]`` cap, using the exact
+same normalization + hash recipe as ``bib.email_ingest``.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+from bib.item import Source
+from bib.store import Store
+
+# ``dev/scripts`` isn't a package on sys.path — load the module by path,
+# matching the convention other one-off dev scripts use in this repo.
+_SCRIPT_PATH = (
+    Path(__file__).resolve().parents[2] / "dev" / "scripts" / "backfill_body_sha1.py"
+)
+_spec = importlib.util.spec_from_file_location("backfill_body_sha1", _SCRIPT_PATH)
+_mod = importlib.util.module_from_spec(_spec)
+sys.modules[_spec.name] = _mod
+_spec.loader.exec_module(_mod)
+backfill_body_sha1 = _mod.backfill_body_sha1
+
+
+def _body_hash(text: str) -> str:
+    return hashlib.sha1(" ".join((text or "").split()).encode()).hexdigest()
+
+
+def _raw_extra_json(store: Store, key: str) -> dict:
+    row = (
+        store._con()  # noqa: SLF001
+        .execute("SELECT extra_json FROM items WHERE key = ?", (key,))
+        .fetchone()
+    )
+    return json.loads(row["extra_json"] or "{}")
+
+
+@pytest.fixture
+def store(tmp_path):
+    db_path = tmp_path / "test-backfill.sqlite"
+    s = Store(db_path)
+    yield s
+    s.close()
+
+
+def _make_email_item(
+    store: Store, *, title: str, abstract: str, url: str, with_body_sha1: bool = False
+) -> str:
+    item = Source(title=title, url=url, abstract=abstract)
+    item.doc_type = "Email"
+    item.add_tag("source:email")
+    key = store.create(item)
+    if with_body_sha1:
+        ej = _raw_extra_json(store, key)
+        ej["body_sha1"] = _body_hash(abstract)
+        store.update(key, extra_json=json.dumps(ej))
+    return key
+
+
+class TestBackfillBodySha1:
+    def test_backfills_untruncated_item(self, store):
+        key = _make_email_item(
+            store,
+            title="Legacy Notice",
+            abstract="Some legacy body content.",
+            url="email:legacy1@x.example",
+        )
+        result = backfill_body_sha1(store, dry_run=False)
+        assert result["candidates"] == 1
+        assert result["backfilled"] == 1
+        assert result["skipped_truncated"] == 0
+
+        ej = _raw_extra_json(store, key)
+        assert ej["body_sha1"] == _body_hash("Some legacy body content.")
+
+    def test_skips_truncated_item(self, store):
+        long_abstract = "x" * 4000
+        key = _make_email_item(
+            store,
+            title="Long Legacy Notice",
+            abstract=long_abstract,
+            url="email:legacy2@x.example",
+        )
+        result = backfill_body_sha1(store, dry_run=False)
+        assert result["candidates"] == 1
+        assert result["backfilled"] == 0
+        assert result["skipped_truncated"] == 1
+
+        ej = _raw_extra_json(store, key)
+        assert "body_sha1" not in ej
+
+    def test_skips_item_with_existing_body_sha1(self, store):
+        _make_email_item(
+            store,
+            title="Already Hashed",
+            abstract="Already has a hash.",
+            url="email:legacy3@x.example",
+            with_body_sha1=True,
+        )
+        result = backfill_body_sha1(store, dry_run=False)
+        assert result["candidates"] == 0
+        assert result["backfilled"] == 0
+
+    def test_dry_run_makes_no_changes(self, store):
+        key = _make_email_item(
+            store,
+            title="Dry Run Notice",
+            abstract="Body content.",
+            url="email:legacy4@x.example",
+        )
+        result = backfill_body_sha1(store, dry_run=True)
+        assert result["candidates"] == 1
+        assert result["backfilled"] == 0
+
+        ej = _raw_extra_json(store, key)
+        assert "body_sha1" not in ej
+
+    def test_second_run_is_idempotent(self, store):
+        _make_email_item(
+            store,
+            title="Idempotent Notice",
+            abstract="Body content.",
+            url="email:legacy5@x.example",
+        )
+        first = backfill_body_sha1(store, dry_run=False)
+        assert first["backfilled"] == 1
+
+        second = backfill_body_sha1(store, dry_run=False)
+        assert second["candidates"] == 0
+        assert second["backfilled"] == 0
+
+    def test_ignores_non_email_items(self, store):
+        item = Source(title="Not an email", url="https://example.com/x")
+        store.create(item)
+        result = backfill_body_sha1(store, dry_run=False)
+        assert result["candidates"] == 0

tests/bib/test_email_ingest.py

@@ -319,6 +319,31 @@ class TestResendAliasing:
         ej = _raw_extra_json(store, items[0].key)
         assert "same-mid@x.example" not in ej.get("alias_mids", [])
 
+    def test_empty_body_attachment_only_resends_stay_separate(
+        self, store, mailbox, tmp_path
+    ):
+        """Attachment-only emails have an empty extracted body, so
+        body_sha1 is the constant sha1(""). Two such messages under the
+        same subject must NOT alias together — aliasing would hit the
+        early return in _ingest_message and silently drop the second
+        message's attachments."""
+        m1 = make_msg(
+            "Attachment Only",
+            "",
+            "att1@x.example",
+            "Mon, 27 Apr 2026 10:00:00 -0400",
+        )
+        m2 = make_msg(
+            "Attachment Only",
+            "   \n  ",
+            "att2@x.example",
+            "Fri, 10 Jul 2026 10:00:00 -0400",
+        )
+        _ingest_message(store, mailbox, m1, tmp_path)
+        _ingest_message(store, mailbox, m2, tmp_path)
+        items = store.list_items(query="Attachment Only")
+        assert len(items) == 2
+
     def test_alias_resend_tags_merge_into_keeper(self, store, tmp_path):
         """A re-send arriving via a different mailbox must contribute its
         mailbox:/sender:/list:/year: tags to the keeper item, not just