feat(memory): built-in FTS5/BM25 retrieval adapter (#4360)

* feat(memory): integrate FTS5 retrieval adapter

* deps: add jieba as default dependency for Chinese tokenization

Without jieba, FTS5 unicode61 tokenizer treats entire Chinese sentences
as single tokens, making single-character or sub-phrase searches
impossible (e.g. '吃' or '油泼面' returns 0 hits against
'用户喜欢吃油泼面'). jieba segments Chinese text into meaningful tokens
before indexing.

* fix(memory): avoid treating hyphens as FTS5 operators

* feat(memory): make Chinese tokenization optional

* fix(memory): warm every requested retrieval scope

* fix(memory): close retrieval resources on shutdown

* fix(memory): close backend when shutdown flush fails

* fix(memory): recreate corrupt retrieval index

* fix(memory): tolerate partial retrieval rebuilds

* fix(memory): warm retrieval index in background

* fix(memory): preserve shutdown flush budget

* fix(memory): stop retrying partial lazy rebuilds

* fix(memory): close retrieval through storage

* refactor(memory): simplify retrieval scope limit

* docs(memory): clarify retrieval shutdown lifecycle

---------

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

README.md

@@ -941,7 +941,7 @@ Across sessions, DeerFlow builds a persistent memory of your profile, preference
 
 Memory updates now skip duplicate fact entries at apply time, so repeated preferences and context do not accumulate endlessly across sessions.
 
-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 a configured retrieval adapter only after durable storage locks are released. Journaled writes, a shared user lock, and optimistic user-memory revisions prevent silent lost updates. Retrieval engines remain optional behind the memory retrieval-adapter contract, with an explicit local substring fallback when no adapter is configured.
+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/`, and rebuilds it in the background during Gateway startup or lazily on the first scoped search. A corrupt derived index is recreated automatically. 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.
 
 Single-fact repository operations are genuinely incremental: an upsert/delete reads, journals, writes, and re-indexes only the addressed fact files, and returns an explicit incomplete delta rather than a cache-dependent fake full document. Summary change sets merge the supplied `user`/`history` child keys over the persisted sections so a partial update cannot erase omitted siblings; full imports normalize both sections to the complete compatibility schema before applying replacement values. Manager/API compatibility methods materialize a fresh full document only when their public response contract requires one. Fact-level point operations use separate expected user-memory and fact revisions and may explicitly rebase when every addressed fact precondition still holds. Snapshot-derived operations such as scoped clear, capped create, consolidation, and trimming never replay stale delete/trim sets: a manifest conflict reloads the complete document and recomputes the operation, with a bounded retry. Fact paths use the first two hexadecimal characters of `SHA-256(fact_id)` so generated `fact_*` IDs distribute across shards. The cache token combines the shared JSON's nanosecond mtime, size, and persisted revision; this prevents coarse-mtime same-size writes from returning stale data without scanning fact files. Direct out-of-band Markdown edits require an explicit reload. Storage-specific conflicts and corruption are translated at the MemoryManager boundary; the Gateway returns conflict as HTTP 409 and a stable, non-sensitive corruption error as HTTP 500. Full-document `save()` remains a compatibility API and computes a diff before writing; malformed or missing `facts` can no longer silently erase an agent's Markdown files. Legacy migration preserves non-empty `user`/`history` before deleting an agent `memory.json`; conflicting summaries keep the legacy file and fail loudly instead of choosing a winner.
 

backend/AGENTS.md

@@ -800,7 +800,8 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_
 - `updater.py` - LLM-based memory updates with fact extraction, whitespace-normalized fact deduplication, optimistic revision checks, and repository change sets
 - `queue.py` - Debounced update queue (per-thread deduplication, configurable wait time); captures `user_id` at enqueue time so it survives the `threading.Timer` boundary
 - `prompt.py` - Prompt templates for memory updates
-- `storage.py` - File repository with one user-global summary JSON, agent-owned single-fact Markdown, target-only journaled changes, strict fact validation, shared-user plus per-fact optimistic revisions, lock-protected migration, deep-copy caching, and an optional retrieval adapter
+- `storage.py` - File repository with one user-global summary JSON, agent-owned single-fact Markdown, target-only journaled changes, strict fact validation, shared-user plus per-fact optimistic revisions, lock-protected migration, deep-copy caching, and a RetrievalPort adapter boundary
+- `retrieval.py` - Built-in scope-aware SQLite FTS5/BM25 adapter; it stores only rebuildable derived data and can be disabled with an empty `retrieval_adapter`. Chinese jieba tokenization is optional via the backend `memory-zh` extra; without it the adapter uses SQLite unicode tokenization and the substring fallback. A corrupt persistent derived database is deleted and recreated once before falling back to substring retrieval. The Gateway closes the derived SQLite connection after its shutdown flush; reads and writes remain serialized by the adapter lock, with connection pooling deferred as a performance follow-up.
 - `tools.py` - Tool-driven memory mode (`memory_search`, `memory_add`, `memory_update`, `memory_delete`) using the same storage/update primitives
 
 **Per-User Isolation**:
@@ -829,7 +830,7 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_
 - Both modes share `FileMemoryStorage`, per-user/per-agent isolation, prompt injection, manual CRUD primitives, and the updater backend.
 - Middleware mode queue debounces (30s default), batches updates, and commits global summaries plus the selected/default agent's fact delta through a user-level lock, optimistic user-memory revisions, per-fact revisions, and a recoverable target-file journal. Only explicitly marked point operations may rebase a stale shared revision, and only while every addressed fact still satisfies its original absent/revision precondition. Snapshot-derived clear/trim/consolidation operations instead reload the complete document and recompute their intent on a manifest conflict, with a bounded retry. Typed manifest/fact conflict subclasses keep that decision independent of exception text, and same-ID creates and stale same-fact writes fail. Scope-lock objects are weakly cached so inactive users do not grow a process-lifetime map. Cache validation does not scale with the fact-file count: its token combines the shared JSON's `(mtime_ns, size, revision)`, so the persisted revision invalidates stale caches even when a coarse-mtime filesystem reports identical metadata for same-size writes; direct out-of-band Markdown edits require `reload()`. Atomic replacement also syncs the parent directory on POSIX so the rename is durable. DeerMem translates private storage conflict/corruption exceptions to the backend-neutral MemoryManager contract; the Gateway maps them to HTTP 409 and a stable HTTP 500 response respectively. A normal default-manager read automatically migrates legacy facts from the global JSON into `__default__`; it also adopts the earlier implicit `lead-agent` fact bucket only when that directory has no custom-agent `config.yaml`, and rejects unexpected files instead of deleting them. The v1-to-v2 migration is one-way for the running application: operators must stop DeerFlow and snapshot the configured storage root before upgrade. Before any destructive v2 write, every migrated JSON source is durably retained as `{manifest_filename}.v1.bak`; a missing-write or mismatched existing backup aborts without modifying v1 data. Legacy per-agent JSON is deleted only after its non-empty summaries are safely adopted or confirmed identical; summary conflicts keep the source file and fail loudly.
 - **Proactive Markdown migration CLI**: from `backend/`, run `PYTHONPATH=. python scripts/migrate_memory_markdown.py --all-users --dry-run` to audit and omit `--dry-run` to migrate before serving traffic. Use repeated `--user-id` values when selecting exact original identities, especially standalone raw IDs containing `@` or other characters that are normalized in directory names; `--storage-path` selects a non-default DeerMem root. The CLI reuses `FileMemoryStorage.migrate`, is idempotent, continues across per-user failures, and exits non-zero if any user fails. It is optional because the first normal read still performs the same migration automatically.
-- A configured `retrieval_adapter` owns indexing and semantic retrieval. File storage sends upsert/remove notifications for normal writes and both explicit and lazy migrations after releasing durable storage locks, then delegates search; without an adapter it declares and uses `substring_fallback`.
+- `retrieval_adapter` owns indexing and retrieval. `fts5` is the DeerMem default and uses a persistent derived SQLite index under `.retrieval/`; an empty value disables the adapter and selects `substring_fallback`. File storage sends upsert/remove notifications for normal writes and both explicit and lazy migrations after releasing durable storage locks, then delegates search. Gateway startup schedules `DeerMem.warm_retrieval()` as a background full rebuild so readiness is not delayed, while a first search lazily rebuilds its exact scope until warm-up completes. Individual malformed facts are logged and skipped without triggering repeated full scans; only a fatal adapter rebuild failure keeps lazy retry enabled. During shutdown, the Gateway waits at most one second for this derived rebuild and leaves the full configured timeout to the canonical memory flush; if the rebuild is still active, its adapter remains open until process exit. Adapter failures mark the scope dirty and fall back to canonical substring search until rebuilding succeeds. `FileMemoryStorage` owns and closes the adapter so higher layers do not reach into private storage state.
 - Staleness pass (same LLM invocation as the regular updater, no extra API call): when `staleness_review_enabled` is `true` and at least `staleness_min_candidates` aged facts exist, `_select_stale_candidates` selects facts older than their individual review window (`expected_valid_days`, or the global `staleness_age_days` fallback) that are not in `staleness_protected_categories` (default: `correction`), surfaces them in the prompt with a `valid:Nd` annotation, and the LLM judges each as KEEP, REMOVE, or EXTEND. REMOVE entries go in `staleFactsToRemove`; EXTEND entries go in `staleFactsToExtend` with an `extend_by_days` value, which sets the fact's `expected_valid_days` to `min(days_since_created + extend_by_days, staleness_max_extension_days)`. The LLM assigns `expected_valid_days` when creating a fact; it is clamped at write time to `staleness_age_days × staleness_max_lifetime_multiplier` (creation cap). `_apply_updates` enforces the guardrail unconditionally at apply time: it intersects both the removal and extension sets with `_select_stale_candidates` output before applying the per-cycle cap (`staleness_max_removals_per_cycle`), so protected and non-aged facts can never be targeted regardless of model behavior or the feature flag setting. Facts the LLM proposed for removal are excluded from extension even if the per-cycle cap prevented their actual deletion that cycle. Extensions use an absolute ceiling (`staleness_max_extension_days`) rather than the creation multiplier so a deliberate review decision can advance the window beyond the initial cap while preventing `timedelta` overflow from a malformed `extend_by_days`.
 - Consolidation pass (same LLM invocation as the regular updater, no extra API call): when `consolidation_enabled` is `true` and at least one category holds `consolidation_min_facts` or more facts, `_select_consolidation_candidates` identifies fragmented categories and surfaces at most `consolidation_max_groups_per_cycle` of them (largest first) in the prompt. The LLM decides which groups to merge and proposes a synthesised fact per group. `_apply_updates` enforces guardrails: source IDs must exist and must not overlap across groups, group size is capped at `consolidation_max_sources`, the merged fact's confidence cannot exceed the source maximum, and facts below `fact_confidence_threshold` are not written. The merged fact carries the newest source's `createdAt` (so the staleness clock reflects the underlying information, not synthesis time) and inherits `expected_valid_days` set so the merged fact is re-reviewed at the earliest source review deadline (`min(createdAt + effective_lifetime)` across sources, where a source's effective lifetime is its `expected_valid_days` or the global `staleness_age_days` fallback for legacy facts without one - so a legacy source's default window is not swallowed by a long-lived sibling), relative to the merged `createdAt`, clamped to a minimal positive window if a source is already past its deadline, then capped at the creation-time `staleness_max_lifetime_multiplier`; this keeps a volatile or legacy sub-detail from inheriting a stable source's long window and escaping staleness review for years, while a merge of uniformly stable sources does not re-enter review prematurely.
 - Next interaction injects selected facts + context into `<memory>` tags in the system prompt when `injection_enabled` is true.
@@ -855,9 +856,9 @@ Focused regression coverage for the updater lives in `backend/tests/test_memory_
 - `strict_user_scope` - Require `user_id` for all storage access (default `false` for no-auth/legacy compatibility)
 - `manifest_filename` - User-global summary JSON filename (kept for configuration compatibility)
 - `file_lock_timeout_seconds` - Scope-lock wait; Markdown facts and the recovery journal are required storage invariants rather than configurable modes
-- `retrieval_adapter` - Optional dotted factory receiving `DeerMemConfig` and returning a retrieval-port implementation
+- `retrieval_adapter` - `fts5` by default, empty to disable, or a dotted factory receiving `DeerMemConfig` and returning a retrieval-port implementation
 - `debounce_seconds` - Wait time before processing (default: 30)
-- `shutdown_flush_timeout_seconds` - Host-shared hard budget (seconds) to drain the memory backend's pending-update buffer on Gateway graceful shutdown (default: 30; 1–300). Each pending item does one LLM call, so large IM batches may need more. The Gateway lifespan calls `MemoryManager.shutdown_flush(timeout)` after channels/scheduler stop; the backend short-circuits on an idle buffer, so the host calls it unconditionally (no pending/processing gate). Must fit inside the pod's K8s `terminationGracePeriodSeconds` (gateway Helm chart sets this; default 45s) or K8s SIGKILLs the drain mid-flight.
+- `shutdown_flush_timeout_seconds` - Hard budget (seconds) reserved for draining the memory backend's pending-update buffer on Gateway graceful shutdown (default: 30; 1–300). Each pending item does one LLM call, so large IM batches may need more. The Gateway lifespan calls `MemoryManager.shutdown_flush(timeout)` after channels/scheduler stop and after waiting at most one additional second for the derived retrieval warm-up; the backend short-circuits on an idle buffer, so the host calls it unconditionally (no pending/processing gate). The retrieval wait does not reduce this canonical flush budget. The combined shutdown hooks, brief retrieval wait, flush budget, and scheduling margin must fit inside the pod's K8s `terminationGracePeriodSeconds` (gateway Helm chart default: 45s) or K8s SIGKILLs the drain mid-flight.
 - `model_name` - LLM for updates (null = default model)
 - `max_facts` / `fact_confidence_threshold` - Fact storage limits (100 / 0.7)
 - `max_injection_tokens` - Token limit for prompt injection (2000)

backend/app/gateway/app.py

@@ -60,6 +60,10 @@ logger = logging.getLogger(__name__)
 # firing signals into a worker that is stuck waiting for shutdown cleanup.
 _SHUTDOWN_HOOK_TIMEOUT_SECONDS = 5.0
 
+# The retrieval index is derived state, so shutdown only waits briefly for its
+# startup rebuild. The canonical memory flush keeps its full configured budget.
+_RETRIEVAL_WARM_SHUTDOWN_TIMEOUT_SECONDS = 1.0
+
 
 async def _ensure_admin_user(app: FastAPI) -> None:
     """Startup hook: handle first boot and migrate orphan threads otherwise.
@@ -170,6 +174,18 @@ async def _migrate_orphaned_threads(store, admin_user_id: str) -> int:
     return migrated
 
 
+async def _warm_memory_retrieval(manager) -> None:
+    """Rebuild the derived retrieval index without delaying Gateway readiness."""
+    try:
+        rebuilt = await asyncio.to_thread(manager.warm_retrieval)
+        if rebuilt:
+            logger.info("Memory retrieval index rebuilt successfully")
+        else:
+            logger.warning("Memory retrieval index rebuild failed; scoped searches will retry lazily")
+    except Exception:
+        logger.warning("Memory retrieval index rebuild skipped", exc_info=True)
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     """Application lifespan handler."""
@@ -204,6 +220,26 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     except Exception:  # observability must never break startup
         logger.exception("Monocle tracing setup failed; continuing without it")
 
+    # Rebuild the derived memory retrieval index in the background. Scoped
+    # searches remain correct while this runs because DeerMem lazily rebuilds
+    # the requested scope when the full warm-up has not completed yet.
+    retrieval_warm_task: asyncio.Task[None] | None = None
+    try:
+        from deerflow.agents.memory import get_memory_manager
+
+        if startup_config.memory.enabled:
+            manager = get_memory_manager()
+            warm_retrieval = getattr(manager, "warm_retrieval", None)
+            if callable(warm_retrieval):
+                retrieval_warm_task = asyncio.create_task(
+                    _warm_memory_retrieval(manager),
+                    name="memory-retrieval-warm-up",
+                )
+        else:
+            logger.info("Memory is disabled; skipping retrieval index rebuild")
+    except Exception:
+        logger.warning("Memory retrieval index rebuild skipped", exc_info=True)
+
     # Pre-warm tiktoken encoding cache so the first memory-injection request
     # never blocks on the BPE data download (which hits an OpenAI/Azure URL
     # that may be unreachable in restricted networks — see issue #3402).
@@ -350,9 +386,26 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         #
         # K8s caveat: ``shutdown_flush_timeout_seconds`` must fit inside the
         # pod's ``terminationGracePeriodSeconds`` (channel stop + browser
-        # session close + this drain + buffer), set on the gateway Helm
-        # deployment -- or K8s SIGKILLs the drain mid-flight and the loss this
-        # is fixing is silently re-introduced.
+        # session close + the brief retrieval-warm wait + this drain + buffer),
+        # set on the gateway Helm deployment -- or K8s SIGKILLs the drain
+        # mid-flight and the loss this is fixing is silently re-introduced.
+        # The retrieval index is derived from canonical memory files, so its
+        # wait is independently capped and never consumes the flush budget.
+        retrieval_warm_finished = True
+        if retrieval_warm_task is not None and not retrieval_warm_task.done():
+            try:
+                await asyncio.wait_for(
+                    asyncio.shield(retrieval_warm_task),
+                    timeout=min(
+                        _RETRIEVAL_WARM_SHUTDOWN_TIMEOUT_SECONDS,
+                        startup_config.memory.shutdown_flush_timeout_seconds,
+                    ),
+                )
+            except TimeoutError:
+                retrieval_warm_finished = False
+                logger.warning("Memory retrieval index rebuild is still running; leaving its connection open during shutdown")
+
+        manager = None
         try:
             app_cfg = get_app_config()
             if app_cfg.memory.enabled:
@@ -370,6 +423,13 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
                     )
         except Exception:
             logger.exception("Failed to flush memory queue on shutdown")
+        finally:
+            close = getattr(manager, "close", None)
+            if callable(close) and retrieval_warm_finished:
+                try:
+                    await asyncio.to_thread(close)
+                except Exception:
+                    logger.exception("Failed to close memory backend on shutdown")
 
     logger.info("Shutting down API Gateway")
 

backend/packages/harness/deerflow/agents/memory/backends/deermem/deer_mem.py

@@ -24,6 +24,7 @@ from __future__ import annotations
 
 import copy
 import logging
+import threading
 from typing import Any, ClassVar, Literal
 
 from pydantic import PrivateAttr
@@ -132,6 +133,11 @@ class DeerMem(MemoryManager):
         # mirroring pre-abstraction `model_name: null`. Standalone (no factory) -> None.
         self._llm = self._config.host_llm if self._config.host_llm is not None else build_llm(self._config.model)
         self._updater = MemoryUpdater(self._config, self._storage, self._llm, prompts_dir=self._config.prompts_dir, callbacks=self.callbacks)
+        # Retrieval is derived data. The first search for a scope lazily
+        # rebuilds it; Gateway warm-up performs the full rebuild off-loop.
+        self._retrieval_lock = threading.RLock()
+        self._retrieval_warmed_scopes: set[tuple[str | None, str | None]] = set()
+        self._retrieval_fully_warmed = False
         # Validate the *global* explicit prompt templates at construction so a
         # misconfigured prompts_dir surfaces at startup rather than as a silent
         # dropped update. Per-agent overrides ({prompts_dir}/{agent}/*.yaml)
@@ -319,40 +325,98 @@ class DeerMem(MemoryManager):
         agent_name: str | None = None,
         category: str | None = None,
     ) -> list[dict[str, Any]]:
-        """Case-insensitive substring search over stored facts.
+        """Search through the configured retrieval adapter.
 
-        Stand-in for the planned BM25+vector+MMR retrieval
-        (``core/retrieval.py``): returns facts whose ``content`` contains the
-        query, ranked by confidence desc, capped at ``top_k``. ``category``
-        filters BEFORE the ``top_k`` slice so a category-scoped search is not
-        starved by higher-confidence facts in other categories. Sufficient for
-        the tool-driven memory mode; upgrade to semantic retrieval later
-        without changing call sites.
+        Retrieval errors never make canonical memory unavailable: the existing
+        case-insensitive substring path remains the last-resort fallback.
         """
         if not query or not query.strip() or top_k <= 0:
             return []
-        query_lower = query.strip().lower()
-        search_facts = getattr(self._storage, "search_facts", None)
         resolved_agent_name = _resolve_agent_name(agent_name)
-        scopes = [{"userId": user_id, "agentName": resolved_agent_name}]
-        indexed = (
-            search_facts(
-                query,
-                scopes=scopes,
-                top_k=top_k,
-                mode="hybrid",
-                filters={"category": category} if category else None,
+        indexed = self._fts5_search(query, top_k=top_k, user_id=user_id, agent_name=resolved_agent_name, category=category)
+        if indexed:
+            return indexed
+        return self._substring_search(query, top_k=top_k, user_id=user_id, agent_name=resolved_agent_name, category=category)
+
+    def _fts5_search(
+        self,
+        query: str,
+        *,
+        top_k: int,
+        user_id: str | None,
+        agent_name: str | None,
+        category: str | None,
+    ) -> list[dict[str, Any]]:
+        """Return adapter results in the public fact shape (compatibility helper)."""
+        agent_name = _resolve_agent_name(agent_name)
+        search_facts = getattr(self._storage, "search_facts", None)
+        scopes = [{"userId": user_id, "agentName": agent_name}]
+        try:
+            self._ensure_retrieval_scopes(scopes)
+            indexed = (
+                search_facts(
+                    query,
+                    scopes=scopes,
+                    top_k=top_k,
+                    mode="hybrid",
+                    filters={"category": category} if category else None,
+                )
+                if callable(search_facts)
+                else []
             )
-            if callable(search_facts)
-            else []
-        )
+        except Exception:
+            logger.exception("Memory retrieval adapter failed; using substring fallback")
+            indexed = []
         if indexed:
             return [_compat_document({"facts": [result.get("fact", result)]})["facts"][0] for result in indexed]
-        memory_data = _call_backend(lambda: self._updater.get_memory_data(agent_name=resolved_agent_name, user_id=user_id))
+
+        return []
+
+    def _substring_search(
+        self,
+        query: str,
+        *,
+        top_k: int,
+        user_id: str | None,
+        agent_name: str | None,
+        category: str | None,
+    ) -> list[dict[str, Any]]:
+        query_lower = query.strip().lower()
+        memory_data = _call_backend(lambda: self._updater.get_memory_data(agent_name=agent_name, user_id=user_id))
         matched = [fact for fact in memory_data.get("facts", []) if isinstance(fact.get("content"), str) and query_lower in fact["content"].lower() and (category is None or fact.get("category") == category)]
         matched.sort(key=_coerce_source_confidence, reverse=True)
         return _compat_document({"facts": matched[:top_k]})["facts"]
 
+    def _ensure_retrieval_scopes(self, scopes: list[dict[str, str | None]]) -> None:
+        """Lazily rebuild every requested scope when warm-up was skipped."""
+        if not hasattr(self, "_retrieval_lock"):
+            self._retrieval_lock = threading.RLock()
+        if not hasattr(self, "_retrieval_warmed_scopes"):
+            self._retrieval_warmed_scopes = set()
+        if not hasattr(self, "_retrieval_fully_warmed"):
+            self._retrieval_fully_warmed = False
+        rebuild = getattr(self._storage, "rebuild_index", None)
+        if not callable(rebuild):
+            return
+        with self._retrieval_lock:
+            if self._retrieval_fully_warmed:
+                return
+            status = getattr(self._storage, "retrieval_status", lambda: {"configured": True})()
+            if not status.get("configured", True):
+                self._retrieval_warmed_scopes.update((scope.get("userId"), scope.get("agentName")) for scope in scopes)
+                return
+            for scope in scopes:
+                key = (scope.get("userId"), scope.get("agentName"))
+                if key in self._retrieval_warmed_scopes:
+                    continue
+                try:
+                    result = rebuild([scope])
+                except Exception:
+                    logger.exception("Failed to lazily rebuild memory retrieval index for scope %r", key)
+                    continue
+                if result.get("supported") and not result.get("fatal"):
+                    self._retrieval_warmed_scopes.add(key)
+
     # ── Manage ───────────────────────────────────────────────────────────
     def get_memory(
         self,
@@ -408,9 +472,13 @@ class DeerMem(MemoryManager):
         """
         return self._queue.flush_sync(timeout)
 
-    # ── Tier 3 hooks (override the base defaults; warm/reload/fact CRUD) ─
+    def close(self) -> None:
+        """Close derived retrieval resources after pending updates drain."""
+        self._storage.close()
+
+    # ── Tier 3 hooks (override the base defaults; warm/reload/fact CRUD) ──
     def warm(self) -> bool:
-        """Pre-warm DeerMem-specific resources (the tiktoken encoding cache).
+        """Pre-warm DeerMem's token-counting resources.
 
         Overrides the base tier-3 hook (default None = nothing to warm). The
         Gateway lifespan calls ``manager.warm()`` directly off the event loop;
@@ -424,6 +492,29 @@ class DeerMem(MemoryManager):
             return True
         return warm_tiktoken_cache()
 
+    def warm_retrieval(self) -> bool:
+        """Rebuild the complete derived retrieval index before serving traffic."""
+        rebuild = getattr(self._storage, "rebuild_index", None)
+        if not callable(rebuild):
+            return True
+        try:
+            result = rebuild()
+            index_ok = not bool(result.get("fatal"))
+            failed = int(result.get("failed") or 0)
+            if failed and index_ok:
+                logger.warning(
+                    "Memory retrieval index rebuilt with %d fact(s) skipped",
+                    failed,
+                )
+            if index_ok:
+                with self._retrieval_lock:
+                    self._retrieval_fully_warmed = True
+                    self._retrieval_warmed_scopes.clear()
+            return index_ok
+        except Exception:
+            logger.exception("Failed to rebuild memory retrieval index during warm-up")
+            return False
+
     def reload_memory(
         self,
         *,

backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/config.py

@@ -69,8 +69,8 @@ class DeerMemConfig(BaseModel):
         description="Maximum wait for the per-scope cross-process advisory file lock.",
     )
     retrieval_adapter: str = Field(
-        default="",
-        description="Optional dotted retrieval-adapter factory. It receives DeerMemConfig and must implement RetrievalPort.",
+        default="fts5",
+        description="Retrieval adapter factory: 'fts5' (default), an empty string to disable, or a dotted factory receiving DeerMemConfig and implementing RetrievalPort.",
     )
     # ── Queue ────────────────────────────────────────────────────────────
     debounce_seconds: int = Field(

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

@@ -0,0 +1,702 @@
+"""FTS5-based retrieval engine for DeerMem.
+
+Provides BM25 full-text search over stored facts with:
+- jieba Chinese tokenization (optional, falls back to whitespace)
+- FTS5 MATCH syntax support (AND/OR/NOT/phrase/prefix) with fallback
+- Time-decay + confidence-weighted ranking
+- Category filtering
+- Scope (user_id) isolation
+
+``FTS5Retrieval`` is the low-level SQLite engine.  The storage integration is
+owned by ``FTS5RetrievalAdapter``, which implements ``storage.RetrievalPort``
+without importing storage and creating a circular dependency.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import math
+import re
+import sqlite3
+import threading
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# ── Scoring weights ──────────────────────────────────────────────────
+#
+# SQLite FTS5 ``bm25(memory_fts)`` (no positional params → defaults K1=1.2,
+# B=0.75) returns a negative value whose magnitude scales with document
+# relevance.  Critically the function takes positional params in order
+# ``(table, k1, b, *column_weights)``: the original code passed
+# ``bm25(..., 0.0, 0.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0)``, which set **k1 = 0**
+# and silently zeroed out the entire BM25 score (disabled tf saturation),
+# collapsing ranking to ``confidence * _CONFIDENCE_WEIGHT``.  Use the
+# no-arg form so SQLite defaults apply and BM25 is actually scored.
+_CONFIDENCE_WEIGHT = 0.2
+
+
+# ── jieba (optional) ──────────────────────────────────────────────────
+
+try:
+    import jieba
+
+    _jieba_available = True
+except ImportError:
+    _jieba_available = False
+
+
+def _tokenize(text: str) -> list[str]:
+    """Tokenize text: jieba for Chinese, whitespace split for English."""
+    if not text or not text.strip():
+        return []
+    if _jieba_available:
+        return [t for t in jieba.cut(text) if t.strip()]
+    return [t for t in text.split() if t.strip()]
+
+
+# ── FTS5 query preprocessing ──────────────────────────────────────────
+
+_FTS5_ADVANCED_RE = re.compile(
+    r"(\bAND\b|\bOR\b|\bNOT\b|\bNEAR\b"
+    r'|"\w.*?"'  # phrase "..."
+    r"|\w+\*"  # prefix prefix*
+    r"|\(.*?\))"  # group (...)
+)
+
+
+def _is_advanced_query(query: str) -> bool:
+    """Detect whether the query uses FTS5 advanced syntax."""
+    return bool(_FTS5_ADVANCED_RE.search(query))
+
+
+def _build_fallback_query(query: str) -> str:
+    """Convert natural-language query to FTS5 OR query (fallback strategy)."""
+    tokens = [token for token in _tokenize(query) if any(char.isalnum() for char in token)]
+    if not tokens:
+        return ""
+    # Quote each token so punctuation in natural-language input cannot become
+    # an FTS5 operator or syntax error. Double quotes inside a token are the
+    # FTS5 escape sequence for a literal quote.
+    return " OR ".join(f'"{token.replace(chr(34), chr(34) * 2)}"' for token in tokens)
+
+
+# ── Core retrieval engine ─────────────────────────────────────────────
+
+
+class FTS5Retrieval:
+    """SQLite FTS5-based retrieval engine.
+
+    Query strategy:
+      1. Advanced FTS5 syntax -> pass through to MATCH
+      2. Natural language -> jieba tokenize + OR join
+      3. Syntax error -> fall back to tokenized OR query
+      4. Still fails -> return empty
+
+    Ranking:
+      BM25 score × time_decay + confidence × 0.2
+    """
+
+    def __init__(self, db_path: str | Path = ":memory:"):
+        self._db_path = str(db_path)
+        # Gateway runs tool calls via asyncio.to_thread / ThreadPoolExecutor;
+        # SQLite connections are not safe to share across threads even when the
+        # top-level instance is single-process. We pick two defensive layers:
+        #   1. ``check_same_thread=False`` so a connection created in thread A
+        #      is accessible from thread B (libsqlite itself is reentrant under
+        #      a serialised wrapper, see #4208 hot-path discussion).
+        #   2. ``_lock`` guards all mutating sqlite calls so concurrent callers
+        #      serialise through here (prevents interleaved writes / FTS5 index
+        #      reorderings). Callers from outside instance methods MUST enter
+        #      the lock via the public API and not bypass into ``self._conn``.
+        self._conn = sqlite3.connect(self._db_path, check_same_thread=False)
+        self._lock = threading.RLock()
+        try:
+            self._conn.execute("PRAGMA journal_mode=WAL")
+            self._conn.execute("PRAGMA busy_timeout=30000")
+            self._init_schema()
+        except Exception:
+            self._conn.close()
+            raise
+
+    def _init_schema(self) -> None:
+        conn = self._conn
+        conn.execute(
+            """
+            CREATE VIRTUAL TABLE IF NOT EXISTS memory_fts USING fts5(
+                doc_id UNINDEXED,
+                content,
+                raw_content UNINDEXED,
+                category UNINDEXED,
+                scope_user UNINDEXED,
+                scope_agent UNINDEXED,
+                created_at UNINDEXED,
+                confidence UNINDEXED,
+                source UNINDEXED,
+                fact_json UNINDEXED,
+                tokenize='unicode61'
+            )
+            """
+        )
+        conn.commit()
+
+    # ── Index operations ───────────────────────────────────────────────
+
+    def _preprocess_content(self, content: str) -> str:
+        """Preprocess content for indexing: jieba tokenize for Chinese."""
+        if not content:
+            return ""
+        if _jieba_available:
+            tokens = _tokenize(content)
+            return " ".join(tokens)
+        return content
+
+    def _row_from_document(self, document: dict[str, Any]) -> tuple[Any, ...]:
+        now = datetime.now(UTC).isoformat().replace("+00:00", "Z")
+        return (
+            document["fact_id"],
+            self._preprocess_content(document["content"]),
+            document["content"],
+            document["category"],
+            document["scope_user"],
+            document["scope_agent"],
+            document.get("created_at") or now,
+            document.get("confidence", 0.5),
+            document.get("source"),
+            json.dumps(document.get("fact_data"), ensure_ascii=False, default=str),
+        )
+
+    def index_fact(
+        self,
+        fact_id: str,
+        content: str,
+        category: str = "context",
+        confidence: float = 0.5,
+        created_at: str | None = None,
+        scope_user: str | None = None,
+        scope_agent: str | None = None,
+        source: str | None = None,
+        fact_data: dict[str, Any] | None = None,
+    ) -> None:
+        """Insert or update a fact in the FTS5 index."""
+        with self._lock:
+            conn = self._conn
+            # Delete existing entry with same doc_id (INSERT OR REPLACE for FTS5)
+            conn.execute("DELETE FROM memory_fts WHERE doc_id = ?", (fact_id,))
+            conn.execute(
+                """
+                INSERT INTO memory_fts(
+                    doc_id, content, raw_content, category, scope_user, scope_agent,
+                    created_at, confidence, source, fact_json
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                self._row_from_document(
+                    {
+                        "fact_id": fact_id,
+                        "content": content,
+                        "category": category,
+                        "scope_user": scope_user or "",
+                        "scope_agent": scope_agent or "",
+                        "created_at": created_at,
+                        "confidence": confidence,
+                        "source": source,
+                        "fact_data": fact_data,
+                    }
+                ),
+            )
+            conn.commit()
+
+    def replace_documents(self, documents: list[dict[str, Any]], *, scopes: list[tuple[str, str]] | None = None) -> None:
+        """Atomically replace all or selected scope rows in one transaction."""
+        with self._lock:
+            conn = self._conn
+            try:
+                conn.execute("BEGIN IMMEDIATE")
+                if scopes is None:
+                    conn.execute("DELETE FROM memory_fts")
+                else:
+                    conn.executemany(
+                        "DELETE FROM memory_fts WHERE scope_user = ? AND scope_agent = ?",
+                        scopes,
+                    )
+                conn.executemany(
+                    """
+                    INSERT INTO memory_fts(
+                        doc_id, content, raw_content, category, scope_user, scope_agent,
+                        created_at, confidence, source, fact_json
+                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    """,
+                    [self._row_from_document(document) for document in documents],
+                )
+                conn.commit()
+            except Exception:
+                conn.rollback()
+                raise
+
+    def remove_fact(self, fact_id: str) -> None:
+        """Remove a fact from the FTS5 index."""
+        with self._lock:
+            conn = self._conn
+            conn.execute("DELETE FROM memory_fts WHERE doc_id = ?", (fact_id,))
+            conn.commit()
+
+    def clear_index(self) -> None:
+        """Clear the entire FTS5 index."""
+        with self._lock:
+            conn = self._conn
+            conn.execute("DELETE FROM memory_fts")
+            conn.commit()
+
+    def clear_scope(self, *, scope_user: str, scope_agent: str) -> None:
+        """Clear one exact adapter scope without affecting other users."""
+        with self._lock:
+            self._conn.execute(
+                "DELETE FROM memory_fts WHERE scope_user = ? AND scope_agent = ?",
+                (scope_user, scope_agent),
+            )
+            self._conn.commit()
+
+    def rebuild_from_facts(
+        self,
+        facts: list[dict[str, Any]],
+        *,
+        scope_user: str | None = None,
+        scope_agent: str | None = None,
+    ) -> None:
+        """Rebuild the entire index from a list of fact dicts."""
+        with self._lock:
+            conn = self._conn
+            try:
+                conn.execute("BEGIN")
+                conn.execute("DELETE FROM memory_fts")
+                for fact in facts:
+                    fact_id = fact.get("id", "")
+                    content = fact.get("content", "")
+                    if not fact_id or not isinstance(content, str) or not content:
+                        continue
+                    now = datetime.now(UTC).isoformat().replace("+00:00", "Z")
+                    conn.execute(
+                        """
+                        INSERT INTO memory_fts(
+                            doc_id, content, raw_content, category, scope_user, scope_agent,
+                            created_at, confidence, source, fact_json
+                        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                        """,
+                        (
+                            fact_id,
+                            self._preprocess_content(content),
+                            content,
+                            fact.get("category", "context"),
+                            scope_user or "",
+                            scope_agent or "",
+                            fact.get("createdAt") or now,
+                            fact.get("confidence", 0.5),
+                            fact.get("source"),
+                            json.dumps(fact, ensure_ascii=False, default=str),
+                        ),
+                    )
+                conn.commit()
+            except Exception:
+                conn.rollback()
+                raise
+
+    # ── Search ─────────────────────────────────────────────────────────
+
+    def search(
+        self,
+        query: str,
+        *,
+        scope_user: str | None = None,
+        scope_agent: str | None = None,
+        category: str | None = None,
+        top_k: int = 5,
+    ) -> list[dict[str, Any]]:
+        """FTS5 BM25 search with category and scope filtering.
+
+        Returns list of fact dicts (id, content, category, confidence,
+        createdAt, source, score, bm25_score) sorted by relevance.
+        """
+        import time
+
+        _t0 = time.perf_counter()
+        if not query or not query.strip() or top_k <= 0:
+            logger.debug("FTS5Retrieval.search: skipped (empty/invalid) query=%r top_k=%d", query, top_k)
+            return []
+
+        query = query.strip()
+
+        # Determine query strategy
+        if _is_advanced_query(query):
+            fts5_query = query
+            strategy = "advanced"
+        else:
+            fts5_query = _build_fallback_query(query)
+            strategy = "tokenized"
+        logger.debug(
+            "FTS5Retrieval.search: query=%r strategy=%r fts5_query=%r scope_user=%r scope_agent=%r category=%r top_k=%d",
+            query,
+            strategy,
+            fts5_query,
+            scope_user,
+            scope_agent,
+            category,
+            top_k,
+        )
+
+        # Try search
+        results = self._execute_search(fts5_query, scope_user, scope_agent, category, top_k)
+        if results is not None:
+            logger.debug(
+                "FTS5Retrieval.search: strategy=%r returned %d results in %.1fms",
+                strategy,
+                len(results),
+                (time.perf_counter() - _t0) * 1000,
+            )
+            return results
+
+        # Fallback: advanced syntax error -> tokenized OR
+        if strategy == "advanced":
+            fallback = _build_fallback_query(query)
+            if fallback and fallback != fts5_query:
+                logger.debug("FTS5Retrieval.search: advanced syntax failed, retrying with tokenized fts5_query=%r", fallback)
+                results = self._execute_search(fallback, scope_user, scope_agent, category, top_k)
+                if results is not None:
+                    logger.debug(
+                        "FTS5Retrieval.search: tokenized fallback returned %d results in %.1fms",
+                        len(results),
+                        (time.perf_counter() - _t0) * 1000,
+                    )
+                    return results
+
+        logger.debug(
+            "FTS5Retrieval.search: returning [] (no path produced results) in %.1fms",
+            (time.perf_counter() - _t0) * 1000,
+        )
+        return []
+
+    def _execute_search(
+        self,
+        fts5_query: str,
+        scope_user: str | None,
+        scope_agent: str | None,
+        category: str | None,
+        top_k: int,
+    ) -> list[dict[str, Any]] | None:
+        """Execute FTS5 query. Return None on syntax error."""
+        if not fts5_query:
+            return []
+
+        conditions = ["memory_fts MATCH ?"]
+        params: list[Any] = [fts5_query]
+
+        if scope_user:
+            conditions.append("scope_user = ?")
+            params.append(scope_user)
+        if scope_agent:
+            conditions.append("scope_agent = ?")
+            params.append(scope_agent)
+        if category:
+            conditions.append("category = ?")
+            params.append(category)
+
+        where_clause = " AND ".join(conditions)
+
+        sql = f"""
+            SELECT doc_id, content, raw_content, category, scope_user, scope_agent,
+                   created_at, confidence, source, fact_json,
+                   bm25(memory_fts) AS bm25_score
+            FROM memory_fts
+            WHERE {where_clause}
+            ORDER BY bm25_score
+            LIMIT ?
+        """
+        params.append(top_k * 2)
+
+        with self._lock:
+            try:
+                conn = self._conn
+                rows = conn.execute(sql, params).fetchall()
+            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:
+            (
+                doc_id,
+                indexed_content,
+                raw_content,
+                cat,
+                s_user,
+                s_agent,
+                created_at,
+                confidence,
+                source,
+                fact_json,
+                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:
+                    decoded = json.loads(fact_json)
+                    if isinstance(decoded, dict):
+                        fact = decoded
+                except (TypeError, ValueError):
+                    logger.debug("FTS5 fact metadata was not valid JSON for doc_id=%r", doc_id)
+            fact.setdefault("id", doc_id)
+            fact.setdefault("content", raw_content if raw_content is not None else indexed_content)
+            fact.setdefault("category", cat)
+            fact.setdefault("confidence", confidence)
+            fact.setdefault("createdAt", created_at)
+            fact.setdefault("source", source if source is not None else "fts5")
+            fact["score"] = score
+            fact["bm25_score"] = -bm25_score
+            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_query,
+            scope_user,
+            scope_agent,
+            category,
+            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]
+
+    def _compute_final_score(
+        self,
+        bm25_score: float,
+        confidence: float,
+        created_at: str,
+    ) -> float:
+        """Combined score: BM25 × time_decay + confidence weight.
+
+        ``bm25_score`` is negative for relevant docs (SQLite FTS5 convention).
+        The caller negates it (``-bm25_score``) before storing in the result
+        dict, so here we treat it as positive relevance magnitude.
+        """
+        score = bm25_score
+
+        try:
+            dt = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
+            age_days = (datetime.now(UTC) - dt).days
+            time_decay = 1.0 if age_days < 30 else math.exp(-0.01 * (age_days - 30))
+            score *= time_decay
+        except (AttributeError, ValueError, TypeError):
+            time_decay = -1.0  # sentinel for "unparseable" → skipped decay
+            age_days = -1
+
+        try:
+            normalized_confidence = float(confidence)
+            if not math.isfinite(normalized_confidence):
+                raise ValueError
+        except (TypeError, ValueError):
+            normalized_confidence = 0.5
+        score += normalized_confidence * _CONFIDENCE_WEIGHT
+
+        logger.debug(
+            "_compute_final_score: bm25_in=%.4f time_decay=%s age_days=%s conf=%.2f -> final=%.4f",
+            bm25_score,
+            time_decay,
+            age_days,
+            normalized_confidence,
+            score,
+        )
+        return score
+
+    # ── Stats ──────────────────────────────────────────────────────────
+
+    def stats(self) -> dict[str, Any]:
+        """Index statistics."""
+        with self._lock:
+            conn = self._conn
+            total = conn.execute("SELECT COUNT(*) FROM memory_fts").fetchone()[0]
+        return {
+            "total_docs": total,
+            "jieba": _jieba_available,
+            "db_path": self._db_path,
+        }
+
+    def close(self) -> None:
+        with self._lock:
+            self._conn.close()
+
+
+def _scope_value(value: str | None) -> str:
+    """Encode a typed scope value so ``None`` cannot collide with a user id."""
+    return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
+
+
+def _scope_key(scope: dict[str, str | None]) -> tuple[str, str]:
+    user_id = scope.get("userId")
+    agent_name = scope.get("agentName")
+    if user_id is not None and not isinstance(user_id, str):
+        raise ValueError("retrieval scope userId must be a string or null")
+    if agent_name is not None and not isinstance(agent_name, str):
+        raise ValueError("retrieval scope agentName must be a string or null")
+    return _scope_value(user_id), _scope_value(agent_name)
+
+
+class FTS5RetrievalAdapter:
+    """Scope-aware ``RetrievalPort`` adapter backed by one SQLite FTS5 DB.
+
+    The index is derived data. Canonical facts remain in Markdown and storage
+    notifications update only the addressed row. A deterministic composite
+    document id prevents equal fact ids in two user/agent scopes from
+    overwriting each other.
+    """
+
+    def __init__(self, db_path: str | Path = ":memory:") -> None:
+        self._engine = FTS5Retrieval(db_path)
+
+    @staticmethod
+    def _document_id(fact_id: str, scope: dict[str, str | None]) -> str:
+        scope_user, scope_agent = _scope_key(scope)
+        return json.dumps([scope_user, scope_agent, fact_id], ensure_ascii=False, separators=(",", ":"))
+
+    def _document(self, fact: dict[str, Any], scope: dict[str, str | None]) -> dict[str, Any]:
+        fact_id = fact.get("id")
+        content = fact.get("content")
+        if not isinstance(fact_id, str) or not fact_id:
+            raise ValueError("retrieval fact.id must be a non-empty string")
+        if not isinstance(content, str) or not content.strip():
+            raise ValueError("retrieval fact.content must be a non-empty string")
+        scope_user, scope_agent = _scope_key(scope)
+        payload = dict(fact)
+        payload["scope"] = {"userId": scope.get("userId"), "agentName": scope.get("agentName")}
+        source = payload.get("source")
+        return {
+            "fact_id": self._document_id(fact_id, scope),
+            "content": content,
+            "category": str(payload.get("category") or "context"),
+            "confidence": float(payload.get("confidence") or 0.5),
+            "created_at": payload.get("createdAt") if isinstance(payload.get("createdAt"), str) else None,
+            "scope_user": scope_user,
+            "scope_agent": scope_agent,
+            "source": source if isinstance(source, str) else json.dumps(source, ensure_ascii=False, default=str),
+            "fact_data": payload,
+        }
+
+    def upsert(self, fact: dict[str, Any], *, scope: dict[str, str | None], path: str) -> None:
+        del path  # Canonical location belongs to storage; the index is rebuildable.
+        document = self._document(fact, scope)
+        self._engine.index_fact(**document)
+
+    def rebuild(self, records: list[tuple[dict[str, Any], dict[str, str | None], str]], *, scopes: list[dict[str, str | None]] | None) -> None:
+        """Atomically replace the records selected by a storage rebuild."""
+        documents = [self._document(fact, scope) for fact, scope, _path in records]
+        encoded_scopes = None
+        if scopes is not None:
+            encoded_scopes = [_scope_key(scope) for scope in scopes]
+        self._engine.replace_documents(documents, scopes=encoded_scopes)
+
+    def remove(self, fact_id: str, *, scope: dict[str, str | None]) -> None:
+        self._engine.remove_fact(self._document_id(fact_id, scope))
+
+    def search(
+        self,
+        query: str,
+        *,
+        scopes: list[dict[str, str | None]],
+        top_k: int,
+        mode: str,
+        filters: dict[str, Any] | None,
+    ) -> list[dict[str, Any]]:
+        if not query.strip() or top_k <= 0:
+            return []
+        if mode not in {"hybrid", "fts5", "lexical"}:
+            raise ValueError(f"unsupported FTS5 retrieval mode: {mode}")
+
+        filters = filters or {}
+        category = filters.get("category")
+        if category is not None and not isinstance(category, str):
+            raise ValueError("retrieval category filter must be a string")
+
+        results: list[dict[str, Any]] = []
+        per_scope_limit = top_k * 4
+        for scope in scopes:
+            scope_user, scope_agent = _scope_key(scope)
+            for candidate in self._engine.search(
+                query,
+                scope_user=scope_user,
+                scope_agent=scope_agent,
+                category=category,
+                top_k=per_scope_limit,
+            ):
+                fact = dict(candidate)
+                score = float(fact.pop("score", 0.0))
+                bm25_score = float(fact.pop("bm25_score", 0.0))
+                if any(fact.get(key) != value for key, value in filters.items()):
+                    continue
+                results.append(
+                    {
+                        "fact": fact,
+                        "score": score,
+                        "matchType": "fts5",
+                        "retrieval": {"bm25": bm25_score},
+                    }
+                )
+
+        results.sort(key=lambda result: result["score"], reverse=True)
+        return results[:top_k]
+
+    def clear(self, *, scopes: list[dict[str, str | None]] | None = None) -> None:
+        if scopes is None:
+            self._engine.clear_index()
+            return
+        for scope in scopes:
+            scope_user, scope_agent = _scope_key(scope)
+            self._engine.clear_scope(scope_user=scope_user, scope_agent=scope_agent)
+
+    def stats(self) -> dict[str, Any]:
+        return self._engine.stats()
+
+    def close(self) -> None:
+        self._engine.close()
+
+
+def create_fts5_retrieval(config: Any) -> FTS5RetrievalAdapter | None:
+    """Build DeerMem's bundled adapter, using a persistent derived index.
+
+    Standalone ``DeerMem`` instances with no configured storage root use an
+    in-memory index. The host factory always injects an absolute storage root,
+    so normal Gateway instances persist the rebuildable index below it.
+    """
+    storage_path = str(getattr(config, "storage_path", "") or "")
+    if not storage_path:
+        db_path: str | Path = ":memory:"
+    else:
+        index_dir = Path(storage_path) / ".retrieval"
+        index_dir.mkdir(parents=True, exist_ok=True)
+        db_path = index_dir / "memory-fts5.sqlite3"
+    try:
+        return FTS5RetrievalAdapter(db_path)
+    except sqlite3.DatabaseError as exc:
+        if db_path == ":memory:":
+            logger.warning("SQLite FTS5 is unavailable; DeerMem will use substring retrieval: %s", exc)
+            return None
+
+        logger.warning("Derived memory retrieval index is invalid; recreating it: %s", exc)
+        try:
+            for path in (Path(db_path), Path(f"{db_path}-wal"), Path(f"{db_path}-shm")):
+                path.unlink(missing_ok=True)
+            return FTS5RetrievalAdapter(db_path)
+        except (OSError, sqlite3.DatabaseError) as retry_exc:
+            logger.warning(
+                "SQLite FTS5 index recreation failed; DeerMem will use substring retrieval: %s",
+                retry_exc,
+            )
+            return None

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

@@ -68,6 +68,10 @@ class RetrievalPort(Protocol):
 
     def search(self, query: str, *, scopes: list[dict[str, str | None]], top_k: int, mode: str, filters: dict[str, Any] | None) -> list[dict[str, Any]]: ...
 
+    def clear(self, *, scopes: list[dict[str, str | None]] | None = None) -> None: ...
+
+    def rebuild(self, records: list[tuple[dict[str, Any], dict[str, str | None], str]], *, scopes: list[dict[str, str | None]] | None) -> None: ...
+
 
 RetrievalNotification = tuple[str, dict[str, Any] | str, str | None]
 ScopedRetrievalNotifications = tuple[str, list[RetrievalNotification]]
@@ -408,6 +412,9 @@ class MemoryStorage(abc.ABC):
         """Clear global summaries and every agent fact bucket for one user."""
         raise NotImplementedError
 
+    def close(self) -> None:
+        """Release optional storage resources."""
+
 
 class FileMemoryStorage(MemoryStorage):
     def __init__(self, config: DeerMemConfig, retrieval: RetrievalPort | None = None):
@@ -416,6 +423,12 @@ class FileMemoryStorage(MemoryStorage):
         self._memory_cache: dict[tuple[str | None, str | None], tuple[dict[str, Any], tuple[Any, ...]]] = {}
         self._cache_lock = threading.Lock()
         self._scope_locks: weakref.WeakValueDictionary[tuple[str | None, str | None], threading.RLock] = weakref.WeakValueDictionary()
+        self._retrieval_dirty_scopes: set[tuple[str | None, str | None]] = set()
+
+    def close(self) -> None:
+        """Release the retrieval adapter owned by this storage instance."""
+        if self._retrieval is not None:
+            self._retrieval.close()
 
     @staticmethod
     def _cache_key(agent_name: str | None = None, *, user_id: str | None = None) -> tuple[str | None, str | None]:
@@ -463,6 +476,8 @@ class FileMemoryStorage(MemoryStorage):
                     self._retrieval.remove(str(value), scope=scope)
             except Exception:
                 logger.exception("Retrieval notification failed for %s", value)
+                with self._cache_lock:
+                    self._retrieval_dirty_scopes.add(self._cache_key(agent_name, user_id=user_id))
 
     @staticmethod
     def _validate_loaded_fact(
@@ -991,7 +1006,7 @@ class FileMemoryStorage(MemoryStorage):
             self._memory_cache[key] = (copy.deepcopy(document), signature)
         return copy.deepcopy(document)
 
-    def reload(self, agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
+    def reload(self, agent_name: str | None = None, *, user_id: str | None = None, _rebuild_retrieval: bool = True) -> dict[str, Any]:
         path = self._get_memory_file_path(agent_name, user_id=user_id)
         key = self._cache_key(agent_name, user_id=user_id)
         legacy_path = self._legacy_agent_memory_path(path, agent_name) if agent_name is not None else None
@@ -1012,6 +1027,8 @@ class FileMemoryStorage(MemoryStorage):
         signature = self._scope_signature(path, agent_name)
         with self._cache_lock:
             self._memory_cache[key] = (copy.deepcopy(document), signature)
+        if _rebuild_retrieval and agent_name is not None and self._retrieval is not None:
+            self.rebuild_index([{"userId": user_id, "agentName": agent_name}])
         return copy.deepcopy(document)
 
     def migrate(
@@ -1029,7 +1046,7 @@ class FileMemoryStorage(MemoryStorage):
             self._recover_if_needed(path)
             migrated, from_version, notifications = self._migrate_locked(path, agent_name, user_id=user_id, include_global=True)
         self._dispatch_retrieval_notifications(notifications, user_id=user_id, agent_name=agent_name)
-        document = self.reload(agent_name, user_id=user_id)
+        document = self.reload(agent_name, user_id=user_id, _rebuild_retrieval=False)
         return {
             "migrated": migrated,
             "fromVersion": from_version,
@@ -1385,7 +1402,25 @@ class FileMemoryStorage(MemoryStorage):
         filters: dict[str, Any] | None = None,
     ) -> list[dict[str, Any]]:
         if self._retrieval is not None:
+            requested_keys = {self._cache_key(self._scope_kwargs(scope).get("agent_name"), user_id=self._scope_kwargs(scope).get("user_id")) for scope in scopes}
+            with self._cache_lock:
+                dirty = requested_keys & self._retrieval_dirty_scopes
+            if dirty:
+                dirty_scopes = [{"userId": user_id, "agentName": agent_name} for user_id, agent_name in dirty]
+                rebuild_result = self.rebuild_index(dirty_scopes)
+                if rebuild_result.get("failed"):
+                    return self._search_substring(query, scopes=scopes, top_k=top_k, filters=filters)
             return self._retrieval.search(query, scopes=scopes, top_k=top_k, mode=mode, filters=filters)
+        return self._search_substring(query, scopes=scopes, top_k=top_k, filters=filters)
+
+    def _search_substring(
+        self,
+        query: str,
+        *,
+        scopes: list[dict[str, str | None]],
+        top_k: int,
+        filters: dict[str, Any] | None,
+    ) -> list[dict[str, Any]]:
         query_lower = query.strip().lower()
         if not query_lower or top_k <= 0:
             return []
@@ -1402,6 +1437,7 @@ class FileMemoryStorage(MemoryStorage):
     def rebuild_index(self, scopes: list[dict[str, str | None]] | None = None) -> dict[str, Any]:
         if self._retrieval is None:
             return {"supported": False, "indexed": 0, "failed": 0, "reason": "retrieval_not_configured"}
+        records: list[tuple[dict[str, Any], dict[str, str | None], str]] = []
         indexed = 0
         failed = 0
         if scopes is None:
@@ -1422,8 +1458,7 @@ class FileMemoryStorage(MemoryStorage):
                         raise MemoryStorageCorruption(f"Fact user scope does not match directory for {path}")
                     validate_agent_name(expected_agent)
                     self._validate_loaded_fact(fact, path, user_id=original_user, agent_name=expected_agent)
-                    self.notify_fact_upsert(fact, path=str(path))
-                    indexed += 1
+                    records.append((fact, _scope_dict(original_user, expected_agent), str(path)))
                 except Exception:
                     logger.exception("Failed to rebuild retrieval index for %s", path)
                     failed += 1
@@ -1436,11 +1471,42 @@ class FileMemoryStorage(MemoryStorage):
                     continue
                 for fact in self.list_facts(**kwargs):
                     try:
-                        self.notify_fact_upsert(fact, path=str(fact_file_path(memory_path, fact["id"], agent_name=agent_name)))
-                        indexed += 1
+                        records.append((fact, _scope_dict(kwargs.get("user_id"), agent_name), str(fact_file_path(memory_path, fact["id"], agent_name=agent_name))))
                     except Exception:
                         logger.exception("Failed to rebuild retrieval index for fact %s", fact.get("id"))
                         failed += 1
+
+        bulk_rebuild = getattr(self._retrieval, "rebuild", None)
+        if callable(bulk_rebuild):
+            try:
+                bulk_rebuild(records, scopes=scopes)
+                indexed = len(records)
+                with self._cache_lock:
+                    if scopes is None:
+                        self._retrieval_dirty_scopes.clear()
+                    else:
+                        self._retrieval_dirty_scopes.difference_update(self._cache_key(self._scope_kwargs(scope).get("agent_name"), user_id=self._scope_kwargs(scope).get("user_id")) for scope in scopes)
+            except Exception:
+                logger.exception("Failed to atomically rebuild retrieval index")
+                failed += len(records) or 1
+                return {"supported": True, "indexed": indexed, "failed": failed, "fatal": True}
+        else:
+            clear = getattr(self._retrieval, "clear", None)
+            if callable(clear):
+                clear(scopes=scopes)
+            for fact, scope, path in records:
+                try:
+                    self.notify_fact_upsert(fact, path=path)
+                    indexed += 1
+                except Exception:
+                    logger.exception("Failed to rebuild retrieval index for fact %s", fact.get("id"))
+                    failed += 1
+            if failed == 0:
+                with self._cache_lock:
+                    if scopes is None:
+                        self._retrieval_dirty_scopes.clear()
+                    else:
+                        self._retrieval_dirty_scopes.difference_update(self._cache_key(self._scope_kwargs(scope).get("agent_name"), user_id=self._scope_kwargs(scope).get("user_id")) for scope in scopes)
         return {"supported": True, "indexed": indexed, "failed": failed}
 
     def retrieval_status(self) -> dict[str, Any]:
@@ -1459,8 +1525,13 @@ class FileMemoryStorage(MemoryStorage):
 def create_storage(config: DeerMemConfig, retrieval: RetrievalPort | None = None) -> MemoryStorage:
     if retrieval is None and config.retrieval_adapter:
         try:
-            module_path, factory_name = config.retrieval_adapter.rsplit(".", 1)
-            factory = getattr(importlib.import_module(module_path), factory_name)
+            if config.retrieval_adapter == "fts5":
+                from .retrieval import create_fts5_retrieval
+
+                factory = create_fts5_retrieval
+            else:
+                module_path, factory_name = config.retrieval_adapter.rsplit(".", 1)
+                factory = getattr(importlib.import_module(module_path), factory_name)
             retrieval = factory(config)
         except Exception as exc:
             raise ValueError(f"backend_config.retrieval_adapter={config.retrieval_adapter!r} failed to load: {exc}") from exc

backend/packages/harness/deerflow/agents/memory/manager.py

@@ -481,6 +481,15 @@ class MemoryManager(BaseModel):
         ``cls(backend_config=backend_config, mode=mode)``.
         """
 
+    def close(self) -> None:
+        """Release backend resources during graceful process shutdown.
+
+        Backends that own external resources may override this hook. The
+        default is intentionally a no-op for lightweight or third-party
+        implementations.
+        """
+        return None
+
 
 # ── Backend discovery (drop-in) ───────────────────────────────────────────
 def _scan_backends() -> dict[str, type[MemoryManager]]:

backend/packages/harness/pyproject.toml

@@ -77,6 +77,8 @@ monocle = ["monocle_apptrace>=0.8.8"]
 # so the core harness install stays lean; import is lazy inside the private
 # Playwright loop. After install, run `playwright install chromium` once.
 browser = ["playwright>=1.40"]
+# Optional Chinese tokenization for the FTS5 memory retrieval adapter.
+memory-zh = ["jieba>=0.42.1"]
 
 [build-system]
 requires = ["hatchling"]

backend/pyproject.toml

@@ -30,6 +30,7 @@ redis = ["deerflow-harness[redis]"]
 discord = ["discord.py>=2.7.0"]
 monocle = ["deerflow-harness[monocle]"]
 browser = ["deerflow-harness[browser]"]
+memory-zh = ["deerflow-harness[memory-zh]"]
 
 [dependency-groups]
 dev = [

backend/tests/test_gateway_lifespan_shutdown.py

@@ -11,6 +11,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import threading
 from contextlib import asynccontextmanager
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -116,7 +117,7 @@ def test_lifespan_sweeps_upload_staging_files_on_startup():
     stop_channel_service.assert_awaited_once()
 
 
-async def _run_lifespan_with_memory_flush(*, enabled: bool, flush_return: bool) -> MagicMock:
+async def _run_lifespan_with_memory_flush(*, enabled: bool, flush_return: bool | Exception) -> MagicMock:
     """Drive lifespan with a spied memory manager.shutdown_flush.
 
     Returns the manager mock so the caller can assert the shutdown flush was
@@ -146,7 +147,10 @@ async def _run_lifespan_with_memory_flush(*, enabled: bool, flush_return: bool)
         return fake_service
 
     manager = MagicMock()
-    manager.shutdown_flush.return_value = flush_return
+    if isinstance(flush_return, Exception):
+        manager.shutdown_flush.side_effect = flush_return
+    else:
+        manager.shutdown_flush.return_value = flush_return
 
     with (
         patch("app.gateway.app.get_app_config", return_value=startup_config),
@@ -191,6 +195,13 @@ def test_lifespan_skips_memory_flush_when_disabled() -> None:
     manager.shutdown_flush.assert_not_called()
 
 
+def test_lifespan_closes_memory_manager_when_flush_raises() -> None:
+    """Derived retrieval resources are released even when queue drain fails."""
+    manager = asyncio.run(_run_lifespan_with_memory_flush(enabled=True, flush_return=RuntimeError("flush failed")))
+    manager.shutdown_flush.assert_called_once_with(5.0)
+    manager.close.assert_called_once_with()
+
+
 # ── startup warm-up log accuracy ────────────────────────────────────────────
 
 
@@ -256,3 +267,113 @@ def test_lifespan_warns_when_warm_returns_false(caplog) -> None:
     manager = asyncio.run(_run_lifespan_with_warm_return(False))
     manager.warm.assert_called_once_with()
     assert any(r.levelno == logging.WARNING and "warm-up failed" in r.message for r in caplog.records)
+
+
+async def _run_lifespan_with_slow_retrieval_warm() -> float:
+    from app.gateway.app import lifespan
+
+    app = FastAPI()
+    startup_config = SimpleNamespace(
+        log_level="INFO",
+        memory=SimpleNamespace(
+            token_counting="char",
+            enabled=True,
+            shutdown_flush_timeout_seconds=5.0,
+        ),
+    )
+    fake_service = MagicMock()
+    fake_service.get_status.return_value = {}
+    release_rebuild = threading.Event()
+    manager = MagicMock()
+    manager.warm_retrieval.side_effect = lambda: release_rebuild.wait(5.0) or True
+    manager.warm.return_value = True
+    manager.shutdown_flush.return_value = True
+
+    async def fake_start(_startup_config):
+        return fake_service
+
+    with (
+        patch("app.gateway.app.get_app_config", return_value=startup_config),
+        patch("app.gateway.app.get_gateway_config", return_value=MagicMock(host="x", port=0)),
+        patch("app.gateway.app.langgraph_runtime", _noop_langgraph_runtime),
+        patch("app.gateway.app.auth.close_oidc_service", AsyncMock()),
+        patch("app.channels.service.start_channel_service", side_effect=fake_start),
+        patch("app.channels.service.stop_channel_service", AsyncMock()),
+        patch("deerflow.agents.memory.get_memory_manager", return_value=manager),
+    ):
+        context = lifespan(app)
+        loop = asyncio.get_running_loop()
+        started_at = loop.time()
+        try:
+            await asyncio.wait_for(context.__aenter__(), timeout=1.0)
+            startup_elapsed = loop.time() - started_at
+        finally:
+            release_rebuild.set()
+        await context.__aexit__(None, None, None)
+    return startup_elapsed
+
+
+def test_lifespan_does_not_wait_for_retrieval_rebuild_before_serving() -> None:
+    assert asyncio.run(_run_lifespan_with_slow_retrieval_warm()) < 1.0
+
+
+async def _run_shutdown_with_blocked_retrieval_warm() -> tuple[float, MagicMock]:
+    from app.gateway.app import lifespan
+
+    app = FastAPI()
+    startup_config = SimpleNamespace(
+        log_level="INFO",
+        memory=SimpleNamespace(
+            token_counting="char",
+            enabled=True,
+            shutdown_flush_timeout_seconds=5.0,
+        ),
+    )
+    fake_service = MagicMock()
+    fake_service.get_status.return_value = {}
+    rebuild_started = threading.Event()
+    release_rebuild = threading.Event()
+    manager = MagicMock()
+
+    def block_rebuild() -> bool:
+        rebuild_started.set()
+        release_rebuild.wait(5.0)
+        return True
+
+    manager.warm_retrieval.side_effect = block_rebuild
+    manager.warm.return_value = True
+    manager.shutdown_flush.return_value = True
+
+    async def fake_start(_startup_config, **_kwargs):
+        return fake_service
+
+    with (
+        patch("app.gateway.app.get_app_config", return_value=startup_config),
+        patch("app.gateway.app.get_gateway_config", return_value=MagicMock(host="x", port=0)),
+        patch("app.gateway.app.langgraph_runtime", _noop_langgraph_runtime),
+        patch("app.gateway.app._RETRIEVAL_WARM_SHUTDOWN_TIMEOUT_SECONDS", 0.01),
+        patch("app.gateway.app.auth.close_oidc_service", AsyncMock()),
+        patch("app.channels.service.start_channel_service", side_effect=fake_start),
+        patch("app.channels.service.stop_channel_service", AsyncMock()),
+        patch("deerflow.agents.memory.get_memory_manager", return_value=manager),
+    ):
+        context = lifespan(app)
+        await context.__aenter__()
+        assert await asyncio.to_thread(rebuild_started.wait, 1.0)
+        loop = asyncio.get_running_loop()
+        started_at = loop.time()
+        try:
+            await context.__aexit__(None, None, None)
+        finally:
+            release_rebuild.set()
+        shutdown_elapsed = loop.time() - started_at
+
+    return shutdown_elapsed, manager
+
+
+def test_lifespan_preserves_flush_budget_when_retrieval_warm_is_still_running() -> None:
+    shutdown_elapsed, manager = asyncio.run(_run_shutdown_with_blocked_retrieval_warm())
+
+    assert shutdown_elapsed < 1.0
+    manager.shutdown_flush.assert_called_once_with(5.0)
+    manager.close.assert_not_called()

backend/tests/test_memory_retrieval_adapter.py

@@ -0,0 +1,289 @@
+"""Regression coverage for DeerMem's #4279 RetrievalPort integration."""
+
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from deerflow.agents.memory.backends.deermem.deer_mem import DeerMem
+from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemConfig
+from deerflow.agents.memory.backends.deermem.deermem.core.retrieval import (
+    FTS5Retrieval,
+    FTS5RetrievalAdapter,
+    _is_advanced_query,
+    _jieba_available,
+    create_fts5_retrieval,
+)
+from deerflow.agents.memory.backends.deermem.deermem.core.storage import FileMemoryStorage
+
+
+def _fact(fact_id: str, content: str, *, category: str = "context") -> dict:
+    return {
+        "id": fact_id,
+        "content": content,
+        "category": category,
+        "confidence": 0.8,
+        "createdAt": "2026-07-21T00:00:00Z",
+        "source": {"type": "test", "threadId": None},
+    }
+
+
+def test_natural_language_hyphens_are_not_treated_as_fts5_syntax() -> None:
+    assert not _is_advanced_query("real-time co-pilot node -js +python")
+
+
+@pytest.mark.skipif(not _jieba_available, reason="install the optional memory-zh extra")
+def test_chinese_subphrase_search_uses_jieba_tokenization(tmp_path: Path) -> None:
+    adapter = FTS5RetrievalAdapter(tmp_path / "facts.sqlite3")
+    scope = {"userId": "alice", "agentName": "agent-a"}
+    try:
+        adapter.upsert(_fact("zh", "中文检索支持验证"), scope=scope, path="")
+        results = adapter.search("检索", scopes=[scope], top_k=5, mode="fts5", filters=None)
+        assert [item["fact"]["id"] for item in results] == ["zh"]
+    finally:
+        adapter.close()
+
+
+def test_score_time_decay_and_advanced_phrase_query(tmp_path: Path) -> None:
+    engine = FTS5Retrieval(tmp_path / "facts.sqlite3")
+    try:
+        assert engine._compute_final_score(1.0, 0.5, "2026-07-21T00:00:00Z") > engine._compute_final_score(1.0, 0.5, "2020-01-01T00:00:00Z")
+        engine.index_fact("phrase", "alpha beta", scope_user='"alice"', scope_agent='"agent-a"')
+        assert engine.search('"alpha beta"', scope_user='"alice"', scope_agent='"agent-a"')
+    finally:
+        engine.close()
+
+
+def test_warm_retrieval_rebuilds_the_complete_index(tmp_path: Path) -> None:
+    manager = DeerMem(backend_config={"storage_path": str(tmp_path), "token_counting": "char"})
+    _, fact_id = manager.create_fact("warm retrieval fact", user_id="alice")
+    assert manager.warm_retrieval()
+    assert any(fact["id"] == fact_id for fact in manager.search("warm", user_id="alice"))
+
+
+def test_warm_retrieval_accepts_partial_fact_failures(tmp_path: Path) -> None:
+    manager = DeerMem(backend_config={"storage_path": str(tmp_path), "token_counting": "char"})
+    manager._storage.rebuild_index = lambda scopes=None: {"supported": True, "indexed": 2, "failed": 1}  # type: ignore[method-assign]
+
+    assert manager.warm_retrieval()
+    assert manager._retrieval_fully_warmed
+
+
+def test_warm_retrieval_retries_after_fatal_rebuild_failure(tmp_path: Path) -> None:
+    manager = DeerMem(backend_config={"storage_path": str(tmp_path), "token_counting": "char"})
+    manager._storage.rebuild_index = lambda scopes=None: {"supported": True, "indexed": 0, "failed": 1, "fatal": True}  # type: ignore[method-assign]
+
+    assert not manager.warm_retrieval()
+    assert not manager._retrieval_fully_warmed
+
+
+def test_lazy_warm_rebuilds_each_requested_scope(tmp_path: Path) -> None:
+    manager = DeerMem(backend_config={"storage_path": str(tmp_path), "token_counting": "char"})
+    calls: list[list[dict[str, str | None]]] = []
+    manager._storage.rebuild_index = lambda scopes=None: calls.append(scopes or []) or {"supported": True, "failed": 0}  # type: ignore[method-assign]
+    scopes = [{"userId": "alice", "agentName": "a"}, {"userId": "bob", "agentName": "b"}]
+    manager._ensure_retrieval_scopes(scopes)
+    assert calls == [[scopes[0]], [scopes[1]]]
+
+
+def test_lazy_warm_does_not_retry_partial_fact_failures(tmp_path: Path) -> None:
+    manager = DeerMem(backend_config={"storage_path": str(tmp_path), "token_counting": "char"})
+    rebuild = MagicMock(return_value={"supported": True, "indexed": 2, "failed": 1})
+    manager._storage.rebuild_index = rebuild  # type: ignore[method-assign]
+    scopes = [{"userId": "alice", "agentName": "a"}]
+
+    manager._ensure_retrieval_scopes(scopes)
+    manager._ensure_retrieval_scopes(scopes)
+
+    rebuild.assert_called_once_with(scopes)
+
+
+def test_deermem_close_releases_retrieval_connection(tmp_path: Path) -> None:
+    manager = DeerMem(backend_config={"storage_path": str(tmp_path), "token_counting": "char"})
+    close_storage = MagicMock()
+    manager._storage.close = close_storage  # type: ignore[attr-defined]
+
+    manager.close()
+
+    close_storage.assert_called_once_with()
+
+
+def test_file_storage_close_releases_retrieval_connection(tmp_path: Path) -> None:
+    retrieval = MagicMock()
+    storage = FileMemoryStorage(DeerMemConfig(storage_path=str(tmp_path)), retrieval=retrieval)
+
+    storage.close()
+
+    retrieval.close.assert_called_once_with()
+
+
+def test_factory_recreates_corrupt_persistent_index(tmp_path: Path) -> None:
+    index_dir = tmp_path / ".retrieval"
+    index_dir.mkdir()
+    db_path = index_dir / "memory-fts5.sqlite3"
+    db_path.write_bytes(b"not a sqlite database")
+
+    adapter = create_fts5_retrieval(DeerMemConfig(storage_path=str(tmp_path)))
+
+    assert adapter is not None
+    try:
+        adapter.upsert(
+            _fact("recovered", "recreated derived index"),
+            scope={"userId": "alice", "agentName": "agent-a"},
+            path="",
+        )
+        assert adapter.search(
+            "recreated",
+            scopes=[{"userId": "alice", "agentName": "agent-a"}],
+            top_k=5,
+            mode="fts5",
+            filters=None,
+        )
+    finally:
+        adapter.close()
+
+
+def test_adapter_isolates_scopes_even_when_fact_ids_repeat(tmp_path: Path) -> None:
+    adapter = FTS5RetrievalAdapter(tmp_path / "facts.sqlite3")
+    try:
+        adapter.upsert(_fact("same", "Alice private alpha"), scope={"userId": "alice", "agentName": "__default__"}, path="")
+        adapter.upsert(_fact("same", "Bob private beta"), scope={"userId": "bob", "agentName": "__default__"}, path="")
+
+        alice = adapter.search("alpha", scopes=[{"userId": "alice", "agentName": "__default__"}], top_k=5, mode="hybrid", filters=None)
+        bob = adapter.search("alpha", scopes=[{"userId": "bob", "agentName": "__default__"}], top_k=5, mode="hybrid", filters=None)
+
+        assert [item["fact"]["content"] for item in alice] == ["Alice private alpha"]
+        assert bob == []
+    finally:
+        adapter.close()
+
+
+def test_adapter_persists_across_restart_and_supports_category_filter(tmp_path: Path) -> None:
+    db_path = tmp_path / "facts.sqlite3"
+    scope = {"userId": "alice", "agentName": "agent-a"}
+    first = FTS5RetrievalAdapter(db_path)
+    first.upsert(_fact("one", "Python deployment preference", category="preference"), scope=scope, path="")
+    first.upsert(_fact("two", "Python deployment context", category="context"), scope=scope, path="")
+    first.close()
+
+    second = FTS5RetrievalAdapter(db_path)
+    try:
+        results = second.search("Python deployment", scopes=[scope], top_k=5, mode="hybrid", filters={"category": "preference"})
+        assert [item["fact"]["id"] for item in results] == ["one"]
+        assert results[0]["matchType"] == "fts5"
+    finally:
+        second.close()
+
+
+def test_file_storage_incremental_notifications_update_and_remove_index(tmp_path: Path) -> None:
+    adapter = FTS5RetrievalAdapter(tmp_path / "facts.sqlite3")
+    storage = FileMemoryStorage(DeerMemConfig(storage_path=str(tmp_path)), retrieval=adapter)
+    scope = {"userId": "alice", "agentName": "agent-a"}
+    try:
+        storage.upsert_fact(_fact("one", "initial searchable value"), user_id="alice", agent_name="agent-a", expected_manifest_revision=0)
+        assert storage.search_facts("initial", scopes=[scope])[0]["fact"]["id"] == "one"
+
+        updated = storage.get_fact("one", user_id="alice", agent_name="agent-a")
+        assert updated is not None
+        updated["content"] = "replacement searchable value"
+        storage.upsert_fact(
+            updated,
+            user_id="alice",
+            agent_name="agent-a",
+            expected_manifest_revision=1,
+            expected_fact_revision=updated["revision"],
+        )
+        assert storage.search_facts("replacement", scopes=[scope])[0]["fact"]["id"] == "one"
+        assert storage.search_facts("initial", scopes=[scope]) == []
+
+        storage.delete_fact("one", user_id="alice", agent_name="agent-a")
+        assert storage.search_facts("replacement", scopes=[scope]) == []
+    finally:
+        adapter.close()
+
+
+def test_full_rebuild_removes_stale_rows_after_markdown_delete(tmp_path: Path) -> None:
+    adapter = FTS5RetrievalAdapter(tmp_path / "facts.sqlite3")
+    storage = FileMemoryStorage(DeerMemConfig(storage_path=str(tmp_path)), retrieval=adapter)
+    scope = {"userId": "alice", "agentName": "agent-a"}
+    try:
+        storage.upsert_fact(_fact("one", "stale markdown value"), user_id="alice", agent_name="agent-a", expected_manifest_revision=0)
+        memory_path = storage._get_memory_file_path("agent-a", user_id="alice")
+        fact_path = next(memory_path.parent.glob("agents/agent-a/facts/**/*.md"))
+        fact_path.unlink()
+
+        result = storage.rebuild_index()
+        assert result == {"supported": True, "indexed": 0, "failed": 0}
+        assert storage.search_facts("stale", scopes=[scope]) == []
+    finally:
+        adapter.close()
+
+
+def test_reload_rebuilds_index_after_out_of_band_markdown_edit(tmp_path: Path) -> None:
+    adapter = FTS5RetrievalAdapter(tmp_path / "facts.sqlite3")
+    storage = FileMemoryStorage(DeerMemConfig(storage_path=str(tmp_path)), retrieval=adapter)
+    scope = {"userId": "alice", "agentName": "agent-a"}
+    try:
+        storage.upsert_fact(_fact("one", "original markdown value"), user_id="alice", agent_name="agent-a", expected_manifest_revision=0)
+        memory_path = storage._get_memory_file_path("agent-a", user_id="alice")
+        fact_path = next(memory_path.parent.glob("agents/agent-a/facts/**/*.md"))
+        raw = fact_path.read_text(encoding="utf-8").replace("original markdown value", "edited markdown value")
+        fact_path.write_text(raw, encoding="utf-8")
+
+        storage.reload("agent-a", user_id="alice")
+        assert storage.search_facts("edited", scopes=[scope])[0]["fact"]["id"] == "one"
+        assert storage.search_facts("original", scopes=[scope]) == []
+    finally:
+        adapter.close()
+
+
+def test_failed_notification_marks_scope_dirty_and_rebuilds_on_search(tmp_path: Path) -> None:
+    adapter = FTS5RetrievalAdapter(tmp_path / "facts.sqlite3")
+    storage = FileMemoryStorage(DeerMemConfig(storage_path=str(tmp_path)), retrieval=adapter)
+    original_upsert = adapter.upsert
+    failed_once = True
+
+    def fail_once(fact, *, scope, path):
+        nonlocal failed_once
+        if failed_once:
+            failed_once = False
+            raise RuntimeError("simulated index outage")
+        original_upsert(fact, scope=scope, path=path)
+
+    adapter.upsert = fail_once  # type: ignore[method-assign]
+    try:
+        storage.upsert_fact(_fact("one", "recoverable indexed value"), user_id="alice", agent_name="agent-a", expected_manifest_revision=0)
+        results = storage.search_facts("recoverable", scopes=[{"userId": "alice", "agentName": "agent-a"}])
+        assert results[0]["fact"]["id"] == "one"
+    finally:
+        adapter.close()
+
+
+def test_concurrent_upserts_are_searchable(tmp_path: Path) -> None:
+    adapter = FTS5RetrievalAdapter(tmp_path / "facts.sqlite3")
+    scope = {"userId": "alice", "agentName": "agent-a"}
+    try:
+
+        def write(index: int) -> None:
+            adapter.upsert(_fact(f"fact-{index}", f"concurrent memory item {index}"), scope=scope, path="")
+
+        with ThreadPoolExecutor(max_workers=8) as pool:
+            list(pool.map(write, range(20)))
+        results = adapter.search("concurrent memory", scopes=[scope], top_k=20, mode="hybrid", filters=None)
+        assert {item["fact"]["id"] for item in results} == {f"fact-{index}" for index in range(20)}
+    finally:
+        adapter.close()
+
+
+def test_deermem_create_and_restart_use_retrieval_adapter(tmp_path: Path) -> None:
+    config = {"storage_path": str(tmp_path), "token_counting": "char"}
+    manager = DeerMem(backend_config=config)
+    _, fact_id = manager.create_fact("BM25 retrieval survives restart", category="knowledge", user_id="alice")
+    assert fact_id is not None
+    assert any(fact["id"] == fact_id for fact in manager.search("retrieval BM25", user_id="alice"))
+
+    restarted = DeerMem(backend_config=config)
+    assert any(fact["id"] == fact_id for fact in restarted.search("restart retrieval", user_id="alice"))

backend/uv.lock

@@ -800,6 +800,9 @@ dependencies = [
 browser = [
     { name = "deerflow-harness", extra = ["browser"] },
 ]
+memory-zh = [
+    { name = "deerflow-harness", extra = ["memory-zh"] },
+]
 discord = [
     { name = "discord-py" },
 ]
@@ -832,6 +835,7 @@ requires-dist = [
     { name = "bcrypt", specifier = ">=4.0.0" },
     { name = "deerflow-harness", editable = "packages/harness" },
     { name = "deerflow-harness", extras = ["browser"], marker = "extra == 'browser'", editable = "packages/harness" },
+    { name = "deerflow-harness", extras = ["memory-zh"], marker = "extra == 'memory-zh'", editable = "packages/harness" },
     { name = "deerflow-harness", extras = ["monocle"], marker = "extra == 'monocle'", editable = "packages/harness" },
     { name = "deerflow-harness", extras = ["postgres"], marker = "extra == 'postgres'", editable = "packages/harness" },
     { name = "deerflow-harness", extras = ["redis"], marker = "extra == 'redis'", editable = "packages/harness" },
@@ -852,7 +856,7 @@ requires-dist = [
     { name = "uvicorn", extras = ["standard"], specifier = ">=0.34.0" },
     { name = "wecom-aibot-python-sdk", specifier = ">=0.1.6" },
 ]
-provides-extras = ["postgres", "redis", "discord", "monocle", "browser"]
+provides-extras = ["postgres", "redis", "discord", "monocle", "browser", "memory-zh"]
 
 [package.metadata.requires-dev]
 dev = [
@@ -917,6 +921,9 @@ boxlite = [
 browser = [
     { name = "playwright" },
 ]
+memory-zh = [
+    { name = "jieba" },
+]
 monocle = [
     { name = "monocle-apptrace" },
 ]
@@ -976,6 +983,7 @@ requires-dist = [
     { name = "langgraph-runtime-inmem", specifier = ">=0.28.0" },
     { name = "langgraph-sdk", specifier = ">=0.1.51" },
     { name = "markdownify", specifier = ">=1.2.2" },
+    { name = "jieba", marker = "extra == 'memory-zh'", specifier = ">=0.42.1" },
     { name = "markitdown", extras = ["all", "xlsx"], specifier = ">=0.0.1a2" },
     { name = "monocle-apptrace", marker = "extra == 'monocle'", specifier = ">=0.8.8" },
     { name = "playwright", marker = "extra == 'browser'", specifier = ">=1.40" },
@@ -992,7 +1000,7 @@ requires-dist = [
     { name = "textual", marker = "extra == 'tui'", specifier = ">=0.80" },
     { name = "tiktoken", specifier = ">=0.8.0" },
 ]
-provides-extras = ["tui", "groundroute", "ollama", "postgres", "redis", "pymupdf", "boxlite", "tenki", "monocle", "browser"]
+provides-extras = ["tui", "groundroute", "ollama", "postgres", "redis", "pymupdf", "boxlite", "tenki", "monocle", "browser", "memory-zh"]
 
 [[package]]
 name = "defusedxml"
@@ -1755,6 +1763,12 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/15/aa/0aca39a37d3c7eb941ba736ede56d689e7be91cab5d9ca846bde3999eba6/isodate-0.7.2-py3-none-any.whl", hash = "sha256:28009937d8031054830160fce6d409ed342816b543597cece116d966c6d99e15", size = 22320, upload-time = "2024-10-08T23:04:09.501Z" },
 ]
 
+[[package]]
+name = "jieba"
+version = "0.42.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/c6/cb/18eeb235f833b726522d7ebed54f2278ce28ba9438e3135ab0278d9792a2/jieba-0.42.1.tar.gz", hash = "sha256:055ca12f62674fafed09427f176506079bc135638a14e23e25be909131928db2", size = 19214172, upload-time = "2020-01-20T14:27:23.5Z" }
+
 [[package]]
 name = "jiter"
 version = "0.14.0"

config.example.yaml

@@ -1668,7 +1668,7 @@ memory:
     strict_user_scope: false    # set true in authenticated deployments after all callers propagate user_id
     manifest_filename: memory.json # user-global JSON: version/revision/time + user/history only; no facts or fact index
     file_lock_timeout_seconds: 10 # per-scope cross-process advisory lock timeout (single-machine local filesystem)
-    retrieval_adapter: ""       # optional dotted factory(config) supplied by the retrieval module
+    retrieval_adapter: fts5     # fts5 (default), empty to disable, or a dotted RetrievalPort factory(config)
     debounce_seconds: 30 # Wait time before processing queued updates
     # Backpressure cap on pending items. 0 = unlimited. At the cap, new
     # non-signal updates are rejected (QueueFull); signal updates are always