run_loop.py

"""
Run-loop orchestration helpers used by the Agent runner. This module coordinates tool execution,
approvals, and turn processing; all symbols here are internal and not part of the public SDK.
"""

from __future__ import annotations

import asyncio
import dataclasses as _dc
import json
from collections.abc import Awaitable, Callable, Mapping
from typing import Any, TypeVar, cast

from openai.types.responses import (
    Response,
    ResponseCompletedEvent,
    ResponseFunctionToolCall,
    ResponseOutputItemDoneEvent,
)
from openai.types.responses.response_output_item import McpCall, McpListTools, ResponseOutputItem
from openai.types.responses.response_prompt_param import ResponsePromptParam
from openai.types.responses.response_reasoning_item import ResponseReasoningItem

from .._mcp_tool_metadata import collect_mcp_list_tools_metadata
from .._tool_identity import (
    NamedToolLookupKey,
    build_function_tool_lookup_map,
    get_function_tool_lookup_key_for_call,
    get_tool_trace_name_for_tool,
)
from ..agent import Agent
from ..agent_output import AgentOutputSchemaBase
from ..exceptions import (
    AgentsException,
    InputGuardrailTripwireTriggered,
    MaxTurnsExceeded,
    ModelBehaviorError,
    OutputGuardrailTripwireTriggered,
    RunErrorDetails,
    UserError,
)
from ..handoffs import Handoff
from ..items import (
    HandoffCallItem,
    ItemHelpers,
    ModelResponse,
    ReasoningItem,
    RunItem,
    ToolApprovalItem,
    ToolCallItem,
    ToolCallItemTypes,
    ToolSearchCallItem,
    ToolSearchOutputItem,
    TResponseInputItem,
    coerce_tool_search_call_raw_item,
    coerce_tool_search_output_raw_item,
)
from ..lifecycle import RunHooks
from ..logger import logger
from ..memory import Session
from ..models._response_terminal import (
    response_error_event_failure_error,
    response_terminal_failure_error,
)
from ..result import RunResultStreaming
from ..run_config import ReasoningItemIdPolicy, RunConfig
from ..run_context import AgentHookContext, RunContextWrapper, TContext
from ..run_error_handlers import RunErrorHandlers
from ..run_state import RunState
from ..sandbox.runtime import SandboxRuntime
from ..stream_events import (
    AgentUpdatedStreamEvent,
    RawResponsesStreamEvent,
    RunItemStreamEvent,
)
from ..tool import (
    FunctionTool,
    Tool,
    ToolOrigin,
    ToolOriginType,
    dispose_resolved_computers,
    get_function_tool_origin,
)
from ..tracing import Span, SpanError, agent_span, get_current_trace, task_span, turn_span
from ..tracing.model_tracing import get_model_tracing_impl
from ..tracing.span_data import AgentSpanData, TaskSpanData
from ..usage import Usage
from ..util import _coro, _error_tracing
from .agent_bindings import AgentBindings, bind_public_agent
from .agent_runner_helpers import (
    apply_resumed_conversation_settings,
    attach_usage_to_span,
    get_unsent_tool_call_ids_for_interrupted_state,
    snapshot_usage,
    usage_delta,
)
from .approvals import approvals_from_step
from .error_handlers import (
    build_run_error_data,
    create_message_output_item,
    format_final_output_text,
    resolve_run_error_handler_result,
    validate_handler_final_output,
)
from .guardrails import (
    input_guardrail_tripwire_triggered_for_stream,
    run_input_guardrails,
    run_input_guardrails_with_queue,
    run_output_guardrails,
    run_single_input_guardrail,
    run_single_output_guardrail,
)
from .items import (
    REJECTION_MESSAGE,
    copy_input_items,
    deduplicate_input_items_preferring_latest,
    ensure_input_item_format,
    normalize_resumed_input,
    prepare_model_input_items,
    run_items_to_input_items,
)
from .model_retry import (
    apply_retry_attempt_usage,
    get_response_with_retry,
    stream_response_with_retry,
)
from .oai_conversation import OpenAIServerConversationTracker
from .prompt_cache_key import PromptCacheKeyResolver, model_settings_with_prompt_cache_key
from .run_steps import (
    NextStepFinalOutput,
    NextStepHandoff,
    NextStepInterruption,
    NextStepRunAgain,
    ProcessedResponse,
    QueueCompleteSentinel,
    SingleStepResult,
    ToolRunApplyPatchCall,
    ToolRunComputerAction,
    ToolRunFunction,
    ToolRunHandoff,
    ToolRunLocalShellCall,
    ToolRunMCPApprovalRequest,
    ToolRunShellCall,
)
from .session_persistence import (
    persist_session_items_for_guardrail_trip,
    prepare_input_with_session,
    resumed_turn_items,
    rewind_session_items,
    save_result_to_session,
    save_resumed_turn_items,
    session_items_for_turn,
    update_run_state_after_resume,
)
from .streaming import stream_step_items_to_queue, stream_step_result_to_queue
from .tool_actions import ApplyPatchAction, ComputerAction, LocalShellAction, ShellAction
from .tool_execution import (
    build_litellm_json_tool_call,
    coerce_shell_call,
    execute_apply_patch_calls,
    execute_computer_actions,
    execute_function_tool_calls,
    execute_local_shell_calls,
    execute_shell_calls,
    extract_tool_call_id,
    initialize_computer_tools,
    maybe_reset_tool_choice,
    normalize_shell_output,
    serialize_shell_output,
)
from .tool_planning import execute_mcp_approval_requests
from .tool_use_tracker import (
    TOOL_CALL_TYPES,
    AgentToolUseTracker,
    hydrate_tool_use_tracker,
    serialize_tool_use_tracker,
)
from .turn_preparation import (
    get_all_tools,
    get_handoffs,
    get_model,
    get_output_schema,
    maybe_filter_model_input,
    validate_run_hooks,
)
from .turn_resolution import (
    check_for_final_output_from_tools,
    execute_final_output,
    execute_handoffs,
    execute_tools_and_side_effects,
    get_single_step_result_from_response,
    process_model_response,
    resolve_interrupted_turn,
    run_final_output_hooks,
)

__all__ = [
    "extract_tool_call_id",
    "coerce_shell_call",
    "normalize_shell_output",
    "serialize_shell_output",
    "ComputerAction",
    "LocalShellAction",
    "ShellAction",
    "ApplyPatchAction",
    "REJECTION_MESSAGE",
    "AgentToolUseTracker",
    "ToolRunHandoff",
    "ToolRunFunction",
    "ToolRunComputerAction",
    "ToolRunMCPApprovalRequest",
    "ToolRunLocalShellCall",
    "ToolRunShellCall",
    "ToolRunApplyPatchCall",
    "ProcessedResponse",
    "NextStepHandoff",
    "NextStepFinalOutput",
    "NextStepRunAgain",
    "NextStepInterruption",
    "SingleStepResult",
    "QueueCompleteSentinel",
    "execute_tools_and_side_effects",
    "resolve_interrupted_turn",
    "execute_function_tool_calls",
    "execute_local_shell_calls",
    "execute_shell_calls",
    "execute_apply_patch_calls",
    "execute_computer_actions",
    "execute_handoffs",
    "execute_mcp_approval_requests",
    "execute_final_output",
    "run_final_output_hooks",
    "run_single_input_guardrail",
    "run_single_output_guardrail",
    "maybe_reset_tool_choice",
    "initialize_computer_tools",
    "process_model_response",
    "stream_step_items_to_queue",
    "stream_step_result_to_queue",
    "check_for_final_output_from_tools",
    "get_model_tracing_impl",
    "validate_run_hooks",
    "maybe_filter_model_input",
    "run_input_guardrails_with_queue",
    "start_streaming",
    "run_single_turn_streamed",
    "run_single_turn",
    "get_single_step_result_from_response",
    "run_input_guardrails",
    "run_output_guardrails",
    "get_new_response",
    "get_output_schema",
    "get_handoffs",
    "get_all_tools",
    "get_model",
    "input_guardrail_tripwire_triggered_for_stream",
]


def _should_attach_generic_agent_error(exc: Exception) -> bool:
    return not isinstance(
        exc,
        ModelBehaviorError | InputGuardrailTripwireTriggered | OutputGuardrailTripwireTriggered,
    )


async def _should_persist_stream_items(
    *,
    session: Session | None,
    server_conversation_tracker: OpenAIServerConversationTracker | None,
    streamed_result: RunResultStreaming,
) -> bool:
    if session is None or server_conversation_tracker is not None:
        return False
    should_skip_session_save = await input_guardrail_tripwire_triggered_for_stream(streamed_result)
    return should_skip_session_save is False


def _prepare_turn_input_items(
    caller_input: str | list[TResponseInputItem],
    generated_items: list[RunItem],
    reasoning_item_id_policy: ReasoningItemIdPolicy | None,
) -> list[TResponseInputItem]:
    caller_items = ItemHelpers.input_to_new_input_list(caller_input)
    continuation_items = run_items_to_input_items(generated_items, reasoning_item_id_policy)
    return prepare_model_input_items(caller_items, continuation_items)


def _complete_stream_interruption(
    streamed_result: RunResultStreaming,
    *,
    interruptions: list[ToolApprovalItem],
    processed_response: ProcessedResponse | None,
) -> None:
    streamed_result.interruptions = interruptions
    streamed_result._last_processed_response = processed_response
    streamed_result.is_complete = True
    streamed_result._event_queue.put_nowait(QueueCompleteSentinel())


async def _save_resumed_stream_items(
    *,
    session: Session | None,
    server_conversation_tracker: OpenAIServerConversationTracker | None,
    streamed_result: RunResultStreaming,
    run_state: RunState | None,
    items: list[RunItem],
    response_id: str | None,
    store: bool | None = None,
) -> None:
    if not await _should_persist_stream_items(
        session=session,
        server_conversation_tracker=server_conversation_tracker,
        streamed_result=streamed_result,
    ):
        return
    streamed_result._current_turn_persisted_item_count = await save_resumed_turn_items(
        session=session,
        items=items,
        persisted_count=streamed_result._current_turn_persisted_item_count,
        response_id=response_id,
        reasoning_item_id_policy=streamed_result._reasoning_item_id_policy,
        store=store,
    )
    if run_state is not None:
        run_state._current_turn_persisted_item_count = (
            streamed_result._current_turn_persisted_item_count
        )


async def _save_stream_items(
    *,
    session: Session | None,
    server_conversation_tracker: OpenAIServerConversationTracker | None,
    streamed_result: RunResultStreaming,
    run_state: RunState | None,
    items: list[RunItem],
    response_id: str | None,
    update_persisted_count: bool,
    store: bool | None = None,
) -> None:
    if not await _should_persist_stream_items(
        session=session,
        server_conversation_tracker=server_conversation_tracker,
        streamed_result=streamed_result,
    ):
        return
    await save_result_to_session(
        session,
        [],
        list(items),
        run_state,
        response_id=response_id,
        store=store,
    )
    if update_persisted_count and streamed_result._state is not None:
        streamed_result._current_turn_persisted_item_count = (
            streamed_result._state._current_turn_persisted_item_count
        )


async def _run_output_guardrails_for_stream(
    *,
    agent: Agent[TContext],
    run_config: RunConfig,
    output: Any,
    context_wrapper: RunContextWrapper[TContext],
    streamed_result: RunResultStreaming,
) -> list[Any]:
    streamed_result._output_guardrails_task = asyncio.create_task(
        run_output_guardrails(
            agent.output_guardrails + (run_config.output_guardrails or []),
            agent,
            output,
            context_wrapper,
        )
    )

    try:
        return cast(list[Any], await streamed_result._output_guardrails_task)
    except OutputGuardrailTripwireTriggered:
        raise
    except asyncio.CancelledError:
        raise
    except Exception:
        logger.error("Unexpected error in output guardrails", exc_info=True)
        return []


async def _finalize_streamed_final_output(
    *,
    streamed_result: RunResultStreaming,
    agent: Agent[TContext],
    run_config: RunConfig,
    output: Any,
    context_wrapper: RunContextWrapper[TContext],
    save_items: Callable[[list[RunItem], str | None, bool | None], Awaitable[None]],
    items: list[RunItem],
    response_id: str | None,
    store_setting: bool | None,
) -> None:
    output_guardrail_results = await _run_output_guardrails_for_stream(
        agent=agent,
        run_config=run_config,
        output=output,
        context_wrapper=context_wrapper,
        streamed_result=streamed_result,
    )
    streamed_result.output_guardrail_results = output_guardrail_results
    streamed_result.final_output = output
    streamed_result.is_complete = True

    await save_items(items, response_id, store_setting)

    streamed_result._event_queue.put_nowait(QueueCompleteSentinel())


async def _finalize_streamed_interruption(
    *,
    streamed_result: RunResultStreaming,
    save_items: Callable[[list[RunItem], str | None, bool | None], Awaitable[None]],
    items: list[RunItem],
    response_id: str | None,
    store_setting: bool | None,
    interruptions: list[ToolApprovalItem],
    processed_response: ProcessedResponse | None,
) -> None:
    await save_items(items, response_id, store_setting)
    _complete_stream_interruption(
        streamed_result,
        interruptions=interruptions,
        processed_response=processed_response,
    )


T = TypeVar("T")


