fix(compaction): bound what a summary request is asked to read

docs/context-compaction-hardening.md

@@ -0,0 +1,64 @@
+# Context compaction hardening follow-up
+
+The immediate oversized-message fix intentionally remains small. A later
+follow-up should turn compaction into a complete, model-aware budgeting and
+recovery pipeline.
+
+## Budgeting
+
+- Represent the compact model's context window, input budget, output budget,
+  and safety margin explicitly.
+- Use the resolved compact model rather than the main model when their windows
+  differ.
+- Count the complete provider request, including wrappers, tool calls, message
+  content, context references, and output allowance.
+- Use the resolved model's tokenizer where available; keep the conservative
+  generic tokenizer only as a fallback.
+- Verify both the summary request and the post-compaction agent request.
+
+## Recoverability
+
+- Record explicit recovery metadata for tool observations, including whether
+  the operation is read-only and the durable arguments or handles needed to
+  repeat the read.
+- Do not infer recoverability from the message role or tool name alone.
+- Preserve file paths, URLs, resource identifiers, artifact handles, and other
+  durable locators separately from large raw payloads.
+- Never re-run writes, sends, executions, or other state-changing operations
+  to recover a dropped observation.
+
+## Oversized unrecoverable messages
+
+- Protect user, system, and assistant messages by default.
+- Summarize natural language in bounded chunks along paragraph or semantic
+  boundaries.
+- Split JSON, code, and tables only at structural boundaries.
+- Merge chunk summaries hierarchically while retaining source message IDs for
+  auditability.
+- Report an explicit blocked state when safe compaction is impossible.
+
+## Result and fallback semantics
+
+- Track which messages a summary covers and which original messages remain
+  protected.
+- Build the next context from the summary, protected originals, durable
+  recovery references, and the latest user request.
+- Ensure protected messages do not trigger the same ineffective compaction on
+  every turn.
+- Allow destructive truncation only for content already represented by a
+  summary or proven recoverable.
+- Distinguish input overflow, output rejection, unusable summaries, tokenizer
+  uncertainty, and unrecoverable content in trace metadata.
+
+## Test matrix
+
+- Every message role and mixtures of recoverable and unrecoverable content.
+- One oversized message and cumulative overflow from individually small ones.
+- Large tool calls, context references, multilingual text, emoji, code, JSON,
+  and tables.
+- Different main-model and compact-model windows.
+- Multiple protected messages, repeated compaction, checkpoint, and resume.
+- Provider input/output rejection and partial chunk-summary failure.
+- Invariants that successful compaction preserves user constraints, never
+  silently deletes unrecoverable data, fits the target window, and actually
+  shrinks the context.

src/xagent/core/agent/context/execution.py

@@ -5,10 +5,13 @@
 from dataclasses import dataclass, field, replace
 from datetime import datetime, timedelta, timezone
 from enum import Enum
+from functools import lru_cache
 from typing import Any
 from uuid import uuid4
 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
+import tiktoken
+
 from ...context_ref import (
     CONTEXT_REFS_KEY,
     ContextReference,
@@ -70,6 +73,24 @@
 # compaction judged worth the budget.
 COMPACT_CONTEXT_REFS_METADATA_KEY = "summary_context_refs"
 
+# Floor for the per-message cap on what compaction is asked to read. Unlike
+# the summary's output budget this has no ceiling: providers cap output far
+# below input, which is why ``COMPACT_SUMMARY_MAX_TOKENS`` exists, but nothing
+# caps a single input message except the window it has to fit in -- so the
+# input cap can scale with the window without ever outgrowing a provider
+# limit. The floor says only that a message this small was never what
+# exhausted a context window; it stops a tiny threshold from capping
+# everything to nothing.
+COMPACT_TRANSCRIPT_MESSAGE_MIN_TOKENS = 2048
+COMPACT_REQUEST_SAFETY_TOKENS = 512
+COMPACT_TOKEN_COUNT_CHUNK_CHARS = 1_024
+# The minimal safe allowlist for whole-message omission. A matching recorded
+# tool call is also required, so the summary retains the source path and exact
+# read arguments. Other tool results may contain one-time handles or write
+# receipts and must not be assumed recoverable merely because their role is
+# ``tool``.
+COMPACT_REREADABLE_TOOL_NAMES = frozenset({"read_file"})
+
 COMPACT_SUMMARY_MAX_TOKENS = 8192
 COMPACT_SUMMARY_MIN_TOKENS = 256
 # Budgets to fall back through when the requested one is refused, largest
@@ -98,6 +119,31 @@
 COMPACT_DROPPED_TOOL_NOTICE_MAX_CHARS = 1024
 COMPACT_DROPPED_TOOL_NAME_MAX_CHARS = 64
 
+
+@lru_cache(maxsize=1)
+def _compact_token_encoding() -> Any:
+    # get_encoding may download its merge table on the first cache miss. Keep
+    # that I/O out of module import so an offline deployment can still start.
+    return tiktoken.get_encoding("cl100k_base")
+
+
+def _count_compact_request_tokens(content: str) -> int:
+    # Encoding very long repetitive strings in one pass can make tiktoken's
+    # BPE merge work disproportionately expensive. Independent chunks also
+    # form a conservative count because tokens cannot merge across a chunk
+    # boundary.
+    encoding = _compact_token_encoding()
+    return sum(
+        len(
+            encoding.encode(
+                content[offset : offset + COMPACT_TOKEN_COUNT_CHUNK_CHARS],
+                disallowed_special=(),
+            )
+        )
+        for offset in range(0, len(content), COMPACT_TOKEN_COUNT_CHUNK_CHARS)
+    )
+
+
 # load_skill retrieves guidance, not evidence, and re-running it restores
 # nothing a dropped observation held.
 NON_EVIDENCE_TOOL_NAMES = CONTROL_TOOL_NAMES | {LOAD_SKILL_TOOL_NAME}
@@ -1125,7 +1171,9 @@ def compact_if_needed(self) -> CompactResult:
             strategy="none",
         )
 
