From 2985127f45baf426be241b10220be4c41f5a95cb Mon Sep 17 00:00:00 2001 From: Hamza Shafique Date: Thu, 6 Aug 2026 03:58:41 +0500 Subject: [PATCH 1/2] feat(rag): audit document ingestion, deletion, and chunk purges MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RAG content entered and left the retrieval corpus with no record of who did it, when, or whether it succeeded (issue #502, epic #435) — a gap under EU AI Act Art. 12 provenance obligations. Three new mutation event types close it: - rag.document_ingested: recorded by ingest_document for every attempt; success joins the chunk-store transaction (fail-closed, so content that cannot be recorded cannot enter the corpus), while ineligible or unextractable files record a failure in its own transaction before the error propagates. - rag.document_deleted: recorded by soft_delete_document in the caller's transaction, attributing corpus removal to the acting user. - rag.chunks_purged: a system-actor event per cleanup run, committed atomically with the hard delete, evidencing that removal happened. Co-Authored-By: Claude Fable 5 --- sparkth/core/audit/events.py | 39 ++++++++++++++ sparkth/core/documents/service.py | 19 ++++++- sparkth/lib/audit/events.py | 6 +++ sparkth/rag/cleanup.py | 19 +++++++ sparkth/rag/ingestion/__init__.py | 86 ++++++++++++++++++++++++++----- tests/core/test_documents.py | 27 ++++++++++ tests/rag/test_cleanup.py | 33 ++++++++++++ tests/rag/test_ingestion_audit.py | 75 +++++++++++++++++++++++++++ 8 files changed, 289 insertions(+), 15 deletions(-) create mode 100644 tests/rag/test_ingestion_audit.py 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..12322bc0 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..cac1abe5 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 @@ -66,6 +69,22 @@ async def cleanup_deleted_documents() -> None: if orphan_chunk_ids: await session.execute(delete(DocumentChunk).where(col(DocumentChunk.id).in_(orphan_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..7e68c1aa 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,12 @@ async def ingest_document( content-hash dedup) -> link chunks to *document_id*. Opens and commits its own database session. + Every attempt is 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) and an + ineligible or unextractable file is recorded immediately in its own + transaction before the error propagates. + Args: document_id: Document.id recorded in the chunk-link table. file_bytes: Raw file content. @@ -33,22 +75,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..e0e1b038 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,35 @@ 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() == [] 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 From db90834bdbf2399ba4610af18b38a02386cd0642 Mon Sep 17 00:00:00 2001 From: Hamza Shafique Date: Thu, 6 Aug 2026 18:05:31 +0500 Subject: [PATCH 2/2] fix(rag): only record a chunks_purged event when a run purges something Review follow-up on #577. Cleanup never hard-deletes the soft-deleted Document rows it processes, so those rows matched on every subsequent run and each run recorded a rag.chunks_purged event with purged_chunk_count 0 and the same document_ids, indefinitely. The record (and the link delete) is now guarded on there having been links to remove this run, so the trail carries one entry per real purge. Also reconciles the ingest_document docstring with the code: only the two declared eligibility/extraction errors produce a failure event, so the docstring no longer claims every attempt is audited, and fixes the mis-wrapped soft_delete_document docstring. --- sparkth/core/documents/service.py | 6 ++--- sparkth/rag/cleanup.py | 45 ++++++++++++++++++++----------- sparkth/rag/ingestion/__init__.py | 14 ++++++---- tests/rag/test_cleanup.py | 28 +++++++++++++++++++ 4 files changed, 69 insertions(+), 24 deletions(-) diff --git a/sparkth/core/documents/service.py b/sparkth/core/documents/service.py index 12322bc0..9aecd974 100644 --- a/sparkth/core/documents/service.py +++ b/sparkth/core/documents/service.py @@ -88,9 +88,9 @@ async def soft_delete_document( ) -> None: """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. + 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() diff --git a/sparkth/rag/cleanup.py b/sparkth/rag/cleanup.py index cac1abe5..9b2ae5b9 100644 --- a/sparkth/rag/cleanup.py +++ b/sparkth/rag/cleanup.py @@ -25,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. """ @@ -64,27 +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))) - # 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), - } + # 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 7e68c1aa..3f609ad7 100644 --- a/sparkth/rag/ingestion/__init__.py +++ b/sparkth/rag/ingestion/__init__.py @@ -57,11 +57,15 @@ async def ingest_document( content-hash dedup) -> link chunks to *document_id*. Opens and commits its own database session. - Every attempt is 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) and an - ineligible or unextractable file is recorded immediately in its own - transaction before the error propagates. + 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. diff --git a/tests/rag/test_cleanup.py b/tests/rag/test_cleanup.py index e0e1b038..f9c5536b 100644 --- a/tests/rag/test_cleanup.py +++ b/tests/rag/test_cleanup.py @@ -188,3 +188,31 @@ async def test_noop_run_records_no_audit_event( 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() == []