mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-14 16:08:41 +00:00
* feat(extensions): observe task lifecycle and system model calls PR 1 (#4636) gave extensions a middleware chain, and a middleware only sees what passes through the agent graph. Two runtime surfaces stay invisible to it: when a lead run or a subagent begins and ends, and the DeerFlow-owned model calls made outside the graph. This slice adds both, with no new Gateway surface -- routers, services, and the reference extension stay in PR 3. Contract (deerflow-extension-api 0.1.1) --------------------------------------- Two contribution kinds join `middlewares` on the registry: `task_lifecycle` (`on_task_start` / `on_task_stop`, receiving a `TaskInfo` and a conservative `TaskOutcome` of completed / aborted / failed) and `system_model_observer` (`on_system_model_call`, receiving a `SystemOperationKind`, a `SystemModelRequest` snapshot, and a `SystemModelResult` carrying either the response or the provider exception plus a duration). `SystemModelRequest.messages` normalizes to a tuple at construction. Goal evaluation and memory extraction pass a message list while title generation and summarization pass one prompt string, and a bare `str` already satisfies `Sequence` -- without normalization an observer iterating `request.messages` would silently walk characters. Copying also makes the frozen snapshot immutable in fact rather than only by declaration, since observations may run after the call site returns and keeps mutating its own list. Registry marks and rollbacks become per-bucket and positional, so an `install()` that fails after registering two different kinds cannot leave one of them behind. `needs_task_store` now covers all three kinds: a deployment that registers only lifecycle hooks still gets a task store. Task lifecycle -------------- The lead worker notifies start after the run has started and stop after completion persistence and the completion hook, but before clearing the finalizing barrier and publishing the stream end -- holding the barrier across stop is what keeps a same-thread replacement run from overlapping this task's lifecycle. Cancellation raised out of the stop notification is deferred, not propagated in place, so a cancelled run still clears the barrier and emits its end frame. A subagent with a parent `run_id` wraps its execution in the same pair inside `finally`, reporting `parent_task_id` so a delegation tree is reconstructable; a subagent without a `run_id` (embedded client, standalone LangGraph Server) logs and skips rather than inventing a parent. Contributors run in registration order inside one shared 3s budget and every failure is logged and failed open. System model calls ------------------ Four kinds cover the model calls the middleware chain cannot see: goal evaluation, memory extraction, title generation, and summarization. Each site reports both terminal paths without changing the provider exception the host observes, short-circuits on `has_system_model_observers`, and passes the live task store when the runtime has one (detached work gets an isolated store). The sync summarization half stays unobserved on purpose -- it and its only host caller are the sync side of an async-only runtime, so notifying there would block a thread on a call site the host never reaches; the reason is recorded at the call site. The DeerMem backend must stay vendorable and cannot import the extension API, so it reports through a new `MemoryCallbacks.on_memory_llm_result` host hook that the DeerFlow-side callbacks translate into an observation. Notification loop ----------------- Extension resources must be touched on the loop that created them, but subagents can execute on isolated loops and DeerMem runs on a worker thread. The Gateway registers its serving loop before any runtime dependency starts and resets it last through the exit stack, so every startup-failure and cancellation path is covered. Awaited hooks raised on another loop are dispatched across with `run_coroutine_threadsafe` and awaited under the same budget; synchronous sites submit fire-and-forget work. Shutdown stops accepting detached observations before the memory flush -- that flush runs on a worker thread and can emit memory observations -- while keeping the loop alive for awaited task hooks until run and subagent drain completes. Tests ----- `test_extension_task_lifecycle.py`, `test_extension_subagent_lifecycle.py`, and `test_extension_system_model_calls.py` cover ordering, fail-open, budget exhaustion, snapshot binding under a concurrent singleton replacement, the loop-dispatch and shutdown-suspension paths, and both terminal paths at every call site. `test_gateway_run_drain_shutdown.py` pins the stop-before-barrier and drain ordering. * fix(extensions): decide notification fail-open by origin, observe cancellation `_notify_each` only guarded `Exception`, so a contributor letting a `CancelledError` escape — an extension implementing an internal timeout with cancellation, say — skipped its successors and reached the worker's deferred-interrupt path, ending an otherwise successful run as cancelled. Fail-open is about where a failure came from, not its base class: only a genuine cancellation of the host task increments `Task.cancelling()`, so propagate on that and contain everything else. `KeyboardInterrupt` / `SystemExit` still propagate. `observe_system_model_call` skipped observers on cancellation for the same base-class reason, leaving goal / title / summarization silent on a terminal path that is routine — interrupt/rollback admission and shutdown both cancel the run task, with the provider tokens already spent. Awaiting observers there is unreliable (a repeated cancel interrupts that await before any of them runs), so report through the same non-blocking submission the synchronous memory bridge uses, then propagate the cancellation untouched. DeerMem keeps `BaseException` around its provider call, now with the reason recorded: that path runs on a worker thread, where cancelling the awaiting side never interrupts the running thread, so `CancelledError` cannot arrive at all. Its host-hook wrapper narrows to `Exception` — only the hook's own failures are non-fatal, and an observability path must not swallow a process teardown signal. * fix(extensions): warn on budget exhaustion, scope observer logs by task, propagate teardown Review response on #4684: - The memory observation bridge caught BaseException, which would swallow a teardown signal raised while dispatching; it now catches Exception, matching the boundary the DeerMem-side call site documents and tests. - A notification-budget timeout raised mid-hook fell into the generic hook-failure path and logged an asyncio-internal traceback; it now logs a warning like the pre-hook budget skip, while a TimeoutError a contributor raises on its own stays classified as a hook failure. - System model observer logs passed the operation kind as the task id, so log lines said "task goal/title/..."; they now carry the task scope id alongside the kind.
255 lines
10 KiB
Python
255 lines
10 KiB
Python
"""Conformance tests for the MemoryManager interface contract.
|
|
|
|
Pins the three-tier ABC + from_config + invariant validator + async a* +
|
|
callbacks surface so the contract stays stable as backends are added. The
|
|
centerpiece is ``_MinimalBackend`` (implements ONLY from_config + add +
|
|
get_context) -- it instantiates via the factory and runs with everything else
|
|
inherited, proving a new backend needs nothing else. That is the direct
|
|
evidence the optimization lowered onboarding cost (no full method surface, no
|
|
factory edit).
|
|
|
|
Each test resets the singleton + restores config so they are order-independent.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
from pydantic import PrivateAttr
|
|
|
|
from deerflow.agents.memory import MemoryManager, get_memory_manager, reset_memory_manager
|
|
from deerflow.agents.memory.manager import MemoryCallbacks
|
|
from deerflow.config.memory_config import MemoryConfig, get_memory_config, set_memory_config
|
|
|
|
|
|
class _MinimalBackend(MemoryManager):
|
|
"""Implements ONLY tier-1 (add/get_context) + from_config.
|
|
|
|
Everything else inherits base defaults -- the minimal onboarding surface a
|
|
new memory system needs. (No search, no management ops, no fact CRUD, no
|
|
cache reload; ``supports_search`` stays False, so it cannot run in tool mode.)
|
|
"""
|
|
|
|
_adds: list = PrivateAttr(default_factory=list)
|
|
|
|
def add(self, thread_id, messages, *, agent_name=None, user_id=None, trace_id=None) -> None:
|
|
self._adds.append((thread_id, user_id))
|
|
|
|
def get_context(self, user_id, *, agent_name=None, thread_id=None) -> str:
|
|
return f"ctx:{user_id}"
|
|
|
|
@classmethod
|
|
def from_config(cls, backend_config, *, mode="middleware", **host_hooks):
|
|
# Consumes nothing from host_hooks -- a truly minimal backend.
|
|
return cls(backend_config=backend_config or {}, mode=mode)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolate_memory_manager():
|
|
orig = get_memory_config()
|
|
reset_memory_manager()
|
|
yield
|
|
set_memory_config(orig)
|
|
reset_memory_manager()
|
|
|
|
|
|
def test_minimal_backend_onboards_via_factory_with_only_add_get_context():
|
|
"""Centerpiece: a backend implementing ONLY from_config + add + get_context
|
|
instantiates via the factory and runs -- all else inherits defaults. Direct
|
|
evidence that onboarding cost dropped (no full method surface, no factory
|
|
edit)."""
|
|
set_memory_config(MemoryConfig(manager_class=f"{__name__}:_MinimalBackend"))
|
|
manager = get_memory_manager()
|
|
assert isinstance(manager, _MinimalBackend)
|
|
|
|
# tier-1 works
|
|
manager.add("t1", [], user_id="u1")
|
|
assert manager._adds == [("t1", "u1")]
|
|
assert manager.get_context("u1") == "ctx:u1"
|
|
|
|
# tier-2 inherits defaults: add_nowait delegates to add; shutdown_flush=True;
|
|
# the rest raise NotImplementedError.
|
|
manager.add_nowait("t1", [], user_id="u1")
|
|
assert manager._adds[-1] == ("t1", "u1")
|
|
assert manager.shutdown_flush(1.0) is True
|
|
with pytest.raises(NotImplementedError):
|
|
manager.search("q")
|
|
with pytest.raises(NotImplementedError):
|
|
manager.get_memory(user_id="u")
|
|
with pytest.raises(NotImplementedError):
|
|
manager.clear_memory(user_id="u")
|
|
with pytest.raises(NotImplementedError):
|
|
manager.import_memory({}, user_id="u")
|
|
with pytest.raises(NotImplementedError):
|
|
manager.export_memory(user_id="u")
|
|
with pytest.raises(NotImplementedError):
|
|
manager.delete_memory(user_id="u")
|
|
|
|
# tier-3 inherits defaults: warm=None (nothing to warm); B-class no-op;
|
|
# A-class (excl. warm) raise.
|
|
assert manager.warm() is None
|
|
assert manager.on_pre_compress([]) == ""
|
|
assert manager.on_turn_start(1, None) is None
|
|
with pytest.raises(NotImplementedError):
|
|
manager.reload_memory(user_id="u")
|
|
with pytest.raises(NotImplementedError):
|
|
manager.create_fact("x", user_id="u")
|
|
with pytest.raises(NotImplementedError):
|
|
manager.delete_fact("x", user_id="u")
|
|
with pytest.raises(NotImplementedError):
|
|
manager.update_fact("x", user_id="u")
|
|
|
|
|
|
def test_tier1_abstract_enforcement():
|
|
"""A backend missing add or get_context cannot instantiate (TypeError at
|
|
construction -- memory is persistent state, missing write/read is a severe
|
|
bug caught eagerly)."""
|
|
|
|
class _NoAdd(MemoryManager):
|
|
def get_context(self, user_id, *, agent_name=None, thread_id=None) -> str:
|
|
return ""
|
|
|
|
class _NoGet(MemoryManager):
|
|
def add(self, thread_id, messages, *, agent_name=None, user_id=None, trace_id=None) -> None:
|
|
pass
|
|
|
|
with pytest.raises(TypeError):
|
|
_NoAdd()
|
|
with pytest.raises(TypeError):
|
|
_NoGet()
|
|
|
|
|
|
def test_invariant_tool_mode_requires_search():
|
|
"""mode='tool' + a non-search backend raises at instantiation (the agent
|
|
calls memory_search in tool mode, so a non-search backend is a
|
|
misconfiguration -- fail fast). Middleware mode is fine for any backend."""
|
|
with pytest.raises(ValueError):
|
|
_MinimalBackend(backend_config={}, mode="tool")
|
|
assert _MinimalBackend(backend_config={}, mode="middleware").mode == "middleware"
|
|
|
|
|
|
def test_invariant_tool_mode_factory_path_raises():
|
|
"""The invariant also fires on the factory path (mode='tool' + non-search
|
|
backend configured via manager_class)."""
|
|
set_memory_config(MemoryConfig(manager_class=f"{__name__}:_MinimalBackend", mode="tool"))
|
|
with pytest.raises(ValueError):
|
|
get_memory_manager()
|
|
|
|
|
|
class _SearchOverrideForgotFlag(_MinimalBackend):
|
|
"""Overrides search() but forgets supports_search=True (flag/impl drift)."""
|
|
|
|
def search(self, query, top_k=5, *, user_id=None, agent_name=None, category=None):
|
|
return []
|
|
|
|
|
|
class _FlagWithoutSearchOverride(_MinimalBackend):
|
|
"""Sets supports_search=True without overriding search() (flag/impl drift)."""
|
|
|
|
supports_search = True
|
|
|
|
|
|
class _ConsistentSearchBackend(_MinimalBackend):
|
|
"""Overrides search() AND sets supports_search=True -- consistent, tool-OK."""
|
|
|
|
supports_search = True
|
|
|
|
def search(self, query, top_k=5, *, user_id=None, agent_name=None, category=None):
|
|
return []
|
|
|
|
|
|
def test_invariant_supports_search_flag_must_match_override():
|
|
"""supports_search (ClassVar) must match whether search() is overridden, so
|
|
the flag can't drift from the implementation -- caught at instantiation, not
|
|
as a misleading tool-mode rejection (override-but-forgot-flag) or a runtime
|
|
NotImplementedError on the first memory_search call (flag-without-override)."""
|
|
with pytest.raises(ValueError, match="inconsistent"):
|
|
_SearchOverrideForgotFlag(backend_config={})
|
|
with pytest.raises(ValueError, match="inconsistent"):
|
|
_FlagWithoutSearchOverride(backend_config={})
|
|
|
|
|
|
def test_invariant_consistent_search_backend_runs_in_tool_mode():
|
|
"""A backend that overrides search() AND sets supports_search=True is
|
|
consistent and may run in tool mode (the override is the real capability;
|
|
the flag agrees with it)."""
|
|
manager = _ConsistentSearchBackend(backend_config={}, mode="tool")
|
|
assert manager.mode == "tool"
|
|
assert manager.search("q") == []
|
|
|
|
|
|
def test_async_defaults_delegate_to_sync():
|
|
"""a* methods default to the sync path (no concurrency benefit); a future
|
|
async LLM client overrides without changing the contract."""
|
|
manager = _MinimalBackend(backend_config={})
|
|
asyncio.run(manager.aadd("t", [], user_id="u"))
|
|
assert manager._adds == [("t", "u")]
|
|
assert asyncio.run(manager.aget_context("u")) == "ctx:u"
|
|
# asearch delegates to search -> raises (default) just like search.
|
|
with pytest.raises(NotImplementedError):
|
|
asyncio.run(manager.asearch("q"))
|
|
|
|
|
|
def test_callbacks_field_optional_and_noop_default():
|
|
"""callbacks is an optional field (default None); the base
|
|
MemoryCallbacks.on_memory_llm_call is a no-op, so a backend with no
|
|
callbacks runs without tracing."""
|
|
assert _MinimalBackend(backend_config={}).callbacks is None
|
|
noop = MemoryCallbacks()
|
|
# no-op: mutates nothing, raises nothing
|
|
noop.on_memory_llm_call({}, thread_id="t", user_id="u", trace_id="tr", model_name="m")
|
|
noop.on_memory_llm_result(
|
|
{},
|
|
prompt="prompt",
|
|
response="response",
|
|
error=None,
|
|
duration_ms=1.0,
|
|
model_name="m",
|
|
)
|
|
manager = _MinimalBackend(backend_config={}, callbacks=noop)
|
|
assert manager.callbacks is noop
|
|
|
|
|
|
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
|
|
backend (DeerMem) consumes the ones it uses -- see test_deermem_self_contained."""
|
|
manager = _MinimalBackend.from_config(
|
|
{"some_key": "v"},
|
|
mode="middleware",
|
|
callbacks=MemoryCallbacks(),
|
|
host_llm_factory=lambda: None,
|
|
should_keep_hidden_message=lambda ak: True,
|
|
)
|
|
assert isinstance(manager, _MinimalBackend)
|
|
assert manager.backend_config == {"some_key": "v"}
|
|
assert manager.mode == "middleware"
|
|
# The minimal backend consumes NO host_hooks (callbacks / host_llm_factory /
|
|
# should_keep_hidden_message are all ignored) -- callbacks stays None.
|
|
assert manager.callbacks is None
|
|
|
|
|
|
def test_unsupported_op_raises_for_caller_try_except():
|
|
"""Callers (router/client/tools) call tier-3 ops directly and catch
|
|
NotImplementedError (no more hasattr probing). A minimal backend's
|
|
unsupported ops raise, so a caller's try/except degrades cleanly (501 /
|
|
fallback / JSON error)."""
|
|
manager = _MinimalBackend(backend_config={})
|
|
try:
|
|
manager.create_fact("x", user_id="u")
|
|
raise AssertionError("create_fact should have raised NotImplementedError")
|
|
except NotImplementedError:
|
|
pass # caller returns 501 / JSON error / falls back
|
|
|
|
|
|
def test_only_add_and_get_context_are_abstract():
|
|
"""The tier-1 abstract set is exactly {add, get_context} (plus from_config);
|
|
tier-2/3 methods carry defaults so a backend implements only what it supports."""
|
|
assert "add" in MemoryManager.__abstractmethods__
|
|
assert "get_context" in MemoryManager.__abstractmethods__
|
|
assert "from_config" in MemoryManager.__abstractmethods__
|
|
# tier-2/3 are NOT abstract (they have defaults)
|
|
for non_abstract in ("search", "get_memory", "shutdown_flush", "warm", "create_fact", "on_pre_compress"):
|
|
assert non_abstract not in MemoryManager.__abstractmethods__, non_abstract
|