async def start_streaming(
    starting_input: str | list[TResponseInputItem],
    streamed_result: RunResultStreaming,
    starting_agent: Agent[TContext],
    max_turns: int,
    hooks: RunHooks[TContext],
    context_wrapper: RunContextWrapper[TContext],
    run_config: RunConfig,
    error_handlers: RunErrorHandlers[TContext] | None,
    previous_response_id: str | None,
    auto_previous_response_id: bool,
    conversation_id: str | None,
    session: Session | None,
    run_state: RunState[TContext] | None = None,
    *,
    is_resumed_state: bool = False,
    sandbox_runtime: SandboxRuntime[TContext] | None = None,
):
    """Run the streaming loop for a run result."""
    if streamed_result.trace:
        streamed_result.trace.start(mark_as_current=True)
    if run_state is not None:
        run_state.set_trace(get_current_trace() or streamed_result.trace)
        streamed_result._trace_state = run_state._trace_state

    if is_resumed_state and run_state is not None:
        (
            conversation_id,
            previous_response_id,
            auto_previous_response_id,
        ) = apply_resumed_conversation_settings(
            run_state=run_state,
            conversation_id=conversation_id,
            previous_response_id=previous_response_id,
            auto_previous_response_id=auto_previous_response_id,
        )

    current_trace = streamed_result.trace or get_current_trace()
    current_task_span: Span[TaskSpanData] | None = (
        task_span(name=current_trace.name) if current_trace else None
    )
    if current_task_span:
        current_task_span.start(mark_as_current=True)
    task_usage_start = snapshot_usage(context_wrapper.usage)

    try:
        resolved_reasoning_item_id_policy: ReasoningItemIdPolicy | None = (
            run_config.reasoning_item_id_policy
            if run_config.reasoning_item_id_policy is not None
            else (run_state._reasoning_item_id_policy if run_state is not None else None)
        )
        if run_state is not None:
            run_state._reasoning_item_id_policy = resolved_reasoning_item_id_policy
        streamed_result._reasoning_item_id_policy = resolved_reasoning_item_id_policy

        if (
            conversation_id is not None
            or previous_response_id is not None
            or auto_previous_response_id
        ):
            server_conversation_tracker = OpenAIServerConversationTracker(
                conversation_id=conversation_id,
                previous_response_id=previous_response_id,
                auto_previous_response_id=auto_previous_response_id,
                reasoning_item_id_policy=resolved_reasoning_item_id_policy,
            )
        else:
            server_conversation_tracker = None

        def _sync_conversation_tracking_from_tracker() -> None:
            if server_conversation_tracker is None:
                return
            if run_state is not None:
                run_state._conversation_id = server_conversation_tracker.conversation_id
                run_state._previous_response_id = server_conversation_tracker.previous_response_id
                run_state._auto_previous_response_id = (
                    server_conversation_tracker.auto_previous_response_id
                )
            streamed_result._conversation_id = server_conversation_tracker.conversation_id
            streamed_result._previous_response_id = server_conversation_tracker.previous_response_id
            streamed_result._auto_previous_response_id = (
                server_conversation_tracker.auto_previous_response_id
            )

        if run_state is None:
            run_state = RunState(
                context=context_wrapper,
                original_input=copy_input_items(starting_input),
                starting_agent=starting_agent,
                max_turns=max_turns,
                conversation_id=conversation_id,
                previous_response_id=previous_response_id,
                auto_previous_response_id=auto_previous_response_id,
            )
            run_state._reasoning_item_id_policy = resolved_reasoning_item_id_policy
            streamed_result._state = run_state
        elif streamed_result._state is None:
            streamed_result._state = run_state
        if run_state is not None:
            streamed_result._model_input_items = list(run_state._generated_items)
            # Streamed follow-ups need the same normalized replay signal as sync runs when the
            # runner's continuation differs from the richer session history.
            streamed_result._replay_from_model_input_items = list(
                run_state._generated_items
            ) != list(run_state._session_items)

        if run_state is not None:
            run_state._conversation_id = conversation_id
            run_state._previous_response_id = previous_response_id
            run_state._auto_previous_response_id = auto_previous_response_id
        streamed_result._conversation_id = conversation_id
        streamed_result._previous_response_id = previous_response_id
        streamed_result._auto_previous_response_id = auto_previous_response_id
        prompt_cache_key_resolver = PromptCacheKeyResolver.from_run_state(
            run_state=run_state,
        )

        current_span: Span[AgentSpanData] | None = None
        if run_state is not None and run_state._current_agent is not None:
            current_agent = run_state._current_agent
        else:
            current_agent = starting_agent
        if run_state is not None:
            current_turn = run_state._current_turn
        else:
            current_turn = 0
        should_run_agent_start_hooks = True
        tool_use_tracker = AgentToolUseTracker()
        if run_state is not None:
            hydrate_tool_use_tracker(tool_use_tracker, run_state, starting_agent)

        pending_server_items: list[RunItem] | None = None
        session_input_items_for_persistence: list[TResponseInputItem] | None = None

        if is_resumed_state and server_conversation_tracker is not None and run_state is not None:
            session_items: list[TResponseInputItem] | None = None
            if session is not None:
                try:
                    session_items = await session.get_items()
                except Exception:
                    session_items = None
            server_conversation_tracker.hydrate_from_state(
                original_input=run_state._original_input,
                generated_items=run_state._generated_items,
                model_responses=run_state._model_responses,
                session_items=session_items,
                unsent_tool_call_ids=get_unsent_tool_call_ids_for_interrupted_state(run_state),
            )

        streamed_result._event_queue.put_nowait(AgentUpdatedStreamEvent(new_agent=current_agent))

        prepared_input: str | list[TResponseInputItem]
        if is_resumed_state and run_state is not None:
            prepared_input = normalize_resumed_input(starting_input)
            streamed_result.input = prepared_input
            streamed_result._original_input_for_persistence = []
            streamed_result._stream_input_persisted = True
        else:
            server_manages_conversation = server_conversation_tracker is not None
            prepared_input, session_items_snapshot = await prepare_input_with_session(
                starting_input,
                session,
                run_config.session_input_callback,
                run_config.session_settings,
                include_history_in_prepared_input=not server_manages_conversation,
                preserve_dropped_new_items=True,
            )
            streamed_result.input = prepared_input
            streamed_result._original_input = copy_input_items(prepared_input)
            if server_manages_conversation:
                streamed_result._original_input_for_persistence = []
                streamed_result._stream_input_persisted = True
            else:
                session_input_items_for_persistence = session_items_snapshot
                streamed_result._original_input_for_persistence = session_items_snapshot

        async def _save_resumed_items(
            items: list[RunItem], response_id: str | None, store_setting: bool | None
        ) -> None:
            await _save_resumed_stream_items(
                session=session,
                server_conversation_tracker=server_conversation_tracker,
                streamed_result=streamed_result,
                run_state=run_state,
                items=items,
                response_id=response_id,
                store=store_setting,
            )

        async def _save_stream_items_with_count(
            items: list[RunItem], response_id: str | None, store_setting: bool | None
        ) -> None:
            await _save_stream_items(
                session=session,
                server_conversation_tracker=server_conversation_tracker,
                streamed_result=streamed_result,
                run_state=run_state,
                items=items,
                response_id=response_id,
                update_persisted_count=True,
                store=store_setting,
            )

        async def _save_stream_items_without_count(
            items: list[RunItem], response_id: str | None, store_setting: bool | None
        ) -> None:
            await _save_stream_items(
                session=session,
                server_conversation_tracker=server_conversation_tracker,
                streamed_result=streamed_result,
                run_state=run_state,
                items=items,
                response_id=response_id,
                update_persisted_count=False,
                store=store_setting,
            )
    except BaseException:
        if current_task_span:
            attach_usage_to_span(
                current_task_span,
                usage_delta(task_usage_start, context_wrapper.usage),
            )
            current_task_span.finish(reset_current=True)
        if streamed_result.trace:
            streamed_result.trace.finish(reset_current=True)
        if not streamed_result.is_complete:
            streamed_result.is_complete = True
            streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
        raise

    try:
        while True:
            all_input_guardrails = (
                starting_agent.input_guardrails + (run_config.input_guardrails or [])
                if current_turn == 0 and not is_resumed_state
                else []
            )
            sequential_guardrails = [g for g in all_input_guardrails if not g.run_in_parallel]
            parallel_guardrails = [g for g in all_input_guardrails if g.run_in_parallel]
            current_bindings = bind_public_agent(current_agent)
            execution_agent = current_bindings.execution_agent
            prepared_turn_input = copy_input_items(streamed_result.input)
            if sandbox_runtime is not None and sandbox_runtime.enabled and sequential_guardrails:
                # Mirror the non-streaming path: a blocking first-turn guardrail should fire
                # before sandbox prep can create, start, or mutate sandbox state.
                existing_input_guardrail_count = len(streamed_result.input_guardrail_results)
                await run_input_guardrails_with_queue(
                    starting_agent,
                    sequential_guardrails,
                    ItemHelpers.input_to_new_input_list(prepared_turn_input),
                    context_wrapper,
                    streamed_result,
                    None,
                )
                for result in streamed_result.input_guardrail_results[
                    existing_input_guardrail_count:
                ]:
                    if result.output.tripwire_triggered:
                        streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
                        session_input_items_for_persistence = (
                            await persist_session_items_for_guardrail_trip(
                                session,
                                server_conversation_tracker,
                                session_input_items_for_persistence,
                                starting_input,
                                run_state,
                                store=current_agent.model_settings.resolve(
                                    run_config.model_settings
                                ).store,
                            )
                        )
                        raise InputGuardrailTripwireTriggered(result)
                sequential_guardrails = []

            if sandbox_runtime is not None:
                prepared_sandbox = await sandbox_runtime.prepare_agent(
                    current_agent=current_agent,
                    current_input=prepared_turn_input,
                    context_wrapper=context_wrapper,
                    is_resumed_state=is_resumed_state,
                )
                current_bindings = prepared_sandbox.bindings
                execution_agent = current_bindings.execution_agent
                prepared_turn_input = copy_input_items(prepared_sandbox.input)
                streamed_result.input = prepared_turn_input
                streamed_result._original_input = copy_input_items(prepared_turn_input)
                if run_state is not None:
                    run_state._original_input = copy_input_items(prepared_turn_input)
                sandbox_runtime.apply_result_metadata(streamed_result)

            if is_resumed_state and run_state is not None and run_state._current_step is not None:
                if isinstance(run_state._current_step, NextStepInterruption):
                    if not run_state._model_responses or not run_state._last_processed_response:
                        raise UserError("No model response found in previous state")

                    last_model_response = run_state._model_responses[-1]

                    turn_result = await resolve_interrupted_turn(
                        bindings=current_bindings,
                        original_input=run_state._original_input,
                        original_pre_step_items=run_state._generated_items,
                        new_response=last_model_response,
                        processed_response=run_state._last_processed_response,
                        hooks=hooks,
                        context_wrapper=context_wrapper,
                        run_config=run_config,
                        server_manages_conversation=server_conversation_tracker is not None,
                        run_state=run_state,
                    )

                    tool_use_tracker.record_processed_response(
                        current_agent, run_state._last_processed_response
                    )
                    streamed_result._tool_use_tracker_snapshot = serialize_tool_use_tracker(
                        tool_use_tracker,
                        starting_agent=(
                            run_state._starting_agent
                            if run_state is not None and run_state._starting_agent is not None
                            else starting_agent
                        ),
                    )

                    streamed_result.input = turn_result.original_input
                    streamed_result._original_input = copy_input_items(turn_result.original_input)
                    generated_items, turn_session_items = resumed_turn_items(turn_result)
                    base_session_items = (
                        list(run_state._session_items) if run_state is not None else []
                    )
                    streamed_result._model_input_items = generated_items
                    streamed_result.new_items = base_session_items + list(turn_session_items)
                    streamed_result._replay_from_model_input_items = list(
                        streamed_result._model_input_items
                    ) != list(streamed_result.new_items)
                    if run_state is not None:
                        update_run_state_after_resume(
                            run_state,
                            turn_result=turn_result,
                            generated_items=generated_items,
                            session_items=streamed_result.new_items,
                        )
                        run_state._current_turn_persisted_item_count = (
                            streamed_result._current_turn_persisted_item_count
                        )

                    stream_step_items_to_queue(
                        list(turn_session_items), streamed_result._event_queue
                    )
                    store_setting = current_agent.model_settings.resolve(
                        run_config.model_settings
                    ).store

                    if isinstance(turn_result.next_step, NextStepInterruption):
                        await _finalize_streamed_interruption(
                            streamed_result=streamed_result,
                            save_items=_save_resumed_items,
                            items=list(turn_session_items),
                            response_id=turn_result.model_response.response_id,
                            store_setting=store_setting,
                            interruptions=approvals_from_step(turn_result.next_step),
                            processed_response=run_state._last_processed_response,
                        )
                        break

                    if isinstance(turn_result.next_step, NextStepHandoff):
                        current_agent = turn_result.next_step.new_agent
                        if run_state is not None:
                            run_state._current_agent = current_agent
                        if current_span:
                            current_span.finish(reset_current=True)
                        current_span = None
                        should_run_agent_start_hooks = True
                        streamed_result._event_queue.put_nowait(
                            AgentUpdatedStreamEvent(new_agent=current_agent)
                        )
                        run_state._current_step = NextStepRunAgain()  # type: ignore[assignment]
                        continue

                    if isinstance(turn_result.next_step, NextStepFinalOutput):
                        await _finalize_streamed_final_output(
                            streamed_result=streamed_result,
                            agent=current_agent,
                            run_config=run_config,
                            output=turn_result.next_step.output,
                            context_wrapper=context_wrapper,
                            save_items=_save_resumed_items,
                            items=list(turn_session_items),
                            response_id=turn_result.model_response.response_id,
                            store_setting=store_setting,
                        )
                        break

                    if isinstance(turn_result.next_step, NextStepRunAgain):
                        await _save_resumed_items(
                            list(turn_session_items),
                            turn_result.model_response.response_id,
                            store_setting,
                        )
                        run_state._current_step = NextStepRunAgain()  # type: ignore[assignment]
                        continue

                    run_state._current_step = None

            if streamed_result._cancel_mode == "after_turn":
                streamed_result.is_complete = True
                streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
                break

            if streamed_result.is_complete:
                break

            all_tools = await get_all_tools(execution_agent, context_wrapper)
            await initialize_computer_tools(tools=all_tools, context_wrapper=context_wrapper)

            if current_span is None:
                handoff_names = [
                    h.agent_name for h in await get_handoffs(execution_agent, context_wrapper)
                ]
                if output_schema := get_output_schema(execution_agent):
                    output_type_name = output_schema.name()
                else:
                    output_type_name = "str"

                current_span = agent_span(
                    name=current_agent.name,
                    handoffs=handoff_names,
                    output_type=output_type_name,
                )
                current_span.start(mark_as_current=True)
                tool_names = [
                    tool_name
                    for tool in all_tools
                    if (tool_name := get_tool_trace_name_for_tool(tool)) is not None
                ]
                current_span.span_data.tools = tool_names

            current_turn += 1
            streamed_result.current_turn = current_turn
            streamed_result._current_turn_persisted_item_count = 0
            if run_state:
                run_state._current_turn_persisted_item_count = 0

            if current_turn > max_turns:
                _error_tracing.attach_error_to_span(
                    current_span,
                    SpanError(
                        message="Max turns exceeded",
                        data={"max_turns": max_turns},
                    ),
                )
                max_turns_error = MaxTurnsExceeded(f"Max turns ({max_turns}) exceeded")
                handler_configured = bool(
                    error_handlers and error_handlers.get("max_turns") is not None
                )
                if handler_configured:
                    streamed_result._max_turns_handled = True
                run_error_data = build_run_error_data(
                    input=streamed_result.input,
                    new_items=streamed_result.new_items,
                    raw_responses=streamed_result.raw_responses,
                    last_agent=current_agent,
                    reasoning_item_id_policy=streamed_result._reasoning_item_id_policy,
                )
                handler_result = await resolve_run_error_handler_result(
                    error_handlers=error_handlers,
                    error=max_turns_error,
                    context_wrapper=context_wrapper,
                    run_data=run_error_data,
                )
                if handler_result is None:
                    if handler_configured:
                        streamed_result._max_turns_handled = False
                    streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
                    break

                validated_output = validate_handler_final_output(
                    current_agent, handler_result.final_output
                )
                output_text = format_final_output_text(current_agent, validated_output)
                synthesized_item = create_message_output_item(current_agent, output_text)
                include_in_history = handler_result.include_in_history
                if include_in_history:
                    streamed_result._model_input_items.append(synthesized_item)
                    streamed_result.new_items.append(synthesized_item)
                    if run_state is not None:
                        run_state._generated_items = list(streamed_result._model_input_items)
                        run_state._clear_generated_items_last_processed_marker()
                        run_state._session_items = list(streamed_result.new_items)
                    stream_step_items_to_queue([synthesized_item], streamed_result._event_queue)
                    store_setting = current_agent.model_settings.resolve(
                        run_config.model_settings
                    ).store
                    if is_resumed_state:
                        await _save_resumed_items([synthesized_item], None, store_setting)
                    else:
                        await _save_stream_items_with_count([synthesized_item], None, store_setting)

                await run_final_output_hooks(
                    current_agent, hooks, context_wrapper, validated_output
                )
                output_guardrail_results = await _run_output_guardrails_for_stream(
                    agent=current_agent,
                    run_config=run_config,
                    output=validated_output,
                    context_wrapper=context_wrapper,
                    streamed_result=streamed_result,
                )
                streamed_result.output_guardrail_results = output_guardrail_results
                streamed_result.final_output = validated_output
                streamed_result.is_complete = True
                streamed_result._stored_exception = None
                streamed_result._max_turns_handled = True
                streamed_result.current_turn = max_turns
                if run_state is not None:
                    run_state._current_turn = max_turns
                    run_state._current_step = None
                streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
                break

            if current_turn == 1:
                if sequential_guardrails:
                    await run_input_guardrails_with_queue(
                        starting_agent,
                        sequential_guardrails,
                        ItemHelpers.input_to_new_input_list(prepared_turn_input),
                        context_wrapper,
                        streamed_result,
                        current_span,
                    )
                    for result in streamed_result.input_guardrail_results:
                        if result.output.tripwire_triggered:
                            streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
                            session_input_items_for_persistence = (
                                await persist_session_items_for_guardrail_trip(
                                    session,
                                    server_conversation_tracker,
                                    session_input_items_for_persistence,
                                    starting_input,
                                    run_state,
                                    store=current_agent.model_settings.resolve(
                                        run_config.model_settings
                                    ).store,
                                )
                            )
                            raise InputGuardrailTripwireTriggered(result)

                if parallel_guardrails:
                    streamed_result._input_guardrails_task = asyncio.create_task(
                        run_input_guardrails_with_queue(
                            starting_agent,
                            parallel_guardrails,
                            ItemHelpers.input_to_new_input_list(prepared_turn_input),
                            context_wrapper,
                            streamed_result,
                            current_span,
                        )
                    )
            try:
                logger.debug(
                    "Starting turn %s, current_agent=%s",
                    current_turn,
                    current_agent.name,
                )
                turn_usage_start = snapshot_usage(context_wrapper.usage)
                current_turn_span = turn_span(
                    turn=current_turn,
                    agent_name=current_agent.name,
                )
                current_turn_span.start(mark_as_current=True)
                try:
                    if (
                        session is not None
                        and server_conversation_tracker is None
                        and not streamed_result._stream_input_persisted
                    ):
                        streamed_result._original_input_for_persistence = (
                            session_input_items_for_persistence
                            if session_input_items_for_persistence is not None
                            else []
                        )
                    turn_result = await run_single_turn_streamed(
                        streamed_result,
                        current_bindings,
                        hooks,
                        context_wrapper,
                        run_config,
                        should_run_agent_start_hooks,
                        tool_use_tracker,
                        all_tools,
                        server_conversation_tracker,
                        pending_server_items=pending_server_items,
                        session=session,
                        session_items_to_rewind=(
                            streamed_result._original_input_for_persistence
                            if session is not None and server_conversation_tracker is None
                            else None
                        ),
                        reasoning_item_id_policy=resolved_reasoning_item_id_policy,
                        prompt_cache_key_resolver=prompt_cache_key_resolver,
                        error_handlers=error_handlers,
                    )
                finally:
                    attach_usage_to_span(
                        current_turn_span,
                        usage_delta(turn_usage_start, context_wrapper.usage),
                    )
                    current_turn_span.finish(reset_current=True)
                logger.debug(
                    "Turn %s complete, next_step type=%s",
                    current_turn,
                    type(turn_result.next_step).__name__,
                )
                should_run_agent_start_hooks = False
                streamed_result._tool_use_tracker_snapshot = serialize_tool_use_tracker(
                    tool_use_tracker,
                    starting_agent=(
                        run_state._starting_agent
                        if run_state is not None and run_state._starting_agent is not None
                        else starting_agent
                    ),
                )

                streamed_result.raw_responses = streamed_result.raw_responses + [
                    turn_result.model_response
                ]
                streamed_result.input = turn_result.original_input
                if isinstance(turn_result.next_step, NextStepHandoff):
                    streamed_result._original_input = copy_input_items(turn_result.original_input)
                    if run_state is not None:
                        run_state._original_input = copy_input_items(turn_result.original_input)
                streamed_result._model_input_items = (
                    turn_result.pre_step_items + turn_result.new_step_items
                )
                turn_session_items = session_items_for_turn(turn_result)
                streamed_result.new_items.extend(turn_session_items)
                streamed_result._replay_from_model_input_items = list(
                    streamed_result._model_input_items
                ) != list(streamed_result.new_items)
                store_setting = current_agent.model_settings.resolve(
                    run_config.model_settings
                ).store
                if server_conversation_tracker is not None:
                    pending_server_items = list(turn_result.new_step_items)

                if isinstance(turn_result.next_step, NextStepRunAgain):
                    streamed_result._current_turn_persisted_item_count = 0
                    if run_state:
                        run_state._current_turn_persisted_item_count = 0

                if server_conversation_tracker is not None:
                    server_conversation_tracker.track_server_items(turn_result.model_response)

                if isinstance(turn_result.next_step, NextStepHandoff):
                    await _save_stream_items_without_count(
                        turn_session_items,
                        turn_result.model_response.response_id,
                        store_setting,
                    )
                    current_agent = turn_result.next_step.new_agent
                    if run_state is not None:
                        run_state._current_agent = current_agent
                    current_span.finish(reset_current=True)
                    current_span = None
                    should_run_agent_start_hooks = True
                    streamed_result._event_queue.put_nowait(
                        AgentUpdatedStreamEvent(new_agent=current_agent)
                    )
                    if streamed_result._state is not None:
                        streamed_result._state._current_step = NextStepRunAgain()

                    if streamed_result._cancel_mode == "after_turn":  # type: ignore[comparison-overlap]
                        streamed_result.is_complete = True
                        streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
                        break
                elif isinstance(turn_result.next_step, NextStepFinalOutput):
                    await _finalize_streamed_final_output(
                        streamed_result=streamed_result,
                        agent=current_agent,
                        run_config=run_config,
                        output=turn_result.next_step.output,
                        context_wrapper=context_wrapper,
                        save_items=_save_stream_items_with_count,
                        items=turn_session_items,
                        response_id=turn_result.model_response.response_id,
                        store_setting=store_setting,
                    )
                    break
                elif isinstance(turn_result.next_step, NextStepInterruption):
                    processed_response_for_state = turn_result.processed_response
                    if processed_response_for_state is None and run_state is not None:
                        processed_response_for_state = run_state._last_processed_response
                    if run_state is not None:
                        run_state._model_responses = streamed_result.raw_responses
                        run_state._last_processed_response = processed_response_for_state
                        run_state._generated_items = streamed_result._model_input_items
                        run_state._mark_generated_items_merged_with_last_processed()
                        run_state._session_items = list(streamed_result.new_items)
                        run_state._current_step = turn_result.next_step
                        run_state._current_turn = current_turn
                        run_state._current_turn_persisted_item_count = (
                            streamed_result._current_turn_persisted_item_count
                        )
                    await _finalize_streamed_interruption(
                        streamed_result=streamed_result,
                        save_items=_save_stream_items_with_count,
                        items=turn_session_items,
                        response_id=turn_result.model_response.response_id,
                        store_setting=store_setting,
                        interruptions=approvals_from_step(turn_result.next_step),
                        processed_response=processed_response_for_state,
                    )
                    break
                elif isinstance(turn_result.next_step, NextStepRunAgain):
                    if streamed_result._state is not None:
                        streamed_result._state._current_step = NextStepRunAgain()

                    await _save_stream_items_with_count(
                        turn_session_items,
                        turn_result.model_response.response_id,
                        store_setting,
                    )

                    if streamed_result._cancel_mode == "after_turn":  # type: ignore[comparison-overlap]
                        streamed_result.is_complete = True
                        streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
                        break
            except Exception as e:
                if current_span and _should_attach_generic_agent_error(e):
                    _error_tracing.attach_error_to_span(
                        current_span,
                        SpanError(
                            message="Error in agent run",
                            data={"error": str(e)},
                        ),
                    )
                raise
    except AgentsException as exc:
        streamed_result.is_complete = True
        streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
        exc.run_data = RunErrorDetails(
            input=streamed_result.input,
            new_items=streamed_result.new_items,
            raw_responses=streamed_result.raw_responses,
            last_agent=current_agent,
            context_wrapper=context_wrapper,
            input_guardrail_results=streamed_result.input_guardrail_results,
            output_guardrail_results=streamed_result.output_guardrail_results,
        )
        raise
    except Exception as e:
        if current_span and _should_attach_generic_agent_error(e):
            _error_tracing.attach_error_to_span(
                current_span,
                SpanError(
                    message="Error in agent run",
                    data={"error": str(e)},
                ),
            )
        streamed_result.is_complete = True
        streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
        raise
    else:
        streamed_result.is_complete = True
    finally:
        _sync_conversation_tracking_from_tracker()
        if streamed_result._input_guardrails_task:
            try:
                triggered = await input_guardrail_tripwire_triggered_for_stream(streamed_result)
                if triggered:
                    first_trigger = next(
                        (
                            result
                            for result in streamed_result.input_guardrail_results
                            if result.output.tripwire_triggered
                        ),
                        None,
                    )
                    if first_trigger is not None:
                        raise InputGuardrailTripwireTriggered(first_trigger)
            except Exception as e:
                logger.debug(
                    f"Error in streamed_result finalize for agent {current_agent.name} - {e}"
                )
        try:
            await dispose_resolved_computers(run_context=context_wrapper)
        except Exception as error:
            logger.warning("Failed to dispose computers after streamed run: %s", error)
        if current_span:
            current_span.finish(reset_current=True)
        if current_task_span:
            attach_usage_to_span(
                current_task_span,
                usage_delta(task_usage_start, context_wrapper.usage),
            )
            current_task_span.finish(reset_current=True)
        if streamed_result.trace:
            streamed_result.trace.finish(reset_current=True)

        if not streamed_result.is_complete:
            streamed_result.is_complete = True
            streamed_result._event_queue.put_nowait(QueueCompleteSentinel())


