Skip to content

Python: Clear stale service session on full-history replay #7354

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
Henry Su (hsusul) wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Python: Clear stale service session on full-history replay #7354

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions python/packages/core/agent_framework/_workflows/_agent_executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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):

@moonbox3 Evan Mattson (moonbox3) Aug 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we avoid using the mere presence of a function_call as the full-history signal? Group-chat orchestrators broadcast each participant's response to every other agent as AgentExecutorRequest(..., should_respond=False) at _base_group_chat_orchestrator.py:433-436, and completed tool turns include their function calls. This therefore clears every receiving participant's valid service continuation after any tool-using turn, so its next turn loses server-held history and reasoning-model histories can fail stateless replay.

@moonbox3 Evan Mattson (moonbox3) Aug 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should full-history handling be shared with the other explicit-message handlers? from_messages at lines 283-294 accepts the same list[Message] history but goes straight to _run_agent_and_emit, so replaying a call/result transcript through that public handler still forwards the stale service session. OpenAI then strips the inline call under previous_response_id, leaving the request attached to the wrong server conversation, so the cross-conversation state corruption this PR addresses remains reachable.

self._session.service_session_id = None
Comment on lines +210 to +211

self._cache.extend(request.messages)

if request.should_respond:
Expand Down
40 changes: 30 additions & 10 deletions python/packages/core/tests/workflow/test_full_conversation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
Loading