diff --git a/src/xagent/core/agent/context/execution.py b/src/xagent/core/agent/context/execution.py
index daacab5347..e0131a91be 100644
--- a/src/xagent/core/agent/context/execution.py
+++ b/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()
diff --git a/src/xagent/core/agent/context/message.py b/src/xagent/core/agent/context/message.py
index db752bfe9f..87a82b68ba 100644
--- a/src/xagent/core/agent/context/message.py
+++ b/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
diff --git a/src/xagent/core/agent/pattern/react/react.py b/src/xagent/core/agent/pattern/react/react.py
index 1e9180c1e2..75c0f2543e 100644
--- a/src/xagent/core/agent/pattern/react/react.py
+++ b/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"
diff --git a/src/xagent/core/agent/runtime.py b/src/xagent/core/agent/runtime.py
index 8ca102c07c..9bcc55e9db 100644
--- a/src/xagent/core/agent/runtime.py
+++ b/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)
diff --git a/src/xagent/core/agent/utils/context_builder.py b/src/xagent/core/agent/utils/context_builder.py
index 3e3f76440e..b7b652a30d 100644
--- a/src/xagent/core/agent/utils/context_builder.py
+++ b/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)
diff --git a/src/xagent/core/agent/utils/llm_utils.py b/src/xagent/core/agent/utils/llm_utils.py
index 3cd6e4832b..e25f9bcd60 100644
--- a/src/xagent/core/agent/utils/llm_utils.py
+++ b/src/xagent/core/agent/utils/llm_utils.py
@@ -5,9 +5,52 @@
 import re
 from typing import Any, Dict, List
 
+from ...model.chat.exceptions import LLMEmptyContentError, LLMNoTextContentError
+from ...model.chat.response_shape import classify_chat_response
+
 logger = logging.getLogger(__name__)
 
 
+def unwrap_chat_text(response: Any) -> str:
+    """Extract the text content of a ``chat()`` response, or raise.
+
+    Adapters return either a plain string (legacy shape) or an envelope dict
+    such as ``{"type": "text", "content": ...}`` /
+    ``{"type": "tool_call", ...}``. Callers that need the text must never
+    fall back to ``str(response)``: on a tool_call envelope that yields the
+    dict's repr, which then leaks into compacted context or API responses as
+    if it were model output (#1714). Shape reading delegates to
+    ``classify_chat_response`` so every consumer shares one structural
+    source of truth.
+
+    Returns:
+        The response itself for plain-string replies, or the ``content`` of
+        a dict envelope carrying a usable non-empty string.
+
+    Raises:
+        LLMEmptyContentError: The response carries string content that is
+            empty or whitespace-only (same transient class the adapters raise
+            for an empty generation) -- envelope or legacy plain string.
+        LLMNoTextContentError: The response is a tool_call envelope, carries
+            non-string content, or has an unrecognized shape.
+    """
+    shape = classify_chat_response(response)
+    if shape.kind == "text":
+        assert shape.text is not None  # classifier invariant
+        return shape.text
+    if shape.kind == "empty":
+        raise LLMEmptyContentError("Chat response content is empty")
+    if isinstance(response, dict):
+        raise LLMNoTextContentError(
+            "Chat response has no usable text content "
+            f"(type={response.get('type')!r}, keys={sorted(response.keys())})"
+        )
+    raise LLMNoTextContentError(
+        "Chat response has no usable text content "
+        f"(response type={type(response).__name__})"
+    )
+
+
 def clean_llm_content(content: str) -> str:
     """Clean content sent to LLM by removing characters that may cause API errors.
 
diff --git a/src/xagent/core/model/chat/basic/base.py b/src/xagent/core/model/chat/basic/base.py
index e5d169e888..30a0aa81f1 100644
--- a/src/xagent/core/model/chat/basic/base.py
+++ b/src/xagent/core/model/chat/basic/base.py
@@ -224,25 +224,29 @@ async def chat(
             **kwargs: Additional parameters specific to the underlying model (e.g. top_p, user, stop).
 
         Returns:
-            The return type is a union; the concrete shape depends on the
-            implementation:
-                -> str: some implementations (e.g. Zhipu, Claude, Gemini)
-                   return the assistant reply content as a bare string.
-                -> dict: other implementations (e.g. the OpenAI family --
-                   OpenAI, OpenRouter, DashScope -- and Xinference) wrap
-                   the reply in an envelope instead:
-                     - {"type": "text", "content": <str>, ...} for a
-                       natural language response
-                     - {"type": "tool_call", "tool_calls": [...], ...} for a
-                       tool call
-                   A "raw" key carrying the provider's full response is
-                   present on some implementations' envelopes and absent on
-                   others, so callers must not require it. Neither list is
-                   exhaustive, and an implementation listed above as
-                   returning a bare string for ordinary replies can still
-                   return a tool-call envelope -- Gemini does exactly that.
-                   Callers that must accept more than one implementation need
-                   to branch on the shape rather than assume a bare string.
+            If the model returns a natural language response:
+                -> dict envelope with fields:
+                    - "type": "text"
+                    - "content": the assistant reply content
+                    - "usage": top-level provider usage stamp, when the provider
+                      reported one (e.g. {"prompt_tokens": ..., "completion_tokens": ...})
+                    - "raw": the provider's full response payload. Present on
+                      some implementations' envelopes and absent on others,
+                      so callers must not require it.
+                (Legacy adapters may still return a plain string reply.)
+
+            If the model triggers a tool call:
+                -> dict envelope with fields:
+                    - "type": "tool_call"
+                    - "tool_calls": list of tool call objects
+                    - "raw": the provider's full response payload -- optional,
+                      provider-dependent (e.g. Gemini's envelope omits it);
+                      callers must not require it
+                    - "usage": top-level provider usage stamp, when reported
+
+            Consumers needing the reply text must classify the envelope
+            structurally (``classify_chat_response`` / ``unwrap_chat_text``)
+            rather than stringifying it.
 
         Raises:
             RuntimeError if the model call fails or returns an unexpected format.
@@ -279,25 +283,13 @@ async def vision_chat(
             **kwargs: Additional parameters specific to the underlying model.
 
         Returns:
-            The return type is a union; the concrete shape depends on the
-            implementation:
-                -> str: some implementations (e.g. Zhipu, Claude, Gemini)
-                   return the assistant reply content as a bare string.
-                -> dict: other implementations (e.g. the OpenAI family --
-                   OpenAI, OpenRouter, DashScope -- and Xinference) wrap
-                   the reply in an envelope instead:
-                     - {"type": "text", "content": <str>, ...} for a
-                       natural language response
-                     - {"type": "tool_call", "tool_calls": [...], ...} for a
-                       tool call
-                   A "raw" key carrying the provider's full response is
-                   present on some implementations' envelopes and absent on
-                   others, so callers must not require it. Neither list is
-                   exhaustive, and an implementation listed above as
-                   returning a bare string for ordinary replies can still
-                   return a tool-call envelope -- Gemini does exactly that.
-                   Callers that must accept more than one implementation need
-                   to branch on the shape rather than assume a bare string.
+            Same envelope contract as ``chat()``: a text envelope
+            (``{"type": "text", "content": ...}``, optionally with ``usage``
+            and ``raw``) for a natural language response, a tool_call
+            envelope (``{"type": "tool_call", "tool_calls": [...]}``) when
+            the model triggers a tool call, or -- for legacy adapters -- a
+            plain string reply. Never stringify the envelope to get the
+            text; use ``classify_chat_response`` / ``unwrap_chat_text``.
 
         Raises:
             RuntimeError if the model doesn't support vision or the call fails.
diff --git a/src/xagent/core/model/chat/basic/claude.py b/src/xagent/core/model/chat/basic/claude.py
index 59e5cf3414..cdf905e6a1 100644
--- a/src/xagent/core/model/chat/basic/claude.py
+++ b/src/xagent/core/model/chat/basic/claude.py
@@ -531,7 +531,8 @@ async def chat(
             **kwargs: Additional parameters to pass to the Anthropic API
 
         Returns:
-            - If normal text reply: return string
+            - If normal text reply: return dict with type "text" and content
+              (plus a top-level "usage" payload when the provider reported one)
             - If tool call triggered: return dict with type "tool_call" and tool_calls list
 
         Raises:
@@ -656,11 +657,23 @@ async def chat(
             # Make the API call
             response = await self._client.messages.create(**completion_params)
 
-            # Record token usage
+            # Record token usage; snapshot it as an OpenAI-style payload so the
+            # result envelopes below can carry a top-level ``usage`` stamp.
+            usage_payload: Optional[Dict[str, Any]] = None
             if hasattr(response, "usage"):
                 usage = response.usage
                 input_tokens, cache_read, cache_write = _anthropic_input_usage(usage)
                 output_tokens = getattr(usage, "output_tokens", 0)
+                usage_payload = {
+                    "prompt_tokens": input_tokens,
+                    "completion_tokens": output_tokens,
+                }
+                # Only stamp cache metrics when non-zero so a default 0 never
+                # shadows fallback fields in downstream extraction.
+                if cache_read > 0:
+                    usage_payload["cached_input_tokens"] = cache_read
+                if cache_write > 0:
+                    usage_payload["cache_write_input_tokens"] = cache_write
                 add_token_usage(
                     input_tokens=input_tokens,
                     output_tokens=output_tokens,
@@ -691,13 +704,16 @@ async def chat(
                         )
 
                 if tool_calls:
-                    return {
+                    result: Dict[str, Any] = {
                         "type": "tool_call",
                         "tool_calls": tool_calls,
                         "raw": response.model_dump()
                         if hasattr(response, "model_dump")
                         else str(response),
                     }
+                    if usage_payload is not None:
+                        result["usage"] = usage_payload
+                    return result
 
             # Extract text content
             text_content = []
@@ -707,6 +723,12 @@ async def chat(
 
             content = "".join(text_content).strip()
 
+            def _text_result(text: str) -> Dict[str, Any]:
+                result: Dict[str, Any] = {"type": "text", "content": text}
+                if usage_payload is not None:
+                    result["usage"] = usage_payload
+                return result
+
             if not content:
                 # Empty response should trigger retry
                 raise LLMRetryableError("LLM returned empty content and no tool calls")
@@ -716,10 +738,12 @@ async def chat(
                 try:
                     # Try to repair the JSON first
                     repaired_content = repair_loads(content, logging=False)
-                    # If repair succeeded, return the repaired JSON as string
-                    # to maintain consistency with normal text response
+                    # If repair succeeded, return the repaired JSON as the text
+                    # envelope's content, consistent with normal text responses
                     logger.info("JSON repair succeeded, returning repaired content")
-                    return json.dumps(repaired_content, ensure_ascii=False)
+                    return _text_result(
+                        json.dumps(repaired_content, ensure_ascii=False)
+                    )
                 except Exception as repair_error:
                     # JSON repair failed - raise retryable error to trigger retry
                     logger.warning(
@@ -736,9 +760,9 @@ async def chat(
                     # When using json_schema, the response is already validated JSON
                     # Return as-is since it's guaranteed to be valid
                     logger.info("Returning JSON schema validated response")
-                    return content
+                    return _text_result(content)
 
-            return content
+            return _text_result(content)
 
         except Exception as e:
             logger.error(f"Claude API error: {str(e)}")
@@ -1197,7 +1221,8 @@ async def vision_chat(
             **kwargs: Additional parameters to pass to the Claude API
 
         Returns:
-            - If normal text reply: return string
+            - If normal text reply: return dict with type "text" and content
+              (plus a top-level "usage" payload when the provider reported one)
             - If tool call triggered: return dict with type "tool_call" and tool_calls list
 
         Raises:
diff --git a/src/xagent/core/model/chat/basic/deepseek_tool_protocol.py b/src/xagent/core/model/chat/basic/deepseek_tool_protocol.py
index 3ca20ca3b9..28dad42a3f 100644
--- a/src/xagent/core/model/chat/basic/deepseek_tool_protocol.py
+++ b/src/xagent/core/model/chat/basic/deepseek_tool_protocol.py
@@ -53,7 +53,12 @@ def normalize_deepseek_response(
     if violation is None:
         return response
     raw = response.get("raw") if isinstance(response, dict) else None
-    return tool_protocol_error_response(violation, raw=raw)
+    error_response = tool_protocol_error_response(violation, raw=raw)
+    # Preserve the top-level usage stamp the adapter put on the original
+    # envelope so token accounting survives the error rebuild.
+    if isinstance(response, dict) and response.get("usage") is not None:
+        error_response["usage"] = response["usage"]
+    return error_response
 
 
 async def adapt_deepseek_stream(
diff --git a/src/xagent/core/model/chat/basic/gemini.py b/src/xagent/core/model/chat/basic/gemini.py
index 54b21adef0..0bd406666f 100644
--- a/src/xagent/core/model/chat/basic/gemini.py
+++ b/src/xagent/core/model/chat/basic/gemini.py
@@ -574,7 +574,9 @@ async def chat(
 
             response = await self._client.aio.models.generate_content(**api_params)
 
-            # Extract token usage
+            # Extract token usage; snapshot it as an OpenAI-style payload so
+            # the result envelopes below can carry a top-level ``usage`` stamp.
+            usage_payload: Optional[Dict[str, Any]] = None
             if hasattr(response, "usage_metadata") and response.usage_metadata:
                 usage_metadata = response.usage_metadata
                 input_tokens = getattr(usage_metadata, "prompt_token_count", 0)
@@ -584,6 +586,14 @@ async def chat(
                 )
 
                 if input_tokens > 0 or output_tokens > 0:
+                    usage_payload = {
+                        "prompt_tokens": input_tokens,
+                        "completion_tokens": output_tokens,
+                    }
+                    # Guard the comparison: real SDKs report int/None here,
+                    # but token accounting must never raise out of chat().
+                    if isinstance(cached_tokens, (int, float)) and cached_tokens > 0:
+                        usage_payload["cached_input_tokens"] = cached_tokens
                     add_token_usage(
                         input_tokens=input_tokens,
                         output_tokens=output_tokens,
@@ -638,10 +648,13 @@ async def chat(
                     text_parts.append(part.text)
 
             if tool_calls:
-                return {
+                tool_result: Dict[str, Any] = {
                     "type": "tool_call",
                     "tool_calls": tool_calls,
                 }
+                if usage_payload is not None:
+                    tool_result["usage"] = usage_payload
+                return tool_result
 
             content = "".join(text_parts).strip()
 
@@ -650,7 +663,10 @@ async def chat(
                     "LLM returned empty content and no tool calls"
                 )
 
-            return content
+            text_result: Dict[str, Any] = {"type": "text", "content": content}
+            if usage_payload is not None:
+                text_result["usage"] = usage_payload
+            return text_result
 
         except Exception as e:
             logger.error("Gemini SDK API error: %s", redact_sensitive_text(str(e)))
diff --git a/src/xagent/core/model/chat/basic/openai.py b/src/xagent/core/model/chat/basic/openai.py
index dd55973f61..30db0fe6a4 100644
--- a/src/xagent/core/model/chat/basic/openai.py
+++ b/src/xagent/core/model/chat/basic/openai.py
@@ -22,6 +22,23 @@
 
 logger = logging.getLogger(__name__)
 
+
+def _response_usage_payload(response: Any) -> Any:
+    """Return the provider usage payload of a raw SDK response, or None.
+
+    Stamped onto every chat/vision result envelope as a top-level ``usage``
+    key so consumers (notably ``PatternRuntime._extract_token_usage``) can
+    read usage without reaching into ``raw``. Read-only: the contextvar
+    ledger write stays with the adapter's single ``add_token_usage`` call.
+    """
+    usage = getattr(response, "usage", None)
+    if usage is None:
+        return None
+    if hasattr(usage, "model_dump"):
+        return usage.model_dump()
+    return usage
+
+
 # ``PROVIDER_STATE_METADATA_KEY`` now lives in ``chat.types`` (see there for
 # why); imported here so existing code and tests that import it from this
 # transport module keep working unchanged.
@@ -609,7 +626,9 @@ def _process_response(
             choice = resp.choices[0]
             message = choice.message
 
-            # Record token usage to context
+            # Record token usage to context; snapshot it so every result
+            # envelope below can carry a top-level ``usage`` stamp.
+            usage_payload = _response_usage_payload(resp)
             if hasattr(resp, "usage") and resp.usage:
                 add_token_usage(
                     input_tokens=resp.usage.prompt_tokens,
@@ -648,6 +667,8 @@ def _process_response(
                     "tool_calls": tool_calls,
                     "raw": resp.model_dump(),
                 }
+                if usage_payload is not None:
+                    result["usage"] = usage_payload
                 has_reasoning_content, reasoning_content = _message_reasoning_content(
                     message
                 )
@@ -695,7 +716,7 @@ def _process_response(
                     and reasoning_content
                     and reasoning_content.strip()
                 ):
-                    return {
+                    result = {
                         "type": "text",
                         "content": reasoning_content,
                         CONTENT_SOURCE_KEY: CONTENT_SOURCE_REASONING_FALLBACK,
@@ -703,6 +724,9 @@ def _process_response(
                         "reasoning": reasoning_content,
                         "raw": resp.model_dump(),
                     }
+                    if usage_payload is not None:
+                        result["usage"] = usage_payload
+                    return result
                 # If there are no tool calls and no content, this is an error
                 raise LLMEmptyContentError(
                     f"LLM returned {'empty' if content == '' else 'None'} content and no tool calls"
@@ -713,6 +737,8 @@ def _process_response(
                 "content": content,
                 "raw": resp.model_dump(),
             }
+            if usage_payload is not None:
+                result["usage"] = usage_payload
             if has_reasoning_content:
                 result["reasoning_content"] = reasoning_content
                 result["reasoning"] = reasoning_content
@@ -981,7 +1007,9 @@ async def _make_api_call() -> Any:
             choice = response.choices[0]
             message = choice.message
 
-            # Record token usage to context
+            # Record token usage to context; snapshot it so every result
+            # envelope below can carry a top-level ``usage`` stamp.
+            usage_payload = _response_usage_payload(response)
             if hasattr(response, "usage") and response.usage:
                 add_token_usage(
                     input_tokens=response.usage.prompt_tokens,
@@ -1020,6 +1048,8 @@ async def _make_api_call() -> Any:
                     "tool_calls": tool_calls,
                     "raw": response.model_dump(),
                 }
+                if usage_payload is not None:
+                    result["usage"] = usage_payload
                 has_reasoning_content, reasoning_content = _message_reasoning_content(
                     message
                 )
@@ -1057,7 +1087,7 @@ async def _make_api_call() -> Any:
                     and reasoning_content
                     and reasoning_content.strip()
                 ):
-                    return {
+                    result = {
                         "type": "text",
                         "content": reasoning_content,
                         CONTENT_SOURCE_KEY: CONTENT_SOURCE_REASONING_FALLBACK,
@@ -1065,6 +1095,9 @@ async def _make_api_call() -> Any:
                         "reasoning": reasoning_content,
                         "raw": response.model_dump(),
                     }
+                    if usage_payload is not None:
+                        result["usage"] = usage_payload
+                    return result
                 # If there are no tool calls and no content, this is an error
                 raise LLMEmptyContentError(
                     f"LLM returned {'empty' if content == '' else 'None'} content and no tool calls"
@@ -1075,6 +1108,8 @@ async def _make_api_call() -> Any:
                 "content": content,
                 "raw": response.model_dump(),
             }
+            if usage_payload is not None:
+                text_result["usage"] = usage_payload
             if has_reasoning_content:
                 text_result["reasoning_content"] = reasoning_content
                 text_result["reasoning"] = reasoning_content
diff --git a/src/xagent/core/model/chat/basic/xinference.py b/src/xagent/core/model/chat/basic/xinference.py
index f504f3348d..fbdce2dc2a 100644
--- a/src/xagent/core/model/chat/basic/xinference.py
+++ b/src/xagent/core/model/chat/basic/xinference.py
@@ -311,7 +311,9 @@ def _process_chat_response(self, response: Any) -> Dict[str, Any]:
         # Xinference returns a dict-like object with various fields
         response_dict = dict(response) if not isinstance(response, dict) else response
 
-        # Record token usage if available
+        # Record token usage if available; the same payload is also stamped
+        # onto the returned envelope (the adapter-boundary usage contract, so
+        # consumers never need to dig through ``raw``).
         usage = response_dict.get("usage", {})
         if usage:
             add_token_usage(
@@ -342,6 +344,8 @@ def _process_chat_response(self, response: Any) -> Dict[str, Any]:
                 if reasoning_content:
                     result["reasoning_content"] = reasoning_content
                     result["reasoning"] = reasoning_content
+                if usage:
+                    result["usage"] = usage
                 return result
 
             # Handle text content
@@ -355,6 +359,8 @@ def _process_chat_response(self, response: Any) -> Dict[str, Any]:
                 if reasoning_content:
                     result["reasoning_content"] = reasoning_content
                     result["reasoning"] = reasoning_content
+                if usage:
+                    result["usage"] = usage
                 return result
 
             # Reasoning models (e.g. qwen3-thinking, deepseek-r1) may emit
@@ -375,22 +381,28 @@ def _process_chat_response(self, response: Any) -> Dict[str, Any]:
                 and reasoning_content
                 and reasoning_content.strip()
             ):
-                return {
+                result = {
                     "type": "text",
                     "content": reasoning_content,
                     "reasoning_content": reasoning_content,
                     "reasoning": reasoning_content,
                     "raw": response_dict,
                 }
+                if usage:
+                    result["usage"] = usage
+                return result
 
         # Fallback: try to get content directly from response
         content = response_dict.get("content", "")
         if content:
-            return {
+            result = {
                 "type": "text",
                 "content": content,
                 "raw": response_dict,
             }
+            if usage:
+                result["usage"] = usage
+            return result
 
         raise RuntimeError(f"Invalid Xinference response: {response_dict}")
 
diff --git a/src/xagent/core/model/chat/basic/zhipu.py b/src/xagent/core/model/chat/basic/zhipu.py
index e77e8ea7ba..dbb835e420 100644
--- a/src/xagent/core/model/chat/basic/zhipu.py
+++ b/src/xagent/core/model/chat/basic/zhipu.py
@@ -160,7 +160,8 @@ async def chat(
             **kwargs: Additional parameters to pass to the Zhipu API
 
         Returns:
-            - If normal text reply: return string
+            - If normal text reply: return dict with type "text" and content
+              (plus a top-level "usage" payload when the provider reported one)
             - If tool call triggered: return dict with type "tool_call" and tool_calls list
 
         Raises:
@@ -247,7 +248,9 @@ async def chat(
                 logger.error(f"Zhipu API response missing choices: {response}")
                 raise RuntimeError("Zhipu API response missing choices")
 
-            # Record token usage
+            # Record token usage; snapshot it as an OpenAI-style payload so the
+            # result envelopes below can carry a top-level ``usage`` stamp.
+            usage_payload: Optional[Dict[str, Any]] = None
             if hasattr(response, "usage"):
                 usage = response.usage
                 input_tokens = getattr(usage, "prompt_tokens", 0) or getattr(
@@ -256,13 +259,20 @@ async def chat(
                 output_tokens = getattr(usage, "completion_tokens", 0) or getattr(
                     usage, "output_tokens", 0
                 )
+                cached_tokens = extract_cached_input_tokens(usage)
+                usage_payload = {
+                    "prompt_tokens": input_tokens,
+                    "completion_tokens": output_tokens,
+                }
+                if cached_tokens > 0:
+                    usage_payload["cached_input_tokens"] = cached_tokens
                 add_token_usage(
                     input_tokens=input_tokens,
                     output_tokens=output_tokens,
                     model=self._model_name,
                     model_id=self.model_id,
                     call_type="chat",
-                    cached_input_tokens=extract_cached_input_tokens(usage),
+                    cached_input_tokens=cached_tokens,
                 )
 
             # Extract the choice
@@ -346,7 +356,7 @@ async def chat(
                         args = {}
 
                     # Return ReAct-compatible tool call format
-                    return {
+                    result: Dict[str, Any] = {
                         "type": "tool_call",
                         "tool_calls": [
                             {
@@ -364,6 +374,9 @@ async def chat(
                         if hasattr(response, "model_dump")
                         else str(response),
                     }
+                    if usage_payload is not None:
+                        result["usage"] = usage_payload
+                    return result
 
             # Handle text content
             content = message.content
@@ -390,7 +403,10 @@ async def chat(
                         "None/empty content but tool calls present, this is expected behavior"
                     )
 
-            return content
+            text_result: Dict[str, Any] = {"type": "text", "content": content}
+            if usage_payload is not None:
+                text_result["usage"] = usage_payload
+            return text_result
 
         except Exception as e:
             # Handle any errors
@@ -803,7 +819,8 @@ async def vision_chat(
             **kwargs: Additional parameters to pass to the Zhipu API
 
         Returns:
-            - If normal text reply: return string
+            - If normal text reply: return dict with type "text" and content
+              (plus a top-level "usage" payload when the provider reported one)
             - If tool call triggered: return dict with type "tool_call" and tool_calls list
 
         Raises:
@@ -891,7 +908,9 @@ async def vision_chat(
                 logger.error(f"Zhipu Vision API response missing choices: {response}")
                 raise RuntimeError("Zhipu Vision API response missing choices")
 
-            # Record token usage
+            # Record token usage; snapshot it as an OpenAI-style payload so the
+            # result envelopes below can carry a top-level ``usage`` stamp.
+            usage_payload: Optional[Dict[str, Any]] = None
             if hasattr(response, "usage"):
                 usage = response.usage
                 input_tokens = getattr(usage, "prompt_tokens", 0) or getattr(
@@ -900,13 +919,20 @@ async def vision_chat(
                 output_tokens = getattr(usage, "completion_tokens", 0) or getattr(
                     usage, "output_tokens", 0
                 )
+                cached_tokens = extract_cached_input_tokens(usage)
+                usage_payload = {
+                    "prompt_tokens": input_tokens,
+                    "completion_tokens": output_tokens,
+                }
+                if cached_tokens > 0:
+                    usage_payload["cached_input_tokens"] = cached_tokens
                 add_token_usage(
                     input_tokens=input_tokens,
                     output_tokens=output_tokens,
                     model=self._model_name,
                     model_id=self.model_id,
                     call_type="vision_chat",
-                    cached_input_tokens=extract_cached_input_tokens(usage),
+                    cached_input_tokens=cached_tokens,
                 )
 
             # Extract the choice
@@ -990,7 +1016,7 @@ async def vision_chat(
                         args = {}
 
                     # Return ReAct-compatible tool call format
-                    return {
+                    result: Dict[str, Any] = {
                         "type": "tool_call",
                         "tool_calls": [
                             {
@@ -1008,6 +1034,9 @@ async def vision_chat(
                         if hasattr(response, "model_dump")
                         else str(response),
                     }
+                    if usage_payload is not None:
+                        result["usage"] = usage_payload
+                    return result
 
             # Handle text content
             content = message.content
@@ -1030,13 +1059,19 @@ async def vision_chat(
                     logger.warning(
                         "No tool calls and None content, returning empty string"
                     )
-                    return ""
+                    empty_result: Dict[str, Any] = {"type": "text", "content": ""}
+                    if usage_payload is not None:
+                        empty_result["usage"] = usage_payload
+                    return empty_result
                 else:
                     logger.info(
                         "None content but tool calls present, this is expected behavior"
                     )
 
-            return content
+            vision_result: Dict[str, Any] = {"type": "text", "content": content}
+            if usage_payload is not None:
+                vision_result["usage"] = usage_payload
+            return vision_result
 
         except Exception as e:
             # Handle any errors
diff --git a/src/xagent/core/model/chat/exceptions.py b/src/xagent/core/model/chat/exceptions.py
index e398dd5dd8..dee1e6a3f6 100644
--- a/src/xagent/core/model/chat/exceptions.py
+++ b/src/xagent/core/model/chat/exceptions.py
@@ -78,6 +78,20 @@ class LLMInvalidResponseError(LLMRetryableError):
     pass
 
 
+class LLMNoTextContentError(LLMInvalidResponseError):
+    """Raised when a chat response carries no usable text content.
+
+    Distinct from ``LLMEmptyContentError`` (the provider answered with an
+    empty string): this means the response has a non-text shape -- a
+    tool_call envelope or an unrecognized payload -- where the caller
+    required text. Stringifying such a response would leak an internal
+    dict repr into compacted context or API responses as if it were model
+    output (#1714), so consumers must fail explicitly instead.
+    """
+
+    pass
+
+
 class LLMTimeoutError(LLMRetryableError):
     """Raised when LLM request times out.
 
diff --git a/src/xagent/core/model/chat/response_shape.py b/src/xagent/core/model/chat/response_shape.py
new file mode 100644
index 0000000000..26b90d81cb
--- /dev/null
+++ b/src/xagent/core/model/chat/response_shape.py
@@ -0,0 +1,62 @@
+"""Structural classification of ``chat()``/``vision_chat()`` response shapes.
+
+Adapters return a small union from the non-streaming chat methods:
+
+- a legacy plain string (the assistant reply),
+- a text envelope ``{"type": "text", "content": ...}`` (optionally with a
+  top-level ``usage`` stamp and a ``raw`` provider payload),
+- a tool-call envelope ``{"type": "tool_call", "tool_calls": [...]}``.
+
+This module is the single, dependency-neutral source of truth for telling
+those shapes apart. Consumers that need the text (``unwrap_chat_text`` in
+the agent layer, ``VisionCore`` in the tools layer, the default
+``stream_chat`` in this package) classify here instead of re-implementing
+isinstance chains -- and none of them ever falls back to ``str(response)``,
+which would leak an internal dict repr as if it were model output (#1714).
+"""
+
+from typing import Any, Literal, NamedTuple
+
+
+class ChatResponseShape(NamedTuple):
+    """Structural reading of a chat response.
+
+    ``kind`` is one of:
+
+    - ``"text"``: usable text is present; ``text`` carries it.
+    - ``"empty"``: a text-bearing shape with no usable text (empty or
+      whitespace-only) -- the same transient condition adapters raise
+      ``LLMEmptyContentError`` for.
+    - ``"tool_call"``: a tool-call envelope; there is no text by design.
+    - ``"unknown"``: any other payload (unrecognized dict, non-string
+      content, non-dict/non-string value).
+    """
+
+    kind: Literal["text", "empty", "tool_call", "unknown"]
+    text: str | None
+
+
+def classify_chat_response(response: Any) -> ChatResponseShape:
+    """Classify a ``chat()``/``vision_chat()`` response by structure.
+
+    Classification is purely structural and never raises: a legacy plain
+    string is text (or empty when whitespace-only); a dict tagged
+    ``type == "tool_call"`` is a tool call; any other dict with string
+    ``content`` is text (or empty when whitespace-only) regardless of its
+    ``type`` tag, matching the duck-typed acceptance ``unwrap_chat_text``
+    has always had; everything else is unknown.
+    """
+    if isinstance(response, str):
+        if response.strip():
+            return ChatResponseShape("text", response)
+        return ChatResponseShape("empty", None)
+    if isinstance(response, dict):
+        if response.get("type") == "tool_call":
+            return ChatResponseShape("tool_call", None)
+        content = response.get("content")
+        if isinstance(content, str):
+            if content.strip():
+                return ChatResponseShape("text", content)
+            return ChatResponseShape("empty", None)
+        return ChatResponseShape("unknown", None)
+    return ChatResponseShape("unknown", None)
diff --git a/src/xagent/web/api/agents.py b/src/xagent/web/api/agents.py
index 2178e9990b..a4c803aa11 100644
--- a/src/xagent/web/api/agents.py
+++ b/src/xagent/web/api/agents.py
@@ -17,6 +17,7 @@
     response_language_rules,
 )
 from ...core.agent.service import AgentService