async def run_single_turn_streamed(
    streamed_result: RunResultStreaming,
    bindings: AgentBindings[TContext],
    hooks: RunHooks[TContext],
    context_wrapper: RunContextWrapper[TContext],
    run_config: RunConfig,
    should_run_agent_start_hooks: bool,
    tool_use_tracker: AgentToolUseTracker,
    all_tools: list[Tool],
    server_conversation_tracker: OpenAIServerConversationTracker | None = None,
    session: Session | None = None,
    session_items_to_rewind: list[TResponseInputItem] | None = None,
    pending_server_items: list[RunItem] | None = None,
    reasoning_item_id_policy: ReasoningItemIdPolicy | None = None,
    prompt_cache_key_resolver: PromptCacheKeyResolver | None = None,
    error_handlers: RunErrorHandlers[TContext] | None = None,
) -> SingleStepResult:
    """Run a single streamed turn and emit events as results arrive."""
    public_agent = bindings.public_agent
    execution_agent = bindings.execution_agent

    async def raise_if_input_guardrail_tripwire_known() -> None:
        tripwire_result = streamed_result._triggered_input_guardrail_result
        if tripwire_result is not None:
            raise InputGuardrailTripwireTriggered(tripwire_result)

        task = streamed_result._input_guardrails_task
        if task is None or not task.done():
            return

        guardrail_exception = task.exception()
        if guardrail_exception is not None:
            raise guardrail_exception

        tripwire_result = streamed_result._triggered_input_guardrail_result
        if tripwire_result is not None:
            raise InputGuardrailTripwireTriggered(tripwire_result)

    emitted_tool_call_ids: set[str] = set()
    emitted_reasoning_item_ids: set[str] = set()
    emitted_tool_search_fingerprints: set[str] = set()
    # Precompute the lookup map used for streaming descriptions. Function tools use the same
    # collision-free lookup keys as runtime dispatch, including deferred top-level aliases.
    tool_map: dict[NamedToolLookupKey, Any] = cast(
        dict[NamedToolLookupKey, Any],
        build_function_tool_lookup_map(
            [tool for tool in all_tools if isinstance(tool, FunctionTool)]
        ),
    )
    for tool in all_tools:
        tool_name = getattr(tool, "name", None)
        if not isinstance(tool_name, str) or not tool_name:
            continue
        if isinstance(tool, FunctionTool):
            continue
        tool_map[tool_name] = tool

    def _tool_search_fingerprint(raw_item: Any) -> str:
        if isinstance(raw_item, Mapping):
            payload: Any = dict(raw_item)
        elif hasattr(raw_item, "model_dump"):
            payload = cast(Any, raw_item).model_dump(exclude_unset=True)
        else:
            payload = {
                "type": getattr(raw_item, "type", None),
                "id": getattr(raw_item, "id", None),
            }
        return json.dumps(payload, sort_keys=True, default=str)

    try:
        turn_input = ItemHelpers.input_to_new_input_list(streamed_result.input)
    except Exception:
        turn_input = []
    context_wrapper.turn_input = list(turn_input)

    if should_run_agent_start_hooks:
        agent_hook_context = AgentHookContext(
            context=context_wrapper.context,
            usage=context_wrapper.usage,
            _approvals=context_wrapper._approvals,
            turn_input=turn_input,
        )
        await asyncio.gather(
            hooks.on_agent_start(agent_hook_context, public_agent),
            (
                public_agent.hooks.on_start(agent_hook_context, public_agent)
                if public_agent.hooks
                else _coro.noop_coroutine()
            ),
        )

    output_schema = get_output_schema(execution_agent)

    streamed_result.current_agent = public_agent
    streamed_result._current_agent_output_schema = get_output_schema(public_agent)

    system_prompt, prompt_config = await asyncio.gather(
        execution_agent.get_system_prompt(context_wrapper),
        execution_agent.get_prompt(context_wrapper),
    )

    handoffs = await get_handoffs(execution_agent, context_wrapper)
    model = get_model(execution_agent, run_config)
    model_settings = execution_agent.model_settings.resolve(run_config.model_settings)
    model_settings = maybe_reset_tool_choice(public_agent, tool_use_tracker, model_settings)

    final_response: ModelResponse | None = None
    streamed_response_output: list[ResponseOutputItem] = []

    if server_conversation_tracker is not None:
        items_for_input = (
            pending_server_items if pending_server_items else streamed_result._model_input_items
        )
        input = server_conversation_tracker.prepare_input(streamed_result.input, items_for_input)
        logger.debug(
            "prepare_input returned %s items; remaining_initial_input=%s",
            len(input),
            len(server_conversation_tracker.remaining_initial_input)
            if server_conversation_tracker.remaining_initial_input
            else 0,
        )
    else:
        input = _prepare_turn_input_items(
            streamed_result.input,
            streamed_result._model_input_items,
            reasoning_item_id_policy,
        )

    filtered = await maybe_filter_model_input(
        agent=public_agent,
        run_config=run_config,
        context_wrapper=context_wrapper,
        input_items=input,
        system_instructions=system_prompt,
    )
    if isinstance(filtered.input, list):
        filtered.input = deduplicate_input_items_preferring_latest(filtered.input)
    hosted_mcp_tool_metadata = collect_mcp_list_tools_metadata(streamed_result._model_input_items)
    if isinstance(filtered.input, list):
        hosted_mcp_tool_metadata.update(collect_mcp_list_tools_metadata(filtered.input))
    if server_conversation_tracker is not None:
        logger.debug(
            "filtered.input has %s items; ids=%s",
            len(filtered.input),
            [id(i) for i in filtered.input],
        )
        # Track only the items actually sent after call_model_input_filter runs. Retry helpers
        # explicitly rewind this state before replaying a failed request.
        server_conversation_tracker.mark_input_as_sent(filtered.input)
    if not filtered.input and server_conversation_tracker is None:
        raise RuntimeError("Prepared model input is empty")

    await asyncio.gather(
        hooks.on_llm_start(context_wrapper, public_agent, filtered.instructions, filtered.input),
        (
            public_agent.hooks.on_llm_start(
                context_wrapper,
                public_agent,
                filtered.instructions,
                filtered.input,
            )
            if public_agent.hooks
            else _coro.noop_coroutine()
        ),
    )

    if (
        not streamed_result._stream_input_persisted
        and session is not None
        and server_conversation_tracker is None
        and streamed_result._original_input_for_persistence is not None
        and len(streamed_result._original_input_for_persistence) > 0
    ):
        streamed_result._stream_input_persisted = True
        input_items_to_save = [
            ensure_input_item_format(item)
            for item in ItemHelpers.input_to_new_input_list(
                streamed_result._original_input_for_persistence
            )
        ]
        if input_items_to_save:
            await save_result_to_session(session, input_items_to_save, [], streamed_result._state)

    previous_response_id = (
        server_conversation_tracker.previous_response_id
        if server_conversation_tracker
        and server_conversation_tracker.previous_response_id is not None
        else None
    )
    conversation_id = (
        server_conversation_tracker.conversation_id if server_conversation_tracker else None
    )
    if conversation_id:
        logger.debug("Using conversation_id=%s", conversation_id)
    else:
        logger.debug("No conversation_id available for request")

    prompt_cache_key = (
        prompt_cache_key_resolver.resolve(
            model_settings,
            model=model,
            conversation_id=conversation_id,
            session=session,
            group_id=run_config.group_id,
        )
        if prompt_cache_key_resolver is not None
        else None
    )
    model_settings = model_settings_with_prompt_cache_key(model_settings, prompt_cache_key)

    async def rewind_model_request() -> None:
        items_to_rewind = session_items_to_rewind if session_items_to_rewind is not None else []
        await rewind_session_items(session, items_to_rewind, server_conversation_tracker)
        if server_conversation_tracker is not None:
            server_conversation_tracker.rewind_input(filtered.input)

    stream_failed_retry_attempts: list[int] = [0]

    retry_stream = stream_response_with_retry(
        get_stream=lambda: model.stream_response(
            filtered.instructions,
            filtered.input,
            model_settings,
            all_tools,
            output_schema,
            handoffs,
            get_model_tracing_impl(
                run_config.tracing_disabled, run_config.trace_include_sensitive_data
            ),
            previous_response_id=previous_response_id,
            conversation_id=conversation_id,
            prompt=prompt_config,
        ),
        rewind=rewind_model_request,
        retry_settings=model_settings.retry,
        get_retry_advice=model.get_retry_advice,
        previous_response_id=previous_response_id,
        conversation_id=conversation_id,
        failed_retry_attempts_out=stream_failed_retry_attempts,
    )

    async for event in retry_stream:
        streamed_result._event_queue.put_nowait(RawResponsesStreamEvent(data=event))

        terminal_response: Response | None = None
        is_completed_event = False
        if isinstance(event, ResponseCompletedEvent):
            is_completed_event = True
            terminal_response = event.response
        elif getattr(event, "type", None) in {"response.incomplete", "response.failed"}:
            event_type = cast(str, event.type)
            maybe_response = getattr(event, "response", None)
            raise response_terminal_failure_error(
                event_type,
                maybe_response if isinstance(maybe_response, Response) else None,
            )
        elif getattr(event, "type", None) in {"error", "response.error"}:
            raise response_error_event_failure_error(cast(str, event.type), event)

        if terminal_response is not None:
            if is_completed_event and not terminal_response.output and streamed_response_output:
                # Some streaming backends emit output items during item.done events while leaving
                # the terminal response output empty. Preserve those items so the runner can
                # resolve the completed step correctly.
                terminal_response.output = list(streamed_response_output)
            usage = (
                apply_retry_attempt_usage(
                    Usage(
                        requests=1,
                        input_tokens=terminal_response.usage.input_tokens,
                        output_tokens=terminal_response.usage.output_tokens,
                        total_tokens=terminal_response.usage.total_tokens,
                        input_tokens_details=terminal_response.usage.input_tokens_details,
                        output_tokens_details=terminal_response.usage.output_tokens_details,
                    ),
                    stream_failed_retry_attempts[0],
                )
                if terminal_response.usage
                else Usage()
            )
            final_response = ModelResponse(
                output=terminal_response.output,
                usage=usage,
                response_id=terminal_response.id,
                request_id=getattr(terminal_response, "_request_id", None),
            )

        if isinstance(event, ResponseOutputItemDoneEvent):
            output_item = event.item
            streamed_response_output.append(output_item)
            output_item_type = getattr(output_item, "type", None)

            if output_item_type == "tool_search_call":
                emitted_tool_search_fingerprints.add(_tool_search_fingerprint(output_item))
                streamed_result._event_queue.put_nowait(
                    RunItemStreamEvent(
                        item=ToolSearchCallItem(
                            raw_item=coerce_tool_search_call_raw_item(output_item),
                            agent=public_agent,
                        ),
                        name="tool_search_called",
                    )
                )

            elif output_item_type == "tool_search_output":
                emitted_tool_search_fingerprints.add(_tool_search_fingerprint(output_item))
                streamed_result._event_queue.put_nowait(
                    RunItemStreamEvent(
                        item=ToolSearchOutputItem(
                            raw_item=coerce_tool_search_output_raw_item(output_item),
                            agent=public_agent,
                        ),
                        name="tool_search_output_created",
                    )
                )

            elif isinstance(output_item, McpListTools):
                hosted_mcp_tool_metadata.update(collect_mcp_list_tools_metadata([output_item]))

            elif isinstance(output_item, TOOL_CALL_TYPES):
                output_call_id: str | None = getattr(
                    output_item, "call_id", getattr(output_item, "id", None)
                )

                if (
                    output_call_id
                    and isinstance(output_call_id, str)
                    and output_call_id not in emitted_tool_call_ids
                ):
                    emitted_tool_call_ids.add(output_call_id)

                    # Look up tool description from precomputed map ("last wins" matches
                    # execution behavior in process_model_response).
                    tool_lookup_key = get_function_tool_lookup_key_for_call(output_item)
                    matched_tool = (
                        tool_map.get(tool_lookup_key) if tool_lookup_key is not None else None
                    )
                    if (
                        matched_tool is None
                        and output_schema is not None
                        and isinstance(output_item, ResponseFunctionToolCall)
                        and output_item.name == "json_tool_call"
                    ):
                        matched_tool = build_litellm_json_tool_call(output_item)
                    tool_description: str | None = None
                    tool_title: str | None = None
                    tool_origin = None
                    if isinstance(output_item, McpCall):
                        metadata = hosted_mcp_tool_metadata.get(
                            (output_item.server_label, output_item.name)
                        )
                        if metadata is not None:
                            tool_description = metadata.description
                            tool_title = metadata.title
                        tool_origin = ToolOrigin(
                            type=ToolOriginType.MCP,
                            mcp_server_name=output_item.server_label,
                        )
                    elif matched_tool is not None:
                        tool_description = getattr(matched_tool, "description", None)
                        tool_title = getattr(matched_tool, "_mcp_title", None)
                        tool_origin = get_function_tool_origin(matched_tool)

                    tool_item = ToolCallItem(
                        raw_item=cast(ToolCallItemTypes, output_item),
                        agent=public_agent,
                        description=tool_description,
                        title=tool_title,
                        tool_origin=tool_origin,
                    )
                    streamed_result._event_queue.put_nowait(
                        RunItemStreamEvent(item=tool_item, name="tool_called")
                    )

            elif isinstance(output_item, ResponseReasoningItem):
                reasoning_id: str | None = getattr(output_item, "id", None)

                if reasoning_id and reasoning_id not in emitted_reasoning_item_ids:
                    emitted_reasoning_item_ids.add(reasoning_id)

                    reasoning_item = ReasoningItem(raw_item=output_item, agent=public_agent)
                    streamed_result._event_queue.put_nowait(
                        RunItemStreamEvent(item=reasoning_item, name="reasoning_item_created")
                    )

    if final_response is not None:
        context_wrapper.usage.add(final_response.usage)
        await asyncio.gather(
            (
                public_agent.hooks.on_llm_end(context_wrapper, public_agent, final_response)
                if public_agent.hooks
                else _coro.noop_coroutine()
            ),
            hooks.on_llm_end(context_wrapper, public_agent, final_response),
        )

    if not final_response:
        raise ModelBehaviorError("Model did not produce a final response!")

    if server_conversation_tracker is not None:
        # Streaming uses the same rewind helper, so a successful retry must restore delivered
        # input tracking before the next turn computes server-managed deltas.
        server_conversation_tracker.mark_input_as_sent(filtered.input)
        server_conversation_tracker.track_server_items(final_response)

    single_step_result = await get_single_step_result_from_response(
        bindings=bindings,
        original_input=streamed_result.input,
        pre_step_items=streamed_result._model_input_items,
        new_response=final_response,
        output_schema=output_schema,
        all_tools=all_tools,
        handoffs=handoffs,
        hooks=hooks,
        context_wrapper=context_wrapper,
        run_config=run_config,
        error_handlers=error_handlers,
        tool_use_tracker=tool_use_tracker,
        server_manages_conversation=server_conversation_tracker is not None,
        event_queue=streamed_result._event_queue,
        before_side_effects=raise_if_input_guardrail_tripwire_known,
    )

    items_to_filter = session_items_for_turn(single_step_result)

    if emitted_tool_call_ids:
        items_to_filter = [
            item
            for item in items_to_filter
            if not (
                isinstance(item, ToolCallItem)
                and (
                    call_id := getattr(item.raw_item, "call_id", getattr(item.raw_item, "id", None))
                )
                and call_id in emitted_tool_call_ids
            )
        ]

    if emitted_reasoning_item_ids:
        items_to_filter = [
            item
            for item in items_to_filter
            if not (
                isinstance(item, ReasoningItem)
                and (reasoning_id := getattr(item.raw_item, "id", None))
                and reasoning_id in emitted_reasoning_item_ids
            )
        ]

    if emitted_tool_search_fingerprints:
        items_to_filter = [
            item
            for item in items_to_filter
            if not (
                isinstance(item, ToolSearchCallItem | ToolSearchOutputItem)
                and _tool_search_fingerprint(item.raw_item) in emitted_tool_search_fingerprints
            )
        ]

    items_to_filter = [item for item in items_to_filter if not isinstance(item, HandoffCallItem)]

    filtered_result = _dc.replace(single_step_result, new_step_items=items_to_filter)
    stream_step_result_to_queue(filtered_result, streamed_result._event_queue)
    return single_step_result


