fix(sdk): summarization: truncate trailing ToolMessages to keep context within keep limit (#3405)

libs/deepagents/deepagents/middleware/_message_eviction.py

@@ -0,0 +1,162 @@
+# ruff: noqa: E501
+"""Shared helpers for evicting/clipping large message content with a head+tail preview.
+
+Used by:
+
+- `FilesystemMiddleware` — proactive per-tool-call offload when a tool result
+    exceeds its configured size threshold.
+- `SummarizationMiddleware` — reactive tail-clipping in the fallback
+    summarization path after a `ContextOverflowError`.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, cast
+
+from langchain_core.messages import BaseMessage, ToolMessage
+
+from deepagents.backends.utils import format_content_with_line_numbers, sanitize_tool_call_id
+
+if TYPE_CHECKING:
+    from langchain_core.messages.content import ContentBlock
+
+    from deepagents.backends.protocol import BackendProtocol
+
+TOO_LARGE_TOOL_MSG = """Tool result too large, the result of this tool call {tool_call_id} was saved in the filesystem at this path: {file_path}
+
+You can read the result from the filesystem by using the read_file tool, but make sure to only read part of the result at a time.
+
+You can do this by specifying an offset and limit in the read_file tool call. For example, to read the first 100 lines, you can use the read_file tool with offset=0 and limit=100.
+
+Here is a preview showing the head and tail of the result (lines of the form `... [N lines truncated] ...` indicate omitted lines in the middle of the content):
+
+{content_sample}
+"""
+
+
+def _create_content_preview(content_str: str, *, head_lines: int = 5, tail_lines: int = 5) -> str:
+    """Create a preview of content showing head and tail with truncation marker.
+
+    Args:
+        content_str: The full content string to preview.
+        head_lines: Number of lines to show from the start.
+        tail_lines: Number of lines to show from the end.
+
+    Returns:
+        Formatted preview string with line numbers.
+    """
+    lines = content_str.splitlines()
+
+    if len(lines) <= head_lines + tail_lines:
+        # If file is small enough, show all lines
+        preview_lines = [line[:1000] for line in lines]
+        return format_content_with_line_numbers(preview_lines, start_line=1)
+
+    # Show head and tail with truncation marker
+    head = [line[:1000] for line in lines[:head_lines]]
+    tail = [line[:1000] for line in lines[-tail_lines:]]
+
+    head_sample = format_content_with_line_numbers(head, start_line=1)
+    truncation_notice = f"\n... [{len(lines) - head_lines - tail_lines} lines truncated] ...\n"
+    tail_sample = format_content_with_line_numbers(tail, start_line=len(lines) - tail_lines + 1)
+
+    return head_sample + truncation_notice + tail_sample
+
+
+def _extract_text_from_message(message: BaseMessage) -> str:
+    """Extract text from a message using its `content_blocks` property.
+
+    Joins all text content blocks and ignores non-text blocks (images, audio, etc.)
+    so that binary payloads don't inflate the size measurement.
+
+    Args:
+        message: The BaseMessage to extract text from.
+
+    Returns:
+        Joined text from all text content blocks, or stringified content as fallback.
+    """
+    texts = [block["text"] for block in message.content_blocks if block["type"] == "text"]
+    return "\n".join(texts)
+
+
+def _build_evicted_content(message: ToolMessage, replacement_text: str) -> str | list[ContentBlock]:
+    """Build replacement content for an evicted message, preserving non-text blocks.
+
+    For plain string content, returns the replacement text directly. For list content
+    with mixed block types (e.g., text + image), replaces all text blocks with a single
+    text block containing the replacement text while keeping non-text blocks intact.
+
+    Args:
+        message: The original ToolMessage being evicted.
+        replacement_text: The truncation notice and preview text.
+
+    Returns:
+        Replacement content: a string or list of content blocks.
+    """
+    if isinstance(message.content, str):
+        return replacement_text
+    media_blocks = [block for block in message.content_blocks if block["type"] != "text"]
+    if not media_blocks:
+        # All content is text, so a plain string replacement is sufficient.
+        return replacement_text
+    return [cast("ContentBlock", {"type": "text", "text": replacement_text}), *media_blocks]
+
+
+def _build_evicted_tool_message(message: ToolMessage, evicted_content: str | list[ContentBlock]) -> ToolMessage:
+    """Build a replacement `ToolMessage` carrying `evicted_content`, preserving identity fields."""
+    return ToolMessage(
+        content=cast("str | list[str | dict]", evicted_content),
+        tool_call_id=message.tool_call_id,
+        name=message.name,
+        id=message.id,
+        artifact=message.artifact,
+        status=message.status,
+        additional_kwargs=dict(message.additional_kwargs),
+        response_metadata=dict(message.response_metadata),
+    )
+
+
+def _offload_tool_message_content(
+    message: ToolMessage,
+    content_str: str,
+    backend: BackendProtocol,
+    large_tool_results_prefix: str,
+) -> ToolMessage | None:
+    """Write `content_str` to `{prefix}/{tool_call_id}` and return a clipped replacement.
+
+    The replacement carries a head+tail preview and the offload path in
+    `TOO_LARGE_TOOL_MSG` format so the agent can `read_file` the full content
+    by tool_call_id. Returns `None` if the backend write fails — caller should
+    keep the original message in that case.
+    """
+    sanitized_id = sanitize_tool_call_id(message.tool_call_id) if message.tool_call_id else "unknown"
+    file_path = f"{large_tool_results_prefix}/{sanitized_id}"
+    result = backend.write(file_path, content_str)
+    if result is None or result.error:
+        return None
+    replacement_text = TOO_LARGE_TOOL_MSG.format(
+        tool_call_id=message.tool_call_id,
+        file_path=file_path,
+        content_sample=_create_content_preview(content_str),
+    )
+    return _build_evicted_tool_message(message, _build_evicted_content(message, replacement_text))
+
+
+async def _aoffload_tool_message_content(
+    message: ToolMessage,
+    content_str: str,
+    backend: BackendProtocol,
+    large_tool_results_prefix: str,
+) -> ToolMessage | None:
+    """Async variant of `_offload_tool_message_content` using `backend.awrite`."""
+    sanitized_id = sanitize_tool_call_id(message.tool_call_id) if message.tool_call_id else "unknown"
+    file_path = f"{large_tool_results_prefix}/{sanitized_id}"
+    result = await backend.awrite(file_path, content_str)
+    if result is None or result.error:
+        return None
+    replacement_text = TOO_LARGE_TOOL_MSG.format(
+        tool_call_id=message.tool_call_id,
+        file_path=file_path,
+        content_sample=_create_content_preview(content_str),
+    )
+    return _build_evicted_tool_message(message, _build_evicted_content(message, replacement_text))

libs/deepagents/deepagents/middleware/_overflow_clip.py

@@ -0,0 +1,206 @@
+"""Read-side clipping for the summarization-on-overflow fallback path.
+
+When `SummarizationMiddleware`'s `wrap_model_call` catches a
+`ContextOverflowError`, it falls through to summarization and *also* invokes
+`_clip_overflow_tail` (or its async variant) to shrink the trailing
+ToolMessage batch in the preserved suffix. Two per-TM paths:
+
+- `read_file` tool result: head-slice the content and append a notice
+    pointing back to the original `file_path` argument. No new backend write
+    is needed because the original file already lives at that path.
+- Any other tool result: full offload to `/large_tool_results/{tool_call_id}`
+    via the shared eviction helper, then replace the message with a
+    `TOO_LARGE_TOOL_MSG` stub.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+from typing import TYPE_CHECKING, Any, cast
+
+from langchain_core.messages import AIMessage, AnyMessage, ToolMessage
+
+from deepagents.middleware._message_eviction import (
+    _aoffload_tool_message_content,
+    _extract_text_from_message,
+    _offload_tool_message_content,
+)
+
+if TYPE_CHECKING:
+    from langchain.agents.middleware.summarization import ContextSize, TokenCounter
+
+    from deepagents.backends.protocol import BackendProtocol
+
+
+def _derive_overflow_clip_threshold_tokens(keep: ContextSize, max_input_tokens: int | None) -> int:
+    """Derive a token threshold for tail-ToolMessage clipping from `keep`.
+
+    Returns the keep token budget. If `keep` is message-based (no token info),
+    falls back to 5_000 -- equivalent to a 20_000-char floor under a `chars / 4`
+    approximation.
+    """
+    kind, value = keep
+    if kind == "tokens":
+        return int(value)
+    if kind == "fraction":
+        if max_input_tokens is None:
+            return 5_000
+        return int(max_input_tokens * value)
+    return 5_000
+
+
+def _find_tail_tool_message_batch(messages: list[AnyMessage]) -> tuple[int, list[ToolMessage]] | None:
+    """Return `(start_index, batch)` if `messages` ends with consecutive ToolMessages."""
+    if not messages or not isinstance(messages[-1], ToolMessage):
+        return None
+    i = len(messages) - 1
+    while i >= 0 and isinstance(messages[i], ToolMessage):
+        i -= 1
+    start = i + 1
+    return start, [cast("ToolMessage", m) for m in messages[start:]]
+
+
+def _build_tool_call_index(messages: list[AnyMessage]) -> dict[str, dict[str, Any]]:
+    """Map `tool_call_id` -> tool_call dict for all AIMessage tool_calls in `messages`."""
+    index: dict[str, dict[str, Any]] = {}
+    for m in messages:
+        if isinstance(m, AIMessage):
+            for tc in m.tool_calls or []:
+                tcid = tc.get("id")
+                if tcid:
+                    index[tcid] = cast("dict[str, Any]", tc)
+    return index
+
+
+def _slice_read_file_tm(msg: ToolMessage, original_path: str) -> ToolMessage:
+    """Slice a `read_file` ToolMessage's content to ~4k head chars and append a path-pointer notice.
+
+    `read_file` results don't need a fresh `/large_tool_results/{tcid}` write -- the
+    full file is already on the backend at `original_path`, and the agent can
+    recover with `read_file(file_path=original_path, offset=N, limit=K)`. The
+    truncation notice mirrors `READ_FILE_TRUNCATION_MSG` in shape so the
+    agent encounters a consistent format whether the tool truncated itself
+    or the middleware did.
+    """
+    content = _extract_text_from_message(msg)
+    notice = (
+        f"\n\n[Output was truncated due to context window size limits. "
+        f"The full content is at {original_path}. "
+        f"Use read_file with offset and limit parameters to retrieve specific portions. "
+        f"For example, to read the first 100 lines, call read_file with file_path='{original_path}', offset=0, limit=100.]"
+    )
+    return msg.model_copy(update={"content": content[:4_000] + notice})
+
+
+def _read_file_original_path(msg: ToolMessage, tc_index: dict[str, dict[str, Any]]) -> str | None:
+    """Return the `file_path` arg from the matching read_file tool_call, or `None`."""
+    tc = tc_index.get(msg.tool_call_id) if msg.tool_call_id else None
+    if not tc or tc.get("name") != "read_file":
+        return None
+    path = tc.get("args", {}).get("file_path")
+    return path if isinstance(path, str) and path else None
+
+
+def _clip_one_tail_message(
+    msg: ToolMessage,
+    tc_index: dict[str, dict[str, Any]],
+    backend: BackendProtocol,
+    large_tool_results_prefix: str,
+) -> ToolMessage | None:
+    """Apply the appropriate per-TM clip: read_file slice vs generic eviction."""
+    original_path = _read_file_original_path(msg, tc_index)
+    if original_path is not None:
+        return _slice_read_file_tm(msg, original_path)
+    return _offload_tool_message_content(msg, _extract_text_from_message(msg), backend, large_tool_results_prefix)
+
+
+async def _aclip_one_tail_message(
+    msg: ToolMessage,
+    tc_index: dict[str, dict[str, Any]],
+    backend: BackendProtocol,
+    large_tool_results_prefix: str,
+) -> ToolMessage | None:
+    """Async variant of `_clip_one_tail_message`."""
+    original_path = _read_file_original_path(msg, tc_index)
+    if original_path is not None:
+        return _slice_read_file_tm(msg, original_path)
+    return await _aoffload_tool_message_content(msg, _extract_text_from_message(msg), backend, large_tool_results_prefix)
+
+
+def _clip_overflow_tail(
+    preserved_messages: list[AnyMessage],
+    backend: BackendProtocol,
+    *,
+    keep: ContextSize,
+    max_input_tokens: int | None,
+    token_counter: TokenCounter,
+    large_tool_results_prefix: str,
+) -> tuple[list[AnyMessage], list[AnyMessage]]:
+    """Offload the trailing ToolMessage batch when it's large enough to matter.
+
+    Engages only when `preserved_messages` ends with consecutive ToolMessages
+    whose combined token count reaches `_derive_overflow_clip_threshold_tokens()`.
+    Each large TM is written under `large_tool_results/{tool_call_id}` and
+    replaced in-place by an offload-pointer ToolMessage.
+
+    Returns `(modified preserved_messages, replacement TMs to persist in
+    state)`. Replacements carry the original ids so the `add_messages`
+    reducer overwrites the originals when the caller propagates them via
+    a `Command` update. The replacements list omits any TM whose backend
+    write failed (those keep their originals in both lists).
+    """
+    found = _find_tail_tool_message_batch(preserved_messages)
+    if found is None:
+        return preserved_messages, []
+    start, tail = found
+    if token_counter(tail) < _derive_overflow_clip_threshold_tokens(keep, max_input_tokens):
+        return preserved_messages, []
+    tc_index = _build_tool_call_index(preserved_messages)
+    new_tail: list[AnyMessage] = []
+    any_clipped = False
+    for m in tail:
+        r = _clip_one_tail_message(m, tc_index, backend, large_tool_results_prefix)
+        if r is not None:
+            if r.id is None:
+                r = r.model_copy(update={"id": str(uuid.uuid4())})
+            new_tail.append(r)
+            any_clipped = True
+        else:
+            new_tail.append(m)
+    if not any_clipped:
+        return preserved_messages, []
+    return [*preserved_messages[:start], *new_tail], new_tail
+
+
+async def _aclip_overflow_tail(
+    preserved_messages: list[AnyMessage],
+    backend: BackendProtocol,
+    *,
+    keep: ContextSize,
+    max_input_tokens: int | None,
+    token_counter: TokenCounter,
+    large_tool_results_prefix: str,
+) -> tuple[list[AnyMessage], list[AnyMessage]]:
+    """Async variant of `_clip_overflow_tail`. Offloads each tail TM concurrently."""
+    found = _find_tail_tool_message_batch(preserved_messages)
+    if found is None:
+        return preserved_messages, []
+    start, tail = found
+    if token_counter(tail) < _derive_overflow_clip_threshold_tokens(keep, max_input_tokens):
+        return preserved_messages, []
+    tc_index = _build_tool_call_index(preserved_messages)
+    results = await asyncio.gather(*(_aclip_one_tail_message(m, tc_index, backend, large_tool_results_prefix) for m in tail))
+    new_tail: list[AnyMessage] = []
+    any_clipped = False
+    for r, m in zip(results, tail, strict=True):
+        if r is not None:
+            if r.id is None:
+                r = r.model_copy(update={"id": str(uuid.uuid4())})  # noqa: PLW2901
+            new_tail.append(r)
+            any_clipped = True
+        else:
+            new_tail.append(m)
+    if not any_clipped:
+        return preserved_messages, []
+    return [*preserved_messages[:start], *new_tail], new_tail

libs/deepagents/deepagents/middleware/filesystem.py

@@ -26,7 +26,7 @@ from langchain.agents.middleware.types import (
 )
 from langchain.tools import ToolRuntime
 from langchain.tools.tool_node import ToolCallRequest
-from langchain_core.messages import AnyMessage, BaseMessage, HumanMessage, ToolMessage
+from langchain_core.messages import AnyMessage, HumanMessage, ToolMessage
 from langchain_core.messages.content import ContentBlock
 from langchain_core.tools import BaseTool, StructuredTool
 from langgraph.channels.delta import DeltaChannel
@@ -53,10 +53,17 @@ from deepagents.backends.utils import (
     check_empty_content,
     format_content_with_line_numbers,
     format_grep_matches,
-    sanitize_tool_call_id,
+    sanitize_tool_call_id as sanitize_tool_call_id,
     truncate_if_too_long,
     validate_path,
 )
+from deepagents.middleware._message_eviction import (
+    TOO_LARGE_TOOL_MSG as TOO_LARGE_TOOL_MSG,
+    _aoffload_tool_message_content,
+    _create_content_preview,
+    _extract_text_from_message,
+    _offload_tool_message_content,
+)
 from deepagents.middleware._utils import append_to_system_message
 
 _FS_WCMATCH_FLAGS = wcglob.BRACE | wcglob.GLOBSTAR
@@ -523,17 +530,6 @@ TOOLS_EXCLUDED_FROM_EVICTION = (
 )
 
 
-TOO_LARGE_TOOL_MSG = """Tool result too large, the result of this tool call {tool_call_id} was saved in the filesystem at this path: {file_path}
-
-You can read the result from the filesystem by using the read_file tool, but make sure to only read part of the result at a time.
-
-You can do this by specifying an offset and limit in the read_file tool call. For example, to read the first 100 lines, you can use the read_file tool with offset=0 and limit=100.
-
-Here is a preview showing the head and tail of the result (lines of the form `... [N lines truncated] ...` indicate omitted lines in the middle of the content):
-
-{content_sample}
-"""
-
 TOO_LARGE_HUMAN_MSG = """Message content too large and was saved to the filesystem at: {file_path}
 
 You can read the full content using the read_file tool with pagination (offset and limit parameters).
@@ -593,74 +589,6 @@ def _build_truncated_human_message(message: HumanMessage, file_path: str) -> Hum
     return message.model_copy(update={"content": evicted})
 
 
-def _create_content_preview(content_str: str, *, head_lines: int = 5, tail_lines: int = 5) -> str:
-    """Create a preview of content showing head and tail with truncation marker.
-
-    Args:
-        content_str: The full content string to preview.
-        head_lines: Number of lines to show from the start.
-        tail_lines: Number of lines to show from the end.
-
-    Returns:
-        Formatted preview string with line numbers.
-    """
-    lines = content_str.splitlines()
-
-    if len(lines) <= head_lines + tail_lines:
-        # If file is small enough, show all lines
-        preview_lines = [line[:1000] for line in lines]
-        return format_content_with_line_numbers(preview_lines, start_line=1)
-
-    # Show head and tail with truncation marker
-    head = [line[:1000] for line in lines[:head_lines]]
-    tail = [line[:1000] for line in lines[-tail_lines:]]
-
-    head_sample = format_content_with_line_numbers(head, start_line=1)
-    truncation_notice = f"\n... [{len(lines) - head_lines - tail_lines} lines truncated] ...\n"
-    tail_sample = format_content_with_line_numbers(tail, start_line=len(lines) - tail_lines + 1)
-
-    return head_sample + truncation_notice + tail_sample
-
-
-def _extract_text_from_message(message: BaseMessage) -> str:
-    """Extract text from a message using its `content_blocks` property.
-
-    Joins all text content blocks and ignores non-text blocks (images, audio, etc.)
-    so that binary payloads don't inflate the size measurement.
-
-    Args:
-        message: The BaseMessage to extract text from.
-
-    Returns:
-        Joined text from all text content blocks, or stringified content as fallback.
-    """
-    texts = [block["text"] for block in message.content_blocks if block["type"] == "text"]
-    return "\n".join(texts)
-
-
-def _build_evicted_content(message: ToolMessage, replacement_text: str) -> str | list[ContentBlock]:
-    """Build replacement content for an evicted message, preserving non-text blocks.
-
-    For plain string content, returns the replacement text directly. For list content
-    with mixed block types (e.g., text + image), replaces all text blocks with a single
-    text block containing the replacement text while keeping non-text blocks intact.
-
-    Args:
-        message: The original ToolMessage being evicted.
-        replacement_text: The truncation notice and preview text.
-
-    Returns:
-        Replacement content: a string or list of content blocks.
-    """
-    if isinstance(message.content, str):
-        return replacement_text
-    media_blocks = [block for block in message.content_blocks if block["type"] != "text"]
-    if not media_blocks:
-        # All content is text, so a plain string replacement is sufficient.
-        return replacement_text
-    return [cast("ContentBlock", {"type": "text", "text": replacement_text}), *media_blocks]
-
-
 class FilesystemMiddleware(AgentMiddleware[FilesystemState, ContextT, ResponseT]):
     """Middleware for providing filesystem and optional execution tools to an agent.
 
@@ -1835,32 +1763,14 @@ class FilesystemMiddleware(AgentMiddleware[FilesystemState, ContextT, ResponseT]
         if len(content_str) <= NUM_CHARS_PER_TOKEN * self._tool_token_limit_before_evict:
             return message, False
 
-        # Write content to filesystem
-        sanitized_id = sanitize_tool_call_id(message.tool_call_id)
-        file_path = f"{self._large_tool_results_prefix}/{sanitized_id}"
-        result = resolved_backend.write(file_path, content_str)
-        if result.error:
+        processed_message = _offload_tool_message_content(
+            message,
+            content_str,
+            resolved_backend,
+            self._large_tool_results_prefix,
+        )
+        if processed_message is None:
             return message, False
-
-        # Create preview showing head and tail of the result
-        content_sample = _create_content_preview(content_str)
-        replacement_text = TOO_LARGE_TOOL_MSG.format(
-            tool_call_id=message.tool_call_id,
-            file_path=file_path,
-            content_sample=content_sample,
-        )
-
-        evicted = _build_evicted_content(message, replacement_text)
-        processed_message = ToolMessage(
-            content=cast("str | list[str | dict]", evicted),
-            tool_call_id=message.tool_call_id,
-            name=message.name,
-            id=message.id,
-            artifact=message.artifact,
-            status=message.status,
-            additional_kwargs=dict(message.additional_kwargs),
-            response_metadata=dict(message.response_metadata),
-        )
         return processed_message, True
 
     async def _aprocess_large_message(
@@ -1882,32 +1792,14 @@ class FilesystemMiddleware(AgentMiddleware[FilesystemState, ContextT, ResponseT]
         if len(content_str) <= NUM_CHARS_PER_TOKEN * self._tool_token_limit_before_evict:
             return message, False
 
-        # Write content to filesystem using async method
-        sanitized_id = sanitize_tool_call_id(message.tool_call_id)
-        file_path = f"{self._large_tool_results_prefix}/{sanitized_id}"
-        result = await resolved_backend.awrite(file_path, content_str)
-        if result.error:
+        processed_message = await _aoffload_tool_message_content(
+            message,
+            content_str,
+            resolved_backend,
+            self._large_tool_results_prefix,
+        )
+        if processed_message is None:
             return message, False
-
-        # Create preview showing head and tail of the result
-        content_sample = _create_content_preview(content_str)
-        replacement_text = TOO_LARGE_TOOL_MSG.format(
-            tool_call_id=message.tool_call_id,
-            file_path=file_path,
-            content_sample=content_sample,
-        )
-
-        evicted = _build_evicted_content(message, replacement_text)
-        processed_message = ToolMessage(
-            content=cast("str | list[str | dict]", evicted),
-            tool_call_id=message.tool_call_id,
-            name=message.name,
-            id=message.id,
-            artifact=message.artifact,
-            status=message.status,
-            additional_kwargs=dict(message.additional_kwargs),
-            response_metadata=dict(message.response_metadata),
-        )
         return processed_message, True
 
     def _get_backend_from_runtime(

libs/deepagents/deepagents/middleware/summarization.py

@@ -70,12 +70,13 @@ from langchain_core.exceptions import ContextOverflowError
 from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, SystemMessage, ToolMessage, get_buffer_string
 from langchain_core.messages.utils import count_tokens_approximately
 from langgraph.config import get_config
-from langgraph.types import Command
+from langgraph.types import Command, Overwrite
 from pydantic import BaseModel
 from typing_extensions import TypedDict
 
 from deepagents._api.deprecation import warn_deprecated
 from deepagents.backends import CompositeBackend
+from deepagents.middleware._overflow_clip import _aclip_overflow_tail, _clip_overflow_tail
 from deepagents.middleware._utils import append_to_system_message
 
 if TYPE_CHECKING:
@@ -315,6 +316,7 @@ class _DeepAgentsSummarizationMiddleware(AgentMiddleware):
         artifacts_root = backend.artifacts_root if isinstance(backend, CompositeBackend) else "/"
         _root = artifacts_root.rstrip("/")
         self._history_path_prefix = f"{_root}/conversation_history"
+        self._large_tool_results_prefix = f"{_root}/large_tool_results"
 
         if _deprecated_history_prefix is not None:
             self._history_path_prefix = _deprecated_history_prefix
@@ -957,11 +959,12 @@ A condensed summary follows:
         should_summarize = self._should_summarize(truncated_messages, total_tokens)
 
         # If no summarization needed, return with truncated messages
+        overflow_triggered = False
         if not should_summarize:
             try:
                 return handler(request.override(messages=truncated_messages))
             except ContextOverflowError:
-                pass
+                overflow_triggered = True
                 # Fallback to summarization on context overflow
 
         # Step 3: Perform summarization
@@ -972,9 +975,21 @@ A condensed summary follows:
 
         messages_to_summarize, preserved_messages = self._partition_messages(truncated_messages, cutoff_index)
 
+        backend = self._get_backend(request.state, request.runtime)
+        # On overflow, offload the large preserved tail TM batch to per-TM files.
+        new_state_tail: list[AnyMessage] = []
+        if overflow_triggered:
+            preserved_messages, new_state_tail = _clip_overflow_tail(
+                preserved_messages,
+                backend,
+                keep=self._lc_helper.keep,
+                max_input_tokens=self._get_profile_limits(),
+                token_counter=self.token_counter,
+                large_tool_results_prefix=self._large_tool_results_prefix,
+            )
+
         # Offload to backend first so history is preserved before summarization.
         # If offload fails, summarization still proceeds (with file_path=None).
-        backend = self._get_backend(request.state, request.runtime)
         file_path = self._offload_to_backend(backend, messages_to_summarize)
         if file_path is None:
             msg = "Offloading conversation history to backend failed during summarization. Older messages will not be recoverable."
@@ -1001,10 +1016,20 @@ A condensed summary follows:
         modified_messages = [*new_messages, *preserved_messages]
         response = handler(request.override(messages=modified_messages))
 
+        # Persist the clipped tail into state via Overwrite. Match the
+        # pattern from filesystem.py: atomically replace the messages channel
+        # so the clipped TMs survive DeltaChannel replay (append + reducer
+        # id-matching is broken until langchain-core auto-assigns ids upstream).
+        update: dict[str, Any] = {"_summarization_event": new_event}
+        if new_state_tail:
+            state_messages = list(request.state.get("messages", []))
+            new_state_messages = [*state_messages[: len(state_messages) - len(new_state_tail)], *new_state_tail]
+            update["messages"] = Overwrite(new_state_messages)
+
         # Return ExtendedModelResponse with state update
         return ExtendedModelResponse(
             model_response=response,
-            command=Command(update={"_summarization_event": new_event}),
+            command=Command(update=update),
         )
 
     async def awrap_model_call(
@@ -1061,11 +1086,12 @@ A condensed summary follows:
         should_summarize = self._should_summarize(truncated_messages, total_tokens)
 
         # If no summarization needed, return with truncated messages
+        overflow_triggered = False
         if not should_summarize:
             try:
                 return await handler(request.override(messages=truncated_messages))
             except ContextOverflowError:
-                pass
+                overflow_triggered = True
                 # Fallback to summarization on context overflow
 
         # Step 3: Perform summarization
@@ -1076,9 +1102,21 @@ A condensed summary follows:
 
         messages_to_summarize, preserved_messages = self._partition_messages(truncated_messages, cutoff_index)
 
+        backend = self._get_backend(request.state, request.runtime)
+        # On overflow, offload the large preserved tail TM batch to per-TM files.
+        new_state_tail: list[AnyMessage] = []
+        if overflow_triggered:
+            preserved_messages, new_state_tail = await _aclip_overflow_tail(
+                preserved_messages,
+                backend,
+                keep=self._lc_helper.keep,
+                max_input_tokens=self._get_profile_limits(),
+                token_counter=self.token_counter,
+                large_tool_results_prefix=self._large_tool_results_prefix,
+            )
+
         # Offload to backend and generate summary concurrently -- they are independent.
         # If offload fails, summarization still proceeds (with file_path=None).
-        backend = self._get_backend(request.state, request.runtime)
         file_path, summary = await asyncio.gather(
             self._aoffload_to_backend(backend, messages_to_summarize),
             self._acreate_summary(messages_to_summarize),
@@ -1105,10 +1143,20 @@ A condensed summary follows:
         modified_messages = [*new_messages, *preserved_messages]
         response = await handler(request.override(messages=modified_messages))
 
+        # Persist the clipped tail into state via Overwrite. Match the
+        # pattern from filesystem.py: atomically replace the messages channel
+        # so the clipped TMs survive DeltaChannel replay (append + reducer
+        # id-matching is broken until langchain-core auto-assigns ids upstream).
+        update: dict[str, Any] = {"_summarization_event": new_event}
+        if new_state_tail:
+            state_messages = list(request.state.get("messages", []))
+            new_state_messages = [*state_messages[: len(state_messages) - len(new_state_tail)], *new_state_tail]
+            update["messages"] = Overwrite(new_state_messages)
+
         # Return ExtendedModelResponse with state update
         return ExtendedModelResponse(
             model_response=response,
-            command=Command(update={"_summarization_event": new_event}),
+            command=Command(update=update),
         )
 
 

libs/deepagents/tests/unit_tests/test_end_to_end.py

@@ -12,6 +12,7 @@ import pytest
 from langchain.agents.middleware import AgentMiddleware
 from langchain.agents.middleware.types import ModelRequest, ModelResponse
 from langchain.tools import ToolRuntime
+from langchain_core.exceptions import ContextOverflowError
 from langchain_core.language_models import LanguageModelInput
 from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
 from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
@@ -3623,3 +3624,177 @@ def test_invalid_tool_call_patched_on_next_turn() -> None:
 
     # Final state must also expose the patched ToolMessage.
     assert any(isinstance(m, ToolMessage) and m.tool_call_id == "call_truncated" for m in result["messages"])
+
+
+_OVERFLOW_INPUT_CHAR_THRESHOLD = 50_000
+
+
+class _OverflowOnLargeInputModel(FakeChatModelWithHistory):
+    """Raises `ContextOverflowError` on any call whose input exceeds the char threshold.
+
+    Calls below the threshold pass through to the iterator (initial agent call,
+    summary-generation calls, post-clip retry). Raising on every over-threshold
+    call -- including the retry -- gives tests a built-in correctness check:
+    if clipping fails to reduce input below the threshold, the retry raises
+    and the test fails visibly.
+    """
+
+    def _generate(self, messages, stop=None, run_manager=None, **kwargs: Any):  # type: ignore[override]
+        total_chars = sum(len(m.content) for m in messages if isinstance(m.content, str))
+        if total_chars > _OVERFLOW_INPUT_CHAR_THRESHOLD:
+            self.call_history.append(
+                {"messages": messages, "kwargs": {"stop": stop, "run_manager": run_manager, **kwargs}, "tools": self.tools, "raised": True}
+            )
+            msg = "simulated overflow"
+            raise ContextOverflowError(msg)
+        return super()._generate(messages, stop, run_manager, **kwargs)
+
+
+def _make_parallel_tool_tail(
+    tool_name: str,
+    per_tm_chars: int,
+    n_tools: int = 4,
+    args_builder: Callable[[int], dict[str, Any]] | None = None,
+) -> list[Any]:
+    """Build [Human, AI(n_tools `tool_name` calls), n_tools x ToolMessage(per_tm_chars)].
+
+    Args:
+        tool_name: The tool name to use in each emitted tool_call.
+        per_tm_chars: Char length of each ToolMessage's content (all 'x').
+        n_tools: Number of parallel tool calls in the AI message.
+        args_builder: Optional `i -> args dict` mapper. Defaults to an empty args dict.
+    """
+    tool_call_ids = [f"tc_{i}" for i in range(n_tools)]
+    args_fn = args_builder if args_builder is not None else (lambda _i: {})
+    ai_msg = AIMessage(
+        content="",
+        tool_calls=[{"id": tcid, "name": tool_name, "args": args_fn(i), "type": "tool_call"} for i, tcid in enumerate(tool_call_ids)],
+    )
+    tool_msgs = [ToolMessage(content="x" * per_tm_chars, tool_call_id=tcid, name=tool_name) for tcid in tool_call_ids]
+    return [HumanMessage(content="query"), ai_msg, *tool_msgs]
+
+
+def test_summarization_clips_read_file_batch_on_overflow() -> None:
+    """End-to-end: 4 parallel real `read_file` calls hit pre-populated big files in state, then the clip path slices each.
+
+    Pre-populates `state["files"]` with 4 large files, has the fake model emit
+    4 parallel `read_file` tool_calls, lets the real `read_file` tool execute
+    against `StateBackend`, then triggers the overflow path and verifies:
+
+    - Each TM in state ends up as the read_file-specific slice (head + path
+      pointer to the original file).
+    - No `/large_tool_results/` files are written -- read_file's full content
+      already lives on the backend at the agent-passed path.
+    - The original files in state are untouched.
+    """
+    big_content = "\n".join(f"line {i}: " + "x" * 200 for i in range(120))  # ~25k chars, multi-line
+    initial_files = {f"/big_{i}.txt": {**create_file_data(big_content)} for i in range(4)}
+
+    fake_model = _OverflowOnLargeInputModel(
+        messages=iter(
+            [
+                AIMessage(
+                    content="",
+                    tool_calls=[
+                        {"id": f"tc_{i}", "name": "read_file", "args": {"file_path": f"/big_{i}.txt"}, "type": "tool_call"} for i in range(4)
+                    ],
+                ),
+                AIMessage(content="summary text"),
+                AIMessage(content="final response"),
+            ]
+        )
+    )
+    fake_model.call_history = []
+    fake_model.profile = {"max_input_tokens": 200_000}
+
+    agent = create_deep_agent(model=fake_model, checkpointer=InMemorySaver())
+    config = {"configurable": {"thread_id": "real-read-file-clip-test"}}
+
+    result = agent.invoke(
+        {"messages": [HumanMessage(content="read the files")], "files": initial_files},
+        config,
+    )
+
+    assert result["messages"][-1].content == "final response"
+
+    # Each ToolMessage in state should be the read_file-specific slice replacement.
+    state_tool_msgs = [m for m in result["messages"] if isinstance(m, ToolMessage)]
+    assert len(state_tool_msgs) == 4
+    for i, tm in enumerate(state_tool_msgs):
+        assert "Tool result too large" not in tm.content
+        assert f"/big_{i}.txt" in tm.content
+        assert "Output was truncated due to context window size limits" in tm.content
+        assert "Use read_file with offset and limit" in tm.content
+        # Sliced content is smaller than the original file.
+        assert len(tm.content) < len(big_content)
+
+    # No per-TM offload files were written -- read_file's full content is at the original path.
+    state = agent.get_state(config)
+    files = state.values.get("files", {})
+    assert not any(k.startswith("/large_tool_results/") for k in files)
+    # Original files are still intact.
+    for i in range(4):
+        assert f"/big_{i}.txt" in files
+
+
+def test_summarization_clips_ls_batch_on_overflow() -> None:
+    """On overflow with 4 parallel `ls` results, each TM is evicted via the generic offload.
+
+    `ls` isn't read_file, so the tail-clip path falls through to
+    `_offload_tool_message_content`: full content written under
+    `/large_tool_results/{tcid}` and the message replaced with a
+    `TOO_LARGE_TOOL_MSG` stub.
+    """
+    fake_model = _OverflowOnLargeInputModel(messages=iter([AIMessage(content="summary text"), AIMessage(content="final response")]))
+    fake_model.call_history = []
+    fake_model.profile = {"max_input_tokens": 200_000}
+
+    agent = create_deep_agent(model=fake_model, checkpointer=InMemorySaver())
+
+    input_messages = _make_parallel_tool_tail(
+        tool_name="ls",
+        per_tm_chars=25_000,
+        n_tools=4,
+        args_builder=lambda i: {"path": f"/dir_{i}"},
+    )
+    config = {"configurable": {"thread_id": "clip-ls-test"}}
+    result = agent.invoke({"messages": input_messages}, config)
+
+    assert len(fake_model.call_history) == 3
+    state_tool_msgs = [m for m in result["messages"] if isinstance(m, ToolMessage)]
+    assert len(state_tool_msgs) == 4
+    for tm in state_tool_msgs:
+        assert "Tool result too large" in tm.content
+        assert "/large_tool_results/" in tm.content
+
+    state = agent.get_state(config)
+    files = state.values.get("files", {})
+    for tcid in ("tc_0", "tc_1", "tc_2", "tc_3"):
+        assert f"/large_tool_results/{tcid}" in files, f"missing offload file for {tcid}"
+
+
+def test_summarization_clips_vanilla_tool_batch_on_overflow() -> None:
+    """On overflow with 4 parallel calls to an arbitrary user tool, each TM is evicted.
+
+    Same expected behavior as the `ls` test -- any non-read_file tool name routes
+    through the generic offload path.
+    """
+    fake_model = _OverflowOnLargeInputModel(messages=iter([AIMessage(content="summary text"), AIMessage(content="final response")]))
+    fake_model.call_history = []
+    fake_model.profile = {"max_input_tokens": 200_000}
+
+    agent = create_deep_agent(model=fake_model, checkpointer=InMemorySaver())
+
+    input_messages = _make_parallel_tool_tail(
+        tool_name="vendor_lookup",
+        per_tm_chars=25_000,
+        n_tools=4,
+        args_builder=lambda i: {"vendor_id": f"v_{i}"},
+    )
+    config = {"configurable": {"thread_id": "clip-vanilla-test"}}
+    _ = agent.invoke({"messages": input_messages}, config)
+
+    state = agent.get_state(config)
+    files = state.values.get("files", {})
+    for tcid in ("tc_0", "tc_1", "tc_2", "tc_3"):
+        assert f"/large_tool_results/{tcid}" in files, f"missing offload file for {tcid}"

libs/deepagents/tests/unit_tests/test_middleware.py

@@ -31,15 +31,17 @@ from deepagents.backends.utils import (
     truncate_if_too_long,
     update_file_data,
 )
+from deepagents.middleware._message_eviction import (
+    _build_evicted_content,
+    _create_content_preview,
+    _extract_text_from_message,
+)
 from deepagents.middleware.filesystem import (
     EMPTY_CONTENT_WARNING,
     NUM_CHARS_PER_TOKEN,
     FileData,
     FilesystemMiddleware,
     FilesystemState,
-    _build_evicted_content,
-    _create_content_preview,
-    _extract_text_from_message,
     supports_execution,
 )
 from deepagents.middleware.patch_tool_calls import PatchToolCallsMiddleware