diff --git a/sdk/nexent/core/agents/context/budget.py b/sdk/nexent/core/agents/context/budget.py
index a8bf2c6572..01a6590256 100644
--- a/sdk/nexent/core/agents/context/budget.py
+++ b/sdk/nexent/core/agents/context/budget.py
@@ -1,3 +1,4 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
 """Small, representation-agnostic helpers for context rendering and summaries."""
 from __future__ import annotations
 
@@ -5,6 +6,9 @@
 import re
 from typing import Any
 
+from ...model_errors import ModelErrorCode
+from ...models.context_overflow import is_provider_context_overflow
+
 
 logger = logging.getLogger("agent_context.budget")
 
@@ -18,12 +22,16 @@ def format_summary_output(raw_output: str) -> str | None:
 
 
 def _is_context_length_error(error: Exception) -> bool:
+    error_code = getattr(error, "error_code", None)
+    if error_code is not None:
+        return error_code == ModelErrorCode.CONTEXT_OVERFLOW
+    if is_provider_context_overflow(error):
+        return True
+    # Legacy providers may omit a machine-readable code. Require explicit
+    # input/prompt pressure; output length is not an input overflow signal.
     text = str(error).lower()
-    return any(marker in text for marker in (
-        "context_length", "context length", "maximum context", "prompt is too long",
-        "reduce the length", "too many tokens", "token limit", "input is too long",
-        "input length", "exceeds context",
-    ))
+    return any(marker in text for marker in ("prompt is too long", "input is too long", "exceeds context"))
+
 
 def message_role(message: Any) -> str:
     role = message.get("role") if isinstance(message, dict) else getattr(message, "role", "")
diff --git a/sdk/nexent/core/agents/context/llm_summary.py b/sdk/nexent/core/agents/context/llm_summary.py
index 0fe4e51f68..6fbae498e4 100644
--- a/sdk/nexent/core/agents/context/llm_summary.py
+++ b/sdk/nexent/core/agents/context/llm_summary.py
@@ -1,3 +1,4 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
 """LLM call orchestration for summary generation with retry and error handling."""
 
 import logging
@@ -21,6 +22,7 @@
 @contextmanager
 def _summary_model_call_context(model, *, has_output_override: bool):
     """Isolate summary streaming and its request-specific output budget."""
+    observation = getattr(model, "last_context_observation", None)
     observer = getattr(model, "observer", None)
     budget_snapshot = getattr(model, "context_budget_snapshot", None)
     extra_body = getattr(model, "extra_body", None)
@@ -40,6 +42,8 @@ def _summary_model_call_context(model, *, has_output_override: bool):
     try:
         yield
     finally:
+        if hasattr(model, "last_context_observation"):
+            model.last_context_observation = observation
         if observer is not None:
             model.observer = observer
         if hasattr(model, "context_budget_snapshot"):
@@ -204,6 +208,11 @@ def _do_generate_summary(
             output_len=len(raw_output),
             response=response, call_type=call_type,
         )
+        finish_reason = getattr(getattr(response, "raw", None), "finish_reason", None)
+        if finish_reason is None:
+            finish_reason = getattr(model, "last_finish_reason", None)
+        if finish_reason in {"length", "content_filter"}:
+            return SummaryResult(records=[record])
         return SummaryResult(summary_text=summary, records=[record])
 
     def _record_failed_compression(self, call_type: str, error_msg: str) -> CompressionCallRecord:
diff --git a/sdk/nexent/core/agents/context/manager.py b/sdk/nexent/core/agents/context/manager.py
index 8ad376aff9..842e1b9c29 100644
--- a/sdk/nexent/core/agents/context/manager.py
+++ b/sdk/nexent/core/agents/context/manager.py
@@ -17,6 +17,7 @@
 from smolagents.models import ChatMessage, MessageRole
 
 from ...context_runtime.contracts import ContextEvidence, FinalContext
+from ...context_runtime.usage import InputUsageObservation, context_model_key, normalized_chars_per_token, request_characters
 from ..summary_cache import CompressionCallRecord
 from .budget import extract_message_text, message_role
 from .config import ContextManagerConfig