async def run_single_turn(
    *,
    bindings: AgentBindings[TContext],
    all_tools: list[Tool],
    original_input: str | list[TResponseInputItem],
    generated_items: list[RunItem],
    hooks: RunHooks[TContext],
    context_wrapper: RunContextWrapper[TContext],
    run_config: RunConfig,
    should_run_agent_start_hooks: bool,
    tool_use_tracker: AgentToolUseTracker,
    server_conversation_tracker: OpenAIServerConversationTracker | None = None,
    session: Session | None = None,
    session_items_to_rewind: list[TResponseInputItem] | None = None,
    reasoning_item_id_policy: ReasoningItemIdPolicy | None = None,
    prompt_cache_key_resolver: PromptCacheKeyResolver | None = None,
    error_handlers: RunErrorHandlers[TContext] | None = None,
) -> SingleStepResult:
    """Run a single non-streaming turn of the agent loop."""
    public_agent = bindings.public_agent
    execution_agent = bindings.execution_agent
    try:
        turn_input = ItemHelpers.input_to_new_input_list(original_input)
    except Exception:
        turn_input = []
    context_wrapper.turn_input = list(turn_input)

    if should_run_agent_start_hooks:
        agent_hook_context = AgentHookContext(
            context=context_wrapper.context,
            usage=context_wrapper.usage,
            _approvals=context_wrapper._approvals,
            turn_input=turn_input,
        )
        await asyncio.gather(
            hooks.on_agent_start(agent_hook_context, public_agent),
            (
                public_agent.hooks.on_start(agent_hook_context, public_agent)
                if public_agent.hooks
                else _coro.noop_coroutine()
            ),
        )

    system_prompt, prompt_config = await asyncio.gather(
        execution_agent.get_system_prompt(context_wrapper),
        execution_agent.get_prompt(context_wrapper),
    )

    output_schema = get_output_schema(execution_agent)
    handoffs = await get_handoffs(execution_agent, context_wrapper)
    if server_conversation_tracker is not None:
        input = server_conversation_tracker.prepare_input(original_input, generated_items)
    else:
        input = _prepare_turn_input_items(original_input, generated_items, reasoning_item_id_policy)

    new_response = await get_new_response(
        bindings,
        system_prompt,
        input,
        output_schema,
        all_tools,
        handoffs,
        hooks,
        context_wrapper,
        run_config,
        tool_use_tracker,
        server_conversation_tracker,
        prompt_config,
        session=session,
        session_items_to_rewind=session_items_to_rewind,
        prompt_cache_key_resolver=prompt_cache_key_resolver,
    )

    return await get_single_step_result_from_response(
        bindings=bindings,
        original_input=original_input,
        pre_step_items=generated_items,
        new_response=new_response,
        output_schema=output_schema,
        all_tools=all_tools,
        handoffs=handoffs,
        hooks=hooks,
        context_wrapper=context_wrapper,
        run_config=run_config,
        error_handlers=error_handlers,
        tool_use_tracker=tool_use_tracker,
        server_manages_conversation=server_conversation_tracker is not None,
    )