+from ...core.agent.utils.llm_utils import unwrap_chat_text
 from ...core.agent.voice_policy import _VOICE_INSTRUCTIONS as _core_voice_instructions
 from ...core.agent.voice_policy import apply_output_voice, voice_from_preferences
 from ...core.memory.in_memory import InMemoryMemoryStore
@@ -575,10 +576,10 @@ async def optimize_instructions(
             ]
         )
 
-        if isinstance(response, dict) and "content" in response:
-            content = str(response["content"])
-        else:
-            content = response if isinstance(response, str) else str(response)
+        # Raises LLMNoTextContentError on a tool_call envelope or any other
+        # text-less shape rather than repr()ing it into a 200 response
+        # (#1714); the outer except turns that into a clear 500.
+        content = unwrap_chat_text(response)
 
         mismatch = detect_prose_script_mismatch(request.instructions, content)
         if mismatch is not None:
diff --git a/tests/core/agent/test_compact_llm_usage_contract.py b/tests/core/agent/test_compact_llm_usage_contract.py
new file mode 100644
index 0000000000..f13344c1d3
--- /dev/null
+++ b/tests/core/agent/test_compact_llm_usage_contract.py
@@ -0,0 +1,1302 @@
+"""Contract tests for compact-path token usage accounting (#520) and
+text-extraction of chat envelopes (#1714).
+
+Mock-boundary philosophy: the adapter under test runs its real code end to
+end; only the provider SDK's transport object is replaced (``AsyncOpenAI``
+for the OpenAI family, a duck-typed attribute stub for Zhipu/Gemini whose
+adapters read attributes rather than construct SDK types). Everything from
+the SDK response object inward -- envelope construction, the top-level
+``usage`` stamp, the runtime extractor, the context ledger and the
+contextvar ledger -- is production code, so a regression anywhere in that
+chain fails these tests. This mirrors test_vector_index_contract.py: the
+rest of the adapter suites mock the adapter's own return value, which is
+exactly why usage buried in ``raw.usage`` (#520) and repr()ed tool_call
+envelopes (#1714) survived unnoticed.
+"""
+
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import AsyncMock
+
+import pytest
+from openai.types.chat import ChatCompletion
+from openai.types.chat.chat_completion import Choice
+from openai.types.chat.chat_completion_message import ChatCompletionMessage
+from openai.types.completion_usage import CompletionUsage
+
+from xagent.core.agent import ExecutionContext, PatternRuntime, ReActPattern
+from xagent.core.agent.utils.context_builder import ContextBuilder, StepExecutionResult
+from xagent.core.agent.utils.llm_utils import unwrap_chat_text
+from xagent.core.model.chat.basic.deepseek import DeepSeekLLM
+from xagent.core.model.chat.basic.gemini import GeminiLLM
+from xagent.core.model.chat.basic.openai import OpenAILLM
+from xagent.core.model.chat.basic.zhipu import ZhipuLLM
+from xagent.core.model.chat.exceptions import (
+    LLMEmptyContentError,
+    LLMNoTextContentError,
+)
+from xagent.core.model.chat.token_context import get_token_usage, reset_token_usage
+
+PROMPT_TOKENS = 10
+COMPLETION_TOKENS = 5
+CACHED_TOKENS = 6
+COMPACT_SUMMARY = "summarized tool result"
+
+
+class FakeLLM:
+    """Queue-based fake for the main-pattern LLM (no usage accounting)."""
+
+    def __init__(self, responses: list[Any]) -> None:
+        self.responses = responses
+        self.calls: list[dict[str, Any]] = []
+
+    model_name = "fake-model"
+
+    async def chat(self, **kwargs: Any) -> Any:
+        self.calls.append(kwargs)
+        return self.responses.pop(0)
+
+
+class TraceEventRecorder:
+    def __init__(self) -> None:
+        self.events: list[dict[str, Any]] = []
+
+    async def trace_event(
+        self,
+        event_type: Any,
+        *,
+        task_id: str | None = None,
+        step_id: str | None = None,
+        data: dict[str, Any] | None = None,
+        **_: Any,
+    ) -> str:
+        self.events.append(
+            {
+                "event_type": getattr(event_type, "value", str(event_type)),
+                "task_id": task_id,
+                "step_id": step_id,
+                "data": data or {},
+            }
+        )
+        return str(len(self.events))
+
+
+def _snapshot_ledger() -> tuple[int, int, int]:
+    """Value snapshot of the contextvar ledger (the object itself is live)."""
+    ledger = get_token_usage()
+    return ledger.llm_calls, ledger.input_tokens, ledger.output_tokens
+
+
+def _openai_completion(content: str) -> ChatCompletion:
+    """A real SDK response carrying real ``CompletionUsage``."""
+    return ChatCompletion(
+        id="compact-contract-completion",
+        choices=[
+            Choice(
+                finish_reason="stop",
+                index=0,
+                message=ChatCompletionMessage(
+                    content=content,
+                    role="assistant",
+                    tool_calls=None,
+                ),
+            )
+        ],
+        created=1234567890,
+        model="gpt-4o-mini",
+        object="chat.completion",
+        usage=CompletionUsage(
+            prompt_tokens=PROMPT_TOKENS,
+            completion_tokens=COMPLETION_TOKENS,
+            total_tokens=PROMPT_TOKENS + COMPLETION_TOKENS,
+        ),
+    )
+
+
+def _zhipu_response(content: str) -> SimpleNamespace:
+    """Duck-typed stand-in for the zai-sdk response: the adapter only reads
+    attributes (``choices[0].message.content``, ``usage.prompt_tokens`` ...),
+    so a namespace at the transport boundary exercises the real adapter code."""
+    return SimpleNamespace(
+        choices=[
+            SimpleNamespace(
+                message=SimpleNamespace(content=content, tool_calls=None),
+                finish_reason="stop",
+            )
+        ],
+        usage=SimpleNamespace(
+            prompt_tokens=PROMPT_TOKENS,
+            completion_tokens=COMPLETION_TOKENS,
+        ),
+    )
+
+
+def _gemini_response(content: str) -> SimpleNamespace:
+    """Duck-typed stand-in for the google-genai response, same reasoning as
+    the Zhipu stub: attribute reads only (``usage_metadata``,
+    ``candidates[0].content.parts[*].text``)."""
+    return SimpleNamespace(
+        usage_metadata=SimpleNamespace(
+            prompt_token_count=PROMPT_TOKENS,
+            candidates_token_count=COMPLETION_TOKENS,
+            cached_content_token_count=0,
+        ),
+        candidates=[
+            SimpleNamespace(
+                content=SimpleNamespace(
+                    parts=[SimpleNamespace(text=content, function_call=None)]
+                )
+            )
+        ],
+    )
+
+
+def _react_compact_scenario() -> tuple[TraceEventRecorder, ExecutionContext]:
+    """threshold=1 context that forces one LLM compaction on the first turn."""
+    tracer = TraceEventRecorder()
+    context = ExecutionContext(execution_id="compact-usage-contract")
+    context.compact_config.threshold = 1
+    context.add_user_message("current request")
+    context.add_assistant_message(
+        "",
+        tool_calls=[
+            {"id": "call-1", "type": "function", "function": {"name": "read_file"}}
+        ],
+    )
+    context.add_tool_result("read_file", {"output": "x" * 200}, tool_call_id="call-1")
+    return tracer, context
+
+
+async def _run_compact_with(compact_llm: Any) -> tuple[TraceEventRecorder, Any]:
+    tracer, context = _react_compact_scenario()
+    runtime = PatternRuntime(tracer=tracer)
+    result = await ReActPattern(max_iterations=1).run(
+        context=context,
+        tools=[],
+        llm=FakeLLM([{"content": "done"}]),
+        compact_llm=compact_llm,
+        runtime=runtime,
+    )
+    assert result["success"] is True
+    return tracer, context
+
+
+def _assert_compact_usage_accounted(
+    tracer: TraceEventRecorder, context: Any, ledger_before: tuple[int, int, int]
+) -> None:
+    """One compact call must surface exactly once in every accounting view."""
+    compact_llm_events = [
+        event
+        for event in tracer.events
+        if event["event_type"] in {"action_start_llm", "action_end_llm"}
+        and event["data"].get("purpose") == "context_compaction"
+    ]
+    assert [event["event_type"] for event in compact_llm_events] == [
+        "action_start_llm",
+        "action_end_llm",
+    ]
+    end_data = compact_llm_events[1]["data"]
+    assert end_data["input_tokens"] == PROMPT_TOKENS
+    assert end_data["output_tokens"] == COMPLETION_TOKENS
+
+    # The execution-context ledger sees exactly one call with exact numbers.
+    assert context.get_total_token_usage() == {
+        "total": PROMPT_TOKENS + COMPLETION_TOKENS,
+        "input": PROMPT_TOKENS,
+        "output": COMPLETION_TOKENS,
+        "call_count": 1,
+    }
+
+    # The contextvar ledger also records exactly one call -- the adapter's own
+    # ``add_token_usage`` write; the stamp/extractor path must not double it.
+    calls_before, input_before, output_before = ledger_before
+    ledger = get_token_usage()
+    assert ledger.llm_calls - calls_before == 1
+    assert ledger.input_tokens - input_before == PROMPT_TOKENS
+    assert ledger.output_tokens - output_before == COMPLETION_TOKENS
+
+
+@pytest.mark.asyncio
+async def test_openai_chat_stamps_top_level_usage(mocker) -> None:
+    """The adapter envelope itself carries the stamp, not just the trace:
+    this is what lets consumers read usage without opening ``raw``."""
+    reset_token_usage()
+    llm = OpenAILLM(model_name="gpt-4o-mini", api_key="test-key")
+    mock_client = mocker.AsyncMock()
+    mock_client.chat.completions.create.return_value = _openai_completion(
+        COMPACT_SUMMARY
+    )
+    mocker.patch(
+        "xagent.core.model.chat.basic.openai.AsyncOpenAI",
+        return_value=mock_client,
+    )
+
+    response = await llm.chat([{"role": "user", "content": "hi"}])
+
+    assert response["type"] == "text"
+    assert response["usage"]["prompt_tokens"] == PROMPT_TOKENS
+    assert response["usage"]["completion_tokens"] == COMPLETION_TOKENS
+    assert response["usage"]["total_tokens"] == PROMPT_TOKENS + COMPLETION_TOKENS
+
+
+@pytest.mark.asyncio
+async def test_openai_tool_call_envelope_stamps_top_level_usage(mocker) -> None:
+    reset_token_usage()
+    llm = OpenAILLM(model_name="gpt-4o-mini", api_key="test-key")
+    mock_client = mocker.AsyncMock()
+    completion = ChatCompletion(
+        id="compact-contract-tool-call",
+        choices=[
+            Choice(
+                finish_reason="tool_calls",
+                index=0,
+                message=ChatCompletionMessage(
+                    content=None,
+                    role="assistant",
+                    tool_calls=[
+                        {
+                            "id": "call_1",
+                            "type": "function",
+                            "function": {
+                                "name": "read_file",
+                                "arguments": "{}",
+                            },
+                        }
+                    ],
+                ),
+            )
+        ],
+        created=1234567890,
+        model="gpt-4o-mini",
+        object="chat.completion",
+        usage=CompletionUsage(
+            prompt_tokens=PROMPT_TOKENS,
+            completion_tokens=COMPLETION_TOKENS,
+            total_tokens=PROMPT_TOKENS + COMPLETION_TOKENS,
+        ),
+    )
+    mock_client.chat.completions.create.return_value = completion
+    mocker.patch(
+        "xagent.core.model.chat.basic.openai.AsyncOpenAI",
+        return_value=mock_client,
+    )
+
+    response = await llm.chat([{"role": "user", "content": "hi"}])
+
+    assert response["type"] == "tool_call"
+    assert response["usage"]["prompt_tokens"] == PROMPT_TOKENS
+    assert response["usage"]["completion_tokens"] == COMPLETION_TOKENS
+
+
+@pytest.mark.asyncio
+async def test_openai_compact_usage_flows_to_trace_and_context(mocker) -> None:
+    reset_token_usage()
+    ledger_before = _snapshot_ledger()
+    llm = OpenAILLM(model_name="gpt-4o-mini", api_key="test-key")
+    mock_client = mocker.AsyncMock()
+    mock_client.chat.completions.create.return_value = _openai_completion(
+        COMPACT_SUMMARY
+    )
+    mocker.patch(
+        "xagent.core.model.chat.basic.openai.AsyncOpenAI",
+        return_value=mock_client,
+    )
+
+    tracer, context = await _run_compact_with(llm)
+
+    assert any(
+        COMPACT_SUMMARY in (message.content or "") for message in context.messages
+    )
+    _assert_compact_usage_accounted(tracer, context, ledger_before)
+
+
+@pytest.mark.asyncio
+async def test_deepseek_compact_usage_flows_to_trace_and_context(mocker) -> None:
+    reset_token_usage()
+    ledger_before = _snapshot_ledger()
+    llm = DeepSeekLLM(model_name="deepseek-v4-flash", api_key="test-key")
+    mock_client = mocker.AsyncMock()
+    mock_client.chat.completions.create.return_value = _openai_completion(
+        COMPACT_SUMMARY
+    )
+    mocker.patch(
+        "xagent.core.model.chat.basic.openai.AsyncOpenAI",
+        return_value=mock_client,
+    )
+
+    tracer, context = await _run_compact_with(llm)
+
+    assert any(
+        COMPACT_SUMMARY in (message.content or "") for message in context.messages
+    )
+    _assert_compact_usage_accounted(tracer, context, ledger_before)
+
+
+@pytest.mark.asyncio
+async def test_zhipu_compact_usage_flows_to_trace_and_context() -> None:
+    reset_token_usage()
+    ledger_before = _snapshot_ledger()
+    llm = ZhipuLLM(model_name="glm-4.5", api_key="test-key")
+    llm._client = SimpleNamespace(
+        chat=SimpleNamespace(
+            completions=SimpleNamespace(
+                create=lambda **kwargs: _zhipu_response(COMPACT_SUMMARY)
+            )
+        )
+    )
+
+    tracer, context = await _run_compact_with(llm)
+
+    assert any(
+        COMPACT_SUMMARY in (message.content or "") for message in context.messages
+    )
+    _assert_compact_usage_accounted(tracer, context, ledger_before)
+
+
+@pytest.mark.asyncio
+async def test_gemini_compact_usage_flows_to_trace_and_context() -> None:
+    reset_token_usage()
+    ledger_before = _snapshot_ledger()
+    llm = GeminiLLM(model_name="gemini-2.5-flash", api_key="test-key")
+    llm._client = SimpleNamespace(
+        aio=SimpleNamespace(
+            models=SimpleNamespace(
+                generate_content=AsyncMock(
+                    return_value=_gemini_response(COMPACT_SUMMARY)
+                )
+            )
+        )
+    )
+
+    tracer, context = await _run_compact_with(llm)
+
+    assert any(
+        COMPACT_SUMMARY in (message.content or "") for message in context.messages
+    )
+    _assert_compact_usage_accounted(tracer, context, ledger_before)
+
+
+@pytest.mark.asyncio
+async def test_compact_dependency_falls_back_when_compact_llm_returns_tool_call_envelope() -> (
+    None
+):
+    """#1714: a tool_call envelope from the compact model must never be
+    repr()ed into the rebuilt context; compaction fails explicitly and the
+    existing truncation/error fallbacks engage instead."""
+    tool_call_envelope = {
+        "type": "tool_call",
+        "tool_calls": [
+            {
+                "id": "call_1",
+                "type": "function",
+                "function": {"name": "read_file", "arguments": "{}"},
+            }
+        ],
+    }
+    compact_llm = FakeLLM([tool_call_envelope, tool_call_envelope])
+    builder = ContextBuilder(
+        llm=FakeLLM([]), compact_threshold=1, compact_llm=compact_llm
+    )
+    dep_result = StepExecutionResult(
+        step_id="dep-1",
+        messages=[
+            {"role": "user", "content": "u" * 100},
+            {"role": "assistant", "content": "a" * 100},
+        ],
+        final_result={},
+        agent_name="dep-agent",
+    )
+
+    messages = await builder.build_context_for_step(
+        step_name="target",
+        step_description="do something with the dependency output",
+        dependencies=["dep-1"],
+        dependency_results={"dep-1": dep_result},
+    )
+
+    # Both the individual and the whole-context compaction consulted the
+    # compact model and both had to fall back.
+    assert len(compact_llm.calls) == 2
+    rendered = json.dumps(messages, ensure_ascii=False)
+    assert "tool_calls" not in rendered
+    # The explicit truncation fallback keeps real history rather than a repr.
+    assert any("a" * 100 in m["content"] for m in messages)
+
+
+@pytest.mark.asyncio
+async def test_execution_context_compact_tolerates_tool_call_envelope() -> None:
+    """#1714 on the runtime path: a text-less envelope yields an empty summary,
+    so the LLM-summary strategy declines and truncation takes over."""
+    tracer, context = _react_compact_scenario()
+    runtime = PatternRuntime(tracer=tracer)
+    compact_llm = FakeLLM(
+        [
+            {
+                "type": "tool_call",
+                "tool_calls": [
+                    {
+                        "id": "call_1",
+                        "type": "function",
+                        "function": {"name": "read_file", "arguments": "{}"},
+                    }
+                ],
+            }
+        ]
+    )
+
+    result = await ReActPattern(max_iterations=1).run(
+        context=context,
+        tools=[],
+        llm=FakeLLM([{"content": "done"}]),
+        compact_llm=compact_llm,
+        runtime=runtime,
+    )
+
+    assert result["success"] is True
+    rendered = json.dumps(
+        [m.content for m in context.messages], ensure_ascii=False, default=str
+    )
+    assert "'tool_calls'" not in rendered
+    assert '"tool_calls"' not in rendered
+
+
+def test_deepseek_violation_rebuild_preserves_usage_stamp() -> None:
+    """normalize_deepseek_response rebuilds the envelope on a protocol
+    violation; the adapter's top-level usage stamp must survive the rebuild."""
+    from xagent.core.model.chat.basic.deepseek_tool_protocol import (
+        normalize_deepseek_response,
+    )
+
+    usage = {"prompt_tokens": PROMPT_TOKENS, "completion_tokens": COMPLETION_TOKENS}
+    response = {
+        "type": "text",
+        "content": "Sure: <｜｜DSML｜｜tool_calls>",
+        "usage": usage,
+    }
+    tool = {
+        "type": "function",
+        "function": {
+            "name": "final_answer",
+            "description": "Call final_answer.",
+            "parameters": {"type": "object", "properties": {}},
+        },
+    }
+
+    normalized = normalize_deepseek_response(response, tools=[tool])
+
+    assert normalized["type"] == "tool_protocol_error"
+    assert normalized["usage"] == usage
+
+
+class TestResolveUsagePayload:
+    """The runtime extractor reads top-level stamps first, then ``raw``."""
+
+    def setup_method(self) -> None:
+        self.runtime = PatternRuntime()
+
+    @pytest.mark.parametrize(
+        ("response", "expected"),
+        [
+            # OpenAI-family envelope without a stamp: usage lives in raw.
+            (
+                {
+                    "type": "text",
+                    "content": "x",
+                    "raw": {"usage": {"prompt_tokens": 10, "completion_tokens": 5}},
+                },
+                (10, 5),
+            ),
+            # Legacy plain-string response: fail open.
+            ("plain string", None),
+            # Explicitly null usage in raw: fail open, never raise.
+            (
+                {"type": "text", "content": "x", "raw": {"usage": None}},
+                None,
+            ),
+            # Top-level stamp (backwards compatible, and preferred over raw).
+            (
+                {
+                    "type": "text",
+                    "content": "x",
+                    "usage": {"prompt_tokens": 3, "completion_tokens": 2},
+                    "raw": {"usage": {"prompt_tokens": 10, "completion_tokens": 5}},
+                },
+                (3, 2),
+            ),
+            # Gemini-style usage_metadata one level down in raw.
+            (
+                {
+                    "type": "text",
+                    "content": "x",
+                    "raw": {
+                        "usage_metadata": {
+                            "prompt_token_count": 7,
+                            "candidates_token_count": 4,
+                        }
+                    },
+                },
+                (7, 4),
+            ),
+            # Zhipu tool_call fallback shape: raw is a plain stringified
+            # response -- fail open, never raise or probe the string.
+            (
+                {
+                    "type": "tool_call",
+                    "tool_calls": [],
+                    "raw": "ChatCompletion(choices=[...])",
+                },
+                None,
+            ),
+            # Top-level usage_metadata (legacy Gemini-style attribute shape
+            # preserved as-is on the response): the original key order applies.
+            (
+                {
+                    "type": "text",
+                    "content": "x",
+                    "usage_metadata": {
+                        "prompt_token_count": 8,
+                        "candidates_token_count": 3,
+                    },
+                },
+                (8, 3),
+            ),
+            # Usage present but all-zero: not a measurement, fail open.
+            (
+                {
+                    "type": "text",
+                    "content": "x",
+                    "usage": {"prompt_tokens": 0, "completion_tokens": 0},
+                },
+                None,
+            ),
+            # All-zero usage in raw behaves the same one level down.
+            (
+                {
+                    "type": "text",
+                    "content": "x",
+                    "raw": {"usage": {"prompt_tokens": 0, "completion_tokens": 0}},
+                },
+                None,
+            ),
+        ],
+    )
+    def test_extract_token_usage_shapes(self, response: Any, expected: Any) -> None:
+        assert self.runtime._extract_token_usage(response) == expected
+
+    def test_extract_cached_tokens_fails_open_on_string_raw(self) -> None:
+        response = {
+            "type": "tool_call",
+            "tool_calls": [],
+            "raw": "ChatCompletion(choices=[...])",
+        }
+        assert self.runtime._extract_cached_tokens(response) == 0
+
+    def test_extract_cached_tokens_reads_raw_usage(self) -> None:
+        response = {
+            "type": "text",
+            "content": "x",
+            "raw": {
+                "usage": {
+                    "prompt_tokens": 10,
+                    "completion_tokens": 5,
+                    "prompt_tokens_details": {"cached_tokens": 6},
+                }
+            },
+        }
+        assert self.runtime._extract_cached_tokens(response) == 6
+        assert self.runtime._extract_cached_tokens("plain string") == 0
+
+
+class TestStrictUsageIntCoercion:
+    """Usage counters are billing inputs, so coercion must be strict:
+    bools, NaN/inf, negatives, non-integral floats, and numeric strings are
+    rejected rather than silently truncated or crashed on, and an invalid
+    earlier alias must not shadow a valid later candidate."""
+
+    def setup_method(self) -> None:
+        self.runtime = PatternRuntime()
+
+    @pytest.mark.parametrize(
+        "bad",
+        [
+            float("nan"),
+            float("inf"),
+            float("-inf"),
+            True,
+            False,
+            -5,
+            10.5,
+            "10",
+            None,
+        ],
+    )
+    def test_first_int_rejects_invalid_values(self, bad: Any) -> None:
+        assert self.runtime._first_int({"prompt_tokens": bad}, ("prompt_tokens",)) == 0
+
+    @pytest.mark.parametrize(
+        ("good", "expected"),
+        [(10, 10), (0, 0), (10.0, 10), (2**40, 2**40)],
+    )
+    def test_first_int_accepts_valid_values(self, good: Any, expected: int) -> None:
+        assert (
+            self.runtime._first_int({"prompt_tokens": good}, ("prompt_tokens",))
+            == expected
+        )
+
+    def test_invalid_first_alias_falls_through_to_valid_candidate(self) -> None:
+        usage = {
+            "prompt_tokens": float("nan"),
+            "input_tokens": 7,
+            "completion_tokens": 3,
+        }
+        response = {"type": "text", "content": "x", "usage": usage}
+        assert self.runtime._extract_token_usage(response) == (7, 3)
+
+    def test_extract_token_usage_rejects_bool_and_negative(self) -> None:
+        response = {
+            "type": "text",
+            "content": "x",
+            "usage": {"prompt_tokens": True, "completion_tokens": -5},
+        }
+        assert self.runtime._extract_token_usage(response) is None
+
+    def test_extract_cached_tokens_rejects_non_finite(self) -> None:
+        response = {
+            "type": "text",
+            "content": "x",
+            "usage": {"cached_input_tokens": float("inf")},
+        }
+        assert self.runtime._extract_cached_tokens(response) == 0
+
+
+class TestUsageStampCachedTokens:
+    """The stamp must carry cache metrics so ``_extract_cached_tokens`` sees
+    prompt-cache hits on non-streaming calls (review findings on PR #1787).
+    Cached keys are stamped only when non-zero, so a default 0 never shadows
+    fallback fields downstream."""
+
+    @pytest.mark.asyncio
+    async def test_zhipu_chat_stamp_includes_cached_tokens(self) -> None:
+        reset_token_usage()
+        llm = ZhipuLLM(model_name="glm-4.5", api_key="test-key")
+        response = _zhipu_response("cached reply")
+        response.usage.prompt_tokens_details = SimpleNamespace(cached_tokens=4)
+        llm._client = SimpleNamespace(
+            chat=SimpleNamespace(
+                completions=SimpleNamespace(create=lambda **kwargs: response)
+            )
+        )
+
+        result = await llm.chat(messages=[{"role": "user", "content": "hi"}])
+
+        assert result["usage"]["cached_input_tokens"] == 4
+
+    @pytest.mark.asyncio
+    async def test_zhipu_chat_stamp_omits_zero_cached_tokens(self) -> None:
+        reset_token_usage()
+        llm = ZhipuLLM(model_name="glm-4.5", api_key="test-key")
+        llm._client = SimpleNamespace(
+            chat=SimpleNamespace(
+                completions=SimpleNamespace(
+                    create=lambda **kwargs: _zhipu_response("plain reply")
+                )
+            )
+        )
+
+        result = await llm.chat(messages=[{"role": "user", "content": "hi"}])
+
+        assert "cached_input_tokens" not in result["usage"]
+
+    @pytest.mark.asyncio
+    async def test_gemini_chat_stamp_includes_cached_tokens(self) -> None:
+        reset_token_usage()
+        llm = GeminiLLM(model_name="gemini-2.5-flash", api_key="test-key")
+        response = _gemini_response("cached reply")
+        response.usage_metadata.cached_content_token_count = 4
+        llm._client = SimpleNamespace(
+            aio=SimpleNamespace(
+                models=SimpleNamespace(
+                    generate_content=AsyncMock(return_value=response)
+                )
+            )
+        )
+
+        result = await llm.chat(messages=[{"role": "user", "content": "hi"}])
+
+        assert result["usage"]["cached_input_tokens"] == 4
+
+
+class TestUnwrapChatText:
+    """``unwrap_chat_text`` distinguishes "no text shape" from "empty text"
+    so retry/fallback semantics stay aligned with the adapters' own classes."""
+
+    def test_plain_string_passthrough(self) -> None:
+        assert unwrap_chat_text("hello") == "hello"
+
+    def test_envelope_content(self) -> None:
+        assert unwrap_chat_text({"type": "text", "content": "hi"}) == "hi"
+
+    @pytest.mark.parametrize("content", ["", "   \n\t"])
+    def test_empty_envelope_content_raises_empty_content(self, content: str) -> None:
+        with pytest.raises(LLMEmptyContentError):
+            unwrap_chat_text({"type": "text", "content": content})
+
+    def test_tool_call_envelope_raises_no_text_content(self) -> None:
+        with pytest.raises(LLMNoTextContentError):
+            unwrap_chat_text({"type": "tool_call", "tool_calls": []})
+
+    def test_unrecognized_shape_raises_no_text_content(self) -> None:
+        with pytest.raises(LLMNoTextContentError):
+            unwrap_chat_text(42)
+
+    def test_empty_plain_string_raises_empty_content(self) -> None:
+        """An empty legacy plain string is the same transient "no content"
+        as an empty envelope -- ``classify_chat_response`` is the single
+        source for that distinction."""
+        with pytest.raises(LLMEmptyContentError):
+            unwrap_chat_text("")
+
+
+@pytest.mark.asyncio
+async def test_openai_compact_cached_tokens_flow_to_trace_event(mocker) -> None:
+    """End-to-end cache loop: a real CompletionUsage carrying
+    ``prompt_tokens_details.cached_tokens`` must surface as
+    ``cached_input_tokens`` on the compact trace event -- the stamp alone is
+    not enough, the extractor must read it back."""
+    from openai.types.completion_usage import PromptTokensDetails
+
+    reset_token_usage()
+    llm = OpenAILLM(model_name="gpt-4o-mini", api_key="test-key")
+    completion = _openai_completion(COMPACT_SUMMARY)
+    completion.usage = CompletionUsage(
+        prompt_tokens=PROMPT_TOKENS,
+        completion_tokens=COMPLETION_TOKENS,
+        total_tokens=PROMPT_TOKENS + COMPLETION_TOKENS,
+        prompt_tokens_details=PromptTokensDetails(cached_tokens=CACHED_TOKENS),
+    )
+    mock_client = mocker.AsyncMock()
+    mock_client.chat.completions.create.return_value = completion
+    mocker.patch(
+        "xagent.core.model.chat.basic.openai.AsyncOpenAI",
+        return_value=mock_client,
+    )
+
+    tracer, context = await _run_compact_with(llm)
+
+    compact_end = next(
+        event
+        for event in tracer.events
+        if event["event_type"] == "action_end_llm"
+        and event["data"].get("purpose") == "context_compaction"
+    )
+    assert compact_end["data"]["cached_input_tokens"] == CACHED_TOKENS
+    assert compact_end["data"]["input_tokens"] == PROMPT_TOKENS
+    assert context.get_total_token_usage()["total"] == (
+        PROMPT_TOKENS + COMPLETION_TOKENS
+    )
+
+
+class TestSyntheticUsageFreshnessBaseline:
+    """A synthetic (context-compaction) usage record must never become the
+    freshness baseline of ``_get_total_tokens``: when an LLM compaction
+    declines (e.g. the compact model returns a tool_call envelope) the
+    messages are unchanged, so the record's fingerprint still matches and a
+    small compact-prompt token count would otherwise be mistaken for the
+    live context size -- suppressing the truncation fallback and letting
+    oversized history flow to the main model."""
+
+    def test_synthetic_record_does_not_hijack_freshness_baseline(self) -> None:
+        context = ExecutionContext()
+        context.compact_config.threshold = 100
+        context.add_user_message("u" * 2000)  # ~500 est tokens, over threshold
+        context.record_llm_usage(
+            input_tokens=10,
+            output_tokens=5,
+            synthetic_purpose="context_compaction",
+        )
+        # The fingerprint matches (messages unchanged) and 10 < threshold,
+        # but the record is synthetic: fall back to the char estimate.
+        assert context.estimate_context_tokens() > 100
+
+    def test_real_record_still_serves_as_freshness_baseline(self) -> None:
+        context = ExecutionContext()
+        context.add_user_message("u" * 2000)
+        context.record_llm_usage(input_tokens=42, output_tokens=5)
+        assert context.estimate_context_tokens() == 42
+
+    def test_synthetic_record_after_real_record_keeps_real_baseline(self) -> None:
+        context = ExecutionContext()
+        context.add_user_message("u" * 2000)
+        context.record_llm_usage(input_tokens=42, output_tokens=5)
+        context.record_llm_usage(
+            input_tokens=10,
+            output_tokens=5,
+            synthetic_purpose="context_compaction",
+        )
+        assert context.estimate_context_tokens() == 42
+
+    def test_synthetic_purpose_survives_checkpoint_roundtrip(self) -> None:
+        context = ExecutionContext()
+        context.add_user_message("hi")
+        context.record_llm_usage(
+            input_tokens=10,
+            output_tokens=5,
+            synthetic_purpose="context_compaction",
+        )
+        context.record_llm_usage(input_tokens=7, output_tokens=3)
+
+        restored = ExecutionContext.from_dict(context.to_dict())
+
+        assert [call.synthetic_purpose for call in restored.llm_calls] == [
+            "context_compaction",
+            None,
+        ]
+
+    def test_old_checkpoint_without_synthetic_purpose_defaults_none(self) -> None:
+        context = ExecutionContext()
+        context.add_user_message("hi")
+        context.record_llm_usage(input_tokens=10, output_tokens=5)
+        data = context.to_dict()
+        for call in data["llm_calls"]:
+            call.pop("synthetic_purpose", None)
+
+        restored = ExecutionContext.from_dict(data)
+
+        assert [call.synthetic_purpose for call in restored.llm_calls] == [None]
+
+    @pytest.mark.asyncio
+    async def test_failed_llm_compaction_still_truncates_oversized_history(
+        self,
+    ) -> None:
+        """Reviewer scenario: oversized history (chars/4 > threshold) and a
+        compact model returning a *stamped* tool_call envelope whose
+        prompt_tokens < threshold. LLM compaction declines, and the
+        truncation fallback must really truncate; the compact usage is
+        recorded exactly once and must not feed the freshness estimate."""
+        tracer = TraceEventRecorder()
+        context = ExecutionContext(execution_id="compact-freshness-contract")
+        context.compact_config.threshold = 100
+        context.compact_config.max_messages = 1
+        context.add_user_message("current request")
+        context.add_assistant_message(
+            "",
+            tool_calls=[
+                {"id": "call-1", "type": "function", "function": {"name": "read_file"}}
+            ],
+        )
+        context.add_tool_result(
+            "read_file", {"output": "x" * 2000}, tool_call_id="call-1"
+        )
+        assert context.estimate_context_tokens() > 100
+
+        compact_llm = FakeLLM(
+            [
+                {
+                    "type": "tool_call",
+                    "tool_calls": [
+                        {
+                            "id": "call_9",
+                            "type": "function",
+                            "function": {"name": "read_file", "arguments": "{}"},
+                        }
+                    ],
+                    "usage": {"prompt_tokens": 10, "completion_tokens": 5},
+                }
+            ]
+        )
+
+        result = await ReActPattern(max_iterations=1).run(
+            context=context,
+            tools=[],
+            llm=FakeLLM([{"content": "done"}]),
+            compact_llm=compact_llm,
+            runtime=PatternRuntime(tracer=tracer),
+        )
+
+        assert result["success"] is True
+        # Truncation really happened: the declined LLM compaction left the
+        # messages untouched, so only the fallback's compact event proves the
+        # small compact-prompt count did not hijack the estimate. (Message
+        # counts alone cannot show it: the pattern appends afterwards.)
+        compact_end = next(
+            (
+                event
+                for event in tracer.events
+                if event["event_type"] == "action_end_compact"
+            ),
+            None,
+        )
+        assert compact_end is not None, "truncation fallback did not fire"
+        assert compact_end["data"]["strategy"] == "truncate"
+        assert compact_end["data"]["removed_count"] >= 1
+        # The compact call is billed exactly once, marked synthetic.
+        assert context.get_total_token_usage() == {
+            "total": 15,
+            "input": 10,
+            "output": 5,
+            "call_count": 1,
+        }
+        assert context.llm_calls[-1].synthetic_purpose == "context_compaction"
+
+    @pytest.mark.asyncio
+    async def test_failed_llm_compaction_empty_envelope_still_truncates(self) -> None:
+        """Empty-text-envelope variant of the freshness scenario: the empty
+        summary declines the LLM compaction and the fallback must truncate
+        regardless of the small stamped usage."""
+        tracer = TraceEventRecorder()
+        context = ExecutionContext(execution_id="compact-freshness-empty")
+        context.compact_config.threshold = 100
+        context.compact_config.max_messages = 1
+        context.add_user_message("current request")
+        context.add_assistant_message(
+            "",
+            tool_calls=[
+                {"id": "call-1", "type": "function", "function": {"name": "read_file"}}
+            ],
+        )
+        context.add_tool_result(
+            "read_file", {"output": "x" * 2000}, tool_call_id="call-1"
+        )
+        assert context.estimate_context_tokens() > 100
+
+        compact_llm = FakeLLM(
+            [
+                {
+                    "type": "text",
+                    "content": "",
+                    "usage": {"prompt_tokens": 10, "completion_tokens": 5},
+                }
+            ]
+        )
+
+        result = await ReActPattern(max_iterations=1).run(
+            context=context,
+            tools=[],
+            llm=FakeLLM([{"content": "done"}]),
+            compact_llm=compact_llm,
+            runtime=PatternRuntime(tracer=tracer),
+        )
+
+        assert result["success"] is True
+        compact_end = next(
+            (
+                event
+                for event in tracer.events
+                if event["event_type"] == "action_end_compact"
+            ),
+            None,
+        )
+        assert compact_end is not None, "truncation fallback did not fire"
+        assert compact_end["data"]["strategy"] == "truncate"
+        assert compact_end["data"]["removed_count"] >= 1
+        assert context.get_total_token_usage()["call_count"] == 1
+        assert context.llm_calls[-1].synthetic_purpose == "context_compaction"
+
+
+class TestVisionChatUsageStamp:
+    """``vision_chat`` envelopes carry the same top-level usage stamp as
+    ``chat()`` -- image inputs account tokens identically (#520)."""
+
+    @pytest.mark.asyncio
+    async def test_openai_vision_chat_stamps_top_level_usage(self, mocker) -> None:
+        llm = OpenAILLM(
+            model_name="gpt-4o-mini",
+            api_key="test-key",
+            abilities=["chat", "vision"],
+        )
+        mock_client = mocker.AsyncMock()
+        mock_client.chat.completions.create.return_value = _openai_completion(
+            "A diagram on a whiteboard."
+        )
+        mocker.patch(
+            "xagent.core.model.chat.basic.openai.AsyncOpenAI",
+            return_value=mock_client,
+        )
+        messages = [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "text", "text": "Describe this image."},
+                    {
+                        "type": "image_url",
+                        "image_url": {
+                            "url": "data:image/jpeg;base64,ZmFrZV9pbWFnZV9kYXRh"
+                        },
+                    },
+                ],
+            }
+        ]
+
+        response = await llm.vision_chat(messages)
+
+        assert response["type"] == "text"
+        assert response["content"] == "A diagram on a whiteboard."
+        assert response["usage"]["prompt_tokens"] == PROMPT_TOKENS
+        assert response["usage"]["completion_tokens"] == COMPLETION_TOKENS
+
+    @pytest.mark.asyncio
+    async def test_zhipu_vision_chat_stamps_usage_with_cached_tokens(self) -> None:
+        llm = ZhipuLLM(model_name="glm-4.5v", api_key="test-key")
+        response = _zhipu_response("A flowchart with three boxes.")
+        response.usage.prompt_tokens_details = SimpleNamespace(
+            cached_tokens=CACHED_TOKENS
+        )
+        llm._client = SimpleNamespace(
+            chat=SimpleNamespace(
+                completions=SimpleNamespace(create=lambda **kwargs: response)
+            )
+        )
+        messages = [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "text", "text": "Describe this image."},
+                    {
+                        "type": "image_url",
+                        "image_url": {
+                            "url": "data:image/jpeg;base64,ZmFrZV9pbWFnZV9kYXRh"
+                        },
+                    },
+                ],
+            }
+        ]
+
+        result = await llm.vision_chat(messages=messages)
+
+        assert result["type"] == "text"
+        assert result["content"] == "A flowchart with three boxes."
+        assert result["usage"] == {
+            "prompt_tokens": PROMPT_TOKENS,
+            "completion_tokens": COMPLETION_TOKENS,
+            "cached_input_tokens": CACHED_TOKENS,
+        }
+
+
+@pytest.mark.asyncio
+async def test_openai_reasoning_truncation_branch_stamps_usage(mocker) -> None:
+    """The reasoning-content early return (content empty, finish_reason
+    "length") is a separate result-construction branch; its envelope must
+    carry the stamp too."""
+    llm = OpenAILLM(model_name="gpt-4o-mini", api_key="test-key")
+    completion = ChatCompletion(
+        id="compact-contract-reasoning",
+        choices=[
+            Choice(
+                finish_reason="length",
+                index=0,
+                message=ChatCompletionMessage(
+                    content="",
+                    role="assistant",
+                    tool_calls=None,
+                    reasoning_content="partial reasoning trace",  # type: ignore[call-arg]
+                ),
+            )
+        ],
+        created=1234567890,
+        model="gpt-4o-mini",
+        object="chat.completion",
+        usage=CompletionUsage(
+            prompt_tokens=PROMPT_TOKENS,
+            completion_tokens=COMPLETION_TOKENS,
+            total_tokens=PROMPT_TOKENS + COMPLETION_TOKENS,
+        ),
+    )
+    mock_client = mocker.AsyncMock()
+    mock_client.chat.completions.create.return_value = completion
+    mocker.patch(
+        "xagent.core.model.chat.basic.openai.AsyncOpenAI",
+        return_value=mock_client,
+    )
+
+    response = await llm.chat([{"role": "user", "content": "hi"}])
+
+    assert response["type"] == "text"
+    assert response["content"] == "partial reasoning trace"
+    assert response["usage"]["prompt_tokens"] == PROMPT_TOKENS
+    assert response["usage"]["completion_tokens"] == COMPLETION_TOKENS
+
+
+@pytest.mark.asyncio
+async def test_compact_usage_survives_checkpoint_roundtrip(mocker) -> None:
+    """Invariant I5: usage recorded during compaction must survive
+    ``to_dict``/``from_dict`` unchanged -- checkpoints are how executions
+    resume, and a lossy roundtrip would silently drop billed tokens."""
+    reset_token_usage()
+    llm = OpenAILLM(model_name="gpt-4o-mini", api_key="test-key")
+    mock_client = mocker.AsyncMock()
+    mock_client.chat.completions.create.return_value = _openai_completion(
+        COMPACT_SUMMARY
+    )
+    mocker.patch(
+        "xagent.core.model.chat.basic.openai.AsyncOpenAI",
+        return_value=mock_client,
+    )
+
+    _, context = await _run_compact_with(llm)
+
+    expected = {
+        "total": PROMPT_TOKENS + COMPLETION_TOKENS,
+        "input": PROMPT_TOKENS,
+        "output": COMPLETION_TOKENS,
+        "call_count": 1,
+    }
+    assert context.get_total_token_usage() == expected
+
+    restored = ExecutionContext.from_dict(context.to_dict())
+
+    assert restored.get_total_token_usage() == expected
+
+
+@pytest.mark.asyncio
+async def test_compact_estimate_reflects_current_messages_not_stale_record(
+    mocker,
+) -> None:
+    """Invariant I4 + the implicit ordering contract in
+    ``PatternRuntime.compact_context_if_needed``: ``on_llm_end`` (which calls
+    ``record_llm_usage``) runs *before* ``compact_with_llm_response`` rewrites
+    ``context.messages``. If someone swaps them, the record's
+    ``prompt_message_count``/``prompt_content_chars`` describe the *rewritten*
+    message list, the staleness check in ``_get_total_tokens`` then passes,
+    and ``estimate_context_tokens`` silently reports the compact call's
+    prompt tokens (~10) instead of estimating the summary text (~hundreds).
+    """
+    reset_token_usage()
+    llm = OpenAILLM(model_name="gpt-4o-mini", api_key="test-key")
+    mock_client = mocker.AsyncMock()
+    mock_client.chat.completions.create.return_value = _openai_completion(
+        COMPACT_SUMMARY
+    )
+    mocker.patch(
+        "xagent.core.model.chat.basic.openai.AsyncOpenAI",
+        return_value=mock_client,
+    )
+
+    tracer = TraceEventRecorder()
+    context = ExecutionContext(execution_id="compact-estimate-contract")
+    context.compact_config.threshold = 1
+    context.add_user_message("current request")
+    context.add_assistant_message(
+        "",
+        tool_calls=[
+            {"id": "call-1", "type": "function", "function": {"name": "read_file"}}
+        ],
+    )
+    # Large enough that the summary is genuinely smaller than the original.
+    context.add_tool_result("read_file", {"output": "x" * 4000}, tool_call_id="call-1")
+    estimate_before = context.estimate_context_tokens()
+    message_count_before = len(context.messages)
+
+    result = await ReActPattern(max_iterations=1).run(
+        context=context,
+        tools=[],
+        llm=FakeLLM([{"content": "done"}]),
+        compact_llm=llm,
+        runtime=PatternRuntime(tracer=tracer),
+    )
+
+    assert result["success"] is True
+    # Compaction really shrank the message list.
+    compact_end = next(
+        event for event in tracer.events if event["event_type"] == "action_end_compact"
+    )
+    assert compact_end["data"]["original_count"] == message_count_before
+    assert compact_end["data"]["final_count"] < message_count_before
+
+    # The post-compaction estimate is a live estimate of the current
+    # messages: smaller than before, positive, and far above the compact
+    # call's own prompt-token count (which a swapped record would leak).
+    estimate_after = context.estimate_context_tokens()
+    assert 0 < estimate_after < estimate_before
+    assert estimate_after > 3 * PROMPT_TOKENS
+
+
+@pytest.mark.asyncio
+async def test_compact_dependency_falls_back_when_compact_llm_returns_empty_content() -> (
+    None
+):
+    """#1714 empty-content path: an empty text envelope now raises
+    ``LLMEmptyContentError`` from ``unwrap_chat_text``; compaction must fail
+    explicitly and engage the truncation fallback, with no residue of the
+    envelope in the rebuilt context."""
+    empty_envelope = {"type": "text", "content": ""}
+    compact_llm = FakeLLM([empty_envelope, empty_envelope])
+    builder = ContextBuilder(
+        llm=FakeLLM([]), compact_threshold=1, compact_llm=compact_llm
+    )
+    dep_result = StepExecutionResult(
+        step_id="dep-1",
+        messages=[
+            {"role": "user", "content": "u" * 100},
+            {"role": "assistant", "content": "a" * 100},
+        ],
+        final_result={},
+        agent_name="dep-agent",
+    )
+
+    messages = await builder.build_context_for_step(
+        step_name="target",
+        step_description="do something with the dependency output",
+        dependencies=["dep-1"],
+        dependency_results={"dep-1": dep_result},
+    )
+
+    # Both compaction levels consulted the compact model and both fell back.
+    assert len(compact_llm.calls) == 2
+    rendered = json.dumps(messages, ensure_ascii=False)
+    assert "'content': ''" not in rendered
+    # The explicit truncation fallback keeps real history rather than a repr.
+    assert any("a" * 100 in m["content"] for m in messages)
+
+
+@pytest.mark.asyncio
+async def test_zhipu_tool_call_envelope_uses_model_dump_raw() -> None:
+    """D8: the real zai-sdk response is a pydantic model, so the production
+    ``raw`` always goes through ``response.model_dump()`` -- the stand-in
+    must exercise that branch, not the ``str(response)`` fallback."""
+
+    class _ZaiLikeResponse(SimpleNamespace):
+        def model_dump(self) -> dict[str, Any]:
+            return {"id": "zai-1", "object": "chat.completion"}
+
+    tool_call = SimpleNamespace(
+        id="call_1",
+        type="function",
+        function=SimpleNamespace(name="calculator", arguments='{"expression": "2+2"}'),
+    )
+    response = _ZaiLikeResponse(
+        choices=[
+            SimpleNamespace(
+                message=SimpleNamespace(content=None, tool_calls=[tool_call]),
+                finish_reason="tool_calls",
+            )
+        ],
+        usage=SimpleNamespace(prompt_tokens=PROMPT_TOKENS, completion_tokens=5),
+    )
+    reset_token_usage()
+    llm = ZhipuLLM(model_name="glm-4.5", api_key="test-key")
+    llm._client = SimpleNamespace(
+        chat=SimpleNamespace(
+            completions=SimpleNamespace(create=lambda **kwargs: response)
+        )
+    )
+
+    result = await llm.chat(
+        [{"role": "user", "content": "2+2?"}],
+        tools=[
+            {
+                "type": "function",
+                "function": {
+                    "name": "calculator",
+                    "description": "calc",
+                    "parameters": {"type": "object", "properties": {}},
+                },
+            }
+        ],
+    )
+
+    assert result["type"] == "tool_call"
+    assert result["raw"] == {"id": "zai-1", "object": "chat.completion"}
+    assert result["usage"]["prompt_tokens"] == PROMPT_TOKENS
diff --git a/tests/core/agent/test_react.py b/tests/core/agent/test_react.py
index 85b8d1426c..ea723512fb 100644
--- a/tests/core/agent/test_react.py
+++ b/tests/core/agent/test_react.py
@@ -8824,3 +8824,33 @@ async def test_react_summarizes_with_the_main_model_when_no_compact_model() -> N
     # One routing decision for the whole turn, taken on the conversation.
     assert len(route_prompts) == 1
     assert "Conversation history to compact" not in route_prompts[0]
