fix(diagnostics): run the duplicate scan off the event loop (F91)

features/diagnostics.py

@@ -163,32 +163,47 @@ async def _audit_contradictions(conn: Any) -> dict[str, Any] | None:
     return None
 
 
-async def _audit_duplicates(conn: Any, user_id: str) -> dict[str, Any] | None:
-    """(b) duplicates: pairwise same-kind, different keys, smart_similarity > 0.85.
+def _pairwise_duplicates(groups: list[list[tuple[str, str, str]]]) -> list[dict[str, Any]]:
+    """Sync pairwise scan over (key, value, kind) triples (F91).
+
+    Runs off the event loop via asyncio.to_thread. Exactly the old loop: skip
+    equal keys, smart_similarity > 0.85, same group/pair order, same element
+    schema.
 
     ponytail: O(n²) pairwise scan — fine at audit scale (<10k facts); blocking or
     FTS prefilter if user memories grow past that.
     """
     from rag.conflict import smart_similarity
 
+    dup_items: list[dict[str, Any]] = []
+    for group in groups:
+        for i, a in enumerate(group):
+            for b in group[i + 1 :]:
+                if a[0] == b[0]:
+                    continue
+                sim = smart_similarity(a[1], b[1])
+                if sim > 0.85:
+                    dup_items.append({"a_key": a[0], "b_key": b[0], "kind": a[2], "similarity": round(sim, 3)})
+    return dup_items
+
+
+async def _audit_duplicates(conn: Any, user_id: str) -> dict[str, Any] | None:
+    """(b) duplicates: pairwise same-kind, different keys, smart_similarity > 0.85.
+
+    F91: the CPU-bound scan is delegated to a worker thread so the MCP server
+    stays responsive; rows are flattened to plain tuples before crossing the
+    thread boundary (sqlite3.Row objects do not cross it).
+    """
     rows = await (
         await conn.execute(
             "SELECT entry_id, memory_kind, key, value FROM core_memory WHERE user_id=? AND memory_kind IS NOT NULL ORDER BY memory_kind",
             (user_id,),
         )
     ).fetchall()
-    by_kind: dict[str, list[Any]] = {}
+    by_kind: dict[str, list[tuple[str, str, str]]] = {}
     for r in rows:
-        by_kind.setdefault(str(r["memory_kind"]), []).append(r)
-    dup_items: list[dict[str, Any]] = []
-    for group in by_kind.values():
-        for i, a in enumerate(group):
-            for b in group[i + 1 :]:
-                if a["key"] == b["key"]:
-                    continue
-                sim = smart_similarity(str(a["value"]), str(b["value"]))
-                if sim > 0.85:
-                    dup_items.append({"a_key": a["key"], "b_key": b["key"], "kind": str(a["memory_kind"]), "similarity": round(sim, 3)})
+        by_kind.setdefault(str(r["memory_kind"]), []).append((str(r["key"]), str(r["value"]), str(r["memory_kind"])))
+    dup_items = await asyncio.to_thread(_pairwise_duplicates, list(by_kind.values()))
     if dup_items:
         return {
             "severity": "warn",

tests/test_features/test_memory_audit.py

@@ -5,6 +5,7 @@
 dead [[fact:key]] links in wiki content.
 """
 
+import asyncio
 import sqlite3
 import time
 
@@ -131,3 +132,44 @@ async def test_diagnose_checks_l0_hash_chain(audit_db):
     res2 = await run_diagnose("u1")
     names2 = {c["name"]: c for c in res2["checks"]}
     assert names2["l0_hash_chain"]["status"] == "fail"
+
+
+async def test_audit_duplicates_does_not_freeze_event_loop(audit_db):
+    """F91: 160 near-identical same-kind facts → 12 720 pairs (~1s naive CPU scan).
+
+    The pairwise duplicate scan must run off the event loop: a heartbeat task
+    ticking every 20ms must never observe a gap >= 0.5s while audit_content
+    works. The first seeded pair is generated first, so the [:100] cap cannot
+    cut it from items.
+    """
+    stem = "the user drinks dark roast coffee with oat milk and one cinnamon stick every morning"
+    for i in range(160):
+        _fact(audit_db, f"dup.{i:03d}", f"{stem} note {i:03d}", kind="preference")
+
+    from features.diagnostics import audit_content
+
+    gaps: list[float] = []
+    stop = asyncio.Event()
+    loop = asyncio.get_running_loop()
+
+    async def heartbeat() -> None:
+        last = loop.time()
+        while not stop.is_set():
+            await asyncio.sleep(0.02)
+            now = loop.time()
+            gaps.append(now - last)
+            last = now
+
+    hb = asyncio.create_task(heartbeat())
+    try:
+        findings = await audit_content("u1")
+    finally:
+        stop.set()
+        await hb  # no dangling tasks
+
+    assert gaps, "heartbeat never ticked"
+    assert max(gaps) < 0.5, f"event loop frozen for {max(gaps):.2f}s during duplicate scan"
+
+    dup = next((c for c in findings if c["type"] == "duplicate"), None)
+    assert dup is not None, "duplicate finding missing"
+    assert any({i["a_key"], i["b_key"]} == {"dup.000", "dup.001"} for i in dup["items"])