From a46d9221f8ab029ef0b7b8caae25c10ab4153395 Mon Sep 17 00:00:00 2001 From: yetuge <182123929+yetuge@users.noreply.github.com> Date: Wed, 4 Nov 2026 21:54:47 +0800 Subject: [PATCH] fix(summarization): build the summary model off the Gateway event loop (#7874) Context compaction runs from abefore_model/aafter_model on the shared loop, and its async candidate loop constructed the summary provider inline. create_chat_model can read local CLI credentials while it is constructed (Codex auth.json, Claude Code credential files), so an automatic compaction stalled every other request for that read. Split construction out of _model_for and add _amodel_for, which builds the provider on the assembly pool; the sync path keeps building inline. Anchor and failure-cache semantics are unchanged, so a broken candidate still falls through to the next one, and inference still awaits on the loop. Co-authored-by: Willem Jiang --- .../middlewares/summarization_middleware.py | 51 ++++++--- .../packages/harness/deerflow/utils/AGENTS.md | 28 ++--- .../test_summarization_model_offloop.py | 108 ++++++++++++++++++ .../test_extension_system_model_calls.py | 6 +- 4 files changed, 166 insertions(+), 27 deletions(-) create mode 100644 backend/tests/blocking_io/test_summarization_model_offloop.py diff --git a/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py index cac88e6ff4..75113a22aa 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py @@ -33,6 +33,7 @@ from deerflow.extensions.notify import notify_context_compacted from deerflow.models import create_chat_model from deerflow.models.reasoning import resolve_reasoning_contract from deerflow.runtime.events.catalog import MIDDLEWARE_SUMMARIZE_TAG +from deerflow.utils.assembly_io import run_assembly logger = logging.getLogger(__name__) _SUMMARY_TRIGGER_MESSAGE_NAME = "summary" @@ -258,18 +259,12 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): deduped.append(name) return deduped - def _model_for(self, name: str | None) -> Any | None: - """The nostream summary model for ``name``, built lazily and guarded. + def _build_model(self, name: str | None) -> Any | None: + """Construct and tag the nostream summary model; ``None`` on failure. - Returns the pre-built anchor when ``name`` matches it (no rebuild), otherwise - constructs and caches. A construction failure is caught and cached as ``None`` - so a broken candidate config never escapes the fail-open boundary, is never - retried this turn, and still lets the next candidate run. + A construction failure must not escape the fail-open boundary: the + caller caches the failure for this turn so the next candidate can run. """ - if name == self._anchor_model_name: - return self._summary_model - if name in self._model_cache: - return self._model_cache[name] try: model = create_chat_model( name=name, @@ -277,10 +272,40 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): app_config=self._app_config, attach_tracing=False, ) - built = self._tag_nostream(model.with_config(tags=["middleware:summarize"])) + return self._tag_nostream(model.with_config(tags=["middleware:summarize"])) except Exception: logger.exception("Failed to build summary model %r; trying the next candidate", name) - built = None + return None + + def _model_for(self, name: str | None) -> Any | None: + """The nostream summary model for ``name`` on the synchronous path. + + Returns the pre-built anchor when ``name`` matches it (no rebuild), otherwise + constructs and caches. A construction failure is cached as ``None`` so a + broken candidate config never escapes the fail-open boundary, is never + retried this turn, and still lets the next candidate run. + """ + if name == self._anchor_model_name: + return self._summary_model + if name in self._model_cache: + return self._model_cache[name] + built = self._build_model(name) + self._model_cache[name] = built + return built + + async def _amodel_for(self, name: str | None) -> Any | None: + """The nostream summary model for ``name``, built off the event loop. + + ``abefore_model``/``aafter_model`` run on the Gateway loop, and provider + construction can read local CLI credentials (Codex auth.json, Claude Code + credential files), so the build runs on the assembly pool. Anchor and + cache semantics stay identical to :meth:`_model_for`. + """ + if name == self._anchor_model_name: + return self._summary_model + if name in self._model_cache: + return self._model_cache[name] + built = await run_assembly(self._build_model, name) self._model_cache[name] = built return built @@ -392,7 +417,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): return reused for index, name in enumerate(names): text = await self._ainvoke_summary( - self._model_for(name), + await self._amodel_for(name), prompt, last=index == len(names) - 1, model_name=name, diff --git a/backend/packages/harness/deerflow/utils/AGENTS.md b/backend/packages/harness/deerflow/utils/AGENTS.md index 481290bb83..ea4b0107c0 100644 --- a/backend/packages/harness/deerflow/utils/AGENTS.md +++ b/backend/packages/harness/deerflow/utils/AGENTS.md @@ -121,20 +121,22 @@ Display summaries use `strip_leading_think_blocks` before limiting text; ### Agent / Tool Assembly Off-Load One-shot input-polish, follow-up-suggestion, and project-file-summary helpers, -skill moderation (`scan_skill_content`), the goal evaluator, and thread title -generation build providers through `run_assembly` before awaiting inference; -the run worker calls its cached evaluator factory the same way. Provider -construction can read local credentials; keep it off the event loop and the -default executor while preserving request context. Cancellation during -construction must not start inference. A scan without `app_config` (in-graph -`skill_manage`) reads the live config in the same pool, and its fail-closed -fallback reuses that config rather than reading it again on the loop. Other -model construction is separate and not covered here: manual compaction -(`compact_thread_context` builds the summarization middleware and its models -inline) and, inside graph runs, summarization and subagent construction. +skill moderation (`scan_skill_content`), the goal evaluator, thread title +generation, and in-graph context compaction build providers through +`run_assembly` before awaiting inference; the run worker calls its cached +evaluator factory the same way. Provider construction can read local +credentials; keep it off the event loop and the default executor while +preserving request context. Cancellation during construction must not start +inference. A scan without `app_config` (in-graph `skill_manage`) reads the live +config in the same pool, and its fail-closed fallback reuses that config rather +than reading it again on the loop. Other model construction is separate and +not covered here: manual compaction (`compact_thread_context` builds the +summarization middleware and its models inline) and, inside graph runs, +subagent construction. Coverage: `tests/blocking_io/test_oneshot_llm_offloop.py`, -`tests/blocking_io/test_system_model_offloop.py` and -`tests/blocking_io/test_title_model_offloop.py`. +`tests/blocking_io/test_system_model_offloop.py`, +`tests/blocking_io/test_title_model_offloop.py` and +`tests/blocking_io/test_summarization_model_offloop.py`. `start_run` also resolves the factory through `run_assembly` before admission: cold imports cannot occupy the loop or its default pool, and import errors diff --git a/backend/tests/blocking_io/test_summarization_model_offloop.py b/backend/tests/blocking_io/test_summarization_model_offloop.py new file mode 100644 index 0000000000..70af0fb669 --- /dev/null +++ b/backend/tests/blocking_io/test_summarization_model_offloop.py @@ -0,0 +1,108 @@ +"""Auto-compaction must build the summary provider off the Gateway event loop. + +Compaction runs from ``abefore_model``/``aafter_model`` on the shared loop, and +``create_chat_model`` can read local CLI credentials while it is constructed +(Codex ``auth.json``, Claude Code credential files). The real Codex provider is +built here under the strict gate; tracing and inference stay offline. +""" + +import json +import threading +from types import SimpleNamespace + +import pytest +from langchain_core.language_models import BaseChatModel +from langchain_core.messages import AIMessage, HumanMessage +from langchain_core.outputs import ChatGeneration, ChatResult +from pydantic import Field + +from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware +from deerflow.config.app_config import AppConfig +from deerflow.models import factory +from deerflow.models.openai_codex_provider import CodexChatModel + +pytestmark = pytest.mark.asyncio + +_SUMMARY = "Configured summary." + + +class AnchorModel(BaseChatModel): + """The run model; used here only to prove the fallback was not reached.""" + + calls: list[str] = Field(default_factory=list) + + @property + def _llm_type(self): + return "summarization-offloop-anchor" + + def _generate(self, messages, stop=None, run_manager=None, **kwargs): + self.calls.append("anchor") + return ChatResult(generations=[ChatGeneration(message=AIMessage(content="Anchor summary."))]) + + async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs): + return self._generate(messages, stop=stop, run_manager=run_manager, **kwargs) + + +def _codex_config() -> AppConfig: + return AppConfig.model_validate( + { + "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}, + "models": [{"name": "fixture-codex", "model": "fixture-model", "use": "deerflow.models.openai_codex_provider:CodexChatModel"}], + } + ) + + +@pytest.fixture +def codex(tmp_path, monkeypatch): + """Real Codex construction (credential file IO included); offline tracing and inference.""" + credentials = tmp_path / "auth.json" + credentials.write_text(json.dumps({"access_token": "fixture-token", "account_id": "fixture-account"}), encoding="utf-8") + monkeypatch.setenv("CODEX_AUTH_PATH", str(credentials)) + monkeypatch.setattr(factory, "build_tracing_callbacks", lambda **kwargs: []) + state = SimpleNamespace(construction_thread=None, calls=[]) + + async def invoke(model, messages, config=None, **kwargs): + state.calls.append((model, threading.get_ident())) + return AIMessage(content=_SUMMARY) + + monkeypatch.setattr(CodexChatModel, "ainvoke", invoke) + return state + + +def _middleware(anchor): + return DeerFlowSummarizationMiddleware( + model=anchor, + trigger=("messages", 100), + keep=("messages", 1), + trim_tokens_to_summarize=4096, + summary_prompt="{messages}", + app_config=_codex_config(), + configured_model_name="fixture-codex", + run_model_name="fixture-codex", + anchor_model_name="offline", + ) + + +async def test_auto_compaction_builds_summary_model_off_loop(codex, monkeypatch): + anchor = AnchorModel() + middleware = _middleware(anchor) + real_build = DeerFlowSummarizationMiddleware._build_model + + def recording_build(self, name): + codex.construction_thread = threading.get_ident() + return real_build(self, name) + + monkeypatch.setattr(DeerFlowSummarizationMiddleware, "_build_model", recording_build) + + summary = await middleware._asummarize_with([HumanMessage(content="old question", id="user"), AIMessage(content="old answer", id="assistant")]) + + # On the loop the gate rejects the credential read, the candidate is cached + # as unavailable, and compaction returns no summary at all. + assert summary == _SUMMARY + assert len(codex.calls) == 1 + assert codex.construction_thread is not None + assert codex.construction_thread != threading.get_ident() + # Inference itself still awaits on the loop once the provider exists. + assert codex.calls[0][1] == threading.get_ident() + # The configured model was usable, so the run-model anchor never ran. + assert anchor.calls == [] diff --git a/backend/tests/test_extension_system_model_calls.py b/backend/tests/test_extension_system_model_calls.py index ce2131c335..a9a748a68d 100644 --- a/backend/tests/test_extension_system_model_calls.py +++ b/backend/tests/test_extension_system_model_calls.py @@ -381,7 +381,11 @@ async def test_async_summarization_observes_each_provider_attempt_and_live_store middleware.trim_tokens_to_summarize = None middleware._generation_candidate_names = lambda: ["first", "second"] models = {"first": _Failing(), "second": _Working()} - middleware._model_for = lambda name: models[name] + + async def _amodel_for(name): + return models[name] + + middleware._amodel_for = _amodel_for result = await middleware._asummarize_prompt("prompt", task_store=store)