Python: Fix code interpreter deltas dropped on prefix collision

python/packages/core/agent_framework/_types.py

@@ -469,6 +469,15 @@ class ComputerSafetyCheck(TypedDict, total=False):
 ContentT = TypeVar("ContentT", bound="Content")
 _MODEL_OUTPUT_KIND_KEY = "model_output_kind"
 _MODEL_OUTPUT_REFUSAL = "refusal"
+# Marks a streamed nested content item (e.g. a code interpreter call's `inputs`/`outputs`
+# text) as a full resend of everything streamed so far, rather than an incremental delta
+# to append. Some providers follow up a run of true deltas with one event that repeats the
+# complete value (e.g. the OpenAI Responses API's `code_interpreter_call_code.done`, sent
+# after a series of `.delta` events for the same call). Text content alone can't reliably
+# tell that apart from a delta that merely happens to start with what came before - e.g. a
+# delta of "(" followed by a delta of "()" is two chunks that combine into "(()", not a
+# resend of "()" - so providers must set this explicitly on a full resend.
+_CONTENT_ITEM_SNAPSHOT_KEY = "content_item_snapshot"
 
 # endregion
 
@@ -2524,21 +2533,55 @@ def _content_items_text(items: Any) -> str | None:
     return "".join(text_parts)
 
 
+def _is_content_item_snapshot(items: Any) -> bool:
+    """Whether a nested content list is tagged as a full resend, see `_CONTENT_ITEM_SNAPSHOT_KEY`."""
+    if not isinstance(items, list):
+        return False
+    return any(
+        isinstance(item, Content) and item.additional_properties.get(_CONTENT_ITEM_SNAPSHOT_KEY)
+        for item in cast("list[object]", items)
+    )
+
+
+def _strip_snapshot_marker(items: Any) -> None:
+    """Remove the merge-only snapshot marker from a nested content list, in place.
+
+    The marker is only an instruction for the merge step that reads it; stripping it
+    keeps it from persisting into finalized content that gets serialized into history
+    or passed to middleware.
+    """
+    if isinstance(items, list):
+        for item in cast("list[object]", items):
+            if isinstance(item, Content):
+                item.additional_properties.pop(_CONTENT_ITEM_SNAPSHOT_KEY, None)
+
+
+def _copy_without_snapshot_marker(items: Any) -> Any:
+    """Deep-copy a nested content list, stripping the merge-only snapshot marker."""
+    copied = deepcopy(items)
+    _strip_snapshot_marker(copied)
+    return copied
+
+
 def _merge_content_item_lists(existing: Any, incoming: Any) -> Any:
     """Merge streamed nested content lists, replacing deltas with a later full value when present."""
     if incoming is None:
         return existing
-    if existing is None:
+    # A snapshot can be the very first (or only) chunk seen for this call - not just one
+    # that arrives after prior deltas - so strip the marker here too, not only below.
+    if existing is None or _is_content_item_snapshot(incoming):
+        return _copy_without_snapshot_marker(incoming)
+
+    # An empty list has no item to fold a delta into (and nothing to add from one),
+    # so hand back whichever side actually has content before indexing into it below.
+    if not existing:
         return deepcopy(incoming)
+    if not incoming:
+        return existing
 
     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
@@ -2585,7 +2628,13 @@ def _coalesce_code_interpreter_content(contents: list[Content]) -> None:
 
         existing = seen.get(key)
         if existing is None:
+            # The call_id's first chunk can itself be a tagged snapshot (a call that
+            # never streams incremental deltas, just one done-style chunk), so this
+            # copy needs the same stripping `_merge_code_interpreter_content` does for
+            # later chunks.
             copied = deepcopy(content)
+            _strip_snapshot_marker(copied.inputs)
+            _strip_snapshot_marker(copied.outputs)
             seen[key] = copied
             coalesced_contents.append(copied)
             continue

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

@@ -47,6 +47,7 @@
     is_local_history_conversation_id,
 )
 from agent_framework._telemetry import FeatureIndex
+from agent_framework._types import _CONTENT_ITEM_SNAPSHOT_KEY
 from agent_framework.exceptions import MiddlewareException
 
 from .test_filesystem import COLLIDING_IDENTIFIERS
@@ -673,7 +674,12 @@ 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")],
+                        inputs=[
+                            Content.from_text(
+                                text="import pandas as pd",
+                                additional_properties={_CONTENT_ITEM_SNAPSHOT_KEY: True},
+                            )
+                        ],
                         additional_properties={"sequence_number": 3},
                     )
                 ],

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

