"""Phase-2 (self-contained DeerMem) tests. Covers: DI construction (owns storage/updater/queue/llm), zero-config defaults, ``trace_id`` threading to the optional ``callbacks`` hook, langfuse being optional, ``hide_from_ui`` default-skip + hook-keep, empty ``storage_class`` (portable default), and portability -- ``backends/deermem/`` has exactly one ``from deerflow`` line (the ABC contract) and can be vendored into another agent by copying the folder and repointing that one line. Storage is isolated via ``$DEERMEM_DATA_DIR`` -> ``tmp_path``; the LLM is a fake injected onto the updater so no network is needed. """ from __future__ import annotations import sys from pathlib import Path import pytest from langchain_core.messages import AIMessage, HumanMessage from deerflow.agents.memory.backends.deermem.deer_mem import DeerMem from deerflow.agents.memory.backends.deermem.deermem.core.message_processing import ( filter_messages_for_memory, ) from deerflow.agents.memory.backends.deermem.deermem.core.storage import FileMemoryStorage from deerflow.agents.memory.backends.deermem.deermem.core.updater import _trim_facts_to_max from deerflow.agents.memory.manager import MemoryCallbacks @pytest.fixture def deermem_data_dir(tmp_path, monkeypatch): """Isolate DeerMem storage under tmp_path via $DEERMEM_DATA_DIR.""" d = tmp_path / "deermem_data" d.mkdir() monkeypatch.setenv("DEERMEM_DATA_DIR", str(d)) yield d class _FakeLLM: """Returns a fixed memory-update JSON so no real LLM/network is needed.""" def __init__(self, payload: str | None = None) -> None: self._payload = payload or '{"user":{},"history":{},"newFacts":[],"factsToRemove":[]}' def invoke(self, prompt, config=None): return type("R", (), {"content": self._payload})() def _deermem_with_fake_llm(backend_config=None, payload=None, callbacks=None) -> DeerMem: dm = DeerMem(backend_config=backend_config, callbacks=callbacks) fake = _FakeLLM(payload) dm._llm = fake dm._updater._llm = fake return dm def test_add_swallows_queue_full_so_backpressure_does_not_break_caller(deermem_data_dir, caplog) -> None: """Regression: QueueFull raised under backpressure is caught in DeerMem.add (the backend owns the queue, so it owns the degradation) so memory backpressure degrades to "update skipped" instead of propagating into MemoryMiddleware.after_agent and breaking the agent run -- peer middlewares self-guard the same way.""" import logging dm = _deermem_with_fake_llm(backend_config={"storage_path": str(deermem_data_dir), "queue_max_depth": 1}) # Stop the debounce timer so enqueued items stay pending (the cap persists # across the second add instead of being drained by a timer fire). dm._queue._schedule_timer = lambda *a, **k: None conv = [HumanMessage("Please explain quantum computing in detail"), AIMessage("Quantum computing uses qubits and superposition.")] # First add fills the queue to its depth cap (non-signal, new key). dm.add("thread-A", conv, agent_name="lead_agent", user_id="u") assert dm._queue.pending_count == 1 # Second add for a different key hits the cap -> QueueFull internally. It # must be caught: no exception escapes DeerMem.add. with caplog.at_level(logging.WARNING, logger="deerflow.agents.memory.backends.deermem.deer_mem"): dm.add("thread-B", conv, agent_name="lead_agent", user_id="u") assert "rejected under backpressure" in caplog.text # thread-B was rejected (not enqueued); only thread-A remains. assert dm._queue.pending_count == 1 def test_di_construction_owns_dependencies(): dm = DeerMem(backend_config={"max_facts": 50, "storage_path": "/tmp/x"}) assert dm._config.max_facts == 50 assert dm._storage is not None and dm._updater is not None and dm._queue is not None # dependencies are wired (DI), not globals: assert dm._updater._storage is dm._storage assert dm._queue._updater is dm._updater def test_from_config_keeps_backend_config_pure_of_injected_hooks(deermem_data_dir): """Host hooks (should_keep_hidden_message / trace_context_manager / host_llm) arrive as from_config kwargs and are parsed into DeerMemConfig (self._config, PrivateAttr); the instance's backend_config field stays the pure data the host passed (no callables / LLM), so it remains serializable and matches the README contract ("host hooks ... NOT in backend_config").""" def _keep(ak): return False trace_cm = object() # sentinel; trace_context_manager is typed Any fake_llm = _FakeLLM() dm = DeerMem.from_config( backend_config={"storage_path": str(deermem_data_dir), "max_facts": 20}, mode="middleware", should_keep_hidden_message=_keep, trace_context_manager=trace_cm, host_llm_factory=lambda: fake_llm, callbacks=None, ) # hooks reached DeerMemConfig (PrivateAttr) -- wired, not lost assert dm._config.should_keep_hidden_message is _keep assert dm._config.trace_context_manager is trace_cm assert dm._config.host_llm is fake_llm # backend_config field is the pure original data (no injected hooks) assert dm.backend_config == {"storage_path": str(deermem_data_dir), "max_facts": 20} assert "should_keep_hidden_message" not in dm.backend_config assert "trace_context_manager" not in dm.backend_config assert "host_llm" not in dm.backend_config def test_zero_config_defaults_run_non_llm_ops(deermem_data_dir): dm = DeerMem(backend_config=None) # zero config assert dm._llm is None # no model -> no LLM dm.import_memory( {"version": "1.0", "lastUpdated": "", "user": {}, "history": {}, "facts": [{"id": "f", "content": "x", "category": "c", "confidence": 0.5, "createdAt": "", "source": "m"}]}, user_id="u", ) assert "x" in dm.get_context(user_id="u") assert dm.get_memory(user_id="u")["facts"][0]["content"] == "x" def test_import_without_agent_name_persists_facts_in_default_markdown_bucket(deermem_data_dir): dm = DeerMem(backend_config=None) dm.import_memory( { "user": {}, "history": {}, "facts": [ { "id": "fact_default_import", "content": "imported through the default manager scope", "category": "context", "confidence": 0.8, "source": "import", } ], }, user_id="alice", ) assert [fact["id"] for fact in dm.get_memory(user_id="alice")["facts"]] == ["fact_default_import"] facts_root = deermem_data_dir / "users" / "alice" / "agents" / "__default__" / "facts" assert [path.stem for path in facts_root.glob("**/*.md")] == ["fact_default_import"] def test_import_empty_summary_sections_replace_existing_summaries_with_complete_defaults(deermem_data_dir): dm = DeerMem(backend_config=None) existing = dm.get_memory(user_id="alice") existing["user"]["workContext"] = {"summary": "old work", "updatedAt": "old"} existing["user"]["personalContext"] = {"summary": "old personal", "updatedAt": "old"} existing["history"]["recentMonths"] = {"summary": "old history", "updatedAt": "old"} dm.import_memory(existing, user_id="alice") imported = dm.import_memory({"user": {}, "history": {}, "facts": []}, user_id="alice") assert imported["user"] == { "workContext": {"summary": "", "updatedAt": ""}, "personalContext": {"summary": "", "updatedAt": ""}, "topOfMind": {"summary": "", "updatedAt": ""}, } assert imported["history"] == { "recentMonths": {"summary": "", "updatedAt": ""}, "earlierContext": {"summary": "", "updatedAt": ""}, "longTermBackground": {"summary": "", "updatedAt": ""}, } def test_trace_id_threads_through_to_callbacks(deermem_data_dir): """trace_id reaches the pre-LLM-call callbacks hook (on_memory_llm_call).""" calls = [] class _RecordingCallbacks(MemoryCallbacks): def on_memory_llm_call(self, invoke_config, *, thread_id, user_id, trace_id, model_name): calls.append((thread_id, trace_id, model_name)) dm = _deermem_with_fake_llm({"model": {"provider": "openai", "model": "gpt-x", "api_key": "k", "base_url": "u"}}, callbacks=_RecordingCallbacks()) dm.add( thread_id="t1", messages=[HumanMessage(content="hi"), AIMessage(content="hello")], agent_name=None, user_id="u1", trace_id="trace-42", ) dm._queue.flush() assert calls and calls[0] == ("t1", "trace-42", "gpt-x") def test_default_passive_update_persists_fact_in_reserved_default_bucket(deermem_data_dir): dm = _deermem_with_fake_llm(payload='{"user":{},"history":{},"newFacts":[{"content":"Default agent fact","category":"context","confidence":0.9}],"factsToRemove":[]}') dm.add( thread_id="default-thread", messages=[HumanMessage(content="remember this"), AIMessage(content="understood")], user_id="alice", ) dm._queue.flush() assert [fact["content"] for fact in dm.get_memory(user_id="alice")["facts"]] == ["Default agent fact"] facts_root = deermem_data_dir / "users" / "alice" / "agents" / "__default__" / "facts" assert list(facts_root.glob("**/*.md")) def test_clear_all_memory_removes_global_summaries_and_every_agent_fact(deermem_data_dir): dm = DeerMem() imported = dm.get_memory(user_id="alice") imported["user"]["workContext"] = {"summary": "shared profile", "updatedAt": "now"} imported["facts"] = [ { "id": "fact_default", "content": "default fact", "category": "context", "confidence": 0.9, "createdAt": "2026-01-01T00:00:00Z", "source": "manual", } ] dm.import_memory(imported, user_id="alice") dm.create_fact("custom fact", agent_name="custom-agent", user_id="alice") custom_dir = deermem_data_dir / "users" / "alice" / "agents" / "custom-agent" config_path = custom_dir / "config.yaml" config_path.write_text("name: custom-agent\n", encoding="utf-8") cleared = dm.clear_memory(user_id="alice") assert cleared["user"]["workContext"]["summary"] == "" assert cleared["facts"] == [] assert dm.get_memory(agent_name="custom-agent", user_id="alice")["facts"] == [] assert config_path.read_text(encoding="utf-8") == "name: custom-agent\n" def test_scoped_clear_preserves_shared_summaries(deermem_data_dir): dm = DeerMem() imported = dm.get_memory(user_id="alice") imported["user"]["workContext"] = {"summary": "shared profile", "updatedAt": "now"} dm.import_memory(imported, user_id="alice") dm.create_fact("custom fact", agent_name="custom-agent", user_id="alice") cleared = dm.clear_memory(agent_name="custom-agent", user_id="alice") assert cleared["facts"] == [] assert cleared["user"]["workContext"]["summary"] == "shared profile" assert dm.get_memory(user_id="alice")["user"]["workContext"]["summary"] == "shared profile" def test_callbacks_optional_no_langfuse(deermem_data_dir): dm = _deermem_with_fake_llm({"model": {"provider": "openai", "model": "gpt-x", "api_key": "k", "base_url": "u"}}) assert dm.callbacks is None # no callbacks = no langfuse (not hard-required) dm.add( thread_id="t2", messages=[HumanMessage(content="hi"), AIMessage(content="hello")], agent_name=None, user_id="u2", trace_id="t-99", ) dm._queue.flush() # no callbacks, no error, update completes def test_hide_from_ui_default_skip_hook_keeps(): hidden = HumanMessage(content="secret", additional_kwargs={"hide_from_ui": True}) normal = HumanMessage(content="hi") ai = AIMessage(content="hello") # default (no hook) -> hide_from_ui skipped assert hidden not in filter_messages_for_memory([hidden, normal, ai]) # hook returns True -> hidden kept assert hidden in filter_messages_for_memory([hidden, normal, ai], should_keep_hidden_message=lambda ak: True) def test_storage_class_empty_uses_filememorystorage(): # empty storage_class (default) -> FileMemoryStorage directly, no importlib (portable, zero noise) dm = DeerMem(backend_config=None) assert dm._config.storage_class == "" assert isinstance(dm._storage, FileMemoryStorage) def test_portability_only_abc_contract_imports_deerflow(): """backends/deermem/ has exactly ONE `from deerflow` line: the ABC contract in deer_mem.py.""" import deerflow.agents.memory.backends.deermem as pkg root = Path(pkg.__file__).parent deerflow_imports = [] for p in root.rglob("*.py"): for line in p.read_text(encoding="utf-8").splitlines(): s = line.strip() if s.startswith("from deerflow") or s.startswith("import 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] # Minimal vendored host contract (what another agent would ship). DeerMem only # needs this ABC -- nothing else from a host. _VENDORED_MANAGER_PY = ''' """Vendored host contract (pydantic BaseModel + three-tier ABC) for the portability demo.""" from abc import abstractmethod from typing import Any, ClassVar, Literal from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator class MemoryManager(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True) backend_config: dict[str, Any] = Field(default_factory=dict) mode: Literal["middleware", "tool"] = "middleware" callbacks: Any = None supports_search: ClassVar[bool] = False @field_validator("backend_config", mode="before") @classmethod def _coerce_backend_config(cls, value): return value or {} @model_validator(mode="after") def _check_invariants(self): if self.mode == "tool" and not type(self).supports_search: raise ValueError("tool mode requires search") return self # Tier 1: abstract (every backend must implement). @abstractmethod def add(self, thread_id, messages, *, agent_name=None, user_id=None, trace_id=None) -> None: ... @abstractmethod def get_context(self, user_id, *, agent_name=None, thread_id=None) -> str: ... @classmethod @abstractmethod def from_config(cls, backend_config, *, mode="middleware", **host_hooks): ... # Tier 2: management defaults (override if supported). def add_nowait(self, thread_id, messages, *, agent_name=None, user_id=None) -> None: self.add(thread_id, messages, agent_name=agent_name, user_id=user_id) def search(self, query, top_k=5, *, user_id=None, agent_name=None, category=None) -> list: raise NotImplementedError def get_memory(self, *, user_id=None, agent_name=None) -> dict: raise NotImplementedError def delete_memory(self, *, user_id=None, agent_name=None) -> None: raise NotImplementedError def clear_memory(self, *, user_id=None, agent_name=None) -> dict: raise NotImplementedError def import_memory(self, memory_data, *, user_id=None, agent_name=None) -> dict: raise NotImplementedError def export_memory(self, *, user_id=None, agent_name=None) -> dict: raise NotImplementedError def shutdown_flush(self, timeout) -> bool: return True # Tier 3: optional hooks (override if supported). def warm(self) -> bool | None: return None def reload_memory(self, *, user_id=None, agent_name=None) -> dict: raise NotImplementedError def create_fact(self, content, category="context", confidence=0.5, *, agent_name=None, user_id=None): raise NotImplementedError def delete_fact(self, fact_id, *, agent_name=None, user_id=None) -> dict: raise NotImplementedError def update_fact(self, fact_id, content=None, category=None, confidence=None, *, agent_name=None, user_id=None) -> dict: raise NotImplementedError def on_pre_compress(self, messages) -> str: return "" def on_turn_start(self, turn_number, message, **kwargs) -> None: return None class MemoryConflictError(RuntimeError): ... class MemoryCorruptionError(RuntimeError): ... ''' def test_portability_vendor_to_other_agent(tmp_path, monkeypatch): """Copy backends/deermem/ into a temp package, repoint the ONE ABC import to a vendored manager, import, and run a round-trip -- proves copy + 1-line + run portability (zero deerflow dependency at runtime).""" import importlib import shutil import deerflow.agents.memory.backends.deermem as pkg src = Path(pkg.__file__).parent # Vendored host package with a minimal manager.py (the contract). host_pkg = tmp_path / "otheragent" host_pkg.mkdir() (host_pkg / "__init__.py").write_text("", encoding="utf-8") (host_pkg / "manager.py").write_text(_VENDORED_MANAGER_PY, encoding="utf-8") # Copy the DeerMem backend folder. dst_pkg = tmp_path / "otheragent_deermem" shutil.copytree(src, dst_pkg) # 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" assert contract_import in text text = text.replace( contract_import, "from otheragent.manager import MemoryConflictError, MemoryCorruptionError, MemoryManager", ) deer_mem_file.write_text(text, encoding="utf-8") monkeypatch.setenv("DEERMEM_DATA_DIR", str(tmp_path / "data")) monkeypatch.syspath_prepend(str(tmp_path)) try: mod = importlib.import_module("otheragent_deermem.deer_mem") assert hasattr(mod, "DeerMem") dm = mod.DeerMem(backend_config=None) # zero config, self._llm=None dm.import_memory( {"version": "1.0", "lastUpdated": "", "user": {}, "history": {}, "facts": [{"id": "f", "content": "y", "category": "c", "confidence": 0.5, "createdAt": "", "source": "m"}]}, user_id="ua", ) assert "y" in dm.get_context(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) def test_per_user_memory_path_matches_host_safe_user_id(deermem_data_dir): """Pin the per-user memory path across the abstraction. DeerMem writes memory to ``{storage_path}/users/{safe_user_id}/memory.json`` where ``safe_user_id`` is byte-identical to the host's ``make_safe_user_id``. The factory injects ``runtime_home()`` (= base_dir) as ``storage_path``, so the on-disk path is ``{base_dir}/users/{uid}/memory.json`` -- identical to pre-abstraction. This locks that equivalence so a future change to DeerMem's path / safe_user_id logic can't silently orphan existing per-user memory (risk:high, persistent state). """ from deerflow.config.paths import make_safe_user_id user_id = "test-user-123@example.com" # storage_path mirrors what the host factory injects (runtime_home / base_dir) dm = DeerMem(backend_config={"storage_path": str(deermem_data_dir)}) dm.create_fact("User prefers concise answers", category="preference", agent_name="default", user_id=user_id) expected_safe = make_safe_user_id(user_id) expected_file = deermem_data_dir / "users" / expected_safe / "memory.json" assert expected_file.is_file(), f"memory not at expected per-user path: {expected_file}" # DeerMem used the host-identical safe_user_id (not some other encoding). user_dirs = [p.name for p in (deermem_data_dir / "users").iterdir() if p.is_dir()] assert user_dirs == [expected_safe], f"safe_user_id diverged from host: {user_dirs}" def test_trim_facts_to_max_coerces_non_float_confidence(): """Non-float stored confidence must not crash the max_facts trim sort. Regression: the vendored copy used ``key=lambda f: f.get("confidence", 0)`` which raised TypeError comparing None/str against float once ``len > max_facts`` (legacy / imported facts with abnormal confidence). This is the #4034 intent that the module-skipped test files never exercised against the vendored updater; pinning it here so the rename can't silently drop the coercion again. """ facts = [ {"id": "a", "confidence": None}, {"id": "b", "confidence": "0.9"}, # numeric string {"id": "c", "confidence": 0.8}, {"id": "d", "confidence": "high"}, # non-numeric ] # No TypeError; coerced ranking: b("0.9"->0.9) > c(0.8) > a(None->0.5)=d("high"->0.5). kept = _trim_facts_to_max(facts, max_facts=2) assert [f["id"] for f in kept] == ["b", "c"] # Below the cap -> returned unchanged (no sort, no crash). assert _trim_facts_to_max(facts, max_facts=10) == facts def test_create_fact_trims_to_max_and_signals_eviction(deermem_data_dir): """create_fact enforces max_facts and signals eviction via None fact_id. Regression: the vendored ``create_memory_fact`` only appended (no trim), so manual / tool adds could grow memory past max_facts. Now it trims (highest confidence wins) and returns ``None`` when the cap evicts the new fact, so the tool reports "not stored" instead of a dangling id + false "added". """ # DeerMemConfig enforces max_facts >= 10, so fill the cap with 10 high-conf facts. dm = DeerMem(backend_config={"max_facts": 10, "storage_path": str(deermem_data_dir)}) for i in range(10): _, fid = dm.create_fact(f"high{i}", category="context", confidence=0.9, user_id="u1") assert fid is not None # Cap is full (10 facts); a lower-confidence 11th is evicted, not stored. memory_data, evicted_id = dm.create_fact("low_evicted", category="context", confidence=0.1, user_id="u1") assert evicted_id is None assert "low_evicted" not in {f["content"] for f in memory_data["facts"]} assert len(memory_data["facts"]) == 10 def test_search_survives_non_float_confidence(deermem_data_dir): """DeerMem.search ranks by _coerce_source_confidence, so non-float stored confidence (null / string / non-numeric, reachable via import / legacy) must not crash the sort. Re-adds the regression guard deleted with the monolithic test_search_memory_facts_sort_survives_non_float_stored_confidence.""" dm = DeerMem(backend_config={"storage_path": str(deermem_data_dir)}) # create_fact validates confidence to float, so seed non-float via import # (simulating imported / legacy data that bypasses _validate_confidence). dm.import_memory( { "user": {}, "history": {}, "facts": [ {"id": "a", "content": "alpha matching query", "confidence": None}, {"id": "b", "content": "bravo matching query", "confidence": "0.9"}, {"id": "c", "content": "charlie matching query", "confidence": "high"}, ], }, user_id="u1", ) results = dm.search("query", top_k=10, user_id="u1") # No TypeError; all three match "query"; ranked by coerced confidence desc: # b("0.9"->0.9) > a(None->0.5)=c("high"->0.5), stable so a before c. assert [r["id"] for r in results] == ["b", "a", "c"] def test_is_human_clarification_response_matches_host_read(): """The standalone mirror must agree with the host's read_human_input_response so hidden-message filtering doesn't diverge between production (host hook) and standalone / test (mirror default). Pins drift (#5).""" from deerflow.agents.human_input import read_human_input_response from deerflow.agents.memory.backends.deermem.deermem.core.message_processing import _is_human_clarification_response def payload(**overrides): base = {"version": 1, "kind": "human_input_response", "source": "s", "request_id": "r", "value": "v", "response_kind": "text"} base.update(overrides) return {"human_input_response": base} cases = [ {}, {"human_input_response": {}}, payload(), # valid text response payload(response_kind="option", option_id="o1"), # valid option response payload(response_kind="option"), # option without option_id -> not valid payload(value=""), # empty value -> not valid payload(source=""), # empty source -> not valid payload(version=2), # wrong version -> not valid payload(kind="other"), # wrong kind -> not valid {"human_input_response": "not a mapping"}, {"other_key": 1}, # no human_input_response key ] for ak in cases: host_keeps = read_human_input_response(ak) is not None mirror_keeps = _is_human_clarification_response(ak) assert host_keeps == mirror_keeps, f"divergence on {ak!r}: host={host_keeps} mirror={mirror_keeps}" def test_build_llm_returns_none_when_no_model_configured(): """Zero-config (no model_config, or model_config with no model) -> None. Non-LLM ops still work; an update raises at runtime.""" from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemModelConfig from deerflow.agents.memory.backends.deermem.deermem.core.llm import build_llm assert build_llm(None) is None assert build_llm(DeerMemModelConfig()) is None # model=None default def test_build_llm_degrades_to_none_on_init_failure(caplog): """build_llm degrades to None (with a WARNING) when init_chat_model fails, mirroring _host_default_llm -- so a misconfigured explicit ``model`` does NOT crash app startup. Memory CRUD/read/search still work; extraction is disabled; an update raises at runtime with the underlying error logged.""" from unittest.mock import patch from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemModelConfig from deerflow.agents.memory.backends.deermem.deermem.core.llm import build_llm model_config = DeerMemModelConfig(provider="openai", model="bogus-model", api_key="k") llm_logger = "deerflow.agents.memory.backends.deermem.deermem.core.llm" with patch("langchain.chat_models.init_chat_model", side_effect=RuntimeError("boom")): with caplog.at_level("WARNING", logger=llm_logger): result = build_llm(model_config) assert result is None assert any("build_llm failed" in r.message for r in caplog.records) def test_from_backend_config_warns_on_unknown_keys(caplog): """Unknown backend_config keys log a WARNING so a typo (e.g. ``storage_pat`` missing the ``h``) does not silently fall back to the default and write memory to an unintended location. Mirrors the host layer's load_memory_config_from_dict warning.""" from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemConfig cfg_logger = "deerflow.agents.memory.backends.deermem.deermem.config" with caplog.at_level("WARNING", logger=cfg_logger): cfg = DeerMemConfig.from_backend_config({"storage_path": "/tmp/x", "storage_pat": "/tmp/y"}) # known key parsed; unknown key ignored but warned about assert cfg.storage_path == "/tmp/x" assert any("Unknown backend_config keys" in r.message for r in caplog.records) assert any("storage_pat" in r.message for r in caplog.records) def test_from_backend_config_silent_on_known_keys(caplog): """No warning when every key is known (regression guard for the typo warning).""" from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemConfig cfg_logger = "deerflow.agents.memory.backends.deermem.deermem.config" with caplog.at_level("WARNING", logger=cfg_logger): DeerMemConfig.from_backend_config({"storage_path": "/tmp/x", "max_facts": 20}) assert not any("Unknown backend_config keys" in r.message for r in caplog.records) def test_from_backend_config_null_values_fall_back_to_defaults(): """Explicit YAML ``null`` values must behave like omitted keys, not crash. ``config.example.yaml`` ships ``backend_config.model:`` as a bare key with commented children, which YAML parses to ``None`` (and ``make config-upgrade`` writes it out as an explicit ``model: null``). Non-Optional fields like ``model: DeerMemModelConfig`` reject an explicit ``None`` even though the omitted key would use the field default — so the shipped example config crashed every run with a DeerMemConfig ValidationError.""" from deerflow.agents.memory.backends.deermem.deermem.config import ( DeerMemConfig, DeerMemModelConfig, ) cfg = DeerMemConfig.from_backend_config({"model": None, "debounce_seconds": None, "storage_path": "/tmp/x"}) # None entries fall back to field defaults; real values still parse assert isinstance(cfg.model, DeerMemModelConfig) assert cfg.model.model is None # default = no extraction LLM configured assert cfg.debounce_seconds == DeerMemConfig().debounce_seconds assert cfg.storage_path == "/tmp/x" def test_from_backend_config_null_values_do_not_warn_as_unknown(caplog): """Dropped ``None`` entries are known keys — they must not trip the unknown-key typo warning.""" from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemConfig cfg_logger = "deerflow.agents.memory.backends.deermem.deermem.config" with caplog.at_level("WARNING", logger=cfg_logger): DeerMemConfig.from_backend_config({"model": None}) assert not any("Unknown backend_config keys" in r.message for r in caplog.records)