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 <willem.jiang@gmail.com>

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

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

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.

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.

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

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))