Post-commit hooks are best-effort observers

CHANGELOG.md

@@ -3,7 +3,7 @@
 
 ### Added
 
-- Client lifecycle hooks (`after_ingest`, `after_delete`, `before_search`, `after_search`) registered under the `haiku.rag.hooks` entry-point group and activated via the `hooks:` config list.
+- Client lifecycle hooks (`after_ingest`, `after_delete`, `before_search`, `after_search`) registered under the `haiku.rag.hooks` entry-point group and activated via the `hooks:` config list. Post-commit observer hooks are best-effort; failures are logged and do not change the completed operation's result.
 - `SearchResult.annotations` carries notes attached by `after_search` hooks, preserved through context expansion and rendered in agent-facing output.
 
 ## [0.78.0] - 2026-08-24

docs/hooks.md

@@ -60,6 +60,8 @@ An unknown name in `hooks:` raises `ValueError` when the client is constructed,
 
 ## Semantics
 
+- **Post-commit hooks are best-effort observers.** By the time `after_ingest` or `after_delete` runs, the operation has committed. A hook failure is logged, subsequent hooks still run, and the operation still returns success (the ingester proceeds through its normal success path). Correctness-critical derived state therefore needs its own retry or reconciliation, such as the backfill loop below. `before_search` and `after_search` failures propagate: nothing has committed and failing the search is visible to the caller.
+- **Post-commit hooks are not a supported transformation point.** Mutating event models does not alter the committed record; explicit client writes are separate operations and are not atomic with the original write.
 - **Update equals ingest.** `after_ingest` fires for both creation and content updates, with `event.operation` set to `"create"` or `"update"` (creation against an already-stored URI reports `"update"`). Treat both as "replace any state you derived from these documents". The operation is informational, for notification or sync hooks. Metadata-only and title-only updates do not fire.
 - **Batch your writes.** A batch import delivers all its documents in one event. A hook keeping LanceDB state should write once per event, not once per document, to avoid creating a table version per document.
 - **Hooks run after the write commits.** They execute outside the store's write lock, so a hook may itself write to the database, and a hook failure never rolls back the document write.

haiku_rag_slim/haiku/rag/client/__init__.py

@@ -17,7 +17,7 @@ import httpx
 from haiku.rag.client.documents import DocumentImport
 from haiku.rag.config import AppConfig, get_config
 from haiku.rag.converters import get_converter
-from haiku.rag.hooks import DeleteEvent, build_hooks, load_hooks
+from haiku.rag.hooks import DeleteEvent, build_hooks, load_hooks, notify
 from haiku.rag.reranking import get_reranker
 from haiku.rag.store.engine import Store
 from haiku.rag.store.models.chunk import Chunk, SearchResult, SearchType
@@ -424,8 +424,7 @@ class HaikuRAG:
         if self._config.storage.auto_vacuum:
             self._schedule_vacuum()
         event = DeleteEvent(documents=docs_to_delete)