+
+
+@pytest.mark.asyncio
+async def test_react_finalize_fallback_never_leaks_envelope_repr() -> None:
+    """C3: when the model's text unwraps to an empty final answer, the
+    finalize fallback must surface the envelope's *text* -- never the whole
+    envelope, whose repr would land in the user-visible transcript."""
+    context = ExecutionContext(execution_id="react-envelope-fallback")
+    context.add_user_message("hi")
+    runtime = PatternRuntime(tracer=TraceEventRecorder())
+    json_text = '{"action": "final_answer", "action_input": ""}'
+
+    result = await ReActPattern(max_iterations=1).run(
+        context=context,
+        tools=[],
+        llm=FakeLLM(
+            [
+                {
+                    "type": "text",
+                    "content": json_text,
+                    "usage": {"prompt_tokens": 10, "completion_tokens": 5},
+                }
+            ]
+        ),
+        runtime=runtime,
+    )
+
+    assert result["success"] is True
+    assert result["output"] == json_text
+    assert "'type': 'text'" not in str(result)
diff --git a/tests/core/model/chat/basic/test_claude.py b/tests/core/model/chat/basic/test_claude.py
index 6e26ed7f2b..8c6b0ef42c 100644
--- a/tests/core/model/chat/basic/test_claude.py
+++ b/tests/core/model/chat/basic/test_claude.py
@@ -163,9 +163,10 @@ async def test_basic_chat_completion(self, llm, mocker):
 
         response = await llm.chat(messages)
 
