Python: Concatenate code-interpreter stream deltas with prefix overlap (#8955)

python/packages/core/agent_framework/_types.py

@@ -2350,21 +2350,48 @@ def _content_items_text(items: Any) -> str | None:
     return "".join(text_parts)
 
 
+def _content_items_are_complete_value(items: Any) -> bool:
+    """Return True when items are marked as an authoritative full value.
+
+    Providers that resend a complete snapshot after incremental deltas should set
+    ``additional_properties["stream_complete"]`` (or ``"complete"``) on the text
+    content. Core only looks at that generic flag -- not provider event shapes --
+    so ordinary deltas always concatenate even when one text is a string prefix of
+    the other (see #8955 / #8903).
+    """
+    if not isinstance(items, list):
+        return False
+    for item in cast(list[object], items):
+        if not isinstance(item, Content):
+            continue
+        props = item.additional_properties or {}
+        if props.get("stream_complete") or props.get("complete"):
+            return True
+    return False
+
+
 def _merge_content_item_lists(existing: Any, incoming: Any) -> Any:
-    """Merge streamed nested content lists, replacing deltas with a later full value when present."""
+    """Merge streamed nested content lists.
+
+    Incremental text deltas are always concatenated. A later complete/full value
+    (``stream_complete`` / ``complete`` on ``additional_properties``) replaces the
+    accumulation.
+    """
     if incoming is None:
         return existing
     if existing is None:
         return deepcopy(incoming)
 
+    if _content_items_are_complete_value(incoming):
+        return deepcopy(incoming)
+
+    # Empty accumulation (e.g. initial outputs=[]) keeps the first real payload.
+    if isinstance(existing, list) and len(cast(list[object], existing)) == 0:
+        return deepcopy(incoming)
+
     existing_text = _content_items_text(existing)
     incoming_text = _content_items_text(incoming)
     if existing_text is not None and incoming_text is not None:
-        if incoming_text.startswith(existing_text):
-            return deepcopy(incoming)
-        if existing_text.startswith(incoming_text):
-            return existing
-
         existing_items = cast(list[Content], existing)
         merged = deepcopy(existing_items[0])
         merged.text = existing_text + incoming_text

python/packages/core/tests/core/test_sessions.py

@@ -673,7 +673,13 @@ async def test_after_run_stores_coalesced_code_interpreter_chunks(self) -> None:
                 contents=[
                     Content.from_code_interpreter_tool_call(
                         call_id="ci_123",
-                        inputs=[Content.from_text(text="import pandas as pd")],
+                        # Final snapshot: authoritative full value via stream_complete.
+                        inputs=[
+                            Content.from_text(
+                                text="import pandas as pd",
+                                additional_properties={"stream_complete": True},
+                            )
+                        ],
                         additional_properties={"sequence_number": 3},
                     )
                 ],

python/packages/core/tests/core/test_types.py

@@ -421,6 +421,81 @@ def test_code_interpreter_tool_call_content_parses_inputs():
     assert call.inputs[0].text == "print('hi')"
 
 
+def test_code_interpreter_tool_call_stream_concatenates_prefix_deltas() -> None:
+    """Genuine deltas must concatenate even when one text is a prefix of the other (#8955)."""
+    updates = [
+        ChatResponseUpdate(
+            contents=[Content.from_code_interpreter_tool_call(call_id="ci_1", inputs=[Content.from_text(text="(")])]
+        ),
+        ChatResponseUpdate(
+            contents=[Content.from_code_interpreter_tool_call(call_id="ci_1", inputs=[Content.from_text(text="()")])]
+        ),
+    ]
+    resp = ChatResponse.from_updates(updates)
+    joined = "".join(i.text or "" for i in resp.messages[0].contents[0].inputs or [])
+    assert joined == "(()"
+
+
+def test_code_interpreter_tool_call_stream_concatenates_indent_prefix_deltas() -> None:
+    deltas = ["    ", "    print(i)\n"]
+    updates = [
+        ChatResponseUpdate(
+            contents=[Content.from_code_interpreter_tool_call(call_id="ci_1", inputs=[Content.from_text(text=d)])]
+        )
+        for d in deltas
+    ]
+    resp = ChatResponse.from_updates(updates)
+    joined = "".join(i.text or "" for i in resp.messages[0].contents[0].inputs or [])
+    assert joined == "        print(i)\n"
+
+
+def test_code_interpreter_tool_call_stream_done_replaces_accumulated_deltas() -> None:
+    """A stream_complete snapshot is authoritative and replaces prior delta text."""
+    updates = [
+        ChatResponseUpdate(
+            contents=[
+                Content.from_code_interpreter_tool_call(call_id="ci_1", inputs=[Content.from_text(text="import os\n")])
+            ]
+        ),
+        ChatResponseUpdate(
+            contents=[
+                Content.from_code_interpreter_tool_call(
+                    call_id="ci_1",
+                    inputs=[
+                        Content.from_text(
+                            text="import os\nimport sys",
+                            additional_properties={"stream_complete": True},
+                        )
+                    ],
+                )
+            ]
+        ),
+    ]
+    resp = ChatResponse.from_updates(updates)
+    joined = "".join(i.text or "" for i in resp.messages[0].contents[0].inputs or [])
+    assert joined == "import os\nimport sys"
+
+
+def test_code_interpreter_tool_result_empty_outputs_then_text() -> None:
+    """Initial empty outputs=[] must not IndexError when a text chunk arrives."""
+    updates = [
+        ChatResponseUpdate(
+            contents=[Content.from_code_interpreter_tool_result(call_id="ci_1", outputs=[])]
+        ),
+        ChatResponseUpdate(
+            contents=[
+                Content.from_code_interpreter_tool_result(
+                    call_id="ci_1",
+                    outputs=[Content.from_text(text="ok")],
+                )
+            ]
+        ),
+    ]
+    resp = ChatResponse.from_updates(updates)
+    outputs = resp.messages[0].contents[0].outputs or []
+    assert "".join(o.text or "" for o in outputs if o.type == "text") == "ok"
+
+
 def test_code_interpreter_tool_result_content_outputs():
     result = Content.from_code_interpreter_tool_result(
         call_id="call-2",

python/packages/openai/agent_framework_openai/_chat_client.py

@@ -3863,6 +3863,8 @@ def parse_annotations(
                     "output_index": event.output_index,
                     "sequence_number": event.sequence_number,
                     "item_id": event.item_id,
+                    # Authoritative full code; core merges replace prior deltas.
+                    "stream_complete": True,
                 }
                 contents.append(
                     Content.from_code_interpreter_tool_call(

python/packages/openai/tests/openai/test_openai_chat_client.py

@@ -8205,10 +8205,12 @@ def test_parse_chunk_from_openai_code_interpreter_done() -> None:
     assert result.contents[0].inputs[0].type == "text"
     assert result.contents[0].inputs[0].text is not None
     assert "import pandas as pd" in result.contents[0].inputs[0].text
-    # Verify additional_properties for stream ordering
+    # Verify additional_properties for stream ordering and authoritative snapshot
     assert result.contents[0].additional_properties["output_index"] == 0
     assert result.contents[0].additional_properties["sequence_number"] == 5
     assert result.contents[0].additional_properties["item_id"] == "ci_456"
+    assert result.contents[0].additional_properties["stream_complete"] is True
+    assert result.contents[0].inputs[0].additional_properties["stream_complete"] is True
 
 
 def test_parse_chunk_from_openai_reasoning() -> None: