From 7af72a212626a450d5bea8ad02c5c5e5ceb9cae2 Mon Sep 17 00:00:00 2001 From: cr-sbarbouche Date: Thu, 1 Oct 2026 21:49:39 +0100 Subject: [PATCH 1/6] Python: stop guessing full-resend code interpreter chunks from their text _merge_content_item_lists decided whether to concatenate two streamed code_interpreter chunks or replace the accumulated text with a new one by checking whether one chunk's text is a string-prefix of the other. That guess exists to handle providers that resend the complete code once streaming finishes (the OpenAI Responses API's code_interpreter_call_code.done event repeats the full code after a run of .delta events for the same call), but it ran on every merge, including between two ordinary deltas. Whenever two genuine deltas happened to line up as a prefix of each other, one got silently discarded. This is not an edge case for code: repeated indentation, a closing paren right after an opening one, a short number followed by a longer one starting with the same digit, all collide with this heuristic. Nothing raised, the result was just quietly wrong. Replace the guess with an explicit marker. A chunk is only treated as a full resend when its additional_properties carries the new _CONTENT_ITEM_SNAPSHOT_KEY; everything else concatenates. The OpenAI client now sets that marker on the done event's content instead of leaving the merge layer to infer it from text. Fixes #8956 --- .../packages/core/agent_framework/_types.py | 27 ++++-- .../packages/core/tests/core/test_sessions.py | 8 +- python/packages/core/tests/core/test_types.py | 89 +++++++++++++++++++ .../agent_framework_openai/_chat_client.py | 7 +- .../tests/openai/test_openai_chat_client.py | 5 ++ 5 files changed, 129 insertions(+), 7 deletions(-) diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index 4722699b7fa..4ad1b3a81ed 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -424,6 +424,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 characters streamed in order, not a +# two-character-value resend - so providers must set this explicitly on a full resend. +_CONTENT_ITEM_SNAPSHOT_KEY = "content_item_snapshot" # endregion @@ -2330,6 +2339,16 @@ 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 _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: @@ -2337,14 +2356,12 @@ def _merge_content_item_lists(existing: Any, incoming: Any) -> Any: if existing is None: return deepcopy(incoming) + if _is_content_item_snapshot(incoming): + 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 diff --git a/python/packages/core/tests/core/test_sessions.py b/python/packages/core/tests/core/test_sessions.py index 1cf5d5adcfc..666aecc0a39 100644 --- a/python/packages/core/tests/core/test_sessions.py +++ b/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}, ) ], diff --git a/python/packages/core/tests/core/test_types.py b/python/packages/core/tests/core/test_types.py index ff4bfedf014..19dae564185 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -40,6 +40,7 @@ GROUP_TOKEN_COUNT_KEY, ) from agent_framework._types import ( + _CONTENT_ITEM_SNAPSHOT_KEY, _append_instructions, _get_data_bytes, _get_data_bytes_as_str, @@ -2157,6 +2158,94 @@ 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" + + # region Role & FinishReason basics diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py index d7b7e0735cb..dce71e03228 100644 --- a/python/packages/openai/agent_framework_openai/_chat_client.py +++ b/python/packages/openai/agent_framework_openai/_chat_client.py @@ -58,6 +58,7 @@ tool, ) from agent_framework._types import ( + _CONTENT_ITEM_SNAPSHOT_KEY, # pyright: ignore[reportPrivateUsage] Annotation, ChatOptions, ChatResponse, @@ -3728,7 +3729,11 @@ def output_text_properties(output: Any) -> dict[str, Any] | None: Content.from_text( text=event.code, raw_representation=event, - additional_properties=ci_additional_properties, + # The done event repeats the complete code generated so far + # rather than a further delta, so it must replace (not be + # concatenated onto) any `.delta` events already accumulated + # for this call. + additional_properties={**ci_additional_properties, _CONTENT_ITEM_SNAPSHOT_KEY: True}, ) ], raw_representation=event, diff --git a/python/packages/openai/tests/openai/test_openai_chat_client.py b/python/packages/openai/tests/openai/test_openai_chat_client.py index fcb31c961b3..9e40d673bde 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_client.py +++ b/python/packages/openai/tests/openai/test_openai_chat_client.py @@ -38,6 +38,7 @@ SessionContext, _filter_approval_control_messages, ) +from agent_framework._types import _CONTENT_ITEM_SNAPSHOT_KEY from agent_framework._workflows._checkpoint_encoding import decode_checkpoint_value, encode_checkpoint_value from agent_framework.exceptions import ( ChatClientException, @@ -7712,6 +7713,10 @@ def test_parse_chunk_from_openai_code_interpreter_done() -> None: 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" + # The done event repeats the full code rather than a further delta, so the merge + # layer needs it tagged as a snapshot to replace (not concatenate onto) accumulated + # `.delta` chunks for the same call. + assert result.contents[0].inputs[0].additional_properties[_CONTENT_ITEM_SNAPSHOT_KEY] is True def test_parse_chunk_from_openai_reasoning() -> None: From 2e60667d9cd146c97e4dff9a1fab50226ca1fdb1 Mon Sep 17 00:00:00 2001 From: cr-sbarbouche Date: Thu, 1 Oct 2026 22:04:20 +0100 Subject: [PATCH 2/6] Python: address review feedback on the snapshot-marker fix Drop the openai package's cross-package import of the new core marker constant and redeclare it locally instead, matching how _MODEL_OUTPUT_KIND_KEY is already duplicated in that file. agent-framework-openai's pyproject.toml floors agent-framework-core at the current release, which doesn't carry the new symbol yet, so importing it directly would break that pinned combination. Guard _merge_content_item_lists against an empty existing or incoming list instead of indexing into it unconditionally. A code interpreter call can open with an empty inputs/outputs placeholder before any text streams in, and the removed prefix checks used to short-circuit that case by accident. Fix a comment that miscounted characters vs. chunks in the snapshot-marker example. --- .../packages/core/agent_framework/_types.py | 11 +++++++-- python/packages/core/tests/core/test_types.py | 24 +++++++++++++++++++ .../agent_framework_openai/_chat_client.py | 2 +- .../tests/openai/test_openai_chat_client.py | 3 +-- 4 files changed, 35 insertions(+), 5 deletions(-) diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index 4ad1b3a81ed..618014d2e4b 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -430,8 +430,8 @@ class ComputerSafetyCheck(TypedDict, total=False): # 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 characters streamed in order, not a -# two-character-value resend - so providers must set this explicitly on a full resend. +# 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 @@ -2359,6 +2359,13 @@ def _merge_content_item_lists(existing: Any, incoming: Any) -> Any: if _is_content_item_snapshot(incoming): return deepcopy(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: diff --git a/python/packages/core/tests/core/test_types.py b/python/packages/core/tests/core/test_types.py index 19dae564185..7edddd9676c 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -2246,6 +2246,30 @@ def test_code_interpreter_snapshot_chunk_replaces_accumulated_deltas(): assert "".join(item.text or "" for item in calls[0].inputs) == "import pandas as pd" +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 diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py index dce71e03228..784ef14698d 100644 --- a/python/packages/openai/agent_framework_openai/_chat_client.py +++ b/python/packages/openai/agent_framework_openai/_chat_client.py @@ -58,7 +58,6 @@ tool, ) from agent_framework._types import ( - _CONTENT_ITEM_SNAPSHOT_KEY, # pyright: ignore[reportPrivateUsage] Annotation, ChatOptions, ChatResponse, @@ -154,6 +153,7 @@ class _PromptCacheOptions(TypedDict, total=False): _MODEL_OUTPUT_KIND_KEY = "model_output_kind" _MODEL_OUTPUT_REFUSAL = "refusal" +_CONTENT_ITEM_SNAPSHOT_KEY = "content_item_snapshot" def _is_refusal_text_content(content: Content) -> bool: diff --git a/python/packages/openai/tests/openai/test_openai_chat_client.py b/python/packages/openai/tests/openai/test_openai_chat_client.py index 9e40d673bde..15ffbeb9da6 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_client.py +++ b/python/packages/openai/tests/openai/test_openai_chat_client.py @@ -38,7 +38,6 @@ SessionContext, _filter_approval_control_messages, ) -from agent_framework._types import _CONTENT_ITEM_SNAPSHOT_KEY from agent_framework._workflows._checkpoint_encoding import decode_checkpoint_value, encode_checkpoint_value from agent_framework.exceptions import ( ChatClientException, @@ -70,7 +69,7 @@ from pytest import param from agent_framework_openai import OpenAIChatClient, OpenAIChatOptions, RawOpenAIChatClient -from agent_framework_openai._chat_client import OPENAI_LOCAL_SHELL_CALL_ITEM_ID_KEY +from agent_framework_openai._chat_client import _CONTENT_ITEM_SNAPSHOT_KEY, OPENAI_LOCAL_SHELL_CALL_ITEM_ID_KEY from agent_framework_openai._exceptions import OpenAIContentFilterException _OPENAI_HTTPX = cast(Any, import_module(DefaultAsyncHttpxClient.__mro__[1].__module__.partition(".")[0])) From dcc31f204a0c16eccbb3e2a8598ec8039c159c2a Mon Sep 17 00:00:00 2001 From: cr-sbarbouche Date: Fri, 2 Oct 2026 06:08:23 +0100 Subject: [PATCH 3/6] Python: strip the snapshot marker and add a back-compat fallback Stop leaking the merge-only snapshot marker into finalized content. _merge_content_item_lists returned the tagged chunk as-is once it recognized a full resend, so _CONTENT_ITEM_SNAPSHOT_KEY stuck around in additional_properties after the merge was done, where to_dict() would persist it into history and middleware would see it. Pop the key off the copy before returning it; it has no meaning past this one merge step. Also recognize the OpenAI Responses API's code_interpreter_call_code.done event from its raw event type as a second, independent snapshot signal. agent-framework-openai's published releases stream that event without the marker, and their agent-framework-core floor allows installing this core version, so without this fallback, upgrading only core would turn a streamed delta followed by that provider's done event into duplicated code instead of a correct replace. --- .../packages/core/agent_framework/_types.py | 22 +++++++++- python/packages/core/tests/core/test_types.py | 41 +++++++++++++++++++ 2 files changed, 61 insertions(+), 2 deletions(-) diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index 618014d2e4b..120ac2182e2 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -433,6 +433,13 @@ class ComputerSafetyCheck(TypedDict, total=False): # 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" +# Recognize the OpenAI Responses API's code interpreter "done" event from its raw event +# shape as a secondary snapshot signal, alongside the marker above. Older +# agent-framework-openai releases stream this event without setting the marker, and +# their agent-framework-core floor allows installing this version of core, so without +# this fallback upgrading core alone would turn a streamed delta followed by that +# provider's done event into duplicated code instead of a correct replace. +_OPENAI_CODE_INTERPRETER_DONE_EVENT_TYPE = "response.code_interpreter_call_code.done" # endregion @@ -2344,7 +2351,11 @@ def _is_content_item_snapshot(items: Any) -> bool: if not isinstance(items, list): return False return any( - isinstance(item, Content) and item.additional_properties.get(_CONTENT_ITEM_SNAPSHOT_KEY) + isinstance(item, Content) + and ( + item.additional_properties.get(_CONTENT_ITEM_SNAPSHOT_KEY) + or getattr(item.raw_representation, "type", None) == _OPENAI_CODE_INTERPRETER_DONE_EVENT_TYPE + ) for item in cast("list[object]", items) ) @@ -2357,7 +2368,14 @@ def _merge_content_item_lists(existing: Any, incoming: Any) -> Any: return deepcopy(incoming) if _is_content_item_snapshot(incoming): - return deepcopy(incoming) + snapshot = deepcopy(incoming) + # The marker is only an instruction for this merge step; strip it so it doesn't + # persist into the finalized content that gets serialized into history or passed + # to middleware. + for item in cast("list[object]", snapshot): + if isinstance(item, Content): + item.additional_properties.pop(_CONTENT_ITEM_SNAPSHOT_KEY, None) + return snapshot # 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. diff --git a/python/packages/core/tests/core/test_types.py b/python/packages/core/tests/core/test_types.py index 7edddd9676c..917c00a3c03 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -8,6 +8,7 @@ from dataclasses import dataclass from datetime import datetime, timezone from functools import partial +from types import SimpleNamespace from typing import Any, Literal, cast import pytest @@ -2244,6 +2245,46 @@ def test_code_interpreter_snapshot_chunk_replaces_accumulated_deltas(): 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_done_event_recognized_without_explicit_marker(): + """A done-shaped chunk still replaces accumulated deltas even without the marker. + + Older `agent-framework-openai` releases stream the done event without tagging it, + and their `agent-framework-core` floor allows installing this version of core, so + the merge also recognizes the done event by its raw event `type` as a fallback - + otherwise upgrading core alone would turn this into duplicated code. + """ + done_event = SimpleNamespace(type="response.code_interpreter_call_code.done") + 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", raw_representation=done_event)], + ) + ] + ), + ] + + 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" def test_code_interpreter_delta_merges_onto_an_empty_placeholder(): From b2e7e4b550a2611715d7ac55210b386888db6fba Mon Sep 17 00:00:00 2001 From: cr-sbarbouche Date: Wed, 7 Oct 2026 14:10:24 +0100 Subject: [PATCH 4/6] Python: strip the snapshot marker on the first chunk too The previous stripping only ran inside _merge_content_item_lists, which never runs for a call_id's first occurrence - _coalesce_code_interpreter_content deep-copies that one directly and stores it in seen without going through any merge function. A call that streams a single tagged snapshot and no prior deltas hit exactly that path, so the marker still reached the finalized content. Factored the stripping into a shared _strip_snapshot_marker helper used by both the first-occurrence copy in _coalesce_code_interpreter_content and the merge path in _merge_content_item_lists, including its own existing-is-None early return, which had the same gap for a field (inputs or outputs) streaming its first chunk as a snapshot. --- .../packages/core/agent_framework/_types.py | 42 +++++++++++++------ python/packages/core/tests/core/test_types.py | 32 ++++++++++++++ 2 files changed, 62 insertions(+), 12 deletions(-) diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index 120ac2182e2..433711c64f6 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -2360,22 +2360,34 @@ def _is_content_item_snapshot(items: Any) -> bool: ) +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: - return deepcopy(incoming) - - if _is_content_item_snapshot(incoming): - snapshot = deepcopy(incoming) - # The marker is only an instruction for this merge step; strip it so it doesn't - # persist into the finalized content that gets serialized into history or passed - # to middleware. - for item in cast("list[object]", snapshot): - if isinstance(item, Content): - item.additional_properties.pop(_CONTENT_ITEM_SNAPSHOT_KEY, None) - return snapshot + # 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. @@ -2433,7 +2445,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 diff --git a/python/packages/core/tests/core/test_types.py b/python/packages/core/tests/core/test_types.py index 917c00a3c03..79181cea8d7 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -2250,6 +2250,38 @@ def test_code_interpreter_snapshot_chunk_replaces_accumulated_deltas(): 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_done_event_recognized_without_explicit_marker(): """A done-shaped chunk still replaces accumulated deltas even without the marker. From e1fcfe7d9d6aef410b25e2a0b3db071433c40d72 Mon Sep 17 00:00:00 2001 From: cr-sbarbouche Date: Thu, 8 Oct 2026 10:31:53 +0100 Subject: [PATCH 5/6] Python: drop the OpenAI-specific snapshot fallback from core _is_content_item_snapshot recognized a code_interpreter_call_code.done event by its raw event type string as well as the explicit marker, to keep working with agent-framework-openai releases published before this fix. That put OpenAI wire-protocol knowledge directly in the provider-agnostic merge code in _types.py, which core should not carry. Back to checking only the explicit _CONTENT_ITEM_SNAPSHOT_KEY marker. A pre-fix agent-framework-openai paired with this core no longer gets the done event recognized as a snapshot, so that specific old-client/new-core combination can duplicate code again; providers that set the marker (including the current agent-framework-openai, already updated in this PR) are unaffected. --- .../packages/core/agent_framework/_types.py | 13 +------ python/packages/core/tests/core/test_types.py | 38 ------------------- 2 files changed, 1 insertion(+), 50 deletions(-) diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index 88503ddc533..c8749ecab59 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -433,13 +433,6 @@ class ComputerSafetyCheck(TypedDict, total=False): # 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" -# Recognize the OpenAI Responses API's code interpreter "done" event from its raw event -# shape as a secondary snapshot signal, alongside the marker above. Older -# agent-framework-openai releases stream this event without setting the marker, and -# their agent-framework-core floor allows installing this version of core, so without -# this fallback upgrading core alone would turn a streamed delta followed by that -# provider's done event into duplicated code instead of a correct replace. -_OPENAI_CODE_INTERPRETER_DONE_EVENT_TYPE = "response.code_interpreter_call_code.done" # endregion @@ -2371,11 +2364,7 @@ def _is_content_item_snapshot(items: Any) -> bool: if not isinstance(items, list): return False return any( - isinstance(item, Content) - and ( - item.additional_properties.get(_CONTENT_ITEM_SNAPSHOT_KEY) - or getattr(item.raw_representation, "type", None) == _OPENAI_CODE_INTERPRETER_DONE_EVENT_TYPE - ) + isinstance(item, Content) and item.additional_properties.get(_CONTENT_ITEM_SNAPSHOT_KEY) for item in cast("list[object]", items) ) diff --git a/python/packages/core/tests/core/test_types.py b/python/packages/core/tests/core/test_types.py index c8000266280..6be05dd32f9 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -9,7 +9,6 @@ from dataclasses import dataclass from datetime import datetime, timezone from functools import partial -from types import SimpleNamespace from typing import Any, Literal, cast import pytest @@ -2367,43 +2366,6 @@ def test_code_interpreter_snapshot_marker_stripped_as_the_first_chunk(): assert _CONTENT_ITEM_SNAPSHOT_KEY not in calls[0].inputs[0].additional_properties -def test_code_interpreter_done_event_recognized_without_explicit_marker(): - """A done-shaped chunk still replaces accumulated deltas even without the marker. - - Older `agent-framework-openai` releases stream the done event without tagging it, - and their `agent-framework-core` floor allows installing this version of core, so - the merge also recognizes the done event by its raw event `type` as a fallback - - otherwise upgrading core alone would turn this into duplicated code. - """ - done_event = SimpleNamespace(type="response.code_interpreter_call_code.done") - 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", raw_representation=done_event)], - ) - ] - ), - ] - - 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" - - 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. From 9c00bbb6329c909c013b8998894b8ba3014993ee Mon Sep 17 00:00:00 2001 From: cr-sbarbouche Date: Fri, 9 Oct 2026 11:37:53 +0100 Subject: [PATCH 6/6] Python: tag the completed code interpreter item as a snapshot too response.output_item.done never had a code_interpreter_call branch, unlike every sibling item type (shell_call, computer_call, image_generation_call, mcp_call, web_search_call) that gets its full state captured there. There is no incremental event for a code interpreter call's outputs (logs/images), so this meant streamed code interpreter results were silently always empty - outputs only ever got set from response.output_item.added, which fires before the call has run. Add that branch, reusing the same item-to-contents conversion already used by the non-streaming parser and response.output_item.added (now extracted into one shared helper instead of three near-identical copies). The done item's code is tagged as a snapshot like code_interpreter_call_code.done's is, since a normal stream sends both for the same call and the merge needs to recognize the second one as a repeat of the authoritative value rather than concatenate onto the first. --- .../agent_framework_openai/_chat_client.py | 138 +++++++++--------- .../tests/openai/test_openai_chat_client.py | 111 ++++++++++++++ 2 files changed, 177 insertions(+), 72 deletions(-) diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py index 25ddb7f623b..8cdab6328b6 100644 --- a/python/packages/openai/agent_framework_openai/_chat_client.py +++ b/python/packages/openai/agent_framework_openai/_chat_client.py @@ -2513,6 +2513,59 @@ def _image_generation_item_to_contents(self, item: Any) -> list[Content]: ), ] + def _code_interpreter_item_to_contents(self, item: Any, *, mark_code_as_snapshot: bool) -> list[Content]: + """Convert a ``code_interpreter_call`` output item into framework ``Content`` objects. + + Used by the non-streaming parser and by both the streaming ``response.output_item.added`` + and ``response.output_item.done`` handlers. ``outputs`` (logs/images) are only ever + populated on the completed item - at ``added`` the call has not run yet - so there is + nothing to mark there; callers pass ``mark_code_as_snapshot=True`` only for a completed + item's code, so the streaming merge layer replaces (rather than concatenates onto) any + `.delta`/`.done` code chunks already accumulated for this call instead of duplicating them. + """ + call_id = getattr(item, "call_id", None) or getattr(item, "id", None) + outputs: list[Content] = [] + if item_outputs := getattr(item, "outputs", None): + for code_output in item_outputs: + if getattr(code_output, "type", None) == "logs": + outputs.append( + Content.from_text( + text=code_output.logs, + raw_representation=code_output, + ) + ) + elif getattr(code_output, "type", None) == "image": + outputs.append( + Content.from_uri( + uri=code_output.url, + raw_representation=code_output, + media_type="image", + ) + ) + contents: list[Content] = [] + if code := getattr(item, "code", None): + contents.append( + Content.from_code_interpreter_tool_call( + call_id=call_id, + inputs=[ + Content.from_text( + text=code, + raw_representation=item, + additional_properties={_CONTENT_ITEM_SNAPSHOT_KEY: True} if mark_code_as_snapshot else None, + ) + ], + raw_representation=item, + ) + ) + contents.append( + Content.from_code_interpreter_tool_result( + call_id=call_id, + outputs=outputs, + raw_representation=item, + ) + ) + return contents + def _shell_item_to_contents(self, item: Any, local_shell_tool_name: str | None) -> list[Content]: """Convert a shell output item into framework ``Content`` objects. @@ -3434,40 +3487,9 @@ def _parse_response_from_openai( ) ) case "code_interpreter_call": # ResponseOutputCodeInterpreterCall - call_id = getattr(item, "call_id", None) or getattr(item, "id", None) - outputs: list[Content] = [] - if item_outputs := getattr(item, "outputs", None): - for code_output in item_outputs: - if getattr(code_output, "type", None) == "logs": - outputs.append( - Content.from_text( - text=code_output.logs, - raw_representation=code_output, - ) - ) - elif getattr(code_output, "type", None) == "image": - outputs.append( - Content.from_uri( - uri=code_output.url, - raw_representation=code_output, - media_type="image", - ) - ) - if code := getattr(item, "code", None): - contents.append( - Content.from_code_interpreter_tool_call( - call_id=call_id, - inputs=[Content.from_text(text=code, raw_representation=item)], - raw_representation=item, - ) - ) - contents.append( - Content.from_code_interpreter_tool_result( - call_id=call_id, - outputs=outputs, - raw_representation=item, - ) - ) + # Non-streaming: the whole item is already complete, so there is nothing to + # merge against and no need to tag the code as a snapshot. + contents.extend(self._code_interpreter_item_to_contents(item, mark_code_as_snapshot=False)) case "function_call": # ResponseOutputFunctionCall contents.append( Content.from_function_call( @@ -3956,44 +3978,10 @@ def parse_annotations( ) # Result deferred to response.output_item.done case "code_interpreter_call": # ResponseOutputCodeInterpreterCall - call_id = getattr(event_item, "call_id", None) or getattr(event_item, "id", None) - outputs: list[Content] = [] - if hasattr(event_item, "outputs") and event_item.outputs: - for code_output in event_item.outputs: - if getattr(code_output, "type", None) == "logs": - outputs.append( - Content.from_text( - text=cast(Any, code_output).logs, - raw_representation=code_output, - ) - ) - elif getattr(code_output, "type", None) == "image": - outputs.append( - Content.from_uri( - uri=cast(Any, code_output).url, - raw_representation=code_output, - media_type="image", - ) - ) - if hasattr(event_item, "code") and event_item.code: - contents.append( - Content.from_code_interpreter_tool_call( - call_id=call_id, - inputs=[ - Content.from_text( - text=event_item.code, - raw_representation=event_item, - ) - ], - raw_representation=event_item, - ) - ) - contents.append( - Content.from_code_interpreter_tool_result( - call_id=call_id, - outputs=outputs, - raw_representation=event_item, - ) + # The call has not run yet at `added` time, so any code seen here is an + # opening fragment deltas build on, not a complete value to snapshot. + contents.extend( + self._code_interpreter_item_to_contents(event_item, mark_code_as_snapshot=False) ) case ( "shell_call" @@ -4191,6 +4179,12 @@ def parse_annotations( seen_function_call_output_ids, done_item ): contents.append(self._parse_function_call_output_content(output_item, done_call_id)) + elif getattr(done_item, "type", None) == "code_interpreter_call": + # The completed item is the only place `outputs` (logs/images) are ever + # delivered during streaming - there is no incremental outputs event - and + # its `code` is the authoritative final value, so it is tagged as a snapshot + # even though `.code_interpreter_call_code.done` (above) usually already set it. + contents.extend(self._code_interpreter_item_to_contents(done_item, mark_code_as_snapshot=True)) elif getattr(done_item, "type", None) == _AZURE_AI_SEARCH_CALL_OUTPUT_TYPE: pass case _: diff --git a/python/packages/openai/tests/openai/test_openai_chat_client.py b/python/packages/openai/tests/openai/test_openai_chat_client.py index 40670ca5f83..2222d79bb1a 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_client.py +++ b/python/packages/openai/tests/openai/test_openai_chat_client.py @@ -8215,6 +8215,117 @@ def test_parse_chunk_from_openai_code_interpreter_done() -> None: assert result.contents[0].inputs[0].additional_properties[_CONTENT_ITEM_SNAPSHOT_KEY] is True +def test_parse_chunk_from_openai_code_interpreter_output_item_done() -> None: + """Test _parse_chunk_from_openai with a completed code_interpreter_call output item. + + `response.output_item.done` is the only place `outputs` (logs/images) are ever delivered + for a streamed code interpreter call - there is no incremental outputs event - so this + completed item must be parsed, and its `code` must be tagged as a snapshot like + `code_interpreter_call_code.done` is, since both can carry the same final code. + """ + client = OpenAIChatClient(model="test-model", api_key="test-key") + chat_options: dict[str, Any] = {} + function_call_ids: dict[int, tuple[str, str]] = {} + + mock_log_output = MagicMock() + mock_log_output.type = "logs" + mock_log_output.logs = "hello" + + mock_done_event = MagicMock() + mock_done_event.type = "response.output_item.done" + mock_done_item = MagicMock() + mock_done_item.type = "code_interpreter_call" + mock_done_item.call_id = "ci_789" + mock_done_item.id = "ci_789" + mock_done_item.code = "print('hello')" + mock_done_item.outputs = [mock_log_output] + mock_done_event.item = mock_done_item + + result = client._parse_chunk_from_openai(mock_done_event, chat_options, function_call_ids) + call_contents = [c for c in result.contents if c.type == "code_interpreter_tool_call"] + result_contents = [c for c in result.contents if c.type == "code_interpreter_tool_result"] + assert len(call_contents) == 1 + assert call_contents[0].call_id == "ci_789" + assert call_contents[0].inputs + assert call_contents[0].inputs[0].text == "print('hello')" + assert call_contents[0].inputs[0].additional_properties[_CONTENT_ITEM_SNAPSHOT_KEY] is True + assert len(result_contents) == 1 + assert result_contents[0].call_id == "ci_789" + assert result_contents[0].outputs + assert any(out.type == "text" and out.text == "hello" for out in result_contents[0].outputs) + + +def test_code_interpreter_streaming_sequence_does_not_duplicate_code_and_captures_outputs() -> None: + """A realistic added/delta/done/output_item.done sequence must not duplicate code. + + `code_interpreter_call_code.done` and `response.output_item.done` can both carry the + same final code for one call; without both being tagged as snapshots, the second one + to arrive would concatenate onto the first instead of being recognized as a repeat of + the same authoritative value. + """ + client = OpenAIChatClient(model="test-model", api_key="test-key") + chat_options: dict[str, Any] = {} + function_call_ids: dict[int, tuple[str, str]] = {} + updates = [] + + added_event = MagicMock() + added_event.type = "response.output_item.added" + added_item = MagicMock() + added_item.type = "code_interpreter_call" + added_item.call_id = "ci_1" + added_item.id = "ci_1" + added_item.code = None + added_item.outputs = None + added_event.item = added_item + updates.append(client._parse_chunk_from_openai(added_event, chat_options, function_call_ids)) + + for index, delta in enumerate(["import pandas", " as pd"]): + delta_event = MagicMock() + delta_event.type = "response.code_interpreter_call_code.delta" + delta_event.item_id = "ci_1" + delta_event.delta = delta + delta_event.output_index = 0 + delta_event.sequence_number = index + 1 + delta_event.call_id = None + delta_event.id = None + updates.append(client._parse_chunk_from_openai(delta_event, chat_options, function_call_ids)) + + done_code_event = MagicMock() + done_code_event.type = "response.code_interpreter_call_code.done" + done_code_event.item_id = "ci_1" + done_code_event.code = "import pandas as pd" + done_code_event.output_index = 0 + done_code_event.sequence_number = 3 + done_code_event.call_id = None + done_code_event.id = None + updates.append(client._parse_chunk_from_openai(done_code_event, chat_options, function_call_ids)) + + mock_log_output = MagicMock() + mock_log_output.type = "logs" + mock_log_output.logs = "pandas 2.1.0" + done_item_event = MagicMock() + done_item_event.type = "response.output_item.done" + done_item = MagicMock() + done_item.type = "code_interpreter_call" + done_item.call_id = "ci_1" + done_item.id = "ci_1" + done_item.code = "import pandas as pd" + done_item.outputs = [mock_log_output] + done_item_event.item = done_item + updates.append(client._parse_chunk_from_openai(done_item_event, chat_options, function_call_ids)) + + response = ChatResponse.from_updates(updates) + calls = [c for c in response.messages[0].contents if c.type == "code_interpreter_tool_call"] + results = [c for c in response.messages[0].contents if c.type == "code_interpreter_tool_result"] + assert len(calls) == 1 + assert calls[0].inputs + assert "".join(item.text or "" for item in calls[0].inputs) == "import pandas as pd" + assert _CONTENT_ITEM_SNAPSHOT_KEY not in calls[0].inputs[0].additional_properties + assert len(results) == 1 + assert results[0].outputs + assert any(out.type == "text" and out.text == "pandas 2.1.0" for out in results[0].outputs) + + def test_parse_chunk_from_openai_reasoning() -> None: """Test _parse_chunk_from_openai with reasoning content.""" client = OpenAIChatClient(model="test-model", api_key="test-key")