Merge branch 'main' into add-dapr-session

Author: filintod

src/agents/extensions/memory/sqlalchemy_session.py

@@ -319,3 +319,16 @@ async def clear_session(self) -> None:
                 await sess.execute(
                     delete(self._sessions).where(self._sessions.c.session_id == self.session_id)
                 )
+
+    @property
+    def engine(self) -> AsyncEngine:
+        """Access the underlying SQLAlchemy AsyncEngine.
+
+        This property provides direct access to the engine for advanced use cases,
+        such as checking connection pool status, configuring engine settings,
+        or manually disposing the engine when needed.
+
+        Returns:
+            AsyncEngine: The SQLAlchemy async engine instance.
+        """
+        return self._engine

src/agents/extensions/models/litellm_model.py

@@ -44,6 +44,7 @@
 from ...models.chatcmpl_stream_handler import ChatCmplStreamHandler
 from ...models.fake_id import FAKE_RESPONSES_ID
 from ...models.interface import Model, ModelTracing
+from ...models.openai_responses import Converter as OpenAIResponsesConverter
 from ...tool import Tool
 from ...tracing import generation_span
 from ...tracing.span_data import GenerationSpanData
@@ -367,15 +368,19 @@ async def _fetch_response(
         if isinstance(ret, litellm.types.utils.ModelResponse):
             return ret
 
+        responses_tool_choice = OpenAIResponsesConverter.convert_tool_choice(
+            model_settings.tool_choice
+        )
+        if responses_tool_choice is None or responses_tool_choice is omit:
+            responses_tool_choice = "auto"
+
         response = Response(
             id=FAKE_RESPONSES_ID,
             created_at=time.time(),
             model=self.model,
             object="response",
             output=[],
-            tool_choice=cast(Literal["auto", "required", "none"], tool_choice)
-            if tool_choice is not omit
-            else "auto",
+            tool_choice=responses_tool_choice,  # type: ignore[arg-type]
             top_p=model_settings.top_p,
             temperature=model_settings.temperature,
             tools=[],

src/agents/realtime/model_inputs.py

@@ -95,6 +95,9 @@ class RealtimeModelSendToolOutput:
 class RealtimeModelSendInterrupt:
     """Send an interrupt to the model."""
 
+    force_response_cancel: bool = False
+    """Force sending a response.cancel event even if automatic cancellation is enabled."""
+
 
 @dataclass
 class RealtimeModelSendSessionUpdate:

src/agents/realtime/openai_realtime.py

@@ -395,36 +395,36 @@ async def _send_interrupt(self, event: RealtimeModelSendInterrupt) -> None:
         current_item_id = playback_state.get("current_item_id")
         current_item_content_index = playback_state.get("current_item_content_index")
         elapsed_ms = playback_state.get("elapsed_ms")
+
         if current_item_id is None or elapsed_ms is None:
             logger.debug(
                 "Skipping interrupt. "
                 f"Item id: {current_item_id}, "
                 f"elapsed ms: {elapsed_ms}, "
                 f"content index: {current_item_content_index}"
             )
-            return
-
-        current_item_content_index = current_item_content_index or 0
-        if elapsed_ms > 0:
-            await self._emit_event(
-                RealtimeModelAudioInterruptedEvent(
-                    item_id=current_item_id,
-                    content_index=current_item_content_index,
-                )
-            )
-            converted = _ConversionHelper.convert_interrupt(
-                current_item_id,
-                current_item_content_index,
-                int(elapsed_ms),
-            )
-            await self._send_raw_message(converted)
         else:
-            logger.debug(
-                "Didn't interrupt bc elapsed ms is < 0. "
-                f"Item id: {current_item_id}, "
-                f"elapsed ms: {elapsed_ms}, "
-                f"content index: {current_item_content_index}"
-            )
+            current_item_content_index = current_item_content_index or 0
+            if elapsed_ms > 0:
+                await self._emit_event(
+                    RealtimeModelAudioInterruptedEvent(
+                        item_id=current_item_id,
+                        content_index=current_item_content_index,
+                    )
+                )
+                converted = _ConversionHelper.convert_interrupt(
+                    current_item_id,
+                    current_item_content_index,
+                    int(elapsed_ms),
+                )
+                await self._send_raw_message(converted)
+            else:
+                logger.debug(
+                    "Didn't interrupt bc elapsed ms is < 0. "
+                    f"Item id: {current_item_id}, "
+                    f"elapsed ms: {elapsed_ms}, "
+                    f"content index: {current_item_content_index}"
+                )
 
         session = self._created_session
         automatic_response_cancellation_enabled = (
@@ -434,12 +434,16 @@ async def _send_interrupt(self, event: RealtimeModelSendInterrupt) -> None:
             and session.audio.input.turn_detection is not None
             and session.audio.input.turn_detection.interrupt_response is True
         )
-        if not automatic_response_cancellation_enabled:
+        should_cancel_response = event.force_response_cancel or (
+            not automatic_response_cancellation_enabled
+        )
+        if should_cancel_response:
             await self._cancel_response()
 
-        self._audio_state_tracker.on_interrupted()
-        if self._playback_tracker:
-            self._playback_tracker.on_interrupted()
+        if current_item_id is not None and elapsed_ms is not None:
+            self._audio_state_tracker.on_interrupted()
+            if self._playback_tracker:
+                self._playback_tracker.on_interrupted()
 
     async def _send_session_update(self, event: RealtimeModelSendSessionUpdate) -> None:
         """Send a session update to the model."""

src/agents/realtime/session.py

@@ -704,7 +704,7 @@ async def _run_output_guardrails(self, text: str, response_id: str) -> bool:
             )
 
             # Interrupt the model
-            await self._model.send_event(RealtimeModelSendInterrupt())
+            await self._model.send_event(RealtimeModelSendInterrupt(force_response_cancel=True))
 
             # Send guardrail triggered message
             guardrail_names = [result.guardrail.get_name() for result in triggered_results]

src/agents/run.py

@@ -1138,6 +1138,15 @@ async def _start_streaming(
 
             streamed_result.is_complete = True
         finally:
+            if streamed_result._input_guardrails_task:
+                try:
+                    await AgentRunner._input_guardrail_tripwire_triggered_for_stream(
+                        streamed_result
+                    )
+                except Exception as e:
+                    logger.debug(
+                        f"Error in streamed_result finalize for agent {current_agent.name} - {e}"
+                    )
             if current_span:
                 current_span.finish(reset_current=True)
             if streamed_result.trace:

tests/extensions/memory/test_sqlalchemy_session.py

@@ -14,6 +14,7 @@
     Summary,
 )
 from sqlalchemy import select, text, update
+from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
 from sqlalchemy.sql import Select
 
 pytest.importorskip("sqlalchemy")  # Skip tests if SQLAlchemy is not installed
@@ -390,3 +391,56 @@ async def recording_execute(statement: Any, *args: Any, **kwargs: Any) -> Any:
 
     assert _item_ids(retrieved_full) == ["rs_first", "msg_second"]
     assert _item_ids(retrieved_limited) == ["rs_first", "msg_second"]
+
+
+async def test_engine_property_from_url():
+    """Test that the engine property returns the AsyncEngine from from_url."""
+    session_id = "engine_property_test"
+    session = SQLAlchemySession.from_url(session_id, url=DB_URL, create_tables=True)
+
+    # Verify engine property returns an AsyncEngine instance
+    assert isinstance(session.engine, AsyncEngine)
+
+    # Verify we can use the engine for advanced operations
+    # For example, check pool status
+    assert session.engine.pool is not None
+
+    # Verify we can manually dispose the engine
+    await session.engine.dispose()
+
+
+async def test_engine_property_from_external_engine():
+    """Test that the engine property returns the external engine."""
+    session_id = "external_engine_test"
+
+    # Create engine externally
+    external_engine = create_async_engine(DB_URL)
+
+    # Create session with external engine
+    session = SQLAlchemySession(session_id, engine=external_engine, create_tables=True)
+
+    # Verify engine property returns the same engine instance
+    assert session.engine is external_engine
+
+    # Verify we can use the engine
+    assert isinstance(session.engine, AsyncEngine)
+
+    # Clean up - user is responsible for disposing external engine
+    await external_engine.dispose()
+
+
+async def test_engine_property_is_read_only():
+    """Test that the engine property cannot be modified."""
+    session_id = "readonly_engine_test"
+    session = SQLAlchemySession.from_url(session_id, url=DB_URL, create_tables=True)
+
+    # Verify engine property exists
+    assert hasattr(session, "engine")
+
+    # Verify it's a property (read-only, cannot be set)
+    # Type ignore needed because mypy correctly detects this is read-only
+    with pytest.raises(AttributeError):
+        session.engine = create_async_engine(DB_URL)  # type: ignore[misc]
+
+    # Clean up
+    await session.engine.dispose()

tests/realtime/test_openai_realtime.py

@@ -1,4 +1,5 @@
 import json
+from types import SimpleNamespace
 from typing import Any, cast
 from unittest.mock import AsyncMock, Mock, patch
 
@@ -509,6 +510,59 @@ async def test_send_event_dispatch(self, model, monkeypatch):
         # session update -> 1
         assert send_raw.await_count == 8
 
+    @pytest.mark.asyncio
+    async def test_interrupt_force_cancel_overrides_auto_cancellation(self, model, monkeypatch):
+        """Interrupt should send response.cancel even when auto cancel is enabled."""
+        model._audio_state_tracker.set_audio_format("pcm16")
+        model._audio_state_tracker.on_audio_delta("item_1", 0, b"\x00" * 4800)
+        model._ongoing_response = True
+        model._created_session = SimpleNamespace(
+            audio=SimpleNamespace(
+                input=SimpleNamespace(
+                    turn_detection=SimpleNamespace(interrupt_response=True)
+                )
+            )
+        )
+
+        send_raw = AsyncMock()
+        emit_event = AsyncMock()
+        monkeypatch.setattr(model, "_send_raw_message", send_raw)
+        monkeypatch.setattr(model, "_emit_event", emit_event)
+
+        await model._send_interrupt(RealtimeModelSendInterrupt(force_response_cancel=True))
+
+        assert send_raw.await_count == 2
+        payload_types = {call.args[0].type for call in send_raw.call_args_list}
+        assert payload_types == {"conversation.item.truncate", "response.cancel"}
+        assert model._ongoing_response is False
+        assert model._audio_state_tracker.get_last_audio_item() is None
+
+    @pytest.mark.asyncio
+    async def test_interrupt_respects_auto_cancellation_when_not_forced(self, model, monkeypatch):
+        """Interrupt should avoid sending response.cancel when relying on automatic cancellation."""
+        model._audio_state_tracker.set_audio_format("pcm16")
+        model._audio_state_tracker.on_audio_delta("item_1", 0, b"\x00" * 4800)
+        model._ongoing_response = True
+        model._created_session = SimpleNamespace(
+            audio=SimpleNamespace(
+                input=SimpleNamespace(
+                    turn_detection=SimpleNamespace(interrupt_response=True)
+                )
+            )
+        )
+
+        send_raw = AsyncMock()
+        emit_event = AsyncMock()
+        monkeypatch.setattr(model, "_send_raw_message", send_raw)
+        monkeypatch.setattr(model, "_emit_event", emit_event)
+
+        await model._send_interrupt(RealtimeModelSendInterrupt())
+
+        assert send_raw.await_count == 1
+        assert send_raw.call_args_list[0].args[0].type == "conversation.item.truncate"
+        assert all(call.args[0].type != "response.cancel" for call in send_raw.call_args_list)
+        assert model._ongoing_response is True
+
     def test_add_remove_listener_and_tools_conversion(self, model):
         listener = AsyncMock()
         model.add_listener(listener)