fix(memory): treat facts deleted during unlocked DeerMem scans as removed (#6255)

* fix(memory): treat facts deleted during unlocked DeerMem scans as removed

load(), reload(), and the full rebuild_index() scan list agent fact
files without the storage locks. A delete committed between the listing
and opening a file raised MemoryStorageCorruption for intact data, which
surfaced as HTTP 500 from the memory API and an empty memory block in
the prompt. A listed fact that no longer exists is now skipped; an entry
that is still present but unreadable is still corruption.

* docs(changelog): reference #6255 for the DeerMem vanished-fact fix

* docs(memory): note lock coverage and heuristic limits of the vanished-fact skip

CHANGELOG.md

@@ -466,6 +466,16 @@ This release closes that milestone with **301 merged pull requests**.
 
 ### Fixed
 
+- **memory:** Reading DeerMem agent memory no longer fails while another write
+  deletes a fact. `load()`, `reload()`, and the full `rebuild_index()` scan list
+  the fact files without the storage locks, so a delete committed between the
+  listing and opening a file raised `MemoryStorageCorruption` for data that was
+  intact: the memory API returned HTTP 500 ("Stored memory data is corrupted"),
+  prompt injection dropped the whole memory block for that turn (or failed the
+  run under `failure_policy.read: fail_closed`), and a full index rebuild counted
+  the fact as failed. A fact that vanishes after the listing is now treated as
+  deleted; an entry that is still present but unreadable, such as a dangling
+  symlink, is still reported as corruption. ([#6255])
 - **memory:** A DeerMem memory reload no longer pins an older document in the
   cache. `reload()` read the document before computing its cache signature, so
   a write committed in between (for example by the background memory updater)
@@ -7687,3 +7697,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6229]: https://github.com/bytedance/deer-flow/pull/6229
 [#6230]: https://github.com/bytedance/deer-flow/pull/6230
 [#6238]: https://github.com/bytedance/deer-flow/pull/6238
+[#6255]: https://github.com/bytedance/deer-flow/pull/6255

CHANGELOG_zh.md

@@ -423,6 +423,13 @@
 
 ### 修复
 
+- **记忆：** 读取 DeerMem 智能体记忆时，不再因另一写入同时删除事实而失败。
+  `load()`、`reload()` 与全量 `rebuild_index()` 扫描在不持有存储锁的情况下列
+  出事实文件，若删除恰好在列出之后、打开文件之前提交，就会对完好的数据抛出
+  `MemoryStorageCorruption`：记忆 API 返回 HTTP 500（"Stored memory data is
+  corrupted"），提示词注入在该轮丢弃整个记忆块（在 `failure_policy.read:
+  fail_closed` 下则使运行失败），全量索引重建则把该事实计为失败。列出后消失的事实现在被视为已删除；仍然存在但无法读取的条目（例如悬空符
+  号链接）仍会报告为损坏。([#6255])
 - **记忆：** DeerMem 记忆重新加载不再把旧文档固定在缓存中。`reload()`
   此前先读取文档、后计算缓存签名，若两者之间有写入提交（例如后台记忆更
   新器），旧文档就会以新签名写入缓存，之后每次 `load()` 都返回过时的记
@@ -6420,3 +6427,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6229]: https://github.com/bytedance/deer-flow/pull/6229
 [#6230]: https://github.com/bytedance/deer-flow/pull/6230
 [#6238]: https://github.com/bytedance/deer-flow/pull/6238
+[#6255]: https://github.com/bytedance/deer-flow/pull/6255

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

@@ -135,6 +135,7 @@ Use the typed conflict classes instead of matching exception text.
 The weak lock cache must not retain inactive user scopes.
 Cache validation uses the manifest metadata and persisted revision.
 Unlocked `load()`/`reload()` compute that signature before reading the document.
+Unlocked fact scans skip entries deleted after listing; present unreadable entries are corruption.
 Out-of-band Markdown edits require `reload()`.
 POSIX atomic replacement must sync the parent directory.
 

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

@@ -410,6 +410,26 @@ def _parse_fact_markdown(path: Path) -> dict[str, Any]:
         raise MemoryStorageCorruption(f"Failed to parse canonical fact {path}: {exc}") from exc
 
 
+def _parse_listed_fact(path: Path) -> dict[str, Any] | None:
+    """Parse a fact found by a directory scan that ran without the scope locks.
+
+    A commit can delete a listed fact before it is opened; that vanished entry
+    returns ``None``. An entry that is still present but unreadable (such as a
+    dangling symlink) remains corruption.
+
+    The ``lexists`` re-check is a best-effort heuristic, not an exact test: a
+    delete followed by a recreate of the same id before the check still raises
+    (the conservative side), and an ``lstat`` failure on a parent directory
+    (such as ``EACCES``) reads as vanished, so that entry is skipped.
+    """
+    try:
+        return _parse_fact_markdown(path)
+    except MemoryStorageCorruption:
+        if os.path.lexists(path):
+            raise
+        return None
+
+
 def _fsync_parent_directory(directory: Path) -> None:
     """Make a completed rename durable on POSIX filesystems."""
     if os.name == "nt":
@@ -869,7 +889,12 @@ class FileMemoryStorage(MemoryStorage):
             return []
         facts: list[dict[str, Any]] = []
         for fact_path in sorted(agent_facts_directory(path, agent_name).glob("**/*.md")):
-            fact = _parse_fact_markdown(fact_path)
+            # Reached both unlocked (load()/reload()) and under the scope locks
+            # (save(), clear_all(), default-bucket migration); the vanished-entry
+            # skip only ever fires on the unlocked paths.
+            fact = _parse_listed_fact(fact_path)
+            if fact is None:
+                continue
             facts.append(self._validate_loaded_fact(fact, fact_path, user_id=user_id, agent_name=agent_name))
         # Shard directories are an internal layout detail and must not change
         # the stable fact order observed by callers.
@@ -1843,7 +1868,9 @@ class FileMemoryStorage(MemoryStorage):
             candidates = root.glob("**/facts/**/*.md")
             for path in candidates:
                 try:
-                    fact = _parse_fact_markdown(path)
+                    fact = _parse_listed_fact(path)
+                    if fact is None:
+                        continue
                     relative_parts = path.relative_to(root).parts
                     agents_index = relative_parts.index("agents")
                     expected_agent = relative_parts[agents_index + 1]

backend/tests/test_memory_storage_markdown.py

@@ -142,6 +142,72 @@ def test_cached_load_does_not_scan_all_fact_files(storage: FileMemoryStorage, mo
     assert storage.load("agent-a", user_id="alice")["facts"][0]["content"] == "Project uses Python 3.12"
 
 
+@pytest.mark.parametrize("read", ["load", "reload"])
+def test_unlocked_read_skips_fact_deleted_after_directory_listing(storage: FileMemoryStorage, monkeypatch: pytest.MonkeyPatch, read: str) -> None:
+    for fact_id in ("fact_keep", "fact_drop"):
+        storage.upsert_fact({"id": fact_id, "content": fact_id, "category": "context", "confidence": 0.9}, user_id="alice", agent_name="agent-a")
+    revision = storage.reload("agent-a", user_id="alice")["revision"]
+    storage._memory_cache.clear()
+    parse_fact_markdown = storage_module._parse_fact_markdown
+    raced = False
+
+    def parse_after_concurrent_delete(fact_path: Path) -> dict:
+        nonlocal raced
+        if fact_path.stem == "fact_drop" and not raced:
+            raced = True
+            # Another writer commits the delete after the unlocked read listed
+            # the fact directory but before it opened this file.
+            storage.delete_fact("fact_drop", user_id="alice", agent_name="agent-a")
+        return parse_fact_markdown(fact_path)
+
+    monkeypatch.setattr(storage_module, "_parse_fact_markdown", parse_after_concurrent_delete)
+    racing_read = getattr(storage, read)("agent-a", user_id="alice")
+    monkeypatch.setattr(storage_module, "_parse_fact_markdown", parse_fact_markdown)
+
+    assert raced
+    assert [fact["id"] for fact in racing_read["facts"]] == ["fact_keep"]
+    settled = storage.load("agent-a", user_id="alice")
+    assert [fact["id"] for fact in settled["facts"]] == ["fact_keep"]
+    assert settled["revision"] == revision + 1
+
+
+@pytest.mark.skipif(os.name == "nt", reason="creating symlinks needs privileges on Windows")
+def test_unlocked_read_still_rejects_dangling_fact_symlink(storage: FileMemoryStorage, tmp_path: Path) -> None:
+    storage.upsert_fact({"id": "fact_keep", "content": "kept", "category": "context", "confidence": 0.9}, user_id="alice", agent_name="agent-a")
+    memory_path = storage._get_memory_file_path("agent-a", user_id="alice")
+    dangling = fact_file_path(memory_path, "fact_dangling", agent_name="agent-a")
+    dangling.parent.mkdir(parents=True, exist_ok=True)
+    dangling.symlink_to(tmp_path / "missing.md")
+    storage._memory_cache.clear()
+
+    with pytest.raises(MemoryStorageCorruption, match="fact_dangling"):
+        storage.load("agent-a", user_id="alice")
+
+
+def test_deermem_read_survives_fact_deleted_by_another_worker(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    writer = DeerMem(backend_config={"storage_path": str(tmp_path)})
+    reader = DeerMem(backend_config={"storage_path": str(tmp_path)})
+    writer.create_fact("kept fact", user_id="alice")
+    _, deleted_id = writer.create_fact("deleted fact", user_id="alice")
+    parse_fact_markdown = storage_module._parse_fact_markdown
+    raced = False
+
+    def parse_after_concurrent_delete(fact_path: Path) -> dict:
+        nonlocal raced
+        if fact_path.stem == deleted_id and not raced:
+            raced = True
+            writer.delete_fact(deleted_id, user_id="alice")
+        return parse_fact_markdown(fact_path)
+
+    monkeypatch.setattr(storage_module, "_parse_fact_markdown", parse_after_concurrent_delete)
+    # Previously MemoryCorruptionError: a 500 from the memory API and an empty
+    # memory block in the prompt.
+    memory = reader.get_memory(user_id="alice")
+
+    assert raced
+    assert [fact["content"] for fact in memory["facts"]] == ["kept fact"]
+
+
 def test_shared_json_signature_invalidates_cache_after_other_storage_writes(tmp_path: Path) -> None:
     config = DeerMemConfig(storage_path=str(tmp_path))
     first = FileMemoryStorage(config)
@@ -1063,6 +1129,43 @@ def test_full_rebuild_index_accepts_original_email_user_scope(tmp_path: Path) ->
     assert retrieval.fact_ids == ["fact_01HZZZZZZZZZZZZZZZZZZZZZZZ"]
 
 
+def test_full_rebuild_index_skips_fact_deleted_after_directory_listing(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    class RecordingRetrieval:
+        def __init__(self) -> None:
+            self.fact_ids: list[str] = []
+
+        def upsert(self, fact, *, scope, path):
+            self.fact_ids.append(fact["id"])
+
+        def remove(self, fact_id, *, scope):
+            pass
+
+        def search(self, query, *, scopes, top_k, mode, filters):
+            return []
+
+    retrieval = RecordingRetrieval()
+    scoped = FileMemoryStorage(DeerMemConfig(storage_path=str(tmp_path)), retrieval=retrieval)
+    for fact_id in ("fact_keep", "fact_drop"):
+        scoped.upsert_fact({"id": fact_id, "content": fact_id, "category": "context", "confidence": 0.9}, user_id="alice", agent_name="agent-a")
+    retrieval.fact_ids.clear()
+    parse_fact_markdown = storage_module._parse_fact_markdown
+    raced = False
+
+    def parse_after_concurrent_delete(fact_path: Path) -> dict:
+        nonlocal raced
+        if fact_path.stem == "fact_drop" and not raced:
+            raced = True
+            scoped.delete_fact("fact_drop", user_id="alice", agent_name="agent-a")
+        return parse_fact_markdown(fact_path)
+
+    monkeypatch.setattr(storage_module, "_parse_fact_markdown", parse_after_concurrent_delete)
+    result = scoped.rebuild_index()
+
+    assert raced
+    assert result == {"supported": True, "indexed": 1, "failed": 0}
+    assert retrieval.fact_ids == ["fact_keep"]
+
+
 def test_scope_lock_cache_releases_unused_entries(storage: FileMemoryStorage) -> None:
     key = storage._cache_key("agent-a", user_id="alice")
     lock = storage._scope_lock(key)