fix(memory): reject duplicate facts inside the create critical section (#4599)

* fix(memory): reject duplicate facts inside the create critical section

memory_add's duplicate check ran outside the storage critical section
(the tools.py comment called this out): two concurrent tool calls for
the same user could both pass the check and both store the same fact.

Move authoritative duplicate rejection into
MemoryUpdater.create_memory_fact: the candidate's normalized content
key is checked against the fresh memory snapshot on every
revision-conflict retry, so the loser of a concurrent create reloads,
sees the winner's fact, and is rejected with ValueError("Duplicate
fact"). The tool-layer pre-check stays as a fast path.

The REST router now maps the duplicate ValueError to 409 with a clear
detail instead of the misleading "content cannot be empty" 400.

Tests: backend-level duplicate rejection, a simulated
concurrent-commit-during-conflict-retry race, and the router 409
mapping.

* fix(memory): retry legacy-path fact creation on save conflict

Wrap the legacy single-file save in the same bounded conflict-retry
loop as the apply_changes path: reload the fresh snapshot and re-run
_raise_if_duplicate_fact_content on every retry, so a concurrent
duplicate commit is rejected with ValueError("Duplicate fact") (tool
error / REST 409) instead of the generic OSError save failure
(REST 500). No-duplicate conflicts now retry and store instead of
failing the create.

Addresses review feedback on bytedance/deer-flow#4599.

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)

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."""

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"})
 

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)

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: