Python: fix thread serialization for multi-turn tool calls (#4684)

* Python: strip fc_id from loaded history

* Move fc_id replay handling into Responses client

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Remove unnecessary pytest asyncio marker

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Add Responses integration test for fc_id replay

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* removed old arg

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: eavanvalkenburg

python/packages/core/agent_framework/openai/_responses_client.py

@@ -1032,24 +1032,27 @@ def _prepare_messages_for_openai(self, chat_messages: Sequence[Message]) -> list
         Returns:
             The prepared chat messages for a request.
         """
-        call_id_to_id: dict[str, str] = {}
-        for message in chat_messages:
-            for content in message.contents:
-                if (
-                    content.type == "function_call"
-                    and content.additional_properties
-                    and "fc_id" in content.additional_properties
-                    and content.additional_properties["fc_id"]
-                ):
-                    call_id_to_id[content.call_id] = content.additional_properties["fc_id"]  # type: ignore[attr-defined, index]
-        list_of_list = [self._prepare_message_for_openai(message, call_id_to_id) for message in chat_messages]
+        list_of_list = [self._prepare_message_for_openai(message) for message in chat_messages]
         # Flatten the list of lists into a single list
         return list(chain.from_iterable(list_of_list))
 
+    @staticmethod
+    def _message_replays_provider_context(message: Message) -> bool:
+        """Return whether the message came from provider-attributed replay context.
+
+        Responses ``fc_id`` values are response-scoped and only valid while replaying
+        the same live tool loop. Once a message comes back through a context provider
+        (for example, loaded session history), that message is historical input and
+        must not reuse the original response-scoped ``fc_id``.
+        """
+        additional_properties = getattr(message, "additional_properties", None)
+        if not additional_properties:
+            return False
+        return "_attribution" in additional_properties
+
     def _prepare_message_for_openai(
         self,
         message: Message,
-        call_id_to_id: dict[str, str],
     ) -> list[dict[str, Any]]:
         """Prepare a chat message for the OpenAI Responses API format."""
         all_messages: list[dict[str, Any]] = []
@@ -1067,39 +1070,41 @@ def _prepare_message_for_openai(
                 case "text_reasoning":
                     if not has_function_call:
                         continue  # reasoning not followed by a function_call is invalid in input
-                    reasoning = self._prepare_content_for_openai(message.role, content, call_id_to_id)  # type: ignore[arg-type]
+                    reasoning = self._prepare_content_for_openai(message.role, content, message=message)
                     if reasoning:
                         all_messages.append(reasoning)
                 case "function_result":
                     new_args: dict[str, Any] = {}
-                    new_args.update(self._prepare_content_for_openai(message.role, content, call_id_to_id))  # type: ignore[arg-type]
+                    new_args.update(self._prepare_content_for_openai(message.role, content, message=message))
                     if new_args:
                         all_messages.append(new_args)
                 case "function_call":
-                    function_call = self._prepare_content_for_openai(message.role, content, call_id_to_id)  # type: ignore[arg-type]
+                    function_call = self._prepare_content_for_openai(message.role, content, message=message)
                     if function_call:
-                        all_messages.append(function_call)  # type: ignore
+                        all_messages.append(function_call)
                 case "function_approval_response" | "function_approval_request":
-                    prepared = self._prepare_content_for_openai(Role(message.role), content, call_id_to_id)
+                    prepared = self._prepare_content_for_openai(message.role, content, message=message)
                     if prepared:
-                        all_messages.append(prepared)  # type: ignore
+                        all_messages.append(prepared)
                 case _:
-                    prepared_content = self._prepare_content_for_openai(message.role, content, call_id_to_id)  # type: ignore
+                    prepared_content = self._prepare_content_for_openai(message.role, content, message=message)
                     if prepared_content:
                         if "content" not in args:
                             args["content"] = []
-                        args["content"].append(prepared_content)  # type: ignore
+                        args["content"].append(prepared_content)  # type: ignore[reportUnknownMemberType]
         if "content" in args or "tool_calls" in args:
             all_messages.append(args)
         return all_messages
 
     def _prepare_content_for_openai(
         self,
-        role: Role,
+        role: Role | str,
         content: Content,
-        call_id_to_id: dict[str, str],
+        *,
+        message: Message | None = None,
     ) -> dict[str, Any]:
         """Prepare content for the OpenAI Responses API format."""
+        role = Role(role)
         match content.type:
             case "text":
                 if role == "assistant":
@@ -1174,8 +1179,15 @@ def _prepare_content_for_openai(
                 if not content.call_id:
                     logger.warning(f"FunctionCallContent missing call_id for function '{content.name}'")
                     return {}
-                # Use fc_id from additional_properties if available, otherwise fallback to call_id
-                fc_id = call_id_to_id.get(content.call_id, content.call_id)
+                fc_id = content.call_id
+                if (
+                    message is not None
+                    and not self._message_replays_provider_context(message)
+                    and content.additional_properties
+                ):
+                    live_fc_id = content.additional_properties.get("fc_id")
+                    if isinstance(live_fc_id, str) and live_fc_id:
+                        fc_id = live_fc_id
                 # OpenAI Responses API requires IDs to start with `fc_`
                 if not fc_id.startswith("fc_"):
                     fc_id = f"fc_{fc_id}"
@@ -1221,7 +1233,7 @@ def _prepare_content_for_openai(
                         if item.type == "text":
                             output_parts.append({"type": "input_text", "text": item.text or ""})
                         else:
-                            part = self._prepare_content_for_openai("user", item, call_id_to_id)  # type: ignore[arg-type]
+                            part = self._prepare_content_for_openai("user", item)
                             if part:
                                 output_parts.append(part)
                     if output_parts:

python/packages/core/tests/core/test_agents.py

@@ -2,9 +2,10 @@
 
 import contextlib
 import inspect
+import json
 from collections.abc import AsyncIterable, MutableSequence
 from typing import Any
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 from uuid import uuid4
 
 import pytest
@@ -1943,6 +1944,128 @@ async def test_stores_by_default_with_store_false_in_default_options_injects_inm
     assert any(isinstance(p, InMemoryHistoryProvider) for p in agent.context_providers)
 
 
+async def test_shared_local_storage_cross_provider_responses_history_does_not_leak_fc_id() -> None:
+    """Responses-specific replay metadata should stay local to Responses when session storage is shared."""
+    from openai.types.chat.chat_completion import ChatCompletion, Choice
+    from openai.types.chat.chat_completion_message import ChatCompletionMessage
+
+    from agent_framework._sessions import InMemoryHistoryProvider
+    from agent_framework.openai import OpenAIChatClient, OpenAIResponsesClient
+
+    @tool(approval_mode="never_require")
+    def search_hotels(city: str) -> str:
+        return f"Found 3 hotels in {city}"
+
+    responses_client = OpenAIResponsesClient(model_id="test-model", api_key="test-key")
+    responses_agent = Agent(
+        client=responses_client,
+        tools=[search_hotels],
+        default_options={"store": False},
+    )
+    session = responses_agent.create_session()
+
+    responses_tool_call = MagicMock()
+    responses_tool_call.type = "function_call"
+    responses_tool_call.id = "fc_provider123"
+    responses_tool_call.call_id = "call_1"
+    responses_tool_call.name = "search_hotels"
+    responses_tool_call.arguments = '{"city": "Paris"}'
+    responses_tool_call.status = "completed"
+
+    responses_first = MagicMock()
+    responses_first.output_parsed = None
+    responses_first.metadata = {}
+    responses_first.usage = None
+    responses_first.id = "resp_1"
+    responses_first.model = "test-model"
+    responses_first.created_at = 1000000000
+    responses_first.status = "completed"
+    responses_first.finish_reason = "tool_calls"
+    responses_first.incomplete = None
+    responses_first.output = [responses_tool_call]
+
+    responses_text_item = MagicMock()
+    responses_text_item.type = "message"
+    responses_text_content = MagicMock()
+    responses_text_content.type = "output_text"
+    responses_text_content.text = "Hotel Lutetia is the cheapest option."
+    responses_text_item.content = [responses_text_content]
+
+    responses_second = MagicMock()
+    responses_second.output_parsed = None
+    responses_second.metadata = {}
+    responses_second.usage = None
+    responses_second.id = "resp_2"
+    responses_second.model = "test-model"
+    responses_second.created_at = 1000000001
+    responses_second.status = "completed"
+    responses_second.finish_reason = "stop"
+    responses_second.incomplete = None
+    responses_second.output = [responses_text_item]
+
+    with patch.object(
+        responses_client.client.responses,
+        "create",
+        side_effect=[responses_first, responses_second],
+    ) as mock_responses_create:
+        responses_result = await responses_agent.run("Find me a hotel in Paris", session=session)
+
+    assert responses_result.text == "Hotel Lutetia is the cheapest option."
+    assert any(isinstance(provider, InMemoryHistoryProvider) for provider in responses_agent.context_providers)
+
+    shared_messages = session.state[InMemoryHistoryProvider.DEFAULT_SOURCE_ID]["messages"]
+    shared_function_call = next(
+        content for message in shared_messages for content in message.contents if content.type == "function_call"
+    )
+    assert shared_function_call.additional_properties is not None
+    assert shared_function_call.additional_properties.get("fc_id") == "fc_provider123"
+
+    responses_replay_input = mock_responses_create.call_args_list[1].kwargs["input"]
+    responses_replay_call = next(item for item in responses_replay_input if item.get("type") == "function_call")
+    assert responses_replay_call["id"] == "fc_provider123"
+
+    chat_client = OpenAIChatClient(model_id="test-model", api_key="test-key")
+    chat_agent = Agent(client=chat_client)
+
+    chat_response = ChatCompletion(
+        id="chatcmpl-test",
+        object="chat.completion",
+        created=1234567890,
+        model="gpt-4o-mini",
+        choices=[
+            Choice(
+                index=0,
+                message=ChatCompletionMessage(role="assistant", content="The cheapest option is still Hotel Lutetia."),
+                finish_reason="stop",
+            )
+        ],
+    )
+
+    with patch.object(
+        chat_client.client.chat.completions,
+        "create",
+        new=AsyncMock(return_value=chat_response),
+    ) as mock_chat_create:
+        chat_result = await chat_agent.run("Which option is cheapest?", session=session)
+
+    assert chat_result.text == "The cheapest option is still Hotel Lutetia."
+
+    chat_request_messages = mock_chat_create.call_args.kwargs["messages"]
+    assistant_tool_call_message = next(
+        message for message in chat_request_messages if message.get("role") == "assistant" and message.get("tool_calls")
+    )
+    assert assistant_tool_call_message["tool_calls"][0]["id"] == "call_1"
+    assert assistant_tool_call_message["tool_calls"][0]["function"]["name"] == "search_hotels"
+
+    tool_result_message = next(
+        message
+        for message in chat_request_messages
+        if message.get("role") == "tool" and message.get("tool_call_id") == "call_1"
+    )
+    assert tool_result_message["content"] == "Found 3 hotels in Paris"
+    assert "fc_provider123" not in json.dumps(chat_request_messages)
+
+
 # region as_tool user_input_request propagation