-        # Verify response is a non-empty string
-        assert isinstance(response, str)
-        assert response == "Hello World"
+        # Verify response is a text envelope carrying the usage stamp
+        assert response["type"] == "text"
+        assert response["content"] == "Hello World"
+        assert response["usage"] == {"prompt_tokens": 10, "completion_tokens": 5}
         print(f"Basic chat response: {response}")
 
         # Verify the API was called with correct parameters
@@ -175,6 +176,43 @@ async def test_basic_chat_completion(self, llm, mocker):
         assert "temperature" in call_args.kwargs
         assert call_args.kwargs["temperature"] == 0.7
 
+    @pytest.mark.asyncio
+    async def test_usage_stamp_includes_cache_metrics(self, llm, mocker):
+        """Cache read/write tokens ride the usage stamp so PatternRuntime's
+        _extract_cached_tokens sees prompt-cache hits on non-streaming calls."""
+        mock_client = mocker.AsyncMock()
+
+        mock_text_block = mocker.Mock()
+        mock_text_block.type = "text"
+        mock_text_block.text = "Cached"
+
+        mock_usage = mocker.Mock()
+        mock_usage.input_tokens = 10
+        mock_usage.output_tokens = 5
+        mock_usage.cache_read_input_tokens = 4
+        mock_usage.cache_creation_input_tokens = 2
+
+        mock_response = mocker.Mock()
+        mock_response.stop_reason = "stop"
+        mock_response.content = [mock_text_block]
+        mock_response.usage = mock_usage
+
+        mock_client.messages.create.return_value = mock_response
+        mocker.patch(
+            "xagent.core.model.chat.basic.claude.AsyncAnthropic",
+            return_value=mock_client,
+        )
+
+        response = await llm.chat([{"role": "user", "content": "hi"}])
+
+        # _anthropic_input_usage re-adds cache tokens into the input total.
+        assert response["usage"] == {
+            "prompt_tokens": 16,
+            "completion_tokens": 5,
+            "cached_input_tokens": 4,
+            "cache_write_input_tokens": 2,
+        }
+
     @pytest.mark.asyncio
     async def test_tool_calling(self, llm, mocker):
         """Test tool calling functionality."""
