Cover PDF attachment ingest through the full create_document_from_source path

CHANGELOG.md

@@ -1,6 +1,10 @@
 # Changelog
 ## [Unreleased]
 
+### Added
+
+- PDF `/EmbeddedFiles` attachments are ingested as separate Documents linked to the wrapper through `metadata.parent_uri`. Child URIs use a `#attachment=<percent-encoded-name>` fragment on the parent URI. Re-ingest reconciles the child set (add / update / delete) against the wrapper's current attachments; `delete_document` cascades through `parent_uri`. Nested chains are bounded at 3 levels. Toggle with `processing.extract_pdf_attachments` (default `true`).
+
 ### Changed
 
 - A successful DELETE job auto-prunes dead jobs with the same `(source_id, uri)`. New `JobRepo.prune_dead(source_id, uri)`.

docs/configuration/processing.md

@@ -23,6 +23,9 @@ processing:
   chunking_merge_peers: true                 # Merge undersized successive chunks
   chunking_use_markdown_tables: false        # Use markdown tables vs narrative format
 
+  # PDF /EmbeddedFiles attachments
+  extract_pdf_attachments: true              # Ingest embedded files as separate Documents
+
   # Automatic title generation
   auto_title: false                          # Auto-generate titles on ingestion
   title_model:                               # LLM for title generation (fallback)