-        for hook in self._hooks:
-            await hook.after_delete(self, event)
+        await notify(self._hooks, "after_delete", self, event)
         return True
 
     async def list_documents(

haiku_rag_slim/haiku/rag/client/documents.py

@@ -16,7 +16,7 @@ from haiku.rag.client.processing import (
 )
 from haiku.rag.client.titles import resolve_title
 from haiku.rag.converters import get_converter
-from haiku.rag.hooks import IngestEvent
+from haiku.rag.hooks import IngestEvent, notify
 from haiku.rag.store.models.chunk import Chunk
 from haiku.rag.store.models.document import Document
 from haiku.rag.store.models.document_item import DocumentItem, extract_items
@@ -162,13 +162,11 @@ async def _store_document_with_chunks(
     if client._config.storage.auto_vacuum:
         client._schedule_vacuum()
 
-
     event = IngestEvent(
         documents=[stored_doc],
         operation="create" if existing is None else "update",
     )
-    for hook in client._hooks:
-        await hook.after_ingest(client, event)
+    await notify(client._hooks, "after_ingest", client, event)
     return stored_doc
 
 
@@ -220,10 +218,8 @@ async def _update_document_with_chunks(
     if client._config.storage.auto_vacuum:
         client._schedule_vacuum()
 
-
     event = IngestEvent(documents=[updated_doc], operation="update")
-    for hook in client._hooks:
-        await hook.after_ingest(client, event)
+    await notify(client._hooks, "after_ingest", client, event)
     return updated_doc
 
 
@@ -326,10 +322,8 @@ async def _store_documents_with_chunks(
     if client._config.storage.auto_vacuum:
         client._schedule_vacuum()
 
-
     event = IngestEvent(documents=created, operation="create")
-    for hook in client._hooks:
-        await hook.after_ingest(client, event)
+    await notify(client._hooks, "after_ingest", client, event)
     return created
 
 

haiku_rag_slim/haiku/rag/hooks.py

@@ -1,7 +1,8 @@
+import logging
 from collections.abc import Callable, Mapping, Sequence
 from dataclasses import dataclass
 from importlib.metadata import entry_points
-from typing import TYPE_CHECKING, Literal, Protocol, runtime_checkable
+from typing import TYPE_CHECKING, Literal, Protocol, overload, runtime_checkable
 
 from haiku.rag.store.models.chunk import SearchResult, SearchType
 from haiku.rag.store.models.document import Document
@@ -13,6 +14,8 @@ if TYPE_CHECKING:
 
 ENTRY_POINT_GROUP = "haiku.rag.hooks"
 
+logger = logging.getLogger(__name__)
+
 IngestOperation = Literal["create", "update"]
 
 
@@ -62,10 +65,19 @@ class Hook:
         document's content was rewritten (including creation against an
         already-stored URI). Replace any state derived from the documents
         regardless of the operation: even a creation may be a retry.
-        Metadata/title-only updates do not fire."""
+        Metadata/title-only updates do not fire.
+
+        Best-effort observer: the operation has already committed, so
+        exceptions are logged and never raised, and subsequent hooks still
+        run. Correctness-critical derived state needs its own
+        reconciliation. Post-commit hooks are not a supported transformation
+        point: mutating event models does not alter the committed record,
+        and explicit client writes are separate operations, not atomic with
+        the original write."""
 
     async def after_delete(self, client: "HaikuRAG", event: DeleteEvent) -> None:
-        """``event.documents`` were deleted; cascades arrive as one event."""
+        """``event.documents`` were deleted; cascades arrive as one event.
+        Best-effort observer with the same contract as ``after_ingest``."""
 
     async def before_search(
         self, client: "HaikuRAG", request: SearchRequest
@@ -86,6 +98,47 @@ class Hook:
         return results
 
 
+@overload
+async def notify(
+    hooks: Sequence[Hook],
+    method: Literal["after_ingest"],
+    client: "HaikuRAG",
+    event: IngestEvent,
+) -> None: ...
+
+
+@overload
+async def notify(
+    hooks: Sequence[Hook],
+    method: Literal["after_delete"],
+    client: "HaikuRAG",
+    event: DeleteEvent,
+) -> None: ...
+
+
+async def notify(
+    hooks: Sequence[Hook],
+    method: Literal["after_ingest", "after_delete"],
+    client: "HaikuRAG",
+    event: IngestEvent | DeleteEvent,
+) -> None:
+    """Fire post-commit observer hooks best-effort: a hook failure is logged
+    and never raised (the operation already committed), and subsequent hooks
+    still run."""
+    for hook in hooks:
+        try:
+            await getattr(hook, method)(client, event)
+        except Exception:
+            cls = type(hook)
+            logger.exception(
+                "%s.%s.%s failed for documents %s",
+                cls.__module__,
+                cls.__qualname__,
+                method,
+                [d.id for d in event.documents],
+            )
+
+
 HookFactory = Callable[[], Hook]
 
 

tests/ingester/test_pipeline.py

@@ -1,4 +1,6 @@
+import logging
 from datetime import UTC, datetime
+from pathlib import Path
 from unittest.mock import AsyncMock
 
 import httpx
@@ -13,10 +15,12 @@ from obstore.exceptions import (
 )
 
 from haiku.rag.client import HaikuRAG
+from haiku.rag.hooks import Hook
 from haiku.rag.ingester.exceptions import PermanentError, TransientError
 from haiku.rag.ingester.queue.models import Job, JobOp, JobStatus
 from haiku.rag.ingester.workers.pipeline import run_job
 from haiku.rag.sources.base import FetchResult, FileTooLargeError, Source
+from haiku.rag.sources.fs import FSSource
 from haiku.rag.store.models.document import Document
 
 
@@ -580,3 +584,37 @@ async def test_other_obstore_errors_classified_transient(exc_class):
     client.create_document_from_source.side_effect = exc_class("upstream hiccup")
     with pytest.raises(TransientError):
         await run_job(client, _job())
+
+
+@pytest.fixture(scope="module")
+def vcr_cassette_dir():
+    return str(Path(__file__).parent.parent / "cassettes" / "test_pipeline")
+
+
+class _ThrowingIngestHook(Hook):
+    async def after_ingest(self, client, event):
+        raise RuntimeError("ingest hook boom")
+
+
+@pytest.mark.asyncio
+@pytest.mark.vcr()
+async def test_after_ingest_hook_failure_does_not_fail_job(
+    temp_db_path, tmp_path, caplog
+):
+    """A throwing post-commit hook must not turn a committed ingest into a
+    failed job: run_job returns a successful JobResult instead of raising
+    into _classify."""
+    file_path = tmp_path / "doc.md"
+    file_path.write_text("hello")
+
+    fs = FSSource(root=tmp_path, source_id="src")
+
+    async with HaikuRAG(temp_db_path, create=True) as client:
+        client._hooks = [_ThrowingIngestHook()]
+
+        with caplog.at_level(logging.ERROR, logger="haiku.rag.hooks"):
+            result = await run_job(client, _job(uri=str(file_path)), sources=[fs])
+
+        assert result.document_id is not None
+        assert result.deleted is False
+        assert any("after_ingest" in r.message for r in caplog.records)

tests/test_hooks.py

@@ -1,3 +1,5 @@
+import logging
+
 import pytest
 
 from haiku.rag.client import HaikuRAG
@@ -372,6 +374,80 @@ async def test_annotations_survive_context_expansion(temp_db_path):
         ]
 
 
+class ThrowingHook(Hook):
+    async def after_ingest(self, client, event):
+        raise RuntimeError("ingest hook boom")
+
+    async def after_delete(self, client, event):
+        raise RuntimeError("delete hook boom")
+
+    async def before_search(self, client, request):
+        raise RuntimeError("search hook boom")
+
+
+@pytest.mark.asyncio
+async def test_after_ingest_hook_failure_is_logged_not_raised(temp_db_path, caplog):
+    spy = RecordingHook()
+    dim = Config.embeddings.model.vector_dim
+
+    async with HaikuRAG(temp_db_path, create=True) as client:
+        client._hooks = [ThrowingHook(), spy]
+
+        with caplog.at_level(logging.ERROR, logger="haiku.rag.hooks"):
+            doc = await client.import_document(
+                _docling_doc("a", "Alpha body"),
+                [Chunk(content="Alpha body", embedding=[0.1] * dim, order=0)],
+                uri="mem://a",
+                title="Alpha",
+            )
+
+        assert doc.id is not None
+        stored = await client.get_document_by_id(doc.id)
+        assert stored is not None
+
+        # Subsequent hooks still run after a failing one.
+        assert spy.events == [("ingest", "create", ((doc.id, "mem://a"),))]
+
+        record = next(r for r in caplog.records if "after_ingest" in r.message)
+        assert "tests.test_hooks.ThrowingHook" in record.message
+        assert str(doc.id) in record.message
+
+
+@pytest.mark.asyncio
+async def test_after_delete_hook_failure_is_logged_not_raised(temp_db_path, caplog):
+    spy = RecordingHook()
+    dim = Config.embeddings.model.vector_dim
+
+    async with HaikuRAG(temp_db_path, create=True) as client:
+        doc = await client.import_document(
+            _docling_doc("a", "Alpha body"),
+            [Chunk(content="Alpha body", embedding=[0.1] * dim, order=0)],
+            uri="mem://a",
+            title="Alpha",
+        )
+        assert doc.id is not None
+        client._hooks = [ThrowingHook(), spy]
+
+        with caplog.at_level(logging.ERROR, logger="haiku.rag.hooks"):
+            assert await client.delete_document(doc.id) is True
+
+        assert await client.get_document_by_id(doc.id) is None
+        assert spy.events == [("delete", ((doc.id, "mem://a"),))]
+
+        record = next(r for r in caplog.records if "after_delete" in r.message)
+        assert "tests.test_hooks.ThrowingHook" in record.message
+        assert str(doc.id) in record.message
+
+
+@pytest.mark.asyncio
+async def test_before_search_hook_failure_propagates(temp_db_path):
+    async with HaikuRAG(temp_db_path, create=True) as client:
+        client._hooks = [ThrowingHook()]
+
+        with pytest.raises(RuntimeError, match="search hook boom"):
+            await client.search("alpha")
+
+
 @pytest.mark.asyncio
 async def test_delete_missing_document_fires_nothing(temp_db_path):
     spy = RecordingHook()