From 2f3ae616972bbc4c75db59f8b5c96abee2a25f6e Mon Sep 17 00:00:00 2001 From: cr-sbarbouche Date: Sat, 12 Sep 2026 20:54:15 +0100 Subject: [PATCH 1/3] Python: fix streamed parallel tool calls merging into the wrong call _process_update only ever merged an incoming function_call chunk into message.contents[-1], so once two tool calls are in flight at the same time their interleaved argument deltas got mashed together or split into duplicate, unmergeable fragments instead of each call ending up complete. Match the merge target by call_id across all in-progress calls, falling back to the trailing item for continuation deltas that never repeat their call_id. Fixes #8336 --- .../packages/core/agent_framework/_types.py | 36 ++++++++++-- python/packages/core/tests/core/test_types.py | 58 +++++++++++++++++++ 2 files changed, 88 insertions(+), 6 deletions(-) diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index 7b0efa634e..1e23e9d086 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -2032,12 +2032,8 @@ def _process_update(response: ChatResponse | AgentResponse, update: ChatResponse logger.warning(f"Skipping unknown content type or invalid content: {exc}") continue match content_type: - # mypy doesn't narrow type based on match/case, but we know these are FunctionCallContents - case "function_call" if message.contents and message.contents[-1].type == "function_call": - try: - message.contents[-1] += content - except (AdditionItemMismatch, ContentError): - message.contents.append(content) + case "function_call": + _merge_function_call_content(message, content) case "usage": if response.usage_details is None: response.usage_details = UsageDetails() @@ -2075,6 +2071,34 @@ def _process_update(response: ChatResponse | AgentResponse, update: ChatResponse response.continuation_token = update.continuation_token +def _merge_function_call_content(message: Message, content: Content) -> None: + """Merge a streamed function_call chunk into the in-progress call it belongs to. + + Providers can stream multiple tool calls in parallel, so the next function_call + chunk is not necessarily a continuation of the most recently appended one; a chunk + with a call_id is matched against existing contents by that id first. Chunks with + no call_id (continuation deltas some providers only stamp on the first chunk) fall + back to merging with the trailing function_call item, preserving prior behavior. + """ + call_id = getattr(content, "call_id", None) + if call_id: + for index in range(len(message.contents) - 1, -1, -1): + existing = message.contents[index] + if existing.type == "function_call" and getattr(existing, "call_id", None) == call_id: + try: + message.contents[index] = existing + content + except (AdditionItemMismatch, ContentError): + break + return + if message.contents and message.contents[-1].type == "function_call": + try: + message.contents[-1] += content + return + except (AdditionItemMismatch, ContentError): + pass + message.contents.append(content) + + def _coalesce_text_content(contents: list[Content], type_str: Literal["text", "text_reasoning"]) -> None: """Take any subsequence Text or TextReasoningContent items and coalesce them into a single item.""" if not contents: diff --git a/python/packages/core/tests/core/test_types.py b/python/packages/core/tests/core/test_types.py index f7669221bd..976b04c652 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -1912,6 +1912,64 @@ def test_function_call_incompatible_ids_are_not_merged(): assert len(fcs) == 2 +def test_function_call_interleaved_parallel_streaming_merges_by_call_id(): + """Argument deltas for two parallel tool calls can interleave; each must land on its own call. + + This mirrors how the OpenAI Responses API streams parallel tool calls: every + ``response.function_call_arguments.delta`` event is tagged with the call's real + call_id (tracked per output_index), but deltas for different calls are not + guaranteed to arrive grouped together. Before this fix, only the trailing + content item was ever considered a merge target, so an out-of-turn delta for an + earlier call_id was appended as a stray duplicate instead of being folded into + its call, leaving both calls with incomplete, unparsable arguments. + """ + updates = [ + ChatResponseUpdate(contents=[Content.from_function_call(call_id="call_1", name="get_weather", arguments="")]), + ChatResponseUpdate(contents=[Content.from_function_call(call_id="call_2", name="get_time", arguments="")]), + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call_1", name="get_weather", arguments='{"location":')] + ), + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call_2", name="get_time", arguments='{"timezone":')] + ), + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call_1", name="get_weather", arguments='"NYC"}')] + ), + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call_2", name="get_time", arguments='"EST"}')] + ), + ] + + resp = ChatResponse.from_updates(updates) + assert len(resp.messages) == 1 + fcs = [c for c in resp.messages[0].contents if c.type == "function_call"] + assert len(fcs) == 2 + + by_call_id = {c.call_id: c for c in fcs} + assert by_call_id["call_1"].arguments == '{"location":"NYC"}' + assert by_call_id["call_2"].arguments == '{"timezone":"EST"}' + + +def test_function_call_merge_falls_back_to_trailing_item_without_call_id(): + """Continuation deltas some providers never re-stamp with a call_id still merge. + + Not every provider repeats the call_id on every streamed chunk (e.g. the OpenAI + Chat Completions API only sends it on the first delta for a tool call), so the + fallback of merging into the trailing function_call item must still hold for the + single-call-in-flight case. + """ + updates = [ + ChatResponseUpdate(contents=[Content.from_function_call(call_id="call_1", name="f", arguments="{")]), + ChatResponseUpdate(contents=[Content.from_function_call(call_id="", name="", arguments="}")]), + ] + + resp = ChatResponse.from_updates(updates) + fcs = [c for c in resp.messages[0].contents if c.type == "function_call"] + assert len(fcs) == 1 + assert fcs[0].call_id == "call_1" + assert fcs[0].arguments == "{}" + + # region Role & FinishReason basics From 3467e10007eda561de7eaf0cce78780cbb9ee0bf Mon Sep 17 00:00:00 2001 From: cr-sbarbouche Date: Sat, 12 Sep 2026 21:11:28 +0100 Subject: [PATCH 2/3] Python: don't let a tagged call absorb into an untagged trailing item Content.__add__ only rejects a function_call merge when both sides carry a call_id and they differ, so a chunk tagged with a brand-new call_id could still fall into the trailing-item fallback and get silently absorbed by an untagged trailing call, adopting its id and mashing two unrelated calls' arguments together. Reserve that fallback for chunks that carry no call_id at all; a tagged chunk with no matching in-progress call is always a new call and should be appended. --- .../packages/core/agent_framework/_types.py | 4 ++++ python/packages/core/tests/core/test_types.py | 23 +++++++++++++++++++ 2 files changed, 27 insertions(+) diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index 1e23e9d086..7b65ec3b2a 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -2090,6 +2090,10 @@ def _merge_function_call_content(message: Message, content: Content) -> None: except (AdditionItemMismatch, ContentError): break return + # A tagged chunk that matches no in-progress call is a new call, not a + # continuation - an untagged trailing item would silently absorb it otherwise. + message.contents.append(content) + return if message.contents and message.contents[-1].type == "function_call": try: message.contents[-1] += content diff --git a/python/packages/core/tests/core/test_types.py b/python/packages/core/tests/core/test_types.py index 976b04c652..ff374a609a 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -1970,6 +1970,29 @@ def test_function_call_merge_falls_back_to_trailing_item_without_call_id(): assert fcs[0].arguments == "{}" +def test_function_call_tagged_chunk_does_not_absorb_into_untagged_trailing_call(): + """A tagged chunk with no matching in-progress call must not merge into an untagged one. + + Content.__add__ only rejects a merge when *both* sides carry a call_id and they + differ, so an untagged trailing item (call_id falsy) would otherwise silently + accept a chunk tagged with a brand-new call_id, adopting that id and + concatenating unrelated arguments. The trailing-item fallback must be reserved + for chunks that carry no call_id at all. + """ + untagged = Content("function_call", call_id=None, name="a", arguments="partial-a") + resp = ChatResponse.from_updates([ + ChatResponseUpdate(contents=[untagged]), + ChatResponseUpdate(contents=[Content.from_function_call(call_id="call_new", name="b", arguments="{}")]), + ]) + + fcs = [c for c in resp.messages[0].contents if c.type == "function_call"] + assert len(fcs) == 2 + assert fcs[0].call_id is None + assert fcs[0].arguments == "partial-a" + assert fcs[1].call_id == "call_new" + assert fcs[1].arguments == "{}" + + # region Role & FinishReason basics From 173add75ec707d1711f70b1d43bb6bdebed411e8 Mon Sep 17 00:00:00 2001 From: cr-sbarbouche Date: Mon, 14 Sep 2026 06:59:59 +0100 Subject: [PATCH 3/3] Python: don't merge a reused call_id into an already-identified call Content.__add__ only rejects a function_call merge on id mismatch when both sides carry an id. The Chat Completions client stamps a stable occurrence id on every chunk it emits, but an untagged chunk (id=None) that happens to share an already-identified call's call_id slipped past that guard and got merged anyway, silently corrupting a finished call's arguments if a provider ever reused the call_id. Skip candidates whose id is already set when the incoming chunk has none, and keep scanning instead of assuming a call_id match is proof enough. --- .../packages/core/agent_framework/_types.py | 21 ++++++++++----- python/packages/core/tests/core/test_types.py | 27 +++++++++++++++++++ 2 files changed, 42 insertions(+), 6 deletions(-) diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index 7b65ec3b2a..a89f3a8825 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -2081,15 +2081,24 @@ def _merge_function_call_content(message: Message, content: Content) -> None: back to merging with the trailing function_call item, preserving prior behavior. """ call_id = getattr(content, "call_id", None) + content_id = getattr(content, "id", None) if call_id: for index in range(len(message.contents) - 1, -1, -1): existing = message.contents[index] - if existing.type == "function_call" and getattr(existing, "call_id", None) == call_id: - try: - message.contents[index] = existing + content - except (AdditionItemMismatch, ContentError): - break - return + if existing.type != "function_call" or getattr(existing, "call_id", None) != call_id: + continue + if existing.id is not None and content_id is None: + # existing already has a stable occurrence id from its client (e.g. the + # Chat Completions client stamps one on every chunk); an untagged chunk + # that merely happens to share its call_id isn't proof it's a continuation + # of that specific occurrence - a provider could reuse a call_id for a + # later, unrelated call. Keep scanning rather than merge on a hunch. + continue + try: + message.contents[index] = existing + content + except (AdditionItemMismatch, ContentError): + break + return # A tagged chunk that matches no in-progress call is a new call, not a # continuation - an untagged trailing item would silently absorb it otherwise. message.contents.append(content) diff --git a/python/packages/core/tests/core/test_types.py b/python/packages/core/tests/core/test_types.py index ff374a609a..ffe25291d8 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -1970,6 +1970,33 @@ def test_function_call_merge_falls_back_to_trailing_item_without_call_id(): assert fcs[0].arguments == "{}" +def test_function_call_reused_call_id_does_not_absorb_untagged_chunk(): + """A reused call_id must not let an untagged chunk merge into an already-identified call. + + The Chat Completions client stamps a stable occurrence ``id`` on every function-call + chunk it emits, precisely so a provider reusing a ``call_id`` (or a client that only + tags the first chunk) can't be confused with an unrelated call. If a later, untagged + chunk happens to carry the same ``call_id`` as a call that already has an occurrence + id, it must not be assumed to be that call's continuation - it should be treated as + a new, separate call instead of corrupting the finished one's arguments. + """ + updates = [ + ChatResponseUpdate( + contents=[Content.from_function_call(id="af-call-1", call_id="call_1", name="get_weather", arguments="{}")] + ), + # No `id` and a reused call_id: an unrelated call, not a continuation. + ChatResponseUpdate(contents=[Content.from_function_call(call_id="call_1", name="get_time", arguments="{}")]), + ] + + resp = ChatResponse.from_updates(updates) + fcs = [c for c in resp.messages[0].contents if c.type == "function_call"] + assert len(fcs) == 2 + assert fcs[0].name == "get_weather" + assert fcs[0].arguments == "{}" + assert fcs[1].name == "get_time" + assert fcs[1].arguments == "{}" + + def test_function_call_tagged_chunk_does_not_absorb_into_untagged_trailing_call(): """A tagged chunk with no matching in-progress call must not merge into an untagged one.