mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-11-12 22:20:33 +00:00
fix(memory): reject duplicate fact IDs during import (#8151)
* fix(memory): reject duplicate fact IDs during import * fix(memory): explain duplicate IDs in import response * docs(changelog): link duplicate import fix PR * fix(memory): keep duplicate validation portable --------- Co-authored-by: AochenShen99 <sac1063467113@hotmail.com> Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
b883deaab2
commit
b80191fad3
@ -257,6 +257,9 @@ This section accumulates work toward the **2.3.0** milestone
|
||||
group binding, which covers every group of the bot, starts a new one in
|
||||
each group. ([#7794])
|
||||
|
||||
- **memory:** DeerMem rejects backups with duplicate fact IDs before replacing
|
||||
saved memory, regardless of entry order. ([#8151])
|
||||
|
||||
- **memory:** Importing a backup without a `facts` array now returns HTTP 422
|
||||
and preserves saved memory. An explicit empty array still clears the selected
|
||||
fact bucket. ([#8131])
|
||||
@ -16325,3 +16328,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
|
||||
[#8118]: https://github.com/bytedance/deer-flow/pull/8118
|
||||
[#8126]: https://github.com/bytedance/deer-flow/pull/8126
|
||||
[#8131]: https://github.com/bytedance/deer-flow/pull/8131
|
||||
[#8151]: https://github.com/bytedance/deer-flow/pull/8151
|
||||
|
||||
@ -11,7 +11,7 @@ from app.gateway.authz import require_permission
|
||||
from app.gateway.internal_auth import get_trusted_internal_owner_user_id
|
||||
from app.gateway.persistent_writes import run_drained_write
|
||||
from app.gateway.secret_masking import mask_secret_values, normalize_config_key
|
||||
from deerflow.agents.memory import MemoryConflictError, MemoryCorruptionError, MemoryManager, get_memory_manager
|
||||
from deerflow.agents.memory import MemoryConflictError, MemoryCorruptionError, MemoryDuplicateFactIdError, MemoryManager, get_memory_manager
|
||||
from deerflow.config.agents_config import AGENT_NAME_PATTERN
|
||||
from deerflow.config.memory_config import get_memory_config
|
||||
from deerflow.config.paths import make_safe_user_id
|
||||
@ -596,6 +596,8 @@ async def import_memory(body: MemoryImportRequest, request: Request, agent_name:
|
||||
raise _unsupported_501(manager, "import memory") from None
|
||||
except (MemoryConflictError, MemoryCorruptionError) as exc:
|
||||
raise _map_memory_manager_error(exc) from exc
|
||||
except MemoryDuplicateFactIdError as exc:
|
||||
raise HTTPException(status_code=400, detail="Invalid memory import: fact IDs must be unique.") from exc
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=400, detail="Invalid memory import: facts must be a list of objects with non-empty content.") from exc
|
||||
except OSError as exc:
|
||||
|
||||
@ -15,6 +15,7 @@ them directly from ``deerflow.agents.memory.backends.deermem.deermem.core.*``.
|
||||
from deerflow.agents.memory.manager import (
|
||||
MemoryConflictError,
|
||||
MemoryCorruptionError,
|
||||
MemoryDuplicateFactIdError,
|
||||
MemoryManager,
|
||||
MemoryManagerError,
|
||||
MemoryReadError,
|
||||
@ -29,6 +30,7 @@ __all__ = [
|
||||
"MemoryReadError",
|
||||
"MemoryConflictError",
|
||||
"MemoryCorruptionError",
|
||||
"MemoryDuplicateFactIdError",
|
||||
"get_memory_manager",
|
||||
"memory_read_failures_are_fatal",
|
||||
"reset_memory_manager",
|
||||
|
||||
@ -29,7 +29,7 @@ from typing import Any, ClassVar, Literal
|
||||
|
||||
from pydantic import PrivateAttr
|
||||
|
||||
from deerflow.agents.memory.manager import MemoryConflictError, MemoryCorruptionError, MemoryManager
|
||||
from deerflow.agents.memory.manager import MemoryConflictError, MemoryCorruptionError, MemoryDuplicateFactIdError, MemoryManager
|
||||
|
||||
from .deermem.config import DeerMemConfig
|
||||
from .deermem.core.eviction import EVICTION_POLICY_HYBRID_V1
|
||||
@ -45,7 +45,7 @@ from .deermem.core.prompt import format_memory_for_injection, load_prompt, load_
|
||||
from .deermem.core.queue import MemoryUpdateQueue, QueueFull
|
||||
from .deermem.core.relevance import build_idf, order_facts_for_query, tokenize, warm_tokenizer
|
||||
from .deermem.core.storage import MemoryRevisionConflict, MemoryStorageCorruption, create_storage
|
||||
from .deermem.core.updater import MemoryUpdater, _coerce_source_confidence
|
||||
from .deermem.core.updater import DuplicateFactIdError, MemoryUpdater, _coerce_source_confidence
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@ -56,13 +56,15 @@ def _resolve_agent_name(agent_name: str | None) -> str:
|
||||
|
||||
|
||||
def _call_backend(operation):
|
||||
"""Translate DeerMem-private storage errors into the public manager contract."""
|
||||
"""Translate DeerMem-private errors into the public manager contract."""
|
||||
try:
|
||||
return operation()
|
||||
except MemoryRevisionConflict as exc:
|
||||
raise MemoryConflictError(str(exc)) from exc
|
||||
except MemoryStorageCorruption as exc:
|
||||
raise MemoryCorruptionError(str(exc)) from exc
|
||||
except DuplicateFactIdError as exc:
|
||||
raise MemoryDuplicateFactIdError(str(exc)) from exc
|
||||
|
||||
|
||||
def _legacy_source_value(source: Any) -> str:
|
||||
|
||||
@ -40,6 +40,10 @@ from .storage import (
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DuplicateFactIdError(ValueError):
|
||||
"""A backup contains more than one entry for a fact ID."""
|
||||
|
||||
|
||||
# Thread pool for offloading sync memory updates when called from an async
|
||||
# context. Unlike the previous asyncio.run() approach, this runs *sync*
|
||||
# model.invoke() calls — no event loop is created, so the langchain async
|
||||
@ -1000,6 +1004,14 @@ 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)
|
||||
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", []))
|
||||
|
||||
@ -113,6 +113,10 @@ class MemoryManagerError(RuntimeError):
|
||||
"""Backend-neutral base error exposed at the MemoryManager boundary."""
|
||||
|
||||
|
||||
class MemoryDuplicateFactIdError(ValueError):
|
||||
"""A memory import contains more than one record for the same fact ID."""
|
||||
|
||||
|
||||
class MemoryReadError(MemoryManagerError):
|
||||
"""A required memory read failed, so callers must not continue without it."""
|
||||
|
||||
|
||||
@ -361,7 +361,7 @@ def test_portability_only_abc_contract_imports_deerflow():
|
||||
deerflow_imports.append((p.relative_to(root).as_posix(), s))
|
||||
assert len(deerflow_imports) == 1, deerflow_imports
|
||||
assert deerflow_imports[0][0] == "deer_mem.py"
|
||||
assert "memory.manager import MemoryConflictError, MemoryCorruptionError, MemoryManager" in deerflow_imports[0][1]
|
||||
assert deerflow_imports[0][1].startswith("from deerflow.agents.memory.manager import ")
|
||||
|
||||
|
||||
# Minimal vendored host contract (what another agent would ship). DeerMem only
|
||||
@ -438,6 +438,7 @@ class MemoryManager(BaseModel):
|
||||
|
||||
class MemoryConflictError(RuntimeError): ...
|
||||
class MemoryCorruptionError(RuntimeError): ...
|
||||
class MemoryDuplicateFactIdError(ValueError): ...
|
||||
'''
|
||||
|
||||
|
||||
@ -462,11 +463,11 @@ def test_portability_vendor_to_other_agent(tmp_path, monkeypatch):
|
||||
# Repoint the single ABC-contract import line to the vendored manager.
|
||||
deer_mem_file = dst_pkg / "deer_mem.py"
|
||||
text = deer_mem_file.read_text(encoding="utf-8")
|
||||
contract_import = "from deerflow.agents.memory.manager import MemoryConflictError, MemoryCorruptionError, MemoryManager"
|
||||
contract_import = "from deerflow.agents.memory.manager import "
|
||||
assert contract_import in text
|
||||
text = text.replace(
|
||||
contract_import,
|
||||
"from otheragent.manager import MemoryConflictError, MemoryCorruptionError, MemoryManager",
|
||||
"from otheragent.manager import ",
|
||||
)
|
||||
deer_mem_file.write_text(text, encoding="utf-8")
|
||||
|
||||
@ -481,6 +482,11 @@ def test_portability_vendor_to_other_agent(tmp_path, monkeypatch):
|
||||
user_id="ua",
|
||||
)
|
||||
assert "y" in dm.get_context(user_id="ua")
|
||||
host_manager = importlib.import_module("otheragent.manager")
|
||||
backup = dm.get_memory(user_id="ua")
|
||||
backup["facts"] *= 2
|
||||
with pytest.raises(host_manager.MemoryDuplicateFactIdError, match="Duplicate fact id"):
|
||||
dm.import_memory(backup, user_id="ua")
|
||||
finally:
|
||||
for k in [k for k in list(sys.modules) if k.startswith("otheragent_deermem") or k == "otheragent"]:
|
||||
sys.modules.pop(k, None)
|
||||
|
||||
@ -25,6 +25,23 @@ def manager(tmp_path: Path) -> Iterator[DeerMem]:
|
||||
backend.close()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("agent_name", [None, "researcher"])
|
||||
@pytest.mark.parametrize("changed_first", [False, True], ids=["unchanged-first", "changed-first"])
|
||||
def test_import_rejects_duplicate_fact_ids_without_replacing_memory(manager: DeerMem, agent_name: str | None, changed_first: bool) -> None:
|
||||
scope = {"user_id": "alice", "agent_name": agent_name}
|
||||
manager.create_fact("Keep PostgreSQL", **scope)
|
||||
before = manager.get_memory(**scope)
|
||||
backup = copy.deepcopy(before)
|
||||
original = backup["facts"][0]
|
||||
changed = {**original, "content": "Use SQLite instead"}
|
||||
backup["facts"] = [changed, original] if changed_first else [original, changed]
|
||||
|
||||
with pytest.raises(ValueError, match="Duplicate fact id"):
|
||||
manager.import_memory(backup, **scope)
|
||||
|
||||
assert manager.get_memory(**scope) == before
|
||||
|
||||
|
||||
@pytest.mark.parametrize("agent_name", [None, "researcher"])
|
||||
@pytest.mark.parametrize("backup_revision", [1, None, 20], ids=["older", "omitted", "newer"])
|
||||
def test_import_restores_backup_with_current_fact_revision(manager: DeerMem, agent_name: str | None, backup_revision: int | None) -> None:
|
||||
|
||||
@ -253,6 +253,35 @@ def test_import_requires_explicit_facts_to_replace_saved_memory(tmp_path, agent_
|
||||
manager.close()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("agent_name", [None, "researcher"])
|
||||
@pytest.mark.parametrize("changed_first", [False, True])
|
||||
def test_import_reports_duplicate_fact_ids_without_replacing_memory(tmp_path, agent_name: str | None, changed_first: bool) -> None:
|
||||
app = make_authed_test_app()
|
||||
app.include_router(memory.router)
|
||||
manager = DeerMem(backend_config={"storage_path": str(tmp_path), "token_counting": "char"})
|
||||
params = {"agent_name": agent_name} if agent_name else {}
|
||||
try:
|
||||
with (
|
||||
patch("app.gateway.routers.memory.get_memory_manager", return_value=manager),
|
||||
patch("app.gateway.routers.memory.get_effective_user_id", return_value="alice"),
|
||||
TestClient(app) as client,
|
||||
):
|
||||
created = client.post("/api/memory/facts", params=params, json={"content": "Keep PostgreSQL"})
|
||||
assert created.status_code == 200
|
||||
before = client.get("/api/memory/export", params=params).json()
|
||||
original = before["facts"][0]
|
||||
changed = {**original, "content": "Use SQLite instead"}
|
||||
facts = [changed, original] if changed_first else [original, changed]
|
||||
|
||||
response = client.post("/api/memory/import", params=params, json={**before, "facts": facts})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.json()["detail"] == "Invalid memory import: fact IDs must be unique."
|
||||
assert client.get("/api/memory/export", params=params).json() == before
|
||||
finally:
|
||||
manager.close()
|
||||
|
||||
|
||||
def test_import_memory_route_preserves_source_error() -> None:
|
||||
app = make_authed_test_app()
|
||||
app.include_router(memory.router)
|
||||
|
||||
@ -162,6 +162,7 @@ DeerMem backup imports restore saved fact contents even after those facts have
|
||||
been edited, using the current stored revisions to protect the restore. Older
|
||||
backups may omit fact revisions. Backup revisions are validated before the
|
||||
importer substitutes current revisions. Rejected imports preserve stored memory.
|
||||
Backups with duplicate fact IDs are rejected before replacing the selected bucket.
|
||||
The import endpoint requires an explicit `facts` array. Omitting it returns
|
||||
HTTP 422, while `facts: []` explicitly clears the selected fact bucket.
|
||||
A concurrent write during import still returns HTTP 409 so the caller can reload
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user