diff --git a/backend/app/gateway/routers/memory.py b/backend/app/gateway/routers/memory.py index 291983df2..2046ee546 100644 --- a/backend/app/gateway/routers/memory.py +++ b/backend/app/gateway/routers/memory.py @@ -116,6 +116,8 @@ def _map_memory_fact_value_error(exc: ValueError) -> HTTPException: 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." + elif exc.args and exc.args[0] == "Duplicate fact": + return HTTPException(status_code=409, detail="A fact with the same content already exists.") else: detail = "Memory fact content cannot be empty." return HTTPException(status_code=400, detail=detail) diff --git a/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/updater.py b/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/updater.py index 7f81c0e3e..11f5f5e45 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/updater.py +++ b/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/updater.py @@ -372,6 +372,20 @@ def _fact_content_key(content: Any) -> str | None: return stripped.casefold() +def _raise_if_duplicate_fact_content(memory_data: dict[str, Any], content_key: str | None) -> None: + """Reject a candidate fact whose normalized content already exists. + + Callers must invoke this against the freshest snapshot available inside + their read-check-write critical section (i.e. on every revision-conflict + retry), so two concurrent creators of the same content cannot both pass + the check and store duplicate facts.""" + if content_key is None: + return + for fact in memory_data.get("facts", []): + if isinstance(fact, dict) and _fact_content_key(fact.get("content")) == content_key: + raise ValueError("Duplicate fact") + + # ── Staleness review helpers ────────────────────────────────────────────── @@ -815,6 +829,13 @@ class MemoryUpdater: "added" status. This restores both the max_facts cap and the post-trim existence check (upstream's ``create_memory_fact_with_created_fact``), which the vendored copy had dropped together to avoid the dangling id. + + Duplicate rejection is enforced here (not only by callers): the + candidate's normalized content key is checked against the fresh + memory snapshot inside the revision-conflict retry loop of both + storage paths (apply_changes and legacy single-file save), so + concurrent creators cannot both store the same content. Raises + ``ValueError("Duplicate fact")`` on a normalized-content match. """ if agent_name is None: raise ValueError("agent_name") @@ -823,6 +844,7 @@ class MemoryUpdater: raise ValueError("content") normalized_category = category.strip() or "context" validated_confidence = _validate_confidence(confidence) + candidate_key = _fact_content_key(normalized_content) now = utc_now_iso_z() fact_id = f"fact_{uuid.uuid4().hex[:8]}" candidate = { @@ -836,6 +858,12 @@ class MemoryUpdater: 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) + # Duplicate rejection lives inside the conflict-retry loop so + # it is re-evaluated against the fresh snapshot after every + # revision conflict: two concurrent creators of the same + # content cannot both store it (the loser reloads, sees the + # winner's fact, and is rejected here). + _raise_if_duplicate_fact_content(memory_data, candidate_key) 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"]} @@ -860,15 +888,24 @@ class MemoryUpdater: 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) - 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". - stored = any(f.get("id") == fact_id for f in updated_memory["facts"]) - return updated_memory, (fact_id if stored else None) + # Legacy single-file path: same duplicate-rejection contract as the + # apply_changes path above. A revision-conflicted save (False) reloads + # the fresh snapshot and re-runs the duplicate check, so a concurrent + # creator's commit is rejected with ValueError("Duplicate fact") + # instead of surfacing as a generic save failure. + 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) + _raise_if_duplicate_fact_content(memory_data, candidate_key) + updated_memory = dict(memory_data) + updated_memory["facts"] = _trim_facts_to_max([*memory_data.get("facts", []), copy.deepcopy(candidate)], self._config.max_facts) + if self._save_memory_to_file(updated_memory, agent_name, user_id=user_id, expected_revision=int(memory_data.get("revision") or 0)): + # If the cap evicted the just-added (lower-confidence) fact, + # signal via None so callers don't report a dangling id as + # "added". + stored = any(f.get("id") == fact_id for f in updated_memory["facts"]) + return updated_memory, (fact_id if stored else None) + logger.info("Retrying capped fact creation from a fresh snapshot after a revision conflict") + raise OSError("Failed to save memory data after creating fact") 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.""" diff --git a/backend/packages/harness/deerflow/agents/memory/tools.py b/backend/packages/harness/deerflow/agents/memory/tools.py index 5be20b046..c32e2ff12 100644 --- a/backend/packages/harness/deerflow/agents/memory/tools.py +++ b/backend/packages/harness/deerflow/agents/memory/tools.py @@ -118,9 +118,11 @@ def memory_add_tool( content_key = _memory_content_key(normalized_content) manager = get_memory_manager() existing_facts = manager.get_memory(agent_name=agent_name, user_id=user_id).get("facts", []) - # Tool calls normally run one-at-a-time per user turn. If tool-mode - # writing broadens to multiple concurrent calls for the same user, - # move duplicate rejection into the storage/update critical section. + # Fast-path duplicate rejection to spare a write attempt in the common + # case. The authoritative check lives in the backend's create critical + # section (DeerMem re-checks against a fresh snapshot on every + # revision-conflict retry in create_memory_fact), so concurrent tool + # calls for the same user cannot both store the same content. if any(_memory_content_key(str(fact.get("content", ""))) == content_key for fact in existing_facts): return json.dumps({"error": "Duplicate fact"}) diff --git a/backend/tests/test_memory_router.py b/backend/tests/test_memory_router.py index 7d2e3bd25..9d7177283 100644 --- a/backend/tests/test_memory_router.py +++ b/backend/tests/test_memory_router.py @@ -213,6 +213,20 @@ def test_create_memory_fact_route_maps_conflict_to_409() -> None: assert response.json()["detail"] == "Memory changed concurrently; reload and retry." +def test_create_memory_fact_route_maps_duplicate_to_409() -> None: + app = FastAPI() + app.include_router(memory.router) + mock_mgr = MagicMock() + mock_mgr.create_fact.side_effect = ValueError("Duplicate fact") + + 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"] == "A fact with the same content already exists." + + def test_get_memory_route_maps_corruption_to_stable_500() -> None: app = FastAPI() app.include_router(memory.router) diff --git a/backend/tests/test_memory_updater.py b/backend/tests/test_memory_updater.py index c17e30f83..b6a41fb51 100644 --- a/backend/tests/test_memory_updater.py +++ b/backend/tests/test_memory_updater.py @@ -5,7 +5,10 @@ from unittest.mock import AsyncMock, MagicMock, patch from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemConfig from deerflow.agents.memory.backends.deermem.deermem.core.prompt import format_conversation_for_update -from deerflow.agents.memory.backends.deermem.deermem.core.storage import MemoryStorage +from deerflow.agents.memory.backends.deermem.deermem.core.storage import ( + MemoryManifestRevisionConflict, + MemoryStorage, +) from deerflow.agents.memory.backends.deermem.deermem.core.updater import ( MemoryUpdater, _build_staleness_section, @@ -444,6 +447,154 @@ def test_create_memory_fact_rejects_invalid_confidence() -> None: raise AssertionError("Expected ValueError for invalid fact confidence") +class _ConcurrentCommitStorage(_MemoryStorage): + """apply_changes stand-in that simulates a concurrent writer committing a + duplicate fact between the caller's snapshot read and its first apply. + + The first apply commits ``concurrent_fact`` (as a winning concurrent + writer would) and then raises a manifest revision conflict, forcing the + caller into its conflict-retry path with a fresh snapshot. Later applies + succeed and persist the upserts so test assertions can observe what the + caller actually stored. + """ + + def __init__(self, concurrent_fact: dict[str, object], memory: dict[str, object] | None = None): + super().__init__(memory) + self._concurrent_fact = concurrent_fact + self._conflict_injected = False + + def apply_changes(self, change_set, **scope): # noqa: ANN001, ANN201, ANN202 - test fake + if not self._conflict_injected: + self._conflict_injected = True + self.memory["facts"] = [*self.memory.get("facts", []), copy.deepcopy(self._concurrent_fact)] + self.memory["revision"] = int(self.memory.get("revision") or 0) + 1 + raise MemoryManifestRevisionConflict("simulated concurrent commit") + self.memory["facts"] = [*self.memory.get("facts", []), *change_set.get("upserts", [])] + self.memory["revision"] = int(self.memory.get("revision") or 0) + 1 + return {"complete": False} + + +def test_create_memory_fact_rejects_duplicate_content() -> None: + """Backend-level dedup: create_memory_fact itself must reject content that + already exists (normalized), not just the tool layer's pre-check.""" + existing = _make_memory( + facts=[ + { + "id": "fact_existing", + "content": "User prefers dark mode", + "category": "preference", + "confidence": 0.9, + "createdAt": "2026-03-18T00:00:00Z", + "source": "manual", + } + ] + ) + + updater = _make_updater(memory=existing) + try: + updater.create_memory_fact(content=" user prefers DARK mode ", agent_name="researcher") + except ValueError as exc: + assert exc.args == ("Duplicate fact",) + else: + raise AssertionError("Expected ValueError for duplicate fact content") + + +def test_create_memory_fact_rejects_duplicate_committed_during_conflict_retry() -> None: + """Race regression: a concurrent writer commits the same content between + this caller's snapshot read and its first apply. After the revision + conflict, the retry must detect the duplicate from the fresh snapshot + instead of storing a second copy.""" + concurrent_fact = { + "id": "fact_concurrent", + "content": "User prefers dark mode", + "category": "preference", + "confidence": 0.9, + "createdAt": "2026-03-18T00:00:00Z", + "source": "manual", + } + storage = _ConcurrentCommitStorage(concurrent_fact=concurrent_fact) + updater = _make_updater(storage=storage) + + try: + updater.create_memory_fact(content="user prefers DARK mode", agent_name="researcher") + except ValueError as exc: + assert exc.args == ("Duplicate fact",) + else: + raise AssertionError("Expected ValueError for duplicate fact content") + + assert [fact["id"] for fact in storage.memory["facts"]] == ["fact_concurrent"] + + +class _LegacyConcurrentCommitStorage(_MemoryStorage): + """Legacy-path stand-in (no apply_changes override) that simulates a + concurrent writer committing a fact between the caller's snapshot read + and its first save: the first save commits ``concurrent_fact`` and + returns False (revision conflict), later saves persist normally.""" + + def __init__(self, concurrent_fact: dict[str, object], memory: dict[str, object] | None = None): + super().__init__(memory) + self._concurrent_fact = concurrent_fact + self._conflict_injected = False + + def save(self, memory_data, agent_name=None, *, user_id=None, expected_revision=None): # noqa: ANN001, ANN201, ANN202 - test fake + self.save_calls.append((agent_name, user_id, expected_revision)) + if not self._conflict_injected: + self._conflict_injected = True + self.memory["facts"] = [*self.memory.get("facts", []), copy.deepcopy(self._concurrent_fact)] + self.memory["revision"] = int(self.memory.get("revision") or 0) + 1 + return False + self.memory = memory_data + return True + + +def test_create_memory_fact_legacy_path_rejects_duplicate_committed_during_save_conflict() -> None: + """Legacy single-file path race regression: a concurrent writer commits + the same content between this caller's snapshot read and its first save. + After the revision conflict, the retry must detect the duplicate from the + fresh snapshot and raise ValueError("Duplicate fact") instead of the + generic OSError save failure.""" + concurrent_fact = { + "id": "fact_concurrent", + "content": "User prefers dark mode", + "category": "preference", + "confidence": 0.9, + "createdAt": "2026-03-18T00:00:00Z", + "source": "manual", + } + storage = _LegacyConcurrentCommitStorage(concurrent_fact=concurrent_fact) + updater = _make_updater(storage=storage) + + try: + updater.create_memory_fact(content="user prefers DARK mode", agent_name="researcher") + except ValueError as exc: + assert exc.args == ("Duplicate fact",) + else: + raise AssertionError("Expected ValueError for duplicate fact content") + + assert [fact["id"] for fact in storage.memory["facts"]] == ["fact_concurrent"] + + +def test_create_memory_fact_legacy_path_retries_save_conflict_and_stores() -> None: + """Legacy single-file path: a save conflict without a duplicate reloads + the fresh snapshot and retries instead of failing the create.""" + concurrent_fact = { + "id": "fact_concurrent", + "content": "An unrelated concurrent fact", + "category": "context", + "confidence": 0.9, + "createdAt": "2026-03-18T00:00:00Z", + "source": "manual", + } + storage = _LegacyConcurrentCommitStorage(concurrent_fact=concurrent_fact) + updater = _make_updater(storage=storage) + + _, fact_id = updater.create_memory_fact(content="Brand new fact", agent_name="researcher") + + assert fact_id is not None + assert len(storage.save_calls) == 2 + assert [fact["id"] for fact in storage.memory["facts"]] == ["fact_concurrent", fact_id] + + def test_delete_memory_fact_raises_for_unknown_id() -> None: updater = _make_updater() try: