40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138

@dataclassclassRunResultBase(abc.ABC):input:str|list[TResponseInputItem]"""The original input items i.e. the items before run() was called. This may be a mutated    version of the input, if there are handoff input filters that mutate the input.    """new_items:list[RunItem]"""The new items generated during the agent run. These include things like new messages, tool    calls and their outputs, etc.    """raw_responses:list[ModelResponse]"""The raw LLM responses generated by the model during the agent run."""final_output:Any"""The output of the last agent."""input_guardrail_results:list[InputGuardrailResult]"""Guardrail results for the input messages."""output_guardrail_results:list[OutputGuardrailResult]"""Guardrail results for the final output of the agent."""tool_input_guardrail_results:list[ToolInputGuardrailResult]"""Tool input guardrail results from all tools executed during the run."""tool_output_guardrail_results:list[ToolOutputGuardrailResult]"""Tool output guardrail results from all tools executed during the run."""context_wrapper:RunContextWrapper[Any]"""The context wrapper for the agent run."""@property@abc.abstractmethoddeflast_agent(self)->Agent[Any]:"""The last agent that was run."""defrelease_agents(self,*,release_new_items:bool=True)->None:"""        Release strong references to agents held by this result. After calling this method,        accessing `item.agent` or `last_agent` may return `None` if the agent has been garbage        collected. Callers can use this when they are done inspecting the result and want to        eagerly drop any associated agent graph.        """ifrelease_new_items:foriteminself.new_items:release=getattr(item,"release_agent",None)ifcallable(release):release()self._release_last_agent_reference()def__del__(self)->None:try:# Fall back to releasing agents automatically in case the caller never invoked# `release_agents()` explicitly so GC of the RunResult drops the last strong reference.# We pass `release_new_items=False` so RunItems that the user intentionally keeps# continue exposing their originating agent until that agent itself is collected.self.release_agents(release_new_items=False)exceptException:# Avoid raising from __del__.pass@abc.abstractmethoddef_release_last_agent_reference(self)->None:"""Release stored agent reference specific to the concrete result type."""deffinal_output_as(self,cls:type[T],raise_if_incorrect_type:bool=False)->T:"""A convenience method to cast the final output to a specific type. By default, the cast        is only for the typechecker. If you set `raise_if_incorrect_type` to True, we'll raise a        TypeError if the final output is not of the given type.        Args:            cls: The type to cast the final output to.            raise_if_incorrect_type: If True, we'll raise a TypeError if the final output is not of                the given type.        Returns:            The final output casted to the given type.        """ifraise_if_incorrect_typeandnotisinstance(self.final_output,cls):raiseTypeError(f"Final output is not of type{cls.__name__}")returncast(T,self.final_output)defto_input_list(self)->list[TResponseInputItem]:"""Creates a new input list, merging the original input with all the new items generated."""original_items:list[TResponseInputItem]=ItemHelpers.input_to_new_input_list(self.input)new_items=[item.to_input_item()foriteminself.new_items]returnoriginal_items+new_items@propertydeflast_response_id(self)->str|None:"""Convenience method to get the response ID of the last model response."""ifnotself.raw_responses:returnNonereturnself.raw_responses[-1].response_id

141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174

@dataclassclassRunResult(RunResultBase):_last_agent:Agent[Any]_last_agent_ref:weakref.ReferenceType[Agent[Any]]|None=field(init=False,repr=False,default=None,)def__post_init__(self)->None:self._last_agent_ref=weakref.ref(self._last_agent)@propertydeflast_agent(self)->Agent[Any]:"""The last agent that was run."""agent=cast("Agent[Any] | None",self.__dict__.get("_last_agent"))ifagentisnotNone:returnagentifself._last_agent_ref:agent=self._last_agent_ref()ifagentisnotNone:returnagentraiseAgentsException("Last agent reference is no longer available.")def_release_last_agent_reference(self)->None:agent=cast("Agent[Any] | None",self.__dict__.get("_last_agent"))ifagentisNone:returnself._last_agent_ref=weakref.ref(agent)# Preserve dataclass field so repr/asdict continue to succeed.self.__dict__["_last_agent"]=Nonedef__str__(self)->str:returnpretty_print_result(self)

