perf: move search persistence off the hot path as one atomic write

Phase 6 persistence gated every /query/hybrid response with ~21+
sequential auto-commit Neo4j queries (SearchQuery node, then one query
per FOUND document link, then one per WebResult). The search_id is now
generated up front and returned immediately; the persistence runs as a
background asyncio task (strong references held against mid-flight GC).

The write itself is collapsed into ONE UNWIND-based execute_write
transaction with aggregating CALL subqueries (so an empty doc-link list
cannot swallow the web-result branch), meaning a mid-way failure can no
longer leave a partial SearchQuery graph behind.

The persisted shape consumed by the consolidation repair loop is
unchanged - SearchQuery {id, query, user, timestamp, processed:false,
total_results, web_count, keywords}, tenant labels, FOUND {rank,
rrf_score} -> WebResult {url, title, content} - and is now pinned by
tests/test_search_persistence.py against exactly what
consolidation_service queries. Tenant scoping of the document MATCH is
preserved and asserted.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L

src/services/hybrid_rag_service.py

@@ -83,6 +83,9 @@ class HybridRAGService:
         self.settings = settings
         self.volatile = volatile_service
         self.reranker_model = settings.ollama_llm_model
+        # Strong references to fire-and-forget persistence tasks so they are
+        # not garbage-collected mid-flight (see Phase 6 in search()).
+        self._background_tasks: set = set()
 
     async def search(
         self,
@@ -182,17 +185,25 @@ class HybridRAGService:
 
         timing["total_ms"] = (time.time() - start_time) * 1000
 
-        # Phase 6: Persistence (async, non-blocking)
-        phase6_start = time.time()
-        search_id = await self._persist_search_for_librarian(
-            query=query,
-            user=user,
-            keywords_data=keywords_data,
-            raw_results=raw_results,
-            final_results=final_results,
-            timing=timing
+        # Phase 6: Persistence — genuinely off the hot path. The search_id is
+        # generated up front and returned immediately; the Neo4j write runs as
+        # a background task (one atomic transaction, see
+        # _persist_search_for_librarian) instead of gating the response.
+        search_id = str(uuid.uuid4())
+        timing["persistence_ms"] = 0.0  # not on the request path anymore
+        persist_task = asyncio.create_task(
+            self._persist_search_for_librarian(
+                search_id=search_id,
+                query=query,
+                user=user,
+                keywords_data=keywords_data,
+                raw_results=raw_results,
+                final_results=final_results,
+                timing=timing
+            )
         )
-        timing["persistence_ms"] = (time.time() - phase6_start) * 1000
+        self._background_tasks.add(persist_task)
+        persist_task.add_done_callback(self._background_tasks.discard)
 
         # Degradation signaling: a failed leg contributes no results, but the
         # response says so instead of silently pretending the leg was empty
@@ -882,6 +893,7 @@ Ranking:"""
 
     async def _persist_search_for_librarian(
         self,
+        search_id: str,
         query: str,
         user: str,
         keywords_data: Dict[str, Any],
@@ -892,10 +904,20 @@ Ranking:"""
         """
         Phase 6: Store search query and results for Librarian processing.
 
-        Creates SearchQuery node in Neo4j with relationships to found documents
-        and web results for offline knowledge consolidation.
+        Creates the SearchQuery node, FOUND links to this tenant's Document
+        nodes, and WebResult nodes in ONE UNWIND-based write transaction
+        (previously ~21+ sequential auto-commit queries), so a mid-way
+        failure can never leave a partial SearchQuery graph behind.
+
+        SHAPE CONTRACT: the consolidation service (consolidation_service.py)
+        consumes exactly this shape — SearchQuery {id, query, user,
+        timestamp, processed:false, total_results, web_count, keywords},
+        (sq)-[f:FOUND {rank, rrf_score}]->(wr:WebResult {url, title,
+        content}) — do not change it without updating both sides
+        (pinned by tests/test_search_persistence.py).
 
         Args:
+            search_id: Pre-generated search ID (already returned to the caller)
             query: Search query
             user: User identifier
             keywords_data: Extracted keywords/synonyms
@@ -904,15 +926,47 @@ Ranking:"""
             timing: Performance timing
 
         Returns:
-            Search ID for tracking
+            Search ID on success, None on failure
         """
         try:
             user_base_label = get_neo4j_user_base_label(user)
             user_doc_label = get_neo4j_user_label(user)
-            search_id = str(uuid.uuid4())
 
-            # Create SearchQuery node
-            create_query = f"""
+            # Links to found wiki documents (top 20).
+            # TENANT ISOLATION: matched against this tenant's Document label
+            # only — an unscoped (d:Document {page_id}) match would attach
+            # FOUND relationships to other tenants' documents that share the
+            # same Wiki.js page id.
+            doc_links = []
+            for rank, result_data in enumerate(final_results[:20], start=1):
+                result = result_data.get("result", {})
+                page_id = result.get("page_id")
+                if page_id:
+                    doc_links.append({
+                        "page_id": page_id,
+                        "source": result_data.get("source_type", "unknown"),
+                        "rank": rank,
+                        "rrf_score": result_data.get("rrf_score", 0),
+                        "final_rank": result_data.get("final_rank", rank)
+                    })
+
+            # Web results as WebResult nodes (top 10)
+            web_links = []
+            web_results = [r for r in final_results[:10] if r.get("result", {}).get("url")]
+            for rank, result_data in enumerate(web_results, start=1):
+                result = result_data.get("result", {})
+                web_links.append({
+                    "url": result.get("url"),
+                    "title": result.get("title", ""),
+                    "content": result.get("content", "")[:1000],  # Truncate
+                    "rank": rank,
+                    "rrf_score": result_data.get("rrf_score", 0)
+                })
+
+            # Single atomic write: node + doc links + web results. The CALL
+            # subqueries aggregate so an empty UNWIND list cannot swallow the
+            # rest of the query.
+            persist_query = f"""
             CREATE (sq:{user_base_label}_SearchQuery:SearchQuery {{
                 id: $search_id,
                 query: $query,
@@ -927,10 +981,39 @@ Ranking:"""
                 synonyms: $synonyms,
                 timing_ms: $timing_ms
             }})