@@ -235,6 +273,8 @@ async def test_tool_calling(self, llm, mocker):
         assert isinstance(response, dict)
         assert response.get("type") == "tool_call"
         assert "tool_calls" in response
+        # The usage stamp rides on tool_call envelopes too.
+        assert response["usage"] == {"prompt_tokens": 20, "completion_tokens": 10}
 
         tool_calls = response["tool_calls"]
         assert len(tool_calls) > 0
@@ -387,8 +427,8 @@ async def test_context_manager(self, claude_llm_config, mocker):
             messages = [{"role": "user", "content": "Say 'test'"}]
             response = await ctx_llm.chat(messages)
 
-            assert isinstance(response, str)
-            assert response == "test"
+            assert response["type"] == "text"
+            assert response["content"] == "test"
             print(f"Context manager response: {response}")
 
         # Verify the client was properly closed
@@ -466,8 +506,8 @@ async def test_custom_parameters(self, llm, mocker):
             max_tokens=50,  # Limit response length
         )
 
-        assert isinstance(response, str)
-        assert response == "Test response"
+        assert response["type"] == "text"
+        assert response["content"] == "Test response"
         print(f"Custom parameters response: {response}")
 
         # Verify custom parameters were passed
@@ -565,7 +605,7 @@ async def test_thinking_mode_enabled(self, llm, mocker):
             messages, thinking={"type": "enabled", "budget_tokens": 20480}
         )
 