-    def build_llm_compact_request_if_needed(self) -> dict[str, Any] | None:
+    def build_llm_compact_request_if_needed(
+        self, *, context_window: int | None = None
+    ) -> dict[str, Any] | None:
         if not self.compact_config.enabled:
             return None
 
@@ -1138,15 +1186,77 @@ def build_llm_compact_request_if_needed(self) -> dict[str, Any] | None:
             return None
 
         max_tokens = self._llm_compact_max_tokens()
+        omitted: list[dict[str, Any]] = []
+        messages = self._build_llm_compact_prompt(visible_messages, omitted=omitted)
+        metadata: dict[str, Any] = {
+            "original_tokens": total_tokens,
+            "threshold": self.compact_config.threshold,
+            "max_summary_tokens": max_tokens,
+        }
+        if not isinstance(context_window, int) or context_window <= 0:
+            metadata["llm_compact_context_window_unknown"] = True
+            return {
+                "blocked": True,
+                "messages": messages,
+                "original_tokens": total_tokens,
+                "max_tokens": 0,
+                "metadata": metadata,
+            }
+        try:
+            request_input_tokens = sum(
+                max(
+                    1,
+                    _count_compact_request_tokens(str(message.get("content") or "")),
+                )
+                for message in messages
+            )
+        except Exception as exc:  # noqa: BLE001
+            metadata.update(
+                {
+                    "llm_compact_tokenizer_unavailable": True,
+                    "compact_tokenizer_error_type": type(exc).__name__,
+                }
+            )
+            return {
+                "blocked": True,
+                "messages": messages,
+                "original_tokens": total_tokens,
+                "max_tokens": 0,
+                "metadata": metadata,
+            }
+        metadata["compact_request_input_tokens"] = request_input_tokens
+        metadata["compact_request_tokenizer"] = "cl100k_base"
+
+        available_output_tokens = (
+            context_window - request_input_tokens - COMPACT_REQUEST_SAFETY_TOKENS
+        )
+        metadata["compact_context_window"] = context_window
+        metadata["compact_request_safety_tokens"] = COMPACT_REQUEST_SAFETY_TOKENS
+        if available_output_tokens < COMPACT_SUMMARY_MIN_TOKENS:
+            metadata["llm_compact_request_too_large"] = True
+            return {
+                "blocked": True,
+                "messages": messages,
+                "original_tokens": total_tokens,
+                "max_tokens": 0,
+                "metadata": metadata,
+            }
+        if available_output_tokens < max_tokens:
+            max_tokens = available_output_tokens
+            metadata["max_summary_tokens"] = max_tokens
+            metadata["compact_budget_reduced_to"] = max_tokens
+        if omitted:
+            # Surfaced so a thin summary has a visible cause. Counts and sizes
+            # only -- never the omitted content, which is the whole reason it
+            # was left out.
+            metadata["omitted_message_count"] = len(omitted)
+            metadata["omitted_messages"] = omitted
+            metadata["max_message_tokens"] = self._compact_message_max_tokens()
         return {
-            "messages": self._build_llm_compact_prompt(visible_messages),
+            "messages": messages,
             "original_tokens": total_tokens,
             "max_tokens": max_tokens,
-            "metadata": {
-                "original_tokens": total_tokens,
-                "threshold": self.compact_config.threshold,
-                "max_summary_tokens": max_tokens,
-            },
+            "metadata": metadata,
         }
 
     def _drop_oldest_messages(self) -> CompactResult:
@@ -1440,9 +1550,11 @@ def _latest_visible_user_message(self) -> Message | None:
         return None
 
     def _build_llm_compact_prompt(
-        self, messages: list[Message]
+        self,
+        messages: list[Message],
+        omitted: list[dict[str, Any]] | None = None,
     ) -> list[dict[str, str]]:
-        transcript = self._compact_transcript(messages)
+        transcript = self._compact_transcript(messages, omitted=omitted)
         return [
             {
                 "role": "system",
@@ -1502,7 +1614,40 @@ def _llm_compact_max_tokens(self) -> int:
             min(COMPACT_SUMMARY_MAX_TOKENS, self.compact_config.threshold // 4),
         )
 
-    def _compact_transcript(self, messages: list[Message]) -> str:
+    def _compact_message_max_tokens(self) -> int:
+        """Cap on what any single message contributes to a summary request.
+
+        A summary is written by reading the history, so a message large enough
+        to exhaust the window on its own makes compaction impossible rather
+        than merely expensive: the request cannot be sent, the backstop then
+        finds a message count small enough that its tail window keeps
+        everything, and the context stays over budget with nothing able to
+        shrink it.
+
+        Shares the summary output budget's ``threshold // 4`` denominator on
+        purpose -- no single message should be able to claim more of the
+        request than the whole summary is allowed to produce.
+        """
+        return max(
+            COMPACT_TRANSCRIPT_MESSAGE_MIN_TOKENS,
+            self.compact_config.threshold // 4,
+        )
+
+    def _compact_transcript(
+        self,
+        messages: list[Message],
+        omitted: list[dict[str, Any]] | None = None,
+    ) -> str:
+        max_message_tokens = self._compact_message_max_tokens()
+        rereadable_tool_call_ids = {
+            str(tool_call["id"])
+            for message in messages
+            for tool_call in message.tool_calls or ()
+            if isinstance(tool_call, dict)
+            and tool_call.get("id")
+            and self._compact_tool_call_name(tool_call) in COMPACT_REREADABLE_TOOL_NAMES
+            and self._compact_tool_call_arguments(tool_call)
+        }
         chunks: list[str] = []
         for index, message in enumerate(messages, start=1):
             header = f"{index}. {message.role.upper()}"
@@ -1514,13 +1659,71 @@ def _compact_transcript(self, messages: list[Message]) -> str:
                     "tool_calls="
                     + json.dumps(message.tool_calls, ensure_ascii=False, default=str)
                 )
-            chunks.append(message.content)
+            content_tokens = max(1, len(message.content) // 4)
+            can_reread = (
+                message.role == "tool"
+                and message.metadata.get("tool_name") in COMPACT_REREADABLE_TOOL_NAMES
+                and message.tool_call_id is not None
+                and str(message.tool_call_id) in rereadable_tool_call_ids
+            )
+            # User, system, assistant, and non-rereadable tool messages are
+            # their own source. Omitting one here would permanently discard it
+            # when the summary replaces the raw history.
+            if can_reread and content_tokens > max_message_tokens:
+                chunks.append(self._omitted_content_notice(message, content_tokens))
+                if omitted is not None:
+                    omitted.append(
+                        {
+                            "index": index,
+                            "role": message.role,
+                            "tool_name": message.metadata.get("tool_name"),
+                            "estimated_tokens": content_tokens,
+                        }
+                    )
+            else:
+                chunks.append(message.content)
             if message.context_refs:
                 context_refs_text = message.context_refs_text()
                 if context_refs_text:
                     chunks.append(context_refs_text)
         return "\n".join(chunks)
 
+    @staticmethod
+    def _compact_tool_call_name(tool_call: dict[str, Any]) -> str:
+        function = tool_call.get("function")
+        if isinstance(function, dict):
+            return str(function.get("name") or "")
+        return str(tool_call.get("name") or "")
+
+    @staticmethod
+    def _compact_tool_call_arguments(tool_call: dict[str, Any]) -> Any:
+        function = tool_call.get("function")
+        if isinstance(function, dict):
+            return function.get("arguments")
+        return tool_call.get("args")
+
+    @staticmethod
+    def _omitted_content_notice(message: Message, content_tokens: int) -> str:
+        """Stand-in for a message too large to include in a summary request.
+
+        Replaces the whole message rather than a prefix of it. A byte-slice
+        can land inside a structured value -- a JSON key, an identifier -- and
+        the model completes the severed token by guessing, which reads as data
+        and is silently wrong. Naming what was dropped instead lets the
+        summary say the work happened and point at where to re-read it.
+
+        Deterministic by construction: the text is derived only from the
+        message, never from the clock or a request id, so retrying the same
+        compaction at a smaller output budget sends a byte-identical request.
+        """
+        tool_name = message.metadata.get("tool_name")
+        subject = f"{tool_name} result" if tool_name else "tool result"
+        return (
+            f"[content omitted from this summary request: {subject}, "
+            f"~{content_tokens} tokens. Re-run the recorded read-only tool "
+            f"call if its contents are needed rather than reconstructing them.]"
+        )
+
     @staticmethod
     def _is_reasoning_fallback(response: Any) -> bool:
         """True when the client substituted a reasoning trace for content.