feat(comments): write per-attachment .md siblings + attach to bib

Each comment dir now produces, alongside combined.md, one
``<original>.md`` sibling per attachment containing just that
attachment's extracted text + a small YAML frontmatter (status,
chars, source_file). Lets Zotero (and anything else consuming the
corpus) navigate to a single attachment's transcript instead of
always opening the aggregated combined.md.

- combine.extract_comment writes siblings during fresh extraction.
- New combine.derive_siblings_from_combined parses an existing
  combined.md and emits the missing siblings without re-decoding the
  underlying PDFs/DOCXs — the backfill path for the 23k items already
  extracted before this change.
- walker renames its callback to on_extracted(comment_id, comment_dir)
  and runs the sibling-backfill for any skipped dir before invoking
  the callback, so callers always see a complete set of MDs.
- cli.comments attach callback now glob('*.md')s the dir and attaches
  each missing one; the existing-attachments cache key changes from
  item_key to (item_key, filename) to keep idempotence at the per-file
  level.

Per-attachment metadata extraction (author/org from letterhead +
signature lines) tracked separately as #416.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/cli/comments.py

@@ -63,26 +63,32 @@ def _build_bib_helpers(use_bib: bool, attach: bool):
     from bib import connect
 
     store = connect()
-    existing: set[str] = {
+    # Pre-load every existing markdown attachment so re-runs are no-ops
-        row[0]
+    # at the per-(item, filename) level. Both combined.md and the per-
+    # attachment <name>.md siblings live in this set.
+    existing: set[tuple[str, str]] = {
+        (row[0], row[1])
         for row in store._con().execute(
-            "SELECT i.key FROM attachments a JOIN items i ON a.item_id = i.id "
+            "SELECT i.key, a.filename FROM attachments a "
-            "WHERE a.filename = 'combined.md'"
+            "JOIN items i ON a.item_id = i.id "
+            "WHERE a.filename LIKE '%.md'"
         )
     }
 
-    def attach_callback(comment_id: str, md_path: Path) -> None:
+    def attach_callback(comment_id: str, comment_dir: Path) -> None:
         info = cache.get(comment_id)
         if not info:
             return  # comment not in bib
         item_key = info[0]
-        if item_key in existing:
+        for md_path in sorted(comment_dir.glob("*.md")):
-            return  # already attached on a prior run
+            key = (item_key, md_path.name)
-        try:
+            if key in existing:
-            store.attach_file(item_key, md_path, title="combined.md")
+                continue
-            existing.add(item_key)
+            try:
-        except Exception as e:  # noqa: BLE001
+                store.attach_file(item_key, md_path, title=md_path.name)
-            log.warning("attach combined.md failed for %s: %s", item_key, e)
+                existing.add(key)
+            except Exception as e:  # noqa: BLE001
+                log.warning("attach %s failed for %s: %s", md_path.name, item_key, e)
 
     return body_lookup, attach_callback
 
@@ -133,7 +139,7 @@ def extract(
         limit=limit or None,
         workers=workers,
         force=force,
-        on_combined=attach_cb,
+        on_extracted=attach_cb,
     )
     for k, v in stats.items():
         typer.echo(f"  {k}: {v}")

src/rex/comments/__init__.py

@@ -7,11 +7,16 @@ Public API:
 
 from __future__ import annotations
 
-from rex.comments.combine import extract_comment, parse_combined
+from rex.comments.combine import (
+    derive_siblings_from_combined,
+    extract_comment,
+    parse_combined,
+)
 from rex.comments.extract import ExtractResult, extract_attachment
 
 __all__ = [
     "ExtractResult",
+    "derive_siblings_from_combined",
     "extract_attachment",
     "extract_comment",
     "parse_combined",

src/rex/comments/combine.py

@@ -27,6 +27,11 @@ def extract_comment(
 ) -> Path:
     """Extract every attachment in *comment_dir* and write *combined.md*.
 
+    Also writes one ``<filename>.md`` sibling per attachment with that
+    attachment's text only — useful when downstream tools (Zotero) want
+    to navigate to a single attachment's transcript instead of the
+    aggregated combined.md.
+
     Returns the path to combined.md. If it already exists and ``force``
     is False, returns it without doing any work (resume rule).
 
@@ -41,13 +46,21 @@ def extract_comment(
 
     attachments_meta: list[dict[str, Any]] = []
     body_sections: list[str] = []
+    sibling_payloads: list[tuple[str, str]] = []  # (sibling_filename, contents)
 
     for path in sorted(_attachment_paths(comment_dir)):
         result = extract_attachment(path)
         attachments_meta.append(
             {"file": path.name, "status": result.status, "chars": result.chars}
         )
-        body_sections.append(_render_section(path.name, result))
+        section = _render_section(path.name, result)
+        body_sections.append(section)
+        sibling_payloads.append(
+            (
+                _sibling_name(path.name),
+                _sibling_contents(path.name, comment_id, docket_id, result),
+            )
+        )
 
     inline_section = (
         f"## Inline comment\n\n{inline_body.strip()}\n" if inline_body.strip() else ""
@@ -68,9 +81,44 @@ def extract_comment(
     tmp = out.with_suffix(out.suffix + ".tmp")
     tmp.write_text(_emit(frontmatter, body_text))
     os.replace(tmp, out)
+
+    for sibling_name, contents in sibling_payloads:
+        _write_sibling(comment_dir, sibling_name, contents)
+
     return out
 
 
+def derive_siblings_from_combined(comment_dir: Path) -> list[Path]:
+    """Backfill: write per-attachment sibling MDs from an existing combined.md.
+
+    For dirs that were extracted before sibling-write existed, this
+    parses combined.md and emits the sibling files without re-decoding
+    the original PDFs/DOCXs. Idempotent — skips siblings that already
+    exist with non-zero size.
+    """
+    combined = comment_dir / _FILENAME
+    if not combined.is_file():
+        return []
+    fm, body = parse_combined(combined.read_text())
+    docket_id = fm.get("docket_id", comment_dir.parent.name)
+    comment_id = fm.get("comment_id", comment_dir.name)
+
+    written: list[Path] = []
+    for filename, section_body in _split_sections(body):
+        if filename.lower() == "inline comment":
+            continue
+        sibling_name = _sibling_name(filename)
+        sibling = comment_dir / sibling_name
+        if sibling.is_file() and sibling.stat().st_size > 0:
+            continue
+        contents = _sibling_text_from_section(
+            filename, comment_id, docket_id, section_body
+        )
+        _write_sibling(comment_dir, sibling_name, contents)
+        written.append(sibling)
+    return written
+
+
 def parse_combined(text: str) -> tuple[dict[str, Any], str]:
     """Split a combined.md into (frontmatter dict, body markdown)."""
     if not text.startswith("---\n"):
@@ -95,17 +143,100 @@ def _attachment_paths(comment_dir: Path) -> list[Path]:
 
 
 def _render_section(filename: str, result: ExtractResult) -> str:
+    return f"## {filename}\n\n{_section_body_for(result)}\n"
+
+
+def _section_body_for(result: ExtractResult) -> str:
     if result.status == "ok":
-        body = result.text.strip()
+        return result.text.strip()
-    elif result.status == "ocr_needed":
+    if result.status == "ocr_needed":
-        body = f"_(ocr_needed — {result.chars} chars extracted)_"
+        return f"_(ocr_needed — {result.chars} chars extracted)_"
-    elif result.status == "failed":
+    if result.status == "failed":
-        body = "_(failed — extraction error, see logs)_"
+        return "_(failed — extraction error, see logs)_"
-    else:  # unsupported
+    return "_(unsupported file type)_"
-        body = "_(unsupported file type)_"
-    return f"## {filename}\n\n{body}\n"
 
 
 def _emit(frontmatter: dict[str, Any], body: str) -> str:
     yaml_text = yaml.safe_dump(frontmatter, sort_keys=False, default_flow_style=False)
     return f"---\n{yaml_text}---\n\n{body}"
+
+
+def _sibling_name(attachment_filename: str) -> str:
+    # Preserve the original extension as part of the sibling stem so
+    # `attachment_1.pdf` and `attachment_1.docx` (rare but possible in
+    # the same dir) don't collide on a shared `attachment_1.md`.
+    return f"{attachment_filename}.md"
+
+
+def _sibling_contents(
+    attachment_filename: str,
+    comment_id: str,
+    docket_id: str,
+    result: ExtractResult,
+) -> str:
+    fm = {
+        "comment_id": comment_id,
+        "docket_id": docket_id,
+        "source_file": attachment_filename,
+        "status": result.status,
+        "chars": result.chars,
+        "extracted_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
+    }
+    return _emit(fm, _section_body_for(result) + "\n")
+
+
+def _sibling_text_from_section(
+    attachment_filename: str,
+    comment_id: str,
+    docket_id: str,
+    section_body: str,
+) -> str:
+    # Backfill path: we only have the rendered section body, not the
+    # original ExtractResult, so chars/status are best-effort.
+    body = section_body.strip()
+    status = "ok"
+    if body.startswith("_(failed"):
+        status = "failed"
+    elif body.startswith("_(ocr_needed"):
+        status = "ocr_needed"
+    elif body.startswith("_(unsupported"):
+        status = "unsupported"
+    fm = {
+        "comment_id": comment_id,
+        "docket_id": docket_id,
+        "source_file": attachment_filename,
+        "status": status,
+        "chars": len(body),
+        "extracted_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
+    }
+    return _emit(fm, body + "\n")
+
+
+def _write_sibling(comment_dir: Path, sibling_name: str, contents: str) -> None:
+    out = comment_dir / sibling_name
+    tmp = out.with_suffix(out.suffix + ".tmp")
+    tmp.write_text(contents)
+    os.replace(tmp, out)
+
+
+def _split_sections(body: str) -> list[tuple[str, str]]:
+    """Split a combined.md body into ``[(filename, section_body), ...]``.
+
+    Sections start with ``## <filename>`` lines. Anything before the
+    first ``## `` line is dropped.
+    """
+    sections: list[tuple[str, str]] = []
+    current_name: str | None = None
+    current_lines: list[str] = []
+    for line in body.splitlines():
+        if line.startswith("## "):
+            if current_name is not None:
+                sections.append((current_name, "\n".join(current_lines).strip()))
+            current_name = line[3:].strip()
+            current_lines = []
+        else:
+            if current_name is not None:
+                current_lines.append(line)
+    if current_name is not None:
+        sections.append((current_name, "\n".join(current_lines).strip()))
+    return sections

src/rex/comments/walker.py

@@ -12,7 +12,7 @@ from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from pathlib import Path
 
-from rex.comments.combine import extract_comment
+from rex.comments.combine import derive_siblings_from_combined, extract_comment
 
 log = logging.getLogger(__name__)
 
@@ -27,7 +27,7 @@ def walk_and_extract(
     limit: int | None = None,
     workers: int = 8,
     force: bool = False,
-    on_combined: Callable[[str, Path], None] | None = None,
+    on_extracted: Callable[[str, Path], None] | None = None,
 ) -> dict[str, int]:
     """Process every comment dir under *root*.
 
@@ -36,10 +36,14 @@ def walk_and_extract(
     ``comment_id`` and returns the inline body text from bib.sqlite (or
     "" if missing).
 
-    *on_combined*, if given, is called on the main thread with
+    *on_extracted*, if given, is called on the main thread with
-    ``(comment_id, combined_md_path)`` for every dir that ends up with a
+    ``(comment_id, comment_dir)`` for every dir that ends up with a
-    combined.md — both newly-written ones and pre-existing skips. Lets
+    combined.md — both newly-written ones and pre-existing skips. The
-    callers attach the markdown to bib without crossing thread boundaries.
+    callback is responsible for discovering whatever markdown files it
+    wants to consume in the dir (combined.md plus per-attachment
+    siblings). Skipped dirs that pre-date the sibling-write feature get
+    siblings derived from combined.md before the callback fires, so the
+    callback always sees a complete set.
 
     Returns counts: ``{"written": N, "skipped": N, "failed": N}``.
     """
@@ -57,13 +61,17 @@ def walk_and_extract(
             for fut in as_completed(futures):
                 status = fut.result()
                 stats[status] += 1
-                if on_combined and status == "written":
+                if on_extracted and status == "written":
                     cdir = futures[fut]
-                    on_combined(cdir.name, cdir / _COMBINED)
+                    on_extracted(cdir.name, cdir)
 
-    if on_combined:
+    if on_extracted:
         for cdir in skipped_dirs:
-            on_combined(cdir.name, cdir / _COMBINED)
+            try:
+                derive_siblings_from_combined(cdir)
+            except Exception as e:  # noqa: BLE001
+                log.warning("sibling backfill failed for %s: %s", cdir, e)
+            on_extracted(cdir.name, cdir)
 
     return stats
 

tests/cli/test_comments.py

@@ -59,8 +59,9 @@ def test_stats_after_extract(tmp_path: Path):
     assert "1" in result.output  # at least one comment counted
 
 
-def test_extract_attaches_combined_md_to_bib(tmp_path: Path, monkeypatch):
+def test_extract_attaches_combined_md_and_siblings_to_bib(tmp_path: Path, monkeypatch):
-    """End-to-end: extract writes combined.md AND attaches it to the bib item."""
+    """End-to-end: extract writes combined.md plus per-attachment sibling MDs,
+    and attaches all of them to the bib item. Idempotent on re-run."""
     from bib import connect
     from bib.item import Source
 
@@ -78,39 +79,27 @@ def test_extract_attaches_combined_md_to_bib(tmp_path: Path, monkeypatch):
     item_key = store.upsert(item)
     store.close()
 
-    result = runner.invoke(
+    result = runner.invoke(app, ["extract", "--root", str(tmp_path), "--workers", "1"])
-        app,
-        [
-            "extract",
-            "--root",
-            str(tmp_path),
-            "--workers",
-            "1",
-        ],
-    )
     assert result.exit_code == 0, result.output
     assert (cdir / "combined.md").is_file()
+    assert (cdir / "attachment_1.pdf.md").is_file()
 
-    store = connect(str(bib_db))
+    def md_attachments():
-    rows = list(
+        store = connect(str(bib_db))
-        store._con().execute(
+        rows = sorted(
-            "SELECT a.filename FROM attachments a JOIN items i ON a.item_id=i.id "
+            r[0]
-            "WHERE i.key=? AND a.filename='combined.md'",
+            for r in store._con().execute(
-            (item_key,),
+                "SELECT a.filename FROM attachments a "
+                "JOIN items i ON a.item_id=i.id "
+                "WHERE i.key=? AND a.filename LIKE '%.md'",
+                (item_key,),
+            )
         )
-    )
+        store.close()
-    store.close()
+        return rows
-    assert len(rows) == 1, "combined.md should be attached exactly once"
 
-    # Re-running must not duplicate the attachment (idempotence)
+    assert md_attachments() == ["attachment_1.pdf.md", "combined.md"]
+
+    # Re-running must not duplicate attachments
     runner.invoke(app, ["extract", "--root", str(tmp_path), "--workers", "1"])
-    store = connect(str(bib_db))
+    assert md_attachments() == ["attachment_1.pdf.md", "combined.md"]
-    rows = list(
-        store._con().execute(
-            "SELECT COUNT(*) FROM attachments a JOIN items i ON a.item_id=i.id "
-            "WHERE i.key=? AND a.filename='combined.md'",
-            (item_key,),
-        )
-    )
-    store.close()
-    assert rows[0][0] == 1, "second run created a duplicate attachment"

tests/rex/comments/test_walker.py

@@ -77,7 +77,7 @@ def test_walk_and_extract_respects_limit(tmp_path: Path):
     assert stats["written"] == 1
 
 
-def test_on_combined_called_for_written_dirs(tmp_path: Path):
+def test_on_extracted_called_for_written_dirs(tmp_path: Path):
     a, b = _setup_two_comments(tmp_path)
     seen: list[tuple[str, Path]] = []
 
@@ -85,27 +85,38 @@ def test_on_combined_called_for_written_dirs(tmp_path: Path):
         tmp_path,
         inline_body_lookup=_stub_inline_body,
         workers=2,
-        on_combined=lambda cid, p: seen.append((cid, p)),
+        on_extracted=lambda cid, d: seen.append((cid, d)),
     )
 
     assert sorted(cid for cid, _ in seen) == sorted([a.name, b.name])
-    assert all(p.name == "combined.md" for _, p in seen)
+    # callback gets the comment dir; sibling + combined.md are inside it
+    for _cid, d in seen:
+        assert (d / "combined.md").is_file()
+        assert (d / "attachment_1.pdf.md").is_file()
 
 
-def test_on_combined_called_for_skipped_dirs(tmp_path: Path):
+def test_on_extracted_called_for_skipped_dirs(tmp_path: Path):
-    """Pre-existing combined.md still gets the callback — so previously
+    """Pre-existing combined.md still gets the callback, and any missing
-    extracted MDs can be backfill-attached on a later run."""
+    sibling MDs are derived from it before the callback fires — so
+    previously-extracted dirs end up with the same set of MDs as
+    freshly-written ones."""
     a, b = _setup_two_comments(tmp_path)
-    (a / "combined.md").write_text("---\n---\nstale\n")
+    # Pre-write a stale combined.md with one section so derive_siblings
+    # has something to split.
+    (a / "combined.md").write_text(
+        "---\ncomment_id: x\ndocket_id: y\n---\n\n## attachment_1.pdf\n\nstale body\n"
+    )
 
     seen: list[str] = []
     stats = walk_and_extract(
         tmp_path,
         inline_body_lookup=_stub_inline_body,
         workers=1,
-        on_combined=lambda cid, _p: seen.append(cid),
+        on_extracted=lambda cid, _d: seen.append(cid),
     )
 
-    assert stats["written"] == 1
+    assert stats["written"] == 1  # b
-    assert stats["skipped"] == 1
+    assert stats["skipped"] == 1  # a
     assert sorted(seen) == sorted([a.name, b.name])
+    # a's sibling was backfilled from its (stale) combined.md
+    assert (a / "attachment_1.pdf.md").is_file()