github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-10-26 13:16:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(memory): keep the retrieval index Pod-local and refresh scopes written by peers (#6494)

Browse Source 
* fix(memory): keep the retrieval index Pod-local and refresh scopes written by peers

DeerMem's derived SQLite FTS5 index for every user was one WAL database pinned
to {storage_path}/.retrieval. With several Gateway Pods sharing one home volume
each Pod opened that file over a network filesystem (unsupported by SQLite),
every Pod start emptied and refilled it under its peers, one Pod's corruption
recovery deleted files the others held open, and a Pod never learned that a
peer had written a user's facts, so memory_search served stale results.

- New backend-private key memory.backend_config.retrieval_index_path: empty
  keeps {storage_path}/.retrieval, a relative path resolves against
  storage_path, an absolute path is used as-is. The full rebuild and the
  corruption recovery act only on that configured index.
- rebuild_index records each agent scope's manifest signature
  (mtime_ns, size, revision) before reading its facts; search_facts rebuilds a
  scope whose live signature differs because another process wrote the user's
  memory. An own commit advances the recorded signatures of that user's scopes
  only when they were in sync at the previous revision, so own writes never
  rebuild while an interleaved peer write still does.
- Gateway startup warns, in the style of _validate_agent_storage, when a
  declared multi-instance deployment keeps the DeerMem index inside
  storage_path.
- config.example.yaml documents the key; config_version 55 -> 56, mirrored in
  the Helm chart and its README.
- Helm mounts an emptyDir at /var/lib/deerflow/memory-index, points the key at
  it, drops the legacy memory.storage_path: memory.json line, and documents the
  memory layout next to the other multi-replica prerequisites.

* docs(changelog): link the memory retrieval index entry to #6494

* fix(memory): never advance the index signature on a no-op commit and resync whole scopes

Two review findings on the cross-replica retrieval re-sync:

- save() and apply_changes() inferred the replaced manifest revision as
  "returned revision - 1". The commit helper returns the existing manifest
  unchanged for a no-op (unchanged-value or empty change set), so after a peer
  bumped the manifest an own no-op write advanced this process's stale index
  signature to the peer's revision without carrying the peer's delta: a fact the
  peer deleted stayed searchable until the manifest changed again. Both paths
  now read the real pre-commit revision under the user lock and pass
  committed=(pre_revision, signature) only when the commit produced a new
  revision; a no-op passes None so the next search still rebuilds the scope.

- The scoped rebuild (signature mismatch and the pre-existing dirty-scope path)
  listed facts through list_facts() with its default limit of 100 and then
  replaced every row of the scope, so a scope with more than 100 facts lost its
  tail on the first re-sync. It now reads the scope's complete fact list.

* fix(memory): fence index signature promotion against a concurrent scope rebuild

Review finding on the cross-replica re-sync: a successful adapter notification
does not prove its delta is still in the index when the signature is promoted.
The SQLite lock covers single adapter calls only, and neither the scope lock
nor DeerMem's warm-up lock spans "notify, then promote", so rebuild_index can
replace a scope's rows in between. After a peer summary-only commit (r+1), A's
search starts a scoped refresh that captures the r+1 signature and reads f1/f2;
A's own deletion of f2 commits r+2 and removes the row; the refresh then
installs its stale f1/f2 and marks the scope synced at r+1; the deletion
resumes and promoted that signature to r+2, so later searches saw the live
signature as equal and kept returning the resurrected f2 until another manifest
change.

Mechanism: a generation fence rather than a lock. rebuild_index bumps a
per-scope generation (scoped rebuild) or a storage-wide generation (full
rebuild) under _cache_lock whenever it publishes or forgets a scope's rows, in
both mark_synced and forget_synced. _dispatch_retrieval_notifications snapshots
the generations under _cache_lock before its first adapter call and, at
promotion time, skips every scope whose generation moved (and everything when
the storage-wide generation moved), leaving the rebuild's own captured
signature in place; that signature predates the commit, so the next search's
comparison rebuilds the scope. A lock would have had to serialize notification
plus promotion with bulk install plus publication while search_facts ->
rebuild_index -> load() already takes the scope, file and cache locks; the
fence adds no lock-order obligations and only costs a redundant rebuild in the
rare race. Every interleaving is covered: a rebuild that completes before the
snapshot is reflected in the captured generation and promotion proceeds on an
index that carries both the rebuild's rows and this write's delta; one that
publishes between snapshot and promotion is skipped; one that publishes after
the promotion records its stale captured signature, which the next search
catches.

Tests reproduce both orders deterministically with threading.Event ordering
(adapter wrapper that holds the bulk install and the remove notification):
test_promotion_is_fenced_against_a_refresh_that_installs_in_between (red before
the fix) and
test_stale_refresh_installing_after_a_promotion_is_caught_by_the_next_search
(regression guard for the mirrored order).

* fix(memory): publish a rebuild's signature under the lock that replaced its rows

Review finding on the cross-replica re-sync: two peer-triggered refreshes of
one scope could replace rows in one order and publish their captured
signatures in the other. On a warmed Pod A, refresh R1 captures r+1 with f1/f2
and waits before installing; a peer deletes f2 at r+2; refresh R2 captures
r+2, installs f1 only and pauses after bulk_rebuild returns, before
mark_synced; R1 installs its stale f1/f2 and publishes r+1; R2 then publishes
r+2 over R1's rows. Later searches saw the live signature as equal and kept
returning the deleted fact. The generation fence only guards own-write
promotion, and the adapter's SQLite lock ends when the install returns.

Mechanism: a per-storage _retrieval_publish_lock held across the row
replacement and the signature publication, for the bulk path (bulk_rebuild +
mark_synced / forget_synced) and the clear+upsert fallback, scoped and full
alike. Overlapping refreshes of a scope therefore serialize their
install+publish pairs: whichever installs last also publishes last, and a stale
captured signature published last differs from the live manifest, so the next
search's compare rebuilds the scope. Fact reading (load() / the glob) stays
outside the lock so a slow read never blocks refreshes of other scopes. The
startup full rebuild holds the lock for its single bulk install; scoped
refreshes wait for it, which is acceptable because before warm-up completes a
search would be rebuilding its scope anyway (DeerMem.warm_retrieval takes its
own _retrieval_lock only after rebuild() returns, so the two never nest).

Lock order: the publish lock is acquired with no storage lock held --
search_facts releases _cache_lock before calling rebuild_index, reload() leaves
its scope/file-lock block before rebuilding, and DeerMem._ensure_retrieval_scopes
holds only its outer RLock, which nothing inside rebuild_index acquires. Inside
the lock only the adapter's own lock and _cache_lock are taken, and both are
leaf locks that never call back into storage, so no cycle is possible.

Test: test_rebuild_publication_stays_ordered_with_row_replacement drives the
reviewer's interleaving with threading.Event ordering (red before the fix). The
pause harness now holds a refresh after its fact read, outside the publish
lock, instead of inside the adapter's rebuild, so the earlier fence tests keep
their semantics without holding the new lock while paused.

* fix(memory): read each manifest once per full rebuild and serve the local index on a compare read error

Two suggestion-level review items on the cross-replica retrieval re-sync.

- The full rebuild cached manifest signatures keyed by (manifest path, agent
  name), but _scope_signature reads only the manifest file -- every agent
  bucket of a user shares one memory.json -- so the startup scan stat+read+
  parsed the same file once per bucket that has facts. The cache is now keyed
  by the manifest path alone while synced signatures are still recorded per
  (user, agent) scope: users manifest reads instead of users x agents.

- The per-search freshness compare put a shared-volume manifest read on every
  search_facts call, and _load_memory_file raises MemoryStorageCorruption on
  OSError and malformed JSON, so a transient network-filesystem read error or a
  corrupt memory.json failed a search the Pod-local index could still answer.
  The compare now catches OSError, ValueError and MemoryStorageCorruption: it
  logs a warning naming the scope (traceback at debug), serves the current
  local index for that scope and leaves the recorded synced signature
  untouched so the next search compares again. It deliberately does not mark
  the scope dirty: a scoped rebuild would hit the same failing read, forget the
  scope's signature and fail the search path too.

* docs(changelog): resolve merge markers left in the Unreleased section

* fix(memory): forget a scope's index signature when an own delta cannot be proved compatible

Review finding on the cross-replica re-sync: skipping promotion left a newer
rebuild's signature trusted even after this dispatcher had overwritten its rows
with an older notification. On a warmed Pod A, an own PATCH of f2 commits r+1
and pauses after releasing the user lock, before its adapter upsert; Pod B
deletes f2 at r+2; a concurrent search on A detects the peer change, rebuilds
the scope without f2 and publishes r+2; the pending r+1 upsert then reinserts
the deleted fact, and the revision check merely skipped because the synced
revision was r+2 rather than the PATCH's previous r. The signature stayed equal
to the live manifest, so searches returned the deleted fact indefinitely. The
generation fence could not help: it only suppresses promotion, the stale
adapter mutation still happens.

Rule: after an own commit's notifications are applied, the mutated scope's
recorded signature may stay only when the delta is provably compatible with
the published snapshot -- the scope is synced at exactly previous_revision and
no rebuild published rows for it (or storage-wide) since the fence snapshot --
in which case it is promoted as before. In every other case where this
dispatcher mutated the adapter for the scope (synced at another revision,
newer or older; generation moved; not synced at all; or notifications without
a committed revision, i.e. migration and clear paths) the scope's signature is
forgotten so the next search's compare rebuilds it. The user's other scopes
were not mutated by this delta: they are promoted when provable and otherwise
left untouched. The decision runs under _cache_lock together with the
generation reads, so a rebuild publishing concurrently is either observed or
observes the forget; the failure path (dirty + forget) is unchanged.

Orders covered: the reviewer's (publish r+2, then the stale upsert) forgets and
the next search rebuilds without f2; the earlier fence order (publish between
adapter op and promotion) now forgets too, one redundant rebuild; no concurrent
rebuild promotes exactly as before.

Test: test_stale_own_delta_replayed_over_a_newer_snapshot_forgets_the_scope
holds the PATCH's adapter upsert after the user lock is released (new
pause_next_upsert hook on the test adapter), Event-ordered, red before the fix.

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
Nan Gao 2026-10-08 21:20:45 +08:00 committed by GitHub
parent fb0ed9c960
commit 5ae953fc2f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
18 changed files with 1181 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
CHANGELOG.md
View File

@ -804,6 +804,26 @@ This release closes that milestone with **439 merged pull requests**.
### Fixed
- **memory:** DeerMem's derived SQLite FTS5 retrieval index can now live
outside the memory root, and a Gateway instance now notices facts another
instance wrote. The index for every user was one SQLite database in WAL mode
at `{storage_path}/.retrieval`, so several Gateway Pods sharing one home
volume opened the same WAL file over a network filesystem, which SQLite does
not support; every Pod start emptied and refilled the shared index under its
peers, one Pod's corruption recovery deleted files the others held open, and
a Pod kept serving its own copy of a user's facts after a peer wrote new ones.
The new `memory.backend_config.retrieval_index_path` places the index
directory elsewhere (empty keeps today's location; a relative path is
resolved against `storage_path`), the startup rebuild and corruption recovery
touch only that local index, and a search now re-syncs a scope whose
`memory.json` revision changed since this process last indexed it, so a
peer's facts appear on the next search while this instance's own writes do
not trigger a rebuild. A declared multi-instance deployment
(`deployment.multi_instance: true` / `DEER_FLOW_MULTI_INSTANCE=1`) that keeps
the index inside `storage_path` logs a startup warning. The Helm chart mounts
a Pod-local `emptyDir` at `/var/lib/deerflow/memory-index`, points the key at
it, and drops the legacy `memory.storage_path: memory.json` line that the
Gateway discarded with a warning at every start. `config_version` is now 56. ([#6494])
- **skills:** A `/skill-name` activation now survives a retried model call. The
activation was marked as done before the model was called, so when the call
failed (rate limit, overload, timeout) or came back empty and was retried, the
@ -9237,5 +9257,6 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
[#6450]: https://github.com/bytedance/deer-flow/pull/6450
[#6481]: https://github.com/bytedance/deer-flow/pull/6481
[#6484]: https://github.com/bytedance/deer-flow/pull/6484
[#6494]: https://github.com/bytedance/deer-flow/pull/6494
[#6495]: https://github.com/bytedance/deer-flow/pull/6495
[#6506]: https://github.com/bytedance/deer-flow/pull/6506

2
README.md
View File

@ -2342,7 +2342,7 @@ In the default DeerMem `middleware` mode, automatic extraction now classifies ev
When a fact scope reaches `max_facts`, DeerMem still uses the historical confidence-only eviction order by default. Operators can opt in to `memory.backend_config.fact_eviction_policy: hybrid-v1`, which combines bounded confidence (65%), explicit-confirmation freshness (25%), and query-driven access heat (10%). Hybrid signal metadata is collected only while hybrid-v1 or shadow mode is enabled. Explicit confirmation is returned as `factsToReinforce` by the existing memory-update LLM call and is accepted only when deterministic message processing also detects a user reinforcement signal; it also resets the fact's staleness-review clock. This deterministic gate is batch-level: it establishes only that a human message among the last six filtered messages in the current extraction batch matched a reinforcement pattern. The LLM-selected `factsToReinforce` ID supplies the fact binding; DeerMem does not independently verify a signal-to-fact correspondence. Repeated extraction or automatic injection never confirms a fact. Custom `memory_update` prompts should add the optional `factsToReinforce` array to participate in confirmation freshness. Access heat is stored in a separate decaying sidecar and increases only when `memory_search` actually returns the fact, so reads do not rewrite canonical Markdown or its `updatedAt`. Hybrid mode also reserves a bounded minimum of correction slots (10% of the cap, at most 10; unused slots return to normal competition). Capacity deletion remains physical, but a bounded metadata-only audit records fact IDs, categories, policy scores, and reasons without copying fact content. `fact_eviction_shadow_enabled: true` evaluates hybrid-v1 alongside the default policy without changing actual retention. This feature adds no LLM invocation and can be rolled back by selecting `confidence`.
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.
File-backed memory now separates global user context from agent facts. Each user has one `memory.json` containing only the project-independent `user` and `history` summaries; every fact is a canonical Markdown file below `agents/{agent_name}/facts/`. Existing lead-agent middleware, API, Settings, import/export, and embedded-client calls that omit `agent_name` resolve inside DeerMem to the reserved `__default__` bucket. That bucket is outside the valid custom-agent name grammar, so a real custom agent named `lead-agent` has a separate fact repository and deleting a custom agent cannot delete a memory-only directory without `config.yaml`. Public agent identifiers are case-insensitive and canonicalized to lowercase. Runtime/API readers still receive a compatibility `facts` array for the selected/default agent, so the frontend does not read agent facts from `memory.json`; structured Markdown `source` metadata is projected to the historical string field at the MemoryManager boundary. An unscoped Clear All first migrates facts from unread legacy per-agent JSON without adopting its soon-to-be-cleared summaries, then removes shared summaries and facts from every agent bucket while preserving agent configuration files, so a later read cannot resurrect skipped legacy facts; an explicitly agent-scoped clear removes only that agent's facts. On first normal read, old facts embedded in the user JSON are migrated automatically to `__default__`; facts written to the earlier implicit `lead-agent` bucket are also moved when that directory is not a real custom agent. Migration and normal writes notify the configured retrieval adapter only after durable storage locks are released. DeerMem uses a scope-aware SQLite FTS5/BM25 adapter by default, stores only rebuildable derived index data under `.retrieval/` (or under `memory.backend_config.retrieval_index_path` when set; a relative value resolves against `storage_path`), and rebuilds it in the background during Gateway startup or lazily on the first scoped search. A corrupt derived index is recreated automatically. When several Gateway instances share one `storage_path`, point `retrieval_index_path` at instance-local disk: SQLite's WAL mode is unsupported on network filesystems, each instance re-syncs a user's index on its next search after a peer writes, and a declared multi-instance deployment (`deployment.multi_instance: true` / `DEER_FLOW_MULTI_INSTANCE=1`) that leaves the index under the memory root logs a warning at startup. The Helm chart mounts a Pod-local `emptyDir` for it. Set `memory.backend_config.retrieval_adapter` to an empty string to disable it and use the local substring fallback. Chinese tokenization is optional; install the backend `memory-zh` extra (`uv sync --extra memory-zh`) for jieba-assisted sub-phrase search. Journaled writes, a shared user lock, and optimistic user-memory revisions prevent silent lost updates.
Set `memory.backend_config.retrieval_relevance_enabled: true` to opt into deterministic relevance/confidence ranking and query-aware injection. This **bypasses `retrieval_adapter` for search**, including the default FTS5/BM25 and custom adapters; indexing remains configured. `retrieval_relevance_weight` controls the blend and `retrieval_diversity_weight` enables near-duplicate penalties (default 0). Scoring uses at most the first 4096 characters and 128 tokens per query/fact. Search diversifies only up to `top_k`; injection diversifies guaranteed and regular facts independently until their token budgets are reached. Leave the feature disabled to retain the existing retrieval and injection behavior.

46
backend/app/gateway/deps.py
View File

@ -22,6 +22,7 @@ import logging
import os
from collections.abc import AsyncGenerator, Callable
from contextlib import AsyncExitStack, asynccontextmanager
from pathlib import Path
from typing import TYPE_CHECKING, Any, TypeVar, cast
from fastapi import FastAPI, HTTPException, Request
@ -258,6 +259,48 @@ def _validate_agent_storage(config: AppConfig) -> None:
)
def _validate_memory_retrieval_index(config: AppConfig) -> None:
"""Warn when a declared multi-instance deployment keeps DeerMem's retrieval index on the shared memory volume.
DeerMem's derived FTS5 index is one SQLite database in WAL mode. When
``storage_path`` sits on the home volume several Gateway instances share,
the default ``{storage_path}/.retrieval`` makes every instance open that
same file over a network filesystem (where SQLite documents WAL as
unsupported), empty and refill it under its peers at startup, and delete
it from under them on corruption recovery. The index is rebuildable, so
each instance should keep its own copy on local disk through
``memory.backend_config.retrieval_index_path``. Only the explicit
declaration counts: uvicorn workers of one process tree share local disk,
where a shared WAL index is supported. Mirrors ``_validate_agent_storage``:
a warning, not a refusal, because memory still works, only slower.
"""
declaration = multi_instance_declaration(config)
if declaration is None:
return
memory = getattr(config, "memory", None)
if memory is None or not getattr(memory, "enabled", False) or getattr(memory, "manager_class", "deermem") != "deermem":
return
backend_config = dict(getattr(memory, "backend_config", None) or {})
if backend_config.get("retrieval_adapter", "fts5") != "fts5":
return # disabled, or a custom RetrievalPort factory that owns its own storage
from deerflow.agents.memory.backends.deermem.deermem.core.paths import retrieval_index_directory
from deerflow.agents.memory.manager import resolve_deermem_storage_path
storage_path = resolve_deermem_storage_path(backend_config)
index_dir = retrieval_index_directory(storage_path, backend_config.get("retrieval_index_path"))
if index_dir is None or not Path(index_dir).resolve().is_relative_to(Path(storage_path).resolve()):
return
logger.warning(
"%s but the DeerMem retrieval index at %s is inside memory storage_path %s: every Gateway instance opens the same "
"SQLite WAL index over the shared memory volume, rebuilds it under its peers at startup and deletes it from under them "
"on corruption recovery. Set memory.backend_config.retrieval_index_path to an instance-local directory (the Helm chart "
"mounts an emptyDir at /var/lib/deerflow/memory-index).",
declaration.knob,
index_dir,
storage_path,
)
async def _drain_inflight_runs(run_manager: RunManager) -> None:
"""Drain in-flight runs before the checkpointer is torn down (issue #3373).
@ -526,6 +569,9 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen
# Reject agent_storage.backend='db' on a non-durable database, and warn on
# node-divergent file storage under multi-worker Postgres.
_validate_agent_storage(startup_config)
# Warn when a declared multi-instance deployment shares DeerMem's SQLite
# retrieval index across instances through the memory volume.
_validate_memory_retrieval_index(startup_config)
async with AsyncExitStack() as stack:
# Lifecycle and system-model hooks can originate on isolated subagent

1
backend/docs/CONFIGURATION.md
View File

@ -585,6 +585,7 @@ Notes:
- `DEER_FLOW_MULTI_INSTANCE` treats blank, `0`, `false`, `no` and `off` as "not declared"; any other value declares a multi-instance deployment, so a typo fails closed.
- `GET /health/ready` pings the Redis stream bridge on every probe and answers 503 (`stream_bridge: unreachable`) while Redis is down, so the orchestrator drains that instance instead of routing it runs it cannot publish or stream; the memory bridge reports `not_configured`. When `sandbox.provisioner_url` is set, the body also carries the provisioner's `/health` verdict (`provisioner: ok|unreachable`), which never changes the status code because every instance shares one provisioner. That probe follows the sandbox clients' proxy policy: loopback, private, link-local and cluster-local provisioner addresses bypass `HTTP_PROXY`, external hosts keep the environment's proxy settings.
- The declaration also drives the `agent_storage.backend: file` divergence warning, the inbound webhook dedupe warning, and the WeChat QR-login guard, which otherwise only look at the worker count.
- With the default DeerMem backend, the declaration also warns when the derived SQLite retrieval index sits inside the shared `storage_path` (the default `{storage_path}/.retrieval`). Set `memory.backend_config.retrieval_index_path` to an instance-local directory (relative values resolve against `storage_path`): SQLite WAL is unsupported on network filesystems, and the index is rebuilt from the Markdown facts at startup and re-synced per user scope after a peer writes.
- Restart-required: the gate runs once at startup. Restart all Gateway instances together after changing it.
### Agent Storage

27
backend/packages/harness/deerflow/agents/memory/AGENTS.md
View File

@ -201,16 +201,39 @@ to `--user-id` (default `default`); `threads/` go to their `threads_meta` owner.
DeerMem selects persistent SQLite FTS5 by default.
An empty value selects the substring fallback.
SQLite index data lives below `.retrieval/` and remains rebuildable.
The SQLite index is rebuildable derived data below `retrieval_index_path`
(empty = `{storage_path}/.retrieval`; relative resolves against `storage_path`;
`paths.retrieval_index_directory` is the one resolver). Instances sharing
`storage_path` keep it instance-local: SQLite WAL is unsupported on network
filesystems, and the full rebuild and one-shot corruption recovery touch only
that local index. `deps._validate_memory_retrieval_index` warns when a declared
multi-instance deployment leaves it inside `storage_path`.
Chinese tokenization uses `jieba` only with the `memory-zh` extra.
Malformed facts are logged and skipped during rebuild.
A fatal rebuild failure keeps lazy retry active.
A corrupt persistent database is deleted and recreated once.
Storage sends adapter updates after it releases durable locks.
Adapter failures mark the scope dirty.
Search then uses canonical substring matching until rebuild succeeds.
Cross-process freshness: `rebuild_index` records each agent scope's manifest
signature `(mtime_ns, size, revision)` before reading that scope's complete
fact list (never a `list_facts` page); `search_facts` rebuilds a scope whose
live signature differs (a peer wrote the user's memory), and a manifest read
failure during that compare logs and serves the local index. A commit that produced
a new revision advances the recorded signatures of that user's scopes that were
in sync at the pre-commit revision read under the user lock; a no-op commit
advances nothing. Own writes therefore never rebuild while an interleaved peer
write still does. Promotion is generation-fenced: `rebuild_index` bumps a
per-scope (full rebuild: storage-wide) generation under `_cache_lock` when it
publishes or forgets rows, the dispatcher snapshots them before its first
adapter call, and a scope whose generation moved keeps the rebuild's own
signature for the next search to compare; an own delta that cannot be proved
compatible with the published snapshot forgets the mutated scope's signature
instead of merely skipping promotion. A rebuild's row replacement and its
publication run as one unit under `_retrieval_publish_lock` (fact reads stay
outside), so overlapping refreshes of a scope publish in install order.
Gateway startup schedules `DeerMem.warm_retrieval()` without delaying readiness.
The first search can rebuild its exact scope.
Shutdown waits one second for retrieval warm-up.

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

@ -84,6 +84,15 @@ class DeerMemConfig(BaseModel):
"Search bypasses this adapter when retrieval_relevance_enabled is true; indexing remains configured."
),
)
retrieval_index_path: str = Field(
default="",
description=(
"Directory holding the derived SQLite FTS5 retrieval index. Empty (default) = ``{storage_path}/.retrieval``; "
"a relative path is resolved against storage_path; an absolute path is used as-is. The index is rebuildable, "
"so Gateway instances that share storage_path must each point this at instance-local disk: SQLite WAL is "
"unsupported on network filesystems, and peers would rebuild or delete each other's index."
),
)
fact_dedup_enabled: bool = Field(
default=False,
description=(

25
backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/paths.py
View File

@ -69,6 +69,31 @@ def _default_root() -> Path:
return Path.home() / ".deermem"
RETRIEVAL_INDEX_DIRNAME = ".retrieval"
def retrieval_index_directory(storage_path: str | None, retrieval_index_path: str | None = None) -> Path | None:
"""Return the directory of DeerMem's derived SQLite retrieval index.
``retrieval_index_path`` wins when set: an absolute path is used as-is and a
relative one is resolved against the data root (``storage_path``, or the
default root when that is empty). Otherwise the index lives below
``{storage_path}/.retrieval``, and a DeerMem without a storage root keeps it
in memory (``None``). The index is rebuildable derived data, so instances
that share ``storage_path`` keep it instance-local through this setting.
"""
configured = str(retrieval_index_path or "")
root = Path(storage_path) if storage_path else None
if configured:
path = Path(configured)
if path.is_absolute():
return path
return (root if root is not None else _default_root()) / path
if root is None:
return None
return root / RETRIEVAL_INDEX_DIRNAME
def memory_file_path(
config: DeerMemConfig,
agent_name: str | None = None,

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

@ -24,6 +24,8 @@ from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from .paths import retrieval_index_directory
logger = logging.getLogger(__name__)
# ── Scoring weights ──────────────────────────────────────────────────
@ -675,13 +677,14 @@ def create_fts5_retrieval(config: Any) -> FTS5RetrievalAdapter | None:
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.
so normal Gateway instances persist the rebuildable index below it, or at
``retrieval_index_path`` when the operator keeps it off a shared volume.
Corruption recovery below deletes only that index's own files.
"""
storage_path = str(getattr(config, "storage_path", "") or "")
if not storage_path:
index_dir = retrieval_index_directory(getattr(config, "storage_path", ""), getattr(config, "retrieval_index_path", ""))
if index_dir is None:
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:

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

@ -581,6 +581,22 @@ class FileMemoryStorage(MemoryStorage):
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()
# Manifest signature at which this process last brought each agent scope's
# index in sync. A different live signature means another process wrote
# the user's memory, so the scope is rebuilt before it is searched.
self._retrieval_synced_signatures: dict[tuple[str | None, str | None], tuple[Any, ...]] = {}
# Generations bumped (under _cache_lock) whenever rebuild_index publishes
# or forgets a scope's rows: per scope for a scoped rebuild, storage-wide
# for a full one. Signature promotion after an own commit is fenced on
# them, because a rebuild that installed rows in between may have read
# its facts before that commit and has recorded its own signature.
self._retrieval_index_generation = 0
self._retrieval_scope_generations: dict[tuple[str | None, str | None], int] = {}
# Serializes a rebuild's row replacement with its signature publication
# (see rebuild_index): the adapter lock ends when the install returns, so
# two overlapping refreshes of one scope could otherwise install in one
# order and publish in the other.
self._retrieval_publish_lock = threading.Lock()
def close(self) -> None:
"""Release the retrieval adapter owned by this storage instance."""
@ -848,11 +864,42 @@ class FileMemoryStorage(MemoryStorage):
*,
user_id: str | None,
agent_name: str | None,
committed: tuple[int, tuple[Any, ...]] | None = None,
) -> None:
"""Notify the optional index only after durable storage locks are released."""
"""Notify the optional index only after durable storage locks are released.
``committed`` is ``(previous_revision, signature)`` of a manifest this
process just replaced, with ``previous_revision`` read under the user lock
right before the commit; a commit that changed nothing passes ``None``.
Once every notification lands, the user's scopes whose index was in sync
at ``previous_revision`` are in sync at ``signature`` too: the write was
the only change in between and the index carries its delta, so the next
search does not rebuild them. A scope synced at an older revision keeps
that older signature and is rebuilt, because a peer wrote the manifest
in between; a no-op commit leaves every recorded signature as it was.
A scope is promoted only when the delta is provably compatible with the
published snapshot: it is synced at exactly ``previous_revision`` and no
rebuild published rows for it (or storage-wide) since the generation
snapshot taken before the first adapter call. The SQLite lock covers
single adapter calls only, so ``rebuild_index`` may replace the scope's
rows at any point around the notifications below, and a rebuild may
have read its facts before this commit (resurrecting a fact this write
deleted) or after it (so that an older delta replayed on top reinserts
a fact a peer has since deleted). The scope this delta mutated therefore
forgets its recorded signature whenever compatibility cannot be proved,
so the next search's comparison rebuilds it; the user's other scopes
were not mutated here and are promoted when provable, else left alone.
"""
if self._retrieval is None:
return
scope = _scope_dict(user_id, agent_name)
key = self._cache_key(agent_name, user_id=user_id)
fence: tuple[int, dict[tuple[str | None, str | None], int]] | None = None
if committed is not None:
with self._cache_lock:
fence = (self._retrieval_index_generation, dict(self._retrieval_scope_generations))
failed = False
for action, value, fact_path in notifications:
try:
if action == "upsert":
@ -860,9 +907,35 @@ class FileMemoryStorage(MemoryStorage):
else:
self._retrieval.remove(str(value), scope=scope)
except Exception:
failed = True
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))
# Decided under _cache_lock together with the generation reads, so a
# rebuild publishing concurrently is either seen here or sees the result.
with self._cache_lock:
if failed:
self._retrieval_dirty_scopes.add(key)
self._retrieval_synced_signatures.pop(key, None)
return
publication_moved = fence is None or fence[0] != self._retrieval_index_generation
def provably_compatible(synced_key: tuple[str | None, str | None]) -> bool:
if committed is None or fence is None or publication_moved:
return False
synced = self._retrieval_synced_signatures.get(synced_key)
if synced is None or (synced[2] or 0) != committed[0]:
return False
return self._retrieval_scope_generations.get(synced_key, 0) == fence[1].get(synced_key, 0)
if notifications and not provably_compatible(key):
# Rows of this scope were mutated by a delta that may be older than
# the published snapshot: drop the signature so the next search rebuilds.
self._retrieval_synced_signatures.pop(key, None)
if committed is None or publication_moved:
return
signature = committed[1]
for synced_key in [synced_key for synced_key in self._retrieval_synced_signatures if synced_key[0] == user_id]:
if provably_compatible(synced_key):
self._retrieval_synced_signatures[synced_key] = signature
@staticmethod
def _validate_loaded_fact(
@ -1475,6 +1548,7 @@ class FileMemoryStorage(MemoryStorage):
lock_path = path.parent / ".memory.lock"
notifications: list[RetrievalNotification] = []
deleted_metadata_ids: list[str] = []
committed: tuple[int, tuple[Any, ...]] | None = None
try:
if not isinstance(memory_data, dict):
raise ValueError("memory_data must be an object")
@ -1497,7 +1571,8 @@ class FileMemoryStorage(MemoryStorage):
summaries = None
if agent_name is None:
summaries = {"user": memory_data.get("user", {}), "history": memory_data.get("history", {})}
_, notifications = self._commit_changes_locked(
pre_revision = int((self._load_memory_file(path) or {}).get("revision") or 0)
memory_file, notifications = self._commit_changes_locked(
path,
user_id=user_id,
agent_name=agent_name,
@ -1508,6 +1583,10 @@ class FileMemoryStorage(MemoryStorage):
)
document = self._read_document(path, agent_name, user_id=user_id)
signature = self._scope_signature(path, agent_name)
# A no-op commit hands back the unchanged manifest: nothing was
# replaced, so no recorded index signature may move.
if int(memory_file.get("revision") or 0) != pre_revision:
committed = (pre_revision, signature)
with self._cache_lock:
self._memory_cache[key] = (copy.deepcopy(document), signature)
except MemoryRevisionConflict:
@ -1516,7 +1595,7 @@ class FileMemoryStorage(MemoryStorage):
logger.error("Failed to save memory scope %s: %s", key, exc)
return False
self._dispatch_retrieval_notifications(notifications, user_id=user_id, agent_name=agent_name)
self._dispatch_retrieval_notifications(notifications, user_id=user_id, agent_name=agent_name, committed=committed)
if agent_name is not None and deleted_metadata_ids:
self.clear_fact_metadata(
agent_name=agent_name,
@ -1686,12 +1765,14 @@ class FileMemoryStorage(MemoryStorage):
expected = expected_manifest_revision
notifications: list[RetrievalNotification] = []
memory_file: dict[str, Any] | None = None
committed: tuple[int, tuple[Any, ...]] | None = None
safe_delete_rebase = not deletes or (isinstance(delete_revisions, dict) and all(str(fact_id) in delete_revisions for fact_id in deletes))
safe_upsert_rebase = all(str(incoming["id"]) in normalized_upsert_revisions for incoming in upserts)
for attempt in range(3):
try:
with self._scope_lock(key), _process_file_lock(path.parent / ".memory.lock", float(getattr(self._config, "file_lock_timeout_seconds", 10))):
self._recover_if_needed(path)
pre_revision = int((self._load_memory_file(path) or {}).get("revision") or 0)
memory_file, notifications = self._commit_changes_locked(
path,
user_id=user_id,
@ -1703,6 +1784,10 @@ class FileMemoryStorage(MemoryStorage):
delete_revisions=copy.deepcopy(delete_revisions),
upsert_revisions=normalized_upsert_revisions,
)
# A no-op commit (unchanged values, empty change set) hands back
# the unchanged manifest; only a real replacement may advance
# the recorded index signatures.
committed = (pre_revision, self._scope_signature(path, agent_name)) if int(memory_file.get("revision") or 0) != pre_revision else None
break
except MemoryManifestRevisionConflict as exc:
can_rebase = allow_manifest_rebase and has_fact_changes and summaries is None and safe_delete_rebase and safe_upsert_rebase and attempt < 2
@ -1711,7 +1796,7 @@ class FileMemoryStorage(MemoryStorage):
current = self._load_memory_file(path)
expected = int((current or {}).get("revision") or 0)
logger.info("Rebasing disjoint memory fact change after revision conflict: %s", exc)
self._dispatch_retrieval_notifications(notifications, user_id=user_id, agent_name=agent_name)
self._dispatch_retrieval_notifications(notifications, user_id=user_id, agent_name=agent_name, committed=committed)
if memory_file is None: # defensive: the bounded loop either commits or raises
raise MemoryStorageError("Memory repository change did not produce a result")
deleted_fact_ids = [str(value) for action, value, _ in notifications if action == "remove"]
@ -1825,6 +1910,26 @@ class FileMemoryStorage(MemoryStorage):
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
synced = {key: self._retrieval_synced_signatures.get(key) for key in requested_keys}
for key in requested_keys - dirty:
user_id, agent_name = key
if agent_name is None:
continue
# Another process wrote this user's memory since this process last
# indexed the scope: the live manifest signature no longer matches.
try:
live_signature = self._scope_signature(self._get_memory_file_path(agent_name, user_id=user_id), agent_name)
except (OSError, ValueError, MemoryStorageCorruption) as exc:
# The local index exists to decouple search from the shared
# volume: a transient read error or a corrupt manifest must not
# fail a search the index can answer. Serve it as is and keep the
# recorded signature so the next search compares again; a dirty
# mark would send the rebuild into the same failing read.
logger.warning("Memory retrieval freshness check skipped for user %r agent %r; serving the local index: %s", user_id, agent_name, exc)
logger.debug("Memory retrieval freshness check failure", exc_info=exc)
continue
if synced[key] != live_signature:
dirty.add(key)
if dirty:
dirty_scopes = [{"userId": user_id, "agentName": agent_name} for user_id, agent_name in dirty]
rebuild_result = self.rebuild_index(dirty_scopes)
@ -1863,15 +1968,28 @@ class FileMemoryStorage(MemoryStorage):
records: list[tuple[dict[str, Any], dict[str, str | None], str]] = []
indexed = 0
failed = 0
# Manifest signatures taken before each scope's facts are read. A write that
# races the rebuild leaves a newer live signature, so the scope is re-synced on
# its next search instead of hiding behind a signature taken after a stale read.
synced_signatures: dict[tuple[str | None, str | None], tuple[Any, ...]] = {}
if scopes is None:
root = Path(self._config.storage_path) if self._config.storage_path else memory_file_path(self._config).parent
candidates = root.glob("**/facts/**/*.md")
# One signature per manifest: every agent bucket of a user shares its
# memory.json, and _scope_signature reads only that file.
manifest_signatures: dict[Path, tuple[Any, ...]] = {}
for path in candidates:
try:
relative_parts = path.relative_to(root).parts
manifest_path: Path | None = None
if "agents" in relative_parts[:-1]:
agents_index = relative_parts.index("agents")
manifest_path = root.joinpath(*relative_parts[:agents_index]) / self._config.manifest_filename
if manifest_path not in manifest_signatures:
manifest_signatures[manifest_path] = self._scope_signature(manifest_path, relative_parts[agents_index + 1])
fact = _parse_listed_fact(path)
if fact is None:
continue
relative_parts = path.relative_to(root).parts
agents_index = relative_parts.index("agents")
expected_agent = relative_parts[agents_index + 1]
expected_user_bucket = relative_parts[1] if len(relative_parts) > 1 and relative_parts[0] == "users" else None
@ -1883,6 +2001,8 @@ 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)
if manifest_path is not None:
synced_signatures.setdefault(self._cache_key(expected_agent, user_id=original_user), manifest_signatures[manifest_path])
records.append((fact, _scope_dict(original_user, expected_agent), str(path)))
except Exception:
logger.exception("Failed to rebuild retrieval index for %s", path)
@ -1894,44 +2014,88 @@ class FileMemoryStorage(MemoryStorage):
agent_name = kwargs.get("agent_name")
if agent_name is None:
continue
for fact in self.list_facts(**kwargs):
synced_signatures[self._cache_key(agent_name, user_id=kwargs.get("user_id"))] = self._scope_signature(memory_path, agent_name)
# The bulk replace below drops every row this rebuild does not
# supply, so read the complete scope rather than a list_facts page.
for fact in self.load(**kwargs).get("facts", []):
try:
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:
requested_keys = None if scopes is None else {self._cache_key(self._scope_kwargs(scope).get("agent_name"), user_id=self._scope_kwargs(scope).get("user_id")) for scope in scopes}
def bump_generations() -> None:
"""Fence in-flight signature promotions off the scopes published here (caller holds _cache_lock)."""
if requested_keys is None:
self._retrieval_index_generation += 1
else:
for key in requested_keys:
self._retrieval_scope_generations[key] = self._retrieval_scope_generations.get(key, 0) + 1
def mark_synced() -> None:
with self._cache_lock:
bump_generations()
if requested_keys is None:
self._retrieval_dirty_scopes.clear()
self._retrieval_synced_signatures = synced_signatures
else:
self._retrieval_dirty_scopes.difference_update(requested_keys)
self._retrieval_synced_signatures.update(synced_signatures)
def forget_synced() -> None:
with self._cache_lock:
bump_generations()
if requested_keys is None:
self._retrieval_synced_signatures.clear()
else:
for key in requested_keys:
self._retrieval_synced_signatures.pop(key, None)
# Row replacement and signature publication form one unit under
# _retrieval_publish_lock. The adapter lock ends when the install
# returns, so two overlapping refreshes of one scope could otherwise
# install in one order and publish in the other, and the later (live)
# signature would certify the earlier refresh's stale rows. Serialized,
# whichever installs last also publishes last, and a stale captured
# signature published last is caught by the next search's compare.
# Fact reading above stays outside the lock so a slow read never blocks
# refreshes of other scopes. The startup full rebuild holds it for its
# single bulk install; scoped refreshes wait for that, which is fine
# because before warm-up completes a search would be rebuilding its
# scope anyway. Lock order: callers hold no storage lock here
# (search_facts releases _cache_lock first; reload() leaves its
# scope/file-lock block before rebuilding; DeerMem's warm-up lock is an
# outer RLock that nothing inside acquires), and only the adapter's own
# lock and _cache_lock are taken inside, both leaf locks.
with self._retrieval_publish_lock:
bulk_rebuild = getattr(self._retrieval, "rebuild", None)
if callable(bulk_rebuild):
try:
self.notify_fact_upsert(fact, path=path)
indexed += 1
bulk_rebuild(records, scopes=scopes)
indexed = len(records)
mark_synced()
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)
logger.exception("Failed to atomically rebuild retrieval index")
failed += len(records) or 1
forget_synced()
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:
mark_synced()
else:
forget_synced()
return {"supported": True, "indexed": indexed, "failed": failed}
def retrieval_status(self) -> dict[str, Any]:

40
backend/packages/harness/deerflow/agents/memory/manager.py
View File

@ -24,7 +24,7 @@ import os
import sys
import threading
from abc import abstractmethod
from collections.abc import Callable
from collections.abc import Callable, Mapping
from pathlib import Path
from types import ModuleType
from typing import Any, ClassVar, Literal
@ -1099,6 +1099,27 @@ def _refresh_judge_for_reloaded_config(manager: MemoryManager) -> None:
logger.info("Memory judge refreshed after a memory judging-config change")
def resolve_deermem_storage_path(backend_config: Mapping[str, Any]) -> str:
"""Return the absolute DeerMem data root the host injects for ``backend_config``.
Zero-config UX: an empty ``storage_path`` is deer-flow's state dir (absolute,
CWD-independent), so memory lands at ``{runtime_home}/users/{user_id}/memory.json``
(deer-flow's base_dir, same as pre-abstraction). A relative value is resolved
against ``runtime_home()`` to preserve those semantics; left as-is it would be
CWD-relative and fragile. Resolved in host code so the portable ``paths.py``
stays free of any runtime_home dependency. The Gateway startup gate reuses it
to locate the derived retrieval index without constructing a backend.
"""
from deerflow.config.runtime_paths import runtime_home
storage_path = str(backend_config.get("storage_path") or "")
if not storage_path:
return str(runtime_home())
if not Path(storage_path).is_absolute():
return str((Path(runtime_home()) / storage_path).resolve())
return storage_path
# ── Singleton factory ─────────────────────────────────────────────────────
def get_memory_manager() -> MemoryManager:
"""Return the singleton :class:`MemoryManager` for the active config.
@ -1134,22 +1155,7 @@ def get_memory_manager() -> MemoryManager:
manager_class = cfg.manager_class
cls = _resolve_manager_class(manager_class)
backend_config = dict(cfg.backend_config or {})
# Zero-config UX: default DeerMem storage to deer-flow's state dir
# (absolute, CWD-independent) so memory lands at
# {runtime_home}/users/{user_id}/memory.json (deer-flow's base_dir,
# same as pre-abstraction) unless the host explicitly sets storage_path.
if not backend_config.get("storage_path"):
from deerflow.config.runtime_paths import runtime_home
backend_config["storage_path"] = str(runtime_home())
elif not Path(backend_config.get("storage_path", "")).is_absolute():
# A relative storage_path is resolved against runtime_home() (base_dir-
# relative, CWD-independent) to preserve pre-abstraction semantics; left
# as-is it would be CWD-relative and fragile. (Resolved here in host code
# so the portable paths.py stays free of any runtime_home dependency.)
from deerflow.config.runtime_paths import runtime_home
backend_config["storage_path"] = str((Path(runtime_home()) / backend_config["storage_path"]).resolve())
backend_config["storage_path"] = resolve_deermem_storage_path(backend_config)
# storage_path-is-a-file guard lives on DeerMemConfig.model_validator
# now (DeerMem-private semantics; fires even when the factory bypassed).
# Host hook providers: the factory supplies these as kwargs; each

109
backend/tests/test_helm_memory_retrieval_index.py Normal file
View File

@ -0,0 +1,109 @@
"""Regression tests for the Helm chart's Pod-local DeerMem retrieval index.
The chart's default ``config`` carried a legacy ``memory.storage_path:
memory.json`` line that the Gateway dropped with a warning at every start, and
nothing kept DeerMem's derived SQLite FTS5 index off the shared home volume: with
``gateway.replicas > 1`` every Pod opened one WAL database over the
ReadWriteMany filesystem, rebuilt it under its peers at startup and deleted it
from under them on corruption recovery. These tests pin the ``emptyDir`` mounted
at the index path, the ``memory.backend_config.retrieval_index_path`` that
points DeerMem at it, the removal of the legacy key, the config_version sync,
and the operator notes.
The ``helm template`` tests skip when helm is not installed; CI's runner has it.
"""
from __future__ import annotations
import re
import shutil
import subprocess
from pathlib import Path
import pytest
import yaml
REPO_ROOT = Path(__file__).resolve().parents[2]
CHART = REPO_ROOT / "deploy" / "helm" / "deer-flow"
VALUES = CHART / "values.yaml"
README = CHART / "README.md"
NOTES = CHART / "templates" / "NOTES.txt"
GATEWAY_TEMPLATE = CHART / "templates" / "gateway-deployment.yaml"
CONFIG_EXAMPLE = REPO_ROOT / "config.example.yaml"
INDEX_PATH = "/var/lib/deerflow/memory-index"
VOLUME_NAME = "memory-index"
def _values() -> dict:
return yaml.safe_load(VALUES.read_text(encoding="utf-8"))
def _rendered_config() -> dict:
return yaml.safe_load(_values()["config"])
def _render_chart(*settings: str) -> list[dict]:
helm = shutil.which("helm")
if helm is None:
pytest.skip("helm is unavailable")
command = [helm, "template", "deer-flow", str(CHART)]
for setting in settings:
command.extend(["--set", setting])
rendered = subprocess.run(command, check=True, capture_output=True, text=True).stdout
return [document for document in yaml.safe_load_all(rendered) if isinstance(document, dict)]
def _by_kind(documents: list[dict], kind: str, name_suffix: str) -> dict:
return next(document for document in documents if document.get("kind") == kind and document["metadata"]["name"].endswith(name_suffix))
def _gateway_pod(documents: list[dict]) -> dict:
return _by_kind(documents, "Deployment", "-gateway")["spec"]["template"]["spec"]
def test_default_config_points_deermem_at_the_pod_local_index() -> None:
memory = _rendered_config()["memory"]
assert memory["backend_config"]["retrieval_index_path"] == INDEX_PATH
assert "storage_path" not in memory, "the legacy file-style key was dropped with a warning at every Gateway start"
assert "storage_path" not in memory["backend_config"], "canonical memory stays at the default root on the home volume"
def test_gateway_template_mounts_an_empty_dir_at_the_index_path() -> None:
template = GATEWAY_TEMPLATE.read_text(encoding="utf-8")
assert re.search(rf"- name: {VOLUME_NAME}\n\s+mountPath: {re.escape(INDEX_PATH)}\n", template), "the gateway container must mount the index volume at the configured path"
assert re.search(rf"- name: {VOLUME_NAME}\n\s+emptyDir: \{{\}}", template), "the index is rebuildable derived data, so a Pod-local emptyDir is enough"
def test_chart_config_version_tracks_the_example() -> None:
example = yaml.safe_load(CONFIG_EXAMPLE.read_text(encoding="utf-8"))["config_version"]
chart = _rendered_config()["config_version"]
assert chart >= example, "scripts/check_config_version.sh fails the chart build when it lags"
readme_versions = {int(value) for value in re.findall(r"^\s+config_version:\s+(\d+)$", README.read_text(encoding="utf-8"), flags=re.MULTILINE)}
assert readme_versions == {chart}, "the README's config example must show the chart's version"
def test_operator_notes_document_the_pod_local_index() -> None:
for path in (VALUES, README, NOTES):
text = path.read_text(encoding="utf-8")
assert "retrieval_index_path" in text, path.name
assert INDEX_PATH in text, path.name
def test_rendered_gateway_pod_keeps_the_retrieval_index_on_an_empty_dir() -> None:
documents = _render_chart("gateway.replicas=2")
pod = _gateway_pod(documents)
gateway = next(container for container in pod["containers"] if container["name"] == "gateway")
mount = next(item for item in gateway["volumeMounts"] if item["name"] == VOLUME_NAME)
assert mount["mountPath"] == INDEX_PATH
assert "subPath" not in mount
volume = next(item for item in pod["volumes"] if item["name"] == VOLUME_NAME)
assert volume == {"name": VOLUME_NAME, "emptyDir": {}}
assert pod["securityContext"]["fsGroup"] == 1000, "the emptyDir is group-writable for the non-root gateway user"
def test_rendered_config_map_carries_the_index_path() -> None:
documents = _render_chart()
config = yaml.safe_load(_by_kind(documents, "ConfigMap", "-config")["data"]["config.yaml"])
assert config["memory"]["backend_config"]["retrieval_index_path"] == INDEX_PATH
assert "storage_path" not in config["memory"]

580
backend/tests/test_memory_retrieval_index_local.py Normal file
View File

@ -0,0 +1,580 @@
"""DeerMem retrieval-index placement and cross-instance freshness.
Several Gateway instances (Kubernetes Pods) can share one ``storage_path`` on a
ReadWriteMany home volume. The derived SQLite FTS5 index used to be pinned to
``{storage_path}/.retrieval``: one WAL database that every instance opened over
the network filesystem, emptied and refilled at every start, and deleted from
under its peers on any instance's corruption recovery. An instance also never
learned that a peer had written a user's facts, so its index served stale
results. These tests pin ``memory.backend_config.retrieval_index_path`` and the
manifest-signature re-sync that makes a peer's write visible on the next search
without rebuilding a scope this process wrote itself.
"""
from __future__ import annotations
import logging
import threading
from collections.abc import Callable
from pathlib import Path
from unittest.mock import patch
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 FTS5RetrievalAdapter, create_fts5_retrieval
from deerflow.agents.memory.backends.deermem.deermem.core.storage import FileMemoryStorage, MemoryStorageCorruption, create_empty_memory
INDEX_FILENAME = "memory-fts5.sqlite3"
SCOPE = {"userId": "alice", "agentName": "agent-a"}
def _fact(fact_id: str, content: str) -> dict:
return {
"id": fact_id,
"content": content,
"category": "context",
"confidence": 0.8,
"createdAt": "2026-10-01T00:00:00Z",
"source": {"type": "test", "threadId": None},
}
def _ids(results: list[dict]) -> list[str]:
return [item["fact"]["id"] for item in results]
def _instance(storage_root: Path, index_dir: Path) -> FileMemoryStorage:
"""One Gateway instance: shared canonical storage, its own derived index."""
index_dir.mkdir(parents=True, exist_ok=True)
return FileMemoryStorage(DeerMemConfig(storage_path=str(storage_root)), retrieval=FTS5RetrievalAdapter(index_dir / INDEX_FILENAME))
# ── retrieval_index_path ─────────────────────────────────────────────────────
def test_absolute_retrieval_index_path_places_the_index_outside_storage_path(tmp_path: Path) -> None:
storage_root = tmp_path / "home"
index_dir = tmp_path / "pod-local-index"
adapter = create_fts5_retrieval(DeerMemConfig(storage_path=str(storage_root), retrieval_index_path=str(index_dir)))
assert adapter is not None
try:
assert (index_dir / INDEX_FILENAME).is_file()
assert not (storage_root / ".retrieval").exists()
finally:
adapter.close()
def test_relative_retrieval_index_path_resolves_against_storage_path(tmp_path: Path) -> None:
adapter = create_fts5_retrieval(DeerMemConfig(storage_path=str(tmp_path), retrieval_index_path="index/local"))
assert adapter is not None
try:
assert (tmp_path / "index" / "local" / INDEX_FILENAME).is_file()
assert not (tmp_path / ".retrieval").exists()
finally:
adapter.close()
def test_default_retrieval_index_path_is_unchanged(tmp_path: Path) -> None:
adapter = create_fts5_retrieval(DeerMemConfig(storage_path=str(tmp_path)))
assert adapter is not None
try:
assert (tmp_path / ".retrieval" / INDEX_FILENAME).is_file()
finally:
adapter.close()
def test_corruption_recovery_touches_only_the_configured_index(tmp_path: Path) -> None:
"""A Pod recreating its own corrupt index must not delete files a peer holds open."""
storage_root = tmp_path / "home"
index_dir = tmp_path / "pod-local-index"
index_dir.mkdir()
(index_dir / INDEX_FILENAME).write_bytes(b"not a sqlite database")
shared_index = storage_root / ".retrieval" / INDEX_FILENAME
shared_index.parent.mkdir(parents=True)
shared_index.write_bytes(b"a peer's file, must survive")
adapter = create_fts5_retrieval(DeerMemConfig(storage_path=str(storage_root), retrieval_index_path=str(index_dir)))
assert adapter is not None
try:
assert shared_index.read_bytes() == b"a peer's file, must survive"
adapter.upsert(_fact("recovered", "recreated local index"), scope=SCOPE, path="")
assert _ids(adapter.search("recreated", scopes=[SCOPE], top_k=5, mode="fts5", filters=None)) == ["recovered"]
finally:
adapter.close()
# ── cross-instance freshness ─────────────────────────────────────────────────
def test_search_sees_facts_a_peer_instance_wrote(tmp_path: Path) -> None:
"""Two Pods, one storage_path, one Pod-local index each: a write lands in the other Pod's next search."""
storage_root = tmp_path / "home"
pod_a = _instance(storage_root, tmp_path / "index-a")
pod_b = _instance(storage_root, tmp_path / "index-b")
try:
assert pod_a.rebuild_index()["failed"] == 0 # startup warm-up on both Pods
assert pod_b.rebuild_index()["failed"] == 0
pod_a.upsert_fact(_fact("one", "alpha written on pod a"), user_id="alice", agent_name="agent-a")
assert _ids(pod_b.search_facts("alpha", scopes=[SCOPE])) == ["one"]
pod_b.upsert_fact(_fact("two", "beta written on pod b"), user_id="alice", agent_name="agent-a")
assert _ids(pod_a.search_facts("beta", scopes=[SCOPE])) == ["two"]
assert _ids(pod_a.search_facts("alpha", scopes=[SCOPE])) == ["one"]
finally:
pod_a.close()
pod_b.close()
def test_search_drops_facts_a_peer_instance_deleted(tmp_path: Path) -> None:
storage_root = tmp_path / "home"
pod_a = _instance(storage_root, tmp_path / "index-a")
pod_b = _instance(storage_root, tmp_path / "index-b")
try:
pod_a.upsert_fact(_fact("one", "alpha written on pod a"), user_id="alice", agent_name="agent-a")
assert pod_b.rebuild_index()["failed"] == 0
assert _ids(pod_b.search_facts("alpha", scopes=[SCOPE])) == ["one"]
pod_a.delete_fact("one", user_id="alice", agent_name="agent-a")
assert pod_b.search_facts("alpha", scopes=[SCOPE]) == []
finally:
pod_a.close()
pod_b.close()
def test_own_writes_do_not_rebuild_a_scope_this_instance_indexed(tmp_path: Path) -> None:
"""Incremental notifications already cover this process's writes; only a peer's write costs a rebuild."""
storage_root = tmp_path / "home"
pod = _instance(storage_root, tmp_path / "index-a")
try:
pod.upsert_fact(_fact("one", "alpha indexed before warm-up"), user_id="alice", agent_name="agent-a")
assert pod.rebuild_index()["failed"] == 0
with patch.object(pod, "rebuild_index", wraps=pod.rebuild_index) as rebuild:
assert _ids(pod.search_facts("alpha", scopes=[SCOPE])) == ["one"]
pod.upsert_fact(_fact("two", "beta written here"), user_id="alice", agent_name="agent-a")
assert _ids(pod.search_facts("beta", scopes=[SCOPE])) == ["two"]
# Summary updates and writes to another agent bump the shared user manifest too.
summaries = create_empty_memory()
summaries["user"]["workContext"]["summary"] = "works on deployments"
assert pod.save(summaries, user_id="alice")
pod.upsert_fact(_fact("three", "gamma for another agent"), user_id="alice", agent_name="agent-b")
assert _ids(pod.search_facts("alpha", scopes=[SCOPE])) == ["one"]
pod.delete_fact("two", user_id="alice", agent_name="agent-a")
assert pod.search_facts("beta", scopes=[SCOPE]) == []
assert rebuild.call_count == 0
finally:
pod.close()
def test_peer_write_between_own_sync_and_own_write_still_rebuilds(tmp_path: Path) -> None:
"""An own write must not paper over a peer write that landed since this instance last synced."""
storage_root = tmp_path / "home"
pod_a = _instance(storage_root, tmp_path / "index-a")
pod_b = _instance(storage_root, tmp_path / "index-b")
try:
assert pod_a.rebuild_index()["failed"] == 0
assert pod_b.rebuild_index()["failed"] == 0
pod_a.upsert_fact(_fact("one", "alpha written on pod a"), user_id="alice", agent_name="agent-a")
pod_b.upsert_fact(_fact("two", "beta written on pod b"), user_id="alice", agent_name="agent-a")
assert set(_ids(pod_b.search_facts("alpha OR beta", scopes=[SCOPE], mode="fts5"))) == {"one", "two"}
finally:
pod_a.close()
pod_b.close()
def test_deermem_instances_with_pod_local_indexes_see_each_other(tmp_path: Path) -> None:
storage_root = tmp_path / "home"
def pod(index_name: str) -> DeerMem:
return DeerMem(backend_config={"storage_path": str(storage_root), "retrieval_index_path": str(tmp_path / index_name), "token_counting": "char"})
pod_a = pod("index-a")
pod_b = pod("index-b")
try:
assert pod_a.warm_retrieval()
assert pod_b.warm_retrieval()
_, fact_id = pod_a.create_fact("pod a remembers the deployment region", user_id="alice")
assert [fact["id"] for fact in pod_b.search("deployment region", user_id="alice")] == [fact_id]
assert (tmp_path / "index-a" / INDEX_FILENAME).is_file()
assert (tmp_path / "index-b" / INDEX_FILENAME).is_file()
assert not (storage_root / ".retrieval").exists()
finally:
pod_a.close()
pod_b.close()
@pytest.mark.parametrize("no_op", ["unchanged_patch", "identical_save"])
def test_no_op_commit_after_a_peer_delete_does_not_mask_the_deletion(tmp_path: Path, no_op: str) -> None:
"""A commit that changes nothing must not advance the synced signature past a peer's write.
Pod A indexed f1/f2 at revision r; Pod B deleted f2 (r+1); A then applies a
supported update to f1 that turns out to be a no-op. The commit helper hands
back the unchanged manifest, so inferring "previous revision = r" from it
would mark A's stale index as in sync at r+1 and keep returning f2 forever.
"""
storage_root = tmp_path / "home"
pod_a = _instance(storage_root, tmp_path / "index-a")
pod_b = _instance(storage_root, tmp_path / "index-b")
try:
pod_a.apply_changes({"upserts": [_fact("f1", "alpha stays"), _fact("f2", "beta goes away")]}, user_id="alice", agent_name="agent-a")
assert pod_a.rebuild_index()["failed"] == 0
assert _ids(pod_a.search_facts("beta", scopes=[SCOPE])) == ["f2"]
pod_b.delete_fact("f2", user_id="alice", agent_name="agent-a")
if no_op == "unchanged_patch":
stored = pod_a.get_fact("f1", user_id="alice", agent_name="agent-a")
assert stored is not None
result = pod_a.upsert_fact(stored, user_id="alice", agent_name="agent-a", expected_fact_revision=stored["revision"])
assert result["upsertedFacts"] == [], "the unchanged-value patch must be a no-op commit"
else:
assert pod_a.save(pod_a.load("agent-a", user_id="alice"), "agent-a", user_id="alice")
assert pod_a.search_facts("beta", scopes=[SCOPE]) == []
assert _ids(pod_a.search_facts("alpha", scopes=[SCOPE])) == ["f1"]
finally:
pod_a.close()
pod_b.close()
def test_scoped_resync_indexes_every_fact_of_a_large_scope(tmp_path: Path) -> None:
"""The re-sync must read the whole scope, not the first page of 100 facts.
With 150 facts (max_facts allows up to 500) both Pods start complete. A
peer's summary-only save changes the manifest signature; A's next search
rebuilds the scope and must still hold all 150 facts afterwards.
"""
storage_root = tmp_path / "home"
config = DeerMemConfig(storage_path=str(storage_root), max_facts=200)
(tmp_path / "index-a").mkdir()
(tmp_path / "index-b").mkdir()
pod_a = FileMemoryStorage(config, retrieval=FTS5RetrievalAdapter(tmp_path / "index-a" / INDEX_FILENAME))
pod_b = FileMemoryStorage(config, retrieval=FTS5RetrievalAdapter(tmp_path / "index-b" / INDEX_FILENAME))
try:
facts = [_fact(f"fact-{index:03d}", f"zeta common memory number {index:03d} unique{index:03d}") for index in range(1, 151)]
pod_a.apply_changes({"upserts": facts}, user_id="alice", agent_name="agent-a")
assert pod_a.rebuild_index()["failed"] == 0
assert pod_b.rebuild_index()["failed"] == 0
assert _ids(pod_a.search_facts("unique150", scopes=[SCOPE])) == ["fact-150"]
summaries = create_empty_memory()
summaries["user"]["workContext"]["summary"] = "peer summary refresh"
assert pod_b.save(summaries, user_id="alice")
assert _ids(pod_a.search_facts("unique150", scopes=[SCOPE])) == ["fact-150"]
assert _ids(pod_a.search_facts("unique001", scopes=[SCOPE])) == ["fact-001"]
assert len(pod_a.search_facts("zeta", scopes=[SCOPE], top_k=200)) == 150
finally:
pod_a.close()
pod_b.close()
class _PausableAdapter:
"""FTS5 adapter whose next bulk install, next upsert and every remove can be held at a chosen point.
With ``pause_after_next_install`` armed, ``rebuild`` performs the real
install, signals ``installed`` and waits for ``allow_publish`` before
returning, which holds the owning rebuild between its row replacement and
its signature publication (one call; later rebuilds run unpaused).
With ``pause_next_upsert`` armed, ``upsert`` signals ``at_upsert`` and
waits for ``allow_upsert`` before the real upsert, which holds the owning
write after it released the user lock and before its adapter mutation.
``remove`` performs the real removal, signals ``removed`` and waits for
``allow_promote`` before returning, which holds the owning write after its
adapter notification and before its signature promotion.
"""
def __init__(self, inner: FTS5RetrievalAdapter) -> None:
self._inner = inner
self.pause_after_next_install = False
self.installed = threading.Event()
self.allow_publish = threading.Event()
self.pause_next_upsert = False
self.at_upsert = threading.Event()
self.allow_upsert = threading.Event()
self.removed = threading.Event()
self.allow_promote = threading.Event()
def rebuild(self, records, *, scopes):
self._inner.rebuild(records, scopes=scopes)
if self.pause_after_next_install:
self.pause_after_next_install = False
self.installed.set()
assert self.allow_publish.wait(10), "test orchestration stalled before the signature publication"
def upsert(self, fact, *, scope, path):
if self.pause_next_upsert:
self.pause_next_upsert = False
self.at_upsert.set()
assert self.allow_upsert.wait(10), "test orchestration stalled before the adapter upsert"
self._inner.upsert(fact, scope=scope, path=path)
def remove(self, fact_id, *, scope):
self._inner.remove(fact_id, scope=scope)
self.removed.set()
assert self.allow_promote.wait(10), "test orchestration stalled before the signature promotion"
def release(self) -> None:
self.allow_publish.set()
self.allow_upsert.set()
self.allow_promote.set()
def __getattr__(self, name: str):
return getattr(self._inner, name)
class _ReadHold:
"""Hold one scoped refresh after it has read the scope's facts and before it installs them.
The scoped ``rebuild_index`` captures the manifest signature, reads the
facts through ``load()`` and only then replaces the index rows, so holding
here models a refresh that carries a snapshot taken before later writes.
"""
def __init__(self, storage: FileMemoryStorage) -> None:
self.at_read = threading.Event()
self.allow_install = threading.Event()
self.armed = True
real_load = storage.load
def load(*args, **kwargs):
document = real_load(*args, **kwargs)
if self.armed:
self.armed = False
self.at_read.set()
assert self.allow_install.wait(10), "test orchestration stalled before the index install"
return document
storage.load = load # type: ignore[method-assign]
def _run(target: Callable[[], object], errors: list[BaseException]) -> threading.Thread:
def body() -> None:
try:
target()
except BaseException as exc: # noqa: BLE001 - surfaced through the errors list
errors.append(exc)
thread = threading.Thread(target=body, daemon=True)
thread.start()
return thread
def _join(thread: threading.Thread, errors: list[BaseException]) -> None:
thread.join(10)
assert not thread.is_alive(), "worker thread did not finish"
assert errors == []
def _racing_pods(tmp_path: Path) -> tuple[FileMemoryStorage, _PausableAdapter, FileMemoryStorage, _ReadHold]:
"""Warmed Pod A with a pausable index, peer Pod B, and A's next scoped refresh held after its fact read."""
storage_root = tmp_path / "home"
(tmp_path / "index-a").mkdir()
adapter = _PausableAdapter(FTS5RetrievalAdapter(tmp_path / "index-a" / INDEX_FILENAME))
pod_a = FileMemoryStorage(DeerMemConfig(storage_path=str(storage_root)), retrieval=adapter) # type: ignore[arg-type]
pod_b = _instance(storage_root, tmp_path / "index-b")
pod_a.apply_changes({"upserts": [_fact("f1", "alpha stays"), _fact("f2", "beta goes away")]}, user_id="alice", agent_name="agent-a")
assert pod_a.rebuild_index()["failed"] == 0
summaries = create_empty_memory()
summaries["user"]["workContext"]["summary"] = "peer summary refresh"
assert pod_b.save(summaries, user_id="alice") # r+1 on the shared manifest: A's next search re-syncs
return pod_a, adapter, pod_b, _ReadHold(pod_a)
def test_promotion_is_fenced_against_a_refresh_that_installs_in_between(tmp_path: Path) -> None:
"""Reviewer interleaving: refresh reads f1/f2, own delete removes f2, refresh installs, delete promotes.
The refresh captured the r+1 signature and the facts before the deletion
committed r+2. Its install resurrects f2; promoting the deletion's r+2
signature afterwards would declare that stale index in sync with the live
manifest and keep returning f2 until the next manifest change.
"""
pod_a, adapter, pod_b, hold = _racing_pods(tmp_path)
errors: list[BaseException] = []
try:
refresh = _run(lambda: pod_a.search_facts("alpha", scopes=[SCOPE]), errors)
assert hold.at_read.wait(10) # refresh holds f1/f2 read at r+1, not yet installed
deletion = _run(lambda: pod_a.delete_fact("f2", user_id="alice", agent_name="agent-a"), errors)
assert adapter.removed.wait(10) # r+2 committed, index row removed, promotion pending
hold.allow_install.set() # stale install resurrects f2 and records the r+1 signature
_join(refresh, errors)
adapter.allow_promote.set() # the deletion must not stamp that index as r+2
_join(deletion, errors)
assert pod_a.search_facts("beta", scopes=[SCOPE]) == []
assert _ids(pod_a.search_facts("alpha", scopes=[SCOPE])) == ["f1"]
finally:
hold.allow_install.set()
adapter.release()
pod_a.close()
pod_b.close()
def test_stale_refresh_installing_after_a_promotion_is_caught_by_the_next_search(tmp_path: Path) -> None:
"""Mirrored order: the promotion lands first, then a stale refresh installs f1/f2.
The stale refresh records the r+1 signature it captured, so the next search
sees it differ from the live r+2 manifest and rebuilds the scope.
"""
pod_a, adapter, pod_b, hold = _racing_pods(tmp_path)
errors: list[BaseException] = []
try:
stale_refresh = _run(lambda: pod_a.search_facts("alpha", scopes=[SCOPE]), errors)
assert hold.at_read.wait(10)
assert pod_a.rebuild_index([SCOPE])["failed"] == 0 # a second, unpaused refresh syncs the scope at r+1
adapter.allow_promote.set()
pod_a.delete_fact("f2", user_id="alice", agent_name="agent-a") # r+2: removes the row and promotes r+1 -> r+2
assert pod_a.search_facts("beta", scopes=[SCOPE]) == []
hold.allow_install.set() # the stale install resurrects f2 under its own r+1 signature
_join(stale_refresh, errors)
assert pod_a.search_facts("beta", scopes=[SCOPE]) == []
assert _ids(pod_a.search_facts("alpha", scopes=[SCOPE])) == ["f1"]
finally:
hold.allow_install.set()
adapter.release()
pod_a.close()
pod_b.close()
def test_rebuild_publication_stays_ordered_with_row_replacement(tmp_path: Path) -> None:
"""Reviewer interleaving: two refreshes install in one order and would publish in the other.
R1 captures r+1 with f1/f2 and is held before installing. A peer deletes f2
(r+2). R2 captures r+2, installs f1 only and is held before publishing. R1
then installs its stale f1/f2 and publishes r+1; R2 publishes r+2 last.
Without serializing install+publish, R2's live-matching signature certifies
R1's stale rows and the deleted f2 is returned until the manifest changes.
"""
pod_a, adapter, pod_b, hold = _racing_pods(tmp_path)
errors: list[BaseException] = []
try:
first = _run(lambda: pod_a.search_facts("alpha", scopes=[SCOPE]), errors)
assert hold.at_read.wait(10) # R1 holds f1/f2 read at r+1
pod_b.delete_fact("f2", user_id="alice", agent_name="agent-a") # r+2
adapter.pause_after_next_install = True
second = _run(lambda: pod_a.search_facts("alpha", scopes=[SCOPE]), errors)
assert adapter.installed.wait(10) # R2 installed f1 at r+2, publication pending
hold.allow_install.set() # R1 proceeds: with install+publish serialized it waits for R2; unfixed, it installs f1/f2 and publishes r+1 now
first.join(1.5)
adapter.allow_publish.set() # R2 publishes r+2
_join(second, errors)
_join(first, errors)
assert pod_a.search_facts("beta", scopes=[SCOPE]) == []
assert _ids(pod_a.search_facts("alpha", scopes=[SCOPE])) == ["f1"]
finally:
hold.allow_install.set()
adapter.release()
pod_a.close()
pod_b.close()
def test_full_rebuild_reads_each_manifest_once(tmp_path: Path) -> None:
"""Every agent bucket of a user shares one memory.json; the startup scan must not re-read it per bucket."""
pod = _instance(tmp_path / "home", tmp_path / "index-a")
try:
for agent in ("agent-a", "agent-b", "agent-c"):
pod.upsert_fact(_fact(f"{agent}-fact", f"fact for {agent}"), user_id="alice", agent_name=agent)
pod.upsert_fact(_fact("bob-fact", "fact for bob"), user_id="bob", agent_name="agent-a")
with patch.object(pod, "_load_memory_file", wraps=pod._load_memory_file) as manifest_reads:
result = pod.rebuild_index()
assert result == {"supported": True, "indexed": 4, "failed": 0}
read_paths = sorted({call.args[0] for call in manifest_reads.call_args_list})
assert read_paths == [pod._get_memory_file_path(user_id="alice"), pod._get_memory_file_path(user_id="bob")]
assert manifest_reads.call_count == 2, "one manifest read per user, not per agent bucket"
for agent in ("agent-a", "agent-b", "agent-c"):
assert _ids(pod.search_facts(f"{agent}", scopes=[{"userId": "alice", "agentName": agent}])) == [f"{agent}-fact"]
finally:
pod.close()
@pytest.mark.parametrize("failure", [OSError("stale NFS handle"), MemoryStorageCorruption("malformed memory.json")])
def test_manifest_read_failure_during_the_freshness_compare_serves_the_local_index(tmp_path: Path, caplog: pytest.LogCaptureFixture, failure: Exception) -> None:
"""The Pod-local index exists to decouple search from the shared volume.
A transient read error or a corrupt manifest during the per-search compare
must log and serve the current index, leave the recorded signature alone
(no dirty mark: a rebuild would hit the same read), and let the next search
compare normally.
"""
storage_root = tmp_path / "home"
pod_a = _instance(storage_root, tmp_path / "index-a")
pod_b = _instance(storage_root, tmp_path / "index-b")
key = ("alice", "agent-a")
try:
pod_a.upsert_fact(_fact("one", "alpha indexed locally"), user_id="alice", agent_name="agent-a")
assert pod_a.rebuild_index()["failed"] == 0
synced_before = pod_a._retrieval_synced_signatures[key]
with patch.object(pod_a, "_load_memory_file", side_effect=failure), patch.object(pod_a, "rebuild_index", wraps=pod_a.rebuild_index) as rebuild, caplog.at_level(logging.WARNING):
assert _ids(pod_a.search_facts("alpha", scopes=[SCOPE])) == ["one"]
assert rebuild.call_count == 0
warnings = [record for record in caplog.records if record.levelno == logging.WARNING and "agent-a" in record.getMessage()]
assert warnings, "the skipped compare must be visible to operators"
assert pod_a._retrieval_synced_signatures[key] == synced_before
assert key not in pod_a._retrieval_dirty_scopes
pod_b.upsert_fact(_fact("two", "beta written on pod b"), user_id="alice", agent_name="agent-a")
assert _ids(pod_a.search_facts("beta", scopes=[SCOPE])) == ["two"] # the compare works again and re-syncs
finally:
pod_a.close()
pod_b.close()
def test_stale_own_delta_replayed_over_a_newer_snapshot_forgets_the_scope(tmp_path: Path) -> None:
"""Reviewer interleaving: own PATCH commits r+1 and is held before its adapter upsert; peer deletes;
a search rebuilds without the fact and publishes r+2; the held upsert then reinserts the deleted fact.
The promotion rule cannot promote (the scope is synced at r+2, not at the
PATCH's previous r), but merely skipping would leave the r+2 signature
trusted over rows this delta just mutated. The scope's signature must be
forgotten so the next search rebuilds without the deleted fact.
"""
storage_root = tmp_path / "home"
(tmp_path / "index-a").mkdir()
adapter = _PausableAdapter(FTS5RetrievalAdapter(tmp_path / "index-a" / INDEX_FILENAME))
pod_a = FileMemoryStorage(DeerMemConfig(storage_path=str(storage_root)), retrieval=adapter) # type: ignore[arg-type]
pod_b = _instance(storage_root, tmp_path / "index-b")
errors: list[BaseException] = []
try:
pod_a.apply_changes({"upserts": [_fact("f1", "alpha stays"), _fact("f2", "beta version one")]}, user_id="alice", agent_name="agent-a")
assert pod_a.rebuild_index()["failed"] == 0 # warmed at r
stored = pod_a.get_fact("f2", user_id="alice", agent_name="agent-a")
assert stored is not None
adapter.pause_next_upsert = True
patch_thread = _run(lambda: pod_a.upsert_fact({**stored, "content": "beta version two"}, user_id="alice", agent_name="agent-a", expected_fact_revision=stored["revision"]), errors)
assert adapter.at_upsert.wait(10) # r+1 committed, user lock released, adapter upsert pending
pod_b.delete_fact("f2", user_id="alice", agent_name="agent-a") # r+2 on the shared manifest
assert pod_a.search_facts("beta", scopes=[SCOPE]) == [] # A re-syncs without f2 and publishes r+2
adapter.allow_upsert.set() # the stale r+1 delta reinserts the deleted fact
_join(patch_thread, errors)
assert pod_a.search_facts("beta", scopes=[SCOPE]) == []
assert _ids(pod_a.search_facts("alpha", scopes=[SCOPE])) == ["f1"]
finally:
adapter.release()
pod_a.close()
pod_b.close()

87
backend/tests/test_multi_worker_postgres_gate.py
View File

@ -18,7 +18,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastapi import FastAPI
from app.gateway.deps import _enforce_postgres_for_multi_worker, _validate_agent_storage, langgraph_runtime
from app.gateway.deps import _enforce_postgres_for_multi_worker, _validate_agent_storage, _validate_memory_retrieval_index, langgraph_runtime
from app.gateway.routers.browser import _browser_tools_enabled
from deerflow.config.database_config import DatabaseConfig
from deerflow.config.deployment_config import MULTI_INSTANCE_ENV_VAR, DeploymentConfig, multi_instance_declaration
@ -707,3 +707,88 @@ def test_deployment_declaration_helpers(monkeypatch):
by_env = multi_instance_declaration(SimpleNamespace(deployment=DeploymentConfig(multi_instance=True)))
assert by_env is not None
assert (by_env.source, by_env.knob, by_env.rollback) == ("env", f"{MULTI_INSTANCE_ENV_VAR}=replicas", f"unset {MULTI_INSTANCE_ENV_VAR}")
# ---------------------------------------------------------------------------
# DeerMem retrieval index: a declared multi-instance deployment must keep the
# derived SQLite index off the shared memory volume.
# ---------------------------------------------------------------------------
def _memory_config(*, enabled: bool = True, manager_class: str = "deermem", **backend_config):
return SimpleNamespace(enabled=enabled, manager_class=manager_class, backend_config=backend_config)
def _retrieval_index_warnings(caplog) -> list[str]:
return [r.message for r in caplog.records if "retrieval_index_path" in r.message]
def test_declared_multi_instance_warns_when_the_retrieval_index_shares_the_memory_root(monkeypatch, caplog, tmp_path):
monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, "true")
config = _cluster_ready()
config.memory = _memory_config(storage_path=str(tmp_path))
with caplog.at_level("WARNING"):
_validate_memory_retrieval_index(config)
messages = _retrieval_index_warnings(caplog)
assert messages, "the default index location is inside storage_path and therefore on the shared volume"
assert f"{MULTI_INSTANCE_ENV_VAR}=true" in messages[0], "must name the knob that declared the topology"
assert str(tmp_path / ".retrieval") in messages[0]
assert str(tmp_path) in messages[0]
def test_declared_multi_instance_warns_when_a_relative_index_path_stays_below_storage_path(caplog, tmp_path):
config = _cluster_ready(deployment_multi_instance=True)
config.memory = _memory_config(storage_path=str(tmp_path), retrieval_index_path="pod-index")
with caplog.at_level("WARNING"):
_validate_memory_retrieval_index(config)
messages = _retrieval_index_warnings(caplog)
assert messages and "deployment.multi_instance=true" in messages[0]
assert str(tmp_path / "pod-index") in messages[0]
def test_declared_multi_instance_defaults_the_memory_root_to_runtime_home(monkeypatch, caplog, tmp_path):
"""An empty storage_path means the host injects runtime_home(), which is the shared home volume."""
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
config = _cluster_ready(deployment_multi_instance=True)
config.memory = _memory_config()
with caplog.at_level("WARNING"):
_validate_memory_retrieval_index(config)
messages = _retrieval_index_warnings(caplog)
assert messages and str(tmp_path.resolve() / ".retrieval") in messages[0]
def test_declared_multi_instance_accepts_an_instance_local_retrieval_index(caplog, tmp_path):
config = _cluster_ready(deployment_multi_instance=True)
config.memory = _memory_config(storage_path=str(tmp_path / "home"), retrieval_index_path=str(tmp_path / "pod-local"))
with caplog.at_level("WARNING"):
_validate_memory_retrieval_index(config)
assert _retrieval_index_warnings(caplog) == []
def test_retrieval_index_warning_requires_the_multi_instance_declaration(monkeypatch, caplog, tmp_path):
"""Workers of one process tree share local disk, where a shared WAL index is supported."""
monkeypatch.setenv("GATEWAY_WORKERS", "2")
config = _cluster_ready()
config.memory = _memory_config(storage_path=str(tmp_path))
with caplog.at_level("WARNING"):
_validate_memory_retrieval_index(config)
assert _retrieval_index_warnings(caplog) == []
@pytest.mark.parametrize(
"memory",
[
None,
_memory_config(enabled=False, storage_path="/shared/home"),
_memory_config(manager_class="mem0", storage_path="/shared/home"),
_memory_config(storage_path="/shared/home", retrieval_adapter=""),
_memory_config(storage_path="/shared/home", retrieval_adapter="my_pkg.retrieval:create"),
],
)
def test_retrieval_index_warning_only_covers_the_bundled_fts5_index(caplog, memory):
config = _cluster_ready(deployment_multi_instance=True)
if memory is not None:
config.memory = memory
with caplog.at_level("WARNING"):
_validate_memory_retrieval_index(config)
assert _retrieval_index_warnings(caplog) == []

8
config.example.yaml
View File

@ -20,7 +20,7 @@
# ============================================================================
# Bump this number when the config schema changes.
# Run `make config-upgrade` to merge new fields into your local config.yaml.
config_version: 55
config_version: 56
# Optional operator-owned literal prompt extensions (see CONFIGURATION.md):
# lead_prompt_overlay:
@ -2200,6 +2200,11 @@ summarization:
# DeerMem-private fields live under ``backend_config`` (NOT at the memory: top level):
# storage_path - Data root. Empty = deer-flow base_dir (factory injects absolute
# runtime_home). Absolute path = that root. Relative = CWD-relative.
# retrieval_index_path - Directory of the derived SQLite FTS5 retrieval index. Empty =
# {storage_path}/.retrieval. Relative = below storage_path. Absolute =
# as-is. Point it at instance-local disk when several Gateway
# instances share storage_path (SQLite WAL is unsupported on network
# filesystems; peers would rebuild or delete each other's index).
# model - LLM config for memory extraction: {provider, model, api_key, base_url,
# temperature}. Omit all fields = no extraction (non-LLM ops still work).
# debounce_seconds - Debounce wait before processing queued updates (default: 30)
@ -2352,6 +2357,7 @@ memory:
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: fts5 # fts5 (default), empty to disable, or a dotted RetrievalPort factory(config)
retrieval_index_path: "" # empty = {storage_path}/.retrieval; relative = below storage_path; absolute as-is. Rebuildable derived data: keep it instance-local when Gateway instances share storage_path (the Helm chart uses an emptyDir)
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

13
deploy/helm/deer-flow/README.md
View File

@ -135,7 +135,7 @@ they resolve from the `secrets` map):
```yaml
config: |
config_version: 55
config_version: 56
models:
- name: gpt-4
use: langchain_openai:ChatOpenAI
@ -307,6 +307,17 @@ kubectl -n deer-flow exec deploy/deer-flow-provisioner -- curl -s localhost:8002
`persistence.home.accessMode: ReadWriteMany` on multi-node clusters (thread
uploads, outputs, memory and `extensions_config.json` live on that volume)
and `agent_storage.backend: db` so custom agents are visible on every Pod.
Memory itself is multi-instance safe on that volume (per-user file locks,
journaled writes, and a Pod re-syncs a user's search index on its next
search after a peer writes), but DeerMem's derived SQLite FTS5 index is
not: SQLite WAL is unsupported on network filesystems, every Pod start would
rebuild a shared index under its peers, and one Pod's corruption recovery
would delete it from under them. The default `config` therefore sets
`memory.backend_config.retrieval_index_path: /var/lib/deerflow/memory-index`,
a Pod-local `emptyDir` the gateway Deployment mounts; keep that line when you
override `config:` — a multi-instance gateway that leaves the index under
the memory root logs a warning at startup. The index is rebuilt from the
Markdown facts on every Pod start, so losing the emptyDir loses nothing.
A `PodDisruptionBudget` (`minAvailable: 1`) is rendered automatically for a
multi-instance gateway (same rule), and the rollout strategy is
surge-then-drain (`maxSurge: 1`, `maxUnavailable: 0`).

4
deploy/helm/deer-flow/templates/NOTES.txt
View File

@ -66,6 +66,10 @@ Provisioner / sandbox notes:
• On multi-node clusters you additionally need
`persistence.home.accessMode: ReadWriteMany` (e.g. NFS) so sandbox Pods on
other nodes can mount the shared user-data PVC.
• Memory data stays on that PVC; DeerMem's rebuildable SQLite search index
is Pod-local (`memory.backend_config.retrieval_index_path`, an emptyDir at
/var/lib/deerflow/memory-index). Keep that config line when you override
`config`, or gateway Pods share one SQLite WAL file over the network.
Generated secrets (persisted across upgrades):

9
deploy/helm/deer-flow/templates/gateway-deployment.yaml
View File

@ -254,6 +254,13 @@ spec:
- name: home
mountPath: /app/backend/.deer-flow
subPath: deer-flow
# Pod-local home for DeerMem's rebuildable SQLite retrieval index
# (config memory.backend_config.retrieval_index_path). SQLite WAL is
# unsupported on network filesystems and peers must not rebuild or
# delete each other's index, so it never lives on the shared home
# volume; fsGroup 1000 above makes the emptyDir writable by uid 1000.
- name: memory-index
mountPath: /var/lib/deerflow/memory-index
volumes:
- name: config
configMap:
@ -278,3 +285,5 @@ spec:
{{- else }}
emptyDir: {}
{{- end }}
- name: memory-index
emptyDir: {}

17
deploy/helm/deer-flow/values.yaml
View File

@ -37,6 +37,11 @@ gateway:
# while channels are enabled), the WeChat QR login, and browser tools.
# Multi-node clusters also need `persistence.home.accessMode: ReadWriteMany`,
# and `agent_storage.backend: db` keeps custom agents visible on every Pod.
# Memory stays on that shared volume, but DeerMem's SQLite retrieval index is
# Pod-local (`memory.backend_config.retrieval_index_path` in `config`, on an
# emptyDir at /var/lib/deerflow/memory-index); keep that line when you
# override `config`, or the Gateway warns at startup and every Pod shares one
# WAL database over the network filesystem.
# Scheduler multi_instance recovery is safe only when its config
# prerequisites are met; max_concurrent_runs is shared across Pods then.
replicas: 1
@ -324,7 +329,7 @@ ingress:
# -- DeerFlow config.yaml content. Secrets MUST stay as $VAR references — never
# inline literal secret values here. The default enables provisioner sandbox.
config: |
config_version: 55
config_version: 56
log_level: info
recursion_limit: 100
max_recursion_limit: 1000
@ -406,8 +411,16 @@ config: |
# agent_storage:
# backend: db
# Canonical memory (users/{uid}/memory.json plus fact Markdown) lives at the
# default root on the home volume. The derived SQLite FTS5 retrieval index is
# rebuildable and must not be shared between Pods: SQLite WAL is unsupported
# on network filesystems, every Pod start would rebuild the shared index
# under its peers, and one Pod's corruption recovery would delete files the
# others hold open. gateway-deployment.yaml mounts an emptyDir at this path;
# a Pod re-syncs a user's index on its next search after a peer writes.
memory:
storage_path: memory.json
backend_config:
retrieval_index_path: /var/lib/deerflow/memory-index
verification:
receipts_enabled: true