diff --git a/agent_core/core/errors.py b/agent_core/core/errors.py
index 26e130a5..d2f96f48 100644
--- a/agent_core/core/errors.py
+++ b/agent_core/core/errors.py
@@ -34,7 +34,8 @@ class ErrorCategory(str, Enum):
     RATE_LIMIT = "rate_limit"  # 429 — transient
     QUOTA = "quota"  # 429 + monthly/account scope (separable from per-min)
     MODEL = "model"  # 404, "model_not_found"
-    BAD_REQUEST = "bad_request"  # 400 — request malformed (context overflow, etc.)
+    BAD_REQUEST = "bad_request"  # 400 — request malformed
+    CONTEXT_OVERFLOW = "context_overflow"  # request exceeds the model's context window
     BLOCKED = "blocked"  # safety filter (Gemini/Anthropic)
     SERVER = "server"  # 5xx, "overloaded_error"
     CONNECTION = "connection"  # network / timeout / DNS
diff --git a/agent_core/core/impl/action/router.py b/agent_core/core/impl/action/router.py
index 5ac68824..4a524b7e 100644
--- a/agent_core/core/impl/action/router.py
+++ b/agent_core/core/impl/action/router.py
@@ -10,7 +10,7 @@
 
 import json
 import ast
-from typing import Optional, List, Dict, Any, Tuple
+from typing import Callable, Optional, List, Dict, Any, Tuple
 
 from agent_core.core.state import get_state, get_session_or_none
 from agent_core.decorators import profile, OperationCategory
@@ -21,6 +21,8 @@
 from agent_core.core.impl.llm.errors import LLMConsecutiveFailureError
 from agent_core.core.errors import ClassifiedError, ErrorCategory, ErrorInfo
 from agent_core.core.prompts import SELECT_ACTION_PROMPT
+from agent_core.core.impl.llm.interface import LLMContextOverflowError
+from agent_core import get_event_stream_manager
 from agent_core.utils.logger import logger
 
 
@@ -135,13 +137,20 @@ async def select_action_in_session(
             action_candidates=self._format_candidates(action_candidates),
             integration_essentials=integration_essentials,
         )
-        full_prompt = SELECT_ACTION_PROMPT.format(
-            session_state=session_state,
-            event_stream=event_stream_content,
-            query=query,
-            action_candidates=self._format_candidates(action_candidates),
-            integration_essentials=integration_essentials,
-        )
+        candidates_text = self._format_candidates(action_candidates)
+
+        def render_prompt() -> str:
+            # Rendered from the CURRENT stream, so a fold made while deciding
+            # (see _prompt_for_decision) is reflected in what is sent.
+            return SELECT_ACTION_PROMPT.format(
+                session_state=session_state,
+                event_stream=self.context_engine.get_event_stream(session_id=session_id),
+                query=query,
+                action_candidates=candidates_text,
+                integration_essentials=integration_essentials,
+            )
+
+        full_prompt = render_prompt()
 
         max_format_retries = 3
         current_prompt = full_prompt
@@ -154,6 +163,7 @@ async def select_action_in_session(
                 call_type=LLMCallType.ACTION_SELECTION,
                 session_id=session_id,
                 prompt_name=decision_prompt_name,
+                render_prompt=render_prompt,
             )
 
             # Parse parallel action decisions with format error detection
