Python: Clear stale service session on full-history replay

python/packages/core/agent_framework/_workflows/_agent_executor.py

@@ -202,9 +202,14 @@ async def run(
     ) -> None:
         """Handle an AgentExecutorRequest (canonical input).
 
-        This is the standard path: extend cache with provided messages; if should_respond
-        run the agent and emit an AgentExecutorResponse downstream.
+        This is the standard path: extend cache with provided messages; if should_respond,
+        run the agent and emit an AgentExecutorResponse downstream. When the request
+        replays function-call history, clear the service session because those messages
+        provide the continuation context explicitly.
         """
+        if any(content.type == "function_call" for message in request.messages for content in message.contents):
+            self._session.service_session_id = None
+
         self._cache.extend(request.messages)
 
         if request.should_respond:

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

@@ -3,7 +3,6 @@
 from collections.abc import AsyncIterable, Awaitable
 from typing import Any, Literal, overload
 
-import pytest
 from pydantic import PrivateAttr
 from typing_extensions import Never
 
@@ -429,15 +428,6 @@ async def handle(
         )
 
 
-@pytest.mark.xfail(
-    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:
     """Replaying a full conversation (including function calls) via AgentExecutorRequest must
     clear service_session_id so the API does not receive both previous_response_id and the
@@ -467,6 +457,36 @@ async def test_run_request_with_full_history_clears_service_session_id() -> None
     assert spy_agent._captured_service_session_id is None  # pyright: ignore[reportPrivateUsage]
 
 
+async def test_run_request_with_function_result_preserves_service_session_id() -> None:
+    """A function result continues the service-stored call instead of replaying it."""
+    spy_agent = _SessionIdCapturingAgent(id="result_spy_agent", name="ResultSpyAgent")
+    spy_exec = AgentExecutor(spy_agent, id="result_spy_agent")
+    spy_exec._session.service_session_id = "resp_PENDING_TOOL_CALL"  # pyright: ignore[reportPrivateUsage]
+
+    @executor(id="function_result_source")
+    async def function_result_source(
+        _: str,
+        ctx: WorkflowContext[AgentExecutorRequest, Any],
+    ) -> None:
+        await ctx.send_message(
+            AgentExecutorRequest(
+                messages=[
+                    Message(
+                        role="tool",
+                        contents=[Content.from_function_result(call_id="call_weather_1", result="Sunny, 72F")],
+                    )
+                ],
+                should_respond=True,
+            )
+        )
+
+    wf = WorkflowBuilder(start_executor=function_result_source).add_edge(function_result_source, spy_exec).build()
+
+    await wf.run("tool result")
+
+    assert spy_agent._captured_service_session_id == "resp_PENDING_TOOL_CALL"  # pyright: ignore[reportPrivateUsage]
+
+
 async def test_from_response_preserves_service_session_id() -> None:
     """from_response hands off a prior agent's full conversation to the next executor.
     The receiving executor's service_session_id is preserved so the API can continue