@

feat(summarization): 非破坏性上下文压缩——保留原始对话仅在请求时瞬态压缩

重写 DeerFlowSummarizationMiddleware：将 stock before_model 的
RemoveMessage(REMOVE_ALL_MESSAGES) 改为 no-op，改在 wrap_model_call 中
瞬态压缩为 [summary, *recent]，完整原始对话仍保留在 state 中，前端继续
渲染用户真实问答。summary 消息标记 hide_from_ui 且永不持久化，按 thread
缓存以避免每次模型调用都触发 LLM。前端 isHiddenFromUIMessage 额外隐藏旧
版本持久化的 name="summary" 消息。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
@

frontend-web/src/core/messages/utils.ts

@@ -434,7 +434,17 @@ export function findToolCallResult(
 }
 
 export function isHiddenFromUIMessage(message: Message) {
-  return message.additional_kwargs?.hide_from_ui === true;
+  if (message.additional_kwargs?.hide_from_ui === true) {
+    return true;
+  }
+  // Conversation-summary messages are context for the model only — never shown
+  // to the user. New summaries are injected transiently (never persisted), but
+  // threads compacted by older builds still carry a persisted `name: "summary"`
+  // HumanMessage, so hide those too.
+  if (message.name === "summary") {
+    return true;
+  }
+  return false;
 }
 
 /**

offline-backend-20260512/backend/CLAUDE.md

@@ -163,7 +163,7 @@ Lead-agent middlewares are assembled in strict append order across `packages/har
 6. **GuardrailMiddleware** - Pre-tool-call authorization via pluggable `GuardrailProvider` protocol (optional, if `guardrails.enabled` in config). Evaluates each tool call and returns error ToolMessage on deny. Three provider options: built-in `AllowlistProvider` (zero deps), OAP policy providers (e.g. `aport-agent-guardrails`), or custom providers. See [docs/GUARDRAILS.md](docs/GUARDRAILS.md) for setup, usage, and how to implement a provider.
 7. **SandboxAuditMiddleware** - Audits sandboxed shell/file operations for security logging before tool execution continues
 8. **ToolErrorHandlingMiddleware** - Converts tool exceptions into error `ToolMessage`s so the run can continue instead of aborting
-9. **SummarizationMiddleware** - Context reduction when approaching token limits (optional, if enabled)
+9. **SummarizationMiddleware** (`DeerFlowSummarizationMiddleware`) - Context reduction when approaching token limits (optional, if enabled). **Non-destructive**: overrides the stock `before_model` (which persists a `RemoveMessage(REMOVE_ALL_MESSAGES)`) to a no-op and instead compresses **transiently** in `wrap_model_call` — the model request is replaced with `[summary, *recent]` while the full original conversation stays in state (so the frontend keeps rendering the user's real Q&A). The `summary` message is `hide_from_ui` and never persisted; summaries are cached per-thread (bounded, sticky boundary) to avoid an LLM call on every model invocation. Tests: `tests/test_summarization_middleware.py`
 10. **TodoListMiddleware** - Task tracking with `write_todos` tool (optional, if plan_mode)
 11. **TokenUsageMiddleware** - Records token usage metrics when token tracking is enabled (optional)
 12. **TitleMiddleware** - Two-phase auto-title (never shows "untitled"): `before_model` sets an instant question-derived *provisional* title the moment the first user message arrives (no LLM, zero latency), then `after_model` asynchronously generates the final LLM title and replaces it (tracked via the `title_provisional` state flag). Normalizes structured message content before prompting the title model
@@ -175,7 +175,7 @@ Lead-agent middlewares are assembled in strict append order across `packages/har
 18. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, interrupts via `Command(goto=END)` (must be last)
 8a. **ToolMetricsMiddleware** - Appended immediately after `ToolErrorHandlingMiddleware` in `_build_runtime_middlewares`, so it sits *deepest* in the `wrap_tool_call` chain. Records one `tool_call_metrics` row per tool/skill/MCP invocation — timing duration and bucketing failures into a coarse `error_category` (rate_limited / ip_blocked / auth_error / timeout / network_error / not_found / empty_result / tool_error). It catches both raised exceptions (before #8 converts them) and *self-handled* errors returned as error `ToolMessage`s (search tools that swallow HTTP 429/403). It also captures the targeted Agent Skill into `tool_call_metrics.skill_name`: the lead agent invokes a skill by `read_file`-ing its `/mnt/skills/<category>/<name>/SKILL.md` path (per the `<skill_system>` prompt block — there is no dedicated "use skill" tool), so `skill_name` is recovered by scanning every tool call's string arguments for the `skills/{public,custom}/{name}/` path shape (`_SKILL_PATH_RE`), plus the explicit `name` argument of `skill_view`/`skill_manage`/`skill_archive`. `skill_view`-specific success detection handles its swallowed failures (`Skill 'x' not found.` returned as a plain string rather than raised). Regression coverage: `tests/test_tool_metrics_middleware.py`. Writes are best-effort fire-and-forget. **Per-run skill merging**: `SqlToolMetricsStore.record` upserts skill rows by `(run_id, skill_name)` instead of inserting one row per call — a skill touched by several tool calls / retries inside one run collapses to a single row. Any success in the run wins (a skill that failed then finally succeeded is recorded as one `success`); a skill that only ever failed yields one `error` row whose `error_message` accumulates every attempt's message; durations are summed. Non-skill tool calls, and skill calls with no `run_id`, keep one row each. The read→merge→write is guarded by a per-event-loop `asyncio.Lock`. Surfaced by the admin `GET /api/admin/tool-metrics` router (which has a `kind=skill` filter and a per-skill `/summary` breakdown).
 9. **PromptPrefixMiddleware** - Injects the admin-configured prompt prefix into the model request only. The prefix is tagged onto the first human message's `additional_kwargs['prompt_prefix']` by the Gateway (`app.gateway.services`) and prepended to content here at call time, so the persisted/displayed message stays prefix-free
-10. **SummarizationMiddleware** - Context reduction when approaching token limits (optional, if enabled)
+10. **SummarizationMiddleware** (`DeerFlowSummarizationMiddleware`) - Context reduction when approaching token limits (optional, if enabled). **Non-destructive**: overrides the stock `before_model` (which persists a `RemoveMessage(REMOVE_ALL_MESSAGES)`) to a no-op and instead compresses **transiently** in `wrap_model_call` — the model request is replaced with `[summary, *recent]` while the full original conversation stays in state (so the frontend keeps rendering the user's real Q&A). The `summary` message is `hide_from_ui` and never persisted; summaries are cached per-thread (bounded, sticky boundary) to avoid an LLM call on every model invocation. Tests: `tests/test_summarization_middleware.py`
 11. **TodoListMiddleware** - Task tracking with `write_todos` tool (optional, if plan_mode)
 12. **TokenUsageMiddleware** - Records token usage metrics when token tracking is enabled (optional)
 13. **TitleMiddleware** - Two-phase auto-title (never shows "untitled"): `before_model` sets an instant question-derived *provisional* title the moment the first user message arrives (no LLM, zero latency), then `after_model` asynchronously generates the final LLM title and replaces it (tracked via the `title_provisional` state flag). Normalizes structured message content before prompting the title model

offline-backend-20260512/backend/docs/summarization.md

@@ -4,13 +4,26 @@ DeerFlow includes automatic conversation summarization to handle long conversati
 
 ## Overview
 
-The summarization feature uses LangChain's `SummarizationMiddleware` to monitor conversation history and trigger summarization based on configurable thresholds. When activated, it:
+The summarization feature uses DeerFlow's `DeerFlowSummarizationMiddleware` (a subclass of LangChain's `SummarizationMiddleware`) to monitor conversation history and trigger summarization based on configurable thresholds. When activated, it:
 
 1. Monitors message token counts in real-time
 2. Triggers summarization when thresholds are met
 3. Keeps recent messages intact while summarizing older exchanges
 4. Maintains AI/Tool message pairs together for context continuity
-5. Injects the summary back into the conversation
+5. Substitutes the summary for the older messages **in the model request only**
+
+### Non-destructive (display-preserving) compression
+
+> **Important:** DeerFlow compresses **transiently**, not destructively.
+
+LangChain's stock `SummarizationMiddleware` runs in `before_model` and persists a `RemoveMessage(REMOVE_ALL_MESSAGES)` to the checkpoint — which permanently deletes the original Q&A from both the model context *and* the conversation the frontend renders. DeerFlow overrides this:
+
+- `before_model` / `abefore_model` are **no-ops** (the destructive path is disabled).
+- Compression runs in `wrap_model_call` / `awrap_model_call`: for a single model call, the message list is replaced with `[summary, *recent]` via `request.override(messages=...)`. **Nothing is written back to state.**
+- The full original messages stay in the checkpoint, so the frontend keeps showing the user's real conversation. The summary message is named `summary` and flagged `hide_from_ui`, and because it only ever lives in the transient request it never reaches the frontend at all.
+- The generated summary is cached **per-thread** so we don't pay an extra LLM call on every model invocation. The cached summary is reused (no LLM call) until enough *new* messages accumulate to push the compressed context back over the trigger threshold (a sticky boundary mirroring the stock middleware's post-compaction behavior). The cache is bounded (oldest threads evicted first).
+
+> **Legacy threads:** conversations compacted by an older (destructive) build already lost their original messages and carry a persisted `summary` message. The frontend hides any `name: "summary"` message, so the stray summary bubble no longer shows — but those already-discarded messages cannot be recovered.
 
 ## Configuration
 
@@ -169,10 +182,7 @@ The default LangChain prompt instructs the model to:
    - Messages to summarize (older messages beyond the `keep` threshold)
    - Messages to preserve (recent messages within the `keep` threshold)
 4. **Summary Generation**: The model generates a concise summary of the older messages
-5. **Context Replacement**: The message history is updated:
-   - All old messages are removed
-   - A single summary message is added
-   - Recent messages are preserved
+5. **Context Replacement (transient)**: The message list **handed to the model for this call** is rebuilt as `[summary, *recent]`. The persisted state is left untouched — old messages are **not** removed, so the frontend still renders the full original conversation.
 6. **AI/Tool Pair Protection**: The system ensures AI messages and their corresponding tool messages stay together
 7. **Skill Rescue**: Before the summary is generated, the most recently loaded skill files (tool results whose tool name is in `skill_file_read_tool_names` and whose target path is under `skills.container_path`) are lifted out of the summarization set and prepended to the preserved tail. Selection walks newest-first under three budgets: `preserve_recent_skill_count`, `preserve_recent_skill_tokens`, and `preserve_recent_skill_tokens_per_skill`. The triggering AIMessage and all of its paired ToolMessages move together so tool_call ↔ tool_result pairing stays intact.
 
@@ -298,8 +308,9 @@ The middleware intelligently preserves message context:
 ### Code Structure
 
 - **Configuration**: `packages/harness/deerflow/config/summarization_config.py`
-- **Integration**: `packages/harness/deerflow/agents/lead_agent/agent.py`
-- **Middleware**: Uses `langchain.agents.middleware.SummarizationMiddleware`
+- **Integration**: `packages/harness/deerflow/agents/lead_agent/agent.py` (`_create_summarization_middleware`)
+- **Middleware**: `packages/harness/deerflow/agents/middlewares/summarization_middleware.py` — `DeerFlowSummarizationMiddleware`, subclassing `langchain.agents.middleware.SummarizationMiddleware` but overriding compaction to be transient (`wrap_model_call`) instead of destructive (`before_model`)
+- **Tests**: `tests/test_summarization_middleware.py`
 
 ### Middleware Order
 
@@ -313,9 +324,9 @@ Summarization runs after ThreadData and Sandbox initialization but before Title
 
 ### State Management
 
-- Summarization is stateless - configuration is loaded once at startup
-- Summaries are added as regular messages in the conversation history
-- The checkpointer persists the summarized history automatically
+- Configuration is loaded once at startup
+- Summarization is **non-destructive**: the summary replaces older messages only in the transient model request (`wrap_model_call`), never in persisted state. The checkpointer therefore keeps the full original conversation
+- The per-thread summary cache lives in the middleware instance (in-memory, bounded). It is a cost optimization only — a cold/stale cache simply re-summarizes, which is always correct
 
 ## Example Configurations
 

offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py

@@ -1,21 +1,41 @@
-"""Summarization middleware extensions for DeerFlow."""
+"""Summarization middleware extensions for DeerFlow.
+
+Unlike LangChain's stock ``SummarizationMiddleware`` (which runs in
+``before_model`` and **persists** a destructive ``RemoveMessage(REMOVE_ALL_MESSAGES)``
+to the checkpoint — wiping the original Q&A from both the model context *and*
+the conversation the frontend renders), this middleware compresses the history
+**transiently in ``wrap_model_call``**. The full original messages stay in state
+(so the UI keeps showing the user's real conversation); only the message list
+handed to the model for a single call is replaced with ``[summary, *recent]``.
+
+The generated summary is cached per-thread so we don't pay an extra LLM call on
+every model invocation: once a thread is summarized, the cached summary is reused
+until enough *new* messages accumulate to push the compressed context back over
+the trigger threshold (sticky boundary, mirroring the stock middleware's
+post-compaction behavior).
+"""
 
 from __future__ import annotations
 
 import logging
-from collections.abc import Collection
+from collections import OrderedDict
+from collections.abc import Awaitable, Callable, Collection
 from dataclasses import dataclass
 from typing import Any, Protocol, override, runtime_checkable
 
 from langchain.agents import AgentState
 from langchain.agents.middleware import SummarizationMiddleware
-from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, RemoveMessage, ToolMessage
+from langchain.agents.middleware.types import ModelCallResult, ModelRequest, ModelResponse
+from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, ToolMessage
 from langgraph.config import get_config
-from langgraph.graph.message import REMOVE_ALL_MESSAGES
 from langgraph.runtime import Runtime
 
 logger = logging.getLogger(__name__)
 
+# Upper bound on the per-thread summary cache so a long-lived process serving
+# many threads can't grow it without limit. Oldest entries are evicted first.
+_SUMMARY_CACHE_MAX_THREADS = 512
+
 
 @dataclass(frozen=True)
 class SummarizationEvent:
@@ -95,6 +115,22 @@ class _SkillBundle:
     skill_key: str
 
 
+@dataclass(frozen=True)
+class _SummaryCacheEntry:
+    """A cached, reusable summary for one thread.
+
+    ``summarized_ids`` are the ids of the original messages folded into
+    ``summary_text``. The cache is only reusable while every one of those ids is
+    still present in the live message list and no AIMessage/ToolMessage pair was
+    split when the boundary was drawn (``reusable``) — otherwise filtering by id
+    could orphan a tool result, so we fall back to re-summarizing.
+    """
+
+    summarized_ids: frozenset[str]
+    summary_text: str
+    reusable: bool
+
+
 class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
     """Summarization middleware with pre-compression hook dispatch and skill rescue."""
 
@@ -116,69 +152,181 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
         self._preserve_recent_skill_count = max(0, preserve_recent_skill_count)
         self._preserve_recent_skill_tokens = max(0, preserve_recent_skill_tokens)
         self._preserve_recent_skill_tokens_per_skill = max(0, preserve_recent_skill_tokens_per_skill)
+        # Per-thread cache of the most recent summary: thread_id -> (summarized message ids, summary text).
+        # Lets us reuse a summary across model calls instead of re-running the LLM every invocation.
+        self._summary_cache: OrderedDict[str, _SummaryCacheEntry] = OrderedDict()
 
+    @override
     def before_model(self, state: AgentState, runtime: Runtime) -> dict | None:
-        return self._maybe_summarize(state, runtime)
+        """Disable the stock middleware's destructive ``before_model`` compaction.
 
+        Compression now happens transiently in :meth:`wrap_model_call`, so the
+        persisted conversation (and the frontend's view of it) is never mutated.
+        """
+        return None
+
+    @override
     async def abefore_model(self, state: AgentState, runtime: Runtime) -> dict | None:
-        return await self._amaybe_summarize(state, runtime)
+        """Async counterpart of :meth:`before_model` — also a no-op."""
+        return None
 
-    def _maybe_summarize(self, state: AgentState, runtime: Runtime) -> dict | None:
-        messages = state["messages"]
-        self._ensure_message_ids(messages)
+    # ---- wrap_model_call: transient, display-preserving compression ----------
 
-        total_tokens = self.token_counter(messages)
-        if not self._should_summarize(messages, total_tokens):
+    @override
+    def wrap_model_call(
+        self,
+        request: ModelRequest,
+        handler: Callable[[ModelRequest], ModelResponse],
+    ) -> ModelCallResult:
+        compressed = self._compressed_messages(request)
+        if compressed is not None:
+            request = request.override(messages=compressed)
+        return handler(request)
+
+    @override
+    async def awrap_model_call(
+        self,
+        request: ModelRequest,
+        handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
+    ) -> ModelCallResult:
+        compressed = await self._acompressed_messages(request)
+        if compressed is not None:
+            request = request.override(messages=compressed)
+        return await handler(request)
+
+    def _compressed_messages(self, request: ModelRequest) -> list[AnyMessage] | None:
+        """Build the compressed message list for this call (sync summary path).
+
+        Returns ``None`` when no compression is needed (the request is passed
+        through untouched).
+        """
+        plan = self._plan_compression(request)
+        if plan is None:
             return None
-
-        cutoff_index = self._determine_cutoff_index(messages)
-        if cutoff_index <= 0:
-            return None
-
-        messages_to_summarize, preserved_messages = self._partition_with_skill_rescue(messages, cutoff_index)
-        self._fire_hooks(messages_to_summarize, preserved_messages, runtime)
+        action, payload = plan
+        if action == "reuse":
+            return payload  # type: ignore[return-value]
+        # action == "summarize"
+        messages_to_summarize, preserved, thread_id = payload  # type: ignore[misc]
+        self._fire_hooks(messages_to_summarize, preserved, getattr(request, "runtime", None))
         summary = self._create_summary(messages_to_summarize)
-        new_messages = self._build_new_messages(summary)
+        return self._finish_summarize(messages_to_summarize, preserved, summary, thread_id)
 
-        return {
-            "messages": [
-                RemoveMessage(id=REMOVE_ALL_MESSAGES),
-                *new_messages,
-                *preserved_messages,
-            ]
-        }
+    async def _acompressed_messages(self, request: ModelRequest) -> list[AnyMessage] | None:
+        """Async counterpart of :meth:`_compressed_messages`."""
+        plan = self._plan_compression(request)
+        if plan is None:
+            return None
+        action, payload = plan
+        if action == "reuse":
+            return payload  # type: ignore[return-value]
+        messages_to_summarize, preserved, thread_id = payload  # type: ignore[misc]
+        self._fire_hooks(messages_to_summarize, preserved, getattr(request, "runtime", None))
+        summary = await self._acreate_summary(messages_to_summarize)
+        return self._finish_summarize(messages_to_summarize, preserved, summary, thread_id)
 
-    async def _amaybe_summarize(self, state: AgentState, runtime: Runtime) -> dict | None:
-        messages = state["messages"]
+    def _plan_compression(
+        self,
+        request: ModelRequest,
+    ) -> tuple[str, Any] | None:
+        """Decide what to do for this model call, doing **no** LLM work.
+
+        Returns one of:
+
+        - ``None`` — pass the request through unchanged (no compression needed).
+        - ``("reuse", compressed_messages)`` — reuse the cached summary.
+        - ``("summarize", (messages_to_summarize, preserved, thread_id))`` — the
+          caller must run the (a)sync summarizer, then call
+          :meth:`_finish_summarize`.
+        """
+        messages = list(getattr(request, "messages", None) or [])
+        if not messages:
+            return None
         self._ensure_message_ids(messages)
+        thread_id = self._resolve_request_thread_id(request)
 
-        total_tokens = self.token_counter(messages)
-        if not self._should_summarize(messages, total_tokens):
+        cached = self._summary_cache.get(thread_id) if thread_id else None
+        present_ids = {m.id for m in messages}
+        if cached is not None and cached.reusable and cached.summarized_ids <= present_ids:
+            preserved = [m for m in messages if m.id not in cached.summarized_ids]
+            effective = [self._summary_message(cached.summary_text), *preserved]
+            if not self._should_summarize(effective, self.token_counter(effective)):
+                # Refresh LRU recency and reuse without an LLM call.
+                self._summary_cache.move_to_end(thread_id)
+                return ("reuse", effective)
+            # New messages have pushed the compressed context back over the
+            # threshold — fall through and re-summarize a larger prefix.
+
+        if not self._should_summarize(messages, self.token_counter(messages)):
             return None
 
         cutoff_index = self._determine_cutoff_index(messages)
         if cutoff_index <= 0:
             return None
 
-        messages_to_summarize, preserved_messages = self._partition_with_skill_rescue(messages, cutoff_index)
-        self._fire_hooks(messages_to_summarize, preserved_messages, runtime)
-        summary = await self._acreate_summary(messages_to_summarize)
-        new_messages = self._build_new_messages(summary)
+        messages_to_summarize, preserved = self._partition_with_skill_rescue(messages, cutoff_index)
+        if not messages_to_summarize:
+            return None
 
-        return {
-            "messages": [
-                RemoveMessage(id=REMOVE_ALL_MESSAGES),
-                *new_messages,
-                *preserved_messages,
-            ]
-        }
+        return ("summarize", (messages_to_summarize, preserved, thread_id))
+
+    def _finish_summarize(
+        self,
+        messages_to_summarize: list[AnyMessage],
+        preserved: list[AnyMessage],
+        summary: str,
+        thread_id: str | None,
+    ) -> list[AnyMessage]:
+        """Assemble the compressed message list and update the per-thread cache."""
+        compressed = [self._summary_message(summary), *preserved]
+
+        if thread_id:
+            summarized_ids = frozenset(m.id for m in messages_to_summarize if m.id is not None)
+            preserved_ids = {m.id for m in preserved if m.id is not None}
+            # A non-empty overlap means skill rescue split an AIMessage across the
+            # boundary; reconstructing ``preserved`` by id filtering later could
+            # orphan a tool result, so mark the entry non-reusable (re-summarize).
+            reusable = summarized_ids.isdisjoint(preserved_ids)
+            self._store_cache_entry(
+                thread_id,
+                _SummaryCacheEntry(summarized_ids=summarized_ids, summary_text=summary, reusable=reusable),
+            )
+
+        return compressed
+
+    def _store_cache_entry(self, thread_id: str, entry: _SummaryCacheEntry) -> None:
+        """Insert/refresh a cache entry, evicting the oldest thread if over capacity."""
+        self._summary_cache[thread_id] = entry
+        self._summary_cache.move_to_end(thread_id)
+        while len(self._summary_cache) > _SUMMARY_CACHE_MAX_THREADS:
+            self._summary_cache.popitem(last=False)
+
+    @staticmethod
+    def _resolve_request_thread_id(request: ModelRequest) -> str | None:
+        """Best-effort thread id from the model request's runtime."""
+        runtime = getattr(request, "runtime", None)
+        if runtime is None:
+            return None
+        return _resolve_thread_id(runtime)
+
+    def _summary_message(self, summary: str) -> HumanMessage:
+        """Build the summary message handed to the model.
+
+        It is named ``summary`` and flagged ``hide_from_ui`` defensively. Because
+        it only ever lives in the transient model request (never in persisted
+        state), the frontend never receives it — the user keeps seeing the
+        original conversation.
+        """
+        return HumanMessage(
+            content=f"Here is a summary of the conversation to date:\n\n{summary}",
+            name="summary",
+            additional_kwargs={"hide_from_ui": True, "lc_source": "summarization"},
+        )
 
     @override
     def _build_new_messages(self, summary: str) -> list[HumanMessage]:
-        """Override the base implementation to let the human message with the special name 'summary'.
-        And this message will be ignored to display in the frontend, but still can be used as context for the model.
-        """
-        return [HumanMessage(content=f"Here is a summary of the conversation to date:\n\n{summary}", name="summary")]
+        """Kept for backwards compatibility; delegates to :meth:`_summary_message`."""
+        return [self._summary_message(summary)]
 
     def _partition_with_skill_rescue(
         self,