From a1835e1c38bbdddcb13c539a1cbd6ec2bfbf775f Mon Sep 17 00:00:00 2001 From: Benke Qu Date: Thu, 14 May 2026 17:16:07 -0700 Subject: [PATCH 1/5] fix: clear service_session_id in _agent_wrapper when propagate_session=True When propagate_session=True, the child agent inherits the parent's service_session_id. After the parent's first LLM call, MAF auto-populates this from the Responses API conversation_id. The child sends it as previous_response_id which the server rejects because the parent's tool_call is still pending (400 error). This fix saves and clears service_session_id before calling the child agent and restores it in a finally block, preserving session.state sharing while isolating the server-side conversation pointer. Fixes #5874 --- .../packages/core/agent_framework/_agents.py | 40 +++++--- .../packages/core/tests/core/test_agents.py | 93 +++++++++++++++++++ 2 files changed, 120 insertions(+), 13 deletions(-) diff --git a/python/packages/core/agent_framework/_agents.py b/python/packages/core/agent_framework/_agents.py index 585898ae52..3f2de9c6b4 100644 --- a/python/packages/core/agent_framework/_agents.py +++ b/python/packages/core/agent_framework/_agents.py @@ -549,19 +549,33 @@ async def _agent_wrapper(ctx: FunctionInvocationContext, **kwargs: Any) -> str: ctx: the function invocation context used **kwargs: only used to dynamically load the argument that is defined for this tool. """ - stream = self.run( - str(kwargs.get(arg_name, "")), - stream=True, - session=ctx.session if propagate_session else None, - function_invocation_kwargs=dict(ctx.kwargs), - ) - if stream_callback is not None: - stream.with_transform_hook(stream_callback) - final_response = await stream.get_final_response() - if final_response.user_input_requests: - raise UserInputRequiredException(contents=final_response.user_input_requests) - # TODO(Copilot): update once #4331 merges - return final_response.text + session = ctx.session if propagate_session else None + + # Isolate the child agent from the parent's server-side conversation. + # service_session_id would cause the child to send previous_response_id + # referencing the parent's pending tool_call, resulting in a 400 error. + saved_service_session_id = None + if session is not None and session.service_session_id is not None: + saved_service_session_id = session.service_session_id + session.service_session_id = None + + try: + stream = self.run( + str(kwargs.get(arg_name, "")), + stream=True, + session=session, + function_invocation_kwargs=dict(ctx.kwargs), + ) + if stream_callback is not None: + stream.with_transform_hook(stream_callback) + final_response = await stream.get_final_response() + if final_response.user_input_requests: + raise UserInputRequiredException(contents=final_response.user_input_requests) + # TODO(Copilot): update once #4331 merges + return final_response.text + finally: + if session is not None and saved_service_session_id is not None: + session.service_session_id = saved_service_session_id return FunctionTool( name=tool_name, diff --git a/python/packages/core/tests/core/test_agents.py b/python/packages/core/tests/core/test_agents.py index f8e460e127..c816d4c7ed 100644 --- a/python/packages/core/tests/core/test_agents.py +++ b/python/packages/core/tests/core/test_agents.py @@ -1532,6 +1532,99 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any: assert parent_session.state["counter"] == 1 +async def test_chat_agent_as_tool_propagate_session_clears_service_session_id(client: SupportsChatGetResponse) -> None: + """Test that propagate_session=True clears service_session_id for the child and restores it after.""" + agent = Agent(client=client, name="SubAgent", description="Sub agent") + tool = agent.as_tool(propagate_session=True) + + parent_session = AgentSession(session_id="shared-session") + parent_session.service_session_id = "resp_parent_abc123" + parent_session.state["data"] = "shared" + + original_run = agent.run + captured_session = None + + def capturing_run(*args: Any, **kwargs: Any) -> Any: + nonlocal captured_session + captured_session = kwargs.get("session") + # The child should see the same session object but with service_session_id cleared + assert captured_session is parent_session + assert captured_session.service_session_id is None + assert captured_session.state["data"] == "shared" + return original_run(*args, **kwargs) + + agent.run = capturing_run # type: ignore[assignment, method-assign] + + await tool.invoke( + context=FunctionInvocationContext( + function=tool, + arguments={"task": "Hello"}, + session=parent_session, + ) + ) + + # After the child finishes, service_session_id is restored + assert parent_session.service_session_id == "resp_parent_abc123" + + +async def test_chat_agent_as_tool_propagate_session_restores_service_session_id_on_error( + client: SupportsChatGetResponse, +) -> None: + """Test that service_session_id is restored even if the child agent raises.""" + agent = Agent(client=client, name="SubAgent", description="Sub agent") + tool = agent.as_tool(propagate_session=True) + + parent_session = AgentSession(session_id="shared-session") + parent_session.service_session_id = "resp_parent_xyz789" + + def failing_run(*args: Any, **kwargs: Any) -> Any: + raise RuntimeError("Child agent failed") + + agent.run = failing_run # type: ignore[assignment, method-assign] + + with raises(RuntimeError, match="Child agent failed"): + await tool.invoke( + context=FunctionInvocationContext( + function=tool, + arguments={"task": "Hello"}, + session=parent_session, + ) + ) + + # service_session_id must be restored even after failure + assert parent_session.service_session_id == "resp_parent_xyz789" + + +async def test_chat_agent_as_tool_propagate_session_no_service_session_id(client: SupportsChatGetResponse) -> None: + """Test that when service_session_id is None, no save/restore is needed.""" + agent = Agent(client=client, name="SubAgent", description="Sub agent") + tool = agent.as_tool(propagate_session=True) + + parent_session = AgentSession(session_id="shared-session") + parent_session.service_session_id = None + + original_run = agent.run + captured_session = None + + def capturing_run(*args: Any, **kwargs: Any) -> Any: + nonlocal captured_session + captured_session = kwargs.get("session") + assert captured_session.service_session_id is None + return original_run(*args, **kwargs) + + agent.run = capturing_run # type: ignore[assignment, method-assign] + + await tool.invoke( + context=FunctionInvocationContext( + function=tool, + arguments={"task": "Hello"}, + session=parent_session, + ) + ) + + assert parent_session.service_session_id is None + + async def test_chat_agent_as_mcp_server_basic(client: SupportsChatGetResponse) -> None: """Test basic as_mcp_server functionality.""" agent = Agent(client=client, name="TestAgent", description="Test agent for MCP") From 81c31449d3e58e1b0bbff5983dd6b2f63b7f07a9 Mon Sep 17 00:00:00 2001 From: Benke Qu Date: Mon, 1 Jun 2026 10:51:26 -0700 Subject: [PATCH 2/5] refactor: use child session copy instead of in-place mutation Address Copilot review comments: - Create a child AgentSession with shared state dict but isolated service_session_id, avoiding race conditions under concurrent asyncio.gather tool invocations. - Update tests to verify child gets a separate session object and that child-set service_session_id does not leak to parent. --- .../packages/core/agent_framework/_agents.py | 48 +++++++++---------- .../packages/core/tests/core/test_agents.py | 19 +++++--- 2 files changed, 35 insertions(+), 32 deletions(-) diff --git a/python/packages/core/agent_framework/_agents.py b/python/packages/core/agent_framework/_agents.py index 3f2de9c6b4..d2ee55ded5 100644 --- a/python/packages/core/agent_framework/_agents.py +++ b/python/packages/core/agent_framework/_agents.py @@ -551,31 +551,29 @@ async def _agent_wrapper(ctx: FunctionInvocationContext, **kwargs: Any) -> str: """ session = ctx.session if propagate_session else None - # Isolate the child agent from the parent's server-side conversation. - # service_session_id would cause the child to send previous_response_id - # referencing the parent's pending tool_call, resulting in a 400 error. - saved_service_session_id = None - if session is not None and session.service_session_id is not None: - saved_service_session_id = session.service_session_id - session.service_session_id = None - - try: - stream = self.run( - str(kwargs.get(arg_name, "")), - stream=True, - session=session, - function_invocation_kwargs=dict(ctx.kwargs), - ) - if stream_callback is not None: - stream.with_transform_hook(stream_callback) - final_response = await stream.get_final_response() - if final_response.user_input_requests: - raise UserInputRequiredException(contents=final_response.user_input_requests) - # TODO(Copilot): update once #4331 merges - return final_response.text - finally: - if session is not None and saved_service_session_id is not None: - session.service_session_id = saved_service_session_id + # Create a child session that shares the parent's state dict but has + # an isolated service_session_id. This avoids mutating the parent + # session in-place, which would race under concurrent asyncio.gather + # tool invocations sharing the same session. + if session is not None: + child_session = AgentSession(session_id=session.session_id) + child_session.state = session.state # shared by reference + child_session.service_session_id = None + session = child_session + + stream = self.run( + str(kwargs.get(arg_name, "")), + stream=True, + session=session, + function_invocation_kwargs=dict(ctx.kwargs), + ) + if stream_callback is not None: + stream.with_transform_hook(stream_callback) + final_response = await stream.get_final_response() + if final_response.user_input_requests: + raise UserInputRequiredException(contents=final_response.user_input_requests) + # TODO(Copilot): update once #4331 merges + return final_response.text return FunctionTool( name=tool_name, diff --git a/python/packages/core/tests/core/test_agents.py b/python/packages/core/tests/core/test_agents.py index c816d4c7ed..19c896f09a 100644 --- a/python/packages/core/tests/core/test_agents.py +++ b/python/packages/core/tests/core/test_agents.py @@ -1533,7 +1533,7 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any: async def test_chat_agent_as_tool_propagate_session_clears_service_session_id(client: SupportsChatGetResponse) -> None: - """Test that propagate_session=True clears service_session_id for the child and restores it after.""" + """Test that propagate_session=True gives the child a separate session with cleared service_session_id.""" agent = Agent(client=client, name="SubAgent", description="Sub agent") tool = agent.as_tool(propagate_session=True) @@ -1547,9 +1547,11 @@ async def test_chat_agent_as_tool_propagate_session_clears_service_session_id(cl def capturing_run(*args: Any, **kwargs: Any) -> Any: nonlocal captured_session captured_session = kwargs.get("session") - # The child should see the same session object but with service_session_id cleared - assert captured_session is parent_session + # The child gets a different session object with isolated service_session_id + assert captured_session is not parent_session assert captured_session.service_session_id is None + # But shares the same state dict by reference + assert captured_session.state is parent_session.state assert captured_session.state["data"] == "shared" return original_run(*args, **kwargs) @@ -1563,14 +1565,14 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any: ) ) - # After the child finishes, service_session_id is restored + # Parent's service_session_id is never mutated assert parent_session.service_session_id == "resp_parent_abc123" async def test_chat_agent_as_tool_propagate_session_restores_service_session_id_on_error( client: SupportsChatGetResponse, ) -> None: - """Test that service_session_id is restored even if the child agent raises.""" + """Test that parent's service_session_id is untouched even if the child agent raises.""" agent = Agent(client=client, name="SubAgent", description="Sub agent") tool = agent.as_tool(propagate_session=True) @@ -1591,12 +1593,12 @@ def failing_run(*args: Any, **kwargs: Any) -> Any: ) ) - # service_session_id must be restored even after failure + # Parent's service_session_id is never mutated — child has its own session assert parent_session.service_session_id == "resp_parent_xyz789" async def test_chat_agent_as_tool_propagate_session_no_service_session_id(client: SupportsChatGetResponse) -> None: - """Test that when service_session_id is None, no save/restore is needed.""" + """Test that child setting service_session_id does not leak back to the parent.""" agent = Agent(client=client, name="SubAgent", description="Sub agent") tool = agent.as_tool(propagate_session=True) @@ -1610,6 +1612,8 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any: nonlocal captured_session captured_session = kwargs.get("session") assert captured_session.service_session_id is None + # Simulate the child's run populating service_session_id + captured_session.service_session_id = "resp_child_leaked" return original_run(*args, **kwargs) agent.run = capturing_run # type: ignore[assignment, method-assign] @@ -1622,6 +1626,7 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any: ) ) + # The child's service_session_id must not leak back to the parent assert parent_session.service_session_id is None From 9eeb6e1b08197abf08d744208d877f3f24d27e13 Mon Sep 17 00:00:00 2001 From: Benke Qu Date: Wed, 8 Jul 2026 08:06:59 -0700 Subject: [PATCH 3/5] fix: update test_chat_agent_as_tool_propagate_session_true for child session isolation The existing test asserted captured_session is parent_session, but since we now create a separate child AgentSession (to avoid racing under concurrent asyncio.gather), the child is a different object. Updated assertions to verify: - child is NOT the parent object (isolation) - child shares the same session_id and state dict (by reference) - child's service_session_id is None (isolated) --- python/packages/core/tests/core/test_agents.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/python/packages/core/tests/core/test_agents.py b/python/packages/core/tests/core/test_agents.py index a4d7182249..b97308e927 100644 --- a/python/packages/core/tests/core/test_agents.py +++ b/python/packages/core/tests/core/test_agents.py @@ -1481,10 +1481,13 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any: ) ) - assert captured_session is parent_session - assert captured_session is not None + # Child receives a separate AgentSession (not the parent object) to isolate + # service_session_id, but shares the same state dict and session_id. + assert captured_session is not parent_session assert captured_session.session_id == "parent-session-123" + assert captured_session.state is parent_session.state assert captured_session.state["shared_key"] == "shared_value" + assert captured_session.service_session_id is None async def test_chat_agent_as_tool_propagate_session_false_by_default(client: SupportsChatGetResponse) -> None: From d291af19ad3cce4b9ee63eee67c2b2e76ee0fcde Mon Sep 17 00:00:00 2001 From: Benke Qu Date: Thu, 9 Jul 2026 10:57:31 -0700 Subject: [PATCH 4/5] fix: add type narrowing asserts for captured_session Add 'assert captured_session is not None' before attribute access to satisfy mypy/pyright type checking on Optional values. --- python/packages/core/tests/core/test_agents.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/python/packages/core/tests/core/test_agents.py b/python/packages/core/tests/core/test_agents.py index 47ce0040ea..cdfe4739f4 100644 --- a/python/packages/core/tests/core/test_agents.py +++ b/python/packages/core/tests/core/test_agents.py @@ -1739,6 +1739,7 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any: # Child receives a separate AgentSession (not the parent object) to isolate # service_session_id, but shares the same state dict and session_id. + assert captured_session is not None assert captured_session is not parent_session assert captured_session.session_id == "parent-session-123" assert captured_session.state is parent_session.state @@ -1821,6 +1822,7 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any: nonlocal captured_session captured_session = kwargs.get("session") # The child gets a different session object with isolated service_session_id + assert captured_session is not None assert captured_session is not parent_session assert captured_session.service_session_id is None # But shares the same state dict by reference @@ -1884,6 +1886,7 @@ async def test_chat_agent_as_tool_propagate_session_no_service_session_id(client def capturing_run(*args: Any, **kwargs: Any) -> Any: nonlocal captured_session captured_session = kwargs.get("session") + assert captured_session is not None assert captured_session.service_session_id is None # Simulate the child's run populating service_session_id captured_session.service_session_id = "resp_child_leaked" From 00fbba6c8ce948834873791d209e8a7f96d0e62b Mon Sep 17 00:00:00 2001 From: Evan Mattson Date: Fri, 10 Jul 2026 14:37:57 +0900 Subject: [PATCH 5/5] Python: Fix test typing checks --- python/packages/core/tests/core/test_agents.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/python/packages/core/tests/core/test_agents.py b/python/packages/core/tests/core/test_agents.py index cdfe4739f4..b443b1d2fc 100644 --- a/python/packages/core/tests/core/test_agents.py +++ b/python/packages/core/tests/core/test_agents.py @@ -1830,7 +1830,7 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any: assert captured_session.state["data"] == "shared" return original_run(*args, **kwargs) - agent.run = capturing_run # type: ignore[assignment, method-assign] + agent.run = capturing_run # type: ignore[assignment, method-assign] # ty: ignore[invalid-assignment] await tool.invoke( context=FunctionInvocationContext( @@ -1857,7 +1857,7 @@ async def test_chat_agent_as_tool_propagate_session_restores_service_session_id_ def failing_run(*args: Any, **kwargs: Any) -> Any: raise RuntimeError("Child agent failed") - agent.run = failing_run # type: ignore[assignment, method-assign] + agent.run = failing_run # type: ignore[assignment, method-assign] # ty: ignore[invalid-assignment] with raises(RuntimeError, match="Child agent failed"): await tool.invoke( @@ -1892,7 +1892,7 @@ def capturing_run(*args: Any, **kwargs: Any) -> Any: captured_session.service_session_id = "resp_child_leaked" return original_run(*args, **kwargs) - agent.run = capturing_run # type: ignore[assignment, method-assign] + agent.run = capturing_run # type: ignore[assignment, method-assign] # ty: ignore[invalid-assignment] await tool.invoke( context=FunctionInvocationContext(