bug:context overflow causing event stream collapsed

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

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

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

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":