fix(gateway): suppress stream consumer on API failure and cap fallback chunks to prevent Telegram flood

agent/agent_init.py

@@ -866,6 +866,13 @@ def init_agent(
     agent._last_activity_desc: str = "initializing"
     agent._current_tool: str | None = None
     agent._api_call_count: int = 0
+    # Short summary of the model-API failure, set when the conversation loop
+    # exhausts retries on a terminal error (rate-limit / 429 / connection
+    # drop).  The gateway's stream consumer reads this via ``api_failed_summary``
+    # to suppress a partial / oversized streamed buffer and deliver a single
+    # clean error instead of flooding Telegram with split messages.  ``None``
+    # when the last turn succeeded (or hasn't failed yet).
+    agent.api_failed_summary: Optional[str] = None
     # Opt-out flag for the between-turns MCP tool refresh (build_turn_context).
     # Set on internal forks (e.g. background_review) that must keep ``tools[]``
     # byte-identical to a parent for provider cache parity.

agent/conversation_loop.py

@@ -1211,6 +1211,10 @@ def run_conversation(
     _last_preflight_pressure: Optional[int] = None
     _preflight_compression_blocked = _ctx.preflight_compression_blocked
     _turn_exit_reason = "unknown"  # Diagnostic: why the loop ended
+    # Reset the model-API failure summary at the start of every turn so a
+    # stale error from a previous failed turn can never suppress a valid
+    # response in the gateway's stream consumer.
+    agent.api_failed_summary = None
     # Last composed answer intentionally held back by a verification gate. If
     # that continuation consumes the remaining budget, this is the best
     # user-facing result available; it must not be confused with error or
@@ -5111,6 +5115,11 @@ def _perform_api_call(next_api_kwargs):
                     # Terminal — flush buffered retry/fallback trace.
                     agent._flush_status_buffer()
                     _final_summary = agent._summarize_api_error(api_error)
+                    # Surface the failure to the gateway's stream consumer so
+                    # it can suppress a partial / oversized streamed buffer
+                    # (e.g. echoed system prompt) and deliver one clean error
+                    # instead of flooding the user with split messages.
+                    agent.api_failed_summary = _final_summary
                     _billing_guidance = ""
                     if classified.reason == FailoverReason.billing:
                         agent._emit_status(f"❌ Billing or credits exhausted — {_final_summary}")
@@ -5442,6 +5451,13 @@ def _perform_api_call(next_api_kwargs):
         if response is None:
             _turn_exit_reason = "all_retries_exhausted_no_response"
             print(f"{agent.log_prefix}❌ All API retries exhausted with no successful response.")
+            # Surface the failure to the gateway's stream consumer so it can
+            # suppress a partial / oversized streamed buffer and deliver one
+            # clean error instead of flooding the user with split messages.
+            agent.api_failed_summary = (
+                getattr(agent, "api_failed_summary", None)
+                or "All API retries exhausted with no successful response."
+            )
             agent._persist_session(messages, conversation_history)
             break
 

gateway/run.py

@@ -4191,6 +4191,18 @@ def run_sync(self):
                         on_before_finalize=_pause_typing_before_finalize,
                         initial_reply_to_id=ctx.event_message_id,
                         run_still_current=ctx._run_still_current,
+                        # Lazily report the agent's model-API failure so
+                        # the stream consumer can suppress a partial /
+                        # oversized streamed buffer (e.g. echoed system
+                        # prompt) and deliver a single clean error instead
+                        # of flooding the user with split messages.
+                        api_error_fn=(
+                            lambda: getattr(
+                                ctx.agent_holder[0], "api_failed_summary", None
+                            )
+                            if ctx.agent_holder and ctx.agent_holder[0] is not None
+                            else None
+                        ),
                     )
                     if _want_stream_deltas:
                         def _stream_delta_cb(text: str) -> None:

gateway/stream_consumer.py

@@ -199,6 +199,7 @@ def __init__(
         on_before_finalize: Optional[Callable[[], Any]] = None,
         initial_reply_to_id: Optional[str] = None,
         run_still_current: Optional[Callable[[], bool]] = None,
+        api_error_fn: Optional[Callable[[], Optional[str]]] = None,
     ):
         self.adapter = adapter
         self.chat_id = chat_id
@@ -283,6 +284,17 @@ def __init__(
         # continuing to edit and deliver stale deltas.
         self._run_still_current = run_still_current or (lambda: True)
 
+        # API-failure detector.  When the agent's model call fails
+        # (rate-limit / 429 / connection drop), this callable returns the
+        # short error summary (or a non-empty truthy string).  The consumer
+        # then suppresses the accumulated streamed content at final flush and
+        # delivers just the clean error instead of blasting the partial
+        # buffer (which may contain the full system prompt / skill context)
+        # to the user — that buffer is exactly what produced the Telegram
+        # flood-of-messages symptom when the model API was down.  Optional;
+        # consumers that don't wire it keep the legacy (dangerous) behaviour.
+        self._api_error_fn = api_error_fn
+
         # Think-block filter state (mirrors CLI's _stream_delta tag suppression)
         self._in_think_block = False
         self._think_buffer = ""
@@ -803,6 +815,17 @@ async def run(self) -> None:
                 ):
                     should_edit = False
                 if should_edit and self._accumulated:
+                    # API-failure guard (see _send_api_error_final): if the
+                    # model call failed, never flush the accumulated buffer to
+                    # the user — deliver only the clean error.  Intercepted
+                    # here so it covers both mid-stream overflow splits and
+                    # the final flush at got_done.
+                    _api_err = (
+                        self._api_error_fn() if self._api_error_fn else None
+                    )
+                    if _api_err:
+                        await self._send_api_error_final(_api_err)
+                        return
                     # Split overflow: if accumulated text exceeds the platform
                     # limit, split into properly sized chunks.
                     if (
@@ -949,6 +972,22 @@ async def run(self) -> None:
                 if got_done:
                     if self._accumulated or self._message_id is not None or self._already_sent:
                         await self._notify_before_finalize()
+
+                    # API-failure guard: if the model call failed (rate-limit /
+                    # 429 / connection drop), the accumulated buffer is partial,
+                    # possibly huge (echoed system prompt / skill context), and
+                    # must NOT be flushed to the user.  Deliver only a single
+                    # clean error message and mark the response as sent so the
+                    # gateway's own final-send path skips re-delivering the raw
+                    # buffer.  This is the primary fix for the Telegram flood
+                    # symptom when the model API is down.
+                    _api_err = (
+                        self._api_error_fn() if self._api_error_fn else None
+                    )
+                    if _api_err:
+                        await self._send_api_error_final(_api_err)
+                        return
+
                     # Final edit without cursor. If progressive editing failed
                     # mid-stream, send a single continuation/fallback message
                     # here instead of letting the base gateway path send the
@@ -1242,6 +1281,38 @@ def _truncate_for_stream(
             return self._split_text_chunks(text, limit, len_fn)
         return list(chunks)
 
+    async def _send_api_error_final(self, error_summary: str) -> None:
+        """Deliver a single clean error message when the model API failed.
+
+        Called from the final-flush path when ``api_error_fn`` reports the
+        agent's model call failed.  The accumulated streamed buffer is
+        discarded (it may be partial and huge — echoed system prompt / skill
+        context) so we never flush it to the user.  We mark the response as
+        delivered so the gateway's own final-send path skips re-delivering the
+        raw buffer (which would otherwise be chunked into another flood).
+        """
+        # Keep it short and user-friendly.  The long error detail stays in the
+        # gateway/agent logs; the user only needs to know to retry.
+        _safe = f"⚠️ Model API error — no response was generated. {error_summary}".strip()
+        try:
+            result = await self.adapter.send(
+                chat_id=self.chat_id,
+                content=_safe,
+                metadata=self._metadata_for_send(final=True),
+            )
+        except Exception:
+            result = None
+        # Mark delivered regardless of send success: if the send itself failed
+        # (e.g. Telegram also unreachable), the gateway's fallback send will
+        # still try the agent's short error message.  We never want the large
+        # accumulated buffer to be re-delivered.
+        self._already_sent = True
+        self._final_response_sent = True
+        self._final_content_delivered = True
+        if result and result.success and result.message_id:
+            self._message_id = str(result.message_id)
+            self._last_sent_text = _safe
+
     async def _send_fallback_final(self, text: str) -> None:
         """Send the final continuation after streaming edits stop working.