fix(streaming): stop duplicating tool-call args from cumulative-resend providers (#35718)

DeepSeek / Baidu Qianfan stream tool-call arguments in cumulative mode:
each chunk resends the full arguments-so-far instead of the new fragment.
The stream accumulator blindly concatenated arg deltas with +=, turning
that into '{...}{...}{...}', which failed json.loads and got nuked to '{}'
— a silently corrupted tool call (#35592). Worse on multi-param tools
(search_files, session_search, memory replace) because longer args take
more chunks, giving more resend opportunities.

- Per-slot cumulative latch in the stream accumulator: a delta that is a
  strict superset of the accumulated buffer marks the slot cumulative and
  replaces (not appends); exact duplicates are dropped only after latching.
  Incremental fragments are untouched (default += path).
- Backstop _collapse_repeated_json_arguments() in the repair pipeline
  collapses pure identical-resend buffers (K exact repeats of a valid-JSON
  unit) for providers that resend the complete object from chunk 1. Only
  reached after json.loads already failed, so compliant single objects are
  never touched.

Not a gateway or DeepSeek-model bug — any OpenAI-wire provider in
cumulative streaming mode is affected.

tests/run_agent/test_repair_tool_call_arguments.py

@@ -140,3 +140,50 @@ class TestRepairToolCallArguments:
         parsed = json.loads(result)
         assert "line" in parsed["msg"]
 
+    # -- Stage 1b: cumulative-resend duplicate collapse (#35592) --
+
+    def test_collapse_duplicate_object_2x(self):
+        """Two concatenated identical objects collapse to one."""
+        raw = '{"path":"x"}{"path":"x"}'
+        result = _repair_tool_call_arguments(raw, "search_files")
+        assert json.loads(result) == {"path": "x"}
+
+    def test_collapse_duplicate_object_many(self):
+        """K concatenated identical objects collapse to one."""
+        raw = '{"a":1,"b":2}' * 100
+        result = _repair_tool_call_arguments(raw, "t")
+        assert json.loads(result) == {"a": 1, "b": 2}
+
+    def test_collapse_nested_object_repeat(self):
+        """Repeated nested-key object collapses correctly."""
+        raw = '{"command":{"command":"x"}}' * 3
+        result = _repair_tool_call_arguments(raw, "t")
+        assert json.loads(result) == {"command": {"command": "x"}}
+
+    def test_single_object_not_touched_by_collapse(self):
+        """A clean single object never enters the collapse path."""
+        raw = '{"path": "x.py"}'
+        result = _repair_tool_call_arguments(raw, "t")
+        assert json.loads(result) == {"path": "x.py"}
+
+    def test_single_nested_object_not_corrupted(self):
+        """Nested-key single object is NOT mistaken for a repeat (safety)."""
+        raw = '{"command":{"command":"x"}}'
+        result = _repair_tool_call_arguments(raw, "t")
+        assert json.loads(result) == {"command": {"command": "x"}}
+
+    def test_two_different_objects_not_collapsed(self):
+        """Distinct concatenated objects are not a clean repeat — collapse
+        declines and the generic repair handles it (returns '{}')."""
+        from agent.message_sanitization import _collapse_repeated_json_arguments
+        assert _collapse_repeated_json_arguments('{"a":1}{"b":2}') is None
+
+    def test_collapse_helper_returns_none_for_valid_single(self):
+        from agent.message_sanitization import _collapse_repeated_json_arguments
+        assert _collapse_repeated_json_arguments('{"a":1}') is None
+
+    def test_collapse_helper_returns_none_for_repeated_non_json(self):
+        from agent.message_sanitization import _collapse_repeated_json_arguments
+        # 'abab' repeats 'ab' but 'ab' is not valid JSON.
+        assert _collapse_repeated_json_arguments('abab') is None
+

tests/run_agent/test_streaming.py

@@ -184,6 +184,143 @@ class TestStreamingAccumulator:
         assert tc[0].function.name == "read_file"
         assert tc[0].function.arguments == '{"path": "x.py"}'
 
+    @patch("run_agent.AIAgent._create_request_openai_client")
+    @patch("run_agent.AIAgent._close_request_openai_client")
+    def test_args_not_duplicated_on_cumulative_growing_resend(self, mock_close, mock_create):
+        """DeepSeek / Baidu Qianfan stream args in cumulative mode (#35592).
+
+        Each chunk resends the full arguments-so-far (monotonic growth)
+        instead of the new fragment.  Blind += produced a duplicated
+        '{...}{...}' string that failed json.loads and got nuked to '{}'.
+        The per-slot cumulative latch must replace (not append) so the
+        final arguments are a single valid object.
+        """
+        import json as _json
+        from run_agent import AIAgent
+
+        full = '{"pattern": "def handle", "path": "gateway", "output_mode": "content"}'
+        # Growing prefixes, then identical full resends at the tail.
+        steps = [full[:12], full[:30], full[:48], full, full, full]
+        chunks = [_make_stream_chunk(tool_calls=[
+            _make_tool_call_delta(index=0, tc_id="call_qf", name="search_files")
+        ])]
+        for s in steps:
+            chunks.append(_make_stream_chunk(tool_calls=[
+                _make_tool_call_delta(index=0, tc_id="call_qf", name="search_files", arguments=s)
+            ]))
+        chunks.append(_make_stream_chunk(finish_reason="tool_calls"))
+
+        mock_client = MagicMock()
+        mock_client.chat.completions.create.return_value = iter(chunks)
+        mock_create.return_value = mock_client
+
+        agent = AIAgent(
+            api_key="test-key",
+            base_url="https://qianfan.baidubce.com/v2",
+            model="deepseek-v4-pro",
+            quiet_mode=True,
+            skip_context_files=True,
+            skip_memory=True,
+        )
+        agent.api_mode = "chat_completions"
+        agent._interrupt_requested = False
+
+        response = agent._interruptible_streaming_api_call({})
+
+        tc = response.choices[0].message.tool_calls
+        assert tc is not None and len(tc) == 1
+        assert _json.loads(tc[0].function.arguments) == _json.loads(full)
+        # Not flagged as truncated (which would set finish_reason='length').
+        assert response.choices[0].finish_reason == "tool_calls"
+
+    @patch("run_agent.AIAgent._create_request_openai_client")
+    @patch("run_agent.AIAgent._close_request_openai_client")
+    def test_args_identical_full_resend_collapsed_by_backstop(self, mock_close, mock_create):
+        """Provider resends the COMPLETE object identically from chunk 1.
+
+        There is never a strict superset to latch on, so the per-chunk
+        guard appends — but the post-loop _collapse_repeated_json_arguments
+        backstop must collapse the K-repeat to one valid object (#35592).
+        """
+        import json as _json
+        from run_agent import AIAgent
+
+        full = '{"action": "replace", "old_text": "a", "new_text": "b"}'
+        chunks = [_make_stream_chunk(tool_calls=[
+            _make_tool_call_delta(index=0, tc_id="call_id", name="memory")
+        ])]
+        for _ in range(5):
+            chunks.append(_make_stream_chunk(tool_calls=[
+                _make_tool_call_delta(index=0, tc_id="call_id", name="memory", arguments=full)
+            ]))
+        chunks.append(_make_stream_chunk(finish_reason="tool_calls"))
+
+        mock_client = MagicMock()
+        mock_client.chat.completions.create.return_value = iter(chunks)
+        mock_create.return_value = mock_client
+
+        agent = AIAgent(
+            api_key="test-key",
+            base_url="https://qianfan.baidubce.com/v2",
+            model="deepseek-v4-pro",
+            quiet_mode=True,
+            skip_context_files=True,
+            skip_memory=True,
+        )
+        agent.api_mode = "chat_completions"
+        agent._interrupt_requested = False
+
+        response = agent._interruptible_streaming_api_call({})
+
+        tc = response.choices[0].message.tool_calls
+        assert tc is not None and len(tc) == 1
+        assert _json.loads(tc[0].function.arguments) == _json.loads(full)
+
+    @patch("run_agent.AIAgent._create_request_openai_client")
+    @patch("run_agent.AIAgent._close_request_openai_client")
+    def test_incremental_args_with_duplicate_leading_fragment(self, mock_close, mock_create):
+        """Genuine incremental stream whose 2nd fragment equals the 1st.
+
+        Safety guard: an exact-equal delta on a NON-latched slot must be
+        treated as a real fragment and appended, never silently dropped.
+        Verifies the cumulative-resend fix does not corrupt nested-key
+        objects (the false-positive class flagged in review of #35592).
+        """
+        from run_agent import AIAgent
+
+        # Fragments deliberately repeat the leading '{"command":' substring.
+        frags = ['{"command":', '{"command":', ' "x"}', '}']
+        chunks = [_make_stream_chunk(tool_calls=[
+            _make_tool_call_delta(index=0, tc_id="c3", name="t")
+        ])]
+        for f in frags:
+            chunks.append(_make_stream_chunk(tool_calls=[
+                _make_tool_call_delta(index=0, tc_id="c3", name="t", arguments=f)
+            ]))
+        chunks.append(_make_stream_chunk(finish_reason="tool_calls"))
+
+        mock_client = MagicMock()
+        mock_client.chat.completions.create.return_value = iter(chunks)
+        mock_create.return_value = mock_client
+
+        agent = AIAgent(
+            api_key="test-key",
+            base_url="https://openrouter.ai/api/v1",
+            model="test/model",
+            quiet_mode=True,
+            skip_context_files=True,
+            skip_memory=True,
+        )
+        agent.api_mode = "chat_completions"
+        agent._interrupt_requested = False
+
+        response = agent._interruptible_streaming_api_call({})
+
+        tc = response.choices[0].message.tool_calls
+        assert tc is not None and len(tc) == 1
+        # All four fragments concatenated — nothing dropped.
+        assert tc[0].function.arguments == '{"command":{"command": "x"}}'
+
     @patch("run_agent.AIAgent._create_request_openai_client")
     @patch("run_agent.AIAgent._close_request_openai_client")
     def test_tool_call_extra_content_preserved(self, mock_close, mock_create):