Skip to content

fix(realtime): restore agent after update failure #4652

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

Open
hsusul wants to merge 2 commits into
base: main
Choose a base branch
from
+192 −11
Open

fix(realtime): restore agent after update failure #4652

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ea26f7e
fix(realtime): restore agent after update failure
hsusul Aug 25, 2026
000eaf1
fix(realtime): serialize agent update transactions
hsusul Aug 26, 2026
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
52 changes: 42 additions & 10 deletions src/agents/realtime/session.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,6 +239,7 @@ def __init__(
self._active_tool_invocations: dict[str, tuple[str, str, str]] = {}
self._pending_tool_outputs: dict[str, _PendingToolOutput] = {}
self._current_dispatch_snapshot: _RealtimeDispatchSnapshot | None = None
self._update_agent_lock = asyncio.Lock()

# Guardrails state tracking
self._interrupted_response_ids: set[str] = set()
Expand Down Expand Up @@ -378,18 +379,49 @@ async def interrupt(self) -> None:

async def update_agent(self, agent: RealtimeAgent) -> None:
"""Update the active agent for this session and apply its settings to the model."""
updated_settings = await self._get_updated_model_settings_from_agent(
starting_settings=None,
agent=agent,
)
updated_snapshot = self._dispatch_snapshot_from_settings(agent, updated_settings)
async with self._update_agent_lock:

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 26, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

P2 Badge Avoid deadlocking reentrant agent updates

When async_tool_calls=False and a custom or ScriptedRealtimeModel emits a function call while processing this session-update send, a tool that invokes the public update_agent() API blocks on this lock. The original update is simultaneously awaiting send_event(), which cannot finish delivering the function call until that tool returns, so both tasks deadlock; serialize transitions without holding a non-reentrant lock across listener delivery.

AGENTS.md reference: AGENTS.md:L149-L149

Useful? React with 👍 / 👎.

@FU-max-boop FU-max-boop Aug 26, 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.

Thanks for the follow-up at exact head 000eaf1. I independently reproduced the requested two-failure and repeated-cancellation send-success/send-failure cases; those now settle correctly on Python 3.10 and 3.13. The full realtime session file is also green (203 tests on each version), along with Ruff, mypy, and Pyright for the changed files.

I reproduced this reentrancy issue with a cleanup-safe public-path probe: async_tool_calls=False, a ScriptedRealtimeModel emits a function call while handling the outer session-update send, and the tool awaits session.update_agent(nested). The outer update holds _update_agent_lock while send_event() awaits listener delivery, and the nested update waits on the same lock.

One caution from prototyping: a ContextVar logical owner is not sufficient by itself. A detached tool task inherits it and can bypass serialization, then overlap a third independent update after the outer transaction releases the lock. Making the nested send inline preserves Scripted transport reentrancy, but a cancelled nested public update can then return before its wire operation settles. Moving that send back to a shielded child task restores cancellation semantics but reintroduces the Scripted delivery-worker deadlock.

I suggest freezing the narrow protocol before adding more owner state: serialize independent outer updates; separate outbound wire commit/failure from synchronous listener completion (or explicitly register listener-originated transitions); shield and drain every outer or nested send before commit/rollback; and enqueue inherited background work as an independent transaction. The identity fence should remain while handoff still mutates the same state outside this protocol. I have deterministic probes for all three boundaries and can contribute them once the intended transport/session contract is agreed.

updated_settings = await self._get_updated_model_settings_from_agent(
starting_settings=None,
agent=agent,
)
updated_snapshot = self._dispatch_snapshot_from_settings(agent, updated_settings)

self._current_agent = agent
self._current_dispatch_snapshot = updated_snapshot
previous_agent = self._current_agent
previous_snapshot = self._current_dispatch_snapshot
self._current_agent = agent
self._current_dispatch_snapshot = updated_snapshot

await self._model.send_event(
RealtimeModelSendSessionUpdate(session_settings=updated_settings)
)
send_task = asyncio.create_task(
self._model.send_event(
RealtimeModelSendSessionUpdate(session_settings=updated_settings)
)
)
try:
await asyncio.shield(send_task)
except asyncio.CancelledError:
while not send_task.done():
try:
await asyncio.shield(send_task)
except asyncio.CancelledError:
continue
except BaseException:
break
if send_task.cancelled() or send_task.exception() is not None:
if (
self._current_agent is agent
and self._current_dispatch_snapshot is updated_snapshot
):
self._current_agent = previous_agent
self._current_dispatch_snapshot = previous_snapshot
raise
except BaseException:
if (
self._current_agent is agent
and self._current_dispatch_snapshot is updated_snapshot
):
self._current_agent = previous_agent
self._current_dispatch_snapshot = previous_snapshot
raise

def _reconcile_output_response(self, response_id: str) -> None:
if self._active_output_response_generation is None:
Expand Down
151 changes: 150 additions & 1 deletion tests/realtime/test_session.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@
_PendingToolOutputSendError,
_serialize_tool_output,
)
from agents.realtime.testing import RealtimeConnectCall, ScriptedRealtimeModel
from agents.realtime.testing import RealtimeConnectCall, RealtimeStep, ScriptedRealtimeModel
from agents.run_context import RunContextWrapper
from agents.tool import FunctionTool, function_tool, tool_namespace
from agents.tool_context import ToolContext
Expand Down Expand Up @@ -6177,6 +6177,155 @@ async def test_update_agent_validation_failure_keeps_current_agent(self, mock_mo
assert session._current_agent is first_agent
assert mock_model.sent_events == ()

@pytest.mark.asyncio
async def test_update_agent_send_failure_keeps_current_agent(self):
first_agent = RealtimeAgent(name="first", instructions="first")
second_agent = RealtimeAgent(name="second", instructions="second")
model = ScriptedRealtimeModel(
steps=[
RealtimeStep(
expect=RealtimeModelSendSessionUpdate,
error=RuntimeError("send failed"),
)
]
)
session = RealtimeSession(model, first_agent, None)

async with session:
with pytest.raises(RuntimeError, match="send failed"):
await session.update_agent(second_agent)

assert session._current_agent is first_agent
assert session._current_dispatch_snapshot is not None
assert session._current_dispatch_snapshot.agent is first_agent

@pytest.mark.asyncio
async def test_overlapping_failed_updates_restore_last_committed_agent(self, mock_model):
first_agent = RealtimeAgent(name="first", instructions="first")
first_failing_agent = RealtimeAgent(name="first-failing", instructions="first-failing")
second_failing_agent = RealtimeAgent(name="second-failing", instructions="second-failing")
first_send_started = asyncio.Event()
second_send_started = asyncio.Event()
release_first_send = asyncio.Event()
release_second_send = asyncio.Event()
send_count = 0

async def send_event(_event):
nonlocal send_count
send_count += 1
if send_count == 1:
first_send_started.set()
await release_first_send.wait()
raise RuntimeError("first send failed")
second_send_started.set()
await release_second_send.wait()
raise RuntimeError("second send failed")

mock_model.send_event = send_event
session = RealtimeSession(mock_model, first_agent, None)

async with session:
first_update = asyncio.create_task(session.update_agent(first_failing_agent))
await first_send_started.wait()
second_update = asyncio.create_task(session.update_agent(second_failing_agent))
for _ in range(10):
await asyncio.sleep(0)

assert not second_send_started.is_set()

release_first_send.set()
with pytest.raises(RuntimeError, match="first send failed"):
await first_update

await second_send_started.wait()
release_second_send.set()
with pytest.raises(RuntimeError, match="second send failed"):
await second_update

assert session._current_agent is first_agent
assert session._current_dispatch_snapshot is not None
assert session._current_dispatch_snapshot.agent is first_agent

@pytest.mark.asyncio
async def test_failed_update_agent_does_not_rollback_handoff_owned_state(self, mock_model):
first_agent = RealtimeAgent(name="first", instructions="first")
updated_agent = RealtimeAgent(name="updated", instructions="updated")
handed_off_agent = RealtimeAgent(name="handed-off", instructions="handed-off")
send_started = asyncio.Event()
release_send = asyncio.Event()

async def send_event(_event):
send_started.set()
await release_send.wait()
raise RuntimeError("send failed")

mock_model.send_event = send_event
session = RealtimeSession(mock_model, first_agent, None)

async with session:
update = asyncio.create_task(session.update_agent(updated_agent))
await send_started.wait()

handoff_settings = await session._get_updated_model_settings_from_agent(
starting_settings=None,
agent=handed_off_agent,
)
handoff_snapshot = session._dispatch_snapshot_from_settings(
handed_off_agent, handoff_settings
)
session._current_agent = handed_off_agent
session._current_dispatch_snapshot = handoff_snapshot

release_send.set()
with pytest.raises(RuntimeError, match="send failed"):
await update

assert session._current_agent is handed_off_agent
assert session._current_dispatch_snapshot is handoff_snapshot

@pytest.mark.asyncio
@pytest.mark.parametrize("send_fails", [False, True])
async def test_repeatedly_cancelled_update_agent_waits_for_send_to_settle(
self, mock_model, send_fails
):
first_agent = RealtimeAgent(name="first", instructions="first")
second_agent = RealtimeAgent(name="second", instructions="second")
send_started = asyncio.Event()
release_send = asyncio.Event()
send_cancelled = asyncio.Event()

async def send_event(_event):
send_started.set()
try:
await release_send.wait()
except asyncio.CancelledError:
send_cancelled.set()
raise
if send_fails:
raise RuntimeError("send failed")

mock_model.send_event = send_event
session = RealtimeSession(mock_model, first_agent, None)

async with session:
update = asyncio.create_task(session.update_agent(second_agent))
await send_started.wait()
update.cancel()
await asyncio.sleep(0)
update.cancel()
await asyncio.sleep(0)

assert not update.done()
assert not send_cancelled.is_set()

release_send.set()
with pytest.raises(asyncio.CancelledError):
await update

assert session._current_agent is (first_agent if send_fails else second_agent)
assert session._current_dispatch_snapshot is not None
assert session._current_dispatch_snapshot.agent is session._current_agent


class TestTranscriptPreservation:
"""Tests ensuring assistant transcripts are preserved across updates."""
Expand Down
Loading