@@ -368,6 +371,35 @@ Explicit titles passed via `title=` parameter always take precedence and are nev
 
 To generate titles for existing untitled documents, use [`rebuild --title-only`](../cli.md#rebuild-database).
 
+### PDF Embedded Attachments
+
+A PDF can carry other files inside it via the `/EmbeddedFiles` table (signed memos, appendices, supporting documents). With `extract_pdf_attachments: true` (the default), each embedded file is ingested as a separate Document linked to the wrapper through `metadata.parent_uri`:
+
+```yaml
+processing:
+  extract_pdf_attachments: true
+```
+
+```python
+# After ingesting a PDF with two attachments:
+parent = await client.create_document_from_source("/path/to/parent.pdf")
+children = await client.list_documents(
+    filter=f"metadata LIKE '%\"parent_uri\": \"{parent.uri}\"%'"
+)
+# children: 2 Documents, each with parent.uri in metadata.parent_uri,
+# URIs like file:///path/to/parent.pdf#attachment=memo.pdf
+```
+
+Behavior:
+
+- Children inherit the standard ingest metadata (`content_type`, `md5`, `source_revision`) plus `parent_uri`.
+- Re-ingesting the wrapper reconciles its current attachment set against existing children: new files are added, changed bytes update in place, and dropped names are deleted.
+- `delete_document(parent_id)` cascades through `parent_uri` and removes all children.
+- Nested attachments (a PDF whose attachment is itself a PDF with attachments) recurse up to 3 levels. Deeper chains log a warning and skip.
+- Attachments whose extension or content type the converter does not support log a warning and are skipped without aborting the rest of the set.
+
+Set `extract_pdf_attachments: false` to ingest only the wrapper.
+
 ## Continuous ingestion
 
 For automatic ingestion of local directories, S3 buckets, or HTTP

docs/python.md

@@ -80,6 +80,8 @@ doc = await client.create_document_from_source(
 )
 ```
 
+PDFs that carry attachments via the `/EmbeddedFiles` table are split into one Document per attachment, linked to the wrapper through `metadata.parent_uri`. See [PDF Embedded Attachments](configuration/processing.md#pdf-embedded-attachments).
+
 ### Retrieving Documents
 
 By ID:
@@ -167,6 +169,8 @@ await client.update_document(document_id=doc.id, chunks=custom_chunks)
 await client.delete_document(doc.id)
 ```
 
+Deleting a document also removes any child Documents linked to it via `metadata.parent_uri` (PDF attachment children, primarily). The cascade is transitive.
+
 ## Searching Documents
 
 The search method performs native hybrid search (vector + full-text) using LanceDB with optional reranking for improved relevance:

tests/test_pdf_attachments.py

@@ -293,3 +293,122 @@ async def test_cascade_delete_removes_reconciled_children(temp_db_path, monkeypa
 
         await client.delete_document(parent.id)
         assert await client.list_documents() == []
+
+
+async def test_create_document_from_source_extracts_attachments(
+    tmp_path, temp_db_path, monkeypatch
+):
+    """The full create_document_from_source path — the same entry point the
+    ingester worker uses for an UPSERT job — produces parent + child docs
+    when the source is a PDF with embedded files on disk."""
+    monkeypatch.setattr(
+        "haiku.rag.client.documents._ingest_fetch_result",
+        fake_ingest_fetch_result,
+    )
+    pdf_path = tmp_path / "parent.pdf"
+    pdf_path.write_bytes(
+        build_pdf([("notes.txt", b"plain text"), ("data.txt", b"more data")])
+    )
+
+    async with HaikuRAG(temp_db_path, create=True) as client:
+        parent = await client.create_document_from_source(pdf_path)
+
+    async with HaikuRAG(temp_db_path) as client:
+        assert isinstance(parent, Document)
+        children = await client.list_documents(filter=parent_uri_filter(parent.uri))
+        assert len(children) == 2
+        assert {c.metadata["parent_uri"] for c in children} == {parent.uri}
+
+
+async def test_create_document_from_source_reingest_after_attachment_edit(
+    tmp_path, temp_db_path, monkeypatch
+):
+    """Mutate the parent PDF's attachments and re-ingest. The md5 short-circuit
+    must NOT fire (parent bytes changed); reconciliation diffs children to add,
+    update, and delete in one pass while leaving unrelated children untouched."""
+    monkeypatch.setattr(
+        "haiku.rag.client.documents._ingest_fetch_result",
+        fake_ingest_fetch_result,
+    )
+    pdf_path = tmp_path / "parent.pdf"
+    pdf_path.write_bytes(
+        build_pdf(
+            [
+                ("stable.txt", b"unchanged across runs"),
+                ("changed.txt", b"old contents"),
+                ("removed.txt", b"goes away"),
+            ]
+        )
+    )
+
+    async with HaikuRAG(temp_db_path, create=True) as client:
+        parent = await client.create_document_from_source(pdf_path)
+        assert isinstance(parent, Document)
+        before_children = {
+            c.uri: c
+            for c in await client.list_documents(filter=parent_uri_filter(parent.uri))
+        }
+        assert set(before_children) == {
+            f"{parent.uri}#attachment=stable.txt",
+            f"{parent.uri}#attachment=changed.txt",
+            f"{parent.uri}#attachment=removed.txt",
+        }
+        stable_id_before = before_children[f"{parent.uri}#attachment=stable.txt"].id
+        changed_id_before = before_children[f"{parent.uri}#attachment=changed.txt"].id
+
+    pdf_path.write_bytes(
+        build_pdf(
+            [
+                ("stable.txt", b"unchanged across runs"),
+                ("changed.txt", b"new contents"),
+                ("added.txt", b"brand new"),
+            ]
+        )
+    )
+
+    async with HaikuRAG(temp_db_path) as client:
+        await client.create_document_from_source(pdf_path)
+        after_children = {
+            c.uri: c
+            for c in await client.list_documents(filter=parent_uri_filter(parent.uri))
+        }
+
+        assert set(after_children) == {
+            f"{parent.uri}#attachment=stable.txt",
+            f"{parent.uri}#attachment=changed.txt",
+            f"{parent.uri}#attachment=added.txt",
+        }
+        stable = after_children[f"{parent.uri}#attachment=stable.txt"]
+        changed = after_children[f"{parent.uri}#attachment=changed.txt"]
+        assert stable.id == stable_id_before
+        assert changed.id == changed_id_before
+        assert (
+            stable.metadata["md5"]
+            == before_children[f"{parent.uri}#attachment=stable.txt"].metadata["md5"]
+        )
+        assert (
+            changed.metadata["md5"]
+            != before_children[f"{parent.uri}#attachment=changed.txt"].metadata["md5"]
+        )
+
+
+async def test_create_document_from_source_delete_cascades(
+    tmp_path, temp_db_path, monkeypatch
+):
+    """The ingester worker's DELETE path is just client.delete_document(doc.id).
+    A parent ingested via the full pipeline must cascade to its children when
+    that path runs — mirrors what happens when a watched file is removed."""
+    monkeypatch.setattr(
+        "haiku.rag.client.documents._ingest_fetch_result",
+        fake_ingest_fetch_result,
+    )
+    pdf_path = tmp_path / "parent.pdf"
+    pdf_path.write_bytes(build_pdf([("a.txt", b"A"), ("b.txt", b"B")]))
+
+    async with HaikuRAG(temp_db_path, create=True) as client:
+        parent = await client.create_document_from_source(pdf_path)
+        assert isinstance(parent, Document)
+        assert len(await client.list_documents()) == 3
+
+        await client.delete_document(parent.id)
+        assert await client.list_documents() == []