diff --git a/CHANGELOG.md b/CHANGELOG.md index 3440b9f50a..8eaca55051 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 8b0c1ce0e6..8f438e6a7b 100644 --- a/CHANGELOG_zh.md +++ b/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 diff --git a/backend/packages/harness/deerflow/agents/memory/AGENTS.md b/backend/packages/harness/deerflow/agents/memory/AGENTS.md index e4f5f31c9e..549216e635 100644 --- a/backend/packages/harness/deerflow/agents/memory/AGENTS.md +++ b/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. diff --git a/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/storage.py b/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/storage.py index b41a731371..3f439c4f70 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/storage.py +++ b/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] diff --git a/backend/tests/test_memory_storage_markdown.py b/backend/tests/test_memory_storage_markdown.py index dd4c80af52..7295a7314d 100644 --- a/backend/tests/test_memory_storage_markdown.py +++ b/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)