diff --git a/sparkth/core/audit/events.py b/sparkth/core/audit/events.py
index 7c304c3e..9b7520c0 100644
--- a/sparkth/core/audit/events.py
+++ b/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"
diff --git a/sparkth/core/documents/service.py b/sparkth/core/documents/service.py
index c6bc84fd..9aecd974 100644
--- a/sparkth/core/documents/service.py
+++ b/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()
diff --git a/sparkth/lib/audit/events.py b/sparkth/lib/audit/events.py
index ad876126..b127eb92 100644
--- a/sparkth/lib/audit/events.py
+++ b/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",
diff --git a/sparkth/rag/cleanup.py b/sparkth/rag/cleanup.py
index 410c380d..9b2ae5b9 100644
--- a/sparkth/rag/cleanup.py
+++ b/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.",
diff --git a/sparkth/rag/ingestion/__init__.py b/sparkth/rag/ingestion/__init__.py
index 697b2698..3f609ad7 100644
--- a/sparkth/rag/ingestion/__init__.py
+++ b/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
diff --git a/tests/core/test_documents.py b/tests/core/test_documents.py
index a4773538..e20a4b27 100644
--- a/tests/core/test_documents.py
+++ b/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() == []
diff --git a/tests/rag/test_cleanup.py b/tests/rag/test_cleanup.py
index f6fcd101..f9c5536b 100644
--- a/tests/rag/test_cleanup.py
+++ b/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() == []
diff --git a/tests/rag/test_ingestion_audit.py b/tests/rag/test_ingestion_audit.py
new file mode 100644
index 00000000..0f7e7c2e
--- /dev/null
+++ b/tests/rag/test_ingestion_audit.py
@@ -0,0 +1,75 @@
+"""Audit capture for RAG ingestion.
+
+Every ``ingest_document`` call must leave an audit record of the target
+document and outcome: content entering the retrieval corpus (or failing to)
+is never silent.
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from sparkth.lib.audit.events import RAGDocumentIngestedAuditEvent
+from sparkth.lib.audit.hooks import AUDIT_EVENTS
+from sparkth.lib.rag import ScannedPDFError, UnsupportedFileTypeError, ingest_document
+from sparkth.lib.testing import AuditEventsFetcher
+
+
+def test_ingested_event_type_is_registered() -> None:
+    assert AUDIT_EVENTS.resolve("rag.document_ingested") is RAGDocumentIngestedAuditEvent
+
+
+class TestIngestionAudit:
+    async def test_successful_ingestion_records_success_event(self, audit_events: AuditEventsFetcher) -> None:
+        extraction = MagicMock(markdown="# H\ntext")
+        chunker = MagicMock()
+        chunker.return_value.chunk.return_value = [MagicMock()]
+        with (
+            patch("sparkth.rag.ingestion.extract_to_markdown", return_value=extraction),
+            patch("sparkth.rag.ingestion.DocumentChunker", chunker),
+            patch("sparkth.rag.ingestion.store_and_link_chunks", return_value=(2, 1)),
+        ):
+            await ingest_document("a.txt", b"x", 10)
+
+        (event,) = await audit_events()
+        assert (event.category, event.action) == ("rag", "document_ingested")
+        assert event.outcome == "success"
+        assert event.target_type == "document"
+        assert event.target_id == "10"
+        assert event.new_values == {"filename": "a.txt", "new_chunks": 2, "reused_chunks": 1}
+
+    async def test_empty_extraction_still_records_success(self, audit_events: AuditEventsFetcher) -> None:
+        """A document that yields no chunks still entered the pipeline; the
+        zero-chunk outcome is part of the trail."""
+        extraction = MagicMock(markdown="")
+        chunker = MagicMock()
+        chunker.return_value.chunk.return_value = []
+        with (
+            patch("sparkth.rag.ingestion.extract_to_markdown", return_value=extraction),
+            patch("sparkth.rag.ingestion.DocumentChunker", chunker),
+        ):
+            await ingest_document("a.txt", b"x", 10)
+
+        (event,) = await audit_events()
+        assert event.outcome == "success"
+        assert event.new_values == {"filename": "a.txt", "new_chunks": 0, "reused_chunks": 0}
+
+    async def test_unsupported_file_type_records_failure(self, audit_events: AuditEventsFetcher) -> None:
+        with pytest.raises(UnsupportedFileTypeError):
+            await ingest_document("img.png", b"x", 10)
+
+        (event,) = await audit_events()
+        assert (event.category, event.action) == ("rag", "document_ingested")
+        assert event.outcome == "failure"
+        assert event.target_id == "10"
+        assert event.error_detail
+        assert event.new_values is None
+
+    async def test_scanned_pdf_records_failure(self, audit_events: AuditEventsFetcher) -> None:
+        with patch("sparkth.rag.ingestion.extract_to_markdown", side_effect=ScannedPDFError("a.pdf")):
+            with pytest.raises(ScannedPDFError):
+                await ingest_document("a.pdf", b"x", 10)
+
+        (event,) = await audit_events()
+        assert event.outcome == "failure"
+        assert event.error_detail