177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424

@dataclassclassRunResultStreaming(RunResultBase):"""The result of an agent run in streaming mode. You can use the `stream_events` method to    receive semantic events as they are generated.    The streaming method will raise:    - A MaxTurnsExceeded exception if the agent exceeds the max_turns limit.    - A GuardrailTripwireTriggered exception if a guardrail is tripped.    """current_agent:Agent[Any]"""The current agent that is running."""current_turn:int"""The current turn number."""max_turns:int"""The maximum number of turns the agent can run for."""final_output:Any"""The final output of the agent. This is None until the agent has finished running."""_current_agent_output_schema:AgentOutputSchemaBase|None=field(repr=False)trace:Trace|None=field(repr=False)is_complete:bool=False"""Whether the agent has finished running."""_current_agent_ref:weakref.ReferenceType[Agent[Any]]|None=field(init=False,repr=False,default=None,)# Queues that the background run_loop writes to_event_queue:asyncio.Queue[StreamEvent|QueueCompleteSentinel]=field(default_factory=asyncio.Queue,repr=False)_input_guardrail_queue:asyncio.Queue[InputGuardrailResult]=field(default_factory=asyncio.Queue,repr=False)# Store the asyncio tasks that we're waiting on_run_impl_task:asyncio.Task[Any]|None=field(default=None,repr=False)_input_guardrails_task:asyncio.Task[Any]|None=field(default=None,repr=False)_output_guardrails_task:asyncio.Task[Any]|None=field(default=None,repr=False)_stored_exception:Exception|None=field(default=None,repr=False)# Soft cancel state_cancel_mode:Literal["none","immediate","after_turn"]=field(default="none",repr=False)def__post_init__(self)->None:self._current_agent_ref=weakref.ref(self.current_agent)@propertydeflast_agent(self)->Agent[Any]:"""The last agent that was run. Updates as the agent run progresses, so the true last agent        is only available after the agent run is complete.        """agent=cast("Agent[Any] | None",self.__dict__.get("current_agent"))ifagentisnotNone:returnagentifself._current_agent_ref:agent=self._current_agent_ref()ifagentisnotNone:returnagentraiseAgentsException("Last agent reference is no longer available.")def_release_last_agent_reference(self)->None:agent=cast("Agent[Any] | None",self.__dict__.get("current_agent"))ifagentisNone:returnself._current_agent_ref=weakref.ref(agent)# Preserve dataclass field so repr/asdict continue to succeed.self.__dict__["current_agent"]=Nonedefcancel(self,mode:Literal["immediate","after_turn"]="immediate")->None:"""Cancel the streaming run.        Args:            mode: Cancellation strategy:                - "immediate": Stop immediately, cancel all tasks, clear queues (default)                - "after_turn": Complete current turn gracefully before stopping                    * Allows LLM response to finish                    * Executes pending tool calls                    * Saves session state properly                    * Tracks usage accurately                    * Stops before next turn begins        Example:            ```python            result = Runner.run_streamed(agent, "Task", session=session)            async for event in result.stream_events():                if user_interrupted():                    result.cancel(mode="after_turn")  # Graceful                    # result.cancel()  # Immediate (default)            ```        Note: After calling cancel(), you should continue consuming stream_events()        to allow the cancellation to complete properly.        """# Store the cancel mode for the background task to checkself._cancel_mode=modeifmode=="immediate":# Existing behavior - immediate shutdownself._cleanup_tasks()# Cancel all running tasksself.is_complete=True# Mark the run as complete to stop event streaming# Optionally, clear the event queue to prevent processing stale eventswhilenotself._event_queue.empty():self._event_queue.get_nowait()whilenotself._input_guardrail_queue.empty():self._input_guardrail_queue.get_nowait()elifmode=="after_turn":# Soft cancel - just set the flag# The streaming loop will check this and stop gracefully# Don't call _cleanup_tasks() or clear queues yetpassasyncdefstream_events(self)->AsyncIterator[StreamEvent]:"""Stream deltas for new items as they are generated. We're using the types from the        OpenAI Responses API, so these are semantic events: each event has a `type` field that        describes the type of the event, along with the data for that event.        This will raise:        - A MaxTurnsExceeded exception if the agent exceeds the max_turns limit.        - A GuardrailTripwireTriggered exception if a guardrail is tripped.        """try:whileTrue:self._check_errors()ifself._stored_exception:logger.debug("Breaking due to stored exception")self.is_complete=Truebreakifself.is_completeandself._event_queue.empty():breaktry:item=awaitself._event_queue.get()exceptasyncio.CancelledError:breakifisinstance(item,QueueCompleteSentinel):# Await input guardrails if they are still running, so late# exceptions are captured.awaitself._await_task_safely(self._input_guardrails_task)self._event_queue.task_done()# Check for errors, in case the queue was completed# due to an exceptionself._check_errors()breakyielditemself._event_queue.task_done()finally:# Ensure main execution completes before cleanup to avoid race conditions# with session operationsawaitself._await_task_safely(self._run_impl_task)# Safely terminate all background tasks after main execution has finishedself._cleanup_tasks()ifself._stored_exception:raiseself._stored_exceptiondef_create_error_details(self)->RunErrorDetails:"""Return a `RunErrorDetails` object considering the current attributes of the class."""returnRunErrorDetails(input=self.input,new_items=self.new_items,raw_responses=self.raw_responses,last_agent=self.current_agent,context_wrapper=self.context_wrapper,input_guardrail_results=self.input_guardrail_results,output_guardrail_results=self.output_guardrail_results,)def_check_errors(self):ifself.current_turn>self.max_turns:max_turns_exc=MaxTurnsExceeded(f"Max turns ({self.max_turns}) exceeded")max_turns_exc.run_data=self._create_error_details()self._stored_exception=max_turns_exc# Fetch all the completed guardrail results from the queue and raise if neededwhilenotself._input_guardrail_queue.empty():guardrail_result=self._input_guardrail_queue.get_nowait()ifguardrail_result.output.tripwire_triggered:tripwire_exc=InputGuardrailTripwireTriggered(guardrail_result)tripwire_exc.run_data=self._create_error_details()self._stored_exception=tripwire_exc# Check the tasks for any exceptionsifself._run_impl_taskandself._run_impl_task.done():run_impl_exc=self._run_impl_task.exception()ifrun_impl_excandisinstance(run_impl_exc,Exception):ifisinstance(run_impl_exc,AgentsException)andrun_impl_exc.run_dataisNone:run_impl_exc.run_data=self._create_error_details()self._stored_exception=run_impl_excifself._input_guardrails_taskandself._input_guardrails_task.done():in_guard_exc=self._input_guardrails_task.exception()ifin_guard_excandisinstance(in_guard_exc,Exception):ifisinstance(in_guard_exc,AgentsException)andin_guard_exc.run_dataisNone:in_guard_exc.run_data=self._create_error_details()self._stored_exception=in_guard_excifself._output_guardrails_taskandself._output_guardrails_task.done():out_guard_exc=self._output_guardrails_task.exception()ifout_guard_excandisinstance(out_guard_exc,Exception):ifisinstance(out_guard_exc,AgentsException)andout_guard_exc.run_dataisNone:out_guard_exc.run_data=self._create_error_details()self._stored_exception=out_guard_excdef_cleanup_tasks(self):ifself._run_impl_taskandnotself._run_impl_task.done():self._run_impl_task.cancel()ifself._input_guardrails_taskandnotself._input_guardrails_task.done():self._input_guardrails_task.cancel()ifself._output_guardrails_taskandnotself._output_guardrails_task.done():self._output_guardrails_task.cancel()def__str__(self)->str:returnpretty_print_run_result_streaming(self)asyncdef_await_task_safely(self,task:asyncio.Task[Any]|None)->None:"""Await a task if present, ignoring cancellation and storing exceptions elsewhere.        This ensures we do not lose late guardrail exceptions while not surfacing        CancelledError to callers of stream_events.        """iftaskandnottask.done():try:awaittaskexceptasyncio.CancelledError:# Task was cancelled (e.g., due to result.cancel()). Nothing to do here.passexceptException:# The exception will be surfaced via _check_errors() if needed.pass