fix(summarization): summarize with the run model, fall back on summary-provider failure (#4361)

* fix(summarization): own the run model for compaction; bound failure

With summarization.model_name: null the summary model resolved to
config.models[0] while the executing model is selected per run; when they
differ and models[0]'s provider is broken (expired key, quota, outage)
compaction silently failed every triggered turn and context grew unbounded
until the main provider 400s the run (#3103's shape), even though the run's
own model was healthy.

Model ownership is now sourced from the builders, not re-derived at runtime:
- The lead, subagent, and manual /compact builders each pass the resolved run
  model into create_summarization_middleware(run_model_name=...). The middleware
  no longer reads runtime.context / get_config(), which do not carry a custom
  agent's or a subagent's resolved model, so a custom-agent lead run and a
  distinct-model subagent now summarize with their own model, not models[0] /
  the parent's. Runtime re-resolution and the per-name model cache are removed.
- model_name: null summarizes with the run's own model; an explicitly configured
  summary model generates and falls back to the run model on failure. The
  fallback is built lazily after the primary fails and its construction is
  guarded, so a broken fallback cannot skip a healthy primary or escape the
  automatic failure boundary.

Failure is bounded and side-effect-safe:
- An empty or whitespace-only response is treated as a generation failure, not a
  valid summary, so compaction never removes all history for an empty replacement.
- compact_state/acompact_state take raise_on_failure independent of force: the
  manual /compact path always surfaces a generation failure (even force=false)
  and routes it to the existing ContextCompactionFailed path (HTTP 500 ->
  frontend error toast) instead of an unconsumed response reason. The automatic
  path leaves compaction state unchanged.
- before_summarization hooks fire only after a replacement summary exists.

SummarizationConfig.model_name, config.example.yaml, and docs/summarization.md
document the final lead/subagent/manual ownership rules.

Part of RFC #4346 (section A). Evaluating fraction/triggers against the run
model's profile (profile ownership) is a separate follow-up.

* fix(summarization): manual /compact model ownership + fail-open construct/parse

Manual /compact carried only agent_name, so it derived the run model from the
custom-agent model or config.models[0] and missed the request-selected model the
run path uses (request -> custom-agent -> default). Carry model_name through
ThreadCompactRequest and the frontend compact call, resolve with the same
precedence, and move the custom-agent config read off the event loop (asyncio
.to_thread) with user_id so the strict blocking-IO gate is not bypassed by the
broad except.

Make one summary attempt own its full lifecycle so the fail-open boundary covers
construction and response parsing, not just invocation: build each candidate model
lazily and guarded (a raising constructor falls through to the healthy run model
instead of breaking agent construction), build the model_name:null primary from the
run model rather than config.models[0], and run response text extraction inside the
invocation try so a failing .text accessor falls back instead of escaping compaction.
Adds factory-level constructor-failure, response-extraction-failure (sync/async), and
route-path model-ownership tests.
This commit is contained in:
Aari 2026-07-26 07:39:39 +08:00 committed by GitHub
parent b41a8d44df
commit 37c343fe30
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 1144 additions and 86 deletions

View File

@ -442,6 +442,7 @@ class ThreadCompactRequest(BaseModel):
force: bool = Field(default=True, description="Run compaction even if automatic summarization thresholds are not met") force: bool = Field(default=True, description="Run compaction even if automatic summarization thresholds are not met")
keep: ContextSize | None = Field(default=None, description="Optional retention policy for this compaction only") keep: ContextSize | None = Field(default=None, description="Optional retention policy for this compaction only")
agent_name: str | None = Field(default=None, max_length=128, description="Optional custom agent name for memory attribution") agent_name: str | None = Field(default=None, max_length=128, description="Optional custom agent name for memory attribution")
model_name: str | None = Field(default=None, max_length=128, description="Optional model to summarize with; resolved request override -> custom-agent model -> default, mirroring run model selection")
class ThreadCompactResponse(BaseModel): class ThreadCompactResponse(BaseModel):
@ -1128,6 +1129,7 @@ async def compact_thread(thread_id: str, body: ThreadCompactRequest, request: Re
force=body.force, force=body.force,
user_id=get_effective_user_id(), user_id=get_effective_user_id(),
agent_name=body.agent_name, agent_name=body.agent_name,
model_name=body.model_name,
) )
except ConflictError: except ConflictError:
raise HTTPException(status_code=409, detail="Thread has a run in flight. Compact after the run finishes.") from None raise HTTPException(status_code=409, detail="Thread has a run in flight. Compact after the run finishes.") from None

View File

@ -21,7 +21,7 @@ Summarization is configured in `config.yaml` under the `summarization` key:
```yaml ```yaml
summarization: summarization:
enabled: true enabled: true
model_name: null # Use default model or specify a lightweight model model_name: null # null = summarize with the run's own model (see below); or name a lightweight model
# Trigger conditions (OR logic - any condition triggers summarization) # Trigger conditions (OR logic - any condition triggers summarization)
trigger: trigger:
@ -61,8 +61,11 @@ summarization:
#### `model_name` #### `model_name`
- **Type**: String or null - **Type**: String or null
- **Default**: `null` (uses default model) - **Default**: `null`
- **Description**: Model to use for generating summaries. Recommended to use a lightweight, cost-effective model like `gpt-4o-mini` or equivalent. - **Description**: Model to use for generating summaries.
- **`null` (model ownership)**: summarize with the model the run actually executes with — the lead run's resolved model, a subagent's own model, or a thread's custom-agent model — **not** `config.models[0]`. This keeps compaction working on a run whose model is healthy even when `models[0]`'s provider is broken (expired key, quota, outage).
- **Set to a model name**: that model generates summaries. If its provider fails, compaction **falls back to the run's own model** so a broken summary provider cannot disable compaction while a working model is available. Recommended to use a lightweight, cost-effective model like `gpt-4o-mini` or equivalent.
- Ownership applies to all three paths — automatic lead compaction, subagent compaction, and manual `/compact`. Manual `/compact` resolves the run model with the same precedence as a normal run: the model selected for the request (`POST /api/threads/{id}/compact` body `model_name`, sent by the frontend from the composer's current model) → the thread's custom-agent model → the default. A whitespace-only summary response is treated as a generation failure (it is never committed as a valid empty summary).
#### `trigger` #### `trigger`
- **Type**: Single `ContextSize` or list of `ContextSize` objects - **Type**: Single `ContextSize` or list of `ContextSize` objects
@ -223,9 +226,9 @@ The middleware intelligently preserves message context:
- Summaries don't require the most powerful models - Summaries don't require the most powerful models
- Significant cost savings on high-volume applications - Significant cost savings on high-volume applications
- **Default**: If `model_name` is `null`, uses the default model - **Default**: If `model_name` is `null`, summarizes with the run's own model (not `models[0]`)
- May be more expensive but ensures consistency - Keeps compaction working when `models[0]`'s provider is broken but the run's model is healthy
- Good for simple setups - Good for simple setups; no separate summary provider to keep credentialed
### Optimization Tips ### Optimization Tips

View File

@ -133,9 +133,14 @@ def _resolve_model_name(requested_model_name: str | None = None, *, app_config:
return default_model_name return default_model_name
def _create_summarization_middleware(*, app_config: AppConfig | None = None) -> DeerFlowSummarizationMiddleware | None: def _create_summarization_middleware(*, app_config: AppConfig | None = None, run_model_name: str | None = None) -> DeerFlowSummarizationMiddleware | None:
"""Create and configure the summarization middleware from config.""" """Create and configure the summarization middleware from config.
return create_summarization_middleware(app_config=app_config)
``run_model_name`` is the resolved run model; it is the source of truth for
``model_name: null`` summarization and the explicit-summary-model fallback, so a
custom agent's model is used instead of ``config.models[0]``.
"""
return create_summarization_middleware(app_config=app_config, run_model_name=run_model_name)
def _create_todo_list_middleware(is_plan_mode: bool) -> TodoMiddleware | None: def _create_todo_list_middleware(is_plan_mode: bool) -> TodoMiddleware | None:
@ -359,7 +364,7 @@ def build_middlewares(
) )
# Add summarization middleware if enabled # Add summarization middleware if enabled
summarization_middleware = _create_summarization_middleware(app_config=resolved_app_config) summarization_middleware = _create_summarization_middleware(app_config=resolved_app_config, run_model_name=model_name)
if summarization_middleware is not None: if summarization_middleware is not None:
middlewares.append(summarization_middleware) middlewares.append(summarization_middleware)

View File

@ -21,6 +21,25 @@ from deerflow.models import create_chat_model
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
_SUMMARY_TRIGGER_MESSAGE_NAME = "summary" _SUMMARY_TRIGGER_MESSAGE_NAME = "summary"
_UNSET = object()
# Valid non-generated summaries for the empty / too-long-to-summarize edges; these
# short-circuit model invocation (and must not be treated as generation failures).
_CANNED_SUMMARIES = frozenset(
{
"No previous conversation history.",
"Previous conversation was too long to summarize.",
}
)
class SummaryGenerationError(RuntimeError):
"""Summary generation failed after exhausting the run-model fallback.
Raised only when a caller opts in via ``raise_on_failure`` (the manual
``/compact`` path) so a real failure is reported distinctly from "nothing to
compact". The automatic path leaves ``raise_on_failure`` False and swallows the
failure, leaving compaction state unchanged for the turn.
"""
@dataclass(frozen=True) @dataclass(frozen=True)
@ -82,22 +101,115 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
self, self,
*args, *args,
before_summarization: list[BeforeSummarizationHook] | None = None, before_summarization: list[BeforeSummarizationHook] | None = None,
app_config: Any | None = None,
configured_model_name: str | None = None,
run_model_name: str | None = None,
anchor_model_name: str | None = _UNSET, # type: ignore[assignment]
**kwargs, **kwargs,
) -> None: ) -> None:
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
self._before_summarization_hooks = before_summarization or [] self._before_summarization_hooks = before_summarization or []
# Model-ownership state. The model that actually executes the run is selected
# per run and is the authoritative source of truth, so the caller (lead /
# subagent / manual builders) supplies it directly as ``run_model_name``
# instead of the middleware re-deriving it from ``runtime.context`` /
# ``get_config()`` — those fields do not carry a custom agent's or a subagent's
# resolved model.
#
# ``configured_model_name`` is the explicitly configured summary model
# (``None`` => summarize with the run's own model). ``run_model_name`` is the
# model the run executes with; when they differ and the summary provider is
# broken (expired key, quota, outage) the run's own working model can still
# compact.
self._app_config = app_config
self._configured_summary_model_name = configured_model_name
self._run_model_name = run_model_name
# The summary LLM call runs inside a LangGraph middleware hook, so its token # The summary LLM call runs inside a LangGraph middleware hook, so its token
# stream would otherwise be captured by the messages-tuple stream callback and # stream would otherwise be captured by the messages-tuple stream callback and
# broadcast to the frontend as a phantom AI message. Tag a dedicated model copy # broadcast to the frontend as a phantom AI message. Tag a dedicated model copy
# with TAG_NOSTREAM so the streaming handler skips it. # with TAG_NOSTREAM so the streaming handler skips it.
# Keep self.model untagged so the parent's profile / ls_params inspection still works. # Keep self.model untagged so the parent's profile / ls_params inspection still works.
# self._summary_model = self._tag_nostream(self.model)
# Preserve any tags already bound on the model (e.g. "middleware:summarize" set in # ``self.model`` is the pre-built *anchor* model: it drives the parent's token
# lead_agent/agent.py for RunJournal attribution): RunnableBinding.with_config does a # counter / profile inspection and is reused verbatim by generation when a
# shallow merge that would otherwise overwrite the existing tags list entirely. # candidate matches its name. The factory builds it guarded and passes its name
existing_tags = list((getattr(self.model, "config", None) or {}).get("tags") or []) # explicitly; direct construction (tests) mirrors the old factory choice
# (configured model, else default) so the passed ``model`` is the primary.
if anchor_model_name is _UNSET:
self._anchor_model_name = configured_model_name or self._default_model_name()
else:
self._anchor_model_name = anchor_model_name
# Nostream generation models built lazily by name and cached (None = a build
# that failed, so a broken candidate config is not retried every turn and does
# not escape the fail-open boundary).
self._model_cache: dict[str | None, Any] = {}
def _tag_nostream(self, model: Any) -> Any:
"""Return a copy of ``model`` carrying TAG_NOSTREAM without clobbering tags.
lead_agent/agent.py binds "middleware:summarize" for RunJournal attribution;
RunnableBinding.with_config shallow-merges config, so existing tags must be
preserved explicitly instead of being overwritten with just [TAG_NOSTREAM].
"""
existing_tags = list((getattr(model, "config", None) or {}).get("tags") or [])
merged_tags = [*existing_tags, TAG_NOSTREAM] if TAG_NOSTREAM not in existing_tags else existing_tags merged_tags = [*existing_tags, TAG_NOSTREAM] if TAG_NOSTREAM not in existing_tags else existing_tags
self._summary_model = self.model.with_config(tags=merged_tags) return model.with_config(tags=merged_tags)
def _default_model_name(self) -> str | None:
if self._app_config is None:
return None
models = getattr(self._app_config, "models", None)
return models[0].name if models else None
def _generation_candidate_names(self) -> list[str | None]:
"""Ordered summary-generation candidates by name (deduplicated).
Explicit summary model: the configured model first, then the run's own model
as a distinct fallback. ``model_name: null``: the run's own model only — its
construction is the primary, so there is no eager dependency on
``config.models[0]`` (a bare default is used only when no run model was
resolved). A ``None`` entry means "let ``create_chat_model`` pick the default",
which only occurs when nothing resolves a name.
"""
default = self._default_model_name()
if self._configured_summary_model_name is not None:
names = [self._configured_summary_model_name, self._run_model_name or default]
else:
names = [self._run_model_name or default]
deduped: list[str | None] = []
seen: set[str | None] = set()
for name in names:
if name in seen:
continue
seen.add(name)
deduped.append(name)
return deduped
def _model_for(self, name: str | None) -> Any | None:
"""The nostream summary model for ``name``, built lazily and guarded.
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.
"""
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,
thinking_enabled=False,
app_config=self._app_config,
attach_tracing=False,
)
built = 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
self._model_cache[name] = built
return built
@override @override
def _create_summary(self, messages_to_summarize: list[AnyMessage]) -> str | None: def _create_summary(self, messages_to_summarize: list[AnyMessage]) -> str | None:
@ -107,6 +219,31 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
async def _acreate_summary(self, messages_to_summarize: list[AnyMessage]) -> str | None: async def _acreate_summary(self, messages_to_summarize: list[AnyMessage]) -> str | None:
return await self._asummarize_with(messages_to_summarize) return await self._asummarize_with(messages_to_summarize)
def _prepare_summary_prompt(self, messages_to_summarize: list[AnyMessage], previous_summary: str | None) -> str | None:
"""Return the formatted prompt, or a canned string for the empty/too-long edges.
A non-``None`` return that is not a real prompt (the two canned strings) is a
valid summary and short-circuits generation; ``None`` means "build a prompt".
"""
if not messages_to_summarize:
return "No previous conversation history."
prompt = self._build_summary_prompt(messages_to_summarize, previous_summary=previous_summary)
if prompt is None:
return "Previous conversation was too long to summarize."
return prompt
@staticmethod
def _nonempty_summary(text: Any) -> str | None:
"""Normalize a model response's text; a blank/whitespace-only body is a failure.
Committing ``""`` as a summary would fire the before_summarization hooks and
remove all prior history for an empty replacement, so an empty body is treated
as generation failure (try the fallback, or leave state unchanged) rather than
a valid summary.
"""
stripped = text.strip() if isinstance(text, str) else ""
return stripped or None
def _summarize_with(self, messages_to_summarize: list[AnyMessage], previous_summary: str | None = None) -> str | None: def _summarize_with(self, messages_to_summarize: list[AnyMessage], previous_summary: str | None = None) -> str | None:
"""Mirror the parent ``_create_summary`` but invoke the nostream-tagged model. """Mirror the parent ``_create_summary`` but invoke the nostream-tagged model.
@ -114,38 +251,84 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
cached and reused across concurrent runs, so a temporary swap would leak the cached and reused across concurrent runs, so a temporary swap would leak the
``RunnableBinding`` to other coroutines during ``await`` and break parent logic ``RunnableBinding`` to other coroutines during ``await`` and break parent logic
that inspects the raw model (``profile`` / ``_get_ls_params``). that inspects the raw model (``profile`` / ``_get_ls_params``).
Generation uses the run's own model (``model_name: null``) or the explicitly
configured summary model, falling back to the run model on failure so a broken
summary provider cannot disable compaction while a working model is available.
""" """
if not messages_to_summarize: prompt = self._prepare_summary_prompt(messages_to_summarize, previous_summary)
return "No previous conversation history." if prompt is None or prompt in _CANNED_SUMMARIES:
prompt = self._build_summary_prompt(messages_to_summarize, previous_summary=previous_summary) return prompt
if prompt is None: # Walk the ordered candidates; each attempt owns its full lifecycle (lazy
return "Previous conversation was too long to summarize." # guarded construction -> invoke -> text extraction -> non-empty validation),
try: # and any failure at any stage falls through to the next candidate. When all
response = self._summary_model.invoke( # candidates fail the caller leaves compaction state unchanged.
prompt, names = self._generation_candidate_names()
config={"metadata": {"lc_source": "summarization"}}, for index, name in enumerate(names):
) text = self._invoke_summary(self._model_for(name), prompt, last=index == len(names) - 1)
return response.text.strip() if text is not None:
except Exception: return text
logger.exception("Summary generation failed; skipping compaction this turn") return None
return None
async def _asummarize_with(self, messages_to_summarize: list[AnyMessage], previous_summary: str | None = None) -> str | None: async def _asummarize_with(self, messages_to_summarize: list[AnyMessage], previous_summary: str | None = None) -> str | None:
"""Async counterpart of :meth:`_summarize_with` using the nostream model.""" """Async counterpart of :meth:`_summarize_with` using the nostream model."""
if not messages_to_summarize: prompt = self._prepare_summary_prompt(messages_to_summarize, previous_summary)
return "No previous conversation history." if prompt is None or prompt in _CANNED_SUMMARIES:
prompt = self._build_summary_prompt(messages_to_summarize, previous_summary=previous_summary) return prompt
if prompt is None: names = self._generation_candidate_names()
return "Previous conversation was too long to summarize." for index, name in enumerate(names):
try: text = await self._ainvoke_summary(self._model_for(name), prompt, last=index == len(names) - 1)
response = await self._summary_model.ainvoke( if text is not None:
prompt, return text
config={"metadata": {"lc_source": "summarization"}}, return None
)
return response.text.strip() def _invoke_summary(self, model: Any | None, prompt: str, *, last: bool = False) -> str | None:
except Exception: """Invoke ``model`` for a summary; ``None`` on error or a blank response.
logger.exception("Summary generation failed; skipping compaction this turn")
Text extraction / non-empty validation runs *inside* the try: reading the
response's ``.text`` is part of consuming the provider result, so a failing
accessor must convert to a candidate failure (fall through) rather than escape
the fail-open boundary.
"""
if model is None:
return None return None
try:
response = model.invoke(prompt, config={"metadata": {"lc_source": "summarization"}})
return self._checked_summary(response, last)
except Exception:
self._log_summary_error(last)
return None
async def _ainvoke_summary(self, model: Any | None, prompt: str, *, last: bool = False) -> str | None:
"""Async counterpart of :meth:`_invoke_summary`."""
if model is None:
return None
try:
response = await model.ainvoke(prompt, config={"metadata": {"lc_source": "summarization"}})
return self._checked_summary(response, last)
except Exception:
self._log_summary_error(last)
return None
def _checked_summary(self, response: Any, last: bool) -> str | None:
summary = self._nonempty_summary(getattr(response, "text", None))
if summary is None:
self._log_summary_empty(last)
return summary
@staticmethod
def _log_summary_error(last: bool) -> None:
if last:
logger.exception("Summary generation failed; skipping compaction this turn")
else:
logger.warning("Summary generation failed; falling back to the run model", exc_info=True)
@staticmethod
def _log_summary_empty(last: bool) -> None:
if last:
logger.warning("Summary model returned empty text; skipping compaction this turn")
else:
logger.warning("Summary model returned empty text; falling back to the run model")
@staticmethod @staticmethod
def _summary_count_message(summary_text: str) -> HumanMessage: def _summary_count_message(summary_text: str) -> HumanMessage:
@ -303,15 +486,30 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
runtime: Runtime, runtime: Runtime,
*, *,
force: bool = False, force: bool = False,
raise_on_failure: bool = False,
) -> ContextCompactionResult | None: ) -> ContextCompactionResult | None:
"""Summarize old context and retain the active tail.
``force`` bypasses the automatic trigger threshold (a manual caller always
wants to compact). ``raise_on_failure`` is a *separate* concern: when set (the
manual ``/compact`` path), a generation failure raises ``SummaryGenerationError``
so it can be reported distinctly from "nothing to compact"; the automatic path
leaves it False and swallows the failure, retrying on a later triggered turn.
"""
prepared = self._prepare_compaction(state, force=force) prepared = self._prepare_compaction(state, force=force)
if prepared is None: if prepared is None:
return None return None
messages_to_summarize, preserved_messages, previous_summary, total_tokens = prepared messages_to_summarize, preserved_messages, previous_summary, total_tokens = prepared
self._fire_hooks(messages_to_summarize, preserved_messages, runtime)
summary = self._summarize_with(messages_to_summarize, previous_summary=previous_summary) summary = self._summarize_with(messages_to_summarize, previous_summary=previous_summary)
if summary is None: if summary is None:
if raise_on_failure:
raise SummaryGenerationError("summary generation failed")
return None return None
# Fire hooks only once a replacement summary exists — flushing pre-compaction
# messages into durable memory for a summary that never materializes would
# duplicate that work on the next attempt. Messages are still removed after
# this returns (in _maybe_summarize), so hooks run before they are gone.
self._fire_hooks(messages_to_summarize, preserved_messages, runtime)
return ContextCompactionResult( return ContextCompactionResult(
summary_text=summary, summary_text=summary,
messages_to_summarize=tuple(messages_to_summarize), messages_to_summarize=tuple(messages_to_summarize),
@ -325,15 +523,20 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
runtime: Runtime, runtime: Runtime,
*, *,
force: bool = False, force: bool = False,
raise_on_failure: bool = False,
) -> ContextCompactionResult | None: ) -> ContextCompactionResult | None:
"""Async counterpart of :meth:`compact_state` (see it for ``raise_on_failure``)."""
prepared = self._prepare_compaction(state, force=force) prepared = self._prepare_compaction(state, force=force)
if prepared is None: if prepared is None:
return None return None
messages_to_summarize, preserved_messages, previous_summary, total_tokens = prepared messages_to_summarize, preserved_messages, previous_summary, total_tokens = prepared
self._fire_hooks(messages_to_summarize, preserved_messages, runtime)
summary = await self._asummarize_with(messages_to_summarize, previous_summary=previous_summary) summary = await self._asummarize_with(messages_to_summarize, previous_summary=previous_summary)
if summary is None: if summary is None:
if raise_on_failure:
raise SummaryGenerationError("summary generation failed")
return None return None
# Fire hooks only once a replacement summary exists (see compact_state).
self._fire_hooks(messages_to_summarize, preserved_messages, runtime)
return ContextCompactionResult( return ContextCompactionResult(
summary_text=summary, summary_text=summary,
messages_to_summarize=tuple(messages_to_summarize), messages_to_summarize=tuple(messages_to_summarize),
@ -444,11 +647,37 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
logger.exception("before_summarization hook %s failed", hook_name) logger.exception("before_summarization hook %s failed", hook_name)
def _build_summary_anchor(candidate_names: list[str | None], app_config: Any) -> tuple[Any | None, str | None]:
"""Build the first constructible model among ``candidate_names`` (guarded).
The returned model is tagged for RunJournal attribution but *not* TAG_NOSTREAM (the
middleware wraps a nostream copy). It becomes the parent's token-counter / profile
anchor and is reused for generation when a candidate matches its name. A per-name
construction failure is swallowed and the next candidate tried, so a broken primary
constructor neither breaks agent construction nor skips the healthy run model; a
trailing ``None`` name asks ``create_chat_model`` for its own default. Returns
``(None, None)`` when nothing can be constructed.
"""
tried: set[str | None] = set()
for name in candidate_names:
if name in tried:
continue
tried.add(name)
try:
model = create_chat_model(name=name, thinking_enabled=False, app_config=app_config, attach_tracing=False)
except Exception:
logger.exception("Failed to build summary anchor model %r; trying the next candidate", name)
continue
return model.with_config(tags=["middleware:summarize"]), name
return None, None
def create_summarization_middleware( def create_summarization_middleware(
*, *,
app_config: Any | None = None, app_config: Any | None = None,
keep: tuple[str, int | float] | None = None, keep: tuple[str, int | float] | None = None,
skip_memory_flush: bool = False, skip_memory_flush: bool = False,
run_model_name: str | None = None,
) -> DeerFlowSummarizationMiddleware | None: ) -> DeerFlowSummarizationMiddleware | None:
"""Create the configured summarization middleware. """Create the configured summarization middleware.
@ -456,6 +685,13 @@ def create_summarization_middleware(
use this factory so model resolution, hooks, prompt config, and retention use this factory so model resolution, hooks, prompt config, and retention
defaults cannot drift. defaults cannot drift.
``run_model_name`` is the model the run actually executes with, resolved by the
caller (the lead / subagent / manual builders each already resolve it) and passed
in as the authoritative source of truth for ``model_name: null`` summarization and
the explicit-summary-model fallback. The middleware does not re-derive it from
``runtime.context`` / ``get_config()``, which do not carry a custom agent's or a
subagent's resolved model.
``skip_memory_flush`` omits the ``memory_flush_hook`` that otherwise ``skip_memory_flush`` omits the ``memory_flush_hook`` that otherwise
flushes pre-compaction messages into the durable memory queue. The lead flushes pre-compaction messages into the durable memory queue. The lead
chain keeps it (research should persist); the subagent chain sets it so a chain keeps it (research should persist); the subagent chain sets it so a
@ -477,23 +713,25 @@ def create_summarization_middleware(
else: else:
trigger = config.trigger.to_tuple() trigger = config.trigger.to_tuple()
if config.model_name: default_name = resolved_app_config.models[0].name if getattr(resolved_app_config, "models", None) else None
model = create_chat_model( # Build the anchor (token-counter / profile model, reused for generation) guarded,
name=config.model_name, # rather than eagerly building the configured/default model and letting a broken
thinking_enabled=False, # constructor escape. Candidates in order: the primary generation model (configured
app_config=resolved_app_config, # summary model, else the run's own model), then the run model, then the default,
attach_tracing=False, # then ``None`` (create_chat_model's default) as a last resort. So the null case
) # builds from ``run_model_name`` — not ``config.models[0]`` — and a broken primary
else: # falls through to the healthy run model instead of failing agent construction.
model = create_chat_model( primary_name = config.model_name or run_model_name or default_name
thinking_enabled=False, anchor_model, anchor_name = _build_summary_anchor(
app_config=resolved_app_config, [primary_name, run_model_name or default_name, default_name, None],
attach_tracing=False, resolved_app_config,
) )
model = model.with_config(tags=["middleware:summarize"]) if anchor_model is None:
logger.warning("Summarization is enabled but no summary model could be constructed; compaction is unavailable for this build")
return None
kwargs: dict[str, Any] = { kwargs: dict[str, Any] = {
"model": model, "model": anchor_model,
"trigger": trigger, "trigger": trigger,
"keep": keep or config.keep.to_tuple(), "keep": keep or config.keep.to_tuple(),
} }
@ -511,4 +749,8 @@ def create_summarization_middleware(
return DeerFlowSummarizationMiddleware( return DeerFlowSummarizationMiddleware(
**kwargs, **kwargs,
before_summarization=hooks, before_summarization=hooks,
app_config=resolved_app_config,
configured_model_name=config.model_name,
run_model_name=run_model_name,
anchor_model_name=anchor_name,
) )

View File

@ -475,6 +475,11 @@ def build_subagent_runtime_middlewares(
summarization_middleware = create_summarization_middleware( summarization_middleware = create_summarization_middleware(
app_config=app_config, app_config=app_config,
skip_memory_flush=True, skip_memory_flush=True,
# The subagent's resolved model is the source of truth for null-model
# summarization: the subagent context/configurable does not carry the child
# model (it inherits the parent's), so passing it directly is what makes a
# distinct-model subagent summarize with its own model, not the parent's.
run_model_name=model_name,
) )
if summarization_middleware is not None: if summarization_middleware is not None:
middlewares.append(summarization_middleware) middlewares.append(summarization_middleware)

View File

@ -28,7 +28,10 @@ class SummarizationConfig(BaseModel):
) )
model_name: str | None = Field( model_name: str | None = Field(
default=None, default=None,
description="Model name to use for summarization (None = use a lightweight model)", description="Model name to use for summarization. None = summarize with the model the run "
"actually executes with (the lead run's model, a subagent's own model, or a thread's "
"custom-agent model), not config.models[0]. When set, that model generates and the run's "
"own model is used as a fallback if the configured summary provider fails.",
) )
trigger: ContextSize | list[ContextSize] | None = Field( trigger: ContextSize | list[ContextSize] | None = Field(
default=None, default=None,

View File

@ -2,15 +2,19 @@
from __future__ import annotations from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass from dataclasses import dataclass
from types import SimpleNamespace from types import SimpleNamespace
from langgraph.types import Overwrite from langgraph.types import Overwrite
from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware, create_summarization_middleware from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware, SummaryGenerationError, create_summarization_middleware
from deerflow.config.app_config import AppConfig, get_app_config from deerflow.config.app_config import AppConfig, get_app_config
from deerflow.runtime.checkpoint_state import CheckpointStateAccessor from deerflow.runtime.checkpoint_state import CheckpointStateAccessor
logger = logging.getLogger(__name__)
class ContextCompactionDisabled(RuntimeError): class ContextCompactionDisabled(RuntimeError):
"""Raised when manual compaction is requested while summarization is disabled.""" """Raised when manual compaction is requested while summarization is disabled."""
@ -38,13 +42,60 @@ def _create_compaction_middleware(
*, *,
app_config: AppConfig, app_config: AppConfig,
keep: tuple[str, int | float] | None, keep: tuple[str, int | float] | None,
run_model_name: str | None = None,
) -> DeerFlowSummarizationMiddleware: ) -> DeerFlowSummarizationMiddleware:
middleware = create_summarization_middleware(app_config=app_config, keep=keep) middleware = create_summarization_middleware(app_config=app_config, keep=keep, run_model_name=run_model_name)
if middleware is None: if middleware is None:
raise ContextCompactionDisabled("Context compaction is disabled.") raise ContextCompactionDisabled("Context compaction is disabled.")
return middleware return middleware
def _safe_load_agent_config(agent_name: str, user_id: str | None):
"""Load a custom agent's config, returning ``None`` on any failure.
A missing / unparseable agent config must not fail compaction; the run model is a
best-effort optimization and the default is a safe fallback. The caller runs this
off the event loop via ``asyncio.to_thread``, so the strict blocking-IO detector
does not flag the filesystem read and the broad ``except`` here cannot mask a
``BlockingError`` raised on the loop.
"""
from deerflow.config.agents_config import load_agent_config
try:
return load_agent_config(agent_name, user_id=user_id)
except Exception:
logger.warning("Could not load agent config for %r; using the default model for summarization", agent_name, exc_info=True)
return None
async def _aresolve_thread_model_name(
model_name: str | None,
agent_name: str | None,
user_id: str | None,
app_config: AppConfig,
) -> str | None:
"""Resolve the model a thread should summarize with, mirroring lead resolution.
Precedence matches ``lead_agent._resolve_model_name``: an explicit request model
override (validated against configured models) wins, else the thread's custom-agent
configured model, else ``config.models[0]``. Manual ``/compact`` does not execute
the agent, so there is no live runtime carrying the selected model the caller
(route / client) supplies it as ``model_name`` the same way a normal run submits
``context.model_name``. The custom-agent config read happens only when no request
model was supplied, runs off the event loop, and passes the owning ``user_id`` so
the per-user agent directory resolves.
"""
default = app_config.models[0].name if getattr(app_config, "models", None) else None
candidate = model_name
if not candidate and agent_name:
agent_config = await asyncio.to_thread(_safe_load_agent_config, agent_name, user_id)
if agent_config and agent_config.model:
candidate = agent_config.model
if candidate and app_config.get_model_config(candidate):
return candidate
return default
async def compact_thread_context( async def compact_thread_context(
accessor: CheckpointStateAccessor, accessor: CheckpointStateAccessor,
thread_id: str, thread_id: str,
@ -53,11 +104,13 @@ async def compact_thread_context(
force: bool = True, force: bool = True,
user_id: str | None = None, user_id: str | None = None,
agent_name: str | None = None, agent_name: str | None = None,
model_name: str | None = None,
app_config: AppConfig | None = None, app_config: AppConfig | None = None,
) -> ThreadCompactionResult: ) -> ThreadCompactionResult:
"""Summarize old messages in a thread and write a compacted checkpoint.""" """Summarize old messages in a thread and write a compacted checkpoint."""
resolved_app_config = app_config or get_app_config() resolved_app_config = app_config or get_app_config()
middleware = _create_compaction_middleware(app_config=resolved_app_config, keep=keep) run_model_name = await _aresolve_thread_model_name(model_name, agent_name, user_id, resolved_app_config)
middleware = _create_compaction_middleware(app_config=resolved_app_config, keep=keep, run_model_name=run_model_name)
read_config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}} read_config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
snapshot = await accessor.aget(read_config) snapshot = await accessor.aget(read_config)
@ -80,7 +133,18 @@ async def compact_thread_context(
if agent_name: if agent_name:
runtime_context["agent_name"] = agent_name runtime_context["agent_name"] = agent_name
runtime = SimpleNamespace(context=runtime_context) runtime = SimpleNamespace(context=runtime_context)
result = await middleware.acompact_state(state, runtime, force=force) # type: ignore[arg-type] try:
# ``raise_on_failure`` is independent of ``force``: a manual caller always wants
# a generation failure surfaced (even a force=False call that met the threshold),
# so it must not collapse into the force=False "nothing to compact" branch below.
result = await middleware.acompact_state(state, runtime, force=force, raise_on_failure=True) # type: ignore[arg-type]
except SummaryGenerationError as exc:
# A compressible thread whose summary LLM failed (after the run-model fallback)
# is a real failure, distinct from "nothing to compact". Route it to the
# already-consumed ContextCompactionFailed path (HTTP 500 -> frontend error
# toast) instead of a compacted=False result that reads as "does not need
# compaction".
raise ContextCompactionFailed("summary generation failed") from exc
if result is None: if result is None:
return ThreadCompactionResult(thread_id=thread_id, compacted=False, reason="not_enough_messages") return ThreadCompactionResult(thread_id=thread_id, compacted=False, reason="not_enough_messages")

View File

@ -10,9 +10,10 @@ from langgraph.graph.message import add_messages
from langgraph.types import Overwrite from langgraph.types import Overwrite
from app.gateway import services as gateway_services from app.gateway import services as gateway_services
from deerflow.agents.middlewares.summarization_middleware import SummaryGenerationError
from deerflow.runtime import context_compaction from deerflow.runtime import context_compaction
from deerflow.runtime.checkpoint_state import CheckpointStateAccessor from deerflow.runtime.checkpoint_state import CheckpointStateAccessor
from deerflow.runtime.context_compaction import compact_thread_context from deerflow.runtime.context_compaction import ContextCompactionFailed, compact_thread_context
class _FakeAccessor: class _FakeAccessor:
@ -56,7 +57,7 @@ class _FakeCompactionMiddleware:
return None return None
return (state["messages"][:-1], state["messages"][-1:], state.get("summary_text"), 123) return (state["messages"][:-1], state["messages"][-1:], state.get("summary_text"), 123)
async def acompact_state(self, state, runtime, *, force=False): async def acompact_state(self, state, runtime, *, force=False, raise_on_failure=False):
self.runtime_contexts.append(dict(runtime.context)) self.runtime_contexts.append(dict(runtime.context))
prepared = self._prepare_compaction(state, force=force) prepared = self._prepare_compaction(state, force=force)
if prepared is None: if prepared is None:
@ -90,6 +91,9 @@ async def test_compact_thread_context_reads_materialized_state_and_overwrites_me
"_create_compaction_middleware", "_create_compaction_middleware",
lambda **_kwargs: middleware, lambda **_kwargs: middleware,
) )
import deerflow.config.agents_config as agents_config
monkeypatch.setattr(agents_config, "load_agent_config", lambda name, **_kw: None)
result = await compact_thread_context( result = await compact_thread_context(
accessor, accessor,
@ -255,3 +259,180 @@ async def test_compact_thread_context_returns_noop_without_writing(monkeypatch):
assert result.reason == "not_enough_messages" assert result.reason == "not_enough_messages"
assert accessor.update_args is None assert accessor.update_args is None
assert middleware.prepare_calls == 1 assert middleware.prepare_calls == 1
@pytest.mark.asyncio
async def test_compact_thread_context_raises_on_summary_generation_failure(monkeypatch):
"""A compressible thread whose summary LLM fails (after the run-model fallback) is
a real failure: it must surface via the already-consumed ``ContextCompactionFailed``
path (HTTP 500 -> frontend error toast), not a ``compacted=False`` result that the
frontend renders as the benign "does not need compaction" toast."""
accessor = _FakeAccessor(
{
"messages": [
HumanMessage(content="old question"),
AIMessage(content="old answer"),
HumanMessage(content="latest question"),
],
}
)
captured: dict[str, object] = {}
class _FailingMiddleware:
async def acompact_state(self, state, runtime, *, force=False, raise_on_failure=False):
# The manual path must opt into raise_on_failure regardless of force.
captured["raise_on_failure"] = raise_on_failure
raise SummaryGenerationError("summary generation failed")
monkeypatch.setattr(
context_compaction,
"_create_compaction_middleware",
lambda **_kwargs: _FailingMiddleware(),
)
with pytest.raises(ContextCompactionFailed):
await compact_thread_context(accessor, "thread-1", app_config=SimpleNamespace())
assert captured["raise_on_failure"] is True
assert accessor.update_args is None
def _model_app_config(*names):
models = [SimpleNamespace(name=n) for n in names]
return SimpleNamespace(
models=models,
get_model_config=lambda n: next((m for m in models if m.name == n), None),
)
@pytest.mark.asyncio
async def test_resolve_request_model_overrides_agent_and_default(monkeypatch):
"""The selected request model wins over both the custom-agent model and the default,
mirroring lead resolution (request -> agent -> default). A supplied request model
also short-circuits the agent-config filesystem read entirely."""
import deerflow.config.agents_config as agents_config
def _fail(name, **_kw):
raise AssertionError("agent config must not be loaded when a request model is supplied")
monkeypatch.setattr(agents_config, "load_agent_config", _fail)
app_config = _model_app_config("default-model", "agent-model", "selected-model")
got = await context_compaction._aresolve_thread_model_name("selected-model", "research-agent", "user-1", app_config)
assert got == "selected-model"
@pytest.mark.asyncio
async def test_resolve_invalid_request_model_falls_to_default_not_agent(monkeypatch):
"""A request model that is not a configured model falls straight to the default —
the agent model is only consulted when no request model was supplied, matching
``lead_agent._resolve_model_name(requested or agent)``."""
import deerflow.config.agents_config as agents_config
monkeypatch.setattr(agents_config, "load_agent_config", lambda name, **_kw: SimpleNamespace(model="agent-model"))
app_config = _model_app_config("default-model", "agent-model")
got = await context_compaction._aresolve_thread_model_name("ghost-model", "research-agent", "user-1", app_config)
assert got == "default-model"
@pytest.mark.asyncio
async def test_resolve_agent_model_when_no_request_model(monkeypatch):
"""With no request model, a custom-agent thread summarizes with its own model, and
the agent config is loaded with the owning ``user_id`` (per-user agent directory)."""
import deerflow.config.agents_config as agents_config
seen: dict[str, object] = {}
def _load(name, *, user_id=None):
seen["name"] = name
seen["user_id"] = user_id
return SimpleNamespace(model="agent-model")
monkeypatch.setattr(agents_config, "load_agent_config", _load)
app_config = _model_app_config("default-model", "agent-model")
got = await context_compaction._aresolve_thread_model_name(None, "research-agent", "user-1", app_config)
assert got == "agent-model"
assert seen == {"name": "research-agent", "user_id": "user-1"}
@pytest.mark.asyncio
async def test_resolve_falls_back_to_default(monkeypatch):
"""No agent, an unloadable agent config, or an agent whose model is not configured all
resolve to the default model rather than failing compaction."""
import deerflow.config.agents_config as agents_config
app_config = _model_app_config("default-model")
# No request model and no agent → default.
assert await context_compaction._aresolve_thread_model_name(None, None, "user-1", app_config) == "default-model"
# Agent config cannot be loaded (missing/unparseable) → default, not a crash.
def _raise(name, **_kw):
raise FileNotFoundError("agent directory not found")
monkeypatch.setattr(agents_config, "load_agent_config", _raise)
assert await context_compaction._aresolve_thread_model_name(None, "ghost-agent", "user-1", app_config) == "default-model"
# Agent's model is not a configured model → default.
monkeypatch.setattr(agents_config, "load_agent_config", lambda name, **_kw: SimpleNamespace(model="unconfigured"))
assert await context_compaction._aresolve_thread_model_name(None, "x-agent", "user-1", app_config) == "default-model"
@pytest.mark.asyncio
async def test_resolve_agent_config_load_runs_off_the_event_loop(monkeypatch):
"""The blocking agent-config read must run off the event loop (``asyncio.to_thread``),
not synchronously before the first await where the strict blocking-IO gate would flag
it and the broad except would then mask the ``BlockingError`` as a default fallback."""
import threading
import deerflow.config.agents_config as agents_config
main_thread = threading.get_ident()
seen: dict[str, object] = {}
def _load(name, *, user_id=None):
seen["thread"] = threading.get_ident()
return SimpleNamespace(model="agent-model")
monkeypatch.setattr(agents_config, "load_agent_config", _load)
app_config = _model_app_config("default-model", "agent-model")
got = await context_compaction._aresolve_thread_model_name(None, "research-agent", "user-1", app_config)
assert got == "agent-model"
assert seen["thread"] != main_thread
@pytest.mark.asyncio
async def test_compact_thread_context_threads_selected_model_to_factory(monkeypatch):
"""Route/client path: the model selected for the thread request reaches the
summarization factory as ``run_model_name`` covering the real call, not only the
resolver in isolation (request override -> agent -> default)."""
import deerflow.config.agents_config as agents_config
monkeypatch.setattr(agents_config, "load_agent_config", lambda name, **_kw: SimpleNamespace(model="agent-model"))
app_config = _model_app_config("default-model", "agent-model", "selected-model")
captured: dict[str, object] = {}
def _capture(**kwargs):
captured["run_model_name"] = kwargs.get("run_model_name")
return _FakeCompactionMiddleware()
monkeypatch.setattr(context_compaction, "_create_compaction_middleware", _capture)
messages = [HumanMessage(content="old"), AIMessage(content="a"), HumanMessage(content="new")]
# 1. Explicit request model wins.
await compact_thread_context(_FakeAccessor({"messages": messages}), "thread-1", app_config=app_config, user_id="user-1", agent_name="research-agent", model_name="selected-model")
assert captured["run_model_name"] == "selected-model"
# 2. No request model → the custom agent's model.
await compact_thread_context(_FakeAccessor({"messages": messages}), "thread-1", app_config=app_config, user_id="user-1", agent_name="research-agent")
assert captured["run_model_name"] == "agent-model"
# 3. No request model, no agent → default.
await compact_thread_context(_FakeAccessor({"messages": messages}), "thread-1", app_config=app_config, user_id="user-1")
assert captured["run_model_name"] == "default-model"

View File

@ -503,7 +503,7 @@ class TestMiddlewareRegistration:
summary_sentinel = object() summary_sentinel = object()
monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: []) monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: summary_sentinel) monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda **_kwargs: summary_sentinel)
monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None) monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
middlewares = lead_agent_module.build_middlewares( middlewares = lead_agent_module.build_middlewares(

View File

@ -609,6 +609,38 @@ def test_build_middlewares_passes_explicit_app_config_to_shared_factory(monkeypa
assert middlewares[0] == "base-middleware" assert middlewares[0] == "base-middleware"
def test_build_middlewares_passes_run_model_name_to_summarization(monkeypatch):
"""The resolved run model (e.g. a custom agent's model, distinct from models[0])
is threaded into the summarization factory, so null-model compaction summarizes
with it rather than config.models[0]. Production-shaped: the model reaches the
factory as an argument, not via runtime.context."""
app_config = _make_app_config(
[
_make_model("default-model", supports_thinking=False),
_make_model("custom-agent-model", supports_thinking=False),
]
)
captured: dict[str, object] = {}
monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init: ["base-middleware"])
monkeypatch.setattr(
lead_agent_module,
"_create_summarization_middleware",
lambda **kwargs: captured.update(kwargs) or None,
)
monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
monkeypatch.setattr(lead_agent_module, "TitleMiddleware", lambda *, app_config: "title-middleware")
monkeypatch.setattr(lead_agent_module, "MemoryMiddleware", lambda agent_name=None, *, memory_config: "memory-middleware")
lead_agent_module.build_middlewares(
{"configurable": {"is_plan_mode": False, "subagent_enabled": False}},
model_name="custom-agent-model",
app_config=app_config,
)
assert captured["run_model_name"] == "custom-agent-model"
def test_build_middlewares_orders_skill_activation_before_policy_and_durable_context(monkeypatch): def test_build_middlewares_orders_skill_activation_before_policy_and_durable_context(monkeypatch):
from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware
from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware
@ -616,7 +648,7 @@ def test_build_middlewares_orders_skill_activation_before_policy_and_durable_con
app_config = _make_app_config([_make_model("safe-model", supports_thinking=False)]) app_config = _make_app_config([_make_model("safe-model", supports_thinking=False)])
monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: []) monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None) monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda **_kwargs: None)
monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None) monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
middlewares = lead_agent_module.build_middlewares( middlewares = lead_agent_module.build_middlewares(
@ -644,7 +676,7 @@ def test_compiled_skill_policy_chain_filters_schema_and_blocks_execution(monkeyp
loop_detection=LoopDetectionConfig(enabled=False), loop_detection=LoopDetectionConfig(enabled=False),
) )
monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: []) monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None) monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda **_kwargs: None)
monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None) monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
middlewares = lead_agent_module.build_middlewares( middlewares = lead_agent_module.build_middlewares(
@ -715,7 +747,7 @@ def test_build_middlewares_places_mcp_routing_before_deferred_filter(monkeypatch
monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config) monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config)
monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: []) monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None) monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda **_kwargs: None)
monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None) monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
middlewares = lead_agent_module.build_middlewares( middlewares = lead_agent_module.build_middlewares(
@ -746,7 +778,7 @@ def test_build_middlewares_uses_loop_detection_config(monkeypatch):
monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config) monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config)
monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: []) monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None) monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda **_kwargs: None)
monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None) monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
middlewares = lead_agent_module.build_middlewares( middlewares = lead_agent_module.build_middlewares(
@ -772,7 +804,7 @@ def test_build_middlewares_omits_loop_detection_when_disabled(monkeypatch):
monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config) monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config)
monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: []) monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None) monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda **_kwargs: None)
monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None) monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
middlewares = lead_agent_module.build_middlewares( middlewares = lead_agent_module.build_middlewares(
@ -799,7 +831,7 @@ def test_build_middlewares_injects_configured_extension_middlewares(monkeypatch)
monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config) monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config)
monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: []) monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None) monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda **_kwargs: None)
monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None) monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
middlewares = lead_agent_module.build_middlewares( middlewares = lead_agent_module.build_middlewares(
@ -829,7 +861,7 @@ def test_build_middlewares_passes_subagent_total_limit_from_app_config(monkeypat
monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config) monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config)
monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: []) monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None) monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda **_kwargs: None)
monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None) monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
middlewares = lead_agent_module.build_middlewares( middlewares = lead_agent_module.build_middlewares(
@ -852,7 +884,7 @@ def test_build_middlewares_allows_runtime_subagent_total_limit_override(monkeypa
monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config) monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config)
monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: []) monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None) monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda **_kwargs: None)
monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None) monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
middlewares = lead_agent_module.build_middlewares( middlewares = lead_agent_module.build_middlewares(
@ -881,7 +913,7 @@ def test_build_middlewares_rejects_invalid_configured_extension_middleware(monke
monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config) monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config)
monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: []) monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None) monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda **_kwargs: None)
monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None) monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
with pytest.raises(ValueError, match="not an instance of type"): with pytest.raises(ValueError, match="not an instance of type"):
@ -901,7 +933,7 @@ def test_build_middlewares_rejects_configured_extension_class_with_wrong_base(mo
monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config) monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config)
monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: []) monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None) monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda **_kwargs: None)
monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None) monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
with pytest.raises(ValueError, match="is not a subclass of AgentMiddleware"): with pytest.raises(ValueError, match="is not a subclass of AgentMiddleware"):
@ -921,7 +953,7 @@ def test_build_middlewares_reraises_configured_extension_instantiation_failure(m
monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config) monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config)
monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: []) monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None) monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda **_kwargs: None)
monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None) monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
with pytest.raises(RuntimeError, match="configured middleware init failed"): with pytest.raises(RuntimeError, match="configured middleware init failed"):
@ -941,7 +973,7 @@ def test_build_middlewares_rejects_missing_configured_extension_module(monkeypat
monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config) monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config)
monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: []) monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None) monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda **_kwargs: None)
monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None) monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
with pytest.raises(ImportError, match="Could not import module definitely_missing_pkg.middlewares_typo"): with pytest.raises(ImportError, match="Could not import module definitely_missing_pkg.middlewares_typo"):
@ -986,7 +1018,10 @@ def test_create_summarization_middleware_uses_configured_model_alias(monkeypatch
fake_model.with_config.assert_called_once_with(tags=["middleware:summarize"]) fake_model.with_config.assert_called_once_with(tags=["middleware:summarize"])
def test_create_summarization_middleware_omits_model_name_when_unconfigured(monkeypatch): def test_create_summarization_middleware_uses_default_when_unconfigured(monkeypatch):
"""Null summary config with no supplied run model builds the anchor from the default
model now with an explicit name rather than relying on create_chat_model's internal
default, so the factory has no implicit models[0] dependency. Same resulting model."""
app_config = _make_app_config([_make_model("default-model", supports_thinking=False)]) app_config = _make_app_config([_make_model("default-model", supports_thinking=False)])
app_config.summarization = SummarizationConfig(enabled=True, model_name=None) app_config.summarization = SummarizationConfig(enabled=True, model_name=None)
app_config.memory = MemoryConfig(enabled=False) app_config.memory = MemoryConfig(enabled=False)
@ -1004,10 +1039,35 @@ def test_create_summarization_middleware_omits_model_name_when_unconfigured(monk
middleware = lead_agent_module._create_summarization_middleware(app_config=app_config) middleware = lead_agent_module._create_summarization_middleware(app_config=app_config)
assert "name" not in captured assert captured["name"] == "default-model"
assert captured["thinking_enabled"] is False assert captured["thinking_enabled"] is False
assert captured["app_config"] is app_config assert captured["app_config"] is app_config
assert middleware["model"] is fake_model assert middleware["model"] is fake_model
assert middleware["anchor_model_name"] == "default-model"
def test_create_summarization_middleware_threads_run_model_name(monkeypatch):
"""A custom agent's resolved model reaches the middleware as run_model_name (the
model-ownership source of truth), while configured_model_name stays None for a
null summarization config."""
app_config = _make_app_config(
[
_make_model("default-model", supports_thinking=False),
_make_model("custom-agent-model", supports_thinking=False),
]
)
app_config.summarization = SummarizationConfig(enabled=True, model_name=None)
app_config.memory = MemoryConfig(enabled=False)
fake_model = MagicMock()
fake_model.with_config.return_value = fake_model
monkeypatch.setattr(summarization_middleware_module, "create_chat_model", lambda **kwargs: fake_model)
monkeypatch.setattr(summarization_middleware_module, "DeerFlowSummarizationMiddleware", lambda **kwargs: kwargs)
middleware = lead_agent_module._create_summarization_middleware(app_config=app_config, run_model_name="custom-agent-model")
assert middleware["run_model_name"] == "custom-agent-model"
assert middleware["configured_model_name"] is None
def test_create_summarization_middleware_uses_frontend_supported_update_key(monkeypatch): def test_create_summarization_middleware_uses_frontend_supported_update_key(monkeypatch):

View File

@ -2,7 +2,7 @@ from __future__ import annotations
from types import SimpleNamespace from types import SimpleNamespace
from unittest import mock from unittest import mock
from unittest.mock import MagicMock from unittest.mock import AsyncMock, MagicMock
import pytest import pytest
from langchain.agents import create_agent from langchain.agents import create_agent
@ -13,7 +13,7 @@ from langgraph.constants import TAG_NOSTREAM
from deerflow.agents.memory.summarization_hook import memory_flush_hook from deerflow.agents.memory.summarization_hook import memory_flush_hook
from deerflow.agents.middlewares.dynamic_context_middleware import _DYNAMIC_CONTEXT_REMINDER_KEY, DynamicContextMiddleware, is_dynamic_context_reminder from deerflow.agents.middlewares.dynamic_context_middleware import _DYNAMIC_CONTEXT_REMINDER_KEY, DynamicContextMiddleware, is_dynamic_context_reminder
from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware, SummarizationEvent, create_summarization_middleware from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware, SummarizationEvent, SummaryGenerationError, create_summarization_middleware
from deerflow.agents.thread_state import ThreadState from deerflow.agents.thread_state import ThreadState
from deerflow.config.memory_config import MemoryConfig from deerflow.config.memory_config import MemoryConfig
from deerflow.config.summarization_config import SummarizationConfig from deerflow.config.summarization_config import SummarizationConfig
@ -78,6 +78,7 @@ def _middleware(
) -> DeerFlowSummarizationMiddleware: ) -> DeerFlowSummarizationMiddleware:
model = MagicMock() model = MagicMock()
model.invoke.return_value = SimpleNamespace(text="compressed summary") model.invoke.return_value = SimpleNamespace(text="compressed summary")
model.ainvoke = AsyncMock(return_value=SimpleNamespace(text="compressed summary"))
model.with_config.return_value = model model.with_config.return_value = model
return DeerFlowSummarizationMiddleware( return DeerFlowSummarizationMiddleware(
model=model, model=model,
@ -727,3 +728,481 @@ def test_benign_summary_input_text_preserved() -> None:
assert out is not None assert out is not None
assert "User: what is the plan" in out assert "User: what is the plan" in out
assert "prior recap text" in out assert "prior recap text" in out
def _app_config(model_names=("default-model",)):
"""Minimal AppConfig-shaped stub: ordered ``models`` + ``get_model_config``."""
models = [SimpleNamespace(name=name) for name in model_names]
def get_model_config(name):
return next((model for model in models if model.name == name), None)
return SimpleNamespace(models=models, get_model_config=get_model_config)
def _tracking_create_chat_model(built: list, *, fail: bool = False):
"""Patch target for ``create_chat_model``: records built names, returns a mock
whose summary text is ``from-<name>`` (or fails on invoke when ``fail``)."""
def _factory(*, name=None, thinking_enabled=False, app_config=None, attach_tracing=True, **kwargs):
model = MagicMock()
model.with_config.return_value = model
if fail:
model.invoke.side_effect = RuntimeError("provider down")
model.ainvoke = AsyncMock(side_effect=RuntimeError("provider down"))
else:
model.invoke.return_value = SimpleNamespace(text=f"from-{name}")
model.ainvoke = AsyncMock(return_value=SimpleNamespace(text=f"from-{name}"))
built.append(name)
return model
return _factory
def _blank_model(*, text: str = " \n\t ") -> MagicMock:
"""A model whose response body is whitespace-only (a generation failure)."""
model = MagicMock()
model.with_config.return_value = model
model.invoke.return_value = SimpleNamespace(text=text)
model.ainvoke = AsyncMock(return_value=SimpleNamespace(text=text))
return model
def test_null_model_summarizes_with_the_run_model(monkeypatch: pytest.MonkeyPatch) -> None:
"""model_name: null must summarize with the model the run actually uses, not
config.models[0]. This is the model-ownership fix: a run on a non-default model
while models[0]'s provider is broken must still compact.
Production-shaped: the run model is supplied at build time (``run_model_name``),
and the runtime carries NO ``model_name`` in its context the previous fixture
injected ``runtime.context['model_name']``, which the production custom-agent /
subagent contexts never populate."""
built: list = []
monkeypatch.setattr("deerflow.agents.middlewares.summarization_middleware.create_chat_model", _tracking_create_chat_model(built))
default_model = MagicMock()
default_model.with_config.return_value = default_model
default_model.invoke.return_value = SimpleNamespace(text="from-default")
middleware = DeerFlowSummarizationMiddleware(
model=default_model,
trigger=("messages", 4),
keep=("messages", 2),
token_counter=len,
app_config=_app_config(("default-model", "run-model")),
configured_model_name=None,
run_model_name="run-model",
)
result = middleware.before_model({"messages": _messages()}, _runtime())
# The run model was built + used; the default (models[0]) never generated a summary.
assert built == ["run-model"]
default_model.invoke.assert_not_called()
assert result is not None
assert result["summary_text"] == "from-run-model"
def test_explicit_summary_model_failure_falls_back_to_run_model(monkeypatch: pytest.MonkeyPatch) -> None:
"""An explicitly configured summary model generates; if its provider is broken,
compaction falls back to the run's own (working) model instead of no-op'ing.
The fallback is built lazily only after the primary fails."""
built: list = []
monkeypatch.setattr("deerflow.agents.middlewares.summarization_middleware.create_chat_model", _tracking_create_chat_model(built))
explicit = MagicMock()
explicit.with_config.return_value = explicit
explicit.invoke.side_effect = RuntimeError("summary provider down")
middleware = DeerFlowSummarizationMiddleware(
model=explicit,
trigger=("messages", 4),
keep=("messages", 2),
token_counter=len,
app_config=_app_config(("default-model", "run-model")),
configured_model_name="summary-model",
run_model_name="run-model",
)
result = middleware.before_model({"messages": _messages()}, _runtime())
explicit.invoke.assert_called_once() # primary tried
assert built == ["run-model"] # fallback built (lazily, after primary failed) + used
assert result is not None
assert result["summary_text"] == "from-run-model"
@pytest.mark.anyio
async def test_async_explicit_failure_falls_back_to_run_model(monkeypatch: pytest.MonkeyPatch) -> None:
"""The async path applies the same run-model fallback as the sync path."""
built: list = []
monkeypatch.setattr("deerflow.agents.middlewares.summarization_middleware.create_chat_model", _tracking_create_chat_model(built))
explicit = MagicMock()
explicit.with_config.return_value = explicit
explicit.ainvoke = AsyncMock(side_effect=RuntimeError("summary provider down"))
middleware = DeerFlowSummarizationMiddleware(
model=explicit,
trigger=("messages", 4),
keep=("messages", 2),
token_counter=len,
app_config=_app_config(("default-model", "run-model")),
configured_model_name="summary-model",
run_model_name="run-model",
)
result = await middleware.abefore_model({"messages": _messages()}, _runtime())
explicit.ainvoke.assert_awaited_once()
assert built == ["run-model"]
assert result is not None
assert result["summary_text"] == "from-run-model"
def test_both_summary_models_failing_returns_none_on_automatic_path(monkeypatch: pytest.MonkeyPatch) -> None:
"""When the explicit model and the run-model fallback both fail, the automatic
path leaves compaction state unchanged (returns None) rather than raising."""
built: list = []
monkeypatch.setattr("deerflow.agents.middlewares.summarization_middleware.create_chat_model", _tracking_create_chat_model(built, fail=True))
explicit = MagicMock()
explicit.with_config.return_value = explicit
explicit.invoke.side_effect = RuntimeError("provider down")
middleware = DeerFlowSummarizationMiddleware(
model=explicit,
trigger=("messages", 4),
keep=("messages", 2),
token_counter=len,
app_config=_app_config(("default-model", "run-model")),
configured_model_name="summary-model",
run_model_name="run-model",
)
result = middleware.before_model({"messages": _messages()}, _runtime())
assert result is None
explicit.invoke.assert_called_once() # primary tried
assert built == ["run-model"] # fallback tried too
def test_explicit_summary_model_equal_to_run_model_is_not_retried(monkeypatch: pytest.MonkeyPatch) -> None:
"""When the configured summary model IS the run model, there is no distinct
fallback: the failed model must not be re-invoked (that would just burn another
call against a provider we already know is down) and no second model is built."""
built: list = []
monkeypatch.setattr("deerflow.agents.middlewares.summarization_middleware.create_chat_model", _tracking_create_chat_model(built, fail=True))
explicit = MagicMock()
explicit.with_config.return_value = explicit
explicit.invoke.side_effect = RuntimeError("provider down")
middleware = DeerFlowSummarizationMiddleware(
model=explicit,
trigger=("messages", 4),
keep=("messages", 2),
token_counter=len,
app_config=_app_config(("summary-model",)),
configured_model_name="summary-model",
run_model_name="summary-model", # run model == configured summary model
)
result = middleware.before_model({"messages": _messages()}, _runtime())
assert result is None
explicit.invoke.assert_called_once() # tried exactly once, not twice
assert built == [] # no distinct fallback model was built
def test_fallback_construction_error_does_not_escape_automatic_path(monkeypatch: pytest.MonkeyPatch) -> None:
"""A broken run-model config must not skip the healthy primary or escape the
automatic failure boundary: the primary is tried first, and a fallback that
fails to *construct* is swallowed (returns None), not raised."""
def _failing_build(*, name=None, **kwargs):
raise RuntimeError("cannot build run model")
monkeypatch.setattr("deerflow.agents.middlewares.summarization_middleware.create_chat_model", _failing_build)
explicit = MagicMock()
explicit.with_config.return_value = explicit
explicit.invoke.side_effect = RuntimeError("summary provider down")
middleware = DeerFlowSummarizationMiddleware(
model=explicit,
trigger=("messages", 4),
keep=("messages", 2),
token_counter=len,
app_config=_app_config(("default-model", "run-model")),
configured_model_name="summary-model",
run_model_name="run-model",
)
result = middleware.before_model({"messages": _messages()}, _runtime())
assert result is None
explicit.invoke.assert_called_once() # healthy-or-not, the primary was still tried first
def test_blank_summary_response_is_not_committed_null_case(monkeypatch: pytest.MonkeyPatch) -> None:
"""A whitespace-only model response is a generation failure, not a valid empty
summary: the automatic path returns None (history preserved, no RemoveMessage)
instead of removing all history for an empty replacement."""
run_model = _blank_model()
monkeypatch.setattr("deerflow.agents.middlewares.summarization_middleware.create_chat_model", lambda **kwargs: run_model)
default_model = MagicMock()
default_model.with_config.return_value = default_model
middleware = DeerFlowSummarizationMiddleware(
model=default_model,
trigger=("messages", 4),
keep=("messages", 2),
token_counter=len,
app_config=_app_config(("default-model", "run-model")),
configured_model_name=None,
run_model_name="run-model",
)
result = middleware.before_model({"messages": _messages()}, _runtime())
assert result is None
run_model.invoke.assert_called_once()
@pytest.mark.anyio
async def test_blank_summary_response_is_not_committed_async(monkeypatch: pytest.MonkeyPatch) -> None:
"""Async counterpart: a whitespace-only response leaves compaction state unchanged."""
run_model = _blank_model()
monkeypatch.setattr("deerflow.agents.middlewares.summarization_middleware.create_chat_model", lambda **kwargs: run_model)
default_model = MagicMock()
default_model.with_config.return_value = default_model
middleware = DeerFlowSummarizationMiddleware(
model=default_model,
trigger=("messages", 4),
keep=("messages", 2),
token_counter=len,
app_config=_app_config(("default-model", "run-model")),
configured_model_name=None,
run_model_name="run-model",
)
result = await middleware.abefore_model({"messages": _messages()}, _runtime())
assert result is None
run_model.ainvoke.assert_awaited_once()
def test_blank_primary_summary_falls_back_to_run_model(monkeypatch: pytest.MonkeyPatch) -> None:
"""A blank primary response is treated as failure and triggers the run-model
fallback, exactly as a raised exception would."""
built: list = []
monkeypatch.setattr("deerflow.agents.middlewares.summarization_middleware.create_chat_model", _tracking_create_chat_model(built))
explicit = _blank_model() # primary returns whitespace
middleware = DeerFlowSummarizationMiddleware(
model=explicit,
trigger=("messages", 4),
keep=("messages", 2),
token_counter=len,
app_config=_app_config(("default-model", "run-model")),
configured_model_name="summary-model",
run_model_name="run-model",
)
result = middleware.before_model({"messages": _messages()}, _runtime())
explicit.invoke.assert_called_once() # blank primary counted as a failure
assert built == ["run-model"] # fallback built + used
assert result is not None
assert result["summary_text"] == "from-run-model"
def test_manual_compaction_failure_raises_summary_generation_error(monkeypatch: pytest.MonkeyPatch) -> None:
"""A manual /compact opts into ``raise_on_failure`` so a generation failure raises,
letting the caller report it distinctly from "nothing to compact". This is now
decoupled from ``force`` (which only bypasses the trigger threshold)."""
default_model = MagicMock()
default_model.with_config.return_value = default_model
default_model.invoke.side_effect = RuntimeError("provider down")
middleware = DeerFlowSummarizationMiddleware(
model=default_model,
trigger=("messages", 4),
keep=("messages", 2),
token_counter=len,
app_config=_app_config(("default-model",)),
configured_model_name=None,
run_model_name="default-model",
)
with pytest.raises(SummaryGenerationError):
middleware.compact_state({"messages": _messages()}, _runtime(), force=True, raise_on_failure=True)
def test_force_alone_does_not_raise_on_failure(monkeypatch: pytest.MonkeyPatch) -> None:
"""``force`` bypasses the threshold but must NOT, on its own, raise on a generation
failure only ``raise_on_failure`` does. The automatic path (force + no
raise_on_failure) leaves state unchanged."""
default_model = MagicMock()
default_model.with_config.return_value = default_model
default_model.invoke.side_effect = RuntimeError("provider down")
middleware = DeerFlowSummarizationMiddleware(
model=default_model,
trigger=("messages", 4),
keep=("messages", 2),
token_counter=len,
app_config=_app_config(("default-model",)),
configured_model_name=None,
run_model_name="default-model",
)
assert middleware.compact_state({"messages": _messages()}, _runtime(), force=True) is None
def test_before_summarization_hook_not_fired_when_summary_fails(monkeypatch: pytest.MonkeyPatch) -> None:
"""Hooks must not enqueue durable-memory work for a summary that never
materializes; on failure the automatic path returns None without firing hooks."""
default_model = MagicMock()
default_model.with_config.return_value = default_model
default_model.invoke.side_effect = RuntimeError("provider down")
captured: list[SummarizationEvent] = []
middleware = DeerFlowSummarizationMiddleware(
model=default_model,
trigger=("messages", 4),
keep=("messages", 2),
token_counter=len,
app_config=_app_config(("default-model",)),
configured_model_name=None,
run_model_name="default-model",
before_summarization=[captured.append],
)
# The run uses the default model, which fails; no distinct fallback in the null case.
assert middleware.before_model({"messages": _messages()}, _runtime()) is None
assert captured == []
def _factory_app_config(model_names, *, summary_model_name=None):
"""AppConfig-shaped stub for the factory: summarization enabled + ordered models."""
models = [SimpleNamespace(name=name) for name in model_names]
return SimpleNamespace(
summarization=SummarizationConfig(enabled=True, model_name=summary_model_name),
memory=MemoryConfig(enabled=False),
models=models,
get_model_config=lambda name: next((model for model in models if model.name == name), None),
)
def test_factory_null_case_anchor_is_run_model_not_models0(monkeypatch):
"""model_name: null builds the summary model from ``run_model_name``, never
config.models[0]. A run on a non-default model whose models[0] provider is broken
still gets a working summarization middleware the factory has no eager models[0]
dependency."""
built: list = []
monkeypatch.setattr("deerflow.agents.middlewares.summarization_middleware.create_chat_model", _tracking_create_chat_model(built))
middleware = create_summarization_middleware(
app_config=_factory_app_config(("models0", "run-model")),
keep=("messages", 2),
run_model_name="run-model",
)
assert middleware is not None
assert built == ["run-model"] # anchor built from the run model, not models0 / None
result = middleware.compact_state({"messages": _messages()}, _runtime(), force=True)
assert result is not None
assert result.summary_text == "from-run-model"
def test_factory_configured_constructor_failure_falls_back_to_run_model(monkeypatch):
"""A configured summary model whose *constructor* raises must not break middleware
creation or skip the healthy run model. The anchor falls through the broken
constructor to the run model, and generation summarizes with it (the reviewer's
``configured='broken-summary'`` reproduction)."""
built: list = []
def _factory(*, name=None, thinking_enabled=False, app_config=None, attach_tracing=True, **kwargs):
if name == "broken-summary":
raise RuntimeError("cannot construct summary provider")
model = MagicMock()
model.with_config.return_value = model
model.invoke.return_value = SimpleNamespace(text=f"from-{name}")
model.ainvoke = AsyncMock(return_value=SimpleNamespace(text=f"from-{name}"))
built.append(name)
return model
monkeypatch.setattr("deerflow.agents.middlewares.summarization_middleware.create_chat_model", _factory)
middleware = create_summarization_middleware(
app_config=_factory_app_config(("models0", "run-model"), summary_model_name="broken-summary"),
keep=("messages", 2),
run_model_name="run-model",
)
assert middleware is not None # a broken configured constructor did not break creation
assert "run-model" in built # the healthy run model was constructed as the anchor
result = middleware.compact_state({"messages": _messages()}, _runtime(), force=True)
assert result is not None
assert result.summary_text == "from-run-model"
class _RaisingTextResponse:
"""A provider response whose ``.text`` accessor fails — a realistic malformed result."""
@property
def text(self):
raise ValueError("malformed content blocks")
def _text_raises_model() -> MagicMock:
model = MagicMock()
model.with_config.return_value = model
model.invoke.return_value = _RaisingTextResponse()
model.ainvoke = AsyncMock(return_value=_RaisingTextResponse())
return model
def test_text_extraction_failure_falls_back_to_run_model(monkeypatch):
"""A response whose ``.text`` accessor raises is part of consuming the provider
result, so it must be a candidate failure that falls back to the run model not an
exception that escapes automatic compaction."""
built: list = []
monkeypatch.setattr("deerflow.agents.middlewares.summarization_middleware.create_chat_model", _tracking_create_chat_model(built))
primary = _text_raises_model()
middleware = DeerFlowSummarizationMiddleware(
model=primary,
trigger=("messages", 4),
keep=("messages", 2),
token_counter=len,
app_config=_app_config(("default-model", "run-model")),
configured_model_name="summary-model",
run_model_name="run-model",
)
result = middleware.before_model({"messages": _messages()}, _runtime())
primary.invoke.assert_called_once() # primary invoked; its .text raised
assert built == ["run-model"] # fell back to the run model
assert result is not None
assert result["summary_text"] == "from-run-model"
@pytest.mark.anyio
async def test_text_extraction_failure_falls_back_to_run_model_async(monkeypatch):
"""Async counterpart: a ``.text`` accessor failure falls back to the run model."""
built: list = []
monkeypatch.setattr("deerflow.agents.middlewares.summarization_middleware.create_chat_model", _tracking_create_chat_model(built))
primary = _text_raises_model()
middleware = DeerFlowSummarizationMiddleware(
model=primary,
trigger=("messages", 4),
keep=("messages", 2),
token_counter=len,
app_config=_app_config(("default-model", "run-model")),
configured_model_name="summary-model",
run_model_name="run-model",
)
result = await middleware.abefore_model({"messages": _messages()}, _runtime())
primary.ainvoke.assert_awaited_once()
assert built == ["run-model"]
assert result is not None
assert result["summary_text"] == "from-run-model"

View File

@ -681,10 +681,11 @@ def test_subagent_runtime_middlewares_attach_durable_context_before_summarizatio
sentinel = object() sentinel = object()
captured: dict[str, object] = {} captured: dict[str, object] = {}
def fake_create_summarization_middleware(*, app_config=None, keep=None, skip_memory_flush=False): def fake_create_summarization_middleware(*, app_config=None, keep=None, skip_memory_flush=False, run_model_name=None):
captured["app_config"] = app_config captured["app_config"] = app_config
captured["keep"] = keep captured["keep"] = keep
captured["skip_memory_flush"] = skip_memory_flush captured["skip_memory_flush"] = skip_memory_flush
captured["run_model_name"] = run_model_name
return sentinel return sentinel
# summarization is enabled by default False; flip it on so the factory path # summarization is enabled by default False; flip it on so the factory path
@ -703,6 +704,10 @@ def test_subagent_runtime_middlewares_attach_durable_context_before_summarizatio
# skip_memory_flush=True so subagent-internal turns are not flushed into the # skip_memory_flush=True so subagent-internal turns are not flushed into the
# PARENT thread's durable memory (#3875 Phase 3 review). # PARENT thread's durable memory (#3875 Phase 3 review).
assert captured["skip_memory_flush"] is True assert captured["skip_memory_flush"] is True
# Model ownership: the subagent's own resolved model is threaded into the factory
# so a distinct-model subagent summarizes with its model, not the parent's — the
# subagent context/configurable never carries the child model.
assert captured["run_model_name"] == "test-model"
durable = [middleware for middleware in middlewares if isinstance(middleware, DurableContextMiddleware)] durable = [middleware for middleware in middlewares if isinstance(middleware, DurableContextMiddleware)]
assert len(durable) == 1 assert len(durable) == 1
# ``_skills_root`` is ``posixpath.normpath(container_path)``, so compare against # ``_skills_root`` is ``posixpath.normpath(container_path)``, so compare against

View File

@ -1522,7 +1522,11 @@ title:
summarization: summarization:
enabled: true enabled: true
# Model to use for summarization (null = use default model) # Model to use for summarization.
# null = summarize with the model the run actually uses (the lead run's model, a
# subagent's own model, or a thread's custom-agent model), NOT models[0].
# set = that model generates; if its provider fails, compaction falls back to the
# run's own model so a broken summary provider cannot disable compaction.
# Recommended: Use a lightweight, cost-effective model like "gpt-4o-mini" or similar # Recommended: Use a lightweight, cost-effective model like "gpt-4o-mini" or similar
model_name: null model_name: null

View File

@ -1001,6 +1001,8 @@ export function InputBox({
signal, signal,
agentName: agentName:
typeof context.agent_name === "string" ? context.agent_name : null, typeof context.agent_name === "string" ? context.agent_name : null,
modelName:
typeof context.model_name === "string" ? context.model_name : null,
}); });
if ( if (
!isCurrentGoalRequest(compactRequestStateRef.current, request, threadId) !isCurrentGoalRequest(compactRequestStateRef.current, request, threadId)
@ -1039,6 +1041,7 @@ export function InputBox({
} }
}, [ }, [
context.agent_name, context.agent_name,
context.model_name,
queryClient, queryClient,
t.inputBox.compactFailed, t.inputBox.compactFailed,
t.inputBox.compactSkipped, t.inputBox.compactSkipped,

View File

@ -17,6 +17,7 @@ export type ThreadCompactResponse = {
export type CompactThreadContextOptions = { export type CompactThreadContextOptions = {
signal?: AbortSignal; signal?: AbortSignal;
agentName?: string | null; agentName?: string | null;
modelName?: string | null;
}; };
export type ThreadBranchResponse = { export type ThreadBranchResponse = {
@ -110,6 +111,7 @@ export async function compactThreadContext(
body: JSON.stringify({ body: JSON.stringify({
force: true, force: true,
...(options.agentName ? { agent_name: options.agentName } : {}), ...(options.agentName ? { agent_name: options.agentName } : {}),
...(options.modelName ? { model_name: options.modelName } : {}),
}), }),
signal: options.signal, signal: options.signal,
}, },