diff --git a/features/diagnostics.py b/features/diagnostics.py index 29e77b9..7af7528 100644 --- a/features/diagnostics.py +++ b/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", diff --git a/tests/test_features/test_memory_audit.py b/tests/test_features/test_memory_audit.py index 8aac6f0..544bfa6 100644 --- a/tests/test_features/test_memory_audit.py +++ b/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"])