fix: stamp provider usage at the adapter boundary and close the compact accounting gap

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

@@ -783,8 +783,14 @@ def record_llm_usage(
         input_tokens: int,
         output_tokens: int,
         prompt_message_count: int | None = None,
+        synthetic_purpose: str | None = None,
     ) -> None:
-        """Record provider usage for an LLM call without appending a message."""
+        """Record provider usage for an LLM call without appending a message.
+
+        ``synthetic_purpose`` marks internal non-conversational calls (e.g.
+        ``"context_compaction"``) whose prompt token count must not serve as
+        the context-freshness baseline.
+        """
 
         if input_tokens <= 0 and output_tokens <= 0:
             return
@@ -802,6 +808,7 @@ def record_llm_usage(
                 prompt_content_chars=self._message_content_chars(
                     self.messages[:prompt_message_count]
                 ),
+                synthetic_purpose=synthetic_purpose,
             )
         )
 
@@ -919,6 +926,7 @@ def _merge_llm_calls(self, contexts: list["ExecutionContext"]) -> None:
                         prompt_message_count=call.prompt_message_count,
                         prompt_content_chars=call.prompt_content_chars,
                         timestamp=call.timestamp,
+                        synthetic_purpose=call.synthetic_purpose,
                     )
                 )
         self.llm_calls = merged_calls
@@ -993,6 +1001,7 @@ def to_dict(self) -> dict[str, Any]:
                     "prompt_message_count": call.prompt_message_count,
                     "prompt_content_chars": call.prompt_content_chars,
                     "timestamp": call.timestamp.isoformat(),
+                    "synthetic_purpose": call.synthetic_purpose,
                 }
                 for call in self.llm_calls
             ],
@@ -1041,6 +1050,9 @@ def from_dict(cls, data: dict[str, Any]) -> "ExecutionContext":
                 prompt_message_count=call.get("prompt_message_count"),
                 prompt_content_chars=call.get("prompt_content_chars"),
                 timestamp=datetime.fromisoformat(call["timestamp"]),
+                # Older checkpoints predate the field; absence means a real
+                # conversational call, i.e. eligible as freshness baseline.
+                synthetic_purpose=call.get("synthetic_purpose"),
             )
             for call in data.get("llm_calls", [])
         ]
@@ -1570,26 +1582,40 @@ def estimate_context_tokens(self) -> int:
         return self._get_total_tokens()
 
     def _get_total_tokens(self) -> int:
