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)