@@ -60,6 +61,11 @@ def __init__(self, config: Optional[ContextManagerConfig] = None, max_steps: int
         self._pending_history_summary_event: dict[str, Any] | None = None
         self._memory_compact_cache: dict[tuple[Any, ...], list[ContextItem]] = {}
         self._run_compaction_attempts = 0
+        self._usage_anchor: InputUsageObservation | None = None
+        self._estimate_basis = "local_estimate"
+        self._usage_scope_initialized = False
+        self._usage_model_key: str | None = None
+        self._usage_baseline: InputUsageObservation | None = None
 
     def _compaction_trigger_threshold_tokens(self) -> int:
         return self.config.compaction_trigger_threshold_tokens or self.config.token_threshold
@@ -84,6 +90,8 @@ def prepare_run_context(
         items: Optional[Sequence[Any]] = None,
     ) -> ManagedRunContext:
         self._history_candidate = None
+        self._usage_anchor = None
+        self._usage_scope_initialized = False
         self._run_compaction_attempts = 0
         self._current_item_cache.clear()
         source = self._item_source(items)
@@ -129,7 +137,9 @@ def assemble_final_context(
     ) -> FinalContext:
         if purpose != "step" and (request_system_messages or request_tail_messages):
             raise ValueError("Request-only action messages require step purpose")
+        self._run_compaction_attempts = 0
         run_context = run_context or self.prepare_run_context(memory, "")
+        self._observe_model_usage(model)
         policy = resolve_policy(self.config.policy_layers)
         persisted_items = list(run_context.items)
         if self._history_candidate is not None:
@@ -206,9 +216,14 @@ def assemble_final_context(
                 # If it is within the useful history target, fixed request
                 # content is the remaining excess and cannot be reclaimed by
                 # repeatedly summarizing the same checkpoint.
-                should_compress_history = original_history_tokens > history_target
+                should_compress_history = bool(
+                    force_compaction or (turns and original_history_tokens > history_target)
+               )
                 if not should_compress_history:
-                    compaction_stop_reason = "history_target_reached"
+                    compaction_stop_reason = (
+                        "no_new_history" if not turns and original_history_tokens > history_target
+                        else "history_target_reached"
+                    )
                 else:
                     history_triggered = True
                     self._emit_history_summary_status("compacting")
@@ -219,15 +234,19 @@ def assemble_final_context(
                 for _ in range(remaining_attempts if should_compress_history else 0):
                     self._run_compaction_attempts += 1
                     attempt = self._run_compaction_attempts
-                    result = self._history_compressor.compress(current_summary, current_turns, model)
+                    try:
+                        result = self._history_compressor.compress(current_summary, current_turns, model)
+                    except Exception:
+                        # Auxiliary maintenance cannot reject a main model request.
+                        # Cooperative cancellation inherits BaseException and propagates.
+                        logger.exception("History maintenance failed; preserving accepted context")
+                        compaction_stop_reason = "maintenance_failed"
+                        break
                     attempts = attempt
                     self._record_compression(result.records)
                     candidate = result.candidate
                     if candidate is None:
                         compaction_stop_reason = "invalid_candidate"
-                        if last_valid is None and result.fallback_turns:
-                            fallback_by_id = {item.id: item for item in result.fallback_turns}
-                            final_items = [fallback_by_id.get(item.id, item) for item in final_items]
                         break
                     candidate_item = candidate.as_item()
                     previous_tokens = sum(item.token_estimate for item in ([current_summary] if current_summary else []))
@@ -281,13 +300,13 @@ def assemble_final_context(
                 elif should_compress_history:
                     self._emit_history_summary_status("idle")
 
-            final_items = self._compact_to_compaction_target(
-                final_items,
-                purpose_stable,
-                purpose_dynamic,
-                canonical_tools,
-                model=model,
-            )
+            try:
+                final_items = self._compact_to_compaction_target(
+                    final_items, purpose_stable, purpose_dynamic, canonical_tools, model=model,
+                )
+            except Exception:
+                logger.exception("Item maintenance failed; preserving accepted context")
+                compaction_stop_reason = "maintenance_failed"
 
         final_items.sort(key=lambda item: item.layout_key)
         rendered = self.build_context_messages(final_items)
@@ -296,7 +315,7 @@ def assemble_final_context(
         dynamic = [message for message in rendered if message_role(message) not in {"system", "developer"}]
         memory_messages = [*stable, *dynamic]
         messages = [*stable, *purpose_stable, *dynamic, *purpose_dynamic]
-        final_tokens = self._message_tokens(messages) + self._tools_tokens(canonical_tools)
+        final_tokens = self._estimate_request(messages, canonical_tools)
         self._last_uncompressed_token_count = raw_tokens
         self._last_compressed_token_count = final_tokens
         effective_limit = self.config.effective_input_limit_tokens
@@ -346,6 +365,8 @@ def assemble_final_context(
                 non_history_tokens=non_history_tokens,
                 compaction_stop_reason=compaction_stop_reason,
                 raw_token_estimate=raw_tokens,
+                token_estimate_basis=self._estimate_basis,
+                observed_provider_input_tokens=(self._usage_anchor.input_tokens if self._usage_anchor else None),
                 final_token_estimate=final_tokens,
                 loaded_summary_unit_id=(loaded.content.get("unit_id") if loaded else None),
                 loaded_summary_coverage=(loaded.content.get("covered_through_message_id") if loaded else None),
@@ -432,7 +453,9 @@ def _compact_to_compaction_target(self, items, purpose_stable, purpose_dynamic,
             for _, _, _, original, compact in sorted(savings, key=lambda row: (-row[0], row[1], row[2])):
                 if compact is None:
                     compact = self._compact_execution(original, model)
-                if compact is None or self._estimate_items([compact], [], [], []) >= self._estimate_items([original], [], [], []):
+                if compact is None or request_characters(self.build_context_messages([compact]), []) >= request_characters(
+                    self.build_context_messages([original]), []
+                ):
                     continue
                 result[result.index(original)] = compact
                 if self._estimate_items(result, purpose_stable, purpose_dynamic, tools) <= self._compaction_target_tokens():
@@ -539,7 +562,7 @@ def _select_long_term_memories(self, result, memory_items, *, model):
             replacements.append(item.__class__(
                 **data, content=content,
                 metadata={**deepcopy(item.metadata), "representation": "selected_blocks"},
-                token_estimate=max(1, math.ceil(len(content[key]) / self.config.chars_per_token)),
+                token_estimate=max(1, math.ceil(len(content[key]) / self._chars_per_token())),
             ))
         self._memory_compact_cache[cache_key] = replacements
         self._record_compression([CompressionCallRecord(
@@ -668,18 +691,44 @@ def _purpose_messages(self, *, purpose, task, final_answer_templates):
         )
 
     def _estimate_items(self, items, stable, dynamic, tools):
-        return self._message_tokens([*self.build_context_messages(items), *stable, *dynamic]) + self._tools_tokens(
-            tools
-        )
+        return self._estimate_request([*self.build_context_messages(items), *stable, *dynamic], tools)
+
+    def _observe_model_usage(self, model):
+        observation = getattr(model, "last_context_observation", None)
+        key = context_model_key(model)
+        if not self._usage_scope_initialized or self._usage_model_key != key:
+            # Ignore a model's pre-run or pre-replacement response (connectivity
+            # probes and other conversations may have used the same instance).
+            self._usage_anchor = None
+            self._usage_baseline = observation if isinstance(observation, InputUsageObservation) else None
+            self._usage_model_key = key
+            self._usage_scope_initialized = True
+            return
+        if (isinstance(observation, InputUsageObservation) and observation.model_key == key
+                and observation != self._usage_baseline):
+            self._usage_anchor = observation
+
+    def _estimate_request(self, messages, tools):
+        if self._usage_anchor is not None:
+            self._estimate_basis = "provider_calibrated_estimate"
+            return self._usage_anchor.estimate(
+                input_characters=request_characters(messages, tools),
+                chars_per_token=self._chars_per_token(),
+            )
+        self._estimate_basis = "local_estimate"
+        return self._message_tokens(messages) + self._tools_tokens(tools)
+
+    def _chars_per_token(self):
+        return normalized_chars_per_token(self.config.chars_per_token)
 
     def _message_tokens(self, messages):
         return max(
-            0, int(sum(len(extract_message_text(message)) for message in messages) / self.config.chars_per_token)
+            0, int(sum(len(extract_message_text(message)) for message in messages) / self._chars_per_token())
         )
 
     def _tools_tokens(self, tools):
         return (
-            int(len(json.dumps(tools, ensure_ascii=False, default=str)) / self.config.chars_per_token) if tools else 0
+            int(len(json.dumps(tools, ensure_ascii=False, default=str)) / self._chars_per_token()) if tools else 0
         )
 
     def _record_compression(self, records):
diff --git a/sdk/nexent/core/agents/context/runtime.py b/sdk/nexent/core/agents/context/runtime.py
index f74321e163..b492f14f8e 100644
--- a/sdk/nexent/core/agents/context/runtime.py
+++ b/sdk/nexent/core/agents/context/runtime.py
@@ -1,3 +1,4 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
 """Managed context path thin adapter.
 
 All context policy and final payload assembly belongs to ContextManager.  This
@@ -12,6 +13,7 @@
 from smolagents.models import ChatMessage, Model
 
 from ...context_runtime.contracts import ContextEvidence, FinalContext, ModelTool
+from ...context_runtime.usage import normalized_chars_per_token
 from .evidence import ContextEvidenceCollector
 from .manager import ContextManager
 from .models import ContextItem, ContextItemInput
@@ -178,7 +180,7 @@ def finalize_evidence(self, *, status: str) -> ContextEvidence:
 
     @property
     def chars_per_token(self) -> float:
-        return self.context_manager.config.chars_per_token
+        return normalized_chars_per_token(self.context_manager.config.chars_per_token)
 
     @property
     def token_threshold(self) -> int | None:
diff --git a/sdk/nexent/core/agents/core_agent.py b/sdk/nexent/core/agents/core_agent.py
index 5f1b3b5388..2c230ef10c 100644
--- a/sdk/nexent/core/agents/core_agent.py
+++ b/sdk/nexent/core/agents/core_agent.py
@@ -1571,18 +1571,9 @@ def rebuild_after_provider_overflow():
             memory_step.code_action = code_action
             history = getattr(self, "_execution_history", None)
             if history is not None:
-                limit = self.context_runtime.effective_input_limit_tokens
-                if limit:
-                    estimate = final_context.evidence.final_token_estimate or 0
-                    usage = getattr(memory_step.model_output_message, "token_usage", None)
-                    provider_input = getattr(usage, "input_tokens", 0)
-                    if isinstance(provider_input, int):
-                        estimate = max(estimate, provider_input)
-                    # Existing context plus the accepted new assistant body must fit
-                    # alongside the read result. Reserve wrapper/error overhead.
-                    remaining = max(0, limit - estimate - 256)
-                    chars = max(0, int(remaining * self.context_runtime.chars_per_token) - len(str(model_output)))
-                    history.reset_read_budget(self._execution_owner, max_rendered_chars=chars)
+                # Keep recall bounded by the existing I/O cap. Local token estimates
+                # are advisory and must not make authorized original output unreadable.
+                history.reset_read_budget(self._execution_owner, max_rendered_chars=16000)
             self._resolve_deferred_model_attempt(
                 memory_step.model_output_message,
                 accepted=True,
@@ -2390,8 +2381,13 @@ def _collect_step_metrics(self, action_step: ActionStep):
 
         # 3. Current memory estimated length
         chars_per_token = self.context_runtime.chars_per_token
-        metric["memory_state"]["estimated_input_tokens"] = msg_token_count(
-            action_step.model_input_messages, chars_per_token
+        context_counts = self.context_runtime.token_counts()
+        compressed_estimate = context_counts.get("compressed") if isinstance(context_counts, dict) else None
+        # Compare the same assembled payload and estimation basis before/after
+        # maintenance, including request overlays and tool schemas.
+        metric["memory_state"]["estimated_input_tokens"] = (
+            compressed_estimate if type(compressed_estimate) is int and compressed_estimate >= 0
+            else msg_token_count(action_step.model_input_messages, chars_per_token)
         )
         metric["memory_state"]["estimated_output_tokens"] = msg_token_count(
             action_step.model_output_message, chars_per_token
diff --git a/sdk/nexent/core/context_runtime/contracts.py b/sdk/nexent/core/context_runtime/contracts.py
index e21e1a5c36..ec879c96e5 100644
--- a/sdk/nexent/core/context_runtime/contracts.py
+++ b/sdk/nexent/core/context_runtime/contracts.py
@@ -1,3 +1,4 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
 """Contracts for the single ContextManager-backed runtime path."""
 from __future__ import annotations
 
@@ -47,6 +48,8 @@ class ContextEvidence:
     compressible_history_tokens: int = 0
     non_history_tokens: int = 0
     compaction_stop_reason: str = "not_triggered"
+    token_estimate_basis: str = "local_estimate"
+    observed_provider_input_tokens: int | None = None
     raw_token_estimate: int = 0
     final_token_estimate: int = 0
     loaded_summary_unit_id: int | None = None
diff --git a/sdk/nexent/core/context_runtime/usage.py b/sdk/nexent/core/context_runtime/usage.py
new file mode 100644
index 0000000000..1e6095c8c7
--- /dev/null
+++ b/sdk/nexent/core/context_runtime/usage.py
@@ -0,0 +1,83 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
+"""Immutable completed-request observations for advisory context maintenance.
+
+Only provider input usage is measured. Differences for a subsequent payload are
+estimates, including new tool schemas and results; they never authorize rejection.
+No request bodies or credentials are retained in an observation.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+from dataclasses import dataclass
+from typing import Any
+
+from pydantic import BaseModel
+
+
+def normalized_chars_per_token(value: Any) -> float:
+    """Keep an unavailable local estimation ratio from blocking inference."""
+    return value if type(value) in (int, float) and math.isfinite(value) and value > 0 else 1.5
+
+
+def _json_value(value: Any) -> Any:
+    if isinstance(value, BaseModel):
+        return value.model_dump(mode='json')
+    return str(value)
+
+
+def request_characters(messages, tools) -> int:
+    """Measure the entire caller payload, preserving native tool boundaries."""
+    normalized = []
+    for message in messages:
+        if isinstance(message, dict):
+            normalized.append({key: message[key] for key in ('role', 'content', 'tool_calls', 'tool_call_id')
+                               if message.get(key) is not None})
+        else:
+            normalized.append({key: getattr(message, key) for key in ('role', 'content', 'tool_calls', 'tool_call_id')
+                               if getattr(message, key, None) is not None})
+    schemas = []
+    for tool in tools or ():
+        if isinstance(tool, dict):
+            schemas.append(tool)
+        else:
+            from smolagents.models import get_tool_json_schema
+            schemas.append(get_tool_json_schema(tool))
+    return len(json.dumps({'messages': normalized, 'tools': schemas}, ensure_ascii=False,
+                          sort_keys=True, default=_json_value))
+
+
+def context_model_key(model) -> str:
+    """Bind observations to the saved outlet and its configuration revision."""
+    profile = getattr(model, 'profile', None)
+    value = profile.model_dump(mode='json') if isinstance(profile, BaseModel) else {
+        'model_id': getattr(model, 'model_id', None),
+        'model_name': getattr(model, 'model_name', None),
+        'model_factory': getattr(model, 'model_factory', None),
+    }
+    return hashlib.sha256(json.dumps({"instance": id(model), "configuration": value},
+                                     sort_keys=True, default=_json_value).encode()).hexdigest()
+
+
+@dataclass(frozen=True)
+class InputUsageObservation:
+    input_tokens: int
+    input_characters: int
+    model_key: str
+    request_id: str
+    attempt: int
+
+    @classmethod
+    def capture(cls, *, usage, messages, tools, model_key, request_id, attempt):
+        """Accept only complete provider observations, retaining unknown as absent."""
+        tokens = getattr(usage, 'input_tokens', None)
+        if getattr(usage, 'origin', None) != 'provider' or type(tokens) is not int or tokens < 0:
+            return None
+        return cls(tokens, request_characters(messages, tools), model_key, request_id, attempt)
+
+    def estimate(self, *, input_characters: int, chars_per_token: float) -> int:
+        """Account for the whole changed payload without calling it measured usage."""
+        delta = (input_characters - self.input_characters) / normalized_chars_per_token(chars_per_token)
+        return max(0, math.ceil(self.input_tokens + delta))
diff --git a/sdk/nexent/core/model_service/compatibility/legacy_chat_model.py b/sdk/nexent/core/model_service/compatibility/legacy_chat_model.py
index c0cf46c8ed..b4547b281f 100644
--- a/sdk/nexent/core/model_service/compatibility/legacy_chat_model.py
+++ b/sdk/nexent/core/model_service/compatibility/legacy_chat_model.py
@@ -20,6 +20,7 @@
 )
 from smolagents.monitoring import TokenUsage
 
+from nexent.core.context_runtime.usage import InputUsageObservation, context_model_key
 from nexent.core.concurrency.cancellation import RunCancellationScope
 from nexent.core.model_errors import ModelErrorCode, ModelInvocationTerminalError
 from nexent.core.model_service.compatibility.invocation_options import LegacyInvocationOptions
@@ -134,6 +135,7 @@ def __init__(
         self.last_provider_cache_advice = None
         self.last_retry_count = 0
         self.last_context_evidence = None
+        self.last_context_observation = None
 
     @classmethod
     def from_profile(cls, profile, api_key, *, http_headers=None, observer=None, cancellation_scope=None):
@@ -284,11 +286,15 @@ def _events(
             raise TypeError("Legacy context_rebuild must be a synchronous callback")
         request, profile = self._request(messages, options, tools)
 
+        observed_messages = messages
+
         def rebuild_request(_error):
+            nonlocal observed_messages
             rebuilt = context_rebuild()
             rebuilt_messages = getattr(rebuilt, "messages", rebuilt)
             if not isinstance(rebuilt_messages, list):
                 raise TypeError("context_rebuild must return FinalContext or a message list")
+            observed_messages = rebuilt_messages
             self.last_context_evidence = getattr(rebuilt, "evidence", None)
             replacement, _ = self._request(rebuilt_messages, options, tools, profile=profile)
             return replacement.model_copy(update={"request_id": request.request_id})
@@ -297,6 +303,7 @@ def rebuild_request(_error):
         self.last_attempt_number = None
         self.last_retry_count = 0
         self.last_context_evidence = None
+        self.last_context_observation = None
         self._reset_response(profile)
         attempt_number = None
         completed = False
@@ -330,6 +337,10 @@ def rebuild_request(_error):
                         if not suppress_stream:
                             self.observer.flush_remaining_tokens()
                         message = event.message
+                        self.last_context_observation = InputUsageObservation.capture(
+                            usage=message.usage, messages=observed_messages, tools=tools,
+                            model_key=context_model_key(self), request_id=request.request_id, attempt=event.attempt,
+                        )
                         self.last_finish_reason = message.finish_reason
                         self.last_input_token_count = message.usage.input_tokens
                         self.last_output_token_count = message.usage.output_tokens
diff --git a/sdk/nexent/core/models/openai_llm.py b/sdk/nexent/core/models/openai_llm.py
index 605b174406..fb653d822e 100644
--- a/sdk/nexent/core/models/openai_llm.py
+++ b/sdk/nexent/core/models/openai_llm.py
@@ -10,6 +10,8 @@
 )
 from ..utils.token_estimation import estimate_tokens_text
 from .response_diagnostics import ResponseDiagnostics
+from ..context_runtime.usage import InputUsageObservation, context_model_key
+from types import SimpleNamespace
 from ..concurrency import RunCancellationScope, run_blocking
 import logging
 import threading
@@ -283,6 +285,7 @@ def __init__(self, observer: MessageObserver = MessageObserver, temperature=DEFA
         self.last_cached_input_token_count = 0
         self.diagnostic_chunks_enabled = diagnostic_chunks_enabled
         self.last_response_diagnostics = None
+        self.last_context_observation = None
         self.last_attempt_id = None
         self.last_attempt_number = None
         self.last_reasoning_preview = None
@@ -370,6 +373,7 @@ def __call__(self, messages: List[Dict[str, Any]], stop_sequences: Optional[List
                  _retry_empty_response: bool = True,
                  _empty_response_retries_used: int = 0,
                  **kwargs, ) -> ChatMessage:
+        self.last_context_observation = None
         messages = merge_system_messages(messages or [])
         _monitoring_operation.set("chat_completion")
 
@@ -893,6 +897,11 @@ def _close_stream_once():
                         "attempt_id": attempt_id,
                         "attempt": attempt,
                     })
+                    self.last_context_observation = InputUsageObservation.capture(
+                        usage=SimpleNamespace(input_tokens=input_tokens, origin="provider" if usage is not None else "unknown"),
+                        messages=messages, tools=tools_to_call_from,
+                        model_key=context_model_key(self), request_id=attempt_id, attempt=attempt,
+                    )
                     return message
 
                 except Exception as e:
diff --git a/sdk/nexent/monitor/monitoring.py b/sdk/nexent/monitor/monitoring.py
index 54d4012f0e..38c0473c1b 100644
--- a/sdk/nexent/monitor/monitoring.py
+++ b/sdk/nexent/monitor/monitoring.py
@@ -1424,6 +1424,8 @@ def record_final_context_evidence(self, evidence: Any, step_number: int) -> None
             "context.compaction_target_tokens": getattr(
                 evidence, "compaction_target_tokens", 0
             ),
+            "context.tokens.estimate_basis": getattr(evidence, "token_estimate_basis", "local_estimate"),
+            "context.tokens.provider_input_known": getattr(evidence, "observed_provider_input_tokens", None) is not None,
             "context.tokens.pre_compression": getattr(evidence, "raw_token_estimate", 0),
             "context.tokens.post_compression": getattr(evidence, "final_token_estimate", 0),
             "context.compaction.exceeds_effective_input_limit": bool(
@@ -1433,6 +1435,9 @@ def record_final_context_evidence(self, evidence: Any, step_number: int) -> None
             "context.compression.fallback_compaction": bool(getattr(evidence, "fallback_compaction_used", False)),
             "context.compression.records": json.dumps(compression_records, ensure_ascii=False, sort_keys=True),
         }
+        observed_input = getattr(evidence, "observed_provider_input_tokens", None)
+        if observed_input is not None:
+            attrs["context.tokens.observed_provider_input"] = observed_input
         self.add_span_event("agent.final_context", attrs)
 
     def set_agent_context_metrics(self, metrics: List[Dict[str, Any]]) -> None:
diff --git a/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/case.yaml b/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/case.yaml
index 574e4161fd..8b6b333566 100644
--- a/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/case.yaml
+++ b/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/case.yaml
@@ -31,11 +31,13 @@ case:
   steps:
   - order: 1
     action: Construct a model through the actual NexentAgent factory with a supplied
-      resolved outlet, then send normal and streaming requests through sync and async APIs; capture
-      requests at the HTTP peer, including the resolved gateway asynchronous stream
-      and authorized resource conversion after restoring a saved row snapshot.
-    expected: Exact root/path/body/auth and one physical request; empty temperature/top_p keys are absent and numeric zero is retained; custom gateway headers stay separate from model-specific Body parameters and credential values stay outside snapshots; content, reasoning,
-      tool arguments and usage are preserved.
+      resolved outlet, then send normal and streaming requests through sync and async
+      APIs; capture requests at the HTTP peer, including the resolved gateway asynchronous
+      stream and authorized resource conversion after restoring a saved row snapshot.
+    expected: Exact root/path/body/auth and one physical request; empty temperature/top_p
+      keys are absent and numeric zero is retained; custom gateway headers stay separate
+      from model-specific Body parameters and credential values stay outside snapshots;
+      content, reasoning, tool arguments and usage are preserved.
   - order: 2
     action: Serve multiline and fragmented SSE frames, including final tool calls
       and usage; repeat identical usage and omit DONE after a valid finish.
@@ -43,13 +45,21 @@ case:
       never become completed calls. Identical usage is emitted once; normal EOF after
       a complete finish is accepted while conflicts and truncation remain failures.
   - order: 3
-    action: Serve transient errors before visible output, context overflow followed by
-      rebuilt messages, authentication errors, malformed/truncated streams and stop a stream early.
+    action: Serve transient errors before visible output, context overflow followed
+      by rebuilt messages, authentication errors, malformed/truncated streams and
+      stop a stream early.
     expected: Only ModelService may retry a transient pre-output failure within its
       declared attempt budget; authentication, visible partial output and cancellation
       never replay. Errors are sanitized and owned HTTP resources and shared permits
       are released. Context recovery retains frozen credentials and changes only messages
       within the same physical attempt budget.
+  - order: 4
+    action: Complete an actual model request through the HTTP peer, apply its provider
+      input usage to shared context assembly, and change messages or tool schemas
+      before the next request. Include unknown usage and failed maintenance.
+    expected: The observation belongs to the completed request; subsequent context
+      accounts for changed input, advisory excess never prevents the peer request,
+      and failed maintenance preserves original content.
   expected_results:
   - Actual HTTP request matches the prepared outlet configuration in sync and async
     paths.
@@ -57,6 +67,8 @@ case:
   - Retry counts obey the explicit ModelService budget before visible output; failure,
     partial output and cancellation release resources and never hide a replay or private
     provider message.
+  - Shared context maintenance uses completed provider input observations without
+    count-based admission rejection or loss of protected input.
   forbidden_side_effects:
   - No hosted provider, production connections, tool execution or real credentials.
   - No retry outside the declared pre-output budget, AgentMode change or secret content
@@ -67,7 +79,8 @@ case:
       required_services: []
   acceptance:
   - id: AC-01
-    description: Exact request reaches the real HTTP peer, omitting empty temperature/top_p while preserving zero and configured numbers.
+    description: Exact request reaches the real HTTP peer, omitting empty temperature/top_p
+      while preserving zero and configured numbers.
     steps:
     - 1
     expected_results:
@@ -94,3 +107,13 @@ case:
     forbidden_side_effects:
     - 1
     - 2
+  - id: AC-04
+    description: Completed provider input usage is paired with its request and shared
+      maintenance remains non-blocking when input changes or a summary fails.
+    steps:
+    - 4
+    expected_results:
+    - 4
+    forbidden_side_effects:
+    - 1
+    - 2
diff --git a/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/context_maintenance_cases.py b/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/context_maintenance_cases.py
new file mode 100644
index 0000000000..3bb4818d77
--- /dev/null
+++ b/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/context_maintenance_cases.py
@@ -0,0 +1,113 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
+"""Shared maintenance through actual model HTTP and SSE boundaries."""
+
+from nexent.core.agents.context import (
+    ContextItemInput,
+    ContextManager,
+    ContextManagerConfig,
+)
+from nexent.core.model_service.compatibility.legacy_chat_model import LegacyChatModel
+from smolagents.memory import AgentMemory, TaskStep
+
+
+def test_shared_context_maintenance_uses_completed_provider_usage(peer, request):
+    module = request.module
+    service, _, profile, limiter, journal, credentials = module.setup(peer, 'compatible_eof')
+    model = LegacyChatModel(profile, service)
+    persisted = []
+    config = ContextManagerConfig(
+        effective_input_limit_tokens=50, history_summary_sink=persisted.append,
+        policy_layers={'request': {'processing_mode': 'passthrough'}},
+    )
+    manager = ContextManager(config)
+    history_text = 'ORIGINAL-HISTORY-SENTINEL ' + 'historical evidence ' * 1000
+    history = ContextItemInput(id='turn:1:2', type='conversation_turn', content={
+        'user_message': history_text, 'assistant_final_answer': 'previous result',
+        'user_message_id': 1, 'assistant_message_id': 2, 'attachments': [],
+    })
+    memory = AgentMemory(system_prompt='Keep original evidence.')
+    memory.steps.append(TaskStep(task='current task'))
+    run = manager.prepare_run_context(memory, 'Keep original evidence.', [history])
+    initial = manager.assemble_final_context(model=model, memory=memory, current_run_start_idx=0, run_context=run)
+    assert initial.evidence.exceeds_effective_input_limit
+    model(initial.messages)
+    observation = model.last_context_observation
+    assert observation.input_tokens == 10
+    same = manager.assemble_final_context(model=model, memory=memory, current_run_start_idx=0, run_context=run)
+    assert same.evidence.final_token_estimate == 10
+    assert same.evidence.observed_provider_input_tokens == 10
+    assert same.evidence.token_estimate_basis == 'provider_calibrated_estimate'
+
+    # Real auxiliary inference returns noncanonical prose, so it must not
+    # replace the accepted checkpoint or the main-request observation.
+    config.policy_layers = {'request': {'processing_mode': 'adaptive_compact'}}
+    memory.steps.append(TaskStep(task='NEW-CURRENT-TASK ' + 'new input ' * 200))
+    changed = manager.assemble_final_context(model=model, memory=memory, current_run_start_idx=0, run_context=run)
+    assert changed.evidence.history_compression_triggered
+    assert changed.evidence.final_token_estimate > 10
+    assert changed.evidence.exceeds_effective_input_limit
+    assert model.last_context_observation is observation
+    assert not persisted and changed.evidence.new_summary_coverage is None
+    response = model(changed.messages)
+    assert response.content == 'compatible'
+    actual = module.records(journal, scenario='compatible_eof')
+    assert len(actual) == len(credentials) == 3  # Initial, rejected summary, next main request.
+    sent = str(actual[-1]['body']['messages'])
+    assert history_text in sent and 'NEW-CURRENT-TASK' in sent
+    assert not limiter._entries
+
+
+def test_shared_context_unknown_usage_does_not_block_request(peer, request):
+    module = request.module
+    service, _, profile, limiter, journal, credentials = module.setup(peer, 'compatible_no_usage')
+    model = LegacyChatModel(profile, service)
+    manager = ContextManager(ContextManagerConfig(effective_input_limit_tokens=1,
+        policy_layers={'request': {'processing_mode': 'passthrough'}}))
+    memory = AgentMemory(system_prompt='Preserve the input.')
+    memory.steps.append(TaskStep(task='PROTECTED INPUT ' * 200))
+    run = manager.prepare_run_context(memory, 'Preserve the input.')
+    for _ in range(2):
+        context = manager.assemble_final_context(model=model, memory=memory, current_run_start_idx=0, run_context=run)
+        assert context.evidence.exceeds_effective_input_limit
+        assert context.evidence.observed_provider_input_tokens is None
+        assert context.evidence.token_estimate_basis == 'local_estimate'
+        response = model(context.messages)
+        assert response.content == 'unknown usage response' and response.token_usage is None
+        assert model.last_context_observation is None
+    assert len(module.records(journal, scenario='compatible_no_usage')) == len(credentials) == 2
+    assert not limiter._entries
+
+
+def test_explicit_overflow_rebuilds_shared_history_even_when_estimate_fits(peer, request):
+    from nexent.core.agents.context.runtime import ManagedContextRuntime
+
+    module = request.module
+    service, _, profile, limiter, journal, credentials = module.setup(peer, 'shared_context_overflow')
+    model = LegacyChatModel(profile, service)
+    saved = []
+    manager = ContextManager(ContextManagerConfig(effective_input_limit_tokens=1_000_000,
+        history_summary_sink=saved.append, policy_layers={'request': {'processing_mode': 'adaptive_compact'}}))
+    history = ContextItemInput(id='turn:1:2', type='conversation_turn', content={
+        'user_message': 'SOURCE-FACT-001 is ORCHID-7429.\n' + 'old evidence ' * 200,
+        'assistant_final_answer': 'Keep the exact fact.', 'user_message_id': 1, 'assistant_message_id': 2,
+        'attachments': [],
+    })
+    memory = AgentMemory(system_prompt='Keep protected context.')
+    runtime = ManagedContextRuntime(manager, [history])
+    runtime.prepare_run(memory=memory, fallback_system_prompt='Keep protected context.')
+    memory.steps.append(TaskStep(task='PROTECTED-CURRENT-TASK'))
+    initial = runtime.prepare_step(model=model, memory=memory, current_run_start_idx=0)
+    assert not initial.evidence.history_compression_triggered
+    assert initial.evidence.final_token_estimate < initial.evidence.compaction_trigger_threshold_tokens
+    response = model(initial.messages, context_rebuild=lambda: runtime.recover_step(
+        model=model, memory=memory, current_run_start_idx=0))
+    assert response.content == 'recovered' and response.model_attempt_number == 2
+    assert len(saved) == 1 and saved[0].covered_through_message_id == 2
+    captured = module.records(journal, scenario='shared_context_overflow')
+    assert len(captured) == 3  # Rejected main, real summary, rebuilt main.
+    assert len(credentials) == 2  # Two logical requests; the main retry retains frozen credentials.
+    assert 'PROTECTED-CURRENT-TASK' not in str(captured[1]['body']['messages'])
+    rebuilt = str(captured[-1]['body']['messages'])
+    assert 'PROTECTED-CURRENT-TASK' in rebuilt and 'ORCHID-7429' in rebuilt
+    assert 'old evidence ' * 200 not in rebuilt
+    assert not limiter._entries
diff --git a/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/execution.yaml b/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/execution.yaml
index df510eaf43..3e0fe0f28a 100644
--- a/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/execution.yaml
+++ b/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/execution.yaml
@@ -5,12 +5,39 @@ implementations:
 - framework: pytest
   file: test.py
   selector: MODEL-SERVICE-TRANSPORT-D3-001
-  profiles: [mock]
+  profiles:
+  - mock
 acceptance_bindings:
-  # Auxiliary backend text flow preserves output and does not add an outer retry.
-  AC-01: [test_all_selected_custom_model_controls_reach_real_peer, test_all_selected_hosted_model_controls_reach_real_peer, test_custom_provider_shutdown_configuration_reaches_peer, test_optional_sampling_reaches_http_peer_without_null, test_exact_request_reaches_real_peer, test_legacy_model_interface_uses_real_transport_and_retains_usage, test_legacy_connectivity_uses_real_service_defaults, test_actual_agent_factory_uses_resolved_outlet, test_resource_conversion_and_saved_snapshot_reach_real_peer, test_auxiliary_prompt_uses_saved_outlet_http_without_outer_retry, test_long_text_preparation_uses_saved_outlet_real_http]
-  AC-02: [test_agent_bridge_uses_vllm_tail_and_shared_localized_errors, test_compatible_stream_tails_complete_once_through_actual_http, test_fragmented_stream_retains_kinds_calls_and_usage, test_resolved_gateway_stream_uses_real_async_transport]
-  AC-03: [test_agent_bridge_uses_vllm_tail_and_shared_localized_errors, test_http_detail_reaches_agent_bridge_without_credentials_or_extra_retry, test_retry_error_and_partial_output_boundaries, test_cancellation_closes_blocked_real_response, test_sync_cancellation_reclaims_blocked_managed_worker, test_legacy_empty_policy, test_legacy_context_recovery_shares_real_http_budget, test_auxiliary_prompt_uses_saved_outlet_http_without_outer_retry]
+  AC-01:
+  - test_all_selected_custom_model_controls_reach_real_peer
+  - test_all_selected_hosted_model_controls_reach_real_peer
+  - test_custom_provider_shutdown_configuration_reaches_peer
+  - test_optional_sampling_reaches_http_peer_without_null
+  - test_exact_request_reaches_real_peer
+  - test_legacy_model_interface_uses_real_transport_and_retains_usage
+  - test_legacy_connectivity_uses_real_service_defaults
+  - test_actual_agent_factory_uses_resolved_outlet
+  - test_resource_conversion_and_saved_snapshot_reach_real_peer
+  - test_auxiliary_prompt_uses_saved_outlet_http_without_outer_retry
+  - test_long_text_preparation_uses_saved_outlet_real_http
+  AC-02:
+  - test_agent_bridge_uses_vllm_tail_and_shared_localized_errors
+  - test_compatible_stream_tails_complete_once_through_actual_http
+  - test_fragmented_stream_retains_kinds_calls_and_usage
+  - test_resolved_gateway_stream_uses_real_async_transport
+  AC-03:
+  - test_agent_bridge_uses_vllm_tail_and_shared_localized_errors
+  - test_http_detail_reaches_agent_bridge_without_credentials_or_extra_retry
+  - test_retry_error_and_partial_output_boundaries
+  - test_cancellation_closes_blocked_real_response
+  - test_sync_cancellation_reclaims_blocked_managed_worker
+  - test_legacy_empty_policy
+  - test_legacy_context_recovery_shares_real_http_budget
+  - test_auxiliary_prompt_uses_saved_outlet_http_without_outer_retry
+  - test_explicit_overflow_rebuilds_shared_history_even_when_estimate_fits
+  AC-04:
+  - test_shared_context_maintenance_uses_completed_provider_usage
+  - test_shared_context_unknown_usage_does_not_block_request
 preparation:
   families: []
   anchors: false
diff --git a/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/http_peer.py b/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/http_peer.py
index 16dd1553c8..c245269d53 100644
--- a/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/http_peer.py
+++ b/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/http_peer.py
@@ -121,6 +121,10 @@ def do_POST(self):
                 return self.json_response(400, {"error": {
                     "code": "context_length_exceeded", "message": "provider rejected dummy-loopback",
                 }})
+            if scenario == "shared_context_overflow" and self.server.counts[scenario] == 1:
+                return self.json_response(400, {"error": {
+                    "code": "context_length_exceeded", "message": "controlled input overflow",
+                }})
             if scenario == "disconnect" and self.server.counts[scenario] == 1:
                 self.connection.shutdown(socket.SHUT_RDWR)
                 return
@@ -165,6 +169,27 @@ def do_POST(self):
             if scenario == "vllm_023_tail":
                 sample = Path(__file__).resolve().parents[3] / "test/fixtures/model_service/vllm_023_glm_tail.sse"
                 self.chunk(sample.read_bytes())
+            elif scenario == "compatible_no_usage":
+                self.chunk(frame({"content": "unknown usage response"}, finish="stop"))
+                self.chunk(b"data: [DONE]\n\n")
+            elif scenario == "shared_context_overflow":
+                if self.server.counts[scenario] == 2:
+                    content = "# Compact Result of History\n\n" + "\n\n".join(
+                        "## " + heading + "\n\n" + value for heading, value in [
+                            ("Task Overview", "Preserve the original fact."),
+                            ("Completed Work", "SOURCE-FACT-001 is ORCHID-7429."),
+                            ("Key Decisions", "Use the observed original."),
+                            ("Unresolved Issues", "None."),
+                            ("Pending Items", "Complete the current task."),
+                            ("Next Steps", "Return the answer."),
+                            ("Context To Preserve", "SOURCE-FACT-001 is ORCHID-7429."),
+                        ]
+                    )
+                else:
+                    content = "recovered"
+                self.chunk(frame({"content": content}, finish="stop",
+                                 usage={"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}))
+                self.chunk(b"data: [DONE]\n\n")
             elif scenario == "late_content":
                 self.chunk(frame({"content": "ok"}, finish="stop"))
                 self.chunk(frame({"content": "late"}))
diff --git a/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/test.py b/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/test.py
index 0a9e51b59e..4b1c9a5a2d 100644
--- a/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/test.py
+++ b/test-e2e/cases/MODEL-SERVICE-TRANSPORT-D3-001/test.py
@@ -57,6 +57,10 @@
 
 from .catalog_control_cases import test_all_selected_custom_model_controls_reach_real_peer, test_all_selected_hosted_model_controls_reach_real_peer  # noqa: F401
 
+from .context_maintenance_cases import test_shared_context_maintenance_uses_completed_provider_usage  # noqa: F401
+
+from .context_maintenance_cases import test_shared_context_unknown_usage_does_not_block_request, test_explicit_overflow_rebuilds_shared_history_even_when_estimate_fits  # noqa: F401
+
 pytestmark = [pytest.mark.case_id("MODEL-SERVICE-TRANSPORT-D3-001"), pytest.mark.stage("D3")]
 
 
diff --git a/test-e2e/changes/requirements/CONTEXT-MAINTENANCE-20261110.yaml b/test-e2e/changes/requirements/CONTEXT-MAINTENANCE-20261110.yaml
new file mode 100644
index 0000000000..8fde1899ab
--- /dev/null
+++ b/test-e2e/changes/requirements/CONTEXT-MAINTENANCE-20261110.yaml
@@ -0,0 +1,54 @@
+schema_version: '1.0'
+change:
+  change_id: CONTEXT-MAINTENANCE-20261110
+  change_type: requirement
+  requirement_id: CONTEXT-MAINTENANCE-20261110
+  title: Share non-blocking context maintenance and provider usage observations
+  description: Keep context estimates advisory, preserve history when maintenance
+    fails, and retain bounded provider overflow recovery for both agent runtimes.
+  affected_features:
+    existing:
+    - SDK-MODEL-TRANSPORT
+    added: []
+    modified:
+    - F-171
+    retired: []
+  affected_cases:
+    existing:
+    - AGENT-HISTORY-D3-001
+    added: []
+    modified:
+    - MODEL-SERVICE-TRANSPORT-D3-001
+    retired: []
+  traditional_ut:
+    decision: strengthen
+    behaviors:
+    - Pair actual provider input usage with the completed request; reject stale or
+      non-provider observations.
+    - Account for changed messages/tools without treating estimates as admission limits.
+    - Preserve original history when maintenance fails; bound each request independently
+      and retain cancellation.
+    - Keep output length separate from explicit input overflow; preserve retry and
+      saved output-cap contracts.
+    reason: Existing history and recovery assertions are reused; focused observation,
+      fail-open, and repeated-maintenance regressions are missing.
+    test_selectors:
+    - test/sdk/core/agents/test_context_usage_observation.py
+    - test/sdk/core/agents/test_context_maintenance_failures.py
+    - test/sdk/core/agents/test_context_helper_contracts.py
+    - test/sdk/core/agents/test_history_context_runtime.py
+    - test/sdk/core/agents/test_execution_history_compaction.py
+    - test/sdk/core/agents/test_context_manager_assembly.py
+    - test/sdk/core/model_service/test_legacy_response_observation.py
+    - test/sdk/core/model_service/test_context_recovery.py
+    - test/sdk/core/model_service/test_legacy_context_recovery.py
+    - test/sdk/core/model_service/test_legacy_context_budget.py
+    - test/sdk/core/models/test_openai_llm.py
+    - test/sdk/monitor/test_monitoring.py
+    - test/sdk/core/context_runtime/test_runtimes.py
+    - test/sdk/core/agents/test_nexent_agent.py
+    - test/sdk/core/agents/test_nexent_agent_context_runtime_factory.py
+    - test/sdk/core/agents/test_execution_history_review_regressions.py
+  notes: Existing runtime/model ownership mappings apply. Transport preparation is
+    self-contained with owned loopback server/client cleanup; real Agent acceptance
+    retains separate Phoenix evidence. No new D1 duplicates UT.
diff --git a/test-e2e/features/F-171/feature.yaml b/test-e2e/features/F-171/feature.yaml
index db7e4ef5d6..5ea658c0c2 100644
--- a/test-e2e/features/F-171/feature.yaml
+++ b/test-e2e/features/F-171/feature.yaml
@@ -10,6 +10,10 @@ feature:
   business_rules:
   - rule_id: F-171-RULE-01
     description: agent/tool/context gateway 契约：上下文预算与压缩、受管 MCP/模型并发、沙箱桥接线程回收、工具与观察者消息
+  - rule_id: F-171-RULE-02
+    description: Context estimates remain advisory; provider input observations inform
+      bounded maintenance without blocking inference or losing protected context and
+      original execution results.
   required_test_stages:
   - D1
   source:
diff --git a/test/sdk/core/agents/test_context_helper_contracts.py b/test/sdk/core/agents/test_context_helper_contracts.py
index ac91e9a52a..711f266421 100644
--- a/test/sdk/core/agents/test_context_helper_contracts.py
+++ b/test/sdk/core/agents/test_context_helper_contracts.py
@@ -55,6 +55,9 @@ def test_summary_output_normalization_and_fallback(caplog):
         (RuntimeError("maximum context length exceeded"), True),
         (RuntimeError("input is too long"), True),
         (RuntimeError("temporary connection failure"), False),
+        (RuntimeError("completion token limit reached; finish_reason=length"), False),
+        (RuntimeError("output token limit reached"), False),
+        (RuntimeError("reduce the length of the generated answer"), False),
     ],
 )
 def test_context_length_error_detection(error, expected):
@@ -580,3 +583,10 @@ def test_context_manager_memory_rendering_and_change_reasons():
     assert manager._change_reasons("third", manager._previous_stable_items) == [
         "unexpected_nondeterminism"
     ]
+
+
+def test_typed_rate_limit_never_enters_context_recovery_from_message_text():
+    from nexent.core.model_errors import ModelErrorCode, ModelInvocationTerminalError
+    error = ModelInvocationTerminalError(ModelErrorCode.RATE_LIMIT_EXHAUSTED, 3,
+                                         cause=RuntimeError('too many tokens per minute'))
+    assert _is_context_length_error(error) is False
diff --git a/test/sdk/core/agents/test_context_maintenance_failures.py b/test/sdk/core/agents/test_context_maintenance_failures.py
new file mode 100644
index 0000000000..5ccbf77e94
--- /dev/null
+++ b/test/sdk/core/agents/test_context_maintenance_failures.py
@@ -0,0 +1,144 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
+"""Main inference survives auxiliary maintenance failures and exhausted estimates."""
+
+import pytest
+from nexent.core.agents.context import (
+    ContextItemInput,
+    ContextManager,
+    ContextManagerConfig,
+)
+from nexent.core.concurrency.cancellation import RunTerminated
+from smolagents.memory import TaskStep
+from test_history_context_runtime import (
+    _Memory,
+    _rendered_text,
+    _summary_and_turns,
+    _SummaryModel,
+    _SystemPrompt,
+)
+
+
+def test_later_request_still_compacts_after_prior_attempt_budget_is_used(monkeypatch):
+    monkeypatch.setattr("smolagents.memory.SystemPromptStep", _SystemPrompt)
+    manager = ContextManager(ContextManagerConfig(
+        soft_input_budget_tokens=20,
+        policy_layers={"request": {"processing_mode": "adaptive_compact"}},
+    ))
+    memory = _Memory([TaskStep(task="current")])
+    run = manager.prepare_run_context(memory, "", _summary_and_turns())
+    manager._run_compaction_attempts = 3
+    model = _SummaryModel()
+    result = manager.assemble_final_context(
+        model=model, memory=memory, current_run_start_idx=0, run_context=run,
+    )
+    assert model.calls
+    assert 1 <= result.evidence.compaction_attempts <= 3
+
+
+def test_auxiliary_failure_does_not_reject_main_context(monkeypatch):
+    monkeypatch.setattr("smolagents.memory.SystemPromptStep", _SystemPrompt)
+    manager = ContextManager(ContextManagerConfig(
+        soft_input_budget_tokens=20,
+        policy_layers={"request": {"processing_mode": "adaptive_compact"}},
+    ))
+    memory = _Memory([TaskStep(task="current protected task")])
+    run = manager.prepare_run_context(memory, "", _summary_and_turns())
+    def failed_compression(*args):
+        raise RuntimeError("auxiliary service unavailable")
+    monkeypatch.setattr(manager._history_compressor, "compress", failed_compression)
+    result = manager.assemble_final_context(
+        model=_SummaryModel(), memory=memory, current_run_start_idx=0, run_context=run,
+    )
+    rendered = _rendered_text(result.messages)
+    assert "new question " * 100 in rendered
+    assert "current protected task" in rendered
+    assert result.evidence.compaction_stop_reason == "maintenance_failed"
+    assert result.evidence.new_summary_coverage is None
+
+
+
+def test_cooperative_cancellation_is_not_swallowed_by_maintenance(monkeypatch):
+    monkeypatch.setattr("smolagents.memory.SystemPromptStep", _SystemPrompt)
+    manager = ContextManager(ContextManagerConfig(
+        soft_input_budget_tokens=20,
+        policy_layers={"request": {"processing_mode": "adaptive_compact"}},
+    ))
+    memory = _Memory([TaskStep(task="current")])
+    run = manager.prepare_run_context(memory, "", _summary_and_turns())
+    def cancelled(*args):
+        raise RunTerminated()
+    monkeypatch.setattr(manager._history_compressor, "compress", cancelled)
+    with pytest.raises(RunTerminated):
+        manager.assemble_final_context(
+            model=_SummaryModel(), memory=memory, current_run_start_idx=0, run_context=run,
+        )
+
+
+def test_truncated_summary_is_rejected_without_input_compression_retry():
+    from nexent.core.agents.context.llm_summary import LLMSummary
+    from nexent.core.agents.context.step_renderer import StepRenderer
+    config = ContextManagerConfig()
+    model = _SummaryModel()
+    model.last_finish_reason = 'length'
+    result = LLMSummary(config, StepRenderer(config)).generate_summary('original facts', model)
+    assert result.summary_text is None
+    assert len(model.calls) == 1
+    assert len(result.records) == 1
+
+
+def test_step_metrics_compare_the_same_context_estimate_basis(monkeypatch):
+    from types import SimpleNamespace
+    from unittest.mock import Mock
+
+    from nexent.core.agents.core_agent import CoreAgent
+    agent = CoreAgent.__new__(CoreAgent)
+    agent.step_metrics = []
+    agent._last_uncompressed_est = 400
+    agent.context_runtime = SimpleNamespace(
+        compression_stats=lambda: {'calls': 0}, token_counts=lambda: {'uncompressed': 400, 'compressed': 400},
+        chars_per_token=1.5, token_threshold=1000,
+    )
+    monkeypatch.setattr('nexent.core.agents.core_agent.get_monitoring_manager', lambda: Mock())
+    step = SimpleNamespace(step_number=1, token_usage=None,
+                           model_input_messages=[{'role': 'user', 'content': 'small'}],
+                           model_output_message={'role': 'assistant', 'content': 'answer'})
+    agent._collect_step_metrics(step)
+    assert agent.step_metrics[0]['memory_state']['estimated_input_tokens'] == 400
+    assert agent.step_metrics[0]['compression_ratio'] == 0.0
+
+
+def test_explicit_overflow_can_compact_history_below_the_estimated_threshold(monkeypatch):
+    monkeypatch.setattr('smolagents.memory.SystemPromptStep', _SystemPrompt)
+    saved = []
+    manager = ContextManager(ContextManagerConfig(
+        effective_input_limit_tokens=1_000_000, history_summary_sink=saved.append,
+        policy_layers={'request': {'processing_mode': 'adaptive_compact'}},
+    ))
+    memory = _Memory([TaskStep(task='protected current task')])
+    run = manager.prepare_run_context(memory, '', _summary_and_turns())
+    model = _SummaryModel()
+    result = manager.assemble_final_context(
+        model=model, memory=memory, current_run_start_idx=0, run_context=run, force_compaction=True,
+    )
+    assert len(model.calls) == 1 and len(saved) == 1
+    assert result.evidence.history_compression_triggered
+    assert 'protected current task' in _rendered_text(result.messages)
+    assert result.evidence.new_summary_coverage == 22
+
+
+def test_accepted_checkpoint_is_not_repeatedly_summarized_without_new_turns(monkeypatch):
+    monkeypatch.setattr('smolagents.memory.SystemPromptStep', _SystemPrompt)
+    manager = ContextManager(ContextManagerConfig(
+        effective_input_limit_tokens=1000,
+        policy_layers={'request': {'processing_mode': 'adaptive_compact'}},
+    ))
+    memory = _Memory([TaskStep(task='current')])
+    summary = ContextItemInput(id='summary:10', type='history_summary', content={
+        'unit_id': 10, 'summary': 'accepted checkpoint ' * 1000, 'covered_through_message_id': 20,
+    })
+    run = manager.prepare_run_context(memory, '', [summary])
+    model = _SummaryModel()
+    result = manager.assemble_final_context(model=model, memory=memory, current_run_start_idx=0, run_context=run)
+    assert model.calls == []
+    assert result.evidence.history_compression_triggered is False
+    assert 'accepted checkpoint ' * 1000 in _rendered_text(result.messages)
diff --git a/test/sdk/core/agents/test_context_usage_observation.py b/test/sdk/core/agents/test_context_usage_observation.py
new file mode 100644
index 0000000000..2a9a757162
--- /dev/null
+++ b/test/sdk/core/agents/test_context_usage_observation.py
@@ -0,0 +1,98 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
+"""Provider observations inform advisory context estimates without admission gates."""
+
+from dataclasses import FrozenInstanceError
+from types import SimpleNamespace
+
+import pytest
+from nexent.core.agents.context import ContextManager, ContextManagerConfig
+from nexent.core.context_runtime.usage import (
+    InputUsageObservation,
+    context_model_key,
+    request_characters,
+)
+
+
+def test_request_measurement_includes_tool_arguments_results_and_schema():
+    messages = [{'role': 'user', 'content': 'hello'}]
+    size = request_characters(messages, [])
+    assert request_characters(messages + [{'role': 'tool', 'tool_call_id': 'call-1', 'content': 'result' * 80}], []) > size
+    assert request_characters(messages, [{'type': 'function', 'function': {'name': 'read', 'description': 'x' * 500}}]) > size
+    assert request_characters([{'role': 'assistant', 'content': '', 'tool_calls': [{'id': 'call-1', 'function': {'arguments': 'x' * 500}}]}], []) > size
+
+
+@pytest.mark.parametrize('tokens,origin', [(None, 'provider'), (10, 'estimated'), (-1, 'provider'), (True, 'provider')])
+def test_invalid_usage_cannot_be_a_provider_anchor(tokens, origin):
+    assert InputUsageObservation.capture(
+        usage=SimpleNamespace(input_tokens=tokens, origin=origin), messages=[], tools=[],
+        model_key='model', request_id='request', attempt=1,
+    ) is None
+
+
+def test_observation_is_immutable_and_calibration_accounts_for_new_payload():
+    messages = [{'role': 'user', 'content': 'x' * 2000}]
+    model = SimpleNamespace(model_id='model')
+    observation = InputUsageObservation.capture(
+        usage=SimpleNamespace(input_tokens=100, origin='provider'), messages=messages, tools=[],
+        model_key=context_model_key(model), request_id='request', attempt=1,
+    )
+    with pytest.raises(FrozenInstanceError):
+        observation.input_tokens = 5
+    manager = ContextManager(ContextManagerConfig(chars_per_token=2))
+    manager._observe_model_usage(model)
+    model.last_context_observation = observation
+    manager._observe_model_usage(model)
+    assert manager._estimate_request(messages, []) == 100
+    assert manager._estimate_request(messages + [{'role': 'tool', 'content': 'y' * 1000}], []) >= 600
+    assert manager._estimate_request(messages, [{'description': 'z' * 1000}]) >= 600
+    assert manager._estimate_request([{'role': 'user', 'content': 'short'}], []) < 100
+    assert manager._estimate_basis == 'provider_calibrated_estimate'
+
+
+def test_switching_model_discards_previous_usage_anchor():
+    messages = [{'role': 'user', 'content': 'x' * 2000}]
+    model = SimpleNamespace(model_id='first')
+    manager = ContextManager(ContextManagerConfig(chars_per_token=2))
+    manager._observe_model_usage(model)
+    model.last_context_observation = InputUsageObservation.capture(
+        usage=SimpleNamespace(input_tokens=100, origin='provider'), messages=messages, tools=[],
+        model_key=context_model_key(model), request_id='request', attempt=1,
+    )
+    manager._observe_model_usage(model)
+    assert manager._estimate_request(messages, []) == 100
+    model.model_id = 'second'
+    manager._observe_model_usage(model)
+    assert manager._estimate_request(messages, []) >= 1000
+    assert manager._estimate_basis == 'local_estimate'
+
+
+@pytest.mark.parametrize('ratio', [0, -1, float('nan'), float('inf')])
+def test_invalid_estimation_ratio_remains_advisory(ratio):
+    manager = ContextManager(ContextManagerConfig(chars_per_token=ratio))
+    assert manager._estimate_request([{'role': 'user', 'content': 'hello'}], []) >= 0
+    from nexent.core.agents.context.runtime import ManagedContextRuntime
+    assert ManagedContextRuntime(manager).chars_per_token > 0
+    observation = InputUsageObservation(100, 50, 'model', 'request', 1)
+    assert observation.estimate(input_characters=100, chars_per_token=ratio) > 100
+
+
+def test_zero_provider_input_is_known_and_cached_tokens_are_not_added():
+    observation = InputUsageObservation.capture(
+        usage=SimpleNamespace(input_tokens=0, cached_input_tokens=100, origin='provider'),
+        messages=[], tools=[], model_key='model', request_id='zero', attempt=1,
+    )
+    assert observation.input_tokens == 0
+    assert observation.estimate(input_characters=observation.input_characters, chars_per_token=1.5) == 0
+
+
+def test_preexisting_response_from_another_run_is_not_used():
+    messages = [{'role': 'user', 'content': 'x' * 2000}]
+    model = SimpleNamespace(model_id='shared')
+    model.last_context_observation = InputUsageObservation.capture(
+        usage=SimpleNamespace(input_tokens=1, origin='provider'), messages=messages, tools=[],
+        model_key=context_model_key(model), request_id='other-conversation', attempt=1,
+    )
+    manager = ContextManager(ContextManagerConfig(chars_per_token=2))
+    manager._observe_model_usage(model)
+    assert manager._estimate_request(messages, []) == 1000
+    assert manager._estimate_basis == 'local_estimate'
diff --git a/test/sdk/core/agents/test_execution_history_compaction.py b/test/sdk/core/agents/test_execution_history_compaction.py
index e431998d79..a43967bb25 100644
--- a/test/sdk/core/agents/test_execution_history_compaction.py
+++ b/test/sdk/core/agents/test_execution_history_compaction.py
@@ -346,13 +346,13 @@ def test_ac003_legacy_observation_fallback_keeps_masking_and_empty_success(obser
     assert len(observer.execution_history._records) == 1
 
 
-def test_ac009_real_agent_read_uses_provider_usage_and_resets_owner_between_runs():
+@pytest.mark.parametrize("provider_input", [5000, 6000, 60000])
+def test_ac009_real_agent_read_is_not_blocked_by_token_counts_and_resets_owner_between_runs(provider_input):
     from smolagents.agents import EMPTY_PROMPT_TEMPLATES
     from smolagents.models import ChatMessage, MessageRole, TokenUsage
     from nexent.core.utils.observer import ProcessType
 
-    # Keep prompt cost controlled so the small budget tests reader clipping,
-    # independently of the SDK's full default instruction/tool template.
+    # A locally exhausted context estimate must not prevent authorized recall.
     templates = deepcopy(EMPTY_PROMPT_TEMPLATES)
     templates['system_prompt'] = 'Execute the requested code and return the final answer.'
     agent, observer, runtime, _ = runtime_agent(prompt_templates=templates)
@@ -388,7 +388,7 @@ def capture_read(*args, **kwargs):
         '<final_answer>done</final_answer>',
     ]
     agent.model.generate = Mock(side_effect=[ChatMessage(role=MessageRole.ASSISTANT, content=body,
-        token_usage=TokenUsage(input_tokens=5000, output_tokens=100)) for body in responses])
+        token_usage=TokenUsage(input_tokens=provider_input, output_tokens=100)) for body in responses])
     assert str(agent.run('Print a report once, read its original range, then finish.', max_steps=3)) == 'done'
     assert agent._execution_owner != old_owner
     assert agent.tools['read_execution_output'].owner == agent._execution_owner
@@ -396,12 +396,12 @@ def capture_read(*args, **kwargs):
     original, loaded = [store.get(key) for key in references]
     assert len(read_payloads) == 1 and 'next_offset' in loaded.text
     payload = read_payloads[0]
-    assert 0 < payload['next_offset'] < 4000
+    assert payload['next_offset'] == 4000
     assert payload['text'] == original.text[:payload['next_offset']] and not payload['eof']
     assert loaded.owner == original.owner == agent._execution_owner
     bounded = [call.kwargs['max_rendered_chars'] for call in store.reset_read_budget.call_args_list
                if 'max_rendered_chars' in call.kwargs]
-    assert len(bounded) == 2 and all(0 < value < 4000 for value in bounded)
+    assert len(bounded) == 2 and all(value == 16000 for value in bounded)
     previous_owner = agent._execution_owner
     agent.model.generate = Mock(return_value=ChatMessage(role=MessageRole.ASSISTANT,
                                                         content='<final_answer>fresh run</final_answer>'))
diff --git a/test/sdk/core/agents/test_history_context_runtime.py b/test/sdk/core/agents/test_history_context_runtime.py
index 5759e80e86..efd39db04c 100644
--- a/test/sdk/core/agents/test_history_context_runtime.py
+++ b/test/sdk/core/agents/test_history_context_runtime.py
@@ -354,7 +354,9 @@ def __call__(self, messages, stop_sequences=None):
     assert result.evidence.summary_persist_status == "not_attempted"
     assert result.evidence.new_summary_coverage is None
     rendered = str(result.messages)
-    assert "history limited" in rendered
+    assert "history limited" not in rendered
+    assert "new question " * 100 in _rendered_text(result.messages)
+    assert "new answer " * 100 in _rendered_text(result.messages)
     assert _summary_and_turns()[1].content["user_message"] == "new question " * 100
 
 
diff --git a/test/sdk/core/agents/test_nexent_agent.py b/test/sdk/core/agents/test_nexent_agent.py
index a230d790fd..c686ba388b 100644
--- a/test/sdk/core/agents/test_nexent_agent.py
+++ b/test/sdk/core/agents/test_nexent_agent.py
@@ -186,7 +186,7 @@ def __init__(self, enabled=False, **kwargs):
 
 
 mock_sdk_context_runtime_module = types.ModuleType("sdk.nexent.core.context_runtime")
-mock_sdk_context_runtime_module.__path__ = []
+mock_sdk_context_runtime_module.__path__ = [str(SDK_SOURCE_ROOT / "nexent" / "core" / "context_runtime")]
 mock_sdk_context_runtime_managed_module = types.ModuleType("sdk.nexent.core.context_runtime.managed")
 mock_sdk_context_runtime_managed_module.__path__ = []
 mock_sdk_context_runtime_managed_runtime_module = types.ModuleType(
diff --git a/test/sdk/core/model_service/test_legacy_response_observation.py b/test/sdk/core/model_service/test_legacy_response_observation.py
index 86e2864c15..e609627ab7 100644
--- a/test/sdk/core/model_service/test_legacy_response_observation.py
+++ b/test/sdk/core/model_service/test_legacy_response_observation.py
@@ -58,7 +58,22 @@ def test_failed_next_invocation_never_reuses_previous_response_usage_or_diagnost
     state['error'] = normalize_model_error(ModelTransportError('HTTP_TLS_ERROR'), request_id='failure', attempt=2, max_attempts=3)
     with pytest.raises(ModelInvocationTerminalError):
         model.generate([{'role': 'user', 'content': 'second'}])
+    assert model.last_context_observation is None
     assert model.last_input_token_count is model.last_output_token_count is model.last_finish_reason is None
     assert model.last_prompt_cache_usage is model.last_cached_input_token_count is None
     assert model.last_response_diagnostics['content_char_count'] == 0
     assert model.last_response_diagnostics['reasoning_char_count'] == 0
+
+
+def test_completed_observation_pairs_provider_input_with_actual_request(setup):
+    from nexent.core.context_runtime.usage import context_model_key, request_characters
+    model, state, calls, _observer = setup
+    state["usage"] = TokenUsage(input_tokens=10, output_tokens=0, origin="provider")
+    messages = [{'role': 'user', 'content': 'request body'}]
+    model.generate(messages)
+    observation = model.last_context_observation
+    assert observation.input_tokens == 10
+    assert observation.input_characters == request_characters(messages, [])
+    assert observation.model_key == context_model_key(model)
+    assert observation.request_id == calls[-1][0].request_id
+    assert observation.attempt == 2
diff --git a/test/sdk/core/models/test_openai_llm.py b/test/sdk/core/models/test_openai_llm.py
index f87905ae91..72d64ad2bf 100644
--- a/test/sdk/core/models/test_openai_llm.py
+++ b/test/sdk/core/models/test_openai_llm.py
@@ -872,6 +872,7 @@ def test_call_normal_operation(openai_model_instance):
         # Verify token counts were set
         assert openai_model_instance.last_input_token_count == 10
         assert openai_model_instance.last_output_token_count == 5
+        assert openai_model_instance.last_context_observation.input_tokens == 10
 
 
 def test_call_with_no_think_token_addition(openai_model_instance):
@@ -1098,6 +1099,7 @@ def test_call_with_no_usage_info(openai_model_instance):
         # The implementation estimates tokens from input/output text
         assert openai_model_instance.last_input_token_count >= 0
         assert openai_model_instance.last_output_token_count >= 0
+        assert openai_model_instance.last_context_observation is None
 
 
 def test_call_with_null_tokens(openai_model_instance):
diff --git a/test/sdk/monitor/test_monitoring.py b/test/sdk/monitor/test_monitoring.py
index 9ffe59dbfa..eda9c03f35 100644
--- a/test/sdk/monitor/test_monitoring.py
+++ b/test/sdk/monitor/test_monitoring.py
@@ -974,6 +974,7 @@ def test_record_final_context_evidence_adds_content_free_event(self, mock_trace)
                 compaction_trigger_threshold_tokens=100,
                 compaction_target_tokens=72,
                 raw_token_estimate=110, final_token_estimate=80,
+                token_estimate_basis="provider_calibrated_estimate", observed_provider_input_tokens=90,
                 messages_fingerprint="messages-fp", tools_fingerprint="tools-fp",
                 message_roles=("system", "user", "assistant"),
                 history_message_roles=("user", "assistant"), compression_attempted=True,
@@ -988,6 +989,9 @@ def test_record_final_context_evidence_adds_content_free_event(self, mock_trace)
             assert event_attrs["context.compaction_target_tokens"] == 72
             assert event_attrs["context.tokens.pre_compression"] == 110
             assert event_attrs["context.tokens.post_compression"] == 80
+            assert event_attrs["context.tokens.estimate_basis"] == "provider_calibrated_estimate"
+            assert event_attrs["context.tokens.provider_input_known"] is True
+            assert event_attrs["context.tokens.observed_provider_input"] == 90
             assert event_attrs["context.compression.attempted"] is True