feat(rag): audit document ingestion, deletion, and chunk purges

sparkth/core/audit/events.py

@@ -170,3 +170,42 @@ class ToolFailedAuditEvent(AIActionAuditEvent):
     a handler (protocol-level: unknown tool, input validation)."""
 
     event_type: ClassVar[str] = "tool.failed"
+
+
+@AUDIT_EVENTS.register
+@dataclass(frozen=True, slots=True, kw_only=True)
+class RAGDocumentIngestedAuditEvent(MutationAuditEvent):
+    """A document's content entered the RAG corpus, or failed to.
+
+    Recorded by :func:`sparkth.lib.rag.ingest_document` for every attempt:
+    ``target`` is the document, ``change.new`` carries the filename and chunk
+    counts on success, and a failed extraction records outcome ``failure``
+    with the scrubbed error instead.
+    """
+
+    event_type: ClassVar[str] = "rag.document_ingested"
+
+
+@AUDIT_EVENTS.register
+@dataclass(frozen=True, slots=True, kw_only=True)
+class RAGDocumentDeletedAuditEvent(MutationAuditEvent):
+    """A document was removed from the retrieval corpus (soft-deleted).
+
+    Recorded by :func:`sparkth.lib.documents.soft_delete_document` in the
+    caller's transaction; ``change.old`` snapshots the document name.
+    """
+
+    event_type: ClassVar[str] = "rag.document_deleted"
+
+
+@AUDIT_EVENTS.register
+@dataclass(frozen=True, slots=True, kw_only=True)
+class RAGChunksPurgedAuditEvent(MutationAuditEvent):
+    """A cleanup run hard-deleted the chunks of soft-deleted documents.
+
+    The system-actor evidence that content removal actually happened:
+    ``change.old`` lists the processed document ids and the purged chunk
+    count. One event per cleanup run that found work.
+    """
+
+    event_type: ClassVar[str] = "rag.chunks_purged"

sparkth/core/documents/service.py

@@ -5,6 +5,10 @@
 from sqlmodel import col, select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
+from sparkth.core.audit.enums import AuditOutcome
+from sparkth.core.audit.events import RAGDocumentDeletedAuditEvent
+from sparkth.core.audit.recorder import record_event
+from sparkth.core.audit.types import AuditChange, AuditTarget
 from sparkth.core.documents.enums import DocumentStatus
 from sparkth.core.documents.models import Document
 from sparkth.lib.log import get_logger
@@ -82,11 +86,24 @@ async def soft_delete_document(
     session: AsyncSession,
     document_id: int,
 ) -> None:
-    """Soft-delete a Document by id. Does not commit."""
+    """Soft-delete a Document by id. Does not commit.
+
+    Records a ``rag.document_deleted`` audit event in the caller's transaction,
+    so the deletion and its record commit or roll back together. A missing id
+    is a no-op and records nothing.
+    """
     result = await session.exec(select(Document).where(col(Document.id) == document_id))
     doc = result.first()
     if doc is None:
         return
     doc.soft_delete()
     session.add(doc)
+    await record_event(
+        session,
+        RAGDocumentDeletedAuditEvent(
+            outcome=AuditOutcome.SUCCESS,
+            target=AuditTarget(type="document", id=str(document_id)),
+            change=AuditChange(old={"name": doc.name}),
+        ),
+    )
     await session.flush()

sparkth/lib/audit/events.py

@@ -12,6 +12,9 @@
     BaseAuditEvent,
     LoginAuditEvent,
     MutationAuditEvent,
+    RAGChunksPurgedAuditEvent,
+    RAGDocumentDeletedAuditEvent,
+    RAGDocumentIngestedAuditEvent,
     ToolCompletedAuditEvent,
     ToolFailedAuditEvent,
     ToolInvokedAuditEvent,
@@ -28,6 +31,9 @@
     "BaseAuditEvent",
     "LoginAuditEvent",
     "MutationAuditEvent",
+    "RAGChunksPurgedAuditEvent",
+    "RAGDocumentDeletedAuditEvent",
+    "RAGDocumentIngestedAuditEvent",
     "ToolCompletedAuditEvent",
     "ToolFailedAuditEvent",
     "ToolInvokedAuditEvent",

sparkth/rag/cleanup.py

@@ -5,6 +5,9 @@
 from sqlalchemy import delete
 from sqlmodel import col, select
 
+from sparkth.lib.audit import record_event
+from sparkth.lib.audit.context import SystemActor
+from sparkth.lib.audit.events import AuditChange, AuditOutcome, AuditTarget, RAGChunksPurgedAuditEvent
 from sparkth.lib.db import session_scope
 from sparkth.lib.documents import Document
 from sparkth.lib.log import configure_logging, get_logger
@@ -22,6 +25,11 @@ async def cleanup_deleted_documents() -> None:
     Cleanup does NOT hard-delete Document rows — that is the responsibility
     of the plugin that owns the document via soft_delete_document().
 
+    A run that removes at least one chunk link or orphaned chunk records a
+    ``rag.chunks_purged`` audit event, committed atomically with the purge.
+    A run that removes nothing records nothing, so repeat runs over the same
+    already-purged documents do not accumulate empty entries.
+
     This is a system-wide background job: it operates across all users
     intentionally, as orphan cleanup is not scoped per user.
     """