-        assert isinstance(response, str)
+        assert response["type"] == "text"
 
         # Verify thinking mode was passed
         call_args = mock_client.messages.create.call_args
@@ -604,7 +644,7 @@ async def test_thinking_mode_disabled(self, llm, mocker):
         # Test with thinking mode explicitly disabled
         response = await llm.chat(messages, thinking={"type": "disabled"})
 
-        assert isinstance(response, str)
+        assert response["type"] == "text"
 
         # Verify thinking mode was set to disabled
         call_args = mock_client.messages.create.call_args
@@ -654,8 +694,8 @@ async def test_vision_chat(self, llm, mocker):
 
         response = await llm.vision_chat(messages)
 
-        assert isinstance(response, str)
-        assert response == "I see an image"
+        assert response["type"] == "text"
+        assert response["content"] == "I see an image"
         print(f"Vision chat response: {response}")
 
     @pytest.mark.asyncio
@@ -1012,10 +1052,10 @@ async def test_output_config_json_schema(self, llm, mocker):
 
         response = await llm.chat(messages, output_config=output_config)
 
-        assert isinstance(response, str)
-        # Verify the response contains the expected JSON
-        assert "joke" in response
-        assert "punchline" in response
+        assert response["type"] == "text"
+        # Verify the response content contains the expected JSON
+        assert "joke" in response["content"]
+        assert "punchline" in response["content"]
 
         # Verify the API was called with output_config
         mock_client.messages.create.assert_called_once()
