fix: improve run_sync's python 3.14 compatibility (openai#2006)

Author: seratch

src/agents/run.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import inspect
 import os
+import warnings
 from dataclasses import dataclass, field
 from typing import Any, Callable, Generic, cast, get_args
 
@@ -720,19 +722,40 @@ def run_sync(
  conversation_id = kwargs.get("conversation_id")
  session = kwargs.get("session")
 
- # Python 3.14 no longer creates a default loop implicitly, so we inspect the running loop.
+ # Python 3.14 stopped implicitly wiring up a default event loop
+ # when synchronous code touches asyncio APIs for the first time.
+ # Several of our synchronous entry points (for example the Redis/SQLAlchemy session helpers)
+ # construct asyncio primitives like asyncio.Lock during __init__,
+ # which binds them to whatever loop happens to be the thread's default at that moment.
+ # To keep those locks usable we must ensure that run_sync reuses that same default loop
+ # instead of hopping over to a brand-new asyncio.run() loop.
  try:
- loop = asyncio.get_running_loop()
+ already_running_loop = asyncio.get_running_loop()
  except RuntimeError:
- loop = None
+ already_running_loop = None
 
- if loop is not None:
+ if already_running_loop is not None:
  # This method is only expected to run when no loop is already active.
+ # (Each thread has its own default loop; concurrent sync runs should happen on
+ # different threads. In a single thread use the async API to interleave work.)
  raise RuntimeError(
  "AgentRunner.run_sync() cannot be called when an event loop is already running."
  )
 
- return asyncio.run(
+ policy = asyncio.get_event_loop_policy()
+ with warnings.catch_warnings():
+ warnings.simplefilter("ignore", DeprecationWarning)
+ try:
+ default_loop = policy.get_event_loop()
+ except RuntimeError:
+ default_loop = policy.new_event_loop()
+ policy.set_event_loop(default_loop)
+
+ # We intentionally leave the default loop open even if we had to create one above. Session
+ # instances and other helpers stash loop-bound primitives between calls and expect to find
+ # the same default loop every time run_sync is invoked on this thread.
+ # Schedule the async run on the default loop so that we can manage cancellation explicitly.
+ task = default_loop.create_task(
  self.run(
  starting_agent,
  input,
@@ -746,6 +769,24 @@ def run_sync(
  )
  )
 
+ try:
+ # Drive the coroutine to completion, harvesting the final RunResult.
+ return default_loop.run_until_complete(task)
+ except BaseException:
+ # If the sync caller aborts (KeyboardInterrupt, etc.), make sure the scheduled task
+ # does not linger on the shared loop by cancelling it and waiting for completion.
+ if not task.done():
+ task.cancel()
+ with contextlib.suppress(asyncio.CancelledError):
+ default_loop.run_until_complete(task)
+ raise
+ finally:
+ if not default_loop.is_closed():
+ # The loop stays open for subsequent runs, but we still need to flush any pending
+ # async generators so their cleanup code executes promptly.
+ with contextlib.suppress(RuntimeError):
+ default_loop.run_until_complete(default_loop.shutdown_asyncgens())
+
  def run_streamed(
  self,
  starting_agent: Agent[TContext],

tests/test_agent_runner_sync.py

@@ -0,0 +1,150 @@
+import asyncio
+from collections.abc import Generator
+from typing import Any
+
+import pytest
+
+from agents.agent import Agent
+from agents.run import AgentRunner
+
+
+@pytest.fixture
+def fresh_event_loop_policy() -> Generator[asyncio.AbstractEventLoopPolicy, None, None]:
+ policy_before = asyncio.get_event_loop_policy()
+ new_policy = asyncio.DefaultEventLoopPolicy()
+ asyncio.set_event_loop_policy(new_policy)
+ try:
+ yield new_policy
+ finally:
+ asyncio.set_event_loop_policy(policy_before)
+
+
+def test_run_sync_reuses_existing_default_loop(monkeypatch, fresh_event_loop_policy):
+ runner = AgentRunner()
+ observed_loops: list[asyncio.AbstractEventLoop] = []
+
+ async def fake_run(self, *_args, **_kwargs):
+ observed_loops.append(asyncio.get_running_loop())
+ return object()
+
+ monkeypatch.setattr(AgentRunner, "run", fake_run, raising=False)
+
+ test_loop = asyncio.new_event_loop()
+ fresh_event_loop_policy.set_event_loop(test_loop)
+
+ try:
+ runner.run_sync(Agent(name="test-agent"), "input")
+ assert observed_loops and observed_loops[0] is test_loop
+ finally:
+ fresh_event_loop_policy.set_event_loop(None)
+ test_loop.close()
+
+
+def test_run_sync_creates_default_loop_when_missing(monkeypatch, fresh_event_loop_policy):
+ runner = AgentRunner()
+ observed_loops: list[asyncio.AbstractEventLoop] = []
+
+ async def fake_run(self, *_args, **_kwargs):
+ observed_loops.append(asyncio.get_running_loop())
+ return object()
+
+ monkeypatch.setattr(AgentRunner, "run", fake_run, raising=False)
+
+ fresh_event_loop_policy.set_event_loop(None)
+
+ runner.run_sync(Agent(name="test-agent"), "input")
+ created_loop = observed_loops[0]
+ assert created_loop is fresh_event_loop_policy.get_event_loop()
+
+ fresh_event_loop_policy.set_event_loop(None)
+ created_loop.close()
+
+
+def test_run_sync_errors_when_loop_already_running(monkeypatch, fresh_event_loop_policy):
+ runner = AgentRunner()
+
+ async def fake_run(self, *_args, **_kwargs):
+ return object()
+
+ monkeypatch.setattr(AgentRunner, "run", fake_run, raising=False)
+
+ async def invoke():
+ with pytest.raises(RuntimeError):
+ runner.run_sync(Agent(name="test-agent"), "input")
+
+ asyncio.run(invoke())
+
+
+def test_run_sync_cancels_task_when_interrupted(monkeypatch, fresh_event_loop_policy):
+ runner = AgentRunner()
+
+ async def fake_run(self, *_args, **_kwargs):
+ await asyncio.sleep(3600)
+
+ monkeypatch.setattr(AgentRunner, "run", fake_run, raising=False)
+
+ test_loop = asyncio.new_event_loop()
+ fresh_event_loop_policy.set_event_loop(test_loop)
+
+ created_tasks: list[asyncio.Task[Any]] = []
+ original_create_task = test_loop.create_task
+
+ def capturing_create_task(coro):
+ task = original_create_task(coro)
+ created_tasks.append(task)
+ return task
+
+ original_run_until_complete = test_loop.run_until_complete
+ call_count = {"value": 0}
+
+ def interrupt_once(future):
+ call_count["value"] += 1
+ if call_count["value"] == 1:
+ raise KeyboardInterrupt()
+ return original_run_until_complete(future)
+
+ monkeypatch.setattr(test_loop, "create_task", capturing_create_task)
+ monkeypatch.setattr(test_loop, "run_until_complete", interrupt_once)
+
+ try:
+ with pytest.raises(KeyboardInterrupt):
+ runner.run_sync(Agent(name="test-agent"), "input")
+
+ assert created_tasks, "Expected run_sync to schedule a task."
+ assert created_tasks[0].done()
+ assert created_tasks[0].cancelled()
+ assert call_count["value"] >= 2
+ finally:
+ monkeypatch.undo()
+ fresh_event_loop_policy.set_event_loop(None)
+ test_loop.close()
+
+
+def test_run_sync_finalizes_async_generators(monkeypatch, fresh_event_loop_policy):
+ runner = AgentRunner()
+ cleanup_markers: list[str] = []
+
+ async def fake_run(self, *_args, **_kwargs):
+ async def agen():
+ try:
+ yield None
+ finally:
+ cleanup_markers.append("done")
+
+ gen = agen()
+ await gen.__anext__()
+ return "ok"
+
+ monkeypatch.setattr(AgentRunner, "run", fake_run, raising=False)
+
+ test_loop = asyncio.new_event_loop()
+ fresh_event_loop_policy.set_event_loop(test_loop)
+
+ try:
+ runner.run_sync(Agent(name="test-agent"), "input")
+ assert cleanup_markers == ["done"], (
+ "Async generators must be finalized after run_sync returns."
+ )
+ finally:
+ fresh_event_loop_policy.set_event_loop(None)
+ test_loop.close()