diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py
index 4ab15a52f79..602da9dc4cb 100644
--- a/python/packages/core/agent_framework/_tools.py
+++ b/python/packages/core/agent_framework/_tools.py
@@ -3029,6 +3029,7 @@ def _bind_approval_responses_to_pending_requests(
     invocation_session: AgentSession | None,
     *,
     consume: bool = True,
+    staged_response_ids: set[int] | None = None,
 ) -> set[int]:
     """Rebind approval responses and remove unissued or duplicate responses."""
     if invocation_session is None:
@@ -3049,6 +3050,8 @@ def _bind_approval_responses_to_pending_requests(
                 consume=consume,
             )
             if rebound is None:
+                if staged_response_ids and id(content) in staged_response_ids:
+                    continue
                 logger.warning(
                     "Ignored an approval response with id %r because it did not match the active approval "
                     "occurrence identity; the pending request was retained for retry.",
@@ -3107,8 +3110,13 @@ def _store_already_approved_approval_requests(
 def _stage_approval_batch_responses(
     invocation_session: AgentSession | None,
     approval_responses: Sequence[Content],
+    *,
+    staged_response_ids: set[int] | None = None,
 ) -> tuple[list[Content], list[dict[str, str | None]], list[Content] | None]:
-    """Accumulate approval decisions and release a batch only when every decision is present."""
+    """Accumulate approval decisions and release a batch only when every decision is present.
+
+    ``staged_response_ids`` records matched source objects for diagnostic-only warning suppression.
+    """
     if not approval_responses:
         return [], [], None
     state = _get_tool_approval_state(invocation_session)
@@ -3154,6 +3162,8 @@ def _stage_approval_batch_responses(
                 continue
             if request_id not in stored_responses:
                 stored_responses[request_id] = rebound
+            if staged_response_ids is not None:
+                staged_response_ids.add(id(response))
             matched = True
 
         if not matched:
@@ -4593,9 +4603,11 @@ async def _resolve_approval_responses(
     pending_responses_before_binding = list(
         _collect_approval_responses(prepared_messages, non_approval_result_ids=host_result_ids).values()
     )
+    staged_response_ids: set[int] = set()
     staged_responses, function_call_order, waiting_requests = _stage_approval_batch_responses(
         approval_session,
         pending_responses_before_binding,
+        staged_response_ids=staged_response_ids,
     )
     if waiting_requests is not None:
         response_messages, streaming_updates = _messages_and_updates_for_terminal_contents(waiting_requests)
@@ -4610,6 +4622,7 @@ async def _resolve_approval_responses(
         prepared_messages,
         approval_session,
         consume=False,
+        staged_response_ids=staged_response_ids,
     )
     active_pending_ids = (
         set(_load_pending_approval_requests(approval_session))
diff --git a/python/packages/core/tests/core/test_harness_tool_approval.py b/python/packages/core/tests/core/test_harness_tool_approval.py
index 57c9f70993e..0bab34d8f5d 100644
--- a/python/packages/core/tests/core/test_harness_tool_approval.py
+++ b/python/packages/core/tests/core/test_harness_tool_approval.py
@@ -6,6 +6,7 @@
 import json
 import warnings
 from collections.abc import Awaitable, Callable, MutableSequence
+from contextlib import nullcontext
 from datetime import timedelta
 from enum import Enum
 from pathlib import Path
@@ -1119,12 +1120,17 @@ async def capture_response(
     assert "function_approval_response" not in replayed_types
 
 
+@pytest.mark.parametrize("legacy_request_id", [False, True], ids=["occurrence-id", "legacy-id"])
+@pytest.mark.parametrize("conflicting_duplicate", [False, True], ids=["single-response", "conflicting-duplicate"])
 @pytest.mark.parametrize("approved", [True, False], ids=["approved", "rejected"])
 @pytest.mark.parametrize("streaming", [False, True], ids=["non-streaming", "streaming"])
 async def test_approval_resume_returns_result_without_mutating_inputs(
     chat_client_base: MockBaseChatClient,
+    caplog: pytest.LogCaptureFixture,
     approved: bool,
     streaming: bool,
+    legacy_request_id: bool,
+    conflicting_duplicate: bool,
 ) -> None:
     """Approval resume should return its terminal result without changing caller-owned messages."""
     calls = 0
@@ -1150,23 +1156,49 @@ def guarded_tool() -> str:
         first_response = await agent.run("run guarded", session=session)
         approval_request = first_response.user_input_requests[0]
 
-    approval_response = approval_request.to_function_approval_response(approved=approved)
-    approval_message = Message(role="user", contents=[approval_response])
+    if legacy_request_id:
+        from agent_framework._tools import _store_already_approved_approval_requests, _store_pending_approval_requests
 
-    if streaming:
-        chat_client_base.streaming_responses = [
-            [ChatResponseUpdate(role="assistant", contents=[Content.from_text("done")])]
-        ]
-        second_stream = agent.run(approval_message, stream=True, session=session)
-        second_updates = [update async for update in second_stream]
-        second_response = await second_stream.get_final_response()
-        assert [[content.type for content in update.contents] for update in second_updates] == [
-            ["function_result"],
-            ["text"],
-        ]
-    else:
-        chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=["done"]))]
-        second_response = await agent.run(approval_message, session=session)
+        approval_request = Content.from_dict(approval_request.to_dict())
+        approval_request.id = "call_guarded"
+        assert approval_request.id != _function_call(approval_request).id
+        _store_pending_approval_requests(session, [approval_request])
+        _store_already_approved_approval_requests(session, [approval_request], [])
+
+    approval_response = approval_request.to_function_approval_response(approved=approved)
+    approval_contents = [approval_response]
+    if conflicting_duplicate:
+        conflicting_call = Content.from_dict(_function_call(approval_request).to_dict())
+        conflicting_call.id = "af-call-conflicting-duplicate"
+        assert approval_response.id is not None
+        approval_contents.append(
+            Content.from_function_approval_response(
+                approved=not approved,
+                id=approval_response.id,
+                function_call=conflicting_call,
+            )
+        )
+    approval_message = Message(role="user", contents=approval_contents.copy())
+
+    with (
+        pytest.warns(FutureWarning, match="legacy provider call_id request binding")
+        if legacy_request_id
+        else nullcontext[None]()
+    ):
+        if streaming:
+            chat_client_base.streaming_responses = [
+                [ChatResponseUpdate(role="assistant", contents=[Content.from_text("done")])]
+            ]
+            second_stream = agent.run(approval_message, stream=True, session=session)
+            second_updates = [update async for update in second_stream]
+            second_response = await second_stream.get_final_response()
+            assert [[content.type for content in update.contents] for update in second_updates] == [
+                ["function_result"],
+                ["text"],
+            ]
+        else:
+            chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=["done"]))]
+            second_response = await agent.run(approval_message, session=session)
 
     assert [[content.type for content in message.contents] for message in second_response.messages] == [
         ["function_result"],
@@ -1176,12 +1208,28 @@ def guarded_tool() -> str:
     assert result.call_id == "call_guarded"
     assert result.result == ("approved result" if approved else "Error: Tool call invocation was rejected by user.")
     assert calls == int(approved)
+    if conflicting_duplicate:
+        assert len(caplog.records) == 1
+        assert "occurrence identity" in caplog.text
+    else:
+        assert caplog.records == []
     assert approval_message.role == "user"
-    assert approval_message.contents == [approval_response]
+    assert approval_message.contents == approval_contents
     assert [[content.type for content in message.contents] for message in first_response.messages] == [
         ["function_call", "function_approval_request"]
     ]
 
+    caplog.clear()
+    replay_response = approval_request.to_function_approval_response(approved=True)
+    if streaming:
+        replay_stream = agent.run(replay_response, stream=True, session=session)
+        _ = [update async for update in replay_stream]
+        await replay_stream.get_final_response()
+    else:
+        await agent.run(replay_response, session=session)
+    assert calls == int(approved)
+    assert "occurrence identity" in caplog.text
+
 
 @pytest.mark.parametrize("streaming", [False, True], ids=["non-streaming", "streaming"])
 async def test_approval_resume_replays_reasoning_with_function_call_group(