diff --git a/python/packages/core/agent_framework/_workflows/_agent_executor.py b/python/packages/core/agent_framework/_workflows/_agent_executor.py index 22e872be07c..5b5f836b012 100644 --- a/python/packages/core/agent_framework/_workflows/_agent_executor.py +++ b/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: diff --git a/python/packages/core/tests/workflow/test_full_conversation.py b/python/packages/core/tests/workflow/test_full_conversation.py index fbb9ac4eeee..c7f345090de 100644 --- a/python/packages/core/tests/workflow/test_full_conversation.py +++ b/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