fix(memory): select FTS5 matches by composite score (#8005)

* fix(memory): select FTS5 matches by composite score

Score every SQL-filtered match before selecting the final composite top-k.
Keep candidate retention bounded with a streaming heap and preserve the
existing scoring formula and adapter-level filtering behavior.

Add regressions for confidence and time-decay reversals, category filters,
result prefixes, ties, and the public DeerMem storage path.

Fixes #8003

* fix(memory): report matched and retained FTS5 rows

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -3690,6 +3690,8 @@ When `lastConfirmedAt` is absent, hybrid confirmation freshness and staleness re
 
 File-backed memory now separates global user context from agent facts. Each user has one `memory.json` containing only the project-independent `user` and `history` summaries; every fact is a canonical Markdown file below `agents/{agent_name}/facts/`. Existing lead-agent middleware, API, Settings, import/export, and embedded-client calls that omit `agent_name` resolve inside DeerMem to the reserved `__default__` bucket. That bucket is outside the valid custom-agent name grammar, so a real custom agent named `lead-agent` has a separate fact repository and deleting a custom agent cannot delete a memory-only directory without `config.yaml`. Public agent identifiers are case-insensitive and canonicalized to lowercase. Runtime/API readers still receive a compatibility `facts` array for the selected/default agent, so the frontend does not read agent facts from `memory.json`; structured Markdown `source` metadata is projected to the historical string field at the MemoryManager boundary. An unscoped Clear All first migrates facts from unread legacy per-agent JSON without adopting its soon-to-be-cleared summaries, then removes shared summaries and facts from every agent bucket while preserving agent configuration files, so a later read cannot resurrect skipped legacy facts; an explicitly agent-scoped clear removes only that agent's facts. On first normal read, old facts embedded in the user JSON are migrated automatically to `__default__`; facts written to the earlier implicit `lead-agent` bucket are also moved when that directory is not a real custom agent. Migration and normal writes notify the configured retrieval adapter only after durable storage locks are released. DeerMem uses a scope-aware SQLite FTS5/BM25 adapter by default, stores only rebuildable derived index data under `.retrieval/` (or under `memory.backend_config.retrieval_index_path` when set; a relative value resolves against `storage_path`), and rebuilds it in the background during Gateway startup or lazily on the first scoped search. A corrupt derived index is recreated automatically. When several Gateway instances share one `storage_path`, point `retrieval_index_path` at instance-local disk: SQLite's WAL mode is unsupported on network filesystems, each instance re-syncs a user's index on its next search after a peer writes, and a declared multi-instance deployment (`deployment.multi_instance: true` / `DEER_FLOW_MULTI_INSTANCE=1`) that leaves the index under the memory root logs a warning at startup. The Helm chart mounts a Pod-local `emptyDir` for it. Set `memory.backend_config.retrieval_adapter` to an empty string to disable it and use the local substring fallback. Chinese tokenization is optional; install the backend `memory-zh` extra (`uv sync --extra memory-zh`) for jieba-assisted sub-phrase search. Journaled writes, a shared user lock, and optimistic user-memory revisions prevent silent lost updates.
 
+FTS5 search selects the highest combined BM25, time-decay, and confidence scores across all matching facts before applying the requested result count. Requesting fewer results preserves the same ranking.
+
 Set `memory.backend_config.retrieval_relevance_enabled: true` to opt into deterministic relevance/confidence ranking and query-aware injection. This **bypasses `retrieval_adapter` for search**, including the default FTS5/BM25 and custom adapters; indexing remains configured. `retrieval_relevance_weight` controls the blend and `retrieval_diversity_weight` enables near-duplicate penalties (default 0). Scoring uses at most the first 4096 characters and 128 tokens per query/fact. Search diversifies only up to `top_k`; injection diversifies guaranteed and regular facts independently until their token budgets are reached. Leave the feature disabled to retain the existing retrieval and injection behavior.
 
 Lexical relevance measures IDF-weighted coverage of distinct query terms, so

backend/packages/harness/deerflow/agents/memory/AGENTS.md

@@ -250,6 +250,10 @@ to `--user-id` (default `default`); `threads/` go to their `threads_meta` owner.
 DeerMem selects persistent SQLite FTS5 by default.
 An empty value selects the substring fallback.
 
+FTS5 scores all SQL-filtered matches before selecting composite `top_k`.
+Do not pre-limit by BM25 alone. A streaming heap bounds retained Python
+rows, not SQLite ranking work.
+
 The SQLite index is rebuildable derived data below `retrieval_index_path`
 (empty = `{storage_path}/.retrieval`; relative resolves against `storage_path`;
 `paths.retrieval_index_directory` is the one resolver). Instances sharing

backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/retrieval.py

@@ -14,12 +14,15 @@ without importing storage and creating a circular dependency.
 
 from __future__ import annotations
 
+import heapq
 import json
 import logging
 import math
 import re
 import sqlite3
 import threading
+from collections.abc import Iterator
+from contextlib import closing
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
@@ -406,6 +409,7 @@ class FTS5Retrieval:
 
         where_clause = " AND ".join(conditions)
 
+        # Stable heap selection uses this BM25 order to break composite score ties.
         sql = f"""
             SELECT doc_id, content, raw_content, category, scope_user, scope_agent,
                    created_at, confidence, source, fact_json,
@@ -413,20 +417,31 @@ class FTS5Retrieval:
             FROM memory_fts
             WHERE {where_clause}
             ORDER BY bm25_score
-            LIMIT ?
         """
-        params.append(top_k * 2)
+
+        matched_count = 0
+
+        def score_matches(cursor: sqlite3.Cursor) -> Iterator[tuple[float, sqlite3.Row]]:
+            nonlocal matched_count
+            for row in cursor:
+                matched_count += 1
+                score = self._compute_final_score(bm25_score=-row["bm25_score"], confidence=row["confidence"], created_at=row["created_at"])
+                yield score, row
 
         with self._lock:
             try:
-                conn = self._conn
-                rows = conn.execute(sql, params).fetchall()
+                # Hold the shared connection lock through cursor iteration and
+                # composite scoring of every match. The heap bounds retained
+                # rows, while lock time grows with the match count.
+                with closing(self._conn.execute(sql, params)) as cursor:
+                    cursor.row_factory = sqlite3.Row
+                    rows = heapq.nlargest(top_k, score_matches(cursor), key=lambda item: item[0])
             except sqlite3.OperationalError as e:
                 logger.debug("FTS5 query syntax error: %s (query: %s)", e, fts5_query)
                 return None
 
         results: list[dict[str, Any]] = []
-        for row in rows:
+        for score, row in rows:
             (
                 doc_id,
                 indexed_content,
@@ -441,12 +456,6 @@ class FTS5Retrieval:
                 bm25_score,
             ) = row
 
-            score = self._compute_final_score(
-                bm25_score=-bm25_score,  # FTS5 returns negative
-                confidence=confidence,
-                created_at=created_at,
-            )
-
             fact: dict[str, Any] = {}
             if fact_json:
                 try:
@@ -466,17 +475,17 @@ class FTS5Retrieval:
             results.append(fact)
 
         logger.debug(
-            "FTS5 raw SQL: fts5_query=%r scope_user=%r scope_agent=%r category=%r -> %d rows. bm25 raw: %s",
+            "FTS5 composite top-k: fts5_query=%r scope_user=%r scope_agent=%r category=%r matched=%d retained=%d bm25=%s",
             fts5_query,
             scope_user,
             scope_agent,
             category,
+            matched_count,
             len(rows),
             [(r["id"], r["bm25_score"]) for r in results[:10]],
         )
 
-        results.sort(key=lambda r: r["score"], reverse=True)
-        return results[:top_k]
+        return results
 
     def _compute_final_score(
         self,

backend/tests/test_memory_retrieval_adapter.py

@@ -344,3 +344,112 @@ def test_substring_fallback_searches_every_fact_of_a_large_scope(tmp_path: Path)
         assert {"fact-101", "fact-150"} <= returned_ids
     finally:
         storage.close()
+
+
+@pytest.mark.parametrize("top_k", [1, 5])
+@pytest.mark.parametrize("reversal", ["confidence", "time_decay"])
+def test_engine_selects_composite_top_k_before_truncating(top_k: int, reversal: str, caplog: pytest.LogCaptureFixture) -> None:
+    from datetime import UTC, datetime, timedelta
+
+    now = datetime.now(UTC)
+    recent = now.isoformat()
+    older = (now - timedelta(days=3650)).isoformat()
+    engine = FTS5Retrieval()
+    try:
+        for index in range(30):
+            engine.index_fact(
+                f"short-{index:02d}",
+                f"Python Python tool{index:02d} reminder",
+                confidence=0.7 if reversal == "confidence" else 0.8,
+                created_at=recent if reversal == "confidence" else older,
+            )
+        engine.index_fact(
+            "best",
+            "Python deployment preference uses project virtual environments and pinned dependencies",
+            confidence=0.95 if reversal == "confidence" else 0.8,
+            created_at=recent,
+        )
+        complete = engine.search("Python", top_k=31)
+        assert len(complete) == 31
+        assert complete[0]["id"] == "best"
+        # The longer fact has weaker BM25 but the highest composite score.
+        assert complete[0]["bm25_score"] < complete[1]["bm25_score"]
+        assert complete[0]["score"] > complete[1]["score"]
+
+        with caplog.at_level("DEBUG", logger=FTS5Retrieval.__module__):
+            limited = engine.search("Python", top_k=top_k)
+        assert [fact["id"] for fact in limited] == [fact["id"] for fact in complete[:top_k]]
+        assert any(f"matched=31 retained={top_k}" in message for message in caplog.messages)
+    finally:
+        engine.close()
+
+
+@pytest.mark.parametrize("category", [None, "preference"])
+@pytest.mark.parametrize("top_k", [1, 5])
+def test_adapter_composite_ranking_preserves_category_and_limit(category: str | None, top_k: int) -> None:
+    from datetime import UTC, datetime
+
+    adapter = FTS5RetrievalAdapter()
+    scope = {"userId": "alice", "agentName": "agent-a"}
+    recent = datetime.now(UTC).isoformat()
+    try:
+        for index in range(45):
+            adapter.upsert(
+                {**_fact(f"short-{index:02d}", f"Python Python tool{index:02d} reminder", category="preference"), "confidence": 0.7, "createdAt": recent},
+                scope=scope,
+                path="",
+            )
+        adapter.upsert(
+            {**_fact("best", "Python deployment preference uses project virtual environments and pinned dependencies", category="preference"), "confidence": 0.95, "createdAt": recent},
+            scope=scope,
+            path="",
+        )
+        adapter.upsert({**_fact("other-category", "Python", category="context"), "confidence": 1.0, "createdAt": recent}, scope=scope, path="")
+        adapter.upsert({**_fact("nonmatch", "Coffee brewing preference", category="preference"), "confidence": 1.0, "createdAt": recent}, scope=scope, path="")
+        filters = {"category": category} if category else None
+        complete = adapter.search("Python", scopes=[scope], top_k=100, mode="fts5", filters=filters)
+        assert complete[0]["fact"]["id"] == ("best" if category else "other-category")
+        limited = adapter.search("Python", scopes=[scope], top_k=top_k, mode="fts5", filters=filters)
+        assert [row["fact"]["id"] for row in limited] == [row["fact"]["id"] for row in complete[:top_k]]
+        assert all(row["matchType"] == "fts5" for row in limited)
+        assert all("Python" in row["fact"]["content"] for row in limited)
+        if category:
+            assert all(row["fact"]["category"] == category for row in limited)
+    finally:
+        adapter.close()
+
+
+def test_engine_composite_score_ties_keep_bm25_and_insertion_order() -> None:
+    from datetime import UTC, datetime
+
+    engine = FTS5Retrieval()
+    recent = datetime.now(UTC).isoformat()
+    try:
+        for fact_id in ("z-first", "a-second", "middle"):
+            engine.index_fact(fact_id, "Python same content", confidence=0.8, created_at=recent)
+        for top_k in (1, 2, 3, 10):
+            assert [fact["id"] for fact in engine.search("Python", top_k=top_k)] == ["z-first", "a-second", "middle"][:top_k]
+    finally:
+        engine.close()
+
+
+def test_deermem_default_tool_limit_keeps_best_composite_fact(tmp_path: Path) -> None:
+    manager = DeerMem(backend_config={"storage_path": str(tmp_path), "token_counting": "char"})
+    try:
+        for index in range(90):
+            manager.create_fact(f"Python Python tool{index:02d} reminder", confidence=0.7, user_id="alice")
+        _, best_id = manager.create_fact(
+            "Python deployment preference uses project virtual environments and pinned dependencies",
+            confidence=0.95,
+            category="preference",
+            user_id="alice",
+        )
+        complete = manager.search("Python", top_k=100, user_id="alice")
+        assert len(complete) == 91
+        assert complete[0]["id"] == best_id
+        # memory_search's default limit is 10. Changing only the requested
+        # count must not decide whether the highest-scoring fact is reachable.
+        for top_k in (10, 12):
+            assert [fact["id"] for fact in manager.search("Python", top_k=top_k, user_id="alice")] == [fact["id"] for fact in complete[:top_k]]
+    finally:
+        manager.close()