From 30b3a012cb6129f7fd06fbf07e2bae160e43fd0b Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Mon, 26 Oct 2026 16:13:00 +0900 Subject: [PATCH] fix(memory): flush pre-compaction memory off the event loop (#7530) * fix(memory): flush pre-compaction memory off the event loop The async compaction path fired the before-summarization hooks inline. memory_flush_hook hands the summarized messages to add_nowait, which is sync HTTP for mem0 and Honcho (inherited default -> add) and OpenViking (its own override), so every automatic compaction with an HTTP memory backend blocked the Gateway event loop for the whole request. acompact_state now runs the hooks through asyncio.to_thread, as MemoryMiddleware.aafter_agent already does for its writes. Ordering, failure isolation, summary-first timing and context propagation are unchanged; the sync compact_state path is untouched. * docs(changelog): reference #7530 for the pre-compaction memory flush fix * test(memory): anchor the OpenViking pre-compaction flush off the event loop --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 8 + CHANGELOG_zh.md | 6 + .../harness/deerflow/agents/memory/AGENTS.md | 2 + .../middlewares/summarization_middleware.py | 27 ++- .../test_summarization_memory_flush.py | 213 ++++++++++++++++++ .../tests/test_summarization_middleware.py | 34 +++ 6 files changed, 288 insertions(+), 2 deletions(-) create mode 100644 backend/tests/blocking_io/test_summarization_memory_flush.py diff --git a/CHANGELOG.md b/CHANGELOG.md index aee605f2a9..ee9fa16edc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1397,6 +1397,13 @@ As of 2026-10-21, the milestone includes **829 merged pull requests**. ### Fixed +- **memory:** Automatic context compaction no longer blocks the event loop + while it saves the messages it is about to summarize to memory. With the + mem0, Honcho or OpenViking backend, that save is a synchronous HTTP request, + so the Gateway stopped serving other requests and streams until the memory + server answered. It now runs in a worker, like the after-turn memory write. + ([#7530]) + - **channels:** IM channel runs now use the top-level `recursion_limit`, as Gateway API runs do, instead of a fixed `100` that could end a multi-step task after a few tool-calling turns without an answer. A `recursion_limit` @@ -13099,3 +13106,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#7449]: https://github.com/bytedance/deer-flow/pull/7449 [#7507]: https://github.com/bytedance/deer-flow/pull/7507 [#7521]: https://github.com/bytedance/deer-flow/pull/7521 +[#7530]: https://github.com/bytedance/deer-flow/pull/7530 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index be7ffe58d8..d25a6d64d5 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -1112,6 +1112,11 @@ ### 修复 +- **记忆：** 自动上下文压缩在把即将被总结的消息保存到记忆时，不再阻塞事件循环。 + 使用 mem0、Honcho 或 OpenViking 后端时，这次保存是同步 HTTP 请求，Gateway 在记忆 + 服务器响应之前无法处理其他请求和流式输出。现在它与每轮结束后的记忆写入一样在工作 + 线程中执行。([#7530]) + - **渠道：** IM 渠道的运行现在与 Gateway API 运行一样使用顶层 `recursion_limit`，不再固定 发送 `100`；此前多步任务可能在几轮工具调用后就中止，用户收不到回答。IM 渠道 `session.config` 中的 `recursion_limit` 仍然优先；早先的 README 示例把它设为 `100`， @@ -10457,3 +10462,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#7449]: https://github.com/bytedance/deer-flow/pull/7449 [#7507]: https://github.com/bytedance/deer-flow/pull/7507 [#7521]: https://github.com/bytedance/deer-flow/pull/7521 +[#7530]: https://github.com/bytedance/deer-flow/pull/7530 diff --git a/backend/packages/harness/deerflow/agents/memory/AGENTS.md b/backend/packages/harness/deerflow/agents/memory/AGENTS.md index 8da58177d0..d1e1152ae9 100644 --- a/backend/packages/harness/deerflow/agents/memory/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/memory/AGENTS.md @@ -24,6 +24,8 @@ This directory owns memory capture, storage, retrieval, prompt injection, and mo - `manager.py` defines the backend-neutral `MemoryManager` contract. - `agents/middlewares/memory_middleware.py` queues filtered conversations for passive capture. - `summarization_hook.py` connects memory work to the summarization lifecycle. + Async compaction runs it in a worker: `add_nowait` is sync HTTP for mem0, + Honcho and OpenViking (`tests/blocking_io/test_summarization_memory_flush.py`). - `tools.py` provides `memory_search`, `memory_get`, `memory_add`, `memory_update`, and `memory_delete`. - `backends/deermem/` contains the default local backend. - `backends/mem0/`, `backends/openviking/`, and `backends/honcho/` contain optional adapters. diff --git a/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py index 4ef5a3a914..13a3f6279c 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import hashlib import html import logging @@ -85,7 +86,11 @@ class ContextCompactionResult: @runtime_checkable class BeforeSummarizationHook(Protocol): - """Hook invoked before summarization removes messages from state.""" + """Hook invoked before summarization removes messages from state. + + Hooks are synchronous and may block. The async compaction path runs them + in a worker thread, so a hook must not assume it runs on the event loop. + """ def __call__(self, event: SummarizationEvent) -> None: ... @@ -933,7 +938,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): summary = str(summary) self._record_summary_telemetry(runtime, previous_summary, summary, len(messages_to_summarize), llm_call_skipped=llm_call_skipped) # Fire hooks only once a replacement summary exists (see compact_state). - self._fire_hooks(messages_to_summarize, preserved_messages, runtime) + await self._afire_hooks(messages_to_summarize, preserved_messages, runtime) self._record_compaction( source_content_hashes, summary=summary, @@ -1038,6 +1043,24 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): hook_name = getattr(hook, "__name__", None) or type(hook).__name__ logger.exception("before_summarization hook %s failed", hook_name) + async def _afire_hooks( + self, + messages_to_summarize: list[AnyMessage], + preserved_messages: list[AnyMessage], + runtime: Runtime, + ) -> None: + """Async counterpart of :meth:`_fire_hooks`, run in a worker thread. + + ``memory_flush_hook`` hands the messages to the memory backend, and + the mem0, Honcho and OpenViking backends write over sync HTTP, so + running the hooks inline would stall the event loop for the whole + request. ``asyncio.to_thread`` copies the context, so hooks still see + the run's LangGraph config and user context. + """ + if not self._before_summarization_hooks: + return + await asyncio.to_thread(self._fire_hooks, messages_to_summarize, preserved_messages, runtime) + def _summary_model_thinking_enabled(name: str | None, app_config: Any) -> bool: """Return a construction mode accepted by the model's reasoning contract. diff --git a/backend/tests/blocking_io/test_summarization_memory_flush.py b/backend/tests/blocking_io/test_summarization_memory_flush.py new file mode 100644 index 0000000000..9f0363d0ec --- /dev/null +++ b/backend/tests/blocking_io/test_summarization_memory_flush.py @@ -0,0 +1,213 @@ +"""Regression anchors: the async compaction path must flush memory off the loop. + +``memory_flush_hook`` hands the about-to-be-summarized messages to the memory +backend's ``add_nowait``. mem0 and Honcho inherit the default, which calls +their sync HTTP ``add``, and OpenViking's override runs its sync session +recorder, so firing the hook inline from ``acompact_state`` blocked the event +loop for the whole backend request. +""" + +from __future__ import annotations + +from contextlib import nullcontext +from pathlib import Path +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from langchain_core.messages import AIMessage, HumanMessage + +from deerflow.agents.memory.backends.honcho.honcho_manager import HonchoMemoryManager +from deerflow.agents.memory.backends.mem0.mem0_manager import Mem0Manager +from deerflow.agents.memory.backends.openviking.openviking_manager import OpenVikingMemoryManager +from deerflow.agents.memory.summarization_hook import memory_flush_hook +from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware +from deerflow.config.memory_config import MemoryConfig + +HOOK_MODULE = "deerflow.agents.memory.summarization_hook" + + +class _BlockingWrites: + """Records a backend write only after doing real blocking file IO.""" + + def __init__(self, probe_dir: Path) -> None: + self.probe_dir = probe_dir + self._probe = probe_dir / "probe.txt" + self.written: list[str] = [] + + def write(self, contents: list[str]) -> None: + self._probe.write_text("blocked io") + self.written.extend(contents) + + +class _FakeMem0Client: + def __init__(self, writes: _BlockingWrites) -> None: + self._writes = writes + + def add_memories(self, *, messages: list[dict[str, str]], **_: Any) -> None: + self._writes.write([message["content"] for message in messages]) + + def close(self) -> None: + pass + + +class _FakeHonchoClient: + def __init__(self, writes: _BlockingWrites) -> None: + self._writes = writes + + def get_or_create_peer(self, ws: str, pid: str) -> None: + pass + + def get_or_create_session(self, ws: str, sid: str) -> None: + pass + + def set_session_peers(self, ws: str, sid: str, pids: list[str]) -> None: + pass + + def add_messages(self, ws: str, sid: str, msgs: list[dict[str, str]]) -> None: + self._writes.write([message["content"] for message in msgs]) + + def close(self) -> None: + pass + + +class _OpenVikingCommitPolicy: + def __init__(self, *, mode: str) -> None: + self.mode = mode + + +class _OpenVikingClient: + supports_request_actor_peer = True + + def health(self) -> bool: + return True + + def close(self) -> None: + pass + + +class _FakeOpenVikingRecorder: + def __init__(self, **_: Any) -> None: + self.client = _OpenVikingClient() + self.writes: _BlockingWrites | None = None + + def record(self, session_id: str, messages: list[Any], peer_id: str | None = None) -> None: + assert self.writes is not None + self.writes.write([message["content"] if isinstance(message, dict) else message.content for message in messages]) + + def flush(self, session_id: str) -> None: + pass + + def close(self) -> None: + pass + + +class _OpenVikingRetriever: + def __init__(self, **kwargs: Any) -> None: + self.__dict__.update(kwargs) + + +def _mem0(writes: _BlockingWrites, monkeypatch: pytest.MonkeyPatch) -> Mem0Manager: + monkeypatch.setenv("MEM0_API_KEY", "test-key") + manager = Mem0Manager(backend_config={}) + manager._client = _FakeMem0Client(writes) + return manager + + +def _honcho(writes: _BlockingWrites, monkeypatch: pytest.MonkeyPatch) -> HonchoMemoryManager: + manager = HonchoMemoryManager.from_config({"base_url": "http://honcho.test"}) + manager._client = _FakeHonchoClient(writes) + return manager + + +def _openviking(writes: _BlockingWrites, monkeypatch: pytest.MonkeyPatch) -> OpenVikingMemoryManager: + import deerflow.agents.memory.backends.openviking.openviking_manager as module + + monkeypatch.setenv("OPENVIKING_API_KEY", "user-key") + monkeypatch.setattr( + module, + "_load_official_integration", + lambda: { + "OpenVikingCommitPolicy": _OpenVikingCommitPolicy, + "OpenVikingPartialWriteError": RuntimeError, + "OpenVikingRetriever": _OpenVikingRetriever, + "OpenVikingSessionRecorder": _FakeOpenVikingRecorder, + "use_actor_peer": lambda peer_id: nullcontext(), + }, + ) + manager = OpenVikingMemoryManager.from_config( + { + "base_url": "http://openviking:1933", + "storage_path": str(writes.probe_dir), + "owner_user_id": "alice", + "startup_policy": "warn", + } + ) + manager._recorder.writes = writes + return manager + + +_BACKENDS = [pytest.param(_mem0, id="mem0"), pytest.param(_honcho, id="honcho"), pytest.param(_openviking, id="openviking")] + + +def _middleware() -> DeerFlowSummarizationMiddleware: + model = MagicMock() + model.ainvoke = AsyncMock(return_value=SimpleNamespace(text="compressed summary")) + model.with_config.return_value = model + return DeerFlowSummarizationMiddleware( + model=model, + trigger=("messages", 4), + keep=("messages", 2), + token_counter=len, + before_summarization=[memory_flush_hook], + ) + + +def _messages() -> list[Any]: + return [ + HumanMessage("I prefer green tea", id="h1"), + AIMessage("Noted", id="a1"), + HumanMessage("Plan my week", id="h2"), + AIMessage("Here is a plan", id="a2"), + ] + + +_RUNTIME = SimpleNamespace(context={"thread_id": "thread-1", "user_id": "alice"}) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("make_manager", _BACKENDS) +async def test_async_compaction_flushes_memory_off_the_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, make_manager: Any) -> None: + writes = _BlockingWrites(tmp_path) + manager = make_manager(writes, monkeypatch) + middleware = _middleware() + + with patch(HOOK_MODULE + ".get_memory_config", return_value=MemoryConfig(enabled=True)), patch(HOOK_MODULE + ".get_memory_manager", return_value=manager): + result = await middleware.abefore_model({"messages": _messages()}, _RUNTIME) + + assert result is not None + assert result["summary_text"] == "compressed summary" + # The hook swallows failures, so a write that tripped the gate on the loop + # would only be logged; the landed write is what proves it ran off-loop. + assert writes.written == ["I prefer green tea", "Noted"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("make_manager", _BACKENDS) +async def test_blocking_probe_inline_flush_actually_trips_the_gate(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, make_manager: Any, caplog: pytest.LogCaptureFixture) -> None: + """Meta-check: firing the hooks inline on the loop is caught by the gate.""" + writes = _BlockingWrites(tmp_path) + manager = make_manager(writes, monkeypatch) + middleware = _middleware() + messages = _messages() + + with ( + patch(HOOK_MODULE + ".get_memory_config", return_value=MemoryConfig(enabled=True)), + patch(HOOK_MODULE + ".get_memory_manager", return_value=manager), + caplog.at_level("DEBUG"), + ): + middleware._fire_hooks(messages[:2], messages[2:], _RUNTIME) + + assert writes.written == [] + assert "Blocking call" in caplog.text diff --git a/backend/tests/test_summarization_middleware.py b/backend/tests/test_summarization_middleware.py index 5212dbafe8..ead7a9bb5a 100644 --- a/backend/tests/test_summarization_middleware.py +++ b/backend/tests/test_summarization_middleware.py @@ -1,5 +1,7 @@ from __future__ import annotations +import contextvars +import threading from types import SimpleNamespace from unittest import mock from unittest.mock import AsyncMock, MagicMock @@ -376,6 +378,38 @@ async def test_abefore_model_calls_hooks_same_as_sync() -> None: assert [message.content for message in captured[0].messages_to_summarize] == ["user-1", "assistant-1"] +_HOOK_CONTEXT: contextvars.ContextVar[str] = contextvars.ContextVar("_HOOK_CONTEXT") + + +@pytest.mark.anyio +async def test_abefore_model_runs_hooks_in_a_worker_with_sync_semantics(caplog: pytest.LogCaptureFixture) -> None: + loop_thread = threading.get_ident() + calls: list[tuple[str, int, str | None]] = [] + + def _hook(name: str): + def hook(_: SummarizationEvent) -> None: + calls.append((name, threading.get_ident(), _HOOK_CONTEXT.get(None))) + + return hook + + def _broken_hook(_: SummarizationEvent) -> None: + raise RuntimeError("hook failure") + + middleware = _middleware(before_summarization=[_hook("first"), _broken_hook, _hook("second")]) + token = _HOOK_CONTEXT.set("run-context") + try: + with caplog.at_level("ERROR"): + result = await middleware.abefore_model({"messages": _messages()}, _runtime()) + finally: + _HOOK_CONTEXT.reset(token) + + assert [name for name, _, _ in calls] == ["first", "second"] + assert all(thread != loop_thread for _, thread, _ in calls) + assert all(context == "run-context" for _, _, context in calls) + assert "before_summarization hook _broken_hook failed" in caplog.text + assert isinstance(result["messages"][0], RemoveMessage) + + def test_memory_flush_hook_skips_when_memory_disabled(monkeypatch: pytest.MonkeyPatch) -> None: manager = MagicMock() monkeypatch.setattr("deerflow.agents.memory.summarization_hook.get_memory_config", lambda: MemoryConfig(enabled=False))