diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index f98743794a4..a3247c10b35 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/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 diff --git a/python/packages/core/tests/core/test_sessions.py b/python/packages/core/tests/core/test_sessions.py index 1cf5d5adcfc..e23d560e284 100644 --- a/python/packages/core/tests/core/test_sessions.py +++ b/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}, ) ], diff --git a/python/packages/core/tests/core/test_types.py b/python/packages/core/tests/core/test_types.py index 05d474687f4..f712f12de33 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/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", 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: