github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-07-27 16:37:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(memory): add incremental agent-scoped Markdown fact storage (#4279)

Browse Source 
* feat(memory): add Markdown fact storage repository

* docs(memory): explain storage rewrite for beginners

* docs(memory): fix plan markdown formatting

* refactor(memory): separate global summaries from agent facts

* fix(memory): make Markdown fact updates incremental and safe

* Update STORAGE_REWRITE_CHANGES.md

* Delete docs/plans/STORAGE_REWRITE_PLAN.md

* Delete docs/plans/STORAGE_REWRITE_CHANGES.md

* fix(memory): address Markdown storage review feedback

* fix(memory): complete review follow-ups

* fix(memory): resolve storage review findings

* feat(memory): add proactive Markdown migration CLI

* fix(memory): harden Markdown storage concurrency

* fix(memory): harden markdown storage migration

* fix(memory): close migration review gaps

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
Co-authored-by: qin-chenghan <qinchenghan@Huawei.com>
This commit is contained in:
RongfuShuiping 2026-07-22 07:46:03 +08:00 committed by GitHub
parent cd86275638
commit 4bf028d048
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
23 changed files with 4136 additions and 232 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
README.md
View File

@ -810,6 +810,23 @@ Across sessions, DeerFlow builds a persistent memory of your profile, preference
Memory updates now skip duplicate fact entries at apply time, so repeated preferences and context do not accumulate endlessly across sessions.
File-backed memory now separates global user context from agent facts. Each user has one `memory.json` containing only the project-independent `user` and `history` summaries; every fact is a canonical Markdown file below `agents/{agent_name}/facts/`. Existing lead-agent middleware, API, Settings, import/export, and embedded-client calls that omit `agent_name` resolve inside DeerMem to the reserved `__default__` bucket. That bucket is outside the valid custom-agent name grammar, so a real custom agent named `lead-agent` has a separate fact repository and deleting a custom agent cannot delete a memory-only directory without `config.yaml`. Public agent identifiers are case-insensitive and canonicalized to lowercase. Runtime/API readers still receive a compatibility `facts` array for the selected/default agent, so the frontend does not read agent facts from `memory.json`; structured Markdown `source` metadata is projected to the historical string field at the MemoryManager boundary. An unscoped Clear All first migrates facts from unread legacy per-agent JSON without adopting its soon-to-be-cleared summaries, then removes shared summaries and facts from every agent bucket while preserving agent configuration files, so a later read cannot resurrect skipped legacy facts; an explicitly agent-scoped clear removes only that agent's facts. On first normal read, old facts embedded in the user JSON are migrated automatically to `__default__`; facts written to the earlier implicit `lead-agent` bucket are also moved when that directory is not a real custom agent. Migration and normal writes notify a configured retrieval adapter only after durable storage locks are released. Journaled writes, a shared user lock, and optimistic user-memory revisions prevent silent lost updates. Retrieval engines remain optional behind the memory retrieval-adapter contract, with an explicit local substring fallback when no adapter is configured.
Single-fact repository operations are genuinely incremental: an upsert/delete reads, journals, writes, and re-indexes only the addressed fact files, and returns an explicit incomplete delta rather than a cache-dependent fake full document. Summary change sets merge the supplied `user`/`history` child keys over the persisted sections so a partial update cannot erase omitted siblings; full imports normalize both sections to the complete compatibility schema before applying replacement values. Manager/API compatibility methods materialize a fresh full document only when their public response contract requires one. Fact-level point operations use separate expected user-memory and fact revisions and may explicitly rebase when every addressed fact precondition still holds. Snapshot-derived operations such as scoped clear, capped create, consolidation, and trimming never replay stale delete/trim sets: a manifest conflict reloads the complete document and recomputes the operation, with a bounded retry. Fact paths use the first two hexadecimal characters of `SHA-256(fact_id)` so generated `fact_*` IDs distribute across shards. The cache token combines the shared JSON's nanosecond mtime, size, and persisted revision; this prevents coarse-mtime same-size writes from returning stale data without scanning fact files. Direct out-of-band Markdown edits require an explicit reload. Storage-specific conflicts and corruption are translated at the MemoryManager boundary; the Gateway returns conflict as HTTP 409 and a stable, non-sensitive corruption error as HTTP 500. Full-document `save()` remains a compatibility API and computes a diff before writing; malformed or missing `facts` can no longer silently erase an agent's Markdown files. Legacy migration preserves non-empty `user`/`history` before deleting an agent `memory.json`; conflicting summaries keep the legacy file and fail loudly instead of choosing a winner.
Legacy facts in `memory.json` migrate automatically into the reserved `__default__` Markdown bucket on the user's first normal memory read. Operators who prefer to audit or complete the migration before serving traffic can run the optional idempotent CLI from `backend/`:
```bash
PYTHONPATH=. python scripts/migrate_memory_markdown.py --all-users --dry-run
PYTHONPATH=. python scripts/migrate_memory_markdown.py --all-users
# A custom DeerMem root or original non-directory-safe identity can be explicit:
PYTHONPATH=. python scripts/migrate_memory_markdown.py --storage-path /path/to/deerflow-home --user-id 'test@example.com'
```
The v1-to-v2 storage migration is one-way for a running application: pre-PR code does not read Markdown facts. Before upgrading a persistent deployment, stop DeerFlow and take a filesystem snapshot or full backup of the configured memory storage root. The migration also durably retains each destructive JSON source beside the original path as `{manifest_filename}.v1.bak` before writing v2 data; an existing mismatched backup or a backup-write failure stops migration without modifying the v1 source. This local backup preserves pre-migration data but is not a substitute for a full snapshot and does not contain facts created after the upgrade.
`--user-id` may be repeated. `--all-users` discovers the existing directory-safe buckets below the selected storage root; standalone integrations that passed raw IDs containing characters such as `@` should use the original value with `--user-id`. A failed user's migration is reported without hiding the rest of the audit, and the command exits non-zero when any user fails. The automatic first-read path remains enabled, so running this CLI is not required for startup.
## Recommended Models
DeerFlow is model-agnostic — it works with any LLM that implements the OpenAI-compatible API. That said, it performs best with models that support:

26
backend/AGENTS.md
View File

@ -604,15 +604,15 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_
### Memory System (`packages/harness/deerflow/agents/memory/`)
**Components**:
- `updater.py` - LLM-based memory updates with fact extraction, whitespace-normalized fact deduplication (trims leading/trailing whitespace before comparing), and atomic file I/O
- `updater.py` - LLM-based memory updates with fact extraction, whitespace-normalized fact deduplication, optimistic revision checks, and repository change sets
- `queue.py` - Debounced update queue (per-thread deduplication, configurable wait time); captures `user_id` at enqueue time so it survives the `threading.Timer` boundary
- `prompt.py` - Prompt templates for memory updates
- `storage.py` - File-based storage with per-user isolation; cache keyed by `(user_id, agent_name)` tuple
- `storage.py` - File repository with one user-global summary JSON, agent-owned single-fact Markdown, target-only journaled changes, strict fact validation, shared-user plus per-fact optimistic revisions, lock-protected migration, deep-copy caching, and an optional retrieval adapter
- `tools.py` - Tool-driven memory mode (`memory_search`, `memory_add`, `memory_update`, `memory_delete`) using the same storage/update primitives
**Per-User Isolation**:
- Memory is stored per-user at `{base_dir}/users/{user_id}/memory.json`
- Per-agent per-user memory at `{base_dir}/users/{user_id}/agents/{agent_name}/memory.json`
- Per-agent facts at `{base_dir}/users/{user_id}/agents/{agent_name}/facts/{sha256-prefix}/{fact-id}.md`, where the prefix is the first two hexadecimal characters of `SHA-256(fact_id)`; there is no per-agent `memory.json`
- Custom agent definitions (`SOUL.md` + `config.yaml`) are also per-user at `{base_dir}/users/{user_id}/agents/{agent_name}/`. The legacy shared layout `{base_dir}/agents/{agent_name}/` remains read-only fallback for unmigrated installations
- Middleware mode captures `user_id` via `get_effective_user_id()` at enqueue time; tool mode resolves `user_id` and `agent_name` from `ToolRuntime.context` via `resolve_runtime_user_id(runtime)` so tool calls stay scoped to the authenticated user and active custom agent
- The `/api/memory*` endpoints resolve the owner through `_resolve_memory_user_id(request)`: trusted internal callers (IM channel workers carrying the `X-DeerFlow-Owner-User-Id` header, e.g. a bound `/memory` command) act for the connection owner; browser/API callers fall back to `get_effective_user_id()`. The header is only honored after `AuthMiddleware` validated the internal token, mirroring `get_trusted_internal_owner_user_id` used by the threads router
@ -620,16 +620,23 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_
- Absolute `storage_path` in config opts out of per-user isolation
- **Migration**: Run `PYTHONPATH=. python scripts/migrate_user_isolation.py` to move legacy `memory.json`, `threads/`, and `agents/` into per-user layout. Supports `--dry-run` (preview changes) and `--user-id USER_ID` (assign unowned legacy data to a user, defaults to `default`).
**Data Structure** (stored in `{base_dir}/users/{user_id}/memory.json`):
**Data Structure**:
- **User Context**: `workContext`, `personalContext`, `topOfMind` (1-3 sentence summaries)
- **History**: `recentMonths`, `earlierContext`, `longTermBackground`
- **Facts**: Discrete facts with `id`, `content`, `category` (preference/knowledge/context/behavior/goal), `confidence` (0-1), `createdAt`, `source`
- **Global JSON**: `{base_dir}/users/{user_id}/memory.json` stores only `version`, shared revision/time, `user`, and `history`; it never stores facts or a fact index
- **Facts**: Schema-v2 Markdown documents under `agents/{agent_name}/facts/{sha256-prefix}/{fact-id}.md`; YAML front matter contains structure and the body contains the atomic fact
- **Default agent compatibility**: DeerMem resolves an omitted `agent_name` to the reserved `__default__` fact bucket at the manager boundary. The sentinel is accepted only by DeerMem storage and is outside the custom-agent name grammar, so a real custom `lead-agent` remains isolated. Public agent identifiers are case-insensitive and canonicalized to lowercase before storage
- **Compatibility view**: direct global storage reads return `facts: []`, while DeerMem Manager/API reads select the explicit agent or reserved default and return its facts, so existing Settings and embedded-client schemas remain stable. Markdown keeps structured `source` metadata internally; the manager projects it to the historical string field before returning a public document
- **Incremental result contract**: `FileMemoryStorage.apply_changes()` returns `complete: false` plus `upsertedFacts`/`deletedFactIds`; it never presents a partial cache as a complete memory document. Public compatibility callers explicitly reload a fresh complete view only where their response contract requires it, including after successful disjoint-create rebases
- **Repository**: `get/list/upsert/delete_fact`, `apply_changes`, summary operations, migration, index lifecycle/status, and scoped search. `apply_changes` and direct fact CRUD touch only target Markdown files; direct fact CRUD accepts separate expected user-memory and fact revisions. Supplied summary child keys merge over their persisted section, while import normalizes complete replacement sections first. Whole-document `load/save` remains for compatibility but validates the complete `facts` list and diffs it before persistence. An unscoped manager clear first migrates facts from unread legacy agent JSON without adopting potentially conflicting summaries, then removes the global summaries and every agent's canonical facts while preserving agent configuration; an explicit agent clear removes only that bucket's facts and preserves the shared summaries
**Workflow**:
- `memory.mode: middleware` (default) keeps the passive path: `MemoryMiddleware` filters messages (user inputs + final AI responses), captures `user_id` via `get_effective_user_id()`, queues conversation with the captured `user_id`, and the debounced background thread invokes the LLM to extract context updates and facts using the stored `user_id`.
- `memory.mode: tool` skips `MemoryMiddleware` and registers `memory_search`, `memory_add`, `memory_update`, and `memory_delete` on the agent. The model decides when to search, add, update, or delete facts; this is opt-in/experimental and should not be described as better than middleware mode without eval evidence.
- Both modes share `FileMemoryStorage`, per-user/per-agent isolation, prompt injection, manual CRUD primitives, and the updater backend.
- Middleware mode queue debounces (30s default), batches updates, deduplicates per-thread, applies updates atomically (temp file + rename) with cache invalidation, and skips duplicate fact content before append.
- Middleware mode queue debounces (30s default), batches updates, and commits global summaries plus the selected/default agent's fact delta through a user-level lock, optimistic user-memory revisions, per-fact revisions, and a recoverable target-file journal. Only explicitly marked point operations may rebase a stale shared revision, and only while every addressed fact still satisfies its original absent/revision precondition. Snapshot-derived clear/trim/consolidation operations instead reload the complete document and recompute their intent on a manifest conflict, with a bounded retry. Typed manifest/fact conflict subclasses keep that decision independent of exception text, and same-ID creates and stale same-fact writes fail. Scope-lock objects are weakly cached so inactive users do not grow a process-lifetime map. Cache validation does not scale with the fact-file count: its token combines the shared JSON's `(mtime_ns, size, revision)`, so the persisted revision invalidates stale caches even when a coarse-mtime filesystem reports identical metadata for same-size writes; direct out-of-band Markdown edits require `reload()`. Atomic replacement also syncs the parent directory on POSIX so the rename is durable. DeerMem translates private storage conflict/corruption exceptions to the backend-neutral MemoryManager contract; the Gateway maps them to HTTP 409 and a stable HTTP 500 response respectively. A normal default-manager read automatically migrates legacy facts from the global JSON into `__default__`; it also adopts the earlier implicit `lead-agent` fact bucket only when that directory has no custom-agent `config.yaml`, and rejects unexpected files instead of deleting them. The v1-to-v2 migration is one-way for the running application: operators must stop DeerFlow and snapshot the configured storage root before upgrade. Before any destructive v2 write, every migrated JSON source is durably retained as `{manifest_filename}.v1.bak`; a missing-write or mismatched existing backup aborts without modifying v1 data. Legacy per-agent JSON is deleted only after its non-empty summaries are safely adopted or confirmed identical; summary conflicts keep the source file and fail loudly.
- **Proactive Markdown migration CLI**: from `backend/`, run `PYTHONPATH=. python scripts/migrate_memory_markdown.py --all-users --dry-run` to audit and omit `--dry-run` to migrate before serving traffic. Use repeated `--user-id` values when selecting exact original identities, especially standalone raw IDs containing `@` or other characters that are normalized in directory names; `--storage-path` selects a non-default DeerMem root. The CLI reuses `FileMemoryStorage.migrate`, is idempotent, continues across per-user failures, and exits non-zero if any user fails. It is optional because the first normal read still performs the same migration automatically.
- A configured `retrieval_adapter` owns indexing and semantic retrieval. File storage sends upsert/remove notifications for normal writes and both explicit and lazy migrations after releasing durable storage locks, then delegates search; without an adapter it declares and uses `substring_fallback`.
- Staleness pass (same LLM invocation as the regular updater, no extra API call): when `staleness_review_enabled` is `true` and at least `staleness_min_candidates` aged facts exist, `_select_stale_candidates` selects facts older than their individual review window (`expected_valid_days`, or the global `staleness_age_days` fallback) that are not in `staleness_protected_categories` (default: `correction`), surfaces them in the prompt with a `valid:Nd` annotation, and the LLM judges each as KEEP, REMOVE, or EXTEND. REMOVE entries go in `staleFactsToRemove`; EXTEND entries go in `staleFactsToExtend` with an `extend_by_days` value, which sets the fact's `expected_valid_days` to `min(days_since_created + extend_by_days, staleness_max_extension_days)`. The LLM assigns `expected_valid_days` when creating a fact; it is clamped at write time to `staleness_age_days × staleness_max_lifetime_multiplier` (creation cap). `_apply_updates` enforces the guardrail unconditionally at apply time: it intersects both the removal and extension sets with `_select_stale_candidates` output before applying the per-cycle cap (`staleness_max_removals_per_cycle`), so protected and non-aged facts can never be targeted regardless of model behavior or the feature flag setting. Facts the LLM proposed for removal are excluded from extension even if the per-cycle cap prevented their actual deletion that cycle. Extensions use an absolute ceiling (`staleness_max_extension_days`) rather than the creation multiplier so a deliberate review decision can advance the window beyond the initial cap while preventing `timedelta` overflow from a malformed `extend_by_days`.
- Consolidation pass (same LLM invocation as the regular updater, no extra API call): when `consolidation_enabled` is `true` and at least one category holds `consolidation_min_facts` or more facts, `_select_consolidation_candidates` identifies fragmented categories and surfaces at most `consolidation_max_groups_per_cycle` of them (largest first) in the prompt. The LLM decides which groups to merge and proposes a synthesised fact per group. `_apply_updates` enforces guardrails: source IDs must exist and must not overlap across groups, group size is capped at `consolidation_max_sources`, the merged fact's confidence cannot exceed the source maximum, and facts below `fact_confidence_threshold` are not written. The merged fact carries the newest source's `createdAt` (so the staleness clock reflects the underlying information, not synthesis time) and inherits `expected_valid_days` set so the merged fact is re-reviewed at the earliest source review deadline (`min(createdAt + effective_lifetime)` across sources, where a source's effective lifetime is its `expected_valid_days` or the global `staleness_age_days` fallback for legacy facts without one - so a legacy source's default window is not swallowed by a long-lived sibling), relative to the merged `createdAt`, clamped to a minimal positive window if a source is already past its deadline, then capped at the creation-time `staleness_max_lifetime_multiplier`; this keeps a volatile or legacy sub-detail from inheriting a stable source's long window and escaping staleness review for years, while a merge of uniformly stable sources does not re-enter review prematurely.
- Next interaction injects selected facts + context into `<memory>` tags in the system prompt when `injection_enabled` is true.
@ -650,7 +657,12 @@ Focused regression coverage for the updater lives in `backend/tests/test_memory_
**Configuration** (`config.yaml` → `memory`):
- `enabled` / `injection_enabled` - Master switches
- `mode` - Operation mode: `middleware` (default passive background extraction) or `tool` (experimental model-driven memory tools). Modes are mutually exclusive.
- `storage_path` - Path to memory.json (absolute path opts out of per-user isolation)
- `storage_path` - DeerMem storage root; one global summary JSON lives under each user and Markdown facts remain under agent buckets
- `storage_class` - `file` or a dotted `MemoryStorage` class; invalid persistent backends fail fast
- `strict_user_scope` - Require `user_id` for all storage access (default `false` for no-auth/legacy compatibility)
- `manifest_filename` - User-global summary JSON filename (kept for configuration compatibility)
- `file_lock_timeout_seconds` - Scope-lock wait; Markdown facts and the recovery journal are required storage invariants rather than configurable modes
- `retrieval_adapter` - Optional dotted factory receiving `DeerMemConfig` and returning a retrieval-port implementation
- `debounce_seconds` - Wait time before processing (default: 30)
- `shutdown_flush_timeout_seconds` - Host-shared hard budget (seconds) to drain the memory backend's pending-update buffer on Gateway graceful shutdown (default: 30; 1–300). Each pending item does one LLM call, so large IM batches may need more. The Gateway lifespan calls `MemoryManager.shutdown_flush(timeout)` after channels/scheduler stop; the backend short-circuits on an idle buffer, so the host calls it unconditionally (no pending/processing gate). Must fit inside the pod's K8s `terminationGracePeriodSeconds` (gateway Helm chart sets this; default 45s) or K8s SIGKILLs the drain mid-flight.
- `model_name` - LLM for updates (null = default model)

7
backend/app/gateway/routers/agents.py
View File

@ -470,6 +470,8 @@ async def delete_agent(name: str) -> None:
if not agent_dir.exists():
outcome = "legacy" if paths.agent_dir(name).exists() else "missing"
return outcome, str(agent_dir)
if not (agent_dir / "config.yaml").is_file():
return "not-custom-agent", str(agent_dir)
shutil.rmtree(agent_dir)
return "deleted", str(agent_dir)
@ -486,5 +488,10 @@ async def delete_agent(name: str) -> None:
)
if outcome == "missing":
raise HTTPException(status_code=404, detail=f"Agent '{name}' not found")
if outcome == "not-custom-agent":
raise HTTPException(
status_code=409,
detail=(f"Directory for '{name}' contains memory data but is not a custom agent because config.yaml is missing; it was preserved."),
)
logger.info(f"Deleted agent '{name}' from {agent_dir}")

86
backend/app/gateway/routers/memory.py
View File

@ -1,12 +1,12 @@
"""Memory API router for retrieving and managing global memory data."""
from typing import Literal
from typing import Any, Literal
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, field_validator
from app.gateway.internal_auth import get_trusted_internal_owner_user_id
from deerflow.agents.memory import get_memory_manager
from deerflow.agents.memory import MemoryConflictError, MemoryCorruptionError, get_memory_manager
from deerflow.config.memory_config import get_memory_config
from deerflow.config.paths import make_safe_user_id
from deerflow.runtime.user_context import get_effective_user_id
@ -65,16 +65,44 @@ class Fact(BaseModel):
id: str = Field(..., description="Unique identifier for the fact")
content: str = Field(..., description="Fact content")
category: str = Field(default="context", description="Fact category")
categoryExtension: str | None = Field(default=None, description="Extension category when category is 'other'")
topics: list[str] | None = Field(default=None, description="Retrieval-oriented topic labels")
confidence: float = Field(default=0.5, description="Confidence score (0-1)")
createdAt: str = Field(default="", description="Creation timestamp")
source: str = Field(default="unknown", description="Source thread ID")
source: str = Field(default="unknown", description="Legacy source string; structured metadata remains internal to storage")
sourceError: str | None = Field(default=None, description="Optional description of the prior mistake or wrong approach")
schemaVersion: int | None = Field(default=None, description="Per-fact schema version")
status: str | None = Field(default=None, description="Fact lifecycle status")
scope: dict[str, str | None] | None = Field(default=None, description="Canonical user/agent scope")
revision: int | None = Field(default=None, description="Fact optimistic revision")
updatedAt: str | None = Field(default=None, description="Last fact update timestamp")
consolidatedAt: str | None = None
consolidatedFrom: list[str] | None = None
@field_validator("source", mode="before")
@classmethod
def _legacy_source_string(cls, value: Any) -> str:
"""Keep the HTTP contract stable while Markdown stores rich metadata."""
if isinstance(value, str):
return value
if not isinstance(value, dict):
return "unknown"
source_type = value.get("type")
thread_id = value.get("threadId")
if source_type == "conversation" and isinstance(thread_id, str) and thread_id:
return thread_id
if isinstance(source_type, str) and source_type:
return source_type
if isinstance(thread_id, str) and thread_id:
return thread_id
return "unknown"
class MemoryResponse(BaseModel):
"""Response model for memory data."""
version: str = Field(default="1.0", description="Memory schema version")
revision: int | None = Field(default=None, description="Manifest revision")
lastUpdated: str = Field(default="", description="Last update timestamp")
user: UserContext = Field(default_factory=UserContext)
history: HistoryContext = Field(default_factory=HistoryContext)
@ -85,11 +113,20 @@ def _map_memory_fact_value_error(exc: ValueError) -> HTTPException:
"""Convert updater validation errors into stable API responses."""
if exc.args and exc.args[0] == "confidence":
detail = "Invalid confidence value; must be between 0 and 1."
elif exc.args and exc.args[0] == "agent_name":
detail = "An agent name is required for fact operations; user-global memory stores summaries only."
else:
detail = "Memory fact content cannot be empty."
return HTTPException(status_code=400, detail=detail)
def _map_memory_manager_error(exc: MemoryConflictError | MemoryCorruptionError) -> HTTPException:
"""Map backend-neutral manager errors without importing a storage plugin."""
if isinstance(exc, MemoryConflictError):
return HTTPException(status_code=409, detail="Memory changed concurrently; reload and retry.")
return HTTPException(status_code=500, detail="Stored memory data is corrupted.")
def _require_capability(name: str, *, label: str):
"""Return a DeerMem-internal capability (bound method) or raise 501.
@ -183,7 +220,10 @@ async def get_memory(http_request: Request) -> MemoryResponse:
}
```
"""
memory_data = get_memory_manager().get_memory(user_id=_resolve_memory_user_id(http_request))
try:
memory_data = get_memory_manager().get_memory(user_id=_resolve_memory_user_id(http_request))
except (MemoryConflictError, MemoryCorruptionError) as exc:
raise _map_memory_manager_error(exc) from exc
return MemoryResponse(**memory_data)
@ -205,14 +245,17 @@ async def reload_memory(http_request: Request) -> MemoryResponse:
"""
user_id = _resolve_memory_user_id(http_request)
manager = get_memory_manager()
if hasattr(manager, "reload_memory"):
memory_data = manager.reload_memory(user_id=user_id)
else:
# Non-DeerMem backends have no reload concept; return current memory.
# (Asymmetry vs fact CRUD, which raises 501 when unsupported: reload is a
# read-only refresh, so degrading to get_memory is safe and still useful;
# silently no-op'ing a write would hide data loss, so writes fail loud.)
memory_data = manager.get_memory(user_id=user_id)
try:
if hasattr(manager, "reload_memory"):
memory_data = manager.reload_memory(user_id=user_id)
else:
# Non-DeerMem backends have no reload concept; return current memory.
# (Asymmetry vs fact CRUD, which raises 501 when unsupported: reload is a
# read-only refresh, so degrading to get_memory is safe and still useful;
# silently no-op'ing a write would hide data loss, so writes fail loud.)
memory_data = manager.get_memory(user_id=user_id)
except (MemoryConflictError, MemoryCorruptionError) as exc:
raise _map_memory_manager_error(exc) from exc
return MemoryResponse(**memory_data)
@ -227,6 +270,8 @@ async def clear_memory(http_request: Request) -> MemoryResponse:
"""Clear all persisted memory data."""
try:
memory_data = get_memory_manager().clear_memory(user_id=_resolve_memory_user_id(http_request))
except (MemoryConflictError, MemoryCorruptionError) as exc:
raise _map_memory_manager_error(exc) from exc
except OSError as exc:
raise HTTPException(status_code=500, detail="Failed to clear memory data.") from exc
@ -252,6 +297,8 @@ async def create_memory_fact_endpoint(request: FactCreateRequest, http_request:
)
except ValueError as exc:
raise _map_memory_fact_value_error(exc) from exc
except (MemoryConflictError, MemoryCorruptionError) as exc:
raise _map_memory_manager_error(exc) from exc
except OSError as exc:
raise HTTPException(status_code=500, detail="Failed to create memory fact.") from exc
@ -275,6 +322,8 @@ async def delete_memory_fact_endpoint(fact_id: str, http_request: Request) -> Me
memory_data = delete_fact(fact_id, user_id=_resolve_memory_user_id(http_request))
except KeyError as exc:
raise HTTPException(status_code=404, detail=f"Memory fact '{fact_id}' not found.") from exc
except (MemoryConflictError, MemoryCorruptionError) as exc:
raise _map_memory_manager_error(exc) from exc
except OSError as exc:
raise HTTPException(status_code=500, detail="Failed to delete memory fact.") from exc
@ -303,6 +352,8 @@ async def update_memory_fact_endpoint(fact_id: str, request: FactPatchRequest, h
raise _map_memory_fact_value_error(exc) from exc
except KeyError as exc:
raise HTTPException(status_code=404, detail=f"Memory fact '{fact_id}' not found.") from exc
except (MemoryConflictError, MemoryCorruptionError) as exc:
raise _map_memory_manager_error(exc) from exc
except OSError as exc:
raise HTTPException(status_code=500, detail="Failed to update memory fact.") from exc
@ -318,7 +369,10 @@ async def update_memory_fact_endpoint(fact_id: str, request: FactPatchRequest, h
)
async def export_memory(http_request: Request) -> MemoryResponse:
"""Export the current memory data."""
memory_data = get_memory_manager().get_memory(user_id=_resolve_memory_user_id(http_request))
try:
memory_data = get_memory_manager().get_memory(user_id=_resolve_memory_user_id(http_request))
except (MemoryConflictError, MemoryCorruptionError) as exc:
raise _map_memory_manager_error(exc) from exc
return MemoryResponse(**memory_data)
@ -332,7 +386,9 @@ async def export_memory(http_request: Request) -> MemoryResponse:
async def import_memory(request: MemoryResponse, http_request: Request) -> MemoryResponse:
"""Import and persist memory data."""
try:
memory_data = get_memory_manager().import_memory(request.model_dump(), user_id=_resolve_memory_user_id(http_request))
memory_data = get_memory_manager().import_memory(request.model_dump(exclude_none=True), user_id=_resolve_memory_user_id(http_request))
except (MemoryConflictError, MemoryCorruptionError) as exc:
raise _map_memory_manager_error(exc) from exc
except OSError as exc:
raise HTTPException(status_code=500, detail="Failed to import memory data.") from exc

6
backend/packages/harness/deerflow/agents/memory/__init__.py
View File

@ -13,13 +13,19 @@ them directly from ``deerflow.agents.memory.backends.deermem.deermem.core.*``.
"""
from deerflow.agents.memory.manager import (
MemoryConflictError,
MemoryCorruptionError,
MemoryManager,
MemoryManagerError,
get_memory_manager,
reset_memory_manager,
)
__all__ = [
"MemoryManager",
"MemoryManagerError",
"MemoryConflictError",
"MemoryCorruptionError",
"get_memory_manager",
"reset_memory_manager",
]

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

@ -23,10 +23,11 @@ never breaks those modules at import time (see MemoryManager plan, step 8).
from __future__ import annotations
import copy
import logging
from typing import Any
from deerflow.agents.memory.manager import MemoryManager
from deerflow.agents.memory.manager import MemoryConflictError, MemoryCorruptionError, MemoryManager
from .deermem.config import DeerMemConfig
from .deermem.core.llm import build_llm
@ -36,14 +37,56 @@ from .deermem.core.message_processing import (
filter_messages_for_memory,
load_patterns,
)
from .deermem.core.paths import DEFAULT_AGENT_BUCKET
from .deermem.core.prompt import format_memory_for_injection, load_prompt, load_prompt_messages, warm_tiktoken_cache
from .deermem.core.queue import MemoryUpdateQueue
from .deermem.core.storage import create_storage
from .deermem.core.storage import MemoryRevisionConflict, MemoryStorageCorruption, create_storage
from .deermem.core.updater import MemoryUpdater, _coerce_source_confidence
logger = logging.getLogger(__name__)
def _resolve_agent_name(agent_name: str | None) -> str:
"""Return DeerFlow's case-insensitive canonical agent identifier."""
return agent_name.lower() if agent_name is not None else DEFAULT_AGENT_BUCKET
def _call_backend(operation):
"""Translate DeerMem-private storage errors into the public manager contract."""
try:
return operation()
except MemoryRevisionConflict as exc:
raise MemoryConflictError(str(exc)) from exc
except MemoryStorageCorruption as exc:
raise MemoryCorruptionError(str(exc)) from exc
def _legacy_source_value(source: Any) -> str:
"""Project structured source metadata back to the legacy public string."""
if isinstance(source, str):
return source
if not isinstance(source, dict):
return "unknown"
source_type = source.get("type")
thread_id = source.get("threadId")
if source_type == "conversation" and isinstance(thread_id, str) and thread_id:
return thread_id
if isinstance(source_type, str) and source_type:
return source_type
if isinstance(thread_id, str) and thread_id:
return thread_id
return "unknown"
def _compat_document(memory_data: dict[str, Any]) -> dict[str, Any]:
"""Return the historical Manager/API shape without changing persistence."""
result = copy.deepcopy(memory_data)
for fact in result.get("facts", []):
if isinstance(fact, dict):
fact["source"] = _legacy_source_value(fact.get("source"))
return result
class DeerMem(MemoryManager):
"""Default memory backend: file-backed facts + debounced LLM extraction."""
@ -109,7 +152,7 @@ class DeerMem(MemoryManager):
self._queue.add(
thread_id=thread_id,
messages=filtered,
agent_name=agent_name,
agent_name=_resolve_agent_name(agent_name),
user_id=user_id,
trace_id=trace_id,
correction_detected=correction_detected,
@ -136,7 +179,7 @@ class DeerMem(MemoryManager):
self._queue.add_nowait(
thread_id=thread_id,
messages=filtered,
agent_name=agent_name,
agent_name=_resolve_agent_name(agent_name),
user_id=user_id,
correction_detected=correction_detected,
reinforcement_detected=reinforcement_detected,
@ -179,7 +222,7 @@ class DeerMem(MemoryManager):
``injection_enabled`` gate and the ``<memory>`` wrapping stay at the
call site (``_get_memory_context``); this returns only the body.
"""
memory_data = self._updater.get_memory_data(agent_name=agent_name, user_id=user_id)
memory_data = _call_backend(lambda: self._updater.get_memory_data(agent_name=_resolve_agent_name(agent_name), user_id=user_id))
return format_memory_for_injection(
memory_data,
max_tokens=self._config.max_injection_tokens,
@ -210,10 +253,26 @@ class DeerMem(MemoryManager):
if not query or not query.strip() or top_k <= 0:
return []
query_lower = query.strip().lower()
memory_data = self._updater.get_memory_data(agent_name=agent_name, user_id=user_id)
search_facts = getattr(self._storage, "search_facts", None)
resolved_agent_name = _resolve_agent_name(agent_name)
scopes = [{"userId": user_id, "agentName": resolved_agent_name}]
indexed = (
search_facts(
query,
scopes=scopes,
top_k=top_k,
mode="hybrid",
filters={"category": category} if category else None,
)
if callable(search_facts)
else []
)
if indexed:
return [_compat_document({"facts": [result.get("fact", result)]})["facts"][0] for result in indexed]
memory_data = _call_backend(lambda: self._updater.get_memory_data(agent_name=resolved_agent_name, user_id=user_id))
matched = [fact for fact in memory_data.get("facts", []) if isinstance(fact.get("content"), str) and query_lower in fact["content"].lower() and (category is None or fact.get("category") == category)]
matched.sort(key=_coerce_source_confidence, reverse=True)
return matched[:top_k]
return _compat_document({"facts": matched[:top_k]})["facts"]
# ── Manage ───────────────────────────────────────────────────────────
def get_memory(
@ -222,7 +281,8 @@ class DeerMem(MemoryManager):
user_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
return self._updater.get_memory_data(agent_name=agent_name, user_id=user_id)
memory_data = _call_backend(lambda: self._updater.get_memory_data(agent_name=_resolve_agent_name(agent_name), user_id=user_id))
return _compat_document(memory_data)
def delete_memory(
self,
@ -239,7 +299,11 @@ class DeerMem(MemoryManager):
user_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
return self._updater.clear_memory_data(agent_name=agent_name, user_id=user_id)
if agent_name is None:
memory_data = _call_backend(lambda: self._updater.clear_all_memory_data(user_id=user_id))
else:
memory_data = _call_backend(lambda: self._updater.clear_memory_data(agent_name=_resolve_agent_name(agent_name), user_id=user_id))
return _compat_document(memory_data)
def import_memory(
self,
@ -248,7 +312,14 @@ class DeerMem(MemoryManager):
user_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
return self._updater.import_memory_data(memory_data, agent_name=agent_name, user_id=user_id)
imported = _call_backend(
lambda: self._updater.import_memory_data(
memory_data,
agent_name=_resolve_agent_name(agent_name),
user_id=user_id,
)
)
return _compat_document(imported)
def export_memory(
self,
@ -295,7 +366,13 @@ class DeerMem(MemoryManager):
agent_name: str | None = None,
) -> dict[str, Any]:
"""Drop the cached memory document and reload from disk."""
return self._updater.reload_memory_data(agent_name=agent_name, user_id=user_id)
memory_data = _call_backend(
lambda: self._updater.reload_memory_data(
agent_name=_resolve_agent_name(agent_name),
user_id=user_id,
)
)
return _compat_document(memory_data)
def create_fact(
self,
@ -306,13 +383,16 @@ class DeerMem(MemoryManager):
agent_name: str | None = None,
user_id: str | None = None,
) -> tuple[dict[str, Any], str | None]:
return self._updater.create_memory_fact(
content,
category=category,
confidence=confidence,
agent_name=agent_name,
user_id=user_id,
memory_data, fact_id = _call_backend(
lambda: self._updater.create_memory_fact(
content,
category=category,
confidence=confidence,
agent_name=_resolve_agent_name(agent_name),
user_id=user_id,
)
)
return _compat_document(memory_data), fact_id
def delete_fact(
self,
@ -321,7 +401,14 @@ class DeerMem(MemoryManager):
agent_name: str | None = None,
user_id: str | None = None,
) -> dict[str, Any]:
return self._updater.delete_memory_fact(fact_id, agent_name=agent_name, user_id=user_id)
memory_data = _call_backend(
lambda: self._updater.delete_memory_fact(
fact_id,
agent_name=_resolve_agent_name(agent_name),
user_id=user_id,
)
)
return _compat_document(memory_data)
def update_fact(
self,
@ -333,11 +420,14 @@ class DeerMem(MemoryManager):
agent_name: str | None = None,
user_id: str | None = None,
) -> dict[str, Any]:
return self._updater.update_memory_fact(
fact_id,
content=content,
category=category,
confidence=confidence,
agent_name=agent_name,
user_id=user_id,
memory_data = _call_backend(
lambda: self._updater.update_memory_fact(
fact_id,
content=content,
category=category,
confidence=confidence,
agent_name=_resolve_agent_name(agent_name),
user_id=user_id,
)
)
return _compat_document(memory_data)

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

@ -52,6 +52,24 @@ class DeerMemConfig(BaseModel):
default="",
description="Dotted class path for an alternative storage provider; empty (default) = FileMemoryStorage (no importlib, portable).",
)
strict_user_scope: bool = Field(
default=False,
description="Require user_id for every storage scope. False preserves no-auth and legacy callers.",
)
manifest_filename: str = Field(
default="memory.json",
description="User-global summary JSON filename. Kept under this name for config compatibility; must be a plain .json filename.",
)
file_lock_timeout_seconds: int = Field(
default=10,
ge=1,
le=120,
description="Maximum wait for the per-scope cross-process advisory file lock.",
)
retrieval_adapter: str = Field(
default="",
description="Optional dotted retrieval-adapter factory. It receives DeerMemConfig and must implement RetrievalPort.",
)
# ── Queue ────────────────────────────────────────────────────────────
debounce_seconds: int = Field(
default=30,
@ -254,6 +272,7 @@ class DeerMemConfig(BaseModel):
"""
if not backend_config:
return cls()
backend_config = dict(backend_config)
known = {k: v for k, v in backend_config.items() if k in cls.model_fields and v is not None}
unknown = sorted(k for k in backend_config if k not in cls.model_fields)
if unknown:

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

@ -2,9 +2,9 @@
The host no longer dictates where DeerMem stores data. Root = ``config.storage_path``
(if set, absolute or relative) or ``$DEERMEM_DATA_DIR`` or ``~/.deermem/``.
Per-user / per-agent / legacy layouts live under the root, mirroring the
pre-abstraction paths so a one-time data migration (old ``{base_dir}/users/*``
-> DeerMem root) is a plain move.
Each user has one global ``memory.json`` for project-independent summaries.
Agent-specific facts live below ``agents/{agent_name}/facts`` and never add a
fact index to that JSON document.
user_id is sanitized in-process (``[A-Za-z0-9_-]`` + SHA-256 digest for lossy
ids) and agent_name validated against an inlined pattern -- DeerMem does not
@ -31,6 +31,9 @@ _SAFE_USER_ID_DIGEST_HEX_LEN = 16
# agent_name validation (inlined; was deer-flow's AGENT_NAME_PATTERN).
AGENT_NAME_PATTERN = re.compile(r"^[A-Za-z0-9-]+$")
# Internal bucket used when callers omit ``agent_name``. The underscores keep
# it outside the public custom-agent namespace accepted by AGENT_NAME_PATTERN.
DEFAULT_AGENT_BUCKET = "__default__"
def safe_user_id(raw: str) -> str:
@ -54,7 +57,7 @@ def validate_agent_name(name: str) -> None:
"""Validate that the agent name is safe to use in filesystem paths."""
if not name:
raise ValueError("Agent name must be a non-empty string.")
if not AGENT_NAME_PATTERN.match(name):
if name != DEFAULT_AGENT_BUCKET and not AGENT_NAME_PATTERN.match(name):
raise ValueError(f"Invalid agent name {name!r}: names must match {AGENT_NAME_PATTERN.pattern}")
@ -74,22 +77,40 @@ def memory_file_path(
) -> Path:
"""Resolve the memory file path under DeerMem's own data root.
``config.storage_path`` (absolute or relative) is the root; per-user /
per-agent / legacy layouts live under it. Empty -> default root
``config.storage_path`` (absolute or relative) is the root. Empty -> default root
(``$DEERMEM_DATA_DIR`` / ``~/.deermem/``). The host (deer-flow factory)
injects an absolute base_dir as ``storage_path`` so memory lands at
``{base_dir}/users/{user_id}/memory.json`` (CWD-independent).
"""
root = Path(config.storage_path) if config.storage_path else _default_root()
if config.strict_user_scope and user_id is None:
raise ValueError("user_id is required when strict_user_scope is enabled.")
manifest_filename = config.manifest_filename
if Path(manifest_filename).name != manifest_filename or not manifest_filename.endswith(".json"):
raise ValueError("manifest_filename must be a plain .json filename.")
if user_id is not None:
uid = safe_user_id(user_id)
if agent_name is not None:
validate_agent_name(agent_name)
return root / "users" / uid / "agents" / agent_name.lower() / "memory.json"
return root / "users" / uid / "memory.json"
bucket = root / "users" / uid
return bucket / manifest_filename
# Legacy: no user_id
if agent_name is not None:
validate_agent_name(agent_name)
return root / "agents" / agent_name.lower() / "memory.json"
return root / "memory.json"
bucket = root
return bucket / manifest_filename
def agent_facts_directory(memory_path: Path, agent_name: str) -> Path:
"""Return the fact root for one required agent below a user's memory file."""
validate_agent_name(agent_name)
return memory_path.parent / "agents" / agent_name.lower() / "facts"
def fact_file_path(memory_path: Path, fact_id: str, *, agent_name: str) -> Path:
"""Return the sharded Markdown path for one agent-owned fact."""
if not fact_id or not re.fullmatch(r"[A-Za-z0-9_-]+", fact_id):
raise ValueError("Fact id may contain only letters, numbers, '_' and '-'.")
prefix = hashlib.sha256(fact_id.encode("utf-8")).hexdigest()[:2]
return agent_facts_directory(memory_path, agent_name) / prefix / f"{fact_id}.md"

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

File diff suppressed because it is too large Load Diff

256
backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/updater.py
View File

@ -20,6 +20,7 @@ from .prompt import (
load_prompt_messages,
)
from .storage import (
MemoryManifestRevisionConflict,
MemoryStorage,
create_empty_memory,
utc_now_iso_z,
@ -630,9 +631,19 @@ class MemoryUpdater:
# ── Data access + fact CRUD (formerly module-level functions; use self._storage) ──
def _save_memory_to_file(self, memory_data: dict[str, Any], agent_name: str | None = None, *, user_id: str | None = None) -> bool:
def _save_memory_to_file(
self,
memory_data: dict[str, Any],
agent_name: str | None = None,
*,
user_id: str | None = None,
expected_revision: int | None = None,
) -> bool:
"""Persist memory data via the injected storage."""
return self._storage.save(memory_data, agent_name, user_id=user_id)
kwargs: dict[str, Any] = {"user_id": user_id}
if expected_revision is not None:
kwargs["expected_revision"] = expected_revision
return self._storage.save(memory_data, agent_name, **kwargs)
def get_memory_data(self, agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
"""Get the current memory data via the injected storage."""
@ -644,14 +655,90 @@ class MemoryUpdater:
def import_memory_data(self, memory_data: dict[str, Any], agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
"""Persist imported memory data via the injected storage."""
if not isinstance(memory_data, dict):
raise ValueError("memory_data")
memory_data = copy.deepcopy(memory_data)
empty = create_empty_memory()
for section in ("user", "history"):
incoming_section = memory_data.get(section, {})
if not isinstance(incoming_section, dict):
raise ValueError(f"memory_data.{section}")
complete_section = copy.deepcopy(empty[section])
for key, value in incoming_section.items():
if key in complete_section and isinstance(complete_section[key], dict) and isinstance(value, dict):
complete_section[key].update(copy.deepcopy(value))
else:
complete_section[key] = copy.deepcopy(value)
memory_data[section] = complete_section
if agent_name is not None and getattr(type(self._storage), "apply_changes", None) is not MemoryStorage.apply_changes:
current = self.get_memory_data(agent_name, user_id=user_id)
incoming_facts = copy.deepcopy(memory_data.get("facts", []))
if not isinstance(incoming_facts, list) or any(not isinstance(fact, dict) for fact in incoming_facts):
raise ValueError("memory_data.facts")
for fact in incoming_facts:
fact["id"] = str(fact.get("id") or f"fact_{uuid.uuid4().hex}")
fact["confidence"] = _coerce_source_confidence(fact)
current_by_id = {str(fact.get("id")): fact for fact in current.get("facts", []) if isinstance(fact, dict)}
incoming_ids = {str(fact.get("id")) for fact in incoming_facts}
self._storage.apply_changes(
{
"upserts": incoming_facts,
"upsertRevisions": {str(fact.get("id")): (int(current_by_id[str(fact.get("id"))].get("revision") or 1) if str(fact.get("id")) in current_by_id else None) for fact in incoming_facts},
"deletes": [fact_id for fact_id in current_by_id if fact_id not in incoming_ids],
"deleteRevisions": {fact_id: int(fact.get("revision") or 1) for fact_id, fact in current_by_id.items() if fact_id not in incoming_ids},
"summaries": {"user": copy.deepcopy(memory_data.get("user", {})), "history": copy.deepcopy(memory_data.get("history", {}))},
},
agent_name=agent_name,
user_id=user_id,
expected_manifest_revision=int(current.get("revision") or 0),
)
return self._storage.load(agent_name, user_id=user_id)
if agent_name is None:
memory_data["facts"] = []
if not self._storage.save(memory_data, agent_name, user_id=user_id):
raise OSError("Failed to save imported memory data")
return self._storage.load(agent_name, user_id=user_id)
def clear_memory_data(self, agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
"""Clear all stored memory data and persist an empty structure."""
"""Clear one selected agent's facts without resetting shared summaries."""
if agent_name is not None and getattr(type(self._storage), "apply_changes", None) is not MemoryStorage.apply_changes:
for attempt in range(3):
current = self.get_memory_data(agent_name, user_id=user_id) if attempt == 0 else self.reload_memory_data(agent_name, user_id=user_id)
facts = [fact for fact in current.get("facts", []) if isinstance(fact, dict)]
try:
self._storage.apply_changes(
{
"deletes": [str(fact.get("id")) for fact in facts],
"deleteRevisions": {str(fact.get("id")): int(fact.get("revision") or 1) for fact in facts},
},
agent_name=agent_name,
user_id=user_id,
expected_manifest_revision=int(current.get("revision") or 0),
)
return self.reload_memory_data(agent_name, user_id=user_id)
except MemoryManifestRevisionConflict:
if attempt == 2:
raise
logger.info("Retrying scoped memory clear from a fresh snapshot after a revision conflict")
raise AssertionError("bounded scoped-clear retry did not return or raise")
current = self.get_memory_data(agent_name, user_id=user_id)
cleared_memory = copy.deepcopy(current)
cleared_memory["facts"] = []
if not self._save_memory_to_file(cleared_memory, agent_name, user_id=user_id, expected_revision=int(current.get("revision") or 0)):
raise OSError("Failed to save cleared memory data")
return cleared_memory
def clear_all_memory_data(self, *, user_id: str | None = None) -> dict[str, Any]:
"""Clear global summaries and every agent fact bucket for one user."""
if getattr(type(self._storage), "clear_all", None) is not MemoryStorage.clear_all:
return self._storage.clear_all(user_id=user_id)
current = self.get_memory_data(user_id=user_id)
cleared_memory = create_empty_memory()
if not self._save_memory_to_file(cleared_memory, agent_name, user_id=user_id):
if not self._save_memory_to_file(
cleared_memory,
user_id=user_id,
expected_revision=int(current.get("revision") or 0),
):
raise OSError("Failed to save cleared memory data")
return cleared_memory
@ -671,28 +758,54 @@ class MemoryUpdater:
existence check (upstream's ``create_memory_fact_with_created_fact``),
which the vendored copy had dropped together to avoid the dangling id.
"""
if agent_name is None:
raise ValueError("agent_name")
normalized_content = content.strip()
if not normalized_content:
raise ValueError("content")
normalized_category = category.strip() or "context"
validated_confidence = _validate_confidence(confidence)
now = utc_now_iso_z()
fact_id = f"fact_{uuid.uuid4().hex[:8]}"
candidate = {
"id": fact_id,
"content": normalized_content,
"category": normalized_category,
"confidence": validated_confidence,
"createdAt": now,
"source": "manual",
}
if getattr(type(self._storage), "apply_changes", None) is not MemoryStorage.apply_changes:
for attempt in range(3):
memory_data = self.get_memory_data(agent_name, user_id=user_id) if attempt == 0 else self.reload_memory_data(agent_name, user_id=user_id)
updated_memory = dict(memory_data)
updated_memory["facts"] = _trim_facts_to_max([*memory_data.get("facts", []), copy.deepcopy(candidate)], self._config.max_facts)
kept_ids = {str(fact.get("id")) for fact in updated_memory["facts"]}
deletions = [str(fact.get("id")) for fact in memory_data.get("facts", []) if str(fact.get("id")) not in kept_ids]
try:
self._storage.apply_changes(
{
"upserts": [fact for fact in updated_memory["facts"] if fact.get("id") == fact_id],
"upsertRevisions": {fact_id: None},
"deletes": deletions,
"deleteRevisions": {str(fact.get("id")): int(fact.get("revision") or 1) for fact in memory_data.get("facts", []) if str(fact.get("id")) in deletions},
},
agent_name=agent_name,
user_id=user_id,
expected_manifest_revision=int(memory_data.get("revision") or 0),
)
fresh_memory = self.reload_memory_data(agent_name, user_id=user_id)
stored = any(fact.get("id") == fact_id for fact in fresh_memory.get("facts", []))
return fresh_memory, (fact_id if stored else None)
except MemoryManifestRevisionConflict:
if attempt == 2:
raise
logger.info("Retrying capped fact creation from a fresh snapshot after a revision conflict")
raise AssertionError("bounded create retry did not return or raise")
memory_data = self.get_memory_data(agent_name, user_id=user_id)
updated_memory = dict(memory_data)
facts = list(memory_data.get("facts", []))
fact_id = f"fact_{uuid.uuid4().hex[:8]}"
facts.append(
{
"id": fact_id,
"content": normalized_content,
"category": normalized_category,
"confidence": validated_confidence,
"createdAt": now,
"source": "manual",
}
)
updated_memory["facts"] = _trim_facts_to_max(facts, self._config.max_facts)
if not self._save_memory_to_file(updated_memory, agent_name, user_id=user_id):
updated_memory["facts"] = _trim_facts_to_max([*memory_data.get("facts", []), candidate], self._config.max_facts)
if not self._save_memory_to_file(updated_memory, agent_name, user_id=user_id, expected_revision=int(memory_data.get("revision") or 0)):
raise OSError("Failed to save memory data after creating fact")
# If the cap evicted the just-added (lower-confidence) fact, signal via
# None so callers don't report a dangling id as "added".
@ -701,19 +814,68 @@ class MemoryUpdater:
def delete_memory_fact(self, fact_id: str, agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
"""Delete a fact by its id and persist the updated memory data."""
if agent_name is None:
raise ValueError("agent_name")
if getattr(type(self._storage), "apply_changes", None) is not MemoryStorage.apply_changes and hasattr(self._storage, "get_fact"):
deleted = self._storage.get_fact(fact_id, agent_name=agent_name, user_id=user_id)
if deleted is None:
raise KeyError(fact_id)
global_memory = self.get_memory_data(user_id=user_id)
self._storage.apply_changes(
{"deletes": [fact_id], "deleteRevisions": {fact_id: int(deleted.get("revision") or 1)}},
agent_name=agent_name,
user_id=user_id,
expected_manifest_revision=int(global_memory.get("revision") or 0),
allow_manifest_rebase=True,
)
return self.get_memory_data(agent_name, user_id=user_id)
memory_data = self.get_memory_data(agent_name, user_id=user_id)
facts = memory_data.get("facts", [])
updated_facts = [fact for fact in facts if fact.get("id") != fact_id]
if len(updated_facts) == len(facts):
raise KeyError(fact_id)
deleted = next(fact for fact in facts if fact.get("id") == fact_id)
if getattr(type(self._storage), "apply_changes", None) is not MemoryStorage.apply_changes:
self._storage.apply_changes(
{"deletes": [fact_id], "deleteRevisions": {fact_id: int(deleted.get("revision") or 1)}},
agent_name=agent_name,
user_id=user_id,
expected_manifest_revision=int(memory_data.get("revision") or 0),
allow_manifest_rebase=True,
)
return self.get_memory_data(agent_name, user_id=user_id)
updated_memory = dict(memory_data)
updated_memory["facts"] = updated_facts
if not self._save_memory_to_file(updated_memory, agent_name, user_id=user_id):
if not self._save_memory_to_file(updated_memory, agent_name, user_id=user_id, expected_revision=int(memory_data.get("revision") or 0)):
raise OSError(f"Failed to save memory data after deleting fact '{fact_id}'")
return updated_memory
def update_memory_fact(self, fact_id: str, content: str | None = None, category: str | None = None, confidence: float | None = None, agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
"""Update an existing fact and persist the updated memory data."""
if agent_name is None:
raise ValueError("agent_name")
if getattr(type(self._storage), "apply_changes", None) is not MemoryStorage.apply_changes and hasattr(self._storage, "get_fact"):
updated_fact = self._storage.get_fact(fact_id, agent_name=agent_name, user_id=user_id)
if updated_fact is None:
raise KeyError(fact_id)
if content is not None:
normalized_content = content.strip()
if not normalized_content:
raise ValueError("content")
updated_fact["content"] = normalized_content
if category is not None:
updated_fact["category"] = category.strip() or "context"
if confidence is not None:
updated_fact["confidence"] = _validate_confidence(confidence)
global_memory = self.get_memory_data(user_id=user_id)
self._storage.apply_changes(
{"upserts": [updated_fact], "upsertRevisions": {fact_id: int(updated_fact.get("revision") or 1)}},
agent_name=agent_name,
user_id=user_id,
expected_manifest_revision=int(global_memory.get("revision") or 0),
allow_manifest_rebase=True,
)
return self.get_memory_data(agent_name, user_id=user_id)
memory_data = self.get_memory_data(agent_name, user_id=user_id)
updated_memory = dict(memory_data)
updated_facts: list[dict[str, Any]] = []
@ -736,8 +898,18 @@ class MemoryUpdater:
updated_facts.append(fact)
if not found:
raise KeyError(fact_id)
if getattr(type(self._storage), "apply_changes", None) is not MemoryStorage.apply_changes:
changed = next(fact for fact in updated_facts if fact.get("id") == fact_id)
self._storage.apply_changes(
{"upserts": [changed], "upsertRevisions": {fact_id: int(changed.get("revision") or 1)}},
agent_name=agent_name,
user_id=user_id,
expected_manifest_revision=int(memory_data.get("revision") or 0),
allow_manifest_rebase=True,
)
return self.get_memory_data(agent_name, user_id=user_id)
updated_memory["facts"] = updated_facts
if not self._save_memory_to_file(updated_memory, agent_name, user_id=user_id):
if not self._save_memory_to_file(updated_memory, agent_name, user_id=user_id, expected_revision=int(memory_data.get("revision") or 0)):
raise OSError(f"Failed to save memory data after updating fact '{fact_id}'")
return updated_memory
@ -829,11 +1001,53 @@ class MemoryUpdater:
) -> bool:
"""Parse the model response, apply updates, and persist memory."""
update_data = _parse_memory_update_response(response_content)
if getattr(type(self._storage), "apply_changes", None) is not MemoryStorage.apply_changes:
for attempt in range(3):
# Deep-copy before in-place mutation so a failed commit cannot
# corrupt the cached snapshot. On a manifest conflict the
# complete extraction result is reapplied to a fresh document;
# its trim/consolidation/delete decisions are snapshot-wide and
# must never be replayed as disjoint point writes.
updated_memory = self._apply_updates(copy.deepcopy(current_memory), update_data, thread_id)
updated_memory = _strip_upload_mentions_from_memory(updated_memory)
current_by_id = {str(fact.get("id")): fact for fact in current_memory.get("facts", [])}
updated_by_id = {str(fact.get("id")): fact for fact in updated_memory.get("facts", [])}
change_set = {
"upserts": [copy.deepcopy(fact) for fact_id, fact in updated_by_id.items() if current_by_id.get(fact_id) != fact],
"upsertRevisions": {fact_id: (int(current_by_id[fact_id].get("revision") or 1) if fact_id in current_by_id else None) for fact_id, fact in updated_by_id.items() if current_by_id.get(fact_id) != fact},
"deletes": [fact_id for fact_id in current_by_id if fact_id not in updated_by_id],
"deleteRevisions": {fact_id: int(current_by_id[fact_id].get("revision") or 1) for fact_id in current_by_id if fact_id not in updated_by_id},
}
summaries_changed = updated_memory.get("user", {}) != current_memory.get("user", {}) or updated_memory.get("history", {}) != current_memory.get("history", {})
if summaries_changed:
change_set["summaries"] = {
"user": copy.deepcopy(updated_memory.get("user", {})),
"history": copy.deepcopy(updated_memory.get("history", {})),
}
try:
self._storage.apply_changes(
change_set,
agent_name=agent_name,
user_id=user_id,
expected_manifest_revision=int(current_memory.get("revision") or 0),
)
return True
except MemoryManifestRevisionConflict:
if attempt == 2:
raise
current_memory = self.reload_memory_data(agent_name, user_id=user_id)
logger.info("Retrying extracted memory update from a fresh snapshot after a revision conflict")
raise AssertionError("bounded extracted-update retry did not return or raise")
# Deep-copy before in-place mutation so a subsequent save() failure
# cannot corrupt the still-cached original object reference.
updated_memory = self._apply_updates(copy.deepcopy(current_memory), update_data, thread_id)
updated_memory = _strip_upload_mentions_from_memory(updated_memory)
return self._storage.save(updated_memory, agent_name, user_id=user_id)
return self._storage.save(
updated_memory,
agent_name,
user_id=user_id,
expected_revision=int(current_memory.get("revision") or 0),
)
async def aupdate_memory(
self,

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

@ -41,6 +41,18 @@ _backends_cache: dict[str, type[MemoryManager]] | None = None
_manager_lock = threading.Lock()
class MemoryManagerError(RuntimeError):
"""Backend-neutral base error exposed at the MemoryManager boundary."""
class MemoryConflictError(MemoryManagerError):
"""The requested write lost an optimistic-concurrency race."""
class MemoryCorruptionError(MemoryManagerError):
"""Persisted memory cannot be read safely."""
class MemoryManager(ABC):
"""Backend-neutral memory manager contract (9 methods).
@ -167,7 +179,12 @@ class MemoryManager(ABC):
user_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
"""Clear the bucket's memory; return the cleared (now-empty) document."""
"""Clear memory and return the empty compatibility document.
``agent_name=None`` means all memory owned by the user. An explicit
agent name clears only that agent's memory and must preserve shared
user-level summaries.
"""
@abstractmethod
def import_memory(

137
backend/scripts/migrate_memory_markdown.py Normal file
View File

@ -0,0 +1,137 @@
"""Proactively migrate legacy global JSON facts into Markdown fact files.
Normal DeerMem reads already perform this migration lazily. This CLI lets an
operator preview or complete the same idempotent migration before serving
traffic, which is useful for multi-user upgrade audits.
Usage from ``backend/``::
PYTHONPATH=. python scripts/migrate_memory_markdown.py --all-users --dry-run
PYTHONPATH=. python scripts/migrate_memory_markdown.py --all-users
PYTHONPATH=. python scripts/migrate_memory_markdown.py --user-id alice
"""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from typing import Any
from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemConfig
from deerflow.agents.memory.backends.deermem.deermem.core.paths import DEFAULT_AGENT_BUCKET, memory_file_path
from deerflow.agents.memory.backends.deermem.deermem.core.storage import DOCUMENT_VERSION, FileMemoryStorage
from deerflow.config.runtime_paths import runtime_home
def discover_user_ids(storage_path: Path) -> list[str]:
"""Return directory-safe user IDs found below one DeerMem storage root."""
users_root = storage_path / "users"
if not users_root.is_dir():
return []
return sorted(entry.name for entry in users_root.iterdir() if entry.is_dir())
def _inspect_legacy_global_json(config: DeerMemConfig, user_id: str) -> tuple[Path, bool]:
path = memory_file_path(config, user_id=user_id)
if not path.exists():
return path, False
try:
document = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
raise ValueError(f"cannot parse {path}: {exc}") from exc
if not isinstance(document, dict):
raise ValueError(f"{path} must contain a JSON object")
return path, "facts" in document or document.get("version") != DOCUMENT_VERSION
def migrate_users(
config: DeerMemConfig,
user_ids: list[str],
*,
dry_run: bool = False,
) -> list[dict[str, Any]]:
"""Migrate selected users independently and return an audit-friendly report."""
storage = FileMemoryStorage(config)
report: list[dict[str, Any]] = []
for user_id in user_ids:
entry: dict[str, Any] = {"user_id": user_id, "status": "current", "path": "", "error": None}
try:
path, needs_migration = _inspect_legacy_global_json(config, user_id)
entry["path"] = str(path)
if not needs_migration:
report.append(entry)
continue
if dry_run:
entry["status"] = "planned"
else:
result = storage.migrate(user_id=user_id, agent_name=DEFAULT_AGENT_BUCKET)
entry["status"] = "migrated" if result.get("migrated") else "current"
entry["from_version"] = result.get("fromVersion")
entry["to_version"] = result.get("toVersion")
except Exception as exc: # noqa: BLE001 - one bad user must not hide the rest of the audit
entry["status"] = "failed"
entry["error"] = str(exc)
report.append(entry)
return report
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=("Proactively migrate legacy facts from each user's memory.json into the reserved __default__ Markdown fact bucket."))
selection = parser.add_mutually_exclusive_group(required=True)
selection.add_argument(
"--all-users",
action="store_true",
help="Migrate every directory-safe user bucket found under STORAGE_PATH/users.",
)
selection.add_argument(
"--user-id",
action="append",
dest="user_ids",
metavar="USER_ID",
help="Migrate one original user ID; repeat this option for multiple users.",
)
parser.add_argument(
"--storage-path",
type=Path,
default=None,
help="DeerMem root directory; defaults to DeerFlow's runtime home.",
)
parser.add_argument("--dry-run", action="store_true", help="Report pending migrations without changing files.")
return parser
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
storage_path = (args.storage_path or runtime_home()).resolve()
config = DeerMemConfig(storage_path=str(storage_path))
user_ids = discover_user_ids(storage_path) if args.all_users else list(dict.fromkeys(args.user_ids or []))
print(f"Storage root: {storage_path}")
if not user_ids:
print("No user buckets found; nothing to migrate.")
return 0
report = migrate_users(config, user_ids, dry_run=args.dry_run)
for entry in report:
status = entry["status"]
user_id = entry["user_id"]
if status == "planned":
print(f"{user_id}: would migrate {entry['path']}")
elif status == "migrated":
print(f"{user_id}: migrated {entry['path']} ({entry.get('from_version')} -> {entry.get('to_version')})")
elif status == "failed":
print(f"{user_id}: FAILED: {entry['error']}")
else:
print(f"{user_id}: already current")
migrated = sum(entry["status"] == "migrated" for entry in report)
planned = sum(entry["status"] == "planned" for entry in report)
current = sum(entry["status"] == "current" for entry in report)
failed = sum(entry["status"] == "failed" for entry in report)
print(f"Summary: migrated={migrated} planned={planned} current={current} failed={failed}")
return 1 if failed else 0
if __name__ == "__main__":
raise SystemExit(main())

35
backend/tests/test_custom_agent.py
View File

@ -9,6 +9,9 @@ import pytest
import yaml
from fastapi.testclient import TestClient
from app.gateway.routers.agents import AGENT_NAME_PATTERN as GATEWAY_AGENT_NAME_PATTERN
from deerflow.agents.memory.backends.deermem.deermem.core.paths import AGENT_NAME_PATTERN as DEERMEM_AGENT_NAME_PATTERN
from deerflow.agents.memory.backends.deermem.deermem.core.paths import DEFAULT_AGENT_BUCKET, validate_agent_name
from deerflow.config.agents_api_config import AgentsApiConfig, get_agents_api_config, set_agents_api_config
# ---------------------------------------------------------------------------
@ -16,6 +19,12 @@ from deerflow.config.agents_api_config import AgentsApiConfig, get_agents_api_co
# ---------------------------------------------------------------------------
def test_reserved_memory_bucket_stays_outside_both_public_agent_patterns() -> None:
assert GATEWAY_AGENT_NAME_PATTERN.fullmatch(DEFAULT_AGENT_BUCKET) is None
assert DEERMEM_AGENT_NAME_PATTERN.fullmatch(DEFAULT_AGENT_BUCKET) is None
validate_agent_name(DEFAULT_AGENT_BUCKET) # Internal storage sentinel remains usable.
def _make_paths(base_dir: Path):
"""Return a Paths instance pointing to base_dir."""
from deerflow.config.paths import Paths
@ -477,16 +486,16 @@ class TestMemoryFilePath:
assert path == tmp_path / "memory.json"
def test_agent_memory_path(self, tmp_path, monkeypatch):
"""Providing agent_name should return per-agent memory file."""
"""All agents share the user-global summary JSON path."""
from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemConfig
from deerflow.agents.memory.backends.deermem.deermem.core.storage import FileMemoryStorage
monkeypatch.setenv("DEERMEM_DATA_DIR", str(tmp_path))
storage = FileMemoryStorage(DeerMemConfig())
path = storage._get_memory_file_path("code-reviewer")
assert path == tmp_path / "agents" / "code-reviewer" / "memory.json"
assert path == tmp_path / "memory.json"
def test_different_paths_for_different_agents(self, tmp_path, monkeypatch):
def test_agents_share_summary_path(self, tmp_path, monkeypatch):
from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemConfig
from deerflow.agents.memory.backends.deermem.deermem.core.storage import FileMemoryStorage
@ -496,9 +505,7 @@ class TestMemoryFilePath:
path_a = storage._get_memory_file_path("agent-a")
path_b = storage._get_memory_file_path("agent-b")
assert path_global != path_a
assert path_global != path_b
assert path_a != path_b
assert path_global == path_a == path_b
# ===========================================================================
@ -730,6 +737,22 @@ class TestAgentsAPI:
response = agent_client.delete("/api/agents/does-not-exist")
assert response.status_code == 404
def test_delete_rejects_memory_only_directory_without_removing_facts(self, agent_client, tmp_path):
agent_dir = tmp_path / "users" / "test-user-autouse" / "agents" / "lead-agent"
facts_dir = agent_dir / "facts"
facts_dir.mkdir(parents=True)
fact_path = facts_dir / "fact_keep.md"
fact_path.write_text("memory data", encoding="utf-8")
response = agent_client.delete("/api/agents/lead-agent")
assert response.status_code == 409
assert fact_path.read_text(encoding="utf-8") == "memory data"
def test_reserved_default_bucket_cannot_be_created_as_custom_agent(self, agent_client):
response = agent_client.post("/api/agents", json={"name": "__default__", "soul": "must fail"})
assert response.status_code == 422
def test_create_agent_with_model_and_tool_groups(self, agent_client):
payload = {
"name": "specialized",

117
backend/tests/test_deermem_self_contained.py
View File

@ -74,6 +74,52 @@ def test_zero_config_defaults_run_non_llm_ops(deermem_data_dir):
assert dm.get_memory(user_id="u")["facts"][0]["content"] == "x"
def test_import_without_agent_name_persists_facts_in_default_markdown_bucket(deermem_data_dir):
dm = DeerMem(backend_config=None)
dm.import_memory(
{
"user": {},
"history": {},
"facts": [
{
"id": "fact_default_import",
"content": "imported through the default manager scope",
"category": "context",
"confidence": 0.8,
"source": "import",
}
],
},
user_id="alice",
)
assert [fact["id"] for fact in dm.get_memory(user_id="alice")["facts"]] == ["fact_default_import"]
facts_root = deermem_data_dir / "users" / "alice" / "agents" / "__default__" / "facts"
assert [path.stem for path in facts_root.glob("**/*.md")] == ["fact_default_import"]
def test_import_empty_summary_sections_replace_existing_summaries_with_complete_defaults(deermem_data_dir):
dm = DeerMem(backend_config=None)
existing = dm.get_memory(user_id="alice")
existing["user"]["workContext"] = {"summary": "old work", "updatedAt": "old"}
existing["user"]["personalContext"] = {"summary": "old personal", "updatedAt": "old"}
existing["history"]["recentMonths"] = {"summary": "old history", "updatedAt": "old"}
dm.import_memory(existing, user_id="alice")
imported = dm.import_memory({"user": {}, "history": {}, "facts": []}, user_id="alice")
assert imported["user"] == {
"workContext": {"summary": "", "updatedAt": ""},
"personalContext": {"summary": "", "updatedAt": ""},
"topOfMind": {"summary": "", "updatedAt": ""},
}
assert imported["history"] == {
"recentMonths": {"summary": "", "updatedAt": ""},
"earlierContext": {"summary": "", "updatedAt": ""},
"longTermBackground": {"summary": "", "updatedAt": ""},
}
def test_trace_id_threads_through_to_tracing_callback(deermem_data_dir):
calls = []
@ -92,6 +138,63 @@ def test_trace_id_threads_through_to_tracing_callback(deermem_data_dir):
assert calls and calls[0] == ("t1", "trace-42", "gpt-x")
def test_default_passive_update_persists_fact_in_reserved_default_bucket(deermem_data_dir):
dm = _deermem_with_fake_llm(payload='{"user":{},"history":{},"newFacts":[{"content":"Default agent fact","category":"context","confidence":0.9}],"factsToRemove":[]}')
dm.add(
thread_id="default-thread",
messages=[HumanMessage(content="remember this"), AIMessage(content="understood")],
user_id="alice",
)
dm._queue.flush()
assert [fact["content"] for fact in dm.get_memory(user_id="alice")["facts"]] == ["Default agent fact"]
facts_root = deermem_data_dir / "users" / "alice" / "agents" / "__default__" / "facts"
assert list(facts_root.glob("**/*.md"))
def test_clear_all_memory_removes_global_summaries_and_every_agent_fact(deermem_data_dir):
dm = DeerMem()
imported = dm.get_memory(user_id="alice")
imported["user"]["workContext"] = {"summary": "shared profile", "updatedAt": "now"}
imported["facts"] = [
{
"id": "fact_default",
"content": "default fact",
"category": "context",
"confidence": 0.9,
"createdAt": "2026-01-01T00:00:00Z",
"source": "manual",
}
]
dm.import_memory(imported, user_id="alice")
dm.create_fact("custom fact", agent_name="custom-agent", user_id="alice")
custom_dir = deermem_data_dir / "users" / "alice" / "agents" / "custom-agent"
config_path = custom_dir / "config.yaml"
config_path.write_text("name: custom-agent\n", encoding="utf-8")
cleared = dm.clear_memory(user_id="alice")
assert cleared["user"]["workContext"]["summary"] == ""
assert cleared["facts"] == []
assert dm.get_memory(agent_name="custom-agent", user_id="alice")["facts"] == []
assert config_path.read_text(encoding="utf-8") == "name: custom-agent\n"
def test_scoped_clear_preserves_shared_summaries(deermem_data_dir):
dm = DeerMem()
imported = dm.get_memory(user_id="alice")
imported["user"]["workContext"] = {"summary": "shared profile", "updatedAt": "now"}
dm.import_memory(imported, user_id="alice")
dm.create_fact("custom fact", agent_name="custom-agent", user_id="alice")
cleared = dm.clear_memory(agent_name="custom-agent", user_id="alice")
assert cleared["facts"] == []
assert cleared["user"]["workContext"]["summary"] == "shared profile"
assert dm.get_memory(user_id="alice")["user"]["workContext"]["summary"] == "shared profile"
def test_tracing_callback_optional_no_langfuse(deermem_data_dir):
dm = _deermem_with_fake_llm({"model": {"provider": "openai", "model": "gpt-x", "api_key": "k", "base_url": "u"}})
assert dm._config.tracing_callback is None # langfuse not hard-required
@ -135,7 +238,7 @@ def test_portability_only_abc_contract_imports_deerflow():
deerflow_imports.append((p.relative_to(root).as_posix(), s))
assert len(deerflow_imports) == 1, deerflow_imports
assert deerflow_imports[0][0] == "deer_mem.py"
assert "memory.manager import MemoryManager" in deerflow_imports[0][1]
assert "memory.manager import MemoryConflictError, MemoryCorruptionError, MemoryManager" in deerflow_imports[0][1]
# Minimal vendored host contract (what another agent would ship). DeerMem only
@ -166,6 +269,9 @@ class MemoryManager(ABC):
def import_memory(self, memory_data, *, user_id=None, agent_name=None) -> dict: ...
@abstractmethod
def export_memory(self, *, user_id=None, agent_name=None) -> dict: ...
class MemoryConflictError(RuntimeError): ...
class MemoryCorruptionError(RuntimeError): ...
'''
@ -190,10 +296,11 @@ def test_portability_vendor_to_other_agent(tmp_path, monkeypatch):
# Repoint the single ABC-contract import line to the vendored manager.
deer_mem_file = dst_pkg / "deer_mem.py"
text = deer_mem_file.read_text(encoding="utf-8")
assert "from deerflow.agents.memory.manager import MemoryManager" in text
contract_import = "from deerflow.agents.memory.manager import MemoryConflictError, MemoryCorruptionError, MemoryManager"
assert contract_import in text
text = text.replace(
"from deerflow.agents.memory.manager import MemoryManager",
"from otheragent.manager import MemoryManager",
contract_import,
"from otheragent.manager import MemoryConflictError, MemoryCorruptionError, MemoryManager",
)
deer_mem_file.write_text(text, encoding="utf-8")
@ -229,7 +336,7 @@ def test_per_user_memory_path_matches_host_safe_user_id(deermem_data_dir):
user_id = "test-user-123@example.com"
# storage_path mirrors what the host factory injects (runtime_home / base_dir)
dm = DeerMem(backend_config={"storage_path": str(deermem_data_dir)})
dm.create_fact("User prefers concise answers", category="preference", user_id=user_id)
dm.create_fact("User prefers concise answers", category="preference", agent_name="default", user_id=user_id)
expected_safe = make_safe_user_id(user_id)
expected_file = deermem_data_dir / "users" / expected_safe / "memory.json"

2
backend/tests/test_memory_manager_pluggable.py
View File

@ -143,7 +143,7 @@ def test_empty_storage_path_factory_injects_runtime_home(tmp_path, monkeypatch)
set_memory_config(MemoryConfig(manager_class="deermem")) # no storage_path
manager = get_memory_manager()
assert Path(manager._config.storage_path) == tmp_path
manager.create_fact("hello", user_id="u1")
manager.create_fact("hello", user_id="u1", agent_name="test-agent")
# per-user dir created under the injected runtime_home root
user_dirs = [p.name for p in (tmp_path / "users").iterdir() if p.is_dir()]
assert len(user_dirs) == 1

99
backend/tests/test_memory_markdown_migration_cli.py Normal file
View File

@ -0,0 +1,99 @@
"""Tests for the proactive JSON-to-Markdown memory migration CLI."""
import json
from pathlib import Path
from deerflow.agents.memory.backends.deermem.deermem.core.paths import fact_file_path
def _legacy_memory(content: str) -> dict:
return {
"version": "1.0",
"revision": 0,
"lastUpdated": "2026-01-01T00:00:00Z",
"user": {"workContext": {"summary": "keep me", "updatedAt": "2026-01-01T00:00:00Z"}},
"history": {},
"facts": [
{
"id": "fact_legacy",
"content": content,
"category": "context",
"confidence": 0.9,
"createdAt": "2026-01-01T00:00:00Z",
"source": "manual",
}
],
}
def _seed_user(root: Path, user_bucket: str, content: str) -> Path:
path = root / "users" / user_bucket / "memory.json"
path.parent.mkdir(parents=True)
path.write_text(json.dumps(_legacy_memory(content)), encoding="utf-8")
return path
def test_cli_migrates_one_explicit_user(tmp_path: Path) -> None:
from scripts.migrate_memory_markdown import main
memory_path = _seed_user(tmp_path, "alice", "legacy alice fact")
exit_code = main(["--storage-path", str(tmp_path), "--user-id", "alice"])
assert exit_code == 0
persisted = json.loads(memory_path.read_text(encoding="utf-8"))
assert "facts" not in persisted
assert persisted["user"]["workContext"]["summary"] == "keep me"
assert fact_file_path(memory_path, "fact_legacy", agent_name="__default__").exists()
def test_cli_dry_run_reports_without_writing(tmp_path: Path, capsys) -> None:
from scripts.migrate_memory_markdown import main
memory_path = _seed_user(tmp_path, "alice", "legacy alice fact")
original = memory_path.read_bytes()
exit_code = main(["--storage-path", str(tmp_path), "--user-id", "alice", "--dry-run"])
assert exit_code == 0
assert memory_path.read_bytes() == original
assert not (memory_path.parent / "agents" / "__default__" / "facts").exists()
assert "would migrate" in capsys.readouterr().out
def test_cli_all_users_migrates_each_discovered_bucket(tmp_path: Path) -> None:
from scripts.migrate_memory_markdown import main
alice_path = _seed_user(tmp_path, "alice", "legacy alice fact")
bob_path = _seed_user(tmp_path, "bob", "legacy bob fact")
exit_code = main(["--storage-path", str(tmp_path), "--all-users"])
assert exit_code == 0
for memory_path in (alice_path, bob_path):
assert "facts" not in json.loads(memory_path.read_text(encoding="utf-8"))
assert fact_file_path(memory_path, "fact_legacy", agent_name="__default__").exists()
def test_cli_is_idempotent_and_reports_current_user(tmp_path: Path, capsys) -> None:
from scripts.migrate_memory_markdown import main
_seed_user(tmp_path, "alice", "legacy alice fact")
arguments = ["--storage-path", str(tmp_path), "--user-id", "alice"]
assert main(arguments) == 0
capsys.readouterr()
assert main(arguments) == 0
assert "already current" in capsys.readouterr().out
def test_cli_requires_user_selection(tmp_path: Path) -> None:
from scripts.migrate_memory_markdown import main
try:
main(["--storage-path", str(tmp_path)])
except SystemExit as exc:
assert exc.code == 2
else:
raise AssertionError("CLI must require --all-users or --user-id")

118
backend/tests/test_memory_router.py
View File

@ -1,4 +1,5 @@
import asyncio
import json
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
@ -6,6 +7,8 @@ from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.gateway.routers import memory
from deerflow.agents.memory import MemoryConflictError, MemoryCorruptionError
from deerflow.agents.memory.backends.deermem.deer_mem import DeerMem
def _sample_memory(facts: list[dict] | None = None) -> dict:
@ -86,6 +89,36 @@ def test_import_memory_route_returns_imported_memory() -> None:
assert response.json()["facts"] == imported_memory["facts"]
def test_import_route_without_agent_name_persists_default_bucket_markdown(tmp_path) -> None:
app = FastAPI()
app.include_router(memory.router)
manager = DeerMem(backend_config={"storage_path": str(tmp_path)})
imported_memory = _sample_memory(
facts=[
{
"id": "fact_gateway_import",
"content": "Gateway imports use the default agent bucket.",
"category": "context",
"confidence": 0.9,
"createdAt": "2026-07-21T00:00:00Z",
"source": "import",
}
]
)
with (
patch("app.gateway.routers.memory.get_memory_manager", return_value=manager),
patch("app.gateway.routers.memory.get_effective_user_id", return_value="alice"),
TestClient(app) as client,
):
response = client.post("/api/memory/import", json=imported_memory)
assert response.status_code == 200
assert [fact["id"] for fact in response.json()["facts"]] == ["fact_gateway_import"]
facts_root = tmp_path / "users" / "alice" / "agents" / "__default__" / "facts"
assert [path.stem for path in facts_root.glob("**/*.md")] == ["fact_gateway_import"]
def test_import_memory_route_preserves_source_error() -> None:
app = FastAPI()
app.include_router(memory.router)
@ -144,6 +177,34 @@ def test_create_memory_fact_route_returns_updated_memory() -> None:
assert response.json()["facts"] == updated_memory["facts"]
def test_create_memory_fact_route_maps_conflict_to_409() -> None:
app = FastAPI()
app.include_router(memory.router)
mock_mgr = MagicMock()
mock_mgr.create_fact.side_effect = MemoryConflictError("stale write")
with patch("app.gateway.routers.memory.get_memory_manager", return_value=mock_mgr):
with TestClient(app) as client:
response = client.post("/api/memory/facts", json={"content": "fact"})
assert response.status_code == 409
assert response.json()["detail"] == "Memory changed concurrently; reload and retry."
def test_get_memory_route_maps_corruption_to_stable_500() -> None:
app = FastAPI()
app.include_router(memory.router)
mock_mgr = MagicMock()
mock_mgr.get_memory.side_effect = MemoryCorruptionError("private path and parser detail")
with patch("app.gateway.routers.memory.get_memory_manager", return_value=mock_mgr):
with TestClient(app) as client:
response = client.get("/api/memory")
assert response.status_code == 500
assert response.json()["detail"] == "Stored memory data is corrupted."
def test_delete_memory_fact_route_returns_updated_memory() -> None:
app = FastAPI()
app.include_router(memory.router)
@ -184,6 +245,63 @@ def test_update_memory_fact_route_returns_updated_memory() -> None:
assert response.json()["facts"] == updated_memory["facts"]
def test_settings_fact_crud_without_agent_name_uses_default_agent(tmp_path) -> None:
"""The current Settings API sends no agent_name; it must remain usable."""
app = FastAPI()
app.include_router(memory.router)
memory_path = tmp_path / "users" / "alice" / "memory.json"
memory_path.parent.mkdir(parents=True)
legacy = _sample_memory(
facts=[
{
"id": "fact_legacy",
"content": "Legacy global fact",
"category": "context",
"confidence": 0.9,
"createdAt": "2026-03-20T00:00:00Z",
"source": "manual",
}
]
)
memory_path.write_text(json.dumps(legacy), encoding="utf-8")
manager = DeerMem(backend_config={"storage_path": str(tmp_path)})
with (
patch("app.gateway.routers.memory.get_memory_manager", return_value=manager),
patch("app.gateway.routers.memory.get_effective_user_id", return_value="alice"),
TestClient(app) as client,
):
fetched = client.get("/api/memory")
assert fetched.status_code == 200
assert [fact["content"] for fact in fetched.json()["facts"]] == ["Legacy global fact"]
exported = client.get("/api/memory/export")
assert exported.status_code == 200
assert [fact["content"] for fact in exported.json()["facts"]] == ["Legacy global fact"]
created = client.post("/api/memory/facts", json={"content": "Project uses Python", "category": "context", "confidence": 0.8})
assert created.status_code == 200
assert all(isinstance(fact["source"], str) for fact in created.json()["facts"])
fact_id = next(fact["id"] for fact in created.json()["facts"] if fact["content"] == "Project uses Python")
updated = client.patch(f"/api/memory/facts/{fact_id}", json={"content": "Project uses Python 3.12"})
assert updated.status_code == 200
assert updated.json()["facts"][0]["content"] == "Project uses Python 3.12"
deleted = client.delete(f"/api/memory/facts/{fact_id}")
assert deleted.status_code == 200
assert [fact["id"] for fact in deleted.json()["facts"]] == ["fact_legacy"]
deleted_legacy = client.delete("/api/memory/facts/fact_legacy")
assert deleted_legacy.status_code == 200
assert deleted_legacy.json()["facts"] == []
facts_root = tmp_path / "users" / "alice" / "agents" / "__default__" / "facts"
assert facts_root.exists()
assert not list(facts_root.glob("**/*.md"))
assert "facts" not in json.loads(memory_path.read_text(encoding="utf-8"))
def test_update_memory_fact_route_preserves_omitted_fields() -> None:
app = FastAPI()
app.include_router(memory.router)

25
backend/tests/test_memory_storage.py
View File

@ -55,11 +55,11 @@ class TestFileMemoryStorage:
assert storage._get_memory_file_path(None) == tmp_path / "memory.json"
def test_get_memory_file_path_agent(self, tmp_path, monkeypatch):
"""Legacy per-agent path lives under the DeerMem root ($DEERMEM_DATA_DIR)."""
"""Agent facts share the user's global summary JSON path."""
monkeypatch.setenv("DEERMEM_DATA_DIR", str(tmp_path))
storage = FileMemoryStorage(DeerMemConfig())
path = storage._get_memory_file_path("test-agent")
assert path == tmp_path / "agents" / "test-agent" / "memory.json"
assert path == tmp_path / "memory.json"
@pytest.mark.parametrize("invalid_name", ["", "../etc/passwd", "agent/name", "agent\\name", "agent name", "agent@123", "agent_name"])
def test_validate_agent_name_invalid(self, invalid_name):
@ -78,9 +78,10 @@ class TestFileMemoryStorage:
monkeypatch.setenv("DEERMEM_DATA_DIR", str(tmp_path))
memory_file = tmp_path / "memory.json"
storage = FileMemoryStorage(DeerMemConfig())
result = storage.save({"version": "1.0", "facts": [{"content": "test fact"}]})
result = storage.save({"version": "1.0", "facts": []})
assert result is True
assert memory_file.exists()
assert "facts" not in memory_file.read_text(encoding="utf-8")
def test_save_does_not_mutate_caller_dict(self, tmp_path, monkeypatch):
monkeypatch.setenv("DEERMEM_DATA_DIR", str(tmp_path))
@ -98,16 +99,18 @@ class TestFileMemoryStorage:
memory_file.parent.mkdir(parents=True, exist_ok=True)
import json as _json
memory_file.write_text(_json.dumps({"version": "1.0", "facts": [{"content": "original"}]}))
memory_file.write_text(_json.dumps({"version": "2.0", "user": {"workContext": {"summary": "original"}}, "history": {}}))
storage = FileMemoryStorage(DeerMemConfig())
cached = storage.load()
assert cached["facts"][0]["content"] == "original"
assert cached["user"]["workContext"]["summary"] == "original"
with patch("builtins.open", side_effect=OSError("disk full")):
result = storage.save({"version": "1.0", "facts": [{"content": "mutated"}]})
changed = create_empty_memory()
changed["user"]["workContext"]["summary"] = "mutated"
result = storage.save(changed)
assert result is False
after = storage.load()
assert after["facts"][0]["content"] == "original"
assert after["user"]["workContext"]["summary"] == "original"
def test_cache_thread_safety(self, tmp_path, monkeypatch):
monkeypatch.setenv("DEERMEM_DATA_DIR", str(tmp_path))
@ -137,13 +140,13 @@ class TestFileMemoryStorage:
monkeypatch.setenv("DEERMEM_DATA_DIR", str(tmp_path))
memory_file = tmp_path / "memory.json"
memory_file.parent.mkdir(parents=True, exist_ok=True)
memory_file.write_text('{"version": "1.0", "facts": [{"content": "initial fact"}]}')
memory_file.write_text('{"version": "2.0", "user": {"workContext": {"summary": "initial"}}, "history": {}}')
storage = FileMemoryStorage(DeerMemConfig())
memory1 = storage.load()
assert memory1["facts"][0]["content"] == "initial fact"
memory_file.write_text('{"version": "1.0", "facts": [{"content": "updated fact"}]}')
assert memory1["user"]["workContext"]["summary"] == "initial"
memory_file.write_text('{"version": "2.0", "user": {"workContext": {"summary": "updated"}}, "history": {}}')
memory2 = storage.reload()
assert memory2["facts"][0]["content"] == "updated fact"
assert memory2["user"]["workContext"]["summary"] == "updated"
class TestCreateStorage:

1189
backend/tests/test_memory_storage_markdown.py Normal file
View File

File diff suppressed because it is too large Load Diff

10
backend/tests/test_memory_storage_user_isolation.py
View File

@ -5,6 +5,7 @@ from pathlib import Path
import pytest
from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemConfig
from deerflow.agents.memory.backends.deermem.deermem.core.paths import fact_file_path
from deerflow.agents.memory.backends.deermem.deermem.core.storage import FileMemoryStorage, create_empty_memory
@ -75,12 +76,15 @@ class TestUserIsolatedStorage:
assert s.load(user_id="alice")["user"]["workContext"]["summary"] == "alice"
def test_user_agent_memory_file_location(self, base_dir: Path):
"""Per-user per-agent memory uses {root}/users/{uid}/agents/{name}/memory.json."""
"""One user JSON stores summaries; agent facts live below the agent directory."""
s = FileMemoryStorage(DeerMemConfig())
memory = create_empty_memory()
memory["user"]["workContext"]["summary"] = "agent scoped"
memory["facts"] = [{"id": "fact_agent", "content": "agent scoped"}]
s.save(memory, "test-agent", user_id="alice")
assert (base_dir / "users" / "alice" / "agents" / "test-agent" / "memory.json").exists()
memory_path = base_dir / "users" / "alice" / "memory.json"
assert memory_path.exists()
assert not (base_dir / "users" / "alice" / "agents" / "test-agent" / "memory.json").exists()
assert fact_file_path(memory_path, "fact_agent", agent_name="test-agent").exists()
def test_cache_key_is_user_agent_tuple(self, base_dir: Path):
"""Cache keys must be (user_id, agent_name) tuples."""

5
config.example.yaml
View File

@ -1638,6 +1638,11 @@ memory:
# Each backend self-interprets it (DeerMem parses it into DeerMemConfig).
backend_config:
storage_path: "" # empty = deer-flow base_dir (factory injects absolute runtime_home); a non-empty path is the root DIRECTORY (per-user memory under {storage_path}/users/{uid}/memory.json)
storage_class: file # file (default) or a dotted MemoryStorage class path; invalid persistent backends fail fast
strict_user_scope: false # set true in authenticated deployments after all callers propagate user_id
manifest_filename: memory.json # user-global JSON: version/revision/time + user/history only; no facts or fact index
file_lock_timeout_seconds: 10 # per-scope cross-process advisory lock timeout (single-machine local filesystem)
retrieval_adapter: "" # optional dotted factory(config) supplied by the retrieval module
debounce_seconds: 30 # Wait time before processing queued updates
model: # LLM for memory extraction; omit all fields = no extraction (non-LLM ops still work; an update raises)
# provider: openai

260
docs/plans/STORAGE_REWRITE_CHANGES.md Normal file
View File

@ -0,0 +1,260 @@
# DeerMem Storage Rewrite: Implementation Guide
> This document explains the final PR behavior for readers who are new to the codebase. It complements `STORAGE_REWRITE_PLAN.md`: the plan explains the intended contract; this file explains where the code changed and how a request travels through it.
## 1. What changed on disk
Before the rewrite, facts and summaries could live together in a coarse JSON document. The final layout is:
```text
users/alice/
├── memory.json
├── memory.json.v1.bak
└── agents/
├── __default__/facts/30/fact_a.md
└── research-agent/facts/ee/fact_b.md
```
`memory.json` owns project-independent summaries. Each Markdown file owns one atomic fact. No JSON fact index duplicates the Markdown repository.
## 2. Main files
### `deer_mem.py`
This is the DeerMem implementation of the public `MemoryManager` plugin interface.
It:
- maps omitted agents to `__default__`;
- canonicalizes explicit agent names to lowercase;
- translates storage-private exceptions to public manager errors;
- converts structured internal source metadata to the legacy public string;
- reloads complete documents where the manager/API contract promises completeness.
Gateway code does not import DeerMem storage internals.
### `deermem/core/paths.py`
This owns user, agent, and fact path validation. The internal `__default__` sentinel is deliberately outside the public custom-agent regular expression.
Fact paths are sharded by the first two hexadecimal characters of `SHA-256(fact_id)`. This gives deterministic direct lookup while distributing generated IDs that all begin with `fact_`.
### `deermem/core/storage.py`
This owns canonical parsing/rendering, locks, revisions, journaling, recovery, migration, cache invalidation, fact repository operations, and retrieval notifications.
### `deermem/core/updater.py`
This converts manual CRUD and LLM extraction results into repository change sets. It decides whether an operation is a point mutation or depends on a complete snapshot.
### `app/gateway/routers/memory.py`
This keeps the HTTP schema compatible. It catches only backend-neutral MemoryManager exceptions and maps concurrency to 409 and corruption to a stable 500.
### `scripts/migrate_memory_markdown.py`
This optional CLI lets operators preview or proactively run the same migration that normal first reads perform lazily.
## 3. Read trajectory
For an ordinary request with no explicit agent:
```text
Gateway / middleware
-> MemoryManager.get_memory(user_id, agent_name=None)
-> DeerMem canonicalizes agent to __default__
-> FileMemoryStorage.load(__default__)
-> recover journal / migrate legacy data when needed
-> compare memory.json (mtime_ns, file_size, revision) with cache
-> cache hit, or parse selected agent Markdown files
-> DeerMem converts source metadata to public strings
-> Gateway validates and returns the compatibility document
```
For a custom agent, the same trajectory reads that lowercase agent bucket and combines it with the user's shared summaries.
## 4. Single-fact update trajectory
```text
MemoryManager.update_fact(fact_id)
-> storage.get_fact(fact_id)
-> build one upsert with expected fact revision
-> acquire in-process + cross-process locks
-> recover an interrupted prior journal if present
-> validate shared revision and target fact revision
-> journal only memory.json + the addressed fact
-> atomically replace the Markdown fact and memory.json
-> fsync files and the POSIX parent directory
-> notify retrieval for that fact only
-> reload a complete compatibility response
```
Unchanged sibling facts are not backed up, rewritten, or re-indexed.
## 5. Why there are two revisions
The shared JSON revision protects the multi-file transaction. The fact revision protects one Markdown object.
They answer different questions:
```text
manifest revision: Did anything in this user's memory change?
fact revision: Is the exact fact I read still the same object version?
```
Direct `upsert_fact` and `delete_fact` therefore accept separate expected manifest and fact revisions.
## 6. Safe rebase versus fresh-snapshot retry
A point update to `fact_a` can safely move from manifest revision 10 to 11 if `fact_a` still has the expected revision. A scoped clear cannot: its meaning depends on every fact that existed at the successful commit.
The final implementation uses this rule:
```text
point mutation + valid fact preconditions
-> explicit bounded manifest rebase is allowed
clear / trim / consolidation / collection-derived set
-> manifest conflict
-> reload full document
-> recompute complete operation
-> bounded retry
```
This fixes two reviewed races:
- a fact created between a scoped-clear snapshot and commit no longer survives behind a successful “empty” response;
- two creators starting from 9/10 facts cannot commit 11 facts.
## 7. Clear semantics
The manager distinguishes two calls:
```text
clear_memory(user_id=alice)
clears shared user/history summaries and every agent fact bucket
preserves custom-agent config.yaml and SOUL.md
clear_memory(user_id=alice, agent_name=research-agent)
clears only research-agent facts
preserves shared user/history summaries
```
The file backend's all-user clear holds the user lock while enumerating buckets. It first migrates facts from any unread legacy per-agent `memory.json` without adopting legacy summaries that are about to be cleared, then deletes the resulting canonical fact files; retained `.v1.bak` files are inactive migration evidence and are never read back, while custom-agent configuration remains untouched.
## 8. Source compatibility
On disk:
```yaml
source:
type: conversation
threadId: thread_123
```
Public API:
```json
{"source": "thread_123"}
```
Manual/import/consolidation sources similarly return their type string. The richer internal form is ready for retrieval metadata without breaking the frontend.
## 9. Cache behavior
The earlier cache signature scanned and statted every fact file on every `load()`. The final cache observes the shared `memory.json` only.
The token is `(mtime_ns, file_size, revision)`. Every supported storage mutation replaces the JSON and increments its revision, so another process's write invalidates cached agent documents even when a coarse-mtime filesystem reports identical metadata for same-size writes. Validation reads one shared JSON and does not scale with the number of fact files. This may invalidate more agents than strictly necessary, but it avoids an O(n) fact-directory walk without introducing a second manifest.
Manual out-of-band Markdown edits are not part of the supported write API and require `reload()`.
Per-scope in-process locks live in a weak-value dictionary, so inactive users do not permanently grow the lock map.
## 10. Migration trajectory
When a v1 `memory.json` still contains facts:
```text
first load/reload
-> detect legacy facts/version
-> acquire normal locks
-> durably retain every source as memory.json.v1.bak
-> normalize facts into __default__ scope
-> commit Markdown files and preserved summaries through the journal
-> rewrite memory.json without facts
-> release storage locks and notify the configured retrieval adapter
-> continue the original read
```
The explicit CLI path emits the same retrieval upserts. Re-running an idempotent migration emits no duplicate notification because no fact is rewritten.
Operators may run:
```bash
cd backend
PYTHONPATH=. python scripts/migrate_memory_markdown.py --all-users --dry-run
PYTHONPATH=. python scripts/migrate_memory_markdown.py --all-users
```
The command is idempotent, accepts repeated `--user-id`, supports `--storage-path`, continues reporting after one user fails, and exits non-zero if any user failed. It is not required for startup.
This is a one-way application migration: pre-PR code cannot read Markdown facts. Before upgrading a persistent deployment, stop DeerFlow and take a filesystem snapshot or full backup of the configured storage root. Storage also retains each destructive v1 JSON source beside its original path as `{source_filename}.v1.bak` before committing v2 data. Existing backups are immutable; a mismatch or backup-write failure aborts before changing the source. The local backup contains pre-migration data only and does not replace the full snapshot requirement.
## 11. Configuration changes
`manifest_filename` and lock timeout remain configurable. Markdown format and journaling are required invariants, not modes.
`fact_format` and `journal_enabled` are not `DeerMemConfig` fields. If supplied in `backend_config`, they follow the normal unknown-key behavior: DeerMem logs a warning and ignores them.
## 12. Agent naming
The Gateway already stores custom-agent names in lowercase. DeerMem now makes the same rule explicit at its manager boundary:
```text
Lead-Agent == lead-agent == LEAD-AGENT
```
This gives Linux, macOS, and Windows the same behavior. A cross-layer contract test pins that both public naming patterns reject `__default__`, while storage may use the sentinel internally.
## 13. Failure behavior
- Stale same-fact write: conflict, no overwrite.
- Same-ID concurrent create: conflict, no conversion into update.
- Continuing collection contention: conflict after bounded retries.
- Corrupt JSON/Markdown: stable public corruption error; original files retained.
- Conflicting legacy summaries/facts: migration fails loudly and preserves the legacy source.
- Missing, unreadable, or mismatched persistent v1 backup: migration stops before changing the source.
- Retrieval adapter failure: persistence remains committed; the failure is logged/reported per fact.
## 14. Review-driven fixes retained in the implementation
- Default bucket changed from colliding `lead-agent` to reserved `__default__`.
- Custom-agent deletion requires a genuine config file.
- Legacy global facts migrate on first read and through an optional CLI.
- Public responses retain string `source`.
- Conflicts map to HTTP 409 through MemoryManager-neutral errors.
- Clear All and scoped clear have distinct semantics.
- Full index rebuild validates original email-style user IDs correctly.
- Lock cache is reclaimable.
- Manifest and fact revisions are separate.
- Retry dispatch uses exception subtypes, not message text.
- POSIX atomic replacement fsyncs the parent directory.
- Stale default-bucket test now asserts `__default__`.
- Snapshot-derived set operations recompute instead of replaying stale intent.
- Cache validation no longer scans every fact on every read.
- Cache tokens include the persisted revision to survive coarse-mtime same-size writes.
- Fact paths shard by `SHA-256(fact_id)[:2]` rather than the constant `fa` prefix.
- Destructive migration writes immutable `.v1.bak` sources before v2 data.
- Fixed storage invariants are no longer presented as configurable features.
## 15. Verification map
Primary coverage lives in:
- `tests/test_memory_storage_markdown.py`
- `tests/test_deermem_self_contained.py`
- `tests/test_memory_router.py`
- `tests/test_memory_markdown_migration_cli.py`
- `tests/test_custom_agent.py`
Tests cover layout, compatibility, incremental writes, concurrency, cap enforcement, clear semantics, corruption, migration, retrieval delegation, cache invalidation, path containment, portability, Windows lock behavior, and POSIX directory sync.

224
docs/plans/STORAGE_REWRITE_PLAN.md Normal file
View File

@ -0,0 +1,224 @@
# DeerMem Storage Rewrite Plan
> Status: implemented in PR #4279 and retained as the design record requested during review.
> Scope: file-backed storage only. Retrieval ranking, embeddings, project scope, and recall policy remain separate work.
## 1. Objective
Separate durable memory into two ownership layers:
```text
user_id
├── memory.json # project-independent user/history summaries
└── agents/{agent_name}/facts/ # agent-related atomic facts
└── {sha256-prefix}/{fact_id}.md
```
The rewrite must:
1. keep one global summary JSON per user;
2. store every fact as one canonical Markdown document;
3. prevent custom agents from sharing a fact repository accidentally;
4. make single-fact writes genuinely incremental;
5. preserve the existing MemoryManager and HTTP response shape;
6. give future retrieval implementations a stable fact repository API;
7. survive stale writers, process concurrency, and interrupted multi-file writes;
8. upgrade existing JSON facts without requiring application downtime.
## 2. Non-goals
- No `project` or `project_id` storage scope in this PR.
- No vector database, BM25, embedding, MMR, or reranking implementation.
- No change to when facts are recalled or injected into prompts.
- No SQLite/Postgres backend.
- No new frontend memory-management experience.
- No support for disabling the safety journal or selecting another fact format.
## 3. Canonical layout
```text
{storage_root}/users/{safe_user_id}/
├── memory.json
├── memory.json.v1.bak # retained only after migrating this v1 source
├── .memory.lock
├── .memory.journal.json # exists only during/recovering a transaction
├── .recovery/ # transaction backups
└── agents/
├── __default__/
│ └── facts/{sha256-prefix}/{fact_id}.md
└── {custom-agent}/
├── config.yaml # owned by the custom-agent subsystem
├── SOUL.md
└── facts/{sha256-prefix}/{fact_id}.md
```
`memory.json` contains only:
```json
{
"version": "2.0",
"revision": 12,
"lastUpdated": "2026-07-19T00:00:00Z",
"user": {},
"history": {}
}
```
It never contains facts, fact paths, hashes, embeddings, or a fact manifest.
`sha256-prefix` is the first two hexadecimal characters of `SHA-256(fact_id)`, giving a deterministic 256-way shard that also distributes generated `fact_*` IDs.
## 4. Scope rules
The storage scope in this PR is:
```text
user_id + agent_name
```
- `thread_id` is source metadata only.
- Omitted `agent_name` resolves inside DeerMem to the reserved `__default__` bucket.
- `__default__` is outside the public custom-agent grammar.
- Public agent identifiers are case-insensitive and canonicalized to lowercase.
- A custom agent named `lead-agent` is distinct from `__default__`.
- The custom-agent delete route must require `config.yaml`; a memory-only directory is preserved.
## 5. Fact document
Each Markdown file has YAML front matter plus one human-readable body:
```markdown
---
id: fact_ab12
schemaVersion: 2
category: constraint
topics: [python, runtime]
confidence: 0.95
status: active
user_id: alice
agent_name: research-agent
source:
type: conversation
threadId: thread_123
createdAt: 2026-07-19T00:00:00Z
updatedAt: 2026-07-19T00:00:00Z
revision: 1
consolidatedFrom: []
---
# Runtime constraint
Project uses Python 3.12.
```
Storage validates IDs, content, category, confidence, timestamps, lifecycle status, scope, source, revision, and consolidation metadata before committing.
## 6. Compatibility boundary
Internally, `source` is structured metadata. Public MemoryManager/API documents preserve the historical string field:
```text
{type: conversation, threadId: thread_123} -> "thread_123"
{type: manual, threadId: null} -> "manual"
```
Manager/API reads materialize a compatibility `facts` array for the selected/default agent. The frontend never reads facts from `memory.json`.
Storage-specific conflict and corruption errors are translated at the MemoryManager boundary. The Gateway maps conflicts to HTTP 409 and returns a stable, non-sensitive HTTP 500 for corruption.
## 7. Repository contract
The file backend exposes:
- `get_fact` / `list_facts`
- `upsert_fact` / `delete_fact`
- `apply_changes`
- summary reads/updates
- migration
- index lifecycle and scoped search
`apply_changes()` returns an explicit incomplete delta. A caller that promises a full compatibility document must reload after committing.
Direct fact CRUD uses separate preconditions:
- expected shared `memory.json` revision;
- expected target fact revision or expected absence.
## 8. Concurrency model
Every transaction holds an in-process scope lock and a cross-process user file lock. Fact files and the shared JSON are protected by a recoverable journal.
Operations are divided by intent:
### Point operations
An update/delete of named facts depends only on addressed fact preconditions. It may explicitly rebase after a manifest conflict if all expected absence/revision checks still hold.
### Snapshot-derived operations
Clear, max-fact trimming, consolidation, and other collection-derived set operations depend on the complete fact snapshot. They must not replay an old delete/trim set against a newer manifest.
On conflict they:
1. reload the complete selected-agent document;
2. recompute the complete operation;
3. retry at most three times;
4. return a conflict if contention continues.
This preserves Clear All/scoped-clear meaning and the `max_facts` invariant.
Summary change sets are patches at the `user`/`history` child-key level: omitted siblings remain persisted. Import remains replacement-oriented by normalizing incoming sections against the complete empty compatibility schema before it calls storage.
## 9. Cache model
Every supported fact mutation advances and atomically replaces the shared `memory.json`. Cache validation uses `(mtime_ns, file_size, revision)` from that file. The persisted revision prevents a stale hit when a coarse-mtime filesystem reports identical metadata for same-size writes.
The read path does not glob/stat every Markdown fact merely to validate a cache hit, so validation cost does not grow with the number of fact files. It reads the shared JSON revision on every check. Direct out-of-band edits to Markdown require an explicit `reload()` or restart.
Inactive per-scope locks are weakly cached and may be garbage-collected.
## 10. Migration
The first normal default read detects legacy facts in `memory.json`, acquires the normal locks, and migrates them into `__default__`. User/history summaries are preserved and the JSON is rewritten without `facts`. Explicit and lazy migrations return their committed fact deltas through the call chain and notify a configured retrieval adapter after releasing storage locks.
Clear All enumerates every agent while holding the user lock. Facts from any unread legacy per-agent JSON are migrated first without adopting legacy summaries that are about to be cleared, and the resulting canonical facts are then deleted with the rest of the bucket. This prevents both summary conflicts from blocking an explicit clear and a later read from resurrecting skipped facts. The immutable `.v1.bak` remains inactive and agent configuration files remain untouched.
The v1-to-v2 migration is one-way for the running application because pre-PR code does not read Markdown facts. Operators must stop DeerFlow and create a filesystem snapshot or full backup of the configured storage root before upgrading a persistent deployment. Before the first destructive v2 write, storage atomically and durably retains every migrated JSON source as `{source_filename}.v1.bak`. An existing backup is never overwritten: if it differs from the source, or if the backup cannot be written, migration stops before changing v1 data. These local backups preserve pre-migration data only and do not replace the required full snapshot.
An optional idempotent operator CLI supports preflight audits and proactive migration:
```bash
cd backend
PYTHONPATH=. python scripts/migrate_memory_markdown.py --all-users --dry-run
PYTHONPATH=. python scripts/migrate_memory_markdown.py --all-users
```
Repeated `--user-id` and custom `--storage-path` are supported. The CLI is optional; lazy first-read migration remains the zero-administration path.
Legacy per-agent JSON is deleted only after its summaries are safely adopted or confirmed identical. Conflicting non-empty summaries and different same-ID facts fail loudly and preserve the source.
## 11. Fixed storage invariants
This release has one canonical fact format and requires crash recovery:
```text
fact format = Markdown
journal = enabled
```
These are implementation invariants, not public `DeerMemConfig` fields. If `fact_format` or `journal_enabled` is supplied in `backend_config`, it follows the normal unknown-key behavior: DeerMem logs a warning and ignores it.
## 12. Acceptance criteria
- Global JSON contains no facts or manifest.
- Default and custom-agent fact sets remain isolated.
- One fact mutation writes/notifies only addressed facts.
- Same-ID creates and stale same-fact writes fail.
- Snapshot-derived operations recompute after conflicts.
- Persisted fact count never exceeds `max_facts` after concurrent creates.
- Scoped clear either clears facts committed before its successful revision or returns conflict.
- Legacy global facts migrate through normal reads and the CLI.
- Public source remains a string.
- Cache validation does not scale with the number of fact files and includes the persisted revision.
- Every destructive v1 JSON source is durably backed up before migration.
- Tests cover Windows locking, POSIX directory sync, recovery, migration, concurrency, API compatibility, and portability.