Python: Fix duplicate item error for tool-using agents under service-side storage (#3295)

python/packages/core/tests/workflow/test_full_conversation.py

@@ -429,7 +429,12 @@ async def handle(
 
 
 @pytest.mark.xfail(
-    reason="reset_service_session support not yet implemented — see #4047",
+    reason=(
+        "Tracks the executor-layer half of #3295: AgentExecutor should clear service_session_id "
+        "when handed a full prior conversation. The wire-level 'Duplicate item' API error is "
+        "already closed by the chat-client strip in #3295; this xfail covers the defense-in-depth "
+        "follow-up that makes the executor wiring reflect intent."
+    ),
     strict=True,
 )
 async def test_run_request_with_full_history_clears_service_session_id() -> None:

python/packages/foundry/tests/foundry/test_foundry_chat_client.py

@@ -435,7 +435,7 @@ async def test_chat_message_parsing_with_function_calls() -> None:
         Message(role="tool", contents=[function_result]),
     ]
 
-    prepared_messages = client._prepare_messages_for_openai(messages)
+    prepared_messages = client._prepare_messages_for_openai(messages, request_uses_service_side_storage=False)
 
     assert prepared_messages == [
         {

python/packages/openai/agent_framework_openai/_chat_client.py

@@ -1409,29 +1409,31 @@ def _prepare_message_for_openai(
         }
         additional_properties = message.additional_properties
         replays_local_storage = "_attribution" in additional_properties
-        uses_service_side_storage = request_uses_service_side_storage and not replays_local_storage
-        # Reasoning items are only valid in input when they directly preceded a function_call
-        # in the same response. Including a reasoning item that preceded a text response
-        # (i.e. no function_call in the same message) causes an API error:
-        # "reasoning was provided without its required following item."
-        #
-        # Local storage is stricter: response-scoped reasoning items (rs_*) cannot be replayed
-        # back to the service unless that message is using service-side storage.
-        # In that mode we omit reasoning items and rely on function call + tool output replay.
-        has_function_call = any(c.type == "function_call" for c in message.contents)
+        # Server-issued response item identities (function_call fc_*, reasoning rs_*, approval IDs,
+        # local-shell-call IDs) must not be re-sent inline when the request carries
+        # previous_response_id / conversation_id / conversation: the server already has them via
+        # the prior response and rejects duplicates with "Duplicate item found with id ...".
+        # function_result keeps its call_id and the server pairs it to the prior function_call via
+        # that key. See microsoft/agent-framework#3295. The strip is gated on the request-level
+        # flag, not a message-level one: HistoryProvider-attributed messages
+        # (replays_local_storage) still need stripping when the request also carries a continuation
+        # marker, since the server-stored items would otherwise duplicate the inline ones. Without
+        # storage, standalone reasoning items are invalid per the API ("reasoning was provided
+        # without its required following item"), so the reasoning branch always drops.
         for content in message.contents:
             match content.type:
                 case "text_reasoning":
-                    if not uses_service_side_storage or 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,
-                        replays_local_storage=replays_local_storage,
-                    )
-                    if reasoning:
-                        all_messages.append(reasoning)
+                    continue
                 case "function_result":
+                    if request_uses_service_side_storage:
+                        props = content.additional_properties or {}
+                        # Local-shell variant serializes as `local_shell_call` carrying a server-issued id;
+                        # plain function_call_output pairs by call_id and is safe under storage.
+                        if (
+                            props.get(OPENAI_SHELL_OUTPUT_TYPE_KEY) == OPENAI_SHELL_OUTPUT_TYPE_LOCAL_SHELL_CALL
+                            and props.get(OPENAI_LOCAL_SHELL_CALL_ITEM_ID_KEY)
+                        ):
+                            continue
                     new_args: dict[str, Any] = {}
                     new_args.update(
                         self._prepare_content_for_openai(
@@ -1443,6 +1445,8 @@ def _prepare_message_for_openai(
                     if new_args:
                         all_messages.append(new_args)
                 case "function_call":
+                    if request_uses_service_side_storage:
+                        continue
                     function_call = self._prepare_content_for_openai(
                         message.role,
                         content,
@@ -1451,6 +1455,8 @@ def _prepare_message_for_openai(
                     if function_call:
                         all_messages.append(function_call)
                 case "function_approval_response" | "function_approval_request":
+                    if request_uses_service_side_storage:
+                        continue
                     prepared = self._prepare_content_for_openai(
                         message.role,
                         content,
@@ -1463,6 +1469,12 @@ def _prepare_message_for_openai(
                     # top-level mcp_call input item; the result side emits an
                     # internal marker that `_prepare_messages_for_openai`
                     # coalesces onto the matching call (or drops if unmatched).
+                    # The mcp_call item carries the model-emitted call_id as its
+                    # server-side `id`, so under continuation it would duplicate
+                    # the prior response's items (#3295). Drop the call here; the
+                    # orphan result is dropped by the coalesce step that follows.
+                    if request_uses_service_side_storage:
+                        continue
                     prepared_mcp = self._prepare_content_for_openai(
                         message.role,
                         content,