mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-11-12 22:20:33 +00:00
fix(memory): reject duplicate fact ids in imports regardless of entry order (#8153)
* fix(memory): reject duplicate fact ids in imports regardless of entry order A backup containing two entries for the same fact id was accepted or rejected depending only on their order. The duplicate check in _commit_changes_locked consulted the map of prepared writes, but an entry whose normalized content matches the stored fact is skipped from that map - so an unchanged entry followed by an edited duplicate passed the check while the reverse order raised, and the saved snapshot ended up with whichever entry came last. Track every incoming id in a seen-set for duplicate detection, before the unchanged-entry skip. The saved snapshot now survives the rejection in both orders. Fixes #8150 * fix(memory): validate duplicate import ids before capacity selection Copilot review follow-up on #8153: the storage-level duplicate guard runs after _select_for_capacity, and the capacity policy can drop one of two same-id entries (e.g. lower-confidence copies trimmed to max_facts) - the duplicate then never reaches the guard and the import outcome still depends on entry order. Validate all normalized incoming ids in import_memory_data before capacity selection. The storage-level guard stays as defense in depth. Regression test with max_facts=10 and 9 distinct high-confidence facts plus two same-id copies. * fix(memory): surface duplicate fact id errors with their own detail willem-bd review on #8153, three follow-ups: 1. P2: the bare ValueError fell through to the router's generic catch, so POST /api/memory/import answered with the facts-array message even when every fact had non-empty content. Add MemoryDuplicateFactIdError (same type as #8151) at the manager boundary and map it to its own 400 detail that names the duplicated id; router test asserts the exact detail. 2. P3: the duplicate check lived inside the apply_changes-override branch, so the base MemoryStorage save() path bypassed both guards. Move it before the branch split, right after normalize_memory_data, matching #8151's validate-before-capacity posture. 3. P3: parametrize the order-dependence regression over the default bucket (agent_name=None) and "researcher", and assert the seeded snapshot survives the rejection in both orders. * test(memory): pin the storage duplicate guard and fix the no-op assertion willem-bd review follow-ups on #8153: 1. The snapshot-survival assertion compared storage.load against get_memory_data, which is itself storage.load - both reads happened after the rejected import, so a regression that partially applied the duplicate would pass. Capture the seeded snapshot before the rejected import and compare against it. 2. The updater gate rejects import duplicates before storage is touched, so nothing exercised the storage-level guard in _commit_changes_locked with an unchanged duplicate first - reverting it would keep the suite green. Add a direct apply_changes regression test; verified it fails with the guard removed. * style(memory): apply ruff fixes after #8151 merge resolution * test(memory): assert untouched store in the storage-guard pin; facade import willem-bd review follow-ups on #8153: 1. The storage-guard regression test pinned rejection but not the untouched-store half: capture the seeded snapshot right after seeding and assert storage.load still equals it after the rejection, so a regression that partially applies the first entry before raising fails. 2. The router regression test imported MemoryDuplicateFactIdError from the manager module directly; use the package-facade import consistent with the other manager-error usage (the symbol is already re-exported from deerflow.agents.memory since #8151). * test(memory): drop the mock router duplicate-id test ShenAC-SAC review on #8153: #8151 already drives the real import route with duplicate IDs and asserts the same 400 response plus an unchanged snapshot, and the mock test's docstring implied a detail the assertion did not check. The direct apply_changes() regression is kept. Also removes the blank line after the docstring that tripped ruff format --check on the repo-pinned Ruff version. * test(memory): remove the orphaned mock-router test body ShenAC-SAC review: my earlier deletion removed the function definition and docstring but left the test body running inside test_clear_route_keeps_working_for_windows_reserved_agent_name. Remove the whole block and the now-unused MemoryDuplicateFactIdError import; #8151 already covers the duplicate-ID response through the real import route.
This commit is contained in:
parent
06634003a5
commit
eb97327ddf
@ -1347,14 +1347,21 @@ class FileMemoryStorage(MemoryStorage):
|
||||
|
||||
scope = _scope_dict(user_id, agent_name)
|
||||
prepared: dict[str, tuple[dict[str, Any], dict[str, Any] | None, Path]] = {}
|
||||
seen_ids: set[str] = set()
|
||||
for incoming in upserts:
|
||||
if not isinstance(incoming, dict):
|
||||
raise ValueError("change_set.upserts must contain fact objects")
|
||||
candidate = copy.deepcopy(incoming)
|
||||
candidate["id"] = str(candidate.get("id") or f"fact_{uuid.uuid4().hex}")
|
||||
fact_id = candidate["id"]
|
||||
if fact_id in prepared:
|
||||
# Duplicate detection must consider every incoming id, not just the
|
||||
# ones scheduled for a write: an unchanged entry (existing equals the
|
||||
# normalized candidate) is skipped from `prepared`, so a duplicate
|
||||
# after it would otherwise pass and the import outcome would depend
|
||||
# on the order of the two entries.
|
||||
if fact_id in seen_ids:
|
||||
raise ValueError(f"Duplicate fact id {fact_id!r} in upserts")
|
||||
seen_ids.add(fact_id)
|
||||
if agent_name is None: # guarded above
|
||||
raise ValueError("agent_name is required for fact repository changes")
|
||||
existing, fact_path = self._read_fact(path, fact_id, user_id=user_id, agent_name=agent_name)
|
||||
|
||||
@ -1004,14 +1004,15 @@ class MemoryUpdater:
|
||||
if isinstance(revision, bool) or not isinstance(revision, int) or revision < 1:
|
||||
raise ValueError("fact.revision must be an integer >= 1")
|
||||
memory_data = normalize_memory_data(memory_data)
|
||||
if agent_name is not None:
|
||||
# Validate before capacity selection can discard a conflicting entry.
|
||||
seen_ids: set[str] = set()
|
||||
for fact in memory_data["facts"]:
|
||||
fact_id = fact["id"]
|
||||
if fact_id in seen_ids:
|
||||
raise DuplicateFactIdError(f"Duplicate fact id {fact_id!r} in memory import")
|
||||
seen_ids.add(fact_id)
|
||||
# Duplicate detection must cover BOTH storage paths (the apply_changes
|
||||
# override and the base save() path) and run before capacity selection,
|
||||
# which can drop one of two same-id entries and hide the duplicate.
|
||||
seen_import_ids: set[str] = set()
|
||||
for fact in memory_data.get("facts", []):
|
||||
fact_id = str(fact.get("id") or f"fact_{uuid.uuid4().hex}")
|
||||
if fact_id in seen_import_ids:
|
||||
raise DuplicateFactIdError(f"Duplicate fact id {fact_id!r} in memory import")
|
||||
seen_import_ids.add(fact_id)
|
||||
if agent_name is not None and getattr(type(self._storage), "apply_changes", None) is not MemoryStorage.apply_changes:
|
||||
current = self.get_memory_data(agent_name, user_id=user_id)
|
||||
incoming_facts = copy.deepcopy(memory_data.get("facts", []))
|
||||
|
||||
@ -4,6 +4,7 @@ import json
|
||||
import logging
|
||||
import threading
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
@ -11,6 +12,7 @@ import pytest
|
||||
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 (
|
||||
FileMemoryStorage,
|
||||
MemoryManifestRevisionConflict,
|
||||
MemoryStorage,
|
||||
)
|
||||
@ -1967,7 +1969,6 @@ class TestOutOfRangeExtractionConfidenceDoesNotBlockUpdates:
|
||||
assert normalized["confidence"] == confidence
|
||||
|
||||
def test_update_memory_drops_the_bad_fact_and_keeps_the_rest(self, tmp_path) -> None:
|
||||
from deerflow.agents.memory.backends.deermem.deermem.core.storage import FileMemoryStorage
|
||||
|
||||
response = {
|
||||
"user": {},
|
||||
@ -2044,3 +2045,90 @@ class TestParseMemoryUpdateFactsToRemoveGate:
|
||||
except json.JSONDecodeError:
|
||||
return
|
||||
raise AssertionError('decoy object {"user": "alice"} must be rejected')
|
||||
|
||||
|
||||
def _fact(content: str, fact_id: str, revision: int = 1) -> dict[str, object]:
|
||||
return {
|
||||
"id": fact_id,
|
||||
"content": content,
|
||||
"category": "context",
|
||||
"confidence": 0.9,
|
||||
"createdAt": "2026-03-20T00:00:00Z",
|
||||
"source": "manual",
|
||||
"revision": revision,
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("agent_name", [None, "researcher"])
|
||||
def test_import_duplicate_id_rejected_regardless_of_entry_order(tmp_path: Path, agent_name: str | None) -> None:
|
||||
"""A backup with two entries for the same fact id must be rejected whether
|
||||
the first entry matches the stored fact or not - the outcome used to depend
|
||||
on entry order because unchanged entries were skipped from the write map.
|
||||
The seeded snapshot must survive the rejection in both orders."""
|
||||
from deerflow.agents.memory.backends.deermem.deermem.core.storage import DeerMemConfig
|
||||
from deerflow.agents.memory.backends.deermem.deermem.core.updater import DuplicateFactIdError
|
||||
|
||||
for order in ("unchanged-first", "edited-first"):
|
||||
config = DeerMemConfig(storage_path=str(tmp_path / order), retrieval_adapter="")
|
||||
storage = FileMemoryStorage(config)
|
||||
updater = MemoryUpdater(config, storage)
|
||||
seed = _make_memory(facts=[_fact("Keep PostgreSQL", "fact_dup")])
|
||||
updater.import_memory_data(seed, agent_name=agent_name)
|
||||
|
||||
unchanged = _fact("Keep PostgreSQL", "fact_dup")
|
||||
edited = _fact("Use SQLite instead", "fact_dup")
|
||||
facts = [unchanged, edited] if order == "unchanged-first" else [edited, unchanged]
|
||||
|
||||
snapshot_after_seed = updater.get_memory_data(agent_name=agent_name)
|
||||
|
||||
with pytest.raises(DuplicateFactIdError, match="Duplicate fact id"):
|
||||
updater.import_memory_data(_make_memory(facts=facts), agent_name=agent_name)
|
||||
|
||||
# The rejection must not have touched the seeded snapshot: compare
|
||||
# against the pre-import read, not a post-rejection one (which could
|
||||
# hide a partially applied duplicate).
|
||||
assert storage.load(agent_name=agent_name) == snapshot_after_seed
|
||||
|
||||
|
||||
def test_import_duplicate_id_rejected_before_capacity_selection(tmp_path: Path) -> None:
|
||||
"""Capacity selection runs before the storage guard and can drop one of two
|
||||
same-id entries, hiding the duplicate - validation must happen first
|
||||
(Copilot review on #8153)."""
|
||||
from deerflow.agents.memory.backends.deermem.deermem.core.updater import DuplicateFactIdError
|
||||
|
||||
config = DeerMemConfig(storage_path=str(tmp_path / "over-capacity"), retrieval_adapter="", max_facts=10)
|
||||
storage = FileMemoryStorage(config)
|
||||
updater = MemoryUpdater(config, storage)
|
||||
|
||||
facts = [_fact(f"distinct {i}", f"fact_{i}") for i in range(9)]
|
||||
facts.append(_fact("first copy", "fact_dup", revision=1))
|
||||
facts.append(_fact("second copy", "fact_dup", revision=1))
|
||||
assert len(facts) > 10 # over capacity
|
||||
|
||||
with pytest.raises(DuplicateFactIdError, match="Duplicate fact id"):
|
||||
updater.import_memory_data(_make_memory(facts=facts), agent_name="researcher")
|
||||
|
||||
|
||||
def test_storage_apply_changes_rejects_unchanged_duplicate_first(tmp_path: Path) -> None:
|
||||
"""Pin the storage-level guard directly: _commit_changes_locked must reject
|
||||
a duplicate whose first entry matches the stored fact (normalized equal, so
|
||||
it is skipped from the prepared writes) - the updater gate alone does not
|
||||
exercise this path."""
|
||||
from deerflow.agents.memory.backends.deermem.deermem.core.storage import DeerMemConfig, FileMemoryStorage
|
||||
|
||||
config = DeerMemConfig(storage_path=str(tmp_path / "storage-guard"), retrieval_adapter="")
|
||||
storage = FileMemoryStorage(config)
|
||||
updater = MemoryUpdater(config, storage)
|
||||
updater.import_memory_data(_make_memory(facts=[_fact("Keep PostgreSQL", "fact_dup")]), agent_name="researcher")
|
||||
expected = storage.load(agent_name="researcher")
|
||||
|
||||
unchanged = _fact("Keep PostgreSQL", "fact_dup", revision=1)
|
||||
edited = _fact("Use SQLite instead", "fact_dup", revision=1)
|
||||
with pytest.raises(ValueError, match="Duplicate fact id"):
|
||||
storage.apply_changes(
|
||||
{"upserts": [unchanged, edited]},
|
||||
user_id=None,
|
||||
agent_name="researcher",
|
||||
)
|
||||
# The rejection must not partially apply the first entry.
|
||||
assert storage.load(agent_name="researcher") == expected
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user