diff --git a/surfsense_backend/app/services/document_revision_recorder.py b/surfsense_backend/app/services/document_revision_recorder.py index d4ca086d3..1a34efd73 100644 --- a/surfsense_backend/app/services/document_revision_recorder.py +++ b/surfsense_backend/app/services/document_revision_recorder.py @@ -1,12 +1,15 @@ -"""Direct-caller adapter: a document save becomes one knowledge-store revision. +"""Direct-caller adapter: document content changes become knowledge-store revisions. -Editor saves and upload-extracted markdown share this path — the same single -write path agent turns use — behind ``KNOWLEDGE_STORE_ENABLED``. +Editor saves, upload-extracted markdown, and connector sync batches share this +path — the same single write path agent turns use — behind +``KNOWLEDGE_STORE_ENABLED``. """ from __future__ import annotations import logging +from collections.abc import Mapping, Sequence +from typing import TYPE_CHECKING from sqlalchemy.ext.asyncio import AsyncSession @@ -18,25 +21,32 @@ from app.knowledge_store import KnowledgeStore from app.knowledge_store.identities import user_identity from app.knowledge_store.settings import load_knowledge_store_settings +if TYPE_CHECKING: + from app.db import Document + logger = logging.getLogger(__name__) -async def record_document_markdown( +async def record_markdown_files( *, workspace_id: int | str, - store_path: str, - markdown: str, + files: Mapping[str, str], + message: str, author_user_id: str | None, ) -> str | None: - """Record one revision for one saved document; ``None`` when disabled.""" - if not load_knowledge_store_settings().enabled: + """Record ``files`` (store path → markdown) as one revision. + + ``None`` when the store is disabled, the batch is empty, or nothing + actually changed (identical content is a no-op by construction). + """ + if not files or not load_knowledge_store_settings().enabled: return None store = KnowledgeStore.for_workspace(workspace_id) - filename = store_path.rsplit("/", 1)[-1] async with store.transaction( - message=f"docs: save {filename}", author=user_identity(author_user_id) + message=message, author=user_identity(author_user_id) ) as tx: - tx.write(store_path, markdown.encode()) + for path, markdown in files.items(): + tx.write(path, markdown.encode()) return tx.revision @@ -50,7 +60,7 @@ async def record_saved_document( markdown: str, author_user_id: str | None, ) -> str | None: - """Resolve the document's canonical store path and record the save. + """Resolve one document's canonical store path and record the save. Never raises: while the store coexists with the Postgres write path (until the Phase 5 cut), a recording failure must not fail the save @@ -63,10 +73,11 @@ async def record_saved_document( virtual_path = doc_to_virtual_path( doc_id=doc_id, title=title, folder_id=folder_id, index=index ) - return await record_document_markdown( + filename = virtual_path.rsplit("/", 1)[-1] + return await record_markdown_files( workspace_id=workspace_id, - store_path=virtual_path.lstrip("/"), - markdown=markdown, + files={virtual_path.lstrip("/"): markdown}, + message=f"docs: save {filename}", author_user_id=author_user_id, ) except Exception: @@ -77,3 +88,44 @@ async def record_saved_document( exc_info=True, ) return None + + +async def record_prepared_documents( + session: AsyncSession, documents: Sequence[Document] +) -> str | None: + """Record a sync batch's accepted markdown as one revision. + + Called after ``prepare_for_indexing`` commits — the moment content becomes + durable — so chunking/embedding failures can never block the record. + Never raises, for the same coexistence reason as ``record_saved_document``. + """ + if not documents or not load_knowledge_store_settings().enabled: + return None + try: + workspace_id = documents[0].workspace_id + index = await build_path_index(session, workspace_id) + files: dict[str, str] = {} + for doc in documents: + if not doc.source_markdown: + continue + virtual_path = doc_to_virtual_path( + doc_id=doc.id, title=doc.title, folder_id=doc.folder_id, index=index + ) + files[virtual_path.lstrip("/")] = doc.source_markdown + return await record_markdown_files( + workspace_id=workspace_id, + files=files, + message=f"sync: index {len(files)} document(s)", + author_user_id=( + str(documents[0].created_by_id) + if documents[0].created_by_id is not None + else None + ), + ) + except Exception: + logger.warning( + "Knowledge store recording failed for a sync batch in workspace %s", + documents[0].workspace_id, + exc_info=True, + ) + return None diff --git a/surfsense_backend/tests/integration/knowledge_store/test_document_recorder.py b/surfsense_backend/tests/integration/knowledge_store/test_document_recorder.py index de95fff92..2e0b5f0b6 100644 --- a/surfsense_backend/tests/integration/knowledge_store/test_document_recorder.py +++ b/surfsense_backend/tests/integration/knowledge_store/test_document_recorder.py @@ -6,7 +6,7 @@ import pytest from app.config import config as app_config from app.knowledge_store import KnowledgeStore -from app.services.document_revision_recorder import record_document_markdown +from app.services.document_revision_recorder import record_markdown_files pytestmark = pytest.mark.integration @@ -19,10 +19,10 @@ def knowledge_root(tmp_path, monkeypatch): async def test_one_save_records_one_revision(knowledge_root, workspace_id): - revision = await record_document_markdown( + revision = await record_markdown_files( workspace_id=workspace_id, - store_path="documents/notes/meeting.md", - markdown="# Meeting", + files={"documents/notes/meeting.md": "# Meeting"}, + message="docs: save meeting.md", author_user_id="1", ) @@ -34,14 +34,54 @@ async def test_one_save_records_one_revision(knowledge_root, workspace_id): assert "meeting.md" in rev.message +async def test_a_sync_batch_records_one_revision(knowledge_root, workspace_id): + revision = await record_markdown_files( + workspace_id=workspace_id, + files={ + "documents/notion/roadmap.md": "# Roadmap", + "documents/notion/okrs.md": "# OKRs", + }, + message="sync: index 2 document(s)", + author_user_id="1", + ) + + store = KnowledgeStore.for_workspace(workspace_id) + revisions = await store.list_revisions() + assert [r.id for r in revisions] == [revision] + assert await store.read_as_of(revision, "documents/notion/okrs.md") == b"# OKRs" + + +async def test_unchanged_content_records_nothing(knowledge_root, workspace_id): + files = {"documents/notion/roadmap.md": "# Roadmap"} + first = await record_markdown_files( + workspace_id=workspace_id, files=files, message="sync", author_user_id="1" + ) + second = await record_markdown_files( + workspace_id=workspace_id, files=files, message="sync", author_user_id="1" + ) + + assert first is not None + assert second is None + assert len(await KnowledgeStore.for_workspace(workspace_id).list_revisions()) == 1 + + +async def test_empty_batch_records_nothing(knowledge_root, workspace_id): + revision = await record_markdown_files( + workspace_id=workspace_id, files={}, message="sync", author_user_id="1" + ) + + assert revision is None + assert not (knowledge_root / str(workspace_id)).exists() + + async def test_disabled_store_records_nothing(monkeypatch, tmp_path, workspace_id): monkeypatch.setattr(app_config, "KNOWLEDGE_STORE_ENABLED", False) monkeypatch.setattr(app_config, "KNOWLEDGE_STORE_ROOT", str(tmp_path)) - revision = await record_document_markdown( + revision = await record_markdown_files( workspace_id=workspace_id, - store_path="documents/notes/meeting.md", - markdown="# Meeting", + files={"documents/notes/meeting.md": "# Meeting"}, + message="docs: save meeting.md", author_user_id="1", )