@@ -61,11 +69,35 @@ async def cleanup_deleted_documents() -> None:
         else:
             logger.info("No orphaned chunks found.")
 
-        await session.execute(delete(DocumentChunkLink).where(col(DocumentChunkLink.document_id).in_(deleted_doc_ids)))
+        if candidate_chunk_ids:
+            await session.execute(
+                delete(DocumentChunkLink).where(col(DocumentChunkLink.document_id).in_(deleted_doc_ids))
+            )
 
         if orphan_chunk_ids:
             await session.execute(delete(DocumentChunk).where(col(DocumentChunk.id).in_(orphan_chunk_ids)))
 
+        # Soft-deleted Documents are never hard-deleted here, so they keep matching on
+        # every later run. Only a run that actually removed something is evidence of a
+        # purge; recording the others would fill the trail with empty repeat entries.
+        # candidate_chunk_ids is non-empty exactly when there were links to delete.
+        if candidate_chunk_ids:
+            # System-actor evidence that the corpus removal happened,
+            # committed atomically with the purge itself.
+            await record_event(
+                session,
+                RAGChunksPurgedAuditEvent(
+                    outcome=AuditOutcome.SUCCESS,
+                    actor=SystemActor(label="rag-cleanup"),
+                    target=AuditTarget(type="rag_corpus"),
+                    change=AuditChange(
+                        old={
+                            "document_ids": sorted(deleted_doc_ids),
+                            "purged_chunk_count": len(orphan_chunk_ids),
+                        }
+                    ),
+                ),
+            )
         await session.commit()
         logger.info(
             "Cleanup complete. Deleted %d orphaned chunks.",

sparkth/rag/ingestion/__init__.py

@@ -2,13 +2,49 @@
 
 import asyncio
 
+from sparkth.lib.audit import record_event, record_event_now, scrub_error_detail
+from sparkth.lib.audit.events import (
+    AuditChange,
+    AuditOutcome,
+    AuditTarget,
+    RAGDocumentIngestedAuditEvent,
+)
 from sparkth.lib.db import session_scope
+from sparkth.lib.log import get_logger
 from sparkth.memory_profiler import profile_memory
+from sparkth.rag.exceptions import ScannedPDFError, UnsupportedFileTypeError
 from sparkth.rag.ingestion.chunking import DocumentChunker
 from sparkth.rag.ingestion.extraction import check_extraction_eligibility, extract_to_markdown
 from sparkth.rag.store import ChunkStoreService, store_and_link_chunks
 from sparkth.rag.types import IngestionResult
 
+logger = get_logger(__name__)
+
+
+def _ingested_event(
+    document_id: int,
+    outcome: AuditOutcome,
+    *,
+    change: AuditChange | None = None,
+    error_detail: str | None = None,
+) -> RAGDocumentIngestedAuditEvent:
+    return RAGDocumentIngestedAuditEvent(
+        outcome=outcome,
+        target=AuditTarget(type="document", id=str(document_id)),
+        change=change,
+        error_detail=error_detail,
+    )
+
+
+def _success_change(filename: str, result: IngestionResult) -> AuditChange:
+    return AuditChange(
+        new={
+            "filename": filename,
+            "new_chunks": result.new_chunks,
+            "reused_chunks": result.reused_chunks,
+        }
+    )
+
 
 async def ingest_document(
     filename: str,
@@ -21,6 +57,16 @@ async def ingest_document(
     content-hash dedup) -> link chunks to *document_id*. Opens and commits its
     own database session.
 
+    Audited as a ``rag.document_ingested`` event. A success is recorded in the
+    same transaction as the chunk write (fail-closed, so unrecordable content
+    cannot enter the corpus). A failure is recorded, in its own transaction
+    before the error propagates, for the two eligibility/extraction errors this
+    function declares below; any other error (a parse failure inside an
+    extractor, a chunker error, a database error from the chunk write)
+    propagates without an audit record. Nothing entered the corpus in those
+    cases, so the trail stays accurate about the corpus itself, but it is not a
+    complete log of attempts.
+
     Args:
         document_id: Document.id recorded in the chunk-link table.
         file_bytes: Raw file content.
@@ -33,22 +79,38 @@ async def ingest_document(
         UnsupportedFileTypeError: type the extractors cannot handle.
         ScannedPDFError: PDF appears scanned/image-only.
     """
-    check_extraction_eligibility(filename)
+    try:
+        check_extraction_eligibility(filename)
 
-    async with profile_memory("pipeline_total", file=filename):
-        async with profile_memory("extraction", file=filename, size_bytes=len(file_bytes)):
-            extraction_result = await asyncio.to_thread(extract_to_markdown, file_bytes, filename)
+        async with profile_memory("pipeline_total", file=filename):
+            async with profile_memory("extraction", file=filename, size_bytes=len(file_bytes)):
+                extraction_result = await asyncio.to_thread(extract_to_markdown, file_bytes, filename)
 
-        async with profile_memory("chunking", file=filename, markdown_chars=len(extraction_result.markdown)):
-            chunks = await asyncio.to_thread(DocumentChunker().chunk, extraction_result)
+            async with profile_memory("chunking", file=filename, markdown_chars=len(extraction_result.markdown)):
+                chunks = await asyncio.to_thread(DocumentChunker().chunk, extraction_result)
 
-        if not chunks:
-            return IngestionResult(new_chunks=0, reused_chunks=0)
+            if not chunks:
+                # Nothing was stored, so there is no store transaction to
+                # join; the attempt is still recorded, in its own transaction.
+                result = IngestionResult(new_chunks=0, reused_chunks=0)
+                await record_event_now(
+                    _ingested_event(document_id, AuditOutcome.SUCCESS, change=_success_change(filename, result))
+                )
+                return result
 
-        store = ChunkStoreService()
-        async with session_scope() as session:
-            async with profile_memory("store_and_link", file=filename, chunks=len(chunks)):
-                new_count, reused_count = await store_and_link_chunks(session, document_id, chunks, store)
-            await session.commit()
+            store = ChunkStoreService()
+            async with session_scope() as session:
+                async with profile_memory("store_and_link", file=filename, chunks=len(chunks)):
+                    new_count, reused_count = await store_and_link_chunks(session, document_id, chunks, store)
+                result = IngestionResult(new_chunks=new_count, reused_chunks=reused_count)
+                await record_event(
+                    session,
+                    _ingested_event(document_id, AuditOutcome.SUCCESS, change=_success_change(filename, result)),
+                )
+                await session.commit()
+    except (UnsupportedFileTypeError, ScannedPDFError) as exc:
+        logger.warning("Ingestion of '%s' (document_id=%d) failed: %s", filename, document_id, exc)
+        await record_event_now(_ingested_event(document_id, AuditOutcome.FAILURE, error_detail=scrub_error_detail(exc)))
+        raise
 
-    return IngestionResult(new_chunks=new_count, reused_chunks=reused_count)
+    return result

tests/core/test_documents.py

@@ -12,6 +12,7 @@
     update_document_status,
 )
 from sparkth.lib.documents import list_ready_documents
+from sparkth.lib.testing import AuditEventsFetcher
 
 
 class TestDocumentModel:
@@ -161,3 +162,29 @@ async def test_sets_is_deleted(self, session: AsyncSession) -> None:
 
         assert doc.is_deleted is True
         assert doc.deleted_at is not None
+
+    async def test_records_deletion_audit_event(self, session: AsyncSession, audit_events: AuditEventsFetcher) -> None:
+        """Removing a document from the retrieval corpus is audited in the
+        same transaction as the soft-delete."""
+        doc = await create_document(session, user_id=1, name="report.pdf", mime_type=None)
+        assert doc.id is not None
+        document_id = doc.id
+        await session.commit()
+
+        await soft_delete_document(session, document_id)
+        await session.commit()
+
+        (event,) = await audit_events()
+        assert (event.category, event.action) == ("rag", "document_deleted")
+        assert event.outcome == "success"
+        assert event.target_type == "document"
+        assert event.target_id == str(document_id)
+        assert event.old_values == {"name": "report.pdf"}
+
+    async def test_missing_document_records_no_audit_event(
+        self, session: AsyncSession, audit_events: AuditEventsFetcher
+    ) -> None:
+        await soft_delete_document(session, 999)
+        await session.commit()
+
+        assert await audit_events() == []

tests/rag/test_cleanup.py

@@ -4,6 +4,7 @@
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from sparkth.lib.documents import Document
+from sparkth.lib.testing import AuditEventsFetcher
 from sparkth.rag.cleanup import cleanup_deleted_documents
 from sparkth.rag.models import DocumentChunk, DocumentChunkLink
 
@@ -155,3 +156,63 @@ async def test_deletes_target_the_right_ids(self, session: AsyncSession) -> None
         assert await _chunk_ids(session) == {20}  # chunk 21 orphaned and deleted
         assert await _link_keys(session) == {(6, 20)}  # doc 5's links gone
         assert await _document_ids(session) == {5, 6}
+
+
+class TestCleanupAudit:
+    async def test_purge_run_records_system_audit_event(
+        self, session: AsyncSession, audit_events: AuditEventsFetcher
+    ) -> None:
+        """A cleanup run that processes deleted documents leaves a system-actor
+        audit record of what was purged, atomic with the purge itself."""
+        await _seed(
+            session,
+            _document(1, is_deleted=True),
+            _chunk(10),
+            DocumentChunkLink(document_id=1, chunk_id=10),
+        )
+
+        await cleanup_deleted_documents()
+
+        (event,) = await audit_events()
+        assert (event.category, event.action) == ("rag", "chunks_purged")
+        assert event.outcome == "success"
+        assert event.actor_type == "system"
+        assert event.old_values == {"document_ids": [1], "purged_chunk_count": 1}
+
+    async def test_noop_run_records_no_audit_event(
+        self, session: AsyncSession, audit_events: AuditEventsFetcher
+    ) -> None:
+        """No soft-deleted documents means no state change and no audit record."""
+        await _seed(session, _document(1, is_deleted=False))
+
+        await cleanup_deleted_documents()
+
+        assert await audit_events() == []
+
+    async def test_repeat_run_over_purged_documents_records_nothing_new(
+        self, session: AsyncSession, audit_events: AuditEventsFetcher
+    ) -> None:
+        """Soft-deleted documents stay behind after a purge, so they match again on the
+        next run — but with nothing left to remove, that run adds no audit record."""
+        await _seed(
+            session,
+            _document(1, is_deleted=True),
+            _chunk(10),
+            DocumentChunkLink(document_id=1, chunk_id=10),
+        )
+
+        await cleanup_deleted_documents()
+        await cleanup_deleted_documents()
+
+        (event,) = await audit_events()
+        assert event.old_values == {"document_ids": [1], "purged_chunk_count": 1}
+
+    async def test_deleted_documents_with_nothing_linked_record_no_audit_event(
+        self, session: AsyncSession, audit_events: AuditEventsFetcher
+    ) -> None:
+        """Soft-deleted documents that never had chunks purge nothing, so nothing is recorded."""
+        await _seed(session, _document(1, is_deleted=True), _document(2, is_deleted=True))
+
+        await cleanup_deleted_documents()
+
+        assert await audit_events() == []