diff --git a/README.md b/README.md index 615ea74f28..5fc18f17ce 100644 --- a/README.md +++ b/README.md @@ -3612,6 +3612,13 @@ An individual Custom Agent can opt out of memory without changing the global set Single-fact repository operations are genuinely incremental: an upsert/delete reads, journals, writes, and re-indexes only the addressed fact files, and returns an explicit incomplete delta rather than a cache-dependent fake full document. Summary change sets merge the supplied `user`/`history` child keys over the persisted sections so a partial update cannot erase omitted siblings; full imports normalize both sections to the complete compatibility schema before applying replacement values. Imports reject malformed fact lists or blank/non-text content with HTTP 400 before changing stored memory; recoverable legacy metadata still receives defaults. An explicit empty fact list remains an intentional clear. Manager/API compatibility methods materialize a fresh full document only when their public response contract requires one. Fact-level point operations use separate expected user-memory and fact revisions and may explicitly rebase when every addressed fact precondition still holds. Snapshot-derived operations such as scoped clear, capped create, consolidation, and trimming never replay stale delete/trim sets: a manifest conflict reloads the complete document and recomputes the operation, with a bounded retry. Fact paths use the first two hexadecimal characters of `SHA-256(fact_id)` so generated `fact_*` IDs distribute across shards. The cache token combines the shared JSON's nanosecond mtime, size, and persisted revision; this prevents coarse-mtime same-size writes from returning stale data without scanning fact files. Direct out-of-band Markdown edits require an explicit reload. Storage-specific conflicts and corruption are translated at the MemoryManager boundary; the Gateway returns conflict as HTTP 409 and a stable, non-sensitive corruption error as HTTP 500. Full-document `save()` remains a compatibility API and computes a diff before writing; malformed or missing `facts` can no longer silently erase an agent's Markdown files. Legacy migration preserves non-empty `user`/`history` before deleting an agent `memory.json`; conflicting summaries keep the legacy file and fail loudly instead of choosing a winner. +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. +A concurrent write during import still returns HTTP 409 so the caller can reload +and retry. + Legacy facts in `memory.json` migrate automatically into the reserved `__default__` Markdown bucket on the user's first normal memory read. Operators who prefer to audit or complete the migration before serving traffic can run the optional idempotent CLI from `backend/`: ```bash diff --git a/backend/packages/harness/deerflow/agents/memory/AGENTS.md b/backend/packages/harness/deerflow/agents/memory/AGENTS.md index a1bd18fdfc..0e128ceb67 100644 --- a/backend/packages/harness/deerflow/agents/memory/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/memory/AGENTS.md @@ -189,6 +189,10 @@ Point operations can rebase only when all original fact preconditions still hold Snapshot operations must reload and recompute after a manifest conflict. Use the typed conflict classes instead of matching exception text. +Replacement imports copy facts, validate supplied revisions as positive integers, +and use live revisions for existing IDs. Omitted revisions stay valid. Preserve +manifest, upsert, absence and delete preconditions. + The weak lock cache must not retain inactive user scopes. Cache validation uses the manifest metadata and persisted revision. Unlocked `load()`/`reload()` compute that signature before reading the document. 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 5afc2ee3ba..86dd9f1ea0 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 @@ -1035,6 +1035,10 @@ class MemoryUpdater: raw_facts = memory_data.get("facts") if not isinstance(raw_facts, list) or any(not isinstance(fact, dict) or not isinstance(fact.get("content"), str) or not fact["content"].strip() for fact in raw_facts): raise ValueError("memory_data.facts must be a list of facts with non-empty content") + for fact in raw_facts: + revision = fact.get("revision", 1) + 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 and getattr(type(self._storage), "apply_changes", None) is not MemoryStorage.apply_changes: current = self.get_memory_data(agent_name, user_id=user_id) @@ -1050,10 +1054,17 @@ class MemoryUpdater: user_id=user_id, ) current_by_id = {str(fact.get("id")): fact for fact in current.get("facts", []) if isinstance(fact, dict)} - incoming_ids = {str(fact.get("id")) for fact in incoming_facts} + upsert_revisions = {fact["id"]: (int(current_by_id[fact["id"]].get("revision") or 1) if fact["id"] in current_by_id else None) for fact in incoming_facts} + # Backup revisions describe the exported snapshot; writes use the + # current snapshot's preconditions, including the fact payload. + for fact in incoming_facts: + current_revision = upsert_revisions[fact["id"]] + if current_revision is not None: + fact["revision"] = current_revision + incoming_ids = set(upsert_revisions) change_set = { "upserts": incoming_facts, - "upsertRevisions": {str(fact.get("id")): (int(current_by_id[str(fact.get("id"))].get("revision") or 1) if str(fact.get("id")) in current_by_id else None) for fact in incoming_facts}, + "upsertRevisions": upsert_revisions, "deletes": [fact_id for fact_id in current_by_id if fact_id not in incoming_ids], "deleteRevisions": {fact_id: int(fact.get("revision") or 1) for fact_id, fact in current_by_id.items() if fact_id not in incoming_ids}, } diff --git a/backend/tests/test_memory_import.py b/backend/tests/test_memory_import.py new file mode 100644 index 0000000000..7d2252988c --- /dev/null +++ b/backend/tests/test_memory_import.py @@ -0,0 +1,234 @@ +"""Backup restoration through DeerMem and its canonical file repository.""" + +import copy +from collections.abc import Iterator +from pathlib import Path + +import pytest + +from deerflow.agents.memory import MemoryConflictError +from deerflow.agents.memory.backends.deermem.deer_mem import DeerMem +from deerflow.agents.memory.backends.deermem.deermem.core.paths import fact_file_path +from deerflow.agents.memory.backends.deermem.deermem.core.storage import ( + MemoryFactRevisionConflict, + MemoryManifestRevisionConflict, + _render_fact_markdown, +) + + +@pytest.fixture +def manager(tmp_path: Path) -> Iterator[DeerMem]: + backend = DeerMem(backend_config={"storage_path": str(tmp_path), "token_counting": "char"}) + try: + yield backend + finally: + backend.close() + + +@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: + scope = {"user_id": "alice", "agent_name": agent_name} + initial, fact_id = manager.create_fact("Use PostgreSQL", **scope) + assert fact_id is not None + manager.import_memory(initial, user_id="alice", agent_name="other-agent") + manager.import_memory(initial, user_id="bob", agent_name=agent_name) + other_agent = manager.get_memory(user_id="alice", agent_name="other-agent")["facts"] + other_user = manager.get_memory(user_id="bob", agent_name=agent_name) + backup = manager.get_memory(**scope) + if backup_revision is None: + backup["facts"][0].pop("revision") + else: + backup["facts"][0]["revision"] = backup_revision + original = copy.deepcopy(backup) + + edited = manager.update_fact(fact_id, content="Use SQLite", **scope) + manager._storage.update_summaries( + { + "user": {"workContext": {"summary": "Current work", "updatedAt": "2026-10-01T00:00:00Z"}}, + "history": {"recentMonths": {"summary": "Current history", "updatedAt": "2026-10-01T00:00:00Z"}}, + }, + user_id="alice", + ) + before = manager.get_memory(**scope) + + restored = manager.import_memory(backup, **scope) + + assert restored["facts"][0]["id"] == fact_id + assert restored["facts"][0]["content"] == "Use PostgreSQL" + assert restored["facts"][0]["revision"] == edited["facts"][0]["revision"] + 1 + assert restored["facts"][0]["createdAt"] == initial["facts"][0]["createdAt"] + assert restored["facts"][0]["scope"] == {"userId": "alice", "agentName": agent_name or "__default__"} + expected_summaries = before if agent_name else backup + assert restored["user"] == expected_summaries["user"] + assert restored["history"] == expected_summaries["history"] + assert manager.reload_memory(**scope) == restored + assert manager.get_memory(user_id="alice", agent_name="other-agent")["facts"] == other_agent + assert manager.get_memory(user_id="bob", agent_name=agent_name) == other_user + assert backup == original + + +@pytest.mark.parametrize("agent_name", [None, "researcher"]) +def test_import_current_backup_is_idempotent(manager: DeerMem, agent_name: str | None) -> None: + scope = {"user_id": "alice", "agent_name": agent_name} + _, fact_id = manager.create_fact("Use PostgreSQL", **scope) + assert fact_id is not None + manager.update_fact(fact_id, content="Use SQLite", **scope) + backup = manager.get_memory(**scope) + original = copy.deepcopy(backup) + + assert manager.import_memory(backup, **scope) == original + assert manager.import_memory(backup, **scope) == original + assert manager.reload_memory(**scope) == original + assert backup == original + + +@pytest.mark.parametrize("agent_name", [None, "researcher"]) +@pytest.mark.parametrize("existing_id", [True, False], ids=["existing", "new"]) +@pytest.mark.parametrize("revision", [0, -1, True, False, 1.5, "1", None], ids=["zero", "negative", "true", "false", "float", "string", "null"]) +def test_import_rejects_invalid_backup_revision(manager: DeerMem, agent_name: str | None, existing_id: bool, revision: object) -> None: + scope = {"user_id": "alice", "agent_name": agent_name} + _, fact_id = manager.create_fact("Use PostgreSQL", **scope) + assert fact_id is not None + backup = manager.get_memory(**scope) + backup["facts"][0]["revision"] = revision + if not existing_id: + backup["facts"][0]["id"] = "new_fact" + original = copy.deepcopy(backup) + manager.update_fact(fact_id, content="Use SQLite", **scope) + manager.create_fact("Keep the post-backup fact", **scope) + before = manager.get_memory(**scope) + + with pytest.raises(ValueError, match=r"fact\.revision must be an integer >= 1"): + manager.import_memory(backup, **scope) + + assert manager.reload_memory(**scope) == before + assert backup == original + + +def test_import_validates_revisions_before_capacity_selection(manager: DeerMem) -> None: + scope = {"user_id": "alice"} + _, fact_id = manager.create_fact("Keep the current preference", **scope) + assert fact_id is not None + before = manager.get_memory(**scope) + backup = copy.deepcopy(before) + backup["facts"].append({"id": "invalid", "content": "Low-confidence fact", "confidence": 0.1, "revision": 0}) + manager._updater._config.max_facts = 1 + + with pytest.raises(ValueError, match=r"fact\.revision must be an integer >= 1"): + manager.import_memory(backup, **scope) + + assert manager.reload_memory(**scope) == before + + +@pytest.mark.parametrize("agent_name", [None, "researcher"]) +def test_import_recreates_deleted_fact_and_removes_post_backup_fact(manager: DeerMem, agent_name: str | None) -> None: + scope = {"user_id": "alice", "agent_name": agent_name} + _, fact_id = manager.create_fact("Use PostgreSQL", **scope) + assert fact_id is not None + manager.update_fact(fact_id, content="Use SQLite", **scope) + backup = manager.get_memory(**scope) + original = copy.deepcopy(backup) + manager.delete_fact(fact_id, **scope) + _, extra_id = manager.create_fact("Added after the backup", **scope) + assert extra_id is not None + + restored = manager.import_memory(backup, **scope) + + assert restored["facts"] == original["facts"] + assert manager.reload_memory(**scope)["facts"] == original["facts"] + assert backup == original + + +@pytest.mark.parametrize("agent_name", [None, "researcher"]) +@pytest.mark.parametrize("concurrent_change", ["update", "delete", "create", "summaries"]) +def test_import_rejects_concurrent_write(manager: DeerMem, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, agent_name: str | None, concurrent_change: str) -> None: + scope = {"user_id": "alice", "agent_name": agent_name} + _, fact_id = manager.create_fact("Use PostgreSQL", **scope) + assert fact_id is not None + backup = manager.get_memory(**scope) + backup["facts"].append({"id": "restored_fact", "content": "Another saved preference"}) + original = copy.deepcopy(backup) + manager.update_fact(fact_id, content="Use SQLite", **scope) + _, extra_id = manager.create_fact("Post-backup fact", **scope) + assert extra_id is not None + before = manager.get_memory(**scope) + peer = DeerMem(backend_config={"storage_path": str(tmp_path), "token_counting": "char"}) + real_apply = manager._storage.apply_changes + committed: list[dict] = [] + + def apply_after_concurrent_write(change_set: dict, **kwargs) -> dict: + assert not committed, "A replacement import must not replay a stale snapshot" + assert kwargs["expected_manifest_revision"] == before["revision"] + assert not kwargs.get("allow_manifest_rebase", False) + assert change_set["upsertRevisions"] == {fact_id: 2, "restored_fact": None} + assert change_set["deleteRevisions"] == {extra_id: 1} + assert change_set["deletes"] == [extra_id] + if concurrent_change == "update": + peer.update_fact(fact_id, content="Concurrent preference", **scope) + elif concurrent_change == "delete": + peer.delete_fact(fact_id, **scope) + elif concurrent_change == "create": + peer.create_fact("Concurrent new fact", **scope) + else: + peer._storage.update_summaries({"user": {"workContext": {"summary": "Concurrent work", "updatedAt": "now"}}}, user_id="alice") + committed.append(peer.get_memory(**scope)) + return real_apply(change_set, **kwargs) + + monkeypatch.setattr(manager._storage, "apply_changes", apply_after_concurrent_write) + try: + with pytest.raises(MemoryConflictError) as error: + manager.import_memory(backup, **scope) + + assert isinstance(error.value.__cause__, MemoryManifestRevisionConflict) + assert len(committed) == 1 + assert manager.reload_memory(**scope) == committed[0] + assert backup == original + finally: + peer.close() + + +@pytest.mark.parametrize("changed_fact", ["upsert", "delete", "create"]) +def test_import_keeps_fact_preconditions_when_manifest_is_unchanged(manager: DeerMem, monkeypatch: pytest.MonkeyPatch, changed_fact: str) -> None: + scope = {"user_id": "alice", "agent_name": "researcher"} + _, fact_id = manager.create_fact("Use PostgreSQL", **scope) + assert fact_id is not None + backup = manager.get_memory(**scope) + new_fact = {**copy.deepcopy(backup["facts"][0]), "id": "restored_fact", "content": "Saved preference"} + backup["facts"].append(new_fact) + original = copy.deepcopy(backup) + manager.update_fact(fact_id, content="Use SQLite", **scope) + _, extra_id = manager.create_fact("Post-backup fact", **scope) + assert extra_id is not None + before = manager.get_memory(**scope) + by_id = {fact["id"]: fact for fact in before["facts"]} + target = {"upsert": by_id[fact_id], "delete": by_id[extra_id], "create": new_fact}[changed_fact] + # Matching revisions isolate the expected-absence check for a new fact. + changed_revision = target["revision"] if changed_fact == "create" else target["revision"] + 1 + changed = {**copy.deepcopy(target), "content": "Concurrent file edit", "revision": changed_revision} + manifest = manager._storage._get_memory_file_path(**scope) + manifest_bytes = manifest.read_bytes() + path = fact_file_path(manifest, changed["id"], agent_name="researcher") + real_apply = manager._storage.apply_changes + observed: list[dict] = [] + + def apply_after_file_edit(change_set: dict, **kwargs) -> dict: + assert not observed + assert change_set["upsertRevisions"] == {fact_id: 2, "restored_fact": None} + assert change_set["deleteRevisions"] == {extra_id: 1} + # An out-of-band Markdown edit leaves the shared manifest unchanged. + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(_render_fact_markdown(changed)) + observed.append(manager.reload_memory(**scope)) + return real_apply(change_set, **kwargs) + + monkeypatch.setattr(manager._storage, "apply_changes", apply_after_file_edit) + + with pytest.raises(MemoryConflictError) as error: + manager.import_memory(backup, **scope) + + assert isinstance(error.value.__cause__, MemoryFactRevisionConflict) + assert len(observed) == 1 + assert manager.reload_memory(**scope) == observed[0] + assert manifest.read_bytes() == manifest_bytes + assert backup == original diff --git a/backend/tests/test_memory_router.py b/backend/tests/test_memory_router.py index f1a6333b55..65709b224a 100644 --- a/backend/tests/test_memory_router.py +++ b/backend/tests/test_memory_router.py @@ -145,6 +145,84 @@ def test_import_route_without_agent_name_persists_default_bucket_markdown(tmp_pa assert [path.stem for path in facts_root.glob("**/*.md")] == ["fact_gateway_import"] +@pytest.mark.parametrize("agent_name", [None, "researcher"]) +@pytest.mark.parametrize("backup_revision", [1, "omitted", None, True, "1", 1.0], ids=["older", "omitted", "null", "boolean", "string", "integral-float"]) +def test_import_route_restores_edited_fact_from_backup(tmp_path, agent_name: str | None, backup_revision: object) -> 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": "Use PostgreSQL", "confidence": 0.9}) + assert created.status_code == 200 + fact_id = created.json()["facts"][0]["id"] + exported = client.get("/api/memory/export", params=params) + assert exported.status_code == 200 + backup = exported.json() + if backup_revision == "omitted": + backup["facts"][0].pop("revision") + else: + backup["facts"][0]["revision"] = backup_revision + edited = client.patch(f"/api/memory/facts/{fact_id}", params=params, json={"content": "Use SQLite"}) + assert edited.status_code == 200 + + restored = client.post("/api/memory/import", params=params, json=backup) + + assert restored.status_code == 200, restored.text + assert restored.json()["facts"][0]["content"] == "Use PostgreSQL" + assert restored.json()["facts"][0]["id"] == fact_id + assert restored.json()["facts"][0]["revision"] == edited.json()["facts"][0]["revision"] + 1 + reloaded = client.post("/api/memory/reload", params=params) + assert reloaded.status_code == 200 + assert reloaded.json()["facts"] == restored.json()["facts"] + finally: + manager.close() + + +@pytest.mark.parametrize("agent_name", [None, "researcher"]) +@pytest.mark.parametrize("existing_id", [True, False], ids=["existing", "new"]) +@pytest.mark.parametrize("revision", [0, -1, False, "0"], ids=["zero", "negative", "false", "string-zero"]) +def test_import_route_rejects_invalid_revision_without_replacing_memory(tmp_path, agent_name: str | None, existing_id: bool, revision: object) -> 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": "Use PostgreSQL"}) + assert created.status_code == 200 + fact_id = created.json()["facts"][0]["id"] + exported = client.get("/api/memory/export", params=params) + assert exported.status_code == 200 + backup = exported.json() + backup["facts"][0]["revision"] = revision + if not existing_id: + backup["facts"][0]["id"] = "new_fact" + assert client.patch(f"/api/memory/facts/{fact_id}", params=params, json={"content": "Use SQLite"}).status_code == 200 + assert client.post("/api/memory/facts", params=params, json={"content": "Keep the post-backup fact"}).status_code == 200 + before = client.get("/api/memory/export", params=params) + assert before.status_code == 200 + + response = client.post("/api/memory/import", params=params, json=backup) + + assert response.status_code == 400, response.text + assert response.json()["detail"] == "Invalid memory import: facts must be a list of objects with non-empty content." + reloaded = client.post("/api/memory/reload", params=params) + assert reloaded.status_code == 200 + assert reloaded.json() == before.json() + finally: + manager.close() + + def test_import_memory_route_preserves_source_error() -> None: app = make_authed_test_app() app.include_router(memory.router)