fix(realtime): address Codex review for input guardrails

Snapshot the active agent and its input guardrails when the transcription event is handled so a concurrent update_agent()/handoff cannot run a different agent's guardrails, and run the input guardrails concurrently so a slow guardrail cannot delay the forced response cancel.

Author: Skyline-9

src/agents/realtime/session.py

@@ -17,6 +17,7 @@
 )
 from ..agent import Agent
 from ..exceptions import ToolInputGuardrailTripwireTriggered, UserError
+from ..guardrail import InputGuardrail, InputGuardrailResult
 from ..handoffs import Handoff
 from ..items import ToolApprovalItem
 from ..logger import logger
@@ -1270,37 +1271,32 @@ async def _run_output_guardrails(self, text: str, response_id: str) -> bool:
 
         return False
 
-    async def _run_input_guardrails(self, text: str, item_id: str) -> bool:
+    async def _run_input_guardrails(
+        self,
+        text: str,
+        item_id: str,
+        agent: RealtimeAgent,
+        input_guardrails: list[InputGuardrail[Any]],
+    ) -> bool:
         """Run input guardrails on the user's transcribed input. Returns True if any guardrail was
         triggered.
-        """
-        combined_guardrails = self._current_agent.input_guardrails + self._run_config.get(
-            "input_guardrails", []
-        )
-        seen_ids: set[int] = set()
-        input_guardrails = []
-        for guardrail in combined_guardrails:
-            guardrail_id = id(guardrail)
-            if guardrail_id not in seen_ids:
-                input_guardrails.append(guardrail)
-                seen_ids.add(guardrail_id)
 
+        ``agent`` and ``input_guardrails`` are snapshotted when the transcription event is handled
+        so that a concurrent ``update_agent()`` or handoff cannot swap in a different agent's
+        guardrails before this background task runs.
+        """
         # If we've already interrupted the response for this user item, skip.
         if not input_guardrails or item_id in self._interrupted_input_item_ids:
             return False
 
-        triggered_results = []
-
-        for guardrail in input_guardrails:
+        async def _run_one(guardrail: InputGuardrail[Any]) -> InputGuardrailResult | None:
             try:
-                result = await guardrail.run(
+                return await guardrail.run(
                     # TODO (rm) Remove this cast, it's wrong
-                    cast(Agent[Any], self._current_agent),
+                    cast(Agent[Any], agent),
                     text,
                     self._context_wrapper,
                 )
-                if result.output.tripwire_triggered:
-                    triggered_results.append(result)
             except Exception as exc:
                 logger.warning(
                     "Input guardrail %r raised %s: %s; skipping it.",
@@ -1309,7 +1305,14 @@ async def _run_input_guardrails(self, text: str, item_id: str) -> bool:
                     exc,
                 )
                 logger.debug("Input guardrail failure details.", exc_info=True)
-                continue
+                return None
+
+        # Run the guardrails concurrently so a slow guardrail cannot delay the forced cancel behind
+        # unrelated guardrails, which would let the unsafe turn keep generating.
+        results = await asyncio.gather(*(_run_one(guardrail) for guardrail in input_guardrails))
+        triggered_results = [
+            result for result in results if result is not None and result.output.tripwire_triggered
+        ]
 
         if triggered_results:
             # Double-check: bail if already interrupted for this user item.
@@ -1353,14 +1356,27 @@ def _enqueue_guardrail_task(self, text: str, response_id: str) -> None:
         task.add_done_callback(self._on_guardrail_task_done)
 
     def _enqueue_input_guardrail_task(self, text: str, item_id: str) -> None:
+        # Snapshot the active agent and its guardrails now; a later update_agent()/handoff must not
+        # change which guardrails run against this transcript.
+        agent = self._current_agent
+        combined_guardrails = agent.input_guardrails + self._run_config.get("input_guardrails", [])
+
+        seen_ids: set[int] = set()
+        input_guardrails: list[InputGuardrail[Any]] = []
+        for guardrail in combined_guardrails:
+            guardrail_id = id(guardrail)
+            if guardrail_id not in seen_ids:
+                input_guardrails.append(guardrail)
+                seen_ids.add(guardrail_id)
+
         # Skip creating a no-op task when no input guardrails are configured.
-        if not self._current_agent.input_guardrails and not self._run_config.get(
-            "input_guardrails"
-        ):
+        if not input_guardrails:
             return
 
         # Runs the input guardrails in a separate task to avoid blocking the main loop.
-        task = asyncio.create_task(self._run_input_guardrails(text, item_id))
+        task = asyncio.create_task(
+            self._run_input_guardrails(text, item_id, agent, input_guardrails)
+        )
         # Reuse the shared guardrail task set + done callback so completed tasks are removed,
         # exceptions surface as events, and close() cancels any still-running task.
         self._guardrail_tasks.add(task)