-            RETURN sq.id as id
+            WITH sq
+            CALL {{
+                WITH sq
+                UNWIND $doc_links AS link
+                MATCH (d:{user_doc_label}:Document {{page_id: link.page_id}})
+                MERGE (sq)-[f:FOUND]->(d)
+                SET f.source = link.source,
+                    f.rank = link.rank,
+                    f.rrf_score = link.rrf_score,
+                    f.final_rank = link.final_rank
+                RETURN count(*) AS docs_linked
+            }}
+            CALL {{
+                WITH sq
+                UNWIND $web_links AS wl
+                CREATE (wr:{user_base_label}_WebResult:WebResult {{
+                    url: wl.url,
+                    title: wl.title,
+                    content: wl.content,
+                    search_id: $search_id,
+                    timestamp: datetime()
+                }})
+                CREATE (sq)-[:FOUND {{
+                    source: "web",
+                    rank: wl.rank,
+                    rrf_score: wl.rrf_score
+                }}]->(wr)
+                RETURN count(*) AS web_created
+            }}
+            RETURN sq.id AS id, docs_linked, web_created
             """
 
-            result = await self.graph.neo4j.execute_query(create_query, {
+            await self.graph.neo4j.execute_write(persist_query, {
                 "search_id": search_id,
                 "query": query,
                 "user": user,
@@ -940,67 +1023,11 @@ Ranking:"""
                 "web_count": len(raw_results.get("web", [])),
                 "keywords": keywords_data.get("core_keywords", []),
                 "synonyms": json.dumps(keywords_data.get("synonyms", {})),
-                "timing_ms": timing.get("total_ms", 0)
+                "timing_ms": timing.get("total_ms", 0),
+                "doc_links": doc_links,
+                "web_links": web_links
             })
 
-            # Link to found wiki documents (top 20)
-            for rank, result_data in enumerate(final_results[:20], start=1):
-                result = result_data.get("result", {})
-                page_id = result.get("page_id")
-
-                if page_id:
-                    # TENANT ISOLATION: only link to this tenant's Document
-                    # nodes. An unscoped (d:Document {page_id}) match would
-                    # attach FOUND relationships to other tenants' documents
-                    # that share the same Wiki.js page id.
-                    link_doc_query = f"""
-                    MATCH (sq:{user_base_label}_SearchQuery:SearchQuery {{id: $search_id}})
-                    MATCH (d:{user_doc_label}:Document {{page_id: $page_id}})
-                    MERGE (sq)-[f:FOUND]->(d)
-                    SET f.source = $source,
-                        f.rank = $rank,
-                        f.rrf_score = $rrf_score,
-                        f.final_rank = $final_rank
-                    """
-
-                    await self.graph.neo4j.execute_query(link_doc_query, {
-                        "search_id": search_id,
-                        "page_id": page_id,
-                        "source": result_data.get("source_type", "unknown"),
-                        "rank": rank,
-                        "rrf_score": result_data.get("rrf_score", 0),
-                        "final_rank": result_data.get("final_rank", rank)
-                    })
-
-            # Store web results as WebResult nodes (top 10)
-            web_results = [r for r in final_results[:10] if r.get("result", {}).get("url")]
-            for rank, result_data in enumerate(web_results, start=1):
-                result = result_data.get("result", {})
-                create_web_query = f"""
-                MATCH (sq:{user_base_label}_SearchQuery:SearchQuery {{id: $search_id}})
-                CREATE (wr:{user_base_label}_WebResult:WebResult {{
-                    url: $url,
-                    title: $title,
-                    content: $content,
-                    search_id: $search_id,
-                    timestamp: datetime()
-                }})
-                CREATE (sq)-[:FOUND {{
-                    source: "web",
-                    rank: $rank,
-                    rrf_score: $rrf_score
-                }}]->(wr)
-                """
-
-                await self.graph.neo4j.execute_query(create_web_query, {
-                    "search_id": search_id,
-                    "url": result.get("url"),
-                    "title": result.get("title", ""),
-                    "content": result.get("content", "")[:1000],  # Truncate
-                    "rank": rank,
-                    "rrf_score": result_data.get("rrf_score", 0)
-                })
-
             logger.info(f"Persisted search {search_id} for Librarian processing")
             return search_id