diff --git a/backend/README.md b/backend/README.md index 056b5ed7c..a3a74c276 100644 --- a/backend/README.md +++ b/backend/README.md @@ -99,6 +99,7 @@ LLM-powered persistent context retention across conversations: - **Debounced updates**: Batches updates to minimize LLM calls (configurable wait time) - **System prompt injection**: Top facts + context injected into agent prompts - **Run-level memory identity**: `GET /api/threads/{thread_id}/runs/{run_id}/events?event_types=context:memory` returns the SHA-256 identity of the effective hidden memory block without copying memory text into the event store +- **Read failures**: Strict backend policies (including legacy `fail_closed`) stop the turn, including at the 5-second async injection deadline. Fail-open reads continue without new context. Timeout handling does not wait for a free worker; a timed-out read may still occupy its worker until the backend returns. - **Storage**: JSON file with mtime-based cache invalidation ### Tool Ecosystem diff --git a/backend/packages/harness/deerflow/agents/lead_agent/prompt.py b/backend/packages/harness/deerflow/agents/lead_agent/prompt.py index a49d884d3..329f25bd7 100644 --- a/backend/packages/harness/deerflow/agents/lead_agent/prompt.py +++ b/backend/packages/harness/deerflow/agents/lead_agent/prompt.py @@ -772,6 +772,8 @@ def _get_memory_context( Returns: Formatted memory context string wrapped in XML tags, or empty string if disabled. """ + from deerflow.agents.memory import MemoryManagerError, MemoryReadError + config = None try: from deerflow.agents.memory import get_memory_manager @@ -799,12 +801,14 @@ def _get_memory_context( {memory_content} """ + except MemoryReadError: + logger.exception("Required memory context could not be loaded") + raise except Exception as exc: logger.exception("Failed to load memory context") - from deerflow.agents.memory import MemoryManagerError - - failure_policy = getattr(config, "backend_config", {}).get("failure_policy", {}) if config is not None else {} - if isinstance(exc, MemoryManagerError) and failure_policy.get("read") == "fail_closed": + backend_config = getattr(config, "backend_config", {}) if config is not None else {} + failure_policy = backend_config.get("failure_policy", {}) if isinstance(backend_config, dict) else {} + if isinstance(exc, MemoryManagerError) and isinstance(failure_policy, dict) and failure_policy.get("read") == "fail_closed": raise return "" diff --git a/backend/packages/harness/deerflow/agents/memory/AGENTS.md b/backend/packages/harness/deerflow/agents/memory/AGENTS.md index aadc1e97d..28150c5b6 100644 --- a/backend/packages/harness/deerflow/agents/memory/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/memory/AGENTS.md @@ -202,6 +202,18 @@ Use the earliest source review deadline for its next review. #### Remote backends +Strict reads use the backend-neutral `MemoryReadError`. +Backends declare their policy through `read_failures_are_fatal_for_config()`. +`DynamicContextMiddleware` preserves that policy at its injection timeout. +Policy methods must use only in-memory config. The non-loading lookup returns +unknown for a cold backend; discovery/config reload runs inside the existing +timed injection worker. Per-call policy state cannot leak across runs, and +timeout handling never submits more executor work. Unknown policy fails closed. +The prompt loader retains `MemoryManagerError` + `fail_closed` compatibility +for third-party backends that have not adopted the typed error. +The base policy resolver also honors legacy `fail_closed` at the timeout boundary; +other settings remain permissive unless the backend overrides the resolver. + OpenViking uses the maintained `langchain-openviking` package. Keep it in middleware mode. One API key is bound to one configured DeerFlow owner. diff --git a/backend/packages/harness/deerflow/agents/memory/__init__.py b/backend/packages/harness/deerflow/agents/memory/__init__.py index 22188cf49..cb85c5d5b 100644 --- a/backend/packages/harness/deerflow/agents/memory/__init__.py +++ b/backend/packages/harness/deerflow/agents/memory/__init__.py @@ -17,15 +17,19 @@ from deerflow.agents.memory.manager import ( MemoryCorruptionError, MemoryManager, MemoryManagerError, + MemoryReadError, get_memory_manager, + memory_read_failures_are_fatal, reset_memory_manager, ) __all__ = [ "MemoryManager", "MemoryManagerError", + "MemoryReadError", "MemoryConflictError", "MemoryCorruptionError", "get_memory_manager", + "memory_read_failures_are_fatal", "reset_memory_manager", ] diff --git a/backend/packages/harness/deerflow/agents/memory/backends/honcho/honcho_manager.py b/backend/packages/harness/deerflow/agents/memory/backends/honcho/honcho_manager.py index fc763a6b8..55b74a614 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/honcho/honcho_manager.py +++ b/backend/packages/harness/deerflow/agents/memory/backends/honcho/honcho_manager.py @@ -41,7 +41,7 @@ from typing import Any, ClassVar, Literal from pydantic import PrivateAttr # ABC contract -- the ONE allowed `from deerflow` import in this backend folder. -from deerflow.agents.memory.manager import MemoryManager, MemoryManagerError +from deerflow.agents.memory.manager import MemoryManager, MemoryManagerError, MemoryReadError from .client import HonchoClient from .config import HonchoConfig, sanitize_id @@ -122,6 +122,13 @@ class HonchoMemoryManager(MemoryManager): """ return cls(backend_config=backend_config, mode=mode) + @classmethod + def read_failures_are_fatal_for_config( + cls, + backend_config: dict[str, Any] | None, + ) -> bool: + return HonchoConfig.from_backend_config(backend_config).read_fail_closed + # ── identity resolution (fail closed) ──────────────────────────────── def _workspace(self, user_id: str | None) -> str | None: if not user_id: @@ -138,7 +145,7 @@ class HonchoMemoryManager(MemoryManager): def _read_or_fallback(self, fallback: Any, fn: Any) -> Any: """Single ``failure_policy.read`` gate for every recall path, mirroring mem0's helper of the same name: fail-open (default) logs and returns - ``fallback``; ``fail_closed`` wraps into ``MemoryManagerError``. The + ``fallback``; ``fail_closed`` wraps into ``MemoryReadError``. The broad ``except Exception`` is the containment boundary — no client exception may escape into ``MemoryMiddleware.after_agent``.""" try: @@ -147,7 +154,7 @@ class HonchoMemoryManager(MemoryManager): raise except Exception as exc: if self._config.read_fail_closed: - raise MemoryManagerError(f"honcho memory recall failed: {exc}") from exc + raise MemoryReadError(f"honcho memory recall failed: {exc}") from exc logger.warning("honcho memory: recall failed (fail-open): %s", exc) return fallback diff --git a/backend/packages/harness/deerflow/agents/memory/backends/mem0/mem0_manager.py b/backend/packages/harness/deerflow/agents/memory/backends/mem0/mem0_manager.py index 54c599b96..e4867f978 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/mem0/mem0_manager.py +++ b/backend/packages/harness/deerflow/agents/memory/backends/mem0/mem0_manager.py @@ -15,7 +15,7 @@ from typing import Any, ClassVar, Literal from pydantic import PrivateAttr # ABC contract -- the ONE allowed `from deerflow` import in this backend folder. -from deerflow.agents.memory.manager import MemoryManager, MemoryManagerError +from deerflow.agents.memory.manager import MemoryManager, MemoryManagerError, MemoryReadError from .client import Mem0APIError, Mem0Client from .config import Mem0Config @@ -106,6 +106,13 @@ class Mem0Manager(MemoryManager): """Release the underlying HTTP connection pool.""" self._client.close() + @classmethod + def read_failures_are_fatal_for_config( + cls, + backend_config: dict[str, Any] | None, + ) -> bool: + return Mem0Config.from_backend_config(backend_config).read_policy == "fail_closed" + # ── Error policies ─────────────────────────────────────────────────── def _read_or_fallback(self, fallback: Any, fn: Any) -> Any: try: @@ -114,7 +121,7 @@ class Mem0Manager(MemoryManager): if self._config.read_policy == "fail_open": logger.warning("mem0 read failed (%s); continuing without memory", e) return fallback - raise MemoryManagerError(f"mem0 read failed: {e}") from e + raise MemoryReadError(f"mem0 read failed: {e}") from e def _write_or_drop(self, fn: Any) -> None: try: diff --git a/backend/packages/harness/deerflow/agents/memory/backends/openviking/openviking_manager.py b/backend/packages/harness/deerflow/agents/memory/backends/openviking/openviking_manager.py index 13493d734..0dd9cf21e 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/openviking/openviking_manager.py +++ b/backend/packages/harness/deerflow/agents/memory/backends/openviking/openviking_manager.py @@ -16,7 +16,7 @@ from typing import Any, ClassVar, Literal from pydantic import PrivateAttr -from deerflow.agents.memory.manager import MemoryManager, MemoryManagerError +from deerflow.agents.memory.manager import MemoryManager, MemoryManagerError, MemoryReadError from .config import OpenVikingConfig from .session import ( @@ -118,6 +118,13 @@ class OpenVikingMemoryManager(MemoryManager): user_id=user_id, ) + @classmethod + def read_failures_are_fatal_for_config( + cls, + backend_config: dict[str, Any] | None, + ) -> bool: + return OpenVikingConfig.from_backend_config(backend_config).read_failure_policy == "raise" + def add_nowait( self, thread_id: str, @@ -163,7 +170,7 @@ class OpenVikingMemoryManager(MemoryManager): if not self._begin_operation(): return "" try: - peer_id = self._resolve_scope(user_id, agent_name) + peer_id = self._resolve_read_scope(user_id, agent_name) retriever = copy.copy(self._retriever) retriever.target_uri = _memory_target_uris(peer_id) if thread_id: @@ -176,9 +183,9 @@ class OpenVikingMemoryManager(MemoryManager): try: with self._actor_peer_scope(peer_id): documents = retriever.invoke(self._config.injection_query) - except Exception: + except Exception as exc: if self._config.read_failure_policy == "raise": - raise + raise MemoryReadError("OpenViking context retrieval failed") from exc logger.warning( "OpenViking context retrieval failed; continuing without injected memory", exc_info=True, @@ -217,7 +224,7 @@ class OpenVikingMemoryManager(MemoryManager): if not query.strip() or not self._begin_operation(): return [] try: - peer_id = self._resolve_scope(user_id, agent_name) + peer_id = self._resolve_read_scope(user_id, agent_name) retriever = copy.copy(self._retriever) retriever.target_uri = _memory_target_uris(peer_id) retriever.search_mode = "find" @@ -232,9 +239,9 @@ class OpenVikingMemoryManager(MemoryManager): try: with self._actor_peer_scope(peer_id): documents = retriever.invoke(query.strip()) - except Exception: + except Exception as exc: if self._config.read_failure_policy == "raise": - raise + raise MemoryReadError("OpenViking memory search failed") from exc logger.warning( "OpenViking memory search failed; returning no results", exc_info=True, @@ -453,6 +460,18 @@ class OpenVikingMemoryManager(MemoryManager): raise MemoryManagerError(f"OpenViking USER API key is bound to DeerFlow owner_user_id {self._config.owner_user_id!r}, but this request belongs to {resolved_user!r}. Refusing to share one credential across users.") return _canonical_peer_id(agent_name, self._config.default_peer_id) + def _resolve_read_scope( + self, + user_id: str | None, + agent_name: str | None, + ) -> str: + try: + return self._resolve_scope(user_id, agent_name) + except MemoryManagerError as exc: + if self._config.read_failure_policy == "raise": + raise MemoryReadError(str(exc)) from exc + raise + def _actor_peer_scope( self, peer_id: str, diff --git a/backend/packages/harness/deerflow/agents/memory/manager.py b/backend/packages/harness/deerflow/agents/memory/manager.py index a9f8ab9e1..1c6bef954 100644 --- a/backend/packages/harness/deerflow/agents/memory/manager.py +++ b/backend/packages/harness/deerflow/agents/memory/manager.py @@ -19,6 +19,7 @@ from __future__ import annotations import importlib import logging import os +import sys import threading from abc import abstractmethod from pathlib import Path @@ -86,6 +87,10 @@ class MemoryManagerError(RuntimeError): """Backend-neutral base error exposed at the MemoryManager boundary.""" +class MemoryReadError(MemoryManagerError): + """A required memory read failed, so callers must not continue without it.""" + + class MemoryConflictError(MemoryManagerError): """The requested write lost an optimistic-concurrency race.""" @@ -166,6 +171,28 @@ class MemoryManager(BaseModel): # search. Most backends keep tool mode fully model-directed. requires_passive_writes_in_tool_mode: ClassVar[bool] = False + @classmethod + def read_failures_are_fatal_for_config( + cls, + backend_config: dict[str, Any] | None, + ) -> bool: + """Honor legacy fail_closed using only in-memory config; do not perform I/O.""" + + failure_policy = backend_config.get("failure_policy") if isinstance(backend_config, dict) else None + return isinstance(failure_policy, dict) and failure_policy.get("read") == "fail_closed" + + @property + def read_failures_are_fatal(self) -> bool: + """Whether caller-owned timeouts must abort instead of degrading. + + Backends that require memory context override the class-level config + resolver so this remains available before or after manager creation. + The default honors legacy ``fail_closed``; other settings are permissive + unless the backend overrides the config resolver. + """ + + return type(self).read_failures_are_fatal_for_config(self.backend_config) + @model_validator(mode="after") def _check_invariants(self) -> MemoryManager: """Cross-field invariants every backend must satisfy at instantiation. @@ -235,6 +262,12 @@ class MemoryManager(BaseModel): the returned string is injected verbatim by call sites. Format parameters are the backend's own private config (received via ``backend_config`` at construction), NOT a host config on this method. + + Backends configured to tolerate read failures return an empty string. + Backends configured to require memory context raise + :class:`MemoryReadError`, which callers must propagate, and expose + ``read_failures_are_fatal`` so caller-owned timeouts preserve the same + policy before a backend call returns. """ # ── Tier 2: management ops with defaults ──────────────────────────── @@ -590,7 +623,7 @@ def _scan_backends() -> dict[str, type[MemoryManager]]: return registry -def _resolve_manager_class(manager_class: str) -> type[MemoryManager]: +def _resolve_manager_class(manager_class: str, *, allow_discovery: bool = True) -> type[MemoryManager]: """Resolve a ``manager_class`` config value to a concrete class. Resolution order: @@ -605,7 +638,9 @@ def _resolve_manager_class(manager_class: str) -> type[MemoryManager]: is resolved eagerly at startup so it can be warmed) so the operator fixes ``memory.manager_class`` instead of discovering the mismatch later. """ - registry = _scan_backends() + # Timeout preparation may only inspect already-loaded backends. Do not + # turn a cold registry or dotted import into event-loop file/import I/O. + registry = _scan_backends() if allow_discovery else (_backends_cache or {}) if manager_class in registry: return registry[manager_class] @@ -617,11 +652,14 @@ def _resolve_manager_class(manager_class: str) -> type[MemoryManager]: module_path, _, attr = manager_class.rpartition(".") if module_path and attr: try: - module = importlib.import_module(module_path) + module = importlib.import_module(module_path) if allow_discovery else sys.modules.get(module_path) except ImportError as e: dotted_error = f"cannot import module {module_path!r}: {e}" else: - cls = getattr(module, attr, None) + if allow_discovery: + cls = getattr(module, attr, None) + else: + cls = vars(module).get(attr) if module is not None else None if cls is None: dotted_error = f"attribute {attr!r} not found in {module_path!r}" elif not (isinstance(cls, type) and issubclass(cls, MemoryManager)): @@ -930,6 +968,33 @@ def get_memory_manager() -> MemoryManager: return _memory_manager +def memory_read_failures_are_fatal( + manager_class: str, + backend_config: dict[str, Any] | None, + *, + resolved_only: bool = False, +) -> bool | None: + """Resolve strict-read capability without constructing a new manager. + + With ``resolved_only``, never scan/import backends; return ``None`` when + the class is not already loaded. The caller can finish discovery inside + its bounded worker. Invalid config and full-resolution failures fail closed. + """ + + try: + cls = _resolve_manager_class(manager_class, allow_discovery=not resolved_only) + except Exception: + if resolved_only: + return None + logger.exception("Could not resolve memory read failure policy; treating the read timeout as fatal") + return True + try: + return cls.read_failures_are_fatal_for_config(backend_config) + except Exception: + logger.exception("Could not resolve memory read failure policy; treating the read timeout as fatal") + return True + + def reset_memory_manager() -> None: """Clear the cached singleton manager and the backend registry. diff --git a/backend/packages/harness/deerflow/agents/middlewares/dynamic_context_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/dynamic_context_middleware.py index b2effeae6..023a56800 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/dynamic_context_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/dynamic_context_middleware.py @@ -384,6 +384,25 @@ class DynamicContextMiddleware(AgentMiddleware): def _build_date_update_reminder(self) -> str: return _format_current_date_reminder(_format_current_date()) + def _read_failures_are_fatal(self, *, allow_io: bool = True) -> bool | None: + from deerflow.agents.memory import memory_read_failures_are_fatal + from deerflow.config.memory_config import get_memory_config + + if self._app_config is None and not allow_io: + return None # get_memory_config() may reload config.yaml from disk. + try: + memory_config = self._app_config.memory if self._app_config else get_memory_config() + if not memory_config.enabled or not memory_config.injection_enabled: + return False + return memory_read_failures_are_fatal( + memory_config.manager_class, + memory_config.backend_config, + resolved_only=not allow_io, + ) + except Exception: + logger.exception("DynamicContextMiddleware: could not resolve memory read failure policy; treating the injection timeout as fatal") + return True + @staticmethod def _make_reminder_and_user_messages( original: HumanMessage, @@ -506,6 +525,18 @@ class DynamicContextMiddleware(AgentMiddleware): @override async def abefore_agent(self, state, runtime: Runtime) -> dict | None: + # The warm path uses only this call's config and already-loaded class. + # Cold discovery/config reload shares the injection's bounded worker, + # never a second executor job after the timeout. Keep this value local: + # a late worker must not overwrite another run's timeout policy. + read_failures_are_fatal = self._read_failures_are_fatal(allow_io=False) + + def inject_with_policy(): + nonlocal read_failures_are_fatal + if read_failures_are_fatal is None: + read_failures_are_fatal = self._read_failures_are_fatal() + return self._inject(state, runtime) + # _inject() performs synchronous file I/O (memory JSON loading) and # potentially blocking network calls (tiktoken encoding download on # first use). Offload to a thread so the event loop is never blocked @@ -519,10 +550,16 @@ class DynamicContextMiddleware(AgentMiddleware): # rather than hanging. Frozen context already in state remains active. try: result = await asyncio.wait_for( - asyncio.to_thread(self._inject, state, runtime), + asyncio.to_thread(inject_with_policy), timeout=_INJECT_TIMEOUT_SECONDS, ) - except TimeoutError: + except TimeoutError as exc: + from deerflow.agents.memory import MemoryReadError + + # A worker that never started (or is still resolving policy) leaves + # the policy unknown. Fail closed without waiting for that worker. + if read_failures_are_fatal is not False: + raise MemoryReadError("Required memory context retrieval timed out") from exc logger.warning( "DynamicContextMiddleware: injection timed out (%.1fs); skipping new memory/date injection for this turn", _INJECT_TIMEOUT_SECONDS, diff --git a/backend/tests/blocking_io/test_dynamic_context_middleware.py b/backend/tests/blocking_io/test_dynamic_context_middleware.py index 6dfb603b6..fa19bc448 100644 --- a/backend/tests/blocking_io/test_dynamic_context_middleware.py +++ b/backend/tests/blocking_io/test_dynamic_context_middleware.py @@ -16,6 +16,7 @@ from __future__ import annotations import asyncio import hashlib import threading +from concurrent.futures import ThreadPoolExecutor from types import SimpleNamespace from unittest import mock @@ -23,11 +24,16 @@ import pytest from langchain.agents import create_agent from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel from langchain_core.messages import AIMessage, HumanMessage +from pydantic import PrivateAttr +from deerflow.agents.lead_agent import prompt as prompt_module +from deerflow.agents.memory import MemoryManager, MemoryReadError, reset_memory_manager +from deerflow.agents.memory.manager import _scan_backends from deerflow.agents.middlewares.dynamic_context_middleware import ( _DYNAMIC_CONTEXT_REMINDER_KEY, DynamicContextMiddleware, ) +from deerflow.config.memory_config import MemoryConfig from deerflow.runtime.context_keys import CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY pytestmark = pytest.mark.asyncio @@ -40,6 +46,34 @@ class _FakeModel(FakeMessagesListChatModel): return self +class _LegacyBackend(MemoryManager): + """Third-party backend that inherits the default timeout-policy resolver.""" + + _release: threading.Event = PrivateAttr(default_factory=threading.Event) + _finished: threading.Event = PrivateAttr(default_factory=threading.Event) + + @classmethod + def from_config(cls, backend_config, *, mode="middleware", **host_hooks): + return cls(backend_config=backend_config, mode=mode) + + def add(self, thread_id, messages, **kwargs): + pass + + def get_context(self, user_id, **kwargs): + try: + self._release.wait(timeout=2) + return "Late memory context" + finally: + self._finished.set() + + +@pytest.fixture(autouse=True) +def _isolate_memory_manager() -> None: + reset_memory_manager() + yield + reset_memory_manager() + + async def test_abefore_agent_does_not_block_event_loop() -> None: """``abefore_agent`` must offload _inject() to a thread pool.""" mw = DynamicContextMiddleware() @@ -57,7 +91,7 @@ async def test_abefore_agent_does_not_block_event_loop() -> None: with ( mock.patch.object(mw, "_build_full_reminder", slow_build_reminder), - mock.patch("deerflow.agents.lead_agent.prompt._get_memory_context", return_value=""), + mock.patch.object(prompt_module, "_get_memory_context", return_value=""), ): agent = await asyncio.to_thread( lambda: create_agent( @@ -84,7 +118,7 @@ async def test_abefore_agent_returns_same_result_as_before_agent() -> None: runtime = SimpleNamespace(context={}) with ( - mock.patch("deerflow.agents.lead_agent.prompt._get_memory_context", return_value=""), + mock.patch.object(prompt_module, "_get_memory_context", return_value=""), mock.patch("deerflow.agents.middlewares.dynamic_context_middleware.datetime") as mock_dt, ): mock_dt.now.return_value.strftime.return_value = "2026-06-05, Friday" @@ -106,9 +140,23 @@ async def test_abefore_agent_returns_same_result_as_before_agent() -> None: assert sync_result["messages"][1].id == async_result["messages"][1].id -async def test_abefore_agent_returns_none_on_timeout() -> None: +async def test_abefore_agent_returns_none_on_timeout( + monkeypatch: pytest.MonkeyPatch, +) -> None: """A timed-out worker must not emit a late, phantom context event.""" - mw = DynamicContextMiddleware() + monkeypatch.setenv("OPENVIKING_API_KEY", "test-key") + await asyncio.to_thread(_scan_backends) + mw = DynamicContextMiddleware( + app_config=SimpleNamespace( + memory=MemoryConfig( + manager_class="openviking", + backend_config={ + "owner_user_id": "alice", + "failure_policy": {"read": "fail_open"}, + }, + ) + ) + ) started = threading.Event() release = threading.Event() finished = threading.Event() @@ -150,6 +198,127 @@ async def test_abefore_agent_returns_none_on_timeout() -> None: journal.record_memory_context.assert_not_called() +async def test_abefore_agent_propagates_strict_memory_timeout( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A strict backend must not degrade after the middleware timeout.""" + monkeypatch.setenv("OPENVIKING_API_KEY", "test-key") + await asyncio.to_thread(_scan_backends) + mw = DynamicContextMiddleware( + app_config=SimpleNamespace( + memory=MemoryConfig( + manager_class="openviking", + backend_config={ + "owner_user_id": "alice", + "failure_policy": {"read": "raise"}, + }, + ) + ) + ) + started = threading.Event() + release = threading.Event() + finished = threading.Event() + + def blocking_inject(state, runtime=None): + started.set() + release.wait(timeout=2) + finished.set() + + with ( + mock.patch.object(mw, "_inject", blocking_inject), + mock.patch( + "deerflow.agents.middlewares.dynamic_context_middleware._INJECT_TIMEOUT_SECONDS", + 0.01, + ), + ): + state = {"messages": [HumanMessage(content="Hello", id="msg-1")]} + runtime = SimpleNamespace(context={}) + with pytest.raises(MemoryReadError) as exc_info: + await mw.abefore_agent(state, runtime) + + assert isinstance(exc_info.value.__cause__, TimeoutError) + assert started.is_set() + release.set() + assert await asyncio.to_thread(finished.wait, 1) + + +@pytest.mark.parametrize( + ("manager_class", "backend_config", "api_key"), + [ + pytest.param( + "openviking", + { + "owner_user_id": "alice", + "failure_policy": {"read": "raise"}, + }, + None, + id="missing_openviking_api_key", + ), + pytest.param( + "openviking", + { + "owner_user_id": "alice", + "failure_policy": {"read": "invalid"}, + }, + "test-key", + id="invalid_backend_config", + ), + pytest.param( + "missing.backend:Manager", + {}, + None, + id="unknown_manager_class", + ), + ], +) +async def test_abefore_agent_policy_resolution_failure_does_not_replace_timeout( + monkeypatch: pytest.MonkeyPatch, + manager_class: str, + backend_config: dict, + api_key: str | None, +) -> None: + """An unresolved timeout policy must fail closed with the original cause.""" + if api_key is None: + monkeypatch.delenv("OPENVIKING_API_KEY", raising=False) + else: + monkeypatch.setenv("OPENVIKING_API_KEY", api_key) + mw = DynamicContextMiddleware( + app_config=SimpleNamespace( + memory=MemoryConfig( + manager_class=manager_class, + backend_config=backend_config, + ) + ) + ) + started = threading.Event() + release = threading.Event() + finished = threading.Event() + + def blocking_inject(state, runtime=None): + started.set() + release.wait(timeout=2) + finished.set() + + try: + with ( + mock.patch.object(mw, "_inject", blocking_inject), + mock.patch( + "deerflow.agents.middlewares.dynamic_context_middleware._INJECT_TIMEOUT_SECONDS", + 0.01, + ), + ): + state = {"messages": [HumanMessage(content="Hello", id="msg-1")]} + runtime = SimpleNamespace(context={}) + with pytest.raises(MemoryReadError) as exc_info: + await mw.abefore_agent(state, runtime) + finally: + release.set() + assert await asyncio.to_thread(finished.wait, 1) + + assert isinstance(exc_info.value.__cause__, TimeoutError) + assert started.is_set() + + async def test_abefore_agent_records_checkpointed_memory_on_timeout() -> None: """A timeout does not hide frozen memory that remains effective for the run.""" mw = DynamicContextMiddleware() @@ -209,3 +378,161 @@ async def test_abefore_agent_records_checkpointed_memory_on_timeout() -> None: content_sha256=hashlib.sha256(memory_content.encode("utf-8")).hexdigest(), ) journal.record_memory_context.assert_called_once() + + +@pytest.mark.parametrize("read_policy", ["fail_open", "raise"]) +@pytest.mark.parametrize("already_saturated", [False, True], ids=["read_occupies_worker", "pool_already_full"]) +async def test_timeout_does_not_wait_for_saturated_executor(monkeypatch, read_policy, already_saturated): + """Neither a running read nor another request may delay timeout handling.""" + monkeypatch.setenv("OPENVIKING_API_KEY", "test-key") + await asyncio.to_thread(_scan_backends) # normal Gateway startup discovery + mw = DynamicContextMiddleware( + app_config=SimpleNamespace( + memory=MemoryConfig( + manager_class="openviking", + backend_config={"owner_user_id": "alice", "failure_policy": {"read": read_policy}}, + ) + ) + ) + entered = threading.Event() + release = threading.Event() + finished = threading.Event() + executor = ThreadPoolExecutor(max_workers=1) + loop = asyncio.get_running_loop() + + def occupy_worker(*_args): + entered.set() + release.wait(timeout=2) + finished.set() + + try: + with mock.patch.object(loop, "_default_executor", executor): + if already_saturated: + executor.submit(occupy_worker) + while not entered.is_set(): + await asyncio.sleep(0) + with ( + mock.patch.object(mw, "_inject", side_effect=occupy_worker) as inject, + mock.patch("deerflow.agents.middlewares.dynamic_context_middleware._INJECT_TIMEOUT_SECONDS", 0.01), + ): + call = mw.abefore_agent({"messages": [HumanMessage(content="hi", id="m1")]}, SimpleNamespace(context={})) + if read_policy == "raise": + with pytest.raises(MemoryReadError) as exc_info: + await asyncio.wait_for(call, 0.25) + assert isinstance(exc_info.value.__cause__, TimeoutError) + else: + assert await asyncio.wait_for(call, 0.25) is None + assert not finished.is_set() # the request returned before its worker + assert inject.call_count == (0 if already_saturated else 1) + finally: + release.set() + await asyncio.to_thread(executor.shutdown, wait=True, cancel_futures=True) + + +@pytest.mark.parametrize("read_policy", ["fail_closed", "fail_open"]) +async def test_legacy_backend_timeout_preserves_read_policy(read_policy): + """The real read path honors legacy policy without waiting for its worker.""" + cfg = MemoryConfig(manager_class=f"{__name__}:_LegacyBackend", backend_config={"failure_policy": {"read": read_policy}}) + backend = _LegacyBackend.from_config(cfg.backend_config) + mw = DynamicContextMiddleware(app_config=SimpleNamespace(memory=cfg)) + executor = ThreadPoolExecutor(max_workers=1) + try: + with ( + mock.patch.object(asyncio.get_running_loop(), "_default_executor", executor), + mock.patch("deerflow.agents.memory.get_memory_manager", return_value=backend), + mock.patch("deerflow.agents.middlewares.dynamic_context_middleware._INJECT_TIMEOUT_SECONDS", 0.01), + ): + call = mw.abefore_agent({"messages": [HumanMessage(content="hi", id="m1")]}, SimpleNamespace(context={})) + if read_policy == "fail_closed": + with pytest.raises(MemoryReadError) as exc_info: + await asyncio.wait_for(call, 0.25) + assert isinstance(exc_info.value.__cause__, TimeoutError) + else: + assert await asyncio.wait_for(call, 0.25) is None + assert not backend._finished.is_set() + finally: + backend._release.set() + await asyncio.to_thread(executor.shutdown, wait=True, cancel_futures=True) + assert backend._finished.is_set() + + +@pytest.mark.parametrize("explicit_config", [True, False], ids=["cold_registry", "config_fallback"]) +async def test_cold_policy_resolution_stays_off_event_loop(monkeypatch, tmp_path, explicit_config): + """Cold discovery and the config-reload fallback remain inside the deadline.""" + from deerflow.agents.memory import manager as manager_module + + monkeypatch.setenv("OPENVIKING_API_KEY", "test-key") + cfg = MemoryConfig(manager_class="openviking", backend_config={"owner_user_id": "alice", "failure_policy": {"read": "fail_open"}}) + policy_file = tmp_path / "policy.txt" + await asyncio.to_thread(policy_file.write_text, "policy", encoding="utf-8") + event_loop_thread = threading.get_ident() + seen = [] + + def cold_scan(): + assert threading.get_ident() != event_loop_thread + policy_file.read_text(encoding="utf-8") + seen.append("scan") + return _scan_backends() + + def reload_config(): + assert threading.get_ident() != event_loop_thread + policy_file.read_text(encoding="utf-8") + seen.append("config") + return cfg + + mw = DynamicContextMiddleware(app_config=SimpleNamespace(memory=cfg) if explicit_config else None) + release = threading.Event() + finished = threading.Event() + + def blocking_inject(*_): + try: + release.wait(timeout=2) + finally: + finished.set() + + try: + with ( + mock.patch.object(manager_module, "_scan_backends", side_effect=cold_scan), + mock.patch("deerflow.config.memory_config.get_memory_config", side_effect=reload_config), + mock.patch.object(mw, "_inject", side_effect=blocking_inject), + mock.patch("deerflow.agents.middlewares.dynamic_context_middleware._INJECT_TIMEOUT_SECONDS", 0.2), + ): + assert await asyncio.wait_for(mw.abefore_agent({}, SimpleNamespace(context={})), 0.5) is None + finally: + release.set() + assert await asyncio.to_thread(finished.wait, 1) + assert "scan" in seen + assert ("config" in seen) is not explicit_config + + +@pytest.mark.parametrize("disabled_field", [None, "enabled", "injection_enabled"], ids=["unknown_policy", "memory_disabled", "injection_disabled"]) +async def test_cold_saturated_timeout_never_starts_discovery(monkeypatch, disabled_field): + """An unknown policy fails closed; disabled memory needs no policy lookup.""" + cfg = MemoryConfig(manager_class="openviking") + if disabled_field: + setattr(cfg, disabled_field, False) + mw = DynamicContextMiddleware(app_config=SimpleNamespace(memory=cfg)) + executor = ThreadPoolExecutor(max_workers=1) + release = threading.Event() + try: + with ( + mock.patch.object(asyncio.get_running_loop(), "_default_executor", executor), + mock.patch("deerflow.agents.memory.manager._scan_backends") as scan, + mock.patch("deerflow.config.memory_config.get_memory_config") as reload_config, + mock.patch.object(mw, "_inject") as inject, + mock.patch("deerflow.agents.middlewares.dynamic_context_middleware._INJECT_TIMEOUT_SECONDS", 0.01), + ): + executor.submit(release.wait, 2) + call = mw.abefore_agent({}, SimpleNamespace(context={})) + if disabled_field is None: + with pytest.raises(MemoryReadError) as exc_info: + await asyncio.wait_for(call, 0.25) + assert isinstance(exc_info.value.__cause__, TimeoutError) + else: + assert await asyncio.wait_for(call, 0.25) is None + scan.assert_not_called() + reload_config.assert_not_called() + inject.assert_not_called() + finally: + release.set() + await asyncio.to_thread(executor.shutdown, wait=True, cancel_futures=True) diff --git a/backend/tests/test_honcho_memory_backend.py b/backend/tests/test_honcho_memory_backend.py index aa913e4eb..28119c99b 100644 --- a/backend/tests/test_honcho_memory_backend.py +++ b/backend/tests/test_honcho_memory_backend.py @@ -11,7 +11,7 @@ import pytest from deerflow.agents.memory.backends.honcho.client import HonchoClient, HonchoRequestError from deerflow.agents.memory.backends.honcho.config import HonchoConfig, sanitize_id from deerflow.agents.memory.backends.honcho.honcho_manager import HonchoMemoryManager, _stable_id -from deerflow.agents.memory.manager import MemoryManagerError +from deerflow.agents.memory.manager import MemoryManagerError, MemoryReadError class TestHonchoConfig: @@ -385,7 +385,8 @@ class TestHonchoManagerRead: def test_get_context_fail_closed_raises_contract_error(self): mgr, fake = _manager(failure_policy={"read": "fail_closed"}) fake.raise_on = "representation" - with pytest.raises(MemoryManagerError): + assert mgr.read_failures_are_fatal is True + with pytest.raises(MemoryReadError): mgr.get_context("u1") def test_get_context_fail_open_swallows_non_honcho_exceptions(self): diff --git a/backend/tests/test_lead_agent_prompt.py b/backend/tests/test_lead_agent_prompt.py index 1344494b2..01f7ad556 100644 --- a/backend/tests/test_lead_agent_prompt.py +++ b/backend/tests/test_lead_agent_prompt.py @@ -394,7 +394,38 @@ def test_get_memory_context_uses_explicit_app_config_without_global_config(monke } -def test_get_memory_context_propagates_fail_closed_manager_error(monkeypatch): +def test_get_memory_context_propagates_required_read_error_without_backend_config(monkeypatch): + from deerflow.agents.memory import MemoryReadError + + explicit_config = SimpleNamespace( + memory=SimpleNamespace( + enabled=True, + injection_enabled=True, + backend_config={}, + ), + ) + manager = SimpleNamespace(get_context=lambda *args, **kwargs: (_ for _ in ()).throw(MemoryReadError("down"))) + monkeypatch.setattr("deerflow.agents.memory.get_memory_manager", lambda: manager) + monkeypatch.setattr("deerflow.runtime.user_context.get_effective_user_id", lambda: "user-1") + + with pytest.raises(MemoryReadError, match="down"): + prompt_module._get_memory_context("agent-a", app_config=explicit_config) + + +def test_get_memory_context_swallows_ordinary_manager_error(monkeypatch): + from deerflow.agents.memory import MemoryManagerError + + explicit_config = SimpleNamespace( + memory=SimpleNamespace(enabled=True, injection_enabled=True, backend_config={}), + ) + manager = SimpleNamespace(get_context=lambda *args, **kwargs: (_ for _ in ()).throw(MemoryManagerError("down"))) + monkeypatch.setattr("deerflow.agents.memory.get_memory_manager", lambda: manager) + monkeypatch.setattr("deerflow.runtime.user_context.get_effective_user_id", lambda: "user-1") + + assert prompt_module._get_memory_context("agent-a", app_config=explicit_config) == "" + + +def test_get_memory_context_preserves_legacy_fail_closed_contract(monkeypatch): from deerflow.agents.memory import MemoryManagerError explicit_config = SimpleNamespace( @@ -406,23 +437,13 @@ def test_get_memory_context_propagates_fail_closed_manager_error(monkeypatch): ) manager = SimpleNamespace(get_context=lambda *args, **kwargs: (_ for _ in ()).throw(MemoryManagerError("down"))) monkeypatch.setattr("deerflow.agents.memory.get_memory_manager", lambda: manager) - monkeypatch.setattr("deerflow.runtime.user_context.get_effective_user_id", lambda: "user-1") with pytest.raises(MemoryManagerError, match="down"): - prompt_module._get_memory_context("agent-a", app_config=explicit_config) - - -def test_get_memory_context_swallows_manager_error_without_fail_closed(monkeypatch): - from deerflow.agents.memory import MemoryManagerError - - explicit_config = SimpleNamespace( - memory=SimpleNamespace(enabled=True, injection_enabled=True, backend_config={}), - ) - manager = SimpleNamespace(get_context=lambda *args, **kwargs: (_ for _ in ()).throw(MemoryManagerError("down"))) - monkeypatch.setattr("deerflow.agents.memory.get_memory_manager", lambda: manager) - monkeypatch.setattr("deerflow.runtime.user_context.get_effective_user_id", lambda: "user-1") - - assert prompt_module._get_memory_context("agent-a", app_config=explicit_config) == "" + prompt_module._get_memory_context( + "agent-a", + app_config=explicit_config, + user_id="user-1", + ) def test_get_memory_context_prefers_explicit_user_id(monkeypatch): diff --git a/backend/tests/test_mem0_memory_backend.py b/backend/tests/test_mem0_memory_backend.py index a77a0f7e9..4ba1ba571 100644 --- a/backend/tests/test_mem0_memory_backend.py +++ b/backend/tests/test_mem0_memory_backend.py @@ -441,6 +441,22 @@ class TestMem0ManagerAdd: class TestMem0ManagerGetContext: + @pytest.mark.parametrize( + ("read_policy", "expected"), + [ + pytest.param("fail_open", False, id="fail_open"), + pytest.param("fail_closed", True, id="fail_closed"), + ], + ) + def test_read_failure_capability_matches_policy( + self, + read_policy: str, + expected: bool, + ) -> None: + mgr, _fake = _manager({"failure_policy": {"read": read_policy}}) + + assert mgr.read_failures_are_fatal is expected + def test_formats_dedupes_and_scopes(self) -> None: mgr, fake = _manager() fake.list_results = [ @@ -466,11 +482,11 @@ class TestMem0ManagerGetContext: assert mgr.get_context("u1") == "" def test_read_error_fail_closed_raises(self) -> None: - from deerflow.agents.memory.manager import MemoryManagerError + from deerflow.agents.memory.manager import MemoryReadError mgr, fake = _manager({"failure_policy": {"read": "fail_closed"}}) fake.error = Mem0APIError("down") - with pytest.raises(MemoryManagerError): + with pytest.raises(MemoryReadError): mgr.get_context("u1") def test_truncates_to_max_injection_chars(self) -> None: diff --git a/backend/tests/test_memory_manager_interface.py b/backend/tests/test_memory_manager_interface.py index df2e3eeac..aab1e2ca6 100644 --- a/backend/tests/test_memory_manager_interface.py +++ b/backend/tests/test_memory_manager_interface.py @@ -14,11 +14,19 @@ Each test resets the singleton + restores config so they are order-independent. from __future__ import annotations import asyncio +from unittest import mock import pytest from pydantic import PrivateAttr -from deerflow.agents.memory import MemoryManager, get_memory_manager, reset_memory_manager +from deerflow.agents.memory import ( + MemoryManager, + MemoryManagerError, + MemoryReadError, + get_memory_manager, + memory_read_failures_are_fatal, + reset_memory_manager, +) from deerflow.agents.memory.manager import MemoryCallbacks from deerflow.config.memory_config import MemoryConfig, get_memory_config, set_memory_config @@ -212,6 +220,131 @@ def test_callbacks_field_optional_and_noop_default(): assert manager.callbacks is noop +def test_required_read_error_shares_manager_boundary(): + assert issubclass(MemoryReadError, MemoryManagerError) + assert _MinimalBackend(backend_config={}).read_failures_are_fatal is False + assert ( + memory_read_failures_are_fatal( + f"{__name__}:_MinimalBackend", + {}, + ) + is False + ) + + +@pytest.mark.parametrize( + ("backend_config", "expected"), + [ + (None, False), + ({}, False), + ({"failure_policy": {"read": "fail_closed"}}, True), + ({"failure_policy": {"read": "fail_open"}}, False), + ({"failure_policy": {"read": "raise"}}, False), + ({"failure_policy": None}, False), + ({"failure_policy": "fail_closed"}, False), + ({"failure_policy": []}, False), + ], +) +def test_base_read_failure_policy_preserves_legacy_config(backend_config, expected): + """A backend without a policy override keeps the prompt's legacy semantics.""" + assert _MinimalBackend(backend_config=backend_config).read_failures_are_fatal is expected + assert memory_read_failures_are_fatal(f"{__name__}:_MinimalBackend", backend_config, resolved_only=True) is expected + + +def test_read_failure_capability_uses_requested_backend_config( + monkeypatch: pytest.MonkeyPatch, +): + set_memory_config(MemoryConfig(manager_class=f"{__name__}:_MinimalBackend")) + manager = get_memory_manager() + monkeypatch.setenv("OPENVIKING_API_KEY", "test-key") + + assert isinstance(manager, _MinimalBackend) + assert ( + memory_read_failures_are_fatal( + "openviking", + { + "owner_user_id": "alice", + "failure_policy": {"read": "raise"}, + }, + ) + is True + ) + + +@pytest.mark.parametrize("selector", [f"{__name__}:_MinimalBackend", f"{__name__}._MinimalBackend", "not_loaded.backend:Manager"]) +def test_resolved_policy_does_not_scan_or_import_backends(selector): + """Loaded dotted backends work; unknown ones remain unknown without imports.""" + with ( + mock.patch("deerflow.agents.memory.manager._scan_backends") as scan, + mock.patch("deerflow.agents.memory.manager.importlib.import_module") as import_module, + ): + result = memory_read_failures_are_fatal(selector, {}, resolved_only=True) + assert result is (None if selector.startswith("not_loaded") else False) + scan.assert_not_called() + import_module.assert_not_called() + + +def test_resolved_policy_tracks_current_config_without_caching_boolean(monkeypatch): + monkeypatch.setenv("OPENVIKING_API_KEY", "test-key") + config = {"owner_user_id": "alice", "failure_policy": {"read": "fail_open"}} + assert memory_read_failures_are_fatal("openviking", config, resolved_only=True) is None + assert memory_read_failures_are_fatal("openviking", config) is False + with mock.patch("deerflow.agents.memory.manager._scan_backends") as scan: + assert memory_read_failures_are_fatal("openviking", config, resolved_only=True) is False + config["failure_policy"]["read"] = "raise" + assert memory_read_failures_are_fatal("openviking", config, resolved_only=True) is True + scan.assert_not_called() + + +@pytest.mark.parametrize( + ("manager_class", "backend_config", "api_key"), + [ + pytest.param( + "openviking", + { + "owner_user_id": "alice", + "failure_policy": {"read": "raise"}, + }, + None, + id="missing_openviking_api_key", + ), + pytest.param( + "openviking", + { + "owner_user_id": "alice", + "failure_policy": {"read": "invalid"}, + }, + "test-key", + id="invalid_backend_config", + ), + pytest.param( + "missing.backend:Manager", + {}, + None, + id="unknown_manager_class", + ), + ], +) +def test_read_failure_capability_fails_closed_when_policy_cannot_be_resolved( + monkeypatch: pytest.MonkeyPatch, + manager_class: str, + backend_config: dict, + api_key: str | None, +) -> None: + if api_key is None: + monkeypatch.delenv("OPENVIKING_API_KEY", raising=False) + else: + monkeypatch.setenv("OPENVIKING_API_KEY", api_key) + + assert ( + memory_read_failures_are_fatal( + manager_class, + backend_config, + ) + is True + ) + + def test_from_config_consumes_host_hooks_it_needs(): """A backend's from_config consumes the host_hooks it needs; the minimal backend consumes none (ignores callbacks / host_llm_factory / etc.). A real diff --git a/backend/tests/test_openviking_memory_backend.py b/backend/tests/test_openviking_memory_backend.py index 947edf71f..c38ec91e4 100644 --- a/backend/tests/test_openviking_memory_backend.py +++ b/backend/tests/test_openviking_memory_backend.py @@ -5,8 +5,10 @@ from __future__ import annotations import copy import gc import json +import socket import threading import weakref +from collections.abc import Iterator from contextlib import contextmanager from contextvars import ContextVar from pathlib import Path @@ -24,9 +26,11 @@ from deerflow.agents.memory.backends.openviking.openviking_manager import ( ) from deerflow.agents.memory.manager import ( MemoryManagerError, + MemoryReadError, _scan_backends, reset_memory_manager, ) +from deerflow.agents.middlewares.dynamic_context_middleware import DynamicContextMiddleware class _CommitPolicy: @@ -210,6 +214,14 @@ def _manager( return OpenVikingMemoryManager.from_config(_backend_config(tmp_path, **overrides)) +@pytest.fixture +def unreachable_openviking_url() -> Iterator[str]: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as reserved_socket: + reserved_socket.bind(("127.0.0.1", 0)) + host, port = reserved_socket.getsockname() + yield f"http://{host}:{port}" + + def test_config_uses_single_user_key_and_rejects_legacy_trusted_fields( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -241,6 +253,28 @@ def test_backend_is_discovered_by_registered_name() -> None: assert _scan_backends()["openviking"] is OpenVikingMemoryManager +@pytest.mark.parametrize( + ("read_policy", "expected"), + [ + pytest.param("fail_open", False, id="fail_open"), + pytest.param("raise", True, id="raise"), + ], +) +def test_read_failure_capability_matches_policy( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + read_policy: str, + expected: bool, +) -> None: + manager = _manager( + tmp_path, + monkeypatch, + failure_policy={"read": read_policy}, + ) + + assert manager.read_failures_are_fatal is expected + + def test_official_loader_uses_standalone_package() -> None: from deerflow.agents.memory.backends.openviking.openviking_manager import ( _load_official_integration, @@ -329,6 +363,150 @@ def test_context_without_thread_uses_existing_find_path( ] +def test_unreachable_context_read_raise_aborts_dynamic_context_injection( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + unreachable_openviking_url: str, +) -> None: + manager = _manager( + tmp_path, + monkeypatch, + base_url=unreachable_openviking_url, + timeout_seconds=0.1, + failure_policy={"read": "raise"}, + ) + monkeypatch.setattr( + "deerflow.agents.memory.get_memory_manager", + lambda: manager, + ) + monkeypatch.setattr( + "deerflow.agents.middlewares.dynamic_context_middleware.resolve_runtime_user_id", + lambda runtime: "alice", + ) + middleware = DynamicContextMiddleware() + state = {"messages": [HumanMessage("answer this", id="message-1")]} + + with pytest.raises(MemoryReadError) as exc_info: + middleware.before_agent(state, None) + + assert exc_info.value.__cause__ is not None + + +def test_strict_scope_mismatch_uses_required_read_error( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager( + tmp_path, + monkeypatch, + failure_policy={"read": "raise"}, + ) + monkeypatch.setattr( + "deerflow.agents.memory.get_memory_manager", + lambda: manager, + ) + monkeypatch.setattr( + "deerflow.agents.middlewares.dynamic_context_middleware.resolve_runtime_user_id", + lambda runtime: "bob", + ) + middleware = DynamicContextMiddleware() + state = {"messages": [HumanMessage("answer this", id="message-1")]} + + with pytest.raises(MemoryReadError, match="owner_user_id 'alice'") as exc_info: + middleware.before_agent(state, None) + with pytest.raises(MemoryReadError, match="owner_user_id 'alice'") as search_error: + manager.search("preferences", user_id="bob", agent_name="research") + + for error in (exc_info.value, search_error.value): + assert isinstance(error.__cause__, MemoryManagerError) + assert "owner_user_id 'alice'" in str(error.__cause__) + assert manager._retriever.calls == [] + + +def test_unreachable_context_read_fail_open_returns_no_injected_context( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + unreachable_openviking_url: str, +) -> None: + manager = _manager( + tmp_path, + monkeypatch, + base_url=unreachable_openviking_url, + timeout_seconds=0.1, + failure_policy={"read": "fail_open"}, + ) + monkeypatch.setattr( + "deerflow.agents.memory.get_memory_manager", + lambda: manager, + ) + monkeypatch.setattr( + "deerflow.agents.middlewares.dynamic_context_middleware.resolve_runtime_user_id", + lambda runtime: "alice", + ) + middleware = DynamicContextMiddleware() + state = {"messages": [HumanMessage("answer this", id="message-1")]} + + assert manager.get_context("alice", agent_name="research") == "" + update = middleware.before_agent(state, None) + assert update is not None + assert all(not str(message.id).endswith("__memory") for message in update["messages"]) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("read_policy", ["raise", "fail_open"]) +async def test_unreachable_async_context_read_honors_policy( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + unreachable_openviking_url: str, + read_policy: str, +) -> None: + manager = _manager( + tmp_path, + monkeypatch, + base_url=unreachable_openviking_url, + timeout_seconds=0.1, + failure_policy={"read": read_policy}, + ) + + if read_policy == "raise": + with pytest.raises(MemoryReadError) as exc_info: + await manager.aget_context("alice", agent_name="research") + assert exc_info.value.__cause__ is not None + else: + assert await manager.aget_context("alice", agent_name="research") == "" + + +@pytest.mark.asyncio +async def test_unreachable_context_read_raise_aborts_async_dynamic_context_injection( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + unreachable_openviking_url: str, +) -> None: + manager = _manager( + tmp_path, + monkeypatch, + base_url=unreachable_openviking_url, + timeout_seconds=0.1, + failure_policy={"read": "raise"}, + ) + monkeypatch.setattr( + "deerflow.agents.memory.get_memory_manager", + lambda: manager, + ) + monkeypatch.setattr( + "deerflow.agents.middlewares.dynamic_context_middleware.resolve_runtime_user_id", + lambda runtime: "alice", + ) + middleware = DynamicContextMiddleware() + state = {"messages": [HumanMessage("answer this", id="message-1")]} + + with pytest.raises(MemoryReadError) as exc_info: + await middleware.abefore_agent(state, None) + + assert exc_info.value.__cause__ is not None + + def test_manager_refuses_to_share_single_user_key_across_deerflow_users( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/docs/OPENVIKING.md b/docs/OPENVIKING.md index 489a6a540..57e9183a5 100644 --- a/docs/OPENVIKING.md +++ b/docs/OPENVIKING.md @@ -155,8 +155,14 @@ boundary; peers separate memory scopes within that user. ## Retry and failure behavior -- `read: fail_open` logs retrieval failures and returns no injected OpenViking - memory. `read: raise` propagates the retrieval failure to its DeerFlow caller. +- `read: fail_open` logs retrieval failures and continues the turn without + recalled OpenViking context. `read: raise` aborts the turn when recall fails. + The same policy applies at DeerFlow's 5-second async injection deadline, + even when the worker pool is full. If configuration or backend discovery + has not finished by then, the unknown policy fails closed. This deadline + limits request waiting, not the blocking operation: an in-flight read keeps + its worker until the backend returns or reaches its own timeout + (`timeout_seconds`, 30 seconds by default). - `write: log_and_drop` logs capture failures without failing an already generated answer. `write: raise` propagates them. - DeerFlow stores only hashes and counters in a bounded local capture cursor