From a6e3f0a95152a4de7225df2b332e527c694337eb Mon Sep 17 00:00:00 2001 From: minelhi <3417378192@qq.com> Date: Wed, 7 Oct 2026 18:15:40 +0800 Subject: [PATCH 1/3] fix(python): concatenate code-interpreter stream deltas with prefix overlap (#8955) Stop treating string-prefix relationships as a full resend. Only replace accumulated text when the incoming payload is a provider *.done / complete value; ordinary deltas always concatenate (also covers #8903). Co-authored-by: Cursor --- .../packages/core/agent_framework/_types.py | 37 ++++++++++-- python/packages/core/tests/core/test_types.py | 59 +++++++++++++++++++ 2 files changed, 90 insertions(+), 6 deletions(-) diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index f98743794a4..fc92779cdb0 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -2350,21 +2350,46 @@ 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 carry an authoritative full value (e.g. a ``*.done`` stream event). + + OpenAI Responses emits ``code_interpreter_call_code.done`` with the full code after + incremental ``.delta`` events. Only that kind of payload should replace prior text; + ordinary deltas must always concatenate even when one happens to be 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 + raw = item.raw_representation + raw_type = getattr(raw, "type", None) if raw is not None else None + if isinstance(raw_type, str) and raw_type.endswith(".done"): + 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 + (provider ``*.done`` event or ``stream_complete`` flag) 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) + 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_types.py b/python/packages/core/tests/core/test_types.py index 05d474687f4..e1955c934f6 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -421,6 +421,65 @@ 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 provider *.done payload is authoritative and replaces prior delta text.""" + + class _DoneEvent: + 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 os\n")]) + ] + ), + ChatResponseUpdate( + contents=[ + Content.from_code_interpreter_tool_call( + call_id="ci_1", + inputs=[ + Content.from_text( + text="import os\nimport sys", + raw_representation=_DoneEvent(), + ) + ], + ) + ] + ), + ] + 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_content_outputs(): result = Content.from_code_interpreter_tool_result( call_id="call-2", From 772ad82fa51f3986f3fb7970ed7681e2c82d0c4e Mon Sep 17 00:00:00 2001 From: minelhi <3417378192@qq.com> Date: Wed, 7 Oct 2026 18:25:16 +0800 Subject: [PATCH 2/3] fix(python): address #8955 review -- empty list and done marker Handle empty accumulated outputs without IndexError, and mark the session-history full-code snapshot as stream_complete so it replaces deltas instead of concatenating. Co-authored-by: Cursor --- .../packages/core/agent_framework/_types.py | 4 ++++ .../packages/core/tests/core/test_sessions.py | 9 ++++++++- python/packages/core/tests/core/test_types.py | 20 +++++++++++++++++++ 3 files changed, 32 insertions(+), 1 deletion(-) diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index fc92779cdb0..5f7a4c8d716 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -2387,6 +2387,10 @@ def _merge_content_item_lists(existing: Any, incoming: Any) -> Any: 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: diff --git a/python/packages/core/tests/core/test_sessions.py b/python/packages/core/tests/core/test_sessions.py index 1cf5d5adcfc..6adbeeac25a 100644 --- a/python/packages/core/tests/core/test_sessions.py +++ b/python/packages/core/tests/core/test_sessions.py @@ -673,7 +673,14 @@ 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 mimics OpenAI code_interpreter_call_code.done + # (authoritative full value, not another incremental delta). + inputs=[ + Content.from_text( + text="import pandas as pd", + additional_properties={"stream_complete": 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 e1955c934f6..346cfd61a01 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -480,6 +480,26 @@ class _DoneEvent: 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", From 463ae9342ee75681c1377d6b123eccf0127232ee Mon Sep 17 00:00:00 2001 From: minelhi <3417378192@qq.com> Date: Fri, 9 Oct 2026 00:32:15 +0800 Subject: [PATCH 3/3] fix(python): use generic stream_complete for code-interpreter merge (#8955) Drop OpenAI *.done raw_representation checks from core. Providers mark authoritative snapshots with additional_properties stream_complete; OpenAI done events now set that flag. Co-authored-by: Cursor --- python/packages/core/agent_framework/_types.py | 18 ++++++++---------- .../packages/core/tests/core/test_sessions.py | 3 +-- python/packages/core/tests/core/test_types.py | 8 ++------ .../agent_framework_openai/_chat_client.py | 2 ++ .../tests/openai/test_openai_chat_client.py | 4 +++- 5 files changed, 16 insertions(+), 19 deletions(-) diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index 5f7a4c8d716..a3247c10b35 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -2351,12 +2351,13 @@ def _content_items_text(items: Any) -> str | None: def _content_items_are_complete_value(items: Any) -> bool: - """Return True when items carry an authoritative full value (e.g. a ``*.done`` stream event). + """Return True when items are marked as an authoritative full value. - OpenAI Responses emits ``code_interpreter_call_code.done`` with the full code after - incremental ``.delta`` events. Only that kind of payload should replace prior text; - ordinary deltas must always concatenate even when one happens to be a string prefix - of the other (see #8955 / #8903). + 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 @@ -2366,10 +2367,6 @@ def _content_items_are_complete_value(items: Any) -> bool: props = item.additional_properties or {} if props.get("stream_complete") or props.get("complete"): return True - raw = item.raw_representation - raw_type = getattr(raw, "type", None) if raw is not None else None - if isinstance(raw_type, str) and raw_type.endswith(".done"): - return True return False @@ -2377,7 +2374,8 @@ def _merge_content_item_lists(existing: Any, incoming: Any) -> Any: """Merge streamed nested content lists. Incremental text deltas are always concatenated. A later complete/full value - (provider ``*.done`` event or ``stream_complete`` flag) replaces the accumulation. + (``stream_complete`` / ``complete`` on ``additional_properties``) replaces the + accumulation. """ if incoming is None: return existing diff --git a/python/packages/core/tests/core/test_sessions.py b/python/packages/core/tests/core/test_sessions.py index 6adbeeac25a..e23d560e284 100644 --- a/python/packages/core/tests/core/test_sessions.py +++ b/python/packages/core/tests/core/test_sessions.py @@ -673,8 +673,7 @@ async def test_after_run_stores_coalesced_code_interpreter_chunks(self) -> None: contents=[ Content.from_code_interpreter_tool_call( call_id="ci_123", - # Final snapshot mimics OpenAI code_interpreter_call_code.done - # (authoritative full value, not another incremental delta). + # Final snapshot: authoritative full value via stream_complete. inputs=[ Content.from_text( text="import pandas as pd", diff --git a/python/packages/core/tests/core/test_types.py b/python/packages/core/tests/core/test_types.py index 346cfd61a01..f712f12de33 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -450,11 +450,7 @@ def test_code_interpreter_tool_call_stream_concatenates_indent_prefix_deltas() - def test_code_interpreter_tool_call_stream_done_replaces_accumulated_deltas() -> None: - """A provider *.done payload is authoritative and replaces prior delta text.""" - - class _DoneEvent: - type = "response.code_interpreter_call_code.done" - + """A stream_complete snapshot is authoritative and replaces prior delta text.""" updates = [ ChatResponseUpdate( contents=[ @@ -468,7 +464,7 @@ class _DoneEvent: inputs=[ Content.from_text( text="import os\nimport sys", - raw_representation=_DoneEvent(), + additional_properties={"stream_complete": True}, ) ], ) diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py index 150160aac62..cb9abc9b071 100644 --- a/python/packages/openai/agent_framework_openai/_chat_client.py +++ b/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( 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 bd576e61321..9753a653f9c 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_client.py +++ b/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: