From bfeafa2d68a00c5684c95fa14df8bbdec75135c1 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Thu, 24 Sep 2026 17:00:24 +0530 Subject: [PATCH] fix(agent): collapse orphan continuation trail on retry exhaustion (#119001) Ports #119027::finalize_continuation_partial: when a stream drop entered the continuation path and the next request exhausted retries before a token, the _length_continuation_fragment/_nudge rows were persisted as-is, so resume replayed a dangling synthetic user nudge. Collapse them into one assistant row before persistence and feed the collapsed text to the #119081 partial-retention path (the fragment rows are gone by then). Co-authored-by: fangliquan --- agent/turn_api_error.py | 13 ++-- agent/turn_recovery.py | 54 +++++++--------- agent/turn_truncation.py | 63 ++++++++++++++----- ...test_retry_exhaustion_partial_retention.py | 17 +++++ 4 files changed, 96 insertions(+), 51 deletions(-) diff --git a/agent/turn_api_error.py b/agent/turn_api_error.py index 63ce59bb06..90c095df3f 100644 --- a/agent/turn_api_error.py +++ b/agent/turn_api_error.py @@ -22,7 +22,7 @@ from agent.turn_recovery import ( _NONRETRYABLE_LABELS, abort_turn_on_interrupt, compute_error_backoff, interruptible_backoff_sleep, log_api_error_attempt, max_retries_exhausted_result, nonretryable_client_error_result, recover_after_classification, - recover_before_classification, route_classified_error, + recover_before_classification, route_classified_error, settle_delivered_partial, ) logger = logging.getLogger("agent.conversation_loop") @@ -55,6 +55,7 @@ def handle_api_error( conversation_history: Any, approx_tokens: Any, retry_count: Any, max_retries: Any, compression_attempts: Any, max_compression_attempts: Any, api_call_count: Any, api_request_id: Any, api_start_time: Any, effective_task_id: Any, turn_id: Any, + current_turn_user_idx: Any = None, ) -> ApiErrorVerdict: """Recover from ``api_error`` in the original order. Every fallback activation must leave the retry loop with ``restart_with_rebuilt_messages`` armed (``"break"``) so the pre-API @@ -212,7 +213,7 @@ def handle_api_error( api_messages=api_messages, api_kwargs=api_kwargs, active_system_prompt=active_system_prompt, conversation_history=conversation_history, approx_tokens=approx_tokens, retry_count=retry_count, max_retries=max_retries, compression_attempts=compression_attempts, - api_call_count=api_call_count, + api_call_count=api_call_count, current_turn_user_idx=current_turn_user_idx, ) active_system_prompt = _ue.active_system_prompt retry_count = _ue.retry_count @@ -255,6 +256,7 @@ def settle_unrecovered_error( _provider: Any, _base: Any, _model: Any, messages: Any, api_messages: Any, api_kwargs: Any, active_system_prompt: Any, conversation_history: Any, approx_tokens: Any, retry_count: Any, max_retries: Any, compression_attempts: Any, api_call_count: Any, error_context: Any = None, + current_turn_user_idx: Any = None, ) -> UnrecoveredErrorVerdict: """Decide the fate of an API error that every recovery chain declined: local validation / non-retryable client errors (Copilot stale-credential self-heal first, then fallback, then a @@ -342,11 +344,13 @@ def settle_unrecovered_error( active_system_prompt = _arm_fallback_restart(agent, api_messages, active_system_prompt, _retry) retry_count = compression_attempts = 0 return _verdict("break") + # Terminal from here: collapse the continuation trail once, before the persist. + _delivered = settle_delivered_partial(agent, messages, current_turn_user_idx) return _verdict("return", nonretryable_client_error_result( agent, api_error, classified, status_code=status_code, api_kwargs=api_kwargs, api_messages=api_messages, messages=messages, conversation_history=conversation_history, api_call_count=api_call_count, approx_tokens=approx_tokens, provider=_provider, - base_url=_base, model=_model, + base_url=_base, model=_model, delivered=_delivered, )) if retry_count >= max_retries: @@ -383,12 +387,13 @@ def settle_unrecovered_error( if _ladder["action"] == "continue": retry_count = 0 return _verdict(_ladder["action"], _ladder.get("result")) + _delivered = settle_delivered_partial(agent, messages, current_turn_user_idx) return _verdict("return", max_retries_exhausted_result( agent, api_error, classified, max_retries=max_retries, is_rate_limited=is_rate_limited, error_msg=error_msg, api_kwargs=api_kwargs, api_messages=api_messages, messages=messages, conversation_history=conversation_history, api_call_count=api_call_count, approx_tokens=approx_tokens, provider=_provider, - base_url=_base, model=_model, + base_url=_base, model=_model, delivered=_delivered, )) wait_time = compute_error_backoff( diff --git a/agent/turn_recovery.py b/agent/turn_recovery.py index 36effa52c9..c00d86ae3c 100644 --- a/agent/turn_recovery.py +++ b/agent/turn_recovery.py @@ -773,37 +773,26 @@ def _failed_turn_result(final_response: str, messages: Any, api_call_count: int, } -def _recover_delivered_partial_text(agent: Any, messages: Any) -> str: - """Visible assistant text already delivered this turn ("" when none). +def settle_delivered_partial(agent: Any, messages: Any, current_turn_user_idx: Any) -> str: + """Visible text already delivered this turn ("" when none), collapsing any continuation + trail first so the terminal persist never keeps a dangling synthetic nudge (#119001). - ``build_api_request`` resets ``_current_streamed_assistant_text`` on every - attempt, so after a mid-stream death + continuation + pre-stream 429 the - live accumulator is empty and the only record is the - ``_length_continuation_fragment`` rows the truncation path appended (#119001). + ``build_api_request`` resets ``_current_streamed_assistant_text`` per attempt, so after a + mid-stream death + continuation + pre-stream error the live accumulator is empty and the + fragment rows (now collapsed into one assistant row) are the only record. """ - try: - _live = getattr(agent, "_current_streamed_assistant_text", "") or "" - except Exception: - _live = "" - if isinstance(_live, str) and _live.strip(): - return _live.strip() - _parts = [ - m["content"].strip() for m in messages or () - if isinstance(m, dict) and m.get("_length_continuation_fragment") - and isinstance(m.get("content"), str) and m["content"].strip() - ] - if not _parts: - return "" - # Same glue as _join_truncated_parts: newline where two parts would stick. - _joined = "" - for _part in _parts: - if _joined and not _joined[-1].isspace() and not _part[0].isspace(): - _joined += "\n" - _joined += _part - return _joined.strip() + from agent.turn_truncation import collapse_continuation_trail + collapsed = collapse_continuation_trail( + agent, messages, current_turn_user_idx, finish_reason="error", + ) + live = getattr(agent, "_current_streamed_assistant_text", "") + if isinstance(live, str) and live.strip(): + from agent.agent_runtime_helpers import strip_think_blocks + return strip_think_blocks(agent, live).strip() or collapsed + return collapsed -def _with_delivered_partial(final_response: str, error_summary: str, agent: Any, messages: Any) -> tuple: +def _with_delivered_partial(final_response: str, error_summary: str, delivered: str) -> tuple: """Prepend delivered partial text to a terminal error body ("" unchanged). Returns ``(final_response, keep_partial)``; callers set ``result["partial"]`` @@ -811,8 +800,8 @@ def _with_delivered_partial(final_response: str, error_summary: str, agent: Any, retain the bubble instead of clearing it. ``final_response`` must stay distinct from ``error`` — that inequality is the retention contract. """ - _delivered = _recover_delivered_partial_text(agent, messages) - if not _delivered or _delivered.strip() == (error_summary or "").strip(): + _delivered = (delivered or "").strip() + if not _delivered or _delivered == (error_summary or "").strip(): return final_response, False return f"{_delivered}\n\n{final_response}", True @@ -984,6 +973,7 @@ def nonretryable_client_error_result( agent: Any, api_error: Exception, classified: Any, *, status_code: Optional[int], api_kwargs: Any, api_messages: Any, messages: List[Dict[str, Any]], conversation_history: Any, api_call_count: int, approx_tokens: int, provider: Any, base_url: Any, model: Any, + delivered: str = "", ) -> Dict[str, Any]: """Terminal path for a non-retryable 4xx once fallback is exhausted: debug dump, flush the retry trace, print auth / billing / content-policy / TLS guidance, persist (skipped @@ -1091,7 +1081,7 @@ def nonretryable_client_error_result( # (agent/error_surface.py) reads a rejected OAuth token as a retryable # "Provider error" and offers Retry instead of a re-login. _final_response, _keep_partial = _with_delivered_partial( - _final_response, _nonretryable_summary, agent, messages, + _final_response, _nonretryable_summary, delivered, ) result = _failed_turn_result(_final_response, messages, api_call_count, _nonretryable_summary) result.update({ @@ -1118,7 +1108,7 @@ def max_retries_exhausted_result( agent: Any, api_error: Exception, classified: Any, *, max_retries: int, is_rate_limited: bool, error_msg: str, api_kwargs: Any, api_messages: Any, messages: List[Dict[str, Any]], conversation_history: Any, api_call_count: int, approx_tokens: int, provider: Any, - base_url: Any, model: Any, + base_url: Any, model: Any, delivered: str = "", ) -> Dict[str, Any]: """Terminal path once retries, transport recovery and fallback all failed: flush the trace, emit the billing / rate-limit / generic status, print stream-drop or thinking-timeout @@ -1243,7 +1233,7 @@ def max_retries_exhausted_result( # shown, so keep it as the reply (marked failed) instead of an error-only # turn — the gateway flags ``partial`` and surfaces retain the bubble. _final_response, _keep_partial = _with_delivered_partial( - _final_response, _final_summary, agent, messages, + _final_response, _final_summary, delivered, ) if _keep_partial: result["final_response"] = _final_response diff --git a/agent/turn_truncation.py b/agent/turn_truncation.py index 7a1dc6487c..895b9251e1 100644 --- a/agent/turn_truncation.py +++ b/agent/turn_truncation.py @@ -42,6 +42,49 @@ _CONTEXT_OVERFLOW_PARTIAL_FINAL = ( "chats are reset automatically)." ) +def collapse_continuation_trail( + agent: Any, messages: List[Dict[str, Any]], current_turn_user_idx: Any, *, + finish_reason: str, parts: Optional[List[str]] = None, +) -> str: + """Drop this turn's ``_length_continuation_fragment``/``_nudge`` rows and append one + assistant row holding the joined, think-stripped partial; returns that text ("" none). + + ``parts=None`` (retry exhaustion, #119001): the text comes from the fragment rows and + nothing happens without a valid turn index or a trail — an unanswered synthetic nudge + must never be persisted, and an earlier turn's rows must never be read. Explicit + ``parts`` (the continuation ceiling) always appends, scanning from 0 without an index. + """ + idx = current_turn_user_idx + valid_idx = isinstance(idx, int) and idx >= 0 + if parts is None and not (valid_idx and idx < len(messages)): + return "" + turn_start = idx + 1 if valid_idx else 0 + fragment_parts: List[str] = [] + retained: List[Any] = [] + found_trail = False + for message in messages[turn_start:]: + if isinstance(message, dict) and ( + message.get("_length_continuation_fragment") or message.get("_length_continuation_nudge") + ): + found_trail = True + content = message.get("content") + if message.get("_length_continuation_fragment") and isinstance(content, str) and content: + fragment_parts.append(content) + continue + retained.append(message) + if parts is None and not found_trail: + return "" + messages[turn_start:] = retained + from agent.conversation_loop import _join_truncated_parts + partial = agent._strip_think_blocks( + _join_truncated_parts(fragment_parts if parts is None else parts) + ).strip() + if partial: + append_message(messages, {"role": "assistant", "content": partial, "finish_reason": finish_reason}) + agent._session_messages = messages + return partial + + _THINKING_EXHAUSTED = ( "💭 Reasoning exhausted the output token budget — no visible response was produced.", "⚠️ **Thinking Budget Exhausted**\n\nThe model used all its output tokens on reasoning " @@ -288,7 +331,11 @@ def _continue_text(st: _Trunc, _retry: TurnRetryState, assistant_message: Any) - _retry.restart_with_length_continuation = True return st.done("break") - partial_response = agent._strip_think_blocks(_join_truncated_parts(st.truncated_response_parts)).strip() + # Unanswered continue nudges made every later turn re-truncate: drop the trail. + partial_response = collapse_continuation_trail( + agent, messages, st.current_turn_user_idx, finish_reason="length", + parts=st.truncated_response_parts, + ) # The one-shot reasoning-off override must not leak into the next turn. agent._ephemeral_reasoning_off = False agent._vprint( @@ -299,20 +346,6 @@ def _continue_text(st: _Trunc, _retry: TurnRetryState, assistant_message: Any) - else "no visible text was produced."), force=True, diagnostic=True, ) - # Unanswered continue nudges made every later turn re-truncate: drop the trail. - idx = st.current_turn_user_idx - _turn_start = idx + 1 if isinstance(idx, int) and idx >= 0 else 0 - messages[_turn_start:] = [ - m for m in messages[_turn_start:] - if not (isinstance(m, dict) and ( - m.get("_length_continuation_fragment") or m.get("_length_continuation_nudge") - )) - ] - if partial_response: - append_message(messages, { - "role": "assistant", "content": partial_response, "finish_reason": "length" - }) - agent._session_messages = messages if filled is not None: notice = _WINDOW_FILLED.format(prompt=filled[0], ctx=filled[1]) return st.end_turn( diff --git a/tests/agent/test_retry_exhaustion_partial_retention.py b/tests/agent/test_retry_exhaustion_partial_retention.py index 759f6b9b29..c30ec8e993 100644 --- a/tests/agent/test_retry_exhaustion_partial_retention.py +++ b/tests/agent/test_retry_exhaustion_partial_retention.py @@ -121,3 +121,20 @@ def test_nonretryable_terminal_keeps_delivered_partial(): assert result.get("partial") is True assert PARTIAL in result["final_response"] assert result["final_response"].strip() != str(result["error"]).strip() + + +def test_exhausted_429_collapses_continuation_trail_into_one_assistant_row(): + messages = _messages_with_fragment() + error = _Http(429, "HTTP 429: RequestBurstTooFast — slow down traffic growth") + classified = classify_api_error(error, provider="openrouter", model="m") + result = max_retries_exhausted_result( + _Agent(), error, classified, max_retries=3, is_rate_limited=True, + error_msg=str(error).lower(), api_kwargs=None, api_messages=[], messages=messages, + conversation_history=None, api_call_count=3, approx_tokens=10, provider="openrouter", + base_url="https://openrouter.ai/api/v1", model="m", current_turn_user_idx=0, + ) + # No dangling synthetic nudge: the turn persists as user -> one assistant row. + assert [m["role"] for m in messages] == ["user", "assistant"] + assert messages[1]["content"] == PARTIAL + assert not any(m.get("_length_continuation_nudge") for m in messages) + assert result.get("partial") is True and PARTIAL in result["final_response"]