@@ -167,7 +177,7 @@ async def select_action_in_session(
 
                 if attempt < max_format_retries - 1:
                     current_prompt = self._augment_prompt_with_format_error(
-                        full_prompt, attempt + 1, decision, format_error
+                        render_prompt(), attempt + 1, decision, format_error
                     )
                     continue
                 else:
@@ -214,6 +224,7 @@ async def _prompt_for_decision(
         call_type: str = LLMCallType.ACTION_SELECTION,
         session_id: Optional[str] = None,
         prompt_name: Optional[str] = None,
+        render_prompt: Optional[Callable[[], str]] = None,
     ) -> Dict[str, Any]:
         """
         Prompt the LLM for an action decision with session caching support.
@@ -230,6 +241,10 @@ async def _prompt_for_decision(
         max_retries = 3
         last_error: Optional[Exception] = None
         current_prompt = prompt
+        # One fold per decision: a request that still does not fit after the
+        # stream has been folded to keep_recent_tokens is a configuration
+        # problem, not something another fold can solve.
+        folded = False
 
         # Get current task_id for session cache (if running in a task)
         # Use session_id if provided, otherwise fall back to global state
@@ -264,8 +279,6 @@ async def _prompt_for_decision(
                     if has_session:
                         # Session is registered (complex task) - use session caching
                         # CRITICAL: Use session-specific stream to prevent event leakage
-                        from agent_core import get_event_stream_manager
-
                         event_stream_manager = get_event_stream_manager()
                         # Use get_stream_by_id with session_id to get the correct task's stream
                         effective_session_id = session_id or current_task_id
@@ -287,7 +300,28 @@ async def _prompt_for_decision(
                                 )
                             )
 
-                            if has_delta:
+                            if (
+                                has_delta
+                                and not folded
+                                and render_prompt is not None
+                                and not self.llm_interface.fits_context(
+                                    current_task_id, call_type, system_prompt, delta_events
+                                )
+                            ):
+                                # The next request would exceed the budget: fold the
+                                # stream now and start a fresh session from it.
+                                logger.info(
+                                    f"[SESSION CACHE] Next request exceeds the context "
+                                    f"budget; folding the event stream for {call_type}"
+                                )
+                                stream.summarize_by_LLM()
+                                folded = True
+                                current_prompt = render_prompt()
+                                self.context_engine.reset_event_stream_sync(
+                                    call_type, session_id=effective_session_id
+                                )
+                                has_synced_before = False
+                            elif has_delta:
                                 # Send only the new events
                                 logger.info(
                                     f"[SESSION CACHE] Sending delta events for {call_type}"
@@ -308,7 +342,7 @@ async def _prompt_for_decision(
                                 logger.info(
                                     f"[SESSION CACHE] No delta events, resetting cache for {call_type}"
                                 )
-                                self.llm_interface.end_session_cache(
+                                self.llm_interface.reset_session_history(
                                     current_task_id, call_type
                                 )
                                 self.context_engine.reset_event_stream_sync(
@@ -318,7 +352,12 @@ async def _prompt_for_decision(
                                 has_synced_before = False
 
                         if not has_synced_before:
-                            # First call with session - send full prompt to establish session
+                            # First call with session - send full prompt to establish session.
+                            # Also reached after the stream folds (the fold clears the
+                            # sync point); any accumulated turns are stale by then.
+                            self.llm_interface.reset_session_history(
+                                current_task_id, call_type
+                            )
                             logger.info(
                                 f"[SESSION CACHE] Creating new session for {call_type} (first call)"
                             )
@@ -375,6 +414,26 @@ async def _prompt_for_decision(
             except LLMConsecutiveFailureError:
                 # Fatal: LLM is in a broken state - re-raise immediately, do not retry
                 raise
+            except LLMContextOverflowError:
+                # The request could not fit (pre-flight, or the provider said so).
+                # Fold once and retry with a prompt rendered from the folded stream.
+                if folded or render_prompt is None or not (current_task_id and is_task):
+                    raise
+                stream = get_event_stream_manager().get_stream_by_id(session_id or current_task_id)
+                if stream is None:
+                    raise
+                logger.warning(
+                    f"[SESSION CACHE] Request exceeded the context budget; folding the "
+                    f"event stream and retrying once for {call_type}"
+                )
+                stream.summarize_by_LLM()
+                folded = True
+                current_prompt = render_prompt()
+                self.llm_interface.reset_session_history(current_task_id, call_type)
+                self.context_engine.reset_event_stream_sync(
+                    call_type, session_id=session_id or current_task_id
+                )
+                continue
             except RuntimeError as e:
                 # LLM provider error (empty response, API error, auth failure, etc.)
                 # — a recognized, user-actionable failure, not a code bug. The
diff --git a/agent_core/core/impl/event_stream/event_stream.py b/agent_core/core/impl/event_stream/event_stream.py
index 97b46313..b492d577 100644
--- a/agent_core/core/impl/event_stream/event_stream.py
+++ b/agent_core/core/impl/event_stream/event_stream.py
@@ -9,7 +9,6 @@
 APIs:
   log(kind, message, severity="INFO") -> int (event index)
   to_prompt_snapshot(max_events=60, include_summary=True) -> str
-  summarize_if_needed()  # auto-rollup when thresholds exceeded
   summarize_by_rule()        # force summarization of oldest chunk
   summarize_by_LLM()        # force summarization of oldest chunk
 """
@@ -32,8 +31,8 @@
 SEVERITIES = ("DEBUG", "INFO", "WARN", "ERROR")
 
 
-def _configured_context_limits() -> Tuple[int, int]:
-    """Read the summarization thresholds from settings.json.
+def _configured_keep_recent_tokens() -> int:
+    """Read context.keep_recent_tokens from settings.json.
 
     app.config owns the defaults and already absorbs a missing file, bad JSON
     and out-of-range values, so there is nothing left to guard here — a raised
@@ -48,9 +47,9 @@ def _configured_context_limits() -> Tuple[int, int]:
     Read once per stream, so a settings.json edit applies to sessions created
     after it; the main session's stream needs a restart.
     """
-    from app.config import get_context_limits
+    from app.config import get_keep_recent_tokens
 
-    return get_context_limits()
+    return get_keep_recent_tokens()
 
 
 # Messages longer than this are externalized to a temp file and replaced with a
@@ -117,35 +116,17 @@ def __init__(
         llm: LLMInterfaceProtocol,
         temp_dir: Path | None = None,
     ) -> None:
-        # Thresholds come from settings.json — there is no per-stream override,
-        # so every session folds on the same rules. Tests pin them by patching
-        # _configured_context_limits (see the event_stream_limits fixture).
-        summarize_at_tokens, tail_keep_after_summarize_tokens = (
-            _configured_context_limits()
-        )
-
+        # The stream never decides to fold on its own. The router asks for a
+        # fold (summarize_by_LLM) when the NEXT REQUEST would not fit the
+        # context budget; the only stream-side setting is how much recent
+        # history a fold keeps. Tests pin it by patching
+        # _configured_keep_recent_tokens (see the event_stream_limits fixture).
         self.head_summary: Optional[str] = None
         self.llm = llm
         self.tail_events: List[EventRecord] = []
-        self.summarize_at_tokens = summarize_at_tokens
-        self.tail_keep_after_summarize_tokens = tail_keep_after_summarize_tokens
+        self.tail_keep_after_summarize_tokens = _configured_keep_recent_tokens()
         self.temp_dir = temp_dir
 
-        MINIMUM_BUFFER_TOKENS_BEFORE_NEXT_SUMMARIZATION = 2000
-        if (
-            tail_keep_after_summarize_tokens
-            + MINIMUM_BUFFER_TOKENS_BEFORE_NEXT_SUMMARIZATION
-            > summarize_at_tokens
-        ):
-            logger.warning(
-                f"[EventStream] Value for tail_keep_after_summarize_tokens ({tail_keep_after_summarize_tokens}) "
-                f"is too large relative to summarize_at_tokens ({summarize_at_tokens}). "
-                f"Resetting tail_keep_after_summarize_tokens to {summarize_at_tokens - MINIMUM_BUFFER_TOKENS_BEFORE_NEXT_SUMMARIZATION}"
-            )
-            self.tail_keep_after_summarize_tokens = (
-                summarize_at_tokens - MINIMUM_BUFFER_TOKENS_BEFORE_NEXT_SUMMARIZATION
-            )
-
         self._lock = threading.RLock()
         self._total_tokens: int = 0
         # Wall-clock of the last `datetime` marker pushed into the stream (None
@@ -320,7 +301,6 @@ def log(
             self._total_tokens += get_cached_token_count(rec)
             # Summarization runs inside the lock - blocks other log() calls
             # until summarization completes
-            self.summarize_if_needed()
             return len(self.tail_events) - 1
 
     # Convenience wrappers for common event families (optional use)
@@ -386,21 +366,6 @@ def _externalize_message(
             )
             return message
 
-    def summarize_if_needed(self) -> None:
-        """
-        Trigger summarization when the tail token count exceeds the configured threshold.
-
-        This is a SYNCHRONOUS blocking call - if summarization is needed, it runs
-        immediately and waits for completion before returning.
-        """
-        if self._total_tokens < self.summarize_at_tokens:
-            return
-
-        logger.debug(
-            f"[EventStream] Triggering summarization: {self._total_tokens} tokens >= {self.summarize_at_tokens} threshold"
-        )
-        self.summarize_by_LLM()
-
     def _find_token_cutoff(self, events: List[EventRecord], keep_tokens: int) -> int:
         """
         Find the cutoff index such that events from cutoff to end have approximately keep_tokens.
@@ -510,8 +475,6 @@ def summarize_by_LLM(self) -> None:
         # verbatim BEFORE deciding whether an LLM call is warranted — that alone
         # often drops the stream back under the threshold for free.
         if self._shrink_pinned_oversize(cutoff):
-            if self._total_tokens < self.summarize_at_tokens:
-                return
             # Budget changed; the fold boundary moves with it.
             cutoff = self._find_token_cutoff(
                 self.tail_events, self.tail_keep_after_summarize_tokens
diff --git a/agent_core/core/impl/llm/errors.py b/agent_core/core/impl/llm/errors.py
index 639d8488..716ddefe 100644
--- a/agent_core/core/impl/llm/errors.py
+++ b/agent_core/core/impl/llm/errors.py
@@ -146,6 +146,10 @@ def provider_display_name(provider: Optional[str]) -> str:
     ErrorCategory.QUOTA: "quota exceeded",
     ErrorCategory.MODEL: "the selected model is not available",
     ErrorCategory.BAD_REQUEST: "the request was rejected",
+    ErrorCategory.CONTEXT_OVERFLOW: "the request exceeded the model's context window",
+    ErrorCategory.CONTEXT_OVERFLOW: "the request exceeded the model's context window",
+    ErrorCategory.CONTEXT_OVERFLOW: "the request exceeded the model's context window",
+    ErrorCategory.CONTEXT_OVERFLOW: "the request exceeded the model's context window",
     ErrorCategory.BLOCKED: "blocked by the provider's safety filter",
     ErrorCategory.SERVER: "the provider is unavailable",
     ErrorCategory.CONNECTION: "unable to reach the provider",
@@ -381,7 +385,7 @@ def _classify_openai_compat(exc: Exception, provider: str) -> LLMErrorInfo:
         elif code == "rate_limit_exceeded":
             category = ErrorCategory.RATE_LIMIT
         elif code == "context_length_exceeded":
-            category = ErrorCategory.BAD_REQUEST
+            category = ErrorCategory.CONTEXT_OVERFLOW
         elif code in ("model_not_found", "invalid_model"):
             category = ErrorCategory.MODEL
         elif code == "invalid_api_key":
diff --git a/agent_core/core/impl/llm/interface.py b/agent_core/core/impl/llm/interface.py
index 38dea7fc..91f40850 100644
--- a/agent_core/core/impl/llm/interface.py
+++ b/agent_core/core/impl/llm/interface.py
@@ -49,7 +49,7 @@
 
 # Logging setup - use shared agent_core logger for consistency
 from agent_core.utils.logger import logger
-from agent_core.utils.token import billable_tokens
+from agent_core.utils.token import billable_tokens, count_tokens
 
 # Per-call metadata (prompt identity + start time) propagated from the public
 # entry methods down to the capture chokepoint (_call_log_to_db) without
@@ -70,6 +70,25 @@
 )
 
 
+# Session key of the session call in flight. Set by the public session entry
+# points and read by _report_usage_async, so the input count the provider
+# reports lands on the right session. A ContextVar rather than an attribute:
+# asyncio.to_thread copies the context into the worker thread, and concurrent
+# sessions run in separate contexts, so they never see each other's value.
+_active_session_key: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar(
+    "_active_session_key", default=None
+)
+
+
+class LLMContextOverflowError(RuntimeError):
+    """The assembled request does not fit the configured context window.
+
+    Raised before anything is sent. It is a local budget error, not a provider
+    failure, so callers must not count it against the provider or retry it on
+    a fallback provider with the same payload.
+    """
+
+
 class _EmptyResponse(Exception):
     """Raised when a provider returns empty/error content and the failure has already been counted.
 
@@ -266,17 +285,14 @@ def __init__(
         #   by OR to the last cacheable block (i.e. last assistant message)
         # - gemini:    growing `contents` array; implicit caching matches
         #   the longest stable prefix automatically (no marker required)
-        self._anthropic_session_messages: Dict[str, List[dict]] = {}
-        self._bedrock_session_messages: Dict[str, List[dict]] = {}
-        self._openrouter_anthropic_session_messages: Dict[str, List[dict]] = {}
-        self._gemini_session_messages: Dict[str, List[dict]] = {}
-        # openai / deepseek / grok / non-Claude openrouter: stateless
-        # chat-completions APIs with no server-side session. We accumulate a
-        # growing [user, assistant, ...] history here and resend it each turn
-        # so the model retains earlier context (the delta-only approach dropped
-        # everything but the newest turn); the stable growing prefix also feeds
-        # prompt_cache_key prefix caching.
-        self._openai_compat_session_messages: Dict[str, List[dict]] = {}
+        # Accumulated multi-turn history per session, keyed by
+        # "<task_id>:<call_type>". One interface serves one provider, so the
+        # message shape inside is whatever that provider's session branch
+        # builds; the container itself knows nothing about providers.
+        self._session_histories: Dict[str, List[dict]] = {}
+        # Input tokens the provider reported for each session's last request:
+        # the exact size of the context that request carried.
+        self._last_input_tokens: Dict[str, int] = {}
 
         if ctx["byteplus"]:
             self.api_key = ctx["byteplus"]["api_key"]
@@ -426,11 +442,8 @@ def reinitialize(
                 # Real provider change: message formats differ across
                 # providers, so the accumulated histories aren't reusable.
                 self._session_system_prompts = {}
-                self._anthropic_session_messages = {}
-                self._bedrock_session_messages = {}
-                self._openrouter_anthropic_session_messages = {}
-                self._gemini_session_messages = {}
-                self._openai_compat_session_messages = {}
+                self._session_histories = {}
+                self._last_input_tokens = {}
 
             # Reinitialize Gemini cache manager
             if self._gemini_client:
@@ -481,6 +494,9 @@ def _report_usage_async(
         cached_tokens: int = 0,
     ) -> None:
         """Report usage asynchronously if hook is set."""
+        session_key = _active_session_key.get()
+        if session_key is not None:
+            self._last_input_tokens[session_key] = input_tokens
         if not self._report_usage:
             return
 
@@ -745,6 +761,44 @@ def _try_fallback(self, response: Dict[str, Any], attempt) -> Optional[str]:
                 return content
         return None
 
+    def _check_context_fits(
+        self,
+        system_prompt: Optional[str],
+        user_prompt: Optional[str] = None,
+        messages: Optional[List[dict]] = None,
+    ) -> None:
+        """Refuse a request that cannot fit the configured context window.
+
+        Counts the payload that is about to be sent: the system prompt plus
+        either the single user prompt or every accumulated message. Input and
+        the output reservation share the window.
+        """
+        from app.config import get_context_window
+
+        window = get_context_window()
+        budget = window - self.max_tokens
+
+        total = count_tokens(system_prompt or "")
+        if messages:
+            for message in messages:
+                content = message.get("content", message.get("parts"))
+                if isinstance(content, str):
+                    total += count_tokens(content)
+                elif isinstance(content, list):
+                    for block in content:
+                        text = block.get("text") if isinstance(block, dict) else None
+                        if isinstance(text, str):
+                            total += count_tokens(text)
+        elif user_prompt:
+            total += count_tokens(user_prompt)
+
+        if total > budget:
+            raise LLMContextOverflowError(
+                f"Request of ~{total} input tokens exceeds the {budget}-token budget "
+                f"({window} window - {self.max_tokens} reserved for output) for "
+                f"{self.provider}/{self.model}."
+            )
+
     def _generate_response_sync(
         self,
         system_prompt: Optional[str] = None,
@@ -790,6 +844,7 @@ def _generate_response_sync(
             )
             if _transport is None:  # pragma: no cover
                 raise RuntimeError(f"Unknown provider {self.provider!r}")
+            self._check_context_fits(system_prompt, user_prompt)
             response = _transport(self, system_prompt, user_prompt, json_mode=json_mode)
             content = response.get("content", "").strip()
 
@@ -814,6 +869,14 @@ def _generate_response_sync(
                 # served fallback turn is a success; an exhausted (or
                 # unconfigured) chain falls through to the exact historical
                 # failure path below.
+                if (
+                    error_info is not None
+                    and error_info.category == ErrorCategory.CONTEXT_OVERFLOW
+                ):
+                    # The provider refused the request for size. Not a provider
+                    # failure, and a fallback provider would get the same payload:
+                    # surface it so the caller can fold the stream and retry.
+                    raise LLMContextOverflowError(error_detail)
                 served = self._try_fallback(
                     response,
                     lambda fb: fb._generate_response_sync(
@@ -862,6 +925,9 @@ def _generate_response_sync(
         except LLMConsecutiveFailureError:
             # Re-raise consecutive failure errors without incrementing counter
             raise
+        except LLMContextOverflowError:
+            # Nothing was sent; not a provider failure. Do not count or fall back.
+            raise
         except _EmptyResponse as e:
             # Failure already counted above; convert back to RuntimeError for callers.
             raise RuntimeError(str(e)) from None
@@ -1021,11 +1087,8 @@ def end_session_cache(self, task_id: str, call_type: str) -> None:
         # Clean up stored system prompt and multi-turn message histories
         session_key = f"{task_id}:{call_type}"
         system_prompt = self._session_system_prompts.pop(session_key, None)
-        self._anthropic_session_messages.pop(session_key, None)
-        self._bedrock_session_messages.pop(session_key, None)
-        self._openrouter_anthropic_session_messages.pop(session_key, None)
-        self._gemini_session_messages.pop(session_key, None)
-        self._openai_compat_session_messages.pop(session_key, None)
+        self._session_histories.pop(session_key, None)
+        self._last_input_tokens.pop(session_key, None)
 
         # Clean up provider-specific caches
         if self.provider == "byteplus" and self._byteplus_cache_manager:
@@ -1034,6 +1097,46 @@ def end_session_cache(self, task_id: str, call_type: str) -> None:
             # Invalidate the explicit cache for this system prompt + call_type
             self._gemini_cache_manager.invalidate_cache(system_prompt, call_type)
 
+    def reset_session_history(self, task_id: str, call_type: str) -> None:
+        """Drop the accumulated turns for a session after its event stream folded.
+
+        The session stays registered, so the router keeps taking the session
+        path and the next call re-establishes the prefix from the current
+        stream. ``end_session_cache`` is for the end of a task; using it here
+        also dropped the registration and pushed every following turn onto the
+        stateless path.
+        """
+        session_key = f"{task_id}:{call_type}"
+        self._session_histories.pop(session_key, None)
+        self._last_input_tokens.pop(session_key, None)
+        if self._byteplus_cache_manager:
+            # This provider keeps the history server-side; end that chain too.
+            self._byteplus_cache_manager.end_session(task_id, call_type)
+
+    def last_input_tokens(self, task_id: str, call_type: str) -> Optional[int]:
+        """Input tokens the provider reported for this session's last request."""
+        return self._last_input_tokens.get(f"{task_id}:{call_type}")
+
+    def fits_context(
+        self, task_id: str, call_type: str, system_prompt: Optional[str], pending: str
+    ) -> bool:
+        """Whether this session's next request fits inside the context budget.
+
+        Projection: the provider's own input count for the previous request
+        (exact, and already paid for) plus a local count of only what is new,
+        plus the output reservation, against the window less the headroom the
+        summary request needs. On a session's first request there is no
+        provider count yet, so the whole prompt is counted locally.
+        """
+        from app.config import get_context_window, get_reserve_tokens
+
+        last = self._last_input_tokens.get(f"{task_id}:{call_type}")
+        if last is None:
+            projected = count_tokens(system_prompt or "") + count_tokens(pending)
+        else:
+            projected = last + count_tokens(pending)
+        return projected + self.max_tokens <= get_context_window() - get_reserve_tokens()
+
     def end_all_session_caches(self, task_id: str) -> None:
         """End ALL session/explicit caches for a task (all call types).
 
@@ -1058,16 +1161,9 @@ def end_all_session_caches(self, task_id: str) -> None:
         # Clean up multi-turn message histories across all providers that
         # accumulate (anthropic, bedrock, openrouter-via-claude, gemini,
         # openai-subscription).
-        for buffer in (
-            self._anthropic_session_messages,
-            self._bedrock_session_messages,
-            self._openrouter_anthropic_session_messages,
-            self._gemini_session_messages,
-            self._openai_compat_session_messages,
-        ):
-            stale = [k for k in buffer if k.startswith(f"{task_id}:")]
-            for key in stale:
-                buffer.pop(key, None)
+        for state in (self._session_histories, self._last_input_tokens):
+            for key in [k for k in state if k.startswith(f"{task_id}:")]:
+                state.pop(key, None)
 
         # Clean up provider-specific caches
         if self.provider == "byteplus" and self._byteplus_cache_manager:
@@ -1077,43 +1173,6 @@ def end_all_session_caches(self, task_id: str) -> None:
             for system_prompt, call_type in prompts_and_types:
                 self._gemini_cache_manager.invalidate_cache(system_prompt, call_type)
 
-    def _trim_openai_compat_history(self, history: List[dict]) -> None:
-        """Bound an accumulated openai-compat session history IN PLACE.
-
-        Stateless resends grow every turn, so cap the history to keep
-        ``[system + history + new turn + response]`` inside the model's context
-        window. This is a safety backstop — the agent's summarization-driven
-        session reset (which clears the whole buffer via ``end_session_cache``)
-        normally fires first.
-
-        Trimming preserves the FIRST user/assistant pair — the grounding turn
-        carrying the original query / Definition of Done — and drops the oldest
-        MIDDLE pairs, so we never re-introduce the amnesia this fix exists to
-        prevent. Uses a chars≈4*tokens heuristic.
-        """
-        # Fixed history budget (~240k chars ≈ 60k tokens), leaving room for the
-        # system prompt, newest turn, and response. Provider-independent by
-        # design: we keep no per-model context-window table (no hardcoded model
-        # list), so a single conservative constant governs trimming for every
-        # provider. A power user can raise it via model.context_window_override.
-        max_history_chars = 240_000
-        try:
-            from app.config import get_settings
-
-            override = get_settings().get("model", {}).get("context_window_override")
-            if override:
-                max_history_chars = max(240_000, int(override) * 4)
-        except Exception:
-            pass
-
-        def _size() -> int:
-            return sum(len(m.get("content", "") or "") for m in history)
-
-        # Keep index 0/1 (grounding) and the most recent pair; trim from the
-        # oldest middle pair inward.
-        while len(history) > 4 and _size() > max_history_chars:
-            del history[2:4]
-
     def has_session_cache(self, task_id: str, call_type: str) -> bool:
         """Check if a session/explicit cache is available for the given task and call type.
 
@@ -1196,6 +1255,11 @@ def _finalize_session_response(
             # on a fallback provider. The fallback interface keeps its own
             # session buffers, so its history accumulates independently and
             # the primary's buffers stay warm for the next-turn retry.
+            if (
+                error_info is not None
+                and error_info.category == ErrorCategory.CONTEXT_OVERFLOW
+            ):
+                raise LLMContextOverflowError(error_detail)
             if self._current_session_call is not None:
                 task_id, call_type, fb_user_prompt = self._current_session_call
                 stored_system = self._session_system_prompts.get(
@@ -1302,9 +1366,7 @@ def _generate_response_with_session_sync(
             if not effective_system_prompt:
                 raise ValueError(f"No system prompt for task {task_id}:{call_type}")
 
-            if session_key not in self._gemini_session_messages:
-                self._gemini_session_messages[session_key] = []
-            history = self._gemini_session_messages[session_key]
+            history = self._session_histories.setdefault(session_key, [])
 
             # Build contents = history + new user turn.
             contents: List[Dict[str, Any]] = []
@@ -1317,6 +1379,7 @@ def _generate_response_with_session_sync(
                 f"sending {len(contents)} total contents"
             )
 
+            self._check_context_fits(effective_system_prompt, messages=contents)
             response = self._generate_gemini(
                 effective_system_prompt,
                 user_prompt,
@@ -1361,9 +1424,7 @@ def _generate_response_with_session_sync(
             )
 
             if is_openrouter_claude:
-                if session_key not in self._openrouter_anthropic_session_messages:
-                    self._openrouter_anthropic_session_messages[session_key] = []
-                history = self._openrouter_anthropic_session_messages[session_key]
+                history = self._session_histories.setdefault(session_key, [])
 
                 # Build OpenAI-shaped messages: [system, user1, assistant1,
                 # ..., new_user]. OpenRouter applies extra_body.cache_control
@@ -1380,6 +1441,7 @@ def _generate_response_with_session_sync(
                     f"{len(history)} history msgs, sending {len(or_messages)} total"
                 )
 
+                self._check_context_fits(None, messages=or_messages)
                 response = self._generate_openai(
                     effective_system_prompt,
                     user_prompt,
@@ -1407,10 +1469,7 @@ def _generate_response_with_session_sync(
                 # resend [system, u1, a1, ..., new_user] every turn. Correctness
                 # aside, the stable growing prefix is exactly what prompt_cache_key
                 # rewards, so most of the resend is served from cache once warm.
-                if session_key not in self._openai_compat_session_messages:
-                    self._openai_compat_session_messages[session_key] = []
-                history = self._openai_compat_session_messages[session_key]
-                self._trim_openai_compat_history(history)
+                history = self._session_histories.setdefault(session_key, [])
 
                 oa_messages: List[Dict[str, Any]] = [
                     {"role": "system", "content": effective_system_prompt}
@@ -1424,6 +1483,7 @@ def _generate_response_with_session_sync(
                     f"{len(history)} history msgs, sending {len(oa_messages)} total"
                 )
 
+                self._check_context_fits(None, messages=oa_messages)
                 response = self._generate_openai(
                     effective_system_prompt,
                     user_prompt,
@@ -1450,10 +1510,8 @@ def _generate_response_with_session_sync(
                 raise ValueError(f"No system prompt for task {task_id}:{call_type}")
 
             # Get or initialize multi-turn message history
-            if session_key not in self._anthropic_session_messages:
-                self._anthropic_session_messages[session_key] = []
 
-            history = self._anthropic_session_messages[session_key]
+            history = self._session_histories.setdefault(session_key, [])
 
             # Build messages: history (with cache_control on last assistant) + new user msg
             messages: List[dict] = []
@@ -1505,6 +1563,7 @@ def _generate_response_with_session_sync(
             )
 
             # Call Anthropic with the full multi-turn messages
+            self._check_context_fits(effective_system_prompt, messages=messages)
             response = self._generate_anthropic(
                 effective_system_prompt,
                 user_prompt,
@@ -1542,9 +1601,7 @@ def _generate_response_with_session_sync(
 
             # Get or initialize multi-turn message history (Bedrock Converse
             # content-block format: {"role": ..., "content": [{"text": ...}]}).
-            if session_key not in self._bedrock_session_messages:
-                self._bedrock_session_messages[session_key] = []
-            history = self._bedrock_session_messages[session_key]
+            history = self._session_histories.setdefault(session_key, [])
 
             # Build messages: history (strip any prior cachePoint blocks, we
             # re-place exactly one) + new user message.
@@ -1579,6 +1636,7 @@ def _generate_response_with_session_sync(
                 f"sending {len(messages)} msgs to Converse"
             )
 
+            self._check_context_fits(effective_system_prompt, messages=messages)
             response = self._generate_bedrock(
                 effective_system_prompt,
                 user_prompt,
@@ -1818,9 +1876,13 @@ def generate_response_with_session(
             prompt_name: Identity of the named prompt, for capture/profiling.
         """
         self._begin_call(prompt_name=prompt_name, call_type=call_type, task_id=task_id)
-        return self._generate_response_with_session_sync(
-            task_id, call_type, user_prompt, system_prompt_for_new_session, log_response
-        )
+        token = _active_session_key.set(f"{task_id}:{call_type}")
+        try:
+            return self._generate_response_with_session_sync(
+                task_id, call_type, user_prompt, system_prompt_for_new_session, log_response
+            )
+        finally:
+            _active_session_key.reset(token)
 
     @profile("llm_generate_response_with_session_async", OperationCategory.LLM)
     async def generate_response_with_session_async(
@@ -1845,14 +1907,18 @@ async def generate_response_with_session_async(
         # Stamp here (caller's context) so asyncio.to_thread copies it into the
         # worker thread where capture runs.
         self._begin_call(prompt_name=prompt_name, call_type=call_type, task_id=task_id)
-        return await asyncio.to_thread(
-            self._generate_response_with_session_sync,
-            task_id,
-            call_type,
-            user_prompt,
-            system_prompt_for_new_session,
-            log_response,
-        )
+        token = _active_session_key.set(f"{task_id}:{call_type}")
+        try:
+            return await asyncio.to_thread(
+                self._generate_response_with_session_sync,
+                task_id,
+                call_type,
+                user_prompt,
+                system_prompt_for_new_session,
+                log_response,
+            )
+        finally:
+            _active_session_key.reset(token)
 
     def _generate_byteplus_with_session(
         self, task_id: str, call_type: str, user_prompt: str
diff --git a/agent_core/core/models/provider_config.py b/agent_core/core/models/provider_config.py
index 2119e12b..51fdaba8 100644
--- a/agent_core/core/models/provider_config.py
+++ b/agent_core/core/models/provider_config.py
@@ -119,7 +119,7 @@ class ProviderProfile:
     uses_max_completion_tokens: bool = False
     # Whether the chat_completions session path accumulates a growing
     # [user, assistant, ...] history for this provider (the
-    # _openai_compat_session_messages buffer). False preserves the
+    # session history). False preserves the
     # historical behavior for minimax/moonshot, whose session turns fall
     # through to stateless generation. Only meaningful on the
     # chat_completions wire.
diff --git a/agent_core/core/models/registry.py b/agent_core/core/models/registry.py
index 1c7a53b4..b19e04a2 100644
--- a/agent_core/core/models/registry.py
+++ b/agent_core/core/models/registry.py
@@ -225,7 +225,7 @@ def default_models_registry() -> Dict[str, Dict[Any, Optional[str]]]:
 
 def session_cc_providers() -> frozenset:
     """chat_completions providers whose session path accumulates history
-    (the _openai_compat_session_messages / openrouter-anthropic buffers).
+    (the accumulated session history).
 
     Replaces the hand-maintained tuple in interface.py's session dispatcher
     and create_session_cache. minimax/moonshot stay excluded
diff --git a/app/config.py b/app/config.py
index 9474db09..3331ae9b 100644
--- a/app/config.py
+++ b/app/config.py
@@ -75,11 +75,17 @@ def invalidate_settings_cache() -> None:
     _settings_cache = None
 
 
-# Event-stream summarization thresholds. Defined here rather than in
-# event_stream.py so settings.json defaults and the runtime fallback cannot
-# drift apart.
-DEFAULT_SUMMARIZE_AT_TOKENS = 100000
-DEFAULT_TAIL_KEEP_AFTER_SUMMARIZE_TOKENS = 10000
+# Context budget settings. The decision to summarize the event stream is made
+# on the WHOLE request (see ActionRouter / LLMInterface.fits_context):
+#   model.context_window        the model's window, in tokens
+#   context.reserve_tokens      headroom the summary request itself needs
+#   context.keep_recent_tokens  recent events kept verbatim after a fold
+# A key that is absent takes the shipped value from _get_default_settings();
+# a key that is present but invalid is a ConfigurationError.
+
+
+class ConfigurationError(ValueError):
+    """A required setting is missing or invalid in settings.json."""
 
 
 def _get_default_settings() -> Dict[str, Any]:
@@ -96,8 +102,8 @@ def _get_default_settings() -> Dict[str, Any]:
         "proactive": {"enabled": True},
         "memory": {"enabled": True},
         "context": {
-            "summarize_at_tokens": DEFAULT_SUMMARIZE_AT_TOKENS,
-            "tail_keep_after_summarize_tokens": DEFAULT_TAIL_KEEP_AFTER_SUMMARIZE_TOKENS,
+            "reserve_tokens": 16384,
+            "keep_recent_tokens": 20000,
         },
         "model": {
             "llm_provider": "anthropic",
@@ -106,6 +112,7 @@ def _get_default_settings() -> Dict[str, Any]:
             "video_gen_provider": "gemini",
             "llm_model": None,
             "vlm_model": None,
+            "context_window": 128000,
             "image_gen_model": None,
             "video_gen_model": None,
             "slow_mode": False,
@@ -208,31 +215,41 @@ def get_app_version() -> str:
     return v or "0.0.0"
 
 
-def get_context_limits() -> Tuple[int, int]:
-    """Get event-stream summarization thresholds from settings.json.
+def _setting(section: str, key: str) -> Any:
+    """``settings.json[section][key]``, or the shipped default when absent."""
+    values = get_settings().get(section)
+    value = values.get(key) if isinstance(values, dict) else None
+    if value is None:
+        return _get_default_settings()[section][key]
+    return value
 
-    Returns ``(summarize_at_tokens, tail_keep_after_summarize_tokens)``.
-    Non-positive or non-integer values fall back to the defaults rather than
-    raising — a bad hand-edit must not take the agent down. EventStream still
-    validates the two against each other; this only guarantees sane types.
-    """
-    context = get_settings().get("context") or {}
-    if not isinstance(context, dict):
-        context = {}
 
-    def _positive_int(key: str, default: int) -> int:
-        value = context.get(key, default)
-        if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
-            return default
-        return value
+def get_context_window() -> int:
+    """The configured model's context window in tokens (model.context_window)."""
+    value = _setting("model", "context_window")
+    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
+        raise ConfigurationError(
+            "settings.json model.context_window must be a positive integer: the "
+            "context window of the configured model, in tokens."
+        )
+    return value
 
-    return (
-        _positive_int("summarize_at_tokens", DEFAULT_SUMMARIZE_AT_TOKENS),
-        _positive_int(
-            "tail_keep_after_summarize_tokens",
-            DEFAULT_TAIL_KEEP_AFTER_SUMMARIZE_TOKENS,
-        ),
-    )
+
+def _get_positive_int(section: str, key: str) -> int:
+    value = _setting(section, key)
+    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
+        raise ConfigurationError(f"settings.json {section}.{key} must be a positive integer.")
+    return value
+
+
+def get_reserve_tokens() -> int:
+    """Headroom kept free for the summary request (context.reserve_tokens)."""
+    return _get_positive_int("context", "reserve_tokens")
+
+
+def get_keep_recent_tokens() -> int:
+    """Recent event-stream tokens kept verbatim after a fold (context.keep_recent_tokens)."""
+    return _get_positive_int("context", "keep_recent_tokens")
 
 
 def get_llm_provider() -> str:
diff --git a/app/config/settings.json b/app/config/settings.json
index 72a93d2a..e385779d 100644
--- a/app/config/settings.json
+++ b/app/config/settings.json
@@ -16,8 +16,8 @@
     "processing_threshold": 23
   },
   "context": {
-    "summarize_at_tokens": 100000,
-    "tail_keep_after_summarize_tokens": 10000
+    "reserve_tokens": 16384,
+    "keep_recent_tokens": 20000
   },
   "model": {
     "slow_mode": true,
@@ -25,7 +25,8 @@
     "llm_provider": "anthropic",
     "vlm_provider": "anthropic",
     "llm_model": null,
-    "vlm_model": null
+    "vlm_model": null,
+    "context_window": 128000
   },
   "api_keys": {
     "openai": "",
@@ -98,4 +99,4 @@
   "agent_app": {
     "marketplace_ref": ""
   }
-}
\ No newline at end of file
+}
diff --git a/app/subagent/runner.py b/app/subagent/runner.py
index f027f25f..fc781edb 100644
--- a/app/subagent/runner.py
+++ b/app/subagent/runner.py
@@ -454,7 +454,7 @@ def _reset_session(self, sub: SubAgent, stream) -> None:
         ``_build_user_prompt`` will resend the full first-turn prompt and
         the LLM interface will lazily recreate the session.
         """
-        self.llm_interface.end_session_cache(sub.id, _SUBAGENT_CALL_TYPE)
+        self.llm_interface.reset_session_history(sub.id, _SUBAGENT_CALL_TYPE)
         stream.reset_session_sync(_SUBAGENT_CALL_TYPE)
 
     # ------------------------------------------------------------------
@@ -584,6 +584,9 @@ def _build_user_prompt(self, sub: SubAgent, stream) -> Tuple[str, bool]:
         model after the cached history vanishes.
         """
         if not stream.has_session_sync(_SUBAGENT_CALL_TYPE):
+            # A true first turn, or the stream folded and cleared the sync
+            # point; either way any accumulated turns are stale.
+            self._reset_session(sub, stream)
             prompt = self.context_engine.make_first_turn_user_prompt(sub)
             return self._with_turn_budget(sub, prompt), True
 
diff --git a/app/ui_layer/browser/frontend/src/constants/errorCategories.ts b/app/ui_layer/browser/frontend/src/constants/errorCategories.ts
index 785ccdbf..18373aee 100644
--- a/app/ui_layer/browser/frontend/src/constants/errorCategories.ts
+++ b/app/ui_layer/browser/frontend/src/constants/errorCategories.ts
@@ -29,6 +29,7 @@ export const ERROR_CATEGORY_STYLE: Record<string, ErrorCategoryStyle> = {
   quota: { icon: CreditCard, colorVar: '--color-warning', labelKey: 'common:errorCategory.quota' },
   model: { icon: AlertCircle, colorVar: '--color-error', labelKey: 'common:errorCategory.model' },
   bad_request: { icon: AlertCircle, colorVar: '--color-error', labelKey: 'common:errorCategory.badRequest' },
+  context_overflow: { icon: AlertCircle, colorVar: '--color-error', labelKey: 'common:errorCategory.contextOverflow' },
   blocked: { icon: ShieldAlert, colorVar: '--color-error', labelKey: 'common:errorCategory.blocked' },
   server: { icon: ServerCrash, colorVar: '--color-error', labelKey: 'common:errorCategory.server' },
   connection: { icon: WifiOff, colorVar: '--color-error', labelKey: 'common:errorCategory.connection' },
diff --git a/app/ui_layer/browser/frontend/src/locales/en/common.json b/app/ui_layer/browser/frontend/src/locales/en/common.json
index 6c254410..8742bc03 100644
--- a/app/ui_layer/browser/frontend/src/locales/en/common.json
+++ b/app/ui_layer/browser/frontend/src/locales/en/common.json
@@ -76,6 +76,7 @@
     "quota": "Quota",
     "model": "Model",
     "badRequest": "Request",
+    "contextOverflow": "Context window exceeded",
     "blocked": "Blocked",
     "server": "Service unavailable",
     "connection": "Connection",
diff --git a/app/ui_layer/browser/frontend/src/locales/es/common.json b/app/ui_layer/browser/frontend/src/locales/es/common.json
index 87249881..5b7bf86d 100644
--- a/app/ui_layer/browser/frontend/src/locales/es/common.json
+++ b/app/ui_layer/browser/frontend/src/locales/es/common.json
@@ -76,6 +76,7 @@
     "quota": "Cuota",
     "model": "Modelo",
     "badRequest": "Solicitud",
+    "contextOverflow": "Ventana de contexto excedida",
     "blocked": "Bloqueado",
     "server": "Servicio no disponible",
     "connection": "Conexión",
diff --git a/app/ui_layer/browser/frontend/src/locales/id/common.json b/app/ui_layer/browser/frontend/src/locales/id/common.json
index 60993a98..33254a53 100644
--- a/app/ui_layer/browser/frontend/src/locales/id/common.json
+++ b/app/ui_layer/browser/frontend/src/locales/id/common.json
@@ -76,6 +76,7 @@
     "quota": "Kuota",
     "model": "Model",
     "badRequest": "Permintaan",
+    "contextOverflow": "Jendela konteks terlampaui",
     "blocked": "Diblokir",
     "server": "Layanan tidak tersedia",
     "connection": "Koneksi",
diff --git a/app/ui_layer/browser/frontend/src/locales/ja/common.json b/app/ui_layer/browser/frontend/src/locales/ja/common.json
index 85ebcaac..e50f07ab 100644
--- a/app/ui_layer/browser/frontend/src/locales/ja/common.json
+++ b/app/ui_layer/browser/frontend/src/locales/ja/common.json
@@ -76,6 +76,7 @@
     "quota": "クォータ",
     "model": "モデル",
     "badRequest": "リクエスト",
+    "contextOverflow": "コンテキストウィンドウ超過",
     "blocked": "ブロック",
     "server": "サービス利用不可",
     "connection": "接続",
diff --git a/app/ui_layer/browser/frontend/src/locales/ko/common.json b/app/ui_layer/browser/frontend/src/locales/ko/common.json
index 7d8cf35f..c0d0b9c8 100644
--- a/app/ui_layer/browser/frontend/src/locales/ko/common.json
+++ b/app/ui_layer/browser/frontend/src/locales/ko/common.json
@@ -76,6 +76,7 @@
     "quota": "할당량",
     "model": "모델",
     "badRequest": "요청",
+    "contextOverflow": "컨텍스트 창 초과",
     "blocked": "차단됨",
     "server": "서비스 사용 불가",
     "connection": "연결",
diff --git a/app/ui_layer/browser/frontend/src/locales/zh-CN/common.json b/app/ui_layer/browser/frontend/src/locales/zh-CN/common.json
index 33861ee3..7e0a75af 100644
--- a/app/ui_layer/browser/frontend/src/locales/zh-CN/common.json
+++ b/app/ui_layer/browser/frontend/src/locales/zh-CN/common.json
@@ -76,6 +76,7 @@
     "quota": "配额",
     "model": "模型",
     "badRequest": "请求",
+    "contextOverflow": "超出上下文窗口",
     "blocked": "已拦截",
     "server": "服务不可用",
     "connection": "连接",
diff --git a/app/ui_layer/browser/frontend/src/locales/zh-TW/common.json b/app/ui_layer/browser/frontend/src/locales/zh-TW/common.json
index b49e7551..cddb9ff2 100644
--- a/app/ui_layer/browser/frontend/src/locales/zh-TW/common.json
+++ b/app/ui_layer/browser/frontend/src/locales/zh-TW/common.json
@@ -76,6 +76,7 @@
     "quota": "配額",
     "model": "模型",
     "badRequest": "請求",
+    "contextOverflow": "超出上下文視窗",
     "blocked": "已封鎖",
     "server": "服務無法使用",
     "connection": "連線",
diff --git a/tests/conftest.py b/tests/conftest.py
index efa31aa8..7b28746c 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -25,28 +25,51 @@ def _safe_load_windows_store_certs(self, storename, purpose):
 os.chdir(PROJECT_ROOT)
 
 
+@pytest.fixture(autouse=True)
+def configured_context_window(monkeypatch):
+    """Give every test a configured model.context_window.
+
+    It is required configuration -- the agent refuses to start without it --
+    and the tracked settings.json ships it as null on purpose. A test that
+    exercises the missing-window error patches get_settings itself, which
+    takes precedence over this fixture for that test.
+    """
+    from app import config as app_config
+
+    real_get_settings = app_config.get_settings
+
+    def _with_window(reload: bool = False):
+        settings = dict(real_get_settings(reload))
+        model = dict(settings.get("model") or {})
+        if not model.get("context_window"):
+            model["context_window"] = 128000
+        settings["model"] = model
+        context = dict(settings.get("context") or {})
+        context.setdefault("reserve_tokens", 16384)
+        context.setdefault("keep_recent_tokens", 20000)
+        settings["context"] = context
+        return settings
+
+    monkeypatch.setattr(app_config, "get_settings", _with_window)
+
+
 @pytest.fixture
 def event_stream_limits(monkeypatch):
-    """Pin EventStream's summarization thresholds for a test.
+    """Pin how much recent history an EventStream keeps after a fold.
 
-    EventStream reads them from settings.json, so without this a local config
-    edit would silently change what an event-stream test exercises. The import
-    is inside the fixture so collecting unrelated tests does not pull in
+    The stream never folds on its own; tests call summarize_by_LLM() when
+    they want one. This pins keep_recent_tokens so a local settings.json
+    edit cannot change what an event-stream test exercises. The import is
+    inside the fixture so collecting unrelated tests does not pull in
     event_stream (and sklearn with it).
-
-    Call with no arguments for thresholds high enough that nothing folds — the
-    right choice for tests that are not about summarization at all.
     """
     from agent_core.core.impl.event_stream import event_stream as event_stream_module
 
-    def _pin(
-        summarize_at_tokens: int = 100000,
-        tail_keep_after_summarize_tokens: int = 10000,
-    ) -> None:
+    def _pin(keep_recent_tokens: int = 10000) -> None:
         monkeypatch.setattr(
             event_stream_module,
-            "_configured_context_limits",
-            lambda: (summarize_at_tokens, tail_keep_after_summarize_tokens),
+            "_configured_keep_recent_tokens",
+            lambda: keep_recent_tokens,
         )
 
     return _pin
diff --git a/tests/test_context_budget.py b/tests/test_context_budget.py
new file mode 100644
index 00000000..8901426b
--- /dev/null
+++ b/tests/test_context_budget.py
@@ -0,0 +1,320 @@
+# -*- coding: utf-8 -*-
+"""Context budget: one decision, on the request, made before it is sent.
+
+The history collapse came from two budgets that could not see each other: a
+threshold on the event stream alone and an independent cap on the session
+history. There is one decision now, and it is made on the whole request the
+way pi and OpenClaw make it -- the provider's own input count for the previous
+request plus what is new -- against the window less the headroom the summary
+request needs. These tests pin:
+
+* the history cap is gone; a fold resets the history WITHOUT ending the
+  session, so the router keeps the session path and its cached prefix;
+* the cache markers the providers need are still on the wire;
+* the provider's input count is recorded per session and drives fits_context;
+* the event stream never folds on its own -- only on request;
+* an overflow is surfaced as one typed error, never counted as a provider
+  failure, never retried on a fallback provider;
+* the settings are configuration with shipped defaults, invalid values error.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from agent_core.core.errors import ErrorCategory
+from agent_core.core.impl.llm.interface import LLMContextOverflowError, LLMInterface
+from agent_core.utils.token import count_tokens
+from app import config as app_config
+from app.models.factory import ModelFactory
+
+WINDOW = 128000
+RESERVE = 16384
+MAX_TOKENS = 8000
+FOLD_POINT = WINDOW - RESERVE - MAX_TOKENS  # 103,616 input tokens
+
+
+def _ctx(provider, model, anthropic_client=None):
+    return {
+        "provider": provider,
+        "model": model,
+        "client": object(),
+        "gemini_client": None,
+        "remote_url": None,
+        "byteplus": None,
+        "anthropic_client": anthropic_client,
+        "bedrock_client": None,
+        "initialized": True,
+        "auth_mode": "api_key",
+    }
+
+
+def _make(provider="grok", model="grok-3", anthropic_client=None):
+    with patch.object(ModelFactory, "create", return_value=_ctx(provider, model, anthropic_client)):
+        return LLMInterface(provider=provider, model=model, api_key="k", base_url="", max_tokens=MAX_TOKENS)
+
+
+def _settings(context_window=WINDOW, reserve_tokens=RESERVE, keep_recent_tokens=20000):
+    return {
+        "model": {"context_window": context_window},
+        "context": {"reserve_tokens": reserve_tokens, "keep_recent_tokens": keep_recent_tokens},
+    }
+
+
+class _FakeAnthropic:
+    def __init__(self):
+        self.calls = []
+        self.messages = self
+
+    def create(self, **kwargs):
+        self.calls.append(kwargs)
+        return SimpleNamespace(
+            content=[SimpleNamespace(type="text", text='{"ok": 1}')],
+            usage=SimpleNamespace(
+                input_tokens=10, output_tokens=2,
+                cache_creation_input_tokens=0, cache_read_input_tokens=0,
+            ),
+        )
+
+
+class _CountingLLM:
+    consecutive_failures = 0
+    _max_consecutive_failures = 5
+
+    def __init__(self):
+        self.calls = 0
+
+    def generate_response(self, user_prompt=None, prompt_name=None, **kw):
+        self.calls += 1
+        return "SUMMARY"
+
+
+# ------------------------------------------------------------- one budget
+
+
+def test_the_independent_history_cap_is_gone():
+    from pathlib import Path
+
+    import agent_core.core.impl.llm.interface as module
+
+    source = Path(module.__file__).read_text(encoding="utf-8")
+    assert "_trim_openai_compat_history" not in source
+    assert "max_history_chars" not in source
+    assert "_session_messages" not in source  # one provider-agnostic container
+
+
+def test_reset_session_history_keeps_the_session_registered():
+    iface = _make()
+    key = "task:action_selection"
+    iface.create_session_cache("task", "action_selection", "SYSTEM")
+    iface._session_histories[key] = [{"role": "user", "content": "stale"}]
+    iface._last_input_tokens[key] = 90000
+
+    iface.reset_session_history("task", "action_selection")
+
+    assert key not in iface._session_histories
+    assert iface.last_input_tokens("task", "action_selection") is None
+    assert iface.has_session_cache("task", "action_selection") is True
+
+
+def test_router_decides_on_the_request_and_restarts_without_ending():
+    from pathlib import Path
+
+    import agent_core.core.impl.action.router as router
+
+    src = Path(router.__file__).read_text(encoding="utf-8")
+    delta_branch = src.index("if has_synced_before:")
+    delta_send = src.index("Sending delta events", delta_branch)
+    assert "fits_context" in src[delta_branch:delta_send]
+    first_call = src.index("if not has_synced_before:")
+    send = src.index("generate_response_with_session_async", first_call)
+    assert "reset_session_history" in src[first_call:send]
+    assert "end_session_cache" not in src[src.index("No delta events"):send]
+    assert "except LLMContextOverflowError" in src
+
+
+def test_subagent_first_turn_resets_history():
+    from pathlib import Path
+
+    import app.subagent.runner as runner
+
+    src = Path(runner.__file__).read_text(encoding="utf-8")
+    branch = src.index("if not stream.has_session_sync(_SUBAGENT_CALL_TYPE):")
+    build = src.index("make_first_turn_user_prompt", branch)
+    assert "_reset_session" in src[branch:build]
+
+
+# ------------------------------------------------------------ KV caching
+
+
+def test_anthropic_session_marks_system_and_last_assistant():
+    fake = _FakeAnthropic()
+    iface = _make("anthropic", "claude-x", anthropic_client=fake)
+    iface._anthropic_client = fake
+    iface.create_session_cache("task", "action_selection", "S" * 5000)
+    for turn in ("turn 1", "turn 2"):
+        iface._generate_response_with_session_sync("task", "action_selection", turn, log_response=False)
+
+    second = fake.calls[1]
+    assert second["system"][0].get("cache_control")
+    assert [m["role"] for m in second["messages"]] == ["user", "assistant", "user"]
+    last_assistant = second["messages"][1]
+    content = last_assistant["content"]
+    assert ("cache_control" in last_assistant) or (
+        isinstance(content, list) and any("cache_control" in block for block in content)
+    )
+
+
+def test_openai_compat_session_sends_a_growing_identical_prefix():
+    sent = []
+
+    def fake_generate_openai(self, system_prompt, user_prompt, call_type=None, messages_override=None, **kw):
+        sent.append([dict(m) for m in messages_override])
+        return {"content": '{"ok": 1}', "tokens_used": 1}
+
+    iface = _make()
+    iface.create_session_cache("task", "action_selection", "SYSTEM")
+    with patch.object(LLMInterface, "_generate_openai", fake_generate_openai):
+        for turn in range(3):
+            iface._generate_response_with_session_sync("task", "action_selection", f"turn {turn}", log_response=False)
+
+    assert [m["role"] for m in sent[2]] == ["system", "user", "assistant", "user", "assistant", "user"]
+    assert sent[1][:3] == sent[2][:3]
+
+
+# --------------------------------------- the decision is made on the request
+
+
+def test_the_providers_input_count_is_recorded_per_session():
+    """Every transport reports it through _report_usage_async; it must land on
+    the session that made the call and nowhere else."""
+    iface = _make()
+    iface.create_session_cache("task", "action_selection", "SYSTEM")
+
+    def fake_generate_openai(self, system_prompt, user_prompt, call_type=None, messages_override=None, **kw):
+        self._report_usage_async("llm_openai", "grok", "grok-3", 4321, 10, 0)
+        return {"content": '{"ok": 1}', "tokens_used": 4331}
+
+    with patch.object(LLMInterface, "_generate_openai", fake_generate_openai):
+        iface.generate_response_with_session("task", "action_selection", "turn 1", log_response=False)
+    assert iface.last_input_tokens("task", "action_selection") == 4321
+
+    iface._report_usage_async("llm_openai", "grok", "grok-3", 999, 1, 0)  # outside any session call
+    assert iface.last_input_tokens("task", "action_selection") == 4321
+
+
+def test_fits_context_uses_the_providers_count_plus_only_what_is_new():
+    iface = _make()
+    pending = "new events " * 300
+    pending_tokens = count_tokens(pending)
+    with patch.object(app_config, "get_settings", return_value=_settings()):
+        iface._last_input_tokens["task:action_selection"] = FOLD_POINT - pending_tokens
+        assert iface.fits_context("task", "action_selection", "SYSTEM", pending) is True
+        iface._last_input_tokens["task:action_selection"] = FOLD_POINT - pending_tokens + 1
+        assert iface.fits_context("task", "action_selection", "SYSTEM", pending) is False
+
+
+def test_fits_context_counts_the_whole_prompt_on_a_sessions_first_request():
+    iface = _make()
+    system, prompt = "system " * 100, "prompt " * 100
+    with patch.object(app_config, "get_settings", return_value=_settings()):
+        assert iface.fits_context("task", "action_selection", system, prompt) is True
+        too_big = "word " * (FOLD_POINT + 5000)
+        assert iface.fits_context("task", "action_selection", system, too_big) is False
+
+
+def test_the_shipped_fold_point_for_a_128k_window():
+    """window 128,000 - reserve 16,384 - output 8,000: fold when input passes 103,616."""
+    iface = _make()
+    with patch.object(app_config, "get_settings", return_value=_settings()):
+        iface._last_input_tokens["task:action_selection"] = 103_616
+        assert iface.fits_context("task", "action_selection", "s", "") is True
+        iface._last_input_tokens["task:action_selection"] = 103_617
+        assert iface.fits_context("task", "action_selection", "s", "") is False
+
+
+# ----------------------------------------- the stream folds only on request
+
+
+def test_the_event_stream_never_folds_on_its_own(event_stream_limits):
+    from agent_core.core.impl.event_stream.event_stream import EventStream
+
+    event_stream_limits(100)
+    llm = _CountingLLM()
+    es = EventStream(llm=llm, temp_dir=None)
+    for i in range(400):
+        es.log("action_end", f"action {i} produced output " + "x " * 50)
+    assert llm.calls == 0 and es.head_summary is None
+
+    es.summarize_by_LLM()
+    assert llm.calls == 1 and es.head_summary is not None
+
+
+# ------------------------------------------------------------- overflow
+
+
+def test_a_provider_size_rejection_is_a_typed_overflow_not_a_failure():
+    iface = _make()
+    refusal = {
+        "content": "",
+        "error": "BadRequestError: context_length_exceeded",
+        "error_info_obj": SimpleNamespace(
+            category=ErrorCategory.CONTEXT_OVERFLOW, message="request too large"
+        ),
+    }
+    fallback_calls = []
+    with patch.dict("agent_core.core.impl.llm.transports.TRANSPORTS",
+                    {"chat_completions": lambda *a, **k: refusal}), \
+         patch.object(LLMInterface, "_try_fallback", lambda self, *a, **k: fallback_calls.append(1)), \
+         patch.object(app_config, "get_settings", return_value=_settings()):
+        with pytest.raises(LLMContextOverflowError):
+            iface._generate_response_sync(system_prompt="s", user_prompt="u", log_response=False)
+    assert iface._consecutive_failures == 0
+    assert fallback_calls == []
+
+
+def test_the_structured_openai_code_maps_to_context_overflow():
+    from pathlib import Path
+
+    import agent_core.core.impl.llm.errors as errors
+
+    src = Path(errors.__file__).read_text(encoding="utf-8")
+    i = src.index('code == "context_length_exceeded"')
+    assert "ErrorCategory.CONTEXT_OVERFLOW" in src[i: i + 120]
+
+
+def test_preflight_refuses_a_request_that_does_not_fit():
+    iface = _make()
+    with patch.object(app_config, "get_settings", return_value=_settings()):
+        with pytest.raises(LLMContextOverflowError):
+            iface._check_context_fits("system", "word " * 200_000)
+    assert iface._consecutive_failures == 0
+
+
+# ------------------------------------------------------------ configuration
+
+
+def test_shipped_defaults_apply_when_keys_are_absent():
+    with patch.object(app_config, "get_settings", return_value={"model": {}, "context": {}}):
+        assert app_config.get_context_window() == 128000
+        assert app_config.get_reserve_tokens() == 16384
+        assert app_config.get_keep_recent_tokens() == 20000
+
+
+@pytest.mark.parametrize("value", [0, -1, True, "16384", 12.5])
+@pytest.mark.parametrize("key", ["reserve_tokens", "keep_recent_tokens"])
+def test_an_invalid_context_setting_is_a_configuration_error(key, value):
+    settings = _settings()
+    settings["context"][key] = value
+    with patch.object(app_config, "get_settings", return_value=settings):
+        with pytest.raises(app_config.ConfigurationError):
+            getattr(app_config, f"get_{key}")()
+
+
+def test_no_fallback_constants_and_no_fractions_remain():
+    from pathlib import Path
+
+    source = Path(app_config.__file__).read_text(encoding="utf-8")
+    for token in ("LEGACY", "v1.4.1", "stream_fraction_of_window", "tail_keep_fraction", "get_context_limits"):
+        assert token not in source, token
diff --git a/tests/test_event_stream_datetime.py b/tests/test_event_stream_datetime.py
index 05cd9021..4fe81eb8 100644
--- a/tests/test_event_stream_datetime.py
+++ b/tests/test_event_stream_datetime.py
@@ -64,10 +64,11 @@ def test_datetime_refreshes_after_interval(event_stream_limits):
 
 
 def test_datetime_restamped_after_summarization(event_stream_limits):
-    event_stream_limits(2100, 100)
+    event_stream_limits(100)
     es = EventStream(llm=_FakeLLM())
     for i in range(400):
         es.log("action_end", f"action {i} produced some output text to add tokens")
-    assert es.head_summary is not None  # summarization happened
+    es.summarize_by_LLM()
+    assert es.head_summary is not None
     # A current datetime marker is always present (re-stamped post-summary).
     assert any(r.event.kind == "datetime" for r in es.tail_events)
diff --git a/tests/test_event_stream_oversize.py b/tests/test_event_stream_oversize.py
index 536302fc..d81ad666 100644
--- a/tests/test_event_stream_oversize.py
+++ b/tests/test_event_stream_oversize.py
@@ -1,15 +1,17 @@
 # -*- coding: utf-8 -*-
 """
-A single oversized event must not cost two blocking summarization passes.
+Oversized retrieval results must not waste summarization passes.
 
 Observed 2026-08-26 in session lui_11e12617: one `grep_files` result of 171,818
 chars (~77k tokens) entered the tail verbatim — grep_files/read_file are exempt
 from log-time externalization because they ARE the retrieval path for
-externalized content. MIN_KEEP_RECENT_EVENTS then pinned it, so the pass it
-triggered went 92,735 -> 77,054 tokens (still over threshold, LLM call wasted)
-and the next event fired a second pass, 79,117 -> 2,960, that folded it anyway.
-Five such events in one 24-minute run; 33 passes totalling 936k uncached input
-tokens and 536s of blocking wall-clock.
+externalized content. MIN_KEEP_RECENT_EVENTS pinned it, and folds that could
+not get under budget were re-triggered on every append.
+
+The stream no longer folds on its own; the router asks for one fold when the
+next request would not fit. So a fold must be worth asking for: it collapses
+oversized pinned events in place first, then summarizes, and a region too
+small to be worth an LLM call is pruned instead.
 
 See _shrink_pinned_oversize / MIN_FOLD_TOKENS in
 agent_core/core/impl/event_stream/event_stream.py.
@@ -34,32 +36,36 @@ def generate_response(self, user_prompt=None, prompt_name=None, **kw):
 
 
 def _stream(tmp_path, llm, event_stream_limits):
-    event_stream_limits(30000, 10000)
+    event_stream_limits(10000)
     return EventStream(llm=llm, temp_dir=tmp_path / "events")
 
 
-def test_oversized_pinned_event_is_collapsed_without_an_llm_call(
-    tmp_path, event_stream_limits
-):
+def test_appending_never_folds_on_its_own(tmp_path, event_stream_limits):
+    """The decision lives with the router; log() only appends."""
     llm = _CountingLLM()
     es = _stream(tmp_path, llm, event_stream_limits)
+    for i in range(400):
+        es.log("action_end", f"action {i} completed " + "x " * 200)
+    assert llm.calls == 0
+    assert es.head_summary is None
 
+
+def test_oversized_pinned_event_is_collapsed_in_place_by_a_fold(tmp_path, event_stream_limits):
+    llm = _CountingLLM()
+    es = _stream(tmp_path, llm, event_stream_limits)
     for i in range(60):
         es.log("action_end", f"action {i} completed " + "x " * 200)
-    assert es._total_tokens < es.summarize_at_tokens
-    baseline_calls = llm.calls
 
     # The grep_files result: exempt from log-time externalization, ~70k tokens,
-    # and the newest event in the tail — exactly what the pin used to hold.
+    # and the newest event in the tail — exactly what the pin holds.
     giant = "matched line " + ("y " * 140_000)
     es.log("action_end", giant, action_name="grep_files")
+    before = es._total_tokens
 
-    # Collapsing it in place is enough on its own: zero passes, where the old
-    # code paid for two.
-    assert llm.calls == baseline_calls
-    assert es._total_tokens < es.summarize_at_tokens
+    es.summarize_by_LLM()
 
-    # The record survives so the UI can still pair action_start ↔ action_end...
+    assert es._total_tokens < before
+    # The record survives so the UI can still pair action_start <-> action_end...
     grep_rec = next(r for r in es.tail_events if r.event.action_name == "grep_files")
     # ...but its message is now a pointer, and the content is on disk.
     assert len(grep_rec.event.message) <= MAX_EVENT_INLINE_CHARS
@@ -68,15 +74,9 @@ def test_oversized_pinned_event_is_collapsed_without_an_llm_call(
     assert written and written[0].read_text(encoding="utf-8") == giant.strip()
 
 
-def test_a_pass_never_finishes_still_over_threshold(tmp_path, event_stream_limits):
-    """The core invariant the double-pass violated.
-
-    A summarization pass that returns with the stream still above
-    summarize_at_tokens has bought nothing — the next log() re-triggers it. Drive
-    a realistic mix (ordinary events, exempt oversized retrieval results, and
-    protected requirements) and assert the stream is back under budget after
-    every single append.
-    """
+def test_a_fold_gets_under_budget_and_a_second_one_is_free(tmp_path, event_stream_limits):
+    """One requested fold must leave the stream near keep_recent_tokens, with
+    protected events intact; asking again with nothing foldable costs no call."""
     llm = _CountingLLM()
     es = _stream(tmp_path, llm, event_stream_limits)
 
@@ -89,14 +89,17 @@ def test_a_pass_never_finishes_still_over_threshold(tmp_path, event_stream_limit
                 "matched line " + ("y " * 90_000),
                 action_name="grep_files" if i % 50 == 0 else "read_file",
             )
-        assert es._total_tokens < es.summarize_at_tokens, (
-            f"stream left at {es._total_tokens} tokens after event {i} — "
-            "the next append will re-trigger summarization immediately"
-        )
 
-    # The protected contract survived all of it.
+    es.summarize_by_LLM()
+    after_first = es._total_tokens
+    assert after_first < 10000 + 4000  # keep_recent_tokens plus a pinned/protected margin
     assert any(r.event.kind == "requirements" for r in es.tail_events)
 
+    calls = llm.calls
+    es.summarize_by_LLM()
+    assert llm.calls == calls
+    assert es._total_tokens <= after_first
+
 
 def test_tiny_foldable_region_is_pruned_not_summarized(tmp_path, event_stream_limits):
     """The 31,907 -> 31,529 case: a 15s LLM call that reclaimed 378 tokens.
@@ -111,10 +114,12 @@ def test_tiny_foldable_region_is_pruned_not_summarized(tmp_path, event_stream_li
     # A small foldable prefix...
     for i in range(3):
         es.log("action_end", f"action {i} completed")
-    # ...behind a wall of protected events that alone breach the threshold.
+    # ...behind a wall of protected events.
     for i in range(80):
         es.log("requirements", f"[ ] requirement {i}: " + "r " * 500)
 
+    es.summarize_by_LLM()
+
     assert llm.calls == 0
     assert not any(r.event.kind == "action_end" for r in es.tail_events)
     assert sum(1 for r in es.tail_events if r.event.kind == "requirements") == 80
diff --git a/tests/test_event_stream_protection.py b/tests/test_event_stream_protection.py
index 1cc8c18b..6d3ad60d 100644
--- a/tests/test_event_stream_protection.py
+++ b/tests/test_event_stream_protection.py
@@ -24,7 +24,7 @@ def generate_response(self, user_prompt=None, prompt_name=None, **kw):
 def test_requirements_survive_summarization(event_stream_limits):
     assert "requirements" in PROTECTED_SUMMARY_KINDS
 
-    event_stream_limits(2100, 100)  # min allowed given the 2000 internal buffer
+    event_stream_limits(100)
     es = EventStream(llm=_FakeLLM())
 
     # The protected contract, logged FIRST so it becomes the oldest event.
@@ -39,6 +39,7 @@ def test_requirements_survive_summarization(event_stream_limits):
             f"action {i} completed and produced some output text to add tokens",
         )
 
+    es.summarize_by_LLM()
     kinds = [r.event.kind for r in es.tail_events]
 
     # Summarization actually happened (old filler collapsed into the summary)…
@@ -56,7 +57,7 @@ def test_requirements_survive_summarization(event_stream_limits):
 def test_protected_only_region_is_noop(event_stream_limits):
     # If the only summarizable-aged content is protected, nothing is collapsed
     # (and it doesn't crash).
-    event_stream_limits(2100, 100)
+    event_stream_limits(100)
     es = EventStream(llm=_FakeLLM())
     es.log("requirements", "\n  [ ] x: y\n         done_when: z")
     es.summarize_by_LLM()  # force; region is tiny + protected
diff --git a/tests/test_llm_reinitialize.py b/tests/test_llm_reinitialize.py
index f21e574f..0b820c6b 100644
--- a/tests/test_llm_reinitialize.py
+++ b/tests/test_llm_reinitialize.py
@@ -3,7 +3,7 @@
 
 Covers the fix for: a model-only (or no-op) Settings save was
 unconditionally wiping every active task's session-cache state
-(_session_system_prompts + per-provider message histories), even though
+(_session_system_prompts + the accumulated session history), even though
 those buffers are only invalidated by an actual *provider* change — the
 message format they hold is provider-specific, not model-specific.
 """
@@ -39,7 +39,7 @@ def llm_interface():
         )
     # Seed accumulated session state as if a task were mid-flight.
     interface._session_system_prompts["task-1:reasoning"] = "system prompt"
-    interface._anthropic_session_messages["task-1:reasoning"] = [
+    interface._session_histories["task-1:reasoning"] = [
         {"role": "user", "content": "hi"},
         {"role": "assistant", "content": "hello"},
     ]
@@ -58,7 +58,7 @@ def test_reinitialize_noop_preserves_everything(llm_interface):
 
     assert ok is True
     mock_create.assert_not_called()
-    assert llm_interface._anthropic_session_messages["task-1:reasoning"]
+    assert llm_interface._session_histories["task-1:reasoning"]
     assert llm_interface._session_system_prompts["task-1:reasoning"] == "system prompt"
 
 
@@ -74,7 +74,7 @@ def test_reinitialize_model_only_preserves_histories(llm_interface):
 
     assert ok is True
     assert llm_interface.model == "claude-b"
-    assert llm_interface._anthropic_session_messages["task-1:reasoning"]
+    assert llm_interface._session_histories["task-1:reasoning"]
     assert llm_interface._session_system_prompts["task-1:reasoning"] == "system prompt"
 
 
@@ -92,5 +92,5 @@ def test_reinitialize_provider_change_clears_histories(llm_interface):
 
     assert ok is True
     assert llm_interface.provider == "openai"
-    assert llm_interface._anthropic_session_messages == {}
+    assert llm_interface._session_histories == {}
     assert llm_interface._session_system_prompts == {}