From d98deef030dd5631eba862b10de0773ac19fb581 Mon Sep 17 00:00:00 2001 From: fangliquanflq Date: Wed, 26 Aug 2026 16:38:40 +0800 Subject: [PATCH] fix(agent): scope continuation dedupe to stream stubs (cherry picked from commit 82acba16e08cce9f613598325d7c566838fb002e) --- agent/conversation_loop.py | 15 ++++++--- agent/turn_final_response.py | 2 +- agent/turn_truncation.py | 6 ++-- .../test_partial_stream_finish_reason.py | 32 +++++++++++++++---- 4 files changed, 40 insertions(+), 15 deletions(-) diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index bc0cc85b22..ad696c9b29 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -288,15 +288,20 @@ def _continuation_overlap_length(previous: str, continuation: str) -> int: return matched if matched >= _MIN_CONTINUATION_OVERLAP else 0 -def _join_truncated_parts(parts: List[str]) -> str: - """Join continuation fragments without repeating a recovered tail.""" +def _join_truncated_parts(parts: List[str | tuple[str, bool]]) -> str: + """Join continuation fragments, deduping only interrupted-stream seams.""" joined = "" - for part in parts: - if joined and part: + previous_was_partial_stub = False + for fragment in parts: + part, is_partial_stub = ( + fragment if isinstance(fragment, tuple) else (fragment, False) + ) + if previous_was_partial_stub and joined and part: part = part[_continuation_overlap_length(joined, part):] if joined and not joined[-1].isspace() and part and not part[0].isspace(): joined += "\n" joined += part + previous_was_partial_stub = is_partial_stub return joined @@ -1377,7 +1382,7 @@ class _LoopState: restart_count: int = 0 _outer_error_count: int = 0 # outer-loop exceptions this turn (#92450), see _MAX_OUTER_LOOP_ERRORS truncated_tool_call_retries: int = 0 - truncated_response_parts: List[str] = field(default_factory=list) + truncated_response_parts: List[tuple[str, bool]] = field(default_factory=list) compression_attempts: int = 0 _last_preflight_pressure: Optional[int] = None # A provider overflow outweighs the rough-estimate calibration that defers preflight after diff --git a/agent/turn_final_response.py b/agent/turn_final_response.py index 779a473003..0c800ab6b0 100644 --- a/agent/turn_final_response.py +++ b/agent/turn_final_response.py @@ -229,7 +229,7 @@ def finish_text_response( codex_ack_continuations = 0 if truncated_response_parts: - final_response = _join_truncated_parts([*truncated_response_parts, final_response]) + final_response = _join_truncated_parts([*truncated_response_parts, (final_response, False)]) truncated_response_parts = [] length_continue_retries = 0 # The continuation recovered, so the fragments stay in the transcript. diff --git a/agent/turn_truncation.py b/agent/turn_truncation.py index 4ed96d7269..0197d9bbf4 100644 --- a/agent/turn_truncation.py +++ b/agent/turn_truncation.py @@ -131,7 +131,7 @@ class TruncationVerdict: result: Optional[Dict[str, Any]] messages: List[Dict[str, Any]] length_continue_retries: int - truncated_response_parts: List[str] + truncated_response_parts: List[Any] truncated_tool_call_retries: int retry_count: int compression_attempts: int @@ -257,7 +257,7 @@ def _continue_text(st: _Trunc, _retry: TurnRetryState, assistant_message: Any) - interim_msg = agent._build_assistant_message(assistant_message, st.finish_reason) interim_msg["_length_continuation_fragment"] = True # ceiling exit drops these append_message(messages, interim_msg) - st.truncated_response_parts.append(_interim_content) + st.truncated_response_parts.append((_interim_content, st.is_stub)) filled = st.window_filled if n < 4 and filled is None: @@ -360,7 +360,7 @@ def recover_from_truncation( agent: Any, response: Any, finish_reason: str, _retry: TurnRetryState, *, messages: List[Dict[str, Any]], conversation_history: Any, api_kwargs: Any, api_call_count: int, effective_task_id: Any, current_turn_user_idx: Any, length_continue_retries: int, - truncated_response_parts: List[str], truncated_tool_call_retries: int, retry_count: int, + truncated_response_parts: List[Any], truncated_tool_call_retries: int, retry_count: int, compression_attempts: int, ) -> TruncationVerdict: """Recover from a truncated response. Order is load-bearing: thinking exhaustion and diff --git a/tests/agent/test_partial_stream_finish_reason.py b/tests/agent/test_partial_stream_finish_reason.py index e66255b24d..76489f01ba 100644 --- a/tests/agent/test_partial_stream_finish_reason.py +++ b/tests/agent/test_partial_stream_finish_reason.py @@ -366,16 +366,16 @@ class TestLengthContinuationAssembly: ) joined = _join_truncated_parts([ - f"Investigation details.\n\n{conclusion}", - conclusion, + (f"Investigation details.\n\n{conclusion}", True), + (conclusion, False), ]) assert joined.count(conclusion) == 1 def test_distinct_continuation_is_preserved(self): assert _join_truncated_parts([ - "The first half ends here", - "and the second half adds new information.", + ("The first half ends here", True), + ("and the second half adds new information.", False), ]) == ( "The first half ends here\n" "and the second half adds new information." @@ -385,8 +385,8 @@ class TestLengthContinuationAssembly: repeated = "A sufficiently long repeated transition sentence ends here." assert _join_truncated_parts([ - f"Existing answer. {repeated}", - f"{repeated} New inbound details remain visible.", + (f"Existing answer. {repeated}", True), + (f"{repeated} New inbound details remain visible.", False), ]) == ( f"Existing answer. {repeated} New inbound details remain visible." ) @@ -491,6 +491,26 @@ class TestConversationLoopPartialStreamContinuation: assert "forty-two" in result["final_response"] assert result["final_response"].count(repeated_tail) == 1 + def test_output_limit_continuation_preserves_intentional_repetition(self, loop_agent): + from tests.run_agent.test_run_agent import _mock_response + + repeated = "This intentionally repeated sentence is longer than thirty-two characters." + first = _mock_response( + content=f"First copy: {repeated}", finish_reason=FINISH_REASON_LENGTH, + ) + continuation = _mock_response(content=repeated, finish_reason="stop") + loop_agent.client.chat.completions.create.side_effect = [first, continuation] + + with ( + patch.object(loop_agent, "_persist_session"), + patch.object(loop_agent, "_save_trajectory"), + patch.object(loop_agent, "_cleanup_task_resources"), + ): + result = loop_agent.run_conversation("repeat this sentence twice") + + assert loop_agent.client.chat.completions.create.call_count == 2 + assert result["final_response"].count(repeated) == 2 + class TestContentFilterStallActivatesFallback: """Regression for #32421: a provider output-layer content safety filter