@@ -42,6 +42,7 @@
     GROUP_TOKEN_COUNT_KEY,
 )
 from agent_framework._types import (
+    _CONTENT_ITEM_SNAPSHOT_KEY,
     _MODEL_OUTPUT_KIND_KEY,
     _append_instructions,
     _get_data_bytes,
@@ -2246,6 +2247,153 @@ def test_function_call_tagged_chunk_does_not_absorb_into_untagged_trailing_call(
     assert fcs[1].arguments == "{}"
 
 
+# region CodeInterpreter streaming merge
+
+
+def test_code_interpreter_deltas_concatenate_even_when_one_starts_with_the_other():
+    """Two genuine deltas must concatenate even if one happens to start with the other.
+
+    The merge previously guessed "is this a full resend?" by checking whether one
+    chunk's text starts with the other's. That guess is wrong whenever two ordinary,
+    unrelated deltas simply happen to line up that way - e.g. a delta of "(" followed
+    by a delta of "()" is two characters streamed in order, not the second one
+    re-sending everything. Before the fix this dropped the first delta entirely and
+    left only "()".
+    """
+    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)
+    calls = [c for c in resp.messages[0].contents if c.type == "code_interpreter_tool_call"]
+    assert len(calls) == 1
+    assert calls[0].inputs is not None
+    assert "".join(item.text or "" for item in calls[0].inputs) == "(()"
+
+
+def test_code_interpreter_deltas_concatenate_in_reverse_prefix_order_too():
+    """Same as above with the longer delta arriving first."""
+    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)
+    calls = [c for c in resp.messages[0].contents if c.type == "code_interpreter_tool_call"]
+    assert len(calls) == 1
+    assert calls[0].inputs is not None
+    assert "".join(item.text or "" for item in calls[0].inputs) == "()("
+
+
+def test_code_interpreter_snapshot_chunk_replaces_accumulated_deltas():
+    """A chunk explicitly tagged as a full resend replaces the accumulated deltas.
+
+    Mirrors the OpenAI Responses API, whose `code_interpreter_call_code.done` event
+    repeats the complete code generated so far after a run of `.delta` events for the
+    same call; the client tags that one chunk so the merge can tell it apart from an
+    ordinary delta instead of guessing from the text.
+    """
+    updates = [
+        ChatResponseUpdate(
+            contents=[
+                Content.from_code_interpreter_tool_call(call_id="ci_1", inputs=[Content.from_text(text="import")])
+            ]
+        ),
+        ChatResponseUpdate(
+            contents=[
+                Content.from_code_interpreter_tool_call(call_id="ci_1", inputs=[Content.from_text(text=" pandas")])
+            ]
+        ),
+        ChatResponseUpdate(
+            contents=[
+                Content.from_code_interpreter_tool_call(
+                    call_id="ci_1",
+                    inputs=[
+                        Content.from_text(
+                            text="import pandas as pd",
+                            additional_properties={_CONTENT_ITEM_SNAPSHOT_KEY: True},
+                        )
+                    ],
+                )
+            ]
+        ),
+    ]
+
+    resp = ChatResponse.from_updates(updates)
+    calls = [c for c in resp.messages[0].contents if c.type == "code_interpreter_tool_call"]
+    assert len(calls) == 1
+    assert calls[0].inputs is not None
+    assert "".join(item.text or "" for item in calls[0].inputs) == "import pandas as pd"
+    # The marker is a merge-time instruction only; it must not survive into the
+    # finalized content, where it would get serialized into history or seen by middleware.
+    assert _CONTENT_ITEM_SNAPSHOT_KEY not in calls[0].inputs[0].additional_properties
+
+
+def test_code_interpreter_snapshot_marker_stripped_as_the_first_chunk():
+    """The marker must not leak even when the snapshot is the only chunk for a call.
+
+    The first delta for a call takes a different code path (`existing` is still `None`,
+    so there's nothing to fold onto yet) than a later one, and that path has to strip the
+    marker too - a call that never streams incremental deltas at all, just one tagged
+    chunk, is exactly this case.
+    """
+    updates = [
+        ChatResponseUpdate(
+            contents=[
+                Content.from_code_interpreter_tool_call(
+                    call_id="ci_1",
+                    inputs=[
+                        Content.from_text(
+                            text="print('hi')",
+                            additional_properties={_CONTENT_ITEM_SNAPSHOT_KEY: True},
+                        )
+                    ],
+                )
+            ]
+        ),
+    ]
+
+    resp = ChatResponse.from_updates(updates)
+    calls = [c for c in resp.messages[0].contents if c.type == "code_interpreter_tool_call"]
+    assert len(calls) == 1
+    assert calls[0].inputs is not None
+    assert "".join(item.text or "" for item in calls[0].inputs) == "print('hi')"
+    assert _CONTENT_ITEM_SNAPSHOT_KEY not in calls[0].inputs[0].additional_properties
+
+
+def test_code_interpreter_delta_merges_onto_an_empty_placeholder():
+    """A placeholder chunk with an empty `inputs`/`outputs` list must not blow up the merge.
+
+    A provider can open a call with an empty list before any text streams in (an
+    `outputs=[]` placeholder is exactly what the OpenAI client sends when a call
+    starts). `existing` is then `[]`, not `None`, so the merge has to hand the
+    first real delta back rather than index into the empty list.
+    """
+    updates = [
+        ChatResponseUpdate(contents=[Content.from_code_interpreter_tool_call(call_id="ci_1", inputs=[])]),
+        ChatResponseUpdate(
+            contents=[
+                Content.from_code_interpreter_tool_call(call_id="ci_1", inputs=[Content.from_text(text="import os")])
+            ]
+        ),
+    ]
+
+    resp = ChatResponse.from_updates(updates)
+    calls = [c for c in resp.messages[0].contents if c.type == "code_interpreter_tool_call"]
+    assert len(calls) == 1
+    assert calls[0].inputs is not None
+    assert "".join(item.text or "" for item in calls[0].inputs) == "import os"
+
+
 # region Role & FinishReason basics