-        if self.llm_calls:
-            latest_call = self.llm_calls[-1]
-            if latest_call.input_tokens > 0:
-                prompt_message_count = latest_call.prompt_message_count
-                prompt_content_chars = latest_call.prompt_content_chars
-                if (
-                    prompt_message_count is not None
-                    and prompt_content_chars is not None
-                    and 0 <= prompt_message_count <= len(self.messages)
-                    and self._message_content_chars(
-                        self.messages[:prompt_message_count]
-                    )
-                    == prompt_content_chars
-                ):
-                    delta_chars = self._message_content_chars(
-                        self.messages[prompt_message_count:]
-                    )
-                    return latest_call.input_tokens + max(0, delta_chars // 4)
+        baseline_call = self._latest_freshness_baseline_call()
+        if baseline_call is not None and baseline_call.input_tokens > 0:
+            prompt_message_count = baseline_call.prompt_message_count
+            prompt_content_chars = baseline_call.prompt_content_chars
+            if (
+                prompt_message_count is not None
+                and prompt_content_chars is not None
+                and 0 <= prompt_message_count <= len(self.messages)
+                and self._message_content_chars(self.messages[:prompt_message_count])
+                == prompt_content_chars
+            ):
+                delta_chars = self._message_content_chars(
+                    self.messages[prompt_message_count:]
+                )
+                return baseline_call.input_tokens + max(0, delta_chars // 4)
         return self._estimate_message_tokens(self.messages)
 
+    def _latest_freshness_baseline_call(self) -> LLMCallRecord | None:
+        """Most recent usage record eligible as the context-size baseline.
+
+        Synthetic records (internal calls such as context compaction) are
+        skipped: their prompt is not the live conversation, so their token
+        count says nothing about the current context size. When a synthetic
+        call did not rewrite the messages (e.g. a declined LLM compaction),
+        its fingerprint still matches -- without this skip, a small
+        compact-prompt count would be mistaken for the live context size
+        and suppress the truncation fallback.
+        """
+        for call in reversed(self.llm_calls):
+            if call.synthetic_purpose:
+                continue
+            return call
+        return None
+
     def _estimate_message_tokens(self, messages: list[Message]) -> int:
         return sum(
             max(1, len(message.content) // 4) + message.context_refs_token_estimate()

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

@@ -104,7 +104,14 @@ def context_refs_token_estimate(self) -> int:
 
 @dataclass
 class LLMCallRecord:
-    """Tracks token usage for a single LLM call."""
+    """Tracks token usage for a single LLM call.
+
+    ``synthetic_purpose`` marks records of internal, non-conversational
+    calls (currently only ``"context_compaction"``): their prompt is not
+    the live conversation, so they are skipped when the context-size
+    estimate picks its freshness baseline (see
+    ``ExecutionContext._get_total_tokens``).
+    """
 
     input_tokens: int
     output_tokens: int
@@ -113,3 +120,4 @@ class LLMCallRecord:
     prompt_message_count: int | None = None
     prompt_content_chars: int | None = None
     timestamp: datetime = field(default_factory=_utcnow)
+    synthetic_purpose: str | None = None

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

@@ -67,6 +67,7 @@
     final_deliverable_file_reference_instructions,
 )
 from ....model.chat.exceptions import LLMToolProtocolError
+from ....model.chat.response_shape import classify_chat_response
 from ....model.chat.tool_protocol import get_tool_protocol_error
 from ....tools.adapters.vibe.interaction_types import INTERACTION_TYPES
 from ....tools.user_interaction import (
@@ -968,10 +969,18 @@ async def _run_tool_calling_loop(
 
             await runtime.checkpoint("after_llm", context=context, pattern=self)
             if normalized.get("done", True):
+                # Fall back to the raw response's usable *text*, never the
+                # raw value itself: for envelope adapters the raw is the
+                # whole envelope dict, and stringifying it downstream would
+                # leak an internal repr into the user-visible transcript.
+                response = assistant_content
+                if not response:
+                    raw_shape = classify_chat_response(normalized.get("raw"))
+                    response = raw_shape.text if raw_shape.kind == "text" else ""
                 return await self._finalize_success(
                     context=context,
                     runtime=runtime,
-                    response=assistant_content or normalized.get("raw"),
+                    response=response,
                 )
 
         self.status = "max_iterations"

src/xagent/core/agent/runtime.py

@@ -3,6 +3,7 @@
 import asyncio
 import inspect
 import logging
+import math
 from collections.abc import Mapping
 from dataclasses import dataclass, field
 from typing import Any, Callable
@@ -1128,11 +1129,16 @@ async def on_llm_end(
     ) -> None:
         event_metadata = metadata or {}
         usage = self._extract_token_usage(response)
+        # Internal calls (currently only "context_compaction") carry a
+        # purpose marker; their prompt is not the live conversation, so the
+        # records must not become the freshness baseline.
+        purpose = event_metadata.get("purpose")
         if usage is not None and callable(getattr(context, "record_llm_usage", None)):
             context.record_llm_usage(
                 input_tokens=usage[0],
                 output_tokens=usage[1],
                 prompt_message_count=len(getattr(context, "messages", [])),
+                synthetic_purpose=purpose,
             )
         cached_tokens = self._extract_cached_tokens(response)
         await self._emit_trace_event(
@@ -1156,9 +1162,39 @@ async def on_llm_end(
             },
         )
 
+    def _resolve_usage_payload(self, response: Any) -> list[tuple[str, Any]]:
+        """Candidate usage payloads of a chat response, most preferred first.
+
+        Looks at the top-level ``usage``/``usage_metadata`` keys first (the
+        adapter stamp and legacy shapes), then one level down inside ``raw``
+        for envelopes that embed the raw provider payload without a stamp.
+        Unknown shapes yield no candidates -- callers fail open to None/0
+        rather than raising.
+        """
+        candidates: list[tuple[str, Any]] = []
+        for key in ("usage", "usage_metadata"):
+            value = self._get_value(response, key)
+            if value is not None:
+                candidates.append((key, value))
+        raw = self._get_value(response, "raw")
+        if raw is not None:
+            for key in ("usage", "usage_metadata"):
+                value = self._get_value(raw, key)
+                if value is not None:
+                    candidates.append((key, value))
+        return candidates
+
     def _extract_token_usage(self, response: Any) -> tuple[int, int] | None:
-        usage = self._get_value(response, "usage")
-        if usage is not None:
+        for key, usage in self._resolve_usage_payload(response):
+            input_tokens, output_tokens = self._usage_pair(key, usage)
+            if input_tokens > 0 or output_tokens > 0:
+                return input_tokens, output_tokens
+
+        return None
+
+    def _usage_pair(self, key: str, usage: Any) -> tuple[int, int]:
+        """(input, output) tokens of one usage payload by candidate kind."""
+        if key == "usage":
             input_tokens = self._first_int(
                 usage, ("prompt_tokens", "input_tokens", "prompt_token_count")
             )
@@ -1171,44 +1207,79 @@ def _extract_token_usage(self, response: Any) -> tuple[int, int] | None:
                     "completion_token_count",
                 ),
             )
-            if input_tokens > 0 or output_tokens > 0:
-                return input_tokens, output_tokens
-
-        usage_metadata = self._get_value(response, "usage_metadata")
-        if usage_metadata is not None:
+        else:
             input_tokens = self._first_int(
-                usage_metadata, ("prompt_token_count", "prompt_tokens", "input_tokens")
+                usage, ("prompt_token_count", "prompt_tokens", "input_tokens")
             )
             output_tokens = self._first_int(
-                usage_metadata,
+                usage,
                 ("candidates_token_count", "completion_tokens", "output_tokens"),
             )
-            if input_tokens > 0 or output_tokens > 0:
-                return input_tokens, output_tokens
-
-        return None
+        return input_tokens, output_tokens
 
     def _extract_cached_tokens(self, response: Any) -> int:
         """Prompt-cache-hit tokens from a response's usage payload, 0 if absent."""
-        usage = self._get_value(response, "usage")
-        if usage is None:
-            return 0
+        for key, usage in self._resolve_usage_payload(response):
+            if key != "usage":
+                continue
+            cached = self._cached_from_payload(usage)
+            if cached:
+                return cached
+        return 0
+
+    def _cached_from_payload(self, usage: Any) -> int:
+        """Strictly-coerced cache-hit tokens from one usage payload.
+
+        Same alias table as the ledger-side ``extract_cached_input_tokens``,
+        but through ``_coerce_usage_int``: malformed values (bools, strings,
+        non-finite or non-integral numbers, negatives) are rejected rather
+        than truncated, and an invalid or zero alias never shadows the
+        nested ``prompt_tokens_details.cached_tokens`` fallback.
+        """
         direct = self._first_int(
             usage, ("cached_input_tokens", "cache_read_input_tokens")
         )
         if direct > 0:
             return direct
-        return extract_cached_input_tokens(usage)
+        hit = self._coerce_usage_int(self._get_value(usage, "prompt_cache_hit_tokens"))
+        if hit:
+            return hit
+        details = self._get_value(usage, "prompt_tokens_details")
+        if details is not None:
+            nested = self._coerce_usage_int(self._get_value(details, "cached_tokens"))
+            if nested:
+                return nested
+        return 0
 
     def _first_int(self, source: Any, keys: tuple[str, ...]) -> int:
         for key in keys:
-            value = self._get_value(source, key)
-            if isinstance(value, int):
-                return value
-            if isinstance(value, float):
-                return int(value)
+            coerced = self._coerce_usage_int(self._get_value(source, key))
+            if coerced is not None:
+                return coerced
         return 0
 
+    @staticmethod
+    def _coerce_usage_int(value: Any) -> int | None:
+        """Coerce a usage counter to a non-negative int, or return None.
+
+        Usage numbers feed billing and context-freshness decisions, so the
+        coercion is strict: bools (an int subclass), non-finite floats
+        (NaN/inf, which ``int()`` would raise on), negatives, and
+        non-integral floats are all rejected instead of being truncated or
+        crashing the extractor. Integral floats (``10.0``) coerce. A None
+        result means "not a usable counter" -- callers skip to the next
+        alias/candidate rather than treating it as a measurement.
+        """
+        if isinstance(value, bool):
+            return None
+        if isinstance(value, int):
+            return value if value >= 0 else None
+        if isinstance(value, float):
+            if not math.isfinite(value) or value < 0 or not value.is_integer():
+                return None
+            return int(value)
+        return None
+
     def _get_value(self, source: Any, key: str) -> Any:
         if isinstance(source, dict):
             return source.get(key)

src/xagent/core/agent/utils/context_builder.py

@@ -11,6 +11,7 @@
 from ...model.chat.basic.base import BaseLLM
 from ..trace import Tracer, trace_compact_end, trace_compact_start
 from .compact import CompactConfig, CompactUtils
+from .llm_utils import unwrap_chat_text
 
 logger = logging.getLogger(__name__)
 
@@ -396,13 +397,11 @@ async def _compact_individual_dependency(
                 },
             ]
 
-            # Get compacted response
+            # Get compacted response. ``unwrap_chat_text`` raises on a
+            # tool_call envelope instead of repr()ing it (#1714); the
+            # surrounding except falls back to truncation.
             response = await self.compact_llm.chat(messages=compact_prompt)
-            content = (
-                response
-                if isinstance(response, str)
-                else response.get("content", str(response))
-            )
+            content = unwrap_chat_text(response)
 
             # Parse back to messages format
             compacted_messages = CompactUtils.parse_compact_response(content)
@@ -489,13 +488,11 @@ async def _compact_dependency_messages(
                 },
             ]
 
-            # Get compacted response
+            # Get compacted response. ``unwrap_chat_text`` raises on a
+            # tool_call envelope instead of repr()ing it (#1714); the
+            # surrounding except falls back to truncation.
             response = await self.compact_llm.chat(messages=compact_prompt)
-            content = (
-                response
-                if isinstance(response, str)
-                else response.get("content", str(response))
-            )
+            content = unwrap_chat_text(response)
 
             # Parse back to messages format
             compacted_messages = CompactUtils.parse_compact_response(content)