From a774f20cc74c52b69107dcd92455d850044e8f7a Mon Sep 17 00:00:00 2001 From: rogercloud Date: Thu, 3 Sep 2026 13:47:05 +0800 Subject: [PATCH 1/5] fix(compaction): bound what a summary request is asked to read One oversized message could make compaction impossible rather than merely expensive, and leave the context with no way to shrink at all. A summary is written by reading the history. When a single tool result is large enough to exhaust the window on its own, the request cannot be sent -- and the backstop cannot rescue it either: a context that short has a tail window wide enough to keep every message, so nothing is dropped, ``removed_count`` is 0, and the next turn arrives in the same state. Stuck, not slow. The threshold already bounds ordinary growth. At the default ratio of 0.75, a 32k window compacts at 24000 tokens and asks for at most 6000 back, so input plus output sits inside the window with room to spare; history reaches that size gradually and is compacted on the way. What is missing is a bound on the outlier that arrives in one step. So this caps a single message, not the total: ``threshold // 4``, with a floor and deliberately no ceiling. It shares the summary output budget's denominator because no one message should be able to claim more of the request than the whole summary may produce. The floor says only that a message that small was never what exhausted a window, and keeps a tiny threshold from capping everything to nothing. No ceiling, because the reason ``COMPACT_SUMMARY_MAX_TOKENS`` exists -- providers cap output far below input -- has no counterpart here: nothing limits one input message except the window it must fit in, so scaling with the window cannot outgrow a provider limit. Oversized content is replaced whole, never sliced. A byte-slice can land inside a structured value, and the model completes the severed token by guessing, which reads as data and is silently wrong; that failure is the reason #1598 exists and it would have been reintroduced here. The stand-in names the tool, gives the size, and says the work happened and where to re-read it. The notice is derived only from the message -- never the clock, never a request id -- so the budget ladder's retries send a byte-identical request instead of defeating prefix caching. Verified: reverting that determinism turns its test red. ``omitted_messages`` and the cap ride on the request metadata into the compact trace event, so a thin summary has a visible cause. Counts and sizes only, never the omitted content. The read path that replays a stored summary looks at three keys and none of them is this one, so none of it reaches a replayed prefix. Each guarantee was mutation-tested: removing the cap, slicing instead of replacing, putting a clock value in the notice, and dropping the scaling each turn exactly their own tests red. --- src/xagent/core/agent/context/execution.py | 101 +++++++++++++++++++-- tests/core/agent/test_context.py | 72 +++++++++++++++ 2 files changed, 163 insertions(+), 10 deletions(-) diff --git a/src/xagent/core/agent/context/execution.py b/src/xagent/core/agent/context/execution.py index 994307f3d..f603f69d8 100644 --- a/src/xagent/core/agent/context/execution.py +++ b/src/xagent/core/agent/context/execution.py @@ -70,6 +70,16 @@ # 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_SUMMARY_MAX_TOKENS = 8192 COMPACT_SUMMARY_MIN_TOKENS = 256 # Budgets to fall back through when the requested one is refused, largest @@ -1138,15 +1148,25 @@ 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 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 +1460,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 +1524,31 @@ 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() chunks: list[str] = [] for index, message in enumerate(messages, start=1): header = f"{index}. {message.role.upper()}" @@ -1514,13 +1560,48 @@ 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) + if 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 _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 f"{message.role} message" + return ( + f"[content omitted from this summary request: {subject}, " + f"~{content_tokens} tokens. It ran and produced output; re-read " + f"the source for its contents rather than reconstructing them.]" + ) + @staticmethod def _is_reasoning_fallback(response: Any) -> bool: """True when the client substituted a reasoning trace for content. diff --git a/tests/core/agent/test_context.py b/tests/core/agent/test_context.py index c7e33e763..9026b4e5a 100644 --- a/tests/core/agent/test_context.py +++ b/tests/core/agent/test_context.py @@ -1993,3 +1993,75 @@ def test_compact_config_round_trip_drops_the_retired_strategy_key() -> None: assert restored.compact_config.threshold == 1234 assert restored.compact_config.max_messages == 7 assert not hasattr(restored.compact_config, "strategy") + + +def test_compact_request_omits_a_message_too_large_to_read() -> None: + """A single oversized result must not make compaction impossible. + + Compaction writes its summary by reading the history, so one message big + enough to exhaust the window on its own cannot be summarized at all -- + and the backstop cannot help either, because a context that short has a + tail window wide enough to keep every message, so nothing is dropped and + the context stays over budget with no way out. + """ + context = ExecutionContext(execution_id="oversized") + context.compact_config.threshold = 24000 + context.add_user_message("summarize the repo") + context.add_tool_result( + "read_file", {"output": "x" * 900_000}, tool_call_id="call-1" + ) + + request = context.build_llm_compact_request_if_needed() + + assert request is not None + assert request["original_tokens"] > 200_000 + prompt_tokens = sum(len(m["content"]) for m in request["messages"]) // 4 + # The point of the cap: the request is now sendable at all. + assert prompt_tokens < context.compact_config.threshold + omitted = request["metadata"]["omitted_messages"] + assert [entry["tool_name"] for entry in omitted] == ["read_file"] + assert request["metadata"]["omitted_message_count"] == 1 + + +def test_compact_request_keeps_messages_under_the_cap_verbatim() -> None: + """The cap is for outliers. Ordinary history must reach the summary + whole, or every summary silently degrades.""" + context = ExecutionContext(execution_id="ordinary") + context.compact_config.threshold = 24000 + context.add_user_message("a question worth summarizing") + context.add_tool_result("read_file", {"output": "y" * 4_000}, tool_call_id="call-1") + + request = context.build_llm_compact_request_if_needed() + + assert request is None or "omitted_messages" not in request["metadata"] + + +def test_oversized_content_is_replaced_whole_not_sliced() -> None: + """Never a half field. A byte-slice can land inside a structured value + and the model completes the severed token by guessing, which reads as + data and is silently wrong -- the failure this replacement exists to + avoid. The stand-in must also be free of any per-turn value, so retrying + the same compaction at a lower output budget sends an identical request. + """ + context = ExecutionContext(execution_id="whole-not-sliced") + context.compact_config.threshold = 24000 + context.add_user_message("go") + context.add_tool_result( + "fetch_page", + { + "output": '{"handle": {"workspace": "4b33784773d5", "branch": "main"}}' + * 5000 + }, + tool_call_id="call-1", + ) + + first = context.build_llm_compact_request_if_needed() + second = context.build_llm_compact_request_if_needed() + + assert first is not None and second is not None + transcript = first["messages"][-1]["content"] + assert "4b33784773d5" not in transcript + assert "fetch_page result" in transcript + # Byte-identical across builds: nothing in the notice comes from the + # clock or a request id. + assert transcript == second["messages"][-1]["content"] From a675d560459276ffd36d1dd1bd2f7c313e870cfe Mon Sep 17 00:00:00 2001 From: rogercloud Date: Sat, 5 Sep 2026 15:18:26 +0800 Subject: [PATCH 2/5] fix(compaction): preserve unrecoverable oversized messages --- docs/context-compaction-hardening.md | 64 ++++++++++ src/xagent/core/agent/context/execution.py | 74 ++++++++++- src/xagent/core/agent/runtime.py | 128 +++++++++++-------- tests/core/agent/test_auto.py | 24 +++- tests/core/agent/test_context.py | 135 ++++++++++++++++++++- tests/core/agent/test_react.py | 19 ++- tests/core/agent/test_runtime.py | 39 ++++++ 7 files changed, 414 insertions(+), 69 deletions(-) create mode 100644 docs/context-compaction-hardening.md diff --git a/docs/context-compaction-hardening.md b/docs/context-compaction-hardening.md new file mode 100644 index 000000000..17e634440 --- /dev/null +++ b/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. +- Prefer a model tokenizer over the current character heuristic where one is + available. +- 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. diff --git a/src/xagent/core/agent/context/execution.py b/src/xagent/core/agent/context/execution.py index f603f69d8..4bf5027c9 100644 --- a/src/xagent/core/agent/context/execution.py +++ b/src/xagent/core/agent/context/execution.py @@ -79,6 +79,13 @@ # 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 +# 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 @@ -1135,7 +1142,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 @@ -1150,11 +1159,34 @@ def build_llm_compact_request_if_needed(self) -> dict[str, Any] | None: max_tokens = self._llm_compact_max_tokens() omitted: list[dict[str, Any]] = [] messages = self._build_llm_compact_prompt(visible_messages, omitted=omitted) + request_input_tokens = sum( + max(1, len(str(message.get("content") or "")) // 4) for message in messages + ) metadata: dict[str, Any] = { "original_tokens": total_tokens, "threshold": self.compact_config.threshold, "max_summary_tokens": max_tokens, + "compact_request_input_tokens": request_input_tokens, } + if isinstance(context_window, int) and context_window > 0: + 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 @@ -1549,6 +1581,15 @@ def _compact_transcript( 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()}" @@ -1561,7 +1602,16 @@ def _compact_transcript( + json.dumps(message.tool_calls, ensure_ascii=False, default=str) ) content_tokens = max(1, len(message.content) // 4) - if content_tokens > max_message_tokens: + 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( @@ -1580,6 +1630,20 @@ def _compact_transcript( 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. @@ -1595,11 +1659,11 @@ def _omitted_content_notice(message: Message, content_tokens: int) -> str: 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 f"{message.role} message" + subject = f"{tool_name} result" if tool_name else "tool result" return ( f"[content omitted from this summary request: {subject}, " - f"~{content_tokens} tokens. It ran and produced output; re-read " - f"the source for its contents rather than reconstructing them.]" + f"~{content_tokens} tokens. Re-run the recorded read-only tool " + f"call if its contents are needed rather than reconstructing them.]" ) @staticmethod diff --git a/src/xagent/core/agent/runtime.py b/src/xagent/core/agent/runtime.py index 8ca102c07..da3417417 100644 --- a/src/xagent/core/agent/runtime.py +++ b/src/xagent/core/agent/runtime.py @@ -34,7 +34,7 @@ WAITING_FOR_USER_STATUS, tool_result_waits_for_user, ) -from .context.execution import COMPACT_SUMMARY_FALLBACK_BUDGETS +from .context.execution import COMPACT_SUMMARY_FALLBACK_BUDGETS, CompactResult from .result import normalize_tool_failure_code, tool_result_succeeded from .streaming import merge_streamed_tool_call_arguments @@ -1355,6 +1355,7 @@ async def compact_context_if_needed( # budget, so the compact trace event can say so alongside the other # degrade markers. compact_outcome: dict[str, Any] = {} + compaction_blocked = False if ( llm is not None and callable(getattr(llm, "chat", None)) @@ -1362,65 +1363,92 @@ async def compact_context_if_needed( and callable(compact_with_llm_response) ): summary_unavailable_metadata = None - request = llm_compact_request_if_needed() + context_window = getattr(llm, "context_window", None) + request = llm_compact_request_if_needed(context_window=context_window) if request is not None: request_metadata = request.get("metadata") or {} - llm_metadata = {**request_metadata, "purpose": "context_compaction"} - try: - await self.on_llm_start( - context=context, - messages=request["messages"], - metadata=llm_metadata, - ) - response = await self._run_compact_llm_call( - llm, - messages=request["messages"], - max_tokens=request["max_tokens"], - metadata=llm_metadata, - outcome=compact_outcome, - ) - await self.on_llm_end( - context=context, - response=response, - metadata=llm_metadata, + if request.get("blocked"): + compaction_blocked = True + message_count = len(getattr(context, "messages", ())) + result = CompactResult( + compacted=False, + original_count=message_count, + final_count=message_count, + strategy="none", + metadata={ + **request_metadata, + "fallback_suppressed": True, + }, ) - result = compact_with_llm_response( - response, - llm=llm, - original_tokens=request.get("original_tokens"), + logger.warning( + "Context compaction request cannot fit the compact " + "model window without discarding unrecoverable " + "messages; preserving the original context. " + "execution_id=%s", + getattr(context, "execution_id", None), ) - for key, value in request_metadata.items(): - result.metadata.setdefault(key, value) - result.metadata.update(compact_outcome) - if not getattr(result, "compacted", False): - logger.warning( - "Context compaction summary was unusable; falling " - "back to dropping messages. execution_id=%s", - getattr(context, "execution_id", None), + else: + llm_metadata = { + **request_metadata, + "purpose": "context_compaction", + } + try: + await self.on_llm_start( + context=context, + messages=request["messages"], + metadata=llm_metadata, ) - unusable_summary_metadata = { - "llm_summary_unusable": True, - **request_metadata, - } - except LLMCallInterrupted: - raise - except Exception as exc: # noqa: BLE001 - await self.on_llm_error( - context=context, - error=exc, - metadata=llm_metadata, - ) - if not callable(compact_if_needed): + response = await self._run_compact_llm_call( + llm, + messages=request["messages"], + max_tokens=request["max_tokens"], + metadata=llm_metadata, + outcome=compact_outcome, + ) + await self.on_llm_end( + context=context, + response=response, + metadata=llm_metadata, + ) + result = compact_with_llm_response( + response, + llm=llm, + original_tokens=request.get("original_tokens"), + ) + for key, value in request_metadata.items(): + result.metadata.setdefault(key, value) + result.metadata.update(compact_outcome) + if not getattr(result, "compacted", False): + logger.warning( + "Context compaction summary was unusable; falling " + "back to dropping messages. execution_id=%s", + getattr(context, "execution_id", None), + ) + unusable_summary_metadata = { + "llm_summary_unusable": True, + **request_metadata, + } + except LLMCallInterrupted: raise - result = compact_if_needed() - result.metadata["llm_compact_error"] = str(exc) - result.metadata["fallback_strategy"] = result.strategy - result.metadata.update(request_metadata) + except Exception as exc: # noqa: BLE001 + await self.on_llm_error( + context=context, + error=exc, + metadata=llm_metadata, + ) + if not callable(compact_if_needed): + raise + result = compact_if_needed() + result.metadata["llm_compact_error"] = str(exc) + result.metadata["fallback_strategy"] = result.strategy + result.metadata.update(request_metadata) # Backstop. ``compact_if_needed`` drops the oldest messages outright, # so it runs only after summarization was skipped, errored, or came # back unusable -- never as an alternative worth choosing. - if result is None or not getattr(result, "compacted", False): + if not compaction_blocked and ( + result is None or not getattr(result, "compacted", False) + ): if not callable(compact_if_needed): return result result = compact_if_needed() diff --git a/tests/core/agent/test_auto.py b/tests/core/agent/test_auto.py index feff9328e..2268547ff 100644 --- a/tests/core/agent/test_auto.py +++ b/tests/core/agent/test_auto.py @@ -2361,11 +2361,12 @@ async def stream_chat(self, messages: Any = None, **kwargs: Any) -> Any: def _auto_routing_router(downstream: Any, route_prompts: list[str]) -> RouterLLM: """A real ``RouterLLM`` with its selection stubbed to record the prompt. - ``context_window`` is set, as production always does via ``adapter.py``; - 4 gives a compaction threshold of 3, so any context compacts. + ``context_window`` is set, as production always does via ``adapter.py``. + The fixture uses a realistic 32k window and enough history below to trigger + compaction. """ router = RouterLLM(downstream_resolver=lambda _model_id: downstream) - router.context_window = 4 + router.context_window = 32_000 async def select_model(prompt: str) -> str: route_prompts.append(prompt) @@ -2394,7 +2395,22 @@ async def test_auto_summarizes_with_the_main_model_when_no_compact_model() -> No router = _auto_routing_router(downstream, route_prompts) context = ExecutionContext() context.add_user_message("hi") - context.add_tool_result("read_file", {"output": "x" * 200}, tool_call_id="call-1") + context.add_assistant_message( + "", + tool_calls=[ + { + "id": "call-1", + "type": "function", + "function": { + "name": "read_file", + "arguments": '{"path":"large.txt"}', + }, + } + ], + ) + context.add_tool_result( + "read_file", {"output": "x" * 120_000}, tool_call_id="call-1" + ) result = await AutoPattern().run( context=context, diff --git a/tests/core/agent/test_context.py b/tests/core/agent/test_context.py index 9026b4e5a..e64cfaad9 100644 --- a/tests/core/agent/test_context.py +++ b/tests/core/agent/test_context.py @@ -2007,6 +2007,19 @@ def test_compact_request_omits_a_message_too_large_to_read() -> None: context = ExecutionContext(execution_id="oversized") context.compact_config.threshold = 24000 context.add_user_message("summarize the repo") + context.add_assistant_message( + "", + tool_calls=[ + { + "id": "call-1", + "type": "function", + "function": { + "name": "read_file", + "arguments": '{"path":"large.txt"}', + }, + } + ], + ) context.add_tool_result( "read_file", {"output": "x" * 900_000}, tool_call_id="call-1" ) @@ -2028,12 +2041,111 @@ def test_compact_request_keeps_messages_under_the_cap_verbatim() -> None: whole, or every summary silently degrades.""" context = ExecutionContext(execution_id="ordinary") context.compact_config.threshold = 24000 - context.add_user_message("a question worth summarizing") - context.add_tool_result("read_file", {"output": "y" * 4_000}, tool_call_id="call-1") + for index in range(7): + context.add_user_message(f"ordinary-{index}:" + "y" * 16_000) request = context.build_llm_compact_request_if_needed() - assert request is None or "omitted_messages" not in request["metadata"] + assert request is not None + transcript = request["messages"][-1]["content"] + for index in range(7): + assert f"ordinary-{index}:" in transcript + assert "omitted_messages" not in request["metadata"] + + +def test_compact_request_never_omits_an_oversized_user_requirement() -> None: + context = ExecutionContext(execution_id="user-requirement") + context.compact_config.threshold = 24000 + marker = "ORIGINAL_REQUIREMENT_MUST_SURVIVE" + context.add_user_message(marker + ":" + "u" * 28_000) + context.add_assistant_message("a" * 70_000) + context.add_user_message("continue") + + request = context.build_llm_compact_request_if_needed(context_window=32_000) + + assert request is not None + assert not request.get("blocked") + assert marker in request["messages"][-1]["content"] + assert "omitted_messages" not in request["metadata"] + + +def test_compact_request_never_omits_an_unrecoverable_tool_result() -> None: + context = ExecutionContext(execution_id="write-receipt") + context.compact_config.threshold = 24000 + marker = "ONE_TIME_WRITE_RECEIPT" + context.add_user_message("create the remote resource") + context.add_assistant_message( + "", + tool_calls=[ + { + "id": "call-1", + "type": "function", + "function": { + "name": "create_resource", + "arguments": '{"name":"report"}', + }, + } + ], + ) + context.add_tool_result( + "create_resource", + {"output": marker + ":" + "r" * 120_000}, + tool_call_id="call-1", + ) + + request = context.build_llm_compact_request_if_needed(context_window=32_000) + + assert request is not None + assert not request.get("blocked") + assert marker in request["messages"][-1]["content"] + assert "omitted_messages" not in request["metadata"] + input_tokens = request["metadata"]["compact_request_input_tokens"] + safety_tokens = request["metadata"]["compact_request_safety_tokens"] + assert input_tokens + request["max_tokens"] + safety_tokens <= 32_000 + + +def test_compact_request_budgets_the_complete_rendered_prompt() -> None: + context = ExecutionContext(execution_id="complete-budget") + context.compact_config.threshold = 24000 + for index in range(5): + context.add_user_message(f"message-{index}:" + "x" * 19_000) + context.add_user_message("y" * 24_000) + + request = context.build_llm_compact_request_if_needed(context_window=32_000) + + assert request is not None + assert not request.get("blocked") + input_tokens = request["metadata"]["compact_request_input_tokens"] + safety_tokens = request["metadata"]["compact_request_safety_tokens"] + assert input_tokens + request["max_tokens"] + safety_tokens <= 32_000 + assert request["max_tokens"] < 6_000 + + +def test_compact_request_blocks_when_tool_calls_overflow_the_window() -> None: + context = ExecutionContext(execution_id="tool-call-budget") + context.compact_config.threshold = 24000 + for _ in range(6): + context.add_user_message("z" * 17_000) + context.add_assistant_message( + "", + tool_calls=[ + { + "id": "call-1", + "type": "function", + "function": { + "name": "large_call", + "arguments": "v" * 200_000, + }, + } + ], + ) + + request = context.build_llm_compact_request_if_needed(context_window=32_000) + + assert request is not None + assert request["blocked"] is True + assert request["metadata"]["llm_compact_request_too_large"] is True + assert request["max_tokens"] == 0 def test_oversized_content_is_replaced_whole_not_sliced() -> None: @@ -2046,8 +2158,21 @@ def test_oversized_content_is_replaced_whole_not_sliced() -> None: context = ExecutionContext(execution_id="whole-not-sliced") context.compact_config.threshold = 24000 context.add_user_message("go") + context.add_assistant_message( + "", + tool_calls=[ + { + "id": "call-1", + "type": "function", + "function": { + "name": "read_file", + "arguments": '{"path":"large.json"}', + }, + } + ], + ) context.add_tool_result( - "fetch_page", + "read_file", { "output": '{"handle": {"workspace": "4b33784773d5", "branch": "main"}}' * 5000 @@ -2061,7 +2186,7 @@ def test_oversized_content_is_replaced_whole_not_sliced() -> None: assert first is not None and second is not None transcript = first["messages"][-1]["content"] assert "4b33784773d5" not in transcript - assert "fetch_page result" in transcript + assert "read_file result" in transcript # Byte-identical across builds: nothing in the notice comes from the # clock or a request id. assert transcript == second["messages"][-1]["content"] diff --git a/tests/core/agent/test_react.py b/tests/core/agent/test_react.py index cd3739f8b..b9f893958 100644 --- a/tests/core/agent/test_react.py +++ b/tests/core/agent/test_react.py @@ -8626,15 +8626,15 @@ async def stream_chat(self, messages: Any = None, **kwargs: Any) -> Any: def _routing_router( - downstream: Any, route_prompts: list[str], *, context_window: int = 4 + downstream: Any, route_prompts: list[str], *, context_window: int = 32_000 ) -> RouterLLM: """A real ``RouterLLM`` whose selection is stubbed to record its prompt. ``context_window`` is deliberately set, matching production: ``adapter.py`` always stamps it from the model row, and ``prepare_llm_for_context`` recomputes the compaction threshold from it. Leaving it unset would put the - fixture in a state a real router never reaches. A window of 4 yields a - threshold of 3, small enough that any context compacts. + fixture in a state a real router never reaches. The fixture uses a realistic + 32k window and enough history below to trigger compaction. """ router = RouterLLM(downstream_resolver=lambda _model_id: downstream) router.context_window = context_window @@ -8653,10 +8653,19 @@ def _react_context_with_tool_history(execution_id: str) -> ExecutionContext: context.add_assistant_message( "", tool_calls=[ - {"id": "call-1", "type": "function", "function": {"name": "read_file"}}, + { + "id": "call-1", + "type": "function", + "function": { + "name": "read_file", + "arguments": '{"path":"large.txt"}', + }, + }, ], ) - context.add_tool_result("read_file", {"output": "x" * 200}, tool_call_id="call-1") + context.add_tool_result( + "read_file", {"output": "x" * 120_000}, tool_call_id="call-1" + ) return context diff --git a/tests/core/agent/test_runtime.py b/tests/core/agent/test_runtime.py index 7033c12c0..c9247e8de 100644 --- a/tests/core/agent/test_runtime.py +++ b/tests/core/agent/test_runtime.py @@ -1721,6 +1721,45 @@ async def chat(self, **_: Any) -> Any: return {"content": "what happened earlier"} +@pytest.mark.asyncio +async def test_compaction_does_not_truncate_an_unsendable_safe_request() -> None: + context = ExecutionContext(execution_id="unsafe-to-truncate") + context.compact_config.threshold = 24000 + context.compact_config.max_messages = 2 + for _ in range(6): + context.add_user_message("z" * 17_000) + context.add_assistant_message( + "", + tool_calls=[ + { + "id": "call-1", + "type": "function", + "function": { + "name": "large_call", + "arguments": "v" * 200_000, + }, + } + ], + ) + original_messages = list(context.messages) + + class UnusedLLM: + context_window = 32_000 + + async def chat(self, **_: Any) -> Any: # pragma: no cover - must not run + raise AssertionError("an oversized compact request must not be sent") + + result = await PatternRuntime().compact_context_if_needed( + context=context, + llm=UnusedLLM(), + ) + + assert not result.compacted + assert result.metadata["llm_compact_request_too_large"] is True + assert result.metadata["fallback_suppressed"] is True + assert context.messages == original_messages + + def _oversized_context(execution_id: str) -> ExecutionContext: context = ExecutionContext(execution_id=execution_id) context.compact_config.threshold = 32000 From c1dfec55969737b784aa020b532fc0d5da31f893 Mon Sep 17 00:00:00 2001 From: rogercloud Date: Sat, 5 Sep 2026 19:51:50 +0800 Subject: [PATCH 3/5] fix(compaction): preserve context on token overflow --- docs/context-compaction-hardening.md | 4 +- src/xagent/core/agent/context/execution.py | 72 ++++++++++++++++------ src/xagent/core/agent/runtime.py | 34 ++++++++-- src/xagent/core/model/chat/error.py | 24 ++++++++ tests/core/agent/test_auto.py | 1 + tests/core/agent/test_context.py | 62 ++++++++++++++----- tests/core/agent/test_dag.py | 1 + tests/core/agent/test_react.py | 5 +- tests/core/agent/test_runtime.py | 40 ++++++++++++ 9 files changed, 198 insertions(+), 45 deletions(-) diff --git a/docs/context-compaction-hardening.md b/docs/context-compaction-hardening.md index 17e634440..aee5a08aa 100644 --- a/docs/context-compaction-hardening.md +++ b/docs/context-compaction-hardening.md @@ -12,8 +12,8 @@ recovery pipeline. differ. - Count the complete provider request, including wrappers, tool calls, message content, context references, and output allowance. -- Prefer a model tokenizer over the current character heuristic where one is - available. +- 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 diff --git a/src/xagent/core/agent/context/execution.py b/src/xagent/core/agent/context/execution.py index 4bf5027c9..d24efef4e 100644 --- a/src/xagent/core/agent/context/execution.py +++ b/src/xagent/core/agent/context/execution.py @@ -9,6 +9,8 @@ from uuid import uuid4 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError +import tiktoken + from ...context_ref import ( CONTEXT_REFS_KEY, ContextReference, @@ -80,6 +82,8 @@ # everything to nothing. COMPACT_TRANSCRIPT_MESSAGE_MIN_TOKENS = 2048 COMPACT_REQUEST_SAFETY_TOKENS = 512 +COMPACT_TOKEN_ENCODING = tiktoken.get_encoding("cl100k_base") +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 @@ -115,6 +119,23 @@ COMPACT_DROPPED_TOOL_NOTICE_MAX_CHARS = 1024 COMPACT_DROPPED_TOOL_NAME_MAX_CHARS = 64 + +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. + return sum( + len( + COMPACT_TOKEN_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} @@ -1160,33 +1181,44 @@ def build_llm_compact_request_if_needed( omitted: list[dict[str, Any]] = [] messages = self._build_llm_compact_prompt(visible_messages, omitted=omitted) request_input_tokens = sum( - max(1, len(str(message.get("content") or "")) // 4) for message in messages + max(1, _count_compact_request_tokens(str(message.get("content") or ""))) + for message in messages ) metadata: dict[str, Any] = { "original_tokens": total_tokens, "threshold": self.compact_config.threshold, "max_summary_tokens": max_tokens, "compact_request_input_tokens": request_input_tokens, + "compact_request_tokenizer": "cl100k_base", } - if isinstance(context_window, int) and context_window > 0: - 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 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, + } + + 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 diff --git a/src/xagent/core/agent/runtime.py b/src/xagent/core/agent/runtime.py index da3417417..d00a8c03f 100644 --- a/src/xagent/core/agent/runtime.py +++ b/src/xagent/core/agent/runtime.py @@ -21,7 +21,7 @@ materialize_llm_kwargs, ) from ..model.chat.basic.base import BaseLLM -from ..model.chat.error import retry_on +from ..model.chat.error import is_context_length_error, retry_on from ..model.chat.exceptions import LLMToolProtocolError from ..model.chat.token_context import extract_cached_input_tokens from ..model.chat.tool_protocol import TOOL_PROTOCOL_ERROR_KEY @@ -1276,6 +1276,8 @@ async def _run_compact_llm_call( except LLMCallInterrupted: raise except Exception as exc: # noqa: BLE001 + if is_context_length_error(exc): + raise budgets = [ budget for budget in COMPACT_SUMMARY_FALLBACK_BUDGETS @@ -1309,7 +1311,11 @@ async def _run_compact_llm_call( raise except Exception as retry_exc: # noqa: BLE001 exc = retry_exc - if retry_on(retry_exc) or self._interrupt_requested: + if ( + is_context_length_error(retry_exc) + or retry_on(retry_exc) + or self._interrupt_requested + ): break continue metadata["compact_budget_reduced_to"] = budget @@ -1438,10 +1444,26 @@ async def compact_context_if_needed( ) if not callable(compact_if_needed): raise - result = compact_if_needed() - result.metadata["llm_compact_error"] = str(exc) - result.metadata["fallback_strategy"] = result.strategy - result.metadata.update(request_metadata) + if is_context_length_error(exc): + compaction_blocked = True + message_count = len(getattr(context, "messages", ())) + result = CompactResult( + compacted=False, + original_count=message_count, + final_count=message_count, + strategy="none", + metadata={ + **request_metadata, + "llm_compact_error": str(exc), + "llm_compact_context_length_error": True, + "fallback_suppressed": True, + }, + ) + else: + result = compact_if_needed() + result.metadata["llm_compact_error"] = str(exc) + result.metadata["fallback_strategy"] = result.strategy + result.metadata.update(request_metadata) # Backstop. ``compact_if_needed`` drops the oldest messages outright, # so it runs only after summarization was skipped, errored, or came diff --git a/src/xagent/core/model/chat/error.py b/src/xagent/core/model/chat/error.py index 5e7d2e874..519fa24ff 100644 --- a/src/xagent/core/model/chat/error.py +++ b/src/xagent/core/model/chat/error.py @@ -9,6 +9,30 @@ ZaiAPIStatusError = None +_CONTEXT_LENGTH_ERROR_MARKERS = ( + "context_length_exceeded", + "context length exceeded", + "maximum context length", + "exceeds the context window", + "input is too long", + "prompt is too long", + "too many input tokens", +) + + +def is_context_length_error(error: BaseException) -> bool: + """Recognize provider context-window failures through wrapper exceptions.""" + seen: set[int] = set() + current: BaseException | None = error + while current is not None and id(current) not in seen: + seen.add(id(current)) + message = str(current).lower() + if any(marker in message for marker in _CONTEXT_LENGTH_ERROR_MARKERS): + return True + current = current.__cause__ or current.__context__ + return False + + def retry_on(e: Exception) -> bool: ERRORS = ( httpx.TimeoutException, diff --git a/tests/core/agent/test_auto.py b/tests/core/agent/test_auto.py index 2268547ff..440af08b6 100644 --- a/tests/core/agent/test_auto.py +++ b/tests/core/agent/test_auto.py @@ -1314,6 +1314,7 @@ async def test_auto_pattern_falls_back_to_the_main_llm_for_compaction() -> None: "react done", ] ) + llm.context_window = 32_000 pattern = AutoPattern() context = ExecutionContext() context.compact_config.threshold = 1 diff --git a/tests/core/agent/test_context.py b/tests/core/agent/test_context.py index e64cfaad9..57bd09cc1 100644 --- a/tests/core/agent/test_context.py +++ b/tests/core/agent/test_context.py @@ -951,7 +951,7 @@ class CompactLLM: ctx.add_tool_result("read_file", {"output": "x" * 200}, tool_call_id="call-1") llm = CompactLLM() - request = ctx.build_llm_compact_request_if_needed() + request = ctx.build_llm_compact_request_if_needed(context_window=32_000) assert request is not None assert request["max_tokens"] == 256 prompt = request["messages"] @@ -1342,7 +1342,7 @@ def test_compact_with_llm_preserves_waiting_for_user_response() -> None: }, ) - request = ctx.build_llm_compact_request_if_needed() + request = ctx.build_llm_compact_request_if_needed(context_window=32_000) assert request is not None result = ctx.compact_with_llm_response( @@ -1433,7 +1433,7 @@ class CompactLLM: assert read_file_message.metadata["raw_result"] == {"output": "kpi table"} llm = CompactLLM() - request = ctx.build_llm_compact_request_if_needed() + request = ctx.build_llm_compact_request_if_needed(context_window=32_000) assert request is not None result = ctx.compact_with_llm_response( @@ -1909,7 +1909,9 @@ def test_llm_compact_budget_scales_with_the_threshold() -> None: The old ceiling of 1024 bound at every realistic window, leaving a reasoning model no room -- its reasoning comes out of this same allowance. """ - request = _context_over_threshold(20_000).build_llm_compact_request_if_needed() + request = _context_over_threshold(20_000).build_llm_compact_request_if_needed( + context_window=32_000 + ) assert request is not None assert request["max_tokens"] == 5_000 @@ -1926,7 +1928,7 @@ def test_llm_compact_budget_stays_under_provider_output_limits() -> None: for window_threshold in (96_000, 150_000, 750_000): request = _context_over_threshold( window_threshold - ).build_llm_compact_request_if_needed() + ).build_llm_compact_request_if_needed(context_window=window_threshold * 2) assert request is not None assert request["max_tokens"] == 8192 @@ -2024,7 +2026,7 @@ def test_compact_request_omits_a_message_too_large_to_read() -> None: "read_file", {"output": "x" * 900_000}, tool_call_id="call-1" ) - request = context.build_llm_compact_request_if_needed() + request = context.build_llm_compact_request_if_needed(context_window=32_000) assert request is not None assert request["original_tokens"] > 200_000 @@ -2044,7 +2046,7 @@ def test_compact_request_keeps_messages_under_the_cap_verbatim() -> None: for index in range(7): context.add_user_message(f"ordinary-{index}:" + "y" * 16_000) - request = context.build_llm_compact_request_if_needed() + request = context.build_llm_compact_request_if_needed(context_window=32_000) assert request is not None transcript = request["messages"][-1]["content"] @@ -2069,7 +2071,7 @@ def test_compact_request_never_omits_an_oversized_user_requirement() -> None: assert "omitted_messages" not in request["metadata"] -def test_compact_request_never_omits_an_unrecoverable_tool_result() -> None: +def test_compact_request_preserves_an_unrecoverable_tool_result_when_blocked() -> None: context = ExecutionContext(execution_id="write-receipt") context.compact_config.threshold = 24000 marker = "ONE_TIME_WRITE_RECEIPT" @@ -2096,20 +2098,18 @@ def test_compact_request_never_omits_an_unrecoverable_tool_result() -> None: request = context.build_llm_compact_request_if_needed(context_window=32_000) assert request is not None - assert not request.get("blocked") + assert request["blocked"] is True assert marker in request["messages"][-1]["content"] assert "omitted_messages" not in request["metadata"] - input_tokens = request["metadata"]["compact_request_input_tokens"] - safety_tokens = request["metadata"]["compact_request_safety_tokens"] - assert input_tokens + request["max_tokens"] + safety_tokens <= 32_000 + assert request["metadata"]["llm_compact_request_too_large"] is True def test_compact_request_budgets_the_complete_rendered_prompt() -> None: context = ExecutionContext(execution_id="complete-budget") context.compact_config.threshold = 24000 for index in range(5): - context.add_user_message(f"message-{index}:" + "x" * 19_000) - context.add_user_message("y" * 24_000) + context.add_user_message(f"message-{index}:" + "word " * 4_500) + context.add_user_message("tail " * 5_000) request = context.build_llm_compact_request_if_needed(context_window=32_000) @@ -2121,6 +2121,36 @@ def test_compact_request_budgets_the_complete_rendered_prompt() -> None: assert request["max_tokens"] < 6_000 +def test_compact_request_counts_cjk_tokens_before_sending() -> None: + context = ExecutionContext(execution_id="cjk-budget") + context.compact_config.threshold = 24_000 + marker = "重要约束必须保留" + context.add_user_message(marker + "重要约束" * 25_000) + context.add_assistant_message("继续处理") + + request = context.build_llm_compact_request_if_needed(context_window=32_000) + + assert request is not None + assert request["blocked"] is True + assert request["metadata"]["compact_request_input_tokens"] > 32_000 + assert request["metadata"]["compact_request_tokenizer"] == "cl100k_base" + assert marker in request["messages"][-1]["content"] + + +def test_compact_request_blocks_when_context_window_is_unknown() -> None: + context = ExecutionContext(execution_id="unknown-window") + context.compact_config.threshold = 1 + context.add_user_message("requirement that must survive") + context.add_assistant_message("work in progress") + + request = context.build_llm_compact_request_if_needed() + + assert request is not None + assert request["blocked"] is True + assert request["metadata"]["llm_compact_context_window_unknown"] is True + assert request["max_tokens"] == 0 + + def test_compact_request_blocks_when_tool_calls_overflow_the_window() -> None: context = ExecutionContext(execution_id="tool-call-budget") context.compact_config.threshold = 24000 @@ -2180,8 +2210,8 @@ def test_oversized_content_is_replaced_whole_not_sliced() -> None: tool_call_id="call-1", ) - first = context.build_llm_compact_request_if_needed() - second = context.build_llm_compact_request_if_needed() + first = context.build_llm_compact_request_if_needed(context_window=32_000) + second = context.build_llm_compact_request_if_needed(context_window=32_000) assert first is not None and second is not None transcript = first["messages"][-1]["content"] diff --git a/tests/core/agent/test_dag.py b/tests/core/agent/test_dag.py index a7482e08b..36b016b3e 100644 --- a/tests/core/agent/test_dag.py +++ b/tests/core/agent/test_dag.py @@ -967,6 +967,7 @@ async def test_dag_pattern_returns_terminal_step_result_as_output() -> None: async def test_dag_pattern_passes_compact_llm_to_step_react_compaction() -> None: llm = SequenceLLM([{"content": "step done", "done": True}]) compact_llm = SequenceLLM([{"content": "compacted dag step context"}]) + compact_llm.context_window = 32_000 plan = build_plan(PlanStep(id="answer", task="Answer with DAG")) pattern = DAGPattern(lambda **_: plan) context = ExecutionContext(execution_id="dag-step-compact-llm") diff --git a/tests/core/agent/test_react.py b/tests/core/agent/test_react.py index b9f893958..f509bbc0c 100644 --- a/tests/core/agent/test_react.py +++ b/tests/core/agent/test_react.py @@ -5696,6 +5696,8 @@ async def test_react_pattern_traces_context_compaction() -> None: context.compact_config.threshold = 1 for index in range(3): context.add_user_message(f"message {index}") + llm = FakeLLM([{"content": "summary"}, {"content": "done"}]) + llm.context_window = 32_000 result = await ReActPattern(max_iterations=1).run( context=context, @@ -5703,7 +5705,7 @@ async def test_react_pattern_traces_context_compaction() -> None: # Two responses: compaction now summarizes with the main model when no # compact model is configured, so it consumes one before the turn's # own call. - llm=FakeLLM([{"content": "summary"}, {"content": "done"}]), + llm=llm, runtime=runtime, ) @@ -5743,6 +5745,7 @@ async def test_react_pattern_uses_compact_llm_for_context_compaction() -> None: } ] ) + compact_llm.context_window = 32_000 result = await ReActPattern(max_iterations=1).run( context=context, diff --git a/tests/core/agent/test_runtime.py b/tests/core/agent/test_runtime.py index c9247e8de..a96e70d56 100644 --- a/tests/core/agent/test_runtime.py +++ b/tests/core/agent/test_runtime.py @@ -1317,6 +1317,7 @@ class RaisingCompactLLM: fallback in ``PatternRuntime.compact_context_if_needed``.""" model_name = "raising-compact-llm" + context_window = 32_000 async def chat(self, **_: Any) -> Any: raise RuntimeError("compact llm exploded") @@ -1459,6 +1460,7 @@ async def trace_event( class EmptySummaryLLM: model_name = "compact-test" + context_window = 32_000 async def chat(self, **_: Any) -> Any: return {"content": ""} @@ -1501,6 +1503,7 @@ async def test_compaction_retries_with_a_smaller_budget_before_truncating() -> N class OutputCappedLLM: model_name = "compact-test" + context_window = 64_000 def __init__(self) -> None: self.budgets: list[int] = [] @@ -1547,6 +1550,7 @@ async def test_compaction_ladder_skips_a_budget_the_model_cannot_use() -> None: class CappedReasoningLLM: model_name = "compact-test" + context_window = 64_000 def __init__(self) -> None: self.budgets: list[int] = [] @@ -1596,6 +1600,7 @@ async def test_compaction_stops_descending_once_a_budget_is_accepted() -> None: class AlwaysReasoningLLM: model_name = "compact-test" + context_window = 64_000 def __init__(self) -> None: self.budgets: list[int] = [] @@ -1694,6 +1699,7 @@ async def test_compaction_does_not_blame_the_model_when_nothing_is_summarizable( class UnusedCompactLLM: model_name = "compact-test" + context_window = 32_000 async def chat(self, **_: Any) -> Any: # pragma: no cover - never called raise AssertionError("no request should have been built") @@ -1716,6 +1722,7 @@ async def chat(self, **_: Any) -> Any: # pragma: no cover - never called class _SummarizingLLM: model_name = "compact-test" + context_window = 32_000 async def chat(self, **_: Any) -> Any: return {"content": "what happened earlier"} @@ -1760,6 +1767,39 @@ async def chat(self, **_: Any) -> Any: # pragma: no cover - must not run assert context.messages == original_messages +@pytest.mark.asyncio +async def test_compaction_preserves_context_after_provider_rejects_its_length() -> None: + context = ExecutionContext(execution_id="provider-context-rejection") + context.compact_config.threshold = 1 + context.compact_config.max_messages = 2 + marker = "ORIGINAL_REQUIREMENT_MUST_SURVIVE" + context.add_user_message(marker) + for index in range(4): + context.add_assistant_message(f"work-{index}") + original_messages = list(context.messages) + + class RejectingLLM: + context_window = 32_000 + + def __init__(self) -> None: + self.budgets: list[int] = [] + + async def chat(self, **kwargs: Any) -> Any: + self.budgets.append(kwargs["max_tokens"]) + raise RuntimeError("maximum context length exceeded") + + llm = RejectingLLM() + result = await PatternRuntime().compact_context_if_needed(context=context, llm=llm) + + assert llm.budgets == [256] + assert not result.compacted + assert result.strategy == "none" + assert result.metadata["llm_compact_context_length_error"] is True + assert result.metadata["fallback_suppressed"] is True + assert context.messages == original_messages + assert marker in context.messages[0].content + + def _oversized_context(execution_id: str) -> ExecutionContext: context = ExecutionContext(execution_id=execution_id) context.compact_config.threshold = 32000 From 2ad6ff57b95059dbc712ccc5fde22bf648d995fb Mon Sep 17 00:00:00 2001 From: rogercloud Date: Sat, 5 Sep 2026 21:01:54 +0800 Subject: [PATCH 4/5] fix(compaction): fail closed on provider overflow --- src/xagent/core/agent/context/execution.py | 42 +++++++++++++++---- src/xagent/core/agent/runtime.py | 8 +++- src/xagent/core/model/chat/basic/gemini.py | 8 ++++ src/xagent/core/model/chat/error.py | 12 +++++- src/xagent/core/model/chat/exceptions.py | 6 +++ tests/core/agent/test_context.py | 28 +++++++++++++ tests/core/agent/test_runtime.py | 5 ++- .../core/model/chat/basic/test_gemini_sdk.py | 34 +++++++++++++++ 8 files changed, 132 insertions(+), 11 deletions(-) diff --git a/src/xagent/core/agent/context/execution.py b/src/xagent/core/agent/context/execution.py index d24efef4e..b17d72b40 100644 --- a/src/xagent/core/agent/context/execution.py +++ b/src/xagent/core/agent/context/execution.py @@ -5,6 +5,7 @@ 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 @@ -82,7 +83,6 @@ # everything to nothing. COMPACT_TRANSCRIPT_MESSAGE_MIN_TOKENS = 2048 COMPACT_REQUEST_SAFETY_TOKENS = 512 -COMPACT_TOKEN_ENCODING = tiktoken.get_encoding("cl100k_base") 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 @@ -120,14 +120,22 @@ 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( - COMPACT_TOKEN_ENCODING.encode( + encoding.encode( content[offset : offset + COMPACT_TOKEN_COUNT_CHUNK_CHARS], disallowed_special=(), ) @@ -1180,16 +1188,10 @@ def build_llm_compact_request_if_needed( max_tokens = self._llm_compact_max_tokens() omitted: list[dict[str, Any]] = [] messages = self._build_llm_compact_prompt(visible_messages, omitted=omitted) - request_input_tokens = sum( - max(1, _count_compact_request_tokens(str(message.get("content") or ""))) - for message in messages - ) metadata: dict[str, Any] = { "original_tokens": total_tokens, "threshold": self.compact_config.threshold, "max_summary_tokens": max_tokens, - "compact_request_input_tokens": request_input_tokens, - "compact_request_tokenizer": "cl100k_base", } if not isinstance(context_window, int) or context_window <= 0: metadata["llm_compact_context_window_unknown"] = True @@ -1200,6 +1202,30 @@ def build_llm_compact_request_if_needed( "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 diff --git a/src/xagent/core/agent/runtime.py b/src/xagent/core/agent/runtime.py index d00a8c03f..a76cda09c 100644 --- a/src/xagent/core/agent/runtime.py +++ b/src/xagent/core/agent/runtime.py @@ -22,7 +22,7 @@ ) from ..model.chat.basic.base import BaseLLM from ..model.chat.error import is_context_length_error, retry_on -from ..model.chat.exceptions import LLMToolProtocolError +from ..model.chat.exceptions import LLMContextLengthError, LLMToolProtocolError from ..model.chat.token_context import extract_cached_input_tokens from ..model.chat.tool_protocol import TOOL_PROTOCOL_ERROR_KEY from ..model.chat.types import ChunkType @@ -290,6 +290,12 @@ async def run_llm_call(self, llm: Any, **kwargs: Any) -> Any: self.interrupt_reason or "interrupted during LLM call" ) from exc raise + except Exception as exc: # noqa: BLE001 + if is_context_length_error(exc): + if isinstance(exc, LLMContextLengthError): + raise + raise LLMContextLengthError(str(exc)) from exc + raise finally: self._active_llm_tasks.discard(task) diff --git a/src/xagent/core/model/chat/basic/gemini.py b/src/xagent/core/model/chat/basic/gemini.py index 54b21adef..1bafe19a1 100644 --- a/src/xagent/core/model/chat/basic/gemini.py +++ b/src/xagent/core/model/chat/basic/gemini.py @@ -10,7 +10,9 @@ from google.genai import errors as genai_errors from ....utils.security import redact_sensitive_text +from ..error import is_context_length_error from ..exceptions import ( + LLMContextLengthError, LLMEmptyContentError, LLMInvalidResponseError, LLMRetryableError, @@ -655,6 +657,9 @@ async def chat( except Exception as e: logger.error("Gemini SDK API error: %s", redact_sensitive_text(str(e))) + if is_context_length_error(e): + raise LLMContextLengthError(str(e)) from e + error_text = str(e) error_text_lower = error_text.lower() @@ -945,6 +950,9 @@ async def stream_chat( "Gemini SDK streaming error: %s", redact_sensitive_text(str(e)) ) + if is_context_length_error(e): + raise LLMContextLengthError(str(e)) from e + error_text = str(e) error_text_lower = error_text.lower() diff --git a/src/xagent/core/model/chat/error.py b/src/xagent/core/model/chat/error.py index 519fa24ff..4f5dac9ad 100644 --- a/src/xagent/core/model/chat/error.py +++ b/src/xagent/core/model/chat/error.py @@ -1,7 +1,11 @@ import httpx import openai -from .exceptions import LLMRetryableError, LLMToolProtocolError +from .exceptions import ( + LLMContextLengthError, + LLMRetryableError, + LLMToolProtocolError, +) try: from zai.core._errors import APIStatusError as ZaiAPIStatusError # type: ignore @@ -14,6 +18,7 @@ "context length exceeded", "maximum context length", "exceeds the context window", + "exceeds the maximum number of tokens allowed", "input is too long", "prompt is too long", "too many input tokens", @@ -26,6 +31,8 @@ def is_context_length_error(error: BaseException) -> bool: current: BaseException | None = error while current is not None and id(current) not in seen: seen.add(id(current)) + if isinstance(current, LLMContextLengthError): + return True message = str(current).lower() if any(marker in message for marker in _CONTEXT_LENGTH_ERROR_MARKERS): return True @@ -34,6 +41,9 @@ def is_context_length_error(error: BaseException) -> bool: def retry_on(e: Exception) -> bool: + if is_context_length_error(e): + return False + ERRORS = ( httpx.TimeoutException, httpx.NetworkError, diff --git a/src/xagent/core/model/chat/exceptions.py b/src/xagent/core/model/chat/exceptions.py index e398dd5dd..c7fc87f5b 100644 --- a/src/xagent/core/model/chat/exceptions.py +++ b/src/xagent/core/model/chat/exceptions.py @@ -3,6 +3,12 @@ from typing import Any +class LLMContextLengthError(RuntimeError): + """The provider rejected an input because it exceeded the model window.""" + + pass + + class LLMRetryableError(RuntimeError): """Base exception for LLM errors that should trigger retry. diff --git a/tests/core/agent/test_context.py b/tests/core/agent/test_context.py index 57bd09cc1..22608968c 100644 --- a/tests/core/agent/test_context.py +++ b/tests/core/agent/test_context.py @@ -13,6 +13,7 @@ MergeStrategy, Message, ) +from xagent.core.agent.context import execution as execution_module from xagent.core.agent.context import enrichment as enrichment_module from xagent.core.agent.context.enrichment import ( MEMORY_CONTEXT_METADATA_KEY, @@ -2151,6 +2152,33 @@ def test_compact_request_blocks_when_context_window_is_unknown() -> None: assert request["max_tokens"] == 0 +def test_compact_request_blocks_when_tokenizer_cannot_load( + monkeypatch: pytest.MonkeyPatch, +) -> None: + context = ExecutionContext(execution_id="offline-tokenizer") + context.compact_config.threshold = 1 + context.add_user_message("requirement that must survive") + original_messages = list(context.messages) + + execution_module._compact_token_encoding.cache_clear() + + def fail_to_load(_: str) -> None: + raise OSError("offline") + + monkeypatch.setattr(execution_module.tiktoken, "get_encoding", fail_to_load) + try: + request = context.build_llm_compact_request_if_needed(context_window=32_000) + finally: + execution_module._compact_token_encoding.cache_clear() + + assert request is not None + assert request["blocked"] is True + assert request["metadata"]["llm_compact_tokenizer_unavailable"] is True + assert request["metadata"]["compact_tokenizer_error_type"] == "OSError" + assert request["max_tokens"] == 0 + assert context.messages == original_messages + + def test_compact_request_blocks_when_tool_calls_overflow_the_window() -> None: context = ExecutionContext(execution_id="tool-call-budget") context.compact_config.threshold = 24000 diff --git a/tests/core/agent/test_runtime.py b/tests/core/agent/test_runtime.py index a96e70d56..a16e8e5eb 100644 --- a/tests/core/agent/test_runtime.py +++ b/tests/core/agent/test_runtime.py @@ -1786,7 +1786,10 @@ def __init__(self) -> None: async def chat(self, **kwargs: Any) -> Any: self.budgets.append(kwargs["max_tokens"]) - raise RuntimeError("maximum context length exceeded") + raise RuntimeError( + "The input token count (461428) exceeds the maximum number " + "of tokens allowed (131072)." + ) llm = RejectingLLM() result = await PatternRuntime().compact_context_if_needed(context=context, llm=llm) diff --git a/tests/core/model/chat/basic/test_gemini_sdk.py b/tests/core/model/chat/basic/test_gemini_sdk.py index b0bb8b018..48feb1817 100644 --- a/tests/core/model/chat/basic/test_gemini_sdk.py +++ b/tests/core/model/chat/basic/test_gemini_sdk.py @@ -674,6 +674,40 @@ async def mock_generate_content_error(*args, **kwargs): print("✅ 500 server error correctly caught as retryable") + @pytest.mark.asyncio + async def test_context_overflow_uses_typed_non_retryable_error( + self, llm: GeminiLLM, mocker: pytest_mock.MockerFixture + ) -> None: + from google.genai import errors as genai_errors + + from xagent.core.model.chat.error import retry_on + from xagent.core.model.chat.exceptions import LLMContextLengthError + + message = ( + "The input token count (461428) exceeds the maximum number of " + "tokens allowed (131072)." + ) + mock_response = mocker.MagicMock() + mock_response.status_code = 400 + mock_response.json.return_value = {"error": {"code": 400, "message": message}} + mock_client = mocker.MagicMock() + + async def mock_generate_content_error(*args: Any, **kwargs: Any) -> None: + raise genai_errors.ClientError( + code=400, + response_json={"error": {"code": 400, "message": message}}, + response=mock_response, + ) + + mock_client.aio.models.generate_content = mock_generate_content_error + mocker.patch("google.genai.Client", return_value=mock_client) + + with pytest.raises(LLMContextLengthError) as exc_info: + await llm.chat([{"role": "user", "content": "Test"}]) + + assert message in str(exc_info.value) + assert retry_on(exc_info.value) is False + @pytest.mark.asyncio async def test_504_deadline_exceeded_error_is_retryable( self, llm: GeminiLLM, mocker: pytest_mock.MockerFixture From 54587a5f9d171776d404ac153978c90dbc2d0c69 Mon Sep 17 00:00:00 2001 From: rogercloud Date: Sat, 5 Sep 2026 21:19:55 +0800 Subject: [PATCH 5/5] style: sort compaction test imports --- tests/core/agent/test_context.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/core/agent/test_context.py b/tests/core/agent/test_context.py index 22608968c..232fc5586 100644 --- a/tests/core/agent/test_context.py +++ b/tests/core/agent/test_context.py @@ -13,8 +13,8 @@ MergeStrategy, Message, ) -from xagent.core.agent.context import execution as execution_module from xagent.core.agent.context import enrichment as enrichment_module +from xagent.core.agent.context import execution as execution_module from xagent.core.agent.context.enrichment import ( MEMORY_CONTEXT_METADATA_KEY, SKILL_CONTEXT_METADATA_KEY,