async def get_new_response(
    bindings: AgentBindings[TContext],
    system_prompt: str | None,
    input: list[TResponseInputItem],
    output_schema: AgentOutputSchemaBase | None,
    all_tools: list[Tool],
    handoffs: list[Handoff],
    hooks: RunHooks[TContext],
    context_wrapper: RunContextWrapper[TContext],
    run_config: RunConfig,
    tool_use_tracker: AgentToolUseTracker,
    server_conversation_tracker: OpenAIServerConversationTracker | None,
    prompt_config: ResponsePromptParam | None,
    session: Session | None = None,
    session_items_to_rewind: list[TResponseInputItem] | None = None,
    prompt_cache_key_resolver: PromptCacheKeyResolver | None = None,
) -> ModelResponse:
    """Call the model and return the raw response, handling retries and hooks."""
    public_agent = bindings.public_agent
    execution_agent = bindings.execution_agent
    filtered = await maybe_filter_model_input(
        agent=public_agent,
        run_config=run_config,
        context_wrapper=context_wrapper,
        input_items=input,
        system_instructions=system_prompt,
    )
    if isinstance(filtered.input, list):
        filtered.input = deduplicate_input_items_preferring_latest(filtered.input)

    model = get_model(execution_agent, run_config)
    model_settings = execution_agent.model_settings.resolve(run_config.model_settings)
    model_settings = maybe_reset_tool_choice(public_agent, tool_use_tracker, model_settings)

    if server_conversation_tracker is not None:
        server_conversation_tracker.mark_input_as_sent(filtered.input)

    await asyncio.gather(
        hooks.on_llm_start(context_wrapper, public_agent, filtered.instructions, filtered.input),
        (
            public_agent.hooks.on_llm_start(
                context_wrapper,
                public_agent,
                filtered.instructions,
                filtered.input,
            )
            if public_agent.hooks
            else _coro.noop_coroutine()
        ),
    )

    previous_response_id = (
        server_conversation_tracker.previous_response_id
        if server_conversation_tracker
        and server_conversation_tracker.previous_response_id is not None
        else None
    )
    conversation_id = (
        server_conversation_tracker.conversation_id if server_conversation_tracker else None
    )
    if conversation_id:
        logger.debug("Using conversation_id=%s", conversation_id)
    else:
        logger.debug("No conversation_id available for request")

    prompt_cache_key = (
        prompt_cache_key_resolver.resolve(
            model_settings,
            model=model,
            conversation_id=conversation_id,
            session=session,
            group_id=run_config.group_id,
        )
        if prompt_cache_key_resolver is not None
        else None
    )
    model_settings = model_settings_with_prompt_cache_key(model_settings, prompt_cache_key)

    async def rewind_model_request() -> None:
        items_to_rewind = session_items_to_rewind if session_items_to_rewind is not None else []
        await rewind_session_items(session, items_to_rewind, server_conversation_tracker)
        if server_conversation_tracker is not None:
            server_conversation_tracker.rewind_input(filtered.input)

    new_response = await get_response_with_retry(
        get_response=lambda: model.get_response(
            system_instructions=filtered.instructions,
            input=filtered.input,
            model_settings=model_settings,
            tools=all_tools,
            output_schema=output_schema,
            handoffs=handoffs,
            tracing=get_model_tracing_impl(
                run_config.tracing_disabled, run_config.trace_include_sensitive_data
            ),
            previous_response_id=previous_response_id,
            conversation_id=conversation_id,
            prompt=prompt_config,
        ),
        rewind=rewind_model_request,
        retry_settings=model_settings.retry,
        get_retry_advice=model.get_retry_advice,
        previous_response_id=previous_response_id,
        conversation_id=conversation_id,
    )
    if server_conversation_tracker is not None:
        # Retry helpers rewind sent-input tracking before replaying a failed request. Mark the
        # filtered input as delivered again once a retry succeeds so subsequent turns only send
        # new deltas.
        server_conversation_tracker.mark_input_as_sent(filtered.input)

    context_wrapper.usage.add(new_response.usage)

    await asyncio.gather(
        (
            public_agent.hooks.on_llm_end(context_wrapper, public_agent, new_response)
            if public_agent.hooks
            else _coro.noop_coroutine()
        ),
        hooks.on_llm_end(context_wrapper, public_agent, new_response),
    )

    return new_response