perf(core): defer local note materialization off the accept path

The local accepted-note write materialized the markdown file AND re-indexed it
inline (awaited in the v2 routes), so the accept paid for the DB-first accept
plus the heavy file write + index. A write-load benchmark showed this made
local writes ~3x slower than main with no offsetting benefit, because the
"offline queue" win is cloud/PGQ-specific.

Bring local into parity with cloud: cloud's materialize_write_change enqueues a
PGQ job and returns immediately; local now schedules the file write + index as
an in-process background task and returns the accepted note_content state at
once. note_content remains the write/read-through cache that serves reads; the
markdown file stays the source of truth, written just after the 202. Test mode
keeps materialization inline so existing tests assert file/search state
synchronously. Clear PARITY INVARIANT comments guard the flow on both the
materializer and the router follow-up scheduler so a later change does not
silently re-inline it.

Refs benchmarks/docs/write-load-benchmark.md.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/api/v2/routers/knowledge_router.py

@@ -89,6 +89,14 @@ def _schedule_post_write_followups(
     always runs so a newly written note back-resolves inbound forward references
     that name it — parity with the watcher's relation repair, which the inline
     write path does not otherwise trigger (#1015). Both are no-ops in test mode.
+
+    CLOUD/LOCAL PARITY (do not break): every follow-up here is a core *router
+    scheduler* that the runtime overrides via dependency injection — local runs
+    it in-process (debounced), cloud overrides each to enqueue a PGQ job. Cloud
+    must override BOTH get_entity_vector_sync_scheduler AND
+    get_relation_resolution_scheduler; if a new follow-up scheduler is added
+    here, cloud needs a matching override or it will run the local in-process
+    work on the cloud API server. See basic_memory_cloud api/deps/cloud_overrides.
     """
     if app_config.semantic_search_enabled:
         vector_sync_scheduler.schedule_entity_vector_sync(

src/basic_memory/cloud/note_content_materialization.py

@@ -2,10 +2,12 @@
 
 from __future__ import annotations
 
+import asyncio
 from collections.abc import Mapping
 from dataclasses import dataclass, replace
 from uuid import UUID
 
+from loguru import logger
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from basic_memory import db, file_utils
@@ -40,6 +42,11 @@
 
 LOCAL_NOTE_CONTENT_TENANT_ID = UUID(int=0)
 
+# Strong references to in-flight deferred materialization tasks. asyncio only
+# keeps weak references to tasks, so without this a fire-and-forget background
+# materialization could be garbage-collected mid-write.
+_pending_materializations: set[asyncio.Task[None]] = set()
+
 
 def note_content_payload_file_path(
     payload: RuntimeNoteContentResponsePayload,
@@ -197,15 +204,58 @@ class LocalNoteContentMaterializationProvider:
     session_maker: async_sessionmaker[AsyncSession]
     file_service: FileService
     file_indexer: IndexFileExecutor | None = None
+    test_mode: bool = False
 
     async def materialize_write_change(
         self,
         accepted: RuntimeAcceptedNoteChange[RuntimeNoteContentResponsePayload],
     ) -> RuntimeAcceptedNoteChange[RuntimeNoteContentResponsePayload]:
-        """Write accepted note content to the local filesystem when needed."""
+        """Materialize an accepted note write OFF the accept path.
+
+        Cloud/local parity (DO NOT UNDO): cloud's materialize_write_change
+        enqueues a PGQ job and returns immediately, letting Tigris object storage
+        + indexing catch up asynchronously because S3 writes are slow. Locally we
+        mirror that with an in-process background task. The accept has already
+        persisted note_content (the write/read-through cache that serves reads);
+        here we only schedule writing the markdown file (the source of truth) and
+        indexing it. Writing + indexing the file is the heavy part of a write, so
+        doing it inline reintroduces a ~3x write-load regression
+        (benchmarks/docs/write-load-benchmark.md).
+
+        PARITY INVARIANT: production must defer. Test mode runs inline ONLY so
+        tests can assert file/search state synchronously — never make the
+        production path synchronous to "simplify" this.
+        """
         if accepted.materialization is None:
             return accepted
+        if self.test_mode:
+            return await self._materialize_write_now(accepted)
+        self._schedule_materialization(accepted)
+        return accepted
+
+    def _schedule_materialization(
+        self,
+        accepted: RuntimeAcceptedNoteChange[RuntimeNoteContentResponsePayload],
+    ) -> None:
+        task = asyncio.create_task(self._materialize_write_logged(accepted))
+        _pending_materializations.add(task)
+        task.add_done_callback(_pending_materializations.discard)
+
+    async def _materialize_write_logged(
+        self,
+        accepted: RuntimeAcceptedNoteChange[RuntimeNoteContentResponsePayload],
+    ) -> None:
+        try:
+            await self._materialize_write_now(accepted)
+        except Exception:  # pragma: no cover - defensive background guard
+            logger.exception("Local note materialization failed")
 
+    async def _materialize_write_now(
+        self,
+        accepted: RuntimeAcceptedNoteChange[RuntimeNoteContentResponsePayload],
+    ) -> RuntimeAcceptedNoteChange[RuntimeNoteContentResponsePayload]:
+        if accepted.materialization is None:  # pragma: no cover - guarded by caller
+            return accepted
         storage = LocalNoteContentStorage(self.file_service)
         cleanup_enqueuer = InlineNoteFileDeleteEnqueuer(storage)
         result = await run_note_materialization(

src/basic_memory/deps/services.py

@@ -732,12 +732,19 @@ async def get_note_content_materialization_provider(
     file_service: FileServiceV2ExternalDep,
     file_indexer: IndexFileExecutorV2ExternalDep,
     session_maker: SessionMakerDep,
+    app_config: AppConfigDep,
 ) -> LocalNoteContentMaterializationProvider:
-    """Create the local inline materializer for accepted-note route writes."""
+    """Create the local materializer for accepted-note route writes.
+
+    test_mode keeps materialization inline so tests can assert file/search state
+    synchronously; production defers the file write + index off the accept path
+    for cloud parity (see LocalNoteContentMaterializationProvider).
+    """
     return LocalNoteContentMaterializationProvider(
         session_maker=session_maker,
         file_service=file_service,
         file_indexer=file_indexer,
+        test_mode=app_config.is_test_env,
     )
 
 

tests/cloud/test_note_content_materialization.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from datetime import UTC, datetime
 from hashlib import sha256
 from typing import Any, cast
@@ -86,11 +87,16 @@ def accepted_materialization_change() -> RuntimeAcceptedNoteChange[
 
 def local_materialization_provider(
     indexer: RecordingFileIndexer,
+    *,
+    test_mode: bool = True,
 ) -> LocalNoteContentMaterializationProvider:
+    # test_mode=True keeps materialization inline so these tests can assert the
+    # result synchronously; production defers it to a background task.
     return LocalNoteContentMaterializationProvider(
         session_maker=cast(async_sessionmaker[AsyncSession], object()),
         file_service=cast(FileService, object()),
         file_indexer=indexer,
+        test_mode=test_mode,
     )
 
 
@@ -153,3 +159,48 @@ async def fake_run_note_materialization(
     assert response_payload["last_materialization_error"] == "Refusing to overwrite notes/test.md"
     assert response_payload["file_checksum"] == "external-checksum"
     assert response_payload["sync_error"] == NOTE_CONTENT_EXTERNAL_CHANGE_SYNC_ERROR
+
+
+@pytest.mark.asyncio
+async def test_local_materialization_defers_write_off_the_accept_path(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Production (test_mode=False) returns the accepted DB state at once, writes async.
+
+    Cloud parity: the accept persists note_content and returns 202; the markdown
+    file (source of truth) and its index are written off the request path.
+    """
+    requests: list[RuntimeNoteMaterializationJobRequest] = []
+
+    async def fake_run_note_materialization(
+        request: RuntimeNoteMaterializationJobRequest,
+        **_: Any,
+    ) -> RuntimeNoteMaterializationResult:
+        requests.append(request)
+        return RuntimeNoteMaterializationResult(
+            entity_id=42,
+            status=RuntimeNoteMaterializationStatus.conflict,
+            reason="deferred",
+            file_path="notes/test.md",
+            file_checksum="c",
+        )
+
+    monkeypatch.setattr(
+        note_content_materialization,
+        "run_note_materialization",
+        fake_run_note_materialization,
+    )
+    indexer = RecordingFileIndexer()
+    accepted = accepted_materialization_change()
+
+    result = await local_materialization_provider(
+        indexer, test_mode=False
+    ).materialize_write_change(accepted)
+
+    # Returned immediately with the accepted DB state — no inline write yet.
+    assert result is accepted
+    assert requests == []
+
+    # The write happens off the accept path; drain the background task to confirm.
+    await asyncio.gather(*list(note_content_materialization._pending_materializations))
+    assert len(requests) == 1