enh: optimize routed agent context usage

src/xagent/core/agent/pattern/auto/auto.py

@@ -24,7 +24,12 @@
     normalize_response_language_label,
     output_language_policy,
 )
-from ...runtime import LLMCallInterrupted, PatternRuntime
+from ...runtime import (
+    LLMCallInterrupted,
+    PatternRuntime,
+    prepare_llm_for_context,
+    resolved_llm_metadata,
+)
 from ..base import (
     REQUIRED_TOOL_CALL_FAILURE_REASON,
     AgentPattern,
@@ -754,6 +759,11 @@ async def _decide(
         # Re-derive the request-scoped language before routing so stale metadata
         # cannot bias the current decision prompt.
         self._clear_response_language(context)
+        call_llm = await prepare_llm_for_context(
+            llm=llm,
+            messages=context.get_messages_for_llm(),
+            context=context,
+        )
         await runtime.compact_context_if_needed(
             context=context,
             llm=compact_llm,
@@ -783,7 +793,10 @@ async def _decide(
                     content=retry_feedback,
                     section_title="Auto routing retry feedback",
                 )
-            metadata: dict[str, Any] = {"phase": "auto_decision"}
+            metadata: dict[str, Any] = {
+                "phase": "auto_decision",
+                **resolved_llm_metadata(call_llm),
+            }
             if attempt:
                 metadata["attempt"] = attempt + 1
             await runtime.on_llm_start(
@@ -806,7 +819,7 @@ async def _decide(
             )
             try:
                 response = await runtime.run_streaming_llm_call(
-                    llm,
+                    call_llm,
                     messages=messages,
                     tools=decision_tools,
                     tool_choice="required",

src/xagent/core/agent/pattern/react/react.py

@@ -47,7 +47,12 @@
 )
 from ...language import final_answer_language_rule
 from ...result import tool_result_succeeded, unwrap_final_answer_content
-from ...runtime import LLMCallInterrupted, PatternRuntime
+from ...runtime import (
+    LLMCallInterrupted,
+    PatternRuntime,
+    prepare_llm_for_context,
+    resolved_llm_metadata,
+)
 from ..base import AgentPattern, PatternResult, truncate_prompt_preview
 from ..final_answer_stream import ReActFinalAnswerStreamer
 
@@ -364,6 +369,21 @@ async def _run_tool_calling_loop(
             if interrupted is not None:
                 return interrupted
 
+            route_messages = self._messages_for_llm(
+                context,
+                has_tools=bool(tool_schemas),
+                force_final_answer=force_final_answer_now,
+                tool_names=self._schema_tool_names(tool_schemas),
+            )
+            call_llm = await prepare_llm_for_context(
+                llm=llm,
+                messages=route_messages,
+                context=context,
+            )
+            llm_metadata = {
+                "iteration": iteration,
+                **resolved_llm_metadata(call_llm),
+            }
             await runtime.compact_context_if_needed(
                 context=context,
                 llm=compact_llm,
@@ -381,7 +401,7 @@ async def _run_tool_calling_loop(
                 context=context,
                 messages=messages,
                 tools=tool_schemas or None,
-                metadata={"iteration": iteration},
+                metadata=llm_metadata,
             )
             answer_streamer: ReActFinalAnswerStreamer | None = None
             try:
@@ -400,12 +420,12 @@ async def _run_tool_calling_loop(
                 if tool_schemas:
                     answer_streamer = ReActFinalAnswerStreamer(runtime)
                     response = await runtime.run_streaming_llm_call(
-                        llm,
+                        call_llm,
                         on_chunk=answer_streamer.handle_chunk,
                         **llm_kwargs,
                     )
                 else:
-                    response = await runtime.stream_final_answer(llm, **llm_kwargs)
+                    response = await runtime.stream_final_answer(call_llm, **llm_kwargs)
             except LLMCallInterrupted:
                 if answer_streamer is not None:
                     await answer_streamer.fail("interrupted during LLM stream")
@@ -428,7 +448,7 @@ async def _run_tool_calling_loop(
                 normalized,
                 force_final_answer=force_final_answer_now,
             )
-            end_metadata: dict[str, Any] = {"iteration": iteration}
+            end_metadata: dict[str, Any] = dict(llm_metadata)
             if requires_protocol_retry:
                 end_metadata.update(
                     success=False,
@@ -448,7 +468,7 @@ async def _run_tool_calling_loop(
                         answer_streamer,
                     ) = await self._retry_tool_protocol_response(
                         context=context,
-                        llm=llm,
+                        llm=call_llm,
                         runtime=runtime,
                         iteration=iteration,
                         tool_schemas=tool_schemas,
@@ -681,6 +701,7 @@ async def _retry_tool_protocol_response(
         metadata = {
             "iteration": iteration,
             "phase": "tool_protocol_retry",
+            **resolved_llm_metadata(llm),
         }
         await runtime.checkpoint(
             "tool_protocol_retry",
@@ -1085,8 +1106,8 @@ def _coerce_arguments(self, arguments: Any) -> dict[str, Any]:
 
     def _build_tool_schema(self, tool: Any) -> dict[str, Any]:
         name = self._tool_name(tool)
-        description = self._tool_description(tool)
-        schema = self._tool_json_schema(tool)
+        description = self._compact_tool_description(self._tool_description(tool))
+        schema = self._compact_tool_json_schema(self._tool_json_schema(tool))
         return {
             "type": "function",
             "function": {
@@ -1096,6 +1117,44 @@ def _build_tool_schema(self, tool: Any) -> dict[str, Any]:
             },
         }
 
+    def _compact_tool_description(self, description: str) -> str:
+        """Trim redundant whitespace while preserving instructional structure."""
+        compacted_lines: list[str] = []
+        for line in description.splitlines():
+            compacted = " ".join(line.split())
+            if compacted:
+                compacted_lines.append(compacted)
+            elif compacted_lines and compacted_lines[-1]:
+                compacted_lines.append("")
+        while compacted_lines and not compacted_lines[-1]:
+            compacted_lines.pop()
+        return "\n".join(compacted_lines)
+
+    def _compact_tool_json_schema(
+        self, value: Any, *, named_schema_mapping: bool = False
+    ) -> Any:
+        """Remove presentation-only Pydantic metadata from provider schemas."""
+        if isinstance(value, list):
+            return [self._compact_tool_json_schema(item) for item in value]
+        if not isinstance(value, dict):
+            return value
+
+        compacted: dict[str, Any] = {}
+        for key, item in value.items():
+            if key == "title" and not named_schema_mapping:
+                continue
+            if named_schema_mapping:
+                compacted[key] = self._compact_tool_json_schema(item)
+            elif key == "description" and isinstance(item, str):
+                compacted[key] = self._compact_tool_description(item)
+            else:
+                compacted[key] = self._compact_tool_json_schema(
+                    item,
+                    named_schema_mapping=key
+                    in {"properties", "patternProperties", "$defs", "definitions"},
+                )
+        return compacted
+
     def _builtin_tool_schemas(self) -> list[dict[str, Any]]:
         return [
             {
@@ -1680,9 +1739,15 @@ async def _run_repeated_tool_decision(
             return None
 
         messages = self._messages_for_repeated_tool_decision(context, metadata)
+        call_llm = await prepare_llm_for_context(
+            llm=llm,
+            messages=messages,
+            context=context,
+        )
         decision_tools = [self._react_decision_tool_schema()]
         llm_metadata = {
             "phase": REPEATED_TOOL_DECISION_REQUESTED_STATUS,
+            **resolved_llm_metadata(call_llm),
             **metadata,
         }
         await runtime.on_llm_start(
@@ -1693,7 +1758,7 @@ async def _run_repeated_tool_decision(
         )
         try:
             response = await runtime.run_streaming_llm_call(
-                llm,
+                call_llm,
                 messages=messages,
                 tools=decision_tools,
                 tool_choice="required",

src/xagent/core/agent/runtime.py

@@ -7,6 +7,7 @@
 from typing import Any, Callable
 from uuid import uuid4
 
+from ...config import get_compact_threshold_ratio
 from ..agent.trace import (
     TraceAction,
     TraceCategory,
@@ -27,6 +28,55 @@ class LLMCallInterrupted(Exception):
     """Raised when an active LLM call is cancelled by an execution interrupt."""
 
 
+async def prepare_llm_for_context(
+    *,
+    llm: Any,
+    messages: list[dict[str, Any]],
+    context: Any,
+) -> Any:
+    """Resolve virtual models before compaction and apply their context window.
+
+    Normal models are returned unchanged. RouterLLM exposes ``prepare_for_call``
+    and returns a one-call wrapper for the concrete xrouter selection, ensuring
+    the selected model is reused after compaction instead of routing twice.
+    """
+    prepared = llm
+    prepare = getattr(llm, "prepare_for_call", None)
+    if callable(prepare):
+        prepared = prepare(messages)
+        if inspect.isawaitable(prepared):
+            prepared = await prepared
+
+    context_window = getattr(prepared, "context_window", None)
+    compact_config = getattr(context, "compact_config", None)
+    # Fixed-model thresholds are initialized once by AgentRunner and restored
+    # verbatim from checkpoints. Recompute only when a virtual model resolves to
+    # a concrete per-call wrapper whose window was unavailable at task start.
+    if (
+        prepared is not llm
+        and isinstance(context_window, int)
+        and context_window > 0
+        and compact_config is not None
+    ):
+        compact_config.threshold = max(
+            1, int(context_window * get_compact_threshold_ratio())
+        )
+
+    return prepared
+
+
+def resolved_llm_metadata(llm: Any) -> dict[str, Any]:
+    """Trace metadata for an already-resolved virtual model."""
+    metadata: dict[str, Any] = {}
+    model_name = getattr(llm, "model_name", None)
+    context_window = getattr(llm, "context_window", None)
+    if isinstance(model_name, str) and model_name:
+        metadata["selected_model"] = model_name
+    if isinstance(context_window, int) and context_window > 0:
+        metadata["context_window"] = context_window
+    return metadata
+
+
 @dataclass
 class PatternRuntime:
     """Thin runtime services shared by execution patterns.