From 09a98e5f7be4acf4956985d2e784cdee75513755 Mon Sep 17 00:00:00 2001 From: Cordis798 <2378285593@qq.com> Date: Mon, 26 Oct 2026 19:44:01 +0800 Subject: [PATCH] fix(memory): honor Mem0 read policies during context deduplication (#7561) Co-authored-by: Cordis798 <284918518+Cordis798@users.noreply.github.com> --- .../harness/deerflow/agents/memory/AGENTS.md | 2 + .../agents/memory/backends/mem0/README.md | 8 ++ .../memory/backends/mem0/mem0_manager.py | 31 ++-- backend/tests/test_mem0_context_record_ids.py | 134 ++++++++++++++++++ 4 files changed, 166 insertions(+), 9 deletions(-) create mode 100644 backend/tests/test_mem0_context_record_ids.py diff --git a/backend/packages/harness/deerflow/agents/memory/AGENTS.md b/backend/packages/harness/deerflow/agents/memory/AGENTS.md index d1e1152ae9..a23c3384d1 100644 --- a/backend/packages/harness/deerflow/agents/memory/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/memory/AGENTS.md @@ -334,6 +334,8 @@ Use the earliest source review deadline for its next review. #### Remote backends Mem0 search and management convert facts inside the read-policy boundary. +Context injection also deduplicates inside that boundary; unhashable record IDs +raise a value-redacted `Mem0APIError` (`tests/test_mem0_context_record_ids.py`). Invalid score/categories/metadata fields omit field values from `Mem0APIError`s. Diagnostics carry only bounded opaque record IDs, otherwise ``; missing/null defaults stay compatible and local programming errors propagate. diff --git a/backend/packages/harness/deerflow/agents/memory/backends/mem0/README.md b/backend/packages/harness/deerflow/agents/memory/backends/mem0/README.md index 09e44ee92d..b728dede03 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/mem0/README.md +++ b/backend/packages/harness/deerflow/agents/memory/backends/mem0/README.md @@ -71,3 +71,11 @@ not block unrelated ASGI handlers or SSE heartbeats. `failure_policy.read: fail_open` logs a recall failure and continues without new memory context. `fail_closed` propagates the backend error through prompt construction and aborts the run instead of silently degrading. + +Context-injection deduplication follows the same read policy when a remote +record has a list or object ID. The whole recall falls back to empty context +or raises `MemoryReadError`, including when the invalid ID appears on a later +page; no partial context is returned. Diagnostics identify the ID field without +echoing the invalid ID or memory content. Existing scalar and absent-ID +handling, first-occurrence ordering, and whole-entry character budgets remain +unchanged. diff --git a/backend/packages/harness/deerflow/agents/memory/backends/mem0/mem0_manager.py b/backend/packages/harness/deerflow/agents/memory/backends/mem0/mem0_manager.py index 41886c29b3..b328fd63bf 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/mem0/mem0_manager.py +++ b/backend/packages/harness/deerflow/agents/memory/backends/mem0/mem0_manager.py @@ -62,6 +62,22 @@ def _record_error(record: dict[str, Any], message: str) -> Mem0APIError: return Mem0APIError(f"{message} (record id={record_id})") +def _deduplicate_context_records(records: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Keep first occurrences, reporting unusable remote IDs as read failures.""" + seen: set[Any] = set() + unique: list[dict[str, Any]] = [] + for record in records: + record_id = record.get("id") + try: + if record_id in seen: + continue + seen.add(record_id) + except TypeError: + raise _record_error(record, "mem0 memory field id must be a scalar") from None + unique.append(record) + return unique + + def _to_fact(record: dict[str, Any]) -> dict[str, Any]: """Map a mem0 record to the backend-neutral fact shape consumed by the host (agents/memory/tools.py): id/content/category/confidence/createdAt/ @@ -230,22 +246,19 @@ class Mem0Manager(MemoryManager): top_k = self._config.top_k records = self._read_or_fallback( [], - lambda: self._client.list_memories( - filters=filters, - page_size=min(top_k, 200), - max_items=top_k, + lambda: _deduplicate_context_records( + self._client.list_memories( + filters=filters, + page_size=min(top_k, 200), + max_items=top_k, + ) ), ) budget = self._config.max_injection_chars - seen: set[str] = set() lines: list[str] = [] used = 0 shortest_line: int | None = None for record in records: - rid = record.get("id") - if rid in seen: - continue - seen.add(rid) text = str(record.get("memory") or "").strip() if not text: continue diff --git a/backend/tests/test_mem0_context_record_ids.py b/backend/tests/test_mem0_context_record_ids.py new file mode 100644 index 0000000000..aa4b9c9644 --- /dev/null +++ b/backend/tests/test_mem0_context_record_ids.py @@ -0,0 +1,134 @@ +"""Mem0 context deduplication failures must obey the configured read policy.""" + +from __future__ import annotations + +import asyncio +import copy +import json +from functools import partial +from types import SimpleNamespace + +import httpx +import pytest + +from deerflow.agents.memory.backends.mem0 import mem0_manager +from deerflow.agents.memory.backends.mem0.client import Mem0APIError, Mem0Client +from deerflow.agents.memory.manager import MemoryReadError + +PRIVATE = "synthetic-private-value" + + +def _read(manager, asynchronous): + scope = {"agent_name": "Research-Agent", "thread_id": "thread-1"} + if asynchronous: + return asyncio.run(manager.aget_context("user-1", **scope)) + return manager.get_context("user-1", **scope) + + +def _manager(monkeypatch, handler, policy="fail_open", *, budget=12000): + monkeypatch.setenv("MEM0_API_KEY", "synthetic-key") + monkeypatch.setattr(mem0_manager, "Mem0Client", partial(Mem0Client, transport=httpx.MockTransport(handler))) + return mem0_manager.Mem0Manager(backend_config={"failure_policy": {"read": policy}, "max_injection_chars": budget}) + + +@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize("policy", ["fail_open", "fail_closed"]) +@pytest.mark.parametrize("record_id", [[PRIVATE], {"private": PRIVATE}], ids=["list-id", "object-id"]) +@pytest.mark.parametrize("later_page", [False, True], ids=["same-page", "later-page"]) +def test_context_id_failure_obeys_read_policy(monkeypatch, caplog, asynchronous, policy, record_id, later_page): + requests = [] + good = {"id": "valid", "memory": PRIVATE} + bad = {"id": record_id, "memory": PRIVATE} + + def handle(request): + requests.append(request) + assert request.url.path == "/v3/memories/" + assert json.loads(request.content) == {"filters": {"AND": [{"user_id": "user-1"}, {"agent_id": "Research-Agent"}, {"run_id": "thread-1"}]}} + assert request.url.params["page_size"] == "8" + if later_page and request.url.params["page"] == "1": + return httpx.Response(200, json={"results": [good], "next": "page-2"}) + return httpx.Response(200, json={"results": [bad] if later_page else [good, bad]}) + + manager = _manager(monkeypatch, handle, policy) + try: + if policy == "fail_open": + assert _read(manager, asynchronous) == "" + assert "mem0 read failed" in caplog.text + assert "field id" in caplog.text + else: + with pytest.raises(MemoryReadError, match="field id") as error: + _read(manager, asynchronous) + assert isinstance(error.value.__cause__, Mem0APIError) + assert PRIVATE not in str(error.value) + assert error.value.__cause__.__cause__ is None + assert len(requests) == (2 if later_page else 1) + assert PRIVATE not in caplog.text + finally: + manager.close() + + +@pytest.mark.parametrize("record_id", ["m1", 17, 1.25, True, None, "missing"], ids=["string", "integer", "float", "boolean", "null", "missing"]) +def test_context_preserves_legacy_scalar_ids_and_entry_budget(monkeypatch, record_id): + first = {"id": record_id, "memory": "short"} + if record_id == "missing": + first.pop("id") + records = [first, dict(first, memory="duplicate"), {"id": "big", "memory": "x" * 30}, {"id": "last", "memory": "end"}] + before = copy.deepcopy(records) + requests = [] + + def handle(request): + requests.append(request) + return httpx.Response(200, json={"results": records}) + + manager = _manager(monkeypatch, handle, budget=15) + try: + assert _read(manager, False) == "- short\n- end" + assert len(requests) == 1 + assert records == before + finally: + manager.close() + + +def test_context_does_not_swallow_unrelated_programming_errors(monkeypatch): + manager = _manager(monkeypatch, lambda request: httpx.Response(200, json={"results": []})) + + def broken_list(**kwargs): + raise TypeError("local programming error") + + monkeypatch.setattr(manager._client, "list_memories", broken_list) + try: + with pytest.raises(TypeError, match="local programming error"): + _read(manager, False) + finally: + manager.close() + + +@pytest.mark.parametrize("policy", ["fail_open", "fail_closed"]) +@pytest.mark.parametrize("record_id", [[PRIVATE], {"private": PRIVATE}], ids=["list-id", "object-id"]) +def test_prompt_loader_preserves_read_policy_for_unusable_ids(monkeypatch, policy, record_id): + from deerflow.agents.lead_agent.prompt import _get_memory_context + + manager = _manager(monkeypatch, lambda request: httpx.Response(200, json={"results": [{"id": record_id, "memory": PRIVATE}]}), policy) + monkeypatch.setattr("deerflow.agents.memory.get_memory_manager", lambda: manager) + config = SimpleNamespace(memory=SimpleNamespace(enabled=True, injection_enabled=True, backend_config={"failure_policy": {"read": policy}})) + try: + if policy == "fail_open": + assert _get_memory_context("Research-Agent", app_config=config, user_id="user-1") == "" + else: + with pytest.raises(MemoryReadError, match="field id"): + _get_memory_context("Research-Agent", app_config=config, user_id="user-1") + finally: + manager.close() + + +@pytest.mark.parametrize("policy", ["fail_open", "fail_closed"]) +def test_prompt_loader_keeps_valid_memory_context(monkeypatch, policy): + from deerflow.agents.lead_agent.prompt import _get_memory_context + + manager = _manager(monkeypatch, lambda request: httpx.Response(200, json={"results": [{"id": "valid", "memory": "a valid fact"}]}), policy) + monkeypatch.setattr("deerflow.agents.memory.get_memory_manager", lambda: manager) + config = SimpleNamespace(memory=SimpleNamespace(enabled=True, injection_enabled=True, backend_config={"failure_policy": {"read": policy}})) + try: + assert _get_memory_context("Research-Agent", app_config=config, user_id="user-1") == "\n- a valid fact\n\n" + finally: + manager.close()