diff --git a/tests/core/model/chat/basic/test_gemini_sdk.py b/tests/core/model/chat/basic/test_gemini_sdk.py
index b0bb8b0187..e5ca63cf2c 100644
--- a/tests/core/model/chat/basic/test_gemini_sdk.py
+++ b/tests/core/model/chat/basic/test_gemini_sdk.py
@@ -90,9 +90,10 @@ async def mock_generate_content(*args, **kwargs):
 
         response = await llm.chat(messages)
 
-        # Verify response
-        assert isinstance(response, str)
-        assert response == "Hello World"
+        # Verify response: a text envelope carrying the usage stamp
+        assert response["type"] == "text"
+        assert response["content"] == "Hello World"
+        assert response["usage"] == {"prompt_tokens": 10, "completion_tokens": 5}
         print(f"Basic chat response: {response}")
 
     @pytest.mark.asyncio
@@ -286,6 +287,8 @@ async def mock_generate_content(*args, **kwargs):
         assert isinstance(response, dict)
         assert response.get("type") == "tool_call"
         assert "tool_calls" in response
+        # The usage stamp rides on tool_call envelopes too (no ``raw`` here).
+        assert response["usage"] == {"prompt_tokens": 15, "completion_tokens": 10}
 
         tool_calls = response["tool_calls"]
         assert len(tool_calls) > 0
@@ -454,10 +457,11 @@ async def mock_generate_content(*args, **kwargs):
 
         response = await llm.chat(messages, response_format={"type": "json_object"})
 
-        # Verify JSON response
-        assert isinstance(response, str)
-        assert "greeting" in response
-        assert "count" in response
+        # Verify JSON response: text envelope, JSON in its content
+        assert response["type"] == "text"
+        assert "greeting" in response["content"]
+        assert "count" in response["content"]
+        assert response["usage"] == {"prompt_tokens": 14, "completion_tokens": 20}
         print(f"JSON mode response: {response}")
 
     @pytest.mark.asyncio
diff --git a/tests/core/model/chat/basic/test_xinference.py b/tests/core/model/chat/basic/test_xinference.py
index 09f2e46b10..3cc484a699 100644
--- a/tests/core/model/chat/basic/test_xinference.py
+++ b/tests/core/model/chat/basic/test_xinference.py
@@ -1022,3 +1022,45 @@ def test_process_chat_response_records_cached_tokens(self) -> None:
         assert usage.input_tokens == 100
         inp = next(d for d in usage.details if d["type"] == "input")
         assert inp["cached_tokens"] == 60
+
+
+class TestXinferenceUsageStamp:
+    """D2: ``_process_chat_response`` stamps the provider usage onto every
+    envelope branch, so consumers never need to dig through ``raw``."""
+
+    def _payload(self) -> dict:
+        return {
+            "choices": [
+                {
+                    "message": {"role": "assistant", "content": "ok"},
+                    "finish_reason": "stop",
+                }
+            ],
+            "usage": {"prompt_tokens": 10, "completion_tokens": 5},
+        }
+
+    def test_text_envelope_stamps_usage(self) -> None:
+        llm = XinferenceLLM(model_name="qwen3.8")
+        result = llm._process_chat_response(self._payload())
+        assert result["usage"] == {"prompt_tokens": 10, "completion_tokens": 5}
+
+    def test_tool_call_envelope_stamps_usage(self) -> None:
+        llm = XinferenceLLM(model_name="qwen3.8")
+        payload = self._payload()
+        payload["choices"][0]["message"]["tool_calls"] = [
+            {
+                "id": "call_1",
+                "type": "function",
+                "function": {"name": "search", "arguments": "{}"},
+            }
+        ]
+        result = llm._process_chat_response(payload)
+        assert result["type"] == "tool_call"
+        assert result["usage"] == {"prompt_tokens": 10, "completion_tokens": 5}
+
+    def test_no_usage_no_stamp(self) -> None:
+        llm = XinferenceLLM(model_name="qwen3.8")
+        payload = self._payload()
+        del payload["usage"]
+        result = llm._process_chat_response(payload)
+        assert "usage" not in result
diff --git a/tests/core/model/chat/basic/test_zhipu.py b/tests/core/model/chat/basic/test_zhipu.py
index 89c4631543..014c3578e3 100644
--- a/tests/core/model/chat/basic/test_zhipu.py
+++ b/tests/core/model/chat/basic/test_zhipu.py
@@ -70,7 +70,8 @@ async def test_normal_text_response(self, zhipu_llm, mock_zhipu_client):
 
         result = await zhipu_llm.chat([{"role": "user", "content": "Hello"}])
 
-        assert result == "Hello, world!"
+        assert result["type"] == "text"
+        assert result["content"] == "Hello, world!"
 
     @pytest.mark.asyncio
     async def test_none_content_response(self, zhipu_llm, mock_zhipu_client):
@@ -134,6 +135,12 @@ async def test_tool_call_response(self, zhipu_llm, mock_zhipu_client):
 
         mock_response = MagicMock()
         mock_response.choices = [mock_choice]
+        mock_response.usage.prompt_tokens = 12
+        mock_response.usage.completion_tokens = 4
+        # Declare "no cache" explicitly: MagicMock auto-attrs would otherwise
+        # coerce to phantom cache tokens (int(MagicMock()) == 1).
+        mock_response.usage.prompt_cache_hit_tokens = 0
+        mock_response.usage.prompt_tokens_details = None
 
         mock_zhipu_client.chat.completions.create.return_value = mock_response
 
@@ -158,6 +165,8 @@ async def test_tool_call_response(self, zhipu_llm, mock_zhipu_client):
         assert result["type"] == "tool_call"
         assert len(result["tool_calls"]) == 1
         assert result["tool_calls"][0]["function"]["name"] == "calculator"
+        # The usage stamp rides on tool_call envelopes too.
+        assert result["usage"] == {"prompt_tokens": 12, "completion_tokens": 4}
 
     @pytest.mark.asyncio
     async def test_stream_chat_yields_tool_call_argument_deltas(
@@ -292,7 +301,8 @@ async def test_thinking_mode_disabled(self, zhipu_llm, mock_zhipu_client):
         assert "thinking" in call_args.kwargs
         assert call_args.kwargs["thinking"]["type"] == "disabled"
 
-        assert result == "Response with thinking disabled"
+        assert result["type"] == "text"
+        assert result["content"] == "Response with thinking disabled"
 
     @pytest.mark.asyncio
     async def test_empty_string_api_key_fallback(self, monkeypatch):
diff --git a/tests/web/api/test_agents_optimize_instructions.py b/tests/web/api/test_agents_optimize_instructions.py
index a6e9933465..79b7ef03ab 100644
--- a/tests/web/api/test_agents_optimize_instructions.py
+++ b/tests/web/api/test_agents_optimize_instructions.py
@@ -80,3 +80,69 @@ async def test_optimize_instructions_falls_back_on_wrong_language_output(
     )
 
     assert result == {"optimized_instructions": draft}
+
+
+@pytest.mark.asyncio
+async def test_optimize_instructions_rejects_tool_call_envelope(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """#1714: a tool_call envelope from the LLM must surface as an explicit
+    500, never a 200 carrying the envelope's repr as "optimized" text."""
+
+    class _ToolCallLLM:
+        async def chat(self, **kwargs: Any) -> dict[str, Any]:
+            return {
+                "type": "tool_call",
+                "tool_calls": [
+                    {
+                        "id": "call_1",
+                        "type": "function",
+                        "function": {"name": "read_file", "arguments": "{}"},
+                    }
+                ],
+            }
+
+    monkeypatch.setattr(
+        agents_api,
+        "UserAwareModelStorage",
+        lambda db: _FakeModelStorage(_ToolCallLLM()),  # type: ignore[arg-type]
+    )
+
+    with pytest.raises(agents_api.HTTPException) as exc_info:
+        await agents_api.optimize_instructions(
+            agents_api.OptimizeInstructionsRequest(instructions="请用中文回答。"),
+            SimpleNamespace(id=7),
+            object(),
+        )
+
+    assert exc_info.value.status_code == 500
+    assert "no usable text content" in exc_info.value.detail
+
+
+@pytest.mark.asyncio
+async def test_optimize_instructions_rejects_empty_content_envelope(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """An empty text envelope raises ``LLMEmptyContentError`` from
+    ``unwrap_chat_text`` and must surface as an explicit 500, not a 200 with
+    an empty "optimization"."""
+
+    class _EmptyContentLLM:
+        async def chat(self, **kwargs: Any) -> dict[str, Any]:
+            return {"type": "text", "content": ""}
+
+    monkeypatch.setattr(
+        agents_api,
+        "UserAwareModelStorage",
+        lambda db: _FakeModelStorage(_EmptyContentLLM()),  # type: ignore[arg-type]
+    )
+
+    with pytest.raises(agents_api.HTTPException) as exc_info:
+        await agents_api.optimize_instructions(
+            agents_api.OptimizeInstructionsRequest(instructions="请用中文回答。"),
+            SimpleNamespace(id=7),
+            object(),
+        )
+
+    assert exc_info.value.status_code == 500
+    assert "empty" in exc_info.value.detail